aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/channel/client_channel.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/channel/client_channel.c')
-rw-r--r--src/core/channel/client_channel.c494
1 files changed, 102 insertions, 392 deletions
diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 9f85557ea1..16d91d4277 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -43,6 +43,7 @@
#include "src/core/channel/channel_args.h"
#include "src/core/channel/connected_channel.h"
+#include "src/core/channel/subchannel_call_holder.h"
#include "src/core/iomgr/iomgr.h"
#include "src/core/profiling/timers.h"
#include "src/core/support/string.h"
@@ -51,7 +52,7 @@
/* Client channel implementation */
-typedef struct call_data call_data;
+typedef grpc_subchannel_call_holder call_data;
typedef struct client_channel_channel_data {
/** metadata context for this channel */
@@ -98,360 +99,22 @@ typedef struct {
grpc_lb_policy *lb_policy;
} lb_policy_connectivity_watcher;
-typedef enum {
- CALL_CREATED,
- CALL_WAITING_FOR_SEND,
- CALL_WAITING_FOR_CONFIG,
- CALL_WAITING_FOR_PICK,
- CALL_WAITING_FOR_CALL,
- CALL_ACTIVE,
- CALL_CANCELLED
-} call_state;
-
-struct call_data {
- /* owning element */
- grpc_call_element *elem;
-
- gpr_mu mu_state;
-
- call_state state;
- gpr_timespec deadline;
- grpc_subchannel *picked_channel;
- grpc_closure async_setup_task;
- grpc_transport_stream_op waiting_op;
- /* our child call stack */
- grpc_subchannel_call *subchannel_call;
- grpc_linked_mdelem status;
- 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 void handle_op_after_cancellation(grpc_exec_ctx *exec_ctx,
- grpc_call_element *elem,
- grpc_transport_stream_op *op) {
- 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(exec_ctx, op->on_done_send->cb_arg, 0);
- }
- 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(exec_ctx, op->on_done_recv->cb_arg, 1);
- }
- if (op->on_consumed) {
- op->on_consumed->cb(exec_ctx, op->on_consumed->cb_arg, 0);
- }
-}
-
typedef struct {
grpc_closure closure;
grpc_call_element *elem;
} waiting_call;
-static void perform_transport_stream_op(grpc_exec_ctx *exec_ctx,
- grpc_call_element *elem,
- grpc_transport_stream_op *op,
- int continuation);
-
-static void continue_with_pick(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
- waiting_call *wc = arg;
- call_data *calld = wc->elem->call_data;
- perform_transport_stream_op(exec_ctx, wc->elem, &calld->waiting_op, 1);
- gpr_free(wc);
-}
-
-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);
- wc->elem = elem;
- grpc_closure_list_add(&chand->waiting_for_config_closures, &wc->closure, 1);
-}
-
-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;
- }
- return 1;
-}
-
-static void started_call_locked(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
- call_data *calld = arg;
- grpc_transport_stream_op op;
- int have_waiting;
-
- 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(exec_ctx, calld->subchannel_call, &op);
- } 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(exec_ctx, calld->subchannel_call,
- &calld->waiting_op);
- }
- } else {
- calld->state = CALL_CANCELLED;
- gpr_mu_unlock(&calld->mu_state);
- if (have_waiting) {
- handle_op_after_cancellation(exec_ctx, calld->elem, &calld->waiting_op);
- }
- }
- } else {
- GPR_ASSERT(calld->state == CALL_CANCELLED);
- gpr_mu_unlock(&calld->mu_state);
- }
-}
-
-static void started_call(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
- call_data *calld = arg;
- gpr_mu_lock(&calld->mu_state);
- started_call_locked(exec_ctx, arg, iomgr_success);
-}
-
-static void picked_target(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
- call_data *calld = arg;
- grpc_pollset *pollset;
- grpc_subchannel_call_create_status call_creation_status;
-
- GPR_TIMER_BEGIN("picked_target", 0);
-
- if (calld->picked_channel == NULL) {
- /* treat this like a cancellation */
- calld->waiting_op.cancel_with_status = GRPC_STATUS_UNAVAILABLE;
- perform_transport_stream_op(exec_ctx, calld->elem, &calld->waiting_op, 1);
- } else {
- gpr_mu_lock(&calld->mu_state);
- if (calld->state == CALL_CANCELLED) {
- gpr_mu_unlock(&calld->mu_state);
- handle_op_after_cancellation(exec_ctx, calld->elem, &calld->waiting_op);
- } else {
- GPR_ASSERT(calld->state == CALL_WAITING_FOR_PICK);
- calld->state = CALL_WAITING_FOR_CALL;
- pollset = calld->waiting_op.bind_pollset;
- grpc_closure_init(&calld->async_setup_task, started_call, calld);
- call_creation_status = grpc_subchannel_create_call(
- exec_ctx, calld->picked_channel, pollset, &calld->subchannel_call,
- &calld->async_setup_task);
- if (call_creation_status == GRPC_SUBCHANNEL_CALL_CREATE_READY) {
- started_call_locked(exec_ctx, calld, iomgr_success);
- } else {
- gpr_mu_unlock(&calld->mu_state);
- }
- }
- }
-
- GPR_TIMER_END("picked_target", 0);
-}
-
-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;
- }
- 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_exec_ctx *exec_ctx, grpc_call_element *elem) {
- 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(exec_ctx, subchannel_call);
- GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, subchannel_call, "get_peer");
- return result;
- } else {
- gpr_mu_unlock(&calld->mu_state);
- return grpc_channel_get_target(chand->master);
- }
-}
-
-static void perform_transport_stream_op(grpc_exec_ctx *exec_ctx,
- grpc_call_element *elem,
- grpc_transport_stream_op *op,
- int continuation) {
- 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_TIMER_BEGIN("perform_transport_stream_op", 0);
- 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) {
- case CALL_ACTIVE:
- GPR_ASSERT(!continuation);
- subchannel_call = calld->subchannel_call;
- gpr_mu_unlock(&calld->mu_state);
- grpc_subchannel_call_process_op(exec_ctx, subchannel_call, op);
- break;
- case CALL_CANCELLED:
- gpr_mu_unlock(&calld->mu_state);
- handle_op_after_cancellation(exec_ctx, elem, op);
- break;
- case CALL_WAITING_FOR_SEND:
- GPR_ASSERT(!continuation);
- grpc_exec_ctx_enqueue(exec_ctx, 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));
- continuation = 1;
- /* 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(exec_ctx, elem, op);
- handle_op_after_cancellation(exec_ctx, elem, &op2);
- } else {
- grpc_exec_ctx_enqueue(exec_ctx, 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(exec_ctx, elem, op);
- } 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 *waiting_op = &calld->waiting_op;
- grpc_pollset *bind_pollset = waiting_op->bind_pollset;
- grpc_metadata_batch *initial_metadata =
- &waiting_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(waiting_op->bind_pollset);
- GPR_ASSERT(waiting_op->send_ops);
- GPR_ASSERT(waiting_op->send_ops->nops >= 1);
- GPR_ASSERT(waiting_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(exec_ctx, lb_policy, bind_pollset,
- initial_metadata, &calld->picked_channel,
- &calld->async_setup_task);
-
- GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "pick");
- } 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(exec_ctx, chand->resolver,
- &chand->incoming_configuration,
- &chand->on_config_changed);
- }
- 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(exec_ctx, elem, op);
- }
- }
- }
- break;
- }
-
- GPR_TIMER_END("perform_transport_stream_op", 0);
+ return grpc_subchannel_call_holder_get_peer(exec_ctx, elem->call_data,
+ chand->master);
}
static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
grpc_transport_stream_op *op) {
- perform_transport_stream_op(exec_ctx, elem, op, 0);
+ GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
+ grpc_subchannel_call_holder_perform_op(exec_ctx, elem->call_data, op);
}
static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand,
@@ -593,11 +256,9 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
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");
- }
+ lb_policy = chand->lb_policy;
+ if (lb_policy) {
+ GRPC_LB_POLICY_REF(lb_policy, "broadcast");
}
if (op->disconnect && chand->resolver != NULL) {
@@ -624,67 +285,110 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
}
}
-/* Constructor for call_data */
-static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
- const void *server_transport_data,
- grpc_transport_stream_op *initial_op) {
+typedef struct {
+ grpc_metadata_batch *initial_metadata;
+ grpc_subchannel **subchannel;
+ grpc_closure *on_ready;
+ grpc_call_element *elem;
+ grpc_closure closure;
+} continue_picking_args;
+
+static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg,
+ grpc_metadata_batch *initial_metadata,
+ grpc_subchannel **subchannel,
+ grpc_closure *on_ready);
+
+static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+ continue_picking_args *cpa = arg;
+ if (!success) {
+ grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 0);
+ } else if (cpa->subchannel == NULL) {
+ /* cancelled, do nothing */
+ } else if (cc_pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata,
+ cpa->subchannel, cpa->on_ready)) {
+ grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 1);
+ }
+ gpr_free(cpa);
+}
+
+static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
+ grpc_metadata_batch *initial_metadata,
+ grpc_subchannel **subchannel,
+ grpc_closure *on_ready) {
+ grpc_call_element *elem = elemp;
+ channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
+ continue_picking_args *cpa;
+ grpc_closure *closure;
- /* TODO(ctiller): is there something useful we can do here? */
- GPR_ASSERT(initial_op == NULL);
+ GPR_ASSERT(subchannel);
- 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);
+ gpr_mu_lock(&chand->mu_config);
+ if (initial_metadata == NULL) {
+ if (chand->lb_policy != NULL) {
+ grpc_lb_policy_cancel_pick(exec_ctx, chand->lb_policy, subchannel);
+ }
+ for (closure = chand->waiting_for_config_closures.head; closure != NULL;
+ closure = grpc_closure_next(closure)) {
+ cpa = closure->cb_arg;
+ if (cpa->subchannel == subchannel) {
+ cpa->subchannel = NULL;
+ grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 0);
+ }
+ }
+ gpr_mu_unlock(&chand->mu_config);
+ return 1;
+ }
+ if (chand->lb_policy != NULL) {
+ int r = grpc_lb_policy_pick(exec_ctx, chand->lb_policy, calld->pollset,
+ initial_metadata, subchannel, on_ready);
+ gpr_mu_unlock(&chand->mu_config);
+ return r;
+ }
+ if (chand->resolver != NULL && !chand->started_resolving) {
+ chand->started_resolving = 1;
+ GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
+ grpc_resolver_next(exec_ctx, chand->resolver,
+ &chand->incoming_configuration,
+ &chand->on_config_changed);
+ }
+ cpa = gpr_malloc(sizeof(*cpa));
+ cpa->initial_metadata = initial_metadata;
+ cpa->subchannel = subchannel;
+ cpa->on_ready = on_ready;
+ cpa->elem = elem;
+ grpc_closure_init(&cpa->closure, continue_picking, cpa);
+ grpc_closure_list_add(&chand->waiting_for_config_closures, &cpa->closure, 1);
+ gpr_mu_unlock(&chand->mu_config);
+ return 0;
+}
+
+/* Constructor for call_data */
+static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+ grpc_call_element_args *args) {
+ grpc_subchannel_call_holder_init(elem->call_data, cc_pick_subchannel, elem);
}
/* Destructor for call_data */
static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem) {
- 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) {
- case CALL_ACTIVE:
- subchannel_call = calld->subchannel_call;
- gpr_mu_unlock(&calld->mu_state);
- GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, subchannel_call, "client_channel");
- break;
- case CALL_CREATED:
- case CALL_CANCELLED:
- 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_UNREACHABLE_CODE(return );
- }
+ grpc_subchannel_call_holder_destroy(exec_ctx, elem->call_data);
}
/* Constructor for channel_data */
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
- grpc_channel_element *elem, grpc_channel *master,
- const grpc_channel_args *args,
- grpc_mdctx *metadata_context, int is_first,
- int is_last) {
+ grpc_channel_element *elem,
+ grpc_channel_element_args *args) {
channel_data *chand = elem->channel_data;
memset(chand, 0, sizeof(*chand));
- GPR_ASSERT(is_last);
+ GPR_ASSERT(args->is_last);
GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
gpr_mu_init(&chand->mu_config);
- chand->mdctx = metadata_context;
- chand->master = master;
+ chand->mdctx = args->metadata_context;
+ chand->master = args->master;
grpc_pollset_set_init(&chand->pollset_set);
grpc_closure_init(&chand->on_config_changed, cc_on_config_changed, chand);
@@ -709,10 +413,16 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
gpr_mu_destroy(&chand->mu_config);
}
+static void cc_set_pollset(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+ grpc_pollset *pollset) {
+ call_data *calld = elem->call_data;
+ calld->pollset = pollset;
+}
+
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",
+ init_call_elem, cc_set_pollset, destroy_call_elem, sizeof(channel_data),
+ init_channel_elem, destroy_channel_elem, cc_get_peer, "client-channel",
};
void grpc_client_channel_set_resolver(grpc_exec_ctx *exec_ctx,