io: fix memory leak during shutdown (#3477)

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,28 @@ 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 --features rt-net --bin test-mem
+        working-directory: tests-integration
+
+      # 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 <team@tokio.rs>"]
 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/CHANGELOG.md

@@ -1,3 +1,8 @@
+# 1.0.3 (January 28, 2020)
+
+### Fixed
+- io: memory leak during shutdown (#3477).
+
 # 1.0.2 (January 14, 2020)
 
 ### Fixed

tokio/Cargo.toml

@@ -8,12 +8,12 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.0.x" git tag.
-version = "1.0.2"
+version = "1.0.3"
 edition = "2018"
 authors = ["Tokio Contributors <team@tokio.rs>"]
 license = "MIT"
 readme = "README.md"
-documentation = "https://docs.rs/tokio/1.0.2/tokio/"
+documentation = "https://docs.rs/tokio/1.0.3/tokio/"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
 description = """

tokio/src/io/driver/registration.rs

@@ -205,6 +205,19 @@ impl Registration {
     }
 }
 
+impl Drop for Registration {
+    fn drop(&mut self) {
+        // It is possible for a cycle to be created between wakers stored in
+        // `ScheduledIo` instances and `Arc<driver::Inner>`. To break this
+        // cycle, wakers are cleared. This is an imperfect solution as it is
+        // possible to store a `Registration` in a waker. In this case, the
+        // cycle would remain.
+        //
+        // See tokio-rs/tokio#3481 for more details.
+        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 {

tokio/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio/1.0.2")]
+#![doc(html_root_url = "https://docs.rs/tokio/1.0.3")]
 #![allow(
     clippy::cognitive_complexity,
     clippy::large_enum_variant,