Merge lp:~pbeaman/akiban-persistit/fix-1174352-commit-thread into lp:akiban-persistit

[Peter Beaman]

Use Semaphore instead of ReentrantLock in TransactionStatus#wwLock. This allows Thread #1 to register the transaction and Thread #2 to commit it. (ReentrantLock throws an IllegalMonitorStateException if a different thread attempts to unlock it.)

This is important to support the documented behavior allowing SessionId management, and that is important to allow a fixed number of threads in a thread pool to handled a potentially much larger number of sessions.

[Nathan Williams (nwilliams)]

Semaphore has availablePermits() which we could use for the removed assert and short sleep, if desired.

This is fine by me as-is, too.

[lp:~pbeaman/akiban-persistit/fix-1174352-commit-thread]

435. By Peter Beaman on 2013-05-01

Reinstate TransactionStatus.isLocked and an assert that uses it.

[Peter Beaman (pbeaman)]

Reinstated TransactionStatus#isLocked and an assert that uses it.

[Nathan Williams (nwilliams)]

Looks good.

review: Approve

=== modified file 'src/main/java/com/persistit/Transaction.java'
--- src/main/java/com/persistit/Transaction.java	2013-05-01 15:05:31 +0000
+++ src/main/java/com/persistit/Transaction.java	2013-05-01 15:05:32 +0000
@@ -517,7 +517,6 @@
         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-05-01 15:05:31 +0000
+++ src/main/java/com/persistit/TransactionIndexBucket.java	2013-05-01 15:05:32 +0000
@@ -605,11 +605,8 @@
 
     private void free(final TransactionStatus status) {
         assert _lock.isHeldByCurrentThread();
-        /*
-         * May be held by another thread briefly while status being checked
-         */
-        assert !status.isHeldByCurrentThread();
-        if (_freeCount < _transactionIndex.getMaxFreeListSize() && !status.isAbandoned()) {
+
+        if (_freeCount < _transactionIndex.getMaxFreeListSize()) {
             status.setNext(_free);
             _free = status;
             _freeCount++;

=== modified file 'src/main/java/com/persistit/TransactionStatus.java'
--- src/main/java/com/persistit/TransactionStatus.java	2013-05-01 15:05:31 +0000
+++ src/main/java/com/persistit/TransactionStatus.java	2013-05-01 15:05:32 +0000
@@ -16,9 +16,9 @@
 
 package com.persistit;
 
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.ReentrantLock;
 
 import com.persistit.Accumulator.Delta;
 
@@ -96,18 +96,20 @@
     private AtomicInteger _mvvCount = new AtomicInteger();
 
     /**
-     * Lock used to manage ww dependencies. An attempt to update an MVV that
-     * already contains a value version from a concurrently executing
-     * transaction must wait for that other transaction to commit or abort.
+     * Semaphore used to manage ww dependencies. An attempt to update an MVV
+     * that already contains a value version from a concurrently executing
+     * transaction must wait for that other transaction to commit or abort. The
+     * protocol uses a Semaphore rather than a ReentrantLock because it may be
+     * acquired in one thread but released in another.
      */
-    private final ReentrantLock _wwLock = new ReentrantLock(true);
+    private final Semaphore _wwLock = new Semaphore(1);
     /**
      * Pointer to next member of singly-linked list.
      */
     private TransactionStatus _next;
 
     /**
-     * Pointer to TransactionStatus on which we intend to claim a lock. (For
+     * Pointer to TransactionStatus on which we intend to claim a permit. (For
      * deadlock detection.)
      */
     private volatile TransactionStatus _depends;
@@ -125,12 +127,6 @@
      */
     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;
     }
@@ -253,17 +249,7 @@
 
     void completeAndUnlock(final long timestamp) {
         complete(timestamp);
-        if (!isAbandoned()) {
-            wwUnlock();
-        }
-    }
-
-    boolean isLocked() {
-        return _wwLock.isLocked();
-    }
-
-    boolean isHeldByCurrentThread() {
-        return _wwLock.isHeldByCurrentThread();
+        wwUnlock();
     }
 
     Delta getDelta() {
@@ -363,16 +349,15 @@
 
     /**
      * <p>
-     * Acquire a lock on this TransactionStatus. This supports the
-     * {@link TransactionIndex#wwDependency(long, long, long)} method. While a
-     * transaction is running this lock is in a locked state. The lock is
-     * acquired when the transaction is registered (see
+     * Acquire a permit on this TransactionStatus. This supports the
+     * {@link TransactionIndex#wwDependency(long, long, long)} method. The
+     * permit is acquired when the transaction is registered (see
      * {@link TransactionIndex#registerTransaction(Transaction)} and released
      * once the transaction is either committed or aborted.
      * </p>
      * <p>
      * The <code>wwDependency</code> method also attempts to acquire, and then
-     * immediately release this lock. This stalls the thread calling
+     * immediately release this permit. This stalls the thread calling
      * wwDependency until the commit/abort status of the current transaction is
      * known.
      * 
@@ -381,15 +366,24 @@
      * @throws InterruptedException
      */
     boolean wwLock(final long timeout) throws InterruptedException {
-        return _wwLock.tryLock(timeout, TimeUnit.MILLISECONDS);
+        return _wwLock.tryAcquire(timeout, TimeUnit.MILLISECONDS);
     }
 
     /**
-     * Release the lock acquired by {@link #wwLock(long)}.
+     * Release the permit acquired by {@link #wwLock(long)}.
      */
     void wwUnlock() {
-        assert !isAbandoned() : "Attempt to unlock abandoned: " + this;
-        _wwLock.unlock();
+        _wwLock.release();
+    }
+
+    /**
+     * Indicate whether this TransactionStatus has been locked. Tested by assert
+     * statements in various places.
+     * 
+     * @return true if a thread has acquired a claim on this TransactionStatus.
+     */
+    boolean isLocked() {
+        return _wwLock.availablePermits() == 0;
     }
 
     /**
@@ -424,20 +418,6 @@
         _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/TransactionSessionSwitchTest.java'
--- src/test/java/com/persistit/TransactionSessionSwitchTest.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/persistit/TransactionSessionSwitchTest.java	2013-05-01 15:05:32 +0000
@@ -0,0 +1,122 @@
+/**
+ * Copyright 2005-2013 Akiban Technologies, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.persistit;
+
+import static org.junit.Assert.*;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import com.persistit.exception.PersistitException;
+import com.persistit.unit.ConcurrentUtil;
+
+/**
+ * <p>
+ * Bug https://bugs.launchpad.net/akiban-persistit/+bug/1174352
+ * </p>
+ * <p>
+ * Documentation for com.persistit.Transaction says that with careful management
+ * of ownership of a SessionId, it is possible to complete a transaction in a
+ * different thread than the one that began it. However, this is not the case:
+ * the second thread receives an IllegalMonitorStateException when attempting to
+ * commit or abort the transaction.
+ * </p>
+ * <p>
+ * The issue is that the TransactionStatus object used to represent transaction
+ * state within the TransactionIndex uses a
+ * java.util.concurrent.locks.ReentrantLock to represent its in-use state, and
+ * only the thread that locked it may unlock it. *
+ * </p>
+ * <p>
+ */
+public class TransactionSessionSwitchTest extends PersistitUnitTestCase {
+
+    private final static int SESSIONS = 100;
+    private final static int STEPS = 197;
+    private final static int THREADS = 17;
+    private final static long TIMEOUT = 10000;
+
+    private Queue<SessionId> sessionQueue = new ArrayBlockingQueue<SessionId>(SESSIONS);
+    private Map<SessionId, AtomicInteger> sessionState = new HashMap<SessionId, AtomicInteger>();
+
+    @Test
+    public void sessionManagement() throws Exception {
+        for (int i = 0; i < SESSIONS; i++) {
+            final SessionId sessionId = new SessionId();
+            sessionQueue.add(sessionId);
+            sessionState.put(sessionId, new AtomicInteger(0));
+        }
+
+        Thread[] threads = new Thread[THREADS];
+
+        final Tree tree = _persistit.getVolume("persistit").getTree("tt", true);
+        for (int i = 0; i < THREADS; i++) {
+            threads[i] = new Thread(new Runnable() {
+                public void run() {
+                    SessionId session;
+                    while ((session = sessionQueue.poll()) != null) {
+                        int state = sessionState.get(session).get();
+                        try {
+                            _persistit.setSessionId(session);
+                            final Transaction txn = _persistit.getTransaction();
+                            if (state == 0) {
+                                txn.begin();
+                            } else if (state > STEPS) {
+                                if ((session.hashCode() % 3) == 0) {
+                                    txn.rollback();
+                                } else {
+                                    txn.commit();
+                                }
+                                txn.end();
+                            } else {
+                                Exchange ex = _persistit.getExchange(tree.getVolume(), tree.getName(), false);
+                                ex.getValue().put(Thread.currentThread().getName());
+                                ex.clear().append(session.hashCode()).append(state).store();
+                                _persistit.releaseExchange(ex);
+                            }
+                        } catch (PersistitException e) {
+                            throw new RuntimeException(e);
+                        } finally {
+                            if (state <= STEPS) {
+                                sessionState.get(session).incrementAndGet();
+                                sessionQueue.add(session);
+                            }
+                        }
+                    }
+                }
+            });
+
+        }
+        ConcurrentUtil.startAndJoinAssertSuccess(TIMEOUT, threads);
+
+        final Exchange ex = _persistit.getExchange(tree.getVolume(), tree.getName(), false);
+        for (SessionId session : sessionState.keySet()) {
+            int count = 0;
+            ex.clear().append(session.hashCode()).append(Key.BEFORE);
+            while (ex.next()) {
+                count++;
+            }
+            final int expected = (session.hashCode() % 3) == 0 ? 0 : STEPS;
+            assertEquals("Mismatched count", expected, count);
+        }
+    }
+}

=== modified file 'src/test/java/com/persistit/stress/unit/AccumulatorRestart.java'
--- src/test/java/com/persistit/stress/unit/AccumulatorRestart.java	2013-05-01 15:05:31 +0000
+++ src/test/java/com/persistit/stress/unit/AccumulatorRestart.java	2013-05-01 15:05:32 +0000
@@ -108,7 +108,7 @@
                         min.minimum(bsum(minValue, r));
                         max.maximum(bsum(maxValue, r));
                         seq.allocate();
-                        sum.add(1);
+                        sum.add(r);
                         seqValue++;
                         final long minWas = getLong(_ex.to("min"), Long.MAX_VALUE);
                         _ex.getValue().put(Math.min(bsum(minValue, r), minWas));