/* Evaluate MSG_ZEROCOPY * * Send traffic between two processes over one of the supported * protocols and modes: * * PF_INET/PF_INET6 * - SOCK_STREAM * - SOCK_DGRAM * - SOCK_DGRAM with UDP_CORK * - SOCK_RAW * - SOCK_RAW with IP_HDRINCL * * PF_PACKET * - SOCK_DGRAM * - SOCK_RAW * * PF_RDS * - SOCK_SEQPACKET * * Start this program on two connected hosts, one in send mode and * the other with option '-r' to put it in receiver mode. * * If zerocopy mode ('-z') is enabled, the sender will verify that * the kernel queues completions on the error queue for all zerocopy * transfers. */ #define _GNU_SOURCE #include <arpa/inet.h>#include <error.h>#include <errno.h>#include <limits.h>#include <linux/errqueue.h>#include <linux/if_packet.h>#include <linux/ipv6.h>#include <linux/socket.h>#include <linux/sockios.h>#include <net/ethernet.h>#include <net/if.h>#include <netinet/ip.h>#include <netinet/ip6.h>#include <netinet/tcp.h>#include <netinet/udp.h>#include <poll.h>#include <sched.h>#include <stdbool.h>#include <stdio.h>#include <stdint.h>#include <stdlib.h>#include <string.h>#include <sys/ioctl.h>#include <sys/socket.h>#include <sys/stat.h>#include <sys/time.h>#include <sys/types.h>#include <sys/wait.h>#include <unistd.h>#include <linux/rds.h> #ifndef SO_EE_ORIGIN_ZEROCOPY#define SO_EE_ORIGIN_ZEROCOPY 5#endif #ifndef SO_ZEROCOPY#define SO_ZEROCOPY 60#endif #ifndef SO_EE_CODE_ZEROCOPY_COPIED#define SO_EE_CODE_ZEROCOPY_COPIED 1#endif #ifndef MSG_ZEROCOPY#define MSG_ZEROCOPY 0x4000000#endif static int cfg_cork;static bool cfg_cork_mixed;static int cfg_cpu = -1; /* default: pin to last cpu */static int cfg_expect_zerocopy = -1;static int cfg_family = PF_UNSPEC;static int cfg_ifindex = 1;static int cfg_payload_len;static int cfg_port = 8000;static bool cfg_rx;static int cfg_runtime_ms = 4200;static int cfg_verbose;static int cfg_waittime_ms = 500;static int cfg_notification_limit = 32;static bool cfg_zerocopy; static socklen_t cfg_alen;static struct sockaddr_storage cfg_dst_addr;static struct sockaddr_storage cfg_src_addr; static int exitcode;static char payload[IP_MAXPACKET];static long packets, bytes, completions, expected_completions;static uint32_t next_completion;static uint32_t sends_since_notify; static unsigned long gettimeofday_ms(void){ struct timeval tv; gettimeofday(&tv, NULL); return (tv.tv_sec * 1000) + (tv.tv_usec / 1000);} static uint16_t get_ip_csum(const uint16_t *start, int num_words){ unsigned long sum = 0; int i; for (i = 0; i < num_words; i++) sum += start[i]; while (sum >> 16) sum = (sum & 0xFFFF) + (sum >> 16); return ~sum;} static int do_setcpu(int cpu){ cpu_set_t mask; CPU_ZERO(&mask); CPU_SET(cpu, &mask); if (sched_setaffinity(0, sizeof(mask), &mask)) fprintf(stderr, "cpu: unable to pin, may increase variance.\n"); else if (cfg_verbose) fprintf(stderr, "cpu: %u\n", cpu); return 0;} static void do_setsockopt(int fd, int level, int optname, int val){ if (setsockopt(fd, level, optname, &val, sizeof(val))) error(1, errno, "setsockopt %d.%d: %d", level, optname, val);} static int do_poll(int fd, int events){ struct pollfd pfd; int ret; pfd.events = events; pfd.revents = 0; pfd.fd = fd; ret = poll(&pfd, 1, cfg_waittime_ms); if (ret == -1) error(1, errno, "poll"); return ret && (pfd.revents & events);} static int do_accept(int fd){ int fda = fd; fd = accept(fda, NULL, NULL); if (fd == -1) error(1, errno, "accept"); if (close(fda)) error(1, errno, "close listen sock"); return fd;} static void add_zcopy_cookie(struct msghdr *msg, uint32_t cookie){ struct cmsghdr *cm; if (!msg->msg_control) error(1, errno, "NULL cookie"); cm = (void *)msg->msg_control; cm->cmsg_len = CMSG_LEN(sizeof(cookie)); cm->cmsg_level = SOL_RDS; cm->cmsg_type = RDS_CMSG_ZCOPY_COOKIE; memcpy(CMSG_DATA(cm), &cookie, sizeof(cookie));} static bool do_sendmsg(int fd, struct msghdr *msg, bool do_zerocopy, int domain){ int ret, len, i, flags; static uint32_t cookie; char ckbuf[CMSG_SPACE(sizeof(cookie))]; len = 0; for (i = 0; i < msg->msg_iovlen; i++) len += msg->msg_iov[i].iov_len; flags = MSG_DONTWAIT; if (do_zerocopy) { flags |= MSG_ZEROCOPY; if (domain == PF_RDS) { memset(&msg->msg_control, 0, sizeof(msg->msg_control)); msg->msg_controllen = CMSG_SPACE(sizeof(cookie)); msg->msg_control = (struct cmsghdr *)ckbuf; add_zcopy_cookie(msg, ++cookie); } } ret = sendmsg(fd, msg, flags); if (ret == -1 && errno == EAGAIN) return false; if (ret == -1) error(1, errno, "send"); if (cfg_verbose && ret != len) fprintf(stderr, "send: ret=%u != %u\n", ret, len); sends_since_notify++; if (len) { packets++; bytes += ret; if (do_zerocopy && ret) expected_completions++; } if (do_zerocopy && domain == PF_RDS) { msg->msg_control = NULL; msg->msg_controllen = 0; } return true;} static void do_sendmsg_corked(int fd, struct msghdr *msg){ bool do_zerocopy = cfg_zerocopy; int i, payload_len, extra_len; /* split up the packet. for non-multiple, make first buffer longer */ payload_len = cfg_payload_len / cfg_cork; extra_len = cfg_payload_len - (cfg_cork * payload_len); do_setsockopt(fd, IPPROTO_UDP, UDP_CORK, 1); for (i = 0; i < cfg_cork; i++) { /* in mixed-frags mode, alternate zerocopy and copy frags * start with non-zerocopy, to ensure attach later works */ if (cfg_cork_mixed) do_zerocopy = (i & 1); msg->msg_iov[0].iov_len = payload_len + extra_len; extra_len = 0; do_sendmsg(fd, msg, do_zerocopy, (cfg_dst_addr.ss_family == AF_INET ? PF_INET : PF_INET6)); } do_setsockopt(fd, IPPROTO_UDP, UDP_CORK, 0);} static int setup_iph(struct iphdr *iph, uint16_t payload_len){ struct sockaddr_in *daddr = (void *) &cfg_dst_addr; struct sockaddr_in *saddr = (void *) &cfg_src_addr; memset(iph, 0, sizeof(*iph)); iph->version = 4; iph->tos = 0; iph->ihl = 5; iph->ttl = 2; iph->saddr = saddr->sin_addr.s_addr; iph->daddr = daddr->sin_addr.s_addr; iph->protocol = IPPROTO_EGP; iph->tot_len = htons(sizeof(*iph) + payload_len); iph->check = get_ip_csum((void *) iph, iph->ihl << 1); return sizeof(*iph);} static int setup_ip6h(struct ipv6hdr *ip6h, uint16_t payload_len){ struct sockaddr_in6 *daddr = (void *) &cfg_dst_addr; struct sockaddr_in6 *saddr = (void *) &cfg_src_addr; memset(ip6h, 0, sizeof(*ip6h)); ip6h->version = 6; ip6h->payload_len = htons(payload_len); ip6h->nexthdr = IPPROTO_EGP; ip6h->hop_limit = 2; ip6h->saddr = saddr->sin6_addr; ip6h->daddr = daddr->sin6_addr; return sizeof(*ip6h);} static void setup_sockaddr(int domain, const char *str_addr, struct sockaddr_storage *sockaddr){ struct sockaddr_in6 *addr6 = (void *) sockaddr; struct sockaddr_in *addr4 = (void *) sockaddr; switch (domain) { case PF_INET: memset(addr4, 0, sizeof(*addr4)); addr4->sin_family = AF_INET; addr4->sin_port = htons(cfg_port); if (str_addr && inet_pton(AF_INET, str_addr, &(addr4->sin_addr)) != 1) error(1, 0, "ipv4 parse error: %s", str_addr); break; case PF_INET6: memset(addr6, 0, sizeof(*addr6)); addr6->sin6_family = AF_INET6; addr6->sin6_port = htons(cfg_port); if (str_addr && inet_pton(AF_INET6, str_addr, &(addr6->sin6_addr)) != 1) error(1, 0, "ipv6 parse error: %s", str_addr); break; default: error(1, 0, "illegal domain"); }} static int do_setup_tx(int domain, int type, int protocol){ int fd; fd = socket(domain, type, protocol); if (fd == -1) error(1, errno, "socket t"); do_setsockopt(fd, SOL_SOCKET, SO_SNDBUF, 1 << 21); if (cfg_zerocopy) do_setsockopt(fd, SOL_SOCKET, SO_ZEROCOPY, 1); if (domain != PF_PACKET && domain != PF_RDS) if (connect(fd, (void *) &cfg_dst_addr, cfg_alen)) error(1, errno, "connect"); if (domain == PF_RDS) { if (bind(fd, (void *) &cfg_src_addr, cfg_alen)) error(1, errno, "bind"); } return fd;} static uint32_t do_process_zerocopy_cookies(struct rds_zcopy_cookies *ck){ int i; if (ck->num > RDS_MAX_ZCOOKIES) error(1, 0, "Returned %d cookies, max expected %d\n", ck->num, RDS_MAX_ZCOOKIES); for (i = 0; i < ck->num; i++) if (cfg_verbose >= 2) fprintf(stderr, "%d\n", ck->cookies[i]); return ck->num;} static bool do_recvmsg_completion(int fd){ char cmsgbuf[CMSG_SPACE(sizeof(struct rds_zcopy_cookies))]; struct rds_zcopy_cookies *ck; struct cmsghdr *cmsg; struct msghdr msg; bool ret = false; memset(&msg, 0, sizeof(msg)); msg.msg_control = cmsgbuf; msg.msg_controllen = sizeof(cmsgbuf); if (recvmsg(fd, &msg, MSG_DONTWAIT)) return ret; if (msg.msg_flags & MSG_CTRUNC) error(1, errno, "recvmsg notification: truncated"); for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) { if (cmsg->cmsg_level == SOL_RDS && cmsg->cmsg_type == RDS_CMSG_ZCOPY_COMPLETION) { ck = (struct rds_zcopy_cookies *)CMSG_DATA(cmsg); completions += do_process_zerocopy_cookies(ck); ret = true; break; } error(0, 0, "ignoring cmsg at level %d type %d\n", cmsg->cmsg_level, cmsg->cmsg_type); } return ret;} static bool do_recv_completion(int fd, int domain){ struct sock_extended_err *serr; struct msghdr msg = {}; struct cmsghdr *cm; uint32_t hi, lo, range; int ret, zerocopy; char control[100]; if (domain == PF_RDS) return do_recvmsg_completion(fd); msg.msg_control = control; msg.msg_controllen = sizeof(control); ret = recvmsg(fd, &msg, MSG_ERRQUEUE); if (ret == -1 && errno == EAGAIN) return false; if (ret == -1) error(1, errno, "recvmsg notification"); if (msg.msg_flags & MSG_CTRUNC) error(1, errno, "recvmsg notification: truncated"); cm = CMSG_FIRSTHDR(&msg); if (!cm) error(1, 0, "cmsg: no cmsg"); if (!((cm->cmsg_level == SOL_IP && cm->cmsg_type == IP_RECVERR) || (cm->cmsg_level == SOL_IPV6 && cm->cmsg_type == IPV6_RECVERR) || (cm->cmsg_level == SOL_PACKET && cm->cmsg_type == PACKET_TX_TIMESTAMP))) error(1, 0, "serr: wrong type: %d.%d", cm->cmsg_level, cm->cmsg_type); serr = (void *) CMSG_DATA(cm); if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY) error(1, 0, "serr: wrong origin: %u", serr->ee_origin); if (serr->ee_errno != 0) error(1, 0, "serr: wrong error code: %u", serr->ee_errno); hi = serr->ee_data; lo = serr->ee_info; range = hi - lo + 1; /* Detect notification gaps. These should not happen often, if at all. * Gaps can occur due to drops, reordering and retransmissions. */ if (cfg_verbose && lo != next_completion) fprintf(stderr, "gap: %u..%u does not append to %u\n", lo, hi, next_completion); next_completion = hi + 1; zerocopy = !(serr->ee_code & SO_EE_CODE_ZEROCOPY_COPIED); if (cfg_expect_zerocopy != -1 && cfg_expect_zerocopy != zerocopy) { fprintf(stderr, "serr: ee_code: %u != expected %u\n", zerocopy, cfg_expect_zerocopy); exitcode = 1; /* suppress repeated messages */ cfg_expect_zerocopy = zerocopy; } if (cfg_verbose >= 2) fprintf(stderr, "completed: %u (h=%u l=%u)\n", range, hi, lo); completions += range; return true;} /* Read all outstanding messages on the errqueue */static void do_recv_completions(int fd, int domain){ while (do_recv_completion(fd, domain)) {} sends_since_notify = 0;} /* Wait for all remaining completions on the errqueue */static void do_recv_remaining_completions(int fd, int domain){ int64_t tstop = gettimeofday_ms() + cfg_waittime_ms; while (completions < expected_completions && gettimeofday_ms() < tstop) { if (do_poll(fd, domain == PF_RDS ? POLLIN : POLLERR)) do_recv_completions(fd, domain); } if (completions < expected_completions) fprintf(stderr, "missing notifications: %lu < %lu\n", completions, expected_completions);} static void do_tx(int domain, int type, int protocol){ struct iovec iov[3] = { {0} }; struct sockaddr_ll laddr; struct msghdr msg = {0}; struct ethhdr eth; union { struct ipv6hdr ip6h; struct iphdr iph; } nh; uint64_t tstop; int fd; Lines 1–500 of 827. Load more
/* Evaluate MSG_ZEROCOPY * * Send traffic between two processes over one of the supported * protocols and modes: * * PF_INET/PF_INET6 * - SOCK_STREAM * - SOCK_DGRAM * - SOCK_DGRAM with UDP_CORK * - SOCK_RAW * - SOCK_RAW with IP_HDRINCL * * PF_PACKET * - SOCK_DGRAM * - SOCK_RAW * * PF_RDS * - SOCK_SEQPACKET * * Start this program on two connected hosts, one in send mode and * the other with option '-r' to put it in receiver mode. * * If zerocopy mode ('-z') is enabled, the sender will verify that * the kernel queues completions on the error queue for all zerocopy * transfers. */ #define _GNU_SOURCE #include <arpa/inet.h> #include <error.h> #include <errno.h> #include <limits.h> #include <linux/errqueue.h> #include <linux/if_packet.h> #include <linux/ipv6.h> #include <linux/socket.h> #include <linux/sockios.h> #include <net/ethernet.h> #include <net/if.h> #include <netinet/ip.h> #include <netinet/ip6.h> #include <netinet/tcp.h> #include <netinet/udp.h> #include <poll.h> #include <sched.h> #include <stdbool.h> #include <stdio.h> #include <stdint.h> #include <stdlib.h> #include <string.h> #include <sys/ioctl.h> #include <sys/socket.h> #include <sys/stat.h> #include <sys/time.h> #include <sys/types.h> #include <sys/wait.h> #include <unistd.h> #include <linux/rds.h> #ifndef SO_EE_ORIGIN_ZEROCOPY #define SO_EE_ORIGIN_ZEROCOPY 5 #endif #ifndef SO_ZEROCOPY #define SO_ZEROCOPY 60 #endif #ifndef SO_EE_CODE_ZEROCOPY_COPIED #define SO_EE_CODE_ZEROCOPY_COPIED 1 #endif #ifndef MSG_ZEROCOPY #define MSG_ZEROCOPY 0x4000000 #endif static int cfg_cork; static bool cfg_cork_mixed; static int cfg_cpu = -1; /* default: pin to last cpu */ static int cfg_expect_zerocopy = -1; static int cfg_family = PF_UNSPEC; static int cfg_ifindex = 1; static int cfg_payload_len; static int cfg_port = 8000; static bool cfg_rx; static int cfg_runtime_ms = 4200; static int cfg_verbose; static int cfg_waittime_ms = 500; static int cfg_notification_limit = 32; static bool cfg_zerocopy; static socklen_t cfg_alen; static struct sockaddr_storage cfg_dst_addr; static struct sockaddr_storage cfg_src_addr; static int exitcode; static char payload[IP_MAXPACKET]; static long packets, bytes, completions, expected_completions; static uint32_t next_completion; static uint32_t sends_since_notify; static unsigned long gettimeofday_ms(void) { struct timeval tv; gettimeofday(&tv, NULL); return (tv.tv_sec * 1000) + (tv.tv_usec / 1000); } static uint16_t get_ip_csum(const uint16_t *start, int num_words) { unsigned long sum = 0; int i; for (i = 0; i < num_words; i++) sum += start[i]; while (sum >> 16) sum = (sum & 0xFFFF) + (sum >> 16); return ~sum; } static int do_setcpu(int cpu) { cpu_set_t mask; CPU_ZERO(&mask); CPU_SET(cpu, &mask); if (sched_setaffinity(0, sizeof(mask), &mask)) fprintf(stderr, "cpu: unable to pin, may increase variance.\n"); else if (cfg_verbose) fprintf(stderr, "cpu: %u\n", cpu); return 0; } static void do_setsockopt(int fd, int level, int optname, int val) { if (setsockopt(fd, level, optname, &val, sizeof(val))) error(1, errno, "setsockopt %d.%d: %d", level, optname, val); } static int do_poll(int fd, int events) { struct pollfd pfd; int ret; pfd.events = events; pfd.revents = 0; pfd.fd = fd; ret = poll(&pfd, 1, cfg_waittime_ms); if (ret == -1) error(1, errno, "poll"); return ret && (pfd.revents & events); } static int do_accept(int fd) { int fda = fd; fd = accept(fda, NULL, NULL); if (fd == -1) error(1, errno, "accept"); if (close(fda)) error(1, errno, "close listen sock"); return fd; } static void add_zcopy_cookie(struct msghdr *msg, uint32_t cookie) { struct cmsghdr *cm; if (!msg->msg_control) error(1, errno, "NULL cookie"); cm = (void *)msg->msg_control; cm->cmsg_len = CMSG_LEN(sizeof(cookie)); cm->cmsg_level = SOL_RDS; cm->cmsg_type = RDS_CMSG_ZCOPY_COOKIE; memcpy(CMSG_DATA(cm), &cookie, sizeof(cookie)); } static bool do_sendmsg(int fd, struct msghdr *msg, bool do_zerocopy, int domain) { int ret, len, i, flags; static uint32_t cookie; char ckbuf[CMSG_SPACE(sizeof(cookie))]; len = 0; for (i = 0; i < msg->msg_iovlen; i++) len += msg->msg_iov[i].iov_len; flags = MSG_DONTWAIT; if (do_zerocopy) { flags |= MSG_ZEROCOPY; if (domain == PF_RDS) { memset(&msg->msg_control, 0, sizeof(msg->msg_control)); msg->msg_controllen = CMSG_SPACE(sizeof(cookie)); msg->msg_control = (struct cmsghdr *)ckbuf; add_zcopy_cookie(msg, ++cookie); } } ret = sendmsg(fd, msg, flags); if (ret == -1 && errno == EAGAIN) return false; if (ret == -1) error(1, errno, "send"); if (cfg_verbose && ret != len) fprintf(stderr, "send: ret=%u != %u\n", ret, len); sends_since_notify++; if (len) { packets++; bytes += ret; if (do_zerocopy && ret) expected_completions++; } if (do_zerocopy && domain == PF_RDS) { msg->msg_control = NULL; msg->msg_controllen = 0; } return true; } static void do_sendmsg_corked(int fd, struct msghdr *msg) { bool do_zerocopy = cfg_zerocopy; int i, payload_len, extra_len; /* split up the packet. for non-multiple, make first buffer longer */ payload_len = cfg_payload_len / cfg_cork; extra_len = cfg_payload_len - (cfg_cork * payload_len); do_setsockopt(fd, IPPROTO_UDP, UDP_CORK, 1); for (i = 0; i < cfg_cork; i++) { /* in mixed-frags mode, alternate zerocopy and copy frags * start with non-zerocopy, to ensure attach later works */ if (cfg_cork_mixed) do_zerocopy = (i & 1); msg->msg_iov[0].iov_len = payload_len + extra_len; extra_len = 0; do_sendmsg(fd, msg, do_zerocopy, (cfg_dst_addr.ss_family == AF_INET ? PF_INET : PF_INET6)); } do_setsockopt(fd, IPPROTO_UDP, UDP_CORK, 0); } static int setup_iph(struct iphdr *iph, uint16_t payload_len) { struct sockaddr_in *daddr = (void *) &cfg_dst_addr; struct sockaddr_in *saddr = (void *) &cfg_src_addr; memset(iph, 0, sizeof(*iph)); iph->version = 4; iph->tos = 0; iph->ihl = 5; iph->ttl = 2; iph->saddr = saddr->sin_addr.s_addr; iph->daddr = daddr->sin_addr.s_addr; iph->protocol = IPPROTO_EGP; iph->tot_len = htons(sizeof(*iph) + payload_len); iph->check = get_ip_csum((void *) iph, iph->ihl << 1); return sizeof(*iph); } static int setup_ip6h(struct ipv6hdr *ip6h, uint16_t payload_len) { struct sockaddr_in6 *daddr = (void *) &cfg_dst_addr; struct sockaddr_in6 *saddr = (void *) &cfg_src_addr; memset(ip6h, 0, sizeof(*ip6h)); ip6h->version = 6; ip6h->payload_len = htons(payload_len); ip6h->nexthdr = IPPROTO_EGP; ip6h->hop_limit = 2; ip6h->saddr = saddr->sin6_addr; ip6h->daddr = daddr->sin6_addr; return sizeof(*ip6h); } static void setup_sockaddr(int domain, const char *str_addr, struct sockaddr_storage *sockaddr) { struct sockaddr_in6 *addr6 = (void *) sockaddr; struct sockaddr_in *addr4 = (void *) sockaddr; switch (domain) { case PF_INET: memset(addr4, 0, sizeof(*addr4)); addr4->sin_family = AF_INET; addr4->sin_port = htons(cfg_port); if (str_addr && inet_pton(AF_INET, str_addr, &(addr4->sin_addr)) != 1) error(1, 0, "ipv4 parse error: %s", str_addr); break; case PF_INET6: memset(addr6, 0, sizeof(*addr6)); addr6->sin6_family = AF_INET6; addr6->sin6_port = htons(cfg_port); if (str_addr && inet_pton(AF_INET6, str_addr, &(addr6->sin6_addr)) != 1) error(1, 0, "ipv6 parse error: %s", str_addr); break; default: error(1, 0, "illegal domain"); } } static int do_setup_tx(int domain, int type, int protocol) { int fd; fd = socket(domain, type, protocol); if (fd == -1) error(1, errno, "socket t"); do_setsockopt(fd, SOL_SOCKET, SO_SNDBUF, 1 << 21); if (cfg_zerocopy) do_setsockopt(fd, SOL_SOCKET, SO_ZEROCOPY, 1); if (domain != PF_PACKET && domain != PF_RDS) if (connect(fd, (void *) &cfg_dst_addr, cfg_alen)) error(1, errno, "connect"); if (domain == PF_RDS) { if (bind(fd, (void *) &cfg_src_addr, cfg_alen)) error(1, errno, "bind"); } return fd; } static uint32_t do_process_zerocopy_cookies(struct rds_zcopy_cookies *ck) { int i; if (ck->num > RDS_MAX_ZCOOKIES) error(1, 0, "Returned %d cookies, max expected %d\n", ck->num, RDS_MAX_ZCOOKIES); for (i = 0; i < ck->num; i++) if (cfg_verbose >= 2) fprintf(stderr, "%d\n", ck->cookies[i]); return ck->num; } static bool do_recvmsg_completion(int fd) { char cmsgbuf[CMSG_SPACE(sizeof(struct rds_zcopy_cookies))]; struct rds_zcopy_cookies *ck; struct cmsghdr *cmsg; struct msghdr msg; bool ret = false; memset(&msg, 0, sizeof(msg)); msg.msg_control = cmsgbuf; msg.msg_controllen = sizeof(cmsgbuf); if (recvmsg(fd, &msg, MSG_DONTWAIT)) return ret; if (msg.msg_flags & MSG_CTRUNC) error(1, errno, "recvmsg notification: truncated"); for (cmsg = CMSG_FIRSTHDR(&msg); cmsg; cmsg = CMSG_NXTHDR(&msg, cmsg)) { if (cmsg->cmsg_level == SOL_RDS && cmsg->cmsg_type == RDS_CMSG_ZCOPY_COMPLETION) { ck = (struct rds_zcopy_cookies *)CMSG_DATA(cmsg); completions += do_process_zerocopy_cookies(ck); ret = true; break; } error(0, 0, "ignoring cmsg at level %d type %d\n", cmsg->cmsg_level, cmsg->cmsg_type); } return ret; } static bool do_recv_completion(int fd, int domain) { struct sock_extended_err *serr; struct msghdr msg = {}; struct cmsghdr *cm; uint32_t hi, lo, range; int ret, zerocopy; char control[100]; if (domain == PF_RDS) return do_recvmsg_completion(fd); msg.msg_control = control; msg.msg_controllen = sizeof(control); ret = recvmsg(fd, &msg, MSG_ERRQUEUE); if (ret == -1 && errno == EAGAIN) return false; if (ret == -1) error(1, errno, "recvmsg notification"); if (msg.msg_flags & MSG_CTRUNC) error(1, errno, "recvmsg notification: truncated"); cm = CMSG_FIRSTHDR(&msg); if (!cm) error(1, 0, "cmsg: no cmsg"); if (!((cm->cmsg_level == SOL_IP && cm->cmsg_type == IP_RECVERR) || (cm->cmsg_level == SOL_IPV6 && cm->cmsg_type == IPV6_RECVERR) || (cm->cmsg_level == SOL_PACKET && cm->cmsg_type == PACKET_TX_TIMESTAMP))) error(1, 0, "serr: wrong type: %d.%d", cm->cmsg_level, cm->cmsg_type); serr = (void *) CMSG_DATA(cm); if (serr->ee_origin != SO_EE_ORIGIN_ZEROCOPY) error(1, 0, "serr: wrong origin: %u", serr->ee_origin); if (serr->ee_errno != 0) error(1, 0, "serr: wrong error code: %u", serr->ee_errno); hi = serr->ee_data; lo = serr->ee_info; range = hi - lo + 1; /* Detect notification gaps. These should not happen often, if at all. * Gaps can occur due to drops, reordering and retransmissions. */ if (cfg_verbose && lo != next_completion) fprintf(stderr, "gap: %u..%u does not append to %u\n", lo, hi, next_completion); next_completion = hi + 1; zerocopy = !(serr->ee_code & SO_EE_CODE_ZEROCOPY_COPIED); if (cfg_expect_zerocopy != -1 && cfg_expect_zerocopy != zerocopy) { fprintf(stderr, "serr: ee_code: %u != expected %u\n", zerocopy, cfg_expect_zerocopy); exitcode = 1; /* suppress repeated messages */ cfg_expect_zerocopy = zerocopy; } if (cfg_verbose >= 2) fprintf(stderr, "completed: %u (h=%u l=%u)\n", range, hi, lo); completions += range; return true; } /* Read all outstanding messages on the errqueue */ static void do_recv_completions(int fd, int domain) { while (do_recv_completion(fd, domain)) {} sends_since_notify = 0; } /* Wait for all remaining completions on the errqueue */ static void do_recv_remaining_completions(int fd, int domain) { int64_t tstop = gettimeofday_ms() + cfg_waittime_ms; while (completions < expected_completions && gettimeofday_ms() < tstop) { if (do_poll(fd, domain == PF_RDS ? POLLIN : POLLERR)) do_recv_completions(fd, domain); } if (completions < expected_completions) fprintf(stderr, "missing notifications: %lu < %lu\n", completions, expected_completions); } static void do_tx(int domain, int type, int protocol) { struct iovec iov[3] = { {0} }; struct sockaddr_ll laddr; struct msghdr msg = {0}; struct ethhdr eth; union { struct ipv6hdr ip6h; struct iphdr iph; } nh; uint64_t tstop; int fd;