* [PATCH v2 01/10] tap: Convert packet pools to per-queue-pair arrays for multiqueue
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety Laurent Vivier
` (8 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
Convert the global pool_tap4 and pool_tap6 packet pools from single
pools to arrays of pools, one for each queue pair. This change is
necessary to support multiqueue operation in vhost-user mode, where
multiple queue pairs may be processing packets concurrently.
The pool storage structures (pool_tap4_storage and pool_tap6_storage)
are now arrays of VHOST_USER_MAX_VQS/2 elements, with corresponding
pointer arrays (pool_tap4 and pool_tap6) for accessing them.
Add a qpair parameter to tap_flush_pools() so it flushes the correct
pool. tap4_handler() and tap6_handler() now use the qpair they
already receive to index into the pool arrays. Add bounds checking
assertions in tap_handler() and tap_add_packet().
In passt and pasta modes, all operations use QPAIR_DEFAULT. In
vhost-user mode, the queue pair is derived from the virtqueue index
via QPAIR_FROM_QUEUE().
All pools within the array share the same buffer pointer:
- In vhost-user mode: Points to the vhost-user memory structure, which
is safe as packet data remains in guest memory and pools only track
iovecs
- In passt/pasta mode: Points to pkt_buf, which is safe as only queue
pair 0 is used
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
tap.c | 77 ++++++++++++++++++++++++++++++-----------------------
tap.h | 2 +-
vu_common.c | 2 +-
3 files changed, 46 insertions(+), 35 deletions(-)
diff --git a/tap.c b/tap.c
index 029888e32db5..6bba436f148f 100644
--- a/tap.c
+++ b/tap.c
@@ -94,9 +94,13 @@ CHECK_FRAME_LEN(L2_MAX_LEN_VU);
DIV_ROUND_UP(sizeof(pkt_buf), \
ETH_HLEN + sizeof(struct ipv6hdr) + sizeof(struct udphdr))
-/* IPv4 (plus ARP) and IPv6 message batches from tap/guest to IP handlers */
-static PACKET_POOL_NOINIT(pool_tap4, TAP_MSGS_IP4);
-static PACKET_POOL_NOINIT(pool_tap6, TAP_MSGS_IP6);
+/* IPv4 (plus ARP) and IPv6 message batches from tap/guest to IP handlers
+ * One pool per queue pair for multiqueue support
+ */
+static PACKET_POOL_DECL(pool_tap4, TAP_MSGS_IP4) pool_tap4_storage[VHOST_USER_MAX_VQS / 2];
+static struct pool *pool_tap4[VHOST_USER_MAX_VQS / 2];
+static PACKET_POOL_DECL(pool_tap6, TAP_MSGS_IP6) pool_tap6_storage[VHOST_USER_MAX_VQS / 2];
+static struct pool *pool_tap6[VHOST_USER_MAX_VQS / 2];
#define TAP_SEQS 128 /* Different L4 tuples in one batch */
@@ -702,12 +706,12 @@ static int tap4_handler(struct ctx *c, unsigned int qpair,
unsigned int i, j, seq_count;
struct tap4_l4_t *seq;
- if (!c->ifi4 || !pool_tap4->count)
- return pool_tap4->count;
+ if (!c->ifi4 || !pool_tap4[qpair]->count)
+ return pool_tap4[qpair]->count;
i = 0;
resume:
- for (seq_count = 0, seq = NULL; i < pool_tap4->count; i++) {
+ for (seq_count = 0, seq = NULL; i < pool_tap4[qpair]->count; i++) {
struct iovec trim_iov[UIO_MAXIOV];
size_t l3len, hlen, l4len, check;
struct ethhdr eh_storage;
@@ -718,7 +722,7 @@ resume:
struct iov_tail data;
struct iphdr *iph;
- if (!packet_get(pool_tap4, i, &data))
+ if (!packet_get(pool_tap4[qpair], i, &data))
continue;
eh = IOV_PEEK_HEADER(&data, eh_storage);
@@ -801,7 +805,7 @@ resume:
if (iph->protocol == IPPROTO_UDP) {
struct iov_tail eh_data;
- packet_get(pool_tap4, i, &eh_data);
+ packet_get(pool_tap4[qpair], i, &eh_data);
if (dhcp(c, qpair, &eh_data))
continue;
}
@@ -832,7 +836,7 @@ resume:
goto append;
if (seq_count == TAP_SEQS)
- break; /* Resume after flushing if i < pool_tap4->count */
+ break; /* Resume after flushing if i < pool_tap4[qpair]->count */
for (seq = tap4_l4 + seq_count - 1; seq >= tap4_l4; seq--) {
if (L4_MATCH(iph, uh, seq)) {
@@ -878,10 +882,10 @@ append:
}
}
- if (i < pool_tap4->count)
+ if (i < pool_tap4[qpair]->count)
goto resume;
- return pool_tap4->count;
+ return pool_tap4[qpair]->count;
}
#define IPV6_NH_OPT(nh) \
@@ -950,12 +954,12 @@ static int tap6_handler(struct ctx *c, unsigned int qpair,
unsigned int i, j, seq_count = 0;
struct tap6_l4_t *seq;
- if (!c->ifi6 || !pool_tap6->count)
- return pool_tap6->count;
+ if (!c->ifi6 || !pool_tap6[qpair]->count)
+ return pool_tap6[qpair]->count;
i = 0;
resume:
- for (seq_count = 0, seq = NULL; i < pool_tap6->count; i++) {
+ for (seq_count = 0, seq = NULL; i < pool_tap6[qpair]->count; i++) {
size_t l4len, plen, check;
struct in6_addr *saddr, *daddr;
struct ipv6hdr ip6h_storage;
@@ -967,7 +971,7 @@ resume:
struct ipv6hdr *ip6h;
uint8_t proto;
- if (!packet_get(pool_tap6, i, &data))
+ if (!packet_get(pool_tap6[qpair], i, &data))
return -1;
eh = IOV_REMOVE_HEADER(&data, eh_storage);
@@ -1082,7 +1086,7 @@ resume:
goto append;
if (seq_count == TAP_SEQS)
- break; /* Resume after flushing if i < pool_tap6->count */
+ break; /* Resume after flushing if i < pool_tap6[qpair]->count */
for (seq = tap6_l4 + seq_count - 1; seq >= tap6_l4; seq--) {
if (L4_MATCH(ip6h, proto, uh, seq)) {
@@ -1129,19 +1133,19 @@ append:
}
}
- if (i < pool_tap6->count)
+ if (i < pool_tap6[qpair]->count)
goto resume;
- return pool_tap6->count;
+ return pool_tap6[qpair]->count;
}
/**
- * tap_flush_pools() - Flush both IPv4 and IPv6 packet pools
+ * tap_flush_pools() - Flush both IPv4 and IPv6 packet pools for a given qpair
*/
-void tap_flush_pools(void)
+void tap_flush_pools(unsigned int qpair)
{
- pool_flush(pool_tap4);
- pool_flush(pool_tap6);
+ pool_flush(pool_tap4[qpair]);
+ pool_flush(pool_tap6[qpair]);
}
/**
@@ -1152,6 +1156,7 @@ void tap_flush_pools(void)
*/
void tap_handler(struct ctx *c, unsigned int qpair, const struct timespec *now)
{
+ assert(qpair < VHOST_USER_MAX_VQS / 2);
tap4_handler(c, qpair, now);
tap6_handler(c, qpair, now);
}
@@ -1186,21 +1191,23 @@ void tap_add_packet(struct ctx *c, unsigned int qpair, struct iov_tail *data,
proto_update_l2_buf(c->guest_mac);
}
+ assert(qpair < VHOST_USER_MAX_VQS / 2);
+
switch (ntohs(eh->h_proto)) {
case ETH_P_ARP:
case ETH_P_IP:
- if (!pool_can_fit(pool_tap4, data)) {
+ if (!pool_can_fit(pool_tap4[qpair], data)) {
tap4_handler(c, qpair, now);
- pool_flush(pool_tap4);
+ pool_flush(pool_tap4[qpair]);
}
- packet_add(pool_tap4, data);
+ packet_add(pool_tap4[qpair], data);
break;
case ETH_P_IPV6:
- if (!pool_can_fit(pool_tap6, data)) {
+ if (!pool_can_fit(pool_tap6[qpair], data)) {
tap6_handler(c, qpair, now);
- pool_flush(pool_tap6);
+ pool_flush(pool_tap6[qpair]);
}
- packet_add(pool_tap6, data);
+ packet_add(pool_tap6[qpair], data);
break;
default:
break;
@@ -1238,7 +1245,7 @@ static void tap_passt_input(struct ctx *c, const struct timespec *now)
ssize_t n;
char *p;
- tap_flush_pools();
+ tap_flush_pools(QPAIR_DEFAULT);
if (partial_len) {
/* We have a partial frame from an earlier pass. Move it to the
@@ -1321,7 +1328,7 @@ static void tap_pasta_input(struct ctx *c, const struct timespec *now)
{
ssize_t n, len;
- tap_flush_pools();
+ tap_flush_pools(QPAIR_DEFAULT);
for (n = 0;
n <= (ssize_t)(sizeof(pkt_buf) - L2_MAX_LEN_PASTA);
@@ -1586,10 +1593,14 @@ static void tap_sock_tun_init(struct ctx *c)
*/
static void tap_sock_update_pool(void *base, size_t size)
{
- int i;
+ unsigned int i;
- pool_tap4_storage = PACKET_INIT(pool_tap4, TAP_MSGS_IP4, base, size);
- pool_tap6_storage = PACKET_INIT(pool_tap6, TAP_MSGS_IP6, base, size);
+ for (i = 0; i < VHOST_USER_MAX_VQS / 2; i++) {
+ pool_tap4_storage[i] = PACKET_INIT(pool_tap4, TAP_MSGS_IP4, base, size);
+ pool_tap4[i] = (struct pool *)&pool_tap4_storage[i];
+ pool_tap6_storage[i] = PACKET_INIT(pool_tap6, TAP_MSGS_IP6, base, size);
+ pool_tap6[i] = (struct pool *)&pool_tap6_storage[i];
+ }
for (i = 0; i < TAP_SEQS; i++) {
tap4_l4[i].p = PACKET_INIT(pool_l4, UIO_MAXIOV, base, size);
diff --git a/tap.h b/tap.h
index 0f8e98fc5f84..7d77e071ba49 100644
--- a/tap.h
+++ b/tap.h
@@ -121,7 +121,7 @@ void tap_handler_passt(struct ctx *c, uint32_t events,
int tap_sock_unix_open(char *sock_path);
void tap_sock_reset(struct ctx *c);
void tap_backend_init(struct ctx *c);
-void tap_flush_pools(void);
+void tap_flush_pools(unsigned int qpair);
void tap_handler(struct ctx *c, unsigned int qpair, const struct timespec *now);
void tap_add_packet(struct ctx *c, unsigned int qpair, struct iov_tail *data,
const struct timespec *now);
diff --git a/vu_common.c b/vu_common.c
index 7239cb42f38b..0cd19b36737d 100644
--- a/vu_common.c
+++ b/vu_common.c
@@ -177,7 +177,7 @@ static void vu_handle_tx(struct vu_dev *vdev, int index,
assert(QPAIR_IS_FROMGUEST(index));
- tap_flush_pools();
+ tap_flush_pools(QPAIR_DEFAULT);
count = 0;
out_sg_count = 0;
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 01/10] tap: Convert packet pools to per-queue-pair arrays for multiqueue Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 03/10] tcp: Make static buffers stack-local " Laurent Vivier
` (7 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
The L4 sequence arrays tap4_l4[] and tap6_l4[] are used to batch
packets with the same L4 tuple within a single tap_handler() call.
They are global, but tap_handler() can be called concurrently from
different worker threads with different qpairs in vhost-user mode.
Make these arrays per-qpair by adding a VHOST_USER_MAX_VQS/2 first
dimension, indexed by the qpair parameter already available in
tap4_handler() and tap6_handler().
Update tap_sock_update_pool() to initialize all qpair*seq entries.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
tap.c | 32 +++++++++++++++++++-------------
1 file changed, 19 insertions(+), 13 deletions(-)
diff --git a/tap.c b/tap.c
index 6bba436f148f..25cde67538ac 100644
--- a/tap.c
+++ b/tap.c
@@ -591,7 +591,7 @@ static struct tap4_l4_t {
struct in_addr daddr;
struct pool_l4_t p;
-} tap4_l4[TAP_SEQS /* Arbitrary: TAP_MSGS in theory, so limit in users */];
+} tap4_l4[VHOST_USER_MAX_VQS / 2][TAP_SEQS /* Arbitrary: TAP_MSGS in theory, so limit in users */];
/**
* struct l4_seq6_t - Message sequence for one protocol handler call, IPv6
@@ -618,7 +618,7 @@ static struct tap6_l4_t {
uint8_t hop_limit;
struct pool_l4_t p;
-} tap6_l4[TAP_SEQS /* Arbitrary: TAP_MSGS in theory, so limit in users */];
+} tap6_l4[VHOST_USER_MAX_VQS / 2][TAP_SEQS /* Arbitrary: TAP_MSGS in theory, so limit in users */];
/**
* tap_packet_debug() - Print debug message for packet(s) from guest/tap
@@ -838,7 +838,7 @@ resume:
if (seq_count == TAP_SEQS)
break; /* Resume after flushing if i < pool_tap4[qpair]->count */
- for (seq = tap4_l4 + seq_count - 1; seq >= tap4_l4; seq--) {
+ for (seq = tap4_l4[qpair] + seq_count - 1; seq >= tap4_l4[qpair]; seq--) {
if (L4_MATCH(iph, uh, seq)) {
if (seq->p.count >= UIO_MAXIOV)
seq = NULL;
@@ -846,8 +846,8 @@ resume:
}
}
- if (!seq || seq < tap4_l4) {
- seq = tap4_l4 + seq_count++;
+ if (!seq || seq < tap4_l4[qpair]) {
+ seq = tap4_l4[qpair] + seq_count++;
L4_SET(iph, uh, seq);
pool_flush((struct pool *)&seq->p);
}
@@ -859,7 +859,7 @@ append:
packet_add((struct pool *)&seq->p, &data);
}
- for (j = 0, seq = tap4_l4; j < seq_count; j++, seq++) {
+ for (j = 0, seq = tap4_l4[qpair]; j < seq_count; j++, seq++) {
const struct pool *p = (const struct pool *)&seq->p;
size_t k;
@@ -1088,7 +1088,7 @@ resume:
if (seq_count == TAP_SEQS)
break; /* Resume after flushing if i < pool_tap6[qpair]->count */
- for (seq = tap6_l4 + seq_count - 1; seq >= tap6_l4; seq--) {
+ for (seq = tap6_l4[qpair] + seq_count - 1; seq >= tap6_l4[qpair]; seq--) {
if (L4_MATCH(ip6h, proto, uh, seq)) {
if (seq->p.count >= UIO_MAXIOV)
seq = NULL;
@@ -1096,8 +1096,8 @@ resume:
}
}
- if (!seq || seq < tap6_l4) {
- seq = tap6_l4 + seq_count++;
+ if (!seq || seq < tap6_l4[qpair]) {
+ seq = tap6_l4[qpair] + seq_count++;
L4_SET(ip6h, proto, uh, seq);
pool_flush((struct pool *)&seq->p);
}
@@ -1109,7 +1109,7 @@ append:
packet_add((struct pool *)&seq->p, &data);
}
- for (j = 0, seq = tap6_l4; j < seq_count; j++, seq++) {
+ for (j = 0, seq = tap6_l4[qpair]; j < seq_count; j++, seq++) {
const struct pool *p = (const struct pool *)&seq->p;
size_t k;
@@ -1602,9 +1602,15 @@ static void tap_sock_update_pool(void *base, size_t size)
pool_tap6[i] = (struct pool *)&pool_tap6_storage[i];
}
- for (i = 0; i < TAP_SEQS; i++) {
- tap4_l4[i].p = PACKET_INIT(pool_l4, UIO_MAXIOV, base, size);
- tap6_l4[i].p = PACKET_INIT(pool_l4, UIO_MAXIOV, base, size);
+ for (i = 0; i < VHOST_USER_MAX_VQS / 2; i++) {
+ unsigned int j;
+
+ for (j = 0; j < TAP_SEQS; j++) {
+ tap4_l4[i][j].p = PACKET_INIT(pool_l4, UIO_MAXIOV,
+ base, size);
+ tap6_l4[i][j].p = PACKET_INIT(pool_l4, UIO_MAXIOV,
+ base, size);
+ }
}
}
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 03/10] tcp: Make static buffers stack-local for thread safety
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 01/10] tap: Convert packet pools to per-queue-pair arrays for multiqueue Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 04/10] udp_vu: Make virtqueue " Laurent Vivier
` (6 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
Static buffers shared across all call sites are not safe when multiple
worker threads handle TCP connections concurrently.
In tcp.c, move tcp_iov[] from file scope into tcp_data_from_tap() where
it is exclusively used. At UIO_MAXIOV (1024) entries of struct iovec
(16 bytes each), this adds 16 KiB to the stack frame.
In tcp_vu.c, move iov_vu[], elem[], and frame[] from file scope into
tcp_vu_data_from_sock() and pass them to tcp_vu_sock_recv() as
parameters. Also make iov_msg[] in tcp_vu_sock_recv() a local variable
instead of static, as it is only used within a single call. Combined,
these add roughly 80 KiB across the nested stack frames, which is
acceptable for per-thread stacks.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
tcp.c | 3 +--
tcp_vu.c | 33 ++++++++++++++++++++-------------
2 files changed, 21 insertions(+), 15 deletions(-)
diff --git a/tcp.c b/tcp.c
index e80a64e38f24..ee4ec599e055 100644
--- a/tcp.c
+++ b/tcp.c
@@ -435,8 +435,6 @@ static socklen_t tcp_info_size;
/* Kernel reports delivery rate in TCP_INFO (kernel commit eb8329e0a04d) */
#define delivery_rate_cap tcp_info_cap(delivery_rate)
-/* sendmsg() to socket */
-static struct iovec tcp_iov [UIO_MAXIOV];
/* Pools for pre-opened sockets (in init) */
int init_sock_pool4 [TCP_SOCK_POOL_SIZE];
@@ -1945,6 +1943,7 @@ static int tcp_data_from_tap(const struct ctx *c, struct tcp_tap_conn *conn,
uint16_t max_ack_seq_wnd = conn->wnd_from_tap;
uint32_t max_ack_seq = conn->seq_ack_from_tap;
uint32_t seq_from_tap = conn->seq_from_tap;
+ struct iovec tcp_iov[UIO_MAXIOV];
struct msghdr mh = { .msg_iov = tcp_iov };
size_t len;
ssize_t n;
diff --git a/tcp_vu.c b/tcp_vu.c
index c9a1e08747b5..b4baecc1f717 100644
--- a/tcp_vu.c
+++ b/tcp_vu.c
@@ -35,9 +35,6 @@
#include "vu_common.h"
#include <time.h>
-static struct iovec iov_vu[VIRTQUEUE_MAX_SIZE];
-static struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE];
-
/**
* struct vu_frame - Descriptor for a TCP frame mapped to virtqueue elements
* @idx_element: Index of first element in elem[] for this frame
@@ -46,13 +43,13 @@ static struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE];
* @num_iovec: Number of iovecs covering this frame's buffers
* @size: Total frame size including all headers
*/
-static struct vu_frame {
+struct vu_frame {
int idx_element;
int num_element;
int idx_iovec;
int num_iovec;
size_t size;
-} frame[VIRTQUEUE_MAX_SIZE];
+};
/**
* tcp_vu_hdrlen() - Sum size of all headers, from TCP to virtio-net
@@ -225,6 +222,9 @@ int tcp_vu_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags,
* @v6: Set for IPv6 connections
* @already_sent: Number of bytes already sent
* @fillsize: Maximum bytes to fill in guest-side receiving window
+ * @iov_vu: IO vector array for virtqueue buffers
+ * @elem: Virtqueue element array
+ * @frame: Frame descriptor array
* @elem_used: number of element (output)
* @frame_cnt: Pointer to store the number of frames (output)
*
@@ -234,9 +234,12 @@ int tcp_vu_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags,
static ssize_t tcp_vu_sock_recv(const struct ctx *c, struct vu_virtq *vq,
const struct tcp_tap_conn *conn, bool v6,
uint32_t already_sent, size_t fillsize,
+ struct iovec *iov_vu,
+ struct vu_virtq_element *elem,
+ struct vu_frame *frame,
int *elem_used, int *frame_cnt)
{
- static struct iovec iov_msg[VIRTQUEUE_MAX_SIZE + DISCARD_IOV_NUM];
+ struct iovec iov_msg[VIRTQUEUE_MAX_SIZE + DISCARD_IOV_NUM];
const struct vu_dev *vdev = c->vdev;
struct msghdr mh_sock = { 0 };
uint16_t mss = MSS_GET(conn);
@@ -253,16 +256,16 @@ static ssize_t tcp_vu_sock_recv(const struct ctx *c, struct vu_virtq *vq,
iov_used = 0;
elem_cnt = 0;
*frame_cnt = 0;
- while (fillsize > 0 && elem_cnt < ARRAY_SIZE(elem) &&
- iov_used < ARRAY_SIZE(iov_vu) &&
- *frame_cnt < ARRAY_SIZE(frame)) {
+ while (fillsize > 0 && elem_cnt < VIRTQUEUE_MAX_SIZE &&
+ iov_used < VIRTQUEUE_MAX_SIZE &&
+ *frame_cnt < VIRTQUEUE_MAX_SIZE) {
size_t frame_size, in_total;
int cnt;
cnt = vu_collect(vdev, vq, &elem[elem_cnt],
- ARRAY_SIZE(elem) - elem_cnt,
+ VIRTQUEUE_MAX_SIZE - elem_cnt,
&iov_vu[iov_used],
- ARRAY_SIZE(iov_vu) - iov_used, &in_total,
+ VIRTQUEUE_MAX_SIZE - iov_used, &in_total,
MIN(mss, fillsize) + hdrlen,
&frame_size);
if (cnt == 0)
@@ -328,7 +331,8 @@ static ssize_t tcp_vu_sock_recv(const struct ctx *c, struct vu_virtq *vq,
if ((size_t)ret <= f->size - hdrlen) {
unsigned cnt;
- cnt = iov_skip_bytes(&iov_vu[f->idx_iovec], f->num_iovec,
+ cnt = iov_skip_bytes(&iov_vu[f->idx_iovec],
+ f->num_iovec,
MAX(hdrlen + ret, VNET_HLEN + ETH_ZLEN),
NULL);
if (cnt < (unsigned)f->num_iovec)
@@ -437,6 +441,9 @@ int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn,
uint32_t already_sent, const struct timespec *now,
unsigned int qpair)
{
+ struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE];
+ struct iovec iov_vu[VIRTQUEUE_MAX_SIZE];
+ struct vu_frame frame[VIRTQUEUE_MAX_SIZE];
uint32_t wnd_scaled = conn->wnd_from_tap << conn->ws_from_tap;
int toguest = QPAIR_TOGUEST_QUEUE(qpair);
struct vu_dev *vdev = c->vdev;
@@ -460,7 +467,7 @@ int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn,
* data from the socket
*/
len = tcp_vu_sock_recv(c, vq, conn, v6, already_sent, fillsize,
- &elem_cnt, &frame_cnt);
+ iov_vu, elem, frame, &elem_cnt, &frame_cnt);
if (len < 0) {
if (len != -EAGAIN && len != -EWOULDBLOCK) {
tcp_rst(c, conn, now, qpair);
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 04/10] udp_vu: Make virtqueue buffers stack-local for thread safety
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
` (2 preceding siblings ...)
2026-07-31 16:23 ` [PATCH v2 03/10] tcp: Make static buffers stack-local " Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 05/10] flow: Make flow timer per-caller " Laurent Vivier
` (5 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
The function-local static buffers elem[] and iov_vu[] in
udp_vu_sock_to_tap() are shared across all threads. When multiple
worker threads process UDP vhost-user data concurrently, they would
stomp on each other's buffers.
Remove the static qualifier so each call gets its own stack-allocated
arrays, eliminating cross-thread sharing.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
udp_vu.c | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/udp_vu.c b/udp_vu.c
index 21fa891b3b51..c045e421001c 100644
--- a/udp_vu.c
+++ b/udp_vu.c
@@ -151,9 +151,9 @@ void udp_vu_sock_to_tap(const struct ctx *c, int s, int n, flow_sidx_t tosidx,
{
const struct flowside *toside = flowside_at_sidx(tosidx);
bool v6 = !(inany_v4(&toside->eaddr) && inany_v4(&toside->oaddr));
- static struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE];
+ struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE];
+ struct iovec iov_vu[VIRTQUEUE_MAX_SIZE];
int toguest = QPAIR_TOGUEST_QUEUE(qpair);
- static struct iovec iov_vu[VIRTQUEUE_MAX_SIZE];
struct vu_dev *vdev = c->vdev;
struct vu_virtq *vq = &vdev->vq[toguest];
size_t hdrlen = udp_vu_hdrlen(v6);
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 05/10] flow: Make flow timer per-caller for thread safety
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
` (3 preceding siblings ...)
2026-07-31 16:23 ` [PATCH v2 04/10] udp_vu: Make virtqueue " Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 06/10] tcp: Make TCP timer state per-caller and guard global tasks Laurent Vivier
` (4 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
Move the static flow_timer_run variable out of flow.c and pass it as a
parameter to flow_defer_handler(). This allows each caller to maintain
its own timer state: each vhost-user queue pair worker uses the per-qpair
context.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
flow.c | 10 ++++------
flow.h | 2 +-
passt.c | 8 +++++---
3 files changed, 10 insertions(+), 10 deletions(-)
diff --git a/flow.c b/flow.c
index 1efa9b966991..59963ea5b1c2 100644
--- a/flow.c
+++ b/flow.c
@@ -142,9 +142,6 @@ static flow_sidx_t flow_hashtab[FLOW_HASH_SIZE];
static_assert(ARRAY_SIZE(flow_hashtab) >= 2 * FLOW_MAX,
"Safe linear probing requires hash table with more entries than the number of sides in the flow table");
-/* Last time the flow timers ran */
-static struct timespec flow_timer_run;
-
/** flowside_from_af() - Initialise flowside from addresses
* @side: flowside to initialise
* @af: Address family (AF_INET or AF_INET6)
@@ -875,10 +872,11 @@ flow_sidx_t flow_lookup_sa(const struct ctx *c, uint8_t proto, uint8_t pif,
* flow_defer_handler() - Handler for per-flow deferred and timed tasks
* @c: Execution context
* @now: Current timestamp
+ * @timer_run: Last time the flow timers ran
* @qpair: Queue pair to process
*/
void flow_defer_handler(const struct ctx *c, const struct timespec *now,
- unsigned int qpair)
+ struct timespec *timer_run, unsigned int qpair)
{
struct flow_free_cluster *free_head = NULL;
unsigned *last_next = &flow_first_free;
@@ -886,9 +884,9 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
bool timer = false;
union flow *flow;
- if (timespec_diff_ms(now, &flow_timer_run) >= FLOW_TIMER_INTERVAL) {
+ if (timespec_diff_ms(now, timer_run) >= FLOW_TIMER_INTERVAL) {
timer = true;
- flow_timer_run = *now;
+ *timer_run = *now;
}
assert(!flow_new_entry); /* Incomplete flow at end of cycle */
diff --git a/flow.h b/flow.h
index 028d42bccc0c..ac1d3897ca98 100644
--- a/flow.h
+++ b/flow.h
@@ -272,7 +272,7 @@ int flow_epoll_set(const struct flow_common *f, int command, uint32_t events,
int fd, unsigned int sidei);
void flow_defer_handler(const struct ctx *c, const struct timespec *now,
- unsigned int qpair);
+ struct timespec *timer_run, unsigned int qpair);
int flow_migrate_source_early(struct ctx *c, const struct migrate_stage *stage,
int fd);
int flow_migrate_source_pre(struct ctx *c, const struct migrate_stage *stage,
diff --git a/passt.c b/passt.c
index bc9f1575ef8c..69fd7b12450a 100644
--- a/passt.c
+++ b/passt.c
@@ -108,15 +108,16 @@ struct passt_stats {
* post_handler() - Run periodic and deferred tasks for L4 protocol handlers
* @c: Execution context
* @now: Current timestamp
+ * @timer_run: Last time the flow timers ran
* @qpair: Queue pair to process
*/
static void post_handler(struct ctx *c, const struct timespec *now,
- unsigned int qpair)
+ struct timespec *timer_run, unsigned int qpair)
{
if (!c->no_tcp)
tcp_defer_handler(c, now, qpair);
- flow_defer_handler(c, now, qpair);
+ flow_defer_handler(c, now, timer_run, qpair);
fwd_scan_ports_timer(c, now);
if (!c->no_ndp)
@@ -231,6 +232,7 @@ static void print_stats(const struct ctx *c, const struct passt_stats *stats,
static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
{
static struct passt_stats stats = { 0 };
+ static struct timespec flow_timer_run;
struct ctx *c = opaque;
struct timespec now;
int i;
@@ -314,7 +316,7 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
print_stats(c, &stats, &now);
}
- post_handler(c, &now, QPAIR_DEFAULT);
+ post_handler(c, &now, &flow_timer_run, QPAIR_DEFAULT);
migrate_handler(c, &now);
}
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 06/10] tcp: Make TCP timer state per-caller and guard global tasks
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
` (4 preceding siblings ...)
2026-07-31 16:23 ` [PATCH v2 05/10] flow: Make flow timer per-caller " Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 07/10] tcp: Protect init socket pools with mutex for thread safety Laurent Vivier
` (3 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
tcp_defer_handler() uses c->tcp.timer_run, c->tcp.keepalive_run, and
c->tcp.inactivity_run as global timer gates shared across all callers.
In multiqueue mode, multiple qpair workers will call tcp_defer_handler()
concurrently, causing races on these fields. It also unconditionally
runs tcp_payload_flush(), tcp_sock_refill_init(), and tcp_splice_refill()
which operate on global state.
Add timer_run, keepalive_run, and inactivity_run as parameters so each
caller provides its own per-qpair timer state. Remove the now-unused
fields from struct tcp_ctx and drop timer_init() which only initialised
c->tcp.timer_run.
Guard tcp_payload_flush() and socket pool refills with qpair == 0 since
they operate on global buffers shared across all queue pairs.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
passt.c | 37 +++++++++++++++++--------------------
tcp.c | 51 +++++++++++++++++++++++++++++++--------------------
tcp.h | 9 ++-------
3 files changed, 50 insertions(+), 47 deletions(-)
diff --git a/passt.c b/passt.c
index 69fd7b12450a..25e1212b9452 100644
--- a/passt.c
+++ b/passt.c
@@ -106,16 +106,23 @@ struct passt_stats {
/**
* post_handler() - Run periodic and deferred tasks for L4 protocol handlers
- * @c: Execution context
- * @now: Current timestamp
- * @timer_run: Last time the flow timers ran
- * @qpair: Queue pair to process
+ * @c: Execution context
+ * @now: Current timestamp
+ * @timer_run: Last time the flow timers ran
+ * @tcp_timer_run: Last time TCP timers ran
+ * @keepalive_run: Last time keepalives ran
+ * @inactivity_run: Last time inactivity scan ran
+ * @qpair: Queue pair to process
*/
static void post_handler(struct ctx *c, const struct timespec *now,
- struct timespec *timer_run, unsigned int qpair)
+ struct timespec *timer_run,
+ struct timespec *tcp_timer_run,
+ time_t *keepalive_run,
+ time_t *inactivity_run, unsigned int qpair)
{
if (!c->no_tcp)
- tcp_defer_handler(c, now, qpair);
+ tcp_defer_handler(c, now, tcp_timer_run, keepalive_run,
+ inactivity_run, qpair);
flow_defer_handler(c, now, timer_run, qpair);
fwd_scan_ports_timer(c, now);
@@ -140,16 +147,6 @@ static void random_init(struct ctx *c)
srandom(seed);
}
-/**
- * timer_init() - Set initial timestamp for timer runs to current time
- * @c: Execution context
- * @now: Current timestamp
- */
-static void timer_init(struct ctx *c, const struct timespec *now)
-{
- c->tcp.timer_run = *now;
-}
-
/**
* proto_update_l2_buf() - Update scatter-gather L2 buffers in protocol handlers
* @eth_d: Ethernet destination address, NULL if unchanged
@@ -231,8 +228,9 @@ static void print_stats(const struct ctx *c, const struct passt_stats *stats,
*/
static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
{
+ static time_t keepalive_run, inactivity_run;
static struct passt_stats stats = { 0 };
- static struct timespec flow_timer_run;
+ static struct timespec flow_timer_run, tcp_timer_run;
struct ctx *c = opaque;
struct timespec now;
int i;
@@ -316,7 +314,8 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
print_stats(c, &stats, &now);
}
- post_handler(c, &now, &flow_timer_run, QPAIR_DEFAULT);
+ post_handler(c, &now, &flow_timer_run, &tcp_timer_run,
+ &keepalive_run, &inactivity_run, QPAIR_DEFAULT);
migrate_handler(c, &now);
}
@@ -453,8 +452,6 @@ int main(int argc, char **argv)
isolate_postfork(c);
- timer_init(c, &now);
-
loop:
/* NOLINTBEGIN(bugprone-branch-clone): intervals can be the same */
/* cppcheck-suppress [duplicateValueTernary, unmatchedSuppression] */
diff --git a/tcp.c b/tcp.c
index ee4ec599e055..ab7cbfa5de83 100644
--- a/tcp.c
+++ b/tcp.c
@@ -2989,17 +2989,19 @@ int tcp_init(struct ctx *c)
/**
* tcp_keepalive() - Send keepalives for connections which need it
* @c: Execution context
+ * @now: Current timestamp
+ * @last_run: Last time keepalives ran, updated on run
* @qpair: Queue pair to process
*/
-static void tcp_keepalive(struct ctx *c, const struct timespec *now,
- unsigned int qpair)
+static void tcp_keepalive(const struct ctx *c, const struct timespec *now,
+ time_t *last_run, unsigned int qpair)
{
union flow *flow;
- if (now->tv_sec - c->tcp.keepalive_run < KEEPALIVE_INTERVAL)
+ if (now->tv_sec - *last_run < KEEPALIVE_INTERVAL)
return;
- c->tcp.keepalive_run = now->tv_sec;
+ *last_run = now->tv_sec;
flow_foreach_of_type(flow, FLOW_TCP) {
struct tcp_tap_conn *conn = &flow->tcp;
@@ -3022,18 +3024,20 @@ static void tcp_keepalive(struct ctx *c, const struct timespec *now,
/**
* tcp_inactivity() - Scan for and close long-inactive connections
* @c: Execution context
+ * @now: Current timestamp
+ * @last_run: Last time inactivity scan ran, updated on run
* @qpair: Queue pair to process
*/
-static void tcp_inactivity(struct ctx *c, const struct timespec *now,
- unsigned int qpair)
+static void tcp_inactivity(const struct ctx *c, const struct timespec *now,
+ time_t *last_run, unsigned int qpair)
{
union flow *flow;
- if (now->tv_sec - c->tcp.inactivity_run < INACTIVITY_INTERVAL)
+ if (now->tv_sec - *last_run < INACTIVITY_INTERVAL)
return;
debug("TCP inactivity scan");
- c->tcp.inactivity_run = now->tv_sec;
+ *last_run = now->tv_sec;
flow_foreach_of_type(flow, FLOW_TCP) {
struct tcp_tap_conn *conn = &flow->tcp;
@@ -3055,27 +3059,34 @@ static void tcp_inactivity(struct ctx *c, const struct timespec *now,
/**
* tcp_defer_handler() - Handler for TCP deferred tasks
- * @c: Execution context
- * @now: Current timestamp
- * @qpair: Queue pair to process
+ * @c: Execution context
+ * @now: Current timestamp
+ * @timer_run: Last time TCP timers ran
+ * @keepalive_run: Last time keepalives ran
+ * @inactivity_run: Last time inactivity scan ran
+ * @qpair: Queue pair to process
*/
/* cppcheck-suppress [constParameterPointer, unmatchedSuppression] */
void tcp_defer_handler(struct ctx *c, const struct timespec *now,
- unsigned int qpair)
+ struct timespec *timer_run, time_t *keepalive_run,
+ time_t *inactivity_run, unsigned int qpair)
{
- tcp_payload_flush(c, now);
+ if (qpair == 0)
+ tcp_payload_flush(c, now);
- if (timespec_diff_ms(now, &c->tcp.timer_run) < TCP_TIMER_INTERVAL)
+ if (timespec_diff_ms(now, timer_run) < TCP_TIMER_INTERVAL)
return;
- c->tcp.timer_run = *now;
+ *timer_run = *now;
- tcp_sock_refill_init(c);
- if (c->mode == MODE_PASTA)
- tcp_splice_refill(c);
+ if (qpair == 0) {
+ tcp_sock_refill_init(c);
+ if (c->mode == MODE_PASTA)
+ tcp_splice_refill(c);
+ }
- tcp_keepalive(c, now, qpair);
- tcp_inactivity(c, now, qpair);
+ tcp_keepalive(c, now, keepalive_run, qpair);
+ tcp_inactivity(c, now, inactivity_run, qpair);
}
/**
diff --git a/tcp.h b/tcp.h
index eb4f8994f4b7..1500fae5162d 100644
--- a/tcp.h
+++ b/tcp.h
@@ -31,7 +31,8 @@ int tcp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif,
const struct timespec *now);
int tcp_init(struct ctx *c);
void tcp_defer_handler(struct ctx *c, const struct timespec *now,
- unsigned int qpair);
+ struct timespec *timer_run, time_t *keepalive_run,
+ time_t *inactivity_run, unsigned int qpair);
void tcp_update_l2_buf(const unsigned char *eth_d);
@@ -41,24 +42,18 @@ extern bool peek_offset_cap;
* struct tcp_ctx - Execution context for TCP routines
* @scan_in: Port scanning state for inbound packets
* @scan_out: Port scanning state for outbound packets
- * @timer_run: Timestamp of most recent timer run
* @pipe_size: Size of pipes for spliced connections
* @rto_max: Maximum retry timeout (in s)
* @syn_retries: SYN retries using exponential backoff timeout
* @syn_linear_timeouts: SYN retries before using exponential backoff timeout
- * @keepalive_run: Time we last issued tap-side keepalives
- * @inactivity_run: Time we last scanned for inactive connections
*/
struct tcp_ctx {
struct fwd_scan scan_in;
struct fwd_scan scan_out;
- struct timespec timer_run;
size_t pipe_size;
int rto_max;
uint8_t syn_retries;
uint8_t syn_linear_timeouts;
- time_t keepalive_run;
- time_t inactivity_run;
};
#endif /* TCP_H */
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 07/10] tcp: Protect init socket pools with mutex for thread safety
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
` (5 preceding siblings ...)
2026-07-31 16:23 ` [PATCH v2 06/10] tcp: Make TCP timer state per-caller and guard global tasks Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 08/10] tcp: Extract tcp_timer_epoll_add() helper Laurent Vivier
` (2 subsequent siblings)
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
The pre-opened socket pools init_sock_pool4/6 are consumed by
tcp_conn_pool_sock() when creating new connections from any worker
thread, and refilled by tcp_sock_refill_pool() from tcp_timer() in
post_handler(). These can run concurrently on different threads.
Add a mutex protecting both operations in tcp_conn_sock() and
tcp_sock_refill_init(), where init namespace pools are accessed.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
tcp.c | 10 +++++++++-
1 file changed, 9 insertions(+), 1 deletion(-)
diff --git a/tcp.c b/tcp.c
index ab7cbfa5de83..aef689faf031 100644
--- a/tcp.c
+++ b/tcp.c
@@ -293,6 +293,7 @@
#include <sys/uio.h>
#include <time.h>
#include <arpa/inet.h>
+#include <pthread.h>
#include <linux/sockios.h>
#include <linux/sock_diag.h>
@@ -439,6 +440,7 @@ static socklen_t tcp_info_size;
/* Pools for pre-opened sockets (in init) */
int init_sock_pool4 [TCP_SOCK_POOL_SIZE];
int init_sock_pool6 [TCP_SOCK_POOL_SIZE];
+static pthread_mutex_t sock_pool_lock = PTHREAD_MUTEX_INITIALIZER;
/**
* conn_at_sidx() - Get TCP connection specific flow at given sidx
@@ -1581,7 +1583,11 @@ int tcp_conn_sock(sa_family_t af)
int *pool = af == AF_INET6 ? init_sock_pool6 : init_sock_pool4;
int s;
- if ((s = tcp_conn_pool_sock(pool)) >= 0)
+ pthread_mutex_lock(&sock_pool_lock);
+ s = tcp_conn_pool_sock(pool);
+ pthread_mutex_unlock(&sock_pool_lock);
+
+ if (s >= 0)
return s;
/* If the pool is empty we just open a new one without refilling the
@@ -2858,6 +2864,7 @@ int tcp_sock_refill_pool(int pool[], sa_family_t af)
*/
static void tcp_sock_refill_init(const struct ctx *c)
{
+ pthread_mutex_lock(&sock_pool_lock);
if (c->ifi4) {
int rc = tcp_sock_refill_pool(init_sock_pool4, AF_INET);
if (rc < 0)
@@ -2870,6 +2877,7 @@ static void tcp_sock_refill_init(const struct ctx *c)
warn("TCP: Error refilling IPv6 host socket pool: %s",
strerror_(-rc));
}
+ pthread_mutex_unlock(&sock_pool_lock);
}
/**
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 08/10] tcp: Extract tcp_timer_epoll_add() helper
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
` (6 preceding siblings ...)
2026-07-31 16:23 ` [PATCH v2 07/10] tcp: Protect init socket pools with mutex for thread safety Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 10/10] flow: Add lazy, lock-free flow migration between queue pairs Laurent Vivier
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
Factor out the epoll registration of TCP timers from tcp_timer_ctl()
into a separate tcp_timer_epoll_add() helper.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
tcp.c | 24 +++++++++++++++++-------
1 file changed, 17 insertions(+), 7 deletions(-)
diff --git a/tcp.c b/tcp.c
index aef689faf031..88caf5d8968b 100644
--- a/tcp.c
+++ b/tcp.c
@@ -545,6 +545,22 @@ static int tcp_epoll_ctl(struct tcp_tap_conn *conn)
return 0;
}
+static int tcp_timer_epoll_add(struct tcp_tap_conn *conn, int fd,
+ const struct timespec *now)
+{
+ union epoll_ref ref;
+
+ ref.type = EPOLL_TYPE_TCP_TIMER;
+ ref.flow = FLOW_IDX(conn);
+ ref.fd = fd;
+ if (epoll_add(flow_epollfd(&conn->f), EPOLLIN | EPOLLET, ref) < 0) {
+ flow_perror_ratelimit(conn, now, "failed to add timer");
+ return -1;
+ }
+
+ return 0;
+}
+
/**
* tcp_timer_ctl() - Set timerfd based on flags/events, create timerfd if needed
* @c: Execution context
@@ -562,7 +578,6 @@ static void tcp_timer_ctl(const struct ctx *c, struct tcp_tap_conn *conn,
return;
if (conn->timer == -1) {
- union epoll_ref ref;
int fd;
fd = timerfd_create(CLOCK_MONOTONIC, 0);
@@ -578,12 +593,7 @@ static void tcp_timer_ctl(const struct ctx *c, struct tcp_tap_conn *conn,
return;
}
- ref.type = EPOLL_TYPE_TCP_TIMER;
- ref.flow = FLOW_IDX(conn);
- ref.fd = fd;
- if (epoll_add(flow_epollfd(&conn->f), EPOLLIN | EPOLLET,
- ref) < 0) {
- flow_perror_ratelimit(conn, now, "failed to add timer");
+ if (tcp_timer_epoll_add(conn, fd, now) < 0) {
close(fd);
return;
}
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
` (7 preceding siblings ...)
2026-07-31 16:23 ` [PATCH v2 08/10] tcp: Extract tcp_timer_epoll_add() helper Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 10/10] flow: Add lazy, lock-free flow migration between queue pairs Laurent Vivier
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
Protect flow table shared state (free list, hash table) with a
pthread_rwlock_t. Make flow_new_entry _Thread_local.
Filter flow_defer_handler() by qpair so each thread only processes
its own flows.
Hold the write lock during flow_defer_handler()'s second pass
(free-list rebuild), since the free list is shared with flow_alloc().
Since the lock is released between flow_alloc() and FLOW_ACTIVATE(),
other threads can observe intermediate flow states (NEW, INI, TGT,
TYPED) during traversal. Adapt flow_foreach() to skip them silently
instead of logging an error, and change flow_defer_handler()'s
free-list rebuild to break cluster merging across them instead of
asserting.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
flow.c | 100 ++++++++++++++++++++++++++++++++++++++++++++-------
flow_table.h | 2 +-
2 files changed, 88 insertions(+), 14 deletions(-)
diff --git a/flow.c b/flow.c
index 59963ea5b1c2..8deca3e3c7f1 100644
--- a/flow.c
+++ b/flow.c
@@ -12,6 +12,8 @@
#include <sched.h>
#include <string.h>
+#include <pthread.h>
+
#include "util.h"
#include "ip.h"
#include "passt.h"
@@ -77,7 +79,10 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES,
/* Global Flow Table */
/**
- * DOC: Theory of Operation - allocating and freeing flow entries
+ * DOC: Theory of Operation
+ *
+ * Allocating and freeing flow entries
+ * ===================================
*
* Flows are entries in flowtab[]. We need to routinely scan the whole table to
* perform deferred bookkeeping tasks on active entries, and sparse empty slots
@@ -125,11 +130,43 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES,
* when we encounter the start of a free cluster, we can immediately skip
* past it, meaning that in practice we only need (number of active
* connections) + (number of free clusters) iterations.
+ *
+ * Flow table locking
+ * ==================
+ *
+ * The flow table has three pieces of shared global state: the free cluster
+ * list (flow_first_free, flowtab[].free), the hash table (flow_hashtab[]),
+ * and the in-progress allocation pointer (flow_new_entry).
+ *
+ * A pthread_rwlock_t (flow_lock) protects the free list and hash table:
+ *
+ * - flow_alloc() and flow_alloc_cancel() take the write lock to modify
+ * the free list.
+ *
+ * - flow_hash_insert() and flow_hash_remove() take the write lock to
+ * modify the hash table.
+ *
+ * - flowside_lookup() takes the read lock to traverse the hash table.
+ *
+ * flow_new_entry is _Thread_local, so each worker thread independently
+ * tracks its own in-progress flow allocation.
+ *
+ * Between flow_alloc() and FLOW_ACTIVATE() (or flow_alloc_cancel()), the
+ * flow is in an intermediate state (NEW, INI, TGT, or TYPED). Other
+ * threads may observe these states during flow table traversal.
+ * flow_foreach() skips them silently, and the free-list rebuild in
+ * flow_defer_handler() breaks cluster merging across them.
+ *
+ * The write lock is held in flow_defer_handler() only for the second
+ * pass (free-list rebuild), which modifies the global free list shared
+ * with flow_alloc(). The first pass (deferred protocol handlers) runs
+ * without the lock and filters by qpair, so each thread only processes
+ * its own flows.
*/
unsigned flow_first_free;
union flow flowtab[FLOW_MAX];
-static const union flow *flow_new_entry; /* = NULL */
+static _Thread_local const union flow *flow_new_entry; /* = NULL */
int qpair_to_fd[FLOW_QPAIR_SIZE];
/* Hash table to index it */
@@ -142,6 +179,8 @@ static flow_sidx_t flow_hashtab[FLOW_HASH_SIZE];
static_assert(ARRAY_SIZE(flow_hashtab) >= 2 * FLOW_MAX,
"Safe linear probing requires hash table with more entries than the number of sides in the flow table");
+static pthread_rwlock_t flow_lock = PTHREAD_RWLOCK_INITIALIZER;
+
/** flowside_from_af() - Initialise flowside from addresses
* @side: flowside to initialise
* @af: Address family (AF_INET or AF_INET6)
@@ -592,12 +631,18 @@ void flow_activate(struct flow_common *f)
*/
union flow *flow_alloc(unsigned int qpair)
{
- union flow *flow = &flowtab[flow_first_free];
+ union flow *flow;
+
+ pthread_rwlock_wrlock(&flow_lock);
+
+ flow = &flowtab[flow_first_free];
assert(!flow_new_entry);
- if (flow_first_free >= FLOW_MAX)
+ if (flow_first_free >= FLOW_MAX) {
+ pthread_rwlock_unlock(&flow_lock);
return NULL;
+ }
assert(flow->f.state == FLOW_STATE_FREE);
assert(flow->f.type == FLOW_TYPE_NONE);
@@ -627,6 +672,8 @@ union flow *flow_alloc(unsigned int qpair)
flow_setqp(&flow->f, qpair);
flow_set_state(&flow->f, FLOW_STATE_NEW);
+ pthread_rwlock_unlock(&flow_lock);
+
return flow;
}
@@ -638,6 +685,8 @@ union flow *flow_alloc(unsigned int qpair)
*/
void flow_alloc_cancel(union flow *flow)
{
+ pthread_rwlock_wrlock(&flow_lock);
+
assert(flow_new_entry == flow);
assert(flow->f.state == FLOW_STATE_NEW ||
flow->f.state == FLOW_STATE_INI ||
@@ -655,6 +704,8 @@ void flow_alloc_cancel(union flow *flow)
flow->free.next = flow_first_free;
flow_first_free = FLOW_IDX(flow);
flow_new_entry = NULL;
+
+ pthread_rwlock_unlock(&flow_lock);
}
/**
@@ -739,10 +790,15 @@ static inline unsigned flow_hash_probe(const struct ctx *c, flow_sidx_t sidx)
*/
uint64_t flow_hash_insert(const struct ctx *c, flow_sidx_t sidx)
{
- uint64_t hash = flow_sidx_hash(c, sidx);
- unsigned b = flow_hash_probe_(hash, sidx);
+ uint64_t hash;
+ unsigned b;
+ pthread_rwlock_wrlock(&flow_lock);
+ hash = flow_sidx_hash(c, sidx);
+ b = flow_hash_probe_(hash, sidx);
flow_hashtab[b] = sidx;
+ pthread_rwlock_unlock(&flow_lock);
+
flow_dbg(flow_at_sidx(sidx), "Side %u hash table insert: bucket: %u",
sidx.sidei, b);
@@ -756,10 +812,15 @@ uint64_t flow_hash_insert(const struct ctx *c, flow_sidx_t sidx)
*/
void flow_hash_remove(const struct ctx *c, flow_sidx_t sidx)
{
- unsigned b = flow_hash_probe(c, sidx), s;
+ unsigned b, s;
+
+ pthread_rwlock_wrlock(&flow_lock);
+ b = flow_hash_probe(c, sidx);
- if (!flow_sidx_valid(flow_hashtab[b]))
+ if (!flow_sidx_valid(flow_hashtab[b])) {
+ pthread_rwlock_unlock(&flow_lock);
return; /* Redundant remove */
+ }
flow_dbg(flow_at_sidx(sidx), "Side %u hash table remove: bucket: %u",
sidx.sidei, b);
@@ -779,6 +840,7 @@ void flow_hash_remove(const struct ctx *c, flow_sidx_t sidx)
}
flow_hashtab[b] = FLOW_SIDX_NONE;
+ pthread_rwlock_unlock(&flow_lock);
}
/**
@@ -793,10 +855,12 @@ void flow_hash_remove(const struct ctx *c, flow_sidx_t sidx)
static flow_sidx_t flowside_lookup(const struct ctx *c, uint8_t proto,
uint8_t pif, const struct flowside *side)
{
- flow_sidx_t sidx;
+ flow_sidx_t sidx, ret;
union flow *flow;
unsigned b;
+ pthread_rwlock_rdlock(&flow_lock);
+
b = flow_hash(c, proto, pif, side) % FLOW_HASH_SIZE;
while ((sidx = flow_hashtab[b], flow = flow_at_sidx(sidx)) &&
!(FLOW_PROTO(&flow->f) == proto &&
@@ -804,7 +868,11 @@ static flow_sidx_t flowside_lookup(const struct ctx *c, uint8_t proto,
flowside_eq(&flow->f.side[sidx.sidei], side)))
b = mod_sub(b, 1, FLOW_HASH_SIZE);
- return flow_hashtab[b];
+ ret = flow_hashtab[b];
+
+ pthread_rwlock_unlock(&flow_lock);
+
+ return ret;
}
/**
@@ -879,8 +947,8 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
struct timespec *timer_run, unsigned int qpair)
{
struct flow_free_cluster *free_head = NULL;
- unsigned *last_next = &flow_first_free;
bool to_free[FLOW_MAX] = { 0 };
+ unsigned *last_next;
bool timer = false;
union flow *flow;
@@ -897,6 +965,9 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
flow_foreach(flow) {
bool closed = false;
+ if (flow->f.qpair != qpair)
+ continue;
+
switch (flow->f.type) {
case FLOW_TYPE_NONE:
assert(false);
@@ -928,6 +999,8 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
}
/* Second step: actually free the flows */
+ pthread_rwlock_wrlock(&flow_lock);
+ last_next = &flow_first_free;
flow_foreach_slot(flow) {
switch (flow->f.state) {
case FLOW_STATE_FREE: {
@@ -956,8 +1029,8 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
case FLOW_STATE_INI:
case FLOW_STATE_TGT:
case FLOW_STATE_TYPED:
- /* Incomplete flow at end of cycle */
- assert(false);
+ /* In-progress allocation on another thread */
+ free_head = NULL;
break;
case FLOW_STATE_ACTIVE:
@@ -989,6 +1062,7 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now,
}
*last_next = FLOW_MAX;
+ pthread_rwlock_unlock(&flow_lock);
}
/**
diff --git a/flow_table.h b/flow_table.h
index 3a33eef15f1e..57944b748920 100644
--- a/flow_table.h
+++ b/flow_table.h
@@ -72,7 +72,7 @@ extern union flow flowtab[];
(flow) += (flow)->free.n - 1; \
/* NOLINTNEXTLINE(readability-inconsistent-ifelse-braces) */\
else if ((flow)->f.state != FLOW_STATE_ACTIVE) { \
- flow_err((flow), "Bad flow state during traversal"); \
+ (void)0; /* Differs from bare continue */ \
continue; \
} else
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread* [PATCH v2 10/10] flow: Add lazy, lock-free flow migration between queue pairs
2026-07-31 16:23 [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers Laurent Vivier
` (8 preceding siblings ...)
2026-07-31 16:23 ` [PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling Laurent Vivier
@ 2026-07-31 16:23 ` Laurent Vivier
9 siblings, 0 replies; 11+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
When the guest steers traffic for an existing flow to a different TX
queue, the flow must migrate to the new queue pair so that socket
events are handled by the correct worker thread.
Use a to_migrate[] array for lazy, lock-free migration:
flow_migrate_mark() records the target qpair from the tap path, and
flow_migrate_epollfd() completes the migration on the next socket
event by moving the fd and updating qpair on the old thread.
TCP timers are migrated the same way in tcp_timer_handler() using
tcp_timer_epoll_add().
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
flow.c | 114 +++++++++++++++++++++++++++++++++++++++++++++++++++++
flow.h | 3 ++
icmp.c | 10 ++++-
tcp.c | 20 ++++++++++
udp.c | 3 ++
udp_flow.c | 4 ++
6 files changed, 152 insertions(+), 2 deletions(-)
diff --git a/flow.c b/flow.c
index 8deca3e3c7f1..3012a07ed680 100644
--- a/flow.c
+++ b/flow.c
@@ -162,6 +162,60 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES,
* with flow_alloc(). The first pass (deferred protocol handlers) runs
* without the lock and filters by qpair, so each thread only processes
* its own flows.
+ *
+ * Flow migration between queue pairs
+ * ==================================
+ *
+ * In vhost-user multiqueue mode, each queue pair has its own worker thread.
+ * A flow is assigned to a queue pair (flow->f.qpair) and its socket is
+ * registered on that queue pair's epollfd. The worker thread for that
+ * queue pair is responsible for processing events and running deferred
+ * handlers (flow_defer_handler) for its own flows.
+ *
+ * When the guest steers traffic for an existing flow to a different TX
+ * queue (e.g. due to RSS), the packet arrives on a queue pair that does
+ * not own the flow. The flow must be migrated to the new queue pair so
+ * that future socket events are handled by the correct worker thread.
+ *
+ * Migration is lazy and lock-free, using a two-step process:
+ *
+ * 1. Mark (tap path, new queue pair's thread):
+ * A packet arrives on queue pair N for a flow currently owned by
+ * queue pair M. flow_migrate_mark() records the target:
+ * to_migrate[flow_index] = N
+ * The flow's qpair field and epollfd registration are unchanged.
+ *
+ * 2. Migrate (socket event, old queue pair M's thread):
+ * When the next event fires on the flow's socket, it is delivered
+ * to queue pair M's epollfd (where the socket is still registered).
+ * flow_migrate_epollfd() compares to_migrate[flow_index] against
+ * flow->f.qpair. If they differ, it performs the migration:
+ * - Removes the socket from queue pair M's epollfd
+ * - Sets flow->f.qpair = N
+ * - Adds the socket to queue pair N's epollfd
+ * The event handler returns true, skipping normal processing for
+ * this event.
+ *
+ * This design has several important properties:
+ *
+ * - No locks or barriers are needed. The old thread performs the
+ * migration itself, so there is no concurrent access to the flow
+ * during the transition.
+ *
+ * - flow_defer_handler() filters flows by flow->f.qpair. Since
+ * qpair is only updated in step 2 (after the epollfd move), the
+ * old queue pair's defer handler continues to process the flow
+ * until migration completes. The new queue pair's defer handler
+ * will not see the flow until the socket is on its epollfd.
+ *
+ * - Between steps 1 and 2, the flow is fully functional on the old
+ * queue pair. The only effect of step 1 is setting the target in
+ * to_migrate[], which has no impact on event processing.
+ *
+ * - If multiple marks occur before the migration completes (e.g. the
+ * guest changes queues again), to_migrate[] simply records the
+ * latest target. Step 2 will migrate to whatever target is current
+ * when the next socket event fires.
*/
unsigned flow_first_free;
@@ -181,6 +235,9 @@ static_assert(ARRAY_SIZE(flow_hashtab) >= 2 * FLOW_MAX,
static pthread_rwlock_t flow_lock = PTHREAD_RWLOCK_INITIALIZER;
+static unsigned int to_migrate[FLOW_MAX];
+static pthread_mutex_t migrate_lock = PTHREAD_MUTEX_INITIALIZER;
+
/** flowside_from_af() - Initialise flowside from addresses
* @side: flowside to initialise
* @af: Address family (AF_INET or AF_INET6)
@@ -443,6 +500,62 @@ static void flow_setqp(struct flow_common *f, unsigned int qpair)
f->qpair = qpair;
}
+/**
+ * flow_migrate_epollfd() - Migrate a flow to a different epollfd if pending
+ * @f: Flow to check for pending migration
+ * @qpair: Queue pair of the current thread (where event fired)
+ * @events: epoll events to watch for
+ * @ref: epoll reference
+ *
+ * Return: true if the flow has been migrated to a new epollfd, false otherwise
+ */
+bool flow_migrate_epollfd(struct flow_common *f, unsigned int qpair,
+ uint32_t events, union epoll_ref ref)
+{
+ unsigned int target;
+ bool ret = false;
+
+ assert(qpair < FLOW_QPAIR_SIZE);
+
+ pthread_mutex_lock(&migrate_lock);
+ target = to_migrate[flow_idx(f)];
+ if (target == f->qpair)
+ goto out;
+
+ flow_trace((union flow *)f,
+ "migrating from qpair %d to %d", qpair, target);
+
+ epoll_del(qpair_to_fd[qpair], ref.fd);
+ flow_setqp(f, target);
+ flow_epoll_set(f, EPOLL_CTL_ADD, events, ref.fd, ref.flowside.sidei);
+ ret = true;
+
+out:
+ pthread_mutex_unlock(&migrate_lock);
+ return ret;
+}
+
+/**
+ * flow_migrate_mark() - Mark a flow for migration to a different qpair
+ * @f: Flow to migrate
+ * @qpair: Target queue pair
+ */
+bool flow_migrate_mark(const struct flow_common *f, unsigned int qpair)
+{
+ bool ret = false;
+
+ pthread_mutex_lock(&migrate_lock);
+ if (f->qpair == qpair)
+ goto out;
+
+ to_migrate[flow_idx(f)] = qpair;
+ ret = true;
+
+out:
+ pthread_mutex_unlock(&migrate_lock);
+ return ret;
+}
+
/**
* flow_initiate_() - Move flow to INI, setting pif[INISIDE]
* @flow: Flow to change state
@@ -669,6 +782,7 @@ union flow *flow_alloc(unsigned int qpair)
flow_new_entry = flow;
memset(flow, 0, sizeof(*flow));
+ to_migrate[FLOW_IDX(flow)] = qpair;
flow_setqp(&flow->f, qpair);
flow_set_state(&flow->f, FLOW_STATE_NEW);
diff --git a/flow.h b/flow.h
index ac1d3897ca98..c31a51a9cc96 100644
--- a/flow.h
+++ b/flow.h
@@ -270,6 +270,9 @@ void flow_init(const struct ctx *c);
int flow_epollfd(const struct flow_common *f);
int flow_epoll_set(const struct flow_common *f, int command, uint32_t events,
int fd, unsigned int sidei);
+bool flow_migrate_epollfd(struct flow_common *f, unsigned int qpair,
+ uint32_t events, union epoll_ref ref);
+bool flow_migrate_mark(const struct flow_common *f, unsigned int qpair);
void flow_defer_handler(const struct ctx *c, const struct timespec *now,
struct timespec *timer_run, unsigned int qpair);
diff --git a/icmp.c b/icmp.c
index 46c5d925600a..802914d329c8 100644
--- a/icmp.c
+++ b/icmp.c
@@ -85,6 +85,9 @@ void icmp_sock_handler(const struct ctx *c, union epoll_ref ref,
assert(pingf);
+ if (flow_migrate_epollfd(&pingf->f, qpair, EPOLLIN, ref))
+ return;
+
n = recvfrom(ref.fd, buf, sizeof(buf), 0, &sr.sa, &sl);
if (n < 0) {
flow_perror_ratelimit(pingf, now, "recvfrom() error");
@@ -307,10 +310,13 @@ int icmp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif,
flow = flow_at_sidx(flow_lookup_af(c, proto, PIF_TAP,
af, saddr, daddr, id, id));
- if (flow)
+ if (flow) {
pingf = &flow->ping;
- else if (!(pingf = icmp_ping_new(c, qpair, af, id, saddr, daddr, now)))
+ if (flow_migrate_mark(&pingf->f, qpair))
+ return 1;
+ } else if (!(pingf = icmp_ping_new(c, qpair, af, id, saddr, daddr, now))) {
return 1;
+ }
tgt = &pingf->f.side[TGTSIDE];
diff --git a/tcp.c b/tcp.c
index 88caf5d8968b..2767a8494107 100644
--- a/tcp.c
+++ b/tcp.c
@@ -2382,6 +2382,9 @@ int tcp_tap_handler(const struct ctx *c, unsigned int qpair, uint8_t pif,
assert(pif_at_sidx(sidx) == PIF_TAP);
conn = &flow->tcp;
+ if (flow_migrate_mark(&conn->f, qpair))
+ return 1;
+
flow_trace(conn, "packet length %zu from tap", l4len);
if (th->rst) {
@@ -2717,6 +2720,17 @@ void tcp_timer_handler(const struct ctx *c, union epoll_ref ref,
assert(!c->no_tcp);
assert(conn->f.type == FLOW_TCP);
+ if (conn->f.qpair != qpair) {
+ int old_epollfd = qpair_to_fd[qpair];
+
+ epoll_del(old_epollfd, conn->timer);
+ if (tcp_timer_epoll_add(conn, conn->timer, now) < 0) {
+ close(conn->timer);
+ conn->timer = -1;
+ }
+ return;
+ }
+
/* We don't reset timers on ~ACK_FROM_TAP_DUE, ~ACK_TO_TAP_DUE. If the
* timer is currently armed, this event came from a previous setting,
* and we just set the timer to a new point in the future: discard it.
@@ -2788,6 +2802,12 @@ void tcp_sock_handler(const struct ctx *c, union epoll_ref ref,
assert(!c->no_tcp);
assert(pif_at_sidx(ref.flowside) != PIF_TAP);
+ if (flow_migrate_epollfd(&conn->f, qpair,
+ tcp_conn_epoll_events(conn->events,
+ conn->flags),
+ ref))
+ return;
+
if (conn->events == CLOSED)
return;
diff --git a/udp.c b/udp.c
index a3321289bb53..6c45eb1a00d5 100644
--- a/udp.c
+++ b/udp.c
@@ -956,6 +956,9 @@ void udp_sock_handler(const struct ctx *c, union epoll_ref ref, uint32_t events,
assert(!c->no_udp && uflow);
+ if (flow_migrate_epollfd(&uflow->f, qpair, EPOLLIN, ref))
+ return;
+
if (events & EPOLLERR) {
if (udp_sock_errs(c, ref.fd, ref.flowside,
PIF_NONE, 0, now, qpair) < 0) {
diff --git a/udp_flow.c b/udp_flow.c
index 8e985df8398d..8894f8cff6bd 100644
--- a/udp_flow.c
+++ b/udp_flow.c
@@ -231,6 +231,8 @@ flow_sidx_t udp_flow_from_sock(const struct ctx *c, unsigned int qpair,
sidx = flow_lookup_sa(c, IPPROTO_UDP, pif, s_in, dst, port);
if ((uflow = udp_at_sidx(sidx))) {
+ if (flow_migrate_mark(&uflow->f, qpair))
+ return FLOW_SIDX_NONE;
udp_flow_activity(uflow, sidx.sidei, now);
return flow_sidx_opposite(sidx);
}
@@ -292,6 +294,8 @@ flow_sidx_t udp_flow_from_tap(const struct ctx *c, unsigned int qpair,
sidx = flow_lookup_af(c, IPPROTO_UDP, pif, af, saddr, daddr,
srcport, dstport);
if ((uflow = udp_at_sidx(sidx))) {
+ if (flow_migrate_mark(&uflow->f, qpair))
+ return FLOW_SIDX_NONE;
udp_flow_activity(uflow, sidx.sidei, now);
return flow_sidx_opposite(sidx);
}
--
2.54.0
^ permalink raw reply [flat|nested] 11+ messages in thread