Make metadata unref atomic

We used to need to lock the metadata context to unref an mdelem. This
change makes it possible to lock only when the mdelem refcount would
reach zero.

src/core/surface/call.c

@@ -1484,7 +1484,6 @@ static void recv_metadata(grpc_exec_ctx *exec_ctx, grpc_call *call,
   grpc_metadata_array *dest;
   grpc_metadata *mdusr;
   int is_trailing;
-  grpc_mdctx *mdctx = call->metadata_context;
 
   is_trailing = call->read_state >= READ_STATE_GOT_INITIAL_METADATA;
   for (l = md->list.head; l != NULL; l = l->next) {
@@ -1532,14 +1531,12 @@ static void recv_metadata(grpc_exec_ctx *exec_ctx, grpc_call *call,
     call->read_state = READ_STATE_GOT_INITIAL_METADATA;
   }
 
-  grpc_mdctx_lock(mdctx);
   for (l = md->list.head; l; l = l->next) {
-    if (l->md) GRPC_MDCTX_LOCKED_MDELEM_UNREF(mdctx, l->md);
+    if (l->md) GRPC_MDELEM_UNREF(l->md);
   }
   for (l = md->garbage.head; l; l = l->next) {
-    GRPC_MDCTX_LOCKED_MDELEM_UNREF(mdctx, l->md);
+    GRPC_MDELEM_UNREF(l->md);
   }
-  grpc_mdctx_unlock(mdctx);
 }
 
 grpc_call_stack *grpc_call_get_call_stack(grpc_call *call) {

src/core/transport/chttp2/stream_encoder.c

@@ -584,7 +584,6 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
   size_t max_take_size;
   gpr_uint32 curop = 0;
   gpr_uint32 unref_op;
-  grpc_mdctx *mdctx = compressor->mdctx;
   grpc_linked_mdelem *l;
   int need_unref = 0;
   gpr_timespec deadline;
@@ -650,17 +649,15 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof,
   finish_frame(&st, 1, eof);
 
   if (need_unref) {
-    grpc_mdctx_lock(mdctx);
     for (unref_op = 0; unref_op < curop; unref_op++) {
       op = &ops[unref_op];
       if (op->type != GRPC_OP_METADATA) continue;
       for (l = op->data.metadata.list.head; l; l = l->next) {
-        if (l->md) GRPC_MDCTX_LOCKED_MDELEM_UNREF(mdctx, l->md);
+        if (l->md) GRPC_MDELEM_UNREF(l->md);
       }
       for (l = op->data.metadata.garbage.head; l; l = l->next) {
-        GRPC_MDCTX_LOCKED_MDELEM_UNREF(mdctx, l->md);
+        GRPC_MDELEM_UNREF(l->md);
       }
     }
-    grpc_mdctx_unlock(mdctx);
   }
 }

src/core/transport/metadata.c

@@ -159,6 +159,10 @@ static void ref_md_locked(internal_metadata *md DEBUG_ARGS) {
           grpc_mdstr_as_c_string((grpc_mdstr *)md->value));
 #endif
   if (0 == gpr_atm_no_barrier_fetch_add(&md->refcnt, 1)) {
+    /* This ref is dropped if grpc_mdelem_unref reaches 1,
+       but allows us to safely unref without taking the mdctx lock
+       until such time */
+    gpr_atm_no_barrier_fetch_add(&md->refcnt, 1);
     md->context->mdtab_free--;
   }
 }
@@ -465,7 +469,7 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdctx *ctx,
 
   /* not found: create a new pair */
   md = gpr_malloc(sizeof(internal_metadata));
-  gpr_atm_rel_store(&md->refcnt, 1);
+  gpr_atm_rel_store(&md->refcnt, 2);
   md->context = ctx;
   md->key = key;
   md->value = value;
@@ -534,8 +538,6 @@ grpc_mdelem *grpc_mdelem_ref(grpc_mdelem *gmd DEBUG_ARGS) {
 
 void grpc_mdelem_unref(grpc_mdelem *gmd DEBUG_ARGS) {
   internal_metadata *md = (internal_metadata *)gmd;
-  grpc_mdctx *ctx = md->context;
-  lock(ctx);
 #ifdef GRPC_METADATA_REFCOUNT_DEBUG
   gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
           "ELM UNREF:%p:%d->%d: '%s' = '%s'", md,
@@ -544,11 +546,13 @@ void grpc_mdelem_unref(grpc_mdelem *gmd DEBUG_ARGS) {
           grpc_mdstr_as_c_string((grpc_mdstr *)md->key),
           grpc_mdstr_as_c_string((grpc_mdstr *)md->value));
 #endif
-  assert(gpr_atm_no_barrier_load(&md->refcnt) >= 1);
-  if (1 == gpr_atm_full_fetch_add(&md->refcnt, -1)) {
+  if (2 == gpr_atm_full_fetch_add(&md->refcnt, -1)) {
+    grpc_mdctx *ctx = md->context;
+    lock(ctx);
+    GPR_ASSERT(1 == gpr_atm_full_fetch_add(&md->refcnt, -1));
     ctx->mdtab_free++;
+    unlock(ctx);
   }
-  unlock(ctx);
 }
 
 const char *grpc_mdstr_as_c_string(grpc_mdstr *s) {
@@ -627,29 +631,6 @@ gpr_slice grpc_mdstr_as_base64_encoded_and_huffman_compressed(grpc_mdstr *gs) {
   return slice;
 }
 
-void grpc_mdctx_lock(grpc_mdctx *ctx) { lock(ctx); }
-
-void grpc_mdctx_locked_mdelem_unref(grpc_mdctx *ctx,
-                                    grpc_mdelem *gmd DEBUG_ARGS) {
-  internal_metadata *md = (internal_metadata *)gmd;
-  grpc_mdctx *elem_ctx = md->context;
-  GPR_ASSERT(ctx == elem_ctx);
-#ifdef GRPC_METADATA_REFCOUNT_DEBUG
-  gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG,
-          "ELM UNREF:%p:%d->%d: '%s' = '%s'", md,
-          gpr_atm_no_barrier_load(&md->refcnt),
-          gpr_atm_no_barrier_load(&md->refcnt) - 1,
-          grpc_mdstr_as_c_string((grpc_mdstr *)md->key),
-          grpc_mdstr_as_c_string((grpc_mdstr *)md->value));
-#endif
-  assert(gpr_atm_no_barrier_load(&md->refcnt) >= 1);
-  if (1 == gpr_atm_full_fetch_add(&md->refcnt, -1)) {
-    ctx->mdtab_free++;
-  }
-}
-
-void grpc_mdctx_unlock(grpc_mdctx *ctx) { unlock(ctx); }
-
 static int conforms_to(grpc_mdstr *s, const gpr_uint8 *legal_bits) {
   const gpr_uint8 *p = GPR_SLICE_START_PTR(s->slice);
   const gpr_uint8 *e = GPR_SLICE_END_PTR(s->slice);

src/core/transport/metadata.h

@@ -155,28 +155,6 @@ int grpc_mdstr_is_legal_header(grpc_mdstr *s);
 int grpc_mdstr_is_legal_nonbin_header(grpc_mdstr *s);
 int grpc_mdstr_is_bin_suffixed(grpc_mdstr *s);
 
-/* Batch mode metadata functions.
-   These API's have equivalents above, but allow taking the mdctx just once,
-   performing a bunch of work, and then leaving the mdctx. */
-
-/* Lock the metadata context: it's only safe to call _locked_ functions against
-   this context from the calling thread until grpc_mdctx_unlock is called */
-void grpc_mdctx_lock(grpc_mdctx *ctx);
-#ifdef GRPC_METADATA_REFCOUNT_DEBUG
-#define GRPC_MDCTX_LOCKED_MDELEM_UNREF(ctx, elem) \
-  grpc_mdctx_locked_mdelem_unref((ctx), (elem), __FILE__, __LINE__)
-/* Unref a metadata element */
-void grpc_mdctx_locked_mdelem_unref(grpc_mdctx *ctx, grpc_mdelem *elem,
-                                    const char *file, int line);
-#else
-#define GRPC_MDCTX_LOCKED_MDELEM_UNREF(ctx, elem) \
-  grpc_mdctx_locked_mdelem_unref((ctx), (elem))
-/* Unref a metadata element */
-void grpc_mdctx_locked_mdelem_unref(grpc_mdctx *ctx, grpc_mdelem *elem);
-#endif
-/* Unlock the metadata context */
-void grpc_mdctx_unlock(grpc_mdctx *ctx);
-
 #define GRPC_MDSTR_KV_HASH(k_hash, v_hash) (GPR_ROTL((k_hash), 2) ^ (v_hash))
 
 #endif /* GRPC_INTERNAL_CORE_TRANSPORT_METADATA_H */