runtime: fix remote abort (#3942)

Co-authored-by: @tijsvd

.github/workflows/ci.yml

@@ -1,8 +1,8 @@
 on:
   push:
-    branches: ["master"]
+    branches: ["v0.3.x"]
   pull_request:
-    branches: ["master"]
+    branches: ["v0.3.x"]
 
 name: CI
 

.github/workflows/stress-test.yml

@@ -19,7 +19,9 @@ jobs:
         run: rustup update stable
 
       - name: Install Valgrind
-        run: sudo apt-get install -y valgrind
+        run: |
+          sudo apt-get update -y
+          sudo apt-get install -y valgrind
 
       # Compiles each of the stress test examples.
       - name: Compile stress test examples

tokio-macros/CHANGELOG.md

@@ -1,3 +1,10 @@
+# 0.3.2 (December 19, 2020)
+
+### Fixed
+
+- fix outdated macro documentation ([#3180])
+- add portability note to `tokio::main` ([#3199])
+
 # 0.3.1 (October 25, 2020)
 
 ### Fixed
@@ -51,3 +58,5 @@
 [#2177]: https://github.com/tokio-rs/tokio/pull/2177
 [#2225]: https://github.com/tokio-rs/tokio/pull/2225
 [#3038]: https://github.com/tokio-rs/tokio/pull/3038
+[#3180]: https://github.com/tokio-rs/tokio/pull/3180
+[#3199]: https://github.com/tokio-rs/tokio/pull/3199

tokio-macros/Cargo.toml

@@ -7,13 +7,13 @@ name = "tokio-macros"
 #   - Cargo.toml
 # - Update CHANGELOG.md.
 # - Create "v0.3.x" git tag.
-version = "0.3.1"
+version = "0.3.2"
 edition = "2018"
 authors = ["Tokio Contributors <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-macros/0.3.1/tokio_macros"
+documentation = "https://docs.rs/tokio-macros/0.3.2/tokio_macros"
 description = """
 Tokio's proc macros.
 """

tokio-macros/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio-macros/0.3.1")]
+#![doc(html_root_url = "https://docs.rs/tokio-macros/0.3.2")]
 #![allow(clippy::needless_doctest_main)]
 #![warn(
     missing_debug_implementations,

tokio-util/src/time/wheel/level.rs

@@ -46,7 +46,7 @@ impl<T: Stack> Level<T> {
             () => {
                 T::default()
             };
-        };
+        }
 
         Level {
             level,

tokio/CHANGELOG.md

@@ -1,3 +1,22 @@
+# 0.3.7 (January 28, 2021)
+
+### Changes
+
+- runtime: update panic messages to include version (#3461)
+
+# 0.3.6 (December 14, 2020)
+
+### Fixed
+- rt: fix deadlock in shutdown (#3228)
+- rt: fix panic in task abort when off rt (#3159)
+- sync: make `add_permits` panic with usize::MAX >> 3 permits (#3188)
+- time: Fix race condition in timer drop (#3229)
+- watch: fix spurious wakeup (#3244)
+
+### Added
+- example: add back udp-codec example (#3205)
+- net: add `TcpStream::into_std` (#3189)
+
 # 0.3.5 (November 30, 2020)
 
 ### Fixed

tokio/Cargo.toml

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

tokio/src/io/driver/mod.rs

@@ -260,7 +260,7 @@ cfg_rt! {
         /// flag is not enabled.
         pub(super) fn current() -> Self {
             crate::runtime::context::io_handle()
-                .expect("there is no reactor running, must be called from the context of Tokio runtime")
+                .expect("there is no reactor running, must be called from the context of a Tokio 0.3.x runtime")
         }
     }
 }
@@ -274,7 +274,7 @@ cfg_not_rt! {
         /// This function panics if there is no current reactor set, or if the `rt`
         /// feature flag is not enabled.
         pub(super) fn current() -> Self {
-            panic!("there is no reactor running, must be called from the context of Tokio runtime with `rt` enabled.")
+            panic!("there is no reactor running, must be called from the context of a Tokio 0.3.x runtime with `rt` enabled.")
         }
     }
 }

tokio/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio/0.3.5")]
+#![doc(html_root_url = "https://docs.rs/tokio/0.3.7")]
 #![allow(
     clippy::cognitive_complexity,
     clippy::large_enum_variant,
@@ -16,6 +16,7 @@
     attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
 ))]
 #![cfg_attr(docsrs, feature(doc_cfg))]
+#![allow(deprecated)]
 
 //! A runtime for writing reliable network applications without compromising speed.
 //!

tokio/src/process/mod.rs

@@ -433,7 +433,7 @@ impl Command {
     /// Basic usage:
     ///
     /// ```no_run
-    /// use tokio::process::Command;;
+    /// use tokio::process::Command;
     /// use std::process::Stdio;
     ///
     /// let command = Command::new("ls")
@@ -457,7 +457,7 @@ impl Command {
     /// Basic usage:
     ///
     /// ```no_run
-    /// use tokio::process::Command;;
+    /// use tokio::process::Command;
     /// use std::process::{Stdio};
     ///
     /// let command = Command::new("ls")

tokio/src/runtime/blocking/pool.rs

@@ -81,7 +81,7 @@ where
     F: FnOnce() -> R + Send + 'static,
     R: Send + 'static,
 {
-    let rt = context::current().expect("not currently running on the Tokio runtime.");
+    let rt = context::current().expect("not currently running on a Tokio 0.3.x runtime.");
     rt.spawn_blocking(func)
 }
 
@@ -91,7 +91,7 @@ where
     F: FnOnce() -> R + Send + 'static,
     R: Send + 'static,
 {
-    let rt = context::current().expect("not currently running on the Tokio runtime.");
+    let rt = context::current().expect("not currently running on a Tokio 0.3.x runtime.");
 
     let (task, _handle) = task::joinable(BlockingTask::new(func));
     rt.blocking_spawner.spawn(task, &rt)

tokio/src/runtime/handle.rs

@@ -97,7 +97,7 @@ impl Handle {
     /// # }
     /// ```
     pub fn current() -> Self {
-        context::current().expect("not currently running on the Tokio runtime.")
+        context::current().expect("not currently running on a Tokio 0.3.x runtime.")
     }
 
     /// Returns a Handle view over the currently running Runtime

tokio/src/runtime/task/harness.rs

@@ -285,6 +285,17 @@ where
         self.cancel_task();
     }
 
+    /// Remotely abort the task
+    ///
+    /// This is similar to `shutdown` except that it asks the runtime to perform
+    /// the shutdown. This is necessary to avoid the shutdown happening in the
+    /// wrong thread for non-Send tasks.
+    pub(super) fn remote_abort(self) {
+        if self.header().state.transition_to_notified_and_cancel() {
+            self.core().schedule(Notified(self.to_task()));
+        }
+    }
+
     // ====== internal ======
 
     fn cancel_task(self) {

tokio/src/runtime/task/join.rs

@@ -192,7 +192,7 @@ impl<T> JoinHandle<T> {
     /// ```
     pub fn abort(&self) {
         if let Some(raw) = self.raw {
-            raw.shutdown();
+            raw.remote_abort();
         }
     }
 }

tokio/src/runtime/task/raw.rs

@@ -22,6 +22,9 @@ pub(super) struct Vtable {
     /// The join handle has been dropped
     pub(super) drop_join_handle_slow: unsafe fn(NonNull<Header>),
 
+    /// The task is remotely aborted
+    pub(super) remote_abort: unsafe fn(NonNull<Header>),
+
     /// Scheduler is being shutdown
     pub(super) shutdown: unsafe fn(NonNull<Header>),
 }
@@ -33,6 +36,7 @@ pub(super) fn vtable<T: Future, S: Schedule>() -> &'static Vtable {
         dealloc: dealloc::<T, S>,
         try_read_output: try_read_output::<T, S>,
         drop_join_handle_slow: drop_join_handle_slow::<T, S>,
+        remote_abort: remote_abort::<T, S>,
         shutdown: shutdown::<T, S>,
     }
 }
@@ -89,6 +93,11 @@ impl RawTask {
         let vtable = self.header().vtable;
         unsafe { (vtable.shutdown)(self.ptr) }
     }
+
+    pub(super) fn remote_abort(self) {
+        let vtable = self.header().vtable;
+        unsafe { (vtable.remote_abort)(self.ptr) }
+    }
 }
 
 impl Clone for RawTask {
@@ -125,6 +134,11 @@ unsafe fn drop_join_handle_slow<T: Future, S: Schedule>(ptr: NonNull<Header>) {
     harness.drop_join_handle_slow()
 }
 
+unsafe fn remote_abort<T: Future, S: Schedule>(ptr: NonNull<Header>) {
+    let harness = Harness::<T, S>::from_raw(ptr);
+    harness.remote_abort()
+}
+
 unsafe fn shutdown<T: Future, S: Schedule>(ptr: NonNull<Header>) {
     let harness = Harness::<T, S>::from_raw(ptr);
     harness.shutdown()

tokio/src/runtime/task/state.rs

@@ -177,6 +177,15 @@ impl State {
         prev.will_need_queueing()
     }
 
+    /// Set the cancelled bit and transition the state to `NOTIFIED`.
+    ///
+    /// Returns `true` if the task needs to be submitted to the pool for
+    /// execution
+    pub(super) fn transition_to_notified_and_cancel(&self) -> bool {
+        let prev = Snapshot(self.val.fetch_or(NOTIFIED | CANCELLED, AcqRel));
+        prev.will_need_queueing()
+    }
+
     /// Set the `CANCELLED` bit and attempt to transition to `Running`.
     ///
     /// Returns `true` if the transition to `Running` succeeded.

tokio/src/sync/watch.rs

@@ -53,10 +53,10 @@
 
 use crate::sync::Notify;
 
+use crate::loom::sync::atomic::AtomicUsize;
+use crate::loom::sync::atomic::Ordering::{Relaxed, SeqCst};
+use crate::loom::sync::{Arc, RwLock, RwLockReadGuard};
 use std::ops;
-use std::sync::atomic::AtomicUsize;
-use std::sync::atomic::Ordering::{Relaxed, SeqCst};
-use std::sync::{Arc, RwLock, RwLockReadGuard};
 
 /// Receives values from the associated [`Sender`](struct@Sender).
 ///
@@ -241,19 +241,19 @@ impl<T> Receiver<T> {
     /// }
     /// ```
     pub async fn changed(&mut self) -> Result<(), error::RecvError> {
-        // In order to avoid a race condition, we first request a notification,
-        // **then** check the current value's version. If a new version exists,
-        // the notification request is dropped.
-        let notified = self.shared.notify_rx.notified();
+        loop {
+            // In order to avoid a race condition, we first request a notification,
+            // **then** check the current value's version. If a new version exists,
+            // the notification request is dropped.
+            let notified = self.shared.notify_rx.notified();
 
-        if let Some(ret) = maybe_changed(&self.shared, &mut self.version) {
-            return ret;
+            if let Some(ret) = maybe_changed(&self.shared, &mut self.version) {
+                return ret;
+            }
+
+            notified.await;
+            // loop around again in case the wake-up was spurious
         }
-
-        notified.await;
-
-        maybe_changed(&self.shared, &mut self.version)
-            .expect("[bug] failed to observe change after notificaton.")
     }
 }
 
@@ -390,3 +390,44 @@ impl<T> ops::Deref for Ref<'_, T> {
         self.inner.deref()
     }
 }
+
+#[cfg(all(test, loom))]
+mod tests {
+    use futures::future::FutureExt;
+    use loom::thread;
+
+    // test for https://github.com/tokio-rs/tokio/issues/3168
+    #[test]
+    fn watch_spurious_wakeup() {
+        loom::model(|| {
+            let (send, mut recv) = crate::sync::watch::channel(0i32);
+
+            send.send(1).unwrap();
+
+            let send_thread = thread::spawn(move || {
+                send.send(2).unwrap();
+                send
+            });
+
+            recv.changed().now_or_never();
+
+            let send = send_thread.join().unwrap();
+            let recv_thread = thread::spawn(move || {
+                recv.changed().now_or_never();
+                recv.changed().now_or_never();
+                recv
+            });
+
+            send.send(3).unwrap();
+
+            let mut recv = recv_thread.join().unwrap();
+            let send_thread = thread::spawn(move || {
+                send.send(2).unwrap();
+            });
+
+            recv.changed().now_or_never();
+
+            send_thread.join().unwrap();
+        });
+    }
+}

tokio/src/task/spawn.rs

@@ -129,7 +129,7 @@ cfg_rt! {
         T::Output: Send + 'static,
     {
         let spawn_handle = runtime::context::spawn_handle()
-        .expect("must be called from the context of Tokio runtime configured with either `basic_scheduler` or `threaded_scheduler`");
+        .expect("must be called from the context of a Tokio 0.3.x runtime configured with either `basic_scheduler` or `threaded_scheduler`");
         let task = crate::util::trace::task(task, "task");
         spawn_handle.spawn(task)
     }

tokio/src/time/driver/handle.rs

@@ -47,7 +47,7 @@ cfg_rt! {
         /// panicking.
         pub(crate) fn current() -> Self {
             crate::runtime::context::time_handle()
-                .expect("there is no timer running, must be called from the context of Tokio runtime")
+                .expect("there is no timer running, must be called from the context of a Tokio 0.3.x runtime")
         }
     }
 }
@@ -71,7 +71,7 @@ cfg_not_rt! {
         /// lazy, and so outside executed inside the runtime successfuly without
         /// panicking.
         pub(crate) fn current() -> Self {
-            panic!("there is no timer running, must be called from the context of Tokio runtime or \
+            panic!("there is no timer running, must be called from the context of a Tokio 0.3.x runtime or \
             `rt` is not enabled")
         }
     }

tokio/tests/macros_select.rs

@@ -359,9 +359,6 @@ async fn join_with_select() {
 async fn use_future_in_if_condition() {
     use tokio::time::{self, Duration};
 
-    let sleep = time::sleep(Duration::from_millis(50));
-    tokio::pin!(sleep);
-
     tokio::select! {
         _ = time::sleep(Duration::from_millis(50)), if false => {
             panic!("if condition ignored")

tokio/tests/task_abort.rs

@@ -1,6 +1,9 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 
+use std::thread::sleep;
+use std::time::Duration;
+
 /// Checks that a suspended task can be aborted without panicking as reported in
 /// issue #3157: <https://github.com/tokio-rs/tokio/issues/3157>.
 #[test]
@@ -24,3 +27,49 @@ fn test_abort_without_panic_3157() {
         let _ = handle.await;
     });
 }
+
+/// Checks that a suspended LocalSet task can be aborted from a remote thread
+/// without panicking and without running the tasks destructor on the wrong thread.
+/// <https://github.com/tokio-rs/tokio/issues/3929>
+#[test]
+fn remote_abort_local_set_3929() {
+    struct DropCheck {
+        created_on: std::thread::ThreadId,
+        not_send: std::marker::PhantomData<*const ()>,
+    }
+
+    impl DropCheck {
+        fn new() -> Self {
+            Self {
+                created_on: std::thread::current().id(),
+                not_send: std::marker::PhantomData,
+            }
+        }
+    }
+    impl Drop for DropCheck {
+        fn drop(&mut self) {
+            if std::thread::current().id() != self.created_on {
+                panic!("non-Send value dropped in another thread!");
+            }
+        }
+    }
+
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .build()
+        .unwrap();
+    let local = tokio::task::LocalSet::new();
+
+    let check = DropCheck::new();
+    let jh = local.spawn_local(async move {
+        futures::future::pending::<()>().await;
+        drop(check);
+    });
+
+    let jh2 = std::thread::spawn(move || {
+        sleep(Duration::from_millis(50));
+        jh.abort();
+    });
+
+    rt.block_on(local);
+    jh2.join().unwrap();
+}