test/cpp/qps/qps_json_driver.cc

   1 /*
   2  *
   3  * Copyright 2015-2016 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <fstream>
  20 #include <iostream>
  21 #include <memory>
  22 #include <set>
  23
  24 #include "absl/flags/flag.h"
  25
  26 #include <grpc/support/log.h>
  27 #include <grpcpp/impl/codegen/config_protobuf.h>
  28
  29 #include "test/core/util/test_config.h"
  30 #include "test/cpp/qps/benchmark_config.h"
  31 #include "test/cpp/qps/driver.h"
  32 #include "test/cpp/qps/parse_json.h"
  33 #include "test/cpp/qps/report.h"
  34 #include "test/cpp/qps/server.h"
  35 #include "test/cpp/util/test_config.h"
  36 #include "test/cpp/util/test_credentials_provider.h"
  37
  38 ABSL_FLAG(std::string, scenarios_file, "",
  39           "JSON file containing an array of Scenario objects");
  40 ABSL_FLAG(std::string, scenarios_json, "",
  41           "JSON string containing an array of Scenario objects");
  42 ABSL_FLAG(bool, quit, false, "Quit the workers");
  43 ABSL_FLAG(std::string, search_param, "",
  44           "The parameter, whose value is to be searched for to achieve "
  45           "targeted cpu load. For now, we have 'offered_load'. Later, "
  46           "'num_channels', 'num_outstanding_requests', etc. shall be "
  47           "added.");
  48 ABSL_FLAG(
  49     double, initial_search_value, 0.0,
  50     "initial parameter value to start the search with (i.e. lower bound)");
  51 ABSL_FLAG(double, targeted_cpu_load, 70.0,
  52           "Targeted cpu load (unit: %, range [0,100])");
  53 ABSL_FLAG(double, stride, 1,
  54           "Defines each stride of the search. The larger the stride is, "
  55           "the coarser the result will be, but will also be faster.");
  56 ABSL_FLAG(double, error_tolerance, 0.01,
  57           "Defines threshold for stopping the search. When current search "
  58           "range is narrower than the error_tolerance computed range, we "
  59           "stop the search.");
  60
  61 ABSL_FLAG(std::string, qps_server_target_override, "",
  62           "Override QPS server target to configure in client configs."
  63           "Only applicable if there is a single benchmark server.");
  64
  65 ABSL_FLAG(std::string, json_file_out, "", "File to write the JSON output to.");
  66
  67 ABSL_FLAG(std::string, credential_type, grpc::testing::kInsecureCredentialsType,
  68           "Credential type for communication with workers");
  69 ABSL_FLAG(
  70     std::string, per_worker_credential_types, "",
  71     "A map of QPS worker addresses to credential types. When creating a "
  72     "channel to a QPS worker's driver port, the qps_json_driver first checks "
  73     "if the 'name:port' string is in the map, and it uses the corresponding "
  74     "credential type if so. If the QPS worker's 'name:port' string is not "
  75     "in the map, then the driver -> worker channel will be created with "
  76     "the credentials specified in --credential_type. The value of this flag "
  77     "is a semicolon-separated list of map entries, where each map entry is "
  78     "a comma-separated pair.");
  79 ABSL_FLAG(bool, run_inproc, false, "Perform an in-process transport test");
  80 ABSL_FLAG(
  81     int32_t, median_latency_collection_interval_millis, 0,
  82     "Specifies the period between gathering latency medians in "
  83     "milliseconds. The medians will be logged out on the client at the "
  84     "end of the benchmark run. If 0, this periodic collection is disabled.");
  85
  86 namespace grpc {
  87 namespace testing {
  88
  89 static std::map<std::string, std::string>
  90 ConstructPerWorkerCredentialTypesMap() {
  91   // Parse a list of the form: "addr1,cred_type1;addr2,cred_type2;..." into
  92   // a map.
  93   std::string remaining = absl::GetFlag(FLAGS_per_worker_credential_types);
  94   std::map<std::string, std::string> out;
  95   while (!remaining.empty()) {
  96     size_t next_semicolon = remaining.find(';');
  97     std::string next_entry = remaining.substr(0, next_semicolon);
  98     if (next_semicolon == std::string::npos) {
  99       remaining = "";
 100     } else {
 101       remaining = remaining.substr(next_semicolon + 1, std::string::npos);
 102     }
 103     size_t comma = next_entry.find(',');
 104     if (comma == std::string::npos) {
 105       gpr_log(GPR_ERROR,
 106               "Expectd --per_worker_credential_types to be a list "
 107               "of the form: 'addr1,cred_type1;addr2,cred_type2;...' "
 108               "into.");
 109       abort();
 110     }
 111     std::string addr = next_entry.substr(0, comma);
 112     std::string cred_type = next_entry.substr(comma + 1, std::string::npos);
 113     if (out.find(addr) != out.end()) {
 114       gpr_log(GPR_ERROR,
 115               "Found duplicate addr in per_worker_credential_types.");
 116       abort();
 117     }
 118     out[addr] = cred_type;
 119   }
 120   return out;
 121 }
 122
 123 static std::unique_ptr<ScenarioResult> RunAndReport(
 124     const Scenario& scenario,
 125     const std::map<std::string, std::string>& per_worker_credential_types,
 126     bool* success) {
 127   std::cerr << "RUNNING SCENARIO: " << scenario.name() << "\n";
 128   auto result = RunScenario(
 129       scenario.client_config(), scenario.num_clients(),
 130       scenario.server_config(), scenario.num_servers(),
 131       scenario.warmup_seconds(), scenario.benchmark_seconds(),
 132       !absl::GetFlag(FLAGS_run_inproc) ? scenario.spawn_local_worker_count()
 133                                        : -2,
 134       absl::GetFlag(FLAGS_qps_server_target_override),
 135       absl::GetFlag(FLAGS_credential_type), per_worker_credential_types,
 136       absl::GetFlag(FLAGS_run_inproc),
 137       absl::GetFlag(FLAGS_median_latency_collection_interval_millis));
 138
 139   // Amend the result with scenario config. Eventually we should adjust
 140   // RunScenario contract so we don't need to touch the result here.
 141   result->mutable_scenario()->CopyFrom(scenario);
 142
 143   GetReporter()->ReportQPS(*result);
 144   GetReporter()->ReportQPSPerCore(*result);
 145   GetReporter()->ReportLatency(*result);
 146   GetReporter()->ReportTimes(*result);
 147   GetReporter()->ReportCpuUsage(*result);
 148   GetReporter()->ReportPollCount(*result);
 149   GetReporter()->ReportQueriesPerCpuSec(*result);
 150
 151   for (int i = 0; *success && i < result->client_success_size(); i++) {
 152     *success = result->client_success(i);
 153   }
 154   for (int i = 0; *success && i < result->server_success_size(); i++) {
 155     *success = result->server_success(i);
 156   }
 157
 158   if (!absl::GetFlag(FLAGS_json_file_out).empty()) {
 159     std::ofstream json_outfile;
 160     json_outfile.open(absl::GetFlag(FLAGS_json_file_out));
 161     json_outfile << "{\"qps\": " << result->summary().qps() << "}\n";
 162     json_outfile.close();
 163   }
 164
 165   return result;
 166 }
 167
 168 static double GetCpuLoad(
 169     Scenario* scenario, double offered_load,
 170     const std::map<std::string, std::string>& per_worker_credential_types,
 171     bool* success) {
 172   scenario->mutable_client_config()
 173       ->mutable_load_params()
 174       ->mutable_poisson()
 175       ->set_offered_load(offered_load);
 176   auto result = RunAndReport(*scenario, per_worker_credential_types, success);
 177   return result->summary().server_cpu_usage();
 178 }
 179
 180 static double BinarySearch(
 181     Scenario* scenario, double targeted_cpu_load, double low, double high,
 182     const std::map<std::string, std::string>& per_worker_credential_types,
 183     bool* success) {
 184   while (low <= high * (1 - absl::GetFlag(FLAGS_error_tolerance))) {
 185     double mid = low + (high - low) / 2;
 186     double current_cpu_load =
 187         GetCpuLoad(scenario, mid, per_worker_credential_types, success);
 188     gpr_log(GPR_DEBUG, "Binary Search: current_offered_load %.0f", mid);
 189     if (!*success) {
 190       gpr_log(GPR_ERROR, "Client/Server Failure");
 191       break;
 192     }
 193     if (targeted_cpu_load <= current_cpu_load) {
 194       high = mid - absl::GetFlag(FLAGS_stride);
 195     } else {
 196       low = mid + absl::GetFlag(FLAGS_stride);
 197     }
 198   }
 199
 200   return low;
 201 }
 202
 203 static double SearchOfferedLoad(
 204     double initial_offered_load, double targeted_cpu_load, Scenario* scenario,
 205     const std::map<std::string, std::string>& per_worker_credential_types,
 206     bool* success) {
 207   std::cerr << "RUNNING SCENARIO: " << scenario->name() << "\n";
 208   double current_offered_load = initial_offered_load;
 209   double current_cpu_load = GetCpuLoad(scenario, current_offered_load,
 210                                        per_worker_credential_types, success);
 211   if (current_cpu_load > targeted_cpu_load) {
 212     gpr_log(GPR_ERROR, "Initial offered load too high");
 213     return -1;
 214   }
 215
 216   while (*success && (current_cpu_load < targeted_cpu_load)) {
 217     current_offered_load *= 2;
 218     current_cpu_load = GetCpuLoad(scenario, current_offered_load,
 219                                   per_worker_credential_types, success);
 220     gpr_log(GPR_DEBUG, "Binary Search: current_offered_load  %.0f",
 221             current_offered_load);
 222   }
 223
 224   double targeted_offered_load =
 225       BinarySearch(scenario, targeted_cpu_load, current_offered_load / 2,
 226                    current_offered_load, per_worker_credential_types, success);
 227
 228   return targeted_offered_load;
 229 }
 230
 231 static bool QpsDriver() {
 232   std::string json;
 233
 234   bool scfile = (!absl::GetFlag(FLAGS_scenarios_file).empty());
 235   bool scjson = (!absl::GetFlag(FLAGS_scenarios_json).empty());
 236   if ((!scfile && !scjson && !absl::GetFlag(FLAGS_quit)) ||
 237       (scfile && (scjson || absl::GetFlag(FLAGS_quit))) ||
 238       (scjson && absl::GetFlag(FLAGS_quit))) {
 239     gpr_log(GPR_ERROR,
 240             "Exactly one of --scenarios_file, --scenarios_json, "
 241             "or --quit must be set");
 242     abort();
 243   }
 244
 245   auto per_worker_credential_types = ConstructPerWorkerCredentialTypesMap();
 246   if (scfile) {
 247     // Read the json data from disk
 248     FILE* json_file = fopen(absl::GetFlag(FLAGS_scenarios_file).c_str(), "r");
 249     GPR_ASSERT(json_file != nullptr);
 250     fseek(json_file, 0, SEEK_END);
 251     long len = ftell(json_file);
 252     char* data = new char[len];
 253     fseek(json_file, 0, SEEK_SET);
 254     GPR_ASSERT(len == (long)fread(data, 1, len, json_file));
 255     fclose(json_file);
 256     json = std::string(data, data + len);
 257     delete[] data;
 258   } else if (scjson) {
 259     json = absl::GetFlag(FLAGS_scenarios_json).c_str();
 260   } else if (absl::GetFlag(FLAGS_quit)) {
 261     return RunQuit(absl::GetFlag(FLAGS_credential_type),
 262                    per_worker_credential_types);
 263   }
 264
 265   // Parse into an array of scenarios
 266   Scenarios scenarios;
 267   ParseJson(json.c_str(), "grpc.testing.Scenarios", &scenarios);
 268   bool success = true;
 269
 270   // Make sure that there is at least some valid scenario here
 271   GPR_ASSERT(scenarios.scenarios_size() > 0);
 272
 273   for (int i = 0; i < scenarios.scenarios_size(); i++) {
 274     if (absl::GetFlag(FLAGS_search_param).empty()) {
 275       const Scenario& scenario = scenarios.scenarios(i);
 276       RunAndReport(scenario, per_worker_credential_types, &success);
 277     } else {
 278       if (absl::GetFlag(FLAGS_search_param) == "offered_load") {
 279         Scenario* scenario = scenarios.mutable_scenarios(i);
 280         double targeted_offered_load =
 281             SearchOfferedLoad(absl::GetFlag(FLAGS_initial_search_value),
 282                               absl::GetFlag(FLAGS_targeted_cpu_load), scenario,
 283                               per_worker_credential_types, &success);
 284         gpr_log(GPR_INFO, "targeted_offered_load %f", targeted_offered_load);
 285         GetCpuLoad(scenario, targeted_offered_load, per_worker_credential_types,
 286                    &success);
 287       } else {
 288         gpr_log(GPR_ERROR, "Unimplemented search param");
 289       }
 290     }
 291   }
 292   return success;
 293 }
 294
 295 }  // namespace testing
 296 }  // namespace grpc
 297
 298 int main(int argc, char** argv) {
 299   grpc::testing::TestEnvironment env(argc, argv);
 300   grpc::testing::InitTest(&argc, &argv, true);
 301
 302   bool ok = grpc::testing::QpsDriver();
 303
 304   return ok ? 0 : 1;
 305 }