Use the unified Hashtable API in client-side stats

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

@@ -55,13 +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) {
-      TagBlockEntry entry = blockedByTag.tryGetOrCreate(tag, TagBlockEntry::new);
-      if (entry != null) {
-        entry.count += count;
-      }
+      blockedByTag.tryGetOrCreate(tag, TagBlockEntry::new).update(count, TagBlockEntry::inc);
     }
   }
 
@@ -106,5 +108,9 @@ private static final class TagBlockEntry extends Hashtable.D1.Entry<String> {
     TagBlockEntry(String tag) {
       super(tag);
     }
+
+    void inc(long n) {
+      count += n;
+    }
   }
 }