diff options
Diffstat (limited to 'src')
18 files changed, 366 insertions, 229 deletions
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/call.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/call.pyx.pxi index 6361669757..0892215b6d 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/call.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/call.pyx.pxi @@ -26,16 +26,13 @@ cdef class Call: def _start_batch(self, operations, tag, retain_self): if not self.is_valid: raise ValueError("invalid call object cannot be used from Python") - cdef OperationTag operation_tag = OperationTag(tag, operations) - if retain_self: - operation_tag.operation_call = self - else: - operation_tag.operation_call = None - operation_tag.store_ops() - cpython.Py_INCREF(operation_tag) + cdef _BatchOperationTag batch_operation_tag = _BatchOperationTag( + tag, operations, self if retain_self else None) + batch_operation_tag.prepare() + cpython.Py_INCREF(batch_operation_tag) return grpc_call_start_batch( - self.c_call, operation_tag.c_ops, operation_tag.c_nops, - <cpython.PyObject *>operation_tag, NULL) + self.c_call, batch_operation_tag.c_ops, batch_operation_tag.c_nops, + <cpython.PyObject *>batch_operation_tag, NULL) def start_client_batch(self, operations, tag): # We don't reference this call in the operations tag because diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi index 644df674cc..443d534d7e 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/channel.pyx.pxi @@ -76,12 +76,12 @@ cdef class Channel: def watch_connectivity_state( self, grpc_connectivity_state last_observed_state, Timespec deadline not None, CompletionQueue queue not None, tag): - cdef OperationTag operation_tag = OperationTag(tag, None) - cpython.Py_INCREF(operation_tag) + cdef _ConnectivityTag connectivity_tag = _ConnectivityTag(tag) + cpython.Py_INCREF(connectivity_tag) with nogil: grpc_channel_watch_connectivity_state( self.c_channel, last_observed_state, deadline.c_time, - queue.c_completion_queue, <cpython.PyObject *>operation_tag) + queue.c_completion_queue, <cpython.PyObject *>connectivity_tag) def target(self): cdef char *target = NULL diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi index 140fc357b9..e259789b35 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/completion_queue.pyx.pxi @@ -37,42 +37,20 @@ cdef class CompletionQueue: self.is_shutdown = False cdef _interpret_event(self, grpc_event event): - cdef OperationTag tag = None - cdef object user_tag = None - cdef Call operation_call = None - cdef CallDetails request_call_details = None - cdef object request_metadata = None - cdef object batch_operations = None + cdef _Tag tag = None if event.type == GRPC_QUEUE_TIMEOUT: - return Event( - event.type, False, None, None, None, None, False, None) + # NOTE(nathaniel): For now we coopt ConnectivityEvent here. + return ConnectivityEvent(GRPC_QUEUE_TIMEOUT, False, None) elif event.type == GRPC_QUEUE_SHUTDOWN: self.is_shutdown = True - return Event( - event.type, True, None, None, None, None, False, None) + # NOTE(nathaniel): For now we coopt ConnectivityEvent here. + return ConnectivityEvent(GRPC_QUEUE_TIMEOUT, True, None) else: - if event.tag != NULL: - tag = <OperationTag>event.tag - # We receive event tags only after they've been inc-ref'd elsewhere in - # the code. - cpython.Py_DECREF(tag) - if tag.shutting_down_server is not None: - tag.shutting_down_server.notify_shutdown_complete() - user_tag = tag.user_tag - operation_call = tag.operation_call - request_call_details = tag.request_call_details - if tag.is_new_request: - request_metadata = _metadata(&tag._c_request_metadata) - grpc_metadata_array_destroy(&tag._c_request_metadata) - batch_operations = tag.release_ops() - if tag.is_new_request: - # Stuff in the tag not explicitly handled by us needs to live through - # the life of the call - operation_call.references.extend(tag.references) - return Event( - event.type, event.success, user_tag, operation_call, - request_call_details, request_metadata, tag.is_new_request, - batch_operations) + tag = <_Tag>event.tag + # We receive event tags only after they've been inc-ref'd elsewhere in + # the code. + cpython.Py_DECREF(tag) + return tag.event(event) def poll(self, Timespec deadline=None): # We name this 'poll' to avoid problems with CPython's expectations for diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/event.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/event.pxd.pxi new file mode 100644 index 0000000000..686199ecf4 --- /dev/null +++ b/src/python/grpcio/grpc/_cython/_cygrpc/event.pxd.pxi @@ -0,0 +1,45 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +cdef class ConnectivityEvent: + + cdef readonly grpc_completion_type completion_type + cdef readonly bint success + cdef readonly object tag + + +cdef class RequestCallEvent: + + cdef readonly grpc_completion_type completion_type + cdef readonly bint success + cdef readonly object tag + cdef readonly Call call + cdef readonly CallDetails call_details + cdef readonly tuple invocation_metadata + + +cdef class BatchOperationEvent: + + cdef readonly grpc_completion_type completion_type + cdef readonly bint success + cdef readonly object tag + cdef readonly object batch_operations + + +cdef class ServerShutdownEvent: + + cdef readonly grpc_completion_type completion_type + cdef readonly bint success + cdef readonly object tag diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/event.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/event.pyx.pxi new file mode 100644 index 0000000000..af26d27318 --- /dev/null +++ b/src/python/grpcio/grpc/_cython/_cygrpc/event.pyx.pxi @@ -0,0 +1,55 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +cdef class ConnectivityEvent: + + def __cinit__( + self, grpc_completion_type completion_type, bint success, object tag): + self.completion_type = completion_type + self.success = success + self.tag = tag + + +cdef class RequestCallEvent: + + def __cinit__( + self, grpc_completion_type completion_type, bint success, object tag, + Call call, CallDetails call_details, tuple invocation_metadata): + self.completion_type = completion_type + self.success = success + self.tag = tag + self.call = call + self.call_details = call_details + self.invocation_metadata = invocation_metadata + + +cdef class BatchOperationEvent: + + def __cinit__( + self, grpc_completion_type completion_type, bint success, object tag, + object batch_operations): + self.completion_type = completion_type + self.success = success + self.tag = tag + self.batch_operations = batch_operations + + +cdef class ServerShutdownEvent: + + def __cinit__( + self, grpc_completion_type completion_type, bint success, object tag): + self.completion_type = completion_type + self.success = success + self.tag = tag diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi index 537cf2b537..7b2482d947 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/records.pxd.pxi @@ -28,43 +28,6 @@ cdef class CallDetails: cdef grpc_call_details c_details -cdef class OperationTag: - - cdef object user_tag - cdef list references - # This allows CompletionQueue to notify the Python Server object that the - # underlying GRPC core server has shutdown - cdef Server shutting_down_server - cdef Call operation_call - cdef CallDetails request_call_details - cdef grpc_metadata_array _c_request_metadata - cdef grpc_op *c_ops - cdef size_t c_nops - cdef readonly object _operations - cdef bint is_new_request - - cdef void store_ops(self) - cdef object release_ops(self) - - -cdef class Event: - - cdef readonly grpc_completion_type type - cdef readonly bint success - cdef readonly object tag - - # For Server.request_call - cdef readonly bint is_new_request - cdef readonly CallDetails request_call_details - cdef readonly object request_metadata - - # For server calls - cdef readonly Call operation_call - - # For Call.start_batch - cdef readonly object batch_operations - - cdef class SslPemKeyCertPair: cdef grpc_ssl_pem_key_cert_pair c_pair diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi index 99f8ffa3e7..1ff3b4471f 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/records.pyx.pxi @@ -218,50 +218,6 @@ cdef class CallDetails: return timespec -cdef class OperationTag: - - def __cinit__(self, user_tag, operations): - self.user_tag = user_tag - self.references = [] - self._operations = operations - - cdef void store_ops(self): - self.c_nops = 0 if self._operations is None else len(self._operations) - if 0 < self.c_nops: - self.c_ops = <grpc_op *>gpr_malloc(sizeof(grpc_op) * self.c_nops) - for index, operation in enumerate(self._operations): - (<Operation>operation).c() - self.c_ops[index] = (<Operation>operation).c_op - - cdef object release_ops(self): - if 0 < self.c_nops: - for index, operation in enumerate(self._operations): - (<Operation>operation).c_op = self.c_ops[index] - (<Operation>operation).un_c() - gpr_free(self.c_ops) - return self._operations - else: - return () - - -cdef class Event: - - def __cinit__(self, grpc_completion_type type, bint success, - object tag, Call operation_call, - CallDetails request_call_details, - object request_metadata, - bint is_new_request, - object batch_operations): - self.type = type - self.success = success - self.tag = tag - self.operation_call = operation_call - self.request_call_details = request_call_details - self.request_metadata = request_metadata - self.batch_operations = batch_operations - self.is_new_request = is_new_request - - cdef class SslPemKeyCertPair: def __cinit__(self, bytes private_key, bytes certificate_chain): diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi index f8d7892858..c19beccde6 100644 --- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi +++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi @@ -78,19 +78,15 @@ cdef class Server: raise ValueError("server must be started and not shutting down") if server_queue not in self.registered_completion_queues: raise ValueError("server_queue must be a registered completion queue") - cdef OperationTag operation_tag = OperationTag(tag, None) - operation_tag.operation_call = Call() - operation_tag.request_call_details = CallDetails() - grpc_metadata_array_init(&operation_tag._c_request_metadata) - operation_tag.references.extend([self, call_queue, server_queue]) - operation_tag.is_new_request = True - cpython.Py_INCREF(operation_tag) + cdef _RequestCallTag request_call_tag = _RequestCallTag(tag) + request_call_tag.prepare() + cpython.Py_INCREF(request_call_tag) return grpc_server_request_call( - self.c_server, &operation_tag.operation_call.c_call, - &operation_tag.request_call_details.c_details, - &operation_tag._c_request_metadata, + self.c_server, &request_call_tag.call.c_call, + &request_call_tag.call_details.c_details, + &request_call_tag.c_invocation_metadata, call_queue.c_completion_queue, server_queue.c_completion_queue, - <cpython.PyObject *>operation_tag) + <cpython.PyObject *>request_call_tag) def register_completion_queue( self, CompletionQueue queue not None): @@ -131,16 +127,14 @@ cdef class Server: cdef _c_shutdown(self, CompletionQueue queue, tag): self.is_shutting_down = True - operation_tag = OperationTag(tag, None) - operation_tag.shutting_down_server = self - cpython.Py_INCREF(operation_tag) + cdef _ServerShutdownTag server_shutdown_tag = _ServerShutdownTag(tag, self) + cpython.Py_INCREF(server_shutdown_tag) with nogil: grpc_server_shutdown_and_notify( self.c_server, queue.c_completion_queue, - <cpython.PyObject *>operation_tag) + <cpython.PyObject *>server_shutdown_tag) def shutdown(self, CompletionQueue queue not None, tag): - cdef OperationTag operation_tag if queue.is_shutting_down: raise ValueError("queue must be live") elif not self.is_started: @@ -153,7 +147,8 @@ cdef class Server: self._c_shutdown(queue, tag) cdef notify_shutdown_complete(self): - # called only by a completion queue on receiving our shutdown operation tag + # called only after our server shutdown tag has emerged from a completion + # queue. self.is_shutdown = True def cancel_all_calls(self): diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/tag.pxd.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/tag.pxd.pxi new file mode 100644 index 0000000000..f9a3b5e8f4 --- /dev/null +++ b/src/python/grpcio/grpc/_cython/_cygrpc/tag.pxd.pxi @@ -0,0 +1,58 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +cdef class _Tag: + + cdef object event(self, grpc_event c_event) + + +cdef class _ConnectivityTag(_Tag): + + cdef readonly object _user_tag + + cdef ConnectivityEvent event(self, grpc_event c_event) + + +cdef class _RequestCallTag(_Tag): + + cdef readonly object _user_tag + cdef Call call + cdef CallDetails call_details + cdef grpc_metadata_array c_invocation_metadata + + cdef void prepare(self) + cdef RequestCallEvent event(self, grpc_event c_event) + + +cdef class _BatchOperationTag(_Tag): + + cdef object _user_tag + cdef readonly object _operations + cdef readonly object _retained_call + cdef grpc_op *c_ops + cdef size_t c_nops + + cdef void prepare(self) + cdef BatchOperationEvent event(self, grpc_event c_event) + + +cdef class _ServerShutdownTag(_Tag): + + cdef readonly object _user_tag + # This allows CompletionQueue to notify the Python Server object that the + # underlying GRPC core server has shutdown + cdef readonly Server _shutting_down_server + + cdef ServerShutdownEvent event(self, grpc_event c_event) diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/tag.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/tag.pyx.pxi new file mode 100644 index 0000000000..aaca458442 --- /dev/null +++ b/src/python/grpcio/grpc/_cython/_cygrpc/tag.pyx.pxi @@ -0,0 +1,87 @@ +# Copyright 2017 gRPC authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +cdef class _Tag: + + cdef object event(self, grpc_event c_event): + raise NotImplementedError() + + +cdef class _ConnectivityTag(_Tag): + + def __cinit__(self, user_tag): + self._user_tag = user_tag + + cdef ConnectivityEvent event(self, grpc_event c_event): + return ConnectivityEvent(c_event.type, c_event.success, self._user_tag) + + +cdef class _RequestCallTag(_Tag): + + def __cinit__(self, user_tag): + self._user_tag = user_tag + self.call = None + self.call_details = None + + cdef void prepare(self): + self.call = Call() + self.call_details = CallDetails() + grpc_metadata_array_init(&self.c_invocation_metadata) + + cdef RequestCallEvent event(self, grpc_event c_event): + cdef tuple invocation_metadata = _metadata(&self.c_invocation_metadata) + grpc_metadata_array_destroy(&self.c_invocation_metadata) + return RequestCallEvent( + c_event.type, c_event.success, self._user_tag, self.call, + self.call_details, invocation_metadata) + + +cdef class _BatchOperationTag: + + def __cinit__(self, user_tag, operations, call): + self._user_tag = user_tag + self._operations = operations + self._retained_call = call + + cdef void prepare(self): + self.c_nops = 0 if self._operations is None else len(self._operations) + if 0 < self.c_nops: + self.c_ops = <grpc_op *>gpr_malloc(sizeof(grpc_op) * self.c_nops) + for index, operation in enumerate(self._operations): + (<Operation>operation).c() + self.c_ops[index] = (<Operation>operation).c_op + + cdef BatchOperationEvent event(self, grpc_event c_event): + if 0 < self.c_nops: + for index, operation in enumerate(self._operations): + (<Operation>operation).c_op = self.c_ops[index] + (<Operation>operation).un_c() + gpr_free(self.c_ops) + return BatchOperationEvent( + c_event.type, c_event.success, self._user_tag, self._operations) + else: + return BatchOperationEvent( + c_event.type, c_event.success, self._user_tag, ()) + + +cdef class _ServerShutdownTag(_Tag): + + def __cinit__(self, user_tag, shutting_down_server): + self._user_tag = user_tag + self._shutting_down_server = shutting_down_server + + cdef ServerShutdownEvent event(self, grpc_event c_event): + self._shutting_down_server.notify_shutdown_complete() + return ServerShutdownEvent(c_event.type, c_event.success, self._user_tag)
\ No newline at end of file diff --git a/src/python/grpcio/grpc/_cython/cygrpc.pxd b/src/python/grpcio/grpc/_cython/cygrpc.pxd index ad229de0ae..b32fa518fc 100644 --- a/src/python/grpcio/grpc/_cython/cygrpc.pxd +++ b/src/python/grpcio/grpc/_cython/cygrpc.pxd @@ -18,8 +18,10 @@ include "_cygrpc/call.pxd.pxi" include "_cygrpc/channel.pxd.pxi" include "_cygrpc/credentials.pxd.pxi" include "_cygrpc/completion_queue.pxd.pxi" +include "_cygrpc/event.pxd.pxi" include "_cygrpc/metadata.pxd.pxi" include "_cygrpc/operation.pxd.pxi" include "_cygrpc/records.pxd.pxi" include "_cygrpc/security.pxd.pxi" include "_cygrpc/server.pxd.pxi" +include "_cygrpc/tag.pxd.pxi" diff --git a/src/python/grpcio/grpc/_cython/cygrpc.pyx b/src/python/grpcio/grpc/_cython/cygrpc.pyx index 0964fb66ab..5106394708 100644 --- a/src/python/grpcio/grpc/_cython/cygrpc.pyx +++ b/src/python/grpcio/grpc/_cython/cygrpc.pyx @@ -25,11 +25,13 @@ include "_cygrpc/call.pyx.pxi" include "_cygrpc/channel.pyx.pxi" include "_cygrpc/credentials.pyx.pxi" include "_cygrpc/completion_queue.pyx.pxi" +include "_cygrpc/event.pyx.pxi" include "_cygrpc/metadata.pyx.pxi" include "_cygrpc/operation.pyx.pxi" include "_cygrpc/records.pyx.pxi" include "_cygrpc/security.pyx.pxi" include "_cygrpc/server.pyx.pxi" +include "_cygrpc/tag.pyx.pxi" # # initialize gRPC diff --git a/src/python/grpcio/grpc/_server.py b/src/python/grpcio/grpc/_server.py index eec31bdcf6..22244b9cec 100644 --- a/src/python/grpcio/grpc/_server.py +++ b/src/python/grpcio/grpc/_server.py @@ -217,11 +217,10 @@ class _Context(grpc.ServicerContext): def time_remaining(self): return max( - float(self._rpc_event.request_call_details.deadline) - time.time(), - 0) + float(self._rpc_event.call_details.deadline) - time.time(), 0) def cancel(self): - self._rpc_event.operation_call.cancel() + self._rpc_event.call.cancel() def add_callback(self, callback): with self._state.condition: @@ -236,23 +235,23 @@ class _Context(grpc.ServicerContext): self._state.disable_next_compression = True def invocation_metadata(self): - return self._rpc_event.request_metadata + return self._rpc_event.invocation_metadata def peer(self): - return _common.decode(self._rpc_event.operation_call.peer()) + return _common.decode(self._rpc_event.call.peer()) def peer_identities(self): - return cygrpc.peer_identities(self._rpc_event.operation_call) + return cygrpc.peer_identities(self._rpc_event.call) def peer_identity_key(self): - id_key = cygrpc.peer_identity_key(self._rpc_event.operation_call) + id_key = cygrpc.peer_identity_key(self._rpc_event.call) return id_key if id_key is None else _common.decode(id_key) def auth_context(self): return { _common.decode(key): value for key, value in six.iteritems( - cygrpc.auth_context(self._rpc_event.operation_call)) + cygrpc.auth_context(self._rpc_event.call)) } def send_initial_metadata(self, initial_metadata): @@ -263,7 +262,7 @@ class _Context(grpc.ServicerContext): if self._state.initial_metadata_allowed: operation = cygrpc.SendInitialMetadataOperation( initial_metadata, _EMPTY_FLAGS) - self._rpc_event.operation_call.start_server_batch( + self._rpc_event.call.start_server_batch( (operation,), _send_initial_metadata(self._state)) self._state.initial_metadata_allowed = False self._state.due.add(_SEND_INITIAL_METADATA_TOKEN) @@ -346,9 +345,9 @@ def _unary_request(rpc_event, state, request_deserializer): if state.client is _CANCELLED or state.statused: return None else: - rpc_event.operation_call.start_server_batch( + rpc_event.call.start_server_batch( (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),), - _receive_message(state, rpc_event.operation_call, + _receive_message(state, rpc_event.call, request_deserializer)) state.due.add(_RECEIVE_MESSAGE_TOKEN) while True: @@ -356,8 +355,8 @@ def _unary_request(rpc_event, state, request_deserializer): if state.request is None: if state.client is _CLOSED: details = '"{}" requires exactly one request message.'.format( - rpc_event.request_call_details.method) - _abort(state, rpc_event.operation_call, + rpc_event.call_details.method) + _abort(state, rpc_event.call, cygrpc.StatusCode.unimplemented, _common.encode(details)) return None @@ -378,13 +377,13 @@ def _call_behavior(rpc_event, state, behavior, argument, request_deserializer): except Exception as exception: # pylint: disable=broad-except with state.condition: if exception is state.abortion: - _abort(state, rpc_event.operation_call, - cygrpc.StatusCode.unknown, b'RPC Aborted') + _abort(state, rpc_event.call, cygrpc.StatusCode.unknown, + b'RPC Aborted') elif exception not in state.rpc_errors: details = 'Exception calling application: {}'.format(exception) logging.exception(details) - _abort(state, rpc_event.operation_call, - cygrpc.StatusCode.unknown, _common.encode(details)) + _abort(state, rpc_event.call, cygrpc.StatusCode.unknown, + _common.encode(details)) return None, False @@ -396,13 +395,13 @@ def _take_response_from_response_iterator(rpc_event, state, response_iterator): except Exception as exception: # pylint: disable=broad-except with state.condition: if exception is state.abortion: - _abort(state, rpc_event.operation_call, - cygrpc.StatusCode.unknown, b'RPC Aborted') + _abort(state, rpc_event.call, cygrpc.StatusCode.unknown, + b'RPC Aborted') elif exception not in state.rpc_errors: details = 'Exception iterating responses: {}'.format(exception) logging.exception(details) - _abort(state, rpc_event.operation_call, - cygrpc.StatusCode.unknown, _common.encode(details)) + _abort(state, rpc_event.call, cygrpc.StatusCode.unknown, + _common.encode(details)) return None, False @@ -410,7 +409,7 @@ def _serialize_response(rpc_event, state, response, response_serializer): serialized_response = _common.serialize(response, response_serializer) if serialized_response is None: with state.condition: - _abort(state, rpc_event.operation_call, cygrpc.StatusCode.internal, + _abort(state, rpc_event.call, cygrpc.StatusCode.internal, b'Failed to serialize response!') return None else: @@ -433,8 +432,8 @@ def _send_response(rpc_event, state, serialized_response): operations = (cygrpc.SendMessageOperation(serialized_response, _EMPTY_FLAGS),) token = _SEND_MESSAGE_TOKEN - rpc_event.operation_call.start_server_batch( - operations, _send_message(state, token)) + rpc_event.call.start_server_batch(operations, + _send_message(state, token)) state.due.add(token) while True: state.condition.wait() @@ -458,7 +457,7 @@ def _status(rpc_event, state, serialized_response): operations.append( cygrpc.SendMessageOperation(serialized_response, _EMPTY_FLAGS)) - rpc_event.operation_call.start_server_batch( + rpc_event.call.start_server_batch( operations, _send_status_from_server(state, _SEND_STATUS_FROM_SERVER_TOKEN)) state.statused = True @@ -525,7 +524,7 @@ def _handle_unary_stream(rpc_event, state, method_handler, thread_pool): def _handle_stream_unary(rpc_event, state, method_handler, thread_pool): - request_iterator = _RequestIterator(state, rpc_event.operation_call, + request_iterator = _RequestIterator(state, rpc_event.call, method_handler.request_deserializer) return thread_pool.submit( _unary_response_in_pool, rpc_event, state, method_handler.stream_unary, @@ -534,7 +533,7 @@ def _handle_stream_unary(rpc_event, state, method_handler, thread_pool): def _handle_stream_stream(rpc_event, state, method_handler, thread_pool): - request_iterator = _RequestIterator(state, rpc_event.operation_call, + request_iterator = _RequestIterator(state, rpc_event.call, method_handler.request_deserializer) return thread_pool.submit( _stream_response_in_pool, rpc_event, state, @@ -552,8 +551,8 @@ def _find_method_handler(rpc_event, generic_handlers, interceptor_pipeline): return None handler_call_details = _HandlerCallDetails( - _common.decode(rpc_event.request_call_details.method), - rpc_event.request_metadata) + _common.decode(rpc_event.call_details.method), + rpc_event.invocation_metadata) if interceptor_pipeline is not None: return interceptor_pipeline.execute(query_handlers, @@ -568,15 +567,15 @@ def _reject_rpc(rpc_event, status, details): cygrpc.SendStatusFromServerOperation(None, status, details, _EMPTY_FLAGS),) rpc_state = _RPCState() - rpc_event.operation_call.start_server_batch( - operations, lambda ignored_event: (rpc_state, (),)) + rpc_event.call.start_server_batch(operations, + lambda ignored_event: (rpc_state, (),)) return rpc_state def _handle_with_method_handler(rpc_event, method_handler, thread_pool): state = _RPCState() with state.condition: - rpc_event.operation_call.start_server_batch( + rpc_event.call.start_server_batch( (cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),), _receive_close_on_server(state)) state.due.add(_RECEIVE_CLOSE_ON_SERVER_TOKEN) @@ -600,7 +599,7 @@ def _handle_call(rpc_event, generic_handlers, interceptor_pipeline, thread_pool, concurrency_exceeded): if not rpc_event.success: return None, None - if rpc_event.request_call_details.method is not None: + if rpc_event.call_details.method is not None: try: method_handler = _find_method_handler(rpc_event, generic_handlers, interceptor_pipeline) diff --git a/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py b/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py index 458b4fe542..7c7353af94 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_cancel_many_calls_test.py @@ -53,7 +53,7 @@ class _Handler(object): self._state = state self._lock = threading.Lock() self._completion_queue = completion_queue - self._call = rpc_event.operation_call + self._call = rpc_event.call def __call__(self): with self._state.condition: diff --git a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py index 41291cc88f..583136cf23 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_server_completion_queue_per_call_test.py @@ -72,7 +72,7 @@ class Test(_common.RpcTest, unittest.TestCase): with server_call_condition: server_send_initial_metadata_start_batch_result = ( - server_request_call_event.operation_call.start_server_batch([ + server_request_call_event.call.start_server_batch([ cygrpc.SendInitialMetadataOperation( _common.INITIAL_METADATA, _common.EMPTY_FLAGS), ], server_send_initial_metadata_tag)) @@ -84,7 +84,7 @@ class Test(_common.RpcTest, unittest.TestCase): with server_call_condition: server_complete_rpc_start_batch_result = ( - server_request_call_event.operation_call.start_server_batch([ + server_request_call_event.call.start_server_batch([ cygrpc.ReceiveCloseOnServerOperation(_common.EMPTY_FLAGS), cygrpc.SendStatusFromServerOperation( _common.TRAILING_METADATA, cygrpc.StatusCode.ok, @@ -101,23 +101,24 @@ class Test(_common.RpcTest, unittest.TestCase): client_complete_rpc_event = self.client_driver.event_with_tag( client_complete_rpc_tag) - return (_common.OperationResult(server_request_call_start_batch_result, - server_request_call_event.type, - server_request_call_event.success), + return (_common.OperationResult( + server_request_call_start_batch_result, + server_request_call_event.completion_type, + server_request_call_event.success), _common.OperationResult( + client_receive_initial_metadata_start_batch_result, + client_receive_initial_metadata_event.completion_type, + client_receive_initial_metadata_event.success), _common.OperationResult( - client_receive_initial_metadata_start_batch_result, - client_receive_initial_metadata_event.type, - client_receive_initial_metadata_event.success), - _common.OperationResult(client_complete_rpc_start_batch_result, - client_complete_rpc_event.type, - client_complete_rpc_event.success), + client_complete_rpc_start_batch_result, + client_complete_rpc_event.completion_type, + client_complete_rpc_event.success), _common.OperationResult( + server_send_initial_metadata_start_batch_result, + server_send_initial_metadata_event.completion_type, + server_send_initial_metadata_event.success), _common.OperationResult( - server_send_initial_metadata_start_batch_result, - server_send_initial_metadata_event.type, - server_send_initial_metadata_event.success), - _common.OperationResult(server_complete_rpc_start_batch_result, - server_complete_rpc_event.type, - server_complete_rpc_event.success),) + server_complete_rpc_start_batch_result, + server_complete_rpc_event.completion_type, + server_complete_rpc_event.success),) def test_rpcs(self): expecteds = [(_common.SUCCESSFUL_OPERATION_RESULT,) * diff --git a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py index b429a20ed7..c5cf606c90 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_no_messages_single_server_completion_queue_test.py @@ -63,7 +63,7 @@ class Test(_common.RpcTest, unittest.TestCase): with self.server_condition: server_send_initial_metadata_start_batch_result = ( - server_request_call_event.operation_call.start_server_batch([ + server_request_call_event.call.start_server_batch([ cygrpc.SendInitialMetadataOperation( _common.INITIAL_METADATA, _common.EMPTY_FLAGS), ], server_send_initial_metadata_tag)) @@ -75,7 +75,7 @@ class Test(_common.RpcTest, unittest.TestCase): with self.server_condition: server_complete_rpc_start_batch_result = ( - server_request_call_event.operation_call.start_server_batch([ + server_request_call_event.call.start_server_batch([ cygrpc.ReceiveCloseOnServerOperation(_common.EMPTY_FLAGS), cygrpc.SendStatusFromServerOperation( _common.TRAILING_METADATA, cygrpc.StatusCode.ok, @@ -92,23 +92,24 @@ class Test(_common.RpcTest, unittest.TestCase): client_complete_rpc_event = self.client_driver.event_with_tag( client_complete_rpc_tag) - return (_common.OperationResult(server_request_call_start_batch_result, - server_request_call_event.type, - server_request_call_event.success), + return (_common.OperationResult( + server_request_call_start_batch_result, + server_request_call_event.completion_type, + server_request_call_event.success), _common.OperationResult( + client_receive_initial_metadata_start_batch_result, + client_receive_initial_metadata_event.completion_type, + client_receive_initial_metadata_event.success), _common.OperationResult( - client_receive_initial_metadata_start_batch_result, - client_receive_initial_metadata_event.type, - client_receive_initial_metadata_event.success), - _common.OperationResult(client_complete_rpc_start_batch_result, - client_complete_rpc_event.type, - client_complete_rpc_event.success), + client_complete_rpc_start_batch_result, + client_complete_rpc_event.completion_type, + client_complete_rpc_event.success), _common.OperationResult( + server_send_initial_metadata_start_batch_result, + server_send_initial_metadata_event.completion_type, + server_send_initial_metadata_event.success), _common.OperationResult( - server_send_initial_metadata_start_batch_result, - server_send_initial_metadata_event.type, - server_send_initial_metadata_event.success), - _common.OperationResult(server_complete_rpc_start_batch_result, - server_complete_rpc_event.type, - server_complete_rpc_event.success),) + server_complete_rpc_start_batch_result, + server_complete_rpc_event.completion_type, + server_complete_rpc_event.success),) def test_rpcs(self): expecteds = [(_common.SUCCESSFUL_OPERATION_RESULT,) * diff --git a/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py b/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py index a6d92f2203..655938e507 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/_read_some_but_not_all_responses_test.py @@ -174,12 +174,12 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase): with server_call_condition: server_send_initial_metadata_start_batch_result = ( - server_rpc_event.operation_call.start_server_batch([ + server_rpc_event.call.start_server_batch([ cygrpc.SendInitialMetadataOperation(_EMPTY_METADATA, _EMPTY_FLAGS), ], server_send_initial_metadata_tag)) server_send_first_message_start_batch_result = ( - server_rpc_event.operation_call.start_server_batch([ + server_rpc_event.call.start_server_batch([ cygrpc.SendMessageOperation(b'\x07', _EMPTY_FLAGS), ], server_send_first_message_tag)) server_send_initial_metadata_event = server_call_driver.event_with_tag( @@ -188,11 +188,11 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase): server_send_first_message_tag) with server_call_condition: server_send_second_message_start_batch_result = ( - server_rpc_event.operation_call.start_server_batch([ + server_rpc_event.call.start_server_batch([ cygrpc.SendMessageOperation(b'\x07', _EMPTY_FLAGS), ], server_send_second_message_tag)) server_complete_rpc_start_batch_result = ( - server_rpc_event.operation_call.start_server_batch([ + server_rpc_event.call.start_server_batch([ cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS), cygrpc.SendStatusFromServerOperation( (), cygrpc.StatusCode.ok, b'test details', @@ -231,9 +231,8 @@ class ReadSomeButNotAllResponsesTest(unittest.TestCase): self.assertEqual(cygrpc.CallError.ok, client_call_cancel_result) self.assertIs(server_rpc_tag, server_rpc_event.tag) self.assertEqual(cygrpc.CompletionType.operation_complete, - server_rpc_event.type) - self.assertIsInstance(server_rpc_event.operation_call, cygrpc.Call) - self.assertEqual(0, len(server_rpc_event.batch_operations)) + server_rpc_event.completion_type) + self.assertIsInstance(server_rpc_event.call, cygrpc.Call) if __name__ == '__main__': diff --git a/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py b/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py index 002fb9b7f4..d22cd79a45 100644 --- a/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py +++ b/src/python/grpcio_tests/tests/unit/_cython/cygrpc_test.py @@ -84,7 +84,8 @@ class TypeSmokeTest(unittest.TestCase): shutdown_tag = object() server.shutdown(completion_queue, shutdown_tag) event = completion_queue.poll() - self.assertEqual(cygrpc.CompletionType.operation_complete, event.type) + self.assertEqual(cygrpc.CompletionType.operation_complete, + event.completion_type) self.assertIs(shutdown_tag, event.tag) del server del completion_queue @@ -143,7 +144,7 @@ class ServerClientMixin(object): self.assertEqual(cygrpc.CallError.ok, call_result) event = queue.poll(deadline) self.assertEqual(cygrpc.CompletionType.operation_complete, - event.type) + event.completion_type) self.assertTrue(event.success) self.assertIs(tag, event.tag) except Exception as error: @@ -201,22 +202,20 @@ class ServerClientMixin(object): request_event = self.server_completion_queue.poll(cygrpc_deadline) self.assertEqual(cygrpc.CompletionType.operation_complete, - request_event.type) - self.assertIsInstance(request_event.operation_call, cygrpc.Call) + request_event.completion_type) + self.assertIsInstance(request_event.call, cygrpc.Call) self.assertIs(server_request_tag, request_event.tag) - self.assertEqual(0, len(request_event.batch_operations)) self.assertTrue( test_common.metadata_transmitted(client_initial_metadata, - request_event.request_metadata)) - self.assertEqual(METHOD, request_event.request_call_details.method) - self.assertEqual(self.expected_host, - request_event.request_call_details.host) + request_event.invocation_metadata)) + self.assertEqual(METHOD, request_event.call_details.method) + self.assertEqual(self.expected_host, request_event.call_details.host) self.assertLess( - abs(DEADLINE - float(request_event.request_call_details.deadline)), + abs(DEADLINE - float(request_event.call_details.deadline)), DEADLINE_TOLERANCE) server_call_tag = object() - server_call = request_event.operation_call + server_call = request_event.call server_initial_metadata = ( (SERVER_INITIAL_METADATA_KEY, SERVER_INITIAL_METADATA_VALUE,),) server_trailing_metadata = ( @@ -318,7 +317,7 @@ class ServerClientMixin(object): ], "Client prologue") request_event = self.server_completion_queue.poll(cygrpc_deadline) - server_call = request_event.operation_call + server_call = request_event.call def perform_server_operations(operations, description): return self._perform_operations(operations, server_call, |