* [PATCH v2 00/10] multithreading: Prepare data structures for concurrent queue pair workers
@ 2026-07-31 16:23 Laurent Vivier
2026-07-31 16:23 ` [PATCH v2 01/10] tap: Convert packet pools to per-queue-pair arrays for multiqueue Laurent Vivier
` (9 more replies)
0 siblings, 10 replies; 22+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:23 UTC (permalink / raw)
To: passt-dev; +Cc: 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
^ permalink raw reply [flat|nested] 22+ messages in thread
* [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-08-03 5:41 ` David Gibson
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, 1 reply; 22+ 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] 22+ 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-08-03 5:43 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 03/10] tcp: Make static buffers stack-local " Laurent Vivier
` (7 subsequent siblings)
9 siblings, 1 reply; 22+ 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] 22+ 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-08-03 5:59 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 04/10] udp_vu: Make virtqueue " Laurent Vivier
` (6 subsequent siblings)
9 siblings, 1 reply; 22+ 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] 22+ 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-08-03 6:00 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 05/10] flow: Make flow timer per-caller " Laurent Vivier
` (5 subsequent siblings)
9 siblings, 1 reply; 22+ 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] 22+ 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-08-03 6:11 ` David Gibson
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, 1 reply; 22+ 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] 22+ 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-08-03 6:40 ` David Gibson
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, 1 reply; 22+ 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] 22+ 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-08-03 6:53 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 08/10] tcp: Extract tcp_timer_epoll_add() helper Laurent Vivier
` (2 subsequent siblings)
9 siblings, 1 reply; 22+ 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] 22+ 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-08-03 6:54 ` David Gibson
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, 1 reply; 22+ 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] 22+ 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-08-03 9:58 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 10/10] flow: Add lazy, lock-free flow migration between queue pairs Laurent Vivier
9 siblings, 1 reply; 22+ 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] 22+ 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; 22+ 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] 22+ messages in thread
* Re: [PATCH v2 01/10] tap: Convert packet pools to per-queue-pair arrays for multiqueue
2026-07-31 16:23 ` [PATCH v2 01/10] tap: Convert packet pools to per-queue-pair arrays for multiqueue Laurent Vivier
@ 2026-08-03 5:41 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 5:41 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 9823 bytes --]
On Fri, Jul 31, 2026 at 06:23:20PM +0200, Laurent Vivier wrote:
> 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>
Reviewed-by: David Gibson <david@gibson.dropbear.id.au>
> ---
> 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
>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety
2026-07-31 16:23 ` [PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety Laurent Vivier
@ 2026-08-03 5:43 ` David Gibson
2026-08-03 5:55 ` David Gibson
0 siblings, 1 reply; 22+ messages in thread
From: David Gibson @ 2026-08-03 5:43 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 4531 bytes --]
On Fri, Jul 31, 2026 at 06:23:21PM +0200, Laurent Vivier wrote:
> 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>
Reviewed-by: David Gibson <david@gibson.dropbear.id.au>
> ---
> 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
>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety
2026-08-03 5:43 ` David Gibson
@ 2026-08-03 5:55 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 5:55 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 1340 bytes --]
On Mon, Aug 03, 2026 at 03:43:15PM +1000, David Gibson wrote:
> On Fri, Jul 31, 2026 at 06:23:21PM +0200, Laurent Vivier wrote:
> > 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>
>
> Reviewed-by: David Gibson <david@gibson.dropbear.id.au>
Although.. it now occurred to me on both this and the previous patch.
IIRC, the effective lifetime of these pool structures is just from
slurping the packets off the tap interface to processing them through
tap[46]_handler().
So, rather than having per queue arrays, would it make more sense to
put these on the stack as well?
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 03/10] tcp: Make static buffers stack-local for thread safety
2026-07-31 16:23 ` [PATCH v2 03/10] tcp: Make static buffers stack-local " Laurent Vivier
@ 2026-08-03 5:59 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 5:59 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 6468 bytes --]
On Fri, Jul 31, 2026 at 06:23:22PM +0200, Laurent Vivier wrote:
> 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>
Reviewed-by: David Gibson <david@gibson.dropbear.id.au>
> ---
> 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
>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 04/10] udp_vu: Make virtqueue buffers stack-local for thread safety
2026-07-31 16:23 ` [PATCH v2 04/10] udp_vu: Make virtqueue " Laurent Vivier
@ 2026-08-03 6:00 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 6:00 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 1588 bytes --]
On Fri, Jul 31, 2026 at 06:23:23PM +0200, Laurent Vivier wrote:
> 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>
Reviewed-by: David Gibson <david@gibson.dropbear.id.au>
> ---
> 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
>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 05/10] flow: Make flow timer per-caller for thread safety
2026-07-31 16:23 ` [PATCH v2 05/10] flow: Make flow timer per-caller " Laurent Vivier
@ 2026-08-03 6:11 ` David Gibson
2026-08-03 6:35 ` David Gibson
0 siblings, 1 reply; 22+ messages in thread
From: David Gibson @ 2026-08-03 6:11 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 5094 bytes --]
On Fri, Jul 31, 2026 at 06:23:24PM +0200, Laurent Vivier wrote:
> 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>
One small misgiving about this, although I doubt it can cause a real
problem. Becayse this timer processing now doesn't happen all at
once, at each queue is on a separate timer cycle, it's theoretically
possible for a flow to change qpair every FLOW_TIMER_INTERVAL and
thereby indefinitely avoid having the timer running on it.
It's pretty tricky to do (depending on how close in real time each
queue's timers end up running), and I don't *think* that can do
anything terribly bad (maybe delay cleanup). But since the guest
could in principle control the qpair placements to specifically
trigger this, it just makes me think it warrants a good close look to
make sure that something bad can't be triggered that way.
> ---
> 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
>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 05/10] flow: Make flow timer per-caller for thread safety
2026-08-03 6:11 ` David Gibson
@ 2026-08-03 6:35 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 6:35 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 2132 bytes --]
On Mon, Aug 03, 2026 at 04:11:24PM +1000, David Gibson wrote:
> On Fri, Jul 31, 2026 at 06:23:24PM +0200, Laurent Vivier wrote:
> > 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>
>
> One small misgiving about this, although I doubt it can cause a real
> problem. Becayse this timer processing now doesn't happen all at
> once, at each queue is on a separate timer cycle, it's theoretically
> possible for a flow to change qpair every FLOW_TIMER_INTERVAL and
> thereby indefinitely avoid having the timer running on it.
>
> It's pretty tricky to do (depending on how close in real time each
> queue's timers end up running), and I don't *think* that can do
> anything terribly bad (maybe delay cleanup). But since the guest
> could in principle control the qpair placements to specifically
> trigger this, it just makes me think it warrants a good close look to
> make sure that something bad can't be triggered that way.
Looking at the next patch, I realised there's a much easier to hit
concern here. Kind of the opposite of the one above, if there's a
migration at the wrong moment, we might run the timer twice on a flow
in a much smaller interval than usual. AFAICT, the worst that can do
- at least with our current flow_defer() and flow_timer() functions is
a little bit of unnecessary work, so I think it's ok. That's because
none of them unconditionally do anything damaging - they always check
an independent timestamp.
However, it does add another wrinkle to what is and isn't safe in the
defer and timer functions: they cannot rely on being called at most
every INTERVAL seconds. We should probably document
that... somewhere.
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 06/10] tcp: Make TCP timer state per-caller and guard global tasks
2026-07-31 16:23 ` [PATCH v2 06/10] tcp: Make TCP timer state per-caller and guard global tasks Laurent Vivier
@ 2026-08-03 6:40 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 6:40 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 10090 bytes --]
On Fri, Jul 31, 2026 at 06:23:25PM +0200, Laurent Vivier wrote:
> 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>
This patch has slightly different variants of the conerns I raised on
the previous one. Unlike the flow_defer and flow_timer functions,
the keepalive and inactivity handling absolutely *does* rely on not
being called for the same flow more often than once every INTERVAL s.
If it happens twice quickly due to a migration we could incorrectly
time out a flow much earlier than it shuold be.
That can be fixed pretty easily by setting conn->tap_inactive and
conn->inactive to false on migration (I haven't looked ahead to see if
you already do that). That would have the cost of maybe sometimes
delaying the keepalive or inactivity handling longer than it should
be. Again, probably harmless in practice (although as noted on the
previous patch, in theory the delay could be indefinite, if the flow
keeps migrating).
> ---
> 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
>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 07/10] tcp: Protect init socket pools with mutex for thread safety
2026-07-31 16:23 ` [PATCH v2 07/10] tcp: Protect init socket pools with mutex for thread safety Laurent Vivier
@ 2026-08-03 6:53 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 6:53 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 3635 bytes --]
On Fri, Jul 31, 2026 at 06:23:26PM +0200, Laurent Vivier wrote:
> 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.
I'm guessing you're going with a mutex rather than a per-thread socket
pool for simplicity? That might be the right choice, but I do wonder
a bit about it. The pool exists to avoid the latency of creating a
new socket for a new connection. If the latency of taking the lock
exceeds that of creating a socket, there's no longer any point to the
pool. In the unconstest case, that's almost certainly not the case -
a happy path futex() lock should be be much faster than a syscall. If
the lock _is_ contested, I suspect opening a socket directly might win
- at least for host ns sockets. For guest ns sockets, the latency is
higher because we need vfork()/setns()/etc.
Then again... taking a socket from the pool also involves writing the
pool, which potentially means a cacheline pingpong. That might incur
a pretty substantial latency.
In theory we could trylock() and open a socket directly if we don't
get the lock immediately, but at that point it would probably be
simpler to have per-thread socket pools anyway.
I guess, since there is no meaningful shared state in the socket pool,
a per-thread pool seems like the more natural approach to me.
>
> 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
>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 08/10] tcp: Extract tcp_timer_epoll_add() helper
2026-07-31 16:23 ` [PATCH v2 08/10] tcp: Extract tcp_timer_epoll_add() helper Laurent Vivier
@ 2026-08-03 6:54 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 6:54 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 501 bytes --]
On Fri, Jul 31, 2026 at 06:23:27PM +0200, Laurent Vivier wrote:
> 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>
Reviewed-by: David Gibson <david@gibson.dropbear.id.au>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
* Re: [PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling
2026-07-31 16:23 ` [PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling Laurent Vivier
@ 2026-08-03 9:58 ` David Gibson
0 siblings, 0 replies; 22+ messages in thread
From: David Gibson @ 2026-08-03 9:58 UTC (permalink / raw)
To: Laurent Vivier; +Cc: passt-dev
[-- Attachment #1: Type: text/plain, Size: 12356 bytes --]
On Fri, Jul 31, 2026 at 06:23:28PM +0200, Laurent Vivier wrote:
> 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.
Oof, unfortunately, I'm not sure this is quite enough. The free list
management was bespoke designed for our single threaded use case, and
I'm not sure the current approach quite works any more with
multithreading. More below.
>
> 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.
The way free cluster consolidation is done is because it was near-free
to do so, because we were already doing a regular linear scan through
the entire table. Now that each defer handler is only handling a
subset of the entries it's not so clear that this approach makes sense
any more. At the very least with free consolidation passes from every
thread, the effective cost might be significantly higher.
> */
>
> 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);
Alas, this is not safe. Restoring flow_first_free to the flow index
relied on the fact that when single threaded, there could be no other
entries allocated between flow_alloc() and FLOW_ACTIVATE(). Because
we always allocated the first free entry, if we cancel it will become
the first free entry again.
But with the lock dropped between flow_alloc() and
flow_alloc_cancel(), this could be restoring an out of date snapshot
of flow_first_free - it might no longer be the actual first free entry
any more if another thread freed something earlier in the table.
Honestly, I'm not sure if flow_alloc_cancel() makes sense any more in
a multi-threaded context. The rules about when it was and wasn't ok
were already pretty confusing, so maybe we should go through the full
flow close path even for flows cancelled very early.
> 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);
It might not be necessary to move this under the lock. This reads the
contents of the individual flow (which IIUC is not covered by the
flow_lock), but doesn't depend on any of the shared state data.
> - 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;
As with insert it might be worth using flow_sidx_hash() and
flow_hash_probe_() separately here, so that the flow_sidx_hash() can
move out of the critical section.
> + 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;
As above, I think the hash can go outside the lock.
> 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;
I think these can all now be additional labels on the
FLOW_STATE_ACTIVE case. At which point, this no longer really wants
to be a switch, but rather just an if (state == FLOW_STATE_FREE) else.
> 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
>
--
David Gibson (he or they) | I'll have my music baroque, and my code
david AT gibson.dropbear.id.au | minimalist, thank you, not the other way
| around.
http://www.ozlabs.org/~dgibson
[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]
^ permalink raw reply [flat|nested] 22+ messages in thread
end of thread, other threads:[~2026-08-03 9:59 UTC | newest]
Thread overview: 22+ messages (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
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-08-03 5:41 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 02/10] tap: Make L4 sequence pools per-qpair for thread safety Laurent Vivier
2026-08-03 5:43 ` David Gibson
2026-08-03 5:55 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 03/10] tcp: Make static buffers stack-local " Laurent Vivier
2026-08-03 5:59 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 04/10] udp_vu: Make virtqueue " Laurent Vivier
2026-08-03 6:00 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 05/10] flow: Make flow timer per-caller " Laurent Vivier
2026-08-03 6:11 ` David Gibson
2026-08-03 6:35 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 06/10] tcp: Make TCP timer state per-caller and guard global tasks Laurent Vivier
2026-08-03 6:40 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 07/10] tcp: Protect init socket pools with mutex for thread safety Laurent Vivier
2026-08-03 6:53 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 08/10] tcp: Extract tcp_timer_epoll_add() helper Laurent Vivier
2026-08-03 6:54 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 09/10] flow: Add locking, per-qpair filtering, and intermediate state handling Laurent Vivier
2026-08-03 9:58 ` David Gibson
2026-07-31 16:23 ` [PATCH v2 10/10] flow: Add lazy, lock-free flow migration between queue pairs Laurent Vivier
Code repositories for project(s) associated with this public inbox
https://passt.top/passt
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for IMAP folder(s).