diff --git a/Makefile b/Makefile index d1417acf07a3e309b6e937a14c6ed92ac98ba90a..e089a7adfc55a92f7d4bd8ea669416150ceba900 100644 --- a/Makefile +++ b/Makefile @@ -3782,6 +3782,7 @@ LIBQPS_SRC = \ test/cpp/qps/client_async.cc \ test/cpp/qps/client_sync.cc \ test/cpp/qps/driver.cc \ + test/cpp/qps/limit_cores.cc \ test/cpp/qps/perf_db_client.cc \ test/cpp/qps/qps_worker.cc \ test/cpp/qps/report.cc \ @@ -3836,6 +3837,7 @@ endif $(OBJDIR)/$(CONFIG)/test/cpp/qps/client_async.o: $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/qps/client_sync.o: $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/qps/driver.o: $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc +$(OBJDIR)/$(CONFIG)/test/cpp/qps/limit_cores.o: $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/qps/perf_db_client.o: $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/qps/qps_worker.o: $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc $(OBJDIR)/$(CONFIG)/test/cpp/qps/report.o: $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc @@ -13036,6 +13038,7 @@ test/cpp/interop/server_helper.cc: $(OPENSSL_DEP) test/cpp/qps/client_async.cc: $(OPENSSL_DEP) test/cpp/qps/client_sync.cc: $(OPENSSL_DEP) test/cpp/qps/driver.cc: $(OPENSSL_DEP) +test/cpp/qps/limit_cores.cc: $(OPENSSL_DEP) test/cpp/qps/perf_db_client.cc: $(OPENSSL_DEP) test/cpp/qps/qps_worker.cc: $(OPENSSL_DEP) test/cpp/qps/report.cc: $(OPENSSL_DEP) diff --git a/build.yaml b/build.yaml index b519f63da829f49c071c8e8d6c8197dea32718eb..188de19e717243a77db3e512a2c55cbe6d4e6da2 100644 --- a/build.yaml +++ b/build.yaml @@ -834,6 +834,7 @@ libs: - test/cpp/qps/driver.h - test/cpp/qps/histogram.h - test/cpp/qps/interarrival.h + - test/cpp/qps/limit_cores.h - test/cpp/qps/perf_db_client.h - test/cpp/qps/qps_worker.h - test/cpp/qps/report.h @@ -851,6 +852,7 @@ libs: - test/cpp/qps/client_async.cc - test/cpp/qps/client_sync.cc - test/cpp/qps/driver.cc + - test/cpp/qps/limit_cores.cc - test/cpp/qps/perf_db_client.cc - test/cpp/qps/qps_worker.cc - test/cpp/qps/report.cc diff --git a/include/grpc/census.h b/include/grpc/census.h index 5faab385b6eda2d6a9fb5fadff6728f27478fae7..6313b196f224238f0fcbb9283116950f40f79d49 100644 --- a/include/grpc/census.h +++ b/include/grpc/census.h @@ -505,6 +505,7 @@ typedef struct census_view census_view; @return A new census view */ + /* TODO(aveitch): consider if context is the right argument type to pass in tags. */ CENSUS_API census_view *census_view_create( diff --git a/include/grpc/impl/codegen/port_platform.h b/include/grpc/impl/codegen/port_platform.h index 30edbc7251f53bf947dff17663da5c86183a66a1..d5294b2efa8b5f8e68873a4fb98e7b21fa1d8f39 100644 --- a/include/grpc/impl/codegen/port_platform.h +++ b/include/grpc/impl/codegen/port_platform.h @@ -153,19 +153,14 @@ #if __GLIBC_PREREQ(2, 10) #define GPR_LINUX_SOCKETUTILS 1 #endif -#if __GLIBC_PREREQ(2, 17) -#define GPR_LINUX_ENV 1 -#endif #endif +#define GPR_LINUX_ENV 1 #ifndef GPR_LINUX_EVENTFD #define GPR_POSIX_NO_SPECIAL_WAKEUP_FD 1 #endif #ifndef GPR_LINUX_SOCKETUTILS #define GPR_POSIX_SOCKETUTILS #endif -#ifndef GPR_LINUX_ENV -#define GPR_POSIX_ENV 1 -#endif #define GPR_POSIX_FILE 1 #define GPR_POSIX_STRING 1 #define GPR_POSIX_SUBPROCESS 1 diff --git a/src/core/support/env_linux.c b/src/core/support/env_linux.c index 2e03365e3383a2899235b3cbb60a5e5ecf6c8556..b5832a591772905819d56df5a760071ed0d8fe51 100644 --- a/src/core/support/env_linux.c +++ b/src/core/support/env_linux.c @@ -1,6 +1,6 @@ /* * - * Copyright 2015, Google Inc. + * Copyright 2015-2016, Google Inc. * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -49,8 +49,28 @@ #include "src/core/support/string.h" +/* Declare weak symbols for versions of secure_getenv that *may* be + * on a users machine. Older libc's call this __secure_getenv, even + * older don't support the functionality. + * + * If a symbol is not present, these will be equal to NULL. + */ +char *__attribute__((weak)) secure_getenv(const char *name); +char *__attribute__((weak)) __secure_getenv(const char *name); + char *gpr_getenv(const char *name) { - char *result = secure_getenv(name); + static char *(*getenv_func)(const char *) = secure_getenv; + /* Check to see which getenv variant is supported (go from most + * to least secure) */ + if (getenv_func == NULL) { + getenv_func = __secure_getenv; + if (getenv_func == NULL) { + gpr_log(GPR_DEBUG, + "No secure_getenv. Please consider upgrading your libc."); + getenv_func = getenv; + } + } + char *result = getenv_func(name); return result == NULL ? result : gpr_strdup(result); } diff --git a/src/csharp/Grpc.Core/Internal/PlatformApis.cs b/src/csharp/Grpc.Core/Internal/PlatformApis.cs index d71e7eccdd64dba12cfc8905ac08cf3e9335a09b..f0c5b0f63f61e1d15d0d04137f279c64477d1e5d 100644 --- a/src/csharp/Grpc.Core/Internal/PlatformApis.cs +++ b/src/csharp/Grpc.Core/Internal/PlatformApis.cs @@ -49,6 +49,7 @@ namespace Grpc.Core.Internal static readonly bool isLinux; static readonly bool isMacOSX; static readonly bool isWindows; + static readonly bool isMono; static PlatformApis() { @@ -58,6 +59,7 @@ namespace Grpc.Core.Internal isMacOSX = (platform == PlatformID.Unix && GetUname() == "Darwin"); isLinux = (platform == PlatformID.Unix && !isMacOSX); isWindows = (platform == PlatformID.Win32NT || platform == PlatformID.Win32S || platform == PlatformID.Win32Windows); + isMono = Type.GetType("Mono.Runtime") != null; } public static bool IsLinux @@ -75,6 +77,11 @@ namespace Grpc.Core.Internal get { return isWindows; } } + public static bool IsMono + { + get { return isMono; } + } + public static bool Is64Bit { get { return IntPtr.Size == 8; } diff --git a/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs b/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs index e614cab6abf19a28ad32cd14a240237dab5bc1a9..95a8797e3eefde879cc645f798b87de8d7cc9380 100644 --- a/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs +++ b/src/csharp/Grpc.Core/Internal/UnmanagedLibrary.cs @@ -91,6 +91,10 @@ namespace Grpc.Core.Internal { if (PlatformApis.IsLinux) { + if (PlatformApis.IsMono) + { + return Mono.dlsym(this.handle, symbolName); + } return Linux.dlsym(this.handle, symbolName); } if (PlatformApis.IsMacOSX) @@ -122,6 +126,10 @@ namespace Grpc.Core.Internal } if (PlatformApis.IsLinux) { + if (PlatformApis.IsMono) + { + return Mono.dlopen(libraryPath, RTLD_GLOBAL + RTLD_LAZY); + } return Linux.dlopen(libraryPath, RTLD_GLOBAL + RTLD_LAZY); } if (PlatformApis.IsMacOSX) @@ -154,5 +162,21 @@ namespace Grpc.Core.Internal [DllImport("libSystem.dylib")] internal static extern IntPtr dlsym(IntPtr handle, string symbol); } + + /// <summary> + /// On Linux systems, using using dlopen and dlsym results in + /// DllNotFoundException("libdl.so not found") if libc6-dev + /// is not installed. As a workaround, we load symbols for + /// dlopen and dlsym from the current process as on Linux + /// Mono sure is linked against these symbols. + /// </summary> + private static class Mono + { + [DllImport("__Internal")] + internal static extern IntPtr dlopen(string filename, int flags); + + [DllImport("__Internal")] + internal static extern IntPtr dlsym(IntPtr handle, string symbol); + } } } diff --git a/src/proto/grpc/testing/control.proto b/src/proto/grpc/testing/control.proto index 2f352e652f323945a613be6a5bb233a8c47b71df..8278836468fd8bf2c8878acba7d9d79d68bfc4de 100644 --- a/src/proto/grpc/testing/control.proto +++ b/src/proto/grpc/testing/control.proto @@ -107,6 +107,10 @@ message ClientConfig { LoadParams load_params = 10; PayloadConfig payload_config = 11; HistogramParams histogram_params = 12; + + // Specify the cores we should run the client on, if desired + repeated int32 core_list = 13; + int32 core_limit = 14; } message ClientStatus { ClientStats stats = 1; } @@ -131,9 +135,13 @@ message ServerConfig { int32 port = 4; // Only for async server. Number of threads used to serve the requests. int32 async_server_threads = 7; - // restrict core usage, currently unused + // Specify the number of cores to limit server to, if desired int32 core_limit = 8; + // payload config, used in generic server PayloadConfig payload_config = 9; + + // Specify the cores we should run the server on, if desired + repeated int32 core_list = 10; } message ServerArgs { @@ -147,6 +155,14 @@ message ServerStatus { ServerStats stats = 1; // the port bound by the server int32 port = 2; - // Number of cores on the server. See gpr_cpu_num_cores. + // Number of cores available to the server int32 cores = 3; } + +message CoreRequest { +} + +message CoreResponse { + // Number of cores available on the server + int32 cores = 1; +} diff --git a/src/proto/grpc/testing/services.proto b/src/proto/grpc/testing/services.proto index af285ceab8130056f02a91bd4945c422a85c4251..4c8e32bb8f36ef0a3d0d003322af6cecdd75eca9 100644 --- a/src/proto/grpc/testing/services.proto +++ b/src/proto/grpc/testing/services.proto @@ -1,4 +1,4 @@ -// Copyright 2015, Google Inc. +// Copyright 2015-2016, Google Inc. // All rights reserved. // // Redistribution and use in source and binary forms, with or without @@ -62,4 +62,7 @@ service WorkerService { // and once the shutdown has finished, the OK status is sent to terminate // this RPC. rpc RunClient(stream ClientArgs) returns (stream ClientStatus); + + // Just return the core count - unary call + rpc CoreCount(CoreRequest) returns (CoreResponse); } diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h index 97487fd0b228b8e0d5adfe9d9c726c3ec2b57820..50b2bf25147341155f98386f925610c1f1b8e105 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 4229e1956e6947f92e92f774038495ee830ddc5f..6b6294ba518b32361d91a4898dd2b5c7ee9d9752 100644 --- a/test/cpp/qps/client_async.cc +++ b/test/cpp/qps/client_async.cc @@ -96,7 +96,8 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { std::function< std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>( BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&, - CompletionQueue*)> start_req, + CompletionQueue*)> + start_req, std::function<void(grpc::Status, ResponseType*)> on_done) : ClientRpcContext(channel_id), context_(), @@ -142,7 +143,8 @@ class ClientRpcContextUnaryImpl : public ClientRpcContext { std::function<void(grpc::Status, ResponseType*)> callback_; std::function<std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>>( BenchmarkService::Stub*, grpc::ClientContext*, const RequestType&, - CompletionQueue*)> start_req_; + CompletionQueue*)> + start_req_; grpc::Status status_; double start_; std::unique_ptr<grpc::ClientAsyncResponseReader<ResponseType>> @@ -159,13 +161,15 @@ 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, - std::function<ClientRpcContext*(int, StubType*, - const RequestType&)> setup_ctx, - std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)> - create_stub) + AsyncClient( + const ClientConfig& config, + std::function<ClientRpcContext*(int, StubType*, const RequestType&)> + setup_ctx, + std::function<std::unique_ptr<StubType>(std::shared_ptr<Channel>)> + create_stub) : ClientImpl<StubType, RequestType>(config, create_stub), num_async_threads_(NumThreads(config)), channel_lock_(new std::mutex[config.client_channels()]), @@ -345,11 +349,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; } @@ -407,7 +411,8 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext { std::function<std::unique_ptr< grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>( BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*, - void*)> start_req, + void*)> + start_req, std::function<void(grpc::Status, ResponseType*)> on_done) : ClientRpcContext(channel_id), context_(), @@ -459,10 +464,10 @@ class ClientRpcContextStreamingImpl : public ClientRpcContext { ResponseType response_; bool (ClientRpcContextStreamingImpl::*next_state_)(bool, Histogram*); std::function<void(grpc::Status, ResponseType*)> callback_; - std::function< - std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>( - BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*, - void*)> start_req_; + std::function<std::unique_ptr< + grpc::ClientAsyncReaderWriter<RequestType, ResponseType>>( + BenchmarkService::Stub*, grpc::ClientContext*, CompletionQueue*, void*)> + start_req_; grpc::Status status_; double start_; std::unique_ptr<grpc::ClientAsyncReaderWriter<RequestType, ResponseType>> @@ -506,7 +511,8 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext { int channel_id, grpc::GenericStub* stub, const ByteBuffer& req, std::function<std::unique_ptr<grpc::GenericClientAsyncReaderWriter>( grpc::GenericStub*, grpc::ClientContext*, - const grpc::string& method_name, CompletionQueue*, void*)> start_req, + const grpc::string& method_name, CompletionQueue*, void*)> + start_req, std::function<void(grpc::Status, ByteBuffer*)> on_done) : ClientRpcContext(channel_id), context_(), @@ -563,7 +569,8 @@ class ClientRpcContextGenericStreamingImpl : public ClientRpcContext { std::function<void(grpc::Status, ByteBuffer*)> callback_; std::function<std::unique_ptr<grpc::GenericClientAsyncReaderWriter>( grpc::GenericStub*, grpc::ClientContext*, const grpc::string&, - CompletionQueue*, void*)> start_req_; + CompletionQueue*, void*)> + start_req_; grpc::Status status_; double start_; std::unique_ptr<grpc::GenericClientAsyncReaderWriter> stream_; diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc index 490156aec2cec0d9cf19927bbe050c040b2b9f58..a4154d68e46584123f10da6b8c6f9833c912473d 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,12 @@ std::unique_ptr<ScenarioResult> RunScenario( } } +<<<<<<< HEAD +======= + // Setup the hosts and core counts + auto hosts_cores = get_hosts_and_cores(workers); + +>>>>>>> master // if num_clients is set to <=0, do dynamic sizing: all workers // except for servers are clients if (num_clients <= 0) { @@ -172,18 +214,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 +264,55 @@ 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++) { +<<<<<<< HEAD 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); +>>>>>>> master 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 0000000000000000000000000000000000000000..c2f3ad8fdeb2a214005718ef4e81956c2dd97259 --- /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 0000000000000000000000000000000000000000..5c0d1e315dc7c4cea150dd5f9fb3a7a0310e5971 --- /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 333f4bd7d05779171b2905e4ac222d5b2976fe41..539da1d893040812fafc2815d10e5ec3e226bea5 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 aa3cb6882146ab771b3d00036e2ecb00ed006ffb..ffc8a83fc53a6c530cea56722951e090efaf49b4 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"); using grpc::testing::ClientConfig; @@ -151,10 +152,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 6316605aafd15bef66cf14c9a0858aad157bb714..7e9e05f7ecc5c1c28a0d65c5609c42b3a987643a 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; + } + private: // Protect against multiple clients using this worker at once. class InstanceGuard { diff --git a/test/cpp/qps/server.h b/test/cpp/qps/server.h index 196fdac8f38a0365a6f50d18e41bbbc274dc5c5f..94a6f8acfab9a99ab40d9eae92aec3c8f4bbbeff 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_; }; diff --git a/tools/run_tests/sources_and_headers.json b/tools/run_tests/sources_and_headers.json index 04035276db7dd70e21f9a20a68c6979d960238ee..82747e85e2e892bcd4b815c71760350cd9b9bdab 100644 --- a/tools/run_tests/sources_and_headers.json +++ b/tools/run_tests/sources_and_headers.json @@ -4607,6 +4607,7 @@ "test/cpp/qps/driver.h", "test/cpp/qps/histogram.h", "test/cpp/qps/interarrival.h", + "test/cpp/qps/limit_cores.h", "test/cpp/qps/perf_db_client.h", "test/cpp/qps/qps_worker.h", "test/cpp/qps/report.h", @@ -4625,6 +4626,8 @@ "test/cpp/qps/driver.h", "test/cpp/qps/histogram.h", "test/cpp/qps/interarrival.h", + "test/cpp/qps/limit_cores.cc", + "test/cpp/qps/limit_cores.h", "test/cpp/qps/perf_db_client.cc", "test/cpp/qps/perf_db_client.h", "test/cpp/qps/qps_worker.cc", diff --git a/vsprojects/vcxproj/qps/qps.vcxproj b/vsprojects/vcxproj/qps/qps.vcxproj index 7df2597b609080f6388d3087096bc8c1952597fb..8306e2e235695d84b82adb36297e5e8683c16218 100644 --- a/vsprojects/vcxproj/qps/qps.vcxproj +++ b/vsprojects/vcxproj/qps/qps.vcxproj @@ -151,6 +151,7 @@ <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\driver.h" /> <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\histogram.h" /> <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\interarrival.h" /> + <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\limit_cores.h" /> <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\perf_db_client.h" /> <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\qps_worker.h" /> <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\report.h" /> @@ -214,6 +215,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\driver.cc"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\limit_cores.cc"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\perf_db_client.cc"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\qps_worker.cc"> diff --git a/vsprojects/vcxproj/qps/qps.vcxproj.filters b/vsprojects/vcxproj/qps/qps.vcxproj.filters index 14e18e2d72fbf7f7bde8ff4047df98d601ee64ca..650116a912ea0799ce8951b12ee52f8f7f1ae7cb 100644 --- a/vsprojects/vcxproj/qps/qps.vcxproj.filters +++ b/vsprojects/vcxproj/qps/qps.vcxproj.filters @@ -28,6 +28,9 @@ <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\driver.cc"> <Filter>test\cpp\qps</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\limit_cores.cc"> + <Filter>test\cpp\qps</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\perf_db_client.cc"> <Filter>test\cpp\qps</Filter> </ClCompile> @@ -63,6 +66,9 @@ <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\interarrival.h"> <Filter>test\cpp\qps</Filter> </ClInclude> + <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\limit_cores.h"> + <Filter>test\cpp\qps</Filter> + </ClInclude> <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\perf_db_client.h"> <Filter>test\cpp\qps</Filter> </ClInclude>