From mboxrd@z Thu Jan 1 00:00:00 1970 Authentication-Results: passt.top; dmarc=pass (p=quarantine dis=none) header.from=redhat.com Authentication-Results: passt.top; dkim=pass (1024-bit key; unprotected) header.d=redhat.com header.i=@redhat.com header.a=rsa-sha256 header.s=mimecast20190719 header.b=KjHSFePU; dkim-atps=neutral Received: from us-smtp-delivery-124.mimecast.com (us-smtp-delivery-124.mimecast.com [170.10.129.124]) by passt.top (Postfix) with ESMTPS id E63035A026E for ; Fri, 31 Jul 2026 18:23:48 +0200 (CEST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=redhat.com; s=mimecast20190719; t=1785515027; h=from:from:reply-to:subject:subject:date:date:message-id:message-id: to:to:cc:cc:mime-version:mime-version:content-type:content-type: content-transfer-encoding:content-transfer-encoding: in-reply-to:in-reply-to:references:references; bh=0tS3S/zgK2jqylgSogsrr8tzlgysdwgnXnCVM7gynV0=; b=KjHSFePUy5TNS50RHqxKTosZNYLFAdeI+v6ABBnHbpJKUgCnKMRJUx9S07FaHjNGw3pPgw puklc1ywzSRF23ebYge1JhCMI5JQtIMo/8pZw+RpREPB6o+JyB2dPUgi5ZYXJCDBg/qXrI ZwSeLEPSsg/mkiHbAc6pnY72IIDOG6U= Received: from mx-prod-mc-03.mail-002.prod.us-west-2.aws.redhat.com (ec2-54-186-198-63.us-west-2.compute.amazonaws.com [54.186.198.63]) by relay.mimecast.com with ESMTP with STARTTLS (version=TLSv1.3, cipher=TLS_AES_256_GCM_SHA384) id us-mta-587-_-C_i0gXPBKVKS3Onp-jbw-1; Fri, 31 Jul 2026 12:23:44 -0400 X-MC-Unique: _-C_i0gXPBKVKS3Onp-jbw-1 X-Mimecast-MFC-AGG-ID: _-C_i0gXPBKVKS3Onp-jbw_1785515023 Received: from mx-prod-int-03.mail-002.prod.us-west-2.aws.redhat.com (mx-prod-int-03.mail-002.prod.us-west-2.aws.redhat.com [10.30.177.12]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange X25519 server-signature RSA-PSS (2048 bits) server-digest SHA256) (No client certificate requested) by mx-prod-mc-03.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTPS id 645B719560B2 for ; Fri, 31 Jul 2026 16:23:43 +0000 (UTC) Received: from lenovo-t14s.redhat.corp (headnet05.pony-001.prod.iad2.dc.redhat.com [10.2.32.117]) by mx-prod-int-03.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTP id 9CE241956089; Fri, 31 Jul 2026 16:23:42 +0000 (UTC) From: Laurent Vivier To: passt-dev@passt.top Subject: [PATCH v2 10/10] flow: Add lazy, lock-free flow migration between queue pairs Date: Fri, 31 Jul 2026 18:23:29 +0200 Message-ID: <20260731162329.3552800-11-lvivier@redhat.com> In-Reply-To: <20260731162329.3552800-1-lvivier@redhat.com> References: <20260731162329.3552800-1-lvivier@redhat.com> MIME-Version: 1.0 X-Scanned-By: MIMEDefang 3.0 on 10.30.177.12 X-Mimecast-Spam-Score: 0 X-Mimecast-MFC-PROC-ID: hHjEkkdC-c7dkWOJBOiYvLkc4CFTm6Z4HWrJN-Jrxxg_1785515023 X-Mimecast-Originator: redhat.com Content-Transfer-Encoding: 8bit content-type: text/plain; charset="US-ASCII"; x-default=true Message-ID-Hash: 7YHT54OQZW6QNINSKC7WPYKR7S5T7TUB X-Message-ID-Hash: 7YHT54OQZW6QNINSKC7WPYKR7S5T7TUB X-MailFrom: lvivier@redhat.com 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: Laurent Vivier 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: 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