Optimize blocking intercepted unary-unary calls

Change the blocking unary-unary call code path to rely
on the underlying synchronous API, as opposed to calling
the Future-based underlying async API and invoking `.result()`
on the returned Future object immediately, which can be
resource-intensive.
pull/14639/head
Mehrdad Afshari 7 years ago
parent 30a0187aaa
commit 32919791c7
  1. 80
      src/python/grpcio/grpc/_interceptor.py

@ -134,6 +134,58 @@ class _LocalFailure(grpc.RpcError, grpc.Future, grpc.Call):
raise self._exception
class _UnaryOutcome(grpc.Call, grpc.Future):
def __init__(self, response, call):
self._response = response
self._call = call
def initial_metadata(self):
return self._call.initial_metadata()
def trailing_metadata(self):
return self._call.trailing_metadata()
def code(self):
return self._call.code()
def details(self):
return self._call.details()
def is_active(self):
return self._call.is_active()
def time_remaining(self):
return self._call.time_remaining()
def cancel(self):
return self._call.cancel()
def add_callback(self, callback):
return self._call.add_callback(callback)
def cancelled(self):
return False
def running(self):
return False
def done(self):
return True
def result(self, ignored_timeout=None):
return self._response
def exception(self, ignored_timeout=None):
return None
def traceback(self, ignored_timeout=None):
return None
def add_done_callback(self, fn):
fn(self)
class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable):
def __init__(self, thunk, method, interceptor):
@ -142,23 +194,35 @@ class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable):
self._interceptor = interceptor
def __call__(self, request, timeout=None, metadata=None, credentials=None):
call_future = self.future(
response, ignored_call = self.with_call(
request,
timeout=timeout,
metadata=metadata,
credentials=credentials)
return call_future.result()
return response
def with_call(self, request, timeout=None, metadata=None, credentials=None):
call_future = self.future(
request,
timeout=timeout,
metadata=metadata,
credentials=credentials)
client_call_details = _ClientCallDetails(self._method, timeout,
metadata, credentials)
def continuation(new_details, request):
new_method, new_timeout, new_metadata, new_credentials = (
_unwrap_client_call_details(new_details, client_call_details))
try:
response, call = self._thunk(new_method).with_call(
request,
timeout=new_timeout,
metadata=new_metadata,
credentials=new_credentials)
return _UnaryOutcome(response, call)
except Exception as exception: # pylint:disable=broad-except
return _LocalFailure(exception, sys.exc_info()[2])
call_future = self._interceptor.intercept_unary_unary(
continuation, client_call_details, request)
return call_future.result(), call_future
def future(self, request, timeout=None, metadata=None, credentials=None):
client_call_details = _ClientCallDetails(self._method, timeout,
metadata, credentials)

Loading…
Cancel
Save