A persistent multi-version map - test concurrent iterate and write

h2/src/test/org/h2/test/store/TestConcurrent.java

@@ -5,6 +5,7 @@
  */
 package org.h2.test.store;
 
+import java.util.Iterator;
 import java.util.Map;
 import java.util.Random;
 import org.h2.dev.store.btree.MVMap;
@@ -27,10 +28,47 @@ public class TestConcurrent extends TestMVStore {
     }
 
     public void test() throws InterruptedException {
+        testConcurrentIterate();
         testConcurrentWrite();
         testConcurrentRead();
     }
 
+    private void testConcurrentIterate() {
+        MVStore s = MVStore.open(null, new TestMapFactory());
+        s.setMaxPageSize(3);
+        final MVMap<Integer, Integer> map = s.openMap("test");
+        final int len = 10;
+        final Random r = new Random();
+        Task t = new Task() {
+            @Override
+            public void call() throws Exception {
+                while (!stop) {
+                    int x = r.nextInt(len);
+                    if (r.nextBoolean()) {
+                        map.remove(x);
+                    } else {
+                        map.put(x, r.nextInt(100));
+                    }
+                }
+            }
+        };
+        t.execute();
+        for (int k = 0; k < 10000; k++) {
+            Iterator<Integer> it = map.keyIterator(r.nextInt(len));
+            long old = s.incrementVersion();
+            s.setRetainVersion(old - 100);
+            while (map.getVersion() == old) {
+                Thread.yield();
+            }
+            while (it.hasNext()) {
+                it.next();
+            }
+        }
+        t.get();
+        s.close();
+    }
+
+
     /**
      * Test what happens on concurrent write. Concurrent write may corrupt the
      * map, so that keys and values may become null.

h2/src/test/org/h2/test/store/TestMVStore.java

@@ -31,6 +31,7 @@ public class TestMVStore extends TestBase {
     }
 
     public void test() throws InterruptedException {
+        testIterateOldVersion();
         testObjects();
         testExample();
         testIterateOverChanges();
@@ -52,6 +53,29 @@ public class TestMVStore extends TestBase {
         testSimple();
     }
 
+    private void testIterateOldVersion() {
+        MVStore s;
+        Map<Integer, Integer> map;
+        s = MVStore.open(null, new TestMapFactory());
+        map = s.openMap("test");
+        int len = 100;
+        for (int i = 0; i < len; i++) {
+            map.put(i, 10 * i);
+        }
+        Iterator<Integer> it = map.keySet().iterator();
+        s.incrementVersion();
+        for (int i = 0; i < len; i += 2) {
+            map.remove(i);
+        }
+        int count = 0;
+        while (it.hasNext()) {
+            it.next();
+            count++;
+        }
+        assertEquals(len, count);
+        s.close();
+    }
+
     private void testObjects() {
         String fileName = getBaseDir() + "/testObjects.h3";
         FileUtils.delete(fileName);

h2/src/tools/org/h2/dev/store/btree/ChangeCursor.java

@@ -16,8 +16,8 @@ public class ChangeCursor<K, V> extends Cursor<K, V> {
 
     private final long minVersion;
 
-    ChangeCursor(MVMap<K, V> map, long minVersion) {
-        super(map);
+    ChangeCursor(MVMap<K, V> map, Page root, K from, long minVersion) {
+        super(map, root, from);
         this.minVersion = minVersion;
     }
 

h2/src/tools/org/h2/dev/store/btree/Cursor.java

@@ -18,28 +18,22 @@ import java.util.Iterator;
 public class Cursor<K, V> implements Iterator<K> {
 
     protected final MVMap<K, V> map;
-    protected final ArrayList<CursorPos> parents = new ArrayList<CursorPos>();
+    protected final Page root;
+    protected final K from;
+    protected ArrayList<CursorPos> parents;
     protected CursorPos currentPos;
     protected K current;
 
-    Cursor(MVMap<K, V> map) {
+    Cursor(MVMap<K, V> map, Page root, K from) {
         this.map = map;
-    }
-
-    /**
-     * Fetch the first key.
-     *
-     * @param root the root page
-     * @param from the key, or null
-     */
-    void start(Page root, K from) {
-        currentPos = min(root, from);
-        if (currentPos != null) {
-            fetchNext();
-        }
+        this.root = root;
+        this.from = from;
     }
 
     public K next() {
+        if (!hasNext()) {
+            return null;
+        }
         K c = current;
         if (c != null) {
             fetchNext();
@@ -56,6 +50,14 @@ public class Cursor<K, V> implements Iterator<K> {
     }
 
     public boolean hasNext() {
+        if (parents == null) {
+            // not initialized yet: fetch the first key
+            parents = new ArrayList<CursorPos>();
+            currentPos = min(root, from);
+            if (currentPos != null) {
+                fetchNext();
+            }
+        }
         return current != null;
     }
 

h2/src/tools/org/h2/dev/store/btree/MVMap.java

@@ -28,9 +28,9 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
     protected final MVStore store;
 
     /**
-     * The root page (may not be null).
+     * The current root page (may not be null).
      */
-    protected Page root;
+    protected volatile Page root;
 
     private final int id;
     private final String name;
@@ -291,7 +291,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
     public void close() {
         closed = true;
         readOnly = true;
-        clearOldVersions();
+        removeAllOldVersions();
         root = null;
     }
 
@@ -366,6 +366,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
 
     protected void setRoot(Page newRoot) {
         if (root != newRoot) {
+            removeUnusedOldVersions();
             if (root.getVersion() != newRoot.getVersion()) {
                 ArrayList<Page> list = oldRoots;
                 if (list.size() > 0) {
@@ -447,9 +448,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
      */
     public Iterator<K> keyIterator(K from) {
         checkOpen();
-        Cursor<K, V> c = new Cursor<K, V>(this);
-        c.start(root, from);
-        return c;
+        return new Cursor<K, V>(this, root, from);
     }
 
     /**
@@ -461,9 +460,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
      */
     public Iterator<K> changeIterator(long minVersion) {
         checkOpen();
-        Cursor<K, V> c = new ChangeCursor<K, V>(this, minVersion);
-        c.start(root, null);
-        return c;
+        return new ChangeCursor<K, V>(this, root, null, minVersion);
     }
 
     public Set<Map.Entry<K, V>> entrySet() {
@@ -481,9 +478,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
 
             @Override
             public Iterator<K> iterator() {
-                Cursor<K, V> c = new Cursor<K, V>(MVMap.this);
-                c.start(root, null);
-                return c;
+                return new Cursor<K, V>(MVMap.this, root, null);
             }
 
             @Override
@@ -532,6 +527,7 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
      */
     void rollbackTo(long version) {
         checkWrite();
+        removeUnusedOldVersions();
         if (version < createVersion) {
             removeMap();
         } else if (root.getVersion() != version) {
@@ -553,12 +549,39 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
     /**
      * Forget all old versions.
      */
-    void clearOldVersions() {
+    void removeAllOldVersions() {
         // create a new instance
         // because another thread might iterate over it
         oldRoots = new ArrayList<Page>();
     }
 
+    /**
+     * Forget those old versions that are no longer needed.
+     */
+    void removeUnusedOldVersions() {
+        long oldest = store.getRetainVersion();
+        if (oldest == -1) {
+            return;
+        }
+        ArrayList<Page> list = oldRoots;
+        int i = 0;
+        // TODO iterate over the list is inefficient
+        for (; i < list.size(); i++) {
+            Page p = list.get(i);
+            if (p.getVersion() > oldest) {
+                break;
+            }
+        }
+        if (i == 0) {
+            return;
+        }
+        // create a new instance
+        // because another thread might iterate over it
+        list = new ArrayList<Page>();
+        list.addAll(oldRoots.subList(i, oldRoots.size()));
+        oldRoots = list;
+    }
+
     public void setReadOnly(boolean readOnly) {
         this.readOnly = readOnly;
     }
@@ -671,4 +694,8 @@ public class MVMap<K, V> extends AbstractMap<K, V> {
         return m;
     }
 
+    public long getVersion() {
+        return root.getVersion();
+    }
+
 }

h2/src/tools/org/h2/dev/store/btree/MVStore.java

@@ -35,7 +35,6 @@ header:
 H:3,blockSize=4096,...
 
 TODO:
-- concurrent iterator (when to increment version; read on first hasNext())
 - how to iterate (just) over deleted pages / entries
 - compact: use total max length instead of page count (liveCount)
 - support background writes (store old version)
@@ -62,6 +61,7 @@ TODO:
 - allocate memory Utils.newBytes
 - unified exception handling
 - check if locale specific string comparison can make data disappear
+- concurrent map; avoid locking during IO (pre-load pages)
 
 */
 
@@ -113,6 +113,7 @@ public class MVStore {
     private int lastMapId;
 
     private boolean reuseSpace = true;
+    private long retainVersion = -1;
     private int retainChunk = -1;
 
     private Compressor compressor = new CompressLZF();
@@ -976,6 +977,19 @@ public class MVStore {
         this.retainChunk = retainChunk;
     }
 
+    /**
+     * Which version to retain. If not set, all versions up to the last stored version are retained.
+     *
+     * @param retainVersion the oldest version to retain
+     */
+    public void setRetainVersion(long retainVersion) {
+        this.retainVersion = retainVersion;
+    }
+
+    public long getRetainVersion() {
+        return retainVersion;
+    }
+
     private boolean isKnownVersion(long version) {
         if (version > currentVersion || version < 0) {
             return false;
@@ -1067,7 +1081,7 @@ public class MVStore {
     private void revertTemp() {
         freedChunks.clear();
         for (MVMap<?, ?> m : mapsChanged.values()) {
-            m.clearOldVersions();
+            m.removeAllOldVersions();
         }
         mapsChanged.clear();
     }