DefaultMaxMessagesRecvByteBufAllocator support an option to ignore maybeMoreData

Motivation:
If large amounts of data is being transferred it is difficult to correlate the amount we attempt to read vs the maximum amount that the OS will actually buffer and deliver to the application. For exmaple some OSes may dynicamlly update the SO_RCVBUF size or otherwise dynamically adjust how much data is delieved to the application. In these circumstances it can reduce latency to just call read() on the socket another time to see if there is really any data remaining instead of giving up the maxMessagesPerRead quantum and going back to the selector to read later.

Motifications:
- Add DefaultMaxMessagesRecvByteBufAllocator#respectMaybeMoreData which provides a way to ignore the maybeMoreData function which may not account for the current data pending, and if it does this maybe racy.

Result:
Option to always use the full maxMessagesPerRead quantum before going back to the selector.

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

@@ -195,4 +195,10 @@ public class AdaptiveRecvByteBufAllocator extends DefaultMaxMessagesRecvByteBufA
     public Handle newHandle() {
         return new HandleImpl(minIndex, maxIndex, initial);
     }
+
+    @Override
+    public AdaptiveRecvByteBufAllocator respectMaybeMoreData(boolean respectMaybeMoreData) {
+        super.respectMaybeMoreData(respectMaybeMoreData);
+        return this;
+    }
 }

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

@@ -25,6 +25,7 @@ import io.netty.util.UncheckedBooleanSupplier;
  */
 public abstract class DefaultMaxMessagesRecvByteBufAllocator implements MaxMessagesRecvByteBufAllocator {
     private volatile int maxMessagesPerRead;
+    private volatile boolean respectMaybeMoreData = true;
 
     public DefaultMaxMessagesRecvByteBufAllocator() {
         this(1);
@@ -48,6 +49,38 @@ public abstract class DefaultMaxMessagesRecvByteBufAllocator implements MaxMessa
         return this;
     }
 
+    /**
+     * Determine if future instances of {@link #newHandle()} will stop reading if we think there is no more data.
+     * @param respectMaybeMoreData
+     * <ul>
+     *     <li>{@code true} to stop reading if we think there is no more data. This may save a system call to read from
+     *          the socket, but if data has arrived in a racy fashion we may give up our {@link #maxMessagesPerRead()}
+     *          quantum and have to wait for the selector to notify us of more data.</li>
+     *     <li>{@code false} to keep reading (up to {@link #maxMessagesPerRead()}) or until there is no data when we
+     *          attempt to read.</li>
+     * </ul>
+     * @return {@code this}.
+     */
+    public DefaultMaxMessagesRecvByteBufAllocator respectMaybeMoreData(boolean respectMaybeMoreData) {
+        this.respectMaybeMoreData = respectMaybeMoreData;
+        return this;
+    }
+
+    /**
+     * Get if future instances of {@link #newHandle()} will stop reading if we think there is no more data.
+     * @return
+     * <ul>
+     *     <li>{@code true} to stop reading if we think there is no more data. This may save a system call to read from
+     *          the socket, but if data has arrived in a racy fashion we may give up our {@link #maxMessagesPerRead()}
+     *          quantum and have to wait for the selector to notify us of more data.</li>
+     *     <li>{@code false} to keep reading (up to {@link #maxMessagesPerRead()}) or until there is no data when we
+     *          attempt to read.</li>
+     * </ul>
+     */
+    public final boolean respectMaybeMoreData() {
+        return respectMaybeMoreData;
+    }
+
     /**
      * Focuses on enforcing the maximum messages per read condition for {@link #continueReading()}.
      */
@@ -58,6 +91,7 @@ public abstract class DefaultMaxMessagesRecvByteBufAllocator implements MaxMessa
         private int totalBytesRead;
         private int attemptedBytesRead;
         private int lastBytesRead;
+        private final boolean respectMaybeMoreData = DefaultMaxMessagesRecvByteBufAllocator.this.respectMaybeMoreData;
         private final UncheckedBooleanSupplier defaultMaybeMoreSupplier = new UncheckedBooleanSupplier() {
             @Override
             public boolean get() {
@@ -106,7 +140,7 @@ public abstract class DefaultMaxMessagesRecvByteBufAllocator implements MaxMessa
         @Override
         public boolean continueReading(UncheckedBooleanSupplier maybeMoreDataSupplier) {
             return config.isAutoRead() &&
-                   maybeMoreDataSupplier.get() &&
+                   (!respectMaybeMoreData || maybeMoreDataSupplier.get()) &&
                    totalMessages < maxMessagePerRead &&
                    totalBytesRead > 0;
         }

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

@@ -53,4 +53,10 @@ public class FixedRecvByteBufAllocator extends DefaultMaxMessagesRecvByteBufAllo
     public Handle newHandle() {
         return new HandleImpl(bufferSize);
     }
+
+    @Override
+    public FixedRecvByteBufAllocator respectMaybeMoreData(boolean respectMaybeMoreData) {
+        super.respectMaybeMoreData(respectMaybeMoreData);
+        return this;
+    }
 }