1 //
2 //
3 // Copyright 2017 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 #include <chrono>
20
21 #include <gtest/gtest.h>
22
23 #include <grpc/support/log.h>
24 #include <grpcpp/channel.h>
25 #include <grpcpp/create_channel.h>
26 #include <grpcpp/impl/grpc_library.h>
27 #include <grpcpp/security/credentials.h>
28 #include <grpcpp/security/server_credentials.h>
29 #include <grpcpp/server.h>
30 #include <grpcpp/server_builder.h>
31
32 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
33 #include "src/core/lib/channel/channel_args.h"
34 #include "src/core/lib/config/core_configuration.h"
35 #include "src/core/lib/debug/stats.h"
36 #include "src/core/lib/event_engine/channel_args_endpoint_config.h"
37 #include "src/core/lib/event_engine/default_event_engine.h"
38 #include "src/core/lib/event_engine/tcp_socket_utils.h"
39 #include "src/core/lib/gprpp/notification.h"
40 #include "src/core/lib/iomgr/endpoint.h"
41 #include "src/core/lib/iomgr/event_engine_shims/endpoint.h"
42 #include "src/core/lib/iomgr/exec_ctx.h"
43 #include "src/core/lib/surface/channel.h"
44 #include "src/core/lib/surface/channel_create.h"
45 #include "src/core/lib/surface/server.h"
46 #include "src/cpp/client/create_channel_internal.h"
47 #include "src/proto/grpc/testing/echo.grpc.pb.h"
48 #include "test/core/event_engine/fuzzing_event_engine/fuzzing_event_engine.h"
49 #include "test/core/util/port.h"
50 #include "test/core/util/test_config.h"
51
52 namespace grpc {
53 namespace testing {
54
55 namespace {
56 using grpc_event_engine::experimental::EventEngine;
57 using grpc_event_engine::experimental::ThreadedFuzzingEventEngine;
58 using grpc_event_engine::experimental::URIToResolvedAddress;
59
tag(intptr_t x)60 void* tag(intptr_t x) { return reinterpret_cast<void*>(x); }
61
62 constexpr int kIterations = 10000;
63 constexpr int kSnapshotEvery = kIterations / 10;
64 } // namespace
65
66 class InProcessCHTTP2 {
67 public:
InProcessCHTTP2(Service * service,std::string & addr,ThreadedFuzzingEventEngine * fuzzing_engine)68 InProcessCHTTP2(Service* service, std::string& addr,
69 ThreadedFuzzingEventEngine* fuzzing_engine) {
70 // TODO(hork): move to a endpoint pair helper
71 // Creating the listener
72 grpc_core::Notification listener_started;
73 std::unique_ptr<EventEngine::Endpoint> listener_endpoint;
74 grpc_core::ChannelArgs args;
75 grpc_event_engine::experimental::ChannelArgsEndpointConfig config(args);
76 auto listener = fuzzing_engine->CreateListener(
77 [&](std::unique_ptr<EventEngine::Endpoint> ep,
78 grpc_core::MemoryAllocator) {
79 listener_endpoint = std::move(ep);
80 listener_started.Notify();
81 },
82 [](absl::Status status) { GPR_ASSERT(status.ok()); }, config,
83 std::make_unique<grpc_core::MemoryQuota>("foo"));
84 if (!listener.ok()) {
85 grpc_core::Crash(absl::StrCat("failed to start listener: ",
86 listener.status().ToString()));
87 }
88 auto target_addr = URIToResolvedAddress(addr);
89 GPR_ASSERT(target_addr.ok());
90 GPR_ASSERT((*listener)->Bind(*target_addr).ok());
91 GPR_ASSERT((*listener)->Start().ok());
92 // Creating the client
93 std::unique_ptr<EventEngine::Endpoint> client_endpoint;
94 grpc_core::Notification client_connected;
95 auto client_memory_quota =
96 std::make_unique<grpc_core::MemoryQuota>("client");
97 std::ignore = fuzzing_engine->Connect(
98 [&](absl::StatusOr<std::unique_ptr<EventEngine::Endpoint>> endpoint) {
99 GPR_ASSERT(endpoint.ok());
100 client_endpoint = std::move(*endpoint);
101 client_connected.Notify();
102 },
103 *target_addr, config,
104 client_memory_quota->CreateMemoryAllocator("conn-1"),
105 grpc_core::Duration::Infinity());
106 client_connected.WaitForNotification();
107 listener_started.WaitForNotification();
108 ServerBuilder b;
109 cq_ = b.AddCompletionQueue(true);
110 b.RegisterService(service);
111 b.SetMaxReceiveMessageSize(INT_MAX);
112 b.SetMaxSendMessageSize(INT_MAX);
113 server_ = b.BuildAndStart();
114 grpc_core::ExecCtx exec_ctx;
115 // add server endpoint to server_
116 {
117 grpc_core::Server* core_server =
118 grpc_core::Server::FromC(server_->c_server());
119 grpc_endpoint* iomgr_server_endpoint =
120 grpc_event_engine_endpoint_create(std::move(listener_endpoint));
121 grpc_core::Transport* transport = grpc_create_chttp2_transport(
122 core_server->channel_args(), iomgr_server_endpoint,
123 /*is_client=*/false);
124 for (grpc_pollset* pollset : core_server->pollsets()) {
125 grpc_endpoint_add_to_pollset(iomgr_server_endpoint, pollset);
126 }
127 GPR_ASSERT(GRPC_LOG_IF_ERROR(
128 "SetupTransport",
129 core_server->SetupTransport(transport, nullptr,
130 core_server->channel_args(), nullptr)));
131 grpc_chttp2_transport_start_reading(transport, nullptr, nullptr, nullptr);
132 }
133 // create channel
134 {
135 grpc_core::ChannelArgs args =
136 grpc_core::CoreConfiguration::Get()
137 .channel_args_preconditioning()
138 .PreconditionChannelArgs(nullptr)
139 .Set(GRPC_ARG_DEFAULT_AUTHORITY, "test.authority");
140 args = args.Set(GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH, INT_MAX)
141 .Set(GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, INT_MAX)
142 .Set(GRPC_ARG_HTTP2_BDP_PROBE, 0);
143 grpc_core::Transport* transport = grpc_create_chttp2_transport(
144 args, grpc_event_engine_endpoint_create(std::move(client_endpoint)),
145 /*is_client=*/true);
146 GPR_ASSERT(transport);
147 grpc_channel* channel =
148 grpc_core::ChannelCreate("target", args, GRPC_CLIENT_DIRECT_CHANNEL,
149 transport)
150 ->release()
151 ->c_ptr();
152 grpc_chttp2_transport_start_reading(transport, nullptr, nullptr, nullptr);
153 channel_ = grpc::CreateChannelInternal(
154 "", channel,
155 std::vector<std::unique_ptr<
156 experimental::ClientInterceptorFactoryInterface>>());
157 }
158 }
159
~InProcessCHTTP2()160 virtual ~InProcessCHTTP2() {
161 server_->Shutdown();
162 cq_->Shutdown();
163 void* tag;
164 bool ok;
165 while (cq_->Next(&tag, &ok)) {
166 }
167 }
168
cq()169 ServerCompletionQueue* cq() { return cq_.get(); }
channel()170 std::shared_ptr<Channel> channel() { return channel_; }
171
172 private:
173 std::unique_ptr<Server> server_;
174 std::unique_ptr<ServerCompletionQueue> cq_;
175 std::shared_ptr<Channel> channel_;
176 };
177
UnaryPingPong(ThreadedFuzzingEventEngine * fuzzing_engine,int request_size,int response_size)178 static double UnaryPingPong(ThreadedFuzzingEventEngine* fuzzing_engine,
179 int request_size, int response_size) {
180 EchoTestService::AsyncService service;
181 std::string target_addr = absl::StrCat(
182 "ipv6:[::1]:", std::to_string(grpc_pick_unused_port_or_die()));
183 std::unique_ptr<InProcessCHTTP2> fixture(
184 new InProcessCHTTP2(&service, target_addr, fuzzing_engine));
185 EchoRequest send_request;
186 EchoResponse send_response;
187 EchoResponse recv_response;
188 if (request_size > 0) {
189 send_request.set_message(std::string(request_size, 'a'));
190 }
191 if (response_size > 0) {
192 send_response.set_message(std::string(response_size, 'a'));
193 }
194 Status recv_status;
195 struct ServerEnv {
196 ServerContext ctx;
197 EchoRequest recv_request;
198 grpc::ServerAsyncResponseWriter<EchoResponse> response_writer;
199 ServerEnv() : response_writer(&ctx) {}
200 };
201 uint8_t server_env_buffer[2 * sizeof(ServerEnv)];
202 ServerEnv* server_env[2] = {
203 reinterpret_cast<ServerEnv*>(server_env_buffer),
204 reinterpret_cast<ServerEnv*>(server_env_buffer + sizeof(ServerEnv))};
205 new (server_env[0]) ServerEnv;
206 new (server_env[1]) ServerEnv;
207 service.RequestEcho(&server_env[0]->ctx, &server_env[0]->recv_request,
208 &server_env[0]->response_writer, fixture->cq(),
209 fixture->cq(), tag(0));
210 service.RequestEcho(&server_env[1]->ctx, &server_env[1]->recv_request,
211 &server_env[1]->response_writer, fixture->cq(),
212 fixture->cq(), tag(1));
213 std::unique_ptr<EchoTestService::Stub> stub(
214 EchoTestService::NewStub(fixture->channel()));
215 auto baseline = grpc_core::global_stats().Collect();
216 auto snapshot = grpc_core::global_stats().Collect();
217 for (int iteration = 0; iteration < kIterations; iteration++) {
218 if (iteration % kSnapshotEvery == 0) {
219 auto new_snapshot = grpc_core::global_stats().Collect();
220 auto diff = new_snapshot->Diff(*snapshot);
221 gpr_log(GPR_DEBUG,
222 " SNAPSHOT: UnaryPingPong(%d, %d): writes_per_iteration=%0.3f "
223 "(total=%lu, i=%d) pings=%lu",
224 request_size, response_size,
225 static_cast<double>(diff->syscall_write) /
226 static_cast<double>(kSnapshotEvery),
227 diff->syscall_write, iteration, diff->http2_pings_sent);
228 snapshot = std::move(new_snapshot);
229 }
230 recv_response.Clear();
231 ClientContext cli_ctx;
232 std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader(
233 stub->AsyncEcho(&cli_ctx, send_request, fixture->cq()));
234 void* t;
235 bool ok;
236 response_reader->Finish(&recv_response, &recv_status, tag(4));
237 GPR_ASSERT(fixture->cq()->Next(&t, &ok));
238 GPR_ASSERT(ok);
239 GPR_ASSERT(t == tag(0) || t == tag(1));
240 intptr_t slot = reinterpret_cast<intptr_t>(t);
241 ServerEnv* senv = server_env[slot];
242 senv->response_writer.Finish(send_response, Status::OK, tag(3));
243 for (int i = (1 << 3) | (1 << 4); i != 0;) {
244 GPR_ASSERT(fixture->cq()->Next(&t, &ok));
245 GPR_ASSERT(ok);
246 int tagnum = static_cast<int>(reinterpret_cast<intptr_t>(t));
247 GPR_ASSERT(i & (1 << tagnum));
248 i -= 1 << tagnum;
249 }
250 GPR_ASSERT(recv_status.ok());
251
252 senv->~ServerEnv();
253 senv = new (senv) ServerEnv();
254 service.RequestEcho(&senv->ctx, &senv->recv_request, &senv->response_writer,
255 fixture->cq(), fixture->cq(), tag(slot));
256 }
257 auto end_stats = grpc_core::global_stats().Collect()->Diff(*baseline);
258 double writes_per_iteration =
259 end_stats->syscall_write / static_cast<double>(kIterations);
260 gpr_log(GPR_DEBUG,
261 "UnaryPingPong(%d, %d): writes_per_iteration=%0.3f (total=%lu)",
262 request_size, response_size, writes_per_iteration,
263 end_stats->syscall_write);
264
265 fixture.reset();
266 server_env[0]->~ServerEnv();
267 server_env[1]->~ServerEnv();
268
269 return writes_per_iteration;
270 }
271
TEST(WritesPerRpcTest,UnaryPingPong)272 TEST(WritesPerRpcTest, UnaryPingPong) {
273 auto fuzzing_engine = std::dynamic_pointer_cast<
274 grpc_event_engine::experimental::ThreadedFuzzingEventEngine>(
275 grpc_event_engine::experimental::GetDefaultEventEngine());
276 EXPECT_LT(UnaryPingPong(fuzzing_engine.get(), 0, 0), 2.2);
277 EXPECT_LT(UnaryPingPong(fuzzing_engine.get(), 1, 0), 2.2);
278 EXPECT_LT(UnaryPingPong(fuzzing_engine.get(), 0, 1), 2.2);
279 EXPECT_LT(UnaryPingPong(fuzzing_engine.get(), 4096, 0), 2.5);
280 EXPECT_LT(UnaryPingPong(fuzzing_engine.get(), 0, 4096), 2.5);
281 }
282
283 } // namespace testing
284 } // namespace grpc
285
main(int argc,char ** argv)286 int main(int argc, char** argv) {
287 grpc_event_engine::experimental::SetEventEngineFactory(
288 []() -> std::unique_ptr<grpc_event_engine::experimental::EventEngine> {
289 return std::make_unique<
290 grpc_event_engine::experimental::ThreadedFuzzingEventEngine>(
291 std::chrono::milliseconds(1));
292 });
293 // avoids a race around gpr_now_impl
294 auto engine = grpc_event_engine::experimental::GetDefaultEventEngine();
295 ::testing::InitGoogleTest(&argc, argv);
296 grpc::testing::TestEnvironment env(&argc, argv);
297 grpc_init();
298 int ret = RUN_ALL_TESTS();
299 grpc_shutdown();
300 return ret;
301 }
302