Merge lp:~nwilliams/akiban-persistit/fix_912514_fetchAndRemove_2 into lp:akiban-persistit

[Nathan Williams]

Fix Exchange store and fetchAndRemove to work correctly both inside and out of a transaction.

The vast majority of this is Peter's original branch. The last few commits were refactorings, to eliminate duplicate code, that I noticed were possible and suggested on the merge prop.

Original branch and description:
lp:~pbeaman/akiban-persistit/fix_912514_fetchAndRemove

The basic strategy to fix this bug is to used a the ThreadLocal-based cached Value object, rather than Exchange._spareValue, as the MVV holding value. The major change was to remove the "_" from instances of _spareValue in storeInternal. However, there were other changes required to reenable StoreOption.FETCH | StoreOption.MVCC, etc.

New test Bug912514 test tests some of the cases. I will also (and have not yet) run PersistitMapStressTest which is where we first observed the bug.

[Peter Beaman (pbeaman)]

Refactorings look good. Thank you.

[Peter Beaman (pbeaman)]

review: Approve

=== modified file 'src/main/java/com/persistit/Exchange.java'
--- src/main/java/com/persistit/Exchange.java	2012-05-25 18:50:59 +0000
+++ src/main/java/com/persistit/Exchange.java	2012-05-29 19:52:29 +0000
@@ -1031,7 +1031,6 @@
      * 
      * @return Encoded key location within the data page. The page itself is
      *         made valid in the level cache.
-     * @throws PMapException
      */
     private int search(Key key, boolean writer) throws PersistitException {
         Buffer buffer = null;
@@ -1099,7 +1098,6 @@
      * 
      * @return Encoded key location within the level. The page itself is valid
      *         within the level cache.
-     * @throws PMapException
      */
     private int searchTree(Key key, int toLevel, boolean writer) throws PersistitException {
         Buffer oldBuffer = null;
@@ -1198,7 +1196,6 @@
      * @param currentLevel
      *            current level in the tree
      * @return Encoded key location within the page.
-     * @throws PMapException
      */
     private int searchLevel(Key key, boolean edge, long pageAddress, int currentLevel, boolean writer)
             throws PersistitException {
@@ -1323,15 +1320,9 @@
      *             uponError
      */
     boolean storeInternal(Key key, Value value, int level, int options) throws PersistitException {
-        if ((options & StoreOptions.FETCH) > 0 && (options & StoreOptions.MVCC) > 0) {
-            throw new IllegalArgumentException("Both fetch and MVCC not supported");
-        }
 
         final boolean doMVCC = (options & StoreOptions.MVCC) > 0;
-        final boolean doAnyFetch = (options & StoreOptions.FETCH) > 0 || doMVCC;
-
-        // spare used for fetch
-        Debug.$assert0.t(!doAnyFetch || value != _spareValue);
+        final boolean doFetch = (options & StoreOptions.FETCH) > 0;
 
         // spares used for new splits/levels
         Debug.$assert0.t(key != _spareKey1);
@@ -1344,6 +1335,8 @@
         boolean incrementMVVCount = false;
 
         final int maxSimpleValueSize = maxValueSize(key.getEncodedSize());
+        final Value spareValue = _persistit.getThreadLocalValue();
+        assert !(doMVCC & value == spareValue || doFetch && value == _spareValue): "storeInternal may be use the supplied Value: " + value;
 
         //
         // First insert the record in the data page
@@ -1369,7 +1362,7 @@
             // This method may delay significantly for I/O and must
             // be called when there are no other claimed resources.
             //
-            newLongRecordPointer = getLongRecordHelper().storeLongRecord(value, _transaction.isActive());
+            newLongRecordPointer = getLongRecordHelper().storeLongRecord(value,  _transaction.isActive());
         }
 
         if (!_ignoreTransactions && ((options & StoreOptions.DONT_JOURNAL) == 0)) {
@@ -1396,7 +1389,7 @@
                 if (!committed && newLongRecordPointerMVV != 0) {
                     _volume.getStructure().deallocateGarbageChain(newLongRecordPointerMVV, 0);
                     newLongRecordPointerMVV = 0;
-                    _spareValue.changeLongRecordMode(false);
+                    spareValue.changeLongRecordMode(false);
                 }
 
                 if (treeClaimRequired && !treeClaimAcquired) {
@@ -1462,37 +1455,38 @@
                         if (keyExisted) {
                             oldLongRecordPointer = buffer.fetchLongRecordPointer(foundAt);
                         }
-                        if (doAnyFetch) {
-                            buffer.fetch(foundAt, _spareValue);
-                            /*
-                             * If we aren't in MVCC we have to un-long-ify as
-                             * fetch was requested. Otherwise only do it if it
-                             * is a long MVV so as to not-needlessly create one.
-                             */
-                            if (!doMVCC) {
-                                fetchFixupForLongRecords(_spareValue, Integer.MAX_VALUE);
-                            } else if (oldLongRecordPointer != 0) {
-                                if (isLongMVV(_spareValue)) {
+
+                        if (doFetch || doMVCC) {
+                            buffer.fetch(foundAt, spareValue);
+                            if (oldLongRecordPointer != 0) {
+                                if (isLongMVV(spareValue)) {
                                     oldLongRecordPointerMVV = oldLongRecordPointer;
-                                    fetchFixupForLongRecords(_spareValue, Integer.MAX_VALUE);
+                                    fetchFixupForLongRecords(spareValue, Integer.MAX_VALUE);
                                 }
-                                /*
-                                 * If it was a long MVV we saved it into the
-                                 * variable above. Otherwise it is a primordial
-                                 * value that we can't get rid of.
-                                 */
-                                oldLongRecordPointer = 0;
+                            }
+                            /*
+                             * If it was a long MVV we saved it into the
+                             * variable above. Otherwise it is a
+                             * primordial value that we can't get rid
+                             * of.
+                             */
+                            oldLongRecordPointer = 0;
+
+                            if (doFetch) {
+                                spareValue.copyTo(_spareValue);
+                                fetchFromValueInternal(_spareValue, Integer.MAX_VALUE, buffer);
                             }
                         }
+
                         if (doMVCC) {
-                            valueToStore = _spareValue;
+                            valueToStore = spareValue;
                             int valueSize = value.getEncodedSize();
                             /*
                              * If key didn't exist the value is truly
                              * non-existent and not just undefined/zero length
                              */
-                            byte[] spareBytes = _spareValue.getEncodedBytes();
-                            int spareSize = keyExisted ? _spareValue.getEncodedSize() : -1;
+                            byte[] spareBytes = spareValue.getEncodedBytes();
+                            int spareSize = keyExisted ? spareValue.getEncodedSize() : -1;
                             spareSize = MVV.prune(spareBytes, 0, spareSize, _persistit.getTransactionIndex(), false,
                                     prunedVersions);
 
@@ -1521,8 +1515,8 @@
                             MVV.visitAllVersions(_mvvVisitor, spareBytes, 0, spareSize);
 
                             int mvvSize = MVV.estimateRequiredLength(spareBytes, spareSize, valueSize);
-                            _spareValue.ensureFit(mvvSize);
-                            spareBytes = _spareValue.getEncodedBytes();
+                            spareValue.ensureFit(mvvSize);
+                            spareBytes = spareValue.getEncodedBytes();
 
                             long versionHandle = TransactionIndex.tss2vh(_transaction.getStartTimestamp(), tStep);
                             int storedLength = MVV.storeVersion(spareBytes, 0, spareSize, spareBytes.length,
@@ -1530,11 +1524,10 @@
 
                             incrementMVVCount = (storedLength & MVV.STORE_EXISTED_MASK) == 0;
                             storedLength &= MVV.STORE_LENGTH_MASK;
-                            _spareValue.setEncodedSize(storedLength);
+                            spareValue.setEncodedSize(storedLength);
 
-                            if (_spareValue.getEncodedSize() > maxSimpleValueSize) {
-                                newLongRecordPointerMVV = getLongRecordHelper().storeLongRecord(_spareValue,
-                                        _transaction.isActive());
+                            if (spareValue.getEncodedSize() > maxSimpleValueSize) {
+                                newLongRecordPointerMVV = getLongRecordHelper().storeLongRecord(spareValue, _transaction.isActive());
                             }
                         }
                     }
@@ -1673,7 +1666,7 @@
             }
 
             value.changeLongRecordMode(false);
-            _spareValue.changeLongRecordMode(false);
+            spareValue.changeLongRecordMode(false);
             if (!committed) {
                 //
                 // We failed to write the new LONG_RECORD. If there was
@@ -1698,7 +1691,7 @@
         }
         _volume.getStatistics().bumpStoreCounter();
         _tree.getStatistics().bumpStoreCounter();
-        if (doAnyFetch) {
+        if (doFetch || doMVCC) {
             _volume.getStatistics().bumpFetchCounter();
             _tree.getStatistics().bumpFetchCounter();
         }
@@ -1761,7 +1754,6 @@
      *            The encoded insert location.
      * @return <code>true</code> if it necessary to insert a key into the
      *         ancestor index page.
-     * @throws PMapException
      */
     // TODO - Check insertIndexLevel timestamps
     private boolean putLevel(LevelCache lc, Key key, ValueHelper valueWriter, Buffer buffer, int foundAt,
@@ -2142,7 +2134,7 @@
                         index = _key.getEncodedSize();
 
                         if (matches) {
-                            matches = fetchInternal(buffer, outValue, foundAt, minimumBytes);
+                            matches = fetchFromBufferInternal(buffer, outValue, foundAt, minimumBytes);
                             if (!matches && direction != EQ) {
                                 nudged = false;
                                 nudgeForMVCC = (direction == GTEQ || direction == LTEQ);
@@ -2162,7 +2154,7 @@
                         if (matches) {
                             index = _key.nextElementIndex(parentIndex);
                             if (index > 0) {
-                                boolean isVisibleMatch = fetchInternal(buffer, outValue, foundAt, minimumBytes);
+                                boolean isVisibleMatch = fetchFromBufferInternal(buffer, outValue, foundAt, minimumBytes);
                                 //
                                 // In any case (matching sibling, child or
                                 // niece/nephew) we need to ignore this
@@ -2569,7 +2561,9 @@
         _persistit.checkClosed();
         _persistit.checkSuspended();
         _key.testValidForStoreAndFetch(_volume.getPageSize());
-        storeInternal(_key, _value, 0, StoreOptions.FETCH | StoreOptions.WAIT);
+        int options = StoreOptions.WAIT | StoreOptions.FETCH;
+        options |= (!_ignoreTransactions && _transaction.isActive()) ? StoreOptions.MVCC : 0;
+        storeInternal(_key, _value, 0, options);
         _spareValue.copyTo(_value);
         return this;
     }
@@ -2727,7 +2721,7 @@
         if (minimumBytes < 0) {
             minimumBytes = 0;
         }
-        fetchInternal(value, minimumBytes);
+        searchAndFetchInternal(value, minimumBytes);
         return this;
     }
 
@@ -2748,9 +2742,29 @@
      *             As thrown from any internal method.
      * @return <code>true</code> if the value was visible.
      */
-    private boolean fetchInternal(Buffer buffer, Value value, int foundAt, int minimumBytes) throws PersistitException {
+    private boolean fetchFromBufferInternal(Buffer buffer, Value value, int foundAt, int minimumBytes) throws PersistitException {
+        buffer.fetch(foundAt, value);
+        return fetchFromValueInternal(value, minimumBytes, buffer);
+    }
+
+    /**
+     * Helper for finalizing the value to return from a, potentially, MVV
+     * contained in the given Value.
+     *
+     * @param value
+     *            Value to finalize.
+     * @param minimumBytes
+     *            Minimum amount of LONG_RECORD to fetch. If &lt;0, the
+     *            <code>value</code> will contain just the descriptor portion.
+     * @param bufferForPruning
+     *            If not <code>null</code> and <code>Value</code> did contain
+     *            an MVV, call {@link Buffer#enqueuePruningAction(int)}.
+     * @throws PersistitException
+     *             As thrown from any internal method.
+     * @return <code>true</code> if the value was visible.
+     */
+    private boolean fetchFromValueInternal(Value value, int minimumBytes, Buffer bufferForPruning) throws PersistitException {
         boolean visible = true;
-        buffer.fetch(foundAt, value);
         /*
          * We must fetch the full LONG_RECORD, if needed, while buffer is
          * claimed from calling code so that it can't be de-allocated as we are
@@ -2763,7 +2777,9 @@
              */
             fetchFixupForLongRecords(value, Integer.MAX_VALUE);
             if (MVV.isArrayMVV(value.getEncodedBytes(), 0, value.getEncodedSize())) {
-                buffer.enqueuePruningAction(_tree.getHandle());
+                if (bufferForPruning != null) {
+                    bufferForPruning.enqueuePruningAction(_tree.getHandle());
+                }
                 visible = mvccFetch(value, minimumBytes);
                 fetchFixupForLongRecords(value, minimumBytes);
             }
@@ -2790,13 +2806,13 @@
      * @throws PersistitException
      *             As thrown from {@link #search(Key, boolean)}
      */
-    private void fetchInternal(Value value, int minimumBytes) throws PersistitException {
+    private void searchAndFetchInternal(Value value, int minimumBytes) throws PersistitException {
         Buffer buffer = null;
         try {
             int foundAt = search(_key, false);
             LevelCache lc = _levelCache[0];
             buffer = lc._buffer;
-            fetchInternal(buffer, value, foundAt, minimumBytes);
+            fetchFromBufferInternal(buffer, value, foundAt, minimumBytes);
             _volume.getStatistics().bumpFetchCounter();
             _tree.getStatistics().bumpFetchCounter();
         } finally {
@@ -3063,7 +3079,7 @@
 
         _value.clear().putAntiValueMVV();
         final int storeOptions = StoreOptions.MVCC | StoreOptions.WAIT | StoreOptions.ONLY_IF_VISIBLE
-                | StoreOptions.DONT_JOURNAL;
+                | StoreOptions.DONT_JOURNAL | (fetchFirst ? StoreOptions.FETCH : 0);
 
         boolean anyRemoved = false;
         boolean keyIsLessThan = true;
@@ -3755,8 +3771,7 @@
     /**
      * Called by Transaction to set up a context for committing updates.
      * 
-     * @param volume
-     * @param _treeName
+     * @param tree
      */
     void setTree(Tree tree) throws PersistitException {
         _persistit.checkClosed();
@@ -3944,7 +3959,7 @@
         boolean savedIgnore = _ignoreMVCCFetch;
         try {
             _ignoreMVCCFetch = true;
-            fetchInternal(_spareValue, -1);
+            searchAndFetchInternal(_spareValue, -1);
             final boolean wasLong = isLongRecord(_spareValue);
             _spareValue.clear();
             return wasLong;
@@ -3966,7 +3981,7 @@
         boolean savedIgnore = _ignoreMVCCFetch;
         try {
             _ignoreMVCCFetch = true;
-            fetchInternal(_spareValue, -1);
+            searchAndFetchInternal(_spareValue, -1);
             final boolean wasLong = isLongMVV(_spareValue);
             _spareValue.clear();
             return wasLong;
 
=== added file 'src/test/java/com/persistit/Bug912514Test.java'
--- src/test/java/com/persistit/Bug912514Test.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/persistit/Bug912514Test.java	2012-05-29 19:52:29 +0000
@@ -0,0 +1,114 @@
+/**
+ * Copyright © 2012 Akiban Technologies, Inc.  All rights reserved.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, version 3 (only) of the
+ * License.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ * 
+ * This program may also be available under different license terms. For more
+ * information, see www.akiban.com or contact [e-mail].
+ */
+
+package com.persistit;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+import com.persistit.unit.PersistitUnitTestCase;
+
+/**
+ * https://bugs.launchpad.net/akiban-persistit/+bug/912514
+ * 
+ * PersistitMapStress1 fails intermittently with messages like this:
+ * 
+ * Finished unit=#1 PersistitMapStress test=PersistitMapStress1 main at ts=8036
+ * - elapsed=5738 - FAILED: value not expected to be null 8036 Failed test
+ * unit=#1 PersistitMapStress test=PersistitMapStress1 main value not expected
+ * to be null
+ * 
+ * I didn't track down the responsible code, but I'm pretty sure the contents of
+ * _spareValue have been mangled.
+ * 
+ * Since akiban-server does not use this method the bug is no more than medium
+ * priority. But it does need to be fixed before akiban-persistit is released as
+ * a standalone library.
+ */
+
+public class Bug912514Test extends PersistitUnitTestCase {
+    
+    private final static String ROLLBACK = "rollback";
+    
+    private void fetchAndStoreAndRemoveHelper(final boolean inTxn, final String... sequence) throws Exception {
+        final Transaction txn = _persistit.getTransaction();
+        final Exchange exchange = _persistit.getExchange("persistit", "Bug912514Test", true);
+        String previous = null;
+        for (String string : sequence) {
+
+            if (inTxn) {
+                txn.begin();
+            }
+            
+            if (string == null) {
+                exchange.to(1).fetchAndRemove();
+            } else {
+                exchange.getValue().put(string);
+                exchange.to(1).fetchAndStore();
+            }
+            compare(previous, exchange.getValue());
+
+            if (inTxn) {
+                if (string.startsWith(ROLLBACK)) {
+                    txn.rollback();
+                } else {
+                    txn.commit();
+                }
+                txn.end();
+                compare(previous, exchange.getValue());
+            }
+
+            if (!inTxn || !string.startsWith(ROLLBACK)) {
+                previous = string;
+            }
+            exchange.fetch();
+            compare(previous, exchange.getValue());
+
+
+        }
+    }
+
+    private void compare(final String string, final Value value) {
+        if (string == null) {
+            assertTrue("Value should be undefined", !value.isDefined());
+        } else {
+            assertEquals("Value should match", string, value.getString());
+        }
+    }
+
+    @Test
+    public void fetchAndStoreTxn() throws Exception {
+        fetchAndStoreAndRemoveHelper(true, RED_FOX, createString(100), createString(1000), createString(10000), RED_FOX);
+    }
+
+    @Test
+    public void fetchAndRemoveNonTxn() throws Exception {
+        fetchAndStoreAndRemoveHelper(false, RED_FOX, null, null, createString(100), null, createString(1000), null,
+                createString(10000), null, null, RED_FOX, null);
+    }
+
+    @Test
+    public void fetchAndStoreTxnWithRollbacks() throws Exception {
+        fetchAndStoreAndRemoveHelper(true, RED_FOX, createString(100), ROLLBACK, createString(1000), ROLLBACK
+                + createString(10000), createString(10000), RED_FOX);
+    }
+
+}