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

[Peter Beaman]

This is a better fix for bug 1036422 CommitPolicy HARD can soak CPU. The previous version delayed transaction commit unnecessarily and resulted in slower update and delete atomics. This code results in a significant improvement. A separate report will detail the benchmark differences.

The key elements in this proposal are (a) don't call waitForDurability() at all in a read-only transaction, and (b) use the Lock in waitDurability rather than a fixed-length sleep to wake up immediately after an I/O that completes in less than 1ms.

[Nathan Williams (nwilliams)]

Makes sense.

review: Approve

=== modified file 'src/main/java/com/persistit/JournalManager.java'
--- src/main/java/com/persistit/JournalManager.java	2012-08-14 14:16:00 +0000
+++ src/main/java/com/persistit/JournalManager.java	2012-08-15 00:37:23 +0000
@@ -2255,7 +2255,13 @@
                     kick();
                     if (delay <= 0) {
                         didWait = true;
-                        Util.spinSleep();
+                        try {
+                            if (_lock.readLock().tryLock(NS_PER_MS, TimeUnit.NANOSECONDS)) {
+                                _lock.readLock().unlock();
+                            }
+                        } catch (InterruptedException e) {
+                            throw new PersistitInterruptedException(e);
+                        }
                     }
                 } else {
                     /*
 
=== modified file 'src/main/java/com/persistit/Transaction.java'
--- src/main/java/com/persistit/Transaction.java	2012-08-04 19:19:44 +0000
+++ src/main/java/com/persistit/Transaction.java	2012-08-15 00:37:23 +0000
@@ -460,7 +460,7 @@
          * higher than with the HARD policy.
          */
         GROUP;
-        
+
         static CommitPolicy forName(final String policyName) {
             for (CommitPolicy policy : values()) {
                 if (policy.name().equalsIgnoreCase(policyName)) {
@@ -594,7 +594,7 @@
             try {
                 _transactionStatus = _persistit.getTransactionIndex().registerTransaction();
             } catch (InterruptedException e) {
-            	_rollbackCompleted = true;
+                _rollbackCompleted = true;
                 throw new PersistitInterruptedException(e);
             }
             _rollbackPending = false;
@@ -621,7 +621,7 @@
             try {
                 _transactionStatus = _persistit.getTransactionIndex().registerCheckpointTransaction();
             } catch (InterruptedException e) {
-            	_rollbackCompleted = true;
+                _rollbackCompleted = true;
                 throw new PersistitInterruptedException(e);
             }
             _rollbackPending = false;
@@ -856,13 +856,14 @@
             sequence(COMMIT_FLUSH_C);
             boolean committed = false;
             try {
-                flushTransactionBuffer(false);
-                _persistit.getJournalManager().waitForDurability(
-                        policy == CommitPolicy.SOFT ? _persistit.getTransactionCommitLeadTime() : 0,
-                        policy == CommitPolicy.GROUP ? _persistit.getTransactionCommitStallTime() : 0);
+                if (flushTransactionBuffer(false)) {
+                    _persistit.getJournalManager().waitForDurability(
+                            policy == CommitPolicy.SOFT ? _persistit.getTransactionCommitLeadTime() : 0,
+                            policy == CommitPolicy.GROUP ? _persistit.getTransactionCommitStallTime() : 0);
+                }
                 committed = true;
             } finally {
-            	
+
                 _persistit.getTransactionIndex().notifyCompleted(_transactionStatus,
                         committed ? _commitTimestamp : TransactionStatus.ABORTED);
                 _commitCompleted = committed;
@@ -1143,17 +1144,20 @@
         }
     }
 
-    synchronized void flushTransactionBuffer(final boolean chain) throws PersistitException {
+    synchronized boolean flushTransactionBuffer(final boolean chain) throws PersistitException {
+        boolean didWrite = false;
         if (_buffer.position() > 0 || _previousJournalAddress != 0) {
             long previousJournalAddress = _persistit.getJournalManager().writeTransactionToJournal(_buffer,
                     _startTimestamp, _commitTimestamp, _previousJournalAddress);
             _buffer.clear();
+            didWrite = true;
             if (chain) {
                 _previousJournalAddress = previousJournalAddress;
             } else {
                 _previousJournalAddress = 0;
             }
         }
+        return didWrite;
     }
 
     synchronized void flushOnCheckpoint(final long timestamp) throws PersistitException {
@@ -1293,7 +1297,7 @@
                     + MAXIMUM_STEP);
         }
     }
-    
+
     private int treeHandle(final Tree tree) {
         final int treeHandle = tree.getHandle();
         assert treeHandle != 0 : "Undefined tree handle in " + tree;
 
=== modified file 'src/test/java/com/persistit/JournalManagerTest.java'
--- src/test/java/com/persistit/JournalManagerTest.java	2012-08-13 22:24:48 +0000
+++ src/test/java/com/persistit/JournalManagerTest.java	2012-08-15 00:37:23 +0000
@@ -540,7 +540,7 @@
         long waitLoopsWithoutDelay  = jman.getWaitLoopsWithNoDelay();
         final Exchange exchange = _persistit.getExchange("persistit", "JournalManagerTest", true);
         final Transaction txn = exchange.getTransaction();
-        for (int count = 0; count < 100; count++) {
+        for (int count = 0; count < 1000; count++) {
             txn.begin();
             exchange.getValue().put(RED_FOX + count);
             exchange.to(count).store();
 
=== modified file 'src/test/java/com/persistit/stress/unit/CommitBench.java'
--- src/test/java/com/persistit/stress/unit/CommitBench.java	2012-08-02 04:45:28 +0000
+++ src/test/java/com/persistit/stress/unit/CommitBench.java	2012-08-15 00:37:23 +0000
@@ -45,8 +45,8 @@
 
     private final int RECORDS = 200000;
     private final int RECORDS_PER_TXN = 10;
-    private final String[] ARG_TEMPLATE = new String[] { "threads|int:1:1000|Number of threads",
-            "duration|int:10:86400|Duration of test in seconds",
+    private final String[] ARG_TEMPLATE = new String[] { "threads|int:1:1:1000|Number of threads",
+            "duration|int:10:10:86400|Duration of test in seconds",
             "policy|String:HARD|Commit policy: SOFT, HARD or GROUP", };
 
     volatile long stopTime;
@@ -59,7 +59,7 @@
     }
 
     public void bench(final String[] args) throws Exception {
-        final ArgParser ap = new ArgParser("CommitBench", args, ARG_TEMPLATE);
+        final ArgParser ap = new ArgParser("CommitBench", args, ARG_TEMPLATE).strict();
         int threadCount = ap.getIntValue("threads");
         int duration = ap.getIntValue("duration");
         String policy = ap.getStringValue("policy");