From 03f7a78880119d4e629646a8fba34c242ef3b04a Mon Sep 17 00:00:00 2001 From: Alice Ryhl Date: Tue, 13 Jul 2021 16:47:04 +0200 Subject: [PATCH] runtime: fix remote abort (#3942) --- tokio-util/src/time/wheel/level.rs | 2 +- tokio/src/lib.rs | 1 + tokio/src/process/mod.rs | 4 +-- tokio/src/runtime/task/harness.rs | 11 +++++++ tokio/src/runtime/task/join.rs | 2 +- tokio/src/runtime/task/raw.rs | 14 +++++++++ tokio/src/runtime/task/state.rs | 9 ++++++ tokio/tests/macros_select.rs | 3 -- tokio/tests/task_abort.rs | 49 ++++++++++++++++++++++++++++++ 9 files changed, 88 insertions(+), 7 deletions(-) diff --git a/tokio-util/src/time/wheel/level.rs b/tokio-util/src/time/wheel/level.rs index 49f9bfb9c..d822155c0 100644 --- a/tokio-util/src/time/wheel/level.rs +++ b/tokio-util/src/time/wheel/level.rs @@ -46,7 +46,7 @@ impl Level { () => { T::default() }; - }; + } Level { level, diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index fa439b40c..51bfb3b1f 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -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. //! diff --git a/tokio/src/process/mod.rs b/tokio/src/process/mod.rs index bd23e1f73..b6deefa08 100644 --- a/tokio/src/process/mod.rs +++ b/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") diff --git a/tokio/src/runtime/task/harness.rs b/tokio/src/runtime/task/harness.rs index 208d48c4d..f6eaa0c93 100644 --- a/tokio/src/runtime/task/harness.rs +++ b/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) { diff --git a/tokio/src/runtime/task/join.rs b/tokio/src/runtime/task/join.rs index dedfb3879..2fe40a721 100644 --- a/tokio/src/runtime/task/join.rs +++ b/tokio/src/runtime/task/join.rs @@ -192,7 +192,7 @@ impl JoinHandle { /// ``` pub fn abort(&self) { if let Some(raw) = self.raw { - raw.shutdown(); + raw.remote_abort(); } } } diff --git a/tokio/src/runtime/task/raw.rs b/tokio/src/runtime/task/raw.rs index cae56d037..39336cee9 100644 --- a/tokio/src/runtime/task/raw.rs +++ b/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
), + /// The task is remotely aborted + pub(super) remote_abort: unsafe fn(NonNull
), + /// Scheduler is being shutdown pub(super) shutdown: unsafe fn(NonNull
), } @@ -33,6 +36,7 @@ pub(super) fn vtable() -> &'static Vtable { dealloc: dealloc::, try_read_output: try_read_output::, drop_join_handle_slow: drop_join_handle_slow::, + remote_abort: remote_abort::, shutdown: shutdown::, } } @@ -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(ptr: NonNull
) { harness.drop_join_handle_slow() } +unsafe fn remote_abort(ptr: NonNull
) { + let harness = Harness::::from_raw(ptr); + harness.remote_abort() +} + unsafe fn shutdown(ptr: NonNull
) { let harness = Harness::::from_raw(ptr); harness.shutdown() diff --git a/tokio/src/runtime/task/state.rs b/tokio/src/runtime/task/state.rs index 21e90430d..da0c567d1 100644 --- a/tokio/src/runtime/task/state.rs +++ b/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. diff --git a/tokio/tests/macros_select.rs b/tokio/tests/macros_select.rs index 3359849db..d0216f523 100644 --- a/tokio/tests/macros_select.rs +++ b/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") diff --git a/tokio/tests/task_abort.rs b/tokio/tests/task_abort.rs index e84f19c3d..b3838e2b0 100644 --- a/tokio/tests/task_abort.rs +++ b/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: . #[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. +/// +#[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(); +}