[netty#3709] Ensure all data is read from socket when EPOLLRDUP is re…

…ceived

Motivation:

When EPOLLRDHUP is received we need to try to read at least one time to ensure
that we read all pending data from the socket. Otherwise we may loose data.

Modifications:

- Ensure we read all data from socket
- Ensure file descriptor is closed on doClose() even if doDeregister() throws an Exception.
- Only handle either EPOLLRDHUP or EPOLLIN as only one is needed to detect connection reset.

Result:

No more data loss on connection reset.

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

@@ -99,12 +99,14 @@ public ChannelMetadata metadata() {
     @Override
     protected void doClose() throws Exception {
         active = false;
-
-        // deregister from epoll now
-        doDeregister();
-
-        FileDescriptor fd = fileDescriptor;
-        fd.close();
+        try {
+            // deregister from epoll now
+            doDeregister();
+        } finally {
+            // Ensure the file descriptor is closed in all cases.
+            FileDescriptor fd = fileDescriptor;
+            fd.close();
+        }
     }
 
     @Override

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

@@ -765,11 +765,14 @@ private boolean doFinishConnect() throws Exception {
 
         @Override
         void epollRdHupReady() {
-            // Just call closeOnRead(). There is no need to trigger a read as this
-            // will result in an IOException anyway.
-            //
-            // See https://github.com/netty/netty/issues/3539
-            closeOnRead(pipeline());
+            if (isActive()) {
+                // If it is still active, we need to call epollInReady as otherwise we may miss to
+                // read pending data from the underyling file descriptor.
+                // See https://github.com/netty/netty/issues/3709
+                epollInReady();
+            } else {
+                closeOnRead(pipeline());
+            }
         }
 
         @Override

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

@@ -310,26 +310,22 @@ private void processReady(EpollEventArray events, int ready) {
                 final long ev = events.events(i);
 
                 AbstractEpollChannel ch = channels.get(fd);
-                if (ch != null && ch.isOpen()) {
-
+                if (ch != null) {
                     AbstractEpollUnsafe unsafe = (AbstractEpollUnsafe) ch.unsafe();
 
-                    // We need to check if the channel is still open before try to trigger the
-                    // callbacks.
-                    // See https://github.com/netty/netty/issues/3443
-                    if ((ev & Native.EPOLLRDHUP) != 0 && ch.isOpen()) {
+                    // First check if EPOLLIN was set, in this case we do not need to check for
+                    // EPOLLRDHUP as EPOLLIN will handle connection-reset case as well.
+                    if ((ev & Native.EPOLLIN) != 0) {
+                        // Something is ready to read, so consume it now
+                        unsafe.epollInReady();
+                    } else if ((ev & Native.EPOLLRDHUP) != 0) {
                         unsafe.epollRdHupReady();
                     }
 
                     if ((ev & Native.EPOLLOUT) != 0 && ch.isOpen()) {
                         // force flush of data as the epoll is writable again
                         unsafe.epollOutReady();
                     }
-
-                    if ((ev & Native.EPOLLIN) != 0 && ch.isOpen()) {
-                        // Something is ready to read, so consume it now
-                        unsafe.epollInReady();
-                    }
                 } else {
                     // We received an event for an fd which we not use anymore. Remove it from the epoll_event set.
                     try {