Refactoring progress towards integrating client configs
diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 726196e..ea79d53 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -36,7 +36,6 @@
#include <stdio.h>
#include "src/core/channel/channel_args.h"
-#include "src/core/channel/child_channel.h"
#include "src/core/channel/connected_channel.h"
#include "src/core/iomgr/iomgr.h"
#include "src/core/iomgr/pollset_set.h"
@@ -51,26 +50,25 @@
typedef struct call_data call_data;
typedef struct {
- /* protects children, child_count, child_capacity, active_child,
- transport_setup_initiated
- does not protect channel stacks held by children
- transport_setup is assumed to be set once during construction */
- gpr_mu mu;
-
- /* the sending child (may be null) */
- grpc_child_channel *active_child;
+ /** metadata context for this channel */
grpc_mdctx *mdctx;
-
- /* calls waiting for a channel to be ready */
- call_data **waiting_children;
- size_t waiting_child_count;
- size_t waiting_child_capacity;
-
- /* transport setup for this channel */
- grpc_transport_setup *transport_setup;
- int transport_setup_initiated;
-
+ /** resolver for this channel */
+ grpc_resolver *resolver;
+ /** channel arguments for this channel
+ TODO(ctiller): still needed? */
grpc_channel_args *args;
+
+ /** mutex protecting waiting list */
+ gpr_mu mu_waiting;
+ /** mutex protecting client configuration, resolution state */
+ gpr_mu mu_config;
+
+ /** currently active load balancer - guarded by mu_config */
+ grpc_lb_policy *lb_policy;
+
+ /** incoming configuration - set by resolver.next
+ guarded by mu_config */
+ grpc_client_config *incoming_configuration;
} channel_data;
typedef enum {
@@ -84,12 +82,14 @@
/* owning element */
grpc_call_element *elem;
+ gpr_mu mu_state;
+
call_state state;
gpr_timespec deadline;
union {
struct {
/* our child call stack */
- grpc_child_call *child_call;
+ grpc_subchannel_call *subchannel_call;
} active;
grpc_transport_op waiting_op;
struct {
@@ -99,6 +99,7 @@
} s;
};
+#if 0
static int prepare_activate(grpc_call_element *elem,
grpc_child_channel *on_child) {
call_data *calld = elem->call_data;
@@ -150,6 +151,7 @@
new_count == chand->waiting_child_count);
chand->waiting_child_count = new_count;
}
+#endif
static void handle_op_after_cancellation(grpc_call_element *elem,
grpc_transport_op *op) {
@@ -183,15 +185,99 @@
}
}
+static void add_to_lb_policy_wait_queue_locked_state_config(channel_data *chand, call_data *calld) {
+ abort();
+}
+
+static void pick_target(grpc_lb_policy *lb_policy, call_data *calld) {
+ abort();
+}
+
static void cc_start_transport_op(grpc_call_element *elem,
grpc_transport_op *op) {
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
- grpc_call_element *child_elem;
+ grpc_subchannel_call *subchannel_call;
+ grpc_lb_policy *lb_policy;
grpc_transport_op waiting_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) {
+ case CALL_ACTIVE:
+ subchannel_call = calld->s.active.subchannel_call;
+ grpc_subchannel_call_ref(subchannel_call);
+ gpr_mu_unlock(&calld->mu_state);
+ grpc_subchannel_call_process_op(subchannel_call, op);
+ grpc_subchannel_call_unref(subchannel_call);
+ break;
+ case CALL_CANCELLED:
+ gpr_mu_unlock(&calld->mu_state);
+ handle_op_after_cancellation(elem, op);
+ break;
+ 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);
+ } else {
+ calld->state = CALL_WAITING;
+ calld->s.waiting_op = *op;
+
+ gpr_mu_lock(&chand->mu_config);
+ lb_policy = chand->lb_policy;
+ if (lb_policy) {
+ grpc_lb_policy_ref(lb_policy);
+ gpr_mu_unlock(&chand->mu_config);
+ gpr_mu_unlock(&calld->mu_state);
+
+ pick_target(lb_policy, calld);
+
+ grpc_lb_policy_unref(lb_policy);
+ } else {
+ add_to_lb_policy_wait_queue_locked_state_config(chand, calld);
+ gpr_mu_unlock(&chand->mu_config);
+ gpr_mu_unlock(&calld->mu_state);
+ }
+ }
+ break;
+ case CALL_WAITING:
+ if (op->cancel_with_status != GRPC_STATUS_OK) {
+ waiting_op = calld->s.waiting_op;
+ calld->state = CALL_CANCELLED;
+ gpr_mu_unlock(&calld->mu_state);
+ handle_op_after_cancellation(elem, &waiting_op);
+ handle_op_after_cancellation(elem, op);
+ } else {
+ GPR_ASSERT((calld->s.waiting_op.send_ops == NULL) !=
+ (op->send_ops == NULL));
+ GPR_ASSERT((calld->s.waiting_op.recv_ops == NULL) !=
+ (op->recv_ops == NULL));
+ if (op->send_ops) {
+ calld->s.waiting_op.send_ops = op->send_ops;
+ calld->s.waiting_op.is_last_send = op->is_last_send;
+ calld->s.waiting_op.on_done_send = op->on_done_send;
+ calld->s.waiting_op.send_user_data = op->send_user_data;
+ }
+ if (op->recv_ops) {
+ calld->s.waiting_op.recv_ops = op->recv_ops;
+ calld->s.waiting_op.recv_state = op->recv_state;
+ calld->s.waiting_op.on_done_recv = op->on_done_recv;
+ calld->s.waiting_op.recv_user_data = op->recv_user_data;
+ }
+ gpr_mu_unlock(&calld->mu_state);
+ if (op->on_consumed) {
+ op->on_consumed(op->on_consumed_user_data, 0);
+ }
+ }
+ break;
+ }
+
+
+
+
+#if 0
gpr_mu_lock(&chand->mu);
switch (calld->state) {
case CALL_ACTIVE:
@@ -285,6 +371,7 @@
handle_op_after_cancellation(elem, op);
break;
}
+#endif
}
static void channel_op(grpc_channel_element *elem,
@@ -536,11 +623,12 @@
return result;
}
-void grpc_client_channel_set_transport_setup(grpc_channel_stack *channel_stack,
- grpc_transport_setup *setup) {
+void grpc_client_channel_set_resolver(grpc_channel_stack *channel_stack,
+ grpc_resolver *resolver) {
/* post construction initialization: set the transport setup pointer */
grpc_channel_element *elem = grpc_channel_stack_last_element(channel_stack);
channel_data *chand = elem->channel_data;
- GPR_ASSERT(!chand->transport_setup);
- chand->transport_setup = setup;
+ GPR_ASSERT(!chand->resolver);
+ chand->resolver = resolver;
+ grpc_resolver_next(resolver, &chand->incoming_configuration, &chand->on_config_changed);
}