Merge pull request #1717 from davidBar-On/issue-1678-SKIP_RX_COPY-MSG_TRUNC

Add SKIP-RX-COPY support - using MSG_TRUNC socket option

Author: swlars

configure.ac

@@ -349,6 +349,20 @@ if test "x$iperf3_cv_header_tcp_info_snd_wnd" = "xyes"; then
   AC_DEFINE([HAVE_TCP_INFO_SND_WND], [1], [Have tcpi_snd_wnd field in tcp_info.])
 fi
 
+
+# Check for MSG_TRUNC (mostly on Linux)
+AC_CACHE_CHECK([MSG_TRUNC recv option],
+[iperf3_cv_header_msg_trunc],
+AC_COMPILE_IFELSE(
+  [AC_LANG_PROGRAM([[#include <sys/types.h>
+                     #include <sys/socket.h>
+                     #include <netinet/in.h>]],
+                   [[int foo = MSG_TRUNC;]])],
+  iperf3_cv_header_msg_trunc=yes,
+  iperf3_cv_header_msg_trunc=no))
+if test "x$iperf3_cv_header_msg_trunc" = "xyes"; then
+    AC_DEFINE([HAVE_MSG_TRUNC], [1], [Have MSG_TRUNC recv option.])
+fi
 # Check for IPPROTO_MPTCP (Linux)
 AC_CACHE_CHECK([MPTCP protocol],
 [iperf3_cv_header_ipproto_mptcp],

src/iperf.h

@@ -180,6 +180,7 @@ struct iperf_settings
     char      *client_password;
     EVP_PKEY  *client_rsa_pubkey;
 #endif // HAVE_SSL
+    int       skip_rx_copy;         /* Whether to ignore received messages data, using MSG_TRUNC option */
     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 */

src/iperf_api.c

@@ -1141,6 +1141,9 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
 #if defined(HAVE_DONT_FRAGMENT)
 	{"dont-fragment", no_argument, NULL, OPT_DONT_FRAGMENT},
 #endif /* HAVE_DONT_FRAGMENT */
+#if defined(HAVE_MSG_TRUNC)
+	{"skip-rx-copy", no_argument, NULL, OPT_SKIP_RX_COPY},
+#endif /* HAVE_MSG_TRUNC */
 #if defined(HAVE_SSL)
     {"username", required_argument, NULL, OPT_CLIENT_USERNAME},
     {"rsa-public-key-path", required_argument, NULL, OPT_CLIENT_RSA_PUBLIC_KEY},
@@ -1687,6 +1690,12 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
 	    test->use_pkcs1_padding = 1;
 	    break;
 #endif /* HAVE_SSL */
+#if defined(HAVE_MSG_TRUNC)
+            case OPT_SKIP_RX_COPY:
+                test->settings->skip_rx_copy = 1;
+                client_flag = 1;
+                break;
+#endif /* HAVE_MSG_TRUNC */
 	    case OPT_PACING_TIMER:
 		test->settings->pacing_timer = unit_atoi(optarg);
 		client_flag = 1;
@@ -2384,6 +2393,8 @@ send_parameters(struct iperf_test *test)
 	    cJSON_AddStringToObject(j, "authtoken", test->settings->authtoken);
 	}
 #endif // HAVE_SSL
+	if (test->settings->skip_rx_copy)
+	    cJSON_AddNumberToObject(j, "skip_rx_copy", test->settings->skip_rx_copy);
 	cJSON_AddStringToObject(j, "client_version", IPERF_VERSION);
 
 	if (test->debug) {
@@ -2494,6 +2505,8 @@ get_parameters(struct iperf_test *test)
 	if ((j_p = iperf_cJSON_GetObjectItemType(j, "authtoken", cJSON_String)) != NULL)
         test->settings->authtoken = strdup(j_p->valuestring);
 #endif //HAVE_SSL
+	if ((j_p = cJSON_GetObjectItem(j, "skip_rx_copy")) != NULL)
+	    test->settings->skip_rx_copy = j_p->valueint;
 	if (test->mode && test->protocol->id == Ptcp && has_tcpinfo_retransmits())
 	    test->sender_has_retransmits = 1;
 	if (test->settings->rate)
@@ -3108,13 +3121,15 @@ iperf_defaults(struct iperf_test *testp)
     testp->settings->rcv_timeout.secs = DEFAULT_NO_MSG_RCVD_TIMEOUT / SEC_TO_mS;
     testp->settings->rcv_timeout.usecs = (DEFAULT_NO_MSG_RCVD_TIMEOUT % SEC_TO_mS) * mS_TO_US;
     testp->zerocopy = 0;
+    testp->settings->skip_rx_copy = 0;
     testp->settings->cntl_ka = 0;
     testp->settings->cntl_ka_keepidle = 0;
     testp->settings->cntl_ka_interval = 0;
     testp->settings->cntl_ka_count = 0;
 
     testp->json_callback = NULL;
 
+
     memset(testp->cookie, 0, COOKIE_SIZE);
 
     testp->multisend = 10;	/* arbitrary */
@@ -3414,6 +3429,7 @@ iperf_reset_test(struct iperf_test *test)
     test->settings->tos = 0;
     test->settings->dont_fragment = 0;
     test->zerocopy = 0;
+    test->settings->skip_rx_copy = 0;
 
 #if defined(HAVE_SSL)
     if (test->settings->authtoken) {

src/iperf_api.h

@@ -101,6 +101,7 @@ typedef atomic_uint_fast64_t atomic_iperf_size_t;
 #define OPT_JSON_STREAM 28
 #define OPT_SND_TIMEOUT 29
 #define OPT_USE_PKCS1_PADDING 30
+#define OPT_SKIP_RX_COPY 32
 #define OPT_CNTL_KA 31
 
 /* states */

src/iperf_locale.c

@@ -209,6 +209,9 @@ 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_MSG_TRUNC)
+                           "  --skip-rx-copy            ignore received messages using MSG_TRUNC option\n"
+#endif /* HAVE_MSG_TRUNC */
                            "  -O, --omit N              perform pre-test for N seconds and omit the pre-test statistics\n"
                            "  -T, --title str           prefix every output line with this string\n"
                            "  --extra-data str          data string to include in client and server JSON\n"

src/iperf_tcp.c

@@ -57,20 +57,28 @@ int
 iperf_tcp_recv(struct iperf_stream *sp)
 {
     int r;
+    int sock_opt;
+
+#if defined(HAVE_MSG_TRUNC)
+    sock_opt = sp->test->settings->skip_rx_copy ? MSG_TRUNC : 0;
+#else
+    sock_opt = 0;
+#endif /* HAVE_MSG_TRUNC */
+
+    r = Nrecv_no_select(sp->socket, sp->buffer, sp->settings->blksize, Ptcp, sock_opt);
 
-    r = Nread_no_select(sp->socket, sp->buffer, sp->settings->blksize, Ptcp);
 
     if (r < 0)
         return r;
 
     /* Only count bytes received while we're in the correct state. */
     if (sp->test->state == TEST_RUNNING) {
-	sp->result->bytes_received += r;
-	sp->result->bytes_received_this_interval += r;
+	      sp->result->bytes_received += r;
+	      sp->result->bytes_received_this_interval += r;
     }
     else {
-	if (sp->test->debug)
-	    printf("Late receive, state = %d-%s\n", sp->test->state, state_to_text(sp->test->state));
+	      if (sp->test->debug)
+	          printf("Late receive, state = %d-%s\n", sp->test->state, state_to_text(sp->test->state));
     }
 
     return r;
@@ -87,12 +95,12 @@ iperf_tcp_send(struct iperf_stream *sp)
     int r;
 
     if (!sp->pending_size)
-	sp->pending_size = sp->settings->blksize;
+	      sp->pending_size = sp->settings->blksize;
 
     if (sp->test->zerocopy)
-	r = Nsendfile(sp->buffer_fd, sp->socket, sp->buffer, sp->pending_size);
+	      r = Nsendfile(sp->buffer_fd, sp->socket, sp->buffer, sp->pending_size);
     else
-	r = Nwrite(sp->socket, sp->buffer, sp->pending_size, Ptcp);
+	      r = Nwrite(sp->socket, sp->buffer, sp->pending_size, Ptcp);
 
     if (r < 0)
         return r;
@@ -102,8 +110,8 @@ iperf_tcp_send(struct iperf_stream *sp)
     sp->result->bytes_sent_this_interval += r;
 
     if (sp->test->debug_level >=  DEBUG_LEVEL_DEBUG)
-	printf("sent %d bytes of %d, pending %d, total %" PRIu64 "\n",
-	    r, sp->settings->blksize, sp->pending_size, sp->result->bytes_sent);
+	      printf("sent %d bytes of %d, pending %d, total %" PRIu64 "\n",
+	          r, sp->settings->blksize, sp->pending_size, sp->result->bytes_sent);
 
     return r;
 }

src/iperf_udp.c

@@ -61,9 +61,18 @@ iperf_udp_recv(struct iperf_stream *sp)
     int       first_packet = 0;
     double    transit = 0, d = 0;
     struct iperf_time sent_time, arrival_time, temp_time;
-    struct iperf_test *test = sp->test;
+    struct iperf_test *test = sp->test;	
+    int sock_opt = 0;
 
-    r = Nread_no_select(sp->socket, sp->buffer, size, Pudp);
+#if defined(HAVE_MSG_TRUNC)
+    // UDP recv() with MSG_TRUNC reads only the size bytes, but return the length of the full packet
+    if (sp->test->settings->skip_rx_copy) {
+        sock_opt = MSG_TRUNC;
+        size = sizeof(sec) + sizeof(usec) + sizeof(pcount);
+    }
+#endif /* HAVE_MSG_TRUNC */
+
+    r = Nrecv_no_select(sp->socket, sp->buffer, size, Pudp, sock_opt);
 
     /*
      * If we got an error in the read, or if we didn't read anything

src/net.c

@@ -365,16 +365,27 @@ netannounce(int domain, int proto, const char *local, const char *bind_dev, int
     return s;
 }
 
-
 /*******************************************************************/
-/* reads 'count' bytes from a socket  */
+/* Nread - reads 'count' bytes from a socket  */
 /********************************************************************/
 
 int
 Nread(int fd, char *buf, size_t count, int prot)
+{
+    return Nrecv(fd, buf, count, prot, 0);
+}
+
+/*******************************************************************/
+/* Nrecv - reads 'count' bytes from a socket  */
+/********************************************************************/
+
+int
+Nrecv(int fd, char *buf, size_t count, int prot, int sock_opt)
 {
     register ssize_t r;
-    register size_t nleft = count;
+    // `nleft` must be signed as it may get negative value for SKIP-RX-COPY UDP (MSG_TRUNC in sock_opt).
+    register ssize_t nleft = count;
+    register size_t total = 0;
     struct iperf_time ftimeout = { 0, 0 };
 
     fd_set rfdset;
@@ -403,7 +414,11 @@ Nread(int fd, char *buf, size_t count, int prot)
     }
 
     while (nleft > 0) {
-        r = read(fd, buf, nleft);
+        if (sock_opt)
+            r = recv(fd, buf, nleft, sock_opt);
+        else
+            r = read(fd, buf, nleft);
+
         if (r < 0) {
             /* XXX EWOULDBLOCK can't happen without non-blocking sockets */
             if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
@@ -413,7 +428,8 @@ Nread(int fd, char *buf, size_t count, int prot)
         } else if (r == 0)
             break;
 
-        nleft -= r;
+        total += r;
+        nleft -= r; 
         buf += r;
 
         /*
@@ -449,20 +465,32 @@ Nread(int fd, char *buf, size_t count, int prot)
             }
         }
     }
-    return count - nleft;
+    return total;
 }
 
 /********************************************************************/
-/* reads 'count' bytes from a socket - but without using select()   */
+/* Nreads 'count' bytes from a socket - but without using select()   */
 /********************************************************************/
 int
 Nread_no_select(int fd, char *buf, size_t count, int prot)
+{
+    return Nrecv_no_select(fd, buf, count, prot, 0);
+}
+
+/********************************************************************/
+/* Nrecv reads 'count' bytes from a socket - but without using select()   */
+/********************************************************************/
+int
+Nrecv_no_select(int fd, char *buf, size_t count, int prot, int sock_opt)
 {
     register ssize_t r;
     register size_t nleft = count;
 
     while (nleft > 0) {
-        r = read(fd, buf, nleft);
+        if (sock_opt)
+            r = recv(fd, buf, nleft, sock_opt);
+        else
+            r = read(fd, buf, nleft);
         if (r < 0) {
             /* XXX EWOULDBLOCK can't happen without non-blocking sockets */
             if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)

src/net.h

@@ -32,7 +32,9 @@ int create_socket(int domain, int type, int proto, const char *local, const char
 int netdial(int domain, int proto, const char *local, const char *bind_dev, int local_port, const char *server, int port, int timeout);
 int netannounce(int domain, int proto, const char *local, const char *bind_dev, int port);
 int Nread(int fd, char *buf, size_t count, int prot);
+int Nrecv(int fd, char *buf, size_t count, int prot, int sock_opt);
 int Nread_no_select(int fd, char *buf, size_t count, int prot);
+int Nrecv_no_select(int fd, char *buf, size_t count, int prot, int sock_opt);
 int Nwrite(int fd, const char *buf, size_t count, int prot) /* __attribute__((hot)) */;
 int has_sendfile(void);
 int Nsendfile(int fromfd, int tofd, const char *buf, size_t count) /* __attribute__((hot)) */;