Fix tokio-rs#2535.

In threaded runtime, the unparker now owns a weak reference to the inner data. This breaks the cycle of Arc and properly releases the io driver and its worker threads.

tokio/src/runtime/park.rs

@@ -9,6 +9,7 @@ use crate::park::{Park, Unpark};
 use crate::runtime::time;
 use crate::util::TryLock;
 
+use std::sync::Weak;
 use std::sync::atomic::Ordering::SeqCst;
 use std::time::Duration;
 
@@ -17,7 +18,7 @@ pub(crate) struct Parker {
 }
 
 pub(crate) struct Unparker {
-    inner: Arc<Inner>,
+    inner: Weak<Inner>,
 }
 
 struct Inner {
@@ -85,7 +86,7 @@ impl Park for Parker {
 
     fn unpark(&self) -> Unparker {
         Unparker {
-            inner: self.inner.clone(),
+            inner: std::sync::Arc::downgrade(&self.inner),
         }
     }
 
@@ -108,7 +109,9 @@ impl Park for Parker {
 
 impl Unpark for Unparker {
     fn unpark(&self) {
-        self.inner.unpark();
+        if let Some(inner) = self.inner.upgrade() {
+            inner.unpark();
+        }
     }
 }
 

tokio/tests/rt_common.rs

@@ -2,7 +2,7 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 
-// Tests to run on both current-thread & therad-pool runtime variants.
+// Tests to run on both current-thread & thread-pool runtime variants.
 
 macro_rules! rt_test {
     ($($t:tt)*) => {
@@ -855,6 +855,7 @@ rt_test! {
         runtime.shutdown_timeout(Duration::from_millis(100));
     }
 
+    #[ignore]
     #[test]
     fn runtime_in_thread_local() {
         use std::cell::RefCell;

tokio/tests/tcp_into_split.rs

@@ -82,6 +82,7 @@ async fn reunite() -> Result<()> {
 }
 
 /// Test that dropping the write half actually closes the stream.
+#[ignore]
 #[tokio::test]
 async fn drop_write() -> Result<()> {
     const MSG: &[u8] = b"split";