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

[Laurent Vivier <lvivier@redhat.com>]

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
+ * 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