Merge github.com:grpc/grpc into cpparena

9 files changed, 115 insertions, 42 deletions

diff --git a/test/cpp/microbenchmarks/bm_call_create.cc b/test/cpp/microbenchmarks/bm_call_create.cc
index 5a209f58c1..3199d2d42d 100644
--- a/test/cpp/microbenchmarks/bm_call_create.cc
+++ b/test/cpp/microbenchmarks/bm_call_create.cc
@@ -46,14 +46,14 @@
 
 extern "C" {
 #include "src/core/ext/filters/client_channel/client_channel.h"
+#include "src/core/ext/filters/deadline/deadline_filter.h"
+#include "src/core/ext/filters/http/client/http_client_filter.h"
+#include "src/core/ext/filters/http/message_compress/message_compress_filter.h"
+#include "src/core/ext/filters/http/server/http_server_filter.h"
 #include "src/core/ext/filters/load_reporting/load_reporting_filter.h"
+#include "src/core/ext/filters/message_size/message_size_filter.h"
 #include "src/core/lib/channel/channel_stack.h"
-#include "src/core/lib/channel/compress_filter.h"
 #include "src/core/lib/channel/connected_channel.h"
-#include "src/core/lib/channel/deadline_filter.h"
-#include "src/core/lib/channel/http_client_filter.h"
-#include "src/core/lib/channel/http_server_filter.h"
-#include "src/core/lib/channel/message_size_filter.h"
 #include "src/core/lib/profiling/timers.h"
 #include "src/core/lib/surface/channel.h"
 #include "src/core/lib/transport/transport_impl.h"
@@ -119,7 +119,7 @@ template <class Fixture>
 static void BM_CallCreateDestroy(benchmark::State &state) {
   TrackCounters track_counters;
   Fixture fixture;
-  grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
+  grpc_completion_queue *cq = grpc_completion_queue_create_for_next(NULL);
   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
   void *method_hdl =
       grpc_channel_register_call(fixture.channel(), "/foo/bar", NULL, NULL);
@@ -461,7 +461,7 @@ BENCHMARK_TEMPLATE(BM_IsolatedFilter, DummyFilter, NoOp);
 BENCHMARK_TEMPLATE(BM_IsolatedFilter, DummyFilter, SendEmptyMetadata);
 typedef Fixture<&grpc_client_channel_filter, 0> ClientChannelFilter;
 BENCHMARK_TEMPLATE(BM_IsolatedFilter, ClientChannelFilter, NoOp);
-typedef Fixture<&grpc_compress_filter, CHECKS_NOT_LAST> CompressFilter;
+typedef Fixture<&grpc_message_compress_filter, CHECKS_NOT_LAST> CompressFilter;
 BENCHMARK_TEMPLATE(BM_IsolatedFilter, CompressFilter, NoOp);
 BENCHMARK_TEMPLATE(BM_IsolatedFilter, CompressFilter, SendEmptyMetadata);
 typedef Fixture<&grpc_client_deadline_filter, CHECKS_NOT_LAST>
@@ -576,7 +576,7 @@ class IsolatedCallFixture : public TrackCounters {
                                                   GRPC_CLIENT_CHANNEL);
       grpc_exec_ctx_finish(&exec_ctx);
     }
-    cq_ = grpc_completion_queue_create(NULL);
+    cq_ = grpc_completion_queue_create_for_next(NULL);
   }
 
   void Finish(benchmark::State &state) {
diff --git a/test/cpp/microbenchmarks/bm_chttp2_transport.cc b/test/cpp/microbenchmarks/bm_chttp2_transport.cc
index 8c5413b5fd..c89f349ca7 100644
--- a/test/cpp/microbenchmarks/bm_chttp2_transport.cc
+++ b/test/cpp/microbenchmarks/bm_chttp2_transport.cc
@@ -569,17 +569,12 @@ static void BM_TransportStreamRecv(benchmark::State &state) {
         grpc_closure_sched(exec_ctx, c.get(), GRPC_ERROR_NONE);
         return;
       }
-    } while (grpc_byte_stream_next(exec_ctx, recv_stream,
+    } while (grpc_byte_stream_next(exec_ctx, recv_stream, &recv_slice,
                                    recv_stream->length - received,
-                                   drain_continue.get()) &&
-             GRPC_ERROR_NONE ==
-                 grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice) &&
-             (received += GRPC_SLICE_LENGTH(recv_slice),
-              grpc_slice_unref_internal(exec_ctx, recv_slice), true));
+                                   drain_continue.get()));
   });
 
   drain_continue = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
-    grpc_byte_stream_pull(exec_ctx, recv_stream, &recv_slice);
     received += GRPC_SLICE_LENGTH(recv_slice);
     grpc_slice_unref_internal(exec_ctx, recv_slice);
     grpc_closure_run(exec_ctx, drain.get(), GRPC_ERROR_NONE);
diff --git a/test/cpp/microbenchmarks/bm_cq.cc b/test/cpp/microbenchmarks/bm_cq.cc
index 38ac9d2705..8b26bf977c 100644
--- a/test/cpp/microbenchmarks/bm_cq.cc
+++ b/test/cpp/microbenchmarks/bm_cq.cc
@@ -62,7 +62,8 @@ BENCHMARK(BM_CreateDestroyCpp);
 static void BM_CreateDestroyCpp2(benchmark::State& state) {
   TrackCounters track_counters;
   while (state.KeepRunning()) {
-    grpc_completion_queue* core_cq = grpc_completion_queue_create(NULL);
+    grpc_completion_queue* core_cq =
+        grpc_completion_queue_create_for_next(NULL);
     CompletionQueue cq(core_cq);
   }
   track_counters.Finish(state);
@@ -72,7 +73,9 @@ BENCHMARK(BM_CreateDestroyCpp2);
 static void BM_CreateDestroyCore(benchmark::State& state) {
   TrackCounters track_counters;
   while (state.KeepRunning()) {
-    grpc_completion_queue_destroy(grpc_completion_queue_create(NULL));
+    // TODO: sreek Templatize this benchmark and pass completion type and
+    // polling type as parameters
+    grpc_completion_queue_destroy(grpc_completion_queue_create_for_next(NULL));
   }
   track_counters.Finish(state);
 }
@@ -108,7 +111,8 @@ BENCHMARK(BM_Pass1Cpp);
 
 static void BM_Pass1Core(benchmark::State& state) {
   TrackCounters track_counters;
-  grpc_completion_queue* cq = grpc_completion_queue_create(NULL);
+  // TODO: sreek Templatize this benchmark and pass polling_type as a param
+  grpc_completion_queue* cq = grpc_completion_queue_create_for_next(NULL);
   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
   while (state.KeepRunning()) {
     grpc_cq_completion completion;
@@ -126,7 +130,8 @@ BENCHMARK(BM_Pass1Core);
 
 static void BM_Pluck1Core(benchmark::State& state) {
   TrackCounters track_counters;
-  grpc_completion_queue* cq = grpc_completion_queue_create(NULL);
+  // TODO: sreek Templatize this benchmark and pass polling_type as a param
+  grpc_completion_queue* cq = grpc_completion_queue_create_for_pluck(NULL);
   gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
   while (state.KeepRunning()) {
     grpc_cq_completion completion;
@@ -144,7 +149,8 @@ BENCHMARK(BM_Pluck1Core);
 
 static void BM_EmptyCore(benchmark::State& state) {
   TrackCounters track_counters;
-  grpc_completion_queue* cq = grpc_completion_queue_create(NULL);
+  // TODO: sreek Templatize this benchmark and pass polling_type as a param
+  grpc_completion_queue* cq = grpc_completion_queue_create_for_next(NULL);
   gpr_timespec deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC);
   while (state.KeepRunning()) {
     grpc_completion_queue_next(cq, deadline, NULL);
diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
index 8627463204..9d7f65d292 100644
--- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
+++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
@@ -108,7 +108,7 @@ static void setup() {
   init_engine_vtable();
   grpc_set_event_engine_test_only(&g_vtable);
 
-  g_cq = grpc_completion_queue_create(NULL);
+  g_cq = grpc_completion_queue_create_for_next(NULL);
 }
 
 static void teardown() {
diff --git a/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc b/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc
index c536e15a2c..fd2210c474 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_streaming_ping_pong.cc
@@ -436,6 +436,18 @@ BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, InProcessCHTTP2, NoOpMutator,
 BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, TCP, NoOpMutator, NoOpMutator)
     ->Range(0, 128 * 1024 * 1024);
 
+BENCHMARK_TEMPLATE(BM_StreamingPingPong, MinInProcessCHTTP2, NoOpMutator,
+                   NoOpMutator)
+    ->Apply(StreamingPingPongArgs);
+BENCHMARK_TEMPLATE(BM_StreamingPingPong, MinTCP, NoOpMutator, NoOpMutator)
+    ->Apply(StreamingPingPongArgs);
+
+BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, MinInProcessCHTTP2, NoOpMutator,
+                   NoOpMutator)
+    ->Range(0, 128 * 1024 * 1024);
+BENCHMARK_TEMPLATE(BM_StreamingPingPongMsgs, MinTCP, NoOpMutator, NoOpMutator)
+    ->Range(0, 128 * 1024 * 1024);
+
 // Generate Args for StreamingPingPongWithCoalescingApi benchmarks. Currently
 // generates args for only "small streams" (i.e streams with 0, 1 or 2 messages)
 static void StreamingPingPongWithCoalescingApiArgs(
@@ -459,6 +471,9 @@ static void StreamingPingPongWithCoalescingApiArgs(
 BENCHMARK_TEMPLATE(BM_StreamingPingPongWithCoalescingApi, InProcessCHTTP2,
                    NoOpMutator, NoOpMutator)
     ->Apply(StreamingPingPongWithCoalescingApiArgs);
+BENCHMARK_TEMPLATE(BM_StreamingPingPongWithCoalescingApi, MinInProcessCHTTP2,
+                   NoOpMutator, NoOpMutator)
+    ->Apply(StreamingPingPongWithCoalescingApiArgs);
 
 }  // namespace testing
 }  // namespace grpc
diff --git a/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc b/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc
index 5c1eb1165b..47705d3031 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_streaming_pump.cc
@@ -189,6 +189,14 @@ BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, SockPair)
     ->Range(0, 128 * 1024 * 1024);
 BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, InProcessCHTTP2)
     ->Range(0, 128 * 1024 * 1024);
+BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinTCP)->Arg(0);
+BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinUDS)->Arg(0);
+BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinSockPair)->Arg(0);
+BENCHMARK_TEMPLATE(BM_PumpStreamClientToServer, MinInProcessCHTTP2)->Arg(0);
+BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinTCP)->Arg(0);
+BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinUDS)->Arg(0);
+BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinSockPair)->Arg(0);
+BENCHMARK_TEMPLATE(BM_PumpStreamServerToClient, MinInProcessCHTTP2)->Arg(0);
 
 }  // namespace testing
 }  // namespace grpc
diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
index c563f28b55..a5cfeb4f95 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
@@ -53,7 +53,8 @@ static void* tag(intptr_t x) { return reinterpret_cast<void*>(x); }
 class TrickledCHTTP2 : public EndpointPairFixture {
  public:
   TrickledCHTTP2(Service* service, size_t megabits_per_second)
-      : EndpointPairFixture(service, MakeEndpoints(megabits_per_second)) {}
+      : EndpointPairFixture(service, MakeEndpoints(megabits_per_second),
+                            FixtureConfiguration()) {}
 
   void AddToLabel(std::ostream& out, benchmark::State& state) {
     out << " writes/iter:"
diff --git a/test/cpp/microbenchmarks/bm_fullstack_unary_ping_pong.cc b/test/cpp/microbenchmarks/bm_fullstack_unary_ping_pong.cc
index 615b05b7c7..7524751fbc 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_unary_ping_pong.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_unary_ping_pong.cc
@@ -141,12 +141,21 @@ static void SweepSizesArgs(benchmark::internal::Benchmark* b) {
 
 BENCHMARK_TEMPLATE(BM_UnaryPingPong, TCP, NoOpMutator, NoOpMutator)
     ->Apply(SweepSizesArgs);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, MinTCP, NoOpMutator, NoOpMutator)
+    ->Apply(SweepSizesArgs);
 BENCHMARK_TEMPLATE(BM_UnaryPingPong, UDS, NoOpMutator, NoOpMutator)
     ->Args({0, 0});
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, MinUDS, NoOpMutator, NoOpMutator)
+    ->Args({0, 0});
 BENCHMARK_TEMPLATE(BM_UnaryPingPong, SockPair, NoOpMutator, NoOpMutator)
     ->Args({0, 0});
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, MinSockPair, NoOpMutator, NoOpMutator)
+    ->Args({0, 0});
 BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2, NoOpMutator, NoOpMutator)
     ->Apply(SweepSizesArgs);
+BENCHMARK_TEMPLATE(BM_UnaryPingPong, MinInProcessCHTTP2, NoOpMutator,
+                   NoOpMutator)
+    ->Apply(SweepSizesArgs);
 BENCHMARK_TEMPLATE(BM_UnaryPingPong, InProcessCHTTP2,
                    Client_AddMetadata<RandomBinaryMetadata<10>, 1>, NoOpMutator)
     ->Args({0, 0});
diff --git a/test/cpp/microbenchmarks/fullstack_fixtures.h b/test/cpp/microbenchmarks/fullstack_fixtures.h
index acc56bf39b..f129ede26a 100644
--- a/test/cpp/microbenchmarks/fullstack_fixtures.h
+++ b/test/cpp/microbenchmarks/fullstack_fixtures.h
@@ -61,29 +61,33 @@ extern "C" {
 namespace grpc {
 namespace testing {
 
-static void ApplyCommonServerBuilderConfig(ServerBuilder* b) {
-  b->SetMaxReceiveMessageSize(INT_MAX);
-  b->SetMaxSendMessageSize(INT_MAX);
-}
+class FixtureConfiguration {
+ public:
+  virtual void ApplyCommonChannelArguments(ChannelArguments* c) const {
+    c->SetInt(GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH, INT_MAX);
+    c->SetInt(GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, INT_MAX);
+  }
 
-static void ApplyCommonChannelArguments(ChannelArguments* c) {
-  c->SetInt(GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH, INT_MAX);
-  c->SetInt(GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, INT_MAX);
-}
+  virtual void ApplyCommonServerBuilderConfig(ServerBuilder* b) const {
+    b->SetMaxReceiveMessageSize(INT_MAX);
+    b->SetMaxSendMessageSize(INT_MAX);
+  }
+};
 
 class BaseFixture : public TrackCounters {};
 
 class FullstackFixture : public BaseFixture {
  public:
-  FullstackFixture(Service* service, const grpc::string& address) {
+  FullstackFixture(Service* service, const FixtureConfiguration& config,
+                   const grpc::string& address) {
     ServerBuilder b;
     b.AddListeningPort(address, InsecureServerCredentials());
     cq_ = b.AddCompletionQueue(true);
     b.RegisterService(service);
-    ApplyCommonServerBuilderConfig(&b);
+    config.ApplyCommonServerBuilderConfig(&b);
     server_ = b.BuildAndStart();
     ChannelArguments args;
-    ApplyCommonChannelArguments(&args);
+    config.ApplyCommonChannelArguments(&args);
     channel_ = CreateCustomChannel(address, InsecureChannelCredentials(), args);
   }
 
@@ -107,7 +111,9 @@ class FullstackFixture : public BaseFixture {
 
 class TCP : public FullstackFixture {
  public:
-  TCP(Service* service) : FullstackFixture(service, MakeAddress()) {}
+  TCP(Service* service, const FixtureConfiguration& fixture_configuration =
+                            FixtureConfiguration())
+      : FullstackFixture(service, fixture_configuration, MakeAddress()) {}
 
  private:
   static grpc::string MakeAddress() {
@@ -120,7 +126,9 @@ class TCP : public FullstackFixture {
 
 class UDS : public FullstackFixture {
  public:
-  UDS(Service* service) : FullstackFixture(service, MakeAddress()) {}
+  UDS(Service* service, const FixtureConfiguration& fixture_configuration =
+                            FixtureConfiguration())
+      : FullstackFixture(service, fixture_configuration, MakeAddress()) {}
 
  private:
   static grpc::string MakeAddress() {
@@ -134,12 +142,13 @@ class UDS : public FullstackFixture {
 
 class EndpointPairFixture : public BaseFixture {
  public:
-  EndpointPairFixture(Service* service, grpc_endpoint_pair endpoints)
+  EndpointPairFixture(Service* service, grpc_endpoint_pair endpoints,
+                      const FixtureConfiguration& fixture_configuration)
       : endpoint_pair_(endpoints) {
     ServerBuilder b;
     cq_ = b.AddCompletionQueue(true);
     b.RegisterService(service);
-    ApplyCommonServerBuilderConfig(&b);
+    fixture_configuration.ApplyCommonServerBuilderConfig(&b);
     server_ = b.BuildAndStart();
 
     grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
@@ -169,7 +178,7 @@ class EndpointPairFixture : public BaseFixture {
     {
       ChannelArguments args;
       args.SetString(GRPC_ARG_DEFAULT_AUTHORITY, "test.authority");
-      ApplyCommonChannelArguments(&args);
+      fixture_configuration.ApplyCommonChannelArguments(&args);
 
       grpc_channel_args c_args = args.c_channel_args();
       client_transport_ =
@@ -211,15 +220,19 @@ class EndpointPairFixture : public BaseFixture {
 
 class SockPair : public EndpointPairFixture {
  public:
-  SockPair(Service* service)
+  SockPair(Service* service, const FixtureConfiguration& fixture_configuration =
+                                 FixtureConfiguration())
       : EndpointPairFixture(service,
-                            grpc_iomgr_create_endpoint_pair("test", NULL)) {}
+                            grpc_iomgr_create_endpoint_pair("test", NULL),
+                            fixture_configuration) {}
 };
 
 class InProcessCHTTP2 : public EndpointPairFixture {
  public:
-  InProcessCHTTP2(Service* service)
-      : EndpointPairFixture(service, MakeEndpoints()) {}
+  InProcessCHTTP2(Service* service,
+                  const FixtureConfiguration& fixture_configuration =
+                      FixtureConfiguration())
+      : EndpointPairFixture(service, MakeEndpoints(), fixture_configuration) {}
 
   void AddToLabel(std::ostream& out, benchmark::State& state) {
     EndpointPairFixture::AddToLabel(out, state);
@@ -238,6 +251,32 @@ class InProcessCHTTP2 : public EndpointPairFixture {
   }
 };
 
+////////////////////////////////////////////////////////////////////////////////
+// Minimal stack fixtures
+
+class MinStackConfiguration : public FixtureConfiguration {
+  void ApplyCommonChannelArguments(ChannelArguments* a) const override {
+    a->SetInt(GRPC_ARG_MINIMAL_STACK, 1);
+    FixtureConfiguration::ApplyCommonChannelArguments(a);
+  }
+
+  void ApplyCommonServerBuilderConfig(ServerBuilder* b) const override {
+    b->AddChannelArgument(GRPC_ARG_MINIMAL_STACK, 1);
+    FixtureConfiguration::ApplyCommonServerBuilderConfig(b);
+  }
+};
+
+template <class Base>
+class MinStackize : public Base {
+ public:
+  MinStackize(Service* service) : Base(service, MinStackConfiguration()) {}
+};
+
+typedef MinStackize<TCP> MinTCP;
+typedef MinStackize<UDS> MinUDS;
+typedef MinStackize<SockPair> MinSockPair;
+typedef MinStackize<InProcessCHTTP2> MinInProcessCHTTP2;
+
 }  // namespace testing
 }  // namespace grpc