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