Progress towards making grpc_slice_unref_internal take an exec_ctx

10 files changed, 57 insertions, 41 deletions

diff --git a/src/core/lib/channel/channel_args.c b/src/core/lib/channel/channel_args.c
index cfc072c0b5..c956b0febc 100644
--- a/src/core/lib/channel/channel_args.c
+++ b/src/core/lib/channel/channel_args.c
@@ -184,7 +184,7 @@ grpc_channel_args *grpc_channel_args_normalize(const grpc_channel_args *a) {
   return b;
 }
 
-void grpc_channel_args_destroy(grpc_channel_args *a) {
+void grpc_channel_args_destroy(grpc_exec_ctx *exec_ctx, grpc_channel_args *a) {
   size_t i;
   if (!a) return;
   for (i = 0; i < a->num_args; i++) {
@@ -195,7 +195,8 @@ void grpc_channel_args_destroy(grpc_channel_args *a) {
       case GRPC_ARG_INTEGER:
         break;
       case GRPC_ARG_POINTER:
-        a->args[i].value.pointer.vtable->destroy(a->args[i].value.pointer.p);
+        a->args[i].value.pointer.vtable->destroy(exec_ctx,
+                                                 a->args[i].value.pointer.p);
         break;
     }
     gpr_free(a->args[i].key);
@@ -249,7 +250,8 @@ static int find_compression_algorithm_states_bitset(const grpc_channel_args *a,
 }
 
 grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
-    grpc_channel_args **a, grpc_compression_algorithm algorithm, int state) {
+    grpc_exec_ctx *exec_ctx, grpc_channel_args **a,
+    grpc_compression_algorithm algorithm, int state) {
   int *states_arg = NULL;
   grpc_channel_args *result = *a;
   const int states_arg_found =
@@ -282,7 +284,7 @@ grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
       GPR_BITCLEAR((unsigned *)&tmp.value.integer, algorithm);
     }
     result = grpc_channel_args_copy_and_add(*a, &tmp, 1);
-    grpc_channel_args_destroy(*a);
+    grpc_channel_args_destroy(exec_ctx, *a);
     *a = result;
   }
   return result;
diff --git a/src/core/lib/channel/channel_args.h b/src/core/lib/channel/channel_args.h
index 1e05303471..5933133413 100644
--- a/src/core/lib/channel/channel_args.h
+++ b/src/core/lib/channel/channel_args.h
@@ -67,7 +67,7 @@ grpc_channel_args *grpc_channel_args_merge(const grpc_channel_args *a,
                                            const grpc_channel_args *b);
 
 /** Destroy arguments created by \a grpc_channel_args_copy */
-void grpc_channel_args_destroy(grpc_channel_args *a);
+void grpc_channel_args_destroy(grpc_exec_ctx *exec_ctx, grpc_channel_args *a);
 
 /** Returns the compression algorithm set in \a a. */
 grpc_compression_algorithm grpc_channel_args_get_compression_algorithm(
@@ -87,7 +87,8 @@ grpc_channel_args *grpc_channel_args_set_compression_algorithm(
  * modified to point to the returned instance (which may be different from the
  * input value of \a a). */
 grpc_channel_args *grpc_channel_args_compression_algorithm_set_state(
-    grpc_channel_args **a, grpc_compression_algorithm algorithm, int enabled);
+    grpc_exec_ctx *exec_ctx, grpc_channel_args **a,
+    grpc_compression_algorithm algorithm, int enabled);
 
 /** Returns the bitset representing the support state (true for enabled, false
  * for disabled) for compression algorithms.
diff --git a/src/core/lib/channel/channel_stack.c b/src/core/lib/channel/channel_stack.c
index 9da81959e7..c32e7e6277 100644
--- a/src/core/lib/channel/channel_stack.c
+++ b/src/core/lib/channel/channel_stack.c
@@ -292,7 +292,7 @@ void grpc_call_element_send_cancel_with_message(grpc_exec_ctx *exec_ctx,
   grpc_transport_stream_op *op = gpr_malloc(sizeof(*op));
   memset(op, 0, sizeof(*op));
   op->on_complete = grpc_closure_create(destroy_op, op);
-  grpc_transport_stream_op_add_cancellation_with_message(op, status,
+  grpc_transport_stream_op_add_cancellation_with_message(exec_ctx, op, status,
                                                          optional_message);
   elem->filter->start_transport_stream_op(exec_ctx, elem, op);
 }
@@ -304,6 +304,6 @@ void grpc_call_element_send_close_with_message(grpc_exec_ctx *exec_ctx,
   grpc_transport_stream_op *op = gpr_malloc(sizeof(*op));
   memset(op, 0, sizeof(*op));
   op->on_complete = grpc_closure_create(destroy_op, op);
-  grpc_transport_stream_op_add_close(op, status, optional_message);
+  grpc_transport_stream_op_add_close(exec_ctx, op, status, optional_message);
   elem->filter->start_transport_stream_op(exec_ctx, elem, op);
 }
diff --git a/src/core/lib/channel/channel_stack_builder.c b/src/core/lib/channel/channel_stack_builder.c
index eda4968f48..366bd0de29 100644
--- a/src/core/lib/channel/channel_stack_builder.c
+++ b/src/core/lib/channel/channel_stack_builder.c
@@ -138,9 +138,10 @@ void grpc_channel_stack_builder_set_name(grpc_channel_stack_builder *builder,
 }
 
 void grpc_channel_stack_builder_set_channel_arguments(
-    grpc_channel_stack_builder *builder, const grpc_channel_args *args) {
+    grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder,
+    const grpc_channel_args *args) {
   if (builder->args != NULL) {
-    grpc_channel_args_destroy(builder->args);
+    grpc_channel_args_destroy(exec_ctx, builder->args);
   }
   builder->args = grpc_channel_args_copy(args);
 }
@@ -213,7 +214,8 @@ bool grpc_channel_stack_builder_add_filter_after(
   return true;
 }
 
-void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder) {
+void grpc_channel_stack_builder_destroy(grpc_exec_ctx *exec_ctx,
+                                        grpc_channel_stack_builder *builder) {
   filter_node *p = builder->begin.next;
   while (p != &builder->end) {
     filter_node *next = p->next;
@@ -221,7 +223,7 @@ void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder) {
     p = next;
   }
   if (builder->args != NULL) {
-    grpc_channel_args_destroy(builder->args);
+    grpc_channel_args_destroy(exec_ctx, builder->args);
   }
   gpr_free(builder->target);
   gpr_free(builder);
@@ -270,7 +272,7 @@ void *grpc_channel_stack_builder_finish(grpc_exec_ctx *exec_ctx,
     i++;
   }
 
-  grpc_channel_stack_builder_destroy(builder);
+  grpc_channel_stack_builder_destroy(exec_ctx, builder);
   gpr_free((grpc_channel_filter **)filters);
 
   return result;
diff --git a/src/core/lib/channel/channel_stack_builder.h b/src/core/lib/channel/channel_stack_builder.h
index 4a00f7bfdb..3532819a0c 100644
--- a/src/core/lib/channel/channel_stack_builder.h
+++ b/src/core/lib/channel/channel_stack_builder.h
@@ -73,7 +73,8 @@ grpc_transport *grpc_channel_stack_builder_get_transport(
 
 /// Set channel arguments: copies args
 void grpc_channel_stack_builder_set_channel_arguments(
-    grpc_channel_stack_builder *builder, const grpc_channel_args *args);
+    grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder,
+    const grpc_channel_args *args);
 
 /// Return a borrowed pointer to the channel arguments
 const grpc_channel_args *grpc_channel_stack_builder_get_channel_arguments(
@@ -158,7 +159,8 @@ void *grpc_channel_stack_builder_finish(grpc_exec_ctx *exec_ctx,
                                         void *destroy_arg);
 
 /// Destroy the builder without creating a channel stack
-void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder);
+void grpc_channel_stack_builder_destroy(grpc_exec_ctx *exec_ctx,
+                                        grpc_channel_stack_builder *builder);
 
 extern int grpc_trace_channel_stack_builder;
 
diff --git a/src/core/lib/channel/compress_filter.c b/src/core/lib/channel/compress_filter.c
index de71bcc22b..9cb52627ce 100644
--- a/src/core/lib/channel/compress_filter.c
+++ b/src/core/lib/channel/compress_filter.c
@@ -44,6 +44,7 @@
 #include "src/core/lib/compression/algorithm_metadata.h"
 #include "src/core/lib/compression/message_compress.h"
 #include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/transport/static_metadata.h"
 
@@ -126,12 +127,14 @@ static int skip_compression(grpc_call_element *elem) {
 
 /** Filter initial metadata */
 static void process_send_initial_metadata(
-    grpc_call_element *elem, grpc_metadata_batch *initial_metadata) {
+    grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+    grpc_metadata_batch *initial_metadata) {
   call_data *calld = elem->call_data;
   channel_data *channeld = elem->channel_data;
   /* Parse incoming request for compression. If any, it'll be available
    * at calld->compression_algorithm */
-  grpc_metadata_batch_filter(initial_metadata, compression_md_filter, elem);
+  grpc_metadata_batch_filter(exec_ctx, initial_metadata, compression_md_filter,
+                             elem);
   if (!calld->has_compression_algorithm) {
     /* If no algorithm was found in the metadata and we aren't
      * exceptionally skipping compression, fall back to the channel
@@ -157,7 +160,7 @@ static void continue_send_message(grpc_exec_ctx *exec_ctx,
 static void send_done(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) {
   grpc_call_element *elem = elemp;
   call_data *calld = elem->call_data;
-  grpc_slice_buffer_reset_and_unref(&calld->slices);
+  grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &calld->slices);
   calld->post_send->cb(exec_ctx, calld->post_send->cb_arg, error);
 }
 
@@ -167,8 +170,8 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx,
   int did_compress;
   grpc_slice_buffer tmp;
   grpc_slice_buffer_init(&tmp);
-  did_compress =
-      grpc_msg_compress(calld->compression_algorithm, &calld->slices, &tmp);
+  did_compress = grpc_msg_compress(exec_ctx, calld->compression_algorithm,
+                                   &calld->slices, &tmp);
   if (did_compress) {
     if (grpc_compression_trace) {
       char *algo_name;
@@ -195,7 +198,7 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx,
     }
   }
 
-  grpc_slice_buffer_destroy(&tmp);
+  grpc_slice_buffer_destroy_internal(exec_ctx, &tmp);
 
   grpc_slice_buffer_stream_init(&calld->replacement_stream, &calld->slices,
                                 calld->send_flags);
@@ -239,7 +242,7 @@ static void compress_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
   GPR_TIMER_BEGIN("compress_start_transport_stream_op", 0);
 
   if (op->send_initial_metadata) {
-    process_send_initial_metadata(elem, op->send_initial_metadata);
+    process_send_initial_metadata(exec_ctx, elem, op->send_initial_metadata);
   }
   if (op->send_message != NULL && !skip_compression(elem) &&
       0 == (op->send_message->flags & GRPC_WRITE_NO_COMPRESS)) {
@@ -277,7 +280,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               void *ignored) {
   /* grab pointers to our data from the call element */
   call_data *calld = elem->call_data;
-  grpc_slice_buffer_destroy(&calld->slices);
+  grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices);
 }
 
 /* Constructor for channel_data */
diff --git a/src/core/lib/channel/deadline_filter.c b/src/core/lib/channel/deadline_filter.c
index 449eb7b8d6..3b24e52ff4 100644
--- a/src/core/lib/channel/deadline_filter.c
+++ b/src/core/lib/channel/deadline_filter.c
@@ -41,6 +41,7 @@
 
 #include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/iomgr/timer.h"
+#include "src/core/lib/slice/slice_internal.h"
 
 //
 // grpc_deadline_state
@@ -58,7 +59,7 @@ static void timer_callback(grpc_exec_ctx* exec_ctx, void* arg,
     grpc_slice msg = grpc_slice_from_static_string("Deadline Exceeded");
     grpc_call_element_send_cancel_with_message(
         exec_ctx, elem, GRPC_STATUS_DEADLINE_EXCEEDED, &msg);
-    grpc_slice_unref(msg);
+    grpc_slice_unref_internal(exec_ctx, msg);
   }
   GRPC_CALL_STACK_UNREF(exec_ctx, deadline_state->call_stack, "deadline_timer");
 }
diff --git a/src/core/lib/channel/http_client_filter.c b/src/core/lib/channel/http_client_filter.c
index 0714f31bdd..026e4d486e 100644
--- a/src/core/lib/channel/http_client_filter.c
+++ b/src/core/lib/channel/http_client_filter.c
@@ -36,6 +36,7 @@
 #include <grpc/support/string_util.h>
 #include <string.h>
 #include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/support/string.h"
 #include "src/core/lib/transport/static_metadata.h"
 #include "src/core/lib/transport/transport_impl.h"
@@ -136,8 +137,8 @@ static void hc_on_recv(grpc_exec_ctx *exec_ctx, void *user_data,
   client_recv_filter_args a;
   a.elem = elem;
   a.exec_ctx = exec_ctx;
-  grpc_metadata_batch_filter(calld->recv_initial_metadata, client_recv_filter,
-                             &a);
+  grpc_metadata_batch_filter(exec_ctx, calld->recv_initial_metadata,
+                             client_recv_filter, &a);
   calld->on_done_recv->cb(exec_ctx, calld->on_done_recv->cb_arg, error);
 }
 
@@ -155,7 +156,7 @@ static void hc_on_complete(grpc_exec_ctx *exec_ctx, void *user_data,
 static void send_done(grpc_exec_ctx *exec_ctx, void *elemp, grpc_error *error) {
   grpc_call_element *elem = elemp;
   call_data *calld = elem->call_data;
-  grpc_slice_buffer_reset_and_unref(&calld->slices);
+  grpc_slice_buffer_reset_and_unref_internal(exec_ctx, &calld->slices);
   calld->post_send->cb(exec_ctx, calld->post_send->cb_arg, error);
 }
 
@@ -244,7 +245,7 @@ static void hc_mutate_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
         /* when all the send_message data is available, then create a MDELEM and
         append to headers */
         grpc_mdelem *payload_bin = grpc_mdelem_from_metadata_strings(
-            GRPC_MDSTR_GRPC_PAYLOAD_BIN,
+            exec_ctx, GRPC_MDSTR_GRPC_PAYLOAD_BIN,
             grpc_mdstr_from_buffer(calld->payload_bytes,
                                    op->send_message->length));
         grpc_metadata_batch_add_tail(op->send_initial_metadata,
@@ -261,8 +262,8 @@ static void hc_mutate_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
       }
     }
 
-    grpc_metadata_batch_filter(op->send_initial_metadata, client_strip_filter,
-                               elem);
+    grpc_metadata_batch_filter(exec_ctx, op->send_initial_metadata,
+                               client_strip_filter, elem);
     /* Send : prefixed headers, which have to be before any application
        layer headers. */
     grpc_metadata_batch_add_head(op->send_initial_metadata, &calld->method,
@@ -324,7 +325,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
                               const grpc_call_final_info *final_info,
                               void *ignored) {
   call_data *calld = elem->call_data;
-  grpc_slice_buffer_destroy(&calld->slices);
+  grpc_slice_buffer_destroy_internal(exec_ctx, &calld->slices);
 }
 
 static grpc_mdelem *scheme_from_args(const grpc_channel_args *args) {
@@ -425,7 +426,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
   chand->max_payload_size_for_get =
       max_payload_size_from_args(args->channel_args);
   chand->user_agent = grpc_mdelem_from_metadata_strings(
-      GRPC_MDSTR_USER_AGENT,
+      exec_ctx, GRPC_MDSTR_USER_AGENT,
       user_agent_from_args(args->channel_args,
                            args->optional_transport->vtable->name));
 }
@@ -434,7 +435,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
 static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
                                  grpc_channel_element *elem) {
   channel_data *chand = elem->channel_data;
-  GRPC_MDELEM_UNREF(chand->user_agent);
+  GRPC_MDELEM_UNREF(exec_ctx, chand->user_agent);
 }
 
 const grpc_channel_filter grpc_http_client_filter = {
diff --git a/src/core/lib/channel/http_server_filter.c b/src/core/lib/channel/http_server_filter.c
index 10631850cd..d09a2b13ee 100644
--- a/src/core/lib/channel/http_server_filter.c
+++ b/src/core/lib/channel/http_server_filter.c
@@ -37,6 +37,7 @@
 #include <grpc/support/log.h>
 #include <string.h>
 #include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/transport/static_metadata.h"
 
 #define EXPECTED_CONTENT_TYPE "application/grpc"
@@ -155,7 +156,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
     /* translate host to :authority since :authority may be
        omitted */
     grpc_mdelem *authority = grpc_mdelem_from_metadata_strings(
-        GRPC_MDSTR_AUTHORITY, GRPC_MDSTR_REF(md->value));
+        a->exec_ctx, GRPC_MDSTR_AUTHORITY, GRPC_MDSTR_REF(md->value));
     calld->seen_authority = 1;
     return authority;
   } else if (md->key == GRPC_MDSTR_GRPC_PAYLOAD_BIN) {
@@ -164,7 +165,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
     calld->seen_payload_bin = 1;
     grpc_slice_buffer_init(&calld->read_slice_buffer);
     grpc_slice_buffer_add(&calld->read_slice_buffer,
-                          grpc_slice_ref(md->value->slice));
+                          grpc_slice_ref_internal(md->value->slice));
     grpc_slice_buffer_stream_init(&calld->read_stream,
                                   &calld->read_slice_buffer, 0);
     return NULL;
@@ -181,7 +182,8 @@ static void hs_on_recv(grpc_exec_ctx *exec_ctx, void *user_data,
     server_filter_args a;
     a.elem = elem;
     a.exec_ctx = exec_ctx;
-    grpc_metadata_batch_filter(calld->recv_initial_metadata, server_filter, &a);
+    grpc_metadata_batch_filter(exec_ctx, calld->recv_initial_metadata,
+                               server_filter, &a);
     /* Have we seen the required http2 transport headers?
        (:method, :scheme, content-type, with :path and :authority covered
        at the channel level right now) */
diff --git a/src/core/lib/channel/message_size_filter.c b/src/core/lib/channel/message_size_filter.c
index 85c307702f..50118b52fd 100644
--- a/src/core/lib/channel/message_size_filter.c
+++ b/src/core/lib/channel/message_size_filter.c
@@ -66,8 +66,10 @@ static int message_size_limits_cmp(void* value1, void* value2) {
   return 0;
 }
 
+static void free_mem(grpc_exec_ctx* exec_ctx, void* p) { gpr_free(p); }
+
 static const grpc_mdstr_hash_table_vtable message_size_limits_vtable = {
-    gpr_free, message_size_limits_copy, message_size_limits_cmp};
+    free_mem, message_size_limits_copy, message_size_limits_cmp};
 
 static void* method_config_convert_value(
     const grpc_method_config* method_config) {
@@ -171,8 +173,8 @@ static grpc_error* init_call_elem(grpc_exec_ctx* exec_ctx,
   calld->max_send_size = chand->max_send_size;
   calld->max_recv_size = chand->max_recv_size;
   if (chand->method_limit_table != NULL) {
-    message_size_limits* limits =
-        grpc_method_config_table_get(chand->method_limit_table, args->path);
+    message_size_limits* limits = grpc_method_config_table_get(
+        exec_ctx, chand->method_limit_table, args->path);
     if (limits != NULL) {
       if (limits->max_send_size >= 0 &&
           (limits->max_send_size < calld->max_send_size ||
@@ -225,7 +227,7 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx,
   if (channel_arg != NULL) {
     GPR_ASSERT(channel_arg->type == GRPC_ARG_POINTER);
     chand->method_limit_table = grpc_method_config_table_convert(
-        (grpc_method_config_table*)channel_arg->value.pointer.p,
+        exec_ctx, (grpc_method_config_table*)channel_arg->value.pointer.p,
         method_config_convert_value, &message_size_limits_vtable);
   }
 }
@@ -234,7 +236,7 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx,
 static void destroy_channel_elem(grpc_exec_ctx* exec_ctx,
                                  grpc_channel_element* elem) {
   channel_data* chand = elem->channel_data;
-  grpc_mdstr_hash_table_unref(chand->method_limit_table);
+  grpc_mdstr_hash_table_unref(exec_ctx, chand->method_limit_table);
 }
 
 const grpc_channel_filter grpc_message_size_filter = {