eagerly allocate the intermediate computation buffers (#3628)

benchmarks/src/main/java/io/druid/benchmark/query/GroupByBenchmark.java

@@ -309,6 +309,7 @@ public class GroupByBenchmark
 
     StupidPool<ByteBuffer> bufferPool = new StupidPool<>(
         new OffheapBufferGenerator("compute", 250_000_000),
+        0,
         Integer.MAX_VALUE
     );
 

benchmarks/src/main/java/io/druid/benchmark/query/TopNBenchmark.java

@@ -272,7 +272,7 @@ public class TopNBenchmark
     }
 
     factory = new TopNQueryRunnerFactory(
-        new StupidPool<>(new OffheapBufferGenerator("compute", 250000000), Integer.MAX_VALUE),
+        new StupidPool<>(new OffheapBufferGenerator("compute", 250000000), 0, Integer.MAX_VALUE),
         new TopNQueryQueryToolChest(new TopNQueryConfig(), QueryBenchmarkUtil.NoopIntervalChunkingQueryRunnerDecorator()),
         QueryBenchmarkUtil.NOOP_QUERYWATCHER
     );

common/src/main/java/io/druid/collections/StupidPool.java

@@ -19,8 +19,8 @@
 
 package io.druid.collections;
 
+import com.google.common.base.Preconditions;
 import com.google.common.base.Supplier;
-
 import io.druid.java.util.common.ISE;
 import io.druid.java.util.common.logger.Logger;
 
@@ -45,17 +45,27 @@ public class StupidPool<T>
       Supplier<T> generator
   )
   {
-    this.generator = generator;
-    this.objectsCacheMaxCount = Integer.MAX_VALUE;
+    this(generator, 0, Integer.MAX_VALUE);
   }
 
   public StupidPool(
       Supplier<T> generator,
+      int initCount,
       int objectsCacheMaxCount
   )
   {
+    Preconditions.checkArgument(
+        initCount <= objectsCacheMaxCount,
+        "initCount[%s] must be less/equal to objectsCacheMaxCount[%s]",
+        initCount,
+        objectsCacheMaxCount
+    );
     this.generator = generator;
     this.objectsCacheMaxCount = objectsCacheMaxCount;
+
+    for (int i = 0; i < initCount; i++) {
+      objects.add(generator.get());
+    }
   }
 
   public ResourceHolder<T> take()

server/src/main/java/io/druid/guice/DruidProcessingModule.java

@@ -110,6 +110,7 @@ public class DruidProcessingModule implements Module
     verifyDirectMemory(config);
     return new StupidPool<>(
         new OffheapBufferGenerator("intermediate processing", config.intermediateComputeSizeBytes()),
+        config.getNumThreads(),
         config.poolCacheMaxCount()
     );
   }