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=a2kzyM3F; dkim-atps=neutral Received: from us-smtp-delivery-124.mimecast.com (us-smtp-delivery-124.mimecast.com [170.10.133.124]) by passt.top (Postfix) with ESMTPS id E7C5F5A061D for ; Fri, 31 Jul 2026 18:46:40 +0200 (CEST) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=redhat.com; s=mimecast20190719; t=1785516399; 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=a09TPSmRNJ5brMTOYQEUorgFklDxXwPlFD+jZo/0hiY=; b=a2kzyM3FhVsUxMFqIt5iV8WamWuro7YlEFT64CWECwsTMhQ4oWviItqF3Xav1Hix6l+YxK eauxgrOPusJ3N9++xqSbDY/V99C22kXopqh7GrwS56bqSLUNXT+Wv8/nR0yvAF8arHvE9T 8gdDKNaH6FGaPXzrRv+82Lx2+Nby290= 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-344-R4oC6_MWO9Od3LshvqGdcA-1; Fri, 31 Jul 2026 12:46:38 -0400 X-MC-Unique: R4oC6_MWO9Od3LshvqGdcA-1 X-Mimecast-MFC-AGG-ID: R4oC6_MWO9Od3LshvqGdcA_1785516397 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 355071800350 for ; Fri, 31 Jul 2026 16:46:37 +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 74EA3423; Fri, 31 Jul 2026 16:46:36 +0000 (UTC) From: Laurent Vivier To: passt-dev@passt.top Subject: [PATCH 6/7] vhost-user: Add per-qpair worker threads Date: Fri, 31 Jul 2026 18:46:27 +0200 Message-ID: <20260731164628.3556997-7-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: hFPCPVfNh2EZTKhglT-_sDwLuLr01baW2ARP3h-zVMY_1785516397 X-Mimecast-Originator: redhat.com Content-Transfer-Encoding: 8bit content-type: text/plain; charset="US-ASCII"; x-default=true Message-ID-Hash: 5JMT42GOE6YRCV3AYYJANSNI7GZFF4NO X-Message-ID-Hash: 5JMT42GOE6YRCV3AYYJANSNI7GZFF4NO 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: 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