expose the max direct memory allowed in jvm info, and guess better th…

…en receive buffer size predictor size based on it

bin/elasticsearch.bat

@@ -29,6 +29,10 @@ if NOT "%ES_HEAP_NEWSIZE%" == "" (
 set JAVA_OPTS=%JAVA_OPTS% -Xmn%ES_HEAP_NEWSIZE%
 )
 
+if NOT "%ES_DIRECT_SIZE%" == "" (
+set JAVA_OPTS=%JAVA_OPTS% -XX:MaxDirectMemorySize=%ES_DIRECT_SIZE%
+)
+
 set JAVA_OPTS=%JAVA_OPTS% -Xss256k
 
 REM Enable aggressive optimizations in the JVM

bin/elasticsearch.in.sh

@@ -25,6 +25,11 @@ if [ "x$ES_HEAP_NEWSIZE" != "x" ]; then
     JAVA_OPTS="$JAVA_OPTS -Xmn${ES_HEAP_NEWSIZE}"
 fi
 
+# max direct memory
+if [ "x$ES_DIRECT_SIZE" != "x" ]; then
+    JAVA_OPTS="$JAVA_OPTS -XX:MaxDirectMemorySize=${ES_DIRECT_SIZE}"
+fi
+
 # reduce the per-thread stack size
 JAVA_OPTS="$JAVA_OPTS -Xss256k"
 
@@ -34,10 +39,6 @@ JAVA_OPTS="$JAVA_OPTS -Djava.awt.headless=true"
 # Force the JVM to use IPv4 stack
 # JAVA_OPTS="$JAVA_OPTS -Djava.net.preferIPv4Stack=true"
 
-# Enable aggressive optimizations in the JVM
-#    - Disabled by default as it might cause the JVM to crash
-# JAVA_OPTS="$JAVA_OPTS -XX:+AggressiveOpts"
-
 JAVA_OPTS="$JAVA_OPTS -XX:+UseParNewGC"
 JAVA_OPTS="$JAVA_OPTS -XX:+UseConcMarkSweepGC"
 

src/deb/default/elasticsearch

@@ -5,6 +5,12 @@
 # Heap Size (defaults to 256m min, 1g max)
 #ES_HEAP_SIZE=2g
 
+# Heap new generation
+#ES_HEAP_NEWSIZE=
+
+# max direct memory
+#ES_DIRECT_SIZE=
+
 # Maximum number of open files, defaults to 65535.
 #MAX_OPEN_FILES=65535
 

src/deb/init.d/elasticsearch

@@ -62,6 +62,12 @@ ES_HOME=/usr/share/$NAME
 # Heap Size (defaults to 256m min, 1g max)
 #ES_HEAP_SIZE=2g
 
+# Heap new generation
+#ES_HEAP_NEWSIZE=
+
+# max direct memory
+#ES_DIRECT_SIZE=
+
 # Additional Java OPTS
 #ES_JAVA_OPTS=
 
@@ -99,6 +105,8 @@ DAEMON=$ES_HOME/bin/elasticsearch
 DAEMON_OPTS="-p $PID_FILE -Des.default.config=$CONF_FILE -Des.default.path.home=$ES_HOME -Des.default.path.logs=$LOG_DIR -Des.default.path.data=$DATA_DIR -Des.default.path.work=$WORK_DIR -Des.default.path.conf=$CONF_DIR"
 
 export ES_HEAP_SIZE
+export ES_HEAP_NEWSIZE
+export ES_DIRECT_SIZE
 export ES_JAVA_OPTS
 
 # Check DAEMON exists

src/main/java/org/elasticsearch/http/netty/NettyHttpServerTransport.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.http.*;
 import org.elasticsearch.http.HttpRequest;
+import org.elasticsearch.monitor.jvm.JvmInfo;
 import org.elasticsearch.transport.BindTransportException;
 import org.jboss.netty.bootstrap.ServerBootstrap;
 import org.jboss.netty.channel.*;
@@ -131,9 +132,16 @@ public NettyHttpServerTransport(Settings settings, NetworkService networkService
         this.tcpSendBufferSize = componentSettings.getAsBytesSize("tcp_send_buffer_size", settings.getAsBytesSize(TCP_SEND_BUFFER_SIZE, TCP_DEFAULT_SEND_BUFFER_SIZE));
         this.tcpReceiveBufferSize = componentSettings.getAsBytesSize("tcp_receive_buffer_size", settings.getAsBytesSize(TCP_RECEIVE_BUFFER_SIZE, TCP_DEFAULT_RECEIVE_BUFFER_SIZE));
 
+        long defaultReceiverPredictor = 512 * 1024;
+        if (JvmInfo.jvmInfo().mem().directMemoryMax().bytes() > 0) {
+            // we can guess a better default...
+            long l = (long) ((0.3 * JvmInfo.jvmInfo().mem().directMemoryMax().bytes()) / workerCount);
+            defaultReceiverPredictor = Math.min(defaultReceiverPredictor, Math.max(l, 64 * 1024));
+        }
+
         // See AdaptiveReceiveBufferSizePredictor#DEFAULT_XXX for default values in netty..., we can use higher ones for us, even fixed one
-        ByteSizeValue receivePredictorMin = componentSettings.getAsBytesSize("receive_predictor_min", componentSettings.getAsBytesSize("receive_predictor_size", ByteSizeValue.parseBytesSizeValue("512k")));
-        ByteSizeValue receivePredictorMax = componentSettings.getAsBytesSize("receive_predictor_max", componentSettings.getAsBytesSize("receive_predictor_size", ByteSizeValue.parseBytesSizeValue("512k")));
+        ByteSizeValue receivePredictorMin = componentSettings.getAsBytesSize("receive_predictor_min", componentSettings.getAsBytesSize("receive_predictor_size", new ByteSizeValue(defaultReceiverPredictor)));
+        ByteSizeValue receivePredictorMax = componentSettings.getAsBytesSize("receive_predictor_max", componentSettings.getAsBytesSize("receive_predictor_size", new ByteSizeValue(defaultReceiverPredictor)));
         if (receivePredictorMax.bytes() == receivePredictorMin.bytes()) {
             receiveBufferSizePredictorFactory = new FixedReceiveBufferSizePredictorFactory((int) receivePredictorMax.bytes());
         } else {
@@ -150,8 +158,8 @@ public NettyHttpServerTransport(Settings settings, NetworkService networkService
         }
         this.maxContentLength = maxContentLength;
 
-        logger.debug("using max_chunk_size[{}], max_header_size[{}], max_initial_line_length[{}], max_content_length[{}]",
-                maxChunkSize, maxHeaderSize, maxInitialLineLength, this.maxContentLength);
+        logger.debug("using max_chunk_size[{}], max_header_size[{}], max_initial_line_length[{}], max_content_length[{}], receive_predictor[{}->{}]",
+                maxChunkSize, maxHeaderSize, maxInitialLineLength, this.maxContentLength, receivePredictorMin, receivePredictorMax);
     }
 
     public void httpServerAdapter(HttpServerAdapter httpServerAdapter) {

src/main/java/org/elasticsearch/monitor/jvm/JvmInfo.java

@@ -67,6 +67,12 @@ public class JvmInfo implements Streamable, Serializable, ToXContent {
         info.mem.heapMax = memoryMXBean.getHeapMemoryUsage().getMax() < 0 ? 0 : memoryMXBean.getHeapMemoryUsage().getMax();
         info.mem.nonHeapInit = memoryMXBean.getNonHeapMemoryUsage().getInit() < 0 ? 0 : memoryMXBean.getNonHeapMemoryUsage().getInit();
         info.mem.nonHeapMax = memoryMXBean.getNonHeapMemoryUsage().getMax() < 0 ? 0 : memoryMXBean.getNonHeapMemoryUsage().getMax();
+        try {
+            Class<?> vmClass = Class.forName("sun.misc.VM");
+            info.mem.directMemoryMax = (Long) vmClass.getMethod("maxDirectMemory").invoke(null);
+        } catch (Throwable t) {
+            // ignore
+        }
         info.inputArguments = runtimeMXBean.getInputArguments().toArray(new String[runtimeMXBean.getInputArguments().size()]);
         info.bootClassPath = runtimeMXBean.getBootClassPath();
         info.classPath = runtimeMXBean.getClassPath();
@@ -276,6 +282,8 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.field(Fields.NON_HEAP_INIT_IN_BYTES, mem.nonHeapInit);
         builder.field(Fields.NON_HEAP_MAX, mem.nonHeapMax().toString());
         builder.field(Fields.NON_HEAP_MAX_IN_BYTES, mem.nonHeapMax);
+        builder.field(Fields.DIRECT_MAX, mem.directMemoryMax().toString());
+        builder.field(Fields.DIRECT_MAX_IN_BYTES, mem.directMemoryMax().bytes());
         builder.endObject();
 
         builder.endObject();
@@ -300,6 +308,8 @@ static final class Fields {
         static final XContentBuilderString NON_HEAP_INIT_IN_BYTES = new XContentBuilderString("non_heap_init_in_bytes");
         static final XContentBuilderString NON_HEAP_MAX = new XContentBuilderString("non_heap_max");
         static final XContentBuilderString NON_HEAP_MAX_IN_BYTES = new XContentBuilderString("non_heap_max_in_bytes");
+        static final XContentBuilderString DIRECT_MAX = new XContentBuilderString("direct_max");
+        static final XContentBuilderString DIRECT_MAX_IN_BYTES = new XContentBuilderString("direct_max_in_bytes");
     }
 
     public static JvmInfo readJvmInfo(StreamInput in) throws IOException {
@@ -359,6 +369,7 @@ public static class Mem implements Streamable, Serializable {
         long heapMax = 0;
         long nonHeapInit = 0;
         long nonHeapMax = 0;
+        long directMemoryMax = 0;
 
         Mem() {
         }
@@ -395,6 +406,14 @@ public ByteSizeValue getNonHeapMax() {
             return nonHeapMax();
         }
 
+        public ByteSizeValue directMemoryMax() {
+            return new ByteSizeValue(directMemoryMax);
+        }
+
+        public ByteSizeValue getDirectMemoryMax() {
+            return directMemoryMax();
+        }
+
         public static Mem readMem(StreamInput in) throws IOException {
             Mem mem = new Mem();
             mem.readFrom(in);
@@ -407,6 +426,7 @@ public void readFrom(StreamInput in) throws IOException {
             heapMax = in.readVLong();
             nonHeapInit = in.readVLong();
             nonHeapMax = in.readVLong();
+            directMemoryMax = in.readVLong();
         }
 
         @Override
@@ -415,6 +435,7 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeVLong(heapMax);
             out.writeVLong(nonHeapInit);
             out.writeVLong(nonHeapMax);
+            out.writeVLong(directMemoryMax);
         }
     }
 }

src/main/java/org/elasticsearch/transport/netty/NettyTransport.java

@@ -40,6 +40,7 @@
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.monitor.jvm.JvmInfo;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.*;
 import org.elasticsearch.transport.support.TransportStreams;
@@ -176,17 +177,24 @@ public NettyTransport(Settings settings, ThreadPool threadPool, NetworkService n
         this.maxCumulationBufferCapacity = componentSettings.getAsBytesSize("max_cumulation_buffer_capacity", null);
         this.maxCompositeBufferComponents = componentSettings.getAsInt("max_composite_buffer_components", -1);
 
+        long defaultReceiverPredictor = 512 * 1024;
+        if (JvmInfo.jvmInfo().mem().directMemoryMax().bytes() > 0) {
+            // we can guess a better default...
+            long l = (long) ((0.3 * JvmInfo.jvmInfo().mem().directMemoryMax().bytes()) / workerCount);
+            defaultReceiverPredictor = Math.min(defaultReceiverPredictor, Math.max(l, 64 * 1024));
+        }
+
         // See AdaptiveReceiveBufferSizePredictor#DEFAULT_XXX for default values in netty..., we can use higher ones for us, even fixed one
-        ByteSizeValue receivePredictorMin = componentSettings.getAsBytesSize("receive_predictor_min", componentSettings.getAsBytesSize("receive_predictor_size", ByteSizeValue.parseBytesSizeValue("512k")));
-        ByteSizeValue receivePredictorMax = componentSettings.getAsBytesSize("receive_predictor_max", componentSettings.getAsBytesSize("receive_predictor_size", ByteSizeValue.parseBytesSizeValue("512k")));
+        ByteSizeValue receivePredictorMin = componentSettings.getAsBytesSize("receive_predictor_min", componentSettings.getAsBytesSize("receive_predictor_size", new ByteSizeValue(defaultReceiverPredictor)));
+        ByteSizeValue receivePredictorMax = componentSettings.getAsBytesSize("receive_predictor_max", componentSettings.getAsBytesSize("receive_predictor_size", new ByteSizeValue(defaultReceiverPredictor)));
         if (receivePredictorMax.bytes() == receivePredictorMin.bytes()) {
             receiveBufferSizePredictorFactory = new FixedReceiveBufferSizePredictorFactory((int) receivePredictorMax.bytes());
         } else {
             receiveBufferSizePredictorFactory = new AdaptiveReceiveBufferSizePredictorFactory((int) receivePredictorMin.bytes(), (int) receivePredictorMin.bytes(), (int) receivePredictorMax.bytes());
         }
 
-        logger.debug("using worker_count[{}], port[{}], bind_host[{}], publish_host[{}], compress[{}], connect_timeout[{}], connections_per_node[{}/{}/{}]",
-                workerCount, port, bindHost, publishHost, compress, connectTimeout, connectionsPerNodeLow, connectionsPerNodeMed, connectionsPerNodeHigh);
+        logger.debug("using worker_count[{}], port[{}], bind_host[{}], publish_host[{}], compress[{}], connect_timeout[{}], connections_per_node[{}/{}/{}], receive_predictor[{}->{}]",
+                workerCount, port, bindHost, publishHost, compress, connectTimeout, connectionsPerNodeLow, connectionsPerNodeMed, connectionsPerNodeHigh, receivePredictorMin, receivePredictorMax);
     }
 
     public Settings settings() {