Merge lp:~nwilliams/akiban-persistit/handle-abandoned-txn-status into lp:akiban-persistit

[Nathan Williams]

Fix closing of transactions from threads that have been abandoned.

If a thread is abandoned with an open transaction the wwLock member of TransactionStatus will be in a locked state. See associated bug for more details.

This isn't easily recoverable as the thread holding the lock is now gone. The only reason it needs to be unlocked is for re-use, mark it as abandoned and throw it away when it is no longer needed.

Minor cleanup and helper usage to eliminate duplicate code.

Includes simple test case and one requiring MVV cleanup.

[Peter Beaman (pbeaman)]

From cell phone, so it's hard to Approve. But it looks fine to me so please
go ahead. Alternatively I was thinking about replacing the lock object and
allowing the ts to be recycled. Either way is fine with me. Good detective
work!
On Feb 20, 2013 12:38 PM, "Nathan Williams" <email address hidden> wrote:

> Nathan Williams has proposed merging
> lp:~nwilliams/akiban-persistit/handle-abandoned-txn-status into
> lp:akiban-persistit.
>
> Requested reviews:
> Akiban Technologies (akiban-technologies)
> Related bugs:
> Bug #1126297 in Akiban Persistit: "Assertion failure in
> TransactionIndexBucket#allocateTransactionStatus"
> https://bugs.launchpad.net/akiban-persistit/+bug/1126297
>
> For more details, see:
>
> https://code.launchpad.net/~nwilliams/akiban-persistit/handle-abandoned-txn-status/+merge/149643
>
> Fix closing of transactions from threads that have been abandoned.
>
> If a thread is abandoned with an open transaction the wwLock member of
> TransactionStatus will be in a locked state. See associated bug for more
> details.
>
> This isn't easily recoverable as the thread holding the lock is now gone.
> The only reason it needs to be unlocked is for re-use, mark it as abandoned
> and throw it away when it is no longer needed.
>
> Minor cleanup and helper usage to eliminate duplicate code.
>
> Includes simple test case and one requiring MVV cleanup.
> --
>
> https://code.launchpad.net/~nwilliams/akiban-persistit/handle-abandoned-txn-status/+merge/149643
> Your team Akiban Technologies is requested to review the proposed merge of
> lp:~nwilliams/akiban-persistit/handle-abandoned-txn-status into
> lp:akiban-persistit.
>
> === modified file 'src/main/java/com/persistit/Persistit.java'
> --- src/main/java/com/persistit/Persistit.java 2013-02-15 15:39:42 +0000
> +++ src/main/java/com/persistit/Persistit.java 2013-02-20 17:37:24 +0000
> @@ -1496,26 +1496,12 @@
> return _bufferPoolTable;
> }
>
> + /**
> + * Remove any sessions that have expired and close transactions
> associated
> + * with them. Also flush statistics for all known volumes.
> + */
> void cleanup() {
> - final Set<SessionId> sessionIds;
> - synchronized (_transactionSessionMap) {
> - sessionIds = new
> HashSet<SessionId>(_transactionSessionMap.keySet());
> - }
> - for (final SessionId sessionId : sessionIds) {
> - if (!sessionId.isAlive()) {
> - Transaction transaction = null;
> - synchronized (_transactionSessionMap) {
> - transaction =
> _transactionSessionMap.remove(sessionId);
> - }
> - if (transaction != null) {
> - try {
> - transaction.close();
> - } catch (final PersistitException e) {
> - _logBase.exception.log(e);
> - }
> - }
> - }
> - }
> + closeZombieTransactions();
> final List<Volume> volumes;
> synchronized (this) {
> volumes = new ArrayList<Volume>(_volumes);
>
> === modified file 'src/main/java/com/persistit/Transaction.jav...

[Nathan Williams (nwilliams)]

I considered that, but the implications regarding other transactions were daunting (e.g. what if someone was waiting for ww-dependency check). That, coupled with the abandonment being rare (i.e. loss of status not a big deal) lead me to this solution.

I'm happy to discuss it further. I don't think there is a huge rush, either, as the customer has a workaround.

[Peter Beaman (pbeaman)]

Looks good. Approving.

[Akiban Build User (build-akiban)]

There were 2 failures during build/test:

* job persistit-build failed at build number 544: http://172.16.20.104:8080/job/persistit-build/544/

* view must-pass failed: persistit-build is aborted

review: Needs Fixing

[Nathan Williams (nwilliams)]

Unrelated.

[Akiban Build User (build-akiban)]

There were 2 failures during build/test:

* job persistit-build failed at build number 546: http://172.16.20.104:8080/job/persistit-build/546/

* view must-pass failed: persistit-build is aborted

review: Needs Fixing

[lp:~nwilliams/akiban-persistit/handle-abandoned-txn-status]

428. By Nathan Williams on 2013-02-26

Add parameter to refactored helper

[Nathan Williams (nwilliams)]

The closeZombieTransactions() helper actually removed all sessions from the map, which of course we don't want to do from cleanup. Add a boolean parameter so both usages can be served.

review: Needs Resubmitting

[Peter Beaman (pbeaman)]

Oops. I didn't notice before that this proposal was still pending. Looks good - Approving.

=== modified file 'src/main/java/com/persistit/Persistit.java'
--- src/main/java/com/persistit/Persistit.java	2013-02-15 15:39:42 +0000
+++ src/main/java/com/persistit/Persistit.java	2013-02-26 17:25:24 +0000
@@ -1496,26 +1496,12 @@
         return _bufferPoolTable;
     }
 
+    /**
+     * Remove any sessions that have expired and close transactions associated
+     * with them. Also flush statistics for all known volumes.
+     */
     void cleanup() {
-        final Set<SessionId> sessionIds;
-        synchronized (_transactionSessionMap) {
-            sessionIds = new HashSet<SessionId>(_transactionSessionMap.keySet());
-        }
-        for (final SessionId sessionId : sessionIds) {
-            if (!sessionId.isAlive()) {
-                Transaction transaction = null;
-                synchronized (_transactionSessionMap) {
-                    transaction = _transactionSessionMap.remove(sessionId);
-                }
-                if (transaction != null) {
-                    try {
-                        transaction.close();
-                    } catch (final PersistitException e) {
-                        _logBase.exception.log(e);
-                    }
-                }
-            }
-        }
+        closeZombieTransactions(false);
         final List<Volume> volumes;
         synchronized (this) {
             volumes = new ArrayList<Volume>(_volumes);
@@ -1676,7 +1662,7 @@
             waitForIOTaskStop(task);
 
             interruptActiveThreads(SHORT_DELAY);
-            closeZombieTransactions();
+            closeZombieTransactions(true);
 
             for (final Volume volume : volumes) {
                 volume.close();
@@ -1695,17 +1681,17 @@
         releaseAllResources();
     }
 
-    private void closeZombieTransactions() {
+    private void closeZombieTransactions(boolean removeAllSessions) {
         final Set<SessionId> sessionIds;
         synchronized (_transactionSessionMap) {
             sessionIds = new HashSet<SessionId>(_transactionSessionMap.keySet());
         }
         for (final SessionId sessionId : sessionIds) {
-            Transaction transaction = null;
-            synchronized (_transactionSessionMap) {
-                transaction = _transactionSessionMap.remove(sessionId);
-            }
-            if (!sessionId.isAlive()) {
+            if (!sessionId.isAlive() || removeAllSessions) {
+                Transaction transaction = null;
+                synchronized (_transactionSessionMap) {
+                    transaction = _transactionSessionMap.remove(sessionId);
+                }
                 if (transaction != null) {
                     try {
                         transaction.close();
 
=== modified file 'src/main/java/com/persistit/Transaction.java'
--- src/main/java/com/persistit/Transaction.java	2013-02-01 11:46:13 +0000
+++ src/main/java/com/persistit/Transaction.java	2013-02-26 17:25:24 +0000
@@ -508,6 +508,7 @@
         if (_nestedDepth > 0 && !_commitCompleted && !_rollbackCompleted) {
             final TransactionStatus ts = _transactionStatus;
             if (ts != null && ts.getTs() == _startTimestamp && !_commitCompleted && !_rollbackCompleted) {
+                _transactionStatus.markAbandoned();
                 rollback();
                 _persistit.getLogBase().txnAbandoned.log(this);
             }
 
=== modified file 'src/main/java/com/persistit/TransactionIndexBucket.java'
--- src/main/java/com/persistit/TransactionIndexBucket.java	2013-02-18 13:57:21 +0000
+++ src/main/java/com/persistit/TransactionIndexBucket.java	2013-02-26 17:25:24 +0000
@@ -258,8 +258,7 @@
         if (ts >= getFloor()) {
             for (TransactionStatus s = getCurrent(); s != null; s = s.getNext()) {
                 if (s == status) {
-                    s.complete(timestamp);
-                    status.wwUnlock();
+                    status.completeAndUnlock(timestamp);
                     if (s.getTs() == getFloor() || hasFloorMoved()) {
                         reduce();
                     }
@@ -272,8 +271,7 @@
                 if (s == status) {
                     final TransactionStatus next = s.getNext();
                     assert s.getTc() != UNCOMMITTED;
-                    s.complete(timestamp);
-                    status.wwUnlock();
+                    status.completeAndUnlock(timestamp);
                     boolean moved = false;
                     if (s.getTc() == ABORTED) {
                         aggregate(s, false);
@@ -610,7 +608,7 @@
          * May be held by another thread briefly while status being checked
          */
         assert !status.isHeldByCurrentThread();
-        if (_freeCount < _transactionIndex.getMaxFreeListSize()) {
+        if (_freeCount < _transactionIndex.getMaxFreeListSize() && !status.isAbandoned()) {
             status.setNext(_free);
             _free = status;
             _freeCount++;
 
=== modified file 'src/main/java/com/persistit/TransactionStatus.java'
--- src/main/java/com/persistit/TransactionStatus.java	2013-02-18 13:57:21 +0000
+++ src/main/java/com/persistit/TransactionStatus.java	2013-02-26 17:25:24 +0000
@@ -124,6 +124,12 @@
      */
     private volatile boolean _notified;
 
+    /**
+     * Indicates whether the owning transaction had been abandoned. This status
+     * may be in a locked state and is not available for re-use.
+     */
+    private volatile boolean _abandoned;
+
     TransactionStatus(final TransactionIndexBucket bucket) {
         _bucket = bucket;
     }
@@ -244,6 +250,13 @@
         _notified = true;
     }
 
+    void completeAndUnlock(final long timestamp) {
+        complete(timestamp);
+        if (!isAbandoned()) {
+            wwUnlock();
+        }
+    }
+
     boolean isLocked() {
         return _wwLock.isLocked();
     }
@@ -374,6 +387,7 @@
      * Release the lock acquired by {@link #wwLock(long)}.
      */
     void wwUnlock() {
+        assert !isAbandoned() : "Attempt to unlock abandoned: " + this;
         _wwLock.unlock();
     }
 
@@ -409,6 +423,20 @@
         _notified = true;
     }
 
+    /**
+     * Make this status as abandoned. This will prevent its unlock and re-use.
+     */
+    void markAbandoned() {
+        _abandoned = true;
+    }
+
+    /**
+     * @return <code>true</code> if this status has been abandoned.
+     */
+    boolean isAbandoned() {
+        return _abandoned;
+    }
+
     @Override
     public String toString() {
         return String.format("<ts=%,d tc=%s mvv=%,d>", _ts, tcString(_tc), _mvvCount.get());
 
=== added file 'src/test/java/com/persistit/TransactionAbandonedTest.java'
--- src/test/java/com/persistit/TransactionAbandonedTest.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/persistit/TransactionAbandonedTest.java	2013-02-26 17:25:24 +0000
@@ -0,0 +1,130 @@
+/**
+ * Copyright © 2005-2013 Akiban Technologies, Inc.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available
+ * under the terms of the Eclipse Public License v1.0 which
+ * accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * This program may also be available under different license terms.
+ * For more information, see www.akiban.com or contact [e-mail].
+ *
+ * Contributors:
+ * Akiban Technologies, Inc.
+ */
+
+package com.persistit;
+
+import com.persistit.exception.PersistitException;
+import com.persistit.unit.ConcurrentUtil;
+import com.persistit.unit.UnitTestProperties;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * <p>
+ * Inspired by bug1126297: Assertion failure in
+ * TransactionIndexBucket#allocateTransactionStatus
+ * </p>
+ * <p>
+ * The symptom was the bug (a locked TransactionStatus on the free list) but the
+ * cause was mishandling of abandoned transactions from the
+ * {@link Persistit#cleanup()} method.
+ * </p>
+ * <p>
+ * When attempting to rollback the abandoned transaction, the status was
+ * notified and then unlocked. Since the lock was held by a now dead thread, an
+ * IllegalMonitorStateException occurred. It was then put on the free list
+ * during the next cleanup of that bucket since it had been notified.
+ * </p>
+ */
+public class TransactionAbandonedTest extends PersistitUnitTestCase {
+    private static final String TREE = TransactionAbandonedTest.class.getSimpleName();
+    private static final int KEY_START = 1;
+    private static final int KEY_RANGE = 10;
+    private static final long MAX_TIMEOUT_MS = 10 * 1000;
+
+    private static class TxnAbandoner extends ConcurrentUtil.ThrowingRunnable {
+        private final Persistit persistit;
+        private final boolean doRead;
+        private final boolean doWrite;
+
+        public TxnAbandoner(Persistit persistit, boolean doRead, boolean doWrite) {
+            this.persistit = persistit;
+            this.doRead = doRead;
+            this.doWrite = doWrite;
+        }
+
+        @Override
+        public void run() throws PersistitException {
+            Transaction txn = persistit.getTransaction();
+            txn.begin();
+            if (doRead) {
+                assertEquals("Traverse count", KEY_RANGE, scanAndCount(getExchange(persistit)));
+            }
+            if (doWrite) {
+                loadData(persistit, KEY_START + KEY_RANGE, KEY_RANGE);
+            }
+        }
+    }
+
+    private static Exchange getExchange(Persistit persistit) throws PersistitException {
+        return persistit.getExchange(UnitTestProperties.VOLUME_NAME, TREE, true);
+    }
+
+    private static void loadData(Persistit persistit, int keyOffset, int count) throws PersistitException {
+        Exchange ex = getExchange(persistit);
+        for (int i = 0; i < count; ++i) {
+            ex.clear().append(keyOffset + i).store();
+        }
+    }
+
+    private static int scanAndCount(Exchange ex) throws PersistitException {
+        ex.clear().append(Key.BEFORE);
+        int saw = 0;
+        while (ex.next()) {
+            ++saw;
+        }
+        return saw;
+    }
+
+    @Before
+    public void disableAndLoad() throws PersistitException {
+        disableBackgroundCleanup();
+        loadData(_persistit, KEY_START, KEY_RANGE);
+    }
+
+    private void runAndCleanup(String name, boolean doRead, boolean doWrite) {
+        Thread t = ConcurrentUtil.createThread(name, new TxnAbandoner(_persistit, false, false));
+        ConcurrentUtil.startAndJoinAssertSuccess(MAX_TIMEOUT_MS, t);
+        // Threw exception before fix
+        _persistit.cleanup();
+    }
+
+    @Test
+    public void noReadsOrWrites() {
+        runAndCleanup("NoReadNoWrite", false, false);
+    }
+
+    @Test
+    public void readOnly() throws PersistitException {
+        runAndCleanup("ReadOnly", true, false);
+        assertEquals("Traversed after abandoned", KEY_RANGE, scanAndCount(getExchange(_persistit)));
+    }
+
+    @Test
+    public void readAndWrite() throws Exception {
+        runAndCleanup("ReadAndWrite", true, true);
+        assertEquals("Traversed after abandoned", KEY_RANGE, scanAndCount(getExchange(_persistit)));
+        // Check that the abandoned was pruned
+        CleanupManager cm = _persistit.getCleanupManager();
+        for (int i = 0; i < 5 && cm.getEnqueuedCount() > 0; ++i) {
+            cm.runTask();
+        }
+        Exchange rawEx = getExchange(_persistit);
+        rawEx.ignoreMVCCFetch(true);
+        assertEquals("Raw traversed after abandoned", KEY_RANGE, scanAndCount(rawEx));
+    }
+}