浏览代码

Merge github.com:grpc/grpc into endpoints

Craig Tiller 10 年之前
父节点
当前提交
c4a9f16f93

+ 12 - 2
src/python/README.md

@@ -52,9 +52,19 @@ BUILDING FROM SOURCE
 ---------------------
 ---------------------
 - Clone this repository
 - Clone this repository
 
 
+- Initialize the git submodules
+```
+$ git submodule update --init
+```
+
+- Make the libraries
+```
+$ make
+```
+
 - Use build_python.sh to build the Python code and install it into a virtual environment
 - Use build_python.sh to build the Python code and install it into a virtual environment
 ```
 ```
-$ tools/run_tests/build_python.sh
+$ CONFIG=opt tools/run_tests/build_python.sh 2.7
 ```
 ```
 
 
 TESTING
 TESTING
@@ -62,7 +72,7 @@ TESTING
 
 
 - Use run_python.sh to run gRPC as it was installed into the virtual environment
 - Use run_python.sh to run gRPC as it was installed into the virtual environment
 ```
 ```
-$ tools/run_tests/run_python.sh
+$ CONFIG=opt PYVER=2.7 tools/run_tests/run_python.sh
 ```
 ```
 
 
 PACKAGING
 PACKAGING

+ 1 - 1
src/python/grpcio/grpc/framework/core/_context.py

@@ -60,7 +60,7 @@ class OperationContext(base.OperationContext):
     with self._lock:
     with self._lock:
       if self._termination_manager.outcome is None:
       if self._termination_manager.outcome is None:
         self._termination_manager.abort(outcome)
         self._termination_manager.abort(outcome)
-        self._transmission_manager.abort(outcome)
+        self._transmission_manager.abort(outcome, None, None)
         self._expiration_manager.terminate()
         self._expiration_manager.terminate()
 
 
   def outcome(self):
   def outcome(self):

+ 2 - 1
src/python/grpcio/grpc/framework/core/_emission.py

@@ -82,7 +82,8 @@ class EmissionManager(_interfaces.EmissionManager):
             completion_present and self._completion_seen or
             completion_present and self._completion_seen or
             allowance_present and allowance <= 0):
             allowance_present and allowance <= 0):
           self._termination_manager.abort(base.Outcome.LOCAL_FAILURE)
           self._termination_manager.abort(base.Outcome.LOCAL_FAILURE)
-          self._transmission_manager.abort(base.Outcome.LOCAL_FAILURE)
+          self._transmission_manager.abort(
+              base.Outcome.LOCAL_FAILURE, None, None)
           self._expiration_manager.terminate()
           self._expiration_manager.terminate()
         else:
         else:
           self._initial_metadata_seen |= initial_metadata_present
           self._initial_metadata_seen |= initial_metadata_present

+ 1 - 1
src/python/grpcio/grpc/framework/core/_expiration.py

@@ -73,7 +73,7 @@ class _ExpirationManager(_interfaces.ExpirationManager):
         if self._future is not None and index == self._index:
         if self._future is not None and index == self._index:
           self._future = None
           self._future = None
           self._termination_manager.expire()
           self._termination_manager.expire()
-          self._transmission_manager.abort(base.Outcome.EXPIRED)
+          self._transmission_manager.abort(base.Outcome.EXPIRED, None, None)
     return expire
     return expire
 
 
   def start(self):
   def start(self):

+ 41 - 25
src/python/grpcio/grpc/framework/core/_ingestion.py

@@ -31,6 +31,7 @@
 
 
 import abc
 import abc
 import collections
 import collections
+import enum
 
 
 from grpc.framework.core import _constants
 from grpc.framework.core import _constants
 from grpc.framework.core import _interfaces
 from grpc.framework.core import _interfaces
@@ -42,21 +43,31 @@ _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE = 'Exception initializing ingestion!'
 _INGESTION_EXCEPTION_LOG_MESSAGE = 'Exception during ingestion!'
 _INGESTION_EXCEPTION_LOG_MESSAGE = 'Exception during ingestion!'
 
 
 
 
-class _SubscriptionCreation(collections.namedtuple(
-    '_SubscriptionCreation', ('subscription', 'remote_error', 'abandoned'))):
+class _SubscriptionCreation(
+    collections.namedtuple(
+        '_SubscriptionCreation',
+        ('kind', 'subscription', 'code', 'message',))):
   """A sum type for the outcome of ingestion initialization.
   """A sum type for the outcome of ingestion initialization.
 
 
-  Either subscription will be non-None, remote_error will be True, or abandoned
-  will be True.
-
   Attributes:
   Attributes:
-    subscription: A base.Subscription describing the customer's interest in
-      operation values from the other side.
-    remote_error: A boolean indicating that the subscription could not be
-      created due to an error on the remote side of the operation.
-    abandoned: A boolean indicating that subscription creation was abandoned.
+    kind: A Kind value coarsely indicating how subscription creation completed.
+    subscription: The created subscription. Only present if kind is
+      Kind.SUBSCRIPTION.
+    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
+      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.
   """
   """
 
 
+  @enum.unique
+  class Kind(enum.Enum):
+    SUBSCRIPTION = 'subscription'
+    REMOTE_ERROR = 'remote error'
+    ABANDONED = 'abandoned'
+
 
 
 class _SubscriptionCreator(object):
 class _SubscriptionCreator(object):
   """Common specification of subscription-creating behavior."""
   """Common specification of subscription-creating behavior."""
@@ -101,12 +112,15 @@ class _ServiceSubscriptionCreator(_SubscriptionCreator):
     try:
     try:
       subscription = self._servicer.service(
       subscription = self._servicer.service(
           group, method, self._operation_context, self._output_operator)
           group, method, self._operation_context, self._output_operator)
-    except base.NoSuchMethodError:
-      return _SubscriptionCreation(None, True, False)
+    except base.NoSuchMethodError as e:
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.REMOTE_ERROR, None, e.code, e.message)
     except abandonment.Abandoned:
     except abandonment.Abandoned:
-      return _SubscriptionCreation(None, False, True)
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.ABANDONED, None, None, None)
     else:
     else:
-      return _SubscriptionCreation(subscription, False, False)
+      return _SubscriptionCreation(
+          _SubscriptionCreation.Kind.SUBSCRIPTION, subscription, None, None)
 
 
 
 
 def _wrap(behavior):
 def _wrap(behavior):
@@ -176,10 +190,10 @@ class _IngestionManager(_interfaces.IngestionManager):
     self._pending_payloads = None
     self._pending_payloads = None
     self._pending_completion = None
     self._pending_completion = None
 
 
-  def _abort_and_notify(self, outcome):
+  def _abort_and_notify(self, outcome, code, message):
     self._abort_internal_only()
     self._abort_internal_only()
     self._termination_manager.abort(outcome)
     self._termination_manager.abort(outcome)
-    self._transmission_manager.abort(outcome)
+    self._transmission_manager.abort(outcome, code, message)
     self._expiration_manager.terminate()
     self._expiration_manager.terminate()
 
 
   def _operator_next(self):
   def _operator_next(self):
@@ -236,12 +250,12 @@ class _IngestionManager(_interfaces.IngestionManager):
         else:
         else:
           with self._lock:
           with self._lock:
             if self._termination_manager.outcome is None:
             if self._termination_manager.outcome is None:
-              self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+              self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
             return
             return
       else:
       else:
         with self._lock:
         with self._lock:
           if self._termination_manager.outcome is None:
           if self._termination_manager.outcome is None:
-            self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
+            self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
           return
           return
 
 
   def _operator_post_create(self, subscription):
   def _operator_post_create(self, subscription):
@@ -260,20 +274,22 @@ class _IngestionManager(_interfaces.IngestionManager):
 
 
   def _create(self, subscription_creator, group, name):
   def _create(self, subscription_creator, group, name):
     outcome = callable_util.call_logging_exceptions(
     outcome = callable_util.call_logging_exceptions(
-        subscription_creator.create, _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE,
-        group, name)
+        subscription_creator.create,
+        _CREATE_SUBSCRIPTION_EXCEPTION_LOG_MESSAGE, group, name)
     if outcome.return_value is None:
     if outcome.return_value is None:
       with self._lock:
       with self._lock:
         if self._termination_manager.outcome is None:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
-    elif outcome.return_value.abandoned:
+          self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
+    elif outcome.return_value.kind is _SubscriptionCreation.Kind.ABANDONED:
       with self._lock:
       with self._lock:
         if self._termination_manager.outcome is None:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.LOCAL_FAILURE)
-    elif outcome.return_value.remote_error:
+          self._abort_and_notify(base.Outcome.LOCAL_FAILURE, None, None)
+    elif outcome.return_value.kind is _SubscriptionCreation.Kind.REMOTE_ERROR:
+      code = outcome.return_value.code
+      message = outcome.return_value.message
       with self._lock:
       with self._lock:
         if self._termination_manager.outcome is None:
         if self._termination_manager.outcome is None:
-          self._abort_and_notify(base.Outcome.REMOTE_FAILURE)
+          self._abort_and_notify(base.Outcome.REMOTE_FAILURE, code, message)
     elif outcome.return_value.subscription.kind is base.Subscription.Kind.FULL:
     elif outcome.return_value.subscription.kind is base.Subscription.Kind.FULL:
       self._operator_post_create(outcome.return_value.subscription)
       self._operator_post_create(outcome.return_value.subscription)
     else:
     else:

+ 7 - 1
src/python/grpcio/grpc/framework/core/_interfaces.py

@@ -155,13 +155,19 @@ class TransmissionManager(object):
     raise NotImplementedError()
     raise NotImplementedError()
 
 
   @abc.abstractmethod
   @abc.abstractmethod
-  def abort(self, outcome):
+  def abort(self, outcome, code, message):
     """Indicates that the operation has aborted.
     """Indicates that the operation has aborted.
 
 
     Args:
     Args:
       outcome: An interfaces.Outcome for the operation. If None, indicates that
       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 abortion should not be communicated to the other side of
         the operation.
         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.
     """
     """
     raise NotImplementedError()
     raise NotImplementedError()
 
 

+ 1 - 1
src/python/grpcio/grpc/framework/core/_operation.py

@@ -79,7 +79,7 @@ class _EasyOperation(_interfaces.Operation):
     with self._lock:
     with self._lock:
       if self._termination_manager.outcome is None:
       if self._termination_manager.outcome is None:
         self._termination_manager.abort(outcome)
         self._termination_manager.abort(outcome)
-        self._transmission_manager.abort(outcome)
+        self._transmission_manager.abort(outcome, None, None)
         self._expiration_manager.terminate()
         self._expiration_manager.terminate()
 
 
 
 

+ 1 - 1
src/python/grpcio/grpc/framework/core/_reception.py

@@ -73,7 +73,7 @@ class ReceptionManager(_interfaces.ReceptionManager):
     self._aborted = True
     self._aborted = True
     if self._termination_manager.outcome is None:
     if self._termination_manager.outcome is None:
       self._termination_manager.abort(outcome)
       self._termination_manager.abort(outcome)
-      self._transmission_manager.abort(None)
+      self._transmission_manager.abort(None, None, None)
       self._expiration_manager.terminate()
       self._expiration_manager.terminate()
 
 
   def _sequence_failure(self, ticket):
   def _sequence_failure(self, ticket):

+ 11 - 3
src/python/grpcio/grpc/framework/core/_transmission.py

@@ -104,9 +104,13 @@ class TransmissionManager(_interfaces.TransmissionManager):
           return None
           return None
         else:
         else:
           self._abortion_outcome = None
           self._abortion_outcome = None
+          if self._completion is None:
+            code, message = None, None
+          else:
+            code, message = self._completion.code, self._completion.message
           return links.Ticket(
           return links.Ticket(
               self._operation_id, self._lowest_unused_sequence_number, None,
               self._operation_id, self._lowest_unused_sequence_number, None,
-              None, None, None, None, None, None, None, None, None,
+              None, None, None, None, None, None, None, code, message,
               termination, None)
               termination, None)
 
 
     action = False
     action = False
@@ -277,7 +281,7 @@ class TransmissionManager(_interfaces.TransmissionManager):
     self._remote_complete = True
     self._remote_complete = True
     self._local_allowance = 0
     self._local_allowance = 0
 
 
-  def abort(self, outcome):
+  def abort(self, outcome, code, message):
     """See _interfaces.TransmissionManager.abort for specification."""
     """See _interfaces.TransmissionManager.abort for specification."""
     if self._transmitting:
     if self._transmitting:
       self._aborted, self._abortion_outcome = True, outcome
       self._aborted, self._abortion_outcome = True, outcome
@@ -287,8 +291,12 @@ class TransmissionManager(_interfaces.TransmissionManager):
         termination = _constants.ABORTION_OUTCOME_TO_TICKET_TERMINATION[
         termination = _constants.ABORTION_OUTCOME_TO_TICKET_TERMINATION[
             outcome]
             outcome]
         if termination is not None:
         if termination is not None:
+          if self._completion is None:
+            code, message = None, None
+          else:
+            code, message = self._completion.code, self._completion.message
           ticket = links.Ticket(
           ticket = links.Ticket(
               self._operation_id, self._lowest_unused_sequence_number, None,
               self._operation_id, self._lowest_unused_sequence_number, None,
-              None, None, None, None, None, None, None, None, None,
+              None, None, None, None, None, None, None, code, message,
               termination, None)
               termination, None)
           self._transmit(ticket)
           self._transmit(ticket)

+ 20 - 1
src/python/grpcio/grpc/framework/interfaces/base/base.py

@@ -47,7 +47,26 @@ from grpc.framework.foundation import abandonment  # pylint: disable=unused-impo
 
 
 
 
 class NoSuchMethodError(Exception):
 class NoSuchMethodError(Exception):
-  """Indicates that an unrecognized operation has been called."""
+  """Indicates that an unrecognized operation has been called.
+
+  Attributes:
+    code: A code value to communicate to the other side of the operation along
+      with indication of operation termination. May be None.
+    details: A details value to communicate to the other side of the operation
+      along with indication of operation termination. May be None.
+  """
+
+  def __init__(self, code, details):
+    """Constructor.
+
+    Args:
+      code: A code value to communicate to the other side of the operation
+        along with indication of operation termination. May be None.
+      details: A details value to communicate to the other side of the
+        operation along with indication of operation termination. May be None.
+    """
+    self.code = code
+    self.details = details
 
 
 
 
 @enum.unique
 @enum.unique

+ 1 - 1
src/python/grpcio_test/grpc_test/framework/interfaces/base/test_cases.py

@@ -134,7 +134,7 @@ class _Servicer(base.Servicer):
       if group != self._group or method != self._method:
       if group != self._group or method != self._method:
         controller.fail(
         controller.fail(
             '%s != %s or %s != %s' % (group, self._group, method, self._method))
             '%s != %s or %s != %s' % (group, self._group, method, self._method))
-        raise base.NoSuchMethodError()
+        raise base.NoSuchMethodError(None, None)
       else:
       else:
         operator = _Operator(
         operator = _Operator(
             controller, controller.on_service_advance, self._pool,
             controller, controller.on_service_advance, self._pool,

+ 0 - 1
tools/run_tests/jobset.py

@@ -174,7 +174,6 @@ class Job(object):
     for k, v in add_env.iteritems():
     for k, v in add_env.iteritems():
       env[k] = v
       env[k] = v
     self._start = time.time()
     self._start = time.time()
-    print spec.cmdline
     self._process = subprocess.Popen(args=spec.cmdline,
     self._process = subprocess.Popen(args=spec.cmdline,
                                      stderr=subprocess.STDOUT,
                                      stderr=subprocess.STDOUT,
                                      stdout=self._tempfile,
                                      stdout=self._tempfile,

+ 5 - 3
tools/run_tests/run_tests.py

@@ -70,13 +70,14 @@ def platform_string():
 # SimpleConfig: just compile with CONFIG=config, and run the binary to test
 # SimpleConfig: just compile with CONFIG=config, and run the binary to test
 class SimpleConfig(object):
 class SimpleConfig(object):
 
 
-  def __init__(self, config, environ=None):
+  def __init__(self, config, environ=None, timeout_seconds=5*60):
     if environ is None:
     if environ is None:
       environ = {}
       environ = {}
     self.build_config = config
     self.build_config = config
     self.allow_hashing = (config != 'gcov')
     self.allow_hashing = (config != 'gcov')
     self.environ = environ
     self.environ = environ
     self.environ['CONFIG'] = config
     self.environ['CONFIG'] = config
+    self.timeout_seconds = timeout_seconds
 
 
   def job_spec(self, cmdline, hash_targets, shortname=None, environ={}):
   def job_spec(self, cmdline, hash_targets, shortname=None, environ={}):
     """Construct a jobset.JobSpec for a test under this config
     """Construct a jobset.JobSpec for a test under this config
@@ -96,6 +97,7 @@ class SimpleConfig(object):
     return jobset.JobSpec(cmdline=cmdline,
     return jobset.JobSpec(cmdline=cmdline,
                           shortname=shortname,
                           shortname=shortname,
                           environ=actual_environ,
                           environ=actual_environ,
+                          timeout_seconds=self.timeout_seconds,
                           hash_targets=hash_targets
                           hash_targets=hash_targets
                               if self.allow_hashing else None)
                               if self.allow_hashing else None)
 
 
@@ -354,11 +356,11 @@ class Build(object):
 _CONFIGS = {
 _CONFIGS = {
     'dbg': SimpleConfig('dbg'),
     'dbg': SimpleConfig('dbg'),
     'opt': SimpleConfig('opt'),
     'opt': SimpleConfig('opt'),
-    'tsan': SimpleConfig('tsan', environ={
+    'tsan': SimpleConfig('tsan', timeout_seconds=10*60, environ={
         'TSAN_OPTIONS': 'suppressions=tools/tsan_suppressions.txt:halt_on_error=1:second_deadlock_stack=1'}),
         'TSAN_OPTIONS': 'suppressions=tools/tsan_suppressions.txt:halt_on_error=1:second_deadlock_stack=1'}),
     'msan': SimpleConfig('msan'),
     'msan': SimpleConfig('msan'),
     'ubsan': SimpleConfig('ubsan'),
     'ubsan': SimpleConfig('ubsan'),
-    'asan': SimpleConfig('asan', environ={
+    'asan': SimpleConfig('asan', timeout_seconds=7*60, environ={
         'ASAN_OPTIONS': 'detect_leaks=1:color=always:suppressions=tools/tsan_suppressions.txt',
         'ASAN_OPTIONS': 'detect_leaks=1:color=always:suppressions=tools/tsan_suppressions.txt',
         'LSAN_OPTIONS': 'report_objects=1'}),
         'LSAN_OPTIONS': 'report_objects=1'}),
     'asan-noleaks': SimpleConfig('asan', environ={
     'asan-noleaks': SimpleConfig('asan', environ={