diff --git a/tokio/Cargo.toml b/tokio/Cargo.toml index 966bfe3ea..6a9a71213 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -90,6 +90,8 @@ time = ["slab"] udp = ["io-driver"] uds = ["io-driver", "mio-uds", "libc"] +compat = ["tokio_03", "full", "parking_lot"] + [dependencies] tokio-macros = { version = "0.2.4", path = "../tokio-macros", optional = true } @@ -108,6 +110,9 @@ parking_lot = { version = "0.11.0", optional = true } # Not in full slab = { version = "0.4.1", optional = true } # Backs `DelayQueue` tracing = { version = "0.1.16", default-features = false, features = ["std"], optional = true } # Not in full +# Compatibility with 0.3 +tokio_03 = { git = "https://github.com/tokio-rs/tokio.git", branch = "master", package = "tokio", features = ["full"], optional = true } + [target.'cfg(unix)'.dependencies] mio-uds = { version = "0.6.5", optional = true } libc = { version = "0.2.42", optional = true } diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index 11a111950..ae99aadc4 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -433,3 +433,6 @@ cfg_macros! { #[cfg(feature = "io-util")] #[cfg(test)] fn is_unpin() {} + +#[cfg(all(feature = "tokio_03", not(feature = "compat")))] +compile_error!("Enable the `compat` feature rather than the `tokio_03` feature."); diff --git a/tokio/src/macros/cfg.rs b/tokio/src/macros/cfg.rs index 4b77544eb..69516c26e 100644 --- a/tokio/src/macros/cfg.rs +++ b/tokio/src/macros/cfg.rs @@ -402,3 +402,21 @@ macro_rules! cfg_coop { )* } } + +macro_rules! cfg_compat { + ($($item:item)*) => { + $( + #[cfg(feature = "compat")] + $item + )* + } +} + +macro_rules! cfg_not_compat { + ($($item:item)*) => { + $( + #[cfg(feature = "compat")] + $item + )* + } +} diff --git a/tokio/src/runtime/builder.rs b/tokio/src/runtime/builder.rs index fad72c7ad..4d7f803e3 100644 --- a/tokio/src/runtime/builder.rs +++ b/tokio/src/runtime/builder.rs @@ -310,12 +310,22 @@ impl Builder { /// }); /// ``` pub fn build(&mut self) -> io::Result { - match self.kind { - Kind::Shell => self.build_shell_runtime(), - #[cfg(feature = "rt-core")] - Kind::Basic => self.build_basic_runtime(), - #[cfg(feature = "rt-threaded")] - Kind::ThreadPool => self.build_threaded_runtime(), + if cfg!(feature = "compat") { + match self.kind { + Kind::Shell => self.build_shell_runtime(), + #[cfg(feature = "rt-core")] + Kind::Basic => self.build_compat_runtime(), + #[cfg(feature = "rt-threaded")] + Kind::ThreadPool => self.build_compat_runtime(), + } + } else { + match self.kind { + Kind::Shell => self.build_shell_runtime(), + #[cfg(feature = "rt-core")] + Kind::Basic => self.build_basic_runtime(), + #[cfg(feature = "rt-threaded")] + Kind::ThreadPool => self.build_threaded_runtime(), + } } } @@ -394,6 +404,47 @@ cfg_time! { } } +cfg_compat! { + impl Builder { + fn make_03_builder(&self) -> tokio_03::runtime::Builder { + let mut builder = match self.kind { + // the make_03_builder function should not be called when building + // a shell runtime + Kind::Shell => unreachable!(), + Kind::Basic => tokio_03::runtime::Builder::new_current_thread(), + Kind::ThreadPool => tokio_03::runtime::Builder::new_multi_thread(), + }; + + if self.enable_io { + builder.enable_io(); + } + if self.enable_time { + builder.enable_time(); + } + + if let Some(core_threads) = self.core_threads { + builder.worker_threads(core_threads); + } + builder.max_threads(self.max_threads); + + if self.thread_name.as_str() != "tokio-runtime-worker" { + builder.thread_name(self.thread_name.as_str()); + } + if let Some(thread_stack_size) = self.thread_stack_size { + builder.thread_stack_size(thread_stack_size); + } + if let Some(after_start) = self.after_start.clone() { + builder.on_thread_start(move || after_start()); + } + if let Some(before_stop) = self.before_stop.clone() { + builder.on_thread_stop(move || before_stop()); + } + + builder + } + } +} + cfg_rt_core! { impl Builder { /// Sets runtime to use a simpler scheduler that runs all tasks on the current-thread. @@ -501,6 +552,38 @@ cfg_rt_threaded! { } } +cfg_compat! { + use crate::runtime::Compat03Runtime; + + impl Builder { + fn build_compat_runtime(&mut self) -> io::Result { + use crate::runtime::Kind; + + let rt02 = self.build_shell_runtime()?; + let rt03 = Compat03Runtime::new(self.make_03_builder().build()?, rt02); + + let spawner = Spawner::Compat03(rt03.handle()); + let io_handle = rt02.handle.io_handle.clone(); + let time_handle = rt02.handle.time_handle.clone(); + let clock = rt02.handle.clock.clone(); + let blocking_pool = blocking::create_blocking_pool(self, self.max_threads); + let blocking_spawner = blocking_pool.spawner().clone(); + + Ok(Runtime { + kind: Kind::Compat03(rt03), + handle: Handle { + spawner, + io_handle, + time_handle, + clock, + blocking_spawner, + }, + blocking_pool, + }) + } + } +} + impl Default for Builder { fn default() -> Self { Self::new() diff --git a/tokio/src/runtime/compat.rs b/tokio/src/runtime/compat.rs new file mode 100644 index 000000000..4645e5374 --- /dev/null +++ b/tokio/src/runtime/compat.rs @@ -0,0 +1,183 @@ +use std::sync::{Arc, Weak}; +use crate::task::JoinHandle; +use std::future::Future; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use parking_lot::RwLock; +use std::time::Duration; +use crate::runtime::task; + +#[derive(Debug)] +pub(crate) struct Compat03Runtime { + shared: Arc, + wait_on_drop: bool, + // This keeps the IO driver alive + rt02_chan: crate::sync::oneshot::Sender<()>, +} + +#[derive(Debug, Clone)] +pub(crate) struct Compat03Handle { + shared: Weak, +} + +#[derive(Debug)] +struct Shared { + runtime: RwLock>, + num_read_locks: AtomicUsize, + kill_on_unlock: AtomicBool, +} + +impl Compat03Runtime { + pub(crate) fn new( + rt: tokio_03::runtime::Runtime, + driver: crate::runtime::Runtime + ) -> Self { + let (rt02_chan, rt02_recv) = crate::sync::oneshot::channel(); + + std::thread::spawn(move || { + let _ = driver.block_on(rt02_recv); + }); + + Self { + shared: Arc::new(Shared { + runtime: RwLock::new(Some(rt)), + num_read_locks: AtomicUsize::new(0), + kill_on_unlock: AtomicBool::new(false), + }), + wait_on_drop: true, + rt02_chan, + } + } + pub(crate) fn handle(&self) -> Compat03Handle { + Compat03Handle { + shared: Arc::downgrade(&self.shared), + } + } + pub(crate) fn take(self, timeout: Duration) -> Option { + self.wait_on_drop = false; + self.shared.take_timeout_or_kill(timeout) + } + + pub(crate) fn spawn(&self, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + match self.shared.spawn(future) { + Some(shared) => shared, + None => task::joinable(async move { unreachable!() }).1, + } + } + + pub(crate) fn block_on(&self, future: F) -> F::Output + where + F: Future, + { + self.shared.block_on(future).expect("Runtime shut down") + } +} + +impl Compat03Handle { + pub(crate) fn spawn(&self, future: F) -> JoinHandle + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + match self.shared.upgrade().and_then(move |shared| shared.spawn(future)) { + Some(shared) => shared, + None => task::joinable(async move { unreachable!() }).1, + } + } + + pub(crate) fn block_on(&self, future: F) -> F::Output + where + F: Future, + { + self.shared.upgrade() + .and_then(|shared| shared.block_on(future)) + .expect("Runtime shut down") + } +} + +impl Shared { + fn spawn(&self, future: F) -> Option> + where + F: Future + Send + 'static, + F::Output: Send + 'static, + { + self.with_runtime(move |runtime| { + JoinHandle::new_compat(runtime.spawn(future)) + }) + } + + fn block_on(&self, future: F) -> Option + where + F: Future, + { + self.with_runtime(move |runtime| { + runtime.block_on(future) + }) + } + + fn with_runtime(&self, func: F) -> Option + where + F: FnOnce(&tokio_03::runtime::Runtime) -> O + { + if self.kill_on_unlock.load(Ordering::SeqCst) { + return None; + } + self.num_read_locks.fetch_add(1, Ordering::SeqCst); + + struct Guard<'a> { + shared: &'a Shared, + } + impl<'a> Drop for Guard<'a> { + fn drop(&mut self) { + let num_locks = self.shared.num_read_locks.fetch_sub(1, Ordering::SeqCst); + let is_last = num_locks == 1; + let kill_on_unlock = self.shared.kill_on_unlock.load(Ordering::SeqCst); + if kill_on_unlock && is_last { + if let Some(rt) = self.shared.runtime.write().take() { + rt.shutdown_background(); + } + } + } + } + + let guard = Guard { shared: self }; + let rw_guard = self.runtime.read(); + let res = rw_guard.as_ref().map(func); + drop(rw_guard); + drop(guard); + + res + } + + // Take out the runtime from the rwlock, waiting at most `dur` for active read locks + // to give up the lock. If the lock is not obtained in this duration, ask the last + // read lock to shut down the runtime when it finishes. + fn take_timeout_or_kill(&self, dur: Duration) -> Option { + match self.runtime.try_write_for(dur) { + Some(runtime_opt) => runtime_opt.take(), + None => { + // Ask the last read lock to kill the runtime when it is done. + self.kill_on_unlock.store(true, Ordering::SeqCst); + + // In case the last read lock exited before we got to make the atomic + // write, try to lock it again. + match self.runtime.try_write() { + Some(runtime_opt) => runtime_opt.take(), + None => None, + } + }, + } + } +} + +impl Drop for Compat03Runtime { + fn drop(&mut self) { + if self.wait_on_drop { + // This will wait indefinitely for the lock. + drop(self.shared.runtime.write().take()); + } + } +} diff --git a/tokio/src/runtime/mod.rs b/tokio/src/runtime/mod.rs index 637f38cab..2db92af09 100644 --- a/tokio/src/runtime/mod.rs +++ b/tokio/src/runtime/mod.rs @@ -208,6 +208,11 @@ cfg_blocking_impl! { mod builder; pub use self::builder::Builder; +cfg_compat! { + pub(crate) mod compat; + use self::compat::Compat03Runtime; +} + pub(crate) mod enter; use self::enter::enter; @@ -242,6 +247,7 @@ cfg_rt_core! { use std::future::Future; use std::time::Duration; +use std::sync::Arc; /// The Tokio runtime. /// @@ -297,10 +303,13 @@ enum Kind { /// Execute tasks across multiple threads. #[cfg(feature = "rt-threaded")] ThreadPool(ThreadPool), + + #[cfg(feature = "compat")] + Compat03(Compat03Runtime), } /// After thread starts / before thread stops -type Callback = std::sync::Arc; +type Callback = Arc; impl Runtime { /// Create a new runtime instance with default configuration values. @@ -395,6 +404,8 @@ impl Runtime { { match &self.kind { Kind::Shell(_) => panic!("task execution disabled"), + #[cfg(feature = "compat")] + Kind::Compat03(handle) => handle.spawn(future), #[cfg(feature = "rt-threaded")] Kind::ThreadPool(exec) => exec.spawn(future), Kind::Basic(exec) => exec.spawn(future), @@ -444,6 +455,8 @@ impl Runtime { Kind::Basic(exec) => exec.block_on(future), #[cfg(feature = "rt-threaded")] Kind::ThreadPool(exec) => exec.block_on(future), + #[cfg(feature = "compat")] + Kind::Compat03(exec) => exec.block_on(future), }) } @@ -546,6 +559,21 @@ impl Runtime { // Wakeup and shutdown all the worker threads self.handle.spawner.shutdown(); self.blocking_pool.shutdown(Some(duration)); + + #[cfg(feature = "compat")] + { + if let Kind::Compat03(compat) = self.kind { + let deadline = std::time::Instant::now() + duration; + if let Some(rt) = compat.take(duration) { + let now = std::time::Instant::now(); + rt.shutdown_timeout( + deadline + .checked_duration_since(now) + .unwrap_or(Duration::from_nanos(0)) + ); + } + } + } } /// Shutdown the runtime, without waiting for any spawned tasks to shutdown. diff --git a/tokio/src/runtime/spawner.rs b/tokio/src/runtime/spawner.rs index c5f2d17cd..bf5f6b934 100644 --- a/tokio/src/runtime/spawner.rs +++ b/tokio/src/runtime/spawner.rs @@ -16,6 +16,8 @@ pub(crate) enum Spawner { Basic(basic_scheduler::Spawner), #[cfg(feature = "rt-threaded")] ThreadPool(thread_pool::Spawner), + #[cfg(feature = "compat")] + Compat03(crate::runtime::compat::Compat03Handle), } impl Spawner { @@ -42,6 +44,8 @@ cfg_rt_core! { Spawner::Basic(spawner) => spawner.spawn(future), #[cfg(feature = "rt-threaded")] Spawner::ThreadPool(spawner) => spawner.spawn(future), + #[cfg(feature = "compat")] + Spawner::Compat03(handle) => handle.spawn(future), } } } diff --git a/tokio/src/runtime/task/error.rs b/tokio/src/runtime/task/error.rs index d5f65a498..f5dd51646 100644 --- a/tokio/src/runtime/task/error.rs +++ b/tokio/src/runtime/task/error.rs @@ -13,6 +13,8 @@ doc_rt_core! { enum Repr { Cancelled, Panic(Mutex>), + #[cfg(feature = "compat")] + Compat03(tokio_03::task::JoinError), } impl JoinError { @@ -44,6 +46,8 @@ impl JoinError { pub fn is_cancelled(&self) -> bool { match &self.repr { Repr::Cancelled => true, + #[cfg(feature = "compat")] + Repr::Compat03(join) => join.is_cancelled(), _ => false, } } @@ -67,6 +71,8 @@ impl JoinError { pub fn is_panic(&self) -> bool { match &self.repr { Repr::Panic(_) => true, + #[cfg(feature = "compat")] + Repr::Compat03(join) => join.is_panic(), _ => false, } } @@ -125,6 +131,8 @@ impl JoinError { pub fn try_into_panic(self) -> Result, JoinError> { match self.repr { Repr::Panic(p) => Ok(p.into_inner().expect("Extracting panic from mutex")), + #[cfg(feature = "compat")] + Repr::Compat03(join) => Ok(join.try_into_panic()?), _ => Err(self), } } @@ -135,6 +143,8 @@ impl fmt::Display for JoinError { match &self.repr { Repr::Cancelled => write!(fmt, "cancelled"), Repr::Panic(_) => write!(fmt, "panic"), + #[cfg(feature = "compat")] + Repr::Compat03(inner) => fmt::Display::fmt(inner, fmt), } } } @@ -144,6 +154,8 @@ impl fmt::Debug for JoinError { match &self.repr { Repr::Cancelled => write!(fmt, "JoinError::Cancelled"), Repr::Panic(_) => write!(fmt, "JoinError::Panic(...)"), + #[cfg(feature = "compat")] + Repr::Compat03(inner) => fmt::Debug::fmt(inner, fmt), } } } @@ -157,7 +169,19 @@ impl From for io::Error { match src.repr { Repr::Cancelled => "task was cancelled", Repr::Panic(_) => "task panicked", + #[cfg(feature = "compat")] + Repr::Compat03(inner) => return io::Error::from(inner), }, ) } } + +cfg_compat! { + impl From for JoinError { + fn from(src: tokio_03::task::JoinError) -> JoinError { + JoinError { + repr: Repr::Compat03(src), + } + } + } +} diff --git a/tokio/src/runtime/task/join.rs b/tokio/src/runtime/task/join.rs index 3c4aabb2e..872758ecb 100644 --- a/tokio/src/runtime/task/join.rs +++ b/tokio/src/runtime/task/join.rs @@ -76,6 +76,8 @@ doc_rt_core! { /// [`task::spawn_blocking`]: crate::task::spawn_blocking /// [`std::thread::JoinHandle`]: std::thread::JoinHandle pub struct JoinHandle { + #[cfg(feature = "compat")] + compat: Option>, raw: Option, _p: PhantomData, } @@ -87,10 +89,20 @@ unsafe impl Sync for JoinHandle {} impl JoinHandle { pub(super) fn new(raw: RawTask) -> JoinHandle { JoinHandle { + #[cfg(feature = "compat")] + compat: None, raw: Some(raw), _p: PhantomData, } } + #[cfg(feature = "compat")] + pub(crate) fn new_compat(compat: tokio_03::task::JoinHandle) -> JoinHandle { + JoinHandle { + compat: Some(compat), + raw: None, + _p: PhantomData, + } + } } impl Unpin for JoinHandle {} @@ -98,7 +110,14 @@ impl Unpin for JoinHandle {} impl Future for JoinHandle { type Output = super::Result; - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + #[cfg(feature = "compat")] + { + if let Some(compat) = &mut self.compat { + return Pin::new(compat).poll(cx)?.map(Ok); + } + } + let mut ret = Poll::Pending; // Keep track of task budget