Merge github.com:grpc/grpc into sceq

6 files changed, 113 insertions, 18 deletions

diff --git a/src/python/grpcio/commands.py b/src/python/grpcio/commands.py
index e1a3f4bed3..bd12c5579c 100644
--- a/src/python/grpcio/commands.py
+++ b/src/python/grpcio/commands.py
@@ -39,17 +39,7 @@ import sys
 import setuptools
 from setuptools.command import build_py
 from setuptools.command import test
-
-# Because we need to support building without Cython but simultaneously need to
-# subclass its command class when we need to and because distutils requires a
-# special hook to acquire a command class, we attempt to import Cython's
-# build_ext, and if that fails we import setuptools'.
-try:
-  # Due to the strange way Cython's Distutils module re-imports build_ext, we
-  # import the build_ext class directly.
-  from Cython.Distutils.build_ext import build_ext
-except ImportError:
-  from setuptools.command.build_ext import build_ext
+from setuptools.command import build_ext
 
 PYTHON_STEM = os.path.dirname(os.path.abspath(__file__))
 
@@ -179,7 +169,7 @@ class BuildPy(build_py.build_py):
     build_py.build_py.run(self)
 
 
-class BuildExt(build_ext):
+class BuildExt(build_ext.build_ext):
   """Custom build_ext command to enable compiler-specific flags."""
 
   C_OPTIONS = {
@@ -196,7 +186,7 @@ class BuildExt(build_ext):
     if compiler in BuildExt.LINK_OPTIONS:
       for extension in self.extensions:
         extension.extra_link_args += list(BuildExt.LINK_OPTIONS[compiler])
-    build_ext.build_extensions(self)
+    build_ext.build_ext.build_extensions(self)
 
 
 class Gather(setuptools.Command):
diff --git a/src/python/grpcio/grpc/framework/core/_termination.py b/src/python/grpcio/grpc/framework/core/_termination.py
index bdb9147e5b..364158b2b8 100644
--- a/src/python/grpcio/grpc/framework/core/_termination.py
+++ b/src/python/grpcio/grpc/framework/core/_termination.py
@@ -1,4 +1,4 @@
-# Copyright 2015, Google Inc.
+# Copyright 2015-2016, Google Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -46,8 +46,8 @@ def _invocation_completion_predicate(
 
 def _service_completion_predicate(
     unused_emission_complete, transmission_complete, unused_reception_complete,
-    unused_ingestion_complete):
-  return transmission_complete
+    ingestion_complete):
+  return transmission_complete and ingestion_complete
 
 
 class TerminationManager(_interfaces.TerminationManager):
diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py
index 8493bbd901..c921c69b4c 100644
--- a/src/python/grpcio/grpc_core_dependencies.py
+++ b/src/python/grpcio/grpc_core_dependencies.py
@@ -222,6 +222,7 @@ CORE_SOURCE_FILES = [
   'src/core/census/context.c',
   'src/core/census/initialize.c',
   'src/core/census/operation.c',
+  'src/core/census/tag_set.c',
   'src/core/census/tracing.c',
   'src/boringssl/err_data.c',
   'third_party/boringssl/crypto/aes/aes.c',
diff --git a/src/python/grpcio/tests/unit/_cython/_channel_test.py b/src/python/grpcio/tests/unit/_cython/_channel_test.py
new file mode 100644
index 0000000000..b414f8e6f6
--- /dev/null
+++ b/src/python/grpcio/tests/unit/_cython/_channel_test.py
@@ -0,0 +1,83 @@
+# Copyright 2016, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import time
+import threading
+import unittest
+
+from grpc._cython import cygrpc
+
+# TODO(nathaniel): This should be at least one hundred. Why not one thousand?
+_PARALLELISM = 4
+
+
+def _channel_and_completion_queue():
+  channel = cygrpc.Channel('localhost:54321', cygrpc.ChannelArgs(()))
+  completion_queue = cygrpc.CompletionQueue()
+  return channel, completion_queue
+
+
+def _connectivity_loop(channel, completion_queue):
+  for _ in range(100):
+    connectivity = channel.check_connectivity_state(True)
+    channel.watch_connectivity_state(
+        connectivity, cygrpc.Timespec(time.time() + 0.2), completion_queue,
+        None)
+    completion_queue.poll(deadline=cygrpc.Timespec(float('+inf')))
+
+
+def _create_loop_destroy():
+  channel, completion_queue = _channel_and_completion_queue()
+  _connectivity_loop(channel, completion_queue)
+  completion_queue.shutdown()
+
+
+def _in_parallel(behavior, arguments):
+  threads = tuple(
+      threading.Thread(target=behavior, args=arguments)
+      for _ in range(_PARALLELISM))
+  for thread in threads:
+    thread.start()
+  for thread in threads:
+    thread.join()
+
+
+class ChannelTest(unittest.TestCase):
+
+  def test_single_channel_lonely_connectivity(self):
+    channel, completion_queue = _channel_and_completion_queue()
+    _in_parallel(_connectivity_loop, (channel, completion_queue,))
+    completion_queue.shutdown()
+
+  def test_multiple_channels_lonely_connectivity(self):
+    _in_parallel(_create_loop_destroy, ())
+
+
+if __name__ == '__main__':
+  unittest.main(verbosity=2)
diff --git a/src/python/grpcio/tests/unit/framework/common/test_constants.py b/src/python/grpcio/tests/unit/framework/common/test_constants.py
index e1d3c2709d..9f1fb8471c 100644
--- a/src/python/grpcio/tests/unit/framework/common/test_constants.py
+++ b/src/python/grpcio/tests/unit/framework/common/test_constants.py
@@ -1,4 +1,4 @@
-# Copyright 2015, Google Inc.
+# Copyright 2015-2016, Google Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -49,5 +49,8 @@ STREAM_LENGTH = 200
 # The size of payloads to transmit in tests.
 PAYLOAD_SIZE = 256 * 1024 + 17
 
+# The parallelism to use in tests of parallel RPCs.
+PARALLELISM = 200
+
 # The size of thread pools to use in tests.
 POOL_SIZE = 10
diff --git a/src/python/grpcio/tests/unit/framework/interfaces/face/_future_invocation_asynchronous_event_service.py b/src/python/grpcio/tests/unit/framework/interfaces/face/_future_invocation_asynchronous_event_service.py
index c178f2f108..fc8daa992f 100644
--- a/src/python/grpcio/tests/unit/framework/interfaces/face/_future_invocation_asynchronous_event_service.py
+++ b/src/python/grpcio/tests/unit/framework/interfaces/face/_future_invocation_asynchronous_event_service.py
@@ -1,4 +1,4 @@
-# Copyright 2015, Google Inc.
+# Copyright 2015-2016, Google Inc.
 # All rights reserved.
 #
 # Redistribution and use in source and binary forms, with or without
@@ -219,6 +219,24 @@ class TestCase(test_coverage.Coverage, unittest.TestCase):
 
         test_messages.verify(second_request, second_response, self)
 
+    for (group, method), test_messages_sequence in (
+        self._digest.unary_unary_messages_sequences.iteritems()):
+      for test_messages in test_messages_sequence:
+        requests = []
+        response_futures = []
+        for _ in range(test_constants.PARALLELISM):
+          request = test_messages.request()
+          response_future = self._invoker.future(group, method)(
+              request, test_constants.LONG_TIMEOUT)
+          requests.append(request)
+          response_futures.append(response_future)
+
+        responses = [
+            response_future.result() for response_future in response_futures]
+
+        for request, response in zip(requests, responses):
+          test_messages.verify(request, response, self)
+
   def testParallelInvocations(self):
     for (group, method), test_messages_sequence in (
         self._digest.unary_unary_messages_sequences.iteritems()):