3 * Copyright 2015 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
23 #include <unordered_map>
26 #include <grpc/support/alloc.h>
27 #include <grpc/support/log.h>
28 #include <grpc/support/string_util.h>
29 #include <grpcpp/channel.h>
30 #include <grpcpp/client_context.h>
31 #include <grpcpp/create_channel.h>
33 #include "src/core/lib/gpr/env.h"
34 #include "src/core/lib/gprpp/host_port.h"
35 #include "src/core/lib/profiling/timers.h"
36 #include "src/proto/grpc/testing/worker_service.grpc.pb.h"
37 #include "test/core/util/port.h"
38 #include "test/core/util/test_config.h"
39 #include "test/cpp/qps/client.h"
40 #include "test/cpp/qps/driver.h"
41 #include "test/cpp/qps/histogram.h"
42 #include "test/cpp/qps/qps_worker.h"
43 #include "test/cpp/qps/stats.h"
44 #include "test/cpp/util/test_credentials_provider.h"
48 using std::unique_ptr;
53 static std::string get_host(const std::string& worker) {
54 absl::string_view host;
55 absl::string_view port;
56 grpc_core::SplitHostPort(worker.c_str(), &host, &port);
57 return std::string(host.data(), host.size());
60 static deque<string> get_workers(const string& env_name) {
62 char* env = gpr_getenv(env_name.c_str());
67 if (strlen(env) != 0) {
69 char* comma = strchr(p, ',');
71 out.emplace_back(p, comma);
81 "Environment variable \"%s\" does not contain a list of QPS "
82 "workers to use. Set it to a comma-separated list of "
83 "hostname:port pairs, starting with hosts that should act as "
84 "servers. E.g. export "
85 "%s=\"serverhost1:1234,clienthost1:1234,clienthost2:1234\"",
86 env_name.c_str(), env_name.c_str());
92 std::string GetCredType(
93 const std::string& worker_addr,
94 const std::map<std::string, std::string>& per_worker_credential_types,
95 const std::string& credential_type) {
96 auto it = per_worker_credential_types.find(worker_addr);
97 if (it != per_worker_credential_types.end()) {
100 return credential_type;
103 // helpers for postprocess_scenario_result
104 static double WallTime(const ClientStats& s) { return s.time_elapsed(); }
105 static double SystemTime(const ClientStats& s) { return s.time_system(); }
106 static double UserTime(const ClientStats& s) { return s.time_user(); }
107 static double CliPollCount(const ClientStats& s) { return s.cq_poll_count(); }
108 static double SvrPollCount(const ServerStats& s) { return s.cq_poll_count(); }
109 static double ServerWallTime(const ServerStats& s) { return s.time_elapsed(); }
110 static double ServerSystemTime(const ServerStats& s) { return s.time_system(); }
111 static double ServerUserTime(const ServerStats& s) { return s.time_user(); }
112 static double ServerTotalCpuTime(const ServerStats& s) {
113 return s.total_cpu_time();
115 static double ServerIdleCpuTime(const ServerStats& s) {
116 return s.idle_cpu_time();
118 static int Cores(int n) { return n; }
120 static bool IsSuccess(const Status& s) {
121 if (s.ok()) return true;
122 // Since we shutdown servers and clients at the same time, they both can
123 // observe cancellation. Thus, we consider CANCELLED as good status.
124 if (static_cast<StatusCode>(s.error_code()) == StatusCode::CANCELLED) {
127 // Since we shutdown servers and clients at the same time, server can close
128 // the socket before the client attempts to do that, and vice versa. Thus
129 // receiving a "Socket closed" error is fine.
130 if (s.error_message() == "Socket closed") return true;
134 // Postprocess ScenarioResult and populate result summary.
135 static void postprocess_scenario_result(ScenarioResult* result) {
136 // Get latencies from ScenarioResult latencies histogram and populate to
139 histogram.MergeProto(result->latencies());
140 result->mutable_summary()->set_latency_50(histogram.Percentile(50));
141 result->mutable_summary()->set_latency_90(histogram.Percentile(90));
142 result->mutable_summary()->set_latency_95(histogram.Percentile(95));
143 result->mutable_summary()->set_latency_99(histogram.Percentile(99));
144 result->mutable_summary()->set_latency_999(histogram.Percentile(99.9));
146 // Calculate qps and cpu load for each client and then aggregate results for
149 double client_system_cpu_load = 0, client_user_cpu_load = 0;
150 for (int i = 0; i < result->client_stats_size(); i++) {
151 auto client_stat = result->client_stats(i);
152 qps += client_stat.latencies().count() / client_stat.time_elapsed();
153 client_system_cpu_load +=
154 client_stat.time_system() / client_stat.time_elapsed();
155 client_user_cpu_load +=
156 client_stat.time_user() / client_stat.time_elapsed();
158 // Calculate cpu load for each server and then aggregate results for all
160 double server_system_cpu_load = 0, server_user_cpu_load = 0;
161 for (int i = 0; i < result->server_stats_size(); i++) {
162 auto server_stat = result->server_stats(i);
163 server_system_cpu_load +=
164 server_stat.time_system() / server_stat.time_elapsed();
165 server_user_cpu_load +=
166 server_stat.time_user() / server_stat.time_elapsed();
168 result->mutable_summary()->set_qps(qps);
169 // Populate the percentage of cpu load to result summary.
170 result->mutable_summary()->set_server_system_time(100 *
171 server_system_cpu_load);
172 result->mutable_summary()->set_server_user_time(100 * server_user_cpu_load);
173 result->mutable_summary()->set_client_system_time(100 *
174 client_system_cpu_load);
175 result->mutable_summary()->set_client_user_time(100 * client_user_cpu_load);
177 // For Non-linux platform, get_cpu_usage() is not implemented. Thus,
178 // ServerTotalCpuTime and ServerIdleCpuTime are both 0.
179 if (average(result->server_stats(), ServerTotalCpuTime) == 0) {
180 result->mutable_summary()->set_server_cpu_usage(0);
182 auto server_cpu_usage =
183 100 - 100 * average(result->server_stats(), ServerIdleCpuTime) /
184 average(result->server_stats(), ServerTotalCpuTime);
185 result->mutable_summary()->set_server_cpu_usage(server_cpu_usage);
188 // Calculate and populate successful request per second and failed requests
189 // per seconds to result summary.
190 auto time_estimate = average(result->client_stats(), WallTime);
191 if (result->request_results_size() > 0) {
192 int64_t successes = 0;
193 int64_t failures = 0;
194 for (int i = 0; i < result->request_results_size(); i++) {
195 const RequestResultCount& rrc = result->request_results(i);
196 if (rrc.status_code() == 0) {
197 successes += rrc.count();
199 failures += rrc.count();
202 result->mutable_summary()->set_successful_requests_per_second(
203 successes / time_estimate);
204 result->mutable_summary()->set_failed_requests_per_second(failures /
208 // Fill in data for other metrics required in result summary
209 auto qps_per_server_core = qps / sum(result->server_cores(), Cores);
210 result->mutable_summary()->set_qps_per_server_core(qps_per_server_core);
211 result->mutable_summary()->set_client_polls_per_request(
212 sum(result->client_stats(), CliPollCount) / histogram.Count());
213 result->mutable_summary()->set_server_polls_per_request(
214 sum(result->server_stats(), SvrPollCount) / histogram.Count());
216 auto server_queries_per_cpu_sec =
217 histogram.Count() / (sum(result->server_stats(), ServerSystemTime) +
218 sum(result->server_stats(), ServerUserTime));
219 auto client_queries_per_cpu_sec =
220 histogram.Count() / (sum(result->client_stats(), SystemTime) +
221 sum(result->client_stats(), UserTime));
223 result->mutable_summary()->set_server_queries_per_cpu_sec(
224 server_queries_per_cpu_sec);
225 result->mutable_summary()->set_client_queries_per_cpu_sec(
226 client_queries_per_cpu_sec);
230 unique_ptr<WorkerService::Stub> stub;
231 unique_ptr<ClientReaderWriter<ClientArgs, ClientStatus>> stream;
235 unique_ptr<WorkerService::Stub> stub;
236 unique_ptr<ClientReaderWriter<ServerArgs, ServerStatus>> stream;
239 static void FinishClients(const std::vector<ClientData>& clients,
240 const ClientArgs& client_mark) {
241 gpr_log(GPR_INFO, "Finishing clients");
242 for (size_t i = 0, i_end = clients.size(); i < i_end; i++) {
243 auto client = &clients[i];
244 if (!client->stream->Write(client_mark)) {
245 gpr_log(GPR_ERROR, "Couldn't write mark to client %zu", i);
248 if (!client->stream->WritesDone()) {
249 gpr_log(GPR_ERROR, "Failed WritesDone for client %zu", i);
255 static void ReceiveFinalStatusFromClients(
256 const std::vector<ClientData>& clients, Histogram& merged_latencies,
257 std::unordered_map<int, int64_t>& merged_statuses, ScenarioResult& result) {
258 gpr_log(GPR_INFO, "Receiving final status from clients");
259 ClientStatus client_status;
260 for (size_t i = 0, i_end = clients.size(); i < i_end; i++) {
261 auto client = &clients[i];
262 // Read the client final status
263 if (client->stream->Read(&client_status)) {
264 gpr_log(GPR_INFO, "Received final status from client %zu", i);
265 const auto& stats = client_status.stats();
266 merged_latencies.MergeProto(stats.latencies());
267 for (int i = 0; i < stats.request_results_size(); i++) {
268 merged_statuses[stats.request_results(i).status_code()] +=
269 stats.request_results(i).count();
271 result.add_client_stats()->CopyFrom(stats);
272 // That final status should be the last message on the client stream
273 GPR_ASSERT(!client->stream->Read(&client_status));
275 gpr_log(GPR_ERROR, "Couldn't get final status from client %zu", i);
281 static void ShutdownClients(const std::vector<ClientData>& clients,
282 ScenarioResult& result) {
283 gpr_log(GPR_INFO, "Shutdown clients");
284 for (size_t i = 0, i_end = clients.size(); i < i_end; i++) {
285 auto client = &clients[i];
286 Status s = client->stream->Finish();
287 // Since we shutdown servers and clients at the same time, clients can
288 // observe cancellation. Thus, we consider both OK and CANCELLED as good
290 const bool success = IsSuccess(s);
291 result.add_client_success(success);
293 gpr_log(GPR_ERROR, "Client %zu had an error %s", i,
294 s.error_message().c_str());
300 static void FinishServers(const std::vector<ServerData>& servers,
301 const ServerArgs& server_mark) {
302 gpr_log(GPR_INFO, "Finishing servers");
303 for (size_t i = 0, i_end = servers.size(); i < i_end; i++) {
304 auto server = &servers[i];
305 if (!server->stream->Write(server_mark)) {
306 gpr_log(GPR_ERROR, "Couldn't write mark to server %zu", i);
309 if (!server->stream->WritesDone()) {
310 gpr_log(GPR_ERROR, "Failed WritesDone for server %zu", i);
316 static void ReceiveFinalStatusFromServer(const std::vector<ServerData>& servers,
317 ScenarioResult& result) {
318 gpr_log(GPR_INFO, "Receiving final status from servers");
319 ServerStatus server_status;
320 for (size_t i = 0, i_end = servers.size(); i < i_end; i++) {
321 auto server = &servers[i];
322 // Read the server final status
323 if (server->stream->Read(&server_status)) {
324 gpr_log(GPR_INFO, "Received final status from server %zu", i);
325 result.add_server_stats()->CopyFrom(server_status.stats());
326 result.add_server_cores(server_status.cores());
327 // That final status should be the last message on the server stream
328 GPR_ASSERT(!server->stream->Read(&server_status));
330 gpr_log(GPR_ERROR, "Couldn't get final status from server %zu", i);
336 static void ShutdownServers(const std::vector<ServerData>& servers,
337 ScenarioResult& result) {
338 gpr_log(GPR_INFO, "Shutdown servers");
339 for (size_t i = 0, i_end = servers.size(); i < i_end; i++) {
340 auto server = &servers[i];
341 Status s = server->stream->Finish();
342 // Since we shutdown servers and clients at the same time, servers can
343 // observe cancellation. Thus, we consider both OK and CANCELLED as good
345 const bool success = IsSuccess(s);
346 result.add_server_success(success);
348 gpr_log(GPR_ERROR, "Server %zu had an error %s", i,
349 s.error_message().c_str());
355 std::vector<grpc::testing::Server*>* g_inproc_servers = nullptr;
357 std::unique_ptr<ScenarioResult> RunScenario(
358 const ClientConfig& initial_client_config, size_t num_clients,
359 const ServerConfig& initial_server_config, size_t num_servers,
360 int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count,
361 const std::string& qps_server_target_override,
362 const std::string& credential_type,
363 const std::map<std::string, std::string>& per_worker_credential_types,
364 bool run_inproc, int32_t median_latency_collection_interval_millis) {
366 g_inproc_servers = new std::vector<grpc::testing::Server*>;
368 // Log everything from the driver
369 gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
371 // ClientContext allocations (all are destroyed at scope exit)
372 list<ClientContext> contexts;
373 auto alloc_context = [](list<ClientContext>* contexts) {
374 contexts->emplace_back();
375 auto context = &contexts->back();
376 context->set_wait_for_ready(true);
380 // To be added to the result, containing the final configuration used for
381 // client and config (including host, etc.)
382 ClientConfig result_client_config;
384 // Get client, server lists; ignore if inproc test
385 auto workers = (!run_inproc) ? get_workers("QPS_WORKERS") : deque<string>();
386 ClientConfig client_config = initial_client_config;
388 // Spawn some local workers if desired
389 vector<unique_ptr<QpsWorker>> local_workers;
390 for (int i = 0; i < abs(spawn_local_worker_count); i++) {
391 // act as if we're a new test -- gets a good rng seed
392 static bool called_init = false;
395 strcpy(args_buf, "some-benchmark");
396 char* args[] = {args_buf};
397 grpc_test_init(1, args);
402 // we use port # of -1 to indicate inproc
403 int driver_port = (!run_inproc) ? grpc_pick_unused_port_or_die() : -1;
404 local_workers.emplace_back(new QpsWorker(driver_port, 0, credential_type));
405 sprintf(addr, "localhost:%d", driver_port);
406 if (spawn_local_worker_count < 0) {
407 workers.push_front(addr);
409 workers.push_back(addr);
412 GPR_ASSERT(!workers.empty());
414 // if num_clients is set to <=0, do dynamic sizing: all workers
415 // except for servers are clients
416 if (num_clients <= 0) {
417 num_clients = workers.size() - num_servers;
420 // TODO(ctiller): support running multiple configurations, and binpack
421 // client/server pairs
422 // to available workers
423 GPR_ASSERT(workers.size() >= num_clients + num_servers);
425 // Trim to just what we need
426 workers.resize(num_clients + num_servers);
429 std::vector<ServerData> servers(num_servers);
430 std::unordered_map<string, std::deque<int>> hosts_cores;
431 ChannelArguments channel_args;
433 for (size_t i = 0; i < num_servers; i++) {
434 gpr_log(GPR_INFO, "Starting server on %s (worker #%" PRIuPTR ")",
435 workers[i].c_str(), i);
437 servers[i].stub = WorkerService::NewStub(grpc::CreateTestChannel(
439 GetCredType(workers[i], per_worker_credential_types, credential_type),
440 nullptr /* call creds */, {} /* interceptor creators */));
442 servers[i].stub = WorkerService::NewStub(
443 local_workers[i]->InProcessChannel(channel_args));
446 const ServerConfig& server_config = initial_server_config;
447 if (server_config.core_limit() != 0) {
449 "server config core limit is set but ignored by driver");
454 *args.mutable_setup() = server_config;
455 servers[i].stream = servers[i].stub->RunServer(alloc_context(&contexts));
456 if (!servers[i].stream->Write(args)) {
457 gpr_log(GPR_ERROR, "Could not write args to server %zu", i);
460 ServerStatus init_status;
461 if (!servers[i].stream->Read(&init_status)) {
462 gpr_log(GPR_ERROR, "Server %zu did not yield initial status", i);
466 std::string cli_target(INPROC_NAME_PREFIX);
467 cli_target += std::to_string(i);
468 client_config.add_server_targets(cli_target);
470 std::string host = get_host(workers[i]);
471 std::string cli_target =
472 grpc_core::JoinHostPort(host.c_str(), init_status.port());
473 client_config.add_server_targets(cli_target.c_str());
476 if (qps_server_target_override.length() > 0) {
477 // overriding the qps server target only makes since if there is <= 1
479 GPR_ASSERT(num_servers <= 1);
480 client_config.add_server_targets(qps_server_target_override);
482 client_config.set_median_latency_collection_interval_millis(
483 median_latency_collection_interval_millis);
485 // Targets are all set by now
486 result_client_config = client_config;
488 std::vector<ClientData> clients(num_clients);
489 size_t channels_allocated = 0;
490 for (size_t i = 0; i < num_clients; i++) {
491 const auto& worker = workers[i + num_servers];
492 gpr_log(GPR_INFO, "Starting client on %s (worker #%" PRIuPTR ")",
493 worker.c_str(), i + num_servers);
495 clients[i].stub = WorkerService::NewStub(grpc::CreateTestChannel(
497 GetCredType(worker, per_worker_credential_types, credential_type),
498 nullptr /* call creds */, {} /* interceptor creators */));
500 clients[i].stub = WorkerService::NewStub(
501 local_workers[i + num_servers]->InProcessChannel(channel_args));
503 ClientConfig per_client_config = client_config;
505 if (initial_client_config.core_limit() != 0) {
506 gpr_log(GPR_ERROR, "client config core limit set but ignored");
510 // Reduce channel count so that total channels specified is held regardless
511 // of the number of clients available
512 size_t num_channels =
513 (client_config.client_channels() - channels_allocated) /
515 channels_allocated += num_channels;
516 gpr_log(GPR_DEBUG, "Client %" PRIdPTR " gets %" PRIdPTR " channels", i,
518 per_client_config.set_client_channels(num_channels);
521 *args.mutable_setup() = per_client_config;
522 clients[i].stream = clients[i].stub->RunClient(alloc_context(&contexts));
523 if (!clients[i].stream->Write(args)) {
524 gpr_log(GPR_ERROR, "Could not write args to client %zu", i);
529 for (size_t i = 0; i < num_clients; i++) {
530 ClientStatus init_status;
531 if (!clients[i].stream->Read(&init_status)) {
532 gpr_log(GPR_ERROR, "Client %zu did not yield initial status", i);
537 // Send an initial mark: clients can use this to know that everything is ready
539 gpr_log(GPR_INFO, "Initiating");
540 ServerArgs server_mark;
541 server_mark.mutable_mark()->set_reset(true);
542 ClientArgs client_mark;
543 client_mark.mutable_mark()->set_reset(true);
544 ServerStatus server_status;
545 ClientStatus client_status;
546 for (size_t i = 0; i < num_clients; i++) {
547 auto client = &clients[i];
548 if (!client->stream->Write(client_mark)) {
549 gpr_log(GPR_ERROR, "Couldn't write mark to client %zu", i);
553 for (size_t i = 0; i < num_clients; i++) {
554 auto client = &clients[i];
555 if (!client->stream->Read(&client_status)) {
556 gpr_log(GPR_ERROR, "Couldn't get status from client %zu", i);
561 // Let everything warmup
562 gpr_log(GPR_INFO, "Warming up");
563 gpr_timespec start = gpr_now(GPR_CLOCK_REALTIME);
565 gpr_time_add(start, gpr_time_from_seconds(warmup_seconds, GPR_TIMESPAN)));
568 gpr_log(GPR_INFO, "Starting");
569 for (size_t i = 0; i < num_servers; i++) {
570 auto server = &servers[i];
571 if (!server->stream->Write(server_mark)) {
572 gpr_log(GPR_ERROR, "Couldn't write mark to server %zu", i);
576 for (size_t i = 0; i < num_clients; i++) {
577 auto client = &clients[i];
578 if (!client->stream->Write(client_mark)) {
579 gpr_log(GPR_ERROR, "Couldn't write mark to client %zu", i);
583 for (size_t i = 0; i < num_servers; i++) {
584 auto server = &servers[i];
585 if (!server->stream->Read(&server_status)) {
586 gpr_log(GPR_ERROR, "Couldn't get status from server %zu", i);
590 for (size_t i = 0; i < num_clients; i++) {
591 auto client = &clients[i];
592 if (!client->stream->Read(&client_status)) {
593 gpr_log(GPR_ERROR, "Couldn't get status from client %zu", i);
599 gpr_log(GPR_INFO, "Running");
600 // Use gpr_sleep_until rather than this_thread::sleep_until to support
601 // compilers that don't work with this_thread
602 gpr_sleep_until(gpr_time_add(
604 gpr_time_from_seconds(warmup_seconds + benchmark_seconds, GPR_TIMESPAN)));
606 gpr_timer_set_enabled(0);
609 std::unique_ptr<ScenarioResult> result(new ScenarioResult);
610 Histogram merged_latencies;
611 std::unordered_map<int, int64_t> merged_statuses;
613 // For the case where clients lead the test such as UNARY and
614 // STREAMING_FROM_CLIENT, clients need to finish completely while a server
615 // is running to prevent the clients from being stuck while waiting for
617 bool client_finish_first =
618 (client_config.rpc_type() != STREAMING_FROM_SERVER);
620 FinishClients(clients, client_mark);
622 if (!client_finish_first) {
623 FinishServers(servers, server_mark);
626 ReceiveFinalStatusFromClients(clients, merged_latencies, merged_statuses,
628 ShutdownClients(clients, *result);
630 if (client_finish_first) {
631 FinishServers(servers, server_mark);
634 ReceiveFinalStatusFromServer(servers, *result);
635 ShutdownServers(servers, *result);
637 if (g_inproc_servers != nullptr) {
638 delete g_inproc_servers;
641 merged_latencies.FillProto(result->mutable_latencies());
642 for (std::unordered_map<int, int64_t>::iterator it = merged_statuses.begin();
643 it != merged_statuses.end(); ++it) {
644 RequestResultCount* rrc = result->add_request_results();
645 rrc->set_status_code(it->first);
646 rrc->set_count(it->second);
648 postprocess_scenario_result(result.get());
653 const std::string& credential_type,
654 const std::map<std::string, std::string>& per_worker_credential_types) {
655 // Get client, server lists
657 auto workers = get_workers("QPS_WORKERS");
658 if (workers.empty()) {
662 for (size_t i = 0; i < workers.size(); i++) {
663 auto stub = WorkerService::NewStub(grpc::CreateTestChannel(
665 GetCredType(workers[i], per_worker_credential_types, credential_type),
666 nullptr /* call creds */, {} /* interceptor creators */));
668 grpc::ClientContext ctx;
669 ctx.set_wait_for_ready(true);
670 Status s = stub->QuitWorker(&ctx, dummy, &dummy);
672 gpr_log(GPR_ERROR, "Worker %zu could not be properly quit because %s", i,
673 s.error_message().c_str());
680 } // namespace testing