[PATCH 6/7] vhost-user: Add per-qpair worker threads

[Laurent Vivier <lvivier@redhat.com>]

Add pthread-based worker threads for vhost-user queue pair processing.
Each queue pair gets its own worker thread with a dedicated epollfd,
enabling parallel packet processing across queue pairs.

Thread 0 handles control events (listen sockets, vhost-user commands,
repair, netlink) plus queue pair 0. Threads 1-15 handle only their
assigned queue pair events.

Threads start/stop dynamically via SET_VRING_ENABLE. Virtqueue kick
eventfds are registered on the appropriate worker thread's epollfd.

Global operations (fwd_scan_ports_timer, ndp_timer) run only on
thread 0.

Thread safety relies on infrastructure from the threadsafe series
(pthread_rwlock on flow table, pthread_mutex on socket pools,
per-qpair buffers, lock-free flow migration).

Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
 flow.c       |   7 +--
 passt.c      | 129 +++++++++++++++++++++++++++++++++++++--------------
 threading.c  |   2 -
 vhost_user.c |  32 ++++++++++---
 vhost_user.h |   1 +
 vu_common.c  |  13 +++---
 6 files changed, 130 insertions(+), 54 deletions(-)

diff --git a/flow.c b/flow.c
index 7821b3f91285..91070c9beb2e 100644
--- a/flow.c
+++ b/flow.c
@@ -457,12 +457,7 @@ static void flow_set_state(struct flow_common *f, enum flow_state state)
  */
 int flow_epollfd(const struct flow_common *f)
 {
-	/* mapping 1:1 between qpair and threadid
-	 * return threading_epollfd(f->qpair);
-	 * but for the moment we have only one thread
-	 */
-	(void)f;
-	return threading_epollfd(THREADING_ID_DEFAULT);
+	return threading_epollfd(f->qpair);
 }
 
 /**
diff --git a/passt.c b/passt.c
index 469e2525f545..33b19264aa6f 100644
--- a/passt.c
+++ b/passt.c
@@ -55,6 +55,7 @@
 #include "netlink.h"
 #include "epoll_ctl.h"
 #include "threading.h"
+#include "flow_table.h"
 
 char pkt_buf[PKT_BUF_BYTES]	__attribute__ ((aligned(PAGE_SIZE)));
 
@@ -100,18 +101,40 @@ struct passt_stats {
 	unsigned long events[EPOLL_NUM_TYPES];
 };
 
+/**
+ * struct qp_context - Per-queue pair context for vhost-user operations
+ * @vdev:		Pointer to vhost-user device
+ * @qpid:		Queue pair identifier
+ * @stats:		Per-queue statistics
+ * @flow_timer_run:	Last time the flow timers ran for this queue pair
+ * @tcp_timer_run:	Last time TCP timers ran for this queue pair
+ * @keepalive_run:	Last time TCP keepalives ran for this queue pair
+ * @inactivity_run:	Last time TCP inactivity scan ran for this queue pair
+ */
+struct qp_context {
+	struct vu_dev *vdev;
+	unsigned int qpid;
+	struct passt_stats stats;
+	struct timespec flow_timer_run;
+	struct timespec tcp_timer_run;
+	time_t keepalive_run;
+	time_t inactivity_run;
+};
+
+static struct qp_context qp_context[VHOST_USER_MAX_VQS / 2];
+
 /**
  * post_handler() - Run periodic and deferred tasks for L4 protocol handlers
  * @c:			Execution context
  * @now:		Current timestamp
- * @timer_run:		Last time the flow timers ran
+ * @flow_timer_run:		Last time the flow timers ran
  * @tcp_timer_run:	Last time TCP timers ran
  * @keepalive_run:	Last time keepalives ran
  * @inactivity_run:	Last time inactivity scan ran
  * @qpair:		Queue pair to process
  */
 static void post_handler(struct ctx *c, const struct timespec *now,
-			 struct timespec *timer_run,
+			 struct timespec *flow_timer_run,
 			 struct timespec *tcp_timer_run,
 			 time_t *keepalive_run,
 			 time_t *inactivity_run, unsigned int qpair)
@@ -120,11 +143,13 @@ static void post_handler(struct ctx *c, const struct timespec *now,
 		tcp_defer_handler(c, now, tcp_timer_run, keepalive_run,
 				  inactivity_run, qpair);
 
-	flow_defer_handler(c, now, timer_run, qpair);
-	fwd_scan_ports_timer(c, now);
+	flow_defer_handler(c, now, flow_timer_run, qpair);
+	if (qpair == 0) {
+		fwd_scan_ports_timer(c, now);
 
-	if (!c->no_ndp)
-		ndp_timer(c, now);
+		if (!c->no_ndp)
+			ndp_timer(c, now);
+	}
 }
 
 /**
@@ -172,16 +197,14 @@ static void exit_handler(int signal)
 /**
  * print_stats() - Print event statistics table to stderr
  * @c:		Execution context
- * @stats:	Event counters
  * @now:	Current timestamp
  */
-static void print_stats(const struct ctx *c, const struct passt_stats *stats,
-			const struct timespec *now)
+static void print_stats(const struct ctx *c, const struct timespec *now)
 {
 	static struct timespec before;
-	static int lines_printed;
+	static int lines_printed = 20;
 	long long elapsed_ns;
-	int i;
+	unsigned int i, j;
 
 	if (!c->stats)
 		return;
@@ -194,11 +217,12 @@ static void print_stats(const struct ctx *c, const struct passt_stats *stats,
 
 	before = *now;
 
-	if (!(lines_printed % 20)) {
+	if (lines_printed >= 20) {
+		lines_printed = 0;
 		/* Table header */
 		for (i = 1; i < EPOLL_NUM_TYPES; i++) {
-			int j;
 
+			FPRINTF(stderr, "    \t");
 			for (j = 0; j < i * (6 + 1); j++) {
 				if (j && !(j % (6 + 1)))
 					FPRINTF(stderr, "|");
@@ -209,11 +233,19 @@ static void print_stats(const struct ctx *c, const struct passt_stats *stats,
 		}
 	}
 
-	FPRINTF(stderr, " ");
-	for (i = 1; i < EPOLL_NUM_TYPES; i++)
-		FPRINTF(stderr, " %6lu", stats->events[i]);
-	FPRINTF(stderr, "\n");
-	lines_printed++;
+	for (j = 0; j < VHOST_USER_MAX_VQS / 2; j++) {
+
+		if (!threading_is_active(j))
+			continue;
+
+		FPRINTF(stderr, "%u\t ", j);
+
+		for (i = 1; i < EPOLL_NUM_TYPES; i++)
+			FPRINTF(stderr, " %6lu", qp_context[j].stats.events[i]);
+
+		FPRINTF(stderr, "\n");
+		lines_printed++;
+	}
 }
 
 /**
@@ -224,10 +256,8 @@ static void print_stats(const struct ctx *c, const struct passt_stats *stats,
  */
 static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
 {
-	static time_t keepalive_run, inactivity_run;
-	static struct passt_stats stats = { 0 };
-	static struct timespec flow_timer_run, tcp_timer_run;
-	struct ctx *c = opaque;
+	struct qp_context *qpc = opaque;
+	struct ctx *c = qpc->vdev->context;
 	struct timespec now;
 	int i;
 
@@ -259,7 +289,7 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
 			pasta_netns_quit_timer_handler(c, ref);
 			break;
 		case EPOLL_TYPE_TCP:
-			tcp_sock_handler(c, ref, eventmask, &now, QPAIR_DEFAULT);
+			tcp_sock_handler(c, ref, eventmask, &now, qpc->qpid);
 			break;
 		case EPOLL_TYPE_TCP_SPLICE:
 			tcp_splice_sock_handler(c, ref, eventmask, &now);
@@ -268,24 +298,24 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
 			tcp_listen_handler(c, ref, &now);
 			break;
 		case EPOLL_TYPE_TCP_TIMER:
-			tcp_timer_handler(c, ref, &now, QPAIR_DEFAULT);
+			tcp_timer_handler(c, ref, &now, qpc->qpid);
 			break;
 		case EPOLL_TYPE_UDP_LISTEN:
 			udp_listen_sock_handler(c, ref, eventmask, &now,
-						QPAIR_DEFAULT);
+						qpc->qpid);
 			break;
 		case EPOLL_TYPE_UDP:
 			udp_sock_handler(c, ref, eventmask, &now,
-					 QPAIR_DEFAULT);
+					 qpc->qpid);
 			break;
 		case EPOLL_TYPE_PING:
-			icmp_sock_handler(c, ref, &now, QPAIR_DEFAULT);
+			icmp_sock_handler(c, ref, &now, qpc->qpid);
 			break;
 		case EPOLL_TYPE_VHOST_CMD:
 			vu_control_handler(c->vdev, c->fd_tap, eventmask);
 			break;
 		case EPOLL_TYPE_VHOST_KICK:
-			vu_kick_cb(c->vdev, ref, &now);
+			vu_kick_cb(qpc->vdev, ref, &now);
 			break;
 		case EPOLL_TYPE_REPAIR_LISTEN:
 			repair_listen_handler(c, eventmask);
@@ -306,14 +336,16 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
 			/* Can't happen */
 			assert(0);
 		}
-		stats.events[ref.type]++;
-		print_stats(c, &stats, &now);
+
+		qpc->stats.events[ref.type]++;
+		print_stats(c, &now);
 	}
 
-	post_handler(c, &now, &flow_timer_run, &tcp_timer_run,
-		     &keepalive_run, &inactivity_run, QPAIR_DEFAULT);
+	post_handler(c, &now, &qpc->flow_timer_run, &qpc->tcp_timer_run,
+		     &qpc->keepalive_run, &qpc->inactivity_run, qpc->qpid);
 
-	migrate_handler(c, &now);
+	if (qpc->qpid == 0)
+		migrate_handler(c, &now);
 }
 
 /**
@@ -376,6 +408,7 @@ int main(int argc, char **argv)
 	madvise(pkt_buf, sizeof(pkt_buf), MADV_HUGEPAGE);
 
 	threading_init();
+	vu_preinit(c);
 
 	if (getrlimit(RLIMIT_NOFILE, &limit))
 		die_perror("Failed to get maximum value of open files limit");
@@ -445,6 +478,34 @@ int main(int argc, char **argv)
 
 	isolate_postfork(c);
 
-	threading_worker_set(THREADING_ID_DEFAULT, passt_worker, c);
+	/* thread #0 is specific to main process */
+	if (c->mode == MODE_VU) {
+		unsigned int i;
+
+		for (i = 0; i < VHOST_USER_MAX_VQS / 2; i++) {
+			struct qp_context *qpc = &qp_context[i];
+
+			qpc->vdev = c->vdev;
+			qpc->qpid = i;
+			qpc->flow_timer_run = now;
+			qpc->tcp_timer_run = now;
+			qpc->keepalive_run = 0;
+			qpc->inactivity_run = 0;
+
+			threading_worker_set(i, passt_worker, qpc);
+		}
+	} else {
+		struct qp_context *qpc = &qp_context[0];
+
+		qpc->vdev = c->vdev;
+		qpc->qpid = 0;
+		qpc->flow_timer_run = now;
+		qpc->tcp_timer_run = now;
+		qpc->keepalive_run = 0;
+		qpc->inactivity_run = 0;
+
+		threading_worker_set(THREADING_ID_DEFAULT, passt_worker, qpc);
+	}
+
 	threading_start_thread(THREADING_ID_DEFAULT);
 }
diff --git a/threading.c b/threading.c
index e1ed644f5864..5f3453b02d99 100644
--- a/threading.c
+++ b/threading.c
@@ -169,7 +169,6 @@ void threading_start_thread(unsigned int threadid)
  * threading_stop_thread() - Stop and join a worker thread
  * @threadid:	Thread index to stop
  */
-/* cppcheck-suppress unusedFunction */
 void threading_stop_thread(unsigned int threadid)
 {
 	if (threadid >= ARRAY_SIZE(threads))
@@ -193,7 +192,6 @@ void threading_stop_thread(unsigned int threadid)
  * Return: true if the thread at the given index has been created and is active,
  *         false otherwise
  */
-/* cppcheck-suppress unusedFunction */
 bool threading_is_active(unsigned int threadid)
 {
 	if (threadid >= ARRAY_SIZE(threads))
diff --git a/vhost_user.c b/vhost_user.c
index ad316aee12c3..ad5e062726a9 100644
--- a/vhost_user.c
+++ b/vhost_user.c
@@ -45,6 +45,7 @@
 #include "migrate.h"
 #include "epoll_ctl.h"
 #include "threading.h"
+#include "vu_common.h"
 
 /* vhost-user version we are compatible with */
 #define VHOST_USER_VERSION 1
@@ -737,7 +738,7 @@ static bool vu_get_vring_base_exec(struct vu_dev *vdev,
 		vdev->vq[idx].call_fd = -1;
 	}
 	if (vdev->vq[idx].kick_fd != -1) {
-		epoll_del(threading_epollfd(THREADING_ID_DEFAULT), vdev->vq[idx].kick_fd);
+		epoll_del(threading_epollfd(idx / 2), vdev->vq[idx].kick_fd);
 		close(vdev->vq[idx].kick_fd);
 		vdev->vq[idx].kick_fd = -1;
 	}
@@ -758,7 +759,7 @@ static void vu_set_watch(const struct vu_dev *vdev, int idx)
 		.queue = idx
 	 };
 
-	epoll_add(threading_epollfd(THREADING_ID_DEFAULT), EPOLLIN, ref);
+	epoll_add(threading_epollfd(idx / 2), EPOLLIN, ref);
 }
 
 /**
@@ -803,7 +804,7 @@ static bool vu_set_vring_kick_exec(struct vu_dev *vdev,
 	vu_check_queue_msg_file(vmsg);
 
 	if (vdev->vq[idx].kick_fd != -1) {
-		epoll_del(threading_epollfd(THREADING_ID_DEFAULT), vdev->vq[idx].kick_fd);
+		epoll_del(threading_epollfd(idx / 2), vdev->vq[idx].kick_fd);
 		close(vdev->vq[idx].kick_fd);
 		vdev->vq[idx].kick_fd = -1;
 	}
@@ -955,6 +956,8 @@ static bool vu_get_queue_num_exec(struct vu_dev *vdev,
  * @vmsg:	vhost-user message
  *
  * Return: false as no reply is requested
+ *
+ * #syscalls:vu madvise
  */
 static bool vu_set_vring_enable_exec(struct vu_dev *vdev,
 				     struct vhost_user_msg *vmsg)
@@ -968,7 +971,16 @@ static bool vu_set_vring_enable_exec(struct vu_dev *vdev,
 	if (idx >= VHOST_USER_MAX_VQS)
 		die("Invalid vring_enable index: %u", idx);
 
+	if (vdev->vq[idx].enable == enable)
+		return false;
+
 	vdev->vq[idx].enable = enable;
+
+	if (vdev->vq[idx & ~1].enable || vdev->vq[(idx & ~1) + 1].enable)
+		threading_start_thread(idx / 2);
+	else
+		threading_stop_thread(idx / 2);
+
 	return false;
 }
 
@@ -1047,6 +1059,16 @@ static bool vu_check_device_state_exec(struct vu_dev *vdev,
 	return true;
 }
 
+/**
+ * vu_preinit() - Pre-initialize vhost-user device storage
+ * @c:		execution context
+ */
+void vu_preinit(struct ctx *c)
+{
+	c->vdev = &vdev_storage;
+	c->vdev->context = c;
+}
+
 /**
  * vu_init() - Initialize vhost-user device structure
  * @c:		execution context
@@ -1055,8 +1077,6 @@ void vu_init(struct ctx *c)
 {
 	unsigned i;
 
-	c->vdev = &vdev_storage;
-	c->vdev->context = c;
 	for (i = 0; i < VHOST_USER_MAX_VQS; i++) {
 		c->vdev->vq[i] = (struct vu_virtq){
 			.call_fd = -1,
@@ -1096,7 +1116,7 @@ void vu_cleanup(struct vu_dev *vdev)
 			vq->err_fd = -1;
 		}
 		if (vq->kick_fd != -1) {
-			epoll_del(threading_epollfd(THREADING_ID_DEFAULT), vq->kick_fd);
+			epoll_del(threading_epollfd(i / 2), vq->kick_fd);
 			close(vq->kick_fd);
 			vq->kick_fd = -1;
 		}
diff --git a/vhost_user.h b/vhost_user.h
index d2e51d3e86c3..1a5614e83c32 100644
--- a/vhost_user.h
+++ b/vhost_user.h
@@ -230,6 +230,7 @@ static inline bool vu_queue_started(const struct vu_virtq *vq)
 }
 
 void vu_print_capabilities(void);
+void vu_preinit(struct ctx *c);
 void vu_init(struct ctx *c);
 void vu_cleanup(struct vu_dev *vdev);
 void vu_log_write(const struct vu_dev *vdev, uint64_t address,
diff --git a/vu_common.c b/vu_common.c
index 0cd19b36737d..c3e3dfc05f26 100644
--- a/vu_common.c
+++ b/vu_common.c
@@ -177,7 +177,7 @@ static void vu_handle_tx(struct vu_dev *vdev, int index,
 
 	assert(QPAIR_IS_FROMGUEST(index));
 
-	tap_flush_pools(QPAIR_DEFAULT);
+	tap_flush_pools(QPAIR_FROM_QUEUE(index));
 
 	count = 0;
 	out_sg_count = 0;
@@ -199,12 +199,13 @@ static void vu_handle_tx(struct vu_dev *vdev, int index,
 
 		data = IOV_TAIL(elem[count].out_sg, elem[count].out_num, 0);
 		if (IOV_DROP_HEADER(&data, struct virtio_net_hdr_mrg_rxbuf)) {
-			tap_add_packet(vdev->context, QPAIR_DEFAULT, &data, now);
+			tap_add_packet(vdev->context, QPAIR_FROM_QUEUE(index),
+				       &data, now);
 		}
 
 		count++;
 	}
-	tap_handler(vdev->context, QPAIR_DEFAULT, now);
+	tap_handler(vdev->context, QPAIR_FROM_QUEUE(index), now);
 
 	if (count) {
 		int i;
@@ -230,12 +231,12 @@ void vu_kick_cb(struct vu_dev *vdev, union epoll_ref ref,
 
 	rc = eventfd_read(ref.fd, &kick_data);
 	if (rc == -1)
-		die_perror("vhost-user kick eventfd_read()");
+		return;
 
 	trace("vhost-user: got kick_data: %016"PRIx64" idx: %d",
 	      kick_data, ref.queue);
-	if (QPAIR_IS_FROMGUEST(ref.queue))
-		vu_handle_tx(vdev, ref.queue, now);
+
+	vu_handle_tx(vdev, ref.queue, now);
 }
 
 /**
-- 
2.54.0