From 9a0503e89df34a331cf715bddbb2994b27103005 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Mon, 26 Jun 2023 20:14:20 +0000 Subject: [PATCH] wip --- tokio/src/runtime/builder.rs | 68 +++++++++++++++++++ tokio/src/runtime/runtime.rs | 16 +++++ tokio/src/runtime/scheduler/block_in_place.rs | 21 ++++++ tokio/src/runtime/scheduler/mod.rs | 3 + tokio/src/task/blocking.rs | 2 +- 5 files changed, 109 insertions(+), 1 deletion(-) create mode 100644 tokio/src/runtime/scheduler/block_in_place.rs diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs index a700342b8..ef636e7d6 100644 --- a/tokio/src/runtime/builder.rs +++ b/tokio/src/runtime/builder.rs @@ -199,6 +199,8 @@ pub(crate) enum Kind { CurrentThread, #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))] MultiThread, + #[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(tokio_wasi)))] + MultiThreadAlt, } impl Builder { @@ -230,6 +232,26 @@ impl Builder { // The number `61` is fairly arbitrary. I believe this value was copied from golang. Builder::new(Kind::MultiThread, 61) } + + cfg_unstable! { + /// Returns a new builder with the alternate multi thread scheduler + /// selected. + /// + /// The alternate multi threaded scheduler is an in-progress + /// candidate to replace the existing multi threaded scheduler. It + /// currently does not scale as well to 16+ processors. + /// + /// This runtime flavor is currently **not considered production + /// ready**. + /// + /// Configuration methods can be chained on the return value. + #[cfg(feature = "rt-multi-thread")] + #[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))] + pub fn new_multi_thread_alt() -> Builder { + // The number `61` is fairly arbitrary. I believe this value was copied from golang. + Builder::new(Kind::MultiThreadAlt, 61) + } + } } /// Returns a new runtime builder initialized with default configuration @@ -656,6 +678,8 @@ impl Builder { Kind::CurrentThread => self.build_current_thread_runtime(), #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))] Kind::MultiThread => self.build_threaded_runtime(), + #[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(tokio_wasi)))] + Kind::MultiThreadAlt => self.build_alt_threaded_runtime(), } } @@ -665,6 +689,8 @@ impl Builder { Kind::CurrentThread => true, #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))] Kind::MultiThread => false, + #[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(tokio_wasi)))] + Kind::MultiThreadAlt => false, }, enable_io: self.enable_io, enable_time: self.enable_time, @@ -1208,6 +1234,48 @@ cfg_rt_multi_thread! { Ok(Runtime::from_parts(Scheduler::MultiThread(scheduler), handle, blocking_pool)) } + + cfg_unstable! { + fn build_alt_threaded_runtime(&mut self) -> io::Result { + use crate::loom::sys::num_cpus; + use crate::runtime::{Config, runtime::Scheduler}; + use crate::runtime::scheduler::MultiThreadAlt; + + let core_threads = self.worker_threads.unwrap_or_else(num_cpus); + + let (driver, driver_handle) = driver::Driver::new(self.get_cfg())?; + + // Create the blocking pool + let blocking_pool = + blocking::create_blocking_pool(self, self.max_blocking_threads + core_threads); + let blocking_spawner = blocking_pool.spawner().clone(); + + // Generate a rng seed for this runtime. + let seed_generator_1 = self.seed_generator.next_generator(); + let seed_generator_2 = self.seed_generator.next_generator(); + + let (scheduler, handle) = MultiThreadAlt::new( + core_threads, + driver, + driver_handle, + blocking_spawner, + seed_generator_2, + Config { + before_park: self.before_park.clone(), + after_unpark: self.after_unpark.clone(), + global_queue_interval: self.global_queue_interval, + event_interval: self.event_interval, + #[cfg(tokio_unstable)] + unhandled_panic: self.unhandled_panic.clone(), + disable_lifo_slot: self.disable_lifo_slot, + seed_generator: seed_generator_1, + metrics_poll_count_histogram: self.metrics_poll_count_histogram_builder(), + }, + ); + + Ok(Runtime::from_parts(Scheduler::MultiThreadAlt(scheduler), handle, blocking_pool)) + } + } } } diff --git a/tokio/src/runtime/runtime.rs b/tokio/src/runtime/runtime.rs index 561285d53..681ba28fc 100644 --- a/tokio/src/runtime/runtime.rs +++ b/tokio/src/runtime/runtime.rs @@ -9,6 +9,10 @@ use std::time::Duration; cfg_rt_multi_thread! { use crate::runtime::Builder; use crate::runtime::scheduler::MultiThread; + + cfg_unstable! { + use crate::runtime::scheduler::MultiThreadAlt; + } } /// The Tokio runtime. @@ -98,6 +102,10 @@ pub(super) enum Scheduler { /// Execute tasks across multiple threads. #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))] MultiThread(MultiThread), + + /// Execute tasks across multiple threads. + #[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(tokio_wasi)))] + MultiThreadAlt(MultiThreadAlt), } impl Runtime { @@ -314,6 +322,8 @@ impl Runtime { Scheduler::CurrentThread(exec) => exec.block_on(&self.handle.inner, future), #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))] Scheduler::MultiThread(exec) => exec.block_on(&self.handle.inner, future), + #[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(tokio_wasi)))] + Scheduler::MultiThreadAlt(exec) => exec.block_on(&self.handle.inner, future), } } @@ -434,6 +444,12 @@ impl Drop for Runtime { // already in the runtime's context. multi_thread.shutdown(&self.handle.inner); } + #[cfg(all(tokio_unstable, feature = "rt-multi-thread", not(tokio_wasi)))] + Scheduler::MultiThreadAlt(multi_thread) => { + // The threaded scheduler drops its tasks on its worker threads, which is + // already in the runtime's context. + multi_thread.shutdown(&self.handle.inner); + } } } } diff --git a/tokio/src/runtime/scheduler/block_in_place.rs b/tokio/src/runtime/scheduler/block_in_place.rs new file mode 100644 index 000000000..803ff4504 --- /dev/null +++ b/tokio/src/runtime/scheduler/block_in_place.rs @@ -0,0 +1,21 @@ +use crate::runtime::scheduler; + +#[track_caller] +pub(crate) fn block_in_place(f: F) -> R +where + F: FnOnce() -> R, +{ + #[cfg(tokio_unstable)] + { + use crate::runtime::{Handle, RuntimeFlavor::MultiThreadAlt}; + + match Handle::try_current().map(|h| h.runtime_flavor()) { + Ok(MultiThreadAlt) => { + return scheduler::multi_thread_alt::block_in_place(f); + } + _ => {} + } + } + + scheduler::multi_thread::block_in_place(f) +} diff --git a/tokio/src/runtime/scheduler/mod.rs b/tokio/src/runtime/scheduler/mod.rs index 620e6841d..e0e1ede65 100644 --- a/tokio/src/runtime/scheduler/mod.rs +++ b/tokio/src/runtime/scheduler/mod.rs @@ -10,6 +10,9 @@ cfg_rt! { } cfg_rt_multi_thread! { + mod block_in_place; + pub(crate) use block_in_place::block_in_place; + mod lock; use lock::Lock; diff --git a/tokio/src/task/blocking.rs b/tokio/src/task/blocking.rs index 9bd15ebd5..1cce46639 100644 --- a/tokio/src/task/blocking.rs +++ b/tokio/src/task/blocking.rs @@ -75,7 +75,7 @@ cfg_rt_multi_thread! { where F: FnOnce() -> R, { - crate::runtime::scheduler::multi_thread::block_in_place(f) + crate::runtime::scheduler::block_in_place(f) } }