diff --git a/Cargo.toml b/Cargo.toml index 0d7cb00fb..0209180c2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,9 +6,9 @@ name = "tokio" # - Update CHANGELOG.md. # - Create "v0.1.x" git tag. version = "0.1.1" - authors = ["Carl Lerche "] license = "MIT/Apache-2.0" +readme = "README.md" repository = "https://github.com/tokio-rs/tokio" homepage = "https://tokio.rs" documentation = "https://docs.rs/tokio/0.1" @@ -16,7 +16,7 @@ description = """ An event-driven, non-blocking I/O platform for writing asynchronous I/O backed applications. """ -categories = ["asynchronous"] +categories = ["asynchronous", "network-programming"] keywords = ["io", "async", "non-blocking", "futures"] [workspace] @@ -25,6 +25,7 @@ members = [ "./", "tokio-executor", "tokio-io", + "tokio-reactor", "tokio-threadpool", ] @@ -35,6 +36,7 @@ appveyor = { repository = "carllerche/tokio" } [dependencies] tokio-io = { version = "0.1", path = "tokio-io" } tokio-executor = { version = "0.1", path = "tokio-executor" } +tokio-reactor = { version = "0.1", path = "tokio-reactor" } tokio-threadpool = { version = "0.1", path = "tokio-threadpool" } bytes = "0.4" log = "0.4" diff --git a/LICENSE-MIT b/LICENSE-MIT index fba4c215a..38c1e27b8 100644 --- a/LICENSE-MIT +++ b/LICENSE-MIT @@ -1,4 +1,4 @@ -Copyright (c) 2018 Tokio Authors +Copyright (c) 2018 Tokio Contributors Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated diff --git a/README.md b/README.md index 33cf17482..50bd3a29a 100644 --- a/README.md +++ b/README.md @@ -87,7 +87,30 @@ fn main() { } ``` -# License +## Project layout + +The `tokio` crate, found at the root, is primarily intended for use by +application developers. Library authors should depend on the sub crates, which +have greater guarantees of stability. + +The crates included as part of Tokio are: + +* [`tokio-executor`]: Task execution related traits and utilities. + +* [`tokio-io`]: Asynchronous I/O related traits and utilities. + +* [`tokio-reactor`]: Event loop that drives I/O resources (like TCP and UDP + sockets). + +* [`tokio-threadpool`]: Schedules the execution of futures across a pool of + threads. + +[`tokio-executor`]: tokio-executor +[`tokio-io`]: tokio-io +[`tokio-reactor`]: tokio-reactor +[`tokio-threadpool`]: tokio-threadpool + +## License This project is licensed under either of diff --git a/src/lib.rs b/src/lib.rs index 4c32bfa86..9de17a9f3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -74,6 +74,7 @@ extern crate slab; #[macro_use] extern crate tokio_io; extern crate tokio_executor; +extern crate tokio_reactor; extern crate tokio_threadpool; #[macro_use] @@ -87,8 +88,6 @@ pub mod runtime; pub use executor::spawn; pub use runtime::run; -mod atomic_task; - pub mod io { //! Asynchronous I/O. //! diff --git a/src/reactor/mod.rs b/src/reactor/mod.rs index 2a936bafb..3f7603ee5 100644 --- a/src/reactor/mod.rs +++ b/src/reactor/mod.rs @@ -1,4 +1,4 @@ -//! Event loop that drives I/O resources. +//! Event loop that drives Tokio I/O resources. //! //! This module contains [`Reactor`], which is the event loop that drives all //! Tokio I/O resources. It is the reactor's job to receive events from the @@ -107,7 +107,7 @@ //! There are a couple of ways to do this. //! //! If the custom I/O resource implements [`mio::Evented`] and implements -//! [`std::Read`] and / or [`std::Write`], then [`PollEvented2`] is the most +//! [`std::Read`] and / or [`std::Write`], then [`PollEvented`] is the most //! suited. //! //! Otherwise, [`Registration`] can be used directly. This provides the lowest @@ -131,742 +131,19 @@ //! [`Reactor::poll`]: struct.Reactor.html#method.poll //! [`Poll::poll`]: https://docs.rs/mio/0.6/mio/struct.Poll.html#method.poll //! [`mio::Evented`]: https://docs.rs/mio/0.6/mio/trait.Evented.html -//! [`PollEvented2`]: struct.PollEvented2.html +//! [`PollEvented`]: struct.PollEvented.html //! [`std::Read`]: https://doc.rust-lang.org/std/io/trait.Read.html //! [`std::Write`]: https://doc.rust-lang.org/std/io/trait.Write.html -use tokio_executor::Enter; -use tokio_executor::park::{Park, Unpark}; - -use atomic_task::AtomicTask; - -use std::{fmt, usize}; -use std::io::{self, ErrorKind}; -use std::mem; -use std::cell::RefCell; -use std::sync::atomic::Ordering::{Relaxed, SeqCst}; -use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT}; -use std::sync::{Arc, Weak, RwLock}; -use std::time::{Duration, Instant}; - -use log::Level; -use mio; -use mio::event::Evented; -use slab::Slab; -use futures::task::Task; - -pub(crate) mod background; -use self::background::Background; +pub use tokio_reactor::{ + Reactor, + Handle, + Background, + Turn, + Registration, + PollEvented as PollEvented2, +}; mod poll_evented; #[allow(deprecated)] pub use self::poll_evented::PollEvented; - -mod registration; -pub use self::registration::Registration; - -mod poll_evented2; -pub use self::poll_evented2::PollEvented as PollEvented2; - -/// The core reactor, or event loop. -/// -/// The event loop is the main source of blocking in an application which drives -/// all other I/O events and notifications happening. Each event loop can have -/// multiple handles pointing to it, each of which can then be used to create -/// various I/O objects to interact with the event loop in interesting ways. -pub struct Reactor { - /// Reuse the `mio::Events` value across calls to poll. - events: mio::Events, - - /// State shared between the reactor and the handles. - inner: Arc, - - _wakeup_registration: mio::Registration, -} - -/// A reference to a reactor. -/// -/// A `Handle` is used for associating I/O objects with an event loop -/// explicitly. Typically though you won't end up using a `Handle` that often -/// and will instead use the default reactor for the execution context. -#[derive(Clone)] -pub struct Handle { - inner: Weak, -} - -/// Return value from the `turn` method on `Reactor`. -/// -/// Currently this value doesn't actually provide any functionality, but it may -/// in the future give insight into what happened during `turn`. -#[derive(Debug)] -pub struct Turn { - _priv: (), -} - -/// Error returned from `Handle::set_fallback`. -#[derive(Clone, Debug)] -pub struct SetFallbackError(()); - -#[deprecated(since = "0.1.2", note = "use SetFallbackError instead")] -#[doc(hidden)] -pub type SetDefaultError = SetFallbackError; - -struct Inner { - /// The underlying system event queue. - io: mio::Poll, - - /// Dispatch slabs for I/O and futures events - io_dispatch: RwLock>, - - /// Used to wake up the reactor from a call to `turn` - wakeup: mio::SetReadiness -} - -struct ScheduledIo { - readiness: AtomicUsize, - reader: AtomicTask, - writer: AtomicTask, -} - -#[derive(Debug, Eq, PartialEq, Clone, Copy)] -pub(crate) enum Direction { - Read, - Write, -} - -/// The global fallback reactor. -static HANDLE_FALLBACK: AtomicUsize = ATOMIC_USIZE_INIT; - -/// Tracks the reactor for the current execution context. -thread_local!(static CURRENT_REACTOR: RefCell> = RefCell::new(None)); - -const TOKEN_WAKEUP: mio::Token = mio::Token(0); -const TOKEN_START: usize = 1; - -// Kind of arbitrary, but this reserves some token space for later usage. -const MAX_SOURCES: usize = usize::MAX >> 4; - -fn _assert_kinds() { - fn _assert() {} - - _assert::(); -} - -// ===== impl Reactor ===== - -/// Set the default reactor for the duration of the closure -/// -/// # Panics -/// -/// This function panics if there already is a default reactor set. -pub(crate) fn with_default(handle: &Handle, enter: &mut Enter, f: F) -> R -where F: FnOnce(&mut Enter) -> R -{ - // Ensure that the executor is removed from the thread-local context - // when leaving the scope. This handles cases that involve panicking. - struct Reset; - - impl Drop for Reset { - fn drop(&mut self) { - CURRENT_REACTOR.with(|current| { - let mut current = current.borrow_mut(); - *current = None; - }); - } - } - - // This ensures the value for the current reactor gets reset even if there - // is a panic. - let _r = Reset; - - CURRENT_REACTOR.with(|current| { - { - let mut current = current.borrow_mut(); - assert!(current.is_none(), "default Tokio reactor already set \ - for execution context"); - *current = Some(handle.clone()); - } - - f(enter) - }) -} - -impl Reactor { - /// Creates a new event loop, returning any error that happened during the - /// creation. - pub fn new() -> io::Result { - let io = mio::Poll::new()?; - let wakeup_pair = mio::Registration::new2(); - - io.register(&wakeup_pair.0, - TOKEN_WAKEUP, - mio::Ready::readable(), - mio::PollOpt::level())?; - - Ok(Reactor { - events: mio::Events::with_capacity(1024), - _wakeup_registration: wakeup_pair.0, - inner: Arc::new(Inner { - io: io, - io_dispatch: RwLock::new(Slab::with_capacity(1)), - wakeup: wakeup_pair.1, - }), - }) - } - - /// Returns a handle to this event loop which can be sent across threads - /// and can be used as a proxy to the event loop itself. - /// - /// Handles are cloneable and clones always refer to the same event loop. - /// This handle is typically passed into functions that create I/O objects - /// to bind them to this event loop. - pub fn handle(&self) -> Handle { - Handle { - inner: Arc::downgrade(&self.inner), - } - } - - /// Configures the fallback handle to be returned from `Handle::default`. - /// - /// The `Handle::default()` function will by default lazily spin up a global - /// thread and run a reactor on this global thread. This behavior is not - /// always desirable in all applications, however, and sometimes a different - /// fallback reactor is desired. - /// - /// This function will attempt to globally alter the return value of - /// `Handle::default()` to return the `handle` specified rather than a - /// lazily initialized global thread. If successful then all future calls to - /// `Handle::default()` which would otherwise fall back to the global thread - /// will instead return a clone of the handle specified. - /// - /// # Errors - /// - /// This function may not always succeed in configuring the fallback handle. - /// If this function was previously called (or perhaps concurrently called - /// on many threads) only the *first* invocation of this function will - /// succeed. All other invocations will return an error. - /// - /// Additionally if the global reactor thread has already been initialized - /// then this function will also return an error. (aka if `Handle::default` - /// has been called previously in this program). - pub fn set_fallback(&self) -> Result<(), SetFallbackError> { - set_fallback(self.handle()) - } - - /// Performs one iteration of the event loop, blocking on waiting for events - /// for at most `max_wait` (forever if `None`). - /// - /// This method is the primary method of running this reactor and processing - /// I/O events that occur. This method executes one iteration of an event - /// loop, blocking at most once waiting for events to happen. - /// - /// If a `max_wait` is specified then the method should block no longer than - /// the duration specified, but this shouldn't be used as a super-precise - /// timer but rather a "ballpark approximation" - /// - /// # Return value - /// - /// This function returns an instance of `Turn` - /// - /// `Turn` as of today has no extra information with it and can be safely - /// discarded. In the future `Turn` may contain information about what - /// happened while this reactor blocked. - /// - /// # Errors - /// - /// This function may also return any I/O error which occurs when polling - /// for readiness of I/O objects with the OS. This is quite unlikely to - /// arise and typically mean that things have gone horribly wrong at that - /// point. Currently this is primarily only known to happen for internal - /// bugs to `tokio` itself. - pub fn turn(&mut self, max_wait: Option) -> io::Result { - self.poll(max_wait)?; - Ok(Turn { _priv: () }) - } - - /// Returns true if the reactor is currently idle. - /// - /// Idle is defined as all tasks that have been spawned have completed, - /// either successfully or with an error. - pub fn is_idle(&self) -> bool { - self.inner.io_dispatch - .read().unwrap() - .is_empty() - } - - /// Run the reactor in the background - pub(crate) fn background(self) -> io::Result { - Background::new(self) - } - - fn poll(&mut self, max_wait: Option) -> io::Result<()> { - // Block waiting for an event to happen, peeling out how many events - // happened. - match self.inner.io.poll(&mut self.events, max_wait) { - Ok(_) => {} - Err(ref e) if e.kind() == ErrorKind::Interrupted => return Ok(()), - Err(e) => return Err(e), - } - - let start = if log_enabled!(Level::Debug) { - Some(Instant::now()) - } else { - None - }; - - // Process all the events that came in, dispatching appropriately - let mut events = 0; - for event in self.events.iter() { - events += 1; - let token = event.token(); - trace!("event {:?} {:?}", event.readiness(), event.token()); - - if token == TOKEN_WAKEUP { - self.inner.wakeup.set_readiness(mio::Ready::empty()).unwrap(); - } else { - self.dispatch(token, event.readiness()); - } - } - - if let Some(start) = start { - let dur = start.elapsed(); - debug!("loop process - {} events, {}.{:03}s", - events, - dur.as_secs(), - dur.subsec_nanos() / 1_000_000); - } - - Ok(()) - } - - fn dispatch(&self, token: mio::Token, ready: mio::Ready) { - let token = usize::from(token) - TOKEN_START; - let io_dispatch = self.inner.io_dispatch.read().unwrap(); - - if let Some(io) = io_dispatch.get(token) { - io.readiness.fetch_or(ready2usize(ready), Relaxed); - - if ready.is_writable() { - io.writer.notify(); - } - - if !(ready & (!mio::Ready::writable())).is_empty() { - io.reader.notify(); - } - } - } -} - -impl Park for Reactor { - type Unpark = Handle; - type Error = io::Error; - - fn unpark(&self) -> Self::Unpark { - self.handle() - } - - fn park(&mut self) -> io::Result<()> { - self.turn(None)?; - Ok(()) - } - - fn park_timeout(&mut self, duration: Duration) -> io::Result<()> { - self.turn(Some(duration))?; - Ok(()) - } -} - -impl fmt::Debug for Reactor { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "Reactor") - } -} - -// ===== impl Handle ===== - -impl Handle { - /// Returns a handle to the current reactor. - pub fn current() -> Handle { - Handle::try_current() - .unwrap_or(Handle { inner: Weak::new() }) - } - - /// Try to get a handle to the current reactor. - /// - /// Returns `Err` if no handle is found. - pub(crate) fn try_current() -> io::Result { - CURRENT_REACTOR.with(|current| { - match *current.borrow() { - Some(ref handle) => Ok(handle.clone()), - None => Handle::fallback(), - } - }) - } - - /// Returns a handle to the fallback reactor. - fn fallback() -> io::Result { - let mut fallback = HANDLE_FALLBACK.load(SeqCst); - - // If the fallback hasn't been previously initialized then let's spin - // up a helper thread and try to initialize with that. If we can't - // actually create a helper thread then we'll just return a "defunct" - // handle which will return errors when I/O objects are attempted to be - // associated. - if fallback == 0 { - let reactor = match Reactor::new() { - Ok(reactor) => reactor, - Err(_) => return Err(io::Error::new(io::ErrorKind::Other, - "failed to create reactor")), - }; - - // If we successfully set ourselves as the actual fallback then we - // want to `forget` the helper thread to ensure that it persists - // globally. If we fail to set ourselves as the fallback that means - // that someone was racing with this call to `Handle::default`. - // They ended up winning so we'll destroy our helper thread (which - // shuts down the thread) and reload the fallback. - if set_fallback(reactor.handle().clone()).is_ok() { - let ret = reactor.handle().clone(); - - match reactor.background() { - Ok(bg) => bg.forget(), - // The global handle is fubar, but y'all probably got bigger - // problems if a thread can't spawn. - Err(_) => {} - } - - return Ok(ret); - } - - fallback = HANDLE_FALLBACK.load(SeqCst); - } - - // At this point our fallback handle global was configured so we use - // its value to reify a handle, clone it, and then forget our reified - // handle as we don't actually have an owning reference to it. - assert!(fallback != 0); - - let ret = unsafe { - let handle = Handle::from_usize(fallback); - let ret = handle.clone(); - drop(handle.into_usize()); - ret - }; - - Ok(ret) - } - - /// Forces a reactor blocked in a call to `turn` to wakeup, or otherwise - /// makes the next call to `turn` return immediately. - /// - /// This method is intended to be used in situations where a notification - /// needs to otherwise be sent to the main reactor. If the reactor is - /// currently blocked inside of `turn` then it will wake up and soon return - /// after this method has been called. If the reactor is not currently - /// blocked in `turn`, then the next call to `turn` will not block and - /// return immediately. - fn wakeup(&self) { - if let Some(inner) = self.inner() { - inner.wakeup.set_readiness(mio::Ready::readable()).unwrap(); - } - } - - fn into_usize(self) -> usize { - unsafe { - mem::transmute::, usize>(self.inner) - } - } - - unsafe fn from_usize(val: usize) -> Handle { - let inner = mem::transmute::>(val);; - Handle { inner } - } - - fn inner(&self) -> Option> { - self.inner.upgrade() - } -} - -impl Unpark for Handle { - fn unpark(&self) { - self.wakeup(); - } -} - -impl Default for Handle { - fn default() -> Handle { - Handle::current() - } -} - -impl fmt::Debug for Handle { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "Handle") - } -} - -fn set_fallback(handle: Handle) -> Result<(), SetFallbackError> { - unsafe { - let val = handle.into_usize(); - match HANDLE_FALLBACK.compare_exchange(0, val, SeqCst, SeqCst) { - Ok(_) => Ok(()), - Err(_) => { - drop(Handle::from_usize(val)); - Err(SetFallbackError(())) - } - } - } -} - -// ===== impl Inner ===== - -impl Inner { - /// Register an I/O resource with the reactor. - /// - /// The registration token is returned. - fn add_source(&self, source: &Evented) - -> io::Result - { - let mut io_dispatch = self.io_dispatch.write().unwrap(); - - if io_dispatch.len() == MAX_SOURCES { - return Err(io::Error::new(io::ErrorKind::Other, "reactor at max \ - registered I/O resources")); - } - - // Acquire a write lock - let key = io_dispatch.insert(ScheduledIo { - readiness: AtomicUsize::new(0), - reader: AtomicTask::new(), - writer: AtomicTask::new(), - }); - - try!(self.io.register(source, - mio::Token(TOKEN_START + key), - mio::Ready::readable() | - mio::Ready::writable() | - platform::all(), - mio::PollOpt::edge())); - - Ok(key) - } - - fn deregister_source(&self, source: &Evented) -> io::Result<()> { - self.io.deregister(source) - } - - fn drop_source(&self, token: usize) { - debug!("dropping I/O source: {}", token); - self.io_dispatch.write().unwrap().remove(token); - } - - /// Registers interest in the I/O resource associated with `token`. - fn register(&self, token: usize, dir: Direction, t: Task) { - debug!("scheduling direction for: {}", token); - let io_dispatch = self.io_dispatch.read().unwrap(); - let sched = io_dispatch.get(token).unwrap(); - - let (task, ready) = match dir { - Direction::Read => (&sched.reader, !mio::Ready::writable()), - Direction::Write => (&sched.writer, mio::Ready::writable()), - }; - - task.register_task(t); - - if sched.readiness.load(SeqCst) & ready2usize(ready) != 0 { - task.notify(); - } - } -} - -impl Drop for Inner { - fn drop(&mut self) { - // When a reactor is dropped it needs to wake up all blocked tasks as - // they'll never receive a notification, and all connected I/O objects - // will start returning errors pretty quickly. - let io = self.io_dispatch.read().unwrap(); - for (_, io) in io.iter() { - io.writer.notify(); - io.reader.notify(); - } - } -} - -impl Direction { - fn ready(&self) -> mio::Ready { - match *self { - Direction::Read => read_ready(), - Direction::Write => write_ready(), - } - } - - fn mask(&self) -> usize { - ready2usize(self.ready()) - } -} - -// ===== misc ===== - -const READ: usize = 1 << 0; -const WRITE: usize = 1 << 1; - -fn read_ready() -> mio::Ready { - mio::Ready::readable() | platform::hup() -} - -fn write_ready() -> mio::Ready { - mio::Ready::writable() -} - -// === legacy - -fn ready2usize(ready: mio::Ready) -> usize { - let mut bits = 0; - if ready.is_readable() { - bits |= READ; - } - if ready.is_writable() { - bits |= WRITE; - } - bits | platform::ready2usize(ready) -} - -fn usize2ready(bits: usize) -> mio::Ready { - let mut ready = mio::Ready::empty(); - if bits & READ != 0 { - ready.insert(mio::Ready::readable()); - } - if bits & WRITE != 0 { - ready.insert(mio::Ready::writable()); - } - ready | platform::usize2ready(bits) -} - -#[cfg(all(unix, not(target_os = "fuchsia")))] -mod platform { - use mio::Ready; - use mio::unix::UnixReady; - - #[cfg(target_os = "dragonfly")] - pub fn all() -> Ready { - hup() | UnixReady::aio() - } - - #[cfg(target_os = "freebsd")] - pub fn all() -> Ready { - hup() | UnixReady::aio() | UnixReady::lio() - } - - #[cfg(not(any(target_os = "dragonfly", target_os = "freebsd")))] - pub fn all() -> Ready { - hup() - } - - pub fn hup() -> Ready { - UnixReady::hup().into() - } - - const HUP: usize = 1 << 2; - const ERROR: usize = 1 << 3; - const AIO: usize = 1 << 4; - const LIO: usize = 1 << 5; - - #[cfg(any(target_os = "dragonfly", target_os = "freebsd"))] - fn is_aio(ready: &Ready) -> bool { - UnixReady::from(*ready).is_aio() - } - - #[cfg(not(any(target_os = "dragonfly", target_os = "freebsd")))] - fn is_aio(_ready: &Ready) -> bool { - false - } - - #[cfg(target_os = "freebsd")] - fn is_lio(ready: &Ready) -> bool { - UnixReady::from(*ready).is_lio() - } - - #[cfg(not(target_os = "freebsd"))] - fn is_lio(_ready: &Ready) -> bool { - false - } - - pub fn ready2usize(ready: Ready) -> usize { - let ready = UnixReady::from(ready); - let mut bits = 0; - if is_aio(&ready) { - bits |= AIO; - } - if is_lio(&ready) { - bits |= LIO; - } - if ready.is_error() { - bits |= ERROR; - } - if ready.is_hup() { - bits |= HUP; - } - bits - } - - #[cfg(any(target_os = "dragonfly", target_os = "freebsd", target_os = "ios", - target_os = "macos"))] - fn usize2ready_aio(ready: &mut UnixReady) { - ready.insert(UnixReady::aio()); - } - - #[cfg(not(any(target_os = "dragonfly", - target_os = "freebsd", target_os = "ios", target_os = "macos")))] - fn usize2ready_aio(_ready: &mut UnixReady) { - // aio not available here → empty - } - - #[cfg(target_os = "freebsd")] - fn usize2ready_lio(ready: &mut UnixReady) { - ready.insert(UnixReady::lio()); - } - - #[cfg(not(target_os = "freebsd"))] - fn usize2ready_lio(_ready: &mut UnixReady) { - // lio not available here → empty - } - - pub fn usize2ready(bits: usize) -> Ready { - let mut ready = UnixReady::from(Ready::empty()); - if bits & AIO != 0 { - usize2ready_aio(&mut ready); - } - if bits & LIO != 0 { - usize2ready_lio(&mut ready); - } - if bits & HUP != 0 { - ready.insert(UnixReady::hup()); - } - if bits & ERROR != 0 { - ready.insert(UnixReady::error()); - } - ready.into() - } -} - -#[cfg(any(windows, target_os = "fuchsia"))] -mod platform { - use mio::Ready; - - pub fn all() -> Ready { - // No platform-specific Readinesses for Windows - Ready::empty() - } - - pub fn hup() -> Ready { - Ready::empty() - } - - pub fn ready2usize(_r: Ready) -> usize { - 0 - } - - pub fn usize2ready(_r: usize) -> Ready { - Ready::empty() - } -} diff --git a/src/reactor/poll_evented.rs b/src/reactor/poll_evented.rs index 16f20d491..ab5376781 100644 --- a/src/reactor/poll_evented.rs +++ b/src/reactor/poll_evented.rs @@ -6,75 +6,36 @@ //! acquisition of a token, and tracking of the readiness state on the //! underlying I/O primitive. -#![allow(deprecated)] +#![allow(deprecated, warnings)] use std::fmt; use std::io::{self, Read, Write}; -use std::sync::atomic::Ordering; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering::Relaxed; use futures::{task, Async, Poll}; use mio::event::Evented; use mio::Ready; use tokio_io::{AsyncRead, AsyncWrite}; -use reactor::{Handle, Direction}; +use reactor::{Handle, Registration}; -struct Registration { - pub token: usize, - pub handle: Handle, - pub readiness: usize, -} - -/// A concrete implementation of a stream of readiness notifications for I/O -/// objects that originates from an event loop. -/// -/// Created by the `PollEvented::new` method, each `PollEvented` is -/// associated with a specific event loop and source of events that will be -/// registered with an event loop. -/// -/// An instance of `PollEvented` is essentially the bridge between the `mio` -/// world and the `tokio-core` world, providing abstractions to receive -/// notifications about changes to an object's `mio::Ready` state. -/// -/// Each readiness stream has a number of methods to test whether the underlying -/// object is readable or writable. Once the methods return that an object is -/// readable/writable, then it will continue to do so until the `need_read` or -/// `need_write` methods are called. -/// -/// That is, this object is typically wrapped in another form of I/O object. -/// It's the responsibility of the wrapper to inform the readiness stream when a -/// "would block" I/O event is seen. The readiness stream will then take care of -/// any scheduling necessary to get notified when the event is ready again. -/// -/// You can find more information about creating a custom I/O object [online]. -/// -/// [online]: https://tokio.rs/docs/going-deeper-tokio/core-low-level/#custom-io -/// -/// ## Readiness to read/write -/// -/// A `PollEvented` allows listening and waiting for an arbitrary `mio::Ready` -/// instance, including the platform-specific contents of `mio::Ready`. At most -/// two future tasks, however, can be waiting on a `PollEvented`. The -/// `need_read` and `need_write` methods can block two separate tasks, one on -/// reading and one on writing. Not all I/O events correspond to read/write, -/// however! -/// -/// To account for this a `PollEvented` gets a little interesting when working -/// with an arbitrary instance of `mio::Ready` that may not map precisely to -/// "write" and "read" tasks. Currently it is defined that instances of -/// `mio::Ready` that do *not* return true from `is_writable` are all notified -/// through `need_read`, or the read task. -/// -/// In other words, `poll_ready` with the `mio::UnixReady::hup` event will block -/// the read task of this `PollEvented` if the `hup` event isn't available. -/// Essentially a good rule of thumb is that if you're using the `poll_ready` -/// method you want to also use `need_read` to signal blocking and you should -/// otherwise probably avoid using two tasks on the same `PollEvented`. #[deprecated(since = "0.1.2", note = "PollEvented2 instead")] #[doc(hidden)] pub struct PollEvented { - registration: Registration, io: E, + inner: Inner, + handle: Handle, +} + +struct Inner { + registration: Registration, + + /// Currently visible read readiness + read_readiness: AtomicUsize, + + /// Currently visible write readiness + write_readiness: AtomicUsize, } impl fmt::Debug for PollEvented { @@ -91,20 +52,17 @@ impl PollEvented { pub fn new(io: E, handle: &Handle) -> io::Result> where E: Evented, { - let token = match handle.inner() { - Some(inner) => inner.add_source(&io)?, - None => { - return Err(io::Error::new(io::ErrorKind::Other, "event loop gone")) - } - }; + let registration = Registration::new(); + registration.register(&io)?; Ok(PollEvented { - registration: Registration { - token: token, - readiness: 0, - handle: handle.clone() - }, io: io, + inner: Inner { + registration, + read_readiness: AtomicUsize::new(0), + write_readiness: AtomicUsize::new(0), + }, + handle: handle.clone(), }) } @@ -123,8 +81,37 @@ impl PollEvented { /// This function will panic if called outside the context of a future's /// task. pub fn poll_read(&mut self) -> Async<()> { - self.poll_ready(super::read_ready()) - .map(|_| ()) + if self.poll_read2().is_ready() { + return ().into(); + } + + Async::NotReady + } + + fn poll_read2(&self) -> Async { + // Load the cached readiness + match self.inner.read_readiness.load(Relaxed) { + 0 => {} + mut n => { + // Check what's new with the reactor. + if let Some(ready) = self.inner.registration.take_read_ready().unwrap() { + n |= ready2usize(ready); + self.inner.read_readiness.store(n, Relaxed); + } + + return usize2ready(n).into(); + } + } + + let ready = match self.inner.registration.poll_read_ready().unwrap() { + Async::Ready(r) => r, + _ => return Async::NotReady, + }; + + // Cache the value + self.inner.read_readiness.store(ready2usize(ready), Relaxed); + + ready.into() } /// Tests to see if this source is ready to be written to or not. @@ -142,8 +129,28 @@ impl PollEvented { /// This function will panic if called outside the context of a future's /// task. pub fn poll_write(&mut self) -> Async<()> { - self.poll_ready(Ready::writable()) - .map(|_| ()) + match self.inner.write_readiness.load(Relaxed) { + 0 => {} + mut n => { + // Check what's new with the reactor. + if let Some(ready) = self.inner.registration.take_write_ready().unwrap() { + n |= ready2usize(ready); + self.inner.write_readiness.store(n, Relaxed); + } + + return ().into(); + } + } + + let ready = match self.inner.registration.poll_write_ready().unwrap() { + Async::Ready(r) => r, + _ => return Async::NotReady, + }; + + // Cache the value + self.inner.write_readiness.store(ready2usize(ready), Relaxed); + + ().into() } /// Test to see whether this source fulfills any condition listed in `mask` @@ -170,35 +177,38 @@ impl PollEvented { /// This function will panic if called outside the context of a future's /// task. pub fn poll_ready(&mut self, mask: Ready) -> Async { - let bits = super::ready2usize(mask); + let mut ret = Ready::empty(); - match self.registration.readiness & bits { - 0 => {} - n => return Async::Ready(super::usize2ready(n)), + if mask.is_empty() { + return ret.into(); } - let token_readiness = self.registration.handle.inner().map(|inner| { - let io_dispatch = inner.io_dispatch.read().unwrap(); - let token = self.registration.token; - io_dispatch[token].readiness.swap(0, Ordering::SeqCst) - }).unwrap_or(0); - - self.registration.readiness |= token_readiness; - - match self.registration.readiness & bits { - 0 => { - if mask.is_writable() { - if self.need_write().is_err() { - return Async::Ready(mask) - } - } else { - if self.need_read().is_err() { - return Async::Ready(mask) - } - } - Async::NotReady + if mask.is_writable() { + if self.poll_write().is_ready() { + ret = Ready::writable(); } - n => Async::Ready(super::usize2ready(n)), + } + + let mask = mask - Ready::writable(); + + if !mask.is_empty() { + if let Async::Ready(v) = self.poll_read2() { + ret |= v & mask; + } + } + + if ret.is_empty() { + if mask.is_writable() { + let _ = self.need_write(); + } + + if mask.is_readable() { + let _ = self.need_read(); + } + + Async::NotReady + } else { + ret.into() } } @@ -234,10 +244,14 @@ impl PollEvented { /// This function will panic if called outside the context of a future's /// task. pub fn need_read(&mut self) -> io::Result<()> { - let bits = super::ready2usize(super::read_ready()); - self.registration.readiness &= !bits; + self.inner.read_readiness.store(0, Relaxed); - self.register(Direction::Read) + if self.poll_read().is_ready() { + // Notify the current task + task::current().notify(); + } + + Ok(()) } /// Indicates to this source of events that the corresponding I/O object is @@ -269,16 +283,20 @@ impl PollEvented { /// This function will panic if called outside the context of a future's /// task. pub fn need_write(&mut self) -> io::Result<()> { - let bits = super::ready2usize(Ready::writable()); - self.registration.readiness &= !bits; + self.inner.write_readiness.store(0, Relaxed); - self.register(Direction::Write) + if self.poll_write().is_ready() { + // Notify the current task + task::current().notify(); + } + + Ok(()) } /// Returns a reference to the event loop handle that this readiness stream /// is associated with. pub fn handle(&self) -> &Handle { - &self.registration.handle + &self.handle } /// Returns a shared reference to the underlying I/O object this readiness @@ -313,21 +331,8 @@ impl PollEvented { pub fn deregister(&self) -> io::Result<()> where E: Evented, { - let inner = match self.handle().inner() { - Some(inner) => inner, - None => return Ok(()), - }; - - inner.deregister_source(&self.io) - } - - fn register(&self, dir: Direction) -> io::Result<()> { - let inner = match self.registration.handle.inner() { - Some(inner) => inner, - None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")), - }; - - inner.register(self.registration.token, dir, task::current()); + // Nothing has to happen here anymore as I/O objects are explicitly + // deregistered before dropped. Ok(()) } } @@ -394,10 +399,147 @@ fn is_wouldblock(r: &io::Result) -> bool { } } -impl Drop for Registration { - fn drop(&mut self) { - if let Some(inner) = self.handle.inner() { - inner.drop_source(self.token); +const READ: usize = 1 << 0; +const WRITE: usize = 1 << 1; + +fn ready2usize(ready: Ready) -> usize { + let mut bits = 0; + if ready.is_readable() { + bits |= READ; + } + if ready.is_writable() { + bits |= WRITE; + } + bits | platform::ready2usize(ready) +} + +fn usize2ready(bits: usize) -> Ready { + let mut ready = Ready::empty(); + if bits & READ != 0 { + ready.insert(Ready::readable()); + } + if bits & WRITE != 0 { + ready.insert(Ready::writable()); + } + ready | platform::usize2ready(bits) +} + +#[cfg(all(unix, not(target_os = "fuchsia")))] +mod platform { + use mio::Ready; + use mio::unix::UnixReady; + + #[cfg(target_os = "dragonfly")] + pub fn all() -> Ready { + hup() | UnixReady::aio() + } + + #[cfg(target_os = "freebsd")] + pub fn all() -> Ready { + hup() | UnixReady::aio() | UnixReady::lio() + } + + const HUP: usize = 1 << 2; + const ERROR: usize = 1 << 3; + const AIO: usize = 1 << 4; + const LIO: usize = 1 << 5; + + #[cfg(any(target_os = "dragonfly", target_os = "freebsd"))] + fn is_aio(ready: &Ready) -> bool { + UnixReady::from(*ready).is_aio() + } + + #[cfg(not(any(target_os = "dragonfly", target_os = "freebsd")))] + fn is_aio(_ready: &Ready) -> bool { + false + } + + #[cfg(target_os = "freebsd")] + fn is_lio(ready: &Ready) -> bool { + UnixReady::from(*ready).is_lio() + } + + #[cfg(not(target_os = "freebsd"))] + fn is_lio(_ready: &Ready) -> bool { + false + } + + pub fn ready2usize(ready: Ready) -> usize { + let ready = UnixReady::from(ready); + let mut bits = 0; + if is_aio(&ready) { + bits |= AIO; } + if is_lio(&ready) { + bits |= LIO; + } + if ready.is_error() { + bits |= ERROR; + } + if ready.is_hup() { + bits |= HUP; + } + bits + } + + #[cfg(any(target_os = "dragonfly", target_os = "freebsd", target_os = "ios", + target_os = "macos"))] + fn usize2ready_aio(ready: &mut UnixReady) { + ready.insert(UnixReady::aio()); + } + + #[cfg(not(any(target_os = "dragonfly", + target_os = "freebsd", target_os = "ios", target_os = "macos")))] + fn usize2ready_aio(_ready: &mut UnixReady) { + // aio not available here → empty + } + + #[cfg(target_os = "freebsd")] + fn usize2ready_lio(ready: &mut UnixReady) { + ready.insert(UnixReady::lio()); + } + + #[cfg(not(target_os = "freebsd"))] + fn usize2ready_lio(_ready: &mut UnixReady) { + // lio not available here → empty + } + + pub fn usize2ready(bits: usize) -> Ready { + let mut ready = UnixReady::from(Ready::empty()); + if bits & AIO != 0 { + usize2ready_aio(&mut ready); + } + if bits & LIO != 0 { + usize2ready_lio(&mut ready); + } + if bits & HUP != 0 { + ready.insert(UnixReady::hup()); + } + if bits & ERROR != 0 { + ready.insert(UnixReady::error()); + } + ready.into() + } +} + +#[cfg(any(windows, target_os = "fuchsia"))] +mod platform { + use mio::Ready; + + pub fn all() -> Ready { + // No platform-specific Readinesses for Windows + Ready::empty() + } + + pub fn hup() -> Ready { + Ready::empty() + } + + pub fn ready2usize(_r: Ready) -> usize { + 0 + } + + pub fn usize2ready(_r: usize) -> Ready { + Ready::empty() } } diff --git a/src/runtime.rs b/src/runtime.rs index 8fb22f61b..efee51d7e 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -104,8 +104,7 @@ //! [idle]: struct.Runtime.html#method.shutdown_on_idle //! [`tokio::spawn`]: ../executor/fn.spawn.html -use reactor::{self, Reactor, Handle}; -use reactor::background::Background; +use reactor::{Reactor, Handle, Background}; use tokio_threadpool::{self as threadpool, ThreadPool, Sender}; use futures::Poll; @@ -221,7 +220,7 @@ impl Runtime { let pool = threadpool::Builder::new() .around_worker(move |w, enter| { - reactor::with_default(&handle, enter, |_| { + ::tokio_reactor::with_default(&handle, enter, |_| { w.run(); }); }) diff --git a/tokio-executor/README.md b/tokio-executor/README.md new file mode 100644 index 000000000..e127c773e --- /dev/null +++ b/tokio-executor/README.md @@ -0,0 +1,54 @@ +# tokio-executor + +Task execution related traits and utilities. + +[Documentation](https://tokio-rs.github.io/tokio/tokio_executor/) + +## Overview + +In the Tokio execution model, futures are lazy. When a future is created, no +work is performed. In order for the work defined by the future to happen, the +future must be submitted to an executor. A future that is submitted to an +executor is called a "task". + +The executor is responsible for ensuring that [`Future::poll`] is called +whenever the task is [notified]. Notification happens when the internal state of +a task transitions from "not ready" to ready. For example, a socket might have +received data and a call to `read` will now be able to succeed. + +This crate provides traits and utilities that are necessary for building an +executor, including: + +* The [`Executor`] trait describes the API for spawning a future onto an + executor. + +* [`enter`] marks that the the current thread is entering an execution + context. This prevents a second executor from accidentally starting from + within the context of one that is already running. + +* [`DefaultExecutor`] spawns tasks onto the default executor for the current + context. + +* [`Park`] abstracts over blocking and unblocking the current thread. + +[`Executor`]: https://tokio-rs.github.io/tokio/tokio_executor/trait.Executor.html +[`enter`]: https://tokio-rs.github.io/tokio/tokio_executor/fn.enter.html +[`DefaultExecutor`]: https://tokio-rs.github.io/tokio/tokio_executor/struct.DefaultExecutor.html +[`Park`]: https://tokio-rs.github.io/tokio/tokio_executor/park/index.html + +## License + +This project is licensed under either of + + * Apache License, Version 2.0, ([LICENSE-APACHE](../LICENSE-APACHE) or + http://www.apache.org/licenses/LICENSE-2.0) + * MIT license ([LICENSE-MIT](../LICENSE-MIT) or + http://opensource.org/licenses/MIT) + +at your option. + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, as defined in the Apache-2.0 license, shall be +dual licensed as above, without any additional terms or conditions. diff --git a/tokio-executor/src/lib.rs b/tokio-executor/src/lib.rs index 299f197d5..14f8ea301 100644 --- a/tokio-executor/src/lib.rs +++ b/tokio-executor/src/lib.rs @@ -1,15 +1,34 @@ -//! Task execution utilities. +//! Task execution related traits and utilities. //! //! In the Tokio execution model, futures are lazy. When a future is created, no //! work is performed. In order for the work defined by the future to happen, //! the future must be submitted to an executor. A future that is submitted to //! an executor is called a "task". //! -//! The executor executor is responsible for ensuring that [`Future::poll`] is -//! called whenever the task is [notified]. Notification happens when the -//! internal state of a task transitions from "not ready" to ready. For -//! example, a socket might have received data and a call to `read` will now be -//! able to succeed. +//! The executor is responsible for ensuring that [`Future::poll`] is called +//! whenever the task is [notified]. Notification happens when the internal +//! state of a task transitions from "not ready" to ready. For example, a socket +//! might have received data and a call to `read` will now be able to succeed. +//! +//! This crate provides traits and utilities that are necessary for building an +//! executor, including: +//! +//! * The [`Executor`] trait describes the API for spawning a future onto an +//! executor. +//! +//! * [`enter`] marks that the the current thread is entering an execution +//! context. This prevents a second executor from accidentally starting from +//! within the context of one that is already running. +//! +//! * [`DefaultExecutor`] spawns tasks onto the default executor for the current +//! context. +//! +//! * [`Park`] abstracts over blocking and unblocking the current thread. +//! +//! [`Executor`]: trait.Executor.html +//! [`enter`]: fn.enter.html +//! [`DefaultExecutor`]: struct.DefaultExecutor.html +//! [`Park`]: park/index.html #![deny(missing_docs, missing_debug_implementations, warnings)] #![doc(html_root_url = "https://docs.rs/tokio-executor/0.1")] diff --git a/tokio-io/LICENSE-APACHE b/tokio-io/LICENSE-APACHE deleted file mode 100644 index 16fe87b06..000000000 --- a/tokio-io/LICENSE-APACHE +++ /dev/null @@ -1,201 +0,0 @@ - Apache License - Version 2.0, January 2004 - http://www.apache.org/licenses/ - -TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION - -1. Definitions. - - "License" shall mean the terms and conditions for use, reproduction, - and distribution as defined by Sections 1 through 9 of this document. - - "Licensor" shall mean the copyright owner or entity authorized by - the copyright owner that is granting the License. - - "Legal Entity" shall mean the union of the acting entity and all - other entities that control, are controlled by, or are under common - control with that entity. For the purposes of this definition, - "control" means (i) the power, direct or indirect, to cause the - direction or management of such entity, whether by contract or - otherwise, or (ii) ownership of fifty percent (50%) or more of the - outstanding shares, or (iii) beneficial ownership of such entity. - - "You" (or "Your") shall mean an individual or Legal Entity - exercising permissions granted by this License. - - "Source" form shall mean the preferred form for making modifications, - including but not limited to software source code, documentation - source, and configuration files. - - "Object" form shall mean any form resulting from mechanical - transformation or translation of a Source form, including but - not limited to compiled object code, generated documentation, - and conversions to other media types. - - "Work" shall mean the work of authorship, whether in Source or - Object form, made available under the License, as indicated by a - copyright notice that is included in or attached to the work - (an example is provided in the Appendix below). - - "Derivative Works" shall mean any work, whether in Source or Object - form, that is based on (or derived from) the Work and for which the - editorial revisions, annotations, elaborations, or other modifications - represent, as a whole, an original work of authorship. For the purposes - of this License, Derivative Works shall not include works that remain - separable from, or merely link (or bind by name) to the interfaces of, - the Work and Derivative Works thereof. - - "Contribution" shall mean any work of authorship, including - the original version of the Work and any modifications or additions - to that Work or Derivative Works thereof, that is intentionally - submitted to Licensor for inclusion in the Work by the copyright owner - or by an individual or Legal Entity authorized to submit on behalf of - the copyright owner. For the purposes of this definition, "submitted" - means any form of electronic, verbal, or written communication sent - to the Licensor or its representatives, including but not limited to - communication on electronic mailing lists, source code control systems, - and issue tracking systems that are managed by, or on behalf of, the - Licensor for the purpose of discussing and improving the Work, but - excluding communication that is conspicuously marked or otherwise - designated in writing by the copyright owner as "Not a Contribution." - - "Contributor" shall mean Licensor and any individual or Legal Entity - on behalf of whom a Contribution has been received by Licensor and - subsequently incorporated within the Work. - -2. Grant of Copyright License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - copyright license to reproduce, prepare Derivative Works of, - publicly display, publicly perform, sublicense, and distribute the - Work and such Derivative Works in Source or Object form. - -3. Grant of Patent License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - (except as stated in this section) patent license to make, have made, - use, offer to sell, sell, import, and otherwise transfer the Work, - where such license applies only to those patent claims licensable - by such Contributor that are necessarily infringed by their - Contribution(s) alone or by combination of their Contribution(s) - with the Work to which such Contribution(s) was submitted. If You - institute patent litigation against any entity (including a - cross-claim or counterclaim in a lawsuit) alleging that the Work - or a Contribution incorporated within the Work constitutes direct - or contributory patent infringement, then any patent licenses - granted to You under this License for that Work shall terminate - as of the date such litigation is filed. - -4. Redistribution. You may reproduce and distribute copies of the - Work or Derivative Works thereof in any medium, with or without - modifications, and in Source or Object form, provided that You - meet the following conditions: - - (a) You must give any other recipients of the Work or - Derivative Works a copy of this License; and - - (b) You must cause any modified files to carry prominent notices - stating that You changed the files; and - - (c) You must retain, in the Source form of any Derivative Works - that You distribute, all copyright, patent, trademark, and - attribution notices from the Source form of the Work, - excluding those notices that do not pertain to any part of - the Derivative Works; and - - (d) If the Work includes a "NOTICE" text file as part of its - distribution, then any Derivative Works that You distribute must - include a readable copy of the attribution notices contained - within such NOTICE file, excluding those notices that do not - pertain to any part of the Derivative Works, in at least one - of the following places: within a NOTICE text file distributed - as part of the Derivative Works; within the Source form or - documentation, if provided along with the Derivative Works; or, - within a display generated by the Derivative Works, if and - wherever such third-party notices normally appear. The contents - of the NOTICE file are for informational purposes only and - do not modify the License. You may add Your own attribution - notices within Derivative Works that You distribute, alongside - or as an addendum to the NOTICE text from the Work, provided - that such additional attribution notices cannot be construed - as modifying the License. - - You may add Your own copyright statement to Your modifications and - may provide additional or different license terms and conditions - for use, reproduction, or distribution of Your modifications, or - for any such Derivative Works as a whole, provided Your use, - reproduction, and distribution of the Work otherwise complies with - the conditions stated in this License. - -5. Submission of Contributions. Unless You explicitly state otherwise, - any Contribution intentionally submitted for inclusion in the Work - by You to the Licensor shall be under the terms and conditions of - this License, without any additional terms or conditions. - Notwithstanding the above, nothing herein shall supersede or modify - the terms of any separate license agreement you may have executed - with Licensor regarding such Contributions. - -6. Trademarks. This License does not grant permission to use the trade - names, trademarks, service marks, or product names of the Licensor, - except as required for reasonable and customary use in describing the - origin of the Work and reproducing the content of the NOTICE file. - -7. Disclaimer of Warranty. Unless required by applicable law or - agreed to in writing, Licensor provides the Work (and each - Contributor provides its Contributions) on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - implied, including, without limitation, any warranties or conditions - of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A - PARTICULAR PURPOSE. You are solely responsible for determining the - appropriateness of using or redistributing the Work and assume any - risks associated with Your exercise of permissions under this License. - -8. Limitation of Liability. In no event and under no legal theory, - whether in tort (including negligence), contract, or otherwise, - unless required by applicable law (such as deliberate and grossly - negligent acts) or agreed to in writing, shall any Contributor be - liable to You for damages, including any direct, indirect, special, - incidental, or consequential damages of any character arising as a - result of this License or out of the use or inability to use the - Work (including but not limited to damages for loss of goodwill, - work stoppage, computer failure or malfunction, or any and all - other commercial damages or losses), even if such Contributor - has been advised of the possibility of such damages. - -9. Accepting Warranty or Additional Liability. While redistributing - the Work or Derivative Works thereof, You may choose to offer, - and charge a fee for, acceptance of support, warranty, indemnity, - or other liability obligations and/or rights consistent with this - License. However, in accepting such obligations, You may act only - on Your own behalf and on Your sole responsibility, not on behalf - of any other Contributor, and only if You agree to indemnify, - defend, and hold each Contributor harmless for any liability - incurred by, or claims asserted against, such Contributor by reason - of your accepting any such warranty or additional liability. - -END OF TERMS AND CONDITIONS - -APPENDIX: How to apply the Apache License to your work. - - To apply the Apache License to your work, attach the following - boilerplate notice, with the fields enclosed by brackets "[]" - replaced with your own identifying information. (Don't include - the brackets!) The text should be enclosed in the appropriate - comment syntax for the file format. We also recommend that a - file or class name and description of purpose be included on the - same "printed page" as the copyright notice for easier - identification within third-party archives. - -Copyright [yyyy] [name of copyright owner] - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. diff --git a/tokio-io/LICENSE-MIT b/tokio-io/LICENSE-MIT deleted file mode 100644 index 2a7bdc066..000000000 --- a/tokio-io/LICENSE-MIT +++ /dev/null @@ -1,25 +0,0 @@ -Copyright (c) 2017 Tokio Authors - -Permission is hereby granted, free of charge, to any -person obtaining a copy of this software and associated -documentation files (the "Software"), to deal in the -Software without restriction, including without -limitation the rights to use, copy, modify, merge, -publish, distribute, sublicense, and/or sell copies of -the Software, and to permit persons to whom the Software -is furnished to do so, subject to the following -conditions: - -The above copyright notice and this permission notice -shall be included in all copies or substantial portions -of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF -ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED -TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A -PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT -SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY -CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION -OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR -IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER -DEALINGS IN THE SOFTWARE. diff --git a/tokio-io/README.md b/tokio-io/README.md index 68e05406d..d1238a57f 100644 --- a/tokio-io/README.md +++ b/tokio-io/README.md @@ -26,14 +26,13 @@ online at [https://tokio.rs](https://tokio.rs). The [API documentation](https://docs.rs/tokio-io) is also a great place to get started for the nitty-gritty. - # License This project is licensed under either of - * Apache License, Version 2.0, ([LICENSE-APACHE](LICENSE-APACHE) or + * Apache License, Version 2.0, ([LICENSE-APACHE](../LICENSE-APACHE) or http://www.apache.org/licenses/LICENSE-2.0) - * MIT license ([LICENSE-MIT](LICENSE-MIT) or + * MIT license ([LICENSE-MIT](../LICENSE-MIT) or http://opensource.org/licenses/MIT) at your option. diff --git a/tokio-reactor/Cargo.toml b/tokio-reactor/Cargo.toml new file mode 100644 index 000000000..366493be8 --- /dev/null +++ b/tokio-reactor/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "tokio-reactor" + +# When releasing to crates.io: +# - Update html_root_url. +# - Update CHANGELOG.md. +# - Create "v0.1.x" git tag. +version = "0.1.0" +authors = ["Carl Lerche "] +license = "MIT/Apache-2.0" +readme = "README.md" +repository = "https://github.com/tokio-rs/tokio" +homepage = "https://tokio.rs" +documentation = "https://docs.rs/tokio-reactor/0.1" +description = """ +Event loop that drives Tokio I/O resources. +""" +categories = ["asynchronous", "network-programming"] + +[dependencies] +futures = "0.1.18" +log = "0.4.1" +mio = "0.6.13" +slab = "0.4.0" +tokio-executor = { version = "0.1", path = "../tokio-executor" } +tokio-io = { version = "0.1", path = "../tokio-io" } diff --git a/tokio-reactor/README.md b/tokio-reactor/README.md new file mode 100644 index 000000000..c2831af1f --- /dev/null +++ b/tokio-reactor/README.md @@ -0,0 +1,49 @@ +# tokio-reactor + +Event loop that drives Tokio I/O resources. + +[Documentation](https://tokio-rs.github.io/tokio/tokio_reactor/) + +## Overview + +The reactor is the engine that drives asynchronous I/O resources (like TCP and +UDP sockets). It is backed by [`mio`] and acts as a bridge between [`mio`] and +[`futures`]. + +The crate provides: + +* [`Reactor`] is the main type of this crate. It performs the event loop logic. + +* [`Handle`] provides a reference to a reactor instance. + +* [`Registration`] and [`PollEvented`] allow third parties to implement I/O + resources that are driven by the reactor. + +Application authors will not use this crate directly. Instead, they will use the +[`tokio`] crate. Library authors should only depend on `tokio-reactor` if they +are building a custom I/O resource. + +[`mio`]: http://github.com/carllerche/mio +[`futures`]: http://github.com/rust-lang-nursery/futures-rs +[`Reactor`]: https://tokio-rs.github.io/tokio/tokio_reactor/struct.Reactor.html +[`Handle`]: https://tokio-rs.github.io/tokio/tokio_reactor/struct.Handle.html +[`Registration`]: https://tokio-rs.github.io/tokio/tokio_reactor/struct.Registration.html +[`PollEvented`]: https://tokio-rs.github.io/tokio/tokio_reactor/struct.PollEvented.html +[`tokio`]: ../ + +## License + +This project is licensed under either of + + * Apache License, Version 2.0, ([LICENSE-APACHE](../LICENSE-APACHE) or + http://www.apache.org/licenses/LICENSE-2.0) + * MIT license ([LICENSE-MIT](../LICENSE-MIT) or + http://opensource.org/licenses/MIT) + +at your option. + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, as defined in the Apache-2.0 license, shall be +dual licensed as above, without any additional terms or conditions. diff --git a/src/atomic_task.rs b/tokio-reactor/src/atomic_task.rs similarity index 100% rename from src/atomic_task.rs rename to tokio-reactor/src/atomic_task.rs diff --git a/src/reactor/background.rs b/tokio-reactor/src/background.rs similarity index 95% rename from src/reactor/background.rs rename to tokio-reactor/src/background.rs index f0836ceb8..1f07c7b89 100644 --- a/src/reactor/background.rs +++ b/tokio-reactor/src/background.rs @@ -1,15 +1,19 @@ +use {Reactor, Handle}; +use atomic_task::AtomicTask; + +use futures::{Future, Async, Poll}; + use std::io; use std::thread; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::SeqCst; -use atomic_task::AtomicTask; - -use reactor::{Reactor, Handle}; -use futures::{Future, Async, Poll}; - /// Handle to the reactor running on a background thread. +/// +/// Instances are created by calling [`Reactor::background`]. +/// +/// [`Reactor::background`]: struct.Reactor.html#method.background #[derive(Debug)] pub struct Background { /// When `None`, the reactor thread will run until the process terminates. @@ -54,7 +58,7 @@ const SHUTDOWN: usize = 3; impl Background { /// Launch a reactor in the background and return a handle to the thread. - pub fn new(reactor: Reactor) -> io::Result { + pub(crate) fn new(reactor: Reactor) -> io::Result { // Grab a handle to the reactor let handle = reactor.handle().clone(); diff --git a/tokio-reactor/src/lib.rs b/tokio-reactor/src/lib.rs new file mode 100644 index 000000000..1934015b3 --- /dev/null +++ b/tokio-reactor/src/lib.rs @@ -0,0 +1,776 @@ +//! Event loop that drives Tokio I/O resources. +//! +//! The reactor is the engine that drives asynchronous I/O resources (like TCP and +//! UDP sockets). It is backed by [`mio`] and acts as a bridge between [`mio`] and +//! [`futures`]. +//! +//! The crate provides: +//! +//! * [`Reactor`] is the main type of this crate. It performs the event loop logic. +//! +//! * [`Handle`] provides a reference to a reactor instance. +//! +//! * [`Registration`] and [`PollEvented`] allow third parties to implement I/O +//! resources that are driven by the reactor. +//! +//! Application authors will not use this crate directly. Instead, they will use the +//! `tokio` crate. Library authors should only depend on `tokio-reactor` if they +//! are building a custom I/O resource. +//! +//! For more details, see [reactor module] documentation in the Tokio crate. +//! +//! [`mio`]: http://github.com/carllerche/mio +//! [`futures`]: http://github.com/rust-lang-nursery/futures-rs +//! [`Reactor`]: struct.Reactor.html +//! [`Handle`]: struct.Handle.html +//! [`Registration`]: struct.Registration.html +//! [`PollEvented`]: struct.PollEvented.html +//! [reactor module]: https://docs.rs/tokio/0.1/tokio/reactor/index.html + +#![doc(html_root_url = "https://docs.rs/tokio-reactor/0.1.0")] +#![deny(missing_docs, warnings, missing_debug_implementations)] + +#[macro_use] +extern crate futures; +#[macro_use] +extern crate log; +extern crate mio; +extern crate slab; +extern crate tokio_executor; +extern crate tokio_io; + +pub(crate) mod background; +mod atomic_task; +mod poll_evented; +mod registration; + +// ===== Public re-exports ===== + +pub use self::background::Background; +pub use self::registration::Registration; +pub use self::poll_evented::PollEvented; + +// ===== Private imports ===== + +use atomic_task::AtomicTask; + +use tokio_executor::Enter; +use tokio_executor::park::{Park, Unpark}; + +use std::{fmt, usize}; +use std::io::{self, ErrorKind}; +use std::mem; +use std::cell::RefCell; +use std::sync::atomic::Ordering::{Relaxed, SeqCst}; +use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT}; +use std::sync::{Arc, Weak, RwLock}; +use std::time::{Duration, Instant}; + +use log::Level; +use mio::event::Evented; +use slab::Slab; +use futures::task::Task; + +/// The core reactor, or event loop. +/// +/// The event loop is the main source of blocking in an application which drives +/// all other I/O events and notifications happening. Each event loop can have +/// multiple handles pointing to it, each of which can then be used to create +/// various I/O objects to interact with the event loop in interesting ways. +pub struct Reactor { + /// Reuse the `mio::Events` value across calls to poll. + events: mio::Events, + + /// State shared between the reactor and the handles. + inner: Arc, + + _wakeup_registration: mio::Registration, +} + +/// A reference to a reactor. +/// +/// A `Handle` is used for associating I/O objects with an event loop +/// explicitly. Typically though you won't end up using a `Handle` that often +/// and will instead use the default reactor for the execution context. +#[derive(Clone)] +pub struct Handle { + inner: Weak, +} + +/// Return value from the `turn` method on `Reactor`. +/// +/// Currently this value doesn't actually provide any functionality, but it may +/// in the future give insight into what happened during `turn`. +#[derive(Debug)] +pub struct Turn { + _priv: (), +} + +/// Error returned from `Handle::set_fallback`. +#[derive(Clone, Debug)] +pub struct SetFallbackError(()); + +#[deprecated(since = "0.1.2", note = "use SetFallbackError instead")] +#[doc(hidden)] +pub type SetDefaultError = SetFallbackError; + +struct Inner { + /// The underlying system event queue. + io: mio::Poll, + + /// Dispatch slabs for I/O and futures events + io_dispatch: RwLock>, + + /// Used to wake up the reactor from a call to `turn` + wakeup: mio::SetReadiness +} + +struct ScheduledIo { + readiness: AtomicUsize, + reader: AtomicTask, + writer: AtomicTask, +} + +#[derive(Debug, Eq, PartialEq, Clone, Copy)] +pub(crate) enum Direction { + Read, + Write, +} + +/// The global fallback reactor. +static HANDLE_FALLBACK: AtomicUsize = ATOMIC_USIZE_INIT; + +/// Tracks the reactor for the current execution context. +thread_local!(static CURRENT_REACTOR: RefCell> = RefCell::new(None)); + +const TOKEN_WAKEUP: mio::Token = mio::Token(0); +const TOKEN_START: usize = 1; + +// Kind of arbitrary, but this reserves some token space for later usage. +const MAX_SOURCES: usize = usize::MAX >> 4; + +fn _assert_kinds() { + fn _assert() {} + + _assert::(); +} + +// ===== impl Reactor ===== + +/// Set the default reactor for the duration of the closure +/// +/// # Panics +/// +/// This function panics if there already is a default reactor set. +pub fn with_default(handle: &Handle, enter: &mut Enter, f: F) -> R +where F: FnOnce(&mut Enter) -> R +{ + // Ensure that the executor is removed from the thread-local context + // when leaving the scope. This handles cases that involve panicking. + struct Reset; + + impl Drop for Reset { + fn drop(&mut self) { + CURRENT_REACTOR.with(|current| { + let mut current = current.borrow_mut(); + *current = None; + }); + } + } + + // This ensures the value for the current reactor gets reset even if there + // is a panic. + let _r = Reset; + + CURRENT_REACTOR.with(|current| { + { + let mut current = current.borrow_mut(); + assert!(current.is_none(), "default Tokio reactor already set \ + for execution context"); + *current = Some(handle.clone()); + } + + f(enter) + }) +} + +impl Reactor { + /// Creates a new event loop, returning any error that happened during the + /// creation. + pub fn new() -> io::Result { + let io = mio::Poll::new()?; + let wakeup_pair = mio::Registration::new2(); + + io.register(&wakeup_pair.0, + TOKEN_WAKEUP, + mio::Ready::readable(), + mio::PollOpt::level())?; + + Ok(Reactor { + events: mio::Events::with_capacity(1024), + _wakeup_registration: wakeup_pair.0, + inner: Arc::new(Inner { + io: io, + io_dispatch: RwLock::new(Slab::with_capacity(1)), + wakeup: wakeup_pair.1, + }), + }) + } + + /// Returns a handle to this event loop which can be sent across threads + /// and can be used as a proxy to the event loop itself. + /// + /// Handles are cloneable and clones always refer to the same event loop. + /// This handle is typically passed into functions that create I/O objects + /// to bind them to this event loop. + pub fn handle(&self) -> Handle { + Handle { + inner: Arc::downgrade(&self.inner), + } + } + + /// Configures the fallback handle to be returned from `Handle::default`. + /// + /// The `Handle::default()` function will by default lazily spin up a global + /// thread and run a reactor on this global thread. This behavior is not + /// always desirable in all applications, however, and sometimes a different + /// fallback reactor is desired. + /// + /// This function will attempt to globally alter the return value of + /// `Handle::default()` to return the `handle` specified rather than a + /// lazily initialized global thread. If successful then all future calls to + /// `Handle::default()` which would otherwise fall back to the global thread + /// will instead return a clone of the handle specified. + /// + /// # Errors + /// + /// This function may not always succeed in configuring the fallback handle. + /// If this function was previously called (or perhaps concurrently called + /// on many threads) only the *first* invocation of this function will + /// succeed. All other invocations will return an error. + /// + /// Additionally if the global reactor thread has already been initialized + /// then this function will also return an error. (aka if `Handle::default` + /// has been called previously in this program). + pub fn set_fallback(&self) -> Result<(), SetFallbackError> { + set_fallback(self.handle()) + } + + /// Performs one iteration of the event loop, blocking on waiting for events + /// for at most `max_wait` (forever if `None`). + /// + /// This method is the primary method of running this reactor and processing + /// I/O events that occur. This method executes one iteration of an event + /// loop, blocking at most once waiting for events to happen. + /// + /// If a `max_wait` is specified then the method should block no longer than + /// the duration specified, but this shouldn't be used as a super-precise + /// timer but rather a "ballpark approximation" + /// + /// # Return value + /// + /// This function returns an instance of `Turn` + /// + /// `Turn` as of today has no extra information with it and can be safely + /// discarded. In the future `Turn` may contain information about what + /// happened while this reactor blocked. + /// + /// # Errors + /// + /// This function may also return any I/O error which occurs when polling + /// for readiness of I/O objects with the OS. This is quite unlikely to + /// arise and typically mean that things have gone horribly wrong at that + /// point. Currently this is primarily only known to happen for internal + /// bugs to `tokio` itself. + pub fn turn(&mut self, max_wait: Option) -> io::Result { + self.poll(max_wait)?; + Ok(Turn { _priv: () }) + } + + /// Returns true if the reactor is currently idle. + /// + /// Idle is defined as all tasks that have been spawned have completed, + /// either successfully or with an error. + pub fn is_idle(&self) -> bool { + self.inner.io_dispatch + .read().unwrap() + .is_empty() + } + + /// Run this reactor on a background thread. + /// + /// This function takes ownership, spawns a new thread, and moves the + /// reactor to this new thread. It then runs the reactor, driving all + /// associated I/O resources, until the `Background` handle is dropped or + /// explicitly shutdown. + pub fn background(self) -> io::Result { + Background::new(self) + } + + fn poll(&mut self, max_wait: Option) -> io::Result<()> { + // Block waiting for an event to happen, peeling out how many events + // happened. + match self.inner.io.poll(&mut self.events, max_wait) { + Ok(_) => {} + Err(ref e) if e.kind() == ErrorKind::Interrupted => return Ok(()), + Err(e) => return Err(e), + } + + let start = if log_enabled!(Level::Debug) { + Some(Instant::now()) + } else { + None + }; + + // Process all the events that came in, dispatching appropriately + let mut events = 0; + for event in self.events.iter() { + events += 1; + let token = event.token(); + trace!("event {:?} {:?}", event.readiness(), event.token()); + + if token == TOKEN_WAKEUP { + self.inner.wakeup.set_readiness(mio::Ready::empty()).unwrap(); + } else { + self.dispatch(token, event.readiness()); + } + } + + if let Some(start) = start { + let dur = start.elapsed(); + debug!("loop process - {} events, {}.{:03}s", + events, + dur.as_secs(), + dur.subsec_nanos() / 1_000_000); + } + + Ok(()) + } + + fn dispatch(&self, token: mio::Token, ready: mio::Ready) { + let token = usize::from(token) - TOKEN_START; + let io_dispatch = self.inner.io_dispatch.read().unwrap(); + + if let Some(io) = io_dispatch.get(token) { + io.readiness.fetch_or(ready2usize(ready), Relaxed); + + if ready.is_writable() { + io.writer.notify(); + } + + if !(ready & (!mio::Ready::writable())).is_empty() { + io.reader.notify(); + } + } + } +} + +impl Park for Reactor { + type Unpark = Handle; + type Error = io::Error; + + fn unpark(&self) -> Self::Unpark { + self.handle() + } + + fn park(&mut self) -> io::Result<()> { + self.turn(None)?; + Ok(()) + } + + fn park_timeout(&mut self, duration: Duration) -> io::Result<()> { + self.turn(Some(duration))?; + Ok(()) + } +} + +impl fmt::Debug for Reactor { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "Reactor") + } +} + +// ===== impl Handle ===== + +impl Handle { + /// Returns a handle to the current reactor. + pub fn current() -> Handle { + Handle::try_current() + .unwrap_or(Handle { inner: Weak::new() }) + } + + /// Try to get a handle to the current reactor. + /// + /// Returns `Err` if no handle is found. + pub(crate) fn try_current() -> io::Result { + CURRENT_REACTOR.with(|current| { + match *current.borrow() { + Some(ref handle) => Ok(handle.clone()), + None => Handle::fallback(), + } + }) + } + + /// Returns a handle to the fallback reactor. + fn fallback() -> io::Result { + let mut fallback = HANDLE_FALLBACK.load(SeqCst); + + // If the fallback hasn't been previously initialized then let's spin + // up a helper thread and try to initialize with that. If we can't + // actually create a helper thread then we'll just return a "defunct" + // handle which will return errors when I/O objects are attempted to be + // associated. + if fallback == 0 { + let reactor = match Reactor::new() { + Ok(reactor) => reactor, + Err(_) => return Err(io::Error::new(io::ErrorKind::Other, + "failed to create reactor")), + }; + + // If we successfully set ourselves as the actual fallback then we + // want to `forget` the helper thread to ensure that it persists + // globally. If we fail to set ourselves as the fallback that means + // that someone was racing with this call to `Handle::default`. + // They ended up winning so we'll destroy our helper thread (which + // shuts down the thread) and reload the fallback. + if set_fallback(reactor.handle().clone()).is_ok() { + let ret = reactor.handle().clone(); + + match reactor.background() { + Ok(bg) => bg.forget(), + // The global handle is fubar, but y'all probably got bigger + // problems if a thread can't spawn. + Err(_) => {} + } + + return Ok(ret); + } + + fallback = HANDLE_FALLBACK.load(SeqCst); + } + + // At this point our fallback handle global was configured so we use + // its value to reify a handle, clone it, and then forget our reified + // handle as we don't actually have an owning reference to it. + assert!(fallback != 0); + + let ret = unsafe { + let handle = Handle::from_usize(fallback); + let ret = handle.clone(); + drop(handle.into_usize()); + ret + }; + + Ok(ret) + } + + /// Forces a reactor blocked in a call to `turn` to wakeup, or otherwise + /// makes the next call to `turn` return immediately. + /// + /// This method is intended to be used in situations where a notification + /// needs to otherwise be sent to the main reactor. If the reactor is + /// currently blocked inside of `turn` then it will wake up and soon return + /// after this method has been called. If the reactor is not currently + /// blocked in `turn`, then the next call to `turn` will not block and + /// return immediately. + fn wakeup(&self) { + if let Some(inner) = self.inner() { + inner.wakeup.set_readiness(mio::Ready::readable()).unwrap(); + } + } + + fn into_usize(self) -> usize { + unsafe { + mem::transmute::, usize>(self.inner) + } + } + + unsafe fn from_usize(val: usize) -> Handle { + let inner = mem::transmute::>(val);; + Handle { inner } + } + + fn inner(&self) -> Option> { + self.inner.upgrade() + } +} + +impl Unpark for Handle { + fn unpark(&self) { + self.wakeup(); + } +} + +impl Default for Handle { + fn default() -> Handle { + Handle::current() + } +} + +impl fmt::Debug for Handle { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "Handle") + } +} + +fn set_fallback(handle: Handle) -> Result<(), SetFallbackError> { + unsafe { + let val = handle.into_usize(); + match HANDLE_FALLBACK.compare_exchange(0, val, SeqCst, SeqCst) { + Ok(_) => Ok(()), + Err(_) => { + drop(Handle::from_usize(val)); + Err(SetFallbackError(())) + } + } + } +} + +// ===== impl Inner ===== + +impl Inner { + /// Register an I/O resource with the reactor. + /// + /// The registration token is returned. + fn add_source(&self, source: &Evented) + -> io::Result + { + let mut io_dispatch = self.io_dispatch.write().unwrap(); + + if io_dispatch.len() == MAX_SOURCES { + return Err(io::Error::new(io::ErrorKind::Other, "reactor at max \ + registered I/O resources")); + } + + // Acquire a write lock + let key = io_dispatch.insert(ScheduledIo { + readiness: AtomicUsize::new(0), + reader: AtomicTask::new(), + writer: AtomicTask::new(), + }); + + try!(self.io.register(source, + mio::Token(TOKEN_START + key), + mio::Ready::readable() | + mio::Ready::writable() | + platform::all(), + mio::PollOpt::edge())); + + Ok(key) + } + + fn drop_source(&self, token: usize) { + debug!("dropping I/O source: {}", token); + self.io_dispatch.write().unwrap().remove(token); + } + + /// Registers interest in the I/O resource associated with `token`. + fn register(&self, token: usize, dir: Direction, t: Task) { + debug!("scheduling direction for: {}", token); + let io_dispatch = self.io_dispatch.read().unwrap(); + let sched = io_dispatch.get(token).unwrap(); + + let (task, ready) = match dir { + Direction::Read => (&sched.reader, !mio::Ready::writable()), + Direction::Write => (&sched.writer, mio::Ready::writable()), + }; + + task.register_task(t); + + if sched.readiness.load(SeqCst) & ready2usize(ready) != 0 { + task.notify(); + } + } +} + +impl Drop for Inner { + fn drop(&mut self) { + // When a reactor is dropped it needs to wake up all blocked tasks as + // they'll never receive a notification, and all connected I/O objects + // will start returning errors pretty quickly. + let io = self.io_dispatch.read().unwrap(); + for (_, io) in io.iter() { + io.writer.notify(); + io.reader.notify(); + } + } +} + +impl Direction { + fn ready(&self) -> mio::Ready { + match *self { + Direction::Read => read_ready(), + Direction::Write => write_ready(), + } + } + + fn mask(&self) -> usize { + ready2usize(self.ready()) + } +} + +// ===== misc ===== + +const READ: usize = 1 << 0; +const WRITE: usize = 1 << 1; + +fn read_ready() -> mio::Ready { + mio::Ready::readable() | platform::hup() +} + +fn write_ready() -> mio::Ready { + mio::Ready::writable() +} + +// === legacy + +fn ready2usize(ready: mio::Ready) -> usize { + let mut bits = 0; + if ready.is_readable() { + bits |= READ; + } + if ready.is_writable() { + bits |= WRITE; + } + bits | platform::ready2usize(ready) +} + +fn usize2ready(bits: usize) -> mio::Ready { + let mut ready = mio::Ready::empty(); + if bits & READ != 0 { + ready.insert(mio::Ready::readable()); + } + if bits & WRITE != 0 { + ready.insert(mio::Ready::writable()); + } + ready | platform::usize2ready(bits) +} + +#[cfg(all(unix, not(target_os = "fuchsia")))] +mod platform { + use mio::Ready; + use mio::unix::UnixReady; + + #[cfg(target_os = "dragonfly")] + pub fn all() -> Ready { + hup() | UnixReady::aio() + } + + #[cfg(target_os = "freebsd")] + pub fn all() -> Ready { + hup() | UnixReady::aio() | UnixReady::lio() + } + + #[cfg(not(any(target_os = "dragonfly", target_os = "freebsd")))] + pub fn all() -> Ready { + hup() + } + + pub fn hup() -> Ready { + UnixReady::hup().into() + } + + const HUP: usize = 1 << 2; + const ERROR: usize = 1 << 3; + const AIO: usize = 1 << 4; + const LIO: usize = 1 << 5; + + #[cfg(any(target_os = "dragonfly", target_os = "freebsd"))] + fn is_aio(ready: &Ready) -> bool { + UnixReady::from(*ready).is_aio() + } + + #[cfg(not(any(target_os = "dragonfly", target_os = "freebsd")))] + fn is_aio(_ready: &Ready) -> bool { + false + } + + #[cfg(target_os = "freebsd")] + fn is_lio(ready: &Ready) -> bool { + UnixReady::from(*ready).is_lio() + } + + #[cfg(not(target_os = "freebsd"))] + fn is_lio(_ready: &Ready) -> bool { + false + } + + pub fn ready2usize(ready: Ready) -> usize { + let ready = UnixReady::from(ready); + let mut bits = 0; + if is_aio(&ready) { + bits |= AIO; + } + if is_lio(&ready) { + bits |= LIO; + } + if ready.is_error() { + bits |= ERROR; + } + if ready.is_hup() { + bits |= HUP; + } + bits + } + + #[cfg(any(target_os = "dragonfly", target_os = "freebsd", target_os = "ios", + target_os = "macos"))] + fn usize2ready_aio(ready: &mut UnixReady) { + ready.insert(UnixReady::aio()); + } + + #[cfg(not(any(target_os = "dragonfly", + target_os = "freebsd", target_os = "ios", target_os = "macos")))] + fn usize2ready_aio(_ready: &mut UnixReady) { + // aio not available here → empty + } + + #[cfg(target_os = "freebsd")] + fn usize2ready_lio(ready: &mut UnixReady) { + ready.insert(UnixReady::lio()); + } + + #[cfg(not(target_os = "freebsd"))] + fn usize2ready_lio(_ready: &mut UnixReady) { + // lio not available here → empty + } + + pub fn usize2ready(bits: usize) -> Ready { + let mut ready = UnixReady::from(Ready::empty()); + if bits & AIO != 0 { + usize2ready_aio(&mut ready); + } + if bits & LIO != 0 { + usize2ready_lio(&mut ready); + } + if bits & HUP != 0 { + ready.insert(UnixReady::hup()); + } + if bits & ERROR != 0 { + ready.insert(UnixReady::error()); + } + ready.into() + } +} + +#[cfg(any(windows, target_os = "fuchsia"))] +mod platform { + use mio::Ready; + + pub fn all() -> Ready { + // No platform-specific Readinesses for Windows + Ready::empty() + } + + pub fn hup() -> Ready { + Ready::empty() + } + + pub fn ready2usize(_r: Ready) -> usize { + 0 + } + + pub fn usize2ready(_r: usize) -> Ready { + Ready::empty() + } +} diff --git a/src/reactor/poll_evented2.rs b/tokio-reactor/src/poll_evented.rs similarity index 97% rename from src/reactor/poll_evented2.rs rename to tokio-reactor/src/poll_evented.rs index 859e49e09..58b4f11cb 100644 --- a/src/reactor/poll_evented2.rs +++ b/tokio-reactor/src/poll_evented.rs @@ -1,7 +1,4 @@ -#![allow(warnings)] - -use reactor::Handle; -use reactor::registration::Registration; +use {Handle, Registration}; use futures::{task, Async, Poll}; use mio; @@ -16,12 +13,12 @@ use std::sync::atomic::Ordering::Relaxed; /// Associates an I/O resource that implements the [`std::Read`] and / or /// [`std::Write`] traits with the reactor that drives it. /// -/// `PollEvented2` uses [`Registration`] internally to take a type that +/// `PollEvented` uses [`Registration`] internally to take a type that /// implements [`mio::Evented`] as well as [`std::Read`] and or [`std::Write`] /// and associate it with a reactor that will drive it. /// -/// Once the [`mio::Evented`] type is wrapped by `PollEvented2`, it can be -/// used from within the future's execution model. As such, the `PollEvented2` +/// Once the [`mio::Evented`] type is wrapped by `PollEvented`, it can be +/// used from within the future's execution model. As such, the `PollEvented` /// type provides [`AsyncRead`] and [`AsyncWrite`] implementations using the /// underlying I/O resource as well as readiness events provided by the reactor. /// @@ -74,7 +71,7 @@ use std::sync::atomic::Ordering::Relaxed; /// /// ## Platform-specific events /// -/// `PollEvented2` also allows receiving platform-specific `mio::Ready` events. +/// `PollEvented` also allows receiving platform-specific `mio::Ready` events. /// These events are included as part of the read readiness event stream. The /// write readiness event stream is only for `Ready::writable()` events. /// diff --git a/src/reactor/registration.rs b/tokio-reactor/src/registration.rs similarity index 99% rename from src/reactor/registration.rs rename to tokio-reactor/src/registration.rs index 7c4f82978..66a8f766f 100644 --- a/src/reactor/registration.rs +++ b/tokio-reactor/src/registration.rs @@ -1,4 +1,4 @@ -use reactor::{Handle, Direction}; +use {Handle, Direction}; use futures::{Async, Poll}; use futures::task::{self, Task}; diff --git a/tokio-threadpool/Cargo.toml b/tokio-threadpool/Cargo.toml index 97e81030f..72c090f68 100644 --- a/tokio-threadpool/Cargo.toml +++ b/tokio-threadpool/Cargo.toml @@ -7,7 +7,7 @@ homepage = "https://github.com/tokio-rs/tokio" license = "MIT/Apache-2.0" authors = ["Carl Lerche "] description = """ -A Future aware thread pool based on work stealing. +A task scheduler backed by a work-stealing thread pool. """ keywords = ["futures", "tokio"] categories = ["concurrency", "asynchronous"] diff --git a/tokio-threadpool/README.md b/tokio-threadpool/README.md index 14edb236e..f83e4a519 100644 --- a/tokio-threadpool/README.md +++ b/tokio-threadpool/README.md @@ -45,8 +45,17 @@ pub fn main() { ## License -`tokio-threadpool` is primarily distributed under the terms of both the MIT -license and the Apache License (Version 2.0), with portions covered by various -BSD-like licenses. +This project is licensed under either of -See LICENSE-APACHE, and LICENSE-MIT for details. + * Apache License, Version 2.0, ([LICENSE-APACHE](../LICENSE-APACHE) or + http://www.apache.org/licenses/LICENSE-2.0) + * MIT license ([LICENSE-MIT](../LICENSE-MIT) or + http://opensource.org/licenses/MIT) + +at your option. + +### Contribution + +Unless you explicitly state otherwise, any contribution intentionally submitted +for inclusion in Tokio by you, as defined in the Apache-2.0 license, shall be +dual licensed as above, without any additional terms or conditions.