From eb1af7f29caaa1cd9e84b9fcbd87b9d62b546de3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Braulio=20Valdivielso=20Mart=C3=ADnez?=
 <brlvldvlsmrtnz@gmail.com>
Date: Fri, 10 Dec 2021 10:08:49 +0000
Subject: [PATCH] io: make `tokio::io::empty` cooperative (#4300)

Reads and buffered reads from a `tokio::io::empty` were always marked
as ready. That makes sense, given that there is nothing to wait for.
However, doing repeated reads on the `empty` could stall the event
loop and prevent other tasks from making progress.

This change uses tokio's coop system to yield control back to the
executor when appropriate.

Note that the issue that originally triggered this PR is not fixed
yet, because the `timeout` function will not poll the timer after
empty::read runs out of budget. A different change will be needed to
address that.

Refs: #4291
---
 tokio/src/io/util/empty.rs   | 20 ++++++++++++++++++--
 tokio/tests/io_util_empty.rs | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 50 insertions(+), 2 deletions(-)
 create mode 100644 tokio/tests/io_util_empty.rs

diff --git a/tokio/src/io/util/empty.rs b/tokio/src/io/util/empty.rs
index f964d18e6..77db60e40 100644
--- a/tokio/src/io/util/empty.rs
+++ b/tokio/src/io/util/empty.rs
@@ -50,16 +50,18 @@ impl AsyncRead for Empty {
     #[inline]
     fn poll_read(
         self: Pin<&mut Self>,
-        _: &mut Context<'_>,
+        cx: &mut Context<'_>,
         _: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
+        ready!(poll_proceed_and_make_progress(cx));
         Poll::Ready(Ok(()))
     }
 }
 
 impl AsyncBufRead for Empty {
     #[inline]
-    fn poll_fill_buf(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        ready!(poll_proceed_and_make_progress(cx));
         Poll::Ready(Ok(&[]))
     }
 
@@ -73,6 +75,20 @@ impl fmt::Debug for Empty {
     }
 }
 
+cfg_coop! {
+    fn poll_proceed_and_make_progress(cx: &mut Context<'_>) -> Poll<()> {
+        let coop = ready!(crate::coop::poll_proceed(cx));
+        coop.made_progress();
+        Poll::Ready(())
+    }
+}
+
+cfg_not_coop! {
+    fn poll_proceed_and_make_progress(_: &mut Context<'_>) -> Poll<()> {
+        Poll::Ready(())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/tokio/tests/io_util_empty.rs b/tokio/tests/io_util_empty.rs
new file mode 100644
index 000000000..e49cd17fc
--- /dev/null
+++ b/tokio/tests/io_util_empty.rs
@@ -0,0 +1,32 @@
+#![cfg(feature = "full")]
+use tokio::io::{AsyncBufReadExt, AsyncReadExt};
+
+#[tokio::test]
+async fn empty_read_is_cooperative() {
+    tokio::select! {
+        biased;
+
+        _ = async {
+            loop {
+                let mut buf = [0u8; 4096];
+                let _ = tokio::io::empty().read(&mut buf).await;
+            }
+        } => {},
+        _ = tokio::task::yield_now() => {}
+    }
+}
+
+#[tokio::test]
+async fn empty_buf_reads_are_cooperative() {
+    tokio::select! {
+        biased;
+
+        _ = async {
+            loop {
+                let mut buf = String::new();
+                let _ = tokio::io::empty().read_line(&mut buf).await;
+            }
+        } => {},
+        _ = tokio::task::yield_now() => {}
+    }
+}