Start work on compat feature

This makes it consistent with `TcpListener::poll_accept` and other
public poll methods the library provides. This particular method is
useful for writing generic code that accepts connections since async
functions can't easily be used with traits. It is possible to
generically accept connections with `Incoming`, however, this doesn't
return the incoming `SocketAddr`.

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/net/tcp/listener.rs

@@ -185,10 +185,10 @@ impl TcpListener {
         poll_fn(|cx| self.poll_accept(cx)).await
     }
 
-    /// Attempts to poll `SocketAddr` and `TcpStream` bound to this address.
+    /// Polls to accept a new incoming connection to this listener.
     ///
-    /// In case if I/O resource isn't ready yet, `Poll::Pending` is returned and
-    /// current task will be notified by a waker.
+    /// If there is no connection to accept, `Poll::Pending` is returned and
+    /// the current task will be notified by a waker.
     pub fn poll_accept(
         &mut self,
         cx: &mut Context<'_>,

tokio/src/net/unix/listener.rs

@@ -104,7 +104,11 @@ impl UnixListener {
         poll_fn(|cx| self.poll_accept(cx)).await
     }
 
-    pub(crate) fn poll_accept(
+    /// Polls to accept a new incoming connection to this listener.
+    ///
+    /// If there is no connection to accept, `Poll::Pending` is returned and
+    /// the current task will be notified by a waker.
+    pub fn poll_accept(
         &mut self,
         cx: &mut Context<'_>,
     ) -> Poll<io::Result<(UnixStream, SocketAddr)>> {

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