Merge branch 'master' of github.com:grpc/grpc into the-ultimate-showdown

Conflicts:
	include/grpc/grpc.h
	src/core/surface/channel.c
	src/core/surface/channel_create.c
	src/core/surface/completion_queue.c
	src/cpp/client/channel.cc
	src/cpp/client/insecure_credentials.cc
	src/csharp/ext/grpc_csharp_ext.c
	src/node/ext/call.cc
	src/node/ext/channel.cc
	src/php/ext/grpc/call.c
	src/php/ext/grpc/channel.c
	src/python/grpcio/grpc/_adapter/_c/types/channel.c
	src/ruby/ext/grpc/rb_channel.c
	test/core/end2end/dualstack_socket_test.c
	test/core/end2end/fixtures/chttp2_fullstack.c
	test/core/end2end/fixtures/chttp2_fullstack_compression.c
	test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
	test/core/end2end/fixtures/chttp2_fullstack_with_poll.c
	test/core/end2end/multiple_server_queues_test.c
	test/core/end2end/no_server_test.c
	test/core/end2end/tests/bad_hostname.c
	test/core/end2end/tests/cancel_after_accept.c
	test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
	test/core/end2end/tests/cancel_after_invoke.c
	test/core/end2end/tests/cancel_before_invoke.c
	test/core/end2end/tests/cancel_in_a_vacuum.c
	test/core/end2end/tests/census_simple_request.c
	test/core/end2end/tests/disappearing_server.c
	test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
	test/core/end2end/tests/empty_batch.c
	test/core/end2end/tests/graceful_server_shutdown.c
	test/core/end2end/tests/invoke_large_request.c
	test/core/end2end/tests/max_concurrent_streams.c
	test/core/end2end/tests/max_message_length.c
	test/core/end2end/tests/ping_pong_streaming.c
	test/core/end2end/tests/registered_call.c
	test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
	test/core/end2end/tests/request_response_with_metadata_and_payload.c
	test/core/end2end/tests/request_response_with_payload.c
	test/core/end2end/tests/request_response_with_payload_and_call_creds.c
	test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
	test/core/end2end/tests/request_with_compressed_payload.c
	test/core/end2end/tests/request_with_flags.c
	test/core/end2end/tests/request_with_large_metadata.c
	test/core/end2end/tests/request_with_payload.c
	test/core/end2end/tests/server_finishes_request.c
	test/core/end2end/tests/simple_delayed_request.c
	test/core/end2end/tests/simple_request.c
	test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
	test/core/fling/client.c
	test/core/fling/server.c
	test/core/surface/lame_client_test.c

17 files changed, 462 insertions, 146 deletions

diff --git a/src/core/transport/chttp2/alpn.c b/src/core/transport/chttp2/alpn.c
index 3ccd5796ba..69da4e6718 100644
--- a/src/core/transport/chttp2/alpn.c
+++ b/src/core/transport/chttp2/alpn.c
@@ -36,8 +36,7 @@
 #include <grpc/support/useful.h>
 
 /* in order of preference */
-static const char *const supported_versions[] = {"h2", "h2-17", "h2-16",
-                                                 "h2-15", "h2-14"};
+static const char *const supported_versions[] = {"h2"};
 
 int grpc_chttp2_is_alpn_version_supported(const char *version, size_t size) {
   size_t i;
diff --git a/src/core/transport/chttp2/frame_data.c b/src/core/transport/chttp2/frame_data.c
index 7a4c355f23..40bf2ebd79 100644
--- a/src/core/transport/chttp2/frame_data.c
+++ b/src/core/transport/chttp2/frame_data.c
@@ -92,7 +92,7 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse(
       p->frame_type = *cur;
       switch (p->frame_type) {
         case 0:
-          /* noop */
+          p->is_frame_compressed = 0;  /* GPR_FALSE */
           break;
         case 1:
           p->is_frame_compressed = 1;  /* GPR_TRUE */
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index e5e6f445b7..42cf0ecd5b 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -60,7 +60,6 @@ typedef enum {
   GRPC_CHTTP2_LIST_WRITABLE,
   GRPC_CHTTP2_LIST_WRITING,
   GRPC_CHTTP2_LIST_WRITTEN,
-  GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE,
   GRPC_CHTTP2_LIST_PARSING_SEEN,
   GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING,
   GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING,
@@ -120,6 +119,10 @@ typedef enum {
   GRPC_WRITE_STATE_SENT_CLOSE
 } grpc_chttp2_write_state;
 
+/* flags that can be or'd into stream_global::writing_now */
+#define GRPC_CHTTP2_WRITING_DATA 1
+#define GRPC_CHTTP2_WRITING_WINDOW 2
+
 typedef enum {
   GRPC_DONT_SEND_CLOSED = 0,
   GRPC_SEND_CLOSED,
@@ -286,6 +289,7 @@ struct grpc_chttp2_transport {
   grpc_endpoint *ep;
   grpc_mdctx *metadata_context;
   gpr_refcount refs;
+  char *peer_string;
 
   gpr_mu mu;
 
@@ -382,6 +386,10 @@ typedef struct {
   gpr_uint8 published_cancelled;
   /** is this stream in the stream map? (boolean) */
   gpr_uint8 in_stream_map;
+  /** bitmask of GRPC_CHTTP2_WRITING_xxx above */
+  gpr_uint8 writing_now;
+  /** has anything been written to this stream? */
+  gpr_uint8 written_anything;
 
   /** stream state already published to the upper layer */
   grpc_stream_state published_state;
@@ -474,11 +482,17 @@ void grpc_chttp2_publish_reads(grpc_chttp2_transport_global *global,
 void grpc_chttp2_list_add_writable_stream(
     grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_stream_global *stream_global);
+void grpc_chttp2_list_add_first_writable_stream(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global);
 int grpc_chttp2_list_pop_writable_stream(
     grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_transport_writing *transport_writing,
     grpc_chttp2_stream_global **stream_global,
     grpc_chttp2_stream_writing **stream_writing);
+void grpc_chttp2_list_remove_writable_stream(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global);
 
 void grpc_chttp2_list_add_incoming_window_updated(
     grpc_chttp2_transport_global *transport_global,
@@ -510,18 +524,6 @@ int grpc_chttp2_list_pop_written_stream(
     grpc_chttp2_stream_global **stream_global,
     grpc_chttp2_stream_writing **stream_writing);
 
-void grpc_chttp2_list_add_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global);
-int grpc_chttp2_list_pop_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_transport_writing *transport_writing,
-    grpc_chttp2_stream_global **stream_global,
-    grpc_chttp2_stream_writing **stream_writing);
-void grpc_chttp2_list_remove_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global);
-
 void grpc_chttp2_list_add_parsing_seen_stream(
     grpc_chttp2_transport_parsing *transport_parsing,
     grpc_chttp2_stream_parsing *stream_parsing);
diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c
index 904b9afce7..d84960009b 100644
--- a/src/core/transport/chttp2/parsing.c
+++ b/src/core/transport/chttp2/parsing.c
@@ -182,8 +182,7 @@ void grpc_chttp2_publish_reads(
       stream_global->max_recv_bytes -= 
           stream_parsing->incoming_window_delta;
       stream_parsing->incoming_window_delta = 0;
-      grpc_chttp2_list_add_writable_window_update_stream(transport_global,
-                                                         stream_global);
+      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
     }
 
     /* update outgoing flow control window */
@@ -607,7 +606,7 @@ static void on_header(void *tp, grpc_mdelem *md) {
     }
     grpc_chttp2_incoming_metadata_buffer_set_deadline(
         &stream_parsing->incoming_metadata,
-        gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), *cached_timeout));
+        gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), *cached_timeout));
     GRPC_MDELEM_UNREF(md);
   } else {
     grpc_chttp2_incoming_metadata_buffer_add(&stream_parsing->incoming_metadata,
diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c
index 65b31a5afd..0f04169741 100644
--- a/src/core/transport/chttp2/stream_encoder.c
+++ b/src/core/transport/chttp2/stream_encoder.c
@@ -441,7 +441,7 @@ static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline,
       gpr_time_sub(deadline, gpr_now(deadline.clock_type)), timeout_str);
   mdelem = grpc_mdelem_from_metadata_strings(
       c->mdctx, GRPC_MDSTR_REF(c->timeout_key_str),
-      grpc_mdstr_from_string(c->mdctx, timeout_str));
+      grpc_mdstr_from_string(c->mdctx, timeout_str, 0));
   mdelem = hpack_enc(c, mdelem, st);
   if (mdelem) GRPC_MDELEM_UNREF(mdelem);
 }
@@ -456,7 +456,7 @@ void grpc_chttp2_hpack_compressor_init(grpc_chttp2_hpack_compressor *c,
                                        grpc_mdctx *ctx) {
   memset(c, 0, sizeof(*c));
   c->mdctx = ctx;
-  c->timeout_key_str = grpc_mdstr_from_string(ctx, "grpc-timeout");
+  c->timeout_key_str = grpc_mdstr_from_string(ctx, "grpc-timeout", 0);
 }
 
 void grpc_chttp2_hpack_compressor_destroy(grpc_chttp2_hpack_compressor *c) {
diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c
index 590f6abfbc..9c3ad7a777 100644
--- a/src/core/transport/chttp2/stream_lists.c
+++ b/src/core/transport/chttp2/stream_lists.c
@@ -108,6 +108,23 @@ static void stream_list_maybe_remove(grpc_chttp2_transport *t,
   }
 }
 
+static void stream_list_add_head(grpc_chttp2_transport *t,
+                                 grpc_chttp2_stream *s,
+                                 grpc_chttp2_stream_list_id id) {
+  grpc_chttp2_stream *old_head;
+  GPR_ASSERT(!s->included[id]);
+  old_head = t->lists[id].head;
+  s->links[id].next = old_head;
+  s->links[id].prev = NULL;
+  if (old_head) {
+    old_head->links[id].prev = s;
+  } else {
+    t->lists[id].tail = s;
+  }
+  t->lists[id].head = s;
+  s->included[id] = 1;
+}
+
 static void stream_list_add_tail(grpc_chttp2_transport *t,
                                  grpc_chttp2_stream *s,
                                  grpc_chttp2_stream_list_id id) {
@@ -119,7 +136,6 @@ static void stream_list_add_tail(grpc_chttp2_transport *t,
   if (old_tail) {
     old_tail->links[id].next = s;
   } else {
-    s->links[id].prev = NULL;
     t->lists[id].head = s;
   }
   t->lists[id].tail = s;
@@ -144,6 +160,15 @@ void grpc_chttp2_list_add_writable_stream(
                   STREAM_FROM_GLOBAL(stream_global), GRPC_CHTTP2_LIST_WRITABLE);
 }
 
+void grpc_chttp2_list_add_first_writable_stream(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global) {
+  GPR_ASSERT(stream_global->id != 0);
+  stream_list_add_head(TRANSPORT_FROM_GLOBAL(transport_global),
+                       STREAM_FROM_GLOBAL(stream_global),
+                       GRPC_CHTTP2_LIST_WRITABLE);
+}
+
 int grpc_chttp2_list_pop_writable_stream(
     grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_transport_writing *transport_writing,
@@ -157,6 +182,14 @@ int grpc_chttp2_list_pop_writable_stream(
   return r;
 }
 
+void grpc_chttp2_list_remove_writable_stream(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global) {
+  stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global),
+                           STREAM_FROM_GLOBAL(stream_global),
+                           GRPC_CHTTP2_LIST_WRITABLE);
+}
+
 void grpc_chttp2_list_add_writing_stream(
     grpc_chttp2_transport_writing *transport_writing,
     grpc_chttp2_stream_writing *stream_writing) {
@@ -202,36 +235,6 @@ int grpc_chttp2_list_pop_written_stream(
   return r;
 }
 
-void grpc_chttp2_list_add_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global) {
-  GPR_ASSERT(stream_global->id != 0);
-  stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global),
-                  STREAM_FROM_GLOBAL(stream_global),
-                  GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
-}
-
-int grpc_chttp2_list_pop_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_transport_writing *transport_writing,
-    grpc_chttp2_stream_global **stream_global,
-    grpc_chttp2_stream_writing **stream_writing) {
-  grpc_chttp2_stream *stream;
-  int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream,
-                          GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
-  *stream_global = &stream->global;
-  *stream_writing = &stream->writing;
-  return r;
-}
-
-void grpc_chttp2_list_remove_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global) {
-  stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global),
-                           STREAM_FROM_GLOBAL(stream_global),
-                           GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
-}
-
 void grpc_chttp2_list_add_parsing_seen_stream(
     grpc_chttp2_transport_parsing *transport_parsing,
     grpc_chttp2_stream_parsing *stream_parsing) {
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index d8ec117aa5..b55e81fdca 100644
--- a/src/core/transport/chttp2/writing.c
+++ b/src/core/transport/chttp2/writing.c
@@ -44,6 +44,7 @@ int grpc_chttp2_unlocking_check_writes(
     grpc_chttp2_transport_writing *transport_writing) {
   grpc_chttp2_stream_global *stream_global;
   grpc_chttp2_stream_writing *stream_writing;
+  grpc_chttp2_stream_global *first_reinserted_stream = NULL;
   gpr_uint32 window_delta;
 
   /* simple writes are queued to qbuf, and flushed here */
@@ -64,50 +65,53 @@ int grpc_chttp2_unlocking_check_writes(
   }
 
   /* for each grpc_chttp2_stream that's become writable, frame it's data
-     (according to
-     available window sizes) and add to the output buffer */
-  while (grpc_chttp2_list_pop_writable_stream(transport_global,
-                                              transport_writing, &stream_global,
-                                              &stream_writing)) {
+     (according to available window sizes) and add to the output buffer */
+  while (grpc_chttp2_list_pop_writable_stream(
+      transport_global, transport_writing, &stream_global, &stream_writing)) {
+    if (stream_global == first_reinserted_stream) {
+      /* prevent infinite loop */
+      grpc_chttp2_list_add_first_writable_stream(transport_global,
+                                                 stream_global);
+      break;
+    }
+
     stream_writing->id = stream_global->id;
-    window_delta = grpc_chttp2_preencode(
-        stream_global->outgoing_sopb->ops, &stream_global->outgoing_sopb->nops,
-        GPR_MIN(transport_global->outgoing_window,
-                stream_global->outgoing_window),
-        &stream_writing->sopb);
-    GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT(
-        "write", transport_global, outgoing_window, -(gpr_int64)window_delta);
-    GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
-                                     outgoing_window, -(gpr_int64)window_delta);
-    transport_global->outgoing_window -= window_delta;
-    stream_global->outgoing_window -= window_delta;
-
-    if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE &&
-        stream_global->outgoing_sopb->nops == 0) {
-      if (!transport_global->is_client && !stream_global->read_closed) {
-        stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM;
-      } else {
-        stream_writing->send_closed = GRPC_SEND_CLOSED;
+    stream_writing->send_closed = GRPC_DONT_SEND_CLOSED;
+
+    if (stream_global->outgoing_sopb) {
+      window_delta =
+          grpc_chttp2_preencode(stream_global->outgoing_sopb->ops,
+                                &stream_global->outgoing_sopb->nops,
+                                GPR_MIN(transport_global->outgoing_window,
+                                        stream_global->outgoing_window),
+                                &stream_writing->sopb);
+      GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT(
+          "write", transport_global, outgoing_window, -(gpr_int64)window_delta);
+      GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
+                                       outgoing_window,
+                                       -(gpr_int64)window_delta);
+      transport_global->outgoing_window -= window_delta;
+      stream_global->outgoing_window -= window_delta;
+
+      if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE &&
+          stream_global->outgoing_sopb->nops == 0) {
+        if (!transport_global->is_client && !stream_global->read_closed) {
+          stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM;
+        } else {
+          stream_writing->send_closed = GRPC_SEND_CLOSED;
+        }
       }
-    }
-    if (stream_writing->sopb.nops > 0 ||
-        stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
-      grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
-    }
 
-    if (stream_global->outgoing_window > 0 &&
-        stream_global->outgoing_sopb->nops != 0) {
-      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+      if (stream_global->outgoing_window > 0 &&
+          stream_global->outgoing_sopb->nops != 0) {
+        grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+        if (first_reinserted_stream == NULL &&
+            transport_global->outgoing_window == 0) {
+          first_reinserted_stream = stream_global;
+        }
+      }
     }
-  }
 
-  /* for each grpc_chttp2_stream that wants to update its window, add that
-   * window here */
-  while (grpc_chttp2_list_pop_writable_window_update_stream(transport_global,
-                                                            transport_writing,
-                                                            &stream_global,
-                                                            &stream_writing)) {
-    stream_writing->id = stream_global->id;
     if (!stream_global->read_closed && stream_global->unannounced_incoming_window > 0) {
       stream_writing->announce_window = stream_global->unannounced_incoming_window;
       GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
@@ -118,6 +122,13 @@ int grpc_chttp2_unlocking_check_writes(
       stream_global->unannounced_incoming_window = 0;
       grpc_chttp2_list_add_incoming_window_updated(transport_global,
                                                    stream_global);
+      stream_global->writing_now |= GRPC_CHTTP2_WRITING_WINDOW;
+    }
+    if (stream_writing->sopb.nops > 0 ||
+        stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
+      stream_global->writing_now |= GRPC_CHTTP2_WRITING_DATA;
+    }
+    if (stream_global->writing_now != 0) {
       grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
     }
   }
@@ -173,6 +184,7 @@ static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing) {
                          stream_writing->send_closed != GRPC_DONT_SEND_CLOSED,
                          stream_writing->id, &transport_writing->hpack_compressor,
                          &transport_writing->outbuf);
+      stream_writing->sopb.nops = 0;
     }
     if (stream_writing->announce_window > 0) {
       gpr_slice_buffer_add(
@@ -181,7 +193,6 @@ static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing) {
               stream_writing->id, stream_writing->announce_window));
       stream_writing->announce_window = 0;
     }
-    stream_writing->sopb.nops = 0;
     if (stream_writing->send_closed == GRPC_SEND_CLOSED_WITH_RST_STREAM) {
       gpr_slice_buffer_add(&transport_writing->outbuf,
                            grpc_chttp2_rst_stream_create(stream_writing->id,
@@ -205,20 +216,25 @@ void grpc_chttp2_cleanup_writing(
 
   while (grpc_chttp2_list_pop_written_stream(
       transport_global, transport_writing, &stream_global, &stream_writing)) {
-    if (stream_global->outgoing_sopb != NULL &&
-        stream_global->outgoing_sopb->nops == 0) {
-      stream_global->outgoing_sopb = NULL;
-      grpc_chttp2_schedule_closure(transport_global,
-                                   stream_global->send_done_closure, 1);
-    }
+    GPR_ASSERT(stream_global->writing_now != 0);
     if (stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
       stream_global->write_state = GRPC_WRITE_STATE_SENT_CLOSE;
       if (!transport_global->is_client) {
         stream_global->read_closed = 1;
       }
-      grpc_chttp2_list_add_read_write_state_changed(transport_global,
-                                                    stream_global);
     }
+    if (stream_global->writing_now & GRPC_CHTTP2_WRITING_DATA) {
+      if (stream_global->outgoing_sopb != NULL &&
+          stream_global->outgoing_sopb->nops == 0) {
+        GPR_ASSERT(stream_global->write_state != GRPC_WRITE_STATE_QUEUED_CLOSE);
+        stream_global->outgoing_sopb = NULL;
+        grpc_chttp2_schedule_closure(transport_global,
+                                     stream_global->send_done_closure, 1);
+      }
+    }
+    stream_global->writing_now = 0;
+    grpc_chttp2_list_add_read_write_state_changed(transport_global,
+                                                  stream_global);
   }
   transport_writing->outbuf.count = 0;
   transport_writing->outbuf.length = 0;
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index c923d5e42f..a9f91b64d5 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -107,9 +107,16 @@ static void cancel_from_api(grpc_chttp2_transport_global *transport_global,
                             grpc_chttp2_stream_global *stream_global,
                             grpc_status_code status);
 
+static void close_from_api(grpc_chttp2_transport_global *transport_global,
+                           grpc_chttp2_stream_global *stream_global,
+                           grpc_status_code status,
+                           gpr_slice *optional_message);
+
 /** 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(
@@ -117,7 +124,7 @@ static void maybe_start_some_streams(
 
 static void connectivity_state_set(
     grpc_chttp2_transport_global *transport_global,
-    grpc_connectivity_state state);
+    grpc_connectivity_state state, const char *reason);
 
 /*
  * CONSTRUCTION/DESTRUCTION/REFCOUNTING
@@ -168,6 +175,7 @@ static void destruct_transport(grpc_chttp2_transport *t) {
 
   grpc_mdctx_unref(t->metadata_context);
 
+  gpr_free(t->peer_string);
   gpr_free(t);
 }
 
@@ -217,6 +225,7 @@ static void init_transport(grpc_chttp2_transport *t,
   gpr_ref_init(&t->refs, 2);
   gpr_mu_init(&t->mu);
   grpc_mdctx_ref(mdctx);
+  t->peer_string = grpc_endpoint_get_peer(ep);
   t->metadata_context = mdctx;
   t->endpoint_reading = 1;
   t->global.next_stream_id = is_client ? 1 : 2;
@@ -228,12 +237,12 @@ static void init_transport(grpc_chttp2_transport *t,
   t->global.pings.next = t->global.pings.prev = &t->global.pings;
   t->parsing.is_client = is_client;
   t->parsing.str_grpc_timeout =
-      grpc_mdstr_from_string(t->metadata_context, "grpc-timeout");
+      grpc_mdstr_from_string(t->metadata_context, "grpc-timeout", 0);
   t->parsing.deframe_state =
       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);
 
@@ -327,7 +336,8 @@ static void destroy_transport(grpc_transport *gt) {
 static void close_transport_locked(grpc_chttp2_transport *t) {
   if (!t->closed) {
     t->closed = 1;
-    connectivity_state_set(&t->global, GRPC_CHANNEL_FATAL_FAILURE);
+    connectivity_state_set(&t->global, GRPC_CHANNEL_FATAL_FAILURE,
+                           "close_transport");
     if (t->ep) {
       grpc_endpoint_shutdown(t->ep);
     }
@@ -393,12 +403,16 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) {
   }
 
   grpc_chttp2_list_remove_incoming_window_updated(&t->global, &s->global);
-  grpc_chttp2_list_remove_writable_window_update_stream(&t->global, &s->global);
+  grpc_chttp2_list_remove_writable_stream(&t->global, &s->global);
 
   gpr_mu_unlock(&t->mu);
 
   for (i = 0; i < STREAM_LIST_COUNT; i++) {
-    GPR_ASSERT(!s->included[i]);
+    if (s->included[i]) {
+      gpr_log(GPR_ERROR, "%s stream %d still included in list %d",
+              t->global.is_client ? "client" : "server", s->global.id, i);
+      abort();
+    }
   }
 
   GPR_ASSERT(s->global.outgoing_sopb == NULL);
@@ -530,7 +544,8 @@ void grpc_chttp2_add_incoming_goaway(
   gpr_free(msg);
   gpr_slice_unref(goaway_text);
   transport_global->seen_goaway = 1;
-  connectivity_state_set(transport_global, GRPC_CHANNEL_FATAL_FAILURE);
+  connectivity_state_set(transport_global, GRPC_CHANNEL_FATAL_FAILURE,
+                         "got_goaway");
 }
 
 static void maybe_start_some_streams(
@@ -555,7 +570,8 @@ static void maybe_start_some_streams(
     transport_global->next_stream_id += 2;
 
     if (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID) {
-      connectivity_state_set(transport_global, GRPC_CHANNEL_TRANSIENT_FAILURE);
+      connectivity_state_set(transport_global, GRPC_CHANNEL_TRANSIENT_FAILURE,
+                             "no_more_stream_ids");
     }
 
     stream_global->outgoing_window =
@@ -574,8 +590,6 @@ static void maybe_start_some_streams(
     grpc_chttp2_list_add_incoming_window_updated(transport_global,
                                                  stream_global);
     grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
-    grpc_chttp2_list_add_writable_window_update_stream(transport_global,
-                                                       stream_global);
 
   }
   /* cancel out streams that will never be started */
@@ -593,10 +607,16 @@ static void perform_stream_op_locked(
     cancel_from_api(transport_global, stream_global, op->cancel_with_status);
   }
 
+  if (op->close_with_status != GRPC_STATUS_OK) {
+    close_from_api(transport_global, stream_global, op->close_with_status,
+                   op->optional_close_message);
+  }
+
   if (op->send_ops) {
     GPR_ASSERT(stream_global->outgoing_sopb == NULL);
     stream_global->send_done_closure = op->on_done_send;
     if (!stream_global->cancelled) {
+      stream_global->written_anything = 1;
       stream_global->outgoing_sopb = op->send_ops;
       if (op->is_last_send &&
           stream_global->write_state == GRPC_WRITE_STATE_OPEN) {
@@ -641,8 +661,7 @@ static void perform_stream_op_locked(
     if (stream_global->id != 0) {
       grpc_chttp2_list_add_read_write_state_changed(transport_global,
                                                     stream_global);
-      grpc_chttp2_list_add_writable_window_update_stream(transport_global,
-                                                         stream_global);
+      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
     }
   }
 
@@ -686,6 +705,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);
 
@@ -705,9 +725,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_locked(t);
-    }
+    close_transport = !grpc_chttp2_has_streams(t);
   }
 
   if (op->set_accept_stream != NULL) {
@@ -720,6 +738,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);
   }
@@ -729,6 +751,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);
+  }
 }
 
 /*
@@ -750,6 +778,7 @@ static void remove_stream(grpc_chttp2_transport *t, gpr_uint32 id) {
   if (!s) {
     s = grpc_chttp2_stream_map_delete(&t->new_stream_map, id);
   }
+  grpc_chttp2_list_remove_writable_stream(&t->global, &s->global);
   GPR_ASSERT(s);
   s->global.in_stream_map = 0;
   if (t->parsing.incoming_stream == &s->parsing) {
@@ -805,6 +834,12 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t) {
                                                            stream_global);
       } else {
         stream_global->write_state = GRPC_WRITE_STATE_SENT_CLOSE;
+        if (stream_global->outgoing_sopb != NULL) {
+          grpc_sopb_reset(stream_global->outgoing_sopb);
+          stream_global->outgoing_sopb = NULL;
+          grpc_chttp2_schedule_closure(transport_global,
+                                       stream_global->send_done_closure, 1);
+        }
         stream_global->read_closed = 1;
         if (!stream_global->published_cancelled) {
           char buffer[GPR_LTOA_MIN_BUFSIZE];
@@ -831,6 +866,9 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t) {
     if (!stream_global->publish_sopb) {
       continue;
     }
+    if (stream_global->writing_now != 0) {
+      continue;
+    }
     /* FIXME(ctiller): we include in_stream_map in our computation of
        whether the stream is write-closed. This is completely bogus,
        but has the effect of delaying stream-closed until the stream
@@ -873,6 +911,108 @@ static void cancel_from_api(grpc_chttp2_transport_global *transport_global,
                                                 stream_global);
 }
 
+static void close_from_api(grpc_chttp2_transport_global *transport_global,
+                           grpc_chttp2_stream_global *stream_global,
+                           grpc_status_code status,
+                           gpr_slice *optional_message) {
+  gpr_slice hdr;
+  gpr_slice status_hdr;
+  gpr_slice message_pfx;
+  gpr_uint8 *p;
+  gpr_uint32 len = 0;
+
+  GPR_ASSERT(status >= 0 && (int)status < 100);
+
+  stream_global->cancelled = 1;
+  stream_global->cancelled_status = status;
+  GPR_ASSERT(stream_global->id != 0);
+  GPR_ASSERT(!stream_global->written_anything);
+
+  /* Hand roll a header block.
+     This is unnecessarily ugly - at some point we should find a more elegant
+     solution.
+     It's complicated by the fact that our send machinery would be dead by the
+     time we got around to sending this, so instead we ignore HPACK compression
+     and just write the uncompressed bytes onto the wire. */
+  status_hdr = gpr_slice_malloc(15 + (status >= 10));
+  p = GPR_SLICE_START_PTR(status_hdr);
+  *p++ = 0x40; /* literal header */
+  *p++ = 11;   /* len(grpc-status) */
+  *p++ = 'g';
+  *p++ = 'r';
+  *p++ = 'p';
+  *p++ = 'c';
+  *p++ = '-';
+  *p++ = 's';
+  *p++ = 't';
+  *p++ = 'a';
+  *p++ = 't';
+  *p++ = 'u';
+  *p++ = 's';
+  if (status < 10) {
+    *p++ = 1;
+    *p++ = '0' + status;
+  } else {
+    *p++ = 2;
+    *p++ = '0' + (status / 10);
+    *p++ = '0' + (status % 10);
+  }
+  GPR_ASSERT(p == GPR_SLICE_END_PTR(status_hdr));
+  len += GPR_SLICE_LENGTH(status_hdr);
+
+  if (optional_message) {
+    GPR_ASSERT(GPR_SLICE_LENGTH(*optional_message) < 127);
+    message_pfx = gpr_slice_malloc(15);
+    p = GPR_SLICE_START_PTR(message_pfx);
+    *p++ = 0x40;
+    *p++ = 12; /* len(grpc-message) */
+    *p++ = 'g';
+    *p++ = 'r';
+    *p++ = 'p';
+    *p++ = 'c';
+    *p++ = '-';
+    *p++ = 'm';
+    *p++ = 'e';
+    *p++ = 's';
+    *p++ = 's';
+    *p++ = 'a';
+    *p++ = 'g';
+    *p++ = 'e';
+    *p++ = GPR_SLICE_LENGTH(*optional_message);
+    GPR_ASSERT(p == GPR_SLICE_END_PTR(message_pfx));
+    len += GPR_SLICE_LENGTH(message_pfx);
+    len += GPR_SLICE_LENGTH(*optional_message);
+  }
+
+  hdr = gpr_slice_malloc(9);
+  p = GPR_SLICE_START_PTR(hdr);
+  *p++ = len >> 16;
+  *p++ = len >> 8;
+  *p++ = len;
+  *p++ = GRPC_CHTTP2_FRAME_HEADER;
+  *p++ = GRPC_CHTTP2_DATA_FLAG_END_STREAM | GRPC_CHTTP2_DATA_FLAG_END_HEADERS;
+  *p++ = stream_global->id >> 24;
+  *p++ = stream_global->id >> 16;
+  *p++ = stream_global->id >> 8;
+  *p++ = stream_global->id;
+  GPR_ASSERT(p == GPR_SLICE_END_PTR(hdr));
+
+  gpr_slice_buffer_add(&transport_global->qbuf, hdr);
+  gpr_slice_buffer_add(&transport_global->qbuf, status_hdr);
+  if (optional_message) {
+    gpr_slice_buffer_add(&transport_global->qbuf, message_pfx);
+    gpr_slice_buffer_add(&transport_global->qbuf,
+                         gpr_slice_ref(*optional_message));
+  }
+
+  gpr_slice_buffer_add(
+      &transport_global->qbuf,
+      grpc_chttp2_rst_stream_create(stream_global->id, GRPC_CHTTP2_NO_ERROR));
+
+  grpc_chttp2_list_add_read_write_state_changed(transport_global,
+                                                stream_global);
+}
+
 static void cancel_stream_cb(grpc_chttp2_transport_global *transport_global,
                              void *user_data,
                              grpc_chttp2_stream_global *stream_global) {
@@ -1001,12 +1141,12 @@ static void schedule_closure_for_connectivity(void *a,
 
 static void connectivity_state_set(
     grpc_chttp2_transport_global *transport_global,
-    grpc_connectivity_state state) {
+    grpc_connectivity_state state, const char *reason) {
   GRPC_CHTTP2_IF_TRACING(
       gpr_log(GPR_DEBUG, "set connectivity_state=%d", state));
   grpc_connectivity_state_set_with_scheduler(
       &TRANSPORT_FROM_GLOBAL(transport_global)->channel_callback.state_tracker,
-      state, schedule_closure_for_connectivity, transport_global);
+      state, schedule_closure_for_connectivity, transport_global, reason);
 }
 
 void grpc_chttp2_schedule_closure(
@@ -1034,6 +1174,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
  */
@@ -1069,9 +1216,17 @@ void grpc_chttp2_flowctl_trace(const char *file, int line, const char *reason,
  * INTEGRATION GLUE
  */
 
-static const grpc_transport_vtable vtable = {
-    sizeof(grpc_chttp2_stream), init_stream,    perform_stream_op,
-    perform_transport_op,       destroy_stream, destroy_transport};
+static char *chttp2_get_peer(grpc_transport *t) {
+  return gpr_strdup(((grpc_chttp2_transport *)t)->peer_string);
+}
+
+static const grpc_transport_vtable vtable = {sizeof(grpc_chttp2_stream),
+                                             init_stream,
+                                             perform_stream_op,
+                                             perform_transport_op,
+                                             destroy_stream,
+                                             destroy_transport,
+                                             chttp2_get_peer};
 
 grpc_transport *grpc_create_chttp2_transport(
     const grpc_channel_args *channel_args, grpc_endpoint *ep, grpc_mdctx *mdctx,
diff --git a/src/core/transport/connectivity_state.c b/src/core/transport/connectivity_state.c
index 1091ceae44..61d26f06f0 100644
--- a/src/core/transport/connectivity_state.c
+++ b/src/core/transport/connectivity_state.c
@@ -34,11 +34,33 @@
 #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 +76,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 +87,11 @@ 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);
@@ -79,12 +107,19 @@ int grpc_connectivity_state_notify_on_state_change(
 
 void grpc_connectivity_state_set_with_scheduler(
     grpc_connectivity_state_tracker *tracker, grpc_connectivity_state state,
-    void (*scheduler)(void *arg, grpc_iomgr_closure *closure), void *arg) {
+    void (*scheduler)(void *arg, grpc_iomgr_closure *closure), void *arg,
+    const char *reason) {
   grpc_connectivity_state_watcher *new = NULL;
   grpc_connectivity_state_watcher *w;
+  if (grpc_connectivity_state_trace) {
+    gpr_log(GPR_DEBUG, "SET: %s: %s --> %s [%s]", tracker->name,
+            grpc_connectivity_state_name(tracker->current_state),
+            grpc_connectivity_state_name(state), reason);
+  }
   if (tracker->current_state == state) {
     return;
   }
+  GPR_ASSERT(tracker->current_state != GRPC_CHANNEL_FATAL_FAILURE);
   tracker->current_state = state;
   while ((w = tracker->watchers)) {
     tracker->watchers = w->next;
@@ -106,7 +141,8 @@ static void default_scheduler(void *ignored, grpc_iomgr_closure *closure) {
 }
 
 void grpc_connectivity_state_set(grpc_connectivity_state_tracker *tracker,
-                                 grpc_connectivity_state state) {
+                                 grpc_connectivity_state state,
+                                 const char *reason) {
   grpc_connectivity_state_set_with_scheduler(tracker, state, default_scheduler,
-                                             NULL);
+                                             NULL, reason);
 }
diff --git a/src/core/transport/connectivity_state.h b/src/core/transport/connectivity_state.h
index bbdcbcb069..a3b0b80c98 100644
--- a/src/core/transport/connectivity_state.h
+++ b/src/core/transport/connectivity_state.h
@@ -51,17 +51,24 @@ 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,
-                                 grpc_connectivity_state state);
+                                 grpc_connectivity_state state,
+                                 const char *reason);
 void grpc_connectivity_state_set_with_scheduler(
     grpc_connectivity_state_tracker *tracker, grpc_connectivity_state state,
-    void (*scheduler)(void *arg, grpc_iomgr_closure *closure), void *arg);
+    void (*scheduler)(void *arg, grpc_iomgr_closure *closure), void *arg,
+    const char *reason);
 
 grpc_connectivity_state grpc_connectivity_state_check(
     grpc_connectivity_state_tracker *tracker);
diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c
index e95b7a21f9..44d32b6cb2 100644
--- a/src/core/transport/metadata.c
+++ b/src/core/transport/metadata.c
@@ -135,7 +135,9 @@ static void unlock(grpc_mdctx *ctx) {
   if (ctx->refs == 0) {
     /* uncomment if you're having trouble diagnosing an mdelem leak to make
        things clearer (slows down destruction a lot, however) */
+#ifdef GRPC_METADATA_REFCOUNT_DEBUG
     gc_mdtab(ctx);
+#endif
     if (ctx->mdtab_count && ctx->mdtab_count == ctx->mdtab_free) {
       discard_metadata(ctx);
     }
@@ -309,7 +311,37 @@ static void slice_unref(void *p) {
   unlock(ctx);
 }
 
-grpc_mdstr *grpc_mdstr_from_string(grpc_mdctx *ctx, const char *str) {
+grpc_mdstr *grpc_mdstr_from_string(grpc_mdctx *ctx, const char *str, int canonicalize_key) {
+  if (canonicalize_key) {
+    size_t len;
+    size_t i;
+    int canonical = 1;
+
+    for (i = 0; str[i]; i++) {
+      if (str[i] >= 'A' && str[i] <= 'Z') {
+        canonical = 0;
+        /* Keep going in loop just to get string length */
+      }
+    }
+    len = i;
+
+    if (canonical) {
+      return grpc_mdstr_from_buffer(ctx, (const gpr_uint8 *)str, len);
+    } else {
+      char *copy = gpr_malloc(len);
+      grpc_mdstr *ret;
+      for (i = 0; i < len; i++) {
+        if (str[i] >= 'A' && str[i] <= 'Z') {
+          copy[i] = str[i] - 'A' + 'a';
+        } else {
+          copy[i] = str[i];
+        }
+      }
+      ret = grpc_mdstr_from_buffer(ctx, (const gpr_uint8 *)copy, len);
+      gpr_free(copy);
+      return ret;
+    }
+  }
   return grpc_mdstr_from_buffer(ctx, (const gpr_uint8 *)str, strlen(str));
 }
 
@@ -491,8 +523,8 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdctx *ctx,
 grpc_mdelem *grpc_mdelem_from_strings(grpc_mdctx *ctx, const char *key,
                                       const char *value) {
   return grpc_mdelem_from_metadata_strings(ctx,
-                                           grpc_mdstr_from_string(ctx, key),
-                                           grpc_mdstr_from_string(ctx, value));
+                                           grpc_mdstr_from_string(ctx, key, 0),
+                                           grpc_mdstr_from_string(ctx, value, 0));
 }
 
 grpc_mdelem *grpc_mdelem_from_slices(grpc_mdctx *ctx, gpr_slice key,
@@ -504,9 +536,10 @@ grpc_mdelem *grpc_mdelem_from_slices(grpc_mdctx *ctx, gpr_slice key,
 grpc_mdelem *grpc_mdelem_from_string_and_buffer(grpc_mdctx *ctx,
                                                 const char *key,
                                                 const gpr_uint8 *value,
-                                                size_t value_length) {
+                                                size_t value_length,
+                                                int canonicalize_key) {
   return grpc_mdelem_from_metadata_strings(
-      ctx, grpc_mdstr_from_string(ctx, key),
+      ctx, grpc_mdstr_from_string(ctx, key, canonicalize_key),
       grpc_mdstr_from_buffer(ctx, value, value_length));
 }
 
diff --git a/src/core/transport/metadata.h b/src/core/transport/metadata.h
index 99b15322c3..15ef9bb555 100644
--- a/src/core/transport/metadata.h
+++ b/src/core/transport/metadata.h
@@ -95,7 +95,7 @@ size_t grpc_mdctx_get_mdtab_free_test_only(grpc_mdctx *mdctx);
 
 /* Constructors for grpc_mdstr instances; take a variety of data types that
    clients may have handy */
-grpc_mdstr *grpc_mdstr_from_string(grpc_mdctx *ctx, const char *str);
+grpc_mdstr *grpc_mdstr_from_string(grpc_mdctx *ctx, const char *str, int perform_key_canonicalization);
 /* Unrefs the slice. */
 grpc_mdstr *grpc_mdstr_from_slice(grpc_mdctx *ctx, gpr_slice slice);
 grpc_mdstr *grpc_mdstr_from_buffer(grpc_mdctx *ctx, const gpr_uint8 *str,
@@ -117,7 +117,8 @@ grpc_mdelem *grpc_mdelem_from_slices(grpc_mdctx *ctx, gpr_slice key,
 grpc_mdelem *grpc_mdelem_from_string_and_buffer(grpc_mdctx *ctx,
                                                 const char *key,
                                                 const gpr_uint8 *value,
-                                                size_t value_length);
+                                                size_t value_length,
+                                                int canonicalize_key);
 
 /* Mutator and accessor for grpc_mdelem user data. The destructor function
    is used as a type tag and is checked during user_data fetch. */
diff --git a/src/core/transport/stream_op.h b/src/core/transport/stream_op.h
index 99ba305a98..37f18b02d9 100644
--- a/src/core/transport/stream_op.h
+++ b/src/core/transport/stream_op.h
@@ -109,7 +109,7 @@ void grpc_metadata_batch_move(grpc_metadata_batch *dst,
                               grpc_metadata_batch *src);
 
 /** Add \a storage to the beginning of \a batch. storage->md is
-    assumed to be valid. 
+    assumed to be valid.
     \a storage is owned by the caller and must survive for the
     lifetime of batch. This usually means it should be around
     for the lifetime of the call. */
diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c
index 2689e3028a..c0d92cf93f 100644
--- a/src/core/transport/transport.c
+++ b/src/core/transport/transport.c
@@ -32,6 +32,8 @@
  */
 
 #include "src/core/transport/transport.h"
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
 #include "src/core/transport/transport_impl.h"
 
 size_t grpc_transport_stream_size(grpc_transport *transport) {
@@ -65,6 +67,10 @@ void grpc_transport_destroy_stream(grpc_transport *transport,
   transport->vtable->destroy_stream(transport, stream);
 }
 
+char *grpc_transport_get_peer(grpc_transport *transport) {
+  return transport->vtable->get_peer(transport);
+}
+
 void grpc_transport_stream_op_finish_with_failure(
     grpc_transport_stream_op *op) {
   if (op->send_ops) {
@@ -79,12 +85,54 @@ void grpc_transport_stream_op_finish_with_failure(
 }
 
 void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op,
-                                               grpc_status_code status,
-                                               grpc_mdstr *message) {
+                                               grpc_status_code status) {
+  GPR_ASSERT(status != GRPC_STATUS_OK);
   if (op->cancel_with_status == GRPC_STATUS_OK) {
     op->cancel_with_status = status;
   }
-  if (message) {
-    GRPC_MDSTR_UNREF(message);
+  if (op->close_with_status != GRPC_STATUS_OK) {
+    op->close_with_status = GRPC_STATUS_OK;
+    if (op->optional_close_message != NULL) {
+      gpr_slice_unref(*op->optional_close_message);
+      op->optional_close_message = NULL;
+    }
+  }
+}
+
+typedef struct {
+  gpr_slice message;
+  grpc_iomgr_closure *then_call;
+  grpc_iomgr_closure closure;
+} close_message_data;
+
+static void free_message(void *p, int iomgr_success) {
+  close_message_data *cmd = p;
+  gpr_slice_unref(cmd->message);
+  if (cmd->then_call != NULL) {
+    cmd->then_call->cb(cmd->then_call->cb_arg, iomgr_success);
+  }
+  gpr_free(cmd);
+}
+
+void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op,
+                                        grpc_status_code status,
+                                        gpr_slice *optional_message) {
+  close_message_data *cmd;
+  GPR_ASSERT(status != GRPC_STATUS_OK);
+  if (op->cancel_with_status != GRPC_STATUS_OK ||
+      op->close_with_status != GRPC_STATUS_OK) {
+    if (optional_message) {
+      gpr_slice_unref(*optional_message);
+    }
+    return;
+  }
+  if (optional_message) {
+    cmd = gpr_malloc(sizeof(*cmd));
+    cmd->message = *optional_message;
+    cmd->then_call = op->on_consumed;
+    grpc_iomgr_closure_init(&cmd->closure, free_message, cmd);
+    op->on_consumed = &cmd->closure;
+    op->optional_close_message = &cmd->message;
   }
+  op->close_with_status = status;
 }
diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h
index 64503604ee..92c1f38c5e 100644
--- a/src/core/transport/transport.h
+++ b/src/core/transport/transport.h
@@ -80,8 +80,14 @@ typedef struct grpc_transport_stream_op {
 
   grpc_pollset *bind_pollset;
 
+  /** If != GRPC_STATUS_OK, cancel this stream */
   grpc_status_code cancel_with_status;
 
+  /** If != GRPC_STATUS_OK, send grpc-status, grpc-message, and close this
+      stream for both reading and writing */
+  grpc_status_code close_with_status;
+  gpr_slice *optional_close_message;
+
   /* Indexes correspond to grpc_context_index enum values */
   grpc_call_context_element *context;
 } grpc_transport_stream_op;
@@ -109,6 +115,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;
@@ -146,8 +154,11 @@ void grpc_transport_destroy_stream(grpc_transport *transport,
 void grpc_transport_stream_op_finish_with_failure(grpc_transport_stream_op *op);
 
 void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op,
-                                               grpc_status_code status,
-                                               grpc_mdstr *message);
+                                               grpc_status_code status);
+
+void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op,
+                                        grpc_status_code status,
+                                        gpr_slice *optional_message);
 
 char *grpc_transport_stream_op_string(grpc_transport_stream_op *op);
 
@@ -182,4 +193,7 @@ void grpc_transport_close(grpc_transport *transport);
 /* Destroy the transport */
 void grpc_transport_destroy(grpc_transport *transport);
 
+/* Get the transports peer */
+char *grpc_transport_get_peer(grpc_transport *transport);
+
 #endif /* GRPC_INTERNAL_CORE_TRANSPORT_TRANSPORT_H */
diff --git a/src/core/transport/transport_impl.h b/src/core/transport/transport_impl.h
index 515721dfb6..d3bbdf6c27 100644
--- a/src/core/transport/transport_impl.h
+++ b/src/core/transport/transport_impl.h
@@ -58,6 +58,9 @@ typedef struct grpc_transport_vtable {
 
   /* implementation of grpc_transport_destroy */
   void (*destroy)(grpc_transport *self);
+
+  /* implementation of grpc_transport_get_peer */
+  char *(*get_peer)(grpc_transport *self);
 } grpc_transport_vtable;
 
 /* an instance of a grpc transport */
diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c
index 10d796fc15..f62c340e97 100644
--- a/src/core/transport/transport_op_string.c
+++ b/src/core/transport/transport_op_string.c
@@ -116,10 +116,9 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) {
   if (op->send_ops) {
     if (!first) gpr_strvec_add(&b, gpr_strdup(" "));
     first = 0;
-    gpr_strvec_add(&b, gpr_strdup("SEND"));
-    if (op->is_last_send) {
-      gpr_strvec_add(&b, gpr_strdup("_LAST"));
-    }
+    gpr_asprintf(&tmp, "SEND%s:%p", op->is_last_send ? "_LAST" : "",
+                 op->on_done_send);
+    gpr_strvec_add(&b, tmp);
     gpr_strvec_add(&b, gpr_strdup("["));
     gpr_strvec_add(&b, grpc_sopb_string(op->send_ops));
     gpr_strvec_add(&b, gpr_strdup("]"));
@@ -128,7 +127,8 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) {
   if (op->recv_ops) {
     if (!first) gpr_strvec_add(&b, gpr_strdup(" "));
     first = 0;
-    gpr_asprintf(&tmp, "RECV:max_recv_bytes=%d", op->max_recv_bytes);
+    gpr_asprintf(&tmp, "RECV:%p:max_recv_bytes=%d", op->on_done_recv,
+                 op->max_recv_bytes);
     gpr_strvec_add(&b, tmp);
   }