Fix timeout when auto read is disabled late in io_uring (#14989)

yawkat · web-flow · commit 9b3a5418e09c · 2025-04-03T14:25:53.000+02:00
Motivation:

If auto read is disabled late (during a channelRead), but there is a
read call during the next channelReadComplete, that read call would be
ignored, leading to the channel freezing up and the connection timing
out.

Modification:

When clearRead cancels the scheduled POLLIN, clear the iostate
immediately, so that the next `read()` will correctly schedule a new
POLLIN.

Not sure if this is the proper way to do it. Maybe this should go into
cancelComplete0 instead? But then we'd have to add some mechanism to
hold back the read call. I think the risk of duplicate reads that this
change adds is better than no reads.

Result:

Echo server does not freeze up anymore after turning off auto read
during a read operation.
diff --git a/transport-classes-io_uring/src/main/java/io/netty/channel/uring/AbstractIoUringChannel.java b/transport-classes-io_uring/src/main/java/io/netty/channel/uring/AbstractIoUringChannel.java
@@ -184,6 +184,7 @@ private void clearRead() {
             long id = registration.submit(
                     IoUringIoOps.newAsyncCancel((byte) 0, pollInId, Native.IORING_OP_POLL_ADD));
             assert id != 0;
+            ioState &= ~POLL_IN_SCHEDULED;
         }
         // Also cancel all outstanding reads as the user did signal there is no more desire to read.
         cancelOutstandingReads(registration(), numOutstandingReads);
diff --git a/transport-native-io_uring/src/test/java/io/netty/channel/uring/IoUringAutoReadTest.java b/transport-native-io_uring/src/test/java/io/netty/channel/uring/IoUringAutoReadTest.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2025 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.uring;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.IoEventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.socket.ServerSocketChannel;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+public class IoUringAutoReadTest {
+
+    @BeforeAll
+    public static void loadJNI() {
+        assumeTrue(IoUring.isAvailable());
+    }
+
+    @Test
+    @Timeout(value = 1, unit = TimeUnit.MINUTES)
+    public void testLateAutoRead() throws Exception {
+        IoEventLoopGroup group = new MultiThreadIoEventLoopGroup(1, IoUringIoHandler.newFactory());
+        try {
+            ServerSocketChannel server = (ServerSocketChannel) new ServerBootstrap()
+                    .group(group)
+                    .channel(IoUringServerSocketChannel.class)
+                    .childHandler(new ChannelInboundHandlerAdapter() {
+                        @Override
+                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                            ctx.channel().config().setAutoRead(false);
+                            ctx.writeAndFlush(msg, ctx.voidPromise());
+                        }
+
+                        @Override
+                        public void channelReadComplete(ChannelHandlerContext ctx) {
+                            ctx.read();
+                        }
+                    })
+                    .bind(0).sync().channel();
+
+            try (Socket sock = new Socket(server.localAddress().getAddress(), server.localAddress().getPort())) {
+                OutputStream out = sock.getOutputStream();
+                InputStream in = sock.getInputStream();
+
+                out.write(1);
+                out.flush();
+                Assertions.assertEquals(1, in.read());
+
+                out.write(2);
+                out.flush();
+                Assertions.assertEquals(2, in.read());
+            }
+        } finally {
+            group.shutdownGracefully();
+        }
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -184,6 +184,7 @@ private void clearRead() {`
`184`	`184`	`long id = registration.submit(`
`185`	`185`	`IoUringIoOps.newAsyncCancel((byte) 0, pollInId, Native.IORING_OP_POLL_ADD));`
`186`	`186`	`assert id != 0;`
	`187`	`+ ioState &= ~POLL_IN_SCHEDULED;`
`187`	`188`	`}`
`188`	`189`	`// Also cancel all outstanding reads as the user did signal there is no more desire to read.`
`189`	`190`	`cancelOutstandingReads(registration(), numOutstandingReads);`