tools: Add unidirectional test to operf

This adds a unidirectional test to operf, which is handy for testing
unidirectional streams.

Signed-off-by: Sander Vrijders <sander.vrijders@ugent.be>
Signed-off-by: Dimitri Staessens <dimitri.staessens@ugent.be>

4 files changed, 208 insertions, 88 deletions

diff --git a/src/tools/operf/operf.c b/src/tools/operf/operf.c
index fe387724..137e8647 100644
--- a/src/tools/operf/operf.c
+++ b/src/tools/operf/operf.c
@@ -62,19 +62,32 @@
 
 #define OPERF_MAX_FLOWS 256
 
+#define TEST_TYPE_UNI 0
+#define TEST_TYPE_BI  1
+
+struct conf {
+        uint32_t test_type;
+} __attribute__((packed));
+
+struct msg {
+        uint32_t id;
+} __attribute__((packed));
+
 struct c {
-        char * s_apn;
-        int    size;
+        char * server_name;
         long   rate;
         bool   flood;
         bool   sleep;
         int    duration;
+        int    size;
 
         unsigned long sent;
         unsigned long rcvd;
 
         pthread_t reader_pt;
         pthread_t writer_pt;
+
+        struct conf conf;
 } client;
 
 struct s {
@@ -82,6 +95,7 @@ struct s {
         fset_t *        flows;
         fqueue_t *      fq;
         pthread_mutex_t lock;
+        struct conf     conf[OPERF_MAX_FLOWS];
 
         uint8_t buffer[OPERF_BUF_SIZE];
         ssize_t timeout;
@@ -99,26 +113,62 @@ static void usage(void)
         printf("Usage: operf [OPTION]...\n"
                "Measures bandwidth between a client and a server\n"
                "  -l, --listen              Run in server mode\n"
-               "\n"
-               "  -n, --server-apn          Name of the operf server\n"
-               "  -d, --duration            Test duration (s, default 60)\n"
+               "  -t, --test                The type of test [uni, bi]"
+               " (default uni)\n"
+               "  -n, --server-name         Name of the operf server\n"
+               "  -d, --duration            Test duration (default 60s)\n"
                "  -r, --rate                Rate (b/s)\n"
                "  -s, --size                Payload size (B, default 1500)\n"
                "  -f, --flood               Send SDUs as fast as possible\n"
                "      --sleep               Sleep in between sending SDUs\n"
+               "\n"
                "      --help                Display this help text and exit\n");
 }
 
+/* Times are in ms. */
+static int time_mul(const char * rem)
+{
+        if (strcmp(rem, "ms") == 0 || strcmp(rem, "") == 0)
+                return 1;
+        else if(strcmp(rem, "s") == 0)
+                return 1000;
+        else if (strcmp(rem, "m") == 0)
+                return 60 * 1000;
+        else if (strcmp(rem, "h") == 0)
+                return 60 * 60 * 1000;
+        else if (strcmp(rem, "d") == 0)
+                return 60 * 60 * 24 * 1000;
+
+        printf("Unknown time unit: %s.\n", rem);
+
+        exit(EXIT_FAILURE);
+}
+
+static int rate_mul(const char * rem)
+{
+        if (strcmp(rem, "k") == 0 || strcmp(rem, "") == 0)
+                return 1000;
+        else if(strcmp(rem, "M") == 0)
+                return MILLION;
+        else if (strcmp(rem, "G") == 0)
+                return BILLION;
+
+        printf("Unknown rate unit: %s.\n", rem);
+
+        exit(EXIT_FAILURE);
+}
+
 int main(int argc, char ** argv)
 {
-        int ret = -1;
-        char * rem = NULL;
-        bool serv = false;
+        int    ret       = -1;
+        char * rem       = NULL;
+        bool   serv      = false;
+        char * type      = "uni";
 
         argc--;
         argv++;
 
-        client.s_apn = NULL;
+        client.server_name = NULL;
         client.size = 1500;
         client.duration = 60000;
         server.timeout = 1000; /* ms */
@@ -128,8 +178,8 @@ int main(int argc, char ** argv)
 
         while (argc > 0) {
                 if (strcmp(*argv, "-n") == 0 ||
-                           strcmp(*argv, "--server_apn") == 0) {
-                        client.s_apn = *(++argv);
+                           strcmp(*argv, "--server-name") == 0) {
+                        client.server_name = *(++argv);
                         --argc;
                 } else if (strcmp(*argv, "-s") == 0 ||
                            strcmp(*argv, "--size") == 0) {
@@ -137,17 +187,13 @@ int main(int argc, char ** argv)
                         --argc;
                 } else if (strcmp(*argv, "-d") == 0 ||
                            strcmp(*argv, "--duration") == 0) {
-                        client.duration = strtol(*(++argv), &rem, 10) * 1000;
+                        client.duration = strtol(*(++argv), &rem, 10);
+                        client.duration *= time_mul(rem);
                         --argc;
                 } else if (strcmp(*argv, "-r") == 0 ||
                            strcmp(*argv, "--rate") == 0) {
                         client.rate = strtol(*(++argv), &rem, 10);
-                        if (*rem == 'k')
-                                client.rate *= 1000;
-                        if (*rem == 'M')
-                                client.rate *= MILLION;
-                        if (*rem == 'G')
-                                client.rate *= BILLION;
+                        client.rate *= rate_mul(rem);
                         --argc;
                 } else if (strcmp(*argv, "-f") == 0 ||
                            strcmp(*argv, "--flood") == 0) {
@@ -157,6 +203,10 @@ int main(int argc, char ** argv)
                 } else if (strcmp(*argv, "-l") == 0 ||
                            strcmp(*argv, "--listen") == 0) {
                         serv = true;
+                } else if (strcmp(*argv, "-t") == 0 ||
+                           strcmp(*argv, "--test") == 0) {
+                        type = *(++argv);
+                        --argc;
                 } else {
                         usage();
                         exit(EXIT_SUCCESS);
@@ -168,11 +218,20 @@ int main(int argc, char ** argv)
         if (serv) {
                 ret = server_main();
         } else {
-                if (client.s_apn == NULL) {
+                if (client.server_name == NULL) {
                         printf("No server specified.\n");
-                        usage();
-                        exit(EXIT_SUCCESS);
+                        exit(EXIT_FAILURE);
                 }
+
+                if (strcmp(type, "uni") == 0)
+                        client.conf.test_type = TEST_TYPE_UNI;
+                else if (strcmp(type, "bi") == 0)
+                        client.conf.test_type = TEST_TYPE_BI;
+                else {
+                        printf("Invalid test type.\n");
+                        exit(EXIT_FAILURE);
+                }
+
                 if (client.size > OPERF_BUF_SIZE) {
                         printf("Packet size truncated to %d bytes.\n",
                                OPERF_BUF_SIZE);
diff --git a/src/tools/operf/operf_client.c b/src/tools/operf/operf_client.c
index d299e239..c8873c54 100644
--- a/src/tools/operf/operf_client.c
+++ b/src/tools/operf/operf_client.c
@@ -36,6 +36,8 @@
  * OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 
+bool stop;
+
 static void busy_wait_until(const struct timespec * deadline)
 {
         struct timespec now;
@@ -45,7 +47,6 @@ static void busy_wait_until(const struct timespec * deadline)
         while (now.tv_sec == deadline->tv_sec
                && now.tv_nsec < deadline->tv_nsec)
                 clock_gettime(CLOCK_REALTIME, &now);
-        pthread_testcancel();
 }
 
 void shutdown_client(int signo, siginfo_t * info, void * c)
@@ -57,8 +58,7 @@ void shutdown_client(int signo, siginfo_t * info, void * c)
         case SIGINT:
         case SIGTERM:
         case SIGHUP:
-                pthread_cancel(client.reader_pt);
-                pthread_cancel(client.writer_pt);
+                stop = true;
         default:
                 return;
         }
@@ -74,7 +74,14 @@ void * reader(void * o)
 
         fccntl(fd, FLOWSRCVTIMEO, &timeout);
 
-        while ((msg_len = flow_read(fd, buf, OPERF_BUF_SIZE)) != -ETIMEDOUT) {
+        while (!stop) {
+                msg_len = flow_read(fd, buf, OPERF_BUF_SIZE);
+                if (msg_len == -ETIMEDOUT) {
+                        printf("Server timed out.\n");
+                        stop = true;
+                        break;
+                }
+
                 if (msg_len != client.size) {
                         printf("Invalid message on fd %d.\n", fd);
                         continue;
@@ -96,7 +103,10 @@ void * writer(void * o)
         struct timespec intv = {(gap / BILLION), gap % BILLION};
         struct timespec end = {0, 0};
 
-        char * buf = malloc(client.size);
+        char *       buf;
+        struct msg * msg;
+
+        buf = malloc(client.size);
         if (buf == NULL)
                 return (void *) -ENOMEM;
 
@@ -107,53 +117,49 @@ void * writer(void * o)
 
         memset(buf, 0, client.size);
 
+        msg = (struct msg *) buf;
+
         if (client.flood)
                 printf("Flooding %s with %d byte SDUs for %d seconds.\n\n",
-                       client.s_apn, client.size, client.duration / 1000);
+                       client.server_name, client.size,
+                       client.duration / 1000);
         else
                 printf("Sending %d byte SDUs for %d s to %s at %.3lf Mb/s.\n\n",
-                       client.size, client.duration / 1000, client.s_apn,
+                       client.size, client.duration / 1000,
+                       client.server_name,
                        client.rate / (double) MILLION);
 
         clock_gettime(CLOCK_REALTIME, &start);
         clock_gettime(CLOCK_REALTIME, &now);
 
-        pthread_cleanup_push((void (*) (void *)) free, buf);
-
-        if (client.flood) {
-                while (ts_diff_ms(&start, &now) < client.duration) {
-                        if (flow_write(*fdp, buf, client.size) == -1) {
-                                printf("Failed to send SDU.\n");
-                                flow_dealloc(*fdp);
-                                free(buf);
-                                return (void *) -1;
-                        }
+        while (!stop && ts_diff_ms(&start, &now) < client.duration) {
+                if (!client.flood) {
+                        clock_gettime(CLOCK_REALTIME, &now);
+                        ts_add(&now, &intv, &end);
+                }
 
-                        ++client.sent;
+                msg->id = client.sent;
 
-                        clock_gettime(CLOCK_REALTIME, &now);
+                if (flow_write(*fdp, buf, client.size) == -1) {
+                        printf("Failed to send SDU.\n");
+                        flow_dealloc(*fdp);
+                        free(buf);
+                        return (void *) -1;
                 }
-        } else {
-                while (ts_diff_ms(&start, &now) < client.duration) {
-                        clock_gettime(CLOCK_REALTIME, &now);
-                        ts_add(&now, &intv, &end);
 
-                        if (flow_write(*fdp, buf, client.size) == -1) {
-                                printf("Failed to send SDU.\n");
-                                flow_dealloc(*fdp);
-                                free(buf);
-                                return (void *) -1;
-                        }
+                ++client.sent;
 
-                        ++client.sent;
+                if (!client.flood) {
                         if (client.sleep)
                                 nanosleep(&intv, NULL);
                         else
                                 busy_wait_until(&end);
+                } else {
+                        clock_gettime(CLOCK_REALTIME, &now);
                 }
         }
 
-        pthread_cleanup_pop(true);
+        free(buf);
 
         printf("Test finished.\n");
 
@@ -183,34 +189,51 @@ int client_main(void)
 
         client.sent = 0;
         client.rcvd = 0;
+        stop = false;
 
-        fd = flow_alloc(client.s_apn, NULL, NULL);
+        /* FIXME: Allow selecting QoS. */
+        fd = flow_alloc(client.server_name, NULL, NULL);
         if (fd < 0) {
                 printf("Failed to allocate flow.\n");
                 return -1;
         }
 
-        clock_gettime(CLOCK_REALTIME, &tic);
+        if (client.conf.test_type == TEST_TYPE_BI)
+                printf("Doing a bidirectional test.\n");
+        else
+                printf("Doing a unidirectional test.\n");
 
-        pthread_create(&client.reader_pt, NULL, reader, &fd);
-        pthread_create(&client.writer_pt, NULL, writer, &fd);
+        if (flow_write(fd, &client.conf, sizeof(client.conf))) {
+                printf("Failed to send configuration.\n");
+                flow_dealloc(fd);
+                return -1;
+        }
 
-        pthread_join(client.writer_pt, NULL);
+        sleep(1);
+
+        clock_gettime(CLOCK_REALTIME, &tic);
 
-        clock_gettime(CLOCK_REALTIME, &toc);
+        if (client.conf.test_type == TEST_TYPE_BI)
+                pthread_create(&client.reader_pt, NULL, reader, &fd);
 
-        pthread_join(client.reader_pt, NULL);
+        pthread_create(&client.writer_pt, NULL, writer, &fd);
+        pthread_join(client.writer_pt, NULL);
 
-        printf("\n");
-        printf("--- %s perf statistics ---\n", client.s_apn);
-        printf("%ld SDUs transmitted, ", client.sent);
-        printf("%ld received, ", client.rcvd);
-        printf("%ld%% packet loss, ", client.sent == 0 ? 0 :
-               100 - ((100 * client.rcvd) / client.sent));
-        printf("time: %.3f ms, ", ts_diff_us(&tic, &toc) / 1000.0);
-        printf("bandwidth: %.3lf Mb/s.\n",
-               (client.rcvd * client.size * 8)
-               / (double) ts_diff_us(&tic, &toc));
+        if (client.conf.test_type == TEST_TYPE_BI){
+                clock_gettime(CLOCK_REALTIME, &toc);
+                pthread_join(client.reader_pt, NULL);
+
+                printf("\n");
+                printf("--- %s perf statistics ---\n", client.server_name);
+                printf("%ld SDUs transmitted, ", client.sent);
+                printf("%ld received, ", client.rcvd);
+                printf("%ld%% packet loss, ", client.sent == 0 ? 0 :
+                       100 - ((100 * client.rcvd) / client.sent));
+                printf("time: %.3f ms, ", ts_diff_us(&tic, &toc) / 1000.0);
+                printf("bandwidth: %.3lf Mb/s.\n",
+                       (client.rcvd * client.size * 8)
+                       / (double) ts_diff_us(&tic, &toc));
+        }
 
         flow_dealloc(fd);
 
diff --git a/src/tools/operf/operf_server.c b/src/tools/operf/operf_server.c
index 11eb92fc..de32f320 100644
--- a/src/tools/operf/operf_server.c
+++ b/src/tools/operf/operf_server.c
@@ -36,6 +36,8 @@
  * OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 
+bool stop;
+
 void shutdown_server(int signo, siginfo_t * info, void * c)
 {
         (void) info;
@@ -46,6 +48,7 @@ void shutdown_server(int signo, siginfo_t * info, void * c)
         case SIGTERM:
         case SIGHUP:
                 pthread_cancel(server.accept_pt);
+                stop = true;
         default:
                 return;
         }
@@ -58,7 +61,7 @@ void * cleaner_thread(void * o)
 
         (void) o;
 
-        while (true) {
+        while (!stop) {
                 clock_gettime(CLOCK_REALTIME, &now);
                 pthread_mutex_lock(&server.lock);
                 for (i = 0; i < OPERF_MAX_FLOWS; ++i)
@@ -73,18 +76,21 @@ void * cleaner_thread(void * o)
                 pthread_mutex_unlock(&server.lock);
                 sleep(1);
         }
+
+        return (void *) 0;
 }
 
-void * server_thread(void *o)
+void * server_thread(void * o)
 {
-        int msg_len = 0;
+        int             msg_len = 0;
         struct timespec timeout = {0, 100 * MILLION};
-        struct timespec now = {0, 0};
-        int fd;
+        struct timespec now     = {0, 0};
+        int             fd;
+        struct msg *    msg;
 
         (void) o;
 
-        while (fevent(server.flows, server.fq, &timeout))
+        while (!stop && fevent(server.flows, server.fq, &timeout))
                 while ((fd = fqueue_next(server.fq)) >= 0) {
                         msg_len = flow_read(fd, server.buffer, OPERF_BUF_SIZE);
                         if (msg_len < 0)
@@ -96,9 +102,21 @@ void * server_thread(void *o)
                         server.times[fd] = now;
                         pthread_mutex_unlock(&server.lock);
 
-                        if (flow_write(fd, server.buffer, msg_len) < 0) {
-                                printf("Error writing to flow (fd %d).\n", fd);
-                                flow_dealloc(fd);
+                        msg = (struct msg *) server.buffer;
+
+                        if (server.conf[fd].test_type == TEST_TYPE_UNI)
+                                printf("Seqno %d from fd %d: %zd.%06zu\n",
+                                       msg->id, fd,
+                                       (ssize_t) now.tv_sec,
+                                       (size_t) now.tv_nsec / 1000);
+
+                        if (server.conf[fd].test_type == TEST_TYPE_BI) {
+                                if (flow_write(fd, server.buffer,
+                                               msg_len) < 0) {
+                                        printf("Error writing to flow "
+                                               "(fd %d).\n", fd);
+                                        flow_dealloc(fd);
+                                }
                         }
                 }
 
@@ -110,12 +128,13 @@ void * accept_thread(void * o)
         int fd = 0;
         struct timespec now = {0, 0};
         qosspec_t qs;
+        int len = 0;
 
         (void) o;
 
         printf("Ouroboros perf server started.\n");
 
-        while (true) {
+        while (!stop) {
                 fd = flow_accept(&qs, NULL);
                 if (fd < 0) {
                         printf("Failed to accept flow.\n");
@@ -124,6 +143,27 @@ void * accept_thread(void * o)
 
                 printf("New flow %d.\n", fd);
 
+                /* Read test type. */
+                len = flow_read(fd, &(server.conf[fd]),
+                                sizeof(server.conf[fd]));
+                if (len == -ETIMEDOUT) {
+                        printf("Failed to read config message.\n");
+                        flow_dealloc(fd);
+                        break;
+                }
+
+                /* Check if length was correct. */
+                if (flow_read(fd, NULL, 0) != 0) {
+                        printf("Invalid config message.\n");
+                        flow_dealloc(fd);
+                        break;
+                }
+
+                if (server.conf[fd].test_type == TEST_TYPE_BI)
+                        printf("Doing a bidirectional test.\n");
+                else
+                        printf("Doing a unidirectional test.\n");
+
                 clock_gettime(CLOCK_REALTIME, &now);
 
                 pthread_mutex_lock(&server.lock);
@@ -161,20 +201,19 @@ int server_main(void)
                 return -1;
         }
 
+        stop = false;
+
         pthread_create(&server.cleaner_pt, NULL, cleaner_thread, NULL);
         pthread_create(&server.accept_pt, NULL, accept_thread, NULL);
         pthread_create(&server.server_pt, NULL, server_thread, NULL);
 
         pthread_join(server.accept_pt, NULL);
 
-        pthread_cancel(server.server_pt);
-        pthread_cancel(server.cleaner_pt);
+        pthread_join(server.server_pt, NULL);
+        pthread_join(server.cleaner_pt, NULL);
 
         fset_destroy(server.flows);
         fqueue_destroy(server.fq);
 
-        pthread_join(server.server_pt, NULL);
-        pthread_join(server.cleaner_pt, NULL);
-
         return 0;
 }
diff --git a/src/tools/oping/oping.c b/src/tools/oping/oping.c
index bc8907e6..c49688b0 100644
--- a/src/tools/oping/oping.c
+++ b/src/tools/oping/oping.c
@@ -135,7 +135,7 @@ static void usage(void)
 /* Times are in ms. */
 static int time_mul(const char * rem)
 {
-        if (strcmp (rem, "ms") == 0 || strcmp(rem, "") == 0)
+        if (strcmp(rem, "ms") == 0 || strcmp(rem, "") == 0)
                 return 1;
         else if(strcmp(rem, "s") == 0)
                 return 1000;
@@ -148,7 +148,6 @@ static int time_mul(const char * rem)
 
         printf("Unknown time unit: %s.\n", rem);
 
-        usage();
         exit(EXIT_FAILURE);
 }
 
@@ -244,7 +243,7 @@ int main(int     argc,
                 if (client.s_apn == NULL) {
                         printf("No server specified.\n");
                         usage();
-                        exit(EXIT_SUCCESS);
+                        exit(EXIT_FAILURE);
                 }
                 if (client.interval > 10000) {
                         printf("Ping interval truncated to 10s.\n");
@@ -270,5 +269,5 @@ int main(int     argc,
 
  fail:
         usage();
-        exit(EXIT_SUCCESS);
+        exit(EXIT_FAILURE);
 }