From 4f91630d6b2ba9fbf0b75e2dbaa1932a739f2b17 Mon Sep 17 00:00:00 2001
From: Muxi Yan <mxyan@google.com>
Date: Fri, 14 Dec 2018 13:46:53 -0800
Subject: CFStream use serial dispatch queue

---
 src/core/lib/iomgr/cfstream_handle.cc | 98 ++++++++++++++++-------------------
 1 file changed, 44 insertions(+), 54 deletions(-)

(limited to 'src/core/lib/iomgr/cfstream_handle.cc')

diff --git a/src/core/lib/iomgr/cfstream_handle.cc b/src/core/lib/iomgr/cfstream_handle.cc
index 827fd24831..bb402f96cf 100644
--- a/src/core/lib/iomgr/cfstream_handle.cc
+++ b/src/core/lib/iomgr/cfstream_handle.cc
@@ -52,62 +52,53 @@ CFStreamHandle* CFStreamHandle::CreateStreamHandle(
 void CFStreamHandle::ReadCallback(CFReadStreamRef stream,
                                   CFStreamEventType type,
                                   void* client_callback_info) {
+  grpc_core::ExecCtx exec_ctx;
   CFStreamHandle* handle = static_cast<CFStreamHandle*>(client_callback_info);
-  CFSTREAM_HANDLE_REF(handle, "read callback");
-  dispatch_async(
-      dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
-        grpc_core::ExecCtx exec_ctx;
-        if (grpc_tcp_trace.enabled()) {
-          gpr_log(GPR_DEBUG, "CFStream ReadCallback (%p, %p, %lu, %p)", handle,
-                  stream, type, client_callback_info);
-        }
-        switch (type) {
-          case kCFStreamEventOpenCompleted:
-            handle->open_event_.SetReady();
-            break;
-          case kCFStreamEventHasBytesAvailable:
-          case kCFStreamEventEndEncountered:
-            handle->read_event_.SetReady();
-            break;
-          case kCFStreamEventErrorOccurred:
-            handle->open_event_.SetReady();
-            handle->read_event_.SetReady();
-            break;
-          default:
-            GPR_UNREACHABLE_CODE(return );
-        }
-        CFSTREAM_HANDLE_UNREF(handle, "read callback");
-      });
+  if (grpc_tcp_trace.enabled()) {
+    gpr_log(GPR_DEBUG, "CFStream ReadCallback (%p, %p, %lu, %p)", handle,
+            stream, type, client_callback_info);
+  }
+  switch (type) {
+    case kCFStreamEventOpenCompleted:
+      handle->open_event_.SetReady();
+      break;
+    case kCFStreamEventHasBytesAvailable:
+    case kCFStreamEventEndEncountered:
+      handle->read_event_.SetReady();
+      break;
+    case kCFStreamEventErrorOccurred:
+      handle->open_event_.SetReady();
+      handle->read_event_.SetReady();
+      break;
+    default:
+      GPR_UNREACHABLE_CODE(return );
+  }
 }
 void CFStreamHandle::WriteCallback(CFWriteStreamRef stream,
                                    CFStreamEventType type,
                                    void* clientCallBackInfo) {
+  grpc_core::ExecCtx exec_ctx;
   CFStreamHandle* handle = static_cast<CFStreamHandle*>(clientCallBackInfo);
-  CFSTREAM_HANDLE_REF(handle, "write callback");
-  dispatch_async(
-      dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), ^{
-        grpc_core::ExecCtx exec_ctx;
-        if (grpc_tcp_trace.enabled()) {
-          gpr_log(GPR_DEBUG, "CFStream WriteCallback (%p, %p, %lu, %p)", handle,
-                  stream, type, clientCallBackInfo);
-        }
-        switch (type) {
-          case kCFStreamEventOpenCompleted:
-            handle->open_event_.SetReady();
-            break;
-          case kCFStreamEventCanAcceptBytes:
-          case kCFStreamEventEndEncountered:
-            handle->write_event_.SetReady();
-            break;
-          case kCFStreamEventErrorOccurred:
-            handle->open_event_.SetReady();
-            handle->write_event_.SetReady();
-            break;
-          default:
-            GPR_UNREACHABLE_CODE(return );
-        }
-        CFSTREAM_HANDLE_UNREF(handle, "write callback");
-      });
+  printf("** CFStreamHandle::WriteCallback\n");
+  if (grpc_tcp_trace.enabled()) {
+    gpr_log(GPR_DEBUG, "CFStream WriteCallback (%p, %p, %lu, %p)", handle,
+            stream, type, clientCallBackInfo);
+  }
+  switch (type) {
+    case kCFStreamEventOpenCompleted:
+      handle->open_event_.SetReady();
+      break;
+    case kCFStreamEventCanAcceptBytes:
+    case kCFStreamEventEndEncountered:
+      handle->write_event_.SetReady();
+      break;
+    case kCFStreamEventErrorOccurred:
+      handle->open_event_.SetReady();
+      handle->write_event_.SetReady();
+      break;
+    default:
+      GPR_UNREACHABLE_CODE(return );
+  }
 }
 
 CFStreamHandle::CFStreamHandle(CFReadStreamRef read_stream,
@@ -116,6 +107,7 @@ CFStreamHandle::CFStreamHandle(CFReadStreamRef read_stream,
   open_event_.InitEvent();
   read_event_.InitEvent();
   write_event_.InitEvent();
+  dispatch_queue_ = dispatch_queue_create(nullptr, DISPATCH_QUEUE_SERIAL);
   CFStreamClientContext ctx = {0, static_cast<void*>(this),
                                CFStreamHandle::Retain, CFStreamHandle::Release,
                                nil};
@@ -129,10 +121,8 @@ CFStreamHandle::CFStreamHandle(CFReadStreamRef read_stream,
       kCFStreamEventOpenCompleted | kCFStreamEventCanAcceptBytes |
           kCFStreamEventErrorOccurred | kCFStreamEventEndEncountered,
       CFStreamHandle::WriteCallback, &ctx);
-  CFReadStreamScheduleWithRunLoop(read_stream, CFRunLoopGetMain(),
-                                  kCFRunLoopCommonModes);
-  CFWriteStreamScheduleWithRunLoop(write_stream, CFRunLoopGetMain(),
-                                   kCFRunLoopCommonModes);
+  CFReadStreamSetDispatchQueue(read_stream, dispatch_queue_);
+  CFWriteStreamSetDispatchQueue(write_stream, dispatch_queue_);
 }
 
 CFStreamHandle::~CFStreamHandle() {
-- 
cgit v1.2.3