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 <lvivier@redhat.com>
> ---
>  flow.c     | 114 +++++++++++++++++++++++++++++++++++++++++++++++++++++
>  flow.h     |   3 ++
>  icmp.c     |  10 ++++-
>  tcp.c      |  20 ++++++++++
>  udp.c      |   3 ++
>  udp_flow.c |   4 ++
>  6 files changed, 152 insertions(+), 2 deletions(-)
> 
> diff --git a/flow.c b/flow.c
> index 8deca3e3c7f1..3012a07ed680 100644
> --- a/flow.c
> +++ b/flow.c
> @@ -162,6 +162,60 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES,
>   * with flow_alloc().  The first pass (deferred protocol handlers) runs
>   * without the lock and filters by qpair, so each thread only processes
>   * its own flows.
> + *
> + * Flow migration between queue pairs
> + * ==================================
> + *
> + * In vhost-user multiqueue mode, each queue pair has its own worker thread.
> + * A flow is assigned to a queue pair (flow->f.qpair) and its socket is
> + * registered on that queue pair's epollfd.  The worker thread for that
> + * queue pair is responsible for processing events and running deferred
> + * handlers (flow_defer_handler) for its own flows.
> + *
> + * When the guest steers traffic for an existing flow to a different TX
> + * queue (e.g. due to RSS), the packet arrives on a queue pair that does

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