rados: split up function into more subfunctions

1 files changed, 159 insertions, 142 deletions

diff --git a/src/rados_bencher.h b/src/rados_bencher.h
index 5075e9e158b..285e42c9931 100644
--- a/src/rados_bencher.h
+++ b/src/rados_bencher.h
@@ -24,136 +24,124 @@
 Mutex dataLock("data mutex");
 
 struct bench_data {
-  bool done;
-  int object_size;
-  int trans_size;
-  int in_flight;
+  bool done; //is the benchmark is done
+  int object_size; //the size of the objects
+  int trans_size; //size of the write/read to perform
+  // same as object_size for write tests
+  int in_flight; //number of reads/writes being waited on
   int started;
   int finished;
   double min_latency;
   double max_latency;
   double avg_latency;
-  utime_t cur_latency;
-  utime_t startTime;
+  utime_t cur_latency; //latency of last completed transaction
+  utime_t startTime; //start time for benchmark
+  char *iTime; //identifier time char[] that the object names are marked with
+  char *object_contents; //pointer to the contents written to each object
 };
 
-static void *status_printer(void * data_store) {
-  bench_data *data = (bench_data *) data_store;
-  Cond cond;
-  int i = 0;
-  int previous_writes = 0;
-  int cycleSinceChange = 0;
-  double avg_bandwidth;
-  double bandwidth;
-  utime_t ONE_SECOND;
-  ONE_SECOND.set_from_double(1.0);
-  dataLock.Lock();
-  while(!data->done) {
-    if (i % 20 == 0) {
-      if (i > 0)
-	cout << "min lat: " << data->min_latency
-	     << " max lat: " << data->max_latency
-	     << " avg lat: " << data->avg_latency << std::endl;
-      //I'm naughty and don't reset the fill
-      cout << setfill(' ') 
-	   << setw(5) << "sec" 
-	   << setw(8) << "Cur ops"
-	   << setw(10) << "started"
-	   << setw(10) << "finished"
-	   << setw(10) << "avg MB/s"
-	   << setw(10) << "cur MB/s"
-	   << setw(10) << "last lat"
-	   << setw(10) << "avg lat" << std::endl;
-    }
-    bandwidth = (double)(data->finished - previous_writes)
-      * (data->trans_size)
-      / (1024*1024)
-      / cycleSinceChange;
-    avg_bandwidth = (double) (data->trans_size) * (data->finished)
-      / (double)(g_clock.now() - data->startTime) / (1024*1024);
-    if (previous_writes != data->finished) {
-      previous_writes = data->finished;
-      cycleSinceChange = 0;
-      cout << setfill(' ') 
-	   << setw(5) << i
-	   << setw(8) << data->in_flight
-	   << setw(10) << data->started
-	   << setw(10) << data->finished
-	   << setw(10) << avg_bandwidth
-	   << setw(10) << bandwidth
-	   << setw(10) << (double)data->cur_latency
-	   << setw(10) << data->avg_latency << std::endl;
-    }
-    else {
-      cout << setfill(' ')
-	   << setw(5) << i
-	   << setw(8) << data->in_flight
-	   << setw(10) << data->started
-	   << setw(10) << data->finished
-	   << setw(10) << avg_bandwidth
-	   << setw(10) << '0'
-	   << setw(10) << '-'
-	   << setw(10) << data->avg_latency << std::endl;
-    }
-    ++i;
-    ++cycleSinceChange;
-    cond.WaitInterval(dataLock, ONE_SECOND);
-  }
-  dataLock.Unlock();
-  return NULL;
-}
+void write_bench(Rados& rados, rados_pool_t pool,
+		 int secondsToRun, int concurrentios, bench_data *data);
+void *status_printer(void * data_store);
+
   
 int aio_bench(Rados& rados, rados_pool_t pool, int secondsToRun,
 	      int concurrentios, int writeSize, int readOffResults) {
-  
-  cout << "Maintaining " << concurrentios << " concurrent writes of "
-       << writeSize << " bytes for at least "
-       << secondsToRun << " seconds." << std::endl;
-  
+
+  char* contentsChars = new char[writeSize];  
   dataLock.Lock();
   bench_data *data = new bench_data();
   data->done = false;
-  data->trans_size = writeSize;
+  data->object_size = writeSize;
+  data->trans_size = writeSize; //just for now
   data->started = 0;
   data->finished = 0;
   data->min_latency = 9999.0; // this better be higher than initial latency!
   data->max_latency = 0;
   data->avg_latency = 0;
+  data->iTime = new char[100];
+  data->object_contents = contentsChars;
   dataLock.Unlock();
 
+
+  //fill in contentsChars deterministically so we can check returns
+  for (int i = 0; i < writeSize; ++i) {
+    contentsChars[i] = i % sizeof(char);
+  }
+  //set up the pool
+  cout << "open pool result = " << rados.open_pool("data",&pool) << " pool = " << pool << std::endl;
+  
+  write_bench(rados, pool, secondsToRun, concurrentios, data);
+
+  //check objects for consistency if requested
+  int errors = 0;
+  if (readOffResults) {
+    char matchName[128];
+    object_t oid;
+    bufferlist actualContents;
+    utime_t start_time;
+    utime_t lat;
+    double total_latency = 0;
+    double avg_latency;
+    double avg_bw;
+    for (int i = 0; i < data->finished; ++i ) {
+      snprintf(matchName, 128, "Object %d:%s", i, data->iTime);
+      oid = object_t(matchName);
+      snprintf(contentsChars, writeSize, "I'm the %dth object!", i);
+      start_time = g_clock.now();
+      rados.read(pool, oid, 0, actualContents, writeSize);
+      lat = g_clock.now() - start_time;
+      total_latency += (double) lat;
+      if (strcmp(contentsChars, actualContents.c_str()) != 0 ) {
+	cerr << "Object " << matchName << " is not correct!";
+	++errors;
+      }
+      actualContents.clear();
+    }
+    avg_latency = total_latency / data->finished;
+    avg_bw = data->finished * writeSize / (total_latency) / (1024 *1024);
+    cout << "read avg latency: " << avg_latency
+	 << " read avg bw: " << avg_bw << std::endl;
+  }
+
+  
+  if (readOffResults) {
+    if (errors) cout << "WARNING: There were " << errors << " total errors in copying!\n";
+    else cout << "No errors in copying!\n";
+  }
+  
+  delete contentsChars;
+  delete data;
+  return 0;
+}
+
+void write_bench(Rados& rados, rados_pool_t pool,
+		 int secondsToRun, int concurrentios, bench_data *data) {
+  cout << "Maintaining " << concurrentios << " concurrent writes of "
+       << data->object_size << " bytes for at least "
+       << secondsToRun << " seconds." << std::endl;
+  
   Rados::AioCompletion* completions[concurrentios];
   char* name[concurrentios];
   bufferlist* contents[concurrentios];
-  char* contentsChars = new char[writeSize];
   double totalLatency = 0;
   utime_t startTimes[concurrentios];
-  char bw[20];
   time_t initialTime;
   utime_t stopTime;
 
-  //fill in contentsChars deterministically so we can check returns
-  for (int i = 0; i < writeSize; ++i) {
-    contentsChars[i] = i % sizeof(char);
-  }
-  
   time(&initialTime);
   stringstream initialTimeS("");
   initialTimeS << initialTime;
-  char iTime[100];
-  strcpy(iTime, initialTimeS.str().c_str());
+  strcpy(data->iTime, initialTimeS.str().c_str());
   //set up writes so I can start them together
   for (int i = 0; i<concurrentios; ++i) {
     name[i] = new char[128];
     contents[i] = new bufferlist();
-    snprintf(name[i], 128, "Object %d:%s", i, iTime);
-    snprintf(contentsChars, writeSize, "I'm the %dth object!", i);
-    contents[i]->append(contentsChars, writeSize);
+    snprintf(name[i], 128, "Object %d:%s", i, data->iTime);
+    snprintf(data->object_contents, data->object_size, "I'm the %dth object!", i);
+    contents[i]->append(data->object_contents, data->object_size);
   }
-  
-  //set up the pool, get start time, and go!
-  cout << "open pool result = " << rados.open_pool("data",&pool) << " pool = " << pool << std::endl;
 
-  
   pthread_t print_thread;
   
   pthread_create(&print_thread, NULL, status_printer, (void *)data);
@@ -162,7 +150,7 @@ int aio_bench(Rados& rados, rados_pool_t pool, int secondsToRun,
   dataLock.Unlock();
   for (int i = 0; i<concurrentios; ++i) {
     startTimes[i] = g_clock.now();
-    rados.aio_write(pool, name[i], 0, *contents[i], writeSize, &completions[i]);
+    rados.aio_write(pool, name[i], 0, *contents[i], data->object_size, &completions[i]);
     dataLock.Lock();
     ++data->started;
     ++data->in_flight;
@@ -185,9 +173,9 @@ int aio_bench(Rados& rados, rados_pool_t pool, int secondsToRun,
     //create new contents and name on the heap, and fill them
     newContents = new bufferlist();
     newName = new char[128];
-    snprintf(newName, 128, "Object %d:%s", data->started, iTime);
-    snprintf(contentsChars, writeSize, "I'm the %dth object!", data->started);
-    newContents->append(contentsChars, writeSize);
+    snprintf(newName, 128, "Object %d:%s", data->started, data->iTime);
+    snprintf(data->object_contents, data->object_size, "I'm the %dth object!", data->started);
+    newContents->append(data->object_contents, data->object_size);
     completions[slot]->wait_for_safe();
     dataLock.Lock();
     data->cur_latency = g_clock.now() - startTimes[slot];
@@ -204,7 +192,7 @@ int aio_bench(Rados& rados, rados_pool_t pool, int secondsToRun,
     //write new stuff to rados, then delete old stuff
     //and save locations of new stuff for later deletion
     startTimes[slot] = g_clock.now();
-    rados.aio_write(pool, newName, 0, *newContents, writeSize, &completions[slot]);
+    rados.aio_write(pool, newName, 0, *newContents, data->object_size, &completions[slot]);
     dataLock.Lock();
     ++data->started;
     ++data->in_flight;
@@ -214,7 +202,7 @@ int aio_bench(Rados& rados, rados_pool_t pool, int secondsToRun,
     name[slot] = newName;
     contents[slot] = newContents;
   }
-  
+
   while (data->finished < data->started) {
     slot = data->finished % concurrentios;
     completions[slot]->wait_for_safe();
@@ -231,62 +219,91 @@ int aio_bench(Rados& rados, rados_pool_t pool, int secondsToRun,
     delete name[slot];
     delete contents[slot];
   }
+
   timePassed = g_clock.now() - data->startTime;
   dataLock.Lock();
   data->done = true;
   dataLock.Unlock();
-  
-  //check objects for consistency if requested
-  int errors = 0;
-  if (readOffResults) {
-    char matchName[128];
-    object_t oid;
-    bufferlist actualContents;
-    utime_t start_time;
-    utime_t lat;
-    double total_latency = 0;
-    double avg_latency;
-    double avg_bw;
-    for (int i = 0; i < data->finished; ++i ) {
-      snprintf(matchName, 128, "Object %d:%d", i, iTime);
-      oid = object_t(matchName);
-      snprintf(contentsChars, writeSize, "I'm the %dth object!", i);
-      start_time = g_clock.now();
-      rados.read(pool, oid, 0, actualContents, writeSize);
-      lat = g_clock.now() - start_time;
-      total_latency += (double) lat;
-      if (strcmp(contentsChars, actualContents.c_str()) != 0 ) {
-	cerr << "Object " << matchName << " is not correct!";
-	++errors;
-      }
-      actualContents.clear();
-    }
-    avg_latency = total_latency / data->finished;
-    avg_bw = data->finished * writeSize / (total_latency) / (1024 *1024);
-    cout << "read avg latency: " << avg_latency
-	 << " read avg bw: " << avg_bw << std::endl;
-  }
+
+  pthread_join(print_thread, NULL);
+
   double bandwidth;
-  bandwidth = ((double)data->finished)*((double)writeSize)/(double)timePassed;
+  bandwidth = ((double)data->finished)*((double)data->object_size)/(double)timePassed;
   bandwidth = bandwidth/(1024*1024); // we want it in MB/sec
+  char bw[20];
   sprintf(bw, "%.3lf \n", bandwidth);
   
   cout << "Total time run:        " << timePassed << std::endl
        << "Total writes made:     " << data->finished << std::endl
-       << "Write size:            " << writeSize << std::endl
+       << "Write size:            " << data->object_size << std::endl
        << "Bandwidth (MB/sec):    " << bw << std::endl
        << "Average Latency:       " << data->avg_latency << std::endl
        << "Max latency:           " << data->max_latency << std::endl
        << "Min latency:           " << data->min_latency << std::endl;
-  
-  if (readOffResults) {
-    if (errors) cout << "WARNING: There were " << errors << " total errors in copying!\n";
-    else cout << "No errors in copying!\n";
+}
+
+void *status_printer(void * data_store) {
+  bench_data *data = (bench_data *) data_store;
+  Cond cond;
+  int i = 0;
+  int previous_writes = 0;
+  int cycleSinceChange = 0;
+  double avg_bandwidth;
+  double bandwidth;
+  utime_t ONE_SECOND;
+  ONE_SECOND.set_from_double(1.0);
+  dataLock.Lock();
+  while(!data->done) {
+    if (i % 20 == 0) {
+      if (i > 0)
+	cout << "min lat: " << data->min_latency
+	     << " max lat: " << data->max_latency
+	     << " avg lat: " << data->avg_latency << std::endl;
+      //I'm naughty and don't reset the fill
+      cout << setfill(' ') 
+	   << setw(5) << "sec" 
+	   << setw(8) << "Cur ops"
+	   << setw(10) << "started"
+	   << setw(10) << "finished"
+	   << setw(10) << "avg MB/s"
+	   << setw(10) << "cur MB/s"
+	   << setw(10) << "last lat"
+	   << setw(10) << "avg lat" << std::endl;
+    }
+    bandwidth = (double)(data->finished - previous_writes)
+      * (data->trans_size)
+      / (1024*1024)
+      / cycleSinceChange;
+    avg_bandwidth = (double) (data->trans_size) * (data->finished)
+      / (double)(g_clock.now() - data->startTime) / (1024*1024);
+    if (previous_writes != data->finished) {
+      previous_writes = data->finished;
+      cycleSinceChange = 0;
+      cout << setfill(' ') 
+	   << setw(5) << i
+	   << setw(8) << data->in_flight
+	   << setw(10) << data->started
+	   << setw(10) << data->finished
+	   << setw(10) << avg_bandwidth
+	   << setw(10) << bandwidth
+	   << setw(10) << (double)data->cur_latency
+	   << setw(10) << data->avg_latency << std::endl;
+    }
+    else {
+      cout << setfill(' ')
+	   << setw(5) << i
+	   << setw(8) << data->in_flight
+	   << setw(10) << data->started
+	   << setw(10) << data->finished
+	   << setw(10) << avg_bandwidth
+	   << setw(10) << '0'
+	   << setw(10) << '-'
+	   << setw(10) << data->avg_latency << std::endl;
+    }
+    ++i;
+    ++cycleSinceChange;
+    cond.WaitInterval(dataLock, ONE_SECOND);
   }
-  
-  pthread_join(print_thread, NULL);
-  
-  delete contentsChars;
-  delete data;
-  return 0;
+  dataLock.Unlock();
+  return NULL;
 }