aboutsummaryrefslogtreecommitdiffhomepage
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/core/transport/bdp_estimator_test.c13
-rw-r--r--test/core/util/trickle_endpoint.c27
-rw-r--r--test/core/util/trickle_endpoint.h2
-rw-r--r--test/cpp/microbenchmarks/BUILD2
-rw-r--r--test/cpp/microbenchmarks/bm_fullstack_trickle.cc156
5 files changed, 176 insertions, 24 deletions
diff --git a/test/core/transport/bdp_estimator_test.c b/test/core/transport/bdp_estimator_test.c
index f55a3ca643..122e097cc4 100644
--- a/test/core/transport/bdp_estimator_test.c
+++ b/test/core/transport/bdp_estimator_test.c
@@ -33,6 +33,7 @@
#include "src/core/lib/transport/bdp_estimator.h"
+#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
@@ -64,6 +65,8 @@ static void add_samples(grpc_bdp_estimator *estimator, int64_t *samples,
GPR_ASSERT(grpc_bdp_estimator_add_incoming_bytes(estimator, samples[i]) ==
false);
}
+ gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
+ gpr_time_from_millis(1, GPR_TIMESPAN)));
grpc_bdp_estimator_complete_ping(estimator);
}
@@ -123,11 +126,11 @@ static void test_get_estimate_random_values(size_t n) {
gpr_log(GPR_INFO, "test_get_estimate_random_values(%" PRIdPTR ")", n);
grpc_bdp_estimator est;
grpc_bdp_estimator_init(&est, "test");
- int min = INT_MAX;
- int max = 65535; // Windows rand() has limited range, make sure the ASSERT
- // passes
+ const int kMaxSample = 65535;
+ int min = kMaxSample;
+ int max = 0;
for (size_t i = 0; i < n; i++) {
- int sample = rand();
+ int sample = rand() % (kMaxSample + 1);
if (sample < min) min = sample;
if (sample > max) max = sample;
add_sample(&est, sample);
@@ -141,6 +144,7 @@ static void test_get_estimate_random_values(size_t n) {
int main(int argc, char **argv) {
grpc_test_init(argc, argv);
+ grpc_init();
test_noop();
test_get_estimate_no_samples();
test_get_estimate_1_sample();
@@ -149,5 +153,6 @@ int main(int argc, char **argv) {
for (size_t i = 3; i < 1000; i = i * 3 / 2) {
test_get_estimate_random_values(i);
}
+ grpc_shutdown();
return 0;
}
diff --git a/test/core/util/trickle_endpoint.c b/test/core/util/trickle_endpoint.c
index 0848147158..02ba257abe 100644
--- a/test/core/util/trickle_endpoint.c
+++ b/test/core/util/trickle_endpoint.c
@@ -44,6 +44,8 @@
#include <grpc/support/useful.h>
#include "src/core/lib/slice/slice_internal.h"
+#define WRITE_BUFFER_SIZE (2 * 1024 * 1024)
+
typedef struct {
grpc_endpoint base;
double bytes_per_second;
@@ -55,6 +57,7 @@ typedef struct {
grpc_slice_buffer writing_buffer;
grpc_error *error;
bool writing;
+ grpc_closure *write_cb;
} trickle_endpoint;
static void te_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
@@ -63,6 +66,15 @@ static void te_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_endpoint_read(exec_ctx, te->wrapped, slices, cb);
}
+static void maybe_call_write_cb_locked(grpc_exec_ctx *exec_ctx,
+ trickle_endpoint *te) {
+ if (te->write_cb != NULL && (te->error != GRPC_ERROR_NONE ||
+ te->write_buffer.length <= WRITE_BUFFER_SIZE)) {
+ grpc_closure_sched(exec_ctx, te->write_cb, GRPC_ERROR_REF(te->error));
+ te->write_cb = NULL;
+ }
+}
+
static void te_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_slice_buffer *slices, grpc_closure *cb) {
trickle_endpoint *te = (trickle_endpoint *)ep;
@@ -70,11 +82,13 @@ static void te_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
grpc_slice_ref_internal(slices->slices[i]);
}
gpr_mu_lock(&te->mu);
+ GPR_ASSERT(te->write_cb == NULL);
if (te->write_buffer.length == 0) {
te->last_write = gpr_now(GPR_CLOCK_MONOTONIC);
}
grpc_slice_buffer_addn(&te->write_buffer, slices->slices, slices->count);
- grpc_closure_sched(exec_ctx, cb, GRPC_ERROR_REF(te->error));
+ te->write_cb = cb;
+ maybe_call_write_cb_locked(exec_ctx, te);
gpr_mu_unlock(&te->mu);
}
@@ -102,6 +116,7 @@ static void te_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
if (te->error == GRPC_ERROR_NONE) {
te->error = GRPC_ERROR_REF(why);
}
+ maybe_call_write_cb_locked(exec_ctx, te);
gpr_mu_unlock(&te->mu);
grpc_endpoint_shutdown(exec_ctx, te->wrapped, why);
}
@@ -157,6 +172,7 @@ grpc_endpoint *grpc_trickle_endpoint_create(grpc_endpoint *wrap,
te->base.vtable = &vtable;
te->wrapped = wrap;
te->bytes_per_second = bytes_per_second;
+ te->write_cb = NULL;
gpr_mu_init(&te->mu);
grpc_slice_buffer_init(&te->write_buffer);
grpc_slice_buffer_init(&te->writing_buffer);
@@ -187,9 +203,18 @@ size_t grpc_trickle_endpoint_trickle(grpc_exec_ctx *exec_ctx,
grpc_endpoint_write(
exec_ctx, te->wrapped, &te->writing_buffer,
grpc_closure_create(te_finish_write, te, grpc_schedule_on_exec_ctx));
+ maybe_call_write_cb_locked(exec_ctx, 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 = (trickle_endpoint *)ep;
+ gpr_mu_lock(&te->mu);
+ size_t backlog = te->write_buffer.length;
+ gpr_mu_unlock(&te->mu);
+ return backlog;
+}
diff --git a/test/core/util/trickle_endpoint.h b/test/core/util/trickle_endpoint.h
index 7e8d9d91e3..e513774eb4 100644
--- a/test/core/util/trickle_endpoint.h
+++ b/test/core/util/trickle_endpoint.h
@@ -43,4 +43,6 @@ grpc_endpoint *grpc_trickle_endpoint_create(grpc_endpoint *wrap,
size_t grpc_trickle_endpoint_trickle(grpc_exec_ctx *exec_ctx,
grpc_endpoint *endpoint);
+size_t grpc_trickle_get_backlog(grpc_endpoint *endpoint);
+
#endif
diff --git a/test/cpp/microbenchmarks/BUILD b/test/cpp/microbenchmarks/BUILD
index cae3fa1a14..208ac6d794 100644
--- a/test/cpp/microbenchmarks/BUILD
+++ b/test/cpp/microbenchmarks/BUILD
@@ -92,7 +92,7 @@ cc_test(
cc_test(
name = "bm_fullstack_trickle",
srcs = ["bm_fullstack_trickle.cc"],
- deps = [":helpers"],
+ deps = [":helpers", "//external:gflags"],
)
cc_test(
diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
index a5cfeb4f95..fc99b06dbb 100644
--- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
+++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc
@@ -34,6 +34,8 @@
/* Benchmark gRPC end2end in various configurations */
#include <benchmark/benchmark.h>
+#include <gflags/gflags.h>
+#include <fstream>
#include "src/core/lib/profiling/timers.h"
#include "src/cpp/client/create_channel_internal.h"
#include "src/proto/grpc/testing/echo.grpc.pb.h"
@@ -45,16 +47,57 @@ extern "C" {
#include "test/core/util/trickle_endpoint.h"
}
+DEFINE_bool(log, false, "Log state to CSV files");
+DEFINE_int32(
+ warmup_megabytes, 1,
+ "Number of megabytes to pump before collecting flow control stats");
+DEFINE_int32(
+ warmup_iterations, 100,
+ "Number of megabytes to pump before collecting flow control stats");
+DEFINE_int32(warmup_max_time_seconds, 10,
+ "Maximum number of seconds to run warmup loop");
+
namespace grpc {
namespace testing {
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, size_t megabits_per_second)
- : EndpointPairFixture(service, MakeEndpoints(megabits_per_second),
- FixtureConfiguration()) {}
+ TrickledCHTTP2(Service* service, size_t message_size,
+ size_t kilobits_per_second)
+ : EndpointPairFixture(service, MakeEndpoints(kilobits_per_second),
+ FixtureConfiguration()) {
+ if (FLAGS_log) {
+ std::ostringstream fn;
+ fn << "trickle." << message_size << "." << kilobits_per_second << ".csv";
+ log_.reset(new 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_outgoing",
+ "server_t_outgoing", "client_t_incoming", "server_t_incoming",
+ "client_s_outgoing_delta", "server_s_outgoing_delta",
+ "client_s_incoming_delta", "server_s_incoming_delta",
+ "client_s_announce_window", "server_s_announce_window",
+ "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");
+ }
+ }
void AddToLabel(std::ostream& out, benchmark::State& state) {
out << " writes/iter:"
@@ -75,7 +118,58 @@ class TrickledCHTTP2 : public EndpointPairFixture {
(double)state.iterations());
}
- void Step() {
+ void Log(int64_t iteration) {
+ 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->outgoing_window, server->outgoing_window,
+ client->incoming_window, server->incoming_window,
+ client_stream ? client_stream->outgoing_window_delta : -1,
+ server_stream ? server_stream->outgoing_window_delta : -1,
+ client_stream ? client_stream->incoming_window_delta : -1,
+ server_stream ? server_stream->incoming_window_delta : -1,
+ client_stream ? client_stream->announce_window : -1,
+ server_stream ? server_stream->announce_window : -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_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
size_t client_backlog =
grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.client);
@@ -83,10 +177,12 @@ class TrickledCHTTP2 : public EndpointPairFixture {
grpc_trickle_endpoint_trickle(&exec_ctx, endpoint_pair_.server);
grpc_exec_ctx_finish(&exec_ctx);
- UpdateStats((grpc_chttp2_transport*)client_transport_, &client_stats_,
- client_backlog);
- UpdateStats((grpc_chttp2_transport*)server_transport_, &server_stats_,
- server_backlog);
+ if (update_stats) {
+ UpdateStats((grpc_chttp2_transport*)client_transport_, &client_stats_,
+ client_backlog);
+ UpdateStats((grpc_chttp2_transport*)server_transport_, &server_stats_,
+ server_backlog);
+ }
}
private:
@@ -97,6 +193,8 @@ class TrickledCHTTP2 : public EndpointPairFixture {
};
Stats client_stats_;
Stats server_stats_;
+ std::unique_ptr<std::ofstream> log_;
+ gpr_timespec start_ = gpr_now(GPR_CLOCK_MONOTONIC);
grpc_endpoint_pair MakeEndpoints(size_t kilobits) {
grpc_endpoint_pair p;
@@ -123,13 +221,15 @@ class TrickledCHTTP2 : public EndpointPairFixture {
// force library initialization
auto& force_library_initialization = Library::get();
-static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok) {
+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)))) {
case CompletionQueue::TIMEOUT:
- fixture->Step();
+ fixture->Step(iteration != -1);
break;
case CompletionQueue::SHUTDOWN:
GPR_ASSERT(false);
@@ -143,7 +243,7 @@ static void TrickleCQNext(TrickledCHTTP2* fixture, void** t, bool* ok) {
static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
EchoTestService::AsyncService service;
std::unique_ptr<TrickledCHTTP2> fixture(
- new TrickledCHTTP2(&service, state.range(1)));
+ new TrickledCHTTP2(&service, state.range(0), state.range(1)));
{
EchoResponse send_response;
EchoResponse recv_response;
@@ -163,18 +263,19 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
void* t;
bool ok;
while (need_tags) {
- TrickleCQNext(fixture.get(), &t, &ok);
+ TrickleCQNext(fixture.get(), &t, &ok, -1);
GPR_ASSERT(ok);
int i = (int)(intptr_t)t;
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
}
request_rw->Read(&recv_response, tag(0));
- while (state.KeepRunning()) {
+ 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);
+ 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)) {
@@ -183,11 +284,26 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
GPR_ASSERT(false);
}
}
+ };
+ gpr_timespec warmup_start = gpr_now(GPR_CLOCK_MONOTONIC);
+ for (int i = 0;
+ i < GPR_MAX(FLAGS_warmup_iterations, 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(FLAGS_warmup_max_time_seconds,
+ GPR_TIMESPAN)) > 0) {
+ break;
+ }
+ }
+ while (state.KeepRunning()) {
+ inner_loop(false);
}
response_rw.Finish(Status::OK, tag(1));
need_tags = (1 << 0) | (1 << 1);
while (need_tags) {
- TrickleCQNext(fixture.get(), &t, &ok);
+ TrickleCQNext(fixture.get(), &t, &ok, -1);
int i = (int)(intptr_t)t;
GPR_ASSERT(need_tags & (1 << i));
need_tags &= ~(1 << i);
@@ -204,10 +320,10 @@ static void BM_PumpStreamServerToClient_Trickle(benchmark::State& state) {
static void TrickleArgs(benchmark::internal::Benchmark* b) {
for (int i = 1; i <= 128 * 1024 * 1024; i *= 8) {
- for (int j = 1; j <= 128 * 1024 * 1024; j *= 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 > 0.01) continue;
+ if (expected_time > 2.0) continue;
b->Args({i, j});
}
}
@@ -217,4 +333,8 @@ BENCHMARK(BM_PumpStreamServerToClient_Trickle)->Apply(TrickleArgs);
}
}
-BENCHMARK_MAIN();
+int main(int argc, char** argv) {
+ ::benchmark::Initialize(&argc, argv);
+ ::google::ParseCommandLineFlags(&argc, &argv, false);
+ ::benchmark::RunSpecifiedBenchmarks();
+}
generated by cgit v1.2.3 (git 2.39.1) at 2025-02-02 15:15:29 +0000