1 files changed, 178 insertions, 21 deletions

diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index c1aa580b2d..a293c93ec6 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -40,7 +40,6 @@
 #include "src/core/channel/connected_channel.h"
 #include "src/core/surface/channel.h"
 #include "src/core/iomgr/iomgr.h"
-#include "src/core/iomgr/pollset_set.h"
 #include "src/core/support/string.h"
 #include "src/core/transport/connectivity_state.h"
 #include <grpc/support/alloc.h>
@@ -57,6 +56,8 @@ typedef struct {
   grpc_mdctx *mdctx;
   /** resolver for this channel */
   grpc_resolver *resolver;
+  /** have we started resolving this channel */
+  int started_resolving;
   /** 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
@@ -77,8 +78,22 @@ typedef struct {
   grpc_iomgr_closure on_config_changed;
   /** connectivity state being tracked */
   grpc_connectivity_state_tracker state_tracker;
+  /** when an lb_policy arrives, should we try to exit idle */
+  int exit_idle_when_lb_policy_arrives;
+  /** pollset_set of interested parties in a new connection */
+  grpc_pollset_set pollset_set;
 } channel_data;
 
+/** We create one watcher for each new lb_policy that is returned from a resolver,
+    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 {
+  channel_data *chand;
+  grpc_iomgr_closure on_changed;
+  grpc_connectivity_state state;
+  grpc_lb_policy *lb_policy;
+} lb_policy_connectivity_watcher;
+
 typedef enum {
   CALL_CREATED,
   CALL_WAITING_FOR_SEND,
@@ -265,6 +280,26 @@ static grpc_iomgr_closure *merge_into_waiting_op(
   return consumed_op;
 }
 
+static char *cc_get_peer(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(subchannel_call);
+    GRPC_SUBCHANNEL_CALL_UNREF(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_call_element *elem,
                                         grpc_transport_stream_op *op,
                                         int continuation) {
@@ -365,6 +400,13 @@ static void perform_transport_stream_op(grpc_call_element *elem,
           } 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);
+            }
             gpr_mu_unlock(&chand->mu_config);
             gpr_mu_unlock(&calld->mu_state);
           } else {
@@ -388,16 +430,53 @@ static void cc_start_transport_stream_op(grpc_call_element *elem,
   perform_transport_stream_op(elem, op, 0);
 }
 
+static void watch_lb_policy(channel_data *chand, grpc_lb_policy *lb_policy, grpc_connectivity_state current_state);
+
+static void on_lb_policy_state_changed(void *arg, int iomgr_success) {
+  lb_policy_connectivity_watcher *w = arg;
+
+  gpr_mu_lock(&w->chand->mu_config);
+  /* check if the notification is for a stale policy */
+  if (w->lb_policy == w->chand->lb_policy) {
+    grpc_connectivity_state_set(&w->chand->state_tracker, w->state,
+                                "lb_changed");
+    if (w->state != GRPC_CHANNEL_FATAL_FAILURE) {
+      watch_lb_policy(w->chand, w->lb_policy, w->state);
+    }
+  }
+  gpr_mu_unlock(&w->chand->mu_config);
+
+  GRPC_CHANNEL_INTERNAL_UNREF(w->chand->master, "watch_lb_policy");
+  gpr_free(w);
+}
+
+static void watch_lb_policy(channel_data *chand, grpc_lb_policy *lb_policy, grpc_connectivity_state current_state) {
+  lb_policy_connectivity_watcher *w = gpr_malloc(sizeof(*w));
+  GRPC_CHANNEL_INTERNAL_REF(chand->master, "watch_lb_policy");
+
+  w->chand = chand;
+  grpc_iomgr_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);
+}
+
 static void cc_on_config_changed(void *arg, int iomgr_success) {
   channel_data *chand = arg;
   grpc_lb_policy *lb_policy = NULL;
   grpc_lb_policy *old_lb_policy;
   grpc_resolver *old_resolver;
   grpc_iomgr_closure *wakeup_closures = NULL;
+  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);
-    GRPC_LB_POLICY_REF(lb_policy, "channel");
+    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);
+    }
 
     grpc_client_config_unref(chand->incoming_configuration);
   }
@@ -411,13 +490,12 @@ static void cc_on_config_changed(void *arg, int iomgr_success) {
     wakeup_closures = chand->waiting_for_config_closures;
     chand->waiting_for_config_closures = NULL;
   }
-  gpr_mu_unlock(&chand->mu_config);
-
-  if (old_lb_policy) {
-    GRPC_LB_POLICY_UNREF(old_lb_policy, "channel");
+  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;
   }
 
-  gpr_mu_lock(&chand->mu_config);
   if (iomgr_success && chand->resolver) {
     grpc_resolver *resolver = chand->resolver;
     GRPC_RESOLVER_REF(resolver, "channel-next");
@@ -426,11 +504,16 @@ static void cc_on_config_changed(void *arg, int iomgr_success) {
     grpc_resolver_next(resolver, &chand->incoming_configuration,
                        &chand->on_config_changed);
     GRPC_RESOLVER_UNREF(resolver, "channel-next");
+    grpc_connectivity_state_set(&chand->state_tracker, state,
+                                "new_lb+resolver");
+    if (lb_policy != NULL) {
+      watch_lb_policy(chand, lb_policy, state);
+    }
   } else {
     old_resolver = chand->resolver;
     chand->resolver = NULL;
     grpc_connectivity_state_set(&chand->state_tracker,
-                                GRPC_CHANNEL_FATAL_FAILURE);
+                                GRPC_CHANNEL_FATAL_FAILURE, "resolver_gone");
     gpr_mu_unlock(&chand->mu_config);
     if (old_resolver != NULL) {
       grpc_resolver_shutdown(old_resolver);
@@ -438,12 +521,24 @@ static void cc_on_config_changed(void *arg, int iomgr_success) {
     }
   }
 
+  if (exit_idle) {
+    grpc_lb_policy_exit_idle(lb_policy);
+    GRPC_LB_POLICY_UNREF(lb_policy, "exit_idle");
+  }
+
+  if (old_lb_policy != NULL) {
+    GRPC_LB_POLICY_UNREF(old_lb_policy, "channel");
+  }
+
   while (wakeup_closures) {
     grpc_iomgr_closure *next = wakeup_closures->next;
-    grpc_iomgr_add_callback(wakeup_closures);
+    wakeup_closures->cb(wakeup_closures->cb_arg, 1);
     wakeup_closures = next;
   }
 
+  if (lb_policy != NULL) {
+    GRPC_LB_POLICY_UNREF(lb_policy, "config_change");
+  }
   GRPC_CHANNEL_INTERNAL_UNREF(chand->master, "resolver");
 }
 
@@ -467,20 +562,22 @@ static void cc_start_transport_op(grpc_channel_element *elem,
     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");
+    }
+  }
+
   if (op->disconnect && chand->resolver != NULL) {
     grpc_connectivity_state_set(&chand->state_tracker,
-                                GRPC_CHANNEL_FATAL_FAILURE);
+                                GRPC_CHANNEL_FATAL_FAILURE, "disconnect");
     destroy_resolver = chand->resolver;
     chand->resolver = NULL;
     if (chand->lb_policy != NULL) {
       grpc_lb_policy_shutdown(chand->lb_policy);
-    }
-  }
-
-  if (!is_empty(op, sizeof(*op))) {
-    lb_policy = chand->lb_policy;
-    if (lb_policy) {
-      GRPC_LB_POLICY_REF(lb_policy, "broadcast");
+      GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel");
+      chand->lb_policy = NULL;
     }
   }
   gpr_mu_unlock(&chand->mu_config);
@@ -561,10 +658,11 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master,
   gpr_mu_init(&chand->mu_config);
   chand->mdctx = metadata_context;
   chand->master = master;
+  grpc_pollset_set_init(&chand->pollset_set);
   grpc_iomgr_closure_init(&chand->on_config_changed, cc_on_config_changed,
                           chand);
 
-  grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE);
+  grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE, "client_channel");
 }
 
 /* Destructor for channel_data */
@@ -578,6 +676,8 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
   if (chand->lb_policy != NULL) {
     GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel");
   }
+  grpc_connectivity_state_destroy(&chand->state_tracker);
+  grpc_pollset_set_destroy(&chand->pollset_set);
   gpr_mu_destroy(&chand->mu_config);
 }
 
@@ -590,6 +690,7 @@ const grpc_channel_filter grpc_client_channel_filter = {
     sizeof(channel_data),
     init_channel_elem,
     destroy_channel_elem,
+    cc_get_peer,
     "client-channel",
 };
 
@@ -598,10 +699,66 @@ void grpc_client_channel_set_resolver(grpc_channel_stack *channel_stack,
   /* 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_mu_lock(&chand->mu_config);
   GPR_ASSERT(!chand->resolver);
   chand->resolver = resolver;
-  GRPC_CHANNEL_INTERNAL_REF(chand->master, "resolver");
   GRPC_RESOLVER_REF(resolver, "channel");
-  grpc_resolver_next(resolver, &chand->incoming_configuration,
-                     &chand->on_config_changed);
+  if (chand->waiting_for_config_closures != NULL ||
+      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);
+  }
+  gpr_mu_unlock(&chand->mu_config);
+}
+
+grpc_connectivity_state grpc_client_channel_check_connectivity_state(
+    grpc_channel_element *elem, int try_to_connect) {
+  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);
+    } 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);
+      }
+    }
+  }
+  gpr_mu_unlock(&chand->mu_config);
+  return out;
+}
+
+void grpc_client_channel_watch_connectivity_state(
+    grpc_channel_element *elem, grpc_connectivity_state *state,
+    grpc_iomgr_closure *on_complete) {
+  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);
+  gpr_mu_unlock(&chand->mu_config);
+}
+
+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) {
+  channel_data *chand = elem->channel_data;
+  grpc_pollset_set_add_pollset(&chand->pollset_set, pollset);
+}
+
+void grpc_client_channel_del_interested_party(grpc_channel_element *elem,
+                                          grpc_pollset *pollset) {
+  channel_data *chand = elem->channel_data;
+  grpc_pollset_set_del_pollset(&chand->pollset_set, pollset);
 }