Merge pull request #456 from metamx/processing-threads-default

more sensible defaults

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

@@ -31,17 +31,15 @@ import com.metamx.common.logger.Logger;
 import com.metamx.emitter.service.ServiceEmitter;
 import com.metamx.emitter.service.ServiceMetricEvent;
 import io.druid.collections.StupidPool;
-import io.druid.concurrent.Execs;
 import io.druid.guice.annotations.Global;
 import io.druid.guice.annotations.Processing;
 import io.druid.query.MetricsEmittingExecutorService;
 import io.druid.query.PrioritizedExecutorService;
 import io.druid.server.DruidProcessingConfig;
+import io.druid.server.VMUtils;
 
-import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -82,37 +80,26 @@ public class DruidProcessingModule implements Module
   public StupidPool<ByteBuffer> getIntermediateResultsPool(DruidProcessingConfig config)
   {
     try {
-      Class<?> vmClass = Class.forName("sun.misc.VM");
+      long maxDirectMemory = VMUtils.getMaxDirectMemory();
-      Object maxDirectMemoryObj = vmClass.getMethod("maxDirectMemory").invoke(null);
 
-      if (maxDirectMemoryObj == null || !(maxDirectMemoryObj instanceof Number)) {
+      final long memoryNeeded = (long) config.intermediateComputeSizeBytes() * (config.getNumThreads() + 1);
-        log.info("Cannot determine maxDirectMemory from[%s]", maxDirectMemoryObj);
+      if (maxDirectMemory < memoryNeeded) {
-      } else {
+        throw new ProvisionException(
-        long maxDirectMemory = ((Number) maxDirectMemoryObj).longValue();
+            String.format(
-
+                "Not enough direct memory.  Please adjust -XX:MaxDirectMemorySize, druid.computation.buffer.size, or druid.processing.numThreads: "
-        final long memoryNeeded = (long) config.intermediateComputeSizeBytes() * (config.getNumThreads() + 1);
+                + "maxDirectMemory[%,d], memoryNeeded[%,d] = druid.computation.buffer.size[%,d] * ( druid.processing.numThreads[%,d] + 1 )",
-        if (maxDirectMemory < memoryNeeded) {
+                maxDirectMemory,
-          throw new ProvisionException(
+                memoryNeeded,
-              String.format(
+                config.intermediateComputeSizeBytes(),
-                  "Not enough direct memory.  Please adjust -XX:MaxDirectMemorySize or druid.computation.buffer.size: "
+                config.getNumThreads()
-                  + "maxDirectMemory[%,d], memoryNeeded[%,d], druid.computation.buffer.size[%,d], numThreads[%,d]",
+            )
-                  maxDirectMemory, memoryNeeded, config.intermediateComputeSizeBytes(), config.getNumThreads()
+        );
-              )
-          );
-        }
       }
+    } catch(UnsupportedOperationException e) {
+      log.info(e.getMessage());
     }
-    catch (ClassNotFoundException e) {
+    catch(RuntimeException e) {
-      log.info("No VM class, cannot do memory check.");
+      log.warn(e, e.getMessage());
-    }
-    catch (NoSuchMethodException e) {
-      log.info("VM.maxDirectMemory doesn't exist, cannot do memory check.");
-    }
-    catch (InvocationTargetException e) {
-      log.warn(e, "static method shouldn't throw this");
-    }
-    catch (IllegalAccessException e) {
-      log.warn(e, "public method, shouldn't throw this");
     }
 
     return new IntermediateProcessingBufferPool(config.intermediateComputeSizeBytes());

server/src/main/java/io/druid/server/DruidProcessingConfig.java

@@ -31,4 +31,12 @@ public abstract class DruidProcessingConfig extends ExecutorServiceConfig
   {
     return 1024 * 1024 * 1024;
   }
+
+  @Override @Config(value = "${base_path}.numThreads")
+  public int getNumThreads()
+  {
+    // default to leaving one core for background tasks
+    final int processors = Runtime.getRuntime().availableProcessors();
+    return processors > 1 ? processors - 1 : processors;
+  }
 }

server/src/main/java/io/druid/server/VMUtils.java

@@ -0,0 +1,51 @@
+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2012, 2013, 2014  Metamarkets Group Inc.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
+
+package io.druid.server;
+
+import java.lang.reflect.InvocationTargetException;
+
+public class VMUtils
+{
+  public static long getMaxDirectMemory() throws UnsupportedOperationException
+  {
+    try {
+      Class<?> vmClass = Class.forName("sun.misc.VM");
+      Object maxDirectMemoryObj = vmClass.getMethod("maxDirectMemory").invoke(null);
+
+      if (maxDirectMemoryObj == null || !(maxDirectMemoryObj instanceof Number)) {
+        throw new UnsupportedOperationException(String.format("Cannot determine maxDirectMemory from [%s]", maxDirectMemoryObj));
+      } else {
+        return ((Number) maxDirectMemoryObj).longValue();
+      }
+    }
+    catch (ClassNotFoundException e) {
+      throw new UnsupportedOperationException("No VM class, cannot do memory check.", e);
+    }
+    catch (NoSuchMethodException e) {
+      throw new UnsupportedOperationException("VM.maxDirectMemory doesn't exist, cannot do memory check.", e);
+    }
+    catch (InvocationTargetException e) {
+      throw new RuntimeException("static method shouldn't throw this", e);
+    }
+    catch (IllegalAccessException e) {
+      throw new RuntimeException("public method, shouldn't throw this", e);
+    }
+  }
+}

services/src/main/java/io/druid/cli/CliRealtimeExample.java

@@ -42,7 +42,7 @@ import java.util.concurrent.Executor;
  */
 @Command(
     name = "realtime",
-    description = "Runs a standalone realtime node for examples, see http://druid.io/docs/0.6.73/Realtime.html for a description"
+    description = "Runs a standalone realtime node for examples, see http://druid.io/docs/latest/Realtime.html for a description"
 )
 public class CliRealtimeExample extends ServerRunnable
 {