indent pass to get logical source lines on one physical line
diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 903f9ea..40b428c 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -51,7 +51,8 @@
typedef struct call_data call_data;
-typedef struct {
+typedef struct
+{
/** metadata context for this channel */
grpc_mdctx *mdctx;
/** resolver for this channel */
@@ -89,14 +90,16 @@
to watch for state changes from the lb_policy. When a state change is seen,
we
update the channel, and create a new watcher */
-typedef struct {
+typedef struct
+{
channel_data *chand;
grpc_closure on_changed;
grpc_connectivity_state state;
grpc_lb_policy *lb_policy;
} lb_policy_connectivity_watcher;
-typedef enum {
+typedef enum
+{
CALL_CREATED,
CALL_WAITING_FOR_SEND,
CALL_WAITING_FOR_CONFIG,
@@ -106,7 +109,8 @@
CALL_CANCELLED
} call_state;
-struct call_data {
+struct call_data
+{
/* owning element */
grpc_call_element *elem;
@@ -123,367 +127,406 @@
grpc_linked_mdelem details;
};
-static grpc_closure *merge_into_waiting_op(grpc_call_element *elem,
- grpc_transport_stream_op *new_op)
- GRPC_MUST_USE_RESULT;
+static grpc_closure *
+merge_into_waiting_op (grpc_call_element * elem, grpc_transport_stream_op * new_op)
+ GRPC_MUST_USE_RESULT;
-static void handle_op_after_cancellation(grpc_call_element *elem,
- grpc_transport_stream_op *op,
- grpc_closure_list *closure_list) {
+ static void handle_op_after_cancellation (grpc_call_element * elem, grpc_transport_stream_op * op, grpc_closure_list * closure_list)
+{
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
- if (op->send_ops) {
- grpc_stream_ops_unref_owned_objects(op->send_ops->ops, op->send_ops->nops);
- op->on_done_send->cb(op->on_done_send->cb_arg, 0, closure_list);
- }
- if (op->recv_ops) {
- char status[GPR_LTOA_MIN_BUFSIZE];
- grpc_metadata_batch mdb;
- gpr_ltoa(GRPC_STATUS_CANCELLED, status);
- calld->status.md =
- grpc_mdelem_from_strings(chand->mdctx, "grpc-status", status);
- calld->details.md =
- grpc_mdelem_from_strings(chand->mdctx, "grpc-message", "Cancelled");
- calld->status.prev = calld->details.next = NULL;
- calld->status.next = &calld->details;
- calld->details.prev = &calld->status;
- mdb.list.head = &calld->status;
- mdb.list.tail = &calld->details;
- mdb.garbage.head = mdb.garbage.tail = NULL;
- mdb.deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
- grpc_sopb_add_metadata(op->recv_ops, mdb);
- *op->recv_state = GRPC_STREAM_CLOSED;
- op->on_done_recv->cb(op->on_done_recv->cb_arg, 1, closure_list);
- }
- if (op->on_consumed) {
- op->on_consumed->cb(op->on_consumed->cb_arg, 0, closure_list);
- }
+ if (op->send_ops)
+ {
+ grpc_stream_ops_unref_owned_objects (op->send_ops->ops, op->send_ops->nops);
+ op->on_done_send->cb (op->on_done_send->cb_arg, 0, closure_list);
+ }
+ if (op->recv_ops)
+ {
+ char status[GPR_LTOA_MIN_BUFSIZE];
+ grpc_metadata_batch mdb;
+ gpr_ltoa (GRPC_STATUS_CANCELLED, status);
+ calld->status.md = grpc_mdelem_from_strings (chand->mdctx, "grpc-status", status);
+ calld->details.md = grpc_mdelem_from_strings (chand->mdctx, "grpc-message", "Cancelled");
+ calld->status.prev = calld->details.next = NULL;
+ calld->status.next = &calld->details;
+ calld->details.prev = &calld->status;
+ mdb.list.head = &calld->status;
+ mdb.list.tail = &calld->details;
+ mdb.garbage.head = mdb.garbage.tail = NULL;
+ mdb.deadline = gpr_inf_future (GPR_CLOCK_REALTIME);
+ grpc_sopb_add_metadata (op->recv_ops, mdb);
+ *op->recv_state = GRPC_STREAM_CLOSED;
+ op->on_done_recv->cb (op->on_done_recv->cb_arg, 1, closure_list);
+ }
+ if (op->on_consumed)
+ {
+ op->on_consumed->cb (op->on_consumed->cb_arg, 0, closure_list);
+ }
}
-typedef struct {
+typedef struct
+{
grpc_closure closure;
grpc_call_element *elem;
} waiting_call;
-static void perform_transport_stream_op(grpc_call_element *elem,
- grpc_transport_stream_op *op,
- int continuation,
- grpc_closure_list *closure_list);
+static void perform_transport_stream_op (grpc_call_element * elem, grpc_transport_stream_op * op, int continuation, grpc_closure_list * closure_list);
-static void continue_with_pick(void *arg, int iomgr_success,
- grpc_closure_list *closure_list) {
+static void
+continue_with_pick (void *arg, int iomgr_success, grpc_closure_list * closure_list)
+{
waiting_call *wc = arg;
call_data *calld = wc->elem->call_data;
- perform_transport_stream_op(wc->elem, &calld->waiting_op, 1, closure_list);
- gpr_free(wc);
+ perform_transport_stream_op (wc->elem, &calld->waiting_op, 1, closure_list);
+ gpr_free (wc);
}
-static void add_to_lb_policy_wait_queue_locked_state_config(
- grpc_call_element *elem) {
+static void
+add_to_lb_policy_wait_queue_locked_state_config (grpc_call_element * elem)
+{
channel_data *chand = elem->channel_data;
- waiting_call *wc = gpr_malloc(sizeof(*wc));
- grpc_closure_init(&wc->closure, continue_with_pick, wc);
+ waiting_call *wc = gpr_malloc (sizeof (*wc));
+ grpc_closure_init (&wc->closure, continue_with_pick, wc);
wc->elem = elem;
- grpc_closure_list_add(&chand->waiting_for_config_closures, &wc->closure, 1);
+ grpc_closure_list_add (&chand->waiting_for_config_closures, &wc->closure, 1);
}
-static int is_empty(void *p, int len) {
+static int
+is_empty (void *p, int len)
+{
char *ptr = p;
int i;
- for (i = 0; i < len; i++) {
- if (ptr[i] != 0) return 0;
- }
+ for (i = 0; i < len; i++)
+ {
+ if (ptr[i] != 0)
+ return 0;
+ }
return 1;
}
-static void started_call(void *arg, int iomgr_success,
- grpc_closure_list *closure_list) {
+static void
+started_call (void *arg, int iomgr_success, grpc_closure_list * closure_list)
+{
call_data *calld = arg;
grpc_transport_stream_op op;
int have_waiting;
- gpr_mu_lock(&calld->mu_state);
- if (calld->state == CALL_CANCELLED && calld->subchannel_call != NULL) {
- memset(&op, 0, sizeof(op));
- op.cancel_with_status = GRPC_STATUS_CANCELLED;
- gpr_mu_unlock(&calld->mu_state);
- grpc_subchannel_call_process_op(calld->subchannel_call, &op, closure_list);
- } else if (calld->state == CALL_WAITING_FOR_CALL) {
- have_waiting = !is_empty(&calld->waiting_op, sizeof(calld->waiting_op));
- if (calld->subchannel_call != NULL) {
- calld->state = CALL_ACTIVE;
- gpr_mu_unlock(&calld->mu_state);
- if (have_waiting) {
- grpc_subchannel_call_process_op(calld->subchannel_call,
- &calld->waiting_op, closure_list);
- }
- } else {
- calld->state = CALL_CANCELLED;
- gpr_mu_unlock(&calld->mu_state);
- if (have_waiting) {
- handle_op_after_cancellation(calld->elem, &calld->waiting_op,
- closure_list);
- }
+ gpr_mu_lock (&calld->mu_state);
+ if (calld->state == CALL_CANCELLED && calld->subchannel_call != NULL)
+ {
+ memset (&op, 0, sizeof (op));
+ op.cancel_with_status = GRPC_STATUS_CANCELLED;
+ gpr_mu_unlock (&calld->mu_state);
+ grpc_subchannel_call_process_op (calld->subchannel_call, &op, closure_list);
}
- } else {
- GPR_ASSERT(calld->state == CALL_CANCELLED);
- gpr_mu_unlock(&calld->mu_state);
- }
+ else if (calld->state == CALL_WAITING_FOR_CALL)
+ {
+ have_waiting = !is_empty (&calld->waiting_op, sizeof (calld->waiting_op));
+ if (calld->subchannel_call != NULL)
+ {
+ calld->state = CALL_ACTIVE;
+ gpr_mu_unlock (&calld->mu_state);
+ if (have_waiting)
+ {
+ grpc_subchannel_call_process_op (calld->subchannel_call, &calld->waiting_op, closure_list);
+ }
+ }
+ else
+ {
+ calld->state = CALL_CANCELLED;
+ gpr_mu_unlock (&calld->mu_state);
+ if (have_waiting)
+ {
+ handle_op_after_cancellation (calld->elem, &calld->waiting_op, closure_list);
+ }
+ }
+ }
+ else
+ {
+ GPR_ASSERT (calld->state == CALL_CANCELLED);
+ gpr_mu_unlock (&calld->mu_state);
+ }
}
-static void picked_target(void *arg, int iomgr_success,
- grpc_closure_list *closure_list) {
+static void
+picked_target (void *arg, int iomgr_success, grpc_closure_list * closure_list)
+{
call_data *calld = arg;
grpc_pollset *pollset;
- if (calld->picked_channel == NULL) {
- /* treat this like a cancellation */
- calld->waiting_op.cancel_with_status = GRPC_STATUS_UNAVAILABLE;
- perform_transport_stream_op(calld->elem, &calld->waiting_op, 1,
- closure_list);
- } else {
- gpr_mu_lock(&calld->mu_state);
- if (calld->state == CALL_CANCELLED) {
- gpr_mu_unlock(&calld->mu_state);
- handle_op_after_cancellation(calld->elem, &calld->waiting_op,
- closure_list);
- } else {
- GPR_ASSERT(calld->state == CALL_WAITING_FOR_PICK);
- calld->state = CALL_WAITING_FOR_CALL;
- pollset = calld->waiting_op.bind_pollset;
- gpr_mu_unlock(&calld->mu_state);
- grpc_closure_init(&calld->async_setup_task, started_call, calld);
- grpc_subchannel_create_call(calld->picked_channel, pollset,
- &calld->subchannel_call,
- &calld->async_setup_task, closure_list);
+ if (calld->picked_channel == NULL)
+ {
+ /* treat this like a cancellation */
+ calld->waiting_op.cancel_with_status = GRPC_STATUS_UNAVAILABLE;
+ perform_transport_stream_op (calld->elem, &calld->waiting_op, 1, closure_list);
}
- }
+ else
+ {
+ gpr_mu_lock (&calld->mu_state);
+ if (calld->state == CALL_CANCELLED)
+ {
+ gpr_mu_unlock (&calld->mu_state);
+ handle_op_after_cancellation (calld->elem, &calld->waiting_op, closure_list);
+ }
+ else
+ {
+ GPR_ASSERT (calld->state == CALL_WAITING_FOR_PICK);
+ calld->state = CALL_WAITING_FOR_CALL;
+ pollset = calld->waiting_op.bind_pollset;
+ gpr_mu_unlock (&calld->mu_state);
+ grpc_closure_init (&calld->async_setup_task, started_call, calld);
+ grpc_subchannel_create_call (calld->picked_channel, pollset, &calld->subchannel_call, &calld->async_setup_task, closure_list);
+ }
+ }
}
-static grpc_closure *merge_into_waiting_op(grpc_call_element *elem,
- grpc_transport_stream_op *new_op) {
+static grpc_closure *
+merge_into_waiting_op (grpc_call_element * elem, grpc_transport_stream_op * new_op)
+{
call_data *calld = elem->call_data;
grpc_closure *consumed_op = NULL;
grpc_transport_stream_op *waiting_op = &calld->waiting_op;
- GPR_ASSERT((waiting_op->send_ops != NULL) + (new_op->send_ops != NULL) <= 1);
- GPR_ASSERT((waiting_op->recv_ops != NULL) + (new_op->recv_ops != NULL) <= 1);
- if (new_op->send_ops != NULL) {
- waiting_op->send_ops = new_op->send_ops;
- waiting_op->is_last_send = new_op->is_last_send;
- waiting_op->on_done_send = new_op->on_done_send;
- }
- if (new_op->recv_ops != NULL) {
- waiting_op->recv_ops = new_op->recv_ops;
- waiting_op->recv_state = new_op->recv_state;
- waiting_op->on_done_recv = new_op->on_done_recv;
- }
- if (new_op->on_consumed != NULL) {
- if (waiting_op->on_consumed != NULL) {
- consumed_op = waiting_op->on_consumed;
+ GPR_ASSERT ((waiting_op->send_ops != NULL) + (new_op->send_ops != NULL) <= 1);
+ GPR_ASSERT ((waiting_op->recv_ops != NULL) + (new_op->recv_ops != NULL) <= 1);
+ if (new_op->send_ops != NULL)
+ {
+ waiting_op->send_ops = new_op->send_ops;
+ waiting_op->is_last_send = new_op->is_last_send;
+ waiting_op->on_done_send = new_op->on_done_send;
}
- waiting_op->on_consumed = new_op->on_consumed;
- }
- if (new_op->cancel_with_status != GRPC_STATUS_OK) {
- waiting_op->cancel_with_status = new_op->cancel_with_status;
- }
+ if (new_op->recv_ops != NULL)
+ {
+ waiting_op->recv_ops = new_op->recv_ops;
+ waiting_op->recv_state = new_op->recv_state;
+ waiting_op->on_done_recv = new_op->on_done_recv;
+ }
+ if (new_op->on_consumed != NULL)
+ {
+ if (waiting_op->on_consumed != NULL)
+ {
+ consumed_op = waiting_op->on_consumed;
+ }
+ waiting_op->on_consumed = new_op->on_consumed;
+ }
+ if (new_op->cancel_with_status != GRPC_STATUS_OK)
+ {
+ waiting_op->cancel_with_status = new_op->cancel_with_status;
+ }
return consumed_op;
}
-static char *cc_get_peer(grpc_call_element *elem,
- grpc_closure_list *closure_list) {
+static char *
+cc_get_peer (grpc_call_element * elem, grpc_closure_list * closure_list)
+{
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
grpc_subchannel_call *subchannel_call;
char *result;
- gpr_mu_lock(&calld->mu_state);
- if (calld->state == CALL_ACTIVE) {
- subchannel_call = calld->subchannel_call;
- GRPC_SUBCHANNEL_CALL_REF(subchannel_call, "get_peer");
- gpr_mu_unlock(&calld->mu_state);
- result = grpc_subchannel_call_get_peer(subchannel_call, closure_list);
- GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call, "get_peer", closure_list);
- return result;
- } else {
- gpr_mu_unlock(&calld->mu_state);
- return grpc_channel_get_target(chand->master);
- }
+ gpr_mu_lock (&calld->mu_state);
+ if (calld->state == CALL_ACTIVE)
+ {
+ subchannel_call = calld->subchannel_call;
+ GRPC_SUBCHANNEL_CALL_REF (subchannel_call, "get_peer");
+ gpr_mu_unlock (&calld->mu_state);
+ result = grpc_subchannel_call_get_peer (subchannel_call, closure_list);
+ GRPC_SUBCHANNEL_CALL_UNREF (subchannel_call, "get_peer", closure_list);
+ return result;
+ }
+ else
+ {
+ gpr_mu_unlock (&calld->mu_state);
+ return grpc_channel_get_target (chand->master);
+ }
}
-static void perform_transport_stream_op(grpc_call_element *elem,
- grpc_transport_stream_op *op,
- int continuation,
- grpc_closure_list *closure_list) {
+static void
+perform_transport_stream_op (grpc_call_element * elem, grpc_transport_stream_op * op, int continuation, grpc_closure_list * closure_list)
+{
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
grpc_subchannel_call *subchannel_call;
grpc_lb_policy *lb_policy;
grpc_transport_stream_op op2;
- GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
- GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
+ GPR_ASSERT (elem->filter == &grpc_client_channel_filter);
+ GRPC_CALL_LOG_OP (GPR_INFO, elem, op);
- gpr_mu_lock(&calld->mu_state);
- switch (calld->state) {
+ gpr_mu_lock (&calld->mu_state);
+ switch (calld->state)
+ {
case CALL_ACTIVE:
- GPR_ASSERT(!continuation);
+ GPR_ASSERT (!continuation);
subchannel_call = calld->subchannel_call;
- gpr_mu_unlock(&calld->mu_state);
- grpc_subchannel_call_process_op(subchannel_call, op, closure_list);
+ gpr_mu_unlock (&calld->mu_state);
+ grpc_subchannel_call_process_op (subchannel_call, op, closure_list);
break;
case CALL_CANCELLED:
- gpr_mu_unlock(&calld->mu_state);
- handle_op_after_cancellation(elem, op, closure_list);
+ gpr_mu_unlock (&calld->mu_state);
+ handle_op_after_cancellation (elem, op, closure_list);
break;
case CALL_WAITING_FOR_SEND:
- GPR_ASSERT(!continuation);
- grpc_closure_list_add(closure_list, merge_into_waiting_op(elem, op), 1);
- if (!calld->waiting_op.send_ops &&
- calld->waiting_op.cancel_with_status == GRPC_STATUS_OK) {
- gpr_mu_unlock(&calld->mu_state);
- break;
- }
+ GPR_ASSERT (!continuation);
+ grpc_closure_list_add (closure_list, merge_into_waiting_op (elem, op), 1);
+ if (!calld->waiting_op.send_ops && calld->waiting_op.cancel_with_status == GRPC_STATUS_OK)
+ {
+ gpr_mu_unlock (&calld->mu_state);
+ break;
+ }
*op = calld->waiting_op;
- memset(&calld->waiting_op, 0, sizeof(calld->waiting_op));
+ memset (&calld->waiting_op, 0, sizeof (calld->waiting_op));
continuation = 1;
- /* fall through */
+ /* fall through */
case CALL_WAITING_FOR_CONFIG:
case CALL_WAITING_FOR_PICK:
case CALL_WAITING_FOR_CALL:
- if (!continuation) {
- if (op->cancel_with_status != GRPC_STATUS_OK) {
- calld->state = CALL_CANCELLED;
- op2 = calld->waiting_op;
- memset(&calld->waiting_op, 0, sizeof(calld->waiting_op));
- if (op->on_consumed) {
- calld->waiting_op.on_consumed = op->on_consumed;
- op->on_consumed = NULL;
- } else if (op2.on_consumed) {
- calld->waiting_op.on_consumed = op2.on_consumed;
- op2.on_consumed = NULL;
- }
- gpr_mu_unlock(&calld->mu_state);
- handle_op_after_cancellation(elem, op, closure_list);
- handle_op_after_cancellation(elem, &op2, closure_list);
- } else {
- grpc_closure_list_add(closure_list, merge_into_waiting_op(elem, op),
- 1);
- gpr_mu_unlock(&calld->mu_state);
- }
- break;
- }
- /* fall through */
+ if (!continuation)
+ {
+ if (op->cancel_with_status != GRPC_STATUS_OK)
+ {
+ calld->state = CALL_CANCELLED;
+ op2 = calld->waiting_op;
+ memset (&calld->waiting_op, 0, sizeof (calld->waiting_op));
+ if (op->on_consumed)
+ {
+ calld->waiting_op.on_consumed = op->on_consumed;
+ op->on_consumed = NULL;
+ }
+ else if (op2.on_consumed)
+ {
+ calld->waiting_op.on_consumed = op2.on_consumed;
+ op2.on_consumed = NULL;
+ }
+ gpr_mu_unlock (&calld->mu_state);
+ handle_op_after_cancellation (elem, op, closure_list);
+ handle_op_after_cancellation (elem, &op2, closure_list);
+ }
+ else
+ {
+ grpc_closure_list_add (closure_list, merge_into_waiting_op (elem, op), 1);
+ gpr_mu_unlock (&calld->mu_state);
+ }
+ break;
+ }
+ /* fall through */
case CALL_CREATED:
- if (op->cancel_with_status != GRPC_STATUS_OK) {
- calld->state = CALL_CANCELLED;
- gpr_mu_unlock(&calld->mu_state);
- handle_op_after_cancellation(elem, op, closure_list);
- } else {
- calld->waiting_op = *op;
+ if (op->cancel_with_status != GRPC_STATUS_OK)
+ {
+ calld->state = CALL_CANCELLED;
+ gpr_mu_unlock (&calld->mu_state);
+ handle_op_after_cancellation (elem, op, closure_list);
+ }
+ else
+ {
+ calld->waiting_op = *op;
- if (op->send_ops == NULL) {
- /* need to have some send ops before we can select the
- lb target */
- calld->state = CALL_WAITING_FOR_SEND;
- gpr_mu_unlock(&calld->mu_state);
- } else {
- gpr_mu_lock(&chand->mu_config);
- lb_policy = chand->lb_policy;
- if (lb_policy) {
- grpc_transport_stream_op *op = &calld->waiting_op;
- grpc_pollset *bind_pollset = op->bind_pollset;
- grpc_metadata_batch *initial_metadata =
- &op->send_ops->ops[0].data.metadata;
- GRPC_LB_POLICY_REF(lb_policy, "pick");
- gpr_mu_unlock(&chand->mu_config);
- calld->state = CALL_WAITING_FOR_PICK;
+ if (op->send_ops == NULL)
+ {
+ /* need to have some send ops before we can select the
+ lb target */
+ calld->state = CALL_WAITING_FOR_SEND;
+ gpr_mu_unlock (&calld->mu_state);
+ }
+ else
+ {
+ gpr_mu_lock (&chand->mu_config);
+ lb_policy = chand->lb_policy;
+ if (lb_policy)
+ {
+ grpc_transport_stream_op *op = &calld->waiting_op;
+ grpc_pollset *bind_pollset = op->bind_pollset;
+ grpc_metadata_batch *initial_metadata = &op->send_ops->ops[0].data.metadata;
+ GRPC_LB_POLICY_REF (lb_policy, "pick");
+ gpr_mu_unlock (&chand->mu_config);
+ calld->state = CALL_WAITING_FOR_PICK;
- GPR_ASSERT(op->bind_pollset);
- GPR_ASSERT(op->send_ops);
- GPR_ASSERT(op->send_ops->nops >= 1);
- GPR_ASSERT(op->send_ops->ops[0].type == GRPC_OP_METADATA);
- gpr_mu_unlock(&calld->mu_state);
+ GPR_ASSERT (op->bind_pollset);
+ GPR_ASSERT (op->send_ops);
+ GPR_ASSERT (op->send_ops->nops >= 1);
+ GPR_ASSERT (op->send_ops->ops[0].type == GRPC_OP_METADATA);
+ gpr_mu_unlock (&calld->mu_state);
- grpc_closure_init(&calld->async_setup_task, picked_target, calld);
- grpc_lb_policy_pick(lb_policy, bind_pollset, initial_metadata,
- &calld->picked_channel,
- &calld->async_setup_task, closure_list);
+ grpc_closure_init (&calld->async_setup_task, picked_target, calld);
+ grpc_lb_policy_pick (lb_policy, bind_pollset, initial_metadata, &calld->picked_channel, &calld->async_setup_task, closure_list);
- GRPC_LB_POLICY_UNREF(lb_policy, "pick", closure_list);
- } else if (chand->resolver != NULL) {
- calld->state = CALL_WAITING_FOR_CONFIG;
- add_to_lb_policy_wait_queue_locked_state_config(elem);
- if (!chand->started_resolving && chand->resolver != NULL) {
- GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
- chand->started_resolving = 1;
- grpc_resolver_next(chand->resolver,
- &chand->incoming_configuration,
- &chand->on_config_changed, closure_list);
- }
- gpr_mu_unlock(&chand->mu_config);
- gpr_mu_unlock(&calld->mu_state);
- } else {
- calld->state = CALL_CANCELLED;
- gpr_mu_unlock(&chand->mu_config);
- gpr_mu_unlock(&calld->mu_state);
- handle_op_after_cancellation(elem, op, closure_list);
- }
- }
- }
+ GRPC_LB_POLICY_UNREF (lb_policy, "pick", closure_list);
+ }
+ else if (chand->resolver != NULL)
+ {
+ calld->state = CALL_WAITING_FOR_CONFIG;
+ add_to_lb_policy_wait_queue_locked_state_config (elem);
+ if (!chand->started_resolving && chand->resolver != NULL)
+ {
+ GRPC_CHANNEL_INTERNAL_REF (chand->master, "resolver");
+ chand->started_resolving = 1;
+ grpc_resolver_next (chand->resolver, &chand->incoming_configuration, &chand->on_config_changed, closure_list);
+ }
+ gpr_mu_unlock (&chand->mu_config);
+ gpr_mu_unlock (&calld->mu_state);
+ }
+ else
+ {
+ calld->state = CALL_CANCELLED;
+ gpr_mu_unlock (&chand->mu_config);
+ gpr_mu_unlock (&calld->mu_state);
+ handle_op_after_cancellation (elem, op, closure_list);
+ }
+ }
+ }
break;
- }
+ }
}
-static void cc_start_transport_stream_op(grpc_call_element *elem,
- grpc_transport_stream_op *op,
- grpc_closure_list *closure_list) {
- perform_transport_stream_op(elem, op, 0, closure_list);
+static void
+cc_start_transport_stream_op (grpc_call_element * elem, grpc_transport_stream_op * op, grpc_closure_list * closure_list)
+{
+ perform_transport_stream_op (elem, op, 0, closure_list);
}
-static void watch_lb_policy(channel_data *chand, grpc_lb_policy *lb_policy,
- grpc_connectivity_state current_state,
- grpc_closure_list *cl);
+static void watch_lb_policy (channel_data * chand, grpc_lb_policy * lb_policy, grpc_connectivity_state current_state, grpc_closure_list * cl);
-static void on_lb_policy_state_changed_locked(lb_policy_connectivity_watcher *w,
- grpc_closure_list *cl) {
+static void
+on_lb_policy_state_changed_locked (lb_policy_connectivity_watcher * w, grpc_closure_list * cl)
+{
/* check if the notification is for a stale policy */
- if (w->lb_policy != w->chand->lb_policy) return;
+ if (w->lb_policy != w->chand->lb_policy)
+ return;
- grpc_connectivity_state_set(&w->chand->state_tracker, w->state, "lb_changed",
- cl);
- if (w->state != GRPC_CHANNEL_FATAL_FAILURE) {
- watch_lb_policy(w->chand, w->lb_policy, w->state, cl);
- }
+ grpc_connectivity_state_set (&w->chand->state_tracker, w->state, "lb_changed", cl);
+ if (w->state != GRPC_CHANNEL_FATAL_FAILURE)
+ {
+ watch_lb_policy (w->chand, w->lb_policy, w->state, cl);
+ }
}
-static void on_lb_policy_state_changed(void *arg, int iomgr_success,
- grpc_closure_list *closure_list) {
+static void
+on_lb_policy_state_changed (void *arg, int iomgr_success, grpc_closure_list * closure_list)
+{
lb_policy_connectivity_watcher *w = arg;
- gpr_mu_lock(&w->chand->mu_config);
- on_lb_policy_state_changed_locked(w, closure_list);
- gpr_mu_unlock(&w->chand->mu_config);
+ gpr_mu_lock (&w->chand->mu_config);
+ on_lb_policy_state_changed_locked (w, closure_list);
+ gpr_mu_unlock (&w->chand->mu_config);
- GRPC_CHANNEL_INTERNAL_UNREF(w->chand->master, "watch_lb_policy",
- closure_list);
- gpr_free(w);
+ GRPC_CHANNEL_INTERNAL_UNREF (w->chand->master, "watch_lb_policy", closure_list);
+ gpr_free (w);
}
-static void watch_lb_policy(channel_data *chand, grpc_lb_policy *lb_policy,
- grpc_connectivity_state current_state,
- grpc_closure_list *closure_list) {
- lb_policy_connectivity_watcher *w = gpr_malloc(sizeof(*w));
- GRPC_CHANNEL_INTERNAL_REF(chand->master, "watch_lb_policy");
+static void
+watch_lb_policy (channel_data * chand, grpc_lb_policy * lb_policy, grpc_connectivity_state current_state, grpc_closure_list * closure_list)
+{
+ lb_policy_connectivity_watcher *w = gpr_malloc (sizeof (*w));
+ GRPC_CHANNEL_INTERNAL_REF (chand->master, "watch_lb_policy");
w->chand = chand;
- grpc_closure_init(&w->on_changed, on_lb_policy_state_changed, w);
+ grpc_closure_init (&w->on_changed, on_lb_policy_state_changed, w);
w->state = current_state;
w->lb_policy = lb_policy;
- grpc_lb_policy_notify_on_state_change(lb_policy, &w->state, &w->on_changed,
- closure_list);
+ grpc_lb_policy_notify_on_state_change (lb_policy, &w->state, &w->on_changed, closure_list);
}
-static void cc_on_config_changed(void *arg, int iomgr_success,
- grpc_closure_list *closure_list) {
+static void
+cc_on_config_changed (void *arg, int iomgr_success, grpc_closure_list * closure_list)
+{
channel_data *chand = arg;
grpc_lb_policy *lb_policy = NULL;
grpc_lb_policy *old_lb_policy;
@@ -491,298 +534,313 @@
grpc_connectivity_state state = GRPC_CHANNEL_TRANSIENT_FAILURE;
int exit_idle = 0;
- if (chand->incoming_configuration != NULL) {
- lb_policy = grpc_client_config_get_lb_policy(chand->incoming_configuration);
- if (lb_policy != NULL) {
- GRPC_LB_POLICY_REF(lb_policy, "channel");
- GRPC_LB_POLICY_REF(lb_policy, "config_change");
- state = grpc_lb_policy_check_connectivity(lb_policy, closure_list);
- }
+ if (chand->incoming_configuration != NULL)
+ {
+ lb_policy = grpc_client_config_get_lb_policy (chand->incoming_configuration);
+ if (lb_policy != NULL)
+ {
+ GRPC_LB_POLICY_REF (lb_policy, "channel");
+ GRPC_LB_POLICY_REF (lb_policy, "config_change");
+ state = grpc_lb_policy_check_connectivity (lb_policy, closure_list);
+ }
- grpc_client_config_unref(chand->incoming_configuration, closure_list);
- }
+ grpc_client_config_unref (chand->incoming_configuration, closure_list);
+ }
chand->incoming_configuration = NULL;
- gpr_mu_lock(&chand->mu_config);
+ gpr_mu_lock (&chand->mu_config);
old_lb_policy = chand->lb_policy;
chand->lb_policy = lb_policy;
- if (lb_policy != NULL || chand->resolver == NULL /* disconnected */) {
- grpc_closure_list_move(&chand->waiting_for_config_closures, closure_list);
- }
- if (lb_policy != NULL && chand->exit_idle_when_lb_policy_arrives) {
- GRPC_LB_POLICY_REF(lb_policy, "exit_idle");
- exit_idle = 1;
- chand->exit_idle_when_lb_policy_arrives = 0;
- }
-
- if (iomgr_success && chand->resolver) {
- grpc_resolver *resolver = chand->resolver;
- GRPC_RESOLVER_REF(resolver, "channel-next");
- grpc_connectivity_state_set(&chand->state_tracker, state, "new_lb+resolver",
- closure_list);
- if (lb_policy != NULL) {
- watch_lb_policy(chand, lb_policy, state, closure_list);
+ if (lb_policy != NULL || chand->resolver == NULL /* disconnected */ )
+ {
+ grpc_closure_list_move (&chand->waiting_for_config_closures, closure_list);
}
- gpr_mu_unlock(&chand->mu_config);
- GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
- grpc_resolver_next(resolver, &chand->incoming_configuration,
- &chand->on_config_changed, closure_list);
- GRPC_RESOLVER_UNREF(resolver, "channel-next", closure_list);
- } else {
- old_resolver = chand->resolver;
- chand->resolver = NULL;
- grpc_connectivity_state_set(&chand->state_tracker,
- GRPC_CHANNEL_FATAL_FAILURE, "resolver_gone",
- closure_list);
- gpr_mu_unlock(&chand->mu_config);
- if (old_resolver != NULL) {
- grpc_resolver_shutdown(old_resolver, closure_list);
- GRPC_RESOLVER_UNREF(old_resolver, "channel", closure_list);
+ if (lb_policy != NULL && chand->exit_idle_when_lb_policy_arrives)
+ {
+ GRPC_LB_POLICY_REF (lb_policy, "exit_idle");
+ exit_idle = 1;
+ chand->exit_idle_when_lb_policy_arrives = 0;
}
- }
- if (exit_idle) {
- grpc_lb_policy_exit_idle(lb_policy, closure_list);
- GRPC_LB_POLICY_UNREF(lb_policy, "exit_idle", closure_list);
- }
+ if (iomgr_success && chand->resolver)
+ {
+ grpc_resolver *resolver = chand->resolver;
+ GRPC_RESOLVER_REF (resolver, "channel-next");
+ grpc_connectivity_state_set (&chand->state_tracker, state, "new_lb+resolver", closure_list);
+ if (lb_policy != NULL)
+ {
+ watch_lb_policy (chand, lb_policy, state, closure_list);
+ }
+ gpr_mu_unlock (&chand->mu_config);
+ GRPC_CHANNEL_INTERNAL_REF (chand->master, "resolver");
+ grpc_resolver_next (resolver, &chand->incoming_configuration, &chand->on_config_changed, closure_list);
+ GRPC_RESOLVER_UNREF (resolver, "channel-next", closure_list);
+ }
+ else
+ {
+ old_resolver = chand->resolver;
+ chand->resolver = NULL;
+ grpc_connectivity_state_set (&chand->state_tracker, GRPC_CHANNEL_FATAL_FAILURE, "resolver_gone", closure_list);
+ gpr_mu_unlock (&chand->mu_config);
+ if (old_resolver != NULL)
+ {
+ grpc_resolver_shutdown (old_resolver, closure_list);
+ GRPC_RESOLVER_UNREF (old_resolver, "channel", closure_list);
+ }
+ }
- if (old_lb_policy != NULL) {
- grpc_lb_policy_shutdown(old_lb_policy, closure_list);
- GRPC_LB_POLICY_UNREF(old_lb_policy, "channel", closure_list);
- }
+ if (exit_idle)
+ {
+ grpc_lb_policy_exit_idle (lb_policy, closure_list);
+ GRPC_LB_POLICY_UNREF (lb_policy, "exit_idle", closure_list);
+ }
- if (lb_policy != NULL) {
- GRPC_LB_POLICY_UNREF(lb_policy, "config_change", closure_list);
- }
+ if (old_lb_policy != NULL)
+ {
+ grpc_lb_policy_shutdown (old_lb_policy, closure_list);
+ GRPC_LB_POLICY_UNREF (old_lb_policy, "channel", closure_list);
+ }
- GRPC_CHANNEL_INTERNAL_UNREF(chand->master, "resolver", closure_list);
+ if (lb_policy != NULL)
+ {
+ GRPC_LB_POLICY_UNREF (lb_policy, "config_change", closure_list);
+ }
+
+ GRPC_CHANNEL_INTERNAL_UNREF (chand->master, "resolver", closure_list);
}
-static void cc_start_transport_op(grpc_channel_element *elem,
- grpc_transport_op *op,
- grpc_closure_list *closure_list) {
+static void
+cc_start_transport_op (grpc_channel_element * elem, grpc_transport_op * op, grpc_closure_list * closure_list)
+{
grpc_lb_policy *lb_policy = NULL;
channel_data *chand = elem->channel_data;
grpc_resolver *destroy_resolver = NULL;
- grpc_closure_list_add(closure_list, op->on_consumed, 1);
+ grpc_closure_list_add (closure_list, op->on_consumed, 1);
- GPR_ASSERT(op->set_accept_stream == NULL);
- GPR_ASSERT(op->bind_pollset == NULL);
+ GPR_ASSERT (op->set_accept_stream == NULL);
+ GPR_ASSERT (op->bind_pollset == NULL);
- gpr_mu_lock(&chand->mu_config);
- if (op->on_connectivity_state_change != NULL) {
- grpc_connectivity_state_notify_on_state_change(
- &chand->state_tracker, op->connectivity_state,
- op->on_connectivity_state_change, closure_list);
- op->on_connectivity_state_change = NULL;
- op->connectivity_state = NULL;
- }
-
- if (!is_empty(op, sizeof(*op))) {
- lb_policy = chand->lb_policy;
- if (lb_policy) {
- GRPC_LB_POLICY_REF(lb_policy, "broadcast");
+ gpr_mu_lock (&chand->mu_config);
+ if (op->on_connectivity_state_change != NULL)
+ {
+ grpc_connectivity_state_notify_on_state_change (&chand->state_tracker, op->connectivity_state, op->on_connectivity_state_change, closure_list);
+ op->on_connectivity_state_change = NULL;
+ op->connectivity_state = NULL;
}
- }
- if (op->disconnect && chand->resolver != NULL) {
- grpc_connectivity_state_set(&chand->state_tracker,
- GRPC_CHANNEL_FATAL_FAILURE, "disconnect",
- closure_list);
- destroy_resolver = chand->resolver;
- chand->resolver = NULL;
- if (chand->lb_policy != NULL) {
- grpc_lb_policy_shutdown(chand->lb_policy, closure_list);
- GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel", closure_list);
- chand->lb_policy = NULL;
+ if (!is_empty (op, sizeof (*op)))
+ {
+ lb_policy = chand->lb_policy;
+ if (lb_policy)
+ {
+ GRPC_LB_POLICY_REF (lb_policy, "broadcast");
+ }
}
- }
- gpr_mu_unlock(&chand->mu_config);
- if (destroy_resolver) {
- grpc_resolver_shutdown(destroy_resolver, closure_list);
- GRPC_RESOLVER_UNREF(destroy_resolver, "channel", closure_list);
- }
+ if (op->disconnect && chand->resolver != NULL)
+ {
+ grpc_connectivity_state_set (&chand->state_tracker, GRPC_CHANNEL_FATAL_FAILURE, "disconnect", closure_list);
+ destroy_resolver = chand->resolver;
+ chand->resolver = NULL;
+ if (chand->lb_policy != NULL)
+ {
+ grpc_lb_policy_shutdown (chand->lb_policy, closure_list);
+ GRPC_LB_POLICY_UNREF (chand->lb_policy, "channel", closure_list);
+ chand->lb_policy = NULL;
+ }
+ }
+ gpr_mu_unlock (&chand->mu_config);
- if (lb_policy) {
- grpc_lb_policy_broadcast(lb_policy, op, closure_list);
- GRPC_LB_POLICY_UNREF(lb_policy, "broadcast", closure_list);
- }
+ if (destroy_resolver)
+ {
+ grpc_resolver_shutdown (destroy_resolver, closure_list);
+ GRPC_RESOLVER_UNREF (destroy_resolver, "channel", closure_list);
+ }
+
+ if (lb_policy)
+ {
+ grpc_lb_policy_broadcast (lb_policy, op, closure_list);
+ GRPC_LB_POLICY_UNREF (lb_policy, "broadcast", closure_list);
+ }
}
/* Constructor for call_data */
-static void init_call_elem(grpc_call_element *elem,
- const void *server_transport_data,
- grpc_transport_stream_op *initial_op,
- grpc_closure_list *closure_list) {
+static void
+init_call_elem (grpc_call_element * elem, const void *server_transport_data, grpc_transport_stream_op * initial_op, grpc_closure_list * closure_list)
+{
call_data *calld = elem->call_data;
/* TODO(ctiller): is there something useful we can do here? */
- GPR_ASSERT(initial_op == NULL);
+ GPR_ASSERT (initial_op == NULL);
- GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
- GPR_ASSERT(server_transport_data == NULL);
- gpr_mu_init(&calld->mu_state);
+ GPR_ASSERT (elem->filter == &grpc_client_channel_filter);
+ GPR_ASSERT (server_transport_data == NULL);
+ gpr_mu_init (&calld->mu_state);
calld->elem = elem;
calld->state = CALL_CREATED;
- calld->deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
+ calld->deadline = gpr_inf_future (GPR_CLOCK_REALTIME);
}
/* Destructor for call_data */
-static void destroy_call_elem(grpc_call_element *elem,
- grpc_closure_list *closure_list) {
+static void
+destroy_call_elem (grpc_call_element * elem, grpc_closure_list * closure_list)
+{
call_data *calld = elem->call_data;
grpc_subchannel_call *subchannel_call;
/* if the call got activated, we need to destroy the child stack also, and
remove it from the in-flight requests tracked by the child_entry we
picked */
- gpr_mu_lock(&calld->mu_state);
- switch (calld->state) {
+ gpr_mu_lock (&calld->mu_state);
+ switch (calld->state)
+ {
case CALL_ACTIVE:
subchannel_call = calld->subchannel_call;
- gpr_mu_unlock(&calld->mu_state);
- GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call, "client_channel",
- closure_list);
+ gpr_mu_unlock (&calld->mu_state);
+ GRPC_SUBCHANNEL_CALL_UNREF (subchannel_call, "client_channel", closure_list);
break;
case CALL_CREATED:
case CALL_CANCELLED:
- gpr_mu_unlock(&calld->mu_state);
+ gpr_mu_unlock (&calld->mu_state);
break;
case CALL_WAITING_FOR_PICK:
case CALL_WAITING_FOR_CONFIG:
case CALL_WAITING_FOR_CALL:
case CALL_WAITING_FOR_SEND:
- gpr_log(GPR_ERROR, "should never reach here");
- abort();
+ gpr_log (GPR_ERROR, "should never reach here");
+ abort ();
break;
- }
+ }
}
/* Constructor for channel_data */
-static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
- const grpc_channel_args *args,
- grpc_mdctx *metadata_context, int is_first,
- int is_last, grpc_closure_list *closure_list) {
+static void
+init_channel_elem (grpc_channel_element * elem, grpc_channel * master, const grpc_channel_args * args, grpc_mdctx * metadata_context, int is_first, int is_last, grpc_closure_list * closure_list)
+{
channel_data *chand = elem->channel_data;
- memset(chand, 0, sizeof(*chand));
+ memset (chand, 0, sizeof (*chand));
- GPR_ASSERT(is_last);
- GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
+ GPR_ASSERT (is_last);
+ GPR_ASSERT (elem->filter == &grpc_client_channel_filter);
- gpr_mu_init(&chand->mu_config);
+ gpr_mu_init (&chand->mu_config);
chand->mdctx = metadata_context;
chand->master = master;
- grpc_pollset_set_init(&chand->pollset_set);
- grpc_closure_init(&chand->on_config_changed, cc_on_config_changed, chand);
+ grpc_pollset_set_init (&chand->pollset_set);
+ grpc_closure_init (&chand->on_config_changed, cc_on_config_changed, chand);
- grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE,
- "client_channel");
+ grpc_connectivity_state_init (&chand->state_tracker, GRPC_CHANNEL_IDLE, "client_channel");
}
/* Destructor for channel_data */
-static void destroy_channel_elem(grpc_channel_element *elem,
- grpc_closure_list *closure_list) {
+static void
+destroy_channel_elem (grpc_channel_element * elem, grpc_closure_list * closure_list)
+{
channel_data *chand = elem->channel_data;
- if (chand->resolver != NULL) {
- grpc_resolver_shutdown(chand->resolver, closure_list);
- GRPC_RESOLVER_UNREF(chand->resolver, "channel", closure_list);
- }
- if (chand->lb_policy != NULL) {
- GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel", closure_list);
- }
- grpc_connectivity_state_destroy(&chand->state_tracker, closure_list);
- grpc_pollset_set_destroy(&chand->pollset_set);
- gpr_mu_destroy(&chand->mu_config);
+ if (chand->resolver != NULL)
+ {
+ grpc_resolver_shutdown (chand->resolver, closure_list);
+ GRPC_RESOLVER_UNREF (chand->resolver, "channel", closure_list);
+ }
+ if (chand->lb_policy != NULL)
+ {
+ GRPC_LB_POLICY_UNREF (chand->lb_policy, "channel", closure_list);
+ }
+ grpc_connectivity_state_destroy (&chand->state_tracker, closure_list);
+ grpc_pollset_set_destroy (&chand->pollset_set);
+ gpr_mu_destroy (&chand->mu_config);
}
const grpc_channel_filter grpc_client_channel_filter = {
- cc_start_transport_stream_op,
- cc_start_transport_op,
- sizeof(call_data),
- init_call_elem,
- destroy_call_elem,
- sizeof(channel_data),
- init_channel_elem,
- destroy_channel_elem,
- cc_get_peer,
- "client-channel",
+ cc_start_transport_stream_op,
+ cc_start_transport_op,
+ sizeof (call_data),
+ init_call_elem,
+ destroy_call_elem,
+ sizeof (channel_data),
+ init_channel_elem,
+ destroy_channel_elem,
+ cc_get_peer,
+ "client-channel",
};
-void grpc_client_channel_set_resolver(grpc_channel_stack *channel_stack,
- grpc_resolver *resolver,
- grpc_closure_list *closure_list) {
+void
+grpc_client_channel_set_resolver (grpc_channel_stack * channel_stack, grpc_resolver * resolver, grpc_closure_list * closure_list)
+{
/* post construction initialization: set the transport setup pointer */
- grpc_channel_element *elem = grpc_channel_stack_last_element(channel_stack);
+ grpc_channel_element *elem = grpc_channel_stack_last_element (channel_stack);
channel_data *chand = elem->channel_data;
- gpr_mu_lock(&chand->mu_config);
- GPR_ASSERT(!chand->resolver);
+ gpr_mu_lock (&chand->mu_config);
+ GPR_ASSERT (!chand->resolver);
chand->resolver = resolver;
- GRPC_RESOLVER_REF(resolver, "channel");
- if (!grpc_closure_list_empty(chand->waiting_for_config_closures) ||
- chand->exit_idle_when_lb_policy_arrives) {
- chand->started_resolving = 1;
- GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
- grpc_resolver_next(resolver, &chand->incoming_configuration,
- &chand->on_config_changed, closure_list);
- }
- gpr_mu_unlock(&chand->mu_config);
+ GRPC_RESOLVER_REF (resolver, "channel");
+ if (!grpc_closure_list_empty (chand->waiting_for_config_closures) || chand->exit_idle_when_lb_policy_arrives)
+ {
+ chand->started_resolving = 1;
+ GRPC_CHANNEL_INTERNAL_REF (chand->master, "resolver");
+ grpc_resolver_next (resolver, &chand->incoming_configuration, &chand->on_config_changed, closure_list);
+ }
+ gpr_mu_unlock (&chand->mu_config);
}
-grpc_connectivity_state grpc_client_channel_check_connectivity_state(
- grpc_channel_element *elem, int try_to_connect,
- grpc_closure_list *closure_list) {
+grpc_connectivity_state
+grpc_client_channel_check_connectivity_state (grpc_channel_element * elem, int try_to_connect, grpc_closure_list * closure_list)
+{
channel_data *chand = elem->channel_data;
grpc_connectivity_state out;
- gpr_mu_lock(&chand->mu_config);
- out = grpc_connectivity_state_check(&chand->state_tracker);
- if (out == GRPC_CHANNEL_IDLE && try_to_connect) {
- if (chand->lb_policy != NULL) {
- grpc_lb_policy_exit_idle(chand->lb_policy, closure_list);
- } else {
- chand->exit_idle_when_lb_policy_arrives = 1;
- if (!chand->started_resolving && chand->resolver != NULL) {
- GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
- chand->started_resolving = 1;
- grpc_resolver_next(chand->resolver, &chand->incoming_configuration,
- &chand->on_config_changed, closure_list);
- }
+ gpr_mu_lock (&chand->mu_config);
+ out = grpc_connectivity_state_check (&chand->state_tracker);
+ if (out == GRPC_CHANNEL_IDLE && try_to_connect)
+ {
+ if (chand->lb_policy != NULL)
+ {
+ grpc_lb_policy_exit_idle (chand->lb_policy, closure_list);
+ }
+ else
+ {
+ chand->exit_idle_when_lb_policy_arrives = 1;
+ if (!chand->started_resolving && chand->resolver != NULL)
+ {
+ GRPC_CHANNEL_INTERNAL_REF (chand->master, "resolver");
+ chand->started_resolving = 1;
+ grpc_resolver_next (chand->resolver, &chand->incoming_configuration, &chand->on_config_changed, closure_list);
+ }
+ }
}
- }
- gpr_mu_unlock(&chand->mu_config);
+ gpr_mu_unlock (&chand->mu_config);
return out;
}
-void grpc_client_channel_watch_connectivity_state(
- grpc_channel_element *elem, grpc_connectivity_state *state,
- grpc_closure *on_complete, grpc_closure_list *closure_list) {
+void
+grpc_client_channel_watch_connectivity_state (grpc_channel_element * elem, grpc_connectivity_state * state, grpc_closure * on_complete, grpc_closure_list * closure_list)
+{
channel_data *chand = elem->channel_data;
- gpr_mu_lock(&chand->mu_config);
- grpc_connectivity_state_notify_on_state_change(&chand->state_tracker, state,
- on_complete, closure_list);
- gpr_mu_unlock(&chand->mu_config);
+ gpr_mu_lock (&chand->mu_config);
+ grpc_connectivity_state_notify_on_state_change (&chand->state_tracker, state, on_complete, closure_list);
+ gpr_mu_unlock (&chand->mu_config);
}
-grpc_pollset_set *grpc_client_channel_get_connecting_pollset_set(
- grpc_channel_element *elem) {
+grpc_pollset_set *
+grpc_client_channel_get_connecting_pollset_set (grpc_channel_element * elem)
+{
channel_data *chand = elem->channel_data;
return &chand->pollset_set;
}
-void grpc_client_channel_add_interested_party(grpc_channel_element *elem,
- grpc_pollset *pollset,
- grpc_closure_list *closure_list) {
+void
+grpc_client_channel_add_interested_party (grpc_channel_element * elem, grpc_pollset * pollset, grpc_closure_list * closure_list)
+{
channel_data *chand = elem->channel_data;
- grpc_pollset_set_add_pollset(&chand->pollset_set, pollset, closure_list);
+ grpc_pollset_set_add_pollset (&chand->pollset_set, pollset, closure_list);
}
-void grpc_client_channel_del_interested_party(grpc_channel_element *elem,
- grpc_pollset *pollset,
- grpc_closure_list *closure_list) {
+void
+grpc_client_channel_del_interested_party (grpc_channel_element * elem, grpc_pollset * pollset, grpc_closure_list * closure_list)
+{
channel_data *chand = elem->channel_data;
- grpc_pollset_set_del_pollset(&chand->pollset_set, pollset, closure_list);
+ grpc_pollset_set_del_pollset (&chand->pollset_set, pollset, closure_list);
}