From mboxrd@z Thu Jan 1 00:00:00 1970 Authentication-Results: passt.top; dmarc=pass (p=quarantine dis=none) header.from=redhat.com Authentication-Results: passt.top; dkim=pass (1024-bit key; unprotected) header.d=redhat.com header.i=@redhat.com header.a=rsa-sha256 header.s=mimecast20190719 header.b=Kd858obM; dkim-atps=neutral Received: from us-smtp-delivery-124.mimecast.com (us-smtp-delivery-124.mimecast.com [170.10.129.124]) by passt.top (Postfix) with ESMTPS id 3B4535A026D for ; Fri, 31 Jul 2026 18:46:36 +0200 (CEST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=redhat.com; s=mimecast20190719; t=1785516395; h=from:from:reply-to:subject:subject:date:date:message-id:message-id: to:to:cc:cc:mime-version:mime-version:content-type:content-type: content-transfer-encoding:content-transfer-encoding: in-reply-to:in-reply-to:references:references; bh=XPXSt6zv9S5yzhAv5y3Vat1rEMuxqguIUmY2DPco4sU=; b=Kd858obM5QbhIQbEogYNntp2Bx+ts+Rt5eds4NgJtZs7YR0KkEQutGJejCdiCdY119ZDf3 HYM8EY/UjlFt4KbAScllJro4H+HHdCx+BPizbLTIoFPw/6gIPY+hi1a3KqJ/U79ZMU4BNE BD3SlmPAhQ3jVTzX0eGsQqqyD5Lbu0c= Received: from mx-prod-mc-06.mail-002.prod.us-west-2.aws.redhat.com (ec2-35-165-154-97.us-west-2.compute.amazonaws.com [35.165.154.97]) by relay.mimecast.com with ESMTP with STARTTLS (version=TLSv1.3, cipher=TLS_AES_256_GCM_SHA384) id us-mta-581-pL0X8naSOSWBysyNCjIxGA-1; Fri, 31 Jul 2026 12:46:33 -0400 X-MC-Unique: pL0X8naSOSWBysyNCjIxGA-1 X-Mimecast-MFC-AGG-ID: pL0X8naSOSWBysyNCjIxGA_1785516392 Received: from mx-prod-int-10.mail-002.prod.us-west-2.aws.redhat.com (mx-prod-int-10.mail-002.prod.us-west-2.aws.redhat.com [10.30.177.95]) (using TLSv1.3 with cipher TLS_AES_256_GCM_SHA384 (256/256 bits) key-exchange X25519 server-signature RSA-PSS (2048 bits) server-digest SHA256) (No client certificate requested) by mx-prod-mc-06.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTPS id C6B091800347 for ; Fri, 31 Jul 2026 16:46:32 +0000 (UTC) Received: from lenovo-t14s.redhat.corp (headnet05.pony-001.prod.iad2.dc.redhat.com [10.2.32.117]) by mx-prod-int-10.mail-002.prod.us-west-2.aws.redhat.com (Postfix) with ESMTP id 11428423; Fri, 31 Jul 2026 16:46:31 +0000 (UTC) From: Laurent Vivier To: passt-dev@passt.top Subject: [PATCH 2/7] threading: Add basic threading infrastructure Date: Fri, 31 Jul 2026 18:46:23 +0200 Message-ID: <20260731164628.3556997-3-lvivier@redhat.com> In-Reply-To: <20260731164628.3556997-1-lvivier@redhat.com> References: <20260731164628.3556997-1-lvivier@redhat.com> MIME-Version: 1.0 X-Scanned-By: MIMEDefang 3.6 on 10.30.177.95 X-Mimecast-Spam-Score: 0 X-Mimecast-MFC-PROC-ID: vCYbLPaI3GXJ0fKw_oBimwOSS5GEdMp7mLz5rgAQVg4_1785516392 X-Mimecast-Originator: redhat.com Content-Transfer-Encoding: 8bit content-type: text/plain; charset="US-ASCII"; x-default=true Message-ID-Hash: 2RM2Y73MMX5KXV7MXTMVBKDX6O56WFQP X-Message-ID-Hash: 2RM2Y73MMX5KXV7MXTMVBKDX6O56WFQP X-MailFrom: lvivier@redhat.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; emergency; loop; banned-address; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header CC: Laurent Vivier X-Mailman-Version: 3.3.8 Precedence: list List-Id: Development discussion and patches for passt Archived-At: Archived-At: List-Archive: List-Archive: List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: 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