legacy-libs/grpc-cloned/deps/grpc/src/core/lib/iomgr/tcp_client_cfstream.cc

   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/gprpp/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
  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
  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
 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
 144 static void ParseResolvedAddress(const grpc_resolved_address* addr,
 145                                  CFStringRef* host, int* port) {
 146   char* host_port;
 147   grpc_sockaddr_to_string(&host_port, addr, 1);
 148   grpc_core::UniquePtr<char> host_string;
 149   grpc_core::UniquePtr<char> port_string;
 150   grpc_core::SplitHostPort(host_port, &host_string, &port_string);
 151   *host =
 152       CFStringCreateWithCString(NULL, host_string.get(), kCFStringEncodingUTF8);
 153   gpr_free(host_port);
 154   *port = grpc_sockaddr_get_port(addr);
 155 }
 156
 157 static void CFStreamClientConnect(grpc_closure* closure, grpc_endpoint** ep,
 158                                   grpc_pollset_set* interested_parties,
 159                                   const grpc_channel_args* channel_args,
 160                                   const grpc_resolved_address* resolved_addr,
 161                                   grpc_millis deadline) {
 162   CFStreamConnect* connect;
 163
 164   connect = (CFStreamConnect*)gpr_zalloc(sizeof(CFStreamConnect));
 165   connect->closure = closure;
 166   connect->endpoint = ep;
 167   connect->addr_name = grpc_sockaddr_to_uri(resolved_addr);
 168   // connect->resource_quota = resource_quota;
 169   connect->refs = 2;  // One for the connect operation, one for the timer.
 170   gpr_ref_init(&connect->refcount, 1);
 171   gpr_mu_init(&connect->mu);
 172
 173   if (grpc_tcp_trace.enabled()) {
 174     gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %p, %s: asynchronously connecting",
 175             connect, connect->addr_name);
 176   }
 177
 178   grpc_resource_quota* resource_quota = grpc_resource_quota_create(NULL);
 179   if (channel_args != NULL) {
 180     for (size_t i = 0; i < channel_args->num_args; i++) {
 181       if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_RESOURCE_QUOTA)) {
 182         grpc_resource_quota_unref_internal(resource_quota);
 183         resource_quota = grpc_resource_quota_ref_internal(
 184             (grpc_resource_quota*)channel_args->args[i].value.pointer.p);
 185       }
 186     }
 187   }
 188   connect->resource_quota = resource_quota;
 189
 190   CFReadStreamRef read_stream;
 191   CFWriteStreamRef write_stream;
 192
 193   CFStringRef host;
 194   int port;
 195   ParseResolvedAddress(resolved_addr, &host, &port);
 196   CFStreamCreatePairWithSocketToHost(NULL, host, port, &read_stream,
 197                                      &write_stream);
 198   CFRelease(host);
 199   connect->read_stream = read_stream;
 200   connect->write_stream = write_stream;
 201   connect->stream_handle =
 202       CFStreamHandle::CreateStreamHandle(read_stream, write_stream);
 203   GRPC_CLOSURE_INIT(&connect->on_open, OnOpen, static_cast<void*>(connect),
 204                     grpc_schedule_on_exec_ctx);
 205   connect->stream_handle->NotifyOnOpen(&connect->on_open);
 206   GRPC_CLOSURE_INIT(&connect->on_alarm, OnAlarm, connect,
 207                     grpc_schedule_on_exec_ctx);
 208   gpr_mu_lock(&connect->mu);
 209   CFReadStreamOpen(read_stream);
 210   CFWriteStreamOpen(write_stream);
 211   grpc_timer_init(&connect->alarm, deadline, &connect->on_alarm);
 212   gpr_mu_unlock(&connect->mu);
 213 }
 214
 215 grpc_tcp_client_vtable grpc_cfstream_client_vtable = {CFStreamClientConnect};
 216
 217 #endif /* GRPC_CFSTREAM_CLIENT */