Merge pull request #1627 from h2database/append-single

Use lock to protect append buffer

h2/src/main/org/h2/mvstore/MVStore.java

@@ -1401,7 +1401,7 @@ public class MVStore implements AutoCloseable {
         try {
             ChunkIdsCollector collector = new ChunkIdsCollector(meta.getId());
             long oldestVersionToKeep = getOldestVersionToKeep();
-            MVMap.RootReference rootReference = meta.getRoot();
+            MVMap.RootReference rootReference = meta.flushAndGetRoot();
             if (fast) {
                 MVMap.RootReference previous;
                 while (rootReference.version >= oldestVersionToKeep && (previous = rootReference.previous) != null) {

h2/src/main/org/h2/mvstore/Page.java

@@ -493,6 +493,16 @@ public abstract class Page implements Cloneable
         return bKeys;
     }
 
+    abstract void expand(int extraKeyCount, Object[] extraKeys, Object[] extraValues);
+
+    final void expandKeys(int extraKeyCount, Object[] extraKeys) {
+        int keyCount = getKeyCount();
+        Object[] newKeys = createKeyStorage(keyCount + extraKeyCount);
+        System.arraycopy(keys, 0, newKeys, 0, keyCount);
+        System.arraycopy(extraKeys, 0, newKeys, keyCount, extraKeyCount);
+        keys = newKeys;
+    }
+
     /**
      * Get the total number of key-value pairs, including child pages.
      *
@@ -1012,6 +1022,11 @@ public abstract class Page implements Cloneable
             return newPage;
         }
 
+        @Override
+        public void expand(int keyCount, Object[] extraKys, Object[] extraValues) {
+            throw new UnsupportedOperationException();
+        }
+
         @Override
         public long getTotalCount() {
             assert !isComplete() || totalCount == calculateTotalCount() :
@@ -1324,6 +1339,21 @@ public abstract class Page implements Cloneable
             return newPage;
         }
 
+        @Override
+        public void expand(int extraKeyCount, Object[] extraKeys, Object[] extraValues) {
+            int keyCount = getKeyCount();
+            expandKeys(extraKeyCount, extraKeys);
+            if(values != null) {
+                Object[] newValues = createValueStorage(keyCount + extraKeyCount);
+                System.arraycopy(values, 0, newValues, 0, keyCount);
+                System.arraycopy(extraValues, 0, newValues, keyCount, extraKeyCount);
+                values = newValues;
+            }
+            if(isPersistent()) {
+                recalculateMemory();
+            }
+        }
+
         @Override
         public long getTotalCount() {
             return getKeyCount();

h2/src/main/org/h2/mvstore/rtree/MVRTreeMap.java

@@ -134,7 +134,7 @@ public final class MVRTreeMap<V> extends MVMap<SpatialKey, V> {
         int attempt = 0;
         while(true) {
             ++attempt;
-            RootReference rootReference = getRoot();
+            RootReference rootReference = flushAndGetRoot();
             Page p = rootReference.root.copy(true);
             V result = operate(p, key, value, decisionMaker);
             if (!p.isLeaf() && p.getTotalCount() == 0) {

h2/src/main/org/h2/mvstore/tx/Transaction.java

@@ -61,7 +61,7 @@ public class Transaction {
      */
     private static final int STATUS_ROLLED_BACK  = 5;
 
-    private static final String STATUS_NAMES[] = {
+    private static final String[] STATUS_NAMES = {
             "CLOSED", "OPEN", "PREPARED", "COMMITTED", "ROLLING_BACK", "ROLLED_BACK"
     };
     static final int LOG_ID_BITS = 40;
@@ -91,7 +91,7 @@ public class Transaction {
     /**
      * This is really a transaction identity, because it's not re-used.
      */
-    public final long sequenceNum;
+    final long sequenceNum;
 
     /*
      * Transaction state is an atomic composite field:
@@ -138,12 +138,17 @@ public class Transaction {
     /**
      * Map on which this transaction is blocked.
      */
-    MVMap<?,VersionedValue> blockingMap;
+    private MVMap<?,VersionedValue> blockingMap;
 
     /**
      * Key in blockingMap on which this transaction is blocked.
      */
-    Object blockingKey;
+    private Object blockingKey;
+
+    /**
+     * Whether other transaction(s) are waiting for this to close.
+     */
+    private volatile boolean notificationRequested;
 
 
     Transaction(TransactionStore store, int transactionId, long sequenceNum, int status,
@@ -237,7 +242,8 @@ public class Transaction {
     }
 
     public int getBlockerId() {
-        return blockingTransaction == null ? 0 : blockingTransaction.ownerId;
+        Transaction blocker = this.blockingTransaction;
+        return blocker == null ? 0 : blocker.ownerId;
     }
 
     /**
@@ -390,19 +396,25 @@ public class Transaction {
     public void rollbackToSavepoint(long savepointId) {
         long lastState = setStatus(STATUS_ROLLING_BACK);
         long logId = getLogId(lastState);
+        boolean success;
         try {
             store.rollbackTo(this, logId, savepointId);
         } finally {
-            notifyAllWaitingTransactions();
+            if (notificationRequested) {
+                notifyAllWaitingTransactions();
+            }
             long expectedState = composeState(STATUS_ROLLING_BACK, logId, hasRollback(lastState));
             long newState = composeState(STATUS_OPEN, savepointId, true);
-            if (!statusAndLogId.compareAndSet(expectedState, newState)) {
-                throw DataUtils.newIllegalStateException(
-                        DataUtils.ERROR_TRANSACTION_ILLEGAL_STATE,
-                        "Transaction {0} concurrently modified " +
-                                "while rollback to savepoint was in progress",
-                        transactionId);
-            }
+            do {
+                success = statusAndLogId.compareAndSet(expectedState, newState);
+            } while (!success && statusAndLogId.get() == expectedState);
+        }
+        // this is moved outside of finally block to avert masking original exception, if any
+        if (!success) {
+            throw DataUtils.newIllegalStateException(
+                    DataUtils.ERROR_TRANSACTION_ILLEGAL_STATE,
+                    "Transaction {0} concurrently modified while rollback to savepoint was in progress",
+                    transactionId);
         }
     }
 
@@ -474,7 +486,7 @@ public class Transaction {
     void closeIt() {
         long lastState = setStatus(STATUS_CLOSED);
         store.store.deregisterVersionUsage(txCounter);
-        if(hasChanges(lastState) || hasRollback(lastState)) {
+        if((hasChanges(lastState) || hasRollback(lastState)) && notificationRequested) {
             notifyAllWaitingTransactions();
         }
     }
@@ -483,7 +495,10 @@ public class Transaction {
         notifyAll();
     }
 
-    public boolean waitFor(Transaction toWaitFor) {
+    public boolean waitFor(Transaction toWaitFor, MVMap<?,VersionedValue> map, Object key) {
+        blockingTransaction = toWaitFor;
+        blockingMap = map;
+        blockingKey = key;
         if (isDeadlocked(toWaitFor)) {
             StringBuilder details = new StringBuilder(
                     String.format("Transaction %d has been chosen as a deadlock victim. Details:%n", transactionId));
@@ -504,7 +519,6 @@ public class Transaction {
             }
         }
 
-        blockingTransaction = toWaitFor;
         try {
             return toWaitFor.waitForThisToEnd(timeoutMillis);
         } finally {
@@ -527,6 +541,7 @@ public class Transaction {
 
     private synchronized boolean waitForThisToEnd(int millis) {
         long until = System.currentTimeMillis() + millis;
+        notificationRequested = true;
         long state;
         int status;
         while((status = getStatus(state = statusAndLogId.get())) != STATUS_CLOSED
@@ -557,8 +572,15 @@ public class Transaction {
 
     @Override
     public String toString() {
-        long state = statusAndLogId.get();
-        return transactionId + "(" + sequenceNum + ") " + STATUS_NAMES[getStatus(state)] + " " + getLogId(state);
+        return transactionId + "(" + sequenceNum + ") " + stateToString();
+    }
+
+    private String stateToString() {
+        return stateToString(statusAndLogId.get());
+    }
+
+    private static String stateToString(long state) {
+        return STATUS_NAMES[getStatus(state)] + (hasRollback(state) ? "<" : "") + " " + getLogId(state);
     }
 
 

h2/src/main/org/h2/mvstore/tx/TransactionMap.java

@@ -46,7 +46,7 @@ public class TransactionMap<K, V> extends AbstractMap<K, V> {
     /**
      * The transaction which is used for this map.
      */
-    final Transaction transaction;
+    private final Transaction transaction;
 
     TransactionMap(Transaction transaction, MVMap<K, VersionedValue> map) {
         this.transaction = transaction;
@@ -105,16 +105,16 @@ public class TransactionMap<K, V> extends AbstractMap<K, V> {
         long undoLogSize;
         do {
             committingTransactions = store.committingTransactions.get();
-            mapRootReference = map.getRoot();
+            mapRootReference = map.flushAndGetRoot();
             BitSet opentransactions = store.openTransactions.get();
             undoLogRootReferences = new MVMap.RootReference[opentransactions.length()];
             undoLogSize = 0;
             for (int i = opentransactions.nextSetBit(0); i >= 0; i = opentransactions.nextSetBit(i+1)) {
                 MVMap<Long, Object[]> undoLog = store.undoLogs[i];
                 if (undoLog != null) {
-                    MVMap.RootReference rootReference = undoLog.getRoot();
+                    MVMap.RootReference rootReference = undoLog.flushAndGetRoot();
                     undoLogRootReferences[i] = rootReference;
-                    undoLogSize += rootReference.root.getTotalCount() + rootReference.getAppendCounter();
+                    undoLogSize += rootReference.getTotalCount();
                 }
             }
         } while(committingTransactions != store.committingTransactions.get() ||
@@ -125,7 +125,7 @@ public class TransactionMap<K, V> extends AbstractMap<K, V> {
         // should be considered as committed.
         // Subsequent processing uses this snapshot info only.
         Page mapRootPage = mapRootReference.root;
-        long size = mapRootPage.getTotalCount();
+        long size = mapRootReference.getTotalCount();
         // if we are looking at the map without any uncommitted values
         if (undoLogSize == 0) {
             return size;
@@ -241,6 +241,16 @@ public class TransactionMap<K, V> extends AbstractMap<K, V> {
         return set(key, decisionMaker);
     }
 
+    /**
+     * Appends entry to uderlying map. This method may be used concurrently,
+     * but latest appended values are not guaranteed to be visible.
+     * @param key should be higher in map's order than any existing key
+     * @param value to be appended
+     */
+    public void append(K key, V value) {
+        map.append(key, VersionedValueUncommitted.getInstance(transaction.log(map.getId(), key, null), value, null));
+    }
+
     /**
      * Lock row for the given key.
      * <p>
@@ -298,16 +308,12 @@ public class TransactionMap<K, V> extends AbstractMap<K, V> {
             assert decision != MVMap.Decision.REPEAT;
             blockingTransaction = decisionMaker.getBlockingTransaction();
             if (decision != MVMap.Decision.ABORT || blockingTransaction == null) {
-                transaction.blockingMap = null;
-                transaction.blockingKey = null;
                 @SuppressWarnings("unchecked")
                 V res = result == null ? null : (V) result.getCurrentValue();
                 return res;
             }
             decisionMaker.reset();
-            transaction.blockingMap = map;
-            transaction.blockingKey = key;
-        } while (blockingTransaction.sequenceNum > sequenceNumWhenStarted || transaction.waitFor(blockingTransaction));
+        } while (blockingTransaction.sequenceNum > sequenceNumWhenStarted || transaction.waitFor(blockingTransaction, map, key));
 
         throw DataUtils.newIllegalStateException(DataUtils.ERROR_TRANSACTION_LOCKED,
                 "Map entry <{0}> with key <{1}> and value {2} is locked by tx {3} and can not be updated by tx {4}"
@@ -669,8 +675,8 @@ public class TransactionMap<K, V> extends AbstractMap<K, V> {
         private final boolean includeAllUncommitted;
         private X current;
 
-        protected TMIterator(TransactionMap<K,?> transactionMap, K from, K to, boolean includeAllUncommitted) {
-            Transaction transaction = transactionMap.transaction;
+        TMIterator(TransactionMap<K,?> transactionMap, K from, K to, boolean includeAllUncommitted) {
+            Transaction transaction = transactionMap.getTransaction();
             this.transactionId = transaction.transactionId;
             TransactionStore store = transaction.store;
             MVMap<K, VersionedValue> map = transactionMap.map;
@@ -683,7 +689,7 @@ public class TransactionMap<K, V> extends AbstractMap<K, V> {
             MVMap.RootReference mapRootReference;
             do {
                 committingTransactions = store.committingTransactions.get();
-                mapRootReference = map.getRoot();
+                mapRootReference = map.flushAndGetRoot();
             } while (committingTransactions != store.committingTransactions.get());
             // Now we have a snapshot, where mapRootReference points to state of the map
             // and committingTransactions mask tells us which of seemingly uncommitted changes

h2/src/main/org/h2/mvstore/tx/TransactionStore.java

@@ -390,7 +390,7 @@ public class TransactionStore {
      */
     long addUndoLogRecord(int transactionId, long logId, Object[] undoLogRecord) {
         MVMap<Long, Object[]> undoLog = undoLogs[transactionId];
-        Long undoKey = getOperationId(transactionId, logId);
+        long undoKey = getOperationId(transactionId, logId);
         if (logId == 0 && !undoLog.isEmpty()) {
             throw DataUtils.newIllegalStateException(
                     DataUtils.ERROR_TOO_MANY_OPEN_TRANSACTIONS,