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

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