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

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 --- 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 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 + */ + +#include +#include +#include + +#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 + */ + +#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