merge

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)

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

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(

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

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);
 }
 

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; }

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);
+        }
     }
 }

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;
+}

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);
 }

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 {

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_;

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));

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

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

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).

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;

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 {

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_;
 };
 

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", 

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">

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>