Start work on compat feature

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 }

tokio/src/lib.rs

@@ -433,3 +433,6 @@ cfg_macros! {
 #[cfg(feature = "io-util")]
 #[cfg(test)]
 fn is_unpin<T: Unpin>() {}
+
+#[cfg(all(feature = "tokio_03", not(feature = "compat")))]
+compile_error!("Enable the `compat` feature rather than the `tokio_03` feature.");

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
+        )*
+    }
+}

tokio/src/runtime/builder.rs

@@ -310,12 +310,22 @@ impl Builder {
     /// });
     /// ```
     pub fn build(&mut self) -> io::Result<Runtime> {
-        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<Runtime> {
+            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()

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<Shared>,
+    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<Shared>,
+}
+
+#[derive(Debug)]
+struct Shared {
+    runtime: RwLock<Option<tokio_03::runtime::Runtime>>,
+    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<tokio_03::runtime::Runtime> {
+        self.wait_on_drop = false;
+        self.shared.take_timeout_or_kill(timeout)
+    }
+
+    pub(crate) fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+    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<F>(&self, future: F) -> F::Output
+    where
+        F: Future,
+    {
+        self.shared.block_on(future).expect("Runtime shut down")
+    }
+}
+
+impl Compat03Handle {
+    pub(crate) fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+    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<F>(&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<F>(&self, future: F) -> Option<JoinHandle<F::Output>>
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        self.with_runtime(move |runtime| {
+            JoinHandle::new_compat(runtime.spawn(future))
+        })
+    }
+
+    fn block_on<F>(&self, future: F) -> Option<F::Output>
+    where
+        F: Future,
+    {
+        self.with_runtime(move |runtime| {
+            runtime.block_on(future)
+        })
+    }
+
+    fn with_runtime<F, O>(&self, func: F) -> Option<O>
+    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<tokio_03::runtime::Runtime> {
+        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());
+        }
+    }
+}

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<dyn Fn() + Send + Sync>;
+type Callback = Arc<dyn Fn() + Send + Sync>;
 
 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.

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),
             }
         }
     }

tokio/src/runtime/task/error.rs

@@ -13,6 +13,8 @@ doc_rt_core! {
 enum Repr {
     Cancelled,
     Panic(Mutex<Box<dyn Any + Send + 'static>>),
+    #[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<Box<dyn Any + Send + 'static>, 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<JoinError> 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<tokio_03::task::JoinError> for JoinError {
+        fn from(src: tokio_03::task::JoinError) -> JoinError {
+            JoinError {
+                repr: Repr::Compat03(src),
+            }
+        }
+    }
+}

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<T> {
+        #[cfg(feature = "compat")]
+        compat: Option<tokio_03::task::JoinHandle<T>>,
         raw: Option<RawTask>,
         _p: PhantomData<T>,
     }
@@ -87,10 +89,20 @@ unsafe impl<T: Send> Sync for JoinHandle<T> {}
 impl<T> JoinHandle<T> {
     pub(super) fn new(raw: RawTask) -> JoinHandle<T> {
         JoinHandle {
+            #[cfg(feature = "compat")]
+            compat: None,
             raw: Some(raw),
             _p: PhantomData,
         }
     }
+    #[cfg(feature = "compat")]
+    pub(crate) fn new_compat(compat: tokio_03::task::JoinHandle<T>) -> JoinHandle<T> {
+        JoinHandle {
+            compat: Some(compat),
+            raw: None,
+            _p: PhantomData,
+        }
+    }
 }
 
 impl<T> Unpin for JoinHandle<T> {}
@@ -98,7 +110,14 @@ impl<T> Unpin for JoinHandle<T> {}
 impl<T> Future for JoinHandle<T> {
     type Output = super::Result<T>;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        #[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