post-merge

16 files changed, 184 insertions, 123 deletions

diff --git a/src/core/transport/chttp2/frame_data.c b/src/core/transport/chttp2/frame_data.c
index 7e3980159e..40bf2ebd79 100644
--- a/src/core/transport/chttp2/frame_data.c
+++ b/src/core/transport/chttp2/frame_data.c
@@ -76,6 +76,7 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse(
   gpr_uint8 *const end = GPR_SLICE_END_PTR(slice);
   gpr_uint8 *cur = beg;
   grpc_chttp2_data_parser *p = parser;
+  gpr_uint32 message_flags = 0;
 
   if (is_last && p->is_last_frame) {
     stream_parsing->received_close = 1;
@@ -91,11 +92,11 @@ 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:
-          gpr_log(GPR_ERROR, "Compressed GRPC frames not yet supported");
-          return GRPC_CHTTP2_STREAM_ERROR;
+          p->is_frame_compressed = 1;  /* GPR_TRUE */
+          break;
         default:
           gpr_log(GPR_ERROR, "Bad GRPC frame type 0x%02x", p->frame_type);
           return GRPC_CHTTP2_STREAM_ERROR;
@@ -130,7 +131,11 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse(
       p->frame_size |= ((gpr_uint32)*cur);
       p->state = GRPC_CHTTP2_DATA_FRAME;
       ++cur;
-      grpc_sopb_add_begin_message(&p->incoming_sopb, p->frame_size, 0);
+      if (p->is_frame_compressed) {
+        message_flags |= GRPC_WRITE_INTERNAL_COMPRESS;
+      }
+      grpc_sopb_add_begin_message(&p->incoming_sopb, p->frame_size,
+                                  message_flags);
     /* fallthrough */
     case GRPC_CHTTP2_DATA_FRAME:
       if (cur == end) {
diff --git a/src/core/transport/chttp2/frame_data.h b/src/core/transport/chttp2/frame_data.h
index 8d6cfcb841..23957b05ad 100644
--- a/src/core/transport/chttp2/frame_data.h
+++ b/src/core/transport/chttp2/frame_data.h
@@ -56,6 +56,7 @@ typedef struct {
   gpr_uint8 frame_type;
   gpr_uint32 frame_size;
 
+  int is_frame_compressed;
   grpc_stream_op_buffer incoming_sopb;
 } grpc_chttp2_data_parser;
 
diff --git a/src/core/transport/chttp2/incoming_metadata.c b/src/core/transport/chttp2/incoming_metadata.c
index 77162a6864..974b864ffb 100644
--- a/src/core/transport/chttp2/incoming_metadata.c
+++ b/src/core/transport/chttp2/incoming_metadata.c
@@ -42,7 +42,7 @@
 
 void grpc_chttp2_incoming_metadata_buffer_init(
     grpc_chttp2_incoming_metadata_buffer *buffer) {
-  buffer->deadline = gpr_inf_future;
+  buffer->deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
 }
 
 void grpc_chttp2_incoming_metadata_buffer_destroy(
@@ -87,7 +87,7 @@ void grpc_chttp2_incoming_metadata_buffer_place_metadata_batch_into(
   b.list.tail = (void *)(gpr_intptr)buffer->count;
   b.garbage.head = b.garbage.tail = NULL;
   b.deadline = buffer->deadline;
-  buffer->deadline = gpr_inf_future;
+  buffer->deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
 
   grpc_sopb_add_metadata(sopb, b);
 }
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index e5e6f445b7..f0eeb6de50 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,
@@ -286,6 +285,7 @@ struct grpc_chttp2_transport {
   grpc_endpoint *ep;
   grpc_mdctx *metadata_context;
   gpr_refcount refs;
+  char *peer_string;
 
   gpr_mu mu;
 
@@ -382,6 +382,8 @@ typedef struct {
   gpr_uint8 published_cancelled;
   /** is this stream in the stream map? (boolean) */
   gpr_uint8 in_stream_map;
+  /** is this stream actively being written? */
+  gpr_uint8 writing_now;
 
   /** stream state already published to the upper layer */
   grpc_stream_state published_state;
@@ -474,11 +476,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 +518,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 82362544d5..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 */
@@ -588,7 +587,7 @@ static void on_header(void *tp, grpc_mdelem *md) {
   GPR_ASSERT(stream_parsing);
 
   GRPC_CHTTP2_IF_TRACING(gpr_log(
-      GPR_INFO, "HTTP:%d:HDR: %s: %s", stream_parsing->id,
+      GPR_INFO, "HTTP:%d:HDR:%s: %s: %s", stream_parsing->id,
       transport_parsing->is_client ? "CLI" : "SVR",
       grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value)));
 
@@ -601,13 +600,13 @@ static void on_header(void *tp, grpc_mdelem *md) {
                                       cached_timeout)) {
         gpr_log(GPR_ERROR, "Ignoring bad timeout value '%s'",
                 grpc_mdstr_as_c_string(md->value));
-        *cached_timeout = gpr_inf_future;
+        *cached_timeout = gpr_inf_future(GPR_CLOCK_REALTIME);
       }
       grpc_mdelem_set_user_data(md, free_timeout, cached_timeout);
     }
     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 d553d80085..65b31a5afd 100644
--- a/src/core/transport/chttp2/stream_encoder.c
+++ b/src/core/transport/chttp2/stream_encoder.c
@@ -438,7 +438,7 @@ static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline,
   char timeout_str[GRPC_CHTTP2_TIMEOUT_ENCODE_MIN_BUFSIZE];
   grpc_mdelem *mdelem;
   grpc_chttp2_encode_timeout(
-      gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME)), timeout_str);
+      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));
@@ -477,6 +477,7 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count,
   gpr_uint32 flow_controlled_bytes_taken = 0;
   gpr_uint32 curop = 0;
   gpr_uint8 *p;
+  int compressed_flag_set = 0;
 
   while (curop < *inops_count) {
     GPR_ASSERT(flow_controlled_bytes_taken <= max_flow_controlled_bytes);
@@ -496,9 +497,12 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count,
       case GRPC_OP_BEGIN_MESSAGE:
         /* begin op: for now we just convert the op to a slice and fall
            through - this lets us reuse the slice framing code below */
+        compressed_flag_set =
+            (op->data.begin_message.flags & GRPC_WRITE_INTERNAL_COMPRESS) != 0;
         slice = gpr_slice_malloc(5);
+
         p = GPR_SLICE_START_PTR(slice);
-        p[0] = 0;
+        p[0] = compressed_flag_set;
         p[1] = op->data.begin_message.length >> 24;
         p[2] = op->data.begin_message.length >> 16;
         p[3] = op->data.begin_message.length >> 8;
@@ -556,6 +560,7 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
   grpc_mdctx *mdctx = compressor->mdctx;
   grpc_linked_mdelem *l;
   int need_unref = 0;
+  gpr_timespec deadline;
 
   GPR_ASSERT(stream_id != 0);
 
@@ -585,8 +590,9 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
           l->md = hpack_enc(compressor, l->md, &st);
           need_unref |= l->md != NULL;
         }
-        if (gpr_time_cmp(op->data.metadata.deadline, gpr_inf_future) != 0) {
-          deadline_enc(compressor, op->data.metadata.deadline, &st);
+        deadline = op->data.metadata.deadline;
+        if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) != 0) {
+          deadline_enc(compressor, deadline, &st);
         }
         curop++;
         break;
diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c
index 590f6abfbc..9e68c1e146 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,18 @@ 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);
+  gpr_log(GPR_DEBUG, "add:%d:%d:%d:%d", stream_global->id,
+          stream_global->write_state, stream_global->in_stream_map,
+          stream_global->read_closed);
+  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 +185,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 +238,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/timeout_encoding.c b/src/core/transport/chttp2/timeout_encoding.c
index 33915c4039..1dd768ada4 100644
--- a/src/core/transport/chttp2/timeout_encoding.c
+++ b/src/core/transport/chttp2/timeout_encoding.c
@@ -147,7 +147,7 @@ int grpc_chttp2_decode_timeout(const char *buffer, gpr_timespec *timeout) {
     gpr_uint32 xp = x * 10 + *p - '0';
     have_digit = 1;
     if (xp < x) {
-      *timeout = gpr_inf_future;
+      *timeout = gpr_inf_future(GPR_CLOCK_REALTIME);
       return 1;
     }
     x = xp;
@@ -159,22 +159,22 @@ int grpc_chttp2_decode_timeout(const char *buffer, gpr_timespec *timeout) {
   /* decode unit specifier */
   switch (*p) {
     case 'n':
-      *timeout = gpr_time_from_nanos(x);
+      *timeout = gpr_time_from_nanos(x, GPR_TIMESPAN);
       break;
     case 'u':
-      *timeout = gpr_time_from_micros(x);
+      *timeout = gpr_time_from_micros(x, GPR_TIMESPAN);
       break;
     case 'm':
-      *timeout = gpr_time_from_millis(x);
+      *timeout = gpr_time_from_millis(x, GPR_TIMESPAN);
       break;
     case 'S':
-      *timeout = gpr_time_from_seconds(x);
+      *timeout = gpr_time_from_seconds(x, GPR_TIMESPAN);
       break;
     case 'M':
-      *timeout = gpr_time_from_minutes(x);
+      *timeout = gpr_time_from_minutes(x, GPR_TIMESPAN);
       break;
     case 'H':
-      *timeout = gpr_time_from_hours(x);
+      *timeout = gpr_time_from_hours(x, GPR_TIMESPAN);
       break;
     default:
       return 0;
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index d8ec117aa5..d39b0c42f7 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,54 @@ 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;
+    GPR_ASSERT(!stream_global->writing_now);
+
+    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 +123,11 @@ 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 = 1;
+      grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
+    } else if (stream_writing->sopb.nops > 0 ||
+               stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
+      stream_global->writing_now = 1;
       grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
     }
   }
@@ -205,6 +215,8 @@ void grpc_chttp2_cleanup_writing(
 
   while (grpc_chttp2_list_pop_written_stream(
       transport_global, transport_writing, &stream_global, &stream_writing)) {
+    GPR_ASSERT(stream_global->writing_now);
+    stream_global->writing_now = 0;
     if (stream_global->outgoing_sopb != NULL &&
         stream_global->outgoing_sopb->nops == 0) {
       stream_global->outgoing_sopb = NULL;
@@ -216,9 +228,9 @@ void grpc_chttp2_cleanup_writing(
       if (!transport_global->is_client) {
         stream_global->read_closed = 1;
       }
-      grpc_chttp2_list_add_read_write_state_changed(transport_global,
-                                                    stream_global);
     }
+    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..5f49b2ddd6 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -168,6 +168,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 +218,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;
@@ -393,12 +395,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);
@@ -574,8 +580,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 */
@@ -641,8 +645,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);
     }
   }
 
@@ -750,6 +753,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) {
@@ -831,6 +835,9 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t) {
     if (!stream_global->publish_sopb) {
       continue;
     }
+    if (stream_global->writing_now) {
+      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
@@ -1069,9 +1076,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/stream_op.c b/src/core/transport/stream_op.c
index fdb50c6b71..a5dfec9d50 100644
--- a/src/core/transport/stream_op.c
+++ b/src/core/transport/stream_op.c
@@ -205,7 +205,7 @@ void grpc_metadata_batch_assert_ok(grpc_metadata_batch *batch) {
 void grpc_metadata_batch_init(grpc_metadata_batch *batch) {
   batch->list.head = batch->list.tail = batch->garbage.head = batch->garbage.tail =
       NULL;
-  batch->deadline = gpr_inf_future;
+  batch->deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
 }
 
 void grpc_metadata_batch_destroy(grpc_metadata_batch *batch) {
@@ -286,6 +286,12 @@ void grpc_metadata_batch_merge(grpc_metadata_batch *target,
   }
 }
 
+void grpc_metadata_batch_move(grpc_metadata_batch *dst,
+                               grpc_metadata_batch *src) {
+  *dst = *src;
+  memset(src, 0, sizeof(grpc_metadata_batch));
+}
+
 void grpc_metadata_batch_filter(grpc_metadata_batch *batch,
                                 grpc_mdelem *(*filter)(void *user_data,
                                                        grpc_mdelem *elem),
diff --git a/src/core/transport/stream_op.h b/src/core/transport/stream_op.h
index 964d39d14f..f27ef1b66b 100644
--- a/src/core/transport/stream_op.h
+++ b/src/core/transport/stream_op.h
@@ -102,6 +102,11 @@ void grpc_metadata_batch_destroy(grpc_metadata_batch *batch);
 void grpc_metadata_batch_merge(grpc_metadata_batch *target,
                                grpc_metadata_batch *add);
 
+/** Moves the metadata information from \a src to \a dst. Upon return, \a src is
+ * zeroed. */
+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. 
     \a storage is owned by the caller and must survive for the
diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c
index 2689e3028a..69c00b6a4f 100644
--- a/src/core/transport/transport.c
+++ b/src/core/transport/transport.c
@@ -65,6 +65,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) {
diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h
index 64503604ee..553779602a 100644
--- a/src/core/transport/transport.h
+++ b/src/core/transport/transport.h
@@ -182,4 +182,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 862eb40c4b..f62c340e97 100644
--- a/src/core/transport/transport_op_string.c
+++ b/src/core/transport/transport_op_string.c
@@ -61,7 +61,7 @@ static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) {
     if (m != md.list.head) gpr_strvec_add(b, gpr_strdup(", "));
     put_metadata(b, m->md);
   }
-  if (gpr_time_cmp(md.deadline, gpr_inf_future) != 0) {
+  if (gpr_time_cmp(md.deadline, gpr_inf_future(md.deadline.clock_type)) != 0) {
     char *tmp;
     gpr_asprintf(&tmp, " deadline=%d.%09d", md.deadline.tv_sec,
                  md.deadline.tv_nsec);
@@ -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);
   }