tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-22 00:00:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

rt, chore: rename internal scheduler types (#4945)

Browse Source 
For historical reasons, the current-thread runtime's scheduler was named
BasicScheduler and the multi-thread runtime's scheduler were named
ThreadPool. This patch renames the schedulers to mirror the runtime
names to increase consistency. This patch also moves the scheduler
implementations into a new `runtime::scheduler` module.
This commit is contained in:
Carl Lerche
2022-08-26 09:47:19 -07:00
committed by GitHub
parent 218f2629ff
commit a3411a412c
25 changed files with 125 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/lib.rs
+2 -2
View File
@@ -153,7 +153,7 @@
//! provide the functionality you need.
//!
//! Using the runtime requires the "rt" or "rt-multi-thread" feature flags, to
//! enable the basic [single-threaded scheduler][rt] and the [thread-pool
//! enable the current-thread [single-threaded scheduler][rt] and the [multi-thread
//! scheduler][rt-multi-thread], respectively. See the [`runtime` module
//! documentation][rt-features] for details. In addition, the "macros" feature
//! flag enables the `#[tokio::main]` and `#[tokio::test]` attributes.
@@ -310,7 +310,7 @@
//! need.
//!
//! - `full`: Enables all features listed below except `test-util` and `tracing`.
//! - `rt`: Enables `tokio::spawn`, the basic (current thread) scheduler,
//! - `rt`: Enables `tokio::spawn`, the current-thread scheduler,
//! and non-scheduler utilities.
//! - `rt-multi-thread`: Enables the heavier, multi-threaded, work-stealing scheduler.
//! - `io-util`: Enables the IO based `Ext` traits.
tokio/src/runtime/builder.rs
+9 -9
View File
@@ -625,7 +625,7 @@ impl Builder {
/// ```
pub fn build(&mut self) -> io::Result<Runtime> {
match &self.kind {
Kind::CurrentThread => self.build_basic_runtime(),
Kind::CurrentThread => self.build_current_thread_runtime(),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Kind::MultiThread => self.build_threaded_runtime(),
}
@@ -831,8 +831,8 @@ impl Builder {
}
}
fn build_basic_runtime(&mut self) -> io::Result<Runtime> {
use crate::runtime::{BasicScheduler, Config, HandleInner, Kind};
fn build_current_thread_runtime(&mut self) -> io::Result<Runtime> {
use crate::runtime::{Config, CurrentThread, HandleInner, Kind};
let (driver, resources) = driver::Driver::new(self.get_cfg())?;
@@ -852,7 +852,7 @@ impl Builder {
// there are no futures ready to do something, it'll let the timer or
// the reactor to generate some new stimuli for the futures to continue
// in their life.
let scheduler = BasicScheduler::new(
let scheduler = CurrentThread::new(
driver,
handle_inner,
Config {
@@ -865,7 +865,7 @@ impl Builder {
disable_lifo_slot: self.disable_lifo_slot,
},
);
let spawner = Spawner::Basic(scheduler.spawner().clone());
let spawner = Spawner::CurrentThread(scheduler.spawner().clone());
Ok(Runtime {
kind: Kind::CurrentThread(scheduler),
@@ -951,7 +951,7 @@ cfg_rt_multi_thread! {
impl Builder {
fn build_threaded_runtime(&mut self) -> io::Result<Runtime> {
use crate::loom::sys::num_cpus;
use crate::runtime::{Config, HandleInner, Kind, ThreadPool};
use crate::runtime::{Config, HandleInner, Kind, MultiThread};
let core_threads = self.worker_threads.unwrap_or_else(num_cpus);
@@ -970,7 +970,7 @@ cfg_rt_multi_thread! {
blocking_spawner,
};
let (scheduler, launch) = ThreadPool::new(
let (scheduler, launch) = MultiThread::new(
core_threads,
driver,
handle_inner,
@@ -984,7 +984,7 @@ cfg_rt_multi_thread! {
disable_lifo_slot: self.disable_lifo_slot,
},
);
let spawner = Spawner::ThreadPool(scheduler.spawner().clone());
let spawner = Spawner::MultiThread(scheduler.spawner().clone());
// Create the runtime handle
let handle = Handle { spawner };
@@ -994,7 +994,7 @@ cfg_rt_multi_thread! {
launch.launch();
Ok(Runtime {
kind: Kind::ThreadPool(scheduler),
kind: Kind::MultiThread(scheduler),
handle,
blocking_pool,
})
tokio/src/runtime/mod.rs
+13 -16
View File
@@ -137,7 +137,7 @@
//! use tokio::runtime;
//!
//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let basic_rt = runtime::Builder::new_current_thread()
//! let rt = runtime::Builder::new_current_thread()
//! .build()?;
//! # Ok(()) }
//! ```
@@ -166,7 +166,6 @@
//! [`tokio::main`]: ../attr.main.html
//! [runtime builder]: crate::runtime::Builder
//! [`Runtime::new`]: crate::runtime::Runtime::new
//! [`Builder::basic_scheduler`]: crate::runtime::Builder::basic_scheduler
//! [`Builder::threaded_scheduler`]: crate::runtime::Builder::threaded_scheduler
//! [`Builder::enable_io`]: crate::runtime::Builder::enable_io
//! [`Builder::enable_time`]: crate::runtime::Builder::enable_time
@@ -187,8 +186,8 @@ cfg_rt! {
pub(crate) mod task;
mod basic_scheduler;
use basic_scheduler::BasicScheduler;
pub(crate) mod scheduler;
use scheduler::CurrentThread;
mod config;
use config::Config;
@@ -244,8 +243,7 @@ cfg_rt! {
cfg_rt_multi_thread! {
use driver::Driver;
pub(crate) mod thread_pool;
use self::thread_pool::ThreadPool;
use scheduler::MultiThread;
}
cfg_rt! {
@@ -304,15 +302,15 @@ cfg_rt! {
blocking_pool: BlockingPool,
}
/// The runtime executor is either a thread-pool or a current-thread executor.
/// The runtime executor is either a multi-thread or a current-thread executor.
#[derive(Debug)]
enum Kind {
/// Execute all tasks on the current-thread.
CurrentThread(BasicScheduler),
CurrentThread(CurrentThread),
/// Execute tasks across multiple threads.
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
ThreadPool(ThreadPool),
MultiThread(MultiThread),
}
/// After thread starts / before thread stops
@@ -347,7 +345,6 @@ cfg_rt! {
/// [mod]: index.html
/// [main]: ../attr.main.html
/// [threaded scheduler]: index.html#threaded-scheduler
/// [basic scheduler]: index.html#basic-scheduler
/// [runtime builder]: crate::runtime::Builder
#[cfg(feature = "rt-multi-thread")]
#[cfg_attr(docsrs, doc(cfg(feature = "rt-multi-thread")))]
@@ -492,7 +489,7 @@ cfg_rt! {
match &self.kind {
Kind::CurrentThread(exec) => exec.block_on(future),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Kind::ThreadPool(exec) => exec.block_on(future),
Kind::MultiThread(exec) => exec.block_on(future),
}
}
@@ -608,11 +605,11 @@ cfg_rt! {
impl Drop for Runtime {
fn drop(&mut self) {
match &mut self.kind {
Kind::CurrentThread(basic) => {
// This ensures that tasks spawned on the basic runtime are dropped inside the
// runtime's context.
Kind::CurrentThread(current_thread) => {
// This ensures that tasks spawned on the current-thread
// runtime are dropped inside the runtime's context.
match self::context::try_enter(self.handle.clone()) {
Some(guard) => basic.set_context_guard(guard),
Some(guard) => current_thread.set_context_guard(guard),
None => {
// The context thread-local has already been destroyed.
//
@@ -622,7 +619,7 @@ cfg_rt! {
}
},
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Kind::ThreadPool(_) => {
Kind::MultiThread(_) => {
// The threaded scheduler drops its tasks on its worker threads, which is
// already in the runtime's context.
},
tokio/src/runtime/basic_scheduler.rs → tokio/src/runtime/scheduler/current_thread.rs
+20 -20
View File
@@ -20,7 +20,7 @@ use std::task::Poll::{Pending, Ready};
use std::time::Duration;
/// Executes tasks on the current thread
pub(crate) struct BasicScheduler {
pub(crate) struct CurrentThread {
/// Core scheduler data is acquired by a thread entering `block_on`.
core: AtomicCell<Core>,
@@ -31,10 +31,10 @@ pub(crate) struct BasicScheduler {
/// Sendable task spawner
spawner: Spawner,
/// This is usually None, but right before dropping the BasicScheduler, it
/// is changed to `Some` with the context being the runtime's own context.
/// This ensures that any tasks dropped in the `BasicScheduler`s destructor
/// run in that runtime's context.
/// This is usually None, but right before dropping the CurrentThread
/// scheduler, it is changed to `Some` with the context being the runtime's
/// own context. This ensures that any tasks dropped in the `CurrentThread`'s
/// destructor run in that runtime's context.
context_guard: Option<EnterGuard>,
}
@@ -108,11 +108,11 @@ struct Context {
/// Initial queue capacity.
const INITIAL_CAPACITY: usize = 64;
// Tracks the current BasicScheduler.
// Tracks the current CurrentThread.
scoped_thread_local!(static CURRENT: Context);
impl BasicScheduler {
pub(crate) fn new(driver: Driver, handle_inner: HandleInner, config: Config) -> BasicScheduler {
impl CurrentThread {
pub(crate) fn new(driver: Driver, handle_inner: HandleInner, config: Config) -> CurrentThread {
let unpark = driver.unpark();
let spawner = Spawner {
@@ -137,7 +137,7 @@ impl BasicScheduler {
unhandled_panic: false,
})));
BasicScheduler {
CurrentThread {
core,
notify: Notify::new(),
spawner,
@@ -193,16 +193,16 @@ impl BasicScheduler {
spawner: self.spawner.clone(),
core: RefCell::new(Some(core)),
},
basic_scheduler: self,
scheduler: self,
})
}
pub(super) fn set_context_guard(&mut self, guard: EnterGuard) {
pub(crate) fn set_context_guard(&mut self, guard: EnterGuard) {
self.context_guard = Some(guard);
}
}
impl Drop for BasicScheduler {
impl Drop for CurrentThread {
fn drop(&mut self) {
// Avoid a double panic if we are currently panicking and
// the lock may be poisoned.
@@ -246,9 +246,9 @@ impl Drop for BasicScheduler {
}
}
impl fmt::Debug for BasicScheduler {
impl fmt::Debug for CurrentThread {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("BasicScheduler").finish()
fmt.debug_struct("CurrentThread").finish()
}
}
@@ -357,8 +357,8 @@ impl Context {
// ===== impl Spawner =====
impl Spawner {
/// Spawns a future onto the basic scheduler
pub(crate) fn spawn<F>(&self, future: F, id: super::task::Id) -> JoinHandle<F::Output>
/// Spawns a future onto the `CurrentThread` scheduler
pub(crate) fn spawn<F>(&self, future: F, id: crate::runtime::task::Id) -> JoinHandle<F::Output>
where
F: crate::future::Future + Send + 'static,
F::Output: Send + 'static,
@@ -503,10 +503,10 @@ impl Wake for Shared {
// ===== CoreGuard =====
/// Used to ensure we always place the `Core` value back into its slot in
/// `BasicScheduler`, even if the future panics.
/// `CurrentThread`, even if the future panics.
struct CoreGuard<'a> {
context: Context,
basic_scheduler: &'a BasicScheduler,
scheduler: &'a CurrentThread,
}
impl CoreGuard<'_> {
@@ -605,10 +605,10 @@ impl Drop for CoreGuard<'_> {
if let Some(core) = self.context.core.borrow_mut().take() {
// Replace old scheduler back into the state to allow
// other threads to pick it up and drive it.
self.basic_scheduler.core.set(core);
self.scheduler.core.set(core);
// Wake up other possible threads that could steal the driver.
self.basic_scheduler.notify.notify_one()
self.scheduler.notify.notify_one()
}
}
}
tokio/src/runtime/scheduler/mod.rs
+7
View File
@@ -0,0 +1,7 @@
pub(crate) mod current_thread;
pub(crate) use current_thread::CurrentThread;
cfg_rt_multi_thread! {
pub(crate) mod multi_thread;
pub(crate) use multi_thread::MultiThread;
}
tokio/src/runtime/thread_pool/idle.rs → tokio/src/runtime/scheduler/multi_thread/idle.rs
View File
tokio/src/runtime/thread_pool/mod.rs → tokio/src/runtime/scheduler/multi_thread/mod.rs
+13 -13
View File
@@ -1,4 +1,4 @@
//! Threadpool
//! Multi-threaded runtime
mod idle;
use self::idle::Idle;
@@ -6,7 +6,7 @@ use self::idle::Idle;
mod park;
pub(crate) use park::{Parker, Unparker};
pub(super) mod queue;
pub(crate) mod queue;
mod worker;
pub(crate) use worker::Launch;
@@ -21,7 +21,7 @@ use std::fmt;
use std::future::Future;
/// Work-stealing based thread pool for executing futures.
pub(crate) struct ThreadPool {
pub(crate) struct MultiThread {
spawner: Spawner,
}
@@ -34,29 +34,29 @@ pub(crate) struct ThreadPool {
/// impact the lifecycle of the thread pool in any way. The thread pool may
/// shut down while there are outstanding `Spawner` instances.
///
/// `Spawner` instances are obtained by calling [`ThreadPool::spawner`].
/// `Spawner` instances are obtained by calling [`MultiThread::spawner`].
///
/// [`ThreadPool::spawner`]: method@ThreadPool::spawner
/// [`MultiThread::spawner`]: method@MultiThread::spawner
#[derive(Clone)]
pub(crate) struct Spawner {
shared: Arc<worker::Shared>,
}
// ===== impl ThreadPool =====
// ===== impl MultiThread =====
impl ThreadPool {
impl MultiThread {
pub(crate) fn new(
size: usize,
driver: Driver,
handle_inner: HandleInner,
config: Config,
) -> (ThreadPool, Launch) {
) -> (MultiThread, Launch) {
let parker = Parker::new(driver);
let (shared, launch) = worker::create(size, parker, handle_inner, config);
let spawner = Spawner { shared };
let thread_pool = ThreadPool { spawner };
let multi_thread = MultiThread { spawner };
(thread_pool, launch)
(multi_thread, launch)
}
/// Returns reference to `Spawner`.
@@ -80,13 +80,13 @@ impl ThreadPool {
}
}
impl fmt::Debug for ThreadPool {
impl fmt::Debug for MultiThread {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
fmt.debug_struct("ThreadPool").finish()
fmt.debug_struct("MultiThread").finish()
}
}
impl Drop for ThreadPool {
impl Drop for MultiThread {
fn drop(&mut self) {
self.spawner.shutdown();
}
tokio/src/runtime/thread_pool/park.rs → tokio/src/runtime/scheduler/multi_thread/park.rs
View File
tokio/src/runtime/thread_pool/queue.rs → tokio/src/runtime/scheduler/multi_thread/queue.rs
View File
tokio/src/runtime/thread_pool/worker.rs → tokio/src/runtime/scheduler/multi_thread/worker.rs
+4 -4
View File
@@ -63,8 +63,8 @@ use crate::loom::sync::{Arc, Mutex};
use crate::park::{Park, Unpark};
use crate::runtime;
use crate::runtime::enter::EnterContext;
use crate::runtime::scheduler::multi_thread::{queue, Idle, Parker, Unparker};
use crate::runtime::task::{Inject, JoinHandle, OwnedTasks};
use crate::runtime::thread_pool::{queue, Idle, Parker, Unparker};
use crate::runtime::{task, Config, HandleInner, MetricsBatch, SchedulerMetrics, WorkerMetrics};
use crate::util::atomic_cell::AtomicCell;
use crate::util::FastRand;
@@ -288,7 +288,7 @@ where
had_entered = true;
return;
} else {
// This probably means we are on the basic_scheduler or in a
// This probably means we are on the current_thread runtime or in a
// LocalSet, where it is _not_ okay to block.
panic!("can call blocking only when running on the multi-threaded runtime");
}
@@ -857,11 +857,11 @@ impl Shared {
impl crate::runtime::ToHandle for Arc<Shared> {
fn to_handle(&self) -> crate::runtime::Handle {
use crate::runtime::thread_pool::Spawner;
use crate::runtime::scheduler::multi_thread::Spawner;
use crate::runtime::{self, Handle};
Handle {
spawner: runtime::Spawner::ThreadPool(Spawner {
spawner: runtime::Spawner::MultiThread(Spawner {
shared: self.clone(),
}),
}
tokio/src/runtime/spawner.rs
+20 -19
View File
@@ -1,24 +1,25 @@
use crate::future::Future;
use crate::runtime::scheduler::current_thread;
use crate::runtime::task::Id;
use crate::runtime::{basic_scheduler, HandleInner};
use crate::runtime::HandleInner;
use crate::task::JoinHandle;
cfg_rt_multi_thread! {
use crate::runtime::thread_pool;
use crate::runtime::scheduler::multi_thread;
}
#[derive(Debug, Clone)]
pub(crate) enum Spawner {
Basic(basic_scheduler::Spawner),
CurrentThread(current_thread::Spawner),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
ThreadPool(thread_pool::Spawner),
MultiThread(multi_thread::Spawner),
}
impl Spawner {
pub(crate) fn shutdown(&mut self) {
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
{
if let Spawner::ThreadPool(spawner) = self {
if let Spawner::MultiThread(spawner) = self {
spawner.shutdown();
}
}
@@ -30,17 +31,17 @@ impl Spawner {
F::Output: Send + 'static,
{
match self {
Spawner::Basic(spawner) => spawner.spawn(future, id),
Spawner::CurrentThread(spawner) => spawner.spawn(future, id),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Spawner::ThreadPool(spawner) => spawner.spawn(future, id),
Spawner::MultiThread(spawner) => spawner.spawn(future, id),
}
}
pub(crate) fn as_handle_inner(&self) -> &HandleInner {
match self {
Spawner::Basic(spawner) => spawner.as_handle_inner(),
Spawner::CurrentThread(spawner) => spawner.as_handle_inner(),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Spawner::ThreadPool(spawner) => spawner.as_handle_inner(),
Spawner::MultiThread(spawner) => spawner.as_handle_inner(),
}
}
}
@@ -51,41 +52,41 @@ cfg_metrics! {
impl Spawner {
pub(crate) fn num_workers(&self) -> usize {
match self {
Spawner::Basic(_) => 1,
Spawner::CurrentThread(_) => 1,
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Spawner::ThreadPool(spawner) => spawner.num_workers(),
Spawner::MultiThread(spawner) => spawner.num_workers(),
}
}
pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
match self {
Spawner::Basic(spawner) => spawner.scheduler_metrics(),
Spawner::CurrentThread(spawner) => spawner.scheduler_metrics(),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Spawner::ThreadPool(spawner) => spawner.scheduler_metrics(),
Spawner::MultiThread(spawner) => spawner.scheduler_metrics(),
}
}
pub(crate) fn worker_metrics(&self, worker: usize) -> &WorkerMetrics {
match self {
Spawner::Basic(spawner) => spawner.worker_metrics(worker),
Spawner::CurrentThread(spawner) => spawner.worker_metrics(worker),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Spawner::ThreadPool(spawner) => spawner.worker_metrics(worker),
Spawner::MultiThread(spawner) => spawner.worker_metrics(worker),
}
}
pub(crate) fn injection_queue_depth(&self) -> usize {
match self {
Spawner::Basic(spawner) => spawner.injection_queue_depth(),
Spawner::CurrentThread(spawner) => spawner.injection_queue_depth(),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Spawner::ThreadPool(spawner) => spawner.injection_queue_depth(),
Spawner::MultiThread(spawner) => spawner.injection_queue_depth(),
}
}
pub(crate) fn worker_local_queue_depth(&self, worker: usize) -> usize {
match self {
Spawner::Basic(spawner) => spawner.worker_metrics(worker).queue_depth(),
Spawner::CurrentThread(spawner) => spawner.worker_metrics(worker).queue_depth(),
#[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
Spawner::ThreadPool(spawner) => spawner.worker_local_queue_depth(worker),
Spawner::MultiThread(spawner) => spawner.worker_local_queue_depth(worker),
}
}
}
tokio/src/runtime/tests/loom_basic_scheduler.rs → tokio/src/runtime/tests/loom_current_thread_scheduler.rs
View File
tokio/src/runtime/tests/loom_queue.rs
+1 -1
View File
@@ -1,6 +1,6 @@
use crate::runtime::blocking::NoopSchedule;
use crate::runtime::scheduler::multi_thread::queue;
use crate::runtime::task::Inject;
use crate::runtime::thread_pool::queue;
use crate::runtime::MetricsBatch;
use loom::thread;
tokio/src/runtime/tests/mod.rs
+1 -1
View File
@@ -33,8 +33,8 @@ mod unowned_wrapper {
}
cfg_loom! {
mod loom_basic_scheduler;
mod loom_blocking;
mod loom_current_thread_scheduler;
mod loom_local;
mod loom_oneshot;
mod loom_pool;
tokio/src/runtime/tests/queue.rs
+1 -1
View File
@@ -1,5 +1,5 @@
use crate::runtime::scheduler::multi_thread::queue;
use crate::runtime::task::{self, Inject, Schedule, Task};
use crate::runtime::thread_pool::queue;
use crate::runtime::MetricsBatch;
use std::thread;
tokio/src/task/blocking.rs
+2 -2
View File
@@ -74,7 +74,7 @@ cfg_rt_multi_thread! {
where
F: FnOnce() -> R,
{
crate::runtime::thread_pool::block_in_place(f)
crate::runtime::scheduler::multi_thread::block_in_place(f)
}
}
@@ -119,7 +119,7 @@ cfg_rt! {
/// is not guaranteed.
///
/// Note that if you are using the single threaded runtime, this function will
/// still spawn additional threads for blocking operations. The basic
/// still spawn additional threads for blocking operations. The current-thread
/// scheduler's single thread is only used for asynchronous code.
///
/// # Related APIs and patterns for bridging asynchronous and blocking code
tokio/tests/rt_common.rs
+1 -1
View File
@@ -2,7 +2,7 @@
#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]
// Tests to run on both current-thread & thread-pool runtime variants.
// Tests to run on both current-thread & multi-thread runtime variants.
macro_rules! rt_test {
($($t:tt)*) => {
tokio/tests/rt_metrics.rs
+12 -12
View File
@@ -6,7 +6,7 @@ use tokio::time::{self, Duration};
#[test]
fn num_workers() {
let rt = basic();
let rt = current_thread();
assert_eq!(1, rt.metrics().num_workers());
let rt = threaded();
@@ -17,7 +17,7 @@ fn num_workers() {
fn remote_schedule_count() {
use std::thread;
let rt = basic();
let rt = current_thread();
let handle = rt.handle().clone();
let task = thread::spawn(move || {
handle.spawn(async {
@@ -48,7 +48,7 @@ fn remote_schedule_count() {
#[test]
fn worker_park_count() {
let rt = basic();
let rt = current_thread();
let metrics = rt.metrics();
rt.block_on(async {
time::sleep(Duration::from_millis(1)).await;
@@ -71,7 +71,7 @@ fn worker_noop_count() {
// There isn't really a great way to generate no-op parks as they happen as
// false-positive events under concurrency.
let rt = basic();
let rt = current_thread();
let metrics = rt.metrics();
rt.block_on(async {
time::sleep(Duration::from_millis(1)).await;
@@ -133,7 +133,7 @@ fn worker_steal_count() {
fn worker_poll_count() {
const N: u64 = 5;
let rt = basic();
let rt = current_thread();
let metrics = rt.metrics();
rt.block_on(async {
for _ in 0..N {
@@ -165,7 +165,7 @@ fn worker_total_busy_duration() {
let zero = Duration::from_millis(0);
let rt = basic();
let rt = current_thread();
let metrics = rt.metrics();
rt.block_on(async {
@@ -204,7 +204,7 @@ fn worker_total_busy_duration() {
#[test]
fn worker_local_schedule_count() {
let rt = basic();
let rt = current_thread();
let metrics = rt.metrics();
rt.block_on(async {
tokio::spawn(async {}).await.unwrap();
@@ -280,7 +280,7 @@ fn worker_overflow_count() {
fn injection_queue_depth() {
use std::thread;
let rt = basic();
let rt = current_thread();
let handle = rt.handle().clone();
let metrics = rt.metrics();
@@ -321,7 +321,7 @@ fn injection_queue_depth() {
fn worker_local_queue_depth() {
const N: usize = 100;
let rt = basic();
let rt = current_thread();
let metrics = rt.metrics();
rt.block_on(async {
for _ in 0..N {
@@ -372,7 +372,7 @@ fn worker_local_queue_depth() {
#[cfg(any(target_os = "linux", target_os = "macos"))]
#[test]
fn io_driver_fd_count() {
let rt = basic();
let rt = current_thread();
let metrics = rt.metrics();
// Since this is enabled w/ the process driver we always
@@ -394,7 +394,7 @@ fn io_driver_fd_count() {
#[cfg(any(target_os = "linux", target_os = "macos"))]
#[test]
fn io_driver_ready_count() {
let rt = basic();
let rt = current_thread();
let metrics = rt.metrics();
let stream = tokio::net::TcpStream::connect("google.com:80");
@@ -403,7 +403,7 @@ fn io_driver_ready_count() {
assert_eq!(metrics.io_driver_ready_count(), 1);
}
fn basic() -> Runtime {
fn current_thread() -> Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
tokio/tests/rt_panic.rs
+2 -2
View File
@@ -26,7 +26,7 @@ fn current_handle_panic_caller() -> Result<(), Box<dyn Error>> {
#[test]
fn into_panic_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(move || {
let rt = basic();
let rt = current_thread();
rt.block_on(async {
let handle = tokio::spawn(future::pending::<()>());
@@ -69,7 +69,7 @@ fn builder_max_blocking_threads_panic_caller() -> Result<(), Box<dyn Error>> {
Ok(())
}
fn basic() -> Runtime {
fn current_thread() -> Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
tokio/tests/sync_panic.rs
+8 -8
View File
@@ -27,7 +27,7 @@ fn broadcast_channel_panic_caller() -> Result<(), Box<dyn Error>> {
#[test]
fn mutex_blocking_lock_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
rt.block_on(async {
let mutex = Mutex::new(5_u32);
let _g = mutex.blocking_lock();
@@ -43,7 +43,7 @@ fn mutex_blocking_lock_panic_caller() -> Result<(), Box<dyn Error>> {
#[test]
fn oneshot_blocking_recv_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
rt.block_on(async {
let (_tx, rx) = oneshot::channel::<u8>();
let _ = rx.blocking_recv();
@@ -71,7 +71,7 @@ fn rwlock_with_max_readers_panic_caller() -> Result<(), Box<dyn Error>> {
#[test]
fn rwlock_blocking_read_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
rt.block_on(async {
let lock = RwLock::<u8>::new(0);
let _ = lock.blocking_read();
@@ -87,7 +87,7 @@ fn rwlock_blocking_read_panic_caller() -> Result<(), Box<dyn Error>> {
#[test]
fn rwlock_blocking_write_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
rt.block_on(async {
let lock = RwLock::<u8>::new(0);
let _ = lock.blocking_write();
@@ -115,7 +115,7 @@ fn mpsc_bounded_channel_panic_caller() -> Result<(), Box<dyn Error>> {
#[test]
fn mpsc_bounded_receiver_blocking_recv_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
let (_tx, mut rx) = mpsc::channel::<u8>(1);
rt.block_on(async {
let _ = rx.blocking_recv();
@@ -131,7 +131,7 @@ fn mpsc_bounded_receiver_blocking_recv_panic_caller() -> Result<(), Box<dyn Erro
#[test]
fn mpsc_bounded_sender_blocking_send_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
let (tx, _rx) = mpsc::channel::<u8>(1);
rt.block_on(async {
let _ = tx.blocking_send(3);
@@ -147,7 +147,7 @@ fn mpsc_bounded_sender_blocking_send_panic_caller() -> Result<(), Box<dyn Error>
#[test]
fn mpsc_unbounded_receiver_blocking_recv_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
let (_tx, mut rx) = mpsc::unbounded_channel::<u8>();
rt.block_on(async {
let _ = rx.blocking_recv();
@@ -160,6 +160,6 @@ fn mpsc_unbounded_receiver_blocking_recv_panic_caller() -> Result<(), Box<dyn Er
Ok(())
}
fn basic() -> Runtime {
fn current_thread() -> Runtime {
Builder::new_current_thread().enable_all().build().unwrap()
}
tokio/tests/task_abort.rs
+1 -1
View File
@@ -89,7 +89,7 @@ fn test_abort_without_panic_3662() {
// Note: We do the following to trigger a deferred task cleanup.
//
// The relevant piece of code you want to look at is in:
// `Inner::block_on` of `basic_scheduler.rs`.
// `Inner::block_on` of `scheduler/current_thread.rs`.
//
// We cause the cleanup to happen by having a poll return Pending once
// so that the scheduler can go into the "auxiliary tasks" mode, at
tokio/tests/task_blocking.rs
+3 -3
View File
@@ -77,7 +77,7 @@ async fn block_in_block() {
#[tokio::test(flavor = "current_thread")]
#[should_panic]
async fn no_block_in_basic_scheduler() {
async fn no_block_in_current_thread_scheduler() {
task::block_in_place(|| {});
}
@@ -91,7 +91,7 @@ fn yes_block_in_threaded_block_on() {
#[test]
#[should_panic]
fn no_block_in_basic_block_on() {
fn no_block_in_current_thread_block_on() {
let rt = runtime::Builder::new_current_thread().build().unwrap();
rt.block_on(async {
task::block_in_place(|| {});
@@ -99,7 +99,7 @@ fn no_block_in_basic_block_on() {
}
#[test]
fn can_enter_basic_rt_from_within_block_in_place() {
fn can_enter_current_thread_rt_from_within_block_in_place() {
let outer = tokio::runtime::Runtime::new().unwrap();
outer.block_on(async {
tokio/tests/task_local_set.rs
+1 -1
View File
@@ -22,7 +22,7 @@ use std::sync::atomic::Ordering::SeqCst;
use std::time::Duration;
#[tokio::test(flavor = "current_thread")]
async fn local_basic_scheduler() {
async fn local_current_thread_scheduler() {
LocalSet::new()
.run_until(async {
task::spawn_local(async {}).await.unwrap();
tokio/tests/time_panic.rs
+3 -3
View File
@@ -15,7 +15,7 @@ use support::panic::test_panic;
#[test]
fn pause_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
rt.block_on(async {
time::pause();
@@ -32,7 +32,7 @@ fn pause_panic_caller() -> Result<(), Box<dyn Error>> {
#[test]
fn resume_panic_caller() -> Result<(), Box<dyn Error>> {
let panic_location_file = test_panic(|| {
let rt = basic();
let rt = current_thread();
rt.block_on(async {
time::resume();
@@ -85,7 +85,7 @@ fn timeout_panic_caller() -> Result<(), Box<dyn Error>> {
Ok(())
}
fn basic() -> Runtime {
fn current_thread() -> Runtime {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
tokio/tests/time_rt.rs
+1 -1
View File
@@ -26,7 +26,7 @@ fn timer_with_threaded_runtime() {
}
#[test]
fn timer_with_basic_scheduler() {
fn timer_with_current_thread_scheduler() {
use tokio::runtime::Builder;
let rt = Builder::new_current_thread().enable_all().build().unwrap();