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

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 --- 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

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 --- 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

On Fri, Jul 31, 2026 at 06:46:22PM +0200, Laurent Vivier wrote:

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

Reviewed-by: David Gibson AFAICT it would make sense to apply this fix immediately, independent of the rest of the series.

--- 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

-- 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

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

On Fri, Jul 31, 2026 at 06:46:24PM +0200, Laurent Vivier wrote:

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 --- 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);

Invoking this through threading_start_thread(), which then special cases thread 0 seems a bit odd. Could you just directly call threading_worker()?

} 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

-- 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