1 files changed, 47 insertions, 19 deletions

diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
index 59fb29dd60..25d243a104 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
@@ -21,17 +21,16 @@
 #include <benchmark/benchmark.h>
 #include <gflags/gflags.h>
 #include <fstream>
+
+#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 "src/core/lib/profiling/timers.h"
-#include "src/cpp/client/create_channel_internal.h"
 #include "src/proto/grpc/testing/echo.grpc.pb.h"
+#include "test/core/util/trickle_endpoint.h"
 #include "test/cpp/microbenchmarks/fullstack_context_mutators.h"
 #include "test/cpp/microbenchmarks/fullstack_fixtures.h"
 #include "test/cpp/util/test_config.h"
-extern "C" {
-#include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
-#include "src/core/ext/transport/chttp2/transport/internal.h"
-#include "test/core/util/trickle_endpoint.h"
-}
 
 DEFINE_bool(log, false, "Log state to CSV files");
 DEFINE_int32(
@@ -46,6 +45,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>
@@ -128,15 +143,18 @@ class TrickledCHTTP2 : public EndpointPairFixture {
         client->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr,
         server->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr,
         server->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr,
-        client->flow_control.remote_window, server->flow_control.remote_window,
-        client->flow_control.announced_window,
-        server->flow_control.announced_window,
-        client_stream ? client_stream->flow_control.remote_window_delta : -1,
-        server_stream ? server_stream->flow_control.remote_window_delta : -1,
-        client_stream ? client_stream->flow_control.local_window_delta : -1,
-        server_stream ? server_stream->flow_control.local_window_delta : -1,
-        client_stream ? client_stream->flow_control.announced_window_delta : -1,
-        server_stream ? server_stream->flow_control.announced_window_delta : -1,
+        client->flow_control->remote_window_,
+        server->flow_control->remote_window_,
+        client->flow_control->announced_window_,
+        server->flow_control->announced_window_,
+        client_stream ? client_stream->flow_control->remote_window_delta_ : -1,
+        server_stream ? server_stream->flow_control->remote_window_delta_ : -1,
+        client_stream ? client_stream->flow_control->local_window_delta_ : -1,
+        server_stream ? server_stream->flow_control->local_window_delta_ : -1,
+        client_stream ? client_stream->flow_control->announced_window_delta_
+                      : -1,
+        server_stream ? server_stream->flow_control->announced_window_delta_
+                      : -1,
         client->settings[GRPC_PEER_SETTINGS]
                         [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
         client->settings[GRPC_LOCAL_SETTINGS]
@@ -159,6 +177,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,9 +232,8 @@ static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok,
                           int64_t iteration) {
   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)))) {
+    switch (
+        fixture->cq()->AsyncNext(t, ok, gpr_inf_past(GPR_CLOCK_MONOTONIC))) {
       case CompletionQueue::TIMEOUT:
         fixture->Step(iteration != -1);
         break;
@@ -290,9 +308,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);
@@ -420,8 +444,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();
 }