Add an option to collect medians every epoch. Useful for gathering per second median latencies

8 files changed, 74 insertions, 14 deletions

diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h
index 9d7469c9b5..a0961dc770 100644
--- a/test/cpp/qps/client.h
+++ b/test/cpp/qps/client.h
@@ -180,6 +180,19 @@ class Client {
       timer_result = timer_->Mark();
     }
 
+    // Print the median latency per interval for one thread.
+    // If the number of warmup seconds is x, then the first x + 1 numbers in the
+    // vector are from the warmup period and should be discarded.
+    if (median_latency_collection_interval_seconds_ > 0) {
+      std::vector<double> medians_per_interval =
+          threads_[0]->GetMedianPerIntervalList();
+      gpr_log(GPR_INFO, "Num threads: %ld", threads_.size());
+      gpr_log(GPR_INFO, "Number of medians: %ld", medians_per_interval.size());
+      for (size_t j = 0; j < medians_per_interval.size(); j++) {
+        gpr_log(GPR_INFO, "%f", medians_per_interval[j]);
+      }
+    }
+
     grpc_stats_data core_stats;
     grpc_stats_collect(&core_stats);
 
@@ -210,6 +223,12 @@ class Client {
     }
   }
 
+  // Returns the interval (in seconds) between collecting latency medians. If 0,
+  // no periodic median latencies will be collected.
+  double GetLatencyCollectionIntervalInSeconds() {
+    return median_latency_collection_interval_seconds_;
+  }
+
   virtual int GetPollCount() {
     // For sync client.
     return 0;
@@ -218,6 +237,7 @@ class Client {
  protected:
   bool closed_loop_;
   gpr_atm thread_pool_done_;
+  double median_latency_collection_interval_seconds_;  // In seconds
 
   void StartThreads(size_t num_threads) {
     gpr_atm_rel_store(&thread_pool_done_, static_cast<gpr_atm>(false));
@@ -299,10 +319,28 @@ class Client {
       MergeStatusHistogram(statuses_, s);
     }
 
+    std::vector<double> GetMedianPerIntervalList() {
+      return medians_each_interval_list_;
+    }
+
+
     void UpdateHistogram(HistogramEntry* entry) {
       std::lock_guard<std::mutex> g(mu_);
       if (entry->value_used()) {
         histogram_.Add(entry->value());
+        if (client_->GetLatencyCollectionIntervalInSeconds() > 0) {
+          histogram_per_interval_.Add(entry->value());
+          double now = UsageTimer::Now();
+          if ((now - interval_start_time_) >=
+              client_->GetLatencyCollectionIntervalInSeconds()) {
+            // Record the median latency of requests from the last interval.
+            // Divide by 1e3 to get microseconds.
+            medians_each_interval_list_.push_back(
+                histogram_per_interval_.Percentile(50) / 1e3);
+            histogram_per_interval_.Reset();
+            interval_start_time_ = now;
+          }
+        }
       }
       if (entry->status_used()) {
         statuses_[entry->status()]++;
@@ -334,6 +372,11 @@ class Client {
     Client* client_;
     const size_t idx_;
     std::thread impl_;
+    // The following are used only if
+    // median_latency_collection_interval_seconds_ is greater than 0
+    Histogram histogram_per_interval_;
+    std::vector<double> medians_each_interval_list_;
+    double interval_start_time_;
   };
 
   bool ThreadCompleted() {
@@ -392,7 +435,8 @@ class ClientImpl : public Client {
     for (auto& t : connecting_threads) {
       t->join();
     }
-
+    median_latency_collection_interval_seconds_ =
+        config.median_latency_collection_interval_millis() / 1e3;
     ClientRequestCreator<RequestType> create_req(&request_,
                                                  config.payload_config());
   }
diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc
index cabbd51843..11cfb4aa05 100644
--- a/test/cpp/qps/driver.cc
+++ b/test/cpp/qps/driver.cc
@@ -198,7 +198,8 @@ std::unique_ptr<ScenarioResult> RunScenario(
     const ServerConfig& initial_server_config, size_t num_servers,
     int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count,
     const grpc::string& qps_server_target_override,
-    const grpc::string& credential_type, bool run_inproc) {
+    const grpc::string& credential_type, bool run_inproc,
+    int32_t median_latency_collection_interval_millis) {
   if (run_inproc) {
     g_inproc_servers = new std::vector<grpc::testing::Server*>;
   }
@@ -317,6 +318,9 @@ std::unique_ptr<ScenarioResult> RunScenario(
     }
   }
 
+  client_config.set_median_latency_collection_interval_millis(
+      median_latency_collection_interval_millis);
+
   // Targets are all set by now
   result_client_config = client_config;
   // Start clients
diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h
index fede4d8045..cda89f7ddf 100644
--- a/test/cpp/qps/driver.h
+++ b/test/cpp/qps/driver.h
@@ -32,7 +32,8 @@ std::unique_ptr<ScenarioResult> RunScenario(
     const grpc::testing::ServerConfig& server_config, size_t num_servers,
     int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count,
     const grpc::string& qps_server_target_override,
-    const grpc::string& credential_type, bool run_inproc);
+    const grpc::string& credential_type, bool run_inproc,
+    int32_t median_latency_collection_interval_millis);
 
 bool RunQuit(const grpc::string& credential_type);
 }  // namespace testing
diff --git a/test/cpp/qps/histogram.h b/test/cpp/qps/histogram.h
index ba72b5b332..6275128f34 100644
--- a/test/cpp/qps/histogram.h
+++ b/test/cpp/qps/histogram.h
@@ -34,6 +34,11 @@ class Histogram {
   ~Histogram() {
     if (impl_) grpc_histogram_destroy(impl_);
   }
+  void Reset() {
+    if (impl_) grpc_histogram_destroy(impl_);
+    impl_ = grpc_histogram_create(default_resolution(), default_max_possible());
+  }
+
   Histogram(Histogram&& other) : impl_(other.impl_) { other.impl_ = nullptr; }
 
   void Merge(const Histogram& h) { grpc_histogram_merge(impl_, h.impl_); }
diff --git a/test/cpp/qps/inproc_sync_unary_ping_pong_test.cc b/test/cpp/qps/inproc_sync_unary_ping_pong_test.cc
index f2e977d48b..56d1730252 100644
--- a/test/cpp/qps/inproc_sync_unary_ping_pong_test.cc
+++ b/test/cpp/qps/inproc_sync_unary_ping_pong_test.cc
@@ -48,7 +48,7 @@ static void RunSynchronousUnaryPingPong() {
 
   const auto result =
       RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2, "",
-                  kInsecureCredentialsType, true);
+                  kInsecureCredentialsType, true, 0);
 
   GetReporter()->ReportQPS(*result);
   GetReporter()->ReportLatency(*result);
diff --git a/test/cpp/qps/qps_json_driver.cc b/test/cpp/qps/qps_json_driver.cc
index 0ff692255c..c95e0df251 100644
--- a/test/cpp/qps/qps_json_driver.cc
+++ b/test/cpp/qps/qps_json_driver.cc
@@ -66,6 +66,11 @@ DEFINE_string(json_file_out, "", "File to write the JSON output to.");
 DEFINE_string(credential_type, grpc::testing::kInsecureCredentialsType,
               "Credential type for communication with workers");
 DEFINE_bool(run_inproc, false, "Perform an in-process transport test");
+DEFINE_int32(
+    median_latency_collection_interval_millis, 0,
+    "Specifies the period between gathering latency medians in "
+    "milliseconds. The medians will be logged out on the client at the "
+    "end of the benchmark run. If 0, this periodic collection is disabled.");
 
 namespace grpc {
 namespace testing {
@@ -73,13 +78,13 @@ namespace testing {
 static std::unique_ptr<ScenarioResult> RunAndReport(const Scenario& scenario,
                                                     bool* success) {
   std::cerr << "RUNNING SCENARIO: " << scenario.name() << "\n";
-  auto result =
-      RunScenario(scenario.client_config(), scenario.num_clients(),
-                  scenario.server_config(), scenario.num_servers(),
-                  scenario.warmup_seconds(), scenario.benchmark_seconds(),
-                  !FLAGS_run_inproc ? scenario.spawn_local_worker_count() : -2,
-                  FLAGS_qps_server_target_override, FLAGS_credential_type,
-                  FLAGS_run_inproc);
+  auto result = RunScenario(
+    scenario.client_config(), scenario.num_clients(),
+    scenario.server_config(), scenario.num_servers(),
+    scenario.warmup_seconds(), scenario.benchmark_seconds(),
+    !FLAGS_run_inproc ? scenario.spawn_local_worker_count() : -2,
+    FLAGS_qps_server_target_override, FLAGS_credential_type, FLAGS_run_inproc,
+    FLAGS_median_latency_collection_interval_millis);
 
   // Amend the result with scenario config. Eventually we should adjust
   // RunScenario contract so we don't need to touch the result here.
@@ -145,7 +150,8 @@ static double SearchOfferedLoad(double initial_offered_load,
                                 bool* success) {
   std::cerr << "RUNNING SCENARIO: " << scenario->name() << "\n";
   double current_offered_load = initial_offered_load;
-  double current_cpu_load = GetCpuLoad(scenario, current_offered_load, success);
+  double current_cpu_load =
+      GetCpuLoad(scenario, current_offered_load, success);
   if (current_cpu_load > targeted_cpu_load) {
     gpr_log(GPR_ERROR, "Initial offered load too high");
     return -1;
diff --git a/test/cpp/qps/qps_openloop_test.cc b/test/cpp/qps/qps_openloop_test.cc
index df929b9811..6044f4265a 100644
--- a/test/cpp/qps/qps_openloop_test.cc
+++ b/test/cpp/qps/qps_openloop_test.cc
@@ -52,7 +52,7 @@ static void RunQPS() {
 
   const auto result =
       RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2, "",
-                  kInsecureCredentialsType, false);
+                  kInsecureCredentialsType, false, 0);
 
   GetReporter()->ReportQPSPerCore(*result);
   GetReporter()->ReportLatency(*result);
diff --git a/test/cpp/qps/secure_sync_unary_ping_pong_test.cc b/test/cpp/qps/secure_sync_unary_ping_pong_test.cc
index bb415e9d63..a559c82cc8 100644
--- a/test/cpp/qps/secure_sync_unary_ping_pong_test.cc
+++ b/test/cpp/qps/secure_sync_unary_ping_pong_test.cc
@@ -55,7 +55,7 @@ static void RunSynchronousUnaryPingPong() {
 
   const auto result =
       RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2, "",
-                  kInsecureCredentialsType, false);
+                  kInsecureCredentialsType, false, 0);
 
   GetReporter()->ReportQPS(*result);
   GetReporter()->ReportLatency(*result);