From 649936a0daaa5ccd50f66b4218775cfcebaa90a9 Mon Sep 17 00:00:00 2001 From: Juanlu Herrero Date: Wed, 9 Sep 2026 14:51:23 -0500 Subject: [PATCH 1/6] selftests: net: fix get_refill_ring_size() to use its local variable In preparation for multi-threaded rss selftests, fix get_refill_ring_size() to use its local `size` variable instead of assigning to the file-global `ring_size`. Signed-off-by: Juanlu Herrero Link: https://patch.msgid.link/20260909-iou-zcrx-v7-1-6b5d48a2a922@fastmail.com Signed-off-by: Jakub Kicinski --- tools/testing/selftests/drivers/net/hw/iou-zcrx.c | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c index f6a8fc5fac24..df6b62204343 100644 --- a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c +++ b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c @@ -132,10 +132,10 @@ static inline size_t get_refill_ring_size(unsigned int rq_entries) { size_t size; - ring_size = rq_entries * sizeof(struct io_uring_zcrx_rqe); + size = rq_entries * sizeof(struct io_uring_zcrx_rqe); /* add space for the header (head/tail/etc.) */ - ring_size += page_size; - return ALIGN_UP(ring_size, page_size); + size += page_size; + return ALIGN_UP(size, page_size); } static void setup_zcrx(struct io_uring *ring) From 3fb2892382446156ef613603200eab6c82a0a033 Mon Sep 17 00:00:00 2001 From: Juanlu Herrero Date: Wed, 9 Sep 2026 14:51:24 -0500 Subject: [PATCH 2/6] selftests: net: remove unused variable in process_recvzc() Remove the unused `sqe` variable in preparation for the multiqueue rss selftest changes to process_recvzc() in the following commits. Signed-off-by: Juanlu Herrero Link: https://patch.msgid.link/20260909-iou-zcrx-v7-2-6b5d48a2a922@fastmail.com Signed-off-by: Jakub Kicinski --- tools/testing/selftests/drivers/net/hw/iou-zcrx.c | 1 - 1 file changed, 1 deletion(-) diff --git a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c index df6b62204343..c6dbd0ad5368 100644 --- a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c +++ b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c @@ -269,7 +269,6 @@ static void process_recvzc(struct io_uring *ring, struct io_uring_cqe *cqe) unsigned rq_mask = rq_ring.ring_entries - 1; struct io_uring_zcrx_cqe *rcqe; struct io_uring_zcrx_rqe *rqe; - struct io_uring_sqe *sqe; uint64_t mask; char *data; ssize_t n; From 72ac4f0c53987284880e5a94a5c91d71ec5091bf Mon Sep 17 00:00:00 2001 From: Juanlu Herrero Date: Wed, 9 Sep 2026 14:51:25 -0500 Subject: [PATCH 3/6] selftests: net: refactor server state into struct thread_ctx Move server-side state (io_uring ring, zcrx area, refill ring, receive tracking) from global variables into a local struct thread_ctx. This is a pure refactor with no behavior change: run_server still allocates a single context on the stack and runs single-threaded, using io_uring accept and recvzc as before. This prepares the ground for the multithread server support in the following commits, which spawns N worker threads each with their own struct thread_ctx. Reviewed-by: David Wei Signed-off-by: Juanlu Herrero Link: https://patch.msgid.link/20260909-iou-zcrx-v7-3-6b5d48a2a922@fastmail.com Signed-off-by: Jakub Kicinski --- .../selftests/drivers/net/hw/iou-zcrx.c | 156 +++++++++--------- 1 file changed, 80 insertions(+), 76 deletions(-) diff --git a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c index c6dbd0ad5368..9b62fd0703e6 100644 --- a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c +++ b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c @@ -87,14 +87,18 @@ static unsigned int cfg_rx_buf_len; static bool cfg_dry_run; static char *payload; -static void *area_ptr; -static void *ring_ptr; -static size_t ring_size; -static struct io_uring_zcrx_rq rq_ring; -static unsigned long area_token; -static int connfd; -static bool stop; -static size_t received; + +struct thread_ctx { + struct io_uring ring; + void *area_ptr; + void *ring_ptr; + size_t ring_size; + struct io_uring_zcrx_rq rq_ring; + unsigned long area_token; + int connfd; + bool stop; + size_t received; +}; static unsigned long gettimeofday_ms(void) { @@ -138,7 +142,7 @@ static inline size_t get_refill_ring_size(unsigned int rq_entries) return ALIGN_UP(size, page_size); } -static void setup_zcrx(struct io_uring *ring) +static void setup_zcrx(struct thread_ctx *ctx) { unsigned int ifindex; unsigned int rq_entries = 4096; @@ -149,44 +153,44 @@ static void setup_zcrx(struct io_uring *ring) error(1, 0, "bad interface name: %s", cfg_ifname); if (cfg_rx_buf_len && cfg_rx_buf_len != page_size) { - area_ptr = mmap(NULL, - AREA_SIZE, - PROT_READ | PROT_WRITE, - MAP_ANONYMOUS | MAP_PRIVATE | - MAP_HUGETLB | MAP_HUGE_2MB, - -1, - 0); - if (area_ptr == MAP_FAILED) { + ctx->area_ptr = mmap(NULL, + AREA_SIZE, + PROT_READ | PROT_WRITE, + MAP_ANONYMOUS | MAP_PRIVATE | + MAP_HUGETLB | MAP_HUGE_2MB, + -1, + 0); + if (ctx->area_ptr == MAP_FAILED) { printf("Can't allocate huge pages\n"); exit(SKIP_CODE); } } else { - area_ptr = mmap(NULL, - AREA_SIZE, - PROT_READ | PROT_WRITE, - MAP_ANONYMOUS | MAP_PRIVATE, - 0, - 0); - if (area_ptr == MAP_FAILED) + ctx->area_ptr = mmap(NULL, + AREA_SIZE, + PROT_READ | PROT_WRITE, + MAP_ANONYMOUS | MAP_PRIVATE, + 0, + 0); + if (ctx->area_ptr == MAP_FAILED) error(1, 0, "mmap(): zero copy area"); } - ring_size = get_refill_ring_size(rq_entries); - ring_ptr = mmap(NULL, - ring_size, - PROT_READ | PROT_WRITE, - MAP_ANONYMOUS | MAP_PRIVATE, - 0, - 0); + ctx->ring_size = get_refill_ring_size(rq_entries); + ctx->ring_ptr = mmap(NULL, + ctx->ring_size, + PROT_READ | PROT_WRITE, + MAP_ANONYMOUS | MAP_PRIVATE, + 0, + 0); struct io_uring_region_desc region_reg = { - .size = ring_size, - .user_addr = (__u64)(unsigned long)ring_ptr, + .size = ctx->ring_size, + .user_addr = (__u64)(unsigned long)ctx->ring_ptr, .flags = IORING_MEM_REGION_TYPE_USER, }; struct io_uring_zcrx_area_reg area_reg = { - .addr = (__u64)(unsigned long)area_ptr, + .addr = (__u64)(unsigned long)ctx->area_ptr, .len = AREA_SIZE, .flags = 0, }; @@ -200,7 +204,7 @@ static void setup_zcrx(struct io_uring *ring) .rx_buf_len = cfg_rx_buf_len, }; - ret = io_uring_register_ifq(ring, (void *)®); + ret = io_uring_register_ifq(&ctx->ring, (void *)®); if (cfg_rx_buf_len && (ret == -EINVAL || ret == -EOPNOTSUPP || ret == -ERANGE)) { printf("Large chunks are not supported %i\n", ret); @@ -209,64 +213,64 @@ static void setup_zcrx(struct io_uring *ring) error(1, 0, "io_uring_register_ifq(): %d", ret); } - rq_ring.khead = (unsigned int *)((char *)ring_ptr + reg.offsets.head); - rq_ring.ktail = (unsigned int *)((char *)ring_ptr + reg.offsets.tail); - rq_ring.rqes = (struct io_uring_zcrx_rqe *)((char *)ring_ptr + reg.offsets.rqes); - rq_ring.rq_tail = 0; - rq_ring.ring_entries = reg.rq_entries; + ctx->rq_ring.khead = (unsigned int *)((char *)ctx->ring_ptr + reg.offsets.head); + ctx->rq_ring.ktail = (unsigned int *)((char *)ctx->ring_ptr + reg.offsets.tail); + ctx->rq_ring.rqes = (struct io_uring_zcrx_rqe *)((char *)ctx->ring_ptr + reg.offsets.rqes); + ctx->rq_ring.rq_tail = 0; + ctx->rq_ring.ring_entries = reg.rq_entries; - area_token = area_reg.rq_area_token; + ctx->area_token = area_reg.rq_area_token; } -static void add_accept(struct io_uring *ring, int sockfd) +static void add_accept(struct thread_ctx *ctx, int sockfd) { struct io_uring_sqe *sqe; - sqe = io_uring_get_sqe(ring); + sqe = io_uring_get_sqe(&ctx->ring); io_uring_prep_accept(sqe, sockfd, NULL, NULL, 0); sqe->user_data = 1; } -static void add_recvzc(struct io_uring *ring, int sockfd) +static void add_recvzc(struct thread_ctx *ctx, int sockfd) { struct io_uring_sqe *sqe; - sqe = io_uring_get_sqe(ring); + sqe = io_uring_get_sqe(&ctx->ring); io_uring_prep_rw(IORING_OP_RECV_ZC, sqe, sockfd, NULL, 0, 0); sqe->ioprio |= IORING_RECV_MULTISHOT; sqe->user_data = 2; } -static void add_recvzc_oneshot(struct io_uring *ring, int sockfd, size_t len) +static void add_recvzc_oneshot(struct thread_ctx *ctx, int sockfd, size_t len) { struct io_uring_sqe *sqe; - sqe = io_uring_get_sqe(ring); + sqe = io_uring_get_sqe(&ctx->ring); io_uring_prep_rw(IORING_OP_RECV_ZC, sqe, sockfd, NULL, len, 0); sqe->ioprio |= IORING_RECV_MULTISHOT; sqe->user_data = 2; } -static void process_accept(struct io_uring *ring, struct io_uring_cqe *cqe) +static void process_accept(struct thread_ctx *ctx, struct io_uring_cqe *cqe) { if (cqe->res < 0) error(1, 0, "accept()"); - if (connfd) + if (ctx->connfd) error(1, 0, "Unexpected second connection"); - connfd = cqe->res; + ctx->connfd = cqe->res; if (cfg_oneshot) - add_recvzc_oneshot(ring, connfd, page_size); + add_recvzc_oneshot(ctx, ctx->connfd, page_size); else - add_recvzc(ring, connfd); + add_recvzc(ctx, ctx->connfd); } -static void process_recvzc(struct io_uring *ring, struct io_uring_cqe *cqe) +static void process_recvzc(struct thread_ctx *ctx, struct io_uring_cqe *cqe) { - unsigned rq_mask = rq_ring.ring_entries - 1; + unsigned int rq_mask = ctx->rq_ring.ring_entries - 1; struct io_uring_zcrx_cqe *rcqe; struct io_uring_zcrx_rqe *rqe; uint64_t mask; @@ -275,7 +279,7 @@ static void process_recvzc(struct io_uring *ring, struct io_uring_cqe *cqe) int i; if (cqe->res == 0 && cqe->flags == 0 && cfg_oneshot_recvs == 0) { - stop = true; + ctx->stop = true; return; } @@ -284,56 +288,56 @@ static void process_recvzc(struct io_uring *ring, struct io_uring_cqe *cqe) if (cfg_oneshot) { if (cqe->res == 0 && cqe->flags == 0 && cfg_oneshot_recvs) { - add_recvzc_oneshot(ring, connfd, page_size); + add_recvzc_oneshot(ctx, ctx->connfd, page_size); cfg_oneshot_recvs--; } } else if (!(cqe->flags & IORING_CQE_F_MORE)) { - add_recvzc(ring, connfd); + add_recvzc(ctx, ctx->connfd); } rcqe = (struct io_uring_zcrx_cqe *)(cqe + 1); n = cqe->res; mask = (1ULL << IORING_ZCRX_AREA_SHIFT) - 1; - data = (char *)area_ptr + (rcqe->off & mask); + data = (char *)ctx->area_ptr + (rcqe->off & mask); for (i = 0; i < n; i++) { - if (*(data + i) != payload[(received + i)]) + if (*(data + i) != payload[(ctx->received + i)]) error(1, 0, "payload mismatch at %d", i); } - received += n; + ctx->received += n; - rqe = &rq_ring.rqes[(rq_ring.rq_tail & rq_mask)]; - rqe->off = (rcqe->off & ~IORING_ZCRX_AREA_MASK) | area_token; + rqe = &ctx->rq_ring.rqes[(ctx->rq_ring.rq_tail & rq_mask)]; + rqe->off = (rcqe->off & ~IORING_ZCRX_AREA_MASK) | ctx->area_token; rqe->len = cqe->res; - io_uring_smp_store_release(rq_ring.ktail, ++rq_ring.rq_tail); + io_uring_smp_store_release(ctx->rq_ring.ktail, ++ctx->rq_ring.rq_tail); } -static void server_loop(struct io_uring *ring) +static void server_loop(struct thread_ctx *ctx) { struct io_uring_cqe *cqe; unsigned int count = 0; unsigned int head; int i, ret; - io_uring_submit_and_wait(ring, 1); + io_uring_submit_and_wait(&ctx->ring, 1); - io_uring_for_each_cqe(ring, head, cqe) { + io_uring_for_each_cqe(&ctx->ring, head, cqe) { if (cqe->user_data == 1) - process_accept(ring, cqe); + process_accept(ctx, cqe); else if (cqe->user_data == 2) - process_recvzc(ring, cqe); + process_recvzc(ctx, cqe); else error(1, 0, "unknown cqe"); count++; } - io_uring_cq_advance(ring, count); + io_uring_cq_advance(&ctx->ring, count); } static void run_server(void) { + struct thread_ctx ctx = {}; unsigned int flags = 0; - struct io_uring ring; int fd, enable, ret; uint64_t tstop; @@ -356,22 +360,22 @@ static void run_server(void) flags |= IORING_SETUP_SUBMIT_ALL; flags |= IORING_SETUP_CQE32; - io_uring_queue_init(512, &ring, flags); + io_uring_queue_init(512, &ctx.ring, flags); - setup_zcrx(&ring); + setup_zcrx(&ctx); if (cfg_dry_run) return; if (listen(fd, 1024) < 0) error(1, 0, "listen()"); - add_accept(&ring, fd); + add_accept(&ctx, fd); tstop = gettimeofday_ms() + 5000; - while (!stop && gettimeofday_ms() < tstop) - server_loop(&ring); + while (!ctx.stop && gettimeofday_ms() < tstop) + server_loop(&ctx); - if (!stop) + if (!ctx.stop) error(1, 0, "test failed\n"); } From 2801895746294168cbf254d5f96372c092576101 Mon Sep 17 00:00:00 2001 From: Juanlu Herrero Date: Wed, 9 Sep 2026 14:51:26 -0500 Subject: [PATCH 4/6] selftests: net: add multithread client support to iou-zcrx Add pthreads to the iou-zcrx client so that multiple connections can be established simultaneously. Each client thread connects to the server and sends its payload independently. Introduce the -t option to control the number of threads (default 1), preserving backwards compatibility with existing tests. Reviewed-by: David Wei Signed-off-by: Juanlu Herrero Link: https://patch.msgid.link/20260909-iou-zcrx-v7-4-6b5d48a2a922@fastmail.com Signed-off-by: Jakub Kicinski --- .../testing/selftests/drivers/net/hw/Makefile | 2 +- .../selftests/drivers/net/hw/iou-zcrx.c | 35 +++++++++++++++++-- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/tools/testing/selftests/drivers/net/hw/Makefile b/tools/testing/selftests/drivers/net/hw/Makefile index bd3b8d2fa47e..69be4d052ef5 100644 --- a/tools/testing/selftests/drivers/net/hw/Makefile +++ b/tools/testing/selftests/drivers/net/hw/Makefile @@ -97,5 +97,5 @@ include ../../../net/ynl.mk include ../../../net/bpf.mk ifeq ($(HAS_IOURING_ZCRX),y) -$(OUTPUT)/iou-zcrx: LDLIBS += -luring +$(OUTPUT)/iou-zcrx: LDLIBS += -luring -lpthread endif diff --git a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c index 9b62fd0703e6..f793a6c04e41 100644 --- a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c +++ b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -85,6 +86,7 @@ static int cfg_send_size = SEND_SIZE; static struct sockaddr_in6 cfg_addr; static unsigned int cfg_rx_buf_len; static bool cfg_dry_run; +static int cfg_num_threads = 1; static char *payload; @@ -379,7 +381,7 @@ static void run_server(void) error(1, 0, "test failed\n"); } -static void run_client(void) +static void *client_worker(void *arg) { ssize_t to_send = cfg_send_size; ssize_t sent = 0; @@ -405,12 +407,36 @@ static void run_client(void) } close(fd); + return NULL; +} + +static void run_client(void) +{ + int total_conns = cfg_num_threads * cfg_num_threads; + pthread_t *threads; + int i, ret; + + threads = calloc(total_conns, sizeof(*threads)); + if (!threads) + error(1, 0, "calloc()"); + + for (i = 0; i < total_conns; i++) { + ret = pthread_create(&threads[i], NULL, client_worker, NULL); + if (ret) + error(1, ret, "pthread_create()"); + } + + for (i = 0; i < total_conns; i++) + pthread_join(threads[i], NULL); + + free(threads); } static void usage(const char *filepath) { error(1, 0, "Usage: %s (-4|-6) (-s|-c) -h -p " - "-l -i -q", filepath); + "-l -i -q -t", + filepath); } static void parse_opts(int argc, char **argv) @@ -428,7 +454,7 @@ static void parse_opts(int argc, char **argv) usage(argv[0]); cfg_payload_len = max_payload_len; - while ((c = getopt(argc, argv, "sch:p:l:i:q:o:z:x:d")) != -1) { + while ((c = getopt(argc, argv, "sch:p:l:i:q:o:z:x:dt:")) != -1) { switch (c) { case 's': if (cfg_client) @@ -469,6 +495,9 @@ static void parse_opts(int argc, char **argv) case 'd': cfg_dry_run = true; break; + case 't': + cfg_num_threads = strtoul(optarg, NULL, 0); + break; } } From 5d3416ebe55568ae84021606c2b1d8ec0ebd7feb Mon Sep 17 00:00:00 2001 From: Juanlu Herrero Date: Wed, 9 Sep 2026 14:51:27 -0500 Subject: [PATCH 5/6] selftests: net: add multithread server support to iou-zcrx Run the iou-zcrx server as N worker threads, each owning one receive queue with its own io_uring and zero-copy receive (zcrx) ifq, so the test can exercise multi-queue zero-copy receive. The main thread owns the listening socket, accepts connections, and dispatches each to the worker owning the queue it landed on by matching SO_INCOMING_NAPI_ID against per-queue NAPI IDs. Signed-off-by: Juanlu Herrero Link: https://patch.msgid.link/20260909-iou-zcrx-v7-5-6b5d48a2a922@fastmail.com Signed-off-by: Jakub Kicinski --- .../testing/selftests/drivers/net/hw/Makefile | 5 +- .../selftests/drivers/net/hw/iou-zcrx.c | 278 +++++++++++++----- 2 files changed, 210 insertions(+), 73 deletions(-) diff --git a/tools/testing/selftests/drivers/net/hw/Makefile b/tools/testing/selftests/drivers/net/hw/Makefile index 69be4d052ef5..8aebdc6feb17 100644 --- a/tools/testing/selftests/drivers/net/hw/Makefile +++ b/tools/testing/selftests/drivers/net/hw/Makefile @@ -18,10 +18,6 @@ else $(warning excluding iouring tests, liburing not installed or too old) endif -TEST_GEN_FILES := \ - $(COND_GEN_FILES) \ -# end of TEST_GEN_FILES - TEST_PROGS = \ csum.py \ devlink_rate_cross_esw.py \ @@ -80,6 +76,7 @@ TEST_INCLUDES := \ YNL_GEN_FILES := \ ncdevmem \ toeplitz \ + $(COND_GEN_FILES) \ # end of YNL_GEN_FILES TEST_GEN_FILES += $(YNL_GEN_FILES) TEST_GEN_FILES += $(patsubst %.c,%.o,$(wildcard *.bpf.c)) diff --git a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c index f793a6c04e41..91b025a8771e 100644 --- a/tools/testing/selftests/drivers/net/hw/iou-zcrx.c +++ b/tools/testing/selftests/drivers/net/hw/iou-zcrx.c @@ -38,6 +38,8 @@ #include #include +#include +#include "netdev-user.h" #define SKIP_CODE 42 @@ -90,6 +92,8 @@ static int cfg_num_threads = 1; static char *payload; +#define MAX_CONNS_PER_THREAD 64 + struct thread_ctx { struct io_uring ring; void *area_ptr; @@ -97,9 +101,15 @@ struct thread_ctx { size_t ring_size; struct io_uring_zcrx_rq rq_ring; unsigned long area_token; - int connfd; - bool stop; - size_t received; + int queue_id; + int napi_id; + pthread_barrier_t *setup_done; + pthread_barrier_t *dispatch_done; + + int connfds[MAX_CONNS_PER_THREAD]; + size_t received[MAX_CONNS_PER_THREAD]; + int oneshot_recvs[MAX_CONNS_PER_THREAD]; + int nr_conns; }; static unsigned long gettimeofday_ms(void) @@ -146,8 +156,8 @@ static inline size_t get_refill_ring_size(unsigned int rq_entries) static void setup_zcrx(struct thread_ctx *ctx) { + unsigned int rq_entries = AREA_SIZE / page_size; unsigned int ifindex; - unsigned int rq_entries = 4096; int ret; ifindex = if_nametoindex(cfg_ifname); @@ -199,7 +209,7 @@ static void setup_zcrx(struct thread_ctx *ctx) struct t_io_uring_zcrx_ifq_reg reg = { .if_idx = ifindex, - .if_rxq = cfg_queue_id, + .if_rxq = ctx->queue_id, .rq_entries = rq_entries, .area_ptr = (__u64)(unsigned long)&area_reg, .region_ptr = (__u64)(unsigned long)®ion_reg, @@ -224,53 +234,32 @@ static void setup_zcrx(struct thread_ctx *ctx) ctx->area_token = area_reg.rq_area_token; } -static void add_accept(struct thread_ctx *ctx, int sockfd) +static void add_recvzc(struct thread_ctx *ctx, int conn_idx) { struct io_uring_sqe *sqe; sqe = io_uring_get_sqe(&ctx->ring); - io_uring_prep_accept(sqe, sockfd, NULL, NULL, 0); - sqe->user_data = 1; -} - -static void add_recvzc(struct thread_ctx *ctx, int sockfd) -{ - struct io_uring_sqe *sqe; - - sqe = io_uring_get_sqe(&ctx->ring); - - io_uring_prep_rw(IORING_OP_RECV_ZC, sqe, sockfd, NULL, 0, 0); + io_uring_prep_rw(IORING_OP_RECV_ZC, sqe, ctx->connfds[conn_idx], + NULL, 0, 0); sqe->ioprio |= IORING_RECV_MULTISHOT; - sqe->user_data = 2; + sqe->user_data = conn_idx; } -static void add_recvzc_oneshot(struct thread_ctx *ctx, int sockfd, size_t len) +static void add_recvzc_oneshot(struct thread_ctx *ctx, int conn_idx, size_t len) { struct io_uring_sqe *sqe; sqe = io_uring_get_sqe(&ctx->ring); - io_uring_prep_rw(IORING_OP_RECV_ZC, sqe, sockfd, NULL, len, 0); + io_uring_prep_rw(IORING_OP_RECV_ZC, sqe, ctx->connfds[conn_idx], + NULL, len, 0); sqe->ioprio |= IORING_RECV_MULTISHOT; - sqe->user_data = 2; + sqe->user_data = conn_idx; } -static void process_accept(struct thread_ctx *ctx, struct io_uring_cqe *cqe) -{ - if (cqe->res < 0) - error(1, 0, "accept()"); - if (ctx->connfd) - error(1, 0, "Unexpected second connection"); - - ctx->connfd = cqe->res; - if (cfg_oneshot) - add_recvzc_oneshot(ctx, ctx->connfd, page_size); - else - add_recvzc(ctx, ctx->connfd); -} - -static void process_recvzc(struct thread_ctx *ctx, struct io_uring_cqe *cqe) +static void process_recvzc(struct thread_ctx *ctx, struct io_uring_cqe *cqe, + int conn_idx) { unsigned int rq_mask = ctx->rq_ring.ring_entries - 1; struct io_uring_zcrx_cqe *rcqe; @@ -280,8 +269,9 @@ static void process_recvzc(struct thread_ctx *ctx, struct io_uring_cqe *cqe) ssize_t n; int i; - if (cqe->res == 0 && cqe->flags == 0 && cfg_oneshot_recvs == 0) { - ctx->stop = true; + if (cqe->res == 0 && cqe->flags == 0 && + ctx->oneshot_recvs[conn_idx] == 0) { + ctx->nr_conns--; return; } @@ -289,12 +279,14 @@ static void process_recvzc(struct thread_ctx *ctx, struct io_uring_cqe *cqe) error(1, 0, "recvzc(): %d", cqe->res); if (cfg_oneshot) { - if (cqe->res == 0 && cqe->flags == 0 && cfg_oneshot_recvs) { - add_recvzc_oneshot(ctx, ctx->connfd, page_size); - cfg_oneshot_recvs--; + if (cqe->res == 0 && cqe->flags == 0 && + ctx->oneshot_recvs[conn_idx]) { + add_recvzc_oneshot(ctx, conn_idx, page_size); + ctx->oneshot_recvs[conn_idx]--; + return; } } else if (!(cqe->flags & IORING_CQE_F_MORE)) { - add_recvzc(ctx, ctx->connfd); + add_recvzc(ctx, conn_idx); } rcqe = (struct io_uring_zcrx_cqe *)(cqe + 1); @@ -304,10 +296,10 @@ static void process_recvzc(struct thread_ctx *ctx, struct io_uring_cqe *cqe) data = (char *)ctx->area_ptr + (rcqe->off & mask); for (i = 0; i < n; i++) { - if (*(data + i) != payload[(ctx->received + i)]) + if (*(data + i) != payload[(ctx->received[conn_idx] + i)]) error(1, 0, "payload mismatch at %d", i); } - ctx->received += n; + ctx->received[conn_idx] += n; rqe = &ctx->rq_ring.rqes[(ctx->rq_ring.rq_tail & rq_mask)]; rqe->off = (rcqe->off & ~IORING_ZCRX_AREA_MASK) | ctx->area_token; @@ -320,28 +312,137 @@ static void server_loop(struct thread_ctx *ctx) struct io_uring_cqe *cqe; unsigned int count = 0; unsigned int head; - int i, ret; io_uring_submit_and_wait(&ctx->ring, 1); io_uring_for_each_cqe(&ctx->ring, head, cqe) { - if (cqe->user_data == 1) - process_accept(ctx, cqe); - else if (cqe->user_data == 2) - process_recvzc(ctx, cqe); - else - error(1, 0, "unknown cqe"); + process_recvzc(ctx, cqe, cqe->user_data); count++; } io_uring_cq_advance(&ctx->ring, count); } +static void *server_worker(void *arg) +{ + struct io_uring_params params = { }; + struct thread_ctx *ctx = arg; + uint64_t tstop; + int nr_conns; + int i; + + params.flags |= IORING_SETUP_COOP_TASKRUN; + params.flags |= IORING_SETUP_SINGLE_ISSUER; + params.flags |= IORING_SETUP_DEFER_TASKRUN; + params.flags |= IORING_SETUP_SUBMIT_ALL; + params.flags |= IORING_SETUP_CQE32; + params.flags |= IORING_SETUP_CQSIZE; + params.cq_entries = AREA_SIZE / page_size; + + io_uring_queue_init_params(512, &ctx->ring, ¶ms); + setup_zcrx(ctx); + + if (cfg_dry_run) + return NULL; + + pthread_barrier_wait(ctx->setup_done); + pthread_barrier_wait(ctx->dispatch_done); + + nr_conns = ctx->nr_conns; + + for (i = 0; i < ctx->nr_conns; i++) { + if (cfg_oneshot) { + ctx->oneshot_recvs[i] = cfg_oneshot_recvs; + add_recvzc_oneshot(ctx, i, page_size); + } else { + add_recvzc(ctx, i); + } + } + + tstop = gettimeofday_ms() + 5000; + while (ctx->nr_conns > 0 && gettimeofday_ms() < tstop) + server_loop(ctx); + + if (ctx->nr_conns != 0) + error(1, 0, "test failed: %d connections incomplete", + ctx->nr_conns); + + for (i = 0; i < nr_conns; i++) { + if (cfg_oneshot) { + if (!ctx->received[i]) + error(1, 0, "connection %d received no data", i); + } else if (ctx->received[i] != (size_t)cfg_send_size) { + error(1, 0, "connection %d received %zu of %d bytes", + i, ctx->received[i], cfg_send_size); + } + } + + return NULL; +} + +static int query_napi_id(unsigned int ifindex, int queue_id) +{ + struct netdev_queue_get_req *req; + struct netdev_queue_get_rsp *rsp; + struct ynl_error yerr; + struct ynl_sock *ys; + int napi_id; + + ys = ynl_sock_create(&ynl_netdev_family, &yerr); + if (!ys) + error(1, 0, "ynl_sock_create: %s", yerr.msg); + + req = netdev_queue_get_req_alloc(); + netdev_queue_get_req_set_ifindex(req, ifindex); + netdev_queue_get_req_set_type(req, NETDEV_QUEUE_TYPE_RX); + netdev_queue_get_req_set_id(req, queue_id); + + rsp = netdev_queue_get(ys, req); + if (!rsp) + error(1, 0, "netdev_queue_get(q=%d): %s", queue_id, + ys->err.msg); + if (!rsp->_present.napi_id) + error(1, 0, "netdev_queue_get(q=%d): napi_id not present", + queue_id); + + napi_id = rsp->napi_id; + + netdev_queue_get_req_free(req); + netdev_queue_get_rsp_free(rsp); + ynl_sock_destroy(ys); + + return napi_id; +} + +static int find_thread_by_conn(struct thread_ctx *ctxs, int connfd) +{ + socklen_t len = sizeof(int); + int napi_id, i; + + if (getsockopt(connfd, SOL_SOCKET, SO_INCOMING_NAPI_ID, &napi_id, &len)) + error(1, errno, "getsockopt(SO_INCOMING_NAPI_ID)"); + + for (i = 0; i < cfg_num_threads; i++) { + if (ctxs[i].napi_id == napi_id) + return i; + } + + error(1, 0, "unknown NAPI ID: %d", napi_id); + return -1; +} + static void run_server(void) { - struct thread_ctx ctx = {}; - unsigned int flags = 0; - int fd, enable, ret; - uint64_t tstop; + pthread_barrier_t setup_done, dispatch_done; + int total_conns, accepted = 0, connfd; + struct thread_ctx *ctxs; + int fd, ret, enable, i; + unsigned int ifindex; + pthread_t *threads; + + ctxs = calloc(cfg_num_threads, sizeof(*ctxs)); + threads = calloc(cfg_num_threads, sizeof(*threads)); + if (!ctxs || !threads) + error(1, 0, "calloc()"); fd = socket(AF_INET6, SOCK_STREAM, 0); if (fd == -1) @@ -356,29 +457,68 @@ static void run_server(void) if (ret < 0) error(1, 0, "bind()"); - flags |= IORING_SETUP_COOP_TASKRUN; - flags |= IORING_SETUP_SINGLE_ISSUER; - flags |= IORING_SETUP_DEFER_TASKRUN; - flags |= IORING_SETUP_SUBMIT_ALL; - flags |= IORING_SETUP_CQE32; + pthread_barrier_init(&setup_done, NULL, cfg_num_threads + 1); + pthread_barrier_init(&dispatch_done, NULL, cfg_num_threads + 1); - io_uring_queue_init(512, &ctx.ring, flags); + for (i = 0; i < cfg_num_threads; i++) { + ctxs[i].queue_id = cfg_queue_id + i; + ctxs[i].setup_done = &setup_done; + ctxs[i].dispatch_done = &dispatch_done; + } + + for (i = 0; i < cfg_num_threads; i++) { + ret = pthread_create(&threads[i], NULL, + server_worker, &ctxs[i]); + if (ret) + error(1, ret, "pthread_create()"); + } - setup_zcrx(&ctx); if (cfg_dry_run) - return; + goto join; + + pthread_barrier_wait(&setup_done); if (listen(fd, 1024) < 0) error(1, 0, "listen()"); - add_accept(&ctx, fd); + if (cfg_num_threads > 1) { + ifindex = if_nametoindex(cfg_ifname); + if (!ifindex) + error(1, 0, "bad interface name: %s", cfg_ifname); + for (i = 0; i < cfg_num_threads; i++) + ctxs[i].napi_id = query_napi_id(ifindex, + ctxs[i].queue_id); + } - tstop = gettimeofday_ms() + 5000; - while (!ctx.stop && gettimeofday_ms() < tstop) - server_loop(&ctx); + total_conns = cfg_num_threads * cfg_num_threads; - if (!ctx.stop) - error(1, 0, "test failed\n"); + while (accepted < total_conns) { + int idx = 0; + + connfd = accept(fd, NULL, NULL); + if (connfd < 0) + error(1, errno, "accept()"); + + if (cfg_num_threads > 1) + idx = find_thread_by_conn(ctxs, connfd); + + if (ctxs[idx].nr_conns >= MAX_CONNS_PER_THREAD) + error(1, 0, "worker %d connection overflow", idx); + ctxs[idx].connfds[ctxs[idx].nr_conns++] = connfd; + accepted++; + } + + pthread_barrier_wait(&dispatch_done); + +join: + for (i = 0; i < cfg_num_threads; i++) + pthread_join(threads[i], NULL); + + pthread_barrier_destroy(&setup_done); + pthread_barrier_destroy(&dispatch_done); + close(fd); + free(threads); + free(ctxs); } static void *client_worker(void *arg) From 9814d2fa87427a2642e54b67d761a499437058fa Mon Sep 17 00:00:00 2001 From: Juanlu Herrero Date: Wed, 9 Sep 2026 14:51:28 -0500 Subject: [PATCH 6/6] selftests: net: add rss_multiqueue test variant to iou-zcrx Add a rss_multiqueue Python test variant that exercises multi-queue zero-copy receive on a single listening socket. Reviewed-by: David Wei Signed-off-by: Juanlu Herrero Link: https://patch.msgid.link/20260909-iou-zcrx-v7-6-6b5d48a2a922@fastmail.com Signed-off-by: Jakub Kicinski --- .../selftests/drivers/net/hw/iou-zcrx.py | 50 ++++++++++++++++++- 1 file changed, 48 insertions(+), 2 deletions(-) diff --git a/tools/testing/selftests/drivers/net/hw/iou-zcrx.py b/tools/testing/selftests/drivers/net/hw/iou-zcrx.py index b7a225fe4bea..c833535d8a03 100755 --- a/tools/testing/selftests/drivers/net/hw/iou-zcrx.py +++ b/tools/testing/selftests/drivers/net/hw/iou-zcrx.py @@ -30,6 +30,13 @@ def create_rss_ctx(cfg): return int(values) +def create_rss_ctx_multi(cfg, start, count): + """Create an RSS context spanning count queues from start, return its ID.""" + output = ethtool(f"-X {cfg.ifname} context new start {start} equal {count}").stdout + values = re.search(r'New RSS context is (\d+)', output).group(1) + return int(values) + + def set_flow_rule(cfg): output = ethtool(f"-N {cfg.ifname} flow-type tcp6 dst-port {cfg.port} action {cfg.target}").stdout values = re.search(r'ID (\d+)', output).group(1) @@ -127,17 +134,56 @@ def _require_ntuple(cfg): defer(ethtool, f"-K {cfg.ifname} ntuple-filters off") +def rss_multiqueue(cfg): + """Steer the test flow to a multi-queue RSS context for multi-thread zcrx.""" + channels = cfg.ethnl.channels_get({'header': {'dev-index': cfg.ifindex}}) + channels = channels['combined-count'] + if channels < 3: + raise KsftSkipEx('Test requires NETIF with at least 3 combined channels') + + rings = cfg.ethnl.rings_get({'header': {'dev-index': cfg.ifindex}}) + rx_rings = rings['rx'] + hds_thresh = rings.get('hds-thresh', 0) + + cfg.ethnl.rings_set({'header': {'dev-index': cfg.ifindex}, + 'tcp-data-split': 'enabled', + 'hds-thresh': 0, + 'rx': 64}) + defer(cfg.ethnl.rings_set, {'header': {'dev-index': cfg.ifindex}, + 'tcp-data-split': 'unknown', + 'hds-thresh': hds_thresh, + 'rx': rx_rings}) + defer(mp_clear_wait, cfg) + + cfg.num_threads = 2 + cfg.target = channels - cfg.num_threads + ethtool(f"-X {cfg.ifname} equal {cfg.target}") + defer(ethtool, f"-X {cfg.ifname} default") + + rss_ctx_id = create_rss_ctx_multi(cfg, cfg.target, cfg.num_threads) + defer(ethtool, f"-X {cfg.ifname} delete context {rss_ctx_id}") + + flow_rule_id = set_flow_rule_rss(cfg, rss_ctx_id) + defer(ethtool, f"-N {cfg.ifname} delete {flow_rule_id}") + + @ksft_variants([ KsftNamedVariant("single", single), KsftNamedVariant("rss", rss), + KsftNamedVariant("rss_multiqueue", rss_multiqueue), ]) def test_zcrx(cfg, setup) -> None: cfg.require_ipver('6') _require_ntuple(cfg) + cfg.num_threads = 1 + setup(cfg) - rx_cmd = f"{cfg.bin_local} -s -p {cfg.port} -i {cfg.ifname} -q {cfg.target}" - tx_cmd = f"{cfg.bin_remote} -c -h {cfg.addr_v['6']} -p {cfg.port} -l 12840" + + rx_cmd = (f"{cfg.bin_local} -s -p {cfg.port} -i {cfg.ifname} " + f"-q {cfg.target} -t {cfg.num_threads}") + tx_cmd = (f"{cfg.bin_remote} -c -h {cfg.addr_v['6']} -p {cfg.port} " + f"-l 12840 -t {cfg.num_threads}") with bkg(rx_cmd, exit_wait=True): wait_port_listen(cfg.port, proto="tcp") cmd(tx_cmd, host=cfg.remote)