Merge github.com:grpc/grpc into filter-selection

8 files changed, 135 insertions, 84 deletions

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 c49e1409a4..620c57f3bc 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;