CacheLongKeyLIRS concurrency improvements

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

@@ -1897,6 +1897,10 @@ public TraceSystem getTraceSystem() {
         return traceSystem;
     }
 
+    public int getCacheSize() {
+        return cacheSize;
+    }
+
     public synchronized void setCacheSize(int kb) {
         if (starting) {
             int max = MathUtils.convertLongToInt(Utils.getMemoryMax()) / 2;

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,10 @@
 package org.h2.mvstore.cache;
 
 import java.lang.ref.WeakReference;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -161,9 +165,13 @@ public V put(long key, V value, int memory) {
         // check whether resize is required: synchronize on s, to avoid
         // concurrent resizes (concurrent reads read
         // from the old segment)
-        synchronized (s) {
-            s = resizeIfNeeded(s, segmentIndex);
-            return s.put(key, hash, value, memory);
+        s.l.lock();
+        try {
+            Segment<V> s2 = resizeIfNeeded(s, segmentIndex);
+            return s2.put(key, hash, value, memory);
+        }
+        finally {
+            s.l.unlock();
         }
     }
 
@@ -208,9 +216,13 @@ public V remove(long key) {
         // check whether resize is required: synchronize on s, to avoid
         // concurrent resizes (concurrent reads read
         // from the old segment)
-        synchronized (s) {
-            s = resizeIfNeeded(s, segmentIndex);
-            return s.remove(key, hash);
+        s.l.lock();
+        try {
+            Segment<V> s2 = resizeIfNeeded(s, segmentIndex);
+            return s2.remove(key, hash);
+        }
+        finally {
+            s.l.unlock();
         }
     }
 
@@ -312,7 +324,7 @@ public long getMaxMemory() {
      *
      * @return the entry set
      */
-    public synchronized Set<Map.Entry<Long, V>> entrySet() {
+    public Set<Map.Entry<Long, V>> entrySet() {
         return getMap().entrySet();
     }
 
@@ -389,7 +401,7 @@ public long getHits() {
     public long getMisses() {
         int x = 0;
         for (Segment<V> s : segments) {
-            x += s.misses;
+            x += s.misses.get();
         }
         return x;
     }
@@ -491,9 +503,13 @@ public void putAll(Map<Long, ? extends V> m) {
      */
     public void trimNonResidentQueue() {
         for (Segment<V> s : segments) {
-            synchronized (s) {
+            s.l.lock();
+            try {
                 s.trimNonResidentQueue();
             }
+            finally {
+                s.l.unlock();
+            }
         }
     }
 
@@ -527,7 +543,7 @@ private static class Segment<V> {
         /**
          * The number of cache misses.
          */
-        long misses;
+        final AtomicLong misses;
 
         /**
          * The map array. The size is always a power of 2.
@@ -602,6 +618,16 @@ 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;
+
         /**
          * Create a new cache segment.
          *  @param maxMemory the maximum memory to use
@@ -632,6 +658,10 @@ private static class Segment<V> {
             @SuppressWarnings("unchecked")
             Entry<V>[] e = new Entry[len];
             entries = e;
+
+            misses = new AtomicLong();
+            concAccess = new ConcurrentSkipListSet<>();
+            l = new ReentrantLock();
         }
 
         /**
@@ -646,7 +676,7 @@ private static class Segment<V> {
             this(old.maxMemory, old.stackMoveDistance, len,
                     old.nonResidentQueueSize, old.nonResidentQueueSizeHigh);
             hits = old.hits;
-            misses = old.misses;
+            misses.set(old.misses.get());
             Entry<V> s = old.stack.stackPrev;
             while (s != old.stack) {
                 Entry<V> e = new Entry<>(s);
@@ -710,17 +740,37 @@ 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 {
+                misses.incrementAndGet();
+                return null;
+            }
+            if (!l.tryLock()) {
+                //concAccess.add(e);
+                return value;
+            }
+            try {
                 access(e);
                 hits++;
+
+                // process entries that were accessed concurrently
+//                while (true) {
+//                    Entry<V> p = concAccess.pollFirst();
+//                    if (p == null) {
+//                        break;
+//                    }
+//                    access(p);
+//                    hits++;
+//                }
+
+                return value;
+            }
+            finally {
+                l.unlock();
             }
-            return value;
         }
 
         /**
@@ -749,9 +799,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 +836,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 +891,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 +966,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 +1099,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 +1131,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);
@@ -1074,7 +1162,6 @@ synchronized Set<Long> keySet() {
         void setMaxMemory(long maxMemory) {
             this.maxMemory = maxMemory;
         }
-
     }
 
     /**
@@ -1086,7 +1173,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 +1235,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 +1258,18 @@ 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;
         }
+
+        @Override
+        public int compareTo(Entry<V> tgt) {
+            return Long.compare(this.key, tgt.key);
+        }
     }
 
     /**

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

@@ -123,6 +123,10 @@ 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);
+            builder.cacheSize(db.getCacheSize());
         }
         this.encrypted = encrypted;
         try {