1 /*
2 *
3 * Copyright 2019 The 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 "src/core/lib/iomgr/port.h"
20
21 #include <algorithm>
22 #include <memory>
23 #include <mutex>
24 #include <random>
25 #include <thread>
26
27 #include <grpc/grpc.h>
28 #include <grpc/support/alloc.h>
29 #include <grpc/support/atm.h>
30 #include <grpc/support/log.h>
31 #include <grpc/support/string_util.h>
32 #include <grpc/support/time.h>
33 #include <grpcpp/channel.h>
34 #include <grpcpp/client_context.h>
35 #include <grpcpp/create_channel.h>
36 #include <grpcpp/health_check_service_interface.h>
37 #include <grpcpp/server.h>
38 #include <grpcpp/server_builder.h>
39 #include <gtest/gtest.h>
40
41 #include "src/core/lib/backoff/backoff.h"
42 #include "src/core/lib/gpr/env.h"
43
44 #include "src/proto/grpc/testing/echo.grpc.pb.h"
45 #include "test/core/util/debugger_macros.h"
46 #include "test/core/util/port.h"
47 #include "test/core/util/test_config.h"
48 #include "test/cpp/end2end/test_service_impl.h"
49 #include "test/cpp/util/test_credentials_provider.h"
50
51 #ifdef GRPC_CFSTREAM
52 using grpc::ClientAsyncResponseReader;
53 using grpc::testing::EchoRequest;
54 using grpc::testing::EchoResponse;
55 using grpc::testing::RequestParams;
56 using std::chrono::system_clock;
57
58 namespace grpc {
59 namespace testing {
60 namespace {
61
62 struct TestScenario {
TestScenariogrpc::testing::__anon5706db810111::TestScenario63 TestScenario(const std::string& creds_type, const std::string& content)
64 : credentials_type(creds_type), message_content(content) {}
65 const std::string credentials_type;
66 const std::string message_content;
67 };
68
69 class CFStreamTest : public ::testing::TestWithParam<TestScenario> {
70 protected:
CFStreamTest()71 CFStreamTest()
72 : server_host_("grpctest"),
73 interface_("lo0"),
74 ipv4_address_("10.0.0.1") {}
75
DNSUp()76 void DNSUp() {
77 std::ostringstream cmd;
78 // Add DNS entry for server_host_ in /etc/hosts
79 cmd << "echo '" << ipv4_address_ << " " << server_host_
80 << " ' | sudo tee -a /etc/hosts";
81 std::system(cmd.str().c_str());
82 }
83
DNSDown()84 void DNSDown() {
85 std::ostringstream cmd;
86 // Remove DNS entry for server_host_ in /etc/hosts
87 cmd << "sudo sed -i '.bak' '/" << server_host_ << "/d' /etc/hosts";
88 std::system(cmd.str().c_str());
89 }
90
InterfaceUp()91 void InterfaceUp() {
92 std::ostringstream cmd;
93 cmd << "sudo /sbin/ifconfig " << interface_ << " alias " << ipv4_address_;
94 std::system(cmd.str().c_str());
95 }
96
InterfaceDown()97 void InterfaceDown() {
98 std::ostringstream cmd;
99 cmd << "sudo /sbin/ifconfig " << interface_ << " -alias " << ipv4_address_;
100 std::system(cmd.str().c_str());
101 }
102
NetworkUp()103 void NetworkUp() {
104 gpr_log(GPR_DEBUG, "Bringing network up");
105 InterfaceUp();
106 DNSUp();
107 }
108
NetworkDown()109 void NetworkDown() {
110 gpr_log(GPR_DEBUG, "Bringing network down");
111 InterfaceDown();
112 DNSDown();
113 }
114
SetUp()115 void SetUp() override {
116 NetworkUp();
117 grpc_init();
118 StartServer();
119 }
120
TearDown()121 void TearDown() override {
122 NetworkDown();
123 StopServer();
124 grpc_shutdown();
125 }
126
StartServer()127 void StartServer() {
128 port_ = grpc_pick_unused_port_or_die();
129 server_.reset(new ServerData(port_, GetParam().credentials_type));
130 server_->Start(server_host_);
131 }
StopServer()132 void StopServer() { server_->Shutdown(); }
133
BuildStub(const std::shared_ptr<Channel> & channel)134 std::unique_ptr<grpc::testing::EchoTestService::Stub> BuildStub(
135 const std::shared_ptr<Channel>& channel) {
136 return grpc::testing::EchoTestService::NewStub(channel);
137 }
138
BuildChannel()139 std::shared_ptr<Channel> BuildChannel() {
140 std::ostringstream server_address;
141 server_address << server_host_ << ":" << port_;
142 ChannelArguments args;
143 auto channel_creds = GetCredentialsProvider()->GetChannelCredentials(
144 GetParam().credentials_type, &args);
145 return CreateCustomChannel(server_address.str(), channel_creds, args);
146 }
147
GetStreamID(ClientContext & context)148 int GetStreamID(ClientContext& context) {
149 int stream_id = 0;
150 grpc_call* call = context.c_call();
151 if (call) {
152 grpc_chttp2_stream* stream = grpc_chttp2_stream_from_call(call);
153 if (stream) {
154 stream_id = stream->id;
155 }
156 }
157 return stream_id;
158 }
159
SendRpc(const std::unique_ptr<grpc::testing::EchoTestService::Stub> & stub,bool expect_success=false)160 void SendRpc(
161 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
162 bool expect_success = false) {
163 auto response = std::unique_ptr<EchoResponse>(new EchoResponse());
164 EchoRequest request;
165 auto& msg = GetParam().message_content;
166 request.set_message(msg);
167 ClientContext context;
168 Status status = stub->Echo(&context, request, response.get());
169 int stream_id = GetStreamID(context);
170 if (status.ok()) {
171 gpr_log(GPR_DEBUG, "RPC with stream_id %d succeeded", stream_id);
172 EXPECT_EQ(msg, response->message());
173 } else {
174 gpr_log(GPR_DEBUG, "RPC with stream_id %d failed: %s", stream_id,
175 status.error_message().c_str());
176 }
177 if (expect_success) {
178 EXPECT_TRUE(status.ok());
179 }
180 }
SendAsyncRpc(const std::unique_ptr<grpc::testing::EchoTestService::Stub> & stub,RequestParams param=RequestParams ())181 void SendAsyncRpc(
182 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
183 RequestParams param = RequestParams()) {
184 EchoRequest request;
185 request.set_message(GetParam().message_content);
186 *request.mutable_param() = std::move(param);
187 AsyncClientCall* call = new AsyncClientCall;
188
189 call->response_reader =
190 stub->PrepareAsyncEcho(&call->context, request, &cq_);
191
192 call->response_reader->StartCall();
193 call->response_reader->Finish(&call->reply, &call->status, (void*)call);
194 }
195
ShutdownCQ()196 void ShutdownCQ() { cq_.Shutdown(); }
197
CQNext(void ** tag,bool * ok)198 bool CQNext(void** tag, bool* ok) {
199 auto deadline = std::chrono::system_clock::now() + std::chrono::seconds(10);
200 auto ret = cq_.AsyncNext(tag, ok, deadline);
201 if (ret == grpc::CompletionQueue::GOT_EVENT) {
202 return true;
203 } else if (ret == grpc::CompletionQueue::SHUTDOWN) {
204 return false;
205 } else {
206 GPR_ASSERT(ret == grpc::CompletionQueue::TIMEOUT);
207 // This can happen if we hit the Apple CFStream bug which results in the
208 // read stream hanging. We are ignoring hangs and timeouts, but these
209 // tests are still useful as they can catch memory memory corruptions,
210 // crashes and other bugs that don't result in test hang/timeout.
211 return false;
212 }
213 }
214
WaitForChannelNotReady(Channel * channel,int timeout_seconds=5)215 bool WaitForChannelNotReady(Channel* channel, int timeout_seconds = 5) {
216 const gpr_timespec deadline =
217 grpc_timeout_seconds_to_deadline(timeout_seconds);
218 grpc_connectivity_state state;
219 while ((state = channel->GetState(false /* try_to_connect */)) ==
220 GRPC_CHANNEL_READY) {
221 if (!channel->WaitForStateChange(state, deadline)) return false;
222 }
223 return true;
224 }
225
WaitForChannelReady(Channel * channel,int timeout_seconds=10)226 bool WaitForChannelReady(Channel* channel, int timeout_seconds = 10) {
227 const gpr_timespec deadline =
228 grpc_timeout_seconds_to_deadline(timeout_seconds);
229 grpc_connectivity_state state;
230 while ((state = channel->GetState(true /* try_to_connect */)) !=
231 GRPC_CHANNEL_READY) {
232 if (!channel->WaitForStateChange(state, deadline)) return false;
233 }
234 return true;
235 }
236
237 struct AsyncClientCall {
238 EchoResponse reply;
239 ClientContext context;
240 Status status;
241 std::unique_ptr<ClientAsyncResponseReader<EchoResponse>> response_reader;
242 };
243
244 private:
245 struct ServerData {
246 int port_;
247 const std::string creds_;
248 std::unique_ptr<Server> server_;
249 TestServiceImpl service_;
250 std::unique_ptr<std::thread> thread_;
251 bool server_ready_ = false;
252
ServerDatagrpc::testing::__anon5706db810111::CFStreamTest::ServerData253 ServerData(int port, const std::string& creds)
254 : port_(port), creds_(creds) {}
255
Startgrpc::testing::__anon5706db810111::CFStreamTest::ServerData256 void Start(const std::string& server_host) {
257 gpr_log(GPR_INFO, "starting server on port %d", port_);
258 std::mutex mu;
259 std::unique_lock<std::mutex> lock(mu);
260 std::condition_variable cond;
261 thread_.reset(new std::thread(
262 std::bind(&ServerData::Serve, this, server_host, &mu, &cond)));
263 cond.wait(lock, [this] { return server_ready_; });
264 server_ready_ = false;
265 gpr_log(GPR_INFO, "server startup complete");
266 }
267
Servegrpc::testing::__anon5706db810111::CFStreamTest::ServerData268 void Serve(const std::string& server_host, std::mutex* mu,
269 std::condition_variable* cond) {
270 std::ostringstream server_address;
271 server_address << server_host << ":" << port_;
272 ServerBuilder builder;
273 auto server_creds =
274 GetCredentialsProvider()->GetServerCredentials(creds_);
275 builder.AddListeningPort(server_address.str(), server_creds);
276 builder.RegisterService(&service_);
277 server_ = builder.BuildAndStart();
278 std::lock_guard<std::mutex> lock(*mu);
279 server_ready_ = true;
280 cond->notify_one();
281 }
282
Shutdowngrpc::testing::__anon5706db810111::CFStreamTest::ServerData283 void Shutdown(bool join = true) {
284 server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0));
285 if (join) thread_->join();
286 }
287 };
288
289 CompletionQueue cq_;
290 const std::string server_host_;
291 const std::string interface_;
292 const std::string ipv4_address_;
293 std::unique_ptr<ServerData> server_;
294 int port_;
295 };
296
CreateTestScenarios()297 std::vector<TestScenario> CreateTestScenarios() {
298 std::vector<TestScenario> scenarios;
299 std::vector<std::string> credentials_types;
300 std::vector<std::string> messages;
301
302 credentials_types.push_back(kInsecureCredentialsType);
303 auto sec_list = GetCredentialsProvider()->GetSecureCredentialsTypeList();
304 for (auto sec = sec_list.begin(); sec != sec_list.end(); sec++) {
305 credentials_types.push_back(*sec);
306 }
307
308 messages.push_back("");
309 for (size_t k = 1; k < GRPC_DEFAULT_MAX_RECV_MESSAGE_LENGTH / 1024; k *= 32) {
310 std::string big_msg;
311 for (size_t i = 0; i < k * 1024; ++i) {
312 char c = 'a' + (i % 26);
313 big_msg += c;
314 }
315 messages.push_back(big_msg);
316 }
317 for (auto cred = credentials_types.begin(); cred != credentials_types.end();
318 ++cred) {
319 for (auto msg = messages.begin(); msg != messages.end(); msg++) {
320 scenarios.emplace_back(*cred, *msg);
321 }
322 }
323
324 return scenarios;
325 }
326
327 INSTANTIATE_TEST_SUITE_P(CFStreamTest, CFStreamTest,
328 ::testing::ValuesIn(CreateTestScenarios()));
329
330 // gRPC should automatically detech network flaps (without enabling keepalives)
331 // when CFStream is enabled
TEST_P(CFStreamTest,NetworkTransition)332 TEST_P(CFStreamTest, NetworkTransition) {
333 auto channel = BuildChannel();
334 auto stub = BuildStub(channel);
335 // Channel should be in READY state after we send an RPC
336 SendRpc(stub, /*expect_success=*/true);
337 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
338
339 std::atomic_bool shutdown{false};
340 std::thread sender = std::thread([this, &stub, &shutdown]() {
341 while (true) {
342 if (shutdown.load()) {
343 return;
344 }
345 SendRpc(stub);
346 std::this_thread::sleep_for(std::chrono::milliseconds(1000));
347 }
348 });
349
350 // bring down network
351 NetworkDown();
352
353 // network going down should be detected by cfstream
354 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
355
356 // bring network interface back up
357 std::this_thread::sleep_for(std::chrono::milliseconds(1000));
358 NetworkUp();
359
360 // channel should reconnect
361 EXPECT_TRUE(WaitForChannelReady(channel.get()));
362 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
363 shutdown.store(true);
364 sender.join();
365 }
366
367 // Network flaps while RPCs are in flight
TEST_P(CFStreamTest,NetworkFlapRpcsInFlight)368 TEST_P(CFStreamTest, NetworkFlapRpcsInFlight) {
369 auto channel = BuildChannel();
370 auto stub = BuildStub(channel);
371 std::atomic_int rpcs_sent{0};
372
373 // Channel should be in READY state after we send some RPCs
374 for (int i = 0; i < 10; ++i) {
375 RequestParams param;
376 param.set_skip_cancelled_check(true);
377 SendAsyncRpc(stub, param);
378 ++rpcs_sent;
379 }
380 EXPECT_TRUE(WaitForChannelReady(channel.get()));
381
382 // Bring down the network
383 NetworkDown();
384
385 std::thread thd = std::thread([this, &rpcs_sent]() {
386 void* got_tag;
387 bool ok = false;
388 bool network_down = true;
389 int total_completions = 0;
390
391 while (CQNext(&got_tag, &ok)) {
392 ++total_completions;
393 GPR_ASSERT(ok);
394 AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag);
395 int stream_id = GetStreamID(call->context);
396 if (!call->status.ok()) {
397 gpr_log(GPR_DEBUG, "RPC with stream_id %d failed with error: %s",
398 stream_id, call->status.error_message().c_str());
399 // Bring network up when RPCs start failing
400 if (network_down) {
401 NetworkUp();
402 network_down = false;
403 }
404 } else {
405 gpr_log(GPR_DEBUG, "RPC with stream_id %d succeeded", stream_id);
406 }
407 delete call;
408 }
409 // Remove line below and uncomment the following line after Apple CFStream
410 // bug has been fixed.
411 (void)rpcs_sent;
412 // EXPECT_EQ(total_completions, rpcs_sent);
413 });
414
415 for (int i = 0; i < 100; ++i) {
416 RequestParams param;
417 param.set_skip_cancelled_check(true);
418 SendAsyncRpc(stub, param);
419 std::this_thread::sleep_for(std::chrono::milliseconds(10));
420 ++rpcs_sent;
421 }
422
423 ShutdownCQ();
424
425 thd.join();
426 }
427
428 // Send a bunch of RPCs, some of which are expected to fail.
429 // We should get back a response for all RPCs
TEST_P(CFStreamTest,ConcurrentRpc)430 TEST_P(CFStreamTest, ConcurrentRpc) {
431 auto channel = BuildChannel();
432 auto stub = BuildStub(channel);
433 std::atomic_int rpcs_sent{0};
434 std::thread thd = std::thread([this, &rpcs_sent]() {
435 void* got_tag;
436 bool ok = false;
437 int total_completions = 0;
438
439 while (CQNext(&got_tag, &ok)) {
440 ++total_completions;
441 GPR_ASSERT(ok);
442 AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag);
443 int stream_id = GetStreamID(call->context);
444 if (!call->status.ok()) {
445 gpr_log(GPR_DEBUG, "RPC with stream_id %d failed with error: %s",
446 stream_id, call->status.error_message().c_str());
447 // Bring network up when RPCs start failing
448 } else {
449 gpr_log(GPR_DEBUG, "RPC with stream_id %d succeeded", stream_id);
450 }
451 delete call;
452 }
453 // Remove line below and uncomment the following line after Apple CFStream
454 // bug has been fixed.
455 (void)rpcs_sent;
456 // EXPECT_EQ(total_completions, rpcs_sent);
457 });
458
459 for (int i = 0; i < 10; ++i) {
460 if (i % 3 == 0) {
461 RequestParams param;
462 ErrorStatus* error = param.mutable_expected_error();
463 error->set_code(StatusCode::INTERNAL);
464 error->set_error_message("internal error");
465 SendAsyncRpc(stub, param);
466 } else if (i % 5 == 0) {
467 RequestParams param;
468 param.set_echo_metadata(true);
469 DebugInfo* info = param.mutable_debug_info();
470 info->add_stack_entries("stack_entry1");
471 info->add_stack_entries("stack_entry2");
472 info->set_detail("detailed debug info");
473 SendAsyncRpc(stub, param);
474 } else {
475 SendAsyncRpc(stub);
476 }
477 ++rpcs_sent;
478 }
479
480 ShutdownCQ();
481
482 thd.join();
483 }
484
485 } // namespace
486 } // namespace testing
487 } // namespace grpc
488 #endif // GRPC_CFSTREAM
489
main(int argc,char ** argv)490 int main(int argc, char** argv) {
491 ::testing::InitGoogleTest(&argc, argv);
492 grpc::testing::TestEnvironment env(argc, argv);
493 gpr_setenv("grpc_cfstream", "1");
494 const auto result = RUN_ALL_TESTS();
495 return result;
496 }
497