GH-5779 reduce cleanup cost ConcurrentCache in LMDB (#5780)

Author: hmottestad

.gitignore

@@ -56,3 +56,4 @@ e2e/test-results
 .serena/
 .vscode
 /.codex/environments/environment.toml
+/.m2_repo_linux_j25

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/ConcurrentCache.java

@@ -8,45 +8,135 @@
  *
  * SPDX-License-Identifier: BSD-3-Clause
  *******************************************************************************/
+// Some portions generated by Codex
 package org.eclipse.rdf4j.sail.lmdb;
 
-import java.util.Iterator;
+import java.util.Arrays;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
- * Limited-size concurrent cache. The actual cleanup to keep the size limited is done once per
- * <code>CLEANUP_INTERVAL</code> invocations of the protected method <code>cleanUp</code>. <code>cleanUp</code> method
- * is called every time by <code>put</code> The maximum size is maintained approximately. Cleanup is not done if size is
- * less than <code>capacity + CLEANUP_INTERVAL / 2</code>.
+ * Fixed-size concurrent cache with approximate FIFO eviction per hash set. The cache never grows beyond its slot
+ * budget; capacity is rounded up to a power-of-two number of slots and entries are evicted in O(1).
  */
 public class ConcurrentCache<K, V> {
 
-	private static final int CLEANUP_INTERVAL = 1024;
-
+	private static final int WAYS = 4;
 	private static final float LOAD_FACTOR = 0.75f;
 
-	private final int capacity;
-
-	private volatile int cleanupTick = 0;
+	private final Entry<K, V>[] entries;
+	private final int[] nextVictim;
+	private final int setMask;
+	private final int setShift;
 
 	protected final ConcurrentHashMap<K, V> cache;
 
+	private volatile long generation = 1;
+
+	@SuppressWarnings("unchecked")
 	public ConcurrentCache(int capacity) {
-		this.capacity = capacity;
-		this.cache = new ConcurrentHashMap<>((int) (capacity / LOAD_FACTOR), LOAD_FACTOR);
+		cache = new ConcurrentHashMap<>((int) (Math.max(1, capacity) / LOAD_FACTOR), LOAD_FACTOR);
+		if (capacity <= 0) {
+			entries = (Entry<K, V>[]) new Entry[0];
+			nextVictim = new int[0];
+			setMask = 0;
+			setShift = 0;
+			return;
+		}
+
+		int slotCount = nextPowerOfTwo(Math.max(WAYS, capacity));
+		int setCount = slotCount / WAYS;
+		entries = (Entry<K, V>[]) new Entry[slotCount];
+		nextVictim = new int[setCount];
+		setMask = setCount - 1;
+		setShift = Integer.numberOfTrailingZeros(setCount);
 	}
 
 	public V get(Object key) {
+		Objects.requireNonNull(key);
+		if (entries.length == 0) {
+			return cache.get(key);
+		}
+
+		int hash = spread(key.hashCode());
+		int base = setBase(hash);
+		int preferredOffset = preferredOffset(hash);
+		long currentGeneration = generation;
+
+		for (int i = 0; i < WAYS; i++) {
+			Entry<K, V> entry = entries[base + ((preferredOffset + i) & (WAYS - 1))];
+			if (entry != null && entry.hash == hash && entry.generation == currentGeneration
+					&& sameKey(key, entry.key)) {
+				return entry.value;
+			}
+		}
 		return cache.get(key);
 	}
 
 	public V put(K key, V value) {
+		Objects.requireNonNull(key);
+		Objects.requireNonNull(value);
 		cleanUp();
-		return cache.put(key, value);
+
+		V previous = cache.put(key, value);
+		if (entries.length == 0) {
+			return previous;
+		}
+
+		int hash = spread(key.hashCode());
+		int setIndex = hash & setMask;
+		int base = setIndex * WAYS;
+		int preferredOffset = preferredOffset(hash);
+		long currentGeneration = generation;
+		int emptySlot = -1;
+
+		for (int i = 0; i < WAYS; i++) {
+			int slot = base + ((preferredOffset + i) & (WAYS - 1));
+			Entry<K, V> entry = entries[slot];
+			if (entry == null) {
+				if (emptySlot < 0) {
+					emptySlot = slot;
+				}
+				continue;
+			}
+
+			if (entry.hash == hash && entry.generation == currentGeneration && sameKey(key, entry.key)) {
+				if (entry.value != value) {
+					entries[slot] = new Entry<>(key, value, hash, currentGeneration);
+				}
+				return previous;
+			}
+		}
+
+		if (emptySlot >= 0) {
+			entries[emptySlot] = new Entry<>(key, value, hash, currentGeneration);
+			return previous;
+		}
+
+		int firstVictim = nextVictim[setIndex] & (WAYS - 1);
+		for (int i = 0; i < WAYS; i++) {
+			int victimOffset = (firstVictim + i) & (WAYS - 1);
+			int slot = base + victimOffset;
+			Entry<K, V> victim = entries[slot];
+			if (victim == null || victim.generation != currentGeneration || onEntryRemoval(victim.key)) {
+				entries[slot] = new Entry<>(key, value, hash, currentGeneration);
+				if (victim != null && victim.generation == currentGeneration && !sameKey(victim.key, key)) {
+					cache.remove(victim.key, victim.value);
+				}
+				nextVictim[setIndex] = (victimOffset + 1) & (WAYS - 1);
+				return previous;
+			}
+		}
+
+		return previous;
 	}
 
 	public void clear() {
+		long nextGeneration = generation + 1;
+		generation = nextGeneration == 0 ? 1 : nextGeneration;
 		cache.clear();
+		Arrays.fill(entries, null);
+		Arrays.fill(nextVictim, 0);
 	}
 
 	/**
@@ -59,34 +149,41 @@ protected boolean onEntryRemoval(K key) {
 	}
 
 	protected void cleanUp() {
-		// This is not thread-safe, but the worst that can happen is that we may (rarely) get slightly longer
-		// cleanup intervals or run cleanUp twice
-		cleanupTick++;
-		if (cleanupTick <= CLEANUP_INTERVAL) {
-			return;
-		}
+		// Legacy hook. Eviction happens during put.
+	}
 
-		cleanupTick %= CLEANUP_INTERVAL;
+	private int setBase(int hash) {
+		return (hash & setMask) * WAYS;
+	}
 
-		synchronized (cache) {
+	private int preferredOffset(int hash) {
+		return (hash >>> setShift) & (WAYS - 1);
+	}
 
-			final int size = cache.size();
-			if (size < capacity + CLEANUP_INTERVAL / 2) {
-				return;
-			}
+	private static int spread(int hash) {
+		return hash ^ (hash >>> 16);
+	}
 
-			Iterator<K> iter = cache.keySet().iterator();
+	private static boolean sameKey(Object key, Object entryKey) {
+		return key == entryKey || key.equals(entryKey);
+	}
 
-			float removeEachTh = (float) size / (size - capacity);
+	private static int nextPowerOfTwo(int n) {
+		return 1 << (Integer.SIZE - Integer.numberOfLeadingZeros(n - 1));
+	}
 
-			for (int i = 0; iter.hasNext(); i++) {
+	private static final class Entry<K, V> {
 
-				K key = iter.next();
+		private final K key;
+		private final V value;
+		private final int hash;
+		private final long generation;
 
-				if (i % removeEachTh < 1) {
-					cache.computeIfPresent(key, (k, v) -> onEntryRemoval(k) ? null : v);
-				}
-			}
+		private Entry(K key, V value, int hash, long generation) {
+			this.key = key;
+			this.value = value;
+			this.hash = hash;
+			this.generation = generation;
 		}
 	}
 }

core/sail/lmdb/src/test/java/org/eclipse/rdf4j/sail/lmdb/ValueStoreNamespaceCacheTest.java

@@ -8,15 +8,19 @@
  *
  * SPDX-License-Identifier: BSD-3-Clause
  *******************************************************************************/
+// Some portions generated by Codex
 package org.eclipse.rdf4j.sail.lmdb;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 import java.io.File;
 import java.lang.invoke.MethodHandle;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.MethodType;
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.eclipse.rdf4j.sail.lmdb.config.LmdbStoreConfig;
@@ -25,6 +29,19 @@
 
 class ValueStoreNamespaceCacheTest {
 
+	@Test
+	void legacyProtectedApiRemainsAvailable() throws Exception {
+		Field cacheField = ConcurrentCache.class.getDeclaredField("cache");
+		assertNotNull(cacheField);
+		assertEquals("cache", cacheField.getName());
+		assertEquals(Modifier.PROTECTED | Modifier.FINAL, cacheField.getModifiers());
+
+		Method cleanUp = ConcurrentCache.class.getDeclaredMethod("cleanUp");
+		assertNotNull(cleanUp);
+		assertEquals(void.class, cleanUp.getReturnType());
+		assertEquals(Modifier.PROTECTED, cleanUp.getModifiers());
+	}
+
 	@Test
 	void getNamespaceUsesLastResult(@TempDir File dataDir) throws Throwable {
 		ValueStore valueStore = new ValueStore(new File(dataDir, "values"), new LmdbStoreConfig());