| /* |
| * |
| * 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/filters/client_channel/subchannel.h" |
| |
| #include <limits.h> |
| #include <string.h> |
| |
| #include <grpc/support/alloc.h> |
| #include <grpc/support/avl.h> |
| #include <grpc/support/string_util.h> |
| |
| #include "src/core/ext/filters/client_channel/client_channel.h" |
| #include "src/core/ext/filters/client_channel/parse_address.h" |
| #include "src/core/ext/filters/client_channel/proxy_mapper_registry.h" |
| #include "src/core/ext/filters/client_channel/subchannel_index.h" |
| #include "src/core/ext/filters/client_channel/uri_parser.h" |
| #include "src/core/lib/channel/channel_args.h" |
| #include "src/core/lib/channel/connected_channel.h" |
| #include "src/core/lib/iomgr/sockaddr_utils.h" |
| #include "src/core/lib/iomgr/timer.h" |
| #include "src/core/lib/profiling/timers.h" |
| #include "src/core/lib/slice/slice_internal.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; |
| |
| grpc_subchannel_key *key; |
| |
| /** set during connection */ |
| grpc_connect_out_args connecting_result; |
| |
| /** callback for connection finishing */ |
| grpc_closure connected; |
| |
| /** callback for our alarm */ |
| grpc_closure on_alarm; |
| |
| /** 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? */ |
| bool disconnected; |
| /** are we connecting */ |
| bool 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? */ |
| bool have_alarm; |
| /** have we started the backoff loop */ |
| bool backoff_begun; |
| /** our alarm */ |
| grpc_timer alarm; |
| }; |
| |
| struct grpc_subchannel_call { |
| grpc_connected_subchannel *connection; |
| grpc_closure *schedule_closure_after_destroy; |
| }; |
| |
| #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); |
| } |
| |
| grpc_connected_subchannel *grpc_connected_subchannel_ref( |
| grpc_connected_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { |
| GRPC_CHANNEL_STACK_REF(CHANNEL_STACK_FROM_CONNECTION(c), REF_REASON); |
| return c; |
| } |
| |
| 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(exec_ctx, c->args); |
| grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker); |
| grpc_connector_unref(exec_ctx, c->connector); |
| grpc_pollset_set_destroy(exec_ctx, 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 %s 0x%08" PRIxPTR " -> 0x%08" PRIxPTR " [%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 = true; |
| grpc_connector_shutdown( |
| exec_ctx, c->connector, |
| GRPC_ERROR_CREATE_FROM_STATIC_STRING("Subchannel disconnected")); |
| 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_closure_sched(exec_ctx, grpc_closure_create(subchannel_destroy, c, |
| grpc_schedule_on_exec_ctx), |
| GRPC_ERROR_NONE); |
| } |
| } |
| |
| grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx, |
| grpc_connector *connector, |
| const 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_zalloc(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->pollset_set = grpc_pollset_set_create(); |
| grpc_resolved_address *addr = gpr_malloc(sizeof(*addr)); |
| grpc_get_subchannel_address_arg(exec_ctx, args->args, addr); |
| grpc_resolved_address *new_address = NULL; |
| grpc_channel_args *new_args = NULL; |
| if (grpc_proxy_mappers_map_address(exec_ctx, addr, args->args, &new_address, |
| &new_args)) { |
| GPR_ASSERT(new_address != NULL); |
| gpr_free(addr); |
| addr = new_address; |
| } |
| static const char *keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS}; |
| grpc_arg new_arg = grpc_create_subchannel_address_arg(addr); |
| gpr_free(addr); |
| c->args = grpc_channel_args_copy_and_add_and_remove( |
| new_args != NULL ? new_args : args->args, keys_to_remove, |
| GPR_ARRAY_SIZE(keys_to_remove), &new_arg, 1); |
| gpr_free(new_arg.value.string); |
| if (new_args != NULL) grpc_channel_args_destroy(exec_ctx, new_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_schedule_on_exec_ctx); |
| grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE, |
| "subchannel"); |
| int initial_backoff_ms = |
| GRPC_SUBCHANNEL_INITIAL_CONNECT_BACKOFF_SECONDS * 1000; |
| int max_backoff_ms = GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS * 1000; |
| int min_backoff_ms = GRPC_SUBCHANNEL_MIN_CONNECT_TIMEOUT_SECONDS * 1000; |
| bool fixed_reconnect_backoff = false; |
| 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_ms")) { |
| fixed_reconnect_backoff = true; |
| initial_backoff_ms = min_backoff_ms = max_backoff_ms = |
| grpc_channel_arg_get_integer( |
| &c->args->args[i], |
| (grpc_integer_options){initial_backoff_ms, 100, INT_MAX}); |
| } else if (0 == strcmp(c->args->args[i].key, |
| GRPC_ARG_MAX_RECONNECT_BACKOFF_MS)) { |
| fixed_reconnect_backoff = false; |
| max_backoff_ms = grpc_channel_arg_get_integer( |
| &c->args->args[i], |
| (grpc_integer_options){max_backoff_ms, 100, INT_MAX}); |
| } else if (0 == strcmp(c->args->args[i].key, |
| GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS)) { |
| fixed_reconnect_backoff = false; |
| initial_backoff_ms = grpc_channel_arg_get_integer( |
| &c->args->args[i], |
| (grpc_integer_options){initial_backoff_ms, 100, INT_MAX}); |
| } |
| } |
| } |
| gpr_backoff_init( |
| &c->backoff_state, initial_backoff_ms, |
| fixed_reconnect_backoff ? 1.0 |
| : GRPC_SUBCHANNEL_RECONNECT_BACKOFF_MULTIPLIER, |
| fixed_reconnect_backoff ? 0.0 : GRPC_SUBCHANNEL_RECONNECT_JITTER, |
| min_backoff_ms, max_backoff_ms); |
| gpr_mu_init(&c->mu); |
| |
| return grpc_subchannel_index_register(exec_ctx, key, c); |
| } |
| |
| static void continue_connect_locked(grpc_exec_ctx *exec_ctx, |
| grpc_subchannel *c) { |
| grpc_connect_in_args args; |
| |
| args.interested_parties = c->pollset_set; |
| args.deadline = c->next_attempt; |
| args.channel_args = c->args; |
| |
| 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); |
| } |
| |
| 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_get(&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); |
| grpc_closure_run(exec_ctx, follow_up, GRPC_ERROR_REF(error)); |
| } |
| |
| 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 = false; |
| if (c->disconnected) { |
| error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING("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_locked(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 maybe_start_connecting_locked(grpc_exec_ctx *exec_ctx, |
| grpc_subchannel *c) { |
| if (c->disconnected) { |
| /* Don't try to connect if we're already disconnected */ |
| return; |
| } |
| |
| if (c->connecting) { |
| /* Already connecting: don't restart */ |
| return; |
| } |
| |
| if (GET_CONNECTED_SUBCHANNEL(c, no_barrier) != NULL) { |
| /* Already connected: don't restart */ |
| return; |
| } |
| |
| if (!grpc_connectivity_state_has_watchers(&c->state_tracker)) { |
| /* Nobody is interested in connecting: so don't just yet */ |
| return; |
| } |
| |
| c->connecting = true; |
| GRPC_SUBCHANNEL_WEAK_REF(c, "connecting"); |
| |
| gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); |
| if (!c->backoff_begun) { |
| c->backoff_begun = true; |
| c->next_attempt = gpr_backoff_begin(&c->backoff_state, now); |
| continue_connect_locked(exec_ctx, c); |
| } else { |
| GPR_ASSERT(!c->have_alarm); |
| c->have_alarm = true; |
| gpr_timespec time_til_next = gpr_time_sub(c->next_attempt, now); |
| 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_closure_init(&c->on_alarm, on_alarm, c, grpc_schedule_on_exec_ctx); |
| grpc_timer_init(exec_ctx, &c->alarm, c->next_attempt, &c->on_alarm, now); |
| } |
| } |
| |
| 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, |
| grpc_schedule_on_exec_ctx); |
| 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; |
| grpc_connectivity_state_notify_on_state_change(exec_ctx, &c->state_tracker, |
| state, &w->closure); |
| maybe_start_connecting_locked(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_make_transport_op(NULL); |
| grpc_channel_element *elem; |
| 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_make_transport_op(NULL); |
| grpc_channel_element *elem; |
| 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( |
| exec_ctx, 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)) { |
| grpc_channel_stack_builder_destroy(exec_ctx, builder); |
| abort(); /* TODO(ctiller): what to do here (previously we just crashed) */ |
| } |
| grpc_error *error = grpc_channel_stack_builder_finish( |
| exec_ctx, builder, 0, 1, connection_destroy, NULL, (void **)&con); |
| if (error != GRPC_ERROR_NONE) { |
| gpr_log(GPR_ERROR, "error initializing subchannel stack: %s", |
| grpc_error_string(error)); |
| GRPC_ERROR_UNREF(error); |
| abort(); /* TODO(ctiller): what to do here? */ |
| } |
| 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, grpc_schedule_on_exec_ctx); |
| |
| 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)); |
| |
| /* 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 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); |
| c->connecting = false; |
| 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 { |
| grpc_connectivity_state_set( |
| exec_ctx, &c->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, |
| grpc_error_set_int(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( |
| "Connect Failed", &error, 1), |
| GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE), |
| "connect_failed"); |
| |
| const char *errmsg = grpc_error_string(error); |
| gpr_log(GPR_INFO, "Connect failed: %s", errmsg); |
| |
| maybe_start_connecting_locked(exec_ctx, c); |
| GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting"); |
| } |
| gpr_mu_unlock(&c->mu); |
| GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connected"); |
| grpc_channel_args_destroy(exec_ctx, 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_ASSERT(c->schedule_closure_after_destroy != NULL); |
| 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->schedule_closure_after_destroy); |
| GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, connection, "subchannel_call"); |
| GPR_TIMER_END("grpc_subchannel_call_unref.destroy", 0); |
| } |
| |
| void grpc_subchannel_call_set_cleanup_closure(grpc_subchannel_call *call, |
| grpc_closure *closure) { |
| GPR_ASSERT(call->schedule_closure_after_destroy == NULL); |
| GPR_ASSERT(closure != NULL); |
| call->schedule_closure_after_destroy = closure; |
| } |
| |
| 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_batch *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_batch(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, |
| const grpc_connected_subchannel_call_args *args, |
| grpc_subchannel_call **call) { |
| grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con); |
| *call = gpr_arena_alloc( |
| args->arena, 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. |
| const grpc_call_element_args call_args = {.call_stack = callstk, |
| .server_transport_data = NULL, |
| .context = NULL, |
| .path = args->path, |
| .start_time = args->start_time, |
| .deadline = args->deadline, |
| .arena = args->arena}; |
| grpc_error *error = grpc_call_stack_init( |
| exec_ctx, chanstk, 1, subchannel_call_destroy, *call, &call_args); |
| if (error != GRPC_ERROR_NONE) { |
| const char *error_string = grpc_error_string(error); |
| gpr_log(GPR_ERROR, "error: %s", 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, args->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); |
| } |
| |
| static void grpc_uri_to_sockaddr(grpc_exec_ctx *exec_ctx, const char *uri_str, |
| grpc_resolved_address *addr) { |
| grpc_uri *uri = grpc_uri_parse(exec_ctx, uri_str, 0 /* suppress_errors */); |
| GPR_ASSERT(uri != NULL); |
| if (strcmp(uri->scheme, "ipv4") == 0) { |
| GPR_ASSERT(parse_ipv4(uri, addr)); |
| } else if (strcmp(uri->scheme, "ipv6") == 0) { |
| GPR_ASSERT(parse_ipv6(uri, addr)); |
| } else { |
| GPR_ASSERT(parse_unix(uri, addr)); |
| } |
| grpc_uri_destroy(uri); |
| } |
| |
| void grpc_get_subchannel_address_arg(grpc_exec_ctx *exec_ctx, |
| const grpc_channel_args *args, |
| grpc_resolved_address *addr) { |
| const char *addr_uri_str = grpc_get_subchannel_address_uri_arg(args); |
| memset(addr, 0, sizeof(*addr)); |
| if (*addr_uri_str != '\0') { |
| grpc_uri_to_sockaddr(exec_ctx, addr_uri_str, addr); |
| } |
| } |
| |
| const char *grpc_get_subchannel_address_uri_arg(const grpc_channel_args *args) { |
| const grpc_arg *addr_arg = |
| grpc_channel_args_find(args, GRPC_ARG_SUBCHANNEL_ADDRESS); |
| GPR_ASSERT(addr_arg != NULL); // Should have been set by LB policy. |
| GPR_ASSERT(addr_arg->type == GRPC_ARG_STRING); |
| return addr_arg->value.string; |
| } |
| |
| grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address *addr) { |
| grpc_arg new_arg; |
| new_arg.key = GRPC_ARG_SUBCHANNEL_ADDRESS; |
| new_arg.type = GRPC_ARG_STRING; |
| new_arg.value.string = |
| addr->len > 0 ? grpc_sockaddr_to_uri(addr) : gpr_strdup(""); |
| return new_arg; |
| } |