implemented concurrency in create_write_persist

include/bake-server.h

@@ -120,22 +120,39 @@ int bake_provider_list_storage_targets(
         bake_target_id_t* targets);
 
 /**
- * @brief Sets the size of the intermediate buffer used for transfering data.
+ * @brief Sets the size and number of intermediate buffers used for transfering data.
  * The size is set to 0 by default. A size of 0 indicates that RDMA will be
  * done all at once and target the backend device directly without using an
  * intermediate buffer.
  *
  * @param provider Bake provider
  * @param target_id Target for which to change the buffer size.
+ * @param count Number of buffers to initialize.
  * @param size Size of the buffer.
  *
  * @return 0 on success, -1 on failure
  */
-int bake_provider_set_target_xfer_buffer_size(
+int bake_provider_set_target_xfer_buffer(
         bake_provider_t provider,
         bake_target_id_t target_id,
+        size_t count,
         size_t size);
 
+/**
+ * @brief Sets the maximum number of ULTs that will be used to concurrently
+ * transfer data.
+ *
+ * @param provider Bake provider
+ * @param target_id Target for which to change the number of ULTs
+ * @param num_threads Number of ULTs
+ *
+ * @return 0 on success, -1 on failure
+ */
+int bake_provider_set_target_xfer_concurrency(
+        bake_provider_t provider,
+        bake_target_id_t target_id,
+        uint32_t num_threads);
+
 #ifdef __cplusplus
 }
 #endif

src/bake-server-daemon.c

@@ -25,6 +25,8 @@ struct options
     char **bake_pools;
     char *host_file;
     size_t buf_size;
+    size_t buf_count;
+    uint32_t num_threads;
     mplex_mode_t mplex_mode;
 };
 
@@ -36,6 +38,8 @@ static void usage(int argc, char **argv)
     fprintf(stderr, "       [-f filename] to write the server address to a file\n");
     fprintf(stderr, "       [-m mode] multiplexing mode (providers or targets) for managing multiple pools (default is targets)\n");
     fprintf(stderr, "       [-b size] buffer size for writes on provider\n");
+    fprintf(stderr, "       [-c count] count of buffers used for accesses on provider\n");
+    fprintf(stderr, "       [-t threads] number of threads used for concurrency\n");
     fprintf(stderr, "Example: ./bake-server-daemon tcp://localhost:1234 /dev/shm/foo.dat /dev/shm/bar.dat\n");
     return;
 }
@@ -47,7 +51,7 @@ static void parse_args(int argc, char **argv, struct options *opts)
     memset(opts, 0, sizeof(*opts));
 
     /* get options */
-    while((opt = getopt(argc, argv, "f:m:b:")) != -1)
+    while((opt = getopt(argc, argv, "f:m:b:t:c:")) != -1)
     {
         switch(opt)
         {
@@ -67,6 +71,12 @@ static void parse_args(int argc, char **argv, struct options *opts)
             case 'b':
                 opts->buf_size = atol(optarg);
                 break;
+            case 'c':
+                opts->buf_count = atol(optarg);
+                break;
+            case 't':
+                opts->num_threads = atol(optarg);
+                break;
             default:
                 usage(argc, argv);
                 exit(EXIT_FAILURE);
@@ -174,7 +184,8 @@ int main(int argc, char **argv)
                 return(-1);
             }
 
-            bake_provider_set_target_xfer_buffer_size(provider, tid, opts.buf_size);
+            bake_provider_set_target_xfer_buffer(provider, tid, opts.buf_count, opts.buf_size);
+            bake_provider_set_target_xfer_concurrency(provider, tid, opts.num_threads);
 
             printf("Provider %d managing new target at multiplex id %d\n", i, i+1);
         }
@@ -205,7 +216,8 @@ int main(int argc, char **argv)
                 return(-1);
             }
 
-            bake_provider_set_target_xfer_buffer_size(provider, tid, opts.buf_size);
+            bake_provider_set_target_xfer_buffer(provider, tid, opts.buf_count, opts.buf_size);
+            bake_provider_set_target_xfer_concurrency(provider, tid, opts.num_threads);
 
             printf("Provider 0 managing new target at multiplex id %d\n", 1);
         }