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