io: fix memory leak on shutdown

In some cases, a cycle is created between I/O driver wakers and the I/O
driver resource slab. This patch clears stored wakers when an I/O
resource is dropped, breaking the cycle.

Fixes #3228

.github/workflows/ci.yml

@@ -28,6 +28,7 @@ jobs:
       - clippy
       - docs
       - loom
+      - valgrind
     steps:
       - run: exit 0
 
@@ -67,6 +68,27 @@ jobs:
         run: cargo hack test --each-feature
         working-directory: tests-build
 
+  valgrind:
+    name: valgrind
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install Rust
+        run: rustup update stable
+
+      - name: Install Valgrind
+        run: |
+          sudo apt-get update -y
+          sudo apt-get install -y valgrind
+
+      # Compile tests
+      - name: cargo build
+        run: cargo build -p tests-integration --bin test-mem
+
+      # Run with valgrind
+      - name: Run valgrind
+        run: valgrind  --leak-check=full --show-leak-kinds=all ./target/debug/test-mem
+
   test-unstable:
     name: test tokio full --unstable
     runs-on: ${{ matrix.os }}

tests-integration/Cargo.toml

@@ -5,7 +5,17 @@ authors = ["Tokio Contributors <[email protected]>"]
 edition = "2018"
 publish = false
 
+[[bin]]
+name = "test-cat"
+
+[[bin]]
+name = "test-mem"
+required-features = ["rt-net"]
+
 [features]
+# For mem check
+rt-net = ["tokio/rt", "tokio/rt-multi-thread", "tokio/net"]
+
 full = [
     "macros",
     "rt",
@@ -23,6 +33,4 @@ rt-multi-thread = ["rt", "tokio/rt-multi-thread"]
 tokio = { path = "../tokio" }
 tokio-test = { path = "../tokio-test", optional = true }
 doc-comment = "0.3.1"
-
-[dev-dependencies]
 futures = { version = "0.3.0", features = ["async-await"] }

tests-integration/src/bin/test-mem.rs

@@ -0,0 +1,21 @@
+use futures::future::poll_fn;
+
+fn main() {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .enable_io()
+        .build()
+        .unwrap();
+
+    rt.block_on(async {
+        let  listener = tokio::net::TcpListener::bind("0.0.0.0:0").await.unwrap();
+        tokio::spawn(async move {
+            loop { 
+                poll_fn(|cx| listener.poll_accept(cx)).await.unwrap();
+            }
+        });
+    });
+
+    std::thread::sleep(std::time::Duration::from_millis(50));
+    drop(rt);
+}

tokio/src/io/driver/registration.rs

@@ -205,6 +205,12 @@ impl Registration {
     }
 }
 
+impl Drop for Registration {
+    fn drop(&mut self) {
+        self.shared.clear_wakers();
+    }
+}
+
 fn gone() -> io::Error {
     io::Error::new(io::ErrorKind::Other, "IO driver has terminated")
 }

tokio/src/io/driver/scheduled_io.rs

@@ -355,6 +355,12 @@ impl ScheduledIo {
         // result isn't important
         let _ = self.set_readiness(None, Tick::Clear(event.tick), |curr| curr - mask_no_closed);
     }
+
+    pub(crate) fn clear_wakers(&self) {
+        let mut waiters = self.waiters.lock();
+        waiters.reader.take();
+        waiters.writer.take();
+    }
 }
 
 impl Drop for ScheduledIo {