1 //
2 //
3 // Copyright 2016 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 <grpc/credentials.h>
20 #include <grpc/grpc.h>
21 #include <grpc/grpc_security.h>
22 #include <grpc/support/alloc.h>
23 #include <grpc/support/sync.h>
24 #include <grpc/support/time.h>
25 #include <string.h>
26
27 #include <algorithm>
28 #include <atomic>
29 #include <string>
30 #include <vector>
31
32 #include "absl/log/log.h"
33 #include "absl/strings/str_cat.h"
34 #include "gtest/gtest.h"
35 #include "src/core/config/core_configuration.h"
36 #include "src/core/lib/channel/channel_args_preconditioning.h"
37 #include "src/core/lib/event_engine/channel_args_endpoint_config.h"
38 #include "src/core/lib/iomgr/closure.h"
39 #include "src/core/lib/iomgr/endpoint.h"
40 #include "src/core/lib/iomgr/error.h"
41 #include "src/core/lib/iomgr/exec_ctx.h"
42 #include "src/core/lib/iomgr/iomgr_fwd.h"
43 #include "src/core/lib/iomgr/pollset.h"
44 #include "src/core/lib/iomgr/resolved_address.h"
45 #include "src/core/lib/iomgr/sockaddr.h"
46 #include "src/core/lib/iomgr/tcp_server.h"
47 #include "src/core/util/thd.h"
48 #include "src/core/util/time.h"
49 #include "test/core/test_util/port.h"
50 #include "test/core/test_util/test_config.h"
51
52 // TODO(yashykt): When our macos testing infrastructure becomes good enough, we
53 // wouldn't need to reduce the number of threads on MacOS
54 #ifdef __APPLE__
55 #define NUM_THREADS 10
56 #else
57 #define NUM_THREADS 100
58 #endif // __APPLE
59
60 #define NUM_OUTER_LOOPS 10
61 #define NUM_INNER_LOOPS 10
62 #define DELAY_MILLIS 10
63 #define POLL_MILLIS 15000
64
65 #define NUM_OUTER_LOOPS_SHORT_TIMEOUTS 10
66 #define NUM_INNER_LOOPS_SHORT_TIMEOUTS 100
67 #define DELAY_MILLIS_SHORT_TIMEOUTS 1
68 // in a successful test run, POLL_MILLIS should never be reached because all
69 // runs should end after the shorter delay_millis
70 #define POLL_MILLIS_SHORT_TIMEOUTS 30000
71 // it should never take longer that this to shutdown the server
72 #define SERVER_SHUTDOWN_TIMEOUT 30000
73
tag(int n)74 static void* tag(int n) { return reinterpret_cast<void*>(n); }
75
create_loop_destroy(void * addr)76 void create_loop_destroy(void* addr) {
77 for (int i = 0; i < NUM_OUTER_LOOPS; ++i) {
78 grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr);
79 grpc_channel_credentials* creds = grpc_insecure_credentials_create();
80 grpc_channel* chan =
81 grpc_channel_create(static_cast<char*>(addr), creds, nullptr);
82 grpc_channel_credentials_release(creds);
83
84 for (int j = 0; j < NUM_INNER_LOOPS; ++j) {
85 gpr_timespec later_time =
86 grpc_timeout_milliseconds_to_deadline(DELAY_MILLIS);
87 grpc_connectivity_state state =
88 grpc_channel_check_connectivity_state(chan, 1);
89 grpc_channel_watch_connectivity_state(chan, state, later_time, cq,
90 nullptr);
91 gpr_timespec poll_time =
92 grpc_timeout_milliseconds_to_deadline(POLL_MILLIS);
93 ASSERT_EQ(grpc_completion_queue_next(cq, poll_time, nullptr).type,
94 GRPC_OP_COMPLETE);
95 }
96 grpc_channel_destroy(chan);
97 grpc_completion_queue_destroy(cq);
98 }
99 }
100
101 // Always stack-allocate or new ServerThreadArgs; never use gpr_malloc since
102 // this contains C++ objects.
103 struct ServerThreadArgs {
104 std::string addr;
105 grpc_server* server = nullptr;
106 grpc_completion_queue* cq = nullptr;
107 std::vector<grpc_pollset*> pollset;
108 gpr_mu* mu = nullptr;
109 gpr_event ready;
110 std::atomic_bool stop{false};
111 };
112
server_thread(void * vargs)113 void server_thread(void* vargs) {
114 struct ServerThreadArgs* args = static_cast<struct ServerThreadArgs*>(vargs);
115 grpc_event ev;
116 gpr_timespec deadline =
117 grpc_timeout_milliseconds_to_deadline(SERVER_SHUTDOWN_TIMEOUT);
118 ev = grpc_completion_queue_next(args->cq, deadline, nullptr);
119 ASSERT_EQ(ev.type, GRPC_OP_COMPLETE);
120 ASSERT_EQ(ev.tag, tag(0xd1e));
121 }
122
on_connect(void * vargs,grpc_endpoint * tcp,grpc_pollset *,grpc_tcp_server_acceptor * acceptor)123 static void on_connect(void* vargs, grpc_endpoint* tcp,
124 grpc_pollset* /*accepting_pollset*/,
125 grpc_tcp_server_acceptor* acceptor) {
126 gpr_free(acceptor);
127 struct ServerThreadArgs* args = static_cast<struct ServerThreadArgs*>(vargs);
128 grpc_endpoint_destroy(tcp);
129 gpr_mu_lock(args->mu);
130 GRPC_LOG_IF_ERROR("pollset_kick",
131 grpc_pollset_kick(args->pollset[0], nullptr));
132 gpr_mu_unlock(args->mu);
133 }
134
bad_server_thread(void * vargs)135 void bad_server_thread(void* vargs) {
136 struct ServerThreadArgs* args = static_cast<struct ServerThreadArgs*>(vargs);
137
138 grpc_core::ExecCtx exec_ctx;
139 grpc_resolved_address resolved_addr;
140 grpc_sockaddr* addr = reinterpret_cast<grpc_sockaddr*>(resolved_addr.addr);
141 int port;
142 grpc_tcp_server* s;
143 auto channel_args = grpc_core::CoreConfiguration::Get()
144 .channel_args_preconditioning()
145 .PreconditionChannelArgs(nullptr);
146 grpc_error_handle error = grpc_tcp_server_create(
147 nullptr,
148 grpc_event_engine::experimental::ChannelArgsEndpointConfig(channel_args),
149 on_connect, args, &s);
150 ASSERT_TRUE(error.ok());
151 memset(&resolved_addr, 0, sizeof(resolved_addr));
152 addr->sa_family = GRPC_AF_INET;
153 resolved_addr.len = sizeof(grpc_sockaddr_in);
154 error = grpc_tcp_server_add_port(s, &resolved_addr, &port);
155 ASSERT_TRUE(GRPC_LOG_IF_ERROR("grpc_tcp_server_add_port", error));
156 ASSERT_GT(port, 0);
157 args->addr = absl::StrCat("localhost:", port);
158
159 grpc_tcp_server_start(s, &args->pollset);
160 gpr_event_set(&args->ready, reinterpret_cast<void*>(1));
161
162 gpr_mu_lock(args->mu);
163 while (!args->stop.load(std::memory_order_acquire)) {
164 grpc_core::Timestamp deadline =
165 grpc_core::Timestamp::Now() + grpc_core::Duration::Milliseconds(100);
166
167 grpc_pollset_worker* worker = nullptr;
168 if (!GRPC_LOG_IF_ERROR(
169 "pollset_work",
170 grpc_pollset_work(args->pollset[0], &worker, deadline))) {
171 args->stop.store(true, std::memory_order_release);
172 }
173 gpr_mu_unlock(args->mu);
174
175 gpr_mu_lock(args->mu);
176 }
177 gpr_mu_unlock(args->mu);
178
179 grpc_tcp_server_unref(s);
180 }
181
done_pollset_shutdown(void * pollset,grpc_error_handle)182 static void done_pollset_shutdown(void* pollset, grpc_error_handle /*error*/) {
183 grpc_pollset_destroy(static_cast<grpc_pollset*>(pollset));
184 gpr_free(pollset);
185 }
186
TEST(ConcurrentConnectivityTest,RunConcurrentConnectivityTest)187 TEST(ConcurrentConnectivityTest, RunConcurrentConnectivityTest) {
188 struct ServerThreadArgs args;
189
190 // First round, no server
191 {
192 VLOG(2) << "Wave 1";
193 grpc_core::Thread threads[NUM_THREADS];
194 args.addr = "localhost:54321";
195 for (auto& th : threads) {
196 th = grpc_core::Thread("grpc_wave_1", create_loop_destroy,
197 const_cast<char*>(args.addr.c_str()));
198 th.Start();
199 }
200 for (auto& th : threads) {
201 th.Join();
202 }
203 }
204
205 // Second round, actual grpc server
206 {
207 VLOG(2) << "Wave 2";
208 int port = grpc_pick_unused_port_or_die();
209 args.addr = absl::StrCat("localhost:", port);
210 args.server = grpc_server_create(nullptr, nullptr);
211 grpc_server_credentials* server_creds =
212 grpc_insecure_server_credentials_create();
213 grpc_server_add_http2_port(args.server, args.addr.c_str(), server_creds);
214 grpc_server_credentials_release(server_creds);
215 args.cq = grpc_completion_queue_create_for_next(nullptr);
216 grpc_server_register_completion_queue(args.server, args.cq, nullptr);
217 grpc_server_start(args.server);
218 grpc_core::Thread server2("grpc_wave_2_server", server_thread, &args);
219 server2.Start();
220
221 grpc_core::Thread threads[NUM_THREADS];
222 for (auto& th : threads) {
223 th = grpc_core::Thread("grpc_wave_2", create_loop_destroy,
224 const_cast<char*>(args.addr.c_str()));
225 th.Start();
226 }
227 for (auto& th : threads) {
228 th.Join();
229 }
230 grpc_server_shutdown_and_notify(args.server, args.cq, tag(0xd1e));
231
232 server2.Join();
233 grpc_server_destroy(args.server);
234 grpc_completion_queue_destroy(args.cq);
235 }
236
237 // Third round, bogus tcp server
238 {
239 VLOG(2) << "Wave 3";
240 auto* pollset = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
241 grpc_pollset_init(pollset, &args.mu);
242 args.pollset.push_back(pollset);
243 gpr_event_init(&args.ready);
244 grpc_core::Thread server3("grpc_wave_3_server", bad_server_thread, &args);
245 server3.Start();
246 gpr_event_wait(&args.ready, gpr_inf_future(GPR_CLOCK_MONOTONIC));
247
248 grpc_core::Thread threads[NUM_THREADS];
249 for (auto& th : threads) {
250 th = grpc_core::Thread("grpc_wave_3", create_loop_destroy,
251 const_cast<char*>(args.addr.c_str()));
252 th.Start();
253 }
254 for (auto& th : threads) {
255 th.Join();
256 }
257
258 args.stop.store(true, std::memory_order_release);
259 server3.Join();
260 {
261 grpc_core::ExecCtx exec_ctx;
262 grpc_pollset_shutdown(
263 args.pollset[0],
264 GRPC_CLOSURE_CREATE(done_pollset_shutdown, args.pollset[0],
265 grpc_schedule_on_exec_ctx));
266 }
267 }
268 }
269
watches_with_short_timeouts(void * addr)270 void watches_with_short_timeouts(void* addr) {
271 for (int i = 0; i < NUM_OUTER_LOOPS_SHORT_TIMEOUTS; ++i) {
272 grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr);
273 grpc_channel_credentials* creds = grpc_insecure_credentials_create();
274 grpc_channel* chan =
275 grpc_channel_create(static_cast<char*>(addr), creds, nullptr);
276 grpc_channel_credentials_release(creds);
277
278 for (int j = 0; j < NUM_INNER_LOOPS_SHORT_TIMEOUTS; ++j) {
279 gpr_timespec later_time =
280 grpc_timeout_milliseconds_to_deadline(DELAY_MILLIS_SHORT_TIMEOUTS);
281 grpc_connectivity_state state =
282 grpc_channel_check_connectivity_state(chan, 0);
283 ASSERT_EQ(state, GRPC_CHANNEL_IDLE);
284 grpc_channel_watch_connectivity_state(chan, state, later_time, cq,
285 nullptr);
286 gpr_timespec poll_time =
287 grpc_timeout_milliseconds_to_deadline(POLL_MILLIS_SHORT_TIMEOUTS);
288 grpc_event ev = grpc_completion_queue_next(cq, poll_time, nullptr);
289 ASSERT_EQ(ev.type, GRPC_OP_COMPLETE);
290 ASSERT_EQ(ev.success, false);
291 }
292 grpc_channel_destroy(chan);
293 grpc_completion_queue_destroy(cq);
294 }
295 }
296
297 // This test tries to catch deadlock situations.
298 // With short timeouts on "watches" and long timeouts on cq next calls,
299 // so that a QUEUE_TIMEOUT likely means that something is stuck.
TEST(ConcurrentConnectivityTest,RunConcurrentWatchesWithShortTimeoutsTest)300 TEST(ConcurrentConnectivityTest, RunConcurrentWatchesWithShortTimeoutsTest) {
301 grpc_core::Thread threads[NUM_THREADS];
302 for (auto& th : threads) {
303 th = grpc_core::Thread("grpc_short_watches", watches_with_short_timeouts,
304 const_cast<char*>("localhost:54321"));
305 th.Start();
306 }
307 for (auto& th : threads) {
308 th.Join();
309 }
310 }
311
main(int argc,char ** argv)312 int main(int argc, char** argv) {
313 grpc::testing::TestEnvironment env(&argc, argv);
314 ::testing::InitGoogleTest(&argc, argv);
315 grpc::testing::TestGrpcScope grpc_scope;
316 return RUN_ALL_TESTS();
317 }
318