Make core limitation work for both client and server so that we can run tests on the same

machine if desired. The core_list flags to qps_driver are comma-separated lists of
core numbers.

Makefile

@@ -3596,6 +3596,7 @@ LIBQPS_SRC = \
     $(GENDIR)/src/proto/grpc/testing/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc \
     test/cpp/qps/client_async.cc \
     test/cpp/qps/client_sync.cc \
+    test/cpp/qps/coresched.cc \
     test/cpp/qps/driver.cc \
     test/cpp/qps/perf_db_client.cc \
     test/cpp/qps/qps_worker.cc \
@@ -3650,6 +3651,7 @@ endif
 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/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.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/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.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/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/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.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/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.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/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc
+$(OBJDIR)/$(CONFIG)/test/cpp/qps/coresched.o: $(GENDIR)/src/proto/grpc/testing/messages.pb.cc $(GENDIR)/src/proto/grpc/testing/messages.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/payloads.pb.cc $(GENDIR)/src/proto/grpc/testing/payloads.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/stats.pb.cc $(GENDIR)/src/proto/grpc/testing/stats.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/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.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/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.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/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/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.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/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.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/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/control.pb.cc $(GENDIR)/src/proto/grpc/testing/control.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/services.pb.cc $(GENDIR)/src/proto/grpc/testing/services.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/perf_db.pb.cc $(GENDIR)/src/proto/grpc/testing/perf_db.grpc.pb.cc
@@ -12812,6 +12814,7 @@ test/cpp/interop/server.cc: $(OPENSSL_DEP)
 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/coresched.cc: $(OPENSSL_DEP)
 test/cpp/qps/driver.cc: $(OPENSSL_DEP)
 test/cpp/qps/perf_db_client.cc: $(OPENSSL_DEP)
 test/cpp/qps/qps_worker.cc: $(OPENSSL_DEP)

build.yaml

@@ -760,6 +760,7 @@ libs:
   language: c++
   headers:
   - test/cpp/qps/client.h
+  - test/cpp/qps/coresched.h
   - test/cpp/qps/driver.h
   - test/cpp/qps/histogram.h
   - test/cpp/qps/interarrival.h
@@ -779,6 +780,7 @@ libs:
   - src/proto/grpc/testing/perf_db.proto
   - test/cpp/qps/client_async.cc
   - test/cpp/qps/client_sync.cc
+  - test/cpp/qps/coresched.cc
   - test/cpp/qps/driver.cc
   - test/cpp/qps/perf_db_client.cc
   - test/cpp/qps/qps_worker.cc

src/proto/grpc/testing/control.proto

@@ -106,6 +106,9 @@ 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;
 }
 
 message ClientStatus { ClientStats stats = 1; }
@@ -135,6 +138,9 @@ message ServerConfig {
   // restrict core usage, currently unused
   int32 core_limit = 8;
   PayloadConfig payload_config = 9;
+
+  // Specify the cores we should run the server on, if desired
+  repeated int32 core_list = 10;
 }
 
 message ServerArgs {

test/cpp/qps/client.h

@@ -36,9 +36,13 @@
 
 #include <condition_variable>
 #include <mutex>
+#include <vector>
+
+#include <grpc/support/log.h>
 
 #include "test/cpp/qps/histogram.h"
 #include "test/cpp/qps/interarrival.h"
+#include "test/cpp/qps/coresched.h"
 #include "test/cpp/qps/timer.h"
 #include "test/cpp/util/create_test_channel.h"
 #include "src/proto/grpc/testing/payloads.grpc.pb.h"
@@ -72,6 +76,15 @@ class Client {
       : channels_(config.client_channels()),
         timer_(new Timer),
         interarrival_timer_() {
+    int clsize = 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));
+      }
+      LimitCores(core_list);
+    }
+
     for (int i = 0; i < config.client_channels(); i++) {
       channels_[i].init(config.server_targets(i % config.server_targets_size()),
                         config);

test/cpp/qps/coresched.cc

@@ -0,0 +1,72 @@
+/*
+ *
+ * Copyright 2015, 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/coresched.h"
+
+#include <vector>
+#include <grpc/support/cpu.h>
+#include <grpc/support/log.h>
+#include <grpc/support/port_platform.h>
+
+namespace grpc {
+namespace testing {
+
+#ifdef GPR_CPU_LINUX
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE
+#endif
+#include <sched.h>
+int LimitCores(std::vector<int> cores) {
+  size_t num_cores = static_cast<size_t>(gpr_cpu_num_cores());
+  if (num_cores > cores.size()) {
+    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_S(cores[i], size, cpup);
+    }
+    GPR_ASSERT(sched_setaffinity(0, size, cpup) == 0);
+    CPU_FREE(cpup);
+    return cores.size();
+  } else {
+    return num_cores;
+  }
+}
+#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/coresched.h

@@ -0,0 +1,45 @@
+/*
+ *
+ * Copyright 2015, 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_CORESCHED_H
+#define TEST_QPS_CORESCHED_H
+
+#include <vector>
+
+namespace grpc {
+namespace testing {
+int LimitCores(std::vector<int> core_vec);
+}  // namespace testing
+}  // namespace grpc
+
+#endif // TEST_QPS_CORESCHED_H

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_string(server_core_list, "", "Comma-separated list of cores for server");
 
 // 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");
@@ -73,6 +72,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_string(client_core_list, "", "Comma-separated list of cores for client");
+
 DEFINE_bool(secure_test, false, "Run a secure test");
 
 using grpc::testing::ClientConfig;
@@ -86,6 +87,22 @@ using grpc::testing::SecurityParams;
 namespace grpc {
 namespace testing {
 
+static std::vector<int> IntParse(const std::string& s) {
+  size_t pos = 0;
+  std::vector<int> res;
+  while (pos < s.size()) {
+    size_t comma = s.find(',', pos);
+    if (comma == std::string::npos) {
+      res.push_back(std::stoi(s.substr(pos)));
+      break;
+    } else {
+      res.push_back(std::stoi(s.substr(pos, comma-pos), nullptr));
+      pos = comma + 1;
+    }
+  }
+  return res;
+}
+
 static void QpsDriver() {
   RpcType rpc_type;
   GPR_ASSERT(RpcType_Parse(FLAGS_rpc_type, &rpc_type));
@@ -142,11 +159,25 @@ static void QpsDriver() {
   client_config.mutable_histogram_params()->set_max_possible(
       Histogram::default_max_possible());
 
+  if (FLAGS_client_core_list.size() > 0) {
+    auto v = IntParse(FLAGS_client_core_list);
+    for (size_t i=0; i<v.size(); i++) {
+      client_config.add_core_list(v[i]);
+    }
+  }
+
   ServerConfig server_config;
   server_config.set_server_type(server_type);
   server_config.set_host("localhost");
   server_config.set_async_server_threads(FLAGS_async_server_threads);
 
+  if (FLAGS_server_core_list.size() > 0) {
+    auto v = IntParse(FLAGS_server_core_list);
+    for (size_t i=0; i<v.size(); i++) {
+      server_config.add_core_list(v[i]);
+    }
+  }
+
   if (FLAGS_secure_test) {
     // Set up security params
     SecurityParams security;

test/cpp/qps/qps_worker.cc

@@ -76,12 +76,7 @@ static std::unique_ptr<Client> CreateClient(const ClientConfig& config) {
   abort();
 }
 
-static void LimitCores(int cores) {}
-
 static std::unique_ptr<Server> CreateServer(const ServerConfig& config) {
-  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

@@ -34,11 +34,13 @@
 #ifndef TEST_QPS_SERVER_H
 #define TEST_QPS_SERVER_H
 
+#include <vector>
 #include <grpc/support/cpu.h>
 #include <grpc++/security/server_credentials.h>
 
 #include "test/core/end2end/data/ssl_test_data.h"
 #include "test/core/util/port.h"
+#include "test/cpp/qps/coresched.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"
@@ -49,6 +51,16 @@ namespace testing {
 class Server {
  public:
   explicit Server(const ServerConfig& config) : timer_(new Timer) {
+    int clsize = 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 {
@@ -87,7 +99,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()) {
@@ -104,6 +116,7 @@ class Server {
 
  private:
   int port_;
+  int cores_;
   std::unique_ptr<Timer> timer_;
 };
 

tools/run_tests/sources_and_headers.json

@@ -4267,6 +4267,7 @@
       "src/proto/grpc/testing/stats.grpc.pb.h", 
       "src/proto/grpc/testing/stats.pb.h", 
       "test/cpp/qps/client.h", 
+      "test/cpp/qps/coresched.h", 
       "test/cpp/qps/driver.h", 
       "test/cpp/qps/histogram.h", 
       "test/cpp/qps/interarrival.h", 
@@ -4284,6 +4285,8 @@
       "test/cpp/qps/client.h", 
       "test/cpp/qps/client_async.cc", 
       "test/cpp/qps/client_sync.cc", 
+      "test/cpp/qps/coresched.cc", 
+      "test/cpp/qps/coresched.h", 
       "test/cpp/qps/driver.cc", 
       "test/cpp/qps/driver.h", 
       "test/cpp/qps/histogram.h", 

vsprojects/vcxproj/qps/qps.vcxproj

@@ -148,6 +148,7 @@
 
   <ItemGroup>
     <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\client.h" />
+    <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\coresched.h" />
     <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\driver.h" />
     <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\histogram.h" />
     <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\interarrival.h" />
@@ -212,6 +213,8 @@
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\client_sync.cc">
     </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\coresched.cc">
+    </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\driver.cc">
     </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\perf_db_client.cc">

vsprojects/vcxproj/qps/qps.vcxproj.filters

@@ -25,6 +25,9 @@
     <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\client_sync.cc">
       <Filter>test\cpp\qps</Filter>
     </ClCompile>
+    <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\coresched.cc">
+      <Filter>test\cpp\qps</Filter>
+    </ClCompile>
     <ClCompile Include="$(SolutionDir)\..\test\cpp\qps\driver.cc">
       <Filter>test\cpp\qps</Filter>
     </ClCompile>
@@ -54,6 +57,9 @@
     <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\client.h">
       <Filter>test\cpp\qps</Filter>
     </ClInclude>
+    <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\coresched.h">
+      <Filter>test\cpp\qps</Filter>
+    </ClInclude>
     <ClInclude Include="$(SolutionDir)\..\test\cpp\qps\driver.h">
       <Filter>test\cpp\qps</Filter>
     </ClInclude>