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