GH-5438 Use thread-local pool in LmdbStore to minimize synchronization

kenwenzel · kenwenzel · commit 18f1bef0d689 · 2025-09-18T20:01:00.000+02:00
diff --git a/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/LmdbContextIdIterator.java b/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/LmdbContextIdIterator.java
@@ -65,8 +65,8 @@ class LmdbContextIdIterator implements Closeable {
 
 	private final Thread ownerThread = Thread.currentThread();
 
-	LmdbContextIdIterator(Pool pool, int dbi, Txn txnRef) throws IOException {
-		this.pool = pool;
+	LmdbContextIdIterator(int dbi, Txn txnRef) throws IOException {
+		this.pool = Pool.get();
 		this.keyData = pool.getVal();
 		this.valueData = pool.getVal();
 
diff --git a/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/LmdbRecordIterator.java b/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/LmdbRecordIterator.java
@@ -79,9 +79,9 @@ class LmdbRecordIterator implements RecordIterator {
 
 	private final Thread ownerThread = Thread.currentThread();
 
-	LmdbRecordIterator(Pool pool, TripleIndex index, boolean rangeSearch, long subj, long pred, long obj,
+	LmdbRecordIterator(TripleIndex index, boolean rangeSearch, long subj, long pred, long obj,
 			long context, boolean explicit, Txn txnRef) throws IOException {
-		this.pool = pool;
+		this.pool = Pool.get();
 		this.keyData = pool.getVal();
 		this.valueData = pool.getVal();
 		this.index = index;
diff --git a/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/LmdbStoreConnection.java b/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/LmdbStoreConnection.java
@@ -200,4 +200,10 @@ public void clearInferred(Resource... contexts) throws SailException {
 		sailChangedEvent.setStatementsRemoved(true);
 	}
 
+	@Override
+	protected void closeInternal() throws SailException {
+		super.closeInternal();
+		// release thread-local pool
+		Pool.release();
+	}
 }
diff --git a/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/Pool.java b/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/Pool.java
@@ -19,81 +19,85 @@
  * A simple pool for {@link MDBVal}, {@link ByteBuffer} and {@link Statistics} instances.
  */
 class Pool {
+	// thread-local pool instance
+	private static final ThreadLocal<Pool> threadlocal = ThreadLocal.withInitial(Pool::new);
 
 	private final MDBVal[] valPool = new MDBVal[1024];
 	private final ByteBuffer[] keyPool = new ByteBuffer[1024];
 	private final Statistics[] statisticsPool = new Statistics[512];
-	private volatile int valPoolIndex = -1;
-	private volatile int keyPoolIndex = -1;
-	private volatile int statisticsPoolIndex = -1;
+	private int valPoolIndex = -1;
+	private int keyPoolIndex = -1;
+	private int statisticsPoolIndex = -1;
 
-	MDBVal getVal() {
-		synchronized (valPool) {
-			if (valPoolIndex >= 0) {
-				return valPool[valPoolIndex--];
-			}
+	final MDBVal getVal() {
+		if (valPoolIndex >= 0) {
+			return valPool[valPoolIndex--];
 		}
 		return MDBVal.malloc();
 	}
 
-	ByteBuffer getKeyBuffer() {
-		synchronized (keyPool) {
-			if (keyPoolIndex >= 0) {
-				ByteBuffer bb = keyPool[keyPoolIndex--];
-				bb.clear();
-				return bb;
-			}
+	final ByteBuffer getKeyBuffer() {
+		if (keyPoolIndex >= 0) {
+			ByteBuffer bb = keyPool[keyPoolIndex--];
+			bb.clear();
+			return bb;
 		}
 		return MemoryUtil.memAlloc(TripleStore.MAX_KEY_LENGTH);
 	}
 
-	Statistics getStatistics() {
-		synchronized (statisticsPool) {
-			if (statisticsPoolIndex >= 0) {
-				return statisticsPool[statisticsPoolIndex--];
-			}
+	final Statistics getStatistics() {
+		if (statisticsPoolIndex >= 0) {
+			return statisticsPool[statisticsPoolIndex--];
 		}
 		return new Statistics();
 	}
 
-	void free(MDBVal val) {
-		synchronized (valPool) {
-			if (valPoolIndex < valPool.length - 1) {
-				valPool[++valPoolIndex] = val;
-			} else {
-				val.close();
-			}
+	final void free(MDBVal val) {
+		if (valPoolIndex < valPool.length - 1) {
+			valPool[++valPoolIndex] = val;
+		} else {
+			val.close();
 		}
 	}
 
-	void free(ByteBuffer bb) {
-		synchronized (keyPool) {
-			if (keyPoolIndex < keyPool.length - 1) {
-				keyPool[++keyPoolIndex] = bb;
-			} else {
-				MemoryUtil.memFree(bb);
-			}
+	final void free(ByteBuffer bb) {
+		if (keyPoolIndex < keyPool.length - 1) {
+			keyPool[++keyPoolIndex] = bb;
+		} else {
+			MemoryUtil.memFree(bb);
 		}
 	}
 
-	void free(Statistics statistics) {
-		synchronized (statisticsPool) {
-			if (statisticsPoolIndex < statisticsPool.length - 1) {
-				statisticsPool[++statisticsPoolIndex] = statistics;
-			}
+	final void free(Statistics statistics) {
+		if (statisticsPoolIndex < statisticsPool.length - 1) {
+			statisticsPool[++statisticsPoolIndex] = statistics;
 		}
 	}
 
-	void close() {
-		synchronized (valPool) {
-			while (valPoolIndex >= 0) {
-				valPool[valPoolIndex--].close();
-			}
+	final void close() {
+		while (valPoolIndex >= 0) {
+			valPool[valPoolIndex--].close();
 		}
-		synchronized (keyPool) {
-			while (keyPoolIndex >= 0) {
-				MemoryUtil.memFree(keyPool[keyPoolIndex--]);
-			}
+		while (keyPoolIndex >= 0) {
+			MemoryUtil.memFree(keyPool[keyPoolIndex--]);
 		}
 	}
+
+	/**
+	 * Get a pool instance for the current thread.
+	 *
+	 * @return a Pool instance
+	 */
+	public static Pool get() {
+		return threadlocal.get();
+	}
+
+	/**
+	 * Release the pool instance for the current thread.
+	 */
+	public static void release() {
+		Pool pool = threadlocal.get();
+		pool.close();
+		threadlocal.remove();
+	}
 }
diff --git a/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/TripleStore.java b/core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/TripleStore.java
@@ -163,7 +163,6 @@ class TripleStore implements Closeable {
 	private long mapSize;
 	private long writeTxn;
 	private final TxnManager txnManager;
-	private final Pool pool = new Pool();
 
 	private TxnRecordCache recordCache = null;
 
@@ -391,7 +390,7 @@ private void reindex(Set<String> currentIndexSpecs, Set<String> newIndexSpecs) t
 						TripleIndex addedIndex = new TripleIndex(fieldSeq);
 						RecordIterator[] sourceIter = { null };
 						try {
-							sourceIter[0] = new LmdbRecordIterator(pool, sourceIndex, false, -1, -1, -1, -1,
+							sourceIter[0] = new LmdbRecordIterator(sourceIndex, false, -1, -1, -1, -1,
 									explicit, txnManager.createTxn(txn));
 
 							RecordIterator it = sourceIter[0];
@@ -482,7 +481,7 @@ public void close() throws IOException {
 	 * @throws IOException
 	 */
 	public LmdbContextIdIterator getContexts(Txn txn) throws IOException {
-		return new LmdbContextIdIterator(this.pool, this.contextsDbi, txn);
+		return new LmdbContextIdIterator(this.contextsDbi, txn);
 	}
 
 	/**
@@ -511,7 +510,7 @@ public RecordIterator getTriples(Txn txn, long subj, long pred, long obj, long c
 
 	private RecordIterator getTriplesUsingIndex(Txn txn, long subj, long pred, long obj, long context,
 			boolean explicit, TripleIndex index, boolean rangeSearch) throws IOException {
-		return new LmdbRecordIterator(pool, index, rangeSearch, subj, pred, obj, context, explicit, txn);
+		return new LmdbRecordIterator(index, rangeSearch, subj, pred, obj, context, explicit, txn);
 	}
 
 	/**
@@ -676,6 +675,7 @@ protected double cardinality(long subj, long pred, long obj, long context) throw
 		}
 
 		return txnManager.doWith((stack, txn) -> {
+			Pool pool = Pool.get();
 			final Statistics s = pool.getStatistics();
 			try {
 				MDBVal maxKey = MDBVal.malloc(stack);
@@ -1329,7 +1329,6 @@ public String toString() {
 		void close() {
 			mdb_dbi_close(env, dbiExplicit);
 			mdb_dbi_close(env, dbiInferred);
-			pool.close();
 		}
 
 		void clear(long txn) {

Original file line number	Diff line number	Diff line change
`@@ -200,4 +200,10 @@ public void clearInferred(Resource... contexts) throws SailException {`
`200`	`200`	`sailChangedEvent.setStatementsRemoved(true);`
`201`	`201`	`}`
`202`	`202`
	`203`	`+ @Override`
	`204`	`+ protected void closeInternal() throws SailException {`
	`205`	`+ super.closeInternal();`
	`206`	`+ // release thread-local pool`
	`207`	`+ Pool.release();`
	`208`	`+ }`
`203`	`209`	`}`