Remove trickle benchmarks (#28000)

* Remove trickle benchmarks * Automated change: Fix sanity tests Co-authored-by: ctiller <ctiller@users.noreply.github.com>
3 years ago · 75a54f55e3
parent c0c0b841ea
commit 75a54f55e3
13 changed files with 3 additions and 828 deletions
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@ -2224,7 +2224,6 @@ add_library(grpc_test_util
  test/core/util/test_tcp_server.cc
  test/core/util/tls_utils.cc
  test/core/util/tracer_util.cc
  test/core/util/trickle_endpoint.cc
 )
 set_target_properties(grpc_test_util PROPERTIES
@ -2293,7 +2292,6 @@ add_library(grpc_test_util_unsecure
  test/core/util/test_config.cc
  test/core/util/test_tcp_server.cc
  test/core/util/tracer_util.cc
  test/core/util/trickle_endpoint.cc
 )
 set_target_properties(grpc_test_util_unsecure PROPERTIES
@ -16790,7 +16788,6 @@ if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
    test/core/util/test_config.cc
    test/core/util/test_tcp_server.cc
    test/core/util/tracer_util.cc
    test/core/util/trickle_endpoint.cc
    test/cpp/performance/writes_per_rpc_test.cc
    third_party/googletest/googletest/src/gtest-all.cc
    third_party/googletest/googlemock/src/gmock-all.cc
--- a/build_autogenerated.yaml
+++ b/build_autogenerated.yaml
@ -1581,7 +1581,6 @@ libs:
  - test/core/util/test_tcp_server.h
  - test/core/util/tls_utils.h
  - test/core/util/tracer_util.h
  - test/core/util/trickle_endpoint.h
  src:
  - test/core/util/cmdline.cc
  - test/core/util/fuzzer_util.cc
@ -1605,7 +1604,6 @@ libs:
  - test/core/util/test_tcp_server.cc
  - test/core/util/tls_utils.cc
  - test/core/util/tracer_util.cc
  - test/core/util/trickle_endpoint.cc
  deps:
  - absl/debugging:failure_signal_handler
  - absl/debugging:stacktrace
@ -1637,7 +1635,6 @@ libs:
  - test/core/util/test_config.h
  - test/core/util/test_tcp_server.h
  - test/core/util/tracer_util.h
  - test/core/util/trickle_endpoint.h
  src:
  - test/core/util/cmdline.cc
  - test/core/util/fuzzer_util.cc
@ -1660,7 +1657,6 @@ libs:
  - test/core/util/test_config.cc
  - test/core/util/test_tcp_server.cc
  - test/core/util/tracer_util.cc
  - test/core/util/trickle_endpoint.cc
  deps:
  - absl/debugging:failure_signal_handler
  - absl/debugging:stacktrace
@ -8504,7 +8500,6 @@ targets:
  - test/core/util/test_config.h
  - test/core/util/test_tcp_server.h
  - test/core/util/tracer_util.h
  - test/core/util/trickle_endpoint.h
  src:
  - src/proto/grpc/testing/echo.proto
  - src/proto/grpc/testing/echo_messages.proto
@ -8530,7 +8525,6 @@ targets:
  - test/core/util/test_config.cc
  - test/core/util/test_tcp_server.cc
  - test/core/util/tracer_util.cc
  - test/core/util/trickle_endpoint.cc
  - test/cpp/performance/writes_per_rpc_test.cc
  deps:
  - absl/debugging:failure_signal_handler
--- a/gRPC-Core.podspec
+++ b/gRPC-Core.podspec
@ -2272,9 +2272,7 @@ Pod::Spec.new do |s|
                      'test/core/util/tls_utils.cc',
                      'test/core/util/tls_utils.h',
                      'test/core/util/tracer_util.cc',
-                      'test/core/util/tracer_util.h',
+                      'test/core/util/tracer_util.h'
                      'test/core/util/trickle_endpoint.cc',
                      'test/core/util/trickle_endpoint.h'
  end
  # patch include of openssl to openssl_grpc
--- a/grpc.gyp
+++ b/grpc.gyp
@ -1101,7 +1101,6 @@
        'test/core/util/test_tcp_server.cc',
        'test/core/util/tls_utils.cc',
        'test/core/util/tracer_util.cc',
        'test/core/util/trickle_endpoint.cc',
      ],
    },
    {
@ -1135,7 +1134,6 @@
        'test/core/util/test_config.cc',
        'test/core/util/test_tcp_server.cc',
        'test/core/util/tracer_util.cc',
        'test/core/util/trickle_endpoint.cc',
      ],
    },
    {
--- a/test/core/util/BUILD
+++ b/test/core/util/BUILD
@ -53,7 +53,6 @@ grpc_cc_library(
        "test_config.cc",
        "test_tcp_server.cc",
        "tracer_util.cc",
        "trickle_endpoint.cc",
    ],
    hdrs = [
        "cmdline.h",
@ -76,7 +75,6 @@ grpc_cc_library(
        "test_config.h",
        "test_tcp_server.h",
        "tracer_util.h",
        "trickle_endpoint.h",
    ],
    external_deps = [
        "absl/debugging:failure_signal_handler",
--- a/test/core/util/trickle_endpoint.cc
+++ b/test/core/util/trickle_endpoint.cc
@ -1,207 +0,0 @@
 /*
 *
 * Copyright 2016 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
 #include <inttypes.h>
 #include <string.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 #include "src/core/lib/gpr/useful.h"
 #include "src/core/lib/iomgr/sockaddr.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "test/core/util/passthru_endpoint.h"
 #define WRITE_BUFFER_SIZE (2 * 1024 * 1024)
 typedef struct {
  grpc_endpoint base;
  double bytes_per_second;
  grpc_endpoint* wrapped;
  gpr_timespec last_write;
  gpr_mu mu;
  grpc_slice_buffer write_buffer;
  grpc_slice_buffer writing_buffer;
  grpc_error_handle error;
  bool writing;
  grpc_closure* write_cb;
 } trickle_endpoint;
 static void te_read(grpc_endpoint* ep, grpc_slice_buffer* slices,
                    grpc_closure* cb, bool urgent) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  grpc_endpoint_read(te->wrapped, slices, cb, urgent);
 }
 static void maybe_call_write_cb_locked(trickle_endpoint* te) {
  if (te->write_cb != nullptr &&
      (te->error != GRPC_ERROR_NONE ||
       te->write_buffer.length <= WRITE_BUFFER_SIZE)) {
    grpc_core::ExecCtx::Run(DEBUG_LOCATION, te->write_cb,
                            GRPC_ERROR_REF(te->error));
    te->write_cb = nullptr;
  }
 }
 static void te_write(grpc_endpoint* ep, grpc_slice_buffer* slices,
                     grpc_closure* cb, void* /*arg*/) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  gpr_mu_lock(&te->mu);
  GPR_ASSERT(te->write_cb == nullptr);
  if (te->write_buffer.length == 0) {
    te->last_write = gpr_now(GPR_CLOCK_MONOTONIC);
  }
  for (size_t i = 0; i < slices->count; i++) {
    grpc_slice_buffer_add(&te->write_buffer,
                          grpc_slice_copy(slices->slices[i]));
  }
  te->write_cb = cb;
  maybe_call_write_cb_locked(te);
  gpr_mu_unlock(&te->mu);
 }
 static void te_add_to_pollset(grpc_endpoint* ep, grpc_pollset* pollset) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  grpc_endpoint_add_to_pollset(te->wrapped, pollset);
 }
 static void te_add_to_pollset_set(grpc_endpoint* ep,
                                  grpc_pollset_set* pollset_set) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  grpc_endpoint_add_to_pollset_set(te->wrapped, pollset_set);
 }
 static void te_delete_from_pollset_set(grpc_endpoint* ep,
                                       grpc_pollset_set* pollset_set) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  grpc_endpoint_delete_from_pollset_set(te->wrapped, pollset_set);
 }
 static void te_shutdown(grpc_endpoint* ep, grpc_error_handle why) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  gpr_mu_lock(&te->mu);
  if (te->error == GRPC_ERROR_NONE) {
    te->error = GRPC_ERROR_REF(why);
  }
  maybe_call_write_cb_locked(te);
  gpr_mu_unlock(&te->mu);
  grpc_endpoint_shutdown(te->wrapped, why);
 }
 static void te_destroy(grpc_endpoint* ep) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  grpc_endpoint_destroy(te->wrapped);
  gpr_mu_destroy(&te->mu);
  grpc_slice_buffer_destroy_internal(&te->write_buffer);
  grpc_slice_buffer_destroy_internal(&te->writing_buffer);
  GRPC_ERROR_UNREF(te->error);
  gpr_free(te);
 }
 static absl::string_view te_get_peer(grpc_endpoint* ep) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  return grpc_endpoint_get_peer(te->wrapped);
 }
 static absl::string_view te_get_local_address(grpc_endpoint* ep) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  return grpc_endpoint_get_local_address(te->wrapped);
 }
 static int te_get_fd(grpc_endpoint* ep) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  return grpc_endpoint_get_fd(te->wrapped);
 }
 static bool te_can_track_err(grpc_endpoint* /*ep*/) { return false; }
 static void te_finish_write(void* arg, grpc_error_handle /*error*/) {
  trickle_endpoint* te = static_cast<trickle_endpoint*>(arg);
  gpr_mu_lock(&te->mu);
  te->writing = false;
  grpc_slice_buffer_reset_and_unref(&te->writing_buffer);
  gpr_mu_unlock(&te->mu);
 }
 static const grpc_endpoint_vtable vtable = {te_read,
                                            te_write,
                                            te_add_to_pollset,
                                            te_add_to_pollset_set,
                                            te_delete_from_pollset_set,
                                            te_shutdown,
                                            te_destroy,
                                            te_get_peer,
                                            te_get_local_address,
                                            te_get_fd,
                                            te_can_track_err};
 grpc_endpoint* grpc_trickle_endpoint_create(grpc_endpoint* wrap,
                                            double bytes_per_second) {
  trickle_endpoint* te =
      static_cast<trickle_endpoint*>(gpr_malloc(sizeof(*te)));
  te->base.vtable = &vtable;
  te->wrapped = wrap;
  te->bytes_per_second = bytes_per_second;
  te->write_cb = nullptr;
  gpr_mu_init(&te->mu);
  grpc_slice_buffer_init(&te->write_buffer);
  grpc_slice_buffer_init(&te->writing_buffer);
  te->error = GRPC_ERROR_NONE;
  te->writing = false;
  return &te->base;
 }
 static double ts2dbl(gpr_timespec s) {
  return static_cast<double>(s.tv_sec) + 1e-9 * static_cast<double>(s.tv_nsec);
 }
 size_t grpc_trickle_endpoint_trickle(grpc_endpoint* ep) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  gpr_mu_lock(&te->mu);
  if (!te->writing && te->write_buffer.length > 0) {
    gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
    double elapsed = ts2dbl(gpr_time_sub(now, te->last_write));
    size_t bytes = static_cast<size_t>(te->bytes_per_second * elapsed);
    // gpr_log(GPR_DEBUG, "%lf elapsed --> %" PRIdPTR " bytes", elapsed, bytes);
    if (bytes > 0) {
      grpc_slice_buffer_move_first(&te->write_buffer,
                                   std::min(bytes, te->write_buffer.length),
                                   &te->writing_buffer);
      te->writing = true;
      te->last_write = now;
      grpc_endpoint_write(
          te->wrapped, &te->writing_buffer,
          GRPC_CLOSURE_CREATE(te_finish_write, te, grpc_schedule_on_exec_ctx),
          nullptr);
      maybe_call_write_cb_locked(te);
    }
  }
  size_t backlog = te->write_buffer.length;
  gpr_mu_unlock(&te->mu);
  return backlog;
 }
 size_t grpc_trickle_get_backlog(grpc_endpoint* ep) {
  trickle_endpoint* te = reinterpret_cast<trickle_endpoint*>(ep);
  gpr_mu_lock(&te->mu);
  size_t backlog = te->write_buffer.length;
  gpr_mu_unlock(&te->mu);
  return backlog;
 }
--- a/test/core/util/trickle_endpoint.h
+++ b/test/core/util/trickle_endpoint.h
@ -1,32 +0,0 @@
 /*
 *
 * Copyright 2016 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
 #ifndef TRICKLE_ENDPOINT_H
 #define TRICKLE_ENDPOINT_H
 #include "src/core/lib/iomgr/endpoint.h"
 grpc_endpoint* grpc_trickle_endpoint_create(grpc_endpoint* wrap,
                                            double bytes_per_second);
 /* Allow up to \a bytes through the endpoint. Returns the new backlog. */
 size_t grpc_trickle_endpoint_trickle(grpc_endpoint* endpoint);
 size_t grpc_trickle_get_backlog(grpc_endpoint* endpoint);
 #endif
--- a/test/cpp/microbenchmarks/BUILD
+++ b/test/cpp/microbenchmarks/BUILD
@ -225,22 +225,6 @@ grpc_cc_test(
    deps = [":fullstack_streaming_pump_h"],
 )
 grpc_cc_test(
    name = "bm_fullstack_trickle",
    size = "large",
    srcs = ["bm_fullstack_trickle.cc"],
    args = grpc_benchmark_args(),
    external_deps = [
        "absl/flags:flag",
    ],
    tags = [
        "manual",
        "no_windows",
        "notap",
    ],
    deps = [":helpers"],
 )
 grpc_cc_library(
    name = "fullstack_unary_ping_pong_h",
    testonly = 1,
--- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
@ -1,480 +0,0 @@
 /*
 *
 * Copyright 2016 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */
 /* Benchmark gRPC end2end in various configurations */
 #include <fstream>
 #include <benchmark/benchmark.h>
 #include "absl/flags/flag.h"
 #include "absl/memory/memory.h"
 #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/proto/grpc/testing/echo.grpc.pb.h"
 #include "test/core/util/resource_user_util.h"
 #include "test/core/util/test_config.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"
 ABSL_FLAG(bool, log, false, "Log state to CSV files");
 ABSL_FLAG(int32_t, warmup_megabytes, 1,
          "Number of megabytes to pump before collecting flow control stats");
 ABSL_FLAG(int32_t, warmup_iterations, 100,
          "Number of iterations to run before collecting flow control stats");
 ABSL_FLAG(int32_t, warmup_max_time_seconds, 10,
          "Maximum number of seconds to run warmup loop");
 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>
 static void write_csv(std::ostream* out, A0&& a0) {
  if (!out) return;
  (*out) << a0 << "\n";
 }
 template <class A0, class... Arg>
 static void write_csv(std::ostream* out, A0&& a0, Arg&&... arg) {
  if (!out) return;
  (*out) << a0 << ",";
  write_csv(out, std::forward<Arg>(arg)...);
 }
 class TrickledCHTTP2 : public EndpointPairFixture {
 public:
  TrickledCHTTP2(Service* service, bool streaming, size_t req_size,
                 size_t resp_size, size_t kilobits_per_second,
                 grpc_passthru_endpoint_stats* stats)
      : EndpointPairFixture(service, MakeEndpoints(kilobits_per_second, stats),
                            FixtureConfiguration()),
        stats_(stats) {
    if (absl::GetFlag(FLAGS_log)) {
      std::ostringstream fn;
      fn << "trickle." << (streaming ? "streaming" : "unary") << "." << req_size
         << "." << resp_size << "." << kilobits_per_second << ".csv";
      log_ = absl::make_unique<std::ofstream>(fn.str().c_str());
      write_csv(log_.get(), "t", "iteration", "client_backlog",
                "server_backlog", "client_t_stall", "client_s_stall",
                "server_t_stall", "server_s_stall", "client_t_remote",
                "server_t_remote", "client_t_announced", "server_t_announced",
                "client_s_remote_delta", "server_s_remote_delta",
                "client_s_local_delta", "server_s_local_delta",
                "client_s_announced_delta", "server_s_announced_delta",
                "client_peer_iws", "client_local_iws", "client_sent_iws",
                "client_acked_iws", "server_peer_iws", "server_local_iws",
                "server_sent_iws", "server_acked_iws", "client_queued_bytes",
                "server_queued_bytes");
    }
  }
  ~TrickledCHTTP2() override {
    if (stats_ != nullptr) {
      grpc_passthru_endpoint_stats_destroy(stats_);
    }
  }
  void AddToLabel(std::ostream& out, benchmark::State& state) override {
    out << " writes/iter:"
        << (static_cast<double>(stats_->num_writes) /
            static_cast<double>(state.iterations()))
        << " cli_transport_stalls/iter:"
        << (static_cast<double>(
                client_stats_.streams_stalled_due_to_transport_flow_control) /
            static_cast<double>(state.iterations()))
        << " cli_stream_stalls/iter:"
        << (static_cast<double>(
                client_stats_.streams_stalled_due_to_stream_flow_control) /
            static_cast<double>(state.iterations()))
        << " svr_transport_stalls/iter:"
        << (static_cast<double>(
                server_stats_.streams_stalled_due_to_transport_flow_control) /
            static_cast<double>(state.iterations()))
        << " svr_stream_stalls/iter:"
        << (static_cast<double>(
                server_stats_.streams_stalled_due_to_stream_flow_control) /
            static_cast<double>(state.iterations()));
  }
  void Log(int64_t iteration) GPR_ATTRIBUTE_NO_TSAN {
    auto now = gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), start_);
    grpc_chttp2_transport* client =
        reinterpret_cast<grpc_chttp2_transport*>(client_transport_);
    grpc_chttp2_transport* server =
        reinterpret_cast<grpc_chttp2_transport*>(server_transport_);
    grpc_chttp2_stream* client_stream =
        client->stream_map.count == 1
            ? static_cast<grpc_chttp2_stream*>(client->stream_map.values[0])
            : nullptr;
    grpc_chttp2_stream* server_stream =
        server->stream_map.count == 1
            ? static_cast<grpc_chttp2_stream*>(server->stream_map.values[0])
            : nullptr;
    write_csv(
        log_.get(),
        static_cast<double>(now.tv_sec) +
            1e-9 * static_cast<double>(now.tv_nsec),
        iteration, grpc_trickle_get_backlog(endpoint_pair_.client),
        grpc_trickle_get_backlog(endpoint_pair_.server),
        client->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr,
        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->settings[GRPC_PEER_SETTINGS]
                        [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
        client->settings[GRPC_LOCAL_SETTINGS]
                        [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
        client->settings[GRPC_SENT_SETTINGS]
                        [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
        client->settings[GRPC_ACKED_SETTINGS]
                        [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
        server->settings[GRPC_PEER_SETTINGS]
                        [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
        server->settings[GRPC_LOCAL_SETTINGS]
                        [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
        server->settings[GRPC_SENT_SETTINGS]
                        [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
        server->settings[GRPC_ACKED_SETTINGS]
                        [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
        client_stream ? client_stream->flow_controlled_buffer.length : 0,
        server_stream ? server_stream->flow_controlled_buffer.length : 0);
  }
  void Step(bool update_stats) {
    grpc_core::ExecCtx exec_ctx;
    inc_time();
    size_t client_backlog =
        grpc_trickle_endpoint_trickle(endpoint_pair_.client);
    size_t server_backlog =
        grpc_trickle_endpoint_trickle(endpoint_pair_.server);
    if (update_stats) {
      UpdateStats(reinterpret_cast<grpc_chttp2_transport*>(client_transport_),
                  &client_stats_, client_backlog);
      UpdateStats(reinterpret_cast<grpc_chttp2_transport*>(server_transport_),
                  &server_stats_, server_backlog);
    }
  }
 private:
  grpc_passthru_endpoint_stats* stats_;
  struct Stats {
    int streams_stalled_due_to_stream_flow_control = 0;
    int streams_stalled_due_to_transport_flow_control = 0;
  };
  Stats client_stats_;
  Stats server_stats_;
  std::unique_ptr<std::ofstream> log_;
  gpr_timespec start_ = gpr_now(GPR_CLOCK_MONOTONIC);
  static grpc_endpoint_pair MakeEndpoints(size_t kilobits,
                                          grpc_passthru_endpoint_stats* stats) {
    grpc_endpoint_pair p;
    grpc_passthru_endpoint_create(&p.client, &p.server, stats);
    double bytes_per_second = 125.0 * kilobits;
    p.client = grpc_trickle_endpoint_create(p.client, bytes_per_second);
    p.server = grpc_trickle_endpoint_create(p.server, bytes_per_second);
    return p;
  }
  void UpdateStats(grpc_chttp2_transport* t, Stats* s,
                   size_t backlog) GPR_ATTRIBUTE_NO_TSAN {
    if (backlog == 0) {
      if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr) {
        s->streams_stalled_due_to_stream_flow_control++;
      }
      if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr) {
        s->streams_stalled_due_to_transport_flow_control++;
      }
    }
  }
 };
 static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok,
                          int64_t iteration) {
  while (true) {
    fixture->Log(iteration);
    switch (
        fixture->cq()->AsyncNext(t, ok, gpr_inf_past(GPR_CLOCK_MONOTONIC))) {
      case CompletionQueue::TIMEOUT:
        fixture->Step(iteration != -1);
        break;
      case CompletionQueue::SHUTDOWN:
        GPR_ASSERT(false);
        break;
      case CompletionQueue::GOT_EVENT:
        return;
    }
  }
 }
 static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
  EchoTestService::AsyncService service;
  std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2(
      &service, true, state.range(0) /* req_size */,
      state.range(0) /* resp_size */, state.range(1) /* bw in kbit/s */,
      grpc_passthru_endpoint_stats_create()));
  {
    EchoResponse send_response;
    EchoResponse recv_response;
    if (state.range(0) > 0) {
      send_response.set_message(std::string(state.range(0), 'a'));
    }
    Status recv_status;
    ServerContext svr_ctx;
    ServerAsyncReaderWriter<EchoResponse, EchoRequest> response_rw(&svr_ctx);
    service.RequestBidiStream(&svr_ctx, &response_rw, fixture->cq(),
                              fixture->cq(), tag(0));
    std::unique_ptr<EchoTestService::Stub> stub(
        EchoTestService::NewStub(fixture->channel()));
    ClientContext cli_ctx;
    auto request_rw = stub->AsyncBidiStream(&cli_ctx, fixture->cq(), tag(1));
    int need_tags = (1 << 0) | (1 << 1);
    void* t;
    bool ok;
    while (need_tags) {
      TrickleCQNext(fixture.get(), &t, &ok, -1);
      GPR_ASSERT(ok);
      int i = static_cast<int>(reinterpret_cast<intptr_t>(t));
      GPR_ASSERT(need_tags & (1 << i));
      need_tags &= ~(1 << i);
    }
    request_rw->Read(&recv_response, tag(0));
    auto inner_loop = [&](bool in_warmup) {
      GPR_TIMER_SCOPE("BenchmarkCycle", 0);
      response_rw.Write(send_response, tag(1));
      while (true) {
        TrickleCQNext(fixture.get(), &t, &ok,
                      in_warmup ? -1 : state.iterations());
        if (t == tag(0)) {
          request_rw->Read(&recv_response, tag(0));
        } else if (t == tag(1)) {
          break;
        } else {
          GPR_ASSERT(false);
        }
      }
    };
    gpr_timespec warmup_start = gpr_now(GPR_CLOCK_MONOTONIC);
    for (int i = 0;
         i < std::max(int64_t(absl::GetFlag(FLAGS_warmup_iterations)),
                      absl::GetFlag(FLAGS_warmup_megabytes) * 1024 * 1024 /
                          (14 + state.range(0)));
         i++) {
      inner_loop(true);
      if (gpr_time_cmp(gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), warmup_start),
                       gpr_time_from_seconds(
                           absl::GetFlag(FLAGS_warmup_max_time_seconds),
                           GPR_TIMESPAN)) > 0) {
        break;
      }
    }
    while (state.KeepRunning()) {
      inner_loop(false);
    }
    response_rw.Finish(Status::OK, tag(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 = static_cast<int>(reinterpret_cast<intptr_t>(t));
      GPR_ASSERT(need_tags & (1 << i));
      need_tags &= ~(1 << i);
    }
  }
  fixture->Finish(state);
  fixture.reset();
  state.SetBytesProcessed(state.range(0) * state.iterations());
 }
 static void StreamingTrickleArgs(benchmark::internal::Benchmark* b) {
  for (int i = 1; i <= 128 * 1024 * 1024; i *= 8) {
    for (int j = 64; j <= 128 * 1024 * 1024; j *= 8) {
      double expected_time =
          static_cast<double>(14 + i) / (125.0 * static_cast<double>(j));
      if (expected_time > 2.0) continue;
      b->Args({i, j});
    }
  }
 }
 BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(StreamingTrickleArgs);
 static void BM_PumpUnbalancedUnary_Trickle(benchmark::State& state) {
  EchoTestService::AsyncService service;
  std::unique_ptr<TrickledCHTTP2> fixture(new TrickledCHTTP2(
      &service, false, state.range(0) /* req_size */,
      state.range(1) /* resp_size */, state.range(2) /* bw in kbit/s */,
      grpc_passthru_endpoint_stats_create()));
  EchoRequest send_request;
  EchoResponse send_response;
  EchoResponse recv_response;
  if (state.range(0) > 0) {
    send_request.set_message(std::string(state.range(0), 'a'));
  }
  if (state.range(1) > 0) {
    send_response.set_message(std::string(state.range(1), 'a'));
  }
  Status recv_status;
  struct ServerEnv {
    ServerContext ctx;
    EchoRequest recv_request;
    grpc::ServerAsyncResponseWriter<EchoResponse> response_writer;
    ServerEnv() : response_writer(&ctx) {}
  };
  uint8_t server_env_buffer[2 * sizeof(ServerEnv)];
  ServerEnv* server_env[2] = {
      reinterpret_cast<ServerEnv*>(server_env_buffer),
      reinterpret_cast<ServerEnv*>(server_env_buffer + sizeof(ServerEnv))};
  new (server_env[0]) ServerEnv;
  new (server_env[1]) ServerEnv;
  service.RequestEcho(&server_env[0]->ctx, &server_env[0]->recv_request,
                      &server_env[0]->response_writer, fixture->cq(),
                      fixture->cq(), tag(0));
  service.RequestEcho(&server_env[1]->ctx, &server_env[1]->recv_request,
                      &server_env[1]->response_writer, fixture->cq(),
                      fixture->cq(), tag(1));
  std::unique_ptr<EchoTestService::Stub> stub(
      EchoTestService::NewStub(fixture->channel()));
  auto inner_loop = [&](bool in_warmup) {
    GPR_TIMER_SCOPE("BenchmarkCycle", 0);
    recv_response.Clear();
    ClientContext cli_ctx;
    std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
        stub->AsyncEcho(&cli_ctx, send_request, fixture->cq()));
    void* t;
    bool ok;
    response_reader->Finish(&recv_response, &recv_status, tag(4));
    TrickleCQNext(fixture.get(), &t, &ok, in_warmup ? -1 : state.iterations());
    GPR_ASSERT(ok);
    GPR_ASSERT(t == tag(0) || t == tag(1));
    intptr_t slot = reinterpret_cast<intptr_t>(t);
    ServerEnv* senv = server_env[slot];
    senv->response_writer.Finish(send_response, Status::OK, tag(3));
    for (int i = (1 << 3) | (1 << 4); i != 0;) {
      TrickleCQNext(fixture.get(), &t, &ok,
                    in_warmup ? -1 : state.iterations());
      GPR_ASSERT(ok);
      int tagnum = static_cast<int>(reinterpret_cast<intptr_t>(t));
      GPR_ASSERT(i & (1 << tagnum));
      i -= 1 << tagnum;
    }
    GPR_ASSERT(recv_status.ok());
    senv->~ServerEnv();
    senv = new (senv) ServerEnv();
    service.RequestEcho(&senv->ctx, &senv->recv_request, &senv->response_writer,
                        fixture->cq(), fixture->cq(), tag(slot));
  };
  gpr_timespec warmup_start = gpr_now(GPR_CLOCK_MONOTONIC);
  for (int i = 0; i < std::max(int64_t(absl::GetFlag(FLAGS_warmup_iterations)),
                               absl::GetFlag(FLAGS_warmup_megabytes) * 1024 *
                                   1024 / (14 + state.range(0)));
       i++) {
    inner_loop(true);
    if (gpr_time_cmp(
            gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), warmup_start),
            gpr_time_from_seconds(absl::GetFlag(FLAGS_warmup_max_time_seconds),
                                  GPR_TIMESPAN)) > 0) {
      break;
    }
  }
  while (state.KeepRunning()) {
    inner_loop(false);
  }
  fixture->Finish(state);
  fixture.reset();
  server_env[0]->~ServerEnv();
  server_env[1]->~ServerEnv();
  state.SetBytesProcessed(state.range(0) * state.iterations() +
                          state.range(1) * state.iterations());
 }
 static void UnaryTrickleArgs(benchmark::internal::Benchmark* b) {
  for (int bw = 64; bw <= 128 * 1024 * 1024; bw *= 16) {
    b->Args({1, 1, bw});
    for (int i = 64; i <= 128 * 1024 * 1024; i *= 64) {
      double expected_time =
          static_cast<double>(14 + i) / (125.0 * static_cast<double>(bw));
      if (expected_time > 2.0) continue;
      b->Args({i, 1, bw});
      b->Args({1, i, bw});
      b->Args({i, i, bw});
    }
  }
 }
 BENCHMARK(BM_PumpUnbalancedUnary_Trickle)->Apply(UnaryTrickleArgs);
 }  // namespace testing
 }  // namespace grpc
 extern gpr_timespec (*gpr_now_impl)(gpr_clock_type clock_type);
 // Some distros have RunSpecifiedBenchmarks under the benchmark namespace,
 // and others do not. This allows us to support both modes.
 namespace benchmark {
 void RunTheBenchmarksNamespaced() { RunSpecifiedBenchmarks(); }
 }  // namespace benchmark
 int main(int argc, char** argv) {
  grpc::testing::TestEnvironment env(argc, argv);
  LibraryInitializer libInit;
  ::benchmark::Initialize(&argc, argv);
  ::grpc::testing::InitTest(&argc, &argv, false);
  grpc_timer_manager_set_threading(false);
  gpr_now_impl = ::grpc::testing::fake_now;
  benchmark::RunTheBenchmarksNamespaced();
 }
--- a/tools/internal_ci/linux/grpc_trickle_diff.sh
+++ b/tools/internal_ci/linux/grpc_trickle_diff.sh
@ -1,42 +0,0 @@
 #!/usr/bin/env bash
 # Copyright 2017 gRPC authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
 # This script is invoked by Jenkins and runs a diff on the microbenchmarks
 set -ex
 # List of benchmarks that provide good signal for analyzing performance changes in pull requests
 BENCHMARKS_TO_RUN="cli_transport_stalls_per_iteration cli_stream_stalls_per_iteration svr_transport_stalls_per_iteration svr_stream_stalls_per_iteration"
 # Enter the gRPC repo root
 cd $(dirname $0)/../../..
 source tools/internal_ci/helper_scripts/prepare_build_linux_perf_rc
 tools/run_tests/start_port_server.py
 tools/internal_ci/linux/run_if_c_cpp_modified.sh tools/profiling/microbenchmarks/bm_diff/bm_main.py \
  -d "origin/$KOKORO_GITHUB_PULL_REQUEST_TARGET_BRANCH" \
  -b bm_fullstack_trickle \
  -l 4 \
  -t $BENCHMARKS_TO_RUN \
  --no-counters \
  --pr_comment_name trickle || FAILED="true"
 # kill port_server.py to prevent the build from freezing
 ps aux | grep port_server\\.py | awk '{print $2}' | xargs kill -9
 if [ "$FAILED" != "" ]
 then
  exit 1
 fi
--- a/tools/internal_ci/linux/pull_request/grpc_trickle_diff.cfg
+++ b/tools/internal_ci/linux/pull_request/grpc_trickle_diff.cfg
@ -1,33 +0,0 @@
 # Copyright 2017 gRPC authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # Config file for the internal CI (in protobuf text format)
 # Location of the continuous shell script in repository.
 build_file: "grpc/tools/internal_ci/linux/grpc_trickle_diff.sh"
 timeout_mins: 120
 before_action {
  fetch_keystore {
    keystore_resource {
      keystore_config_id: 73836
      keyname: "grpc_checks_private_key"
    }
  }
 }
 action {
  define_artifacts {
    regex: "**/*sponge_log.*"
    regex: "github/grpc/reports/**"
  }
 }
--- a/tools/profiling/microbenchmarks/bm_diff/bm_constants.py
+++ b/tools/profiling/microbenchmarks/bm_diff/bm_constants.py
@ -19,7 +19,7 @@ _AVAILABLE_BENCHMARK_TESTS = [
    'bm_fullstack_unary_ping_pong', 'bm_fullstack_streaming_ping_pong',
    'bm_fullstack_streaming_pump', 'bm_closure', 'bm_cq', 'bm_call_create',
    'bm_error', 'bm_chttp2_hpack', 'bm_chttp2_transport', 'bm_pollset',
-    'bm_metadata', 'bm_fullstack_trickle'
+    'bm_metadata'
 ]
 _INTERESTING = ('cpu_time', 'real_time', 'locks_per_iteration',
--- a/tools/profiling/microbenchmarks/bm_diff/bm_diff.py
+++ b/tools/profiling/microbenchmarks/bm_diff/bm_diff.py
@ -118,7 +118,7 @@ class Benchmark:
                         (f, new_name, new, old_name, old, mdn_diff))
            s = bm_speedup.speedup(new, old, 1e-5)
            if abs(s) > 3:
-                if mdn_diff > 0.5 or 'trickle' in f:
+                if mdn_diff > 0.5:
                    self.final[f] = '%+d%%' % s
        return self.final.keys()