Plumbing transport_op changes through

7 files changed, 110 insertions, 195 deletions

diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index ddff3efb32..7a8eb8c54f 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -1154,7 +1154,7 @@ static void execute_op(grpc_call *call, grpc_transport_stream_op *op) {
 
   elem = CALL_ELEM_FROM_CALL(call, 0);
   op->context = call->context;
-  elem->filter->start_transport_op(elem, op);
+  elem->filter->start_transport_stream_op(elem, op);
 }
 
 grpc_call *grpc_call_from_top_element(grpc_call_element *elem) {
diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c
index a3c4dcebc1..6c4b407a85 100644
--- a/src/core/surface/channel.c
+++ b/src/core/surface/channel.c
@@ -39,7 +39,6 @@
 #include "src/core/iomgr/iomgr.h"
 #include "src/core/support/string.h"
 #include "src/core/surface/call.h"
-#include "src/core/surface/client.h"
 #include "src/core/surface/init.h"
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
@@ -238,22 +237,15 @@ void grpc_channel_internal_unref(grpc_channel *channel) {
   }
 }
 
-void grpc_channel_destroy(grpc_channel *channel) {
-  grpc_channel_op op;
-  grpc_channel_element *elem;
-
-  elem = grpc_channel_stack_element(CHANNEL_STACK_FROM_CHANNEL(channel), 0);
-
-  op.type = GRPC_CHANNEL_GOAWAY;
-  op.dir = GRPC_CALL_DOWN;
-  op.data.goaway.status = GRPC_STATUS_OK;
-  op.data.goaway.message = gpr_slice_from_copied_string("Client disconnect");
-  elem->filter->channel_op(elem, NULL, &op);
-
-  op.type = GRPC_CHANNEL_DISCONNECT;
-  op.dir = GRPC_CALL_DOWN;
-  elem->filter->channel_op(elem, NULL, &op);
+static void execute_op(grpc_channel *channel, grpc_transport_op *op) {
+  abort();
+}
 
+void grpc_channel_destroy(grpc_channel *channel) {
+  grpc_transport_op op;
+  memset(&op, 0, sizeof(op));
+  op.disconnect = 1;
+  execute_op(channel, &op);
   GRPC_CHANNEL_INTERNAL_UNREF(channel, "channel");
 }
 
diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c
index 8efd86b9f6..14ff63a2e3 100644
--- a/src/core/surface/channel_create.c
+++ b/src/core/surface/channel_create.c
@@ -38,9 +38,7 @@
 
 #include "src/core/channel/client_channel.h"
 #include "src/core/client_config/resolver_registry.h"
-#include "src/core/client_config/subchannels/tcp_subchannel.h"
 #include "src/core/surface/channel.h"
-#include "src/core/surface/client.h"
 
 /* Create a client channel:
    Asynchronously: - resolve target
@@ -53,7 +51,6 @@ grpc_channel *grpc_channel_create(const char *target,
   const grpc_channel_filter *filters[MAX_FILTERS];
   grpc_resolver *resolver;
   int n = 0;
-  filters[n++] = &grpc_client_surface_filter;
   /* TODO(census)
   if (grpc_channel_args_is_census_enabled(args)) {
     filters[n++] = &grpc_client_census_filter;
@@ -61,7 +58,8 @@ grpc_channel *grpc_channel_create(const char *target,
   filters[n++] = &grpc_client_channel_filter;
   GPR_ASSERT(n <= MAX_FILTERS);
 
-  resolver = grpc_resolver_create(target, grpc_create_tcp_subchannel_factory());
+  GPR_ASSERT(!"NULL should be a subchannel factory creation below");
+  resolver = grpc_resolver_create(target, NULL);
   if (!resolver) {
     return NULL;
   }
diff --git a/src/core/surface/client.c b/src/core/surface/client.c
deleted file mode 100644
index 9c9cba5771..0000000000
--- a/src/core/surface/client.c
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- *
- * 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/surface/client.h"
-
-#include "src/core/surface/call.h"
-#include "src/core/surface/channel.h"
-#include "src/core/support/string.h"
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
-
-typedef struct { void *unused; } call_data;
-
-typedef struct { void *unused; } channel_data;
-
-static void client_start_transport_op(grpc_call_element *elem,
-                                      grpc_transport_stream_op *op) {
-  GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
-  grpc_call_next_op(elem, op);
-}
-
-static void channel_op(grpc_channel_element *elem,
-                       grpc_channel_element *from_elem, grpc_channel_op *op) {
-  switch (op->type) {
-    case GRPC_ACCEPT_CALL:
-      gpr_log(GPR_ERROR, "Client cannot accept new calls");
-      break;
-    case GRPC_TRANSPORT_CLOSED:
-      grpc_client_channel_closed(elem);
-      break;
-    case GRPC_TRANSPORT_GOAWAY:
-      gpr_slice_unref(op->data.goaway.message);
-      break;
-    default:
-      GPR_ASSERT(op->dir == GRPC_CALL_DOWN);
-      grpc_channel_next_op(elem, op);
-  }
-}
-
-static void init_call_elem(grpc_call_element *elem,
-                           const void *transport_server_data,
-                           grpc_transport_stream_op *initial_op) {}
-
-static void destroy_call_elem(grpc_call_element *elem) {}
-
-static void init_channel_elem(grpc_channel_element *elem,
-                              const grpc_channel_args *args, grpc_mdctx *mdctx,
-                              int is_first, int is_last) {
-  GPR_ASSERT(is_first);
-  GPR_ASSERT(!is_last);
-}
-
-static void destroy_channel_elem(grpc_channel_element *elem) {}
-
-const grpc_channel_filter grpc_client_surface_filter = {
-    client_start_transport_op, channel_op, sizeof(call_data), init_call_elem,
-    destroy_call_elem, sizeof(channel_data), init_channel_elem,
-    destroy_channel_elem, "client",
-};
diff --git a/src/core/surface/client.h b/src/core/surface/client.h
deleted file mode 100644
index 9db2ccf3d2..0000000000
--- a/src/core/surface/client.h
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- *
- * 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_SURFACE_CLIENT_H
-#define GRPC_INTERNAL_CORE_SURFACE_CLIENT_H
-
-#include "src/core/channel/channel_stack.h"
-
-extern const grpc_channel_filter grpc_client_surface_filter;
-
-#endif  /* GRPC_INTERNAL_CORE_SURFACE_CLIENT_H */
diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c
index 4b55e9dc91..5235d3f7f4 100644
--- a/src/core/surface/lame_client.c
+++ b/src/core/surface/lame_client.c
@@ -49,16 +49,16 @@ typedef struct {
 
 typedef struct { grpc_mdctx *mdctx; } channel_data;
 
-static void lame_start_transport_op(grpc_call_element *elem,
-                                    grpc_transport_stream_op *op) {
+static void lame_start_transport_stream_op(grpc_call_element *elem,
+                                           grpc_transport_stream_op *op) {
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
   GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
-  if (op->send_ops) {
+  if (op->send_ops != NULL) {
     grpc_stream_ops_unref_owned_objects(op->send_ops->ops, op->send_ops->nops);
     op->on_done_send->cb(op->on_done_send->cb_arg, 0);
   }
-  if (op->recv_ops) {
+  if (op->recv_ops != NULL) {
     char tmp[GPR_LTOA_MIN_BUFSIZE];
     grpc_metadata_batch mdb;
     gpr_ltoa(GRPC_STATUS_UNKNOWN, tmp);
@@ -77,22 +77,21 @@ static void lame_start_transport_op(grpc_call_element *elem,
     *op->recv_state = GRPC_STREAM_CLOSED;
     op->on_done_recv->cb(op->on_done_recv->cb_arg, 1);
   }
-  if (op->on_consumed) {
+  if (op->on_consumed != NULL) {
     op->on_consumed->cb(op->on_consumed->cb_arg, 0);
   }
 }
 
-static void channel_op(grpc_channel_element *elem,
-                       grpc_channel_element *from_elem, grpc_channel_op *op) {
-  switch (op->type) {
-    case GRPC_CHANNEL_GOAWAY:
-      gpr_slice_unref(op->data.goaway.message);
-      break;
-    case GRPC_CHANNEL_DISCONNECT:
-      grpc_client_channel_closed(elem);
-      break;
-    default:
-      break;
+static void lame_start_transport_op(grpc_channel_element *elem,
+                                    grpc_transport_op *op) {
+  if (op->on_connectivity_state_change) {
+    GPR_ASSERT(*op->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE);
+    *op->connectivity_state = GRPC_CHANNEL_FATAL_FAILURE;
+    op->on_connectivity_state_change->cb(
+        op->on_connectivity_state_change->cb_arg, 1);
+  }
+  if (op->on_consumed != NULL) {
+    op->on_consumed->cb(op->on_consumed->cb_arg, 1);
   }
 }
 
@@ -118,9 +117,15 @@ static void init_channel_elem(grpc_channel_element *elem,
 static void destroy_channel_elem(grpc_channel_element *elem) {}
 
 static const grpc_channel_filter lame_filter = {
-    lame_start_transport_op, channel_op,           sizeof(call_data),
-    init_call_elem,          destroy_call_elem,    sizeof(channel_data),
-    init_channel_elem,       destroy_channel_elem, "lame-client",
+    lame_start_transport_stream_op,
+    lame_start_transport_op,
+    sizeof(call_data),
+    init_call_elem,
+    destroy_call_elem,
+    sizeof(channel_data),
+    init_channel_elem,
+    destroy_channel_elem,
+    "lame-client",
 };
 
 grpc_channel *grpc_lame_client_channel_create(void) {
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 607344a7a6..568f7925dd 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -115,6 +115,7 @@ typedef struct channel_registered_method {
 struct channel_data {
   grpc_server *server;
   size_t num_calls;
+  grpc_connectivity_state connectivity_state;
   grpc_channel *channel;
   grpc_mdstr *path_key;
   grpc_mdstr *authority_key;
@@ -125,6 +126,7 @@ struct channel_data {
   gpr_uint32 registered_method_slots;
   gpr_uint32 registered_method_max_probes;
   grpc_iomgr_closure finish_destroy_channel_closure;
+  grpc_iomgr_closure channel_connectivity_changed;
 };
 
 typedef struct shutdown_tag {
@@ -539,13 +541,41 @@ static void server_mutate_op(grpc_call_element *elem,
   }
 }
 
-static void server_start_transport_op(grpc_call_element *elem,
-                                      grpc_transport_stream_op *op) {
+static void server_start_transport_stream_op(grpc_call_element *elem,
+                                             grpc_transport_stream_op *op) {
   GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
   server_mutate_op(elem, op);
   grpc_call_next_op(elem, op);
 }
 
+static void accept_stream(void *cd, grpc_transport *transport,
+                          const void *transport_server_data) {
+  channel_data *chand = cd;
+  /* create a call */
+  grpc_call_create(chand->channel, NULL, transport_server_data, NULL, 0,
+                   gpr_inf_future);
+}
+
+static void channel_connectivity_changed(void *cd, int iomgr_status_ignored) {
+  channel_data *chand = cd;
+  grpc_server *server = chand->server;
+  if (chand->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) {
+    grpc_transport_op op;
+    memset(&op, 0, sizeof(op));
+    op.on_connectivity_state_change = &chand->channel_connectivity_changed,
+    op.connectivity_state = &chand->connectivity_state;
+    grpc_channel_next_op(grpc_channel_stack_element(
+                             grpc_channel_get_channel_stack(chand->channel), 0),
+                         &op);
+  } else {
+    gpr_mu_lock(&server->mu_global);
+    destroy_channel(chand);
+    gpr_mu_unlock(&server->mu_global);
+    GRPC_CHANNEL_INTERNAL_UNREF(chand->channel, "connectivity");
+  }
+}
+
+#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;
@@ -576,39 +606,45 @@ static void channel_op(grpc_channel_element *elem,
       break;
   }
 }
+#endif
 
 typedef struct {
   channel_data *chand;
   int send_goaway;
   int send_disconnect;
   grpc_iomgr_closure finish_shutdown_channel_closure;
+
+  /* for use during shutdown: the goaway message to send */
+  gpr_slice goaway_message;
 } shutdown_channel_args;
 
-static void finish_shutdown_channel(void *p, int success) {
+static void destroy_shutdown_channel_args(void *p, int success) {
   shutdown_channel_args *sca = p;
-  grpc_channel_op op;
-
-  if (sca->send_goaway) {
-    op.type = GRPC_CHANNEL_GOAWAY;
-    op.dir = GRPC_CALL_DOWN;
-    op.data.goaway.status = GRPC_STATUS_OK;
-    op.data.goaway.message = gpr_slice_from_copied_string("Server shutdown");
-    channel_op(grpc_channel_stack_element(
-                   grpc_channel_get_channel_stack(sca->chand->channel), 0),
-               NULL, &op);
-  }
-  if (sca->send_disconnect) {
-    op.type = GRPC_CHANNEL_DISCONNECT;
-    op.dir = GRPC_CALL_DOWN;
-    channel_op(grpc_channel_stack_element(
-                   grpc_channel_get_channel_stack(sca->chand->channel), 0),
-               NULL, &op);
-  }
   GRPC_CHANNEL_INTERNAL_UNREF(sca->chand->channel, "shutdown");
-
+  gpr_slice_unref(sca->goaway_message);
   gpr_free(sca);
 }
 
+static void finish_shutdown_channel(void *p, int success) {
+  shutdown_channel_args *sca = p;
+  grpc_transport_op op;
+  memset(&op, 0, sizeof(op));
+
+  op.send_goaway = sca->send_goaway;
+  sca->goaway_message = gpr_slice_from_copied_string("Server shutdown");
+  op.goaway_message = &sca->goaway_message;
+  op.goaway_status = GRPC_STATUS_OK;
+  op.disconnect = sca->send_disconnect;
+  grpc_iomgr_closure_init(&sca->finish_shutdown_channel_closure,
+                          destroy_shutdown_channel_args, sca);
+  op.on_consumed = &sca->finish_shutdown_channel_closure;
+
+  grpc_channel_next_op(
+      grpc_channel_stack_element(
+          grpc_channel_get_channel_stack(sca->chand->channel), 0),
+      &op);
+}
+
 static void shutdown_channel(channel_data *chand, int send_goaway,
                              int send_disconnect) {
   shutdown_channel_args *sca;
@@ -687,6 +723,9 @@ static void init_channel_elem(grpc_channel_element *elem,
   chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority");
   chand->next = chand->prev = chand;
   chand->registered_methods = NULL;
+  chand->connectivity_state = GRPC_CHANNEL_IDLE;
+  grpc_iomgr_closure_init(&chand->channel_connectivity_changed,
+                          channel_connectivity_changed, chand);
 }
 
 static void destroy_channel_elem(grpc_channel_element *elem) {
@@ -717,8 +756,8 @@ static void destroy_channel_elem(grpc_channel_element *elem) {
 }
 
 static const grpc_channel_filter server_surface_filter = {
-    server_start_transport_op,
-    channel_op,
+    server_start_transport_stream_op,
+    grpc_channel_next_op,
     sizeof(call_data),
     init_call_elem,
     destroy_call_elem,
@@ -852,6 +891,7 @@ grpc_transport_setup_result grpc_server_setup_transport(
   gpr_uint32 slots;
   gpr_uint32 probes;
   gpr_uint32 max_probes = 0;
+  grpc_transport_op op;
   grpc_transport_setup_result result;
 
   for (i = 0; i < s->channel_filter_count; i++) {
@@ -863,7 +903,9 @@ grpc_transport_setup_result grpc_server_setup_transport(
   filters[i] = &grpc_connected_channel_filter;
 
   for (i = 0; i < s->cq_count; i++) {
-    grpc_transport_add_to_pollset(transport, grpc_cq_pollset(s->cqs[i]));
+    memset(&op, 0, sizeof(op));
+    op.bind_pollset = grpc_cq_pollset(s->cqs[i]);
+    grpc_transport_perform_op(transport, &op);
   }
 
   channel =
@@ -875,6 +917,14 @@ grpc_transport_setup_result grpc_server_setup_transport(
   server_ref(s);
   chand->channel = channel;
 
+  GRPC_CHANNEL_INTERNAL_REF(channel, "connectivity");
+  memset(&op, 0, sizeof(op));
+  op.set_accept_stream = accept_stream;
+  op.set_accept_stream_user_data = chand;
+  op.on_connectivity_state_change = &chand->channel_connectivity_changed;
+  op.connectivity_state = &chand->connectivity_state;
+  grpc_transport_perform_op(transport, &op);
+
   num_registered_methods = 0;
   for (rm = s->registered_methods; rm; rm = rm->next) {
     num_registered_methods++;