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 <string.h>
27
28 #include "src/core/lib/iomgr/tcp_client.h"
29
30 #include <grpc/grpc.h>
31 #include <grpc/support/alloc.h>
32 #include <grpc/support/log.h>
33 #include <grpc/support/time.h>
34
35 #include "src/core/lib/iomgr/iomgr.h"
36 #include "src/core/lib/iomgr/pollset.h"
37 #include "src/core/lib/iomgr/timer.h"
38 #include "test/core/util/test_config.h"
39
40 static gpr_mu* g_mu;
41 static grpc_pollset* g_pollset;
42 static int g_connections_complete = 0;
43 static grpc_endpoint* g_connecting = NULL;
44
test_deadline(void)45 static grpc_millis test_deadline(void) {
46 return grpc_timespec_to_millis_round_up(grpc_timeout_seconds_to_deadline(10));
47 }
48
finish_connection()49 static void finish_connection() {
50 gpr_mu_lock(g_mu);
51 g_connections_complete++;
52 GPR_ASSERT(
53 GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(g_pollset, NULL)));
54 gpr_mu_unlock(g_mu);
55 }
56
must_succeed(void * arg,grpc_error * error)57 static void must_succeed(void* arg, grpc_error* error) {
58 GPR_ASSERT(g_connecting != NULL);
59 GPR_ASSERT(error == GRPC_ERROR_NONE);
60 grpc_endpoint_shutdown(g_connecting, GRPC_ERROR_CREATE_FROM_STATIC_STRING(
61 "must_succeed called"));
62 grpc_endpoint_destroy(g_connecting);
63 g_connecting = NULL;
64 finish_connection();
65 }
66
must_fail(void * arg,grpc_error * error)67 static void must_fail(void* arg, grpc_error* error) {
68 GPR_ASSERT(g_connecting == NULL);
69 GPR_ASSERT(error != GRPC_ERROR_NONE);
70 finish_connection();
71 }
72
close_cb(uv_handle_t * handle)73 static void close_cb(uv_handle_t* handle) { gpr_free(handle); }
74
connection_cb(uv_stream_t * server,int status)75 static void connection_cb(uv_stream_t* server, int status) {
76 uv_tcp_t* client_handle =
77 static_cast<uv_tcp_t*>(gpr_malloc(sizeof(uv_tcp_t)));
78 GPR_ASSERT(0 == status);
79 GPR_ASSERT(0 == uv_tcp_init(uv_default_loop(), client_handle));
80 GPR_ASSERT(0 == uv_accept(server, (uv_stream_t*)client_handle));
81 uv_close((uv_handle_t*)client_handle, close_cb);
82 }
83
test_succeeds(void)84 void test_succeeds(void) {
85 grpc_resolved_address resolved_addr;
86 struct sockaddr_in* addr = (struct sockaddr_in*)resolved_addr.addr;
87 uv_tcp_t* svr_handle = static_cast<uv_tcp_t*>(gpr_malloc(sizeof(uv_tcp_t)));
88 int connections_complete_before;
89 grpc_closure done;
90 grpc_core::ExecCtx exec_ctx;
91
92 gpr_log(GPR_DEBUG, "test_succeeds");
93
94 memset(&resolved_addr, 0, sizeof(resolved_addr));
95 resolved_addr.len = sizeof(struct sockaddr_in);
96 addr->sin_family = AF_INET;
97
98 /* create a dummy server */
99 GPR_ASSERT(0 == uv_tcp_init(uv_default_loop(), svr_handle));
100 GPR_ASSERT(0 == uv_tcp_bind(svr_handle, (struct sockaddr*)addr, 0));
101 GPR_ASSERT(0 == uv_listen((uv_stream_t*)svr_handle, 1, connection_cb));
102
103 gpr_mu_lock(g_mu);
104 connections_complete_before = g_connections_complete;
105 gpr_mu_unlock(g_mu);
106
107 /* connect to it */
108 GPR_ASSERT(uv_tcp_getsockname(svr_handle, (struct sockaddr*)addr,
109 (int*)&resolved_addr.len) == 0);
110 GRPC_CLOSURE_INIT(&done, must_succeed, NULL, grpc_schedule_on_exec_ctx);
111 grpc_tcp_client_connect(&done, &g_connecting, NULL, NULL, &resolved_addr,
112 GRPC_MILLIS_INF_FUTURE);
113
114 gpr_mu_lock(g_mu);
115
116 while (g_connections_complete == connections_complete_before) {
117 grpc_pollset_worker* worker = NULL;
118 GPR_ASSERT(GRPC_LOG_IF_ERROR(
119 "pollset_work",
120 grpc_pollset_work(g_pollset, &worker,
121 grpc_timespec_to_millis_round_up(
122 grpc_timeout_seconds_to_deadline(5)))));
123 gpr_mu_unlock(g_mu);
124 grpc_core::ExecCtx::Get()->Flush();
125 gpr_mu_lock(g_mu);
126 }
127
128 // This will get cleaned up when the pollset runs again or gets shutdown
129 uv_close((uv_handle_t*)svr_handle, close_cb);
130
131 gpr_mu_unlock(g_mu);
132 }
133
test_fails(void)134 void test_fails(void) {
135 grpc_resolved_address resolved_addr;
136 struct sockaddr_in* addr = (struct sockaddr_in*)resolved_addr.addr;
137 int connections_complete_before;
138 grpc_closure done;
139 grpc_core::ExecCtx exec_ctx;
140
141 gpr_log(GPR_DEBUG, "test_fails");
142
143 memset(&resolved_addr, 0, sizeof(resolved_addr));
144 resolved_addr.len = sizeof(struct sockaddr_in);
145 addr->sin_family = AF_INET;
146
147 gpr_mu_lock(g_mu);
148 connections_complete_before = g_connections_complete;
149 gpr_mu_unlock(g_mu);
150
151 /* connect to a broken address */
152 GRPC_CLOSURE_INIT(&done, must_fail, NULL, grpc_schedule_on_exec_ctx);
153 grpc_tcp_client_connect(&done, &g_connecting, NULL, NULL, &resolved_addr,
154 GRPC_MILLIS_INF_FUTURE);
155
156 gpr_mu_lock(g_mu);
157
158 /* wait for the connection callback to finish */
159 while (g_connections_complete == connections_complete_before) {
160 grpc_pollset_worker* worker = NULL;
161 gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
162 grpc_millis polling_deadline = test_deadline();
163 switch (grpc_timer_check(&polling_deadline)) {
164 case GRPC_TIMERS_FIRED:
165 break;
166 case GRPC_TIMERS_NOT_CHECKED:
167 polling_deadline = grpc_timespec_to_millis_round_up(now);
168 /* fall through */
169 case GRPC_TIMERS_CHECKED_AND_EMPTY:
170 GPR_ASSERT(GRPC_LOG_IF_ERROR(
171 "pollset_work",
172 grpc_pollset_work(g_pollset, &worker, polling_deadline)));
173 break;
174 }
175 gpr_mu_unlock(g_mu);
176 grpc_core::ExecCtx::Get()->Flush();
177 gpr_mu_lock(g_mu);
178 }
179
180 gpr_mu_unlock(g_mu);
181 }
182
destroy_pollset(void * p,grpc_error * error)183 static void destroy_pollset(void* p, grpc_error* error) {
184 grpc_pollset_destroy(static_cast<grpc_pollset*>(p));
185 }
186
main(int argc,char ** argv)187 int main(int argc, char** argv) {
188 grpc_closure destroyed;
189 grpc_core::ExecCtx exec_ctx;
190 grpc_test_init(argc, argv);
191 grpc_init();
192 g_pollset = static_cast<grpc_pollset*>(gpr_malloc(grpc_pollset_size()));
193 grpc_pollset_init(g_pollset, &g_mu);
194
195 test_succeeds();
196 gpr_log(GPR_ERROR, "End of first test");
197 test_fails();
198 GRPC_CLOSURE_INIT(&destroyed, destroy_pollset, g_pollset,
199 grpc_schedule_on_exec_ctx);
200 grpc_pollset_shutdown(g_pollset, &destroyed);
201
202 grpc_shutdown();
203 gpr_free(g_pollset);
204 return 0;
205 }
206
207 #else /* GRPC_UV */
208
main(int argc,char ** argv)209 int main(int argc, char** argv) { return 1; }
210
211 #endif /* GRPC_UV */
212