From mboxrd@z Thu Jan 1 00:00:00 1970 Authentication-Results: passt.top; dmarc=none (p=none dis=none) header.from=gibson.dropbear.id.au Authentication-Results: passt.top; dkim=pass (2048-bit key; secure) header.d=gibson.dropbear.id.au header.i=@gibson.dropbear.id.au header.a=rsa-sha256 header.s=202608 header.b=Uba5GW04; dkim-atps=neutral Received: from mail.ozlabs.org (gandalf.ozlabs.org [150.107.74.76]) by passt.top (Postfix) with ESMTPS id 2F59B5A0272 for ; Wed, 05 Aug 2026 02:31:09 +0200 (CEST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gibson.dropbear.id.au; s=202608; t=1785889865; bh=nGED4avYpOUcvMK/KPbai1gV4+C5LrgtsfZfEJQqwwQ=; h=Date:From:To:Cc:Subject:References:In-Reply-To:From; b=Uba5GW04eWrj64UwXuEHsvJt804bf62HcfduOiu/MKr/RWhSNiboNUSHyj9V2BBES oRiw75Pvfnhe4v+Ah3uNCSzD+s/tUSlk72U3CDFa2WcTimJASsK8xCdNOwhwTMi1B4 l6mVCgQdRHUwDbQH5qRspLIq/oVusXMSq6/XPMzggj3Vi6MuDqlMkcJHByia5+zZME CPWTjB7GPYJw6R0YvG3LpB0A8D9yXBCRitGmKIInBbKBhi1x/IfzXWpm8y1rRNdAC9 mEiSbQjqMguC8FFfI+nP+gvzqiGuTwGgFCTrcX8PgOL6RMDzgwn0b4pWtBoa5tIT0/ qC+bwSFg+Bk5Q== Received: by gandalf.ozlabs.org (Postfix, from userid 1007) id 4hFBCs5d1Yz4w2B; Wed, 05 Aug 2026 10:31:05 +1000 (AEST) Date: Wed, 5 Aug 2026 10:30:47 +1000 From: David Gibson To: Laurent Vivier Subject: Re: [PATCH v2 10/10] flow: Add lazy, lock-free flow migration between queue pairs Message-ID: References: <20260731162329.3552800-1-lvivier@redhat.com> <20260731162329.3552800-11-lvivier@redhat.com> MIME-Version: 1.0 Content-Type: multipart/signed; micalg=pgp-sha512; protocol="application/pgp-signature"; boundary="dyyE++pbkY8TER7o" Content-Disposition: inline In-Reply-To: <20260731162329.3552800-11-lvivier@redhat.com> Message-ID-Hash: QME3YY75TX3IQXV2O64WDM3GTGT5ZUYO X-Message-ID-Hash: QME3YY75TX3IQXV2O64WDM3GTGT5ZUYO X-MailFrom: dgibson@gandalf.ozlabs.org X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; emergency; loop; banned-address; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header CC: passt-dev@passt.top X-Mailman-Version: 3.3.8 Precedence: list List-Id: Development discussion and patches for passt Archived-At: Archived-At: List-Archive: List-Archive: List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: --dyyE++pbkY8TER7o Content-Type: text/plain; charset=us-ascii Content-Disposition: inline Content-Transfer-Encoding: quoted-printable 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. >=20 > 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. >=20 > TCP timers are migrated the same way in tcp_timer_handler() using > tcp_timer_epoll_add(). >=20 > 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(-) >=20 > 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) =3D=3D 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 > + * =3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D= =3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D > + * > + * In vhost-user multiqueue mode, each queue pair has its own worker thr= ead. > + * 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] =3D 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 =3D 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. > */ > =20 > unsigned flow_first_free; > @@ -181,6 +235,9 @@ static_assert(ARRAY_SIZE(flow_hashtab) >=3D 2 * FLOW_= MAX, > =20 > static pthread_rwlock_t flow_lock =3D PTHREAD_RWLOCK_INITIALIZER; > =20 > +static unsigned int to_migrate[FLOW_MAX]; > +static pthread_mutex_t migrate_lock =3D 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, unsign= ed int qpair) > f->qpair =3D qpair; > } > =20 > +/** > + * flow_migrate_epollfd() - Migrate a flow to a different epollfd if pen= ding > + * @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 ot= herwise > + */ > +bool flow_migrate_epollfd(struct flow_common *f, unsigned int qpair, > + uint32_t events, union epoll_ref ref) > +{ > + unsigned int target; > + bool ret =3D false; > + > + assert(qpair < FLOW_QPAIR_SIZE); > + > + pthread_mutex_lock(&migrate_lock); > + target =3D to_migrate[flow_idx(f)]; > + if (target =3D=3D 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 =3D 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 =3D false; > + > + pthread_mutex_lock(&migrate_lock); > + if (f->qpair =3D=3D qpair) > + goto out; > + > + to_migrate[flow_idx(f)] =3D qpair; > + ret =3D 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) > =20 > flow_new_entry =3D flow; > memset(flow, 0, sizeof(*flow)); > + to_migrate[FLOW_IDX(flow)] =3D qpair; > flow_setqp(&flow->f, qpair); > flow_set_state(&flow->f, FLOW_STATE_NEW); > =20 > 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 ev= ents, > 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); > =20 > 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, > =20 > assert(pingf); > =20 > + if (flow_migrate_epollfd(&pingf->f, qpair, EPOLLIN, ref)) > + return; > + > n =3D 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 =3D flow_at_sidx(flow_lookup_af(c, proto, PIF_TAP, > af, saddr, daddr, id, id)); > =20 > - if (flow) > + if (flow) { > pingf =3D &flow->ping; > - else if (!(pingf =3D 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 =3D ' 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 =3D icmp_ping_new(c, qpair, af, id, saddr, daddr, no= w))) { > return 1; > + } > =20 > tgt =3D &pingf->f.side[TGTSIDE]; > =20 > 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 i= nt qpair, uint8_t pif, > assert(pif_at_sidx(sidx) =3D=3D 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 =3D &flow->tcp; > =20 > + if (flow_migrate_mark(&conn->f, qpair)) > + return 1; > + > flow_trace(conn, "packet length %zu from tap", l4len); > =20 > 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 =3D=3D FLOW_TCP); > =20 > + if (conn->f.qpair !=3D qpair) { Here you're accessing f.qpair without the lock at all. > + int old_epollfd =3D 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 =3D -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 e= poll_ref ref, > assert(!c->no_tcp); > assert(pif_at_sidx(ref.flowside) !=3D PIF_TAP); > =20 > + if (flow_migrate_epollfd(&conn->f, qpair, > + tcp_conn_epoll_events(conn->events, > + conn->flags), > + ref)) > + return; > + > if (conn->events =3D=3D CLOSED) > return; > =20 > 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 epol= l_ref ref, uint32_t events, > =20 > assert(!c->no_udp && uflow); > =20 > + 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, u= nsigned int qpair, > =20 > sidx =3D flow_lookup_sa(c, IPPROTO_UDP, pif, s_in, dst, port); > if ((uflow =3D 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, un= signed int qpair, > sidx =3D flow_lookup_af(c, IPPROTO_UDP, pif, af, saddr, daddr, > srcport, dstport); > if ((uflow =3D 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); > } > --=20 > 2.54.0 >=20 --=20 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 --dyyE++pbkY8TER7o Content-Type: application/pgp-signature; name=signature.asc -----BEGIN PGP SIGNATURE----- iQIzBAEBCgAdFiEEO+dNsU4E3yXUXRK2zQJF27ox2GcFAmpyhDEACgkQzQJF27ox 2Gew1w//Y1rR88//DvFrNS9Xdm/30b86euj4ynzdV6v2WJwfh/Dlsmv+zZ+WYpz1 O9E0GcPHLMe9h8CSiZ4Cj40DFJD1OosxJE+RahkbNRn7esSs8c0SBhCmYS+lSuHa A3A5k0v3RQbhovo1WtZxDbkpj5VxmrV4LHzTkFJzxlVf5HsA/+R/IxzMtVt22u1R WnnuoBQunDHQNriqwgtMvByzK6U9Ynm5sepc/ZTf9ptT0OpzkObLvYORMD1ST2dn xNYTR2n8WwyF5WCq7GmIctBr7tnZSk7q6/26ecs9xA0fbdtbdLTuzCaDHuthbwkY ijiSiJwkfrUQXLgqPD8PGr7AM747VLe8FayFxeqHAk0cVDepPJtCePcN3YyB4Axx bp/PLEc/2zV+zNyPmF6kgGY9x68VRIjakY6ojeLquBdHGdjEcqcKCCxYmfjM5uy4 4KbJ3LZ4ArfAQ6FIbr7MkLfRuOSHjLa1wiDKlXtLpfSc7WQFpp1SY3/EnBdNElvK Wc4ngulSrhV8FCU0FZyuYmucxVuRkTU83hwFEiZKxsF6+5iMP+LEf/YzSeD1VsNA 4fAEoYPeBjGv8i+lNhHtq/R/9B1zVwVUJDu8D0RvTcTYlEIQ8TSDDvbVIeKQ1CCZ xatfATfXx4lhiHsVSzUC/vgXEPoh0UQlW5U5I8wQwtNqwnJC3Vs= =RB89 -----END PGP SIGNATURE----- --dyyE++pbkY8TER7o--