[PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers

[PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers

[Laurent Vivier]

This series makes the passt data structures safe for concurrent access
by multiple worker threads, each handling a different queue pair.  It
builds on the multiqueue series which threads the queue pair parameter
through the network stack.

The changes fall into four categories:

1. Per-queue-pair data isolation:
   - Convert packet pools (pool_tap4, pool_tap6) to per-queue-pair arrays
   - Convert L4 sequence batching arrays (tap4_l4, tap6_l4) similarly
   - Move flow and TCP timer state out of global/context structures into
     per-caller parameters

2. Stack-local buffers:
   - Move static iovec and virtqueue buffers onto the stack in tcp.c,
     tcp_vu.c, and udp_vu.c so each thread operates on its own copy

3. Locking for remaining shared state:
   - pthread_mutex on pre-opened socket pools (init_sock_pool4/6)
   - pthread_rwlock on the flow table, hash table, and free list, with
     read locks for lookups and write locks for mutations
   - _Thread_local for flow_new_entry so each thread independently
     tracks its in-progress allocation
   - Per-qpair filtering in flow_defer_handler() so each worker only
     processes its own flows

4. Lock-free flow migration between queue pairs:
   - When guest steers traffic to a different TX queue, flows migrate
     lazily using a to_migrate[] array
   - flow_migrate_mark() records target qpair from tap path (write-once,
     no lock needed)
   - flow_migrate_epollfd() completes migration on next socket/timer
     event by moving fd and updating qpair on the old thread
   - TCP timers migrate the same way using new tcp_timer_epoll_add()
     helper

Global operations that don't need per-queue-pair handling
(tcp_payload_flush, socket pool refills) are guarded to run only on
queue pair 0.

v2:
- Rebase on latest master and multiqueue v6
- Split v1's "flow: Add mutex and per-qpair filtering..." into two patches:
  - "flow: Add locking, per-qpair filtering, and intermediate state handling"
    (locking only)
  - "flow: Add lazy, lock-free flow migration between queue pairs" (migration
    logic separated)
- Changed from simple mutex to pthread_rwlock for flow table (read locks for
  lookups, write locks for mutations and free-list rebuild)
- Updated flow_foreach() and flow_defer_handler() to handle intermediate flow
  states (NEW, INI, TGT, TYPED) that can be observed between flow_alloc() and
  FLOW_ACTIVATE() when lock is released
- New patch: "tcp: Extract tcp_timer_epoll_add() helper" - factors out epoll
  registration for timer migration support
- flow_migrate_epollfd() now uses tcp_timer_epoll_add() for timer migration

v1:
- Initial submission

Based-on: 20260616125130.1324274-1-lvivier@redhat.com

Laurent Vivier (10):
  tap: Convert packet pools to per-queue-pair arrays for multiqueue
  tap: Make L4 sequence pools per-qpair for thread safety
  tcp: Make static buffers stack-local for thread safety
  udp_vu: Make virtqueue buffers stack-local for thread safety
  flow: Make flow timer per-caller for thread safety
  tcp: Make TCP timer state per-caller and guard global tasks
  tcp: Protect init socket pools with mutex for thread safety
  tcp: Extract tcp_timer_epoll_add() helper
  flow: Add locking, per-qpair filtering, and intermediate state
    handling
  flow: Add lazy, lock-free flow migration between queue pairs

 flow.c       | 222 ++++++++++++++++++++++++++++++++++++++++++++++-----
 flow.h       |   5 +-
 flow_table.h |   2 +-
 icmp.c       |  10 ++-
 passt.c      |  37 +++++----
 tap.c        | 109 ++++++++++++++-----------
 tap.h        |   2 +-
 tcp.c        | 108 ++++++++++++++++++-------
 tcp.h        |   9 +--
 tcp_vu.c     |  33 +++++---
 udp.c        |   3 +
 udp_flow.c   |   4 +
 udp_vu.c     |   4 +-
 vu_common.c  |   2 +-
 14 files changed, 409 insertions(+), 141 deletions(-)

-- 
2.54.0

[PATCH v2 01/10] tap: Convert packet pools to per-queue-pair arrays for multiqueue

[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

[PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety

[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

[PATCH v2 03/10] tcp: Make static buffers stack-local for thread safety

[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

[PATCH v2 04/10] udp_vu: Make virtqueue buffers stack-local for thread safety

[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

[PATCH v2 05/10] flow: Make flow timer per-caller for thread safety

[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

[PATCH v2 06/10] tcp: Make TCP timer state per-caller and guard global tasks

[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

[PATCH v2 07/10] tcp: Protect init socket pools with mutex for thread safety

[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

[PATCH v2 08/10] tcp: Extract tcp_timer_epoll_add() helper

[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

[PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling

[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

[PATCH v2 10/10] flow: Add lazy, lock-free flow migration between queue pairs

[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