Added connectivity tests, fixed bugs

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>
@@ -79,8 +78,20 @@ typedef struct {
   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,
@@ -394,17 +405,61 @@ 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;
+  int start_new = 0;
+
+  gpr_log(GPR_DEBUG, "on_lb_policy_state_changed: %p %d", w->lb_policy, w->state);
+
+  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);
+    start_new = 1;
+  } else {
+    gpr_log(GPR_DEBUG, "stale state change: %p vs %p", w->lb_policy, w->chand->lb_policy);
+  }
+  gpr_mu_unlock(&w->chand->mu_config);
+
+  if (start_new) {
+    watch_lb_policy(w->chand, w->lb_policy, w->state);
+  }
+
+  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");
+
+  gpr_log(GPR_DEBUG, "watch_lb_policy: %p %d", lb_policy, current_state);
+
+  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);
+    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);
   }
@@ -423,18 +478,7 @@ static void cc_on_config_changed(void *arg, int iomgr_success) {
     exit_idle = 1;
     chand->exit_idle_when_lb_policy_arrives = 0;
   }
-  gpr_mu_unlock(&chand->mu_config);
-
-  if (exit_idle) {
-    grpc_lb_policy_exit_idle(lb_policy);
-    GRPC_LB_POLICY_UNREF(lb_policy, "exit_idle");
-  }
 
-  if (old_lb_policy) {
-    GRPC_LB_POLICY_UNREF(old_lb_policy, "channel");
-  }
-
-  gpr_mu_lock(&chand->mu_config);
   if (iomgr_success && chand->resolver) {
     grpc_resolver *resolver = chand->resolver;
     GRPC_RESOLVER_REF(resolver, "channel-next");
@@ -443,6 +487,10 @@ 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);
+    if (lb_policy != NULL) {
+      watch_lb_policy(chand, lb_policy, state);
+    }
   } else {
     old_resolver = chand->resolver;
     chand->resolver = NULL;
@@ -455,12 +503,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 = next;
   }
 
+  if (lb_policy != NULL) {
+    GRPC_LB_POLICY_UNREF(lb_policy, "config_change");
+  }
   GRPC_CHANNEL_INTERNAL_UNREF(chand->master, "resolver");
 }
 
@@ -491,6 +551,8 @@ static void cc_start_transport_op(grpc_channel_element *elem,
     chand->resolver = NULL;
     if (chand->lb_policy != NULL) {
       grpc_lb_policy_shutdown(chand->lb_policy);
+      GRPC_LB_POLICY_UNREF(chand->lb_policy, "channel");
+      chand->lb_policy = NULL;
     }
   }
 
@@ -578,10 +640,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 */
@@ -595,6 +658,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);
 }
 
@@ -649,3 +714,20 @@ void grpc_client_channel_watch_connectivity_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);
+}

src/core/channel/client_channel.h

@@ -59,4 +59,11 @@ void grpc_client_channel_watch_connectivity_state(
     grpc_channel_element *elem, grpc_connectivity_state *state,
     grpc_iomgr_closure *on_complete);
 
+grpc_pollset_set *grpc_client_channel_get_connecting_pollset_set(grpc_channel_element *elem);
+
+void grpc_client_channel_add_interested_party(grpc_channel_element *channel,
+                                          grpc_pollset *pollset);
+void grpc_client_channel_del_interested_party(grpc_channel_element *channel,
+                                          grpc_pollset *pollset);
+
 #endif /* GRPC_INTERNAL_CORE_CHANNEL_CLIENT_CHANNEL_H */

src/core/client_config/lb_policies/pick_first.c

@@ -73,12 +73,30 @@ typedef struct {
   grpc_connectivity_state_tracker state_tracker;
 } pick_first_lb_policy;
 
+static void del_interested_parties_locked(pick_first_lb_policy *p) {
+  pending_pick *pp;
+  for (pp = p->pending_picks; pp; pp = pp->next) {
+    grpc_subchannel_del_interested_party(p->subchannels[p->checking_subchannel],
+                                         pp->pollset);
+  }
+}
+
+static void add_interested_parties_locked(pick_first_lb_policy *p) {
+  pending_pick *pp;
+  for (pp = p->pending_picks; pp; pp = pp->next) {
+    grpc_subchannel_add_interested_party(p->subchannels[p->checking_subchannel],
+                                         pp->pollset);
+  }
+}
+
 void pf_destroy(grpc_lb_policy *pol) {
   pick_first_lb_policy *p = (pick_first_lb_policy *)pol;
   size_t i;
+  del_interested_parties_locked(p);
   for (i = 0; i < p->num_subchannels; i++) {
     GRPC_SUBCHANNEL_UNREF(p->subchannels[i], "pick_first");
   }
+  grpc_connectivity_state_destroy(&p->state_tracker);
   gpr_free(p->subchannels);
   gpr_mu_destroy(&p->mu);
   gpr_free(p);
@@ -88,6 +106,7 @@ void pf_shutdown(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(p);
   while ((pp = p->pending_picks)) {
     p->pending_picks = pp->next;
     *pp->target = NULL;
@@ -142,31 +161,25 @@ void pf_pick(grpc_lb_policy *pol, grpc_pollset *pollset,
   }
 }
 
-static void del_interested_parties_locked(pick_first_lb_policy *p) {
-  pending_pick *pp;
-  for (pp = p->pending_picks; pp; pp = pp->next) {
-    grpc_subchannel_del_interested_party(p->subchannels[p->checking_subchannel],
-                                         pp->pollset);
-  }
-}
-
-static void add_interested_parties_locked(pick_first_lb_policy *p) {
-  pending_pick *pp;
-  for (pp = p->pending_picks; pp; pp = pp->next) {
-    grpc_subchannel_add_interested_party(p->subchannels[p->checking_subchannel],
-                                         pp->pollset);
-  }
-}
-
 static void pf_connectivity_changed(void *arg, int iomgr_success) {
   pick_first_lb_policy *p = arg;
   pending_pick *pp;
   int unref = 0;
 
   gpr_mu_lock(&p->mu);
+
+  if (p->selected != NULL) {
+    grpc_connectivity_state_set(&p->state_tracker, p->checking_connectivity);
+    if (p->checking_connectivity != GRPC_CHANNEL_FATAL_FAILURE) {
+      grpc_subchannel_notify_on_state_change(p->selected, &p->checking_connectivity, &p->connectivity_changed);
+    } else {
+      unref = 1;
+    }
+  } else {
 loop:
     switch (p->checking_connectivity) {
       case GRPC_CHANNEL_READY:
+        grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_READY);
         p->selected = p->subchannels[p->checking_subchannel];
         while ((pp = p->pending_picks)) {
           p->pending_picks = pp->next;
@@ -175,18 +188,25 @@ loop:
           grpc_iomgr_add_delayed_callback(pp->on_complete, 1);
           gpr_free(pp);
         }
-      unref = 1;
+        grpc_subchannel_notify_on_state_change(p->selected, &p->checking_connectivity, &p->connectivity_changed);
         break;
       case GRPC_CHANNEL_TRANSIENT_FAILURE:
+        grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE);
         del_interested_parties_locked(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(p);
+        if (p->checking_connectivity == GRPC_CHANNEL_TRANSIENT_FAILURE) {
+          grpc_subchannel_notify_on_state_change(p->subchannels[p->checking_subchannel], &p->checking_connectivity, &p->connectivity_changed);
+        } else {
           goto loop;
+        }
+        break;
       case GRPC_CHANNEL_CONNECTING:
       case GRPC_CHANNEL_IDLE:
+        grpc_connectivity_state_set(&p->state_tracker, p->checking_connectivity);
         grpc_subchannel_notify_on_state_change(
             p->subchannels[p->checking_subchannel], &p->checking_connectivity,
             &p->connectivity_changed);
@@ -198,6 +218,7 @@ loop:
         p->num_subchannels--;
         GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first");
         if (p->num_subchannels == 0) {
+          grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_FATAL_FAILURE);
           while ((pp = p->pending_picks)) {
             p->pending_picks = pp->next;
             *pp->target = NULL;
@@ -206,6 +227,7 @@ loop:
           }
           unref = 1;
         } else {
+          grpc_connectivity_state_set(&p->state_tracker, GRPC_CHANNEL_TRANSIENT_FAILURE);
           p->checking_subchannel %= p->num_subchannels;
           p->checking_connectivity = grpc_subchannel_check_connectivity(
               p->subchannels[p->checking_subchannel]);
@@ -213,6 +235,8 @@ loop:
           goto loop;
         }
     }
+  }
+
   gpr_mu_unlock(&p->mu);
 
   if (unref) {
@@ -278,6 +302,7 @@ grpc_lb_policy *grpc_create_pick_first_lb_policy(grpc_subchannel **subchannels,
   grpc_lb_policy_init(&p->base, &pick_first_lb_policy_vtable);
   p->subchannels = gpr_malloc(sizeof(grpc_subchannel *) * num_subchannels);
   p->num_subchannels = num_subchannels;
+  grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE, "pick_first");
   memcpy(p->subchannels, subchannels,
          sizeof(grpc_subchannel *) * num_subchannels);
   grpc_iomgr_closure_init(&p->connectivity_changed, pf_connectivity_changed, p);

src/core/client_config/lb_policy.c

@@ -81,3 +81,12 @@ void grpc_lb_policy_broadcast(grpc_lb_policy *policy, grpc_transport_op *op) {
 void grpc_lb_policy_exit_idle(grpc_lb_policy *policy) {
   policy->vtable->exit_idle(policy);
 }
+
+void grpc_lb_policy_notify_on_state_change(grpc_lb_policy *policy, grpc_connectivity_state *state,
+                                 grpc_iomgr_closure *closure) {
+  policy->vtable->notify_on_state_change(policy, state, closure);
+}
+
+grpc_connectivity_state grpc_lb_policy_check_connectivity(grpc_lb_policy *policy) {
+  return policy->vtable->check_connectivity(policy);
+}

src/core/client_config/lb_policy.h

@@ -75,6 +75,8 @@ struct grpc_lb_policy_vtable {
                                  grpc_iomgr_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))
@@ -111,4 +113,9 @@ void grpc_lb_policy_broadcast(grpc_lb_policy *policy, grpc_transport_op *op);
 
 void grpc_lb_policy_exit_idle(grpc_lb_policy *policy);
 
+void grpc_lb_policy_notify_on_state_change(grpc_lb_policy *policy, grpc_connectivity_state *state,
+                                 grpc_iomgr_closure *closure);
+
+grpc_connectivity_state grpc_lb_policy_check_connectivity(grpc_lb_policy *policy);
+
 #endif /* GRPC_INTERNAL_CORE_CONFIG_LB_POLICY_H */

src/core/client_config/subchannel.c

@@ -38,9 +38,11 @@
 #include <grpc/support/alloc.h>
 
 #include "src/core/channel/channel_args.h"
+#include "src/core/channel/client_channel.h"
 #include "src/core/channel/connected_channel.h"
 #include "src/core/iomgr/alarm.h"
 #include "src/core/transport/connectivity_state.h"
+#include "src/core/surface/channel.h"
 
 typedef struct {
   /* all fields protected by subchannel->mu */
@@ -94,8 +96,9 @@ struct grpc_subchannel {
   grpc_iomgr_closure connected;
 
   /** pollset_set tracking who's interested in a connection
-      being setup */
+      being setup - owned by the master channel (in particular the client_channel
-  grpc_pollset_set pollset_set;
+      filter there-in) */
+  grpc_pollset_set *pollset_set;
 
   /** mutex protecting remaining elements */
   gpr_mu mu;
@@ -244,7 +247,6 @@ static void subchannel_destroy(grpc_subchannel *c) {
   grpc_channel_args_destroy(c->args);
   gpr_free(c->addr);
   grpc_mdctx_unref(c->mdctx);
-  grpc_pollset_set_destroy(&c->pollset_set);
   grpc_connectivity_state_destroy(&c->state_tracker);
   grpc_connector_unref(c->connector);
   gpr_free(c);
@@ -252,17 +254,18 @@ static void subchannel_destroy(grpc_subchannel *c) {
 
 void grpc_subchannel_add_interested_party(grpc_subchannel *c,
                                           grpc_pollset *pollset) {
-  grpc_pollset_set_add_pollset(&c->pollset_set, pollset);
+  grpc_pollset_set_add_pollset(c->pollset_set, pollset);
 }
 
 void grpc_subchannel_del_interested_party(grpc_subchannel *c,
                                           grpc_pollset *pollset) {
-  grpc_pollset_set_del_pollset(&c->pollset_set, pollset);
+  grpc_pollset_set_del_pollset(c->pollset_set, pollset);
 }
 
 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;
   c->connector = connector;
@@ -277,10 +280,10 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
   c->args = grpc_channel_args_copy(args->args);
   c->mdctx = args->mdctx;
   c->master = args->master;
+  c->pollset_set = grpc_client_channel_get_connecting_pollset_set(parent_elem);
   grpc_mdctx_ref(c->mdctx);
-  grpc_pollset_set_init(&c->pollset_set);
   grpc_iomgr_closure_init(&c->connected, subchannel_connected, c);
-  grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE);
+  grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE, "subchannel");
   gpr_mu_init(&c->mu);
   return c;
 }
@@ -288,7 +291,7 @@ grpc_subchannel *grpc_subchannel_create(grpc_connector *connector,
 static void continue_connect(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);
@@ -309,6 +312,7 @@ static void start_connect(grpc_subchannel *c) {
 
 static void continue_creating_call(void *arg, int iomgr_success) {
   waiting_for_connect *w4c = arg;
+  grpc_subchannel_del_interested_party(w4c->subchannel, w4c->pollset);
   grpc_subchannel_create_call(w4c->subchannel, w4c->pollset, w4c->target,
                               w4c->notify);
   GRPC_SUBCHANNEL_UNREF(w4c->subchannel, "waiting_for_connect");
@@ -344,6 +348,7 @@ void grpc_subchannel_create_call(grpc_subchannel *c, grpc_pollset *pollset,
       connectivity_state_changed_locked(c);
       /* released by connection */
       SUBCHANNEL_REF_LOCKED(c, "connecting");
+      GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting");
       gpr_mu_unlock(&c->mu);
 
       start_connect(c);
@@ -372,6 +377,7 @@ void grpc_subchannel_notify_on_state_change(grpc_subchannel *c,
     c->connecting = 1;
     /* released by connection */
     SUBCHANNEL_REF_LOCKED(c, "connecting");
+    GRPC_CHANNEL_INTERNAL_REF(c->master, "connecting");
     connectivity_state_changed_locked(c);
   }
   gpr_mu_unlock(&c->mu);
@@ -536,7 +542,9 @@ static void publish_transport(grpc_subchannel *c) {
   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(c->master, "connecting");
   GPR_ASSERT(!SUBCHANNEL_UNREF_LOCKED(c, "connecting"));
   elem =
       grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(c->active), 0);
@@ -570,6 +578,7 @@ static void on_alarm(void *arg, int iomgr_success) {
   if (iomgr_success) {
     continue_connect(c);
   } else {
+    GRPC_CHANNEL_INTERNAL_UNREF(c->master, "connecting");
     GRPC_SUBCHANNEL_UNREF(c, "connecting");
   }
 }
@@ -580,9 +589,9 @@ static void subchannel_connected(void *arg, int iomgr_success) {
     publish_transport(c);
   } else {
     gpr_mu_lock(&c->mu);
-    connectivity_state_changed_locked(c);
     GPR_ASSERT(!c->have_alarm);
     c->have_alarm = 1;
+    connectivity_state_changed_locked(c);
     c->next_attempt = gpr_time_add(c->next_attempt, c->backoff_delta);
     c->backoff_delta = gpr_time_add(c->backoff_delta, c->backoff_delta);
     grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_REALTIME));

src/core/iomgr/endpoint.c

@@ -50,6 +50,10 @@ void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) {
   ep->vtable->add_to_pollset(ep, pollset);
 }
 
+void grpc_endpoint_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pollset_set) {
+  ep->vtable->add_to_pollset_set(ep, pollset_set);
+}
+
 void grpc_endpoint_shutdown(grpc_endpoint *ep) { ep->vtable->shutdown(ep); }
 
 void grpc_endpoint_destroy(grpc_endpoint *ep) { ep->vtable->destroy(ep); }

src/core/iomgr/endpoint.h

@@ -35,6 +35,7 @@
 #define GRPC_INTERNAL_CORE_IOMGR_ENDPOINT_H
 
 #include "src/core/iomgr/pollset.h"
+#include "src/core/iomgr/pollset_set.h"
 #include <grpc/support/slice.h>
 #include <grpc/support/time.h>
 
@@ -70,6 +71,7 @@ struct grpc_endpoint_vtable {
                                       size_t nslices, grpc_endpoint_write_cb cb,
                                       void *user_data);
   void (*add_to_pollset)(grpc_endpoint *ep, grpc_pollset *pollset);
+  void (*add_to_pollset_set)(grpc_endpoint *ep, grpc_pollset_set *pollset);
   void (*shutdown)(grpc_endpoint *ep);
   void (*destroy)(grpc_endpoint *ep);
 };
@@ -98,6 +100,7 @@ void grpc_endpoint_destroy(grpc_endpoint *ep);
 /* Add an endpoint to a pollset, so that when the pollset is polled, events from
    this endpoint are considered */
 void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset);
+void grpc_endpoint_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pollset_set);
 
 struct grpc_endpoint {
   const grpc_endpoint_vtable *vtable;

src/core/iomgr/pollset_set_posix.c

@@ -60,7 +60,7 @@ void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set) {
 
 void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set,
                                   grpc_pollset *pollset) {
-  size_t i;
+  size_t i, j;
   gpr_mu_lock(&pollset_set->mu);
   if (pollset_set->pollset_count == pollset_set->pollset_capacity) {
     pollset_set->pollset_capacity =
@@ -70,9 +70,15 @@ void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set,
                                                sizeof(*pollset_set->pollsets));
   }
   pollset_set->pollsets[pollset_set->pollset_count++] = pollset;
-  for (i = 0; i < pollset_set->fd_count; i++) {
+  for (i = 0, j = 0; i < pollset_set->fd_count; i++) {
+    if (grpc_fd_is_orphaned(pollset_set->fds[i])) {
+      GRPC_FD_UNREF(pollset_set->fds[i], "pollset");
+    } else {
       grpc_pollset_add_fd(pollset, pollset_set->fds[i]);
+      pollset_set->fds[j++] = pollset_set->fds[i];
+    }
   }
+  pollset_set->fd_count = j;
   gpr_mu_unlock(&pollset_set->mu);
 }
 

src/core/iomgr/tcp_posix.c

@@ -567,9 +567,14 @@ static void grpc_tcp_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) {
   grpc_pollset_add_fd(pollset, tcp->em_fd);
 }
 
+static void grpc_tcp_add_to_pollset_set(grpc_endpoint *ep, grpc_pollset_set *pollset_set) {
+  grpc_tcp *tcp = (grpc_tcp *)ep;
+  grpc_pollset_set_add_fd(pollset_set, tcp->em_fd);
+}
+
 static const grpc_endpoint_vtable vtable = {
     grpc_tcp_notify_on_read, grpc_tcp_write, grpc_tcp_add_to_pollset,
-    grpc_tcp_shutdown, grpc_tcp_destroy};
+    grpc_tcp_add_to_pollset_set, grpc_tcp_shutdown, grpc_tcp_destroy};
 
 grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd, size_t slice_size) {
   grpc_tcp *tcp = (grpc_tcp *)gpr_malloc(sizeof(grpc_tcp));

src/core/security/secure_endpoint.c

@@ -331,9 +331,15 @@ static void endpoint_add_to_pollset(grpc_endpoint *secure_ep,
   grpc_endpoint_add_to_pollset(ep->wrapped_ep, pollset);
 }
 
+static void endpoint_add_to_pollset_set(grpc_endpoint *secure_ep,
+                                    grpc_pollset_set *pollset_set) {
+  secure_endpoint *ep = (secure_endpoint *)secure_ep;
+  grpc_endpoint_add_to_pollset_set(ep->wrapped_ep, pollset_set);
+}
+
 static const grpc_endpoint_vtable vtable = {
     endpoint_notify_on_read, endpoint_write, endpoint_add_to_pollset,
-    endpoint_shutdown, endpoint_unref};
+    endpoint_add_to_pollset_set, endpoint_shutdown, endpoint_unref};
 
 grpc_endpoint *grpc_secure_endpoint_create(
     struct tsi_frame_protector *protector, grpc_endpoint *transport,

src/core/surface/channel.h

@@ -58,6 +58,8 @@ grpc_mdstr *grpc_channel_get_compresssion_level_string(grpc_channel *channel);
 grpc_mdstr *grpc_channel_get_message_string(grpc_channel *channel);
 gpr_uint32 grpc_channel_get_max_message_length(grpc_channel *channel);
 
+#define GRPC_CHANNEL_REF_COUNT_DEBUG
+
 #ifdef GRPC_CHANNEL_REF_COUNT_DEBUG
 void grpc_channel_internal_ref(grpc_channel *channel, const char *reason);
 void grpc_channel_internal_unref(grpc_channel *channel, const char *reason);

src/core/surface/channel_connectivity.c

@@ -73,10 +73,15 @@ typedef struct {
   grpc_connectivity_state *optional_new_state;
   grpc_completion_queue *cq;
   grpc_cq_completion completion_storage;
+  grpc_channel *channel;
   void *tag;
 } state_watcher;
 
 static void delete_state_watcher(state_watcher *w) {
+  grpc_channel_element *client_channel_elem =
+      grpc_channel_stack_last_element(grpc_channel_get_channel_stack(w->channel));
+  grpc_client_channel_del_interested_party(client_channel_elem, grpc_cq_pollset(w->cq));
+  GRPC_CHANNEL_INTERNAL_UNREF(w->channel, "watch_connectivity");
   gpr_mu_destroy(&w->mu);
   gpr_free(w);
 }
@@ -143,9 +148,9 @@ static void partly_done(state_watcher *w, int due_to_completion) {
   }
 }
 
-static void watch_complete(void *pw, int success) { partly_done(pw, 0); }
+static void watch_complete(void *pw, int success) { partly_done(pw, 1); }
 
-static void timeout_complete(void *pw, int success) { partly_done(pw, 1); }
+static void timeout_complete(void *pw, int success) { partly_done(pw, 0); }
 
 void grpc_channel_watch_connectivity_state(
     grpc_channel *channel, grpc_connectivity_state last_observed_state,
@@ -165,6 +170,7 @@ void grpc_channel_watch_connectivity_state(
   w->optional_new_state = optional_new_state;
   w->cq = cq;
   w->tag = tag;
+  w->channel = channel;
 
   grpc_alarm_init(&w->alarm, deadline, timeout_complete, w,
                   gpr_now(GPR_CLOCK_REALTIME));
@@ -176,6 +182,8 @@ void grpc_channel_watch_connectivity_state(
             client_channel_elem->filter->name);
     grpc_iomgr_add_delayed_callback(&w->on_complete, 1);
   } else {
+    GRPC_CHANNEL_INTERNAL_REF(channel, "watch_connectivity");
+    grpc_client_channel_add_interested_party(client_channel_elem, grpc_cq_pollset(cq));
     grpc_client_channel_watch_connectivity_state(client_channel_elem, &w->state,
                                                  &w->on_complete);
   }

src/core/surface/channel_create.c

@@ -108,6 +108,7 @@ typedef struct {
   gpr_refcount refs;
   grpc_mdctx *mdctx;
   grpc_channel_args *merge_args;
+  grpc_channel *master;
 } subchannel_factory;
 
 static void subchannel_factory_ref(grpc_subchannel_factory *scf) {
@@ -118,6 +119,7 @@ static void subchannel_factory_ref(grpc_subchannel_factory *scf) {
 static void subchannel_factory_unref(grpc_subchannel_factory *scf) {
   subchannel_factory *f = (subchannel_factory *)scf;
   if (gpr_unref(&f->refs)) {
+    GRPC_CHANNEL_INTERNAL_UNREF(f->master, "subchannel_factory");
     grpc_channel_args_destroy(f->merge_args);
     grpc_mdctx_unref(f->mdctx);
     gpr_free(f);
@@ -136,6 +138,7 @@ static grpc_subchannel *subchannel_factory_create_subchannel(
   gpr_ref_init(&c->refs, 1);
   args->mdctx = f->mdctx;
   args->args = final_args;
+  args->master = f->master;
   s = grpc_subchannel_create(&c->base, args);
   grpc_connector_unref(&c->base);
   grpc_channel_args_destroy(final_args);
@@ -166,18 +169,21 @@ grpc_channel *grpc_channel_create(const char *target,
   filters[n++] = &grpc_client_channel_filter;
   GPR_ASSERT(n <= MAX_FILTERS);
 
+  channel = grpc_channel_create_from_filters(filters, n, args, mdctx, 1);
+
   f = gpr_malloc(sizeof(*f));
   f->base.vtable = &subchannel_factory_vtable;
   gpr_ref_init(&f->refs, 1);
   grpc_mdctx_ref(mdctx);
   f->mdctx = mdctx;
   f->merge_args = grpc_channel_args_copy(args);
+  f->master = channel;
+  GRPC_CHANNEL_INTERNAL_REF(f->master, "subchannel_factory");
   resolver = grpc_resolver_create(target, &f->base);
   if (!resolver) {
     return NULL;
   }
 
-  channel = grpc_channel_create_from_filters(filters, n, args, mdctx, 1);
   grpc_client_channel_set_resolver(grpc_channel_get_channel_stack(channel),
                                    resolver);
   GRPC_RESOLVER_UNREF(resolver, "create");

src/core/surface/init.c

@@ -46,6 +46,7 @@
 #include "src/core/surface/init.h"
 #include "src/core/surface/surface_trace.h"
 #include "src/core/transport/chttp2_transport.h"
+#include "src/core/transport/connectivity_state.h"
 
 #ifdef GPR_POSIX_SOCKET
 #include "src/core/client_config/resolvers/unix_resolver_posix.h"
@@ -76,6 +77,7 @@ void grpc_init(void) {
     grpc_register_tracer("http", &grpc_http_trace);
     grpc_register_tracer("flowctl", &grpc_flowctl_trace);
     grpc_register_tracer("batch", &grpc_trace_batch);
+    grpc_register_tracer("connectivity_state", &grpc_connectivity_state_trace);
     grpc_security_pre_init();
     grpc_iomgr_init();
     grpc_tracer_init("GRPC_TRACE");

src/core/surface/secure_channel_create.c

@@ -132,6 +132,7 @@ typedef struct {
   grpc_mdctx *mdctx;
   grpc_channel_args *merge_args;
   grpc_channel_security_connector *security_connector;
+  grpc_channel *master;
 } subchannel_factory;
 
 static void subchannel_factory_ref(grpc_subchannel_factory *scf) {
@@ -144,6 +145,7 @@ static void subchannel_factory_unref(grpc_subchannel_factory *scf) {
   if (gpr_unref(&f->refs)) {
     GRPC_SECURITY_CONNECTOR_UNREF(&f->security_connector->base,
                                   "subchannel_factory");
+    GRPC_CHANNEL_INTERNAL_UNREF(f->master, "subchannel_factory");
     grpc_channel_args_destroy(f->merge_args);
     grpc_mdctx_unref(f->mdctx);
     gpr_free(f);
@@ -163,6 +165,7 @@ static grpc_subchannel *subchannel_factory_create_subchannel(
   gpr_ref_init(&c->refs, 1);
   args->mdctx = f->mdctx;
   args->args = final_args;
+  args->master = f->master;
   s = grpc_subchannel_create(&c->base, args);
   grpc_connector_unref(&c->base);
   grpc_channel_args_destroy(final_args);
@@ -215,6 +218,8 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds,
   filters[n++] = &grpc_client_channel_filter;
   GPR_ASSERT(n <= MAX_FILTERS);
 
+  channel = grpc_channel_create_from_filters(filters, n, args_copy, mdctx, 1);
+
   f = gpr_malloc(sizeof(*f));
   f->base.vtable = &subchannel_factory_vtable;
   gpr_ref_init(&f->refs, 1);
@@ -223,12 +228,13 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds,
   GRPC_SECURITY_CONNECTOR_REF(&connector->base, "subchannel_factory");
   f->security_connector = connector;
   f->merge_args = grpc_channel_args_copy(args_copy);
+  f->master = channel;
+  GRPC_CHANNEL_INTERNAL_REF(channel, "subchannel_factory");
   resolver = grpc_resolver_create(target, &f->base);
   if (!resolver) {
     return NULL;
   }
 
-  channel = grpc_channel_create_from_filters(filters, n, args_copy, mdctx, 1);
   grpc_client_channel_set_resolver(grpc_channel_get_channel_stack(channel),
                                    resolver);
   GRPC_RESOLVER_UNREF(resolver, "create");

src/core/transport/chttp2_transport.c

@@ -110,6 +110,8 @@ static void cancel_from_api(grpc_chttp2_transport_global *transport_global,
 /** Add endpoint from this transport to pollset */
 static void add_to_pollset_locked(grpc_chttp2_transport *t,
                                   grpc_pollset *pollset);
+static void add_to_pollset_set_locked(grpc_chttp2_transport *t,
+                                  grpc_pollset_set *pollset_set);
 
 /** Start new streams that have been created if we can */
 static void maybe_start_some_streams(
@@ -233,7 +235,7 @@ static void init_transport(grpc_chttp2_transport *t,
       is_client ? GRPC_DTS_FH_0 : GRPC_DTS_CLIENT_PREFIX_0;
   t->writing.is_client = is_client;
   grpc_connectivity_state_init(&t->channel_callback.state_tracker,
-                               GRPC_CHANNEL_READY);
+                               GRPC_CHANNEL_READY, "transport");
 
   gpr_slice_buffer_init(&t->global.qbuf);
 
@@ -668,6 +670,7 @@ static void send_ping_locked(grpc_chttp2_transport *t,
 
 static void perform_transport_op(grpc_transport *gt, grpc_transport_op *op) {
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
+  int close_transport = 0;
 
   lock(t);
 
@@ -687,9 +690,7 @@ static void perform_transport_op(grpc_transport *gt, grpc_transport_op *op) {
         t->global.last_incoming_stream_id,
         grpc_chttp2_grpc_status_to_http2_error(op->goaway_status),
         gpr_slice_ref(*op->goaway_message), &t->global.qbuf);
-    if (!grpc_chttp2_has_streams(t)) {
+    close_transport = !grpc_chttp2_has_streams(t);
-      close_transport_locked(t);
-    }
   }
 
   if (op->set_accept_stream != NULL) {
@@ -702,6 +703,10 @@ static void perform_transport_op(grpc_transport *gt, grpc_transport_op *op) {
     add_to_pollset_locked(t, op->bind_pollset);
   }
 
+  if (op->bind_pollset_set) {
+    add_to_pollset_set_locked(t, op->bind_pollset_set);
+  }
+
   if (op->send_ping) {
     send_ping_locked(t, op->send_ping);
   }
@@ -711,6 +716,12 @@ static void perform_transport_op(grpc_transport *gt, grpc_transport_op *op) {
   }
 
   unlock(t);
+
+  if (close_transport) {
+    lock(t);
+    close_transport_locked(t);
+    unlock(t);
+  }
 }
 
 /*
@@ -1016,6 +1027,13 @@ static void add_to_pollset_locked(grpc_chttp2_transport *t,
   }
 }
 
+static void add_to_pollset_set_locked(grpc_chttp2_transport *t,
+                                  grpc_pollset_set *pollset_set) {
+  if (t->ep) {
+    grpc_endpoint_add_to_pollset_set(t->ep, pollset_set);
+  }
+}
+
 /*
  * TRACING
  */

src/core/transport/connectivity_state.c

@@ -34,11 +34,27 @@
 #include "src/core/transport/connectivity_state.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+
+int grpc_connectivity_state_trace = 0;
+
+const char *grpc_connectivity_state_name(grpc_connectivity_state state) {
+  switch (state) {
+    case GRPC_CHANNEL_IDLE: return "IDLE";
+    case GRPC_CHANNEL_CONNECTING: return "CONNECTING";
+    case GRPC_CHANNEL_READY: return "READY";
+    case GRPC_CHANNEL_TRANSIENT_FAILURE: return "TRANSIENT_FAILURE";
+    case GRPC_CHANNEL_FATAL_FAILURE: return "FATAL_FAILURE";
+  }
+  abort();
+  return "UNKNOWN";
+}
 
 void grpc_connectivity_state_init(grpc_connectivity_state_tracker *tracker,
-                                  grpc_connectivity_state init_state) {
+                                  grpc_connectivity_state init_state, const char *name) {
   tracker->current_state = init_state;
   tracker->watchers = NULL;
+  tracker->name = gpr_strdup(name);
 }
 
 void grpc_connectivity_state_destroy(grpc_connectivity_state_tracker *tracker) {
@@ -54,6 +70,7 @@ void grpc_connectivity_state_destroy(grpc_connectivity_state_tracker *tracker) {
     }
     gpr_free(w);
   }
+  gpr_free(tracker->name);
 }
 
 grpc_connectivity_state grpc_connectivity_state_check(
@@ -64,6 +81,9 @@ grpc_connectivity_state grpc_connectivity_state_check(
 int grpc_connectivity_state_notify_on_state_change(
     grpc_connectivity_state_tracker *tracker, grpc_connectivity_state *current,
     grpc_iomgr_closure *notify) {
+  if (grpc_connectivity_state_trace) {
+    gpr_log(GPR_DEBUG, "CONWATCH: %s: from %s [cur=%s]", tracker->name, grpc_connectivity_state_name(*current), grpc_connectivity_state_name(tracker->current_state));
+  }
   if (tracker->current_state != *current) {
     *current = tracker->current_state;
     grpc_iomgr_add_callback(notify);
@@ -82,6 +102,9 @@ void grpc_connectivity_state_set_with_scheduler(
     void (*scheduler)(void *arg, grpc_iomgr_closure *closure), void *arg) {
   grpc_connectivity_state_watcher *new = NULL;
   grpc_connectivity_state_watcher *w;
+  if (grpc_connectivity_state_trace) {
+    gpr_log(GPR_DEBUG, "SET: %s: %s --> %s", tracker->name, grpc_connectivity_state_name(tracker->current_state), grpc_connectivity_state_name(state));
+  }
   if (tracker->current_state == state) {
     return;
   }

src/core/transport/connectivity_state.h

@@ -51,10 +51,14 @@ typedef struct {
   grpc_connectivity_state current_state;
   /** all our watchers */
   grpc_connectivity_state_watcher *watchers;
+  /** a name to help debugging */
+  char *name;
 } grpc_connectivity_state_tracker;
 
+extern int grpc_connectivity_state_trace;
+
 void grpc_connectivity_state_init(grpc_connectivity_state_tracker *tracker,
-                                  grpc_connectivity_state init_state);
+                                  grpc_connectivity_state init_state, const char *name);
 void grpc_connectivity_state_destroy(grpc_connectivity_state_tracker *tracker);
 
 void grpc_connectivity_state_set(grpc_connectivity_state_tracker *tracker,

src/core/transport/transport.h

@@ -105,6 +105,8 @@ typedef struct grpc_transport_op {
   void *set_accept_stream_user_data;
   /** add this transport to a pollset */
   grpc_pollset *bind_pollset;
+  /** add this transport to a pollset_set */
+  grpc_pollset_set *bind_pollset_set;
   /** send a ping, call this back if not NULL */
   grpc_iomgr_closure *send_ping;
 } grpc_transport_op;

test/core/end2end/fixtures/chttp2_fullstack_uds_posix_with_poll.c

@@ -0,0 +1,124 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#include "src/core/channel/client_channel.h"
+#include "src/core/channel/connected_channel.h"
+#include "src/core/channel/http_server_filter.h"
+#include "src/core/support/string.h"
+#include "src/core/surface/channel.h"
+#include "src/core/surface/server.h"
+#include "src/core/transport/chttp2_transport.h"
+#include <grpc/support/alloc.h>
+#include <grpc/support/host_port.h>
+#include <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/sync.h>
+#include <grpc/support/thd.h>
+#include <grpc/support/useful.h>
+#include "test/core/util/port.h"
+#include "test/core/util/test_config.h"
+
+typedef struct fullstack_fixture_data {
+  char *localaddr;
+} fullstack_fixture_data;
+
+static int unique = 1;
+
+static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
+    grpc_channel_args *client_args, grpc_channel_args *server_args) {
+  grpc_end2end_test_fixture f;
+  fullstack_fixture_data *ffd = gpr_malloc(sizeof(fullstack_fixture_data));
+  memset(&f, 0, sizeof(f));
+
+  gpr_asprintf(&ffd->localaddr, "unix:/tmp/grpc_fullstack_test.%d.%d", getpid(),
+               unique++);
+
+  f.fixture_data = ffd;
+  f.cq = grpc_completion_queue_create();
+
+  return f;
+}
+
+void chttp2_init_client_fullstack(grpc_end2end_test_fixture *f,
+                                  grpc_channel_args *client_args) {
+  fullstack_fixture_data *ffd = f->fixture_data;
+  f->client = grpc_channel_create(ffd->localaddr, client_args);
+}
+
+void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f,
+                                  grpc_channel_args *server_args) {
+  fullstack_fixture_data *ffd = f->fixture_data;
+  if (f->server) {
+    grpc_server_destroy(f->server);
+  }
+  f->server = grpc_server_create(server_args);
+  grpc_server_register_completion_queue(f->server, f->cq);
+  GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
+  grpc_server_start(f->server);
+}
+
+void chttp2_tear_down_fullstack(grpc_end2end_test_fixture *f) {
+  fullstack_fixture_data *ffd = f->fixture_data;
+  gpr_free(ffd->localaddr);
+  gpr_free(ffd);
+}
+
+/* All test configurations */
+static grpc_end2end_test_config configs[] = {
+    {"chttp2/fullstack_uds", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION,
+     chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
+     chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
+};
+
+int main(int argc, char **argv) {
+  size_t i;
+
+  grpc_platform_become_multipoller = grpc_poll_become_multipoller;
+
+  grpc_test_init(argc, argv);
+  grpc_init();
+
+  for (i = 0; i < sizeof(configs) / sizeof(*configs); i++) {
+    grpc_end2end_tests(configs[i]);
+  }
+
+  grpc_shutdown();
+
+  return 0;
+}

test/core/end2end/gen_build_json.py

@@ -36,26 +36,29 @@ import simplejson
 import collections
 
 
-FixtureOptions = collections.namedtuple('FixtureOptions', 'secure platforms')
+FixtureOptions = collections.namedtuple('FixtureOptions', 'fullstack secure platforms')
-default_unsecure_fixture_options = FixtureOptions(False, ['windows', 'posix'])
+default_unsecure_fixture_options = FixtureOptions(True, False, ['windows', 'posix'])
-default_secure_fixture_options = FixtureOptions(True, ['windows', 'posix'])
+socketpair_unsecure_fixture_options = FixtureOptions(False, False, ['windows', 'posix'])
+default_secure_fixture_options = FixtureOptions(True, True, ['windows', 'posix'])
 
 # maps fixture name to whether it requires the security library
 END2END_FIXTURES = {
     'chttp2_fake_security': default_secure_fixture_options,
     'chttp2_fullstack': default_unsecure_fixture_options,
-    'chttp2_fullstack_with_poll': FixtureOptions(False, ['posix']),
+    'chttp2_fullstack_with_poll': FixtureOptions(True, False, ['posix']),
-    'chttp2_fullstack_uds_posix': FixtureOptions(False, ['posix']),
+    'chttp2_fullstack_uds_posix': FixtureOptions(True, False, ['posix']),
+    'chttp2_fullstack_uds_posix_with_poll': FixtureOptions(True, False, ['posix']),
     'chttp2_simple_ssl_fullstack': default_secure_fixture_options,
-    'chttp2_simple_ssl_fullstack_with_poll': FixtureOptions(True, ['posix']),
+    'chttp2_simple_ssl_fullstack_with_poll': FixtureOptions(True, True, ['posix']),
     'chttp2_simple_ssl_with_oauth2_fullstack': default_secure_fixture_options,
-    'chttp2_socket_pair': default_unsecure_fixture_options,
+    'chttp2_socket_pair': socketpair_unsecure_fixture_options,
-    'chttp2_socket_pair_one_byte_at_a_time': default_unsecure_fixture_options,
+    'chttp2_socket_pair_one_byte_at_a_time': socketpair_unsecure_fixture_options,
-    'chttp2_socket_pair_with_grpc_trace': default_unsecure_fixture_options,
+    'chttp2_socket_pair_with_grpc_trace': socketpair_unsecure_fixture_options,
 }
 
-TestOptions = collections.namedtuple('TestOptions', 'flaky secure')
+TestOptions = collections.namedtuple('TestOptions', 'needs_fullstack flaky secure')
-default_test_options = TestOptions(False, False)
+default_test_options = TestOptions(False, False, False)
+connectivity_test_options = TestOptions(True, False, False)
 
 # maps test names to options
 END2END_TESTS = {
@@ -66,7 +69,8 @@ END2END_TESTS = {
     'cancel_before_invoke': default_test_options,
     'cancel_in_a_vacuum': default_test_options,
     'census_simple_request': default_test_options,
-    'disappearing_server': default_test_options,
+    'channel_connectivity': connectivity_test_options,
+    'disappearing_server': connectivity_test_options,
     'early_server_shutdown_finishes_inflight_calls': default_test_options,
     'early_server_shutdown_finishes_tags': default_test_options,
     'empty_batch': default_test_options,
@@ -81,17 +85,24 @@ END2END_TESTS = {
     'request_response_with_trailing_metadata_and_payload': default_test_options,
     'request_response_with_metadata_and_payload': default_test_options,
     'request_response_with_payload': default_test_options,
-    'request_response_with_payload_and_call_creds': TestOptions(flaky=False, secure=True),
+    'request_response_with_payload_and_call_creds': TestOptions(needs_fullstack=False, flaky=False, secure=True),
     'request_with_large_metadata': default_test_options,
     'request_with_payload': default_test_options,
     'request_with_flags': default_test_options,
     'server_finishes_request': default_test_options,
-    'simple_delayed_request': default_test_options,
+    'simple_delayed_request': connectivity_test_options,
     'simple_request': default_test_options,
     'simple_request_with_high_initial_sequence_number': default_test_options,
 }
 
 
+def compatible(f, t):
+  if END2END_TESTS[t].needs_fullstack:
+    if not END2END_FIXTURES[f].fullstack:
+      return False
+  return True
+
+
 def main():
   sec_deps = [
     'end2end_certs',
@@ -155,7 +166,8 @@ def main():
                   'end2end_test_%s' % t] + sec_deps
           }
       for f in sorted(END2END_FIXTURES.keys())
-      for t in sorted(END2END_TESTS.keys())] + [
+      for t in sorted(END2END_TESTS.keys())
+      if compatible(f, t)] + [
           {
               'name': '%s_%s_unsecure_test' % (f, t),
               'build': 'test',
@@ -169,7 +181,7 @@ def main():
                   'end2end_test_%s' % t] + unsec_deps
           }
       for f in sorted(END2END_FIXTURES.keys()) if not END2END_FIXTURES[f].secure
-      for t in sorted(END2END_TESTS.keys()) if not END2END_TESTS[t].secure]}
+      for t in sorted(END2END_TESTS.keys()) if compatible(f, t) and not END2END_TESTS[t].secure]}
   print simplejson.dumps(json, sort_keys=True, indent=2 * ' ')
 
 

test/core/end2end/tests/channel_connectivity.c

@@ -0,0 +1,123 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "test/core/end2end/end2end_tests.h"
+
+#include <grpc/support/log.h>
+#include <grpc/support/time.h>
+
+#include "test/core/end2end/cq_verifier.h"
+
+static void *tag(gpr_intptr t) { return (void *)t; }
+
+static void test_connectivity(grpc_end2end_test_config config) {
+  grpc_end2end_test_fixture f = config.create_fixture(NULL, NULL);
+  grpc_connectivity_state state;
+  cq_verifier *cqv = cq_verifier_create(f.cq);
+
+  config.init_client(&f, NULL);
+
+  /* channels should start life in IDLE, and stay there */
+  GPR_ASSERT(grpc_channel_check_connectivity_state(f.client, 0) == GRPC_CHANNEL_IDLE);
+  gpr_sleep_until(GRPC_TIMEOUT_MILLIS_TO_DEADLINE(100));
+  GPR_ASSERT(grpc_channel_check_connectivity_state(f.client, 0) == GRPC_CHANNEL_IDLE);
+
+  /* start watching for a change */
+  grpc_channel_watch_connectivity_state(
+  	f.client, GRPC_CHANNEL_IDLE, &state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f.cq, tag(1));
+  /* nothing should happen */
+  cq_verify_empty(cqv);
+
+  /* check that we're still in idle, and start connecting */
+  GPR_ASSERT(grpc_channel_check_connectivity_state(f.client, 1) == GRPC_CHANNEL_IDLE);
+
+  /* and now the watch should trigger */
+  cq_expect_completion(cqv, tag(1), 1);
+  cq_verify(cqv);
+  GPR_ASSERT(state == GRPC_CHANNEL_CONNECTING);
+
+  /* quickly followed by a transition to TRANSIENT_FAILURE */
+  grpc_channel_watch_connectivity_state(
+  	f.client, GRPC_CHANNEL_CONNECTING, &state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f.cq, tag(2));
+  cq_expect_completion(cqv, tag(2), 1);
+  cq_verify(cqv);
+  GPR_ASSERT(state == GRPC_CHANNEL_TRANSIENT_FAILURE);
+
+  gpr_log(GPR_DEBUG, "*** STARTING SERVER ***");
+
+  /* now let's bring up a server to connect to */
+  config.init_server(&f, NULL);
+
+  gpr_log(GPR_DEBUG, "*** STARTED SERVER ***");
+
+  /* we'll go through some set of transitions (some might be missed), until
+     READY is reached */
+  while (state != GRPC_CHANNEL_READY) {
+  	grpc_channel_watch_connectivity_state(
+  		f.client, state, &state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f.cq, tag(3));
+  	cq_expect_completion(cqv, tag(3), 1);
+    cq_verify(cqv);
+  	GPR_ASSERT(state == GRPC_CHANNEL_READY || state == GRPC_CHANNEL_CONNECTING || state == GRPC_CHANNEL_TRANSIENT_FAILURE);
+  }
+
+  /* bring down the server again */
+  /* we should go immediately to TRANSIENT_FAILURE */
+  gpr_log(GPR_DEBUG, "*** SHUTTING DOWN SERVER ***");
+
+  grpc_channel_watch_connectivity_state(
+  	f.client, GRPC_CHANNEL_READY, &state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f.cq, tag(4));
+
+  grpc_server_shutdown_and_notify(f.server, f.cq, tag(0xdead));
+
+  cq_expect_completion(cqv, tag(4), 1);
+  cq_expect_completion(cqv, tag(0xdead), 1);
+  cq_verify(cqv);
+  GPR_ASSERT(state == GRPC_CHANNEL_TRANSIENT_FAILURE);
+
+  /* cleanup server */
+  grpc_server_destroy(f.server);
+
+  gpr_log(GPR_DEBUG, "*** SHUTDOWN SERVER ***");
+
+  grpc_channel_destroy(f.client);
+  grpc_completion_queue_shutdown(f.cq);
+  grpc_completion_queue_destroy(f.cq);
+  config.tear_down_data(&f);
+
+  cq_verifier_destroy(cqv);
+}
+
+void grpc_end2end_tests(grpc_end2end_test_config config) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION);
+  test_connectivity(config);
+}

test/core/end2end/tests/disappearing_server.c

@@ -199,7 +199,6 @@ static void disappearing_server_test(grpc_end2end_test_config config) {
 }
 
 void grpc_end2end_tests(grpc_end2end_test_config config) {
-  if (config.feature_mask & FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION);
   disappearing_server_test(config);
 }
-}

test/core/end2end/tests/simple_delayed_request.c

@@ -207,8 +207,7 @@ static void test_simple_delayed_request_long(grpc_end2end_test_config config) {
 }
 
 void grpc_end2end_tests(grpc_end2end_test_config config) {
-  if (config.feature_mask & FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION) {
+  GPR_ASSERT(config.feature_mask & FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION);
   test_simple_delayed_request_short(config);
   test_simple_delayed_request_long(config);
 }
-}

tools/run_tests/run_tests.py

@@ -346,7 +346,7 @@ _CONFIGS = {
     'dbg': SimpleConfig('dbg'),
     'opt': SimpleConfig('opt'),
     'tsan': SimpleConfig('tsan', environ={
-        'TSAN_OPTIONS': 'suppressions=tools/tsan_suppressions.txt:halt_on_error=1'}),
+        'TSAN_OPTIONS': 'suppressions=tools/tsan_suppressions.txt:halt_on_error=1:second_deadlock_stack=1'}),
     'msan': SimpleConfig('msan'),
     'ubsan': SimpleConfig('ubsan'),
     'asan': SimpleConfig('asan', environ={