Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Initial pass through to make subchannels single connect

Browse Source
reviewable/pr4232/r1
Craig Tiller 9 years ago
parent 860f484f43
commit b5585d4f72
12 changed files with 232 additions and 476 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 22
      src/core/channel/client_channel.c
  2. 20
      src/core/channel/client_uchannel.c
  3. 4
      src/core/channel/client_uchannel.h
  4. 54
      src/core/channel/subchannel_call_holder.c
  5. 7
      src/core/channel/subchannel_call_holder.h
  6. 37
      src/core/client_config/lb_policies/pick_first.c
  7. 18
      src/core/client_config/lb_policies/round_robin.c
  8. 4
      src/core/client_config/lb_policy.c
  9. 8
      src/core/client_config/lb_policy.h
  10. 481
      src/core/client_config/subchannel.c
  11. 51
      src/core/client_config/subchannel.h
  12. 2
      src/core/transport/transport.h

22
src/core/channel/client_channel.c
Unescape View File

@ -287,7 +287,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
typedef struct { typedef struct {
grpc_metadata_batch *initial_metadata; grpc_metadata_batch *initial_metadata;
grpc_subchannel **subchannel; grpc_connected_subchannel **connected_subchannel;
grpc_closure *on_ready; grpc_closure *on_ready;
grpc_call_element *elem; grpc_call_element *elem;
grpc_closure closure; grpc_closure closure;
@ -295,17 +295,17 @@ typedef struct {
static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg, static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg,
grpc_metadata_batch *initial_metadata, grpc_metadata_batch *initial_metadata,
grpc_subchannel **subchannel, grpc_connected_subchannel **connected_subchannel,
grpc_closure *on_ready); grpc_closure *on_ready);
static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, int success) { static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, int success) {
continue_picking_args *cpa = arg; continue_picking_args *cpa = arg;
if (!success) { if (!success) {
grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 0); grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 0);
} else if (cpa->subchannel == NULL) { } else if (cpa->connected_subchannel == NULL) {
/* cancelled, do nothing */ /* cancelled, do nothing */
} else if (cc_pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata, } else if (cc_pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata,
cpa->subchannel, cpa->on_ready)) { cpa->connected_subchannel, cpa->on_ready)) {
grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 1); grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 1);
} }
gpr_free(cpa); gpr_free(cpa);
@ -313,7 +313,7 @@ static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, int success) {
static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
grpc_metadata_batch *initial_metadata, grpc_metadata_batch *initial_metadata,
grpc_subchannel **subchannel, grpc_connected_subchannel **connected_subchannel,
grpc_closure *on_ready) { grpc_closure *on_ready) {
grpc_call_element *elem = elemp; grpc_call_element *elem = elemp;
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
@ -321,18 +321,18 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
continue_picking_args *cpa; continue_picking_args *cpa;
grpc_closure *closure; grpc_closure *closure;
GPR_ASSERT(subchannel); GPR_ASSERT(connected_subchannel);
gpr_mu_lock(&chand->mu_config); gpr_mu_lock(&chand->mu_config);
if (initial_metadata == NULL) { if (initial_metadata == NULL) {
if (chand->lb_policy != NULL) { if (chand->lb_policy != NULL) {
grpc_lb_policy_cancel_pick(exec_ctx, chand->lb_policy, subchannel); grpc_lb_policy_cancel_pick(exec_ctx, chand->lb_policy, connected_subchannel);
} }
for (closure = chand->waiting_for_config_closures.head; closure != NULL; for (closure = chand->waiting_for_config_closures.head; closure != NULL;
closure = grpc_closure_next(closure)) { closure = grpc_closure_next(closure)) {
cpa = closure->cb_arg; cpa = closure->cb_arg;
if (cpa->subchannel == subchannel) { if (cpa->connected_subchannel == connected_subchannel) {
cpa->subchannel = NULL; cpa->connected_subchannel = NULL;
grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 0); grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 0);
} }
} }
@ -341,7 +341,7 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
} }
if (chand->lb_policy != NULL) { if (chand->lb_policy != NULL) {
int r = grpc_lb_policy_pick(exec_ctx, chand->lb_policy, calld->pollset, int r = grpc_lb_policy_pick(exec_ctx, chand->lb_policy, calld->pollset,
initial_metadata, subchannel, on_ready); initial_metadata, connected_subchannel, on_ready);
gpr_mu_unlock(&chand->mu_config); gpr_mu_unlock(&chand->mu_config);
return r; return r;
} }
@ -354,7 +354,7 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
} }
cpa = gpr_malloc(sizeof(*cpa)); cpa = gpr_malloc(sizeof(*cpa));
cpa->initial_metadata = initial_metadata; cpa->initial_metadata = initial_metadata;
cpa->subchannel = subchannel; cpa->connected_subchannel = connected_subchannel;
cpa->on_ready = on_ready; cpa->on_ready = on_ready;
cpa->elem = elem; cpa->elem = elem;
grpc_closure_init(&cpa->closure, continue_picking, cpa); grpc_closure_init(&cpa->closure, continue_picking, cpa);

20
src/core/channel/client_uchannel.c
Unescape View File

@ -67,7 +67,7 @@ typedef struct client_uchannel_channel_data {
grpc_connectivity_state_tracker state_tracker; grpc_connectivity_state_tracker state_tracker;
/** the subchannel wrapped by the microchannel */ /** the subchannel wrapped by the microchannel */
grpc_subchannel *subchannel; grpc_connected_subchannel *connected_subchannel;
/** the callback used to stay subscribed to subchannel connectivity /** the callback used to stay subscribed to subchannel connectivity
* notifications */ * notifications */
@ -87,7 +87,7 @@ static void monitor_subchannel(grpc_exec_ctx *exec_ctx, void *arg,
grpc_connectivity_state_set(exec_ctx, &chand->state_tracker, grpc_connectivity_state_set(exec_ctx, &chand->state_tracker,
chand->subchannel_connectivity, chand->subchannel_connectivity,
"uchannel_monitor_subchannel"); "uchannel_monitor_subchannel");
grpc_subchannel_notify_on_state_change(exec_ctx, chand->subchannel, grpc_connected_subchannel_notify_on_state_change(exec_ctx, chand->connected_subchannel,
&chand->subchannel_connectivity, &chand->subchannel_connectivity,
&chand->connectivity_cb); &chand->connectivity_cb);
} }
@ -131,11 +131,11 @@ static void cuc_start_transport_op(grpc_exec_ctx *exec_ctx,
static int cuc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg, static int cuc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg,
grpc_metadata_batch *initial_metadata, grpc_metadata_batch *initial_metadata,
grpc_subchannel **subchannel, grpc_connected_subchannel **connected_subchannel,
grpc_closure *on_ready) { grpc_closure *on_ready) {
channel_data *chand = arg; channel_data *chand = arg;
GPR_ASSERT(initial_metadata != NULL); GPR_ASSERT(initial_metadata != NULL);
*subchannel = chand->subchannel; *connected_subchannel = chand->connected_subchannel;
return 1; return 1;
} }
@ -172,7 +172,7 @@ static void cuc_init_channel_elem(grpc_exec_ctx *exec_ctx,
static void cuc_destroy_channel_elem(grpc_exec_ctx *exec_ctx, static void cuc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem) { grpc_channel_element *elem) {
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
grpc_subchannel_state_change_unsubscribe(exec_ctx, chand->subchannel, grpc_connected_subchannel_state_change_unsubscribe(exec_ctx, chand->connected_subchannel,
&chand->connectivity_cb); &chand->connectivity_cb);
grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker); grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker);
gpr_mu_destroy(&chand->mu_state); gpr_mu_destroy(&chand->mu_state);
@ -202,7 +202,7 @@ grpc_connectivity_state grpc_client_uchannel_check_connectivity_state(
GRPC_CHANNEL_CONNECTING, GRPC_CHANNEL_CONNECTING,
"uchannel_connecting_changed"); "uchannel_connecting_changed");
chand->subchannel_connectivity = out; chand->subchannel_connectivity = out;
grpc_subchannel_notify_on_state_change(exec_ctx, chand->subchannel, grpc_connected_subchannel_notify_on_state_change(exec_ctx, chand->connected_subchannel,
&chand->subchannel_connectivity, &chand->subchannel_connectivity,
&chand->connectivity_cb); &chand->connectivity_cb);
} }
@ -226,7 +226,7 @@ grpc_pollset_set *grpc_client_uchannel_get_connecting_pollset_set(
grpc_channel_element *parent_elem; grpc_channel_element *parent_elem;
gpr_mu_lock(&chand->mu_state); gpr_mu_lock(&chand->mu_state);
parent_elem = grpc_channel_stack_last_element(grpc_channel_get_channel_stack( parent_elem = grpc_channel_stack_last_element(grpc_channel_get_channel_stack(
grpc_subchannel_get_master(chand->subchannel))); chand->master));
gpr_mu_unlock(&chand->mu_state); gpr_mu_unlock(&chand->mu_state);
return grpc_client_channel_get_connecting_pollset_set(parent_elem); return grpc_client_channel_get_connecting_pollset_set(parent_elem);
} }
@ -273,13 +273,13 @@ grpc_channel *grpc_client_uchannel_create(grpc_subchannel *subchannel,
return channel; return channel;
} }
void grpc_client_uchannel_set_subchannel(grpc_channel *uchannel, void grpc_client_uchannel_set_connected_subchannel(grpc_channel *uchannel,
grpc_subchannel *subchannel) { grpc_connected_subchannel *connected_subchannel) {
grpc_channel_element *elem = grpc_channel_element *elem =
grpc_channel_stack_last_element(grpc_channel_get_channel_stack(uchannel)); grpc_channel_stack_last_element(grpc_channel_get_channel_stack(uchannel));
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
GPR_ASSERT(elem->filter == &grpc_client_uchannel_filter); GPR_ASSERT(elem->filter == &grpc_client_uchannel_filter);
gpr_mu_lock(&chand->mu_state); gpr_mu_lock(&chand->mu_state);
chand->subchannel = subchannel; chand->connected_subchannel = connected_subchannel;
gpr_mu_unlock(&chand->mu_state); gpr_mu_unlock(&chand->mu_state);
} }

4
src/core/channel/client_uchannel.h
Unescape View File

@ -64,7 +64,7 @@ void grpc_client_uchannel_del_interested_party(grpc_exec_ctx *exec_ctx,
grpc_channel *grpc_client_uchannel_create(grpc_subchannel *subchannel, grpc_channel *grpc_client_uchannel_create(grpc_subchannel *subchannel,
grpc_channel_args *args); grpc_channel_args *args);
void grpc_client_uchannel_set_subchannel(grpc_channel *uchannel, void grpc_client_uchannel_set_connected_subchannel(grpc_channel *uchannel,
grpc_subchannel *subchannel); grpc_connected_subchannel *connected_subchannel);
#endif /* GRPC_INTERNAL_CORE_CHANNEL_CLIENT_MICROCHANNEL_H */ #endif /* GRPC_INTERNAL_CORE_CHANNEL_CLIENT_MICROCHANNEL_H */

54
src/core/channel/subchannel_call_holder.c
Unescape View File

@ -44,7 +44,6 @@
static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *holder, static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *holder,
int success); int success);
static void call_ready(grpc_exec_ctx *exec_ctx, void *holder, int success);
static void retry_ops(grpc_exec_ctx *exec_ctx, void *retry_ops_args, static void retry_ops(grpc_exec_ctx *exec_ctx, void *retry_ops_args,
int success); int success);
@ -63,7 +62,7 @@ void grpc_subchannel_call_holder_init(
holder->pick_subchannel = pick_subchannel; holder->pick_subchannel = pick_subchannel;
holder->pick_subchannel_arg = pick_subchannel_arg; holder->pick_subchannel_arg = pick_subchannel_arg;
gpr_mu_init(&holder->mu); gpr_mu_init(&holder->mu);
holder->subchannel = NULL; holder->connected_subchannel = NULL;
holder->waiting_ops = NULL; holder->waiting_ops = NULL;
holder->waiting_ops_count = 0; holder->waiting_ops_count = 0;
holder->waiting_ops_capacity = 0; holder->waiting_ops_capacity = 0;
@ -125,13 +124,9 @@ retry:
case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING: case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING:
fail_locked(exec_ctx, holder); fail_locked(exec_ctx, holder);
break; break;
case GRPC_SUBCHANNEL_CALL_HOLDER_CREATING_CALL:
grpc_subchannel_cancel_create_call(exec_ctx, holder->subchannel,
&holder->subchannel_call);
break;
case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL: case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL:
holder->pick_subchannel(exec_ctx, holder->pick_subchannel_arg, NULL, holder->pick_subchannel(exec_ctx, holder->pick_subchannel_arg, NULL,
&holder->subchannel, NULL); &holder->connected_subchannel, NULL);
break; break;
} }
gpr_mu_unlock(&holder->mu); gpr_mu_unlock(&holder->mu);
@ -142,28 +137,21 @@ retry:
} }
/* if we don't have a subchannel, try to get one */ /* if we don't have a subchannel, try to get one */
if (holder->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && if (holder->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING &&
holder->subchannel == NULL && op->send_initial_metadata != NULL) { holder->connected_subchannel == NULL && op->send_initial_metadata != NULL) {
holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL; holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL;
grpc_closure_init(&holder->next_step, subchannel_ready, holder); grpc_closure_init(&holder->next_step, subchannel_ready, holder);
if (holder->pick_subchannel(exec_ctx, holder->pick_subchannel_arg, if (holder->pick_subchannel(exec_ctx, holder->pick_subchannel_arg,
op->send_initial_metadata, &holder->subchannel, op->send_initial_metadata, &holder->connected_subchannel,
&holder->next_step)) { &holder->next_step)) {
holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
} }
} }
/* if we've got a subchannel, then let's ask it to create a call */ /* if we've got a subchannel, then let's ask it to create a call */
if (holder->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && if (holder->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING &&
holder->subchannel != NULL) { holder->connected_subchannel != NULL) {
holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_CREATING_CALL; gpr_atm_rel_store(&holder->subchannel_call, grpc_connected_subchannel_create_call(exec_ctx, holder->connected_subchannel, holder->pollset));
grpc_closure_init(&holder->next_step, call_ready, holder); retry_waiting_locked(exec_ctx, holder);
if (grpc_subchannel_create_call(exec_ctx, holder->subchannel, goto retry;
holder->pollset, &holder->subchannel_call,
&holder->next_step)) {
/* got one immediately - continue the op (and any waiting ops) */
holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
retry_waiting_locked(exec_ctx, holder);
goto retry;
}
} }
/* nothing to be done but wait */ /* nothing to be done but wait */
add_waiting_locked(holder, op); add_waiting_locked(holder, op);
@ -179,36 +167,14 @@ static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, int success) {
GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL); GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL);
call = GET_CALL(holder); call = GET_CALL(holder);
GPR_ASSERT(call == NULL || call == CANCELLED_CALL); GPR_ASSERT(call == NULL || call == CANCELLED_CALL);
if (holder->subchannel == NULL) { if (holder->connected_subchannel == NULL) {
holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
fail_locked(exec_ctx, holder); fail_locked(exec_ctx, holder);
} else { } else {
grpc_closure_init(&holder->next_step, call_ready, holder); gpr_atm_rel_store(&holder->subchannel_call, grpc_connected_subchannel_create_call(exec_ctx, holder->connected_subchannel, holder->pollset));
if (grpc_subchannel_create_call(exec_ctx, holder->subchannel,
holder->pollset, &holder->subchannel_call,
&holder->next_step)) {
holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
/* got one immediately - continue the op (and any waiting ops) */
retry_waiting_locked(exec_ctx, holder);
}
}
gpr_mu_unlock(&holder->mu);
}
static void call_ready(grpc_exec_ctx *exec_ctx, void *arg, int success) {
grpc_subchannel_call_holder *holder = arg;
GPR_TIMER_BEGIN("call_ready", 0);
gpr_mu_lock(&holder->mu);
GPR_ASSERT(holder->creation_phase ==
GRPC_SUBCHANNEL_CALL_HOLDER_CREATING_CALL);
holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
if (GET_CALL(holder) != NULL) {
retry_waiting_locked(exec_ctx, holder); retry_waiting_locked(exec_ctx, holder);
} else {
fail_locked(exec_ctx, holder);
} }
gpr_mu_unlock(&holder->mu); gpr_mu_unlock(&holder->mu);
GPR_TIMER_END("call_ready", 0);
} }
typedef struct { typedef struct {

7
src/core/channel/subchannel_call_holder.h
Unescape View File

@ -42,12 +42,11 @@
called when the subchannel is available) */ called when the subchannel is available) */
typedef int (*grpc_subchannel_call_holder_pick_subchannel)( typedef int (*grpc_subchannel_call_holder_pick_subchannel)(
grpc_exec_ctx *exec_ctx, void *arg, grpc_metadata_batch *initial_metadata, grpc_exec_ctx *exec_ctx, void *arg, grpc_metadata_batch *initial_metadata,
grpc_subchannel **subchannel, grpc_closure *on_ready); grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready);
typedef enum { typedef enum {
GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING, GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING,
GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL, GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL
GRPC_SUBCHANNEL_CALL_HOLDER_CREATING_CALL
} grpc_subchannel_call_holder_creation_phase; } grpc_subchannel_call_holder_creation_phase;
/** Wrapper for holding a pointer to grpc_subchannel_call, and the /** Wrapper for holding a pointer to grpc_subchannel_call, and the
@ -71,7 +70,7 @@ typedef struct grpc_subchannel_call_holder {
gpr_mu mu; gpr_mu mu;
grpc_subchannel_call_holder_creation_phase creation_phase; grpc_subchannel_call_holder_creation_phase creation_phase;
grpc_subchannel *subchannel; grpc_connected_subchannel *connected_subchannel;
grpc_pollset *pollset; grpc_pollset *pollset;
grpc_transport_stream_op *waiting_ops; grpc_transport_stream_op *waiting_ops;

37
src/core/client_config/lb_policies/pick_first.c
Unescape View File

@ -42,7 +42,7 @@
typedef struct pending_pick { typedef struct pending_pick {
struct pending_pick *next; struct pending_pick *next;
grpc_pollset *pollset; grpc_pollset *pollset;
grpc_subchannel **target; grpc_connected_subchannel **target;
grpc_closure *on_complete; grpc_closure *on_complete;
} pending_pick; } pending_pick;
@ -60,7 +60,7 @@ typedef struct {
/** the selected channel /** the selected channel
TODO(ctiller): this should be atomically set so we don't TODO(ctiller): this should be atomically set so we don't
need to take a mutex in the common case */ need to take a mutex in the common case */
grpc_subchannel *selected; grpc_connected_subchannel *selected;
/** have we started picking? */ /** have we started picking? */
int started_picking; int started_picking;
/** are we shut down? */ /** are we shut down? */
@ -102,7 +102,7 @@ void pf_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[i], "pick_first"); GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[i], "pick_first");
} }
if (p->selected) { if (p->selected) {
GRPC_SUBCHANNEL_UNREF(exec_ctx, p->selected, "picked_first"); GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, p->selected, "picked_first");
} }
grpc_connectivity_state_destroy(exec_ctx, &p->state_tracker); grpc_connectivity_state_destroy(exec_ctx, &p->state_tracker);
gpr_free(p->subchannels); gpr_free(p->subchannels);
@ -131,7 +131,7 @@ void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
} }
static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
grpc_subchannel **target) { grpc_connected_subchannel **target) {
pick_first_lb_policy *p = (pick_first_lb_policy *)pol; pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
pending_pick *pp; pending_pick *pp;
gpr_mu_lock(&p->mu); gpr_mu_lock(&p->mu);
@ -174,7 +174,7 @@ void pf_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
} }
int pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset, int pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
grpc_metadata_batch *initial_metadata, grpc_subchannel **target, grpc_metadata_batch *initial_metadata, grpc_connected_subchannel **target,
grpc_closure *on_complete) { grpc_closure *on_complete) {
pick_first_lb_policy *p = (pick_first_lb_policy *)pol; pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
pending_pick *pp; pending_pick *pp;
@ -207,7 +207,7 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
grpc_transport_op op; grpc_transport_op op;
size_t num_subchannels = p->num_subchannels; size_t num_subchannels = p->num_subchannels;
grpc_subchannel **subchannels; grpc_subchannel **subchannels;
grpc_subchannel *exclude_subchannel; grpc_connected_subchannel *exclude_subchannel;
gpr_mu_lock(&p->mu); gpr_mu_lock(&p->mu);
subchannels = p->subchannels; subchannels = p->subchannels;
@ -218,7 +218,7 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "destroy_subchannels"); GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "destroy_subchannels");
for (i = 0; i < num_subchannels; i++) { for (i = 0; i < num_subchannels; i++) {
if (subchannels[i] != exclude_subchannel) { if (grpc_subchannel_get_connected_subchannel(subchannels[i]) != exclude_subchannel) {
memset(&op, 0, sizeof(op)); memset(&op, 0, sizeof(op));
op.disconnect = 1; op.disconnect = 1;
grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], &op); grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], &op);
@ -232,6 +232,7 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg, static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
int iomgr_success) { int iomgr_success) {
pick_first_lb_policy *p = arg; pick_first_lb_policy *p = arg;
grpc_subchannel *selected_subchannel;
pending_pick *pp; pending_pick *pp;
gpr_mu_lock(&p->mu); gpr_mu_lock(&p->mu);
@ -244,7 +245,7 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
grpc_connectivity_state_set(exec_ctx, &p->state_tracker, grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
p->checking_connectivity, "selected_changed"); p->checking_connectivity, "selected_changed");
if (p->checking_connectivity != GRPC_CHANNEL_FATAL_FAILURE) { if (p->checking_connectivity != GRPC_CHANNEL_FATAL_FAILURE) {
grpc_subchannel_notify_on_state_change(exec_ctx, p->selected, grpc_connected_subchannel_notify_on_state_change(exec_ctx, p->selected,
&p->checking_connectivity, &p->checking_connectivity,
&p->connectivity_changed); &p->connectivity_changed);
} else { } else {
@ -256,8 +257,10 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
case GRPC_CHANNEL_READY: case GRPC_CHANNEL_READY:
grpc_connectivity_state_set(exec_ctx, &p->state_tracker, grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_READY, "connecting_ready"); GRPC_CHANNEL_READY, "connecting_ready");
p->selected = p->subchannels[p->checking_subchannel]; selected_subchannel = p->subchannels[p->checking_subchannel];
GRPC_SUBCHANNEL_REF(p->selected, "picked_first"); p->selected = grpc_subchannel_get_connected_subchannel(selected_subchannel);
GPR_ASSERT(p->selected);
GRPC_CONNECTED_SUBCHANNEL_REF(p->selected, "picked_first");
/* drop the pick list: we are connected now */ /* drop the pick list: we are connected now */
GRPC_LB_POLICY_REF(&p->base, "destroy_subchannels"); GRPC_LB_POLICY_REF(&p->base, "destroy_subchannels");
grpc_exec_ctx_enqueue(exec_ctx, grpc_exec_ctx_enqueue(exec_ctx,
@ -266,12 +269,12 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
while ((pp = p->pending_picks)) { while ((pp = p->pending_picks)) {
p->pending_picks = pp->next; p->pending_picks = pp->next;
*pp->target = p->selected; *pp->target = p->selected;
grpc_subchannel_del_interested_party(exec_ctx, p->selected, grpc_subchannel_del_interested_party(exec_ctx, selected_subchannel,
pp->pollset); pp->pollset);
grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1); grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
gpr_free(pp); gpr_free(pp);
} }
grpc_subchannel_notify_on_state_change(exec_ctx, p->selected, grpc_connected_subchannel_notify_on_state_change(exec_ctx, p->selected,
&p->checking_connectivity, &p->checking_connectivity,
&p->connectivity_changed); &p->connectivity_changed);
break; break;
@ -342,14 +345,14 @@ static void pf_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
size_t i; size_t i;
size_t n; size_t n;
grpc_subchannel **subchannels; grpc_subchannel **subchannels;
grpc_subchannel *selected; grpc_connected_subchannel *selected;
gpr_mu_lock(&p->mu); gpr_mu_lock(&p->mu);
n = p->num_subchannels; n = p->num_subchannels;
subchannels = gpr_malloc(n * sizeof(*subchannels)); subchannels = gpr_malloc(n * sizeof(*subchannels));
selected = p->selected; selected = p->selected;
if (selected) { if (selected) {
GRPC_SUBCHANNEL_REF(selected, "pf_broadcast_to_selected"); GRPC_CONNECTED_SUBCHANNEL_REF(selected, "pf_broadcast_to_selected");
} }
for (i = 0; i < n; i++) { for (i = 0; i < n; i++) {
subchannels[i] = p->subchannels[i]; subchannels[i] = p->subchannels[i];
@ -358,13 +361,13 @@ static void pf_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
gpr_mu_unlock(&p->mu); gpr_mu_unlock(&p->mu);
for (i = 0; i < n; i++) { for (i = 0; i < n; i++) {
if (selected == subchannels[i]) continue; if (selected == grpc_subchannel_get_connected_subchannel(subchannels[i])) continue;
grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], op); grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], op);
GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannels[i], "pf_broadcast"); GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannels[i], "pf_broadcast");
} }
if (p->selected) { if (p->selected) {
grpc_subchannel_process_transport_op(exec_ctx, selected, op); grpc_connected_subchannel_process_transport_op(exec_ctx, selected, op);
GRPC_SUBCHANNEL_UNREF(exec_ctx, selected, "pf_broadcast_to_selected"); GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, selected, "pf_broadcast_to_selected");
} }
gpr_free(subchannels); gpr_free(subchannels);
} }

18
src/core/client_config/lb_policies/round_robin.c
Unescape View File

@ -46,7 +46,7 @@ int grpc_lb_round_robin_trace = 0;
typedef struct pending_pick { typedef struct pending_pick {
struct pending_pick *next; struct pending_pick *next;
grpc_pollset *pollset; grpc_pollset *pollset;
grpc_subchannel **target; grpc_connected_subchannel **target;
grpc_closure *on_complete; grpc_closure *on_complete;
} pending_pick; } pending_pick;
@ -144,9 +144,9 @@ static void advance_last_picked_locked(round_robin_lb_policy *p) {
/** Prepends (relative to the root at p->ready_list) the connected subchannel \a /** Prepends (relative to the root at p->ready_list) the connected subchannel \a
* csc to the list of ready subchannels. */ * csc to the list of ready subchannels. */
static ready_list *add_connected_sc_locked(round_robin_lb_policy *p, static ready_list *add_connected_sc_locked(round_robin_lb_policy *p,
grpc_subchannel *csc) { grpc_subchannel *sc) {
ready_list *new_elem = gpr_malloc(sizeof(ready_list)); ready_list *new_elem = gpr_malloc(sizeof(ready_list));
new_elem->subchannel = csc; new_elem->subchannel = sc;
if (p->ready_list.prev == NULL) { if (p->ready_list.prev == NULL) {
/* first element */ /* first element */
new_elem->next = &p->ready_list; new_elem->next = &p->ready_list;
@ -160,7 +160,7 @@ static ready_list *add_connected_sc_locked(round_robin_lb_policy *p,
p->ready_list.prev = new_elem; p->ready_list.prev = new_elem;
} }
if (grpc_lb_round_robin_trace) { if (grpc_lb_round_robin_trace) {
gpr_log(GPR_DEBUG, "[READYLIST] ADDING NODE %p (SC %p)", new_elem, csc); gpr_log(GPR_DEBUG, "[READYLIST] ADDING NODE %p (SC %p)", new_elem, sc);
} }
return new_elem; return new_elem;
} }
@ -265,7 +265,7 @@ void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
} }
static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
grpc_subchannel **target) { grpc_connected_subchannel **target) {
round_robin_lb_policy *p = (round_robin_lb_policy *)pol; round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
pending_pick *pp; pending_pick *pp;
size_t i; size_t i;
@ -314,7 +314,7 @@ void rr_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
} }
int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset, int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
grpc_metadata_batch *initial_metadata, grpc_subchannel **target, grpc_metadata_batch *initial_metadata, grpc_connected_subchannel **target,
grpc_closure *on_complete) { grpc_closure *on_complete) {
size_t i; size_t i;
round_robin_lb_policy *p = (round_robin_lb_policy *)pol; round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
@ -323,9 +323,9 @@ int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
gpr_mu_lock(&p->mu); gpr_mu_lock(&p->mu);
if ((selected = peek_next_connected_locked(p))) { if ((selected = peek_next_connected_locked(p))) {
gpr_mu_unlock(&p->mu); gpr_mu_unlock(&p->mu);
*target = selected->subchannel; *target = grpc_subchannel_get_connected_subchannel(selected->subchannel);
if (grpc_lb_round_robin_trace) { if (grpc_lb_round_robin_trace) {
gpr_log(GPR_DEBUG, "[RR PICK] TARGET <-- SUBCHANNEL %p (NODE %p)", gpr_log(GPR_DEBUG, "[RR PICK] TARGET <-- CONNECTED SUBCHANNEL %p (NODE %p)",
selected->subchannel, selected); selected->subchannel, selected);
} }
/* only advance the last picked pointer if the selection was used */ /* only advance the last picked pointer if the selection was used */
@ -390,7 +390,7 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
} }
while ((pp = p->pending_picks)) { while ((pp = p->pending_picks)) {
p->pending_picks = pp->next; p->pending_picks = pp->next;
*pp->target = selected->subchannel; *pp->target = grpc_subchannel_get_connected_subchannel(selected->subchannel);
if (grpc_lb_round_robin_trace) { if (grpc_lb_round_robin_trace) {
gpr_log(GPR_DEBUG, gpr_log(GPR_DEBUG,
"[RR CONN CHANGED] TARGET <-- SUBCHANNEL %p (NODE %p)", "[RR CONN CHANGED] TARGET <-- SUBCHANNEL %p (NODE %p)",

4
src/core/client_config/lb_policy.c
Unescape View File

@ -71,13 +71,13 @@ void grpc_lb_policy_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
grpc_pollset *pollset, grpc_pollset *pollset,
grpc_metadata_batch *initial_metadata, grpc_metadata_batch *initial_metadata,
grpc_subchannel **target, grpc_closure *on_complete) { grpc_connected_subchannel **target, grpc_closure *on_complete) {
return policy->vtable->pick(exec_ctx, policy, pollset, initial_metadata, return policy->vtable->pick(exec_ctx, policy, pollset, initial_metadata,
target, on_complete); target, on_complete);
} }
void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
grpc_subchannel **target) { grpc_connected_subchannel **target) {
policy->vtable->cancel_pick(exec_ctx, policy, target); policy->vtable->cancel_pick(exec_ctx, policy, target);
} }

8
src/core/client_config/lb_policy.h
Unescape View File

@ -58,9 +58,9 @@ struct grpc_lb_policy_vtable {
/** implement grpc_lb_policy_pick */ /** implement grpc_lb_policy_pick */
int (*pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, int (*pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
grpc_pollset *pollset, grpc_metadata_batch *initial_metadata, grpc_pollset *pollset, grpc_metadata_batch *initial_metadata,
grpc_subchannel **target, grpc_closure *on_complete); grpc_connected_subchannel **target, grpc_closure *on_complete);
void (*cancel_pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, void (*cancel_pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
grpc_subchannel **target); grpc_connected_subchannel **target);
/** try to enter a READY connectivity state */ /** try to enter a READY connectivity state */
void (*exit_idle)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy); void (*exit_idle)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
@ -111,10 +111,10 @@ void grpc_lb_policy_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
grpc_pollset *pollset, grpc_pollset *pollset,
grpc_metadata_batch *initial_metadata, grpc_metadata_batch *initial_metadata,
grpc_subchannel **target, grpc_closure *on_complete); grpc_connected_subchannel **target, grpc_closure *on_complete);
void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
grpc_subchannel **target); grpc_connected_subchannel **target);
void grpc_lb_policy_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, void grpc_lb_policy_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
grpc_transport_op *op); grpc_transport_op *op);

481
src/core/client_config/subchannel.c
Unescape View File

@ -52,33 +52,29 @@
#define GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS 120 #define GRPC_SUBCHANNEL_RECONNECT_MAX_BACKOFF_SECONDS 120
#define GRPC_SUBCHANNEL_RECONNECT_JITTER 0.2 #define GRPC_SUBCHANNEL_RECONNECT_JITTER 0.2
typedef struct { #define GET_CONNECTED_SUBCHANNEL(subchannel, barrier) \
/* all fields protected by subchannel->mu */ ((grpc_connected_subchannel *)(gpr_atm_##barrier##_load(&(subchannel)->connected_subchannel)))
struct grpc_connected_subchannel {
/** refcount */ /** refcount */
int refs; gpr_refcount refs;
/** parent subchannel */ };
grpc_subchannel *subchannel;
} connection;
typedef struct { typedef struct {
grpc_closure closure; grpc_closure closure;
size_t version; union {
grpc_subchannel *subchannel; grpc_subchannel *subchannel;
grpc_connected_subchannel *connected_subchannel;
} whom;
grpc_connectivity_state connectivity_state; grpc_connectivity_state connectivity_state;
} state_watcher; } state_watcher;
typedef struct waiting_for_connect {
struct waiting_for_connect *next;
grpc_closure *notify;
grpc_pollset *pollset;
gpr_atm *target;
grpc_subchannel *subchannel;
grpc_closure continuation;
} waiting_for_connect;
struct grpc_subchannel { struct grpc_subchannel {
grpc_connector *connector; grpc_connector *connector;
/** refcount */
gpr_refcount refs;
/** non-transport related channel filters */ /** non-transport related channel filters */
const grpc_channel_filter **filters; const grpc_channel_filter **filters;
size_t num_filters; size_t num_filters;
@ -94,8 +90,6 @@ struct grpc_subchannel {
We occasionally use this to bump the refcount on the master channel We occasionally use this to bump the refcount on the master channel
to keep ourselves alive through an asynchronous operation. */ to keep ourselves alive through an asynchronous operation. */
grpc_channel *master; grpc_channel *master;
/** have we seen a disconnection? */
int disconnected;
/** set during connection */ /** set during connection */
grpc_connect_out_args connecting_result; grpc_connect_out_args connecting_result;
@ -109,19 +103,16 @@ struct grpc_subchannel {
filter there-in) */ filter there-in) */
grpc_pollset_set *pollset_set; grpc_pollset_set *pollset_set;
/** active connection, or null; of type grpc_connected_subchannel */
gpr_atm connected_subchannel;
/** mutex protecting remaining elements */ /** mutex protecting remaining elements */
gpr_mu mu; gpr_mu mu;
/** active connection */ /** have we seen a disconnection? */
connection *active; int disconnected;
/** version number for the active connection */
size_t active_version;
/** refcount */
int refs;
/** are we connecting */ /** are we connecting */
int connecting; int connecting;
/** things waiting for a connection */
waiting_for_connect *waiting;
/** connectivity state tracking */ /** connectivity state tracking */
grpc_connectivity_state_tracker state_tracker; grpc_connectivity_state_tracker state_tracker;
@ -138,7 +129,7 @@ struct grpc_subchannel {
}; };
struct grpc_subchannel_call { struct grpc_subchannel_call {
connection *connection; grpc_connected_subchannel *connection;
}; };
#define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1)) #define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1))
@ -146,27 +137,10 @@ struct grpc_subchannel_call {
#define CALLSTACK_TO_SUBCHANNEL_CALL(callstack) \ #define CALLSTACK_TO_SUBCHANNEL_CALL(callstack) \
(((grpc_subchannel_call *)(callstack)) - 1) (((grpc_subchannel_call *)(callstack)) - 1)
static grpc_subchannel_call *create_call(grpc_exec_ctx *exec_ctx,
connection *con,
grpc_pollset *pollset);
static void connectivity_state_changed_locked(grpc_exec_ctx *exec_ctx,
grpc_subchannel *c,
const char *reason);
static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c);
static gpr_timespec compute_connect_deadline(grpc_subchannel *c); static gpr_timespec compute_connect_deadline(grpc_subchannel *c);
static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *subchannel, static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *subchannel,
int iomgr_success); int iomgr_success);
static void subchannel_ref_locked(grpc_subchannel *c
GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
static int subchannel_unref_locked(
grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT;
static void connection_ref_locked(connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
static grpc_subchannel *connection_unref_locked(
grpc_exec_ctx *exec_ctx,
connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT;
static void subchannel_destroy(grpc_exec_ctx *exec_ctx, grpc_subchannel *c);
#ifdef GRPC_STREAM_REFCOUNT_DEBUG #ifdef GRPC_STREAM_REFCOUNT_DEBUG
#define SUBCHANNEL_REF_LOCKED(p, r) \ #define SUBCHANNEL_REF_LOCKED(p, r) \
subchannel_ref_locked((p), __FILE__, __LINE__, (r)) subchannel_ref_locked((p), __FILE__, __LINE__, (r))
@ -203,66 +177,35 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, grpc_subchannel *c);
* connection implementation * connection implementation
*/ */
static void connection_destroy(grpc_exec_ctx *exec_ctx, connection *c) { static void connection_destroy(grpc_exec_ctx *exec_ctx, void *arg, int success) {
GPR_ASSERT(c->refs == 0); grpc_connected_subchannel *c = arg;
grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c)); grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c));
gpr_free(c); gpr_free(c);
} }
static void connection_ref_locked(connection *c void grpc_connected_subchannel_ref(grpc_connected_subchannel *c
GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
REF_LOG("CONNECTION", c); REF_LOG("CONNECTION", c);
subchannel_ref_locked(c->subchannel REF_PASS_ARGS); gpr_ref(&c->refs);
++c->refs;
} }
static grpc_subchannel *connection_unref_locked( void grpc_connected_subchannel_unref(
grpc_exec_ctx *exec_ctx, connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
grpc_subchannel *destroy = NULL;
UNREF_LOG("CONNECTION", c); UNREF_LOG("CONNECTION", c);
if (subchannel_unref_locked(c->subchannel REF_PASS_ARGS)) { if (gpr_unref(&c->refs)) {
destroy = c->subchannel; grpc_exec_ctx_enqueue(exec_ctx, grpc_closure_create(connection_destroy, c), 1);
}
if (--c->refs == 0 && c->subchannel->active != c) {
connection_destroy(exec_ctx, c);
} }
return destroy;
} }
/* /*
* grpc_subchannel implementation * grpc_subchannel implementation
*/ */
static void subchannel_ref_locked(grpc_subchannel *c static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg, int success) {
GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { grpc_subchannel *c = arg;
REF_LOG("SUBCHANNEL", c); grpc_connected_subchannel *con = GET_CONNECTED_SUBCHANNEL(c, no_barrier);
++c->refs; if (con != NULL) {
} GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, con, "connection");
static int subchannel_unref_locked(grpc_subchannel *c
GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
UNREF_LOG("SUBCHANNEL", c);
return --c->refs == 0;
}
void grpc_subchannel_ref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
gpr_mu_lock(&c->mu);
subchannel_ref_locked(c REF_PASS_ARGS);
gpr_mu_unlock(&c->mu);
}
void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
int destroy;
gpr_mu_lock(&c->mu);
destroy = subchannel_unref_locked(c REF_PASS_ARGS);
gpr_mu_unlock(&c->mu);
if (destroy) subchannel_destroy(exec_ctx, c);
}
static void subchannel_destroy(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
if (c->active != NULL) {
connection_destroy(exec_ctx, c->active);
} }
gpr_free((void *)c->filters); gpr_free((void *)c->filters);
grpc_channel_args_destroy(c->args); grpc_channel_args_destroy(c->args);
@ -273,6 +216,17 @@ static void subchannel_destroy(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
gpr_free(c); gpr_free(c);
} }
void grpc_subchannel_ref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
gpr_ref(&c->refs);
}
void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
if (gpr_unref(&c->refs)) {
grpc_exec_ctx_enqueue(exec_ctx, grpc_closure_create(subchannel_destroy, c), 1);
}
}
void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx, void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx,
grpc_subchannel *c, grpc_subchannel *c,
grpc_pollset *pollset) { grpc_pollset *pollset) {
@ -295,7 +249,7 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
grpc_channel_element *parent_elem = grpc_channel_stack_last_element( grpc_channel_element *parent_elem = grpc_channel_stack_last_element(
grpc_channel_get_channel_stack(args->master)); grpc_channel_get_channel_stack(args->master));
memset(c, 0, sizeof(*c)); memset(c, 0, sizeof(*c));
c->refs = 1; gpr_ref_init(&c->refs, 1);
c->connector = connector; c->connector = connector;
grpc_connector_ref(c->connector); grpc_connector_ref(c->connector);
c->num_filters = args->filter_count; c->num_filters = args->filter_count;
@ -318,60 +272,6 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
return c; return c;
} }
static void cancel_waiting_calls(grpc_exec_ctx *exec_ctx,
grpc_subchannel *subchannel,
int iomgr_success) {
waiting_for_connect *w4c;
gpr_mu_lock(&subchannel->mu);
w4c = subchannel->waiting;
subchannel->waiting = NULL;
gpr_mu_unlock(&subchannel->mu);
while (w4c != NULL) {
waiting_for_connect *next = w4c->next;
grpc_subchannel_del_interested_party(exec_ctx, w4c->subchannel,
w4c->pollset);
if (w4c->notify) {
w4c->notify->cb(exec_ctx, w4c->notify->cb_arg, iomgr_success);
}
GRPC_SUBCHANNEL_UNREF(exec_ctx, w4c->subchannel, "waiting_for_connect");
gpr_free(w4c);
w4c = next;
}
}
void grpc_subchannel_cancel_create_call(grpc_exec_ctx *exec_ctx,
grpc_subchannel *subchannel,
gpr_atm *target) {
waiting_for_connect *w4c;
int unref_count = 0;
gpr_mu_lock(&subchannel->mu);
w4c = subchannel->waiting;
subchannel->waiting = NULL;
while (w4c != NULL) {
waiting_for_connect *next = w4c->next;
if (w4c->target == target) {
grpc_subchannel_del_interested_party(exec_ctx, w4c->subchannel,
w4c->pollset);
grpc_exec_ctx_enqueue(exec_ctx, w4c->notify, 0);
unref_count++;
gpr_free(w4c);
} else {
w4c->next = subchannel->waiting;
subchannel->waiting = w4c;
}
w4c = next;
}
gpr_mu_unlock(&subchannel->mu);
while (unref_count-- > 0) {
GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannel, "waiting_for_connect");
}
}
static void continue_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) { static void continue_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
grpc_connect_in_args args; grpc_connect_in_args args;
@ -381,6 +281,7 @@ static void continue_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
args.deadline = compute_connect_deadline(c); args.deadline = compute_connect_deadline(c);
args.channel_args = c->args; args.channel_args = c->args;
grpc_connectivity_state_set(exec_ctx, &c->state_tracker, GRPC_CHANNEL_CONNECTING, "state_change");
grpc_connector_connect(exec_ctx, c->connector, &args, &c->connecting_result, grpc_connector_connect(exec_ctx, c->connector, &args, &c->connecting_result,
&c->connected); &c->connected);
} }
@ -393,66 +294,6 @@ static void start_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
continue_connect(exec_ctx, c); continue_connect(exec_ctx, c);
} }
static void continue_creating_call(grpc_exec_ctx *exec_ctx, void *arg,
int iomgr_success) {
int call_creation_finished_ok;
waiting_for_connect *w4c = arg;
grpc_subchannel_del_interested_party(exec_ctx, w4c->subchannel, w4c->pollset);
call_creation_finished_ok = grpc_subchannel_create_call(
exec_ctx, w4c->subchannel, w4c->pollset, w4c->target, w4c->notify);
GPR_ASSERT(call_creation_finished_ok == 1);
w4c->notify->cb(exec_ctx, w4c->notify->cb_arg, iomgr_success);
GRPC_SUBCHANNEL_UNREF(exec_ctx, w4c->subchannel, "waiting_for_connect");
gpr_free(w4c);
}
int grpc_subchannel_create_call(grpc_exec_ctx *exec_ctx, grpc_subchannel *c,
grpc_pollset *pollset, gpr_atm *target,
grpc_closure *notify) {
connection *con;
grpc_subchannel_call *call;
GPR_TIMER_BEGIN("grpc_subchannel_create_call", 0);
gpr_mu_lock(&c->mu);
if (c->active != NULL) {
con = c->active;
CONNECTION_REF_LOCKED(con, "call");
gpr_mu_unlock(&c->mu);
call = create_call(exec_ctx, con, pollset);
if (!gpr_atm_rel_cas(target, 0, (gpr_atm)(gpr_uintptr)call)) {
GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "failed to set");
}
GPR_TIMER_END("grpc_subchannel_create_call", 0);
return 1;
} else {
waiting_for_connect *w4c = gpr_malloc(sizeof(*w4c));
w4c->next = c->waiting;
w4c->notify = notify;
w4c->pollset = pollset;
w4c->target = target;
w4c->subchannel = c;
/* released when clearing w4c */
SUBCHANNEL_REF_LOCKED(c, "waiting_for_connect");
grpc_closure_init(&w4c->continuation, continue_creating_call, w4c);
c->waiting = w4c;
grpc_subchannel_add_interested_party(exec_ctx, c, pollset);
if (!c->connecting) {
c->connecting = 1;
connectivity_state_changed_locked(exec_ctx, c, "create_call");
/* released by connection */
SUBCHANNEL_REF_LOCKED(c, "connecting");
GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting");
gpr_mu_unlock(&c->mu);
start_connect(exec_ctx, c);
} else {
gpr_mu_unlock(&c->mu);
}
GPR_TIMER_END("grpc_subchannel_create_call", 0);
return 0;
}
}
grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) { grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) {
grpc_connectivity_state state; grpc_connectivity_state state;
gpr_mu_lock(&c->mu); gpr_mu_lock(&c->mu);
@ -472,9 +313,8 @@ void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
do_connect = 1; do_connect = 1;
c->connecting = 1; c->connecting = 1;
/* released by connection */ /* released by connection */
SUBCHANNEL_REF_LOCKED(c, "connecting"); GRPC_SUBCHANNEL_REF(c, "connecting");
GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting"); GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting");
connectivity_state_changed_locked(exec_ctx, c, "state_change");
} }
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
@ -483,31 +323,28 @@ void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
} }
} }
int grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx, void grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
grpc_subchannel *c, grpc_subchannel *c,
grpc_closure *subscribed_notify) { grpc_closure *subscribed_notify) {
int success;
gpr_mu_lock(&c->mu); gpr_mu_lock(&c->mu);
success = grpc_connectivity_state_change_unsubscribe( grpc_connectivity_state_change_unsubscribe(
exec_ctx, &c->state_tracker, subscribed_notify); exec_ctx, &c->state_tracker, subscribed_notify);
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
return success;
} }
void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx, void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
grpc_subchannel *c, grpc_subchannel *c,
grpc_transport_op *op) { grpc_transport_op *op) {
connection *con = NULL; grpc_connected_subchannel *con;
grpc_subchannel *destroy;
int cancel_alarm = 0; int cancel_alarm = 0;
gpr_mu_lock(&c->mu); gpr_mu_lock(&c->mu);
if (c->active != NULL) { con = GET_CONNECTED_SUBCHANNEL(c, no_barrier);
con = c->active; if (con != NULL) {
CONNECTION_REF_LOCKED(con, "transport-op"); GRPC_CONNECTED_SUBCHANNEL_REF(con, "transport-op");
} }
if (op->disconnect) { if (op->disconnect) {
c->disconnected = 1; c->disconnected = 1;
connectivity_state_changed_locked(exec_ctx, c, "disconnect"); grpc_connectivity_state_set(exec_ctx, &c->state_tracker, GRPC_CHANNEL_FATAL_FAILURE, "disconnect");
if (c->have_alarm) { if (c->have_alarm) {
cancel_alarm = 1; cancel_alarm = 1;
} }
@ -515,17 +352,8 @@ void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
if (con != NULL) { if (con != NULL) {
grpc_channel_stack *channel_stack = CHANNEL_STACK_FROM_CONNECTION(con); grpc_connected_subchannel_process_transport_op(exec_ctx, con, op);
grpc_channel_element *top_elem = GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, con, "transport-op");
grpc_channel_stack_element(channel_stack, 0);
top_elem->filter->start_transport_op(exec_ctx, top_elem, op);
gpr_mu_lock(&c->mu);
destroy = CONNECTION_UNREF_LOCKED(exec_ctx, con, "transport-op");
gpr_mu_unlock(&c->mu);
if (destroy) {
subchannel_destroy(exec_ctx, destroy);
}
} }
if (cancel_alarm) { if (cancel_alarm) {
@ -537,77 +365,62 @@ void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
} }
} }
static void on_state_changed(grpc_exec_ctx *exec_ctx, void *p, 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,
int iomgr_success) { int iomgr_success) {
state_watcher *sw = p; state_watcher *sw = p;
grpc_subchannel *c = sw->subchannel; grpc_subchannel *c = sw->whom.subchannel;
gpr_mu *mu = &c->mu; gpr_mu *mu = &c->mu;
int destroy;
grpc_transport_op op;
grpc_channel_element *elem;
connection *destroy_connection = NULL;
gpr_mu_lock(mu); gpr_mu_lock(mu);
/* if we failed or there is a version number mismatch, just leave /* if we failed just leave this closure */
this closure */ if (iomgr_success) {
if (!iomgr_success || sw->subchannel->active_version != sw->version) { grpc_connectivity_state_set(exec_ctx, &c->state_tracker, sw->connectivity_state, "reflect_child");
goto done; if (sw->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) {
} grpc_connected_subchannel_notify_on_state_change(exec_ctx, GET_CONNECTED_SUBCHANNEL(c, no_barrier), &sw->connectivity_state, &sw->closure);
GRPC_SUBCHANNEL_REF(c, "state_watcher");
switch (sw->connectivity_state) { sw = NULL;
case GRPC_CHANNEL_CONNECTING: }
case GRPC_CHANNEL_READY:
case GRPC_CHANNEL_IDLE:
/* all is still good: keep watching */
memset(&op, 0, sizeof(op));
op.connectivity_state = &sw->connectivity_state;
op.on_connectivity_state_change = &sw->closure;
elem = grpc_channel_stack_element(
CHANNEL_STACK_FROM_CONNECTION(c->active), 0);
elem->filter->start_transport_op(exec_ctx, elem, &op);
/* early out */
gpr_mu_unlock(mu);
return;
case GRPC_CHANNEL_FATAL_FAILURE:
case GRPC_CHANNEL_TRANSIENT_FAILURE:
/* things have gone wrong, deactivate and enter idle */
if (sw->subchannel->active->refs == 0) {
destroy_connection = sw->subchannel->active;
}
sw->subchannel->active = NULL;
grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
c->disconnected
? GRPC_CHANNEL_FATAL_FAILURE
: GRPC_CHANNEL_TRANSIENT_FAILURE,
"connection_failed");
break;
} }
done:
connectivity_state_changed_locked(exec_ctx, c, "transport_state_changed");
destroy = SUBCHANNEL_UNREF_LOCKED(c, "state_watcher");
gpr_free(sw);
gpr_mu_unlock(mu); gpr_mu_unlock(mu);
if (destroy) { GRPC_SUBCHANNEL_UNREF(exec_ctx, c, "state_watcher");
subchannel_destroy(exec_ctx, c); gpr_free(sw);
} }
if (destroy_connection != NULL) {
connection_destroy(exec_ctx, destroy_connection); static void connected_subchannel_state_op(grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con, 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;
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_connectivity_state *state, grpc_closure *closure) {
GPR_ASSERT(state != NULL);
connected_subchannel_state_op(exec_ctx, con, state, closure);
}
void grpc_connected_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con, grpc_closure *closure) {
connected_subchannel_state_op(exec_ctx, con, NULL, closure);
} }
static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) { static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
size_t channel_stack_size; size_t channel_stack_size;
connection *con; grpc_connected_subchannel *con;
grpc_channel_stack *stk; grpc_channel_stack *stk;
size_t num_filters; size_t num_filters;
const grpc_channel_filter **filters; const grpc_channel_filter **filters;
waiting_for_connect *w4c; state_watcher *sw_subchannel;
grpc_transport_op op;
state_watcher *sw;
connection *destroy_connection = NULL;
grpc_channel_element *elem;
/* build final filter list */ /* build final filter list */
num_filters = c->num_filters + c->connecting_result.num_filters + 1; num_filters = c->num_filters + c->connecting_result.num_filters + 1;
@ -619,10 +432,9 @@ static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
/* construct channel stack */ /* construct channel stack */
channel_stack_size = grpc_channel_stack_size(filters, num_filters); channel_stack_size = grpc_channel_stack_size(filters, num_filters);
con = gpr_malloc(sizeof(connection) + channel_stack_size); con = gpr_malloc(sizeof(grpc_connected_subchannel) + channel_stack_size);
stk = (grpc_channel_stack *)(con + 1); stk = (grpc_channel_stack *)(con + 1);
con->refs = 0; gpr_ref_init(&c->refs, 1);
con->subchannel = c;
grpc_channel_stack_init(exec_ctx, filters, num_filters, c->master, c->args, grpc_channel_stack_init(exec_ctx, filters, num_filters, c->master, c->args,
c->mdctx, stk); c->mdctx, stk);
grpc_connected_channel_bind_transport(stk, c->connecting_result.transport); grpc_connected_channel_bind_transport(stk, c->connecting_result.transport);
@ -630,16 +442,16 @@ static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
memset(&c->connecting_result, 0, sizeof(c->connecting_result)); memset(&c->connecting_result, 0, sizeof(c->connecting_result));
/* initialize state watcher */ /* initialize state watcher */
sw = gpr_malloc(sizeof(*sw)); sw_subchannel = gpr_malloc(sizeof(*sw_subchannel));
grpc_closure_init(&sw->closure, on_state_changed, sw); sw_subchannel->whom.subchannel = c;
sw->subchannel = c; sw_subchannel->connectivity_state = GRPC_CHANNEL_READY;
sw->connectivity_state = GRPC_CHANNEL_READY; grpc_closure_init(&sw_subchannel->closure, subchannel_on_child_state_changed, sw_subchannel);
gpr_mu_lock(&c->mu); gpr_mu_lock(&c->mu);
if (c->disconnected) { if (c->disconnected) {
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
gpr_free(sw); gpr_free(sw_subchannel);
gpr_free((void *)filters); gpr_free((void *)filters);
grpc_channel_stack_destroy(exec_ctx, stk); grpc_channel_stack_destroy(exec_ctx, stk);
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting"); GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
@ -648,45 +460,35 @@ static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
} }
/* publish */ /* publish */
if (c->active != NULL && c->active->refs == 0) { GPR_ASSERT(gpr_atm_no_barrier_cas(&c->connected_subchannel, 0, (gpr_atm)con));
destroy_connection = c->active;
}
c->active = con;
c->active_version++;
sw->version = c->active_version;
c->connecting = 0; c->connecting = 0;
/* watch for changes; subchannel ref for connecting is donated /* setup subchannel watching connected subchannel for changes; subchannel ref for connecting is donated
to the state watcher */ to the state watcher */
GRPC_SUBCHANNEL_REF(c, "state_watcher");
GRPC_SUBCHANNEL_UNREF(exec_ctx, c, "connecting");
grpc_connected_subchannel_notify_on_state_change(exec_ctx, con, &sw_subchannel->connectivity_state, &sw_subchannel->closure);
#if 0
grpc_transport_op op;
grpc_channel_element *elem;
/* setup connected subchannel watching transport for changes */
memset(&op, 0, sizeof(op)); memset(&op, 0, sizeof(op));
op.connectivity_state = &sw->connectivity_state; op.connectivity_state = &sw_connected_subchannel->connectivity_state;
op.on_connectivity_state_change = &sw->closure; op.on_connectivity_state_change = &sw_connected_subchannel->closure;
op.bind_pollset_set = c->pollset_set; op.bind_pollset_set = c->pollset_set;
SUBCHANNEL_REF_LOCKED(c, "state_watcher");
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
GPR_ASSERT(!SUBCHANNEL_UNREF_LOCKED(c, "connecting"));
elem = elem =
grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(c->active), 0); grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
elem->filter->start_transport_op(exec_ctx, elem, &op); elem->filter->start_transport_op(exec_ctx, elem, &op);
#endif
/* signal completion */ /* signal completion */
connectivity_state_changed_locked(exec_ctx, c, "connected"); grpc_connectivity_state_set(exec_ctx, &c->state_tracker, GRPC_CHANNEL_READY, "connected");
w4c = c->waiting;
c->waiting = NULL;
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
while (w4c != NULL) {
waiting_for_connect *next = w4c->next;
grpc_exec_ctx_enqueue(exec_ctx, &w4c->continuation, 1);
w4c = next;
}
gpr_free((void *)filters); gpr_free((void *)filters);
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
if (destroy_connection != NULL) {
connection_destroy(exec_ctx, destroy_connection);
}
} }
/* Generate a random number between 0 and 1. */ /* Generate a random number between 0 and 1. */
@ -725,13 +527,11 @@ static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, int iomgr_success) {
if (c->disconnected) { if (c->disconnected) {
iomgr_success = 0; iomgr_success = 0;
} }
connectivity_state_changed_locked(exec_ctx, c, "alarm");
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
if (iomgr_success) { if (iomgr_success) {
update_reconnect_parameters(c); update_reconnect_parameters(c);
continue_connect(exec_ctx, c); continue_connect(exec_ctx, c);
} else { } else {
cancel_waiting_calls(exec_ctx, c, iomgr_success);
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting"); GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
GRPC_SUBCHANNEL_UNREF(exec_ctx, c, "connecting"); GRPC_SUBCHANNEL_UNREF(exec_ctx, c, "connecting");
} }
@ -742,12 +542,14 @@ static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg,
grpc_subchannel *c = arg; grpc_subchannel *c = arg;
if (c->connecting_result.transport != NULL) { if (c->connecting_result.transport != NULL) {
publish_transport(exec_ctx, c); publish_transport(exec_ctx, c);
} else if (c->disconnected) {
/* do nothing */
} else { } else {
gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
gpr_mu_lock(&c->mu); gpr_mu_lock(&c->mu);
GPR_ASSERT(!c->have_alarm); GPR_ASSERT(!c->have_alarm);
c->have_alarm = 1; c->have_alarm = 1;
connectivity_state_changed_locked(exec_ctx, c, "connect_failed"); grpc_connectivity_state_set(exec_ctx, &c->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE, "connect_failed");
grpc_timer_init(exec_ctx, &c->alarm, c->next_attempt, on_alarm, c, now); grpc_timer_init(exec_ctx, &c->alarm, c->next_attempt, on_alarm, c, now);
gpr_mu_unlock(&c->mu); gpr_mu_unlock(&c->mu);
} }
@ -764,29 +566,6 @@ static gpr_timespec compute_connect_deadline(grpc_subchannel *c) {
: min_deadline; : min_deadline;
} }
static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) {
if (c->disconnected) {
return GRPC_CHANNEL_FATAL_FAILURE;
}
if (c->connecting) {
if (c->have_alarm) {
return GRPC_CHANNEL_TRANSIENT_FAILURE;
}
return GRPC_CHANNEL_CONNECTING;
}
if (c->active) {
return GRPC_CHANNEL_READY;
}
return GRPC_CHANNEL_IDLE;
}
static void connectivity_state_changed_locked(grpc_exec_ctx *exec_ctx,
grpc_subchannel *c,
const char *reason) {
grpc_connectivity_state current = compute_connectivity_locked(c);
grpc_connectivity_state_set(exec_ctx, &c->state_tracker, current, reason);
}
/* /*
* grpc_subchannel_call implementation * grpc_subchannel_call implementation
*/ */
@ -794,17 +573,9 @@ static void connectivity_state_changed_locked(grpc_exec_ctx *exec_ctx,
static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call, static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call,
int success) { int success) {
grpc_subchannel_call *c = call; grpc_subchannel_call *c = call;
gpr_mu *mu = &c->connection->subchannel->mu;
grpc_subchannel *destroy;
GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0); GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0);
grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c)); grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c));
gpr_mu_lock(mu);
destroy = CONNECTION_UNREF_LOCKED(exec_ctx, c->connection, "call");
gpr_mu_unlock(mu);
gpr_free(c); gpr_free(c);
if (destroy != NULL) {
subchannel_destroy(exec_ctx, destroy);
}
GPR_TIMER_END("grpc_subchannel_call_unref.destroy", 0); GPR_TIMER_END("grpc_subchannel_call_unref.destroy", 0);
} }
@ -842,8 +613,12 @@ void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx,
top_elem->filter->start_transport_stream_op(exec_ctx, top_elem, op); top_elem->filter->start_transport_stream_op(exec_ctx, top_elem, op);
} }
static grpc_subchannel_call *create_call(grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(grpc_subchannel *c) {
connection *con, return GET_CONNECTED_SUBCHANNEL(c, acq);
}
grpc_subchannel_call *grpc_connected_subchannel_create_call(grpc_exec_ctx *exec_ctx,
grpc_connected_subchannel *con,
grpc_pollset *pollset) { grpc_pollset *pollset) {
grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con); grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
grpc_subchannel_call *call = grpc_subchannel_call *call =

51
src/core/client_config/subchannel.h
Unescape View File

@ -41,6 +41,7 @@
/** A (sub-)channel that knows how to connect to exactly one target /** A (sub-)channel that knows how to connect to exactly one target
address. Provides a target for load balancing. */ address. Provides a target for load balancing. */
typedef struct grpc_subchannel grpc_subchannel; 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_call grpc_subchannel_call;
typedef struct grpc_subchannel_args grpc_subchannel_args; typedef struct grpc_subchannel_args grpc_subchannel_args;
@ -49,6 +50,10 @@ typedef struct grpc_subchannel_args grpc_subchannel_args;
grpc_subchannel_ref((p), __FILE__, __LINE__, (r)) grpc_subchannel_ref((p), __FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_UNREF(cl, p, r) \ #define GRPC_SUBCHANNEL_UNREF(cl, p, r) \
grpc_subchannel_unref((cl), (p), __FILE__, __LINE__, (r)) grpc_subchannel_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) \ #define GRPC_SUBCHANNEL_CALL_REF(p, r) \
grpc_subchannel_call_ref((p), __FILE__, __LINE__, (r)) grpc_subchannel_call_ref((p), __FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \ #define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \
@ -58,6 +63,8 @@ typedef struct grpc_subchannel_args grpc_subchannel_args;
#else #else
#define GRPC_SUBCHANNEL_REF(p, r) grpc_subchannel_ref((p)) #define GRPC_SUBCHANNEL_REF(p, r) grpc_subchannel_ref((p))
#define GRPC_SUBCHANNEL_UNREF(cl, p, r) grpc_subchannel_unref((cl), (p)) #define GRPC_SUBCHANNEL_UNREF(cl, p, r) grpc_subchannel_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_REF(p, r) grpc_subchannel_call_ref((p))
#define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \ #define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \
grpc_subchannel_call_unref((cl), (p)) grpc_subchannel_call_unref((cl), (p))
@ -69,33 +76,29 @@ void grpc_subchannel_ref(grpc_subchannel *channel
void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx, void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
grpc_subchannel *channel grpc_subchannel *channel
GRPC_SUBCHANNEL_REF_EXTRA_ARGS); 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 void grpc_subchannel_call_ref(grpc_subchannel_call *call
GRPC_SUBCHANNEL_REF_EXTRA_ARGS); GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx, void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
grpc_subchannel_call *call grpc_subchannel_call *call
GRPC_SUBCHANNEL_REF_EXTRA_ARGS); GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
/** construct a subchannel call (possibly asynchronously). /** construct a subchannel call */
* grpc_subchannel_call *grpc_connected_subchannel_create_call(grpc_exec_ctx *exec_ctx,
* If the returned status is 1, the call will return immediately and \a target grpc_connected_subchannel *connected_subchannel,
* will point to a connected \a subchannel_call instance. Note that \a notify grpc_pollset *pollset);
* will \em not be invoked in this case.
* Otherwise, if the returned status is 0, the subchannel call will be created
* asynchronously, invoking the \a notify callback upon completion. */
int grpc_subchannel_create_call(grpc_exec_ctx *exec_ctx,
grpc_subchannel *subchannel,
grpc_pollset *pollset, gpr_atm *target,
grpc_closure *notify);
/** cancel \a call in the waiting state. */
void grpc_subchannel_cancel_create_call(grpc_exec_ctx *exec_ctx,
grpc_subchannel *subchannel,
gpr_atm *target);
/** process a transport level op */ /** process a transport level op */
void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx, void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
grpc_subchannel *subchannel, grpc_subchannel *subchannel,
grpc_transport_op *op); grpc_transport_op *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 */ /** poll the current connectivity state of a channel */
grpc_connectivity_state grpc_subchannel_check_connectivity( grpc_connectivity_state grpc_subchannel_check_connectivity(
@ -107,13 +110,19 @@ void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
grpc_subchannel *channel, grpc_subchannel *channel,
grpc_connectivity_state *state, grpc_connectivity_state *state,
grpc_closure *notify); grpc_closure *notify);
void grpc_connected_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
grpc_connected_subchannel *channel,
grpc_connectivity_state *state,
grpc_closure *notify);
/** Remove \a subscribed_notify from the list of closures to be called on a /** Remove \a subscribed_notify from the list of closures to be called on a
* state change if present, returning 1. Otherwise, nothing is done and return * state change if present. */
* 0. */ void grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
int grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
grpc_subchannel *channel, grpc_subchannel *channel,
grpc_closure *subscribed_notify); grpc_closure *subscribed_notify);
void grpc_connected_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
grpc_connected_subchannel *channel,
grpc_closure *subscribed_notify);
/** express interest in \a channel's activities through \a pollset. */ /** express interest in \a channel's activities through \a pollset. */
void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx, void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx,
@ -124,6 +133,10 @@ void grpc_subchannel_del_interested_party(grpc_exec_ctx *exec_ctx,
grpc_subchannel *channel, grpc_subchannel *channel,
grpc_pollset *pollset); grpc_pollset *pollset);
/** 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 */ /** continue processing a transport op */
void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx, void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx,
grpc_subchannel_call *subchannel_call, grpc_subchannel_call *subchannel_call,

2
src/core/transport/transport.h
Unescape View File

@ -96,7 +96,7 @@ typedef struct grpc_transport_stream_op {
typedef struct grpc_transport_op { typedef struct grpc_transport_op {
/** called when processing of this op is done */ /** called when processing of this op is done */
grpc_closure *on_consumed; grpc_closure *on_consumed;
/** connectivity monitoring */ /** connectivity monitoring - set connectivity_state to NULL to unsubscribe */
grpc_closure *on_connectivity_state_change; grpc_closure *on_connectivity_state_change;
grpc_connectivity_state *connectivity_state; grpc_connectivity_state *connectivity_state;
/** should the transport be disconnected */ /** should the transport be disconnected */

Write Preview
Loading…
Cancel
Save