Merge branch 'master' of https://github.com/grpc/grpc into master-upmerge-from-deep-under

13 files changed, 589 insertions, 170 deletions

diff --git a/src/core/lib/channel/channel_stack.c b/src/core/lib/channel/channel_stack.c
index 0655b9353f..57d34d9e9a 100644
--- a/src/core/lib/channel/channel_stack.c
+++ b/src/core/lib/channel/channel_stack.c
@@ -158,13 +158,11 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx,
   }
 }
 
-grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx,
-                                 grpc_channel_stack *channel_stack,
-                                 int initial_refs, grpc_iomgr_cb_func destroy,
-                                 void *destroy_arg,
-                                 grpc_call_context_element *context,
-                                 const void *transport_server_data,
-                                 grpc_call_stack *call_stack) {
+grpc_error *grpc_call_stack_init(
+    grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack,
+    int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg,
+    grpc_call_context_element *context, const void *transport_server_data,
+    gpr_timespec deadline, grpc_call_stack *call_stack) {
   grpc_channel_element *channel_elems = CHANNEL_ELEMS_FROM_STACK(channel_stack);
   grpc_call_element_args args;
   size_t count = channel_stack->count;
@@ -185,6 +183,7 @@ grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx,
     args.call_stack = call_stack;
     args.server_transport_data = transport_server_data;
     args.context = context;
+    args.deadline = deadline;
     call_elems[i].filter = channel_elems[i].filter;
     call_elems[i].channel_data = channel_elems[i].channel_data;
     call_elems[i].call_data = user_data;
@@ -276,16 +275,16 @@ static void destroy_op(grpc_exec_ctx *exec_ctx, void *op, grpc_error *error) {
 }
 
 void grpc_call_element_send_cancel(grpc_exec_ctx *exec_ctx,
-                                   grpc_call_element *cur_elem) {
+                                   grpc_call_element *elem) {
   grpc_transport_stream_op *op = gpr_malloc(sizeof(*op));
   memset(op, 0, sizeof(*op));
   op->cancel_error = GRPC_ERROR_CANCELLED;
   op->on_complete = grpc_closure_create(destroy_op, op);
-  grpc_call_next_op(exec_ctx, cur_elem, op);
+  elem->filter->start_transport_stream_op(exec_ctx, elem, op);
 }
 
 void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx,
-                                                grpc_call_element *cur_elem,
+                                                grpc_call_element *elem,
                                                 grpc_status_code status,
                                                 gpr_slice *optional_message) {
   grpc_transport_stream_op *op = gpr_malloc(sizeof(*op));
@@ -293,5 +292,16 @@ void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx,
   op->on_complete = grpc_closure_create(destroy_op, op);
   grpc_transport_stream_op_add_cancellation_with_message(op, status,
                                                          optional_message);
-  grpc_call_next_op(exec_ctx, cur_elem, op);
+  elem->filter->start_transport_stream_op(exec_ctx, elem, op);
+}
+
+void grpc_call_element_send_close_with_message(grpc_exec_ctx *exec_ctx,
+                                               grpc_call_element *elem,
+                                               grpc_status_code status,
+                                               gpr_slice *optional_message) {
+  grpc_transport_stream_op *op = gpr_malloc(sizeof(*op));
+  memset(op, 0, sizeof(*op));
+  op->on_complete = grpc_closure_create(destroy_op, op);
+  grpc_transport_stream_op_add_close(op, status, optional_message);
+  elem->filter->start_transport_stream_op(exec_ctx, elem, op);
 }
diff --git a/src/core/lib/channel/channel_stack.h b/src/core/lib/channel/channel_stack.h
index 6b73cce380..1cfe2885d8 100644
--- a/src/core/lib/channel/channel_stack.h
+++ b/src/core/lib/channel/channel_stack.h
@@ -74,6 +74,7 @@ typedef struct {
   grpc_call_stack *call_stack;
   const void *server_transport_data;
   grpc_call_context_element *context;
+  gpr_timespec deadline;
 } grpc_call_element_args;
 
 typedef struct {
@@ -220,13 +221,11 @@ void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx,
 /* Initialize a call stack given a channel stack. transport_server_data is
    expected to be NULL on a client, or an opaque transport owned pointer on the
    server. */
-grpc_error *grpc_call_stack_init(grpc_exec_ctx *exec_ctx,
-                                 grpc_channel_stack *channel_stack,
-                                 int initial_refs, grpc_iomgr_cb_func destroy,
-                                 void *destroy_arg,
-                                 grpc_call_context_element *context,
-                                 const void *transport_server_data,
-                                 grpc_call_stack *call_stack);
+grpc_error *grpc_call_stack_init(
+    grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack,
+    int initial_refs, grpc_iomgr_cb_func destroy, void *destroy_arg,
+    grpc_call_context_element *context, const void *transport_server_data,
+    gpr_timespec deadline, grpc_call_stack *call_stack);
 /* Set a pollset or a pollset_set for a call stack: must occur before the first
  * op is started */
 void grpc_call_stack_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx,
@@ -290,6 +289,11 @@ void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx,
                                                 grpc_status_code status,
                                                 gpr_slice *optional_message);
 
+void grpc_call_element_send_close_with_message(grpc_exec_ctx *exec_ctx,
+                                               grpc_call_element *cur_elem,
+                                               grpc_status_code status,
+                                               gpr_slice *optional_message);
+
 extern int grpc_trace_channel;
 
 #define GRPC_CALL_LOG_OP(sev, elem, op) \
diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c
new file mode 100644
index 0000000000..079b98a2f8
--- /dev/null
+++ b/src/core/lib/channel/deadline_filter.c
@@ -0,0 +1,302 @@
+//
+// Copyright 2016, 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/lib/channel/deadline_filter.h"
+
+#include <stdbool.h>
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/sync.h>
+#include <grpc/support/time.h>
+
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/iomgr/timer.h"
+
+//
+// grpc_deadline_state
+//
+
+// Timer callback.
+static void timer_callback(grpc_exec_ctx* exec_ctx, void* arg,
+                           grpc_error* error) {
+  grpc_call_element* elem = arg;
+  grpc_deadline_state* deadline_state = elem->call_data;
+  gpr_mu_lock(&deadline_state->timer_mu);
+  deadline_state->timer_pending = false;
+  gpr_mu_unlock(&deadline_state->timer_mu);
+  if (error != GRPC_ERROR_CANCELLED) {
+    gpr_slice msg = gpr_slice_from_static_string("Deadline Exceeded");
+    grpc_call_element_send_cancel_with_message(
+        exec_ctx, elem, GRPC_STATUS_DEADLINE_EXCEEDED, &msg);
+    gpr_slice_unref(msg);
+  }
+  GRPC_CALL_STACK_UNREF(exec_ctx, deadline_state->call_stack, "deadline_timer");
+}
+
+// Starts the deadline timer.
+static void start_timer_if_needed(grpc_exec_ctx* exec_ctx,
+                                  grpc_call_element* elem,
+                                  gpr_timespec deadline) {
+  grpc_deadline_state* deadline_state = elem->call_data;
+  deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
+  if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_MONOTONIC)) != 0) {
+    // Take a reference to the call stack, to be owned by the timer.
+    GRPC_CALL_STACK_REF(deadline_state->call_stack, "deadline_timer");
+    gpr_mu_lock(&deadline_state->timer_mu);
+    deadline_state->timer_pending = true;
+    grpc_timer_init(exec_ctx, &deadline_state->timer, deadline, timer_callback,
+                    elem, gpr_now(GPR_CLOCK_MONOTONIC));
+    gpr_mu_unlock(&deadline_state->timer_mu);
+  }
+}
+
+// Cancels the deadline timer.
+static void cancel_timer_if_needed(grpc_exec_ctx* exec_ctx,
+                                   grpc_deadline_state* deadline_state) {
+  gpr_mu_lock(&deadline_state->timer_mu);
+  if (deadline_state->timer_pending) {
+    grpc_timer_cancel(exec_ctx, &deadline_state->timer);
+    deadline_state->timer_pending = false;
+  }
+  gpr_mu_unlock(&deadline_state->timer_mu);
+}
+
+// Callback run when the call is complete.
+static void on_complete(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) {
+  grpc_deadline_state* deadline_state = arg;
+  cancel_timer_if_needed(exec_ctx, deadline_state);
+  // Invoke the next callback.
+  deadline_state->next_on_complete->cb(
+      exec_ctx, deadline_state->next_on_complete->cb_arg, error);
+}
+
+// Inject our own on_complete callback into op.
+static void inject_on_complete_cb(grpc_deadline_state* deadline_state,
+                                  grpc_transport_stream_op* op) {
+  deadline_state->next_on_complete = op->on_complete;
+  grpc_closure_init(&deadline_state->on_complete, on_complete, deadline_state);
+  op->on_complete = &deadline_state->on_complete;
+}
+
+// Callback and associated state for starting the timer after call stack
+// initialization has been completed.
+struct start_timer_after_init_state {
+  grpc_call_element* elem;
+  gpr_timespec deadline;
+  grpc_closure closure;
+};
+static void start_timer_after_init(grpc_exec_ctx* exec_ctx, void* arg,
+                                   grpc_error* error) {
+  struct start_timer_after_init_state* state = arg;
+  start_timer_if_needed(exec_ctx, state->elem, state->deadline);
+  gpr_free(state);
+}
+
+void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
+                              grpc_call_element_args* args) {
+  grpc_deadline_state* deadline_state = elem->call_data;
+  memset(deadline_state, 0, sizeof(*deadline_state));
+  deadline_state->call_stack = args->call_stack;
+  gpr_mu_init(&deadline_state->timer_mu);
+  // Deadline will always be infinite on servers, so the timer will only be
+  // set on clients with a finite deadline.
+  const gpr_timespec deadline =
+      gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC);
+  if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_MONOTONIC)) != 0) {
+    // When the deadline passes, we indicate the failure by sending down
+    // an op with cancel_error set.  However, we can't send down any ops
+    // until after the call stack is fully initialized.  If we start the
+    // timer here, we have no guarantee that the timer won't pop before
+    // call stack initialization is finished.  To avoid that problem, we
+    // create a closure to start the timer, and we schedule that closure
+    // to be run after call stack initialization is done.
+    struct start_timer_after_init_state* state = gpr_malloc(sizeof(*state));
+    state->elem = elem;
+    state->deadline = deadline;
+    grpc_closure_init(&state->closure, start_timer_after_init, state);
+    grpc_exec_ctx_sched(exec_ctx, &state->closure, GRPC_ERROR_NONE, NULL);
+  }
+}
+
+void grpc_deadline_state_destroy(grpc_exec_ctx* exec_ctx,
+                                 grpc_call_element* elem) {
+  grpc_deadline_state* deadline_state = elem->call_data;
+  cancel_timer_if_needed(exec_ctx, deadline_state);
+  gpr_mu_destroy(&deadline_state->timer_mu);
+}
+
+void grpc_deadline_state_client_start_transport_stream_op(
+    grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
+    grpc_transport_stream_op* op) {
+  grpc_deadline_state* deadline_state = elem->call_data;
+  if (op->cancel_error != GRPC_ERROR_NONE ||
+      op->close_error != GRPC_ERROR_NONE) {
+    cancel_timer_if_needed(exec_ctx, deadline_state);
+  } else {
+    // Make sure we know when the call is complete, so that we can cancel
+    // the timer.
+    if (op->recv_trailing_metadata != NULL) {
+      inject_on_complete_cb(deadline_state, op);
+    }
+  }
+}
+
+//
+// filter code
+//
+
+// Constructor for channel_data.  Used for both client and server filters.
+static void init_channel_elem(grpc_exec_ctx* exec_ctx,
+                              grpc_channel_element* elem,
+                              grpc_channel_element_args* args) {
+  GPR_ASSERT(!args->is_last);
+}
+
+// Destructor for channel_data.  Used for both client and server filters.
+static void destroy_channel_elem(grpc_exec_ctx* exec_ctx,
+                                 grpc_channel_element* elem) {}
+
+// Call data used for both client and server filter.
+typedef struct base_call_data {
+  grpc_deadline_state deadline_state;
+} base_call_data;
+
+// Additional call data used only for the server filter.
+typedef struct server_call_data {
+  base_call_data base;  // Must be first.
+  // The closure for receiving initial metadata.
+  grpc_closure recv_initial_metadata_ready;
+  // Received initial metadata batch.
+  grpc_metadata_batch* recv_initial_metadata;
+  // The original recv_initial_metadata_ready closure, which we chain to
+  // after our own closure is invoked.
+  grpc_closure* next_recv_initial_metadata_ready;
+} server_call_data;
+
+// Constructor for call_data.  Used for both client and server filters.
+static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx,
+                                  grpc_call_element* elem,
+                                  grpc_call_element_args* args) {
+  // Note: size of call data is different between client and server.
+  memset(elem->call_data, 0, elem->filter->sizeof_call_data);
+  grpc_deadline_state_init(exec_ctx, elem, args);
+  return GRPC_ERROR_NONE;
+}
+
+// Destructor for call_data.  Used for both client and server filters.
+static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
+                              const grpc_call_final_info* final_info,
+                              void* and_free_memory) {
+  grpc_deadline_state_destroy(exec_ctx, elem);
+}
+
+// Method for starting a call op for client filter.
+static void client_start_transport_stream_op(grpc_exec_ctx* exec_ctx,
+                                             grpc_call_element* elem,
+                                             grpc_transport_stream_op* op) {
+  grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op);
+  // Chain to next filter.
+  grpc_call_next_op(exec_ctx, elem, op);
+}
+
+// Callback for receiving initial metadata on the server.
+static void recv_initial_metadata_ready(grpc_exec_ctx* exec_ctx, void* arg,
+                                        grpc_error* error) {
+  grpc_call_element* elem = arg;
+  server_call_data* calld = elem->call_data;
+  // Get deadline from metadata and start the timer if needed.
+  start_timer_if_needed(exec_ctx, elem, calld->recv_initial_metadata->deadline);
+  // Invoke the next callback.
+  calld->next_recv_initial_metadata_ready->cb(
+      exec_ctx, calld->next_recv_initial_metadata_ready->cb_arg, error);
+}
+
+// Method for starting a call op for server filter.
+static void server_start_transport_stream_op(grpc_exec_ctx* exec_ctx,
+                                             grpc_call_element* elem,
+                                             grpc_transport_stream_op* op) {
+  server_call_data* calld = elem->call_data;
+  if (op->cancel_error != GRPC_ERROR_NONE ||
+      op->close_error != GRPC_ERROR_NONE) {
+    cancel_timer_if_needed(exec_ctx, &calld->base.deadline_state);
+  } else {
+    // If we're receiving initial metadata, we need to get the deadline
+    // from the recv_initial_metadata_ready callback.  So we inject our
+    // own callback into that hook.
+    if (op->recv_initial_metadata_ready != NULL) {
+      calld->next_recv_initial_metadata_ready = op->recv_initial_metadata_ready;
+      calld->recv_initial_metadata = op->recv_initial_metadata;
+      grpc_closure_init(&calld->recv_initial_metadata_ready,
+                        recv_initial_metadata_ready, elem);
+      op->recv_initial_metadata_ready = &calld->recv_initial_metadata_ready;
+    }
+    // Make sure we know when the call is complete, so that we can cancel
+    // the timer.
+    // Note that we trigger this on recv_trailing_metadata, even though
+    // the client never sends trailing metadata, because this is the
+    // hook that tells us when the call is complete on the server side.
+    if (op->recv_trailing_metadata != NULL) {
+      inject_on_complete_cb(&calld->base.deadline_state, op);
+    }
+  }
+  // Chain to next filter.
+  grpc_call_next_op(exec_ctx, elem, op);
+}
+
+const grpc_channel_filter grpc_client_deadline_filter = {
+    client_start_transport_stream_op,
+    grpc_channel_next_op,
+    sizeof(base_call_data),
+    init_call_elem,
+    grpc_call_stack_ignore_set_pollset_or_pollset_set,
+    destroy_call_elem,
+    0,  // sizeof(channel_data)
+    init_channel_elem,
+    destroy_channel_elem,
+    grpc_call_next_get_peer,
+    "deadline",
+};
+
+const grpc_channel_filter grpc_server_deadline_filter = {
+    server_start_transport_stream_op,
+    grpc_channel_next_op,
+    sizeof(server_call_data),
+    init_call_elem,
+    grpc_call_stack_ignore_set_pollset_or_pollset_set,
+    destroy_call_elem,
+    0,  // sizeof(channel_data)
+    init_channel_elem,
+    destroy_channel_elem,
+    grpc_call_next_get_peer,
+    "deadline",
+};
diff --git a/src/core/lib/channel/deadline_filter.h b/src/core/lib/channel/deadline_filter.h
new file mode 100644
index 0000000000..685df87761
--- /dev/null
+++ b/src/core/lib/channel/deadline_filter.h
@@ -0,0 +1,79 @@
+//
+// Copyright 2016, 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_CORE_LIB_CHANNEL_DEADLINE_FILTER_H
+#define GRPC_CORE_LIB_CHANNEL_DEADLINE_FILTER_H
+
+#include "src/core/lib/channel/channel_stack.h"
+#include "src/core/lib/iomgr/timer.h"
+
+// State used for filters that enforce call deadlines.
+// Must be the first field in the filter's call_data.
+typedef struct grpc_deadline_state {
+  // We take a reference to the call stack for the timer callback.
+  grpc_call_stack* call_stack;
+  // Guards access to timer_pending and timer.
+  gpr_mu timer_mu;
+  // True if the timer callback is currently pending.
+  bool timer_pending;
+  // The deadline timer.
+  grpc_timer timer;
+  // Closure to invoke when the call is complete.
+  // We use this to cancel the timer.
+  grpc_closure on_complete;
+  // The original on_complete closure, which we chain to after our own
+  // closure is invoked.
+  grpc_closure* next_on_complete;
+} grpc_deadline_state;
+
+// To be used in a filter's init_call_elem(), destroy_call_elem(), and
+// start_transport_stream_op() methods to enforce call deadlines.
+//
+// REQUIRES: The first field in elem->call_data is a grpc_deadline_state.
+//
+// For grpc_deadline_state_client_start_transport_stream_op(), it is the
+// caller's responsibility to chain to the next filter if necessary
+// after the function returns.
+void grpc_deadline_state_init(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
+                              grpc_call_element_args* args);
+void grpc_deadline_state_destroy(grpc_exec_ctx* exec_ctx,
+                                 grpc_call_element* elem);
+void grpc_deadline_state_client_start_transport_stream_op(
+    grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
+    grpc_transport_stream_op* op);
+
+// Deadline filters for direct client channels and server channels.
+// Note: Deadlines for non-direct client channels are handled by the
+// client_channel filter.
+extern const grpc_channel_filter grpc_client_deadline_filter;
+extern const grpc_channel_filter grpc_server_deadline_filter;
+
+#endif /* GRPC_CORE_LIB_CHANNEL_DEADLINE_FILTER_H */
diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c
index edcc741ff6..1dc05fb20d 100644
--- a/src/core/lib/channel/http_client_filter.c
+++ b/src/core/lib/channel/http_client_filter.c
@@ -103,8 +103,8 @@ static grpc_mdelem *client_recv_filter(void *user_data, grpc_mdelem *md) {
                  grpc_mdstr_as_c_string(md->value));
     gpr_slice message = gpr_slice_from_copied_string(message_string);
     gpr_free(message_string);
-    grpc_call_element_send_cancel_with_message(a->exec_ctx, a->elem,
-                                               GRPC_STATUS_CANCELLED, &message);
+    grpc_call_element_send_close_with_message(a->exec_ctx, a->elem,
+                                              GRPC_STATUS_CANCELLED, &message);
     return NULL;
   } else if (md == GRPC_MDELEM_CONTENT_TYPE_APPLICATION_SLASH_GRPC) {
     return NULL;
diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c
index f8a6dd3491..02fc68fc3a 100644
--- a/src/core/lib/channel/message_size_filter.c
+++ b/src/core/lib/channel/message_size_filter.c
@@ -75,7 +75,7 @@ static void recv_message_ready(grpc_exec_ctx* exec_ctx, void* user_data,
                  (*calld->recv_message)->length, chand->max_recv_size);
     gpr_slice message = gpr_slice_from_copied_string(message_string);
     gpr_free(message_string);
-    grpc_call_element_send_cancel_with_message(
+    grpc_call_element_send_close_with_message(
         exec_ctx, elem, GRPC_STATUS_INVALID_ARGUMENT, &message);
   }
   // Invoke the next callback.
@@ -96,7 +96,7 @@ static void start_transport_stream_op(grpc_exec_ctx* exec_ctx,
                  op->send_message->length, chand->max_send_size);
     gpr_slice message = gpr_slice_from_copied_string(message_string);
     gpr_free(message_string);
-    grpc_call_element_send_cancel_with_message(
+    grpc_call_element_send_close_with_message(
         exec_ctx, elem, GRPC_STATUS_INVALID_ARGUMENT, &message);
   }
   // Inject callback for receiving a message.
diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c
index e366961936..31c80260f8 100644
--- a/src/core/lib/iomgr/error.c
+++ b/src/core/lib/iomgr/error.c
@@ -324,6 +324,64 @@ const char *grpc_error_get_str(grpc_error *err, grpc_error_strs which) {
   return gpr_avl_get(err->strs, (void *)(uintptr_t)which);
 }
 
+typedef struct {
+  grpc_error *error;
+  grpc_status_code code;
+  const char *msg;
+} special_error_status_map;
+static special_error_status_map error_status_map[] = {
+    {GRPC_ERROR_NONE, GRPC_STATUS_OK, ""},
+    {GRPC_ERROR_CANCELLED, GRPC_STATUS_CANCELLED, "RPC cancelled"},
+    {GRPC_ERROR_OOM, GRPC_STATUS_RESOURCE_EXHAUSTED, "Out of memory"},
+};
+
+static grpc_error *recursively_find_error_with_status(grpc_error *error,
+                                                      intptr_t *status) {
+  // If the error itself has a status code, return it.
+  if (grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, status)) {
+    return error;
+  }
+  // Otherwise, search through its children.
+  intptr_t key = 0;
+  while (true) {
+    grpc_error *child_error = gpr_avl_get(error->errs, (void *)key++);
+    if (child_error == NULL) break;
+    grpc_error *result =
+        recursively_find_error_with_status(child_error, status);
+    if (result != NULL) return result;
+  }
+  return NULL;
+}
+
+void grpc_error_get_status(grpc_error *error, grpc_status_code *code,
+                           const char **msg) {
+  // Handle special errors via the static map.
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(error_status_map); ++i) {
+    if (error == error_status_map[i].error) {
+      *code = error_status_map[i].code;
+      *msg = error_status_map[i].msg;
+      return;
+    }
+  }
+  // Populate code.
+  // Start with the parent error and recurse through the tree of children
+  // until we find the first one that has a status code.
+  intptr_t status = GRPC_STATUS_UNKNOWN;  // Default in case we don't find one.
+  grpc_error *found_error = recursively_find_error_with_status(error, &status);
+  *code = (grpc_status_code)status;
+  // Now populate msg.
+  // If we found an error with a status code above, use that; otherwise,
+  // fall back to using the parent error.
+  if (found_error == NULL) found_error = error;
+  // If the error has a status message, use it.  Otherwise, fall back to
+  // the error description.
+  *msg = grpc_error_get_str(found_error, GRPC_ERROR_STR_GRPC_MESSAGE);
+  if (*msg == NULL) {
+    *msg = grpc_error_get_str(found_error, GRPC_ERROR_STR_DESCRIPTION);
+    if (*msg == NULL) *msg = "uknown error";  // Just in case.
+  }
+}
+
 grpc_error *grpc_error_add_child(grpc_error *src, grpc_error *child) {
   GPR_TIMER_BEGIN("grpc_error_add_child", 0);
   grpc_error *new = copy_error_and_unref(src);
diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h
index 6c769accdb..7e2fd7a3bd 100644
--- a/src/core/lib/iomgr/error.h
+++ b/src/core/lib/iomgr/error.h
@@ -37,6 +37,7 @@
 #include <stdbool.h>
 #include <stdint.h>
 
+#include <grpc/status.h>
 #include <grpc/support/time.h>
 
 /// Opaque representation of an error.
@@ -175,6 +176,13 @@ grpc_error *grpc_error_set_str(grpc_error *src, grpc_error_strs which,
 /// Returns NULL if the specified string is not set.
 /// Caller does NOT own return value.
 const char *grpc_error_get_str(grpc_error *error, grpc_error_strs which);
+
+/// A utility function to get the status code and message to be returned
+/// to the application.  If not set in the top-level message, looks
+/// through child errors until it finds the first one with these attributes.
+void grpc_error_get_status(grpc_error *error, grpc_status_code *code,
+                           const char **msg);
+
 /// Add a child error: an error that is believed to have contributed to this
 /// error occurring. Allows root causing high level errors from lower level
 /// errors that contributed to them.
diff --git a/src/core/lib/iomgr/tcp_client_posix.c b/src/core/lib/iomgr/tcp_client_posix.c
index 80c7a3f128..3496b6094f 100644
--- a/src/core/lib/iomgr/tcp_client_posix.c
+++ b/src/core/lib/iomgr/tcp_client_posix.c
@@ -146,61 +146,57 @@ static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) {
   grpc_timer_cancel(exec_ctx, &ac->alarm);
 
   gpr_mu_lock(&ac->mu);
-  if (error == GRPC_ERROR_NONE) {
-    do {
-      so_error_size = sizeof(so_error);
-      err = getsockopt(grpc_fd_wrapped_fd(fd), SOL_SOCKET, SO_ERROR, &so_error,
-                       &so_error_size);
-    } while (err < 0 && errno == EINTR);
-    if (err < 0) {
-      error = GRPC_OS_ERROR(errno, "getsockopt");
-      goto finish;
-    } else if (so_error != 0) {
-      if (so_error == ENOBUFS) {
-        /* We will get one of these errors if we have run out of
-           memory in the kernel for the data structures allocated
-           when you connect a socket.  If this happens it is very
-           likely that if we wait a little bit then try again the
-           connection will work (since other programs or this
-           program will close their network connections and free up
-           memory).  This does _not_ indicate that there is anything
-           wrong with the server we are connecting to, this is a
-           local problem.
-
-           If you are looking at this code, then chances are that
-           your program or another program on the same computer
-           opened too many network connections.  The "easy" fix:
-           don't do that! */
-        gpr_log(GPR_ERROR, "kernel out of buffers");
-        gpr_mu_unlock(&ac->mu);
-        grpc_fd_notify_on_write(exec_ctx, fd, &ac->write_closure);
-        return;
-      } else {
-        switch (so_error) {
-          case ECONNREFUSED:
-            error = grpc_error_set_int(error, GRPC_ERROR_INT_ERRNO, errno);
-            error = grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR,
-                                       "Connection refused");
-            break;
-          default:
-            error = GRPC_OS_ERROR(errno, "getsockopt(SO_ERROR)");
-            break;
-        }
-        goto finish;
-      }
-    } else {
-      grpc_pollset_set_del_fd(exec_ctx, ac->interested_parties, fd);
-      *ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str);
-      fd = NULL;
-      goto finish;
-    }
-  } else {
+  if (error != GRPC_ERROR_NONE) {
     error =
         grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, "Timeout occurred");
     goto finish;
   }
 
-  GPR_UNREACHABLE_CODE(return );
+  do {
+    so_error_size = sizeof(so_error);
+    err = getsockopt(grpc_fd_wrapped_fd(fd), SOL_SOCKET, SO_ERROR, &so_error,
+                     &so_error_size);
+  } while (err < 0 && errno == EINTR);
+  if (err < 0) {
+    error = GRPC_OS_ERROR(errno, "getsockopt");
+    goto finish;
+  }
+
+  switch (so_error) {
+    case 0:
+      grpc_pollset_set_del_fd(exec_ctx, ac->interested_parties, fd);
+      *ep = grpc_tcp_create(fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, ac->addr_str);
+      fd = NULL;
+      break;
+    case ENOBUFS:
+      /* We will get one of these errors if we have run out of
+         memory in the kernel for the data structures allocated
+         when you connect a socket.  If this happens it is very
+         likely that if we wait a little bit then try again the
+         connection will work (since other programs or this
+         program will close their network connections and free up
+         memory).  This does _not_ indicate that there is anything
+         wrong with the server we are connecting to, this is a
+         local problem.
+
+         If you are looking at this code, then chances are that
+         your program or another program on the same computer
+         opened too many network connections.  The "easy" fix:
+         don't do that! */
+      gpr_log(GPR_ERROR, "kernel out of buffers");
+      gpr_mu_unlock(&ac->mu);
+      grpc_fd_notify_on_write(exec_ctx, fd, &ac->write_closure);
+      return;
+    case ECONNREFUSED:
+      /* This error shouldn't happen for anything other than connect(). */
+      error = GRPC_OS_ERROR(so_error, "connect");
+      break;
+    default:
+      /* We don't really know which syscall triggered the problem here,
+         so punt by reporting getsockopt(). */
+      error = GRPC_OS_ERROR(so_error, "getsockopt(SO_ERROR)");
+      break;
+  }
 
 finish:
   if (fd != NULL) {
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index 979e0aeaea..5690bcab1e 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -126,8 +126,6 @@ struct grpc_call {
 
   /* client or server call */
   bool is_client;
-  /* is the alarm set */
-  bool have_alarm;
   /** has grpc_call_destroy been called */
   bool destroy_called;
   /** flag indicating that cancellation is inherited */
@@ -170,9 +168,6 @@ struct grpc_call {
   /* Contexts for various subsystems (security, tracing, ...). */
   grpc_call_context_element context[GRPC_CONTEXT_COUNT];
 
-  /* Deadline alarm - if have_alarm is non-zero */
-  grpc_timer alarm;
-
   /* for the client, extra metadata is initial metadata; for the
      server, it's trailing metadata */
   grpc_linked_mdelem send_extra_metadata[MAX_SEND_EXTRA_METADATA_COUNT];
@@ -215,8 +210,6 @@ struct grpc_call {
 #define CALL_FROM_TOP_ELEM(top_elem) \
   CALL_FROM_CALL_STACK(grpc_call_stack_from_top_element(top_elem))
 
-static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call,
-                               gpr_timespec deadline);
 static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
                        grpc_transport_stream_op *op);
 static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
@@ -264,39 +257,8 @@ grpc_call *grpc_call_create(
       call->metadata_batch[i][j].deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
     }
   }
-  call->send_deadline =
-      gpr_convert_clock_type(send_deadline, GPR_CLOCK_MONOTONIC);
-  GRPC_CHANNEL_INTERNAL_REF(channel, "call");
-  /* initial refcount dropped by grpc_call_destroy */
-  grpc_error *error = grpc_call_stack_init(
-      &exec_ctx, channel_stack, 1, destroy_call, call, call->context,
-      server_transport_data, CALL_STACK_FROM_CALL(call));
-  if (error != GRPC_ERROR_NONE) {
-    intptr_t status;
-    if (!grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status))
-      status = GRPC_STATUS_UNKNOWN;
-    const char *error_str =
-        grpc_error_get_str(error, GRPC_ERROR_STR_DESCRIPTION);
-    close_with_status(&exec_ctx, call, (grpc_status_code)status,
-                      error_str == NULL ? "unknown error" : error_str);
-    GRPC_ERROR_UNREF(error);
-  }
-  if (cq != NULL) {
-    GPR_ASSERT(
-        pollset_set_alternative == NULL &&
-        "Only one of 'cq' and 'pollset_set_alternative' should be non-NULL.");
-    GRPC_CQ_INTERNAL_REF(cq, "bind");
-    call->pollent =
-        grpc_polling_entity_create_from_pollset(grpc_cq_pollset(cq));
-  }
-  if (pollset_set_alternative != NULL) {
-    call->pollent =
-        grpc_polling_entity_create_from_pollset_set(pollset_set_alternative);
-  }
-  if (!grpc_polling_entity_is_empty(&call->pollent)) {
-    grpc_call_stack_set_pollset_or_pollset_set(
-        &exec_ctx, CALL_STACK_FROM_CALL(call), &call->pollent);
-  }
+  send_deadline = gpr_convert_clock_type(send_deadline, GPR_CLOCK_MONOTONIC);
+
   if (parent_call != NULL) {
     GRPC_CALL_INTERNAL_REF(parent_call, "child");
     GPR_ASSERT(call->is_client);
@@ -338,10 +300,38 @@ grpc_call *grpc_call_create(
 
     gpr_mu_unlock(&parent_call->mu);
   }
-  if (gpr_time_cmp(send_deadline, gpr_inf_future(send_deadline.clock_type)) !=
-      0) {
-    set_deadline_alarm(&exec_ctx, call, send_deadline);
+
+  call->send_deadline = send_deadline;
+
+  GRPC_CHANNEL_INTERNAL_REF(channel, "call");
+  /* initial refcount dropped by grpc_call_destroy */
+  grpc_error *error = grpc_call_stack_init(
+      &exec_ctx, channel_stack, 1, destroy_call, call, call->context,
+      server_transport_data, send_deadline, CALL_STACK_FROM_CALL(call));
+  if (error != GRPC_ERROR_NONE) {
+    grpc_status_code status;
+    const char *error_str;
+    grpc_error_get_status(error, &status, &error_str);
+    close_with_status(&exec_ctx, call, status, error_str);
+    GRPC_ERROR_UNREF(error);
   }
+  if (cq != NULL) {
+    GPR_ASSERT(
+        pollset_set_alternative == NULL &&
+        "Only one of 'cq' and 'pollset_set_alternative' should be non-NULL.");
+    GRPC_CQ_INTERNAL_REF(cq, "bind");
+    call->pollent =
+        grpc_polling_entity_create_from_pollset(grpc_cq_pollset(cq));
+  }
+  if (pollset_set_alternative != NULL) {
+    call->pollent =
+        grpc_polling_entity_create_from_pollset_set(pollset_set_alternative);
+  }
+  if (!grpc_polling_entity_is_empty(&call->pollent)) {
+    grpc_call_stack_set_pollset_or_pollset_set(
+        &exec_ctx, CALL_STACK_FROM_CALL(call), &call->pollent);
+  }
+
   grpc_exec_ctx_finish(&exec_ctx);
   GPR_TIMER_END("grpc_call_create", 0);
   return call;
@@ -454,20 +444,11 @@ static void set_status_details(grpc_call *call, status_source source,
 
 static void set_status_from_error(grpc_call *call, status_source source,
                                   grpc_error *error) {
-  intptr_t status;
-  if (grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status)) {
-    set_status_code(call, source, (uint32_t)status);
-  } else {
-    set_status_code(call, source, GRPC_STATUS_INTERNAL);
-  }
-  const char *msg = grpc_error_get_str(error, GRPC_ERROR_STR_GRPC_MESSAGE);
-  bool free_msg = false;
-  if (msg == NULL) {
-    free_msg = true;
-    msg = grpc_error_string(error);
-  }
+  grpc_status_code status;
+  const char *msg;
+  grpc_error_get_status(error, &status, &msg);
+  set_status_code(call, source, (uint32_t)status);
   set_status_details(call, source, grpc_mdstr_from_string(msg));
-  if (free_msg) grpc_error_free_string(msg);
 }
 
 static void set_incoming_compression_algorithm(
@@ -740,9 +721,6 @@ void grpc_call_destroy(grpc_call *c) {
   gpr_mu_lock(&c->mu);
   GPR_ASSERT(!c->destroy_called);
   c->destroy_called = 1;
-  if (c->have_alarm) {
-    grpc_timer_cancel(&exec_ctx, &c->alarm);
-  }
   cancel = !c->received_final_op;
   gpr_mu_unlock(&c->mu);
   if (cancel) grpc_call_cancel(c, NULL);
@@ -780,7 +758,6 @@ typedef struct termination_closure {
   grpc_closure closure;
   grpc_call *call;
   grpc_error *error;
-  grpc_closure *op_closure;
   enum { TC_CANCEL, TC_CLOSE } type;
   grpc_transport_stream_op op;
 } termination_closure;
@@ -797,7 +774,6 @@ static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp,
       break;
   }
   GRPC_ERROR_UNREF(tc->error);
-  grpc_exec_ctx_sched(exec_ctx, tc->op_closure, GRPC_ERROR_NONE, NULL);
   gpr_free(tc);
 }
 
@@ -817,7 +793,6 @@ static void send_close(grpc_exec_ctx *exec_ctx, void *tcp, grpc_error *error) {
   tc->op.close_error = tc->error;
   /* reuse closure to catch completion */
   grpc_closure_init(&tc->closure, done_termination, tc);
-  tc->op_closure = tc->op.on_complete;
   tc->op.on_complete = &tc->closure;
   execute_op(exec_ctx, tc->call, &tc->op);
 }
@@ -900,32 +875,6 @@ grpc_call *grpc_call_from_top_element(grpc_call_element *elem) {
   return CALL_FROM_TOP_ELEM(elem);
 }
 
-static void call_alarm(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
-  grpc_call *call = arg;
-  gpr_mu_lock(&call->mu);
-  call->have_alarm = 0;
-  if (error != GRPC_ERROR_CANCELLED) {
-    cancel_with_status(exec_ctx, call, GRPC_STATUS_DEADLINE_EXCEEDED,
-                       "Deadline Exceeded");
-  }
-  gpr_mu_unlock(&call->mu);
-  GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "alarm");
-}
-
-static void set_deadline_alarm(grpc_exec_ctx *exec_ctx, grpc_call *call,
-                               gpr_timespec deadline) {
-  if (call->have_alarm) {
-    gpr_log(GPR_ERROR, "Attempt to set deadline alarm twice");
-    assert(0);
-    return;
-  }
-  GRPC_CALL_INTERNAL_REF(call, "alarm");
-  call->have_alarm = 1;
-  call->send_deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC);
-  grpc_timer_init(exec_ctx, &call->alarm, call->send_deadline, call_alarm, call,
-                  gpr_now(GPR_CLOCK_MONOTONIC));
-}
-
 /* we offset status by a small amount when storing it into transport metadata
    as metadata cannot store a 0 value (which is used as OK for grpc_status_codes
    */
@@ -1267,9 +1216,8 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
     if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) !=
             0 &&
         !call->is_client) {
-      GPR_TIMER_BEGIN("set_deadline_alarm", 0);
-      set_deadline_alarm(exec_ctx, call, md->deadline);
-      GPR_TIMER_END("set_deadline_alarm", 0);
+      call->send_deadline =
+          gpr_convert_clock_type(md->deadline, GPR_CLOCK_MONOTONIC);
     }
   }
 
@@ -1298,9 +1246,17 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp,
   GRPC_ERROR_REF(error);
 
   gpr_mu_lock(&call->mu);
+
+  // If the error has an associated status code, set the call's status.
+  intptr_t status;
+  if (error != GRPC_ERROR_NONE &&
+      grpc_error_get_int(error, GRPC_ERROR_INT_GRPC_STATUS, &status)) {
+    set_status_from_error(call, STATUS_FROM_CORE, error);
+  }
+
   if (bctl->send_initial_metadata) {
     if (error != GRPC_ERROR_NONE) {
-      set_status_code(call, STATUS_FROM_CORE, GRPC_STATUS_UNAVAILABLE);
+      set_status_from_error(call, STATUS_FROM_CORE, error);
     }
     grpc_metadata_batch_destroy(
         &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]);
@@ -1318,9 +1274,6 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp,
     grpc_metadata_batch_filter(md, recv_trailing_filter, call);
 
     call->received_final_op = true;
-    if (call->have_alarm) {
-      grpc_timer_cancel(exec_ctx, &call->alarm);
-    }
     /* propagate cancellation to any interested children */
     child_call = call->first_child;
     if (child_call != NULL) {
diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h
index 3049284f68..4dbf3aae63 100644
--- a/src/core/lib/surface/completion_queue.h
+++ b/src/core/lib/surface/completion_queue.h
@@ -57,6 +57,8 @@ typedef struct grpc_cq_completion {
   uintptr_t next;
 } grpc_cq_completion;
 
+//#define GRPC_CQ_REF_COUNT_DEBUG
+
 #ifdef GRPC_CQ_REF_COUNT_DEBUG
 void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason,
                           const char *file, int line);
diff --git a/src/core/lib/surface/init.c b/src/core/lib/surface/init.c
index 64bdfc3446..289f4ce8e8 100644
--- a/src/core/lib/surface/init.c
+++ b/src/core/lib/surface/init.c
@@ -43,6 +43,7 @@
 #include "src/core/lib/channel/channel_stack.h"
 #include "src/core/lib/channel/compress_filter.h"
 #include "src/core/lib/channel/connected_channel.h"
+#include "src/core/lib/channel/deadline_filter.h"
 #include "src/core/lib/channel/http_client_filter.h"
 #include "src/core/lib/channel/http_server_filter.h"
 #include "src/core/lib/channel/message_size_filter.h"
@@ -100,6 +101,12 @@ static bool maybe_add_http_filter(grpc_channel_stack_builder *builder,
 
 static void register_builtin_channel_init() {
   grpc_channel_init_register_stage(
+      GRPC_CLIENT_DIRECT_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
+      prepend_filter, (void *)&grpc_client_deadline_filter);
+  grpc_channel_init_register_stage(
+      GRPC_SERVER_CHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY, prepend_filter,
+      (void *)&grpc_server_deadline_filter);
+  grpc_channel_init_register_stage(
       GRPC_CLIENT_SUBCHANNEL, GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
       prepend_filter, (void *)&grpc_message_size_filter);
   grpc_channel_init_register_stage(
diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c
index 08f9d7e8d9..82fc605218 100644
--- a/src/core/lib/transport/transport.c
+++ b/src/core/lib/transport/transport.c
@@ -47,7 +47,7 @@
 void grpc_stream_ref(grpc_stream_refcount *refcount, const char *reason) {
   gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count);
   gpr_log(GPR_DEBUG, "%s %p:%p   REF %d->%d %s", refcount->object_type,
-          refcount, refcount->destroy.cb_arg, val, val + 1, reason);
+          refcount, refcount->destroy.cb_arg, (int)val, (int)val + 1, reason);
 #else
 void grpc_stream_ref(grpc_stream_refcount *refcount) {
 #endif
@@ -59,7 +59,7 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount,
                        const char *reason) {
   gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count);
   gpr_log(GPR_DEBUG, "%s %p:%p UNREF %d->%d %s", refcount->object_type,
-          refcount, refcount->destroy.cb_arg, val, val - 1, reason);
+          refcount, refcount->destroy.cb_arg, (int)val, (int)val - 1, reason);
 #else
 void grpc_stream_unref(grpc_exec_ctx *exec_ctx,
                        grpc_stream_refcount *refcount) {
@@ -224,7 +224,7 @@ void grpc_transport_stream_op_add_cancellation_with_message(
     error = GRPC_ERROR_CREATE("Call cancelled");
   }
   error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, status);
-  add_error(op, &op->close_error, error);
+  add_error(op, &op->cancel_error, error);
 }
 
 void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op,