make tokio::io::empty cooperative

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<()>> {
+        let _ = 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]>> {
+        let _ = 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::*;

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() => {}
+    }
+}