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.

3 files changed, 12 insertions, 56 deletions

diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c
index 83227e677d..2cbf90fb84 100644
--- a/src/core/transport/chttp2/stream_encoder.c
+++ b/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);
   }
 }
diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c
index 3dbb9f0b53..02f9cda754 100644
--- a/src/core/transport/metadata.c
+++ b/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);
diff --git a/src/core/transport/metadata.h b/src/core/transport/metadata.h
index 136a65f288..9a8164037c 100644
--- a/src/core/transport/metadata.h
+++ b/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 */