ndb - threads and timings to hugo* pk ops

storage/ndb/test/include/HugoTransactions.hpp:
  threads and timings to hugo* pk ops
storage/ndb/test/include/NDBT_Thread.hpp:
  threads and timings to hugo* pk ops
storage/ndb/test/src/HugoTransactions.cpp:
  threads and timings to hugo* pk ops
storage/ndb/test/src/Makefile.am:
  threads and timings to hugo* pk ops
storage/ndb/test/src/NDBT_Thread.cpp:
  threads and timings to hugo* pk ops
storage/ndb/test/tools/hugoPkDelete.cpp:
  threads and timings to hugo* pk ops
storage/ndb/test/tools/hugoPkRead.cpp:
  threads and timings to hugo* pk ops
storage/ndb/test/tools/hugoPkUpdate.cpp:
  threads and timings to hugo* pk ops

1 files changed, 91 insertions, 7 deletions

diff --git a/storage/ndb/test/tools/hugoPkUpdate.cpp b/storage/ndb/test/tools/hugoPkUpdate.cpp
index 3e950bc96cd..b920a4f396a 100644
--- a/storage/ndb/test/tools/hugoPkUpdate.cpp
+++ b/storage/ndb/test/tools/hugoPkUpdate.cpp
@@ -20,24 +20,43 @@
 #include <NdbApi.hpp>
 #include <NdbMain.h>
 #include <NDBT.hpp> 
+#include <NDBT_Thread.hpp>
+#include <NDBT_Stats.hpp>
 #include <NdbSleep.h>
 #include <getarg.h>
 
 #include <HugoTransactions.hpp>
 
+static NDBT_ThreadFunc hugoPkUpdate;
+
+struct ThrInput {
+  const NdbDictionary::Table* pTab;
+  int records;
+  int batch;
+  int stats;
+};
+
+struct ThrOutput {
+  NDBT_Stats latency;
+};
+
 int main(int argc, const char** argv){
   ndb_init();
 
   int _records = 0;
   int _loops = 1;
+  int _threads = 1;
+  int _stats = 0;
   int _abort = 0;
-  int _batch = 0;
+  int _batch = 1;
   const char* _tabname = NULL, *db = 0;
   int _help = 0;
 
   struct getargs args[] = {
     { "aborts", 'a', arg_integer, &_abort, "percent of transactions that are aborted", "abort%" },
     { "loops", 'l', arg_integer, &_loops, "number of times to run this program(0=infinite loop)", "loops" },
+    { "threads", 't', arg_integer, &_threads, "number of threads (default 1)", "threads" },
+    { "stats", 's', arg_flag, &_stats, "report latency per batch", "stats" },
     //    { "batch", 'b', arg_integer, &_batch, "batch value", "batch" },
     { "records", 'r', arg_integer, &_records, "Number of records", "records" },
     { "usage", '?', arg_flag, &_help, "Print help", "" },
@@ -83,16 +102,81 @@ int main(int argc, const char** argv){
     return NDBT_ProgramExit(NDBT_WRONGARGS);
   }
 
-  HugoTransactions hugoTrans(*pTab);
+  // threads
+  NDBT_ThreadSet ths(_threads);
+
+  // create Ndb object for each thread
+  if (ths.connect(&con, "TEST_DB") == -1) {
+    ndbout << "connect failed: err=" << ths.get_err() << endl;
+    return NDBT_ProgramExit(NDBT_FAILED);
+  }
+
+  // input is options
+  ThrInput input;
+  ths.set_input(&input);
+  input.pTab = pTab;
+  input.records = _records;
+  input.batch = _batch;
+  input.stats = _stats;
+
+  // output is stats
+  ThrOutput output;
+  ths.set_output<ThrOutput>();
+
   int i = 0;
-  while (i<_loops || _loops==0) {
-    ndbout << "loop " << i << ": ";
-    if (hugoTrans.pkUpdateRecords(&MyNdb, 
-				  _records) != 0){
-      return NDBT_ProgramExit(NDBT_FAILED);
+  while (i < _loops || _loops == 0) {
+    ndbout << i << ": ";
+
+    ths.set_func(hugoPkUpdate);
+    ths.start();
+    ths.stop();
+
+    if (ths.get_err())
+      NDBT_ProgramExit(NDBT_FAILED);
+
+    if (_stats) {
+      NDBT_Stats latency;
+
+      // add stats from each thread
+      int n;
+      for (n = 0; n < ths.get_count(); n++) {
+        NDBT_Thread& thr = ths.get_thread(n);
+        ThrOutput* output = (ThrOutput*)thr.get_output();
+        latency += output->latency;
+      }
+
+      ndbout
+        << "latency per batch (us): "
+        << " samples=" << latency.getCount()
+        << " min=" << (int)latency.getMin()
+        << " max=" << (int)latency.getMax()
+        << " mean=" << (int)latency.getMean()
+        << " stddev=" << (int)latency.getStddev()
+        << endl;
     }
     i++;
   }
 
   return NDBT_ProgramExit(NDBT_OK);
 }
+
+static void hugoPkUpdate(NDBT_Thread& thr)
+{
+  const ThrInput* input = (const ThrInput*)thr.get_input();
+  ThrOutput* output = (ThrOutput*)thr.get_output();
+
+  HugoTransactions hugoTrans(*input->pTab);
+  output->latency.reset();
+  if (input->stats)
+    hugoTrans.setStatsLatency(&output->latency);
+
+  NDBT_ThreadSet& ths = thr.get_thread_set();
+  hugoTrans.setThrInfo(ths.get_count(), thr.get_thread_no());
+
+  int ret;
+  ret = hugoTrans.pkUpdateRecords(thr.get_ndb(),
+                                  input->records,
+                                  input->batch);
+  if (ret != 0)
+    thr.set_err(ret);
+}