io: fix possible I/O resource hang (#6134)

Use a per-resource tick instead of a single global tick counter. This
prevents the hang issue described by #6133.

Fixes: #6133

tokio/src/runtime/io/driver.rs

@@ -18,10 +18,6 @@ use std::time::Duration;
 
 /// I/O driver, backed by Mio.
 pub(crate) struct Driver {
-    /// Tracks the number of times `turn` is called. It is safe for this to wrap
-    /// as it is mostly used to determine when to call `compact()`.
-    tick: u8,
-
     /// True when an event with the signal token is received
     signal_ready: bool,
 
@@ -77,7 +73,7 @@ pub(super) enum Direction {
 }
 
 pub(super) enum Tick {
-    Set(u8),
+    Set,
     Clear(u8),
 }
 
@@ -102,7 +98,6 @@ impl Driver {
         let registry = poll.registry().try_clone()?;
 
         let driver = Driver {
-            tick: 0,
             signal_ready: false,
             events: mio::Events::with_capacity(nevents),
             poll,
@@ -145,8 +140,6 @@ impl Driver {
     fn turn(&mut self, handle: &Handle, max_wait: Option<Duration>) {
         debug_assert!(!handle.registrations.is_shutdown(&handle.synced.lock()));
 
-        self.tick = self.tick.wrapping_add(1);
-
         handle.release_pending_registrations();
 
         let events = &mut self.events;
@@ -184,7 +177,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(self.tick), |curr| curr | ready);
+                io.set_readiness(Tick::Set, |curr| curr | ready);
                 io.wake(ready);
 
                 ready_count += 1;

tokio/src/runtime/io/scheduled_io.rs

@@ -219,17 +219,21 @@ impl ScheduledIo {
             let current_readiness = Ready::from_usize(current);
             let new = f(current_readiness);
 
-            let next = match tick {
-                Tick::Set(t) => TICK.pack(t as usize, new.as_usize()),
+            let new_tick = match tick {
+                Tick::Set => {
+                    let current = TICK.unpack(current);
+                    current.wrapping_add(1) % (TICK.max_value() + 1)
+                }
                 Tick::Clear(t) => {
                     if TICK.unpack(current) as u8 != t {
                         // Trying to clear readiness with an old event!
                         return;
                     }
 
-                    TICK.pack(t as usize, new.as_usize())
+                    t as usize
                 }
             };
+            let next = TICK.pack(new_tick, new.as_usize());
 
             match self
                 .readiness