Drop usage of partial byte buf allocator.

Motivation:

Vert.x partial byte buf allocator implies the usage of unpooled buffers which can lead to increase significantly the memory footprint for SSL.

Changes:

Delete the partial byte buf allocator and use instead the same allocator than the non SSL case.

vertx-core/src/main/java/io/vertx/core/impl/buffer/VertxByteBufAllocator.java

@@ -30,11 +30,6 @@ public abstract class VertxByteBufAllocator extends AbstractByteBufAllocator {
     POOLED_ALLOCATOR = pooledAllocator;
   }
 
-  /**
-   * Vert.x shared un-pooled allocator.
-   */
-  public static final ByteBufAllocator UNPOOLED_ALLOCATOR = new UnpooledByteBufAllocator(false);
-
   private static final VertxByteBufAllocator UNSAFE_IMPL = new VertxByteBufAllocator() {
     @Override
     protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
@@ -53,7 +48,7 @@ protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
 
   @Override
   protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
-    return UNPOOLED_ALLOCATOR.directBuffer(initialCapacity, maxCapacity);
+    return UnpooledByteBufAllocator.DEFAULT.directBuffer(initialCapacity, maxCapacity);
   }
 
   @Override

vertx-core/src/main/java/io/vertx/core/net/impl/NetServerImpl.java

@@ -22,7 +22,6 @@
 import io.netty.handler.traffic.GlobalTrafficShapingHandler;
 import io.netty.util.concurrent.GenericFutureListener;
 import io.vertx.core.*;
-import io.vertx.core.buffer.impl.PartialPooledByteBufAllocator;
 import io.vertx.core.http.ClientAuth;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.http.impl.HttpUtils;
@@ -507,13 +506,8 @@ private void bind(
     channelBalancer.addWorker(eventLoop, worker);
     ServerBootstrap bootstrap = new ServerBootstrap();
     bootstrap.group(vertx.getAcceptorEventLoopGroup(), channelBalancer.workers());
-    if (options.isSsl()) {
-      bootstrap.childOption(ChannelOption.ALLOCATOR, PartialPooledByteBufAllocator.INSTANCE);
-    } else {
-      bootstrap.childOption(ChannelOption.ALLOCATOR, VertxByteBufAllocator.POOLED_ALLOCATOR);
-    }
-
     bootstrap.childHandler(channelBalancer);
+    bootstrap.childOption(ChannelOption.ALLOCATOR, VertxByteBufAllocator.POOLED_ALLOCATOR);
     applyConnectionOptions(localAddress.isDomainSocket(), bootstrap);
 
     // Actual bind

vertx-core/src/main/java/io/vertx/core/spi/transport/Transport.java

@@ -117,7 +117,6 @@ default EventLoopGroup eventLoopGroup(int type, int nThreads, ThreadFactory thre
   ChannelFactory<? extends ServerChannel> serverChannelFactory(boolean domainSocket);
 
   default void configure(DatagramChannel channel, DatagramSocketOptions options) {
-    channel.config().setAllocator(PartialPooledByteBufAllocator.INSTANCE);
     if (options.getSendBufferSize() != -1) {
       channel.config().setSendBufferSize(options.getSendBufferSize());
     }

vertx-core/src/test/java/io/vertx/it/buffer/TcpAllocationTest.java

@@ -11,24 +11,40 @@
 package io.vertx.it.buffer;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
-import io.netty.buffer.UnpooledByteBufAllocator;
 import io.netty.util.ReferenceCountUtil;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.buffer.impl.BufferImpl;
 import io.vertx.core.impl.buffer.VertxByteBufAllocator;
 import io.vertx.core.internal.net.NetSocketInternal;
-import io.vertx.core.net.NetClient;
-import io.vertx.core.net.NetServer;
-import io.vertx.core.net.NetSocket;
+import io.vertx.core.net.*;
 import io.vertx.test.core.VertxTestBase;
+import io.vertx.test.tls.Cert;
+import org.junit.Assume;
 import org.junit.Test;
 
 public class TcpAllocationTest extends VertxTestBase {
 
   @Test
   public void testByteBufOriginateFromDefaultByteBufAllocator() {
-    NetServer server = vertx.createNetServer();
+    testByteBufOriginateFromDefaultByteBufAllocator(null);
+  }
+
+  @Test
+  public void testByteBufOriginateFromDefaultByteBufAllocatorWithJdkSsl() {
+    testByteBufOriginateFromDefaultByteBufAllocator(new JdkSSLEngineOptions());
+  }
+
+  @Test
+  public void testByteBufOriginateFromDefaultByteBufAllocatorWithOpenSsl() {
+    Assume.assumeTrue(OpenSSLEngineOptions.isAvailable());
+    testByteBufOriginateFromDefaultByteBufAllocator(new OpenSSLEngineOptions());
+  }
+
+  private void testByteBufOriginateFromDefaultByteBufAllocator(SSLEngineOptions sslEngineOptions) {
+    NetServer server = vertx.createNetServer(new NetServerOptions()
+      .setSsl(sslEngineOptions != null)
+      .setSslEngineOptions(sslEngineOptions)
+      .setKeyCertOptions(Cert.SERVER_JKS.get()));
     server.connectHandler(so -> {
       NetSocketInternal soi = (NetSocketInternal) so;
       soi.messageHandler(msg -> {
@@ -38,13 +54,26 @@ public void testByteBufOriginateFromDefaultByteBufAllocator() {
         } finally {
           ReferenceCountUtil.release(msg);
         }
-        testComplete();
+        soi.write(Buffer.buffer("pong"));
       });
     });
     server.listen(1234, "localhost").await();
-    NetClient client = vertx.createNetClient();
-    NetSocket so = client.connect(1234, "localhost").await();
-    so.write(Buffer.buffer("ping"));
+    NetClient client = vertx.createNetClient(new NetClientOptions()
+      .setSsl(sslEngineOptions != null)
+      .setTrustAll(true)
+      .setHostnameVerificationAlgorithm("")
+    );
+    NetSocketInternal soi = (NetSocketInternal) client.connect(1234, "localhost").await();
+    soi.messageHandler(msg -> {
+      try {
+        ByteBuf bbuf = (ByteBuf) msg;
+        assertSame(VertxByteBufAllocator.POOLED_ALLOCATOR, bbuf.alloc());
+      } finally {
+        ReferenceCountUtil.release(msg);
+      }
+      testComplete();
+    });
+    soi.write(Buffer.buffer("ping"));
     await();
   }