mirror of https://github.com/grpc/grpc.git
Craig Tiller
9 years ago
commit
d925c93047
193 changed files with 9858 additions and 2289 deletions
@ -0,0 +1,132 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include <limits.h> |
||||
#include <string.h> |
||||
|
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/sync.h> |
||||
|
||||
#include "src/core/ext/load_reporting/load_reporting.h" |
||||
#include "src/core/ext/load_reporting/load_reporting_filter.h" |
||||
#include "src/core/lib/channel/channel_stack_builder.h" |
||||
#include "src/core/lib/surface/channel_init.h" |
||||
|
||||
struct grpc_load_reporting_config { |
||||
grpc_load_reporting_fn fn; |
||||
void *user_data; |
||||
}; |
||||
|
||||
grpc_load_reporting_config *grpc_load_reporting_config_create( |
||||
grpc_load_reporting_fn fn, void *user_data) { |
||||
GPR_ASSERT(fn != NULL); |
||||
grpc_load_reporting_config *lrc = |
||||
gpr_malloc(sizeof(grpc_load_reporting_config)); |
||||
lrc->fn = fn; |
||||
lrc->user_data = user_data; |
||||
return lrc; |
||||
} |
||||
|
||||
grpc_load_reporting_config *grpc_load_reporting_config_copy( |
||||
grpc_load_reporting_config *src) { |
||||
return grpc_load_reporting_config_create(src->fn, src->user_data); |
||||
} |
||||
|
||||
void grpc_load_reporting_config_destroy(grpc_load_reporting_config *lrc) { |
||||
gpr_free(lrc); |
||||
} |
||||
|
||||
void grpc_load_reporting_config_call( |
||||
grpc_load_reporting_config *lrc, |
||||
const grpc_load_reporting_call_data *call_data) { |
||||
lrc->fn(call_data, lrc->user_data); |
||||
} |
||||
|
||||
static bool is_load_reporting_enabled(const grpc_channel_args *a) { |
||||
if (a == NULL) return false; |
||||
for (size_t i = 0; i < a->num_args; i++) { |
||||
if (0 == strcmp(a->args[i].key, GRPC_ARG_ENABLE_LOAD_REPORTING)) { |
||||
return a->args[i].value.pointer.p != NULL; |
||||
} |
||||
} |
||||
return false; |
||||
} |
||||
|
||||
static bool maybe_add_load_reporting_filter(grpc_channel_stack_builder *builder, |
||||
void *arg) { |
||||
const grpc_channel_args *args = |
||||
grpc_channel_stack_builder_get_channel_arguments(builder); |
||||
if (is_load_reporting_enabled(args)) { |
||||
return grpc_channel_stack_builder_prepend_filter( |
||||
builder, (const grpc_channel_filter *)arg, NULL, NULL); |
||||
} |
||||
return true; |
||||
} |
||||
|
||||
static void lrd_arg_destroy(void *p) { grpc_load_reporting_config_destroy(p); } |
||||
|
||||
static void *lrd_arg_copy(void *p) { |
||||
return grpc_load_reporting_config_copy(p); |
||||
} |
||||
|
||||
static int lrd_arg_cmp(void *a, void *b) { |
||||
grpc_load_reporting_config *lhs = a; |
||||
grpc_load_reporting_config *rhs = b; |
||||
return !(lhs->fn == rhs->fn && lhs->user_data == rhs->user_data); |
||||
} |
||||
|
||||
static const grpc_arg_pointer_vtable lrd_ptr_vtable = { |
||||
lrd_arg_copy, lrd_arg_destroy, lrd_arg_cmp}; |
||||
|
||||
grpc_arg grpc_load_reporting_config_create_arg( |
||||
grpc_load_reporting_config *lrc) { |
||||
grpc_arg arg; |
||||
arg.type = GRPC_ARG_POINTER; |
||||
arg.key = GRPC_ARG_ENABLE_LOAD_REPORTING; |
||||
arg.value.pointer.p = lrc; |
||||
arg.value.pointer.vtable = &lrd_ptr_vtable; |
||||
return arg; |
||||
} |
||||
|
||||
/* Plugin registration */ |
||||
|
||||
void grpc_load_reporting_plugin_init(void) { |
||||
grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MAX, |
||||
maybe_add_load_reporting_filter, |
||||
(void *)&grpc_load_reporting_filter); |
||||
grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, INT_MAX, |
||||
maybe_add_load_reporting_filter, |
||||
(void *)&grpc_load_reporting_filter); |
||||
} |
||||
|
||||
void grpc_load_reporting_plugin_shutdown() {} |
||||
@ -0,0 +1,75 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_H |
||||
#define GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_H |
||||
|
||||
#include "src/core/lib/iomgr/closure.h" |
||||
#include "src/core/lib/surface/call.h" |
||||
|
||||
typedef struct grpc_load_reporting_config grpc_load_reporting_config; |
||||
|
||||
/** Call information to be passed to the provided load reporting function upon
|
||||
* completion of the call */ |
||||
typedef struct grpc_load_reporting_call_data { |
||||
const grpc_call_stats *stats; /**< Stats for the call */ |
||||
const char *trailing_md_string; /**< LR trailing metadata info */ |
||||
} grpc_load_reporting_call_data; |
||||
|
||||
/** Custom function to be called by the load reporting filter. */ |
||||
typedef void (*grpc_load_reporting_fn)( |
||||
const grpc_load_reporting_call_data *call_data, void *user_data); |
||||
|
||||
/** Register \a fn as the function to be invoked by the load reporting filter.
|
||||
* \a fn will be invoked at the beginning and at the end of the call. |
||||
* |
||||
* For the first invocation, \a fn's first argument |
||||
* (grpc_load_reporting_call_data*) will be NULL. \a user_data is always passed |
||||
* as-is. */ |
||||
grpc_load_reporting_config *grpc_load_reporting_config_create( |
||||
grpc_load_reporting_fn fn, void *user_data); |
||||
|
||||
grpc_load_reporting_config *grpc_load_reporting_config_copy( |
||||
grpc_load_reporting_config *src); |
||||
|
||||
void grpc_load_reporting_config_destroy(grpc_load_reporting_config *lrc); |
||||
|
||||
/** Invoke the function registered by \a grpc_load_reporting_init. */ |
||||
void grpc_load_reporting_config_call( |
||||
grpc_load_reporting_config *lrc, |
||||
const grpc_load_reporting_call_data *call_data); |
||||
|
||||
/** Return a \a grpc_arg enabling load reporting */ |
||||
grpc_arg grpc_load_reporting_config_create_arg(grpc_load_reporting_config *lrc); |
||||
|
||||
#endif /* GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_H */ |
||||
@ -0,0 +1,151 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
#include <grpc/support/sync.h> |
||||
#include <string.h> |
||||
|
||||
#include "src/core/ext/load_reporting/load_reporting.h" |
||||
#include "src/core/ext/load_reporting/load_reporting_filter.h" |
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/profiling/timers.h" |
||||
#include "src/core/lib/transport/static_metadata.h" |
||||
|
||||
typedef struct call_data { const char *trailing_md_string; } call_data; |
||||
typedef struct channel_data { |
||||
gpr_mu mu; |
||||
grpc_load_reporting_config *lrc; |
||||
} channel_data; |
||||
|
||||
static void invoke_lr_fn_locked(grpc_load_reporting_config *lrc, |
||||
grpc_load_reporting_call_data *lr_call_data) { |
||||
GPR_TIMER_BEGIN("load_reporting_config_fn", 0); |
||||
grpc_load_reporting_config_call(lrc, lr_call_data); |
||||
GPR_TIMER_END("load_reporting_config_fn", 0); |
||||
} |
||||
|
||||
/* Constructor for call_data */ |
||||
static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, |
||||
grpc_call_element_args *args) { |
||||
call_data *calld = elem->call_data; |
||||
memset(calld, 0, sizeof(call_data)); |
||||
} |
||||
|
||||
/* Destructor for call_data */ |
||||
static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, |
||||
const grpc_call_stats *stats, void *ignored) { |
||||
channel_data *chand = elem->channel_data; |
||||
call_data *calld = elem->call_data; |
||||
|
||||
grpc_load_reporting_call_data lr_call_data = {stats, |
||||
calld->trailing_md_string}; |
||||
|
||||
gpr_mu_lock(&chand->mu); |
||||
invoke_lr_fn_locked(chand->lrc, &lr_call_data); |
||||
gpr_mu_unlock(&chand->mu); |
||||
} |
||||
|
||||
/* Constructor for channel_data */ |
||||
static void init_channel_elem(grpc_exec_ctx *exec_ctx, |
||||
grpc_channel_element *elem, |
||||
grpc_channel_element_args *args) { |
||||
GPR_ASSERT(!args->is_last); |
||||
|
||||
channel_data *chand = elem->channel_data; |
||||
memset(chand, 0, sizeof(channel_data)); |
||||
|
||||
gpr_mu_init(&chand->mu); |
||||
for (size_t i = 0; i < args->channel_args->num_args; i++) { |
||||
if (0 == strcmp(args->channel_args->args[i].key, |
||||
GRPC_ARG_ENABLE_LOAD_REPORTING)) { |
||||
grpc_load_reporting_config *arg_lrc = |
||||
args->channel_args->args[i].value.pointer.p; |
||||
chand->lrc = grpc_load_reporting_config_copy(arg_lrc); |
||||
GPR_ASSERT(chand->lrc != NULL); |
||||
break; |
||||
} |
||||
} |
||||
GPR_ASSERT(chand->lrc != NULL); /* arg actually found */ |
||||
|
||||
gpr_mu_lock(&chand->mu); |
||||
invoke_lr_fn_locked(chand->lrc, NULL); |
||||
gpr_mu_unlock(&chand->mu); |
||||
} |
||||
|
||||
/* Destructor for channel data */ |
||||
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, |
||||
grpc_channel_element *elem) { |
||||
channel_data *chand = elem->channel_data; |
||||
gpr_mu_destroy(&chand->mu); |
||||
grpc_load_reporting_config_destroy(chand->lrc); |
||||
} |
||||
|
||||
static grpc_mdelem *lr_trailing_md_filter(void *user_data, grpc_mdelem *md) { |
||||
grpc_call_element *elem = user_data; |
||||
call_data *calld = elem->call_data; |
||||
|
||||
if (md->key == GRPC_MDSTR_LOAD_REPORTING) { |
||||
calld->trailing_md_string = gpr_strdup(grpc_mdstr_as_c_string(md->value)); |
||||
return NULL; |
||||
} |
||||
|
||||
return md; |
||||
} |
||||
|
||||
static void lr_start_transport_stream_op(grpc_exec_ctx *exec_ctx, |
||||
grpc_call_element *elem, |
||||
grpc_transport_stream_op *op) { |
||||
GPR_TIMER_BEGIN("lr_start_transport_stream_op", 0); |
||||
|
||||
if (op->send_trailing_metadata) { |
||||
grpc_metadata_batch_filter(op->send_trailing_metadata, |
||||
lr_trailing_md_filter, elem); |
||||
} |
||||
grpc_call_next_op(exec_ctx, elem, op); |
||||
|
||||
GPR_TIMER_END("lr_start_transport_stream_op", 0); |
||||
} |
||||
|
||||
const grpc_channel_filter grpc_load_reporting_filter = { |
||||
lr_start_transport_stream_op, |
||||
grpc_channel_next_op, |
||||
sizeof(call_data), |
||||
init_call_elem, |
||||
grpc_call_stack_ignore_set_pollset, |
||||
destroy_call_elem, |
||||
sizeof(channel_data), |
||||
init_channel_elem, |
||||
destroy_channel_elem, |
||||
grpc_call_next_get_peer, |
||||
"load_reporting"}; |
||||
@ -0,0 +1,41 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2016, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_FILTER_H |
||||
#define GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_FILTER_H |
||||
|
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
|
||||
extern const grpc_channel_filter grpc_load_reporting_filter; |
||||
|
||||
#endif /* GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_FILTER_H */ |
||||
@ -1,69 +0,0 @@ |
||||
// 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. |
||||
|
||||
syntax = "proto3"; |
||||
|
||||
import "src/proto/grpc/testing/control.proto"; |
||||
|
||||
package grpc.testing; |
||||
|
||||
service PerfDbTransfer { |
||||
// Sends client info |
||||
rpc RecordSingleClientData(SingleUserRecordRequest) |
||||
returns (SingleUserRecordReply) {} |
||||
} |
||||
|
||||
// Metrics to be stored |
||||
message Metrics { |
||||
double qps = 1; |
||||
double qps_per_core = 2; |
||||
double perc_lat_50 = 3; |
||||
double perc_lat_90 = 4; |
||||
double perc_lat_95 = 5; |
||||
double perc_lat_99 = 6; |
||||
double perc_lat_99_point_9 = 7; |
||||
double server_system_time = 8; |
||||
double server_user_time = 9; |
||||
double client_system_time = 10; |
||||
double client_user_time = 11; |
||||
} |
||||
|
||||
// Request for storing a single user's data |
||||
message SingleUserRecordRequest { |
||||
string hashed_id = 1; |
||||
string test_name = 2; |
||||
string sys_info = 3; |
||||
string tag = 4; |
||||
Metrics metrics = 5; |
||||
ClientConfig client_config = 6; |
||||
ServerConfig server_config = 7; |
||||
} |
||||
|
||||
// Reply to request for storing single user's data |
||||
message SingleUserRecordReply {} |
||||
@ -0,0 +1,852 @@ |
||||
# Copyright 2016, Google Inc. |
||||
# All rights reserved. |
||||
# |
||||
# Redistribution and use in source and binary forms, with or without |
||||
# modification, are permitted provided that the following conditions are |
||||
# met: |
||||
# |
||||
# * Redistributions of source code must retain the above copyright |
||||
# notice, this list of conditions and the following disclaimer. |
||||
# * Redistributions in binary form must reproduce the above |
||||
# copyright notice, this list of conditions and the following disclaimer |
||||
# in the documentation and/or other materials provided with the |
||||
# distribution. |
||||
# * Neither the name of Google Inc. nor the names of its |
||||
# contributors may be used to endorse or promote products derived from |
||||
# this software without specific prior written permission. |
||||
# |
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
|
||||
"""Invocation-side implementation of gRPC Python.""" |
||||
|
||||
import sys |
||||
import threading |
||||
import time |
||||
|
||||
import grpc |
||||
from grpc import _common |
||||
from grpc import _grpcio_metadata |
||||
from grpc.framework.foundation import callable_util |
||||
from grpc._cython import cygrpc |
||||
|
||||
_USER_AGENT = 'Python-gRPC-{}'.format(_grpcio_metadata.__version__) |
||||
|
||||
_EMPTY_FLAGS = 0 |
||||
_INFINITE_FUTURE = cygrpc.Timespec(float('+inf')) |
||||
_EMPTY_METADATA = cygrpc.Metadata(()) |
||||
|
||||
_UNARY_UNARY_INITIAL_DUE = ( |
||||
cygrpc.OperationType.send_initial_metadata, |
||||
cygrpc.OperationType.send_message, |
||||
cygrpc.OperationType.send_close_from_client, |
||||
cygrpc.OperationType.receive_initial_metadata, |
||||
cygrpc.OperationType.receive_message, |
||||
cygrpc.OperationType.receive_status_on_client, |
||||
) |
||||
_UNARY_STREAM_INITIAL_DUE = ( |
||||
cygrpc.OperationType.send_initial_metadata, |
||||
cygrpc.OperationType.send_message, |
||||
cygrpc.OperationType.send_close_from_client, |
||||
cygrpc.OperationType.receive_initial_metadata, |
||||
cygrpc.OperationType.receive_status_on_client, |
||||
) |
||||
_STREAM_UNARY_INITIAL_DUE = ( |
||||
cygrpc.OperationType.send_initial_metadata, |
||||
cygrpc.OperationType.receive_initial_metadata, |
||||
cygrpc.OperationType.receive_message, |
||||
cygrpc.OperationType.receive_status_on_client, |
||||
) |
||||
_STREAM_STREAM_INITIAL_DUE = ( |
||||
cygrpc.OperationType.send_initial_metadata, |
||||
cygrpc.OperationType.receive_initial_metadata, |
||||
cygrpc.OperationType.receive_status_on_client, |
||||
) |
||||
|
||||
_CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE = ( |
||||
'Exception calling channel subscription callback!') |
||||
|
||||
|
||||
def _deadline(timeout): |
||||
if timeout is None: |
||||
return None, _INFINITE_FUTURE |
||||
else: |
||||
deadline = time.time() + timeout |
||||
return deadline, cygrpc.Timespec(deadline) |
||||
|
||||
|
||||
def _unknown_code_details(unknown_cygrpc_code, details): |
||||
return b'Server sent unknown code {} and details "{}"'.format( |
||||
unknown_cygrpc_code, details) |
||||
|
||||
|
||||
def _wait_once_until(condition, until): |
||||
if until is None: |
||||
condition.wait() |
||||
else: |
||||
remaining = until - time.time() |
||||
if remaining < 0: |
||||
raise grpc.FutureTimeoutError() |
||||
else: |
||||
condition.wait(timeout=remaining) |
||||
|
||||
|
||||
class _RPCState(object): |
||||
|
||||
def __init__(self, due, initial_metadata, trailing_metadata, code, details): |
||||
self.condition = threading.Condition() |
||||
# The cygrpc.OperationType objects representing events due from the RPC's |
||||
# completion queue. |
||||
self.due = set(due) |
||||
self.initial_metadata = initial_metadata |
||||
self.response = None |
||||
self.trailing_metadata = trailing_metadata |
||||
self.code = code |
||||
self.details = details |
||||
# The semantics of grpc.Future.cancel and grpc.Future.cancelled are |
||||
# slightly wonky, so they have to be tracked separately from the rest of the |
||||
# result of the RPC. This field tracks whether cancellation was requested |
||||
# prior to termination of the RPC. |
||||
self.cancelled = False |
||||
self.callbacks = [] |
||||
|
||||
|
||||
def _abort(state, code, details): |
||||
if state.code is None: |
||||
state.code = code |
||||
state.details = details |
||||
if state.initial_metadata is None: |
||||
state.initial_metadata = _EMPTY_METADATA |
||||
state.trailing_metadata = _EMPTY_METADATA |
||||
|
||||
|
||||
def _handle_event(event, state, response_deserializer): |
||||
callbacks = [] |
||||
for batch_operation in event.batch_operations: |
||||
operation_type = batch_operation.type |
||||
state.due.remove(operation_type) |
||||
if operation_type is cygrpc.OperationType.receive_initial_metadata: |
||||
state.initial_metadata = batch_operation.received_metadata |
||||
elif operation_type is cygrpc.OperationType.receive_message: |
||||
serialized_response = batch_operation.received_message.bytes() |
||||
if serialized_response is not None: |
||||
response = _common.deserialize( |
||||
serialized_response, response_deserializer) |
||||
if response is None: |
||||
details = b'Exception deserializing response!' |
||||
_abort(state, grpc.StatusCode.INTERNAL, details) |
||||
else: |
||||
state.response = response |
||||
elif operation_type is cygrpc.OperationType.receive_status_on_client: |
||||
state.trailing_metadata = batch_operation.received_metadata |
||||
if state.code is None: |
||||
code = _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE.get( |
||||
batch_operation.received_status_code) |
||||
if code is None: |
||||
state.code = grpc.StatusCode.UNKNOWN |
||||
state.details = _unknown_code_details( |
||||
batch_operation.received_status_code, |
||||
batch_operation.received_status_details) |
||||
else: |
||||
state.code = code |
||||
state.details = batch_operation.received_status_details |
||||
callbacks.extend(state.callbacks) |
||||
state.callbacks = None |
||||
return callbacks |
||||
|
||||
|
||||
def _event_handler(state, call, response_deserializer): |
||||
def handle_event(event): |
||||
with state.condition: |
||||
callbacks = _handle_event(event, state, response_deserializer) |
||||
state.condition.notify_all() |
||||
done = not state.due |
||||
for callback in callbacks: |
||||
callback() |
||||
return call if done else None |
||||
return handle_event |
||||
|
||||
|
||||
def _consume_request_iterator( |
||||
request_iterator, state, call, request_serializer): |
||||
event_handler = _event_handler(state, call, None) |
||||
def consume_request_iterator(): |
||||
for request in request_iterator: |
||||
serialized_request = _common.serialize(request, request_serializer) |
||||
with state.condition: |
||||
if state.code is None and not state.cancelled: |
||||
if serialized_request is None: |
||||
call.cancel() |
||||
details = b'Exception serializing request!' |
||||
_abort(state, grpc.StatusCode.INTERNAL, details) |
||||
return |
||||
else: |
||||
operations = ( |
||||
cygrpc.operation_send_message( |
||||
serialized_request, _EMPTY_FLAGS), |
||||
) |
||||
call.start_batch(cygrpc.Operations(operations), event_handler) |
||||
state.due.add(cygrpc.OperationType.send_message) |
||||
while True: |
||||
state.condition.wait() |
||||
if state.code is None: |
||||
if cygrpc.OperationType.send_message not in state.due: |
||||
break |
||||
else: |
||||
return |
||||
else: |
||||
return |
||||
with state.condition: |
||||
if state.code is None: |
||||
operations = ( |
||||
cygrpc.operation_send_close_from_client(_EMPTY_FLAGS), |
||||
) |
||||
call.start_batch(cygrpc.Operations(operations), event_handler) |
||||
state.due.add(cygrpc.OperationType.send_close_from_client) |
||||
thread = threading.Thread(target=consume_request_iterator) |
||||
thread.start() |
||||
|
||||
|
||||
class _Rendezvous(grpc.RpcError, grpc.Future, grpc.Call): |
||||
|
||||
def __init__(self, state, call, response_deserializer, deadline): |
||||
super(_Rendezvous, self).__init__() |
||||
self._state = state |
||||
self._call = call |
||||
self._response_deserializer = response_deserializer |
||||
self._deadline = deadline |
||||
|
||||
def cancel(self): |
||||
with self._state.condition: |
||||
if self._state.code is None: |
||||
self._call.cancel() |
||||
self._state.cancelled = True |
||||
_abort(self._state, grpc.StatusCode.CANCELLED, b'Cancelled!') |
||||
self._state.condition.notify_all() |
||||
return False |
||||
|
||||
def cancelled(self): |
||||
with self._state.condition: |
||||
return self._state.cancelled |
||||
|
||||
def running(self): |
||||
with self._state.condition: |
||||
return self._state.code is None |
||||
|
||||
def done(self): |
||||
with self._state.condition: |
||||
return self._state.code is not None |
||||
|
||||
def result(self, timeout=None): |
||||
until = None if timeout is None else time.time() + timeout |
||||
with self._state.condition: |
||||
while True: |
||||
if self._state.code is None: |
||||
_wait_once_until(self._state.condition, until) |
||||
elif self._state.code is grpc.StatusCode.OK: |
||||
return self._state.response |
||||
elif self._state.cancelled: |
||||
raise grpc.FutureCancelledError() |
||||
else: |
||||
raise self |
||||
|
||||
def exception(self, timeout=None): |
||||
until = None if timeout is None else time.time() + timeout |
||||
with self._state.condition: |
||||
while True: |
||||
if self._state.code is None: |
||||
_wait_once_until(self._state.condition, until) |
||||
elif self._state.code is grpc.StatusCode.OK: |
||||
return None |
||||
elif self._state.cancelled: |
||||
raise grpc.FutureCancelledError() |
||||
else: |
||||
return self |
||||
|
||||
def traceback(self, timeout=None): |
||||
until = None if timeout is None else time.time() + timeout |
||||
with self._state.condition: |
||||
while True: |
||||
if self._state.code is None: |
||||
_wait_once_until(self._state.condition, until) |
||||
elif self._state.code is grpc.StatusCode.OK: |
||||
return None |
||||
elif self._state.cancelled: |
||||
raise grpc.FutureCancelledError() |
||||
else: |
||||
try: |
||||
raise self |
||||
except grpc.RpcError: |
||||
return sys.exc_info()[2] |
||||
|
||||
def add_done_callback(self, fn): |
||||
with self._state.condition: |
||||
if self._state.code is None: |
||||
self._state.callbacks.append(lambda: fn(self)) |
||||
return |
||||
|
||||
fn(self) |
||||
|
||||
def _next(self): |
||||
with self._state.condition: |
||||
if self._state.code is None: |
||||
event_handler = _event_handler( |
||||
self._state, self._call, self._response_deserializer) |
||||
self._call.start_batch( |
||||
cygrpc.Operations( |
||||
(cygrpc.operation_receive_message(_EMPTY_FLAGS),)), |
||||
event_handler) |
||||
self._state.due.add(cygrpc.OperationType.receive_message) |
||||
elif self._state.code is grpc.StatusCode.OK: |
||||
raise StopIteration() |
||||
else: |
||||
raise self |
||||
while True: |
||||
self._state.condition.wait() |
||||
if self._state.response is not None: |
||||
response = self._state.response |
||||
self._state.response = None |
||||
return response |
||||
elif cygrpc.OperationType.receive_message not in self._state.due: |
||||
if self._state.code is grpc.StatusCode.OK: |
||||
raise StopIteration() |
||||
elif self._state.code is not None: |
||||
raise self |
||||
|
||||
def __iter__(self): |
||||
return self |
||||
|
||||
def __next__(self): |
||||
return self._next() |
||||
|
||||
def next(self): |
||||
return self._next() |
||||
|
||||
def is_active(self): |
||||
with self._state.condition: |
||||
return self._state.code is None |
||||
|
||||
def time_remaining(self): |
||||
if self._deadline is None: |
||||
return None |
||||
else: |
||||
return max(self._deadline - time.time(), 0) |
||||
|
||||
def add_cancellation_callback(self, callback): |
||||
with self._state.condition: |
||||
if self._state.callbacks is None: |
||||
return False |
||||
else: |
||||
self._state.callbacks.append(lambda unused_future: callback()) |
||||
return True |
||||
|
||||
def initial_metadata(self): |
||||
with self._state.condition: |
||||
while self._state.initial_metadata is None: |
||||
self._state.condition.wait() |
||||
return self._state.initial_metadata |
||||
|
||||
def trailing_metadata(self): |
||||
with self._state.condition: |
||||
while self._state.trailing_metadata is None: |
||||
self._state.condition.wait() |
||||
return self._state.trailing_metadata |
||||
|
||||
def code(self): |
||||
with self._state.condition: |
||||
while self._state.code is None: |
||||
self._state.condition.wait() |
||||
return self._state.code |
||||
|
||||
def details(self): |
||||
with self._state.condition: |
||||
while self._state.details is None: |
||||
self._state.condition.wait() |
||||
return self._state.details |
||||
|
||||
def _repr(self): |
||||
with self._state.condition: |
||||
if self._state.code is None: |
||||
return '<_Rendezvous object of in-flight RPC>' |
||||
else: |
||||
return '<_Rendezvous of RPC that terminated with ({}, {})>'.format( |
||||
self._state.code, self._state.details) |
||||
|
||||
def __repr__(self): |
||||
return self._repr() |
||||
|
||||
def __str__(self): |
||||
return self._repr() |
||||
|
||||
def __del__(self): |
||||
with self._state.condition: |
||||
if self._state.code is None: |
||||
self._call.cancel() |
||||
self._state.cancelled = True |
||||
self._state.code = grpc.StatusCode.CANCELLED |
||||
self._state.condition.notify_all() |
||||
|
||||
|
||||
def _start_unary_request(request, timeout, request_serializer): |
||||
deadline, deadline_timespec = _deadline(timeout) |
||||
serialized_request = _common.serialize(request, request_serializer) |
||||
if serialized_request is None: |
||||
state = _RPCState( |
||||
(), _EMPTY_METADATA, _EMPTY_METADATA, grpc.StatusCode.INTERNAL, |
||||
b'Exception serializing request!') |
||||
rendezvous = _Rendezvous(state, None, None, deadline) |
||||
return deadline, deadline_timespec, None, rendezvous |
||||
else: |
||||
return deadline, deadline_timespec, serialized_request, None |
||||
|
||||
|
||||
def _end_unary_response_blocking(state, with_call, deadline): |
||||
if state.code is grpc.StatusCode.OK: |
||||
if with_call: |
||||
rendezvous = _Rendezvous(state, None, None, deadline) |
||||
return state.response, rendezvous |
||||
else: |
||||
return state.response |
||||
else: |
||||
raise _Rendezvous(state, None, None, deadline) |
||||
|
||||
|
||||
class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): |
||||
|
||||
def __init__( |
||||
self, channel, create_managed_call, method, request_serializer, |
||||
response_deserializer): |
||||
self._channel = channel |
||||
self._create_managed_call = create_managed_call |
||||
self._method = method |
||||
self._request_serializer = request_serializer |
||||
self._response_deserializer = response_deserializer |
||||
|
||||
def _prepare(self, request, timeout, metadata): |
||||
deadline, deadline_timespec, serialized_request, rendezvous = ( |
||||
_start_unary_request(request, timeout, self._request_serializer)) |
||||
if serialized_request is None: |
||||
return None, None, None, None, rendezvous |
||||
else: |
||||
state = _RPCState(_UNARY_UNARY_INITIAL_DUE, None, None, None, None) |
||||
operations = ( |
||||
cygrpc.operation_send_initial_metadata( |
||||
_common.metadata(metadata), _EMPTY_FLAGS), |
||||
cygrpc.operation_send_message(serialized_request, _EMPTY_FLAGS), |
||||
cygrpc.operation_send_close_from_client(_EMPTY_FLAGS), |
||||
cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS), |
||||
cygrpc.operation_receive_message(_EMPTY_FLAGS), |
||||
cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), |
||||
) |
||||
return state, operations, deadline, deadline_timespec, None |
||||
|
||||
def __call__( |
||||
self, request, timeout=None, metadata=None, credentials=None, |
||||
with_call=False): |
||||
state, operations, deadline, deadline_timespec, rendezvous = self._prepare( |
||||
request, timeout, metadata) |
||||
if rendezvous: |
||||
raise rendezvous |
||||
else: |
||||
completion_queue = cygrpc.CompletionQueue() |
||||
call = self._channel.create_call( |
||||
None, 0, completion_queue, self._method, None, deadline_timespec) |
||||
if credentials is not None: |
||||
call.set_credentials(credentials._credentials) |
||||
call.start_batch(cygrpc.Operations(operations), None) |
||||
_handle_event(completion_queue.poll(), state, self._response_deserializer) |
||||
return _end_unary_response_blocking(state, with_call, deadline) |
||||
|
||||
def future(self, request, timeout=None, metadata=None, credentials=None): |
||||
state, operations, deadline, deadline_timespec, rendezvous = self._prepare( |
||||
request, timeout, metadata) |
||||
if rendezvous: |
||||
return rendezvous |
||||
else: |
||||
call = self._create_managed_call( |
||||
None, 0, self._method, None, deadline_timespec) |
||||
if credentials is not None: |
||||
call.set_credentials(credentials._credentials) |
||||
event_handler = _event_handler(state, call, self._response_deserializer) |
||||
with state.condition: |
||||
call.start_batch(cygrpc.Operations(operations), event_handler) |
||||
return _Rendezvous(state, call, self._response_deserializer, deadline) |
||||
|
||||
|
||||
class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): |
||||
|
||||
def __init__( |
||||
self, channel, create_managed_call, method, request_serializer, |
||||
response_deserializer): |
||||
self._channel = channel |
||||
self._create_managed_call = create_managed_call |
||||
self._method = method |
||||
self._request_serializer = request_serializer |
||||
self._response_deserializer = response_deserializer |
||||
|
||||
def __call__(self, request, timeout=None, metadata=None, credentials=None): |
||||
deadline, deadline_timespec, serialized_request, rendezvous = ( |
||||
_start_unary_request(request, timeout, self._request_serializer)) |
||||
if serialized_request is None: |
||||
raise rendezvous |
||||
else: |
||||
state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None) |
||||
call = self._create_managed_call( |
||||
None, 0, self._method, None, deadline_timespec) |
||||
if credentials is not None: |
||||
call.set_credentials(credentials._credentials) |
||||
event_handler = _event_handler(state, call, self._response_deserializer) |
||||
with state.condition: |
||||
call.start_batch( |
||||
cygrpc.Operations( |
||||
(cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),)), |
||||
event_handler) |
||||
operations = ( |
||||
cygrpc.operation_send_initial_metadata( |
||||
_common.metadata(metadata), _EMPTY_FLAGS), |
||||
cygrpc.operation_send_message(serialized_request, _EMPTY_FLAGS), |
||||
cygrpc.operation_send_close_from_client(_EMPTY_FLAGS), |
||||
cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), |
||||
) |
||||
call.start_batch(cygrpc.Operations(operations), event_handler) |
||||
return _Rendezvous(state, call, self._response_deserializer, deadline) |
||||
|
||||
|
||||
class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): |
||||
|
||||
def __init__( |
||||
self, channel, create_managed_call, method, request_serializer, |
||||
response_deserializer): |
||||
self._channel = channel |
||||
self._create_managed_call = create_managed_call |
||||
self._method = method |
||||
self._request_serializer = request_serializer |
||||
self._response_deserializer = response_deserializer |
||||
|
||||
def __call__( |
||||
self, request_iterator, timeout=None, metadata=None, credentials=None, |
||||
with_call=False): |
||||
deadline, deadline_timespec = _deadline(timeout) |
||||
state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) |
||||
completion_queue = cygrpc.CompletionQueue() |
||||
call = self._channel.create_call( |
||||
None, 0, completion_queue, self._method, None, deadline_timespec) |
||||
if credentials is not None: |
||||
call.set_credentials(credentials._credentials) |
||||
with state.condition: |
||||
call.start_batch( |
||||
cygrpc.Operations( |
||||
(cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),)), |
||||
None) |
||||
operations = ( |
||||
cygrpc.operation_send_initial_metadata( |
||||
_common.metadata(metadata), _EMPTY_FLAGS), |
||||
cygrpc.operation_receive_message(_EMPTY_FLAGS), |
||||
cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), |
||||
) |
||||
call.start_batch(cygrpc.Operations(operations), None) |
||||
_consume_request_iterator( |
||||
request_iterator, state, call, self._request_serializer) |
||||
while True: |
||||
event = completion_queue.poll() |
||||
with state.condition: |
||||
_handle_event(event, state, self._response_deserializer) |
||||
state.condition.notify_all() |
||||
if not state.due: |
||||
break |
||||
return _end_unary_response_blocking(state, with_call, deadline) |
||||
|
||||
def future( |
||||
self, request_iterator, timeout=None, metadata=None, credentials=None): |
||||
deadline, deadline_timespec = _deadline(timeout) |
||||
state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) |
||||
call = self._create_managed_call( |
||||
None, 0, self._method, None, deadline_timespec) |
||||
if credentials is not None: |
||||
call.set_credentials(credentials._credentials) |
||||
event_handler = _event_handler(state, call, self._response_deserializer) |
||||
with state.condition: |
||||
call.start_batch( |
||||
cygrpc.Operations( |
||||
(cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),)), |
||||
event_handler) |
||||
operations = ( |
||||
cygrpc.operation_send_initial_metadata( |
||||
_common.metadata(metadata), _EMPTY_FLAGS), |
||||
cygrpc.operation_receive_message(_EMPTY_FLAGS), |
||||
cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), |
||||
) |
||||
call.start_batch(cygrpc.Operations(operations), event_handler) |
||||
_consume_request_iterator( |
||||
request_iterator, state, call, self._request_serializer) |
||||
return _Rendezvous(state, call, self._response_deserializer, deadline) |
||||
|
||||
|
||||
class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): |
||||
|
||||
def __init__( |
||||
self, channel, create_managed_call, method, request_serializer, |
||||
response_deserializer): |
||||
self._channel = channel |
||||
self._create_managed_call = create_managed_call |
||||
self._method = method |
||||
self._request_serializer = request_serializer |
||||
self._response_deserializer = response_deserializer |
||||
|
||||
def __call__( |
||||
self, request_iterator, timeout=None, metadata=None, credentials=None): |
||||
deadline, deadline_timespec = _deadline(timeout) |
||||
state = _RPCState(_STREAM_STREAM_INITIAL_DUE, None, None, None, None) |
||||
call = self._create_managed_call( |
||||
None, 0, self._method, None, deadline_timespec) |
||||
if credentials is not None: |
||||
call.set_credentials(credentials._credentials) |
||||
event_handler = _event_handler(state, call, self._response_deserializer) |
||||
with state.condition: |
||||
call.start_batch( |
||||
cygrpc.Operations( |
||||
(cygrpc.operation_receive_initial_metadata(_EMPTY_FLAGS),)), |
||||
event_handler) |
||||
operations = ( |
||||
cygrpc.operation_send_initial_metadata( |
||||
_common.metadata(metadata), _EMPTY_FLAGS), |
||||
cygrpc.operation_receive_status_on_client(_EMPTY_FLAGS), |
||||
) |
||||
call.start_batch(cygrpc.Operations(operations), event_handler) |
||||
_consume_request_iterator( |
||||
request_iterator, state, call, self._request_serializer) |
||||
return _Rendezvous(state, call, self._response_deserializer, deadline) |
||||
|
||||
|
||||
class _ChannelCallState(object): |
||||
|
||||
def __init__(self, channel): |
||||
self.lock = threading.Lock() |
||||
self.channel = channel |
||||
self.completion_queue = cygrpc.CompletionQueue() |
||||
self.managed_calls = None |
||||
|
||||
|
||||
def _call_spin(state): |
||||
while True: |
||||
event = state.completion_queue.poll() |
||||
completed_call = event.tag(event) |
||||
if completed_call is not None: |
||||
with state.lock: |
||||
state.managed_calls.remove(completed_call) |
||||
if not state.managed_calls: |
||||
state.managed_calls = None |
||||
return |
||||
|
||||
|
||||
def _create_channel_managed_call(state): |
||||
def create_channel_managed_call(parent, flags, method, host, deadline): |
||||
"""Creates a managed cygrpc.Call. |
||||
|
||||
Callers of this function must conduct at least one operation on the returned |
||||
call. The tags associated with operations conducted on the returned call |
||||
must be no-argument callables that return None to indicate that this channel |
||||
should continue polling for events associated with the call and return the |
||||
call itself to indicate that no more events associated with the call will be |
||||
generated. |
||||
|
||||
Args: |
||||
parent: A cygrpc.Call to be used as the parent of the created call. |
||||
flags: An integer bitfield of call flags. |
||||
method: The RPC method. |
||||
host: A host string for the created call. |
||||
deadline: A cygrpc.Timespec to be the deadline of the created call. |
||||
|
||||
Returns: |
||||
A cygrpc.Call with which to conduct an RPC. |
||||
""" |
||||
with state.lock: |
||||
call = state.channel.create_call( |
||||
parent, flags, state.completion_queue, method, host, deadline) |
||||
if state.managed_calls is None: |
||||
state.managed_calls = set((call,)) |
||||
spin_thread = threading.Thread(target=_call_spin, args=(state,)) |
||||
spin_thread.start() |
||||
else: |
||||
state.managed_calls.add(call) |
||||
return call |
||||
return create_channel_managed_call |
||||
|
||||
|
||||
class _ChannelConnectivityState(object): |
||||
|
||||
def __init__(self, channel): |
||||
self.lock = threading.Lock() |
||||
self.channel = channel |
||||
self.polling = False |
||||
self.connectivity = None |
||||
self.try_to_connect = False |
||||
self.callbacks_and_connectivities = [] |
||||
self.delivering = False |
||||
|
||||
|
||||
def _deliveries(state): |
||||
callbacks_needing_update = [] |
||||
for callback_and_connectivity in state.callbacks_and_connectivities: |
||||
callback, callback_connectivity, = callback_and_connectivity |
||||
if callback_connectivity is not state.connectivity: |
||||
callbacks_needing_update.append(callback) |
||||
callback_and_connectivity[1] = state.connectivity |
||||
return callbacks_needing_update |
||||
|
||||
|
||||
def _deliver(state, initial_connectivity, initial_callbacks): |
||||
connectivity = initial_connectivity |
||||
callbacks = initial_callbacks |
||||
while True: |
||||
for callback in callbacks: |
||||
callable_util.call_logging_exceptions( |
||||
callback, _CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE, |
||||
connectivity) |
||||
with state.lock: |
||||
callbacks = _deliveries(state) |
||||
if callbacks: |
||||
connectivity = state.connectivity |
||||
else: |
||||
state.delivering = False |
||||
return |
||||
|
||||
|
||||
def _spawn_delivery(state, callbacks): |
||||
delivering_thread = threading.Thread( |
||||
target=_deliver, args=(state, state.connectivity, callbacks,)) |
||||
delivering_thread.start() |
||||
state.delivering = True |
||||
|
||||
|
||||
# NOTE(https://github.com/grpc/grpc/issues/3064): We'd rather not poll. |
||||
def _poll_connectivity(state, channel, initial_try_to_connect): |
||||
try_to_connect = initial_try_to_connect |
||||
connectivity = channel.check_connectivity_state(try_to_connect) |
||||
with state.lock: |
||||
state.connectivity = ( |
||||
_common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[ |
||||
connectivity]) |
||||
callbacks = tuple( |
||||
callback for callback, unused_but_known_to_be_none_connectivity |
||||
in state.callbacks_and_connectivities) |
||||
for callback_and_connectivity in state.callbacks_and_connectivities: |
||||
callback_and_connectivity[1] = state.connectivity |
||||
if callbacks: |
||||
_spawn_delivery(state, callbacks) |
||||
completion_queue = cygrpc.CompletionQueue() |
||||
while True: |
||||
channel.watch_connectivity_state( |
||||
connectivity, cygrpc.Timespec(time.time() + 0.2), |
||||
completion_queue, None) |
||||
event = completion_queue.poll() |
||||
with state.lock: |
||||
if not state.callbacks_and_connectivities and not state.try_to_connect: |
||||
state.polling = False |
||||
state.connectivity = None |
||||
break |
||||
try_to_connect = state.try_to_connect |
||||
state.try_to_connect = False |
||||
if event.success or try_to_connect: |
||||
connectivity = channel.check_connectivity_state(try_to_connect) |
||||
with state.lock: |
||||
state.connectivity = ( |
||||
_common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[ |
||||
connectivity]) |
||||
if not state.delivering: |
||||
callbacks = _deliveries(state) |
||||
if callbacks: |
||||
_spawn_delivery(state, callbacks) |
||||
|
||||
|
||||
def _subscribe(state, callback, try_to_connect): |
||||
with state.lock: |
||||
if not state.callbacks_and_connectivities and not state.polling: |
||||
polling_thread = threading.Thread( |
||||
target=_poll_connectivity, |
||||
args=(state, state.channel, bool(try_to_connect))) |
||||
polling_thread.start() |
||||
state.polling = True |
||||
state.callbacks_and_connectivities.append([callback, None]) |
||||
elif not state.delivering and state.connectivity is not None: |
||||
_spawn_delivery(state, (callback,)) |
||||
state.try_to_connect |= bool(try_to_connect) |
||||
state.callbacks_and_connectivities.append( |
||||
[callback, state.connectivity]) |
||||
else: |
||||
state.try_to_connect |= bool(try_to_connect) |
||||
state.callbacks_and_connectivities.append([callback, None]) |
||||
|
||||
|
||||
def _unsubscribe(state, callback): |
||||
with state.lock: |
||||
for index, (subscribed_callback, unused_connectivity) in enumerate( |
||||
state.callbacks_and_connectivities): |
||||
if callback == subscribed_callback: |
||||
state.callbacks_and_connectivities.pop(index) |
||||
break |
||||
|
||||
|
||||
def _moot(state): |
||||
with state.lock: |
||||
del state.callbacks_and_connectivities[:] |
||||
|
||||
|
||||
def _options(options): |
||||
if options is None: |
||||
pairs = ((cygrpc.ChannelArgKey.primary_user_agent_string, _USER_AGENT),) |
||||
else: |
||||
pairs = list(options) + [ |
||||
(cygrpc.ChannelArgKey.primary_user_agent_string, _USER_AGENT)] |
||||
return cygrpc.ChannelArgs( |
||||
cygrpc.ChannelArg(arg_name, arg_value) for arg_name, arg_value in pairs) |
||||
|
||||
|
||||
class Channel(grpc.Channel): |
||||
|
||||
def __init__(self, target, options, credentials): |
||||
self._channel = cygrpc.Channel(target, _options(options), credentials) |
||||
self._call_state = _ChannelCallState(self._channel) |
||||
self._connectivity_state = _ChannelConnectivityState(self._channel) |
||||
|
||||
def subscribe(self, callback, try_to_connect=None): |
||||
_subscribe(self._connectivity_state, callback, try_to_connect) |
||||
|
||||
def unsubscribe(self, callback): |
||||
_unsubscribe(self._connectivity_state, callback) |
||||
|
||||
def unary_unary( |
||||
self, method, request_serializer=None, response_deserializer=None): |
||||
return _UnaryUnaryMultiCallable( |
||||
self._channel, _create_channel_managed_call(self._call_state), method, |
||||
request_serializer, response_deserializer) |
||||
|
||||
def unary_stream( |
||||
self, method, request_serializer=None, response_deserializer=None): |
||||
return _UnaryStreamMultiCallable( |
||||
self._channel, _create_channel_managed_call(self._call_state), method, |
||||
request_serializer, response_deserializer) |
||||
|
||||
def stream_unary( |
||||
self, method, request_serializer=None, response_deserializer=None): |
||||
return _StreamUnaryMultiCallable( |
||||
self._channel, _create_channel_managed_call(self._call_state), method, |
||||
request_serializer, response_deserializer) |
||||
|
||||
def stream_stream( |
||||
self, method, request_serializer=None, response_deserializer=None): |
||||
return _StreamStreamMultiCallable( |
||||
self._channel, _create_channel_managed_call(self._call_state), method, |
||||
request_serializer, response_deserializer) |
||||
|
||||
def __del__(self): |
||||
_moot(self._connectivity_state) |
||||
@ -0,0 +1,99 @@ |
||||
# Copyright 2016, Google Inc. |
||||
# All rights reserved. |
||||
# |
||||
# Redistribution and use in source and binary forms, with or without |
||||
# modification, are permitted provided that the following conditions are |
||||
# met: |
||||
# |
||||
# * Redistributions of source code must retain the above copyright |
||||
# notice, this list of conditions and the following disclaimer. |
||||
# * Redistributions in binary form must reproduce the above |
||||
# copyright notice, this list of conditions and the following disclaimer |
||||
# in the documentation and/or other materials provided with the |
||||
# distribution. |
||||
# * Neither the name of Google Inc. nor the names of its |
||||
# contributors may be used to endorse or promote products derived from |
||||
# this software without specific prior written permission. |
||||
# |
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
|
||||
"""Shared implementation.""" |
||||
|
||||
import logging |
||||
|
||||
import six |
||||
|
||||
import grpc |
||||
from grpc._cython import cygrpc |
||||
|
||||
_EMPTY_METADATA = cygrpc.Metadata(()) |
||||
|
||||
CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = { |
||||
cygrpc.ConnectivityState.idle: grpc.ChannelConnectivity.IDLE, |
||||
cygrpc.ConnectivityState.connecting: grpc.ChannelConnectivity.CONNECTING, |
||||
cygrpc.ConnectivityState.ready: grpc.ChannelConnectivity.READY, |
||||
cygrpc.ConnectivityState.transient_failure: |
||||
grpc.ChannelConnectivity.TRANSIENT_FAILURE, |
||||
cygrpc.ConnectivityState.fatal_failure: |
||||
grpc.ChannelConnectivity.FATAL_FAILURE, |
||||
} |
||||
|
||||
CYGRPC_STATUS_CODE_TO_STATUS_CODE = { |
||||
cygrpc.StatusCode.ok: grpc.StatusCode.OK, |
||||
cygrpc.StatusCode.cancelled: grpc.StatusCode.CANCELLED, |
||||
cygrpc.StatusCode.unknown: grpc.StatusCode.UNKNOWN, |
||||
cygrpc.StatusCode.invalid_argument: grpc.StatusCode.INVALID_ARGUMENT, |
||||
cygrpc.StatusCode.deadline_exceeded: grpc.StatusCode.DEADLINE_EXCEEDED, |
||||
cygrpc.StatusCode.not_found: grpc.StatusCode.NOT_FOUND, |
||||
cygrpc.StatusCode.already_exists: grpc.StatusCode.ALREADY_EXISTS, |
||||
cygrpc.StatusCode.permission_denied: grpc.StatusCode.PERMISSION_DENIED, |
||||
cygrpc.StatusCode.unauthenticated: grpc.StatusCode.UNAUTHENTICATED, |
||||
cygrpc.StatusCode.resource_exhausted: grpc.StatusCode.RESOURCE_EXHAUSTED, |
||||
cygrpc.StatusCode.failed_precondition: grpc.StatusCode.FAILED_PRECONDITION, |
||||
cygrpc.StatusCode.aborted: grpc.StatusCode.ABORTED, |
||||
cygrpc.StatusCode.out_of_range: grpc.StatusCode.OUT_OF_RANGE, |
||||
cygrpc.StatusCode.unimplemented: grpc.StatusCode.UNIMPLEMENTED, |
||||
cygrpc.StatusCode.internal: grpc.StatusCode.INTERNAL, |
||||
cygrpc.StatusCode.unavailable: grpc.StatusCode.UNAVAILABLE, |
||||
cygrpc.StatusCode.data_loss: grpc.StatusCode.DATA_LOSS, |
||||
} |
||||
STATUS_CODE_TO_CYGRPC_STATUS_CODE = { |
||||
grpc_code: cygrpc_code |
||||
for cygrpc_code, grpc_code in six.iteritems( |
||||
CYGRPC_STATUS_CODE_TO_STATUS_CODE) |
||||
} |
||||
|
||||
|
||||
def metadata(application_metadata): |
||||
return _EMPTY_METADATA if application_metadata is None else cygrpc.Metadata( |
||||
cygrpc.Metadatum(key, value) for key, value in application_metadata) |
||||
|
||||
|
||||
def _transform(message, transformer, exception_message): |
||||
if transformer is None: |
||||
return message |
||||
else: |
||||
try: |
||||
return transformer(message) |
||||
except Exception: # pylint: disable=broad-except |
||||
logging.exception(exception_message) |
||||
return None |
||||
|
||||
|
||||
def serialize(message, serializer): |
||||
return _transform(message, serializer, 'Exception serializing message!') |
||||
|
||||
|
||||
def deserialize(serialized_message, deserializer): |
||||
return _transform(serialized_message, deserializer, |
||||
'Exception deserializing message!') |
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue