diff options
Diffstat (limited to 'src/core')
29 files changed, 1451 insertions, 243 deletions
diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c index 2c8113c1db..7a80975278 100644 --- a/src/core/ext/client_config/client_channel.c +++ b/src/core/ext/client_config/client_channel.c @@ -43,15 +43,20 @@ #include <grpc/support/useful.h> #include "src/core/ext/client_config/lb_policy_registry.h" +#include "src/core/ext/client_config/method_config.h" #include "src/core/ext/client_config/subchannel.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/connected_channel.h" +#include "src/core/lib/channel/deadline_filter.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/channel.h" #include "src/core/lib/transport/connectivity_state.h" +#include "src/core/lib/transport/metadata.h" +#include "src/core/lib/transport/metadata_batch.h" +#include "src/core/lib/transport/static_metadata.h" /* Client channel implementation */ @@ -67,12 +72,13 @@ typedef struct client_channel_channel_data { /** client channel factory */ grpc_client_channel_factory *client_channel_factory; - /** mutex protecting client configuration, including all - variables below in this data structure */ + /** mutex protecting all variables below in this data structure */ gpr_mu mu; - /** currently active load balancer - guarded by mu */ + /** currently active load balancer */ grpc_lb_policy *lb_policy; - /** incoming resolver result - set by resolver.next(), guarded by mu */ + /** method config table */ + grpc_method_config_table *method_config_table; + /** incoming resolver result - set by resolver.next() */ grpc_resolver_result *resolver_result; /** a list of closures that are all waiting for config to come in */ grpc_closure_list waiting_for_config_closures; @@ -114,7 +120,7 @@ static void set_channel_connectivity_state_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy_cancel_picks( exec_ctx, chand->lb_policy, /* mask= */ GRPC_INITIAL_METADATA_WAIT_FOR_READY, - /* check= */ 0); + /* check= */ 0, GRPC_ERROR_REF(error)); } grpc_connectivity_state_set(exec_ctx, &chand->state_tracker, state, error, reason); @@ -171,6 +177,7 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, channel_data *chand = arg; grpc_lb_policy *lb_policy = NULL; grpc_lb_policy *old_lb_policy; + grpc_method_config_table *method_config_table = NULL; grpc_connectivity_state state = GRPC_CHANNEL_TRANSIENT_FAILURE; bool exit_idle = false; grpc_error *state_error = GRPC_ERROR_CREATE("No load balancing policy"); @@ -194,6 +201,13 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, state = grpc_lb_policy_check_connectivity(exec_ctx, lb_policy, &state_error); } + const grpc_arg *channel_arg = grpc_channel_args_find( + lb_policy_args.additional_args, GRPC_ARG_SERVICE_CONFIG); + if (channel_arg != NULL) { + GPR_ASSERT(channel_arg->type == GRPC_ARG_POINTER); + method_config_table = grpc_method_config_table_ref( + (grpc_method_config_table *)channel_arg->value.pointer.p); + } grpc_resolver_result_unref(exec_ctx, chand->resolver_result); chand->resolver_result = NULL; } @@ -206,6 +220,10 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, gpr_mu_lock(&chand->mu); old_lb_policy = chand->lb_policy; chand->lb_policy = lb_policy; + if (chand->method_config_table != NULL) { + grpc_method_config_table_unref(chand->method_config_table); + } + chand->method_config_table = method_config_table; if (lb_policy != NULL) { grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures, NULL); @@ -366,6 +384,9 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx, chand->interested_parties); GRPC_LB_POLICY_UNREF(exec_ctx, chand->lb_policy, "channel"); } + if (chand->method_config_table != NULL) { + grpc_method_config_table_unref(chand->method_config_table); + } grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker); grpc_pollset_set_destroy(chand->interested_parties); gpr_mu_destroy(&chand->mu); @@ -391,6 +412,20 @@ typedef enum { for initial metadata before trying to create a call object, and handling cancellation gracefully. */ typedef struct client_channel_call_data { + // State for handling deadlines. + // The code in deadline_filter.c requires this to be the first field. + // TODO(roth): This is slightly sub-optimal in that grpc_deadline_state + // and this struct both independently store a pointer to the call + // stack and each has its own mutex. If/when we have time, find a way + // to avoid this without breaking the grpc_deadline_state abstraction. + grpc_deadline_state deadline_state; + gpr_timespec deadline; + + // Request path. + grpc_mdstr *path; + + grpc_error *cancel_error; + /** either 0 for no call, 1 for cancelled, or a pointer to a grpc_subchannel_call */ gpr_atm subchannel_call; @@ -485,16 +520,17 @@ static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, gpr_atm_no_barrier_store(&calld->subchannel_call, 1); fail_locked(exec_ctx, calld, GRPC_ERROR_CREATE_REFERENCING( "Failed to create subchannel", &error, 1)); - } else if (1 == gpr_atm_acq_load(&calld->subchannel_call)) { + } else if (GET_CALL(calld) == CANCELLED_CALL) { /* already cancelled before subchannel became ready */ fail_locked(exec_ctx, calld, GRPC_ERROR_CREATE_REFERENCING( "Cancelled before creating subchannel", &error, 1)); } else { + /* Create call on subchannel. */ grpc_subchannel_call *subchannel_call = NULL; grpc_error *new_error = grpc_connected_subchannel_create_call( - exec_ctx, calld->connected_subchannel, calld->pollent, - &subchannel_call); + exec_ctx, calld->connected_subchannel, calld->pollent, calld->path, + calld->deadline, &subchannel_call); if (new_error != GRPC_ERROR_NONE) { new_error = grpc_error_add_child(new_error, error); subchannel_call = CANCELLED_CALL; @@ -531,7 +567,7 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_metadata_batch *initial_metadata, uint32_t initial_metadata_flags, grpc_connected_subchannel **connected_subchannel, - grpc_closure *on_ready); + grpc_closure *on_ready, grpc_error *error); static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { @@ -542,7 +578,8 @@ static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_REF(error), NULL); } else if (pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata, cpa->initial_metadata_flags, - cpa->connected_subchannel, cpa->on_ready)) { + cpa->connected_subchannel, cpa->on_ready, + GRPC_ERROR_NONE)) { grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE, NULL); } gpr_free(cpa); @@ -552,7 +589,7 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_metadata_batch *initial_metadata, uint32_t initial_metadata_flags, grpc_connected_subchannel **connected_subchannel, - grpc_closure *on_ready) { + grpc_closure *on_ready, grpc_error *error) { GPR_TIMER_BEGIN("pick_subchannel", 0); channel_data *chand = elem->channel_data; @@ -566,21 +603,24 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, if (initial_metadata == NULL) { if (chand->lb_policy != NULL) { grpc_lb_policy_cancel_pick(exec_ctx, chand->lb_policy, - connected_subchannel); + connected_subchannel, GRPC_ERROR_REF(error)); } for (closure = chand->waiting_for_config_closures.head; closure != NULL; closure = closure->next_data.next) { cpa = closure->cb_arg; if (cpa->connected_subchannel == connected_subchannel) { cpa->connected_subchannel = NULL; - grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, - GRPC_ERROR_CREATE("Pick cancelled"), NULL); + grpc_exec_ctx_sched( + exec_ctx, cpa->on_ready, + GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1), NULL); } } gpr_mu_unlock(&chand->mu); GPR_TIMER_END("pick_subchannel", 0); + GRPC_ERROR_UNREF(error); return true; } + GPR_ASSERT(error == GRPC_ERROR_NONE); if (chand->lb_policy != NULL) { grpc_lb_policy *lb_policy = chand->lb_policy; int r; @@ -631,12 +671,13 @@ static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op) { call_data *calld = elem->call_data; GRPC_CALL_LOG_OP(GPR_INFO, elem, op); + grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op); /* try to (atomically) get the call */ grpc_subchannel_call *call = GET_CALL(calld); GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0); if (call == CANCELLED_CALL) { - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); + grpc_transport_stream_op_finish_with_failure( + exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error)); GPR_TIMER_END("cc_start_transport_stream_op", 0); return; } @@ -652,8 +693,8 @@ retry: call = GET_CALL(calld); if (call == CANCELLED_CALL) { gpr_mu_unlock(&calld->mu); - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); + grpc_transport_stream_op_finish_with_failure( + exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error)); GPR_TIMER_END("cc_start_transport_stream_op", 0); return; } @@ -669,18 +710,24 @@ retry: (gpr_atm)(uintptr_t)CANCELLED_CALL)) { goto retry; } else { + // Stash a copy of cancel_error in our call data, so that we can use + // it for subsequent operations. This ensures that if the call is + // cancelled before any ops are passed down (e.g., if the deadline + // is in the past when the call starts), we can return the right + // error to the caller when the first op does get passed down. + calld->cancel_error = GRPC_ERROR_REF(op->cancel_error); switch (calld->creation_phase) { case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING: fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error)); break; case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL: pick_subchannel(exec_ctx, elem, NULL, 0, &calld->connected_subchannel, - NULL); + NULL, GRPC_ERROR_REF(op->cancel_error)); break; } gpr_mu_unlock(&calld->mu); - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); + grpc_transport_stream_op_finish_with_failure( + exec_ctx, op, GRPC_ERROR_REF(op->cancel_error)); GPR_TIMER_END("cc_start_transport_stream_op", 0); return; } @@ -694,7 +741,8 @@ retry: GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel"); if (pick_subchannel(exec_ctx, elem, op->send_initial_metadata, op->send_initial_metadata_flags, - &calld->connected_subchannel, &calld->next_step)) { + &calld->connected_subchannel, &calld->next_step, + GRPC_ERROR_NONE)) { calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); } @@ -704,8 +752,8 @@ retry: calld->connected_subchannel != NULL) { grpc_subchannel_call *subchannel_call = NULL; grpc_error *error = grpc_connected_subchannel_create_call( - exec_ctx, calld->connected_subchannel, calld->pollent, - &subchannel_call); + exec_ctx, calld->connected_subchannel, calld->pollent, calld->path, + calld->deadline, &subchannel_call); if (error != GRPC_ERROR_NONE) { subchannel_call = CANCELLED_CALL; fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error)); @@ -726,7 +774,19 @@ retry: static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_call_element_args *args) { + channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; + gpr_mu_lock(&chand->mu); + grpc_method_config_table *method_config_table = + chand->method_config_table == NULL + ? NULL + : grpc_method_config_table_ref(chand->method_config_table); + gpr_mu_unlock(&chand->mu); + grpc_deadline_state_init(exec_ctx, elem, args, method_config_table); + grpc_method_config_table_unref(chand->method_config_table); + calld->deadline = args->deadline; + calld->path = GRPC_MDSTR_REF(args->path); + calld->cancel_error = GRPC_ERROR_NONE; gpr_atm_rel_store(&calld->subchannel_call, 0); gpr_mu_init(&calld->mu); calld->connected_subchannel = NULL; @@ -745,6 +805,9 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx, const grpc_call_final_info *final_info, void *and_free_memory) { call_data *calld = elem->call_data; + grpc_deadline_state_destroy(exec_ctx, elem); + GRPC_MDSTR_UNREF(calld->path); + GRPC_ERROR_UNREF(calld->cancel_error); grpc_subchannel_call *call = GET_CALL(calld); if (call != NULL && call != CANCELLED_CALL) { GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call"); diff --git a/src/core/ext/client_config/lb_policy.c b/src/core/ext/client_config/lb_policy.c index 903563ef6b..46391272a6 100644 --- a/src/core/ext/client_config/lb_policy.c +++ b/src/core/ext/client_config/lb_policy.c @@ -108,16 +108,18 @@ int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, } void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, - grpc_connected_subchannel **target) { - policy->vtable->cancel_pick(exec_ctx, policy, target); + grpc_connected_subchannel **target, + grpc_error *error) { + policy->vtable->cancel_pick(exec_ctx, policy, target, error); } void grpc_lb_policy_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, uint32_t initial_metadata_flags_mask, - uint32_t initial_metadata_flags_eq) { + uint32_t initial_metadata_flags_eq, + grpc_error *error) { policy->vtable->cancel_picks(exec_ctx, policy, initial_metadata_flags_mask, - initial_metadata_flags_eq); + initial_metadata_flags_eq, error); } void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) { diff --git a/src/core/ext/client_config/lb_policy.h b/src/core/ext/client_config/lb_policy.h index 37c93d707c..7fb3e08cb3 100644 --- a/src/core/ext/client_config/lb_policy.h +++ b/src/core/ext/client_config/lb_policy.h @@ -77,12 +77,12 @@ struct grpc_lb_policy_vtable { /** \see grpc_lb_policy_cancel_pick */ void (*cancel_pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, - grpc_connected_subchannel **target); + grpc_connected_subchannel **target, grpc_error *error); /** \see grpc_lb_policy_cancel_picks */ void (*cancel_picks)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, uint32_t initial_metadata_flags_mask, - uint32_t initial_metadata_flags_eq); + uint32_t initial_metadata_flags_eq, grpc_error *error); /** \see grpc_lb_policy_ping_one */ void (*ping_one)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, @@ -161,7 +161,8 @@ void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, The \a on_complete callback of the pending picks will be invoked with \a *target set to NULL. */ void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, - grpc_connected_subchannel **target); + grpc_connected_subchannel **target, + grpc_error *error); /** Cancel all pending picks for which their \a initial_metadata_flags (as given in the call to \a grpc_lb_policy_pick) matches \a initial_metadata_flags_eq @@ -169,7 +170,8 @@ void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, void grpc_lb_policy_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, uint32_t initial_metadata_flags_mask, - uint32_t initial_metadata_flags_eq); + uint32_t initial_metadata_flags_eq, + grpc_error *error); /** Try to enter a READY connectivity state */ void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy); diff --git a/src/core/ext/client_config/method_config.c b/src/core/ext/client_config/method_config.c new file mode 100644 index 0000000000..553a7be496 --- /dev/null +++ b/src/core/ext/client_config/method_config.c @@ -0,0 +1,289 @@ +// +// 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. +// + +#include "src/core/ext/client_config/method_config.h" + +#include <string.h> + +#include <grpc/impl/codegen/grpc_types.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include <grpc/support/time.h> + +#include "src/core/lib/transport/hashtable.h" +#include "src/core/lib/transport/metadata.h" + +// +// grpc_method_config +// + +// bool vtable + +static void* bool_copy(void* valuep) { + bool value = *(bool*)valuep; + bool* new_value = gpr_malloc(sizeof(bool)); + *new_value = value; + return new_value; +} + +static int bool_cmp(void* v1, void* v2) { + bool b1 = *(bool*)v1; + bool b2 = *(bool*)v2; + if (!b1 && b2) return -1; + if (b1 && !b2) return 1; + return 0; +} + +static grpc_hash_table_vtable bool_vtable = {gpr_free, bool_copy, bool_cmp}; + +// timespec vtable + +static void* timespec_copy(void* valuep) { + gpr_timespec value = *(gpr_timespec*)valuep; + gpr_timespec* new_value = gpr_malloc(sizeof(gpr_timespec)); + *new_value = value; + return new_value; +} + +static int timespec_cmp(void* v1, void* v2) { + return gpr_time_cmp(*(gpr_timespec*)v1, *(gpr_timespec*)v2); +} + +static grpc_hash_table_vtable timespec_vtable = { + gpr_free, timespec_copy, timespec_cmp}; + +// int32 vtable + +static void* int32_copy(void* valuep) { + int32_t value = *(int32_t*)valuep; + int32_t* new_value = gpr_malloc(sizeof(int32_t)); + *new_value = value; + return new_value; +} + +static int int32_cmp(void* v1, void* v2) { + int32_t i1 = *(int32_t*)v1; + int32_t i2 = *(int32_t*)v2; + if (i1 < i2) return -1; + if (i1 > i2) return 1; + return 0; +} + +static grpc_hash_table_vtable int32_vtable = {gpr_free, int32_copy, int32_cmp}; + +// Hash table keys. +#define GRPC_METHOD_CONFIG_WAIT_FOR_READY "grpc.wait_for_ready" // bool +#define GRPC_METHOD_CONFIG_TIMEOUT "grpc.timeout" // gpr_timespec +#define GRPC_METHOD_CONFIG_MAX_REQUEST_MESSAGE_BYTES \ + "grpc.max_request_message_bytes" // int32 +#define GRPC_METHOD_CONFIG_MAX_RESPONSE_MESSAGE_BYTES \ + "grpc.max_response_message_bytes" // int32 + +struct grpc_method_config { + grpc_hash_table* table; + grpc_mdstr* wait_for_ready_key; + grpc_mdstr* timeout_key; + grpc_mdstr* max_request_message_bytes_key; + grpc_mdstr* max_response_message_bytes_key; +}; + +grpc_method_config* grpc_method_config_create( + bool* wait_for_ready, gpr_timespec* timeout, + int32_t* max_request_message_bytes, int32_t* max_response_message_bytes) { + grpc_method_config* method_config = gpr_malloc(sizeof(grpc_method_config)); + memset(method_config, 0, sizeof(grpc_method_config)); + grpc_hash_table_entry entries[4]; + size_t num_entries = 0; + if (wait_for_ready != NULL) { + method_config->wait_for_ready_key = + grpc_mdstr_from_string(GRPC_METHOD_CONFIG_WAIT_FOR_READY); + entries[num_entries].key = method_config->wait_for_ready_key; + entries[num_entries].value = wait_for_ready; + entries[num_entries].vtable = &bool_vtable; + ++num_entries; + } + if (timeout != NULL) { + method_config->timeout_key = + grpc_mdstr_from_string(GRPC_METHOD_CONFIG_TIMEOUT); + entries[num_entries].key = method_config->timeout_key; + entries[num_entries].value = timeout; + entries[num_entries].vtable = ×pec_vtable; + ++num_entries; + } + if (max_request_message_bytes != NULL) { + method_config->max_request_message_bytes_key = + grpc_mdstr_from_string(GRPC_METHOD_CONFIG_MAX_REQUEST_MESSAGE_BYTES); + entries[num_entries].key = method_config->max_request_message_bytes_key; + entries[num_entries].value = max_request_message_bytes; + entries[num_entries].vtable = &int32_vtable; + ++num_entries; + } + if (max_response_message_bytes != NULL) { + method_config->max_response_message_bytes_key = + grpc_mdstr_from_string(GRPC_METHOD_CONFIG_MAX_RESPONSE_MESSAGE_BYTES); + entries[num_entries].key = method_config->max_response_message_bytes_key; + entries[num_entries].value = max_response_message_bytes; + entries[num_entries].vtable = &int32_vtable; + ++num_entries; + } + method_config->table = grpc_hash_table_create(num_entries, entries); + return method_config; +} + +grpc_method_config* grpc_method_config_ref(grpc_method_config* method_config) { + grpc_hash_table_ref(method_config->table); + return method_config; +} + +void grpc_method_config_unref(grpc_method_config* method_config) { + if (grpc_hash_table_unref(method_config->table)) { + GRPC_MDSTR_UNREF(method_config->wait_for_ready_key); + GRPC_MDSTR_UNREF(method_config->timeout_key); + GRPC_MDSTR_UNREF(method_config->max_request_message_bytes_key); + GRPC_MDSTR_UNREF(method_config->max_response_message_bytes_key); + } +} + +int grpc_method_config_cmp(grpc_method_config* method_config1, + grpc_method_config* method_config2) { + return grpc_hash_table_cmp(method_config1->table, method_config2->table); +} + +bool* grpc_method_config_get_wait_for_ready(grpc_method_config* method_config) { + return grpc_hash_table_get(method_config->table, + method_config->wait_for_ready_key); +} + +gpr_timespec* grpc_method_config_get_timeout( + grpc_method_config* method_config) { + return grpc_hash_table_get(method_config->table, method_config->timeout_key); +} + +int32_t* grpc_method_config_get_max_request_message_bytes( + grpc_method_config* method_config) { + return grpc_hash_table_get(method_config->table, + method_config->max_request_message_bytes_key); +} + +int32_t* grpc_method_config_get_max_response_message_bytes( + grpc_method_config* method_config) { + return grpc_hash_table_get(method_config->table, + method_config->max_response_message_bytes_key); +} + +// +// grpc_method_config_table +// + +static void method_config_unref(void* valuep) { + grpc_method_config_unref(valuep); +} + +static void* method_config_ref(void* valuep) { + return grpc_method_config_ref(valuep); +} + +static int method_config_cmp(void* valuep1, void* valuep2) { + return grpc_method_config_cmp(valuep1, valuep2); +} + +static const grpc_hash_table_vtable method_config_table_vtable = { + method_config_unref, method_config_ref, method_config_cmp}; + +grpc_method_config_table* grpc_method_config_table_create( + size_t num_entries, grpc_method_config_table_entry* entries) { + grpc_hash_table_entry hash_table_entries[num_entries]; + for (size_t i = 0; i < num_entries; ++i) { + hash_table_entries[i].key = entries[i].method_name; + hash_table_entries[i].value = entries[i].method_config; + hash_table_entries[i].vtable = &method_config_table_vtable; + } + return grpc_hash_table_create(num_entries, hash_table_entries); +} + +grpc_method_config_table* grpc_method_config_table_ref( + grpc_method_config_table* table) { + return grpc_hash_table_ref(table); +} + +void grpc_method_config_table_unref(grpc_method_config_table* table) { + grpc_hash_table_unref(table); +} + +int grpc_method_config_table_cmp(grpc_method_config_table* table1, + grpc_method_config_table* table2) { + return grpc_hash_table_cmp(table1, table2); +} + +grpc_method_config* grpc_method_config_table_get_method_config( + grpc_method_config_table* table, grpc_mdstr* path) { + grpc_method_config* method_config = grpc_hash_table_get(table, path); + // If we didn't find a match for the path, try looking for a wildcard + // entry (i.e., change "/service/method" to "/service/*"). + if (method_config == NULL) { + const char* path_str = grpc_mdstr_as_c_string(path); + const char* sep = strrchr(path_str, '/') + 1; + const size_t len = (size_t)(sep - path_str); + char buf[len + 2]; // '*' and NUL + memcpy(buf, path_str, len); + buf[len] = '*'; + buf[len + 1] = '\0'; + grpc_mdstr* wildcard_path = grpc_mdstr_from_string(buf); + method_config = grpc_hash_table_get(table, wildcard_path); + GRPC_MDSTR_UNREF(wildcard_path); + } + return grpc_method_config_ref(method_config); +} + +static void* copy_arg(void* p) { + return grpc_method_config_table_ref(p); +} + +static void destroy_arg(void* p) { + grpc_method_config_table_unref(p); +} + +static int cmp_arg(void* p1, void* p2) { + return grpc_method_config_table_cmp(p1, p2); +} + +static grpc_arg_pointer_vtable arg_vtable = {copy_arg, destroy_arg, cmp_arg}; + +grpc_arg grpc_method_config_table_create_channel_arg( + grpc_method_config_table* table) { + grpc_arg arg; + arg.type = GRPC_ARG_POINTER; + arg.key = GRPC_ARG_SERVICE_CONFIG; + arg.value.pointer.p = grpc_method_config_table_ref(table); + arg.value.pointer.vtable = &arg_vtable; + return arg; +} diff --git a/src/core/ext/client_config/method_config.h b/src/core/ext/client_config/method_config.h new file mode 100644 index 0000000000..e95a5583be --- /dev/null +++ b/src/core/ext/client_config/method_config.h @@ -0,0 +1,97 @@ +// +// 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_CLIENT_CONFIG_METHOD_CONFIG_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_METHOD_CONFIG_H + +#include <stdbool.h> + +#include <grpc/impl/codegen/gpr_types.h> +#include <grpc/impl/codegen/grpc_types.h> + +#include "src/core/lib/transport/hashtable.h" +#include "src/core/lib/transport/metadata.h" + +/// Per-method configuration. +typedef struct grpc_method_config grpc_method_config; + +/// Any parameter may be NULL to indicate that the value is unset. +grpc_method_config* grpc_method_config_create( + bool* wait_for_ready, gpr_timespec* timeout, + int32_t* max_request_message_bytes, int32_t* max_response_message_bytes); + +grpc_method_config* grpc_method_config_ref(grpc_method_config* method_config); +void grpc_method_config_unref(grpc_method_config* method_config); + +int grpc_method_config_cmp(grpc_method_config* method_config1, + grpc_method_config* method_config2); + +/// These methods return NULL if the requested field is unset. +/// The caller does NOT take ownership of the result. +bool* grpc_method_config_get_wait_for_ready(grpc_method_config* method_config); +gpr_timespec* grpc_method_config_get_timeout(grpc_method_config* method_config); +int32_t* grpc_method_config_get_max_request_message_bytes( + grpc_method_config* method_config); +int32_t* grpc_method_config_get_max_response_message_bytes( + grpc_method_config* method_config); + +/// A table of method configs. +typedef grpc_hash_table grpc_method_config_table; + +typedef struct grpc_method_config_table_entry { + /// The name is of one of the following forms: + /// service/method -- specifies exact service and method name + /// service/* -- matches all methods for the specified service + grpc_mdstr* method_name; + grpc_method_config* method_config; +} grpc_method_config_table_entry; + +/// Takes new references to all keys and values in \a entries. +grpc_method_config_table* grpc_method_config_table_create( + size_t num_entries, grpc_method_config_table_entry* entries); + +grpc_method_config_table* grpc_method_config_table_ref( + grpc_method_config_table* table); +void grpc_method_config_table_unref(grpc_method_config_table* table); + +int grpc_method_config_table_cmp(grpc_method_config_table* table1, + grpc_method_config_table* table2); + +/// Returns NULL if the method has no config. +/// Caller owns a reference to result. +grpc_method_config* grpc_method_config_table_get_method_config( + grpc_method_config_table* table, grpc_mdstr* path); + +/// Returns a channel arg containing \a table. +grpc_arg grpc_method_config_table_create_channel_arg( + grpc_method_config_table* table); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_METHOD_CONFIG_H */ diff --git a/src/core/ext/client_config/resolver_result.c b/src/core/ext/client_config/resolver_result.c index 63480d152b..16d124d205 100644 --- a/src/core/ext/client_config/resolver_result.c +++ b/src/core/ext/client_config/resolver_result.c @@ -36,8 +36,13 @@ #include <grpc/support/alloc.h> #include <grpc/support/string_util.h> +#include "src/core/lib/transport/metadata.h" #include "src/core/lib/channel/channel_args.h" +// +// grpc_resolver_result +// + struct grpc_resolver_result { gpr_refcount refs; char* server_name; diff --git a/src/core/ext/client_config/resolver_result.h b/src/core/ext/client_config/resolver_result.h index 414c2e2482..707c8c2cf5 100644 --- a/src/core/ext/client_config/resolver_result.h +++ b/src/core/ext/client_config/resolver_result.h @@ -48,26 +48,21 @@ /// Results reported from a grpc_resolver. typedef struct grpc_resolver_result grpc_resolver_result; -/// Takes ownership of \a addresses and \a lb_policy_args. +/// Takes ownership of \a addresses, \a lb_policy_args. grpc_resolver_result* grpc_resolver_result_create( const char* server_name, grpc_lb_addresses* addresses, const char* lb_policy_name, grpc_channel_args* lb_policy_args); + void grpc_resolver_result_ref(grpc_resolver_result* result); void grpc_resolver_result_unref(grpc_exec_ctx* exec_ctx, grpc_resolver_result* result); -/// Caller does NOT take ownership of result. +/// Accessors. Caller does NOT take ownership of results. const char* grpc_resolver_result_get_server_name(grpc_resolver_result* result); - -/// Caller does NOT take ownership of result. grpc_lb_addresses* grpc_resolver_result_get_addresses( grpc_resolver_result* result); - -/// Caller does NOT take ownership of result. const char* grpc_resolver_result_get_lb_policy_name( grpc_resolver_result* result); - -/// Caller does NOT take ownership of result. grpc_channel_args* grpc_resolver_result_get_lb_policy_args( grpc_resolver_result* result); diff --git a/src/core/ext/client_config/subchannel.c b/src/core/ext/client_config/subchannel.c index 1141844634..6674dbbe68 100644 --- a/src/core/ext/client_config/subchannel.c +++ b/src/core/ext/client_config/subchannel.c @@ -220,8 +220,8 @@ static gpr_atm ref_mutate(grpc_subchannel *c, gpr_atm delta, : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta); #ifdef GRPC_STREAM_REFCOUNT_DEBUG gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "SUBCHANNEL: %p % 12s 0x%08x -> 0x%08x [%s]", c, purpose, old_val, - old_val + delta, reason); + "SUBCHANNEL: %p %12s 0x%08d -> 0x%08d [%s]", c, purpose, (int)old_val, + (int)(old_val + delta), reason); #endif return old_val; } @@ -629,7 +629,9 @@ static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg, c->have_alarm = 1; grpc_connectivity_state_set( exec_ctx, &c->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, - GRPC_ERROR_CREATE_REFERENCING("Connect Failed", &error, 1), + grpc_error_set_int( + GRPC_ERROR_CREATE_REFERENCING("Connect Failed", &error, 1), + GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE), "connect_failed"); gpr_timespec time_til_next = gpr_time_sub(c->next_attempt, now); const char *errmsg = grpc_error_string(error); @@ -698,14 +700,15 @@ grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel( grpc_error *grpc_connected_subchannel_create_call( grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con, - grpc_polling_entity *pollent, grpc_subchannel_call **call) { + grpc_polling_entity *pollent, grpc_mdstr *path, gpr_timespec deadline, + grpc_subchannel_call **call) { grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con); *call = gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size); grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(*call); (*call)->connection = con; // Ref is added below. grpc_error *error = grpc_call_stack_init(exec_ctx, chanstk, 1, subchannel_call_destroy, *call, - NULL, NULL, callstk); + NULL, NULL, path, deadline, callstk); if (error != GRPC_ERROR_NONE) { const char *error_string = grpc_error_string(error); gpr_log(GPR_ERROR, "error: %s", error_string); diff --git a/src/core/ext/client_config/subchannel.h b/src/core/ext/client_config/subchannel.h index 218bb43e0a..f8de26dfd5 100644 --- a/src/core/ext/client_config/subchannel.h +++ b/src/core/ext/client_config/subchannel.h @@ -38,6 +38,7 @@ #include "src/core/lib/channel/channel_stack.h" #include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/transport/connectivity_state.h" +#include "src/core/lib/transport/metadata.h" /** A (sub-)channel that knows how to connect to exactly one target address. Provides a target for load balancing. */ @@ -110,7 +111,8 @@ void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx, /** construct a subchannel call */ grpc_error *grpc_connected_subchannel_create_call( grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *connected_subchannel, - grpc_polling_entity *pollent, grpc_subchannel_call **subchannel_call); + grpc_polling_entity *pollent, grpc_mdstr *path, gpr_timespec deadline, + grpc_subchannel_call **subchannel_call); /** process a transport level op */ void grpc_connected_subchannel_process_transport_op( diff --git a/src/core/ext/lb_policy/grpclb/grpclb.c b/src/core/ext/lb_policy/grpclb/grpclb.c index f45cff1a17..cda518394f 100644 --- a/src/core/ext/lb_policy/grpclb/grpclb.c +++ b/src/core/ext/lb_policy/grpclb/grpclb.c @@ -112,6 +112,7 @@ #include "src/core/ext/client_config/parse_address.h" #include "src/core/ext/lb_policy/grpclb/grpclb.h" #include "src/core/ext/lb_policy/grpclb/load_balancer_api.h" +#include "src/core/lib/channel/channel_args.h" #include "src/core/lib/iomgr/sockaddr.h" #include "src/core/lib/iomgr/sockaddr_utils.h" #include "src/core/lib/support/string.h" @@ -285,6 +286,7 @@ typedef struct glb_lb_policy { const char *server_name; grpc_client_channel_factory *cc_factory; + grpc_channel_args *args; /** for communicating with the LB server */ grpc_channel *lb_channel; @@ -442,6 +444,7 @@ static grpc_lb_policy *create_rr(grpc_exec_ctx *exec_ctx, args.server_name = glb_policy->server_name; args.client_channel_factory = glb_policy->cc_factory; args.addresses = process_serverlist(serverlist); + args.additional_args = glb_policy->args; grpc_lb_policy *rr = grpc_lb_policy_create(exec_ctx, "round_robin", &args); @@ -567,6 +570,7 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, * Create a client channel over them to communicate with a LB service */ glb_policy->server_name = gpr_strdup(args->server_name); glb_policy->cc_factory = args->client_channel_factory; + glb_policy->args = grpc_channel_args_copy(args->additional_args); GPR_ASSERT(glb_policy->cc_factory != NULL); /* construct a target from the addresses in args, given in the form @@ -633,6 +637,7 @@ static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { GPR_ASSERT(glb_policy->pending_picks == NULL); GPR_ASSERT(glb_policy->pending_pings == NULL); gpr_free((void *)glb_policy->server_name); + grpc_channel_args_destroy(glb_policy->args); grpc_channel_destroy(glb_policy->lb_channel); glb_policy->lb_channel = NULL; grpc_connectivity_state_destroy(exec_ctx, &glb_policy->state_tracker); @@ -688,7 +693,8 @@ static void glb_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { } static void glb_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, - grpc_connected_subchannel **target) { + grpc_connected_subchannel **target, + grpc_error *error) { glb_lb_policy *glb_policy = (glb_lb_policy *)pol; gpr_mu_lock(&glb_policy->mu); pending_pick *pp = glb_policy->pending_picks; @@ -699,8 +705,10 @@ static void glb_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_polling_entity_del_from_pollset_set( exec_ctx, pp->pollent, glb_policy->base.interested_parties); *target = NULL; - grpc_exec_ctx_sched(exec_ctx, &pp->wrapped_on_complete, - GRPC_ERROR_CANCELLED, NULL); + grpc_exec_ctx_sched( + exec_ctx, &pp->wrapped_on_complete, + GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1), NULL); + gpr_free(pp); } else { pp->next = glb_policy->pending_picks; glb_policy->pending_picks = pp; @@ -708,12 +716,14 @@ static void glb_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pp = next; } gpr_mu_unlock(&glb_policy->mu); + GRPC_ERROR_UNREF(error); } static grpc_call *lb_client_data_get_call(struct lb_client_data *lb_client); static void glb_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, uint32_t initial_metadata_flags_mask, - uint32_t initial_metadata_flags_eq) { + uint32_t initial_metadata_flags_eq, + grpc_error *error) { glb_lb_policy *glb_policy = (glb_lb_policy *)pol; gpr_mu_lock(&glb_policy->mu); if (glb_policy->lb_client != NULL) { @@ -728,8 +738,10 @@ static void glb_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, initial_metadata_flags_eq) { grpc_polling_entity_del_from_pollset_set( exec_ctx, pp->pollent, glb_policy->base.interested_parties); - grpc_exec_ctx_sched(exec_ctx, &pp->wrapped_on_complete, - GRPC_ERROR_CANCELLED, NULL); + grpc_exec_ctx_sched( + exec_ctx, &pp->wrapped_on_complete, + GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1), NULL); + gpr_free(pp); } else { pp->next = glb_policy->pending_picks; glb_policy->pending_picks = pp; @@ -737,6 +749,7 @@ static void glb_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pp = next; } gpr_mu_unlock(&glb_policy->mu); + GRPC_ERROR_UNREF(error); } static void query_for_backends(grpc_exec_ctx *exec_ctx, diff --git a/src/core/ext/lb_policy/pick_first/pick_first.c b/src/core/ext/lb_policy/pick_first/pick_first.c index 5af484eccf..c0efbe511b 100644 --- a/src/core/ext/lb_policy/pick_first/pick_first.c +++ b/src/core/ext/lb_policy/pick_first/pick_first.c @@ -128,7 +128,8 @@ static void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { } static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, - grpc_connected_subchannel **target) { + grpc_connected_subchannel **target, + grpc_error *error) { pick_first_lb_policy *p = (pick_first_lb_policy *)pol; pending_pick *pp; gpr_mu_lock(&p->mu); @@ -140,8 +141,9 @@ static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_polling_entity_del_from_pollset_set(exec_ctx, pp->pollent, p->base.interested_parties); *target = NULL; - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, - GRPC_ERROR_CREATE("Pick Cancelled"), NULL); + grpc_exec_ctx_sched( + exec_ctx, pp->on_complete, + GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1), NULL); gpr_free(pp); } else { pp->next = p->pending_picks; @@ -150,11 +152,13 @@ static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pp = next; } gpr_mu_unlock(&p->mu); + GRPC_ERROR_UNREF(error); } static void pf_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, uint32_t initial_metadata_flags_mask, - uint32_t initial_metadata_flags_eq) { + uint32_t initial_metadata_flags_eq, + grpc_error *error) { pick_first_lb_policy *p = (pick_first_lb_policy *)pol; pending_pick *pp; gpr_mu_lock(&p->mu); @@ -166,8 +170,9 @@ static void pf_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, initial_metadata_flags_eq) { grpc_polling_entity_del_from_pollset_set(exec_ctx, pp->pollent, p->base.interested_parties); - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, - GRPC_ERROR_CREATE("Pick Cancelled"), NULL); + grpc_exec_ctx_sched( + exec_ctx, pp->on_complete, + GRPC_ERROR_CREATE_REFERENCING("Pick Cancelled", &error, 1), NULL); gpr_free(pp); } else { pp->next = p->pending_picks; @@ -176,6 +181,7 @@ static void pf_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pp = next; } gpr_mu_unlock(&p->mu); + GRPC_ERROR_UNREF(error); } static void start_picking(grpc_exec_ctx *exec_ctx, pick_first_lb_policy *p) { @@ -472,6 +478,7 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx, sc_args.addr = (struct sockaddr *)(&args->addresses->addresses[i].address.addr); sc_args.addr_len = args->addresses->addresses[i].address.len; + sc_args.args = args->additional_args; grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel( exec_ctx, args->client_channel_factory, &sc_args); diff --git a/src/core/ext/lb_policy/round_robin/round_robin.c b/src/core/ext/lb_policy/round_robin/round_robin.c index dfd4b1929c..37856c228d 100644 --- a/src/core/ext/lb_policy/round_robin/round_robin.c +++ b/src/core/ext/lb_policy/round_robin/round_robin.c @@ -308,7 +308,8 @@ static void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { } static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, - grpc_connected_subchannel **target) { + grpc_connected_subchannel **target, + grpc_error *error) { round_robin_lb_policy *p = (round_robin_lb_policy *)pol; pending_pick *pp; gpr_mu_lock(&p->mu); @@ -320,8 +321,9 @@ static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_polling_entity_del_from_pollset_set(exec_ctx, pp->pollent, p->base.interested_parties); *target = NULL; - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, GRPC_ERROR_CANCELLED, - NULL); + grpc_exec_ctx_sched( + exec_ctx, pp->on_complete, + GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1), NULL); gpr_free(pp); } else { pp->next = p->pending_picks; @@ -330,11 +332,13 @@ static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pp = next; } gpr_mu_unlock(&p->mu); + GRPC_ERROR_UNREF(error); } static void rr_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, uint32_t initial_metadata_flags_mask, - uint32_t initial_metadata_flags_eq) { + uint32_t initial_metadata_flags_eq, + grpc_error *error) { round_robin_lb_policy *p = (round_robin_lb_policy *)pol; pending_pick *pp; gpr_mu_lock(&p->mu); @@ -347,8 +351,9 @@ static void rr_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_polling_entity_del_from_pollset_set(exec_ctx, pp->pollent, p->base.interested_parties); *pp->target = NULL; - grpc_exec_ctx_sched(exec_ctx, pp->on_complete, GRPC_ERROR_CANCELLED, - NULL); + grpc_exec_ctx_sched( + exec_ctx, pp->on_complete, + GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1), NULL); gpr_free(pp); } else { pp->next = p->pending_picks; @@ -357,6 +362,7 @@ static void rr_cancel_picks(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, pp = next; } gpr_mu_unlock(&p->mu); + GRPC_ERROR_UNREF(error); } static void start_picking(grpc_exec_ctx *exec_ctx, round_robin_lb_policy *p) { @@ -635,6 +641,7 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx, sc_args.addr = (struct sockaddr *)(&args->addresses->addresses[i].address.addr); sc_args.addr_len = args->addresses->addresses[i].address.len; + sc_args.args = args->additional_args; grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel( exec_ctx, args->client_channel_factory, &sc_args); diff --git a/src/core/lib/channel/channel_args.c b/src/core/lib/channel/channel_args.c index 3a56b1ff20..bab6fcd9fa 100644 --- a/src/core/lib/channel/channel_args.c +++ b/src/core/lib/channel/channel_args.c @@ -272,6 +272,16 @@ int grpc_channel_args_compare(const grpc_channel_args *a, return 0; } +const grpc_arg *grpc_channel_args_find(const grpc_channel_args *args, + const char *name) { + if (args != NULL) { + for (size_t i = 0; i < args->num_args; ++i) { + if (args->args[i].key == name) return &args->args[i]; + } + } + return NULL; +} + int grpc_channel_arg_get_integer(grpc_arg *arg, grpc_integer_options options) { if (arg->type != GRPC_ARG_INTEGER) { gpr_log(GPR_ERROR, "%s ignored: it must be an integer", arg->key); diff --git a/src/core/lib/channel/channel_args.h b/src/core/lib/channel/channel_args.h index 586a296d1f..38fb4c55d4 100644 --- a/src/core/lib/channel/channel_args.h +++ b/src/core/lib/channel/channel_args.h @@ -89,6 +89,12 @@ uint32_t grpc_channel_args_compression_algorithm_get_states( int grpc_channel_args_compare(const grpc_channel_args *a, const grpc_channel_args *b); +/** Returns the value of argument \a name from \a args, or NULL if not found. + Note: \a name is matched using pointer equality, so it must be the + same instance of the string used to create the grpc_arg key. */ +const grpc_arg *grpc_channel_args_find(const grpc_channel_args *args, + const char *name); + typedef struct grpc_integer_options { int default_value; // Return this if value is outside of expected bounds. int min_value; diff --git a/src/core/lib/channel/channel_stack.c b/src/core/lib/channel/channel_stack.c index 0655b9353f..205496f2f2 100644 --- a/src/core/lib/channel/channel_stack.c +++ b/src/core/lib/channel/channel_stack.c @@ -158,13 +158,11 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx, } } -grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx, - grpc_channel_stack *channel_stack, - int initial_refs, grpc_iomgr_cb_func destroy, - void *destroy_arg, - grpc_call_context_element *context, - const void *transport_server_data, - grpc_call_stack *call_stack) { +grpc_error *grpc_call_stack_init( + grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack, + int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg, + grpc_call_context_element *context, const void *transport_server_data, + grpc_mdstr *path, gpr_timespec deadline, grpc_call_stack *call_stack) { grpc_channel_element *channel_elems = CHANNEL_ELEMS_FROM_STACK(channel_stack); grpc_call_element_args args; size_t count = channel_stack->count; @@ -185,6 +183,8 @@ grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx, args.call_stack = call_stack; args.server_transport_data = transport_server_data; args.context = context; + args.path = path; + args.deadline = deadline; call_elems[i].filter = channel_elems[i].filter; call_elems[i].channel_data = channel_elems[i].channel_data; call_elems[i].call_data = user_data; @@ -276,16 +276,16 @@ static void destroy_op(grpc_exec_ctx *exec_ctx, void *op, grpc_error *error) { } void grpc_call_element_send_cancel(grpc_exec_ctx *exec_ctx, - grpc_call_element *cur_elem) { + grpc_call_element *elem) { grpc_transport_stream_op *op = gpr_malloc(sizeof(*op)); memset(op, 0, sizeof(*op)); op->cancel_error = GRPC_ERROR_CANCELLED; op->on_complete = grpc_closure_create(destroy_op, op); - grpc_call_next_op(exec_ctx, cur_elem, op); + elem->filter->start_transport_stream_op(exec_ctx, elem, op); } void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx, - grpc_call_element *cur_elem, + grpc_call_element *elem, grpc_status_code status, gpr_slice *optional_message) { grpc_transport_stream_op *op = gpr_malloc(sizeof(*op)); @@ -293,5 +293,16 @@ void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx, op->on_complete = grpc_closure_create(destroy_op, op); grpc_transport_stream_op_add_cancellation_with_message(op, status, optional_message); - grpc_call_next_op(exec_ctx, cur_elem, op); + elem->filter->start_transport_stream_op(exec_ctx, elem, op); +} + +void grpc_call_element_send_close_with_message(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_status_code status, + gpr_slice *optional_message) { + grpc_transport_stream_op *op = gpr_malloc(sizeof(*op)); + memset(op, 0, sizeof(*op)); + op->on_complete = grpc_closure_create(destroy_op, op); + grpc_transport_stream_op_add_close(op, status, optional_message); + elem->filter->start_transport_stream_op(exec_ctx, elem, op); } diff --git a/src/core/lib/channel/channel_stack.h b/src/core/lib/channel/channel_stack.h index 6b73cce380..5b46cd32a3 100644 --- a/src/core/lib/channel/channel_stack.h +++ b/src/core/lib/channel/channel_stack.h @@ -74,6 +74,8 @@ typedef struct { grpc_call_stack *call_stack; const void *server_transport_data; grpc_call_context_element *context; + grpc_mdstr *path; + gpr_timespec deadline; } grpc_call_element_args; typedef struct { @@ -220,13 +222,11 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx, /* Initialize a call stack given a channel stack. transport_server_data is expected to be NULL on a client, or an opaque transport owned pointer on the server. */ -grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx, - grpc_channel_stack *channel_stack, - int initial_refs, grpc_iomgr_cb_func destroy, - void *destroy_arg, - grpc_call_context_element *context, - const void *transport_server_data, - grpc_call_stack *call_stack); +grpc_error *grpc_call_stack_init( + grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack, + int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg, + grpc_call_context_element *context, const void *transport_server_data, + grpc_mdstr *path, gpr_timespec deadline, grpc_call_stack *call_stack); /* Set a pollset or a pollset_set for a call stack: must occur before the first * op is started */ void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, @@ -290,6 +290,11 @@ void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx, grpc_status_code status, gpr_slice *optional_message); +void grpc_call_element_send_close_with_message(grpc_exec_ctx *exec_ctx, + grpc_call_element *cur_elem, + grpc_status_code status, + gpr_slice *optional_message); + extern int grpc_trace_channel; #define GRPC_CALL_LOG_OP(sev, elem, op) \ diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c new file mode 100644 index 0000000000..89dc8dd6e8 --- /dev/null +++ b/src/core/lib/channel/deadline_filter.c @@ -0,0 +1,316 @@ +// +// 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 "src/core/lib/channel/deadline_filter.h" + +#include <stdbool.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/sync.h> +#include <grpc/support/time.h> + +#include "src/core/lib/iomgr/exec_ctx.h" +#include "src/core/lib/iomgr/timer.h" + +// +// grpc_deadline_state +// + +// Timer callback. +static void timer_callback(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { + grpc_call_element* elem = arg; + grpc_deadline_state* deadline_state = elem->call_data; + gpr_mu_lock(&deadline_state->timer_mu); + deadline_state->timer_pending = false; + gpr_mu_unlock(&deadline_state->timer_mu); + if (error != GRPC_ERROR_CANCELLED) { + gpr_slice msg = gpr_slice_from_static_string("Deadline Exceeded"); + grpc_call_element_send_cancel_with_message( + exec_ctx, elem, GRPC_STATUS_DEADLINE_EXCEEDED, &msg); + gpr_slice_unref(msg); + } + GRPC_CALL_STACK_UNREF(exec_ctx, deadline_state->call_stack, "deadline_timer"); +} + +// Starts the deadline timer. +static void start_timer_if_needed(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + gpr_timespec deadline) { + grpc_deadline_state* deadline_state = elem->call_data; + deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); + if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_MONOTONIC)) != 0) { + // Take a reference to the call stack, to be owned by the timer. + GRPC_CALL_STACK_REF(deadline_state->call_stack, "deadline_timer"); + gpr_mu_lock(&deadline_state->timer_mu); + deadline_state->timer_pending = true; + grpc_timer_init(exec_ctx, &deadline_state->timer, deadline, timer_callback, + elem, gpr_now(GPR_CLOCK_MONOTONIC)); + gpr_mu_unlock(&deadline_state->timer_mu); + } +} + +// Cancels the deadline timer. +static void cancel_timer_if_needed(grpc_exec_ctx* exec_ctx, + grpc_deadline_state* deadline_state) { + gpr_mu_lock(&deadline_state->timer_mu); + if (deadline_state->timer_pending) { + grpc_timer_cancel(exec_ctx, &deadline_state->timer); + deadline_state->timer_pending = false; + } + gpr_mu_unlock(&deadline_state->timer_mu); +} + +// Callback run when the call is complete. +static void on_complete(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { + grpc_deadline_state* deadline_state = arg; + cancel_timer_if_needed(exec_ctx, deadline_state); + // Invoke the next callback. + deadline_state->next_on_complete->cb( + exec_ctx, deadline_state->next_on_complete->cb_arg, error); +} + +// Inject our own on_complete callback into op. +static void inject_on_complete_cb(grpc_deadline_state* deadline_state, + grpc_transport_stream_op* op) { + deadline_state->next_on_complete = op->on_complete; + grpc_closure_init(&deadline_state->on_complete, on_complete, deadline_state); + op->on_complete = &deadline_state->on_complete; +} + +// Callback and associated state for starting the timer after call stack +// initialization has been completed. +struct start_timer_after_init_state { + grpc_call_element* elem; + gpr_timespec deadline; + grpc_closure closure; +}; +static void start_timer_after_init(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { + struct start_timer_after_init_state* state = arg; + start_timer_if_needed(exec_ctx, state->elem, state->deadline); + gpr_free(state); +} + +void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + grpc_call_element_args* args, + grpc_method_config_table* method_config_table) { + grpc_deadline_state* deadline_state = elem->call_data; + memset(deadline_state, 0, sizeof(*deadline_state)); + deadline_state->call_stack = args->call_stack; + gpr_mu_init(&deadline_state->timer_mu); + // Deadline will always be infinite on servers, so the timer will only be + // set on clients with a finite deadline. + gpr_timespec deadline = + gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC); + if (method_config_table != NULL) { + grpc_method_config* method_config = + grpc_method_config_table_get_method_config(method_config_table, + args->path); + if (method_config != NULL) { + gpr_timespec* per_method_deadline = + grpc_method_config_get_timeout(method_config); + if (per_method_deadline != NULL) { + deadline = gpr_time_min(deadline, *per_method_deadline); + } + } + } + if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_MONOTONIC)) != 0) { + // When the deadline passes, we indicate the failure by sending down + // an op with cancel_error set. However, we can't send down any ops + // until after the call stack is fully initialized. If we start the + // timer here, we have no guarantee that the timer won't pop before + // call stack initialization is finished. To avoid that problem, we + // create a closure to start the timer, and we schedule that closure + // to be run after call stack initialization is done. + struct start_timer_after_init_state* state = gpr_malloc(sizeof(*state)); + state->elem = elem; + state->deadline = deadline; + grpc_closure_init(&state->closure, start_timer_after_init, state); + grpc_exec_ctx_sched(exec_ctx, &state->closure, GRPC_ERROR_NONE, NULL); + } +} + +void grpc_deadline_state_destroy(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem) { + grpc_deadline_state* deadline_state = elem->call_data; + cancel_timer_if_needed(exec_ctx, deadline_state); + gpr_mu_destroy(&deadline_state->timer_mu); +} + +void grpc_deadline_state_client_start_transport_stream_op( + grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + grpc_transport_stream_op* op) { + grpc_deadline_state* deadline_state = elem->call_data; + if (op->cancel_error != GRPC_ERROR_NONE || + op->close_error != GRPC_ERROR_NONE) { + cancel_timer_if_needed(exec_ctx, deadline_state); + } else { + // Make sure we know when the call is complete, so that we can cancel + // the timer. + if (op->recv_trailing_metadata != NULL) { + inject_on_complete_cb(deadline_state, op); + } + } +} + +// +// filter code +// + +// Constructor for channel_data. Used for both client and server filters. +static void init_channel_elem(grpc_exec_ctx* exec_ctx, + grpc_channel_element* elem, + grpc_channel_element_args* args) { + GPR_ASSERT(!args->is_last); +} + +// Destructor for channel_data. Used for both client and server filters. +static void destroy_channel_elem(grpc_exec_ctx* exec_ctx, + grpc_channel_element* elem) {} + +// Call data used for both client and server filter. +typedef struct base_call_data { + grpc_deadline_state deadline_state; +} base_call_data; + +// Additional call data used only for the server filter. +typedef struct server_call_data { + base_call_data base; // Must be first. + // The closure for receiving initial metadata. + grpc_closure recv_initial_metadata_ready; + // Received initial metadata batch. + grpc_metadata_batch* recv_initial_metadata; + // The original recv_initial_metadata_ready closure, which we chain to + // after our own closure is invoked. + grpc_closure* next_recv_initial_metadata_ready; +} server_call_data; + +// Constructor for call_data. Used for both client and server filters. +static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + grpc_call_element_args* args) { + // Note: size of call data is different between client and server. + memset(elem->call_data, 0, elem->filter->sizeof_call_data); + grpc_deadline_state_init(exec_ctx, elem, args, + NULL /* method_config_table */); + return GRPC_ERROR_NONE; +} + +// Destructor for call_data. Used for both client and server filters. +static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + const grpc_call_final_info* final_info, + void* and_free_memory) { + grpc_deadline_state_destroy(exec_ctx, elem); +} + +// Method for starting a call op for client filter. +static void client_start_transport_stream_op(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + grpc_transport_stream_op* op) { + grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op); + // Chain to next filter. + grpc_call_next_op(exec_ctx, elem, op); +} + +// Callback for receiving initial metadata on the server. +static void recv_initial_metadata_ready(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { + grpc_call_element* elem = arg; + server_call_data* calld = elem->call_data; + // Get deadline from metadata and start the timer if needed. + start_timer_if_needed(exec_ctx, elem, calld->recv_initial_metadata->deadline); + // Invoke the next callback. + calld->next_recv_initial_metadata_ready->cb( + exec_ctx, calld->next_recv_initial_metadata_ready->cb_arg, error); +} + +// Method for starting a call op for server filter. +static void server_start_transport_stream_op(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem, + grpc_transport_stream_op* op) { + server_call_data* calld = elem->call_data; + if (op->cancel_error != GRPC_ERROR_NONE || + op->close_error != GRPC_ERROR_NONE) { + cancel_timer_if_needed(exec_ctx, &calld->base.deadline_state); + } else { + // If we're receiving initial metadata, we need to get the deadline + // from the recv_initial_metadata_ready callback. So we inject our + // own callback into that hook. + if (op->recv_initial_metadata_ready != NULL) { + calld->next_recv_initial_metadata_ready = op->recv_initial_metadata_ready; + calld->recv_initial_metadata = op->recv_initial_metadata; + grpc_closure_init(&calld->recv_initial_metadata_ready, + recv_initial_metadata_ready, elem); + op->recv_initial_metadata_ready = &calld->recv_initial_metadata_ready; + } + // Make sure we know when the call is complete, so that we can cancel + // the timer. + // Note that we trigger this on recv_trailing_metadata, even though + // the client never sends trailing metadata, because this is the + // hook that tells us when the call is complete on the server side. + if (op->recv_trailing_metadata != NULL) { + inject_on_complete_cb(&calld->base.deadline_state, op); + } + } + // Chain to next filter. + grpc_call_next_op(exec_ctx, elem, op); +} + +const grpc_channel_filter grpc_client_deadline_filter = { + client_start_transport_stream_op, + grpc_channel_next_op, + sizeof(base_call_data), + init_call_elem, + grpc_call_stack_ignore_set_pollset_or_pollset_set, + destroy_call_elem, + 0, // sizeof(channel_data) + init_channel_elem, + destroy_channel_elem, + grpc_call_next_get_peer, + "deadline", +}; + +const grpc_channel_filter grpc_server_deadline_filter = { + server_start_transport_stream_op, + grpc_channel_next_op, + sizeof(server_call_data), + init_call_elem, + grpc_call_stack_ignore_set_pollset_or_pollset_set, + destroy_call_elem, + 0, // sizeof(channel_data) + init_channel_elem, + destroy_channel_elem, + grpc_call_next_get_peer, + "deadline", +}; diff --git a/src/core/lib/channel/deadline_filter.h b/src/core/lib/channel/deadline_filter.h new file mode 100644 index 0000000000..2fcee0b9ef --- /dev/null +++ b/src/core/lib/channel/deadline_filter.h @@ -0,0 +1,82 @@ +// +// 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_LIB_CHANNEL_DEADLINE_FILTER_H +#define GRPC_CORE_LIB_CHANNEL_DEADLINE_FILTER_H + +#include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/iomgr/timer.h" + +#include "src/core/ext/client_config/method_config.h" + +// State used for filters that enforce call deadlines. +// Must be the first field in the filter's call_data. +typedef struct grpc_deadline_state { + // We take a reference to the call stack for the timer callback. + grpc_call_stack* call_stack; + // Guards access to timer_pending and timer. + gpr_mu timer_mu; + // True if the timer callback is currently pending. + bool timer_pending; + // The deadline timer. + grpc_timer timer; + // Closure to invoke when the call is complete. + // We use this to cancel the timer. + grpc_closure on_complete; + // The original on_complete closure, which we chain to after our own + // closure is invoked. + grpc_closure* next_on_complete; +} grpc_deadline_state; + +// To be used in a filter's init_call_elem(), destroy_call_elem(), and +// start_transport_stream_op() methods to enforce call deadlines. +// +// REQUIRES: The first field in elem->call_data is a grpc_deadline_state. +// +// For grpc_deadline_state_client_start_transport_stream_op(), it is the +// caller's responsibility to chain to the next filter if necessary +// after the function returns. +void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + grpc_call_element_args* args, + grpc_method_config_table* method_config_table); +void grpc_deadline_state_destroy(grpc_exec_ctx* exec_ctx, + grpc_call_element* elem); +void grpc_deadline_state_client_start_transport_stream_op( + grpc_exec_ctx* exec_ctx, grpc_call_element* elem, + grpc_transport_stream_op* op); + +// Deadline filters for direct client channels and server channels. +// Note: Deadlines for non-direct client channels are handled by the +// client_channel filter. +extern const grpc_channel_filter grpc_client_deadline_filter; +extern const grpc_channel_filter grpc_server_deadline_filter; + +#endif /* GRPC_CORE_LIB_CHANNEL_DEADLINE_FILTER_H */ diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c index edcc741ff6..1dc05fb20d 100644 --- a/src/core/lib/channel/http_client_filter.c +++ b/src/core/lib/channel/http_client_filter.c @@ -103,8 +103,8 @@ static grpc_mdelem *client_recv_filter(void *user_data, grpc_mdelem *md) { grpc_mdstr_as_c_string(md->value)); gpr_slice message = gpr_slice_from_copied_string(message_string); gpr_free(message_string); - grpc_call_element_send_cancel_with_message(a->exec_ctx, a->elem, - GRPC_STATUS_CANCELLED, &message); + grpc_call_element_send_close_with_message(a->exec_ctx, a->elem, + GRPC_STATUS_CANCELLED, &message); return NULL; } else if (md == GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC) { return NULL; diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c index f8a6dd3491..4efa6f07be 100644 --- a/src/core/lib/channel/message_size_filter.c +++ b/src/core/lib/channel/message_size_filter.c @@ -39,12 +39,15 @@ #include <grpc/support/string_util.h> #include "src/core/lib/channel/channel_args.h" +#include "src/core/ext/client_config/method_config.h" #define DEFAULT_MAX_SEND_MESSAGE_LENGTH -1 // Unlimited. // The protobuf library will (by default) start warning at 100 megs. #define DEFAULT_MAX_RECV_MESSAGE_LENGTH (4 * 1024 * 1024) typedef struct call_data { + int max_send_size; + int max_recv_size; // Receive closures are chained: we inject this closure as the // recv_message_ready up-call on transport_stream_op, and remember to // call our next_recv_message_ready member after handling it. @@ -58,6 +61,8 @@ typedef struct call_data { typedef struct channel_data { int max_send_size; int max_recv_size; + // Method config table. + grpc_method_config_table* method_config_table; } channel_data; // Callback invoked when we receive a message. Here we check the max @@ -66,16 +71,15 @@ static void recv_message_ready(grpc_exec_ctx* exec_ctx, void* user_data, grpc_error* error) { grpc_call_element* elem = user_data; call_data* calld = elem->call_data; - channel_data* chand = elem->channel_data; - if (*calld->recv_message != NULL && chand->max_recv_size >= 0 && - (*calld->recv_message)->length > (size_t)chand->max_recv_size) { + if (*calld->recv_message != NULL && calld->max_recv_size >= 0 && + (*calld->recv_message)->length > (size_t)calld->max_recv_size) { char* message_string; gpr_asprintf(&message_string, "Received message larger than max (%u vs. %d)", - (*calld->recv_message)->length, chand->max_recv_size); + (*calld->recv_message)->length, calld->max_recv_size); gpr_slice message = gpr_slice_from_copied_string(message_string); gpr_free(message_string); - grpc_call_element_send_cancel_with_message( + grpc_call_element_send_close_with_message( exec_ctx, elem, GRPC_STATUS_INVALID_ARGUMENT, &message); } // Invoke the next callback. @@ -87,16 +91,15 @@ static void start_transport_stream_op(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, grpc_transport_stream_op* op) { call_data* calld = elem->call_data; - channel_data* chand = elem->channel_data; // Check max send message size. - if (op->send_message != NULL && chand->max_send_size >= 0 && - op->send_message->length > (size_t)chand->max_send_size) { + if (op->send_message != NULL && calld->max_send_size >= 0 && + op->send_message->length > (size_t)calld->max_send_size) { char* message_string; gpr_asprintf(&message_string, "Sent message larger than max (%u vs. %d)", - op->send_message->length, chand->max_send_size); + op->send_message->length, calld->max_send_size); gpr_slice message = gpr_slice_from_copied_string(message_string); gpr_free(message_string); - grpc_call_element_send_cancel_with_message( + grpc_call_element_send_close_with_message( exec_ctx, elem, GRPC_STATUS_INVALID_ARGUMENT, &message); } // Inject callback for receiving a message. @@ -113,9 +116,37 @@ static void start_transport_stream_op(grpc_exec_ctx* exec_ctx, static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem, grpc_call_element_args* args) { + channel_data* chand = elem->channel_data; call_data* calld = elem->call_data; calld->next_recv_message_ready = NULL; grpc_closure_init(&calld->recv_message_ready, recv_message_ready, elem); + // Get max sizes from channel data, then merge in per-method config values. + // Note: Per-method config is only available on the client, so we + // apply the max request size to the send limit and the max response + // size to the receive limit. + calld->max_send_size = chand->max_send_size; + calld->max_recv_size = chand->max_recv_size; + if (chand->method_config_table != NULL) { + grpc_method_config* method_config = + grpc_method_config_table_get_method_config(chand->method_config_table, + args->path); + if (method_config != NULL) { + int32_t* max_request_message_bytes = + grpc_method_config_get_max_request_message_bytes(method_config); + if (max_request_message_bytes != NULL && + (*max_request_message_bytes < calld->max_send_size || + calld->max_send_size < 0)) { + calld->max_send_size = *max_request_message_bytes; + } + int32_t* max_response_message_bytes = + grpc_method_config_get_max_response_message_bytes(method_config); + if (max_response_message_bytes != NULL && + (*max_response_message_bytes < calld->max_recv_size || + calld->max_recv_size < 0)) { + calld->max_recv_size = *max_response_message_bytes; + } + } + } return GRPC_ERROR_NONE; } @@ -149,11 +180,22 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx, grpc_channel_arg_get_integer(&args->channel_args->args[i], options); } } + // Get method config table from channel args. + const grpc_arg *channel_arg = grpc_channel_args_find( + args->channel_args, GRPC_ARG_SERVICE_CONFIG); + if (channel_arg != NULL) { + GPR_ASSERT(channel_arg->type == GRPC_ARG_POINTER); + chand->method_config_table = grpc_method_config_table_ref( + (grpc_method_config_table *)channel_arg->value.pointer.p); + } } // Destructor for channel_data. static void destroy_channel_elem(grpc_exec_ctx* exec_ctx, - grpc_channel_element* elem) {} + grpc_channel_element* elem) { + channel_data* chand = elem->channel_data; + grpc_method_config_table_unref(chand->method_config_table); +} const grpc_channel_filter grpc_message_size_filter = { start_transport_stream_op, diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c index e366961936..31c80260f8 100644 --- a/src/core/lib/iomgr/error.c +++ b/src/core/lib/iomgr/error.c @@ -324,6 +324,64 @@ const char *grpc_error_get_str(grpc_error *err, grpc_error_strs which) { return gpr_avl_get(err->strs, (void *)(uintptr_t)which); } +typedef struct { + grpc_error *error; + grpc_status_code code; + const char *msg; +} special_error_status_map; +static special_error_status_map error_status_map[] = { + {GRPC_ERROR_NONE, GRPC_STATUS_OK, ""}, + {GRPC_ERROR_CANCELLED, GRPC_STATUS_CANCELLED, "RPC cancelled"}, + {GRPC_ERROR_OOM, GRPC_STATUS_RESOURCE_EXHAUSTED, "Out of memory"}, +}; + +static grpc_error *recursively_find_error_with_status(grpc_error *error, + intptr_t *status) { + // If the error itself has a status code, return it. + if (grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, status)) { + return error; + } + // Otherwise, search through its children. + intptr_t key = 0; + while (true) { + grpc_error *child_error = gpr_avl_get(error->errs, (void *)key++); + if (child_error == NULL) break; + grpc_error *result = + recursively_find_error_with_status(child_error, status); + if (result != NULL) return result; + } + return NULL; +} + +void grpc_error_get_status(grpc_error *error, grpc_status_code *code, + const char **msg) { + // Handle special errors via the static map. + for (size_t i = 0; i < GPR_ARRAY_SIZE(error_status_map); ++i) { + if (error == error_status_map[i].error) { + *code = error_status_map[i].code; + *msg = error_status_map[i].msg; + return; + } + } + // Populate code. + // Start with the parent error and recurse through the tree of children + // until we find the first one that has a status code. + intptr_t status = GRPC_STATUS_UNKNOWN; // Default in case we don't find one. + grpc_error *found_error = recursively_find_error_with_status(error, &status); + *code = (grpc_status_code)status; + // Now populate msg. + // If we found an error with a status code above, use that; otherwise, + // fall back to using the parent error. + if (found_error == NULL) found_error = error; + // If the error has a status message, use it. Otherwise, fall back to + // the error description. + *msg = grpc_error_get_str(found_error, GRPC_ERROR_STR_GRPC_MESSAGE); + if (*msg == NULL) { + *msg = grpc_error_get_str(found_error, GRPC_ERROR_STR_DESCRIPTION); + if (*msg == NULL) *msg = "uknown error"; // Just in case. + } +} + grpc_error *grpc_error_add_child(grpc_error *src, grpc_error *child) { GPR_TIMER_BEGIN("grpc_error_add_child", 0); grpc_error *new = copy_error_and_unref(src); diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h index 6c769accdb..7e2fd7a3bd 100644 --- a/src/core/lib/iomgr/error.h +++ b/src/core/lib/iomgr/error.h @@ -37,6 +37,7 @@ #include <stdbool.h> #include <stdint.h> +#include <grpc/status.h> #include <grpc/support/time.h> /// Opaque representation of an error. @@ -175,6 +176,13 @@ grpc_error *grpc_error_set_str(grpc_error *src, grpc_error_strs which, /// Returns NULL if the specified string is not set. /// Caller does NOT own return value. const char *grpc_error_get_str(grpc_error *error, grpc_error_strs which); + +/// A utility function to get the status code and message to be returned +/// to the application. If not set in the top-level message, looks +/// through child errors until it finds the first one with these attributes. +void grpc_error_get_status(grpc_error *error, grpc_status_code *code, + const char **msg); + /// Add a child error: an error that is believed to have contributed to this /// error occurring. Allows root causing high level errors from lower level /// errors that contributed to them. diff --git a/src/core/lib/iomgr/tcp_client_posix.c b/src/core/lib/iomgr/tcp_client_posix.c index 80c7a3f128..3496b6094f 100644 --- a/src/core/lib/iomgr/tcp_client_posix.c +++ b/src/core/lib/iomgr/tcp_client_posix.c @@ -146,61 +146,57 @@ static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) { grpc_timer_cancel(exec_ctx, &ac->alarm); gpr_mu_lock(&ac->mu); - if (error == GRPC_ERROR_NONE) { - do { - so_error_size = sizeof(so_error); - err = getsockopt(grpc_fd_wrapped_fd(fd), SOL_SOCKET, SO_ERROR, &so_error, - &so_error_size); - } while (err < 0 && errno == EINTR); - if (err < 0) { - error = GRPC_OS_ERROR(errno, "getsockopt"); - goto finish; - } else if (so_error != 0) { - if (so_error == ENOBUFS) { - /* We will get one of these errors if we have run out of - memory in the kernel for the data structures allocated - when you connect a socket. If this happens it is very - likely that if we wait a little bit then try again the - connection will work (since other programs or this - program will close their network connections and free up - memory). This does _not_ indicate that there is anything - wrong with the server we are connecting to, this is a - local problem. - - If you are looking at this code, then chances are that - your program or another program on the same computer - opened too many network connections. The "easy" fix: - don't do that! */ - gpr_log(GPR_ERROR, "kernel out of buffers"); - gpr_mu_unlock(&ac->mu); - grpc_fd_notify_on_write(exec_ctx, fd, &ac->write_closure); - return; - } else { - switch (so_error) { - case ECONNREFUSED: - error = grpc_error_set_int(error, GRPC_ERROR_INT_ERRNO, errno); - error = grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, - "Connection refused"); - break; - default: - error = GRPC_OS_ERROR(errno, "getsockopt(SO_ERROR)"); - break; - } - goto finish; - } - } else { - grpc_pollset_set_del_fd(exec_ctx, ac->interested_parties, fd); - *ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str); - fd = NULL; - goto finish; - } - } else { + if (error != GRPC_ERROR_NONE) { error = grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, "Timeout occurred"); goto finish; } - GPR_UNREACHABLE_CODE(return ); + do { + so_error_size = sizeof(so_error); + err = getsockopt(grpc_fd_wrapped_fd(fd), SOL_SOCKET, SO_ERROR, &so_error, + &so_error_size); + } while (err < 0 && errno == EINTR); + if (err < 0) { + error = GRPC_OS_ERROR(errno, "getsockopt"); + goto finish; + } + + switch (so_error) { + case 0: + grpc_pollset_set_del_fd(exec_ctx, ac->interested_parties, fd); + *ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str); + fd = NULL; + break; + case ENOBUFS: + /* We will get one of these errors if we have run out of + memory in the kernel for the data structures allocated + when you connect a socket. If this happens it is very + likely that if we wait a little bit then try again the + connection will work (since other programs or this + program will close their network connections and free up + memory). This does _not_ indicate that there is anything + wrong with the server we are connecting to, this is a + local problem. + + If you are looking at this code, then chances are that + your program or another program on the same computer + opened too many network connections. The "easy" fix: + don't do that! */ + gpr_log(GPR_ERROR, "kernel out of buffers"); + gpr_mu_unlock(&ac->mu); + grpc_fd_notify_on_write(exec_ctx, fd, &ac->write_closure); + return; + case ECONNREFUSED: + /* This error shouldn't happen for anything other than connect(). */ + error = GRPC_OS_ERROR(so_error, "connect"); + break; + default: + /* We don't really know which syscall triggered the problem here, + so punt by reporting getsockopt(). */ + error = GRPC_OS_ERROR(so_error, "getsockopt(SO_ERROR)"); + break; + } finish: if (fd != NULL) { diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c index 979e0aeaea..0af72547e3 100644 --- a/src/core/lib/surface/call.c +++ b/src/core/lib/surface/call.c @@ -126,8 +126,6 @@ struct grpc_call { /* client or server call */ bool is_client; - /* is the alarm set */ - bool have_alarm; /** has grpc_call_destroy been called */ bool destroy_called; /** flag indicating that cancellation is inherited */ @@ -170,9 +168,6 @@ struct grpc_call { /* Contexts for various subsystems (security, tracing, ...). */ grpc_call_context_element context[GRPC_CONTEXT_COUNT]; - /* Deadline alarm - if have_alarm is non-zero */ - grpc_timer alarm; - /* for the client, extra metadata is initial metadata; for the server, it's trailing metadata */ grpc_linked_mdelem send_extra_metadata[MAX_SEND_EXTRA_METADATA_COUNT]; @@ -215,8 +210,6 @@ struct grpc_call { #define CALL_FROM_TOP_ELEM(top_elem) \ CALL_FROM_CALL_STACK(grpc_call_stack_from_top_element(top_elem)) -static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call, - gpr_timespec deadline); static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call, grpc_transport_stream_op *op); static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c, @@ -249,10 +242,14 @@ grpc_call *grpc_call_create( /* Always support no compression */ GPR_BITSET(&call->encodings_accepted_by_peer, GRPC_COMPRESS_NONE); call->is_client = server_transport_data == NULL; + grpc_mdstr *path = NULL; if (call->is_client) { GPR_ASSERT(add_initial_metadata_count < MAX_SEND_EXTRA_METADATA_COUNT); for (i = 0; i < add_initial_metadata_count; i++) { call->send_extra_metadata[i].md = add_initial_metadata[i]; + if (add_initial_metadata[i]->key == GRPC_MDSTR_PATH) { + path = GRPC_MDSTR_REF(add_initial_metadata[i]->value); + } } call->send_extra_metadata_count = (int)add_initial_metadata_count; } else { @@ -264,39 +261,8 @@ grpc_call *grpc_call_create( call->metadata_batch[i][j].deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); } } - call->send_deadline = - gpr_convert_clock_type(send_deadline, GPR_CLOCK_MONOTONIC); - GRPC_CHANNEL_INTERNAL_REF(channel, "call"); - /* initial refcount dropped by grpc_call_destroy */ - grpc_error *error = grpc_call_stack_init( - &exec_ctx, channel_stack, 1, destroy_call, call, call->context, - server_transport_data, CALL_STACK_FROM_CALL(call)); - if (error != GRPC_ERROR_NONE) { - intptr_t status; - if (!grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status)) - status = GRPC_STATUS_UNKNOWN; - const char *error_str = - grpc_error_get_str(error, GRPC_ERROR_STR_DESCRIPTION); - close_with_status(&exec_ctx, call, (grpc_status_code)status, - error_str == NULL ? "unknown error" : error_str); - GRPC_ERROR_UNREF(error); - } - if (cq != NULL) { - GPR_ASSERT( - pollset_set_alternative == NULL && - "Only one of 'cq' and 'pollset_set_alternative' should be non-NULL."); - GRPC_CQ_INTERNAL_REF(cq, "bind"); - call->pollent = - grpc_polling_entity_create_from_pollset(grpc_cq_pollset(cq)); - } - if (pollset_set_alternative != NULL) { - call->pollent = - grpc_polling_entity_create_from_pollset_set(pollset_set_alternative); - } - if (!grpc_polling_entity_is_empty(&call->pollent)) { - grpc_call_stack_set_pollset_or_pollset_set( - &exec_ctx, CALL_STACK_FROM_CALL(call), &call->pollent); - } + send_deadline = gpr_convert_clock_type(send_deadline, GPR_CLOCK_MONOTONIC); + if (parent_call != NULL) { GRPC_CALL_INTERNAL_REF(parent_call, "child"); GPR_ASSERT(call->is_client); @@ -338,10 +304,40 @@ grpc_call *grpc_call_create( gpr_mu_unlock(&parent_call->mu); } - if (gpr_time_cmp(send_deadline, gpr_inf_future(send_deadline.clock_type)) != - 0) { - set_deadline_alarm(&exec_ctx, call, send_deadline); + + call->send_deadline = send_deadline; + + GRPC_CHANNEL_INTERNAL_REF(channel, "call"); + /* initial refcount dropped by grpc_call_destroy */ + grpc_error *error = grpc_call_stack_init( + &exec_ctx, channel_stack, 1, destroy_call, call, call->context, + server_transport_data, path, send_deadline, CALL_STACK_FROM_CALL(call)); + if (error != GRPC_ERROR_NONE) { + grpc_status_code status; + const char *error_str; + grpc_error_get_status(error, &status, &error_str); + close_with_status(&exec_ctx, call, status, error_str); + GRPC_ERROR_UNREF(error); } + if (cq != NULL) { + GPR_ASSERT( + pollset_set_alternative == NULL && + "Only one of 'cq' and 'pollset_set_alternative' should be non-NULL."); + GRPC_CQ_INTERNAL_REF(cq, "bind"); + call->pollent = + grpc_polling_entity_create_from_pollset(grpc_cq_pollset(cq)); + } + if (pollset_set_alternative != NULL) { + call->pollent = + grpc_polling_entity_create_from_pollset_set(pollset_set_alternative); + } + if (!grpc_polling_entity_is_empty(&call->pollent)) { + grpc_call_stack_set_pollset_or_pollset_set( + &exec_ctx, CALL_STACK_FROM_CALL(call), &call->pollent); + } + + if (path != NULL) GRPC_MDSTR_UNREF(path); + grpc_exec_ctx_finish(&exec_ctx); GPR_TIMER_END("grpc_call_create", 0); return call; @@ -454,20 +450,11 @@ static void set_status_details(grpc_call *call, status_source source, static void set_status_from_error(grpc_call *call, status_source source, grpc_error *error) { - intptr_t status; - if (grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status)) { - set_status_code(call, source, (uint32_t)status); - } else { - set_status_code(call, source, GRPC_STATUS_INTERNAL); - } - const char *msg = grpc_error_get_str(error, GRPC_ERROR_STR_GRPC_MESSAGE); - bool free_msg = false; - if (msg == NULL) { - free_msg = true; - msg = grpc_error_string(error); - } + grpc_status_code status; + const char *msg; + grpc_error_get_status(error, &status, &msg); + set_status_code(call, source, (uint32_t)status); set_status_details(call, source, grpc_mdstr_from_string(msg)); - if (free_msg) grpc_error_free_string(msg); } static void set_incoming_compression_algorithm( @@ -740,9 +727,6 @@ void grpc_call_destroy(grpc_call *c) { gpr_mu_lock(&c->mu); GPR_ASSERT(!c->destroy_called); c->destroy_called = 1; - if (c->have_alarm) { - grpc_timer_cancel(&exec_ctx, &c->alarm); - } cancel = !c->received_final_op; gpr_mu_unlock(&c->mu); if (cancel) grpc_call_cancel(c, NULL); @@ -780,7 +764,6 @@ typedef struct termination_closure { grpc_closure closure; grpc_call *call; grpc_error *error; - grpc_closure *op_closure; enum { TC_CANCEL, TC_CLOSE } type; grpc_transport_stream_op op; } termination_closure; @@ -797,7 +780,6 @@ static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp, break; } GRPC_ERROR_UNREF(tc->error); - grpc_exec_ctx_sched(exec_ctx, tc->op_closure, GRPC_ERROR_NONE, NULL); gpr_free(tc); } @@ -817,7 +799,6 @@ static void send_close(grpc_exec_ctx *exec_ctx, void *tcp, grpc_error *error) { tc->op.close_error = tc->error; /* reuse closure to catch completion */ grpc_closure_init(&tc->closure, done_termination, tc); - tc->op_closure = tc->op.on_complete; tc->op.on_complete = &tc->closure; execute_op(exec_ctx, tc->call, &tc->op); } @@ -900,32 +881,6 @@ grpc_call *grpc_call_from_top_element(grpc_call_element *elem) { return CALL_FROM_TOP_ELEM(elem); } -static void call_alarm(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { - grpc_call *call = arg; - gpr_mu_lock(&call->mu); - call->have_alarm = 0; - if (error != GRPC_ERROR_CANCELLED) { - cancel_with_status(exec_ctx, call, GRPC_STATUS_DEADLINE_EXCEEDED, - "Deadline Exceeded"); - } - gpr_mu_unlock(&call->mu); - GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "alarm"); -} - -static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call, - gpr_timespec deadline) { - if (call->have_alarm) { - gpr_log(GPR_ERROR, "Attempt to set deadline alarm twice"); - assert(0); - return; - } - GRPC_CALL_INTERNAL_REF(call, "alarm"); - call->have_alarm = 1; - call->send_deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); - grpc_timer_init(exec_ctx, &call->alarm, call->send_deadline, call_alarm, call, - gpr_now(GPR_CLOCK_MONOTONIC)); -} - /* we offset status by a small amount when storing it into transport metadata as metadata cannot store a 0 value (which is used as OK for grpc_status_codes */ @@ -1267,9 +1222,8 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx, if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) != 0 && !call->is_client) { - GPR_TIMER_BEGIN("set_deadline_alarm", 0); - set_deadline_alarm(exec_ctx, call, md->deadline); - GPR_TIMER_END("set_deadline_alarm", 0); + call->send_deadline = + gpr_convert_clock_type(md->deadline, GPR_CLOCK_MONOTONIC); } } @@ -1298,9 +1252,17 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, GRPC_ERROR_REF(error); gpr_mu_lock(&call->mu); + + // If the error has an associated status code, set the call's status. + intptr_t status; + if (error != GRPC_ERROR_NONE && + grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status)) { + set_status_from_error(call, STATUS_FROM_CORE, error); + } + if (bctl->send_initial_metadata) { if (error != GRPC_ERROR_NONE) { - set_status_code(call, STATUS_FROM_CORE, GRPC_STATUS_UNAVAILABLE); + set_status_from_error(call, STATUS_FROM_CORE, error); } grpc_metadata_batch_destroy( &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]); @@ -1318,9 +1280,6 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, grpc_metadata_batch_filter(md, recv_trailing_filter, call); call->received_final_op = true; - if (call->have_alarm) { - grpc_timer_cancel(exec_ctx, &call->alarm); - } /* propagate cancellation to any interested children */ child_call = call->first_child; if (child_call != NULL) { diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index 3049284f68..4dbf3aae63 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -57,6 +57,8 @@ typedef struct grpc_cq_completion { uintptr_t next; } grpc_cq_completion; +//#define GRPC_CQ_REF_COUNT_DEBUG + #ifdef GRPC_CQ_REF_COUNT_DEBUG void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason, const char *file, int line); diff --git a/src/core/lib/surface/init.c b/src/core/lib/surface/init.c index 64bdfc3446..289f4ce8e8 100644 --- a/src/core/lib/surface/init.c +++ b/src/core/lib/surface/init.c @@ -43,6 +43,7 @@ #include "src/core/lib/channel/channel_stack.h" #include "src/core/lib/channel/compress_filter.h" #include "src/core/lib/channel/connected_channel.h" +#include "src/core/lib/channel/deadline_filter.h" #include "src/core/lib/channel/http_client_filter.h" #include "src/core/lib/channel/http_server_filter.h" #include "src/core/lib/channel/message_size_filter.h" @@ -100,6 +101,12 @@ static bool maybe_add_http_filter(grpc_channel_stack_builder *builder, static void register_builtin_channel_init() { grpc_channel_init_register_stage( + GRPC_CLIENT_DIRECT_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, + prepend_filter, (void *)&grpc_client_deadline_filter); + grpc_channel_init_register_stage( + GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, prepend_filter, + (void *)&grpc_server_deadline_filter); + grpc_channel_init_register_stage( GRPC_CLIENT_SUBCHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, prepend_filter, (void *)&grpc_message_size_filter); grpc_channel_init_register_stage( diff --git a/src/core/lib/transport/hashtable.c b/src/core/lib/transport/hashtable.c new file mode 100644 index 0000000000..89e2640969 --- /dev/null +++ b/src/core/lib/transport/hashtable.c @@ -0,0 +1,139 @@ +// +// 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 "src/core/lib/transport/hashtable.h" + +#include <stdbool.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> + +#include "src/core/lib/transport/metadata.h" + +struct grpc_hash_table { + gpr_refcount refs; + size_t num_entries; + grpc_hash_table_entry* entries; +}; + +// Helper function for insert and get operations that performs quadratic +// probing (https://en.wikipedia.org/wiki/Quadratic_probing). +static size_t grpc_hash_table_find_index( + grpc_hash_table* table, grpc_mdstr* key, bool find_empty) { + for (size_t i = 0; i < table->num_entries; ++i) { + const size_t idx = (key->hash + i * i) % table->num_entries; + if (table->entries[idx].key == NULL) + return find_empty ? idx : table->num_entries; + if (table->entries[idx].key == key) return idx; + } + return table->num_entries; // Not found. +} + +static void grpc_hash_table_add(grpc_hash_table* table, grpc_mdstr* key, + void* value, + const grpc_hash_table_vtable* vtable) { + GPR_ASSERT(value != NULL); + const size_t idx = + grpc_hash_table_find_index(table, key, true /* find_empty */); + // This can happen if the table is full. + GPR_ASSERT(idx != table->num_entries); + grpc_hash_table_entry* entry = &table->entries[idx]; + entry->key = GRPC_MDSTR_REF(key); + entry->value = vtable->copy_value(value); + entry->vtable = vtable; +} + +grpc_hash_table* grpc_hash_table_create(size_t num_entries, + grpc_hash_table_entry* entries) { + grpc_hash_table* table = gpr_malloc(sizeof(*table)); + memset(table, 0, sizeof(*table)); + gpr_ref_init(&table->refs, 1); + // Quadratic chaining gets best performance when the table is no more + // than half full. + table->num_entries = num_entries * 2; + const size_t entry_size = sizeof(grpc_hash_table_entry) * table->num_entries; + table->entries = gpr_malloc(entry_size); + memset(table->entries, 0, entry_size); + for (size_t i = 0; i < num_entries; ++i) { + grpc_hash_table_entry* entry = &entries[i]; + grpc_hash_table_add(table, entry->key, entry->value, entry->vtable); + } + return table; +} + +grpc_hash_table* grpc_hash_table_ref(grpc_hash_table* table) { + if (table != NULL) gpr_ref(&table->refs); + return table; +} + +int grpc_hash_table_unref(grpc_hash_table* table) { + if (table != NULL && gpr_unref(&table->refs)) { + for (size_t i = 0; i < table->num_entries; ++i) { + grpc_hash_table_entry* entry = &table->entries[i]; + if (entry->key != NULL) { + GRPC_MDSTR_UNREF(entry->key); + entry->vtable->destroy_value(entry->value); + } + } + gpr_free(table->entries); + gpr_free(table); + return 1; + } + return 0; +} + +void* grpc_hash_table_get(grpc_hash_table* table, grpc_mdstr* key) { + const size_t idx = + grpc_hash_table_find_index(table, key, false /* find_empty */); + if (idx == table->num_entries) return NULL; // Not found. + return table->entries[idx].value; +} + +int grpc_hash_table_cmp(grpc_hash_table* table1, grpc_hash_table* table2) { + // Compare by num_entries. + if (table1->num_entries < table2->num_entries) return -1; + if (table1->num_entries > table2->num_entries) return 1; + for (size_t i = 0; i < table1->num_entries; ++i) { + grpc_hash_table_entry* e1 = &table1->entries[i]; + grpc_hash_table_entry* e2 = &table2->entries[i]; + // Compare keys by hash value. + if (e1->key->hash < e2->key->hash) return -1; + if (e1->key->hash > e2->key->hash) return 1; + // Compare by vtable (pointer equality). + if (e1->vtable < e2->vtable) return -1; + if (e1->vtable > e2->vtable) return 1; + // Compare values via vtable. + const int value_result = e1->vtable->compare_value(e1->value, e2->value); + if (value_result != 0) return value_result; + } + return 0; +} diff --git a/src/core/lib/transport/hashtable.h b/src/core/lib/transport/hashtable.h new file mode 100644 index 0000000000..93f0642c17 --- /dev/null +++ b/src/core/lib/transport/hashtable.h @@ -0,0 +1,82 @@ +/* + * 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_LIB_TRANSPORT_HASHTABLE_H +#define GRPC_CORE_LIB_TRANSPORT_HASHTABLE_H + +#include "src/core/lib/transport/metadata.h" + +/** Hash table implementation. + * + * This implementation uses open addressing + * (https://en.wikipedia.org/wiki/Open_addressing) with quadratic + * probing (https://en.wikipedia.org/wiki/Quadratic_probing). + * This means that the hash table is of fixed size and cannot contain + * more than that number of elements. + * + * The keys are grpc_mdstr objects. The values are arbitrary pointers + * with a common vtable. + * + * Hash tables are intentionally immutable, to avoid the need for locking. + */ + +typedef struct grpc_hash_table grpc_hash_table; + +typedef struct grpc_hash_table_vtable { + void (*destroy_value)(void* value); + void* (*copy_value)(void* value); + int (*compare_value)(void* value1, void* value2); +} grpc_hash_table_vtable; + +typedef struct grpc_hash_table_entry { + grpc_mdstr* key; + void* value; /* Must not be NULL. */ + const grpc_hash_table_vtable* vtable; +} grpc_hash_table_entry; + +/** Creates a new hash table of containing \a entries, which is an array + of length \a num_entries. + Creates its own copy of all keys and values from \a entries. */ +grpc_hash_table* grpc_hash_table_create(size_t num_entries, + grpc_hash_table_entry* entries); + +grpc_hash_table* grpc_hash_table_ref(grpc_hash_table* table); +/** Returns 1 when \a table is destroyed. */ +int grpc_hash_table_unref(grpc_hash_table* table); + +/** Returns the value from \a table associated with \a key. + Returns NULL if \a key is not found. */ +void* grpc_hash_table_get(grpc_hash_table* table, grpc_mdstr* key); + +/** Compares two hash tables. */ +int grpc_hash_table_cmp(grpc_hash_table* table1, grpc_hash_table* table2); + +#endif /* GRPC_CORE_LIB_TRANSPORT_HASHTABLE_H */ diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c index 08f9d7e8d9..82fc605218 100644 --- a/src/core/lib/transport/transport.c +++ b/src/core/lib/transport/transport.c @@ -47,7 +47,7 @@ void grpc_stream_ref(grpc_stream_refcount *refcount, const char *reason) { gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count); gpr_log(GPR_DEBUG, "%s %p:%p REF %d->%d %s", refcount->object_type, - refcount, refcount->destroy.cb_arg, val, val + 1, reason); + refcount, refcount->destroy.cb_arg, (int)val, (int)val + 1, reason); #else void grpc_stream_ref(grpc_stream_refcount *refcount) { #endif @@ -59,7 +59,7 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount, const char *reason) { gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count); gpr_log(GPR_DEBUG, "%s %p:%p UNREF %d->%d %s", refcount->object_type, - refcount, refcount->destroy.cb_arg, val, val - 1, reason); + refcount, refcount->destroy.cb_arg, (int)val, (int)val - 1, reason); #else void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount) { @@ -224,7 +224,7 @@ void grpc_transport_stream_op_add_cancellation_with_message( error = GRPC_ERROR_CREATE("Call cancelled"); } error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, status); - add_error(op, &op->close_error, error); + add_error(op, &op->cancel_error, error); } void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op, |