Merge lp:~pbeaman/akiban-persistit/fix-hard-commit-infinite-loop into lp:akiban-persistit

[Peter Beaman]

Fixes bug 1036422 - CommitPolicy HARD can soak CPU.

Modify logic in JournalManager.JournalFlusher#waitForDurability to make sure no loop gets executed without at least a short (1ms) sleep interval. Add logic to detect a failure to do that and a unit test to demonstrate the problem and that it is fixed.

[Nathan Williams (nwilliams)]

The vast majority of this diff is formatting changes. Haven't commented on that in a while, so it seemed appropriate :)

Otherwise, looks good -- simple and tested.

review: Approve

=== modified file 'src/main/java/com/persistit/JournalManager.java'
--- src/main/java/com/persistit/JournalManager.java	2012-08-02 14:19:26 +0000
+++ src/main/java/com/persistit/JournalManager.java	2012-08-14 14:22:24 +0000
@@ -212,6 +212,8 @@
 
     private boolean _allowHandlesForTempVolumesAndTrees;
 
+    private AtomicLong _waitLoopsWithNoDelay = new AtomicLong();
+
     /**
      * <p>
      * Initialize the new journal. This method takes its information from the
@@ -648,8 +650,8 @@
         synchronized (this) {
             if (address >= _writeBufferAddress && address + length <= _currentAddress) {
                 assert _writeBufferAddress + _writeBuffer.position() == _currentAddress : String.format(
-                        "writeBufferAddress=%,d position=%,d currentAddress=%,d", _writeBufferAddress, _writeBuffer
-                                .position(), _currentAddress);
+                        "writeBufferAddress=%,d position=%,d currentAddress=%,d", _writeBufferAddress,
+                        _writeBuffer.position(), _currentAddress);
                 final int wbPosition = _writeBuffer.position();
                 final int wbLimit = _writeBuffer.limit();
                 _writeBuffer.position((int) (address - _writeBufferAddress));
@@ -948,8 +950,8 @@
         advance(TM.OVERHEAD);
         int offset = 0;
         for (final TransactionMapItem ts : _liveTransactionMap.values()) {
-            TM.putEntry(_writeBuffer, offset / TM.ENTRY_SIZE, ts.getStartTimestamp(), ts.getCommitTimestamp(), ts
-                    .getStartAddress(), ts.getLastRecordAddress());
+            TM.putEntry(_writeBuffer, offset / TM.ENTRY_SIZE, ts.getStartTimestamp(), ts.getCommitTimestamp(),
+                    ts.getStartAddress(), ts.getLastRecordAddress());
             offset += TM.ENTRY_SIZE;
             count--;
             if (count == 0 || offset + TM.ENTRY_SIZE >= _writeBuffer.remaining()) {
@@ -1303,8 +1305,8 @@
         if (address != Long.MAX_VALUE && _writeBuffer != null) {
 
             assert _writeBufferAddress + _writeBuffer.position() == _currentAddress : String.format(
-                    "writeBufferAddress=%,d position=%,d currentAddress=%,d", _writeBufferAddress, _writeBuffer
-                            .position(), _currentAddress);
+                    "writeBufferAddress=%,d position=%,d currentAddress=%,d", _writeBufferAddress,
+                    _writeBuffer.position(), _currentAddress);
 
             try {
                 if (_writeBuffer.position() > 0) {
@@ -1360,8 +1362,8 @@
                     }
 
                     assert _writeBufferAddress + _writeBuffer.position() == _currentAddress : String.format(
-                            "writeBufferAddress=%,d position=%,d currentAddress=%,d", _writeBufferAddress, _writeBuffer
-                                    .position(), _currentAddress);
+                            "writeBufferAddress=%,d position=%,d currentAddress=%,d", _writeBufferAddress,
+                            _writeBuffer.position(), _currentAddress);
 
                     _persistit.getIOMeter().chargeFlushJournal(written, address);
                     return _writeBufferAddress;
@@ -2220,6 +2222,7 @@
                 }
 
                 long remainingSleepNanos;
+                boolean didWait = false;
                 if (estimatedRemainingIoNanos == -1) {
                     remainingSleepNanos = Math.max(0, _flushInterval - (now - endTime));
                 } else {
@@ -2247,22 +2250,33 @@
                     long delay = stallTime * NS_PER_MS - estimatedNanosToFinish;
                     if (delay > 0) {
                         Util.sleep(delay / NS_PER_MS);
+                        didWait = true;
                     }
                     kick();
+                    if (delay <= 0) {
+                        didWait = true;
+                        Util.spinSleep();
+                    }
                 } else {
                     /*
                      * Otherwise, wait until the I/O is about half done and then
                      * retry.
                      */
-                    long delay = (estimatedNanosToFinish - leadTime * NS_PER_MS) / 2;
+                    long delay = ((estimatedNanosToFinish - leadTime * NS_PER_MS) / 2) + NS_PER_MS;
                     try {
-                        if (delay > 0 && _lock.readLock().tryLock(delay, TimeUnit.NANOSECONDS)) {
-                            _lock.readLock().unlock();
+                        if (delay > 0) {
+                            didWait = true;
+                            if (_lock.readLock().tryLock(delay, TimeUnit.NANOSECONDS)) {
+                                _lock.readLock().unlock();
+                            }
                         }
                     } catch (InterruptedException e) {
                         throw new PersistitInterruptedException(e);
                     }
                 }
+                if (!didWait) {
+                    _waitLoopsWithNoDelay.incrementAndGet();
+                }
             }
             if (_lastExceptionTimestamp > timestamp) {
                 final Exception e = _lastException;
@@ -2406,9 +2420,10 @@
                 }
                 pageAddress = readPageBufferFromJournal(stablePageNode, bb);
             } catch (PersistitException ioe) {
-                _persistit.getAlertMonitor().post(
-                        new Event(AlertLevel.ERROR, _persistit.getLogBase().copyException, ioe, volume, pageNode
-                                .getPageAddress(), pageNode.getJournalAddress()), AlertMonitor.JOURNAL_CATEGORY);
+                _persistit
+                        .getAlertMonitor()
+                        .post(new Event(AlertLevel.ERROR, _persistit.getLogBase().copyException, ioe, volume,
+                                pageNode.getPageAddress(), pageNode.getJournalAddress()), AlertMonitor.JOURNAL_CATEGORY);
                 throw ioe;
             }
 
@@ -2454,8 +2469,8 @@
             }
             if (volume == null) {
                 _persistit.getAlertMonitor().post(
-                        new Event(AlertLevel.WARN, _persistit.getLogBase().missingVolume, volumeRef, pageNode
-                                .getJournalAddress()), AlertMonitor.MISSING_VOLUME_CATEGORY);
+                        new Event(AlertLevel.WARN, _persistit.getLogBase().missingVolume, volumeRef,
+                                pageNode.getJournalAddress()), AlertMonitor.MISSING_VOLUME_CATEGORY);
                 if (_ignoreMissingVolume.get()) {
                     _persistit.getLogBase().lostPageFromMissingVolume.log(pageNode.getPageAddress(), volumeRef,
                             pageNode.getJournalAddress());
@@ -2483,8 +2498,8 @@
             try {
                 volume.getStorage().writePage(bb, pageAddress);
             } catch (PersistitException ioe) {
-                _persistit.getLogBase().copyException.log(ioe, volume, pageNode.getPageAddress(), pageNode
-                        .getJournalAddress());
+                _persistit.getLogBase().copyException.log(ioe, volume, pageNode.getPageAddress(),
+                        pageNode.getJournalAddress());
                 throw ioe;
             }
 
@@ -2771,8 +2786,8 @@
              */
             if (ts != null) {
                 if (ts.getMvvCount() > 0 && _persistit.isInitialized()) {
-                    _persistit.getLogBase().pruningIncomplete.log(ts, TransactionPlayer.addressToString(address,
-                            timestamp));
+                    _persistit.getLogBase().pruningIncomplete.log(ts,
+                            TransactionPlayer.addressToString(address, timestamp));
                 }
             }
         }
@@ -2905,4 +2920,8 @@
     public SortedMap<Integer, TreeDescriptor> queryTreeMap() {
         return new TreeMap<Integer, TreeDescriptor>(_handleToTreeMap);
     }
+
+    long getWaitLoopsWithNoDelay() {
+        return _waitLoopsWithNoDelay.get();
+    }
 }
 
=== modified file 'src/test/java/com/persistit/JournalManagerTest.java'
--- src/test/java/com/persistit/JournalManagerTest.java	2012-08-02 14:19:26 +0000
+++ src/test/java/com/persistit/JournalManagerTest.java	2012-08-14 14:22:24 +0000
@@ -43,6 +43,7 @@
 
 import com.persistit.CheckpointManager.Checkpoint;
 import com.persistit.JournalManager.PageNode;
+import com.persistit.Transaction.CommitPolicy;
 import com.persistit.TransactionPlayer.TransactionPlayerListener;
 import com.persistit.exception.PersistitException;
 import com.persistit.unit.ConcurrentUtil.ThrowingRunnable;
@@ -531,6 +532,24 @@
         assertTrue("Should have failed updates during recovery", _persistit.getRecoveryManager().getPlayer()
                 .getIgnoredUpdates() > 0);
     }
+    
+    @Test
+    public void waitForDurabilitySoaksCPU() throws Exception {
+        _persistit.setDefaultTransactionCommitPolicy(CommitPolicy.HARD);
+        final JournalManager jman = _persistit.getJournalManager();
+        long waitLoopsWithoutDelay  = jman.getWaitLoopsWithNoDelay();
+        final Exchange exchange = _persistit.getExchange("persistit", "JournalManagerTest", true);
+        final Transaction txn = exchange.getTransaction();
+        for (int count = 0; count < 100; count++) {
+            txn.begin();
+            exchange.getValue().put(RED_FOX + count);
+            exchange.to(count).store();
+            txn.commit();
+            txn.end();
+        }
+        waitLoopsWithoutDelay = jman.getWaitLoopsWithNoDelay() - waitLoopsWithoutDelay;
+        assertEquals("Wait loops without delay", 0, waitLoopsWithoutDelay);
+    }
 
     private List<PageNode> testCleanupPageListSource(final int size) {
         final List<PageNode> source = new ArrayList<PageNode>(size);