io: split read/write ticks in ScheduledIo to fix poll_read livelock

tokio/src/runtime/io/driver.rs

@@ -72,7 +72,11 @@ pub(crate) struct Handle {
 
 #[derive(Debug)]
 pub(crate) struct ReadyEvent {
-    pub(super) tick: u8,
+    /// Generation counter for read-side readiness (readable, read closed,
+    /// error, and on Linux, priority).
+    pub(super) read_tick: u8,
+    /// Generation counter for write-side readiness (writable, write closed).
+    pub(super) write_tick: u8,
     pub(crate) ready: Ready,
     pub(super) is_shutdown: bool,
 }
@@ -82,7 +86,8 @@ cfg_net_unix!(
         pub(crate) fn with_ready(&self, ready: Ready) -> Self {
             Self {
                 ready,
-                tick: self.tick,
+                read_tick: self.read_tick,
+                write_tick: self.write_tick,
                 is_shutdown: self.is_shutdown,
             }
         }
@@ -96,8 +101,12 @@ pub(super) enum Direction {
 }
 
 pub(super) enum Tick {
-    Set,
-    Clear(u8),
+    /// Clear readiness bits. The read and write generation counters are checked
+    /// independently: each direction's tick is only validated when that
+    /// direction is `Some`, so clearing read readiness is not invalidated by a
+    /// concurrent write-only event (and vice versa). See
+    /// <https://github.com/tokio-rs/tokio/issues/8054>.
+    Clear { read: Option<u8>, write: Option<u8> },
 }
 
 const TOKEN_WAKEUP: mio::Token = mio::Token(0);
@@ -215,7 +224,7 @@ impl Driver {
                 // an `Arc<ScheduledIo>` so we can safely cast this to a ref.
                 let io: &ScheduledIo = unsafe { &*ptr };
 
-                io.set_readiness(Tick::Set, |curr| curr | ready);
+                io.merge_readiness_from_driver(ready);
                 io.wake(ready);
 
                 ready_count += 1;