* [PATCH 1/7] vhost_user: Reset vq enable flag in vu_cleanup()
2026-07-31 16:46 [PATCH 0/7] multithreading: Add worker threads for queue pair processing Laurent Vivier
@ 2026-07-31 16:46 ` Laurent Vivier
2026-07-31 16:46 ` [PATCH 2/7] threading: Add basic threading infrastructure Laurent Vivier
` (5 subsequent siblings)
6 siblings, 0 replies; 8+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:46 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
vu_cleanup() resets most virtqueue state (started, notification,
file descriptors) but does not reset the enable flag. After a
QEMU disconnect and reconnect, the stale enable flag causes
vu_set_vring_enable_exec() to hit its early return check
(vq->enable == enable), skipping the threading_start_thread() call
and leaving the worker thread stopped.
Reset vq->enable to false in vu_cleanup() alongside the other
virtqueue state.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
vhost_user.c | 1 +
1 file changed, 1 insertion(+)
diff --git a/vhost_user.c b/vhost_user.c
index 9e382c21f963..8ba3949e4e9d 100644
--- a/vhost_user.c
+++ b/vhost_user.c
@@ -1083,6 +1083,7 @@ void vu_cleanup(struct vu_dev *vdev)
struct vu_virtq *vq = &vdev->vq[i];
vq->started = false;
+ vq->enable = false;
vq->notification = true;
if (vq->call_fd != -1) {
--
2.54.0
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 2/7] threading: Add basic threading infrastructure
2026-07-31 16:46 [PATCH 0/7] multithreading: Add worker threads for queue pair processing Laurent Vivier
2026-07-31 16:46 ` [PATCH 1/7] vhost_user: Reset vq enable flag in vu_cleanup() Laurent Vivier
@ 2026-07-31 16:46 ` Laurent Vivier
2026-07-31 16:46 ` [PATCH 3/7] passt: Integrate main event loop with " Laurent Vivier
` (4 subsequent siblings)
6 siblings, 0 replies; 8+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:46 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
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
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 3/7] passt: Integrate main event loop with threading infrastructure
2026-07-31 16:46 [PATCH 0/7] multithreading: Add worker threads for queue pair processing Laurent Vivier
2026-07-31 16:46 ` [PATCH 1/7] vhost_user: Reset vq enable flag in vu_cleanup() Laurent Vivier
2026-07-31 16:46 ` [PATCH 2/7] threading: Add basic threading infrastructure Laurent Vivier
@ 2026-07-31 16:46 ` Laurent Vivier
2026-07-31 16:46 ` [PATCH 4/7] flow: Delegate epoll file descriptor management to threading subsystem Laurent Vivier
` (3 subsequent siblings)
6 siblings, 0 replies; 8+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:46 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
Convert the main event loop to use the threading subsystem. The main
process now registers passt_worker() as thread #0 and starts it through
the threading infrastructure instead of running a manual epoll_wait loop.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
passt.c | 20 ++++----------------
threading.c | 3 ---
2 files changed, 4 insertions(+), 19 deletions(-)
diff --git a/passt.c b/passt.c
index 8a06838c5ed8..2cedb7ba0756 100644
--- a/passt.c
+++ b/passt.c
@@ -333,8 +333,7 @@ static void passt_worker(void *opaque, int nfds, struct epoll_event *events)
*/
int main(int argc, char **argv)
{
- struct epoll_event events[NUM_EPOLL_EVENTS];
- int nfds, devnull_fd = -1, fd;
+ int devnull_fd = -1, fd;
struct ctx *c = &passt_ctx;
struct rlimit limit;
struct timespec now;
@@ -377,9 +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");
+ c->epollfd = threading_epollfd(THREADING_ID_DEFAULT);
if (getrlimit(RLIMIT_NOFILE, &limit))
die_perror("Failed to get maximum value of open files limit");
@@ -449,15 +446,6 @@ int main(int argc, char **argv)
isolate_postfork(c);
-loop:
- /* NOLINTBEGIN(bugprone-branch-clone): intervals can be the same */
- /* cppcheck-suppress [duplicateValueTernary, unmatchedSuppression] */
- nfds = epoll_wait(c->epollfd, events, NUM_EPOLL_EVENTS, TIMER_INTERVAL);
- /* NOLINTEND(bugprone-branch-clone) */
- if (nfds == -1 && errno != EINTR)
- die_perror("epoll_wait() failed in main loop");
-
- passt_worker(c, nfds, events);
-
- goto loop;
+ threading_worker_set(THREADING_ID_DEFAULT, passt_worker, c);
+ threading_start_thread(THREADING_ID_DEFAULT);
}
diff --git a/threading.c b/threading.c
index fd5b30ed7ff6..e1ed644f5864 100644
--- a/threading.c
+++ b/threading.c
@@ -66,7 +66,6 @@ void threading_init(void)
*
* 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)
@@ -90,7 +89,6 @@ int threading_worker_set(unsigned int threadid,
*
* 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))
@@ -135,7 +133,6 @@ static void *threading_worker(void *opaque)
*
* #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;
--
2.54.0
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 4/7] flow: Delegate epoll file descriptor management to threading subsystem
2026-07-31 16:46 [PATCH 0/7] multithreading: Add worker threads for queue pair processing Laurent Vivier
` (2 preceding siblings ...)
2026-07-31 16:46 ` [PATCH 3/7] passt: Integrate main event loop with " Laurent Vivier
@ 2026-07-31 16:46 ` Laurent Vivier
2026-07-31 16:46 ` [PATCH 5/7] ctx: Remove epollfd from context structure Laurent Vivier
` (2 subsequent siblings)
6 siblings, 0 replies; 8+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:46 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
Remove the flow-local epoll_id_to_fd mapping array and instead
rely on the threading subsystem to provide the epoll file descriptor
for a given thread number.
Update all protocol handlers (ICMP, TCP, TCP splice, UDP).
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
flow.c | 17 +++++++++--------
flow.h | 4 +---
passt.c | 2 +-
tcp.c | 3 ++-
4 files changed, 13 insertions(+), 13 deletions(-)
diff --git a/flow.c b/flow.c
index 3012a07ed680..7821b3f91285 100644
--- a/flow.c
+++ b/flow.c
@@ -24,6 +24,7 @@
#include "repair.h"
#include "epoll_ctl.h"
#include "serialise.h"
+#include "threading.h"
const char *flow_state_str[] = {
[FLOW_STATE_FREE] = "FREE",
@@ -221,7 +222,6 @@ static_assert(ARRAY_SIZE(flow_epoll) == FLOW_NUM_TYPES,
unsigned flow_first_free;
union flow flowtab[FLOW_MAX];
static _Thread_local const union flow *flow_new_entry; /* = NULL */
-int qpair_to_fd[FLOW_QPAIR_SIZE];
/* Hash table to index it */
#define FLOW_HASH_LOAD 70 /* % */
@@ -457,7 +457,12 @@ static void flow_set_state(struct flow_common *f, enum flow_state state)
*/
int flow_epollfd(const struct flow_common *f)
{
- return qpair_to_fd[f->qpair];
+ /* 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);
}
/**
@@ -525,7 +530,7 @@ bool flow_migrate_epollfd(struct flow_common *f, unsigned int qpair,
flow_trace((union flow *)f,
"migrating from qpair %d to %d", qpair, target);
- epoll_del(qpair_to_fd[qpair], ref.fd);
+ epoll_del(threading_epollfd(qpair), ref.fd);
flow_setqp(f, target);
flow_epoll_set(f, EPOLL_CTL_ADD, events, ref.fd, ref.flowside.sidei);
ret = true;
@@ -1444,9 +1449,8 @@ int flow_migrate_target(struct ctx *c, const struct migrate_stage *stage,
/**
* flow_init() - Initialise flow related data structures
- * @c: Execution context
*/
-void flow_init(const struct ctx *c)
+void flow_init(void)
{
unsigned b;
@@ -1456,7 +1460,4 @@ void flow_init(const struct ctx *c)
for (b = 0; b < FLOW_HASH_SIZE; b++)
flow_hashtab[b] = FLOW_SIDX_NONE;
-
- for (b = 0; b < FLOW_QPAIR_SIZE; b++)
- qpair_to_fd[b] = c->epollfd;
}
diff --git a/flow.h b/flow.h
index c31a51a9cc96..e2693efd0f70 100644
--- a/flow.h
+++ b/flow.h
@@ -157,8 +157,6 @@ struct flowside {
in_port_t eport;
};
-extern int qpair_to_fd[];
-
/**
* flowside_eq() - Check if two flowsides are equal
* @left, @right: Flowsides to compare
@@ -266,7 +264,7 @@ flow_sidx_t flow_lookup_sa(const struct ctx *c, uint8_t proto, uint8_t pif,
union flow;
-void flow_init(const struct ctx *c);
+void flow_init(void);
int flow_epollfd(const struct flow_common *f);
int flow_epoll_set(const struct flow_common *f, int command, uint32_t events,
int fd, unsigned int sidei);
diff --git a/passt.c b/passt.c
index 2cedb7ba0756..57290a5a7c6e 100644
--- a/passt.c
+++ b/passt.c
@@ -399,7 +399,7 @@ int main(int argc, char **argv)
if (clock_gettime(CLOCK_MONOTONIC, &now))
die_perror("Failed to get CLOCK_MONOTONIC time");
- flow_init(c);
+ flow_init();
fwd_scan_ports_init(c);
if ((!c->no_udp && udp_init(c)) || (!c->no_tcp && tcp_init(c)))
diff --git a/tcp.c b/tcp.c
index 2767a8494107..e8da65410bcf 100644
--- a/tcp.c
+++ b/tcp.c
@@ -317,6 +317,7 @@
#include "tcp_buf.h"
#include "tcp_vu.h"
#include "epoll_ctl.h"
+#include "threading.h"
/*
* The size of TCP header (including options) is given by doff (Data Offset)
@@ -2721,7 +2722,7 @@ void tcp_timer_handler(const struct ctx *c, union epoll_ref ref,
assert(conn->f.type == FLOW_TCP);
if (conn->f.qpair != qpair) {
- int old_epollfd = qpair_to_fd[qpair];
+ int old_epollfd = threading_epollfd(qpair);
epoll_del(old_epollfd, conn->timer);
if (tcp_timer_epoll_add(conn, conn->timer, now) < 0) {
--
2.54.0
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 5/7] ctx: Remove epollfd from context structure
2026-07-31 16:46 [PATCH 0/7] multithreading: Add worker threads for queue pair processing Laurent Vivier
` (3 preceding siblings ...)
2026-07-31 16:46 ` [PATCH 4/7] flow: Delegate epoll file descriptor management to threading subsystem Laurent Vivier
@ 2026-07-31 16:46 ` Laurent Vivier
2026-07-31 16:46 ` [PATCH 6/7] vhost-user: Add per-qpair worker threads Laurent Vivier
2026-07-31 16:46 ` [PATCH 7/7] virtio: Prevent crash on virtqueue exhaustion with temporary workaround Laurent Vivier
6 siblings, 0 replies; 8+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:46 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
The main epoll file descriptor is now managed by the threading subsystem,
so there's no need to store it in the context structure.
All code that previously accessed c->epollfd now calls
threading_epollfd(THREADING_ID_DEFAULT) to get the main event loop's
epoll file descriptor.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
conf.c | 10 +++++++---
fwd.c | 4 +++-
netlink.c | 4 +++-
passt.c | 1 -
passt.h | 2 --
pasta.c | 3 ++-
pif.c | 3 ++-
repair.c | 12 ++++++++----
tap.c | 9 ++++++---
tcp_splice.c | 1 +
vhost_user.c | 9 +++++----
11 files changed, 37 insertions(+), 21 deletions(-)
diff --git a/conf.c b/conf.c
index faf268140321..733bad4228cb 100644
--- a/conf.c
+++ b/conf.c
@@ -53,6 +53,7 @@
#include "pesto.h"
#include "serialise.h"
#include "parse.h"
+#include "threading.h"
#define NETNS_RUN_DIR "/run/netns"
@@ -1164,7 +1165,8 @@ static void conf_sock_listen(const struct ctx *c)
die_perror("Couldn't listen on configuration socket");
ref.fd = c->fd_control_listen;
- if (epoll_add(c->epollfd, EPOLLIN | EPOLLET, ref))
+ if (epoll_add(threading_epollfd(THREADING_ID_DEFAULT),
+ EPOLLIN | EPOLLET, ref))
die_perror("Couldn't add configuration socket to epoll");
}
@@ -2117,7 +2119,8 @@ static int conf_recv_rules(const struct ctx *c, int fd)
static void conf_close(struct ctx *c)
{
debug("Closing configuration socket");
- epoll_ctl(c->epollfd, EPOLL_CTL_DEL, c->fd_control, NULL);
+ epoll_ctl(threading_epollfd(THREADING_ID_DEFAULT), EPOLL_CTL_DEL,
+ c->fd_control, NULL);
close(c->fd_control);
c->fd_control = -1;
}
@@ -2180,7 +2183,8 @@ retry:
warn_perror("Can't get configuration client credentials");
c->fd_control = ref.fd = fd;
- rc = epoll_add(c->epollfd, EPOLLIN | EPOLLET, ref);
+ rc = epoll_add(threading_epollfd(THREADING_ID_DEFAULT),
+ EPOLLIN | EPOLLET, ref);
if (rc < 0) {
warn_perror("epoll_ctl() on configuration socket");
goto fail;
diff --git a/fwd.c b/fwd.c
index 5679a3d606ea..fa6679ea1b2e 100644
--- a/fwd.c
+++ b/fwd.c
@@ -33,6 +33,7 @@
#include "netlink.h"
#include "arp.h"
#include "ndp.h"
+#include "threading.h"
#define NEIGH_TABLE_SLOTS 1024
#define NEIGH_TABLE_SIZE (NEIGH_TABLE_SLOTS / 2)
@@ -377,7 +378,8 @@ static int fwd_sync_one(const struct ctx *c, uint8_t pif, unsigned idx,
/* We don't want to listen on this port */
if (fd >= 0) {
/* We already are, so stop */
- epoll_del(c->epollfd, fd);
+ epoll_del(threading_epollfd(THREADING_ID_DEFAULT),
+ fd);
close(fd);
socks[port - rule->first] = -1;
}
diff --git a/netlink.c b/netlink.c
index 8d20dbb57fb8..8fffe490b6f3 100644
--- a/netlink.c
+++ b/netlink.c
@@ -37,6 +37,7 @@
#include "ip.h"
#include "netlink.h"
#include "epoll_ctl.h"
+#include "threading.h"
/* Same as RTA_NEXT() but for nexthops: RTNH_NEXT() doesn't take 'attrlen' */
#define RTNH_NEXT_AND_DEC(rtnh, attrlen) \
@@ -1292,7 +1293,8 @@ int nl_neigh_notify_init(const struct ctx *c)
}
ev.data.u64 = ref.u64;
- if (epoll_ctl(c->epollfd, EPOLL_CTL_ADD, nl_sock_neigh, &ev) == -1) {
+ if (epoll_ctl(threading_epollfd(THREADING_ID_DEFAULT), EPOLL_CTL_ADD,
+ nl_sock_neigh, &ev) == -1) {
warn_perror("epoll_ctl() on neighbour notifier socket failed");
close(nl_sock_neigh);
nl_sock_neigh = -1;
diff --git a/passt.c b/passt.c
index 57290a5a7c6e..469e2525f545 100644
--- a/passt.c
+++ b/passt.c
@@ -376,7 +376,6 @@ int main(int argc, char **argv)
madvise(pkt_buf, sizeof(pkt_buf), MADV_HUGEPAGE);
threading_init();
- c->epollfd = threading_epollfd(THREADING_ID_DEFAULT);
if (getrlimit(RLIMIT_NOFILE, &limit))
die_perror("Failed to get maximum value of open files limit");
diff --git a/passt.h b/passt.h
index 00ef5a8f85e5..40abc5995ffd 100644
--- a/passt.h
+++ b/passt.h
@@ -190,7 +190,6 @@ struct ip6_ctx {
* @no_netns_quit: In pasta mode, don't exit if fs-bound namespace is gone
* @netns_base: Base name for fs-bound namespace, if any, in pasta mode
* @netns_dir: Directory of fs-bound namespace, if any, in pasta mode
- * @epollfd: File descriptor for epoll instance
* @fd_tap_listen: File descriptor for listening AF_UNIX socket, if any
* @fd_tap: AF_UNIX socket, tuntap device, or pre-opened socket
* @fd_control_listen: Listening control/configuration socket, if any
@@ -266,7 +265,6 @@ struct ctx {
char netns_base[PATH_MAX];
char netns_dir[PATH_MAX];
- int epollfd;
int fd_tap_listen;
int fd_tap;
int fd_control_listen;
diff --git a/pasta.c b/pasta.c
index 5aa56b78d262..0a742a4f9abe 100644
--- a/pasta.c
+++ b/pasta.c
@@ -50,6 +50,7 @@
#include "netlink.h"
#include "log.h"
#include "epoll_ctl.h"
+#include "threading.h"
#define HOSTNAME_PREFIX "pasta-"
@@ -517,7 +518,7 @@ void pasta_netns_quit_init(const struct ctx *c)
ref.fd = fd;
- epoll_add(c->epollfd, EPOLLIN, ref);
+ epoll_add(threading_epollfd(THREADING_ID_DEFAULT), EPOLLIN, ref);
}
/**
diff --git a/pif.c b/pif.c
index 8ade587c623a..e0a36413878b 100644
--- a/pif.c
+++ b/pif.c
@@ -16,6 +16,7 @@
#include "ip.h"
#include "inany.h"
#include "epoll_ctl.h"
+#include "threading.h"
const char pif_type_str[][PIF_NAME_SIZE] = {
[PIF_NONE] = "<none>",
@@ -130,7 +131,7 @@ int pif_listen(const struct ctx *c, uint8_t proto, uint8_t pif,
goto fail;
}
- ret = epoll_add(c->epollfd, EPOLLIN, ref);
+ ret = epoll_add(threading_epollfd(THREADING_ID_DEFAULT), EPOLLIN, ref);
if (ret < 0)
goto fail;
diff --git a/repair.c b/repair.c
index f31ccceed7c6..7560242cf1c1 100644
--- a/repair.c
+++ b/repair.c
@@ -23,6 +23,7 @@
#include "flow.h"
#include "flow_table.h"
#include "epoll_ctl.h"
+#include "threading.h"
#include "repair.h"
@@ -58,8 +59,9 @@ void repair_sock_init(const struct ctx *c)
}
ref.fd = c->fd_repair_listen;
- if (epoll_add(c->epollfd, EPOLLIN | EPOLLHUP | EPOLLET, ref))
- err("repair helper socket epoll_ctl(), won't migrate");
+ if (epoll_add(threading_epollfd(THREADING_ID_DEFAULT),
+ EPOLLIN | EPOLLHUP | EPOLLET, ref))
+ err_perror("repair helper socket epoll_ctl(), won't migrate");
}
/**
@@ -115,7 +117,8 @@ int repair_listen_handler(struct ctx *c, uint32_t events)
ref.fd = c->fd_repair;
- rc = epoll_add(c->epollfd, EPOLLHUP | EPOLLET, ref);
+ rc = epoll_add(threading_epollfd(THREADING_ID_DEFAULT),
+ EPOLLHUP | EPOLLET, ref);
if (rc < 0) {
debug("epoll_ctl() on TCP_REPAIR helper socket");
close(c->fd_repair);
@@ -134,7 +137,8 @@ void repair_close(struct ctx *c)
{
debug("Closing TCP_REPAIR helper socket");
- epoll_ctl(c->epollfd, EPOLL_CTL_DEL, c->fd_repair, NULL);
+ epoll_ctl(threading_epollfd(THREADING_ID_DEFAULT), EPOLL_CTL_DEL,
+ c->fd_repair, NULL);
close(c->fd_repair);
c->fd_repair = -1;
}
diff --git a/tap.c b/tap.c
index 25cde67538ac..d9be6bcc41a0 100644
--- a/tap.c
+++ b/tap.c
@@ -61,6 +61,7 @@
#include "vhost_user.h"
#include "vu_common.h"
#include "epoll_ctl.h"
+#include "threading.h"
/* Maximum allowed frame lengths (including L2 header) */
@@ -1226,7 +1227,7 @@ void tap_sock_reset(struct ctx *c)
passt_exit(EXIT_SUCCESS);
/* Close the connected socket, wait for a new connection */
- epoll_del(c->epollfd, c->fd_tap);
+ epoll_del(threading_epollfd(THREADING_ID_DEFAULT), c->fd_tap);
close(c->fd_tap);
c->fd_tap = -1;
if (c->mode == MODE_VU)
@@ -1417,7 +1418,8 @@ static void tap_sock_unix_init(const struct ctx *c)
ref.fd = c->fd_tap_listen;
- epoll_add(c->epollfd, EPOLLIN | EPOLLET, ref);
+ epoll_add(threading_epollfd(THREADING_ID_DEFAULT), EPOLLIN | EPOLLET,
+ ref);
}
/**
@@ -1465,7 +1467,8 @@ static void tap_start_connection(const struct ctx *c)
break;
}
- epoll_add(c->epollfd, EPOLLIN | EPOLLRDHUP, ref);
+ epoll_add(threading_epollfd(THREADING_ID_DEFAULT),
+ EPOLLIN | EPOLLRDHUP, ref);
if (!tap_is_ready(c))
return;
diff --git a/tcp_splice.c b/tcp_splice.c
index 8617e62cea9a..fc5b6c25670d 100644
--- a/tcp_splice.c
+++ b/tcp_splice.c
@@ -56,6 +56,7 @@
#include "inany.h"
#include "flow.h"
#include "epoll_ctl.h"
+#include "threading.h"
#include "flow_table.h"
diff --git a/vhost_user.c b/vhost_user.c
index 8ba3949e4e9d..ad316aee12c3 100644
--- a/vhost_user.c
+++ b/vhost_user.c
@@ -44,6 +44,7 @@
#include "pcap.h"
#include "migrate.h"
#include "epoll_ctl.h"
+#include "threading.h"
/* vhost-user version we are compatible with */
#define VHOST_USER_VERSION 1
@@ -736,7 +737,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(vdev->context->epollfd, vdev->vq[idx].kick_fd);
+ epoll_del(threading_epollfd(THREADING_ID_DEFAULT), vdev->vq[idx].kick_fd);
close(vdev->vq[idx].kick_fd);
vdev->vq[idx].kick_fd = -1;
}
@@ -757,7 +758,7 @@ static void vu_set_watch(const struct vu_dev *vdev, int idx)
.queue = idx
};
- epoll_add(vdev->context->epollfd, EPOLLIN, ref);
+ epoll_add(threading_epollfd(THREADING_ID_DEFAULT), EPOLLIN, ref);
}
/**
@@ -802,7 +803,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(vdev->context->epollfd, vdev->vq[idx].kick_fd);
+ epoll_del(threading_epollfd(THREADING_ID_DEFAULT), vdev->vq[idx].kick_fd);
close(vdev->vq[idx].kick_fd);
vdev->vq[idx].kick_fd = -1;
}
@@ -1095,7 +1096,7 @@ void vu_cleanup(struct vu_dev *vdev)
vq->err_fd = -1;
}
if (vq->kick_fd != -1) {
- epoll_del(vdev->context->epollfd, vq->kick_fd);
+ epoll_del(threading_epollfd(THREADING_ID_DEFAULT), vq->kick_fd);
close(vq->kick_fd);
vq->kick_fd = -1;
}
--
2.54.0
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 6/7] vhost-user: Add per-qpair worker threads
2026-07-31 16:46 [PATCH 0/7] multithreading: Add worker threads for queue pair processing Laurent Vivier
` (4 preceding siblings ...)
2026-07-31 16:46 ` [PATCH 5/7] ctx: Remove epollfd from context structure Laurent Vivier
@ 2026-07-31 16:46 ` Laurent Vivier
2026-07-31 16:46 ` [PATCH 7/7] virtio: Prevent crash on virtqueue exhaustion with temporary workaround Laurent Vivier
6 siblings, 0 replies; 8+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:46 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
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];
+
/**
* 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;
/* 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]);
+
+ 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;
+ 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);
+ }
+
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
*/
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));
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);
}
count++;
}
- tap_handler(vdev->context, QPAIR_DEFAULT, now);
+ tap_handler(vdev->context, QPAIR_FROM_QUEUE(index), now);
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;
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);
}
/**
--
2.54.0
^ permalink raw reply [flat|nested] 8+ messages in thread* [PATCH 7/7] virtio: Prevent crash on virtqueue exhaustion with temporary workaround
2026-07-31 16:46 [PATCH 0/7] multithreading: Add worker threads for queue pair processing Laurent Vivier
` (5 preceding siblings ...)
2026-07-31 16:46 ` [PATCH 6/7] vhost-user: Add per-qpair worker threads Laurent Vivier
@ 2026-07-31 16:46 ` Laurent Vivier
6 siblings, 0 replies; 8+ messages in thread
From: Laurent Vivier @ 2026-07-31 16:46 UTC (permalink / raw)
To: passt-dev; +Cc: Laurent Vivier
When vq->inuse reaches vq->vring.num, instead of calling die(), yield
and retry. This is a temporary hack to avoid crashes when multiple
worker threads try to pop from an exhausted virtqueue.
The proper fix will involve coordination between worker threads to
prevent this condition entirely, but this keeps the system running for
now during multithreading development.
Signed-off-by: Laurent Vivier <lvivier@redhat.com>
---
virtio.c | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git a/virtio.c b/virtio.c
index d7016cc3d580..21bdcb3c9a4d 100644
--- a/virtio.c
+++ b/virtio.c
@@ -77,6 +77,7 @@
#include <endian.h>
#include <string.h>
#include <errno.h>
+#include <sched.h>
#include <sys/eventfd.h>
#include <sys/socket.h>
@@ -531,6 +532,7 @@ int vu_queue_pop(const struct vu_dev *dev, struct vu_virtq *vq,
unsigned int head;
int ret;
+again:
if (!vq->vring.avail)
return -1;
@@ -542,8 +544,10 @@ int vu_queue_pop(const struct vu_dev *dev, struct vu_virtq *vq,
*/
smp_rmb();
- if (vq->inuse >= vq->vring.num)
- die("vhost-user queue size exceeded");
+ if (vq->inuse >= vq->vring.num) {
+ sched_yield();
+ goto again;
+ }
virtqueue_get_head(vq, vq->last_avail_idx++, &head);
--
2.54.0
^ permalink raw reply [flat|nested] 8+ messages in thread