Line data Source code
1 : /*
2 : *
3 : * Copyright 2015, Google Inc.
4 : * All rights reserved.
5 : *
6 : * Redistribution and use in source and binary forms, with or without
7 : * modification, are permitted provided that the following conditions are
8 : * met:
9 : *
10 : * * Redistributions of source code must retain the above copyright
11 : * notice, this list of conditions and the following disclaimer.
12 : * * Redistributions in binary form must reproduce the above
13 : * copyright notice, this list of conditions and the following disclaimer
14 : * in the documentation and/or other materials provided with the
15 : * distribution.
16 : * * Neither the name of Google Inc. nor the names of its
17 : * contributors may be used to endorse or promote products derived from
18 : * this software without specific prior written permission.
19 : *
20 : * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 : * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 : * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 : * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 : * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 : * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 : * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 : * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 : * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 : * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 : * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 : *
32 : */
33 :
34 : #include <grpc/support/port_platform.h>
35 :
36 : #ifdef GPR_POSIX_SOCKET
37 :
38 : #include "src/core/iomgr/tcp_client.h"
39 :
40 : #include <errno.h>
41 : #include <netinet/in.h>
42 : #include <string.h>
43 : #include <unistd.h>
44 :
45 : #include "src/core/iomgr/timer.h"
46 : #include "src/core/iomgr/iomgr_posix.h"
47 : #include "src/core/iomgr/pollset_posix.h"
48 : #include "src/core/iomgr/sockaddr_utils.h"
49 : #include "src/core/iomgr/socket_utils_posix.h"
50 : #include "src/core/iomgr/tcp_posix.h"
51 : #include "src/core/support/string.h"
52 : #include <grpc/support/alloc.h>
53 : #include <grpc/support/log.h>
54 : #include <grpc/support/string_util.h>
55 : #include <grpc/support/time.h>
56 :
57 : extern int grpc_tcp_trace;
58 :
59 : typedef struct {
60 : gpr_mu mu;
61 : grpc_fd *fd;
62 : gpr_timespec deadline;
63 : grpc_timer alarm;
64 : int refs;
65 : grpc_closure write_closure;
66 : grpc_pollset_set *interested_parties;
67 : char *addr_str;
68 : grpc_endpoint **ep;
69 : grpc_closure *closure;
70 : } async_connect;
71 :
72 11797 : static int prepare_socket(const struct sockaddr *addr, int fd) {
73 11797 : if (fd < 0) {
74 0 : goto error;
75 : }
76 :
77 23594 : if (!grpc_set_socket_nonblocking(fd, 1) || !grpc_set_socket_cloexec(fd, 1) ||
78 34899 : (addr->sa_family != AF_UNIX && !grpc_set_socket_low_latency(fd, 1)) ||
79 11797 : !grpc_set_socket_no_sigpipe_if_possible(fd)) {
80 0 : gpr_log(GPR_ERROR, "Unable to configure socket %d: %s", fd,
81 0 : strerror(errno));
82 0 : goto error;
83 : }
84 :
85 11712 : return 1;
86 :
87 : error:
88 0 : if (fd >= 0) {
89 0 : close(fd);
90 : }
91 0 : return 0;
92 : }
93 :
94 11263 : static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int success) {
95 : int done;
96 11220 : async_connect *ac = acp;
97 11263 : if (grpc_tcp_trace) {
98 0 : gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: on_alarm: success=%d", ac->addr_str,
99 : success);
100 : }
101 11263 : gpr_mu_lock(&ac->mu);
102 11263 : if (ac->fd != NULL) {
103 277 : grpc_fd_shutdown(exec_ctx, ac->fd);
104 : }
105 11263 : done = (--ac->refs == 0);
106 11263 : gpr_mu_unlock(&ac->mu);
107 11263 : if (done) {
108 10986 : gpr_mu_destroy(&ac->mu);
109 10986 : gpr_free(ac->addr_str);
110 10986 : gpr_free(ac);
111 : }
112 11263 : }
113 :
114 11263 : static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, int success) {
115 11220 : async_connect *ac = acp;
116 11263 : int so_error = 0;
117 : socklen_t so_error_size;
118 : int err;
119 : int done;
120 11263 : grpc_endpoint **ep = ac->ep;
121 11263 : grpc_closure *closure = ac->closure;
122 : grpc_fd *fd;
123 :
124 11263 : if (grpc_tcp_trace) {
125 0 : gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: on_writable: success=%d",
126 : ac->addr_str, success);
127 : }
128 :
129 11263 : gpr_mu_lock(&ac->mu);
130 11263 : GPR_ASSERT(ac->fd);
131 11220 : fd = ac->fd;
132 11263 : ac->fd = NULL;
133 11263 : gpr_mu_unlock(&ac->mu);
134 :
135 11263 : grpc_timer_cancel(exec_ctx, &ac->alarm);
136 :
137 11263 : gpr_mu_lock(&ac->mu);
138 11263 : if (success) {
139 : do {
140 10986 : so_error_size = sizeof(so_error);
141 10986 : err = getsockopt(fd->fd, SOL_SOCKET, SO_ERROR, &so_error, &so_error_size);
142 10986 : } while (err < 0 && errno == EINTR);
143 10986 : if (err < 0) {
144 0 : gpr_log(GPR_ERROR, "failed to connect to '%s': getsockopt(ERROR): %s",
145 0 : ac->addr_str, strerror(errno));
146 0 : goto finish;
147 10986 : } else if (so_error != 0) {
148 4746 : if (so_error == ENOBUFS) {
149 : /* We will get one of these errors if we have run out of
150 : memory in the kernel for the data structures allocated
151 : when you connect a socket. If this happens it is very
152 : likely that if we wait a little bit then try again the
153 : connection will work (since other programs or this
154 : program will close their network connections and free up
155 : memory). This does _not_ indicate that there is anything
156 : wrong with the server we are connecting to, this is a
157 : local problem.
158 :
159 : If you are looking at this code, then chances are that
160 : your program or another program on the same computer
161 : opened too many network connections. The "easy" fix:
162 : don't do that! */
163 0 : gpr_log(GPR_ERROR, "kernel out of buffers");
164 0 : gpr_mu_unlock(&ac->mu);
165 0 : grpc_fd_notify_on_write(exec_ctx, fd, &ac->write_closure);
166 11263 : return;
167 : } else {
168 4746 : switch (so_error) {
169 : case ECONNREFUSED:
170 4746 : gpr_log(
171 : GPR_ERROR,
172 : "failed to connect to '%s': socket error: connection refused",
173 : ac->addr_str);
174 4746 : break;
175 : default:
176 0 : gpr_log(GPR_ERROR, "failed to connect to '%s': socket error: %d",
177 : ac->addr_str, so_error);
178 0 : break;
179 : }
180 4744 : goto finish;
181 : }
182 : } else {
183 6240 : grpc_pollset_set_del_fd(exec_ctx, ac->interested_parties, fd);
184 6240 : *ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str);
185 6199 : fd = NULL;
186 6240 : goto finish;
187 : }
188 : } else {
189 277 : gpr_log(GPR_ERROR, "failed to connect to '%s': timeout occurred",
190 : ac->addr_str);
191 277 : goto finish;
192 : }
193 :
194 : GPR_UNREACHABLE_CODE(return );
195 :
196 : finish:
197 11263 : if (fd != NULL) {
198 5023 : grpc_pollset_set_del_fd(exec_ctx, ac->interested_parties, fd);
199 5023 : grpc_fd_orphan(exec_ctx, fd, NULL, NULL, "tcp_client_orphan");
200 5021 : fd = NULL;
201 : }
202 11263 : done = (--ac->refs == 0);
203 11263 : gpr_mu_unlock(&ac->mu);
204 11263 : if (done) {
205 277 : gpr_mu_destroy(&ac->mu);
206 277 : gpr_free(ac->addr_str);
207 277 : gpr_free(ac);
208 : }
209 11263 : grpc_exec_ctx_enqueue(exec_ctx, closure, *ep != NULL);
210 : }
211 :
212 11797 : void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
213 : grpc_endpoint **ep,
214 : grpc_pollset_set *interested_parties,
215 : const struct sockaddr *addr, size_t addr_len,
216 : gpr_timespec deadline) {
217 : int fd;
218 : grpc_dualstack_mode dsmode;
219 : int err;
220 : async_connect *ac;
221 : struct sockaddr_in6 addr6_v4mapped;
222 : struct sockaddr_in addr4_copy;
223 : grpc_fd *fdobj;
224 : char *name;
225 : char *addr_str;
226 :
227 11797 : *ep = NULL;
228 :
229 : /* Use dualstack sockets where available. */
230 11797 : if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
231 5137 : addr = (const struct sockaddr *)&addr6_v4mapped;
232 5137 : addr_len = sizeof(addr6_v4mapped);
233 : }
234 :
235 11797 : fd = grpc_create_dualstack_socket(addr, SOCK_STREAM, 0, &dsmode);
236 11797 : if (fd < 0) {
237 0 : gpr_log(GPR_ERROR, "Unable to create socket: %s", strerror(errno));
238 : }
239 11797 : if (dsmode == GRPC_DSMODE_IPV4) {
240 : /* If we got an AF_INET socket, map the address back to IPv4. */
241 106 : GPR_ASSERT(grpc_sockaddr_is_v4mapped(addr, &addr4_copy));
242 106 : addr = (struct sockaddr *)&addr4_copy;
243 106 : addr_len = sizeof(addr4_copy);
244 : }
245 11797 : if (!prepare_socket(addr, fd)) {
246 0 : grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
247 11797 : return;
248 : }
249 :
250 : do {
251 11797 : GPR_ASSERT(addr_len < ~(socklen_t)0);
252 11797 : err = connect(fd, addr, (socklen_t)addr_len);
253 11797 : } while (err < 0 && errno == EINTR);
254 :
255 11797 : addr_str = grpc_sockaddr_to_uri(addr);
256 11797 : gpr_asprintf(&name, "tcp-client:%s", addr_str);
257 :
258 11797 : fdobj = grpc_fd_create(fd, name);
259 :
260 11797 : if (err >= 0) {
261 456 : *ep = grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str);
262 456 : grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
263 456 : goto done;
264 : }
265 :
266 11341 : if (errno != EWOULDBLOCK && errno != EINPROGRESS) {
267 36 : gpr_log(GPR_ERROR, "connect error to '%s': %s", addr_str, strerror(errno));
268 36 : grpc_fd_orphan(exec_ctx, fdobj, NULL, NULL, "tcp_client_connect_error");
269 36 : grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
270 36 : goto done;
271 : }
272 :
273 11305 : grpc_pollset_set_add_fd(exec_ctx, interested_parties, fdobj);
274 :
275 11305 : ac = gpr_malloc(sizeof(async_connect));
276 11305 : ac->closure = closure;
277 11305 : ac->ep = ep;
278 11305 : ac->fd = fdobj;
279 11305 : ac->interested_parties = interested_parties;
280 11305 : ac->addr_str = addr_str;
281 11220 : addr_str = NULL;
282 11305 : gpr_mu_init(&ac->mu);
283 11305 : ac->refs = 2;
284 11305 : ac->write_closure.cb = on_writable;
285 11305 : ac->write_closure.cb_arg = ac;
286 :
287 11305 : if (grpc_tcp_trace) {
288 0 : gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: asynchronously connecting",
289 : ac->addr_str);
290 : }
291 :
292 11305 : gpr_mu_lock(&ac->mu);
293 11305 : grpc_timer_init(exec_ctx, &ac->alarm,
294 : gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC),
295 : tc_on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC));
296 11305 : grpc_fd_notify_on_write(exec_ctx, ac->fd, &ac->write_closure);
297 11305 : gpr_mu_unlock(&ac->mu);
298 :
299 : done:
300 11797 : gpr_free(name);
301 11797 : gpr_free(addr_str);
302 : }
303 :
304 : #endif
|
