xref: /linux/tools/testing/selftests/drivers/net/hw/ncdevmem.c (revision b803c4a4f78834b31ebfbbcea350473333760559)
1 // SPDX-License-Identifier: GPL-2.0
2 /*
3  * tcpdevmem netcat. Works similarly to netcat but does device memory TCP
4  * instead of regular TCP. Uses udmabuf to mock a dmabuf provider.
5  *
6  * Usage:
7  *
8  *     On server:
9  *     ncdevmem -s <server IP> [-c <client IP>] -f eth1 -l -p 5201
10  *
11  *     On client:
12  *     echo -n "hello\nworld" | \
13  *		ncdevmem -s <server IP> [-c <client IP>] -p 5201 -f eth1
14  *
15  * Note this is compatible with regular netcat. i.e. the sender or receiver can
16  * be replaced with regular netcat to test the RX or TX path in isolation.
17  *
18  * Test data validation (devmem TCP on RX only):
19  *
20  *     On server:
21  *     ncdevmem -s <server IP> [-c <client IP>] -f eth1 -l -p 5201 -v 7
22  *
23  *     On client:
24  *     yes $(echo -e \\x01\\x02\\x03\\x04\\x05\\x06) | \
25  *             head -c 1G | \
26  *             nc <server IP> 5201 -p 5201
27  *
28  * Test data validation (devmem TCP on RX and TX, validation happens on RX):
29  *
30  *	On server:
31  *	ncdevmem -s <server IP> [-c <client IP>] -l -p 5201 -v 8 -f eth1
32  *
33  *	On client:
34  *	yes $(echo -e \\x01\\x02\\x03\\x04\\x05\\x06\\x07) | \
35  *		head -c 1M | \
36  *		ncdevmem -s <server IP> [-c <client IP>] -p 5201 -f eth1
37  */
38 #define _GNU_SOURCE
39 #define __EXPORTED_HEADERS__
40 
41 #include <linux/uio.h>
42 #include <stdio.h>
43 #include <stdlib.h>
44 #include <unistd.h>
45 #include <stdbool.h>
46 #include <string.h>
47 #include <errno.h>
48 #define __iovec_defined
49 #include <fcntl.h>
50 #include <malloc.h>
51 #include <error.h>
52 #include <poll.h>
53 
54 #include <arpa/inet.h>
55 #include <sys/socket.h>
56 #include <sys/mman.h>
57 #include <sys/ioctl.h>
58 #include <sys/syscall.h>
59 #include <sys/time.h>
60 
61 #include <linux/memfd.h>
62 #include <linux/dma-buf.h>
63 #include <linux/errqueue.h>
64 #include <linux/udmabuf.h>
65 #include <linux/types.h>
66 #include <linux/netlink.h>
67 #include <linux/genetlink.h>
68 #include <linux/netdev.h>
69 #include <linux/ethtool_netlink.h>
70 #include <time.h>
71 #include <net/if.h>
72 
73 #include "netdev-user.h"
74 #include "ethtool-user.h"
75 #include <ynl.h>
76 
77 #define PAGE_SHIFT 12
78 #define TEST_PREFIX "ncdevmem"
79 #define NUM_PAGES 16000
80 
81 #ifndef MSG_SOCK_DEVMEM
82 #define MSG_SOCK_DEVMEM 0x2000000
83 #endif
84 
85 static char *server_ip;
86 static char *client_ip;
87 static char *port;
88 static size_t do_validation;
89 static int start_queue = -1;
90 static int num_queues = -1;
91 static char *ifname;
92 static unsigned int ifindex;
93 static unsigned int dmabuf_id;
94 static uint32_t tx_dmabuf_id;
95 static int waittime_ms = 500;
96 
97 struct memory_buffer {
98 	int fd;
99 	size_t size;
100 
101 	int devfd;
102 	int memfd;
103 	char *buf_mem;
104 };
105 
106 struct memory_provider {
107 	struct memory_buffer *(*alloc)(size_t size);
108 	void (*free)(struct memory_buffer *ctx);
109 	void (*memcpy_to_device)(struct memory_buffer *dst, size_t off,
110 				 void *src, int n);
111 	void (*memcpy_from_device)(void *dst, struct memory_buffer *src,
112 				   size_t off, int n);
113 };
114 
115 static struct memory_buffer *udmabuf_alloc(size_t size)
116 {
117 	struct udmabuf_create create;
118 	struct memory_buffer *ctx;
119 	int ret;
120 
121 	ctx = malloc(sizeof(*ctx));
122 	if (!ctx)
123 		error(1, ENOMEM, "malloc failed");
124 
125 	ctx->size = size;
126 
127 	ctx->devfd = open("/dev/udmabuf", O_RDWR);
128 	if (ctx->devfd < 0)
129 		error(1, errno,
130 		      "%s: [skip,no-udmabuf: Unable to access DMA buffer device file]\n",
131 		      TEST_PREFIX);
132 
133 	ctx->memfd = memfd_create("udmabuf-test", MFD_ALLOW_SEALING);
134 	if (ctx->memfd < 0)
135 		error(1, errno, "%s: [skip,no-memfd]\n", TEST_PREFIX);
136 
137 	ret = fcntl(ctx->memfd, F_ADD_SEALS, F_SEAL_SHRINK);
138 	if (ret < 0)
139 		error(1, errno, "%s: [skip,fcntl-add-seals]\n", TEST_PREFIX);
140 
141 	ret = ftruncate(ctx->memfd, size);
142 	if (ret == -1)
143 		error(1, errno, "%s: [FAIL,memfd-truncate]\n", TEST_PREFIX);
144 
145 	memset(&create, 0, sizeof(create));
146 
147 	create.memfd = ctx->memfd;
148 	create.offset = 0;
149 	create.size = size;
150 	ctx->fd = ioctl(ctx->devfd, UDMABUF_CREATE, &create);
151 	if (ctx->fd < 0)
152 		error(1, errno, "%s: [FAIL, create udmabuf]\n", TEST_PREFIX);
153 
154 	ctx->buf_mem = mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED,
155 			    ctx->fd, 0);
156 	if (ctx->buf_mem == MAP_FAILED)
157 		error(1, errno, "%s: [FAIL, map udmabuf]\n", TEST_PREFIX);
158 
159 	return ctx;
160 }
161 
162 static void udmabuf_free(struct memory_buffer *ctx)
163 {
164 	munmap(ctx->buf_mem, ctx->size);
165 	close(ctx->fd);
166 	close(ctx->memfd);
167 	close(ctx->devfd);
168 	free(ctx);
169 }
170 
171 static void udmabuf_memcpy_to_device(struct memory_buffer *dst, size_t off,
172 				     void *src, int n)
173 {
174 	struct dma_buf_sync sync = {};
175 
176 	sync.flags = DMA_BUF_SYNC_START | DMA_BUF_SYNC_WRITE;
177 	ioctl(dst->fd, DMA_BUF_IOCTL_SYNC, &sync);
178 
179 	memcpy(dst->buf_mem + off, src, n);
180 
181 	sync.flags = DMA_BUF_SYNC_END | DMA_BUF_SYNC_WRITE;
182 	ioctl(dst->fd, DMA_BUF_IOCTL_SYNC, &sync);
183 }
184 
185 static void udmabuf_memcpy_from_device(void *dst, struct memory_buffer *src,
186 				       size_t off, int n)
187 {
188 	struct dma_buf_sync sync = {};
189 
190 	sync.flags = DMA_BUF_SYNC_START;
191 	ioctl(src->fd, DMA_BUF_IOCTL_SYNC, &sync);
192 
193 	memcpy(dst, src->buf_mem + off, n);
194 
195 	sync.flags = DMA_BUF_SYNC_END;
196 	ioctl(src->fd, DMA_BUF_IOCTL_SYNC, &sync);
197 }
198 
199 static struct memory_provider udmabuf_memory_provider = {
200 	.alloc = udmabuf_alloc,
201 	.free = udmabuf_free,
202 	.memcpy_to_device = udmabuf_memcpy_to_device,
203 	.memcpy_from_device = udmabuf_memcpy_from_device,
204 };
205 
206 static struct memory_provider *provider = &udmabuf_memory_provider;
207 
208 static void print_nonzero_bytes(void *ptr, size_t size)
209 {
210 	unsigned char *p = ptr;
211 	unsigned int i;
212 
213 	for (i = 0; i < size; i++)
214 		putchar(p[i]);
215 }
216 
217 void validate_buffer(void *line, size_t size)
218 {
219 	static unsigned char seed = 1;
220 	unsigned char *ptr = line;
221 	unsigned char expected;
222 	static int errors;
223 	size_t i;
224 
225 	for (i = 0; i < size; i++) {
226 		expected = seed ? seed : '\n';
227 		if (ptr[i] != expected) {
228 			fprintf(stderr,
229 				"Failed validation: expected=%u, actual=%u, index=%lu\n",
230 				expected, ptr[i], i);
231 			errors++;
232 			if (errors > 20)
233 				error(1, 0, "validation failed.");
234 		}
235 		seed++;
236 		if (seed == do_validation)
237 			seed = 0;
238 	}
239 
240 	fprintf(stdout, "Validated buffer\n");
241 }
242 
243 static int rxq_num(int ifindex)
244 {
245 	struct ethtool_channels_get_req *req;
246 	struct ethtool_channels_get_rsp *rsp;
247 	struct ynl_error yerr;
248 	struct ynl_sock *ys;
249 	int num = -1;
250 
251 	ys = ynl_sock_create(&ynl_ethtool_family, &yerr);
252 	if (!ys) {
253 		fprintf(stderr, "YNL: %s\n", yerr.msg);
254 		return -1;
255 	}
256 
257 	req = ethtool_channels_get_req_alloc();
258 	ethtool_channels_get_req_set_header_dev_index(req, ifindex);
259 	rsp = ethtool_channels_get(ys, req);
260 	if (rsp)
261 		num = rsp->rx_count + rsp->combined_count;
262 	ethtool_channels_get_req_free(req);
263 	ethtool_channels_get_rsp_free(rsp);
264 
265 	ynl_sock_destroy(ys);
266 
267 	return num;
268 }
269 
270 #define run_command(cmd, ...)                                           \
271 	({                                                              \
272 		char command[256];                                      \
273 		memset(command, 0, sizeof(command));                    \
274 		snprintf(command, sizeof(command), cmd, ##__VA_ARGS__); \
275 		fprintf(stderr, "Running: %s\n", command);                       \
276 		system(command);                                        \
277 	})
278 
279 static int reset_flow_steering(void)
280 {
281 	/* Depending on the NIC, toggling ntuple off and on might not
282 	 * be allowed. Additionally, attempting to delete existing filters
283 	 * will fail if no filters are present. Therefore, do not enforce
284 	 * the exit status.
285 	 */
286 
287 	run_command("sudo ethtool -K %s ntuple off >&2", ifname);
288 	run_command("sudo ethtool -K %s ntuple on >&2", ifname);
289 	run_command(
290 		"sudo ethtool -n %s | grep 'Filter:' | awk '{print $2}' | xargs -n1 ethtool -N %s delete >&2",
291 		ifname, ifname);
292 	return 0;
293 }
294 
295 static const char *tcp_data_split_str(int val)
296 {
297 	switch (val) {
298 	case 0:
299 		return "off";
300 	case 1:
301 		return "auto";
302 	case 2:
303 		return "on";
304 	default:
305 		return "?";
306 	}
307 }
308 
309 static int configure_headersplit(bool on)
310 {
311 	struct ethtool_rings_get_req *get_req;
312 	struct ethtool_rings_get_rsp *get_rsp;
313 	struct ethtool_rings_set_req *req;
314 	struct ynl_error yerr;
315 	struct ynl_sock *ys;
316 	int ret;
317 
318 	ys = ynl_sock_create(&ynl_ethtool_family, &yerr);
319 	if (!ys) {
320 		fprintf(stderr, "YNL: %s\n", yerr.msg);
321 		return -1;
322 	}
323 
324 	req = ethtool_rings_set_req_alloc();
325 	ethtool_rings_set_req_set_header_dev_index(req, ifindex);
326 	/* 0 - off, 1 - auto, 2 - on */
327 	ethtool_rings_set_req_set_tcp_data_split(req, on ? 2 : 0);
328 	ret = ethtool_rings_set(ys, req);
329 	if (ret < 0)
330 		fprintf(stderr, "YNL failed: %s\n", ys->err.msg);
331 	ethtool_rings_set_req_free(req);
332 
333 	if (ret == 0) {
334 		get_req = ethtool_rings_get_req_alloc();
335 		ethtool_rings_get_req_set_header_dev_index(get_req, ifindex);
336 		get_rsp = ethtool_rings_get(ys, get_req);
337 		ethtool_rings_get_req_free(get_req);
338 		if (get_rsp)
339 			fprintf(stderr, "TCP header split: %s\n",
340 				tcp_data_split_str(get_rsp->tcp_data_split));
341 		ethtool_rings_get_rsp_free(get_rsp);
342 	}
343 
344 	ynl_sock_destroy(ys);
345 
346 	return ret;
347 }
348 
349 static int configure_rss(void)
350 {
351 	return run_command("sudo ethtool -X %s equal %d >&2", ifname, start_queue);
352 }
353 
354 static int configure_channels(unsigned int rx, unsigned int tx)
355 {
356 	return run_command("sudo ethtool -L %s rx %u tx %u", ifname, rx, tx);
357 }
358 
359 static int configure_flow_steering(struct sockaddr_in6 *server_sin)
360 {
361 	const char *type = "tcp6";
362 	const char *server_addr;
363 	char buf[40];
364 
365 	inet_ntop(AF_INET6, &server_sin->sin6_addr, buf, sizeof(buf));
366 	server_addr = buf;
367 
368 	if (IN6_IS_ADDR_V4MAPPED(&server_sin->sin6_addr)) {
369 		type = "tcp4";
370 		server_addr = strrchr(server_addr, ':') + 1;
371 	}
372 
373 	return run_command("sudo ethtool -N %s flow-type %s %s %s dst-ip %s %s %s dst-port %s queue %d >&2",
374 			   ifname,
375 			   type,
376 			   client_ip ? "src-ip" : "",
377 			   client_ip ?: "",
378 			   server_addr,
379 			   client_ip ? "src-port" : "",
380 			   client_ip ? port : "",
381 			   port, start_queue);
382 }
383 
384 static int bind_rx_queue(unsigned int ifindex, unsigned int dmabuf_fd,
385 			 struct netdev_queue_id *queues,
386 			 unsigned int n_queue_index, struct ynl_sock **ys)
387 {
388 	struct netdev_bind_rx_req *req = NULL;
389 	struct netdev_bind_rx_rsp *rsp = NULL;
390 	struct ynl_error yerr;
391 
392 	*ys = ynl_sock_create(&ynl_netdev_family, &yerr);
393 	if (!*ys) {
394 		fprintf(stderr, "YNL: %s\n", yerr.msg);
395 		return -1;
396 	}
397 
398 	req = netdev_bind_rx_req_alloc();
399 	netdev_bind_rx_req_set_ifindex(req, ifindex);
400 	netdev_bind_rx_req_set_fd(req, dmabuf_fd);
401 	__netdev_bind_rx_req_set_queues(req, queues, n_queue_index);
402 
403 	rsp = netdev_bind_rx(*ys, req);
404 	if (!rsp) {
405 		perror("netdev_bind_rx");
406 		goto err_close;
407 	}
408 
409 	if (!rsp->_present.id) {
410 		perror("id not present");
411 		goto err_close;
412 	}
413 
414 	fprintf(stderr, "got dmabuf id=%d\n", rsp->id);
415 	dmabuf_id = rsp->id;
416 
417 	netdev_bind_rx_req_free(req);
418 	netdev_bind_rx_rsp_free(rsp);
419 
420 	return 0;
421 
422 err_close:
423 	fprintf(stderr, "YNL failed: %s\n", (*ys)->err.msg);
424 	netdev_bind_rx_req_free(req);
425 	ynl_sock_destroy(*ys);
426 	return -1;
427 }
428 
429 static int bind_tx_queue(unsigned int ifindex, unsigned int dmabuf_fd,
430 			 struct ynl_sock **ys)
431 {
432 	struct netdev_bind_tx_req *req = NULL;
433 	struct netdev_bind_tx_rsp *rsp = NULL;
434 	struct ynl_error yerr;
435 
436 	*ys = ynl_sock_create(&ynl_netdev_family, &yerr);
437 	if (!*ys) {
438 		fprintf(stderr, "YNL: %s\n", yerr.msg);
439 		return -1;
440 	}
441 
442 	req = netdev_bind_tx_req_alloc();
443 	netdev_bind_tx_req_set_ifindex(req, ifindex);
444 	netdev_bind_tx_req_set_fd(req, dmabuf_fd);
445 
446 	rsp = netdev_bind_tx(*ys, req);
447 	if (!rsp) {
448 		perror("netdev_bind_tx");
449 		goto err_close;
450 	}
451 
452 	if (!rsp->_present.id) {
453 		perror("id not present");
454 		goto err_close;
455 	}
456 
457 	fprintf(stderr, "got tx dmabuf id=%d\n", rsp->id);
458 	tx_dmabuf_id = rsp->id;
459 
460 	netdev_bind_tx_req_free(req);
461 	netdev_bind_tx_rsp_free(rsp);
462 
463 	return 0;
464 
465 err_close:
466 	fprintf(stderr, "YNL failed: %s\n", (*ys)->err.msg);
467 	netdev_bind_tx_req_free(req);
468 	ynl_sock_destroy(*ys);
469 	return -1;
470 }
471 
472 static void enable_reuseaddr(int fd)
473 {
474 	int opt = 1;
475 	int ret;
476 
477 	ret = setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));
478 	if (ret)
479 		error(1, errno, "%s: [FAIL, SO_REUSEPORT]\n", TEST_PREFIX);
480 
481 	ret = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
482 	if (ret)
483 		error(1, errno, "%s: [FAIL, SO_REUSEADDR]\n", TEST_PREFIX);
484 }
485 
486 static int parse_address(const char *str, int port, struct sockaddr_in6 *sin6)
487 {
488 	int ret;
489 
490 	sin6->sin6_family = AF_INET6;
491 	sin6->sin6_port = htons(port);
492 
493 	ret = inet_pton(sin6->sin6_family, str, &sin6->sin6_addr);
494 	if (ret != 1) {
495 		/* fallback to plain IPv4 */
496 		ret = inet_pton(AF_INET, str, &sin6->sin6_addr.s6_addr32[3]);
497 		if (ret != 1)
498 			return -1;
499 
500 		/* add ::ffff prefix */
501 		sin6->sin6_addr.s6_addr32[0] = 0;
502 		sin6->sin6_addr.s6_addr32[1] = 0;
503 		sin6->sin6_addr.s6_addr16[4] = 0;
504 		sin6->sin6_addr.s6_addr16[5] = 0xffff;
505 	}
506 
507 	return 0;
508 }
509 
510 static struct netdev_queue_id *create_queues(void)
511 {
512 	struct netdev_queue_id *queues;
513 	size_t i = 0;
514 
515 	queues = calloc(num_queues, sizeof(*queues));
516 	for (i = 0; i < num_queues; i++) {
517 		queues[i]._present.type = 1;
518 		queues[i]._present.id = 1;
519 		queues[i].type = NETDEV_QUEUE_TYPE_RX;
520 		queues[i].id = start_queue + i;
521 	}
522 
523 	return queues;
524 }
525 
526 static int do_server(struct memory_buffer *mem)
527 {
528 	char ctrl_data[sizeof(int) * 20000];
529 	struct netdev_queue_id *queues;
530 	size_t non_page_aligned_frags = 0;
531 	struct sockaddr_in6 client_addr;
532 	struct sockaddr_in6 server_sin;
533 	size_t page_aligned_frags = 0;
534 	size_t total_received = 0;
535 	socklen_t client_addr_len;
536 	bool is_devmem = false;
537 	char *tmp_mem = NULL;
538 	struct ynl_sock *ys;
539 	char iobuf[819200];
540 	char buffer[256];
541 	int socket_fd;
542 	int client_fd;
543 	int ret;
544 
545 	ret = parse_address(server_ip, atoi(port), &server_sin);
546 	if (ret < 0)
547 		error(1, 0, "parse server address");
548 
549 	if (reset_flow_steering())
550 		error(1, 0, "Failed to reset flow steering\n");
551 
552 	if (configure_headersplit(1))
553 		error(1, 0, "Failed to enable TCP header split\n");
554 
555 	/* Configure RSS to divert all traffic from our devmem queues */
556 	if (configure_rss())
557 		error(1, 0, "Failed to configure rss\n");
558 
559 	/* Flow steer our devmem flows to start_queue */
560 	if (configure_flow_steering(&server_sin))
561 		error(1, 0, "Failed to configure flow steering\n");
562 
563 	sleep(1);
564 
565 	if (bind_rx_queue(ifindex, mem->fd, create_queues(), num_queues, &ys))
566 		error(1, 0, "Failed to bind\n");
567 
568 	tmp_mem = malloc(mem->size);
569 	if (!tmp_mem)
570 		error(1, ENOMEM, "malloc failed");
571 
572 	socket_fd = socket(AF_INET6, SOCK_STREAM, 0);
573 	if (socket_fd < 0)
574 		error(1, errno, "%s: [FAIL, create socket]\n", TEST_PREFIX);
575 
576 	enable_reuseaddr(socket_fd);
577 
578 	fprintf(stderr, "binding to address %s:%d\n", server_ip,
579 		ntohs(server_sin.sin6_port));
580 
581 	ret = bind(socket_fd, &server_sin, sizeof(server_sin));
582 	if (ret)
583 		error(1, errno, "%s: [FAIL, bind]\n", TEST_PREFIX);
584 
585 	ret = listen(socket_fd, 1);
586 	if (ret)
587 		error(1, errno, "%s: [FAIL, listen]\n", TEST_PREFIX);
588 
589 	client_addr_len = sizeof(client_addr);
590 
591 	inet_ntop(AF_INET6, &server_sin.sin6_addr, buffer,
592 		  sizeof(buffer));
593 	fprintf(stderr, "Waiting or connection on %s:%d\n", buffer,
594 		ntohs(server_sin.sin6_port));
595 	client_fd = accept(socket_fd, &client_addr, &client_addr_len);
596 
597 	inet_ntop(AF_INET6, &client_addr.sin6_addr, buffer,
598 		  sizeof(buffer));
599 	fprintf(stderr, "Got connection from %s:%d\n", buffer,
600 		ntohs(client_addr.sin6_port));
601 
602 	while (1) {
603 		struct iovec iov = { .iov_base = iobuf,
604 				     .iov_len = sizeof(iobuf) };
605 		struct dmabuf_cmsg *dmabuf_cmsg = NULL;
606 		struct cmsghdr *cm = NULL;
607 		struct msghdr msg = { 0 };
608 		struct dmabuf_token token;
609 		ssize_t ret;
610 
611 		is_devmem = false;
612 
613 		msg.msg_iov = &iov;
614 		msg.msg_iovlen = 1;
615 		msg.msg_control = ctrl_data;
616 		msg.msg_controllen = sizeof(ctrl_data);
617 		ret = recvmsg(client_fd, &msg, MSG_SOCK_DEVMEM);
618 		fprintf(stderr, "recvmsg ret=%ld\n", ret);
619 		if (ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK))
620 			continue;
621 		if (ret < 0) {
622 			perror("recvmsg");
623 			continue;
624 		}
625 		if (ret == 0) {
626 			fprintf(stderr, "client exited\n");
627 			goto cleanup;
628 		}
629 
630 		for (cm = CMSG_FIRSTHDR(&msg); cm; cm = CMSG_NXTHDR(&msg, cm)) {
631 			if (cm->cmsg_level != SOL_SOCKET ||
632 			    (cm->cmsg_type != SCM_DEVMEM_DMABUF &&
633 			     cm->cmsg_type != SCM_DEVMEM_LINEAR)) {
634 				fprintf(stderr, "skipping non-devmem cmsg\n");
635 				continue;
636 			}
637 
638 			dmabuf_cmsg = (struct dmabuf_cmsg *)CMSG_DATA(cm);
639 			is_devmem = true;
640 
641 			if (cm->cmsg_type == SCM_DEVMEM_LINEAR) {
642 				/* TODO: process data copied from skb's linear
643 				 * buffer.
644 				 */
645 				fprintf(stderr,
646 					"SCM_DEVMEM_LINEAR. dmabuf_cmsg->frag_size=%u\n",
647 					dmabuf_cmsg->frag_size);
648 
649 				continue;
650 			}
651 
652 			token.token_start = dmabuf_cmsg->frag_token;
653 			token.token_count = 1;
654 
655 			total_received += dmabuf_cmsg->frag_size;
656 			fprintf(stderr,
657 				"received frag_page=%llu, in_page_offset=%llu, frag_offset=%llu, frag_size=%u, token=%u, total_received=%lu, dmabuf_id=%u\n",
658 				dmabuf_cmsg->frag_offset >> PAGE_SHIFT,
659 				dmabuf_cmsg->frag_offset % getpagesize(),
660 				dmabuf_cmsg->frag_offset,
661 				dmabuf_cmsg->frag_size, dmabuf_cmsg->frag_token,
662 				total_received, dmabuf_cmsg->dmabuf_id);
663 
664 			if (dmabuf_cmsg->dmabuf_id != dmabuf_id)
665 				error(1, 0,
666 				      "received on wrong dmabuf_id: flow steering error\n");
667 
668 			if (dmabuf_cmsg->frag_size % getpagesize())
669 				non_page_aligned_frags++;
670 			else
671 				page_aligned_frags++;
672 
673 			provider->memcpy_from_device(tmp_mem, mem,
674 						     dmabuf_cmsg->frag_offset,
675 						     dmabuf_cmsg->frag_size);
676 
677 			if (do_validation)
678 				validate_buffer(tmp_mem,
679 						dmabuf_cmsg->frag_size);
680 			else
681 				print_nonzero_bytes(tmp_mem,
682 						    dmabuf_cmsg->frag_size);
683 
684 			ret = setsockopt(client_fd, SOL_SOCKET,
685 					 SO_DEVMEM_DONTNEED, &token,
686 					 sizeof(token));
687 			if (ret != 1)
688 				error(1, 0,
689 				      "SO_DEVMEM_DONTNEED not enough tokens");
690 		}
691 		if (!is_devmem)
692 			error(1, 0, "flow steering error\n");
693 
694 		fprintf(stderr, "total_received=%lu\n", total_received);
695 	}
696 
697 	fprintf(stderr, "%s: ok\n", TEST_PREFIX);
698 
699 	fprintf(stderr, "page_aligned_frags=%lu, non_page_aligned_frags=%lu\n",
700 		page_aligned_frags, non_page_aligned_frags);
701 
702 cleanup:
703 
704 	free(tmp_mem);
705 	close(client_fd);
706 	close(socket_fd);
707 	ynl_sock_destroy(ys);
708 
709 	return 0;
710 }
711 
712 void run_devmem_tests(void)
713 {
714 	struct memory_buffer *mem;
715 	struct ynl_sock *ys;
716 
717 	mem = provider->alloc(getpagesize() * NUM_PAGES);
718 
719 	/* Configure RSS to divert all traffic from our devmem queues */
720 	if (configure_rss())
721 		error(1, 0, "rss error\n");
722 
723 	if (configure_headersplit(1))
724 		error(1, 0, "Failed to configure header split\n");
725 
726 	if (!bind_rx_queue(ifindex, mem->fd,
727 			   calloc(num_queues, sizeof(struct netdev_queue_id)),
728 			   num_queues, &ys))
729 		error(1, 0, "Binding empty queues array should have failed\n");
730 
731 	if (configure_headersplit(0))
732 		error(1, 0, "Failed to configure header split\n");
733 
734 	if (!bind_rx_queue(ifindex, mem->fd, create_queues(), num_queues, &ys))
735 		error(1, 0, "Configure dmabuf with header split off should have failed\n");
736 
737 	if (configure_headersplit(1))
738 		error(1, 0, "Failed to configure header split\n");
739 
740 	if (bind_rx_queue(ifindex, mem->fd, create_queues(), num_queues, &ys))
741 		error(1, 0, "Failed to bind\n");
742 
743 	/* Deactivating a bound queue should not be legal */
744 	if (!configure_channels(num_queues, num_queues - 1))
745 		error(1, 0, "Deactivating a bound queue should be illegal.\n");
746 
747 	/* Closing the netlink socket does an implicit unbind */
748 	ynl_sock_destroy(ys);
749 
750 	provider->free(mem);
751 }
752 
753 static uint64_t gettimeofday_ms(void)
754 {
755 	struct timeval tv;
756 
757 	gettimeofday(&tv, NULL);
758 	return (tv.tv_sec * 1000ULL) + (tv.tv_usec / 1000ULL);
759 }
760 
761 static int do_poll(int fd)
762 {
763 	struct pollfd pfd;
764 	int ret;
765 
766 	pfd.revents = 0;
767 	pfd.fd = fd;
768 
769 	ret = poll(&pfd, 1, waittime_ms);
770 	if (ret == -1)
771 		error(1, errno, "poll");
772 
773 	return ret && (pfd.revents & POLLERR);
774 }
775 
776 static void wait_compl(int fd)
777 {
778 	int64_t tstop = gettimeofday_ms() + waittime_ms;
779 	char control[CMSG_SPACE(100)] = {};
780 	struct sock_extended_err *serr;
781 	struct msghdr msg = {};
782 	struct cmsghdr *cm;
783 	__u32 hi, lo;
784 	int ret;
785 
786 	msg.msg_control = control;
787 	msg.msg_controllen = sizeof(control);
788 
789 	while (gettimeofday_ms() < tstop) {
790 		if (!do_poll(fd))
791 			continue;
792 
793 		ret = recvmsg(fd, &msg, MSG_ERRQUEUE);
794 		if (ret < 0) {
795 			if (errno == EAGAIN)
796 				continue;
797 			error(1, errno, "recvmsg(MSG_ERRQUEUE)");
798 			return;
799 		}
800 		if (msg.msg_flags & MSG_CTRUNC)
801 			error(1, 0, "MSG_CTRUNC\n");
802 
803 		for (cm = CMSG_FIRSTHDR(&msg); cm; cm = CMSG_NXTHDR(&msg, cm)) {
804 			if (cm->cmsg_level != SOL_IP &&
805 			    cm->cmsg_level != SOL_IPV6)
806 				continue;
807 			if (cm->cmsg_level == SOL_IP &&
808 			    cm->cmsg_type != IP_RECVERR)
809 				continue;
810 			if (cm->cmsg_level == SOL_IPV6 &&
811 			    cm->cmsg_type != IPV6_RECVERR)
812 				continue;
813 
814 			serr = (void *)CMSG_DATA(cm);
815 			if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY)
816 				error(1, 0, "wrong origin %u", serr->ee_origin);
817 			if (serr->ee_errno != 0)
818 				error(1, 0, "wrong errno %d", serr->ee_errno);
819 
820 			hi = serr->ee_data;
821 			lo = serr->ee_info;
822 
823 			fprintf(stderr, "tx complete [%d,%d]\n", lo, hi);
824 			return;
825 		}
826 	}
827 
828 	error(1, 0, "did not receive tx completion");
829 }
830 
831 static int do_client(struct memory_buffer *mem)
832 {
833 	char ctrl_data[CMSG_SPACE(sizeof(__u32))];
834 	struct sockaddr_in6 server_sin;
835 	struct sockaddr_in6 client_sin;
836 	struct ynl_sock *ys = NULL;
837 	struct msghdr msg = {};
838 	ssize_t line_size = 0;
839 	struct cmsghdr *cmsg;
840 	struct iovec iov[2];
841 	char *line = NULL;
842 	unsigned long mid;
843 	size_t len = 0;
844 	int socket_fd;
845 	__u32 ddmabuf;
846 	int opt = 1;
847 	int ret;
848 
849 	ret = parse_address(server_ip, atoi(port), &server_sin);
850 	if (ret < 0)
851 		error(1, 0, "parse server address");
852 
853 	socket_fd = socket(AF_INET6, SOCK_STREAM, 0);
854 	if (socket_fd < 0)
855 		error(1, socket_fd, "create socket");
856 
857 	enable_reuseaddr(socket_fd);
858 
859 	ret = setsockopt(socket_fd, SOL_SOCKET, SO_BINDTODEVICE, ifname,
860 			 strlen(ifname) + 1);
861 	if (ret)
862 		error(1, errno, "bindtodevice");
863 
864 	if (bind_tx_queue(ifindex, mem->fd, &ys))
865 		error(1, 0, "Failed to bind\n");
866 
867 	if (client_ip) {
868 		ret = parse_address(client_ip, atoi(port), &client_sin);
869 		if (ret < 0)
870 			error(1, 0, "parse client address");
871 
872 		ret = bind(socket_fd, &client_sin, sizeof(client_sin));
873 		if (ret)
874 			error(1, errno, "bind");
875 	}
876 
877 	ret = setsockopt(socket_fd, SOL_SOCKET, SO_ZEROCOPY, &opt, sizeof(opt));
878 	if (ret)
879 		error(1, errno, "set sock opt");
880 
881 	fprintf(stderr, "Connect to %s %d (via %s)\n", server_ip,
882 		ntohs(server_sin.sin6_port), ifname);
883 
884 	ret = connect(socket_fd, &server_sin, sizeof(server_sin));
885 	if (ret)
886 		error(1, errno, "connect");
887 
888 	while (1) {
889 		free(line);
890 		line = NULL;
891 		line_size = getline(&line, &len, stdin);
892 
893 		if (line_size < 0)
894 			break;
895 
896 		mid = (line_size / 2) + 1;
897 
898 		iov[0].iov_base = (void *)1;
899 		iov[0].iov_len = mid;
900 		iov[1].iov_base = (void *)(mid + 2);
901 		iov[1].iov_len = line_size - mid;
902 
903 		provider->memcpy_to_device(mem, (size_t)iov[0].iov_base, line,
904 					   iov[0].iov_len);
905 		provider->memcpy_to_device(mem, (size_t)iov[1].iov_base,
906 					   line + iov[0].iov_len,
907 					   iov[1].iov_len);
908 
909 		fprintf(stderr,
910 			"read line_size=%ld iov[0].iov_base=%lu, iov[0].iov_len=%lu, iov[1].iov_base=%lu, iov[1].iov_len=%lu\n",
911 			line_size, (unsigned long)iov[0].iov_base,
912 			iov[0].iov_len, (unsigned long)iov[1].iov_base,
913 			iov[1].iov_len);
914 
915 		msg.msg_iov = iov;
916 		msg.msg_iovlen = 2;
917 
918 		msg.msg_control = ctrl_data;
919 		msg.msg_controllen = sizeof(ctrl_data);
920 
921 		cmsg = CMSG_FIRSTHDR(&msg);
922 		cmsg->cmsg_level = SOL_SOCKET;
923 		cmsg->cmsg_type = SCM_DEVMEM_DMABUF;
924 		cmsg->cmsg_len = CMSG_LEN(sizeof(__u32));
925 
926 		ddmabuf = tx_dmabuf_id;
927 
928 		*((__u32 *)CMSG_DATA(cmsg)) = ddmabuf;
929 
930 		ret = sendmsg(socket_fd, &msg, MSG_ZEROCOPY);
931 		if (ret < 0)
932 			error(1, errno, "Failed sendmsg");
933 
934 		fprintf(stderr, "sendmsg_ret=%d\n", ret);
935 
936 		if (ret != line_size)
937 			error(1, errno, "Did not send all bytes");
938 
939 		wait_compl(socket_fd);
940 	}
941 
942 	fprintf(stderr, "%s: tx ok\n", TEST_PREFIX);
943 
944 	free(line);
945 	close(socket_fd);
946 
947 	if (ys)
948 		ynl_sock_destroy(ys);
949 
950 	return 0;
951 }
952 
953 int main(int argc, char *argv[])
954 {
955 	struct memory_buffer *mem;
956 	int is_server = 0, opt;
957 	int ret;
958 
959 	while ((opt = getopt(argc, argv, "ls:c:p:v:q:t:f:")) != -1) {
960 		switch (opt) {
961 		case 'l':
962 			is_server = 1;
963 			break;
964 		case 's':
965 			server_ip = optarg;
966 			break;
967 		case 'c':
968 			client_ip = optarg;
969 			break;
970 		case 'p':
971 			port = optarg;
972 			break;
973 		case 'v':
974 			do_validation = atoll(optarg);
975 			break;
976 		case 'q':
977 			num_queues = atoi(optarg);
978 			break;
979 		case 't':
980 			start_queue = atoi(optarg);
981 			break;
982 		case 'f':
983 			ifname = optarg;
984 			break;
985 		case '?':
986 			fprintf(stderr, "unknown option: %c\n", optopt);
987 			break;
988 		}
989 	}
990 
991 	if (!ifname)
992 		error(1, 0, "Missing -f argument\n");
993 
994 	ifindex = if_nametoindex(ifname);
995 
996 	fprintf(stderr, "using ifindex=%u\n", ifindex);
997 
998 	if (!server_ip && !client_ip) {
999 		if (start_queue < 0 && num_queues < 0) {
1000 			num_queues = rxq_num(ifindex);
1001 			if (num_queues < 0)
1002 				error(1, 0, "couldn't detect number of queues\n");
1003 			if (num_queues < 2)
1004 				error(1, 0,
1005 				      "number of device queues is too low\n");
1006 			/* make sure can bind to multiple queues */
1007 			start_queue = num_queues / 2;
1008 			num_queues /= 2;
1009 		}
1010 
1011 		if (start_queue < 0 || num_queues < 0)
1012 			error(1, 0, "Both -t and -q are required\n");
1013 
1014 		run_devmem_tests();
1015 		return 0;
1016 	}
1017 
1018 	if (start_queue < 0 && num_queues < 0) {
1019 		num_queues = rxq_num(ifindex);
1020 		if (num_queues < 2)
1021 			error(1, 0, "number of device queues is too low\n");
1022 
1023 		num_queues = 1;
1024 		start_queue = rxq_num(ifindex) - num_queues;
1025 
1026 		if (start_queue < 0)
1027 			error(1, 0, "couldn't detect number of queues\n");
1028 
1029 		fprintf(stderr, "using queues %d..%d\n", start_queue, start_queue + num_queues);
1030 	}
1031 
1032 	for (; optind < argc; optind++)
1033 		fprintf(stderr, "extra arguments: %s\n", argv[optind]);
1034 
1035 	if (start_queue < 0)
1036 		error(1, 0, "Missing -t argument\n");
1037 
1038 	if (num_queues < 0)
1039 		error(1, 0, "Missing -q argument\n");
1040 
1041 	if (!server_ip)
1042 		error(1, 0, "Missing -s argument\n");
1043 
1044 	if (!port)
1045 		error(1, 0, "Missing -p argument\n");
1046 
1047 	mem = provider->alloc(getpagesize() * NUM_PAGES);
1048 	ret = is_server ? do_server(mem) : do_client(mem);
1049 	provider->free(mem);
1050 
1051 	return ret;
1052 }
1053