From d311d8018456851700cc59eb4c437e15976e9f8e Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Thu, 17 Sep 2026 22:32:55 +0300 Subject: [PATCH] [#1064] Replay a JE transaction ended by a lock conflict instead of giving it up at once JEStorage.write made one attempt: the DeadlockException JE throws at the victim of a deadlock - the one conflict left with je.lock.timeout at 0 - reached the caller bare after that attempt, and the LDAP client was answered with a server error carrying JE's account of its lock table. PDBStorage and JDBCStorage replay under an attempt cap and a retry window; JEStorage now does the same for a LockConflictException, and gives it up the way they do. Three points are JE's own. The conflict is raised inside the operation and, should the operation swallow it, again by commit(), so both forms are matched and a swallowed conflict commits nothing. The replay backs off first: a record is locked by the LSN of its current version, the victim's abort undoes the version the survivor was granted the lock on, and a replay which comes back at once wins the race for the version put back and forms the deadlock again. And the interrupt flag is not restored when the backoff is interrupted, since JE invalidates the whole environment when a thread carrying the flag makes any call - which the caller's failure road does. --- .../backends/jeb/ConfigurableEnvironment.java | 9 +- .../opends/server/backends/jeb/JEStorage.java | 204 ++++++- .../server/backends/jeb/JEStorageTest.java | 514 +++++++++++++++++- 3 files changed, 709 insertions(+), 18 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/ConfigurableEnvironment.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/ConfigurableEnvironment.java index 896d0910ff..ab669e870e 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/ConfigurableEnvironment.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/ConfigurableEnvironment.java @@ -13,6 +13,7 @@ * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2010-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC. */ package org.opends.server.backends.jeb; @@ -353,9 +354,11 @@ private static EnvironmentConfig defaultConfig() envConfig.setConfigParam(LOG_FAULT_READ_SIZE, String.valueOf(4 * 1024)); } - // Disable lock timeouts, meaning that no lock wait - // timelimit is enforced and a deadlocked operation - // will block indefinitely. + // Disable lock timeouts, meaning that no lock wait timelimit is enforced: a writer waiting for a lock + // blocks until it is granted. A deadlock does not block: JE detects the cycle as soon as a wait would + // close it and ends one of its transactions, chosen at random, with a DeadlockException - the one + // conflict left under this setting, which JEStorage.write replays. An operator may set a timeout back + // through ds-cfg-je-property, and a wait that long is then reported and replayed the same way. envConfig.setLockTimeout(0, TimeUnit.MICROSECONDS); //FIX https://github.com/OpenIdentityPlatform/OpenDJ/issues/53 https://docs.oracle.com/cd/E17277_02/html/java/com/sleepycat/je/EnvironmentConfig.html#FREE_DISK diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/JEStorage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/JEStorage.java index 11fa660077..73016610ff 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/JEStorage.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jeb/JEStorage.java @@ -91,6 +91,7 @@ import com.sleepycat.je.Durability; import com.sleepycat.je.Environment; import com.sleepycat.je.EnvironmentConfig; +import com.sleepycat.je.LockConflictException; import com.sleepycat.je.OperationStatus; import com.sleepycat.je.Transaction; import com.sleepycat.je.TransactionConfig; @@ -682,6 +683,37 @@ private WriteableTransaction newWriteableTransaction(Transaction txn) private static final int IMPORT_DB_CACHE_SIZE = 32 * MB; + /** + * Number of attempts a {@link #write} makes before it propagates the conflict to the caller. + *

+ * It is a budget of attempts and not of time, so it is only ever reached by the conflicts that report quickly. + * With the shipped configuration every conflict does: {@code je.lock.timeout} is 0, so a writer waiting for a + * lock never times out, and the only conflict JE raises is the deadlock, which it detects as soon as the wait + * would close a cycle. An operator who sets a lock timeout through {@code ds-cfg-je-property} turns a wait that + * long into a conflict as well, and a conflict slower to report than {@link #MAX_RETRY_WINDOW_NANOS} spends the + * whole window inside its first attempt, is granted the single replay that window's exemption guarantees, and + * gives up on the window after two attempts rather than after this many. + */ + static final int MAX_RETRIES = 10; + + /** + * Wall-clock budget the replays of a {@link #write} may spend, in nanoseconds. It is checked between attempts, + * so an attempt already running is never interrupted, and never before one replay has been made: the loop + * returns after at most this window plus two attempts. It bounds the conflicts that are slow to report, which + * {@link #MAX_RETRIES} alone does not - an operation whose own work takes seconds would otherwise multiply that + * wait by the attempt count. + */ + static final long MAX_RETRY_WINDOW_NANOS = 10L * 1000L * 1000L * 1000L; //10 s + + /** + * Upper bound of the random delay before the second attempt, in milliseconds; it doubles with every attempt. + * The ladder is {@code PDBStorage}'s, kept so that the two engines back off alike. + */ + private static final double BASE_SLEEP_ON_RETRY_MS = 50.0; + + /** Upper bound the doubled delay is capped at, in milliseconds. */ + private static final double MAX_SLEEP_ON_RETRY_MS = 1000.0; + private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); /** Use read committed isolation instead of the default which is repeatable read. */ @@ -700,6 +732,10 @@ private WriteableTransaction newWriteableTransaction(Transaction txn) private DiskSpaceMonitor diskMonitor; private StorageStatus storageStatus = StorageStatus.working(); private final ConcurrentMap trees = new ConcurrentHashMap<>(); + /** Attempt bound of a {@link #write}, {@link #MAX_RETRIES} outside the tests. */ + private final int maxRetries; + /** Wall-clock bound of a {@link #write}, {@link #MAX_RETRY_WINDOW_NANOS} outside the tests. */ + private final long retryWindowNanos; /** * Creates a new JE storage with the provided configuration. @@ -712,8 +748,36 @@ private WriteableTransaction newWriteableTransaction(Transaction txn) * if memory cannot be reserved */ JEStorage(final JEBackendCfg cfg, ServerContext serverContext) throws ConfigException + { + this(cfg, serverContext, MAX_RETRIES, MAX_RETRY_WINDOW_NANOS); + } + + /** + * Creates a new JE storage whose replay bounds are the given ones rather than {@link #MAX_RETRIES} and + * {@link #MAX_RETRY_WINDOW_NANOS}. + *

+ * Only a test builds one of these, and it does so to stop the two bounds racing each other: with the shipped + * values a run of replays spends a random share of the window on backoff alone, so a test of the attempt cap + * can be ended by the window on a loaded machine, and a test of the window has to spend seconds of build time + * to reach it. + * + * @param cfg + * The configuration. + * @param serverContext + * This server instance context + * @param maxRetries + * Number of attempts a write makes before it propagates the conflict to the caller. + * @param retryWindowNanos + * Wall-clock budget the replays of a write may spend, in nanoseconds. + * @throws ConfigException + * if memory cannot be reserved + */ + JEStorage(final JEBackendCfg cfg, ServerContext serverContext, int maxRetries, long retryWindowNanos) + throws ConfigException { this.serverContext = serverContext; + this.maxRetries = maxRetries; + this.retryWindowNanos = retryWindowNanos; backendDirectory = getBackendDirectory(cfg); config = cfg; cfg.addJEChangeListener(this); @@ -963,27 +1027,143 @@ private static String toDatabaseName(final TreeName treeName) return treeName.toString(); } + /** + * {@inheritDoc} + *

+ * A transaction JE ends with a {@link LockConflictException} is replayed, bounded twice: by {@link #MAX_RETRIES} + * attempts and by the {@link #MAX_RETRY_WINDOW_NANOS} wall-clock window, whichever is spent first - except that + * the window alone never ends the replays before one has been made. It is bounded for the reason + * {@code PDBStorage} bounds its loop: the configuration change paths of the pluggable backend hold an entry + * container's exclusive lock across this method, and every reader of that suffix then waits - untimed and + * uninterruptibly - until it returns, so a conflict that never clears would park every worker thread of that + * suffix rather than fail one operation. + *

+ * JE raises the conflict from inside the operation - a record read or write, never the commit, which takes no + * lock - and the transaction wraps it in a {@link StorageRuntimeException}, which is unwrapped below before it is + * matched. The operation may catch it there; the transaction is then abort-only and {@code commit()} raises the + * conflict again, bare, so an attempt which swallowed its conflict commits nothing and is replayed all the same. + * With the shipped configuration the only conflict is the deadlock: {@code je.lock.timeout} is 0, so a writer + * waits for a lock rather than times out, and JE ends one transaction of a cycle - chosen at random - as soon as + * a wait would close it. A lock timeout set through {@code ds-cfg-je-property} makes a wait that long a conflict + * as well. + *

+ * The replay backs off first, though JE would make it wait for the locks it lost anyway: JE locks a record by + * the LSN of its current version, and the abort of the victim, which hands the waiting survivor the lock it + * asked for, undoes the version that lock belongs to - the survivor then has to lock the version the undo put + * back, and a replay which comes back at once wins that race, holds the record when the survivor asks again, + * and the same deadlock forms with the roles drawn afresh. JE's own retry example sleeps before retrying for + * this reason; without the sleep the deadlock of two writers was seen to form three times in a row. + *

+ * An interrupt reaches the loop only in that sleep, and the loop does not restore the flag the sleep cleared: + * JE invalidates the whole environment when a thread carrying the interrupt flag touches it - the transaction + * registry's latch is acquired interruptibly - and the caller's own failure road makes such a call + * ({@code EntryContainer.writeTrustState}). The interrupt is reported instead, with the conflict being replayed, + * as the suppressed exceptions of the {@link StorageRuntimeException} thrown. + *

+ * Once the bound is spent the conflict is reported as a {@link StorageRuntimeException} naming the backend, the + * attempts spent, the time they took and which of the two bounds ran out. It carries the conflict as a + * suppressed exception rather than as its cause: a cause is unwrapped below and thrown in its place, and + * {@code EntryContainer.throwAllowedExceptionTypes} likewise passes a {@link StorageRuntimeException} through + * untouched only while it has no cause. Given a cause, both hand the caller the bare conflict instead, whose + * message is JE's account of its lock table - all an LDAP client used to be told after the single attempt. + */ @Override public void write(final WriteOperation operation) throws Exception { - final Transaction txn = beginTransaction(); - try + final long startedAt = System.nanoTime(); + final long giveUpAt = startedAt + retryWindowNanos; + for (int attempt = 1;; attempt++) { - operation.run(newWriteableTransaction(txn)); - commit(txn); - } - catch (final StorageRuntimeException e) - { - if (e.getCause() != null) + final LockConflictException conflict; + final Transaction txn = beginTransaction(); + try { - throw (Exception) e.getCause(); + try + { + operation.run(newWriteableTransaction(txn)); + commit(txn); + return; + } + catch (final StorageRuntimeException e) + { + throw unwrap(e); + } + } + catch (final LockConflictException e) + { + conflict = e; + } + finally + { + abort(txn); + } + // System.nanoTime() - giveUpAt is the overflow safe form of the comparison, and attempt > 1 keeps the window + // from ending the loop before a single replay: a lock timeout set by the operator can outlast the window on + // its own, and it is the attempt after that one which is likeliest to succeed, the transaction that blocked + // it having just finished. One clock sample for both, so that the elapsed time reported is the one the give + // up was decided on + final long now = System.nanoTime(); + final boolean capSpent = attempt >= maxRetries; + if (capSpent || (attempt > 1 && now - giveUpAt >= 0)) + { + final long elapsedMs = TimeUnit.NANOSECONDS.toMillis(now - startedAt); + // which of the two bounds was spent, so that the config change paths - which report this as a bare stack + // trace with no message id - say whether raising the attempts or the window is what would have helped + final String boundSpent = capSpent ? "attempt cap" : "retry window"; + final StorageRuntimeException spent = new StorageRuntimeException( + "je: backend '" + config.getBackendId() + "' did not apply the transaction after " + attempt + + " attempts in " + elapsedMs + " ms, the " + boundSpent + " being spent; the last conflict was " + + conflict); + spent.addSuppressed(conflict); + // warned once, at exhaustion only: a conflict is answered by a replay, and the trace below is what says + // so. stackTraceToSingleLineString, the form the config change paths report this exception with, walks + // the causes and never prints a suppressed exception, so this line is the only rendering that carries the + // stack of the conflict + logger.warn(LocalizableMessage.raw("je: giving up on the transaction of backend '%s' after %d attempts" + + " in %d ms, the %s being spent: %s", config.getBackendId(), attempt, elapsedMs, boundSpent, + stackTraceToSingleLineString(conflict))); + throw spent; + } + if (logger.isTraceEnabled()) + { + logger.trace("je: replaying the transaction after %s, attempt %d of %d", conflict, attempt, maxRetries); + } + try + { + // slept for after the finally has ended the rolled back transaction, so that nothing of it is held + // through the delay; random and growing with every attempt, so that two writers replaying the same + // deadlock do not come back in step + Thread.sleep(retryDelayMillis(attempt)); + } + catch (final InterruptedException e) + { + // the flag stays cleared - see above - and the conflict being replayed is reported next to the interrupt, + // wrapped the way the exhausted loop wraps it and for the same reason: every caller strips a cause off + final StorageRuntimeException interrupted = new StorageRuntimeException( + "je: backend '" + config.getBackendId() + "' was interrupted while replaying the transaction after " + + attempt + " attempts; the last conflict was " + conflict); + interrupted.addSuppressed(conflict); + interrupted.addSuppressed(e); + throw interrupted; } - throw e; } - finally + } + + /** Returns the randomized delay before the given attempt is replayed, doubling with each attempt up to a cap. */ + // package private like the PDBStorage copy it is taken from, so that a test can pin the growth and the cap + static long retryDelayMillis(int attempt) + { + final double bound = Math.min(MAX_SLEEP_ON_RETRY_MS, BASE_SLEEP_ON_RETRY_MS * (1 << Math.min(attempt - 1, 5))); + return (long) (Math.random() * bound); + } + + private static Exception unwrap(final StorageRuntimeException e) throws Exception + { + if (e.getCause() != null) { - abort(txn); + throw (Exception) e.getCause(); } + throw e; } private Transaction beginTransaction() diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jeb/JEStorageTest.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jeb/JEStorageTest.java index ef1e762f86..a3494c3da1 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jeb/JEStorageTest.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jeb/JEStorageTest.java @@ -17,13 +17,20 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.failBecauseExceptionWasNotThrown; import static org.forgerock.opendj.config.ConfigurationMock.mockCfg; import static org.forgerock.opendj.ldap.ByteString.valueOfUtf8; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import static org.opends.server.util.CollectionUtils.newTreeSet; import java.io.File; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.forgerock.opendj.config.server.ConfigException; import org.forgerock.opendj.ldap.ByteString; @@ -34,6 +41,7 @@ import org.opends.server.backends.pluggable.spi.AccessMode; import org.opends.server.backends.pluggable.spi.ReadOperation; import org.opends.server.backends.pluggable.spi.ReadableTransaction; +import org.opends.server.backends.pluggable.spi.StorageRuntimeException; import org.opends.server.backends.pluggable.spi.TreeName; import org.opends.server.backends.pluggable.spi.WriteOperation; import org.opends.server.backends.pluggable.spi.WriteableTransaction; @@ -45,9 +53,21 @@ import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; +import com.sleepycat.je.LockConflictException; +import com.sleepycat.je.LockTimeoutException; + /** * Tests what a {@link JEStorage} takes as it opens and gives back when the open fails - the twin - * of the same cases on {@code PDBStorageTest}. + * of the same cases on {@code PDBStorageTest} - and the replay of a {@link JEStorage#write} whose + * transaction JE ends with a {@link LockConflictException}. + *

+ * The conflicts are the engine's own. A deadlock is made by two writers locking two records in opposite order, + * which JE resolves by throwing at a random victim; a conflict on every attempt is made by a transaction which + * keeps a record locked while the storage runs with a {@code je.lock.timeout} - the shipped configuration sets + * none, so a writer waits for a lock rather than times out, but an operator can set one through + * {@code ds-cfg-je-property}, and JE then reports plain contention as a {@link LockTimeoutException}, which it + * documents as "abort and retry" just like the deadlock. A {@code DeadlockException} cannot be built by a test: + * its constructor needs the internal locker it invalidates. */ @SuppressWarnings("javadoc") public class JEStorageTest extends DirectoryServerTestCase @@ -59,6 +79,16 @@ public class JEStorageTest extends DirectoryServerTestCase * what a backend whose directory the server cannot use meets. */ private static final String BLOCKED_DB_DIRECTORY = BACKEND_ID + "-blocked"; + /** A window no run of replays can spend, so that a test of the attempt cap is only ever ended by the cap. */ + private static final long UNREACHABLE_RETRY_WINDOW_NANOS = 300L * 1000L * 1000L * 1000L; //5 min + /** A window a single attempt outlasts, so that a test of the window reaches it without seconds of build time. */ + private static final long SHORT_RETRY_WINDOW_NANOS = 200L * 1000L * 1000L; //200 ms + /** A lock wait shorter than any bound, so that a conflict is reported promptly on every attempt. */ + private static final String SHORT_LOCK_TIMEOUT = "20 ms"; + /** A lock wait longer than {@link #SHORT_RETRY_WINDOW_NANOS}, so that one attempt outlasts the window on its own. */ + private static final String LOCK_TIMEOUT_LONGER_THAN_SHORT_WINDOW = "300 ms"; + /** How long a test waits for a thread it started, in seconds; well past any bound the tests configure. */ + private static final long WAIT_SECONDS = 60; private final TreeName treeName = new TreeName("dc=test", "test"); private ServerContext serverContext; @@ -82,6 +112,7 @@ public void setUp() throws Exception // tearDown(), and this class shares a fixed db-directory across methods and across builds storage.removeStorageFiles(); storage.open(AccessMode.READ_WRITE); + createTreeWithTwoRecords(); } @AfterMethod @@ -197,7 +228,6 @@ public void closingAStorageWhoseOpenFailedTakesNothingMore() throws Exception @Test public void openingAnOpenStorageIsRefusedAndTakesNothing() throws Exception { - createTree(); final MemoryQuota quota = serverContext.getMemoryQuota(); final long availableBefore = quota.getAvailableMemory(); try @@ -237,7 +267,469 @@ private static void unblock(JEStorage blocked) directory.getParentFile().delete(); } - private void createTree() throws Exception + /** + * Replaces the storage under test with one bounded by the given values and, when a lock timeout is given, one + * whose lock waits end in a {@link LockTimeoutException} after that long, the way an operator's + * {@code ds-cfg-je-property: je.lock.timeout=...} makes them end. Bounding the loop stops the two bounds racing + * each other: with the shipped values a run of replays spends a random share of the window on backoff alone, + * so an attempt cap test can be ended by the ten second window instead, and a window test has to make every + * attempt outlast seconds of that window to reach it. + */ + private void reopenWithReplayBounds(int maxRetries, long retryWindowNanos, String lockTimeout) throws Exception + { + closeAndRemove(storage); + storage = new JEStorage(createBackendCfg(lockTimeout), serverContext, maxRetries, retryWindowNanos); + storage.open(AccessMode.READ_WRITE); + createTreeWithTwoRecords(); + } + + /** + * Two writers which lock the two records in opposite order, each holding its first record's write lock before + * asking for the other's. JE detects the cycle and ends one of the two transactions - chosen at random - with a + * {@code DeadlockException}; the other is granted its lock once the victim has aborted. + */ + private final class OpposedWriter implements Runnable + { + private final String name; + private final String first; + private final String second; + private final CyclicBarrier bothHoldTheirFirstLock; + private final boolean swallowTheConflict; + final AtomicInteger attempts = new AtomicInteger(); + final AtomicReference failure = new AtomicReference<>(); + final Thread thread; + + OpposedWriter(String name, String first, String second, CyclicBarrier bothHoldTheirFirstLock, + boolean swallowTheConflict) + { + this.name = name; + this.first = first; + this.second = second; + this.bothHoldTheirFirstLock = bothHoldTheirFirstLock; + this.swallowTheConflict = swallowTheConflict; + this.thread = new Thread(this, name); + } + + @Override + public void run() + { + try + { + storage.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + final int attempt = attempts.incrementAndGet(); + txn.put(treeName, valueOfUtf8(first), valueOfUtf8(name + attempt)); + if (attempt == 1) + { + // only the first attempt meets the other writer there: a replay would wait on the barrier forever + bothHoldTheirFirstLock.await(WAIT_SECONDS, TimeUnit.SECONDS); + } + try + { + txn.put(treeName, valueOfUtf8(second), valueOfUtf8(name + attempt)); + } + catch (StorageRuntimeException e) + { + if (!swallowTheConflict) + { + throw e; + } + // an operation which catches what the transaction raised, the way DN2URI.targetEntryReferrals does + } + } + }); + } + catch (Throwable e) + { + failure.set(e); + } + } + + void startAndJoin(OpposedWriter other) throws InterruptedException + { + thread.start(); + other.thread.start(); + thread.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS)); + other.thread.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS)); + assertThat(thread.isAlive()).as(name + " finished").isFalse(); + assertThat(other.thread.isAlive()).as(other.name + " finished").isFalse(); + } + } + + /** + * Both writers commit, and the victim's transaction was rolled back whole and replayed. How many times is + * JE's to decide: the abort of the victim hands the survivor the lock it waited for, but on the record version + * the abort undoes, so the survivor has to lock the version put back, and a replay which wins that race forms + * the deadlock again with the roles drawn afresh - the backoff makes that rare, not impossible. + */ + private void assertDeadlockVictimReplayedAndBothCommitted(OpposedWriter ab, OpposedWriter ba) throws Exception + { + assertThat(ab.failure.get()).as("ab").isNull(); + assertThat(ba.failure.get()).as("ba").isNull(); + assertThat(ab.attempts.get() + ba.attempts.get()).as("at least one of the two was replayed") + .isGreaterThanOrEqualTo(3); + assertThat(Math.max(ab.attempts.get(), ba.attempts.get())).isLessThanOrEqualTo(JEStorage.MAX_RETRIES); + // whichever committed last wrote both records with the number of the attempt which committed, so the + // records agree - which they would not, had a victim's first record survived its abort + final ByteString a = read("a"); + assertThat(a).isEqualTo(read("b")); + final OpposedWriter last = a.toString().startsWith("ab") ? ab : ba; + assertThat(a).isEqualTo(valueOfUtf8(last.name + last.attempts.get())); + } + + @Test + public void testDeadlockVictimIsReplayed() throws Exception + { + final CyclicBarrier barrier = new CyclicBarrier(2); + final OpposedWriter ab = new OpposedWriter("ab", "a", "b", barrier, false); + final OpposedWriter ba = new OpposedWriter("ba", "b", "a", barrier, false); + + ab.startAndJoin(ba); + + assertDeadlockVictimReplayedAndBothCommitted(ab, ba); + } + + /** + * JE raises a lock conflict from inside the operation, and an operation may catch it there. The transaction is + * then abort-only, and it is {@code commit()} which raises the conflict again - so the loop has to treat a + * conflict raised by the commit as one to replay, not only one raised by the operation. Unlike PersistIt, an + * attempt which swallowed its conflict therefore commits nothing. + */ + @Test + public void testConflictSwallowedInsideTheOperationIsRaisedAgainByCommitAndReplayed() throws Exception + { + final CyclicBarrier barrier = new CyclicBarrier(2); + final OpposedWriter ab = new OpposedWriter("ab", "a", "b", barrier, true); + final OpposedWriter ba = new OpposedWriter("ba", "b", "a", barrier, true); + + ab.startAndJoin(ba); + + assertDeadlockVictimReplayedAndBothCommitted(ab, ba); + } + + /** + * A transaction of its own which keeps a record write-locked until released, so that every attempt of a write + * asking for that record ends the way the storage's lock timeout ends it. + */ + private final class LockHolder implements Runnable + { + private final String key; + private final CountDownLatch held = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private final AtomicReference failure = new AtomicReference<>(); + private final Thread thread = new Thread(this, "lock holder"); + + LockHolder(String key) + { + this.key = key; + } + + @Override + public void run() + { + try + { + storage.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + txn.put(treeName, valueOfUtf8(key), valueOfUtf8("held")); + held.countDown(); + release.await(WAIT_SECONDS, TimeUnit.SECONDS); + } + }); + } + catch (Throwable e) + { + failure.set(e); + } + } + + LockHolder start() throws InterruptedException + { + thread.start(); + assertThat(held.await(WAIT_SECONDS, TimeUnit.SECONDS)).as("the holder took the lock").isTrue(); + return this; + } + + void releaseAndJoin() throws InterruptedException + { + release.countDown(); + thread.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS)); + assertThat(thread.isAlive()).as("the holder finished").isFalse(); + assertThat(failure.get()).as("the holder's own write").isNull(); + } + } + + @Test + public void testWriteGivesUpAfterTheAttemptCap() throws Exception + { + // the message is the same at any cap, so this one is spent in two backoffs rather than in the shipped ladder + final int maxRetries = 3; + reopenWithReplayBounds(maxRetries, UNREACHABLE_RETRY_WINDOW_NANOS, SHORT_LOCK_TIMEOUT); + final LockHolder holder = new LockHolder("a").start(); + try + { + final AtomicInteger attempts = new AtomicInteger(); + try + { + storage.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + attempts.incrementAndGet(); + txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("abandoned")); + } + }); + failBecauseExceptionWasNotThrown(StorageRuntimeException.class); + } + catch (StorageRuntimeException e) + { + assertThat(e.getMessage()).contains("JEStorageTest").contains(maxRetries + " attempts"); + // and which of the two bounds ran out, since the attempt count alone does not say + assertThat(e.getMessage()).contains("attempt cap"); + // write() unwraps a StorageRuntimeException that carries a cause, which would replace this message with + // the bare conflict, and it is the message the config change paths report + assertThat(e.getCause()).isNull(); + assertThat(e.getSuppressed()).hasSize(1); + assertThat(e.getSuppressed()[0]).isInstanceOf(LockTimeoutException.class); + } + assertThat(attempts.get()).isEqualTo(maxRetries); + } + finally + { + holder.releaseAndJoin(); + } + assertThat(read("a")).isEqualTo(valueOfUtf8("held")); + } + + @Test + public void testWriteIsReplayedUntilTheConflictClears() throws Exception + { + reopenWithReplayBounds(JEStorage.MAX_RETRIES, UNREACHABLE_RETRY_WINDOW_NANOS, SHORT_LOCK_TIMEOUT); + final LockHolder holder = new LockHolder("a").start(); + try + { + final AtomicInteger attempts = new AtomicInteger(); + storage.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + if (attempts.incrementAndGet() == 4) + { + // released, and committed, before the lock is asked for, so that this attempt is the one which + // gets it rather than the one which times out on the holder's commit + holder.releaseAndJoin(); + } + txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("applied")); + } + }); + assertThat(attempts.get()).isEqualTo(4); + } + finally + { + holder.releaseAndJoin(); + } + assertThat(read("a")).isEqualTo(valueOfUtf8("applied")); + } + + /** + * With a lock timeout longer than the window, a single attempt outlasts the whole window. Giving up on the + * window alone would then replay nothing, in the very case where the replay is likeliest to succeed: the + * transaction that was blocking this one has just finished. + */ + @Test + public void testWriteIsReplayedOnceWhenTheFirstAttemptOutlastsTheWindow() throws Exception + { + reopenWithReplayBounds(JEStorage.MAX_RETRIES, SHORT_RETRY_WINDOW_NANOS, LOCK_TIMEOUT_LONGER_THAN_SHORT_WINDOW); + final LockHolder holder = new LockHolder("a").start(); + try + { + final AtomicInteger attempts = new AtomicInteger(); + storage.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + if (attempts.incrementAndGet() == 2) + { + holder.releaseAndJoin(); + } + txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("outlasted")); + } + }); + assertThat(attempts.get()).isEqualTo(2); + } + finally + { + holder.releaseAndJoin(); + } + assertThat(read("a")).isEqualTo(valueOfUtf8("outlasted")); + } + + @Test + public void testWriteGivesUpOnTheWindowWhenAttemptsAreSlow() throws Exception + { + reopenWithReplayBounds(JEStorage.MAX_RETRIES, SHORT_RETRY_WINDOW_NANOS, LOCK_TIMEOUT_LONGER_THAN_SHORT_WINDOW); + final LockHolder holder = new LockHolder("a").start(); + try + { + final AtomicInteger attempts = new AtomicInteger(); + try + { + storage.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + attempts.incrementAndGet(); + // a conflict this slow to report spends the wall clock window long before the attempt cap + txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("abandoned")); + } + }); + failBecauseExceptionWasNotThrown(StorageRuntimeException.class); + } + catch (StorageRuntimeException e) + { + // the window is what ended it, and it says so: an assertion on the attempt count alone would also pass + // for a give up on attempt 1, which is the regression the attempt > 1 exemption exists to prevent + assertThat(e.getMessage()).contains("retry window"); + } + // one attempt beyond the first: the first spends the window, the exemption grants the replay, and the + // check after that replay is the one that gives up + assertThat(attempts.get()).isEqualTo(2); + } + finally + { + holder.releaseAndJoin(); + } + assertThat(read("a")).isEqualTo(valueOfUtf8("held")); + } + + /** + * An interrupt reaches the loop in its backoff sleep alone, and it must not reach JE afterwards: a thread + * carrying the interrupt flag invalidates the whole environment on its next call - the transaction registry's + * latch is acquired interruptibly - which is why the loop leaves the flag as the sleep cleared it, and reports + * the interrupt with the conflict instead. Delivering the interrupt to the sleep alone takes a write which makes + * no call of JE at all while the flag is set: this one runs on the storage's import environment, whose writes + * open no transaction, with an operation which raises a conflict JE raised earlier rather than asking JE for a + * new one - a transaction aborted with the flag set would take the environment down before the sleep is reached. + */ + @Test + public void testInterruptedWriteReportsTheConflictItWasReplaying() throws Exception + { + final LockConflictException conflict = aConflictOfJEsOwn(); + // the storage that raised it gave up at its first attempt; this one is bounded as shipped + closeAndRemove(storage); + storage = new JEStorage(createBackendCfg(), serverContext); + storage.startImport(); + + final AtomicInteger attempts = new AtomicInteger(); + final boolean interruptedAfterwards; + try + { + storage.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + attempts.incrementAndGet(); + Thread.currentThread().interrupt(); + throw new StorageRuntimeException(conflict); + } + }); + failBecauseExceptionWasNotThrown(StorageRuntimeException.class); + return; + } + catch (StorageRuntimeException e) + { + interruptedAfterwards = Thread.interrupted(); + // the conflict, not the interrupt, is what the caller is told about - but through the same shape the + // exhausted loop uses, since a bare conflict reaches every caller as its own class name + assertThat(e.getMessage()).contains("JEStorageTest").contains("interrupted"); + assertThat(e.getSuppressed()).contains(conflict).hasAtLeastOneElementOfType(InterruptedException.class); + assertThat(e.getCause()).isNull(); + } + finally + { + Thread.interrupted(); + } + // the flag the sleep cleared stays clear: restored, it would end the environment on the caller's next call + assertThat(interruptedAfterwards).as("interrupt flag after the write").isFalse(); + // one attempt even though the first backoff is a random 0-49 ms and so is sometimes 0: Thread.sleep() checks + // the interrupt flag before it checks for a zero duration, so the replay is never reached + assertThat(attempts.get()).isEqualTo(1); + } + + /** A conflict raised by JE itself: the lock timeout a write gave up on at its first attempt. */ + private LockConflictException aConflictOfJEsOwn() throws Exception + { + reopenWithReplayBounds(1, UNREACHABLE_RETRY_WINDOW_NANOS, SHORT_LOCK_TIMEOUT); + final LockHolder holder = new LockHolder("a").start(); + try + { + storage.write(new WriteOperation() + { + @Override + public void run(WriteableTransaction txn) throws Exception + { + txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("abandoned")); + } + }); + throw new AssertionError("the write was applied although its record was held"); + } + catch (StorageRuntimeException e) + { + assertThat(e.getSuppressed()).hasSize(1); + return (LockConflictException) e.getSuppressed()[0]; + } + finally + { + holder.releaseAndJoin(); + } + } + + /** + * The delay grows with the attempt and stays under the cap, so that a contention the first delays did not + * outlast still has a chance to clear without the replays overrunning the window on sleep alone. + */ + @Test + public void testRetryDelayGrowsAndStaysBounded() + { + long previousBound = 0; + for (int attempt = 1; attempt <= JEStorage.MAX_RETRIES; attempt++) + { + long bound = 0; + for (int i = 0; i < 100; i++) + { + final long delay = JEStorage.retryDelayMillis(attempt); + assertThat(delay).as("attempt %d", attempt).isGreaterThanOrEqualTo(0).isLessThan(1000); + bound = Math.max(bound, delay); + } + if (attempt == 1) + { + assertThat(bound).as("attempt 1 delays past the first tier").isLessThan(50); + } + assertThat(bound).as("attempt %d did not grow past attempt %d", attempt, attempt - 1) + .isGreaterThanOrEqualTo(previousBound / 2); + previousBound = bound; + } + // and the growth is real rather than a delay that never leaves the first tier + long grown = 0; + for (int i = 0; i < 100; i++) + { + grown = Math.max(grown, JEStorage.retryDelayMillis(JEStorage.MAX_RETRIES)); + } + assertThat(grown).as("the last attempts still sleep within the first attempt's bound").isGreaterThan(500); + } + + private void createTreeWithTwoRecords() throws Exception { storage.write(new WriteOperation() { @@ -245,6 +737,8 @@ private void createTree() throws Exception public void run(WriteableTransaction txn) throws Exception { txn.openTree(treeName, true); + txn.put(treeName, valueOfUtf8("a"), valueOfUtf8("0")); + txn.put(treeName, valueOfUtf8("b"), valueOfUtf8("0")); } }); } @@ -274,4 +768,18 @@ private static JEBackendCfg createBackendCfg() when(backendCfg.getDBNumLockTables()).thenReturn(63); return backendCfg; } + + /** + * The configuration of the storage under test, with the lock timeout an operator would set through + * {@code ds-cfg-je-property}, or the shipped one - none - when null. + */ + private static JEBackendCfg createBackendCfg(String lockTimeout) + { + final JEBackendCfg backendCfg = createBackendCfg(); + if (lockTimeout != null) + { + when(backendCfg.getJEProperty()).thenReturn(newTreeSet("je.lock.timeout=" + lockTimeout)); + } + return backendCfg; + } }