Remove option to enable direct buffer pooling

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("es.unsafe.use_unpooled_allocator") == false) {
-            if (heapSize <= 1 << 30) {
-                ergonomicChoices.add("-Des.unsafe.use_unpooled_allocator=true");
-            } else {
-                ergonomicChoices.add("-Des.unsafe.use_unpooled_allocator=false");
-            }
-        }
         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("-Des.unsafe.use_unpooled_allocator=true"));
-    }
-
-    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("-Des.unsafe.use_unpooled_allocator=false"));
-    }
-
     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 'es.unsafe.use_unpooled_allocator', 'false'
+  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 'es.unsafe.use_unpooled_allocator', 'false'
+  systemProperty 'es.use_unpooled_allocator', 'false'
 }
 check.dependsOn(pooledTest, pooledIntegTest)
 

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

@@ -61,7 +61,6 @@
 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;
 
@@ -185,14 +184,12 @@ protected void doStart() {
             serverBootstrap.group(new NioEventLoopGroup(workerCount, daemonThreadFactory(settings,
                 HTTP_SERVER_WORKER_THREAD_NAME_PREFIX)));
 
-            // 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.
-            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.ALLOCATOR);
-            serverBootstrap.childOption(ChannelOption.ALLOCATOR, NettyAllocator.ALLOCATOR);
+            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

@@ -24,22 +24,27 @@
 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 {
 
-    public static final ByteBufAllocator ALLOCATOR;
+    private static final ByteBufAllocator ALLOCATOR;
 
-    private static final String USE_UNPOOLED = "es.unsafe.use_unpooled_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 (Booleans.parseBoolean(System.getProperty(USE_UNPOOLED), false)) {
-                delegate = UnpooledByteBufAllocator.DEFAULT;
+            if (useUnpooled()) {
+                delegate = new NoDirectBuffers(UnpooledByteBufAllocator.DEFAULT);
             } else {
                 int nHeapArena = PooledByteBufAllocator.defaultNumHeapArena();
                 int pageSize = PooledByteBufAllocator.defaultPageSize();
@@ -55,6 +60,39 @@ public class NettyAllocator {
         }
     }
 
+    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 (Booleans.parseBoolean(System.getProperty(USE_UNPOOLED), false)) {
+            return true;
+        } else {
+            long heapSize = JvmInfo.jvmInfo().getMem().getHeapMax().getBytes();
+            return heapSize <= 1 << 30;
+        }
+    }
+
     private static class NoDirectBuffers implements ByteBufAllocator {
 
         private final ByteBufAllocator delegate;

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

@@ -56,8 +56,6 @@
 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;
@@ -149,8 +147,10 @@ protected void doStart() {
     private Bootstrap createClientBootstrap(NioEventLoopGroup eventLoopGroup) {
         final Bootstrap bootstrap = new Bootstrap();
         bootstrap.group(eventLoopGroup);
-        bootstrap.channel(CopyBytesSocketChannel.class);
-        bootstrap.option(ChannelOption.ALLOCATOR, NettyAllocator.ALLOCATOR);
+
+        // 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));
@@ -208,14 +208,12 @@ private void createServerBootstrap(ProfileSettings profileSettings, NioEventLoop
 
         serverBootstrap.group(eventLoopGroup);
 
-        // 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.
-        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.ALLOCATOR);
-        serverBootstrap.childOption(ChannelOption.ALLOCATOR, NettyAllocator.ALLOCATOR);
+        serverBootstrap.option(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator());
+        serverBootstrap.childOption(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator());
 
         serverBootstrap.childHandler(getServerChannelInitializer(name));
         serverBootstrap.handler(new ServerChannelExceptionHandler());

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpClient.java

@@ -45,7 +45,6 @@
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.tasks.Task;
-import org.elasticsearch.transport.CopyBytesSocketChannel;
 import org.elasticsearch.transport.NettyAllocator;
 
 import java.io.Closeable;
@@ -87,8 +86,8 @@ static Collection<String> returnOpaqueIds(Collection<FullHttpResponse> responses
 
     Netty4HttpClient() {
         clientBootstrap = new Bootstrap()
-            .channel(CopyBytesSocketChannel.class)
-            .option(ChannelOption.ALLOCATOR, NettyAllocator.ALLOCATOR)
+            .channel(NettyAllocator.getChannelType())
+            .option(ChannelOption.ALLOCATOR, NettyAllocator.getAllocator())
             .group(new NioEventLoopGroup());
     }