Merge lp:~pbeaman/akiban-persistit/add-lock into lp:akiban-persistit

[Peter Beaman]

Supports a new method Exchange#lock. This method causes a write-write conflict when two concurrent transactions attempt to lock the same key. This could be done by an application storing a value under the same key, but the lock method is more efficient because it uses a temporary tree that seldom causes disk I/O, and because the value written is an AntiValue which causes default pruning to remove the key as soon as it is no longer needed.

To support this we now have a designated temporary volume, the lock volume, maintained by Persistit. This is created lazily on the first call to Persistit#getLockVolume(). So that various operations including pruning work correctly, the lock volume is given a reserved volume handle (Integer.MAX_VALUE) and its trees are managed by the JournalManager's treeHandleMap even though no entires for its tries are written to the journal.

Also in this update are changes in documentation and a new set of tests.

[lp:~pbeaman/akiban-persistit/add-lock]

413. By Peter Beaman on 2013-01-19

Fix up javadoc, make Exchange#setTimoutMillis public, and add test

414. By Peter Beaman on 2013-01-19

Fix up javadoc, make Exchange#setTimoutMillis public, and add test

[Peter Beaman (pbeaman)]

Ran into some build issues with this in Java 7. Specifically, javadoc is a little fussier (needs @return tags to have text).

[lp:~pbeaman/akiban-persistit/add-lock]

415. By Peter Beaman on 2013-01-21

Fix ExchangeTest unit test

416. By Peter Beaman on 2013-01-21

zero timeout means default

417. By Peter Beaman on 2013-01-21

Tweak a variable name

418. By Peter Beaman on 2013-01-22

Merge from trunk. Merge from java7-issues

419. By Peter Beaman on 2013-01-22

Rev version because server should depend on new lock API

420. By Peter Beaman on 2013-01-24

Merge from trunk and modify for review comments

421. By Peter Beaman on 2013-01-24

Clarify JavaDoc

422. By Peter Beaman on 2013-01-24

Source code cleanup

423. By Peter Beaman on 2013-01-25

Add extra prune cycle in ExchangeLockTest#lockTablePruning

424. By Peter Beaman on 2013-01-25

Explicit call to pruneObsoleteTransactions in ExchangeLockTest

425. By Peter Beaman on 2013-01-25

Merge 3.2.6-upgrade

426. By Peter Beaman on 2013-01-25

Source code cleanup

427. By Peter Beaman on 2013-01-25

Another attempt at fixing ExchangeLockTest

Unmerged revisions

=== modified file 'doc/Transactions.rst'
--- doc/Transactions.rst	2012-06-18 17:49:04 +0000
+++ doc/Transactions.rst	2013-01-22 16:32:21 +0000
@@ -116,6 +116,8 @@
 
 Note that many common transaction patterns, including those defined by the TPC-C benchmark, do not experience write-skew and therefore *are* serializable under SI.
 
+In the unusual case where transactions are susceptible to write skew, the Exchange#lock method offers a way for applications to create a write-write dependency explicitly. This mechanism provides an efficient mechanism for ensuring serializable behavior.
+
 Durability Options: ``CommitPolicy``
 ------------------------------------
 

=== modified file 'src/main/java/com/persistit/Accumulator.java'
--- src/main/java/com/persistit/Accumulator.java	2012-10-11 20:37:43 +0000
+++ src/main/java/com/persistit/Accumulator.java	2013-01-22 16:32:21 +0000
@@ -426,7 +426,7 @@
      * 
      * @param a
      * @param b
-     * @return
+     * @return the result of the commutative operation on a and b
      */
     abstract long applyValue(long a, long b);
 
@@ -439,7 +439,7 @@
      * 
      * @param a
      * @param b
-     * @return
+     * @return the result of the commutative operation on a and b
      */
     abstract long updateValue(long a, long b);
 

=== modified file 'src/main/java/com/persistit/AlertMonitor.java'
--- src/main/java/com/persistit/AlertMonitor.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/AlertMonitor.java	2013-01-22 16:32:21 +0000
@@ -794,7 +794,7 @@
      * </p>
      * 
      * @param event
-     * @return
+     * @return formatted event
      */
     private String format(final Event event) {
         return event == null ? "null" : event.toString();

=== modified file 'src/main/java/com/persistit/Buffer.java'
--- src/main/java/com/persistit/Buffer.java	2012-11-26 17:19:45 +0000
+++ src/main/java/com/persistit/Buffer.java	2013-01-22 16:32:21 +0000
@@ -1219,7 +1219,7 @@
      * @param key
      * @param mode
      * @param foundAt
-     * @return
+     * @return foundAt value
      */
     int traverse(final Key key, final Key.Direction mode, final int foundAt) {
         final boolean exactMatch = (foundAt & EXACT_MASK) > 0;
@@ -1239,7 +1239,7 @@
      * 
      * @param key
      * @param foundAt
-     * @return
+     * @return foundAt value
      */
     int previousKey(final Key key, final int foundAt) {
         int p = (foundAt & P_MASK) - KEYBLOCK_LENGTH;
@@ -1312,7 +1312,7 @@
      * 
      * @param key
      * @param foundAt
-     * @return
+     * @return foundAt value
      */
     int nextKey(final Key key, final int foundAt) {
         int p = foundAt & P_MASK;
@@ -1347,7 +1347,7 @@
      * 
      * @param value
      * @param foundAt
-     * @return
+     * @return foundAt value
      */
     int nextLongRecord(final Value value, final int foundAt) {
         Debug.$assert0.t(isDataPage());
@@ -1568,7 +1568,7 @@
      * been inserted or removed.
      * 
      * @param p
-     * @return
+     * @return when key is adjacent
      */
     private boolean adjacentKeyCheck(int p) {
         p &= P_MASK;
@@ -3595,7 +3595,8 @@
      * 
      * @param tree
      * @param spareKey
-     * @return
+     * @return <code>true</code> if this method changed the contents of the
+     *         buffer
      * @throws PersistitException
      */
     boolean pruneMvvValues(final Tree tree, final boolean pruneLongMVVs) throws PersistitException {
@@ -3814,7 +3815,7 @@
         System.arraycopy(bytes, offset, value.getEncodedBytes(), 0, oldSize);
         value.setEncodedSize(oldSize);
         final LongRecordHelper helper = new LongRecordHelper(_persistit, _vol);
-        helper.fetchLongRecord(value, Integer.MAX_VALUE);
+        helper.fetchLongRecord(value, Integer.MAX_VALUE, SharedResource.DEFAULT_MAX_WAIT_TIME);
         final byte[] rawBytes = value.getEncodedBytes();
         final int oldLongSize = value.getEncodedSize();
         // TODO - perhaps remove. Done as a precaution for now.

=== modified file 'src/main/java/com/persistit/BufferPool.java'
--- src/main/java/com/persistit/BufferPool.java	2013-01-02 22:41:20 +0000
+++ src/main/java/com/persistit/BufferPool.java	2013-01-22 16:32:21 +0000
@@ -697,6 +697,30 @@
     /**
      * Find or load a page given its Volume and address. The returned page has a
      * reader or a writer lock, depending on whether the writer parameter is
+     * true on entry. Waits up to {@value SharedResource#DEFAULT_MAX_WAIT_TIME}
+     * milliseconds to acquire the desired lock on the page
+     * 
+     * @param vol
+     *            The Volume
+     * @param page
+     *            The address of the page
+     * @param writer
+     *            <i>true</i> if a write lock is required.
+     * @param wantRead
+     *            <i>true</i> if the caller wants the page read from disk.
+     *            <i>false</i> to allocate a new blank page.)
+     * @return Buffer The Buffer describing the buffer containing the page.
+     * @throws InUseException
+     *             if the specific lock could not be acquired
+     */
+    Buffer get(final Volume vol, final long page, final boolean writer, final boolean wantRead)
+            throws PersistitException {
+        return get(vol, page, writer, wantRead, SharedResource.DEFAULT_MAX_WAIT_TIME);
+    }
+
+    /**
+     * Find or load a page given its Volume and address. The returned page has a
+     * reader or a writer lock, depending on whether the writer parameter is
      * true on entry.
      * 
      * @param vol
@@ -708,10 +732,15 @@
      * @param wantRead
      *            <i>true</i> if the caller wants the page read from disk.
      *            <i>false</i> to allocate a new blank page.)
+     * @param timeout
+     *            maximum time to wait for the page to become available before
+     *            throwing an InUseException
      * @return Buffer The Buffer describing the buffer containing the page.
      * @throws InUseException
+     *             if the specific lock could not be acquired within the
+     *             specified timeout
      */
-    Buffer get(final Volume vol, final long page, final boolean writer, final boolean wantRead)
+    Buffer get(final Volume vol, final long page, final boolean writer, final boolean wantRead, final long timeout)
             throws PersistitException {
         final int hash = hashIndex(vol, page);
         Buffer buffer = null;
@@ -776,7 +805,7 @@
             if (mustClaim) {
                 boolean claimed = false;
                 boolean same = true;
-                final long expires = System.currentTimeMillis() + SharedResource.DEFAULT_MAX_WAIT_TIME;
+                final long expires = System.currentTimeMillis() + timeout;
                 while (same && !claimed && System.currentTimeMillis() < expires) {
                     /*
                      * We're here because we found the page we want, but another

=== modified file 'src/main/java/com/persistit/CLI.java'
--- src/main/java/com/persistit/CLI.java	2012-11-20 17:45:51 +0000
+++ src/main/java/com/persistit/CLI.java	2013-01-22 16:32:21 +0000
@@ -683,7 +683,7 @@
      * @param volumepath
      * @param rmiport
      * @param y
-     * @return
+     * @return Result description
      * @throws Exception
      */
     @Cmd("open")

=== modified file 'src/main/java/com/persistit/Configuration.java'
--- src/main/java/com/persistit/Configuration.java	2012-11-30 23:06:38 +0000
+++ src/main/java/com/persistit/Configuration.java	2013-01-22 16:32:21 +0000
@@ -909,7 +909,7 @@
      *            Properties containing substitution values
      * @param depth
      *            Count of recursive calls - maximum depth is 20.
-     * @return
+     * @return String value resulting from substitutions
      */
     String substituteProperties(String text, final Properties properties, final int depth) {
         int p = text.indexOf("${");

=== modified file 'src/main/java/com/persistit/Exchange.java'
--- src/main/java/com/persistit/Exchange.java	2013-01-07 23:10:42 +0000
+++ src/main/java/com/persistit/Exchange.java	2013-01-22 16:32:21 +0000
@@ -266,6 +266,7 @@
     private BufferPool _pool;
     private Volume _volume;
     private Tree _tree;
+    private long _timeoutMillis = SharedResource.DEFAULT_MAX_WAIT_TIME;
 
     private volatile long _cachedTreeGeneration = -1;
     private volatile int _cacheDepth = 0;
@@ -300,7 +301,7 @@
 
     private volatile Thread _thread;
 
-    private Exchange(final Persistit persistit) {
+    Exchange(final Persistit persistit) {
         _persistit = persistit;
         _key = new Key(_persistit);
         _spareKey1 = new Key(_persistit);
@@ -1127,7 +1128,7 @@
      * @param buffer
      * @param key
      * @param lc
-     * @return
+     * @return foundAt value
      * @throws PersistitInterruptedException
      */
     private int findKey(final Buffer buffer, final Key key, final LevelCache lc) throws PersistitInterruptedException {
@@ -1281,7 +1282,7 @@
                 }
 
                 if (buffer == null) {
-                    buffer = _pool.get(_volume, pageAddress, writer, true);
+                    buffer = _pool.get(_volume, pageAddress, writer, true, _timeoutMillis);
                 }
                 checkPageType(buffer, currentLevel + PAGE_TYPE_DATA, true);
 
@@ -1714,8 +1715,7 @@
                     try {
                         sequence(WRITE_WRITE_STORE_A);
                         final long depends = _persistit.getTransactionIndex().wwDependency(re.getVersionHandle(),
-                        // TODO - timeout?
-                                _transaction.getTransactionStatus(), SharedResource.DEFAULT_MAX_WAIT_TIME);
+                                _transaction.getTransactionStatus(), _timeoutMillis);
                         if (depends != 0 && depends != TransactionStatus.ABORTED) {
                             // version is from concurrent txn that already
                             // committed
@@ -1738,7 +1738,7 @@
                         treeClaimAcquired = false;
                     }
                     final boolean doWait = (options & StoreOptions.WAIT) != 0;
-                    treeClaimAcquired = _treeHolder.claim(true, doWait ? SharedResource.DEFAULT_MAX_WAIT_TIME : 0);
+                    treeClaimAcquired = _treeHolder.claim(true, doWait ? _timeoutMillis : 0);
                     if (!treeClaimAcquired) {
                         if (!doWait) {
                             throw re;
@@ -2175,7 +2175,8 @@
 
                         Debug.$assert0.t(rightSiblingPage >= 0 && rightSiblingPage <= MAX_VALID_PAGE_ADDR);
                         if (rightSiblingPage > 0) {
-                            final Buffer rightSibling = _pool.get(_volume, rightSiblingPage, false, true);
+                            final Buffer rightSibling = _pool.get(_volume, rightSiblingPage, false, true,
+                                    _timeoutMillis);
                             buffer.releaseTouched();
                             //
                             // Reset foundAtNext to point to the first key block
@@ -2741,6 +2742,105 @@
     }
 
     /**
+     * Invoke {@link #lock(Key, long)} with the current key and a default
+     * timeout value of
+     * {@value com.persistit.SharedResource#DEFAULT_MAX_WAIT_TIME} milliseconds.
+     * 
+     * @throws PersistitException
+     */
+    public void lock() throws PersistitException {
+        lock(_key, SharedResource.DEFAULT_MAX_WAIT_TIME);
+    }
+
+    /**
+     * <p>
+     * Within a transaction, enforces a constraint that no other concurrent
+     * transaction also successfully locks the same key. This method must run
+     * within the scope of an active transaction.
+     * </p>
+     * <p>
+     * This method is designed to help applications overcome problems with
+     * "write skew" which is a type of isolation anomaly permitted by Snapshot
+     * Isolation. See, for example,
+     * http://en.wikipedia.org/wiki/Snapshot_isolation for a concise explanation
+     * of Snapshot Isolation and the write skew anomaly.
+     * <p>
+     * </p>
+     * To use this facility an application specifies a key which may or may not
+     * be associated with an actual storage location, but which is designed to
+     * conflict with any other transaction that could participate in a write
+     * skew. Thus the operation serves as a way of ensuring serializable
+     * execution of transactions that could otherwise experience write skew.
+     * </p>
+     * <p>
+     * This method does not actually use any locking mechanism; rather, it
+     * creates a write-write conflict with another transaction when both
+     * transactions are concurrent and when both transactions attempt to lock
+     * the same key. The result in that case is that one of the transactions
+     * receives a {@link RollbackException}. An application using this facility
+     * simply retries the transaction, at which point it is likely to
+     * successfully execute the call to {@link #lock()}.
+     * </p>
+     * <p>
+     * This method works by writing a short value associated with the provided
+     * key into a temporary volume (accessible through the
+     * {@link Persistit#getLockVolume()} method). The value is removed through
+     * the normal pruning process soon after the all potentially conflicting
+     * transactions have either rolled back or committed.
+     * </p>
+     * <p>
+     * All of these interactions are performed through the normal MVCC
+     * transaction mechanism. This method differs from the {@link #store()}
+     * method only in that the {@link Tree} to which a value is written is
+     * located in a reserved temporary volume and is therefore normally not
+     * written to disk. The key is removed by pruning once there is are no
+     * longer any concurrent transactions that could conflict with it.
+     * </p>
+     * <p>
+     * As part of the normal MVCC process, if this method detects a potentially
+     * conflicting lock written by another active concurrent transaction, this
+     * transaction waits until the other transaction either commits or aborts.
+     * To prevent an unbounded wait time this method accepts a timeout value in
+     * milliseconds. If the potentially conflicting transaction neither commits
+     * nor aborts during the timeout interval, this method throws a
+     * <code>RollbackException</code>. In the event this method attempts to
+     * enter a deadlock state with another current transaction; the potential
+     * deadlock is detected immediately and this method immediately throws a
+     * <code>RollbackException</code>.
+     * </p>
+     * 
+     * @param lockKey
+     *            the source Key
+     * @param timeout
+     *            timeout interval in milliseconds, zero for default timeout
+     * @throws PersistitException
+     * @throws RollbackException
+     *             in the specific case that another concurrent transaction has
+     *             also locked the same key
+     * @throws IllegalStateException
+     *             if this Thread does not have an active transaction scope
+     * @see Transaction
+     */
+    public void lock(final Key lockKey, final long timeout) throws PersistitException {
+        assertCorrectThread(true);
+        _persistit.checkClosed();
+        if (!_transaction.isActive()) {
+            throw new IllegalStateException("No active transaction scope");
+        }
+        final Exchange lockExchange = _persistit.getExchange(_persistit.getLockVolume(), _tree.getName(), true);
+        /**
+         * Lock table trees need tree handles for pruning
+         */
+        _persistit.getJournalManager().handleForTree(lockExchange.getTree());
+        lockExchange.setTimeoutMillis(timeout == 0 ? getTimeoutMillis() : timeout);
+        lockKey.copyTo(lockExchange.getKey());
+        lockExchange.getKey().testValidForStoreAndFetch(_pool.getBufferSize());
+        lockExchange.getValue().clear().putAntiValueMVV();
+        final int options = StoreOptions.WAIT | StoreOptions.DONT_JOURNAL | StoreOptions.MVCC;
+        lockExchange.storeInternal(lockExchange.getKey(), lockExchange.getValue(), 0, options);
+    }
+
+    /**
      * Fetches the value associated with the <code>Key</code>, then inserts or
      * updates the value. Effectively this swaps the content of
      * <code>Value</code> with the database record associated with the current
@@ -3051,7 +3151,7 @@
             // claim is held for the duration to prevent a non-atomic
             // update.
             //
-            getLongRecordHelper().fetchLongRecord(value, minimumBytes);
+            getLongRecordHelper().fetchLongRecord(value, minimumBytes, _timeoutMillis);
         }
     }
 
@@ -4116,7 +4216,7 @@
                 if (buffer.isAfterRightEdge(foundAt)) {
                     final long rightSiblingPage = buffer.getRightSibling();
                     if (rightSiblingPage > 0) {
-                        final Buffer rightSibling = _pool.get(_volume, rightSiblingPage, false, true);
+                        final Buffer rightSibling = _pool.get(_volume, rightSiblingPage, false, true, _timeoutMillis);
                         buffer.releaseTouched();
                         //
                         // Reset foundAtNext to point to the first key block
@@ -4179,6 +4279,50 @@
     }
 
     /**
+     * @return The standard timeout setting in milliseconds for this
+     *         <code>Exchange</code>
+     * @see Exchange#setTimeoutMillis(long)
+     */
+    public long getTimeoutMillis() {
+        assertCorrectThread(true);
+        return _timeoutMillis;
+    }
+
+    /**
+     * <p>
+     * Set the standard timeout for this <code>Exchange</code>. The timeout
+     * value represents an approximate upper bound on the wait time for various
+     * methods that wait for actions by other threads. For example, if a thread
+     * needs to read a value from a {@link Buffer} that is currently be updated
+     * by another thread, the read operation waits up to <code>timeout</code>
+     * milliseconds for the other thread to release the <code>Buffer</code>.
+     * </p>
+     * <p>
+     * The timeout value is advisory, and some operations may stall for a longer
+     * period of time than specified. Setting a timeout does not guarantee
+     * real-time behavior.
+     * </p>
+     * <p>
+     * The supplied value must be non-negative. If it is zero, the default
+     * timeout value {@value com.persistit.SharedResource#DEFAULT_MAX_WAIT_TIME}
+     * milliseconds is set.
+     * </p>
+     * 
+     * @param timeout
+     *            Standard timeout setting, in milliseconds, for operations that
+     *            wait.
+     */
+    public void setTimeoutMillis(final long timeout) {
+        if (timeout == 0) {
+            _timeoutMillis = SharedResource.DEFAULT_MAX_WAIT_TIME;
+        } else {
+            // Clipping this to avoid potential overflows when computing
+            // intervals
+            _timeoutMillis = Util.rangeCheck(timeout, 0, Math.max(timeout, Long.MAX_VALUE / 2));
+        }
+    }
+
+    /**
      * Returns a copy of either the data page or a page on the index path to the
      * data page containing the current key. This method looks up the current
      * key, then copies and returns the page at the specified tree level in a

=== modified file 'src/main/java/com/persistit/JournalManager.java'
--- src/main/java/com/persistit/JournalManager.java	2012-10-26 19:44:17 +0000
+++ src/main/java/com/persistit/JournalManager.java	2013-01-22 16:32:21 +0000
@@ -587,6 +587,9 @@
     }
 
     int handleForVolume(final Volume volume) throws PersistitException {
+        if (volume.getHandle() != 0) {
+            return volume.getHandle();
+        }
         if (!_allowHandlesForTempVolumesAndTrees && volume.isTemporary()) {
             throw new IllegalStateException("Creating handle for temporary volume " + volume);
         }
@@ -621,7 +624,9 @@
             }
             handle = Integer.valueOf(++_handleCounter);
             Debug.$assert0.t(!_handleToTreeMap.containsKey(handle));
-            writeTreeHandleToJournal(td, handle.intValue());
+            if (td.getVolumeHandle() != Volume.LOCK_VOLUME_HANDLE) {
+                writeTreeHandleToJournal(td, handle.intValue());
+            }
             _treeToHandleMap.put(td, handle);
             _handleToTreeMap.put(handle, td);
         }
@@ -629,7 +634,7 @@
     }
 
     int handleForTree(final Tree tree) throws PersistitException {
-        if (!_allowHandlesForTempVolumesAndTrees && tree.getVolume().isTemporary()) {
+        if (!_allowHandlesForTempVolumesAndTrees && tree.getVolume().isTemporary() && !tree.getVolume().isLockVolume()) {
             throw new IllegalStateException("Creating handle for temporary tree " + tree);
         }
         if (tree.getHandle() != 0) {
@@ -659,7 +664,11 @@
     Volume volumeForHandle(final int handle) throws PersistitException {
         final Volume volume = lookupVolumeHandle(handle);
         if (volume == null) {
-            return null;
+            if (handle == Volume.LOCK_VOLUME_HANDLE) {
+                return _persistit.getLockVolume();
+            } else {
+                return null;
+            }
         }
         if (!volume.isOpened()) {
             volume.open(_persistit);

=== modified file 'src/main/java/com/persistit/Key.java'
--- src/main/java/com/persistit/Key.java	2012-11-07 14:03:51 +0000
+++ src/main/java/com/persistit/Key.java	2013-01-22 16:32:21 +0000
@@ -1748,7 +1748,7 @@
      * @param v
      * @param bytes
      * @param offset
-     * @return
+     * @return size of appended segment
      */
     private int appendIntInternal(final int v) {
         int size = _size;

=== modified file 'src/main/java/com/persistit/KeyFilter.java'
--- src/main/java/com/persistit/KeyFilter.java	2012-10-19 14:10:24 +0000
+++ src/main/java/com/persistit/KeyFilter.java	2013-01-22 16:32:21 +0000
@@ -1524,7 +1524,7 @@
      * @param level
      * @param forward
      * @param eq
-     * @return
+     * @return whether there may be more matching keys
      */
     private boolean next(final Key key, final int index, final int level, final boolean forward, final boolean eq) {
 

=== modified file 'src/main/java/com/persistit/KeyParser.java'
--- src/main/java/com/persistit/KeyParser.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/KeyParser.java	2013-01-22 16:32:21 +0000
@@ -218,7 +218,7 @@
      * successful, append the segment to the key.
      * 
      * @param key
-     * @return
+     * @return <code>true</code> a valid key segment was parsed
      */
     private boolean parseKeySegment(final Key key) {
         final int index = _index;

=== modified file 'src/main/java/com/persistit/LongRecordHelper.java'
--- src/main/java/com/persistit/LongRecordHelper.java	2012-08-24 13:57:19 +0000
+++ src/main/java/com/persistit/LongRecordHelper.java	2013-01-22 16:32:21 +0000
@@ -60,7 +60,8 @@
      * @param minimumBytesToFetch
      * @throws PersistitException
      */
-    void fetchLongRecord(final Value value, final int minimumBytesToFetch) throws PersistitException {
+    void fetchLongRecord(final Value value, final int minimumBytesToFetch, final long timeout)
+            throws PersistitException {
 
         Buffer buffer = null;
 
@@ -93,7 +94,7 @@
                     corrupt("Invalid LONG_RECORD remaining size=" + remainingSize + " of " + rawSize + " in page "
                             + page);
                 }
-                buffer = _volume.getPool().get(_volume, page, false, true);
+                buffer = _volume.getPool().get(_volume, page, false, true, timeout);
                 if (buffer.getPageType() != PAGE_TYPE_LONG_RECORD) {
                     corrupt("LONG_RECORD chain is invalid at page " + page + " - invalid page type: " + buffer);
                 }

=== modified file 'src/main/java/com/persistit/Persistit.java'
--- src/main/java/com/persistit/Persistit.java	2012-12-31 01:47:18 +0000
+++ src/main/java/com/persistit/Persistit.java	2013-01-22 16:32:21 +0000
@@ -284,6 +284,8 @@
 
     private final ThreadLocal<SoftReference<Value>> _valueThreadLocal = new ThreadLocal<SoftReference<Value>>();
 
+    private volatile Volume _lockVolume;
+
     /**
      * Construct a hollow Persistit instance. To be useful, the instance must
      * receive a <code>Configuration</code> through one of the methods
@@ -1284,6 +1286,18 @@
     }
 
     /**
+     * @return reserved temporary volume for locks
+     * @throws PersistitException
+     */
+    public synchronized Volume getLockVolume() throws PersistitException {
+        if (_lockVolume == null) {
+            _lockVolume = createTemporaryVolume();
+            _lockVolume.setHandle(Volume.LOCK_VOLUME_HANDLE);
+        }
+        return _lockVolume;
+    }
+
+    /**
      * @return The {@link SplitPolicy} that will by applied by default to newly
      *         created or allocated {@link Exchange}s.
      */

=== modified file 'src/main/java/com/persistit/Tree.java'
--- src/main/java/com/persistit/Tree.java	2012-09-27 18:22:25 +0000
+++ src/main/java/com/persistit/Tree.java	2013-01-22 16:32:21 +0000
@@ -274,8 +274,8 @@
     }
 
     /**
-     * Assign are set the tree handle. The tree must may not be a member of a
-     * temporary volume.
+     * Set the tree handle. The tree must may not be a member of a temporary
+     * volume.
      * 
      * @throws PersistitException
      */
@@ -332,7 +332,7 @@
      * only once.
      * 
      * @param handle
-     * @return
+     * @return the handle
      * @throws IllegalStateException
      *             if the handle has already been set
      */

=== modified file 'src/main/java/com/persistit/TreeBuilder.java'
--- src/main/java/com/persistit/TreeBuilder.java	2013-01-04 00:16:13 +0000
+++ src/main/java/com/persistit/TreeBuilder.java	2013-01-22 16:32:21 +0000
@@ -657,7 +657,8 @@
      * @param key
      * @param v1
      * @param v2
-     * @return
+     * @return If <code>true</code>, the resulting value is v2. If
+     *         <code>false</code>, the resulting value is v1.
      * @throws DuplicateKeyException
      *             if a key being inserted or merged matches a key already
      *             exists

=== modified file 'src/main/java/com/persistit/Volume.java'
--- src/main/java/com/persistit/Volume.java	2013-01-02 22:41:20 +0000
+++ src/main/java/com/persistit/Volume.java	2013-01-22 16:32:21 +0000
@@ -48,6 +48,8 @@
  */
 public class Volume {
 
+    final static int LOCK_VOLUME_HANDLE = Integer.MAX_VALUE;
+
     private final String _name;
     private long _id;
     private final AtomicBoolean _closing = new AtomicBoolean();
@@ -336,6 +338,10 @@
         return getStorage().isTemp();
     }
 
+    boolean isLockVolume() {
+        return getHandle() == LOCK_VOLUME_HANDLE;
+    }
+
     /**
      * @return The size in bytes of one page in this <code>Volume</code>.
      */
@@ -546,7 +552,7 @@
      * only once.
      * 
      * @param handle
-     * @return
+     * @return the handle
      * @throws IllegalStateException
      *             if the handle has already been set
      */

=== added file 'src/test/java/com/persistit/ExchangeLockTest.java'
--- src/test/java/com/persistit/ExchangeLockTest.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/persistit/ExchangeLockTest.java	2013-01-22 16:32:21 +0000
@@ -0,0 +1,313 @@
+/**
+ * Copyright © 2012 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.concurrent.Semaphore;
+
+import org.junit.Test;
+
+import com.persistit.exception.InUseException;
+import com.persistit.exception.InvalidKeyException;
+import com.persistit.exception.PersistitException;
+
+public class ExchangeLockTest extends PersistitUnitTestCase {
+    private final static long DMILLIS = SharedResource.DEFAULT_MAX_WAIT_TIME;
+    private final Semaphore _coordinator = new Semaphore(0);
+
+    @Test
+    public void singleThreadedLock() throws Exception {
+        final Exchange ex = _persistit.getExchange("persistit", "ExchangeLockTest", true);
+        final Transaction txn = ex.getTransaction();
+        try {
+            ex.lock();
+            fail("Expected to fail");
+        } catch (final IllegalStateException e) {
+            // expected
+        }
+        txn.begin();
+        try {
+            try {
+                ex.lock();
+                fail("Expected to fail");
+            } catch (final InvalidKeyException e) {
+                // expected
+            }
+            ex.append("motor");
+            ex.lock();
+            final Tree tree = _persistit.getLockVolume().getTree("ExchangeLockTest", false);
+            assertTrue("Expected tree to be defined", tree != null);
+            final Exchange ex2 = new Exchange(tree);
+            ex2.ignoreMVCCFetch(true);
+            assertTrue("Expect a key in the temp volume", ex2.next(true));
+            txn.commit();
+        } catch (final Exception e) {
+            e.printStackTrace();
+        } finally {
+            txn.end();
+        }
+
+        txn.begin();
+        try {
+            ex.lock();
+            txn.commit();
+        } finally {
+            txn.end();
+        }
+    }
+
+    private class Locker implements Runnable {
+        final Semaphore _semaphore = new Semaphore(0);
+        final long _timeout;
+        final int[] _sequence;
+        Exception _exception;
+        volatile int _expectedReleases;
+
+        volatile boolean _committed;
+
+        private Locker(final long timeout, final int... sequence) {
+            _timeout = timeout;
+            _sequence = sequence;
+        }
+
+        private void go(final int waitFor) throws InterruptedException {
+            _semaphore.release();
+            _coordinator.acquire(waitFor);
+        }
+
+        private int cycles() {
+            return _sequence.length + 2;
+        }
+
+        @Override
+        public void run() {
+            try {
+                final Exchange ex = _persistit.getExchange("persistit", "ExchangeLockTest", true);
+                final Transaction txn = ex.getTransaction();
+                _expectedReleases = cycles();
+                txn.begin();
+                try {
+                    for (final int k : _sequence) {
+                        _semaphore.acquire();
+                        ex.clear().append(k).lock(ex.getKey(), _timeout);
+                        _coordinator.release();
+                        _expectedReleases--;
+                    }
+                    _semaphore.acquire();
+                    txn.commit();
+                    _committed = true;
+                    _coordinator.release();
+                    _expectedReleases--;
+                } catch (final Exception e) {
+                    _exception = e;
+                    txn.rollback();
+                } finally {
+                    if (_expectedReleases > 0) {
+                        _coordinator.release(_expectedReleases);
+                    }
+                    _semaphore.acquire();
+                    txn.end();
+                    _coordinator.release();
+                    _expectedReleases--;
+                }
+
+            } catch (final Exception e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    private Thread[] start(final Locker... lockers) {
+        final Thread[] threads = new Thread[lockers.length];
+        int count = 0;
+        for (final Locker locker : lockers) {
+            final Thread t = new Thread(locker);
+            t.start();
+            threads[count++] = t;
+        }
+        return threads;
+    }
+
+    private void join(final Thread[] threads) throws InterruptedException {
+        for (final Thread t : threads) {
+            t.join();
+        }
+    }
+
+    @Test
+    public void nonConflictingLocks() throws Exception {
+        final Locker a = new Locker(DMILLIS, 1, 5);
+        final Locker b = new Locker(DMILLIS, 2, 6);
+        final Locker c = new Locker(DMILLIS, 3, 7);
+        final Locker d = new Locker(DMILLIS, 4, 8);
+        final Thread[] threads = start(a, b, c, d);
+        for (int i = 0; i < a.cycles(); i++) {
+            a.go(1);
+            b.go(1);
+            c.go(1);
+            d.go(1);
+        }
+        join(threads);
+        assertTrue(a._committed);
+        assertTrue(b._committed);
+        assertTrue(c._committed);
+        assertTrue(d._committed);
+    }
+
+    @Test
+    public void simpleConflictingLocks() throws Exception {
+        final Locker a = new Locker(1000, 1);
+        final Locker b = new Locker(1000, 1);
+        final Thread[] threads = start(a, b);
+        final long start = System.currentTimeMillis();
+        for (int i = 0; i < a.cycles(); i++) {
+            a.go(1);
+            b.go(1);
+        }
+        join(threads);
+        final long end = System.currentTimeMillis();
+        assertTrue(end - start >= 1000);
+        assertTrue(a._committed ^ b._committed);
+    }
+
+    @Test
+    public void deadlock() throws Exception {
+        final Locker a = new Locker(DMILLIS, 1, 2);
+        final Locker b = new Locker(DMILLIS, 2, 1);
+        final Thread[] threads = start(a, b);
+        final long start = System.currentTimeMillis();
+        for (int i = 0; i < a.cycles(); i++) {
+            a.go(i == 0 ? 1 : 0);
+            b.go(i == 0 ? 1 : 0);
+        }
+        join(threads);
+        final long end = System.currentTimeMillis();
+        assertTrue(end - start < DMILLIS);
+        assertTrue(a._committed ^ b._committed);
+    }
+
+    @Test
+    public void multiWayDeadlock() throws Exception {
+        final Locker a = new Locker(DMILLIS, 1, 2, 3, 4, 5);
+        final Locker b = new Locker(DMILLIS, 2, 3, 4, 5, 1);
+        final Locker c = new Locker(DMILLIS, 3, 4, 5, 1, 2);
+        final Locker d = new Locker(DMILLIS, 4, 5, 1, 2, 3);
+        final Locker e = new Locker(DMILLIS, 5, 1, 2, 3, 4);
+        final Thread[] threads = start(a, b, c, d, e);
+        final long start = System.currentTimeMillis();
+        for (int i = 0; i < a.cycles(); i++) {
+            final int w = (i == 0) ? 1 : 0;
+            a.go(w);
+            b.go(w);
+            c.go(w);
+            d.go(w);
+            e.go(w);
+        }
+        join(threads);
+        final long end = System.currentTimeMillis();
+        assertTrue(end - start < DMILLIS);
+        int succeeded = 0;
+        for (final Locker l : new Locker[] { a, b, c, d, e }) {
+            if (l._committed) {
+                succeeded++;
+            }
+        }
+        assertEquals(1, succeeded);
+    }
+
+    @Test
+    public void lockTablePruning() throws Exception {
+        final Exchange ex = _persistit.getExchange("persistit", "ExchangeLockTest", true);
+        final Transaction txn = ex.getTransaction();
+        txn.begin();
+        for (int i = 0; i < 10000; i++) {
+            ex.clear().append(i).append(RED_FOX).lock();
+        }
+        txn.commit();
+        txn.end();
+
+        final Exchange lockExchange = new Exchange(_persistit.getLockVolume().getTree("ExchangeLockTest", false));
+        lockExchange.ignoreMVCCFetch(true);
+        final int count1 = keyCount(lockExchange);
+        assertTrue(count1 > 0);
+        for (int i = 0; i < 10000; i++) {
+            lockExchange.clear().append(i).append(RED_FOX).prune();
+        }
+
+        final int count2 = keyCount(lockExchange);
+        assertTrue(count2 < count1);
+        _persistit.getTransactionIndex().updateActiveTransactionCache();
+        _persistit.getCleanupManager().poll();
+
+        final int count3 = keyCount(lockExchange);
+        assertEquals(0, count3);
+
+    }
+
+    private int keyCount(final Exchange ex) throws PersistitException {
+        int count = 0;
+        ex.clear();
+        while (ex.next(true)) {
+            count++;
+        }
+        return count;
+    }
+
+    @Test
+    public void timeout() throws Exception {
+        /*
+         * A cursory check of the Exchange timeout value
+         */
+        final Exchange ex = _persistit.getExchange("persistit", "gogo", true);
+        ex.to("a").store();
+        final long latchedPage = ex.fetchBufferCopy(0).getPageAddress();
+        final BufferPool pool = ex.getBufferPool();
+        final Volume volume = ex.getVolume();
+        ex.setTimeoutMillis(1000);
+        final long start = System.currentTimeMillis();
+        final Semaphore a = new Semaphore(0);
+        final Semaphore b = new Semaphore(0);
+
+        new Thread(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    final Buffer buffer = pool.get(volume, latchedPage, true, true);
+                    a.release();
+                    b.acquire();
+                    buffer.release();
+                } catch (final Exception e) {
+                    e.printStackTrace();
+                }
+            }
+        }).start();
+        a.acquire();
+        try {
+            ex.to("a").store();
+            fail("Expected an InUseException");
+
+        } catch (final InUseException e) {
+            // expected
+        }
+        final long end = System.currentTimeMillis();
+        b.release();
+        final long interval = end - start;
+        assertTrue("Should have waited about 1 second", interval >= 1000 && interval < 2000);
+    }
+}

=== modified file 'src/test/java/com/persistit/MediatedFileChannelTest.java'
--- src/test/java/com/persistit/MediatedFileChannelTest.java	2012-08-24 13:57:19 +0000
+++ src/test/java/com/persistit/MediatedFileChannelTest.java	2013-01-22 16:32:21 +0000
@@ -84,7 +84,6 @@
             for (int i = 0; i < 65536; i++) {
                 bb.array()[i] = (byte) (i % 32 + 64);
             }
-
             final long start = System.nanoTime();
             while (errors == 0 && System.nanoTime() - start < TIME) {
                 try {
@@ -94,6 +93,7 @@
                     fc.force(true);
                     count++;
                 } catch (final InterruptedIOException e) {
+                    bb.position(0);
                     // ignore -- expected
                     interrupts++;
                     // need to clear the interrupted status

=== modified file 'src/test/java/com/persistit/unit/UnitTestProperties.java'
--- src/test/java/com/persistit/unit/UnitTestProperties.java	2012-08-10 16:18:11 +0000
+++ src/test/java/com/persistit/unit/UnitTestProperties.java	2013-01-22 16:32:21 +0000
@@ -41,7 +41,6 @@
         p.setProperty("journalpath", "${datapath}/persistit_journal");
         p.setProperty("logfile", "${datapath}/persistit_${timestamp}.log");
         p.setProperty("tmpvoldir", "${datapath}");
-        p.setProperty("rmiport", System.getProperty("rmiport", "8081"));
         p.setProperty("jmx", "true");
         return p;
     }
@@ -62,7 +61,6 @@
         p.setProperty("journalpath", "${datapath}/persistit_journal");
         p.setProperty("logfile", "${datapath}/persistit_${timestamp}.log");
         p.setProperty("tmpvoldir", "${datapath}");
-        p.setProperty("rmiport", System.getProperty("rmiport", "8081"));
         return p;
     }
 
@@ -78,7 +76,6 @@
         p.setProperty("journalpath", "${datapath}/persistit_alt_journal");
         p.setProperty("logfile", "${datapath}/persistit_${timestamp}.log");
         p.setProperty("tmpvoldir", "${datapath}");
-        p.setProperty("rmiport", System.getProperty("rmiport", "8081"));
         return p;
     }
 
@@ -94,8 +91,6 @@
         p.setProperty("journalpath", "${datapath}/persistit_journal");
         p.setProperty("logfile", "${datapath}/persistit_${timestamp}.log");
         p.setProperty("tmpvoldir", "${datapath}");
-        p.setProperty("rmiport", System.getProperty("rmiport", "8081"));
-        p.setProperty("jmx", "true");
         return p;
     }