diff options
39 files changed, 677 insertions, 368 deletions
diff --git a/src/compiler/objective_c_generator.cc b/src/compiler/objective_c_generator.cc index 69b3805bb1..483c6573a8 100644 --- a/src/compiler/objective_c_generator.cc +++ b/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"); diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index 9a8195e9d0..055aff1444 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/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" /> diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs index 9fa0baca87..6ca4bbdafc 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs +++ b/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 { diff --git a/src/csharp/Grpc.Core/OperationFailedException.cs b/src/csharp/Grpc.Core/OperationFailedException.cs deleted file mode 100644 index 9b1c24d0c1..0000000000 --- a/src/csharp/Grpc.Core/OperationFailedException.cs +++ /dev/null @@ -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) - { - } - } -} diff --git a/src/node/binding.gyp b/src/node/binding.gyp index 6ba233388a..734dc8410b 100644 --- a/src/node/binding.gyp +++ b/src/node/binding.gyp @@ -11,7 +11,8 @@ '-pedantic', '-g', '-zdefs', - '-Werror' + '-Werror', + '-Wno-error=deprecated-declarations' ], 'ldflags': [ '-g' diff --git a/src/node/interop/interop_client.js b/src/node/interop/interop_client.js index 27f6c19c13..6d6f9a349e 100644 --- a/src/node/interop/interop_client.js +++ b/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); diff --git a/src/node/src/client.js b/src/node/src/client.js index 87c7690dc0..5cde438572 100644 --- a/src/node/src/client.js +++ b/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; }; diff --git a/src/objective-c/RxLibrary/GRXWriteable.h b/src/objective-c/RxLibrary/GRXWriteable.h index 216de30735..45613d6dd0 100644 --- a/src/objective-c/RxLibrary/GRXWriteable.h +++ b/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 diff --git a/src/objective-c/RxLibrary/GRXWriteable.m b/src/objective-c/RxLibrary/GRXWriteable.m index 63f7c3e7f3..2729d62b72 100644 --- a/src/objective-c/RxLibrary/GRXWriteable.m +++ b/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]; } diff --git a/src/objective-c/tests/RxLibraryUnitTests.m b/src/objective-c/tests/RxLibraryUnitTests.m index 5e3162875a..a67a4c6cd9 100644 --- a/src/objective-c/tests/RxLibraryUnitTests.m +++ b/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: diff --git a/src/python/grpcio/grpc/_adapter/_c/types.h b/src/python/grpcio/grpc/_adapter/_c/types.h index 4e0da4a28a..f646465c63 100644 --- a/src/python/grpcio/grpc/_adapter/_c/types.h +++ b/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); diff --git a/src/python/grpcio/grpc/_adapter/_c/types/call.c b/src/python/grpcio/grpc/_adapter/_c/types/call.c index 9cdd5de13d..42a50151f6 100644 --- a/src/python/grpcio/grpc/_adapter/_c/types/call.c +++ b/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; +} diff --git a/src/python/grpcio/grpc/_adapter/_c/types/channel.c b/src/python/grpcio/grpc/_adapter/_c/types/channel.c index 2a8477a334..c577ac05eb 100644 --- a/src/python/grpcio/grpc/_adapter/_c/types/channel.c +++ b/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; +} diff --git a/src/python/grpcio/grpc/_adapter/_c/utility.c b/src/python/grpcio/grpc/_adapter/_c/utility.c index 51f3c9be01..2eea0e18ef 100644 --- a/src/python/grpcio/grpc/_adapter/_c/utility.c +++ b/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( diff --git a/src/python/grpcio/grpc/_adapter/_intermediary_low.py b/src/python/grpcio/grpc/_adapter/_intermediary_low.py index 3c7f0a2619..e7bf9dc462 100644 --- a/src/python/grpcio/grpc/_adapter/_intermediary_low.py +++ b/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): diff --git a/src/python/grpcio/grpc/_adapter/_low.py b/src/python/grpcio/grpc/_adapter/_low.py index 239aac81b2..147086e725 100644 --- a/src/python/grpcio/grpc/_adapter/_low.py +++ b/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() diff --git a/src/python/grpcio/grpc/_adapter/_types.py b/src/python/grpcio/grpc/_adapter/_types.py index 5ddb1774ea..5470d2de4a 100644 --- a/src/python/grpcio/grpc/_adapter/_types.py +++ b/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 diff --git a/src/python/grpcio_test/grpc_test/_adapter/_low_test.py b/src/python/grpcio_test/grpc_test/_adapter/_low_test.py index b6583662f3..44fe760fbc 100644 --- a/src/python/grpcio_test/grpc_test/_adapter/_low_test.py +++ b/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) diff --git a/src/python/grpcio_test/grpc_test/framework/face/testing/blocking_invocation_inline_service_test_case.py b/src/python/grpcio_test/grpc_test/framework/face/testing/blocking_invocation_inline_service_test_case.py index 7e1158f96b..251e1eb68e 100644 --- a/src/python/grpcio_test/grpc_test/framework/face/testing/blocking_invocation_inline_service_test_case.py +++ b/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) diff --git a/src/python/grpcio_test/grpc_test/framework/face/testing/event_invocation_synchronous_event_service_test_case.py b/src/python/grpcio_test/grpc_test/framework/face/testing/event_invocation_synchronous_event_service_test_case.py index 18eed53d6e..9df77678eb 100644 --- a/src/python/grpcio_test/grpc_test/framework/face/testing/event_invocation_synchronous_event_service_test_case.py +++ b/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() diff --git a/src/python/grpcio_test/grpc_test/framework/face/testing/future_invocation_asynchronous_event_service_test_case.py b/src/python/grpcio_test/grpc_test/framework/face/testing/future_invocation_asynchronous_event_service_test_case.py index 3b42914342..70d86a0422 100644 --- a/src/python/grpcio_test/grpc_test/framework/face/testing/future_invocation_asynchronous_event_service_test_case.py +++ b/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): diff --git a/src/ruby/ext/grpc/rb_channel.c b/src/ruby/ext/grpc/rb_channel.c index 5e65bf72d5..7d62e9110c 100644 --- a/src/ruby/ext/grpc/rb_channel.c +++ b/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 */ diff --git a/src/ruby/grpc.gemspec b/src/ruby/grpc.gemspec index 45f31329e9..eb748458b9 100755 --- a/src/ruby/grpc.gemspec +++ b/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 diff --git a/src/ruby/lib/grpc/generic/client_stub.rb b/src/ruby/lib/grpc/generic/client_stub.rb index a2f1ec612d..cce718537c 100644 --- a/src/ruby/lib/grpc/generic/client_stub.rb +++ b/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 diff --git a/src/ruby/spec/call_spec.rb b/src/ruby/spec/call_spec.rb index 36a442faed..3c5d33ffcd 100644 --- a/src/ruby/spec/call_spec.rb +++ b/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 diff --git a/src/ruby/spec/channel_spec.rb b/src/ruby/spec/channel_spec.rb index 9081f0e20c..25cefcdfb7 100644 --- a/src/ruby/spec/channel_spec.rb +++ b/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 diff --git a/src/ruby/spec/client_server_spec.rb b/src/ruby/spec/client_server_spec.rb index 57c9a8de9b..2e673ff413 100644 --- a/src/ruby/spec/client_server_spec.rb +++ b/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 diff --git a/src/ruby/spec/generic/active_call_spec.rb b/src/ruby/spec/generic/active_call_spec.rb index 424b2dbdeb..0bf65ba2e9 100644 --- a/src/ruby/spec/generic/active_call_spec.rb +++ b/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 diff --git a/test/cpp/qps/client.h b/test/cpp/qps/client.h index 28cd32a197..1c4f46328f 100644 --- a/test/cpp/qps/client.h +++ b/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_; }; diff --git a/test/cpp/qps/client_async.cc b/test/cpp/qps/client_async.cc index e1e44f9ac0..a337610cbf 100644 --- a/test/cpp/qps/client_async.cc +++ b/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); } }; diff --git a/test/cpp/qps/client_sync.cc b/test/cpp/qps/client_sync.cc index 718698bfe1..db5416a707 100644 --- a/test/cpp/qps/client_sync.cc +++ b/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( diff --git a/test/cpp/qps/driver.cc b/test/cpp/qps/driver.cc index a0360295e0..78e3720938 100644 --- a/test/cpp/qps/driver.cc +++ b/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 diff --git a/test/cpp/qps/driver.h b/test/cpp/qps/driver.h index 5e9d4b3cb9..9a29df8d49 100644 --- a/test/cpp/qps/driver.h +++ b/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 { diff --git a/test/cpp/qps/interarrival.h b/test/cpp/qps/interarrival.h index f90a17a894..04d14f689f 100644 --- a/test/cpp/qps/interarrival.h +++ b/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++) { diff --git a/test/cpp/qps/qps_driver.cc b/test/cpp/qps/qps_driver.cc index d534846365..b1463be8f6 100644 --- a/test/cpp/qps/qps_driver.cc +++ b/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> diff --git a/test/cpp/qps/report.cc b/test/cpp/qps/report.cc index ff01ec1501..e03e8e1fb0 100644 --- a/test/cpp/qps/report.cc +++ b/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); } diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc index 33b6fa55c3..b4fc49c31c 100644 --- a/test/cpp/qps/server_async.cc +++ b/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() {} diff --git a/tools/dockerfile/grpc_go/Dockerfile b/tools/dockerfile/grpc_go/Dockerfile index 06bb3e2d5e..7f3bd8719c 100644 --- a/tools/dockerfile/grpc_go/Dockerfile +++ b/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 diff --git a/tools/dockerfile/grpc_go/build.sh b/tools/dockerfile/grpc_go/build.sh new file mode 100755 index 0000000000..254942fe01 --- /dev/null +++ b/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 |