Fix some stalling in bm_fullstack_trickle, also switch to synthetic time

test/cpp/microbenchmarks/bm_fullstack_trickle.cc

@@ -29,6 +29,7 @@
 extern "C" {
 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
 #include "src/core/ext/transport/chttp2/transport/internal.h"
+#include "src/core/lib/iomgr/timer_manager.h"
 #include "test/core/util/trickle_endpoint.h"
 }
 
@@ -45,6 +46,22 @@ DEFINE_int32(warmup_max_time_seconds, 10,
 namespace grpc {
 namespace testing {
 
+gpr_atm g_now_us = 0;
+
+static gpr_timespec fake_now(gpr_clock_type clock_type) {
+  gpr_timespec t;
+  gpr_atm now = gpr_atm_no_barrier_load(&g_now_us);
+  t.tv_sec = now / GPR_US_PER_SEC;
+  t.tv_nsec = (now % GPR_US_PER_SEC) * GPR_NS_PER_US;
+  t.clock_type = clock_type;
+  return t;
+}
+
+static void inc_time() {
+  gpr_atm_no_barrier_fetch_add(&g_now_us, 100);
+  grpc_timer_manager_tick();
+}
+
 static void* tag(intptr_t x) { return reinterpret_cast<void*>(x); }
 
 template <class A0>
@@ -158,6 +175,7 @@ class TrickledCHTTP2 : public EndpointPairFixture {
 
   void Step(bool update_stats) {
     grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+    inc_time();
     size_t client_backlog =
         grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.client);
     size_t server_backlog =
@@ -213,8 +231,7 @@ static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok,
   while (true) {
     fixture->Log(iteration);
     switch (fixture->cq()->AsyncNext(
-        t, ok, gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
-                            gpr_time_from_micros(100, GPR_TIMESPAN)))) {
+        t, ok, gpr_inf_past(GPR_CLOCK_MONOTONIC))) {
       case CompletionQueue::TIMEOUT:
         fixture->Step(iteration != -1);
         break;
@@ -289,9 +306,15 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
       inner_loop(false);
     }
     response_rw.Finish(Status::OK, tag(1));
-    need_tags = (1 << 0) | (1 << 1);
+    grpc::Status status;
+    request_rw->Finish(&status, tag(2));
+    need_tags = (1 << 0) | (1 << 1) | (1 << 2);
     while (need_tags) {
       TrickleCQNext(fixture.get(), &t, &ok, -1);
+      if (t == tag(0) && ok) {
+        request_rw->Read(&recv_response, tag(0));
+        continue;
+      }
       int i = (int)(intptr_t)t;
       GPR_ASSERT(need_tags & (1 << i));
       need_tags &= ~(1 << i);
@@ -419,8 +442,12 @@ BENCHMARK(BM_PumpUnbalancedUnary_Trickle)->Apply(UnaryTrickleArgs);
 }
 }
 
+extern "C" gpr_timespec (*gpr_now_impl)(gpr_clock_type clock_type);
+
 int main(int argc, char** argv) {
   ::benchmark::Initialize(&argc, argv);
   ::grpc::testing::InitTest(&argc, &argv, false);
+  grpc_timer_manager_set_threading(false);
+  gpr_now_impl = ::grpc::testing::fake_now;
   ::benchmark::RunSpecifiedBenchmarks();
 }

test/cpp/microbenchmarks/fullstack_fixtures.h

@@ -85,7 +85,7 @@ class FullstackFixture : public BaseFixture {
   }
 
   virtual ~FullstackFixture() {
-    server_->Shutdown();
+    server_->Shutdown(gpr_inf_past(GPR_CLOCK_MONOTONIC));
     cq_->Shutdown();
     void* tag;
     bool ok;
@@ -212,7 +212,7 @@ class EndpointPairFixture : public BaseFixture {
   }
 
   virtual ~EndpointPairFixture() {
-    server_->Shutdown();
+    server_->Shutdown(gpr_inf_past(GPR_CLOCK_MONOTONIC));
     cq_->Shutdown();
     void* tag;
     bool ok;