drain() {
/**
* Finds a log group with the same level, message, and
- * throwable in the selected bucket.
+ * throwable as {@code keyHash}'s candidates.
*
* Note, throwables are matched by identity or by class and stack trace.
*
*
The bucket chain supports lock-free reads. A caller that inserts after a miss must repeat
* the search under the table write lock.
*
- * @param bucketIndex bucket selected for {@code keyHash}
* @param keyHash precomputed hash of the level, message, and throwable class
* @param logLevel log level to match
* @param message message to match
@@ -150,19 +125,10 @@ public Collection drain() {
*/
@Nullable
private RawLogMessage find(
- int bucketIndex,
- long keyHash,
- String logLevel,
- String message,
- @Nullable Throwable throwable) {
- // Start searching from given bucket, and follow the entry next links
+ long keyHash, String logLevel, String message, @Nullable Throwable throwable) {
StackTraceElement[] stackTrace = null;
- for (RawLogMessage entry = bucketAt(rawLogMessages, bucketIndex);
- entry != null;
- entry = entry.next()) {
- if (entry.keyHash != keyHash
- || !Objects.equals(logLevel, entry.logLevel)
- || !Objects.equals(message, entry.message)) {
+ for (RawLogMessage entry : ConcurrentHashtable.hashIterable(rawLogMessages, keyHash)) {
+ if (!Objects.equals(logLevel, entry.logLevel) || !Objects.equals(message, entry.message)) {
continue;
}
// throwables are more costly to compare, check first the identity
@@ -190,7 +156,7 @@ private RawLogMessage find(
* match. The first message supplies the tags and timestamp; later messages only increment the
* occurrence count.
*/
- public static final class RawLogMessage extends ConcurrentHashtable.Entry {
+ public static final class RawLogMessage extends ConcurrentHashtable.Entry {
private static final AtomicIntegerFieldUpdater LIVE_OCCURRENCE_COUNT_UPDATER =
AtomicIntegerFieldUpdater.newUpdater(RawLogMessage.class, "liveOccurrenceCount");
@@ -203,8 +169,12 @@ public static final class RawLogMessage extends ConcurrentHashtable.Entry {
/** Number of equivalent log messages captured when this group was drained. */
public int count;
- /** Live counter equivalent log messages accumulated in this group. */
- private volatile int liveOccurrenceCount = 1;
+ /**
+ * Live counter equivalent log messages accumulated in this group. Starts zeroed so a caller can
+ * unconditionally {@link #increment()} once after a find-or-insert, whether it landed this
+ * instance or an existing match.
+ */
+ private volatile int liveOccurrenceCount = 0;
private volatile StackTraceElement[] cachedStackTrace = null;
@@ -240,11 +210,7 @@ private void snapshotCount() {
}
@Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- RawLogMessage that = (RawLogMessage) o;
-
+ public boolean matches(RawLogMessage that) {
if (!Objects.equals(logLevel, that.logLevel)) return false;
if (!Objects.equals(message, that.message)) return false;
@@ -264,6 +230,13 @@ public boolean equals(Object o) {
}
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ return matches((RawLogMessage) o);
+ }
+
@Override
public int hashCode() {
return (int) keyHash;
diff --git a/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java b/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java
index cc0b53ee10e..48cdecb1c65 100644
--- a/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java
+++ b/internal-api/src/main/java/datadog/trace/util/ConcurrentHashtable.java
@@ -1,9 +1,14 @@
package datadog.trace.util;
+import datadog.trace.api.function.Strategy;
+import datadog.trace.api.function.StrategyConsumer;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReferenceArray;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
@@ -69,10 +74,16 @@ private ConcurrentHashtable() {}
* 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}.
+ *
+ * The self-bound type parameter ({@code TEntry extends Entry}, mirroring {@code Enum>}) exists so {@link #matches(Entry)} can compare two already-built entries
+ * without an {@code Object}/{@code instanceof} boundary. It requires every concrete subclass to
+ * declare itself as its own type argument (see {@link D1.Entry}/{@link D2.Entry}); Java has no
+ * true {@code Self} type, so this is enforced by convention, not the compiler.
*/
- public abstract static class Entry {
+ public abstract static class Entry> {
public final long keyHash;
- private volatile Entry next = null;
+ private volatile TEntry next = null;
protected Entry(long keyHash) {
this.keyHash = keyHash;
@@ -81,15 +92,26 @@ protected Entry(long 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) {
+ final void setNext(TEntry next) {
this.next = next;
}
- @SuppressWarnings("unchecked")
@Nullable
- public final TEntry next() {
- return (TEntry) this.next;
+ public final TEntry next() {
+ return this.next;
}
+
+ /**
+ * Returns {@code true} if {@code other} is logically the same entry as this one (same key(s)),
+ * used to compare two already-built entries under the write lock (e.g. in {@link
+ * Reservation#tryGetOrInsertOrNull}). Deliberately narrower than {@code equals}/{@code
+ * hashCode}: this class doesn't need reflexive/symmetric-with-null-and-unrelated-types contract
+ * baggage, and a bespoke method avoids entries accidentally working as {@code HashSet}/{@code
+ * HashMap} keys via an unrelated identity notion. {@link D1.Entry}/{@link D2.Entry} implement
+ * this in terms of their existing key-based {@code matches(...)}, so most callers never write
+ * it directly.
+ */
+ public abstract boolean matches(@Nonnull TEntry other);
}
/**
@@ -105,10 +127,17 @@ 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}.
*
+ * Deliberately parameterized on {@code K} alone, not self-bound on the concrete subclass:
+ * {@link D1} stores and links entries internally as {@code Entry} and casts back to {@code
+ * TEntry} at its API boundary, trading one unchecked cast (always sound -- the table only ever
+ * holds instances the caller's own {@code creator} produced) for a simpler subclass signature,
+ * e.g. {@code class MyEntry extends D1.Entry} rather than {@code D1.Entry}.
+ *
* @param the key type
*/
- public abstract static class Entry extends ConcurrentHashtable.Entry {
- final K key;
+ public abstract static class Entry extends ConcurrentHashtable.Entry> {
+ @Nullable final K key;
protected Entry(@Nullable K key) {
super(hash(key));
@@ -127,6 +156,12 @@ public boolean matches(@Nullable Object key) {
return Objects.equals(key, this.key);
}
+ /** {@link ConcurrentHashtable.Entry#matches(Entry)} in terms of the key-based overload. */
+ @Override
+ public final boolean matches(@Nonnull Entry other) {
+ return matches(other.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
@@ -137,21 +172,51 @@ public static long hash(@Nullable Object key) {
}
}
- private final State state;
+ private final State> state;
- private D1(State state) {
+ private D1(State> state) {
this.state = state;
}
/**
* Creates a fixed-size table holding at most {@code maxCapacity} entries. {@code entryClass} is
- * used only to infer the concrete entry type; entries are created by the functions passed to
- * the insertion methods. The table does not resize.
+ * used only to allocate the backing array with the right component type; entries themselves are
+ * created by the {@code creator}/{@code evictable} functions passed to the insertion methods.
+ * The table does not resize.
*/
@Nonnull
+ @SuppressWarnings("unchecked")
public static > D1 createBounded(
@Nonnull Class entryClass, int maxCapacity) {
- return new D1<>(State.createBounded(entryClass, maxCapacity));
+ // entryClass is erased away (see ConcurrentHashtable#createFixedBuckets), so treating it as
+ // Class> instead of the caller's concrete Class is safe.
+ Class> baseEntryClass = (Class>) (Class) entryClass;
+ return new D1<>(ConcurrentHashtable.createBounded(baseEntryClass, maxCapacity));
+ }
+
+ /**
+ * Sound because every entry ever inserted into {@link #state} was produced as a {@code TEntry}.
+ */
+ @SuppressWarnings("unchecked")
+ private TEntry cast(@Nullable Entry entry) {
+ return (TEntry) entry;
+ }
+
+ /** See {@link #cast(Entry)}; casts the functional-interface reference, not each element. */
+ @SuppressWarnings("unchecked")
+ private Predicate> castPredicate(Predicate predicate) {
+ return (Predicate>) predicate;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Consumer> castConsumer(Consumer consumer) {
+ return (Consumer>) consumer;
+ }
+
+ @SuppressWarnings("unchecked")
+ private BiConsumer> castConsumer(
+ BiConsumer consumer) {
+ return (BiConsumer>) consumer;
}
public int size() {
@@ -165,11 +230,11 @@ public boolean isFull() {
@Nullable
public TEntry get(@Nullable K key) {
long keyHash = D1.Entry.hash(key);
- for (TEntry curEntry = bucketFor(state, keyHash);
+ for (Entry curEntry = bucketFor(state, keyHash);
curEntry != null;
curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key)) {
- return curEntry;
+ return cast(curEntry);
}
}
return null;
@@ -182,7 +247,7 @@ public TEntry get(@Nullable K key) {
*/
@Nonnull
public Maybe tryGetOrCreate(
- @Nullable K key, @Nonnull Function creator) {
+ @Nullable K key, @Strategy @Nonnull Function creator) {
return Maybe.of(tryGetOrCreateOrNull(key, creator));
}
@@ -192,22 +257,27 @@ public Maybe tryGetOrCreate(
* {@code key} was not already present. Re-checks under the lock to avoid duplicate entries
* under concurrent misses.
*/
+ @StrategyConsumer
@Nullable
public TEntry tryGetOrCreateOrNull(
- @Nullable K key, @Nonnull Function creator) {
+ @Nullable K key, @Strategy @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()) {
+ for (Entry curEntry = bucketAt(state, index);
+ curEntry != null;
+ curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key)) {
- return curEntry;
+ return cast(curEntry);
}
}
- synchronized (getTableWriteLock(state)) {
- for (TEntry curEntry = bucketAt(state, index);
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
+ for (Entry curEntry = bucketAt(state, index);
curEntry != null;
curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key)) {
- return curEntry;
+ return cast(curEntry);
}
}
// isFull() is checked before creating the entry, not before reserving a slot for it:
@@ -221,6 +291,8 @@ public TEntry tryGetOrCreateOrNull(
insertHeadEntryAt(state, index, newEntry);
state.sizeManager.increment();
return newEntry;
+ } finally {
+ lock.unlock();
}
}
@@ -233,8 +305,8 @@ public TEntry tryGetOrCreateOrNull(
@Nonnull
public Maybe tryGetOrCreateOrEvict(
@Nullable K key,
- @Nonnull Function creator,
- @Nonnull Predicate evictable) {
+ @Strategy @Nonnull Function creator,
+ @Strategy @Nonnull Predicate evictable) {
return Maybe.of(tryGetOrCreateOrEvictOrNull(key, creator, evictable));
}
@@ -245,34 +317,41 @@ public Maybe tryGetOrCreateOrEvict(
* 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.
*/
+ @StrategyConsumer
@Nullable
public TEntry tryGetOrCreateOrEvictOrNull(
@Nullable K key,
- @Nonnull Function creator,
- @Nonnull Predicate evictable) {
+ @Strategy @Nonnull Function creator,
+ @Strategy @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()) {
+ for (Entry curEntry = bucketAt(state, index);
+ curEntry != null;
+ curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key)) {
- return curEntry;
+ return cast(curEntry);
}
}
- synchronized (getTableWriteLock(state)) {
- for (TEntry curEntry = bucketAt(state, index);
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
+ for (Entry curEntry = bucketAt(state, index);
curEntry != null;
curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key)) {
- return curEntry;
+ return cast(curEntry);
}
}
if (state.sizeManager.isFull()
- && state.sizeManager.evictOne(state.buckets, evictable) == null) {
+ && state.sizeManager.evictOne(state.buckets, castPredicate(evictable)) == null) {
return null;
}
TEntry newEntry = creator.apply(key);
insertHeadEntryAt(state, index, newEntry);
state.sizeManager.increment();
return newEntry;
+ } finally {
+ lock.unlock();
}
}
@@ -285,18 +364,22 @@ public TEntry tryGetOrCreateOrEvictOrNull(
public TEntry remove(@Nullable K key) {
long keyHash = D1.Entry.hash(key);
int index = bucketIndex(state.buckets, keyHash);
- synchronized (getTableWriteLock(state)) {
- TEntry prev = null;
- for (TEntry curEntry = bucketAt(state, index);
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
+ Entry prev = null;
+ for (Entry 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 cast(curEntry);
}
}
return null;
+ } finally {
+ lock.unlock();
}
}
@@ -305,31 +388,38 @@ public TEntry remove(@Nullable K key) {
* 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);
+ public boolean removeIf(@Strategy @Nonnull Predicate predicate) {
+ return ConcurrentHashtable.removeIf(state, castPredicate(predicate));
}
/**
* Removes all entries and invokes {@code drainedEntryConsumer} for each one.
*
- * The drain holds the table write lock while detaching buckets and invoking the consumer.
- * For each removed entry, the consumer is invoked synchronously after its bucket is detached.
- * Capacity is released only after all invocations return. The consumer should be quick and must
- * not throw; failures are not rolled back.
+ *
The lock is acquired and released bucket by bucket rather than held for the whole sweep,
+ * so a table-level operation (an insert, a reservation, an eviction) can interleave between
+ * buckets instead of waiting out the entire drain. This means the drain is no longer an atomic
+ * snapshot: an entry inserted into a not-yet-visited bucket while the drain is in progress is
+ * swept up too, and each bucket's capacity slot is released immediately before its consumer
+ * invocation runs, rather than once for the whole sweep at the end. The consumer should be
+ * quick and must not throw; entries not yet reached are left in the table with their capacity
+ * still counted, but entries already detached keep their capacity released regardless.
*
*
Use {@link #drain(Object, BiConsumer)} to avoid a capturing lambda.
*
* @param drainedEntryConsumer action invoked for each removed entry
*/
- public void drain(@Nonnull Consumer drainedEntryConsumer) {
- ConcurrentHashtable.drain(state, drainedEntryConsumer);
+ public void drain(@Strategy @Nonnull Consumer drainedEntryConsumer) {
+ ConcurrentHashtable.drain(state, castConsumer(drainedEntryConsumer));
}
/**
- * Context-passing {@link #drain(Consumer)}. The drain holds the table write lock while
- * detaching buckets and invoking {@code drainedEntryConsumer}. For each removed entry, the
- * consumer is invoked synchronously after its bucket is detached. Capacity is released only
- * after all invocations return.
+ * Context-passing {@link #drain(Consumer)}. The lock is acquired and released bucket by bucket
+ * rather than held for the whole sweep, so a table-level operation (an insert, a reservation,
+ * an eviction) can interleave between buckets instead of waiting out the entire drain. This
+ * means the drain is no longer an atomic snapshot: an entry inserted into a not-yet-visited
+ * bucket while the drain is in progress is swept up too, and each bucket's capacity slot is
+ * released immediately before its consumer invocation runs, rather than once for the whole
+ * sweep at the end.
*
*
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
@@ -340,8 +430,8 @@ public void drain(@Nonnull Consumer drainedEntryConsumer) {
* @param drainedEntryConsumer action invoked with the context and each removed entry
*/
public void drain(
- C context, @Nonnull BiConsumer drainedEntryConsumer) {
- ConcurrentHashtable.drain(state, context, drainedEntryConsumer);
+ C context, @Strategy @Nonnull BiConsumer drainedEntryConsumer) {
+ ConcurrentHashtable.drain(state, context, castConsumer(drainedEntryConsumer));
}
/** Removes all entries. Lock-free readers mid-walk complete against the entries they hold. */
@@ -349,16 +439,17 @@ public void clear() {
ConcurrentHashtable.clear(state);
}
- public void forEach(@Nonnull Consumer consumer) {
- ConcurrentHashtable.forEach(state, consumer);
+ public void forEach(@Strategy @Nonnull Consumer consumer) {
+ ConcurrentHashtable.forEach(state, castConsumer(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);
+ public void forEach(
+ C context, @Strategy @Nonnull BiConsumer consumer) {
+ ConcurrentHashtable.forEach(state, context, castConsumer(consumer));
}
}
@@ -376,14 +467,17 @@ public static final class D2> {
/**
* Abstract base for {@link D2} entries. Subclass to add value fields you wish to mutate in
- * place.
+ * place after retrieving the entry via {@link D2#get}.
+ *
+ * Deliberately parameterized on {@code K1}/{@code K2} alone, not self-bound on the concrete
+ * subclass -- see {@link D1.Entry} for why.
*
* @param first key type
* @param second key type
*/
- public abstract static class Entry extends ConcurrentHashtable.Entry {
- final K1 key1;
- final K2 key2;
+ public abstract static class Entry extends ConcurrentHashtable.Entry> {
+ @Nullable final K1 key1;
+ @Nullable final K2 key2;
protected Entry(@Nullable K1 key1, @Nullable K2 key2) {
super(hash(key1, key2));
@@ -409,27 +503,63 @@ public boolean matches(@Nullable K1 key1, @Nullable K2 key2) {
return Objects.equals(key1, this.key1) && Objects.equals(key2, this.key2);
}
+ /** {@link ConcurrentHashtable.Entry#matches(Entry)} in terms of the key-based overload. */
+ @Override
+ public final boolean matches(@Nonnull Entry other) {
+ return matches(other.key1, other.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 final State> state;
- private D2(State state) {
+ private D2(State> state) {
this.state = state;
}
/**
* Creates a fixed-size table holding at most {@code maxCapacity} entries. {@code entryClass} is
- * used only to infer the concrete entry type; entries are created by the functions passed to
- * the insertion methods. The table does not resize.
+ * used only to allocate the backing array with the right component type; entries themselves are
+ * created by the {@code creator}/{@code evictable} functions passed to the insertion methods.
+ * The table does not resize.
*/
@Nonnull
+ @SuppressWarnings("unchecked")
public static > D2 createBounded(
@Nonnull Class entryClass, int maxCapacity) {
- return new D2<>(State.createBounded(entryClass, maxCapacity));
+ // entryClass is erased away (see ConcurrentHashtable#createFixedBuckets), so treating it as
+ // Class> instead of the caller's concrete Class is safe.
+ Class> baseEntryClass = (Class>) (Class) entryClass;
+ return new D2<>(ConcurrentHashtable.createBounded(baseEntryClass, maxCapacity));
+ }
+
+ /**
+ * Sound because every entry ever inserted into {@link #state} was produced as a {@code TEntry}.
+ */
+ @SuppressWarnings("unchecked")
+ private TEntry cast(@Nullable Entry entry) {
+ return (TEntry) entry;
+ }
+
+ /** See {@link #cast(Entry)}; casts the functional-interface reference, not each element. */
+ @SuppressWarnings("unchecked")
+ private Predicate> castPredicate(Predicate predicate) {
+ return (Predicate>) predicate;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Consumer> castConsumer(Consumer consumer) {
+ return (Consumer>) consumer;
+ }
+
+ @SuppressWarnings("unchecked")
+ private BiConsumer> castConsumer(
+ BiConsumer consumer) {
+ return (BiConsumer>) consumer;
}
public int size() {
@@ -443,11 +573,11 @@ public boolean isFull() {
@Nullable
public TEntry get(@Nullable K1 key1, @Nullable K2 key2) {
long keyHash = D2.Entry.hash(key1, key2);
- for (TEntry curEntry = bucketFor(state, keyHash);
+ for (Entry curEntry = bucketFor(state, keyHash);
curEntry != null;
curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) {
- return curEntry;
+ return cast(curEntry);
}
}
return null;
@@ -457,15 +587,12 @@ public TEntry get(@Nullable K1 key1, @Nullable K2 key2) {
* 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) {
+ @Strategy @Nonnull BiFunction creator) {
return Maybe.of(tryGetOrCreateOrNull(key1, key2, creator));
}
@@ -475,24 +602,29 @@ public Maybe tryGetOrCreate(
* {@code (key1, key2)} was not already present. Re-checks under the lock to avoid duplicate
* entries under concurrent misses.
*/
+ @StrategyConsumer
@Nullable
public TEntry tryGetOrCreateOrNull(
@Nullable K1 key1,
@Nullable K2 key2,
- @Nonnull BiFunction creator) {
+ @Strategy @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()) {
+ for (Entry curEntry = bucketAt(state, index);
+ curEntry != null;
+ curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) {
- return curEntry;
+ return cast(curEntry);
}
}
- synchronized (getTableWriteLock(state)) {
- for (TEntry curEntry = bucketAt(state, index);
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
+ for (Entry curEntry = bucketAt(state, index);
curEntry != null;
curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) {
- return curEntry;
+ return cast(curEntry);
}
}
// isFull() is checked before creating the entry, not before reserving a slot for it:
@@ -506,6 +638,8 @@ public TEntry tryGetOrCreateOrNull(
insertHeadEntryAt(state, index, newEntry);
state.sizeManager.increment();
return newEntry;
+ } finally {
+ lock.unlock();
}
}
@@ -519,8 +653,8 @@ public TEntry tryGetOrCreateOrNull(
public Maybe tryGetOrCreateOrEvict(
@Nullable K1 key1,
@Nullable K2 key2,
- @Nonnull BiFunction creator,
- @Nonnull Predicate evictable) {
+ @Strategy @Nonnull BiFunction creator,
+ @Strategy @Nonnull Predicate evictable) {
return Maybe.of(tryGetOrCreateOrEvictOrNull(key1, key2, creator, evictable));
}
@@ -531,35 +665,42 @@ public Maybe tryGetOrCreateOrEvict(
* 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.
*/
+ @StrategyConsumer
@Nullable
public TEntry tryGetOrCreateOrEvictOrNull(
@Nullable K1 key1,
@Nullable K2 key2,
- @Nonnull BiFunction creator,
- @Nonnull Predicate evictable) {
+ @Strategy @Nonnull BiFunction creator,
+ @Strategy @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()) {
+ for (Entry curEntry = bucketAt(state, index);
+ curEntry != null;
+ curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) {
- return curEntry;
+ return cast(curEntry);
}
}
- synchronized (getTableWriteLock(state)) {
- for (TEntry curEntry = bucketAt(state, index);
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
+ for (Entry curEntry = bucketAt(state, index);
curEntry != null;
curEntry = curEntry.next()) {
if (curEntry.keyHash == keyHash && curEntry.matches(key1, key2)) {
- return curEntry;
+ return cast(curEntry);
}
}
if (state.sizeManager.isFull()
- && state.sizeManager.evictOne(state.buckets, evictable) == null) {
+ && state.sizeManager.evictOne(state.buckets, castPredicate(evictable)) == null) {
return null;
}
TEntry newEntry = creator.apply(key1, key2);
insertHeadEntryAt(state, index, newEntry);
state.sizeManager.increment();
return newEntry;
+ } finally {
+ lock.unlock();
}
}
@@ -572,18 +713,22 @@ public TEntry tryGetOrCreateOrEvictOrNull(
public TEntry remove(@Nullable K1 key1, @Nullable K2 key2) {
long keyHash = D2.Entry.hash(key1, key2);
int index = bucketIndex(state.buckets, keyHash);
- synchronized (getTableWriteLock(state)) {
- TEntry prev = null;
- for (TEntry curEntry = bucketAt(state, index);
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
+ Entry prev = null;
+ for (Entry 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 cast(curEntry);
}
}
return null;
+ } finally {
+ lock.unlock();
}
}
@@ -592,31 +737,38 @@ public TEntry remove(@Nullable K1 key1, @Nullable K2 key2) {
* 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);
+ public boolean removeIf(@Strategy @Nonnull Predicate predicate) {
+ return ConcurrentHashtable.removeIf(state, castPredicate(predicate));
}
/**
* Removes all entries and invokes {@code drainedEntryConsumer} for each one.
*
- * The drain holds the table write lock while detaching buckets and invoking the consumer.
- * For each removed entry, the consumer is invoked synchronously after its bucket is detached.
- * Capacity is released only after all invocations return. The consumer should be quick and must
- * not throw; failures are not rolled back.
+ *
The lock is acquired and released bucket by bucket rather than held for the whole sweep,
+ * so a table-level operation (an insert, a reservation, an eviction) can interleave between
+ * buckets instead of waiting out the entire drain. This means the drain is no longer an atomic
+ * snapshot: an entry inserted into a not-yet-visited bucket while the drain is in progress is
+ * swept up too, and each bucket's capacity slot is released immediately before its consumer
+ * invocation runs, rather than once for the whole sweep at the end. The consumer should be
+ * quick and must not throw; entries not yet reached are left in the table with their capacity
+ * still counted, but entries already detached keep their capacity released regardless.
*
*
Use {@link #drain(Object, BiConsumer)} to avoid a capturing lambda.
*
* @param drainedEntryConsumer action invoked for each removed entry
*/
- public void drain(@Nonnull Consumer drainedEntryConsumer) {
- ConcurrentHashtable.drain(state, drainedEntryConsumer);
+ public void drain(@Strategy @Nonnull Consumer drainedEntryConsumer) {
+ ConcurrentHashtable.drain(state, castConsumer(drainedEntryConsumer));
}
/**
- * Context-passing {@link #drain(Consumer)}. The drain holds the table write lock while
- * detaching buckets and invoking {@code drainedEntryConsumer}. For each removed entry, the
- * consumer is invoked synchronously after its bucket is detached. Capacity is released only
- * after all invocations return.
+ * Context-passing {@link #drain(Consumer)}. The lock is acquired and released bucket by bucket
+ * rather than held for the whole sweep, so a table-level operation (an insert, a reservation,
+ * an eviction) can interleave between buckets instead of waiting out the entire drain. This
+ * means the drain is no longer an atomic snapshot: an entry inserted into a not-yet-visited
+ * bucket while the drain is in progress is swept up too, and each bucket's capacity slot is
+ * released immediately before its consumer invocation runs, rather than once for the whole
+ * sweep at the end.
*
*
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
@@ -627,8 +779,8 @@ public void drain(@Nonnull Consumer drainedEntryConsumer) {
* @param drainedEntryConsumer action invoked with the context and each removed entry
*/
public void drain(
- C context, @Nonnull BiConsumer drainedEntryConsumer) {
- ConcurrentHashtable.drain(state, context, drainedEntryConsumer);
+ C context, @Strategy @Nonnull BiConsumer drainedEntryConsumer) {
+ ConcurrentHashtable.drain(state, context, castConsumer(drainedEntryConsumer));
}
/** Removes all entries. Lock-free readers mid-walk complete against the entries they hold. */
@@ -636,16 +788,17 @@ public void clear() {
ConcurrentHashtable.clear(state);
}
- public void forEach(@Nonnull Consumer consumer) {
- ConcurrentHashtable.forEach(state, consumer);
+ public void forEach(@Strategy @Nonnull Consumer consumer) {
+ ConcurrentHashtable.forEach(state, castConsumer(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);
+ public void forEach(
+ C context, @Strategy @Nonnull BiConsumer consumer) {
+ ConcurrentHashtable.forEach(state, context, castConsumer(consumer));
}
}
@@ -666,7 +819,7 @@ public static final class SizeManager {
* 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("getTableWriteLock(buckets)")
+ @GuardedBy("the corresponding State's getTableWriteLock(state)")
private int evictionCursor;
public SizeManager(int capacity) {
@@ -693,8 +846,9 @@ public boolean isFull() {
* cannot both acquire the last slot. Returns {@code false} with the count unchanged when the
* table is full.
*
- * Build the entry before reserving: there is no cancellation operation, so abandoning a
- * successful reservation permanently consumes capacity.
+ *
Prefer {@link #cancelReservation()} plus deferred entry construction (see {@link
+ * ConcurrentHashtable#reserve}) over abandoning a reservation outright: this method by itself
+ * still has no way to give back a slot once claimed.
*/
public boolean tryReserve() {
if (size.incrementAndGet() > capacity) {
@@ -704,6 +858,27 @@ public boolean tryReserve() {
return true;
}
+ /**
+ * Gives back a slot claimed by {@link #tryReserve()} that was never filled — e.g. a concurrent
+ * match was found under the write lock instead of inserting. Lock-free, symmetric with {@link
+ * #decrement()}.
+ *
+ *
Caller must call this at most once per successful {@link #tryReserve()}; double-cancelling
+ * corrupts the count the same way double-incrementing would. {@link Reservation#close()}
+ * handles this bookkeeping automatically and should be preferred over calling this directly.
+ *
+ *
Under heavy contention on the same logical duplicate, multiple threads can each reserve a
+ * slot for what turns out to be the same entry before any of them cancels, transiently
+ * inflating {@code size} above the table's true occupancy. This can cause an unrelated,
+ * genuinely distinct concurrent insert to see the table as full when it isn't, until the losing
+ * reservations cancel. The effect is self-correcting (bounded by in-flight reservations, not
+ * sustained) and considered an acceptable tradeoff for tables expecting bursts of identical
+ * inserts (e.g. deduplication).
+ */
+ public void cancelReservation() {
+ size.decrementAndGet();
+ }
+
/**
* Reserves one slot, evicting an entry matching {@code evictable} when the table is full.
* Returns {@code false} without changing the table when no entry can be evicted.
@@ -711,10 +886,10 @@ public boolean tryReserve() {
*
The reservation survives concurrent drain and clear operations. The caller must fill it;
* an abandoned reservation permanently consumes capacity.
*/
- @GuardedBy("getTableWriteLock(buckets)")
- public boolean tryReserveOrEvict(
+ @GuardedBy("the corresponding State's getTableWriteLock(state)")
+ public > boolean tryReserveOrEvict(
@Nonnull AtomicReferenceArray buckets,
- @Nonnull Predicate evictable) {
+ @Strategy @Nonnull Predicate evictable) {
if (tryReserve()) {
return true;
}
@@ -740,12 +915,12 @@ public void decrement() {
* Releases {@code removed} slots after a sweep and resets the {@code evictionCursor}.
* Outstanding reservations remain counted.
*/
- @GuardedBy("getTableWriteLock(buckets)")
+ @GuardedBy("the corresponding State's getTableWriteLock(state)")
@SuppressFBWarnings(
value = "AT_STALE_THREAD_WRITE_OF_PRIMITIVE",
justification =
- "evictionCursor is read and written only under synchronized (getTableWriteLock(buckets)); SpotBugs"
- + " cannot model that dynamic guard")
+ "evictionCursor is read and written only while the corresponding State's write lock is"
+ + " held; SpotBugs cannot model that dynamic guard")
public void release(int removed) {
if (removed != 0) {
size.addAndGet(-removed);
@@ -761,11 +936,11 @@ public void release(int removed) {
* This operation may inspect every live entry while holding the table write lock, so the
* predicate should be quick.
*/
- @GuardedBy("getTableWriteLock(buckets)")
+ @GuardedBy("the corresponding State's getTableWriteLock(state)")
@Nullable
- public TEntry evictOne(
+ public > TEntry evictOne(
@Nonnull AtomicReferenceArray buckets,
- @Nonnull Predicate evictable) {
+ @Strategy @Nonnull Predicate evictable) {
TEntry evicted = evictOneInRange(buckets, evictable, evictionCursor, buckets.length());
if (evicted == null && evictionCursor != 0) {
evicted = evictOneInRange(buckets, evictable, 0, evictionCursor);
@@ -781,23 +956,24 @@ public TEntry evictOne(
return null;
}
- @GuardedBy("getTableWriteLock(buckets)")
+ @GuardedBy("the corresponding State's getTableWriteLock(state)")
@SuppressFBWarnings(
value = "AT_STALE_THREAD_WRITE_OF_PRIMITIVE",
justification =
- "evictionCursor is read and written only under synchronized (getTableWriteLock(buckets)); SpotBugs"
- + " cannot model that dynamic guard")
+ "evictionCursor is read and written only while the corresponding State's write lock is"
+ + " held; SpotBugs cannot model that dynamic guard")
+ @StrategyConsumer
@Nullable
- private TEntry evictOneInRange(
+ private > TEntry evictOneInRange(
@Nonnull AtomicReferenceArray buckets,
- @Nonnull Predicate evictable,
+ @Strategy @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);
+ unlinkUnchecked(buckets, i, prev, e);
evictionCursor = i;
return e;
}
@@ -812,21 +988,22 @@ private TEntry evictOneInRange(
* each, and returns how many were removed. Resets the scan position, since a full pass leaves
* nothing later to resume from.
*/
- @GuardedBy("getTableWriteLock(buckets)")
+ @GuardedBy("the corresponding State's getTableWriteLock(state)")
@SuppressFBWarnings(
value = "AT_STALE_THREAD_WRITE_OF_PRIMITIVE",
justification =
- "evictionCursor is read and written only under synchronized (getTableWriteLock(buckets)); SpotBugs"
- + " cannot model that dynamic guard")
- public int evictAll(
+ "evictionCursor is read and written only while the corresponding State's write lock is"
+ + " held; SpotBugs cannot model that dynamic guard")
+ @StrategyConsumer
+ public > int evictAll(
@Nonnull AtomicReferenceArray buckets,
- @Nonnull Predicate evictable) {
+ @Strategy @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);
+ unlinkUnchecked(buckets, i, prev, e);
size.decrementAndGet();
count++;
} else {
@@ -842,25 +1019,38 @@ public int evictAll(
/**
* Bucket array and occupancy manager for a caller-defined capped table. Keep them paired and
* prefer the {@code State}-accepting helpers so structural changes update the count consistently.
+ *
+ * {@code sizeManager} is intentionally package-private: callers outside this class must go
+ * through the {@code State}-accepting static helpers ({@link #estimateSize}, {@link #isFull},
+ * {@link #tryReserve}, {@link #tryReserveOrEvict}, {@link #evictOne}, {@link #evictAll}) rather
+ * than reach into the manager directly.
*/
- public static final class State {
+ public static final class State> {
public final AtomicReferenceArray buckets;
- public final SizeManager sizeManager;
+ final SizeManager sizeManager;
+
+ /**
+ * The table's write lock. Unlike the bare-array helpers' opaque {@code synchronized} monitor,
+ * this is a real {@link ReentrantLock} so a {@link Reservation} can acquire it in {@link
+ * #tryReserve} and release it later in {@link Reservation#finish}/{@link Reservation#close},
+ * across separate calls -- something a lexically-scoped {@code synchronized} block cannot do.
+ */
+ final ReentrantLock writeLock = new ReentrantLock();
private State(AtomicReferenceArray buckets, int maxCapacity) {
this.buckets = buckets;
this.sizeManager = new SizeManager(maxCapacity);
}
+ }
- /**
- * Creates a bucket array for {@code maxCapacity} entries and pairs it with a manager enforcing
- * that cap. {@code entryClass} is used only to infer {@code TEntry}.
- */
- @Nonnull
- public static State createBounded(
- @Nonnull Class entryClass, int maxCapacity) {
- return new State<>(createFixedBuckets(entryClass, maxCapacity), maxCapacity);
- }
+ /**
+ * Creates a bucket array for {@code maxCapacity} entries and pairs it with a manager enforcing
+ * that cap. {@code entryClass} is used only to infer {@code TEntry}.
+ */
+ @Nonnull
+ public static > State createBounded(
+ @Nonnull Class entryClass, int maxCapacity) {
+ return new State<>(createFixedBuckets(entryClass, maxCapacity), maxCapacity);
}
/** Live entries in {@code state}; see {@link SizeManager#estimateSize()}. Lock-free. */
@@ -875,6 +1065,308 @@ public static boolean isFull(@Nonnull State state) {
return state.sizeManager.isFull();
}
+ /**
+ * Reserves one slot in {@code state} without evicting; see {@link SizeManager#tryReserve()}.
+ * Lock-free — does not acquire the table write lock. Returns {@code false} with the table
+ * unchanged when it is full.
+ *
+ * Complete it with {@link #insertReserved}, or prefer {@link #tryReserve} for a higher-level,
+ * auto-cancelling handle that also defers entry construction until the reservation succeeds.
+ */
+ public static > boolean tryReserveSlot(
+ @Nonnull State state) {
+ return state.sizeManager.tryReserve();
+ }
+
+ /**
+ * Claims one slot in {@code state} for {@code keyHash} and returns a handle for completing the
+ * find-or-insert protocol, or an empty handle if the table is full. Never returns {@code null},
+ * so this always composes with try-with-resources:
+ *
+ * {@code
+ * try (Reservation r = ConcurrentHashtable.tryReserve(state, keyHash)) {
+ * return r.tryGetOrInsertOrNull(component1, component2, component3, TEntry::new);
+ * }
+ * }
+ *
+ * Run a lock-free scan first (see {@link #bucketFor}/{@link #bucketAt}) and only call this
+ * once that scan has missed — a successful reservation isn't required for correctness (the
+ * reservation itself, and the locked comparison inside {@link Reservation#tryGetOrInsertOrNull},
+ * are the source of truth), it just avoids paying for a lock and a factory call when a hit was
+ * already visible lock-free.
+ *
+ *
Checks {@link SizeManager#isFull()} lock-free first. If the table looks full, {@code
+ * keyHash}'s bucket is checked too: if it's empty, nothing could possibly match this key, so this
+ * returns a definitively empty reservation without ever taking the write lock. Otherwise it
+ * acquires the write lock, re-checks (the lock-free peek may be stale), and either reserves the
+ * slot -- returning a real reservation that holds the lock open across this call, to be
+ * released later by {@link Reservation#finish}/{@link Reservation#close} -- or, if the table is
+ * genuinely still full, keeps the lock open anyway (rather than releasing it immediately) so
+ * {@link Reservation#finish} can still scan {@code keyHash}'s bucket the next time the caller
+ * supplies a candidate entry. That's what closes the race where a concurrent insert for the same
+ * logical duplicate lands in the exact window between this caller's own lock-free scan and its
+ * reservation attempt: without the lock held open here, that concurrent duplicate would go
+ * uncounted. Holding the lock for the reservation's whole lifetime (rather than just across this
+ * call, the way {@link SizeManager#tryReserve()} does) is also what lets {@link
+ * Reservation#tryGetOrInsertOrNull} skip a second, separately-locked comparison pass: the
+ * reserve-then-insert-or-discard sequence is one uninterrupted critical section, so no concurrent
+ * reservation for the same logical duplicate can slip in between.
+ *
+ *
Always returns a non-null handle — even when the table is full — so the caller must check
+ * {@link Reservation#isReserved()} (or simply call {@link Reservation#tryGetOrInsertOrNull},
+ * which returns {@code null} on an absent reservation) rather than assume every reservation is
+ * real. A reservation can also report {@link Reservation#isReserved()} {@code true} while still
+ * having no slot to insert into (the lock-held-but-full case above); either way, {@code
+ * tryGetOrInsertOrNull}/{@code tryGetOrInsert} return the concurrent match found under the lock,
+ * or {@code null} -- never a newly linked entry when no slot was claimed.
+ *
+ * @param keyHash hash of the key the caller is about to look up or insert
+ */
+ @Nonnull
+ public static > Reservation tryReserve(
+ @Nonnull State state, long keyHash) {
+ if (state.sizeManager.isFull()
+ && bucketAt(state, bucketIndex(state.buckets, keyHash)) == null) {
+ return new Reservation<>(null, false);
+ }
+ ReentrantLock lock = state.writeLock;
+ lock.lock();
+ if (state.sizeManager.tryReserve()) {
+ return new Reservation<>(state, true);
+ }
+ return new Reservation<>(state, false);
+ }
+
+ /**
+ * Handle returned by {@link #tryReserve}, gating {@link #tryGetOrInsertOrNull} behind a claimed
+ * slot and auto-cancelling it on {@link #close} if it's never consumed. A single {@code
+ * Reservation} must be used for at most one {@code tryGetOrInsertOrNull} call.
+ *
+ * Overloaded up to 4 key components ({@link #tryGetOrInsertOrNull(Object, Function)} through
+ * {@link #tryGetOrInsertOrNull(Object, Object, Object, Object, Function4)}) so a non-capturing
+ * method reference can build the entry directly from its natural constructor arguments, without
+ * an intermediate holder object or a capturing lambda.
+ *
+ *
A real (non-empty) reservation holds the table's write lock from the moment {@link
+ * #tryReserve} returns until {@link #finish} or {@link #close} releases it -- unlike a
+ * lexically-scoped {@code synchronized} block, whose acquisition and release can't span two
+ * separate calls. Keep a reservation's lifetime short: nothing else can write to the table while
+ * one is open, and (having deliberately dropped the CAS-based fast path that would let two
+ * threads race for the same slot) nothing else can even reserve.
+ *
+ * @param the table's entry type, itself self-bound (see {@link
+ * ConcurrentHashtable.Entry})
+ */
+ public static final class Reservation> implements AutoCloseable {
+ @Nullable private final State state;
+ private final boolean slotClaimed;
+ private boolean consumed;
+
+ private Reservation(@Nullable State state, boolean slotClaimed) {
+ this.state = state;
+ this.slotClaimed = slotClaimed;
+ }
+
+ /**
+ * {@code true} unless this reservation is definitively empty -- i.e. {@link #tryReserve} could
+ * rule out a match for the target key without even taking the write lock. A {@code true}
+ * reservation may still have no slot to insert into: {@link #tryReserve} keeps the lock open on
+ * a full table when the target bucket is populated, purely so {@link #finish} can still scan it
+ * for a concurrent match. Callers should call {@code tryGetOrInsertOrNull}/{@code
+ * tryGetOrInsert} and check its result rather than assume {@code true} means a slot was
+ * claimed.
+ */
+ public boolean isReserved() {
+ return state != null;
+ }
+
+ /**
+ * Escape hatch for a caller that already built {@code newEntry} itself -- e.g. more than 4 key
+ * components, or components the caller wants to keep as primitives rather than boxing them into
+ * a {@code Function}'s type argument:
+ *
+ * {@code
+ * try (Reservation r = ConcurrentHashtable.tryReserve(state, keyHash)) {
+ * if (!r.isReserved()) {
+ * return null;
+ * }
+ * return r.tryGetOrInsertOrNull(new TEntry(longComponent1, longComponent2));
+ * }
+ * }
+ *
+ * See {@link #tryGetOrInsertOrNull(Object, Object, BiFunction)} for the general contract.
+ */
+ @Nullable
+ public TEntry tryGetOrInsertOrNull(@Nonnull TEntry newEntry) {
+ return state == null ? null : finish(newEntry);
+ }
+
+ /**
+ * One key component; see {@link #tryGetOrInsertOrNull(Object, Object, BiFunction)} for the
+ * general contract.
+ */
+ @StrategyConsumer
+ @Nullable
+ public TEntry tryGetOrInsertOrNull(
+ A a, @Strategy @Nonnull Function factory) {
+ return state == null ? null : finish(factory.apply(a));
+ }
+
+ /**
+ * Two key components. Builds {@code factory.apply(...)} (skipped entirely if this reservation
+ * is definitively empty — see {@link #isReserved()}) and either links the result as a new entry
+ * or discards it in favor of an existing match found under the write lock. Returns {@code null}
+ * when there is neither a slot to claim nor a match to return -- either because this
+ * reservation is definitively empty, or (see {@link #tryReserve}) because the table was still
+ * full even under the lock and no concurrent match was found either; otherwise always returns a
+ * real entry (the newly built one, or the concurrent match).
+ *
+ * Building the entry here, after the reservation already succeeded, keeps the write lock's
+ * critical section limited to the comparison/link/discard decision rather than whatever
+ * construction cost {@code factory} pays. See {@link ConcurrentHashtable.Entry#matches} — the
+ * under-lock comparison is entry-to-entry, so it needs {@code newEntry} already built.
+ */
+ @StrategyConsumer
+ @Nullable
+ public TEntry tryGetOrInsertOrNull(
+ A a, B b, @Strategy @Nonnull BiFunction factory) {
+ return state == null ? null : finish(factory.apply(a, b));
+ }
+
+ /**
+ * Three key components; see {@link #tryGetOrInsertOrNull(Object, Object, BiFunction)} for the
+ * general contract.
+ */
+ @StrategyConsumer
+ @Nullable
+ public TEntry tryGetOrInsertOrNull(
+ A a,
+ B b,
+ C c,
+ @Strategy @Nonnull Function3 factory) {
+ return state == null ? null : finish(factory.apply(a, b, c));
+ }
+
+ /** Four key components; see {@link #tryGetOrInsertOrNull(Object, Object, BiFunction)}. */
+ @StrategyConsumer
+ @Nullable
+ public TEntry tryGetOrInsertOrNull(
+ A a,
+ B b,
+ C c,
+ D d,
+ @Strategy @Nonnull
+ Function4 factory) {
+ return state == null ? null : finish(factory.apply(a, b, c, d));
+ }
+
+ /** {@link Maybe}-wrapping counterpart of {@link #tryGetOrInsertOrNull(Entry)}. */
+ @Nonnull
+ public Maybe tryGetOrInsert(@Nonnull TEntry newEntry) {
+ return Maybe.of(tryGetOrInsertOrNull(newEntry));
+ }
+
+ /**
+ * {@link Maybe}-wrapping counterpart of {@link #tryGetOrInsertOrNull(Object, Function)}, for
+ * callers who'd rather make the "this can fail unlike unbounded collections" outcome visible in
+ * the return type than rely on a {@code null} check — mirrors {@link D1#tryGetOrCreate} /
+ * {@link D2#tryGetOrCreate} wrapping their own {@code ...OrNull} methods.
+ */
+ @Nonnull
+ public Maybe tryGetOrInsert(
+ A a, @Strategy @Nonnull Function factory) {
+ return Maybe.of(tryGetOrInsertOrNull(a, factory));
+ }
+
+ /** Two key components; see {@link #tryGetOrInsert(Object, Function)}. */
+ @Nonnull
+ public Maybe tryGetOrInsert(
+ A a, B b, @Strategy @Nonnull BiFunction factory) {
+ return Maybe.of(tryGetOrInsertOrNull(a, b, factory));
+ }
+
+ /** Three key components; see {@link #tryGetOrInsert(Object, Function)}. */
+ @Nonnull
+ public Maybe tryGetOrInsert(
+ A a,
+ B b,
+ C c,
+ @Strategy @Nonnull Function3 factory) {
+ return Maybe.of(tryGetOrInsertOrNull(a, b, c, factory));
+ }
+
+ /** Four key components; see {@link #tryGetOrInsert(Object, Function)}. */
+ @Nonnull
+ public Maybe tryGetOrInsert(
+ A a,
+ B b,
+ C c,
+ D d,
+ @Strategy @Nonnull
+ Function4 factory) {
+ return Maybe.of(tryGetOrInsertOrNull(a, b, c, d, factory));
+ }
+
+ /**
+ * Runs the locked comparison/link decision. The write lock is already held -- acquired by
+ * {@link #tryReserve} and not yet released -- so this needs no {@code synchronized}/{@code
+ * lock()} of its own.
+ *
+ * Always scans for a match first, whether or not a slot was actually claimed: a {@link
+ * #tryReserve} reservation with no slot still holds the lock specifically so this scan can run.
+ * Only links {@code newEntry} when a slot was claimed; otherwise a miss here means there really
+ * is nothing to return, so this returns {@code null}.
+ */
+ @Nullable
+ private TEntry finish(@Nonnull TEntry newEntry) {
+ int index = bucketIndex(state.buckets, newEntry.keyHash);
+ for (TEntry curEntry = bucketAt(state, index); curEntry != null; curEntry = curEntry.next()) {
+ if (curEntry.keyHash == newEntry.keyHash && curEntry.matches(newEntry)) {
+ return curEntry;
+ }
+ }
+ if (!slotClaimed) {
+ return null;
+ }
+ insertHeadEntryAt(state, index, newEntry);
+ consumed = true;
+ return newEntry;
+ }
+
+ /**
+ * Releases the write lock {@link #tryReserve} acquired, giving back a claimed slot first if it
+ * was never consumed. A no-op on a definitively empty reservation, which never acquired the
+ * lock.
+ */
+ @Override
+ public void close() {
+ if (state == null) {
+ return;
+ }
+ try {
+ if (slotClaimed && !consumed) {
+ state.sizeManager.decrement();
+ }
+ } finally {
+ state.writeLock.unlock();
+ }
+ }
+ }
+
+ /** Three-argument analogue of {@link java.util.function.BiFunction}. */
+ @Strategy
+ @FunctionalInterface
+ public interface Function3 {
+ R apply(A a, B b, C c);
+ }
+
+ /** Four-argument analogue of {@link java.util.function.BiFunction}. */
+ @Strategy
+ @FunctionalInterface
+ public interface Function4 {
+ R apply(A a, B b, C c, D d);
+ }
+
/**
* Reserves one slot in {@code state}, evicting an entry matching {@code evictable} when
* necessary. Returns {@code false} if the table is full and nothing can be evicted. This method
@@ -883,10 +1375,14 @@ public static boolean isFull(@Nonnull State state) {
* The reservation survives drain and clear operations. Complete it with {@link
* #insertReserved}; abandoning it permanently consumes capacity.
*/
- public static boolean tryReserveOrEvict(
- @Nonnull State state, @Nonnull Predicate evictable) {
- synchronized (getTableWriteLock(state)) {
+ public static > boolean tryReserveOrEvict(
+ @Nonnull State state, @Strategy @Nonnull Predicate evictable) {
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
return state.sizeManager.tryReserveOrEvict(state.buckets, evictable);
+ } finally {
+ lock.unlock();
}
}
@@ -896,10 +1392,14 @@ public static boolean tryReserveOrEvict(
* Self-locking.
*/
@Nullable
- public static TEntry evictOne(
- @Nonnull State state, @Nonnull Predicate evictable) {
- synchronized (getTableWriteLock(state)) {
+ public static > TEntry evictOne(
+ @Nonnull State state, @Strategy @Nonnull Predicate evictable) {
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
return state.sizeManager.evictOne(state.buckets, evictable);
+ } finally {
+ lock.unlock();
}
}
@@ -907,10 +1407,14 @@ public static TEntry evictOne(
* 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 (getTableWriteLock(state)) {
+ public static > int evictAll(
+ @Nonnull State state, @Strategy @Nonnull Predicate evictable) {
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
return state.sizeManager.evictAll(state.buckets, evictable);
+ } finally {
+ lock.unlock();
}
}
@@ -931,7 +1435,7 @@ public static int evictAll(
* reflective allocation or runtime type checks; it only lets the compiler infer {@code TEntry}.
*/
@Nonnull
- public static AtomicReferenceArray createFixedBuckets(
+ public static > AtomicReferenceArray createFixedBuckets(
@Nonnull Class entryClass, int capacity) {
return new AtomicReferenceArray<>(sizeFor(capacity));
}
@@ -971,10 +1475,15 @@ public static Object getWriteLock(@Nonnull AtomicReferenceArray buckets, long
return getWriteLockAt(buckets, bucketIndex(buckets, keyHash));
}
- /** {@link #getWriteLock(AtomicReferenceArray, long)} over a {@link State}. */
+ /**
+ * Returns the {@link State}'s {@link ReentrantLock}, held for the same keyed scan-and-mutate
+ * operations as {@link #getWriteLock(AtomicReferenceArray, long)}. Unlike the bare-array
+ * overload's opaque {@code synchronized} monitor, this lock can be acquired in one call and
+ * released in a later one -- see {@link Reservation}.
+ */
@Nonnull
- public static Object getWriteLock(@Nonnull State state, long keyHash) {
- return getWriteLock(state.buckets, keyHash);
+ public static ReentrantLock getWriteLock(@Nonnull State state, long keyHash) {
+ return state.writeLock;
}
/**
@@ -988,10 +1497,10 @@ public static Object getWriteLockAt(@Nonnull AtomicReferenceArray buckets, in
return buckets;
}
- /** {@link #getWriteLockAt(AtomicReferenceArray, int)} over a {@link State}. */
+ /** {@link #getWriteLock(State, long)}, for a bucket index that has already been computed. */
@Nonnull
- public static Object getWriteLockAt(@Nonnull State state, int bucketIndex) {
- return getWriteLockAt(state.buckets, bucketIndex);
+ public static ReentrantLock getWriteLockAt(@Nonnull State state, int bucketIndex) {
+ return state.writeLock;
}
/**
@@ -1003,10 +1512,14 @@ public static Object getTableWriteLock(@Nonnull AtomicReferenceArray buckets)
return buckets;
}
- /** {@link #getTableWriteLock(AtomicReferenceArray)} over a {@link State}. */
+ /**
+ * Returns the {@link State}'s {@link ReentrantLock}, guarding operations that span all buckets.
+ * Unlike {@link #getTableWriteLock(AtomicReferenceArray)}'s opaque {@code synchronized} monitor,
+ * this lock can be acquired in one call and released in a later one -- see {@link Reservation}.
+ */
@Nonnull
- public static Object getTableWriteLock(@Nonnull State state) {
- return getTableWriteLock(state.buckets);
+ public static ReentrantLock getTableWriteLock(@Nonnull State state) {
+ return state.writeLock;
}
public static int bucketIndex(@Nonnull AtomicReferenceArray buckets, long keyHash) {
@@ -1026,14 +1539,14 @@ public static int bucketIndex(@Nonnull AtomicReferenceArray buckets, long key
* bucket.
*/
@Nullable
- public static TEntry bucketFor(
+ 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(
+ public static > TEntry bucketFor(
@Nonnull State state, long keyHash) {
return bucketFor(state.buckets, keyHash);
}
@@ -1045,17 +1558,86 @@ public static TEntry bucketFor(
* overload of it.
*/
@Nullable
- public static TEntry bucketAt(
+ 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) {
+ public static > TEntry bucketAt(
+ @Nonnull State state, int index) {
return bucketAt(state.buckets, index);
}
+ /**
+ * Returns a lock-free iterator over the candidates for {@code keyHash}: entries in the bucket
+ * chain that {@code keyHash} maps to (starting from {@link #bucketFor(AtomicReferenceArray,
+ * long)}) whose own {@link Entry#keyHash} equals it, skipping any other entry sharing the same
+ * bucket via hash collision on {@link #bucketIndex}. Callers only need a {@code matches} check
+ * against the entries this yields, not a {@code keyHash} check of their own.
+ *
+ * Each step follows {@link Entry#next()}, so the iterator reflects entries linked at the time
+ * each step runs rather than a point-in-time snapshot -- entries inserted ahead of the iterator's
+ * current position after iteration starts may or may not be observed, and a concurrently removed
+ * entry remains reachable because {@code unlink()} deliberately retains its {@code next} link for
+ * in-flight readers.
+ */
+ @Nonnull
+ public static > Iterator hashIterator(
+ @Nonnull AtomicReferenceArray buckets, long keyHash) {
+ return new Iterator() {
+ private TEntry next = advance(bucketFor(buckets, keyHash));
+
+ private TEntry advance(TEntry candidate) {
+ while (candidate != null && candidate.keyHash != keyHash) {
+ candidate = candidate.next();
+ }
+ return candidate;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return next != null;
+ }
+
+ @Override
+ public TEntry next() {
+ TEntry current = next;
+ if (current == null) {
+ throw new NoSuchElementException();
+ }
+ next = advance(current.next());
+ return current;
+ }
+ };
+ }
+
+ /** {@link #hashIterator(AtomicReferenceArray, long)} over a {@link State}. */
+ @Nonnull
+ public static > Iterator hashIterator(
+ @Nonnull State state, long keyHash) {
+ return hashIterator(state.buckets, keyHash);
+ }
+
+ /**
+ * {@link Iterable} wrapper around {@link #hashIterator(AtomicReferenceArray, long)}, for callers
+ * that want a plain for-each loop over the candidates for {@code keyHash} rather than driving the
+ * {@link Iterator} by hand.
+ */
+ @Nonnull
+ public static > Iterable hashIterable(
+ @Nonnull AtomicReferenceArray buckets, long keyHash) {
+ return () -> hashIterator(buckets, keyHash);
+ }
+
+ /** {@link #hashIterable(AtomicReferenceArray, long)} over a {@link State}. */
+ @Nonnull
+ public static > Iterable hashIterable(
+ @Nonnull State state, long keyHash) {
+ return hashIterable(state.buckets, keyHash);
+ }
+
/**
* Publishes {@code entry} as the head of bucket {@code index}. The helper writes the entry's
* {@code next} link before the volatile {@link AtomicReferenceArray#set}; a volatile bucket read
@@ -1066,10 +1648,25 @@ public static TEntry bucketAt(@Nonnull State stat
* retains its {@code next} link for readers already traversing that chain.
*/
@GuardedBy("getWriteLockAt(buckets, index)")
- public static void insertHeadEntryAt(
+ public static > void insertHeadEntryAt(
@Nonnull AtomicReferenceArray buckets, int index, @Nonnull TEntry entry) {
assert Thread.holdsLock(getWriteLockAt(buckets, index))
: "insertHeadEntryAt called without holding getWriteLockAt(buckets, index)";
+ insertHeadEntryAtUnchecked(buckets, index, entry);
+ }
+
+ /** {@link #insertHeadEntryAt(AtomicReferenceArray, int, Entry)} over a {@link State}. */
+ @GuardedBy("getWriteLockAt(state, index)")
+ public static > void insertHeadEntryAt(
+ @Nonnull State state, int index, @Nonnull TEntry entry) {
+ assert getWriteLockAt(state, index).isHeldByCurrentThread()
+ : "insertHeadEntryAt called without holding getWriteLockAt(state, index)";
+ insertHeadEntryAtUnchecked(state.buckets, index, entry);
+ }
+
+ /** Splicing logic shared by both {@link #insertHeadEntryAt} overloads, without their assert. */
+ private static > void insertHeadEntryAtUnchecked(
+ @Nonnull AtomicReferenceArray buckets, int index, @Nonnull TEntry entry) {
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"
@@ -1078,20 +1675,13 @@ public static void insertHeadEntryAt(
buckets.set(index, entry);
}
- /** {@link #insertHeadEntryAt(AtomicReferenceArray, int, Entry)} over a {@link State}. */
- @GuardedBy("getWriteLockAt(state, index)")
- 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, keyHash)")
- public static void insertHeadEntryFor(
+ public static > void insertHeadEntryFor(
@Nonnull AtomicReferenceArray buckets, long keyHash, @Nonnull TEntry entry) {
insertHeadEntryAt(buckets, bucketIndex(buckets, keyHash), entry);
}
@@ -1105,9 +1695,11 @@ public static void insertHeadEntryFor(
* reservations.
*/
@GuardedBy("getTableWriteLock(state)")
- public static void insertReserved(
+ public static > void insertReserved(
@Nonnull State state, long keyHash, @Nonnull TEntry entry) {
- insertHeadEntryFor(state.buckets, keyHash, entry);
+ assert getTableWriteLock(state).isHeldByCurrentThread()
+ : "insertReserved called without holding getTableWriteLock(state)";
+ insertHeadEntryAtUnchecked(state.buckets, bucketIndex(state.buckets, keyHash), entry);
}
/**
@@ -1120,13 +1712,31 @@ public static void insertReserved(
* Does not touch size accounting.
*/
@GuardedBy("getWriteLockAt(buckets, index)")
- public static void unlink(
+ public static > void unlink(
@Nonnull AtomicReferenceArray buckets,
int index,
@Nullable TEntry prev,
@Nonnull TEntry entry) {
assert Thread.holdsLock(getWriteLockAt(buckets, index))
: "unlink called without holding getWriteLockAt(buckets, index)";
+ unlinkUnchecked(buckets, index, prev, entry);
+ }
+
+ /** {@link #unlink(AtomicReferenceArray, int, Entry, Entry)} over a {@link State}. */
+ @GuardedBy("getWriteLockAt(state, index)")
+ public static > void unlink(
+ @Nonnull State state, int index, @Nullable TEntry prev, @Nonnull TEntry entry) {
+ assert getWriteLockAt(state, index).isHeldByCurrentThread()
+ : "unlink called without holding getWriteLockAt(state, index)";
+ unlinkUnchecked(state.buckets, index, prev, entry);
+ }
+
+ /** Splicing logic shared by both {@link #unlink} overloads, without their assert. */
+ private static > void unlinkUnchecked(
+ @Nonnull AtomicReferenceArray buckets,
+ int index,
+ @Nullable TEntry prev,
+ @Nonnull TEntry entry) {
TEntry next = entry.next();
if (prev == null) {
buckets.set(index, next);
@@ -1135,23 +1745,17 @@ public static void unlink(
}
}
- /** {@link #unlink(AtomicReferenceArray, int, Entry, Entry)} over a {@link State}. */
- @GuardedBy("getWriteLockAt(state, index)")
- 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(
+ @StrategyConsumer
+ public static > boolean removeIf(
@Nonnull AtomicReferenceArray buckets,
@Nonnull AtomicInteger size,
- @Nonnull Predicate predicate) {
+ @Strategy @Nonnull Predicate predicate) {
synchronized (getTableWriteLock(buckets)) {
boolean removed = false;
for (int i = 0; i < buckets.length(); i++) {
@@ -1176,16 +1780,21 @@ public static boolean removeIf(
* 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) {
+ @StrategyConsumer
+ public static > boolean removeIf(
+ @Nonnull State state, @Strategy @Nonnull Predicate predicate) {
+ // TODO: no caller contends on this lock today; if one does, consider the same per-bucket
+ // locking drain() uses, trading the "predicate sees a stable table" guarantee for throughput.
AtomicReferenceArray buckets = state.buckets;
- synchronized (getTableWriteLock(state)) {
+ ReentrantLock lock = getTableWriteLock(state);
+ lock.lock();
+ try {
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);
+ unlink(state, i, prev, e);
state.sizeManager.decrement();
removed = true;
} else {
@@ -1194,6 +1803,8 @@ public static boolean removeIf(
}
}
return removed;
+ } finally {
+ lock.unlock();
}
}
@@ -1211,9 +1822,9 @@ public static boolean removeIf(
* @param buckets bucket array to drain
* @param drainedEntryConsumer action invoked for each entry after its bucket is detached
*/
- public static void drain(
+ public static > void drain(
@Nonnull AtomicReferenceArray buckets,
- @Nonnull Consumer drainedEntryConsumer) {
+ @Strategy @Nonnull Consumer drainedEntryConsumer) {
drainCounting(buckets, drainedEntryConsumer);
}
@@ -1221,9 +1832,10 @@ public static void drain(
* {@link #drain(AtomicReferenceArray, Consumer)} returning the number of entries passed to {@code
* drainedEntryConsumer} for size accounting.
*/
- private static int drainCounting(
+ @StrategyConsumer
+ private static > int drainCounting(
@Nonnull AtomicReferenceArray buckets,
- @Nonnull Consumer drainedEntryConsumer) {
+ @Strategy @Nonnull Consumer drainedEntryConsumer) {
int removed = 0;
synchronized (getTableWriteLock(buckets)) {
for (int i = 0; i < buckets.length(); i++) {
@@ -1256,18 +1868,19 @@ private static int drainCounting(
* @param context context passed to each invocation of {@code drainedEntryConsumer}
* @param drainedEntryConsumer action invoked with the context and each removed entry
*/
- public static