Use the unified Hashtable API in client-side stats (perf toolbox)

dd-trace-core/src/jmh/java/datadog/trace/common/metrics/CardinalityLimitReporterBenchmark.java

@@ -0,0 +1,61 @@
+package datadog.trace.common.metrics;
+
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * Steady-state {@code record()} acceptance check: once every tag in the working set has an entry,
+ * every call should be a lookup + in-place count bump through the {@link
+ * datadog.trace.util.Hashtable.D1#tryGetOrCreate} {@code Maybe}, with no per-call allocation. Run
+ * with {@code -prof gc} -- B/op should read ~0.
+ *
+ * <p>Not thread-safe by design (see {@link CardinalityLimitReporter}'s class javadoc), so each
+ * thread gets its own reporter and tag pool rather than sharing one instance.
+ */
+@State(Scope.Thread)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Warmup(iterations = 5, time = 2)
+@Measurement(iterations = 5, time = 2)
+@Fork(3)
+@Threads(8)
+public class CardinalityLimitReporterBenchmark {
+
+  private static final int DISTINCT_TAGS = 32;
+
+  private CardinalityLimitReporter reporter;
+  private String[] tags;
+  private int cursor;
+
+  @Setup(Level.Trial)
+  public void setup() {
+    this.reporter = new CardinalityLimitReporter();
+    this.tags = new String[DISTINCT_TAGS];
+    for (int i = 0; i < DISTINCT_TAGS; i++) {
+      tags[i] = "tag-" + i;
+    }
+    // Pre-populate every entry so the measured path is pure lookup + update, not creation.
+    for (String tag : tags) {
+      reporter.record(tag, 1);
+    }
+  }
+
+  @Benchmark
+  public void record() {
+    String tag = tags[cursor++ & (DISTINCT_TAGS - 1)];
+    long count = 1L + (ThreadLocalRandom.current().nextLong() & 0xFF);
+    reporter.record(tag, count);
+  }
+}

dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java

@@ -244,6 +244,14 @@ public int getHitCount() {
     return hitCount;
   }
 
+  /**
+   * {@code true} if nothing hit this entry in the current reporting cycle, making it the first
+   * thing worth evicting when the table is full.
+   */
+  public boolean isStale() {
+    return hitCount == 0;
+  }
+
   public int getErrorCount() {
     return errorCount;
   }

dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateTable.java

@@ -2,7 +2,6 @@
 
 import datadog.trace.core.monitor.HealthMetrics;
 import datadog.trace.util.Hashtable;
-import datadog.trace.util.Hashtable.MutatingTableIterator;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
@@ -25,17 +24,9 @@
  */
 final class AggregateTable {
 
-  private final Hashtable.Entry[] buckets;
-  private final int maxAggregates;
-  private final AggregateEntry.Canonical canonical;
-  private int size;
+  private final Hashtable.State<AggregateEntry> state;
 
-  /**
-   * Bucket index where the last {@link #evictOneStale} successfully removed an entry. The next call
-   * resumes from this bucket so a fast-evicting workload doesn't repeatedly re-walk the same hot
-   * entries clustered near bucket 0. Reset to {@code 0} by {@link #clear}.
-   */
-  private int evictCursor;
+  private final AggregateEntry.Canonical canonical;
 
   AggregateTable(int maxAggregates) {
     this(maxAggregates, AdditionalTagsSchema.EMPTY);
@@ -47,118 +38,83 @@ final class AggregateTable {
 
   AggregateTable(
       int maxAggregates, CoreHandlers handlers, AdditionalTagsSchema additionalTagsSchema) {
-    this.buckets = Hashtable.Support.create(maxAggregates, Hashtable.Support.MAX_RATIO);
-    this.maxAggregates = maxAggregates;
+    this.state = Hashtable.createCapped(maxAggregates);
     this.canonical = new AggregateEntry.Canonical(handlers, additionalTagsSchema);
   }
 
   void resetCoreHandlers(HealthMetrics healthMetrics, CardinalityLimitReporter reporter) {
     canonical.handlers.reset(healthMetrics, reporter);
   }
 
+  /**
+   * Live aggregate count. Exact from this class's point of view: {@link Hashtable#estimateSize} is
+   * an estimate only across a reservation window, and {@link #findOrInsert} reserves and links
+   * without yielding, so no caller can observe one.
+   */
   int size() {
-    return size;
+    return Hashtable.estimateSize(state);
   }
 
   boolean isEmpty() {
-    return size == 0;
+    return Hashtable.isLikelyEmpty(state);
   }
 
   /**
    * Returns the {@link AggregateEntry} to update for {@code snapshot}, lazily creating one on miss.
    * Returns {@code null} when the table is at capacity and no stale entry can be evicted -- the
-   * caller should drop the data point in that case.
+   * caller should drop the data point in that case (reported via {@code onStatsAggregateDropped}).
+   * Dropping the new key rather than evicting an established one is deliberate: the cap is sized to
+   * the steady-state working set, so a full table of entries that were all used this cycle means
+   * the new key is the outlier.
+   *
+   * <p>Cardinality limiting (see {@link MetricCardinalityLimits#USE_BLOCKED_SENTINEL}) reduces how
+   * often eviction fires but doesn't eliminate it. Over-cap values for a single field collapse into
+   * the shared {@code tracer_blocked_value} sentinel, so no one field can fill the table on its
+   * own. But distinct in-budget combinations across fields (resource x service x operation x ...)
+   * can still drive the entry count to {@code maxAggregates}, so eviction remains the backstop.
+   *
+   * <p>The scan that finds a stale entry, and its resume-where-it-left-off amortization, live in
+   * {@link Hashtable#tryReserveOrEvict} -- this class only supplies {@link AggregateEntry#isStale}.
    */
   AggregateEntry findOrInsert(SpanSnapshot snapshot) {
     canonical.populateFrom(snapshot);
     long keyHash = canonical.keyHash;
-    for (AggregateEntry candidate = Hashtable.Support.bucket(buckets, keyHash);
+    for (AggregateEntry candidate = Hashtable.bucketFor(state, keyHash);
         candidate != null;
         candidate = candidate.next()) {
       if (candidate.keyHash == keyHash && canonical.matches(candidate)) {
         return candidate;
       }
     }
-    // Miss path.
-    if (size >= maxAggregates && !evictOneStale()) {
+    // Miss path. Reserve before building the entry so a refused insert costs no allocation; the
+    // reservation evicts a stale entry to make room if the table is already full.
+    if (!Hashtable.tryReserveOrEvict(state, AggregateEntry::isStale)) {
       return null;
     }
     AggregateEntry entry = canonical.createEntry();
-    Hashtable.Support.insertHeadEntry(buckets, keyHash, entry);
-    size++;
+    Hashtable.insertReserved(state, keyHash, entry);
     return entry;
   }
 
-  /**
-   * Unlinks the first entry whose {@code getHitCount() == 0}, resuming the scan from {@link
-   * #evictCursor} so consecutive evictions amortize to O(1) per call. Worst case for a single call
-   * is still O(N) when nearly every entry is hot, but a sustained eviction stream never re-scans
-   * the hot prefix more than twice across N evictions.
-   *
-   * <p>If the table is full and every entry was used in this cycle, drop the new key (reported via
-   * {@code onStatsAggregateDropped}) rather than evicting an established one. Cap is sized to the
-   * steady-state working set, so eviction is rare in the common case.
-   *
-   * <p>Cardinality limiting (see {@link MetricCardinalityLimits#USE_BLOCKED_SENTINEL}) reduces how
-   * often this fires but doesn't eliminate it. Over-cap values for a single field collapse into the
-   * shared {@code tracer_blocked_value} sentinel, so no one field can fill the table on its own.
-   * But distinct in-budget combinations across fields (resource x service x operation x ...) can
-   * still drive the entry count to {@code maxAggregates}, so this cursor-resumed scan remains the
-   * backstop.
-   */
-  private boolean evictOneStale() {
-    // Two passes -- [cursor, length) then [0, cursor) -- using the half-open-range iterator. The
-    // second pass is naturally empty when cursor==0, so no extra check needed.
-    return evictOneStaleInRange(evictCursor, buckets.length)
-        || evictOneStaleInRange(0, evictCursor);
-  }
-
-  /** Scans {@code [startBucket, endBucket)} for the first stale entry and unlinks it. */
-  private boolean evictOneStaleInRange(int startBucket, int endBucket) {
-    MutatingTableIterator<AggregateEntry> iter =
-        Hashtable.Support.mutatingTableIterator(buckets, startBucket, endBucket);
-    while (iter.hasNext()) {
-      AggregateEntry e = iter.next();
-      if (e.getHitCount() == 0) {
-        int bucket = iter.currentBucket();
-        iter.remove();
-        size--;
-        evictCursor = bucket;
-        return true;
-      }
-    }
-    return false;
-  }
-
   void forEach(Consumer<AggregateEntry> consumer) {
-    Hashtable.Support.forEach(buckets, consumer);
+    Hashtable.forEach(state, consumer);
   }
 
   /**
    * Context-passing forEach. Useful for callers that want to avoid a capturing-lambda allocation on
    * each invocation -- pass a non-capturing {@link BiConsumer} (typically a {@code static final})
    * plus whatever side-band state it needs as {@code context}.
    */
-  <T> void forEach(T context, BiConsumer<T, AggregateEntry> consumer) {
-    Hashtable.Support.forEach(buckets, context, consumer);
+  <C> void forEach(C context, BiConsumer<C, AggregateEntry> consumer) {
+    Hashtable.forEach(state, context, consumer);
   }
 
   /** Removes entries whose {@code getHitCount() == 0}. */
   void expungeStaleAggregates() {
-    for (MutatingTableIterator<AggregateEntry> iter =
-            Hashtable.Support.mutatingTableIterator(buckets);
-        iter.hasNext(); ) {
-      AggregateEntry e = iter.next();
-      if (e.getHitCount() == 0) {
-        iter.remove();
-        size--;
-      }
-    }
+    Hashtable.evictAll(state, AggregateEntry::isStale);
   }
 
   void clear() {
-    Hashtable.Support.clear(buckets);
-    size = 0;
-    evictCursor = 0;
+    Hashtable.clear(state);
   }
 }

dd-trace-core/src/main/java/datadog/trace/common/metrics/CardinalityLimitReporter.java

@@ -32,9 +32,10 @@ final class CardinalityLimitReporter {
 
   // Distinct blocked tag names in a window: 9 property fields + the configured peer tags + up to
   // AdditionalTagsSchema.MAX_ADDITIONAL_TAG_KEYS + base.service, with headroom for the brief
-  // overlap
-  // of old and new peer names across a schema rebuild. Fixed capacity; the table chains on overflow
-  // rather than dropping, so an underestimate only adds chain depth on this cold path.
+  // overlap of old and new peer names across a schema rebuild. Fixed, strict-cap capacity: if this
+  // is ever underestimated, excess distinct tags are silently dropped from the summary rather than
+  // recorded (see the null-check in record()) -- this is a cold, best-effort logging path, not a
+  // correctness-sensitive one.
   private static final int TAG_CAPACITY = 64;
 
   // Rough width of one "<tag>=<count>, " entry, used to pre-size the summary builder. Cold path, so
@@ -43,7 +44,8 @@ final class CardinalityLimitReporter {
 
   private final RatelimitedLogger rlLog;
   // Tag name -> blocked count accumulated since the last emitted summary.
-  private final Hashtable.D1<String, TagBlockEntry> blockedByTag = new Hashtable.D1<>(TAG_CAPACITY);
+  private final Hashtable.D1<String, TagBlockEntry> blockedByTag =
+      Hashtable.D1.createCapped(TagBlockEntry.class, TAG_CAPACITY);
 
   CardinalityLimitReporter() {
     this(new RatelimitedLogger(log, 5, MINUTES));
@@ -53,10 +55,15 @@ final class CardinalityLimitReporter {
     this.rlLog = rlLog;
   }
 
-  /** Records {@code count} values blocked for {@code tag} in the current reporting cycle. */
+  /**
+   * Records {@code count} values blocked for {@code tag} in the current reporting cycle.
+   *
+   * <p>A refused create -- the tag table is itself at capacity -- is deliberately ignored: this is
+   * a log sink, and the durable counts still reach {@code onTagCardinalityBlocked}.
+   */
   void record(String tag, long count) {
     if (count > 0) {
-      blockedByTag.getOrCreate(tag, TagBlockEntry::new).count += count;
+      blockedByTag.tryGetOrCreate(tag, TagBlockEntry::new).update(count, TagBlockEntry::inc);
     }
   }
 
@@ -101,5 +108,9 @@ private static final class TagBlockEntry extends Hashtable.D1.Entry<String> {
     TagBlockEntry(String tag) {
       super(tag);
     }
+
+    void inc(long n) {
+      count += n;
+    }
   }
 }

internal-api/src/jmh/java/datadog/trace/api/TagMapAccessBenchmark.java

@@ -1,5 +1,6 @@
 package datadog.trace.api;
 
+import datadog.trace.util.BenchmarkUtils;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -56,6 +57,31 @@
  * TagMapAccessBenchmark.insert_hashMap_builderStyle  thrpt    5  28057827.189 ± 1359655.664  ops/s
  * TagMapAccessBenchmark.insert_via_ledger            thrpt    5  41169656.095 ±  773264.754  ops/s
  * </code>
+ *
+ * <p>Rerun on JDK 8 with a new top-level {@code @Setup(Level.Trial)} calling {@link
+ * BenchmarkUtils#polluteHashDispatch()} (this file had none before). M ops/s, 8 threads:
+ *
+ * <pre>{@code
+ * getEntry                        83   getObject                    87
+ * insert                          37   insert_hashMap                48
+ * insert_hashMap_builderStyle     20   insert_via_ledger             37*
+ * }</pre>
+ *
+ * <p>* = error bar about a quarter of the mean at {@code @Fork(2)} — directional only.
+ *
+ * <p>Every number here is 9-29% below the Java 17 table above, with no clear split between the
+ * TagMap paths and the HashMap paths this pollution should affect. The table above is Java 17; this
+ * rerun is JDK 8, whose C2 backend for Apple Silicon (AArch64) is far less mature than JDK 17+'s —
+ * a broad-based slowdown across every entry is expected from that JDK gap alone, independent of
+ * pollution — the same JDK-crossing explanation applies to {@link
+ * datadog.trace.util.CaseInsensitiveMapBenchmark}'s rerun. ({@link
+ * datadog.trace.util.HashtableD1Benchmark} and {@link datadog.trace.util.HashtableD2Benchmark} saw
+ * a similar broad drop despite holding the JDK constant — that one is same-session run-to-run
+ * noise, not a JDK effect.) The relative story survives: {@code insert_hashMap} (48M) still beats
+ * {@code insert} (37M) for plain insertion, and {@code insert_via_ledger} (37M) still clearly beats
+ * the HashMap builder-style path (20M); {@code insert_via_ledger} landing roughly level with {@code
+ * insert} here (vs. clearly behind it in the table above) is within that path's own wide error bar,
+ * not a new finding.
  */
 @BenchmarkMode(Mode.Throughput)
 @OutputTimeUnit(TimeUnit.SECONDS)
@@ -99,6 +125,11 @@ public class TagMapAccessBenchmark {
    * Pre-populated read map, PER-THREAD ({@code Scope.Thread}): each thread owns its own map so
    * reads don't contend on shared mutable state under {@code @Threads(8)}.
    */
+  @Setup(Level.Trial)
+  public void setUp() {
+    BenchmarkUtils.polluteHashDispatch();
+  }
+
   @State(Scope.Thread)
   public static class ReadMap {
     TagMap map;