Clean-up core list usage and make it possible to reset the core list

src/proto/grpc/testing/control.proto

@@ -134,8 +134,7 @@ 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
+  // payload config, used in generic server
-  int32 core_limit = 8;
   PayloadConfig payload_config = 9;
 
   // Specify the cores we should run the server on, if desired

test/cpp/qps/client.h

@@ -324,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) {
+    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_);

test/cpp/qps/limit_cores.cc

@@ -46,23 +46,31 @@ namespace testing {
 #define _GNU_SOURCE
 #endif
 #include <sched.h>
-int LimitCores(std::vector<int> cores) {
+int LimitCores(const int *cores, int cores_size) {
-  size_t num_cores = static_cast<size_t>(gpr_cpu_num_cores());
+  int num_cores = gpr_cpu_num_cores();
-  if (num_cores > cores.size()) {
+  int cores_set = 0;
-    cpu_set_t *cpup = CPU_ALLOC(num_cores);
-    GPR_ASSERT(cpup);
-    size_t size = CPU_ALLOC_SIZE(num_cores);
-    CPU_ZERO_S(size, cpup);
 
-    for (size_t i = 0; i < cores.size(); i++) {
+  cpu_set_t *cpup = CPU_ALLOC(num_cores);
-      CPU_SET_S(cores[i], size, cpup);
+  GPR_ASSERT(cpup);
+  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++;
+      }
     }
-    GPR_ASSERT(sched_setaffinity(0, size, cpup) == 0);
-    CPU_FREE(cpup);
-    return cores.size();
   } else {
-    return num_cores;
+    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

test/cpp/qps/limit_cores.h

@@ -38,7 +38,10 @@
 
 namespace grpc {
 namespace testing {
-int LimitCores(std::vector<int> core_vec);
+// LimitCores takes array and size arguments (instead of vector) for more 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
 

test/cpp/qps/qps_worker.cc

@@ -87,10 +87,6 @@ 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);

test/cpp/qps/server.h

@@ -51,18 +51,10 @@ namespace testing {
 class Server {
  public:
   explicit Server(const ServerConfig& config) : timer_(new Timer) {
-    int clsize = config.core_list_size();
+    cores_ = LimitCores(config.core_list().data(), config.core_list_size());
-    if (clsize > 0) {
-      std::vector<int> core_list;
-      for (int i = 0; i < clsize; i++) {
-        core_list.push_back(config.core_list(i));
-      }
-      cores_ = LimitCores(core_list);
-    } else {
-      cores_ = gpr_cpu_num_cores();
-    }
     if (config.port()) {
       port_ = config.port();
+
     } else {
       port_ = grpc_pick_unused_port_or_die();
     }