diff --git a/persistit/core/src/main/java/com/persistit/JournalManager.java b/persistit/core/src/main/java/com/persistit/JournalManager.java index 0a8874405..99b7f01a1 100644 --- a/persistit/core/src/main/java/com/persistit/JournalManager.java +++ b/persistit/core/src/main/java/com/persistit/JournalManager.java @@ -61,6 +61,7 @@ import com.persistit.Persistit.FatalErrorException; import com.persistit.TransactionPlayer.TransactionPlayerListener; import com.persistit.exception.CorruptJournalException; +import com.persistit.exception.PersistitClosedException; import com.persistit.exception.PersistitException; import com.persistit.exception.PersistitIOException; import com.persistit.exception.PersistitInterruptedException; @@ -1849,6 +1850,8 @@ void pruneObsoleteTransactions(final boolean rollbackPruningEnabled) { * complete the I/O when requested. In particular, a value of * zero indicates the I/O should start immediately. * @throws PersistitInterruptedException if the thread is interrupted while waiting + * @throws PersistitClosedException if Persistit was closed or crashed before + * durability could be confirmed */ void waitForDurability(final long flushedTimestamp, final long leadTime, final long stallTime) @@ -1857,7 +1860,7 @@ void waitForDurability(final long flushedTimestamp, final long leadTime, final l if (flusher != null) { flusher.waitForDurability(flushedTimestamp, leadTime, stallTime); } else { - throw new IllegalStateException("JOURNAL_FLUSHER is not running"); + throw new PersistitClosedException("JOURNAL_FLUSHER is not running"); } } @@ -2349,6 +2352,21 @@ private void waitForDurability(final long flushedTimestamp, final long leadTime, break; } + if (isStopped()) { + /* + * The flusher publishes its final _startTimestamp and + * _endTimestamp before stopping, so one re-read after + * observing the stop is exact: either the final flush + * cycle covered flushedTimestamp, or no future cycle ever + * will and polling would spin forever against a stopped + * thread. + */ + if (_endTimestamp > flushedTimestamp && _startTimestamp > flushedTimestamp) { + break; + } + throw new PersistitClosedException("JOURNAL_FLUSHER stopped before durability was confirmed"); + } + long remainingSleepNanos; if (estimatedRemainingIoNanos == -1) { remainingSleepNanos = Math.max(0, _flushInterval - (now - endTime)); diff --git a/persistit/core/src/test/java/com/persistit/JournalManagerTest.java b/persistit/core/src/test/java/com/persistit/JournalManagerTest.java index 0350e9578..3d23220bb 100644 --- a/persistit/core/src/test/java/com/persistit/JournalManagerTest.java +++ b/persistit/core/src/test/java/com/persistit/JournalManagerTest.java @@ -22,6 +22,7 @@ import com.persistit.CheckpointManager.Checkpoint; import com.persistit.JournalManager.PageNode; import com.persistit.TransactionPlayer.TransactionPlayerListener; +import com.persistit.exception.PersistitClosedException; import com.persistit.exception.PersistitException; import com.persistit.unit.ConcurrentUtil.ThrowingRunnable; import com.persistit.util.Util; @@ -678,6 +679,50 @@ public void run() throws Exception { disableSequencer(); } + @Test + public void waitForDurabilityAfterCloseThrowsPersistitClosedException() throws Exception { + final JournalManager jman = _persistit.getJournalManager(); + final long flushedTimestamp = _persistit.getTimestampAllocator().updateTimestamp(); + _persistit.close(); + try { + jman.waitForDurability(flushedTimestamp, 0, 0); + fail("Expected PersistitClosedException"); + } catch (final PersistitClosedException expected) { + /* + * A commit racing Persistit.close() must observe the documented + * closed-state exception, not a raw IllegalStateException that + * bypasses callers' shutdown handling (issue #304). + */ + } + } + + @Test + public void waitForDurabilityOnStoppedFlusherThrowsInsteadOfSpinning() throws Exception { + final JournalManager jman = _persistit.getJournalManager(); + _persistit.crash(); + /* + * crash() stops JOURNAL_FLUSHER without clearing the JournalManager's + * reference to it, and a timestamp allocated after the flusher's final + * cycle can never be covered by its durability marks - the shape of + * the shutdown race in which a committing thread captured the flusher + * reference just before it was cleared. Without the stopped-flusher + * check this wait polls forever. + */ + final long flushedTimestamp = _persistit.getTimestampAllocator().updateTimestamp(); + final Thread waiter = createThread("WAIT_FOR_DURABILITY", new ThrowingRunnable() { + @Override + public void run() throws Exception { + try { + jman.waitForDurability(flushedTimestamp, 0, 0); + fail("Expected PersistitClosedException"); + } catch (final PersistitClosedException expected) { + // expected + } + } + }); + startAndJoinAssertSuccess(10000, waiter); + } + private int countKeys(final boolean mvcc) throws PersistitException { final Exchange exchange = _persistit.getExchange(_volumeName, "JournalManagerTest1", false); exchange.ignoreMVCCFetch(!mvcc);