Merge pull request #17578 from muxi/grpccall-safety

Make gRPC ObjC thread safety right

1 files changed, 14 insertions, 7 deletions

diff --git a/src/objective-c/RxLibrary/GRXBufferedPipe.m b/src/objective-c/RxLibrary/GRXBufferedPipe.m
index 546d46cba3..74e2f03da6 100644
--- a/src/objective-c/RxLibrary/GRXBufferedPipe.m
+++ b/src/objective-c/RxLibrary/GRXBufferedPipe.m
@@ -51,16 +51,22 @@
     // We need a copy, so that it doesn't mutate before it's written at the other end of the pipe.
     value = [value copy];
   }
-  __weak GRXBufferedPipe *weakSelf = self;
   dispatch_async(_writeQueue, ^(void) {
-    [weakSelf.writeable writeValue:value];
+    @synchronized(self) {
+      if (self->_state == GRXWriterStateFinished) {
+        return;
+      }
+      [self.writeable writeValue:value];
+    }
   });
 }
 
 - (void)writesFinishedWithError:(NSError *)errorOrNil {
-  __weak GRXBufferedPipe *weakSelf = self;
   dispatch_async(_writeQueue, ^{
-    [weakSelf finishWithError:errorOrNil];
+    if (self->_state == GRXWriterStateFinished) {
+      return;
+    }
+    [self finishWithError:errorOrNil];
   });
 }
 
@@ -100,14 +106,15 @@
 }
 
 - (void)startWithWriteable:(id<GRXWriteable>)writeable {
-  self.writeable = writeable;
-  _state = GRXWriterStateStarted;
+  @synchronized(self) {
+    self.writeable = writeable;
+    _state = GRXWriterStateStarted;
+  }
   dispatch_resume(_writeQueue);
 }
 
 - (void)finishWithError:(NSError *)errorOrNil {
   [self.writeable writesFinishedWithError:errorOrNil];
-  self.state = GRXWriterStateFinished;
 }
 
 - (void)dealloc {