diff options
author | Vijay Pai <vpai@google.com> | 2016-02-02 10:36:12 -0800 |
---|---|---|
committer | Vijay Pai <vpai@google.com> | 2016-02-02 10:36:12 -0800 |
commit | b47b556d33c0e22601e773cad17d576197b4942b (patch) | |
tree | 96f3900c391a24e27182e00c4bf020a55b4cfdb9 /test | |
parent | d99ed6b636eca5d7ba38f19337f447aa0fbeebb9 (diff) | |
parent | 71a083b1b14cf6f9844d51e7e7c3b75de16c0501 (diff) |
Merge branch 'master' into worker_quit
Diffstat (limited to 'test')
-rw-r--r-- | test/cpp/qps/client.h | 11 | ||||
-rw-r--r-- | test/cpp/qps/client_async.cc | 7 | ||||
-rw-r--r-- | test/cpp/qps/driver.cc | 131 | ||||
-rw-r--r-- | test/cpp/qps/limit_cores.cc | 80 | ||||
-rw-r--r-- | test/cpp/qps/limit_cores.h | 51 | ||||
-rwxr-xr-x | test/cpp/qps/qps-sweep.sh | 14 | ||||
-rw-r--r-- | test/cpp/qps/qps_driver.cc | 15 | ||||
-rw-r--r-- | test/cpp/qps/qps_worker.cc | 12 | ||||
-rw-r--r-- | test/cpp/qps/server.h | 13 |
9 files changed, 302 insertions, 32 deletions
diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h index 97487fd0b2..50b2bf2514 100644 --- a/test/cpp/qps/client.h +++ b/test/cpp/qps/client.h @@ -36,16 +36,20 @@ #include <condition_variable> #include <mutex> +#include <vector> #include <grpc++/support/byte_buffer.h> #include <grpc++/support/slice.h> +#include <grpc/support/log.h> +#include "src/proto/grpc/testing/payloads.grpc.pb.h" +#include "src/proto/grpc/testing/services.grpc.pb.h" + +#include "test/cpp/qps/limit_cores.h" #include "test/cpp/qps/histogram.h" #include "test/cpp/qps/interarrival.h" #include "test/cpp/qps/timer.h" #include "test/cpp/util/create_test_channel.h" -#include "src/proto/grpc/testing/payloads.grpc.pb.h" -#include "src/proto/grpc/testing/services.grpc.pb.h" namespace grpc { @@ -320,6 +324,8 @@ class ClientImpl : public Client { std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)> create_stub) : channels_(config.client_channels()), create_stub_(create_stub) { + cores_ = LimitCores(config.core_list().data(), config.core_list_size()); + for (int i = 0; i < config.client_channels(); i++) { channels_[i].init(config.server_targets(i % config.server_targets_size()), config, create_stub_); @@ -331,6 +337,7 @@ class ClientImpl : public Client { virtual ~ClientImpl() {} protected: + int cores_; RequestType request_; class ClientChannelInfo { diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc index 4229e1956e..f3f8f37051 100644 --- a/test/cpp/qps/client_async.cc +++ b/test/cpp/qps/client_async.cc @@ -159,6 +159,7 @@ class AsyncClient : public ClientImpl<StubType, RequestType> { using Client::SetupLoadTest; using Client::NextIssueTime; using Client::closed_loop_; + using ClientImpl<StubType, RequestType>::cores_; using ClientImpl<StubType, RequestType>::channels_; using ClientImpl<StubType, RequestType>::request_; AsyncClient(const ClientConfig& config, @@ -345,11 +346,11 @@ class AsyncClient : public ClientImpl<StubType, RequestType> { private: bool val_; }; - static int NumThreads(const ClientConfig& config) { + int NumThreads(const ClientConfig& config) { int num_threads = config.async_client_threads(); if (num_threads <= 0) { // Use dynamic sizing - num_threads = gpr_cpu_num_cores(); - gpr_log(GPR_INFO, "Sizing client server to %d threads", num_threads); + num_threads = cores_; + gpr_log(GPR_INFO, "Sizing async client to %d threads", num_threads); } return num_threads; } diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc index 5fd0f21d1c..a08aad0e68 100644 --- a/test/cpp/qps/driver.cc +++ b/test/cpp/qps/driver.cc @@ -34,6 +34,7 @@ #include <deque> #include <list> #include <thread> +#include <unordered_map> #include <vector> #include <grpc++/channel.h> @@ -59,7 +60,42 @@ using std::vector; namespace grpc { namespace testing { -static deque<string> get_hosts(const string& name) { +static std::string get_host(const std::string& worker) { + char* host; + char* port; + + gpr_split_host_port(worker.c_str(), &host, &port); + const string s(host); + + gpr_free(host); + gpr_free(port); + return s; +} + +static std::unordered_map<string, std::deque<int>> get_hosts_and_cores( + const deque<string>& workers) { + std::unordered_map<string, std::deque<int>> hosts; + for (auto it = workers.begin(); it != workers.end(); it++) { + const string host = get_host(*it); + if (hosts.find(host) == hosts.end()) { + auto stub = WorkerService::NewStub( + CreateChannel(*it, InsecureChannelCredentials())); + grpc::ClientContext ctx; + CoreRequest dummy; + CoreResponse cores; + grpc::Status s = stub->CoreCount(&ctx, dummy, &cores); + assert(s.ok()); + std::deque<int> dq; + for (int i = 0; i < cores.cores(); i++) { + dq.push_back(i); + } + hosts[host] = dq; + } + } + return hosts; +} + +static deque<string> get_workers(const string& name) { char* env = gpr_getenv(name.c_str()); if (!env) return deque<string>(); @@ -105,18 +141,18 @@ struct ClientData { std::unique_ptr<ScenarioResult> RunScenario( const ClientConfig& initial_client_config, size_t num_clients, - const ServerConfig& server_config, size_t num_servers, int warmup_seconds, - int benchmark_seconds, int spawn_local_worker_count) { + const ServerConfig& initial_server_config, size_t num_servers, + int warmup_seconds, int benchmark_seconds, int spawn_local_worker_count) { // ClientContext allocations (all are destroyed at scope exit) list<ClientContext> contexts; // To be added to the result, containing the final configuration used for // client and config (including host, etc.) ClientConfig result_client_config; - ServerConfig result_server_config; + const ServerConfig result_server_config = initial_server_config; // Get client, server lists - auto workers = get_hosts("QPS_WORKERS"); + auto workers = get_workers("QPS_WORKERS"); ClientConfig client_config = initial_client_config; // Spawn some local workers if desired @@ -143,6 +179,9 @@ std::unique_ptr<ScenarioResult> RunScenario( } } + // Setup the hosts and core counts + auto hosts_cores = get_hosts_and_cores(workers); + // if num_clients is set to <=0, do dynamic sizing: all workers // except for servers are clients if (num_clients <= 0) { @@ -172,18 +211,49 @@ std::unique_ptr<ScenarioResult> RunScenario( i); servers[i].stub = WorkerService::NewStub( CreateChannel(workers[i], InsecureChannelCredentials())); + + ServerConfig server_config = initial_server_config; + char* host; + char* driver_port; + char* cli_target; + gpr_split_host_port(workers[i].c_str(), &host, &driver_port); + string host_str(host); + int server_core_limit = initial_server_config.core_limit(); + int client_core_limit = initial_client_config.core_limit(); + + if (server_core_limit == 0 && client_core_limit > 0) { + // In this case, limit the server cores if it matches the + // same host as one or more clients + const auto& dq = hosts_cores.at(host_str); + bool match = false; + int limit = dq.size(); + for (size_t cli = 0; cli < num_clients; cli++) { + if (host_str == get_host(workers[cli + num_servers])) { + limit -= client_core_limit; + match = true; + } + } + if (match) { + GPR_ASSERT(limit > 0); + server_core_limit = limit; + } + } + if (server_core_limit > 0) { + auto& dq = hosts_cores.at(host_str); + GPR_ASSERT(dq.size() >= static_cast<size_t>(server_core_limit)); + for (int core = 0; core < server_core_limit; core++) { + server_config.add_core_list(dq.front()); + dq.pop_front(); + } + } + ServerArgs args; - result_server_config = server_config; *args.mutable_setup() = server_config; servers[i].stream = servers[i].stub->RunServer(runsc::AllocContext(&contexts, deadline)); GPR_ASSERT(servers[i].stream->Write(args)); ServerStatus init_status; GPR_ASSERT(servers[i].stream->Read(&init_status)); - char* host; - char* driver_port; - char* cli_target; - gpr_split_host_port(workers[i].c_str(), &host, &driver_port); gpr_join_host_port(&cli_target, host, init_status.port()); client_config.add_server_targets(cli_target); gpr_free(host); @@ -191,19 +261,50 @@ std::unique_ptr<ScenarioResult> RunScenario( gpr_free(cli_target); } + // Targets are all set by now + result_client_config = client_config; // Start clients using runsc::ClientData; // clients is array rather than std::vector to avoid gcc-4.4 issues // where class contained in std::vector must have a copy constructor auto* clients = new ClientData[num_clients]; for (size_t i = 0; i < num_clients; i++) { - gpr_log(GPR_INFO, "Starting client on %s (worker #%d)", - workers[i + num_servers].c_str(), i + num_servers); + const auto& worker = workers[i + num_servers]; + gpr_log(GPR_INFO, "Starting client on %s (worker #%d)", worker.c_str(), + i + num_servers); clients[i].stub = WorkerService::NewStub( - CreateChannel(workers[i + num_servers], InsecureChannelCredentials())); + CreateChannel(worker, InsecureChannelCredentials())); + ClientConfig per_client_config = client_config; + + int server_core_limit = initial_server_config.core_limit(); + int client_core_limit = initial_client_config.core_limit(); + if ((server_core_limit > 0) || (client_core_limit > 0)) { + auto& dq = hosts_cores.at(get_host(worker)); + if (client_core_limit == 0) { + // limit client cores if it matches a server host + bool match = false; + int limit = dq.size(); + for (size_t srv = 0; srv < num_servers; srv++) { + if (get_host(worker) == get_host(workers[srv])) { + match = true; + } + } + if (match) { + GPR_ASSERT(limit > 0); + client_core_limit = limit; + } + } + if (client_core_limit > 0) { + GPR_ASSERT(dq.size() >= static_cast<size_t>(client_core_limit)); + for (int core = 0; core < client_core_limit; core++) { + per_client_config.add_core_list(dq.front()); + dq.pop_front(); + } + } + } + ClientArgs args; - result_client_config = client_config; - *args.mutable_setup() = client_config; + *args.mutable_setup() = per_client_config; clients[i].stream = clients[i].stub->RunClient(runsc::AllocContext(&contexts, deadline)); GPR_ASSERT(clients[i].stream->Write(args)); diff --git a/test/cpp/qps/limit_cores.cc b/test/cpp/qps/limit_cores.cc new file mode 100644 index 0000000000..c2f3ad8fde --- /dev/null +++ b/test/cpp/qps/limit_cores.cc @@ -0,0 +1,80 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "test/cpp/qps/limit_cores.h" + +#include <grpc/support/cpu.h> +#include <grpc/support/log.h> +#include <grpc/support/port_platform.h> +#include <vector> + +namespace grpc { +namespace testing { + +#ifdef GPR_CPU_LINUX +#ifndef _GNU_SOURCE +#define _GNU_SOURCE +#endif +#include <sched.h> +int LimitCores(const int *cores, int cores_size) { + const int num_cores = gpr_cpu_num_cores(); + int cores_set = 0; + + cpu_set_t *cpup = CPU_ALLOC(num_cores); + GPR_ASSERT(cpup); + const size_t size = CPU_ALLOC_SIZE(num_cores); + CPU_ZERO_S(size, cpup); + + if (cores_size > 0) { + for (int i = 0; i < cores_size; i++) { + if (cores[i] < num_cores) { + CPU_SET_S(cores[i], size, cpup); + cores_set++; + } + } + } else { + for (int i = 0; i < num_cores; i++) { + CPU_SET_S(i, size, cpup); + cores_set++; + } + } + GPR_ASSERT(sched_setaffinity(0, size, cpup) == 0); + CPU_FREE(cpup); + return cores_set; +} +#else +// LimitCores is not currently supported for non-Linux platforms +int LimitCores(std::vector<int> core_vec) { return gpr_cpu_num_cores(); } +#endif +} // namespace testing +} // namespace grpc diff --git a/test/cpp/qps/limit_cores.h b/test/cpp/qps/limit_cores.h new file mode 100644 index 0000000000..5c0d1e315d --- /dev/null +++ b/test/cpp/qps/limit_cores.h @@ -0,0 +1,51 @@ +/* + * + * Copyright 2016, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef TEST_QPS_LIMIT_CORES_H +#define TEST_QPS_LIMIT_CORES_H + +#include <vector> + +namespace grpc { +namespace testing { +/// LimitCores: allow this worker to only run on the cores specified in the +/// array \a cores, which is of length \a cores_size. +/// +/// LimitCores takes array and size arguments (instead of vector) for direct +/// conversion from repeated field of protobuf. Use a cores_size of 0 to remove +/// existing limits (from an empty repeated field) +int LimitCores(const int *cores, int cores_size); +} // namespace testing +} // namespace grpc + +#endif // TEST_QPS_LIMIT_CORES_H diff --git a/test/cpp/qps/qps-sweep.sh b/test/cpp/qps/qps-sweep.sh index 333f4bd7d0..539da1d893 100755 --- a/test/cpp/qps/qps-sweep.sh +++ b/test/cpp/qps/qps-sweep.sh @@ -57,6 +57,20 @@ for secure in true false; do --async_client_threads=0 --async_server_threads=0 --secure_test=$secure \ --num_servers=1 --num_clients=0 + # Scenario 2b: QPS with a single server core + "$bins"/opt/qps_driver --rpc_type=STREAMING --client_type=ASYNC_CLIENT \ + --server_type=ASYNC_GENERIC_SERVER --outstanding_rpcs_per_channel=100 \ + --client_channels=64 --bbuf_req_size=0 --bbuf_resp_size=0 \ + --async_client_threads=0 --async_server_threads=0 --secure_test=$secure \ + --num_servers=1 --num_clients=0 --server_core_limit=1 + + # Scenario 2c: protobuf-based QPS + "$bins"/opt/qps_driver --rpc_type=STREAMING --client_type=ASYNC_CLIENT \ + --server_type=ASYNC_SERVER --outstanding_rpcs_per_channel=100 \ + --client_channels=64 --simple_req_size=0 --simple_resp_size=0 \ + --async_client_threads=0 --async_server_threads=0 --secure_test=$secure \ + --num_servers=1 --num_clients=0 + # Scenario 3: Latency at near-peak load (TBD) # Scenario 4: Single-channel bidirectional throughput test (like TCP_STREAM). diff --git a/test/cpp/qps/qps_driver.cc b/test/cpp/qps/qps_driver.cc index 1fe37b1667..69fb4d75e8 100644 --- a/test/cpp/qps/qps_driver.cc +++ b/test/cpp/qps/qps_driver.cc @@ -48,14 +48,13 @@ DEFINE_int32(warmup_seconds, 5, "Warmup time (in seconds)"); DEFINE_int32(benchmark_seconds, 30, "Benchmark time (in seconds)"); DEFINE_int32(local_workers, 0, "Number of local workers to start"); -// Common config -DEFINE_string(rpc_type, "UNARY", "Type of RPC: UNARY or STREAMING"); - // Server config DEFINE_int32(async_server_threads, 1, "Number of threads for async servers"); DEFINE_string(server_type, "SYNC_SERVER", "Server type"); +DEFINE_int32(server_core_limit, -1, "Limit on server cores to use"); // Client config +DEFINE_string(rpc_type, "UNARY", "Type of RPC: UNARY or STREAMING"); DEFINE_int32(outstanding_rpcs_per_channel, 1, "Number of outstanding rpcs per channel"); DEFINE_int32(client_channels, 1, "Number of client channels"); @@ -75,6 +74,8 @@ DEFINE_double(determ_load, -1.0, "Deterministic offered load (qps)"); DEFINE_double(pareto_base, -1.0, "Pareto base interarrival time (us)"); DEFINE_double(pareto_alpha, -1.0, "Pareto alpha value"); +DEFINE_int32(client_core_limit, -1, "Limit on client cores to use"); + DEFINE_bool(secure_test, false, "Run a secure test"); DEFINE_bool(quit, false, "Quit the workers"); @@ -158,10 +159,18 @@ static void QpsDriver() { client_config.mutable_histogram_params()->set_max_possible( Histogram::default_max_possible()); + if (FLAGS_client_core_limit > 0) { + client_config.set_core_limit(FLAGS_client_core_limit); + } + ServerConfig server_config; server_config.set_server_type(server_type); server_config.set_async_server_threads(FLAGS_async_server_threads); + if (FLAGS_server_core_limit > 0) { + server_config.set_core_limit(FLAGS_server_core_limit); + } + if (FLAGS_secure_test) { // Set up security params SecurityParams security; diff --git a/test/cpp/qps/qps_worker.cc b/test/cpp/qps/qps_worker.cc index 47b51fe3dd..9442017ddf 100644 --- a/test/cpp/qps/qps_worker.cc +++ b/test/cpp/qps/qps_worker.cc @@ -47,6 +47,7 @@ #include <grpc++/server_builder.h> #include <grpc/grpc.h> #include <grpc/support/alloc.h> +#include <grpc/support/cpu.h> #include <grpc/support/histogram.h> #include <grpc/support/host_port.h> #include <grpc/support/log.h> @@ -83,15 +84,10 @@ static std::unique_ptr<Client> CreateClient(const ClientConfig& config) { abort(); } -static void LimitCores(int cores) {} - static std::unique_ptr<Server> CreateServer(const ServerConfig& config) { gpr_log(GPR_INFO, "Starting server of type %s", ServerType_Name(config.server_type()).c_str()); - if (config.core_limit() > 0) { - LimitCores(config.core_limit()); - } switch (config.server_type()) { case ServerType::SYNC_SERVER: return CreateSynchronousServer(config); @@ -138,6 +134,12 @@ class WorkerServiceImpl GRPC_FINAL : public WorkerService::Service { return ret; } + Status CoreCount(ServerContext* ctx, const CoreRequest*, + CoreResponse* resp) GRPC_OVERRIDE { + resp->set_cores(gpr_cpu_num_cores()); + return Status::OK; + } + Status QuitWorker(ServerContext* ctx, const Void*, Void*) GRPC_OVERRIDE { InstanceGuard g(this); if (!g.Acquired()) { diff --git a/test/cpp/qps/server.h b/test/cpp/qps/server.h index 196fdac8f3..94a6f8acfa 100644 --- a/test/cpp/qps/server.h +++ b/test/cpp/qps/server.h @@ -34,14 +34,16 @@ #ifndef TEST_QPS_SERVER_H #define TEST_QPS_SERVER_H -#include <grpc/support/cpu.h> #include <grpc++/security/server_credentials.h> +#include <grpc/support/cpu.h> +#include <vector> +#include "src/proto/grpc/testing/control.grpc.pb.h" +#include "src/proto/grpc/testing/messages.grpc.pb.h" #include "test/core/end2end/data/ssl_test_data.h" #include "test/core/util/port.h" +#include "test/cpp/qps/limit_cores.h" #include "test/cpp/qps/timer.h" -#include "src/proto/grpc/testing/messages.grpc.pb.h" -#include "src/proto/grpc/testing/control.grpc.pb.h" namespace grpc { namespace testing { @@ -49,8 +51,10 @@ namespace testing { class Server { public: explicit Server(const ServerConfig& config) : timer_(new Timer) { + cores_ = LimitCores(config.core_list().data(), config.core_list_size()); if (config.port()) { port_ = config.port(); + } else { port_ = grpc_pick_unused_port_or_die(); } @@ -86,7 +90,7 @@ class Server { } int port() const { return port_; } - int cores() const { return gpr_cpu_num_cores(); } + int cores() const { return cores_; } static std::shared_ptr<ServerCredentials> CreateServerCredentials( const ServerConfig& config) { if (config.has_security_params()) { @@ -103,6 +107,7 @@ class Server { private: int port_; + int cores_; std::unique_ptr<Timer> timer_; }; |