Repository: netty/netty
Issue: Epoll transport doesn't work with custom FileRegion implementation
Labels: defect

- Netty version: 4.1.5.Final, 5.0.0.Alpha2
- Tried to use custom FileRegion implementation with epoll transport
- Expected: any FileRegion DMA implementation works with epoll transport as with NIO transport
- Observerd: epoll transport expects only the particular DefaultFileRegion implementation:

```
java.lang.UnsupportedOperationException: unsupported message type: DataItemFileRegion (expected: ByteBuf, DefaultFileRegion)
    at io.netty.channel.epoll.AbstractEpollStreamChannel.filterOutboundMessage(AbstractEpollStreamChannel.java:539)
    at io.netty.channel.AbstractChannel$AbstractUnsafe.write(AbstractChannel.java:799)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.write(DefaultChannelPipeline.java:1291)
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:748)
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:740)
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:826)
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:733)
    at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.write(CombinedChannelDuplexHandler.java:525)
    at io.netty.handler.codec.MessageToMessageEncoder.write(MessageToMessageEncoder.java:111)
    at io.netty.channel.CombinedChannelDuplexHandler.write(CombinedChannelDuplexHandler.java:345)
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:748)
    at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:740)
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:826)
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:733)
    at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:714)
```


Issue comments:
@akurilov let me fix this.

also FYI Netty 5.x has been discontinued.

Fixed by https://github.com/netty/netty/pull/5825


Pull request review comments:
src.limit() -> limit

Can you add a comment why you explicitly use `Unpooled` here instead of just using `alloc`?

on line 402 ... can you change `region.transfered()` -> `region.transferred()` while you are here?

![MAJOR](https://raw.githubusercontent.com/SonarCommunity/sonar-github/master/images/severity-major.png) Add the missing @deprecated Javadoc tag. [![rule](https://raw.githubusercontent.com/SonarCommunity/sonar-github/master/images/rule.png)](https://garage.netty.io/sonarqube/coding_rules#rule_key=squid%3AMissingDeprecatedCheck)

actually I should use the allocator.

false positive

fixed

fixed


Patch that resolved the issue:
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
index f18e4c74303..78507d70e28 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
@@ -17,7 +17,9 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelFuture;
@@ -29,6 +31,7 @@
 import io.netty.channel.ConnectTimeoutException;
 import io.netty.channel.DefaultFileRegion;
 import io.netty.channel.EventLoop;
+import io.netty.channel.FileRegion;
 import io.netty.channel.RecvByteBufAllocator;
 import io.netty.channel.socket.DuplexChannel;
 import io.netty.channel.unix.FileDescriptor;
@@ -44,6 +47,7 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.ConnectionPendingException;
+import java.nio.channels.WritableByteChannel;
 import java.util.Queue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledFuture;
@@ -83,6 +87,8 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel im
     private FileDescriptor pipeIn;
     private FileDescriptor pipeOut;
 
+    private WritableByteChannel byteChannel;
+
     /**
      * @deprecated Use {@link #AbstractEpollStreamChannel(Channel, Socket)}.
      */
@@ -372,7 +378,7 @@ private boolean writeBytesMultiple(
      * @param region        the {@link DefaultFileRegion} from which the bytes should be written
      * @return amount       the amount of written bytes
      */
-    private boolean writeFileRegion(
+    private boolean writeDefaultFileRegion(
             ChannelOutboundBuffer in, DefaultFileRegion region, int writeSpinCount) throws Exception {
         final long regionCount = region.count();
         if (region.transferred() >= regionCount) {
@@ -393,7 +399,43 @@ private boolean writeFileRegion(
             }
 
             flushedAmount += localFlushedAmount;
-            if (region.transfered() >= regionCount) {
+            if (region.transferred() >= regionCount) {
+                done = true;
+                break;
+            }
+        }
+
+        if (flushedAmount > 0) {
+            in.progress(flushedAmount);
+        }
+
+        if (done) {
+            in.remove();
+        }
+        return done;
+    }
+
+    private boolean writeFileRegion(
+            ChannelOutboundBuffer in, FileRegion region, final int writeSpinCount) throws Exception {
+        if (region.transferred() >= region.count()) {
+            in.remove();
+            return true;
+        }
+
+        boolean done = false;
+        long flushedAmount = 0;
+
+        if (byteChannel == null) {
+            byteChannel = new SocketWritableByteChannel();
+        }
+        for (int i = writeSpinCount - 1; i >= 0; i--) {
+            final long localFlushedAmount = region.transferTo(byteChannel, region.transferred());
+            if (localFlushedAmount == 0) {
+                break;
+            }
+
+            flushedAmount += localFlushedAmount;
+            if (region.transferred() >= region.count()) {
                 done = true;
                 break;
             }
@@ -448,15 +490,19 @@ protected boolean doWriteSingle(ChannelOutboundBuffer in, int writeSpinCount) th
         // The outbound buffer contains only one message or it contains a file region.
         Object msg = in.current();
         if (msg instanceof ByteBuf) {
-            ByteBuf buf = (ByteBuf) msg;
-            if (!writeBytes(in, buf, writeSpinCount)) {
+            if (!writeBytes(in, (ByteBuf) msg, writeSpinCount)) {
                 // was not able to write everything so break here we will get notified later again once
                 // the network stack can handle more writes.
                 return false;
             }
         } else if (msg instanceof DefaultFileRegion) {
-            DefaultFileRegion region = (DefaultFileRegion) msg;
-            if (!writeFileRegion(in, region, writeSpinCount)) {
+            if (!writeDefaultFileRegion(in, (DefaultFileRegion) msg, writeSpinCount)) {
+                // was not able to write everything so break here we will get notified later again once
+                // the network stack can handle more writes.
+                return false;
+            }
+        } else if (msg instanceof FileRegion) {
+            if (!writeFileRegion(in, (FileRegion) msg, writeSpinCount)) {
                 // was not able to write everything so break here we will get notified later again once
                 // the network stack can handle more writes.
                 return false;
@@ -533,7 +579,7 @@ protected Object filterOutboundMessage(Object msg) {
             return buf;
         }
 
-        if (msg instanceof DefaultFileRegion || msg instanceof SpliceOutTask) {
+        if (msg instanceof FileRegion || msg instanceof SpliceOutTask) {
             return msg;
         }
 
@@ -1211,4 +1257,56 @@ public boolean spliceIn(RecvByteBufAllocator.Handle handle) {
             }
         }
     }
+
+    private final class SocketWritableByteChannel implements WritableByteChannel {
+
+        @Override
+        public int write(ByteBuffer src) throws IOException {
+            final int written;
+            int position = src.position();
+            int limit = src.limit();
+            if (src.isDirect()) {
+                written = fd().write(src, position, limit);
+            } else {
+                final int readableBytes = limit - position;
+                ByteBuf buffer = null;
+                try {
+                    if (readableBytes == 0) {
+                        buffer = Unpooled.EMPTY_BUFFER;
+                    } else {
+                        final ByteBufAllocator alloc = alloc();
+                        if (alloc.isDirectBufferPooled()) {
+                            buffer = alloc.directBuffer(readableBytes);
+                        } else {
+                            buffer = ByteBufUtil.threadLocalDirectBuffer();
+                            if (buffer == null) {
+                                buffer = alloc.directBuffer(readableBytes);
+                            }
+                        }
+                    }
+                    buffer.writeBytes(src.duplicate());
+                    ByteBuffer nioBuffer = buffer.internalNioBuffer(buffer.readerIndex(), readableBytes);
+                    written = fd().write(nioBuffer, nioBuffer.position(), nioBuffer.limit());
+                } finally {
+                    if (buffer != null) {
+                        buffer.release();
+                    }
+                }
+            }
+            if (written > 0) {
+                src.position(position + written);
+            }
+            return written;
+        }
+
+        @Override
+        public boolean isOpen() {
+            return fd().isOpen();
+        }
+
+        @Override
+        public void close() throws IOException {
+            fd().close();
+        }
+    }
 }


Test patch:
diff --git a/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketFileRegionTest.java b/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketFileRegionTest.java
index 4a44c62171c..6b7ad3ca0e5 100644
--- a/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketFileRegionTest.java
+++ b/testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketFileRegionTest.java
@@ -33,6 +33,7 @@
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
 import java.io.IOException;
+import java.nio.channels.WritableByteChannel;
 import java.util.Random;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -52,6 +53,11 @@ public void testFileRegion() throws Throwable {
         run();
     }
 
+    @Test
+    public void testCustomFileRegion() throws Throwable {
+        run();
+    }
+
     @Test
     public void testFileRegionNotAutoRead() throws Throwable {
         run();
@@ -68,23 +74,28 @@ public void testFileRegionVoidPromiseNotAutoRead() throws Throwable {
     }
 
     public void testFileRegion(ServerBootstrap sb, Bootstrap cb) throws Throwable {
-        testFileRegion0(sb, cb, false, true);
+        testFileRegion0(sb, cb, false, true, true);
+    }
+
+    public void testCustomFileRegion(ServerBootstrap sb, Bootstrap cb) throws Throwable {
+        testFileRegion0(sb, cb, false, true, false);
     }
 
     public void testFileRegionVoidPromise(ServerBootstrap sb, Bootstrap cb) throws Throwable {
-        testFileRegion0(sb, cb, true, true);
+        testFileRegion0(sb, cb, true, true, true);
     }
 
     public void testFileRegionNotAutoRead(ServerBootstrap sb, Bootstrap cb) throws Throwable {
-        testFileRegion0(sb, cb, false, false);
+        testFileRegion0(sb, cb, false, false, true);
     }
 
     public void testFileRegionVoidPromiseNotAutoRead(ServerBootstrap sb, Bootstrap cb) throws Throwable {
-        testFileRegion0(sb, cb, true, false);
+        testFileRegion0(sb, cb, true, false, true);
     }
 
     private static void testFileRegion0(
-            ServerBootstrap sb, Bootstrap cb, boolean voidPromise, final boolean autoRead) throws Throwable {
+            ServerBootstrap sb, Bootstrap cb, boolean voidPromise, final boolean autoRead, boolean defaultFileRegion)
+            throws Throwable {
         sb.childOption(ChannelOption.AUTO_READ, autoRead);
         cb.option(ChannelOption.AUTO_READ, autoRead);
 
@@ -140,6 +151,10 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
                 new FileInputStream(file).getChannel(), startOffset, data.length - bufferSize);
         FileRegion emptyRegion = new DefaultFileRegion(new FileInputStream(file).getChannel(), 0, 0);
 
+        if (!defaultFileRegion) {
+            region = new FileRegionWrapper(region);
+            emptyRegion = new FileRegionWrapper(emptyRegion);
+        }
         // Do write ByteBuf and then FileRegion to ensure that mixed writes work
         // Also, write an empty FileRegion to test if writing an empty FileRegion does not cause any issues.
         //
@@ -229,4 +244,77 @@ public void exceptionCaught(ChannelHandlerContext ctx,
             }
         }
     }
+
+    private static final class FileRegionWrapper implements FileRegion {
+        private final FileRegion region;
+
+        FileRegionWrapper(FileRegion region) {
+            this.region = region;
+        }
+
+        @Override
+        public int refCnt() {
+            return region.refCnt();
+        }
+
+        @Override
+        public long position() {
+            return region.position();
+        }
+
+        @Override
+        @Deprecated
+        public long transfered() {
+            return region.transfered();
+        }
+
+        @Override
+        public boolean release() {
+            return region.release();
+        }
+
+        @Override
+        public long transferred() {
+            return region.transferred();
+        }
+
+        @Override
+        public long count() {
+            return region.count();
+        }
+
+        @Override
+        public boolean release(int decrement) {
+            return region.release(decrement);
+        }
+
+        @Override
+        public long transferTo(WritableByteChannel target, long position) throws IOException {
+            return region.transferTo(target, position);
+        }
+
+        @Override
+        public FileRegion retain() {
+            region.retain();
+            return this;
+        }
+
+        @Override
+        public FileRegion retain(int increment) {
+            region.retain(increment);
+            return this;
+        }
+
+        @Override
+        public FileRegion touch() {
+            region.touch();
+            return this;
+        }
+
+        @Override
+        public FileRegion touch(Object hint) {
+            region.touch(hint);
+            return this;
+        }
+    }
 }