rxperf: use parallel connections
[openafs.git] / src / rx / test / rxperf.c
index 88cb2ce..3f4a918 100644 (file)
@@ -79,9 +79,9 @@ nn * We are using getopt since we want it to be possible to link to
                                 * is hopefully a err.h */
 #endif
 #include <getopt.h>
-#include "rx.h"
-#include "rx_null.h"
-#include "rx_globals.h"
+#include <rx/rx.h>
+#include <rx/rx_null.h>
+#include <rx/rx_globals.h>
 
 #ifdef AFS_PTHREAD_ENV
 #include <pthread.h>
@@ -277,18 +277,30 @@ char somebuf[RXPERF_BUFSIZE];
 
 afs_int32 rxwrite_size = sizeof(somebuf);
 afs_int32 rxread_size = sizeof(somebuf);
+afs_int32 use_rx_readv = 0;
 
 static int
 do_readbytes(struct rx_call *call, afs_int32 bytes)
 {
+    struct iovec tiov[RX_MAXIOVECS];
     afs_int32 size;
+    int tnio;
+    int code;
 
     while (bytes > 0) {
        size = rxread_size;
+
        if (size > bytes)
            size = bytes;
-       if (rx_Read(call, somebuf, size) != size)
-           return 1;
+       if (use_rx_readv) {
+            if (size > RX_MAX_PACKET_DATA_SIZE)
+                size = RX_MAX_PACKET_DATA_SIZE;
+            code = rx_Readv(call, tiov, &tnio, RX_MAXIOVECS, size);
+        } else
+            code = rx_Read(call, somebuf, size);
+        if (code != size)
+            return 1;
+
        bytes -= size;
     }
     return 0;
@@ -820,23 +832,6 @@ do_client(const char *server, short port, char *filename, afs_int32 command,
 
     get_sec(0, &secureobj, &secureindex);
 
-    conn = rx_NewConnection(addr, htons(port), RX_SERVER_ID, secureobj, secureindex);
-    if (conn == NULL)
-       errx(1, "failed to contact server");
-
-#ifdef AFS_PTHREAD_ENV
-    pthread_attr_init(&tattr);
-    pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_JOINABLE);
-#endif
-
-    params->conn = conn;
-    params->filename = filename;
-    params->command = command;
-    params->times = times;
-    params->bytes = bytes;
-    params->sendbytes = sendbytes;
-    params->readbytes = readbytes;
-
     switch (command) {
     case RX_PERF_RPC:
         sprintf(stamp, "RPC: threads\t%d, times\t%d, write bytes\t%d, read bytes\t%d",
@@ -856,11 +851,38 @@ do_client(const char *server, short port, char *filename, afs_int32 command,
         break;
     }
 
+    conn = rx_NewConnection(addr, htons(port), RX_SERVER_ID, secureobj, secureindex);
+    if (conn == NULL)
+       errx(1, "failed to contact server");
+
+#ifdef AFS_PTHREAD_ENV
+    pthread_attr_init(&tattr);
+    pthread_attr_setdetachstate(&tattr, PTHREAD_CREATE_JOINABLE);
+#endif
+
+    params->conn = conn;
+    params->filename = filename;
+    params->command = command;
+    params->times = times;
+    params->bytes = bytes;
+    params->sendbytes = sendbytes;
+    params->readbytes = readbytes;
+
     start_timer();
 
 #ifdef AFS_PTHREAD_ENV
-    for ( i=0; i<threads; i++)
+    for ( i=0; i<threads; i++) {
         pthread_create(&thread[i], &tattr, client_thread, params);
+        if ( (i + 1) % RX_MAXCALLS == 0 ) {
+            conn = rx_NewConnection(addr, htons(port), RX_SERVER_ID, secureobj, secureindex);
+            if (conn != NULL) {
+                struct client_data *new_params = malloc(sizeof(struct client_data));
+                memcpy(new_params, params, sizeof(struct client_data));
+                new_params->conn = conn;
+                params = new_params;
+            }
+        }
+    }
 #else
         client_thread(params);
 #endif
@@ -928,7 +950,7 @@ rxperf_server(int argc, char **argv)
     char *ptr;
     int ch;
 
-    while ((ch = getopt(argc, argv, "r:d:p:P:w:W:HNjm:u:4:s:S")) != -1) {
+    while ((ch = getopt(argc, argv, "r:d:p:P:w:W:HNjm:u:4:s:SV")) != -1) {
        switch (ch) {
        case 'd':
 #ifdef RXDEBUG
@@ -994,6 +1016,9 @@ rxperf_server(int argc, char **argv)
            if (ptr && *ptr != '\0')
                errx(1, "can't resolve upd buffer size (Kbytes)");
            break;
+       case 'V':
+           use_rx_readv = 1;
+           break;
        case 'W':
            maxwsize = strtol(optarg, &ptr, 0);
            if (ptr && *ptr != '\0')
@@ -1045,7 +1070,7 @@ rxperf_client(int argc, char **argv)
 
     cmd = RX_PERF_UNKNOWN;
 
-    while ((ch = getopt(argc, argv, "T:S:R:b:c:d:p:P:r:s:w:W:f:HDNjm:u:4:t:")) != -1) {
+    while ((ch = getopt(argc, argv, "T:S:R:b:c:d:p:P:r:s:w:W:f:HDNjm:u:4:t:V")) != -1) {
        switch (ch) {
        case 'b':
            bytes = strtol(optarg, &ptr, 0);
@@ -1096,6 +1121,9 @@ rxperf_client(int argc, char **argv)
            if (host == NULL)
                err(1, "strdup");
            break;
+       case 'V':
+           use_rx_readv = 1;
+           break;
        case 'w':
            rxwrite_size = strtol(optarg, &ptr, 0);
            if (ptr != 0 && ptr[0] != '\0')