code cleanup

Author: hmottestad

core/sail/api/src/main/java/org/eclipse/rdf4j/sail/helpers/AbstractSail.java

@@ -229,13 +229,15 @@ public void shutDown() throws SailException {
 
 					if (con instanceof AbstractSailConnection) {
 						AbstractSailConnection sailCon = (AbstractSailConnection) con;
-						if (sailCon.getOwner() != Thread.currentThread()) {
-							sailCon.getOwner().interrupt();
-							sailCon.getOwner().join(1000);
-							if (sailCon.getOwner().isAlive()) {
+						Thread owner = sailCon.getOwner();
+						if (owner != Thread.currentThread()) {
+							owner.interrupt();
+							// wait up to 1 second for the owner thread to die
+							owner.join(1000);
+							if (owner.isAlive()) {
 								logger.error(
 										"Closing active connection due to shut down and interrupted the owning thread of the connection {} but thread is still alive after 1000 ms!",
-										sailCon.getOwner());
+										owner);
 							}
 						}
 					}

core/sail/api/src/main/java/org/eclipse/rdf4j/sail/helpers/AbstractSailConnection.java

@@ -265,7 +265,12 @@ public final void close() throws SailException {
 				if (sumDone == sumBlocking) {
 					break;
 				} else {
-					LockSupport.parkNanos(Duration.ofMillis(10).toNanos());
+					if (Thread.currentThread().isInterrupted()) {
+						throw new SailException(
+								"Connection was interrupted while waiting on active operations before it could be closed.");
+					} else {
+						LockSupport.parkNanos(Duration.ofMillis(10).toNanos());
+					}
 				}
 			}
 
@@ -317,26 +322,35 @@ private Thread startDeadlockPreventionThread() {
 		Thread deadlockPreventionThread = null;
 
 		if (Thread.currentThread() != owner) {
+
 			if (logger.isInfoEnabled()) {
+				// use info level for this because FedX prevalently closes connections from different threads
 				logger.info(
-						"Closing connection from a different thread than the one that opened it. Connections should not be shared between threads. Opened by "
-								+ owner + " closed by " + Thread.currentThread(),
-						new Throwable("Throwable used for stacktrace"));
+						"Closing connection from a different thread than the one that opened it. Connections should not be shared between threads. Opened by {} closed by {}",
+						owner, Thread.currentThread(), new Throwable("Throwable used for stacktrace"));
 			}
+
 			deadlockPreventionThread = new Thread(() -> {
 				try {
+					// This thread should sleep for a while so that the callee has a chance to finish.
+					// The callee will interrupt this thread when it is finished, which means that there were no
+					// deadlocks and we can exit.
 					Thread.sleep(sailBase.connectionTimeOut / 2);
 
 					owner.interrupt();
+					// wait for up to 1 second for the owner thread to die
 					owner.join(1000);
 					if (owner.isAlive()) {
 						logger.error("Interrupted thread {} but thread is still alive after 1000 ms!", owner);
 					}
 
 				} catch (InterruptedException ignored) {
+					// this thread is interrupted as a signal that there were no deadlocks, so the exception can be
+					// ignored and we can simply exit
 				}
 
 			});
+
 			deadlockPreventionThread.setDaemon(true);
 			deadlockPreventionThread.start();
 

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

@@ -12,16 +12,15 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.eclipse.rdf4j.common.iteration.CloseableIteration;
@@ -235,13 +234,21 @@ public void close() throws SailException {
 						}
 					} finally {
 						if (tripleStore != null) {
-							running.set(false);
-							tripleStoreExecutor.shutdown();
-							while (!tripleStoreExecutor.isTerminated()) {
-								tripleStoreExecutor.shutdownNow();
-								LockSupport.parkNanos(Duration.ofMillis(100).toNanos());
+							try {
+								running.set(false);
+								tripleStoreExecutor.shutdown();
+								try {
+									while (!tripleStoreExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+										logger.warn("Waiting for triple store executor to terminate");
+									}
+								} catch (InterruptedException e) {
+									Thread.currentThread().interrupt();
+									throw new SailException(e);
+								}
+							} finally {
+								tripleStore.close();
 							}
-							tripleStore.close();
+
 						}
 					}
 				}
@@ -564,17 +571,28 @@ private void startTransaction(boolean preferThreading) throws SailException {
 														op.execute();
 													}
 												} else {
-													if (Thread.interrupted()) {
+													if (!running.get()) {
+														logger.warn(
+																"LmdbSailStore was closed while active transaction was waiting for the next operation. Forcing a rollback!");
+														opQueue.add(ROLLBACK_TRANSACTION);
+													} else if (Thread.interrupted()) {
 														throw new InterruptedException();
+													} else {
+														Thread.yield();
 													}
-													Thread.yield();
 												}
 											}
 
 											// keep thread running for at least 2ms to lock-free wait for the next
 											// transaction
-											long start = System.currentTimeMillis();
+											long start = 0;
 											while (running.get() && !nextTransactionAsync) {
+												if (start == 0) {
+													// System.currentTimeMillis() is expensive, so only call it when we
+													// are sure we need to wait
+													start = System.currentTimeMillis();
+												}
+
 												if (System.currentTimeMillis() - start > 2) {
 													synchronized (storeTxnStarted) {
 														if (!nextTransactionAsync) {