aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/client_config
diff options
context:
space:
mode:
authorGravatar Craig Tiller <craig.tiller@gmail.com>2015-12-07 15:14:14 -0800
committerGravatar Craig Tiller <craig.tiller@gmail.com>2015-12-07 15:14:14 -0800
commit87b71e2b55d0b56b4c4778ee842224fcd8f46744 (patch)
treedacbf03e8f819f5a272a024a0ca29a22fb2b0db7 /src/core/client_config
parent26dab31f6ec8c6b6ede97aa00eaaa4b68d6363a7 (diff)
parentf62c4d5a988f37d812373580137ff69a77305102 (diff)
Merge branch 'connected-subchannel' into ping-ping-ping-ping-ping-ping-ping-ping-ping
Diffstat (limited to 'src/core/client_config')
-rw-r--r--src/core/client_config/lb_policies/pick_first.c151
-rw-r--r--src/core/client_config/lb_policies/round_robin.c240
-rw-r--r--src/core/client_config/lb_policy.c78
-rw-r--r--src/core/client_config/lb_policy.h35
-rw-r--r--src/core/client_config/resolver.c7
-rw-r--r--src/core/client_config/resolver.h10
-rw-r--r--src/core/client_config/resolvers/dns_resolver.c7
-rw-r--r--src/core/client_config/resolvers/sockaddr_resolver.c13
-rw-r--r--src/core/client_config/resolvers/zookeeper_resolver.c7
-rw-r--r--src/core/client_config/subchannel.c726
-rw-r--r--src/core/client_config/subchannel.h89
11 files changed, 545 insertions, 818 deletions
diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c
index c007e4c584..8a8b60ba39 100644
--- a/src/core/client_config/lb_policies/pick_first.c
+++ b/src/core/client_config/lb_policies/pick_first.c
@@ -42,7 +42,7 @@
typedef struct pending_pick {
struct pending_pick *next;
grpc_pollset *pollset;
- grpc_subchannel **target;
+ grpc_connected_subchannel **target;
grpc_closure *on_complete;
} pending_pick;
@@ -60,7 +60,7 @@ typedef struct {
/** the selected channel
TODO(ctiller): this should be atomically set so we don't
need to take a mutex in the common case */
- grpc_subchannel *selected;
+ grpc_connected_subchannel *selected;
/** have we started picking? */
int started_picking;
/** are we shut down? */
@@ -76,24 +76,6 @@ typedef struct {
grpc_connectivity_state_tracker state_tracker;
} pick_first_lb_policy;
-static void del_interested_parties_locked(grpc_exec_ctx *exec_ctx,
- pick_first_lb_policy *p) {
- pending_pick *pp;
- for (pp = p->pending_picks; pp; pp = pp->next) {
- grpc_subchannel_del_interested_party(
- exec_ctx, p->subchannels[p->checking_subchannel], pp->pollset);
- }
-}
-
-static void add_interested_parties_locked(grpc_exec_ctx *exec_ctx,
- pick_first_lb_policy *p) {
- pending_pick *pp;
- for (pp = p->pending_picks; pp; pp = pp->next) {
- grpc_subchannel_add_interested_party(
- exec_ctx, p->subchannels[p->checking_subchannel], pp->pollset);
- }
-}
-
void pf_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
size_t i;
@@ -102,7 +84,7 @@ void pf_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[i], "pick_first");
}
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);
gpr_free(p->subchannels);
@@ -114,16 +96,26 @@ void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
pending_pick *pp;
gpr_mu_lock(&p->mu);
- del_interested_parties_locked(exec_ctx, p);
p->shutdown = 1;
pp = p->pending_picks;
p->pending_picks = NULL;
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_FATAL_FAILURE, "shutdown");
+ /* cancel subscription */
+ if (p->selected != NULL) {
+ grpc_connected_subchannel_notify_on_state_change(
+ exec_ctx, p->selected, NULL, NULL, &p->connectivity_changed);
+ } else {
+ grpc_subchannel_notify_on_state_change(
+ exec_ctx, p->subchannels[p->checking_subchannel], NULL, NULL,
+ &p->connectivity_changed);
+ }
gpr_mu_unlock(&p->mu);
while (pp != NULL) {
pending_pick *next = pp->next;
*pp->target = NULL;
+ grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+ pp->pollset);
grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
gpr_free(pp);
pp = next;
@@ -131,7 +123,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,
- grpc_subchannel **target) {
+ grpc_connected_subchannel **target) {
pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
pending_pick *pp;
gpr_mu_lock(&p->mu);
@@ -140,8 +132,8 @@ static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
while (pp != NULL) {
pending_pick *next = pp->next;
if (pp->target == target) {
- grpc_subchannel_del_interested_party(
- exec_ctx, p->subchannels[p->checking_subchannel], pp->pollset);
+ grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+ pp->pollset);
*target = NULL;
grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 0);
gpr_free(pp);
@@ -158,10 +150,11 @@ static void start_picking(grpc_exec_ctx *exec_ctx, pick_first_lb_policy *p) {
p->started_picking = 1;
p->checking_subchannel = 0;
p->checking_connectivity = GRPC_CHANNEL_IDLE;
- GRPC_LB_POLICY_REF(&p->base, "pick_first_connectivity");
+ GRPC_LB_POLICY_WEAK_REF(&p->base, "pick_first_connectivity");
grpc_subchannel_notify_on_state_change(
exec_ctx, p->subchannels[p->checking_subchannel],
- &p->checking_connectivity, &p->connectivity_changed);
+ &p->base.interested_parties, &p->checking_connectivity,
+ &p->connectivity_changed);
}
void pf_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
@@ -174,8 +167,8 @@ 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,
- grpc_metadata_batch *initial_metadata, grpc_subchannel **target,
- grpc_closure *on_complete) {
+ grpc_metadata_batch *initial_metadata,
+ grpc_connected_subchannel **target, grpc_closure *on_complete) {
pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
pending_pick *pp;
gpr_mu_lock(&p->mu);
@@ -187,8 +180,8 @@ int pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
if (!p->started_picking) {
start_picking(exec_ctx, p);
}
- grpc_subchannel_add_interested_party(
- exec_ctx, p->subchannels[p->checking_subchannel], pollset);
+ grpc_pollset_set_add_pollset(exec_ctx, &p->base.interested_parties,
+ pollset);
pp = gpr_malloc(sizeof(*pp));
pp->next = p->pending_picks;
pp->pollset = pollset;
@@ -204,25 +197,17 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
int iomgr_success) {
pick_first_lb_policy *p = arg;
size_t i;
- grpc_transport_op op;
size_t num_subchannels = p->num_subchannels;
grpc_subchannel **subchannels;
- grpc_subchannel *exclude_subchannel;
gpr_mu_lock(&p->mu);
subchannels = p->subchannels;
p->num_subchannels = 0;
p->subchannels = NULL;
- exclude_subchannel = p->selected;
gpr_mu_unlock(&p->mu);
- GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "destroy_subchannels");
+ GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "destroy_subchannels");
for (i = 0; i < num_subchannels; i++) {
- if (subchannels[i] != exclude_subchannel) {
- memset(&op, 0, sizeof(op));
- op.disconnect = 1;
- grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], &op);
- }
GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannels[i], "pick_first");
}
@@ -232,23 +217,28 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
int iomgr_success) {
pick_first_lb_policy *p = arg;
+ grpc_subchannel *selected_subchannel;
pending_pick *pp;
gpr_mu_lock(&p->mu);
if (p->shutdown) {
gpr_mu_unlock(&p->mu);
- GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
+ GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
return;
} else if (p->selected != NULL) {
+ if (p->checking_connectivity == GRPC_CHANNEL_TRANSIENT_FAILURE) {
+ /* if the selected channel goes bad, we're done */
+ p->checking_connectivity = GRPC_CHANNEL_FATAL_FAILURE;
+ }
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
p->checking_connectivity, "selected_changed");
if (p->checking_connectivity != GRPC_CHANNEL_FATAL_FAILURE) {
- grpc_subchannel_notify_on_state_change(exec_ctx, p->selected,
- &p->checking_connectivity,
- &p->connectivity_changed);
+ grpc_connected_subchannel_notify_on_state_change(
+ exec_ctx, p->selected, &p->base.interested_parties,
+ &p->checking_connectivity, &p->connectivity_changed);
} else {
- GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
+ GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
}
} else {
loop:
@@ -256,39 +246,41 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
case GRPC_CHANNEL_READY:
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_READY, "connecting_ready");
- p->selected = p->subchannels[p->checking_subchannel];
- GRPC_SUBCHANNEL_REF(p->selected, "picked_first");
+ selected_subchannel = p->subchannels[p->checking_subchannel];
+ 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 */
- GRPC_LB_POLICY_REF(&p->base, "destroy_subchannels");
+ GRPC_LB_POLICY_WEAK_REF(&p->base, "destroy_subchannels");
grpc_exec_ctx_enqueue(exec_ctx,
grpc_closure_create(destroy_subchannels, p), 1);
/* update any calls that were waiting for a pick */
while ((pp = p->pending_picks)) {
p->pending_picks = pp->next;
*pp->target = p->selected;
- grpc_subchannel_del_interested_party(exec_ctx, p->selected,
- pp->pollset);
+ grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+ pp->pollset);
grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
gpr_free(pp);
}
- grpc_subchannel_notify_on_state_change(exec_ctx, p->selected,
- &p->checking_connectivity,
- &p->connectivity_changed);
+ grpc_connected_subchannel_notify_on_state_change(
+ exec_ctx, p->selected, &p->base.interested_parties,
+ &p->checking_connectivity, &p->connectivity_changed);
break;
case GRPC_CHANNEL_TRANSIENT_FAILURE:
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_TRANSIENT_FAILURE,
"connecting_transient_failure");
- del_interested_parties_locked(exec_ctx, p);
p->checking_subchannel =
(p->checking_subchannel + 1) % p->num_subchannels;
p->checking_connectivity = grpc_subchannel_check_connectivity(
p->subchannels[p->checking_subchannel]);
- add_interested_parties_locked(exec_ctx, p);
if (p->checking_connectivity == GRPC_CHANNEL_TRANSIENT_FAILURE) {
grpc_subchannel_notify_on_state_change(
exec_ctx, p->subchannels[p->checking_subchannel],
- &p->checking_connectivity, &p->connectivity_changed);
+ &p->base.interested_parties, &p->checking_connectivity,
+ &p->connectivity_changed);
} else {
goto loop;
}
@@ -300,13 +292,13 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
"connecting_changed");
grpc_subchannel_notify_on_state_change(
exec_ctx, p->subchannels[p->checking_subchannel],
- &p->checking_connectivity, &p->connectivity_changed);
+ &p->base.interested_parties, &p->checking_connectivity,
+ &p->connectivity_changed);
break;
case GRPC_CHANNEL_FATAL_FAILURE:
- del_interested_parties_locked(exec_ctx, p);
- GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel],
- p->subchannels[p->num_subchannels - 1]);
p->num_subchannels--;
+ GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel],
+ p->subchannels[p->num_subchannels]);
GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[p->num_subchannels],
"pick_first");
if (p->num_subchannels == 0) {
@@ -319,7 +311,8 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
gpr_free(pp);
}
- GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "pick_first_connectivity");
+ GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base,
+ "pick_first_connectivity");
} else {
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_TRANSIENT_FAILURE,
@@ -327,7 +320,6 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
p->checking_subchannel %= p->num_subchannels;
p->checking_connectivity = grpc_subchannel_check_connectivity(
p->subchannels[p->checking_subchannel]);
- add_interested_parties_locked(exec_ctx, p);
goto loop;
}
}
@@ -336,39 +328,6 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
gpr_mu_unlock(&p->mu);
}
-static void pf_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
- grpc_transport_op *op) {
- pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
- size_t i;
- size_t n;
- grpc_subchannel **subchannels;
- grpc_subchannel *selected;
-
- gpr_mu_lock(&p->mu);
- n = p->num_subchannels;
- subchannels = gpr_malloc(n * sizeof(*subchannels));
- selected = p->selected;
- if (selected) {
- GRPC_SUBCHANNEL_REF(selected, "pf_broadcast_to_selected");
- }
- for (i = 0; i < n; i++) {
- subchannels[i] = p->subchannels[i];
- GRPC_SUBCHANNEL_REF(subchannels[i], "pf_broadcast");
- }
- gpr_mu_unlock(&p->mu);
-
- for (i = 0; i < n; i++) {
- if (selected == subchannels[i]) continue;
- grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], op);
- GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannels[i], "pf_broadcast");
- }
- if (p->selected) {
- grpc_subchannel_process_transport_op(exec_ctx, selected, op);
- GRPC_SUBCHANNEL_UNREF(exec_ctx, selected, "pf_broadcast_to_selected");
- }
- gpr_free(subchannels);
-}
-
static grpc_connectivity_state pf_check_connectivity(grpc_exec_ctx *exec_ctx,
grpc_lb_policy *pol) {
pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
@@ -391,7 +350,7 @@ void pf_notify_on_state_change(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
static const grpc_lb_policy_vtable pick_first_lb_policy_vtable = {
pf_destroy, pf_shutdown, pf_pick, pf_cancel_pick, pf_ping_one, pf_exit_idle,
- pf_broadcast, pf_check_connectivity, pf_notify_on_state_change};
+ pf_check_connectivity, pf_notify_on_state_change};
static void pick_first_factory_ref(grpc_lb_policy_factory *factory) {}
diff --git a/src/core/client_config/lb_policies/round_robin.c b/src/core/client_config/lb_policies/round_robin.c
index 1ffe32fff2..b86dba20ee 100644
--- a/src/core/client_config/lb_policies/round_robin.c
+++ b/src/core/client_config/lb_policies/round_robin.c
@@ -38,6 +38,8 @@
#include <grpc/support/alloc.h>
#include "src/core/transport/connectivity_state.h"
+typedef struct round_robin_lb_policy round_robin_lb_policy;
+
int grpc_lb_round_robin_trace = 0;
/** List of entities waiting for a pick.
@@ -46,7 +48,7 @@ int grpc_lb_round_robin_trace = 0;
typedef struct pending_pick {
struct pending_pick *next;
grpc_pollset *pollset;
- grpc_subchannel **target;
+ grpc_connected_subchannel **target;
grpc_closure *on_complete;
} pending_pick;
@@ -58,22 +60,27 @@ typedef struct ready_list {
} ready_list;
typedef struct {
- size_t subchannel_idx; /**< Index over p->subchannels */
- void *p; /**< round_robin_lb_policy instance */
-} connectivity_changed_cb_arg;
-
-typedef struct {
+ /** index within policy->subchannels */
+ size_t index;
+ /** backpointer to owning policy */
+ round_robin_lb_policy *policy;
+ /** subchannel itself */
+ grpc_subchannel *subchannel;
+ /** notification that connectivity has changed on subchannel */
+ grpc_closure connectivity_changed_closure;
+ /** this subchannels current position in subchannel->ready_list */
+ ready_list *ready_list_node;
+ /** last observed connectivity */
+ grpc_connectivity_state connectivity_state;
+} subchannel_data;
+
+struct round_robin_lb_policy {
/** base policy: must be first */
grpc_lb_policy base;
/** all our subchannels */
- grpc_subchannel **subchannels;
size_t num_subchannels;
-
- /** Callbacks, one per subchannel being watched, to be called when their
- * respective connectivity changes */
- grpc_closure *connectivity_changed_cbs;
- connectivity_changed_cb_arg *cb_args;
+ subchannel_data **subchannels;
/** mutex protecting remaining members */
gpr_mu mu;
@@ -81,8 +88,6 @@ typedef struct {
int started_picking;
/** are we shutting down? */
int shutdown;
- /** Connectivity state of the subchannels being watched */
- grpc_connectivity_state *subchannel_connectivity;
/** List of picks that are waiting on connectivity */
pending_pick *pending_picks;
@@ -93,13 +98,7 @@ typedef struct {
ready_list ready_list;
/** Last pick from the ready list. */
ready_list *ready_list_last_pick;
-
- /** Subchannel index to ready_list node.
- *
- * Kept in order to remove nodes from the ready list associated with a
- * subchannel */
- ready_list **subchannel_index_to_readylist_node;
-} round_robin_lb_policy;
+};
/** Returns the next subchannel from the connected list or NULL if the list is
* empty.
@@ -144,9 +143,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
* csc to the list of ready subchannels. */
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));
- new_elem->subchannel = csc;
+ new_elem->subchannel = sc;
if (p->ready_list.prev == NULL) {
/* first element */
new_elem->next = &p->ready_list;
@@ -160,7 +159,7 @@ static ready_list *add_connected_sc_locked(round_robin_lb_policy *p,
p->ready_list.prev = new_elem;
}
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;
}
@@ -200,28 +199,15 @@ static void remove_disconnected_sc_locked(round_robin_lb_policy *p,
gpr_free(node);
}
-static void del_interested_parties_locked(grpc_exec_ctx *exec_ctx,
- round_robin_lb_policy *p,
- const size_t subchannel_idx) {
- pending_pick *pp;
- for (pp = p->pending_picks; pp; pp = pp->next) {
- grpc_subchannel_del_interested_party(
- exec_ctx, p->subchannels[subchannel_idx], pp->pollset);
- }
-}
-
void rr_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
size_t i;
ready_list *elem;
for (i = 0; i < p->num_subchannels; i++) {
- del_interested_parties_locked(exec_ctx, p, i);
+ subchannel_data *sd = p->subchannels[i];
+ GRPC_SUBCHANNEL_UNREF(exec_ctx, sd->subchannel, "round_robin");
+ gpr_free(sd);
}
- for (i = 0; i < p->num_subchannels; i++) {
- GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[i], "round_robin");
- }
- gpr_free(p->connectivity_changed_cbs);
- gpr_free(p->subchannel_connectivity);
grpc_connectivity_state_destroy(exec_ctx, &p->state_tracker);
gpr_free(p->subchannels);
@@ -237,20 +223,15 @@ void rr_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
gpr_free(elem);
elem = tmp;
}
- gpr_free(p->subchannel_index_to_readylist_node);
- gpr_free(p->cb_args);
gpr_free(p);
}
void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
- size_t i;
round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
pending_pick *pp;
- gpr_mu_lock(&p->mu);
+ size_t i;
- for (i = 0; i < p->num_subchannels; i++) {
- del_interested_parties_locked(exec_ctx, p, i);
- }
+ gpr_mu_lock(&p->mu);
p->shutdown = 1;
while ((pp = p->pending_picks)) {
@@ -261,24 +242,26 @@ void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
}
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_FATAL_FAILURE, "shutdown");
+ for (i = 0; i < p->num_subchannels; i++) {
+ subchannel_data *sd = p->subchannels[i];
+ grpc_subchannel_notify_on_state_change(exec_ctx, sd->subchannel, NULL, NULL,
+ &sd->connectivity_changed_closure);
+ }
gpr_mu_unlock(&p->mu);
}
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;
pending_pick *pp;
- size_t i;
gpr_mu_lock(&p->mu);
pp = p->pending_picks;
p->pending_picks = NULL;
while (pp != NULL) {
pending_pick *next = pp->next;
if (pp->target == target) {
- for (i = 0; i < p->num_subchannels; i++) {
- grpc_subchannel_add_interested_party(exec_ctx, p->subchannels[i],
- pp->pollset);
- }
+ grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+ pp->pollset);
*target = NULL;
grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 0);
gpr_free(pp);
@@ -295,12 +278,16 @@ static void start_picking(grpc_exec_ctx *exec_ctx, round_robin_lb_policy *p) {
size_t i;
p->started_picking = 1;
+ gpr_log(GPR_DEBUG, "LB_POLICY: p=%p num_subchannels=%d", p,
+ p->num_subchannels);
+
for (i = 0; i < p->num_subchannels; i++) {
- p->subchannel_connectivity[i] = GRPC_CHANNEL_IDLE;
- grpc_subchannel_notify_on_state_change(exec_ctx, p->subchannels[i],
- &p->subchannel_connectivity[i],
- &p->connectivity_changed_cbs[i]);
- GRPC_LB_POLICY_REF(&p->base, "round_robin_connectivity");
+ subchannel_data *sd = p->subchannels[i];
+ sd->connectivity_state = GRPC_CHANNEL_IDLE;
+ grpc_subchannel_notify_on_state_change(
+ exec_ctx, sd->subchannel, &p->base.interested_parties,
+ &sd->connectivity_state, &sd->connectivity_changed_closure);
+ GRPC_LB_POLICY_WEAK_REF(&p->base, "round_robin_connectivity");
}
}
@@ -314,18 +301,18 @@ 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,
- grpc_metadata_batch *initial_metadata, grpc_subchannel **target,
- grpc_closure *on_complete) {
- size_t i;
+ grpc_metadata_batch *initial_metadata,
+ grpc_connected_subchannel **target, grpc_closure *on_complete) {
round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
pending_pick *pp;
ready_list *selected;
gpr_mu_lock(&p->mu);
if ((selected = peek_next_connected_locked(p))) {
gpr_mu_unlock(&p->mu);
- *target = selected->subchannel;
+ *target = grpc_subchannel_get_connected_subchannel(selected->subchannel);
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);
}
/* only advance the last picked pointer if the selection was used */
@@ -335,10 +322,8 @@ int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
if (!p->started_picking) {
start_picking(exec_ctx, p);
}
- for (i = 0; i < p->num_subchannels; i++) {
- grpc_subchannel_add_interested_party(exec_ctx, p->subchannels[i],
- pollset);
- }
+ grpc_pollset_set_add_pollset(exec_ctx, &p->base.interested_parties,
+ pollset);
pp = gpr_malloc(sizeof(*pp));
pp->next = p->pending_picks;
pp->pollset = pollset;
@@ -352,33 +337,25 @@ int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
int iomgr_success) {
- connectivity_changed_cb_arg *cb_arg = arg;
- round_robin_lb_policy *p = cb_arg->p;
- /* index over p->subchannels of this cb's subchannel */
- const size_t this_idx = cb_arg->subchannel_idx;
+ subchannel_data *sd = arg;
+ round_robin_lb_policy *p = sd->policy;
pending_pick *pp;
ready_list *selected;
int unref = 0;
- /* connectivity state of this cb's subchannel */
- grpc_connectivity_state *this_connectivity;
-
gpr_mu_lock(&p->mu);
- this_connectivity = &p->subchannel_connectivity[this_idx];
-
if (p->shutdown) {
unref = 1;
} else {
- switch (*this_connectivity) {
+ switch (sd->connectivity_state) {
case GRPC_CHANNEL_READY:
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_READY, "connecting_ready");
/* add the newly connected subchannel to the list of connected ones.
* Note that it goes to the "end of the line". */
- p->subchannel_index_to_readylist_node[this_idx] =
- add_connected_sc_locked(p, p->subchannels[this_idx]);
+ sd->ready_list_node = add_connected_sc_locked(p, sd->subchannel);
/* at this point we know there's at least one suitable subchannel. Go
* ahead and pick one and notify the pending suitors in
* p->pending_picks. This preemtively replicates rr_pick()'s actions. */
@@ -390,60 +367,60 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
}
while ((pp = p->pending_picks)) {
p->pending_picks = pp->next;
- *pp->target = selected->subchannel;
+ *pp->target =
+ grpc_subchannel_get_connected_subchannel(selected->subchannel);
if (grpc_lb_round_robin_trace) {
gpr_log(GPR_DEBUG,
"[RR CONN CHANGED] TARGET <-- SUBCHANNEL %p (NODE %p)",
selected->subchannel, selected);
}
- grpc_subchannel_del_interested_party(exec_ctx, selected->subchannel,
- pp->pollset);
+ grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
+ pp->pollset);
grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
gpr_free(pp);
}
grpc_subchannel_notify_on_state_change(
- exec_ctx, p->subchannels[this_idx], this_connectivity,
- &p->connectivity_changed_cbs[this_idx]);
+ exec_ctx, sd->subchannel, &p->base.interested_parties,
+ &sd->connectivity_state, &sd->connectivity_changed_closure);
break;
case GRPC_CHANNEL_CONNECTING:
case GRPC_CHANNEL_IDLE:
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
- *this_connectivity, "connecting_changed");
+ sd->connectivity_state,
+ "connecting_changed");
grpc_subchannel_notify_on_state_change(
- exec_ctx, p->subchannels[this_idx], this_connectivity,
- &p->connectivity_changed_cbs[this_idx]);
+ exec_ctx, sd->subchannel, &p->base.interested_parties,
+ &sd->connectivity_state, &sd->connectivity_changed_closure);
break;
case GRPC_CHANNEL_TRANSIENT_FAILURE:
- del_interested_parties_locked(exec_ctx, p, this_idx);
/* renew state notification */
grpc_subchannel_notify_on_state_change(
- exec_ctx, p->subchannels[this_idx], this_connectivity,
- &p->connectivity_changed_cbs[this_idx]);
+ exec_ctx, sd->subchannel, &p->base.interested_parties,
+ &sd->connectivity_state, &sd->connectivity_changed_closure);
/* remove from ready list if still present */
- if (p->subchannel_index_to_readylist_node[this_idx] != NULL) {
- remove_disconnected_sc_locked(
- p, p->subchannel_index_to_readylist_node[this_idx]);
- p->subchannel_index_to_readylist_node[this_idx] = NULL;
+ if (sd->ready_list_node != NULL) {
+ remove_disconnected_sc_locked(p, sd->ready_list_node);
+ sd->ready_list_node = NULL;
}
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_TRANSIENT_FAILURE,
"connecting_transient_failure");
break;
case GRPC_CHANNEL_FATAL_FAILURE:
- del_interested_parties_locked(exec_ctx, p, this_idx);
- if (p->subchannel_index_to_readylist_node[this_idx] != NULL) {
- remove_disconnected_sc_locked(
- p, p->subchannel_index_to_readylist_node[this_idx]);
- p->subchannel_index_to_readylist_node[this_idx] = NULL;
+ if (sd->ready_list_node != NULL) {
+ remove_disconnected_sc_locked(p, sd->ready_list_node);
+ sd->ready_list_node = NULL;
}
- GPR_SWAP(grpc_subchannel *, p->subchannels[this_idx],
- p->subchannels[p->num_subchannels - 1]);
p->num_subchannels--;
- GRPC_SUBCHANNEL_UNREF(exec_ctx, p->subchannels[p->num_subchannels],
- "round_robin");
+ GPR_SWAP(subchannel_data *, p->subchannels[sd->index],
+ p->subchannels[p->num_subchannels]);
+ GRPC_SUBCHANNEL_UNREF(exec_ctx, sd->subchannel, "round_robin");
+ p->subchannels[sd->index]->index = sd->index;
+ gpr_free(sd);
+ unref = 1;
if (p->num_subchannels == 0) {
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_FATAL_FAILURE,
@@ -454,7 +431,6 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
gpr_free(pp);
}
- unref = 1;
} else {
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
GRPC_CHANNEL_TRANSIENT_FAILURE,
@@ -466,31 +442,8 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
gpr_mu_unlock(&p->mu);
if (unref) {
- GRPC_LB_POLICY_UNREF(exec_ctx, &p->base, "round_robin_connectivity");
- }
-}
-
-static void rr_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
- grpc_transport_op *op) {
- round_robin_lb_policy *p = (round_robin_lb_policy *)pol;
- size_t i;
- size_t n;
- grpc_subchannel **subchannels;
-
- gpr_mu_lock(&p->mu);
- n = p->num_subchannels;
- subchannels = gpr_malloc(n * sizeof(*subchannels));
- for (i = 0; i < n; i++) {
- subchannels[i] = p->subchannels[i];
- GRPC_SUBCHANNEL_REF(subchannels[i], "rr_broadcast");
+ GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base, "round_robin_connectivity");
}
- gpr_mu_unlock(&p->mu);
-
- for (i = 0; i < n; i++) {
- grpc_subchannel_process_transport_op(exec_ctx, subchannels[i], op);
- GRPC_SUBCHANNEL_UNREF(exec_ctx, subchannels[i], "rr_broadcast");
- }
- gpr_free(subchannels);
}
static grpc_connectivity_state rr_check_connectivity(grpc_exec_ctx *exec_ctx,
@@ -516,7 +469,7 @@ static void rr_notify_on_state_change(grpc_exec_ctx *exec_ctx,
static const grpc_lb_policy_vtable round_robin_lb_policy_vtable = {
rr_destroy, rr_shutdown, rr_pick, rr_cancel_pick, rr_exit_idle,
- rr_broadcast, rr_check_connectivity, rr_notify_on_state_change};
+ rr_check_connectivity, rr_notify_on_state_change};
static void round_robin_factory_ref(grpc_lb_policy_factory *factory) {}
@@ -529,27 +482,22 @@ static grpc_lb_policy *create_round_robin(grpc_lb_policy_factory *factory,
GPR_ASSERT(args->num_subchannels > 0);
memset(p, 0, sizeof(*p));
grpc_lb_policy_init(&p->base, &round_robin_lb_policy_vtable);
- p->subchannels =
- gpr_malloc(sizeof(grpc_subchannel *) * args->num_subchannels);
p->num_subchannels = args->num_subchannels;
+ p->subchannels = gpr_malloc(sizeof(*p->subchannels) * p->num_subchannels);
+ memset(p->subchannels, 0, sizeof(*p->subchannels) * p->num_subchannels);
grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE,
"round_robin");
- memcpy(p->subchannels, args->subchannels,
- sizeof(grpc_subchannel *) * args->num_subchannels);
gpr_mu_init(&p->mu);
- p->connectivity_changed_cbs =
- gpr_malloc(sizeof(grpc_closure) * args->num_subchannels);
- p->subchannel_connectivity =
- gpr_malloc(sizeof(grpc_connectivity_state) * args->num_subchannels);
-
- p->cb_args =
- gpr_malloc(sizeof(connectivity_changed_cb_arg) * args->num_subchannels);
for (i = 0; i < args->num_subchannels; i++) {
- p->cb_args[i].subchannel_idx = i;
- p->cb_args[i].p = p;
- grpc_closure_init(&p->connectivity_changed_cbs[i], rr_connectivity_changed,
- &p->cb_args[i]);
+ subchannel_data *sd = gpr_malloc(sizeof(*sd));
+ memset(sd, 0, sizeof(*sd));
+ p->subchannels[i] = sd;
+ sd->policy = p;
+ sd->index = i;
+ sd->subchannel = args->subchannels[i];
+ grpc_closure_init(&sd->connectivity_changed_closure,
+ rr_connectivity_changed, sd);
}
/* The (dummy node) root of the ready list */
@@ -558,10 +506,6 @@ static grpc_lb_policy *create_round_robin(grpc_lb_policy_factory *factory,
p->ready_list.next = NULL;
p->ready_list_last_pick = &p->ready_list;
- p->subchannel_index_to_readylist_node =
- gpr_malloc(sizeof(grpc_subchannel *) * args->num_subchannels);
- memset(p->subchannel_index_to_readylist_node, 0,
- sizeof(grpc_subchannel *) * args->num_subchannels);
return &p->base;
}
diff --git a/src/core/client_config/lb_policy.c b/src/core/client_config/lb_policy.c
index 36a2454309..d254161546 100644
--- a/src/core/client_config/lb_policy.c
+++ b/src/core/client_config/lb_policy.c
@@ -33,59 +33,85 @@
#include "src/core/client_config/lb_policy.h"
+#define WEAK_REF_BITS 16
+
void grpc_lb_policy_init(grpc_lb_policy *policy,
const grpc_lb_policy_vtable *vtable) {
policy->vtable = vtable;
- gpr_ref_init(&policy->refs, 1);
+ gpr_atm_no_barrier_store(&policy->ref_pair, 1 << WEAK_REF_BITS);
+ grpc_pollset_set_init(&policy->interested_parties);
}
#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG
-void grpc_lb_policy_ref(grpc_lb_policy *policy, const char *file, int line,
- const char *reason) {
- gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "LB_POLICY:%p ref %d -> %d %s",
- policy, (int)policy->refs.count, (int)policy->refs.count + 1, reason);
+#define REF_FUNC_EXTRA_ARGS , const char *file, int line, const char *reason
+#define REF_MUTATE_EXTRA_ARGS REF_FUNC_EXTRA_ARGS, const char *purpose
+#define REF_FUNC_PASS_ARGS(new_reason) , file, line, new_reason
+#define REF_MUTATE_PASS_ARGS(purpose) , file, line, reason, purpose
#else
-void grpc_lb_policy_ref(grpc_lb_policy *policy) {
+#define REF_FUNC_EXTRA_ARGS
+#define REF_MUTATE_EXTRA_ARGS
+#define REF_FUNC_PASS_ARGS(new_reason)
+#define REF_MUTATE_PASS_ARGS(x)
#endif
- gpr_ref(&policy->refs);
-}
+static gpr_atm ref_mutate(grpc_lb_policy *c, gpr_atm delta,
+ int barrier REF_MUTATE_EXTRA_ARGS) {
+ gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, delta)
+ : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta);
#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG
-void grpc_lb_policy_unref(grpc_lb_policy *policy,
- grpc_closure_list *closure_list, const char *file,
- int line, const char *reason) {
- gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "LB_POLICY:%p unref %d -> %d %s",
- policy, (int)policy->refs.count, (int)policy->refs.count - 1, reason);
-#else
-void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
+ gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
+ "LB_POLICY: %p % 12s 0x%08x -> 0x%08x [%s]", c, purpose, old_val,
+ old_val + delta, reason);
#endif
- if (gpr_unref(&policy->refs)) {
- policy->vtable->destroy(exec_ctx, policy);
+ return old_val;
+}
+
+void grpc_lb_policy_ref(grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) {
+ ref_mutate(policy, 1 << WEAK_REF_BITS, 0 REF_MUTATE_PASS_ARGS("STRONG_REF"));
+}
+
+void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx,
+ grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) {
+ gpr_atm old_val =
+ ref_mutate(policy, (gpr_atm)1 - (gpr_atm)(1 << WEAK_REF_BITS),
+ 1 REF_MUTATE_PASS_ARGS("STRONG_UNREF"));
+ gpr_atm mask = ~(gpr_atm)((1 << WEAK_REF_BITS) - 1);
+ gpr_atm check = 1 << WEAK_REF_BITS;
+ if ((old_val & mask) == check) {
+ policy->vtable->shutdown(exec_ctx, policy);
}
+ grpc_lb_policy_weak_unref(exec_ctx,
+ policy REF_FUNC_PASS_ARGS("strong-unref"));
}
-void grpc_lb_policy_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
- policy->vtable->shutdown(exec_ctx, policy);
+void grpc_lb_policy_weak_ref(grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) {
+ ref_mutate(policy, 1, 0 REF_MUTATE_PASS_ARGS("WEAK_REF"));
+}
+
+void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx,
+ grpc_lb_policy *policy REF_FUNC_EXTRA_ARGS) {
+ gpr_atm old_val =
+ ref_mutate(policy, -(gpr_atm)1, 1 REF_MUTATE_PASS_ARGS("WEAK_UNREF"));
+ if (old_val == 1) {
+ grpc_pollset_set_destroy(&policy->interested_parties);
+ policy->vtable->destroy(exec_ctx, policy);
+ }
}
int grpc_lb_policy_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
grpc_pollset *pollset,
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,
target, on_complete);
}
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);
}
-void grpc_lb_policy_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
- grpc_transport_op *op) {
- policy->vtable->broadcast(exec_ctx, policy, op);
-}
-
void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy) {
policy->vtable->exit_idle(exec_ctx, policy);
}
diff --git a/src/core/client_config/lb_policy.h b/src/core/client_config/lb_policy.h
index 894a94e0ec..01c57a24ff 100644
--- a/src/core/client_config/lb_policy.h
+++ b/src/core/client_config/lb_policy.h
@@ -47,7 +47,8 @@ typedef void (*grpc_lb_completion)(void *cb_arg, grpc_subchannel *subchannel,
struct grpc_lb_policy {
const grpc_lb_policy_vtable *vtable;
- gpr_refcount refs;
+ gpr_atm ref_pair;
+ grpc_pollset_set interested_parties;
};
struct grpc_lb_policy_vtable {
@@ -58,19 +59,15 @@ struct grpc_lb_policy_vtable {
/** implement grpc_lb_policy_pick */
int (*pick)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
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,
- grpc_subchannel **target);
+ grpc_connected_subchannel **target);
void (*ping_one)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, grpc_pollset *pollset, grpc_closure *closure);
/** try to enter a READY connectivity state */
void (*exit_idle)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
- /** broadcast a transport op to all subchannels */
- void (*broadcast)(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
- grpc_transport_op *op);
-
/** check the current connectivity of the lb_policy */
grpc_connectivity_state (*check_connectivity)(grpc_exec_ctx *exec_ctx,
grpc_lb_policy *policy);
@@ -83,29 +80,39 @@ struct grpc_lb_policy_vtable {
grpc_closure *closure);
};
+/*#define GRPC_LB_POLICY_REFCOUNT_DEBUG*/
#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG
#define GRPC_LB_POLICY_REF(p, r) \
grpc_lb_policy_ref((p), __FILE__, __LINE__, (r))
#define GRPC_LB_POLICY_UNREF(exec_ctx, p, r) \
grpc_lb_policy_unref((exec_ctx), (p), __FILE__, __LINE__, (r))
+#define GRPC_LB_POLICY_WEAK_REF(p, r) \
+ grpc_lb_policy_weak_ref((p), __FILE__, __LINE__, (r))
+#define GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, p, r) \
+ grpc_lb_policy_weak_unref((exec_ctx), (p), __FILE__, __LINE__, (r))
void grpc_lb_policy_ref(grpc_lb_policy *policy, const char *file, int line,
const char *reason);
void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
const char *file, int line, const char *reason);
+void grpc_lb_policy_weak_ref(grpc_lb_policy *policy, const char *file, int line,
+ const char *reason);
+void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
+ const char *file, int line, const char *reason);
#else
#define GRPC_LB_POLICY_REF(p, r) grpc_lb_policy_ref((p))
#define GRPC_LB_POLICY_UNREF(cl, p, r) grpc_lb_policy_unref((cl), (p))
+#define GRPC_LB_POLICY_WEAK_REF(p, r) grpc_lb_policy_weak_ref((p))
+#define GRPC_LB_POLICY_WEAK_UNREF(cl, p, r) grpc_lb_policy_weak_unref((cl), (p))
void grpc_lb_policy_ref(grpc_lb_policy *policy);
void grpc_lb_policy_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
+void grpc_lb_policy_weak_ref(grpc_lb_policy *policy);
+void grpc_lb_policy_weak_unref(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
#endif
/** called by concrete implementations to initialize the base struct */
void grpc_lb_policy_init(grpc_lb_policy *policy,
const grpc_lb_policy_vtable *vtable);
-/** Start shutting down (fail any pending picks) */
-void grpc_lb_policy_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
-
/** Given initial metadata in \a initial_metadata, find an appropriate
target for this rpc, and 'return' it by calling \a on_complete after setting
\a target.
@@ -113,15 +120,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,
grpc_pollset *pollset,
grpc_metadata_batch *initial_metadata,
- grpc_subchannel **target, grpc_closure *on_complete);
+ grpc_connected_subchannel **target,
+ grpc_closure *on_complete);
void grpc_lb_policy_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, grpc_pollset *pollset, grpc_closure *closure);
void grpc_lb_policy_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
- grpc_subchannel **target);
-
-void grpc_lb_policy_broadcast(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy,
- grpc_transport_op *op);
+ grpc_connected_subchannel **target);
void grpc_lb_policy_exit_idle(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy);
diff --git a/src/core/client_config/resolver.c b/src/core/client_config/resolver.c
index 081097eb19..eda01e72ba 100644
--- a/src/core/client_config/resolver.c
+++ b/src/core/client_config/resolver.c
@@ -71,11 +71,8 @@ void grpc_resolver_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver) {
}
void grpc_resolver_channel_saw_error(grpc_exec_ctx *exec_ctx,
- grpc_resolver *resolver,
- struct sockaddr *failing_address,
- int failing_address_len) {
- resolver->vtable->channel_saw_error(exec_ctx, resolver, failing_address,
- failing_address_len);
+ grpc_resolver *resolver) {
+ resolver->vtable->channel_saw_error(exec_ctx, resolver);
}
void grpc_resolver_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
diff --git a/src/core/client_config/resolver.h b/src/core/client_config/resolver.h
index 7ba0cd5bd4..e612eaf3b3 100644
--- a/src/core/client_config/resolver.h
+++ b/src/core/client_config/resolver.h
@@ -35,8 +35,8 @@
#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_H
#include "src/core/client_config/client_config.h"
+#include "src/core/client_config/subchannel.h"
#include "src/core/iomgr/iomgr.h"
-#include "src/core/iomgr/sockaddr.h"
typedef struct grpc_resolver grpc_resolver;
typedef struct grpc_resolver_vtable grpc_resolver_vtable;
@@ -51,9 +51,7 @@ struct grpc_resolver {
struct grpc_resolver_vtable {
void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver);
void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver);
- void (*channel_saw_error)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
- struct sockaddr *failing_address,
- int failing_address_len);
+ void (*channel_saw_error)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver);
void (*next)(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
grpc_client_config **target_config, grpc_closure *on_complete);
};
@@ -81,9 +79,7 @@ void grpc_resolver_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver);
/** Notification that the channel has seen an error on some address.
Can be used as a hint that re-resolution is desirable soon. */
void grpc_resolver_channel_saw_error(grpc_exec_ctx *exec_ctx,
- grpc_resolver *resolver,
- struct sockaddr *failing_address,
- int failing_address_len);
+ grpc_resolver *resolver);
/** Get the next client config. Called by the channel to fetch a new
configuration. Expected to set *target_config with a new configuration,
diff --git a/src/core/client_config/resolvers/dns_resolver.c b/src/core/client_config/resolvers/dns_resolver.c
index 7f9dd2543f..a467340ba2 100644
--- a/src/core/client_config/resolvers/dns_resolver.c
+++ b/src/core/client_config/resolvers/dns_resolver.c
@@ -81,9 +81,7 @@ static void dns_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
dns_resolver *r);
static void dns_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
-static void dns_channel_saw_error(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
- struct sockaddr *failing_address,
- int failing_address_len);
+static void dns_channel_saw_error(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
static void dns_next(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
grpc_client_config **target_config,
grpc_closure *on_complete);
@@ -103,8 +101,7 @@ static void dns_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver) {
}
static void dns_channel_saw_error(grpc_exec_ctx *exec_ctx,
- grpc_resolver *resolver, struct sockaddr *sa,
- int len) {
+ grpc_resolver *resolver) {
dns_resolver *r = (dns_resolver *)resolver;
gpr_mu_lock(&r->mu);
if (!r->resolving) {
diff --git a/src/core/client_config/resolvers/sockaddr_resolver.c b/src/core/client_config/resolvers/sockaddr_resolver.c
index 0b017f06c7..fd0212a1e7 100644
--- a/src/core/client_config/resolvers/sockaddr_resolver.c
+++ b/src/core/client_config/resolvers/sockaddr_resolver.c
@@ -83,9 +83,7 @@ static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
static void sockaddr_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
static void sockaddr_channel_saw_error(grpc_exec_ctx *exec_ctx,
- grpc_resolver *r,
- struct sockaddr *failing_address,
- int failing_address_len);
+ grpc_resolver *r);
static void sockaddr_next(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
grpc_client_config **target_config,
grpc_closure *on_complete);
@@ -107,8 +105,13 @@ static void sockaddr_shutdown(grpc_exec_ctx *exec_ctx,
}
static void sockaddr_channel_saw_error(grpc_exec_ctx *exec_ctx,
- grpc_resolver *resolver,
- struct sockaddr *sa, int len) {}
+ grpc_resolver *resolver) {
+ sockaddr_resolver *r = (sockaddr_resolver *)resolver;
+ gpr_mu_lock(&r->mu);
+ r->published = 0;
+ sockaddr_maybe_finish_next_locked(exec_ctx, r);
+ gpr_mu_unlock(&r->mu);
+}
static void sockaddr_next(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver,
grpc_client_config **target_config,
diff --git a/src/core/client_config/resolvers/zookeeper_resolver.c b/src/core/client_config/resolvers/zookeeper_resolver.c
index 136197d4c6..4924ca77d6 100644
--- a/src/core/client_config/resolvers/zookeeper_resolver.c
+++ b/src/core/client_config/resolvers/zookeeper_resolver.c
@@ -96,9 +96,7 @@ static void zookeeper_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
static void zookeeper_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *r);
static void zookeeper_channel_saw_error(grpc_exec_ctx *exec_ctx,
- grpc_resolver *r,
- struct sockaddr *failing_address,
- int failing_address_len);
+ grpc_resolver *r);
static void zookeeper_next(grpc_exec_ctx *exec_ctx, grpc_resolver *r,
grpc_client_config **target_config,
grpc_closure *on_complete);
@@ -125,8 +123,7 @@ static void zookeeper_shutdown(grpc_exec_ctx *exec_ctx,
}
static void zookeeper_channel_saw_error(grpc_exec_ctx *exec_ctx,
- grpc_resolver *resolver,
- struct sockaddr *sa, int len) {
+ grpc_resolver *resolver) {
zookeeper_resolver *r = (zookeeper_resolver *)resolver;
gpr_mu_lock(&r->mu);
if (r->resolving == 0) {
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index 28496ac2c9..6631e9bae2 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -47,39 +47,44 @@
#include "src/core/transport/connectivity_state.h"
#include "src/core/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
-typedef struct {
- /* all fields protected by subchannel->mu */
- /** refcount */
- int refs;
- /** parent subchannel */
- grpc_subchannel *subchannel;
-} connection;
+#define GET_CONNECTED_SUBCHANNEL(subchannel, barrier) \
+ ((grpc_connected_subchannel *)(gpr_atm_##barrier##_load( \
+ &(subchannel)->connected_subchannel)))
typedef struct {
grpc_closure closure;
- size_t version;
grpc_subchannel *subchannel;
grpc_connectivity_state connectivity_state;
} state_watcher;
-typedef struct waiting_for_connect {
- struct waiting_for_connect *next;
- grpc_closure *notify;
- grpc_pollset *pollset;
- gpr_atm *target;
+typedef struct external_state_watcher {
grpc_subchannel *subchannel;
- grpc_closure continuation;
-} waiting_for_connect;
+ 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;
@@ -88,15 +93,9 @@ struct grpc_subchannel {
/** address to connect to */
struct sockaddr *addr;
size_t addr_len;
+
/** initial string to send to peer */
gpr_slice initial_connect_string;
- /** master channel - the grpc_channel instance that ultimately owns
- this channel_data via its channel stack.
- We occasionally use this to bump the refcount on the master channel
- to keep ourselves alive through an asynchronous operation. */
- grpc_channel *master;
- /** have we seen a disconnection? */
- int disconnected;
/** set during connection */
grpc_connect_out_args connecting_result;
@@ -105,27 +104,24 @@ struct grpc_subchannel {
grpc_closure connected;
/** pollset_set tracking who's interested in a connection
- being setup - owned by the master channel (in particular the
- client_channel
- filter there-in) */
- grpc_pollset_set *pollset_set;
+ 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;
- /** active connection */
- connection *active;
- /** version number for the active connection */
- size_t active_version;
- /** refcount */
- int refs;
+ /** have we seen a disconnection? */
+ int disconnected;
/** are we connecting */
int connecting;
- /** things waiting for a connection */
- waiting_for_connect *waiting;
/** connectivity state tracking */
grpc_connectivity_state_tracker state_tracker;
+ external_state_watcher root_external_state_watcher;
+
/** next connect attempt time */
gpr_timespec next_attempt;
/** amount to backoff each failure */
@@ -139,151 +135,141 @@ struct grpc_subchannel {
};
struct grpc_subchannel_call {
- connection *connection;
+ grpc_connected_subchannel *connection;
};
#define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1))
-#define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)((con) + 1))
+#define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)(con))
#define CALLSTACK_TO_SUBCHANNEL_CALL(callstack) \
(((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 void subchannel_connected(grpc_exec_ctx *exec_ctx, void *subchannel,
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
-#define SUBCHANNEL_REF_LOCKED(p, r) \
- subchannel_ref_locked((p), __FILE__, __LINE__, (r))
-#define SUBCHANNEL_UNREF_LOCKED(p, r) \
- subchannel_unref_locked((p), __FILE__, __LINE__, (r))
-#define CONNECTION_REF_LOCKED(p, r) \
- connection_ref_locked((p), __FILE__, __LINE__, (r))
-#define CONNECTION_UNREF_LOCKED(cl, p, r) \
- connection_unref_locked((cl), (p), __FILE__, __LINE__, (r))
-#define REF_PASS_ARGS , file, line, reason
-#define REF_PASS_REASON , reason
+#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, (p)->refs + 1, reason)
+ (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, (p)->refs - 1, reason)
+ (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 SUBCHANNEL_REF_LOCKED(p, r) subchannel_ref_locked((p))
-#define SUBCHANNEL_UNREF_LOCKED(p, r) subchannel_unref_locked((p))
-#define CONNECTION_REF_LOCKED(p, r) connection_ref_locked((p))
-#define CONNECTION_UNREF_LOCKED(cl, p, r) connection_unref_locked((cl), (p))
-#define REF_PASS_ARGS
-#define REF_PASS_REASON
+#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, connection *c) {
- GPR_ASSERT(c->refs == 0);
+static void connection_destroy(grpc_exec_ctx *exec_ctx, void *arg,
+ int success) {
+ grpc_connected_subchannel *c = arg;
grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c));
gpr_free(c);
}
-static void connection_ref_locked(connection *c
- GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
- REF_LOG("CONNECTION", c);
- subchannel_ref_locked(c->subchannel REF_PASS_ARGS);
- ++c->refs;
+void grpc_connected_subchannel_ref(grpc_connected_subchannel *c
+ GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+ GRPC_CHANNEL_STACK_REF(CHANNEL_STACK_FROM_CONNECTION(c), REF_REASON);
}
-static grpc_subchannel *connection_unref_locked(
- grpc_exec_ctx *exec_ctx, connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
- grpc_subchannel *destroy = NULL;
- UNREF_LOG("CONNECTION", c);
- if (subchannel_unref_locked(c->subchannel REF_PASS_ARGS)) {
- destroy = c->subchannel;
- }
- if (--c->refs == 0 && c->subchannel->active != c) {
- connection_destroy(exec_ctx, c);
- }
- return destroy;
+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_ref_locked(grpc_subchannel *c
- GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
- REF_LOG("SUBCHANNEL", c);
- ++c->refs;
+static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
+ int success) {
+ grpc_subchannel *c = arg;
+ gpr_free((void *)c->filters);
+ grpc_channel_args_destroy(c->args);
+ gpr_free(c->addr);
+ gpr_slice_unref(c->initial_connect_string);
+ grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
+ grpc_connector_unref(exec_ctx, c->connector);
+ grpc_pollset_set_destroy(&c->pollset_set);
+ gpr_free(c);
}
-static int subchannel_unref_locked(grpc_subchannel *c
- GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
- UNREF_LOG("SUBCHANNEL", c);
- return --c->refs == 0;
+static gpr_atm ref_mutate(grpc_subchannel *c, gpr_atm delta,
+ int barrier REF_MUTATE_EXTRA_ARGS) {
+ gpr_atm old_val = barrier ? gpr_atm_full_fetch_add(&c->ref_pair, delta)
+ : gpr_atm_no_barrier_fetch_add(&c->ref_pair, delta);
+#ifdef GRPC_STREAM_REFCOUNT_DEBUG
+ gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
+ "SUBCHANNEL: %p % 12s 0x%08x -> 0x%08x [%s]", c, purpose, old_val,
+ old_val + delta, reason);
+#endif
+ return old_val;
}
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);
+ 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);
}
-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);
+void 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);
}
-static void subchannel_destroy(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
- if (c->active != NULL) {
- connection_destroy(exec_ctx, c->active);
+static void disconnect(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
+ grpc_connected_subchannel *con;
+ gpr_mu_lock(&c->mu);
+ GPR_ASSERT(!c->disconnected);
+ c->disconnected = 1;
+ grpc_connector_shutdown(exec_ctx, c->connector);
+ con = GET_CONNECTED_SUBCHANNEL(c, no_barrier);
+ if (con != NULL) {
+ GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, con, "connection");
+ gpr_atm_no_barrier_store(&c->connected_subchannel, 0xdeadbeef);
}
- gpr_free((void *)c->filters);
- grpc_channel_args_destroy(c->args);
- gpr_free(c->addr);
- gpr_slice_unref(c->initial_connect_string);
- grpc_connectivity_state_destroy(exec_ctx, &c->state_tracker);
- grpc_connector_unref(exec_ctx, c->connector);
- gpr_free(c);
+ gpr_mu_unlock(&c->mu);
}
-void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *c,
- grpc_pollset *pollset) {
- grpc_pollset_set_add_pollset(exec_ctx, c->pollset_set, pollset);
+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_del_interested_party(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *c,
- grpc_pollset *pollset) {
- grpc_pollset_set_del_pollset(exec_ctx, c->pollset_set, pollset);
+void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx,
+ grpc_subchannel *c
+ GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
+ gpr_atm old_refs;
+ old_refs = ref_mutate(c, -(gpr_atm)1, 1 REF_MUTATE_PURPOSE("WEAK_UNREF"));
+ if (old_refs == 1) {
+ grpc_exec_ctx_enqueue(exec_ctx, grpc_closure_create(subchannel_destroy, c),
+ 1);
+ }
}
static gpr_uint32 random_seed() {
@@ -293,10 +279,8 @@ static gpr_uint32 random_seed() {
grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
grpc_subchannel_args *args) {
grpc_subchannel *c = gpr_malloc(sizeof(*c));
- grpc_channel_element *parent_elem = grpc_channel_stack_last_element(
- grpc_channel_get_channel_stack(args->master));
memset(c, 0, sizeof(*c));
- c->refs = 1;
+ 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;
@@ -305,13 +289,14 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
sizeof(grpc_channel_filter *) * c->num_filters);
c->addr = gpr_malloc(args->addr_len);
memcpy(c->addr, args->addr, args->addr_len);
+ grpc_pollset_set_init(&c->pollset_set);
c->addr_len = args->addr_len;
grpc_set_initial_connect_string(&c->addr, &c->addr_len,
&c->initial_connect_string);
c->args = grpc_channel_args_copy(args->args);
- c->master = args->master;
- c->pollset_set = grpc_client_channel_get_connecting_pollset_set(parent_elem);
c->random = random_seed();
+ c->root_external_state_watcher.next = c->root_external_state_watcher.prev =
+ &c->root_external_state_watcher;
grpc_closure_init(&c->connected, subchannel_connected, c);
grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE,
"subchannel");
@@ -319,70 +304,18 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
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) {
grpc_connect_in_args args;
- args.interested_parties = c->pollset_set;
+ args.interested_parties = &c->pollset_set;
args.addr = c->addr;
args.addr_len = c->addr_len;
args.deadline = compute_connect_deadline(c);
args.channel_args = c->args;
args.initial_connect_string = c->initial_connect_string;
+ grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
+ GRPC_CHANNEL_CONNECTING, "state_change");
grpc_connector_connect(exec_ctx, c->connector, &args, &c->connecting_result,
&c->connected);
}
@@ -395,66 +328,6 @@ static void start_connect(grpc_exec_ctx *exec_ctx, grpc_subchannel *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 state;
gpr_mu_lock(&c->mu);
@@ -463,153 +336,138 @@ grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) {
return state;
}
-void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *c,
- grpc_connectivity_state *state,
- grpc_closure *notify) {
- int do_connect = 0;
- gpr_mu_lock(&c->mu);
- if (grpc_connectivity_state_notify_on_state_change(
- exec_ctx, &c->state_tracker, state, notify)) {
- do_connect = 1;
- c->connecting = 1;
- /* released by connection */
- SUBCHANNEL_REF_LOCKED(c, "connecting");
- GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting");
- connectivity_state_changed_locked(exec_ctx, c, "state_change");
+static void on_external_state_watcher_done(grpc_exec_ctx *exec_ctx, void *arg,
+ int success) {
+ 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_unlock(&c->mu);
-
- if (do_connect) {
- start_connect(exec_ctx, c);
- }
-}
-
-int grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *c,
- grpc_closure *subscribed_notify) {
- int success;
- gpr_mu_lock(&c->mu);
- success = grpc_connectivity_state_change_unsubscribe(
- exec_ctx, &c->state_tracker, subscribed_notify);
- gpr_mu_unlock(&c->mu);
- return success;
-}
-
-void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *c,
- grpc_transport_op *op) {
- connection *con = NULL;
- grpc_subchannel *destroy;
- int cancel_alarm = 0;
- gpr_mu_lock(&c->mu);
- if (c->active != NULL) {
- con = c->active;
- CONNECTION_REF_LOCKED(con, "transport-op");
- }
- if (op->disconnect) {
- c->disconnected = 1;
- connectivity_state_changed_locked(exec_ctx, c, "disconnect");
- if (c->have_alarm) {
- cancel_alarm = 1;
- }
- }
- gpr_mu_unlock(&c->mu);
-
- if (con != NULL) {
- 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);
+ gpr_mu_lock(&w->subchannel->mu);
+ w->next->prev = w->prev;
+ w->prev->next = w->next;
+ gpr_mu_unlock(&w->subchannel->mu);
+ GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, w->subchannel, "external_state_watcher");
+ gpr_free(w);
+ follow_up->cb(exec_ctx, follow_up->cb_arg, success);
+}
+
+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) {
+ int do_connect = 0;
+ external_state_watcher *w;
+ if (state == NULL) {
gpr_mu_lock(&c->mu);
- destroy = CONNECTION_UNREF_LOCKED(exec_ctx, con, "transport-op");
+ 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);
- if (destroy) {
- subchannel_destroy(exec_ctx, destroy);
+ } else {
+ w = gpr_malloc(sizeof(*w));
+ w->subchannel = c;
+ w->pollset_set = interested_parties;
+ w->notify = notify;
+ grpc_closure_init(&w->closure, on_external_state_watcher_done, w);
+ if (interested_parties != NULL) {
+ grpc_pollset_set_add_pollset_set(exec_ctx, &c->pollset_set,
+ interested_parties);
}
+ GRPC_SUBCHANNEL_WEAK_REF(c, "external_state_watcher");
+ gpr_mu_lock(&c->mu);
+ w->next = &c->root_external_state_watcher;
+ w->prev = w->next->prev;
+ w->next->prev = w->prev->next = w;
+ if (grpc_connectivity_state_notify_on_state_change(
+ exec_ctx, &c->state_tracker, state, &w->closure)) {
+ do_connect = 1;
+ c->connecting = 1;
+ /* released by connection */
+ GRPC_SUBCHANNEL_WEAK_REF(c, "connecting");
+ }
+ gpr_mu_unlock(&c->mu);
}
- if (cancel_alarm) {
- grpc_timer_cancel(exec_ctx, &c->alarm);
+ if (do_connect) {
+ start_connect(exec_ctx, c);
}
+}
- if (op->disconnect) {
- grpc_connector_shutdown(exec_ctx, c->connector);
- }
+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 on_state_changed(grpc_exec_ctx *exec_ctx, void *p,
- int iomgr_success) {
+static void subchannel_on_child_state_changed(grpc_exec_ctx *exec_ctx, void *p,
+ int iomgr_success) {
state_watcher *sw = p;
grpc_subchannel *c = sw->subchannel;
gpr_mu *mu = &c->mu;
- int destroy;
- grpc_transport_op op;
- grpc_channel_element *elem;
- connection *destroy_connection = NULL;
gpr_mu_lock(mu);
- /* if we failed or there is a version number mismatch, just leave
- this closure */
- if (!iomgr_success || sw->subchannel->active_version != sw->version) {
- goto done;
- }
-
- switch (sw->connectivity_state) {
- 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;
+ /* if we failed just leave this closure */
+ if (iomgr_success) {
+ 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_FATAL_FAILURE;
+ }
+ grpc_connectivity_state_set(exec_ctx, &c->state_tracker,
+ sw->connectivity_state, "reflect_child");
+ if (sw->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) {
+ 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;
+ }
}
-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);
- if (destroy) {
- subchannel_destroy(exec_ctx, c);
- }
- if (destroy_connection != NULL) {
- connection_destroy(exec_ctx, destroy_connection);
- }
+ GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "state_watcher");
+ gpr_free(sw);
+}
+
+static void connected_subchannel_state_op(grpc_exec_ctx *exec_ctx,
+ grpc_connected_subchannel *con,
+ grpc_pollset_set *interested_parties,
+ grpc_connectivity_state *state,
+ grpc_closure *closure) {
+ grpc_transport_op op;
+ grpc_channel_element *elem;
+ memset(&op, 0, sizeof(op));
+ op.connectivity_state = state;
+ op.on_connectivity_state_change = closure;
+ op.bind_pollset_set = interested_parties;
+ elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(con), 0);
+ elem->filter->start_transport_op(exec_ctx, elem, &op);
+}
+
+void grpc_connected_subchannel_notify_on_state_change(
+ grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *con,
+ grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
+ grpc_closure *closure) {
+ connected_subchannel_state_op(exec_ctx, con, interested_parties, state,
+ closure);
}
static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
size_t channel_stack_size;
- connection *con;
+ grpc_connected_subchannel *con;
grpc_channel_stack *stk;
size_t num_filters;
const grpc_channel_filter **filters;
- waiting_for_connect *w4c;
- grpc_transport_op op;
- state_watcher *sw;
- connection *destroy_connection = NULL;
- grpc_channel_element *elem;
+ state_watcher *sw_subchannel;
/* build final filter list */
num_filters = c->num_filters + c->connecting_result.num_filters + 1;
@@ -621,74 +479,52 @@ static void publish_transport(grpc_exec_ctx *exec_ctx, grpc_subchannel *c) {
/* construct channel stack */
channel_stack_size = grpc_channel_stack_size(filters, num_filters);
- con = gpr_malloc(sizeof(connection) + channel_stack_size);
- stk = (grpc_channel_stack *)(con + 1);
- con->refs = 0;
- con->subchannel = c;
- grpc_channel_stack_init(exec_ctx, filters, num_filters, c->master, c->args,
- stk);
+ con = gpr_malloc(channel_stack_size);
+ stk = CHANNEL_STACK_FROM_CONNECTION(con);
+ grpc_channel_stack_init(exec_ctx, 1, connection_destroy, con, filters,
+ num_filters, c->args, "CONNECTED_SUBCHANNEL", stk);
grpc_connected_channel_bind_transport(stk, c->connecting_result.transport);
gpr_free((void *)c->connecting_result.filters);
memset(&c->connecting_result, 0, sizeof(c->connecting_result));
/* initialize state watcher */
- sw = gpr_malloc(sizeof(*sw));
- grpc_closure_init(&sw->closure, on_state_changed, sw);
- sw->subchannel = c;
- sw->connectivity_state = GRPC_CHANNEL_READY;
+ 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);
gpr_mu_lock(&c->mu);
if (c->disconnected) {
gpr_mu_unlock(&c->mu);
- gpr_free(sw);
+ gpr_free(sw_subchannel);
gpr_free((void *)filters);
grpc_channel_stack_destroy(exec_ctx, stk);
- GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
- GRPC_SUBCHANNEL_UNREF(exec_ctx, c, "connecting");
+ gpr_free(con);
+ GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
return;
}
/* publish */
- if (c->active != NULL && c->active->refs == 0) {
- destroy_connection = c->active;
- }
- c->active = con;
- c->active_version++;
- sw->version = c->active_version;
+ GPR_ASSERT(gpr_atm_no_barrier_cas(&c->connected_subchannel, 0, (gpr_atm)con));
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 */
- memset(&op, 0, sizeof(op));
- op.connectivity_state = &sw->connectivity_state;
- op.on_connectivity_state_change = &sw->closure;
- 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 =
- grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(c->active), 0);
- elem->filter->start_transport_op(exec_ctx, elem, &op);
+ 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 */
- connectivity_state_changed_locked(exec_ctx, c, "connected");
- w4c = c->waiting;
- c->waiting = NULL;
+ grpc_connectivity_state_set(exec_ctx, &c->state_tracker, GRPC_CHANNEL_READY,
+ "connected");
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);
-
- if (destroy_connection != NULL) {
- connection_destroy(exec_ctx, destroy_connection);
- }
}
/* Generate a random number between 0 and 1. */
@@ -742,29 +578,31 @@ static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, int iomgr_success) {
if (c->disconnected) {
iomgr_success = 0;
}
- connectivity_state_changed_locked(exec_ctx, c, "alarm");
gpr_mu_unlock(&c->mu);
if (iomgr_success) {
update_reconnect_parameters(c);
continue_connect(exec_ctx, c);
} else {
- cancel_waiting_calls(exec_ctx, c, iomgr_success);
- GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, c->master, "connecting");
- GRPC_SUBCHANNEL_UNREF(exec_ctx, c, "connecting");
+ GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
}
}
static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg,
int iomgr_success) {
grpc_subchannel *c = arg;
+
if (c->connecting_result.transport != NULL) {
publish_transport(exec_ctx, c);
+ } else if (c->disconnected) {
+ GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
} else {
gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
gpr_mu_lock(&c->mu);
GPR_ASSERT(!c->have_alarm);
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);
gpr_mu_unlock(&c->mu);
}
@@ -781,29 +619,6 @@ static gpr_timespec compute_connect_deadline(grpc_subchannel *c) {
: 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
*/
@@ -811,37 +626,22 @@ static void connectivity_state_changed_locked(grpc_exec_ctx *exec_ctx,
static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call,
int success) {
grpc_subchannel_call *c = call;
- gpr_mu *mu = &c->connection->subchannel->mu;
- grpc_subchannel *destroy;
GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0);
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);
+ GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, c->connection, "subchannel_call");
gpr_free(c);
- if (destroy != NULL) {
- subchannel_destroy(exec_ctx, destroy);
- }
GPR_TIMER_END("grpc_subchannel_call_unref.destroy", 0);
}
void grpc_subchannel_call_ref(grpc_subchannel_call *c
GRPC_SUBCHANNEL_REF_EXTRA_ARGS) {
-#ifdef GRPC_STREAM_REFCOUNT_DEBUG
- grpc_call_stack_ref(SUBCHANNEL_CALL_TO_CALL_STACK(c), reason);
-#else
- grpc_call_stack_ref(SUBCHANNEL_CALL_TO_CALL_STACK(c));
-#endif
+ 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) {
-#ifdef GRPC_STREAM_REFCOUNT_DEBUG
- grpc_call_stack_unref(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), reason);
-#else
- grpc_call_stack_unref(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c));
-#endif
+ 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,
@@ -859,24 +659,26 @@ void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx,
top_elem->filter->start_transport_stream_op(exec_ctx, top_elem, op);
}
-static grpc_subchannel_call *create_call(grpc_exec_ctx *exec_ctx,
- connection *con,
- grpc_pollset *pollset) {
+grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(
+ grpc_subchannel *c) {
+ 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_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con);
grpc_subchannel_call *call =
gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size);
grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(call);
call->connection = con;
+ GRPC_CONNECTED_SUBCHANNEL_REF(con, "subchannel_call");
grpc_call_stack_init(exec_ctx, chanstk, 1, subchannel_call_destroy, call,
NULL, NULL, callstk);
grpc_call_stack_set_pollset(exec_ctx, callstk, pollset);
return call;
}
-grpc_channel *grpc_subchannel_get_master(grpc_subchannel *subchannel) {
- return subchannel->master;
-}
-
grpc_call_stack *grpc_subchannel_call_get_call_stack(
grpc_subchannel_call *subchannel_call) {
return SUBCHANNEL_CALL_TO_CALL_STACK(subchannel_call);
diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h
index 85ea3739e4..74ebcecfba 100644
--- a/src/core/client_config/subchannel.h
+++ b/src/core/client_config/subchannel.h
@@ -41,6 +41,7 @@
/** A (sub-)channel that knows how to connect to exactly one target
address. Provides a target for load balancing. */
typedef struct grpc_subchannel grpc_subchannel;
+typedef struct grpc_connected_subchannel grpc_connected_subchannel;
typedef struct grpc_subchannel_call grpc_subchannel_call;
typedef struct grpc_subchannel_args grpc_subchannel_args;
@@ -49,6 +50,14 @@ typedef struct grpc_subchannel_args grpc_subchannel_args;
grpc_subchannel_ref((p), __FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_UNREF(cl, p, r) \
grpc_subchannel_unref((cl), (p), __FILE__, __LINE__, (r))
+#define GRPC_SUBCHANNEL_WEAK_REF(p, r) \
+ grpc_subchannel_weak_ref((p), __FILE__, __LINE__, (r))
+#define GRPC_SUBCHANNEL_WEAK_UNREF(cl, p, r) \
+ grpc_subchannel_weak_unref((cl), (p), __FILE__, __LINE__, (r))
+#define GRPC_CONNECTED_SUBCHANNEL_REF(p, r) \
+ grpc_connected_subchannel_ref((p), __FILE__, __LINE__, (r))
+#define GRPC_CONNECTED_SUBCHANNEL_UNREF(cl, p, r) \
+ grpc_connected_subchannel_unref((cl), (p), __FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_CALL_REF(p, r) \
grpc_subchannel_call_ref((p), __FILE__, __LINE__, (r))
#define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \
@@ -58,6 +67,12 @@ typedef struct grpc_subchannel_args grpc_subchannel_args;
#else
#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_WEAK_REF(p, r) grpc_subchannel_weak_ref((p))
+#define GRPC_SUBCHANNEL_WEAK_UNREF(cl, p, r) \
+ grpc_subchannel_weak_unref((cl), (p))
+#define GRPC_CONNECTED_SUBCHANNEL_REF(p, r) grpc_connected_subchannel_ref((p))
+#define GRPC_CONNECTED_SUBCHANNEL_UNREF(cl, p, r) \
+ grpc_connected_subchannel_unref((cl), (p))
#define GRPC_SUBCHANNEL_CALL_REF(p, r) grpc_subchannel_call_ref((p))
#define GRPC_SUBCHANNEL_CALL_UNREF(cl, p, r) \
grpc_subchannel_call_unref((cl), (p))
@@ -69,33 +84,31 @@ void grpc_subchannel_ref(grpc_subchannel *channel
void grpc_subchannel_unref(grpc_exec_ctx *exec_ctx,
grpc_subchannel *channel
GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_subchannel_weak_ref(grpc_subchannel *channel
+ GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx,
+ grpc_subchannel *channel
+ GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_connected_subchannel_ref(grpc_connected_subchannel *channel
+ GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
+void grpc_connected_subchannel_unref(grpc_exec_ctx *exec_ctx,
+ grpc_connected_subchannel *channel
+ GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
void grpc_subchannel_call_ref(grpc_subchannel_call *call
GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
void grpc_subchannel_call_unref(grpc_exec_ctx *exec_ctx,
grpc_subchannel_call *call
GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
-/** construct a subchannel call (possibly asynchronously).
- *
- * If the returned status is 1, the call will return immediately and \a target
- * will point to a connected \a subchannel_call instance. Note that \a notify
- * 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);
+/** construct a subchannel call */
+grpc_subchannel_call *grpc_connected_subchannel_create_call(
+ grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *connected_subchannel,
+ grpc_pollset *pollset);
/** process a transport level op */
-void grpc_subchannel_process_transport_op(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *subchannel,
- 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 */
grpc_connectivity_state grpc_subchannel_check_connectivity(
@@ -103,26 +116,19 @@ grpc_connectivity_state grpc_subchannel_check_connectivity(
/** call notify when the connectivity state of a channel changes from *state.
Updates *state with the new state of the channel */
-void grpc_subchannel_notify_on_state_change(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *channel,
- grpc_connectivity_state *state,
- grpc_closure *notify);
-
-/** 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
- * 0. */
-int grpc_subchannel_state_change_unsubscribe(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *channel,
- grpc_closure *subscribed_notify);
-
-/** express interest in \a channel's activities through \a pollset. */
-void grpc_subchannel_add_interested_party(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *channel,
- grpc_pollset *pollset);
-/** stop following \a channel's activity through \a pollset. */
-void grpc_subchannel_del_interested_party(grpc_exec_ctx *exec_ctx,
- grpc_subchannel *channel,
- grpc_pollset *pollset);
+void grpc_subchannel_notify_on_state_change(
+ grpc_exec_ctx *exec_ctx, grpc_subchannel *channel,
+ grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
+ grpc_closure *notify);
+void grpc_connected_subchannel_notify_on_state_change(
+ grpc_exec_ctx *exec_ctx, grpc_connected_subchannel *channel,
+ grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
+ grpc_closure *notify);
+
+/** retrieve the grpc_connected_subchannel - or NULL if called before
+ the subchannel becomes connected */
+grpc_connected_subchannel *grpc_subchannel_get_connected_subchannel(
+ grpc_subchannel *subchannel);
/** continue processing a transport op */
void grpc_subchannel_call_process_op(grpc_exec_ctx *exec_ctx,
@@ -147,15 +153,10 @@ struct grpc_subchannel_args {
/** Address to connect to */
struct sockaddr *addr;
size_t addr_len;
- /** master channel */
- grpc_channel *master;
};
/** create a subchannel given a connector */
grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
grpc_subchannel_args *args);
-/** Return the master channel associated with the subchannel */
-grpc_channel *grpc_subchannel_get_master(grpc_subchannel *subchannel);
-
#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_H */
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-17 16:33:05 +0000