Add Channel#bufferAllocator() (#11651)

__Motivation__

As we start to migrate codecs to use the new `Buffer` API, we need a way for them to get a handle of `BufferAllocator`.

__Modification__

Added `bufferAllocator()` method to `ChannelConfig`, `Channel` and `ChannelHandlerContext`

__Result__

Codecs can allocate `Buffer` instances

buffer/src/main/java/io/netty/buffer/api/DefaultGlobalBufferAllocator.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 2021 The Netty Project
+ *
+ * The Netty Project 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:
+ *
+ * https://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 io.netty.buffer.api;
+
+import io.netty.util.internal.PlatformDependent;
+import io.netty.util.internal.SystemPropertyUtil;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+
+import java.util.Locale;
+import java.util.function.Supplier;
+
+import static io.netty.buffer.api.BufferAllocator.offHeapPooled;
+import static io.netty.buffer.api.BufferAllocator.offHeapUnpooled;
+import static io.netty.buffer.api.BufferAllocator.onHeapPooled;
+import static io.netty.buffer.api.BufferAllocator.onHeapUnpooled;
+import static io.netty.util.internal.ObjectUtil.checkNotNullWithIAE;
+import static io.netty.util.internal.PlatformDependent.directBufferPreferred;
+import static java.lang.Runtime.getRuntime;
+
+/**
+ * A {@link BufferAllocator} which is {@link #close() disposed} when the {@link Runtime} is shutdown.
+ */
+public final class DefaultGlobalBufferAllocator implements BufferAllocator {
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultGlobalBufferAllocator.class);
+    public static final BufferAllocator DEFAUL_GLOBAL_BUFFER_ALLOCATOR;
+
+    static {
+        String allocType = SystemPropertyUtil.get(
+                "io.netty.allocator.type", PlatformDependent.isAndroid() ? "unpooled" : "pooled");
+        allocType = allocType.toLowerCase(Locale.US).trim();
+
+        BufferAllocator alloc;
+        if ("unpooled".equals(allocType)) {
+            alloc = directBufferPreferred() ? offHeapUnpooled() : onHeapUnpooled();
+            logger.debug("-Dio.netty.allocator.type: {}", allocType);
+        } else if ("pooled".equals(allocType)) {
+            alloc = directBufferPreferred() ? offHeapPooled() : onHeapPooled();
+            logger.debug("-Dio.netty.allocator.type: {}", allocType);
+        } else {
+            alloc = directBufferPreferred() ? offHeapPooled() : onHeapPooled();
+            logger.debug("-Dio.netty.allocator.type: pooled (unknown: {})", allocType);
+        }
+        DEFAUL_GLOBAL_BUFFER_ALLOCATOR = new DefaultGlobalBufferAllocator(alloc);
+    }
+
+    private final BufferAllocator delegate;
+
+    private DefaultGlobalBufferAllocator(BufferAllocator delegate) {
+        this.delegate = checkNotNullWithIAE(delegate, "delegate");
+        getRuntime().addShutdownHook(new Thread(this.delegate::close));
+    }
+
+    @Override
+    public Buffer allocate(int size) {
+        return delegate.allocate(size);
+    }
+
+    @Override
+    public Supplier<Buffer> constBufferSupplier(byte[] bytes) {
+        return delegate.constBufferSupplier(bytes);
+    }
+
+    /**
+     * @throws UnsupportedOperationException Close is not supported as this is a shared allocator.
+     */
+    @Override
+    public void close() {
+        throw new UnsupportedOperationException("Global buffer allocator can not be closed explicitly.");
+    }
+}

codec-http/src/main/java/io/netty/handler/codec/http/DelegatingChannelHandlerContext.java

@@ -16,6 +16,7 @@
 package io.netty.handler.codec.http;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -139,6 +140,11 @@ abstract class DelegatingChannelHandlerContext implements ChannelHandlerContext
         return ctx.alloc();
     }
 
+    @Override
+    public BufferAllocator bufferAllocator() {
+        return ctx.bufferAllocator();
+    }
+
     @Deprecated
     public <T> Attribute<T> attr(AttributeKey<T> key) {
         return ctx.attr(key);

codec-http2/src/test/java/io/netty/handler/codec/http2/Http2FrameInboundWriter.java

@@ -18,6 +18,7 @@ package io.netty.handler.codec.http2;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -213,6 +214,11 @@ final class Http2FrameInboundWriter {
             return channel.alloc();
         }
 
+        @Override
+        public BufferAllocator bufferAllocator() {
+            return channel.bufferAllocator();
+        }
+
         @Override
         public <T> Attribute<T> attr(AttributeKey<T> key) {
             return channel.attr(key);

codec-memcache/src/main/java/io/netty/handler/codec/memcache/binary/BinaryMemcacheClientCodec.java

@@ -17,6 +17,7 @@ package io.netty.handler.codec.memcache.binary;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -207,6 +208,11 @@ public final class BinaryMemcacheClientCodec extends
                     return ctx.alloc();
                 }
 
+                @Override
+                public BufferAllocator bufferAllocator() {
+                    return ctx.bufferAllocator();
+                }
+
                 @Override
                 @Deprecated
                 public <T> Attribute<T> attr(AttributeKey<T> key) {

codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java

@@ -20,6 +20,7 @@ import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufConvertible;
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelHandler;
@@ -626,6 +627,11 @@ public abstract class ByteToMessageDecoder extends ChannelHandlerAdapter {
             return ctx.alloc();
         }
 
+        @Override
+        public BufferAllocator bufferAllocator() {
+            return ctx.bufferAllocator();
+        }
+
         @Override
         @Deprecated
         public <T> Attribute<T> attr(AttributeKey<T> key) {

microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelHandlerContext.java

@@ -15,6 +15,7 @@
 package io.netty.microbench.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
@@ -37,11 +38,22 @@ public abstract class EmbeddedChannelHandlerContext implements ChannelHandlerCon
     private final EventLoop eventLoop;
     private final Channel channel;
     private final ByteBufAllocator alloc;
+    private final BufferAllocator bufferAllocator;
     private final ChannelHandler handler;
     private SocketAddress localAddress;
 
     protected EmbeddedChannelHandlerContext(ByteBufAllocator alloc, ChannelHandler handler, EmbeddedChannel channel) {
         this.alloc = requireNonNull(alloc, "alloc");
+        this.bufferAllocator = null;
+        this.channel = requireNonNull(channel, "channel");
+        this.handler = requireNonNull(handler, "handler");
+        eventLoop = requireNonNull(channel.executor(), "eventLoop");
+    }
+
+    protected EmbeddedChannelHandlerContext(BufferAllocator bufferAllocator, ChannelHandler handler,
+                                            EmbeddedChannel channel) {
+        this.bufferAllocator = requireNonNull(bufferAllocator, "bufferAllocator");
+        this.alloc = null;
         this.channel = requireNonNull(channel, "channel");
         this.handler = requireNonNull(handler, "handler");
         eventLoop = requireNonNull(channel.executor(), "eventLoop");
@@ -244,6 +256,11 @@ public abstract class EmbeddedChannelHandlerContext implements ChannelHandlerCon
         return alloc;
     }
 
+    @Override
+    public BufferAllocator bufferAllocator() {
+        return bufferAllocator;
+    }
+
     @Override
     public final Promise<Void> newPromise() {
         return channel().newPromise();

microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteAccumulatingHandlerContext.java

@@ -16,6 +16,7 @@ package io.netty.microbench.channel;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.codec.ByteToMessageDecoder;
@@ -39,6 +40,18 @@ public abstract class EmbeddedChannelWriteAccumulatingHandlerContext extends Emb
         cumulator = requireNonNull(writeCumulator, "writeCumulator");
     }
 
+    protected EmbeddedChannelWriteAccumulatingHandlerContext(BufferAllocator alloc, ChannelHandler handler,
+                                                             ByteToMessageDecoder.Cumulator writeCumulator) {
+        this(alloc, handler, writeCumulator, new EmbeddedChannel());
+    }
+
+    protected EmbeddedChannelWriteAccumulatingHandlerContext(BufferAllocator alloc, ChannelHandler handler,
+                                                             ByteToMessageDecoder.Cumulator writeCumulator,
+                                                             EmbeddedChannel channel) {
+        super(alloc, handler, channel);
+        cumulator = requireNonNull(writeCumulator, "writeCumulator");
+    }
+
     public final ByteBuf cumulation() {
         return cumulation;
     }

microbench/src/main/java/io/netty/microbench/channel/EmbeddedChannelWriteReleaseHandlerContext.java

@@ -15,6 +15,7 @@
 package io.netty.microbench.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.util.ReferenceCounted;
@@ -29,6 +30,14 @@ public abstract class EmbeddedChannelWriteReleaseHandlerContext extends Embedded
             EmbeddedChannel channel) {
         super(alloc, handler, channel);
     }
+    protected EmbeddedChannelWriteReleaseHandlerContext(BufferAllocator alloc, ChannelHandler handler) {
+        this(alloc, handler, new EmbeddedChannel());
+    }
+
+    protected EmbeddedChannelWriteReleaseHandlerContext(BufferAllocator alloc, ChannelHandler handler,
+            EmbeddedChannel channel) {
+        super(alloc, handler, channel);
+    }
 
     @Override
     protected abstract void handleException(Throwable t);

microbench/src/main/java/io/netty/microbench/handler/ssl/AbstractSslHandlerBenchmark.java

@@ -17,6 +17,7 @@ package io.netty.microbench.handler.ssl;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelHandler;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.ssl.SslContext;
@@ -168,6 +169,11 @@ public class AbstractSslHandlerBenchmark extends AbstractMicrobenchmark {
             super(alloc, handler, writeCumulator);
         }
 
+        SslThroughputBenchmarkHandlerContext(BufferAllocator alloc, ChannelHandler handler,
+                                             ByteToMessageDecoder.Cumulator writeCumulator) {
+            super(alloc, handler, writeCumulator);
+        }
+
         @Override
         protected void handleException(Throwable t) {
             handleUnexpectedException(t);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollChannelConfig.java

@@ -17,6 +17,8 @@ package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
 
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.DefaultChannelConfig;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -56,6 +58,12 @@ public class EpollChannelConfig extends DefaultChannelConfig {
         return this;
     }
 
+    @Override
+    public EpollChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         if (!(allocator.newHandle() instanceof RecvByteBufAllocator.ExtendedHandle)) {

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannelConfig.java

@@ -17,6 +17,7 @@ package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.FixedRecvByteBufAllocator;
@@ -220,6 +221,12 @@ public final class EpollDatagramChannelConfig extends EpollChannelConfig impleme
         return this;
     }
 
+    @Override
+    public EpollDatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollDatagramChannelConfig setConnectTimeoutMillis(int connectTimeoutMillis) {
         super.setConnectTimeoutMillis(connectTimeoutMillis);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainDatagramChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.FixedRecvByteBufAllocator;
@@ -96,6 +97,12 @@ public final class EpollDomainDatagramChannelConfig extends EpollChannelConfig i
         return this;
     }
 
+    @Override
+    public EpollDomainDatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollDomainDatagramChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDomainSocketChannelConfig.java

@@ -18,6 +18,7 @@ package io.netty.channel.epoll;
 import static java.util.Objects.requireNonNull;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -113,6 +114,12 @@ public final class EpollDomainSocketChannelConfig extends EpollDuplexChannelConf
         return this;
     }
 
+    @Override
+    public EpollDomainSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollDomainSocketChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollDuplexChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -94,6 +95,12 @@ public class EpollDuplexChannelConfig extends EpollChannelConfig implements Dupl
         return this;
     }
 
+    @Override
+    public EpollDuplexChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollDuplexChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -187,6 +188,12 @@ public class EpollServerChannelConfig extends EpollChannelConfig implements Serv
         return this;
     }
 
+    @Override
+    public EpollServerChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollServerChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollServerSocketChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -134,6 +135,12 @@ public final class EpollServerSocketChannelConfig extends EpollServerChannelConf
         return this;
     }
 
+    @Override
+    public EpollServerSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollServerSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport-native-epoll/src/main/java/io/netty/channel/epoll/EpollSocketChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.epoll;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -593,6 +594,12 @@ public final class EpollSocketChannelConfig extends EpollDuplexChannelConfig imp
         return this;
     }
 
+    @Override
+    public EpollSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public EpollSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueChannelConfig.java

@@ -16,6 +16,8 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
 import io.netty.channel.MessageSizeEstimator;
@@ -108,6 +110,12 @@ public class KQueueChannelConfig extends DefaultChannelConfig {
         return this;
     }
 
+    @Override
+    public KQueueChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         if (!(allocator.newHandle() instanceof RecvByteBufAllocator.ExtendedHandle)) {

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDatagramChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.FixedRecvByteBufAllocator;
@@ -235,6 +236,12 @@ public final class KQueueDatagramChannelConfig extends KQueueChannelConfig imple
         return this;
     }
 
+    @Override
+    public KQueueDatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueDatagramChannelConfig setConnectTimeoutMillis(int connectTimeoutMillis) {
         super.setConnectTimeoutMillis(connectTimeoutMillis);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainDatagramChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.FixedRecvByteBufAllocator;
@@ -96,6 +97,12 @@ public final class KQueueDomainDatagramChannelConfig
         return this;
     }
 
+    @Override
+    public KQueueDomainDatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueDomainDatagramChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDomainSocketChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -115,6 +116,12 @@ public final class KQueueDomainSocketChannelConfig extends KQueueDuplexChannelCo
         return this;
     }
 
+    @Override
+    public KQueueDomainSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueDomainSocketChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueDuplexChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
@@ -96,6 +97,12 @@ public class KQueueDuplexChannelConfig extends KQueueChannelConfig implements Du
         return this;
     }
 
+    @Override
+    public KQueueDuplexChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueDuplexChannelConfig setAutoClose(boolean autoClose) {
         super.setAutoClose(autoClose);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -164,6 +165,12 @@ public class KQueueServerChannelConfig extends KQueueChannelConfig implements Se
         return this;
     }
 
+    @Override
+    public KQueueServerChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueServerChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueServerSocketChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -161,6 +162,12 @@ public class KQueueServerSocketChannelConfig extends KQueueServerChannelConfig i
         return this;
     }
 
+    @Override
+    public KQueueServerSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueServerSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/KQueueSocketChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.kqueue;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
@@ -349,6 +350,12 @@ public final class KQueueSocketChannelConfig extends KQueueDuplexChannelConfig i
         return this;
     }
 
+    @Override
+    public KQueueSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public KQueueSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpChannelConfig.java

@@ -18,6 +18,8 @@ package io.netty.channel.sctp;
 import com.sun.nio.sctp.SctpChannel;
 import com.sun.nio.sctp.SctpStandardSocketOptions;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -202,6 +204,12 @@ public class DefaultSctpChannelConfig extends DefaultChannelConfig implements Sc
         return this;
     }
 
+    @Override
+    public SctpChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public SctpChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpServerChannelConfig.java

@@ -21,6 +21,8 @@ import static java.util.Objects.requireNonNull;
 import com.sun.nio.sctp.SctpServerChannel;
 import com.sun.nio.sctp.SctpStandardSocketOptions;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -183,6 +185,12 @@ public class DefaultSctpServerChannelConfig extends DefaultChannelConfig impleme
         return this;
     }
 
+    @Override
+    public SctpServerChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public SctpServerChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport/src/main/java/io/netty/channel/Channel.java

@@ -17,6 +17,7 @@ package io.netty.channel;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.socket.DatagramChannel;
 import io.netty.channel.socket.DatagramPacket;
 import io.netty.channel.socket.ServerSocketChannel;
@@ -199,11 +200,20 @@ public interface Channel extends AttributeMap, ChannelOutboundInvoker, Comparabl
 
     /**
      * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
+     * @deprecated Use {@link #bufferAllocator()}
      */
+    @Deprecated
     default ByteBufAllocator alloc() {
         return config().getAllocator();
     }
 
+    /**
+     * Return the assigned {@link BufferAllocator} which will be used to allocate {@link io.netty.buffer.api.Buffer}s.
+     */
+    default BufferAllocator bufferAllocator() {
+        return config().getBufferAllocator();
+    }
+
     @Override
     default Channel read() {
         pipeline().read();

transport/src/main/java/io/netty/channel/ChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.channel.socket.SocketChannelConfig;
 
 import java.nio.ByteBuffer;
@@ -165,15 +166,31 @@ public interface ChannelConfig {
     /**
      * Returns {@link ByteBufAllocator} which is used for the channel
      * to allocate buffers.
+     * @deprecated use {@link #getBufferAllocator()}
      */
+    @Deprecated
     ByteBufAllocator getAllocator();
 
     /**
      * Set the {@link ByteBufAllocator} which is used for the channel
      * to allocate buffers.
+     * @deprecated use {@link #setBufferAllocator(BufferAllocator)}
      */
+    @Deprecated
     ChannelConfig setAllocator(ByteBufAllocator allocator);
 
+    /**
+     * Returns {@link BufferAllocator} which is used for the channel
+     * to allocate buffers.
+     */
+    BufferAllocator getBufferAllocator();
+
+    /**
+     * Set the {@link ByteBufAllocator} which is used for the channel
+     * to allocate buffers.
+     */
+    ChannelConfig setBufferAllocator(BufferAllocator allocator);
+
     /**
      * Returns {@link RecvByteBufAllocator} which is used for the channel to allocate receive buffers.
      */

transport/src/main/java/io/netty/channel/ChannelHandlerContext.java

@@ -17,6 +17,7 @@ package io.netty.channel;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 import io.netty.util.AttributeMap;
@@ -186,8 +187,14 @@ public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvok
     /**
      * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
      */
+    @Deprecated
     ByteBufAllocator alloc();
 
+    /**
+     * Return the assigned {@link ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
+     */
+    BufferAllocator bufferAllocator();
+
     /**
      * @deprecated Use {@link Channel#attr(AttributeKey)}
      */

transport/src/main/java/io/netty/channel/ChannelOption.java

@@ -18,6 +18,7 @@ package io.netty.channel;
 import static java.util.Objects.requireNonNull;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.util.AbstractConstant;
 import io.netty.util.ConstantPool;
 
@@ -77,6 +78,7 @@ public class ChannelOption<T> extends AbstractConstant<ChannelOption<T>> {
     }
 
     public static final ChannelOption<ByteBufAllocator> ALLOCATOR = valueOf("ALLOCATOR");
+    public static final ChannelOption<BufferAllocator> BUFFER_ALLOCATOR = valueOf("BUFFER_ALLOCATOR");
     public static final ChannelOption<RecvByteBufAllocator> RCVBUF_ALLOCATOR = valueOf("RCVBUF_ALLOCATOR");
     public static final ChannelOption<MessageSizeEstimator> MESSAGE_SIZE_ESTIMATOR = valueOf("MESSAGE_SIZE_ESTIMATOR");
 

transport/src/main/java/io/netty/channel/CombinedChannelDuplexHandler.java

@@ -16,6 +16,7 @@
 package io.netty.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 import io.netty.util.concurrent.EventExecutor;
@@ -493,6 +494,11 @@ public class CombinedChannelDuplexHandler<I extends ChannelHandler, O extends Ch
             return ctx.alloc();
         }
 
+        @Override
+        public BufferAllocator bufferAllocator() {
+            return ctx.bufferAllocator();
+        }
+
         @Override
         public Promise<Void> newPromise() {
             return ctx.newPromise();

transport/src/main/java/io/netty/channel/DefaultChannelConfig.java

@@ -16,6 +16,8 @@
 package io.netty.channel;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.buffer.api.DefaultGlobalBufferAllocator;
 import io.netty.util.internal.ObjectUtil;
 
 import java.util.IdentityHashMap;
@@ -27,6 +29,7 @@ import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import static io.netty.channel.ChannelOption.ALLOCATOR;
 import static io.netty.channel.ChannelOption.AUTO_CLOSE;
 import static io.netty.channel.ChannelOption.AUTO_READ;
+import static io.netty.channel.ChannelOption.BUFFER_ALLOCATOR;
 import static io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS;
 import static io.netty.channel.ChannelOption.MAX_MESSAGES_PER_READ;
 import static io.netty.channel.ChannelOption.MAX_MESSAGES_PER_WRITE;
@@ -57,6 +60,7 @@ public class DefaultChannelConfig implements ChannelConfig {
     protected final Channel channel;
 
     private volatile ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
+    private volatile BufferAllocator bufferAllocator = DefaultGlobalBufferAllocator.DEFAUL_GLOBAL_BUFFER_ALLOCATOR;
     private volatile RecvByteBufAllocator rcvBufAllocator;
     private volatile MessageSizeEstimator msgSizeEstimator = DEFAULT_MSG_SIZE_ESTIMATOR;
 
@@ -85,7 +89,7 @@ public class DefaultChannelConfig implements ChannelConfig {
         return getOptions(
                 null,
                 CONNECT_TIMEOUT_MILLIS, MAX_MESSAGES_PER_READ, WRITE_SPIN_COUNT,
-                ALLOCATOR, AUTO_READ, AUTO_CLOSE, RCVBUF_ALLOCATOR, WRITE_BUFFER_HIGH_WATER_MARK,
+                ALLOCATOR, BUFFER_ALLOCATOR, AUTO_READ, AUTO_CLOSE, RCVBUF_ALLOCATOR, WRITE_BUFFER_HIGH_WATER_MARK,
                 WRITE_BUFFER_LOW_WATER_MARK, WRITE_BUFFER_WATER_MARK, MESSAGE_SIZE_ESTIMATOR, MAX_MESSAGES_PER_WRITE);
     }
 
@@ -132,6 +136,9 @@ public class DefaultChannelConfig implements ChannelConfig {
         if (option == ALLOCATOR) {
             return (T) getAllocator();
         }
+        if (option == BUFFER_ALLOCATOR) {
+            return (T) getBufferAllocator();
+        }
         if (option == RCVBUF_ALLOCATOR) {
             return (T) getRecvByteBufAllocator();
         }
@@ -172,6 +179,8 @@ public class DefaultChannelConfig implements ChannelConfig {
             setWriteSpinCount((Integer) value);
         } else if (option == ALLOCATOR) {
             setAllocator((ByteBufAllocator) value);
+        } else if (option == BUFFER_ALLOCATOR) {
+            setBufferAllocator((BufferAllocator) value);
         } else if (option == RCVBUF_ALLOCATOR) {
             setRecvByteBufAllocator((RecvByteBufAllocator) value);
         } else if (option == AUTO_READ) {
@@ -297,6 +306,18 @@ public class DefaultChannelConfig implements ChannelConfig {
         return this;
     }
 
+    @Override
+    public BufferAllocator getBufferAllocator() {
+        return bufferAllocator;
+    }
+
+    @Override
+    public ChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        requireNonNull(bufferAllocator, "bufferAllocator");
+        this.bufferAllocator = bufferAllocator;
+        return this;
+    }
+
     @SuppressWarnings("unchecked")
     @Override
     public <T extends RecvByteBufAllocator> T getRecvByteBufAllocator() {

transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java

@@ -18,6 +18,7 @@ package io.netty.channel;
 import static java.util.Objects.requireNonNull;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
 import io.netty.util.Attribute;
 import io.netty.util.AttributeKey;
 import io.netty.util.ReferenceCountUtil;
@@ -131,6 +132,11 @@ final class DefaultChannelHandlerContext implements ChannelHandlerContext, Resou
         return channel().config().getAllocator();
     }
 
+    @Override
+    public BufferAllocator bufferAllocator() {
+        return channel().config().getBufferAllocator();
+    }
+
     @Override
     public String name() {
         return name;

transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java

@@ -16,6 +16,8 @@
 package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -386,6 +388,12 @@ public class DefaultDatagramChannelConfig extends DefaultChannelConfig implement
         return this;
     }
 
+    @Override
+    public DatagramChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public DatagramChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java

@@ -16,6 +16,8 @@
 package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -171,6 +173,12 @@ public class DefaultServerSocketChannelConfig extends DefaultChannelConfig
         return this;
     }
 
+    @Override
+    public ServerSocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public ServerSocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);

transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java

@@ -16,6 +16,8 @@
 package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.api.BufferAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -304,6 +306,12 @@ public class DefaultSocketChannelConfig extends DefaultChannelConfig
         return this;
     }
 
+    @Override
+    public SocketChannelConfig setBufferAllocator(BufferAllocator bufferAllocator) {
+        super.setBufferAllocator(bufferAllocator);
+        return this;
+    }
+
     @Override
     public SocketChannelConfig setRecvByteBufAllocator(RecvByteBufAllocator allocator) {
         super.setRecvByteBufAllocator(allocator);