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

[Laurent Vivier <lvivier@redhat.com>]

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))
 	 */
-	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
+ * @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 <lvivier@redhat.com>
+ */
+
+#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