Re: [PATCH 2/7] threading: Add basic threading infrastructure

[David Gibson <david@gibson.dropbear.id.au>]

[-- Attachment #1: Type: text/plain, Size: 13295 bytes --]

On Fri, Jul 31, 2026 at 06:46:23PM +0200, Laurent Vivier wrote:
> Introduce a threading framework to support multi-threaded operation,
> particularly needed for vhost-user with multiqueue support. The
> infrastructure provides:
> 
> - Thread pool management with epoll-based event loops
> - Worker thread creation and teardown
> - Per-thread epoll file descriptors for event handling
> - Integration with existing context structure
> 
> This change modifies PID namespace handling to be compatible with
> pthread_create(), as CLONE_THREAD and CLONE_NEWPID are mutually
> exclusive for vhost-user mode.
> 
> Signed-off-by: Laurent Vivier <lvivier@redhat.com>
> ---
>  Makefile    |   8 +-
>  isolation.c |   6 +-
>  passt.c     |   7 +-
>  passt.h     |   5 ++
>  threading.c | 207 ++++++++++++++++++++++++++++++++++++++++++++++++++++
>  threading.h |  22 ++++++
>  6 files changed, 245 insertions(+), 10 deletions(-)
>  create mode 100644 threading.c
>  create mode 100644 threading.h
> 
> diff --git a/Makefile b/Makefile
> index 77474d695915..1907646f9e93 100644
> --- a/Makefile
> +++ b/Makefile
> @@ -37,8 +37,8 @@ PASST_SRCS = arch.c arp.c bitmap.c checksum.c conf.c dhcp.c dhcpv6.c \
>  	epoll_ctl.c flow.c fwd.c fwd_rule.c icmp.c igmp.c inany.c iov.c ip.c \
>  	isolation.c lineread.c log.c mld.c ndp.c netlink.c migrate.c packet.c \
>  	parse.c passt.c pasta.c pcap.c pif.c repair.c serialise.c tap.c tcp.c \
> -	tcp_buf.c tcp_splice.c tcp_vu.c udp.c udp_flow.c udp_vu.c util.c \
> -	vhost_user.c virtio.c vu_common.c
> +	tcp_buf.c tcp_splice.c tcp_vu.c threading.c udp.c udp_flow.c udp_vu.c \
> +	util.c vhost_user.c virtio.c vu_common.c
>  PASST_REPAIR_SRCS = passt-repair.c
>  PESTO_SRCS = pesto.c bitmap.c fwd_rule.c inany.c ip.c lineread.c parse.c \
>  	serialise.c
> @@ -51,8 +51,8 @@ PASST_HEADERS = arch.h arp.h bitmap.h checksum.h conf.h dhcp.h dhcpv6.h \
>  	inany.h iov.h ip.h isolation.h lineread.h log.h migrate.h ndp.h \
>  	netlink.h packet.h parse.h passt.h pasta.h pcap.h pif.h repair.h \
>  	serialise.h siphash.h tap.h tcp.h tcp_buf.h tcp_conn.h tcp_internal.h \
> -	tcp_splice.h tcp_vu.h udp.h udp_flow.h udp_internal.h udp_vu.h util.h \
> -	vhost_user.h virtio.h vu_common.h
> +	tcp_splice.h tcp_vu.h threading.h udp.h udp_flow.h udp_internal.h \
> +	udp_vu.h util.h vhost_user.h virtio.h vu_common.h
>  PASST_REPAIR_HEADERS = linux_dep.h
>  PESTO_HEADERS = bitmap.h common.h fwd_rule.h inany.h ip.h log.h parse.h \
>  	pesto.h serialise.h
> diff --git a/isolation.c b/isolation.c
> index 94cbe7f843fe..a8055b6e21ee 100644
> --- a/isolation.c
> +++ b/isolation.c
> @@ -394,8 +394,12 @@ int isolate_prefork(const struct ctx *c)
>  	/* If we run in foreground, we have no chance to actually move to a new
>  	 * PID namespace. For passt, use CLONE_NEWPID anyway, in case somebody
>  	 * ever gets around seccomp profiles -- there's no harm in passing it.
> +	 * We cannot use CLONE_NEWPID with vhost-user as we need to use
> +	 * pthread_create() for the multithreading. pthread_create() calls
> +	 * clone3(), with CLONE_THREAD that is not compatible with CLONE_NEWPID.
> +	 * (see EINVAL in clone(2))

That's a bit of a bummer.  I can see one possible way around it - the
main thing we want pthreads for is the mutexes.  If initialized with
the PTHREAD_PROCESS_SHARED attribute, these can be used between
different processes, not just threads in the same process, provided
the mutex lives in memory shared between them.  So, one possibility is
that we could manage our own threads directly with clone() rather than
using pthread_create() and still use the mutexes from libpthread.

>  	 */
> -	if (!c->foreground || c->mode != MODE_PASTA)
> +	if (!c->foreground || c->mode == MODE_PASST)
>  		flags |= CLONE_NEWPID;
>  
>  	if (unshare(flags)) {
> diff --git a/passt.c b/passt.c
> index 25e1212b9452..8a06838c5ed8 100644
> --- a/passt.c
> +++ b/passt.c
> @@ -54,11 +54,7 @@
>  #include "repair.h"
>  #include "netlink.h"
>  #include "epoll_ctl.h"
> -
> -#define NUM_EPOLL_EVENTS	8
> -
> -#define TIMER_INTERVAL_		MIN(TCP_TIMER_INTERVAL, FWD_PORT_SCAN_INTERVAL)
> -#define TIMER_INTERVAL		MIN(TIMER_INTERVAL_, FLOW_TIMER_INTERVAL)
> +#include "threading.h"
>  
>  char pkt_buf[PKT_BUF_BYTES]	__attribute__ ((aligned(PAGE_SIZE)));
>  
> @@ -380,6 +376,7 @@ int main(int argc, char **argv)
>  
>  	madvise(pkt_buf, sizeof(pkt_buf), MADV_HUGEPAGE);
>  
> +	threading_init();
>  	c->epollfd = epoll_create1(EPOLL_CLOEXEC);
>  	if (c->epollfd == -1)
>  		die_perror("Failed to create epoll file descriptor");
> diff --git a/passt.h b/passt.h
> index ec936c3d3f32..00ef5a8f85e5 100644
> --- a/passt.h
> +++ b/passt.h
> @@ -9,6 +9,11 @@
>  #define UNIX_SOCK_MAX		100
>  #define UNIX_SOCK_PATH		"/tmp/passt_%i.socket"
>  
> +#define NUM_EPOLL_EVENTS		8
> +
> +#define TIMER_INTERVAL_		MIN(TCP_TIMER_INTERVAL, FWD_PORT_SCAN_INTERVAL)
> +#define TIMER_INTERVAL		MIN(TIMER_INTERVAL_, FLOW_TIMER_INTERVAL)
> +
>  union epoll_ref;
>  
>  #include <stdbool.h>
> diff --git a/threading.c b/threading.c
> new file mode 100644
> index 000000000000..fd5b30ed7ff6
> --- /dev/null
> +++ b/threading.c
> @@ -0,0 +1,207 @@
> +// SPDX-License-Identifier: GPL-2.0-or-later
> +
> +/* PASST - Plug A Simple Socket Transport
> + *  for qemu/UNIX domain socket mode
> + *
> + * PASTA - Pack A Subtle Tap Abstraction
> + *  for network namespace/tap device mode
> + *
> + * threading.c - Basic threading infrastructure
> + *
> + * Copyright Red Hat
> + * Author: Laurent Vivier <lvivier@redhat.com>
> + */
> +
> +#include <errno.h>
> +#include <pthread.h>
> +#include <stdatomic.h>
> +
> +#include "util.h"
> +#include "passt.h"
> +#include "threading.h"
> +
> +/**
> + * struct threading_context - Per-thread execution context
> + * @epollfd:	Epoll file descriptor for this thread
> + * @pthread:	Pthread identifier
> + * @worker:	Worker function to execute in this thread

Why do we need a function pointer here?  Wouldn't we always be using
the same worker function?

> + * @opaque:	Opaque data passed to worker function
> + * @quit:	Flag to signal thread to exit

Might be worth elaborating on what synchronization is needed here.
AFAICT all these fields are read-only while the thread is active,
except for @quit, hence the _Atomic.  Is that correct?

> + */
> +struct threading_context {
> +	pthread_t pthread;
> +	void (*worker)(void *, int, struct epoll_event *);
> +	void *opaque;
> +	int epollfd;
> +	_Atomic bool quit;

Since these are all (nearly) read only, I guess we don't need to pad
this to a cacheline.  Might be worth noting that, though, because it
we did add a field that the thread worker writes to, it could cause
bad cacheline pingponging.

> +};
> +
> +/*
> + *  index 0 is reserved for main process
> + *  new threads start at index 1
> + */
> +struct threading_context threads[MAX_NUM_THREADS + 1];
> +
> +/**
> + * threading_init() - Initialize threading infrastructure
> + */
> +void threading_init(void)
> +{
> +	int i;
> +
> +	for (i = 0; i < ARRAY_SIZE(threads); i++) {
> +		threads[i].epollfd = epoll_create1(EPOLL_CLOEXEC);
> +		if (threads[i].epollfd == -1)
> +			die_perror("Failed to create epoll for threads #%d",
> +				   i);
> +	}
> +}
> +
> +/**
> + * threading_worker_set() - Set worker function for a thread
> + * @threadid:		Thread index
> + * @worker:		Worker function to assign to the thread
> + * @is_valid:	Validation function to select a thread
> + * @opaque:		Opaque data to pass to worker function
> + *
> + * Return: 0 on success, -1 if thread index is invalid
> + */
> +/* cppcheck-suppress unusedFunction */
> +int threading_worker_set(unsigned int threadid,
> +			 void (*worker)(void *, int, struct epoll_event *),
> +			 void *opaque)
> +{
> +	struct threading_context *tc;
> +
> +	if (threadid >= ARRAY_SIZE(threads))
> +		return -1;

Calling this with a bad threadid would necessarily be a bug, no?  In
which case assert() would make more sense.

> +
> +	tc = &threads[threadid];
> +
> +	tc->worker = worker;
> +	tc->opaque = opaque;
> +
> +	return 0;
> +}
> +
> +/**
> + * threading_epollfd() - Get epoll file descriptor for a thread
> + * @threadid:	Thread index
> + *
> + * Return: epoll file descriptor for the specified thread, -1 if index invalid
> + */
> +/* cppcheck-suppress unusedFunction */
> +int threading_epollfd(unsigned int threadid)
> +{
> +	if (threadid >= ARRAY_SIZE(threads))
> +		return -1;

assert?

> +
> +	return threads[threadid].epollfd;
> +}
> +
> +/**
> + * threading_worker() - Main worker thread function
> + * @opaque:	Pointer to threading_context for this thread
> + *
> + * Return: NULL on thread exit
> + *
> + * #syscalls poll write futex
> + */
> +static void *threading_worker(void *opaque)
> +{
> +	struct threading_context *tc = opaque;
> +
> +	while (!tc->quit) {
> +		struct epoll_event events[NUM_EPOLL_EVENTS];
> +		int nfds;
> +
> +		/* NOLINTBEGIN(bugprone-branch-clone): intervals can be the same */
> +		/* cppcheck-suppress [duplicateValueTernary, unmatchedSuppression] */
> +		nfds = epoll_wait(tc->epollfd, events, NUM_EPOLL_EVENTS,
> +				  TIMER_INTERVAL);
> +		/* NOLINTEND(bugprone-branch-clone) */
> +		if (nfds == -1 && errno != EINTR)
> +			die_perror("epoll_wait() failed in thread loop");
> +
> +		tc->worker(tc->opaque, nfds, events);
> +	}
> +
> +	return NULL;
> +}
> +
> +/**
> + * threading_start_thread() - Start a worker thread
> + * @threadid:	Thread index to start
> + *
> + * #syscalls rt_sigaction rt_sigprocmask mprotect getrandom brk clone3 rseq set_robust_list clock_nanosleep
> + */
> +/* cppcheck-suppress unusedFunction */
> +void threading_start_thread(unsigned int threadid)
> +{
> +	struct threading_context *tc;
> +	int ret;
> +
> +	if (threadid >= ARRAY_SIZE(threads))
> +		die_perror("Invalid thread index %u, max is %u\n",
> +			   threadid, MAX_NUM_THREADS);

assert?

> +	tc = &threads[threadid];
> +
> +	if (!tc->worker)
> +		die_perror("No worker for thread #%u\n", threadid);
> +
> +	if (tc->pthread)
> +		return;
> +
> +	/* thread #0 is the main process */
> +	if (threadid == THREADING_ID_DEFAULT) {
> +		tc->pthread = -1; /* Mark thread is active */
> +		tc->quit = false;
> +		/* Thread #0 is main process, call is blocking */
> +		threading_worker(tc);	/* no return */
> +	}
> +
> +	tc->quit = false;
> +	ret = pthread_create(&tc->pthread, NULL, &threading_worker, tc);
> +	if (ret != 0)
> +		die_perror("pthread_create() failed: thread index %u ret %d",
> +			   threadid, ret);
> +}
> +
> +/**
> + * 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))
> +		die_perror("Invalid thread index %u, max is %u\n",
> +		           threadid, ARRAY_SIZE(threads));
> +
> +	if (threadid == THREADING_ID_DEFAULT) {
> +		/* Thread #0 is main process, cannot be stopped */
> +		return;
> +	}
> +
> +	threads[threadid].quit = true;

You're writing this from one thread, then reading it from another.
Don't you need some sort of memory barrier?

> +	pthread_join(threads[threadid].pthread, NULL);
> +	threads[threadid].pthread = 0;
> +}
> +
> +/**
> + * threading_is_active() - Check if a worker thread is active
> + * @threadid:	Thread index to check
> + *
> + * 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))
> +		die_perror("Invalid thread index %u, max is %u\n",
> +		           threadid, ARRAY_SIZE(threads));
> +
> +	return !!threads[threadid].pthread;
> +}
> diff --git a/threading.h b/threading.h
> new file mode 100644
> index 000000000000..88f397a80441
> --- /dev/null
> +++ b/threading.h
> @@ -0,0 +1,22 @@
> +/* SPDX-License-Identifier: GPL-2.0-or-later
> + * Copyright (c) 2025 Red Hat
> + * Author: Laurent Vivier <lvivier@redhat.com>
> + */
> +
> +#ifndef THREADING_H
> +#define THREADING_H
> +
> +#define MAX_NUM_THREADS 32

AIUI, the threading model you have in mind is pretty firmly tied to
the queue pairs.  So, is there any point defining this separately from
the max number of qpairs?

> +#define THREADING_ID_DEFAULT	0
> +
> +void threading_init(void);
> +int threading_worker_set(unsigned int threadid,
> +			 void (*worker)(void *, int, struct epoll_event *),
> +			 void *opaque);
> +int threading_epollfd(unsigned int threadid);
> +void threading_start_thread(unsigned int threadid);
> +void threading_stop_thread(unsigned int threadid);
> +bool threading_is_active(unsigned int threadid);
> +
> +#endif /* THREADING_H */
> -- 
> 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

[-- Attachment #2: signature.asc --]
[-- Type: application/pgp-signature, Size: 833 bytes --]