On Fri, Jul 31, 2026 at 06:46:27PM +0200, Laurent Vivier wrote:
> 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];

Since this will have a 1 to 1 correspondance with a threading_context,
is there a reason to make them separate structures?

>  /**
>   * 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;

I'm not really sure what this change is for.

>  		/* 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]);

This is an unsynchronized access to a variable being written by
another thread.  Since it is just stats, it's probably not harmful,
but could certainly return temporarily nonsense results.

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

AFAICT we controll the dispatch from threading_worker() to the worker
function to here.  Why do we need an opaque pointer when we could
resolve the thread id at the top level and pass that down typed in a
more natural way?


> +	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);

The two if branches are uncomfortably similar.  It should be ok to
unconditionally initialise all the thread and qpair contexts, as long
as you don't actually start the secondary threads if !vhost-user.

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

Uh.. I don't see an 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));

Doesn't this change belong in the earlier patches introducing the queues?

>  	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);

Likewise this one.

>  		}
>  
>  		count++;
>  	}
> -	tap_handler(vdev->context, QPAIR_DEFAULT, now);
> +	tap_handler(vdev->context, QPAIR_FROM_QUEUE(index), now);

And this.

>  	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;

Not sure why this change.

>  	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);

Or this one.

>  }
>  
>  /**
> -- 
> 2.54.0
> 

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