diff --git a/field-inject/src/jmh/java/datadog/instrument/fieldinject/ObjectStoreBenchmark.java b/field-inject/src/jmh/java/datadog/instrument/fieldinject/ObjectStoreBenchmark.java index a0424bc..2b5a25e 100644 --- a/field-inject/src/jmh/java/datadog/instrument/fieldinject/ObjectStoreBenchmark.java +++ b/field-inject/src/jmh/java/datadog/instrument/fieldinject/ObjectStoreBenchmark.java @@ -9,6 +9,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.concurrent.ThreadLocalRandom; import java.util.function.Function; import java.util.stream.IntStream; import org.openjdk.jmh.annotations.Benchmark; @@ -105,7 +106,13 @@ public void multiThreaded(Blackhole blackhole, ThreadData threadData) { generateKeys(); } - private static final Function allocator = key -> new byte[1024]; + // simulates a small, but non-trivial, value-construction cost + private static final Function allocator = + key -> { + byte[] value = new byte[1024]; + ThreadLocalRandom.current().nextBytes(value); + return value; + }; @Setup public void setup() { diff --git a/field-inject/src/main/java/datadog/instrument/fieldinject/GlobalObjectStore.java b/field-inject/src/main/java/datadog/instrument/fieldinject/GlobalObjectStore.java index 8dc7c8e..8162d78 100644 --- a/field-inject/src/main/java/datadog/instrument/fieldinject/GlobalObjectStore.java +++ b/field-inject/src/main/java/datadog/instrument/fieldinject/GlobalObjectStore.java @@ -9,41 +9,71 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.lang.ref.ReferenceQueue; import java.lang.ref.WeakReference; -import java.util.HashSet; +import java.util.Collections; +import java.util.Iterator; import java.util.Map; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; +import java.util.function.Supplier; import javax.annotation.Nullable; /** * Global key-value store used when field-injection is not possible. Since the same object may * participate in multiple stores each global key captures the store identity along with a weak - * reference to the original object key. + * reference to the owning key object. + * + *

The store is split into two maps with separate reference queues: young and old. Ageing the + * store by one generation creates a new young map; the previous young map becomes the old map. */ public final class GlobalObjectStore { - /** Never allow more than this number of objects in the global store. */ + /** Target ceiling for the total number of objects in the global store, young and old. */ private static final int GLOBAL_HARD_LIMIT = 100_000; - /** Temporarily allow more than this number of objects, but start removing old content. */ - private static final int GLOBAL_SOFT_LIMIT = 50_000; + /** Threshold at which we age the current store by one generation. */ + private static final int AGEING_THRESHOLD = GLOBAL_HARD_LIMIT / 2; + + /** Target ceiling for the total number of objects allowed after background eviction. */ + private static final int GLOBAL_SOFT_LIMIT = (GLOBAL_HARD_LIMIT + AGEING_THRESHOLD) / 2; /** Threshold at which we start doing limited cleanup at the same time as put operations. */ private static final int INLINE_CLEANUP_THRESHOLD = 5_000; - /** Threshold at which we start sampling keys to track old content. */ - private static final int OLD_KEYS_THRESHOLD = 512; + /** Constant supplier used when nothing in the store is considered old. */ + private static final Supplier NO_OLD_STALE_KEYS = () -> null; + + /** The current generation of the global object store. */ + private static volatile GlobalObjectStore store = new GlobalObjectStore(); + + /** Token used to decide which thread gets to age the store. */ + private static final AtomicBoolean ageing = new AtomicBoolean(); - private static final Object staleEntriesLock = new Object(); + // the following fields represent a generation of the object store - private static final Map weakMap = new ConcurrentHashMap<>(); + /** Supplies store keys where the key object is unused and eligible for collection. */ + private final ReferenceQueue staleKeys = new ReferenceQueue<>(); - private static final Set oldKeys = new HashSet<>(); + /** Map of weak store keys to value objects. */ + private final ConcurrentHashMap map; - private static int previousEstimate = 0; + /** Supplies old keys where the key object is unused and eligible for collection. */ + private final Supplier oldStaleKeys; + + /** Map of old store keys to value objects. */ + private final Map oldMap; + + private GlobalObjectStore() { + this.map = new ConcurrentHashMap<>(); + this.oldStaleKeys = NO_OLD_STALE_KEYS; + this.oldMap = Collections.emptyMap(); + } - private GlobalObjectStore() {} + private GlobalObjectStore(GlobalObjectStore oldStore) { + this.map = new ConcurrentHashMap<>(INLINE_CLEANUP_THRESHOLD); + this.oldStaleKeys = oldStore.staleKeys::poll; + this.oldMap = oldStore.map; + } /** * Removes stale entries from the global object-store, where the key object is now unused. @@ -54,55 +84,35 @@ private GlobalObjectStore() {} * @return the estimated remaining size of the global object-store */ public static int removeStaleEntries() { - synchronized (staleEntriesLock) { - Map weakMap = GlobalObjectStore.weakMap; - int estimatedSize = weakMap.size(); // capture size before any cleanup - StoreKey key; - while ((key = StoreKey.pollStaleKeys()) != null) { - if (weakMap.remove(key) != null) { - estimatedSize--; - } - } + return store.doRemoveStaleEntries(); + } - // The following code handles proactively removing old content in an attempt to guide the - // store below its soft limit. We remove older objects before recent additions, assuming - // that older objects are less likely to be used. For performance reasons this only runs - // after observed periods of growth or reduction, or if the store is near its hard limit. - - // We deliberately avoid tracking exact age, and instead regularly sample keys to maintain - // a small set that we know are still alive after a couple of calls to removeStaleEntries. - - if (Math.abs(estimatedSize - previousEstimate) > OLD_KEYS_THRESHOLD - || estimatedSize >= (GLOBAL_HARD_LIMIT + GLOBAL_SOFT_LIMIT) / 2) { - - if (estimatedSize >= GLOBAL_SOFT_LIMIT) { - // start proactively removing old content to keep growth in check - for (StoreKey oldKey : oldKeys) { - if (weakMap.remove(oldKey) != null) { - estimatedSize--; - } - } - oldKeys.clear(); - } else { - // have any of the old previously sampled keys been collected? - oldKeys.removeIf(StoreKey::isStale); - } + private int doRemoveStaleEntries() { + Object staleKey; - int refill = OLD_KEYS_THRESHOLD - oldKeys.size(); - if (refill > 0) { - // sample of keys at this time, don't need strict age ordering - for (StoreKey sampleKey : weakMap.keySet()) { - if (oldKeys.add(sampleKey) && --refill == 0) { - break; - } - } - } + // first remove stale entries from the young map + while ((staleKey = staleKeys.poll()) != null) { + //noinspection All: we know staleKey is a store key + map.remove(staleKey); + } - previousEstimate = estimatedSize; - } + // next remove stale entries from the old map + while ((staleKey = oldStaleKeys.get()) != null) { + //noinspection All: we know staleKey is a store key + oldMap.remove(staleKey); + } - return estimatedSize; + int estimatedSize = map.size() + oldMap.size(); + + // randomly evict old content to keep us below the soft limit + Iterator itr = oldMap.keySet().iterator(); + while (estimatedSize >= GLOBAL_SOFT_LIMIT && itr.hasNext()) { + itr.next(); + itr.remove(); + estimatedSize--; } + + return estimatedSize; } /** @@ -114,10 +124,17 @@ public static int removeStaleEntries() { */ @Nullable public static Object get(Object key, int storeId) { + return store.doGet(key, storeId); + } + + @Nullable + private Object doGet(Object key, int storeId) { LookupKey lookupKey = LookupKey.with(key, storeId); try { //noinspection All: intentionally use lookup key without reference overhead - return weakMap.get(lookupKey); + Object value = map.get(lookupKey); + //noinspection All: intentionally use lookup key without reference overhead + return value != null ? value : oldMap.get(lookupKey); } finally { lookupKey.reset(); } @@ -131,13 +148,18 @@ public static Object get(Object key, int storeId) { * @param value the new value */ public static void put(Object key, int storeId, @Nullable Object value) { + GlobalObjectStore s = store; if (value == null) { - remove(key, storeId); - } else if (checkCapacity()) { - weakMap.put(new StoreKey(key, storeId), value); + s.doRemove(key, storeId); + } else { + s.checkCapacity().doPut(key, storeId, value); } } + private void doPut(Object key, int storeId, Object value) { + map.put(new StoreKey(staleKeys, key, storeId), value); + } + /** * Gets the value currently associated with the given key and store-id. If no value exists then * associate the key and store-id with the given value and return that. @@ -148,12 +170,17 @@ public static void put(Object key, int storeId, @Nullable Object value) { * @return existing value if present, otherwise the new value */ public static Object getOrPut(Object key, int storeId, @Nullable Object value) { - Object existing = get(key, storeId); + GlobalObjectStore s = store; + Object existing = s.doGet(key, storeId); // avoids creating unnecessary store key if (existing != null || value == null) { return existing; - } else if (checkCapacity()) { - existing = weakMap.putIfAbsent(new StoreKey(key, storeId), value); + } else { + return s.checkCapacity().doGetOrPut(key, storeId, value); } + } + + private Object doGetOrPut(Object key, int storeId, Object value) { + Object existing = map.putIfAbsent(new StoreKey(staleKeys, key, storeId), value); return existing != null ? existing : value; } @@ -166,16 +193,21 @@ public static Object getOrPut(Object key, int storeId, @Nullable Object value) { * @param valueFunction function to compute values from keys * @return existing value if present, otherwise the new computed value */ - @SuppressWarnings({"rawtypes", "unchecked"}) + @SuppressWarnings({"rawtypes"}) public static Object getOrCompute(Object key, int storeId, Function valueFunction) { - Object existing = get(key, storeId); + GlobalObjectStore s = store; + Object existing = s.doGet(key, storeId); // avoids creating unnecessary store key if (existing != null) { return existing; - } else if (checkCapacity()) { - return weakMap.computeIfAbsent( - new StoreKey(key, storeId), storeKey -> valueFunction.apply(storeKey.get())); + } else { + return s.checkCapacity().doGetOrCompute(key, storeId, valueFunction); } - return valueFunction.apply(key); + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private Object doGetOrCompute(Object key, int storeId, Function valueFunction) { + return map.computeIfAbsent( + new StoreKey(staleKeys, key, storeId), unused -> valueFunction.apply(key)); } /** @@ -187,54 +219,79 @@ public static Object getOrCompute(Object key, int storeId, Function valueFunctio */ @Nullable public static Object remove(Object key, int storeId) { + return store.doRemove(key, storeId); + } + + @Nullable + private Object doRemove(Object key, int storeId) { LookupKey lookupKey = LookupKey.with(key, storeId); try { //noinspection All: intentionally use lookup key without reference overhead - return weakMap.remove(lookupKey); + Object value = map.remove(lookupKey); + //noinspection All: intentionally use lookup key without reference overhead + Object oldValue = oldMap.remove(lookupKey); + return value != null ? value : oldValue; } finally { lookupKey.reset(); } } /** - * @return {@code true} if there is space to add new objects. + * Checks store capacity, performing inline eviction or ageing if appropriate. + * + * @return the latest generation of the global store + */ + private GlobalObjectStore checkCapacity() { + int youngSize = map.size(); + if (youngSize >= INLINE_CLEANUP_THRESHOLD) { + Object staleKey = staleKeys.poll(); + if (staleKey != null) { + //noinspection All: we know staleKey is a store key + map.remove(staleKey); + } + if (youngSize >= AGEING_THRESHOLD) { + return maybeAgeStore(); + } + } + return this; + } + + /** + * Attempts to age this store by one generation; if already ageing don't block, use latest. + * + * @return the latest generation of the global store */ - private static boolean checkCapacity() { - int estimatedSize = weakMap.size(); - if (estimatedSize > INLINE_CLEANUP_THRESHOLD) { - // periodic cleanup may not be enough, start performing inline cleanup - StoreKey staleKey = StoreKey.pollStaleKeys(); - if (staleKey == null) { - return estimatedSize < GLOBAL_HARD_LIMIT; + @SuppressFBWarnings("ST") // we want to update the global object store + private GlobalObjectStore maybeAgeStore() { + // first try to get the token that allows us to age the global store + boolean attemptAgeing = ageing.compareAndSet(false, true); + // only after this get the latest generation of the store + GlobalObjectStore s = store; + if (attemptAgeing) { + try { + if (s == this) { + // our store is still the latest; go ahead and age it + s = store = new GlobalObjectStore(this); + } + } finally { + ageing.set(false); // relinquish the token } - weakMap.remove(staleKey); } - return true; + return s; // always return the latest generation of the store } /** Key used to weakly associate a non-injected key and store-id with a value. */ private static final class StoreKey extends WeakReference { - // stale store keys where the key object is unused and eligible for collection - private static final ReferenceQueue staleKeys = new ReferenceQueue<>(); - final int hash; final int storeId; - StoreKey(Object key, int storeId) { + StoreKey(ReferenceQueue staleKeys, Object key, int storeId) { super(key, staleKeys); this.hash = (31 * storeId) + System.identityHashCode(key); this.storeId = storeId; } - static StoreKey pollStaleKeys() { - return (StoreKey) staleKeys.poll(); - } - - boolean isStale() { - return get() == null; - } - @Override public int hashCode() { return hash; diff --git a/field-inject/src/test/java/datadog/instrument/fieldinject/ObjectStoreTest.java b/field-inject/src/test/java/datadog/instrument/fieldinject/ObjectStoreTest.java index 86d9048..a3982c4 100644 --- a/field-inject/src/test/java/datadog/instrument/fieldinject/ObjectStoreTest.java +++ b/field-inject/src/test/java/datadog/instrument/fieldinject/ObjectStoreTest.java @@ -219,6 +219,45 @@ void removeStaleEntriesIsIdempotent() { assertEquals("stable", store.get(key)); } + // --- generational capacity / eviction --- + + // Mirrors the private thresholds in GlobalObjectStore; kept here so the test intent is clear + // without exposing internals. If those thresholds change this test should be revisited. + private static final int AGEING_THRESHOLD = 50_000; + private static final int GLOBAL_SOFT_LIMIT = 75_000; + private static final int GLOBAL_HARD_LIMIT = 100_000; + + @Test + void sustainedInsertionIsBoundedByAgeingAndSoftLimitTrim() { + ObjectStore capStore = + ObjectStore.of("test.Capacity.Key", "test.Capacity.Value"); + + // Insert enough distinct, strongly-referenced keys to drive the young generation past + // AGEING_THRESHOLD several times over, landing mid-cycle (comfortably above the soft + // limit) so both inline ageing and removeStaleEntries' soft-limit trim get exercised. + int totalInserts = (AGEING_THRESHOLD * 4) + 40_000; + List keys = new ArrayList<>(totalInserts); + for (int i = 0; i < totalInserts; i++) { + Object key = new Object(); + keys.add(key); + capStore.put(key, i); + } + + // Inline enforceCapacity keeps young+old from ever exceeding the hard limit by ageing + // young into old before that point is reached, so recently inserted keys must still be + // retrievable even after hundreds of thousands of insertions. + Object lastKey = keys.get(keys.size() - 1); + assertEquals(totalInserts - 1, capStore.get(lastKey)); + + int finalSize = ObjectStore.removeStaleEntries(); + assertTrue( + finalSize < GLOBAL_HARD_LIMIT, + "Sustained insertion should have triggered eviction rather than unbounded growth"); + assertTrue( + finalSize <= GLOBAL_SOFT_LIMIT, + "removeStaleEntries should trim content back to the soft limit, observed " + finalSize); + } + // --- concurrency --- @Test