diff --git a/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapBenchmark.java deleted file mode 100644 index 63ccb734e9c..00000000000 --- a/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapBenchmark.java +++ /dev/null @@ -1,293 +0,0 @@ -package datadog.trace.util; - -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentSkipListMap; -import java.util.function.Supplier; -import org.openjdk.jmh.annotations.Benchmark; -import org.openjdk.jmh.annotations.Fork; -import org.openjdk.jmh.annotations.Measurement; -import org.openjdk.jmh.annotations.Scope; -import org.openjdk.jmh.annotations.State; -import org.openjdk.jmh.annotations.Threads; -import org.openjdk.jmh.annotations.Warmup; - -/** - * - * - *

- * - *

In most situations in dd-java-agent, ConcurrentMaps are not necessarily needed and incur - * additional overhead. ConcurrentMaps make sense when concurrent writers are likely. - * - *

If a Map can be created atomically in one thread and then stored into a volatile, that is the - * preferred solution. For example, requesting an update from agent / API and then exposing to the - * rest of the tracer via a global. - * - *

If a Map needs to be written in a thread-safe manner, but is primarily accessed from one - * thread at a time, then a synchronized HashMap is usually the best option. - * MacBook M1 with 1 thread (Java 21) - * - * Benchmark Mode Cnt Score Error Units - * ThreadSafeMapBenchmark.create_concHashMap thrpt 6 8081979.153 ± 261559.222 ops/s - * ThreadSafeMapBenchmark.create_concSkipListMap thrpt 6 2998832.124 ± 103708.038 ops/s - * ThreadSafeMapBenchmark.create_hashMap thrpt 6 24938311.610 ± 673725.902 ops/s - * ThreadSafeMapBenchmark.create_hashMap_synchronized thrpt 6 7971740.607 ± 121986.296 ops/s - * - * ThreadSafeMapBenchmark.get_concHashMap thrpt 6 173942565.340 ± 12003493.448 ops/s - * ThreadSafeMapBenchmark.get_concSkipListMap thrpt 6 79230298.061 ± 13007895.765 ops/s - * ThreadSafeMapBenchmark.get_hashMap_synchronized thrpt 6 98056657.832 ± 3413815.061 ops/s - * ThreadSafeMapBenchmark.get_hashMap_volatile thrpt 6 210511753.596 ± 5017502.317 ops/s - * - * MacBook M1 with 8 threads (Java 21) - * - * Benchmark Mode Cnt Score Error Units - * ThreadSafeMapBenchmark.create_concHashMap thrpt 6 58015351.219 ± 6201384.867 ops/s - * ThreadSafeMapBenchmark.create_concSkipListMap thrpt 6 19296105.790 ± 4516587.751 ops/s - * ThreadSafeMapBenchmark.create_hashMap thrpt 6 147917381.815 ± 22901897.589 ops/s - * ThreadSafeMapBenchmark.create_hashMap_synchronized thrpt 6 56466354.962 ± 13202034.783 ops/s - * - * ThreadSafeMapBenchmark.get_concHashMap thrpt 6 849986442.797 ± 14499355.893 ops/s - * ThreadSafeMapBenchmark.get_concSkipListMap thrpt 6 26828246.629 ± 2772377.532 ops/s - * ThreadSafeMapBenchmark.get_hashMap_synchronized thrpt 6 20123419.604 ± 4858466.787 ops/s - * ThreadSafeMapBenchmark.get_hashMap_volatile thrpt 6 286024211.995 ± 114449056.603 ops/s - * - */ -@Fork(2) -@Warmup(iterations = 2) -@Measurement(iterations = 3) -@Threads(8) -@State(Scope.Thread) -public class ThreadSafeMapBenchmark { - static final String[] INSERTION_KEYS = { - "foo", "bar", "baz", "quux", "foobar", "foobaz", "key0", "key1", "key2", "key3" - }; - - static final String[] EQUAL_KEYS = - init( - () -> { - String[] keys = new String[INSERTION_KEYS.length]; - for (int i = 0; i < INSERTION_KEYS.length; ++i) { - keys[i] = new String(INSERTION_KEYS[i]); - } - return keys; - }); - - static T init(Supplier supplier) { - return supplier.get(); - } - - // Per-thread (@State(Scope.Thread)) so cycling the lookup key doesn't contend a shared counter. - // The maps below stay static/shared (the point — concurrent reads of one map); only the index is - // per-thread. A shared counter's cache-line ping-pong would otherwise floor the fastest reads - // (e.g. FlatHashtable's lock-free probe), hiding exactly the differences this benchmark compares. - int lookupIndex = 0; - - String nextLookupKey() { - return nextLookupKey(EQUAL_KEYS); - } - - String nextLookupKey(String[] keys) { - int localIndex = ++lookupIndex; - if (localIndex >= keys.length) { - lookupIndex = localIndex = 0; - } - return keys[localIndex]; - } - - static void fill(Map map) { - for (int i = 0; i < INSERTION_KEYS.length; ++i) { - map.put(INSERTION_KEYS[i], i); - } - } - - // FlatHashtable's contribution here is the lock-free concurrent read: get() is a plain array - // probe - // with no lock and no volatile — safe under concurrency because the table is published once (a - // final static field) and each entry's identity fields are final. (Fixture mirrors the one in - // SingleThreadedMapBenchmark; the benchmarks are self-contained.) - static final class IntEntry { - final String key; - final int value; - - IntEntry(String key, int value) { - this.key = key; - this.value = value; - } - } - - static final class IntEntryKeyStrategy extends FlatHashtable.EntryStrategy { - static final IntEntryKeyStrategy INSTANCE = new IntEntryKeyStrategy(); - - private IntEntryKeyStrategy() {} - - @Override - public boolean matches(IntEntry entry, String key) { - return key.equals(entry.key); - } - - @Override - public long hashOf(IntEntry entry) { - return entry.key.hashCode(); // consistent with the default hashKey - } - } - - // --- CHA-defeat decoys --------------------------------------------------------------------- - // These are never used to build a table; they exist only to be *loaded* (see CHA_DEFEAT), so - // MatchingStrategy.matches and .hashKey each have >=2 concrete implementors. That denies C2 the - // single-implementor CHA devirtualization of matchStrat.hashKey/matches inside get(). If the - // strategy calls still inline afterward, the win is structural (the constant INSTANCE's exact - // type propagated through the inlined get), not a CHA bet that would deopt on a second subclass. - - // Second matches impl -> MatchingStrategy.matches is polymorphic. - static final class DecoyMatchStrategy extends FlatHashtable.EntryStrategy { - static final DecoyMatchStrategy INSTANCE = new DecoyMatchStrategy(); - - private DecoyMatchStrategy() {} - - @Override - public boolean matches(IntEntry entry, String key) { - return key == entry.key; // deliberately different body from IntEntryKeyStrategy - } - - @Override - public long hashOf(IntEntry entry) { - return entry.key.hashCode(); - } - } - - // Overrides hashKey -> MatchingStrategy.hashKey is polymorphic too (default + this override). - static final class DecoyHashKeyStrategy extends FlatHashtable.EntryStrategy { - static final DecoyHashKeyStrategy INSTANCE = new DecoyHashKeyStrategy(); - - private DecoyHashKeyStrategy() {} - - @Override - public long hashKey(String key) { - return key.length(); - } - - @Override - public boolean matches(IntEntry entry, String key) { - return key.equals(entry.key); - } - - @Override - public long hashOf(IntEntry entry) { - return entry.key.length(); - } - } - - // Referenced only so these three concrete implementors load at benchmark class-init, before the - // hot method compiles — see the CHA-defeat note above. - @SuppressWarnings("unused") - static final Object[] CHA_DEFEAT = { - IntEntryKeyStrategy.INSTANCE, DecoyMatchStrategy.INSTANCE, DecoyHashKeyStrategy.INSTANCE - }; - - static IntEntry[] _create_flat() { - // Sized to the key count (FlatHashtable is fixed-capacity, no resize): load factor <= 0.5. - IntEntry[] table = FlatHashtable.create(IntEntry.class, INSERTION_KEYS.length); - for (int i = 0; i < INSERTION_KEYS.length; ++i) { - FlatHashtable.insert(table, new IntEntry(INSERTION_KEYS[i], i), IntEntryKeyStrategy.INSTANCE); - } - return table; - } - - static final HashMap _create_hashMap() { - HashMap map = new HashMap<>(); - fill(map); - return map; - } - - @Benchmark - public Map create_hashMap() { - return _create_hashMap(); - } - - static volatile HashMap VOLATILE_HASH_MAP = _create_hashMap(); - - @Benchmark - public Integer get_hashMap_volatile() { - Map map = VOLATILE_HASH_MAP; - return map.get(nextLookupKey()); - } - - static final Map _create_hashMap_synchronized() { - Map map = Collections.synchronizedMap(new HashMap<>()); - fill(map); - return map; - } - - @Benchmark - public Map create_hashMap_synchronized() { - return _create_hashMap_synchronized(); - } - - static final Map SYNC_HASH_MAP = _create_hashMap_synchronized(); - - @Benchmark - public Integer get_hashMap_synchronized() { - return SYNC_HASH_MAP.get(nextLookupKey()); - } - - static ConcurrentHashMap _create_concHashMap() { - ConcurrentHashMap map = new ConcurrentHashMap<>(); - fill(map); - return map; - } - - @Benchmark - public ConcurrentHashMap create_concHashMap() { - return _create_concHashMap(); - } - - static final ConcurrentHashMap CONC_HASH_MAP = _create_concHashMap(); - - @Benchmark - public Integer get_concHashMap() { - return CONC_HASH_MAP.get(nextLookupKey()); - } - - static ConcurrentSkipListMap _create_concSkipListMap() { - ConcurrentSkipListMap map = new ConcurrentSkipListMap<>(); - fill(map); - return map; - } - - @Benchmark - public ConcurrentSkipListMap create_concSkipListMap() { - return _create_concSkipListMap(); - } - - static final ConcurrentSkipListMap CONC_SKIP_LIST_MAP = - _create_concSkipListMap(); - - @Benchmark - public Integer get_concSkipListMap() { - return CONC_SKIP_LIST_MAP.get(nextLookupKey()); - } - - @Benchmark - public IntEntry[] create_flatHashtable() { - return _create_flat(); - } - - static final IntEntry[] FLAT_TABLE = _create_flat(); - - @Benchmark - public IntEntry get_flatHashtable() { - // Lock-free concurrent read of the shared, once-published table. - return FlatHashtable.get(FLAT_TABLE, nextLookupKey(), IntEntryKeyStrategy.INSTANCE); - } -} diff --git a/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapCounterBenchmark.java b/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapCounterBenchmark.java new file mode 100644 index 00000000000..a78a66f6672 --- /dev/null +++ b/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapCounterBenchmark.java @@ -0,0 +1,152 @@ +package datadog.trace.util; + +import static java.util.concurrent.TimeUnit.MICROSECONDS; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; +import java.util.concurrent.atomic.LongAdder; +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; + +/** + * Benchmarks the "find and increment" pattern: look up an entry by key, then atomically increment + * its counter. Models per-class or per-method hit counters in the tracer. + * + *

The key insight is that {@link ConcurrentHashtable.D1} allows the counter to be embedded + * directly in the entry as a {@code volatile long} updated via {@link AtomicLongFieldUpdater}, + * avoiding the extra object allocation that {@link ConcurrentHashMap} requires when pairing each + * key with an {@link AtomicLong} or {@link LongAdder}. + * + *

Strategies compared: + * + *

+ * + *

Key identity. Lookups reuse the same interned {@code KEYS} instances used to populate + * the table, so they hit the {@code ==} identity fast path rather than {@code equals()}. This is + * deliberate and realistic for the tracer, whose keys are typically interned string literals + * (tag-name constants); it is not an oversight. + * + *

Java 17 results ({@code @Fork(2)}, {@code @Threads(8)}, 64 pre-populated keys): + * + *

{@code
+ * Benchmark                          Score   Units
+ * increment_longAdder                   79   ops/us  (fastest)
+ * increment_atomicLong                  71   ops/us
+ * increment_concurrentHashtable         69   ops/us
+ * }
+ * + *

Key findings: + * + *

+ */ +@Fork(2) +@Warmup(iterations = 2) +@Measurement(iterations = 3) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(MICROSECONDS) +@Threads(8) +public class ThreadSafeMapCounterBenchmark { + + static final int N_KEYS = 64; + static final int CAPACITY = 128; + + static final String[] KEYS = new String[N_KEYS]; + + static { + for (int i = 0; i < N_KEYS; ++i) { + KEYS[i] = "key-" + i; + } + } + + static final class CounterEntry extends ConcurrentHashtable.D1.Entry { + private static final AtomicLongFieldUpdater COUNT = + AtomicLongFieldUpdater.newUpdater(CounterEntry.class, "count"); + + volatile long count; + + CounterEntry(String key) { + super(key); + } + + long increment() { + return COUNT.incrementAndGet(this); + } + } + + /** + * Shared state ({@link Scope#Benchmark}): one instance of each map across all threads, modelling + * a shared instrumentation counter table. + */ + @State(Scope.Benchmark) + public static class SharedState { + ConcurrentHashtable.D1 table; + ConcurrentHashMap atomicLongMap; + ConcurrentHashMap longAdderMap; + + @Setup(Level.Iteration) + public void setUp() { + table = ConcurrentHashtable.D1.createCapped(CounterEntry.class, CAPACITY); + atomicLongMap = new ConcurrentHashMap<>(CAPACITY); + longAdderMap = new ConcurrentHashMap<>(CAPACITY); + for (int i = 0; i < N_KEYS; ++i) { + table.tryGetOrCreateOrNull(KEYS[i], CounterEntry::new); + atomicLongMap.put(KEYS[i], new AtomicLong()); + longAdderMap.put(KEYS[i], new LongAdder()); + } + } + } + + /** Per-thread cursor so each thread cycles through keys independently. */ + @State(Scope.Thread) + public static class ThreadState { + int cursor; + + int next() { + int i = cursor; + cursor = (i + 1) & (N_KEYS - 1); + return i; + } + } + + @Benchmark + public long increment_concurrentHashtable(SharedState s, ThreadState t) { + return s.table.get(KEYS[t.next()]).increment(); + } + + @Benchmark + public long increment_atomicLong(SharedState s, ThreadState t) { + return s.atomicLongMap.get(KEYS[t.next()]).incrementAndGet(); + } + + @Benchmark + public void increment_longAdder(SharedState s, ThreadState t) { + s.longAdderMap.get(KEYS[t.next()]).increment(); + } +} diff --git a/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapD1Benchmark.java b/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapD1Benchmark.java new file mode 100644 index 00000000000..fcf5b07c433 --- /dev/null +++ b/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapD1Benchmark.java @@ -0,0 +1,198 @@ +package datadog.trace.util; + +import static java.util.concurrent.TimeUnit.MICROSECONDS; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListMap; +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; + +/** + * Compares thread-safe map strategies for shared, concurrent single-key lookups. + * + *

See {@link ThreadSafeMapD2Benchmark} for the composite-key variant, which adds the cost of + * hashing two keys and a wrapper object allocation for map-based alternatives. + * + *

The table is shared across all threads ({@link Scope#Benchmark}) and pre-populated before the + * measurement iteration — modelling the steady-state read-mostly pattern that the tracer uses (a + * per-class or per-method instrumentation cache consulted on every invocation). + * + *

Strategies compared: + * + *

    + *
  • {@link ConcurrentHashtable.D1} — lock-free reads, no extra allocation per lookup. + *
  • {@link ConcurrentHashMap} — striped locking; the key is the string itself, no wrapper. + *
  • {@link ConcurrentSkipListMap} — fully lock-free (CAS), but pays tree traversal and {@link + * Comparable} overhead on every operation. + *
  • {@link Collections#synchronizedMap} wrapping {@link HashMap} — global lock on every + * operation. Establishes the coarse-locking baseline. + *
+ * + *

Key identity. Lookups reuse the same interned {@code KEYS} instances used to populate + * the table, so they hit the {@code ==} identity fast path rather than {@code equals()}. This is + * deliberate and realistic for the tracer, whose map keys are typically interned string literals + * (tag-name constants); it is not an oversight. ({@code ImmutableMapBenchmark} covers the + * distinct-instance {@code equals()} path explicitly via its {@code _sameKey} vs default variants.) + * + *

Java 17 results ({@code @Fork(2)}, {@code @Threads(8)}, 64 pre-populated keys): + * + *

{@code
+ * Benchmark                             Score   Units
+ * get_concurrentHashtable               1583   ops/us  (fastest)
+ * get_concurrentHashMap                 1145   ops/us
+ * get_concurrentSkipListMap              170   ops/us
+ * get_synchronizedHashMap                 33   ops/us
+ *
+ * getOrCreate_concurrentHashtable       1450   ops/us  (fastest)
+ * getOrCreate_concurrentHashMap         1125   ops/us
+ * getOrCreate_synchronizedHashMap         31   ops/us
+ * }
+ * + *

Key findings: + * + *

    + *
  • {@code ConcurrentHashtable} is ~38% faster than {@code ConcurrentHashMap} on {@code get} + * (1583 vs 1145 ops/us); avoids the hash-to-segment translation CHM pays even on its fast + * path. + *
  • {@code ConcurrentSkipListMap} is ~9× slower than {@code ConcurrentHashMap} — tree traversal + * cost is high even under lock-free CAS. + *
  • Synchronized {@code HashMap} is ~47× slower than {@code ConcurrentHashtable}; the global + * lock serializes all 8 threads. + *
  • {@code getOrCreate} is near-identical to {@code get} because all keys are pre-populated — + * the lock branch is never taken during measurement. + *
+ */ +@Fork(2) +@Warmup(iterations = 2) +@Measurement(iterations = 3) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(MICROSECONDS) +@Threads(8) +public class ThreadSafeMapD1Benchmark { + + static final int N_KEYS = 64; + static final int CAPACITY = 128; + + static final String[] KEYS = new String[N_KEYS]; + + static { + for (int i = 0; i < N_KEYS; ++i) { + KEYS[i] = "key-" + i; + } + } + + static final class D1Entry extends ConcurrentHashtable.D1.Entry { + final long value; + + D1Entry(String key) { + super(key); + this.value = 1L; + } + } + + /** + * Shared state ({@link Scope#Benchmark}): one instance of each map across all threads, modelling + * a shared instrumentation cache. + */ + @State(Scope.Benchmark) + public static class SharedState { + ConcurrentHashtable.D1 table; + ConcurrentHashMap concurrentHashMap; + ConcurrentSkipListMap skipListMap; + Map synchronizedHashMap; + + @Setup(Level.Iteration) + public void setUp() { + table = ConcurrentHashtable.D1.createCapped(D1Entry.class, CAPACITY); + concurrentHashMap = new ConcurrentHashMap<>(CAPACITY); + skipListMap = new ConcurrentSkipListMap<>(); + synchronizedHashMap = Collections.synchronizedMap(new HashMap<>(CAPACITY)); + for (int i = 0; i < N_KEYS; ++i) { + table.tryGetOrCreateOrNull(KEYS[i], D1Entry::new); + concurrentHashMap.put(KEYS[i], (long) i); + skipListMap.put(KEYS[i], (long) i); + synchronizedHashMap.put(KEYS[i], (long) i); + } + } + } + + /** Per-thread cursor so each thread cycles through keys independently. */ + @State(Scope.Thread) + public static class ThreadState { + int cursor; + + int next() { + int i = cursor; + cursor = (i + 1) & (N_KEYS - 1); + return i; + } + } + + @Benchmark + public D1Entry get_concurrentHashtable(SharedState s, ThreadState t) { + return s.table.get(KEYS[t.next()]); + } + + @Benchmark + public Long get_concurrentHashMap(SharedState s, ThreadState t) { + return s.concurrentHashMap.get(KEYS[t.next()]); + } + + @Benchmark + public Long get_concurrentSkipListMap(SharedState s, ThreadState t) { + return s.skipListMap.get(KEYS[t.next()]); + } + + @Benchmark + public Long get_synchronizedHashMap(SharedState s, ThreadState t) { + return s.synchronizedHashMap.get(KEYS[t.next()]); + } + + @Benchmark + public D1Entry getOrCreate_concurrentHashtable(SharedState s, ThreadState t) { + return s.table.tryGetOrCreateOrNull(KEYS[t.next()], D1Entry::new); + } + + /** + * get-first pattern for CHM — the idiomatic equivalent of D1.getOrCreate on a mostly-populated + * table. + */ + @Benchmark + public Long getOrCreate_concurrentHashMap(SharedState s, ThreadState t) { + String key = KEYS[t.next()]; + Long existing = s.concurrentHashMap.get(key); + if (existing != null) { + return existing; + } + return s.concurrentHashMap.computeIfAbsent(key, k -> 0L); + } + + /** + * get-first pattern for synchronized HashMap. On hit: one lock acquire/release for get. On miss: + * a second synchronized block for the double-checked put. + */ + @Benchmark + public Long getOrCreate_synchronizedHashMap(SharedState s, ThreadState t) { + String key = KEYS[t.next()]; + Long existing = s.synchronizedHashMap.get(key); + if (existing != null) { + return existing; + } + synchronized (s.synchronizedHashMap) { + return s.synchronizedHashMap.computeIfAbsent(key, k -> 0L); + } + } +} diff --git a/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapD2Benchmark.java b/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapD2Benchmark.java new file mode 100644 index 00000000000..c5b9122ec13 --- /dev/null +++ b/internal-api/src/jmh/java/datadog/trace/util/ThreadSafeMapD2Benchmark.java @@ -0,0 +1,353 @@ +package datadog.trace.util; + +import static java.util.concurrent.TimeUnit.MICROSECONDS; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentSkipListMap; +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; + +/** + * Compares thread-safe map strategies for shared, concurrent composite-key lookups. + * + *

See {@link ThreadSafeMapD1Benchmark} for the single-key variant. + * + *

The table is shared across all threads ({@link Scope#Benchmark}) and pre-populated before the + * measurement iteration — modelling the steady-state read-mostly pattern that the tracer uses (a + * per-class or per-method instrumentation cache consulted on every invocation). + * + *

Strategies compared: + * + *

    + *
  • {@link ConcurrentHashtable.D2} — lock-free reads, no composite key allocation per lookup. + * K2 is {@link Integer} (boxed), so EA may still eliminate the box on hits, but the + * allocation is observable on misses. + *
  • {@link ConcurrentHashtable} building blocks (custom entry) — same lock-free read path, but + * K2 is a primitive {@code int} embedded directly in the entry. No boxing at any point; + * demonstrates the flexibility available when {@code D2}'s object-key constraint is too + * limiting. + *
  • {@link ConcurrentHashMap} — striped locking, allocates a {@link Key2} wrapper per lookup + * (boxes the {@code int} K2 inside). + *
  • {@link ConcurrentSkipListMap} — fully lock-free (CAS), but pays tree traversal and {@link + * Comparable} overhead; allocates {@link Key2} per lookup. {@code getOrCreate} uses + * get-then-{@code putIfAbsent} (no native {@code computeIfAbsent}). + *
  • {@link Collections#synchronizedMap} wrapping {@link HashMap} — global lock on every + * operation; allocates {@link Key2} per lookup. Establishes the coarse-locking baseline. + *
+ * + *

Key identity. Lookups reuse the same interned {@code SOURCE_K1} strings and cached + * {@code SOURCE_K2} Integers used to populate the table, so the key-part comparisons hit the {@code + * ==} identity fast path rather than {@code equals()}. This is deliberate and realistic for the + * tracer, whose keys are typically interned literals (tag-name constants) and small boxed ints; it + * is not an oversight. + * + *

Java 17 results ({@code @Fork(2)}, {@code @Threads(8)}, 64 pre-populated keys): + * + *

{@code
+ * Benchmark                              Score   Units
+ * get_concurrentHashtable                1452   ops/us  (tied fastest)
+ * get_support                            1450   ops/us  (primitive int K2)
+ * get_concurrentHashMap                   777   ops/us  (allocates Key2 wrapper)
+ * get_concurrentSkipListMap               146   ops/us
+ * get_synchronizedHashMap                  27   ops/us
+ *
+ * getOrCreate_support                    1379   ops/us  (fastest)
+ * getOrCreate_concurrentHashtable        1119   ops/us
+ * getOrCreate_concurrentHashMap           769   ops/us
+ * getOrCreate_concurrentSkipListMap       151   ops/us
+ * getOrCreate_synchronizedHashMap          28   ops/us
+ * }
+ * + *

Key findings: + * + *

    + *
  • {@code ConcurrentHashtable} and {@code Support} are neck-and-neck on {@code get} (1452 vs + * 1450 ops/us); both avoid the {@link Key2} wrapper allocation that {@code ConcurrentHashMap} + * requires on every lookup. + *
  • {@code ConcurrentHashMap} is ~2× slower than {@code ConcurrentHashtable} on {@code get} + * (777 vs 1452 ops/us) — the {@link Key2} allocation plus two-level hash lookup adds up. + *
  • {@code Support} shows slightly higher {@code getOrCreate} throughput than {@code D2} (1379 + * vs 1119 ops/us) because its primitive {@code int} K2 field avoids boxing inside the entry + * match on the write-path re-check. + *
  • {@code ConcurrentSkipListMap} is ~5× slower than {@code ConcurrentHashMap} due to tree + * traversal; the two-traversal {@code getOrCreate} pattern adds further overhead on misses. + *
  • Synchronized {@code HashMap} is ~50× slower than {@code ConcurrentHashtable}. + *
+ */ +@Fork(2) +@Warmup(iterations = 2) +@Measurement(iterations = 3) +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(MICROSECONDS) +@Threads(8) +public class ThreadSafeMapD2Benchmark { + + static final int N_KEYS = 64; + static final int CAPACITY = 128; + + static final String[] SOURCE_K1 = new String[N_KEYS]; + static final Integer[] SOURCE_K2 = new Integer[N_KEYS]; + static final int[] SOURCE_K2_INT = new int[N_KEYS]; + + static { + for (int i = 0; i < N_KEYS; ++i) { + SOURCE_K1[i] = "key-" + i; + SOURCE_K2_INT[i] = i * 31 + 17; + SOURCE_K2[i] = SOURCE_K2_INT[i]; + } + } + + static final class D2Entry extends ConcurrentHashtable.D2.Entry { + final long value; + + D2Entry(String k1, Integer k2) { + super(k1, k2); + this.value = 1L; + } + } + + /** + * Support-based entry with a primitive {@code int} K2 — no boxing at any point. The hash is + * computed with the same formula as {@link Hashtable.D2.Entry#hash} but avoids the {@link + * Integer#hashCode(int)} boxing path by calling {@link LongHashingUtils} directly. + */ + static final class SupportEntry extends ConcurrentHashtable.Entry { + final String k1; + final int k2; + final long value; + + SupportEntry(String k1, int k2) { + super(hash(k1, k2)); + this.k1 = k1; + this.k2 = k2; + this.value = 1L; + } + + static long hash(String k1, int k2) { + return LongHashingUtils.hash(k1.hashCode(), Integer.hashCode(k2)); + } + + boolean matches(String k1, int k2) { + return this.k2 == k2 && this.k1.equals(k1); + } + } + + /** Composite key for map-based baselines. */ + static final class Key2 implements Comparable { + final String k1; + final Integer k2; + final int hash; + + Key2(String k1, Integer k2) { + this.k1 = k1; + this.k2 = k2; + // Varargs-free hash: Objects.hash(k1, k2) would allocate an Object[] per key, penalizing the + // map baselines with an allocation the wrapper itself doesn't need and overstating the + // ConcurrentHashtable advantage this benchmark measures. + this.hash = 31 * k1.hashCode() + k2.hashCode(); + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof Key2)) { + return false; + } + Key2 other = (Key2) o; + return Objects.equals(k1, other.k1) && Objects.equals(k2, other.k2); + } + + @Override + public int hashCode() { + return hash; + } + + @Override + public int compareTo(Key2 other) { + int c = k1.compareTo(other.k1); + return c != 0 ? c : k2.compareTo(other.k2); + } + } + + /** + * Shared state ({@link Scope#Benchmark}): one instance of each map across all threads, modelling + * a shared instrumentation cache. + */ + @State(Scope.Benchmark) + public static class SharedState { + ConcurrentHashtable.D2 table; + java.util.concurrent.atomic.AtomicReferenceArray supportBuckets; + ConcurrentHashMap concurrentHashMap; + ConcurrentSkipListMap skipListMap; + Map synchronizedHashMap; + + @Setup(Level.Iteration) + public void setUp() { + table = ConcurrentHashtable.D2.createCapped(D2Entry.class, CAPACITY); + supportBuckets = ConcurrentHashtable.createFixedBuckets(SupportEntry.class, CAPACITY); + concurrentHashMap = new ConcurrentHashMap<>(CAPACITY); + skipListMap = new ConcurrentSkipListMap<>(); + synchronizedHashMap = Collections.synchronizedMap(new HashMap<>(CAPACITY)); + for (int i = 0; i < N_KEYS; ++i) { + int k2 = SOURCE_K2[i]; + table.tryGetOrCreateOrNull(SOURCE_K1[i], SOURCE_K2[i], D2Entry::new); + // populate support table + SupportEntry se = new SupportEntry(SOURCE_K1[i], k2); + synchronized (ConcurrentHashtable.getWriteLock(supportBuckets)) { + ConcurrentHashtable.insertHeadEntryFor(supportBuckets, se.keyHash, se); + } + Key2 key = new Key2(SOURCE_K1[i], SOURCE_K2[i]); + concurrentHashMap.put(key, (long) i); + skipListMap.put(key, (long) i); + synchronizedHashMap.put(key, (long) i); + } + } + } + + /** Per-thread cursor so each thread cycles through keys independently. */ + @State(Scope.Thread) + public static class ThreadState { + int cursor; + + int next() { + int i = cursor; + cursor = (i + 1) & (N_KEYS - 1); + return i; + } + } + + @Benchmark + public D2Entry get_concurrentHashtable(SharedState s, ThreadState t) { + int i = t.next(); + return s.table.get(SOURCE_K1[i], SOURCE_K2[i]); + } + + @Benchmark + public SupportEntry get_support(SharedState s, ThreadState t) { + int i = t.next(); + String k1 = SOURCE_K1[i]; + int k2 = SOURCE_K2_INT[i]; + long keyHash = SupportEntry.hash(k1, k2); + for (SupportEntry e = ConcurrentHashtable.bucketFor(s.supportBuckets, keyHash); + e != null; + e = e.next()) { + if (e.keyHash == keyHash && e.matches(k1, k2)) { + return e; + } + } + return null; + } + + @Benchmark + public Long get_concurrentHashMap(SharedState s, ThreadState t) { + int i = t.next(); + return s.concurrentHashMap.get(new Key2(SOURCE_K1[i], SOURCE_K2[i])); + } + + @Benchmark + public Long get_concurrentSkipListMap(SharedState s, ThreadState t) { + int i = t.next(); + return s.skipListMap.get(new Key2(SOURCE_K1[i], SOURCE_K2[i])); + } + + @Benchmark + public Long get_synchronizedHashMap(SharedState s, ThreadState t) { + int i = t.next(); + return s.synchronizedHashMap.get(new Key2(SOURCE_K1[i], SOURCE_K2[i])); + } + + @Benchmark + public D2Entry getOrCreate_concurrentHashtable(SharedState s, ThreadState t) { + int i = t.next(); + return s.table.tryGetOrCreateOrNull(SOURCE_K1[i], SOURCE_K2[i], D2Entry::new); + } + + @Benchmark + public SupportEntry getOrCreate_support(SharedState s, ThreadState t) { + int i = t.next(); + String k1 = SOURCE_K1[i]; + int k2 = SOURCE_K2_INT[i]; + long keyHash = SupportEntry.hash(k1, k2); + int index = ConcurrentHashtable.bucketIndex(s.supportBuckets, keyHash); + for (SupportEntry e = ConcurrentHashtable.bucketAt(s.supportBuckets, index); + e != null; + e = e.next()) { + if (e.keyHash == keyHash && e.matches(k1, k2)) { + return e; + } + } + synchronized (ConcurrentHashtable.getWriteLock(s.supportBuckets)) { + for (SupportEntry e = ConcurrentHashtable.bucketAt(s.supportBuckets, index); + e != null; + e = e.next()) { + if (e.keyHash == keyHash && e.matches(k1, k2)) { + return e; + } + } + SupportEntry newEntry = new SupportEntry(k1, k2); + ConcurrentHashtable.insertHeadEntryAt(s.supportBuckets, index, newEntry); + return newEntry; + } + } + + /** + * get-first pattern for CHM to avoid capturing-lambda allocation on hits — the idiomatic + * equivalent of D2.getOrCreate on a mostly-populated table. + */ + @Benchmark + public Long getOrCreate_concurrentHashMap(SharedState s, ThreadState t) { + int i = t.next(); + Key2 key = new Key2(SOURCE_K1[i], SOURCE_K2[i]); + Long existing = s.concurrentHashMap.get(key); + if (existing != null) { + return existing; + } + return s.concurrentHashMap.computeIfAbsent(key, k -> 0L); + } + + /** + * get-first pattern for ConcurrentSkipListMap — manual get-then-putIfAbsent since CSLM has no + * computeIfAbsent. Two traversals on miss; one on hit. + */ + @Benchmark + public Long getOrCreate_concurrentSkipListMap(SharedState s, ThreadState t) { + int i = t.next(); + Key2 key = new Key2(SOURCE_K1[i], SOURCE_K2[i]); + Long existing = s.skipListMap.get(key); + if (existing != null) { + return existing; + } + Long prev = s.skipListMap.putIfAbsent(key, 0L); + return prev != null ? prev : 0L; + } + + /** + * get-first pattern for synchronized HashMap. On hit: one lock acquire/release for get. On miss: + * a second synchronized block for the double-checked put. + */ + @Benchmark + public Long getOrCreate_synchronizedHashMap(SharedState s, ThreadState t) { + int i = t.next(); + Key2 key = new Key2(SOURCE_K1[i], SOURCE_K2[i]); + Long existing = s.synchronizedHashMap.get(key); + if (existing != null) { + return existing; + } + synchronized (s.synchronizedHashMap) { + return s.synchronizedHashMap.computeIfAbsent(key, k -> 0L); + } + } +} diff --git a/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java b/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java new file mode 100644 index 00000000000..92bda67c0c8 --- /dev/null +++ b/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java @@ -0,0 +1,1383 @@ +package datadog.trace.util; + +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReferenceArray; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Predicate; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import javax.annotation.concurrent.GuardedBy; +import javax.annotation.concurrent.ThreadSafe; + +/** + * Concurrent hash table providing lock-free reads and locked writes for {@link D1} (single-key) and + * {@link D2} (composite-key) tables. + * + *

The API deliberately mirrors {@link Hashtable} so the two are familiar to use, but the two + * share no implementation: {@code ConcurrentHashtable} carries its own {@link Entry} + * hierarchy with a {@code volatile} chain pointer and its own write paths. The single-threaded and + * concurrent variants evolve under different constraints (the concurrent one must reason about the + * memory model on every mutation), so coupling them through a shared base would be a hazard, not a + * convenience. + * + *

Like {@link Hashtable}, capacity is fixed at construction and the table does not resize. + * Unlike {@link Hashtable}, all operations are safe for concurrent access without external + * synchronization. + * + *

The primary advantage over {@link java.util.concurrent.ConcurrentHashMap} for composite-key + * use cases is that {@link D2#get(Object, Object)} and {@link D2#tryGetOrCreate(Object, Object, + * BiFunction)} accept key parts directly — no composite key object is allocated for the lookup. + * {@code ConcurrentHashMap} requires a wrapper object whose ownership may transfer to the map on + * insert; escape analysis must conservatively assume the key escapes even on hit paths, preventing + * scalar replacement. + * + *

Memory model. Bucket slots are held in an {@link AtomicReferenceArray}, so each {@link + * D1#get}/{@link D2#get} begins with a volatile read of the slot. The chain {@code next} pointer is + * {@code volatile} as well, so every step of a chain walk is a volatile read. This is what makes + * removal safe: a splice (re-pointing a predecessor's {@code next} past the removed entry, + * or replacing the bucket head) is a volatile write that lock-free readers observe. The cost is a + * volatile read per chain step and a slightly more expensive insert; the benefit is that the table + * supports removal — {@link D1#remove}, {@link D1#removeIf}, {@link D1#drain}, and {@link D1#clear} + * — rather than being append-only. {@link D1#drain} is the read-and-reset primitive for flush/ + * publish workflows: it removes every entry while handing each to a caller-supplied sink. + * + *

Removal and in-flight readers. A removed entry's own {@code next} pointer is left + * intact (it is never nulled). A reader that had already advanced onto the entry being removed must + * still be able to follow {@code next} forward to the rest of the chain; the detached entry is + * simply unreachable for new lookups and becomes garbage once no in-flight reader references it. A + * concurrent lookup racing a removal may observe either the pre- or post-removal state — both are + * valid linearizations. + * + *

Custom tables (higher arity / primitive keys). Use {@link D1} or {@link D2} when their + * object-key constraints are acceptable — they handle synchronization internally. When you need + * primitive key components, three-or-more key parts, or extra per-entry value fields, drive the + * table yourself with the static building blocks on this class: allocate the spine with {@link + * #createFixedBuckets(Class, int)}, then operate on it with {@link #bucketFor} / {@link #bucketAt}, + * {@link #unlink}, {@link #removeIf}, {@link #drain}, {@link #clear}, and {@link #forEach}. This is + * the same "static functions over a caller-owned array" shape as {@link Hashtable} (see how {@code + * AggregateTable} uses {@code Hashtable}); the calling class then owns the array and exposes + * whatever operations it needs. Subclass {@link Entry} directly for such tables. + * + *

Locking model. Writes are guarded by a per-table monitor obtained from {@link + * #getWriteLock(AtomicReferenceArray)} — treat it as opaque rather than assuming it is the array. + * Reads are lock-free: {@link #bucketFor} / {@link #bucketAt} walks and {@link #forEach} take no + * lock and are safe from any thread. The whole-table mutators — {@link #removeIf}, {@link #drain}, + * {@link #clear} — are self-locking ({@code synchronized (getWriteLock(buckets))} + * internally), so a custom table calls them directly with no lock of its own. The only writes a + * custom table performs by hand are single-key insert and remove; each is an atomic + * check-then-write that the caller wraps in {@code synchronized (getWriteLock(buckets))} so it + * excludes other writers and the self-locking mutators (same monitor, so it nests cleanly with the + * built-ins): + * + *

    + *
  1. Lock-free pre-check: walk the chain via {@link #bucketFor} / {@link #bucketAt}; return if + * found. + *
  2. {@code synchronized (getWriteLock(buckets))} — take the table's write monitor. + *
  3. Re-check under the lock (another thread may have inserted between step 1 and step 2). + *
  4. Insert: build the entry and publish it with {@link #insertHeadEntryFor} / {@link + * #insertHeadEntryAt}. Remove: splice it out with {@link #unlink}. Both are volatile writes + * that lock-free readers observe atomically. + *
+ * + *

{@link #bucketFor} / {@link #bucketAt} (a lock-free read), {@link #insertHeadEntryFor} / + * {@link #insertHeadEntryAt}, and {@link #unlink} are the single-slot primitives for that + * hand-written path; the two mutating ones do not lock, so call them only inside the + * caller's {@code synchronized (getWriteLock(buckets))} block. The entry's chain pointer is written + * for you by those helpers — custom tables never touch it directly. + * + *

A sequence of self-locking calls is not atomic. Each self-locking helper takes and + * releases the monitor on its own, so two of them in a row leave a window in between. That matters + * for any multi-step protocol over one table — notably reserving a slot with {@link + * #tryReserveOrEvict} and then filling it with {@link #insertReserved}: a {@link #drain} or {@link + * #clear} landing in the gap resets the {@link SizeManager} while the reservation is outstanding, + * and the later insert then links an entry the count no longer knows about, so a capped table + * drifts silently past its cap. Hold {@code synchronized (getWriteLock(state))} across the whole + * protocol; the monitor is reentrant, so the self-locking calls nest inside it cleanly. + */ +public final class ConcurrentHashtable { + private ConcurrentHashtable() {} + + /** + * Internal base class for concurrent entries. Stores the precomputed 64-bit keyHash and a {@code + * volatile} chain-next pointer used to link colliding entries within a single bucket. + * + *

The {@code next} pointer is {@code volatile} (unlike {@link Hashtable.Entry}) so that chain + * splices performed by {@link D1#remove}/{@link D2#remove} are visible to lock-free readers. + * + *

Subclasses add the key field(s) and a {@code matches(...)} method tailored to their key + * arity. See {@link D1.Entry} and {@link D2.Entry}; for higher arities, or for primitive key + * components, subclass this directly and drive the table with the static building blocks on + * {@link ConcurrentHashtable}. + */ + public abstract static class Entry { + public final long keyHash; + private volatile Entry next = null; + + protected Entry(long keyHash) { + this.keyHash = keyHash; + } + + // Package-private: the only writers are the static insert/remove building blocks + // (insertHeadEntry, unlink) on the enclosing class, which reach it via the Entry bound. Custom + // tables mutate chains through those helpers, never by touching next directly. + final void setNext(TEntry next) { + this.next = next; + } + + @SuppressWarnings("unchecked") + @Nullable + public final TEntry next() { + return (TEntry) this.next; + } + } + + /** + * Single-key concurrent hash table. Lock-free on hit; locked on miss/mutation. + * + * @param the key type + * @param the user's {@link D1.Entry D1.Entry<K>} subclass + */ + @ThreadSafe + public static final class D1> { + + /** + * Abstract base for {@link D1} entries. Subclass to add value fields you wish to mutate in + * place after retrieving the entry via {@link D1#get}. + * + * @param the key type + */ + public abstract static class Entry extends ConcurrentHashtable.Entry { + final K key; + + protected Entry(@Nullable K key) { + super(hash(key)); + this.key = key; + } + + /** The key this entry was created with. */ + @Nullable + public K key() { + return this.key; + } + + public boolean matches(@Nullable Object key) { + // equals() on the lookup param, not the field, so the JIT can devirtualize it once + // matches() inlines into get/getOrCreate (the caller's key type is known there). + return Objects.equals(key, this.key); + } + + /** + * Returns the 64-bit lookup hash for {@code key}. Null keys map to {@link Long#MIN_VALUE} so + * they don't collide with a real key that hashes to 0; real-key collisions in chains are + * resolved by {@link #matches(Object)}. + */ + public static long hash(@Nullable Object key) { + return (key == null) ? Long.MIN_VALUE : key.hashCode(); + } + } + + private final State state; + + private D1(State state) { + this.state = state; + } + + /** + * Creates a single-key table capped at {@code maxCapacity} entries: a {@link State} whose + * bucket array is sized with load-factor headroom over {@code maxCapacity} and whose {@link + * SizeManager} enforces {@code maxCapacity} as the strict entry-count limit consulted by {@link + * #tryGetOrCreate}. The {@code entryClass} pins the concrete entry type so the compiler infers + * both {@code K} and {@code TEntry} at the call site — e.g. {@code + * D1.createCapped(MyEntry.class, 64)}. Capacity is fixed; the table does not resize. + */ + @Nonnull + public static > D1 createCapped( + @Nonnull Class entryClass, int maxCapacity) { + return new D1<>(State.createCapped(entryClass, maxCapacity)); + } + + public int size() { + return state.sizeManager.estimateSize(); + } + + public boolean isFull() { + return state.sizeManager.isFull(); + } + + @Nullable + public TEntry get(@Nullable K key) { + long keyHash = D1.Entry.hash(key); + for (TEntry curEntry = bucketFor(state, keyHash); + curEntry != null; + curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key)) { + return curEntry; + } + } + return null; + } + + /** + * Returns the entry for {@code key}, creating one via {@code creator} if absent and the table + * is under capacity. Lock-free on hit; acquires a table-level lock on miss. Wraps {@link + * #tryGetOrCreateOrNull} — see that method for the refusal and ordering details. + */ + @Nonnull + public Maybe tryGetOrCreate( + @Nullable K key, @Nonnull Function creator) { + return Maybe.of(tryGetOrCreateOrNull(key, creator)); + } + + /** + * Escape hatch for {@link #tryGetOrCreate} for callers that want the nullable entry directly + * rather than a {@link Maybe} wrapper. Returns {@code null} when the table is at capacity and + * {@code key} was not already present. Re-checks under the lock to avoid duplicate entries + * under concurrent misses. + */ + @Nullable + public TEntry tryGetOrCreateOrNull( + @Nullable K key, @Nonnull Function creator) { + long keyHash = D1.Entry.hash(key); + int index = bucketIndex(state.buckets, keyHash); + for (TEntry curEntry = bucketAt(state, index); curEntry != null; curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key)) { + return curEntry; + } + } + synchronized (getWriteLock(state)) { + for (TEntry curEntry = bucketAt(state, index); + curEntry != null; + curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key)) { + return curEntry; + } + } + // Deliberately isFull() -> create -> increment, not a pre-reserved slot: creator runs + // between the check and the link and may throw, so reserving up front could leak a slot. + if (state.sizeManager.isFull()) { + return null; + } + TEntry newEntry = creator.apply(key); + insertHeadEntryAt(state, index, newEntry); + state.sizeManager.increment(); + return newEntry; + } + } + + /** + * {@link #tryGetOrCreate}, but when the table is full, evicts one entry matching {@code + * evictable} to make room instead of refusing the insert. Refuses only when the table is full + * and nothing matches {@code evictable} — see {@link #tryGetOrCreateOrEvictOrNull} for + * the null-returning form and the eviction/creation ordering. + */ + @Nonnull + public Maybe tryGetOrCreateOrEvict( + @Nullable K key, + @Nonnull Function creator, + @Nonnull Predicate evictable) { + return Maybe.of(tryGetOrCreateOrEvictOrNull(key, creator, evictable)); + } + + /** + * Escape hatch for {@link #tryGetOrCreateOrEvict} for callers that want the nullable entry + * directly. Eviction runs before {@code creator}, not after: {@code creator} may throw, so + * freeing a slot and only then attempting the fallible create keeps a thrown exception from + * ever leaving a slot double-booked. A creator that throws after a successful eviction simply + * leaves the table one entry smaller — no corruption, just a wasted eviction. + */ + @Nullable + public TEntry tryGetOrCreateOrEvictOrNull( + @Nullable K key, + @Nonnull Function creator, + @Nonnull Predicate evictable) { + long keyHash = D1.Entry.hash(key); + int index = bucketIndex(state.buckets, keyHash); + for (TEntry curEntry = bucketAt(state, index); curEntry != null; curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key)) { + return curEntry; + } + } + synchronized (getWriteLock(state)) { + for (TEntry curEntry = bucketAt(state, index); + curEntry != null; + curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key)) { + return curEntry; + } + } + if (state.sizeManager.isFull() + && state.sizeManager.evictOne(state.buckets, evictable) == null) { + return null; + } + TEntry newEntry = creator.apply(key); + insertHeadEntryAt(state, index, newEntry); + state.sizeManager.increment(); + return newEntry; + } + } + + /** + * Removes and returns the entry for {@code key}, or {@code null} if absent. Acquires the + * table-level lock to splice the chain; lock-free readers observe the removal via the volatile + * write of the predecessor's {@code next} (or the bucket head). + */ + @Nullable + public TEntry remove(@Nullable K key) { + long keyHash = D1.Entry.hash(key); + int index = bucketIndex(state.buckets, keyHash); + synchronized (getWriteLock(state)) { + TEntry prev = null; + for (TEntry curEntry = bucketAt(state, index); + curEntry != null; + prev = curEntry, curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key)) { + unlink(state, index, prev, curEntry); + state.sizeManager.decrement(); + return curEntry; + } + } + return null; + } + } + + /** + * Removes every entry matching {@code predicate}, returning {@code true} if any were removed. + * Holds the table-level lock for the whole sweep, so the predicate sees a stable table and + * concurrent writers are excluded; lock-free readers continue throughout. + */ + public boolean removeIf(@Nonnull Predicate predicate) { + return ConcurrentHashtable.removeIf(state, predicate); + } + + /** + * Removes every entry, passing each removed entry to {@code sink} as it is unlinked — the + * read-and-reset primitive for flush/publish workflows (drain the table into a telemetry batch, + * an event emitter, etc.). The whole drain runs under the table-level lock, so it is atomic + * with respect to other writers; {@code sink} therefore runs under the lock and should be cheap + * (accumulate into a collection rather than doing heavy work inline). Equivalent to {@code + * forEach}-then-{@code clear} but in a single locked pass that observes exactly what was + * removed. + * + *

A capturing-lambda {@code sink} is fine here — drain is a rare flush operation — but a + * context-passing overload is offered for callers that prefer to avoid the allocation. + * + *

Contract: {@code sink} must not throw. Entries are detached as the sweep proceeds + * and {@code size} is reset only after it completes, so a {@code sink} that throws part-way + * leaves those already-detached entries gone while {@code size()} still reports the pre-drain + * count. The drain is not rolled back; a throwing sink is a caller error that also means a + * half-published flush. This is intentional — the alternative is per-entry size bookkeeping on + * a path that only matters when the caller is already in error. + */ + public void drain(@Nonnull Consumer sink) { + ConcurrentHashtable.drain(state, sink); + } + + /** + * Context-passing {@link #drain(Consumer)}. Pass a non-capturing {@link BiConsumer} (typically + * a {@code static final}) plus the accumulator as {@code context} (e.g. the target list or + * event builder) to avoid a capturing-lambda allocation. + */ + public void drain(C context, @Nonnull BiConsumer sink) { + ConcurrentHashtable.drain(state, context, sink); + } + + /** Removes all entries. Lock-free readers mid-walk complete against the entries they hold. */ + public void clear() { + ConcurrentHashtable.clear(state); + } + + public void forEach(@Nonnull Consumer consumer) { + ConcurrentHashtable.forEach(state, consumer); + } + + /** + * Context-passing forEach. Avoids a capturing-lambda allocation — pass a non-capturing {@link + * BiConsumer} (typically a {@code static final}) plus whatever side-band state it needs. + */ + public void forEach(C context, @Nonnull BiConsumer consumer) { + ConcurrentHashtable.forEach(state, context, consumer); + } + } + + /** + * Two-key (composite-key) concurrent hash table. Lock-free on hit; locked on miss/mutation. + * + *

Key parts are passed directly to {@link #get} and {@link #getOrCreate}, eliminating the + * per-lookup composite key object allocation that {@code ConcurrentHashMap, V>} + * requires. + * + * @param first key type + * @param second key type + * @param the user's {@link D2.Entry D2.Entry<K1, K2>} subclass + */ + @ThreadSafe + public static final class D2> { + + /** + * Abstract base for {@link D2} entries. Subclass to add value fields you wish to mutate in + * place. + * + * @param first key type + * @param second key type + */ + public abstract static class Entry extends ConcurrentHashtable.Entry { + final K1 key1; + final K2 key2; + + protected Entry(@Nullable K1 key1, @Nullable K2 key2) { + super(hash(key1, key2)); + this.key1 = key1; + this.key2 = key2; + } + + /** The first key part this entry was created with. */ + @Nullable + public K1 key1() { + return this.key1; + } + + /** The second key part this entry was created with. */ + @Nullable + public K2 key2() { + return this.key2; + } + + public boolean matches(@Nullable K1 key1, @Nullable K2 key2) { + // equals() on the lookup params, not the fields, so the JIT can devirtualize them once + // matches() inlines into get/getOrCreate (the caller's key types are known there). + return Objects.equals(key1, this.key1) && Objects.equals(key2, this.key2); + } + + /** Returns the 64-bit lookup hash combining both key parts via {@link LongHashingUtils}. */ + public static long hash(@Nullable Object key1, @Nullable Object key2) { + return LongHashingUtils.hash(key1, key2); + } + } + + private final State state; + + private D2(State state) { + this.state = state; + } + + /** + * Creates a composite-key table capped at {@code maxCapacity} entries: a {@link State} whose + * bucket array is sized with load-factor headroom over {@code maxCapacity} and whose {@link + * SizeManager} enforces {@code maxCapacity} as the strict entry-count limit consulted by {@link + * #tryGetOrCreate}. The {@code entryClass} pins the concrete entry type so the compiler infers + * {@code K1}, {@code K2}, and {@code TEntry} at the call site — e.g. {@code + * D2.createCapped(MyEntry.class, 64)}. Capacity is fixed; the table does not resize. + */ + @Nonnull + public static > D2 createCapped( + @Nonnull Class entryClass, int maxCapacity) { + return new D2<>(State.createCapped(entryClass, maxCapacity)); + } + + public int size() { + return state.sizeManager.estimateSize(); + } + + public boolean isFull() { + return state.sizeManager.isFull(); + } + + @Nullable + public TEntry get(@Nullable K1 key1, @Nullable K2 key2) { + long keyHash = D2.Entry.hash(key1, key2); + for (TEntry curEntry = bucketFor(state, keyHash); + curEntry != null; + curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) { + return curEntry; + } + } + return null; + } + + /** + * Returns the entry for {@code (key1, key2)}, creating one via {@code creator} if absent and + * the table is under capacity. Lock-free on hit; acquires a table-level lock on miss. Wraps + * {@link #tryGetOrCreateOrNull} — see that method for the refusal and ordering details. + * + *

The {@code creator} should build an entry whose {@code keyHash} equals {@link + * D2.Entry#hash(Object, Object) D2.Entry.hash(key1, key2)}. + */ + @Nonnull + public Maybe tryGetOrCreate( + @Nullable K1 key1, + @Nullable K2 key2, + @Nonnull BiFunction creator) { + return Maybe.of(tryGetOrCreateOrNull(key1, key2, creator)); + } + + /** + * Escape hatch for {@link #tryGetOrCreate} for callers that want the nullable entry directly + * rather than a {@link Maybe} wrapper. Returns {@code null} when the table is at capacity and + * {@code (key1, key2)} was not already present. Re-checks under the lock to avoid duplicate + * entries under concurrent misses. + */ + @Nullable + public TEntry tryGetOrCreateOrNull( + @Nullable K1 key1, + @Nullable K2 key2, + @Nonnull BiFunction creator) { + long keyHash = D2.Entry.hash(key1, key2); + int index = bucketIndex(state.buckets, keyHash); + for (TEntry curEntry = bucketAt(state, index); curEntry != null; curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) { + return curEntry; + } + } + synchronized (getWriteLock(state)) { + for (TEntry curEntry = bucketAt(state, index); + curEntry != null; + curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) { + return curEntry; + } + } + // Deliberately isFull() -> create -> increment, not a pre-reserved slot: creator runs + // between the check and the link and may throw, so reserving up front could leak a slot. + if (state.sizeManager.isFull()) { + return null; + } + TEntry newEntry = creator.apply(key1, key2); + insertHeadEntryAt(state, index, newEntry); + state.sizeManager.increment(); + return newEntry; + } + } + + /** + * {@link #tryGetOrCreate}, but when the table is full, evicts one entry matching {@code + * evictable} to make room instead of refusing the insert. Refuses only when the table is full + * and nothing matches {@code evictable} — see {@link #tryGetOrCreateOrEvictOrNull} for + * the null-returning form and the eviction/creation ordering. + */ + @Nonnull + public Maybe tryGetOrCreateOrEvict( + @Nullable K1 key1, + @Nullable K2 key2, + @Nonnull BiFunction creator, + @Nonnull Predicate evictable) { + return Maybe.of(tryGetOrCreateOrEvictOrNull(key1, key2, creator, evictable)); + } + + /** + * Escape hatch for {@link #tryGetOrCreateOrEvict} for callers that want the nullable entry + * directly. Eviction runs before {@code creator}, not after: {@code creator} may throw, so + * freeing a slot and only then attempting the fallible create keeps a thrown exception from + * ever leaving a slot double-booked. A creator that throws after a successful eviction simply + * leaves the table one entry smaller — no corruption, just a wasted eviction. + */ + @Nullable + public TEntry tryGetOrCreateOrEvictOrNull( + @Nullable K1 key1, + @Nullable K2 key2, + @Nonnull BiFunction creator, + @Nonnull Predicate evictable) { + long keyHash = D2.Entry.hash(key1, key2); + int index = bucketIndex(state.buckets, keyHash); + for (TEntry curEntry = bucketAt(state, index); curEntry != null; curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) { + return curEntry; + } + } + synchronized (getWriteLock(state)) { + for (TEntry curEntry = bucketAt(state, index); + curEntry != null; + curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) { + return curEntry; + } + } + if (state.sizeManager.isFull() + && state.sizeManager.evictOne(state.buckets, evictable) == null) { + return null; + } + TEntry newEntry = creator.apply(key1, key2); + insertHeadEntryAt(state, index, newEntry); + state.sizeManager.increment(); + return newEntry; + } + } + + /** + * Removes and returns the entry for {@code (key1, key2)}, or {@code null} if absent. Acquires + * the table-level lock to splice the chain; lock-free readers observe the removal via the + * volatile write of the predecessor's {@code next} (or the bucket head). + */ + @Nullable + public TEntry remove(@Nullable K1 key1, @Nullable K2 key2) { + long keyHash = D2.Entry.hash(key1, key2); + int index = bucketIndex(state.buckets, keyHash); + synchronized (getWriteLock(state)) { + TEntry prev = null; + for (TEntry curEntry = bucketAt(state, index); + curEntry != null; + prev = curEntry, curEntry = curEntry.next()) { + if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) { + unlink(state, index, prev, curEntry); + state.sizeManager.decrement(); + return curEntry; + } + } + return null; + } + } + + /** + * Removes every entry matching {@code predicate}, returning {@code true} if any were removed. + * Holds the table-level lock for the whole sweep, so the predicate sees a stable table and + * concurrent writers are excluded; lock-free readers continue throughout. + */ + public boolean removeIf(@Nonnull Predicate predicate) { + return ConcurrentHashtable.removeIf(state, predicate); + } + + /** + * Removes every entry, passing each removed entry to {@code sink} as it is unlinked — the + * read-and-reset primitive for flush/publish workflows (drain the table into a telemetry batch, + * an event emitter, etc.). The whole drain runs under the table-level lock, so it is atomic + * with respect to other writers; {@code sink} therefore runs under the lock and should be cheap + * (accumulate into a collection rather than doing heavy work inline). Equivalent to {@code + * forEach}-then-{@code clear} but in a single locked pass that observes exactly what was + * removed. + * + *

A capturing-lambda {@code sink} is fine here — drain is a rare flush operation — but a + * context-passing overload is offered for callers that prefer to avoid the allocation. + * + *

Contract: {@code sink} must not throw. Entries are detached as the sweep proceeds + * and {@code size} is reset only after it completes, so a {@code sink} that throws part-way + * leaves those already-detached entries gone while {@code size()} still reports the pre-drain + * count. The drain is not rolled back; a throwing sink is a caller error that also means a + * half-published flush. This is intentional — the alternative is per-entry size bookkeeping on + * a path that only matters when the caller is already in error. + */ + public void drain(@Nonnull Consumer sink) { + ConcurrentHashtable.drain(state, sink); + } + + /** + * Context-passing {@link #drain(Consumer)}. Pass a non-capturing {@link BiConsumer} (typically + * a {@code static final}) plus the accumulator as {@code context} (e.g. the target list or + * event builder) to avoid a capturing-lambda allocation. + */ + public void drain(C context, @Nonnull BiConsumer sink) { + ConcurrentHashtable.drain(state, context, sink); + } + + /** Removes all entries. Lock-free readers mid-walk complete against the entries they hold. */ + public void clear() { + ConcurrentHashtable.clear(state); + } + + public void forEach(@Nonnull Consumer consumer) { + ConcurrentHashtable.forEach(state, consumer); + } + + /** + * Context-passing forEach. Avoids a capturing-lambda allocation — pass a non-capturing {@link + * BiConsumer} (typically a {@code static final}) plus whatever side-band state it needs. + */ + public void forEach(C context, @Nonnull BiConsumer consumer) { + ConcurrentHashtable.forEach(state, context, consumer); + } + } + + /** + * Concurrent counterpart to {@link Hashtable.SizeManager}: manages a table's occupancy against a + * fixed cap in both directions — reserving a slot for an insert and evicting to make room — so a + * caller never has to remember to decrement after unlinking, nor wire up a second object + * alongside the count. + * + *

{@link D1} and {@link D2} each hold one (via {@link State}) for their strict entry-count + * cap; composers driving an {@link AtomicReferenceArray} through the static building blocks can + * pair one the same way instead of hand-rolling the increment/decrement/cap-check bookkeeping — + * see {@link State#createCapped}. + * + *

Locking. {@link #estimateSize()}, {@link #capacity()}, and {@link #isFull()} read + * only the atomic counter and need no lock. Every other method walks or mutates the chains (or + * the eviction cursor) and must be called under {@code synchronized (getWriteLock(buckets))} — + * the same monitor guarding the table's other writes — so a scan never races a concurrent insert + * or remove. Unlike {@link Hashtable.SizeManager}'s plain {@code int}, the live count here is an + * {@link AtomicInteger}: {@link #estimateSize()} and {@link #isFull()} are read without the lock + * (e.g. from {@link D1#size()}), which a plain field could not support safely. + */ + @ThreadSafe + public static final class SizeManager { + private final AtomicInteger size = new AtomicInteger(); + private final int capacity; + + /** + * Bucket index the last eviction removed from. The next scan resumes here, so a sustained + * eviction stream doesn't repeatedly re-walk the same hot entries clustered near bucket 0. + */ + @GuardedBy("getWriteLock(buckets)") + private int cursor; + + public SizeManager(int capacity) { + this.capacity = capacity; + } + + /** Live entries. Safe to call without the write lock. */ + public int estimateSize() { + return size.get(); + } + + public int capacity() { + return capacity; + } + + /** {@code true} once {@link #estimateSize()} has reached {@link #capacity()}. */ + public boolean isFull() { + return size.get() >= capacity; + } + + /** + * Reserves a slot for a fresh insert: increments and returns {@code true}, or leaves the count + * unchanged and returns {@code false} if already at capacity. Use this when the entry to link + * is already fully built (nothing between the check and the increment can fail). When building + * the entry is itself fallible, check {@link #isFull()} first, do the fallible work, then call + * {@link #increment()} only once linking actually succeeds — see {@link + * D1#tryGetOrCreateOrNull} for that ordering. + */ + @GuardedBy("getWriteLock(buckets)") + public boolean tryReserve() { + if (isFull()) { + return false; + } + size.incrementAndGet(); + return true; + } + + /** + * {@link #tryReserve()}, falling back to evicting one entry matching {@code evictable} when the + * table is full. Returns {@code true} with a slot reserved, or {@code false} if the table was + * full and nothing was evictable — in which case {@code buckets} is untouched and the caller + * should drop the datum. + * + *

The write lock must be held across the insert that consumes the reservation, not merely + * across this call: {@link #reset()} (via a table-level drain or clear) zeroes the count, and a + * reservation taken before it is silently voided. + */ + @GuardedBy("getWriteLock(buckets)") + public boolean tryReserveOrEvict( + @Nonnull AtomicReferenceArray buckets, + @Nonnull Predicate evictable) { + if (tryReserve()) { + return true; + } + if (evictOne(buckets, evictable) == null) { + return false; + } + // evictOne already decremented; the slot it freed is ours. + size.incrementAndGet(); + return true; + } + + /** Call after successfully linking a new entry. */ + public void increment() { + size.incrementAndGet(); + } + + /** Call after successfully unlinking an entry. */ + public void decrement() { + size.decrementAndGet(); + } + + /** Zeroes both the live count and the eviction scan position. */ + @GuardedBy("getWriteLock(buckets)") + @SuppressFBWarnings( + value = "AT_STALE_THREAD_WRITE_OF_PRIMITIVE", + justification = + "cursor is read and written only under synchronized (getWriteLock(buckets)); SpotBugs" + + " cannot model that dynamic guard") + public void reset() { + size.set(0); + cursor = 0; + } + + /** + * Scans {@code buckets} for the first entry matching {@code evictable}, starting where the last + * eviction left off and wrapping around if needed. Unlinks and returns the evicted entry, + * decrementing the count; returns {@code null} (count untouched) if nothing matched anywhere. + * + *

Resuming from the previous position amortizes a sustained eviction stream: no successful + * eviction re-scans the hot prefix more than twice. A call that matches nothing has, by + * definition, tested every live entry, so a table that is full and entirely hot pays a full + * pass per attempt; the cursor still steps on so repeated refusals at least start from a + * different bucket next time. Size the cap to the steady-state working set so this stays the + * rare path, and keep {@code evictable} cheap — it is called once per live entry on every + * refusal. + */ + @GuardedBy("getWriteLock(buckets)") + @Nullable + public TEntry evictOne( + @Nonnull AtomicReferenceArray buckets, + @Nonnull Predicate evictable) { + TEntry evicted = evictOneInRange(buckets, evictable, cursor, buckets.length()); + if (evicted == null && cursor != 0) { + evicted = evictOneInRange(buckets, evictable, 0, cursor); + } + if (evicted != null) { + size.decrementAndGet(); + return evicted; + } + // Nothing matched anywhere; step the cursor on regardless so repeated refusals don't all + // restart the (wasted) scan from the same bucket. + cursor = bucketIndex(buckets, cursor + 1); + return null; + } + + @GuardedBy("getWriteLock(buckets)") + @SuppressFBWarnings( + value = "AT_STALE_THREAD_WRITE_OF_PRIMITIVE", + justification = + "cursor is read and written only under synchronized (getWriteLock(buckets)); SpotBugs" + + " cannot model that dynamic guard") + @Nullable + private TEntry evictOneInRange( + @Nonnull AtomicReferenceArray buckets, + @Nonnull Predicate evictable, + int startBucket, + int endBucket) { + for (int i = startBucket; i < endBucket; i++) { + TEntry prev = null; + for (TEntry e = buckets.get(i); e != null; e = e.next()) { + if (evictable.test(e)) { + unlink(buckets, i, prev, e); + cursor = i; + return e; + } + prev = e; + } + } + return null; + } + + /** + * Unlinks every entry matching {@code evictable} in one full pass, decrementing the count for + * each, and returns how many were removed. Resets the scan position, since a full pass leaves + * nothing later to resume from. + */ + @GuardedBy("getWriteLock(buckets)") + @SuppressFBWarnings( + value = "AT_STALE_THREAD_WRITE_OF_PRIMITIVE", + justification = + "cursor is read and written only under synchronized (getWriteLock(buckets)); SpotBugs" + + " cannot model that dynamic guard") + public int evictAll( + @Nonnull AtomicReferenceArray buckets, + @Nonnull Predicate evictable) { + int count = 0; + for (int i = 0; i < buckets.length(); i++) { + TEntry prev = null; + for (TEntry e = buckets.get(i); e != null; e = e.next()) { + if (evictable.test(e)) { + unlink(buckets, i, prev, e); + size.decrementAndGet(); + count++; + } else { + prev = e; + } + } + } + cursor = 0; + return count; + } + } + + /** + * The mutable state of a caller-driven table: a bucket array and the {@link SizeManager} sized + * and capped to match it. Both halves are stateful and neither is much use without the other, + * which is what the name is getting at — the spine holds the entries, the manager holds how many + * there are and where the last eviction looked. + * + *

Hold this, rather than unpacking it. Keeping one field instead of two is not just + * tidier: an array and a manager stored separately can drift apart, which is the mistake this + * type exists to prevent. {@link D1} and {@link D2} hold one internally; composers reach through + * it — {@code state.buckets}, {@code state.sizeManager} — when calling the static building blocks + * directly, or use the {@code State}-taking overloads on this class. + * + *

Same headroom idiom as {@link D1}/{@link D2}: {@code maxCapacity} is the strict cap on live + * entries, and the backing array is sized with load-factor headroom over it. + */ + public static final class State { + public final AtomicReferenceArray buckets; + public final SizeManager sizeManager; + + private State(AtomicReferenceArray buckets, int maxCapacity) { + this.buckets = buckets; + this.sizeManager = new SizeManager(maxCapacity); + } + + /** + * Creates a {@link State}: a bucket array sized with load-factor headroom over {@code + * maxCapacity} (via {@link #createFixedBuckets(Class, int)}), paired with a {@link SizeManager} + * capped at the strict {@code maxCapacity}. {@code entryClass} is a type token only — see + * {@link #createFixedBuckets(Class, int)} for why it's needed despite not being used to + * allocate. + */ + @Nonnull + public static State createCapped( + @Nonnull Class entryClass, int maxCapacity) { + return new State<>(createFixedBuckets(entryClass, maxCapacity), maxCapacity); + } + } + + /** Live entries in {@code state}; see {@link SizeManager#estimateSize()}. Lock-free. */ + public static int estimateSize(@Nonnull State state) { + return state.sizeManager.estimateSize(); + } + + /** + * {@code true} once {@code state} is at capacity; see {@link SizeManager#isFull()}. Lock-free. + */ + public static boolean isFull(@Nonnull State state) { + return state.sizeManager.isFull(); + } + + /** + * Reserves a slot in {@code state} for a fresh insert, evicting one entry matching {@code + * evictable} if the table is full. {@code false} means full with nothing evictable — the caller + * should drop the datum. Self-locking. + * + *

Pairing with an insert: the reservation this takes is only meaningful until the next + * {@link #drain} or {@link #clear}, either of which resets the {@link SizeManager}. Because this + * call releases the monitor before returning, a caller that follows it with {@link + * #insertReserved} must hold {@code synchronized (getWriteLock(state))} across both calls + * — see {@link #insertReserved} for the shape. + */ + public static boolean tryReserveOrEvict( + @Nonnull State state, @Nonnull Predicate evictable) { + synchronized (getWriteLock(state)) { + return state.sizeManager.tryReserveOrEvict(state.buckets, evictable); + } + } + + /** + * Unlinks the first entry in {@code state} matching {@code evictable}, resuming from where the + * last eviction looked, and decrements the count. {@code null} if nothing matched anywhere. + * Self-locking. + */ + @Nullable + public static TEntry evictOne( + @Nonnull State state, @Nonnull Predicate evictable) { + synchronized (getWriteLock(state)) { + return state.sizeManager.evictOne(state.buckets, evictable); + } + } + + /** + * Unlinks every entry in {@code state} matching {@code evictable}, decrementing per removal, and + * returns how many went. Self-locking. + */ + public static int evictAll( + @Nonnull State state, @Nonnull Predicate evictable) { + synchronized (getWriteLock(state)) { + return state.sizeManager.evictAll(state.buckets, evictable); + } + } + + // --------------------------------------------------------------------------------------------- + // Static building blocks over a caller-owned bucket array (formerly the nested Support class). + // Use these to assemble a custom table (higher arity, primitive keys, extra value fields) when + // D1/D2 don't fit; D1/D2 delegate to them internally. The whole-table mutators (removeIf, drain, + // clear) self-lock on the array; the single-slot write primitives (insertHeadEntry, unlink) do + // not lock and must be called under the caller's own synchronized (getWriteLock(buckets)) block. + // Readers + // (bucket walks, forEach) are lock-free. + // --------------------------------------------------------------------------------------------- + + /** + * Allocates a fixed-size bucket array sized to hold {@code capacity} entries: {@code capacity} + * rounded up to the next power of two. + * + *

Unlike {@code FlatHashtable}, whose open-addressing spine is a genuine {@code E[]} that must + * be reflectively allocated from {@code entryClass}, the concurrent spine is an {@link + * AtomicReferenceArray} whose element type is erased — so {@code entryClass} is not used + * to allocate here. It is accepted purely to (a) keep the factory symmetric with the rest of the + * flat-collections family and (b) act as a type-inference anchor so callers write {@code + * createFixedBuckets(MyEntry.class, n)} and get back a precisely typed {@code + * AtomicReferenceArray} without an explicit witness. + */ + @Nonnull + public static AtomicReferenceArray createFixedBuckets( + @Nonnull Class entryClass, int capacity) { + return new AtomicReferenceArray<>(sizeFor(capacity)); + } + + /** + * Returns the bucket-array length to allocate for a table sized to hold {@code requestedSize} + * entries: {@code requestedSize} rounded up to the next power of two. Shares {@link Hashtable}'s + * sizing so the two families round identically. + */ + public static int sizeFor(int requestedSize) { + return Hashtable.Support.sizeFor(requestedSize); + } + + /** + * Returns the monitor that guards writes to {@code buckets}. A custom table locks on this — + * {@code synchronized (getWriteLock(buckets)) { … }} — around its scan-then-insert/remove so it + * excludes other writers and the self-locking whole-table mutators (they lock on the same + * monitor, so the blocks nest). Treat the returned object as opaque: it happens to be the + * array today, but obtain it here rather than assuming that, so callers stay correct if the + * monitor ever changes. + */ + @Nonnull + public static Object getWriteLock(@Nonnull AtomicReferenceArray buckets) { + return buckets; + } + + /** {@link #getWriteLock(AtomicReferenceArray)} over a {@link State}. */ + @Nonnull + public static Object getWriteLock(@Nonnull State state) { + return getWriteLock(state.buckets); + } + + public static int bucketIndex(@Nonnull AtomicReferenceArray buckets, long keyHash) { + return (int) (keyHash & (buckets.length() - 1)); + } + + /** + * Returns the head entry of the bucket that {@code keyHash} maps to. The bucket read is a + * volatile read of the slot, so it is safe from any thread without a lock. + * + *

Named distinctly from {@link #bucketAt} (rather than overloaded on {@code long} vs. {@code + * int}) deliberately: a caller with a primitive {@code int}-typed key hash that called an + * overloaded {@code bucket(buckets, intHash)} would silently bind to the {@code int}-index + * overload instead of widening to this one, reading the raw hash as an array index — out-of-range + * hashes throw {@link IndexOutOfBoundsException}, in-range-but-wrong ones silently read the wrong + * bucket. + */ + @Nullable + public static TEntry bucketFor( + @Nonnull AtomicReferenceArray buckets, long keyHash) { + return buckets.get(bucketIndex(buckets, keyHash)); + } + + /** {@link #bucketFor(AtomicReferenceArray, long)} over a {@link State}. */ + @Nullable + public static TEntry bucketFor( + @Nonnull State state, long keyHash) { + return bucketFor(state.buckets, keyHash); + } + + /** + * Returns the head entry of the bucket at {@code index}. Use when the bucket index is already + * computed (e.g. inside {@code getOrCreate} where the same index is reused across the lock + * boundary). See {@link #bucketFor} for why this is a distinct name rather than an {@code int} + * overload of it. + */ + @Nullable + public static TEntry bucketAt( + @Nonnull AtomicReferenceArray buckets, int index) { + return buckets.get(index); + } + + /** {@link #bucketAt(AtomicReferenceArray, int)} over a {@link State}. */ + @Nullable + public static TEntry bucketAt(@Nonnull State state, int index) { + return bucketAt(state.buckets, index); + } + + /** + * Splices {@code entry} in as the new head of the chain at {@code index}, publishing it with a + * volatile {@link AtomicReferenceArray#set} so lock-free readers observe the whole entry (its + * {@code next} already points at the old head) atomically. Single-slot primitive: it does not + * lock, so call it inside the caller's {@code synchronized (getWriteLock(buckets))} block, after + * re-checking the chain for the key under that lock. Does not touch size accounting. + * + *

See {@link #bucketFor} for why this is a distinct name rather than an {@code int} overload + * of {@link #insertHeadEntryFor}. + */ + @GuardedBy("getWriteLock(buckets)") + public static void insertHeadEntryAt( + @Nonnull AtomicReferenceArray buckets, int index, @Nonnull TEntry entry) { + assert Thread.holdsLock(getWriteLock(buckets)) + : "insertHeadEntryAt called without holding getWriteLock(buckets)"; + assert entry.next() == null + : "Entry already linked -- inserting the same Entry instance twice corrupts the chain" + + " (unlink() deliberately leaves a removed entry's next intact for in-flight" + + " readers, so a removed entry must never be reinserted)"; + entry.setNext(buckets.get(index)); + buckets.set(index, entry); + } + + /** {@link #insertHeadEntryAt(AtomicReferenceArray, int, Entry)} over a {@link State}. */ + @GuardedBy("getWriteLock(state)") + public static void insertHeadEntryAt( + @Nonnull State state, int index, @Nonnull TEntry entry) { + insertHeadEntryAt(state.buckets, index, entry); + } + + /** + * Convenience form of {@link #insertHeadEntryAt} that derives the bucket index from {@code + * keyHash}. Prefer {@link #insertHeadEntryAt} when the index is already computed (e.g. a {@code + * getOrCreate} that reuses it across the lock-free pre-check). + */ + @GuardedBy("getWriteLock(buckets)") + public static void insertHeadEntryFor( + @Nonnull AtomicReferenceArray buckets, long keyHash, @Nonnull TEntry entry) { + insertHeadEntryAt(buckets, bucketIndex(buckets, keyHash), entry); + } + + /** + * Splices {@code entry} in as the new head of its bucket without touching the count, + * because the caller already holds a reservation for it -- from {@link #tryReserveOrEvict} or a + * bare {@link SizeManager#tryReserve()}. Pairing those is the shape of a miss path that wants to + * refuse before it builds anything: + * + *

{@code
+   * synchronized (getWriteLock(state)) {     // ONE critical section for both steps
+   *   if (!tryReserveOrEvict(state, evictable)) {
+   *     return null;                         // refused -- no entry was built
+   *   }
+   *   insertReserved(state, keyHash, buildEntry());
+   * }
+   * }
+ * + *

The enclosing block is required, not stylistic. {@link #tryReserveOrEvict} is self-locking + * and releases the monitor before it returns, so without it a {@link #drain} or {@link #clear} + * can land between the reservation and this insert, reset the {@link SizeManager}, and leave this + * insert linking an entry that the count no longer accounts for -- an undercount that never + * heals, and on a {@link State#createCapped} table a cap that is quietly exceeded from then on. + * The monitor is reentrant, so wrapping the self-locking call costs nothing. + * + *

Because the entry is built inside that block, {@code buildEntry()} must not throw: a throw + * after the reservation is taken leaks a slot for the life of the table. When the build is + * fallible, use the {@link D1#tryGetOrCreateOrNull} shape instead, which checks capacity, builds, + * links, and only then increments. + * + *

Distinct from {@link #insertHeadEntryFor(AtomicReferenceArray, long, Entry)}, which reserves + * as it inserts; calling that one here would count the entry twice. {@link D1} and {@link D2} do + * not use this: their {@code creator} is fallible, so they check/evict, build the entry, link it, + * and only then call {@link SizeManager#increment} -- reserving up front could leak a slot if the + * build throws (see {@link D1#tryGetOrCreateOrNull}). Use this only when the entry is already + * fully built before the reservation is taken. + */ + @GuardedBy("getWriteLock(state)") + public static void insertReserved( + @Nonnull State state, long keyHash, @Nonnull TEntry entry) { + insertHeadEntryFor(state.buckets, keyHash, entry); + } + + /** + * Splices {@code entry} out of the chain at {@code index}. {@code prev} is the in-chain + * predecessor, or {@code null} when {@code entry} is the bucket head. Re-points the predecessor + * (or the bucket head slot) past {@code entry} via a volatile write so lock-free readers see the + * removal. {@code entry}'s own {@code next} is deliberately left intact so a reader already + * positioned on it can still traverse forward. This is a single-slot primitive: it does not lock, + * so call it inside the caller's {@code synchronized (getWriteLock(buckets))} block. Does not + * touch size accounting. + */ + @GuardedBy("getWriteLock(buckets)") + public static void unlink( + @Nonnull AtomicReferenceArray buckets, + int index, + @Nullable TEntry prev, + @Nonnull TEntry entry) { + assert Thread.holdsLock(getWriteLock(buckets)) + : "unlink called without holding getWriteLock(buckets)"; + TEntry next = entry.next(); + if (prev == null) { + buckets.set(index, next); + } else { + prev.setNext(next); + } + } + + /** {@link #unlink(AtomicReferenceArray, int, Entry, Entry)} over a {@link State}. */ + @GuardedBy("getWriteLock(state)") + public static void unlink( + @Nonnull State state, int index, @Nullable TEntry prev, @Nonnull TEntry entry) { + unlink(state.buckets, index, prev, entry); + } + + /** + * Removes every entry matching {@code predicate} from {@code buckets}, decrementing {@code size} + * once per removal. Self-locking: synchronizes on {@code buckets} for the whole sweep, so the + * predicate sees a stable table and concurrent writers are excluded; lock-free readers continue + * throughout. + */ + public static boolean removeIf( + @Nonnull AtomicReferenceArray buckets, + @Nonnull AtomicInteger size, + @Nonnull Predicate predicate) { + synchronized (getWriteLock(buckets)) { + boolean removed = false; + for (int i = 0; i < buckets.length(); i++) { + TEntry prev = null; + for (TEntry e = buckets.get(i); e != null; e = e.next()) { + if (predicate.test(e)) { + unlink(buckets, i, prev, e); + size.decrementAndGet(); + removed = true; + // prev stays put: e is now unlinked, so the last survivor remains the predecessor. + } else { + prev = e; + } + } + } + return removed; + } + } + + /** + * {@link #removeIf(AtomicReferenceArray, AtomicInteger, Predicate)} variant for callers tracking + * occupancy with a {@link State} instead of a bare counter — used by {@link D1#removeIf} and + * {@link D2#removeIf}. + */ + public static boolean removeIf( + @Nonnull State state, @Nonnull Predicate predicate) { + AtomicReferenceArray buckets = state.buckets; + synchronized (getWriteLock(state)) { + boolean removed = false; + for (int i = 0; i < buckets.length(); i++) { + TEntry prev = null; + for (TEntry e = buckets.get(i); e != null; e = e.next()) { + if (predicate.test(e)) { + unlink(buckets, i, prev, e); + state.sizeManager.decrement(); + removed = true; + } else { + prev = e; + } + } + } + return removed; + } + } + + /** + * Removes every entry, passing each to {@code sink} as its bucket is cleared. Each bucket head is + * nulled (a volatile write that publishes the removal) before its chain is fed to {@code sink}, + * so new readers see an empty bucket while the detached chain — whose {@code next} pointers stay + * intact — is handed to the caller. Self-locking: synchronizes on {@code buckets} for the whole + * pass. Does not touch size accounting, so a caller tracking size resets it inside its own {@code + * synchronized (getWriteLock(buckets))} block (which nests with this one on the same monitor). + * + *

{@code sink} must not throw: buckets are detached as the sweep proceeds, so a sink that + * throws part-way leaves earlier buckets drained and later ones intact, and any caller-side size + * reset never runs. The drain is not rolled back — a throwing sink is a caller error. + */ + public static void drain( + @Nonnull AtomicReferenceArray buckets, @Nonnull Consumer sink) { + synchronized (getWriteLock(buckets)) { + for (int i = 0; i < buckets.length(); i++) { + TEntry head = buckets.get(i); + if (head == null) { + continue; + } + buckets.set(i, null); + for (TEntry e = head; e != null; e = e.next()) { + sink.accept(e); + } + } + } + } + + /** Context-passing variant of {@link #drain(AtomicReferenceArray, Consumer)}. Self-locking. */ + public static void drain( + @Nonnull AtomicReferenceArray buckets, + C context, + @Nonnull BiConsumer sink) { + synchronized (getWriteLock(buckets)) { + for (int i = 0; i < buckets.length(); i++) { + TEntry head = buckets.get(i); + if (head == null) { + continue; + } + buckets.set(i, null); + for (TEntry e = head; e != null; e = e.next()) { + sink.accept(context, e); + } + } + } + } + + /** + * {@link #drain(AtomicReferenceArray, Consumer)} plus the matching bookkeeping: empties {@code + * state} into {@code sink} and resets its {@link SizeManager} to zero. Draining without resetting + * leaves the cap permanently consumed, so the two belong in one call rather than as a pair the + * caller has to remember. + */ + public static void drain( + @Nonnull State state, @Nonnull Consumer sink) { + synchronized (getWriteLock(state)) { + drain(state.buckets, sink); + state.sizeManager.reset(); + } + } + + /** Context-passing form of {@link #drain(State, Consumer)}. */ + public static void drain( + @Nonnull State state, + C context, + @Nonnull BiConsumer sink) { + synchronized (getWriteLock(state)) { + drain(state.buckets, context, sink); + state.sizeManager.reset(); + } + } + + /** Nulls every bucket head. Self-locking: synchronizes on {@code buckets}. */ + public static void clear(@Nonnull AtomicReferenceArray buckets) { + synchronized (getWriteLock(buckets)) { + for (int i = 0; i < buckets.length(); i++) { + buckets.set(i, null); + } + } + } + + /** + * {@link #clear(AtomicReferenceArray)} over a {@link State}: also resets its {@link SizeManager}. + */ + public static void clear(@Nonnull State state) { + synchronized (getWriteLock(state)) { + clear(state.buckets); + state.sizeManager.reset(); + } + } + + public static void forEach( + @Nonnull AtomicReferenceArray buckets, @Nonnull Consumer consumer) { + for (int i = 0; i < buckets.length(); i++) { + for (TEntry curEntry = buckets.get(i); curEntry != null; curEntry = curEntry.next()) { + consumer.accept(curEntry); + } + } + } + + public static void forEach( + @Nonnull AtomicReferenceArray buckets, + C context, + @Nonnull BiConsumer consumer) { + for (int i = 0; i < buckets.length(); i++) { + for (TEntry curEntry = buckets.get(i); curEntry != null; curEntry = curEntry.next()) { + consumer.accept(context, curEntry); + } + } + } + + /** {@link #forEach(AtomicReferenceArray, Consumer)} over a {@link State}. */ + public static void forEach( + @Nonnull State state, @Nonnull Consumer consumer) { + forEach(state.buckets, consumer); + } + + /** {@link #forEach(AtomicReferenceArray, Object, BiConsumer)} over a {@link State}. */ + public static void forEach( + @Nonnull State state, + C context, + @Nonnull BiConsumer consumer) { + forEach(state.buckets, context, consumer); + } +} diff --git a/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableD1Test.java b/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableD1Test.java new file mode 100644 index 00000000000..f95e0657211 --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableD1Test.java @@ -0,0 +1,506 @@ +package datadog.trace.util; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class ConcurrentHashtableD1Test { + + @Test + void getReturnsMappedEntry() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + StringEntry e = table.tryGetOrCreateOrNull("hello", k -> new StringEntry(k, 42)); + assertSame(e, table.get("hello")); + assertNull(table.get("world")); + } + + @Test + void getOrCreateOnMissBuildsEntry() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + int[] createCount = {0}; + StringEntry created = + table.tryGetOrCreateOrNull( + "a", + k -> { + createCount[0]++; + return new StringEntry(k, 1); + }); + assertNotNull(created); + assertEquals(1, table.size()); + assertEquals(1, createCount[0]); + assertSame(created, table.get("a")); + } + + @Test + void getOrCreateOnHitSkipsCreator() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + StringEntry seeded = table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 100)); + int[] createCount = {0}; + StringEntry got = + table.tryGetOrCreateOrNull( + "a", + k -> { + createCount[0]++; + return new StringEntry(k, 999); + }); + assertSame(seeded, got); + assertEquals(1, table.size()); + assertEquals(0, createCount[0]); + } + + @Test + void nullKeyIsSupported() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + StringEntry e = table.tryGetOrCreateOrNull(null, k -> new StringEntry(k, 0)); + assertNotNull(e); + assertSame(e, table.get(null)); + } + + @Test + void forEachVisitsAllEntries() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 1)); + table.tryGetOrCreateOrNull("b", k -> new StringEntry(k, 2)); + table.tryGetOrCreateOrNull("c", k -> new StringEntry(k, 3)); + Set seen = new HashSet<>(); + table.forEach(e -> seen.add(e.key)); + assertEquals(3, seen.size()); + assertTrue(seen.contains("a")); + assertTrue(seen.contains("b")); + assertTrue(seen.contains("c")); + } + + @Test + void forEachWithContextPassesContext() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + table.tryGetOrCreateOrNull("x", k -> new StringEntry(k, 10)); + table.tryGetOrCreateOrNull("y", k -> new StringEntry(k, 20)); + Set seen = new HashSet<>(); + table.forEach(seen, (ctx, e) -> ctx.add(e.key)); + assertEquals(2, seen.size()); + assertTrue(seen.contains("x")); + assertTrue(seen.contains("y")); + } + + @Test + void concurrentGetOrCreateProducesExactlyOneEntry() throws InterruptedException { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + int threads = 16; + CountDownLatch ready = new CountDownLatch(threads); + CountDownLatch go = new CountDownLatch(1); + AtomicInteger createCount = new AtomicInteger(); + + Thread[] workers = new Thread[threads]; + for (int i = 0; i < threads; i++) { + workers[i] = + new Thread( + () -> { + ready.countDown(); + try { + go.await(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + return; + } + table.tryGetOrCreateOrNull( + "shared", + k -> { + createCount.incrementAndGet(); + return new StringEntry(k, 1); + }); + }); + workers[i].start(); + } + ready.await(); + go.countDown(); + for (Thread w : workers) { + w.join(); + } + + assertEquals(1, table.size()); + assertEquals(1, createCount.get()); + } + + @Test + void chainedEntriesInSameBucketAreAllReachable() { + // All three keys share hash 0, so they land in the same bucket regardless of table size. + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(CollidingEntry.class, 8); + CollidingKey a = new CollidingKey("a", 0); + CollidingKey b = new CollidingKey("b", 0); // same bucket as a + CollidingKey c = new CollidingKey("c", 0); // same bucket + CollidingEntry ea = table.tryGetOrCreateOrNull(a, CollidingEntry::new); + CollidingEntry eb = table.tryGetOrCreateOrNull(b, CollidingEntry::new); + CollidingEntry ec = table.tryGetOrCreateOrNull(c, CollidingEntry::new); + assertEquals(3, table.size()); + assertSame(ea, table.get(a)); + assertSame(eb, table.get(b)); + assertSame(ec, table.get(c)); + assertNull(table.get(new CollidingKey("d", 0))); // same bucket, different label → miss + } + + @Test + void concurrentDistinctKeyInsertionsAreAllRetained() throws InterruptedException { + int threads = 16; + String[] keys = new String[threads]; + for (int i = 0; i < threads; i++) { + keys[i] = "key-" + i; + } + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, threads * 2); + CountDownLatch ready = new CountDownLatch(threads); + CountDownLatch go = new CountDownLatch(1); + + Thread[] workers = new Thread[threads]; + for (int i = 0; i < threads; i++) { + final String key = keys[i]; + workers[i] = + new Thread( + () -> { + ready.countDown(); + try { + go.await(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + return; + } + table.tryGetOrCreateOrNull(key, k -> new StringEntry(k, 1)); + }); + workers[i].start(); + } + ready.await(); + go.countDown(); + for (Thread w : workers) { + w.join(); + } + + assertEquals(threads, table.size()); + for (String key : keys) { + assertNotNull(table.get(key)); + } + } + + @Test + void removeReturnsEntryAndShrinks() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + StringEntry a = table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 1)); + table.tryGetOrCreateOrNull("b", k -> new StringEntry(k, 2)); + assertSame(a, table.remove("a")); + assertEquals(1, table.size()); + assertNull(table.get("a")); + assertNotNull(table.get("b")); + } + + @Test + void removeAbsentKeyReturnsNull() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 1)); + assertNull(table.remove("missing")); + assertEquals(1, table.size()); + } + + @Test + void removeHeadMiddleAndTailOfSameBucketChain() { + // All three keys share hash 0, so a, b, c land in the same bucket and form one chain. + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(CollidingEntry.class, 8); + CollidingKey a = new CollidingKey("a", 0); + CollidingKey b = new CollidingKey("b", 0); + CollidingKey c = new CollidingKey("c", 0); + table.tryGetOrCreateOrNull(a, CollidingEntry::new); + table.tryGetOrCreateOrNull(b, CollidingEntry::new); + table.tryGetOrCreateOrNull(c, CollidingEntry::new); + + // Remove a middle element; the other two stay reachable. + assertNotNull(table.remove(b)); + assertNull(table.get(b)); + assertNotNull(table.get(a)); + assertNotNull(table.get(c)); + assertEquals(2, table.size()); + + // Drain the rest. + assertNotNull(table.remove(c)); + assertNotNull(table.remove(a)); + assertEquals(0, table.size()); + assertNull(table.get(a)); + } + + @Test + void removeIfRemovesMatchingEntries() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 16); + for (int i = 0; i < 10; i++) { + final int v = i; + table.tryGetOrCreateOrNull("k" + i, k -> new StringEntry(k, v)); + } + boolean removed = table.removeIf(e -> e.value % 2 == 0); // removes values 0,2,4,6,8 + assertTrue(removed); + assertEquals(5, table.size()); + Set seen = new HashSet<>(); + table.forEach(e -> seen.add(e.key)); + assertEquals(5, seen.size()); + for (String key : seen) { + assertNotNull(table.get(key)); + } + } + + @Test + void removeIfReturnsFalseWhenNothingMatches() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 1)); + assertFalse(table.removeIf(e -> false)); + assertEquals(1, table.size()); + } + + @Test + void clearEmptiesTableAndLeavesItUsable() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 1)); + table.tryGetOrCreateOrNull("b", k -> new StringEntry(k, 2)); + table.clear(); + assertEquals(0, table.size()); + assertNull(table.get("a")); + assertNull(table.get("b")); + StringEntry c = table.tryGetOrCreateOrNull("c", k -> new StringEntry(k, 3)); + assertSame(c, table.get("c")); + assertEquals(1, table.size()); + } + + @Test + void drainRemovesEveryEntryAndFeedsSink() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 1)); + table.tryGetOrCreateOrNull("b", k -> new StringEntry(k, 2)); + table.tryGetOrCreateOrNull("c", k -> new StringEntry(k, 3)); + + Set drained = new HashSet<>(); + int[] sum = {0}; + table.drain( + e -> { + drained.add(e.key); + sum[0] += e.value; + }); + + assertEquals(new HashSet<>(Arrays.asList("a", "b", "c")), drained); + assertEquals(6, sum[0]); + assertEquals(0, table.size()); + assertNull(table.get("a")); + // table remains usable after drain + StringEntry d = table.tryGetOrCreateOrNull("d", k -> new StringEntry(k, 4)); + assertSame(d, table.get("d")); + assertEquals(1, table.size()); + } + + @Test + void drainWithContextFeedsSink() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 1)); + table.tryGetOrCreateOrNull("b", k -> new StringEntry(k, 2)); + + Set drained = new HashSet<>(); + table.drain(drained, (ctx, e) -> ctx.add(e.key)); + + assertEquals(new HashSet<>(Arrays.asList("a", "b")), drained); + assertEquals(0, table.size()); + } + + @Test + void drainOnEmptyTableInvokesSinkZeroTimes() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + int[] count = {0}; + table.drain(e -> count[0]++); + assertEquals(0, count[0]); + assertEquals(0, table.size()); + } + + /** + * Exercises the volatile-{@code next} removal contract: while one key is repeatedly removed and + * re-added in a shared collision chain, the other keys in that chain must remain continuously + * visible to a concurrent lock-free reader. + */ + @Test + void concurrentReadsStaySafeWhileOneChainMemberChurns() throws InterruptedException { + // All keys share hash 0, putting every key in one bucket so removal splices a chain the + // reader is walking. + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(CollidingEntry.class, 16); + int n = 8; + CollidingKey[] keys = new CollidingKey[n]; + for (int i = 0; i < n; i++) { + keys[i] = new CollidingKey("k" + i, 0); + table.tryGetOrCreateOrNull(keys[i], CollidingEntry::new); + } + CollidingKey churn = keys[0]; // keys[1..] are stable and must never vanish + + AtomicBoolean stop = new AtomicBoolean(false); + AtomicInteger missed = new AtomicInteger(); + Thread reader = + new Thread( + () -> { + while (!stop.get()) { + for (int i = 1; i < n; i++) { + if (table.get(keys[i]) == null) { + missed.incrementAndGet(); + } + } + } + }); + reader.start(); + for (int r = 0; r < 100_000; r++) { + table.remove(churn); + table.tryGetOrCreateOrNull(churn, CollidingEntry::new); + } + stop.set(true); + reader.join(); + + assertEquals(0, missed.get(), "stable chain members must never be unreachable during removal"); + } + + @Test + void tryGetOrCreateOrEvictInsertsWithoutEvictingWhenUnderCapacity() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 8); + Maybe created = + table.tryGetOrCreateOrEvict("a", k -> new StringEntry(k, 1), e -> true); + assertTrue(created.isPresent()); + assertEquals(1, table.size()); + assertSame(created.getOrNull(), table.get("a")); + } + + @Test + void tryGetOrCreateOrEvictReturnsExistingEntryOnHitWithoutEvicting() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 1); + StringEntry a = table.tryGetOrCreateOrNull("a", k -> new StringEntry(k, 1)); + Maybe got = + table.tryGetOrCreateOrEvict( + "a", + k -> { + throw new AssertionError("creator must not run on a hit"); + }, + e -> { + throw new AssertionError("evictable must not run on a hit"); + }); + assertSame(a, got.getOrNull()); + assertEquals(1, table.size()); + } + + @Test + void tryGetOrCreateOrEvictEvictsWhenFullAndInsertsNewEntry() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 1); + table.tryGetOrCreateOrNull("old", k -> new StringEntry(k, 1)); + assertTrue(table.isFull()); + + Maybe created = + table.tryGetOrCreateOrEvict("new", k -> new StringEntry(k, 2), e -> true); + assertTrue(created.isPresent()); + assertEquals("new", created.getOrNull().key); + assertEquals(1, table.size()); + assertNull(table.get("old")); + assertSame(created.getOrNull(), table.get("new")); + } + + @Test + void tryGetOrCreateOrEvictOrNullRefusesWhenFullAndNothingEvictable() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 1); + table.tryGetOrCreateOrNull("old", k -> new StringEntry(k, 1)); + + StringEntry result = + table.tryGetOrCreateOrEvictOrNull("new", k -> new StringEntry(k, 2), e -> false); + assertNull(result); + assertEquals(1, table.size()); + assertNotNull(table.get("old")); + assertNull(table.get("new")); + } + + @Test + void tryGetOrCreateOrEvictOrNullEvictionRunsBeforeThrowingCreator() { + ConcurrentHashtable.D1 table = + ConcurrentHashtable.D1.createCapped(StringEntry.class, 1); + table.tryGetOrCreateOrNull("old", k -> new StringEntry(k, 1)); + + assertThrows( + RuntimeException.class, + () -> + table.tryGetOrCreateOrEvictOrNull( + "new", + k -> { + throw new RuntimeException("boom"); + }, + e -> true)); + + // Eviction already happened before the creator threw: the table is left one entry smaller, + // not corrupted or double-booked. + assertEquals(0, table.size()); + assertNull(table.get("old")); + assertNull(table.get("new")); + } + + private static final class StringEntry extends ConcurrentHashtable.D1.Entry { + final int value; + + StringEntry(String key, int value) { + super(key); + this.value = value; + } + } + + /** Key with a fixed hashCode to force deterministic bucket placement. */ + private static final class CollidingKey { + final String label; + final int fixedHash; + + CollidingKey(String label, int fixedHash) { + this.label = label; + this.fixedHash = fixedHash; + } + + @Override + public int hashCode() { + return fixedHash; + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof CollidingKey)) { + return false; + } + CollidingKey that = (CollidingKey) o; + return fixedHash == that.fixedHash && label.equals(that.label); + } + } + + private static final class CollidingEntry extends ConcurrentHashtable.D1.Entry { + CollidingEntry(CollidingKey key) { + super(key); + } + } +} diff --git a/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableD2Test.java b/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableD2Test.java new file mode 100644 index 00000000000..ae6978a1ebc --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableD2Test.java @@ -0,0 +1,399 @@ +package datadog.trace.util; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; + +class ConcurrentHashtableD2Test { + + @Test + void pairKeysParticipateInIdentity() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + PairEntry ab = table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + PairEntry ac = table.tryGetOrCreateOrNull("a", 2, PairEntry::new); + PairEntry bb = table.tryGetOrCreateOrNull("b", 1, PairEntry::new); + assertEquals(3, table.size()); + assertSame(ab, table.get("a", 1)); + assertSame(ac, table.get("a", 2)); + assertSame(bb, table.get("b", 1)); + assertNull(table.get("a", 3)); + } + + @Test + void getOrCreateOnMissBuildsEntryViaCreator() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + int[] createCount = {0}; + PairEntry created = + table.tryGetOrCreateOrNull( + "a", + 1, + (k1, k2) -> { + createCount[0]++; + return new PairEntry(k1, k2); + }); + assertNotNull(created); + assertEquals("a", created.key1); + assertEquals(Integer.valueOf(1), created.key2); + assertEquals(1, table.size()); + assertEquals(1, createCount[0]); + assertSame(created, table.get("a", 1)); + } + + @Test + void getOrCreateOnHitSkipsCreator() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + PairEntry seeded = table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + int[] createCount = {0}; + PairEntry got = + table.tryGetOrCreateOrNull( + "a", + 1, + (k1, k2) -> { + createCount[0]++; + return new PairEntry(k1, k2); + }); + assertSame(seeded, got); + assertEquals(1, table.size()); + assertEquals(0, createCount[0]); + } + + @Test + void forEachVisitsBothPairs() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + table.tryGetOrCreateOrNull("b", 2, PairEntry::new); + Set seen = new HashSet<>(); + table.forEach(e -> seen.add(e.key1 + ":" + e.key2)); + assertEquals(2, seen.size()); + assertTrue(seen.contains("a:1")); + assertTrue(seen.contains("b:2")); + } + + @Test + void forEachWithContextPassesContextToConsumer() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + table.tryGetOrCreateOrNull("b", 2, PairEntry::new); + Set seen = new HashSet<>(); + table.forEach(seen, (ctx, e) -> ctx.add(e.key1 + ":" + e.key2)); + assertEquals(2, seen.size()); + assertTrue(seen.contains("a:1")); + assertTrue(seen.contains("b:2")); + } + + @Test + void concurrentGetOrCreateProducesExactlyOneEntry() throws InterruptedException { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + int threads = 16; + CountDownLatch ready = new CountDownLatch(threads); + CountDownLatch go = new CountDownLatch(1); + AtomicInteger createCount = new AtomicInteger(); + + Thread[] workers = new Thread[threads]; + for (int i = 0; i < threads; i++) { + workers[i] = + new Thread( + () -> { + ready.countDown(); + try { + go.await(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + return; + } + table.tryGetOrCreateOrNull( + "shared", + 42, + (k1, k2) -> { + createCount.incrementAndGet(); + return new PairEntry(k1, k2); + }); + }); + workers[i].start(); + } + ready.await(); + go.countDown(); + for (Thread w : workers) { + w.join(); + } + + assertEquals(1, table.size()); + assertEquals(1, createCount.get()); + } + + @Test + void chainedEntriesInSameBucketAreAllReachable() { + // key2 = -31 * key1.hashCode() zeroes the combined hash, so all four land in bucket 0 + // regardless of table size. + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + PairEntry e1 = table.tryGetOrCreateOrNull("a", -31 * "a".hashCode(), PairEntry::new); + PairEntry e2 = table.tryGetOrCreateOrNull("b", -31 * "b".hashCode(), PairEntry::new); + PairEntry e3 = table.tryGetOrCreateOrNull("c", -31 * "c".hashCode(), PairEntry::new); + PairEntry e4 = table.tryGetOrCreateOrNull("d", -31 * "d".hashCode(), PairEntry::new); + assertEquals(4, table.size()); + assertSame(e1, table.get("a", -31 * "a".hashCode())); + assertSame(e2, table.get("b", -31 * "b".hashCode())); + assertSame(e3, table.get("c", -31 * "c".hashCode())); + assertSame(e4, table.get("d", -31 * "d".hashCode())); + assertNull(table.get("a", 3)); + } + + @Test + void concurrentDistinctKeyInsertionsAreAllRetained() throws InterruptedException { + int threads = 16; + String[] k1s = new String[threads]; + Integer[] k2s = new Integer[threads]; + for (int i = 0; i < threads; i++) { + k1s[i] = "key-" + i; + k2s[i] = i; + } + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, threads * 2); + CountDownLatch ready = new CountDownLatch(threads); + CountDownLatch go = new CountDownLatch(1); + + Thread[] workers = new Thread[threads]; + for (int i = 0; i < threads; i++) { + final String k1 = k1s[i]; + final Integer k2 = k2s[i]; + workers[i] = + new Thread( + () -> { + ready.countDown(); + try { + go.await(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + return; + } + table.tryGetOrCreateOrNull(k1, k2, PairEntry::new); + }); + workers[i].start(); + } + ready.await(); + go.countDown(); + for (Thread w : workers) { + w.join(); + } + + assertEquals(threads, table.size()); + for (int i = 0; i < threads; i++) { + assertNotNull(table.get(k1s[i], k2s[i])); + } + } + + @Test + void removeReturnsEntryAndShrinks() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + PairEntry ab = table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + table.tryGetOrCreateOrNull("a", 2, PairEntry::new); + assertSame(ab, table.remove("a", 1)); + assertEquals(1, table.size()); + assertNull(table.get("a", 1)); + assertNotNull(table.get("a", 2)); + } + + @Test + void removeAbsentKeyReturnsNull() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + assertNull(table.remove("a", 99)); + assertNull(table.remove("z", 1)); + assertEquals(1, table.size()); + } + + @Test + void removeMiddleOfSameBucketChainKeepsOthersReachable() { + // key2 = -31 * key1.hashCode() zeroes the combined hash, so all three land in one bucket + // chain regardless of table size. + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + table.tryGetOrCreateOrNull("a", -31 * "a".hashCode(), PairEntry::new); + PairEntry mid = table.tryGetOrCreateOrNull("b", -31 * "b".hashCode(), PairEntry::new); + table.tryGetOrCreateOrNull("c", -31 * "c".hashCode(), PairEntry::new); + + assertSame(mid, table.remove("b", -31 * "b".hashCode())); + assertNull(table.get("b", -31 * "b".hashCode())); + assertNotNull(table.get("a", -31 * "a".hashCode())); + assertNotNull(table.get("c", -31 * "c".hashCode())); + assertEquals(2, table.size()); + } + + @Test + void removeIfRemovesMatchingEntries() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 16); + for (int i = 0; i < 10; i++) { + table.tryGetOrCreateOrNull("k", i, PairEntry::new); + } + boolean removed = table.removeIf(e -> e.key2 % 2 == 0); // removes key2 0,2,4,6,8 + assertTrue(removed); + assertEquals(5, table.size()); + Set seen = new HashSet<>(); + table.forEach(e -> seen.add(e.key1 + ":" + e.key2)); + assertEquals(5, seen.size()); + } + + @Test + void removeIfReturnsFalseWhenNothingMatches() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + assertFalse(table.removeIf(e -> false)); + assertEquals(1, table.size()); + } + + @Test + void clearEmptiesTableAndLeavesItUsable() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + table.tryGetOrCreateOrNull("b", 2, PairEntry::new); + table.clear(); + assertEquals(0, table.size()); + assertNull(table.get("a", 1)); + PairEntry c = table.tryGetOrCreateOrNull("c", 3, PairEntry::new); + assertSame(c, table.get("c", 3)); + assertEquals(1, table.size()); + } + + @Test + void drainRemovesEveryEntryAndFeedsSink() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + table.tryGetOrCreateOrNull("a", 2, PairEntry::new); + table.tryGetOrCreateOrNull("b", 1, PairEntry::new); + + Set drained = new HashSet<>(); + table.drain(e -> drained.add(e.key1 + ":" + e.key2)); + + assertEquals(new HashSet<>(Arrays.asList("a:1", "a:2", "b:1")), drained); + assertEquals(0, table.size()); + assertNull(table.get("a", 1)); + PairEntry c = table.tryGetOrCreateOrNull("c", 3, PairEntry::new); + assertSame(c, table.get("c", 3)); + assertEquals(1, table.size()); + } + + @Test + void drainWithContextFeedsSink() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + table.tryGetOrCreateOrNull("b", 2, PairEntry::new); + + Set drained = new HashSet<>(); + table.drain(drained, (ctx, e) -> ctx.add(e.key1 + ":" + e.key2)); + + assertEquals(new HashSet<>(Arrays.asList("a:1", "b:2")), drained); + assertEquals(0, table.size()); + } + + @Test + void tryGetOrCreateOrEvictInsertsWithoutEvictingWhenUnderCapacity() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 8); + Maybe created = table.tryGetOrCreateOrEvict("a", 1, PairEntry::new, e -> true); + assertTrue(created.isPresent()); + assertEquals(1, table.size()); + assertSame(created.getOrNull(), table.get("a", 1)); + } + + @Test + void tryGetOrCreateOrEvictReturnsExistingEntryOnHitWithoutEvicting() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 1); + PairEntry a = table.tryGetOrCreateOrNull("a", 1, PairEntry::new); + Maybe got = + table.tryGetOrCreateOrEvict( + "a", + 1, + (k1, k2) -> { + throw new AssertionError("creator must not run on a hit"); + }, + e -> { + throw new AssertionError("evictable must not run on a hit"); + }); + assertSame(a, got.getOrNull()); + assertEquals(1, table.size()); + } + + @Test + void tryGetOrCreateOrEvictEvictsWhenFullAndInsertsNewEntry() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 1); + table.tryGetOrCreateOrNull("old", 1, PairEntry::new); + assertTrue(table.isFull()); + + Maybe created = table.tryGetOrCreateOrEvict("new", 2, PairEntry::new, e -> true); + assertTrue(created.isPresent()); + assertEquals("new", created.getOrNull().key1); + assertEquals(1, table.size()); + assertNull(table.get("old", 1)); + assertSame(created.getOrNull(), table.get("new", 2)); + } + + @Test + void tryGetOrCreateOrEvictOrNullRefusesWhenFullAndNothingEvictable() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 1); + table.tryGetOrCreateOrNull("old", 1, PairEntry::new); + + PairEntry result = table.tryGetOrCreateOrEvictOrNull("new", 2, PairEntry::new, e -> false); + assertNull(result); + assertEquals(1, table.size()); + assertNotNull(table.get("old", 1)); + assertNull(table.get("new", 2)); + } + + @Test + void tryGetOrCreateOrEvictOrNullEvictionRunsBeforeThrowingCreator() { + ConcurrentHashtable.D2 table = + ConcurrentHashtable.D2.createCapped(PairEntry.class, 1); + table.tryGetOrCreateOrNull("old", 1, PairEntry::new); + + assertThrows( + RuntimeException.class, + () -> + table.tryGetOrCreateOrEvictOrNull( + "new", + 2, + (k1, k2) -> { + throw new RuntimeException("boom"); + }, + e -> true)); + + // Eviction already happened before the creator threw: the table is left one entry smaller, + // not corrupted or double-booked. + assertEquals(0, table.size()); + assertNull(table.get("old", 1)); + assertNull(table.get("new", 2)); + } + + private static final class PairEntry extends ConcurrentHashtable.D2.Entry { + PairEntry(String key1, Integer key2) { + super(key1, key2); + } + } +} diff --git a/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableSizeManagerTest.java b/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableSizeManagerTest.java new file mode 100644 index 00000000000..dbcd7b794e9 --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableSizeManagerTest.java @@ -0,0 +1,357 @@ +package datadog.trace.util; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.test.util.PollingConditions; +import java.util.function.Predicate; +import org.junit.jupiter.api.Test; + +/** + * Exercises {@link ConcurrentHashtable.SizeManager} and {@link ConcurrentHashtable.State} against a + * {@link ConcurrentHashtable.State}, the same shape a custom table driving the static building + * blocks would use. {@link ConcurrentHashtableD1Test} and {@link ConcurrentHashtableD2Test} cover + * the eviction-aware {@code tryGetOrCreateOrEvict} methods built on top of this. + */ +class ConcurrentHashtableSizeManagerTest { + + @Test + void tryReserveSucceedsUnderCapacityAndFailsWhenFull() { + ConcurrentHashtable.SizeManager sizeManager = new ConcurrentHashtable.SizeManager(2); + assertEquals(0, sizeManager.estimateSize()); + assertFalse(sizeManager.isFull()); + + assertTrue(sizeManager.tryReserve()); + assertEquals(1, sizeManager.estimateSize()); + assertFalse(sizeManager.isFull()); + + assertTrue(sizeManager.tryReserve()); + assertEquals(2, sizeManager.estimateSize()); + assertTrue(sizeManager.isFull()); + + assertFalse(sizeManager.tryReserve()); + assertEquals(2, sizeManager.estimateSize()); + } + + @Test + void tryReserveOrEvictReservesDirectlyWhenUnderCapacity() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 2); + + boolean reserved = tryReserveOrEvict(state, e -> true); + assertTrue(reserved); + assertEquals(1, state.sizeManager.estimateSize()); + assertNull(state.buckets.get(0)); // nothing was evicted + } + + @Test + void tryReserveOrEvictEvictsWhenFullAndSomethingMatches() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 1); + TestEntry existing = insertAt(state, 0, "existing"); + assertTrue(state.sizeManager.tryReserve()); + assertTrue(state.sizeManager.isFull()); + + boolean reserved = tryReserveOrEvict(state, e -> true); + assertTrue(reserved); + assertEquals(1, state.sizeManager.estimateSize()); // one evicted, one reserved: net unchanged + assertNull(state.buckets.get(0)); // existing was unlinked + } + + @Test + void tryReserveOrEvictFailsAndLeavesTableUntouchedWhenNothingEvictable() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 1); + TestEntry existing = insertAt(state, 0, "existing"); + assertTrue(state.sizeManager.tryReserve()); + + boolean reserved = tryReserveOrEvict(state, e -> false); + assertFalse(reserved); + assertEquals(1, state.sizeManager.estimateSize()); + assertSame(existing, state.buckets.get(0)); + } + + @Test + void insertReservedSplicesWithoutTouchingTheCountAfterATryReserve() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 2); + assertTrue(state.sizeManager.tryReserve()); + assertEquals(1, state.sizeManager.estimateSize()); + + TestEntry entry = new TestEntry(0, "reserved"); + synchronized (ConcurrentHashtable.getWriteLock(state)) { + ConcurrentHashtable.insertReserved(state, entry.keyHash, entry); + } + + assertSame(entry, state.buckets.get(0)); + // Count reflects only the earlier tryReserve() -- insertReserved must not increment again. + assertEquals(1, state.sizeManager.estimateSize()); + } + + @Test + void evictOneReturnsNullAndLeavesCountUnchangedWhenNothingMatches() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 4); + insertAt(state, 0, "a"); + state.sizeManager.increment(); + + TestEntry evicted = evictOne(state, e -> false); + assertNull(evicted); + assertEquals(1, state.sizeManager.estimateSize()); + } + + @Test + void evictOneUnlinksMatchAndDecrementsCount() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 4); + TestEntry a = insertAt(state, 0, "a"); + TestEntry b = insertAt(state, 1, "b"); + state.sizeManager.increment(); + state.sizeManager.increment(); + + TestEntry evicted = evictOne(state, e -> e.label.equals("a")); + assertSame(a, evicted); + assertNull(state.buckets.get(0)); + assertSame(b, state.buckets.get(1)); // untouched + assertEquals(1, state.sizeManager.estimateSize()); + } + + /** + * Verifies the cursor-resume contract from {@link ConcurrentHashtable.SizeManager#evictOne}: each + * scan resumes where the previous eviction left off, so among several equally-matching candidates + * the one nearest (forward from the cursor, wrapping) is picked first -- not always the lowest + * bucket index. + */ + @Test + void evictOneResumesFromLastEvictedBucketAndWrapsAround() { + // Bucket-array length 4: keyHash i lands in bucket i. + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 4); + TestEntry e0 = insertAt(state, 0, "e0"); + insertAt(state, 2, "e2"); + TestEntry e3 = insertAt(state, 3, "e3"); + state.sizeManager.increment(); + state.sizeManager.increment(); + state.sizeManager.increment(); + + // First eviction: scan starts at cursor 0, finds bucket 2 first among evictable entries + // (only e2 matches here) -- sets the cursor to 2. + TestEntry firstEvicted = evictOne(state, e -> e.label.equals("e2")); + assertEquals("e2", firstEvicted.label); + + // Second eviction: both e0 (bucket 0) and e3 (bucket 3) match. Scanning resumes at the + // cursor (2) and goes forward before wrapping, so bucket 3 (e3) is found before bucket 0. + TestEntry secondEvicted = evictOne(state, e -> true); + assertSame(e3, secondEvicted); + assertSame(e0, state.buckets.get(0)); // e0 not yet touched + + // Third eviction: only e0 remains. The cursor is now past bucket 3, so the scan must wrap + // around to bucket 0 to find it. + TestEntry thirdEvicted = evictOne(state, e -> true); + assertSame(e0, thirdEvicted); + assertEquals(0, state.sizeManager.estimateSize()); + } + + @Test + void evictAllRemovesEveryMatchAndReturnsCount() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 8); + for (int i = 0; i < 6; i++) { + insertAt(state, i, "e" + i); + state.sizeManager.increment(); + } + // Evict everything except bucket 1 and bucket 4. + int count = evictAll(state, e -> !e.label.equals("e1") && !e.label.equals("e4")); + + assertEquals(4, count); + assertEquals(2, state.sizeManager.estimateSize()); + assertNotNullLabel(state, 1, "e1"); + assertNotNullLabel(state, 4, "e4"); + for (int i : new int[] {0, 2, 3, 5}) { + assertNull(state.buckets.get(i)); + } + } + + @Test + void evictAllResetsCursorSoSubsequentEvictOneScansFromBucketZero() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 4); + insertAt(state, 2, "a"); + state.sizeManager.increment(); + // Advance the cursor away from 0 via a successful eviction at bucket 2. + evictOne(state, e -> true); + + // A full pass that removes nothing still resets the scan position (per evictAll's contract). + int count = evictAll(state, e -> false); + assertEquals(0, count); + + TestEntry e0 = insertAt(state, 0, "e0"); + TestEntry e3 = insertAt(state, 3, "e3"); + state.sizeManager.increment(); + state.sizeManager.increment(); + + // With the cursor reset to 0, the forward scan reaches bucket 0 before bucket 3. + TestEntry evicted = evictOne(state, e -> true); + assertSame(e0, evicted); + assertSame(e3, state.buckets.get(3)); + } + + @Test + void resetZeroesCountAndScanPosition() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 4); + insertAt(state, 2, "a"); + state.sizeManager.increment(); + evictOne(state, e -> true); // advances the cursor to 2, count back to 0 + state.sizeManager.increment(); // pretend a fresh entry was inserted + + state.sizeManager.reset(); + assertEquals(0, state.sizeManager.estimateSize()); + + TestEntry e0 = insertAt(state, 0, "e0"); + TestEntry e3 = insertAt(state, 3, "e3"); + state.sizeManager.increment(); + state.sizeManager.increment(); + TestEntry evicted = evictOne(state, e -> true); + assertSame(e0, evicted); // scan restarted from bucket 0, per reset() + assertSame(e3, state.buckets.get(3)); + } + + @Test + void stateCreateCappedBundlesBucketsAndSizeManager() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 3); + assertEquals(0, state.sizeManager.estimateSize()); + assertEquals(3, state.sizeManager.capacity()); + assertTrue(state.buckets.length() >= 3); + } + + @Test + void stateLevelTryReserveOrEvictAndEvictOneAndEvictAllDelegateToSizeManager() { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 1); + synchronized (ConcurrentHashtable.getWriteLock(state)) { + ConcurrentHashtable.insertHeadEntryAt(state, 0, new TestEntry(0, "a")); + } + state.sizeManager.increment(); + assertTrue(ConcurrentHashtable.isFull(state)); + + // Table is full: tryReserveOrEvict evicts "a" and reserves the freed slot for the caller, who + // is now responsible for splicing in the entry that occupies it -- via insertReserved, since + // the reservation already happened and a plain insertHeadEntryAt/increment would double-count. + // Both steps go in ONE critical section: tryReserveOrEvict is self-locking, so on its own it + // leaves a window where a drain/clear could reset the count out from under the reservation. + synchronized (ConcurrentHashtable.getWriteLock(state)) { + boolean reserved = ConcurrentHashtable.tryReserveOrEvict(state, e -> true); + assertTrue(reserved); + assertEquals(1, ConcurrentHashtable.estimateSize(state)); + assertNull(state.buckets.get(0)); // "a" was evicted; the reserved slot has no entry yet + ConcurrentHashtable.insertReserved(state, 0, new TestEntry(0, "reserved")); + } + + int evicted = ConcurrentHashtable.evictAll(state, e -> true); + assertEquals(1, evicted); + assertEquals(0, ConcurrentHashtable.estimateSize(state)); + assertFalse(ConcurrentHashtable.isFull(state)); + + synchronized (ConcurrentHashtable.getWriteLock(state)) { + ConcurrentHashtable.insertHeadEntryAt(state, 0, new TestEntry(0, "b")); + } + state.sizeManager.increment(); + TestEntry viaEvictOne = ConcurrentHashtable.evictOne(state, e -> e.label.equals("b")); + assertNotNull(viaEvictOne); + assertEquals("b", viaEvictOne.label); + assertEquals(0, ConcurrentHashtable.estimateSize(state)); + } + + /** + * Holding the write lock across {@code tryReserveOrEvict} + {@code insertReserved} keeps a + * concurrent {@link ConcurrentHashtable#clear(ConcurrentHashtable.State)} out of the gap. Were + * the clear able to land in between, its {@code SizeManager.reset()} would void the outstanding + * reservation and the insert would link an entry the count never learns about -- an undercount + * that never heals, and a capped table quietly over its cap from then on. + */ + @Test + void clearCannotInterleaveBetweenReservationAndInsert() throws InterruptedException { + ConcurrentHashtable.State state = + ConcurrentHashtable.State.createCapped(TestEntry.class, 1); + insertAt(state, 0, "a"); + state.sizeManager.increment(); + assertTrue(ConcurrentHashtable.isFull(state)); + + Thread clearer = new Thread(() -> ConcurrentHashtable.clear(state), "clearer"); + synchronized (ConcurrentHashtable.getWriteLock(state)) { + clearer.start(); + // Wait until the clear is definitely queued on the monitor we hold, so the interleaving under + // test is the one actually attempted rather than one the scheduler happened to avoid. + new PollingConditions() + .eventually(() -> assertEquals(Thread.State.BLOCKED, clearer.getState())); + + assertTrue(ConcurrentHashtable.tryReserveOrEvict(state, e -> true)); + ConcurrentHashtable.insertReserved(state, 0, new TestEntry(0, "reserved")); + assertEquals(1, ConcurrentHashtable.estimateSize(state)); + } + clearer.join(); + + // The clear ran strictly after the pair, so the table is exactly post-clear: no entry, no + // count, and -- the point -- the two agree. + assertEquals(0, ConcurrentHashtable.estimateSize(state)); + assertNull(state.buckets.get(0)); + assertFalse(ConcurrentHashtable.isFull(state)); + } + + private static void assertNotNullLabel( + ConcurrentHashtable.State state, int index, String label) { + TestEntry e = state.buckets.get(index); + assertNotNull(e); + assertEquals(label, e.label); + } + + /** Inserts a fresh entry at the given bucket index. Bucket-array length must exceed index. */ + private static TestEntry insertAt( + ConcurrentHashtable.State state, int index, String label) { + TestEntry entry = new TestEntry(index, label); + synchronized (ConcurrentHashtable.getWriteLock(state)) { + ConcurrentHashtable.insertHeadEntryAt(state, index, entry); + } + return entry; + } + + /** {@code sizeManager.tryReserveOrEvict}, taking the write lock {@code @GuardedBy} requires. */ + private static boolean tryReserveOrEvict( + ConcurrentHashtable.State state, Predicate evictable) { + synchronized (ConcurrentHashtable.getWriteLock(state)) { + return state.sizeManager.tryReserveOrEvict(state.buckets, evictable); + } + } + + /** {@code sizeManager.evictOne}, taking the write lock {@code @GuardedBy} requires. */ + private static TestEntry evictOne( + ConcurrentHashtable.State state, Predicate evictable) { + synchronized (ConcurrentHashtable.getWriteLock(state)) { + return state.sizeManager.evictOne(state.buckets, evictable); + } + } + + /** {@code sizeManager.evictAll}, taking the write lock {@code @GuardedBy} requires. */ + private static int evictAll( + ConcurrentHashtable.State state, Predicate evictable) { + synchronized (ConcurrentHashtable.getWriteLock(state)) { + return state.sizeManager.evictAll(state.buckets, evictable); + } + } + + /** Entry with a caller-controlled {@code keyHash} so tests can place it in an exact bucket. */ + private static final class TestEntry extends ConcurrentHashtable.Entry { + final String label; + + TestEntry(long keyHash, String label) { + super(keyHash); + this.label = label; + } + } +} diff --git a/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableStaticsTest.java b/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableStaticsTest.java new file mode 100644 index 00000000000..0a2839b6fa7 --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/util/ConcurrentHashtableStaticsTest.java @@ -0,0 +1,407 @@ +package datadog.trace.util; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReferenceArray; +import org.junit.jupiter.api.Test; + +/** + * Exercises the {@code static} building blocks that {@link ConcurrentHashtable} exposes for the + * caller-owned-array path — the custom-table API used when {@link ConcurrentHashtable.D1}/{@link + * ConcurrentHashtable.D2}'s object-key constraints don't fit (primitive keys, higher arity, extra + * per-entry fields). {@link IntTable} below is a minimal hand-written table with a primitive {@code + * int} key, driving the same lock-free-read / locked-write recipe the class Javadoc documents. + */ +class ConcurrentHashtableStaticsTest { + + @Test + void sizeForRoundsUpToPowerOfTwo() { + assertEquals(1, ConcurrentHashtable.sizeFor(1)); + assertEquals(8, ConcurrentHashtable.sizeFor(5)); + assertEquals(8, ConcurrentHashtable.sizeFor(8)); + assertEquals(16, ConcurrentHashtable.sizeFor(9)); + } + + @Test + void createFixedBucketsAllocatesPowerOfTwoSpine() { + AtomicReferenceArray buckets = + ConcurrentHashtable.createFixedBuckets(IntEntry.class, 10); + assertEquals(16, buckets.length()); + assertNull(buckets.get(0)); + } + + @Test + void getWriteLockIsStableAndNonNull() { + AtomicReferenceArray buckets = + ConcurrentHashtable.createFixedBuckets(IntEntry.class, 8); + Object lock = ConcurrentHashtable.getWriteLock(buckets); + assertNotNull(lock); + assertSame(lock, ConcurrentHashtable.getWriteLock(buckets)); + } + + @Test + void bucketIndexMasksToArrayLength() { + AtomicReferenceArray buckets = + ConcurrentHashtable.createFixedBuckets(IntEntry.class, 8); // length 8, mask 7 + assertEquals(0, ConcurrentHashtable.bucketIndex(buckets, 8L)); + assertEquals(1, ConcurrentHashtable.bucketIndex(buckets, 9L)); + assertEquals(7, ConcurrentHashtable.bucketIndex(buckets, 7L)); + } + + @Test + void insertGetAndRemoveViaStatics() { + IntTable table = new IntTable(8); + IntEntry a = table.getOrCreate(1, 100); + IntEntry b = table.getOrCreate(2, 200); + assertEquals(2, table.size.get()); + assertSame(a, table.get(1)); + assertSame(b, table.get(2)); + assertNull(table.get(3)); + + // getOrCreate on a hit returns the existing entry, no new insert. + assertSame(a, table.getOrCreate(1, 999)); + assertEquals(2, table.size.get()); + + assertSame(a, table.remove(1)); + assertNull(table.get(1)); + assertNull(table.remove(1)); // already gone + assertEquals(1, table.size.get()); + } + + @Test + void insertHeadEntryForPlacesInBucketMaskedFromKeyHash() { + AtomicReferenceArray buckets = + ConcurrentHashtable.createFixedBuckets(IntEntry.class, 8); // mask 7 + IntEntry e = new IntEntry(9, 1); // keyHash 9 → bucket 1 + synchronized (ConcurrentHashtable.getWriteLock(buckets)) { + ConcurrentHashtable.insertHeadEntryFor(buckets, e.keyHash, e); + } + assertSame(e, ConcurrentHashtable.bucketFor(buckets, 9L)); // masks keyHash to the bucket index + assertSame( + e, ConcurrentHashtable.bucketAt(buckets, 1)); // same slot, addressed directly by index + assertNull(buckets.get(0)); + } + + @Test + void unlinkRemovesHeadMiddleAndTailOfChain() { + // Capacity 1 → mask 0 → every key lands in bucket 0, forming one chain. + IntTable table = new IntTable(1); + IntEntry a = table.getOrCreate(1, 1); + IntEntry b = table.getOrCreate(2, 2); + IntEntry c = table.getOrCreate(3, 3); + assertEquals(3, table.size.get()); + + // Remove the middle: head and tail stay reachable. + assertSame(b, table.remove(2)); + assertNull(table.get(2)); + assertSame(a, table.get(1)); + assertSame(c, table.get(3)); + assertEquals(2, table.size.get()); + + // Remove the head, then the last remaining. + assertSame(c, table.remove(3)); + assertSame(a, table.remove(1)); + assertEquals(0, table.size.get()); + assertNull(table.get(1)); + } + + @Test + void staticRemoveIfRemovesMatchingAndDecrementsSize() { + IntTable table = new IntTable(16); + for (int i = 0; i < 10; i++) { + table.getOrCreate(i, i); + } + boolean removed = + ConcurrentHashtable.removeIf(table.buckets, table.size, e -> e.value % 2 == 0); + assertTrue(removed); + assertEquals(5, table.size.get()); + for (int i = 0; i < 10; i++) { + if (i % 2 == 0) { + assertNull(table.get(i)); + } else { + assertNotNull(table.get(i)); + } + } + } + + @Test + void staticRemoveIfReturnsFalseWhenNothingMatches() { + IntTable table = new IntTable(8); + table.getOrCreate(1, 1); + assertFalse(ConcurrentHashtable.removeIf(table.buckets, table.size, e -> false)); + assertEquals(1, table.size.get()); + } + + @Test + void staticDrainRemovesEveryEntryAndFeedsSink() { + IntTable table = new IntTable(8); + table.getOrCreate(1, 10); + table.getOrCreate(2, 20); + table.getOrCreate(3, 30); + + Set keys = new HashSet<>(); + int[] sum = {0}; + ConcurrentHashtable.drain( + table.buckets, + e -> { + keys.add(e.key); + sum[0] += e.value; + }); + + assertEquals(new HashSet<>(java.util.Arrays.asList(1, 2, 3)), keys); + assertEquals(60, sum[0]); + // drain does not touch size accounting on the static path — the caller resets it. + for (int i = 1; i <= 3; i++) { + assertNull(table.get(i)); + } + } + + @Test + void staticDrainWithContextFeedsSink() { + IntTable table = new IntTable(8); + table.getOrCreate(1, 10); + table.getOrCreate(2, 20); + + Set keys = new HashSet<>(); + ConcurrentHashtable.drain(table.buckets, keys, (ctx, e) -> ctx.add(e.key)); + + assertEquals(new HashSet<>(java.util.Arrays.asList(1, 2)), keys); + assertNull(table.get(1)); + } + + @Test + void staticClearEmptiesEveryBucket() { + IntTable table = new IntTable(8); + table.getOrCreate(1, 1); + table.getOrCreate(2, 2); + ConcurrentHashtable.clear(table.buckets); + assertNull(table.get(1)); + assertNull(table.get(2)); + for (int i = 0; i < table.buckets.length(); i++) { + assertNull(table.buckets.get(i)); + } + } + + @Test + void staticForEachVisitsEveryEntry() { + IntTable table = new IntTable(8); + table.getOrCreate(1, 1); + table.getOrCreate(2, 2); + table.getOrCreate(3, 3); + + Set seen = new HashSet<>(); + ConcurrentHashtable.forEach(table.buckets, e -> seen.add(e.key)); + assertEquals(new HashSet<>(java.util.Arrays.asList(1, 2, 3)), seen); + + Set seenCtx = new HashSet<>(); + ConcurrentHashtable.forEach(table.buckets, seenCtx, (ctx, e) -> ctx.add(e.key)); + assertEquals(new HashSet<>(java.util.Arrays.asList(1, 2, 3)), seenCtx); + } + + @Test + void insertHeadEntryWithoutLockTripsAssertion() { + assumeTrue(assertionsEnabled(), "assert-guard test requires -ea"); + AtomicReferenceArray buckets = + ConcurrentHashtable.createFixedBuckets(IntEntry.class, 8); + assertThrows( + AssertionError.class, + () -> ConcurrentHashtable.insertHeadEntryAt(buckets, 0, new IntEntry(1, 1))); + } + + @Test + void unlinkWithoutLockTripsAssertion() { + assumeTrue(assertionsEnabled(), "assert-guard test requires -ea"); + AtomicReferenceArray buckets = + ConcurrentHashtable.createFixedBuckets(IntEntry.class, 8); + IntEntry e = new IntEntry(1, 1); + synchronized (ConcurrentHashtable.getWriteLock(buckets)) { + ConcurrentHashtable.insertHeadEntryAt(buckets, 0, e); + } + assertThrows(AssertionError.class, () -> ConcurrentHashtable.unlink(buckets, 0, null, e)); + } + + @Test + void concurrentGetOrCreateViaStaticsProducesExactlyOneEntry() throws InterruptedException { + IntTable table = new IntTable(8); + int threads = 16; + CountDownLatch ready = new CountDownLatch(threads); + CountDownLatch go = new CountDownLatch(1); + AtomicInteger createCount = new AtomicInteger(); + + Thread[] workers = new Thread[threads]; + for (int i = 0; i < threads; i++) { + workers[i] = + new Thread( + () -> { + ready.countDown(); + try { + go.await(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + return; + } + table.getOrCreateCounting(7, createCount); + }); + workers[i].start(); + } + ready.await(); + go.countDown(); + for (Thread w : workers) { + w.join(); + } + + assertEquals(1, table.size.get()); + assertEquals(1, createCount.get()); + } + + @Test + void concurrentReadsStaySafeWhileOneChainMemberChurnsViaStatics() throws InterruptedException { + // Capacity 1 puts every key in one bucket so unlink splices a chain the reader is walking. + IntTable table = new IntTable(1); + int n = 8; + for (int i = 0; i < n; i++) { + table.getOrCreate(i, i); + } + int churn = 0; // keys 1..n-1 are stable and must never vanish + + AtomicBoolean stop = new AtomicBoolean(false); + AtomicInteger missed = new AtomicInteger(); + Thread reader = + new Thread( + () -> { + while (!stop.get()) { + for (int i = 1; i < n; i++) { + if (table.get(i) == null) { + missed.incrementAndGet(); + } + } + } + }); + reader.start(); + for (int r = 0; r < 100_000; r++) { + table.remove(churn); + table.getOrCreate(churn, churn); + } + stop.set(true); + reader.join(); + + assertEquals(0, missed.get(), "stable chain members must never be unreachable during removal"); + } + + private static boolean assertionsEnabled() { + boolean enabled = false; + assert enabled = true; + return enabled; + } + + /** Primitive-{@code int}-key entry: no boxing, keyHash is the key itself. */ + private static final class IntEntry extends ConcurrentHashtable.Entry { + final int key; + final int value; + + IntEntry(int key, int value) { + super(key); + this.key = key; + this.value = value; + } + + boolean matches(int key) { + return this.key == key; + } + } + + /** + * Minimal hand-written table over a caller-owned {@link AtomicReferenceArray}, following the + * documented recipe: lock-free pre-check, then re-check + mutate under {@code + * getWriteLock(buckets)}. + */ + private static final class IntTable { + final AtomicReferenceArray buckets; + final AtomicInteger size = new AtomicInteger(); + + IntTable(int capacity) { + this.buckets = ConcurrentHashtable.createFixedBuckets(IntEntry.class, capacity); + } + + IntEntry get(int key) { + for (IntEntry e = ConcurrentHashtable.bucketFor(buckets, (long) key); + e != null; + e = e.next()) { + if (e.matches(key)) { + return e; + } + } + return null; + } + + IntEntry getOrCreate(int key, int value) { + int index = ConcurrentHashtable.bucketIndex(buckets, key); + for (IntEntry e = ConcurrentHashtable.bucketAt(buckets, index); e != null; e = e.next()) { + if (e.matches(key)) { + return e; + } + } + synchronized (ConcurrentHashtable.getWriteLock(buckets)) { + for (IntEntry e = ConcurrentHashtable.bucketAt(buckets, index); e != null; e = e.next()) { + if (e.matches(key)) { + return e; + } + } + IntEntry created = new IntEntry(key, value); + ConcurrentHashtable.insertHeadEntryAt(buckets, index, created); + size.incrementAndGet(); + return created; + } + } + + /** {@link #getOrCreate} variant that counts real creations, for the exactly-once race test. */ + IntEntry getOrCreateCounting(int key, AtomicInteger createCount) { + int index = ConcurrentHashtable.bucketIndex(buckets, key); + for (IntEntry e = ConcurrentHashtable.bucketAt(buckets, index); e != null; e = e.next()) { + if (e.matches(key)) { + return e; + } + } + synchronized (ConcurrentHashtable.getWriteLock(buckets)) { + for (IntEntry e = ConcurrentHashtable.bucketAt(buckets, index); e != null; e = e.next()) { + if (e.matches(key)) { + return e; + } + } + createCount.incrementAndGet(); + IntEntry created = new IntEntry(key, 0); + ConcurrentHashtable.insertHeadEntryAt(buckets, index, created); + size.incrementAndGet(); + return created; + } + } + + IntEntry remove(int key) { + int index = ConcurrentHashtable.bucketIndex(buckets, key); + synchronized (ConcurrentHashtable.getWriteLock(buckets)) { + IntEntry prev = null; + for (IntEntry e = ConcurrentHashtable.bucketAt(buckets, index); e != null; e = e.next()) { + if (e.matches(key)) { + ConcurrentHashtable.unlink(buckets, index, prev, e); + size.decrementAndGet(); + return e; + } + prev = e; + } + return null; + } + } + } +}