Remove option to enable direct buffer pooling

buildSrc/src/main/groovy/org/elasticsearch/gradle/BuildPlugin.groovy

@@ -889,7 +889,6 @@ class BuildPlugin implements Plugin<Project> {
                 test.systemProperty('io.netty.noUnsafe', 'true')
                 test.systemProperty('io.netty.noKeySetOptimization', 'true')
                 test.systemProperty('io.netty.recycler.maxCapacityPerThread', '0')
-                test.systemProperty('io.netty.allocator.numDirectArenas', '0')
 
                 test.testLogging { TestLoggingContainer logging ->
                     logging.showExceptions = true

distribution/src/config/jvm.options

@@ -80,7 +80,6 @@
 -Dio.netty.noUnsafe=true
 -Dio.netty.noKeySetOptimization=true
 -Dio.netty.recycler.maxCapacityPerThread=0
--Dio.netty.allocator.numDirectArenas=0
 
 # log4j 2
 -Dlog4j.shutdownHookEnabled=false

distribution/tools/launchers/src/main/java/org/elasticsearch/tools/launchers/JvmErgonomics.java

@@ -55,14 +55,6 @@ static List<String> choose(final List<String> userDefinedJvmOptions) throws Inte
         final List<String> ergonomicChoices = new ArrayList<>();
         final Map<String, Optional<String>> finalJvmOptions = finalJvmOptions(userDefinedJvmOptions);
         final long heapSize = extractHeapSize(finalJvmOptions);
-        final Map<String, String> systemProperties = extractSystemProperties(userDefinedJvmOptions);
-        if (systemProperties.containsKey("io.netty.allocator.type") == false) {
-            if (heapSize <= 1 << 30) {
-                ergonomicChoices.add("-Dio.netty.allocator.type=unpooled");
-            } else {
-                ergonomicChoices.add("-Dio.netty.allocator.type=pooled");
-            }
-        }
         final long maxDirectMemorySize = extractMaxDirectMemorySize(finalJvmOptions);
         if (maxDirectMemorySize == 0) {
             ergonomicChoices.add("-XX:MaxDirectMemorySize=" + heapSize / 2);

distribution/tools/launchers/src/test/java/org/elasticsearch/tools/launchers/JvmErgonomicsTests.java

@@ -117,20 +117,6 @@ public void testExtractNoSystemProperties() {
         assertTrue(parsedSystemProperties.isEmpty());
     }
 
-    public void testPooledMemoryChoiceOnSmallHeap() throws InterruptedException, IOException {
-        final String smallHeap = randomFrom(Arrays.asList("64M", "512M", "1024M", "1G"));
-        assertThat(
-                JvmErgonomics.choose(Arrays.asList("-Xms" + smallHeap, "-Xmx" + smallHeap)),
-                hasItem("-Dio.netty.allocator.type=unpooled"));
-    }
-
-    public void testPooledMemoryChoiceOnNotSmallHeap() throws InterruptedException, IOException {
-        final String largeHeap = randomFrom(Arrays.asList("1025M", "2048M", "2G", "8G"));
-        assertThat(
-                JvmErgonomics.choose(Arrays.asList("-Xms" + largeHeap, "-Xmx" + largeHeap)),
-                hasItem("-Dio.netty.allocator.type=pooled"));
-    }
-
     public void testMaxDirectMemorySizeChoice() throws InterruptedException, IOException {
         final Map<String, String> heapMaxDirectMemorySize = Map.of(
                         "64M", Long.toString((64L << 20) / 2),

modules/transport-netty4/build.gradle

@@ -66,7 +66,7 @@ integTestRunner {
 TaskProvider<Test> pooledTest = tasks.register("pooledTest", Test) {
   include '**/*Tests.class'
   systemProperty 'es.set.netty.runtime.available.processors', 'false'
-  systemProperty 'io.netty.allocator.type', 'pooled'
+  systemProperty 'es.use_unpooled_allocator', 'false'
 }
 // TODO: we can't use task avoidance here because RestIntegTestTask does the testcluster creation
 RestIntegTestTask pooledIntegTest = tasks.create("pooledIntegTest", RestIntegTestTask) {
@@ -75,7 +75,7 @@ RestIntegTestTask pooledIntegTest = tasks.create("pooledIntegTest", RestIntegTes
   }
 }
 testClusters.pooledIntegTest {
-  systemProperty 'io.netty.allocator.type', 'pooled'
+  systemProperty 'es.use_unpooled_allocator', 'false'
 }
 check.dependsOn(pooledTest, pooledIntegTest)
 

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpServerTransport.java

@@ -20,7 +20,6 @@
 package org.elasticsearch.http.netty4;
 
 import io.netty.bootstrap.ServerBootstrap;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandler;
@@ -32,7 +31,6 @@
 import io.netty.channel.RecvByteBufAllocator;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioChannelOption;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.codec.http.HttpContentCompressor;
 import io.netty.handler.codec.http.HttpContentDecompressor;
@@ -63,7 +61,7 @@
 import org.elasticsearch.http.HttpServerChannel;
 import org.elasticsearch.http.netty4.cors.Netty4CorsHandler;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.CopyBytesServerSocketChannel;
+import org.elasticsearch.transport.NettyAllocator;
 import org.elasticsearch.transport.netty4.Netty4Utils;
 
 import java.net.InetSocketAddress;
@@ -186,14 +184,12 @@ protected void doStart() {
             serverBootstrap.group(new NioEventLoopGroup(workerCount, daemonThreadFactory(settings,
                 HTTP_SERVER_WORKER_THREAD_NAME_PREFIX)));
 
-            // If direct buffer pooling is disabled, use the CopyBytesServerSocketChannel which will create child
-            // channels of type CopyBytesSocketChannel. CopyBytesSocketChannel pool a single direct buffer
-            // per-event-loop thread to be used for IO operations.
-            if (ByteBufAllocator.DEFAULT.isDirectBufferPooled()) {
-                serverBootstrap.channel(NioServerSocketChannel.class);
-            } else {
-                serverBootstrap.channel(CopyBytesServerSocketChannel.class);
-            }
+            // NettyAllocator will return the channel type designed to work with the configuredAllocator
+            serverBootstrap.channel(NettyAllocator.getServerChannelType());
+
+            // Set the allocators for both the server channel and the child channels created
+            serverBootstrap.option(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator());
+            serverBootstrap.childOption(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator());
 
             serverBootstrap.childHandler(configureServerChannelHandler());
             serverBootstrap.handler(new ServerChannelExceptionHandler(this));

modules/transport-netty4/src/main/java/org/elasticsearch/transport/NettyAllocator.java

@@ -0,0 +1,214 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.transport;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ServerChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import org.elasticsearch.common.Booleans;
+import org.elasticsearch.monitor.jvm.JvmInfo;
+
+public class NettyAllocator {
+
+    private static final ByteBufAllocator ALLOCATOR;
+
+    private static final String USE_UNPOOLED = "es.use_unpooled_allocator";
+    private static final String USE_NETTY_DEFAULT = "es.unsafe.use_netty_default_allocator";
+
+    static {
+        if (Booleans.parseBoolean(System.getProperty(USE_NETTY_DEFAULT), false)) {
+            ALLOCATOR = ByteBufAllocator.DEFAULT;
+        } else {
+            ByteBufAllocator delegate;
+            if (useUnpooled()) {
+                delegate = new NoDirectBuffers(UnpooledByteBufAllocator.DEFAULT);
+            } else {
+                int nHeapArena = PooledByteBufAllocator.defaultNumHeapArena();
+                int pageSize = PooledByteBufAllocator.defaultPageSize();
+                int maxOrder = PooledByteBufAllocator.defaultMaxOrder();
+                int tinyCacheSize = PooledByteBufAllocator.defaultTinyCacheSize();
+                int smallCacheSize = PooledByteBufAllocator.defaultSmallCacheSize();
+                int normalCacheSize = PooledByteBufAllocator.defaultNormalCacheSize();
+                boolean useCacheForAllThreads = PooledByteBufAllocator.defaultUseCacheForAllThreads();
+                delegate = new PooledByteBufAllocator(false, nHeapArena, 0, pageSize, maxOrder, tinyCacheSize,
+                    smallCacheSize, normalCacheSize, useCacheForAllThreads);
+            }
+            ALLOCATOR = new NoDirectBuffers(delegate);
+        }
+    }
+
+    public static boolean useCopySocket() {
+        return ALLOCATOR instanceof NoDirectBuffers;
+    }
+
+    public static ByteBufAllocator getAllocator() {
+        return ALLOCATOR;
+    }
+
+    public static Class<? extends Channel> getChannelType() {
+        if (ALLOCATOR instanceof NoDirectBuffers) {
+            return CopyBytesSocketChannel.class;
+        } else {
+            return NioSocketChannel.class;
+        }
+    }
+
+    public static Class<? extends ServerChannel> getServerChannelType() {
+        if (ALLOCATOR instanceof NoDirectBuffers) {
+            return CopyBytesServerSocketChannel.class;
+        } else {
+            return NioServerSocketChannel.class;
+        }
+    }
+
+    private static boolean useUnpooled() {
+        if (System.getProperty(USE_UNPOOLED) != null) {
+            return Booleans.parseBoolean(System.getProperty(USE_UNPOOLED));
+        } else {
+            long heapSize = JvmInfo.jvmInfo().getMem().getHeapMax().getBytes();
+            return heapSize <= 1 << 30;
+        }
+    }
+
+    private static class NoDirectBuffers implements ByteBufAllocator {
+
+        private final ByteBufAllocator delegate;
+
+        private NoDirectBuffers(ByteBufAllocator delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public ByteBuf buffer() {
+            return heapBuffer();
+        }
+
+        @Override
+        public ByteBuf buffer(int initialCapacity) {
+            return heapBuffer(initialCapacity);
+        }
+
+        @Override
+        public ByteBuf buffer(int initialCapacity, int maxCapacity) {
+            return heapBuffer(initialCapacity, maxCapacity);
+        }
+
+        @Override
+        public ByteBuf ioBuffer() {
+            return heapBuffer();
+        }
+
+        @Override
+        public ByteBuf ioBuffer(int initialCapacity) {
+            return heapBuffer(initialCapacity);
+        }
+
+        @Override
+        public ByteBuf ioBuffer(int initialCapacity, int maxCapacity) {
+            return heapBuffer(initialCapacity, maxCapacity);
+        }
+
+        @Override
+        public ByteBuf heapBuffer() {
+            return delegate.heapBuffer();
+        }
+
+        @Override
+        public ByteBuf heapBuffer(int initialCapacity) {
+            return delegate.heapBuffer(initialCapacity);
+        }
+
+        @Override
+        public ByteBuf heapBuffer(int initialCapacity, int maxCapacity) {
+            return delegate.heapBuffer(initialCapacity, maxCapacity);
+        }
+
+        @Override
+        public ByteBuf directBuffer() {
+            // TODO: Currently the Netty SslHandler requests direct ByteBufs even when interacting with the
+            //  JDK SSLEngine. This will be fixed in a future version of Netty. For now, return a heap
+            //  ByteBuf. After a Netty upgrade, return to throwing UnsupportedOperationException
+            return heapBuffer();
+        }
+
+        @Override
+        public ByteBuf directBuffer(int initialCapacity) {
+            // TODO: Currently the Netty SslHandler requests direct ByteBufs even when interacting with the
+            //  JDK SSLEngine. This will be fixed in a future version of Netty. For now, return a heap
+            //  ByteBuf. After a Netty upgrade, return to throwing UnsupportedOperationException
+            return heapBuffer(initialCapacity);
+        }
+
+        @Override
+        public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
+            // TODO: Currently the Netty SslHandler requests direct ByteBufs even when interacting with the
+            //  JDK SSLEngine. This will be fixed in a future version of Netty. For now, return a heap
+            //  ByteBuf. After a Netty upgrade, return to throwing UnsupportedOperationException
+            return heapBuffer(initialCapacity, maxCapacity);
+        }
+
+        @Override
+        public CompositeByteBuf compositeBuffer() {
+            return compositeHeapBuffer();
+        }
+
+        @Override
+        public CompositeByteBuf compositeBuffer(int maxNumComponents) {
+            return compositeHeapBuffer(maxNumComponents);
+        }
+
+        @Override
+        public CompositeByteBuf compositeHeapBuffer() {
+            return delegate.compositeHeapBuffer();
+        }
+
+        @Override
+        public CompositeByteBuf compositeHeapBuffer(int maxNumComponents) {
+            return delegate.compositeHeapBuffer(maxNumComponents);
+        }
+
+        @Override
+        public CompositeByteBuf compositeDirectBuffer() {
+            throw new UnsupportedOperationException("Direct buffers not supported.");
+        }
+
+        @Override
+        public CompositeByteBuf compositeDirectBuffer(int maxNumComponents) {
+            throw new UnsupportedOperationException("Direct buffers not supported.");
+        }
+
+        @Override
+        public boolean isDirectBufferPooled() {
+            assert delegate.isDirectBufferPooled() == false;
+            return false;
+        }
+
+        @Override
+        public int calculateNewCapacity(int minNewCapacity, int maxCapacity) {
+            return delegate.calculateNewCapacity(minNewCapacity, maxCapacity);
+        }
+    }
+}

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java

@@ -21,7 +21,6 @@
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
-import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.AdaptiveRecvByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
@@ -34,8 +33,6 @@
 import io.netty.channel.RecvByteBufAllocator;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioChannelOption;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
-import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.util.AttributeKey;
 import io.netty.util.concurrent.Future;
 import org.apache.logging.log4j.LogManager;
@@ -59,8 +56,7 @@
 import org.elasticsearch.core.internal.net.NetUtils;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.threadpool.ThreadPool;
-import org.elasticsearch.transport.CopyBytesServerSocketChannel;
-import org.elasticsearch.transport.CopyBytesSocketChannel;
+import org.elasticsearch.transport.NettyAllocator;
 import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.TransportSettings;
 
@@ -152,13 +148,9 @@ private Bootstrap createClientBootstrap(NioEventLoopGroup eventLoopGroup) {
         final Bootstrap bootstrap = new Bootstrap();
         bootstrap.group(eventLoopGroup);
 
-        // If direct buffer pooling is disabled, use the CopyBytesSocketChannel which will pool a single
-        // direct buffer per-event-loop thread which will be used for IO operations.
-        if (ByteBufAllocator.DEFAULT.isDirectBufferPooled()) {
-            bootstrap.channel(NioSocketChannel.class);
-        } else {
-            bootstrap.channel(CopyBytesSocketChannel.class);
-        }
+        // NettyAllocator will return the channel type designed to work with the configured allocator
+        bootstrap.channel(NettyAllocator.getChannelType());
+        bootstrap.option(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator());
 
         bootstrap.option(ChannelOption.TCP_NODELAY, TransportSettings.TCP_NO_DELAY.get(settings));
         bootstrap.option(ChannelOption.SO_KEEPALIVE, TransportSettings.TCP_KEEP_ALIVE.get(settings));
@@ -216,14 +208,12 @@ private void createServerBootstrap(ProfileSettings profileSettings, NioEventLoop
 
         serverBootstrap.group(eventLoopGroup);
 
-        // If direct buffer pooling is disabled, use the CopyBytesServerSocketChannel which will create child
-        // channels of type CopyBytesSocketChannel. CopyBytesSocketChannel pool a single direct buffer
-        // per-event-loop thread to be used for IO operations.
-        if (ByteBufAllocator.DEFAULT.isDirectBufferPooled()) {
-            serverBootstrap.channel(NioServerSocketChannel.class);
-        } else {
-            serverBootstrap.channel(CopyBytesServerSocketChannel.class);
-        }
+        // NettyAllocator will return the channel type designed to work with the configuredAllocator
+        serverBootstrap.channel(NettyAllocator.getServerChannelType());
+
+        // Set the allocators for both the server channel and the child channels created
+        serverBootstrap.option(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator());
+        serverBootstrap.childOption(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator());
 
         serverBootstrap.childHandler(getServerChannelInitializer(name));
         serverBootstrap.handler(new ServerChannelExceptionHandler());