Merge pull request #5724 from grpc/revert-5607-revert-5544-master
Redo "Pass a non-infinite deadline to grpc_completion_queue_next() to prevent queues from blocking indefinitely in poll()"
diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h
index 7b66cd4..a52095d 100644
--- a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h
+++ b/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 @@
*/
@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
diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
index ff30316..250bbf0 100644
--- a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
+++ b/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];
}
};
});