Factor out channel state watching

3 files changed, 158 insertions, 166 deletions

diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index dc838de715..4d082aceb8 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -426,88 +426,6 @@ static void cc_on_config_changed(void *arg, int iomgr_success) {
   }
 }
 
-#if 0
-static void channel_op(grpc_channel_element *elem,
-                       grpc_channel_element *from_elem, grpc_channel_op *op) {
-  channel_data *chand = elem->channel_data;
-  grpc_child_channel *child_channel;
-  grpc_channel_op rop;
-  GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
-
-  switch (op->type) {
-    case GRPC_CHANNEL_GOAWAY:
-      /* sending goaway: clear out the active child on the way through */
-      gpr_mu_lock(&chand->mu);
-      child_channel = chand->active_child;
-      chand->active_child = NULL;
-      gpr_mu_unlock(&chand->mu);
-      if (child_channel) {
-        grpc_child_channel_handle_op(child_channel, op);
-        grpc_child_channel_destroy(child_channel, 1);
-      } else {
-        gpr_slice_unref(op->data.goaway.message);
-      }
-      break;
-    case GRPC_CHANNEL_DISCONNECT:
-      /* sending disconnect: clear out the active child on the way through */
-      gpr_mu_lock(&chand->mu);
-      child_channel = chand->active_child;
-      chand->active_child = NULL;
-      gpr_mu_unlock(&chand->mu);
-      if (child_channel) {
-        grpc_child_channel_destroy(child_channel, 1);
-      }
-      /* fake a transport closed to satisfy the refcounting in client */
-      rop.type = GRPC_TRANSPORT_CLOSED;
-      rop.dir = GRPC_CALL_UP;
-      grpc_channel_next_op(elem, &rop);
-      break;
-    case GRPC_TRANSPORT_GOAWAY:
-      /* receiving goaway: if it's from our active child, drop the active child;
-         in all cases consume the event here */
-      gpr_mu_lock(&chand->mu);
-      child_channel = grpc_channel_stack_from_top_element(from_elem);
-      if (child_channel == chand->active_child) {
-        chand->active_child = NULL;
-      } else {
-        child_channel = NULL;
-      }
-      gpr_mu_unlock(&chand->mu);
-      if (child_channel) {
-        grpc_child_channel_destroy(child_channel, 0);
-      }
-      gpr_slice_unref(op->data.goaway.message);
-      break;
-    case GRPC_TRANSPORT_CLOSED:
-      /* receiving disconnect: if it's from our active child, drop the active
-         child; in all cases consume the event here */
-      gpr_mu_lock(&chand->mu);
-      child_channel = grpc_channel_stack_from_top_element(from_elem);
-      if (child_channel == chand->active_child) {
-        chand->active_child = NULL;
-      } else {
-        child_channel = NULL;
-      }
-      gpr_mu_unlock(&chand->mu);
-      if (child_channel) {
-        grpc_child_channel_destroy(child_channel, 0);
-      }
-      break;
-    default:
-      switch (op->dir) {
-        case GRPC_CALL_UP:
-          grpc_channel_next_op(elem, op);
-          break;
-        case GRPC_CALL_DOWN:
-          gpr_log(GPR_ERROR, "unhandled channel op: %d", op->type);
-          abort();
-          break;
-      }
-      break;
-  }
-}
-#endif
-
 static void cc_start_transport_op(grpc_channel_element *elem, grpc_transport_op *op) {}
 
 /* Constructor for call_data */
@@ -591,90 +509,6 @@ const grpc_channel_filter grpc_client_channel_filter = {
     init_channel_elem,     destroy_channel_elem, "client-channel",
 };
 
-#if 0
-grpc_transport_setup_result grpc_client_channel_transport_setup_complete(
-    grpc_channel_stack *channel_stack, grpc_transport *transport,
-    grpc_channel_filter const **channel_filters, size_t num_channel_filters,
-    grpc_mdctx *mdctx) {
-  /* we just got a new transport: lets create a child channel stack for it */
-  grpc_channel_element *elem = grpc_channel_stack_last_element(channel_stack);
-  channel_data *chand = elem->channel_data;
-  size_t num_child_filters = 2 + num_channel_filters;
-  grpc_channel_filter const **child_filters;
-  grpc_transport_setup_result result;
-  grpc_child_channel *old_active = NULL;
-  call_data **waiting_children;
-  size_t waiting_child_count;
-  size_t i;
-  grpc_transport_stream_op *call_ops;
-
-  /* build the child filter stack */
-  child_filters = gpr_malloc(sizeof(grpc_channel_filter *) * num_child_filters);
-  /* we always need a link back filter to get back to the connected channel */
-  child_filters[0] = &grpc_child_channel_top_filter;
-  for (i = 0; i < num_channel_filters; i++) {
-    child_filters[i + 1] = channel_filters[i];
-  }
-  /* and we always need a connected channel to talk to the transport */
-  child_filters[num_child_filters - 1] = &grpc_connected_channel_filter;
-
-  GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
-
-  /* BEGIN LOCKING CHANNEL */
-  gpr_mu_lock(&chand->mu);
-  chand->transport_setup_initiated = 0;
-
-  if (chand->active_child) {
-    old_active = chand->active_child;
-  }
-  chand->active_child = grpc_child_channel_create(
-      elem, child_filters, num_child_filters, chand->args, mdctx);
-  result =
-      grpc_connected_channel_bind_transport(chand->active_child, transport);
-
-  /* capture the waiting children - we'll activate them outside the lock
-     to avoid re-entrancy problems */
-  waiting_children = chand->waiting_children;
-  waiting_child_count = chand->waiting_child_count;
-  /* bumping up inflight_requests here avoids taking a lock per rpc below */
-
-  chand->waiting_children = NULL;
-  chand->waiting_child_count = 0;
-  chand->waiting_child_capacity = 0;
-
-  call_ops = gpr_malloc(sizeof(*call_ops) * waiting_child_count);
-
-  for (i = 0; i < waiting_child_count; i++) {
-    call_ops[i] = waiting_children[i]->waiting_op;
-    if (!prepare_activate(waiting_children[i]->elem, chand->active_child)) {
-      waiting_children[i] = NULL;
-      grpc_transport_stream_op_finish_with_failure(&call_ops[i]);
-    }
-  }
-
-  /* END LOCKING CHANNEL */
-  gpr_mu_unlock(&chand->mu);
-
-  /* activate any pending operations - this is safe to do as we guarantee one
-     and only one write operation per request at the surface api - if we lose
-     that guarantee we need to do some curly locking here */
-  for (i = 0; i < waiting_child_count; i++) {
-    if (waiting_children[i]) {
-      complete_activate(waiting_children[i]->elem, &call_ops[i]);
-    }
-  }
-  gpr_free(waiting_children);
-  gpr_free(call_ops);
-  gpr_free(child_filters);
-
-  if (old_active) {
-    grpc_child_channel_destroy(old_active, 1);
-  }
-
-  return result;
-}
-#endif
-
 void grpc_client_channel_set_resolver(grpc_channel_stack *channel_stack,
                                       grpc_resolver *resolver) {
   /* post construction initialization: set the transport setup pointer */
diff --git a/src/core/channel/connectivity_state.c b/src/core/channel/connectivity_state.c
new file mode 100644
index 0000000000..566a2c3344
--- /dev/null
+++ b/src/core/channel/connectivity_state.c
@@ -0,0 +1,92 @@
+/*
+ *
+ * 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 "src/core/channel/connectivity_state.h"
+#include <grpc/support/alloc.h>
+
+void grpc_connectivity_state_init(grpc_connectivity_state_tracker *tracker, grpc_connectivity_state init_state) {
+	tracker->current_state = init_state;
+	tracker->watchers = NULL;
+}
+
+void grpc_connectivity_state_destroy(grpc_connectivity_state_tracker *tracker) {
+  grpc_connectivity_state_watcher *w;
+  while ((w = tracker->watchers)) {
+  	tracker->watchers = w->next;
+
+  	if (GRPC_CHANNEL_FATAL_FAILURE != *w->current) {
+  		*w->current = GRPC_CHANNEL_FATAL_FAILURE;
+  		grpc_iomgr_add_callback(w->notify);
+  	} else {
+  		grpc_iomgr_add_delayed_callback(w->notify, 0);
+  	}
+  	gpr_free(w);
+  }
+}
+
+grpc_connectivity_state grpc_connectivity_state_check(grpc_connectivity_state_tracker *tracker) {
+	return tracker->current_state;
+}
+
+int grpc_connectivity_state_notify_on_state_change(grpc_connectivity_state_tracker *tracker, grpc_connectivity_state *current, grpc_iomgr_closure *notify) {
+  if (tracker->current_state != *current) {
+    *current = tracker->current_state;
+    grpc_iomgr_add_callback(notify);
+  } else {
+	  grpc_connectivity_state_watcher *w = gpr_malloc(sizeof(*w));
+	  w->current = current;
+	  w->notify = notify;
+    w->next = tracker->watchers;
+    tracker->watchers = w;
+  }
+  return tracker->current_state == GRPC_CHANNEL_IDLE;
+}
+
+void grpc_connectivity_state_set(grpc_connectivity_state_tracker *tracker, grpc_connectivity_state state) {
+  grpc_connectivity_state_watcher *new = NULL;
+  grpc_connectivity_state_watcher *w;
+  tracker->current_state = state;
+  while ((w = tracker->watchers)) {
+    tracker->watchers = w->next;
+
+    if (state != *w->current) {
+      *w->current = state;
+      grpc_iomgr_add_callback(w->notify);
+      gpr_free(w);
+    } else {
+      w->next = new;
+      new = w;
+    }
+  }
+  tracker->watchers = new;
+}
diff --git a/src/core/channel/connectivity_state.h b/src/core/channel/connectivity_state.h
new file mode 100644
index 0000000000..ebc1acc559
--- /dev/null
+++ b/src/core/channel/connectivity_state.h
@@ -0,0 +1,66 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CORE_CHANNEL_CONNECTIVITY_STATE_H
+#define GRPC_INTERNAL_CORE_CHANNEL_CONNECTIVITY_STATE_H
+
+#include <grpc/grpc.h>
+#include "src/core/iomgr/iomgr.h"
+
+typedef struct grpc_connectivity_state_watcher {
+	/** we keep watchers in a linked list */
+  struct grpc_connectivity_state_watcher *next;
+  /** closure to notify on change */
+  grpc_iomgr_closure *notify;
+  /** the current state as believed by the watcher */
+  grpc_connectivity_state *current;
+} grpc_connectivity_state_watcher;
+
+typedef struct {
+	/** current connectivity state */
+  grpc_connectivity_state current_state;
+  /** all our watchers */
+  grpc_connectivity_state_watcher *watchers;
+} grpc_connectivity_state_tracker;
+
+void grpc_connectivity_state_init(grpc_connectivity_state_tracker *tracker, grpc_connectivity_state init_state);
+void grpc_connectivity_state_destroy(grpc_connectivity_state_tracker *tracker);
+
+void grpc_connectivity_state_set(grpc_connectivity_state_tracker *tracker, grpc_connectivity_state state);
+
+grpc_connectivity_state grpc_connectivity_state_check(grpc_connectivity_state_tracker *tracker);
+
+/** Return 1 if the channel should start connecting, 0 otherwise */
+int grpc_connectivity_state_notify_on_state_change(grpc_connectivity_state_tracker *tracker, grpc_connectivity_state *current, grpc_iomgr_closure *notify);
+
+#endif /* GRPC_INTERNAL_CORE_CHANNEL_CONNECTIVITY_STATE_H */