Merge lp:~pbeaman/akiban-persistit/fix-1032701-interrupts-leave-latches into lp:akiban-persistit

[Peter Beaman]

Fixes bug 1032701 - Interrupt causes Thread to exit without releasing claims. This bug was caused by an interrupt occurring at a dangerous place in the code. Specifically, in the BufferPool#get(...) method, the try/finally around the actual reading of a page from disk contained a call to BufferPool#invalidate(Buffer). In the event the read failed (in the tested case, due to an interrupt) and then a second interrupt occurred in the invalidate method, the subsequent call to release() never happened.

This branch fixes this by ensuring that the invalidate(Buffer) method does not throw an interrupt. Because this is asserted by a change in the signature of the invalidate(Buffer) and because injecting timing conditions to test this is a bit awkward, there is no unit test for this. Also, there is no proof that no other code paths can cause a similar condition. However, I carefully reviewed all the finally{} blocks that could have the same effect, and also have verified that the test which originally detected the failure no longer fails.

Note that the invalidate() method polls the detach(Buffer) method, which is necessary to avoid deadlock. That it does so is not a change; the only change is that this loop is no longer interruptible.

Also changed are a couple of other items noticed while debugging this:

If a commit() fails due to an I/O failure or an interrupt, the intended action is to abort the transaction. That was being done correctly, but the _rollbackPending flag was not set, causing a spurious warning from the end() method such as:

[TransactionThread_344] WARNING Transaction neither committed nor rolled back Transaction_100000347 depth=1 status=<ts=3,480,654 tc=ABORTED mvv=1> owner=TransactionThread_344

(That the TransactionStatus is ABORTED is inconsistent with the warning.) The _rollbackPending and _rollbackCompleted flags are now managemed more correctly.

Also, an attempt to begin a nested transaction within an aborted transaction context now correctly issues an IllegalStateException. New unit tests verify this.

[Nathan Williams (nwilliams)]

Explanation and change make sense. Other fixes are good, too.

review: Approve

=== modified file 'src/main/java/com/persistit/BufferPool.java'
--- src/main/java/com/persistit/BufferPool.java	2012-08-02 04:45:28 +0000
+++ src/main/java/com/persistit/BufferPool.java	2012-08-04 19:33:20 +0000
@@ -613,11 +613,22 @@
         return result;
     }
 
-    private void invalidate(Buffer buffer) throws PersistitInterruptedException {
+    private void invalidate(Buffer buffer) {
         Debug.$assert0.t(buffer.isValid() && buffer.isMine());
 
         while (!detach(buffer)) {
-            Util.spinSleep();
+            //
+            // Spin until detach succeeds. Note: this method must not throw an Exception
+            // because it is called in at at critical time when cleanup must be done.
+            // It is not possible to lock the hash bucket here due to possible deadlock.
+            // However, the likelihood of a lengthy live-lock is infinitesimal so polling
+            // is acceptable.
+            //
+        	try {
+        		Thread.sleep(1);
+        	} catch (InterruptedException ie) {
+        		// ignore
+        	}
         }
         buffer.clearValid();
         buffer.clearDirty();

=== modified file 'src/main/java/com/persistit/Transaction.java'
--- src/main/java/com/persistit/Transaction.java	2012-08-02 04:45:28 +0000
+++ src/main/java/com/persistit/Transaction.java	2012-08-04 19:33:20 +0000
@@ -586,11 +586,15 @@
         if (_commitCompleted) {
             throw new IllegalStateException("Attempt to begin a committed transaction " + this);
         }
+        if (_rollbackPending) {
+            throw new IllegalStateException("Attempt to begin a transaction with pending rollback" + this);
+        }
         if (_nestedDepth == 0) {
             flushTransactionBuffer(false);
             try {
                 _transactionStatus = _persistit.getTransactionIndex().registerTransaction();
             } catch (InterruptedException e) {
+            	_rollbackCompleted = true;
                 throw new PersistitInterruptedException(e);
             }
             _rollbackPending = false;
@@ -609,11 +613,15 @@
         if (_commitCompleted) {
             throw new IllegalStateException("Attempt to begin a committed transaction " + this);
         }
+        if (_rollbackPending) {
+            throw new IllegalStateException("Attmpet to begin a transaction with pending rollback" + this);
+        }
         if (_nestedDepth == 0) {
             flushTransactionBuffer(false);
             try {
                 _transactionStatus = _persistit.getTransactionIndex().registerCheckpointTransaction();
             } catch (InterruptedException e) {
+            	_rollbackCompleted = true;
                 throw new PersistitInterruptedException(e);
             }
             _rollbackPending = false;
@@ -854,10 +862,11 @@
                         policy == CommitPolicy.GROUP ? _persistit.getTransactionCommitStallTime() : 0);
                 committed = true;
             } finally {
+            	
                 _persistit.getTransactionIndex().notifyCompleted(_transactionStatus,
                         committed ? _commitTimestamp : TransactionStatus.ABORTED);
                 _commitCompleted = committed;
-                _rollbackCompleted = !committed;
+                _rollbackPending = _rollbackCompleted = !committed;
             }
         }
     }

=== modified file 'src/main/java/com/persistit/TransactionIndex.java'
--- src/main/java/com/persistit/TransactionIndex.java	2012-08-02 04:45:28 +0000
+++ src/main/java/com/persistit/TransactionIndex.java	2012-08-04 19:33:20 +0000
@@ -1073,8 +1073,9 @@
                     }
                     bucket.checkpointAccumulatorSnapshots(timestamp);
                     for (final Accumulator accumulator : accumulators) {
-                        accumulator.setCheckpointValueAndTimestamp(accumulator.applyValue(accumulator
-                                .getCheckpointValue(), accumulator.getCheckpointTemp()), timestamp);
+                        accumulator.setCheckpointValueAndTimestamp(
+                                accumulator.applyValue(accumulator.getCheckpointValue(),
+                                        accumulator.getCheckpointTemp()), timestamp);
                     }
                 } catch (RetryException e) {
                     again = true;
@@ -1086,11 +1087,12 @@
     }
 
     /**
-     * Create and return a brand new delta associated with the given
-     * status. Note that it is completely uninitialized and always
-     * allocated from the bucket.
-     *
-     * @param status Status to add the delta to.
+     * Create and return a brand new delta associated with the given status.
+     * Note that it is completely uninitialized and always allocated from the
+     * bucket.
+     * 
+     * @param status
+     *            Status to add the delta to.
      * @return The new Delta.
      */
     Delta addDelta(final TransactionStatus status) {
@@ -1107,18 +1109,22 @@
     }
 
     /**
-     * Create, or combine, new delta information for the given status.
-     * This method attempts to find a compatible delta
-     * (see {@link Delta#canMerge(Accumulator, int)}) to combine with
-     * before allocating a new one. If one is not found,
-     * {@link #addDelta(TransactionStatus)} is called and initialized
-     * before returning.
-     *
-     * @param status Status to add, or combine, delta to.
-     * @param accumulator Accumulator being modified.
-     * @param step Step value of modification.
-     * @param value The value to add or combine.
-     *
+     * Create, or combine, new delta information for the given status. This
+     * method attempts to find a compatible delta (see
+     * {@link Delta#canMerge(Accumulator, int)}) to combine with before
+     * allocating a new one. If one is not found,
+     * {@link #addDelta(TransactionStatus)} is called and initialized before
+     * returning.
+     * 
+     * @param status
+     *            Status to add, or combine, delta to.
+     * @param accumulator
+     *            Accumulator being modified.
+     * @param step
+     *            Step value of modification.
+     * @param value
+     *            The value to add or combine.
+     * 
      * @return Delta that was created or modified.
      */
     Delta addOrCombineDelta(TransactionStatus status, Accumulator accumulator, int step, long value) {

=== modified file 'src/test/java/com/persistit/TransactionTest2.java'
--- src/test/java/com/persistit/TransactionTest2.java	2012-08-02 04:45:28 +0000
+++ src/test/java/com/persistit/TransactionTest2.java	2012-08-04 19:33:20 +0000
@@ -48,7 +48,7 @@
 
     final static CommitPolicy policy = CommitPolicy.SOFT;
 
-    final static long TIMEOUT = 20000; // seconds
+    final static long TIMEOUT = 20000; // 20 seconds
 
     static int _threadCount = 8;
     static int _iterationsPerThread = 25000;
@@ -222,6 +222,20 @@
         assertTrue("ATC has very old transaction",
                 ti.getActiveTransactionCeiling() - ti.getActiveTransactionFloor() < 10000);
     }
+    
+    @Test(expected = IllegalStateException.class)
+    public void illegalStateExceptionOnRollback() throws Exception {
+        Transaction txn = _persistit.getTransaction();
+        txn.rollback();
+        txn.begin();
+    }
+
+    @Test(expected = IllegalStateException.class)
+    public void illegalStateExceptionOnCommit() throws Exception {
+        Transaction txn = _persistit.getTransaction();
+        txn.commit();
+        txn.begin();
+    }
 
     public void runIt() {
         try {