Merge github.com:grpc/grpc into cleaner-posix2

18 files changed, 256 insertions, 193 deletions

diff --git a/src/core/census/context.c b/src/core/census/context.c
index e60330de64..89b8ee0b39 100644
--- a/src/core/census/context.c
+++ b/src/core/census/context.c
@@ -60,10 +60,10 @@
 //   limit of 255 for both CENSUS_MAX_TAG_KV_LEN and CENSUS_MAX_PROPAGATED_TAGS.
 // * Keep all tag information (keys/values/flags) in a single memory buffer,
 //   that can be directly copied to the wire.
-// * Binary tags share the same structure as, but are encoded separately from,
-//   non-binary tags. This is primarily because non-binary tags are far more
-//   likely to be repeated across multiple RPC calls, so are more efficiently
-//   cached and compressed in any metadata schemes.
+
+// min and max valid chars in tag keys and values. All printable ASCII is OK.
+#define MIN_VALID_TAG_CHAR 32   // ' '
+#define MAX_VALID_TAG_CHAR 126  // '~'
 
 // Structure representing a set of tags. Essentially a count of number of tags
 // present, and pointer to a chunk of memory that contains the per-tag details.
@@ -77,7 +77,7 @@ struct tag_set {
   char *kvm;        // key/value memory. Consists of repeated entries of:
   //   Offset  Size  Description
   //     0      1    Key length, including trailing 0. (K)
-  //     1      1    Value length. (V)
+  //     1      1    Value length, including trailing 0 (V)
   //     2      1    Flags
   //     3      K    Key bytes
   //     3 + K  V    Value bytes
@@ -108,19 +108,36 @@ struct raw_tag {
 #define CENSUS_TAG_DELETED CENSUS_TAG_RESERVED
 #define CENSUS_TAG_IS_DELETED(flags) (flags & CENSUS_TAG_DELETED)
 
-// Primary (external) representation of a context. Composed of 3 underlying
-// tag_set structs, one for each of the binary/printable propagated tags, and
-// one for everything else. This is to efficiently support tag
-// encoding/decoding.
+// Primary representation of a context. Composed of 2 underlying tag_set
+// structs, one each for propagated and local (non-propagated) tags. This is
+// to efficiently support tag encoding/decoding.
+// TODO(aveitch): need to add tracing id's/structure.
 struct census_context {
-  struct tag_set tags[3];
+  struct tag_set tags[2];
   census_context_status status;
 };
 
 // Indices into the tags member of census_context
 #define PROPAGATED_TAGS 0
-#define PROPAGATED_BINARY_TAGS 1
-#define LOCAL_TAGS 2
+#define LOCAL_TAGS 1
+
+// Validate (check all characters are in range and size is less than limit) a
+// key or value string. Returns 0 if the string is invalid, or the length
+// (including terminator) if valid.
+static size_t validate_tag(const char *kv) {
+  size_t len = 1;
+  char ch;
+  while ((ch = *kv++) != 0) {
+    if (ch < MIN_VALID_TAG_CHAR || ch > MAX_VALID_TAG_CHAR) {
+      return 0;
+    }
+    len++;
+  }
+  if (len > CENSUS_MAX_TAG_KV_LEN) {
+    return 0;
+  }
+  return len;
+}
 
 // Extract a raw tag given a pointer (raw) to the tag header. Allow for some
 // extra bytes in the tag header (see encode/decode functions for usage: this
@@ -166,9 +183,7 @@ static bool context_delete_tag(census_context *context, const census_tag *tag,
                                size_t key_len) {
   return (
       tag_set_delete_tag(&context->tags[LOCAL_TAGS], tag->key, key_len) ||
-      tag_set_delete_tag(&context->tags[PROPAGATED_TAGS], tag->key, key_len) ||
-      tag_set_delete_tag(&context->tags[PROPAGATED_BINARY_TAGS], tag->key,
-                         key_len));
+      tag_set_delete_tag(&context->tags[PROPAGATED_TAGS], tag->key, key_len));
 }
 
 // Add a tag to a tag_set. Return true on success, false if the tag could
@@ -176,11 +191,11 @@ static bool context_delete_tag(census_context *context, const census_tag *tag,
 // not be called if the tag may already exist (in a non-deleted state) in
 // the tag_set, as that would result in two tags with the same key.
 static bool tag_set_add_tag(struct tag_set *tags, const census_tag *tag,
-                            size_t key_len) {
+                            size_t key_len, size_t value_len) {
   if (tags->ntags == CENSUS_MAX_PROPAGATED_TAGS) {
     return false;
   }
-  const size_t tag_size = key_len + tag->value_len + TAG_HEADER_SIZE;
+  const size_t tag_size = key_len + value_len + TAG_HEADER_SIZE;
   if (tags->kvm_used + tag_size > tags->kvm_size) {
     // allocate new memory if needed
     tags->kvm_size += 2 * CENSUS_MAX_TAG_KV_LEN + TAG_HEADER_SIZE;
@@ -191,13 +206,12 @@ static bool tag_set_add_tag(struct tag_set *tags, const census_tag *tag,
   }
   char *kvp = tags->kvm + tags->kvm_used;
   *kvp++ = (char)key_len;
-  *kvp++ = (char)tag->value_len;
+  *kvp++ = (char)value_len;
   // ensure reserved flags are not used.
-  *kvp++ = (char)(tag->flags & (CENSUS_TAG_PROPAGATE | CENSUS_TAG_STATS |
-                                CENSUS_TAG_BINARY));
+  *kvp++ = (char)(tag->flags & (CENSUS_TAG_PROPAGATE | CENSUS_TAG_STATS));
   memcpy(kvp, tag->key, key_len);
   kvp += key_len;
-  memcpy(kvp, tag->value, tag->value_len);
+  memcpy(kvp, tag->value, value_len);
   tags->kvm_used += tag_size;
   tags->ntags++;
   tags->ntags_alloc++;
@@ -207,30 +221,20 @@ static bool tag_set_add_tag(struct tag_set *tags, const census_tag *tag,
 // Add/modify/delete a tag to/in a context. Caller must validate that tag key
 // etc. are valid.
 static void context_modify_tag(census_context *context, const census_tag *tag,
-                               size_t key_len) {
+                               size_t key_len, size_t value_len) {
   // First delete the tag if it is already present.
   bool deleted = context_delete_tag(context, tag, key_len);
-  // Determine if we need to add it back.
-  bool call_add = tag->value != NULL && tag->value_len != 0;
   bool added = false;
-  if (call_add) {
-    if (CENSUS_TAG_IS_PROPAGATED(tag->flags)) {
-      if (CENSUS_TAG_IS_BINARY(tag->flags)) {
-        added = tag_set_add_tag(&context->tags[PROPAGATED_BINARY_TAGS], tag,
-                                key_len);
-      } else {
-        added = tag_set_add_tag(&context->tags[PROPAGATED_TAGS], tag, key_len);
-      }
-    } else {
-      added = tag_set_add_tag(&context->tags[LOCAL_TAGS], tag, key_len);
-    }
+  if (CENSUS_TAG_IS_PROPAGATED(tag->flags)) {
+    added = tag_set_add_tag(&context->tags[PROPAGATED_TAGS], tag, key_len,
+                            value_len);
+  } else {
+    added =
+        tag_set_add_tag(&context->tags[LOCAL_TAGS], tag, key_len, value_len);
   }
+
   if (deleted) {
-    if (call_add) {
-      context->status.n_modified_tags++;
-    } else {
-      context->status.n_deleted_tags++;
-    }
+    context->status.n_modified_tags++;
   } else {
     if (added) {
       context->status.n_added_tags++;
@@ -292,8 +296,6 @@ census_context *census_context_create(const census_context *base,
     memset(context, 0, sizeof(census_context));
   } else {
     tag_set_copy(&context->tags[PROPAGATED_TAGS], &base->tags[PROPAGATED_TAGS]);
-    tag_set_copy(&context->tags[PROPAGATED_BINARY_TAGS],
-                 &base->tags[PROPAGATED_BINARY_TAGS]);
     tag_set_copy(&context->tags[LOCAL_TAGS], &base->tags[LOCAL_TAGS]);
     memset(&context->status, 0, sizeof(context->status));
   }
@@ -301,22 +303,29 @@ census_context *census_context_create(const census_context *base,
   // the context to add/replace/delete as required.
   for (int i = 0; i < ntags; i++) {
     const census_tag *tag = &tags[i];
-    size_t key_len = strlen(tag->key) + 1;
-    // ignore the tag if it is too long/short.
-    if (key_len != 1 && key_len <= CENSUS_MAX_TAG_KV_LEN &&
-        tag->value_len <= CENSUS_MAX_TAG_KV_LEN) {
-      context_modify_tag(context, tag, key_len);
-    } else {
+    size_t key_len = validate_tag(tag->key);
+    // ignore the tag if it is invalid or too short.
+    if (key_len <= 1) {
       context->status.n_invalid_tags++;
+    } else {
+      if (tag->value != NULL) {
+        size_t value_len = validate_tag(tag->value);
+        if (value_len != 0) {
+          context_modify_tag(context, tag, key_len, value_len);
+        } else {
+          context->status.n_invalid_tags++;
+        }
+      } else {
+        if (context_delete_tag(context, tag, key_len)) {
+          context->status.n_deleted_tags++;
+        }
+      }
     }
   }
   // Remove any deleted tags, update status if needed, and return.
   tag_set_flatten(&context->tags[PROPAGATED_TAGS]);
-  tag_set_flatten(&context->tags[PROPAGATED_BINARY_TAGS]);
   tag_set_flatten(&context->tags[LOCAL_TAGS]);
   context->status.n_propagated_tags = context->tags[PROPAGATED_TAGS].ntags;
-  context->status.n_propagated_binary_tags =
-      context->tags[PROPAGATED_BINARY_TAGS].ntags;
   context->status.n_local_tags = context->tags[LOCAL_TAGS].ntags;
   if (status) {
     *status = &context->status;
@@ -331,7 +340,6 @@ const census_context_status *census_context_get_status(
 
 void census_context_destroy(census_context *context) {
   gpr_free(context->tags[PROPAGATED_TAGS].kvm);
-  gpr_free(context->tags[PROPAGATED_BINARY_TAGS].kvm);
   gpr_free(context->tags[LOCAL_TAGS].kvm);
   gpr_free(context);
 }
@@ -343,9 +351,6 @@ void census_context_initialize_iterator(const census_context *context,
   if (context->tags[PROPAGATED_TAGS].ntags != 0) {
     iterator->base = PROPAGATED_TAGS;
     iterator->kvm = context->tags[PROPAGATED_TAGS].kvm;
-  } else if (context->tags[PROPAGATED_BINARY_TAGS].ntags != 0) {
-    iterator->base = PROPAGATED_BINARY_TAGS;
-    iterator->kvm = context->tags[PROPAGATED_BINARY_TAGS].kvm;
   } else if (context->tags[LOCAL_TAGS].ntags != 0) {
     iterator->base = LOCAL_TAGS;
     iterator->kvm = context->tags[LOCAL_TAGS].kvm;
@@ -363,7 +368,6 @@ int census_context_next_tag(census_context_iterator *iterator,
   iterator->kvm = decode_tag(&raw, iterator->kvm, 0);
   tag->key = raw.key;
   tag->value = raw.value;
-  tag->value_len = raw.value_len;
   tag->flags = raw.flags;
   if (++iterator->index == iterator->context->tags[iterator->base].ntags) {
     do {
@@ -388,7 +392,6 @@ static bool tag_set_get_tag(const struct tag_set *tags, const char *key,
     if (key_len == raw.key_len && memcmp(raw.key, key, key_len) == 0) {
       tag->key = raw.key;
       tag->value = raw.value;
-      tag->value_len = raw.value_len;
       tag->flags = raw.flags;
       return true;
     }
@@ -403,8 +406,6 @@ int census_context_get_tag(const census_context *context, const char *key,
     return 0;
   }
   if (tag_set_get_tag(&context->tags[PROPAGATED_TAGS], key, key_len, tag) ||
-      tag_set_get_tag(&context->tags[PROPAGATED_BINARY_TAGS], key, key_len,
-                      tag) ||
       tag_set_get_tag(&context->tags[LOCAL_TAGS], key, key_len, tag)) {
     return 1;
   }
@@ -447,21 +448,9 @@ static size_t tag_set_encode(const struct tag_set *tags, char *buffer,
   return ENCODED_HEADER_SIZE + tags->kvm_used;
 }
 
-char *census_context_encode(const census_context *context, char *buffer,
-                            size_t buf_size, size_t *print_buf_size,
-                            size_t *bin_buf_size) {
-  *print_buf_size =
-      tag_set_encode(&context->tags[PROPAGATED_TAGS], buffer, buf_size);
-  if (*print_buf_size == 0) {
-    return NULL;
-  }
-  char *b_buffer = buffer + *print_buf_size;
-  *bin_buf_size = tag_set_encode(&context->tags[PROPAGATED_BINARY_TAGS],
-                                 b_buffer, buf_size - *print_buf_size);
-  if (*bin_buf_size == 0) {
-    return NULL;
-  }
-  return b_buffer;
+size_t census_context_encode(const census_context *context, char *buffer,
+                             size_t buf_size) {
+  return tag_set_encode(&context->tags[PROPAGATED_TAGS], buffer, buf_size);
 }
 
 // Decode a tag set.
@@ -506,8 +495,7 @@ static void tag_set_decode(struct tag_set *tags, const char *buffer,
   }
 }
 
-census_context *census_context_decode(const char *buffer, size_t size,
-                                      const char *bin_buffer, size_t bin_size) {
+census_context *census_context_decode(const char *buffer, size_t size) {
   census_context *context = gpr_malloc(sizeof(census_context));
   memset(&context->tags[LOCAL_TAGS], 0, sizeof(struct tag_set));
   if (buffer == NULL) {
@@ -515,16 +503,7 @@ census_context *census_context_decode(const char *buffer, size_t size,
   } else {
     tag_set_decode(&context->tags[PROPAGATED_TAGS], buffer, size);
   }
-  if (bin_buffer == NULL) {
-    memset(&context->tags[PROPAGATED_BINARY_TAGS], 0, sizeof(struct tag_set));
-  } else {
-    tag_set_decode(&context->tags[PROPAGATED_BINARY_TAGS], bin_buffer,
-                   bin_size);
-  }
   memset(&context->status, 0, sizeof(context->status));
   context->status.n_propagated_tags = context->tags[PROPAGATED_TAGS].ntags;
-  context->status.n_propagated_binary_tags =
-      context->tags[PROPAGATED_BINARY_TAGS].ntags;
-  // TODO(aveitch): check that BINARY flag is correct for each type.
   return context;
 }
diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index eeac3c146c..d4ba950818 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -251,7 +251,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
 
   grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, true, NULL);
 
-  GPR_ASSERT(op->set_accept_stream == NULL);
+  GPR_ASSERT(op->set_accept_stream == false);
   if (op->bind_pollset != NULL) {
     grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties,
                                  op->bind_pollset);
diff --git a/src/core/channel/client_uchannel.c b/src/core/channel/client_uchannel.c
index b1e7155773..83fcc3a87f 100644
--- a/src/core/channel/client_uchannel.c
+++ b/src/core/channel/client_uchannel.c
@@ -107,7 +107,7 @@ static void cuc_start_transport_op(grpc_exec_ctx *exec_ctx,
 
   grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, true, NULL);
 
-  GPR_ASSERT(op->set_accept_stream == NULL);
+  GPR_ASSERT(op->set_accept_stream == false);
   GPR_ASSERT(op->bind_pollset == NULL);
 
   if (op->on_connectivity_state_change != NULL) {
diff --git a/src/core/channel/subchannel_call_holder.c b/src/core/channel/subchannel_call_holder.c
index 81297c8d44..8f46885a04 100644
--- a/src/core/channel/subchannel_call_holder.c
+++ b/src/core/channel/subchannel_call_holder.c
@@ -168,21 +168,23 @@ retry:
 
 static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
   grpc_subchannel_call_holder *holder = arg;
-  grpc_subchannel_call *call;
   gpr_mu_lock(&holder->mu);
   GPR_ASSERT(holder->creation_phase ==
              GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL);
-  call = GET_CALL(holder);
-  GPR_ASSERT(call == NULL || call == CANCELLED_CALL);
   holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
   if (holder->connected_subchannel == NULL) {
     fail_locked(exec_ctx, holder);
   } else {
-    gpr_atm_rel_store(
-        &holder->subchannel_call,
-        (gpr_atm)(uintptr_t)grpc_connected_subchannel_create_call(
-            exec_ctx, holder->connected_subchannel, holder->pollset));
-    retry_waiting_locked(exec_ctx, holder);
+    if (!gpr_atm_rel_cas(
+            &holder->subchannel_call, 0,
+            (gpr_atm)(uintptr_t)grpc_connected_subchannel_create_call(
+                exec_ctx, holder->connected_subchannel, holder->pollset))) {
+      GPR_ASSERT(gpr_atm_acq_load(&holder->subchannel_call) == 1);
+      /* if this cas fails, the call was cancelled before the pick completed */
+      fail_locked(exec_ctx, holder);
+    } else {
+      retry_waiting_locked(exec_ctx, holder);
+    }
   }
   gpr_mu_unlock(&holder->mu);
   GRPC_CALL_STACK_UNREF(exec_ctx, holder->owning_call, "pick_subchannel");
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index bec06bf414..d91dd116b8 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -395,7 +395,6 @@ void grpc_subchannel_notify_on_state_change(
     grpc_exec_ctx *exec_ctx, grpc_subchannel *c,
     grpc_pollset_set *interested_parties, grpc_connectivity_state *state,
     grpc_closure *notify) {
-  int do_connect = 0;
   external_state_watcher *w;
 
   if (state == NULL) {
@@ -425,17 +424,13 @@ void grpc_subchannel_notify_on_state_change(
     w->next->prev = w->prev->next = w;
     if (grpc_connectivity_state_notify_on_state_change(
             exec_ctx, &c->state_tracker, state, &w->closure)) {
-      do_connect = 1;
       c->connecting = 1;
       /* released by connection */
       GRPC_SUBCHANNEL_WEAK_REF(c, "connecting");
+      start_connect(exec_ctx, c);
     }
     gpr_mu_unlock(&c->mu);
   }
-
-  if (do_connect) {
-    start_connect(exec_ctx, c);
-  }
 }
 
 void grpc_connected_subchannel_process_transport_op(
@@ -635,11 +630,12 @@ static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, bool iomgr_success) {
   if (c->disconnected) {
     iomgr_success = 0;
   }
-  gpr_mu_unlock(&c->mu);
   if (iomgr_success) {
     update_reconnect_parameters(c);
     continue_connect(exec_ctx, c);
+    gpr_mu_unlock(&c->mu);
   } else {
+    gpr_mu_unlock(&c->mu);
     GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, c, "connecting");
   }
 }
diff --git a/src/core/iomgr/ev_poll_and_epoll_posix.c b/src/core/iomgr/ev_poll_and_epoll_posix.c
index 5ff02bb216..935b4a2874 100644
--- a/src/core/iomgr/ev_poll_and_epoll_posix.c
+++ b/src/core/iomgr/ev_poll_and_epoll_posix.c
@@ -1323,6 +1323,7 @@ static void multipoll_with_poll_pollset_maybe_work_and_unlock(
     } else {
       h->fds[fd_count++] = h->fds[i];
       watchers[pfd_count].fd = h->fds[i];
+      GRPC_FD_REF(watchers[pfd_count].fd, "multipoller_start");
       pfds[pfd_count].fd = h->fds[i]->fd;
       pfds[pfd_count].revents = 0;
       pfd_count++;
@@ -1336,8 +1337,10 @@ static void multipoll_with_poll_pollset_maybe_work_and_unlock(
   gpr_mu_unlock(&pollset->mu);
 
   for (i = 2; i < pfd_count; i++) {
-    pfds[i].events = (short)fd_begin_poll(watchers[i].fd, pollset, worker,
-                                          POLLIN, POLLOUT, &watchers[i]);
+    grpc_fd *fd = watchers[i].fd;
+    pfds[i].events = (short)grpc_fd_begin_poll(fd, pollset, worker, POLLIN,
+                                               POLLOUT, &watchers[i]);
+    GRPC_FD_UNREF(fd, "multipoller_start");
   }
 
   /* TODO(vpai): Consider first doing a 0 timeout poll here to avoid
diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c
index 4b5d8e9fab..3fe12ca76a 100644
--- a/src/core/iomgr/iomgr.c
+++ b/src/core/iomgr/iomgr.c
@@ -41,9 +41,11 @@
 #include <grpc/support/string_util.h>
 #include <grpc/support/sync.h>
 #include <grpc/support/thd.h>
+#include <grpc/support/useful.h>
 
 #include "src/core/iomgr/iomgr_internal.h"
 #include "src/core/iomgr/timer.h"
+#include "src/core/support/env.h"
 #include "src/core/support/string.h"
 
 static gpr_mu g_mu;
@@ -115,6 +117,9 @@ void grpc_iomgr_shutdown(void) {
                     "memory leaks are likely",
                     count_objects());
             dump_objects("LEAKED");
+            if (grpc_iomgr_abort_on_leaks()) {
+              abort();
+            }
           }
           break;
         }
@@ -152,3 +157,14 @@ void grpc_iomgr_unregister_object(grpc_iomgr_object *obj) {
   gpr_mu_unlock(&g_mu);
   gpr_free(obj->name);
 }
+
+bool grpc_iomgr_abort_on_leaks(void) {
+  char *env = gpr_getenv("GRPC_ABORT_ON_LEAKS");
+  if (env == NULL) return false;
+  static const char *truthy[] = {"yes",  "Yes",  "YES", "true",
+                                 "True", "TRUE", "1"};
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(truthy); i++) {
+    if (0 == strcmp(env, truthy[i])) return true;
+  }
+  return false;
+}
diff --git a/src/core/iomgr/iomgr_internal.h b/src/core/iomgr/iomgr_internal.h
index ccf0dee3a4..ac14914b76 100644
--- a/src/core/iomgr/iomgr_internal.h
+++ b/src/core/iomgr/iomgr_internal.h
@@ -34,7 +34,8 @@
 #ifndef GRPC_INTERNAL_CORE_IOMGR_IOMGR_INTERNAL_H
 #define GRPC_INTERNAL_CORE_IOMGR_IOMGR_INTERNAL_H
 
-#include <grpc/support/sync.h>
+#include <stdbool.h>
+
 #include "src/core/iomgr/iomgr.h"
 
 typedef struct grpc_iomgr_object {
@@ -52,4 +53,6 @@ void grpc_iomgr_platform_flush(void);
 /** tear down all platform specific global iomgr structures */
 void grpc_iomgr_platform_shutdown(void);
 
+bool grpc_iomgr_abort_on_leaks(void);
+
 #endif /* GRPC_INTERNAL_CORE_IOMGR_IOMGR_INTERNAL_H */
diff --git a/src/core/support/sync.c b/src/core/support/sync.c
index d368422d9e..69e3e39c5c 100644
--- a/src/core/support/sync.c
+++ b/src/core/support/sync.c
@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -98,6 +98,11 @@ void gpr_ref_init(gpr_refcount *r, int n) { gpr_atm_rel_store(&r->count, n); }
 
 void gpr_ref(gpr_refcount *r) { gpr_atm_no_barrier_fetch_add(&r->count, 1); }
 
+void gpr_ref_non_zero(gpr_refcount *r) {
+  gpr_atm prior = gpr_atm_no_barrier_fetch_add(&r->count, 1);
+  GPR_ASSERT(prior > 0);
+}
+
 void gpr_refn(gpr_refcount *r, int n) {
   gpr_atm_no_barrier_fetch_add(&r->count, n);
 }
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index fb5e0d4b9e..5b13d4ba52 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -407,8 +407,15 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) {
   maybe_finish_shutdown(exec_ctx, chand->server);
   chand->finish_destroy_channel_closure.cb = finish_destroy_channel;
   chand->finish_destroy_channel_closure.cb_arg = chand;
-  grpc_exec_ctx_enqueue(exec_ctx, &chand->finish_destroy_channel_closure, true,
-                        NULL);
+
+  grpc_transport_op op;
+  memset(&op, 0, sizeof(op));
+  op.set_accept_stream = true;
+  op.on_consumed = &chand->finish_destroy_channel_closure;
+  grpc_channel_next_op(exec_ctx,
+                       grpc_channel_stack_element(
+                           grpc_channel_get_channel_stack(chand->channel), 0),
+                       &op);
 }
 
 static void finish_start_new_rpc(grpc_exec_ctx *exec_ctx, grpc_server *server,
@@ -971,7 +978,8 @@ void grpc_server_setup_transport(grpc_exec_ctx *exec_ctx, grpc_server *s,
 
   GRPC_CHANNEL_INTERNAL_REF(channel, "connectivity");
   memset(&op, 0, sizeof(op));
-  op.set_accept_stream = accept_stream;
+  op.set_accept_stream = true;
+  op.set_accept_stream_fn = accept_stream;
   op.set_accept_stream_user_data = chand;
   op.on_connectivity_state_change = &chand->channel_connectivity_changed;
   op.connectivity_state = &chand->connectivity_state;
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index d76d31be23..b720d1ab3e 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -358,6 +358,9 @@ struct grpc_chttp2_transport {
     /** connectivity tracking */
     grpc_connectivity_state_tracker state_tracker;
   } channel_callback;
+
+  /** Transport op to be applied post-parsing */
+  grpc_transport_op *post_parsing_op;
 };
 
 typedef struct {
@@ -417,7 +420,7 @@ typedef struct {
   /** HTTP2 stream id for this stream, or zero if one has not been assigned */
   uint32_t id;
   uint8_t fetching;
-  uint8_t sent_initial_metadata;
+  bool sent_initial_metadata;
   uint8_t sent_message;
   uint8_t sent_trailing_metadata;
   uint8_t read_closed;
@@ -509,7 +512,7 @@ void grpc_chttp2_publish_reads(grpc_exec_ctx *exec_ctx,
                                grpc_chttp2_transport_global *global,
                                grpc_chttp2_transport_parsing *parsing);
 
-void grpc_chttp2_list_add_writable_stream(
+bool grpc_chttp2_list_add_writable_stream(
     grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_stream_global *stream_global);
 /** Get a writable stream
@@ -519,14 +522,13 @@ int grpc_chttp2_list_pop_writable_stream(
     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(
+bool grpc_chttp2_list_remove_writable_stream(
     grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global);
+    grpc_chttp2_stream_global *stream_global) GRPC_MUST_USE_RESULT;
 
-/* returns 1 if stream added, 0 if it was already present */
-int grpc_chttp2_list_add_writing_stream(
+void grpc_chttp2_list_add_writing_stream(
     grpc_chttp2_transport_writing *transport_writing,
-    grpc_chttp2_stream_writing *stream_writing) GRPC_MUST_USE_RESULT;
+    grpc_chttp2_stream_writing *stream_writing);
 int grpc_chttp2_list_have_writing_streams(
     grpc_chttp2_transport_writing *transport_writing);
 int grpc_chttp2_list_pop_writing_stream(
@@ -770,4 +772,9 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
                           grpc_chttp2_transport_parsing *parsing,
                           const uint8_t *opaque_8bytes);
 
+/** add a ref to the stream and add it to the writable list;
+    ref will be dropped in writing.c */
+void grpc_chttp2_become_writable(grpc_chttp2_transport_global *transport_global,
+                                 grpc_chttp2_stream_global *stream_global);
+
 #endif
diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c
index 8fdebd7f13..0516f39fa9 100644
--- a/src/core/transport/chttp2/parsing.c
+++ b/src/core/transport/chttp2/parsing.c
@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -149,7 +149,7 @@ void grpc_chttp2_publish_reads(
   if (was_zero && !is_zero) {
     while (grpc_chttp2_list_pop_stalled_by_transport(transport_global,
                                                      &stream_global)) {
-      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+      grpc_chttp2_become_writable(transport_global, stream_global);
     }
   }
 
@@ -178,7 +178,7 @@ void grpc_chttp2_publish_reads(
                                  outgoing_window);
     is_zero = stream_global->outgoing_window <= 0;
     if (was_zero && !is_zero) {
-      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+      grpc_chttp2_become_writable(transport_global, stream_global);
     }
 
     stream_global->max_recv_bytes -= (uint32_t)GPR_MIN(
diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c
index b284c78818..60fe735cfc 100644
--- a/src/core/transport/chttp2/stream_lists.c
+++ b/src/core/transport/chttp2/stream_lists.c
@@ -100,11 +100,14 @@ static void stream_list_remove(grpc_chttp2_transport *t, grpc_chttp2_stream *s,
   }
 }
 
-static void stream_list_maybe_remove(grpc_chttp2_transport *t,
+static bool stream_list_maybe_remove(grpc_chttp2_transport *t,
                                      grpc_chttp2_stream *s,
                                      grpc_chttp2_stream_list_id id) {
   if (s->included[id]) {
     stream_list_remove(t, s, id);
+    return true;
+  } else {
+    return false;
   }
 }
 
@@ -125,23 +128,24 @@ static void stream_list_add_tail(grpc_chttp2_transport *t,
   s->included[id] = 1;
 }
 
-static int stream_list_add(grpc_chttp2_transport *t, grpc_chttp2_stream *s,
-                           grpc_chttp2_stream_list_id id) {
+static bool stream_list_add(grpc_chttp2_transport *t, grpc_chttp2_stream *s,
+                            grpc_chttp2_stream_list_id id) {
   if (s->included[id]) {
-    return 0;
+    return false;
   }
   stream_list_add_tail(t, s, id);
-  return 1;
+  return true;
 }
 
 /* wrappers for specializations */
 
-void grpc_chttp2_list_add_writable_stream(
+bool grpc_chttp2_list_add_writable_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);
+  return stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global),
+                         STREAM_FROM_GLOBAL(stream_global),
+                         GRPC_CHTTP2_LIST_WRITABLE);
 }
 
 int grpc_chttp2_list_pop_writable_stream(
@@ -159,20 +163,20 @@ int grpc_chttp2_list_pop_writable_stream(
   return r;
 }
 
-void grpc_chttp2_list_remove_writable_stream(
+bool 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);
+  return stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global),
+                                  STREAM_FROM_GLOBAL(stream_global),
+                                  GRPC_CHTTP2_LIST_WRITABLE);
 }
 
-int grpc_chttp2_list_add_writing_stream(
+void grpc_chttp2_list_add_writing_stream(
     grpc_chttp2_transport_writing *transport_writing,
     grpc_chttp2_stream_writing *stream_writing) {
-  return stream_list_add(TRANSPORT_FROM_WRITING(transport_writing),
-                         STREAM_FROM_WRITING(stream_writing),
-                         GRPC_CHTTP2_LIST_WRITING);
+  GPR_ASSERT(stream_list_add(TRANSPORT_FROM_WRITING(transport_writing),
+                             STREAM_FROM_WRITING(stream_writing),
+                             GRPC_CHTTP2_LIST_WRITING));
 }
 
 int grpc_chttp2_list_have_writing_streams(
@@ -332,7 +336,7 @@ void grpc_chttp2_list_flush_writing_stalled_by_transport(
   while (stream_list_pop(transport, &stream,
                          GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT)) {
     if (is_window_available) {
-      grpc_chttp2_list_add_writable_stream(&transport->global, &stream->global);
+      grpc_chttp2_become_writable(&transport->global, &stream->global);
     } else {
       grpc_chttp2_list_add_stalled_by_transport(transport_writing,
                                                 &stream->writing);
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index 356fd8174a..107725cbc7 100644
--- a/src/core/transport/chttp2/writing.c
+++ b/src/core/transport/chttp2/writing.c
@@ -83,7 +83,8 @@ int grpc_chttp2_unlocking_check_writes(
      (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)) {
-    uint8_t sent_initial_metadata;
+    bool sent_initial_metadata = stream_writing->sent_initial_metadata;
+    bool become_writable = false;
 
     stream_writing->id = stream_global->id;
     stream_writing->read_closed = stream_global->read_closed;
@@ -92,16 +93,12 @@ int grpc_chttp2_unlocking_check_writes(
                                  outgoing_window, stream_global,
                                  outgoing_window);
 
-    sent_initial_metadata = stream_writing->sent_initial_metadata;
     if (!sent_initial_metadata && stream_global->send_initial_metadata) {
       stream_writing->send_initial_metadata =
           stream_global->send_initial_metadata;
       stream_global->send_initial_metadata = NULL;
-      if (grpc_chttp2_list_add_writing_stream(transport_writing,
-                                              stream_writing)) {
-        GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing");
-      }
-      sent_initial_metadata = 1;
+      become_writable = true;
+      sent_initial_metadata = true;
     }
     if (sent_initial_metadata) {
       if (stream_global->send_message != NULL) {
@@ -128,10 +125,7 @@ int grpc_chttp2_unlocking_check_writes(
            stream_writing->flow_controlled_buffer.length > 0) &&
           stream_writing->outgoing_window > 0) {
         if (transport_writing->outgoing_window > 0) {
-          if (grpc_chttp2_list_add_writing_stream(transport_writing,
-                                                  stream_writing)) {
-            GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing");
-          }
+          become_writable = true;
         } else {
           grpc_chttp2_list_add_stalled_by_transport(transport_writing,
                                                     stream_writing);
@@ -141,10 +135,7 @@ int grpc_chttp2_unlocking_check_writes(
         stream_writing->send_trailing_metadata =
             stream_global->send_trailing_metadata;
         stream_global->send_trailing_metadata = NULL;
-        if (grpc_chttp2_list_add_writing_stream(transport_writing,
-                                                stream_writing)) {
-          GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing");
-        }
+        become_writable = true;
       }
     }
 
@@ -153,10 +144,13 @@ int grpc_chttp2_unlocking_check_writes(
       GRPC_CHTTP2_FLOW_MOVE_STREAM("write", transport_global, stream_writing,
                                    announce_window, stream_global,
                                    unannounced_incoming_window_for_writing);
-      if (grpc_chttp2_list_add_writing_stream(transport_writing,
-                                              stream_writing)) {
-        GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing");
-      }
+      become_writable = true;
+    }
+
+    if (become_writable) {
+      grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
+    } else {
+      GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2_writing");
     }
   }
 
@@ -310,10 +304,7 @@ static void finalize_outbuf(grpc_exec_ctx *exec_ctx,
          (stream_writing->send_message && !stream_writing->fetching)) &&
         stream_writing->outgoing_window > 0) {
       if (transport_writing->outgoing_window > 0) {
-        if (grpc_chttp2_list_add_writing_stream(transport_writing,
-                                                stream_writing)) {
-          /* do nothing - already reffed */
-        }
+        grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
       } else {
         grpc_chttp2_list_add_writing_stalled_by_transport(transport_writing,
                                                           stream_writing);
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index b9f511e946..19265252ca 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -142,7 +142,7 @@ static void incoming_byte_stream_update_flow_control(
 static void fail_pending_writes(grpc_exec_ctx *exec_ctx,
                                 grpc_chttp2_stream_global *stream_global);
 
-/*
+/*******************************************************************************
  * CONSTRUCTION/DESTRUCTION/REFCOUNTING
  */
 
@@ -432,6 +432,14 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx,
     if (t->ep) {
       allow_endpoint_shutdown_locked(exec_ctx, t);
     }
+
+    /* flush writable stream list to avoid dangling references */
+    grpc_chttp2_stream_global *stream_global;
+    grpc_chttp2_stream_writing *stream_writing;
+    while (grpc_chttp2_list_pop_writable_stream(
+        &t->global, &t->writing, &stream_global, &stream_writing)) {
+      GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2_writing");
+    }
   }
 }
 
@@ -521,7 +529,6 @@ static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
                                            s->global.id) == NULL);
   }
 
-  grpc_chttp2_list_remove_writable_stream(&t->global, &s->global);
   grpc_chttp2_list_remove_unannounced_incoming_window_available(&t->global,
                                                                 &s->global);
   grpc_chttp2_list_remove_stalled_by_transport(&t->global, &s->global);
@@ -583,7 +590,7 @@ grpc_chttp2_stream_parsing *grpc_chttp2_parsing_accept_stream(
   return &accepting->parsing;
 }
 
-/*
+/*******************************************************************************
  * LOCK MANAGEMENT
  */
 
@@ -611,10 +618,18 @@ static void unlock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
   GPR_TIMER_END("unlock", 0);
 }
 
-/*
+/*******************************************************************************
  * OUTPUT PROCESSING
  */
 
+void grpc_chttp2_become_writable(grpc_chttp2_transport_global *transport_global,
+                                 grpc_chttp2_stream_global *stream_global) {
+  if (!TRANSPORT_FROM_GLOBAL(transport_global)->closed &&
+      grpc_chttp2_list_add_writable_stream(transport_global, stream_global)) {
+    GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing");
+  }
+}
+
 static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id,
                          uint32_t value) {
   const grpc_chttp2_setting_parameters *sp =
@@ -732,7 +747,7 @@ static void maybe_start_some_streams(
         stream_global->id, STREAM_FROM_GLOBAL(stream_global));
     stream_global->in_stream_map = 1;
     transport_global->concurrent_stream_count++;
-    grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+    grpc_chttp2_become_writable(transport_global, stream_global);
   }
   /* cancel out streams that will never be started */
   while (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID &&
@@ -821,7 +836,7 @@ static void perform_stream_op_locked(
         maybe_start_some_streams(exec_ctx, transport_global);
       } else {
         GPR_ASSERT(stream_global->id != 0);
-        grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+        grpc_chttp2_become_writable(transport_global, stream_global);
       }
     } else {
       grpc_chttp2_complete_closure_step(
@@ -838,7 +853,7 @@ static void perform_stream_op_locked(
           exec_ctx, &stream_global->send_message_finished, 0);
     } else if (stream_global->id != 0) {
       stream_global->send_message = op->send_message;
-      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+      grpc_chttp2_become_writable(transport_global, stream_global);
     }
   }
 
@@ -858,7 +873,7 @@ static void perform_stream_op_locked(
     } else if (stream_global->id != 0) {
       /* TODO(ctiller): check if there's flow control for any outstanding
          bytes before going writable */
-      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+      grpc_chttp2_become_writable(transport_global, stream_global);
     }
   }
 
@@ -944,12 +959,10 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
   unlock(exec_ctx, t);
 }
 
-static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
-                                 grpc_transport_op *op) {
-  grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
-  int close_transport = 0;
-
-  lock(t);
+static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx,
+                                        grpc_chttp2_transport *t,
+                                        grpc_transport_op *op) {
+  bool close_transport = false;
 
   grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, true, NULL);
 
@@ -968,8 +981,8 @@ static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
     close_transport = !grpc_chttp2_has_streams(t);
   }
 
-  if (op->set_accept_stream != NULL) {
-    t->channel_callback.accept_stream = op->set_accept_stream;
+  if (op->set_accept_stream) {
+    t->channel_callback.accept_stream = op->set_accept_stream_fn;
     t->channel_callback.accept_stream_user_data =
         op->set_accept_stream_user_data;
   }
@@ -990,16 +1003,31 @@ static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
     close_transport_locked(exec_ctx, t);
   }
 
-  unlock(exec_ctx, t);
-
   if (close_transport) {
-    lock(t);
     close_transport_locked(exec_ctx, t);
-    unlock(exec_ctx, t);
   }
 }
 
-/*
+static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
+                                 grpc_transport_op *op) {
+  grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
+
+  lock(t);
+
+  /* If there's a set_accept_stream ensure that we're not parsing
+     to avoid changing things out from underneath */
+  if (t->parsing_active && op->set_accept_stream) {
+    GPR_ASSERT(t->post_parsing_op == NULL);
+    t->post_parsing_op = gpr_malloc(sizeof(*op));
+    memcpy(t->post_parsing_op, op, sizeof(*op));
+  } else {
+    perform_transport_op_locked(exec_ctx, t, op);
+  }
+
+  unlock(exec_ctx, t);
+}
+
+/*******************************************************************************
  * INPUT PROCESSING
  */
 
@@ -1064,7 +1092,6 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   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) {
@@ -1080,6 +1107,9 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   if (grpc_chttp2_unregister_stream(t, s) && t->global.sent_goaway) {
     close_transport_locked(exec_ctx, t);
   }
+  if (grpc_chttp2_list_remove_writable_stream(&t->global, &s->global)) {
+    GRPC_CHTTP2_STREAM_UNREF(exec_ctx, &s->global, "chttp2_writing");
+  }
 
   new_stream_count = grpc_chttp2_stream_map_size(&t->parsing_stream_map) +
                      grpc_chttp2_stream_map_size(&t->new_stream_map);
@@ -1331,7 +1361,7 @@ static void update_global_window(void *args, uint32_t id, void *stream) {
   is_zero = stream_global->outgoing_window <= 0;
 
   if (was_zero && !is_zero) {
-    grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+    grpc_chttp2_become_writable(transport_global, stream_global);
   }
 }
 
@@ -1392,6 +1422,13 @@ static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, bool success) {
     /* handle higher level things */
     grpc_chttp2_publish_reads(exec_ctx, transport_global, transport_parsing);
     t->parsing_active = 0;
+    /* handle delayed transport ops (if there is one) */
+    if (t->post_parsing_op) {
+      grpc_transport_op *op = t->post_parsing_op;
+      t->post_parsing_op = NULL;
+      perform_transport_op_locked(exec_ctx, t, op);
+      gpr_free(op);
+    }
     /* if a stream is in the stream map, and gets cancelled, we need to ensure
      * we are not parsing before continuing the cancellation to keep things in
      * a sane state */
@@ -1426,7 +1463,7 @@ static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, bool success) {
   GPR_TIMER_END("recv_data", 0);
 }
 
-/*
+/*******************************************************************************
  * CALLBACK LOOP
  */
 
@@ -1440,7 +1477,7 @@ static void connectivity_state_set(
                               state, reason);
 }
 
-/*
+/*******************************************************************************
  * POLLSET STUFF
  */
 
@@ -1468,7 +1505,7 @@ static void set_pollset(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
   unlock(exec_ctx, t);
 }
 
-/*
+/*******************************************************************************
  * BYTE STREAM
  */
 
@@ -1508,7 +1545,7 @@ static void incoming_byte_stream_update_flow_control(
                                    add_max_recv_bytes);
     grpc_chttp2_list_add_unannounced_incoming_window_available(transport_global,
                                                                stream_global);
-    grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+    grpc_chttp2_become_writable(transport_global, stream_global);
   }
 }
 
@@ -1623,7 +1660,7 @@ grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
   return incoming_byte_stream;
 }
 
-/*
+/*******************************************************************************
  * TRACING
  */
 
@@ -1709,7 +1746,7 @@ void grpc_chttp2_flowctl_trace(const char *file, int line, const char *phase,
   gpr_free(prefix);
 }
 
-/*
+/*******************************************************************************
  * INTEGRATION GLUE
  */
 
diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c
index 14912af7df..807ae071a3 100644
--- a/src/core/transport/metadata.c
+++ b/src/core/transport/metadata.c
@@ -43,11 +43,13 @@
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 #include <grpc/support/time.h>
+
 #include "src/core/profiling/timers.h"
 #include "src/core/support/murmur_hash.h"
 #include "src/core/support/string.h"
 #include "src/core/transport/chttp2/bin_encoder.h"
 #include "src/core/transport/static_metadata.h"
+#include "src/core/iomgr/iomgr_internal.h"
 
 /* There are two kinds of mdelem and mdstr instances.
  * Static instances are declared in static_metadata.{h,c} and
@@ -227,6 +229,9 @@ void grpc_mdctx_global_shutdown(void) {
     if (shard->count != 0) {
       gpr_log(GPR_DEBUG, "WARNING: %d metadata elements were leaked",
               shard->count);
+      if (grpc_iomgr_abort_on_leaks()) {
+        abort();
+      }
     }
     gpr_free(shard->elems);
   }
@@ -237,6 +242,9 @@ void grpc_mdctx_global_shutdown(void) {
     if (shard->count != 0) {
       gpr_log(GPR_DEBUG, "WARNING: %d metadata strings were leaked",
               shard->count);
+      if (grpc_iomgr_abort_on_leaks()) {
+        abort();
+      }
     }
     gpr_free(shard->strs);
   }
diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c
index 6e154b629a..3b555fa933 100644
--- a/src/core/transport/transport.c
+++ b/src/core/transport/transport.c
@@ -45,7 +45,7 @@ void grpc_stream_ref(grpc_stream_refcount *refcount, const char *reason) {
 #else
 void grpc_stream_ref(grpc_stream_refcount *refcount) {
 #endif
-  gpr_ref(&refcount->refs);
+  gpr_ref_non_zero(&refcount->refs);
 }
 
 #ifdef GRPC_STREAM_REFCOUNT_DEBUG
diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h
index 8902c5d2f6..ed6e121c9c 100644
--- a/src/core/transport/transport.h
+++ b/src/core/transport/transport.h
@@ -123,7 +123,7 @@ typedef struct grpc_transport_stream_op {
 
 /** Transport op: a set of operations to perform on a transport as a whole */
 typedef struct grpc_transport_op {
-  /** called when processing of this op is done */
+  /** Called when processing of this op is done. */
   grpc_closure *on_consumed;
   /** connectivity monitoring - set connectivity_state to NULL to unsubscribe */
   grpc_closure *on_connectivity_state_change;
@@ -138,9 +138,13 @@ typedef struct grpc_transport_op {
   grpc_status_code goaway_status;
   gpr_slice *goaway_message;
   /** set the callback for accepting new streams;
-      this is a permanent callback, unlike the other one-shot closures */
-  void (*set_accept_stream)(grpc_exec_ctx *exec_ctx, void *user_data,
-                            grpc_transport *transport, const void *server_data);
+      this is a permanent callback, unlike the other one-shot closures.
+      If true, the callback is set to set_accept_stream_fn, with its
+      user_data argument set to set_accept_stream_user_data */
+  bool set_accept_stream;
+  void (*set_accept_stream_fn)(grpc_exec_ctx *exec_ctx, void *user_data,
+                               grpc_transport *transport,
+                               const void *server_data);
   void *set_accept_stream_user_data;
   /** add this transport to a pollset */
   grpc_pollset *bind_pollset;