Merge pull request #16261 from muxi/dynamic-cfstream-init

Use environment variable to enable CFStream

BUILD

@@ -1010,6 +1010,7 @@ grpc_cc_library(
         "src/core/lib/iomgr/cfstream_handle.cc",
         "src/core/lib/iomgr/endpoint_cfstream.cc",
         "src/core/lib/iomgr/error_cfstream.cc",
+        "src/core/lib/iomgr/iomgr_posix_cfstream.cc",
         "src/core/lib/iomgr/tcp_client_cfstream.cc",
     ],
     hdrs = [

build.yaml

@@ -548,6 +548,7 @@ filegroups:
   - src/core/lib/iomgr/cfstream_handle.cc
   - src/core/lib/iomgr/endpoint_cfstream.cc
   - src/core/lib/iomgr/error_cfstream.cc
+  - src/core/lib/iomgr/iomgr_posix_cfstream.cc
   - src/core/lib/iomgr/tcp_client_cfstream.cc
   uses:
   - grpc_base_headers

doc/environment_variables.md

@@ -135,3 +135,7 @@ some configuration as environment variables that can be set.
   if set, flow control will be effectively disabled. Max out all values and
   assume the remote peer does the same. Thus we can ignore any flow control
   bookkeeping, error checking, and decision making
+
+* grpc_cfstream
+  set to 1 to turn on CFStream experiment. With this experiment gRPC uses CFStream API to make TCP
+  connections. The option is only available on iOS platform and when macro GRPC_CFSTREAM is defined.

gRPC-Core.podspec

@@ -1118,6 +1118,7 @@ Pod::Spec.new do |s|
     ss.source_files = 'src/core/lib/iomgr/cfstream_handle.cc',
                       'src/core/lib/iomgr/endpoint_cfstream.cc',
                       'src/core/lib/iomgr/error_cfstream.cc',
+                      'src/core/lib/iomgr/iomgr_posix_cfstream.cc',
                       'src/core/lib/iomgr/tcp_client_cfstream.cc',
                       'src/core/lib/iomgr/cfstream_handle.h',
                       'src/core/lib/iomgr/endpoint_cfstream.h',

src/core/lib/iomgr/iomgr_posix_cfstream.cc

@@ -0,0 +1,75 @@
+/*
+ *
+ * Copyright 2015 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+#include <grpc/support/port_platform.h>
+
+#include "src/core/lib/iomgr/port.h"
+
+#ifdef GRPC_CFSTREAM_IOMGR
+
+#include "src/core/lib/debug/trace.h"
+#include "src/core/lib/iomgr/ev_posix.h"
+#include "src/core/lib/iomgr/iomgr_internal.h"
+#include "src/core/lib/iomgr/iomgr_posix.h"
+#include "src/core/lib/iomgr/resolve_address.h"
+#include "src/core/lib/iomgr/tcp_client.h"
+#include "src/core/lib/iomgr/tcp_posix.h"
+#include "src/core/lib/iomgr/tcp_server.h"
+#include "src/core/lib/iomgr/timer.h"
+
+static const char* grpc_cfstream_env_var = "grpc_cfstream";
+
+extern grpc_tcp_server_vtable grpc_posix_tcp_server_vtable;
+extern grpc_tcp_client_vtable grpc_posix_tcp_client_vtable;
+extern grpc_tcp_client_vtable grpc_cfstream_client_vtable;
+extern grpc_timer_vtable grpc_generic_timer_vtable;
+extern grpc_pollset_vtable grpc_posix_pollset_vtable;
+extern grpc_pollset_set_vtable grpc_posix_pollset_set_vtable;
+extern grpc_address_resolver_vtable grpc_posix_resolver_vtable;
+
+static void iomgr_platform_init(void) {
+  grpc_wakeup_fd_global_init();
+  grpc_event_engine_init();
+}
+
+static void iomgr_platform_flush(void) {}
+
+static void iomgr_platform_shutdown(void) {
+  grpc_event_engine_shutdown();
+  grpc_wakeup_fd_global_destroy();
+}
+
+static grpc_iomgr_platform_vtable vtable = {
+    iomgr_platform_init, iomgr_platform_flush, iomgr_platform_shutdown};
+
+void grpc_set_default_iomgr_platform() {
+  char* enable_cfstream = getenv(grpc_cfstream_env_var);
+  grpc_tcp_client_vtable* client_vtable = &grpc_posix_tcp_client_vtable;
+  if (enable_cfstream != nullptr && enable_cfstream[0] == '1') {
+    client_vtable = &grpc_cfstream_client_vtable;
+  }
+  grpc_set_tcp_client_impl(client_vtable);
+  grpc_set_tcp_server_impl(&grpc_posix_tcp_server_vtable);
+  grpc_set_timer_impl(&grpc_generic_timer_vtable);
+  grpc_set_pollset_vtable(&grpc_posix_pollset_vtable);
+  grpc_set_pollset_set_vtable(&grpc_posix_pollset_set_vtable);
+  grpc_set_resolver_impl(&grpc_posix_resolver_vtable);
+  grpc_set_iomgr_platform_vtable(&vtable);
+}
+
+#endif /* GRPC_CFSTREAM_IOMGR */

src/core/lib/iomgr/port.h

@@ -98,9 +98,9 @@
 #define GRPC_POSIX_FORK 1
 #define GRPC_POSIX_NO_SPECIAL_WAKEUP_FD 1
 #ifdef GRPC_CFSTREAM
-#define GRPC_POSIX_SOCKET_IOMGR 1
-#define GRPC_CFSTREAM_ENDPOINT 1
+#define GRPC_CFSTREAM_IOMGR 1
 #define GRPC_CFSTREAM_CLIENT 1
+#define GRPC_CFSTREAM_ENDPOINT 1
 #define GRPC_POSIX_SOCKET_ARES_EV_DRIVER 1
 #define GRPC_POSIX_SOCKET_EV 1
 #define GRPC_POSIX_SOCKET_EV_EPOLL1 1
@@ -111,6 +111,7 @@
 #define GRPC_POSIX_SOCKET_SOCKADDR 1
 #define GRPC_POSIX_SOCKET_SOCKET_FACTORY 1
 #define GRPC_POSIX_SOCKET_TCP 1
+#define GRPC_POSIX_SOCKET_TCP_CLIENT 1
 #define GRPC_POSIX_SOCKET_TCP_SERVER 1
 #define GRPC_POSIX_SOCKET_TCP_SERVER_UTILS_COMMON 1
 #define GRPC_POSIX_SOCKET_UTILS_COMMON 1

src/core/lib/iomgr/tcp_client_cfstream.cc

@@ -211,6 +211,6 @@ static void CFStreamClientConnect(grpc_closure* closure, grpc_endpoint** ep,
   gpr_mu_unlock(&connect->mu);
 }
 
-grpc_tcp_client_vtable grpc_posix_tcp_client_vtable = {CFStreamClientConnect};
+grpc_tcp_client_vtable grpc_cfstream_client_vtable = {CFStreamClientConnect};
 
 #endif /* GRPC_CFSTREAM_CLIENT */

src/objective-c/GRPCClient/GRPCCall.m

@@ -45,6 +45,8 @@ static NSMutableDictionary *callFlags;
 static NSString *const kAuthorizationHeader = @"authorization";
 static NSString *const kBearerPrefix = @"Bearer ";
 
+const char *kCFStreamVarName = "grpc_cfstream";
+
 @interface GRPCCall ()<GRXWriteable>
 // Make them read-write.
 @property(atomic, strong) NSDictionary *responseHeaders;
@@ -206,9 +208,12 @@ static NSString *const kBearerPrefix = @"Bearer ";
   } else {
     [_responseWriteable enqueueSuccessfulCompletion];
   }
-#ifndef GRPC_CFSTREAM
-  [GRPCConnectivityMonitor unregisterObserver:self];
-#endif
+
+  // Connectivity monitor is not required for CFStream
+  char *enableCFStream = getenv(kCFStreamVarName);
+  if (enableCFStream == nil || enableCFStream[0] != '1') {
+    [GRPCConnectivityMonitor unregisterObserver:self];
+  }
 
   // If the call isn't retained anywhere else, it can be deallocated now.
   _retainSelf = nil;
@@ -462,9 +467,11 @@ static NSString *const kBearerPrefix = @"Bearer ";
   [self sendHeaders:_requestHeaders];
   [self invokeCall];
 
-#ifndef GRPC_CFSTREAM
-  [GRPCConnectivityMonitor registerObserver:self selector:@selector(connectivityChanged:)];
-#endif
+  // Connectivity monitor is not required for CFStream
+  char *enableCFStream = getenv(kCFStreamVarName);
+  if (enableCFStream == nil || enableCFStream[0] != '1') {
+    [GRPCConnectivityMonitor registerObserver:self selector:@selector(connectivityChanged:)];
+  }
 }
 
 - (void)startWithWriteable:(id<GRXWriteable>)writeable {

src/objective-c/GRPCClient/private/GRPCCompletionQueue.m

@@ -20,13 +20,8 @@
 
 #import <grpc/grpc.h>
 
-#ifdef GRPC_CFSTREAM
-const grpc_completion_queue_attributes kCompletionQueueAttr = {GRPC_CQ_CURRENT_VERSION,
-                                                               GRPC_CQ_NEXT, GRPC_CQ_NON_POLLING};
-#else
 const grpc_completion_queue_attributes kCompletionQueueAttr = {
     GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING};
-#endif
 
 @implementation GRPCCompletionQueue
 

src/objective-c/GRPCClient/private/GRPCHost.m

@@ -34,6 +34,8 @@
 
 NS_ASSUME_NONNULL_BEGIN
 
+extern const char *kCFStreamVarName;
+
 static NSMutableDictionary *kHostCache;
 
 @implementation GRPCHost {
@@ -49,9 +51,11 @@ static NSMutableDictionary *kHostCache;
   if (_channelCreds != nil) {
     grpc_channel_credentials_release(_channelCreds);
   }
-#ifndef GRPC_CFSTREAM
-  [GRPCConnectivityMonitor unregisterObserver:self];
-#endif
+  // Connectivity monitor is not required for CFStream
+  char *enableCFStream = getenv(kCFStreamVarName);
+  if (enableCFStream == nil || enableCFStream[0] != '1') {
+    [GRPCConnectivityMonitor unregisterObserver:self];
+  }
 }
 
 // Default initializer.
@@ -87,9 +91,12 @@ static NSMutableDictionary *kHostCache;
       _compressAlgorithm = GRPC_COMPRESS_NONE;
       _retryEnabled = YES;
     }
-#ifndef GRPC_CFSTREAM
-    [GRPCConnectivityMonitor registerObserver:self selector:@selector(connectivityChange:)];
-#endif
+
+    // Connectivity monitor is not required for CFStream
+    char *enableCFStream = getenv(kCFStreamVarName);
+    if (enableCFStream == nil || enableCFStream[0] != '1') {
+      [GRPCConnectivityMonitor registerObserver:self selector:@selector(connectivityChange:)];
+    }
   }
   return self;
 }

src/objective-c/README-CFSTREAM.md

@@ -13,17 +13,17 @@ interface that gRPC uses when it is ready for production.
 ## Usage
 If you use gRPC following the instructions in
 [README.md](https://github.com/grpc/grpc/blob/master/src/objective-c/README.md):
-- Simply replace the
-dependency on `gRPC-ProtoRPC` with `gRPC-ProtoRPC/CFStream`. The build system will take care of
-everything else and switch networking to CFStream.
+- Replace the
+dependency on `gRPC-ProtoRPC` with `gRPC-ProtoRPC/CFStream`.
+- Enable CFStream with environment variable `grpc_cfstream=1`. This can be done either in Xcode
+  console or by your code with `setenv()` before gRPC is initialized.
 
 If your project directly depends on podspecs other than `gRPC-ProtoRPC` (e.g. `gRPC` or
 `gRPC-Core`):
 
-- Make your projects depend on subspecs corresponding to CFStream in each gRPC podspec. For
-  `gRPC-Core`, you will need to make sure that the completion queue you create is of type
-  `GRPC_CQ_NON_POLLING`. This is expected to be fixed soon so that you do not have to modify the
-  completion queue type.
+- Make your projects depend on subspecs corresponding to CFStream in each gRPC podspec.
+- Enable CFStream with environment variable `grpc_cfstream=1`. This can be done either in Xcode
+  console or by your code with `setenv()` before gRPC is initialized.
 
 ## Notes
 

src/objective-c/tests/InteropTests.m

@@ -36,6 +36,8 @@
 
 #define TEST_TIMEOUT 32
 
+extern const char *kCFStreamVarName;
+
 // Convenience constructors for the generated proto messages:
 
 @interface RMTStreamingOutputCallRequest (Constructors)
@@ -97,6 +99,9 @@ BOOL isRemoteInteropTest(NSString *host) {
   [Cronet start];
   [GRPCCall useCronetWithEngine:[Cronet getGlobalEngine]];
 #endif
+#ifdef GRPC_CFSTREAM
+  setenv(kCFStreamVarName, "1", 1);
+#endif
 }
 
 - (void)setUp {

src/objective-c/tests/Tests.xcodeproj/project.pbxproj

@@ -1982,6 +1982,16 @@
 				CODE_SIGN_IDENTITY = "iPhone Developer";
 				CODE_SIGN_STYLE = Automatic;
 				GCC_C_LANGUAGE_STANDARD = gnu11;
+				GCC_PREPROCESSOR_DEFINITIONS = (
+					"$(inherited)",
+					"COCOAPODS=1",
+					"$(inherited)",
+					"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
+					"$(inherited)",
+					"PB_FIELD_32BIT=1",
+					"PB_NO_PACKED_STRUCTS=1",
+					"GRPC_CFSTREAM=1",
+				);
 				INFOPLIST_FILE = Info.plist;
 				IPHONEOS_DEPLOYMENT_TARGET = 11.2;
 				LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
@@ -2100,6 +2110,16 @@
 				CODE_SIGN_IDENTITY = "iPhone Developer";
 				CODE_SIGN_STYLE = Automatic;
 				GCC_C_LANGUAGE_STANDARD = gnu11;
+				GCC_PREPROCESSOR_DEFINITIONS = (
+					"$(inherited)",
+					"COCOAPODS=1",
+					"$(inherited)",
+					"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
+					"$(inherited)",
+					"PB_FIELD_32BIT=1",
+					"PB_NO_PACKED_STRUCTS=1",
+					"GRPC_CFSTREAM=1",
+				);
 				INFOPLIST_FILE = Info.plist;
 				IPHONEOS_DEPLOYMENT_TARGET = 11.2;
 				LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";
@@ -2218,6 +2238,16 @@
 				CODE_SIGN_IDENTITY = "iPhone Developer";
 				CODE_SIGN_STYLE = Automatic;
 				GCC_C_LANGUAGE_STANDARD = gnu11;
+				GCC_PREPROCESSOR_DEFINITIONS = (
+					"$(inherited)",
+					"COCOAPODS=1",
+					"$(inherited)",
+					"GPB_USE_PROTOBUF_FRAMEWORK_IMPORTS=1",
+					"$(inherited)",
+					"PB_FIELD_32BIT=1",
+					"PB_NO_PACKED_STRUCTS=1",
+					"GRPC_CFSTREAM=1",
+				);
 				INFOPLIST_FILE = Info.plist;
 				IPHONEOS_DEPLOYMENT_TARGET = 11.2;
 				LD_RUNPATH_SEARCH_PATHS = "$(inherited) @executable_path/Frameworks @loader_path/Frameworks";

tools/run_tests/generated/sources_and_headers.json

@@ -9921,6 +9921,7 @@
       "src/core/lib/iomgr/endpoint_cfstream.h", 
       "src/core/lib/iomgr/error_cfstream.cc", 
       "src/core/lib/iomgr/error_cfstream.h", 
+      "src/core/lib/iomgr/iomgr_posix_cfstream.cc", 
       "src/core/lib/iomgr/tcp_client_cfstream.cc"
     ], 
     "third_party": false,