1 /*
2 *
3 * Copyright 2015 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 posix sockets enabled
22 #ifdef GRPC_POSIX_SOCKET_TCP_SERVER
23
24 #include "src/core/lib/iomgr/tcp_server.h"
25
26 #include <errno.h>
27 #include <ifaddrs.h>
28 #include <netinet/in.h>
29 #include <stdio.h>
30 #include <string.h>
31 #include <sys/socket.h>
32 #include <sys/types.h>
33 #include <unistd.h>
34
35 #include <string>
36
37 #include <grpc/grpc.h>
38 #include <grpc/support/alloc.h>
39 #include <grpc/support/log.h>
40 #include <grpc/support/sync.h>
41 #include <grpc/support/time.h>
42
43 #include "src/core/lib/iomgr/error.h"
44 #include "src/core/lib/iomgr/iomgr.h"
45 #include "src/core/lib/iomgr/resolve_address.h"
46 #include "src/core/lib/iomgr/sockaddr_utils.h"
47 #include "test/core/util/port.h"
48 #include "test/core/util/test_config.h"
49
50 #define LOG_TEST(x) gpr_log(GPR_INFO, "%s", #x)
51
52 static gpr_mu* g_mu;
53 static grpc_pollset* g_pollset;
54 static int g_nconnects = 0;
55
56 typedef struct {
57 /* Owns a ref to server. */
58 grpc_tcp_server* server;
59 unsigned port_index;
60 unsigned fd_index;
61 int server_fd;
62 } on_connect_result;
63
64 typedef struct {
65 grpc_tcp_server* server;
66
67 /* arg is this server_weak_ref. */
68 grpc_closure server_shutdown;
69 } server_weak_ref;
70
71 #define MAX_URI 1024
72 typedef struct {
73 grpc_resolved_address addr;
74 char str[MAX_URI];
75 } test_addr;
76
77 #define MAX_ADDRS 100
78 typedef struct {
79 size_t naddrs;
80 test_addr addrs[MAX_ADDRS];
81 } test_addrs;
82
83 static on_connect_result g_result = {nullptr, 0, 0, -1};
84
85 static char family_name_buf[1024];
sock_family_name(int family)86 static const char* sock_family_name(int family) {
87 if (family == AF_INET) {
88 return "AF_INET";
89 } else if (family == AF_INET6) {
90 return "AF_INET6";
91 } else if (family == AF_UNSPEC) {
92 return "AF_UNSPEC";
93 } else {
94 sprintf(family_name_buf, "%d", family);
95 return family_name_buf;
96 }
97 }
98
on_connect_result_init(on_connect_result * result)99 static void on_connect_result_init(on_connect_result* result) {
100 result->server = nullptr;
101 result->port_index = 0;
102 result->fd_index = 0;
103 result->server_fd = -1;
104 }
105
on_connect_result_set(on_connect_result * result,const grpc_tcp_server_acceptor * acceptor)106 static void on_connect_result_set(on_connect_result* result,
107 const grpc_tcp_server_acceptor* acceptor) {
108 result->server = grpc_tcp_server_ref(acceptor->from_server);
109 result->port_index = acceptor->port_index;
110 result->fd_index = acceptor->fd_index;
111 result->server_fd = grpc_tcp_server_port_fd(
112 result->server, acceptor->port_index, acceptor->fd_index);
113 }
114
server_weak_ref_shutdown(void * arg,grpc_error *)115 static void server_weak_ref_shutdown(void* arg, grpc_error* /*error*/) {
116 server_weak_ref* weak_ref = static_cast<server_weak_ref*>(arg);
117 weak_ref->server = nullptr;
118 }
119
server_weak_ref_init(server_weak_ref * weak_ref)120 static void server_weak_ref_init(server_weak_ref* weak_ref) {
121 weak_ref->server = nullptr;
122 GRPC_CLOSURE_INIT(&weak_ref->server_shutdown, server_weak_ref_shutdown,
123 weak_ref, grpc_schedule_on_exec_ctx);
124 }
125
126 /* Make weak_ref->server_shutdown a shutdown_starting cb on server.
127 grpc_tcp_server promises that the server object will live until
128 weak_ref->server_shutdown has returned. A strong ref on grpc_tcp_server
129 should be held until server_weak_ref_set() returns to avoid a race where the
130 server is deleted before the shutdown_starting cb is added. */
server_weak_ref_set(server_weak_ref * weak_ref,grpc_tcp_server * server)131 static void server_weak_ref_set(server_weak_ref* weak_ref,
132 grpc_tcp_server* server) {
133 grpc_tcp_server_shutdown_starting_add(server, &weak_ref->server_shutdown);
134 weak_ref->server = server;
135 }
136
test_addr_init_str(test_addr * addr)137 static void test_addr_init_str(test_addr* addr) {
138 std::string str = grpc_sockaddr_to_string(&addr->addr, false);
139 size_t str_len = std::min(str.size(), sizeof(addr->str) - 1);
140 memcpy(addr->str, str.c_str(), str_len);
141 addr->str[str_len] = '\0';
142 }
143
on_connect(void *,grpc_endpoint * tcp,grpc_pollset *,grpc_tcp_server_acceptor * acceptor)144 static void on_connect(void* /*arg*/, grpc_endpoint* tcp,
145 grpc_pollset* /*pollset*/,
146 grpc_tcp_server_acceptor* acceptor) {
147 grpc_endpoint_shutdown(tcp,
148 GRPC_ERROR_CREATE_FROM_STATIC_STRING("Connected"));
149 grpc_endpoint_destroy(tcp);
150
151 on_connect_result temp_result;
152 on_connect_result_set(&temp_result, acceptor);
153 gpr_free(acceptor);
154
155 gpr_mu_lock(g_mu);
156 g_result = temp_result;
157 g_nconnects++;
158 GPR_ASSERT(
159 GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(g_pollset, nullptr)));
160 gpr_mu_unlock(g_mu);
161 }
162
test_no_op(void)163 static void test_no_op(void) {
164 grpc_core::ExecCtx exec_ctx;
165 grpc_tcp_server* s;
166 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(nullptr, nullptr, &s));
167 grpc_tcp_server_unref(s);
168 }
169
test_no_op_with_start(void)170 static void test_no_op_with_start(void) {
171 grpc_core::ExecCtx exec_ctx;
172 grpc_tcp_server* s;
173 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(nullptr, nullptr, &s));
174 LOG_TEST("test_no_op_with_start");
175 std::vector<grpc_pollset*> empty_pollset;
176 grpc_tcp_server_start(s, &empty_pollset, on_connect, nullptr);
177 grpc_tcp_server_unref(s);
178 }
179
test_no_op_with_port(void)180 static void test_no_op_with_port(void) {
181 grpc_core::ExecCtx exec_ctx;
182 grpc_resolved_address resolved_addr;
183 struct sockaddr_in* addr =
184 reinterpret_cast<struct sockaddr_in*>(resolved_addr.addr);
185 grpc_tcp_server* s;
186 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(nullptr, nullptr, &s));
187 LOG_TEST("test_no_op_with_port");
188
189 memset(&resolved_addr, 0, sizeof(resolved_addr));
190 resolved_addr.len = static_cast<socklen_t>(sizeof(struct sockaddr_in));
191 addr->sin_family = AF_INET;
192 int port = -1;
193 GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr, &port) ==
194 GRPC_ERROR_NONE &&
195 port > 0);
196
197 grpc_tcp_server_unref(s);
198 }
199
test_no_op_with_port_and_start(void)200 static void test_no_op_with_port_and_start(void) {
201 grpc_core::ExecCtx exec_ctx;
202 grpc_resolved_address resolved_addr;
203 struct sockaddr_in* addr =
204 reinterpret_cast<struct sockaddr_in*>(resolved_addr.addr);
205 grpc_tcp_server* s;
206 GPR_ASSERT(GRPC_ERROR_NONE == grpc_tcp_server_create(nullptr, nullptr, &s));
207 LOG_TEST("test_no_op_with_port_and_start");
208 int port = -1;
209
210 memset(&resolved_addr, 0, sizeof(resolved_addr));
211 resolved_addr.len = static_cast<socklen_t>(sizeof(struct sockaddr_in));
212 addr->sin_family = AF_INET;
213 GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr, &port) ==
214 GRPC_ERROR_NONE &&
215 port > 0);
216
217 std::vector<grpc_pollset*> empty_pollset;
218 grpc_tcp_server_start(s, &empty_pollset, on_connect, nullptr);
219
220 grpc_tcp_server_unref(s);
221 }
222
tcp_connect(const test_addr * remote,on_connect_result * result)223 static grpc_error* tcp_connect(const test_addr* remote,
224 on_connect_result* result) {
225 grpc_millis deadline =
226 grpc_timespec_to_millis_round_up(grpc_timeout_seconds_to_deadline(10));
227 int clifd;
228 int nconnects_before;
229 const struct sockaddr* remote_addr =
230 reinterpret_cast<const struct sockaddr*>(remote->addr.addr);
231
232 gpr_log(GPR_INFO, "Connecting to %s", remote->str);
233 gpr_mu_lock(g_mu);
234 nconnects_before = g_nconnects;
235 on_connect_result_init(&g_result);
236 clifd = socket(remote_addr->sa_family, SOCK_STREAM, 0);
237 if (clifd < 0) {
238 gpr_mu_unlock(g_mu);
239 return GRPC_OS_ERROR(errno, "Failed to create socket");
240 }
241 gpr_log(GPR_DEBUG, "start connect to %s", remote->str);
242 if (connect(clifd, remote_addr, static_cast<socklen_t>(remote->addr.len)) !=
243 0) {
244 gpr_mu_unlock(g_mu);
245 close(clifd);
246 return GRPC_OS_ERROR(errno, "connect");
247 }
248 gpr_log(GPR_DEBUG, "wait");
249 while (g_nconnects == nconnects_before &&
250 deadline > grpc_core::ExecCtx::Get()->Now()) {
251 grpc_pollset_worker* worker = nullptr;
252 grpc_error* err;
253 if ((err = grpc_pollset_work(g_pollset, &worker, deadline)) !=
254 GRPC_ERROR_NONE) {
255 gpr_mu_unlock(g_mu);
256 close(clifd);
257 return err;
258 }
259 gpr_mu_unlock(g_mu);
260
261 gpr_mu_lock(g_mu);
262 }
263 gpr_log(GPR_DEBUG, "wait done");
264 if (g_nconnects != nconnects_before + 1) {
265 gpr_mu_unlock(g_mu);
266 close(clifd);
267 return GRPC_ERROR_CREATE_FROM_STATIC_STRING("Didn't connect");
268 }
269 close(clifd);
270 *result = g_result;
271
272 gpr_mu_unlock(g_mu);
273 gpr_log(GPR_INFO, "Result (%d, %d) fd %d", result->port_index,
274 result->fd_index, result->server_fd);
275 grpc_tcp_server_unref(result->server);
276 return GRPC_ERROR_NONE;
277 }
278
279 /* Tests a tcp server on "::" listeners with multiple ports. If channel_args is
280 non-NULL, pass them to the server. If dst_addrs is non-NULL, use valid addrs
281 as destination addrs (port is not set). If dst_addrs is NULL, use listener
282 addrs as destination addrs. If test_dst_addrs is true, test connectivity with
283 each destination address, set grpc_resolved_address::len=0 for failures, but
284 don't fail the overall unitest. */
test_connect(size_t num_connects,const grpc_channel_args * channel_args,test_addrs * dst_addrs,bool test_dst_addrs)285 static void test_connect(size_t num_connects,
286 const grpc_channel_args* channel_args,
287 test_addrs* dst_addrs, bool test_dst_addrs) {
288 grpc_core::ExecCtx exec_ctx;
289 grpc_resolved_address resolved_addr;
290 grpc_resolved_address resolved_addr1;
291 struct sockaddr_storage* const addr =
292 reinterpret_cast<struct sockaddr_storage*>(resolved_addr.addr);
293 struct sockaddr_storage* const addr1 =
294 reinterpret_cast<struct sockaddr_storage*>(resolved_addr1.addr);
295 unsigned svr_fd_count;
296 int port;
297 int svr_port;
298 unsigned svr1_fd_count;
299 int svr1_port;
300 grpc_tcp_server* s;
301 const unsigned num_ports = 2;
302 GPR_ASSERT(GRPC_ERROR_NONE ==
303 grpc_tcp_server_create(nullptr, channel_args, &s));
304 unsigned port_num;
305 server_weak_ref weak_ref;
306 server_weak_ref_init(&weak_ref);
307 server_weak_ref_set(&weak_ref, s);
308 LOG_TEST("test_connect");
309 gpr_log(GPR_INFO,
310 "clients=%lu, num chan args=%lu, remote IP=%s, test_dst_addrs=%d",
311 static_cast<unsigned long>(num_connects),
312 static_cast<unsigned long>(
313 channel_args != nullptr ? channel_args->num_args : 0),
314 dst_addrs != nullptr ? "<specific>" : "::", test_dst_addrs);
315 memset(&resolved_addr, 0, sizeof(resolved_addr));
316 memset(&resolved_addr1, 0, sizeof(resolved_addr1));
317 resolved_addr.len = static_cast<socklen_t>(sizeof(struct sockaddr_storage));
318 resolved_addr1.len = static_cast<socklen_t>(sizeof(struct sockaddr_storage));
319 addr->ss_family = addr1->ss_family = AF_INET;
320 GPR_ASSERT(GRPC_LOG_IF_ERROR(
321 "grpc_tcp_server_add_port",
322 grpc_tcp_server_add_port(s, &resolved_addr, &svr_port)));
323 gpr_log(GPR_INFO, "Allocated port %d", svr_port);
324 GPR_ASSERT(svr_port > 0);
325 /* Cannot use wildcard (port==0), because add_port() will try to reuse the
326 same port as a previous add_port(). */
327 svr1_port = grpc_pick_unused_port_or_die();
328 GPR_ASSERT(svr1_port > 0);
329 gpr_log(GPR_INFO, "Picked unused port %d", svr1_port);
330 grpc_sockaddr_set_port(&resolved_addr1, svr1_port);
331 GPR_ASSERT(grpc_tcp_server_add_port(s, &resolved_addr1, &port) ==
332 GRPC_ERROR_NONE &&
333 port == svr1_port);
334
335 /* Bad port_index. */
336 GPR_ASSERT(grpc_tcp_server_port_fd_count(s, 2) == 0);
337 GPR_ASSERT(grpc_tcp_server_port_fd(s, 2, 0) < 0);
338
339 /* Bad fd_index. */
340 GPR_ASSERT(grpc_tcp_server_port_fd(s, 0, 100) < 0);
341 GPR_ASSERT(grpc_tcp_server_port_fd(s, 1, 100) < 0);
342
343 /* Got at least one fd per port. */
344 svr_fd_count = grpc_tcp_server_port_fd_count(s, 0);
345 GPR_ASSERT(svr_fd_count >= 1);
346 svr1_fd_count = grpc_tcp_server_port_fd_count(s, 1);
347 GPR_ASSERT(svr1_fd_count >= 1);
348
349 std::vector<grpc_pollset*> test_pollset;
350 test_pollset.push_back(g_pollset);
351 grpc_tcp_server_start(s, &test_pollset, on_connect, nullptr);
352
353 if (dst_addrs != nullptr) {
354 int ports[] = {svr_port, svr1_port};
355 for (port_num = 0; port_num < num_ports; ++port_num) {
356 size_t dst_idx;
357 size_t num_tested = 0;
358 for (dst_idx = 0; dst_idx < dst_addrs->naddrs; ++dst_idx) {
359 test_addr dst = dst_addrs->addrs[dst_idx];
360 on_connect_result result;
361 grpc_error* err;
362 if (dst.addr.len == 0) {
363 gpr_log(GPR_DEBUG, "Skipping test of non-functional local IP %s",
364 dst.str);
365 continue;
366 }
367 GPR_ASSERT(grpc_sockaddr_set_port(&dst.addr, ports[port_num]));
368 test_addr_init_str(&dst);
369 ++num_tested;
370 on_connect_result_init(&result);
371 if ((err = tcp_connect(&dst, &result)) == GRPC_ERROR_NONE &&
372 result.server_fd >= 0 && result.server == s) {
373 continue;
374 }
375 gpr_log(GPR_ERROR, "Failed to connect to %s: %s", dst.str,
376 grpc_error_string(err));
377 GPR_ASSERT(test_dst_addrs);
378 dst_addrs->addrs[dst_idx].addr.len = 0;
379 GRPC_ERROR_UNREF(err);
380 }
381 GPR_ASSERT(num_tested > 0);
382 }
383 } else {
384 for (port_num = 0; port_num < num_ports; ++port_num) {
385 const unsigned num_fds = grpc_tcp_server_port_fd_count(s, port_num);
386 unsigned fd_num;
387 for (fd_num = 0; fd_num < num_fds; ++fd_num) {
388 int fd = grpc_tcp_server_port_fd(s, port_num, fd_num);
389 size_t connect_num;
390 test_addr dst;
391 GPR_ASSERT(fd >= 0);
392 dst.addr.len = static_cast<socklen_t>(sizeof(dst.addr.addr));
393 GPR_ASSERT(getsockname(fd, (struct sockaddr*)dst.addr.addr,
394 (socklen_t*)&dst.addr.len) == 0);
395 GPR_ASSERT(dst.addr.len <= sizeof(dst.addr.addr));
396 test_addr_init_str(&dst);
397 gpr_log(GPR_INFO, "(%d, %d) fd %d family %s listening on %s", port_num,
398 fd_num, fd, sock_family_name(addr->ss_family), dst.str);
399 for (connect_num = 0; connect_num < num_connects; ++connect_num) {
400 on_connect_result result;
401 on_connect_result_init(&result);
402 GPR_ASSERT(
403 GRPC_LOG_IF_ERROR("tcp_connect", tcp_connect(&dst, &result)));
404 GPR_ASSERT(result.server_fd == fd);
405 GPR_ASSERT(result.port_index == port_num);
406 GPR_ASSERT(result.fd_index == fd_num);
407 GPR_ASSERT(result.server == s);
408 GPR_ASSERT(
409 grpc_tcp_server_port_fd(s, result.port_index, result.fd_index) ==
410 result.server_fd);
411 }
412 }
413 }
414 }
415 /* Weak ref to server valid until final unref. */
416 GPR_ASSERT(weak_ref.server != nullptr);
417 GPR_ASSERT(grpc_tcp_server_port_fd(s, 0, 0) >= 0);
418
419 grpc_tcp_server_unref(s);
420 grpc_core::ExecCtx::Get()->Flush();
421
422 /* Weak ref lost. */
423 GPR_ASSERT(weak_ref.server == nullptr);
424 }
425
destroy_pollset(void * p,grpc_error *)426 static void destroy_pollset(void* p, grpc_error* /*error*/) {
427 grpc_pollset_destroy(static_cast<grpc_pollset*>(p));
428 }
429
main(int argc,char ** argv)430 int main(int argc, char** argv) {
431 grpc_closure destroyed;
432 grpc_arg chan_args[1];
433 chan_args[0].type = GRPC_ARG_INTEGER;
434 chan_args[0].key = const_cast<char*>(GRPC_ARG_EXPAND_WILDCARD_ADDRS);
435 chan_args[0].value.integer = 1;
436 const grpc_channel_args channel_args = {1, chan_args};
437 struct ifaddrs* ifa = nullptr;
438 struct ifaddrs* ifa_it;
439 // Zalloc dst_addrs to avoid oversized frames.
440 test_addrs* dst_addrs =
441 static_cast<test_addrs*>(gpr_zalloc(sizeof(*dst_addrs)));
442 grpc::testing::TestEnvironment env(argc, argv);
443 grpc_init();
444 // wait a few seconds to make sure IPv6 link-local addresses can be bound
445 // if we are running under docker container that has just started.
446 // See https://github.com/moby/moby/issues/38491
447 // See https://github.com/grpc/grpc/issues/15610
448 gpr_sleep_until(grpc_timeout_seconds_to_deadline(4));
449 {
450 grpc_core::ExecCtx exec_ctx;
451 g_pollset = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
452 grpc_pollset_init(g_pollset, &g_mu);
453
454 test_no_op();
455 test_no_op_with_start();
456 test_no_op_with_port();
457 test_no_op_with_port_and_start();
458
459 if (getifaddrs(&ifa) != 0 || ifa == nullptr) {
460 gpr_log(GPR_ERROR, "getifaddrs: %s", strerror(errno));
461 return EXIT_FAILURE;
462 }
463 dst_addrs->naddrs = 0;
464 for (ifa_it = ifa; ifa_it != nullptr && dst_addrs->naddrs < MAX_ADDRS;
465 ifa_it = ifa_it->ifa_next) {
466 if (ifa_it->ifa_addr == nullptr) {
467 continue;
468 } else if (ifa_it->ifa_addr->sa_family == AF_INET) {
469 dst_addrs->addrs[dst_addrs->naddrs].addr.len =
470 static_cast<socklen_t>(sizeof(struct sockaddr_in));
471 } else if (ifa_it->ifa_addr->sa_family == AF_INET6) {
472 dst_addrs->addrs[dst_addrs->naddrs].addr.len =
473 static_cast<socklen_t>(sizeof(struct sockaddr_in6));
474 } else {
475 continue;
476 }
477 memcpy(dst_addrs->addrs[dst_addrs->naddrs].addr.addr, ifa_it->ifa_addr,
478 dst_addrs->addrs[dst_addrs->naddrs].addr.len);
479 GPR_ASSERT(
480 grpc_sockaddr_set_port(&dst_addrs->addrs[dst_addrs->naddrs].addr, 0));
481 test_addr_init_str(&dst_addrs->addrs[dst_addrs->naddrs]);
482 ++dst_addrs->naddrs;
483 }
484 freeifaddrs(ifa);
485 ifa = nullptr;
486
487 /* Connect to same addresses as listeners. */
488 test_connect(1, nullptr, nullptr, false);
489 test_connect(10, nullptr, nullptr, false);
490
491 /* Set dst_addrs->addrs[i].len=0 for dst_addrs that are unreachable with a
492 "::" listener. */
493 test_connect(1, nullptr, dst_addrs, true);
494
495 /* Test connect(2) with dst_addrs. */
496 test_connect(1, &channel_args, dst_addrs, false);
497 /* Test connect(2) with dst_addrs. */
498 test_connect(10, &channel_args, dst_addrs, false);
499
500 GRPC_CLOSURE_INIT(&destroyed, destroy_pollset, g_pollset,
501 grpc_schedule_on_exec_ctx);
502 grpc_pollset_shutdown(g_pollset, &destroyed);
503 }
504 grpc_shutdown();
505 gpr_free(dst_addrs);
506 gpr_free(g_pollset);
507 return EXIT_SUCCESS;
508 }
509
510 #else /* GRPC_POSIX_SOCKET_SERVER */
511
main(int argc,char ** argv)512 int main(int argc, char** argv) { return 1; }
513
514 #endif /* GRPC_POSIX_SOCKET_SERVER */
515