GH-4784 interrupt thread that is owning a connection before closing if we are closing from a different thread

Author: hmottestad

core/repository/sail/src/main/java/org/eclipse/rdf4j/repository/sail/SailRepositoryConnection.java

@@ -349,11 +349,19 @@ public RepositoryResult<Statement> getStatements(Resource subj, IRI pred, Value
 			Resource... contexts) throws RepositoryException {
 		Objects.requireNonNull(contexts,
 				"contexts argument may not be null; either the value should be cast to Resource or an empty array should be supplied");
-
+		CloseableIteration<? extends Statement, SailException> statements = null;
 		try {
-			return createRepositoryResult(sailConnection.getStatements(subj, pred, obj, includeInferred, contexts));
-		} catch (SailException e) {
-			throw new RepositoryException("Unable to get statements from Sail", e);
+			statements = sailConnection.getStatements(subj, pred, obj, includeInferred, contexts);
+			return createRepositoryResult(statements);
+		} catch (Throwable t) {
+			if (statements != null) {
+				statements.close();
+			}
+			if (t instanceof SailException) {
+				throw new RepositoryException("Unable to get statements from Sail", t);
+			} else {
+				throw t;
+			}
 		}
 	}
 

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

@@ -222,6 +222,36 @@ public void shutDown() throws SailException {
 				activeConnectionsCopy = new IdentityHashMap<>(activeConnections);
 			}
 
+			// Interrupt any threads that are still using a connection, in case they are waiting for a lock
+			for (Map.Entry<SailConnection, Throwable> entry : activeConnectionsCopy.entrySet()) {
+				try {
+					SailConnection con = entry.getKey();
+
+					if (con instanceof AbstractSailConnection) {
+						AbstractSailConnection sailCon = (AbstractSailConnection) con;
+						if (sailCon.getOwner() != Thread.currentThread()) {
+							sailCon.getOwner().interrupt();
+							sailCon.getOwner().join(1000);
+							if (sailCon.getOwner().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());
+							}
+						}
+					}
+
+				} catch (Throwable e) {
+					if (e instanceof InterruptedException) {
+						throw new SailException(e);
+					} else if (e instanceof AssertionError) {
+						// ignore assertions errors
+					} else if (e instanceof Error) {
+						throw (Error) e;
+					}
+					// ignore all other exceptions
+				}
+			}
+
 			// Forcefully close any connections that are still open
 			for (Map.Entry<SailConnection, Throwable> entry : activeConnectionsCopy.entrySet()) {
 				SailConnection con = entry.getKey();

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

@@ -12,13 +12,15 @@
 
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.VarHandle;
+import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.IdentityHashMap;
 import java.util.LinkedList;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.LongAdder;
+import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
@@ -98,15 +100,7 @@ public abstract class AbstractSailConnection implements SailConnection {
 	private boolean isOpen = true;
 	private static final VarHandle IS_OPEN;
 
-	static {
-		try {
-			IS_OPEN = MethodHandles.lookup()
-					.in(AbstractSailConnection.class)
-					.findVarHandle(AbstractSailConnection.class, "isOpen", boolean.class);
-		} catch (ReflectiveOperationException e) {
-			throw new Error(e);
-		}
-	}
+	private Thread owner;
 
 	/**
 	 * Lock used to prevent concurrent calls to update methods like addStatement, clear, commit, etc. within a
@@ -161,6 +155,7 @@ public AbstractSailConnection(AbstractSail sailBase) {
 		} else {
 			activeIterationsDebug = Collections.emptyMap();
 		}
+		owner = Thread.currentThread();
 	}
 
 	/*---------*
@@ -252,6 +247,8 @@ public boolean isActive() throws UnknownSailTransactionStateException {
 
 	@Override
 	public final void close() throws SailException {
+		Thread deadlockPreventionThread = startDeadlockPreventionThread();
+
 		// obtain an exclusive lock so that any further operations on this
 		// connection (including those from any concurrent threads) are blocked.
 		if (!IS_OPEN.compareAndSet(this, true, false)) {
@@ -268,7 +265,7 @@ public final void close() throws SailException {
 				if (sumDone == sumBlocking) {
 					break;
 				} else {
-					Thread.onSpinWait();
+					LockSupport.parkNanos(Duration.ofMillis(10).toNanos());
 				}
 			}
 
@@ -297,13 +294,56 @@ public final void close() throws SailException {
 				sailBase.connectionClosed(this);
 			}
 		} finally {
-			if (useConnectionLock) {
-				connectionLock.writeLock().unlock();
+			try {
+				if (deadlockPreventionThread != null) {
+					deadlockPreventionThread.interrupt();
+				}
+			} finally {
+				if (useConnectionLock) {
+					connectionLock.writeLock().unlock();
+				}
 			}
+
 		}
 
 	}
 
+	/**
+	 * If the current thread is not the owner, starts a thread to handle potential deadlocks by interrupting the owner.
+	 *
+	 * @return The started deadlock prevention thread or null if the current thread is the owner.
+	 */
+	private Thread startDeadlockPreventionThread() {
+		Thread deadlockPreventionThread = null;
+
+		if (Thread.currentThread() != owner) {
+			if (logger.isInfoEnabled()) {
+				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"));
+			}
+			deadlockPreventionThread = new Thread(() -> {
+				try {
+					Thread.sleep(sailBase.connectionTimeOut / 2);
+
+					owner.interrupt();
+					owner.join(1000);
+					if (owner.isAlive()) {
+						logger.error("Interrupted thread {} but thread is still alive after 1000 ms!", owner);
+					}
+
+				} catch (InterruptedException ignored) {
+				}
+
+			});
+			deadlockPreventionThread.setDaemon(true);
+			deadlockPreventionThread.start();
+
+		}
+		return deadlockPreventionThread;
+	}
+
 	@Override
 	public final CloseableIteration<? extends BindingSet, QueryEvaluationException> evaluate(TupleExpr tupleExpr,
 			Dataset dataset, BindingSet bindings, boolean includeInferred) throws SailException {
@@ -876,6 +916,16 @@ protected Lock getTransactionLock() throws SailException {
 		return new JavaLock(updateLock);
 	}
 
+	/**
+	 * This is for internal use only. It returns the thread that opened this connection.
+	 *
+	 * @return the thread that opened this connection.
+	 */
+	@InternalUseOnly
+	public Thread getOwner() {
+		return owner;
+	}
+
 	/**
 	 * Registers an iteration as active by wrapping it in a {@link SailBaseIteration} object and adding it to the list
 	 * of active iterations.
@@ -959,41 +1009,51 @@ protected AbstractSail getSailBase() {
 	}
 
 	private void forceCloseActiveOperations() throws SailException {
-		for (int i = 0; i < 10 && isActiveOperation() && !debugEnabled; i++) {
-			System.gc();
-			try {
-				Thread.sleep(1);
-			} catch (InterruptedException e) {
-				Thread.currentThread().interrupt();
+		Thread deadlockPreventionThread = startDeadlockPreventionThread();
+		try {
+			for (int i = 0; i < 10 && isActiveOperation() && !debugEnabled; i++) {
+				System.gc();
+				try {
+					Thread.sleep(1);
+				} catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+					throw new SailException(e);
+				}
 			}
-		}
 
-		if (debugEnabled) {
-
-			var activeIterationsCopy = new IdentityHashMap<>(activeIterationsDebug);
-			activeIterationsDebug.clear();
-
-			if (!activeIterationsCopy.isEmpty()) {
-				for (var entry : activeIterationsCopy.entrySet()) {
-					try {
-						logger.warn("Unclosed iteration", entry.getValue());
-						entry.getKey().close();
-					} catch (Exception e) {
-						if (e instanceof InterruptedException) {
-							Thread.currentThread().interrupt();
+			if (debugEnabled) {
+
+				var activeIterationsCopy = new IdentityHashMap<>(activeIterationsDebug);
+				activeIterationsDebug.clear();
+
+				if (!activeIterationsCopy.isEmpty()) {
+					for (var entry : activeIterationsCopy.entrySet()) {
+						try {
+							logger.warn("Unclosed iteration", entry.getValue());
+							entry.getKey().close();
+						} catch (Exception e) {
+							if (e instanceof InterruptedException) {
+								Thread.currentThread().interrupt();
+								throw new SailException(e);
+							}
+							logger.warn("Exception occurred while closing unclosed iterations.", e);
 						}
-						logger.warn("Exception occurred while closing unclosed iterations.", e);
 					}
-				}
 
-				var entry = activeIterationsCopy.entrySet().stream().findAny().orElseThrow();
+					var entry = activeIterationsCopy.entrySet().stream().findAny().orElseThrow();
 
-				throw new SailException(
-						"Connection closed before all iterations were closed: " + entry.getKey().toString(),
-						entry.getValue());
-			}
+					throw new SailException(
+							"Connection closed before all iterations were closed: " + entry.getKey().toString(),
+							entry.getValue());
+				}
 
+			}
+		} finally {
+			if (deadlockPreventionThread != null) {
+				deadlockPreventionThread.interrupt();
+			}
 		}
+
 	}
 
 	/**
@@ -1138,4 +1198,14 @@ public synchronized void release() {
 			}
 		}
 	}
+
+	static {
+		try {
+			IS_OPEN = MethodHandles.lookup()
+					.in(AbstractSailConnection.class)
+					.findVarHandle(AbstractSailConnection.class, "isOpen", boolean.class);
+		} catch (ReflectiveOperationException e) {
+			throw new Error(e);
+		}
+	}
 }

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

@@ -150,4 +150,5 @@ public Supplier<CollectionFactory> getCollectionFactory() {
 		verifyBaseSailSet();
 		return baseSail.getCollectionFactory();
 	}
+
 }

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

@@ -12,6 +12,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -20,6 +21,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 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;
@@ -236,7 +238,8 @@ public void close() throws SailException {
 							running.set(false);
 							tripleStoreExecutor.shutdown();
 							while (!tripleStoreExecutor.isTerminated()) {
-								Thread.yield();
+								tripleStoreExecutor.shutdownNow();
+								LockSupport.parkNanos(Duration.ofMillis(100).toNanos());
 							}
 							tripleStore.close();
 						}
@@ -561,6 +564,9 @@ private void startTransaction(boolean preferThreading) throws SailException {
 														op.execute();
 													}
 												} else {
+													if (Thread.interrupted()) {
+														throw new InterruptedException();
+													}
 													Thread.yield();
 												}
 											}