diff options
author | 2016-10-19 15:19:00 -0700 | |
---|---|---|
committer | 2016-10-19 15:19:00 -0700 | |
commit | 10edf68eec9ec2d5520cb2cb88010090257ea946 (patch) | |
tree | 86f985cea32f234371c6f4b868312e2186b4d569 /src/core/ext/client_config | |
parent | b7a83818f8dc3d8f01a62613471951446fff87c6 (diff) | |
parent | 70ee74c4798afe781788d40afe1923feff37d657 (diff) |
Merge branch 'master' into uv_core_transport
Diffstat (limited to 'src/core/ext/client_config')
-rw-r--r-- | src/core/ext/client_config/client_channel.c | 193 | ||||
-rw-r--r-- | src/core/ext/client_config/method_config.c | 296 | ||||
-rw-r--r-- | src/core/ext/client_config/method_config.h | 116 | ||||
-rw-r--r-- | src/core/ext/client_config/resolver_result.h | 9 | ||||
-rw-r--r-- | src/core/ext/client_config/subchannel.c | 4 | ||||
-rw-r--r-- | src/core/ext/client_config/subchannel.h | 3 |
6 files changed, 592 insertions, 29 deletions
diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c index cbf79afa17..beaa9637c3 100644 --- a/src/core/ext/client_config/client_channel.c +++ b/src/core/ext/client_config/client_channel.c @@ -43,6 +43,7 @@ #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" @@ -53,6 +54,9 @@ #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 */ @@ -68,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; @@ -172,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"); @@ -220,6 +226,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; } @@ -232,6 +245,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); @@ -392,6 +409,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); @@ -424,7 +444,16 @@ typedef struct client_channel_call_data { // 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; + + grpc_mdstr *path; // Request path. + gpr_timespec call_start_time; gpr_timespec deadline; + enum { + WAIT_FOR_READY_UNSET, + WAIT_FOR_READY_FALSE, + WAIT_FOR_READY_TRUE + } wait_for_ready_from_service_config; + grpc_closure read_service_config; grpc_error *cancel_error; @@ -532,10 +561,11 @@ static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, 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, calld->deadline, - &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; @@ -584,11 +614,15 @@ static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, /* cancelled, do nothing */ } else if (error != GRPC_ERROR_NONE) { 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, - GRPC_ERROR_NONE)) { - grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE, NULL); + } else { + call_data *calld = cpa->elem->call_data; + gpr_mu_lock(&calld->mu); + if (pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata, + cpa->initial_metadata_flags, cpa->connected_subchannel, + cpa->on_ready, GRPC_ERROR_NONE)) { + grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE, NULL); + } + gpr_mu_unlock(&calld->mu); } gpr_free(cpa); } @@ -631,18 +665,33 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, GPR_ASSERT(error == GRPC_ERROR_NONE); if (chand->lb_policy != NULL) { grpc_lb_policy *lb_policy = chand->lb_policy; - int r; GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel"); gpr_mu_unlock(&chand->mu); + // If the application explicitly set wait_for_ready, use that. + // Otherwise, if the service config specified a value for this + // method, use that. + const bool wait_for_ready_set_from_api = + initial_metadata_flags & + GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET; + const bool wait_for_ready_set_from_service_config = + calld->wait_for_ready_from_service_config != WAIT_FOR_READY_UNSET; + if (!wait_for_ready_set_from_api && + wait_for_ready_set_from_service_config) { + if (calld->wait_for_ready_from_service_config == WAIT_FOR_READY_TRUE) { + initial_metadata_flags |= GRPC_INITIAL_METADATA_WAIT_FOR_READY; + } else { + initial_metadata_flags &= ~GRPC_INITIAL_METADATA_WAIT_FOR_READY; + } + } // TODO(dgq): make this deadline configurable somehow. const grpc_lb_policy_pick_args inputs = { initial_metadata, initial_metadata_flags, &calld->lb_token_mdelem, gpr_inf_future(GPR_CLOCK_MONOTONIC)}; - r = grpc_lb_policy_pick(exec_ctx, lb_policy, &inputs, connected_subchannel, - NULL, on_ready); + const bool result = grpc_lb_policy_pick( + exec_ctx, lb_policy, &inputs, connected_subchannel, NULL, on_ready); GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "pick_subchannel"); GPR_TIMER_END("pick_subchannel", 0); - return r; + return result; } if (chand->resolver != NULL && !chand->started_resolving) { chand->started_resolving = true; @@ -768,8 +817,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, calld->deadline, - &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)); @@ -786,13 +835,69 @@ retry: GPR_TIMER_END("cc_start_transport_stream_op", 0); } +// Gets data from the service config. Invoked when the resolver returns +// its initial result. +static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_call_element *elem = arg; + channel_data *chand = elem->channel_data; + call_data *calld = elem->call_data; + // If this is an error, there's no point in looking at the service config. + if (error == GRPC_ERROR_NONE) { + // Get the method config table from channel data. + gpr_mu_lock(&chand->mu); + grpc_method_config_table *method_config_table = NULL; + if (chand->method_config_table != NULL) { + method_config_table = + grpc_method_config_table_ref(chand->method_config_table); + } + gpr_mu_unlock(&chand->mu); + // If the method config table was present, use it. + if (method_config_table != NULL) { + const grpc_method_config *method_config = + grpc_method_config_table_get_method_config(method_config_table, + calld->path); + if (method_config != NULL) { + const gpr_timespec *per_method_timeout = + grpc_method_config_get_timeout(method_config); + const bool *wait_for_ready = + grpc_method_config_get_wait_for_ready(method_config); + if (per_method_timeout != NULL || wait_for_ready != NULL) { + gpr_mu_lock(&calld->mu); + if (per_method_timeout != NULL) { + gpr_timespec per_method_deadline = + gpr_time_add(calld->call_start_time, *per_method_timeout); + if (gpr_time_cmp(per_method_deadline, calld->deadline) < 0) { + calld->deadline = per_method_deadline; + // Reset deadline timer. + grpc_deadline_state_reset(exec_ctx, elem, calld->deadline); + } + } + if (wait_for_ready != NULL) { + calld->wait_for_ready_from_service_config = + *wait_for_ready ? WAIT_FOR_READY_TRUE : WAIT_FOR_READY_FALSE; + } + gpr_mu_unlock(&calld->mu); + } + } + grpc_method_config_table_unref(method_config_table); + } + } + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config"); +} + /* Constructor for call_data */ 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; - grpc_deadline_state_init(exec_ctx, elem, args); - calld->deadline = args->deadline; + // Initialize data members. + grpc_deadline_state_init(exec_ctx, elem, args->call_stack); + calld->path = GRPC_MDSTR_REF(args->path); + calld->call_start_time = args->start_time; + calld->deadline = gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC); + calld->wait_for_ready_from_service_config = WAIT_FOR_READY_UNSET; calld->cancel_error = GRPC_ERROR_NONE; gpr_atm_rel_store(&calld->subchannel_call, 0); gpr_mu_init(&calld->mu); @@ -803,6 +908,55 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx, calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; calld->owning_call = args->call_stack; calld->pollent = NULL; + // If the resolver has already returned results, then we can access + // the service config parameters immediately. Otherwise, we need to + // defer that work until the resolver returns an initial result. + // TODO(roth): This code is almost but not quite identical to the code + // in read_service_config() above. It would be nice to find a way to + // combine them, to avoid having to maintain it twice. + gpr_mu_lock(&chand->mu); + if (chand->lb_policy != NULL) { + // We already have a resolver result, so check for service config. + if (chand->method_config_table != NULL) { + grpc_method_config_table *method_config_table = + grpc_method_config_table_ref(chand->method_config_table); + gpr_mu_unlock(&chand->mu); + grpc_method_config *method_config = + grpc_method_config_table_get_method_config(method_config_table, + args->path); + if (method_config != NULL) { + const gpr_timespec *per_method_timeout = + grpc_method_config_get_timeout(method_config); + if (per_method_timeout != NULL) { + gpr_timespec per_method_deadline = + gpr_time_add(calld->call_start_time, *per_method_timeout); + calld->deadline = gpr_time_min(calld->deadline, per_method_deadline); + } + const bool *wait_for_ready = + grpc_method_config_get_wait_for_ready(method_config); + if (wait_for_ready != NULL) { + calld->wait_for_ready_from_service_config = + *wait_for_ready ? WAIT_FOR_READY_TRUE : WAIT_FOR_READY_FALSE; + } + } + grpc_method_config_table_unref(method_config_table); + } else { + gpr_mu_unlock(&chand->mu); + } + } else { + // We don't yet have a resolver result, so register a callback to + // get the service config data once the resolver returns. + // Take a reference to the call stack to be owned by the callback. + GRPC_CALL_STACK_REF(calld->owning_call, "read_service_config"); + grpc_closure_init(&calld->read_service_config, read_service_config, elem); + grpc_closure_list_append(&chand->waiting_for_config_closures, + &calld->read_service_config, GRPC_ERROR_NONE); + gpr_mu_unlock(&chand->mu); + } + // Start the deadline timer with the current deadline value. If we + // do not yet have service config data, then the timer may be reset + // later. + grpc_deadline_state_start(exec_ctx, elem, calld->deadline); return GRPC_ERROR_NONE; } @@ -813,6 +967,7 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx, 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) { 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..f8a82323e7 --- /dev/null +++ b/src/core/ext/client_config/method_config.c @@ -0,0 +1,296 @@ +// +// 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/mdstr_hash_table.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_mdstr_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_mdstr_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_mdstr_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_mdstr_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)); + method_config->wait_for_ready_key = + grpc_mdstr_from_string(GRPC_METHOD_CONFIG_WAIT_FOR_READY); + method_config->timeout_key = + grpc_mdstr_from_string(GRPC_METHOD_CONFIG_TIMEOUT); + method_config->max_request_message_bytes_key = + grpc_mdstr_from_string(GRPC_METHOD_CONFIG_MAX_REQUEST_MESSAGE_BYTES); + method_config->max_response_message_bytes_key = + grpc_mdstr_from_string(GRPC_METHOD_CONFIG_MAX_RESPONSE_MESSAGE_BYTES); + grpc_mdstr_hash_table_entry entries[4]; + size_t num_entries = 0; + if (wait_for_ready != NULL) { + 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) { + 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) { + 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) { + 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_mdstr_hash_table_create(num_entries, entries); + return method_config; +} + +grpc_method_config* grpc_method_config_ref(grpc_method_config* method_config) { + grpc_mdstr_hash_table_ref(method_config->table); + return method_config; +} + +void grpc_method_config_unref(grpc_method_config* method_config) { + if (grpc_mdstr_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); + gpr_free(method_config); + } +} + +int grpc_method_config_cmp(const grpc_method_config* method_config1, + const grpc_method_config* method_config2) { + return grpc_mdstr_hash_table_cmp(method_config1->table, + method_config2->table); +} + +const bool* grpc_method_config_get_wait_for_ready( + const grpc_method_config* method_config) { + return grpc_mdstr_hash_table_get(method_config->table, + method_config->wait_for_ready_key); +} + +const gpr_timespec* grpc_method_config_get_timeout( + const grpc_method_config* method_config) { + return grpc_mdstr_hash_table_get(method_config->table, + method_config->timeout_key); +} + +const int32_t* grpc_method_config_get_max_request_message_bytes( + const grpc_method_config* method_config) { + return grpc_mdstr_hash_table_get( + method_config->table, method_config->max_request_message_bytes_key); +} + +const int32_t* grpc_method_config_get_max_response_message_bytes( + const grpc_method_config* method_config) { + return grpc_mdstr_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_mdstr_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_mdstr_hash_table_entry* hash_table_entries = + gpr_malloc(sizeof(grpc_mdstr_hash_table_entry) * 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; + } + grpc_method_config_table* method_config_table = + grpc_mdstr_hash_table_create(num_entries, hash_table_entries); + gpr_free(hash_table_entries); + return method_config_table; +} + +grpc_method_config_table* grpc_method_config_table_ref( + grpc_method_config_table* table) { + return grpc_mdstr_hash_table_ref(table); +} + +void grpc_method_config_table_unref(grpc_method_config_table* table) { + grpc_mdstr_hash_table_unref(table); +} + +int grpc_method_config_table_cmp(const grpc_method_config_table* table1, + const grpc_method_config_table* table2) { + return grpc_mdstr_hash_table_cmp(table1, table2); +} + +grpc_method_config* grpc_method_config_table_get_method_config( + const grpc_method_config_table* table, const grpc_mdstr* path) { + grpc_method_config* method_config = grpc_mdstr_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 = gpr_malloc(len + 2); // '*' and NUL + memcpy(buf, path_str, len); + buf[len] = '*'; + buf[len + 1] = '\0'; + grpc_mdstr* wildcard_path = grpc_mdstr_from_string(buf); + gpr_free(buf); + method_config = grpc_mdstr_hash_table_get(table, wildcard_path); + GRPC_MDSTR_UNREF(wildcard_path); + } + return 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 = 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..1302ac425d --- /dev/null +++ b/src/core/ext/client_config/method_config.h @@ -0,0 +1,116 @@ +// +// 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/mdstr_hash_table.h" +#include "src/core/lib/transport/metadata.h" + +/// Per-method configuration. +typedef struct grpc_method_config grpc_method_config; + +/// Creates a grpc_method_config with the specified parameters. +/// Any parameter may be NULL to indicate that the value is unset. +/// +/// \a wait_for_ready indicates whether the client should wait until the +/// request deadline for the channel to become ready, even if there is a +/// temporary failure before the deadline while attempting to connect. +/// +/// \a timeout indicates the timeout for calls. +/// +/// \a max_request_message_bytes and \a max_response_message_bytes +/// indicate the maximum sizes of the request (checked when sending) and +/// response (checked when receiving) messages. +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); + +/// Compares two grpc_method_configs. +/// The sort order is stable but undefined. +int grpc_method_config_cmp(const grpc_method_config* method_config1, + const grpc_method_config* method_config2); + +/// These methods return NULL if the requested field is unset. +/// The caller does NOT take ownership of the result. +const bool* grpc_method_config_get_wait_for_ready( + const grpc_method_config* method_config); +const gpr_timespec* grpc_method_config_get_timeout( + const grpc_method_config* method_config); +const int32_t* grpc_method_config_get_max_request_message_bytes( + const grpc_method_config* method_config); +const int32_t* grpc_method_config_get_max_response_message_bytes( + const grpc_method_config* method_config); + +/// A table of method configs. +typedef grpc_mdstr_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); + +/// Compares two grpc_method_config_tables. +/// The sort order is stable but undefined. +int grpc_method_config_table_cmp(const grpc_method_config_table* table1, + const grpc_method_config_table* table2); + +/// Gets the method config for the specified \a path, which should be of +/// the form "/service/method". +/// Returns NULL if the method has no config. +/// Caller does NOT own a reference to the result. +grpc_method_config* grpc_method_config_table_get_method_config( + const grpc_method_config_table* table, const 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.h b/src/core/ext/client_config/resolver_result.h index 414c2e2482..a7ea7c0f4b 100644 --- a/src/core/ext/client_config/resolver_result.h +++ b/src/core/ext/client_config/resolver_result.h @@ -52,22 +52,17 @@ typedef struct grpc_resolver_result grpc_resolver_result; 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 e52e5a2a32..b66a6556a2 100644 --- a/src/core/ext/client_config/subchannel.c +++ b/src/core/ext/client_config/subchannel.c @@ -701,7 +701,7 @@ 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, gpr_timespec deadline, + 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); @@ -709,7 +709,7 @@ grpc_error *grpc_connected_subchannel_create_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, deadline, 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 5b1910213c..7061ef17b8 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,7 @@ 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, gpr_timespec deadline, + grpc_polling_entity *pollent, grpc_mdstr *path, gpr_timespec deadline, grpc_subchannel_call **subchannel_call); /** process a transport level op */ |