Use the unified Hashtable API in client-side stats

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

@@ -244,6 +244,17 @@ 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. Encapsulates the staleness rule on the entry so
+   * the table doesn't have to know it is spelled {@code hitCount == 0}, and reads as {@code
+   * AggregateEntry::isStale} at an eviction call site -- an unbound method reference, so it is
+   * non-capturing and costs no allocation.
+   */
+  public boolean isStale() {
+    return hitCount == 0;
+  }
+
   public int getErrorCount() {
     return errorCount;
   }

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

@@ -2,9 +2,9 @@
 
 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;
+import java.util.function.Predicate;
 
 /**
  * The {@link AggregateEntry} store of the consuming aggregator thread, keyed on the canonical
@@ -25,17 +25,15 @@
  */
 final class AggregateTable {
 
-  private final Hashtable.Entry[] buckets;
-  private final int maxAggregates;
-  private final AggregateEntry.Canonical canonical;
-  private int size;
-
   /**
-   * 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}.
+   * Stale means "not used in this reporting cycle". Held as a {@code static final} so it is a
+   * non-capturing singleton rather than a fresh lambda per eviction.
    */
-  private int evictCursor;
+  private static final Predicate<AggregateEntry> STALE = AggregateEntry::isStale;
+
+  private final Hashtable.State<AggregateEntry> state;
+
+  private final AggregateEntry.Canonical canonical;
 
   AggregateTable(int maxAggregates) {
     this(maxAggregates, AdditionalTagsSchema.EMPTY);
@@ -47,118 +45,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 #STALE}.
    */
   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, STALE)) {
       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, STALE);
   }
 
   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

@@ -4,6 +4,7 @@
 
 import datadog.logging.RatelimitedLogger;
 import datadog.trace.util.Hashtable;
+import java.util.function.ObjLongConsumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,13 +56,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 {@code false} return -- 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.tryGetOrUpdate(tag, TagBlockEntry::new, count, ADD_BLOCKED);
     }
   }
 
@@ -100,6 +103,9 @@ private String summarize() {
   /**
    * Single-key counter entry: the tag name (via {@link #key()}) plus its in-place-mutated count.
    */
+  /** Non-capturing, so {@link #record} allocates nothing per call. */
+  private static final ObjLongConsumer<TagBlockEntry> ADD_BLOCKED = (entry, n) -> entry.count += n;
+
   private static final class TagBlockEntry extends Hashtable.D1.Entry<String> {
     long count;