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 grpc_core::ExecCtx::Get()->Flush();
133 }
134
test_fails(void)135 void test_fails(void) {
136 grpc_resolved_address resolved_addr;
137 struct sockaddr_in* addr = (struct sockaddr_in*)resolved_addr.addr;
138 int connections_complete_before;
139 grpc_closure done;
140 grpc_core::ExecCtx exec_ctx;
141
142 gpr_log(GPR_DEBUG, "test_fails");
143
144 memset(&resolved_addr, 0, sizeof(resolved_addr));
145 resolved_addr.len = sizeof(struct sockaddr_in);
146 addr->sin_family = AF_INET;
147
148 gpr_mu_lock(g_mu);
149 connections_complete_before = g_connections_complete;
150 gpr_mu_unlock(g_mu);
151
152 /* connect to a broken address */
153 GRPC_CLOSURE_INIT(&done, must_fail, NULL, grpc_schedule_on_exec_ctx);
154 grpc_tcp_client_connect(&done, &g_connecting, NULL, NULL, &resolved_addr,
155 GRPC_MILLIS_INF_FUTURE);
156
157 gpr_mu_lock(g_mu);
158
159 /* wait for the connection callback to finish */
160 while (g_connections_complete == connections_complete_before) {
161 grpc_pollset_worker* worker = NULL;
162 gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
163 grpc_millis polling_deadline = test_deadline();
164 switch (grpc_timer_check(&polling_deadline)) {
165 case GRPC_TIMERS_FIRED:
166 break;
167 case GRPC_TIMERS_NOT_CHECKED:
168 polling_deadline = grpc_timespec_to_millis_round_up(now);
169 // fallthrough
170 case GRPC_TIMERS_CHECKED_AND_EMPTY:
171 GPR_ASSERT(GRPC_LOG_IF_ERROR(
172 "pollset_work",
173 grpc_pollset_work(g_pollset, &worker, polling_deadline)));
174 break;
175 }
176 gpr_mu_unlock(g_mu);
177 grpc_core::ExecCtx::Get()->Flush();
178 gpr_mu_lock(g_mu);
179 }
180
181 gpr_mu_unlock(g_mu);
182 grpc_core::ExecCtx::Get()->Flush();
183 }
184
destroy_pollset(void * p,grpc_error * error)185 static void destroy_pollset(void* p, grpc_error* error) {
186 grpc_pollset_destroy(static_cast<grpc_pollset*>(p));
187 }
188
main(int argc,char ** argv)189 int main(int argc, char** argv) {
190 grpc_closure destroyed;
191 grpc::testing::TestEnvironment env(argc, argv);
192 grpc_init();
193 {
194 grpc_core::ExecCtx exec_ctx;
195 g_pollset = static_cast<grpc_pollset*>(gpr_malloc(grpc_pollset_size()));
196 grpc_pollset_init(g_pollset, &g_mu);
197
198 test_succeeds();
199 gpr_log(GPR_ERROR, "End of first test");
200 test_fails();
201 GRPC_CLOSURE_INIT(&destroyed, destroy_pollset, g_pollset,
202 grpc_schedule_on_exec_ctx);
203 grpc_pollset_shutdown(g_pollset, &destroyed);
204 gpr_free(g_pollset);
205 }
206 grpc_shutdown();
207 return 0;
208 }
209
210 #else /* GRPC_UV */
211
main(int,char **)212 int main(int /*argc*/, char** /*argv*/) { return 1; }
213
214 #endif /* GRPC_UV */
215