Microbenchmark for measuring impact of multiple threads calling cq_next

9 files changed, 333 insertions, 4 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index e3b2558ad2..d260a3797b 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -626,6 +626,9 @@ if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 add_dependencies(buildtests_cxx bm_cq)
 endif()
 if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
+add_dependencies(buildtests_cxx bm_cq_multiple_threads)
+endif()
+if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 add_dependencies(buildtests_cxx bm_error)
 endif()
 if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
@@ -8982,6 +8985,49 @@ endif (gRPC_BUILD_TESTS)
 if (gRPC_BUILD_TESTS)
 if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
 
+add_executable(bm_cq_multiple_threads
+  test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
+  third_party/googletest/src/gtest-all.cc
+)
+
+
+target_include_directories(bm_cq_multiple_threads
+  PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}
+  PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/include
+  PRIVATE ${BORINGSSL_ROOT_DIR}/include
+  PRIVATE ${PROTOBUF_ROOT_DIR}/src
+  PRIVATE ${BENCHMARK_ROOT_DIR}/include
+  PRIVATE ${ZLIB_ROOT_DIR}
+  PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/third_party/zlib
+  PRIVATE ${CARES_BUILD_INCLUDE_DIR}
+  PRIVATE ${CARES_INCLUDE_DIR}
+  PRIVATE ${CARES_PLATFORM_INCLUDE_DIR}
+  PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/third_party/cares/cares
+  PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/third_party/gflags/include
+  PRIVATE third_party/googletest/include
+  PRIVATE third_party/googletest
+  PRIVATE ${_gRPC_PROTO_GENS_DIR}
+)
+
+target_link_libraries(bm_cq_multiple_threads
+  ${_gRPC_PROTOBUF_LIBRARIES}
+  ${_gRPC_ALLTARGETS_LIBRARIES}
+  grpc_benchmark
+  benchmark
+  grpc++_test_util
+  grpc_test_util
+  grpc++
+  grpc
+  gpr_test_util
+  gpr
+  ${_gRPC_GFLAGS_LIBRARIES}
+)
+
+endif()
+endif (gRPC_BUILD_TESTS)
+if (gRPC_BUILD_TESTS)
+if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
+
 add_executable(bm_error
   test/cpp/microbenchmarks/bm_error.cc
   third_party/googletest/src/gtest-all.cc
diff --git a/Makefile b/Makefile
index 7499dbbf50..0d52cb4005 100644
--- a/Makefile
+++ b/Makefile
@@ -1105,6 +1105,7 @@ bm_chttp2_hpack: $(BINDIR)/$(CONFIG)/bm_chttp2_hpack
 bm_chttp2_transport: $(BINDIR)/$(CONFIG)/bm_chttp2_transport
 bm_closure: $(BINDIR)/$(CONFIG)/bm_closure
 bm_cq: $(BINDIR)/$(CONFIG)/bm_cq
+bm_cq_multiple_threads: $(BINDIR)/$(CONFIG)/bm_cq_multiple_threads
 bm_error: $(BINDIR)/$(CONFIG)/bm_error
 bm_fullstack_streaming_ping_pong: $(BINDIR)/$(CONFIG)/bm_fullstack_streaming_ping_pong
 bm_fullstack_streaming_pump: $(BINDIR)/$(CONFIG)/bm_fullstack_streaming_pump
@@ -1530,6 +1531,7 @@ buildtests_cxx: privatelibs_cxx \
   $(BINDIR)/$(CONFIG)/bm_chttp2_transport \
   $(BINDIR)/$(CONFIG)/bm_closure \
   $(BINDIR)/$(CONFIG)/bm_cq \
+  $(BINDIR)/$(CONFIG)/bm_cq_multiple_threads \
   $(BINDIR)/$(CONFIG)/bm_error \
   $(BINDIR)/$(CONFIG)/bm_fullstack_streaming_ping_pong \
   $(BINDIR)/$(CONFIG)/bm_fullstack_streaming_pump \
@@ -1648,6 +1650,7 @@ buildtests_cxx: privatelibs_cxx \
   $(BINDIR)/$(CONFIG)/bm_chttp2_transport \
   $(BINDIR)/$(CONFIG)/bm_closure \
   $(BINDIR)/$(CONFIG)/bm_cq \
+  $(BINDIR)/$(CONFIG)/bm_cq_multiple_threads \
   $(BINDIR)/$(CONFIG)/bm_error \
   $(BINDIR)/$(CONFIG)/bm_fullstack_streaming_ping_pong \
   $(BINDIR)/$(CONFIG)/bm_fullstack_streaming_pump \
@@ -1997,6 +2000,8 @@ test_cxx: buildtests_cxx
 	$(Q) $(BINDIR)/$(CONFIG)/bm_closure || ( echo test bm_closure failed ; exit 1 )
 	$(E) "[RUN]     Testing bm_cq"
 	$(Q) $(BINDIR)/$(CONFIG)/bm_cq || ( echo test bm_cq failed ; exit 1 )
+	$(E) "[RUN]     Testing bm_cq_multiple_threads"
+	$(Q) $(BINDIR)/$(CONFIG)/bm_cq_multiple_threads || ( echo test bm_cq_multiple_threads failed ; exit 1 )
 	$(E) "[RUN]     Testing bm_error"
 	$(Q) $(BINDIR)/$(CONFIG)/bm_error || ( echo test bm_error failed ; exit 1 )
 	$(E) "[RUN]     Testing bm_fullstack_streaming_ping_pong"
@@ -13363,6 +13368,50 @@ endif
 endif
 
 
+BM_CQ_MULTIPLE_THREADS_SRC = \
+    test/cpp/microbenchmarks/bm_cq_multiple_threads.cc \
+
+BM_CQ_MULTIPLE_THREADS_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(BM_CQ_MULTIPLE_THREADS_SRC))))
+ifeq ($(NO_SECURE),true)
+
+# You can't build secure targets if you don't have OpenSSL.
+
+$(BINDIR)/$(CONFIG)/bm_cq_multiple_threads: openssl_dep_error
+
+else
+
+
+
+
+ifeq ($(NO_PROTOBUF),true)
+
+# You can't build the protoc plugins or protobuf-enabled targets if you don't have protobuf 3.0.0+.
+
+$(BINDIR)/$(CONFIG)/bm_cq_multiple_threads: protobuf_dep_error
+
+else
+
+$(BINDIR)/$(CONFIG)/bm_cq_multiple_threads: $(PROTOBUF_DEP) $(BM_CQ_MULTIPLE_THREADS_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_benchmark.a $(LIBDIR)/$(CONFIG)/libbenchmark.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
+	$(E) "[LD]      Linking $@"
+	$(Q) mkdir -p `dirname $@`
+	$(Q) $(LDXX) $(LDFLAGS) $(BM_CQ_MULTIPLE_THREADS_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_benchmark.a $(LIBDIR)/$(CONFIG)/libbenchmark.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBSXX) $(LDLIBS_PROTOBUF) $(LDLIBS) $(LDLIBS_SECURE) $(GTEST_LIB) -o $(BINDIR)/$(CONFIG)/bm_cq_multiple_threads
+
+endif
+
+endif
+
+$(BM_CQ_MULTIPLE_THREADS_OBJS): CPPFLAGS += -Ithird_party/benchmark/include -DHAVE_POSIX_REGEX
+$(OBJDIR)/$(CONFIG)/test/cpp/microbenchmarks/bm_cq_multiple_threads.o:  $(LIBDIR)/$(CONFIG)/libgrpc_benchmark.a $(LIBDIR)/$(CONFIG)/libbenchmark.a $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
+
+deps_bm_cq_multiple_threads: $(BM_CQ_MULTIPLE_THREADS_OBJS:.o=.dep)
+
+ifneq ($(NO_SECURE),true)
+ifneq ($(NO_DEPS),true)
+-include $(BM_CQ_MULTIPLE_THREADS_OBJS:.o=.dep)
+endif
+endif
+
+
 BM_ERROR_SRC = \
     test/cpp/microbenchmarks/bm_error.cc \
 
diff --git a/build.yaml b/build.yaml
index 8aff16854c..c88b1161d2 100644
--- a/build.yaml
+++ b/build.yaml
@@ -3205,6 +3205,27 @@ targets:
   - mac
   - linux
   - posix
+- name: bm_cq_multiple_threads
+  build: test
+  language: c++
+  src:
+  - test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
+  deps:
+  - grpc_benchmark
+  - benchmark
+  - grpc++_test_util
+  - grpc_test_util
+  - grpc++
+  - grpc
+  - gpr_test_util
+  - gpr
+  args:
+  - --benchmark_min_time=0
+  defaults: benchmark
+  platforms:
+  - mac
+  - linux
+  - posix
 - name: bm_error
   build: test
   language: c++
diff --git a/src/core/lib/iomgr/ev_posix.c b/src/core/lib/iomgr/ev_posix.c
index b5be5504b9..13409a4de8 100644
--- a/src/core/lib/iomgr/ev_posix.c
+++ b/src/core/lib/iomgr/ev_posix.c
@@ -111,6 +111,12 @@ static void try_engine(const char *engine) {
   }
 }
 
+/* This should be used for testing purposes ONLY */
+void grpc_set_event_engine_test_only(
+    const grpc_event_engine_vtable *ev_engine) {
+  g_event_engine = ev_engine;
+}
+
 /* Call this only after calling grpc_event_engine_init() */
 const char *grpc_get_poll_strategy_name() { return g_poll_strategy_name; }
 
diff --git a/src/core/lib/iomgr/ev_posix.h b/src/core/lib/iomgr/ev_posix.h
index 1a9e5c115a..becc4d359e 100644
--- a/src/core/lib/iomgr/ev_posix.h
+++ b/src/core/lib/iomgr/ev_posix.h
@@ -183,4 +183,7 @@ void grpc_pollset_set_del_fd(grpc_exec_ctx *exec_ctx,
 typedef int (*grpc_poll_function_type)(struct pollfd *, nfds_t, int);
 extern grpc_poll_function_type grpc_poll_function;
 
+/* This should be used for testing purposes ONLY */
+void grpc_set_event_engine_test_only(const grpc_event_engine_vtable *);
+
 #endif /* GRPC_CORE_LIB_IOMGR_EV_POSIX_H */
diff --git a/test/cpp/microbenchmarks/BUILD b/test/cpp/microbenchmarks/BUILD
index 38619666dc..cae3fa1a14 100644
--- a/test/cpp/microbenchmarks/BUILD
+++ b/test/cpp/microbenchmarks/BUILD
@@ -32,16 +32,25 @@ licenses(["notice"])  # 3-clause BSD
 cc_test(
     name = "noop-benchmark",
     srcs = ["noop-benchmark.cc"],
-    deps = ["//external:benchmark"],
     linkopts = ["-pthread"],
+    deps = ["//external:benchmark"],
 )
 
 cc_library(
     name = "helpers",
     srcs = ["helpers.cc"],
-    hdrs = ["helpers.h", "fullstack_fixtures.h", "fullstack_context_mutators.h"],
-    deps = ["//:grpc++", "//external:benchmark", "//test/core/util:grpc_test_util", "//src/proto/grpc/testing:echo_proto"],
+    hdrs = [
+        "fullstack_context_mutators.h",
+        "fullstack_fixtures.h",
+        "helpers.h",
+    ],
     linkopts = ["-pthread"],
+    deps = [
+        "//:grpc++",
+        "//external:benchmark",
+        "//src/proto/grpc/testing:echo_proto",
+        "//test/core/util:grpc_test_util",
+    ],
 )
 
 cc_test(
@@ -57,6 +66,12 @@ cc_test(
 )
 
 cc_test(
+    name = "bm_cq_multiple_threads",
+    srcs = ["bm_cq_multiple_threads.cc"],
+    deps = [":helpers"],
+)
+
+cc_test(
     name = "bm_error",
     srcs = ["bm_error.cc"],
     deps = [":helpers"],
@@ -66,8 +81,8 @@ cc_test(
     name = "bm_fullstack_streaming_ping_pong",
     srcs = ["bm_fullstack_streaming_ping_pong.cc"],
     deps = [":helpers"],
+)
 
-    )
 cc_test(
     name = "bm_fullstack_streaming_pump",
     srcs = ["bm_fullstack_streaming_pump.cc"],
diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
new file mode 100644
index 0000000000..f68baad2b0
--- /dev/null
+++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
@@ -0,0 +1,146 @@
+/*
+ *
+ * Copyright 2017, 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 <benchmark/benchmark.h>
+#include <string.h>
+#include <atomic>
+
+#include <grpc/grpc.h>
+#include "test/cpp/microbenchmarks/helpers.h"
+
+extern "C" {
+#include "src/core/lib/iomgr/ev_posix.h"
+#include "src/core/lib/iomgr/port.h"
+#include "src/core/lib/surface/completion_queue.h"
+}
+
+struct grpc_pollset {
+  gpr_mu mu;
+};
+
+namespace grpc {
+namespace testing {
+
+static void* make_tag(int i) { return (void*)(intptr_t)i; }
+static grpc_completion_queue* g_cq;
+static grpc_event_engine_vtable g_vtable;
+
+static __thread int g_thread_idx;
+static __thread grpc_cq_completion g_cq_completion;
+
+static void pollset_shutdown(grpc_exec_ctx* exec_ctx, grpc_pollset* ps,
+                             grpc_closure* closure) {
+  grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_NONE);
+}
+
+static void pollset_init(grpc_pollset* ps, gpr_mu** mu) {
+  gpr_mu_init(&ps->mu);
+  *mu = &ps->mu;
+}
+
+static void pollset_destroy(grpc_pollset* ps) { gpr_mu_destroy(&ps->mu); }
+
+static grpc_error* pollset_kick(grpc_pollset* p, grpc_pollset_worker* worker) {
+  return GRPC_ERROR_NONE;
+}
+
+/* Callback when the tag is dequeued from the completion queue. Does nothing */
+static void cq_done_cb(grpc_exec_ctx* exec_ctx, void* done_arg,
+                       grpc_cq_completion* cq_completion) {}
+
+/* Queues a completion tag. ZERO polling overhead */
+static grpc_error* pollset_work(grpc_exec_ctx* exec_ctx, grpc_pollset* ps,
+                                grpc_pollset_worker** worker, gpr_timespec now,
+                                gpr_timespec deadline) {
+  gpr_mu_unlock(&ps->mu);
+  grpc_cq_end_op(exec_ctx, g_cq, make_tag(g_thread_idx), GRPC_ERROR_NONE,
+                 cq_done_cb, NULL, &g_cq_completion);
+  grpc_exec_ctx_flush(exec_ctx);
+  gpr_mu_lock(&ps->mu);
+  return GRPC_ERROR_NONE;
+}
+
+static void init_engine_vtable() {
+  memset(&g_vtable, 0, sizeof(g_vtable));
+
+  g_vtable.pollset_size = sizeof(grpc_pollset);
+  g_vtable.pollset_init = pollset_init;
+  g_vtable.pollset_shutdown = pollset_shutdown;
+  g_vtable.pollset_destroy = pollset_destroy;
+  g_vtable.pollset_work = pollset_work;
+  g_vtable.pollset_kick = pollset_kick;
+}
+
+static void setup() {
+  grpc_init();
+  init_engine_vtable();
+  grpc_set_event_engine_test_only(&g_vtable);
+
+  g_cq = grpc_completion_queue_create(NULL);
+}
+
+static void BM_Cq_Throughput(benchmark::State& state) {
+  TrackCounters track_counters;
+  gpr_timespec deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC);
+
+  if (state.thread_index == 0) {
+    setup();
+  }
+
+  while (state.KeepRunning()) {
+    g_thread_idx = state.thread_index;
+    void* dummy_tag = make_tag(g_thread_idx);
+    grpc_cq_begin_op(g_cq, dummy_tag);
+    grpc_completion_queue_next(g_cq, deadline, NULL);
+  }
+
+  if (state.thread_index == 0) {
+    grpc_completion_queue_shutdown(g_cq);
+    grpc_completion_queue_destroy(g_cq);
+
+    state.SetItemsProcessed(state.iterations());
+  }
+
+  track_counters.Finish(state);
+}
+
+BENCHMARK(BM_Cq_Throughput)->Threads(1);
+BENCHMARK(BM_Cq_Throughput)->Threads(2);
+BENCHMARK(BM_Cq_Throughput)->Threads(4);
+BENCHMARK(BM_Cq_Throughput)->Threads(8);
+BENCHMARK(BM_Cq_Throughput)->Threads(16);
+
+}  // namespace testing
+}  // namespace grpc
+
+BENCHMARK_MAIN();
diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json
index be1d8768bd..412eada7c0 100644
--- a/tools/run_tests/generated/sources_and_headers.json
+++ b/tools/run_tests/generated/sources_and_headers.json
@@ -2566,6 +2566,27 @@
     "headers": [], 
     "is_filegroup": false, 
     "language": "c++", 
+    "name": "bm_cq_multiple_threads", 
+    "src": [
+      "test/cpp/microbenchmarks/bm_cq_multiple_threads.cc"
+    ], 
+    "third_party": false, 
+    "type": "target"
+  }, 
+  {
+    "deps": [
+      "benchmark", 
+      "gpr", 
+      "gpr_test_util", 
+      "grpc", 
+      "grpc++", 
+      "grpc++_test_util", 
+      "grpc_benchmark", 
+      "grpc_test_util"
+    ], 
+    "headers": [], 
+    "is_filegroup": false, 
+    "language": "c++", 
     "name": "bm_error", 
     "src": [
       "test/cpp/microbenchmarks/bm_error.cc"
diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json
index 7d2c51a7d4..bda62794d2 100644
--- a/tools/run_tests/generated/tests.json
+++ b/tools/run_tests/generated/tests.json
@@ -2752,6 +2752,28 @@
     "flaky": false, 
     "gtest": false, 
     "language": "c++", 
+    "name": "bm_cq_multiple_threads", 
+    "platforms": [
+      "linux", 
+      "mac", 
+      "posix"
+    ]
+  }, 
+  {
+    "args": [
+      "--benchmark_min_time=0"
+    ], 
+    "ci_platforms": [
+      "linux", 
+      "mac", 
+      "posix"
+    ], 
+    "cpu_cost": 1.0, 
+    "exclude_configs": [], 
+    "exclude_iomgrs": [], 
+    "flaky": false, 
+    "gtest": false, 
+    "language": "c++", 
     "name": "bm_error", 
     "platforms": [
       "linux",