Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Polish threading + something else

Browse Source
pull/16190/head
Muxi Yan 6 years ago
parent ffeb0e6823
commit 512c01bc57
4 changed files with 209 additions and 157 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 241
      src/objective-c/GRPCClient/GRPCCall.m
  2. 1
      src/objective-c/GRPCClient/private/GRPCChannel.m
  3. 1
      src/objective-c/ProtoRPC/ProtoRPC.h
  4. 123
      src/objective-c/ProtoRPC/ProtoRPC.m

241
src/objective-c/GRPCClient/GRPCCall.m
Unescape View File

@ -155,7 +155,7 @@ const char *kCFStreamVarName = "grpc_cfstream";
// Fallback on earlier versions
_dispatchQueue = dispatch_queue_create(NULL, DISPATCH_QUEUE_SERIAL);
}
dispatch_set_target_queue(responseHandler.dispatchQueue, _dispatchQueue);
dispatch_set_target_queue(_dispatchQueue ,responseHandler.dispatchQueue);
_started = NO;
_canceled = NO;
_finished = NO;
@ -171,161 +171,194 @@ const char *kCFStreamVarName = "grpc_cfstream";
}
- (void)start {
dispatch_async(_dispatchQueue, ^{
NSAssert(!self->_started, @"Call already started.");
NSAssert(!self->_canceled, @"Call already canceled.");
if (self->_started) {
GRPCCall *call = nil;
@synchronized (self) {
NSAssert(!_started, @"Call already started.");
NSAssert(!_canceled, @"Call already canceled.");
if (_started) {
return;
}
if (self->_canceled) {
if (_canceled) {
return;
}
self->_started = YES;
if (!self->_callOptions) {
self->_callOptions = [[GRPCCallOptions alloc] init];
_started = YES;
if (!_callOptions) {
_callOptions = [[GRPCCallOptions alloc] init];
}
self->_call = [[GRPCCall alloc] initWithHost:self->_requestOptions.host
path:self->_requestOptions.path
callSafety:self->_requestOptions.safety
requestsWriter:self->_pipe
callOptions:self->_callOptions];
if (self->_callOptions.initialMetadata) {
[self->_call.requestHeaders addEntriesFromDictionary:self->_callOptions.initialMetadata];
_call = [[GRPCCall alloc] initWithHost:_requestOptions.host
path:_requestOptions.path
callSafety:_requestOptions.safety
requestsWriter:_pipe
callOptions:_callOptions];
if (_callOptions.initialMetadata) {
[_call.requestHeaders addEntriesFromDictionary:_callOptions.initialMetadata];
}
call = _call;
}
void (^valueHandler)(id value) = ^(id value) {
dispatch_async(self->_dispatchQueue, ^{
if (self->_handler) {
if (!self->_initialMetadataPublished) {
self->_initialMetadataPublished = YES;
[self issueInitialMetadata:self->_call.responseHeaders];
}
if (value) {
[self issueMessage:value];
}
void (^valueHandler)(id value) = ^(id value) {
@synchronized (self) {
if (self->_handler) {
if (!self->_initialMetadataPublished) {
self->_initialMetadataPublished = YES;
[self issueInitialMetadata:self->_call.responseHeaders];
}
});
};
void (^completionHandler)(NSError *errorOrNil) = ^(NSError *errorOrNil) {
dispatch_async(self->_dispatchQueue, ^{
if (self->_handler) {
if (!self->_initialMetadataPublished) {
self->_initialMetadataPublished = YES;
[self issueInitialMetadata:self->_call.responseHeaders];
}
[self issueClosedWithTrailingMetadata:self->_call.responseTrailers error:errorOrNil];
// Clean up _handler so that no more responses are reported to the handler.
self->_handler = nil;
if (value) {
[self issueMessage:value];
}
// Clearing _call must happen *after* dispatching close in order to get trailing
// metadata from _call.
if (self->_call) {
// Clean up the request writers. This should have no effect to _call since its
// response writeable is already nullified.
[self->_pipe writesFinishedWithError:nil];
self->_call = nil;
self->_pipe = nil;
}
}
};
void (^completionHandler)(NSError *errorOrNil) = ^(NSError *errorOrNil) {
@synchronized(self) {
if (self->_handler) {
if (!self->_initialMetadataPublished) {
self->_initialMetadataPublished = YES;
[self issueInitialMetadata:self->_call.responseHeaders];
}
});
};
id<GRXWriteable> responseWriteable =
[[GRXWriteable alloc] initWithValueHandler:valueHandler
completionHandler:completionHandler];
[self->_call startWithWriteable:responseWriteable];
});
[self issueClosedWithTrailingMetadata:self->_call.responseTrailers error:errorOrNil];
}
// Clearing _call must happen *after* dispatching close in order to get trailing
// metadata from _call.
if (self->_call) {
// Clean up the request writers. This should have no effect to _call since its
// response writeable is already nullified.
[self->_pipe writesFinishedWithError:nil];
self->_call = nil;
self->_pipe = nil;
}
}
};
id<GRXWriteable> responseWriteable =
[[GRXWriteable alloc] initWithValueHandler:valueHandler
completionHandler:completionHandler];
[call startWithWriteable:responseWriteable];
}
- (void)cancel {
dispatch_async(_dispatchQueue, ^{
NSAssert(!self->_canceled, @"Call already canceled.");
if (self->_canceled) {
GRPCCall *call = nil;
@synchronized (self) {
if (_canceled) {
return;
}
self->_canceled = YES;
if (self->_call) {
[self->_call cancel];
self->_call = nil;
self->_pipe = nil;
}
if (self->_handler) {
id<GRPCResponseHandler> handler = self->_handler;
dispatch_async(handler.dispatchQueue, ^{
if ([handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
[handler closedWithTrailingMetadata:nil
error:[NSError errorWithDomain:kGRPCErrorDomain
code:GRPCErrorCodeCancelled
userInfo:@{
NSLocalizedDescriptionKey :
@"Canceled by app"
}]];
_canceled = YES;
call = _call;
_call = nil;
_pipe = nil;
if ([_handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
dispatch_async(_dispatchQueue, ^{
// Copy to local so that block is freed after cancellation completes.
id<GRPCResponseHandler> copiedHandler = nil;
@synchronized (self) {
copiedHandler = self->_handler;
self->_handler = nil;
}
});
// Clean up _handler so that no more responses are reported to the handler.
self->_handler = nil;
[copiedHandler closedWithTrailingMetadata:nil
error:[NSError errorWithDomain:kGRPCErrorDomain
code:GRPCErrorCodeCancelled
userInfo:@{
NSLocalizedDescriptionKey :
@"Canceled by app"
}]];
});
}
});
}
[call cancel];
}
- (void)writeData:(NSData *)data {
dispatch_async(_dispatchQueue, ^{
NSAssert(!self->_canceled, @"Call arleady canceled.");
NSAssert(!self->_finished, @"Call is half-closed before sending data.");
if (self->_canceled) {
GRXBufferedPipe *pipe = nil;
@synchronized(self) {
NSAssert(!_canceled, @"Call arleady canceled.");
NSAssert(!_finished, @"Call is half-closed before sending data.");
if (_canceled) {
return;
}
if (self->_finished) {
if (_finished) {
return;
}
if (self->_pipe) {
[self->_pipe writeValue:data];
if (_pipe) {
pipe = _pipe;
}
});
}
[pipe writeValue:data];
}
- (void)finish {
dispatch_async(_dispatchQueue, ^{
NSAssert(self->_started, @"Call not started.");
NSAssert(!self->_canceled, @"Call arleady canceled.");
NSAssert(!self->_finished, @"Call already half-closed.");
if (!self->_started) {
GRXBufferedPipe *pipe = nil;
@synchronized(self) {
NSAssert(_started, @"Call not started.");
NSAssert(!_canceled, @"Call arleady canceled.");
NSAssert(!_finished, @"Call already half-closed.");
if (!_started) {
return;
}
if (self->_canceled) {
if (_canceled) {
return;
}
if (self->_finished) {
if (_finished) {
return;
}
if (self->_pipe) {
[self->_pipe writesFinishedWithError:nil];
if (_pipe) {
pipe = _pipe;
_pipe = nil;
}
self->_pipe = nil;
self->_finished = YES;
});
_finished = YES;
}
[pipe writesFinishedWithError:nil];
}
- (void)issueInitialMetadata:(NSDictionary *)initialMetadata {
if (initialMetadata != nil && [_handler respondsToSelector:@selector(receivedInitialMetadata:)]) {
[_handler receivedInitialMetadata:initialMetadata];
@synchronized (self) {
if (initialMetadata != nil && [_handler respondsToSelector:@selector(receivedInitialMetadata:)]) {
dispatch_async(_dispatchQueue, ^{
id<GRPCResponseHandler> handler = nil;
@synchronized (self) {
handler = self->_handler;
}
[handler receivedInitialMetadata:initialMetadata];
});
}
}
}
- (void)issueMessage:(id)message {
if (message != nil && [_handler respondsToSelector:@selector(receivedRawMessage:)]) {
[_handler receivedRawMessage:message];
@synchronized (self) {
if (message != nil && [_handler respondsToSelector:@selector(receivedRawMessage:)]) {
dispatch_async(_dispatchQueue, ^{
id<GRPCResponseHandler> handler = nil;
@synchronized (self) {
handler = self->_handler;
}
[handler receivedRawMessage:message];
});
}
}
}
- (void)issueClosedWithTrailingMetadata:(NSDictionary *)trailingMetadata error:(NSError *)error {
if ([_handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
[_handler closedWithTrailingMetadata:trailingMetadata error:error];
@synchronized (self) {
if ([_handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
dispatch_async(_dispatchQueue, ^{
id<GRPCResponseHandler> handler = nil;
@synchronized (self) {
handler = self->_handler;
// Clean up _handler so that no more responses are reported to the handler.
self->_handler = nil;
}
[handler closedWithTrailingMetadata:trailingMetadata
error:error];
});
}
}
}

1
src/objective-c/GRPCClient/private/GRPCChannel.m
Unescape View File

@ -302,7 +302,6 @@ static const NSTimeInterval kDefaultChannelDestroyDelay = 30;
}
- (void)unref {
NSLog(@"unref");
dispatch_async(_dispatchQueue, ^{
NSAssert(self->_refcount > 0, @"Illegal reference count.");
if (self->_refcount == 0) {

1
src/objective-c/ProtoRPC/ProtoRPC.h
Unescape View File

@ -32,6 +32,7 @@ NS_ASSUME_NONNULL_BEGIN
/**
* Issued when initial metadata is received from the server.
*/
- (void)receivedInitialMetadata:(nullable NSDictionary *)initialMetadata;
/**

123
src/objective-c/ProtoRPC/ProtoRPC.m
Unescape View File

@ -112,7 +112,7 @@ static NSError *ErrorForBadProto(id proto, Class expectedClass, NSError *parsing
} else {
_dispatchQueue = dispatch_queue_create(nil, DISPATCH_QUEUE_SERIAL);
}
dispatch_set_target_queue(handler.dispatchQueue, _dispatchQueue);
dispatch_set_target_queue(_dispatchQueue, handler.dispatchQueue);
[self start];
}
@ -127,15 +127,17 @@ static NSError *ErrorForBadProto(id proto, Class expectedClass, NSError *parsing
}
- (void)cancel {
dispatch_async(_dispatchQueue, ^{
if (_call) {
[_call cancel];
_call = nil;
}
if (_handler) {
id<GRPCProtoResponseHandler> handler = _handler;
if ([handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
dispatch_async(handler.dispatchQueue, ^{
GRPCCall2 *call;
@synchronized(self) {
call = _call;
_call = nil;
if ([_handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
dispatch_async(_handler.dispatchQueue, ^{
id<GRPCProtoResponseHandler> handler = nil;
@synchronized(self) {
handler = self->_handler;
self->_handler = nil;
}
[handler closedWithTrailingMetadata:nil
error:[NSError errorWithDomain:kGRPCErrorDomain
code:GRPCErrorCodeCancelled
@ -145,9 +147,8 @@ static NSError *ErrorForBadProto(id proto, Class expectedClass, NSError *parsing
}]];
});
}
_handler = nil;
}
});
}
[call cancel];
}
- (void)writeMessage:(GPBMessage *)message {
@ -155,63 +156,81 @@ static NSError *ErrorForBadProto(id proto, Class expectedClass, NSError *parsing
[NSException raise:NSInvalidArgumentException format:@"Data must be a valid protobuf type."];
}
dispatch_async(_dispatchQueue, ^{
if (_call) {
[_call writeData:[message data]];
}
});
GRPCCall2 *call;
@synchronized(self) {
call = _call;
}
[call writeData:[message data]];
}
- (void)finish {
dispatch_async(_dispatchQueue, ^{
if (_call) {
[_call finish];
_call = nil;
}
});
GRPCCall2 *call;
@synchronized(self) {
call = _call;
_call = nil;
}
[call finish];
}
- (void)receivedInitialMetadata:(NSDictionary *)initialMetadata {
dispatch_async(_dispatchQueue, ^{
if (initialMetadata != nil && [self->_handler respondsToSelector:@selector(initialMetadata:)]) {
[self->_handler receivedInitialMetadata:initialMetadata];
@synchronized (self) {
if (initialMetadata != nil && [_handler respondsToSelector:@selector(initialMetadata:)]) {
dispatch_async(_dispatchQueue, ^{
id<GRPCProtoResponseHandler> handler = nil;
@synchronized (self) {
handler = self->_handler;
}
[handler receivedInitialMetadata:initialMetadata];
});
}
});
}
}
- (void)receivedRawMessage:(NSData *)message {
dispatch_async(_dispatchQueue, ^{
if (self->_handler && message != nil) {
NSError *error = nil;
GPBMessage *parsed = [self->_responseClass parseFromData:message error:&error];
if (parsed) {
if ([self->_handler respondsToSelector:@selector(receivedProtoMessage:)]) {
[self->_handler receivedProtoMessage:parsed];
if (message == nil) return;
NSError *error = nil;
GPBMessage *parsed = [_responseClass parseFromData:message error:&error];
@synchronized (self) {
if (parsed && [_handler respondsToSelector:@selector(receivedProtoMessage:)]) {
dispatch_async(_dispatchQueue, ^{
id<GRPCProtoResponseHandler> handler = nil;
@synchronized (self) {
handler = self->_handler;
}
} else {
if ([self->_handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
[self->_handler
closedWithTrailingMetadata:nil
error:ErrorForBadProto(message, _responseClass, error)];
[handler receivedProtoMessage:parsed];
});
} else if (!parsed && [_handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]){
dispatch_async(_dispatchQueue, ^{
id<GRPCProtoResponseHandler> handler = nil;
@synchronized (self) {
handler = self->_handler;
self->_handler = nil;
}
self->_handler = nil;
[self->_call cancel];
self->_call = nil;
}
[handler closedWithTrailingMetadata:nil
error:ErrorForBadProto(message, _responseClass, error)];
});
[_call cancel];
_call = nil;
}
});
}
}
- (void)closedWithTrailingMetadata:(NSDictionary *)trailingMetadata
error:(NSError *)error {
dispatch_async(_dispatchQueue, ^{
if ([self->_handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
[self->_handler closedWithTrailingMetadata:trailingMetadata error:error];
@synchronized (self) {
if ([_handler respondsToSelector:@selector(closedWithTrailingMetadata:error:)]) {
dispatch_async(_dispatchQueue, ^{
id<GRPCProtoResponseHandler> handler = nil;
@synchronized (self) {
handler = self->_handler;
self->_handler = nil;
}
[handler closedWithTrailingMetadata:trailingMetadata error:error];
});
}
self->_handler = nil;
[self->_call cancel];
self->_call = nil;
});
_call = nil;
}
}
- (dispatch_queue_t)dispatchQueue {

Write Preview
Loading…
Cancel
Save