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