Increase no. of injection threads to 2

Test

swiftmpistepsim.c

@@ -38,6 +38,9 @@ static int verbose = 0;
 /* Attempt to keep original injection time differences. */
 static int usetics = 1;
 
+/* The wait between injections, nanosecs. */
+static long long waitns = 0;
+
 /* Integer types of send and recv tasks, must match log. */
 static const int task_type_send = 22;
 static const int task_type_recv = 23;
@@ -47,14 +50,16 @@ static const int task_subtype_count = 30;  // Just some upper limit on subtype.
 static MPI_Comm subtypeMPI_comms[30];
 
 /* The local queues. */
-static struct mpiuse_log_entry **volatile reqs_queue;
-static int volatile ind_req = 0;
-static int volatile nr_reqs = 0;
-static int volatile injecting = 1;
+static struct mpiuse_log_entry **volatile reqs_queue[2];
+static int volatile ind_req[2] = {0,0};
+static int volatile nr_reqs[2] = {0,0};
+static int volatile injecting[2] = {1,1};
+
 static struct mpiuse_log_entry **volatile recvs_queue;
 static int volatile nr_recvs = 0;
 static int volatile ind_recv = 0;
 static int volatile todo_recv = 0;
+
 static struct mpiuse_log_entry **volatile sends_queue;
 static int volatile nr_sends = 0;
 static int volatile ind_send = 0;
@@ -64,27 +69,23 @@ static int volatile todo_send = 0;
 // XXX need to store this in the data file.
 static double log_clocks_cpufreq = 2194844448.0;
 
-/**
- * @brief Injection thread, initiates MPI_Isend and MPI_Irecv requests.
- *
- * The requests are initiated in the time order of the original log and an
- * attempt to start these with the same interval gap is made if usetics is
- * set, otherwise we just do them as quickly as possible.
- */
-static void *inject_thread(void *arg) {
-
-  if (verbose) message("%d: injection thread starts", *((int *)arg));
+static void injection_runner(int qid) {
+  if (verbose) message("%d: injection thread starts", qid);
   ticks starttics = getticks();
+  struct mpiuse_log_entry **reqs = reqs_queue[qid];
 
   /* Ticks of our last attempt and ticks the first loop takes (usetics == 1). */
-  ticks basetic = reqs_queue[0]->tic;
+  ticks basetic = reqs[0]->tic;
   ticks looptics = 0;
   double deadtime = 0.0;
 
-  while (ind_req < nr_reqs) {
-    struct mpiuse_log_entry *log = reqs_queue[ind_req];
+  struct timespec sleep;
+  sleep.tv_sec = 0;
 
-    if (usetics) {
+  while (ind_req[qid] < nr_reqs[qid]) {
+    struct mpiuse_log_entry *log = reqs[ind_req[qid]];
+
+    if (usetics && waitns == 0) {
       /* Expect time between this request and the previous one. */
       ticks dt = log->tic - basetic;
       basetic = log->tic;
@@ -113,6 +114,10 @@ static void *inject_thread(void *arg) {
         nanosleep(&sleep, NULL);
         deadtime += sleep.tv_nsec;
       }
+    } else if (waitns != 0) {
+      sleep.tv_nsec = waitns;
+      nanosleep(&sleep, NULL);
+      deadtime += sleep.tv_nsec;
     }
 
     /* Initialise new log elements. */
@@ -146,12 +151,12 @@ static void *inject_thread(void *arg) {
     }
     if (err != MPI_SUCCESS) error("Failed to activate send or recv");
 
-    ind_req++;
+    ind_req[qid]++;
 
     /* Set looptics on the first pass. Assumes MPI_Isend and MPI_Irecv are
      * equally timed. */
     if (looptics == 0 && usetics) {
-      looptics = getticks() - starttics;
+        looptics = getticks() - starttics;
       if (verbose)
         message("injection loop took %.3f %s.", clocks_from_ticks(looptics),
                 clocks_getunit());
@@ -160,16 +165,39 @@ static void *inject_thread(void *arg) {
 
   /* All done, thread exiting. */
   if (verbose) {
-    message("%d injections completed, sends = %d, recvs = %d", ind_req,
+    message("%d injections completed, sends = %d, recvs = %d", ind_req[qid],
             nr_sends, nr_recvs);
     message("remaining sends = %d, recvs = %d", todo_send, todo_recv);
     if (usetics) message("deadtime %.3f ms", deadtime / 1.0e6);
   }
   message("took %.3f %s.", clocks_from_ticks(getticks() - starttics),
           clocks_getunit());
-  atomic_dec(&injecting);
+  atomic_dec(&injecting[qid]);
+}
+
+/**
+ * @brief Injection thread, initiates MPI_Isend and MPI_Irecv requests.
+ *
+ * The requests are initiated in the time order of the original log and an
+ * attempt to start these with the same interval gap is made if usetics is
+ * set, otherwise we just do them as quickly as possible.
+ */
+static void *inject_thread1(void *arg) {
+  injection_runner(0);
   return NULL;
 }
+/**
+ * @brief Injection thread, initiates MPI_Isend and MPI_Irecv requests.
+ *
+ * The requests are initiated in the time order of the original log and an
+ * attempt to start these with the same interval gap is made if usetics is
+ * set, otherwise we just do them as quickly as possible.
+ */
+static void *inject_thread2(void *arg) {
+  injection_runner(1);
+  return NULL;
+}
+
 
 /**
  * @brief main loop to run over a queue of MPI requests and test for when they
@@ -198,7 +226,7 @@ static void queue_runner(struct mpiuse_log_entry **logs, int volatile *nr_logs,
 
   /* We loop while new requests are being injected and we still have requests
    * to complete. */
-  while (injecting || (!injecting && *todos > 0)) {
+  while ((injecting[0] || injecting[1]) || (!injecting[0] && !injecting[1] && *todos > 0)) {
     int nlogs = *nr_logs;
     for (int k = 0; k < nlogs; k++) {
       struct mpiuse_log_entry *log = logs[k];
@@ -248,7 +276,7 @@ static void queue_runner(struct mpiuse_log_entry **logs, int volatile *nr_logs,
  */
 static void *send_thread(void *arg) {
 
-  if (verbose) message("%d: send thread starts (%d)", *((int *)arg), injecting);
+  if (verbose) message("%d: send thread starts", *((int *)arg));
   ticks starttics = getticks();
 
   int ncalls;
@@ -319,9 +347,8 @@ static void pick_logs(void) {
   size_t nlogs = mpiuse_nr_logs();
 
   /* Duplicate of logs. */
-  reqs_queue = (struct mpiuse_log_entry **)malloc(
-      sizeof(struct mpiuse_log_entry *) * nlogs);
-  nr_reqs = 0;
+  struct mpiuse_log_entry **reqs = (struct mpiuse_log_entry **)malloc(sizeof(struct mpiuse_log_entry *) * nlogs);
+  int nreqs = 0;
   sends_queue = (struct mpiuse_log_entry **)malloc(
       sizeof(struct mpiuse_log_entry *) * nlogs);
   nr_sends = 0;
@@ -338,8 +365,8 @@ static void pick_logs(void) {
         log->data = calloc(log->size, 1);
 
         /* And keep this log. */
-        reqs_queue[nr_reqs] = log;
-        nr_reqs++;
+        reqs[nreqs] = log;
+        nreqs++;
 
       } else {
         error("task type '%d' is not a known send or recv task", log->type);
@@ -348,16 +375,36 @@ static void pick_logs(void) {
   }
 
   /* Sort into increasing time. */
-  qsort(reqs_queue, nr_reqs, sizeof(struct mpiuse_log_entry *), cmp_logs);
+  qsort(reqs, nreqs, sizeof(struct mpiuse_log_entry *), cmp_logs);
+
+  /* And partition into queues for injection. Use interleave pick so that
+   * close in time injects are on different queues. */
+  reqs_queue[0] = (struct mpiuse_log_entry **)malloc(
+      sizeof(struct mpiuse_log_entry *) * nlogs);
+  nr_reqs[0] = 0;
+  reqs_queue[1] = (struct mpiuse_log_entry **)malloc(
+      sizeof(struct mpiuse_log_entry *) * nlogs);
+  nr_reqs[1] = 0;
+
+  for (int k = 0; k < nreqs; k++) {
+    int qid = k % 2;
+    reqs_queue[qid][nr_reqs[qid]] = reqs[k];
+    nr_reqs[qid]++;
+  }
+
+  message("nr_reqs = %d, %d", nr_reqs[0], nr_reqs[1]);
+
+  free(reqs);
 }
 
 /**
  * @brief usage help.
  */
 static void usage(char *argv[]) {
-  fprintf(stderr, "Usage: %s [-vf] SWIFT_mpiuse-log-file.dat logfile.dat\n",
-          argv[0]);
-  fprintf(stderr, " options: -v verbose, -f fast injections\n");
+  fprintf(stderr, "Usage: %s [-vf] [-n nanosec] SWIFT_mpiuse-log-file.dat "
+          "logfile.dat\n", argv[0]);
+  fprintf(stderr, " options: -v verbose, -f fast injections, "
+          "-n nanoseconds to wait\n");
   fflush(stderr);
 }
 
@@ -383,7 +430,7 @@ int main(int argc, char *argv[]) {
   /* Handle the command-line, we expect a mpiuse data file to read and various
    * options. */
   int opt;
-  while ((opt = getopt(argc, argv, "vf")) != -1) {
+  while ((opt = getopt(argc, argv, "vfn:")) != -1) {
     switch (opt) {
       case 'f':
         usetics = 0;
@@ -391,6 +438,9 @@ int main(int argc, char *argv[]) {
       case 'v':
         verbose = 1;
         break;
+     case 'n':
+        waitns = atoll(optarg);
+        break;
       default:
         if (myrank == 0) usage(argv);
         return 1;
@@ -428,12 +478,16 @@ int main(int argc, char *argv[]) {
   if (myrank == 0) {
     message("Start of MPI tests");
     message("==================");
+    if (waitns > 0) message("adding fixed waits of %lld ns", waitns);
   }
 
   /* Make three threads, one for injecting tasks and two to check for
    * completions of the sends and recv independently. */
-  pthread_t injectthread;
-  if (pthread_create(&injectthread, NULL, &inject_thread, &myrank) != 0)
+  pthread_t injectthread1;
+  if (pthread_create(&injectthread1, NULL, &inject_thread1, &myrank) != 0)
+    error("Failed to create injection thread.");
+  pthread_t injectthread2;
+  if (pthread_create(&injectthread2, NULL, &inject_thread2, &myrank) != 0)
     error("Failed to create injection thread.");
   pthread_t sendthread;
   if (pthread_create(&sendthread, NULL, &send_thread, &myrank) != 0)
@@ -443,7 +497,8 @@ int main(int argc, char *argv[]) {
     error("Failed to create recv thread.");
 
   /* Wait until all threads have exited and all MPI requests have completed. */
-  pthread_join(injectthread, NULL);
+  pthread_join(injectthread1, NULL);
+  pthread_join(injectthread2, NULL);
   pthread_join(sendthread, NULL);
   pthread_join(recvthread, NULL);