Merge github.com:grpc/grpc into new_transport_op

8 files changed, 239 insertions, 112 deletions

diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c
index 7cdbe30198..1127fff756 100644
--- a/src/core/lib/iomgr/error.c
+++ b/src/core/lib/iomgr/error.c
@@ -140,14 +140,16 @@ grpc_error *grpc_error_ref(grpc_error *err, const char *file, int line,
                            const char *func) {
   if (grpc_error_is_special(err)) return err;
   gpr_log(GPR_DEBUG, "%p: %" PRIdPTR " -> %" PRIdPTR " [%s:%d %s]", err,
-          err->refs.count, err->refs.count + 1, file, line, func);
-  gpr_ref(&err->refs);
+          gpr_atm_no_barrier_load(&err->atomics.refs.count),
+          gpr_atm_no_barrier_load(&err->atomics.refs.count) + 1, file, line,
+          func);
+  gpr_ref(&err->atomics.refs);
   return err;
 }
 #else
 grpc_error *grpc_error_ref(grpc_error *err) {
   if (grpc_error_is_special(err)) return err;
-  gpr_ref(&err->refs);
+  gpr_ref(&err->atomics.refs);
   return err;
 }
 #endif
@@ -182,7 +184,7 @@ static void error_destroy(grpc_error *err) {
   GPR_ASSERT(!grpc_error_is_special(err));
   unref_errs(err);
   unref_strs(err);
-  gpr_free((void *)gpr_atm_acq_load(&err->error_string));
+  gpr_free((void *)gpr_atm_acq_load(&err->atomics.error_string));
   gpr_free(err);
 }
 
@@ -191,15 +193,17 @@ void grpc_error_unref(grpc_error *err, const char *file, int line,
                       const char *func) {
   if (grpc_error_is_special(err)) return;
   gpr_log(GPR_DEBUG, "%p: %" PRIdPTR " -> %" PRIdPTR " [%s:%d %s]", err,
-          err->refs.count, err->refs.count - 1, file, line, func);
-  if (gpr_unref(&err->refs)) {
+          gpr_atm_no_barrier_load(&err->atomics.refs.count),
+          gpr_atm_no_barrier_load(&err->atomics.refs.count) - 1, file, line,
+          func);
+  if (gpr_unref(&err->atomics.refs)) {
     error_destroy(err);
   }
 }
 #else
 void grpc_error_unref(grpc_error *err) {
   if (grpc_error_is_special(err)) return;
-  if (gpr_unref(&err->refs)) {
+  if (gpr_unref(&err->atomics.refs)) {
     error_destroy(err);
   }
 }
@@ -328,8 +332,8 @@ grpc_error *grpc_error_create(const char *file, int line, const char *desc,
 
   internal_set_time(&err, GRPC_ERROR_TIME_CREATED, gpr_now(GPR_CLOCK_REALTIME));
 
-  gpr_atm_no_barrier_store(&err->error_string, 0);
-  gpr_ref_init(&err->refs, 1);
+  gpr_atm_no_barrier_store(&err->atomics.error_string, 0);
+  gpr_ref_init(&err->atomics.refs, 1);
   GPR_TIMER_END("grpc_error_create", 0);
   return err;
 }
@@ -369,7 +373,7 @@ static grpc_error *copy_error_and_unref(grpc_error *in) {
                        grpc_slice_from_static_string("cancelled"));
       internal_set_int(&out, GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_CANCELLED);
     }
-  } else if (gpr_ref_is_unique(&in->refs)) {
+  } else if (gpr_ref_is_unique(&in->atomics.refs)) {
     out = in;
   } else {
     uint8_t new_arena_capacity = in->arena_capacity;
@@ -382,10 +386,14 @@ static grpc_error *copy_error_and_unref(grpc_error *in) {
 #ifdef GRPC_ERROR_REFCOUNT_DEBUG
     gpr_log(GPR_DEBUG, "%p create copying %p", out, in);
 #endif
-    memcpy(out, in, sizeof(*in) + in->arena_size * sizeof(intptr_t));
+    // bulk memcpy of the rest of the struct.
+    size_t skip = sizeof(&out->atomics);
+    memcpy((void *)((uintptr_t)out + skip), (void *)((uintptr_t)in + skip),
+           sizeof(*in) + (in->arena_size * sizeof(intptr_t)) - skip);
+    // manually set the atomics and the new capacity
+    gpr_atm_no_barrier_store(&out->atomics.error_string, 0);
+    gpr_ref_init(&out->atomics.refs, 1);
     out->arena_capacity = new_arena_capacity;
-    gpr_atm_no_barrier_store(&out->error_string, 0);
-    gpr_ref_init(&out->refs, 1);
     ref_strs(out);
     ref_errs(out);
     GRPC_ERROR_UNREF(in);
@@ -692,7 +700,7 @@ const char *grpc_error_string(grpc_error *err) {
   if (err == GRPC_ERROR_OOM) return oom_error_string;
   if (err == GRPC_ERROR_CANCELLED) return cancelled_error_string;
 
-  void *p = (void *)gpr_atm_acq_load(&err->error_string);
+  void *p = (void *)gpr_atm_acq_load(&err->atomics.error_string);
   if (p != NULL) {
     GPR_TIMER_END("grpc_error_string", 0);
     return p;
@@ -712,9 +720,9 @@ const char *grpc_error_string(grpc_error *err) {
 
   char *out = finish_kvs(&kvs);
 
-  if (!gpr_atm_rel_cas(&err->error_string, 0, (gpr_atm)out)) {
+  if (!gpr_atm_rel_cas(&err->atomics.error_string, 0, (gpr_atm)out)) {
     gpr_free(out);
-    out = (char *)gpr_atm_no_barrier_load(&err->error_string);
+    out = (char *)gpr_atm_no_barrier_load(&err->atomics.error_string);
   }
 
   GPR_TIMER_END("grpc_error_string", 0);
diff --git a/src/core/lib/iomgr/error_internal.h b/src/core/lib/iomgr/error_internal.h
index fb4814e41f..7f204df1b2 100644
--- a/src/core/lib/iomgr/error_internal.h
+++ b/src/core/lib/iomgr/error_internal.h
@@ -46,14 +46,25 @@ struct grpc_linked_error {
   uint8_t next;
 };
 
+// c core representation of an error. See error.h for high level description of
+// this object.
 struct grpc_error {
-  gpr_refcount refs;
+  // All atomics in grpc_error must be stored in this nested struct. The rest of
+  // the object is memcpy-ed in bulk in copy_and_unref.
+  struct atomics {
+    gpr_refcount refs;
+    gpr_atm error_string;
+  } atomics;
+  // These arrays index into dynamic arena at the bottom of the struct.
+  // UINT8_MAX is used as a sentinel value.
   uint8_t ints[GRPC_ERROR_INT_MAX];
   uint8_t strs[GRPC_ERROR_STR_MAX];
   uint8_t times[GRPC_ERROR_TIME_MAX];
+  // The child errors are stored in the arena, but are effectively a linked list
+  // structure, since they are contained withing grpc_linked_error objects.
   uint8_t first_err;
   uint8_t last_err;
-  gpr_atm error_string;
+  // The arena is dynamically reallocated with a grow factor of 1.5.
   uint8_t arena_size;
   uint8_t arena_capacity;
   intptr_t arena[0];
diff --git a/src/core/lib/iomgr/udp_server.c b/src/core/lib/iomgr/udp_server.c
index d1bcd89af1..71e295770a 100644
--- a/src/core/lib/iomgr/udp_server.c
+++ b/src/core/lib/iomgr/udp_server.c
@@ -109,8 +109,8 @@ struct grpc_udp_server {
   grpc_pollset **pollsets;
   /* number of pollsets in the pollsets array */
   size_t pollset_count;
-  /* The parent grpc server */
-  grpc_server *grpc_server;
+  /* opaque object to pass to callbacks */
+  void *user_data;
 };
 
 grpc_udp_server *grpc_udp_server_create(void) {
@@ -178,7 +178,7 @@ static void deactivated_all_ports(grpc_exec_ctx *exec_ctx, grpc_udp_server *s) {
       /* Call the orphan_cb to signal that the FD is about to be closed and
        * should no longer be used. */
       GPR_ASSERT(sp->orphan_cb);
-      sp->orphan_cb(exec_ctx, sp->emfd);
+      sp->orphan_cb(exec_ctx, sp->emfd, sp->server->user_data);
 
       grpc_fd_orphan(exec_ctx, sp->emfd, &sp->destroyed_closure, NULL,
                      "udp_listener_shutdown");
@@ -204,7 +204,7 @@ void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
   if (s->active_ports) {
     for (sp = s->head; sp; sp = sp->next) {
       GPR_ASSERT(sp->orphan_cb);
-      sp->orphan_cb(exec_ctx, sp->emfd);
+      sp->orphan_cb(exec_ctx, sp->emfd, sp->server->user_data);
       grpc_fd_shutdown(exec_ctx, sp->emfd,
                        GRPC_ERROR_CREATE("Server destroyed"));
     }
@@ -299,7 +299,7 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
 
   /* Tell the registered callback that data is available to read. */
   GPR_ASSERT(sp->read_cb);
-  sp->read_cb(exec_ctx, sp->emfd, sp->server->grpc_server);
+  sp->read_cb(exec_ctx, sp->emfd, sp->server->user_data);
 
   /* Re-arm the notification event so we get another chance to read. */
   grpc_fd_notify_on_read(exec_ctx, sp->emfd, &sp->read_closure);
@@ -322,7 +322,7 @@ static void on_write(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
 
   /* Tell the registered callback that the socket is writeable. */
   GPR_ASSERT(sp->write_cb);
-  sp->write_cb(exec_ctx, sp->emfd);
+  sp->write_cb(exec_ctx, sp->emfd, sp->server->user_data);
 
   /* Re-arm the notification event so we get another chance to write. */
   grpc_fd_notify_on_write(exec_ctx, sp->emfd, &sp->write_closure);
@@ -464,13 +464,13 @@ int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index) {
 
 void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
                            grpc_pollset **pollsets, size_t pollset_count,
-                           grpc_server *server) {
+                           void *user_data) {
   size_t i;
   gpr_mu_lock(&s->mu);
   grpc_udp_listener *sp;
   GPR_ASSERT(s->active_ports == 0);
   s->pollsets = pollsets;
-  s->grpc_server = server;
+  s->user_data = user_data;
 
   sp = s->head;
   while (sp != NULL) {
diff --git a/src/core/lib/iomgr/udp_server.h b/src/core/lib/iomgr/udp_server.h
index ed63fa7d81..90842a47f0 100644
--- a/src/core/lib/iomgr/udp_server.h
+++ b/src/core/lib/iomgr/udp_server.h
@@ -47,23 +47,23 @@ typedef struct grpc_udp_server grpc_udp_server;
 
 /* Called when data is available to read from the socket. */
 typedef void (*grpc_udp_server_read_cb)(grpc_exec_ctx *exec_ctx, grpc_fd *emfd,
-                                        struct grpc_server *server);
+                                        void *user_data);
 
 /* Called when the socket is writeable. */
-typedef void (*grpc_udp_server_write_cb)(grpc_exec_ctx *exec_ctx,
-                                         grpc_fd *emfd);
+typedef void (*grpc_udp_server_write_cb)(grpc_exec_ctx *exec_ctx, grpc_fd *emfd,
+                                         void *user_data);
 
 /* Called when the grpc_fd is about to be orphaned (and the FD closed). */
 typedef void (*grpc_udp_server_orphan_cb)(grpc_exec_ctx *exec_ctx,
-                                          grpc_fd *emfd);
+                                          grpc_fd *emfd, void *user_data);
 
 /* Create a server, initially not bound to any ports */
 grpc_udp_server *grpc_udp_server_create(void);
 
-/* Start listening to bound ports */
+/* Start listening to bound ports. user_data is passed to callbacks. */
 void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *udp_server,
                            grpc_pollset **pollsets, size_t pollset_count,
-                           struct grpc_server *server);
+                           void *user_data);
 
 int grpc_udp_server_get_fd(grpc_udp_server *s, unsigned port_index);
 
diff --git a/src/core/lib/support/arena.c b/src/core/lib/support/arena.c
new file mode 100644
index 0000000000..7bcb983f24
--- /dev/null
+++ b/src/core/lib/support/arena.c
@@ -0,0 +1,98 @@
+/*
+ *
+ * Copyright 2017, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include "src/core/lib/support/arena.h"
+#include <grpc/support/alloc.h>
+#include <grpc/support/atm.h>
+#include <grpc/support/log.h>
+#include <grpc/support/useful.h>
+
+#define ROUND_UP_TO_ALIGNMENT_SIZE(x) \
+  (((x) + GPR_MAX_ALIGNMENT - 1u) & ~(GPR_MAX_ALIGNMENT - 1u))
+
+typedef struct zone {
+  size_t size_begin;
+  size_t size_end;
+  gpr_atm next_atm;
+} zone;
+
+struct gpr_arena {
+  gpr_atm size_so_far;
+  zone initial_zone;
+};
+
+gpr_arena *gpr_arena_create(size_t initial_size) {
+  initial_size = ROUND_UP_TO_ALIGNMENT_SIZE(initial_size);
+  gpr_arena *a = gpr_zalloc(sizeof(gpr_arena) + initial_size);
+  a->initial_zone.size_end = initial_size;
+  return a;
+}
+
+size_t gpr_arena_destroy(gpr_arena *arena) {
+  gpr_atm size = gpr_atm_no_barrier_load(&arena->size_so_far);
+  zone *z = (zone *)gpr_atm_no_barrier_load(&arena->initial_zone.next_atm);
+  gpr_free(arena);
+  while (z) {
+    zone *next_z = (zone *)gpr_atm_no_barrier_load(&z->next_atm);
+    gpr_free(z);
+    z = next_z;
+  }
+  return (size_t)size;
+}
+
+void *gpr_arena_alloc(gpr_arena *arena, size_t size) {
+  size = ROUND_UP_TO_ALIGNMENT_SIZE(size);
+  size_t start =
+      (size_t)gpr_atm_no_barrier_fetch_add(&arena->size_so_far, size);
+  zone *z = &arena->initial_zone;
+  while (start > z->size_end) {
+    zone *next_z = (zone *)gpr_atm_acq_load(&z->next_atm);
+    if (next_z == NULL) {
+      size_t next_z_size = (size_t)gpr_atm_no_barrier_load(&arena->size_so_far);
+      next_z = gpr_zalloc(sizeof(zone) + next_z_size);
+      next_z->size_begin = z->size_end;
+      next_z->size_end = z->size_end + next_z_size;
+      if (!gpr_atm_rel_cas(&z->next_atm, (gpr_atm)NULL, (gpr_atm)next_z)) {
+        gpr_free(next_z);
+        next_z = (zone *)gpr_atm_acq_load(&z->next_atm);
+      }
+    }
+    z = next_z;
+  }
+  if (start + size > z->size_end) {
+    return gpr_arena_alloc(arena, size);
+  }
+  GPR_ASSERT(start >= z->size_begin);
+  GPR_ASSERT(start + size <= z->size_end);
+  return ((char *)(z + 1)) + start - z->size_begin;
+}
diff --git a/src/core/lib/support/arena.h b/src/core/lib/support/arena.h
new file mode 100644
index 0000000000..c28033ffc3
--- /dev/null
+++ b/src/core/lib/support/arena.h
@@ -0,0 +1,54 @@
+/*
+ *
+ * Copyright 2017, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+// \file Arena based allocator
+// Allows very fast allocation of memory, but that memory cannot be freed until
+// the arena as a whole is freed
+// Tracks the total memory allocated against it, so that future arenas can
+// pre-allocate the right amount of memory
+
+#ifndef GRPC_CORE_LIB_SUPPORT_ARENA_H
+#define GRPC_CORE_LIB_SUPPORT_ARENA_H
+
+#include <stddef.h>
+
+typedef struct gpr_arena gpr_arena;
+
+// Create an arena, with \a initial_size bytes in the first allocated buffer
+gpr_arena *gpr_arena_create(size_t initial_size);
+// Allocate \a size bytes from the arena
+void *gpr_arena_alloc(gpr_arena *arena, size_t size);
+// Destroy an arena, returning the total number of bytes allocated
+size_t gpr_arena_destroy(gpr_arena *arena);
+
+#endif /* GRPC_CORE_LIB_SUPPORT_ARENA_H */
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index 0c803c717b..bd34ed1b91 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -139,8 +139,8 @@ struct grpc_call {
   grpc_call *parent;
   grpc_call *first_child;
   gpr_timespec start_time;
-  /* TODO(ctiller): share with cq if possible? */
-  gpr_mu mu;
+  /* protects first_child, and child next/prev links */
+  gpr_mu child_list_mu;
 
   /* client or server call */
   bool is_client;
@@ -155,8 +155,8 @@ struct grpc_call {
   bool received_initial_metadata;
   bool receiving_message;
   bool requested_final_op;
-  bool received_final_op;
-  bool sent_any_op;
+  gpr_atm any_ops_sent_atm;
+  gpr_atm received_final_op_atm;
 
   /* have we received initial metadata */
   bool has_initial_md_been_received;
@@ -271,7 +271,7 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
   GPR_TIMER_BEGIN("grpc_call_create", 0);
   call = gpr_zalloc(sizeof(grpc_call) + channel_stack->call_stack_size);
   *out_call = call;
-  gpr_mu_init(&call->mu);
+  gpr_mu_init(&call->child_list_mu);
   call->channel = args->channel;
   call->cq = args->cq;
   call->parent = args->parent_call;
@@ -310,7 +310,7 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
     GPR_ASSERT(call->is_client);
     GPR_ASSERT(!args->parent_call->is_client);
 
-    gpr_mu_lock(&args->parent_call->mu);
+    gpr_mu_lock(&args->parent_call->child_list_mu);
 
     if (args->propagation_mask & GRPC_PROPAGATE_DEADLINE) {
       send_deadline = gpr_time_min(
@@ -338,6 +338,10 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
     }
     if (args->propagation_mask & GRPC_PROPAGATE_CANCELLATION) {
       call->cancellation_is_inherited = 1;
+      if (gpr_atm_acq_load(&args->parent_call->received_final_op_atm)) {
+        cancel_with_error(exec_ctx, call, STATUS_FROM_API_OVERRIDE,
+                          GRPC_ERROR_CANCELLED);
+      }
     }
 
     if (args->parent_call->first_child == NULL) {
@@ -350,7 +354,7 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
           call;
     }
 
-    gpr_mu_unlock(&args->parent_call->mu);
+    gpr_mu_unlock(&args->parent_call->child_list_mu);
   }
 
   call->send_deadline = send_deadline;
@@ -432,7 +436,7 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call,
   if (c->receiving_stream != NULL) {
     grpc_byte_stream_destroy(exec_ctx, c->receiving_stream);
   }
-  gpr_mu_destroy(&c->mu);
+  gpr_mu_destroy(&c->child_list_mu);
   for (ii = 0; ii < c->send_extra_metadata_count; ii++) {
     GRPC_MDELEM_UNREF(exec_ctx, c->send_extra_metadata[ii].md);
   }
@@ -453,7 +457,7 @@ static void destroy_call(grpc_exec_ctx *exec_ctx, void *call,
 
   for (i = 0; i < STATUS_SOURCE_COUNT; i++) {
     GRPC_ERROR_UNREF(
-        unpack_received_status(gpr_atm_no_barrier_load(&c->status[i])).error);
+        unpack_received_status(gpr_atm_acq_load(&c->status[i])).error);
   }
 
   grpc_call_stack_destroy(exec_ctx, CALL_STACK_FROM_CALL(c), &c->final_info, c);
@@ -470,7 +474,7 @@ void grpc_call_destroy(grpc_call *c) {
   GRPC_API_TRACE("grpc_call_destroy(c=%p)", 1, (c));
 
   if (parent) {
-    gpr_mu_lock(&parent->mu);
+    gpr_mu_lock(&parent->child_list_mu);
     if (c == parent->first_child) {
       parent->first_child = c->sibling_next;
       if (c == parent->first_child) {
@@ -479,15 +483,14 @@ void grpc_call_destroy(grpc_call *c) {
       c->sibling_prev->sibling_next = c->sibling_next;
       c->sibling_next->sibling_prev = c->sibling_prev;
     }
-    gpr_mu_unlock(&parent->mu);
+    gpr_mu_unlock(&parent->child_list_mu);
     GRPC_CALL_INTERNAL_UNREF(&exec_ctx, parent, "child");
   }
 
-  gpr_mu_lock(&c->mu);
   GPR_ASSERT(!c->destroy_called);
   c->destroy_called = 1;
-  cancel = c->sent_any_op && !c->received_final_op;
-  gpr_mu_unlock(&c->mu);
+  cancel = gpr_atm_acq_load(&c->any_ops_sent_atm) &&
+           !gpr_atm_acq_load(&c->received_final_op_atm);
   if (cancel) {
     cancel_with_error(&exec_ctx, c, STATUS_FROM_API_OVERRIDE,
                       GRPC_ERROR_CANCELLED);
@@ -551,56 +554,26 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
       "c=%p, status=%d, description=%s, reserved=%p)",
       4, (c, (int)status, description, reserved));
   GPR_ASSERT(reserved == NULL);
-  gpr_mu_lock(&c->mu);
   cancel_with_status(&exec_ctx, c, STATUS_FROM_API_OVERRIDE, status,
                      description);
-  gpr_mu_unlock(&c->mu);
   grpc_exec_ctx_finish(&exec_ctx);
   return GRPC_CALL_OK;
 }
 
-typedef struct termination_closure {
-  grpc_closure closure;
-  grpc_call *call;
-  grpc_transport_stream_op op;
-  grpc_transport_stream_op_payload payload;
-} termination_closure;
-
-static void done_termination(grpc_exec_ctx *exec_ctx, void *tcp,
-                             grpc_error *error) {
-  termination_closure *tc = tcp;
-  GRPC_CALL_INTERNAL_UNREF(exec_ctx, tc->call, "termination");
-  gpr_free(tc);
-}
-
-static void send_termination(grpc_exec_ctx *exec_ctx, void *tcp,
+static void done_termination(grpc_exec_ctx *exec_ctx, void *call,
                              grpc_error *error) {
-  termination_closure *tc = tcp;
-  memset(&tc->op, 0, sizeof(tc->op));
-  tc->op.payload = &tc->payload;
-  tc->op.cancel_stream = true;
-  tc->op.payload->cancel_stream.cancel_error = GRPC_ERROR_REF(error);
-  /* reuse closure to catch completion */
-  tc->op.on_complete = grpc_closure_init(&tc->closure, done_termination, tc,
-                                         grpc_schedule_on_exec_ctx);
-  execute_op(exec_ctx, tc->call, &tc->op);
-}
-
-static void terminate_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c,
-                                 grpc_error *error) {
-  termination_closure *tc = gpr_malloc(sizeof(*tc));
-  memset(tc, 0, sizeof(*tc));
-  tc->call = c;
-  GRPC_CALL_INTERNAL_REF(tc->call, "termination");
-  grpc_closure_sched(exec_ctx, grpc_closure_init(&tc->closure, send_termination,
-                                                 tc, grpc_schedule_on_exec_ctx),
-                     error);
+  GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "termination");
 }
 
 static void cancel_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c,
                               status_source source, grpc_error *error) {
+  GRPC_CALL_INTERNAL_REF(c, "termination");
   set_status_from_error(exec_ctx, c, source, GRPC_ERROR_REF(error));
-  terminate_with_error(exec_ctx, c, error);
+  grpc_transport_stream_op *op = grpc_make_transport_stream_op(
+      grpc_closure_create(done_termination, c, grpc_schedule_on_exec_ctx));
+  op->cancel_stream = true;
+  op->payload->cancel_stream.cancel_error = error;
+  execute_op(exec_ctx, c, op);
 }
 
 static grpc_error *error_from_status(grpc_status_code status,
@@ -714,9 +687,7 @@ static void set_incoming_compression_algorithm(
 grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm(
     grpc_call *call) {
   grpc_compression_algorithm algorithm;
-  gpr_mu_lock(&call->mu);
   algorithm = call->incoming_compression_algorithm;
-  gpr_mu_unlock(&call->mu);
   return algorithm;
 }
 
@@ -728,9 +699,7 @@ static grpc_compression_algorithm compression_algorithm_for_level_locked(
 
 uint32_t grpc_call_test_only_get_message_flags(grpc_call *call) {
   uint32_t flags;
-  gpr_mu_lock(&call->mu);
   flags = call->test_only_last_message_flags;
-  gpr_mu_unlock(&call->mu);
   return flags;
 }
 
@@ -784,9 +753,7 @@ static void set_encodings_accepted_by_peer(grpc_exec_ctx *exec_ctx,
 
 uint32_t grpc_call_test_only_get_encodings_accepted_by_peer(grpc_call *call) {
   uint32_t encodings_accepted_by_peer;
-  gpr_mu_lock(&call->mu);
   encodings_accepted_by_peer = call->encodings_accepted_by_peer;
-  gpr_mu_unlock(&call->mu);
   return encodings_accepted_by_peer;
 }
 
@@ -1052,7 +1019,7 @@ static void finish_batch_completion(grpc_exec_ctx *exec_ctx, void *user_data,
 }
 
 static grpc_error *consolidate_batch_errors(batch_control *bctl) {
-  size_t n = (size_t)gpr_atm_no_barrier_load(&bctl->num_errors);
+  size_t n = (size_t)gpr_atm_acq_load(&bctl->num_errors);
   if (n == 0) {
     return GRPC_ERROR_NONE;
   } else if (n == 1) {
@@ -1079,8 +1046,6 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
   grpc_call *call = bctl->call;
   grpc_error *error = consolidate_batch_errors(bctl);
 
-  gpr_mu_lock(&call->mu);
-
   if (bctl->op.send_initial_metadata) {
     grpc_metadata_batch_destroy(
         exec_ctx,
@@ -1099,20 +1064,23 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
         &call->metadata_batch[1 /* is_receiving */][1 /* is_trailing */];
     recv_trailing_filter(exec_ctx, call, md);
 
-    call->received_final_op = true;
     /* propagate cancellation to any interested children */
+    gpr_atm_rel_store(&call->received_final_op_atm, 1);
+    gpr_mu_lock(&call->child_list_mu);
     child_call = call->first_child;
     if (child_call != NULL) {
       do {
         next_child_call = child_call->sibling_next;
         if (child_call->cancellation_is_inherited) {
           GRPC_CALL_INTERNAL_REF(child_call, "propagate_cancel");
-          grpc_call_cancel(child_call, NULL);
+          cancel_with_error(exec_ctx, child_call, STATUS_FROM_API_OVERRIDE,
+                            GRPC_ERROR_CANCELLED);
           GRPC_CALL_INTERNAL_UNREF(exec_ctx, child_call, "propagate_cancel");
         }
         child_call = next_child_call;
       } while (child_call != call->first_child);
     }
+    gpr_mu_unlock(&call->child_list_mu);
 
     if (call->is_client) {
       get_final_status(call, set_status_value_directly,
@@ -1126,7 +1094,6 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
     GRPC_ERROR_UNREF(error);
     error = GRPC_ERROR_NONE;
   }
-  gpr_mu_unlock(&call->mu);
 
   if (bctl->completion_data.notify_tag.is_closure) {
     /* unrefs bctl->error */
@@ -1218,7 +1185,6 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
                                    grpc_error *error) {
   batch_control *bctl = bctlp;
   grpc_call *call = bctl->call;
-  gpr_mu_lock(&bctl->call->mu);
   if (error != GRPC_ERROR_NONE) {
     if (call->receiving_stream != NULL) {
       grpc_byte_stream_destroy(exec_ctx, call->receiving_stream);
@@ -1230,11 +1196,9 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
   }
   if (call->has_initial_md_been_received || error != GRPC_ERROR_NONE ||
       call->receiving_stream == NULL) {
-    gpr_mu_unlock(&bctl->call->mu);
     process_data_after_md(exec_ctx, bctlp);
   } else {
     call->saved_receiving_stream_ready_bctlp = bctlp;
-    gpr_mu_unlock(&bctl->call->mu);
   }
 }
 
@@ -1293,7 +1257,7 @@ static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx,
 static void add_batch_error(grpc_exec_ctx *exec_ctx, batch_control *bctl,
                             grpc_error *error, bool has_cancelled) {
   if (error == GRPC_ERROR_NONE) return;
-  int idx = (int)gpr_atm_no_barrier_fetch_add(&bctl->num_errors, 1);
+  int idx = (int)gpr_atm_full_fetch_add(&bctl->num_errors, 1);
   if (idx == 0 && !has_cancelled) {
     cancel_with_error(exec_ctx, bctl->call, STATUS_FROM_CORE,
                       GRPC_ERROR_REF(error));
@@ -1306,8 +1270,6 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
   batch_control *bctl = bctlp;
   grpc_call *call = bctl->call;
 
-  gpr_mu_lock(&call->mu);
-
   add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error), false);
   if (error == GRPC_ERROR_NONE) {
     grpc_metadata_batch *md =
@@ -1333,11 +1295,9 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
         receiving_stream_ready, call->saved_receiving_stream_ready_bctlp,
         grpc_schedule_on_exec_ctx);
     call->saved_receiving_stream_ready_bctlp = NULL;
-    grpc_closure_sched(exec_ctx, saved_rsr_closure, GRPC_ERROR_REF(error));
+    grpc_closure_run(exec_ctx, saved_rsr_closure, GRPC_ERROR_REF(error));
   }
 
-  gpr_mu_unlock(&call->mu);
-
   finish_batch_step(exec_ctx, bctl);
 }
 
@@ -1391,7 +1351,6 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
   bctl->completion_data.notify_tag.is_closure =
       (uint8_t)(is_notify_tag_closure != 0);
 
-  gpr_mu_lock(&call->mu);
   grpc_transport_stream_op *stream_op = &bctl->op;
   grpc_transport_stream_op_payload *stream_op_payload =
       &call->stream_op_payload;
@@ -1682,8 +1641,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
   grpc_closure_init(&bctl->finish_batch, finish_batch, bctl,
                     grpc_schedule_on_exec_ctx);
   stream_op->on_complete = &bctl->finish_batch;
-  call->sent_any_op = true;
-  gpr_mu_unlock(&call->mu);
+  gpr_atm_rel_store(&call->any_ops_sent_atm, 1);
 
   execute_op(exec_ctx, call, stream_op);
 
@@ -1714,7 +1672,6 @@ done_with_error:
   if (stream_op->recv_trailing_metadata) {
     call->requested_final_op = 0;
   }
-  gpr_mu_unlock(&call->mu);
   goto done;
 }
 
@@ -1763,10 +1720,8 @@ uint8_t grpc_call_is_client(grpc_call *call) { return call->is_client; }
 
 grpc_compression_algorithm grpc_call_compression_for_level(
     grpc_call *call, grpc_compression_level level) {
-  gpr_mu_lock(&call->mu);
   grpc_compression_algorithm algo =
       compression_algorithm_for_level_locked(call, level);
-  gpr_mu_unlock(&call->mu);
   return algo;
 }
 
diff --git a/src/core/lib/transport/transport.c b/src/core/lib/transport/transport.c
index 3b0d18c789..ce06b4d004 100644
--- a/src/core/lib/transport/transport.c
+++ b/src/core/lib/transport/transport.c
@@ -264,8 +264,9 @@ typedef struct {
 static void destroy_made_transport_stream_op(grpc_exec_ctx *exec_ctx, void *arg,
                                              grpc_error *error) {
   made_transport_stream_op *op = arg;
-  grpc_closure_sched(exec_ctx, op->inner_on_complete, GRPC_ERROR_REF(error));
+  grpc_closure *c = op->inner_on_complete;
   gpr_free(op);
+  grpc_closure_run(exec_ctx, c, GRPC_ERROR_REF(error));
 }
 
 grpc_transport_stream_op *grpc_make_transport_stream_op(