rt: add support for non-send closures for thread (un)parking #7420

aryan9600 · 2025-06-25T11:14:44Z

Add support for non Send+Sync closures for thread parking and unparking callbacks when using a LocalRuntime. Since a LocalRuntime will always run its tasks on the same thread, its safe to accept a non Send+Sync closure.

Motivation

Since LocalRuntime can run non Send+Sync functions, we should accept thread parking/unparking callbacks that do not implement Send and Sync.

Solution

Add two methods on LocalOptions, on_thread_park and on_thread_unpark that accept a Fn() + 'static. These callbacks are then converted into Fn() + Send + Sync + 'static. This requires unsafe code, but unsafe is acceptable here because of the behaviour of LocalRuntime

fixes #7370

Daksh14 · 2025-06-25T16:41:46Z

tokio/src/runtime/local_runtime/options.rs

+// This is specifically used for executing callbacks when using a `LocalRuntime`, since the
+// callbacks will never be sent across threads and thus do not need to be Send or Sync.
+struct UnsafeSendSync<T>(T);
+unsafe impl<T> Send for UnsafeSendSync<T> {}


shouldn't this be unsafe impl<T: Send> Send for UnsafeSendSync<T> {}
and unsafe impl<T: Send + Sync> Sync for UnsafeSendSync<T> {}

Any unsafe usage should be documented with // SAFETY: at least

the whole point is that T is not Send or Sync.

sorry, my bad, will update the comments to begin with // SAFETY:

Add support for non `Send`+`Sync` closures for thread parking and unparking callbacks when using a `LocalRuntime`. Since a `LocalRuntime` will always run its tasks on the same thread, its safe to accept a non `Send`+`Sync` closure. Signed-off-by: Sanskar Jaiswal <[email protected]>

ADD-SP · 2025-06-29T04:49:09Z

tokio/src/runtime/builder.rs

+// use super::LocalOptions;
+


Suggested change

// use super::LocalOptions;

Outdated lines?

ADD-SP · 2025-06-29T04:52:57Z

tokio/src/runtime/local_runtime/options.rs

+    /// This can be used to start work only when the executor is idle, or for bookkeeping
+    /// and monitoring purposes.
+    ///
+    /// This differs from the `Builder::on_thread_park` method in that it accepts a non Send + Sync


Suggested change

/// This differs from the `Builder::on_thread_park` method in that it accepts a non Send + Sync

/// This differs from the [`Builder::on_thread_park`] method in that it accepts a non Send + Sync

ADD-SP · 2025-06-29T11:27:49Z

tokio/src/runtime/local_runtime/options.rs

 #[non_exhaustive]
+#[allow(missing_debug_implementations)]


I guess this is because we cannot #[derive(Debug)] for Fn. We could try to impl Debug manually, so that we don't need this clippy exception.

ADD-SP · 2025-06-29T11:31:53Z

tokio/src/runtime/local_runtime/options.rs

+    /// Callback for a worker parking itself
+    pub(crate) before_park: Option<Callback>,


Suggested change

/// Callback for a worker parking itself

pub(crate) before_park: Option<Callback>,

/// To run before the local runtime is parked.

pub(crate) before_park: Option<Callback>,

This is to align with

tokio/tokio/src/runtime/builder.rs

Lines 82 to 83 in b7a75b5

/// To run before each worker thread is parked.

pub(super) before_park: Option<Callback>,

ADD-SP · 2025-06-29T11:33:35Z

tokio/src/runtime/local_runtime/options.rs

+    /// Callback for a worker unparking itself
+    pub(crate) after_unpark: Option<Callback>,


Suggested change

/// Callback for a worker unparking itself

pub(crate) after_unpark: Option<Callback>,

/// To run before the local runtime is spawned.

pub(crate) after_unpark: Option<Callback>,

ADD-SP · 2025-06-29T11:39:24Z

tokio/src/runtime/local_runtime/options.rs

+    /// ```
+    /// # use tokio::runtime::{Builder, LocalOptions};
+    /// # pub fn main() {
+    /// let (tx, rx) = std::sync::mpsc::channel();
+    /// let mut opts = LocalOptions::default();
+    /// opts.on_thread_park(move || match rx.recv() {
+    ///    Ok(x) => println!("Received from channel: {}", x),
+    ///    Err(e) => println!("Error receiving from channel: {}", e),
+    /// });
+    ///
+    /// let runtime = Builder::new_current_thread()
+    ///     .enable_time()
+    ///     .build_local(opts)
+    ///     .unwrap();
+    ///
+    /// runtime.block_on(async {
+    ///     tokio::task::spawn_local(async move {
+    ///         tx.send(42).unwrap();
+    ///     });
+    ///     tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+    /// })
+    /// }
+    /// ```


This might not be rendered correctly.

ADD-SP · 2025-06-29T11:39:31Z

tokio/src/runtime/local_runtime/options.rs

+    /// ```
+    /// # use tokio::runtime::{Builder, LocalOptions};
+    /// # pub fn main() {
+    /// let (tx, rx) = std::sync::mpsc::channel();
+    /// let mut opts = LocalOptions::default();
+    /// opts.on_thread_unpark(move || match rx.recv() {
+    ///    Ok(x) => println!("Received from channel: {}", x),
+    ///    Err(e) => println!("Error receiving from channel: {}", e),
+    /// });
+    ///
+    /// let runtime = Builder::new_current_thread()
+    ///     .enable_time()
+    ///     .build_local(opts)
+    ///     .unwrap();
+    ///
+    /// runtime.block_on(async {
+    ///     tokio::task::spawn_local(async move {
+    ///         tx.send(42).unwrap();
+    ///     });
+    ///     tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+    /// })
+    /// }
+    /// ```


This might not be rendered correctly.

ADD-SP · 2025-06-29T11:48:31Z

tokio/src/runtime/local_runtime/options.rs

+        self
+    }
+
+    /// Executes function `f` just after a thread unparks (starts executing tasks).


Suggested change

/// Executes function `f` just after a thread unparks (starts executing tasks).

/// Executes function `f` just after the local runtime unparks (starts executing tasks).

ADD-SP · 2025-06-29T13:06:14Z

tokio/tests/rt_local.rs

We need a test to make sure callbacks will not be executed when using the Handle::block_on.

ADD-SP · 2025-06-29T13:13:05Z

tokio/tests/rt_local.rs

+#[test]
+fn test_on_thread_park_in_runtime() {
+    let mut opts = LocalOptions::default();
+    let called = std::rc::Rc::new(std::cell::RefCell::new(false));


Suggested change

let called = std::rc::Rc::new(std::cell::RefCell::new(false));

// this makes the `on_thread_park` callback `!Send + !Sync`

let called = std::rc::Rc::new(std::cell::RefCell::new(false));

ADD-SP · 2025-06-29T13:24:11Z

tokio/tests/rt_local.rs

We also want to cover the on_thread_unpark callback.

aryan9600 force-pushed the non-send-thread-park branch from be90205 to eeb45d3 Compare June 25, 2025 11:16

Daksh14 reviewed Jun 25, 2025

View reviewed changes

aryan9600 force-pushed the non-send-thread-park branch from eeb45d3 to c124e08 Compare June 25, 2025 17:21

aryan9600 force-pushed the non-send-thread-park branch from c124e08 to 4f8b562 Compare June 26, 2025 05:34

mox692 added A-tokio Area: The main tokio crate M-runtime Module: tokio/runtime labels Jun 27, 2025

ADD-SP added the C-enhancement Category: A PR with an enhancement or bugfix. label Jun 29, 2025

ADD-SP reviewed Jun 29, 2025

View reviewed changes

tokio/tests/rt_local.rs

Copy link

Member

ADD-SP Jun 29, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We also want to cover the on_thread_unpark callback.

ADD-SP added the R-loom-current-thread Run loom current-thread tests on this PR label Jun 29, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

rt: add support for non-send closures for thread (un)parking #7420

rt: add support for non-send closures for thread (un)parking #7420

Uh oh!

aryan9600 commented Jun 25, 2025

Uh oh!

Daksh14 Jun 25, 2025 •

edited

Loading

Uh oh!

aryan9600 Jun 25, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

ADD-SP Jun 29, 2025

Uh oh!

Uh oh!

	/// This differs from the `Builder::on_thread_park` method in that it accepts a non Send + Sync
	/// This differs from the [`Builder::on_thread_park`] method in that it accepts a non Send + Sync

		/// Callback for a worker parking itself
		pub(crate) before_park: Option<Callback>,

	/// To run before each worker thread is parked.
	pub(super) before_park: Option<Callback>,

		/// Callback for a worker unparking itself
		pub(crate) after_unpark: Option<Callback>,

	/// Executes function `f` just after a thread unparks (starts executing tasks).
	/// Executes function `f` just after the local runtime unparks (starts executing tasks).

	let called = std::rc::Rc::new(std::cell::RefCell::new(false));
	// this makes the `on_thread_park` callback `!Send + !Sync`
	let called = std::rc::Rc::new(std::cell::RefCell::new(false));

		#[non_exhaustive]
		#[allow(missing_debug_implementations)]

Uh oh!

rt: add support for non-send closures for thread (un)parking #7420

Are you sure you want to change the base?

rt: add support for non-send closures for thread (un)parking #7420

Uh oh!

Conversation

aryan9600 commented Jun 25, 2025

Motivation

Solution

Uh oh!

Daksh14 Jun 25, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Daksh14 Jun 25, 2025 •

edited

Loading