1 //
2 //
3 // Copyright 2015 gRPC authors.
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 // http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16 //
17 //
18
19 // Microbenchmarks around CHTTP2 transport operations
20
21 #include <string.h>
22
23 #include <memory>
24 #include <queue>
25 #include <sstream>
26
27 #include <benchmark/benchmark.h>
28
29 #include <grpc/support/alloc.h>
30 #include <grpc/support/log.h>
31 #include <grpc/support/string_util.h>
32 #include <grpcpp/support/channel_arguments.h>
33
34 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
35 #include "src/core/ext/transport/chttp2/transport/internal.h"
36 #include "src/core/lib/gprpp/crash.h"
37 #include "src/core/lib/iomgr/closure.h"
38 #include "src/core/lib/resource_quota/api.h"
39 #include "src/core/lib/slice/slice_internal.h"
40 #include "test/core/util/test_config.h"
41 #include "test/cpp/microbenchmarks/helpers.h"
42 #include "test/cpp/util/test_config.h"
43
44 ////////////////////////////////////////////////////////////////////////////////
45 // Helper classes
46 //
47
48 class PhonyEndpoint : public grpc_endpoint {
49 public:
PhonyEndpoint()50 PhonyEndpoint() {
51 static const grpc_endpoint_vtable my_vtable = {read,
52 write,
53 add_to_pollset,
54 add_to_pollset_set,
55 delete_from_pollset_set,
56 shutdown,
57 destroy,
58 get_peer,
59 get_local_address,
60 get_fd,
61 can_track_err};
62 grpc_endpoint::vtable = &my_vtable;
63 }
64
PushInput(grpc_slice slice)65 void PushInput(grpc_slice slice) {
66 if (read_cb_ == nullptr) {
67 GPR_ASSERT(!have_slice_);
68 buffered_slice_ = slice;
69 have_slice_ = true;
70 return;
71 }
72 grpc_slice_buffer_add(slices_, slice);
73 grpc_core::ExecCtx::Run(DEBUG_LOCATION, read_cb_, absl::OkStatus());
74 read_cb_ = nullptr;
75 }
76
77 private:
78 grpc_closure* read_cb_ = nullptr;
79 grpc_slice_buffer* slices_ = nullptr;
80 bool have_slice_ = false;
81 grpc_slice buffered_slice_;
82
QueueRead(grpc_slice_buffer * slices,grpc_closure * cb)83 void QueueRead(grpc_slice_buffer* slices, grpc_closure* cb) {
84 GPR_ASSERT(read_cb_ == nullptr);
85 if (have_slice_) {
86 have_slice_ = false;
87 grpc_slice_buffer_add(slices, buffered_slice_);
88 grpc_core::ExecCtx::Run(DEBUG_LOCATION, cb, absl::OkStatus());
89 return;
90 }
91 read_cb_ = cb;
92 slices_ = slices;
93 }
94
read(grpc_endpoint * ep,grpc_slice_buffer * slices,grpc_closure * cb,bool,int)95 static void read(grpc_endpoint* ep, grpc_slice_buffer* slices,
96 grpc_closure* cb, bool /*urgent*/,
97 int /*min_progress_size*/) {
98 static_cast<PhonyEndpoint*>(ep)->QueueRead(slices, cb);
99 }
100
write(grpc_endpoint *,grpc_slice_buffer *,grpc_closure * cb,void *,int)101 static void write(grpc_endpoint* /*ep*/, grpc_slice_buffer* /*slices*/,
102 grpc_closure* cb, void* /*arg*/, int /*max_frame_size*/) {
103 grpc_core::ExecCtx::Run(DEBUG_LOCATION, cb, absl::OkStatus());
104 }
105
add_to_pollset(grpc_endpoint *,grpc_pollset *)106 static void add_to_pollset(grpc_endpoint* /*ep*/, grpc_pollset* /*pollset*/) {
107 }
108
add_to_pollset_set(grpc_endpoint *,grpc_pollset_set *)109 static void add_to_pollset_set(grpc_endpoint* /*ep*/,
110 grpc_pollset_set* /*pollset*/) {}
111
delete_from_pollset_set(grpc_endpoint *,grpc_pollset_set *)112 static void delete_from_pollset_set(grpc_endpoint* /*ep*/,
113 grpc_pollset_set* /*pollset*/) {}
114
shutdown(grpc_endpoint * ep,grpc_error_handle why)115 static void shutdown(grpc_endpoint* ep, grpc_error_handle why) {
116 grpc_core::ExecCtx::Run(DEBUG_LOCATION,
117 static_cast<PhonyEndpoint*>(ep)->read_cb_, why);
118 }
119
destroy(grpc_endpoint * ep)120 static void destroy(grpc_endpoint* ep) {
121 delete static_cast<PhonyEndpoint*>(ep);
122 }
123
get_peer(grpc_endpoint *)124 static absl::string_view get_peer(grpc_endpoint* /*ep*/) { return "test"; }
get_local_address(grpc_endpoint *)125 static absl::string_view get_local_address(grpc_endpoint* /*ep*/) {
126 return "test";
127 }
get_fd(grpc_endpoint *)128 static int get_fd(grpc_endpoint* /*ep*/) { return 0; }
can_track_err(grpc_endpoint *)129 static bool can_track_err(grpc_endpoint* /*ep*/) { return false; }
130 };
131
132 class Fixture {
133 public:
Fixture(const grpc::ChannelArguments & args,bool client)134 Fixture(const grpc::ChannelArguments& args, bool client) {
135 grpc_channel_args c_args = args.c_channel_args();
136 ep_ = new PhonyEndpoint;
137 auto final_args = grpc_core::CoreConfiguration::Get()
138 .channel_args_preconditioning()
139 .PreconditionChannelArgs(&c_args);
140 t_ = grpc_create_chttp2_transport(final_args, ep_, client);
141 grpc_chttp2_transport_start_reading(t_, nullptr, nullptr, nullptr);
142 FlushExecCtx();
143 }
144
FlushExecCtx()145 void FlushExecCtx() { grpc_core::ExecCtx::Get()->Flush(); }
146
~Fixture()147 ~Fixture() { t_->Orphan(); }
148
chttp2_transport()149 grpc_chttp2_transport* chttp2_transport() {
150 return reinterpret_cast<grpc_chttp2_transport*>(t_);
151 }
transport()152 grpc_core::Transport* transport() { return t_; }
153
PushInput(grpc_slice slice)154 void PushInput(grpc_slice slice) { ep_->PushInput(slice); }
155
156 private:
157 PhonyEndpoint* ep_;
158 grpc_core::Transport* t_;
159 };
160
161 class TestClosure : public grpc_closure {
162 public:
~TestClosure()163 virtual ~TestClosure() {}
164 };
165
166 template <class F>
MakeTestClosure(F f)167 std::unique_ptr<TestClosure> MakeTestClosure(F f) {
168 struct C : public TestClosure {
169 explicit C(const F& f) : f_(f) {
170 GRPC_CLOSURE_INIT(this, Execute, this, nullptr);
171 }
172 F f_;
173 static void Execute(void* arg, grpc_error_handle error) {
174 static_cast<C*>(arg)->f_(error);
175 }
176 };
177 return std::unique_ptr<TestClosure>(new C(f));
178 }
179
180 template <class F>
MakeOnceClosure(F f)181 grpc_closure* MakeOnceClosure(F f) {
182 struct C : public grpc_closure {
183 explicit C(const F& f) : f_(f) {}
184 F f_;
185 static void Execute(void* arg, grpc_error_handle error) {
186 static_cast<C*>(arg)->f_(error);
187 delete static_cast<C*>(arg);
188 }
189 };
190 auto* c = new C{f};
191 return GRPC_CLOSURE_INIT(c, C::Execute, c, nullptr);
192 }
193
194 class Stream {
195 public:
Stream(Fixture * f)196 explicit Stream(Fixture* f) : f_(f) {
197 stream_size_ = f->transport()->filter_stack_transport()->SizeOfStream();
198 stream_ = gpr_malloc(stream_size_);
199 arena_ = grpc_core::Arena::Create(4096, &memory_allocator_);
200 }
201
~Stream()202 ~Stream() {
203 gpr_event_wait(&done_, gpr_inf_future(GPR_CLOCK_REALTIME));
204 gpr_free(stream_);
205 arena_->Destroy();
206 }
207
Init(benchmark::State & state)208 void Init(benchmark::State& state) {
209 GRPC_STREAM_REF_INIT(&refcount_, 1, &Stream::FinishDestroy, this,
210 "test_stream");
211 gpr_event_init(&done_);
212 memset(stream_, 0, stream_size_);
213 if ((state.iterations() & 0xffff) == 0) {
214 arena_->Destroy();
215 arena_ = grpc_core::Arena::Create(4096, &memory_allocator_);
216 }
217 f_->transport()->filter_stack_transport()->InitStream(
218 static_cast<grpc_stream*>(stream_), &refcount_, nullptr, arena_);
219 }
220
DestroyThen(grpc_closure * closure)221 void DestroyThen(grpc_closure* closure) {
222 destroy_closure_ = closure;
223 #ifndef NDEBUG
224 grpc_stream_unref(&refcount_, "DestroyThen");
225 #else
226 grpc_stream_unref(&refcount_);
227 #endif
228 }
229
Op(grpc_transport_stream_op_batch * op)230 void Op(grpc_transport_stream_op_batch* op) {
231 f_->transport()->filter_stack_transport()->PerformStreamOp(
232 static_cast<grpc_stream*>(stream_), op);
233 }
234
chttp2_stream()235 grpc_chttp2_stream* chttp2_stream() {
236 return static_cast<grpc_chttp2_stream*>(stream_);
237 }
238
239 private:
FinishDestroy(void * arg,grpc_error_handle)240 static void FinishDestroy(void* arg, grpc_error_handle /*error*/) {
241 auto stream = static_cast<Stream*>(arg);
242 stream->f_->transport()->filter_stack_transport()->DestroyStream(
243 static_cast<grpc_stream*>(stream->stream_), stream->destroy_closure_);
244 gpr_event_set(&stream->done_, reinterpret_cast<void*>(1));
245 }
246
247 Fixture* f_;
248 grpc_stream_refcount refcount_;
249 grpc_core::MemoryAllocator memory_allocator_ =
250 grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
251 ->memory_quota()
252 ->CreateMemoryAllocator("test"));
253 grpc_core::Arena* arena_;
254 size_t stream_size_;
255 void* stream_;
256 grpc_closure* destroy_closure_ = nullptr;
257 gpr_event done_;
258 };
259
260 ////////////////////////////////////////////////////////////////////////////////
261 // Benchmarks
262 //
263 std::vector<std::unique_ptr<gpr_event>> done_events;
264
BM_StreamCreateDestroy(benchmark::State & state)265 static void BM_StreamCreateDestroy(benchmark::State& state) {
266 grpc_core::ExecCtx exec_ctx;
267 Fixture f(grpc::ChannelArguments(), true);
268 auto* s = new Stream(&f);
269 grpc_transport_stream_op_batch op;
270 grpc_transport_stream_op_batch_payload op_payload(nullptr);
271 op = {};
272 op.cancel_stream = true;
273 op.payload = &op_payload;
274 op_payload.cancel_stream.cancel_error = absl::CancelledError();
275 std::unique_ptr<TestClosure> next =
276 MakeTestClosure([&, s](grpc_error_handle /*error*/) {
277 if (!state.KeepRunning()) {
278 delete s;
279 return;
280 }
281 s->Init(state);
282 s->Op(&op);
283 s->DestroyThen(next.get());
284 });
285 grpc_core::Closure::Run(DEBUG_LOCATION, next.get(), absl::OkStatus());
286 f.FlushExecCtx();
287 }
288 BENCHMARK(BM_StreamCreateDestroy);
289
290 class RepresentativeClientInitialMetadata {
291 public:
Prepare(grpc_metadata_batch * b)292 static void Prepare(grpc_metadata_batch* b) {
293 b->Set(grpc_core::HttpSchemeMetadata(),
294 grpc_core::HttpSchemeMetadata::kHttp);
295 b->Set(grpc_core::HttpMethodMetadata(),
296 grpc_core::HttpMethodMetadata::kPost);
297 b->Set(grpc_core::HttpPathMetadata(),
298 grpc_core::Slice(grpc_core::StaticSlice::FromStaticString(
299 "/foo/bar/bm_chttp2_transport")));
300 b->Set(grpc_core::HttpAuthorityMetadata(),
301 grpc_core::Slice(grpc_core::StaticSlice::FromStaticString(
302 "foo.test.google.fr:1234")));
303 b->Set(
304 grpc_core::GrpcAcceptEncodingMetadata(),
305 grpc_core::CompressionAlgorithmSet(
306 {GRPC_COMPRESS_NONE, GRPC_COMPRESS_DEFLATE, GRPC_COMPRESS_GZIP}));
307 b->Set(grpc_core::TeMetadata(), grpc_core::TeMetadata::kTrailers);
308 b->Set(grpc_core::ContentTypeMetadata(),
309 grpc_core::ContentTypeMetadata::kApplicationGrpc);
310 b->Set(grpc_core::UserAgentMetadata(),
311 grpc_core::Slice(grpc_core::StaticSlice::FromStaticString(
312 "grpc-c/3.0.0-dev (linux; chttp2; green)")));
313 }
314 };
315
316 template <class Metadata>
BM_StreamCreateSendInitialMetadataDestroy(benchmark::State & state)317 static void BM_StreamCreateSendInitialMetadataDestroy(benchmark::State& state) {
318 grpc_core::ExecCtx exec_ctx;
319 Fixture f(grpc::ChannelArguments(), true);
320 auto* s = new Stream(&f);
321 grpc_transport_stream_op_batch op;
322 grpc_transport_stream_op_batch_payload op_payload(nullptr);
323 std::unique_ptr<TestClosure> start;
324 std::unique_ptr<TestClosure> done;
325
326 auto reset_op = [&]() {
327 op = {};
328 op.payload = &op_payload;
329 };
330
331 grpc_core::MemoryAllocator memory_allocator =
332 grpc_core::MemoryAllocator(grpc_core::ResourceQuota::Default()
333 ->memory_quota()
334 ->CreateMemoryAllocator("test"));
335 auto arena = grpc_core::MakeScopedArena(1024, &memory_allocator);
336 grpc_metadata_batch b;
337 Metadata::Prepare(&b);
338
339 f.FlushExecCtx();
340 gpr_event bm_done;
341 gpr_event_init(&bm_done);
342 start = MakeTestClosure([&, s](grpc_error_handle /*error*/) {
343 if (!state.KeepRunning()) {
344 delete s;
345 gpr_event_set(&bm_done, (void*)1);
346 return;
347 }
348 s->Init(state);
349 reset_op();
350 op.on_complete = done.get();
351 op.send_initial_metadata = true;
352 op.payload->send_initial_metadata.send_initial_metadata = &b;
353 s->Op(&op);
354 });
355 done = MakeTestClosure([&](grpc_error_handle /*error*/) {
356 reset_op();
357 op.cancel_stream = true;
358 op.payload->cancel_stream.cancel_error = absl::CancelledError();
359 s->Op(&op);
360 s->DestroyThen(start.get());
361 });
362 grpc_core::ExecCtx::Run(DEBUG_LOCATION, start.get(), absl::OkStatus());
363 f.FlushExecCtx();
364 gpr_event_wait(&bm_done, gpr_inf_future(GPR_CLOCK_REALTIME));
365 }
366 BENCHMARK_TEMPLATE(BM_StreamCreateSendInitialMetadataDestroy,
367 RepresentativeClientInitialMetadata);
368
BM_TransportEmptyOp(benchmark::State & state)369 static void BM_TransportEmptyOp(benchmark::State& state) {
370 grpc_core::ExecCtx exec_ctx;
371 Fixture f(grpc::ChannelArguments(), true);
372 auto* s = new Stream(&f);
373 s->Init(state);
374 grpc_transport_stream_op_batch op;
375 grpc_transport_stream_op_batch_payload op_payload(nullptr);
376 auto reset_op = [&]() {
377 op = {};
378 op.payload = &op_payload;
379 };
380 std::unique_ptr<TestClosure> c =
381 MakeTestClosure([&](grpc_error_handle /*error*/) {
382 if (!state.KeepRunning()) return;
383 reset_op();
384 op.on_complete = c.get();
385 s->Op(&op);
386 });
387 grpc_core::ExecCtx::Run(DEBUG_LOCATION, c.get(), absl::OkStatus());
388 f.FlushExecCtx();
389 reset_op();
390 op.cancel_stream = true;
391 op_payload.cancel_stream.cancel_error = absl::CancelledError();
392 gpr_event* stream_cancel_done = new gpr_event;
393 gpr_event_init(stream_cancel_done);
394 std::unique_ptr<TestClosure> stream_cancel_closure =
395 MakeTestClosure([&](grpc_error_handle error) {
396 GPR_ASSERT(error.ok());
397 gpr_event_set(stream_cancel_done, reinterpret_cast<void*>(1));
398 });
399 op.on_complete = stream_cancel_closure.get();
400 s->Op(&op);
401 f.FlushExecCtx();
402 gpr_event_wait(stream_cancel_done, gpr_inf_future(GPR_CLOCK_REALTIME));
403 done_events.emplace_back(stream_cancel_done);
404 s->DestroyThen(
405 MakeOnceClosure([s](grpc_error_handle /*error*/) { delete s; }));
406 f.FlushExecCtx();
407 }
408 BENCHMARK(BM_TransportEmptyOp);
409
410 // Some distros have RunSpecifiedBenchmarks under the benchmark namespace,
411 // and others do not. This allows us to support both modes.
412 namespace benchmark {
RunTheBenchmarksNamespaced()413 void RunTheBenchmarksNamespaced() { RunSpecifiedBenchmarks(); }
414 } // namespace benchmark
415
main(int argc,char ** argv)416 int main(int argc, char** argv) {
417 grpc::testing::TestEnvironment env(&argc, argv);
418 LibraryInitializer libInit;
419 ::benchmark::Initialize(&argc, argv);
420 grpc::testing::InitTest(&argc, &argv, false);
421 benchmark::RunTheBenchmarksNamespaced();
422 return 0;
423 }
424