Ensure ServerChannel implementations accept multiple connections per …

…read loop (netty#11729)


Motivation:

Due how the DefaultMaxMessagesRecvByteBufAllocator is implemented we did only ever accept one connection per read loop even thought there might be more pending.

Modifications:

- Introduce ServerChannelRecvByteBufAllocator and use it for all our ServerChannel implementations

Result:

Fixes netty#11708

pom.xml

@@ -1033,6 +1033,7 @@
               <!-- Added to fix false-positive -->
               <exclude>io.netty.handler.codec.dns.TcpDnsQueryDecoder</exclude>
               <exclude>io.netty.handler.codec.dns.TcpDnsResponseEncoder</exclude>
+              <exclude>io.netty.channel.ServerChannelRecvByteBufAllocator</exclude>
             </excludes>
           </parameter>
         </configuration>

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

@@ -36,6 +36,10 @@ public class EpollChannelConfig extends DefaultChannelConfig {
         super(channel);
     }
 
+    EpollChannelConfig(AbstractEpollChannel channel, RecvByteBufAllocator recvByteBufAllocator) {
+        super(channel, recvByteBufAllocator);
+    }
+
     @Override
     public Map<ChannelOption<?>, Object> getOptions() {
         return getOptions(super.getOptions(), EpollChannelOption.EPOLL_MODE);

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

@@ -20,6 +20,7 @@
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
+import io.netty.channel.ServerChannelRecvByteBufAllocator;
 import io.netty.channel.WriteBufferWaterMark;
 import io.netty.channel.socket.ServerSocketChannelConfig;
 import io.netty.util.NetUtil;
@@ -38,7 +39,7 @@ public class EpollServerChannelConfig extends EpollChannelConfig implements Serv
     private volatile int pendingFastOpenRequestsThreshold;
 
     EpollServerChannelConfig(AbstractEpollChannel channel) {
-        super(channel);
+        super(channel, new ServerChannelRecvByteBufAllocator());
     }
 
     @Override

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

@@ -38,6 +38,10 @@ public class KQueueChannelConfig extends DefaultChannelConfig {
         super(channel);
     }
 
+    KQueueChannelConfig(AbstractKQueueChannel channel, RecvByteBufAllocator recvByteBufAllocator) {
+        super(channel, recvByteBufAllocator);
+    }
+
     @Override
     @SuppressWarnings("deprecation")
     public Map<ChannelOption<?>, Object> getOptions() {

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

@@ -20,6 +20,7 @@
 import io.netty.channel.ChannelOption;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
+import io.netty.channel.ServerChannelRecvByteBufAllocator;
 import io.netty.channel.WriteBufferWaterMark;
 import io.netty.channel.socket.ServerSocketChannelConfig;
 import io.netty.util.NetUtil;
@@ -38,7 +39,7 @@ public class KQueueServerChannelConfig extends KQueueChannelConfig implements Se
     private volatile int backlog = NetUtil.SOMAXCONN;
 
     KQueueServerChannelConfig(AbstractKQueueChannel channel) {
-        super(channel);
+        super(channel, new ServerChannelRecvByteBufAllocator());
     }
 
     @Override

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

@@ -25,6 +25,7 @@
 import io.netty.channel.DefaultChannelConfig;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
+import io.netty.channel.ServerChannelRecvByteBufAllocator;
 import io.netty.channel.WriteBufferWaterMark;
 import io.netty.util.NetUtil;
 import io.netty.util.internal.ObjectUtil;
@@ -45,7 +46,7 @@ public class DefaultSctpServerChannelConfig extends DefaultChannelConfig impleme
      */
     public DefaultSctpServerChannelConfig(
             io.netty.channel.sctp.SctpServerChannel channel, SctpServerChannel javaChannel) {
-        super(channel);
+        super(channel, new ServerChannelRecvByteBufAllocator());
         this.javaChannel = ObjectUtil.checkNotNull(javaChannel, "javaChannel");
     }
 

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

@@ -26,6 +26,7 @@
  * and also prevents overflow.
  */
 public abstract class DefaultMaxMessagesRecvByteBufAllocator implements MaxMessagesRecvByteBufAllocator {
+    private final boolean ignoreBytesRead;
     private volatile int maxMessagesPerRead;
     private volatile boolean respectMaybeMoreData = true;
 
@@ -34,6 +35,11 @@ public DefaultMaxMessagesRecvByteBufAllocator() {
     }
 
     public DefaultMaxMessagesRecvByteBufAllocator(int maxMessagesPerRead) {
+        this(maxMessagesPerRead, false);
+    }
+
+    DefaultMaxMessagesRecvByteBufAllocator(int maxMessagesPerRead, boolean ignoreBytesRead) {
+        this.ignoreBytesRead = ignoreBytesRead;
         maxMessagesPerRead(maxMessagesPerRead);
     }
 
@@ -141,8 +147,7 @@ public boolean continueReading() {
         public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
             return config.isAutoRead() &&
                    (!respectMaybeMoreData || maybeMoreDataSupplier.get()) &&
-                   totalMessages < maxMessagePerRead &&
-                   totalBytesRead > 0;
+                   totalMessages < maxMessagePerRead && (ignoreBytesRead || totalBytesRead > 0);
         }
 
         @Override

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

@@ -0,0 +1,35 @@
+/*
+ * 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.channel;
+
+/**
+ * {@link MaxMessagesRecvByteBufAllocator} implementation which should be used for {@link ServerChannel}s.
+ */
+public final class ServerChannelRecvByteBufAllocator extends DefaultMaxMessagesRecvByteBufAllocator {
+    public ServerChannelRecvByteBufAllocator() {
+        super(1, true);
+    }
+
+    @Override
+    public Handle newHandle() {
+        return new MaxMessageHandle() {
+            @Override
+            public int guess() {
+                return 128;
+            }
+        };
+    }
+}

transport/src/main/java/io/netty/channel/local/LocalServerChannel.java

@@ -23,6 +23,7 @@
 import io.netty.channel.PreferHeapByteBufAllocator;
 import io.netty.channel.RecvByteBufAllocator;
 import io.netty.channel.ServerChannel;
+import io.netty.channel.ServerChannelRecvByteBufAllocator;
 import io.netty.channel.SingleThreadEventLoop;
 import io.netty.util.concurrent.SingleThreadEventExecutor;
 
@@ -35,7 +36,8 @@
  */
 public class LocalServerChannel extends AbstractServerChannel {
 
-    private final ChannelConfig config = new DefaultChannelConfig(this);
+    private final ChannelConfig config =
+            new DefaultChannelConfig(this, new ServerChannelRecvByteBufAllocator()) { };
     private final Queue<Object> inboundBuffer = new ArrayDeque<Object>();
     private final Runnable shutdownHook = new Runnable() {
         @Override

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

@@ -21,6 +21,7 @@
 import io.netty.channel.DefaultChannelConfig;
 import io.netty.channel.MessageSizeEstimator;
 import io.netty.channel.RecvByteBufAllocator;
+import io.netty.channel.ServerChannelRecvByteBufAllocator;
 import io.netty.channel.WriteBufferWaterMark;
 import io.netty.util.NetUtil;
 import io.netty.util.internal.ObjectUtil;
@@ -47,7 +48,7 @@ public class DefaultServerSocketChannelConfig extends DefaultChannelConfig
      * Creates a new instance.
      */
     public DefaultServerSocketChannelConfig(ServerSocketChannel channel, ServerSocket javaSocket) {
-        super(channel);
+        super(channel, new ServerChannelRecvByteBufAllocator());
         this.javaSocket = ObjectUtil.checkNotNull(javaSocket, "javaSocket");
     }
 

transport/src/test/java/io/netty/channel/DefaultMaxMessagesRecvByteBufAllocatorTest.java

@@ -0,0 +1,76 @@
+/*
+ * 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.channel;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DefaultMaxMessagesRecvByteBufAllocatorTest {
+
+    private DefaultMaxMessagesRecvByteBufAllocator newAllocator(boolean ignoreReadBytes) {
+        return new DefaultMaxMessagesRecvByteBufAllocator(2, ignoreReadBytes) {
+            @Override
+            public Handle newHandle() {
+                return new MaxMessageHandle() {
+                    @Override
+                    public int guess() {
+                        return 0;
+                    }
+                };
+            }
+        };
+    }
+
+    @Test
+    public void testRespectReadBytes() {
+        DefaultMaxMessagesRecvByteBufAllocator allocator = newAllocator(false);
+        RecvByteBufAllocator.Handle handle = allocator.newHandle();
+
+        EmbeddedChannel channel = new EmbeddedChannel();
+        handle.reset(channel.config());
+        handle.incMessagesRead(1);
+        assertFalse(handle.continueReading());
+
+        handle.reset(channel.config());
+        handle.incMessagesRead(1);
+        handle.attemptedBytesRead(1);
+        handle.lastBytesRead(1);
+        assertTrue(handle.continueReading());
+        channel.finish();
+    }
+
+    @Test
+    public void testIgnoreReadBytes() {
+        DefaultMaxMessagesRecvByteBufAllocator allocator = newAllocator(true);
+        RecvByteBufAllocator.Handle handle = allocator.newHandle();
+
+        EmbeddedChannel channel = new EmbeddedChannel();
+        handle.reset(channel.config());
+        handle.incMessagesRead(1);
+        assertTrue(handle.continueReading());
+        handle.incMessagesRead(1);
+        assertFalse(handle.continueReading());
+
+        handle.reset(channel.config());
+        handle.attemptedBytesRead(0);
+        handle.lastBytesRead(0);
+        assertTrue(handle.continueReading());
+        channel.finish();
+    }
+}