benchmarks: sftp_async + few changes

tests/benchmarks/bench_raw.c

@@ -47,7 +47,7 @@ const char python_eater[]=
 static char *get_python_eater(unsigned long bytes){
   char *eater=malloc(sizeof(python_eater));
   char *ptr;
-  char buffer[12];
+  char buf[12];
 
   memcpy(eater,python_eater,sizeof(python_eater));
   ptr=strstr(eater,"XXXXXXXXXX");
@@ -55,8 +55,8 @@ static char *get_python_eater(unsigned long bytes){
     free(eater);
     return NULL;
   }
-  sprintf(buffer,"0x%.8lx",bytes);
-  memcpy(ptr,buffer,10);
+  sprintf(buf,"0x%.8lx",bytes);
+  memcpy(ptr,buf,10);
   return eater;
 }
 
@@ -106,18 +106,16 @@ error:
  */
 int benchmarks_raw_up (ssh_session session, struct argument_s *args,
     float *bps){
-  unsigned long bytes=0x1000000;
+  unsigned long bytes;
   char *script;
   char cmd[128];
-  static char buffer[0x10000];
   int err;
   ssh_channel channel;
   struct timestamp_struct ts;
   float ms=0.0;
   unsigned long total=0;
 
-  if(args->data != 0)
-    bytes = args->data * 1024 * 1024;
+  bytes = args->datasize * 1024 * 1024;
   script =get_python_eater(bytes);
   err=upload_script(session,"/tmp/eater.py",script);
   free(script);
@@ -146,8 +144,8 @@ int benchmarks_raw_up (ssh_session session, struct argument_s *args,
   while(total < bytes){
     unsigned long towrite = bytes - total;
     int w;
-    if(towrite > 0x10000)
-      towrite = 32758;
+    if(towrite > args->chunksize)
+      towrite = args->chunksize;
     w=ssh_channel_write(channel,buffer,towrite);
     if(w == SSH_ERROR)
       goto error;
@@ -205,7 +203,7 @@ const char python_giver[] =
 static char *get_python_giver(unsigned long bytes){
   char *giver=malloc(sizeof(python_giver));
   char *ptr;
-  char buffer[12];
+  char buf[12];
 
   memcpy(giver,python_giver,sizeof(python_giver));
   ptr=strstr(giver,"XXXXXXXXXX");
@@ -213,8 +211,8 @@ static char *get_python_giver(unsigned long bytes){
     free(giver);
     return NULL;
   }
-  sprintf(buffer,"0x%.8lx",bytes);
-  memcpy(ptr,buffer,10);
+  sprintf(buf,"0x%.8lx",bytes);
+  memcpy(ptr,buf,10);
   return giver;
 }
 
@@ -228,18 +226,16 @@ static char *get_python_giver(unsigned long bytes){
  */
 int benchmarks_raw_down (ssh_session session, struct argument_s *args,
     float *bps){
-  unsigned long bytes=0x1000000;
+  unsigned long bytes;
   char *script;
   char cmd[128];
-  static char buffer[0x10000];
   int err;
   ssh_channel channel;
   struct timestamp_struct ts;
   float ms=0.0;
   unsigned long total=0;
 
-  if(args->data != 0)
-    bytes = args->data * 1024 * 1024;
+  bytes = args->datasize * 1024 * 1024;
   script =get_python_giver(bytes);
   err=upload_script(session,"/tmp/giver.py",script);
   free(script);
@@ -261,8 +257,8 @@ int benchmarks_raw_down (ssh_session session, struct argument_s *args,
   while(total < bytes){
     unsigned long toread = bytes - total;
     int r;
-    if(toread > sizeof(buffer))
-      toread = sizeof(buffer);
+    if(toread > args->chunksize)
+      toread = args->chunksize;
     r=ssh_channel_read(channel,buffer,toread,0);
     if(r == SSH_ERROR)
       goto error;

tests/benchmarks/bench_scp.c

@@ -37,15 +37,13 @@
  */
 int benchmarks_scp_up (ssh_session session, struct argument_s *args,
     float *bps){
-  unsigned long bytes=0x1000000;
-  static char buffer[0x10000];
+  unsigned long bytes;
   struct timestamp_struct ts;
   float ms=0.0;
   unsigned long total=0;
   ssh_scp scp;
 
-  if(args->data != 0)
-    bytes = args->data * 1024 * 1024;
+  bytes = args->datasize * 1024 * 1024;
   scp = ssh_scp_new(session,SSH_SCP_WRITE,SCPDIR);
   if(scp == NULL)
     goto error;
@@ -59,8 +57,8 @@ int benchmarks_scp_up (ssh_session session, struct argument_s *args,
   while(total < bytes){
     unsigned long towrite = bytes - total;
     int w;
-    if(towrite > 32758)
-      towrite = 32758;
+    if(towrite > args->chunksize)
+      towrite = args->chunksize;
     w=ssh_scp_write(scp,buffer,towrite);
     if(w == SSH_ERROR)
       goto error;
@@ -93,8 +91,7 @@ error:
  */
 int benchmarks_scp_down (ssh_session session, struct argument_s *args,
     float *bps){
-  unsigned long bytes=0x1000000;
-  static char buffer[0x10000];
+  unsigned long bytes;
   struct timestamp_struct ts;
   float ms=0.0;
   unsigned long total=0;
@@ -102,8 +99,7 @@ int benchmarks_scp_down (ssh_session session, struct argument_s *args,
   int r;
   size_t size;
 
-  if(args->data != 0)
-    bytes = args->data * 1024 * 1024;
+  bytes = args->datasize * 1024 * 1024;
   scp = ssh_scp_new(session,SSH_SCP_READ,SCPDIR SCPFILE);
   if(scp == NULL)
     goto error;
@@ -125,8 +121,8 @@ int benchmarks_scp_down (ssh_session session, struct argument_s *args,
     ssh_scp_accept_request(scp);
     while(total < bytes){
       unsigned long toread = bytes - total;
-      if(toread > sizeof(buffer))
-        toread = sizeof(buffer);
+      if(toread > args->chunksize)
+        toread = args->chunksize;
       r=ssh_scp_read(scp,buffer,toread);
       if(r == SSH_ERROR || r == 0)
         goto error;

tests/benchmarks/bench_sftp.c

@@ -25,6 +25,7 @@
 #include <libssh/sftp.h>
 #include <stdio.h>
 #include <fcntl.h>
+#include <stdlib.h>
 
 #define SFTPDIR "/tmp/"
 #define SFTPFILE "scpbenchmark"
@@ -39,16 +40,14 @@
  */
 int benchmarks_sync_sftp_up (ssh_session session, struct argument_s *args,
     float *bps){
-  unsigned long bytes=0x1000000;
-  static char buffer[0x10000];
+  unsigned long bytes;
   struct timestamp_struct ts;
   float ms=0.0;
   unsigned long total=0;
   sftp_session sftp;
   sftp_file file;
 
-  if(args->data != 0)
-    bytes = args->data * 1024 * 1024;
+  bytes = args->datasize * 1024 * 1024;
   sftp = sftp_new(session);
   if(sftp == NULL)
     goto error;
@@ -63,8 +62,8 @@ int benchmarks_sync_sftp_up (ssh_session session, struct argument_s *args,
   while(total < bytes){
     unsigned long towrite = bytes - total;
     int w;
-    if(towrite > 32758)
-      towrite = 32758;
+    if(towrite > args->chunksize)
+      towrite = args->chunksize;
     w=sftp_write(file,buffer,towrite);
     if(w == SSH_ERROR)
       goto error;
@@ -97,8 +96,7 @@ error:
  */
 int benchmarks_sync_sftp_down (ssh_session session, struct argument_s *args,
     float *bps){
-  unsigned long bytes=0x1000000;
-  static char buffer[0x10000];
+  unsigned long bytes;
   struct timestamp_struct ts;
   float ms=0.0;
   unsigned long total=0;
@@ -106,8 +104,7 @@ int benchmarks_sync_sftp_down (ssh_session session, struct argument_s *args,
   sftp_file file;
   int r;
 
-  if(args->data != 0)
-    bytes = args->data * 1024 * 1024;
+  bytes = args->datasize * 1024 * 1024;
   sftp = sftp_new(session);
   if(sftp == NULL)
     goto error;
@@ -121,8 +118,8 @@ int benchmarks_sync_sftp_down (ssh_session session, struct argument_s *args,
   timestamp_init(&ts);
   while(total < bytes){
     unsigned long toread = bytes - total;
-    if(toread > sizeof(buffer))
-      toread = sizeof(buffer);
+    if(toread > args->chunksize)
+      toread = args->chunksize;
     r=sftp_read(file,buffer,toread);
     if(r == SSH_ERROR)
       goto error;
@@ -150,3 +147,93 @@ error:
     sftp_free(sftp);
   return -1;
 }
+
+/** @internal
+ * @brief benchmarks an asynchronous sftp download using an
+ * existing SSH session.
+ * @param[in] session Open SSH session
+ * @param[in] args Parsed command line arguments
+ * @param[out] bps The calculated bytes per second obtained via benchmark.
+ * @return 0 on success, -1 on error.
+ */
+int benchmarks_async_sftp_down (ssh_session session, struct argument_s *args,
+    float *bps){
+  unsigned long bytes;
+  struct timestamp_struct ts;
+  float ms=0.0;
+  unsigned long total=0;
+  sftp_session sftp;
+  sftp_file file;
+  int r,i;
+  int warned = 0;
+  unsigned long toread;
+  int *ids=NULL;
+  int concurrent_downloads = args->concurrent_requests;
+
+  bytes = args->datasize * 1024 * 1024;
+  sftp = sftp_new(session);
+  if(sftp == NULL)
+    goto error;
+  if(sftp_init(sftp)==SSH_ERROR)
+    goto error;
+  file = sftp_open(sftp,SFTPDIR SFTPFILE,O_RDONLY,0);
+  if(!file)
+    goto error;
+  ids = malloc(concurrent_downloads * sizeof(int));
+  if(args->verbose>0)
+    fprintf(stdout,"Starting download of %lu bytes now, using %d concurrent downloads\n",bytes,
+        concurrent_downloads);
+  timestamp_init(&ts);
+  for (i=0;i<concurrent_downloads;++i){
+    ids[i]=sftp_async_read_begin(file, args->chunksize);
+    if(ids[i]==SSH_ERROR)
+        goto error;
+  }
+  i=0;
+  while(total < bytes){
+    r = sftp_async_read(file, buffer, args->chunksize, ids[i]);
+    if(r == SSH_ERROR)
+      goto error;
+    total += r;
+    if(r != (int)args->chunksize && total != bytes && !warned){
+      fprintf(stderr,"async_sftp_download : receiving short reads (%d, requested %d) "
+          "the received file will be corrupted and shorted. Adapt chunksize to %d\n",
+          r, args->chunksize,r);
+      warned = 1;
+    }
+    /* we had a smaller file */
+    if(r==0){
+      fprintf(stdout,"File smaller than expected : %lu (expected %lu).\n",total,bytes);
+      bytes = total;
+      break;
+    }
+    toread = bytes - total;
+    if(toread < args->chunksize * concurrent_downloads){
+      /* we've got enough launched downloads */
+      ids[i]=-1;
+    }
+    if(toread > args->chunksize)
+      toread = args->chunksize;
+    ids[i]=sftp_async_read_begin(file,toread);
+    if(ids[i] == SSH_ERROR)
+      goto error;
+    i = (i+1) % concurrent_downloads;
+  }
+  sftp_close(file);
+  ms=elapsed_time(&ts);
+  *bps=8000 * (float)bytes / ms;
+  if(args->verbose > 0)
+    fprintf(stdout,"download took %f ms for %lu bytes, at %f bps\n",ms,
+        bytes,*bps);
+  sftp_free(sftp);
+  free(ids);
+  return 0;
+error:
+  fprintf(stderr,"Error during sftp download : %s\n",ssh_get_error(session));
+  if(file)
+    sftp_close(file);
+  if(sftp)
+    sftp_free(sftp);
+  free(ids);
+  return -1;
+}

tests/benchmarks/benchmarks.c

@@ -57,6 +57,11 @@ struct benchmark benchmarks[]= {
         .name="benchmark_sync_sftp_download",
         .fct=benchmarks_sync_sftp_down,
         .enabled=0
+    },
+    {
+        .name="benchmark_async_sftp_download",
+        .fct=benchmarks_async_sftp_down,
+        .enabled=0
     }
 };
 
@@ -128,7 +133,16 @@ static struct argp_option options[] = {
     .key   = '6',
     .arg   = NULL,
     .flags = 0,
-    .doc   = "Download data using synchronous SFTP",
+    .doc   = "Download data using synchronous SFTP (slow)",
+    .group = 0
+
+  },
+  {
+    .name  = "async-sftp-download",
+    .key   = '7',
+    .arg   = NULL,
+    .flags = 0,
+    .doc   = "Download data using asynchronous SFTP (fast)",
     .group = 0
 
   },
@@ -141,13 +155,29 @@ static struct argp_option options[] = {
     .group = 0
   },
   {
-    .name  = "data",
-    .key   = 'd',
+    .name  = "size",
+    .key   = 's',
     .arg   = "MBYTES",
     .flags = 0,
     .doc   = "MBytes of data to send/receive per test",
     .group = 0
   },
+  {
+    .name  = "chunk",
+    .key   = 'c',
+    .arg   = "bytes",
+    .flags = 0,
+    .doc   = "size of data chunks to send/receive",
+    .group = 0
+  },
+  {
+    .name  = "prequests",
+    .key   = 'p',
+    .arg   = "number [20]",
+    .flags = 0,
+    .doc   = "[async SFTP] number of concurrent requests",
+    .group = 0
+  },
   {NULL, 0, NULL, 0, NULL, 0}
 };
 
@@ -168,14 +198,21 @@ static error_t parse_opt (int key, char *arg, struct argp_state *state) {
     case '4':
     case '5':
     case '6':
+    case '7':
       benchmarks[key - '1'].enabled = 1;
       arguments->ntests ++;
       break;
     case 'v':
       arguments->verbose++;
       break;
-    case 'd':
-      arguments->data = atoi(arg);
+    case 's':
+      arguments->datasize = atoi(arg);
+      break;
+    case 'p':
+      arguments->concurrent_requests = atoi(arg);
+      break;
+    case 'c':
+      arguments->chunksize = atoi(arg);
       break;
     case 'h':
       if(arguments->nhosts >= MAX_HOSTS_CONNECT){
@@ -218,6 +255,9 @@ static void cmdline_parse(int argc, char **argv, struct argument_s *arguments) {
 
 static void arguments_init(struct argument_s *arguments){
   memset(arguments,0,sizeof(*arguments));
+  arguments->chunksize=32758;
+  arguments->concurrent_requests=20;
+  arguments->datasize = 10;
 }
 
 static ssh_session connect_host(const char *host, int verbose){
@@ -239,19 +279,19 @@ error:
 }
 
 static char *network_speed(float bps){
-  static char buffer[128];
+  static char buf[128];
   if(bps > 1000*1000*1000){
     /* Gbps */
-    snprintf(buffer,sizeof(buffer),"%f Gbps",bps/(1000*1000*1000));
+    snprintf(buf,sizeof(buf),"%f Gbps",bps/(1000*1000*1000));
   } else if(bps > 1000*1000){
     /* Mbps */
-    snprintf(buffer,sizeof(buffer),"%f Mbps",bps/(1000*1000));
+    snprintf(buf,sizeof(buf),"%f Mbps",bps/(1000*1000));
   } else if(bps > 1000){
-    snprintf(buffer,sizeof(buffer),"%f Kbps",bps/1000);
+    snprintf(buf,sizeof(buf),"%f Kbps",bps/1000);
   } else {
-    snprintf(buffer,sizeof(buffer),"%f bps",bps);
+    snprintf(buf,sizeof(buf),"%f bps",bps);
   }
-  return buffer;
+  return buf;
 }
 
 static void do_benchmarks(ssh_session session, struct argument_s *arguments,
@@ -284,6 +324,8 @@ static void do_benchmarks(ssh_session session, struct argument_s *arguments,
   }
 }
 
+char *buffer;
+
 int main(int argc, char **argv){
   struct argument_s arguments;
   ssh_session session;
@@ -301,6 +343,11 @@ int main(int argc, char **argv){
     }
     arguments.ntests=BENCHMARK_NUMBER;
   }
+  buffer=malloc(arguments.chunksize);
+  if(buffer == NULL){
+    fprintf(stderr,"Allocation of chunk buffer failed\n");
+    return EXIT_FAILURE;
+  }
   if (arguments.verbose > 0){
     fprintf(stdout, "Will try hosts ");
     for(i=0;i<arguments.nhosts;++i){

tests/benchmarks/benchmarks.h

@@ -36,6 +36,7 @@ enum libssh_benchmarks {
     BENCHMARK_SCP_DOWNLOAD,
     BENCHMARK_SYNC_SFTP_UPLOAD,
     BENCHMARK_SYNC_SFTP_DOWNLOAD,
+    BENCHMARK_ASYNC_SFTP_DOWNLOAD,
     BENCHMARK_NUMBER
 };
 
@@ -44,9 +45,13 @@ struct argument_s {
   int verbose;
   int nhosts;
   int ntests;
-  int data;
+  unsigned int datasize;
+  unsigned int chunksize;
+  int concurrent_requests;
 };
 
+extern char *buffer;
+
 typedef int (*bench_fct)(ssh_session session, struct argument_s *args,
     float *bps);
 
@@ -88,4 +93,6 @@ int benchmarks_sync_sftp_up (ssh_session session, struct argument_s *args,
     float *bps);
 int benchmarks_sync_sftp_down (ssh_session session, struct argument_s *args,
     float *bps);
+int benchmarks_async_sftp_down (ssh_session session, struct argument_s *args,
+    float *bps);
 #endif /* BENCHMARKS_H_ */