diff options
Diffstat (limited to 'src/python')
17 files changed, 218 insertions, 148 deletions
diff --git a/src/python/grpcio/grpc/framework/core/_constants.py b/src/python/grpcio/grpc/framework/core/_constants.py index d3be3a4c4a..0f47cb48e0 100644 --- a/src/python/grpcio/grpc/framework/core/_constants.py +++ b/src/python/grpcio/grpc/framework/core/_constants.py @@ -44,14 +44,15 @@ TICKET_SUBSCRIPTION_FOR_BASE_SUBSCRIPTION_KIND = { # ticket should be sent to the other side in the event of such an # outcome. ABORTION_OUTCOME_TO_TICKET_TERMINATION = { - base.Outcome.CANCELLED: links.Ticket.Termination.CANCELLATION, - base.Outcome.EXPIRED: links.Ticket.Termination.EXPIRATION, - base.Outcome.LOCAL_SHUTDOWN: links.Ticket.Termination.SHUTDOWN, - base.Outcome.REMOTE_SHUTDOWN: None, - base.Outcome.RECEPTION_FAILURE: links.Ticket.Termination.RECEPTION_FAILURE, - base.Outcome.TRANSMISSION_FAILURE: None, - base.Outcome.LOCAL_FAILURE: links.Ticket.Termination.LOCAL_FAILURE, - base.Outcome.REMOTE_FAILURE: links.Ticket.Termination.REMOTE_FAILURE, + base.Outcome.Kind.CANCELLED: links.Ticket.Termination.CANCELLATION, + base.Outcome.Kind.EXPIRED: links.Ticket.Termination.EXPIRATION, + base.Outcome.Kind.LOCAL_SHUTDOWN: links.Ticket.Termination.SHUTDOWN, + base.Outcome.Kind.REMOTE_SHUTDOWN: None, + base.Outcome.Kind.RECEPTION_FAILURE: + links.Ticket.Termination.RECEPTION_FAILURE, + base.Outcome.Kind.TRANSMISSION_FAILURE: None, + base.Outcome.Kind.LOCAL_FAILURE: links.Ticket.Termination.LOCAL_FAILURE, + base.Outcome.Kind.REMOTE_FAILURE: links.Ticket.Termination.REMOTE_FAILURE, } INTERNAL_ERROR_LOG_MESSAGE = ':-( RPC Framework (Core) internal error! )-:' diff --git a/src/python/grpcio/grpc/framework/core/_context.py b/src/python/grpcio/grpc/framework/core/_context.py index 76b3534530..a346e9d478 100644 --- a/src/python/grpcio/grpc/framework/core/_context.py +++ b/src/python/grpcio/grpc/framework/core/_context.py @@ -33,6 +33,7 @@ import time # _interfaces is referenced from specification in this module. from grpc.framework.core import _interfaces # pylint: disable=unused-import +from grpc.framework.core import _utilities from grpc.framework.interfaces.base import base @@ -56,11 +57,12 @@ class OperationContext(base.OperationContext): self._transmission_manager = transmission_manager self._expiration_manager = expiration_manager - def _abort(self, outcome): + def _abort(self, outcome_kind): with self._lock: if self._termination_manager.outcome is None: + outcome = _utilities.Outcome(outcome_kind, None, None) self._termination_manager.abort(outcome) - self._transmission_manager.abort(outcome, None, None) + self._transmission_manager.abort(outcome) self._expiration_manager.terminate() def outcome(self): @@ -85,8 +87,8 @@ class OperationContext(base.OperationContext): def cancel(self): """See base.OperationContext.cancel for specification.""" - self._abort(base.Outcome.CANCELLED) + self._abort(base.Outcome.Kind.CANCELLED) def fail(self, exception): """See base.OperationContext.fail for specification.""" - self._abort(base.Outcome.LOCAL_FAILURE) + self._abort(base.Outcome.Kind.LOCAL_FAILURE) diff --git a/src/python/grpcio/grpc/framework/core/_emission.py b/src/python/grpcio/grpc/framework/core/_emission.py index 2d7b2e2f10..8ab59dc3e5 100644 --- a/src/python/grpcio/grpc/framework/core/_emission.py +++ b/src/python/grpcio/grpc/framework/core/_emission.py @@ -30,6 +30,7 @@ """State and behavior for handling emitted values.""" from grpc.framework.core import _interfaces +from grpc.framework.core import _utilities from grpc.framework.interfaces.base import base @@ -81,9 +82,10 @@ class EmissionManager(_interfaces.EmissionManager): payload_present and self._completion_seen or completion_present and self._completion_seen or allowance_present and allowance <= 0): - self._termination_manager.abort(base.Outcome.LOCAL_FAILURE) - self._transmission_manager.abort( - base.Outcome.LOCAL_FAILURE, None, None) + outcome = _utilities.Outcome( + base.Outcome.Kind.LOCAL_FAILURE, None, None) + self._termination_manager.abort(outcome) + self._transmission_manager.abort(outcome) self._expiration_manager.terminate() else: self._initial_metadata_seen |= initial_metadata_present diff --git a/src/python/grpcio/grpc/framework/core/_end.py b/src/python/grpcio/grpc/framework/core/_end.py index f57cde4e58..336e9c21fd 100644 --- a/src/python/grpcio/grpc/framework/core/_end.py +++ b/src/python/grpcio/grpc/framework/core/_end.py @@ -69,7 +69,7 @@ class _Cycle(object): def _abort(operations): for operation in operations: - operation.abort(base.Outcome.LOCAL_SHUTDOWN) + operation.abort(base.Outcome.Kind.LOCAL_SHUTDOWN) def _cancel_futures(futures): @@ -90,19 +90,19 @@ def _termination_action(lock, stats, operation_id, cycle): Args: lock: A lock to hold during the termination action. - states: A mapping from base.Outcome values to integers to increment with - the outcome given to the termination action. + stats: A mapping from base.Outcome.Kind values to integers to increment + with the outcome kind given to the termination action. operation_id: The operation ID for the termination action. cycle: A _Cycle value to be updated during the termination action. Returns: - A callable that takes an operation outcome as its sole parameter and that - should be used as the termination action for the operation associated - with the given operation ID. + A callable that takes an operation outcome kind as its sole parameter and + that should be used as the termination action for the operation + associated with the given operation ID. """ - def termination_action(outcome): + def termination_action(outcome_kind): with lock: - stats[outcome] += 1 + stats[outcome_kind] += 1 cycle.operations.pop(operation_id, None) if not cycle.operations: for action in cycle.idle_actions: @@ -127,7 +127,7 @@ class _End(End): self._lock = threading.Condition() self._servicer_package = servicer_package - self._stats = {outcome: 0 for outcome in base.Outcome} + self._stats = {outcome_kind: 0 for outcome_kind in base.Outcome.Kind} self._mate = None diff --git a/src/python/grpcio/grpc/framework/core/_expiration.py b/src/python/grpcio/grpc/framework/core/_expiration.py index d8690b3a02..ded0ab6bce 100644 --- a/src/python/grpcio/grpc/framework/core/_expiration.py +++ b/src/python/grpcio/grpc/framework/core/_expiration.py @@ -32,6 +32,7 @@ import time from grpc.framework.core import _interfaces +from grpc.framework.core import _utilities from grpc.framework.foundation import later from grpc.framework.interfaces.base import base @@ -73,7 +74,8 @@ class _ExpirationManager(_interfaces.ExpirationManager): if self._future is not None and index == self._index: self._future = None self._termination_manager.expire() - self._transmission_manager.abort(base.Outcome.EXPIRED, None, None) + self._transmission_manager.abort( + _utilities.Outcome(base.Outcome.Kind.EXPIRED, None, None)) return expire def start(self): diff --git a/src/python/grpcio/grpc/framework/core/_ingestion.py b/src/python/grpcio/grpc/framework/core/_ingestion.py index 766d57f931..9a7959a2dd 100644 --- a/src/python/grpcio/grpc/framework/core/_ingestion.py +++ b/src/python/grpcio/grpc/framework/core/_ingestion.py @@ -35,6 +35,7 @@ import enum from grpc.framework.core import _constants from grpc.framework.core import _interfaces +from grpc.framework.core import _utilities from grpc.framework.foundation import abandonment from grpc.framework.foundation import callable_util from grpc.framework.interfaces.base import base @@ -46,7 +47,7 @@ _INGESTION_EXCEPTION_LOG_MESSAGE = 'Exception during ingestion!' class _SubscriptionCreation( collections.namedtuple( '_SubscriptionCreation', - ('kind', 'subscription', 'code', 'message',))): + ('kind', 'subscription', 'code', 'details',))): """A sum type for the outcome of ingestion initialization. Attributes: @@ -56,7 +57,7 @@ class _SubscriptionCreation( code: A code value to be sent to the other side of the operation along with an indication that the operation is being aborted due to an error on the remote side of the operation. Only present if kind is Kind.REMOTE_ERROR. - message: A message value to be sent to the other side of the operation + details: A details value to be sent to the other side of the operation along with an indication that the operation is being aborted due to an error on the remote side of the operation. Only present if kind is Kind.REMOTE_ERROR. @@ -190,11 +191,13 @@ class _IngestionManager(_interfaces.IngestionManager): self._pending_payloads = None self._pending_completion = None - def _abort_and_notify(self, outcome, code, message): + def _abort_and_notify(self, outcome_kind, code, details): self._abort_internal_only() - self._termination_manager.abort(outcome) - self._transmission_manager.abort(outcome, code, message) - self._expiration_manager.terminate() + if self._termination_manager.outcome is None: + outcome = _utilities.Outcome(outcome_kind, code, details) + self._termination_manager.abort(outcome) + self._transmission_manager.abort(outcome) + self._expiration_manager.terminate() def _operator_next(self): """Computes the next step for full-subscription ingestion. @@ -250,12 +253,13 @@ class _IngestionManager(_interfaces.IngestionManager): else: with self._lock: if self._termination_manager.outcome is None: - self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None) + self._abort_and_notify( + base.Outcome.Kind.LOCAL_FAILURE, None, None) return else: with self._lock: if self._termination_manager.outcome is None: - self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None) + self._abort_and_notify(base.Outcome.Kind.LOCAL_FAILURE, None, None) return def _operator_post_create(self, subscription): @@ -279,17 +283,18 @@ class _IngestionManager(_interfaces.IngestionManager): if outcome.return_value is None: with self._lock: if self._termination_manager.outcome is None: - self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None) + self._abort_and_notify(base.Outcome.Kind.LOCAL_FAILURE, None, None) elif outcome.return_value.kind is _SubscriptionCreation.Kind.ABANDONED: with self._lock: if self._termination_manager.outcome is None: - self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None) + self._abort_and_notify(base.Outcome.Kind.LOCAL_FAILURE, None, None) elif outcome.return_value.kind is _SubscriptionCreation.Kind.REMOTE_ERROR: code = outcome.return_value.code - message = outcome.return_value.message + details = outcome.return_value.details with self._lock: if self._termination_manager.outcome is None: - self._abort_and_notify(base.Outcome.REMOTE_FAILURE, code, message) + self._abort_and_notify( + base.Outcome.Kind.REMOTE_FAILURE, code, details) elif outcome.return_value.subscription.kind is base.Subscription.Kind.FULL: self._operator_post_create(outcome.return_value.subscription) else: diff --git a/src/python/grpcio/grpc/framework/core/_interfaces.py b/src/python/grpcio/grpc/framework/core/_interfaces.py index deb5f34f9b..2a534cb7e7 100644 --- a/src/python/grpcio/grpc/framework/core/_interfaces.py +++ b/src/python/grpcio/grpc/framework/core/_interfaces.py @@ -50,13 +50,13 @@ class TerminationManager(object): If the operation has already terminated the callback will not be called. Args: - callback: A callable that will be passed an interfaces.Outcome value. + callback: A callable that will be passed a base.Outcome value. Returns: None if the operation has not yet terminated and the passed callback will - be called when it does, or a base.Outcome value describing the operation - termination if the operation has terminated and the callback will not be - called as a result of this method call. + be called when it does, or a base.Outcome value describing the + operation termination if the operation has terminated and the callback + will not be called as a result of this method call. """ raise NotImplementedError() @@ -76,8 +76,13 @@ class TerminationManager(object): raise NotImplementedError() @abc.abstractmethod - def reception_complete(self): - """Indicates that reception from the other side is complete.""" + def reception_complete(self, code, details): + """Indicates that reception from the other side is complete. + + Args: + code: An application-specific code value. + details: An application-specific details value. + """ raise NotImplementedError() @abc.abstractmethod @@ -95,7 +100,7 @@ class TerminationManager(object): """Indicates that the operation must abort for the indicated reason. Args: - outcome: An interfaces.Outcome indicating operation abortion. + outcome: A base.Outcome indicating operation abortion. """ raise NotImplementedError() @@ -155,19 +160,13 @@ class TransmissionManager(object): raise NotImplementedError() @abc.abstractmethod - def abort(self, outcome, code, message): + def abort(self, outcome): """Indicates that the operation has aborted. Args: - outcome: An interfaces.Outcome for the operation. If None, indicates that - the operation abortion should not be communicated to the other side of - the operation. - code: A code value to communicate to the other side of the operation - along with indication of operation abortion. May be None, and has no - effect if outcome is None. - message: A message value to communicate to the other side of the - operation along with indication of operation abortion. May be None, and - has no effect if outcome is None. + outcome: A base.Outcome for the operation. If None, indicates that the + operation abortion should not be communicated to the other side of the + operation. """ raise NotImplementedError() @@ -279,8 +278,7 @@ class ReceptionManager(object): """Handle a ticket from the other side of the operation. Args: - ticket: An interfaces.BackToFrontTicket or interfaces.FrontToBackTicket - appropriate to this end of the operation and this object. + ticket: A links.Ticket for the operation. """ raise NotImplementedError() @@ -305,10 +303,10 @@ class Operation(object): raise NotImplementedError() @abc.abstractmethod - def abort(self, outcome): + def abort(self, outcome_kind): """Aborts the operation. Args: - outcome: A base.Outcome value indicating operation abortion. + outcome_kind: A base.Outcome.Kind value indicating operation abortion. """ raise NotImplementedError() diff --git a/src/python/grpcio/grpc/framework/core/_operation.py b/src/python/grpcio/grpc/framework/core/_operation.py index cc873c03f9..f5679d0356 100644 --- a/src/python/grpcio/grpc/framework/core/_operation.py +++ b/src/python/grpcio/grpc/framework/core/_operation.py @@ -31,7 +31,6 @@ import threading -# _utilities is referenced from specification in this module. from grpc.framework.core import _context from grpc.framework.core import _emission from grpc.framework.core import _expiration @@ -40,7 +39,7 @@ from grpc.framework.core import _interfaces from grpc.framework.core import _reception from grpc.framework.core import _termination from grpc.framework.core import _transmission -from grpc.framework.core import _utilities # pylint: disable=unused-import +from grpc.framework.core import _utilities class _EasyOperation(_interfaces.Operation): @@ -75,11 +74,12 @@ class _EasyOperation(_interfaces.Operation): with self._lock: self._reception_manager.receive_ticket(ticket) - def abort(self, outcome): + def abort(self, outcome_kind): with self._lock: if self._termination_manager.outcome is None: + outcome = _utilities.Outcome(outcome_kind, None, None) self._termination_manager.abort(outcome) - self._transmission_manager.abort(outcome, None, None) + self._transmission_manager.abort(outcome) self._expiration_manager.terminate() diff --git a/src/python/grpcio/grpc/framework/core/_reception.py b/src/python/grpcio/grpc/framework/core/_reception.py index 1cebe3874b..d374cf0c8c 100644 --- a/src/python/grpcio/grpc/framework/core/_reception.py +++ b/src/python/grpcio/grpc/framework/core/_reception.py @@ -30,21 +30,26 @@ """State and behavior for ticket reception.""" from grpc.framework.core import _interfaces +from grpc.framework.core import _utilities from grpc.framework.interfaces.base import base from grpc.framework.interfaces.base import utilities from grpc.framework.interfaces.links import links -_REMOTE_TICKET_TERMINATION_TO_LOCAL_OUTCOME = { - links.Ticket.Termination.CANCELLATION: base.Outcome.CANCELLED, - links.Ticket.Termination.EXPIRATION: base.Outcome.EXPIRED, - links.Ticket.Termination.SHUTDOWN: base.Outcome.REMOTE_SHUTDOWN, - links.Ticket.Termination.RECEPTION_FAILURE: base.Outcome.RECEPTION_FAILURE, +_REMOTE_TICKET_TERMINATION_TO_LOCAL_OUTCOME_KIND = { + links.Ticket.Termination.CANCELLATION: base.Outcome.Kind.CANCELLED, + links.Ticket.Termination.EXPIRATION: base.Outcome.Kind.EXPIRED, + links.Ticket.Termination.SHUTDOWN: base.Outcome.Kind.REMOTE_SHUTDOWN, + links.Ticket.Termination.RECEPTION_FAILURE: + base.Outcome.Kind.RECEPTION_FAILURE, links.Ticket.Termination.TRANSMISSION_FAILURE: - base.Outcome.TRANSMISSION_FAILURE, - links.Ticket.Termination.LOCAL_FAILURE: base.Outcome.REMOTE_FAILURE, - links.Ticket.Termination.REMOTE_FAILURE: base.Outcome.LOCAL_FAILURE, + base.Outcome.Kind.TRANSMISSION_FAILURE, + links.Ticket.Termination.LOCAL_FAILURE: base.Outcome.Kind.REMOTE_FAILURE, + links.Ticket.Termination.REMOTE_FAILURE: base.Outcome.Kind.LOCAL_FAILURE, } +_RECEPTION_FAILURE_OUTCOME = _utilities.Outcome( + base.Outcome.Kind.RECEPTION_FAILURE, None, None) + class ReceptionManager(_interfaces.ReceptionManager): """A ReceptionManager based around a _Receiver passed to it.""" @@ -73,7 +78,7 @@ class ReceptionManager(_interfaces.ReceptionManager): self._aborted = True if self._termination_manager.outcome is None: self._termination_manager.abort(outcome) - self._transmission_manager.abort(None, None, None) + self._transmission_manager.abort(None) self._expiration_manager.terminate() def _sequence_failure(self, ticket): @@ -102,6 +107,7 @@ class ReceptionManager(_interfaces.ReceptionManager): else: completion = utilities.completion( ticket.terminal_metadata, ticket.code, ticket.message) + self._termination_manager.reception_complete(ticket.code, ticket.message) self._ingestion_manager.advance( ticket.initial_metadata, ticket.payload, completion, ticket.allowance) if ticket.allowance is not None: @@ -129,10 +135,12 @@ class ReceptionManager(_interfaces.ReceptionManager): if self._aborted: return elif self._sequence_failure(ticket): - self._abort(base.Outcome.RECEPTION_FAILURE) + self._abort(_RECEPTION_FAILURE_OUTCOME) elif ticket.termination not in (None, links.Ticket.Termination.COMPLETION): - outcome = _REMOTE_TICKET_TERMINATION_TO_LOCAL_OUTCOME[ticket.termination] - self._abort(outcome) + outcome_kind = _REMOTE_TICKET_TERMINATION_TO_LOCAL_OUTCOME_KIND[ + ticket.termination] + self._abort( + _utilities.Outcome(outcome_kind, ticket.code, ticket.message)) elif ticket.sequence_number == self._lowest_unseen_sequence_number: self._process(ticket) else: diff --git a/src/python/grpcio/grpc/framework/core/_termination.py b/src/python/grpcio/grpc/framework/core/_termination.py index ad9f6123d8..3bf7ade30f 100644 --- a/src/python/grpcio/grpc/framework/core/_termination.py +++ b/src/python/grpcio/grpc/framework/core/_termination.py @@ -33,6 +33,7 @@ import abc from grpc.framework.core import _constants from grpc.framework.core import _interfaces +from grpc.framework.core import _utilities from grpc.framework.foundation import callable_util from grpc.framework.interfaces.base import base @@ -74,7 +75,8 @@ class _TerminationManager(TerminationManager): predicate: One of _invocation_completion_predicate or _service_completion_predicate to be used to determine when the operation has completed. - action: A behavior to pass the operation outcome on operation termination. + action: A behavior to pass the operation outcome's kind on operation + termination. pool: A thread pool. """ self._predicate = predicate @@ -82,14 +84,19 @@ class _TerminationManager(TerminationManager): self._pool = pool self._expiration_manager = None - self.outcome = None self._callbacks = [] + self._code = None + self._details = None self._emission_complete = False self._transmission_complete = False self._reception_complete = False self._ingestion_complete = False + # The None-ness of outcome is the operation-wide record of whether and how + # the operation has terminated. + self.outcome = None + def set_expiration_manager(self, expiration_manager): self._expiration_manager = expiration_manager @@ -106,8 +113,8 @@ class _TerminationManager(TerminationManager): act = callable_util.with_exceptions_logged( self._action, _constants.INTERNAL_ERROR_LOG_MESSAGE) - if outcome is base.Outcome.LOCAL_FAILURE: - self._pool.submit(act, outcome) + if outcome.kind is base.Outcome.Kind.LOCAL_FAILURE: + self._pool.submit(act, base.Outcome.Kind.LOCAL_FAILURE) else: def call_callbacks_and_act(callbacks, outcome): for callback in callbacks: @@ -115,9 +122,11 @@ class _TerminationManager(TerminationManager): callback, _constants.TERMINATION_CALLBACK_EXCEPTION_LOG_MESSAGE, outcome) if callback_outcome.exception is not None: - outcome = base.Outcome.LOCAL_FAILURE + act_outcome_kind = base.Outcome.Kind.LOCAL_FAILURE break - act(outcome) + else: + act_outcome_kind = outcome.kind + act(act_outcome_kind) self._pool.submit( callable_util.with_exceptions_logged( @@ -132,7 +141,9 @@ class _TerminationManager(TerminationManager): if self._predicate( self._emission_complete, self._transmission_complete, self._reception_complete, self._ingestion_complete): - self._terminate_and_notify(base.Outcome.COMPLETED) + self._terminate_and_notify( + _utilities.Outcome( + base.Outcome.Kind.COMPLETED, self._code, self._details)) return True else: return False @@ -163,10 +174,12 @@ class _TerminationManager(TerminationManager): else: return False - def reception_complete(self): + def reception_complete(self, code, details): """See superclass method for specification.""" if self.outcome is None: self._reception_complete = True + self._code = code + self._details = details self._perhaps_complete() def ingestion_complete(self): @@ -177,7 +190,8 @@ class _TerminationManager(TerminationManager): def expire(self): """See _interfaces.TerminationManager.expire for specification.""" - self._terminate_internal_only(base.Outcome.EXPIRED) + self._terminate_internal_only( + _utilities.Outcome(base.Outcome.Kind.EXPIRED, None, None)) def abort(self, outcome): """See _interfaces.TerminationManager.abort for specification.""" diff --git a/src/python/grpcio/grpc/framework/core/_transmission.py b/src/python/grpcio/grpc/framework/core/_transmission.py index 202a71dd71..8f852cfe9f 100644 --- a/src/python/grpcio/grpc/framework/core/_transmission.py +++ b/src/python/grpcio/grpc/framework/core/_transmission.py @@ -34,12 +34,16 @@ import enum from grpc.framework.core import _constants from grpc.framework.core import _interfaces +from grpc.framework.core import _utilities from grpc.framework.foundation import callable_util from grpc.framework.interfaces.base import base from grpc.framework.interfaces.links import links _TRANSMISSION_EXCEPTION_LOG_MESSAGE = 'Exception during transmission!' +_TRANSMISSION_FAILURE_OUTCOME = _utilities.Outcome( + base.Outcome.Kind.TRANSMISSION_FAILURE, None, None) + def _explode_completion(completion): if completion is None: @@ -194,7 +198,7 @@ class TransmissionManager(_interfaces.TransmissionManager): with self._lock: self._abort = _ABORTED_NO_NOTIFY if self._termination_manager.outcome is None: - self._termination_manager.abort(base.Outcome.TRANSMISSION_FAILURE) + self._termination_manager.abort(_TRANSMISSION_FAILURE_OUTCOME) self._expiration_manager.terminate() return @@ -307,19 +311,24 @@ class TransmissionManager(_interfaces.TransmissionManager): self._remote_complete = True self._local_allowance = 0 - def abort(self, outcome, code, message): + def abort(self, outcome): """See _interfaces.TransmissionManager.abort for specification.""" if self._abort.kind is _Abort.Kind.NOT_ABORTED: - termination = _constants.ABORTION_OUTCOME_TO_TICKET_TERMINATION.get( - outcome) - if termination is None: + if outcome is None: self._abort = _ABORTED_NO_NOTIFY - elif self._transmitting: - self._abort = _Abort( - _Abort.Kind.ABORTED_NOTIFY_NEEDED, termination, code, message) else: - ticket = links.Ticket( - self._operation_id, self._lowest_unused_sequence_number, None, - None, None, None, None, None, None, None, code, message, - termination, None) - self._transmit(ticket) + termination = _constants.ABORTION_OUTCOME_TO_TICKET_TERMINATION.get( + outcome.kind) + if termination is None: + self._abort = _ABORTED_NO_NOTIFY + elif self._transmitting: + self._abort = _Abort( + _Abort.Kind.ABORTED_NOTIFY_NEEDED, termination, outcome.code, + outcome.details) + else: + ticket = links.Ticket( + self._operation_id, self._lowest_unused_sequence_number, None, + None, None, None, None, None, None, None, outcome.code, + outcome.details, termination, None) + self._transmit(ticket) + self._abort = _ABORTED_NO_NOTIFY diff --git a/src/python/grpcio/grpc/framework/core/_utilities.py b/src/python/grpcio/grpc/framework/core/_utilities.py index 5b0d798751..abedc727e4 100644 --- a/src/python/grpcio/grpc/framework/core/_utilities.py +++ b/src/python/grpcio/grpc/framework/core/_utilities.py @@ -31,6 +31,8 @@ import collections +from grpc.framework.interfaces.base import base + class ServicerPackage( collections.namedtuple( @@ -44,3 +46,9 @@ class ServicerPackage( maximum_timeout: A float indicating the maximum length of time in seconds to allow for an operation. """ + + +class Outcome( + base.Outcome, + collections.namedtuple('Outcome', ('kind', 'code', 'details',))): + """A trivial implementation of base.Outcome.""" diff --git a/src/python/grpcio/grpc/framework/crust/_control.py b/src/python/grpcio/grpc/framework/crust/_control.py index 01de3c15bd..7bddf46a57 100644 --- a/src/python/grpcio/grpc/framework/crust/_control.py +++ b/src/python/grpcio/grpc/framework/crust/_control.py @@ -110,30 +110,31 @@ class _Termination( _NOT_TERMINATED = _Termination(False, None, None) -_OPERATION_OUTCOME_TO_TERMINATION_CONSTRUCTOR = { - base.Outcome.COMPLETED: lambda *unused_args: _Termination(True, None, None), - base.Outcome.CANCELLED: lambda *args: _Termination( +_OPERATION_OUTCOME_KIND_TO_TERMINATION_CONSTRUCTOR = { + base.Outcome.Kind.COMPLETED: lambda *unused_args: _Termination( + True, None, None), + base.Outcome.Kind.CANCELLED: lambda *args: _Termination( True, face.Abortion(face.Abortion.Kind.CANCELLED, *args), face.CancellationError(*args)), - base.Outcome.EXPIRED: lambda *args: _Termination( + base.Outcome.Kind.EXPIRED: lambda *args: _Termination( True, face.Abortion(face.Abortion.Kind.EXPIRED, *args), face.ExpirationError(*args)), - base.Outcome.LOCAL_SHUTDOWN: lambda *args: _Termination( + base.Outcome.Kind.LOCAL_SHUTDOWN: lambda *args: _Termination( True, face.Abortion(face.Abortion.Kind.LOCAL_SHUTDOWN, *args), face.LocalShutdownError(*args)), - base.Outcome.REMOTE_SHUTDOWN: lambda *args: _Termination( + base.Outcome.Kind.REMOTE_SHUTDOWN: lambda *args: _Termination( True, face.Abortion(face.Abortion.Kind.REMOTE_SHUTDOWN, *args), face.RemoteShutdownError(*args)), - base.Outcome.RECEPTION_FAILURE: lambda *args: _Termination( + base.Outcome.Kind.RECEPTION_FAILURE: lambda *args: _Termination( True, face.Abortion(face.Abortion.Kind.NETWORK_FAILURE, *args), face.NetworkError(*args)), - base.Outcome.TRANSMISSION_FAILURE: lambda *args: _Termination( + base.Outcome.Kind.TRANSMISSION_FAILURE: lambda *args: _Termination( True, face.Abortion(face.Abortion.Kind.NETWORK_FAILURE, *args), face.NetworkError(*args)), - base.Outcome.LOCAL_FAILURE: lambda *args: _Termination( + base.Outcome.Kind.LOCAL_FAILURE: lambda *args: _Termination( True, face.Abortion(face.Abortion.Kind.LOCAL_FAILURE, *args), face.LocalError(*args)), - base.Outcome.REMOTE_FAILURE: lambda *args: _Termination( + base.Outcome.Kind.REMOTE_FAILURE: lambda *args: _Termination( True, face.Abortion(face.Abortion.Kind.REMOTE_FAILURE, *args), face.RemoteError(*args)), } @@ -247,13 +248,17 @@ class Rendezvous(base.Operator, future.Future, stream.Consumer, face.Call): else: initial_metadata = self._up_initial_metadata.value if self._up_completion.kind is _Awaited.Kind.NOT_YET_ARRIVED: - terminal_metadata, code, details = None, None, None + terminal_metadata = None else: terminal_metadata = self._up_completion.value.terminal_metadata + if outcome.kind is base.Outcome.Kind.COMPLETED: code = self._up_completion.value.code details = self._up_completion.value.message - self._termination = _OPERATION_OUTCOME_TO_TERMINATION_CONSTRUCTOR[ - outcome](initial_metadata, terminal_metadata, code, details) + else: + code = outcome.code + details = outcome.details + self._termination = _OPERATION_OUTCOME_KIND_TO_TERMINATION_CONSTRUCTOR[ + outcome.kind](initial_metadata, terminal_metadata, code, details) self._condition.notify_all() diff --git a/src/python/grpcio/grpc/framework/interfaces/base/base.py b/src/python/grpcio/grpc/framework/interfaces/base/base.py index bc52efb4c5..0d9d6b464e 100644 --- a/src/python/grpcio/grpc/framework/interfaces/base/base.py +++ b/src/python/grpcio/grpc/framework/interfaces/base/base.py @@ -40,7 +40,7 @@ applications choose. # threading is referenced from specification in this module. import abc import enum -import threading +import threading # pylint: disable=unused-import # abandonment is referenced from specification in this module. from grpc.framework.foundation import abandonment # pylint: disable=unused-import @@ -69,19 +69,30 @@ class NoSuchMethodError(Exception): self.details = details -@enum.unique -class Outcome(enum.Enum): - """Operation outcomes.""" +class Outcome(object): + """The outcome of an operation. - COMPLETED = 'completed' - CANCELLED = 'cancelled' - EXPIRED = 'expired' - LOCAL_SHUTDOWN = 'local shutdown' - REMOTE_SHUTDOWN = 'remote shutdown' - RECEPTION_FAILURE = 'reception failure' - TRANSMISSION_FAILURE = 'transmission failure' - LOCAL_FAILURE = 'local failure' - REMOTE_FAILURE = 'remote failure' + Attributes: + kind: A Kind value coarsely identifying how the operation terminated. + code: An application-specific code value or None if no such value was + provided. + details: An application-specific details value or None if no such value was + provided. + """ + + @enum.unique + class Kind(enum.Enum): + """Ways in which an operation can terminate.""" + + COMPLETED = 'completed' + CANCELLED = 'cancelled' + EXPIRED = 'expired' + LOCAL_SHUTDOWN = 'local shutdown' + REMOTE_SHUTDOWN = 'remote shutdown' + RECEPTION_FAILURE = 'reception failure' + TRANSMISSION_FAILURE = 'transmission failure' + LOCAL_FAILURE = 'local failure' + REMOTE_FAILURE = 'remote failure' class Completion(object): @@ -294,8 +305,8 @@ class End(object): """Reports the number of terminated operations broken down by outcome. Returns: - A dictionary from Outcome value to an integer identifying the number - of operations that terminated with that outcome. + A dictionary from Outcome.Kind value to an integer identifying the number + of operations that terminated with that outcome kind. """ raise NotImplementedError() diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/_control.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/_control.py index e4d2a7a0d7..46a01876d8 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/_control.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/_control.py @@ -236,8 +236,8 @@ class Instruction( collections.namedtuple( 'Instruction', ('kind', 'advance_args', 'advance_kwargs', 'conclude_success', - 'conclude_message', 'conclude_invocation_outcome', - 'conclude_service_outcome',))): + 'conclude_message', 'conclude_invocation_outcome_kind', + 'conclude_service_outcome_kind',))): """""" @enum.unique @@ -532,24 +532,24 @@ class _SequenceController(Controller): self._state.service_side_outcome = outcome if self._todo is not None or self._remaining_elements: self._failed('Premature service-side outcome %s!' % (outcome,)) - elif outcome is not self._sequence.outcome.service: + elif outcome.kind is not self._sequence.outcome_kinds.service: self._failed( - 'Incorrect service-side outcome: %s should have been %s' % ( - outcome, self._sequence.outcome.service)) + 'Incorrect service-side outcome kind: %s should have been %s' % ( + outcome.kind, self._sequence.outcome_kinds.service)) elif self._state.invocation_side_outcome is not None: - self._passed(self._state.invocation_side_outcome, outcome) + self._passed(self._state.invocation_side_outcome.kind, outcome.kind) def invocation_on_termination(self, outcome): with self._condition: self._state.invocation_side_outcome = outcome if self._todo is not None or self._remaining_elements: self._failed('Premature invocation-side outcome %s!' % (outcome,)) - elif outcome is not self._sequence.outcome.invocation: + elif outcome.kind is not self._sequence.outcome_kinds.invocation: self._failed( - 'Incorrect invocation-side outcome: %s should have been %s' % ( - outcome, self._sequence.outcome.invocation)) + 'Incorrect invocation-side outcome kind: %s should have been %s' % ( + outcome.kind, self._sequence.outcome_kinds.invocation)) elif self._state.service_side_outcome is not None: - self._passed(outcome, self._state.service_side_outcome) + self._passed(outcome.kind, self._state.service_side_outcome.kind) class _SequenceControllerCreator(ControllerCreator): diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/_sequence.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/_sequence.py index 1d77aaebe6..f547d91681 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/_sequence.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/_sequence.py @@ -103,13 +103,14 @@ class Element(collections.namedtuple('Element', ('kind', 'transmission',))): SERVICE_FAILURE = 'service failure' -class Outcome(collections.namedtuple('Outcome', ('invocation', 'service',))): +class OutcomeKinds( + collections.namedtuple('Outcome', ('invocation', 'service',))): """A description of the expected outcome of an operation test. Attributes: - invocation: The base.Outcome value expected on the invocation side of the - operation. - service: The base.Outcome value expected on the service side of the + invocation: The base.Outcome.Kind value expected on the invocation side of + the operation. + service: The base.Outcome.Kind value expected on the service side of the operation. """ @@ -117,7 +118,8 @@ class Outcome(collections.namedtuple('Outcome', ('invocation', 'service',))): class Sequence( collections.namedtuple( 'Sequence', - ('name', 'maximum_duration', 'invocation', 'elements', 'outcome',))): + ('name', 'maximum_duration', 'invocation', 'elements', + 'outcome_kinds',))): """Describes at a high level steps to perform in a test. Attributes: @@ -128,7 +130,8 @@ class Sequence( under test. elements: A sequence of Element values describing at coarse granularity actions to take during the operation under test. - outcome: An Outcome value describing the expected outcome of the test. + outcome_kinds: An OutcomeKinds value describing the expected outcome kinds + of the test. """ _EASY = Sequence( @@ -139,7 +142,7 @@ _EASY = Sequence( Element( Element.Kind.SERVICE_TRANSMISSION, Transmission(True, True, True)), ), - Outcome(base.Outcome.COMPLETED, base.Outcome.COMPLETED)) + OutcomeKinds(base.Outcome.Kind.COMPLETED, base.Outcome.Kind.COMPLETED)) _PEASY = Sequence( 'Peasy', @@ -154,7 +157,7 @@ _PEASY = Sequence( Element( Element.Kind.SERVICE_TRANSMISSION, Transmission(False, True, True)), ), - Outcome(base.Outcome.COMPLETED, base.Outcome.COMPLETED)) + OutcomeKinds(base.Outcome.Kind.COMPLETED, base.Outcome.Kind.COMPLETED)) # TODO(issue 2959): Finish this test suite. This tuple of sequences should diff --git a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py index 87332cf612..5065a3f38a 100644 --- a/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py +++ b/src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py @@ -44,7 +44,8 @@ from grpc_test.framework.interfaces.base import test_interfaces _SYNCHRONICITY_VARIATION = (('Sync', False), ('Async', True)) -_EMPTY_OUTCOME_DICT = {outcome: 0 for outcome in base.Outcome} +_EMPTY_OUTCOME_KIND_DICT = { + outcome_kind: 0 for outcome_kind in base.Outcome.Kind} class _Serialization(test_interfaces.Serialization): @@ -119,7 +120,7 @@ class _Operator(base.Operator): class _Servicer(base.Servicer): - """An base.Servicer with instrumented for testing.""" + """A base.Servicer with instrumented for testing.""" def __init__(self, group, method, controllers, pool): self._condition = threading.Condition() @@ -223,11 +224,12 @@ class _OperationTest(unittest.TestCase): self.assertTrue( instruction.conclude_success, msg=instruction.conclude_message) - expected_invocation_stats = dict(_EMPTY_OUTCOME_DICT) - expected_invocation_stats[instruction.conclude_invocation_outcome] += 1 + expected_invocation_stats = dict(_EMPTY_OUTCOME_KIND_DICT) + expected_invocation_stats[ + instruction.conclude_invocation_outcome_kind] += 1 self.assertDictEqual(expected_invocation_stats, invocation_stats) - expected_service_stats = dict(_EMPTY_OUTCOME_DICT) - expected_service_stats[instruction.conclude_service_outcome] += 1 + expected_service_stats = dict(_EMPTY_OUTCOME_KIND_DICT) + expected_service_stats[instruction.conclude_service_outcome_kind] += 1 self.assertDictEqual(expected_service_stats, service_stats) |