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.java'
> --- src/main/java/com/persistit/Transaction.java        2013-02-01
> 11:46:13 +0000
> +++ src/main/java/com/persistit/Transaction.java        2013-02-20
> 17:37: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-20
> 17:37: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-20
> 17:37: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-20
> 17:37: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
> <email address hidden>.
> + *
> + * 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));
> +    }
> +}
>
>
>