feat: scatter streams to different cpu cores

Signed-off-by: Changliang Wu <changliang.wu@smartx.com>

Author: wu-cl

src/iperf.h

@@ -202,6 +202,7 @@ struct iperf_stream
     pthread_t thr;
     int thread_created;
     int       done;
+    int       affinity;
 
     /* configurable members */
     int       local_port;
@@ -293,6 +294,7 @@ enum debug_level {
 };
 
 
+#define MAX_STREAMS 128
 struct iperf_test
 {
     pthread_mutex_t print_mutex;
@@ -314,6 +316,9 @@ struct iperf_test
     int       duration;                         /* total duration of test (-t flag) */
     char     *diskfile_name;			/* -F option */
     int       affinity, server_affinity;	/* -A option */
+#if defined(HAVE_SCHED_SETAFFINITY)
+    cpu_set_t cpumask;
+#endif /* HAVE_CPUSET_SETAFFINITY */
 #if defined(HAVE_CPUSET_SETAFFINITY)
     cpuset_t cpumask;
 #endif /* HAVE_CPUSET_SETAFFINITY */
@@ -385,7 +390,8 @@ struct iperf_test
     double cpu_util[3];                            /* cpu utilization of the test - total, user, system */
     double remote_cpu_util[3];                     /* cpu utilization for the remote host/client - total, user, system */
 
-    int       num_streams;                      /* total streams in the test (-P) */
+    int num_streams; /* total streams in the test (-P) */
+    int streams_affinity[MAX_STREAMS];
 
     atomic_iperf_size_t bytes_sent;
     atomic_iperf_size_t blocks_sent;
@@ -463,7 +469,6 @@ struct iperf_test
 #define MAX_OMIT_TIME 600
 #define MAX_BURST 1000
 #define MAX_MSS (9 * 1024)
-#define MAX_STREAMS 128
 
 #define TIMESTAMP_FORMAT "%c "
 

src/iperf_api.c

@@ -4635,6 +4635,7 @@ iperf_new_stream(struct iperf_test *test, int s, int sender)
 
     memset(sp, 0, sizeof(struct iperf_stream));
 
+    sp->affinity = -1;
     sp->sender = sender;
     sp->test = test;
     sp->settings = test->settings;
@@ -5169,6 +5170,85 @@ iperf_json_finish(struct iperf_test *test)
 
 
 /* CPU affinity stuff - Linux, FreeBSD, and Windows only. */
+/* init streams cpu map
+   If the number of CPUs >= streams (-P), map each stream to a specific CPU core.
+ */
+void
+iperf_affinity_streams_init(struct iperf_test *test)
+{
+#if defined(HAVE_SCHED_SETAFFINITY)
+    if (sched_getaffinity(0, sizeof(cpu_set_t), &test->cpumask) != 0) {
+        return;
+    }
+
+    /* Assign bindings according to main process affinity */
+    int available_cpu_num = CPU_COUNT(&test->cpumask);
+    int affinity_stream_cnt = 0;
+    if (available_cpu_num > 0) {
+        affinity_stream_cnt = (test->num_streams / available_cpu_num) * available_cpu_num;
+    }
+
+    for (int i = 0; i < test->num_streams; i++) {
+        if (affinity_stream_cnt > 0) {
+            int offset = i % available_cpu_num;
+            int j = 0;
+            for (; j < CPU_SETSIZE; j++) {
+                if (CPU_ISSET(j, &test->cpumask)) {
+                    if (offset == 0) {
+                        break;
+                    }
+                    offset--;
+                }
+            }
+            test->streams_affinity[i] = j;
+            affinity_stream_cnt--;
+        } else {
+            test->streams_affinity[i] = -1;
+        }
+    }
+#else
+    for (int i = 0; i < test->num_streams; i++) {
+        test->streams_affinity[i] = -1;
+    }
+#endif
+}
+
+void
+iperf_setaffinity_streams_raw(int affinity)
+{
+#if defined(HAVE_SCHED_SETAFFINITY)
+    cpu_set_t cpu_set;
+    CPU_ZERO(&cpu_set);
+    CPU_SET(affinity, &cpu_set);
+    if (sched_setaffinity(0, sizeof(cpu_set_t), &cpu_set) != 0) {
+        return;
+    }
+#endif
+}
+
+void
+iperf_setaffinity_streams(struct iperf_test *test, int index)
+{
+#if defined(HAVE_SCHED_SETAFFINITY)
+    iperf_setaffinity_streams_raw(test->streams_affinity[index]);
+#endif
+}
+
+void
+iperf_setaffinity_streams_post(struct iperf_test *test)
+{
+#if defined(HAVE_SCHED_SETAFFINITY)
+    if (sched_setaffinity(0, sizeof(cpu_set_t), &test->cpumask) != 0) {
+        return;
+    }
+
+    struct iperf_stream *sp;
+    int index = 0;
+    SLIST_FOREACH(sp, &test->streams, streams){
+        sp->affinity = test->streams_affinity[index++ % test->num_streams];
+    }
+#endif
+}
 
 int
 iperf_setaffinity(struct iperf_test *test, int affinity)

src/iperf_api.h

@@ -385,6 +385,10 @@ int iperf_json_finish(struct iperf_test *);
 /* CPU affinity routines */
 int iperf_setaffinity(struct iperf_test *, int affinity);
 int iperf_clearaffinity(struct iperf_test *);
+void iperf_affinity_streams_init(struct iperf_test *test);
+void iperf_setaffinity_streams(struct iperf_test *test, int index);
+void iperf_setaffinity_streams_raw(int affinity);
+void iperf_setaffinity_streams_post(struct iperf_test *test);
 
 /* Custom printf routine. */
 int iperf_printf(struct iperf_test *test, const char *format, ...) __attribute__ ((format(printf,2,3)));

src/iperf_client_api.c

@@ -57,6 +57,10 @@ iperf_client_worker_run(void *s) {
     struct iperf_stream *sp = (struct iperf_stream *) s;
     struct iperf_test *test = sp->test;
 
+    if (sp->affinity >= 0) {
+        iperf_setaffinity_streams_raw(sp->affinity);
+    }
+
     /* Blocking signal to make sure that signal will be handled by main thread */
     sigset_t set;
     sigemptyset(&set);
@@ -112,6 +116,13 @@ iperf_create_streams(struct iperf_test *test, int sender)
 
     int orig_bind_port = test->bind_port;
     for (i = 0; i < test->num_streams; ++i) {
+        /* -> Create streams on different cpus
+           The network protocol stack will save the cpu core when creating the socket. 
+           With a multi-queue NIC, the queue selection may relate to the cpu core num
+           (XPS, RPS, sk_*_queue_mapping...), 
+           which helps improve the performance of multi-queue physical network cards.
+         */
+        iperf_setaffinity_streams(test,i);
 
         test->bind_port = orig_bind_port;
 	if (orig_bind_port) {
@@ -339,6 +350,7 @@ iperf_handle_message_client(struct iperf_test *test)
                 test->on_connect(test);
             break;
         case CREATE_STREAMS:
+            iperf_affinity_streams_init(test);
             if (test->mode == BIDIRECTIONAL)
             {
                 if (iperf_create_streams(test, 1) < 0)
@@ -348,6 +360,7 @@ iperf_handle_message_client(struct iperf_test *test)
             }
             else if (iperf_create_streams(test, test->mode) < 0)
                 return -1;
+            iperf_setaffinity_streams_post(test);
             break;
         case TEST_START:
             if (iperf_init_test(test) < 0)

src/iperf_server_api.c

@@ -70,6 +70,10 @@ iperf_server_worker_run(void *s) {
     struct iperf_stream *sp = (struct iperf_stream *) s;
     struct iperf_test *test = sp->test;
 
+    if (sp->affinity >= 0) {
+        iperf_setaffinity_streams_raw(sp->affinity);
+    }
+
     /* Blocking signal to make sure that signal will be handled by main thread */
     sigset_t set;
     sigemptyset(&set);
@@ -713,6 +717,7 @@ iperf_run_server(struct iperf_test *test)
                         streams_to_send = test->num_streams;
                         streams_to_rec = 0;
                     }
+                    iperf_affinity_streams_init(test);
                 }
             }
             if (FD_ISSET(test->ctrl_sck, &read_set)) {
@@ -725,7 +730,7 @@ iperf_run_server(struct iperf_test *test)
 
             if (test->state == CREATE_STREAMS) {
                 if (FD_ISSET(test->prot_listener, &read_set)) {
-
+                    iperf_setaffinity_streams(test,rec_streams_accepted);
                     if ((s = test->protocol->accept(test)) < 0) {
 			cleanup_server(test);
                         return -1;
@@ -844,6 +849,8 @@ iperf_run_server(struct iperf_test *test)
 
 
                 if (rec_streams_accepted == streams_to_rec && send_streams_accepted == streams_to_send) {
+                   iperf_setaffinity_streams_post(test);
+
                     if (test->protocol->id != Ptcp) {
                         FD_CLR(test->prot_listener, &test->read_set);
                         close(test->prot_listener);