UDP Connect retry mechanism

.gitignore

@@ -30,3 +30,5 @@ examples/.libs
 examples/Makefile
 examples/mic
 examples/mis
+
+.vscode/

src/iperf.h

@@ -165,6 +165,8 @@ struct iperf_settings
 #endif // HAVE_SSL
     int	      connect_timeout;	    /* socket connection timeout, in ms */
     int       idle_timeout;         /* server idle time timeout */
+    unsigned int udp_connect_retries; /* Number of UDP connection retries */
+    unsigned int udp_connect_retry_timeout; /* UDP connection retries timeout in secs */
     unsigned int snd_timeout; /* Timeout for sending tcp messages in active mode, in us */
     struct iperf_time rcv_timeout;  /* Timeout for receiving messages in active mode, in us */
 };
@@ -433,6 +435,8 @@ extern int gerror; /* error value from getaddrinfo(3), for use in internal error
 #define UDP_CONNECT_MSG 0x36373839          // "6789" - legacy value was 123456789
 #define UDP_CONNECT_REPLY 0x39383736        // "9876" - legacy value was 987654321
 #define LEGACY_UDP_CONNECT_REPLY 987654321  // Old servers may still reply with the legacy value
+#define UDP_ALL_STREAMS_CONNECTED_MSG 0x32333435  // "2345"
+#define UDP_ALL_STREAMS_CONNECTED_REPLY 0x35343332  // "5432"
 
 /* In Reverse mode, maximum number of packets to wait for "accept" response - to handle out of order packets */
 #define MAX_REVERSE_OUT_OF_ORDER_PACKETS 2

src/iperf_api.c

@@ -116,7 +116,7 @@ usage()
 void
 usage_long(FILE *f)
 {
-    fprintf(f, usage_longstr, DEFAULT_NO_MSG_RCVD_TIMEOUT, UDP_RATE / (1024*1024), DEFAULT_PACING_TIMER, DURATION, DEFAULT_TCP_BLKSIZE / 1024, DEFAULT_UDP_BLKSIZE);
+    fprintf(f, usage_longstr, DEFAULT_NO_MSG_RCVD_TIMEOUT, DEFAULT_UDP_CONNECT_RETRY_NUM, DEFAULT_UDP_CONNECT_RETRY_TIMEOUT, UDP_RATE / (1024*1024), DEFAULT_PACING_TIMER, DURATION, DEFAULT_TCP_BLKSIZE / 1024, DEFAULT_UDP_BLKSIZE);
 }
 
 
@@ -1012,6 +1012,7 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
         {"server", no_argument, NULL, 's'},
         {"client", required_argument, NULL, 'c'},
         {"udp", no_argument, NULL, 'u'},
+        {"udp-retry", optional_argument, NULL, OPT_UDP_RETRIES},
         {"bitrate", required_argument, NULL, 'b'},
         {"bandwidth", required_argument, NULL, 'b'},
 	{"server-bitrate-limit", required_argument, NULL, OPT_SERVER_BITRATE_LIMIT},
@@ -1095,6 +1096,7 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
     struct xbind_entry *xbe;
     double farg;
     int rcv_timeout_in = 0;
+    int udp_retries_timeout_specified = 0;
 
     blksize = 0;
     server_flag = client_flag = rate_flag = duration_flag = rcv_timeout_flag = snd_timeout_flag =0;
@@ -1191,6 +1193,36 @@ iperf_parse_arguments(struct iperf_test *test, int argc, char **argv)
                 set_protocol(test, Pudp);
 		client_flag = 1;
                 break;
+            case OPT_UDP_RETRIES:
+                set_protocol(test, Pudp); /* UDP connection retries implies UDP */
+                test->settings->udp_connect_retries = DEFAULT_UDP_CONNECT_RETRY_NUM;
+                test->settings->udp_connect_retry_timeout = DEFAULT_UDP_CONNECT_RETRY_TIMEOUT;
+                udp_retries_timeout_specified = 0;
+                if (optarg) {
+                    slash = strchr(optarg, '/');
+                    if (slash) {
+                        *slash = '\0';
+                        ++slash;
+                        if (strlen(optarg) > 0) { /* if retries timeout was specified */
+                            udp_retries_timeout_specified = 1;
+                            test->settings->udp_connect_retry_timeout = atof(slash);
+                            if (test->settings->udp_connect_retry_timeout < 1 || test->settings->udp_connect_retries > MAX_TIME) {
+                                i_errno = IEUDPCONNECT;
+                                return -1;
+                            }
+                        }
+                    }
+                    if (strlen(optarg) > 0) { /* if retries number was specified */
+                        test->settings->udp_connect_retries = atof(optarg);
+                        if (test->settings->udp_connect_retries < 1 ||
+                            (udp_retries_timeout_specified && test->settings->udp_connect_retries == 1)) {
+                            i_errno = IEUDPCONNECT;
+                            return -1;                        
+                        }
+                    }
+                }
+		client_flag = 1;
+                break;
             case OPT_SCTP:
 #if defined(HAVE_SCTP_H)
                 set_protocol(test, Psctp);
@@ -2170,6 +2202,8 @@ send_parameters(struct iperf_test *test)
 	    cJSON_AddNumberToObject(j, "repeating_payload", test->repeating_payload);
 	if (test->zerocopy)
 	    cJSON_AddNumberToObject(j, "zerocopy", test->zerocopy);
+        cJSON_AddNumberToObject(j, "udpconretry", test->settings->udp_connect_retries);
+        cJSON_AddNumberToObject(j, "udpconretry_timeout", test->settings->udp_connect_retry_timeout);
 #if defined(HAVE_DONT_FRAGMENT)
 	if (test->settings->dont_fragment)
 	    cJSON_AddNumberToObject(j, "dont_fragment", test->settings->dont_fragment);
@@ -2286,6 +2320,10 @@ get_parameters(struct iperf_test *test)
 	    test->repeating_payload = 1;
 	if ((j_p = cJSON_GetObjectItem(j, "zerocopy")) != NULL)
 	    test->zerocopy = j_p->valueint;
+	if ((j_p = cJSON_GetObjectItem(j, "udpconretry")) != NULL)
+	    test->settings->udp_connect_retries = j_p->valueint;
+	if ((j_p = cJSON_GetObjectItem(j, "udpconretry_timeout")) != NULL)
+	    test->settings->udp_connect_retry_timeout = j_p->valueint;
 #if defined(HAVE_DONT_FRAGMENT)
 	if ((j_p = cJSON_GetObjectItem(j, "dont_fragment")) != NULL)
 	    test->settings->dont_fragment = j_p->valueint;
@@ -2804,6 +2842,9 @@ iperf_defaults(struct iperf_test *testp)
     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;
+    testp->settings->udp_connect_retries = 1;
+    testp->settings->udp_connect_retry_timeout = DEFAULT_UDP_CONNECT_RETRY_TIMEOUT;
+
     testp->zerocopy = 0;
 
     memset(testp->cookie, 0, COOKIE_SIZE);
@@ -3092,6 +3133,9 @@ iperf_reset_test(struct iperf_test *test)
     test->settings->dont_fragment = 0;
     test->zerocopy = 0;
 
+    test->settings->udp_connect_retries = 1;
+    test->settings->udp_connect_retry_timeout = DEFAULT_UDP_CONNECT_RETRY_TIMEOUT;
+
 #if defined(HAVE_SSL)
     if (test->settings->authtoken) {
         free(test->settings->authtoken);

src/iperf_api.h

@@ -56,9 +56,11 @@ 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
+#define DEFAULT_PACING_TIMER 1000 /* [ms] */
+#define DEFAULT_NO_MSG_RCVD_TIMEOUT 120000 /* [ms] */
+#define MIN_NO_MSG_RCVD_TIMEOUT 100 /* [ms] */
+#define DEFAULT_UDP_CONNECT_RETRY_NUM 3
+#define DEFAULT_UDP_CONNECT_RETRY_TIMEOUT 10 /* [sec] */
 
 #define WARN_STR_LEN 128
 
@@ -90,6 +92,7 @@ typedef uint64_t iperf_size_t;
 #define OPT_DONT_FRAGMENT 26
 #define OPT_RCV_TIMEOUT 27
 #define OPT_SND_TIMEOUT 28
+#define OPT_UDP_RETRIES 29
 
 /* states */
 #define TEST_START 1
@@ -390,6 +393,7 @@ enum {
     IERCVTIMEOUT = 31,      // Illegal message receive timeout
     IERVRSONLYRCVTIMEOUT = 32,  // Client receive timeout is valid only in reverse mode
     IESNDTIMEOUT = 33,      // Illegal message send timeout
+    IEUDPCONNECT = 34,      // illegal optional arguments for udp-retry option
     /* Test errors */
     IENEWTEST = 100,        // Unable to create a new test (check perror)
     IEINITTEST = 101,       // Test initialization failed (check perror)
@@ -450,6 +454,9 @@ enum {
     IESTREAMREAD = 206,     // Unable to read from stream (check perror)
     IESTREAMCLOSE = 207,    // Stream has closed unexpectedly
     IESTREAMID = 208,       // Stream has invalid ID
+    IESTREAMCNCTSEND = 209, // Failed to send stream connection mesage/reply (UDP) 
+    IESTREAMCNCTED = 210,   // Server did not receive a response that all streams are connected (UDP)
+    IESTREAMCNCTEDREPLY = 211, // Client did not receive ack reply that the server received the response that all streams are connected (UDP)
     /* Timer errors */
     IENEWTIMER = 300,       // Unable to create new timer (check perror)
     IEUPDATETIMER = 301,    // Unable to update timer (check perror)

src/iperf_client_api.c

@@ -39,6 +39,7 @@
 
 #include "iperf.h"
 #include "iperf_api.h"
+#include "iperf_udp.h"
 #include "iperf_util.h"
 #include "iperf_locale.h"
 #include "iperf_time.h"
@@ -295,6 +296,10 @@ iperf_handle_message_client(struct iperf_test *test)
             }
             else if (iperf_create_streams(test, test->mode) < 0)
                 return -1;
+            if (test->protocol->id == Pudp) {
+                if (iperf_udp_send_all_streams_connected_msgs(test) < 0)
+                    return -1;
+            }
             break;
         case TEST_START:
             if (iperf_init_test(test) < 0)

src/iperf_error.c

@@ -350,6 +350,10 @@ iperf_strerror(int int_errno)
             snprintf(errstr, len, "send timeout value is incorrect or not in range");
             perr = 1;
             break;
+        case IEUDPCONNECT:
+            snprintf(errstr, len, "illegal optional arguments for udp-retry option");
+            perr = 1;
+            break;
         case IERVRSONLYRCVTIMEOUT:
             snprintf(errstr, len, "client receive timeout is valid only in receiving mode");
             perr = 1;
@@ -396,6 +400,15 @@ iperf_strerror(int int_errno)
         case IESTREAMID:
             snprintf(errstr, len, "stream has an invalid id");
             break;
+        case IESTREAMCNCTSEND:
+            snprintf(errstr, len, "failed to send stream connection mesage/reply");
+            break;
+        case IESTREAMCNCTED:
+            snprintf(errstr, len, "server did not receive a response that all streams are connected");
+            break;
+        case IESTREAMCNCTEDREPLY:
+            snprintf(errstr, len, "Client did not receive ack reply that the server received the response that all streams are connected");
+            break;
         case IENEWTIMER:
             snprintf(errstr, len, "unable to create new timer");
             perr = 1;

src/iperf_locale.c

@@ -154,6 +154,9 @@ const char usage_longstr[] = "Usage: iperf3 [-s|-c host] [options]\n"
                            "  --nstreams      #         number of SCTP streams\n"
 #endif /* HAVE_SCTP_H */
                            "  -u, --udp                 use UDP rather than TCP\n"
+                           "  --udp-retry[=#[/#]]       send UDP connection retries (implies also using UDP)\n"
+                           "                            (1st optional: number of connection retries (default %d))\n"
+                           "                            (2nd optional: timeout [sec] to receive connection msg (default %d))\n"
                            "  --connect-timeout #       timeout for control connection setup (ms)\n"
                            "  -b, --bitrate #[KMG][/#]  target bitrate in bits/sec (0 for unlimited)\n"
                            "                            (default %d Mbit/sec for UDP, unlimited for TCP)\n"

src/iperf_server_api.c

@@ -451,6 +451,8 @@ iperf_run_server(struct iperf_test *test)
     int64_t t_usecs;
     int64_t timeout_us;
     int64_t rcv_timeout_us;
+    struct sockaddr_storage sa_peer;
+    socklen_t sa_peer_len;
 
     if (test->logfile)
         if (iperf_open_logfile(test) < 0)
@@ -602,6 +604,8 @@ iperf_run_server(struct iperf_test *test)
             }
 
             if (test->state == CREATE_STREAMS) {
+                iperf_udp_discard_old_connect_messages(test, &read_set, 1); /* discard old connect requests but reply to them */
+
                 if (FD_ISSET(test->prot_listener, &read_set)) {
 
                     if ((s = test->protocol->accept(test)) < 0) {
@@ -725,12 +729,39 @@ iperf_run_server(struct iperf_test *test)
 
                             flag = -1;
                         }
-                    }
+                    } /* if !is_closed(s) */
                     FD_CLR(test->prot_listener, &read_set);
-                }
+                } /* input received in prot_listener */
 
 
+                /* check if all streams connections accepted */
                 if (rec_streams_accepted == streams_to_rec && send_streams_accepted == streams_to_send) {
+                    /* receive cleint's ACK that last connection ack was received */
+                    if (test->protocol->id == Pudp) {
+                        if (iperf_udp_acceppt_all_streams_connected_msgs(test, UDP_ALL_STREAMS_CONNECTED_MSG , test->prot_listener, &sa_peer, &sa_peer_len) == 0) {
+                            cleanup_server(test);
+                            i_errno = IESTREAMCNCTED;
+                            return -1;
+                        }
+
+                        /* send acks that all connected msg received -
+                           check status only for first message, as client may close socket before all messages are sent */
+                        if (test->settings->udp_connect_retries > 1) {
+                            if (test->debug_level >= DEBUG_LEVEL_INFO) {
+                                iperf_printf(test, "Sending %d replies to ack that all streams connected message was received (on Socket %d)\n", test->settings->udp_connect_retries, test->prot_listener);
+                            }
+                            /* bind the remote side of the socket to the client */
+                            if (iperf_udp_bind_to_accepted(test, test->prot_listener, &sa_peer, sa_peer_len) < 0) {
+                                return -1;
+                            }
+
+                            if (iperf_udp_send_connect_msg(test, test->prot_listener, UDP_ALL_STREAMS_CONNECTED_REPLY, 1) < 0) {
+                                cleanup_server(test);
+                                return -1;
+                            }
+                        }
+                    }
+
                     if (test->protocol->id != Ptcp) {
                         FD_CLR(test->prot_listener, &test->read_set);
                         close(test->prot_listener);
@@ -792,7 +823,7 @@ iperf_run_server(struct iperf_test *test)
                         return -1;
 		    }
                 }
-            }
+            } /* if CREATE_STREAMS */
 
             if (test->state == TEST_RUNNING) {
                 if (test->mode == BIDIRECTIONAL) {
@@ -818,15 +849,15 @@ iperf_run_server(struct iperf_test *test)
 		    }
                 }
 	    }
-        }
+        } /* if result > 0 */
 
 	if (result == 0 ||
 	    (timeout != NULL && timeout->tv_sec == 0 && timeout->tv_usec == 0)) {
 	    /* Run the timers. */
 	    iperf_time_now(&now);
 	    tmr_run(&now);
 	}
-    }
+    } /* while not IPERF_DONE */
 
 
     if (test->json_output) {