Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

GH-5107 shacl sail lost connection causes deadlock #5110

Draft
wants to merge 4 commits into
base: main
Choose a base branch
from
Draft

GH-5107 shacl sail lost connection causes deadlock #5110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8355988
reproduce test failure
hmottestad Aug 14, 2024
2779ef3
various attempts at fixing issue
hmottestad Aug 14, 2024
b8cf968
wip
hmottestad Dec 29, 2024
dbcdf26
wip
hmottestad Jan 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 21 additions & 5 deletions ...rc/main/java/org/eclipse/rdf4j/common/concurrent/locks/ExclusiveReentrantLockManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ public class ExclusiveReentrantLockManager {

private final int waitToCollect;

LockMonitoring<ExclusiveReentrantLock> lockMonitoring;
LockMonitoring<Lock> lockMonitoring;

public ExclusiveReentrantLockManager() {
this(false);
Expand All @@ -50,18 +50,18 @@ public ExclusiveReentrantLockManager(boolean trackLocks, int collectionFrequency

if (trackLocks || Properties.lockTrackingEnabled()) {

lockMonitoring = new LockTracking(
lockMonitoring = new LockTracking<>(
true,
"ExclusiveReentrantLockManager",
"ExclusiveReentrantLockManager(w/tracking)",
LoggerFactory.getLogger(this.getClass()),
waitToCollect,
Lock.ExtendedSupplier.wrap(this::getExclusiveLockInner, this::tryExclusiveLockInner)
);

} else {
lockMonitoring = new LockCleaner(
lockMonitoring = new LockCleaner<>(
false,
"ExclusiveReentrantLockManager",
"ExclusiveReentrantLockManager(w/cleaner)",
LoggerFactory.getLogger(this.getClass()),
Lock.ExtendedSupplier.wrap(this::getExclusiveLockInner, this::tryExclusiveLockInner)
);
Expand All @@ -87,6 +87,8 @@ private Lock tryExclusiveLockInner() {

}

private final AtomicLong ownerIsDead = new AtomicLong();

private Lock getExclusiveLockInner() throws InterruptedException {

synchronized (owner) {
Expand All @@ -100,6 +102,14 @@ private Lock getExclusiveLockInner() throws InterruptedException {
if (lock != null) {
return lock;
} else {
if (!owner.get().isAlive()) {
long l = ownerIsDead.incrementAndGet();
if (l > 10) {
ownerIsDead.set(0);
continue;
}

}
lockMonitoring.runCleanup();
owner.wait(waitToCollect);
}
Expand All @@ -113,6 +123,12 @@ private Lock getExclusiveLockInner() throws InterruptedException {
if (lock != null) {
return lock;
} else {
if (!owner.get().isAlive()) {
System.out.println("Owner thread is dead");
// activeLocks.set(0);
// owner.set(null);
// continue;
}
owner.wait(waitToCollect);
}
}
Expand Down
2 changes: 2 additions & 0 deletions core/sail/api/src/main/java/org/eclipse/rdf4j/sail/helpers/AbstractSail.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.WeakHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.eclipse.rdf4j.common.transaction.IsolationLevel;
Expand Down Expand Up @@ -119,6 +120,7 @@ protected static boolean debugEnabled() {
* debugging was disable at the time the connection was acquired.
*/
private final Map<SailConnection, Throwable> activeConnections = new IdentityHashMap<>();
// private final Map<SailConnection, Throwable> activeConnections = new WeakHashMap<>();

/*
* constructors
Expand Down
79 changes: 56 additions & 23 deletions core/sail/api/src/main/java/org/eclipse/rdf4j/sail/helpers/AbstractSailConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,9 @@
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;

import org.eclipse.rdf4j.common.annotation.InternalUseOnly;
import org.eclipse.rdf4j.common.concurrent.locks.ExclusiveReentrantLockManager;
import org.eclipse.rdf4j.common.concurrent.locks.Lock;
import org.eclipse.rdf4j.common.concurrent.locks.diagnostics.ConcurrentCleaner;
import org.eclipse.rdf4j.common.iteration.CloseableIteration;
Expand Down Expand Up @@ -98,17 +98,16 @@ public abstract class AbstractSailConnection implements SailConnection {
private final AtomicReference<Thread> activeThread = new AtomicReference<>();

@SuppressWarnings("FieldMayBeFinal")
private boolean isOpen = true;
private volatile boolean isOpen = true;
private static final VarHandle IS_OPEN;

private Thread owner;

/**
* Lock used to prevent concurrent calls to update methods like addStatement, clear, commit, etc. within a
* transaction.
*
*/
private final ReentrantLock updateLock = new ReentrantLock();
private final ExclusiveReentrantLockManager updateLock = new ExclusiveReentrantLockManager();
private final LongAdder iterationsOpened = new LongAdder();
private final LongAdder iterationsClosed = new LongAdder();

Expand Down Expand Up @@ -200,8 +199,7 @@ public void begin(IsolationLevel isolationLevel) throws SailException {
activeThread.setRelease(Thread.currentThread());

verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();
try {
if (isActive()) {
throw new SailException("a transaction is already active on this connection.");
Expand All @@ -210,8 +208,11 @@ public void begin(IsolationLevel isolationLevel) throws SailException {
startTransactionInternal();
txnActive = true;
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand Down Expand Up @@ -505,15 +506,19 @@ public final void prepare() throws SailException {
activeThread.setRelease(Thread.currentThread());
verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();

try {
if (txnActive) {
prepareInternal();
txnPrepared = true;
}
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand All @@ -535,7 +540,8 @@ public final void commit() throws SailException {

verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();

try {
if (txnActive) {
if (!txnPrepared) {
Expand All @@ -546,8 +552,11 @@ public final void commit() throws SailException {
txnPrepared = false;
}
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand All @@ -572,7 +581,8 @@ public final void rollback() throws SailException {

verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();

try {
if (txnActive) {
try {
Expand All @@ -586,8 +596,11 @@ public final void rollback() throws SailException {
debugEnabled ? new Throwable() : null);
}
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand Down Expand Up @@ -694,13 +707,17 @@ public final void endUpdate(UpdateContext op) throws SailException {

verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();

try {
verifyIsActive();
endUpdateInternal(op);
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand Down Expand Up @@ -750,14 +767,18 @@ public final void clear(Resource... contexts) throws SailException {

verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();

try {
verifyIsActive();
clearInternal(contexts);
statementsRemoved = true;
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand Down Expand Up @@ -820,13 +841,17 @@ public final void setNamespace(String prefix, String name) throws SailException

verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();

try {
verifyIsActive();
setNamespaceInternal(prefix, name);
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand All @@ -848,13 +873,17 @@ public final void removeNamespace(String prefix) throws SailException {

verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();

try {
verifyIsActive();
removeNamespaceInternal(prefix);
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand All @@ -873,13 +902,17 @@ public final void clearNamespaces() throws SailException {

verifyIsOpen();

updateLock.lock();
Lock exclusiveLock = updateLock.getExclusiveLock();

try {
verifyIsActive();
clearNamespacesInternal();
} finally {
updateLock.unlock();
exclusiveLock.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SailException(e);
} finally {
try {
activeThread.setRelease(null);
Expand Down
3 changes: 2 additions & 1 deletion ...est/java/org/eclipse/rdf4j/common/concurrent/locks/ExclusiveReentrantLockManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,8 @@ void stalledTest() throws InterruptedException {

assertThat(memoryAppender.countEventsForLogger(ExclusiveReentrantLockManager.class.getName()))
.isGreaterThanOrEqualTo(1);
memoryAppender.assertContains("is waiting on a possibly stalled lock \"ExclusiveReentrantLockManager\" with id",
memoryAppender.assertContains(
"is waiting on a possibly stalled lock \"ExclusiveReentrantLockManager(w/tracking)\" with id",
Level.INFO);
memoryAppender.assertContains(
"at org.eclipse.rdf4j.common.concurrent.locks.ExclusiveReentrantLockManagerTest.lambda$stalledTest$0(ExclusiveReentrantLockManagerTest.java:",
Expand Down
Loading
Loading