multiclient loadgen

commit 4810fd8dd7db362a86e568caa5434832622706e2
Author: Andrew Wong <awong@cloudera.com>
Date:   Wed May 23 15:24:27 2018 -0700

    multiclient perf
    
    Change-Id: Ie1e7764a11261c9d337853370ee4b5c745557fa7

diff --git a/src/kudu/tools/tool_action_perf.cc b/src/kudu/tools/tool_action_perf.cc
index 2aeb514..9235905 100644
--- a/src/kudu/tools/tool_action_perf.cc
+++ b/src/kudu/tools/tool_action_perf.cc
@@ -254,6 +254,9 @@ DEFINE_bool(use_random, false,
             "Whether to use random numbers instead of sequential ones. "
             "In case of using random numbers collisions are possible over "
             "the data for columns with unique constraint (e.g. primary key).");
+DEFINE_int32(num_clients, 1,
+             "Number of client connections to spread the workload across. Does "
+             "a simple round-robining of clients");
 
 namespace kudu {
 namespace tools {
@@ -456,7 +459,7 @@ void GeneratorThread(
   }
 }
 
-Status GenerateInsertRows(const shared_ptr<KuduClient>& client,
+Status GenerateInsertRows(const vector<shared_ptr<KuduClient>>& clients,
                           const string& table_name,
                           uint64_t* total_row_count,
                           uint64_t* total_err_count) {
@@ -467,7 +470,7 @@ Status GenerateInsertRows(const shared_ptr<KuduClient>& client,
   vector<uint64_t> err_count(gen_num, 0);
   vector<thread> threads;
   for (size_t i = 0; i < gen_num; ++i) {
-    threads.emplace_back(&GeneratorThread, client, table_name, i, gen_num,
+    threads.emplace_back(&GeneratorThread, clients[i % clients.size()], table_name, i, gen_num,
                          &status[i], &row_count[i], &err_count[i]);
   }
   for (auto& t : threads) {
@@ -556,10 +559,14 @@ Status TestLoadGenerator(const RunnerContext& context) {
     return Status::InvalidArgument(
         "At least one master address must be specified");
   }
-  shared_ptr<KuduClient> client;
-  RETURN_NOT_OK(KuduClientBuilder()
-                .master_server_addrs(master_addrs)
-                .Build(&client));
+  vector<shared_ptr<KuduClient>> clients;
+  for (int i = 0; i < FLAGS_num_clients; i++) {
+    shared_ptr<KuduClient> client;
+    RETURN_NOT_OK(KuduClientBuilder()
+                  .master_server_addrs(master_addrs)
+                  .Build(&client));
+    clients.emplace_back(std::move(client));
+  }
   string table_name;
   bool is_auto_table = false;
   if (!FLAGS_table_name.empty()) {
@@ -578,7 +585,7 @@ Status TestLoadGenerator(const RunnerContext& context) {
     b.AddColumn("string_val")->Type(KuduColumnSchema::STRING);
     RETURN_NOT_OK(b.Build(&schema));
 
-    unique_ptr<KuduTableCreator> table_creator(client->NewTableCreator());
+    unique_ptr<KuduTableCreator> table_creator(clients[0]->NewTableCreator());
     RETURN_NOT_OK(table_creator->table_name(table_name)
                   .schema(&schema)
                   .num_replicas(FLAGS_table_num_replicas)
@@ -594,7 +601,7 @@ Status TestLoadGenerator(const RunnerContext& context) {
   uint64_t total_err_count = 0;
   Stopwatch sw(Stopwatch::ALL_THREADS);
   sw.start();
-  Status status = GenerateInsertRows(client, table_name,
+  Status status = GenerateInsertRows(clients, table_name,
                                      &total_row_count, &total_err_count);
   sw.stop();
   const double total = sw.elapsed().wall_millis();
@@ -621,7 +628,7 @@ Status TestLoadGenerator(const RunnerContext& context) {
   if (FLAGS_run_scan) {
     // Run a table scan to count inserted rows.
     uint64_t count;
-    RETURN_NOT_OK(CountTableRows(client, table_name, &count));
+    RETURN_NOT_OK(CountTableRows(clients[0], table_name, &count));
     cout << endl << "Scanner report" << endl
          << "  expected rows: " << total_row_count << endl
          << "  actual rows  : " << count << endl;
@@ -635,7 +642,7 @@ Status TestLoadGenerator(const RunnerContext& context) {
   if (is_auto_table && !FLAGS_keep_auto_table) {
     cout << "Dropping auto-created table '" << table_name << "'" << endl;
     // Drop the table which was automatically created to run the test.
-    RETURN_NOT_OK(client->DeleteTable(table_name));
+    RETURN_NOT_OK(clients[0]->DeleteTable(table_name));
   }
 
   return Status::OK();