diff options
Diffstat (limited to 'src/core/ext/client_channel')
36 files changed, 4888 insertions, 0 deletions
diff --git a/src/core/ext/client_channel/README.md b/src/core/ext/client_channel/README.md new file mode 100644 index 0000000000..eda01e3e71 --- /dev/null +++ b/src/core/ext/client_channel/README.md @@ -0,0 +1,66 @@ +Client Configuration Support for GRPC +===================================== + +This library provides high level configuration machinery to construct client +channels and load balance between them. + +Each grpc_channel is created with a grpc_resolver. It is the resolver's duty +to resolve a name into configuration data for the channel. Such configuration +data might include: + +- a list of (ip, port) addresses to connect to +- a load balancing policy to decide which server to send a request to +- a set of filters to mutate outgoing requests (say, by adding metadata) + +The resolver provides this data as a stream of grpc_resolver_result objects to +the channel. We represent configuration as a stream so that it can be changed +by the resolver during execution, by reacting to external events (such as a +new configuration file being pushed to some store). + + +Load Balancing +-------------- + +Load balancing configuration is provided by a grpc_lb_policy object, stored as +part of grpc_resolver_result. + +The primary job of the load balancing policies is to pick a target server given only the +initial metadata for a request. It does this by providing a grpc_subchannel +object to the owning channel. + + +Sub-Channels +------------ + +A sub-channel provides a connection to a server for a client channel. It has a +connectivity state like a regular channel, and so can be connected or +disconnected. This connectivity state can be used to inform load balancing +decisions (for example, by avoiding disconnected backends). + +Configured sub-channels are fully setup to participate in the grpc data plane. +Their behavior is specified by a set of grpc channel filters defined at their +construction. To customize this behavior, resolvers build +grpc_client_channel_factory objects, which use the decorator pattern to customize +construction arguments for concrete grpc_subchannel instances. + + +Naming for GRPC +=============== + +Names in GRPC are represented by a URI (as defined in +[RFC 3986](https://tools.ietf.org/html/rfc3986)). + +The following schemes are currently supported: + +dns:///host:port - dns schemes are currently supported so long as authority is + empty (authority based dns resolution is expected in a future + release) + +unix:path - the unix scheme is used to create and connect to unix domain + sockets - the authority must be empty, and the path + represents the absolute or relative path to the desired + socket + +ipv4:host:port - a pre-resolved ipv4 dotted decimal address/port combination + +ipv6:[host]:port - a pre-resolved ipv6 address/port combination diff --git a/src/core/ext/client_channel/channel_connectivity.c b/src/core/ext/client_channel/channel_connectivity.c new file mode 100644 index 0000000000..9797e66564 --- /dev/null +++ b/src/core/ext/client_channel/channel_connectivity.c @@ -0,0 +1,222 @@ +/* + * + * 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/lib/surface/channel.h" + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> + +#include "src/core/ext/client_channel/client_channel.h" +#include "src/core/lib/iomgr/timer.h" +#include "src/core/lib/surface/api_trace.h" +#include "src/core/lib/surface/completion_queue.h" + +grpc_connectivity_state grpc_channel_check_connectivity_state( + grpc_channel *channel, int try_to_connect) { + /* forward through to the underlying client channel */ + grpc_channel_element *client_channel_elem = + grpc_channel_stack_last_element(grpc_channel_get_channel_stack(channel)); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_connectivity_state state; + GRPC_API_TRACE( + "grpc_channel_check_connectivity_state(channel=%p, try_to_connect=%d)", 2, + (channel, try_to_connect)); + if (client_channel_elem->filter == &grpc_client_channel_filter) { + state = grpc_client_channel_check_connectivity_state( + &exec_ctx, client_channel_elem, try_to_connect); + grpc_exec_ctx_finish(&exec_ctx); + return state; + } + gpr_log(GPR_ERROR, + "grpc_channel_check_connectivity_state called on something that is " + "not a client channel, but '%s'", + client_channel_elem->filter->name); + grpc_exec_ctx_finish(&exec_ctx); + return GRPC_CHANNEL_SHUTDOWN; +} + +typedef enum { + WAITING, + CALLING_BACK, + CALLING_BACK_AND_FINISHED, + CALLED_BACK +} callback_phase; + +typedef struct { + gpr_mu mu; + callback_phase phase; + grpc_closure on_complete; + grpc_timer alarm; + grpc_connectivity_state state; + grpc_completion_queue *cq; + grpc_cq_completion completion_storage; + grpc_channel *channel; + void *tag; +} state_watcher; + +static void delete_state_watcher(grpc_exec_ctx *exec_ctx, state_watcher *w) { + grpc_channel_element *client_channel_elem = grpc_channel_stack_last_element( + grpc_channel_get_channel_stack(w->channel)); + if (client_channel_elem->filter == &grpc_client_channel_filter) { + GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, w->channel, + "watch_channel_connectivity"); + } else { + abort(); + } + gpr_mu_destroy(&w->mu); + gpr_free(w); +} + +static void finished_completion(grpc_exec_ctx *exec_ctx, void *pw, + grpc_cq_completion *ignored) { + int delete = 0; + state_watcher *w = pw; + gpr_mu_lock(&w->mu); + switch (w->phase) { + case WAITING: + case CALLED_BACK: + GPR_UNREACHABLE_CODE(return ); + case CALLING_BACK: + w->phase = CALLED_BACK; + break; + case CALLING_BACK_AND_FINISHED: + delete = 1; + break; + } + gpr_mu_unlock(&w->mu); + + if (delete) { + delete_state_watcher(exec_ctx, w); + } +} + +static void partly_done(grpc_exec_ctx *exec_ctx, state_watcher *w, + bool due_to_completion, grpc_error *error) { + int delete = 0; + + if (due_to_completion) { + grpc_timer_cancel(exec_ctx, &w->alarm); + } + + gpr_mu_lock(&w->mu); + + if (due_to_completion) { + if (grpc_trace_operation_failures) { + GRPC_LOG_IF_ERROR("watch_completion_error", GRPC_ERROR_REF(error)); + } + GRPC_ERROR_UNREF(error); + error = GRPC_ERROR_NONE; + } else { + if (error == GRPC_ERROR_NONE) { + error = + GRPC_ERROR_CREATE("Timed out waiting for connection state change"); + } else if (error == GRPC_ERROR_CANCELLED) { + error = GRPC_ERROR_NONE; + } + } + switch (w->phase) { + case WAITING: + w->phase = CALLING_BACK; + grpc_cq_end_op(exec_ctx, w->cq, w->tag, GRPC_ERROR_REF(error), + finished_completion, w, &w->completion_storage); + break; + case CALLING_BACK: + w->phase = CALLING_BACK_AND_FINISHED; + break; + case CALLING_BACK_AND_FINISHED: + GPR_UNREACHABLE_CODE(return ); + case CALLED_BACK: + delete = 1; + break; + } + gpr_mu_unlock(&w->mu); + + if (delete) { + delete_state_watcher(exec_ctx, w); + } + + GRPC_ERROR_UNREF(error); +} + +static void watch_complete(grpc_exec_ctx *exec_ctx, void *pw, + grpc_error *error) { + partly_done(exec_ctx, pw, true, GRPC_ERROR_REF(error)); +} + +static void timeout_complete(grpc_exec_ctx *exec_ctx, void *pw, + grpc_error *error) { + partly_done(exec_ctx, pw, false, GRPC_ERROR_REF(error)); +} + +void grpc_channel_watch_connectivity_state( + grpc_channel *channel, grpc_connectivity_state last_observed_state, + gpr_timespec deadline, grpc_completion_queue *cq, void *tag) { + grpc_channel_element *client_channel_elem = + grpc_channel_stack_last_element(grpc_channel_get_channel_stack(channel)); + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + state_watcher *w = gpr_malloc(sizeof(*w)); + + GRPC_API_TRACE( + "grpc_channel_watch_connectivity_state(" + "channel=%p, last_observed_state=%d, " + "deadline=gpr_timespec { tv_sec: %" PRId64 + ", tv_nsec: %d, clock_type: %d }, " + "cq=%p, tag=%p)", + 7, (channel, (int)last_observed_state, deadline.tv_sec, deadline.tv_nsec, + (int)deadline.clock_type, cq, tag)); + + grpc_cq_begin_op(cq, tag); + + gpr_mu_init(&w->mu); + grpc_closure_init(&w->on_complete, watch_complete, w); + w->phase = WAITING; + w->state = last_observed_state; + w->cq = cq; + w->tag = tag; + w->channel = channel; + + grpc_timer_init(&exec_ctx, &w->alarm, + gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), + timeout_complete, w, gpr_now(GPR_CLOCK_MONOTONIC)); + + if (client_channel_elem->filter == &grpc_client_channel_filter) { + GRPC_CHANNEL_INTERNAL_REF(channel, "watch_channel_connectivity"); + grpc_client_channel_watch_connectivity_state(&exec_ctx, client_channel_elem, + grpc_cq_pollset(cq), &w->state, + &w->on_complete); + } else { + abort(); + } + + grpc_exec_ctx_finish(&exec_ctx); +} diff --git a/src/core/ext/client_channel/client_channel.c b/src/core/ext/client_channel/client_channel.c new file mode 100644 index 0000000000..7e8579de60 --- /dev/null +++ b/src/core/ext/client_channel/client_channel.c @@ -0,0 +1,840 @@ +/* + * + * 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_channel/client_channel.h" + +#include <stdbool.h> +#include <stdio.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/sync.h> +#include <grpc/support/useful.h> + +#include "src/core/ext/client_channel/subchannel.h" +#include "src/core/lib/channel/channel_args.h" +#include "src/core/lib/channel/connected_channel.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" + +/* Client channel implementation */ + +/************************************************************************* + * CHANNEL-WIDE FUNCTIONS + */ + +typedef struct client_channel_channel_data { + /** resolver for this channel */ + grpc_resolver *resolver; + /** have we started resolving this channel */ + bool started_resolving; + + /** mutex protecting client configuration, including all + variables below in this data structure */ + gpr_mu mu; + /** currently active load balancer - guarded by mu */ + grpc_lb_policy *lb_policy; + /** incoming resolver result - set by resolver.next(), guarded by mu */ + 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; + /** resolver callback */ + grpc_closure on_resolver_result_changed; + /** connectivity state being tracked */ + grpc_connectivity_state_tracker state_tracker; + /** when an lb_policy arrives, should we try to exit idle */ + bool exit_idle_when_lb_policy_arrives; + /** owning stack */ + grpc_channel_stack *owning_stack; + /** interested parties (owned) */ + grpc_pollset_set *interested_parties; +} channel_data; + +/** We create one watcher for each new lb_policy that is returned from a + resolver, to watch for state changes from the lb_policy. When a state + change is seen, we update the channel, and create a new watcher. */ +typedef struct { + channel_data *chand; + grpc_closure on_changed; + grpc_connectivity_state state; + grpc_lb_policy *lb_policy; +} lb_policy_connectivity_watcher; + +static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand, + grpc_lb_policy *lb_policy, + grpc_connectivity_state current_state); + +static void set_channel_connectivity_state_locked(grpc_exec_ctx *exec_ctx, + channel_data *chand, + grpc_connectivity_state state, + grpc_error *error, + const char *reason) { + if ((state == GRPC_CHANNEL_TRANSIENT_FAILURE || + state == GRPC_CHANNEL_SHUTDOWN) && + chand->lb_policy != NULL) { + /* cancel fail-fast picks */ + grpc_lb_policy_cancel_picks( + exec_ctx, chand->lb_policy, + /* mask= */ GRPC_INITIAL_METADATA_IGNORE_CONNECTIVITY, + /* check= */ 0); + } + grpc_connectivity_state_set(exec_ctx, &chand->state_tracker, state, error, + reason); +} + +static void on_lb_policy_state_changed_locked(grpc_exec_ctx *exec_ctx, + lb_policy_connectivity_watcher *w, + grpc_error *error) { + grpc_connectivity_state publish_state = w->state; + /* check if the notification is for a stale policy */ + if (w->lb_policy != w->chand->lb_policy) return; + + if (publish_state == GRPC_CHANNEL_SHUTDOWN && w->chand->resolver != NULL) { + publish_state = GRPC_CHANNEL_TRANSIENT_FAILURE; + grpc_resolver_channel_saw_error(exec_ctx, w->chand->resolver); + GRPC_LB_POLICY_UNREF(exec_ctx, w->chand->lb_policy, "channel"); + w->chand->lb_policy = NULL; + } + set_channel_connectivity_state_locked(exec_ctx, w->chand, publish_state, + GRPC_ERROR_REF(error), "lb_changed"); + if (w->state != GRPC_CHANNEL_SHUTDOWN) { + watch_lb_policy(exec_ctx, w->chand, w->lb_policy, w->state); + } +} + +static void on_lb_policy_state_changed(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + lb_policy_connectivity_watcher *w = arg; + + gpr_mu_lock(&w->chand->mu); + on_lb_policy_state_changed_locked(exec_ctx, w, error); + gpr_mu_unlock(&w->chand->mu); + + GRPC_CHANNEL_STACK_UNREF(exec_ctx, w->chand->owning_stack, "watch_lb_policy"); + gpr_free(w); +} + +static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand, + grpc_lb_policy *lb_policy, + grpc_connectivity_state current_state) { + lb_policy_connectivity_watcher *w = gpr_malloc(sizeof(*w)); + GRPC_CHANNEL_STACK_REF(chand->owning_stack, "watch_lb_policy"); + + w->chand = chand; + grpc_closure_init(&w->on_changed, on_lb_policy_state_changed, w); + w->state = current_state; + w->lb_policy = lb_policy; + grpc_lb_policy_notify_on_state_change(exec_ctx, lb_policy, &w->state, + &w->on_changed); +} + +static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + channel_data *chand = arg; + grpc_lb_policy *lb_policy = NULL; + grpc_lb_policy *old_lb_policy; + grpc_connectivity_state state = GRPC_CHANNEL_TRANSIENT_FAILURE; + bool exit_idle = false; + grpc_error *state_error = GRPC_ERROR_CREATE("No load balancing policy"); + + if (chand->resolver_result != NULL) { + lb_policy = grpc_resolver_result_get_lb_policy(chand->resolver_result); + if (lb_policy != NULL) { + GRPC_LB_POLICY_REF(lb_policy, "channel"); + GRPC_LB_POLICY_REF(lb_policy, "config_change"); + GRPC_ERROR_UNREF(state_error); + state = + grpc_lb_policy_check_connectivity(exec_ctx, lb_policy, &state_error); + } + + grpc_resolver_result_unref(exec_ctx, chand->resolver_result); + } + + chand->resolver_result = NULL; + + if (lb_policy != NULL) { + grpc_pollset_set_add_pollset_set(exec_ctx, lb_policy->interested_parties, + chand->interested_parties); + } + + gpr_mu_lock(&chand->mu); + old_lb_policy = chand->lb_policy; + chand->lb_policy = lb_policy; + if (lb_policy != NULL) { + grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures, + NULL); + } else if (chand->resolver == NULL /* disconnected */) { + grpc_closure_list_fail_all( + &chand->waiting_for_config_closures, + GRPC_ERROR_CREATE_REFERENCING("Channel disconnected", &error, 1)); + grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures, + NULL); + } + if (lb_policy != NULL && chand->exit_idle_when_lb_policy_arrives) { + GRPC_LB_POLICY_REF(lb_policy, "exit_idle"); + exit_idle = true; + chand->exit_idle_when_lb_policy_arrives = false; + } + + if (error == GRPC_ERROR_NONE && chand->resolver) { + set_channel_connectivity_state_locked( + exec_ctx, chand, state, GRPC_ERROR_REF(state_error), "new_lb+resolver"); + if (lb_policy != NULL) { + watch_lb_policy(exec_ctx, chand, lb_policy, state); + } + GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver"); + grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result, + &chand->on_resolver_result_changed); + gpr_mu_unlock(&chand->mu); + } else { + if (chand->resolver != NULL) { + grpc_resolver_shutdown(exec_ctx, chand->resolver); + GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel"); + chand->resolver = NULL; + } + grpc_error *refs[] = {error, state_error}; + set_channel_connectivity_state_locked( + exec_ctx, chand, GRPC_CHANNEL_SHUTDOWN, + GRPC_ERROR_CREATE_REFERENCING("Got config after disconnection", refs, + GPR_ARRAY_SIZE(refs)), + "resolver_gone"); + gpr_mu_unlock(&chand->mu); + } + + if (exit_idle) { + grpc_lb_policy_exit_idle(exec_ctx, lb_policy); + GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "exit_idle"); + } + + if (old_lb_policy != NULL) { + grpc_pollset_set_del_pollset_set( + exec_ctx, old_lb_policy->interested_parties, chand->interested_parties); + GRPC_LB_POLICY_UNREF(exec_ctx, old_lb_policy, "channel"); + } + + if (lb_policy != NULL) { + GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "config_change"); + } + + GRPC_CHANNEL_STACK_UNREF(exec_ctx, chand->owning_stack, "resolver"); + GRPC_ERROR_UNREF(state_error); +} + +static void cc_start_transport_op(grpc_exec_ctx *exec_ctx, + grpc_channel_element *elem, + grpc_transport_op *op) { + channel_data *chand = elem->channel_data; + + grpc_exec_ctx_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE, NULL); + + GPR_ASSERT(op->set_accept_stream == false); + if (op->bind_pollset != NULL) { + grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties, + op->bind_pollset); + } + + gpr_mu_lock(&chand->mu); + if (op->on_connectivity_state_change != NULL) { + grpc_connectivity_state_notify_on_state_change( + exec_ctx, &chand->state_tracker, op->connectivity_state, + op->on_connectivity_state_change); + op->on_connectivity_state_change = NULL; + op->connectivity_state = NULL; + } + + if (op->send_ping != NULL) { + if (chand->lb_policy == NULL) { + grpc_exec_ctx_sched(exec_ctx, op->send_ping, + GRPC_ERROR_CREATE("Ping with no load balancing"), + NULL); + } else { + grpc_lb_policy_ping_one(exec_ctx, chand->lb_policy, op->send_ping); + op->bind_pollset = NULL; + } + op->send_ping = NULL; + } + + if (op->disconnect_with_error != GRPC_ERROR_NONE) { + if (chand->resolver != NULL) { + set_channel_connectivity_state_locked( + exec_ctx, chand, GRPC_CHANNEL_SHUTDOWN, + GRPC_ERROR_REF(op->disconnect_with_error), "disconnect"); + grpc_resolver_shutdown(exec_ctx, chand->resolver); + GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel"); + chand->resolver = NULL; + if (!chand->started_resolving) { + grpc_closure_list_fail_all(&chand->waiting_for_config_closures, + GRPC_ERROR_REF(op->disconnect_with_error)); + grpc_exec_ctx_enqueue_list(exec_ctx, + &chand->waiting_for_config_closures, NULL); + } + if (chand->lb_policy != NULL) { + grpc_pollset_set_del_pollset_set(exec_ctx, + chand->lb_policy->interested_parties, + chand->interested_parties); + GRPC_LB_POLICY_UNREF(exec_ctx, chand->lb_policy, "channel"); + chand->lb_policy = NULL; + } + } + GRPC_ERROR_UNREF(op->disconnect_with_error); + } + gpr_mu_unlock(&chand->mu); +} + +/* Constructor for channel_data */ +static void cc_init_channel_elem(grpc_exec_ctx *exec_ctx, + grpc_channel_element *elem, + grpc_channel_element_args *args) { + channel_data *chand = elem->channel_data; + + memset(chand, 0, sizeof(*chand)); + + GPR_ASSERT(args->is_last); + GPR_ASSERT(elem->filter == &grpc_client_channel_filter); + + gpr_mu_init(&chand->mu); + grpc_closure_init(&chand->on_resolver_result_changed, + on_resolver_result_changed, chand); + chand->owning_stack = args->channel_stack; + + grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE, + "client_channel"); + chand->interested_parties = grpc_pollset_set_create(); +} + +/* Destructor for channel_data */ +static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx, + grpc_channel_element *elem) { + channel_data *chand = elem->channel_data; + + if (chand->resolver != NULL) { + grpc_resolver_shutdown(exec_ctx, chand->resolver); + GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel"); + } + if (chand->lb_policy != NULL) { + grpc_pollset_set_del_pollset_set(exec_ctx, + chand->lb_policy->interested_parties, + chand->interested_parties); + GRPC_LB_POLICY_UNREF(exec_ctx, chand->lb_policy, "channel"); + } + grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker); + grpc_pollset_set_destroy(chand->interested_parties); + gpr_mu_destroy(&chand->mu); +} + +/************************************************************************* + * PER-CALL FUNCTIONS + */ + +#define GET_CALL(call_data) \ + ((grpc_subchannel_call *)(gpr_atm_acq_load(&(call_data)->subchannel_call))) + +#define CANCELLED_CALL ((grpc_subchannel_call *)1) + +typedef enum { + GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING, + GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL +} subchannel_creation_phase; + +/** Call data. Holds a pointer to grpc_subchannel_call and the + associated machinery to create such a pointer. + Handles queueing of stream ops until a call object is ready, waiting + for initial metadata before trying to create a call object, + and handling cancellation gracefully. */ +typedef struct client_channel_call_data { + /** either 0 for no call, 1 for cancelled, or a pointer to a + grpc_subchannel_call */ + gpr_atm subchannel_call; + + gpr_mu mu; + + subchannel_creation_phase creation_phase; + grpc_connected_subchannel *connected_subchannel; + grpc_polling_entity *pollent; + + grpc_transport_stream_op *waiting_ops; + size_t waiting_ops_count; + size_t waiting_ops_capacity; + + grpc_closure next_step; + + grpc_call_stack *owning_call; +} call_data; + +static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) { + GPR_TIMER_BEGIN("add_waiting_locked", 0); + if (calld->waiting_ops_count == calld->waiting_ops_capacity) { + calld->waiting_ops_capacity = GPR_MAX(3, 2 * calld->waiting_ops_capacity); + calld->waiting_ops = + gpr_realloc(calld->waiting_ops, + calld->waiting_ops_capacity * sizeof(*calld->waiting_ops)); + } + calld->waiting_ops[calld->waiting_ops_count++] = *op; + GPR_TIMER_END("add_waiting_locked", 0); +} + +static void fail_locked(grpc_exec_ctx *exec_ctx, call_data *calld, + grpc_error *error) { + size_t i; + for (i = 0; i < calld->waiting_ops_count; i++) { + grpc_transport_stream_op_finish_with_failure( + exec_ctx, &calld->waiting_ops[i], GRPC_ERROR_REF(error)); + } + calld->waiting_ops_count = 0; + GRPC_ERROR_UNREF(error); +} + +typedef struct { + grpc_transport_stream_op *ops; + size_t nops; + grpc_subchannel_call *call; +} retry_ops_args; + +static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) { + retry_ops_args *a = args; + size_t i; + for (i = 0; i < a->nops; i++) { + grpc_subchannel_call_process_op(exec_ctx, a->call, &a->ops[i]); + } + GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops"); + gpr_free(a->ops); + gpr_free(a); +} + +static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) { + retry_ops_args *a = gpr_malloc(sizeof(*a)); + a->ops = calld->waiting_ops; + a->nops = calld->waiting_ops_count; + a->call = GET_CALL(calld); + if (a->call == CANCELLED_CALL) { + gpr_free(a); + fail_locked(exec_ctx, calld, GRPC_ERROR_CANCELLED); + return; + } + calld->waiting_ops = NULL; + calld->waiting_ops_count = 0; + calld->waiting_ops_capacity = 0; + GRPC_SUBCHANNEL_CALL_REF(a->call, "retry_ops"); + grpc_exec_ctx_sched(exec_ctx, grpc_closure_create(retry_ops, a), + GRPC_ERROR_NONE, NULL); +} + +static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + call_data *calld = arg; + gpr_mu_lock(&calld->mu); + GPR_ASSERT(calld->creation_phase == + GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL); + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; + if (calld->connected_subchannel == NULL) { + 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)) { + /* already cancelled before subchannel became ready */ + fail_locked(exec_ctx, calld, + GRPC_ERROR_CREATE_REFERENCING( + "Cancelled before creating subchannel", &error, 1)); + } else { + grpc_subchannel_call *subchannel_call = NULL; + grpc_error *new_error = grpc_connected_subchannel_create_call( + exec_ctx, calld->connected_subchannel, calld->pollent, + &subchannel_call); + if (new_error != GRPC_ERROR_NONE) { + new_error = grpc_error_add_child(new_error, error); + subchannel_call = CANCELLED_CALL; + fail_locked(exec_ctx, calld, new_error); + } + gpr_atm_rel_store(&calld->subchannel_call, + (gpr_atm)(uintptr_t)subchannel_call); + retry_waiting_locked(exec_ctx, calld); + } + gpr_mu_unlock(&calld->mu); + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); +} + +static char *cc_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { + call_data *calld = elem->call_data; + grpc_subchannel_call *subchannel_call = GET_CALL(calld); + if (subchannel_call == NULL || subchannel_call == CANCELLED_CALL) { + return NULL; + } else { + return grpc_subchannel_call_get_peer(exec_ctx, subchannel_call); + } +} + +typedef struct { + grpc_metadata_batch *initial_metadata; + uint32_t initial_metadata_flags; + grpc_connected_subchannel **connected_subchannel; + grpc_closure *on_ready; + grpc_call_element *elem; + grpc_closure closure; +} continue_picking_args; + +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); + +static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + continue_picking_args *cpa = arg; + if (cpa->connected_subchannel == NULL) { + /* 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_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE, NULL); + } + gpr_free(cpa); +} + +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) { + GPR_TIMER_BEGIN("pick_subchannel", 0); + + channel_data *chand = elem->channel_data; + call_data *calld = elem->call_data; + continue_picking_args *cpa; + grpc_closure *closure; + + GPR_ASSERT(connected_subchannel); + + gpr_mu_lock(&chand->mu); + if (initial_metadata == NULL) { + if (chand->lb_policy != NULL) { + grpc_lb_policy_cancel_pick(exec_ctx, chand->lb_policy, + connected_subchannel); + } + 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); + } + } + gpr_mu_unlock(&chand->mu); + GPR_TIMER_END("pick_subchannel", 0); + return true; + } + 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); + r = grpc_lb_policy_pick(exec_ctx, lb_policy, calld->pollent, + initial_metadata, initial_metadata_flags, + connected_subchannel, on_ready); + GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "pick_subchannel"); + GPR_TIMER_END("pick_subchannel", 0); + return r; + } + if (chand->resolver != NULL && !chand->started_resolving) { + chand->started_resolving = true; + GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver"); + grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result, + &chand->on_resolver_result_changed); + } + if (chand->resolver != NULL) { + cpa = gpr_malloc(sizeof(*cpa)); + cpa->initial_metadata = initial_metadata; + cpa->initial_metadata_flags = initial_metadata_flags; + cpa->connected_subchannel = connected_subchannel; + cpa->on_ready = on_ready; + cpa->elem = elem; + grpc_closure_init(&cpa->closure, continue_picking, cpa); + grpc_closure_list_append(&chand->waiting_for_config_closures, &cpa->closure, + GRPC_ERROR_NONE); + } else { + grpc_exec_ctx_sched(exec_ctx, on_ready, GRPC_ERROR_CREATE("Disconnected"), + NULL); + } + gpr_mu_unlock(&chand->mu); + + GPR_TIMER_END("pick_subchannel", 0); + return false; +} + +// The logic here is fairly complicated, due to (a) the fact that we +// need to handle the case where we receive the send op before the +// initial metadata op, and (b) the need for efficiency, especially in +// the streaming case. +// TODO(ctiller): Explain this more thoroughly. +static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_transport_stream_op *op) { + call_data *calld = elem->call_data; + GRPC_CALL_LOG_OP(GPR_INFO, 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); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + if (call != NULL) { + grpc_subchannel_call_process_op(exec_ctx, call, op); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + /* we failed; lock and figure out what to do */ + gpr_mu_lock(&calld->mu); +retry: + /* need to recheck that another thread hasn't set the call */ + 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); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + if (call != NULL) { + gpr_mu_unlock(&calld->mu); + grpc_subchannel_call_process_op(exec_ctx, call, op); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + /* if this is a cancellation, then we can raise our cancelled flag */ + if (op->cancel_error != GRPC_ERROR_NONE) { + if (!gpr_atm_rel_cas(&calld->subchannel_call, 0, + (gpr_atm)(uintptr_t)CANCELLED_CALL)) { + goto retry; + } else { + 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); + break; + } + gpr_mu_unlock(&calld->mu); + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, + GRPC_ERROR_CANCELLED); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + } + /* if we don't have a subchannel, try to get one */ + if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && + calld->connected_subchannel == NULL && + op->send_initial_metadata != NULL) { + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL; + grpc_closure_init(&calld->next_step, subchannel_ready, calld); + 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->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); + } + } + /* if we've got a subchannel, then let's ask it to create a call */ + if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && + 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); + if (error != GRPC_ERROR_NONE) { + subchannel_call = CANCELLED_CALL; + fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error)); + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error); + } + gpr_atm_rel_store(&calld->subchannel_call, + (gpr_atm)(uintptr_t)subchannel_call); + retry_waiting_locked(exec_ctx, calld); + goto retry; + } + /* nothing to be done but wait */ + add_waiting_locked(calld, op); + gpr_mu_unlock(&calld->mu); + GPR_TIMER_END("cc_start_transport_stream_op", 0); +} + +/* 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) { + call_data *calld = elem->call_data; + gpr_atm_rel_store(&calld->subchannel_call, 0); + gpr_mu_init(&calld->mu); + calld->connected_subchannel = NULL; + calld->waiting_ops = NULL; + calld->waiting_ops_count = 0; + calld->waiting_ops_capacity = 0; + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; + calld->owning_call = args->call_stack; + calld->pollent = NULL; + return GRPC_ERROR_NONE; +} + +/* Destructor for call_data */ +static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + const grpc_call_final_info *final_info, + void *and_free_memory) { + call_data *calld = elem->call_data; + grpc_subchannel_call *call = GET_CALL(calld); + if (call != NULL && call != CANCELLED_CALL) { + GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call"); + } + GPR_ASSERT(calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING); + gpr_mu_destroy(&calld->mu); + GPR_ASSERT(calld->waiting_ops_count == 0); + gpr_free(calld->waiting_ops); + gpr_free(and_free_memory); +} + +static void cc_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_polling_entity *pollent) { + call_data *calld = elem->call_data; + calld->pollent = pollent; +} + +/************************************************************************* + * EXPORTED SYMBOLS + */ + +const grpc_channel_filter grpc_client_channel_filter = { + cc_start_transport_stream_op, + cc_start_transport_op, + sizeof(call_data), + cc_init_call_elem, + cc_set_pollset_or_pollset_set, + cc_destroy_call_elem, + sizeof(channel_data), + cc_init_channel_elem, + cc_destroy_channel_elem, + cc_get_peer, + "client-channel", +}; + +void grpc_client_channel_set_resolver(grpc_exec_ctx *exec_ctx, + grpc_channel_stack *channel_stack, + grpc_resolver *resolver) { + /* post construction initialization: set the transport setup pointer */ + grpc_channel_element *elem = grpc_channel_stack_last_element(channel_stack); + channel_data *chand = elem->channel_data; + gpr_mu_lock(&chand->mu); + GPR_ASSERT(!chand->resolver); + chand->resolver = resolver; + GRPC_RESOLVER_REF(resolver, "channel"); + if (!grpc_closure_list_empty(chand->waiting_for_config_closures) || + chand->exit_idle_when_lb_policy_arrives) { + chand->started_resolving = true; + GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver"); + grpc_resolver_next(exec_ctx, resolver, &chand->resolver_result, + &chand->on_resolver_result_changed); + } + gpr_mu_unlock(&chand->mu); +} + +grpc_connectivity_state grpc_client_channel_check_connectivity_state( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, int try_to_connect) { + channel_data *chand = elem->channel_data; + grpc_connectivity_state out; + gpr_mu_lock(&chand->mu); + out = grpc_connectivity_state_check(&chand->state_tracker, NULL); + if (out == GRPC_CHANNEL_IDLE && try_to_connect) { + if (chand->lb_policy != NULL) { + grpc_lb_policy_exit_idle(exec_ctx, chand->lb_policy); + } else { + chand->exit_idle_when_lb_policy_arrives = true; + if (!chand->started_resolving && chand->resolver != NULL) { + GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver"); + chand->started_resolving = true; + grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result, + &chand->on_resolver_result_changed); + } + } + } + gpr_mu_unlock(&chand->mu); + return out; +} + +typedef struct { + channel_data *chand; + grpc_pollset *pollset; + grpc_closure *on_complete; + grpc_closure my_closure; +} external_connectivity_watcher; + +static void on_external_watch_complete(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + external_connectivity_watcher *w = arg; + grpc_closure *follow_up = w->on_complete; + grpc_pollset_set_del_pollset(exec_ctx, w->chand->interested_parties, + w->pollset); + GRPC_CHANNEL_STACK_UNREF(exec_ctx, w->chand->owning_stack, + "external_connectivity_watcher"); + gpr_free(w); + follow_up->cb(exec_ctx, follow_up->cb_arg, error); +} + +void grpc_client_channel_watch_connectivity_state( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_pollset *pollset, + grpc_connectivity_state *state, grpc_closure *on_complete) { + channel_data *chand = elem->channel_data; + external_connectivity_watcher *w = gpr_malloc(sizeof(*w)); + w->chand = chand; + w->pollset = pollset; + w->on_complete = on_complete; + grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties, pollset); + grpc_closure_init(&w->my_closure, on_external_watch_complete, w); + GRPC_CHANNEL_STACK_REF(w->chand->owning_stack, + "external_connectivity_watcher"); + gpr_mu_lock(&chand->mu); + grpc_connectivity_state_notify_on_state_change( + exec_ctx, &chand->state_tracker, state, &w->my_closure); + gpr_mu_unlock(&chand->mu); +} diff --git a/src/core/ext/client_channel/client_channel.h b/src/core/ext/client_channel/client_channel.h new file mode 100644 index 0000000000..5dea45de43 --- /dev/null +++ b/src/core/ext/client_channel/client_channel.h @@ -0,0 +1,63 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_CLIENT_CHANNEL_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_CLIENT_CHANNEL_H + +#include "src/core/ext/client_channel/resolver.h" +#include "src/core/lib/channel/channel_stack.h" + +/* A client channel is a channel that begins disconnected, and can connect + to some endpoint on demand. If that endpoint disconnects, it will be + connected to again later. + + Calls on a disconnected client channel are queued until a connection is + established. */ + +extern const grpc_channel_filter grpc_client_channel_filter; + +/* post-construction initializer to let the client channel know which + transport setup it should cancel upon destruction, or initiate when it needs + a connection */ +void grpc_client_channel_set_resolver(grpc_exec_ctx *exec_ctx, + grpc_channel_stack *channel_stack, + grpc_resolver *resolver); + +grpc_connectivity_state grpc_client_channel_check_connectivity_state( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, int try_to_connect); + +void grpc_client_channel_watch_connectivity_state( + grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, grpc_pollset *pollset, + grpc_connectivity_state *state, grpc_closure *on_complete); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_CLIENT_CHANNEL_H */ diff --git a/src/core/ext/client_channel/client_channel_factory.c b/src/core/ext/client_channel/client_channel_factory.c new file mode 100644 index 0000000000..db1cc9093c --- /dev/null +++ b/src/core/ext/client_channel/client_channel_factory.c @@ -0,0 +1,57 @@ +/* + * + * 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_channel/client_channel_factory.h" + +void grpc_client_channel_factory_ref(grpc_client_channel_factory* factory) { + factory->vtable->ref(factory); +} + +void grpc_client_channel_factory_unref(grpc_exec_ctx* exec_ctx, + grpc_client_channel_factory* factory) { + factory->vtable->unref(exec_ctx, factory); +} + +grpc_subchannel* grpc_client_channel_factory_create_subchannel( + grpc_exec_ctx* exec_ctx, grpc_client_channel_factory* factory, + grpc_subchannel_args* args) { + return factory->vtable->create_subchannel(exec_ctx, factory, args); +} + +grpc_channel* grpc_client_channel_factory_create_channel( + grpc_exec_ctx* exec_ctx, grpc_client_channel_factory* factory, + const char* target, grpc_client_channel_type type, + grpc_channel_args* args) { + return factory->vtable->create_client_channel(exec_ctx, factory, target, type, + args); +} diff --git a/src/core/ext/client_channel/client_channel_factory.h b/src/core/ext/client_channel/client_channel_factory.h new file mode 100644 index 0000000000..cf6724a24b --- /dev/null +++ b/src/core/ext/client_channel/client_channel_factory.h @@ -0,0 +1,85 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_CLIENT_CHANNEL_FACTORY_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_CLIENT_CHANNEL_FACTORY_H + +#include <grpc/impl/codegen/grpc_types.h> + +#include "src/core/ext/client_channel/subchannel.h" +#include "src/core/lib/channel/channel_stack.h" + +typedef struct grpc_client_channel_factory grpc_client_channel_factory; +typedef struct grpc_client_channel_factory_vtable + grpc_client_channel_factory_vtable; + +typedef enum { + GRPC_CLIENT_CHANNEL_TYPE_REGULAR, /** for the user-level regular calls */ + GRPC_CLIENT_CHANNEL_TYPE_LOAD_BALANCING, /** for communication with a load + balancing service */ +} grpc_client_channel_type; + +/** Constructor for new configured channels. + Creating decorators around this type is encouraged to adapt behavior. */ +struct grpc_client_channel_factory { + const grpc_client_channel_factory_vtable *vtable; +}; + +struct grpc_client_channel_factory_vtable { + void (*ref)(grpc_client_channel_factory *factory); + void (*unref)(grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *factory); + grpc_subchannel *(*create_subchannel)(grpc_exec_ctx *exec_ctx, + grpc_client_channel_factory *factory, + grpc_subchannel_args *args); + grpc_channel *(*create_client_channel)(grpc_exec_ctx *exec_ctx, + grpc_client_channel_factory *factory, + const char *target, + grpc_client_channel_type type, + grpc_channel_args *args); +}; + +void grpc_client_channel_factory_ref(grpc_client_channel_factory *factory); +void grpc_client_channel_factory_unref(grpc_exec_ctx *exec_ctx, + grpc_client_channel_factory *factory); + +/** Create a new grpc_subchannel */ +grpc_subchannel *grpc_client_channel_factory_create_subchannel( + grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *factory, + grpc_subchannel_args *args); + +/** Create a new grpc_channel */ +grpc_channel *grpc_client_channel_factory_create_channel( + grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *factory, + const char *target, grpc_client_channel_type type, grpc_channel_args *args); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_CLIENT_CHANNEL_FACTORY_H */ diff --git a/src/core/ext/client_channel/client_channel_plugin.c b/src/core/ext/client_channel/client_channel_plugin.c new file mode 100644 index 0000000000..954f0db3bb --- /dev/null +++ b/src/core/ext/client_channel/client_channel_plugin.c @@ -0,0 +1,95 @@ +/* + * + * 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 <limits.h> +#include <stdbool.h> +#include <string.h> + +#include <grpc/support/alloc.h> + +#include "src/core/ext/client_channel/client_channel.h" +#include "src/core/ext/client_channel/lb_policy_registry.h" +#include "src/core/ext/client_channel/resolver_registry.h" +#include "src/core/ext/client_channel/subchannel_index.h" +#include "src/core/lib/surface/channel_init.h" + +#ifndef GRPC_DEFAULT_NAME_PREFIX +#define GRPC_DEFAULT_NAME_PREFIX "dns:///" +#endif + +static bool append_filter(grpc_channel_stack_builder *builder, void *arg) { + return grpc_channel_stack_builder_append_filter( + builder, (const grpc_channel_filter *)arg, NULL, NULL); +} + +static bool set_default_host_if_unset(grpc_channel_stack_builder *builder, + void *unused) { + const grpc_channel_args *args = + grpc_channel_stack_builder_get_channel_arguments(builder); + for (size_t i = 0; i < args->num_args; i++) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY) || + 0 == strcmp(args->args[i].key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)) { + return true; + } + } + char *default_authority = grpc_get_default_authority( + grpc_channel_stack_builder_get_target(builder)); + if (default_authority != NULL) { + grpc_arg arg; + arg.type = GRPC_ARG_STRING; + arg.key = GRPC_ARG_DEFAULT_AUTHORITY; + arg.value.string = default_authority; + grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1); + grpc_channel_stack_builder_set_channel_arguments(builder, new_args); + gpr_free(default_authority); + grpc_channel_args_destroy(new_args); + } + return true; +} + +void grpc_client_channel_init(void) { + grpc_lb_policy_registry_init(); + grpc_resolver_registry_init(GRPC_DEFAULT_NAME_PREFIX); + grpc_subchannel_index_init(); + grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MIN, + set_default_host_if_unset, NULL); + grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MAX, append_filter, + (void *)&grpc_client_channel_filter); +} + +void grpc_client_channel_shutdown(void) { + grpc_subchannel_index_shutdown(); + grpc_channel_init_shutdown(); + grpc_resolver_registry_shutdown(); + grpc_lb_policy_registry_shutdown(); +} diff --git a/src/core/ext/client_channel/connector.c b/src/core/ext/client_channel/connector.c new file mode 100644 index 0000000000..0582e5b372 --- /dev/null +++ b/src/core/ext/client_channel/connector.c @@ -0,0 +1,55 @@ +/* + * + * 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_channel/connector.h" + +grpc_connector* grpc_connector_ref(grpc_connector* connector) { + connector->vtable->ref(connector); + return connector; +} + +void grpc_connector_unref(grpc_exec_ctx* exec_ctx, grpc_connector* connector) { + connector->vtable->unref(exec_ctx, connector); +} + +void grpc_connector_connect(grpc_exec_ctx* exec_ctx, grpc_connector* connector, + const grpc_connect_in_args* in_args, + grpc_connect_out_args* out_args, + grpc_closure* notify) { + connector->vtable->connect(exec_ctx, connector, in_args, out_args, notify); +} + +void grpc_connector_shutdown(grpc_exec_ctx* exec_ctx, + grpc_connector* connector) { + connector->vtable->shutdown(exec_ctx, connector); +} diff --git a/src/core/ext/client_channel/connector.h b/src/core/ext/client_channel/connector.h new file mode 100644 index 0000000000..ea9d23706e --- /dev/null +++ b/src/core/ext/client_channel/connector.h @@ -0,0 +1,92 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_CONNECTOR_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_CONNECTOR_H + +#include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/iomgr/sockaddr.h" +#include "src/core/lib/transport/transport.h" + +typedef struct grpc_connector grpc_connector; +typedef struct grpc_connector_vtable grpc_connector_vtable; + +struct grpc_connector { + const grpc_connector_vtable *vtable; +}; + +typedef struct { + /** set of pollsets interested in this connection */ + grpc_pollset_set *interested_parties; + /** address to connect to */ + const struct sockaddr *addr; + size_t addr_len; + /** initial connect string to send */ + gpr_slice initial_connect_string; + /** deadline for connection */ + gpr_timespec deadline; + /** channel arguments (to be passed to transport) */ + const grpc_channel_args *channel_args; +} grpc_connect_in_args; + +typedef struct { + /** the connected transport */ + grpc_transport *transport; + + /** channel arguments (to be passed to the filters) */ + grpc_channel_args *channel_args; +} grpc_connect_out_args; + +struct grpc_connector_vtable { + void (*ref)(grpc_connector *connector); + void (*unref)(grpc_exec_ctx *exec_ctx, grpc_connector *connector); + /** Implementation of grpc_connector_shutdown */ + void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_connector *connector); + /** Implementation of grpc_connector_connect */ + void (*connect)(grpc_exec_ctx *exec_ctx, grpc_connector *connector, + const grpc_connect_in_args *in_args, + grpc_connect_out_args *out_args, grpc_closure *notify); +}; + +grpc_connector *grpc_connector_ref(grpc_connector *connector); +void grpc_connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *connector); +/** Connect using the connector: max one outstanding call at a time */ +void grpc_connector_connect(grpc_exec_ctx *exec_ctx, grpc_connector *connector, + const grpc_connect_in_args *in_args, + grpc_connect_out_args *out_args, + grpc_closure *notify); +/** Cancel any pending connection */ +void grpc_connector_shutdown(grpc_exec_ctx *exec_ctx, + grpc_connector *connector); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_CONNECTOR_H */ diff --git a/src/core/ext/client_channel/default_initial_connect_string.c b/src/core/ext/client_channel/default_initial_connect_string.c new file mode 100644 index 0000000000..a70da4a84a --- /dev/null +++ b/src/core/ext/client_channel/default_initial_connect_string.c @@ -0,0 +1,39 @@ +/* + * + * 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 <grpc/support/slice.h> +#include "src/core/lib/iomgr/sockaddr.h" + +void grpc_set_default_initial_connect_string(struct sockaddr **addr, + size_t *addr_len, + gpr_slice *initial_str) {} diff --git a/src/core/ext/client_channel/initial_connect_string.c b/src/core/ext/client_channel/initial_connect_string.c new file mode 100644 index 0000000000..fd8ddb83a8 --- /dev/null +++ b/src/core/ext/client_channel/initial_connect_string.c @@ -0,0 +1,53 @@ +/* + * + * 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_channel/initial_connect_string.h" + +#include <stddef.h> + +extern void grpc_set_default_initial_connect_string(struct sockaddr **addr, + size_t *addr_len, + gpr_slice *initial_str); + +static grpc_set_initial_connect_string_func g_set_initial_connect_string_func = + grpc_set_default_initial_connect_string; + +void grpc_test_set_initial_connect_string_function( + grpc_set_initial_connect_string_func func) { + g_set_initial_connect_string_func = func; +} + +void grpc_set_initial_connect_string(struct sockaddr **addr, size_t *addr_len, + gpr_slice *initial_str) { + g_set_initial_connect_string_func(addr, addr_len, initial_str); +} diff --git a/src/core/ext/client_channel/initial_connect_string.h b/src/core/ext/client_channel/initial_connect_string.h new file mode 100644 index 0000000000..06f0767832 --- /dev/null +++ b/src/core/ext/client_channel/initial_connect_string.h @@ -0,0 +1,50 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_INITIAL_CONNECT_STRING_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_INITIAL_CONNECT_STRING_H + +#include <grpc/support/slice.h> +#include "src/core/lib/iomgr/sockaddr.h" + +typedef void (*grpc_set_initial_connect_string_func)(struct sockaddr **addr, + size_t *addr_len, + gpr_slice *initial_str); +void grpc_test_set_initial_connect_string_function( + grpc_set_initial_connect_string_func func); + +/** Set a string to be sent once connected. Optionally reset addr. */ +void grpc_set_initial_connect_string(struct sockaddr **addr, size_t *addr_len, + gpr_slice *connect_string); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_INITIAL_CONNECT_STRING_H */ diff --git a/src/core/ext/client_channel/lb_policy.c b/src/core/ext/client_channel/lb_policy.c new file mode 100644 index 0000000000..9ca895ec6c --- /dev/null +++ b/src/core/ext/client_channel/lb_policy.c @@ -0,0 +1,146 @@ +/* + * + * 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_channel/lb_policy.h" + +#define WEAK_REF_BITS 16 + +void grpc_lb_policy_init(grpc_lb_policy *policy, + const grpc_lb_policy_vtable *vtable) { + policy->vtable = vtable; + gpr_atm_no_barrier_store(&policy->ref_pair, 1 << WEAK_REF_BITS); + policy->interested_parties = grpc_pollset_set_create(); +} + +#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG +#define REF_FUNC_EXTRA_ARGS , const char *file, int line, const char *reason +#define REF_MUTATE_EXTRA_ARGS REF_FUNC_EXTRA_ARGS, const char *purpose +#define REF_FUNC_PASS_ARGS(new_reason) , file, line, new_reason +#define REF_MUTATE_PASS_ARGS(purpose) , file, line, reason, purpose +#else +#define REF_FUNC_EXTRA_ARGS +#define REF_MUTATE_EXTRA_ARGS +#define REF_FUNC_PASS_ARGS(new_reason) +#define REF_MUTATE_PASS_ARGS(x) +#endif + +static gpr_atm ref_mutate(grpc_lb_policy *c, gpr_atm delta, + int barrier REF_MUTATE_EXTRA_ARGS) { + gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, delta) + : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta); +#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, + "LB_POLICY: 0x%" PRIxPTR " %12s 0x%" PRIxPTR " -> 0x%" PRIxPTR + " [%s]", + (intptr_t)c, purpose, old_val, old_val + delta, reason); +#endif + return old_val; +} + +void grpc_lb_policy_ref(grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) { + ref_mutate(policy, 1 << WEAK_REF_BITS, 0 REF_MUTATE_PASS_ARGS("STRONG_REF")); +} + +void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, + grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) { + gpr_atm old_val = + ref_mutate(policy, (gpr_atm)1 - (gpr_atm)(1 << WEAK_REF_BITS), + 1 REF_MUTATE_PASS_ARGS("STRONG_UNREF")); + gpr_atm mask = ~(gpr_atm)((1 << WEAK_REF_BITS) - 1); + gpr_atm check = 1 << WEAK_REF_BITS; + if ((old_val & mask) == check) { + policy->vtable->shutdown(exec_ctx, policy); + } + grpc_lb_policy_weak_unref(exec_ctx, + policy REF_FUNC_PASS_ARGS("strong-unref")); +} + +void grpc_lb_policy_weak_ref(grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) { + ref_mutate(policy, 1, 0 REF_MUTATE_PASS_ARGS("WEAK_REF")); +} + +void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx, + grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) { + gpr_atm old_val = + ref_mutate(policy, -(gpr_atm)1, 1 REF_MUTATE_PASS_ARGS("WEAK_UNREF")); + if (old_val == 1) { + grpc_pollset_set_destroy(policy->interested_parties); + policy->vtable->destroy(exec_ctx, policy); + } +} + +int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_polling_entity *pollent, + grpc_metadata_batch *initial_metadata, + uint32_t initial_metadata_flags, + grpc_connected_subchannel **target, + grpc_closure *on_complete) { + return policy->vtable->pick(exec_ctx, policy, pollent, initial_metadata, + initial_metadata_flags, target, on_complete); +} + +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); +} + +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) { + policy->vtable->cancel_picks(exec_ctx, policy, initial_metadata_flags_mask, + initial_metadata_flags_eq); +} + +void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) { + policy->vtable->exit_idle(exec_ctx, policy); +} + +void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_closure *closure) { + policy->vtable->ping_one(exec_ctx, policy, closure); +} + +void grpc_lb_policy_notify_on_state_change(grpc_exec_ctx *exec_ctx, + grpc_lb_policy *policy, + grpc_connectivity_state *state, + grpc_closure *closure) { + policy->vtable->notify_on_state_change(exec_ctx, policy, state, closure); +} + +grpc_connectivity_state grpc_lb_policy_check_connectivity( + grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_error **connectivity_error) { + return policy->vtable->check_connectivity(exec_ctx, policy, + connectivity_error); +} diff --git a/src/core/ext/client_channel/lb_policy.h b/src/core/ext/client_channel/lb_policy.h new file mode 100644 index 0000000000..a5838cdb9f --- /dev/null +++ b/src/core/ext/client_channel/lb_policy.h @@ -0,0 +1,166 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_H + +#include "src/core/ext/client_channel/subchannel.h" +#include "src/core/lib/iomgr/polling_entity.h" +#include "src/core/lib/transport/connectivity_state.h" + +/** A load balancing policy: specified by a vtable and a struct (which + is expected to be extended to contain some parameters) */ +typedef struct grpc_lb_policy grpc_lb_policy; +typedef struct grpc_lb_policy_vtable grpc_lb_policy_vtable; + +typedef void (*grpc_lb_completion)(void *cb_arg, grpc_subchannel *subchannel, + grpc_status_code status, const char *errmsg); + +struct grpc_lb_policy { + const grpc_lb_policy_vtable *vtable; + gpr_atm ref_pair; + /* owned pointer to interested parties in load balancing decisions */ + grpc_pollset_set *interested_parties; +}; + +struct grpc_lb_policy_vtable { + void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy); + + void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy); + + /** implement grpc_lb_policy_pick */ + int (*pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_polling_entity *pollent, + grpc_metadata_batch *initial_metadata, + uint32_t initial_metadata_flags, + grpc_connected_subchannel **target, grpc_closure *on_complete); + void (*cancel_pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_connected_subchannel **target); + void (*cancel_picks)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + uint32_t initial_metadata_flags_mask, + uint32_t initial_metadata_flags_eq); + + void (*ping_one)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_closure *closure); + + /** Try to enter a READY connectivity state */ + void (*exit_idle)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy); + + /** check the current connectivity of the lb_policy */ + grpc_connectivity_state (*check_connectivity)( + grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_error **connectivity_error); + + /** call notify when the connectivity state of a channel changes from *state. + Updates *state with the new state of the policy. Calling with a NULL \a + state cancels the subscription. + */ + void (*notify_on_state_change)(grpc_exec_ctx *exec_ctx, + grpc_lb_policy *policy, + grpc_connectivity_state *state, + grpc_closure *closure); +}; + +/*#define GRPC_LB_POLICY_REFCOUNT_DEBUG*/ +#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG +#define GRPC_LB_POLICY_REF(p, r) \ + grpc_lb_policy_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_LB_POLICY_UNREF(exec_ctx, p, r) \ + grpc_lb_policy_unref((exec_ctx), (p), __FILE__, __LINE__, (r)) +#define GRPC_LB_POLICY_WEAK_REF(p, r) \ + grpc_lb_policy_weak_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, p, r) \ + grpc_lb_policy_weak_unref((exec_ctx), (p), __FILE__, __LINE__, (r)) +void grpc_lb_policy_ref(grpc_lb_policy *policy, const char *file, int line, + const char *reason); +void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + const char *file, int line, const char *reason); +void grpc_lb_policy_weak_ref(grpc_lb_policy *policy, const char *file, int line, + const char *reason); +void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + const char *file, int line, const char *reason); +#else +#define GRPC_LB_POLICY_REF(p, r) grpc_lb_policy_ref((p)) +#define GRPC_LB_POLICY_UNREF(cl, p, r) grpc_lb_policy_unref((cl), (p)) +#define GRPC_LB_POLICY_WEAK_REF(p, r) grpc_lb_policy_weak_ref((p)) +#define GRPC_LB_POLICY_WEAK_UNREF(cl, p, r) grpc_lb_policy_weak_unref((cl), (p)) +void grpc_lb_policy_ref(grpc_lb_policy *policy); +void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy); +void grpc_lb_policy_weak_ref(grpc_lb_policy *policy); +void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy); +#endif + +/** called by concrete implementations to initialize the base struct */ +void grpc_lb_policy_init(grpc_lb_policy *policy, + const grpc_lb_policy_vtable *vtable); + +/** Given initial metadata in \a initial_metadata, find an appropriate + target for this rpc, and 'return' it by calling \a on_complete after setting + \a target. + Picking can be asynchronous. Any IO should be done under \a pollent. */ +int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_polling_entity *pollent, + grpc_metadata_batch *initial_metadata, + uint32_t initial_metadata_flags, + grpc_connected_subchannel **target, + grpc_closure *on_complete); + +void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_closure *closure); + +void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_connected_subchannel **target); + +/** Cancel all pending picks which have: + (initial_metadata_flags & initial_metadata_flags_mask) == + initial_metadata_flags_eq */ +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); + +/** Try to enter a READY connectivity state */ +void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy); + +/* Call notify when the connectivity state of a channel changes from \a *state. + * Updates \a *state with the new state of the policy */ +void grpc_lb_policy_notify_on_state_change(grpc_exec_ctx *exec_ctx, + grpc_lb_policy *policy, + grpc_connectivity_state *state, + grpc_closure *closure); + +grpc_connectivity_state grpc_lb_policy_check_connectivity( + grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, + grpc_error **connectivity_error); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_H */ diff --git a/src/core/ext/client_channel/lb_policy_factory.c b/src/core/ext/client_channel/lb_policy_factory.c new file mode 100644 index 0000000000..56deac7d3d --- /dev/null +++ b/src/core/ext/client_channel/lb_policy_factory.c @@ -0,0 +1,49 @@ +/* + * + * 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_channel/lb_policy_factory.h" + +void grpc_lb_policy_factory_ref(grpc_lb_policy_factory* factory) { + factory->vtable->ref(factory); +} + +void grpc_lb_policy_factory_unref(grpc_lb_policy_factory* factory) { + factory->vtable->unref(factory); +} + +grpc_lb_policy* grpc_lb_policy_factory_create_lb_policy( + grpc_exec_ctx* exec_ctx, grpc_lb_policy_factory* factory, + grpc_lb_policy_args* args) { + if (factory == NULL) return NULL; + return factory->vtable->create_lb_policy(exec_ctx, factory, args); +} diff --git a/src/core/ext/client_channel/lb_policy_factory.h b/src/core/ext/client_channel/lb_policy_factory.h new file mode 100644 index 0000000000..eedade191d --- /dev/null +++ b/src/core/ext/client_channel/lb_policy_factory.h @@ -0,0 +1,76 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_FACTORY_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_FACTORY_H + +#include "src/core/ext/client_channel/client_channel_factory.h" +#include "src/core/ext/client_channel/lb_policy.h" +#include "src/core/lib/iomgr/resolve_address.h" + +#include "src/core/lib/iomgr/exec_ctx.h" + +typedef struct grpc_lb_policy_factory grpc_lb_policy_factory; +typedef struct grpc_lb_policy_factory_vtable grpc_lb_policy_factory_vtable; + +struct grpc_lb_policy_factory { + const grpc_lb_policy_factory_vtable *vtable; +}; + +typedef struct grpc_lb_policy_args { + grpc_resolved_addresses *addresses; + grpc_client_channel_factory *client_channel_factory; +} grpc_lb_policy_args; + +struct grpc_lb_policy_factory_vtable { + void (*ref)(grpc_lb_policy_factory *factory); + void (*unref)(grpc_lb_policy_factory *factory); + + /** Implementation of grpc_lb_policy_factory_create_lb_policy */ + grpc_lb_policy *(*create_lb_policy)(grpc_exec_ctx *exec_ctx, + grpc_lb_policy_factory *factory, + grpc_lb_policy_args *args); + + /** Name for the LB policy this factory implements */ + const char *name; +}; + +void grpc_lb_policy_factory_ref(grpc_lb_policy_factory *factory); +void grpc_lb_policy_factory_unref(grpc_lb_policy_factory *factory); + +/** Create a lb_policy instance. */ +grpc_lb_policy *grpc_lb_policy_factory_create_lb_policy( + grpc_exec_ctx *exec_ctx, grpc_lb_policy_factory *factory, + grpc_lb_policy_args *args); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_FACTORY_H */ diff --git a/src/core/ext/client_channel/lb_policy_registry.c b/src/core/ext/client_channel/lb_policy_registry.c new file mode 100644 index 0000000000..f46a721f9d --- /dev/null +++ b/src/core/ext/client_channel/lb_policy_registry.c @@ -0,0 +1,83 @@ +/* + * + * 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_channel/lb_policy_registry.h" + +#include <string.h> + +#define MAX_POLICIES 10 + +static grpc_lb_policy_factory *g_all_of_the_lb_policies[MAX_POLICIES]; +static int g_number_of_lb_policies = 0; + +void grpc_lb_policy_registry_init(void) { g_number_of_lb_policies = 0; } + +void grpc_lb_policy_registry_shutdown(void) { + int i; + for (i = 0; i < g_number_of_lb_policies; i++) { + grpc_lb_policy_factory_unref(g_all_of_the_lb_policies[i]); + } +} + +void grpc_register_lb_policy(grpc_lb_policy_factory *factory) { + int i; + for (i = 0; i < g_number_of_lb_policies; i++) { + GPR_ASSERT(0 != strcmp(factory->vtable->name, + g_all_of_the_lb_policies[i]->vtable->name)); + } + GPR_ASSERT(g_number_of_lb_policies != MAX_POLICIES); + grpc_lb_policy_factory_ref(factory); + g_all_of_the_lb_policies[g_number_of_lb_policies++] = factory; +} + +static grpc_lb_policy_factory *lookup_factory(const char *name) { + int i; + + if (name == NULL) return NULL; + + for (i = 0; i < g_number_of_lb_policies; i++) { + if (0 == strcmp(name, g_all_of_the_lb_policies[i]->vtable->name)) { + return g_all_of_the_lb_policies[i]; + } + } + + return NULL; +} + +grpc_lb_policy *grpc_lb_policy_create(grpc_exec_ctx *exec_ctx, const char *name, + grpc_lb_policy_args *args) { + grpc_lb_policy_factory *factory = lookup_factory(name); + grpc_lb_policy *lb_policy = + grpc_lb_policy_factory_create_lb_policy(exec_ctx, factory, args); + return lb_policy; +} diff --git a/src/core/ext/client_channel/lb_policy_registry.h b/src/core/ext/client_channel/lb_policy_registry.h new file mode 100644 index 0000000000..301861553a --- /dev/null +++ b/src/core/ext/client_channel/lb_policy_registry.h @@ -0,0 +1,55 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_REGISTRY_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_REGISTRY_H + +#include "src/core/ext/client_channel/lb_policy_factory.h" +#include "src/core/lib/iomgr/exec_ctx.h" + +/** Initialize the registry and set \a default_factory as the factory to be + * returned when no name is provided in a lookup */ +void grpc_lb_policy_registry_init(void); +void grpc_lb_policy_registry_shutdown(void); + +/** Register a LB policy factory. */ +void grpc_register_lb_policy(grpc_lb_policy_factory *factory); + +/** Create a \a grpc_lb_policy instance. + * + * If \a name is NULL, the default factory from \a grpc_lb_policy_registry_init + * will be returned. */ +grpc_lb_policy *grpc_lb_policy_create(grpc_exec_ctx *exec_ctx, const char *name, + grpc_lb_policy_args *args); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_LB_POLICY_REGISTRY_H */ diff --git a/src/core/ext/client_channel/parse_address.c b/src/core/ext/client_channel/parse_address.c new file mode 100644 index 0000000000..31ca44299b --- /dev/null +++ b/src/core/ext/client_channel/parse_address.c @@ -0,0 +1,137 @@ +/* + * + * 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/ext/client_channel/parse_address.h" + +#include <stdio.h> +#include <string.h> +#ifdef GPR_HAVE_UNIX_SOCKET +#include <sys/un.h> +#endif + +#include <grpc/support/alloc.h> +#include <grpc/support/host_port.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + +#ifdef GPR_HAVE_UNIX_SOCKET +int parse_unix(grpc_uri *uri, struct sockaddr_storage *addr, size_t *len) { + struct sockaddr_un *un = (struct sockaddr_un *)addr; + + un->sun_family = AF_UNIX; + strcpy(un->sun_path, uri->path); + *len = strlen(un->sun_path) + sizeof(un->sun_family) + 1; + + return 1; +} +#endif + +int parse_ipv4(grpc_uri *uri, struct sockaddr_storage *addr, size_t *len) { + const char *host_port = uri->path; + char *host; + char *port; + int port_num; + int result = 0; + struct sockaddr_in *in = (struct sockaddr_in *)addr; + + if (*host_port == '/') ++host_port; + if (!gpr_split_host_port(host_port, &host, &port)) { + return 0; + } + + memset(in, 0, sizeof(*in)); + *len = sizeof(*in); + in->sin_family = AF_INET; + if (inet_pton(AF_INET, host, &in->sin_addr) == 0) { + gpr_log(GPR_ERROR, "invalid ipv4 address: '%s'", host); + goto done; + } + + if (port != NULL) { + if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 || + port_num > 65535) { + gpr_log(GPR_ERROR, "invalid ipv4 port: '%s'", port); + goto done; + } + in->sin_port = htons((uint16_t)port_num); + } else { + gpr_log(GPR_ERROR, "no port given for ipv4 scheme"); + goto done; + } + + result = 1; +done: + gpr_free(host); + gpr_free(port); + return result; +} + +int parse_ipv6(grpc_uri *uri, struct sockaddr_storage *addr, size_t *len) { + const char *host_port = uri->path; + char *host; + char *port; + int port_num; + int result = 0; + struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)addr; + + if (*host_port == '/') ++host_port; + if (!gpr_split_host_port(host_port, &host, &port)) { + return 0; + } + + memset(in6, 0, sizeof(*in6)); + *len = sizeof(*in6); + in6->sin6_family = AF_INET6; + if (inet_pton(AF_INET6, host, &in6->sin6_addr) == 0) { + gpr_log(GPR_ERROR, "invalid ipv6 address: '%s'", host); + goto done; + } + + if (port != NULL) { + if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 || + port_num > 65535) { + gpr_log(GPR_ERROR, "invalid ipv6 port: '%s'", port); + goto done; + } + in6->sin6_port = htons((uint16_t)port_num); + } else { + gpr_log(GPR_ERROR, "no port given for ipv6 scheme"); + goto done; + } + + result = 1; +done: + gpr_free(host); + gpr_free(port); + return result; +} diff --git a/src/core/ext/client_channel/parse_address.h b/src/core/ext/client_channel/parse_address.h new file mode 100644 index 0000000000..1504d3916d --- /dev/null +++ b/src/core/ext/client_channel/parse_address.h @@ -0,0 +1,56 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_PARSE_ADDRESS_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_PARSE_ADDRESS_H + +#include <stddef.h> + +#include "src/core/ext/client_channel/uri_parser.h" +#include "src/core/lib/iomgr/sockaddr.h" + +#ifdef GPR_HAVE_UNIX_SOCKET +/** Populate \a addr and \a len from \a uri, whose path is expected to contain a + * unix socket path. Returns true upon success. */ +int parse_unix(grpc_uri *uri, struct sockaddr_storage *addr, size_t *len); +#endif + +/** Populate /a addr and \a len from \a uri, whose path is expected to contain a + * host:port pair. Returns true upon success. */ +int parse_ipv4(grpc_uri *uri, struct sockaddr_storage *addr, size_t *len); + +/** Populate /a addr and \a len from \a uri, whose path is expected to contain a + * host:port pair. Returns true upon success. */ +int parse_ipv6(grpc_uri *uri, struct sockaddr_storage *addr, size_t *len); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_PARSE_ADDRESS_H */ diff --git a/src/core/ext/client_channel/resolver.c b/src/core/ext/client_channel/resolver.c new file mode 100644 index 0000000000..0c26d95f2f --- /dev/null +++ b/src/core/ext/client_channel/resolver.c @@ -0,0 +1,82 @@ +/* + * + * 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_channel/resolver.h" + +void grpc_resolver_init(grpc_resolver *resolver, + const grpc_resolver_vtable *vtable) { + resolver->vtable = vtable; + gpr_ref_init(&resolver->refs, 1); +} + +#ifdef GRPC_RESOLVER_REFCOUNT_DEBUG +void grpc_resolver_ref(grpc_resolver *resolver, grpc_closure_list *closure_list, + const char *file, int line, const char *reason) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "RESOLVER:%p ref %d -> %d %s", + resolver, (int)resolver->refs.count, (int)resolver->refs.count + 1, + reason); +#else +void grpc_resolver_ref(grpc_resolver *resolver) { +#endif + gpr_ref(&resolver->refs); +} + +#ifdef GRPC_RESOLVER_REFCOUNT_DEBUG +void grpc_resolver_unref(grpc_resolver *resolver, + grpc_closure_list *closure_list, const char *file, + int line, const char *reason) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "RESOLVER:%p unref %d -> %d %s", + resolver, (int)resolver->refs.count, (int)resolver->refs.count - 1, + reason); +#else +void grpc_resolver_unref(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver) { +#endif + if (gpr_unref(&resolver->refs)) { + resolver->vtable->destroy(exec_ctx, resolver); + } +} + +void grpc_resolver_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver) { + resolver->vtable->shutdown(exec_ctx, resolver); +} + +void grpc_resolver_channel_saw_error(grpc_exec_ctx *exec_ctx, + grpc_resolver *resolver) { + resolver->vtable->channel_saw_error(exec_ctx, resolver); +} + +void grpc_resolver_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver, + grpc_resolver_result **result, + grpc_closure *on_complete) { + resolver->vtable->next(exec_ctx, resolver, result, on_complete); +} diff --git a/src/core/ext/client_channel/resolver.h b/src/core/ext/client_channel/resolver.h new file mode 100644 index 0000000000..017c55bb80 --- /dev/null +++ b/src/core/ext/client_channel/resolver.h @@ -0,0 +1,94 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_H + +#include "src/core/ext/client_channel/resolver_result.h" +#include "src/core/ext/client_channel/subchannel.h" +#include "src/core/lib/iomgr/iomgr.h" + +typedef struct grpc_resolver grpc_resolver; +typedef struct grpc_resolver_vtable grpc_resolver_vtable; + +/** grpc_resolver provides grpc_resolver_result objects to grpc_channel + objects */ +struct grpc_resolver { + const grpc_resolver_vtable *vtable; + gpr_refcount refs; +}; + +struct grpc_resolver_vtable { + void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver); + void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver); + void (*channel_saw_error)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver); + void (*next)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver, + grpc_resolver_result **result, grpc_closure *on_complete); +}; + +#ifdef GRPC_RESOLVER_REFCOUNT_DEBUG +#define GRPC_RESOLVER_REF(p, r) grpc_resolver_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_RESOLVER_UNREF(cl, p, r) \ + grpc_resolver_unref((cl), (p), __FILE__, __LINE__, (r)) +void grpc_resolver_ref(grpc_resolver *policy, const char *file, int line, + const char *reason); +void grpc_resolver_unref(grpc_resolver *policy, grpc_closure_list *closure_list, + const char *file, int line, const char *reason); +#else +#define GRPC_RESOLVER_REF(p, r) grpc_resolver_ref((p)) +#define GRPC_RESOLVER_UNREF(cl, p, r) grpc_resolver_unref((cl), (p)) +void grpc_resolver_ref(grpc_resolver *policy); +void grpc_resolver_unref(grpc_exec_ctx *exec_ctx, grpc_resolver *policy); +#endif + +void grpc_resolver_init(grpc_resolver *resolver, + const grpc_resolver_vtable *vtable); + +void grpc_resolver_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver); + +/** Notification that the channel has seen an error on some address. + Can be used as a hint that re-resolution is desirable soon. */ +void grpc_resolver_channel_saw_error(grpc_exec_ctx *exec_ctx, + grpc_resolver *resolver); + +/** Get the next client config. Called by the channel to fetch a new + configuration. Expected to set *result with a new configuration, + and then schedule on_complete for execution. + + If resolution is fatally broken, set *result to NULL and + schedule on_complete. */ +void grpc_resolver_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver, + grpc_resolver_result **result, + grpc_closure *on_complete); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_H */ diff --git a/src/core/ext/client_channel/resolver_factory.c b/src/core/ext/client_channel/resolver_factory.c new file mode 100644 index 0000000000..7c3d644257 --- /dev/null +++ b/src/core/ext/client_channel/resolver_factory.c @@ -0,0 +1,55 @@ +/* + * + * 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_channel/resolver_factory.h" + +void grpc_resolver_factory_ref(grpc_resolver_factory* factory) { + factory->vtable->ref(factory); +} + +void grpc_resolver_factory_unref(grpc_resolver_factory* factory) { + factory->vtable->unref(factory); +} + +/** Create a resolver instance for a name */ +grpc_resolver* grpc_resolver_factory_create_resolver( + grpc_resolver_factory* factory, grpc_resolver_args* args) { + if (factory == NULL) return NULL; + return factory->vtable->create_resolver(factory, args); +} + +char* grpc_resolver_factory_get_default_authority( + grpc_resolver_factory* factory, grpc_uri* uri) { + if (factory == NULL) return NULL; + return factory->vtable->get_default_authority(factory, uri); +} diff --git a/src/core/ext/client_channel/resolver_factory.h b/src/core/ext/client_channel/resolver_factory.h new file mode 100644 index 0000000000..1068acfdd4 --- /dev/null +++ b/src/core/ext/client_channel/resolver_factory.h @@ -0,0 +1,82 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_FACTORY_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_FACTORY_H + +#include "src/core/ext/client_channel/client_channel_factory.h" +#include "src/core/ext/client_channel/resolver.h" +#include "src/core/ext/client_channel/uri_parser.h" + +typedef struct grpc_resolver_factory grpc_resolver_factory; +typedef struct grpc_resolver_factory_vtable grpc_resolver_factory_vtable; + +/** grpc_resolver provides grpc_resolver_result objects to grpc_channel + objects */ +struct grpc_resolver_factory { + const grpc_resolver_factory_vtable *vtable; +}; + +typedef struct grpc_resolver_args { + grpc_uri *uri; + grpc_client_channel_factory *client_channel_factory; +} grpc_resolver_args; + +struct grpc_resolver_factory_vtable { + void (*ref)(grpc_resolver_factory *factory); + void (*unref)(grpc_resolver_factory *factory); + + /** Implementation of grpc_resolver_factory_create_resolver */ + grpc_resolver *(*create_resolver)(grpc_resolver_factory *factory, + grpc_resolver_args *args); + + /** Implementation of grpc_resolver_factory_get_default_authority */ + char *(*get_default_authority)(grpc_resolver_factory *factory, grpc_uri *uri); + + /** URI scheme that this factory implements */ + const char *scheme; +}; + +void grpc_resolver_factory_ref(grpc_resolver_factory *resolver); +void grpc_resolver_factory_unref(grpc_resolver_factory *resolver); + +/** Create a resolver instance for a name */ +grpc_resolver *grpc_resolver_factory_create_resolver( + grpc_resolver_factory *factory, grpc_resolver_args *args); + +/** Return a (freshly allocated with gpr_malloc) string representing + the default authority to use for this scheme. */ +char *grpc_resolver_factory_get_default_authority( + grpc_resolver_factory *factory, grpc_uri *uri); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_FACTORY_H */ diff --git a/src/core/ext/client_channel/resolver_registry.c b/src/core/ext/client_channel/resolver_registry.c new file mode 100644 index 0000000000..5827819efb --- /dev/null +++ b/src/core/ext/client_channel/resolver_registry.c @@ -0,0 +1,151 @@ +/* + * + * 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_channel/resolver_registry.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + +#define MAX_RESOLVERS 10 + +static grpc_resolver_factory *g_all_of_the_resolvers[MAX_RESOLVERS]; +static int g_number_of_resolvers = 0; + +static char *g_default_resolver_prefix; + +void grpc_resolver_registry_init(const char *default_resolver_prefix) { + g_default_resolver_prefix = gpr_strdup(default_resolver_prefix); +} + +void grpc_resolver_registry_shutdown(void) { + int i; + for (i = 0; i < g_number_of_resolvers; i++) { + grpc_resolver_factory_unref(g_all_of_the_resolvers[i]); + } + gpr_free(g_default_resolver_prefix); + // FIXME(ctiller): this should live in grpc_resolver_registry_init, + // however that would have the client_channel plugin call this AFTER we start + // registering resolvers from third party plugins, and so they'd never show + // up. + // We likely need some kind of dependency system for plugins.... what form + // that takes is TBD. + g_number_of_resolvers = 0; +} + +void grpc_register_resolver_type(grpc_resolver_factory *factory) { + int i; + for (i = 0; i < g_number_of_resolvers; i++) { + GPR_ASSERT(0 != strcmp(factory->vtable->scheme, + g_all_of_the_resolvers[i]->vtable->scheme)); + } + GPR_ASSERT(g_number_of_resolvers != MAX_RESOLVERS); + grpc_resolver_factory_ref(factory); + g_all_of_the_resolvers[g_number_of_resolvers++] = factory; +} + +static grpc_resolver_factory *lookup_factory(const char *name) { + int i; + + for (i = 0; i < g_number_of_resolvers; i++) { + if (0 == strcmp(name, g_all_of_the_resolvers[i]->vtable->scheme)) { + return g_all_of_the_resolvers[i]; + } + } + + return NULL; +} + +grpc_resolver_factory *grpc_resolver_factory_lookup(const char *name) { + grpc_resolver_factory *f = lookup_factory(name); + if (f) grpc_resolver_factory_ref(f); + return f; +} + +static grpc_resolver_factory *lookup_factory_by_uri(grpc_uri *uri) { + if (!uri) return NULL; + return lookup_factory(uri->scheme); +} + +static grpc_resolver_factory *resolve_factory(const char *target, + grpc_uri **uri) { + char *tmp; + grpc_resolver_factory *factory = NULL; + + GPR_ASSERT(uri != NULL); + *uri = grpc_uri_parse(target, 1); + factory = lookup_factory_by_uri(*uri); + if (factory == NULL) { + if (g_default_resolver_prefix != NULL) { + grpc_uri_destroy(*uri); + gpr_asprintf(&tmp, "%s%s", g_default_resolver_prefix, target); + *uri = grpc_uri_parse(tmp, 1); + factory = lookup_factory_by_uri(*uri); + if (factory == NULL) { + grpc_uri_destroy(grpc_uri_parse(target, 0)); + grpc_uri_destroy(grpc_uri_parse(tmp, 0)); + gpr_log(GPR_ERROR, "don't know how to resolve '%s' or '%s'", target, + tmp); + } + gpr_free(tmp); + } else { + grpc_uri_destroy(grpc_uri_parse(target, 0)); + gpr_log(GPR_ERROR, "don't know how to resolve '%s'", target); + } + } + return factory; +} + +grpc_resolver *grpc_resolver_create( + const char *target, grpc_client_channel_factory *client_channel_factory) { + grpc_uri *uri = NULL; + grpc_resolver_factory *factory = resolve_factory(target, &uri); + grpc_resolver *resolver; + grpc_resolver_args args; + memset(&args, 0, sizeof(args)); + args.uri = uri; + args.client_channel_factory = client_channel_factory; + resolver = grpc_resolver_factory_create_resolver(factory, &args); + grpc_uri_destroy(uri); + return resolver; +} + +char *grpc_get_default_authority(const char *target) { + grpc_uri *uri = NULL; + grpc_resolver_factory *factory = resolve_factory(target, &uri); + char *authority = grpc_resolver_factory_get_default_authority(factory, uri); + grpc_uri_destroy(uri); + return authority; +} diff --git a/src/core/ext/client_channel/resolver_registry.h b/src/core/ext/client_channel/resolver_registry.h new file mode 100644 index 0000000000..cdf5dff8ba --- /dev/null +++ b/src/core/ext/client_channel/resolver_registry.h @@ -0,0 +1,69 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_REGISTRY_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_REGISTRY_H + +#include "src/core/ext/client_channel/resolver_factory.h" + +void grpc_resolver_registry_init(const char *default_prefix); +void grpc_resolver_registry_shutdown(void); + +/** Register a resolver type. + URI's of \a scheme will be resolved with the given resolver. + If \a priority is greater than zero, then the resolver will be eligible + to resolve names that are passed in with no scheme. Higher priority + resolvers will be tried before lower priority schemes. */ +void grpc_register_resolver_type(grpc_resolver_factory *factory); + +/** Create a resolver given \a target. + First tries to parse \a target as a URI. If this succeeds, tries + to locate a registered resolver factory based on the URI scheme. + If parsing or location fails, prefixes default_prefix from + grpc_resolver_registry_init to target, and tries again (if default_prefix + was not NULL). + If a resolver factory was found, use it to instantiate a resolver and + return it. + If a resolver factory was not found, return NULL. */ +grpc_resolver *grpc_resolver_create( + const char *target, grpc_client_channel_factory *client_channel_factory); + +/** Find a resolver factory given a name and return an (owned-by-the-caller) + * reference to it */ +grpc_resolver_factory *grpc_resolver_factory_lookup(const char *name); + +/** Given a target, return a (freshly allocated with gpr_malloc) string + representing the default authority to pass from a client. */ +char *grpc_get_default_authority(const char *target); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_REGISTRY_H */ diff --git a/src/core/ext/client_channel/resolver_result.c b/src/core/ext/client_channel/resolver_result.c new file mode 100644 index 0000000000..3a4162b0ff --- /dev/null +++ b/src/core/ext/client_channel/resolver_result.c @@ -0,0 +1,75 @@ +/* + * + * 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_channel/resolver_result.h" + +#include <string.h> + +#include <grpc/support/alloc.h> + +struct grpc_resolver_result { + gpr_refcount refs; + grpc_lb_policy *lb_policy; +}; + +grpc_resolver_result *grpc_resolver_result_create() { + grpc_resolver_result *c = gpr_malloc(sizeof(*c)); + memset(c, 0, sizeof(*c)); + gpr_ref_init(&c->refs, 1); + return c; +} + +void grpc_resolver_result_ref(grpc_resolver_result *c) { gpr_ref(&c->refs); } + +void grpc_resolver_result_unref(grpc_exec_ctx *exec_ctx, + grpc_resolver_result *c) { + if (gpr_unref(&c->refs)) { + if (c->lb_policy != NULL) { + GRPC_LB_POLICY_UNREF(exec_ctx, c->lb_policy, "resolver_result"); + } + gpr_free(c); + } +} + +void grpc_resolver_result_set_lb_policy(grpc_resolver_result *c, + grpc_lb_policy *lb_policy) { + GPR_ASSERT(c->lb_policy == NULL); + if (lb_policy) { + GRPC_LB_POLICY_REF(lb_policy, "resolver_result"); + } + c->lb_policy = lb_policy; +} + +grpc_lb_policy *grpc_resolver_result_get_lb_policy(grpc_resolver_result *c) { + return c->lb_policy; +} diff --git a/src/core/ext/client_channel/resolver_result.h b/src/core/ext/client_channel/resolver_result.h new file mode 100644 index 0000000000..258ac7c36f --- /dev/null +++ b/src/core/ext/client_channel/resolver_result.h @@ -0,0 +1,52 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_RESULT_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_RESULT_H + +#include "src/core/ext/client_channel/lb_policy.h" + +/** Results reported from a grpc_resolver. */ +typedef struct grpc_resolver_result grpc_resolver_result; + +grpc_resolver_result *grpc_resolver_result_create(); +void grpc_resolver_result_ref(grpc_resolver_result *client_config); +void grpc_resolver_result_unref(grpc_exec_ctx *exec_ctx, + grpc_resolver_result *client_config); + +void grpc_resolver_result_set_lb_policy(grpc_resolver_result *client_config, + grpc_lb_policy *lb_policy); +grpc_lb_policy *grpc_resolver_result_get_lb_policy( + grpc_resolver_result *client_config); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_RESOLVER_RESULT_H */ diff --git a/src/core/ext/client_channel/subchannel.c b/src/core/ext/client_channel/subchannel.c new file mode 100644 index 0000000000..c97ff39e40 --- /dev/null +++ b/src/core/ext/client_channel/subchannel.c @@ -0,0 +1,730 @@ +/* + * + * 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_channel/subchannel.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/avl.h> + +#include "src/core/ext/client_channel/client_channel.h" +#include "src/core/ext/client_channel/initial_connect_string.h" +#include "src/core/ext/client_channel/subchannel_index.h" +#include "src/core/lib/channel/channel_args.h" +#include "src/core/lib/channel/connected_channel.h" +#include "src/core/lib/iomgr/timer.h" +#include "src/core/lib/profiling/timers.h" +#include "src/core/lib/support/backoff.h" +#include "src/core/lib/surface/channel.h" +#include "src/core/lib/surface/channel_init.h" +#include "src/core/lib/transport/connectivity_state.h" + +#define INTERNAL_REF_BITS 16 +#define STRONG_REF_MASK (~(gpr_atm)((1 << INTERNAL_REF_BITS) - 1)) + +#define GRPC_SUBCHANNEL_MIN_CONNECT_TIMEOUT_SECONDS 20 +#define GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS 1 +#define GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER 1.6 +#define GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS 120 +#define GRPC_SUBCHANNEL_RECONNECT_JITTER 0.2 + +#define GET_CONNECTED_SUBCHANNEL(subchannel, barrier) \ + ((grpc_connected_subchannel *)(gpr_atm_##barrier##_load( \ + &(subchannel)->connected_subchannel))) + +typedef struct { + grpc_closure closure; + grpc_subchannel *subchannel; + grpc_connectivity_state connectivity_state; +} state_watcher; + +typedef struct external_state_watcher { + grpc_subchannel *subchannel; + grpc_pollset_set *pollset_set; + grpc_closure *notify; + grpc_closure closure; + struct external_state_watcher *next; + struct external_state_watcher *prev; +} external_state_watcher; + +struct grpc_subchannel { + grpc_connector *connector; + + /** refcount + - lower INTERNAL_REF_BITS bits are for internal references: + these do not keep the subchannel open. + - upper remaining bits are for public references: these do + keep the subchannel open */ + gpr_atm ref_pair; + + /** non-transport related channel filters */ + const grpc_channel_filter **filters; + size_t num_filters; + /** channel arguments */ + grpc_channel_args *args; + /** address to connect to */ + struct sockaddr *addr; + size_t addr_len; + + grpc_subchannel_key *key; + + /** initial string to send to peer */ + gpr_slice initial_connect_string; + + /** set during connection */ + grpc_connect_out_args connecting_result; + + /** callback for connection finishing */ + grpc_closure connected; + + /** pollset_set tracking who's interested in a connection + being setup */ + grpc_pollset_set *pollset_set; + + /** active connection, or null; of type grpc_connected_subchannel */ + gpr_atm connected_subchannel; + + /** mutex protecting remaining elements */ + gpr_mu mu; + + /** have we seen a disconnection? */ + int disconnected; + /** are we connecting */ + int connecting; + /** connectivity state tracking */ + grpc_connectivity_state_tracker state_tracker; + + external_state_watcher root_external_state_watcher; + + /** next connect attempt time */ + gpr_timespec next_attempt; + /** backoff state */ + gpr_backoff backoff_state; + /** do we have an active alarm? */ + int have_alarm; + /** our alarm */ + grpc_timer alarm; +}; + +struct grpc_subchannel_call { + grpc_connected_subchannel *connection; +}; + +#define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1)) +#define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)(con)) +#define CALLSTACK_TO_SUBCHANNEL_CALL(callstack) \ + (((grpc_subchannel_call *)(callstack)) - 1) + +static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *subchannel, + grpc_error *error); + +#ifdef GRPC_STREAM_REFCOUNT_DEBUG +#define REF_REASON reason +#define REF_LOG(name, p) \ + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p ref %d -> %d %s", \ + (name), (p), (p)->refs.count, (p)->refs.count + 1, reason) +#define UNREF_LOG(name, p) \ + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p unref %d -> %d %s", \ + (name), (p), (p)->refs.count, (p)->refs.count - 1, reason) +#define REF_MUTATE_EXTRA_ARGS \ + GRPC_SUBCHANNEL_REF_EXTRA_ARGS, const char *purpose +#define REF_MUTATE_PURPOSE(x) , file, line, reason, x +#else +#define REF_REASON "" +#define REF_LOG(name, p) \ + do { \ + } while (0) +#define UNREF_LOG(name, p) \ + do { \ + } while (0) +#define REF_MUTATE_EXTRA_ARGS +#define REF_MUTATE_PURPOSE(x) +#endif + +/* + * connection implementation + */ + +static void connection_destroy(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_connected_subchannel *c = arg; + grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c)); + gpr_free(c); +} + +void grpc_connected_subchannel_ref( + grpc_connected_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + GRPC_CHANNEL_STACK_REF(CHANNEL_STACK_FROM_CONNECTION(c), REF_REASON); +} + +void grpc_connected_subchannel_unref(grpc_exec_ctx *exec_ctx, + grpc_connected_subchannel *c + GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + GRPC_CHANNEL_STACK_UNREF(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c), + REF_REASON); +} + +/* + * grpc_subchannel implementation + */ + +static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_subchannel *c = arg; + gpr_free((void *)c->filters); + grpc_channel_args_destroy(c->args); + gpr_free(c->addr); + gpr_slice_unref(c->initial_connect_string); + grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker); + grpc_connector_unref(exec_ctx, c->connector); + grpc_pollset_set_destroy(c->pollset_set); + grpc_subchannel_key_destroy(exec_ctx, c->key); + gpr_free(c); +} + +static gpr_atm ref_mutate(grpc_subchannel *c, gpr_atm delta, + int barrier REF_MUTATE_EXTRA_ARGS) { + gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, 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); +#endif + return old_val; +} + +grpc_subchannel *grpc_subchannel_ref( + grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + gpr_atm old_refs; + old_refs = ref_mutate(c, (1 << INTERNAL_REF_BITS), + 0 REF_MUTATE_PURPOSE("STRONG_REF")); + GPR_ASSERT((old_refs & STRONG_REF_MASK) != 0); + return c; +} + +grpc_subchannel *grpc_subchannel_weak_ref( + grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + gpr_atm old_refs; + old_refs = ref_mutate(c, 1, 0 REF_MUTATE_PURPOSE("WEAK_REF")); + GPR_ASSERT(old_refs != 0); + return c; +} + +grpc_subchannel *grpc_subchannel_ref_from_weak_ref( + grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + if (!c) return NULL; + for (;;) { + gpr_atm old_refs = gpr_atm_acq_load(&c->ref_pair); + if (old_refs >= (1 << INTERNAL_REF_BITS)) { + gpr_atm new_refs = old_refs + (1 << INTERNAL_REF_BITS); + if (gpr_atm_rel_cas(&c->ref_pair, old_refs, new_refs)) { + return c; + } + } else { + return NULL; + } + } +} + +static void disconnect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) { + grpc_connected_subchannel *con; + grpc_subchannel_index_unregister(exec_ctx, c->key, c); + gpr_mu_lock(&c->mu); + GPR_ASSERT(!c->disconnected); + c->disconnected = 1; + grpc_connector_shutdown(exec_ctx, c->connector); + con = GET_CONNECTED_SUBCHANNEL(c, no_barrier); + if (con != NULL) { + GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, con, "connection"); + gpr_atm_no_barrier_store(&c->connected_subchannel, (gpr_atm)0xdeadbeef); + } + gpr_mu_unlock(&c->mu); +} + +void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx, + grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + gpr_atm old_refs; + old_refs = ref_mutate(c, (gpr_atm)1 - (gpr_atm)(1 << INTERNAL_REF_BITS), + 1 REF_MUTATE_PURPOSE("STRONG_UNREF")); + if ((old_refs & STRONG_REF_MASK) == (1 << INTERNAL_REF_BITS)) { + disconnect(exec_ctx, c); + } + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "strong-unref"); +} + +void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx, + grpc_subchannel *c + GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + gpr_atm old_refs; + old_refs = ref_mutate(c, -(gpr_atm)1, 1 REF_MUTATE_PURPOSE("WEAK_UNREF")); + if (old_refs == 1) { + grpc_exec_ctx_sched(exec_ctx, grpc_closure_create(subchannel_destroy, c), + GRPC_ERROR_NONE, NULL); + } +} + +grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx, + grpc_connector *connector, + grpc_subchannel_args *args) { + grpc_subchannel_key *key = grpc_subchannel_key_create(connector, args); + grpc_subchannel *c = grpc_subchannel_index_find(exec_ctx, key); + if (c) { + grpc_subchannel_key_destroy(exec_ctx, key); + return c; + } + + c = gpr_malloc(sizeof(*c)); + memset(c, 0, sizeof(*c)); + c->key = key; + gpr_atm_no_barrier_store(&c->ref_pair, 1 << INTERNAL_REF_BITS); + c->connector = connector; + grpc_connector_ref(c->connector); + c->num_filters = args->filter_count; + if (c->num_filters > 0) { + c->filters = gpr_malloc(sizeof(grpc_channel_filter *) * c->num_filters); + memcpy((void *)c->filters, args->filters, + sizeof(grpc_channel_filter *) * c->num_filters); + } else { + c->filters = NULL; + } + c->addr = gpr_malloc(args->addr_len); + if (args->addr_len) memcpy(c->addr, args->addr, args->addr_len); + c->pollset_set = grpc_pollset_set_create(); + c->addr_len = args->addr_len; + grpc_set_initial_connect_string(&c->addr, &c->addr_len, + &c->initial_connect_string); + c->args = grpc_channel_args_copy(args->args); + c->root_external_state_watcher.next = c->root_external_state_watcher.prev = + &c->root_external_state_watcher; + grpc_closure_init(&c->connected, subchannel_connected, c); + grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE, + "subchannel"); + gpr_backoff_init(&c->backoff_state, + GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER, + GRPC_SUBCHANNEL_RECONNECT_JITTER, + GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS * 1000, + GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS * 1000); + if (c->args) { + for (size_t i = 0; i < c->args->num_args; i++) { + if (0 == strcmp(c->args->args[i].key, + "grpc.testing.fixed_reconnect_backoff")) { + GPR_ASSERT(c->args->args[i].type == GRPC_ARG_INTEGER); + gpr_backoff_init(&c->backoff_state, 1.0, 0.0, + c->args->args[i].value.integer, + c->args->args[i].value.integer); + } + if (0 == + strcmp(c->args->args[i].key, GRPC_ARG_MAX_RECONNECT_BACKOFF_MS)) { + if (c->args->args[i].type == GRPC_ARG_INTEGER) { + if (c->args->args[i].value.integer >= 0) { + gpr_backoff_init( + &c->backoff_state, GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER, + GRPC_SUBCHANNEL_RECONNECT_JITTER, + GPR_MIN(c->args->args[i].value.integer, + GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS * 1000), + c->args->args[i].value.integer); + } else { + gpr_log(GPR_ERROR, GRPC_ARG_MAX_RECONNECT_BACKOFF_MS + " : must be non-negative"); + } + } else { + gpr_log(GPR_ERROR, + GRPC_ARG_MAX_RECONNECT_BACKOFF_MS " : must be an integer"); + } + } + } + } + gpr_mu_init(&c->mu); + + return grpc_subchannel_index_register(exec_ctx, key, c); +} + +static void continue_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) { + grpc_connect_in_args args; + + args.interested_parties = c->pollset_set; + args.addr = c->addr; + args.addr_len = c->addr_len; + args.deadline = c->next_attempt; + args.channel_args = c->args; + args.initial_connect_string = c->initial_connect_string; + + grpc_connectivity_state_set(exec_ctx, &c->state_tracker, + GRPC_CHANNEL_CONNECTING, GRPC_ERROR_NONE, + "state_change"); + grpc_connector_connect(exec_ctx, c->connector, &args, &c->connecting_result, + &c->connected); +} + +static void start_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) { + c->next_attempt = + gpr_backoff_begin(&c->backoff_state, gpr_now(GPR_CLOCK_MONOTONIC)); + continue_connect(exec_ctx, c); +} + +grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c, + grpc_error **error) { + grpc_connectivity_state state; + gpr_mu_lock(&c->mu); + state = grpc_connectivity_state_check(&c->state_tracker, error); + gpr_mu_unlock(&c->mu); + return state; +} + +static void on_external_state_watcher_done(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + external_state_watcher *w = arg; + grpc_closure *follow_up = w->notify; + if (w->pollset_set != NULL) { + grpc_pollset_set_del_pollset_set(exec_ctx, w->subchannel->pollset_set, + w->pollset_set); + } + gpr_mu_lock(&w->subchannel->mu); + w->next->prev = w->prev; + w->prev->next = w->next; + gpr_mu_unlock(&w->subchannel->mu); + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, w->subchannel, "external_state_watcher"); + gpr_free(w); + follow_up->cb(exec_ctx, follow_up->cb_arg, error); +} + +void grpc_subchannel_notify_on_state_change( + grpc_exec_ctx *exec_ctx, grpc_subchannel *c, + grpc_pollset_set *interested_parties, grpc_connectivity_state *state, + grpc_closure *notify) { + external_state_watcher *w; + + if (state == NULL) { + gpr_mu_lock(&c->mu); + for (w = c->root_external_state_watcher.next; + w != &c->root_external_state_watcher; w = w->next) { + if (w->notify == notify) { + grpc_connectivity_state_notify_on_state_change( + exec_ctx, &c->state_tracker, NULL, &w->closure); + } + } + gpr_mu_unlock(&c->mu); + } else { + w = gpr_malloc(sizeof(*w)); + w->subchannel = c; + w->pollset_set = interested_parties; + w->notify = notify; + grpc_closure_init(&w->closure, on_external_state_watcher_done, w); + if (interested_parties != NULL) { + grpc_pollset_set_add_pollset_set(exec_ctx, c->pollset_set, + interested_parties); + } + GRPC_SUBCHANNEL_WEAK_REF(c, "external_state_watcher"); + gpr_mu_lock(&c->mu); + w->next = &c->root_external_state_watcher; + w->prev = w->next->prev; + w->next->prev = w->prev->next = w; + if (grpc_connectivity_state_notify_on_state_change( + exec_ctx, &c->state_tracker, state, &w->closure)) { + c->connecting = 1; + /* released by connection */ + GRPC_SUBCHANNEL_WEAK_REF(c, "connecting"); + start_connect(exec_ctx, c); + } + gpr_mu_unlock(&c->mu); + } +} + +void grpc_connected_subchannel_process_transport_op( + grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con, + grpc_transport_op *op) { + grpc_channel_stack *channel_stack = CHANNEL_STACK_FROM_CONNECTION(con); + grpc_channel_element *top_elem = grpc_channel_stack_element(channel_stack, 0); + top_elem->filter->start_transport_op(exec_ctx, top_elem, op); +} + +static void subchannel_on_child_state_changed(grpc_exec_ctx *exec_ctx, void *p, + grpc_error *error) { + state_watcher *sw = p; + grpc_subchannel *c = sw->subchannel; + gpr_mu *mu = &c->mu; + + gpr_mu_lock(mu); + + /* if we failed just leave this closure */ + if (sw->connectivity_state == GRPC_CHANNEL_TRANSIENT_FAILURE) { + /* any errors on a subchannel ==> we're done, create a new one */ + sw->connectivity_state = GRPC_CHANNEL_SHUTDOWN; + } + grpc_connectivity_state_set(exec_ctx, &c->state_tracker, + sw->connectivity_state, GRPC_ERROR_REF(error), + "reflect_child"); + if (sw->connectivity_state != GRPC_CHANNEL_SHUTDOWN) { + grpc_connected_subchannel_notify_on_state_change( + exec_ctx, GET_CONNECTED_SUBCHANNEL(c, no_barrier), NULL, + &sw->connectivity_state, &sw->closure); + GRPC_SUBCHANNEL_WEAK_REF(c, "state_watcher"); + sw = NULL; + } + + gpr_mu_unlock(mu); + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "state_watcher"); + gpr_free(sw); +} + +static void connected_subchannel_state_op(grpc_exec_ctx *exec_ctx, + grpc_connected_subchannel *con, + grpc_pollset_set *interested_parties, + grpc_connectivity_state *state, + grpc_closure *closure) { + grpc_transport_op op; + grpc_channel_element *elem; + memset(&op, 0, sizeof(op)); + op.connectivity_state = state; + op.on_connectivity_state_change = closure; + op.bind_pollset_set = interested_parties; + elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0); + elem->filter->start_transport_op(exec_ctx, elem, &op); +} + +void grpc_connected_subchannel_notify_on_state_change( + grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con, + grpc_pollset_set *interested_parties, grpc_connectivity_state *state, + grpc_closure *closure) { + connected_subchannel_state_op(exec_ctx, con, interested_parties, state, + closure); +} + +void grpc_connected_subchannel_ping(grpc_exec_ctx *exec_ctx, + grpc_connected_subchannel *con, + grpc_closure *closure) { + grpc_transport_op op; + grpc_channel_element *elem; + memset(&op, 0, sizeof(op)); + op.send_ping = closure; + elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0); + elem->filter->start_transport_op(exec_ctx, elem, &op); +} + +static void publish_transport_locked(grpc_exec_ctx *exec_ctx, + grpc_subchannel *c) { + grpc_connected_subchannel *con; + grpc_channel_stack *stk; + state_watcher *sw_subchannel; + + /* construct channel stack */ + grpc_channel_stack_builder *builder = grpc_channel_stack_builder_create(); + grpc_channel_stack_builder_set_channel_arguments( + builder, c->connecting_result.channel_args); + grpc_channel_stack_builder_set_transport(builder, + c->connecting_result.transport); + + if (grpc_channel_init_create_stack(exec_ctx, builder, + GRPC_CLIENT_SUBCHANNEL)) { + con = grpc_channel_stack_builder_finish(exec_ctx, builder, 0, 1, + connection_destroy, NULL); + } else { + grpc_channel_stack_builder_destroy(builder); + abort(); /* TODO(ctiller): what to do here (previously we just crashed) */ + } + stk = CHANNEL_STACK_FROM_CONNECTION(con); + memset(&c->connecting_result, 0, sizeof(c->connecting_result)); + + /* initialize state watcher */ + sw_subchannel = gpr_malloc(sizeof(*sw_subchannel)); + sw_subchannel->subchannel = c; + sw_subchannel->connectivity_state = GRPC_CHANNEL_READY; + grpc_closure_init(&sw_subchannel->closure, subchannel_on_child_state_changed, + sw_subchannel); + + if (c->disconnected) { + gpr_free(sw_subchannel); + grpc_channel_stack_destroy(exec_ctx, stk); + gpr_free(con); + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting"); + return; + } + + /* publish */ + /* TODO(ctiller): this full barrier seems to clear up a TSAN failure. + I'd have expected the rel_cas below to be enough, but + seemingly it's not. + Re-evaluate if we really need this. */ + gpr_atm_full_barrier(); + GPR_ASSERT(gpr_atm_rel_cas(&c->connected_subchannel, 0, (gpr_atm)con)); + c->connecting = 0; + + /* setup subchannel watching connected subchannel for changes; subchannel + ref + for connecting is donated + to the state watcher */ + GRPC_SUBCHANNEL_WEAK_REF(c, "state_watcher"); + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting"); + grpc_connected_subchannel_notify_on_state_change( + exec_ctx, con, c->pollset_set, &sw_subchannel->connectivity_state, + &sw_subchannel->closure); + + /* signal completion */ + grpc_connectivity_state_set(exec_ctx, &c->state_tracker, GRPC_CHANNEL_READY, + GRPC_ERROR_NONE, "connected"); +} + +static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { + grpc_subchannel *c = arg; + gpr_mu_lock(&c->mu); + c->have_alarm = 0; + if (c->disconnected) { + error = GRPC_ERROR_CREATE_REFERENCING("Disconnected", &error, 1); + } else { + GRPC_ERROR_REF(error); + } + if (error == GRPC_ERROR_NONE) { + gpr_log(GPR_INFO, "Failed to connect to channel, retrying"); + c->next_attempt = + gpr_backoff_step(&c->backoff_state, gpr_now(GPR_CLOCK_MONOTONIC)); + continue_connect(exec_ctx, c); + gpr_mu_unlock(&c->mu); + } else { + gpr_mu_unlock(&c->mu); + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting"); + } + GRPC_ERROR_UNREF(error); +} + +static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_subchannel *c = arg; + grpc_channel_args *delete_channel_args = c->connecting_result.channel_args; + + GRPC_SUBCHANNEL_WEAK_REF(c, "connected"); + gpr_mu_lock(&c->mu); + if (c->connecting_result.transport != NULL) { + publish_transport_locked(exec_ctx, c); + } else if (c->disconnected) { + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting"); + } else { + gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); + GPR_ASSERT(!c->have_alarm); + 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), + "connect_failed"); + gpr_timespec time_til_next = gpr_time_sub(c->next_attempt, now); + const char *errmsg = grpc_error_string(error); + gpr_log(GPR_INFO, "Connect failed: %s", errmsg); + if (gpr_time_cmp(time_til_next, gpr_time_0(time_til_next.clock_type)) <= + 0) { + gpr_log(GPR_INFO, "Retry immediately"); + } else { + gpr_log(GPR_INFO, "Retry in %" PRId64 ".%09d seconds", + time_til_next.tv_sec, time_til_next.tv_nsec); + } + grpc_timer_init(exec_ctx, &c->alarm, c->next_attempt, on_alarm, c, now); + grpc_error_free_string(errmsg); + } + gpr_mu_unlock(&c->mu); + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting"); + grpc_channel_args_destroy(delete_channel_args); +} + +/* + * grpc_subchannel_call implementation + */ + +static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call, + grpc_error *error) { + grpc_subchannel_call *c = call; + GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0); + grpc_connected_subchannel *connection = c->connection; + grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), NULL, c); + GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, connection, "subchannel_call"); + GPR_TIMER_END("grpc_subchannel_call_unref.destroy", 0); +} + +void grpc_subchannel_call_ref( + grpc_subchannel_call *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + GRPC_CALL_STACK_REF(SUBCHANNEL_CALL_TO_CALL_STACK(c), REF_REASON); +} + +void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx, + grpc_subchannel_call *c + GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + GRPC_CALL_STACK_UNREF(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), REF_REASON); +} + +char *grpc_subchannel_call_get_peer(grpc_exec_ctx *exec_ctx, + grpc_subchannel_call *call) { + grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call); + grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0); + return top_elem->filter->get_peer(exec_ctx, top_elem); +} + +void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx, + grpc_subchannel_call *call, + grpc_transport_stream_op *op) { + GPR_TIMER_BEGIN("grpc_subchannel_call_process_op", 0); + grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call); + grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0); + top_elem->filter->start_transport_stream_op(exec_ctx, top_elem, op); + GPR_TIMER_END("grpc_subchannel_call_process_op", 0); +} + +grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel( + grpc_subchannel *c) { + return GET_CONNECTED_SUBCHANNEL(c, acq); +} + +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_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); + if (error != GRPC_ERROR_NONE) { + const char *error_string = grpc_error_string(error); + gpr_log(GPR_ERROR, "error: %s", error_string); + grpc_error_free_string(error_string); + gpr_free(*call); + return error; + } + GRPC_CONNECTED_SUBCHANNEL_REF(con, "subchannel_call"); + grpc_call_stack_set_pollset_or_pollset_set(exec_ctx, callstk, pollent); + return GRPC_ERROR_NONE; +} + +grpc_call_stack *grpc_subchannel_call_get_call_stack( + grpc_subchannel_call *subchannel_call) { + return SUBCHANNEL_CALL_TO_CALL_STACK(subchannel_call); +} diff --git a/src/core/ext/client_channel/subchannel.h b/src/core/ext/client_channel/subchannel.h new file mode 100644 index 0000000000..b4183b459f --- /dev/null +++ b/src/core/ext/client_channel/subchannel.h @@ -0,0 +1,175 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_H + +#include "src/core/ext/client_channel/connector.h" +#include "src/core/lib/channel/channel_stack.h" +#include "src/core/lib/iomgr/polling_entity.h" +#include "src/core/lib/transport/connectivity_state.h" + +/** A (sub-)channel that knows how to connect to exactly one target + address. Provides a target for load balancing. */ +typedef struct grpc_subchannel grpc_subchannel; +typedef struct grpc_connected_subchannel grpc_connected_subchannel; +typedef struct grpc_subchannel_call grpc_subchannel_call; +typedef struct grpc_subchannel_args grpc_subchannel_args; + +#ifdef GRPC_STREAM_REFCOUNT_DEBUG +#define GRPC_SUBCHANNEL_REF(p, r) \ + grpc_subchannel_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(p, r) \ + grpc_subchannel_ref_from_weak_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_UNREF(cl, p, r) \ + grpc_subchannel_unref((cl), (p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_WEAK_REF(p, r) \ + grpc_subchannel_weak_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_WEAK_UNREF(cl, p, r) \ + grpc_subchannel_weak_unref((cl), (p), __FILE__, __LINE__, (r)) +#define GRPC_CONNECTED_SUBCHANNEL_REF(p, r) \ + grpc_connected_subchannel_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_CONNECTED_SUBCHANNEL_UNREF(cl, p, r) \ + grpc_connected_subchannel_unref((cl), (p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_CALL_REF(p, r) \ + grpc_subchannel_call_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \ + grpc_subchannel_call_unref((cl), (p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_REF_EXTRA_ARGS \ + , const char *file, int line, const char *reason +#else +#define GRPC_SUBCHANNEL_REF(p, r) grpc_subchannel_ref((p)) +#define GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(p, r) \ + grpc_subchannel_ref_from_weak_ref((p)) +#define GRPC_SUBCHANNEL_UNREF(cl, p, r) grpc_subchannel_unref((cl), (p)) +#define GRPC_SUBCHANNEL_WEAK_REF(p, r) grpc_subchannel_weak_ref((p)) +#define GRPC_SUBCHANNEL_WEAK_UNREF(cl, p, r) \ + grpc_subchannel_weak_unref((cl), (p)) +#define GRPC_CONNECTED_SUBCHANNEL_REF(p, r) grpc_connected_subchannel_ref((p)) +#define GRPC_CONNECTED_SUBCHANNEL_UNREF(cl, p, r) \ + grpc_connected_subchannel_unref((cl), (p)) +#define GRPC_SUBCHANNEL_CALL_REF(p, r) grpc_subchannel_call_ref((p)) +#define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \ + grpc_subchannel_call_unref((cl), (p)) +#define GRPC_SUBCHANNEL_REF_EXTRA_ARGS +#endif + +grpc_subchannel *grpc_subchannel_ref( + grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +grpc_subchannel *grpc_subchannel_ref_from_weak_ref( + grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx, + grpc_subchannel *channel + GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +grpc_subchannel *grpc_subchannel_weak_ref( + grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx, + grpc_subchannel *channel + GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_connected_subchannel_ref( + grpc_connected_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_connected_subchannel_unref(grpc_exec_ctx *exec_ctx, + grpc_connected_subchannel *channel + GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_call_ref( + grpc_subchannel_call *call GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx, + grpc_subchannel_call *call + GRPC_SUBCHANNEL_REF_EXTRA_ARGS); + +/** 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); + +/** process a transport level op */ +void grpc_connected_subchannel_process_transport_op( + grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *subchannel, + grpc_transport_op *op); + +/** poll the current connectivity state of a channel */ +grpc_connectivity_state grpc_subchannel_check_connectivity( + grpc_subchannel *channel, grpc_error **error); + +/** call notify when the connectivity state of a channel changes from *state. + Updates *state with the new state of the channel */ +void grpc_subchannel_notify_on_state_change( + grpc_exec_ctx *exec_ctx, grpc_subchannel *channel, + grpc_pollset_set *interested_parties, grpc_connectivity_state *state, + grpc_closure *notify); +void grpc_connected_subchannel_notify_on_state_change( + grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *channel, + grpc_pollset_set *interested_parties, grpc_connectivity_state *state, + grpc_closure *notify); +void grpc_connected_subchannel_ping(grpc_exec_ctx *exec_ctx, + grpc_connected_subchannel *channel, + grpc_closure *notify); + +/** retrieve the grpc_connected_subchannel - or NULL if called before + the subchannel becomes connected */ +grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel( + grpc_subchannel *subchannel); + +/** continue processing a transport op */ +void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx, + grpc_subchannel_call *subchannel_call, + grpc_transport_stream_op *op); + +/** continue querying for peer */ +char *grpc_subchannel_call_get_peer(grpc_exec_ctx *exec_ctx, + grpc_subchannel_call *subchannel_call); + +grpc_call_stack *grpc_subchannel_call_get_call_stack( + grpc_subchannel_call *subchannel_call); + +struct grpc_subchannel_args { + /* When updating this struct, also update subchannel_index.c */ + + /** Channel filters for this channel - wrapped factories will likely + want to mutate this */ + const grpc_channel_filter **filters; + /** The number of filters in the above array */ + size_t filter_count; + /** Channel arguments to be supplied to the newly created channel */ + const grpc_channel_args *args; + /** Address to connect to */ + struct sockaddr *addr; + size_t addr_len; +}; + +/** create a subchannel given a connector */ +grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx, + grpc_connector *connector, + grpc_subchannel_args *args); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_H */ diff --git a/src/core/ext/client_channel/subchannel_factory.c b/src/core/ext/client_channel/subchannel_factory.c new file mode 100644 index 0000000000..bfb979159b --- /dev/null +++ b/src/core/ext/client_channel/subchannel_factory.c @@ -0,0 +1,49 @@ +/* + * + * 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_channel/subchannel_factory.h" + +void grpc_subchannel_factory_ref(grpc_subchannel_factory* factory) { + factory->vtable->ref(factory); +} + +void grpc_subchannel_factory_unref(grpc_exec_ctx* exec_ctx, + grpc_subchannel_factory* factory) { + factory->vtable->unref(exec_ctx, factory); +} + +grpc_subchannel* grpc_subchannel_factory_create_subchannel( + grpc_exec_ctx* exec_ctx, grpc_subchannel_factory* factory, + grpc_subchannel_args* args) { + return factory->vtable->create_subchannel(exec_ctx, factory, args); +} diff --git a/src/core/ext/client_channel/subchannel_factory.h b/src/core/ext/client_channel/subchannel_factory.h new file mode 100644 index 0000000000..2b67806767 --- /dev/null +++ b/src/core/ext/client_channel/subchannel_factory.h @@ -0,0 +1,66 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_FACTORY_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_FACTORY_H + +#include "src/core/ext/client_channel/subchannel.h" +#include "src/core/lib/channel/channel_stack.h" + +typedef struct grpc_subchannel_factory grpc_subchannel_factory; +typedef struct grpc_subchannel_factory_vtable grpc_subchannel_factory_vtable; + +/** Constructor for new configured channels. + Creating decorators around this type is encouraged to adapt behavior. */ +struct grpc_subchannel_factory { + const grpc_subchannel_factory_vtable *vtable; +}; + +struct grpc_subchannel_factory_vtable { + void (*ref)(grpc_subchannel_factory *factory); + void (*unref)(grpc_exec_ctx *exec_ctx, grpc_subchannel_factory *factory); + grpc_subchannel *(*create_subchannel)(grpc_exec_ctx *exec_ctx, + grpc_subchannel_factory *factory, + grpc_subchannel_args *args); +}; + +void grpc_subchannel_factory_ref(grpc_subchannel_factory *factory); +void grpc_subchannel_factory_unref(grpc_exec_ctx *exec_ctx, + grpc_subchannel_factory *factory); + +/** Create a new grpc_subchannel */ +grpc_subchannel *grpc_subchannel_factory_create_subchannel( + grpc_exec_ctx *exec_ctx, grpc_subchannel_factory *factory, + grpc_subchannel_args *args); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_FACTORY_H */ diff --git a/src/core/ext/client_channel/subchannel_index.c b/src/core/ext/client_channel/subchannel_index.c new file mode 100644 index 0000000000..daf62eac17 --- /dev/null +++ b/src/core/ext/client_channel/subchannel_index.c @@ -0,0 +1,273 @@ +// +// +// 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/ext/client_channel/subchannel_index.h" + +#include <stdbool.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/avl.h> +#include <grpc/support/tls.h> + +#include "src/core/lib/channel/channel_args.h" + +// a map of subchannel_key --> subchannel, used for detecting connections +// to the same destination in order to share them +static gpr_avl g_subchannel_index; + +static gpr_mu g_mu; + +struct grpc_subchannel_key { + grpc_connector *connector; + grpc_subchannel_args args; +}; + +GPR_TLS_DECL(subchannel_index_exec_ctx); + +static void enter_ctx(grpc_exec_ctx *exec_ctx) { + GPR_ASSERT(gpr_tls_get(&subchannel_index_exec_ctx) == 0); + gpr_tls_set(&subchannel_index_exec_ctx, (intptr_t)exec_ctx); +} + +static void leave_ctx(grpc_exec_ctx *exec_ctx) { + GPR_ASSERT(gpr_tls_get(&subchannel_index_exec_ctx) == (intptr_t)exec_ctx); + gpr_tls_set(&subchannel_index_exec_ctx, 0); +} + +static grpc_exec_ctx *current_ctx() { + grpc_exec_ctx *c = (grpc_exec_ctx *)gpr_tls_get(&subchannel_index_exec_ctx); + GPR_ASSERT(c != NULL); + return c; +} + +static grpc_subchannel_key *create_key( + grpc_connector *connector, grpc_subchannel_args *args, + grpc_channel_args *(*copy_channel_args)(const grpc_channel_args *args)) { + grpc_subchannel_key *k = gpr_malloc(sizeof(*k)); + k->connector = grpc_connector_ref(connector); + k->args.filter_count = args->filter_count; + if (k->args.filter_count > 0) { + k->args.filters = + gpr_malloc(sizeof(*k->args.filters) * k->args.filter_count); + memcpy((grpc_channel_filter *)k->args.filters, args->filters, + sizeof(*k->args.filters) * k->args.filter_count); + } else { + k->args.filters = NULL; + } + k->args.addr_len = args->addr_len; + k->args.addr = gpr_malloc(args->addr_len); + if (k->args.addr_len > 0) { + memcpy(k->args.addr, args->addr, k->args.addr_len); + } + k->args.args = copy_channel_args(args->args); + return k; +} + +grpc_subchannel_key *grpc_subchannel_key_create(grpc_connector *connector, + grpc_subchannel_args *args) { + return create_key(connector, args, grpc_channel_args_normalize); +} + +static grpc_subchannel_key *subchannel_key_copy(grpc_subchannel_key *k) { + return create_key(k->connector, &k->args, grpc_channel_args_copy); +} + +static int subchannel_key_compare(grpc_subchannel_key *a, + grpc_subchannel_key *b) { + int c = GPR_ICMP(a->connector, b->connector); + if (c != 0) return c; + c = GPR_ICMP(a->args.addr_len, b->args.addr_len); + if (c != 0) return c; + c = GPR_ICMP(a->args.filter_count, b->args.filter_count); + if (c != 0) return c; + if (a->args.addr_len) { + c = memcmp(a->args.addr, b->args.addr, a->args.addr_len); + if (c != 0) return c; + } + if (a->args.filter_count > 0) { + c = memcmp(a->args.filters, b->args.filters, + a->args.filter_count * sizeof(*a->args.filters)); + if (c != 0) return c; + } + return grpc_channel_args_compare(a->args.args, b->args.args); +} + +void grpc_subchannel_key_destroy(grpc_exec_ctx *exec_ctx, + grpc_subchannel_key *k) { + grpc_connector_unref(exec_ctx, k->connector); + gpr_free(k->args.addr); + gpr_free((grpc_channel_args *)k->args.filters); + grpc_channel_args_destroy((grpc_channel_args *)k->args.args); + gpr_free(k); +} + +static void sck_avl_destroy(void *p) { + grpc_subchannel_key_destroy(current_ctx(), p); +} + +static void *sck_avl_copy(void *p) { return subchannel_key_copy(p); } + +static long sck_avl_compare(void *a, void *b) { + return subchannel_key_compare(a, b); +} + +static void scv_avl_destroy(void *p) { + GRPC_SUBCHANNEL_WEAK_UNREF(current_ctx(), p, "subchannel_index"); +} + +static void *scv_avl_copy(void *p) { + GRPC_SUBCHANNEL_WEAK_REF(p, "subchannel_index"); + return p; +} + +static const gpr_avl_vtable subchannel_avl_vtable = { + .destroy_key = sck_avl_destroy, + .copy_key = sck_avl_copy, + .compare_keys = sck_avl_compare, + .destroy_value = scv_avl_destroy, + .copy_value = scv_avl_copy}; + +void grpc_subchannel_index_init(void) { + g_subchannel_index = gpr_avl_create(&subchannel_avl_vtable); + gpr_mu_init(&g_mu); + gpr_tls_init(&subchannel_index_exec_ctx); +} + +void grpc_subchannel_index_shutdown(void) { + gpr_mu_destroy(&g_mu); + gpr_avl_unref(g_subchannel_index); + gpr_tls_destroy(&subchannel_index_exec_ctx); +} + +grpc_subchannel *grpc_subchannel_index_find(grpc_exec_ctx *exec_ctx, + grpc_subchannel_key *key) { + enter_ctx(exec_ctx); + + // Lock, and take a reference to the subchannel index. + // We don't need to do the search under a lock as avl's are immutable. + gpr_mu_lock(&g_mu); + gpr_avl index = gpr_avl_ref(g_subchannel_index); + gpr_mu_unlock(&g_mu); + + grpc_subchannel *c = + GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(gpr_avl_get(index, key), "index_find"); + gpr_avl_unref(index); + + leave_ctx(exec_ctx); + return c; +} + +grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx, + grpc_subchannel_key *key, + grpc_subchannel *constructed) { + enter_ctx(exec_ctx); + + grpc_subchannel *c = NULL; + + while (c == NULL) { + // Compare and swap loop: + // - take a reference to the current index + gpr_mu_lock(&g_mu); + gpr_avl index = gpr_avl_ref(g_subchannel_index); + gpr_mu_unlock(&g_mu); + + // - Check to see if a subchannel already exists + c = gpr_avl_get(index, key); + if (c != NULL) { + // yes -> we're done + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, constructed, "index_register"); + } else { + // no -> update the avl and compare/swap + gpr_avl updated = + gpr_avl_add(gpr_avl_ref(index), subchannel_key_copy(key), + GRPC_SUBCHANNEL_WEAK_REF(constructed, "index_register")); + + // it may happen (but it's expected to be unlikely) + // that some other thread has changed the index: + // compare/swap here to check that, and retry as necessary + gpr_mu_lock(&g_mu); + if (index.root == g_subchannel_index.root) { + GPR_SWAP(gpr_avl, updated, g_subchannel_index); + c = constructed; + } + gpr_mu_unlock(&g_mu); + + gpr_avl_unref(updated); + } + gpr_avl_unref(index); + } + + leave_ctx(exec_ctx); + + return c; +} + +void grpc_subchannel_index_unregister(grpc_exec_ctx *exec_ctx, + grpc_subchannel_key *key, + grpc_subchannel *constructed) { + enter_ctx(exec_ctx); + + bool done = false; + while (!done) { + // Compare and swap loop: + // - take a reference to the current index + gpr_mu_lock(&g_mu); + gpr_avl index = gpr_avl_ref(g_subchannel_index); + gpr_mu_unlock(&g_mu); + + // Check to see if this key still refers to the previously + // registered subchannel + grpc_subchannel *c = gpr_avl_get(index, key); + if (c != constructed) { + gpr_avl_unref(index); + break; + } + + // compare and swap the update (some other thread may have + // mutated the index behind us) + gpr_avl updated = gpr_avl_remove(gpr_avl_ref(index), key); + + gpr_mu_lock(&g_mu); + if (index.root == g_subchannel_index.root) { + GPR_SWAP(gpr_avl, updated, g_subchannel_index); + done = true; + } + gpr_mu_unlock(&g_mu); + + gpr_avl_unref(updated); + gpr_avl_unref(index); + } + + leave_ctx(exec_ctx); +} diff --git a/src/core/ext/client_channel/subchannel_index.h b/src/core/ext/client_channel/subchannel_index.h new file mode 100644 index 0000000000..a34389d564 --- /dev/null +++ b/src/core/ext/client_channel/subchannel_index.h @@ -0,0 +1,77 @@ +/* + * + * 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_SUBCHANNEL_INDEX_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_INDEX_H + +#include "src/core/ext/client_channel/connector.h" +#include "src/core/ext/client_channel/subchannel.h" + +/** \file Provides an index of active subchannels so that they can be + shared amongst channels */ + +typedef struct grpc_subchannel_key grpc_subchannel_key; + +/** Create a key that can be used to uniquely identify a subchannel */ +grpc_subchannel_key *grpc_subchannel_key_create(grpc_connector *con, + grpc_subchannel_args *args); + +/** Destroy a subchannel key */ +void grpc_subchannel_key_destroy(grpc_exec_ctx *exec_ctx, + grpc_subchannel_key *key); + +/** Given a subchannel key, find the subchannel registered for it. + Returns NULL if no such channel exists. + Thread-safe. */ +grpc_subchannel *grpc_subchannel_index_find(grpc_exec_ctx *exec_ctx, + grpc_subchannel_key *key); + +/** Register a subchannel against a key. + Takes ownership of \a constructed. + Returns the registered subchannel. This may be different from + \a constructed in the case of a registration race. */ +grpc_subchannel *grpc_subchannel_index_register(grpc_exec_ctx *exec_ctx, + grpc_subchannel_key *key, + grpc_subchannel *constructed); + +/** Remove \a constructed as the registered subchannel for \a key. */ +void grpc_subchannel_index_unregister(grpc_exec_ctx *exec_ctx, + grpc_subchannel_key *key, + grpc_subchannel *constructed); + +/** Initialize the subchannel index (global) */ +void grpc_subchannel_index_init(void); +/** Shutdown the subchannel index (global) */ +void grpc_subchannel_index_shutdown(void); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_INDEX_H */ diff --git a/src/core/ext/client_channel/uri_parser.c b/src/core/ext/client_channel/uri_parser.c new file mode 100644 index 0000000000..bcb6a1dee4 --- /dev/null +++ b/src/core/ext/client_channel/uri_parser.c @@ -0,0 +1,310 @@ +/* + * + * 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_channel/uri_parser.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/port_platform.h> +#include <grpc/support/slice.h> +#include <grpc/support/slice_buffer.h> +#include <grpc/support/string_util.h> + +#include "src/core/lib/support/string.h" + +/** a size_t default value... maps to all 1's */ +#define NOT_SET (~(size_t)0) + +static grpc_uri *bad_uri(const char *uri_text, size_t pos, const char *section, + int suppress_errors) { + char *line_prefix; + size_t pfx_len; + + if (!suppress_errors) { + gpr_asprintf(&line_prefix, "bad uri.%s: '", section); + pfx_len = strlen(line_prefix) + pos; + gpr_log(GPR_ERROR, "%s%s'", line_prefix, uri_text); + gpr_free(line_prefix); + + line_prefix = gpr_malloc(pfx_len + 1); + memset(line_prefix, ' ', pfx_len); + line_prefix[pfx_len] = 0; + gpr_log(GPR_ERROR, "%s^ here", line_prefix); + gpr_free(line_prefix); + } + + return NULL; +} + +/** Returns a copy of \a src[begin, end) */ +static char *copy_component(const char *src, size_t begin, size_t end) { + char *out = gpr_malloc(end - begin + 1); + memcpy(out, src + begin, end - begin); + out[end - begin] = 0; + return out; +} + +/** Returns how many chars to advance if \a uri_text[i] begins a valid \a pchar + * production. If \a uri_text[i] introduces an invalid \a pchar (such as percent + * sign not followed by two hex digits), NOT_SET is returned. */ +static size_t parse_pchar(const char *uri_text, size_t i) { + /* pchar = unreserved / pct-encoded / sub-delims / ":" / "@" + * unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~" + * pct-encoded = "%" HEXDIG HEXDIG + * sub-delims = "!" / "$" / "&" / "'" / "(" / ")" + / "*" / "+" / "," / ";" / "=" */ + char c = uri_text[i]; + if (((c >= 'A') && (c <= 'Z')) || ((c >= 'a') && (c <= 'z')) || + ((c >= '0') && (c <= '9')) || + (c == '-' || c == '.' || c == '_' || c == '~') || /* unreserved */ + (c == '!' || c == '$' || c == '&' || c == '\'' || c == '$' || c == '&' || + c == '(' || c == ')' || c == '*' || c == '+' || c == ',' || c == ';' || + c == '=') /* sub-delims */) { + return 1; + } + if (c == '%') { /* pct-encoded */ + size_t j; + if (uri_text[i + 1] == 0 || uri_text[i + 2] == 0) { + return NOT_SET; + } + for (j = i + 1; j < 2; j++) { + c = uri_text[j]; + if (!(((c >= '0') && (c <= '9')) || ((c >= 'a') && (c <= 'f')) || + ((c >= 'A') && (c <= 'F')))) { + return NOT_SET; + } + } + return 2; + } + return 0; +} + +/* *( pchar / "?" / "/" ) */ +static int parse_fragment_or_query(const char *uri_text, size_t *i) { + char c; + while ((c = uri_text[*i]) != 0) { + const size_t advance = parse_pchar(uri_text, *i); /* pchar */ + switch (advance) { + case 0: /* uri_text[i] isn't in pchar */ + /* maybe it's ? or / */ + if (uri_text[*i] == '?' || uri_text[*i] == '/') { + (*i)++; + break; + } else { + return 1; + } + GPR_UNREACHABLE_CODE(return 0); + default: + (*i) += advance; + break; + case NOT_SET: /* uri_text[i] introduces an invalid URI */ + return 0; + } + } + /* *i is the first uri_text position past the \a query production, maybe \0 */ + return 1; +} + +static void do_nothing(void *ignored) {} +static void parse_query_parts(grpc_uri *uri) { + static const char *QUERY_PARTS_SEPARATOR = "&"; + static const char *QUERY_PARTS_VALUE_SEPARATOR = "="; + GPR_ASSERT(uri->query != NULL); + if (uri->query[0] == '\0') { + uri->query_parts = NULL; + uri->query_parts_values = NULL; + uri->num_query_parts = 0; + return; + } + gpr_slice query_slice = + gpr_slice_new(uri->query, strlen(uri->query), do_nothing); + gpr_slice_buffer query_parts; /* the &-separated elements of the query */ + gpr_slice_buffer query_param_parts; /* the =-separated subelements */ + + gpr_slice_buffer_init(&query_parts); + gpr_slice_buffer_init(&query_param_parts); + + gpr_slice_split(query_slice, QUERY_PARTS_SEPARATOR, &query_parts); + uri->query_parts = gpr_malloc(query_parts.count * sizeof(char *)); + uri->query_parts_values = gpr_malloc(query_parts.count * sizeof(char *)); + uri->num_query_parts = query_parts.count; + for (size_t i = 0; i < query_parts.count; i++) { + gpr_slice_split(query_parts.slices[i], QUERY_PARTS_VALUE_SEPARATOR, + &query_param_parts); + GPR_ASSERT(query_param_parts.count > 0); + uri->query_parts[i] = + gpr_dump_slice(query_param_parts.slices[0], GPR_DUMP_ASCII); + if (query_param_parts.count > 1) { + /* TODO(dgq): only the first value after the separator is considered. + * Perhaps all chars after the first separator for the query part should + * be included, even if they include the separator. */ + uri->query_parts_values[i] = + gpr_dump_slice(query_param_parts.slices[1], GPR_DUMP_ASCII); + } else { + uri->query_parts_values[i] = NULL; + } + gpr_slice_buffer_reset_and_unref(&query_param_parts); + } + gpr_slice_buffer_destroy(&query_parts); + gpr_slice_buffer_destroy(&query_param_parts); + gpr_slice_unref(query_slice); +} + +grpc_uri *grpc_uri_parse(const char *uri_text, int suppress_errors) { + grpc_uri *uri; + size_t scheme_begin = 0; + size_t scheme_end = NOT_SET; + size_t authority_begin = NOT_SET; + size_t authority_end = NOT_SET; + size_t path_begin = NOT_SET; + size_t path_end = NOT_SET; + size_t query_begin = NOT_SET; + size_t query_end = NOT_SET; + size_t fragment_begin = NOT_SET; + size_t fragment_end = NOT_SET; + size_t i; + + for (i = scheme_begin; uri_text[i] != 0; i++) { + if (uri_text[i] == ':') { + scheme_end = i; + break; + } + if (uri_text[i] >= 'a' && uri_text[i] <= 'z') continue; + if (uri_text[i] >= 'A' && uri_text[i] <= 'Z') continue; + if (i != scheme_begin) { + if (uri_text[i] >= '0' && uri_text[i] <= '9') continue; + if (uri_text[i] == '+') continue; + if (uri_text[i] == '-') continue; + if (uri_text[i] == '.') continue; + } + break; + } + if (scheme_end == NOT_SET) { + return bad_uri(uri_text, i, "scheme", suppress_errors); + } + + if (uri_text[scheme_end + 1] == '/' && uri_text[scheme_end + 2] == '/') { + authority_begin = scheme_end + 3; + for (i = authority_begin; uri_text[i] != 0 && authority_end == NOT_SET; + i++) { + if (uri_text[i] == '/' || uri_text[i] == '?' || uri_text[i] == '#') { + authority_end = i; + } + } + if (authority_end == NOT_SET && uri_text[i] == 0) { + authority_end = i; + } + if (authority_end == NOT_SET) { + return bad_uri(uri_text, i, "authority", suppress_errors); + } + /* TODO(ctiller): parse the authority correctly */ + path_begin = authority_end; + } else { + path_begin = scheme_end + 1; + } + + for (i = path_begin; uri_text[i] != 0; i++) { + if (uri_text[i] == '?' || uri_text[i] == '#') { + path_end = i; + break; + } + } + if (path_end == NOT_SET && uri_text[i] == 0) { + path_end = i; + } + if (path_end == NOT_SET) { + return bad_uri(uri_text, i, "path", suppress_errors); + } + + if (uri_text[i] == '?') { + query_begin = ++i; + if (!parse_fragment_or_query(uri_text, &i)) { + return bad_uri(uri_text, i, "query", suppress_errors); + } else if (uri_text[i] != 0 && uri_text[i] != '#') { + /* We must be at the end or at the beginning of a fragment */ + return bad_uri(uri_text, i, "query", suppress_errors); + } + query_end = i; + } + if (uri_text[i] == '#') { + fragment_begin = ++i; + if (!parse_fragment_or_query(uri_text, &i)) { + return bad_uri(uri_text, i - fragment_end, "fragment", suppress_errors); + } else if (uri_text[i] != 0) { + /* We must be at the end */ + return bad_uri(uri_text, i, "fragment", suppress_errors); + } + fragment_end = i; + } + + uri = gpr_malloc(sizeof(*uri)); + memset(uri, 0, sizeof(*uri)); + uri->scheme = copy_component(uri_text, scheme_begin, scheme_end); + uri->authority = copy_component(uri_text, authority_begin, authority_end); + uri->path = copy_component(uri_text, path_begin, path_end); + uri->query = copy_component(uri_text, query_begin, query_end); + uri->fragment = copy_component(uri_text, fragment_begin, fragment_end); + parse_query_parts(uri); + + return uri; +} + +const char *grpc_uri_get_query_arg(const grpc_uri *uri, const char *key) { + GPR_ASSERT(key != NULL); + if (key[0] == '\0') return NULL; + + for (size_t i = 0; i < uri->num_query_parts; ++i) { + if (0 == strcmp(key, uri->query_parts[i])) { + return uri->query_parts_values[i]; + } + } + return NULL; +} + +void grpc_uri_destroy(grpc_uri *uri) { + if (!uri) return; + gpr_free(uri->scheme); + gpr_free(uri->authority); + gpr_free(uri->path); + gpr_free(uri->query); + for (size_t i = 0; i < uri->num_query_parts; ++i) { + gpr_free(uri->query_parts[i]); + gpr_free(uri->query_parts_values[i]); + } + gpr_free(uri->query_parts); + gpr_free(uri->query_parts_values); + gpr_free(uri->fragment); + gpr_free(uri); +} diff --git a/src/core/ext/client_channel/uri_parser.h b/src/core/ext/client_channel/uri_parser.h new file mode 100644 index 0000000000..875a7cb07c --- /dev/null +++ b/src/core/ext/client_channel/uri_parser.h @@ -0,0 +1,63 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_URI_PARSER_H +#define GRPC_CORE_EXT_CLIENT_CONFIG_URI_PARSER_H + +#include <stddef.h> + +typedef struct { + char *scheme; + char *authority; + char *path; + char *query; + /** Query substrings separated by '&' */ + char **query_parts; + /** Number of elements in \a query_parts and \a query_parts_values */ + size_t num_query_parts; + /** Split each query part by '='. NULL if not present. */ + char **query_parts_values; + char *fragment; +} grpc_uri; + +/** parse a uri, return NULL on failure */ +grpc_uri *grpc_uri_parse(const char *uri_text, int suppress_errors); + +/** return the part of a query string after the '=' in "?key=xxx&...", or NULL + * if key is not present */ +const char *grpc_uri_get_query_arg(const grpc_uri *uri, const char *key); + +/** destroy a uri */ +void grpc_uri_destroy(grpc_uri *uri); + +#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_URI_PARSER_H */ |