CacheLongKeyLIRS concurrency improvements

h2/src/main/org/h2/engine/DbSettings.java

@@ -8,6 +8,7 @@
 import java.util.HashMap;
 
 import org.h2.api.ErrorCode;
+import org.h2.util.Utils;
 import org.h2.message.DbException;
 
 /**
@@ -306,6 +307,26 @@ public class DbSettings extends SettingsBase {
      */
     public final boolean compressData = get("COMPRESS", false);
 
+    /**
+     * Database setting <code>CACHE_CONCURRENCY</code>
+     * (default: 16).<br />
+     * Set the read cache concurrency.
+     */
+    public final int cacheConcurrency = get("CACHE_CONCURRENCY", 16);
+
+    /**
+     * Database setting <code>AUTO_COMMIT_BUFFER_SIZE_KB</code>
+     * (default: depends on max heap).<br />
+     * Set the size of the write buffer, in KB disk space (for file-based
+     * stores). Unless auto-commit is disabled, changes are automatically
+     * saved if there are more than this amount of changes.
+     *
+     * When the value is set to 0 or lower, data is not automatically
+     * stored.
+     */
+    public final int autoCommitBufferSize = get("AUTO_COMMIT_BUFFER_SIZE_KB",
+        Math.max(1, Math.min(19, Utils.scaleForAvailableMemory(64))) * 1024);
+
     /**
      * Database setting <code>IGNORE_CATALOGS</code>
      * (default: false).<br />

h2/src/main/org/h2/mvstore/cache/CacheLongKeyLIRS.java

@@ -6,6 +6,8 @@
 package org.h2.mvstore.cache;
 
 import java.lang.ref.WeakReference;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -602,6 +604,21 @@ private static class Segment<V> {
          */
         private int stackMoveCounter;
 
+        /*
+         * Holds entries that were concurrently get()
+         */
+        private final ConcurrentSkipListSet<Entry<V>> concAccess;
+
+        /*
+         * Serialize access to this segments
+         */
+        private final ReentrantLock l;
+
+        /*
+         * Used as null value for ConcurrentSkipListSet
+         */
+        private final Entry<V> ENTRY_NULL = new Entry<>();
+
         /**
          * Create a new cache segment.
          *  @param maxMemory the maximum memory to use
@@ -632,6 +649,9 @@ private static class Segment<V> {
             @SuppressWarnings("unchecked")
             Entry<V>[] e = new Entry[len];
             entries = e;
+
+            concAccess = new ConcurrentSkipListSet<>();
+            l = new ReentrantLock();
         }
 
         /**
@@ -710,17 +730,42 @@ private void addToMap(Entry<V> e) {
          * @param e the entry
          * @return the value, or null if there is no resident entry
          */
-        synchronized V get(Entry<V> e) {
+        V get(Entry<V> e) {
             V value = e == null ? null : e.getValue();
-            if (value == null) {
-                // the entry was not found
-                // or it was a non-resident entry
-                misses++;
-            } else {
-                access(e);
-                hits++;
+            if (!l.tryLock()) {
+                Entry<V> e2 = value == null ? ENTRY_NULL : e;
+                concAccess.add(e2);
+                return value;
+            }
+            try {
+                if (value == null) {
+                    // the entry was not found
+                    // or it was a non-resident entry
+                    misses++;
+                } else {
+                    access(e);
+                    hits++;
+                }
+
+                // process entries that were accessed concurrently
+                while (true) {
+                    Entry<V> p = concAccess.pollFirst();
+                    if (p == null) {
+                        break;
+                    }
+                    if (p == ENTRY_NULL) {
+                        misses++;
+                    } else {
+                        access(p);
+                        hits++;
+                    }
+                }
+
+                return value;
+            }
+            finally {
+                l.unlock();
             }
-            return value;
         }
 
         /**
@@ -749,9 +794,8 @@ private void access(Entry<V> e) {
                 V v = e.getValue();
                 if (v != null) {
                     removeFromQueue(e);
-                    if (e.reference != null) {
+                    if (e.value == null) {
                         e.value = v;
-                        e.reference = null;
                         usedMemory += e.memory;
                     }
                     if (e.stackNext != null) {
@@ -787,7 +831,17 @@ private void access(Entry<V> e) {
          * @param memory the memory used for the given entry
          * @return the old value, or null if there was no resident entry
          */
-        synchronized V put(long key, int hash, V value, int memory) {
+        V put(long key, int hash, V value, int memory) {
+            l.lock();
+            try {
+                return putUnlocked(key, hash, value, memory);
+            }
+            finally {
+                l.unlock();
+            }
+        }
+
+        private V putUnlocked(long key, int hash, V value, int memory) {
             Entry<V> e = find(key, hash);
             boolean existed = e != null;
             V old = null;
@@ -832,7 +886,17 @@ synchronized V put(long key, int hash, V value, int memory) {
          * @param hash the hash
          * @return the old value, or null if there was no resident entry
          */
-        synchronized V remove(long key, int hash) {
+        V remove(long key, int hash) {
+            l.lock();
+            try {
+                return removeUnlocked(key, hash);
+            }
+            finally {
+                l.unlock();
+            }
+        }
+
+        private V removeUnlocked(long key, int hash) {
             int index = hash & mask;
             Entry<V> e = entries[index];
             if (e == null) {
@@ -897,7 +961,6 @@ private void evictBlock() {
                 Entry<V> e = queue.queuePrev;
                 usedMemory -= e.memory;
                 removeFromQueue(e);
-                e.reference = new WeakReference<>(e.value);
                 e.value = null;
                 addToQueue(queue2, e);
                 // the size of the non-resident-cold entries needs to be limited
@@ -1031,7 +1094,17 @@ private void removeFromQueue(Entry<V> e) {
          * @param nonResident true for non-resident entries
          * @return the key list
          */
-        synchronized List<Long> keys(boolean cold, boolean nonResident) {
+        List<Long> keys(boolean cold, boolean nonResident) {
+            l.lock();
+            try {
+                return keysUnlocked(cold, nonResident);
+            }
+            finally {
+                l.unlock();
+            }
+        }
+
+        private List<Long> keysUnlocked(boolean cold, boolean nonResident) {
             ArrayList<Long> keys = new ArrayList<>();
             if (cold) {
                 Entry<V> start = nonResident ? queue2 : queue;
@@ -1053,7 +1126,17 @@ synchronized List<Long> keys(boolean cold, boolean nonResident) {
          *
          * @return the set of keys
          */
-        synchronized Set<Long> keySet() {
+        Set<Long> keySet() {
+            l.lock();
+            try {
+                return keySetUnlocked();
+            }
+            finally {
+                l.unlock();
+            }
+        }
+
+        private Set<Long> keySetUnlocked() {
             HashSet<Long> set = new HashSet<>();
             for (Entry<V> e = stack.stackNext; e != stack; e = e.stackNext) {
                 set.add(e.key);
@@ -1086,7 +1169,7 @@ void setMaxMemory(long maxMemory) {
      *
      * @param <V> the value type
      */
-    static class Entry<V> {
+    static class Entry<V> implements Comparable<Entry<V>> {
 
         /**
          * The key.
@@ -1148,10 +1231,15 @@ static class Entry<V> {
             this.key = key;
             this.memory = memory;
             this.value = value;
+            if (value != null) {
+                this.reference = new WeakReference<>(value);
+            }
         }
 
         Entry(Entry<V> old) {
-            this(old.key, old.value, old.memory);
+            this.key = old.key;
+            this.memory = old.memory;
+            this.value = old.value;
             this.reference = old.reference;
             this.topMove = old.topMove;
         }
@@ -1166,12 +1254,17 @@ boolean isHot() {
         }
 
         V getValue() {
-            return value == null ? reference.get() : value;
+            final V v = value;
+            return v == null ? reference.get() : v;
         }
 
         int getMemory() {
             return value == null ? 0 : memory;
         }
+
+        public int compareTo(Entry<V> tgt) {
+            return key == tgt.key ? 0 : key < tgt.key ? -1 : 1;
+        }
     }
 
     /**

h2/src/main/org/h2/mvstore/db/Store.java

@@ -123,6 +123,9 @@ public Store(Database db) {
             // otherwise background thread would compete for store lock
             // with maps opening procedure
             builder.autoCommitDisabled();
+
+            builder.cacheConcurrency(db.getSettings().cacheConcurrency);
+            builder.autoCommitBufferSize(db.getSettings().autoCommitBufferSize);
         }
         this.encrypted = encrypted;
         try {