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 "src/core/lib/iomgr/port.h"
20
21 // This test won't work except with libuv
22 #ifdef GRPC_UV
23
24 #include <uv.h>
25
26 #include "src/core/lib/iomgr/tcp_server.h"
27
28 #include <string.h>
29
30 #include <grpc/grpc.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/iomgr/iomgr.h"
37 #include "src/core/lib/iomgr/resolve_address.h"
38 #include "src/core/lib/iomgr/sockaddr_utils.h"
39 #include "test/core/util/port.h"
40 #include "test/core/util/test_config.h"
41
42 #define LOG_TEST(x) gpr_log(GPR_INFO, "%s", #x)
43
44 static gpr_mu* g_mu;
45 static grpc_pollset* g_pollset;
46 static int g_nconnects = 0;
47
48 typedef struct on_connect_result {
49 /* Owns a ref to server. */
50 grpc_tcp_server* server;
51 unsigned port_index;
52 unsigned fd_index;
53 } on_connect_result;
54
55 typedef struct server_weak_ref {
56 grpc_tcp_server* server;
57
58 /* arg is this server_weak_ref. */
59 grpc_closure server_shutdown;
60 } server_weak_ref;
61
62 static on_connect_result g_result = {NULL, 0, 0};
63
on_connect_result_init(on_connect_result * result)64 static void on_connect_result_init(on_connect_result* result) {
65 result->server = NULL;
66 result->port_index = 0;
67 result->fd_index = 0;
68 }
69
on_connect_result_set(on_connect_result * result,const grpc_tcp_server_acceptor * acceptor)70 static void on_connect_result_set(on_connect_result* result,
71 const grpc_tcp_server_acceptor* acceptor) {
72 result->server = grpc_tcp_server_ref(acceptor->from_server);
73 result->port_index = acceptor->port_index;
74 result->fd_index = acceptor->fd_index;
75 }
76
server_weak_ref_shutdown(void * arg,grpc_error * error)77 static void server_weak_ref_shutdown(void* arg, grpc_error* error) {
78 server_weak_ref* weak_ref = static_cast<server_weak_ref*>(arg);
79 weak_ref->server = NULL;
80 }
81
server_weak_ref_init(server_weak_ref * weak_ref)82 static void server_weak_ref_init(server_weak_ref* weak_ref) {
83 weak_ref->server = NULL;
84 GRPC_CLOSURE_INIT(&weak_ref->server_shutdown, server_weak_ref_shutdown,
85 weak_ref, grpc_schedule_on_exec_ctx);
86 }
87
88 /* Make weak_ref->server_shutdown a shutdown_starting cb on server.
89 grpc_tcp_server promises that the server object will live until
90 weak_ref->server_shutdown has returned. A strong ref on grpc_tcp_server
91 should be held until server_weak_ref_set() returns to avoid a race where the
92 server is deleted before the shutdown_starting cb is added. */
server_weak_ref_set(server_weak_ref * weak_ref,grpc_tcp_server * server)93 static void server_weak_ref_set(server_weak_ref* weak_ref,
94 grpc_tcp_server* server) {
95 grpc_tcp_server_shutdown_starting_add(server, &weak_ref->server_shutdown);
96 weak_ref->server = server;
97 }
98
on_connect(void * arg,grpc_endpoint * tcp,grpc_pollset * pollset,grpc_tcp_server_acceptor * acceptor)99 static void on_connect(void* arg, grpc_endpoint* tcp, grpc_pollset* pollset,
100 grpc_tcp_server_acceptor* acceptor) {
101 grpc_endpoint_shutdown(tcp,
102 GRPC_ERROR_CREATE_FROM_STATIC_STRING("Connected"));
103 grpc_endpoint_destroy(tcp);
104
105 on_connect_result temp_result;
106 on_connect_result_set(&temp_result, acceptor);
107 gpr_free(acceptor);
108
109 gpr_mu_lock(g_mu);
110 g_result = temp_result;
111 g_nconnects++;
112 GPR_ASSERT(
113 GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(g_pollset, NULL)));
114 gpr_mu_unlock(g_mu);
115 }
116
test_no_op(void)117 static void test_no_op(void) {
118 grpc_core::ExecCtx exec_ctx;
119 grpc_tcp_server* s;
120 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s));
121 grpc_tcp_server_unref(s);
122 grpc_core::ExecCtx::Get()->Flush();
123 }
124
test_no_op_with_start(void)125 static void test_no_op_with_start(void) {
126 grpc_core::ExecCtx exec_ctx;
127 grpc_tcp_server* s;
128 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s));
129 LOG_TEST("test_no_op_with_start");
130 grpc_tcp_server_start(s, NULL, 0, on_connect, NULL);
131 grpc_tcp_server_unref(s);
132 grpc_core::ExecCtx::Get()->Flush();
133 }
134
test_no_op_with_port(void)135 static void test_no_op_with_port(void) {
136 grpc_core::ExecCtx exec_ctx;
137 grpc_resolved_address resolved_addr;
138 struct sockaddr_in* addr = (struct sockaddr_in*)resolved_addr.addr;
139 grpc_tcp_server* s;
140 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s));
141 LOG_TEST("test_no_op_with_port");
142
143 memset(&resolved_addr, 0, sizeof(resolved_addr));
144 resolved_addr.len = sizeof(struct sockaddr_in);
145 addr->sin_family = AF_INET;
146 int port;
147 GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr, &port) ==
148 GRPC_ERROR_NONE &&
149 port > 0);
150
151 grpc_tcp_server_unref(s);
152 grpc_core::ExecCtx::Get()->Flush();
153 }
154
test_no_op_with_port_and_start(void)155 static void test_no_op_with_port_and_start(void) {
156 grpc_core::ExecCtx exec_ctx;
157 grpc_resolved_address resolved_addr;
158 struct sockaddr_in* addr = (struct sockaddr_in*)resolved_addr.addr;
159 grpc_tcp_server* s;
160 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s));
161 LOG_TEST("test_no_op_with_port_and_start");
162 int port;
163
164 memset(&resolved_addr, 0, sizeof(resolved_addr));
165 resolved_addr.len = sizeof(struct sockaddr_in);
166 addr->sin_family = AF_INET;
167 GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr, &port) ==
168 GRPC_ERROR_NONE &&
169 port > 0);
170
171 grpc_tcp_server_start(s, NULL, 0, on_connect, NULL);
172
173 grpc_tcp_server_unref(s);
174 grpc_core::ExecCtx::Get()->Flush();
175 }
176
connect_cb(uv_connect_t * req,int status)177 static void connect_cb(uv_connect_t* req, int status) {
178 GPR_ASSERT(status == 0);
179 gpr_free(req);
180 }
181
close_cb(uv_handle_t * handle)182 static void close_cb(uv_handle_t* handle) { gpr_free(handle); }
183
tcp_connect(const struct sockaddr * remote,socklen_t remote_len,on_connect_result * result)184 static void tcp_connect(const struct sockaddr* remote, socklen_t remote_len,
185 on_connect_result* result) {
186 gpr_timespec deadline = grpc_timeout_seconds_to_deadline(10);
187 uv_tcp_t* client_handle =
188 static_cast<uv_tcp_t*>(gpr_malloc(sizeof(uv_tcp_t)));
189 uv_connect_t* req =
190 static_cast<uv_connect_t*>(gpr_malloc(sizeof(uv_connect_t)));
191 int nconnects_before;
192
193 gpr_mu_lock(g_mu);
194 nconnects_before = g_nconnects;
195 on_connect_result_init(&g_result);
196 GPR_ASSERT(uv_tcp_init(uv_default_loop(), client_handle) == 0);
197 gpr_log(GPR_DEBUG, "start connect");
198 GPR_ASSERT(uv_tcp_connect(req, client_handle, remote, connect_cb) == 0);
199 gpr_log(GPR_DEBUG, "wait");
200 while (g_nconnects == nconnects_before &&
201 gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) > 0) {
202 grpc_pollset_worker* worker = NULL;
203 GPR_ASSERT(GRPC_LOG_IF_ERROR(
204 "pollset_work",
205 grpc_pollset_work(g_pollset, &worker,
206 grpc_timespec_to_millis_round_up(deadline))));
207 gpr_mu_unlock(g_mu);
208
209 gpr_mu_lock(g_mu);
210 }
211 gpr_log(GPR_DEBUG, "wait done");
212 GPR_ASSERT(g_nconnects == nconnects_before + 1);
213 uv_close((uv_handle_t*)client_handle, close_cb);
214 *result = g_result;
215
216 gpr_mu_unlock(g_mu);
217 }
218
219 /* Tests a tcp server with multiple ports. */
test_connect(unsigned n)220 static void test_connect(unsigned n) {
221 grpc_core::ExecCtx exec_ctx;
222 grpc_resolved_address resolved_addr;
223 grpc_resolved_address resolved_addr1;
224 struct sockaddr_storage* addr = (struct sockaddr_storage*)resolved_addr.addr;
225 struct sockaddr_storage* addr1 =
226 (struct sockaddr_storage*)resolved_addr1.addr;
227 int svr_port;
228 int svr1_port;
229 grpc_tcp_server* s;
230 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(NULL, NULL, &s));
231 unsigned i;
232 server_weak_ref weak_ref;
233 server_weak_ref_init(&weak_ref);
234 LOG_TEST("test_connect");
235 gpr_log(GPR_INFO, "clients=%d", n);
236 memset(&resolved_addr, 0, sizeof(resolved_addr));
237 memset(&resolved_addr1, 0, sizeof(resolved_addr1));
238 resolved_addr.len = sizeof(struct sockaddr_storage);
239 resolved_addr1.len = sizeof(struct sockaddr_storage);
240 addr->ss_family = addr1->ss_family = AF_INET;
241 GPR_ASSERT(GRPC_ERROR_NONE ==
242 grpc_tcp_server_add_port(s, &resolved_addr, &svr_port));
243 GPR_ASSERT(svr_port > 0);
244 GPR_ASSERT((uv_ip6_addr("::", svr_port, (struct sockaddr_in6*)addr)) == 0);
245 /* Cannot use wildcard (port==0), because add_port() will try to reuse the
246 same port as a previous add_port(). */
247 svr1_port = grpc_pick_unused_port_or_die();
248 grpc_sockaddr_set_port(&resolved_addr1, svr1_port);
249 GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr1, &svr_port) ==
250 GRPC_ERROR_NONE &&
251 svr_port == svr1_port);
252
253 grpc_tcp_server_start(s, &g_pollset, 1, on_connect, NULL);
254
255 GPR_ASSERT(uv_ip6_addr("::", svr_port, (struct sockaddr_in6*)addr1) == 0);
256
257 for (i = 0; i < n; i++) {
258 on_connect_result result;
259 on_connect_result_init(&result);
260 tcp_connect((struct sockaddr*)addr, (socklen_t)resolved_addr.len, &result);
261 GPR_ASSERT(result.port_index == 0);
262 GPR_ASSERT(result.server == s);
263 if (weak_ref.server == NULL) {
264 server_weak_ref_set(&weak_ref, result.server);
265 }
266 grpc_tcp_server_unref(result.server);
267
268 on_connect_result_init(&result);
269 tcp_connect((struct sockaddr*)addr1, (socklen_t)resolved_addr1.len,
270 &result);
271 GPR_ASSERT(result.port_index == 1);
272 GPR_ASSERT(result.server == s);
273 grpc_tcp_server_unref(result.server);
274 }
275
276 /* Weak ref to server valid until final unref. */
277 GPR_ASSERT(weak_ref.server != NULL);
278
279 grpc_tcp_server_unref(s);
280 grpc_core::ExecCtx::Get()->Flush();
281 /* Weak ref lost. */
282 GPR_ASSERT(weak_ref.server == NULL);
283 }
284
destroy_pollset(void * p,grpc_error * error)285 static void destroy_pollset(void* p, grpc_error* error) {
286 grpc_pollset_destroy(static_cast<grpc_pollset*>(p));
287 }
288
main(int argc,char ** argv)289 int main(int argc, char** argv) {
290 grpc_closure destroyed;
291 grpc::testing::TestEnvironment env(argc, argv);
292 grpc_init();
293 {
294 grpc_core::ExecCtx exec_ctx;
295 g_pollset = static_cast<grpc_pollset*>(gpr_malloc(grpc_pollset_size()));
296 grpc_pollset_init(g_pollset, &g_mu);
297
298 test_no_op();
299 test_no_op_with_start();
300 test_no_op_with_port();
301 test_no_op_with_port_and_start();
302 test_connect(1);
303 test_connect(10);
304
305 GRPC_CLOSURE_INIT(&destroyed, destroy_pollset, g_pollset,
306 grpc_schedule_on_exec_ctx);
307 grpc_pollset_shutdown(g_pollset, &destroyed);
308
309 gpr_free(g_pollset);
310 }
311 grpc_shutdown();
312 return 0;
313 }
314
315 #else /* GRPC_UV */
316
main(int,char **)317 int main(int /*argc*/, char** /*argv*/) { return 1; }
318
319 #endif /* GRPC_UV */
320