Fix AssertionError from AsyncSocketChannel.beginRead()

An AssertionError is triggered by a ByteBuf when beginRead() attempts to
access the buffer which has been freed already.  This commit ensures the
buffer is not freed before performing an I/O operation.

To determine if the buffer has been freed, UnsafeByteBuf.isFreed() has
been added.

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

@@ -1536,6 +1536,10 @@ public class DefaultCompositeByteBuf extends AbstractByteBuf implements Composit
         discardReadComponents();
     }
 
+    public boolean isFreed() {
+        return freed;
+    }
+
     @Override
     public void free() {
         if (freed) {

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

@@ -252,6 +252,11 @@ public class DuplicatedByteBuf extends AbstractByteBuf {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public boolean isFreed() {
+        return buffer.isFreed();
+    }
+
     @Override
     public void free() {
 

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

@@ -252,6 +252,11 @@ public class ReadOnlyByteBuf extends AbstractByteBuf {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public boolean isFreed() {
+        return buffer.isFreed();
+    }
+
     @Override
     public void free() { }
 }

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

@@ -324,6 +324,11 @@ public class SlicedByteBuf extends AbstractByteBuf {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public boolean isFreed() {
+        return buffer.isFreed();
+    }
+
     @Override
     public void free() { }
 }

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

@@ -809,6 +809,11 @@ public class SwappedByteBuf implements UnsafeByteBuf {
         buf.discardSomeReadBytes();
     }
 
+    @Override
+    public boolean isFreed() {
+        return buf.isFreed();
+    }
+
     @Override
     public void free() { }
 }

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

@@ -498,6 +498,11 @@ public class UnpooledDirectByteBuf extends AbstractByteBuf {
         }
     }
 
+    @Override
+    public boolean isFreed() {
+        return freed;
+    }
+
     @Override
     public void free() {
         if (freed) {

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

@@ -375,6 +375,11 @@ public class UnpooledHeapByteBuf extends AbstractByteBuf {
         }
     }
 
+    @Override
+    public boolean isFreed() {
+        return freed;
+    }
+
     @Override
     public void free() {
         freed = true;

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

@@ -42,6 +42,11 @@ public interface UnsafeByteBuf extends ByteBuf {
      */
     void discardSomeReadBytes();
 
+    /**
+     * Returns {@code true} if and only if this buffer has been deallocated by {@link #free()}.
+     */
+    boolean isFreed();
+
     /**
      * Deallocates the internal memory block of this buffer or returns it to the {@link ByteBufAllocator} it came
      * from.  The result of accessing a released buffer is unspecified and can even cause JVM crash.

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

@@ -865,6 +865,11 @@ class ReplayingDecoderBuffer implements UnsafeByteBuf {
         throw new UnreplayableOperationException();
     }
 
+    @Override
+    public boolean isFreed() {
+        return ((UnsafeByteBuf) buffer).isFreed();
+    }
+
     @Override
     public void free() {
         throw new UnreplayableOperationException();

transport/src/main/java/io/netty/channel/socket/aio/AioSocketChannel.java

@@ -17,6 +17,7 @@ package io.netty.channel.socket.aio;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ChannelBufType;
+import io.netty.buffer.UnsafeByteBuf;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelFlushFutureNotifier;
 import io.netty.channel.ChannelFuture;
@@ -244,6 +245,9 @@ public class AioSocketChannel extends AbstractAioChannel implements SocketChanne
         try {
             if (buf.readable()) {
                 for (;;) {
+                    if (((UnsafeByteBuf) buf).isFreed()) {
+                        break;
+                    }
                     // Ensure the readerIndex of the buffer is 0 before beginning an async write.
                     // Otherwise, JDK can write into a wrong region of the buffer when a handler calls
                     // discardReadBytes() later, modifying the readerIndex and the writerIndex unexpectedly.
@@ -288,7 +292,7 @@ public class AioSocketChannel extends AbstractAioChannel implements SocketChanne
     }
 
     private void beginRead() {
-        if (inBeginRead || asyncReadInProgress || inputShutdown || readSuspended.get()) {
+        if (inBeginRead || asyncReadInProgress || readSuspended.get()) {
             return;
         }
 
@@ -297,6 +301,10 @@ public class AioSocketChannel extends AbstractAioChannel implements SocketChanne
         try {
             for (;;) {
                 ByteBuf byteBuf = pipeline().inboundByteBuffer();
+                if (((UnsafeByteBuf) byteBuf).isFreed()) {
+                    break;
+                }
+
                 if (!byteBuf.readable()) {
                     byteBuf.discardReadBytes();
                 }