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))
*/
- 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
+ * @opaque: Opaque data passed to worker function
+ * @quit: Flag to signal thread to exit
+ */
+struct threading_context {
+ pthread_t pthread;
+ void (*worker)(void *, int, struct epoll_event *);
+ void *opaque;
+ int epollfd;
+ _Atomic bool quit;
+};
+
+/*
+ * 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;
+
+ 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;
+
+ 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);
+
+ 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;
+ 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
+
+#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