Change the default recycler type.

Recycling is not thread-local anymore by default but instead there are several
pools of objects to recycle that threads may use depending on their id.
Each pool is protected by its own lock so up to ${number of pools} threads may
recycler objects concurrently.

Recyclers have also been refactored for better composability, for example there
is a soft recycler that creates a recycler that wraps data around a
SoftReference and a thread-local recycler that can take any factory or recyclers
and instantiates a dedicated instance per thread.

RecyclerBenchmark has been added to try to figure out the overhead of object
recycling depending on the recycler type and the number of threads trying to
recycle objects concurrently.

Close elastic#4647

Author: jpountz

src/main/java/org/elasticsearch/cache/recycler/CacheRecycler.java

@@ -24,11 +24,14 @@
 import org.elasticsearch.ElasticsearchIllegalArgumentException;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.recycler.*;
+import org.elasticsearch.common.recycler.Recycler;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 
 import java.util.Locale;
 
+import static org.elasticsearch.common.recycler.Recyclers.*;
+
 @SuppressWarnings("unchecked")
 public class CacheRecycler extends AbstractComponent {
 
@@ -66,8 +69,9 @@ public CacheRecycler(Settings settings) {
         final Type type = Type.parse(settings.get("type"));
         int limit = settings.getAsInt("limit", 10);
         int smartSize = settings.getAsInt("smart_size", 1024);
+        final int availableProcessors = EsExecutors.boundedNumberOfProcessors(settings);
 
-        hashMap = build(type, limit, smartSize, new Recycler.C<ObjectObjectOpenHashMap>() {
+        hashMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<ObjectObjectOpenHashMap>() {
             @Override
             public ObjectObjectOpenHashMap newInstance(int sizing) {
                 return new ObjectObjectOpenHashMap(size(sizing));
@@ -78,7 +82,7 @@ public void clear(ObjectObjectOpenHashMap value) {
                 value.clear();
             }
         });
-        hashSet = build(type, limit, smartSize, new Recycler.C<ObjectOpenHashSet>() {
+        hashSet = build(type, limit, smartSize, availableProcessors, new Recycler.C<ObjectOpenHashSet>() {
             @Override
             public ObjectOpenHashSet newInstance(int sizing) {
                 return new ObjectOpenHashSet(size(sizing), 0.5f);
@@ -89,7 +93,7 @@ public void clear(ObjectOpenHashSet value) {
                 value.clear();
             }
         });
-        doubleObjectMap = build(type, limit, smartSize, new Recycler.C<DoubleObjectOpenHashMap>() {
+        doubleObjectMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<DoubleObjectOpenHashMap>() {
             @Override
             public DoubleObjectOpenHashMap newInstance(int sizing) {
                 return new DoubleObjectOpenHashMap(size(sizing));
@@ -100,7 +104,7 @@ public void clear(DoubleObjectOpenHashMap value) {
                 value.clear();
             }
         });
-        longObjectMap = build(type, limit, smartSize, new Recycler.C<LongObjectOpenHashMap>() {
+        longObjectMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<LongObjectOpenHashMap>() {
             @Override
             public LongObjectOpenHashMap newInstance(int sizing) {
                 return new LongObjectOpenHashMap(size(sizing));
@@ -111,7 +115,7 @@ public void clear(LongObjectOpenHashMap value) {
                 value.clear();
             }
         });
-        longLongMap = build(type, limit, smartSize, new Recycler.C<LongLongOpenHashMap>() {
+        longLongMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<LongLongOpenHashMap>() {
             @Override
             public LongLongOpenHashMap newInstance(int sizing) {
                 return new LongLongOpenHashMap(size(sizing));
@@ -122,7 +126,7 @@ public void clear(LongLongOpenHashMap value) {
                 value.clear();
             }
         });
-        intIntMap = build(type, limit, smartSize, new Recycler.C<IntIntOpenHashMap>() {
+        intIntMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<IntIntOpenHashMap>() {
             @Override
             public IntIntOpenHashMap newInstance(int sizing) {
                 return new IntIntOpenHashMap(size(sizing));
@@ -133,7 +137,7 @@ public void clear(IntIntOpenHashMap value) {
                 value.clear();
             }
         });
-        floatIntMap = build(type, limit, smartSize, new Recycler.C<FloatIntOpenHashMap>() {
+        floatIntMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<FloatIntOpenHashMap>() {
             @Override
             public FloatIntOpenHashMap newInstance(int sizing) {
                 return new FloatIntOpenHashMap(size(sizing));
@@ -144,7 +148,7 @@ public void clear(FloatIntOpenHashMap value) {
                 value.clear();
             }
         });
-        doubleIntMap = build(type, limit, smartSize, new Recycler.C<DoubleIntOpenHashMap>() {
+        doubleIntMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<DoubleIntOpenHashMap>() {
             @Override
             public DoubleIntOpenHashMap newInstance(int sizing) {
                 return new DoubleIntOpenHashMap(size(sizing));
@@ -155,7 +159,7 @@ public void clear(DoubleIntOpenHashMap value) {
                 value.clear();
             }
         });
-        longIntMap = build(type, limit, smartSize, new Recycler.C<LongIntOpenHashMap>() {
+        longIntMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<LongIntOpenHashMap>() {
             @Override
             public LongIntOpenHashMap newInstance(int sizing) {
                 return new LongIntOpenHashMap(size(sizing));
@@ -166,7 +170,7 @@ public void clear(LongIntOpenHashMap value) {
                 value.clear();
             }
         });
-        objectIntMap = build(type, limit, smartSize, new Recycler.C<ObjectIntOpenHashMap>() {
+        objectIntMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<ObjectIntOpenHashMap>() {
             @Override
             public ObjectIntOpenHashMap newInstance(int sizing) {
                 return new ObjectIntOpenHashMap(size(sizing));
@@ -177,7 +181,7 @@ public void clear(ObjectIntOpenHashMap value) {
                 value.clear();
             }
         });
-        intObjectMap = build(type, limit, smartSize, new Recycler.C<IntObjectOpenHashMap>() {
+        intObjectMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<IntObjectOpenHashMap>() {
             @Override
             public IntObjectOpenHashMap newInstance(int sizing) {
                 return new IntObjectOpenHashMap(size(sizing));
@@ -188,7 +192,7 @@ public void clear(IntObjectOpenHashMap value) {
                 value.clear();
             }
         });
-        objectFloatMap = build(type, limit, smartSize, new Recycler.C<ObjectFloatOpenHashMap>() {
+        objectFloatMap = build(type, limit, smartSize, availableProcessors, new Recycler.C<ObjectFloatOpenHashMap>() {
             @Override
             public ObjectFloatOpenHashMap newInstance(int sizing) {
                 return new ObjectFloatOpenHashMap(size(sizing));
@@ -253,12 +257,12 @@ static int size(int sizing) {
         return sizing > 0 ? sizing : 256;
     }
 
-    private <T> Recycler<T> build(Type type, int limit, int smartSize, Recycler.C<T> c) {
+    private <T> Recycler<T> build(Type type, int limit, int smartSize, int availableProcessors, Recycler.C<T> c) {
         Recycler<T> recycler;
         try {
-            recycler = type.build(c, limit);
+            recycler = type.build(c, limit, availableProcessors);
             if (smartSize > 0) {
-                recycler = new Recycler.Sizing<T>(recycler, smartSize);
+                recycler = sizing(recycler, none(c), smartSize);
             }
         } catch (IllegalArgumentException ex) {
             throw new ElasticsearchIllegalArgumentException("no type support [" + type + "] for recycler");
@@ -270,40 +274,44 @@ private <T> Recycler<T> build(Type type, int limit, int smartSize, Recycler.C<T>
     public static enum Type {
         SOFT_THREAD_LOCAL {
             @Override
-            <T> Recycler<T> build(Recycler.C<T> c, int limit) {
-                return new SoftThreadLocalRecycler<T>(c, limit);
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int availableProcessors) {
+                return threadLocal(softFactory(dequeFactory(c, limit)));
             }
+        },
+        THREAD_LOCAL {
             @Override
-            boolean perThread() {
-                return true;
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int availableProcessors) {
+                return threadLocal(dequeFactory(c, limit));
             }
         },
-        THREAD_LOCAL {
+        QUEUE {
             @Override
-            <T> Recycler<T> build(Recycler.C<T> c, int limit) {
-                return new ThreadLocalRecycler<T>(c, limit);
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int availableProcessors) {
+                return concurrentDeque(c, limit);
             }
+        },
+        SOFT_CONCURRENT {
             @Override
-            boolean perThread() {
-                return true;
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int availableProcessors) {
+                return concurrent(softFactory(dequeFactory(c, limit)), availableProcessors);
             }
         },
-        QUEUE {
+        CONCURRENT {
             @Override
-            <T> Recycler<T> build(Recycler.C<T> c, int limit) {
-                return new QueueRecycler<T>(c, limit);
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int availableProcessors) {
+                return concurrent(dequeFactory(c, limit), availableProcessors);
             }
         },
         NONE {
             @Override
-            <T> Recycler<T> build(Recycler.C<T> c, int limit) {
-                return new NoneRecycler<T>(c);
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int availableProcessors) {
+                return none(c);
             }
         };
 
         public static Type parse(String type) {
             if (Strings.isNullOrEmpty(type)) {
-                return SOFT_THREAD_LOCAL;
+                return SOFT_CONCURRENT;
             }
             try {
                 return Type.valueOf(type.toUpperCase(Locale.ROOT));
@@ -312,9 +320,6 @@ public static Type parse(String type) {
             }
         }
 
-        abstract <T> Recycler<T> build(Recycler.C<T> c, int limit);
-        boolean perThread() {
-            return false;
-        }
+        abstract <T> Recycler<T> build(Recycler.C<T> c, int limit, int availableProcessors);
     }
 }

src/main/java/org/elasticsearch/cache/recycler/PageCacheRecycler.java

@@ -19,18 +19,20 @@
 
 package org.elasticsearch.cache.recycler;
 
-import org.elasticsearch.cache.recycler.CacheRecycler.Type;
+import com.google.common.base.Strings;
+import org.elasticsearch.ElasticsearchIllegalArgumentException;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.inject.Inject;
-import org.elasticsearch.common.recycler.NoneRecycler;
 import org.elasticsearch.common.recycler.Recycler;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.Arrays;
+import java.util.Locale;
+
+import static org.elasticsearch.common.recycler.Recyclers.*;
 
 /** A recycler of fixed-size pages. */
 public class PageCacheRecycler extends AbstractComponent {
@@ -79,17 +81,9 @@ private static int maxCount(long limit, long pageSize, double weight, double tot
     public PageCacheRecycler(Settings settings, ThreadPool threadPool) {
         super(settings);
         final Type type = Type.parse(componentSettings.get(TYPE));
-        long limit = componentSettings.getAsMemory(LIMIT_HEAP, "10%").bytes();
-        if (type.perThread()) {
-            final long limitPerThread = componentSettings.getAsBytesSize(LIMIT_PER_THREAD, new ByteSizeValue(-1)).bytes();
-            if (limitPerThread != -1) {
-                // if the per_thread limit is set, it has precedence
-                limit = limitPerThread;
-            } else {
-                // divide memory equally to all search threads
-                limit /= maximumSearchThreadPoolSize(threadPool, settings);
-            }
-        }
+        final long limit = componentSettings.getAsMemory(LIMIT_HEAP, "10%").bytes();
+        final int availableProcessors = EsExecutors.boundedNumberOfProcessors(settings);
+        final int searchThreadPoolSize = maximumSearchThreadPoolSize(threadPool, settings);
 
         // We have a global amount of memory that we need to divide across data types.
         // Since some types are more useful than other ones we give them different weights.
@@ -113,39 +107,39 @@ public PageCacheRecycler(Settings settings, ThreadPool threadPool) {
 
         final double totalWeight = bytesWeight + intsWeight + longsWeight + doublesWeight + objectsWeight;
 
-        bytePage = build(type, maxCount(limit, BigArrays.BYTE_PAGE_SIZE, bytesWeight, totalWeight), new Recycler.C<byte[]>() {
+        bytePage = build(type, maxCount(limit, BigArrays.BYTE_PAGE_SIZE, bytesWeight, totalWeight), searchThreadPoolSize, availableProcessors, new Recycler.C<byte[]>() {
             @Override
             public byte[] newInstance(int sizing) {
                 return new byte[BigArrays.BYTE_PAGE_SIZE];
             }
             @Override
             public void clear(byte[] value) {}
         });
-        intPage = build(type, maxCount(limit, BigArrays.INT_PAGE_SIZE, intsWeight, totalWeight), new Recycler.C<int[]>() {
+        intPage = build(type, maxCount(limit, BigArrays.INT_PAGE_SIZE, intsWeight, totalWeight), searchThreadPoolSize, availableProcessors, new Recycler.C<int[]>() {
             @Override
             public int[] newInstance(int sizing) {
                 return new int[BigArrays.INT_PAGE_SIZE];
             }
             @Override
             public void clear(int[] value) {}
         });
-        longPage = build(type, maxCount(limit, BigArrays.LONG_PAGE_SIZE, longsWeight, totalWeight), new Recycler.C<long[]>() {
+        longPage = build(type, maxCount(limit, BigArrays.LONG_PAGE_SIZE, longsWeight, totalWeight), searchThreadPoolSize, availableProcessors, new Recycler.C<long[]>() {
             @Override
             public long[] newInstance(int sizing) {
                 return new long[BigArrays.LONG_PAGE_SIZE];
             }
             @Override
             public void clear(long[] value) {}
         });
-        doublePage = build(type, maxCount(limit, BigArrays.DOUBLE_PAGE_SIZE, doublesWeight, totalWeight), new Recycler.C<double[]>() {
+        doublePage = build(type, maxCount(limit, BigArrays.DOUBLE_PAGE_SIZE, doublesWeight, totalWeight), searchThreadPoolSize, availableProcessors, new Recycler.C<double[]>() {
             @Override
             public double[] newInstance(int sizing) {
                 return new double[BigArrays.DOUBLE_PAGE_SIZE];
             }
             @Override
             public void clear(double[] value) {}
         });
-        objectPage = build(type, maxCount(limit, BigArrays.OBJECT_PAGE_SIZE, objectsWeight, totalWeight), new Recycler.C<Object[]>() {
+        objectPage = build(type, maxCount(limit, BigArrays.OBJECT_PAGE_SIZE, objectsWeight, totalWeight), searchThreadPoolSize, availableProcessors, new Recycler.C<Object[]>() {
             @Override
             public Object[] newInstance(int sizing) {
                 return new Object[BigArrays.OBJECT_PAGE_SIZE];
@@ -194,13 +188,65 @@ public Recycler.V<Object[]> objectPage() {
         return objectPage.obtain();
     }
 
-    private static <T> Recycler<T> build(Type type, int limit, Recycler.C<T> c) {
+    private static <T> Recycler<T> build(Type type, int limit, int estimatedThreadPoolSize, int availableProcessors, Recycler.C<T> c) {
         final Recycler<T> recycler;
         if (limit == 0) {
-            recycler = new NoneRecycler<T>(c);
+            recycler = none(c);
         } else {
-            recycler = type.build(c, limit);
+            recycler = type.build(c, limit, estimatedThreadPoolSize, availableProcessors);
         }
         return recycler;
     }
+
+    public static enum Type {
+        SOFT_THREAD_LOCAL {
+            @Override
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int estimatedThreadPoolSize, int availableProcessors) {
+                return threadLocal(softFactory(dequeFactory(c, limit / estimatedThreadPoolSize)));
+            }
+        },
+        THREAD_LOCAL {
+            @Override
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int estimatedThreadPoolSize, int availableProcessors) {
+                return threadLocal(dequeFactory(c, limit / estimatedThreadPoolSize));
+            }
+        },
+        QUEUE {
+            @Override
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int estimatedThreadPoolSize, int availableProcessors) {
+                return concurrentDeque(c, limit);
+            }
+        },
+        SOFT_CONCURRENT {
+            @Override
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int estimatedThreadPoolSize, int availableProcessors) {
+                return concurrent(softFactory(dequeFactory(c, limit / availableProcessors)), availableProcessors);
+            }
+        },
+        CONCURRENT {
+            @Override
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int estimatedThreadPoolSize, int availableProcessors) {
+                return concurrent(dequeFactory(c, limit / availableProcessors), availableProcessors);
+            }
+        },
+        NONE {
+            @Override
+            <T> Recycler<T> build(Recycler.C<T> c, int limit, int estimatedThreadPoolSize, int availableProcessors) {
+                return none(c);
+            }
+        };
+
+        public static Type parse(String type) {
+            if (Strings.isNullOrEmpty(type)) {
+                return SOFT_CONCURRENT;
+            }
+            try {
+                return Type.valueOf(type.toUpperCase(Locale.ROOT));
+            } catch (IllegalArgumentException e) {
+                throw new ElasticsearchIllegalArgumentException("no type support [" + type + "]");
+            }
+        }
+
+        abstract <T> Recycler<T> build(Recycler.C<T> c, int limit, int estimatedThreadPoolSize, int availableProcessors);
+    }
 }