enh: Add --rcv-timeout option (esnet#1125)

Enhancement to PR esnet#1101 - add --rcv-timeout option to allow setting the timeout for receiving packet from the sender in a running test, instead of the constant 120 seconds set in esnet#1101. This is to allow short timeout, so the server will not be "busy" for long time doing nothing. Also, the resolution is set to ms (with minimum of 100ms), as at least in fast networks that may be required.

Since both the server and the client can be a sender, the option is allowed for both. The server setting is for all tests - the client is not sending its --rcv-timeout setting to the server.

While changing the help message, few printf constants from other help lines were changed to parameters.

src/iperf.h

@@ -162,6 +162,7 @@ struct iperf_settings
 #endif // HAVE_SSL
     int	      connect_timeout;	    /* socket connection timeout, in ms */
     int       idle_timeout;         /* server idle time timeout */
+    struct iperf_time rcv_timeout;  /* Timeout for receiving messages in active mode, in us */
 };
 
 struct iperf_test;
@@ -383,6 +384,8 @@ struct iperf_test
 /* default settings */
 #define PORT 5201  /* default port to listen on (don't use the same port as iperf2) */
 #define uS_TO_NS 1000
+#define mS_TO_US 1000
+#define SEC_TO_mS 1000
 #define SEC_TO_US 1000000LL
 #define UDP_RATE (1024 * 1024) /* 1 Mbps */
 #define OMIT 0 /* seconds */
@@ -407,7 +410,6 @@ struct iperf_test
 #define MAX_BURST 1000
 #define MAX_MSS (9 * 1024)
 #define MAX_STREAMS 128
-#define NO_MSG_RCVD_TIMEOUT 120
 
 #define TIMESTAMP_FORMAT "%c "
 

src/iperf_api.c

@@ -117,7 +117,7 @@ usage()
 void
 usage_long(FILE *f)
 {
-    fprintf(f, usage_longstr, UDP_RATE / (1024*1024), DURATION, DEFAULT_TCP_BLKSIZE / 1024, DEFAULT_UDP_BLKSIZE);
+    fprintf(f, usage_longstr, DEFAULT_NO_MSG_RCVD_TIMEOUT, UDP_RATE / (1024*1024), DEFAULT_PACING_TIMER, DURATION, DEFAULT_TCP_BLKSIZE / 1024, DEFAULT_UDP_BLKSIZE);
 }
 
 
@@ -400,6 +400,12 @@ iperf_get_dont_fragment(struct iperf_test *ipt)
     return ipt->settings->dont_fragment;
 }
 
+struct iperf_time*
+iperf_get_test_rcv_timeout(struct iperf_test *ipt)
+{
+    return &ipt->settings->rcv_timeout;
+}
+
 char*
 iperf_get_test_congestion_control(struct iperf_test* ipt)
 {
@@ -749,6 +755,13 @@ iperf_set_dont_fragment(struct iperf_test* ipt, int dnf)
     ipt->settings->dont_fragment = dnf;
 }
 
+void
+iperf_set_test_rcv_timeout(struct iperf_test* ipt, struct iperf_time* to)
+{
+    ipt->settings->rcv_timeout.secs = to->secs;
+    ipt->settings->rcv_timeout.usecs = to->usecs;
+}
+
 void
 iperf_set_test_congestion_control(struct iperf_test* ipt, char* cc)
 {
@@ -994,24 +1007,26 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
 	{"pacing-timer", required_argument, NULL, OPT_PACING_TIMER},
 	{"connect-timeout", required_argument, NULL, OPT_CONNECT_TIMEOUT},
         {"idle-timeout", required_argument, NULL, OPT_IDLE_TIMEOUT},
+        {"rcv-timeout", required_argument, NULL, OPT_RCV_TIMEOUT},
         {"debug", no_argument, NULL, 'd'},
         {"help", no_argument, NULL, 'h'},
         {NULL, 0, NULL, 0}
     };
     int flag;
     int portno;
     int blksize;
-    int server_flag, client_flag, rate_flag, duration_flag;
+    int server_flag, client_flag, rate_flag, duration_flag, rcv_timeout_flag;
     char *endptr;
 #if defined(HAVE_CPU_AFFINITY)
     char* comma;
 #endif /* HAVE_CPU_AFFINITY */
     char* slash;
     struct xbind_entry *xbe;
     double farg;
+    int rcv_timeout_in = 0;
 
     blksize = 0;
-    server_flag = client_flag = rate_flag = duration_flag = 0;
+    server_flag = client_flag = rate_flag = duration_flag = rcv_timeout_flag = 0;
 #if defined(HAVE_SSL)
     char *client_username = NULL, *client_rsa_public_key = NULL, *server_rsa_private_key = NULL;
 #endif /* HAVE_SSL */
@@ -1323,6 +1338,16 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
                 }
 		server_flag = 1;
 	        break;
+            case OPT_RCV_TIMEOUT:
+                rcv_timeout_in = atoi(optarg);
+                if (rcv_timeout_in < MIN_NO_MSG_RCVD_TIMEOUT || rcv_timeout_in > MAX_TIME * SEC_TO_mS) {
+                    i_errno = IERCVTIMEOUT;
+                    return -1;
+                }
+                test->settings->rcv_timeout.secs = rcv_timeout_in / SEC_TO_mS;
+                test->settings->rcv_timeout.usecs = (rcv_timeout_in % SEC_TO_mS) * mS_TO_US;
+                rcv_timeout_flag = 1;
+	        break;
             case 'A':
 #if defined(HAVE_CPU_AFFINITY)
                 test->affinity = strtol(optarg, &endptr, 0);
@@ -1489,6 +1514,9 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
     } else if (test->role == 'c' && (test->server_skew_threshold != 0)){
         i_errno = IESERVERONLY;
         return -1;
+    } else if (test->role == 'c' && rcv_timeout_flag && test->mode == SENDER){
+        i_errno = IERVRSONLYRCVTIMEOUT;
+        return -1;
     } else if (test->role == 's' && (server_rsa_private_key || test->server_authorized_users) && 
         !(server_rsa_private_key && test->server_authorized_users)) {
          i_errno = IESETSERVERAUTH;
@@ -2621,12 +2649,15 @@ iperf_defaults(struct iperf_test *testp)
     testp->settings->bitrate_limit_interval = 5;
     testp->settings->bitrate_limit_stats_per_interval = 0;
     testp->settings->fqrate = 0;
-    testp->settings->pacing_timer = 1000;
+    testp->settings->pacing_timer = DEFAULT_PACING_TIMER;
     testp->settings->burst = 0;
     testp->settings->mss = 0;
     testp->settings->bytes = 0;
     testp->settings->blocks = 0;
     testp->settings->connect_timeout = -1;
+    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;
+
     memset(testp->cookie, 0, COOKIE_SIZE);
 
     testp->multisend = 10;	/* arbitrary */

src/iperf_api.h

@@ -56,6 +56,9 @@ typedef uint64_t iperf_size_t;
 #define DEFAULT_UDP_BLKSIZE 1460 /* default is dynamically set, else this */
 #define DEFAULT_TCP_BLKSIZE (128 * 1024)  /* default read/write block size */
 #define DEFAULT_SCTP_BLKSIZE (64 * 1024)
+#define DEFAULT_PACING_TIMER 1000
+#define DEFAULT_NO_MSG_RCVD_TIMEOUT 120000
+#define MIN_NO_MSG_RCVD_TIMEOUT 100
 
 /* short option equivalents, used to support options that only have long form */
 #define OPT_SCTP 1
@@ -83,6 +86,7 @@ typedef uint64_t iperf_size_t;
 #define OPT_BIND_DEV 24
 #define OPT_IDLE_TIMEOUT 25
 #define OPT_DONT_FRAGMENT 26
+#define OPT_RCV_TIMEOUT 27
 
 /* states */
 #define TEST_START 1
@@ -372,6 +376,8 @@ enum {
     IETOTALINTERVAL = 28,   // Invalid time interval for calculating average data rate
     IESKEWTHRESHOLD = 29,   // Invalid value specified as skew threshold
     IEIDLETIMEOUT = 30,     // Invalid value specified as idle state timeout
+    IERCVTIMEOUT = 31,      // Illegal message receive timeout
+    IERVRSONLYRCVTIMEOUT = 32,  // Client receive timeout is valid only in reverse mode
     /* Test errors */
     IENEWTEST = 100,        // Unable to create a new test (check perror)
     IEINITTEST = 101,       // Test initialization failed (check perror)

src/iperf_client_api.c

@@ -468,6 +468,12 @@ iperf_run_client(struct iperf_test * test)
     struct iperf_time now;
     struct timeval* timeout = NULL;
     struct iperf_stream *sp;
+    struct iperf_time last_receive_time;
+    struct iperf_time diff_time;
+    struct timeval used_timeout;
+    int64_t t_usecs;
+    int64_t timeout_us;
+    int64_t rcv_timeout_us;
 
     if (test->logfile)
         if (iperf_open_logfile(test) < 0)
@@ -497,19 +503,55 @@ iperf_run_client(struct iperf_test * test)
 
     /* Begin calculating CPU utilization */
     cpu_util(NULL);
+    if (test->mode != SENDER)
+        rcv_timeout_us = (test->settings->rcv_timeout.secs * SEC_TO_US) + test->settings->rcv_timeout.usecs;
+    else
+        rcv_timeout_us = 0;
 
     startup = 1;
     while (test->state != IPERF_DONE) {
 	memcpy(&read_set, &test->read_set, sizeof(fd_set));
 	memcpy(&write_set, &test->write_set, sizeof(fd_set));
 	iperf_time_now(&now);
 	timeout = tmr_timeout(&now);
+
+        // In reverse active mode client ensures data is received
+        if (test->state == TEST_RUNNING && rcv_timeout_us > 0) {
+            timeout_us = -1;
+            if (timeout != NULL) {
+                used_timeout.tv_sec = timeout->tv_sec;
+                used_timeout.tv_usec = timeout->tv_usec;
+                timeout_us = (timeout->tv_sec * SEC_TO_US) + timeout->tv_usec;
+            }
+            if (timeout_us < 0 || timeout_us > rcv_timeout_us) {
+                used_timeout.tv_sec = test->settings->rcv_timeout.secs;
+                used_timeout.tv_usec = test->settings->rcv_timeout.usecs;
+            }
+            timeout = &used_timeout;
+        }
+
 	result = select(test->max_fd + 1, &read_set, &write_set, NULL, timeout);
 	if (result < 0 && errno != EINTR) {
   	    i_errno = IESELECT;
 	    goto cleanup_and_fail;
-	}
+        } else if (result == 0 && test->state == TEST_RUNNING && rcv_timeout_us > 0) {
+            // If nothing was received in non-reverse running state then probably something got stack -
+            // either client, server or network, and test should be terminated.
+            iperf_time_now(&now);
+            if (iperf_time_diff(&now, &last_receive_time, &diff_time) == 0) {
+                t_usecs = iperf_time_in_usecs(&diff_time);
+                if (t_usecs > rcv_timeout_us) {
+                    i_errno = IENOMSG;
+                    goto cleanup_and_fail;
+                }
+
+            }
+        }
+
 	if (result > 0) {
+            if (rcv_timeout_us > 0) {
+                iperf_time_now(&last_receive_time);
+            }
 	    if (FD_ISSET(test->ctrl_sck, &read_set)) {
  	        if (iperf_handle_message_client(test) < 0) {
 		    goto cleanup_and_fail;

src/iperf_error.c

@@ -341,6 +341,14 @@ iperf_strerror(int int_errno)
             snprintf(errstr, len, "unable to set CPU affinity");
             perr = 1;
             break;
+        case IERCVTIMEOUT:
+            snprintf(errstr, len, "receive timeout value is incorrect or not in range");
+            perr = 1;
+            break;
+        case IERVRSONLYRCVTIMEOUT:
+            snprintf(errstr, len, "client receive timeout is valid only in receiving mode");
+            perr = 1;
+            break;
 	case IEDAEMON:
 	    snprintf(errstr, len, "unable to become a daemon");
 	    perr = 1;
@@ -431,7 +439,7 @@ iperf_strerror(int int_errno)
 	    snprintf(errstr, len, "idle timeout parameter is not positive or larget then allowed limit");
             break;
 	case IENOMSG:
-	    snprintf(errstr, len, "no message was received for %d seconds", NO_MSG_RCVD_TIMEOUT);
+	    snprintf(errstr, len, "message receiving timedout");
             break;
     case IESETDONTFRAGMENT:
 	    snprintf(errstr, len, "unable to set IP Do-Not-Fragment flag");

src/iperf_locale.c

@@ -119,21 +119,23 @@ const char usage_longstr[] = "Usage: iperf3 [-s|-c host] [options]\n"
                            "  -d, --debug               emit debugging output\n"
                            "  -v, --version             show version information and quit\n"
                            "  -h, --help                show this message and quit\n"
+                           "  --rcv-timeout #           timeout for receiving a message in active mode (ms)\n"
+                           "                            on receving side.  Default is %d\n"
                            "Server specific:\n"
                            "  -s, --server              run in server mode\n"
                            "  -D, --daemon              run the server as a daemon\n"
                            "  -1, --one-off             handle one client connection then exit\n"
 			   "  --server-bitrate-limit #[KMG][/#]   server's total bit rate limit (default 0 = no limit)\n"
 			   "                            (optional slash and number of secs interval for averaging\n"
 			   "                            total data rate.  Default is 5 seconds)\n"
-                           "  --idle-timeout #         restart server if is idle for #[sec] to overcome\n"
-                           "                           stacked server for different reasone\n"
+                           "  --idle-timeout #          restart server if is idle for #[sec] to overcome stacked\n"
+                           "                            server for different reasone (default - no timeout)\n"
 #if defined(HAVE_SSL)
                            "  --rsa-private-key-path    path to the RSA private key used to decrypt\n"
 			   "                            authentication credentials\n"
                            "  --authorized-users-path   path to the configuration file containing user\n"
                            "                            credentials\n"
-                           "  --time-skew-threshold    time skew threshold (in seconds) between the server\n"
+                           "  --time-skew-threshold     time skew threshold (in seconds) between the server\n"
                            "                            and client during the authentication process\n"
 #endif //HAVE_SSL
                            "Client specific:\n"
@@ -148,7 +150,7 @@ const char usage_longstr[] = "Usage: iperf3 [-s|-c host] [options]\n"
                            "  -b, --bitrate #[KMG][/#]  target bitrate in bits/sec (0 for unlimited)\n"
                            "                            (default %d Mbit/sec for UDP, unlimited for TCP)\n"
                            "                            (optional slash and packet count for burst mode)\n"
-			   "  --pacing-timer #[KMG]     set the timing for pacing, in microseconds (default 1000)\n"
+			   "  --pacing-timer #[KMG]     set the timing for pacing, in microseconds (default %d)\n"
 #if defined(HAVE_SO_MAX_PACING_RATE)
                            "  --fq-rate #[KMG]          enable fair-queuing based socket pacing in\n"
 			   "                            bits/sec (Linux only)\n"

src/iperf_server_api.c

@@ -424,9 +424,11 @@ iperf_run_server(struct iperf_test *test)
     struct iperf_time last_receive_time;
     struct iperf_time diff_time;
     struct timeval* timeout;
-    struct timeval default_timeout;
+    struct timeval used_timeout;
     int flag;
     int64_t t_usecs;
+    int64_t timeout_us;
+    int64_t rcv_timeout_us;
 
     if (test->logfile)
         if (iperf_open_logfile(test) < 0)
@@ -460,6 +462,7 @@ iperf_run_server(struct iperf_test *test)
     test->state = IPERF_START;
     send_streams_accepted = 0;
     rec_streams_accepted = 0;
+    rcv_timeout_us = (test->settings->rcv_timeout.secs * SEC_TO_US) + test->settings->rcv_timeout.usecs;
 
     while (test->state != IPERF_DONE) {
 
@@ -477,18 +480,24 @@ iperf_run_server(struct iperf_test *test)
 	timeout = tmr_timeout(&now);
 
         // Ensure select() will timeout to allow handling error cases that require server restart
-        if (timeout == NULL) {
-            default_timeout.tv_sec = 0;
-            if (test->state == IPERF_START) {
-                if (test-> settings->idle_timeout > 0)
-                    default_timeout.tv_sec = test-> settings->idle_timeout;
-            } else {
-                default_timeout.tv_sec = NO_MSG_RCVD_TIMEOUT;
+        if (test->state == IPERF_START) {       // In idle mode server may need to restart
+            if (timeout == NULL && test->settings->idle_timeout > 0) {
+                used_timeout.tv_sec = test->settings->idle_timeout;
+                used_timeout.tv_usec = 0;
+                timeout = &used_timeout;
             }
-            if (default_timeout.tv_sec > 0) {
-                default_timeout.tv_usec = 0;
-                timeout = &default_timeout;
+        } else if (test->mode != SENDER) {     // In non-reverse active mode server ensures data is received
+            timeout_us = -1;
+            if (timeout != NULL) {
+                used_timeout.tv_sec = timeout->tv_sec;
+                used_timeout.tv_usec = timeout->tv_usec;
+                timeout_us = (timeout->tv_sec * SEC_TO_US) + timeout->tv_usec;
             }
+            if (timeout_us < 0 || timeout_us > rcv_timeout_us) {
+                used_timeout.tv_sec = test->settings->rcv_timeout.secs;
+                used_timeout.tv_usec = test->settings->rcv_timeout.usecs;
+            }
+            timeout = &used_timeout;
         }
 
         result = select(test->max_fd + 1, &read_set, &write_set, NULL, timeout);
@@ -497,9 +506,9 @@ iperf_run_server(struct iperf_test *test)
             i_errno = IESELECT;
             return -1;
         } else if (result == 0) {
-            // If nothing was received during the last ... time
+            // If nothing was received during the specified time (per state)
             // then probably something got stack either at the client, server or network,
-            // and Test should forced to end.
+            // and Test should be forced to end.
             iperf_time_now(&now);
             t_usecs = 0;
             if (iperf_time_diff(&now, &last_receive_time, &diff_time) == 0) {
@@ -509,17 +518,17 @@ iperf_run_server(struct iperf_test *test)
                         test->server_forced_idle_restarts_count += 1;
                         if (test->debug)
                             printf("Server restart (#%d) in idle state as no connection request was received for %d sec\n",
-                                test->server_forced_idle_restarts_count, test-> settings->idle_timeout);
+                                test->server_forced_idle_restarts_count, test->settings->idle_timeout);
                         cleanup_server(test);
                         return 2;
                     }
                 }
-                else if (t_usecs > NO_MSG_RCVD_TIMEOUT * SEC_TO_US) {
+                else if (test->mode != SENDER && t_usecs > rcv_timeout_us) {
                     test->server_forced_no_msg_restarts_count += 1;
                     i_errno = IENOMSG;
                     if (iperf_get_verbose(test))
-                        iperf_err(test, "Server restart (#%d) in active test as no message was received for %d sec",
-                                  test->server_forced_no_msg_restarts_count, NO_MSG_RCVD_TIMEOUT);
+                        iperf_err(test, "Server restart (#%d) in active test as message receive timed-out",
+                                  test->server_forced_no_msg_restarts_count);
                     cleanup_server(test);
                     return -1;
                 }