Port ChannelOutboundBuffer and related changes from 4.0

Motivation:

We did various changes related to the ChannelOutboundBuffer in 4.0 branch. This commit port all of them over and so make sure our branches are synced in terms of these changes.

Related to [netty#2734], [netty#2709], [netty#2729], [netty#2710] and [netty#2693] .

Modification:
Port all changes that was done on the ChannelOutboundBuffer.

This includes the port of the following commits:
 - 73dfd7c
 - 997d8c3
 - e282e50
 - 5e5d1a5
 - 8ee3575
 - d6f0d12
 - 16e5076
 - 3f3e66c

Result:
 - Less memory usage by ChannelOutboundBuffer
 - Same code as in 4.0 branch
 - Make it possible to use ChannelOutboundBuffer with Channel implementation that not extends AbstractChannel

buffer/src/main/java/io/netty/buffer/ByteBufUtil.java

@@ -16,6 +16,8 @@
 package io.netty.buffer;
 
 import io.netty.util.CharsetUtil;
+import io.netty.util.Recycler;
+import io.netty.util.Recycler.Handle;
 import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.SystemPropertyUtil;
 import io.netty.util.internal.logging.InternalLogger;
@@ -42,16 +44,16 @@ public final class ByteBufUtil {
 
     static final ByteBufAllocator DEFAULT_ALLOCATOR;
 
+    private static final int THREAD_LOCAL_BUFFER_SIZE;
+
     static {
         final char[] DIGITS = "0123456789abcdef".toCharArray();
         for (int i = 0; i < 256; i ++) {
             HEXDUMP_TABLE[ i << 1     ] = DIGITS[i >>> 4 & 0x0F];
             HEXDUMP_TABLE[(i << 1) + 1] = DIGITS[i       & 0x0F];
         }
 
-        String allocType = SystemPropertyUtil.get(
-                "io.netty.allocator.type", PlatformDependent.isAndroid() ? "unpooled" : "pooled");
-        allocType = allocType.toLowerCase(Locale.US).trim();
+        String allocType = SystemPropertyUtil.get("io.netty.allocator.type", "unpooled").toLowerCase(Locale.US).trim();
 
         ByteBufAllocator alloc;
         if ("unpooled".equals(allocType)) {
@@ -66,6 +68,9 @@ public final class ByteBufUtil {
         }
 
         DEFAULT_ALLOCATOR = alloc;
+
+        THREAD_LOCAL_BUFFER_SIZE = SystemPropertyUtil.getInt("io.netty.threadLocalDirectBufferSize", 64 * 1024);
+        logger.debug("-Dio.netty.threadLocalDirectBufferSize: {}", THREAD_LOCAL_BUFFER_SIZE);
     }
 
     /**
@@ -414,5 +419,89 @@ static String decodeString(ByteBuffer src, Charset charset) {
         return dst.flip().toString();
     }
 
+    /**
+     * Returns a cached thread-local direct buffer, if available.
+     *
+     * @return a cached thread-local direct buffer, if available.  {@code null} otherwise.
+     */
+    public static ByteBuf threadLocalDirectBuffer() {
+        if (THREAD_LOCAL_BUFFER_SIZE <= 0) {
+            return null;
+        }
+
+        if (PlatformDependent.hasUnsafe()) {
+            return ThreadLocalUnsafeDirectByteBuf.newInstance();
+        } else {
+            return ThreadLocalDirectByteBuf.newInstance();
+        }
+    }
+
+    static final class ThreadLocalUnsafeDirectByteBuf extends UnpooledUnsafeDirectByteBuf {
+
+        private static final Recycler<ThreadLocalUnsafeDirectByteBuf> RECYCLER =
+                new Recycler<ThreadLocalUnsafeDirectByteBuf>() {
+                    @Override
+                    protected ThreadLocalUnsafeDirectByteBuf newObject(Handle handle) {
+                        return new ThreadLocalUnsafeDirectByteBuf(handle);
+                    }
+                };
+
+        static ThreadLocalUnsafeDirectByteBuf newInstance() {
+            ThreadLocalUnsafeDirectByteBuf buf = RECYCLER.get();
+            buf.setRefCnt(1);
+            return buf;
+        }
+
+        private final Handle handle;
+
+        private ThreadLocalUnsafeDirectByteBuf(Handle handle) {
+            super(UnpooledByteBufAllocator.DEFAULT, 256, Integer.MAX_VALUE);
+            this.handle = handle;
+        }
+
+        @Override
+        protected void deallocate() {
+            if (capacity() > THREAD_LOCAL_BUFFER_SIZE) {
+                super.deallocate();
+            } else {
+                clear();
+                RECYCLER.recycle(this, handle);
+            }
+        }
+    }
+
+    static final class ThreadLocalDirectByteBuf extends UnpooledDirectByteBuf {
+
+        private static final Recycler<ThreadLocalDirectByteBuf> RECYCLER = new Recycler<ThreadLocalDirectByteBuf>() {
+            @Override
+            protected ThreadLocalDirectByteBuf newObject(Handle handle) {
+                return new ThreadLocalDirectByteBuf(handle);
+            }
+        };
+
+        static ThreadLocalDirectByteBuf newInstance() {
+            ThreadLocalDirectByteBuf buf = RECYCLER.get();
+            buf.setRefCnt(1);
+            return buf;
+        }
+
+        private final Handle handle;
+
+        private ThreadLocalDirectByteBuf(Handle handle) {
+            super(UnpooledByteBufAllocator.DEFAULT, 256, Integer.MAX_VALUE);
+            this.handle = handle;
+        }
+
+        @Override
+        protected void deallocate() {
+            if (capacity() > THREAD_LOCAL_BUFFER_SIZE) {
+                super.deallocate();
+            } else {
+                clear();
+                RECYCLER.recycle(this, handle);
+            }
+        }
+    }
+
     private ByteBufUtil() { }
 }

common/src/main/java/io/netty/util/ReferenceCountUtil.java

@@ -39,7 +39,7 @@ public static <T> T retain(T msg) {
     }
 
     /**
-     * Try to call {@link ReferenceCounted#retain()} if the specified message implements {@link ReferenceCounted}.
+     * Try to call {@link ReferenceCounted#retain(int)} if the specified message implements {@link ReferenceCounted}.
      * If the specified message doesn't implement {@link ReferenceCounted}, this method does nothing.
      */
     @SuppressWarnings("unchecked")
@@ -87,7 +87,7 @@ public static boolean release(Object msg) {
     }
 
     /**
-     * Try to call {@link ReferenceCounted#release()} if the specified message implements {@link ReferenceCounted}.
+     * Try to call {@link ReferenceCounted#release(int)} if the specified message implements {@link ReferenceCounted}.
      * If the specified message doesn't implement {@link ReferenceCounted}, this method does nothing.
      */
     public static boolean release(Object msg, int decrement) {
@@ -97,6 +97,38 @@ public static boolean release(Object msg, int decrement) {
         return false;
     }
 
+    /**
+     * Try to call {@link ReferenceCounted#release()} if the specified message implements {@link ReferenceCounted}.
+     * If the specified message doesn't implement {@link ReferenceCounted}, this method does nothing.
+     * Unlike {@link #release(Object)} this method catches an exception raised by {@link ReferenceCounted#release()}
+     * and logs it, rather than rethrowing it to the caller.  It is usually recommended to use {@link #release(Object)}
+     * instead, unless you absolutely need to swallow an exception.
+     */
+    public static void safeRelease(Object msg) {
+        try {
+            release(msg);
+        } catch (Throwable t) {
+            logger.warn("Failed to release a message: {}", msg, t);
+        }
+    }
+
+    /**
+     * Try to call {@link ReferenceCounted#release(int)} if the specified message implements {@link ReferenceCounted}.
+     * If the specified message doesn't implement {@link ReferenceCounted}, this method does nothing.
+     * Unlike {@link #release(Object)} this method catches an exception raised by {@link ReferenceCounted#release(int)}
+     * and logs it, rather than rethrowing it to the caller.  It is usually recommended to use
+     * {@link #release(Object, int)} instead, unless you absolutely need to swallow an exception.
+     */
+    public static void safeRelease(Object msg, int decrement) {
+        try {
+            release(msg, decrement);
+        } catch (Throwable t) {
+            if (logger.isWarnEnabled()) {
+                logger.warn("Failed to release a message: {} (decrement: {})", msg, decrement, t);
+            }
+        }
+    }
+
     /**
      * Schedules the specified object to be released when the caller thread terminates. Note that this operation is
      * intended to simplify reference counting of ephemeral objects during unit tests. Do not use it beyond the

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

@@ -15,10 +15,15 @@
  */
 package io.netty.channel.epoll;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.AbstractChannel;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelMetadata;
 import io.netty.channel.EventLoop;
+import io.netty.util.ReferenceCountUtil;
 import io.netty.util.internal.OneTimeTask;
 
 import java.net.InetSocketAddress;
@@ -98,6 +103,9 @@ protected void doDeregister() throws Exception {
 
     @Override
     protected void doBeginRead() throws Exception {
+        // Channel.read() or ChannelHandlerContext.read() was called
+        ((AbstractEpollUnsafe) unsafe()).readPending = true;
+
         if ((flags & readFlag) == 0) {
             flags |= readFlag;
             modifyEvents();
@@ -159,6 +167,47 @@ protected void doRegister() throws Exception {
     @Override
     protected abstract AbstractEpollUnsafe newUnsafe();
 
+    /**
+     * Returns an off-heap copy of the specified {@link ByteBuf}, and releases the original one.
+     */
+    protected final ByteBuf newDirectBuffer(ByteBuf buf) {
+        return newDirectBuffer(buf, buf);
+    }
+
+    /**
+     * Returns an off-heap copy of the specified {@link ByteBuf}, and releases the specified holder.
+     * The caller must ensure that the holder releases the original {@link ByteBuf} when the holder is released by
+     * this method.
+     */
+    protected final ByteBuf newDirectBuffer(Object holder, ByteBuf buf) {
+        final int readableBytes = buf.readableBytes();
+        if (readableBytes == 0) {
+            ReferenceCountUtil.safeRelease(holder);
+            return Unpooled.EMPTY_BUFFER;
+        }
+
+        final ByteBufAllocator alloc = alloc();
+        if (alloc.isDirectBufferPooled()) {
+            return newDirectBuffer0(holder, buf, alloc, readableBytes);
+        }
+
+        final ByteBuf directBuf = ByteBufUtil.threadLocalDirectBuffer();
+        if (directBuf == null) {
+            return newDirectBuffer0(holder, buf, alloc, readableBytes);
+        }
+
+        directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
+        ReferenceCountUtil.safeRelease(holder);
+        return directBuf;
+    }
+
+    private static ByteBuf newDirectBuffer0(Object holder, ByteBuf buf, ByteBufAllocator alloc, int capacity) {
+        final ByteBuf directBuf = alloc.directBuffer(capacity);
+        directBuf.writeBytes(buf, buf.readerIndex(), capacity);
+        ReferenceCountUtil.safeRelease(holder);
+        return directBuf;
+    }
+
     protected static void checkResolvable(InetSocketAddress addr) {
         if (addr.isUnresolved()) {
             throw new UnresolvedAddressException();
@@ -180,13 +229,6 @@ void epollRdHupReady() {
             // NOOP
         }
 
-        @Override
-        public void beginRead() {
-            // Channel.read() or ChannelHandlerContext.read() was called
-            readPending = true;
-            super.beginRead();
-        }
-
         @Override
         protected void flush0() {
             // Flush immediately only when there's no pending flush.