Merge pull request #5544 from gresrun/master

Pass a non-infinite deadline to grpc_completion_queue_next() to prevent queues from blocking indefinitely in poll()

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

@@ -36,6 +36,8 @@
 
 typedef void(^GRPCQueueCompletionHandler)(bool success);
 
+extern const int64_t kGRPCCompletionQueueDefaultTimeoutSecs;
+
 /**
  * This class lets one more easily use |grpc_completion_queue|. To use it, pass the value of the
  * |unmanagedQueue| property of an instance of this class to |grpc_channel_create_call|. Then for
@@ -49,6 +51,11 @@ typedef void(^GRPCQueueCompletionHandler)(bool success);
  */
 @interface GRPCCompletionQueue : NSObject
 @property(nonatomic, readonly) grpc_completion_queue *unmanagedQueue;
+@property(nonatomic, readonly) int64_t timeoutSecs;
 
 + (instancetype)completionQueue;
+
+- (instancetype)init;
+- (instancetype)initWithTimeout:(int64_t)timeoutSecs NS_DESIGNATED_INITIALIZER;
+
 @end

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

@@ -35,6 +35,9 @@
 
 #import <grpc/grpc.h>
 
+
+const int64_t kGRPCCompletionQueueDefaultTimeoutSecs = 60;
+
 @implementation GRPCCompletionQueue
 
 + (instancetype)completionQueue {
@@ -42,8 +45,13 @@
 }
 
 - (instancetype)init {
+  return [self initWithTimeout:kGRPCCompletionQueueDefaultTimeoutSecs];
+}
+
+- (instancetype)initWithTimeout:(int64_t)timeoutSecs {
   if ((self = [super init])) {
     _unmanagedQueue = grpc_completion_queue_create(NULL);
+    _timeoutSecs = timeoutSecs;
 
     // This is for the following block to capture the pointer by value (instead
     // of retaining self and doing self->_unmanagedQueue). This is essential
@@ -61,22 +69,28 @@
       gDefaultConcurrentQueue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0);
     });
     dispatch_async(gDefaultConcurrentQueue, ^{
+      // Using a non-infinite deadline to re-enter grpc_completion_queue_next()
+      // alleviates https://github.com/grpc/grpc/issues/5593
+      gpr_timespec deadline = (timeoutSecs < 0)
+          ? gpr_inf_future(GPR_CLOCK_REALTIME)
+          : gpr_time_from_seconds(timeoutSecs, GPR_CLOCK_REALTIME);
       while (YES) {
-        // The following call blocks until an event is available.
-        grpc_event event = grpc_completion_queue_next(unmanagedQueue,
-                                                      gpr_inf_future(GPR_CLOCK_REALTIME),
-                                                      NULL);
+        // The following call blocks until an event is available or the deadline elapses.
+        grpc_event event = grpc_completion_queue_next(unmanagedQueue, deadline, NULL);
         GRPCQueueCompletionHandler handler;
         switch (event.type) {
           case GRPC_OP_COMPLETE:
             handler = (__bridge_transfer GRPCQueueCompletionHandler)event.tag;
             handler(event.success);
             break;
+          case GRPC_QUEUE_TIMEOUT:
+            // Nothing to do here
+            break;
           case GRPC_QUEUE_SHUTDOWN:
             grpc_completion_queue_destroy(unmanagedQueue);
             return;
           default:
-            [NSException raise:@"Unrecognized completion type" format:@""];
+            [NSException raise:@"Unrecognized completion type" format:@"type=%d", event.type];
         }
       };
     });