[PATCH 0/8] multithreading: Prepare data structures for concurrent queue pair workers

newer
[PATCH 1/3] fwd_rule: Parse target...

Laurent Vivier

16 Jun 2026 16 Jun '26

7:10 p.m.

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 three 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 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. Based-on: 20260616125130.1324274-1-lvivier@redhat.com Laurent Vivier (8): 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 flow: Add mutex and per-qpair filtering to flow table operations flow.c | 68 +++++++++++++++++++++++++------- flow.h | 2 +- flow_table.h | 2 +- passt.c | 37 +++++++++-------- tap.c | 109 +++++++++++++++++++++++++++++---------------------- tap.h | 2 +- tcp.c | 64 +++++++++++++++++++----------- tcp.h | 9 +---- tcp_vu.c | 33 ++++++++++------ udp_vu.c | 4 +- vu_common.c | 2 +- 11 files changed, 203 insertions(+), 129 deletions(-) -- 2.54.0

Show replies by date

Laurent Vivier

16 Jun 16 Jun

7:10 p.m.

New subject: [PATCH 1/8] tap: Convert packet pools to per-queue-pair arrays for multiqueue

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 --- 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 5e9c7a1701bf..80912372e216 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 */ @@ -717,12 +721,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++) { size_t l3len, hlen, l4len; struct ethhdr eh_storage; struct iphdr iph_storage; @@ -732,7 +736,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); @@ -804,7 +808,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; } @@ -835,7 +839,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)) { @@ -881,10 +885,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) \ @@ -953,12 +957,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; @@ -970,7 +974,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); @@ -1083,7 +1087,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)) { @@ -1130,19 +1134,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]); } /** @@ -1153,6 +1157,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); } @@ -1187,21 +1192,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; @@ -1239,7 +1246,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 @@ -1322,7 +1329,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); @@ -1591,10 +1598,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 ecb12de372b5..8a1d8f933a5b 100644 --- a/tap.h +++ b/tap.h @@ -123,7 +123,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 6aa1ba768136..2e69c1945055 100644 --- a/vu_common.c +++ b/vu_common.c @@ -178,7 +178,7 @@ static void vu_handle_tx(struct vu_dev *vdev, int index, assert(QPAIR_FROMGUEST_QUEUE(QPAIR_FROM_QUEUE(index)) == (unsigned int)index); - tap_flush_pools(); + tap_flush_pools(QPAIR_FROM_QUEUE(index)); count = 0; out_sg_count = 0; -- 2.54.0

Laurent Vivier

7:10 p.m.

New subject: [PATCH 2/8] tap: Make L4 sequence pools per-qpair for thread safety

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 --- tap.c | 32 +++++++++++++++++++------------- 1 file changed, 19 insertions(+), 13 deletions(-) diff --git a/tap.c b/tap.c index 80912372e216..659df9d560d3 100644 --- a/tap.c +++ b/tap.c @@ -606,7 +606,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 @@ -633,7 +633,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 @@ -841,7 +841,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; @@ -849,8 +849,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); } @@ -862,7 +862,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; @@ -1089,7 +1089,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; @@ -1097,8 +1097,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); } @@ -1110,7 +1110,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; @@ -1607,9 +1607,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

Laurent Vivier

7:10 p.m.

New subject: [PATCH 4/8] udp_vu: Make virtqueue buffers stack-local for thread safety

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 --- udp_vu.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/udp_vu.c b/udp_vu.c index 864e7a99b8d9..342673dc7e6d 100644 --- a/udp_vu.c +++ b/udp_vu.c @@ -146,8 +146,8 @@ 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]; - static struct iovec iov_vu[VIRTQUEUE_MAX_SIZE]; + struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE]; + struct iovec iov_vu[VIRTQUEUE_MAX_SIZE]; int rx_queue = QPAIR_TOGUEST_QUEUE(qpair); struct vu_dev *vdev = c->vdev; struct vu_virtq *vq = &vdev->vq[rx_queue]; -- 2.54.0

Laurent Vivier

7:10 p.m.

New subject: [PATCH 5/8] flow: Make flow timer per-caller for thread safety

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 --- flow.c | 11 +++++------ flow.h | 2 +- passt.c | 8 +++++--- 3 files changed, 11 insertions(+), 10 deletions(-) diff --git a/flow.c b/flow.c index 787a7139cfc1..08c7620c7b0f 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) @@ -898,9 +895,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; @@ -908,9 +907,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 53e0408a9ee5..10634e64a7fc 100644 --- a/flow.h +++ b/flow.h @@ -280,7 +280,7 @@ void flow_migrate(struct flow_common *f, unsigned int qpair, uint32_t events, (flow_migrate(&(flow_)->f, qpair_, events_, fd_, 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 3afc59b19120..bebc2b99f523 100644 --- a/passt.c +++ b/passt.c @@ -98,15 +98,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) @@ -221,6 +222,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; @@ -304,7 +306,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); } -- 2.54.0

Laurent Vivier

7:10 p.m.

New subject: [PATCH 3/8] tcp: Make static buffers stack-local for thread safety

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 --- tcp.c | 3 +-- tcp_vu.c | 33 ++++++++++++++++++++------------- 2 files changed, 21 insertions(+), 15 deletions(-) diff --git a/tcp.c b/tcp.c index 1549e14adaf4..f4fe866ba7c3 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]; @@ -1900,6 +1898,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 4f76f599156f..9270ece43d17 100644 --- a/tcp_vu.c +++ b/tcp_vu.c @@ -35,9 +35,6 @@ #include "vu_common.h" #include -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 @@ -224,6 +221,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) * @@ -233,9 +233,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); @@ -252,16 +255,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) @@ -327,7 +330,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) @@ -433,6 +437,9 @@ static void tcp_vu_prepare(const struct ctx *c, struct tcp_tap_conn *conn, int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn, 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 rx_queue = QPAIR_TOGUEST_QUEUE(qpair); struct vu_dev *vdev = c->vdev; @@ -477,7 +484,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, qpair); -- 2.54.0

Laurent Vivier

7:10 p.m.

New subject: [PATCH 7/8] tcp: Protect init socket pools with mutex for thread safety

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 --- tcp.c | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/tcp.c b/tcp.c index 955012355d69..019340c1c348 100644 --- a/tcp.c +++ b/tcp.c @@ -293,6 +293,7 @@ #include #include #include +#include #include #include @@ -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 @@ -1568,7 +1570,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 @@ -2857,6 +2863,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) @@ -2869,6 +2876,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

Laurent Vivier

7:10 p.m.

New subject: [PATCH 6/8] tcp: Make TCP timer state per-caller and guard global tasks

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 --- 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 bebc2b99f523..ca5973e17317 100644 --- a/passt.c +++ b/passt.c @@ -96,16 +96,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); @@ -130,16 +137,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 @@ -221,8 +218,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; @@ -306,7 +304,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); } @@ -433,8 +432,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 f4fe866ba7c3..955012355d69 100644 --- a/tcp.c +++ b/tcp.c @@ -2988,17 +2988,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; @@ -3021,18 +3023,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; @@ -3054,27 +3058,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); + if (qpair == 0) + tcp_payload_flush(c); - 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 490f1b140e44..64c75ba481bd 100644 --- a/tcp.h +++ b/tcp.h @@ -32,7 +32,8 @@ int tcp_listen(const struct ctx *c, uint8_t pif, unsigned rule, const union inany_addr *addr, const char *ifname, in_port_t port); 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); @@ -42,24 +43,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

Laurent Vivier

7:10 p.m.

New subject: [PATCH 8/8] flow: Add mutex and per-qpair filtering to flow table operations

The flow table free list, hash table, and flow_new_entry are global shared state accessed from multiple threads. Protect flow_alloc(), flow_alloc_cancel(), flow_hash_insert(), flow_hash_remove(), and the free list rebuild in flow_defer_handler() with a pthread_rwlock_t: writers for mutations, readers for lookups. Make flow_new_entry _Thread_local so each thread independently tracks its own in-progress allocation. 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() and flow_defer_handler() accordingly: skip these entries silently rather than treating them as errors, and break free-list cluster merging across them. Filter flow_defer_handler()'s first loop by qpair, so each thread only processes its own flows. Signed-off-by: Laurent Vivier --- flow.c | 59 +++++++++++++++++++++++++++++++++++++++++++--------- flow_table.h | 2 +- 2 files changed, 50 insertions(+), 11 deletions(-) diff --git a/flow.c b/flow.c index 08c7620c7b0f..149360c3ec87 100644 --- a/flow.c +++ b/flow.c @@ -12,6 +12,8 @@ #include #include +#include + #include "util.h" #include "ip.h" #include "passt.h" @@ -129,7 +131,7 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES, 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 +144,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) @@ -616,12 +620,18 @@ void flow_activate(struct flow_common *f) */ union flow *flow_alloc(void) { - 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); @@ -650,6 +660,8 @@ union flow *flow_alloc(void) memset(flow, 0, sizeof(*flow)); flow_set_state(&flow->f, FLOW_STATE_NEW); + pthread_rwlock_unlock(&flow_lock); + return flow; } @@ -661,6 +673,8 @@ union flow *flow_alloc(void) */ 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 || @@ -678,6 +692,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); } /** @@ -763,9 +779,13 @@ 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); + unsigned b; + pthread_rwlock_wrlock(&flow_lock); + 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); @@ -779,10 +799,16 @@ 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; - if (!flow_sidx_valid(flow_hashtab[b])) + pthread_rwlock_wrlock(&flow_lock); + + b = flow_hash_probe(c, sidx); + + 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); @@ -802,6 +828,8 @@ void flow_hash_remove(const struct ctx *c, flow_sidx_t sidx) } flow_hashtab[b] = FLOW_SIDX_NONE; + + pthread_rwlock_unlock(&flow_lock); } /** @@ -816,10 +844,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 && @@ -827,7 +857,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; } /** @@ -920,6 +954,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); @@ -951,6 +988,7 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now, } /* Second step: actually free the flows */ + pthread_rwlock_wrlock(&flow_lock); flow_foreach_slot(flow) { switch (flow->f.state) { case FLOW_STATE_FREE: { @@ -979,8 +1017,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: @@ -1012,6 +1050,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 e4ff6f73c35c..f2545390205a 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

29 Jun 29 Jun

11:59 a.m.

New subject: [PATCH 1/8] tap: Convert packet pools to per-queue-pair arrays for multiqueue

On Tue, Jun 16, 2026 at 07:10:45PM +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

Generally LGTM, a few notes below.

...

--- 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 5e9c7a1701bf..80912372e216 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];

We probably want to introduce an explicit cacheline alignment - otherwise the last part of one array element could occupy the same cacheline as the first part of the next element. Since those will be used more or less exclusively by different threads, it could be a pretty bad case for cacheline pingponging. I mean, probably not, because TAP_MSGS_* is large enough that we'll rarely be using the last pieces of each entry, but still. Also this is the only place PACKET_POOL_NOINIT() was used, so the macro can be removed too. A possible longer term idea: we could consider restructuring things so that the pool sizes aren't compile time fixed: for the non-vu cases we'd assign the whole iovec array to the single pool that's in use. For vu, we'd split the iovec array between the number of queues actually in use.

...

#define TAP_SEQS 128 /* Different L4 tuples in one batch */

@@ -717,12 +721,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)

Here and elsewhere, I'd kind of prefer to see a local, rather than repeating pool_tapX[qpair] a bunch of times.

...

+ 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++) { size_t l3len, hlen, l4len; struct ethhdr eh_storage; struct iphdr iph_storage; @@ -732,7 +736,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); @@ -804,7 +808,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; } @@ -835,7 +839,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)) { @@ -881,10 +885,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) \ @@ -953,12 +957,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; @@ -970,7 +974,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); @@ -1083,7 +1087,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)) { @@ -1130,19 +1134,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]); }

/** @@ -1153,6 +1157,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); } @@ -1187,21 +1192,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; @@ -1239,7 +1246,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 @@ -1322,7 +1329,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); @@ -1591,10 +1598,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 ecb12de372b5..8a1d8f933a5b 100644 --- a/tap.h +++ b/tap.h @@ -123,7 +123,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 6aa1ba768136..2e69c1945055 100644 --- a/vu_common.c +++ b/vu_common.c @@ -178,7 +178,7 @@ static void vu_handle_tx(struct vu_dev *vdev, int index, assert(QPAIR_FROMGUEST_QUEUE(QPAIR_FROM_QUEUE(index)) == (unsigned int)index);

- tap_flush_pools(); + tap_flush_pools(QPAIR_FROM_QUEUE(index));

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

David Gibson

2 Jul 2 Jul

4:37 a.m.

New subject: [PATCH 4/8] udp_vu: Make virtqueue buffers stack-local for thread safety

On Tue, Jun 16, 2026 at 07:10:48PM +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

Reviewed-by: David Gibson I generally dislike static locals anyway, though they have their uses. I'd be happy to see this applied independent of the rest of the series.

...

--- udp_vu.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/udp_vu.c b/udp_vu.c index 864e7a99b8d9..342673dc7e6d 100644 --- a/udp_vu.c +++ b/udp_vu.c @@ -146,8 +146,8 @@ 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]; - static struct iovec iov_vu[VIRTQUEUE_MAX_SIZE]; + struct vu_virtq_element elem[VIRTQUEUE_MAX_SIZE]; + struct iovec iov_vu[VIRTQUEUE_MAX_SIZE];

Nit, this needs a little re-ordering to preserve the reverse christmas tree.

...

int rx_queue = QPAIR_TOGUEST_QUEUE(qpair); struct vu_dev *vdev = c->vdev; struct vu_virtq *vq = &vdev->vq[rx_queue]; -- 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

David Gibson

4:32 a.m.

New subject: [PATCH 3/8] tcp: Make static buffers stack-local for thread safety

On Tue, Jun 16, 2026 at 07:10:47PM +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

Reviewed-by: David Gibson Regardless of the thread implications, I think this is nicer than the global just used by one function. I'd be happy to see this applied independent of the rest of the series. Some tiny nits noted below, though:

...

--- tcp.c | 3 +-- tcp_vu.c | 33 ++++++++++++++++++++------------- 2 files changed, 21 insertions(+), 15 deletions(-)

diff --git a/tcp.c b/tcp.c index 1549e14adaf4..f4fe866ba7c3 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]; @@ -1900,6 +1898,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 4f76f599156f..9270ece43d17 100644 --- a/tcp_vu.c +++ b/tcp_vu.c @@ -35,9 +35,6 @@ #include "vu_common.h" #include

-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 @@ -224,6 +221,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) * @@ -233,9 +233,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); @@ -252,16 +255,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) @@ -327,7 +330,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,

Nit: Unrelated whitespace change

...

MAX(hdrlen + ret, VNET_HLEN + ETH_ZLEN), NULL); if (cnt < (unsigned)f->num_iovec) @@ -433,6 +437,9 @@ static void tcp_vu_prepare(const struct ctx *c, struct tcp_tap_conn *conn, int tcp_vu_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn, 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];

These aren't ordered according to our reverse-christmas tree convention.

...

uint32_t wnd_scaled = conn->wnd_from_tap << conn->ws_from_tap; int rx_queue = QPAIR_TOGUEST_QUEUE(qpair); struct vu_dev *vdev = c->vdev; @@ -477,7 +484,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, 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

David Gibson

4:27 a.m.

New subject: [PATCH 2/8] tap: Make L4 sequence pools per-qpair for thread safety

On Tue, Jun 16, 2026 at 07:10:46PM +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

Generally LGTM. Couple of nits and one broader design thought.

...

--- tap.c | 32 +++++++++++++++++++------------- 1 file changed, 19 insertions(+), 13 deletions(-)

diff --git a/tap.c b/tap.c index 80912372e216..659df9d560d3 100644 --- a/tap.c +++ b/tap.c @@ -606,7 +606,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 */];

Nit: VHOST_USER_MAX_VQS is being used often enough for sizing data structures that we might want a new #define for it. As in 1/8, I think an explicit cacheline alignment would be a good idea here too.

...

/** * struct l4_seq6_t - Message sequence for one protocol handler call, IPv6 @@ -633,7 +633,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 */];

Ditto. Also... maybe this would be more trouble for a first pass. But we're introducing a bunch of parallel per-qpair arrays. I'm wondering if it might be nicer to instead gather all the per-qpair structures together into a struct qpair_ctx or struct qpair_tap_ctx. Couple of advantages I can see (neither of which are must haves on their own): * Clearer to describe the locking semantics ("none, because it's thread private") for a single big structure than parallel slices of a whole bunch of other structures * Might give better memory locality for each thread

...

/** * tap_packet_debug() - Print debug message for packet(s) from guest/tap @@ -841,7 +841,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; @@ -849,8 +849,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); } @@ -862,7 +862,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;

@@ -1089,7 +1089,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; @@ -1097,8 +1097,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); } @@ -1110,7 +1110,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;

@@ -1607,9 +1607,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

David Gibson

4:49 a.m.

New subject: [PATCH 5/8] flow: Make flow timer per-caller for thread safety

On Tue, Jun 16, 2026 at 07:10:49PM +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

LGTM

...

--- flow.c | 11 +++++------ flow.h | 2 +- passt.c | 8 +++++--- 3 files changed, 11 insertions(+), 10 deletions(-)

diff --git a/flow.c b/flow.c index 787a7139cfc1..08c7620c7b0f 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) @@ -898,9 +895,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; @@ -908,9 +907,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 53e0408a9ee5..10634e64a7fc 100644 --- a/flow.h +++ b/flow.h @@ -280,7 +280,7 @@ void flow_migrate(struct flow_common *f, unsigned int qpair, uint32_t events, (flow_migrate(&(flow_)->f, qpair_, events_, fd_, 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 3afc59b19120..bebc2b99f523 100644 --- a/passt.c +++ b/passt.c @@ -98,15 +98,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) @@ -221,6 +222,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; @@ -304,7 +306,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); } -- 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

David Gibson

4:55 a.m.

New subject: [PATCH 6/8] tcp: Make TCP timer state per-caller and guard global tasks

On Tue, Jun 16, 2026 at 07:10:50PM +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 --- 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 bebc2b99f523..ca5973e17317 100644 --- a/passt.c +++ b/passt.c @@ -96,16 +96,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)

This is pretty bulk, and also threads several TCP specific things through this not TCP specific code path. This seems like a place where a qpair_ctx which holds these timestamps would be useful.

...

{ 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); @@ -130,16 +137,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 @@ -221,8 +218,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; @@ -306,7 +304,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); } @@ -433,8 +432,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 f4fe866ba7c3..955012355d69 100644 --- a/tcp.c +++ b/tcp.c @@ -2988,17 +2988,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; @@ -3021,18 +3023,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; @@ -3054,27 +3058,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)

This one is TCP specific, but it's still pretty bulky.

...

{ - tcp_payload_flush(c); + if (qpair == 0) + tcp_payload_flush(c);

- 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 490f1b140e44..64c75ba481bd 100644 --- a/tcp.h +++ b/tcp.h @@ -32,7 +32,8 @@ int tcp_listen(const struct ctx *c, uint8_t pif, unsigned rule, const union inany_addr *addr, const char *ifname, in_port_t port); 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);

@@ -42,24 +43,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

David Gibson

5:03 a.m.

New subject: [PATCH 8/8] flow: Add mutex and per-qpair filtering to flow table operations

On Tue, Jun 16, 2026 at 07:10:52PM +0200, Laurent Vivier wrote:

...

The flow table free list, hash table, and flow_new_entry are global shared state accessed from multiple threads.

Protect flow_alloc(), flow_alloc_cancel(), flow_hash_insert(), flow_hash_remove(), and the free list rebuild in flow_defer_handler() with a pthread_rwlock_t: writers for mutations, readers for lookups.

Make flow_new_entry _Thread_local so each thread independently tracks its own in-progress allocation.

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() and flow_defer_handler() accordingly: skip these entries silently rather than treating them as errors, and break free-list cluster merging across them.

Filter flow_defer_handler()'s first loop by qpair, so each thread only processes its own flows.

Signed-off-by: Laurent Vivier

I'm skipping review of this, since it sounds like you're reworking the flow table locking significantly anyway.

...

--- flow.c | 59 +++++++++++++++++++++++++++++++++++++++++++--------- flow_table.h | 2 +- 2 files changed, 50 insertions(+), 11 deletions(-)

diff --git a/flow.c b/flow.c index 08c7620c7b0f..149360c3ec87 100644 --- a/flow.c +++ b/flow.c @@ -12,6 +12,8 @@ #include #include

+#include + #include "util.h" #include "ip.h" #include "passt.h" @@ -129,7 +131,7 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES,

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 +144,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) @@ -616,12 +620,18 @@ void flow_activate(struct flow_common *f) */ union flow *flow_alloc(void) { - 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); @@ -650,6 +660,8 @@ union flow *flow_alloc(void) memset(flow, 0, sizeof(*flow)); flow_set_state(&flow->f, FLOW_STATE_NEW);

+ pthread_rwlock_unlock(&flow_lock); + return flow; }

@@ -661,6 +673,8 @@ union flow *flow_alloc(void) */ 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 || @@ -678,6 +692,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); }

/** @@ -763,9 +779,13 @@ 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); + unsigned b;

+ pthread_rwlock_wrlock(&flow_lock); + 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);

@@ -779,10 +799,16 @@ 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;

- if (!flow_sidx_valid(flow_hashtab[b])) + pthread_rwlock_wrlock(&flow_lock); + + b = flow_hash_probe(c, sidx); + + 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); @@ -802,6 +828,8 @@ void flow_hash_remove(const struct ctx *c, flow_sidx_t sidx) }

flow_hashtab[b] = FLOW_SIDX_NONE; + + pthread_rwlock_unlock(&flow_lock); }

/** @@ -816,10 +844,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 && @@ -827,7 +857,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; }

/** @@ -920,6 +954,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); @@ -951,6 +988,7 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now, }

/* Second step: actually free the flows */ + pthread_rwlock_wrlock(&flow_lock); flow_foreach_slot(flow) { switch (flow->f.state) { case FLOW_STATE_FREE: { @@ -979,8 +1017,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: @@ -1012,6 +1050,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 e4ff6f73c35c..f2545390205a 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

David Gibson

4:59 a.m.

New subject: [PATCH 7/8] tcp: Protect init socket pools with mutex for thread safety

On Tue, Jun 16, 2026 at 07:10:51PM +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 wonder if we might be better off with a per-qpair/thread socket pool instead of a mutexed common pool. The obvious tradeoff would be that the pool is smaller so a thread could run out of sockets while we still have some available on another thread. But, having a common pool only gains us there if the mutex lock is significantly cheaper than the system calls to refill the pool. That might be the case, but I'm not certain.

...

Signed-off-by: Laurent Vivier --- tcp.c | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/tcp.c b/tcp.c index 955012355d69..019340c1c348 100644 --- a/tcp.c +++ b/tcp.c @@ -293,6 +293,7 @@ #include #include #include +#include

#include #include @@ -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 @@ -1568,7 +1570,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 @@ -2857,6 +2863,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) @@ -2869,6 +2876,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

Age (days ago)

Last active (days ago)

List overview

Download

16 comments

2 participants

participants (2)

David Gibson
Laurent Vivier

[PATCH 0/8] multithreading: Prepare data structures for concurrent queue pair workers

tags

participants (2)