Merge branch 'master' of github.com:grpc/grpc into the-ultimate-showdown

Conflicts:
	src/ruby/ext/grpc/rb_channel.c

src/compiler/objective_c_generator.cc

@@ -154,9 +154,9 @@ void PrintAdvancedImplementation(Printer *printer,
 
   printer->Print("        responsesWriteable:[GRXWriteable ");
   if (method->server_streaming()) {
-    printer->Print("writeableWithStreamHandler:eventHandler]];\n");
+    printer->Print("writeableWithEventHandler:eventHandler]];\n");
   } else {
-    printer->Print("writeableWithSingleValueHandler:handler]];\n");
+    printer->Print("writeableWithSingleHandler:handler]];\n");
   }
 
   printer->Print("}\n");

src/csharp/Grpc.Core/Grpc.Core.csproj

@@ -83,7 +83,6 @@
     <Compile Include="Internal\AsyncCompletion.cs" />
     <Compile Include="Internal\AsyncCallBase.cs" />
     <Compile Include="Internal\AsyncCallServer.cs" />
-    <Compile Include="OperationFailedException.cs" />
     <Compile Include="Internal\AsyncCall.cs" />
     <Compile Include="Utils\Preconditions.cs" />
     <Compile Include="Internal\ServerCredentialsSafeHandle.cs" />

src/csharp/Grpc.Core/Internal/AsyncCallBase.cs

@@ -293,7 +293,7 @@ namespace Grpc.Core.Internal
 
             if (!success)
             {
-                FireCompletion(origCompletionDelegate, null, new OperationFailedException("Send failed"));
+                FireCompletion(origCompletionDelegate, null, new InvalidOperationException("Send failed"));
             }
             else
             {
@@ -318,7 +318,7 @@ namespace Grpc.Core.Internal
 
             if (!success)
             {
-                FireCompletion(origCompletionDelegate, null, new OperationFailedException("Halfclose failed"));
+                FireCompletion(origCompletionDelegate, null, new InvalidOperationException("Halfclose failed"));
             }
             else
             {

src/csharp/Grpc.Core/OperationFailedException.cs

@@ -1,47 +0,0 @@
-#region Copyright notice and license
-
-// Copyright 2015, 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.
-
-#endregion
-
-using System;
-
-namespace Grpc.Core
-{
-    /// <summary>
-    /// Thrown when gRPC operation fails.
-    /// </summary>
-    public class OperationFailedException : Exception
-    {
-        public OperationFailedException(string message) : base(message)
-        {
-        }
-    }
-}

src/node/binding.gyp

@@ -11,7 +11,8 @@
         '-pedantic',
         '-g',
         '-zdefs',
-        '-Werror'
+        '-Werror',
+        '-Wno-error=deprecated-declarations'
       ],
       'ldflags': [
         '-g'

src/node/interop/interop_client.js

@@ -298,7 +298,9 @@ function authTest(expected_user, scope, client, done) {
       assert.strictEqual(resp.payload.type, 'COMPRESSABLE');
       assert.strictEqual(resp.payload.body.length, 314159);
       assert.strictEqual(resp.username, expected_user);
-      assert.strictEqual(resp.oauth_scope, AUTH_SCOPE_RESPONSE);
+      if (scope) {
+        assert.strictEqual(resp.oauth_scope, AUTH_SCOPE_RESPONSE);
+      }
       if (done) {
         done();
       }
@@ -335,7 +337,7 @@ function oauth2Test(expected_user, scope, per_rpc, client, done) {
           if (done) {
             done();
           }
-        });
+        }, client_metadata);
       };
       if (per_rpc) {
         updateMetadata('', {}, makeTestCall);

src/node/src/client.js

@@ -526,7 +526,7 @@ var requester_makers = {
  * requestSerialize: function to serialize request objects
  * responseDeserialize: function to deserialize response objects
  * @param {Object} methods An object mapping method names to method attributes
- * @param {string} serviceName The name of the service
+ * @param {string} serviceName The fully qualified name of the service
  * @return {function(string, Object)} New client constructor
  */
 exports.makeClientConstructor = function(methods, serviceName) {
@@ -551,8 +551,10 @@ exports.makeClientConstructor = function(methods, serviceName) {
     }
     options['grpc.primary_user_agent'] = 'grpc-node/' + version;
     this.channel = new grpc.Channel(address, credentials, options);
-    this.server_address = address.replace(/\/$/, '');
-    this.auth_uri = this.server_address + '/' + serviceName;
+    // Remove the optional DNS scheme, trailing port, and trailing backslash
+    address = address.replace(/^(dns:\/{3})?([^:\/]+)(:\d+)?\/?$/, '$2');
+    this.server_address = address;
+    this.auth_uri = 'https://' + this.server_address + '/' + serviceName;
     this.updateMetadata = updateMetadata;
   }
 
@@ -590,7 +592,8 @@ exports.makeClientConstructor = function(methods, serviceName) {
  */
 exports.makeProtobufClientConstructor =  function(service) {
   var method_attrs = common.getProtobufServiceAttrs(service, service.name);
-  var Client = exports.makeClientConstructor(method_attrs);
+  var Client = exports.makeClientConstructor(
+      method_attrs, common.fullyQualifiedName(service));
   Client.service = service;
   return Client;
 };

src/objective-c/RxLibrary/GRXWriteable.h

@@ -48,15 +48,15 @@
 
 typedef void (^GRXValueHandler)(id value);
 typedef void (^GRXCompletionHandler)(NSError *errorOrNil);
-typedef void (^GRXSingleValueHandler)(id value, NSError *errorOrNil);
-typedef void (^GRXStreamHandler)(BOOL done, id value, NSError *error);
+typedef void (^GRXSingleHandler)(id value, NSError *errorOrNil);
+typedef void (^GRXEventHandler)(BOOL done, id value, NSError *error);
 
 // Utility to create objects that conform to the GRXWriteable protocol, from
 // blocks that handle each of the two methods of the protocol.
 @interface GRXWriteable : NSObject<GRXWriteable>
 
-+ (instancetype)writeableWithSingleValueHandler:(GRXSingleValueHandler)handler;
-+ (instancetype)writeableWithStreamHandler:(GRXStreamHandler)handler;
++ (instancetype)writeableWithSingleHandler:(GRXSingleHandler)handler;
++ (instancetype)writeableWithEventHandler:(GRXEventHandler)handler;
 
 - (instancetype)initWithValueHandler:(GRXValueHandler)valueHandler
                    completionHandler:(GRXCompletionHandler)completionHandler

src/objective-c/RxLibrary/GRXWriteable.m

@@ -38,7 +38,7 @@
   GRXCompletionHandler _completionHandler;
 }
 
-+ (instancetype)writeableWithSingleValueHandler:(GRXSingleValueHandler)handler {
++ (instancetype)writeableWithSingleHandler:(GRXSingleHandler)handler {
   if (!handler) {
     return [[self alloc] init];
   }
@@ -51,7 +51,7 @@
   }];
 }
 
-+ (instancetype)writeableWithStreamHandler:(GRXStreamHandler)handler {
++ (instancetype)writeableWithEventHandler:(GRXEventHandler)handler {
   if (!handler) {
     return [[self alloc] init];
   }

src/objective-c/tests/RxLibraryUnitTests.m

@@ -55,7 +55,7 @@
   return [[self alloc] init];
 }
 
-- (GRXSingleValueHandler)block {
+- (GRXSingleHandler)block {
   return ^(id value, NSError *errorOrNil) {
     ++_timesCalled;
     _value = value;
@@ -71,13 +71,13 @@
 
 #pragma mark Writeable
 
-- (void)testWriteableSingleValueHandlerIsCalledForValue {
+- (void)testWriteableSingleHandlerIsCalledForValue {
   // Given:
   CapturingSingleValueHandler *handler = [CapturingSingleValueHandler handler];
   id anyValue = @7;
 
   // If:
-  id<GRXWriteable> writeable = [GRXWriteable writeableWithSingleValueHandler:handler.block];
+  id<GRXWriteable> writeable = [GRXWriteable writeableWithSingleHandler:handler.block];
   [writeable writeValue:anyValue];
 
   // Then:
@@ -86,13 +86,13 @@
   XCTAssertEqualObjects(handler.errorOrNil, nil);
 }
 
-- (void)testWriteableSingleValueHandlerIsCalledForError {
+- (void)testWriteableSingleHandlerIsCalledForError {
   // Given:
   CapturingSingleValueHandler *handler = [CapturingSingleValueHandler handler];
   NSError *anyError = [NSError errorWithDomain:@"domain" code:7 userInfo:nil];
 
   // If:
-  id<GRXWriteable> writeable = [GRXWriteable writeableWithSingleValueHandler:handler.block];
+  id<GRXWriteable> writeable = [GRXWriteable writeableWithSingleHandler:handler.block];
   [writeable writesFinishedWithError:anyError];
 
   // Then:
@@ -106,7 +106,7 @@
 - (void)testBufferedPipePropagatesValue {
   // Given:
   CapturingSingleValueHandler *handler = [CapturingSingleValueHandler handler];
-  id<GRXWriteable> writeable = [GRXWriteable writeableWithSingleValueHandler:handler.block];
+  id<GRXWriteable> writeable = [GRXWriteable writeableWithSingleHandler:handler.block];
   id anyValue = @7;
 
   // If:
@@ -123,7 +123,7 @@
 - (void)testBufferedPipePropagatesError {
   // Given:
   CapturingSingleValueHandler *handler = [CapturingSingleValueHandler handler];
-  id<GRXWriteable> writeable = [GRXWriteable writeableWithSingleValueHandler:handler.block];
+  id<GRXWriteable> writeable = [GRXWriteable writeableWithSingleHandler:handler.block];
   NSError *anyError = [NSError errorWithDomain:@"domain" code:7 userInfo:nil];
 
   // If:

src/python/grpcio/grpc/_adapter/_c/types.h

@@ -113,6 +113,7 @@ Call *pygrpc_Call_new_empty(CompletionQueue *cq);
 void pygrpc_Call_dealloc(Call *self);
 PyObject *pygrpc_Call_start_batch(Call *self, PyObject *args, PyObject *kwargs);
 PyObject *pygrpc_Call_cancel(Call *self, PyObject *args, PyObject *kwargs);
+PyObject *pygrpc_Call_peer(Call *self);
 extern PyTypeObject pygrpc_Call_type;
 
 
@@ -129,6 +130,11 @@ Channel *pygrpc_Channel_new(
 void pygrpc_Channel_dealloc(Channel *self);
 Call *pygrpc_Channel_create_call(
     Channel *self, PyObject *args, PyObject *kwargs);
+PyObject *pygrpc_Channel_check_connectivity_state(Channel *self, PyObject *args,
+                                                  PyObject *kwargs);
+PyObject *pygrpc_Channel_watch_connectivity_state(Channel *self, PyObject *args,
+                                                  PyObject *kwargs);
+PyObject *pygrpc_Channel_target(Channel *self);
 extern PyTypeObject pygrpc_Channel_type;
 
 
@@ -181,6 +187,9 @@ pygrpc_tag *pygrpc_produce_request_tag(PyObject *user_tag, Call *empty_call);
 /* Construct a tag associated with a server shutdown. */
 pygrpc_tag *pygrpc_produce_server_shutdown_tag(PyObject *user_tag);
 
+/* Construct a tag associated with a channel state change. */
+pygrpc_tag *pygrpc_produce_channel_state_change_tag(PyObject *user_tag);
+
 /* Frees all resources owned by the tag and the tag itself. */
 void pygrpc_discard_tag(pygrpc_tag *tag);
 

src/python/grpcio/grpc/_adapter/_c/types/call.c

@@ -42,6 +42,7 @@
 PyMethodDef pygrpc_Call_methods[] = {
     {"start_batch", (PyCFunction)pygrpc_Call_start_batch, METH_KEYWORDS, ""},
     {"cancel", (PyCFunction)pygrpc_Call_cancel, METH_KEYWORDS, ""},
+    {"peer", (PyCFunction)pygrpc_Call_peer, METH_NOARGS, ""},
     {NULL}
 };
 const char pygrpc_Call_doc[] = "See grpc._adapter._types.Call.";
@@ -161,3 +162,10 @@ PyObject *pygrpc_Call_cancel(Call *self, PyObject *args, PyObject *kwargs) {
   }
   return PyInt_FromLong(errcode);
 }
+
+PyObject *pygrpc_Call_peer(Call *self) {
+  char *peer = grpc_call_get_peer(self->c_call);
+  PyObject *py_peer = PyString_FromString(peer);
+  gpr_free(peer);
+  return py_peer;
+}

src/python/grpcio/grpc/_adapter/_c/types/channel.c

@@ -36,10 +36,14 @@
 #define PY_SSIZE_T_CLEAN
 #include <Python.h>
 #include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
 
 
 PyMethodDef pygrpc_Channel_methods[] = {
     {"create_call", (PyCFunction)pygrpc_Channel_create_call, METH_KEYWORDS, ""},
+    {"check_connectivity_state", (PyCFunction)pygrpc_Channel_check_connectivity_state, METH_KEYWORDS, ""},
+    {"watch_connectivity_state", (PyCFunction)pygrpc_Channel_watch_connectivity_state, METH_KEYWORDS, ""},
+    {"target", (PyCFunction)pygrpc_Channel_target, METH_NOARGS, ""},
     {NULL}
 };
 const char pygrpc_Channel_doc[] = "See grpc._adapter._types.Channel.";
@@ -122,7 +126,7 @@ Call *pygrpc_Channel_create_call(
   const char *host;
   double deadline;
   char *keywords[] = {"cq", "method", "host", "deadline", NULL};
-  if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O!ssd:create_call", keywords,
+  if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O!szd:create_call", keywords,
         &pygrpc_CompletionQueue_type, &cq, &method, &host, &deadline)) {
     return NULL;
   }
@@ -132,3 +136,51 @@ Call *pygrpc_Channel_create_call(
       pygrpc_cast_double_to_gpr_timespec(deadline), NULL);
   return call;
 }
+
+PyObject *pygrpc_Channel_check_connectivity_state(
+    Channel *self, PyObject *args, PyObject *kwargs) {
+  PyObject *py_try_to_connect;
+  int try_to_connect;
+  char *keywords[] = {"try_to_connect", NULL};
+  grpc_connectivity_state state;
+  if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O:connectivity_state", keywords,
+                                   &py_try_to_connect)) {
+    return NULL;
+  }
+  if (!PyBool_Check(py_try_to_connect)) {
+    Py_XDECREF(py_try_to_connect);
+    return NULL;
+  }
+  try_to_connect = Py_True == py_try_to_connect;
+  Py_DECREF(py_try_to_connect);
+  state = grpc_channel_check_connectivity_state(self->c_chan, try_to_connect);
+  return PyInt_FromLong(state);
+}
+
+PyObject *pygrpc_Channel_watch_connectivity_state(
+    Channel *self, PyObject *args, PyObject *kwargs) {
+  PyObject *tag;
+  double deadline;
+  int last_observed_state;
+  CompletionQueue *completion_queue;
+  char *keywords[] = {"last_observed_state", "deadline",
+                      "completion_queue", "tag"};
+  if (!PyArg_ParseTupleAndKeywords(
+      args, kwargs, "idO!O:watch_connectivity_state", keywords,
+      &last_observed_state, &deadline, &pygrpc_CompletionQueue_type,
+      &completion_queue, &tag)) {
+    return NULL;
+  }
+  grpc_channel_watch_connectivity_state(
+      self->c_chan, (grpc_connectivity_state)last_observed_state,
+      pygrpc_cast_double_to_gpr_timespec(deadline), completion_queue->c_cq,
+      pygrpc_produce_channel_state_change_tag(tag));
+  Py_RETURN_NONE;
+}
+
+PyObject *pygrpc_Channel_target(Channel *self) {
+  char *target = grpc_channel_get_target(self->c_chan);
+  PyObject *py_target = PyString_FromString(target);
+  gpr_free(target);
+  return py_target;
+}

src/python/grpcio/grpc/_adapter/_c/utility.c

@@ -88,6 +88,19 @@ pygrpc_tag *pygrpc_produce_server_shutdown_tag(PyObject *user_tag) {
   return tag;
 }
 
+pygrpc_tag *pygrpc_produce_channel_state_change_tag(PyObject *user_tag) {
+  pygrpc_tag *tag = gpr_malloc(sizeof(pygrpc_tag));
+  tag->user_tag = user_tag;
+  Py_XINCREF(tag->user_tag);
+  tag->call = NULL;
+  tag->ops = NULL;
+  tag->nops = 0;
+  grpc_call_details_init(&tag->request_call_details);
+  grpc_metadata_array_init(&tag->request_metadata);
+  tag->is_new_call = 0;
+  return tag;
+}
+
 void pygrpc_discard_tag(pygrpc_tag *tag) {
   if (!tag) {
     return;
@@ -139,7 +152,7 @@ PyObject *pygrpc_consume_event(grpc_event event) {
 }
 
 int pygrpc_produce_op(PyObject *op, grpc_op *result) {
-  static const int OP_TUPLE_SIZE = 5;
+  static const int OP_TUPLE_SIZE = 6;
   static const int STATUS_TUPLE_SIZE = 2;
   static const int TYPE_INDEX = 0;
   static const int INITIAL_METADATA_INDEX = 1;
@@ -148,6 +161,7 @@ int pygrpc_produce_op(PyObject *op, grpc_op *result) {
   static const int STATUS_INDEX = 4;
   static const int STATUS_CODE_INDEX = 0;
   static const int STATUS_DETAILS_INDEX = 1;
+  static const int WRITE_FLAGS_INDEX = 5;
   int type;
   Py_ssize_t message_size;
   char *message;
@@ -170,7 +184,10 @@ int pygrpc_produce_op(PyObject *op, grpc_op *result) {
     return 0;
   }
   c_op.op = type;
-  c_op.flags = 0;
+  c_op.flags = PyInt_AsLong(PyTuple_GET_ITEM(op, WRITE_FLAGS_INDEX));
+  if (PyErr_Occurred()) {
+    return 0;
+  }
   switch (type) {
   case GRPC_OP_SEND_INITIAL_METADATA:
     if (!pygrpc_cast_pyseq_to_send_metadata(

src/python/grpcio/grpc/_adapter/_intermediary_low.py

@@ -127,7 +127,7 @@ class Call(object):
 
   def write(self, message, tag):
     return self._internal.start_batch([
-          _types.OpArgs.send_message(message)
+          _types.OpArgs.send_message(message, 0)
       ], _TagAdapter(tag, Event.Kind.WRITE_ACCEPTED))
 
   def complete(self, tag):

src/python/grpcio/grpc/_adapter/_low.py

@@ -75,6 +75,9 @@ class Call(_types.Call):
     else:
       return self.call.cancel(code, details)
 
+  def peer(self):
+    return self.call.peer()
+
 
 class Channel(_types.Channel):
 
@@ -88,6 +91,17 @@ class Channel(_types.Channel):
   def create_call(self, completion_queue, method, host, deadline=None):
     return Call(self.channel.create_call(completion_queue.completion_queue, method, host, deadline))
 
+  def check_connectivity_state(self, try_to_connect):
+    return self.channel.check_connectivity_state(try_to_connect)
+
+  def watch_connectivity_state(self, last_observed_state, deadline,
+                               completion_queue, tag):
+    self.channel.watch_connectivity_state(
+        last_observed_state, deadline, completion_queue.completion_queue, tag)
+
+  def target(self):
+    return self.channel.target()
+
 
 _NO_TAG = object()
 

src/python/grpcio/grpc/_adapter/_types.py

@@ -31,13 +31,12 @@ import abc
 import collections
 import enum
 
-# TODO(atash): decide whether or not to move these enums to the _c module to
-# force build errors with upstream changes.
 
 class GrpcChannelArgumentKeys(enum.Enum):
   """Mirrors keys used in grpc_channel_args for GRPC-specific arguments."""
   SSL_TARGET_NAME_OVERRIDE = 'grpc.ssl_target_name_override'
 
+
 @enum.unique
 class CallError(enum.IntEnum):
   """Mirrors grpc_call_error in the C core."""
@@ -53,6 +52,7 @@ class CallError(enum.IntEnum):
   ERROR_INVALID_FLAGS       = 9
   ERROR_INVALID_METADATA    = 10
 
+
 @enum.unique
 class StatusCode(enum.IntEnum):
   """Mirrors grpc_status_code in the C core."""
@@ -74,6 +74,14 @@ class StatusCode(enum.IntEnum):
   DATA_LOSS           = 15
   UNAUTHENTICATED     = 16
 
+
+@enum.unique
+class OpWriteFlags(enum.IntEnum):
+  """Mirrors defined write-flag constants in the C core."""
+  WRITE_BUFFER_HINT = 1
+  WRITE_NO_COMPRESS = 2
+
+
 @enum.unique
 class OpType(enum.IntEnum):
   """Mirrors grpc_op_type in the C core."""
@@ -86,12 +94,24 @@ class OpType(enum.IntEnum):
   RECV_STATUS_ON_CLIENT   = 6
   RECV_CLOSE_ON_SERVER    = 7
 
+
 @enum.unique
 class EventType(enum.IntEnum):
   """Mirrors grpc_completion_type in the C core."""
-  QUEUE_SHUTDOWN  = 0
-  QUEUE_TIMEOUT   = 1  # if seen on the Python side, something went horridly wrong
-  OP_COMPLETE     = 2
+  QUEUE_SHUTDOWN = 0
+  QUEUE_TIMEOUT  = 1  # if seen on the Python side, something went horridly wrong
+  OP_COMPLETE    = 2
+
+
+@enum.unique
+class ConnectivityState(enum.IntEnum):
+  """Mirrors grpc_connectivity_state in the C core."""
+  IDLE              = 0
+  CONNECTING        = 1
+  READY             = 2
+  TRANSIENT_FAILURE = 3
+  FATAL_FAILURE     = 4
+
 
 class Status(collections.namedtuple(
     'Status', [
@@ -105,6 +125,7 @@ class Status(collections.namedtuple(
     details (str): ...
   """
 
+
 class CallDetails(collections.namedtuple(
     'CallDetails', [
         'method',
@@ -119,6 +140,7 @@ class CallDetails(collections.namedtuple(
     deadline (float): ...
   """
 
+
 class OpArgs(collections.namedtuple(
     'OpArgs', [
         'type',
@@ -126,6 +148,7 @@ class OpArgs(collections.namedtuple(
         'trailing_metadata',
         'message',
         'status',
+        'write_flags',
     ])):
   """Arguments passed into a GRPC operation.
 
@@ -138,39 +161,40 @@ class OpArgs(collections.namedtuple(
     message (bytes): Only valid if type == OpType.SEND_MESSAGE, else is None.
     status (Status): Only valid if type == OpType.SEND_STATUS_FROM_SERVER, else
       is None.
+    write_flags (int): a bit OR'ing of 0 or more OpWriteFlags values.
   """
 
   @staticmethod
   def send_initial_metadata(initial_metadata):
-    return OpArgs(OpType.SEND_INITIAL_METADATA, initial_metadata, None, None, None)
+    return OpArgs(OpType.SEND_INITIAL_METADATA, initial_metadata, None, None, None, 0)
 
   @staticmethod
-  def send_message(message):
-    return OpArgs(OpType.SEND_MESSAGE, None, None, message, None)
+  def send_message(message, flags):
+    return OpArgs(OpType.SEND_MESSAGE, None, None, message, None, flags)
 
   @staticmethod
   def send_close_from_client():
-    return OpArgs(OpType.SEND_CLOSE_FROM_CLIENT, None, None, None, None)
+    return OpArgs(OpType.SEND_CLOSE_FROM_CLIENT, None, None, None, None, 0)
 
   @staticmethod
   def send_status_from_server(trailing_metadata, status_code, status_details):
-    return OpArgs(OpType.SEND_STATUS_FROM_SERVER, None, trailing_metadata, None, Status(status_code, status_details))
+    return OpArgs(OpType.SEND_STATUS_FROM_SERVER, None, trailing_metadata, None, Status(status_code, status_details), 0)
 
   @staticmethod
   def recv_initial_metadata():
-    return OpArgs(OpType.RECV_INITIAL_METADATA, None, None, None, None);
+    return OpArgs(OpType.RECV_INITIAL_METADATA, None, None, None, None, 0);
 
   @staticmethod
   def recv_message():
-    return OpArgs(OpType.RECV_MESSAGE, None, None, None, None)
+    return OpArgs(OpType.RECV_MESSAGE, None, None, None, None, 0)
 
   @staticmethod
   def recv_status_on_client():
-    return OpArgs(OpType.RECV_STATUS_ON_CLIENT, None, None, None, None)
+    return OpArgs(OpType.RECV_STATUS_ON_CLIENT, None, None, None, None, 0)
 
   @staticmethod
   def recv_close_on_server():
-    return OpArgs(OpType.RECV_CLOSE_ON_SERVER, None, None, None, None)
+    return OpArgs(OpType.RECV_CLOSE_ON_SERVER, None, None, None, None, 0)
 
 
 class OpResult(collections.namedtuple(
@@ -290,6 +314,15 @@ class Call:
     """
     return CallError.ERROR
 
+  @abc.abstractmethod
+  def peer(self):
+    """Get the peer of this call.
+
+    Returns:
+      str: the peer of this call.
+    """
+    return None
+
 
 class Channel:
   __metaclass__ = abc.ABCMeta
@@ -321,6 +354,40 @@ class Channel:
     """
     return None
 
+  @abc.abstractmethod
+  def check_connectivity_state(self, try_to_connect):
+    """Check and optionally repair the connectivity state of the channel.
+
+    Args:
+      try_to_connect (bool): whether or not to try to connect the channel if
+      disconnected.
+
+    Returns:
+      ConnectivityState: state of the channel at the time of this invocation.
+    """
+    return None
+
+  @abc.abstractmethod
+  def watch_connectivity_state(self, last_observed_state, deadline,
+                               completion_queue, tag):
+    """Watch for connectivity state changes from the last_observed_state.
+
+    Args:
+      last_observed_state (ConnectivityState): ...
+      deadline (float): ...
+      completion_queue (CompletionQueue): ...
+      tag (object) ...
+    """
+
+  @abc.abstractmethod
+  def target(self):
+    """Get the target of this channel.
+
+    Returns:
+      str: the target of this channel.
+    """
+    return None
+
 
 class Server:
   __metaclass__ = abc.ABCMeta

src/python/grpcio_test/grpc_test/_adapter/_low_test.py

@@ -117,7 +117,7 @@ class InsecureServerInsecureClient(unittest.TestCase):
     client_initial_metadata = [(CLIENT_METADATA_ASCII_KEY, CLIENT_METADATA_ASCII_VALUE), (CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE)]
     client_start_batch_result = client_call.start_batch([
         _types.OpArgs.send_initial_metadata(client_initial_metadata),
-        _types.OpArgs.send_message(REQUEST),
+        _types.OpArgs.send_message(REQUEST, 0),
         _types.OpArgs.send_close_from_client(),
         _types.OpArgs.recv_initial_metadata(),
         _types.OpArgs.recv_message(),
@@ -144,6 +144,15 @@ class InsecureServerInsecureClient(unittest.TestCase):
     self.assertEquals(HOST, request_event.call_details.host)
     self.assertLess(abs(DEADLINE - request_event.call_details.deadline), DEADLINE_TOLERANCE)
 
+    # Check that the channel is connected, and that both it and the call have
+    # the proper target and peer; do this after the first flurry of messages to
+    # avoid the possibility that connection was delayed by the core until the
+    # first message was sent.
+    self.assertEqual(_types.ConnectivityState.READY,
+                     self.client_channel.check_connectivity_state(False))
+    self.assertIsNotNone(self.client_channel.target())
+    self.assertIsNotNone(client_call.peer())
+
     server_call_tag = object()
     server_call = request_event.call
     server_initial_metadata = [(SERVER_INITIAL_METADATA_KEY, SERVER_INITIAL_METADATA_VALUE)]
@@ -151,7 +160,7 @@ class InsecureServerInsecureClient(unittest.TestCase):
     server_start_batch_result = server_call.start_batch([
         _types.OpArgs.send_initial_metadata(server_initial_metadata),
         _types.OpArgs.recv_message(),
-        _types.OpArgs.send_message(RESPONSE),
+        _types.OpArgs.send_message(RESPONSE, 0),
         _types.OpArgs.recv_close_on_server(),
         _types.OpArgs.send_status_from_server(server_trailing_metadata, SERVER_STATUS_CODE, SERVER_STATUS_DETAILS)
     ], server_call_tag)

src/python/grpcio_test/grpc_test/framework/face/testing/blocking_invocation_inline_service_test_case.py

@@ -34,15 +34,13 @@ import abc
 import unittest  # pylint: disable=unused-import
 
 from grpc.framework.face import exceptions
+from grpc_test.framework.common import test_constants
 from grpc_test.framework.face.testing import control
 from grpc_test.framework.face.testing import coverage
 from grpc_test.framework.face.testing import digest
 from grpc_test.framework.face.testing import stock_service
 from grpc_test.framework.face.testing import test_case
 
-_TIMEOUT = 3
-_LONG_TIMEOUT = 45
-
 
 class BlockingInvocationInlineServiceTestCase(
     test_case.FaceTestCase, coverage.BlockingCoverage):
@@ -79,7 +77,7 @@ class BlockingInvocationInlineServiceTestCase(
         request = test_messages.request()
 
         response = self.stub.blocking_value_in_value_out(
-            name, request, _LONG_TIMEOUT)
+            name, request, test_constants.LONG_TIMEOUT)
 
         test_messages.verify(request, response, self)
 
@@ -90,7 +88,7 @@ class BlockingInvocationInlineServiceTestCase(
         request = test_messages.request()
 
         response_iterator = self.stub.inline_value_in_stream_out(
-            name, request, _LONG_TIMEOUT)
+            name, request, test_constants.LONG_TIMEOUT)
         responses = list(response_iterator)
 
         test_messages.verify(request, responses, self)
@@ -102,7 +100,7 @@ class BlockingInvocationInlineServiceTestCase(
         requests = test_messages.requests()
 
         response = self.stub.blocking_stream_in_value_out(
-            name, iter(requests), _LONG_TIMEOUT)
+            name, iter(requests), test_constants.LONG_TIMEOUT)
 
         test_messages.verify(requests, response, self)
 
@@ -113,7 +111,7 @@ class BlockingInvocationInlineServiceTestCase(
         requests = test_messages.requests()
 
         response_iterator = self.stub.inline_stream_in_stream_out(
-            name, iter(requests), _LONG_TIMEOUT)
+            name, iter(requests), test_constants.LONG_TIMEOUT)
         responses = list(response_iterator)
 
         test_messages.verify(requests, responses, self)
@@ -126,12 +124,12 @@ class BlockingInvocationInlineServiceTestCase(
         second_request = test_messages.request()
 
         first_response = self.stub.blocking_value_in_value_out(
-            name, first_request, _TIMEOUT)
+            name, first_request, test_constants.SHORT_TIMEOUT)
 
         test_messages.verify(first_request, first_response, self)
 
         second_response = self.stub.blocking_value_in_value_out(
-            name, second_request, _TIMEOUT)
+            name, second_request, test_constants.SHORT_TIMEOUT)
 
         test_messages.verify(second_request, second_response, self)
 
@@ -144,7 +142,7 @@ class BlockingInvocationInlineServiceTestCase(
         with self.control.pause(), self.assertRaises(
             exceptions.ExpirationError):
           multi_callable = self.stub.unary_unary_multi_callable(name)
-          multi_callable(request, _TIMEOUT)
+          multi_callable(request, test_constants.SHORT_TIMEOUT)
 
   def testExpiredUnaryRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -155,7 +153,7 @@ class BlockingInvocationInlineServiceTestCase(
         with self.control.pause(), self.assertRaises(
             exceptions.ExpirationError):
           response_iterator = self.stub.inline_value_in_stream_out(
-              name, request, _TIMEOUT)
+              name, request, test_constants.SHORT_TIMEOUT)
           list(response_iterator)
 
   def testExpiredStreamRequestUnaryResponse(self):
@@ -167,7 +165,7 @@ class BlockingInvocationInlineServiceTestCase(
         with self.control.pause(), self.assertRaises(
             exceptions.ExpirationError):
           multi_callable = self.stub.stream_unary_multi_callable(name)
-          multi_callable(iter(requests), _TIMEOUT)
+          multi_callable(iter(requests), test_constants.SHORT_TIMEOUT)
 
   def testExpiredStreamRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -178,7 +176,7 @@ class BlockingInvocationInlineServiceTestCase(
         with self.control.pause(), self.assertRaises(
             exceptions.ExpirationError):
           response_iterator = self.stub.inline_stream_in_stream_out(
-              name, iter(requests), _TIMEOUT)
+              name, iter(requests), test_constants.SHORT_TIMEOUT)
           list(response_iterator)
 
   def testFailedUnaryRequestUnaryResponse(self):
@@ -188,7 +186,8 @@ class BlockingInvocationInlineServiceTestCase(
         request = test_messages.request()
 
         with self.control.fail(), self.assertRaises(exceptions.ServicerError):
-          self.stub.blocking_value_in_value_out(name, request, _TIMEOUT)
+          self.stub.blocking_value_in_value_out(name, request,
+                                                test_constants.SHORT_TIMEOUT)
 
   def testFailedUnaryRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -198,7 +197,7 @@ class BlockingInvocationInlineServiceTestCase(
 
         with self.control.fail(), self.assertRaises(exceptions.ServicerError):
           response_iterator = self.stub.inline_value_in_stream_out(
-              name, request, _TIMEOUT)
+              name, request, test_constants.SHORT_TIMEOUT)
           list(response_iterator)
 
   def testFailedStreamRequestUnaryResponse(self):
@@ -208,7 +207,8 @@ class BlockingInvocationInlineServiceTestCase(
         requests = test_messages.requests()
 
         with self.control.fail(), self.assertRaises(exceptions.ServicerError):
-          self.stub.blocking_stream_in_value_out(name, iter(requests), _TIMEOUT)
+          self.stub.blocking_stream_in_value_out(name, iter(requests),
+                                                 test_constants.SHORT_TIMEOUT)
 
   def testFailedStreamRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -218,5 +218,5 @@ class BlockingInvocationInlineServiceTestCase(
 
         with self.control.fail(), self.assertRaises(exceptions.ServicerError):
           response_iterator = self.stub.inline_stream_in_stream_out(
-              name, iter(requests), _TIMEOUT)
+              name, iter(requests), test_constants.SHORT_TIMEOUT)
           list(response_iterator)

src/python/grpcio_test/grpc_test/framework/face/testing/event_invocation_synchronous_event_service_test_case.py

@@ -33,6 +33,7 @@ import abc
 import unittest
 
 from grpc.framework.face import interfaces
+from grpc_test.framework.common import test_constants
 from grpc_test.framework.face.testing import callback as testing_callback
 from grpc_test.framework.face.testing import control
 from grpc_test.framework.face.testing import coverage
@@ -40,8 +41,6 @@ from grpc_test.framework.face.testing import digest
 from grpc_test.framework.face.testing import stock_service
 from grpc_test.framework.face.testing import test_case
 
-_TIMEOUT = 3
-
 
 class EventInvocationSynchronousEventServiceTestCase(
     test_case.FaceTestCase, coverage.FullCoverage):
@@ -79,7 +78,8 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         self.stub.event_value_in_value_out(
-            name, request, callback.complete, callback.abort, _TIMEOUT)
+            name, request, callback.complete, callback.abort,
+            test_constants.SHORT_TIMEOUT)
         callback.block_until_terminated()
         response = callback.response()
 
@@ -93,7 +93,8 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         self.stub.event_value_in_stream_out(
-            name, request, callback, callback.abort, _TIMEOUT)
+            name, request, callback, callback.abort,
+            test_constants.SHORT_TIMEOUT)
         callback.block_until_terminated()
         responses = callback.responses()
 
@@ -107,7 +108,8 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         unused_call, request_consumer = self.stub.event_stream_in_value_out(
-            name, callback.complete, callback.abort, _TIMEOUT)
+            name, callback.complete, callback.abort,
+            test_constants.SHORT_TIMEOUT)
         for request in requests:
           request_consumer.consume(request)
         request_consumer.terminate()
@@ -124,7 +126,7 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         unused_call, request_consumer = self.stub.event_stream_in_stream_out(
-            name, callback, callback.abort, _TIMEOUT)
+            name, callback, callback.abort, test_constants.SHORT_TIMEOUT)
         for request in requests:
           request_consumer.consume(request)
         request_consumer.terminate()
@@ -147,11 +149,11 @@ class EventInvocationSynchronousEventServiceTestCase(
           first_callback.complete(first_response)
           self.stub.event_value_in_value_out(
               name, second_request, second_callback.complete,
-              second_callback.abort, _TIMEOUT)
+              second_callback.abort, test_constants.SHORT_TIMEOUT)
 
         self.stub.event_value_in_value_out(
             name, first_request, make_second_invocation, first_callback.abort,
-            _TIMEOUT)
+           test_constants.SHORT_TIMEOUT)
         second_callback.block_until_terminated()
 
         first_response = first_callback.response()
@@ -168,7 +170,8 @@ class EventInvocationSynchronousEventServiceTestCase(
 
         with self.control.pause():
           self.stub.event_value_in_value_out(
-              name, request, callback.complete, callback.abort, _TIMEOUT)
+              name, request, callback.complete, callback.abort,
+              test_constants.SHORT_TIMEOUT)
           callback.block_until_terminated()
 
         self.assertEqual(interfaces.Abortion.EXPIRED, callback.abortion())
@@ -182,7 +185,8 @@ class EventInvocationSynchronousEventServiceTestCase(
 
         with self.control.pause():
           self.stub.event_value_in_stream_out(
-              name, request, callback, callback.abort, _TIMEOUT)
+              name, request, callback, callback.abort,
+              test_constants.SHORT_TIMEOUT)
           callback.block_until_terminated()
 
         self.assertEqual(interfaces.Abortion.EXPIRED, callback.abortion())
@@ -194,7 +198,8 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         self.stub.event_stream_in_value_out(
-            name, callback.complete, callback.abort, _TIMEOUT)
+            name, callback.complete, callback.abort,
+            test_constants.SHORT_TIMEOUT)
         callback.block_until_terminated()
 
         self.assertEqual(interfaces.Abortion.EXPIRED, callback.abortion())
@@ -207,7 +212,7 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         unused_call, request_consumer = self.stub.event_stream_in_stream_out(
-            name, callback, callback.abort, _TIMEOUT)
+            name, callback, callback.abort, test_constants.SHORT_TIMEOUT)
         for request in requests:
           request_consumer.consume(request)
         callback.block_until_terminated()
@@ -223,10 +228,12 @@ class EventInvocationSynchronousEventServiceTestCase(
 
         with self.control.fail():
           self.stub.event_value_in_value_out(
-              name, request, callback.complete, callback.abort, _TIMEOUT)
+              name, request, callback.complete, callback.abort,
+              test_constants.SHORT_TIMEOUT)
           callback.block_until_terminated()
 
-        self.assertEqual(interfaces.Abortion.SERVICER_FAILURE, callback.abortion())
+        self.assertEqual(interfaces.Abortion.SERVICER_FAILURE,
+                         callback.abortion())
 
   def testFailedUnaryRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -237,10 +244,12 @@ class EventInvocationSynchronousEventServiceTestCase(
 
         with self.control.fail():
           self.stub.event_value_in_stream_out(
-              name, request, callback, callback.abort, _TIMEOUT)
+              name, request, callback, callback.abort,
+              test_constants.SHORT_TIMEOUT)
           callback.block_until_terminated()
 
-        self.assertEqual(interfaces.Abortion.SERVICER_FAILURE, callback.abortion())
+        self.assertEqual(interfaces.Abortion.SERVICER_FAILURE,
+                         callback.abortion())
 
   def testFailedStreamRequestUnaryResponse(self):
     for name, test_messages_sequence in (
@@ -251,13 +260,15 @@ class EventInvocationSynchronousEventServiceTestCase(
 
         with self.control.fail():
           unused_call, request_consumer = self.stub.event_stream_in_value_out(
-              name, callback.complete, callback.abort, _TIMEOUT)
+              name, callback.complete, callback.abort,
+              test_constants.SHORT_TIMEOUT)
           for request in requests:
             request_consumer.consume(request)
           request_consumer.terminate()
           callback.block_until_terminated()
 
-        self.assertEqual(interfaces.Abortion.SERVICER_FAILURE, callback.abortion())
+        self.assertEqual(interfaces.Abortion.SERVICER_FAILURE,
+                         callback.abortion())
 
   def testFailedStreamRequestStreamResponse(self):
     for name, test_messages_sequence in (
@@ -268,7 +279,7 @@ class EventInvocationSynchronousEventServiceTestCase(
 
         with self.control.fail():
           unused_call, request_consumer = self.stub.event_stream_in_stream_out(
-              name, callback, callback.abort, _TIMEOUT)
+              name, callback, callback.abort, test_constants.SHORT_TIMEOUT)
           for request in requests:
             request_consumer.consume(request)
           request_consumer.terminate()
@@ -287,10 +298,10 @@ class EventInvocationSynchronousEventServiceTestCase(
 
         self.stub.event_value_in_value_out(
             name, first_request, first_callback.complete, first_callback.abort,
-            _TIMEOUT)
+           test_constants.SHORT_TIMEOUT)
         self.stub.event_value_in_value_out(
             name, second_request, second_callback.complete,
-            second_callback.abort, _TIMEOUT)
+            second_callback.abort, test_constants.SHORT_TIMEOUT)
         first_callback.block_until_terminated()
         second_callback.block_until_terminated()
 
@@ -312,7 +323,8 @@ class EventInvocationSynchronousEventServiceTestCase(
 
         with self.control.pause():
           call = self.stub.event_value_in_value_out(
-              name, request, callback.complete, callback.abort, _TIMEOUT)
+              name, request, callback.complete, callback.abort,
+              test_constants.SHORT_TIMEOUT)
           call.cancel()
           callback.block_until_terminated()
 
@@ -326,7 +338,8 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         call = self.stub.event_value_in_stream_out(
-            name, request, callback, callback.abort, _TIMEOUT)
+            name, request, callback, callback.abort,
+            test_constants.SHORT_TIMEOUT)
         call.cancel()
         callback.block_until_terminated()
 
@@ -340,7 +353,8 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         call, request_consumer = self.stub.event_stream_in_value_out(
-            name, callback.complete, callback.abort, _TIMEOUT)
+            name, callback.complete, callback.abort,
+            test_constants.SHORT_TIMEOUT)
         for request in requests:
           request_consumer.consume(request)
         call.cancel()
@@ -355,7 +369,7 @@ class EventInvocationSynchronousEventServiceTestCase(
         callback = testing_callback.Callback()
 
         call, unused_request_consumer = self.stub.event_stream_in_stream_out(
-            name, callback, callback.abort, _TIMEOUT)
+            name, callback, callback.abort, test_constants.SHORT_TIMEOUT)
         call.cancel()
         callback.block_until_terminated()
 

src/python/grpcio_test/grpc_test/framework/face/testing/future_invocation_asynchronous_event_service_test_case.py

@@ -37,13 +37,13 @@ import unittest
 from grpc.framework.face import exceptions
 from grpc.framework.foundation import future
 from grpc.framework.foundation import logging_pool
+from grpc_test.framework.common import test_constants
 from grpc_test.framework.face.testing import control
 from grpc_test.framework.face.testing import coverage
 from grpc_test.framework.face.testing import digest
 from grpc_test.framework.face.testing import stock_service
 from grpc_test.framework.face.testing import test_case
 
-_TIMEOUT = 3
 _MAXIMUM_POOL_SIZE = 10
 
 
@@ -110,7 +110,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         request = test_messages.request()
 
         response_future = self.stub.future_value_in_value_out(
-            name, request, _TIMEOUT)
+            name, request, test_constants.SHORT_TIMEOUT)
         response = response_future.result()
 
         test_messages.verify(request, response, self)
@@ -122,7 +122,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         request = test_messages.request()
 
         response_iterator = self.stub.inline_value_in_stream_out(
-            name, request, _TIMEOUT)
+            name, request, test_constants.SHORT_TIMEOUT)
         responses = list(response_iterator)
 
         test_messages.verify(request, responses, self)
@@ -138,7 +138,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         # returned to calling code before the iterator yields any requests.
         with request_iterator.pause():
           response_future = self.stub.future_stream_in_value_out(
-              name, request_iterator, _TIMEOUT)
+              name, request_iterator, test_constants.SHORT_TIMEOUT)
         response = response_future.result()
 
         test_messages.verify(requests, response, self)
@@ -154,7 +154,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         # returned to calling code before the iterator yields any requests.
         with request_iterator.pause():
           response_iterator = self.stub.inline_stream_in_stream_out(
-              name, request_iterator, _TIMEOUT)
+              name, request_iterator, test_constants.SHORT_TIMEOUT)
         responses = list(response_iterator)
 
         test_messages.verify(requests, responses, self)
@@ -167,13 +167,13 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         second_request = test_messages.request()
 
         first_response_future = self.stub.future_value_in_value_out(
-            name, first_request, _TIMEOUT)
+            name, first_request, test_constants.SHORT_TIMEOUT)
         first_response = first_response_future.result()
 
         test_messages.verify(first_request, first_response, self)
 
         second_response_future = self.stub.future_value_in_value_out(
-            name, second_request, _TIMEOUT)
+            name, second_request, test_constants.SHORT_TIMEOUT)
         second_response = second_response_future.result()
 
         test_messages.verify(second_request, second_response, self)
@@ -186,7 +186,8 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.pause():
           multi_callable = self.stub.unary_unary_multi_callable(name)
-          response_future = multi_callable.future(request, _TIMEOUT)
+          response_future = multi_callable.future(request,
+                                                  test_constants.SHORT_TIMEOUT)
           self.assertIsInstance(
               response_future.exception(), exceptions.ExpirationError)
           with self.assertRaises(exceptions.ExpirationError):
@@ -200,7 +201,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.pause():
           response_iterator = self.stub.inline_value_in_stream_out(
-              name, request, _TIMEOUT)
+              name, request, test_constants.SHORT_TIMEOUT)
           with self.assertRaises(exceptions.ExpirationError):
             list(response_iterator)
 
@@ -212,7 +213,8 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.pause():
           multi_callable = self.stub.stream_unary_multi_callable(name)
-          response_future = multi_callable.future(iter(requests), _TIMEOUT)
+          response_future = multi_callable.future(iter(requests),
+                                                  test_constants.SHORT_TIMEOUT)
           self.assertIsInstance(
               response_future.exception(), exceptions.ExpirationError)
           with self.assertRaises(exceptions.ExpirationError):
@@ -226,7 +228,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.pause():
           response_iterator = self.stub.inline_stream_in_stream_out(
-              name, iter(requests), _TIMEOUT)
+              name, iter(requests), test_constants.SHORT_TIMEOUT)
           with self.assertRaises(exceptions.ExpirationError):
             list(response_iterator)
 
@@ -238,7 +240,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.fail():
           response_future = self.stub.future_value_in_value_out(
-              name, request, _TIMEOUT)
+              name, request, test_constants.SHORT_TIMEOUT)
 
           # Because the servicer fails outside of the thread from which the
           # servicer-side runtime called into it its failure is
@@ -261,7 +263,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         # expiration of the RPC.
         with self.control.fail(), self.assertRaises(exceptions.ExpirationError):
           response_iterator = self.stub.inline_value_in_stream_out(
-              name, request, _TIMEOUT)
+              name, request, test_constants.SHORT_TIMEOUT)
           list(response_iterator)
 
   def testFailedStreamRequestUnaryResponse(self):
@@ -272,7 +274,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.fail():
           response_future = self.stub.future_stream_in_value_out(
-              name, iter(requests), _TIMEOUT)
+              name, iter(requests), test_constants.SHORT_TIMEOUT)
 
           # Because the servicer fails outside of the thread from which the
           # servicer-side runtime called into it its failure is
@@ -295,7 +297,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         # expiration of the RPC.
         with self.control.fail(), self.assertRaises(exceptions.ExpirationError):
           response_iterator = self.stub.inline_stream_in_stream_out(
-              name, iter(requests), _TIMEOUT)
+              name, iter(requests), test_constants.SHORT_TIMEOUT)
           list(response_iterator)
 
   def testParallelInvocations(self):
@@ -305,10 +307,11 @@ class FutureInvocationAsynchronousEventServiceTestCase(
         first_request = test_messages.request()
         second_request = test_messages.request()
 
+        # TODO(bug 2039): use LONG_TIMEOUT instead
         first_response_future = self.stub.future_value_in_value_out(
-            name, first_request, _TIMEOUT)
+            name, first_request, test_constants.SHORT_TIMEOUT)
         second_response_future = self.stub.future_value_in_value_out(
-            name, second_request, _TIMEOUT)
+            name, second_request, test_constants.SHORT_TIMEOUT)
         first_response = first_response_future.result()
         second_response = second_response_future.result()
 
@@ -327,7 +330,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.pause():
           response_future = self.stub.future_value_in_value_out(
-              name, request, _TIMEOUT)
+              name, request, test_constants.SHORT_TIMEOUT)
           cancel_method_return_value = response_future.cancel()
 
         self.assertFalse(cancel_method_return_value)
@@ -341,7 +344,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.pause():
           response_iterator = self.stub.inline_value_in_stream_out(
-              name, request, _TIMEOUT)
+              name, request, test_constants.SHORT_TIMEOUT)
           response_iterator.cancel()
 
         with self.assertRaises(future.CancelledError):
@@ -355,7 +358,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.pause():
           response_future = self.stub.future_stream_in_value_out(
-              name, iter(requests), _TIMEOUT)
+              name, iter(requests), test_constants.SHORT_TIMEOUT)
           cancel_method_return_value = response_future.cancel()
 
         self.assertFalse(cancel_method_return_value)
@@ -369,7 +372,7 @@ class FutureInvocationAsynchronousEventServiceTestCase(
 
         with self.control.pause():
           response_iterator = self.stub.inline_stream_in_stream_out(
-              name, iter(requests), _TIMEOUT)
+              name, iter(requests), test_constants.SHORT_TIMEOUT)
           response_iterator.cancel()
 
         with self.assertRaises(future.CancelledError):

src/ruby/ext/grpc/rb_channel.c

@@ -195,18 +195,28 @@ static VALUE grpc_rb_channel_init_copy(VALUE copy, VALUE orig) {
 
 /* Create a call given a grpc_channel, in order to call method. The request
    is not sent until grpc_call_invoke is called. */
-static VALUE grpc_rb_channel_create_call(VALUE self, VALUE cqueue, VALUE method,
-                                         VALUE host, VALUE deadline) {
+static VALUE grpc_rb_channel_create_call(VALUE self, VALUE cqueue,
+                                         VALUE parent, VALUE mask,
+                                         VALUE method, VALUE host,
+                                         VALUE deadline) {
   VALUE res = Qnil;
   grpc_rb_channel *wrapper = NULL;
   grpc_call *call = NULL;
+  grpc_call *parent_call = NULL;
   grpc_channel *ch = NULL;
   grpc_completion_queue *cq = NULL;
+  int flags = GRPC_PROPAGATE_DEFAULTS;
   char *method_chars = StringValueCStr(method);
   char *host_chars = NULL;
   if (host != Qnil) {
     host_chars = StringValueCStr(host);
   }
+  if (mask != Qnil) {
+    flags = NUM2UINT(mask);
+  }
+  if (parent != Qnil) {
+    parent_call = grpc_rb_get_wrapped_call(parent);
+  }
 
   cq = grpc_rb_get_wrapped_completion_queue(cqueue);
   TypedData_Get_Struct(self, grpc_rb_channel, &grpc_channel_data_type, wrapper);
@@ -216,11 +226,10 @@ static VALUE grpc_rb_channel_create_call(VALUE self, VALUE cqueue, VALUE method,
     return Qnil;
   }
 
-  call = grpc_channel_create_call(ch, NULL, GRPC_PROPAGATE_DEFAULTS, cq,
-                                  method_chars, host_chars,
-                                  grpc_rb_time_timeval(deadline,
-                                                       /* absolute time */ 0),
-                                  NULL);
+  call = grpc_channel_create_call(ch, parent_call, flags, cq, method_chars,
+                                  host_chars, grpc_rb_time_timeval(
+                                      deadline,
+                                      /* absolute time */ 0), NULL);
   if (call == NULL) {
     rb_raise(rb_eRuntimeError, "cannot create call with method %s",
              method_chars);
@@ -238,6 +247,7 @@ static VALUE grpc_rb_channel_create_call(VALUE self, VALUE cqueue, VALUE method,
   return res;
 }
 
+
 /* Closes the channel, calling it's destroy method */
 static VALUE grpc_rb_channel_destroy(VALUE self) {
   grpc_rb_channel *wrapper = NULL;
@@ -269,6 +279,22 @@ static VALUE grpc_rb_channel_get_target(VALUE self) {
   return res;
 }
 
+static void Init_grpc_propagate_masks() {
+  /* Constants representing call propagation masks in grpc.h */
+  VALUE grpc_rb_mPropagateMasks = rb_define_module_under(
+      grpc_rb_mGrpcCore, "PropagateMasks");
+  rb_define_const(grpc_rb_mPropagateMasks, "DEADLINE",
+                  UINT2NUM(GRPC_PROPAGATE_DEADLINE));
+  rb_define_const(grpc_rb_mPropagateMasks, "CENSUS_STATS_CONTEXT",
+                  UINT2NUM(GRPC_PROPAGATE_CENSUS_STATS_CONTEXT));
+  rb_define_const(grpc_rb_mPropagateMasks, "CENSUS_TRACING_CONTEXT",
+                  UINT2NUM(GRPC_PROPAGATE_CENSUS_TRACING_CONTEXT));
+  rb_define_const(grpc_rb_mPropagateMasks, "CANCELLATION",
+                  UINT2NUM(GRPC_PROPAGATE_CANCELLATION));
+  rb_define_const(grpc_rb_mPropagateMasks, "DEFAULTS",
+                  UINT2NUM(GRPC_PROPAGATE_DEFAULTS));
+}
+
 void Init_grpc_channel() {
   grpc_rb_cChannelArgs = rb_define_class("TmpChannelArgs", rb_cObject);
   grpc_rb_cChannel =
@@ -284,7 +310,7 @@ void Init_grpc_channel() {
 
   /* Add ruby analogues of the Channel methods. */
   rb_define_method(grpc_rb_cChannel, "create_call",
-                   grpc_rb_channel_create_call, 4);
+                   grpc_rb_channel_create_call, 6);
   rb_define_method(grpc_rb_cChannel, "target", grpc_rb_channel_get_target, 0);
   rb_define_method(grpc_rb_cChannel, "destroy", grpc_rb_channel_destroy, 0);
   rb_define_alias(grpc_rb_cChannel, "close", "destroy");
@@ -300,6 +326,7 @@ void Init_grpc_channel() {
                   ID2SYM(rb_intern(GRPC_ARG_MAX_CONCURRENT_STREAMS)));
   rb_define_const(grpc_rb_cChannel, "MAX_MESSAGE_LENGTH",
                   ID2SYM(rb_intern(GRPC_ARG_MAX_MESSAGE_LENGTH)));
+  Init_grpc_propagate_masks();
 }
 
 /* Gets the wrapped channel from the ruby wrapper */

src/ruby/grpc.gemspec

@@ -22,7 +22,7 @@ Gem::Specification.new do |s|
   s.files += Dir.glob('bin/**/*')
   s.test_files = Dir.glob('spec/**/*')
   %w(math noproto).each do |b|
-    s.executables += [ "#{b}_client.rb", "#{b}_server.rb" ]
+    s.executables += ["#{b}_client.rb", "#{b}_server.rb"]
   end
   s.require_paths = %w( bin lib )
   s.platform      = Gem::Platform::RUBY

src/ruby/lib/grpc/generic/client_stub.rb

@@ -32,6 +32,8 @@ require 'grpc/version'
 
 # GRPC contains the General RPC module.
 module GRPC
+  # rubocop:disable Metrics/ParameterLists
+
   # ClientStub represents an endpoint used to send requests to GRPC servers.
   class ClientStub
     include Core::StatusCodes
@@ -68,6 +70,12 @@ module GRPC
       update_metadata
     end
 
+    # Allows users of the stub to modify the propagate mask.
+    #
+    # This is an advanced feature for use when making calls to another gRPC
+    # server whilst running in the handler of an existing one.
+    attr_writer :propagate_mask
+
     # Creates a new ClientStub.
     #
     # Minimally, a stub is created with the just the host of the gRPC service
@@ -91,8 +99,8 @@ module GRPC
     #
     # - :update_metadata
     # when present, this a func that takes a hash and returns a hash
-    # it can be used to update metadata, i.e, remove, change or update
-    # amend metadata values.
+    # it can be used to update metadata, i.e, remove, or amend
+    # metadata values.
     #
     # @param host [String] the host the stub connects to
     # @param q [Core::CompletionQueue] used to wait for events
@@ -105,6 +113,7 @@ module GRPC
                    channel_override: nil,
                    timeout: nil,
                    creds: nil,
+                   propagate_mask: nil,
                    update_metadata: nil,
                    **kw)
       fail(TypeError, '!CompletionQueue') unless q.is_a?(Core::CompletionQueue)
@@ -113,6 +122,7 @@ module GRPC
       @update_metadata = ClientStub.check_update_metadata(update_metadata)
       alt_host = kw[Core::Channel::SSL_TARGET]
       @host = alt_host.nil? ? host : alt_host
+      @propagate_mask = propagate_mask
       @timeout = timeout.nil? ? DEFAULT_TIMEOUT : timeout
     end
 
@@ -151,11 +161,15 @@ module GRPC
     # @param marshal [Function] f(obj)->string that marshals requests
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
     # @param timeout [Numeric] (optional) the max completion time in seconds
+    # @param parent [Core::Call] a prior call whose reserved metadata
+    #   will be propagated by this one.
     # @param return_op [true|false] return an Operation if true
     # @return [Object] the response received from the server
     def request_response(method, req, marshal, unmarshal, timeout = nil,
-                         return_op: false, **kw)
-      c = new_active_call(method, marshal, unmarshal, timeout)
+                         return_op: false,
+                         parent: parent,
+                         **kw)
+      c = new_active_call(method, marshal, unmarshal, timeout, parent: parent)
       kw_with_jwt_uri = self.class.update_with_jwt_aud_uri(kw, @host, method)
       md = @update_metadata.nil? ? kw : @update_metadata.call(kw_with_jwt_uri)
       return c.request_response(req, **md) unless return_op
@@ -210,10 +224,14 @@ module GRPC
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
     # @param timeout [Numeric] the max completion time in seconds
     # @param return_op [true|false] return an Operation if true
+    # @param parent [Core::Call] a prior call whose reserved metadata
+    #   will be propagated by this one.
     # @return [Object|Operation] the response received from the server
     def client_streamer(method, requests, marshal, unmarshal, timeout = nil,
-                        return_op: false, **kw)
-      c = new_active_call(method, marshal, unmarshal, timeout)
+                        return_op: false,
+                        parent: nil,
+                        **kw)
+      c = new_active_call(method, marshal, unmarshal, timeout, parent: parent)
       kw_with_jwt_uri = self.class.update_with_jwt_aud_uri(kw, @host, method)
       md = @update_metadata.nil? ? kw : @update_metadata.call(kw_with_jwt_uri)
       return c.client_streamer(requests, **md) unless return_op
@@ -276,11 +294,16 @@ module GRPC
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
     # @param timeout [Numeric] the max completion time in seconds
     # @param return_op [true|false]return an Operation if true
+    # @param parent [Core::Call] a prior call whose reserved metadata
+    #   will be propagated by this one.
     # @param blk [Block] when provided, is executed for each response
     # @return [Enumerator|Operation|nil] as discussed above
     def server_streamer(method, req, marshal, unmarshal, timeout = nil,
-                        return_op: false, **kw, &blk)
-      c = new_active_call(method, marshal, unmarshal, timeout)
+                        return_op: false,
+                        parent: nil,
+                        **kw,
+                        &blk)
+      c = new_active_call(method, marshal, unmarshal, timeout, parent: parent)
       kw_with_jwt_uri = self.class.update_with_jwt_aud_uri(kw, @host, method)
       md = @update_metadata.nil? ? kw : @update_metadata.call(kw_with_jwt_uri)
       return c.server_streamer(req, **md, &blk) unless return_op
@@ -381,12 +404,17 @@ module GRPC
     # @param marshal [Function] f(obj)->string that marshals requests
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
     # @param timeout [Numeric] (optional) the max completion time in seconds
-    # @param blk [Block] when provided, is executed for each response
+    # @param parent [Core::Call] a prior call whose reserved metadata
+    #   will be propagated by this one.
     # @param return_op [true|false] return an Operation if true
+    # @param blk [Block] when provided, is executed for each response
     # @return [Enumerator|nil|Operation] as discussed above
     def bidi_streamer(method, requests, marshal, unmarshal, timeout = nil,
-                      return_op: false, **kw, &blk)
-      c = new_active_call(method, marshal, unmarshal, timeout)
+                      return_op: false,
+                      parent: nil,
+                      **kw,
+                      &blk)
+      c = new_active_call(method, marshal, unmarshal, timeout, parent: parent)
       kw_with_jwt_uri = self.class.update_with_jwt_aud_uri(kw, @host, method)
       md = @update_metadata.nil? ? kw : @update_metadata.call(kw_with_jwt_uri)
       return c.bidi_streamer(requests, **md, &blk) unless return_op
@@ -407,10 +435,17 @@ module GRPC
     # @param method [string] the method being called.
     # @param marshal [Function] f(obj)->string that marshals requests
     # @param unmarshal [Function] f(string)->obj that unmarshals responses
+    # @param parent [Grpc::Call] a parent call, available when calls are
+    #   made from server
     # @param timeout [TimeConst]
-    def new_active_call(method, marshal, unmarshal, timeout = nil)
+    def new_active_call(method, marshal, unmarshal, timeout = nil, parent: nil)
       deadline = from_relative_time(timeout.nil? ? @timeout : timeout)
-      call = @ch.create_call(@queue, method, nil, deadline)
+      call = @ch.create_call(@queue,
+                             parent, # parent call
+                             @propagate_mask, # propagation options
+                             method,
+                             nil, # host use nil,
+                             deadline)
       ActiveCall.new(call, @queue, marshal, unmarshal, deadline, started: false)
     end
   end

src/ruby/spec/call_spec.rb

@@ -137,7 +137,7 @@ describe GRPC::Core::Call do
   end
 
   def make_test_call
-    @ch.create_call(client_queue, 'dummy_method', nil, deadline)
+    @ch.create_call(client_queue, nil, nil, 'dummy_method', nil, deadline)
   end
 
   def deadline

src/ruby/spec/channel_spec.rb

@@ -117,7 +117,7 @@ describe GRPC::Core::Channel do
       deadline = Time.now + 5
 
       blk = proc do
-        ch.create_call(cq, 'dummy_method', nil, deadline)
+        ch.create_call(cq, nil, nil, 'dummy_method', nil, deadline)
       end
       expect(&blk).to_not raise_error
     end
@@ -128,7 +128,7 @@ describe GRPC::Core::Channel do
 
       deadline = Time.now + 5
       blk = proc do
-        ch.create_call(cq, 'dummy_method', nil, deadline)
+        ch.create_call(cq, nil, nil, 'dummy_method', nil, deadline)
       end
       expect(&blk).to raise_error(RuntimeError)
     end

src/ruby/spec/client_server_spec.rb

@@ -61,7 +61,7 @@ shared_context 'setup: tags' do
   end
 
   def new_client_call
-    @ch.create_call(@client_queue, '/method', nil, deadline)
+    @ch.create_call(@client_queue, nil, nil, '/method', nil, deadline)
   end
 end
 

src/ruby/spec/generic/active_call_spec.rb

@@ -338,7 +338,7 @@ describe GRPC::ActiveCall do
   end
 
   def make_test_call
-    @ch.create_call(@client_queue, '/method', nil, deadline)
+    @ch.create_call(@client_queue, nil, nil, '/method', nil, deadline)
   end
 
   def deadline

test/cpp/qps/client.h

@@ -41,6 +41,7 @@
 
 #include <condition_variable>
 #include <mutex>
+#include <grpc++/config.h>
 
 namespace grpc {
 
@@ -67,10 +68,12 @@ typedef std::chrono::time_point<grpc_time_source> grpc_time;
 class Client {
  public:
   explicit Client(const ClientConfig& config)
-      : timer_(new Timer), interarrival_timer_() {
+      : channels_(config.client_channels()),
+        timer_(new Timer),
+        interarrival_timer_() {
     for (int i = 0; i < config.client_channels(); i++) {
-      channels_.push_back(ClientChannelInfo(
-          config.server_targets(i % config.server_targets_size()), config));
+      channels_[i].init(config.server_targets(i % config.server_targets_size()),
+                        config);
     }
     request_.set_response_type(grpc::testing::PayloadType::COMPRESSABLE);
     request_.set_response_size(config.payload_size());
@@ -79,7 +82,8 @@ class Client {
 
   ClientStats Mark() {
     Histogram latencies;
-    std::vector<Histogram> to_merge(threads_.size());
+    // avoid std::vector for old compilers that expect a copy constructor
+    Histogram* to_merge = new Histogram[threads_.size()];
     for (size_t i = 0; i < threads_.size(); i++) {
       threads_[i]->BeginSwap(&to_merge[i]);
     }
@@ -89,6 +93,7 @@ class Client {
       threads_[i]->EndSwap();
       latencies.Merge(&to_merge[i]);
     }
+    delete[] to_merge;
 
     auto timer_result = timer->Mark();
 
@@ -106,9 +111,20 @@ class Client {
 
   class ClientChannelInfo {
    public:
-    ClientChannelInfo(const grpc::string& target, const ClientConfig& config)
-        : channel_(CreateTestChannel(target, config.enable_ssl())),
-          stub_(TestService::NewStub(channel_)) {}
+    ClientChannelInfo() {}
+    ClientChannelInfo(const ClientChannelInfo& i) {
+      // The copy constructor is to satisfy old compilers
+      // that need it for using std::vector . It is only ever
+      // used for empty entries
+      GPR_ASSERT(!i.channel_ && !i.stub_);
+    }
+    void init(const grpc::string& target, const ClientConfig& config) {
+      // We have to use a 2-phase init like this with a default
+      // constructor followed by an initializer function to make
+      // old compilers happy with using this in std::vector
+      channel_ = CreateTestChannel(target, config.enable_ssl());
+      stub_ = TestService::NewStub(channel_);
+    }
     ChannelInterface* get_channel() { return channel_.get(); }
     TestService::Stub* get_stub() { return stub_.get(); }
 
@@ -189,27 +205,9 @@ class Client {
     Thread(Client* client, size_t idx)
         : done_(false),
           new_(nullptr),
-          impl_([this, idx, client]() {
-            for (;;) {
-              // run the loop body
-              bool thread_still_ok = client->ThreadFunc(&histogram_, idx);
-              // lock, see if we're done
-              std::lock_guard<std::mutex> g(mu_);
-              if (!thread_still_ok) {
-                gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
-                done_ = true;
-              }
-              if (done_) {
-                return;
-              }
-              // check if we're marking, swap out the histogram if so
-              if (new_) {
-                new_->Swap(&histogram_);
-                new_ = nullptr;
-                cv_.notify_one();
-              }
-            }
-          }) {}
+          client_(client),
+          idx_(idx),
+          impl_(&Thread::ThreadFunc, this) {}
 
     ~Thread() {
       {
@@ -226,13 +224,37 @@ class Client {
 
     void EndSwap() {
       std::unique_lock<std::mutex> g(mu_);
-      cv_.wait(g, [this]() { return new_ == nullptr; });
+      while (new_ != nullptr) {
+        cv_.wait(g);
+      };
     }
 
    private:
     Thread(const Thread&);
     Thread& operator=(const Thread&);
 
+    void ThreadFunc() {
+      for (;;) {
+        // run the loop body
+        const bool thread_still_ok = client_->ThreadFunc(&histogram_, idx_);
+        // lock, see if we're done
+        std::lock_guard<std::mutex> g(mu_);
+        if (!thread_still_ok) {
+          gpr_log(GPR_ERROR, "Finishing client thread due to RPC error");
+          done_ = true;
+        }
+        if (done_) {
+          return;
+        }
+        // check if we're marking, swap out the histogram if so
+        if (new_) {
+          new_->Swap(&histogram_);
+          new_ = nullptr;
+          cv_.notify_one();
+        }
+      }
+    }
+
     TestService::Stub* stub_;
     ClientConfig config_;
     std::mutex mu_;
@@ -240,6 +262,8 @@ class Client {
     bool done_;
     Histogram* new_;
     Histogram histogram_;
+    Client* client_;
+    size_t idx_;
     std::thread impl_;
   };
 

test/cpp/qps/client_async.cc

@@ -156,7 +156,7 @@ class AsyncClient : public Client {
       std::function<ClientRpcContext*(int, TestService::Stub*,
                                       const SimpleRequest&)> setup_ctx)
       : Client(config),
-        channel_lock_(config.client_channels()),
+        channel_lock_(new std::mutex[config.client_channels()]),
         contexts_(config.client_channels()),
         max_outstanding_per_channel_(config.outstanding_rpcs_per_channel()),
         channel_count_(config.client_channels()),
@@ -208,6 +208,7 @@ class AsyncClient : public Client {
         delete ctx;
       }
     }
+    delete[] channel_lock_;
   }
 
   bool ThreadFunc(Histogram* histogram,
@@ -316,23 +317,28 @@ class AsyncClient : public Client {
   }
 
  private:
-  class boolean { // exists only to avoid data-race on vector<bool>
+  class boolean {  // exists only to avoid data-race on vector<bool>
    public:
-    boolean(): val_(false) {}
-    boolean(bool b): val_(b) {}
-    operator bool() const {return val_;}
-    boolean& operator=(bool b) {val_=b; return *this;}
+    boolean() : val_(false) {}
+    boolean(bool b) : val_(b) {}
+    operator bool() const { return val_; }
+    boolean& operator=(bool b) {
+      val_ = b;
+      return *this;
+    }
+
    private:
     bool val_;
   };
   std::vector<std::unique_ptr<CompletionQueue>> cli_cqs_;
 
   std::vector<deadline_list> rpc_deadlines_;  // per thread deadlines
-  std::vector<int> next_channel_;      // per thread round-robin channel ctr
-  std::vector<boolean> issue_allowed_; // may this thread attempt to issue
-  std::vector<grpc_time> next_issue_;  // when should it issue?
+  std::vector<int> next_channel_;       // per thread round-robin channel ctr
+  std::vector<boolean> issue_allowed_;  // may this thread attempt to issue
+  std::vector<grpc_time> next_issue_;   // when should it issue?
 
-  std::vector<std::mutex> channel_lock_;
+  std::mutex*
+      channel_lock_;  // a vector, but avoid std::vector for old compilers
   std::vector<context_list> contexts_;  // per-channel list of idle contexts
   int max_outstanding_per_channel_;
   int channel_count_;
@@ -348,15 +354,17 @@ class AsyncUnaryClient GRPC_FINAL : public AsyncClient {
   ~AsyncUnaryClient() GRPC_OVERRIDE { EndThreads(); }
 
  private:
+  static void CheckDone(grpc::Status s, SimpleResponse* response) {}
+  static std::unique_ptr<grpc::ClientAsyncResponseReader<SimpleResponse>>
+  StartReq(TestService::Stub* stub, grpc::ClientContext* ctx,
+           const SimpleRequest& request, CompletionQueue* cq) {
+    return stub->AsyncUnaryCall(ctx, request, cq);
+  };
   static ClientRpcContext* SetupCtx(int channel_id, TestService::Stub* stub,
                                     const SimpleRequest& req) {
-    auto check_done = [](grpc::Status s, SimpleResponse* response) {};
-    auto start_req = [](TestService::Stub* stub, grpc::ClientContext* ctx,
-                        const SimpleRequest& request, CompletionQueue* cq) {
-      return stub->AsyncUnaryCall(ctx, request, cq);
-    };
     return new ClientRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
-        channel_id, stub, req, start_req, check_done);
+        channel_id, stub, req, AsyncUnaryClient::StartReq,
+        AsyncUnaryClient::CheckDone);
   }
 };
 
@@ -442,16 +450,19 @@ class AsyncStreamingClient GRPC_FINAL : public AsyncClient {
   ~AsyncStreamingClient() GRPC_OVERRIDE { EndThreads(); }
 
  private:
+  static void CheckDone(grpc::Status s, SimpleResponse* response) {}
+  static std::unique_ptr<
+      grpc::ClientAsyncReaderWriter<SimpleRequest, SimpleResponse>>
+  StartReq(TestService::Stub* stub, grpc::ClientContext* ctx,
+           CompletionQueue* cq, void* tag) {
+    auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
+    return stream;
+  };
   static ClientRpcContext* SetupCtx(int channel_id, TestService::Stub* stub,
                                     const SimpleRequest& req) {
-    auto check_done = [](grpc::Status s, SimpleResponse* response) {};
-    auto start_req = [](TestService::Stub* stub, grpc::ClientContext* ctx,
-                        CompletionQueue* cq, void* tag) {
-      auto stream = stub->AsyncStreamingCall(ctx, cq, tag);
-      return stream;
-    };
     return new ClientRpcContextStreamingImpl<SimpleRequest, SimpleResponse>(
-        channel_id, stub, req, start_req, check_done);
+        channel_id, stub, req, AsyncStreamingClient::StartReq,
+        AsyncStreamingClient::CheckDone);
   }
 };
 

test/cpp/qps/client_sync.cc

@@ -45,8 +45,9 @@
 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/histogram.h>
-#include <grpc/support/log.h>
 #include <grpc/support/host_port.h>
+#include <grpc/support/log.h>
+#include <grpc/support/time.h>
 #include <gflags/gflags.h>
 #include <grpc++/client_context.h>
 #include <grpc++/server.h>
@@ -79,7 +80,9 @@ class SynchronousClient : public Client {
   void WaitToIssue(int thread_idx) {
     grpc_time next_time;
     if (NextIssueTime(thread_idx, &next_time)) {
-      std::this_thread::sleep_until(next_time);
+      gpr_timespec next_timespec;
+      TimepointHR2Timespec(next_time, &next_timespec);
+      gpr_sleep_until(next_timespec);
     }
   }
 
@@ -110,9 +113,10 @@ class SynchronousUnaryClient GRPC_FINAL : public SynchronousClient {
 class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
  public:
   SynchronousStreamingClient(const ClientConfig& config)
-      : SynchronousClient(config),
-        context_(num_threads_),
-        stream_(num_threads_) {
+      : SynchronousClient(config) {
+    context_ = new grpc::ClientContext[num_threads_];
+    stream_ = new std::unique_ptr<
+        grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>[num_threads_];
     for (size_t thread_idx = 0; thread_idx < num_threads_; thread_idx++) {
       auto* stub = channels_[thread_idx % channels_.size()].get_stub();
       stream_[thread_idx] = stub->StreamingCall(&context_[thread_idx]);
@@ -121,12 +125,15 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
   }
   ~SynchronousStreamingClient() {
     EndThreads();
-    for (auto stream = stream_.begin(); stream != stream_.end(); stream++) {
+    for (auto stream = &stream_[0]; stream != &stream_[num_threads_];
+         stream++) {
       if (*stream) {
         (*stream)->WritesDone();
         EXPECT_TRUE((*stream)->Finish().ok());
       }
     }
+    delete[] stream_;
+    delete[] context_;
   }
 
   bool ThreadFunc(Histogram* histogram, size_t thread_idx) GRPC_OVERRIDE {
@@ -141,9 +148,11 @@ class SynchronousStreamingClient GRPC_FINAL : public SynchronousClient {
   }
 
  private:
-  std::vector<grpc::ClientContext> context_;
-  std::vector<std::unique_ptr<
-      grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>> stream_;
+  // These are both conceptually std::vector but cannot be for old compilers
+  // that expect contained classes to support copy constructors
+  grpc::ClientContext* context_;
+  std::unique_ptr<grpc::ClientReaderWriter<SimpleRequest, SimpleResponse>>*
+      stream_;
 };
 
 std::unique_ptr<Client> CreateSynchronousUnaryClient(

test/cpp/qps/driver.cc

@@ -77,16 +77,34 @@ static deque<string> get_hosts(const string& name) {
   }
 }
 
+// Namespace for classes and functions used only in RunScenario
+// Using this rather than local definitions to workaround gcc-4.4 limitations
+// regarding using templates without linkage
+namespace runsc {
+
+// ClientContext allocator
+static ClientContext* AllocContext(list<ClientContext>* contexts) {
+  contexts->emplace_back();
+  return &contexts->back();
+}
+
+struct ServerData {
+  unique_ptr<Worker::Stub> stub;
+  unique_ptr<ClientReaderWriter<ServerArgs, ServerStatus>> stream;
+};
+
+struct ClientData {
+  unique_ptr<Worker::Stub> stub;
+  unique_ptr<ClientReaderWriter<ClientArgs, ClientStatus>> stream;
+};
+}  // namespace runsc
+
 std::unique_ptr<ScenarioResult> RunScenario(
     const ClientConfig& initial_client_config, size_t num_clients,
     const ServerConfig& server_config, size_t num_servers, int warmup_seconds,
     int benchmark_seconds, int spawn_local_worker_count) {
-  // ClientContext allocator (all are destroyed at scope exit)
+  // ClientContext allocations (all are destroyed at scope exit)
   list<ClientContext> contexts;
-  auto alloc_context = [&contexts]() {
-    contexts.emplace_back();
-    return &contexts.back();
-  };
 
   // To be added to the result, containing the final configuration used for
   // client and config (incluiding host, etc.)
@@ -131,23 +149,22 @@ std::unique_ptr<ScenarioResult> RunScenario(
   workers.resize(num_clients + num_servers);
 
   // Start servers
-  struct ServerData {
-    unique_ptr<Worker::Stub> stub;
-    unique_ptr<ClientReaderWriter<ServerArgs, ServerStatus>> stream;
-  };
-  vector<ServerData> servers;
+  using runsc::ServerData;
+  // servers is array rather than std::vector to avoid gcc-4.4 issues
+  // where class contained in std::vector must have a copy constructor
+  auto* servers = new ServerData[num_servers];
   for (size_t i = 0; i < num_servers; i++) {
-    ServerData sd;
-    sd.stub = std::move(Worker::NewStub(
+    servers[i].stub = std::move(Worker::NewStub(
         CreateChannel(workers[i], InsecureCredentials(), ChannelArguments())));
     ServerArgs args;
     result_server_config = server_config;
     result_server_config.set_host(workers[i]);
     *args.mutable_setup() = server_config;
-    sd.stream = std::move(sd.stub->RunServer(alloc_context()));
-    GPR_ASSERT(sd.stream->Write(args));
+    servers[i].stream =
+        std::move(servers[i].stub->RunServer(runsc::AllocContext(&contexts)));
+    GPR_ASSERT(servers[i].stream->Write(args));
     ServerStatus init_status;
-    GPR_ASSERT(sd.stream->Read(&init_status));
+    GPR_ASSERT(servers[i].stream->Read(&init_status));
     char* host;
     char* driver_port;
     char* cli_target;
@@ -157,30 +174,25 @@ std::unique_ptr<ScenarioResult> RunScenario(
     gpr_free(host);
     gpr_free(driver_port);
     gpr_free(cli_target);
-
-    servers.push_back(std::move(sd));
   }
 
   // Start clients
-  struct ClientData {
-    unique_ptr<Worker::Stub> stub;
-    unique_ptr<ClientReaderWriter<ClientArgs, ClientStatus>> stream;
-  };
-  vector<ClientData> clients;
+  using runsc::ClientData;
+  // clients is array rather than std::vector to avoid gcc-4.4 issues
+  // where class contained in std::vector must have a copy constructor
+  auto* clients = new ClientData[num_clients];
   for (size_t i = 0; i < num_clients; i++) {
-    ClientData cd;
-    cd.stub = std::move(Worker::NewStub(CreateChannel(
+    clients[i].stub = std::move(Worker::NewStub(CreateChannel(
         workers[i + num_servers], InsecureCredentials(), ChannelArguments())));
     ClientArgs args;
     result_client_config = client_config;
     result_client_config.set_host(workers[i + num_servers]);
     *args.mutable_setup() = client_config;
-    cd.stream = std::move(cd.stub->RunTest(alloc_context()));
-    GPR_ASSERT(cd.stream->Write(args));
+    clients[i].stream =
+        std::move(clients[i].stub->RunTest(runsc::AllocContext(&contexts)));
+    GPR_ASSERT(clients[i].stream->Write(args));
     ClientStatus init_status;
-    GPR_ASSERT(cd.stream->Read(&init_status));
-
-    clients.push_back(std::move(cd));
+    GPR_ASSERT(clients[i].stream->Read(&init_status));
   }
 
   // Let everything warmup
@@ -195,23 +207,25 @@ std::unique_ptr<ScenarioResult> RunScenario(
   server_mark.mutable_mark();
   ClientArgs client_mark;
   client_mark.mutable_mark();
-  for (auto server = servers.begin(); server != servers.end(); server++) {
+  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
     GPR_ASSERT(server->stream->Write(server_mark));
   }
-  for (auto client = clients.begin(); client != clients.end(); client++) {
+  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
     GPR_ASSERT(client->stream->Write(client_mark));
   }
   ServerStatus server_status;
   ClientStatus client_status;
-  for (auto server = servers.begin(); server != servers.end(); server++) {
+  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
     GPR_ASSERT(server->stream->Read(&server_status));
   }
-  for (auto client = clients.begin(); client != clients.end(); client++) {
+  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
     GPR_ASSERT(client->stream->Read(&client_status));
   }
 
   // Wait some time
   gpr_log(GPR_INFO, "Running");
+  // Use gpr_sleep_until rather than this_thread::sleep_until to support
+  // compilers that don't work with this_thread
   gpr_sleep_until(gpr_time_add(
       start, gpr_time_from_seconds(benchmark_seconds, GPR_TIMESPAN)));
 
@@ -220,34 +234,36 @@ std::unique_ptr<ScenarioResult> RunScenario(
   result->client_config = result_client_config;
   result->server_config = result_server_config;
   gpr_log(GPR_INFO, "Finishing");
-  for (auto server = servers.begin(); server != servers.end(); server++) {
+  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
     GPR_ASSERT(server->stream->Write(server_mark));
   }
-  for (auto client = clients.begin(); client != clients.end(); client++) {
+  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
     GPR_ASSERT(client->stream->Write(client_mark));
   }
-  for (auto server = servers.begin(); server != servers.end(); server++) {
+  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
     GPR_ASSERT(server->stream->Read(&server_status));
     const auto& stats = server_status.stats();
-    result->server_resources.push_back(ResourceUsage{
-        stats.time_elapsed(), stats.time_user(), stats.time_system()});
+    result->server_resources.emplace_back(
+        stats.time_elapsed(), stats.time_user(), stats.time_system());
   }
-  for (auto client = clients.begin(); client != clients.end(); client++) {
+  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
     GPR_ASSERT(client->stream->Read(&client_status));
     const auto& stats = client_status.stats();
     result->latencies.MergeProto(stats.latencies());
-    result->client_resources.push_back(ResourceUsage{
-        stats.time_elapsed(), stats.time_user(), stats.time_system()});
+    result->client_resources.emplace_back(
+        stats.time_elapsed(), stats.time_user(), stats.time_system());
   }
 
-  for (auto client = clients.begin(); client != clients.end(); client++) {
+  for (auto client = &clients[0]; client != &clients[num_clients]; client++) {
     GPR_ASSERT(client->stream->WritesDone());
     GPR_ASSERT(client->stream->Finish().ok());
   }
-  for (auto server = servers.begin(); server != servers.end(); server++) {
+  for (auto server = &servers[0]; server != &servers[num_servers]; server++) {
     GPR_ASSERT(server->stream->WritesDone());
     GPR_ASSERT(server->stream->Finish().ok());
   }
+  delete[] clients;
+  delete[] servers;
   return result;
 }
 }  // namespace testing

test/cpp/qps/driver.h

@@ -41,10 +41,18 @@
 
 namespace grpc {
 namespace testing {
-struct ResourceUsage {
-  double wall_time;
-  double user_time;
-  double system_time;
+class ResourceUsage {
+ public:
+  ResourceUsage(double w, double u, double s)
+      : wall_time_(w), user_time_(u), system_time_(s) {}
+  double wall_time() const { return wall_time_; }
+  double user_time() const { return user_time_; }
+  double system_time() const { return system_time_; }
+
+ private:
+  double wall_time_;
+  double user_time_;
+  double system_time_;
 };
 
 struct ScenarioResult {

test/cpp/qps/interarrival.h

@@ -36,7 +36,8 @@
 
 #include <chrono>
 #include <cmath>
-#include <random>
+#include <cstdlib>
+#include <vector>
 
 #include <grpc++/config.h>
 
@@ -141,17 +142,16 @@ class ParetoDist GRPC_FINAL : public RandomDist {
 // in an efficient re-entrant way. The random table is built at construction
 // time, and each call must include the thread id of the invoker
 
-typedef std::default_random_engine qps_random_engine;
-
 class InterarrivalTimer {
  public:
   InterarrivalTimer() {}
   void init(const RandomDist& r, int threads, int entries = 1000000) {
-    qps_random_engine gen;
-    std::uniform_real_distribution<double> uniform(0.0, 1.0);
     for (int i = 0; i < entries; i++) {
-      random_table_.push_back(std::chrono::nanoseconds(
-          static_cast<int64_t>(1e9 * r(uniform(gen)))));
+      // rand is the only choice that is portable across POSIX and Windows
+      // and that supports new and old compilers
+      const double uniform_0_1 = rand() / RAND_MAX;
+      random_table_.push_back(
+          std::chrono::nanoseconds(static_cast<int64_t>(1e9 * r(uniform_0_1))));
     }
     // Now set up the thread positions
     for (int i = 0; i < threads; i++) {

test/cpp/qps/qps_driver.cc

@@ -33,6 +33,7 @@
 
 #include <memory>
 #include <set>
+#include <signal.h>
 
 #include <gflags/gflags.h>
 #include <grpc/support/log.h>

test/cpp/qps/report.cc

@@ -34,11 +34,16 @@
 #include "test/cpp/qps/report.h"
 
 #include <grpc/support/log.h>
+#include "test/cpp/qps/driver.h"
 #include "test/cpp/qps/stats.h"
 
 namespace grpc {
 namespace testing {
 
+static double WallTime(ResourceUsage u) { return u.wall_time(); }
+static double UserTime(ResourceUsage u) { return u.user_time(); }
+static double SystemTime(ResourceUsage u) { return u.system_time(); }
+
 void CompositeReporter::add(std::unique_ptr<Reporter> reporter) {
   reporters_.emplace_back(std::move(reporter));
 }
@@ -68,16 +73,14 @@ void CompositeReporter::ReportTimes(const ScenarioResult& result) {
 }
 
 void GprLogReporter::ReportQPS(const ScenarioResult& result) {
-  gpr_log(GPR_INFO, "QPS: %.1f",
-          result.latencies.Count() /
-              average(result.client_resources,
-                      [](ResourceUsage u) { return u.wall_time; }));
+  gpr_log(
+      GPR_INFO, "QPS: %.1f",
+      result.latencies.Count() / average(result.client_resources, WallTime));
 }
 
 void GprLogReporter::ReportQPSPerCore(const ScenarioResult& result) {
-  auto qps = result.latencies.Count() /
-             average(result.client_resources,
-                     [](ResourceUsage u) { return u.wall_time; });
+  auto qps =
+      result.latencies.Count() / average(result.client_resources, WallTime);
 
   gpr_log(GPR_INFO, "QPS: %.1f (%.1f/server core)", qps,
           qps / result.server_config.threads());
@@ -95,40 +98,30 @@ void GprLogReporter::ReportLatency(const ScenarioResult& result) {
 
 void GprLogReporter::ReportTimes(const ScenarioResult& result) {
   gpr_log(GPR_INFO, "Server system time: %.2f%%",
-          100.0 * sum(result.server_resources,
-                      [](ResourceUsage u) { return u.system_time; }) /
-              sum(result.server_resources,
-                  [](ResourceUsage u) { return u.wall_time; }));
+          100.0 * sum(result.server_resources, SystemTime) /
+              sum(result.server_resources, WallTime));
   gpr_log(GPR_INFO, "Server user time:   %.2f%%",
-          100.0 * sum(result.server_resources,
-                      [](ResourceUsage u) { return u.user_time; }) /
-              sum(result.server_resources,
-                  [](ResourceUsage u) { return u.wall_time; }));
+          100.0 * sum(result.server_resources, UserTime) /
+              sum(result.server_resources, WallTime));
   gpr_log(GPR_INFO, "Client system time: %.2f%%",
-          100.0 * sum(result.client_resources,
-                      [](ResourceUsage u) { return u.system_time; }) /
-              sum(result.client_resources,
-                  [](ResourceUsage u) { return u.wall_time; }));
+          100.0 * sum(result.client_resources, SystemTime) /
+              sum(result.client_resources, WallTime));
   gpr_log(GPR_INFO, "Client user time:   %.2f%%",
-          100.0 * sum(result.client_resources,
-                      [](ResourceUsage u) { return u.user_time; }) /
-              sum(result.client_resources,
-                  [](ResourceUsage u) { return u.wall_time; }));
+          100.0 * sum(result.client_resources, UserTime) /
+              sum(result.client_resources, WallTime));
 }
 
 void PerfDbReporter::ReportQPS(const ScenarioResult& result) {
-  auto qps = result.latencies.Count() /
-             average(result.client_resources,
-                     [](ResourceUsage u) { return u.wall_time; });
+  auto qps =
+      result.latencies.Count() / average(result.client_resources, WallTime);
 
   perf_db_client_.setQps(qps);
   perf_db_client_.setConfigs(result.client_config, result.server_config);
 }
 
 void PerfDbReporter::ReportQPSPerCore(const ScenarioResult& result) {
-  auto qps = result.latencies.Count() /
-             average(result.client_resources,
-                     [](ResourceUsage u) { return u.wall_time; });
+  auto qps =
+      result.latencies.Count() / average(result.client_resources, WallTime);
 
   auto qpsPerCore = qps / result.server_config.threads();
 
@@ -139,33 +132,29 @@ void PerfDbReporter::ReportQPSPerCore(const ScenarioResult& result) {
 
 void PerfDbReporter::ReportLatency(const ScenarioResult& result) {
   perf_db_client_.setLatencies(result.latencies.Percentile(50) / 1000,
-                             result.latencies.Percentile(90) / 1000,
-                             result.latencies.Percentile(95) / 1000,
-                             result.latencies.Percentile(99) / 1000,
-                             result.latencies.Percentile(99.9) / 1000);
+                               result.latencies.Percentile(90) / 1000,
+                               result.latencies.Percentile(95) / 1000,
+                               result.latencies.Percentile(99) / 1000,
+                               result.latencies.Percentile(99.9) / 1000);
   perf_db_client_.setConfigs(result.client_config, result.server_config);
 }
 
 void PerfDbReporter::ReportTimes(const ScenarioResult& result) {
-  double server_system_time =
-      100.0 * sum(result.server_resources,
-                  [](ResourceUsage u) { return u.system_time; }) /
-      sum(result.server_resources, [](ResourceUsage u) { return u.wall_time; });
-  double server_user_time =
-      100.0 * sum(result.server_resources,
-                  [](ResourceUsage u) { return u.user_time; }) /
-      sum(result.server_resources, [](ResourceUsage u) { return u.wall_time; });
-  double client_system_time =
-      100.0 * sum(result.client_resources,
-                  [](ResourceUsage u) { return u.system_time; }) /
-      sum(result.client_resources, [](ResourceUsage u) { return u.wall_time; });
-  double client_user_time =
-      100.0 * sum(result.client_resources,
-                  [](ResourceUsage u) { return u.user_time; }) /
-      sum(result.client_resources, [](ResourceUsage u) { return u.wall_time; });
-
-  perf_db_client_.setTimes(server_system_time, server_user_time, client_system_time,
-                         client_user_time);
+  const double server_system_time = 100.0 *
+                                    sum(result.server_resources, SystemTime) /
+                                    sum(result.server_resources, WallTime);
+  const double server_user_time = 100.0 *
+                                  sum(result.server_resources, UserTime) /
+                                  sum(result.server_resources, WallTime);
+  const double client_system_time = 100.0 *
+                                    sum(result.client_resources, SystemTime) /
+                                    sum(result.client_resources, WallTime);
+  const double client_user_time = 100.0 *
+                                  sum(result.client_resources, UserTime) /
+                                  sum(result.client_resources, WallTime);
+
+  perf_db_client_.setTimes(server_system_time, server_user_time,
+                           client_system_time, client_user_time);
   perf_db_client_.setConfigs(result.client_config, result.server_config);
 }
 

test/cpp/qps/server_async.cc

@@ -99,25 +99,7 @@ class AsyncQpsServerTest : public Server {
       shutdown_state_.emplace_back(new PerThreadShutdownState());
     }
     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_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);
-          if (!shutdown_state_[i]->shutdown()) {
-            // this RPC context is done, so refresh it
-            if (!still_going) {
-              ctx->Reset();
-            }
-          } else {
-            return;
-          }
-        }
-        return;
-      }));
+      threads_.emplace_back(&AsyncQpsServerTest::ThreadFunc, this, i);
     }
   }
   ~AsyncQpsServerTest() {
@@ -142,6 +124,26 @@ class AsyncQpsServerTest : public Server {
   }
 
  private:
+  void ThreadFunc(int rank) {
+    // Wait until work is available or we are shutting down
+    bool ok;
+    void *got_tag;
+    while (srv_cqs_[rank]->Next(&got_tag, &ok)) {
+      ServerRpcContext *ctx = detag(got_tag);
+      // The tag is a pointer to an RPC context to invoke
+      const bool still_going = ctx->RunNextState(ok);
+      if (!shutdown_state_[rank]->shutdown()) {
+        // this RPC context is done, so refresh it
+        if (!still_going) {
+          ctx->Reset();
+        }
+      } else {
+        return;
+      }
+    }
+    return;
+  }
+
   class ServerRpcContext {
    public:
     ServerRpcContext() {}

tools/dockerfile/grpc_go/Dockerfile

@@ -32,6 +32,8 @@ FROM golang:1.4
 
 # Get the source from GitHub
 RUN go get google.golang.org/grpc
+RUN go get golang.org/x/oauth2
+RUN go get google.golang.org/cloud
 
 # Add a service_account directory containing the auth creds file
 ADD service_account service_account

tools/dockerfile/grpc_go/build.sh

@@ -0,0 +1,34 @@
+#!/bin/bash
+# Copyright 2015, 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.
+
+cp -R /var/local/git-clone/grpc-go/. /go/
+go get golang.org/x/oauth2
+go get google.golang.org/cloud
+cd src/google.golang.org/grpc/interop/client && go install