Merge branch 'master' of https://github.com/grpc/grpc into ct-you-complete-me

3 files changed, 181 insertions, 168 deletions

diff --git a/test/compiler/python_plugin_test.py b/test/compiler/python_plugin_test.py
index 367effdb1a..0e58d912b9 100644
--- a/test/compiler/python_plugin_test.py
+++ b/test/compiler/python_plugin_test.py
@@ -36,6 +36,7 @@ import shutil
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import unittest
 
@@ -49,13 +50,13 @@ STUB_IDENTIFIER = 'EarlyAdopterTestServiceStub'
 SERVER_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_server'
 STUB_FACTORY_IDENTIFIER = 'early_adopter_create_TestService_stub'
 
-# Timeouts and delays.
-SHORT_TIMEOUT = 0.1
-NORMAL_TIMEOUT = 1
-LONG_TIMEOUT = 2
-DOES_NOT_MATTER_DELAY = 0
+# The timeout used in tests of RPCs that are supposed to expire.
+SHORT_TIMEOUT = 2
+# The timeout used in tests of RPCs that are not supposed to expire. The
+# absurdly large value doesn't matter since no passing execution of this test
+# module will ever wait the duration.
+LONG_TIMEOUT = 600
 NO_DELAY = 0
-LONG_DELAY = 1
 
 # Build mode environment variable set by tools/run_tests/run_tests.py.
 _build_mode = os.environ['CONFIG']
@@ -64,47 +65,54 @@ _build_mode = os.environ['CONFIG']
 class _ServicerMethods(object):
 
   def __init__(self, test_pb2, delay):
+    self._condition = threading.Condition()
+    self._delay = delay
     self._paused = False
-    self._failed = False
-    self.test_pb2 = test_pb2
-    self.delay = delay
+    self._fail = False
+    self._test_pb2 = test_pb2
 
   @contextlib.contextmanager
   def pause(self):  # pylint: disable=invalid-name
-    self._paused = True
+    with self._condition:
+      self._paused = True
     yield
-    self._paused = False
+    with self._condition:
+      self._paused = False
+      self._condition.notify_all()
 
   @contextlib.contextmanager
   def fail(self):  # pylint: disable=invalid-name
-    self._failed = True
+    with self._condition:
+      self._fail = True
     yield
-    self._failed = False
+    with self._condition:
+      self._fail = False
 
   def _control(self):  # pylint: disable=invalid-name
-    if self._failed:
-      raise ValueError()
-    time.sleep(self.delay)
-    while self._paused:
-      time.sleep(0)
-
-  def UnaryCall(self, request, unused_context):
-    response = self.test_pb2.SimpleResponse()
-    response.payload.payload_type = self.test_pb2.COMPRESSABLE
+    with self._condition:
+      if self._fail:
+        raise ValueError()
+      while self._paused:
+        self._condition.wait()
+    time.sleep(self._delay)
+
+  def UnaryCall(self, request, unused_rpc_context):
+    response = self._test_pb2.SimpleResponse()
+    response.payload.payload_type = self._test_pb2.COMPRESSABLE
     response.payload.payload_compressable = 'a' * request.response_size
     self._control()
     return response
 
-  def StreamingOutputCall(self, request, unused_context):
+  def StreamingOutputCall(self, request, unused_rpc_context):
     for parameter in request.response_parameters:
-      response = self.test_pb2.StreamingOutputCallResponse()
-      response.payload.payload_type = self.test_pb2.COMPRESSABLE
+      response = self._test_pb2.StreamingOutputCallResponse()
+      response.payload.payload_type = self._test_pb2.COMPRESSABLE
       response.payload.payload_compressable = 'a' * parameter.size
       self._control()
       yield response
 
-  def StreamingInputCall(self, request_iter, unused_context):
-    response = self.test_pb2.StreamingInputCallResponse()
+  def StreamingInputCall(self, request_iter, unused_rpc_context):
+    response = self._test_pb2.StreamingInputCallResponse()
     aggregated_payload_size = 0
     for request in request_iter:
       aggregated_payload_size += len(request.payload.payload_compressable)
@@ -112,21 +120,21 @@ class _ServicerMethods(object):
     self._control()
     return response
 
-  def FullDuplexCall(self, request_iter, unused_context):
+  def FullDuplexCall(self, request_iter, unused_rpc_context):
     for request in request_iter:
       for parameter in request.response_parameters:
-        response = self.test_pb2.StreamingOutputCallResponse()
-        response.payload.payload_type = self.test_pb2.COMPRESSABLE
+        response = self._test_pb2.StreamingOutputCallResponse()
+        response.payload.payload_type = self._test_pb2.COMPRESSABLE
         response.payload.payload_compressable = 'a' * parameter.size
         self._control()
         yield response
 
-  def HalfDuplexCall(self, request_iter, unused_context):
+  def HalfDuplexCall(self, request_iter, unused_rpc_context):
     responses = []
     for request in request_iter:
       for parameter in request.response_parameters:
-        response = self.test_pb2.StreamingOutputCallResponse()
-        response.payload.payload_type = self.test_pb2.COMPRESSABLE
+        response = self._test_pb2.StreamingOutputCallResponse()
+        response.payload.payload_type = self._test_pb2.COMPRESSABLE
         response.payload.payload_compressable = 'a' * parameter.size
         self._control()
         responses.append(response)
@@ -147,12 +155,11 @@ def _CreateService(test_pb2, delay):
   waiting for the service.
 
   Args:
-    test_pb2: the test_pb2 module generated by this test
-    delay: delay in seconds per response from the servicer
-    timeout: how long the stub will wait for the servicer by default.
+    test_pb2: The test_pb2 module generated by this test.
+    delay: Delay in seconds per response from the servicer.
 
   Yields:
-    A three-tuple (servicer_methods, servicer, stub), where the servicer is
+    A (servicer_methods, servicer, stub) three-tuple where servicer_methods is
       the back-end of the service bound to the stub and the server and stub
       are both activated and ready for use.
   """
@@ -185,7 +192,7 @@ def _CreateService(test_pb2, delay):
       yield servicer_methods, stub, server
 
 
-def StreamingInputRequest(test_pb2):
+def _streaming_input_request_iterator(test_pb2):
   for _ in range(3):
     request = test_pb2.StreamingInputCallRequest()
     request.payload.payload_type = test_pb2.COMPRESSABLE
@@ -193,7 +200,7 @@ def StreamingInputRequest(test_pb2):
     yield request
 
 
-def StreamingOutputRequest(test_pb2):
+def _streaming_output_request(test_pb2):
   request = test_pb2.StreamingOutputCallRequest()
   sizes = [1, 2, 3]
   request.response_parameters.add(size=sizes[0], interval_us=0)
@@ -202,7 +209,7 @@ def StreamingOutputRequest(test_pb2):
   return request
 
 
-def FullDuplexRequest(test_pb2):
+def _full_duplex_request_iterator(test_pb2):
   request = test_pb2.StreamingOutputCallRequest()
   request.response_parameters.add(size=1, interval_us=0)
   yield request
@@ -250,7 +257,7 @@ class PythonPluginTest(unittest.TestCase):
       if exc.errno != errno.ENOENT:
         raise
 
-  # TODO(atash): Figure out which of theses tests is hanging flakily with small
+  # TODO(atash): Figure out which of these tests is hanging flakily with small
   # probability.
 
   def testImportAttributes(self):
@@ -265,37 +272,36 @@ class PythonPluginTest(unittest.TestCase):
   def testUpDown(self):
     import test_pb2
     with _CreateService(
-        test_pb2, DOES_NOT_MATTER_DELAY) as (servicer, stub, unused_server):
+        test_pb2, NO_DELAY) as (servicer, stub, unused_server):
       request = test_pb2.SimpleRequest(response_size=13)
 
   def testUnaryCall(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server):
+    with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server):
+      timeout = 6  # TODO(issue 2039): LONG_TIMEOUT like the other methods.
       request = test_pb2.SimpleRequest(response_size=13)
-      response = stub.UnaryCall(request, NORMAL_TIMEOUT)
-    expected_response = servicer.UnaryCall(request, None)
+      response = stub.UnaryCall(request, timeout)
+    expected_response = methods.UnaryCall(request, 'not a real RpcContext!')
     self.assertEqual(expected_response, response)
 
   def testUnaryCallAsync(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
     request = test_pb2.SimpleRequest(response_size=13)
-    with _CreateService(test_pb2, LONG_DELAY) as (
-        servicer, stub, unused_server):
-      start_time = time.clock()
-      response_future = stub.UnaryCall.async(request, LONG_TIMEOUT)
-      # Check that we didn't block on the asynchronous call.
-      self.assertGreater(LONG_DELAY, time.clock() - start_time)
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      # Check that the call does not block waiting for the server to respond.
+      with methods.pause():
+        response_future = stub.UnaryCall.async(request, LONG_TIMEOUT)
       response = response_future.result()
-    expected_response = servicer.UnaryCall(request, None)
+    expected_response = methods.UnaryCall(request, 'not a real RpcContext!')
     self.assertEqual(expected_response, response)
 
   def testUnaryCallAsyncExpired(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    # set the timeout super low...
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
       request = test_pb2.SimpleRequest(response_size=13)
-      with servicer.pause():
+      with methods.pause():
         response_future = stub.UnaryCall.async(request, SHORT_TIMEOUT)
         with self.assertRaises(exceptions.ExpirationError):
           response_future.result()
@@ -305,9 +311,9 @@ class PythonPluginTest(unittest.TestCase):
   def testUnaryCallAsyncCancelled(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
     request = test_pb2.SimpleRequest(response_size=13)
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.pause():
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.pause():
         response_future = stub.UnaryCall.async(request, 1)
         response_future.cancel()
         self.assertTrue(response_future.cancelled())
@@ -315,30 +321,31 @@ class PythonPluginTest(unittest.TestCase):
   def testUnaryCallAsyncFailed(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
     request = test_pb2.SimpleRequest(response_size=13)
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.fail():
-        response_future = stub.UnaryCall.async(request, NORMAL_TIMEOUT)
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.fail():
+        response_future = stub.UnaryCall.async(request, LONG_TIMEOUT)
         self.assertIsNotNone(response_future.exception())
 
   def testStreamingOutputCall(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    request = StreamingOutputRequest(test_pb2)
-    with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server):
-      responses = stub.StreamingOutputCall(request, NORMAL_TIMEOUT)
-      expected_responses = servicer.StreamingOutputCall(request, None)
-      for check in itertools.izip_longest(expected_responses, responses):
-        expected_response, response = check
+    request = _streaming_output_request(test_pb2)
+    with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server):
+      responses = stub.StreamingOutputCall(request, LONG_TIMEOUT)
+      expected_responses = methods.StreamingOutputCall(
+          request, 'not a real RpcContext!')
+      for expected_response, response in itertools.izip_longest(
+          expected_responses, responses):
         self.assertEqual(expected_response, response)
 
   @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
                  'forever and fix.')
   def testStreamingOutputCallExpired(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    request = StreamingOutputRequest(test_pb2)
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.pause():
+    request = _streaming_output_request(test_pb2)
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.pause():
         responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT)
         with self.assertRaises(exceptions.ExpirationError):
           list(responses)
@@ -347,9 +354,9 @@ class PythonPluginTest(unittest.TestCase):
                  'forever and fix.')
   def testStreamingOutputCallCancelled(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    request = StreamingOutputRequest(test_pb2)
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        unused_servicer, stub, unused_server):
+    request = _streaming_output_request(test_pb2)
+    with _CreateService(test_pb2, NO_DELAY) as (
+        unused_methods, stub, unused_server):
       responses = stub.StreamingOutputCall(request, SHORT_TIMEOUT)
       next(responses)
       responses.cancel()
@@ -360,10 +367,10 @@ class PythonPluginTest(unittest.TestCase):
                  'instead of raising the proper error.')
   def testStreamingOutputCallFailed(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    request = StreamingOutputRequest(test_pb2)
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.fail():
+    request = _streaming_output_request(test_pb2)
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.fail():
         responses = stub.StreamingOutputCall(request, 1)
         self.assertIsNotNone(responses)
         with self.assertRaises(exceptions.ServicerError):
@@ -373,34 +380,32 @@ class PythonPluginTest(unittest.TestCase):
                  'forever and fix.')
   def testStreamingInputCall(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server):
-      response = stub.StreamingInputCall(StreamingInputRequest(test_pb2),
-                                         NORMAL_TIMEOUT)
-    expected_response = servicer.StreamingInputCall(
-        StreamingInputRequest(test_pb2), None)
+    with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server):
+      response = stub.StreamingInputCall(
+          _streaming_input_request_iterator(test_pb2), LONG_TIMEOUT)
+    expected_response = methods.StreamingInputCall(
+        _streaming_input_request_iterator(test_pb2), 'not a real RpcContext!')
     self.assertEqual(expected_response, response)
 
   def testStreamingInputCallAsync(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    with _CreateService(test_pb2, LONG_DELAY) as (
-        servicer, stub, unused_server):
-      start_time = time.clock()
-      response_future = stub.StreamingInputCall.async(
-          StreamingInputRequest(test_pb2), LONG_TIMEOUT)
-      self.assertGreater(LONG_DELAY, time.clock() - start_time)
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.pause():
+        response_future = stub.StreamingInputCall.async(
+            _streaming_input_request_iterator(test_pb2), LONG_TIMEOUT)
       response = response_future.result()
-    expected_response = servicer.StreamingInputCall(
-        StreamingInputRequest(test_pb2), None)
+    expected_response = methods.StreamingInputCall(
+        _streaming_input_request_iterator(test_pb2), 'not a real RpcContext!')
     self.assertEqual(expected_response, response)
 
   def testStreamingInputCallAsyncExpired(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    # set the timeout super low...
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.pause():
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.pause():
         response_future = stub.StreamingInputCall.async(
-            StreamingInputRequest(test_pb2), SHORT_TIMEOUT)
+            _streaming_input_request_iterator(test_pb2), SHORT_TIMEOUT)
         with self.assertRaises(exceptions.ExpirationError):
           response_future.result()
         self.assertIsInstance(
@@ -408,11 +413,12 @@ class PythonPluginTest(unittest.TestCase):
 
   def testStreamingInputCallAsyncCancelled(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.pause():
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.pause():
+        timeout = 6  # TODO(issue 2039): LONG_TIMEOUT like the other methods.
         response_future = stub.StreamingInputCall.async(
-            StreamingInputRequest(test_pb2), NORMAL_TIMEOUT)
+            _streaming_input_request_iterator(test_pb2), timeout)
         response_future.cancel()
         self.assertTrue(response_future.cancelled())
       with self.assertRaises(future.CancelledError):
@@ -420,33 +426,33 @@ class PythonPluginTest(unittest.TestCase):
 
   def testStreamingInputCallAsyncFailed(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.fail():
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.fail():
         response_future = stub.StreamingInputCall.async(
-            StreamingInputRequest(test_pb2), SHORT_TIMEOUT)
+            _streaming_input_request_iterator(test_pb2), SHORT_TIMEOUT)
         self.assertIsNotNone(response_future.exception())
 
   def testFullDuplexCall(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server):
-      responses = stub.FullDuplexCall(FullDuplexRequest(test_pb2),
-                                      NORMAL_TIMEOUT)
-      expected_responses = servicer.FullDuplexCall(FullDuplexRequest(test_pb2),
-                                                   None)
-      for check in itertools.izip_longest(expected_responses, responses):
-        expected_response, response = check
+    with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server):
+      responses = stub.FullDuplexCall(
+          _full_duplex_request_iterator(test_pb2), LONG_TIMEOUT)
+      expected_responses = methods.FullDuplexCall(
+          _full_duplex_request_iterator(test_pb2), 'not a real RpcContext!')
+      for expected_response, response in itertools.izip_longest(
+          expected_responses, responses):
         self.assertEqual(expected_response, response)
 
   @unittest.skip('TODO(atash,nathaniel): figure out why this flakily hangs '
                  'forever and fix.')
   def testFullDuplexCallExpired(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    request = FullDuplexRequest(test_pb2)
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.pause():
-        responses = stub.FullDuplexCall(request, SHORT_TIMEOUT)
+    request_iterator = _full_duplex_request_iterator(test_pb2)
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.pause():
+        responses = stub.FullDuplexCall(request_iterator, SHORT_TIMEOUT)
         with self.assertRaises(exceptions.ExpirationError):
           list(responses)
 
@@ -454,9 +460,9 @@ class PythonPluginTest(unittest.TestCase):
                  'forever and fix.')
   def testFullDuplexCallCancelled(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server):
-      request = FullDuplexRequest(test_pb2)
-      responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT)
+    with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server):
+      request_iterator = _full_duplex_request_iterator(test_pb2)
+      responses = stub.FullDuplexCall(request_iterator, LONG_TIMEOUT)
       next(responses)
       responses.cancel()
       with self.assertRaises(future.CancelledError):
@@ -466,11 +472,11 @@ class PythonPluginTest(unittest.TestCase):
                  'and fix.')
   def testFullDuplexCallFailed(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    request = FullDuplexRequest(test_pb2)
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      with servicer.fail():
-        responses = stub.FullDuplexCall(request, NORMAL_TIMEOUT)
+    request_iterator = _full_duplex_request_iterator(test_pb2)
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      with methods.fail():
+        responses = stub.FullDuplexCall(request_iterator, LONG_TIMEOUT)
         self.assertIsNotNone(responses)
         with self.assertRaises(exceptions.ServicerError):
           next(responses)
@@ -479,9 +485,9 @@ class PythonPluginTest(unittest.TestCase):
                  'forever and fix.')
   def testHalfDuplexCall(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    with _CreateService(test_pb2, DOES_NOT_MATTER_DELAY) as (
-        servicer, stub, unused_server):
-      def HalfDuplexRequest():
+    with _CreateService(test_pb2, NO_DELAY) as (
+        methods, stub, unused_server):
+      def half_duplex_request_iterator():
         request = test_pb2.StreamingOutputCallRequest()
         request.response_parameters.add(size=1, interval_us=0)
         yield request
@@ -489,30 +495,38 @@ class PythonPluginTest(unittest.TestCase):
         request.response_parameters.add(size=2, interval_us=0)
         request.response_parameters.add(size=3, interval_us=0)
         yield request
-      responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT)
-      expected_responses = servicer.HalfDuplexCall(HalfDuplexRequest(), None)
+      responses = stub.HalfDuplexCall(
+          half_duplex_request_iterator(), LONG_TIMEOUT)
+      expected_responses = methods.HalfDuplexCall(
+          half_duplex_request_iterator(), 'not a real RpcContext!')
       for check in itertools.izip_longest(expected_responses, responses):
         expected_response, response = check
         self.assertEqual(expected_response, response)
 
   def testHalfDuplexCallWedged(self):
     import test_pb2  # pylint: disable=g-import-not-at-top
-    wait_flag = [False]
+    condition = threading.Condition()
+    wait_cell = [False]
     @contextlib.contextmanager
     def wait():  # pylint: disable=invalid-name
       # Where's Python 3's 'nonlocal' statement when you need it?
-      wait_flag[0] = True
+      with condition:
+        wait_cell[0] = True
       yield
-      wait_flag[0] = False
-    def HalfDuplexRequest():
+      with condition:
+        wait_cell[0] = False
+        condition.notify_all()
+    def half_duplex_request_iterator():
       request = test_pb2.StreamingOutputCallRequest()
       request.response_parameters.add(size=1, interval_us=0)
       yield request
-      while wait_flag[0]:
-        time.sleep(0.1)
-    with _CreateService(test_pb2, NO_DELAY) as (servicer, stub, unused_server):
+      with condition:
+        while wait_cell[0]:
+          condition.wait()
+    with _CreateService(test_pb2, NO_DELAY) as (methods, stub, unused_server):
       with wait():
-        responses = stub.HalfDuplexCall(HalfDuplexRequest(), NORMAL_TIMEOUT)
+        responses = stub.HalfDuplexCall(
+            half_duplex_request_iterator(), SHORT_TIMEOUT)
         # half-duplex waits for the client to send all info
         with self.assertRaises(exceptions.ExpirationError):
           next(responses)
diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc
index 921836e201..1b7a8d26b2 100644
--- a/test/cpp/qps/client_async.cc
+++ b/test/cpp/qps/client_async.cc
@@ -62,7 +62,7 @@ typedef std::list<grpc_time> deadline_list;
 
 class ClientRpcContext {
  public:
-  ClientRpcContext(int ch) : channel_id_(ch) {}
+  explicit ClientRpcContext(int ch) : channel_id_(ch) {}
   virtual ~ClientRpcContext() {}
   // next state, return false if done. Collect stats when appropriate
   virtual bool RunNextState(bool, Histogram* hist) = 0;
diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc
index 4b0678bb2c..210aef4fd6 100644
--- a/test/cpp/qps/server_async.cc
+++ b/test/cpp/qps/server_async.cc
@@ -73,31 +73,35 @@ class AsyncQpsServerTest : public Server {
     gpr_free(server_address);
 
     builder.RegisterAsyncService(&async_service_);
-    srv_cq_ = builder.AddCompletionQueue();
+    for (int i = 0; i < config.threads(); i++) {
+      srv_cqs_.emplace_back(std::move(builder.AddCompletionQueue()));
+    }
 
     server_ = builder.BuildAndStart();
 
     using namespace std::placeholders;
-    request_unary_ =
-        std::bind(&TestService::AsyncService::RequestUnaryCall, &async_service_,
-                  _1, _2, _3, srv_cq_.get(), srv_cq_.get(), _4);
-    request_streaming_ =
-        std::bind(&TestService::AsyncService::RequestStreamingCall,
-                  &async_service_, _1, _2, srv_cq_.get(), srv_cq_.get(), _3);
-    for (int i = 0; i < 100; i++) {
-      contexts_.push_front(
-          new ServerRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
-              request_unary_, ProcessRPC));
-      contexts_.push_front(
-          new ServerRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
-              request_streaming_, ProcessRPC));
+    for (int i = 0; i < 10; i++) {
+      for (int j = 0; j < config.threads(); j++) {
+        auto request_unary = std::bind(
+            &TestService::AsyncService::RequestUnaryCall, &async_service_, _1,
+            _2, _3, srv_cqs_[j].get(), srv_cqs_[j].get(), _4);
+        auto request_streaming = std::bind(
+            &TestService::AsyncService::RequestStreamingCall, &async_service_,
+            _1, _2, srv_cqs_[j].get(), srv_cqs_[j].get(), _3);
+        contexts_.push_front(
+            new ServerRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
+                request_unary, ProcessRPC));
+        contexts_.push_front(
+            new ServerRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
+                request_streaming, ProcessRPC));
+      }
     }
     for (int i = 0; i < config.threads(); i++) {
       threads_.push_back(std::thread([=]() {
         // Wait until work is available or we are shutting down
         bool ok;
         void *got_tag;
-        while (srv_cq_->Next(&got_tag, &ok)) {
+        while (srv_cqs_[i]->Next(&got_tag, &ok)) {
           ServerRpcContext *ctx = detag(got_tag);
           // The tag is a pointer to an RPC context to invoke
           bool still_going = ctx->RunNextState(ok);
@@ -125,11 +129,13 @@ class AsyncQpsServerTest : public Server {
     for (auto thr = threads_.begin(); thr != threads_.end(); thr++) {
       thr->join();
     }
-    srv_cq_->Shutdown();
-    bool ok;
-    void *got_tag;
-    while (srv_cq_->Next(&got_tag, &ok))
-      ;
+    for (auto cq = srv_cqs_.begin(); cq != srv_cqs_.end(); ++cq) {
+      (*cq)->Shutdown();
+      bool ok;
+      void *got_tag;
+      while ((*cq)->Next(&got_tag, &ok))
+        ;
+    }
     while (!contexts_.empty()) {
       delete contexts_.front();
       contexts_.pop_front();
@@ -306,15 +312,8 @@ class AsyncQpsServerTest : public Server {
   }
   std::vector<std::thread> threads_;
   std::unique_ptr<grpc::Server> server_;
-  std::unique_ptr<grpc::ServerCompletionQueue> srv_cq_;
+  std::vector<std::unique_ptr<grpc::ServerCompletionQueue>> srv_cqs_;
   TestService::AsyncService async_service_;
-  std::function<void(ServerContext *, SimpleRequest *,
-                     grpc::ServerAsyncResponseWriter<SimpleResponse> *, void *)>
-      request_unary_;
-  std::function<void(
-      ServerContext *,
-      grpc::ServerAsyncReaderWriter<SimpleResponse, SimpleRequest> *, void *)>
-      request_streaming_;
   std::forward_list<ServerRpcContext *> contexts_;
 
   std::mutex shutdown_mutex_;