Loading...
1/* Evaluate MSG_ZEROCOPY
2 *
3 * Send traffic between two processes over one of the supported
4 * protocols and modes:
5 *
6 * PF_INET/PF_INET6
7 * - SOCK_STREAM
8 * - SOCK_DGRAM
9 * - SOCK_DGRAM with UDP_CORK
10 * - SOCK_RAW
11 * - SOCK_RAW with IP_HDRINCL
12 *
13 * PF_PACKET
14 * - SOCK_DGRAM
15 * - SOCK_RAW
16 *
17 * PF_RDS
18 * - SOCK_SEQPACKET
19 *
20 * Start this program on two connected hosts, one in send mode and
21 * the other with option '-r' to put it in receiver mode.
22 *
23 * If zerocopy mode ('-z') is enabled, the sender will verify that
24 * the kernel queues completions on the error queue for all zerocopy
25 * transfers.
26 */
27
28#define _GNU_SOURCE
29
30#include <arpa/inet.h>
31#include <error.h>
32#include <errno.h>
33#include <limits.h>
34#include <linux/errqueue.h>
35#include <linux/if_packet.h>
36#include <linux/ipv6.h>
37#include <linux/socket.h>
38#include <linux/sockios.h>
39#include <net/ethernet.h>
40#include <net/if.h>
41#include <netinet/ip.h>
42#include <netinet/ip6.h>
43#include <netinet/tcp.h>
44#include <netinet/udp.h>
45#include <poll.h>
46#include <sched.h>
47#include <stdbool.h>
48#include <stdio.h>
49#include <stdint.h>
50#include <stdlib.h>
51#include <string.h>
52#include <sys/ioctl.h>
53#include <sys/socket.h>
54#include <sys/stat.h>
55#include <sys/time.h>
56#include <sys/types.h>
57#include <sys/wait.h>
58#include <unistd.h>
59#include <linux/rds.h>
60
61#ifndef SO_EE_ORIGIN_ZEROCOPY
62#define SO_EE_ORIGIN_ZEROCOPY 5
63#endif
64
65#ifndef SO_ZEROCOPY
66#define SO_ZEROCOPY 60
67#endif
68
69#ifndef SO_EE_CODE_ZEROCOPY_COPIED
70#define SO_EE_CODE_ZEROCOPY_COPIED 1
71#endif
72
73#ifndef MSG_ZEROCOPY
74#define MSG_ZEROCOPY 0x4000000
75#endif
76
77static int cfg_cork;
78static bool cfg_cork_mixed;
79static int cfg_cpu = -1; /* default: pin to last cpu */
80static int cfg_family = PF_UNSPEC;
81static int cfg_ifindex = 1;
82static int cfg_payload_len;
83static int cfg_port = 8000;
84static bool cfg_rx;
85static int cfg_runtime_ms = 4200;
86static int cfg_verbose;
87static int cfg_waittime_ms = 500;
88static bool cfg_zerocopy;
89
90static socklen_t cfg_alen;
91static struct sockaddr_storage cfg_dst_addr;
92static struct sockaddr_storage cfg_src_addr;
93
94static char payload[IP_MAXPACKET];
95static long packets, bytes, completions, expected_completions;
96static int zerocopied = -1;
97static uint32_t next_completion;
98
99static unsigned long gettimeofday_ms(void)
100{
101 struct timeval tv;
102
103 gettimeofday(&tv, NULL);
104 return (tv.tv_sec * 1000) + (tv.tv_usec / 1000);
105}
106
107static uint16_t get_ip_csum(const uint16_t *start, int num_words)
108{
109 unsigned long sum = 0;
110 int i;
111
112 for (i = 0; i < num_words; i++)
113 sum += start[i];
114
115 while (sum >> 16)
116 sum = (sum & 0xFFFF) + (sum >> 16);
117
118 return ~sum;
119}
120
121static int do_setcpu(int cpu)
122{
123 cpu_set_t mask;
124
125 CPU_ZERO(&mask);
126 CPU_SET(cpu, &mask);
127 if (sched_setaffinity(0, sizeof(mask), &mask))
128 fprintf(stderr, "cpu: unable to pin, may increase variance.\n");
129 else if (cfg_verbose)
130 fprintf(stderr, "cpu: %u\n", cpu);
131
132 return 0;
133}
134
135static void do_setsockopt(int fd, int level, int optname, int val)
136{
137 if (setsockopt(fd, level, optname, &val, sizeof(val)))
138 error(1, errno, "setsockopt %d.%d: %d", level, optname, val);
139}
140
141static int do_poll(int fd, int events)
142{
143 struct pollfd pfd;
144 int ret;
145
146 pfd.events = events;
147 pfd.revents = 0;
148 pfd.fd = fd;
149
150 ret = poll(&pfd, 1, cfg_waittime_ms);
151 if (ret == -1)
152 error(1, errno, "poll");
153
154 return ret && (pfd.revents & events);
155}
156
157static int do_accept(int fd)
158{
159 int fda = fd;
160
161 fd = accept(fda, NULL, NULL);
162 if (fd == -1)
163 error(1, errno, "accept");
164 if (close(fda))
165 error(1, errno, "close listen sock");
166
167 return fd;
168}
169
170static void add_zcopy_cookie(struct msghdr *msg, uint32_t cookie)
171{
172 struct cmsghdr *cm;
173
174 if (!msg->msg_control)
175 error(1, errno, "NULL cookie");
176 cm = (void *)msg->msg_control;
177 cm->cmsg_len = CMSG_LEN(sizeof(cookie));
178 cm->cmsg_level = SOL_RDS;
179 cm->cmsg_type = RDS_CMSG_ZCOPY_COOKIE;
180 memcpy(CMSG_DATA(cm), &cookie, sizeof(cookie));
181}
182
183static bool do_sendmsg(int fd, struct msghdr *msg, bool do_zerocopy, int domain)
184{
185 int ret, len, i, flags;
186 static uint32_t cookie;
187 char ckbuf[CMSG_SPACE(sizeof(cookie))];
188
189 len = 0;
190 for (i = 0; i < msg->msg_iovlen; i++)
191 len += msg->msg_iov[i].iov_len;
192
193 flags = MSG_DONTWAIT;
194 if (do_zerocopy) {
195 flags |= MSG_ZEROCOPY;
196 if (domain == PF_RDS) {
197 memset(&msg->msg_control, 0, sizeof(msg->msg_control));
198 msg->msg_controllen = CMSG_SPACE(sizeof(cookie));
199 msg->msg_control = (struct cmsghdr *)ckbuf;
200 add_zcopy_cookie(msg, ++cookie);
201 }
202 }
203
204 ret = sendmsg(fd, msg, flags);
205 if (ret == -1 && errno == EAGAIN)
206 return false;
207 if (ret == -1)
208 error(1, errno, "send");
209 if (cfg_verbose && ret != len)
210 fprintf(stderr, "send: ret=%u != %u\n", ret, len);
211
212 if (len) {
213 packets++;
214 bytes += ret;
215 if (do_zerocopy && ret)
216 expected_completions++;
217 }
218 if (do_zerocopy && domain == PF_RDS) {
219 msg->msg_control = NULL;
220 msg->msg_controllen = 0;
221 }
222
223 return true;
224}
225
226static void do_sendmsg_corked(int fd, struct msghdr *msg)
227{
228 bool do_zerocopy = cfg_zerocopy;
229 int i, payload_len, extra_len;
230
231 /* split up the packet. for non-multiple, make first buffer longer */
232 payload_len = cfg_payload_len / cfg_cork;
233 extra_len = cfg_payload_len - (cfg_cork * payload_len);
234
235 do_setsockopt(fd, IPPROTO_UDP, UDP_CORK, 1);
236
237 for (i = 0; i < cfg_cork; i++) {
238
239 /* in mixed-frags mode, alternate zerocopy and copy frags
240 * start with non-zerocopy, to ensure attach later works
241 */
242 if (cfg_cork_mixed)
243 do_zerocopy = (i & 1);
244
245 msg->msg_iov[0].iov_len = payload_len + extra_len;
246 extra_len = 0;
247
248 do_sendmsg(fd, msg, do_zerocopy,
249 (cfg_dst_addr.ss_family == AF_INET ?
250 PF_INET : PF_INET6));
251 }
252
253 do_setsockopt(fd, IPPROTO_UDP, UDP_CORK, 0);
254}
255
256static int setup_iph(struct iphdr *iph, uint16_t payload_len)
257{
258 struct sockaddr_in *daddr = (void *) &cfg_dst_addr;
259 struct sockaddr_in *saddr = (void *) &cfg_src_addr;
260
261 memset(iph, 0, sizeof(*iph));
262
263 iph->version = 4;
264 iph->tos = 0;
265 iph->ihl = 5;
266 iph->ttl = 2;
267 iph->saddr = saddr->sin_addr.s_addr;
268 iph->daddr = daddr->sin_addr.s_addr;
269 iph->protocol = IPPROTO_EGP;
270 iph->tot_len = htons(sizeof(*iph) + payload_len);
271 iph->check = get_ip_csum((void *) iph, iph->ihl << 1);
272
273 return sizeof(*iph);
274}
275
276static int setup_ip6h(struct ipv6hdr *ip6h, uint16_t payload_len)
277{
278 struct sockaddr_in6 *daddr = (void *) &cfg_dst_addr;
279 struct sockaddr_in6 *saddr = (void *) &cfg_src_addr;
280
281 memset(ip6h, 0, sizeof(*ip6h));
282
283 ip6h->version = 6;
284 ip6h->payload_len = htons(payload_len);
285 ip6h->nexthdr = IPPROTO_EGP;
286 ip6h->hop_limit = 2;
287 ip6h->saddr = saddr->sin6_addr;
288 ip6h->daddr = daddr->sin6_addr;
289
290 return sizeof(*ip6h);
291}
292
293
294static void setup_sockaddr(int domain, const char *str_addr,
295 struct sockaddr_storage *sockaddr)
296{
297 struct sockaddr_in6 *addr6 = (void *) sockaddr;
298 struct sockaddr_in *addr4 = (void *) sockaddr;
299
300 switch (domain) {
301 case PF_INET:
302 memset(addr4, 0, sizeof(*addr4));
303 addr4->sin_family = AF_INET;
304 addr4->sin_port = htons(cfg_port);
305 if (str_addr &&
306 inet_pton(AF_INET, str_addr, &(addr4->sin_addr)) != 1)
307 error(1, 0, "ipv4 parse error: %s", str_addr);
308 break;
309 case PF_INET6:
310 memset(addr6, 0, sizeof(*addr6));
311 addr6->sin6_family = AF_INET6;
312 addr6->sin6_port = htons(cfg_port);
313 if (str_addr &&
314 inet_pton(AF_INET6, str_addr, &(addr6->sin6_addr)) != 1)
315 error(1, 0, "ipv6 parse error: %s", str_addr);
316 break;
317 default:
318 error(1, 0, "illegal domain");
319 }
320}
321
322static int do_setup_tx(int domain, int type, int protocol)
323{
324 int fd;
325
326 fd = socket(domain, type, protocol);
327 if (fd == -1)
328 error(1, errno, "socket t");
329
330 do_setsockopt(fd, SOL_SOCKET, SO_SNDBUF, 1 << 21);
331 if (cfg_zerocopy)
332 do_setsockopt(fd, SOL_SOCKET, SO_ZEROCOPY, 1);
333
334 if (domain != PF_PACKET && domain != PF_RDS)
335 if (connect(fd, (void *) &cfg_dst_addr, cfg_alen))
336 error(1, errno, "connect");
337
338 if (domain == PF_RDS) {
339 if (bind(fd, (void *) &cfg_src_addr, cfg_alen))
340 error(1, errno, "bind");
341 }
342
343 return fd;
344}
345
346static uint32_t do_process_zerocopy_cookies(struct rds_zcopy_cookies *ck)
347{
348 int i;
349
350 if (ck->num > RDS_MAX_ZCOOKIES)
351 error(1, 0, "Returned %d cookies, max expected %d\n",
352 ck->num, RDS_MAX_ZCOOKIES);
353 for (i = 0; i < ck->num; i++)
354 if (cfg_verbose >= 2)
355 fprintf(stderr, "%d\n", ck->cookies[i]);
356 return ck->num;
357}
358
359static bool do_recvmsg_completion(int fd)
360{
361 char cmsgbuf[CMSG_SPACE(sizeof(struct rds_zcopy_cookies))];
362 struct rds_zcopy_cookies *ck;
363 struct cmsghdr *cmsg;
364 struct msghdr msg;
365 bool ret = false;
366
367 memset(&msg, 0, sizeof(msg));
368 msg.msg_control = cmsgbuf;
369 msg.msg_controllen = sizeof(cmsgbuf);
370
371 if (recvmsg(fd, &msg, MSG_DONTWAIT))
372 return ret;
373
374 if (msg.msg_flags & MSG_CTRUNC)
375 error(1, errno, "recvmsg notification: truncated");
376
377 for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
378 if (cmsg->cmsg_level == SOL_RDS &&
379 cmsg->cmsg_type == RDS_CMSG_ZCOPY_COMPLETION) {
380
381 ck = (struct rds_zcopy_cookies *)CMSG_DATA(cmsg);
382 completions += do_process_zerocopy_cookies(ck);
383 ret = true;
384 break;
385 }
386 error(0, 0, "ignoring cmsg at level %d type %d\n",
387 cmsg->cmsg_level, cmsg->cmsg_type);
388 }
389 return ret;
390}
391
392static bool do_recv_completion(int fd, int domain)
393{
394 struct sock_extended_err *serr;
395 struct msghdr msg = {};
396 struct cmsghdr *cm;
397 uint32_t hi, lo, range;
398 int ret, zerocopy;
399 char control[100];
400
401 if (domain == PF_RDS)
402 return do_recvmsg_completion(fd);
403
404 msg.msg_control = control;
405 msg.msg_controllen = sizeof(control);
406
407 ret = recvmsg(fd, &msg, MSG_ERRQUEUE);
408 if (ret == -1 && errno == EAGAIN)
409 return false;
410 if (ret == -1)
411 error(1, errno, "recvmsg notification");
412 if (msg.msg_flags & MSG_CTRUNC)
413 error(1, errno, "recvmsg notification: truncated");
414
415 cm = CMSG_FIRSTHDR(&msg);
416 if (!cm)
417 error(1, 0, "cmsg: no cmsg");
418 if (!((cm->cmsg_level == SOL_IP && cm->cmsg_type == IP_RECVERR) ||
419 (cm->cmsg_level == SOL_IPV6 && cm->cmsg_type == IPV6_RECVERR) ||
420 (cm->cmsg_level == SOL_PACKET && cm->cmsg_type == PACKET_TX_TIMESTAMP)))
421 error(1, 0, "serr: wrong type: %d.%d",
422 cm->cmsg_level, cm->cmsg_type);
423
424 serr = (void *) CMSG_DATA(cm);
425
426 if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY)
427 error(1, 0, "serr: wrong origin: %u", serr->ee_origin);
428 if (serr->ee_errno != 0)
429 error(1, 0, "serr: wrong error code: %u", serr->ee_errno);
430
431 hi = serr->ee_data;
432 lo = serr->ee_info;
433 range = hi - lo + 1;
434
435 /* Detect notification gaps. These should not happen often, if at all.
436 * Gaps can occur due to drops, reordering and retransmissions.
437 */
438 if (lo != next_completion)
439 fprintf(stderr, "gap: %u..%u does not append to %u\n",
440 lo, hi, next_completion);
441 next_completion = hi + 1;
442
443 zerocopy = !(serr->ee_code & SO_EE_CODE_ZEROCOPY_COPIED);
444 if (zerocopied == -1)
445 zerocopied = zerocopy;
446 else if (zerocopied != zerocopy) {
447 fprintf(stderr, "serr: inconsistent\n");
448 zerocopied = zerocopy;
449 }
450
451 if (cfg_verbose >= 2)
452 fprintf(stderr, "completed: %u (h=%u l=%u)\n",
453 range, hi, lo);
454
455 completions += range;
456 return true;
457}
458
459/* Read all outstanding messages on the errqueue */
460static void do_recv_completions(int fd, int domain)
461{
462 while (do_recv_completion(fd, domain)) {}
463}
464
465/* Wait for all remaining completions on the errqueue */
466static void do_recv_remaining_completions(int fd, int domain)
467{
468 int64_t tstop = gettimeofday_ms() + cfg_waittime_ms;
469
470 while (completions < expected_completions &&
471 gettimeofday_ms() < tstop) {
472 if (do_poll(fd, domain == PF_RDS ? POLLIN : POLLERR))
473 do_recv_completions(fd, domain);
474 }
475
476 if (completions < expected_completions)
477 fprintf(stderr, "missing notifications: %lu < %lu\n",
478 completions, expected_completions);
479}
480
481static void do_tx(int domain, int type, int protocol)
482{
483 struct iovec iov[3] = { {0} };
484 struct sockaddr_ll laddr;
485 struct msghdr msg = {0};
486 struct ethhdr eth;
487 union {
488 struct ipv6hdr ip6h;
489 struct iphdr iph;
490 } nh;
491 uint64_t tstop;
492 int fd;
493
494 fd = do_setup_tx(domain, type, protocol);
495
496 if (domain == PF_PACKET) {
497 uint16_t proto = cfg_family == PF_INET ? ETH_P_IP : ETH_P_IPV6;
498
499 /* sock_raw passes ll header as data */
500 if (type == SOCK_RAW) {
501 memset(eth.h_dest, 0x06, ETH_ALEN);
502 memset(eth.h_source, 0x02, ETH_ALEN);
503 eth.h_proto = htons(proto);
504 iov[0].iov_base = ð
505 iov[0].iov_len = sizeof(eth);
506 msg.msg_iovlen++;
507 }
508
509 /* both sock_raw and sock_dgram expect name */
510 memset(&laddr, 0, sizeof(laddr));
511 laddr.sll_family = AF_PACKET;
512 laddr.sll_ifindex = cfg_ifindex;
513 laddr.sll_protocol = htons(proto);
514 laddr.sll_halen = ETH_ALEN;
515
516 memset(laddr.sll_addr, 0x06, ETH_ALEN);
517
518 msg.msg_name = &laddr;
519 msg.msg_namelen = sizeof(laddr);
520 }
521
522 /* packet and raw sockets with hdrincl must pass network header */
523 if (domain == PF_PACKET || protocol == IPPROTO_RAW) {
524 if (cfg_family == PF_INET)
525 iov[1].iov_len = setup_iph(&nh.iph, cfg_payload_len);
526 else
527 iov[1].iov_len = setup_ip6h(&nh.ip6h, cfg_payload_len);
528
529 iov[1].iov_base = (void *) &nh;
530 msg.msg_iovlen++;
531 }
532
533 if (domain == PF_RDS) {
534 msg.msg_name = &cfg_dst_addr;
535 msg.msg_namelen = (cfg_dst_addr.ss_family == AF_INET ?
536 sizeof(struct sockaddr_in) :
537 sizeof(struct sockaddr_in6));
538 }
539
540 iov[2].iov_base = payload;
541 iov[2].iov_len = cfg_payload_len;
542 msg.msg_iovlen++;
543 msg.msg_iov = &iov[3 - msg.msg_iovlen];
544
545 tstop = gettimeofday_ms() + cfg_runtime_ms;
546 do {
547 if (cfg_cork)
548 do_sendmsg_corked(fd, &msg);
549 else
550 do_sendmsg(fd, &msg, cfg_zerocopy, domain);
551
552 while (!do_poll(fd, POLLOUT)) {
553 if (cfg_zerocopy)
554 do_recv_completions(fd, domain);
555 }
556
557 } while (gettimeofday_ms() < tstop);
558
559 if (cfg_zerocopy)
560 do_recv_remaining_completions(fd, domain);
561
562 if (close(fd))
563 error(1, errno, "close");
564
565 fprintf(stderr, "tx=%lu (%lu MB) txc=%lu zc=%c\n",
566 packets, bytes >> 20, completions,
567 zerocopied == 1 ? 'y' : 'n');
568}
569
570static int do_setup_rx(int domain, int type, int protocol)
571{
572 int fd;
573
574 /* If tx over PF_PACKET, rx over PF_INET(6)/SOCK_RAW,
575 * to recv the only copy of the packet, not a clone
576 */
577 if (domain == PF_PACKET)
578 error(1, 0, "Use PF_INET/SOCK_RAW to read");
579
580 if (type == SOCK_RAW && protocol == IPPROTO_RAW)
581 error(1, 0, "IPPROTO_RAW: not supported on Rx");
582
583 fd = socket(domain, type, protocol);
584 if (fd == -1)
585 error(1, errno, "socket r");
586
587 do_setsockopt(fd, SOL_SOCKET, SO_RCVBUF, 1 << 21);
588 do_setsockopt(fd, SOL_SOCKET, SO_RCVLOWAT, 1 << 16);
589 do_setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, 1);
590
591 if (bind(fd, (void *) &cfg_dst_addr, cfg_alen))
592 error(1, errno, "bind");
593
594 if (type == SOCK_STREAM) {
595 if (listen(fd, 1))
596 error(1, errno, "listen");
597 fd = do_accept(fd);
598 }
599
600 return fd;
601}
602
603/* Flush all outstanding bytes for the tcp receive queue */
604static void do_flush_tcp(int fd)
605{
606 int ret;
607
608 /* MSG_TRUNC flushes up to len bytes */
609 ret = recv(fd, NULL, 1 << 21, MSG_TRUNC | MSG_DONTWAIT);
610 if (ret == -1 && errno == EAGAIN)
611 return;
612 if (ret == -1)
613 error(1, errno, "flush");
614 if (!ret)
615 return;
616
617 packets++;
618 bytes += ret;
619}
620
621/* Flush all outstanding datagrams. Verify first few bytes of each. */
622static void do_flush_datagram(int fd, int type)
623{
624 int ret, off = 0;
625 char buf[64];
626
627 /* MSG_TRUNC will return full datagram length */
628 ret = recv(fd, buf, sizeof(buf), MSG_DONTWAIT | MSG_TRUNC);
629 if (ret == -1 && errno == EAGAIN)
630 return;
631
632 /* raw ipv4 return with header, raw ipv6 without */
633 if (cfg_family == PF_INET && type == SOCK_RAW) {
634 off += sizeof(struct iphdr);
635 ret -= sizeof(struct iphdr);
636 }
637
638 if (ret == -1)
639 error(1, errno, "recv");
640 if (ret != cfg_payload_len)
641 error(1, 0, "recv: ret=%u != %u", ret, cfg_payload_len);
642 if (ret > sizeof(buf) - off)
643 ret = sizeof(buf) - off;
644 if (memcmp(buf + off, payload, ret))
645 error(1, 0, "recv: data mismatch");
646
647 packets++;
648 bytes += cfg_payload_len;
649}
650
651static void do_rx(int domain, int type, int protocol)
652{
653 const int cfg_receiver_wait_ms = 400;
654 uint64_t tstop;
655 int fd;
656
657 fd = do_setup_rx(domain, type, protocol);
658
659 tstop = gettimeofday_ms() + cfg_runtime_ms + cfg_receiver_wait_ms;
660 do {
661 if (type == SOCK_STREAM)
662 do_flush_tcp(fd);
663 else
664 do_flush_datagram(fd, type);
665
666 do_poll(fd, POLLIN);
667
668 } while (gettimeofday_ms() < tstop);
669
670 if (close(fd))
671 error(1, errno, "close");
672
673 fprintf(stderr, "rx=%lu (%lu MB)\n", packets, bytes >> 20);
674}
675
676static void do_test(int domain, int type, int protocol)
677{
678 int i;
679
680 if (cfg_cork && (domain == PF_PACKET || type != SOCK_DGRAM))
681 error(1, 0, "can only cork udp sockets");
682
683 do_setcpu(cfg_cpu);
684
685 for (i = 0; i < IP_MAXPACKET; i++)
686 payload[i] = 'a' + (i % 26);
687
688 if (cfg_rx)
689 do_rx(domain, type, protocol);
690 else
691 do_tx(domain, type, protocol);
692}
693
694static void usage(const char *filepath)
695{
696 error(1, 0, "Usage: %s [options] <test>", filepath);
697}
698
699static void parse_opts(int argc, char **argv)
700{
701 const int max_payload_len = sizeof(payload) -
702 sizeof(struct ipv6hdr) -
703 sizeof(struct tcphdr) -
704 40 /* max tcp options */;
705 int c;
706 char *daddr = NULL, *saddr = NULL;
707 char *cfg_test;
708
709 cfg_payload_len = max_payload_len;
710
711 while ((c = getopt(argc, argv, "46c:C:D:i:mp:rs:S:t:vz")) != -1) {
712 switch (c) {
713 case '4':
714 if (cfg_family != PF_UNSPEC)
715 error(1, 0, "Pass one of -4 or -6");
716 cfg_family = PF_INET;
717 cfg_alen = sizeof(struct sockaddr_in);
718 break;
719 case '6':
720 if (cfg_family != PF_UNSPEC)
721 error(1, 0, "Pass one of -4 or -6");
722 cfg_family = PF_INET6;
723 cfg_alen = sizeof(struct sockaddr_in6);
724 break;
725 case 'c':
726 cfg_cork = strtol(optarg, NULL, 0);
727 break;
728 case 'C':
729 cfg_cpu = strtol(optarg, NULL, 0);
730 break;
731 case 'D':
732 daddr = optarg;
733 break;
734 case 'i':
735 cfg_ifindex = if_nametoindex(optarg);
736 if (cfg_ifindex == 0)
737 error(1, errno, "invalid iface: %s", optarg);
738 break;
739 case 'm':
740 cfg_cork_mixed = true;
741 break;
742 case 'p':
743 cfg_port = strtoul(optarg, NULL, 0);
744 break;
745 case 'r':
746 cfg_rx = true;
747 break;
748 case 's':
749 cfg_payload_len = strtoul(optarg, NULL, 0);
750 break;
751 case 'S':
752 saddr = optarg;
753 break;
754 case 't':
755 cfg_runtime_ms = 200 + strtoul(optarg, NULL, 10) * 1000;
756 break;
757 case 'v':
758 cfg_verbose++;
759 break;
760 case 'z':
761 cfg_zerocopy = true;
762 break;
763 }
764 }
765
766 cfg_test = argv[argc - 1];
767 if (strcmp(cfg_test, "rds") == 0) {
768 if (!daddr)
769 error(1, 0, "-D <server addr> required for PF_RDS\n");
770 if (!cfg_rx && !saddr)
771 error(1, 0, "-S <client addr> required for PF_RDS\n");
772 }
773 setup_sockaddr(cfg_family, daddr, &cfg_dst_addr);
774 setup_sockaddr(cfg_family, saddr, &cfg_src_addr);
775
776 if (cfg_payload_len > max_payload_len)
777 error(1, 0, "-s: payload exceeds max (%d)", max_payload_len);
778 if (cfg_cork_mixed && (!cfg_zerocopy || !cfg_cork))
779 error(1, 0, "-m: cork_mixed requires corking and zerocopy");
780
781 if (optind != argc - 1)
782 usage(argv[0]);
783}
784
785int main(int argc, char **argv)
786{
787 const char *cfg_test;
788
789 parse_opts(argc, argv);
790
791 cfg_test = argv[argc - 1];
792
793 if (!strcmp(cfg_test, "packet"))
794 do_test(PF_PACKET, SOCK_RAW, 0);
795 else if (!strcmp(cfg_test, "packet_dgram"))
796 do_test(PF_PACKET, SOCK_DGRAM, 0);
797 else if (!strcmp(cfg_test, "raw"))
798 do_test(cfg_family, SOCK_RAW, IPPROTO_EGP);
799 else if (!strcmp(cfg_test, "raw_hdrincl"))
800 do_test(cfg_family, SOCK_RAW, IPPROTO_RAW);
801 else if (!strcmp(cfg_test, "tcp"))
802 do_test(cfg_family, SOCK_STREAM, 0);
803 else if (!strcmp(cfg_test, "udp"))
804 do_test(cfg_family, SOCK_DGRAM, 0);
805 else if (!strcmp(cfg_test, "rds"))
806 do_test(PF_RDS, SOCK_SEQPACKET, 0);
807 else
808 error(1, 0, "unknown cfg_test %s", cfg_test);
809
810 return 0;
811}
1/* Evaluate MSG_ZEROCOPY
2 *
3 * Send traffic between two processes over one of the supported
4 * protocols and modes:
5 *
6 * PF_INET/PF_INET6
7 * - SOCK_STREAM
8 * - SOCK_DGRAM
9 * - SOCK_DGRAM with UDP_CORK
10 * - SOCK_RAW
11 * - SOCK_RAW with IP_HDRINCL
12 *
13 * PF_PACKET
14 * - SOCK_DGRAM
15 * - SOCK_RAW
16 *
17 * PF_RDS
18 * - SOCK_SEQPACKET
19 *
20 * Start this program on two connected hosts, one in send mode and
21 * the other with option '-r' to put it in receiver mode.
22 *
23 * If zerocopy mode ('-z') is enabled, the sender will verify that
24 * the kernel queues completions on the error queue for all zerocopy
25 * transfers.
26 */
27
28#define _GNU_SOURCE
29
30#include <arpa/inet.h>
31#include <error.h>
32#include <errno.h>
33#include <limits.h>
34#include <linux/errqueue.h>
35#include <linux/if_packet.h>
36#include <linux/ipv6.h>
37#include <linux/socket.h>
38#include <linux/sockios.h>
39#include <net/ethernet.h>
40#include <net/if.h>
41#include <netinet/ip.h>
42#include <netinet/ip6.h>
43#include <netinet/tcp.h>
44#include <netinet/udp.h>
45#include <poll.h>
46#include <sched.h>
47#include <stdbool.h>
48#include <stdio.h>
49#include <stdint.h>
50#include <stdlib.h>
51#include <string.h>
52#include <sys/ioctl.h>
53#include <sys/socket.h>
54#include <sys/stat.h>
55#include <sys/time.h>
56#include <sys/types.h>
57#include <sys/wait.h>
58#include <unistd.h>
59#include <linux/rds.h>
60
61#ifndef SO_EE_ORIGIN_ZEROCOPY
62#define SO_EE_ORIGIN_ZEROCOPY 5
63#endif
64
65#ifndef SO_ZEROCOPY
66#define SO_ZEROCOPY 60
67#endif
68
69#ifndef SO_EE_CODE_ZEROCOPY_COPIED
70#define SO_EE_CODE_ZEROCOPY_COPIED 1
71#endif
72
73#ifndef MSG_ZEROCOPY
74#define MSG_ZEROCOPY 0x4000000
75#endif
76
77static int cfg_cork;
78static bool cfg_cork_mixed;
79static int cfg_cpu = -1; /* default: pin to last cpu */
80static int cfg_family = PF_UNSPEC;
81static int cfg_ifindex = 1;
82static int cfg_payload_len;
83static int cfg_port = 8000;
84static bool cfg_rx;
85static int cfg_runtime_ms = 4200;
86static int cfg_verbose;
87static int cfg_waittime_ms = 500;
88static bool cfg_zerocopy;
89
90static socklen_t cfg_alen;
91static struct sockaddr_storage cfg_dst_addr;
92static struct sockaddr_storage cfg_src_addr;
93
94static char payload[IP_MAXPACKET];
95static long packets, bytes, completions, expected_completions;
96static int zerocopied = -1;
97static uint32_t next_completion;
98
99static unsigned long gettimeofday_ms(void)
100{
101 struct timeval tv;
102
103 gettimeofday(&tv, NULL);
104 return (tv.tv_sec * 1000) + (tv.tv_usec / 1000);
105}
106
107static uint16_t get_ip_csum(const uint16_t *start, int num_words)
108{
109 unsigned long sum = 0;
110 int i;
111
112 for (i = 0; i < num_words; i++)
113 sum += start[i];
114
115 while (sum >> 16)
116 sum = (sum & 0xFFFF) + (sum >> 16);
117
118 return ~sum;
119}
120
121static int do_setcpu(int cpu)
122{
123 cpu_set_t mask;
124
125 CPU_ZERO(&mask);
126 CPU_SET(cpu, &mask);
127 if (sched_setaffinity(0, sizeof(mask), &mask))
128 error(1, 0, "setaffinity %d", cpu);
129
130 if (cfg_verbose)
131 fprintf(stderr, "cpu: %u\n", cpu);
132
133 return 0;
134}
135
136static void do_setsockopt(int fd, int level, int optname, int val)
137{
138 if (setsockopt(fd, level, optname, &val, sizeof(val)))
139 error(1, errno, "setsockopt %d.%d: %d", level, optname, val);
140}
141
142static int do_poll(int fd, int events)
143{
144 struct pollfd pfd;
145 int ret;
146
147 pfd.events = events;
148 pfd.revents = 0;
149 pfd.fd = fd;
150
151 ret = poll(&pfd, 1, cfg_waittime_ms);
152 if (ret == -1)
153 error(1, errno, "poll");
154
155 return ret && (pfd.revents & events);
156}
157
158static int do_accept(int fd)
159{
160 int fda = fd;
161
162 fd = accept(fda, NULL, NULL);
163 if (fd == -1)
164 error(1, errno, "accept");
165 if (close(fda))
166 error(1, errno, "close listen sock");
167
168 return fd;
169}
170
171static void add_zcopy_cookie(struct msghdr *msg, uint32_t cookie)
172{
173 struct cmsghdr *cm;
174
175 if (!msg->msg_control)
176 error(1, errno, "NULL cookie");
177 cm = (void *)msg->msg_control;
178 cm->cmsg_len = CMSG_LEN(sizeof(cookie));
179 cm->cmsg_level = SOL_RDS;
180 cm->cmsg_type = RDS_CMSG_ZCOPY_COOKIE;
181 memcpy(CMSG_DATA(cm), &cookie, sizeof(cookie));
182}
183
184static bool do_sendmsg(int fd, struct msghdr *msg, bool do_zerocopy, int domain)
185{
186 int ret, len, i, flags;
187 static uint32_t cookie;
188 char ckbuf[CMSG_SPACE(sizeof(cookie))];
189
190 len = 0;
191 for (i = 0; i < msg->msg_iovlen; i++)
192 len += msg->msg_iov[i].iov_len;
193
194 flags = MSG_DONTWAIT;
195 if (do_zerocopy) {
196 flags |= MSG_ZEROCOPY;
197 if (domain == PF_RDS) {
198 memset(&msg->msg_control, 0, sizeof(msg->msg_control));
199 msg->msg_controllen = CMSG_SPACE(sizeof(cookie));
200 msg->msg_control = (struct cmsghdr *)ckbuf;
201 add_zcopy_cookie(msg, ++cookie);
202 }
203 }
204
205 ret = sendmsg(fd, msg, flags);
206 if (ret == -1 && errno == EAGAIN)
207 return false;
208 if (ret == -1)
209 error(1, errno, "send");
210 if (cfg_verbose && ret != len)
211 fprintf(stderr, "send: ret=%u != %u\n", ret, len);
212
213 if (len) {
214 packets++;
215 bytes += ret;
216 if (do_zerocopy && ret)
217 expected_completions++;
218 }
219 if (do_zerocopy && domain == PF_RDS) {
220 msg->msg_control = NULL;
221 msg->msg_controllen = 0;
222 }
223
224 return true;
225}
226
227static void do_sendmsg_corked(int fd, struct msghdr *msg)
228{
229 bool do_zerocopy = cfg_zerocopy;
230 int i, payload_len, extra_len;
231
232 /* split up the packet. for non-multiple, make first buffer longer */
233 payload_len = cfg_payload_len / cfg_cork;
234 extra_len = cfg_payload_len - (cfg_cork * payload_len);
235
236 do_setsockopt(fd, IPPROTO_UDP, UDP_CORK, 1);
237
238 for (i = 0; i < cfg_cork; i++) {
239
240 /* in mixed-frags mode, alternate zerocopy and copy frags
241 * start with non-zerocopy, to ensure attach later works
242 */
243 if (cfg_cork_mixed)
244 do_zerocopy = (i & 1);
245
246 msg->msg_iov[0].iov_len = payload_len + extra_len;
247 extra_len = 0;
248
249 do_sendmsg(fd, msg, do_zerocopy,
250 (cfg_dst_addr.ss_family == AF_INET ?
251 PF_INET : PF_INET6));
252 }
253
254 do_setsockopt(fd, IPPROTO_UDP, UDP_CORK, 0);
255}
256
257static int setup_iph(struct iphdr *iph, uint16_t payload_len)
258{
259 struct sockaddr_in *daddr = (void *) &cfg_dst_addr;
260 struct sockaddr_in *saddr = (void *) &cfg_src_addr;
261
262 memset(iph, 0, sizeof(*iph));
263
264 iph->version = 4;
265 iph->tos = 0;
266 iph->ihl = 5;
267 iph->ttl = 2;
268 iph->saddr = saddr->sin_addr.s_addr;
269 iph->daddr = daddr->sin_addr.s_addr;
270 iph->protocol = IPPROTO_EGP;
271 iph->tot_len = htons(sizeof(*iph) + payload_len);
272 iph->check = get_ip_csum((void *) iph, iph->ihl << 1);
273
274 return sizeof(*iph);
275}
276
277static int setup_ip6h(struct ipv6hdr *ip6h, uint16_t payload_len)
278{
279 struct sockaddr_in6 *daddr = (void *) &cfg_dst_addr;
280 struct sockaddr_in6 *saddr = (void *) &cfg_src_addr;
281
282 memset(ip6h, 0, sizeof(*ip6h));
283
284 ip6h->version = 6;
285 ip6h->payload_len = htons(payload_len);
286 ip6h->nexthdr = IPPROTO_EGP;
287 ip6h->hop_limit = 2;
288 ip6h->saddr = saddr->sin6_addr;
289 ip6h->daddr = daddr->sin6_addr;
290
291 return sizeof(*ip6h);
292}
293
294
295static void setup_sockaddr(int domain, const char *str_addr,
296 struct sockaddr_storage *sockaddr)
297{
298 struct sockaddr_in6 *addr6 = (void *) sockaddr;
299 struct sockaddr_in *addr4 = (void *) sockaddr;
300
301 switch (domain) {
302 case PF_INET:
303 memset(addr4, 0, sizeof(*addr4));
304 addr4->sin_family = AF_INET;
305 addr4->sin_port = htons(cfg_port);
306 if (str_addr &&
307 inet_pton(AF_INET, str_addr, &(addr4->sin_addr)) != 1)
308 error(1, 0, "ipv4 parse error: %s", str_addr);
309 break;
310 case PF_INET6:
311 memset(addr6, 0, sizeof(*addr6));
312 addr6->sin6_family = AF_INET6;
313 addr6->sin6_port = htons(cfg_port);
314 if (str_addr &&
315 inet_pton(AF_INET6, str_addr, &(addr6->sin6_addr)) != 1)
316 error(1, 0, "ipv6 parse error: %s", str_addr);
317 break;
318 default:
319 error(1, 0, "illegal domain");
320 }
321}
322
323static int do_setup_tx(int domain, int type, int protocol)
324{
325 int fd;
326
327 fd = socket(domain, type, protocol);
328 if (fd == -1)
329 error(1, errno, "socket t");
330
331 do_setsockopt(fd, SOL_SOCKET, SO_SNDBUF, 1 << 21);
332 if (cfg_zerocopy)
333 do_setsockopt(fd, SOL_SOCKET, SO_ZEROCOPY, 1);
334
335 if (domain != PF_PACKET && domain != PF_RDS)
336 if (connect(fd, (void *) &cfg_dst_addr, cfg_alen))
337 error(1, errno, "connect");
338
339 if (domain == PF_RDS) {
340 if (bind(fd, (void *) &cfg_src_addr, cfg_alen))
341 error(1, errno, "bind");
342 }
343
344 return fd;
345}
346
347static uint32_t do_process_zerocopy_cookies(struct rds_zcopy_cookies *ck)
348{
349 int i;
350
351 if (ck->num > RDS_MAX_ZCOOKIES)
352 error(1, 0, "Returned %d cookies, max expected %d\n",
353 ck->num, RDS_MAX_ZCOOKIES);
354 for (i = 0; i < ck->num; i++)
355 if (cfg_verbose >= 2)
356 fprintf(stderr, "%d\n", ck->cookies[i]);
357 return ck->num;
358}
359
360static bool do_recvmsg_completion(int fd)
361{
362 char cmsgbuf[CMSG_SPACE(sizeof(struct rds_zcopy_cookies))];
363 struct rds_zcopy_cookies *ck;
364 struct cmsghdr *cmsg;
365 struct msghdr msg;
366 bool ret = false;
367
368 memset(&msg, 0, sizeof(msg));
369 msg.msg_control = cmsgbuf;
370 msg.msg_controllen = sizeof(cmsgbuf);
371
372 if (recvmsg(fd, &msg, MSG_DONTWAIT))
373 return ret;
374
375 if (msg.msg_flags & MSG_CTRUNC)
376 error(1, errno, "recvmsg notification: truncated");
377
378 for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) {
379 if (cmsg->cmsg_level == SOL_RDS &&
380 cmsg->cmsg_type == RDS_CMSG_ZCOPY_COMPLETION) {
381
382 ck = (struct rds_zcopy_cookies *)CMSG_DATA(cmsg);
383 completions += do_process_zerocopy_cookies(ck);
384 ret = true;
385 break;
386 }
387 error(0, 0, "ignoring cmsg at level %d type %d\n",
388 cmsg->cmsg_level, cmsg->cmsg_type);
389 }
390 return ret;
391}
392
393static bool do_recv_completion(int fd, int domain)
394{
395 struct sock_extended_err *serr;
396 struct msghdr msg = {};
397 struct cmsghdr *cm;
398 uint32_t hi, lo, range;
399 int ret, zerocopy;
400 char control[100];
401
402 if (domain == PF_RDS)
403 return do_recvmsg_completion(fd);
404
405 msg.msg_control = control;
406 msg.msg_controllen = sizeof(control);
407
408 ret = recvmsg(fd, &msg, MSG_ERRQUEUE);
409 if (ret == -1 && errno == EAGAIN)
410 return false;
411 if (ret == -1)
412 error(1, errno, "recvmsg notification");
413 if (msg.msg_flags & MSG_CTRUNC)
414 error(1, errno, "recvmsg notification: truncated");
415
416 cm = CMSG_FIRSTHDR(&msg);
417 if (!cm)
418 error(1, 0, "cmsg: no cmsg");
419 if (!((cm->cmsg_level == SOL_IP && cm->cmsg_type == IP_RECVERR) ||
420 (cm->cmsg_level == SOL_IPV6 && cm->cmsg_type == IPV6_RECVERR) ||
421 (cm->cmsg_level == SOL_PACKET && cm->cmsg_type == PACKET_TX_TIMESTAMP)))
422 error(1, 0, "serr: wrong type: %d.%d",
423 cm->cmsg_level, cm->cmsg_type);
424
425 serr = (void *) CMSG_DATA(cm);
426
427 if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY)
428 error(1, 0, "serr: wrong origin: %u", serr->ee_origin);
429 if (serr->ee_errno != 0)
430 error(1, 0, "serr: wrong error code: %u", serr->ee_errno);
431
432 hi = serr->ee_data;
433 lo = serr->ee_info;
434 range = hi - lo + 1;
435
436 /* Detect notification gaps. These should not happen often, if at all.
437 * Gaps can occur due to drops, reordering and retransmissions.
438 */
439 if (lo != next_completion)
440 fprintf(stderr, "gap: %u..%u does not append to %u\n",
441 lo, hi, next_completion);
442 next_completion = hi + 1;
443
444 zerocopy = !(serr->ee_code & SO_EE_CODE_ZEROCOPY_COPIED);
445 if (zerocopied == -1)
446 zerocopied = zerocopy;
447 else if (zerocopied != zerocopy) {
448 fprintf(stderr, "serr: inconsistent\n");
449 zerocopied = zerocopy;
450 }
451
452 if (cfg_verbose >= 2)
453 fprintf(stderr, "completed: %u (h=%u l=%u)\n",
454 range, hi, lo);
455
456 completions += range;
457 return true;
458}
459
460/* Read all outstanding messages on the errqueue */
461static void do_recv_completions(int fd, int domain)
462{
463 while (do_recv_completion(fd, domain)) {}
464}
465
466/* Wait for all remaining completions on the errqueue */
467static void do_recv_remaining_completions(int fd, int domain)
468{
469 int64_t tstop = gettimeofday_ms() + cfg_waittime_ms;
470
471 while (completions < expected_completions &&
472 gettimeofday_ms() < tstop) {
473 if (do_poll(fd, domain == PF_RDS ? POLLIN : POLLERR))
474 do_recv_completions(fd, domain);
475 }
476
477 if (completions < expected_completions)
478 fprintf(stderr, "missing notifications: %lu < %lu\n",
479 completions, expected_completions);
480}
481
482static void do_tx(int domain, int type, int protocol)
483{
484 struct iovec iov[3] = { {0} };
485 struct sockaddr_ll laddr;
486 struct msghdr msg = {0};
487 struct ethhdr eth;
488 union {
489 struct ipv6hdr ip6h;
490 struct iphdr iph;
491 } nh;
492 uint64_t tstop;
493 int fd;
494
495 fd = do_setup_tx(domain, type, protocol);
496
497 if (domain == PF_PACKET) {
498 uint16_t proto = cfg_family == PF_INET ? ETH_P_IP : ETH_P_IPV6;
499
500 /* sock_raw passes ll header as data */
501 if (type == SOCK_RAW) {
502 memset(eth.h_dest, 0x06, ETH_ALEN);
503 memset(eth.h_source, 0x02, ETH_ALEN);
504 eth.h_proto = htons(proto);
505 iov[0].iov_base = ð
506 iov[0].iov_len = sizeof(eth);
507 msg.msg_iovlen++;
508 }
509
510 /* both sock_raw and sock_dgram expect name */
511 memset(&laddr, 0, sizeof(laddr));
512 laddr.sll_family = AF_PACKET;
513 laddr.sll_ifindex = cfg_ifindex;
514 laddr.sll_protocol = htons(proto);
515 laddr.sll_halen = ETH_ALEN;
516
517 memset(laddr.sll_addr, 0x06, ETH_ALEN);
518
519 msg.msg_name = &laddr;
520 msg.msg_namelen = sizeof(laddr);
521 }
522
523 /* packet and raw sockets with hdrincl must pass network header */
524 if (domain == PF_PACKET || protocol == IPPROTO_RAW) {
525 if (cfg_family == PF_INET)
526 iov[1].iov_len = setup_iph(&nh.iph, cfg_payload_len);
527 else
528 iov[1].iov_len = setup_ip6h(&nh.ip6h, cfg_payload_len);
529
530 iov[1].iov_base = (void *) &nh;
531 msg.msg_iovlen++;
532 }
533
534 if (domain == PF_RDS) {
535 msg.msg_name = &cfg_dst_addr;
536 msg.msg_namelen = (cfg_dst_addr.ss_family == AF_INET ?
537 sizeof(struct sockaddr_in) :
538 sizeof(struct sockaddr_in6));
539 }
540
541 iov[2].iov_base = payload;
542 iov[2].iov_len = cfg_payload_len;
543 msg.msg_iovlen++;
544 msg.msg_iov = &iov[3 - msg.msg_iovlen];
545
546 tstop = gettimeofday_ms() + cfg_runtime_ms;
547 do {
548 if (cfg_cork)
549 do_sendmsg_corked(fd, &msg);
550 else
551 do_sendmsg(fd, &msg, cfg_zerocopy, domain);
552
553 while (!do_poll(fd, POLLOUT)) {
554 if (cfg_zerocopy)
555 do_recv_completions(fd, domain);
556 }
557
558 } while (gettimeofday_ms() < tstop);
559
560 if (cfg_zerocopy)
561 do_recv_remaining_completions(fd, domain);
562
563 if (close(fd))
564 error(1, errno, "close");
565
566 fprintf(stderr, "tx=%lu (%lu MB) txc=%lu zc=%c\n",
567 packets, bytes >> 20, completions,
568 zerocopied == 1 ? 'y' : 'n');
569}
570
571static int do_setup_rx(int domain, int type, int protocol)
572{
573 int fd;
574
575 /* If tx over PF_PACKET, rx over PF_INET(6)/SOCK_RAW,
576 * to recv the only copy of the packet, not a clone
577 */
578 if (domain == PF_PACKET)
579 error(1, 0, "Use PF_INET/SOCK_RAW to read");
580
581 if (type == SOCK_RAW && protocol == IPPROTO_RAW)
582 error(1, 0, "IPPROTO_RAW: not supported on Rx");
583
584 fd = socket(domain, type, protocol);
585 if (fd == -1)
586 error(1, errno, "socket r");
587
588 do_setsockopt(fd, SOL_SOCKET, SO_RCVBUF, 1 << 21);
589 do_setsockopt(fd, SOL_SOCKET, SO_RCVLOWAT, 1 << 16);
590 do_setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, 1);
591
592 if (bind(fd, (void *) &cfg_dst_addr, cfg_alen))
593 error(1, errno, "bind");
594
595 if (type == SOCK_STREAM) {
596 if (listen(fd, 1))
597 error(1, errno, "listen");
598 fd = do_accept(fd);
599 }
600
601 return fd;
602}
603
604/* Flush all outstanding bytes for the tcp receive queue */
605static void do_flush_tcp(int fd)
606{
607 int ret;
608
609 /* MSG_TRUNC flushes up to len bytes */
610 ret = recv(fd, NULL, 1 << 21, MSG_TRUNC | MSG_DONTWAIT);
611 if (ret == -1 && errno == EAGAIN)
612 return;
613 if (ret == -1)
614 error(1, errno, "flush");
615 if (!ret)
616 return;
617
618 packets++;
619 bytes += ret;
620}
621
622/* Flush all outstanding datagrams. Verify first few bytes of each. */
623static void do_flush_datagram(int fd, int type)
624{
625 int ret, off = 0;
626 char buf[64];
627
628 /* MSG_TRUNC will return full datagram length */
629 ret = recv(fd, buf, sizeof(buf), MSG_DONTWAIT | MSG_TRUNC);
630 if (ret == -1 && errno == EAGAIN)
631 return;
632
633 /* raw ipv4 return with header, raw ipv6 without */
634 if (cfg_family == PF_INET && type == SOCK_RAW) {
635 off += sizeof(struct iphdr);
636 ret -= sizeof(struct iphdr);
637 }
638
639 if (ret == -1)
640 error(1, errno, "recv");
641 if (ret != cfg_payload_len)
642 error(1, 0, "recv: ret=%u != %u", ret, cfg_payload_len);
643 if (ret > sizeof(buf) - off)
644 ret = sizeof(buf) - off;
645 if (memcmp(buf + off, payload, ret))
646 error(1, 0, "recv: data mismatch");
647
648 packets++;
649 bytes += cfg_payload_len;
650}
651
652static void do_rx(int domain, int type, int protocol)
653{
654 const int cfg_receiver_wait_ms = 400;
655 uint64_t tstop;
656 int fd;
657
658 fd = do_setup_rx(domain, type, protocol);
659
660 tstop = gettimeofday_ms() + cfg_runtime_ms + cfg_receiver_wait_ms;
661 do {
662 if (type == SOCK_STREAM)
663 do_flush_tcp(fd);
664 else
665 do_flush_datagram(fd, type);
666
667 do_poll(fd, POLLIN);
668
669 } while (gettimeofday_ms() < tstop);
670
671 if (close(fd))
672 error(1, errno, "close");
673
674 fprintf(stderr, "rx=%lu (%lu MB)\n", packets, bytes >> 20);
675}
676
677static void do_test(int domain, int type, int protocol)
678{
679 int i;
680
681 if (cfg_cork && (domain == PF_PACKET || type != SOCK_DGRAM))
682 error(1, 0, "can only cork udp sockets");
683
684 do_setcpu(cfg_cpu);
685
686 for (i = 0; i < IP_MAXPACKET; i++)
687 payload[i] = 'a' + (i % 26);
688
689 if (cfg_rx)
690 do_rx(domain, type, protocol);
691 else
692 do_tx(domain, type, protocol);
693}
694
695static void usage(const char *filepath)
696{
697 error(1, 0, "Usage: %s [options] <test>", filepath);
698}
699
700static void parse_opts(int argc, char **argv)
701{
702 const int max_payload_len = sizeof(payload) -
703 sizeof(struct ipv6hdr) -
704 sizeof(struct tcphdr) -
705 40 /* max tcp options */;
706 int c;
707 char *daddr = NULL, *saddr = NULL;
708 char *cfg_test;
709
710 cfg_payload_len = max_payload_len;
711
712 while ((c = getopt(argc, argv, "46c:C:D:i:mp:rs:S:t:vz")) != -1) {
713 switch (c) {
714 case '4':
715 if (cfg_family != PF_UNSPEC)
716 error(1, 0, "Pass one of -4 or -6");
717 cfg_family = PF_INET;
718 cfg_alen = sizeof(struct sockaddr_in);
719 break;
720 case '6':
721 if (cfg_family != PF_UNSPEC)
722 error(1, 0, "Pass one of -4 or -6");
723 cfg_family = PF_INET6;
724 cfg_alen = sizeof(struct sockaddr_in6);
725 break;
726 case 'c':
727 cfg_cork = strtol(optarg, NULL, 0);
728 break;
729 case 'C':
730 cfg_cpu = strtol(optarg, NULL, 0);
731 break;
732 case 'D':
733 daddr = optarg;
734 break;
735 case 'i':
736 cfg_ifindex = if_nametoindex(optarg);
737 if (cfg_ifindex == 0)
738 error(1, errno, "invalid iface: %s", optarg);
739 break;
740 case 'm':
741 cfg_cork_mixed = true;
742 break;
743 case 'p':
744 cfg_port = strtoul(optarg, NULL, 0);
745 break;
746 case 'r':
747 cfg_rx = true;
748 break;
749 case 's':
750 cfg_payload_len = strtoul(optarg, NULL, 0);
751 break;
752 case 'S':
753 saddr = optarg;
754 break;
755 case 't':
756 cfg_runtime_ms = 200 + strtoul(optarg, NULL, 10) * 1000;
757 break;
758 case 'v':
759 cfg_verbose++;
760 break;
761 case 'z':
762 cfg_zerocopy = true;
763 break;
764 }
765 }
766
767 cfg_test = argv[argc - 1];
768 if (strcmp(cfg_test, "rds") == 0) {
769 if (!daddr)
770 error(1, 0, "-D <server addr> required for PF_RDS\n");
771 if (!cfg_rx && !saddr)
772 error(1, 0, "-S <client addr> required for PF_RDS\n");
773 }
774 setup_sockaddr(cfg_family, daddr, &cfg_dst_addr);
775 setup_sockaddr(cfg_family, saddr, &cfg_src_addr);
776
777 if (cfg_payload_len > max_payload_len)
778 error(1, 0, "-s: payload exceeds max (%d)", max_payload_len);
779 if (cfg_cork_mixed && (!cfg_zerocopy || !cfg_cork))
780 error(1, 0, "-m: cork_mixed requires corking and zerocopy");
781
782 if (optind != argc - 1)
783 usage(argv[0]);
784}
785
786int main(int argc, char **argv)
787{
788 const char *cfg_test;
789
790 parse_opts(argc, argv);
791
792 cfg_test = argv[argc - 1];
793
794 if (!strcmp(cfg_test, "packet"))
795 do_test(PF_PACKET, SOCK_RAW, 0);
796 else if (!strcmp(cfg_test, "packet_dgram"))
797 do_test(PF_PACKET, SOCK_DGRAM, 0);
798 else if (!strcmp(cfg_test, "raw"))
799 do_test(cfg_family, SOCK_RAW, IPPROTO_EGP);
800 else if (!strcmp(cfg_test, "raw_hdrincl"))
801 do_test(cfg_family, SOCK_RAW, IPPROTO_RAW);
802 else if (!strcmp(cfg_test, "tcp"))
803 do_test(cfg_family, SOCK_STREAM, 0);
804 else if (!strcmp(cfg_test, "udp"))
805 do_test(cfg_family, SOCK_DGRAM, 0);
806 else if (!strcmp(cfg_test, "rds"))
807 do_test(PF_RDS, SOCK_SEQPACKET, 0);
808 else
809 error(1, 0, "unknown cfg_test %s", cfg_test);
810
811 return 0;
812}