1
2 /*
3 *
4 * Copyright 2018 gRPC authors.
5 *
6 * Licensed under the Apache License, Version 2.0 (the "License");
7 * you may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
9 *
10 * http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
17 *
18 */
19
20 #include <grpc/support/port_platform.h>
21
22 #include "src/core/lib/iomgr/port.h"
23
24 #ifdef GRPC_CFSTREAM_CLIENT
25
26 #include <CoreFoundation/CoreFoundation.h>
27
28 #include <string.h>
29
30 #include <grpc/support/alloc.h>
31 #include <grpc/support/log.h>
32 #include <grpc/support/sync.h>
33
34 #include <netinet/in.h>
35
36 #include "src/core/lib/channel/channel_args.h"
37 #include "src/core/lib/gpr/host_port.h"
38 #include "src/core/lib/iomgr/cfstream_handle.h"
39 #include "src/core/lib/iomgr/closure.h"
40 #include "src/core/lib/iomgr/endpoint_cfstream.h"
41 #include "src/core/lib/iomgr/error.h"
42 #include "src/core/lib/iomgr/error_cfstream.h"
43 #include "src/core/lib/iomgr/sockaddr_utils.h"
44 #include "src/core/lib/iomgr/tcp_client.h"
45 #include "src/core/lib/iomgr/timer.h"
46
47 extern grpc_core::TraceFlag grpc_tcp_trace;
48
49 typedef struct CFStreamConnect {
50 gpr_mu mu;
51 gpr_refcount refcount;
52
53 CFReadStreamRef read_stream;
54 CFWriteStreamRef write_stream;
55 CFStreamHandle* stream_handle;
56
57 grpc_timer alarm;
58 grpc_closure on_alarm;
59 grpc_closure on_open;
60
61 bool read_stream_open;
62 bool write_stream_open;
63 bool failed;
64
65 grpc_closure* closure;
66 grpc_endpoint** endpoint;
67 int refs;
68 char* addr_name;
69 grpc_resource_quota* resource_quota;
70 } CFStreamConnect;
71
CFStreamConnectCleanup(CFStreamConnect * connect)72 static void CFStreamConnectCleanup(CFStreamConnect* connect) {
73 grpc_resource_quota_unref_internal(connect->resource_quota);
74 CFSTREAM_HANDLE_UNREF(connect->stream_handle, "async connect clean up");
75 CFRelease(connect->read_stream);
76 CFRelease(connect->write_stream);
77 gpr_mu_destroy(&connect->mu);
78 gpr_free(connect->addr_name);
79 gpr_free(connect);
80 }
81
OnAlarm(void * arg,grpc_error * error)82 static void OnAlarm(void* arg, grpc_error* error) {
83 CFStreamConnect* connect = static_cast<CFStreamConnect*>(arg);
84 if (grpc_tcp_trace.enabled()) {
85 gpr_log(GPR_DEBUG, "CLIENT_CONNECT :%p OnAlarm, error:%p", connect, error);
86 }
87 gpr_mu_lock(&connect->mu);
88 grpc_closure* closure = connect->closure;
89 connect->closure = nil;
90 const bool done = (--connect->refs == 0);
91 gpr_mu_unlock(&connect->mu);
92 // Only schedule a callback once, by either OnAlarm or OnOpen. The
93 // first one issues callback while the second one does cleanup.
94 if (done) {
95 CFStreamConnectCleanup(connect);
96 } else {
97 grpc_error* error =
98 GRPC_ERROR_CREATE_FROM_STATIC_STRING("connect() timed out");
99 GRPC_CLOSURE_SCHED(closure, error);
100 }
101 }
102
OnOpen(void * arg,grpc_error * error)103 static void OnOpen(void* arg, grpc_error* error) {
104 CFStreamConnect* connect = static_cast<CFStreamConnect*>(arg);
105 if (grpc_tcp_trace.enabled()) {
106 gpr_log(GPR_DEBUG, "CLIENT_CONNECT :%p OnOpen, error:%p", connect, error);
107 }
108 gpr_mu_lock(&connect->mu);
109 grpc_timer_cancel(&connect->alarm);
110 grpc_closure* closure = connect->closure;
111 connect->closure = nil;
112
113 bool done = (--connect->refs == 0);
114 grpc_endpoint** endpoint = connect->endpoint;
115
116 // Only schedule a callback once, by either OnAlarm or OnOpen. The
117 // first one issues callback while the second one does cleanup.
118 if (done) {
119 gpr_mu_unlock(&connect->mu);
120 CFStreamConnectCleanup(connect);
121 } else {
122 if (error == GRPC_ERROR_NONE) {
123 CFErrorRef stream_error = CFReadStreamCopyError(connect->read_stream);
124 if (stream_error == NULL) {
125 stream_error = CFWriteStreamCopyError(connect->write_stream);
126 }
127 if (stream_error) {
128 error = GRPC_ERROR_CREATE_FROM_CFERROR(stream_error, "connect() error");
129 CFRelease(stream_error);
130 }
131 if (error == GRPC_ERROR_NONE) {
132 *endpoint = grpc_cfstream_endpoint_create(
133 connect->read_stream, connect->write_stream, connect->addr_name,
134 connect->resource_quota, connect->stream_handle);
135 }
136 } else {
137 GRPC_ERROR_REF(error);
138 }
139 gpr_mu_unlock(&connect->mu);
140 GRPC_CLOSURE_SCHED(closure, error);
141 }
142 }
143
ParseResolvedAddress(const grpc_resolved_address * addr,CFStringRef * host,int * port)144 static void ParseResolvedAddress(const grpc_resolved_address* addr,
145 CFStringRef* host, int* port) {
146 char *host_port, *host_string, *port_string;
147 grpc_sockaddr_to_string(&host_port, addr, 1);
148 gpr_split_host_port(host_port, &host_string, &port_string);
149 *host = CFStringCreateWithCString(NULL, host_string, kCFStringEncodingUTF8);
150 gpr_free(host_string);
151 gpr_free(port_string);
152 gpr_free(host_port);
153 *port = grpc_sockaddr_get_port(addr);
154 }
155
CFStreamClientConnect(grpc_closure * closure,grpc_endpoint ** ep,grpc_pollset_set * interested_parties,const grpc_channel_args * channel_args,const grpc_resolved_address * resolved_addr,grpc_millis deadline)156 static void CFStreamClientConnect(grpc_closure* closure, grpc_endpoint** ep,
157 grpc_pollset_set* interested_parties,
158 const grpc_channel_args* channel_args,
159 const grpc_resolved_address* resolved_addr,
160 grpc_millis deadline) {
161 CFStreamConnect* connect;
162
163 connect = (CFStreamConnect*)gpr_zalloc(sizeof(CFStreamConnect));
164 connect->closure = closure;
165 connect->endpoint = ep;
166 connect->addr_name = grpc_sockaddr_to_uri(resolved_addr);
167 // connect->resource_quota = resource_quota;
168 connect->refs = 2; // One for the connect operation, one for the timer.
169 gpr_ref_init(&connect->refcount, 1);
170 gpr_mu_init(&connect->mu);
171
172 if (grpc_tcp_trace.enabled()) {
173 gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %p, %s: asynchronously connecting",
174 connect, connect->addr_name);
175 }
176
177 grpc_resource_quota* resource_quota = grpc_resource_quota_create(NULL);
178 if (channel_args != NULL) {
179 for (size_t i = 0; i < channel_args->num_args; i++) {
180 if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_RESOURCE_QUOTA)) {
181 grpc_resource_quota_unref_internal(resource_quota);
182 resource_quota = grpc_resource_quota_ref_internal(
183 (grpc_resource_quota*)channel_args->args[i].value.pointer.p);
184 }
185 }
186 }
187 connect->resource_quota = resource_quota;
188
189 CFReadStreamRef read_stream;
190 CFWriteStreamRef write_stream;
191
192 CFStringRef host;
193 int port;
194 ParseResolvedAddress(resolved_addr, &host, &port);
195 CFStreamCreatePairWithSocketToHost(NULL, host, port, &read_stream,
196 &write_stream);
197 CFRelease(host);
198 connect->read_stream = read_stream;
199 connect->write_stream = write_stream;
200 connect->stream_handle =
201 CFStreamHandle::CreateStreamHandle(read_stream, write_stream);
202 GRPC_CLOSURE_INIT(&connect->on_open, OnOpen, static_cast<void*>(connect),
203 grpc_schedule_on_exec_ctx);
204 connect->stream_handle->NotifyOnOpen(&connect->on_open);
205 GRPC_CLOSURE_INIT(&connect->on_alarm, OnAlarm, connect,
206 grpc_schedule_on_exec_ctx);
207 gpr_mu_lock(&connect->mu);
208 CFReadStreamOpen(read_stream);
209 CFWriteStreamOpen(write_stream);
210 grpc_timer_init(&connect->alarm, deadline, &connect->on_alarm);
211 gpr_mu_unlock(&connect->mu);
212 }
213
214 grpc_tcp_client_vtable grpc_cfstream_client_vtable = {CFStreamClientConnect};
215
216 #endif /* GRPC_CFSTREAM_CLIENT */
217