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 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

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 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

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 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

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 --- 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

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 --- 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

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

Reviewed-by: David Gibson

--- 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

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

Reviewed-by: David Gibson

--- 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

-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

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

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

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

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

On Fri, Jul 31, 2026 at 06:23:29PM +0200, Laurent Vivier wrote:

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 --- 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

I'm not sure what RSS means in this context.

+ * 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

The lazy approach means we'll get at least one socket event on the wrong thread/epollfd, which isn't ideal. Might be worth it for the other benefits, though.

+ * 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.

When does the processing happen instead for this event? We use edge triggered events (EPOLLET) in a number of places, so we can't count on the event being re-raised by the kernel.

+ * 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.

IIUC, if any packets are sent to tap during that window, they'll go to the "wrong" queue. That should be fine - a guest would always have to deal with some "lagging" packets on the old queue, I'm just making sure I understand the expected behaviour.

+ * - 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;

So, not entirely lock free. What exactly does migrate_lock protect? I guess it must be at least the to_migrate[] array, plus all the f->qpair fields, yes? Except.. I don't think this quite works to protect f->qpair. It's a bitfield which means that accesses to surrounding fields might access it incidentally and non-atomically - so all of those would also have to go under the lock.

/** 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

Return: value comment would be helpful.

+ */ +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))

It's just pointer math, so it's technically safe, but conceptually the 'pingf = ' shouldn't happen until after the migrate: if you don't know you own the flow entry, you don't know it's (still) a ping flow.

+ 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);

The assert is accessing a field of the flow entry before you've verified this thread still owns it (as does the one before it that fell out of the diff context). Also, there's a more subtle race here. It'd be weird, but it's possible we could get two SYNs for the same connection on different queues at basically the same time.

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) {

Here you're accessing f.qpair without the lock at all.

+ 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;

This means we're in a broken state and should force an RST, no?

+ } + 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

-- 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