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

Use environment variable to enable CFStream

14 files changed, 155 insertions, 27 deletions

diff --git a/BUILD b/BUILD
index 35cf86288d..c1b573f9cf 100644
--- a/BUILD
+++ b/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 = [
diff --git a/build.yaml b/build.yaml
index bd9c4237a1..9827d0956d 100644
--- a/build.yaml
+++ b/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
diff --git a/doc/environment_variables.md b/doc/environment_variables.md
index bf2de927a4..b472f7e126 100644
--- a/doc/environment_variables.md
+++ b/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.
diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec
index 1998bc8b4c..f828185752 100644
--- a/gRPC-Core.podspec
+++ b/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',
diff --git a/src/core/lib/iomgr/iomgr_posix_cfstream.cc b/src/core/lib/iomgr/iomgr_posix_cfstream.cc
new file mode 100644
index 0000000000..235a9e0712
--- /dev/null
+++ b/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 */
diff --git a/src/core/lib/iomgr/port.h b/src/core/lib/iomgr/port.h
index 80d8e63cdd..1d0ecff802 100644
--- a/src/core/lib/iomgr/port.h
+++ b/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
diff --git a/src/core/lib/iomgr/tcp_client_cfstream.cc b/src/core/lib/iomgr/tcp_client_cfstream.cc
index 5acea91792..4b21322d74 100644
--- a/src/core/lib/iomgr/tcp_client_cfstream.cc
+++ b/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 */
diff --git a/src/objective-c/GRPCClient/GRPCCall.m b/src/objective-c/GRPCClient/GRPCCall.m
index 8ce88c7db2..084fbdeb49 100644
--- a/src/objective-c/GRPCClient/GRPCCall.m
+++ b/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 {
diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
index bda1c3360b..f454a6dc57 100644
--- a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
+++ b/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
 
diff --git a/src/objective-c/GRPCClient/private/GRPCHost.m b/src/objective-c/GRPCClient/private/GRPCHost.m
index 2e9f9f243b..862909f238 100644
--- a/src/objective-c/GRPCClient/private/GRPCHost.m
+++ b/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;
 }
diff --git a/src/objective-c/README-CFSTREAM.md b/src/objective-c/README-CFSTREAM.md
index 0b5215a7b3..0cb25ab237 100644
--- a/src/objective-c/README-CFSTREAM.md
+++ b/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
 
diff --git a/src/objective-c/tests/InteropTests.m b/src/objective-c/tests/InteropTests.m
index 1e1da2dd66..5750dccd89 100644
--- a/src/objective-c/tests/InteropTests.m
+++ b/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 {
diff --git a/src/objective-c/tests/Tests.xcodeproj/project.pbxproj b/src/objective-c/tests/Tests.xcodeproj/project.pbxproj
index 8ff4633582..ea1066219d 100644
--- a/src/objective-c/tests/Tests.xcodeproj/project.pbxproj
+++ b/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";
diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json
index 34e23f09c2..a9327bded1 100644
--- a/tools/run_tests/generated/sources_and_headers.json
+++ b/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,