diff --git a/configure.ac b/configure.ac index 11f458690..89f95d251 100644 --- a/configure.ac +++ b/configure.ac @@ -253,6 +253,10 @@ AC_CHECK_FUNCS([sendfile]) # connections. AC_CHECK_FUNCS([getline]) +# Check for sendmmsg/recvmmsg support for better throughput. +AC_CHECK_FUNCS([sendmmsg recvmmsg], + AC_DEFINE([HAVE_SEND_RECVMMSG], [1], [Have sendmmsg/recvmmsg functions.])) + # Check for packet pacing socket option (Linux only for now). AC_CACHE_CHECK([SO_MAX_PACING_RATE socket option], [iperf3_cv_header_so_max_pacing_rate], diff --git a/src/iperf.h b/src/iperf.h index e1dd3326c..a79805fe4 100644 --- a/src/iperf.h +++ b/src/iperf.h @@ -91,6 +91,16 @@ struct iperf_sctp_info uint32_t cwnd; }; +#ifndef HAVE_SENDMMSG +struct dummy_mmsghdr { + struct msghdr msg_hdr; /* Message header */ + unsigned int msg_len; /* Number of received bytes for header */ +}; +#endif /* HAVE_SENDMMSG */ + +/* macro to calculate buffer sizes used for a stream (to support hse of send/recvmmsg) */ +#define STREAM_BUFSIZE(prefix) (((prefix)->settings->blksize) * (((prefix)->settings->send_recvmmsg == 0 || (prefix)->settings->burst == 0)? 1 : (prefix)->settings->burst)) + struct iperf_interval_results { atomic_iperf_size_t bytes_transferred; /* bytes transferred in this interval */ @@ -186,6 +196,7 @@ struct iperf_settings int connect_timeout; /* socket connection timeout, in ms */ int idle_timeout; /* server idle time timeout */ unsigned int snd_timeout; /* Timeout for sending tcp messages in active mode, in us */ + int send_recvmmsg; /* whether sendmmsg/recvmmsg should be used - mainly per the -Z option */ struct iperf_time rcv_timeout; /* Timeout for receiving messages in active mode, in us */ int cntl_ka; /* Use Control TCP connection Keepalive */ int cntl_ka_keepidle; /* Control TCP connection Keepalive idle time (TCP_KEEPIDLE) */ @@ -237,6 +248,16 @@ struct iperf_stream int64_t omitted_outoforder_packets; int64_t cnt_error; int64_t omitted_cnt_error; + +#ifdef HAVE_SENDMMSG + struct mmsghdr *msg; /* For some reason, just struct and not a pointer failes on "field ‘msg’ has incomplete type" */ +#else + struct dummy_mmsghdr *msg; +#endif /* HAVE_SENDMMSG */ + struct iovec *pmsg_iov; /* Pointer to the array of iov_msg per mmsg message */ + int sendmmsg_buffered_packets_count; /* number of buffered packets for sendmmsg */ + char *pbuf; /* Pointer to current vailable space in buffer */ + uint64_t target; struct sockaddr_storage local_addr; diff --git a/src/iperf_api.c b/src/iperf_api.c index c645ef5cf..2690ca42f 100644 --- a/src/iperf_api.c +++ b/src/iperf_api.c @@ -43,6 +43,7 @@ #include #include #include +#include #include #include #include @@ -101,6 +102,7 @@ static int JSON_write(int fd, cJSON *json); static void print_interval_results(struct iperf_test *test, struct iperf_stream *sp, cJSON *json_interval_streams); static cJSON *JSON_read(int fd, int max_size); static int JSONStream_Output(struct iperf_test *test, const char* event_name, cJSON* obj); +void iperf_init_send_recvmmsg(struct iperf_test *test); /*************************** Print usage functions ****************************/ @@ -1971,6 +1973,9 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv) 1 : round(test->settings->bitrate_limit_interval/test->stats_interval) ); } + /* Set whether send/recvmmsg is supported along with related settings */ + iperf_init_send_recvmmsg(test); + /* Show warning if JSON output is used with explicit report format */ if ((test->json_output) && (test->settings->unit_format != 'a')) { warning("Report format (-f) flag ignored with JSON output (-J)"); @@ -2335,6 +2340,9 @@ iperf_exchange_parameters(struct iperf_test *test) } return -1; } + + /* Set whether send/recvmmsg is supported along with related settings */ + iperf_init_send_recvmmsg(test); #if defined(HAVE_SSL) if (test_is_authorized(test) < 0){ @@ -3255,6 +3263,8 @@ iperf_defaults(struct iperf_test *testp) testp->json_callback = NULL; + testp->zerocopy = 0; + testp->settings->send_recvmmsg = 0; memset(testp->cookie, 0, COOKIE_SIZE); testp->multisend = 10; /* arbitrary */ @@ -3555,6 +3565,7 @@ iperf_reset_test(struct iperf_test *test) test->settings->dont_fragment = 0; test->zerocopy = 0; test->settings->skip_rx_copy = 0; + test->settings->send_recvmmsg = 0; #if defined(HAVE_SSL) if (test->settings->authtoken) { @@ -3616,6 +3627,8 @@ iperf_reset_stats(struct iperf_test *test) sp->omitted_cnt_error = sp->cnt_error; sp->omitted_outoforder_packets = sp->outoforder_packets; sp->jitter = 0; + sp->sendmmsg_buffered_packets_count = 0; + sp->pbuf = sp->buffer; rp = sp->result; rp->bytes_sent_omit = rp->bytes_sent; rp->bytes_received = 0; @@ -4694,9 +4707,15 @@ void iperf_free_stream(struct iperf_stream *sp) { struct iperf_interval_results *irp, *nirp; + int r; /* XXX: need to free interval list too! */ - munmap(sp->buffer, sp->test->settings->blksize); + r = munmap(sp->buffer, STREAM_BUFSIZE(sp->test)); + if (r < 0) + printf("Failed to release stream buffer at %p with size %d - errno=%d: %s)\n", sp->buffer, STREAM_BUFSIZE(sp->test), errno, strerror(errno)); + else if (sp->test->debug) + printf("Successfully released stream buffer at %p with size %d\n", sp->buffer, STREAM_BUFSIZE(sp->test)); + close(sp->buffer_fd); if (sp->diskfile_fd >= 0) close(sp->diskfile_fd); @@ -4707,6 +4726,11 @@ iperf_free_stream(struct iperf_stream *sp) free(sp->result); if (sp->send_timer != NULL) tmr_cancel(sp->send_timer); + + if (sp->pmsg_iov != NULL) + free(sp->pmsg_iov); + if (sp->msg != NULL) + free(sp->msg); free(sp); } @@ -4716,6 +4740,9 @@ iperf_new_stream(struct iperf_test *test, int s, int sender) { struct iperf_stream *sp; int ret = 0; + int burst; + int i; + char *pbuf; char template[1024]; if (test->tmp_template) { @@ -4774,19 +4801,50 @@ iperf_new_stream(struct iperf_test *test, int s, int sender) free(sp); return NULL; } - if (ftruncate(sp->buffer_fd, test->settings->blksize) < 0) { + if (ftruncate(sp->buffer_fd, STREAM_BUFSIZE(sp)) < 0) { i_errno = IECREATESTREAM; free(sp->result); free(sp); return NULL; } - sp->buffer = (char *) mmap(NULL, test->settings->blksize, PROT_READ|PROT_WRITE, MAP_SHARED, sp->buffer_fd, 0); + + sp->buffer = (char *) mmap(NULL, STREAM_BUFSIZE(sp), PROT_READ|PROT_WRITE, MAP_PRIVATE, sp->buffer_fd, 0); if (sp->buffer == MAP_FAILED) { i_errno = IECREATESTREAM; free(sp->result); free(sp); return NULL; } + else if (sp->test->debug) + printf("Successfully allocated stream buffer at addr %p with length %d\n", sp->buffer, STREAM_BUFSIZE(sp)); + + /* Allocate and initialize pointers for the messages in the buffer for UDP multi-msg send/receive */ + if (test->protocol->id == Pudp) { + burst = (sp->settings->send_recvmmsg == 0)? 1 : sp->settings->burst; + sp->msg = malloc(burst * sizeof(*sp->msg)); + if (sp->msg == NULL) { + i_errno = IECREATESTREAM; + iperf_free_stream(sp); + return NULL; + } + memset(sp->msg, 0, burst * sizeof(*sp->msg)); + + sp->pmsg_iov = malloc(burst * sizeof(struct iovec)); + if (sp->pmsg_iov == NULL) { + i_errno = IECREATESTREAM; + iperf_free_stream(sp); + return NULL; + } + memset(sp->pmsg_iov, 0, burst * sizeof(struct iovec)); + + for (i = 0, pbuf = sp->buffer; i < burst; i++, pbuf += sp->settings->blksize) { + sp->msg[i].msg_hdr.msg_iov = &sp->pmsg_iov[i]; + sp->msg[i].msg_hdr.msg_iov->iov_base = pbuf; + sp->msg[i].msg_hdr.msg_iov->iov_len = sp->settings->blksize; + sp->msg[i].msg_hdr.msg_iovlen = 1; + } + } + sp->pending_size = 0; /* Set socket */ @@ -4795,11 +4853,14 @@ iperf_new_stream(struct iperf_test *test, int s, int sender) sp->snd = test->protocol->send; sp->rcv = test->protocol->recv; + sp->sendmmsg_buffered_packets_count = 0; + sp->pbuf = sp->buffer; + if (test->diskfile_name != (char*) 0) { sp->diskfile_fd = open(test->diskfile_name, sender ? O_RDONLY : (O_WRONLY|O_CREAT|O_TRUNC), S_IRUSR|S_IWUSR); if (sp->diskfile_fd == -1) { i_errno = IEFILE; - munmap(sp->buffer, sp->test->settings->blksize); + munmap(sp->buffer, STREAM_BUFSIZE(sp)); free(sp->result); free(sp); return NULL; @@ -4813,13 +4874,13 @@ iperf_new_stream(struct iperf_test *test, int s, int sender) /* Initialize stream */ if (test->repeating_payload) - fill_with_repeating_pattern(sp->buffer, test->settings->blksize); + fill_with_repeating_pattern(sp->buffer, STREAM_BUFSIZE(sp)); else - ret = readentropy(sp->buffer, test->settings->blksize); + ret = readentropy(sp->buffer, STREAM_BUFSIZE(sp)); if ((ret < 0) || (iperf_init_stream(sp, test) < 0)) { close(sp->buffer_fd); - munmap(sp->buffer, sp->test->settings->blksize); + munmap(sp->buffer, STREAM_BUFSIZE(sp)); free(sp->result); free(sp); return NULL; @@ -4972,7 +5033,7 @@ diskfile_send(struct iperf_stream *sp) /* if needed, read enough data from the disk to fill up the buffer */ if (sp->diskfile_left < sp->test->settings->blksize && !sp->test->done) { - r = read(sp->diskfile_fd, sp->buffer, sp->test->settings->blksize - + r = read(sp->diskfile_fd, sp->buffer, STREAM_BUFSIZE(sp) - sp->diskfile_left); buffer_left += r; rtot += r; @@ -4982,8 +5043,8 @@ diskfile_send(struct iperf_stream *sp) // If the buffer doesn't contain a full buffer at this point, // adjust the size of the data to send. - if (buffer_left != sp->test->settings->blksize) { - if (sp->test->debug) + if (buffer_left != STREAM_BUFSIZE(sp)) { + if (sp->test->debug) printf("possible eof\n"); // setting data size to be sent, // which is less than full block/buffer size @@ -5572,3 +5633,27 @@ iperf_set_control_keepalive(struct iperf_test *test) return 0; } #endif //HAVE_TCP_KEEPALIVE + +/* + * Initialize whether sed/recvmmsg is supported for the test, + * along with initializing other related settings. + */ +void +iperf_init_send_recvmmsg(struct iperf_test *test) +{ +#ifdef HAVE_SENDMMSG + /* Set UDP `send_recvmmsg` from `zerocopy` settings (not used for disk file) */ + if (test->protocol->id == Pudp && test->zerocopy && test->diskfile_name == (char *)0) { + test->settings->send_recvmmsg = 1; + /* Ensure non-zero burst number of packets when sendmmsg/recvmmsg are used */ + if (test->settings->burst == 0) + test->settings->burst = 1; +#ifdef UIO_MAXIOV + /* Ensure burst size is appropriate for sendmmsg() */ + else if (test->settings->burst > UIO_MAXIOV) + test->settings->burst = UIO_MAXIOV; +#endif /* UIO_MAXIOV */ + } +#endif /* HAVE_SENDMMSG */ + return; +} diff --git a/src/iperf_locale.c b/src/iperf_locale.c index 8ed673866..c2b8fb667 100644 --- a/src/iperf_locale.c +++ b/src/iperf_locale.c @@ -211,6 +211,10 @@ const char usage_longstr[] = "Usage: iperf3 [-s|-c host] [options]\n" " -L, --flowlabel N set the IPv6 flow label (only supported on Linux)\n" #endif /* HAVE_FLOWLABEL */ " -Z, --zerocopy use a 'zero copy' method of sending data\n" +#if defined(HAVE_SENDMMSG) + " UDP: use `sendmmsg` to send `-b` bust number of messages;\n" + " (not supported when reading from a file).\n" +#endif /* HAVE_SENDMMSG */ #if defined(HAVE_MSG_TRUNC) " --skip-rx-copy ignore received messages using MSG_TRUNC option\n" #endif /* HAVE_MSG_TRUNC */ diff --git a/src/iperf_udp.c b/src/iperf_udp.c index c8835e6d7..f7823d731 100644 --- a/src/iperf_udp.c +++ b/src/iperf_udp.c @@ -24,6 +24,13 @@ * This code is distributed under a BSD style license, see the LICENSE * file for complete information. */ + +#include "iperf_config.h" + +#ifdef HAVE_SENDMMSG +#define _GNU_SOURCE /* required for sendmmg() */ +#endif /* HAVE_SENDMMSG */ + #include #include #include @@ -73,6 +80,52 @@ iperf_udp_recv(struct iperf_stream *sp) #endif /* HAVE_MSG_TRUNC */ r = Nrecv_no_select(sp->socket, sp->buffer, size, Pudp, sock_opt); +/* + * Tests under WSL Ubunto 20 shows that using `mmsgrecv()` has no prformance advantage + * over `Nread()`. It may be that is is because of extra "read" required to find that + * there are is more data to read. + * In addition, `mmsgrecv` requires the use of timeout parameter, + * as otherwise it can get stack after receiving the last message send. + * However, adding the timeout may have major impact on the performance. + * Therefore, using `mmsgrecv` is commented out. + * To fully support `mmsgrecv`, in addition to uncommenting (here and in other parts + * of `iperf_udp_recv`), `sp->buffer` should be changed to `pbuf`. + */ +/**************** NOT USING `mmsgrecv()` ****************************** + int msgs_recvd; + int i; + char *pbuf; + + // Select message reading method +#ifdef HAVE_SEND_RECVMMSG + + struct timespec tmo; + + if (sp->settings->send_recvmmsg == 1) { // Use recvmmsg() + // Set read timeout + tmo.tv_sec = sp->settings->rcv_timeout.secs; + tmo.tv_nsec = sp->settings->rcv_timeout.usecs; + + // Receive at least one message + do { + msgs_recvd = recvmmsg(sp->socket, sp->msg, sp->settings->burst, MSG_WAITFORONE, &tmo); + } while (msgs_recvd < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)); + if (msgs_recvd <= 0) { + r = msgs_recvd; + } else { + for (i = 0, r = 0; i < msgs_recvd; i++) { + r += sp->msg[i].msg_hdr.msg_iov->iov_len; + } + } + } // recvmmsg + else +#endif // HAVE_SEND_RECVMMSG + { // Nread + r = Nread(sp->socket, sp->buffer, size, Pudp); + msgs_recvd = 1; + // `sp->msg[0].msg_hdr.msg_iov->iov_base = sp->buffer;` - not needed as initialized as part of Stream init + } +**************** NOT USING `mmsgrecv()` ******************************/ /* * If we got an error in the read, or if we didn't read anything @@ -96,6 +149,12 @@ iperf_udp_recv(struct iperf_stream *sp) sp->result->bytes_received += r; sp->result->bytes_received_this_interval += r; +/**************** NOT USING `mmsgrecv()` ****************************** + // Go over all nessages received to evalauet packet count and timings + for (i = 0; i < msgs_recvd; i++) { + pbuf = sp->msg[i].msg_hdr.msg_iov->iov_base; // current msg buffer +**************** NOT USING `mmsgrecv()` ******************************/ + /* Dig the various counters out of the incoming UDP packet */ if (test->udp_counters_64bit) { memcpy(&sec, sp->buffer, sizeof(sec)); @@ -148,11 +207,11 @@ iperf_udp_recv(struct iperf_stream *sp) sp->packet_count = pcount; } else { - /* - * Sequence number went backward (or was stationary?!?). - * This counts as an out-of-order packet. - */ - sp->outoforder_packets++; + /* + * Sequence number went backward (or was stationary?!?). + * This counts as an out-of-order packet. + */ + sp->outoforder_packets++; /* * If we have lost packets, then the fact that we are now @@ -184,16 +243,23 @@ iperf_udp_recv(struct iperf_stream *sp) iperf_time_diff(&arrival_time, &sent_time, &temp_time); transit = iperf_time_in_secs(&temp_time); - /* Hack to handle the first packet by initializing prev_transit. */ - if (first_packet) - sp->prev_transit = transit; + /* Hack to handle the first packet by initializing prev_transit. */ + if (first_packet) { + sp->prev_transit = transit; + first_packet = 0; + } d = transit - sp->prev_transit; if (d < 0) d = -d; sp->prev_transit = transit; sp->jitter += (d - sp->jitter) / 16.0; - } + +/**************** NOT USING `mmsgrecv()` ****************************** + } // for over all received messages +**************** NOT USING `mmsgrecv()` ******************************/ + + } /* if state is TEST_RUNNING */ else { if (test->debug_level >= DEBUG_LEVEL_INFO) printf("Late receive, state = %d\n", test->state); @@ -210,44 +276,98 @@ iperf_udp_recv(struct iperf_stream *sp) int iperf_udp_send(struct iperf_stream *sp) { - int r; - int size = sp->settings->blksize; + int r = 0; + int size = sp->settings->blksize; struct iperf_time before; + char *buf = sp->buffer; + uint32_t sec, usec; - iperf_time_now(&before); +#ifdef HAVE_SENDMMSG + int i, j, k; + char *b; +#endif /* HAVE_SENDMMSG */ - ++sp->packet_count; +#ifdef HAVE_SENDMMSG + /* if sendmmsg is used - set buffer pointer to next buffer */ + if (sp->settings->send_recvmmsg == 1) { + i = sp->sendmmsg_buffered_packets_count++; + sp->msg[i].msg_hdr.msg_iovlen = 1; + buf = sp->pbuf; + sp->msg[i].msg_hdr.msg_iov->iov_base = buf; + sp->pbuf += sp->settings->blksize; + sp->msg[i].msg_hdr.msg_iov->iov_len = size; + } +#endif /* HAVE_SENDMMSG */ + /* Set message packet count */ + ++sp->packet_count; if (sp->test->udp_counters_64bit) { - - uint32_t sec, usec; uint64_t pcount; - - sec = htonl(before.secs); - usec = htonl(before.usecs); pcount = htobe64(sp->packet_count); + memcpy(buf+8, &pcount, sizeof(pcount)); + } else { + uint32_t pcount; + pcount = htonl(sp->packet_count); + memcpy(buf+8, &pcount, sizeof(pcount)); + } + +/* Use sendmmsg when approriate to send the packet, else use Nwrite */ +#ifdef HAVE_SENDMMSG + if (sp->settings->send_recvmmsg == 1) { + /* When enough "burst" packets were buffered - send them */ + if (sp->sendmmsg_buffered_packets_count >= sp->settings->burst) { + /* Set actual sending time to all packets*/ + iperf_time_now(&before); + sec = htonl(before.secs); + usec = htonl(before.usecs); + for (i = 0; i < sp->sendmmsg_buffered_packets_count; i++) { + b = sp->msg[i].msg_hdr.msg_iov->iov_base; + memcpy(b, &sec, sizeof(sec)); + memcpy(b+4, &usec, sizeof(usec)); + } - memcpy(sp->buffer, &sec, sizeof(sec)); - memcpy(sp->buffer+4, &usec, sizeof(usec)); - memcpy(sp->buffer+8, &pcount, sizeof(pcount)); + /* Sending messages and making sure all packets are sent */ + i = 0; /* count of messages sent */ + r = 0; /* total bytes sent */ + while (i < sp->sendmmsg_buffered_packets_count) { + j = sendmmsg(sp->socket, &sp->msg[i], sp->sendmmsg_buffered_packets_count - i, MSG_DONTWAIT); + if (j < 0) { + r = j; + break; + } + + if (sp->test->debug && i+j < sp->sendmmsg_buffered_packets_count) + printf("sendmmsg() sent only %d messges out of %d still bufferred\n", + j, sp->sendmmsg_buffered_packets_count-i); + + for (k = i; k < i+j; k++) { /* accumulate number of bytes sent */ + r += sp->msg[k].msg_len; + } + i += j; /* accumulate number of messages received */ + } + if (sp->test->debug_level >= DEBUG_LEVEL_DEBUG) + printf("sendmmsg() %s. Sent %d messges out of %d bufferred. %d bytes sent. (errno=%d: %s)\n", + ((r > 0)? "succesful":"FAILED"), i, sp->sendmmsg_buffered_packets_count, r, errno, strerror(errno)); + + sp->sendmmsg_buffered_packets_count = 0; + sp->pbuf = sp->buffer; + } } - else { - - uint32_t sec, usec, pcount; - - sec = htonl(before.secs); + else +#endif /* HAVE_SENDMMSG */ + { + /* Set sending time */ + iperf_time_now(&before); + sec = htonl(before.secs); usec = htonl(before.usecs); - pcount = htonl(sp->packet_count); - - memcpy(sp->buffer, &sec, sizeof(sec)); - memcpy(sp->buffer+4, &usec, sizeof(usec)); - memcpy(sp->buffer+8, &pcount, sizeof(pcount)); + memcpy(buf, &sec, sizeof(sec)); + memcpy(buf+4, &usec, sizeof(usec)); + /* Write singe message */ + r = Nwrite(sp->socket, buf, size, Pudp); } - r = Nwrite(sp->socket, sp->buffer, size, Pudp); - if (r <= 0) { --sp->packet_count; /* Don't count messages that no data was sent from them. * Allows "resending" a massage with the same numbering */ @@ -261,8 +381,10 @@ iperf_udp_send(struct iperf_stream *sp) sp->result->bytes_sent += r; sp->result->bytes_sent_this_interval += r; - if (sp->test->debug_level >= DEBUG_LEVEL_DEBUG) - printf("sent %d bytes of %d, total %" PRIu64 "\n", r, sp->settings->blksize, sp->result->bytes_sent); + if (sp->test->debug_level >= DEBUG_LEVEL_DEBUG) { + if (sp->settings->send_recvmmsg == 0 || r > 0) + printf("sent %d bytes of %d bytes buffers, total %" PRIu64 "\n", r, size, sp->result->bytes_sent); + } return r; } @@ -325,7 +447,7 @@ iperf_udp_buffercheck(struct iperf_test *test, int s) char str[WARN_STR_LEN]; snprintf(str, sizeof(str), "Block size %d > sending socket buffer size %d", - test->settings->blksize, sndbuf_actual); + STREAM_BUFSIZE(test), sndbuf_actual); warning(str); rc = 1; } @@ -343,11 +465,11 @@ iperf_udp_buffercheck(struct iperf_test *test, int s) i_errno = IESETBUF2; return -1; } - if (test->settings->blksize > rcvbuf_actual) { + if (STREAM_BUFSIZE(test) > rcvbuf_actual) { char str[WARN_STR_LEN]; snprintf(str, sizeof(str), "Block size %d > receiving socket buffer size %d", - test->settings->blksize, rcvbuf_actual); + STREAM_BUFSIZE(test), rcvbuf_actual); warning(str); rc = 1; } @@ -537,7 +659,7 @@ iperf_udp_connect(struct iperf_test *test) if (rc > 0) { if (test->settings->socket_bufsize == 0) { char str[WARN_STR_LEN]; - int bufsize = test->settings->blksize + UDP_BUFFER_EXTRA; + int bufsize = STREAM_BUFSIZE(test) + UDP_BUFFER_EXTRA; snprintf(str, sizeof(str), "Increasing socket buffer size to %d", bufsize); warning(str);