From df19119c0a702508ab14d96145680a54e8bc3ea4 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Thu, 1 Mar 2018 21:48:18 -0800 Subject: [PATCH] Add `io` facade and update `reactor` docs (#166) This patch updates the documentation for a number of APIs. It also introduces a prelude module and an io facade module, re-exporting types from tokio-io. --- Cargo.toml | 5 +- README.md | 45 +++++++++ src/lib.rs | 156 ++++++++++++++++++++++------- src/reactor/mod.rs | 145 ++++++++++++++++++++++++--- src/reactor/poll_evented2.rs | 188 +++++++++++++++++------------------ src/reactor/registration.rs | 114 ++++++++++++++++++--- tokio-io/src/async_read.rs | 16 +-- tokio-io/src/async_write.rs | 16 +-- 8 files changed, 508 insertions(+), 177 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 0a5c5d7ef..0d7cb00fb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +33,7 @@ travis-ci = { repository = "tokio-rs/tokio" } appveyor = { repository = "carllerche/tokio" } [dependencies] -tokio-io = "0.1" +tokio-io = { version = "0.1", path = "tokio-io" } tokio-executor = { version = "0.1", path = "tokio-executor" } tokio-threadpool = { version = "0.1", path = "tokio-threadpool" } bytes = "0.4" @@ -55,3 +55,6 @@ serde = "1.0" serde_derive = "1.0" serde_json = "1.0" time = "0.1" + +[patch.crates-io] +tokio-io = { path = "tokio-io" } diff --git a/README.md b/README.md index 3b98d7c6b..33cf17482 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,51 @@ an asynchronous application. [reactor]: https://docs.rs/tokio/0.1.1/tokio/reactor/index.html [scheduler]: https://tokio-rs.github.io/tokio/tokio/runtime/index.html +## Example + +A basic TCP echo server with Tokio: + +```rust +extern crate tokio; + +use tokio::prelude::*; +use tokio::io::copy; +use tokio::net::TcpListener; + +fn main() { + // Bind the server's socket. + let addr = "127.0.0.1:12345".parse().unwrap(); + let listener = TcpListener::bind(&addr) + .expect("unable to bind TCP listener"); + + // Pull out a stream of sockets for incoming connections + let server = listener.incoming() + .map_err(|e| eprintln!("accept failed = {:?}", e)) + .for_each(|sock| { + // Split up the reading and writing parts of the + // socket. + let (reader, writer) = sock.split(); + + // A future that echos the data and returns how + // many bytes were copied... + let bytes_copied = copy(reader, writer); + + // ... after which we'll print what happened. + let handle_conn = bytes_copied.map(|amt| { + println!("wrote {:?} bytes", amt) + }).map_err(|err| { + eprintln!("IO error {:?}", err) + }); + + // Spawn the future as a concurrent task. + tokio::spawn(handle_conn) + }); + + // Start the Tokio runtime + tokio::run(server); +} +``` + # License This project is licensed under either of diff --git a/src/lib.rs b/src/lib.rs index f36ce326d..4c32bfa86 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,50 +1,31 @@ -//! `Future`-powered I/O at the core of Tokio +//! A runtime for writing reliable, asynchronous, and slim applications. //! -//! This crate uses the [`futures`] crate to provide an event loop ("reactor -//! core") which can be used to drive I/O like TCP and UDP. All asynchronous I/O -//! is powered by the [`mio`] crate. +//! Tokio is an event-driven, non-blocking I/O platform for writing asynchronous +//! applications with the Rust programming language. At a high level, it +//! provides a few major components: //! -//! [`futures`]: ../futures/index.html -//! [`mio`]: ../mio/index.html +//! * A multi threaded, work-stealing based task [scheduler][runtime]. +//! * A [reactor][reactor] backed by the operating system's event queue (epoll, kqueue, +//! IOCP, etc...). +//! * Asynchronous [TCP and UDP][net] sockets. //! -//! The concrete types provided in this crate are relatively bare bones but are -//! intended to be the essential foundation for further projects needing an -//! event loop. In this crate you'll find: +//! Tokio is built using futures (provided by the [futures] crate) as the +//! abstraction for managing the complexity of asynchronous programming. //! -//! * TCP, both streams and listeners. -//! * UDP sockets. -//! * An event loop to run futures. +//! Guide level documentation is found on the [website]. //! -//! More functionality is likely to be added over time, but otherwise the crate -//! is intended to be flexible, with the [`PollEvented`] type accepting any -//! type that implements [`mio::Evented`]. For example, the [`tokio-uds`] crate -//! uses [`PollEvented`] to provide support for Unix domain sockets. -//! -//! [`PollEvented`]: ./reactor/struct.PollEvented.html -//! [`mio::Evented`]: ../mio/event/trait.Evented.html -//! [`tokio-uds`]: https://crates.io/crates/tokio-uds -//! -//! Some other important tasks covered by this crate are: -//! -//! * All I/O is futures-aware. If any action in this crate returns "not ready" -//! or "would block", then the current future task is scheduled to receive a -//! notification when it would otherwise make progress. -//! -//! You can find more extensive documentation in terms of tutorials at -//! [https://tokio.rs](https://tokio.rs). +//! [website]: https://tokio.rs/docs/getting-started/hello-world/ +//! [futures]: http://docs.rs/futures //! //! # Examples //! //! A simple TCP echo server: //! //! ```no_run -//! extern crate futures; //! extern crate tokio; -//! extern crate tokio_io; //! -//! use futures::prelude::*; -//! use tokio_io::AsyncRead; -//! use tokio_io::io::copy; +//! use tokio::prelude::*; +//! use tokio::io::copy; //! use tokio::net::TcpListener; //! //! fn main() { @@ -55,7 +36,7 @@ //! //! // Pull out a stream of sockets for incoming connections //! let server = listener.incoming() -//! .map_err(|e| println!("accept failed = {:?}", e)) +//! .map_err(|e| eprintln!("accept failed = {:?}", e)) //! .for_each(|sock| { //! // Split up the reading and writing parts of the //! // socket. @@ -82,9 +63,7 @@ //! ``` #![doc(html_root_url = "https://docs.rs/tokio/0.1.1")] -#![deny(missing_docs)] -#![deny(warnings)] -#![warn(missing_debug_implementations)] +#![deny(missing_docs, warnings, missing_debug_implementations)] extern crate bytes; #[macro_use] @@ -109,3 +88,104 @@ pub use executor::spawn; pub use runtime::run; mod atomic_task; + +pub mod io { + //! Asynchronous I/O. + //! + //! This module is the asynchronous version of `std::io`. Primarily, it + //! defines two traits, [`AsyncRead`] and [`AsyncWrite`], which extend the + //! `Read` and `Write` traits of the standard library. + //! + //! [`AsyncRead`] and [`AsyncWrite`] must only be implemented for + //! non-blocking I/O types that integrate with the futures type system. In + //! other words, these types must never block the thread, and instead the + //! current task is notified when the I/O resource is ready. + //! + //! Utilities functions are provided for working with [`AsyncRead`] / + //! [`AsyncWrite`] types. For example, [`copy`] asynchronously copies all + //! data from a source to a destination. + //! + //! Additionally, [`Read`], [`Write`], [`Error`], [`ErrorKind`], and + //! [`Result`] are re-exported from `std::io` for ease of use. + //! + //! [`AsyncRead`]: trait.AsyncRead.html + //! [`AsyncWrite`]: trait.AsyncWrite.html + //! [`copy`]: fn.copy.html + //! [`Read`]: trait.Read.html + //! [`Write`]: trait.Write.html + //! [`Error`]: struct.Error.html + //! [`ErrorKind`]: enum.ErrorKind.html + //! [`Result`]: type.Result.html + + pub use tokio_io::{ + AsyncRead, + AsyncWrite, + }; + + // Utils + pub use tokio_io::io::{ + copy, + Copy, + flush, + Flush, + lines, + Lines, + read_exact, + ReadExact, + read_to_end, + ReadToEnd, + read_until, + ReadUntil, + shutdown, + Shutdown, + write_all, + WriteAll, + }; + + // Re-export io::Error so that users don't have to deal + // with conflicts when `use`ing `futures::io` and `std::io`. + pub use ::std::io::{ + Error, + ErrorKind, + Result, + Read, + Write, + }; +} + +pub mod prelude { + //! A "prelude" for users of the `tokio` crate. + //! + //! This prelude is similar to the standard library's prelude in that you'll + //! almost always want to import its entire contents, but unlike the standard + //! library's prelude you'll have to do so manually: + //! + //! ``` + //! use tokio::prelude::*; + //! ``` + //! + //! The prelude may grow over time as additional items see ubiquitous use. + + pub use tokio_io::{ + AsyncRead, + AsyncWrite, + }; + + pub use ::std::io::{ + Read, + Write, + }; + + pub use futures::{ + Future, + future, + Stream, + stream, + Sink, + IntoFuture, + Async, + AsyncSink, + Poll, + task, + }; +} diff --git a/src/reactor/mod.rs b/src/reactor/mod.rs index 4e62eb1d2..2a936bafb 100644 --- a/src/reactor/mod.rs +++ b/src/reactor/mod.rs @@ -1,20 +1,139 @@ -//! The core reactor driving all I/O. +//! Event loop that drives I/O resources. //! -//! This module contains the [`Reactor`] reactor type which is the event loop for -//! all I/O happening in `tokio`. This core reactor (or event loop) is used to -//! drive 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 +//! operating system ([epoll], [kqueue], [IOCP], etc...) and forward them to +//! waiting tasks. It is the bridge between operating system and the futures +//! model. //! -//! The [`Handle`] struct, created by [`handle`][handle_method], is a reference -//! to the event loop and can be used when constructing I/O objects. +//! # Overview //! -//! Lastly [`PollEvented`] can be used to construct I/O objects that interact -//! with the event loop, e.g. [`TcpStream`] in the net module. +//! When using Tokio, all operations are asynchronous and represented by +//! futures. These futures, representing the application logic, are scheduled by +//! an executor (see [runtime model] for more details). Executors wait for +//! notifications before scheduling the future for execution time, i.e., nothing +//! happens until an event is received indicating that the task can make +//! progress. +//! +//! The reactor receives events from the operating system and notifies the +//! executor. +//! +//! Let's start with a basic example, establishing a TCP connection. +//! +//! ```rust +//! # extern crate tokio; +//! # fn dox() { +//! use tokio::prelude::*; +//! use tokio::net::TcpStream; +//! +//! let addr = "93.184.216.34:9243".parse().unwrap(); +//! +//! let connect_future = TcpStream::connect(&addr); +//! +//! let task = connect_future +//! .and_then(|socket| { +//! println!("successfully connected"); +//! Ok(()) +//! }) +//! .map_err(|e| println!("failed to connect; err={:?}", e)); +//! +//! tokio::run(task); +//! # } +//! # fn main() {} +//! ``` +//! +//! Establishing a TCP connection usually cannot be completed immediately. +//! [`TcpStream::connect`] does not block the current thread. Instead, it +//! returns a [future][connect-future] that resolves once the TCP connection has +//! been established. The connect future itself has no way of knowing when the +//! TCP connection has been established. +//! +//! Before returning the future, [`TcpStream::connect`] registers the socket +//! with a reactor. This registration process, handled by [`Registration`], is +//! what links the [`TcpStream`] with the [`Reactor`] instance. At this point, +//! the reactor starts listening for connection events from the operating system +//! for that socket. +//! +//! Once the connect future is passed to [`tokio::run`], it is spawned onto a +//! thread pool. The thread pool waits until it is notified that the connection +//! has completed. +//! +//! When the TCP connection is established, the reactor receives an event from +//! the operating system. It then notifies the thread pool, telling it that the +//! connect future can complete. At this point, the thread pool will schedule +//! the task to run on one of its worker threads. This results in the `and_then` +//! closure to get executed. +//! +//! ## Lazy registration +//! +//! Notice how the snippet above does not explicitly reference a reactor. When +//! [`TcpStream::connect`] is called, it registers the socket with a reactor, +//! but no reactor is specified. This works because the registration process +//! mentioned above is actually lazy. It doesn't *actually* happen in the +//! [`connect`] function. Instead, the registration is established the first +//! time that the task is polled (again, see [runtime model]). +//! +//! A reactor instance is automatically made available when using the Tokio +//! [runtime], which is done using [`tokio::run`]. The Tokio runtime's executor +//! sets a thread-local variable referencing the associated [`Reactor`] instance +//! and [`Handle::current`] (used by [`Registration`]) returns the reference. +//! +//! ## Implementation +//! +//! The reactor implementation uses [`mio`] to interface with the operating +//! system's event queue. A call to [`Reactor::poll`] results in in a single +//! call to [`Poll::poll`] which in turn results in a single call to the +//! operating system's selector. +//! +//! The reactor maintains state for each registered I/O resource. This tracks +//! the executor task to notify when events are provided by the operating +//! system's selector. This state is stored in a `Sync` data structure and +//! referenced by [`Registration`]. When the [`Registration`] instance is +//! dropped, this state is cleaned up. Because the state is stored in a `Sync` +//! data structure, the [`Registration`] instance is able to be moved to other +//! threads. +//! +//! By default, a runtime's default reactor runs on a background thread. This +//! ensures that application code cannot significantly impact the reactor's +//! responsiveness. +//! +//! ## Integrating with the reactor +//! +//! Tokio comes with a number of I/O resources, like TCP and UDP sockets, that +//! automatically integrate with the reactor. However, library authors or +//! applications may wish to implement their own resources that are also backed +//! by the reactor. +//! +//! 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 +//! suited. +//! +//! Otherwise, [`Registration`] can be used directly. This provides the lowest +//! level primitive needed for integrating with the reactor: a stream of +//! readiness events. //! //! [`Reactor`]: struct.Reactor.html -//! [`Handle`]: struct.Handle.html -//! [handle_method]: struct.Reactor.html#method.handle -//! [`PollEvented`]: struct.PollEvented.html +//! [`Registration`]: struct.Registration.html +//! [runtime model]: https://tokio.rs/docs/getting-started/runtime-model/ +//! [epoll]: http://man7.org/linux/man-pages/man7/epoll.7.html +//! [kqueue]: https://www.freebsd.org/cgi/man.cgi?query=kqueue&sektion=2 +//! [IOCP]: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365198(v=vs.85).aspx +//! [`TcpStream::connect`]: ../net/struct.TcpStream.html#method.connect +//! [`connect`]: ../net/struct.TcpStream.html#method.connect +//! [connect-future]: ../net/struct.ConnectFuture.html +//! [`tokio::run`]: ../runtime/fn.run.html //! [`TcpStream`]: ../net/struct.TcpStream.html +//! [runtime]: ../runtime +//! [`Handle::current`]: struct.Handle.html#method.current +//! [`mio`]: https://github.com/carllerche/mio +//! [`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 +//! [`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}; @@ -65,11 +184,11 @@ pub struct Reactor { _wakeup_registration: mio::Registration, } -/// A handle to an event loop. +/// 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 an implicitly configured handle for your thread. +/// and will instead use the default reactor for the execution context. #[derive(Clone)] pub struct Handle { inner: Weak, diff --git a/src/reactor/poll_evented2.rs b/src/reactor/poll_evented2.rs index 9affad0e5..859e49e09 100644 --- a/src/reactor/poll_evented2.rs +++ b/src/reactor/poll_evented2.rs @@ -1,12 +1,3 @@ -//! -//! Readiness tracking streams, backing I/O objects. -//! -//! This module contains the core type which is used to back all I/O on object -//! in `tokio-core`. The `PollEvented` type is the implementation detail of -//! all I/O. Each `PollEvented` manages registration with a reactor, -//! acquisition of a token, and tracking of the readiness state on the -//! underlying I/O primitive. - #![allow(warnings)] use reactor::Handle; @@ -22,51 +13,78 @@ use std::io::{self, Read, Write}; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::Relaxed; -/// A concrete implementation of a stream of readiness notifications for I/O -/// objects that originates from an event loop. +/// Associates an I/O resource that implements the [`std::Read`] and / or +/// [`std::Write`] traits with the reactor that drives it. /// -/// 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. +/// `PollEvented2` 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. /// -/// 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. +/// Once the [`mio::Evented`] type is wrapped by `PollEvented2`, it can be +/// used from within the future's execution model. As such, the `PollEvented2` +/// type provides [`AsyncRead`] and [`AsyncWrite`] implementations using the +/// underlying I/O resource as well as readiness events provided by the reactor. /// -/// 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. +/// **Note**: While `PollEvented` is `Sync` (if the underlying I/O type is +/// `Sync`), the caller must ensure that there are at most two tasks that use a +/// `PollEvented` instance concurrenty. One for reading and one for writing. +/// While violating this requirement is "safe" from a Rust memory model point of +/// view, it will result in unexpected behavior in the form of lost +/// notifications and tasks hanging. /// -/// 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. +/// ## Readiness events /// -/// You can find more information about creating a custom I/O object [online]. +/// Besides just providing [`AsyncRead`] and [`AsyncWrite`] implementations, +/// this type also supports access to the underlying readiness event stream. +/// While similar in function to what [`Registration`] provides, the semantics +/// are a bit different. /// -/// [online]: https://tokio.rs/docs/going-deeper-tokio/core-low-level/#custom-io +/// Two functions are provided to access the readiness events: +/// [`poll_read_ready`] and [`poll_write_ready`]. These functions return the +/// current readiness state of the `PollEvented` instance. If +/// [`poll_read_ready`] indicates read readiness, immediately calling +/// [`poll_read_ready`] again will also indicate read readiness. /// -/// ## Readiness to read/write +/// When the operation is attempted and is unable to succeed due to the I/O +/// resource not being ready, the caller must call [`need_read`] or +/// [`need_write`]. This clears the readiness state until a new readiness event +/// is received. /// -/// 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! +/// This allows the caller to implement additional funcitons. For example, +/// [`TcpListener`] implements accept by using [`poll_read_ready`] and +/// [`need_read`]. /// -/// 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. +/// ```rust,ignore +/// pub fn accept(&mut self) -> io::Result<(net::TcpStream, SocketAddr)> { +/// if let Async::NotReady = self.poll_evented.poll_read_ready()? { +/// return Err(io::ErrorKind::WouldBlock.into()) +/// } /// -/// 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`. +/// match self.poll_evented.get_ref().accept_std() { +/// Ok(pair) => Ok(pair), +/// Err(e) => { +/// if e.kind() == io::ErrorKind::WouldBlock { +/// self.poll_evented.need_read()?; +/// } +/// Err(e) +/// } +/// } +/// } +/// ``` +/// +/// ## Platform-specific events +/// +/// `PollEvented2` 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. +/// +/// [`std::Read`]: https://doc.rust-lang.org/std/io/trait.Read.html +/// [`std::Write`]: https://doc.rust-lang.org/std/io/trait.Write.html +/// [`AsyncRead`]: ../io/trait.AsyncRead.html +/// [`AsyncWrite`]: ../io/trait.AsyncWrite.html +/// [`mio::Evented`]: https://docs.rs/mio/0.6/mio/trait.Evented.html +/// [`Registration`]: struct.Registration.html +/// [`TcpListener`]: ../net/struct.TcpListener.html pub struct PollEvented { io: E, inner: Inner, @@ -106,18 +124,19 @@ where E: Evented Ok(ret) } - /// Tests to see if this source is ready to be read from or not. + /// Check the I/O resource's read readiness state. /// - /// If this stream is not ready for a read then `Async::NotReady` will be - /// returned and the current task will be scheduled to receive a - /// notification when the stream is readable again. In other words, this - /// method is only safe to call from within the context of a future's task, - /// typically done in a `Future::poll` method. + /// If the resource is not ready for a read then `Async::NotReady` is + /// returned and the current task is notified once a new event is received. + /// + /// The I/O resource will remain in a read-ready state until readiness is + /// cleared by calling [`need_read`]. + /// + /// [`need_read`]: #method.need_read /// /// # Panics /// - /// This function will panic if called outside the context of a future's - /// task. + /// This function will panic if called from outside of a task context. pub fn poll_read_ready(&self) -> Poll { self.register()?; @@ -143,29 +162,18 @@ where E: Evented Ok(ready.into()) } - /// Indicates to this source of events that the corresponding I/O object is - /// no longer readable, but it needs to be. + /// Resets the I/O resource's read readiness state and registers the current + /// task to be notified once a read readiness event is received. /// - /// This function, like `poll_read`, is only safe to call from the context - /// of a future's task (typically in a `Future::poll` implementation). It - /// informs this readiness stream that the underlying object is no longer - /// readable, typically because a "would block" error was seen. + /// After calling this function, `poll_read_ready` will return `NotReady` + /// until a new read readiness event has been received. /// - /// *All* readiness bits associated with this stream except the writable bit - /// will be reset when this method is called. The current task is then - /// scheduled to receive a notification whenever anything changes other than - /// the writable bit. Note that this typically just means the readable bit - /// is used here, but if you're using a custom I/O object for events like - /// hup/error this may also be relevant. - /// - /// Note that it is also only valid to call this method if `poll_read` - /// previously indicated that the object is readable. That is, this function - /// must always be paired with calls to `poll_read` previously. + /// This function clears **all** readiness state **except** write readiness. + /// This includes any platform-specific readiness bits. /// /// # Panics /// - /// This function will panic if called outside the context of a future's - /// task. + /// This function will panic if called from outside of a task context. pub fn need_read(&self) -> io::Result<()> { self.inner.read_readiness.store(0, Relaxed); @@ -177,18 +185,19 @@ where E: Evented Ok(()) } - /// Tests to see if this source is ready to be written to or not. + /// Check the I/O resource's write readiness state. /// - /// If this stream is not ready for a write then `Async::NotReady` will be - /// returned and the current task will be scheduled to receive a - /// notification when the stream is writable again. In other words, this - /// method is only safe to call from within the context of a future's task, - /// typically done in a `Future::poll` method. + /// If the resource is not ready for a write then `Async::NotReady` is + /// returned and the current task is notified once a new event is received. + /// + /// The I/O resource will remain in a write-ready state until readiness is + /// cleared by calling [`need_write`]. + /// + /// [`need_write`]: #method.need_write /// /// # Panics /// - /// This function will panic if called outside the context of a future's - /// task. + /// This function will panic if called from outside of a task context. pub fn poll_write_ready(&self) -> Poll { self.register()?; @@ -213,28 +222,15 @@ where E: Evented Ok(ready.into()) } - /// Indicates to this source of events that the corresponding I/O object is - /// no longer writable, but it needs to be. + /// Resets the I/O resource's write readiness state and registers the current + /// task to be notified once a write readiness event is received. /// - /// This function, like `poll_write_ready`, is only safe to call from the - /// context of a future's task (typically in a `Future::poll` - /// implementation). It informs this readiness stream that the underlying - /// object is no longer writable, typically because a "would block" error - /// was seen. - /// - /// The flag indicating that this stream is writable is unset and the - /// current task is scheduled to receive a notification when the stream is - /// then again writable. - /// - /// Note that it is also only valid to call this method if - /// `poll_write_ready` previously indicated that the object is writable. - /// That is, this function must always be paired with calls to `poll_write` - /// previously. + /// After calling this function, `poll_write_ready` will return `NotReady` + /// until a new read readiness event has been received. /// /// # Panics /// - /// This function will panic if called outside the context of a future's - /// task. + /// This function will panic if called from outside of a task context. pub fn need_write(&self) -> io::Result<()> { self.inner.write_readiness.store(0, Relaxed); diff --git a/src/reactor/registration.rs b/src/reactor/registration.rs index 69d4d00e8..7c4f82978 100644 --- a/src/reactor/registration.rs +++ b/src/reactor/registration.rs @@ -9,17 +9,37 @@ use std::cell::UnsafeCell; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering::SeqCst; -/// Handle to a reactor registration. +/// Associates an I/O resource with the reactor instance that drives it. /// /// A registration represents an I/O resource registered with a Reactor such -/// that it will receive task notifications on readiness. +/// that it will receive task notifications on readiness. This is the lowest +/// level API for integrating with a reactor. /// -/// The registration is lazily made and supports concurrent operations. This -/// allows a `Registration` instance to be created without the reactor handle -/// that will eventually be used to drive the resource. +/// The association between an I/O resource is made by calling [`register`]. +/// Once the association is established, it remains established until the +/// registration instance is dropped. Subsequent calls to [`register`] are +/// no-ops. /// -/// The difficulty is due to the fact that a single registration drives two -/// separate tasks -- A read half and a write half. +/// A registration instance represents two separate readiness streams. One for +/// the read readiness and one for write readiness. These streams are +/// independent and can be consumed from separate tasks. +/// +/// **Note**: while `Registration` is `Sync`, the caller must ensure that there +/// are at most two tasks that use a registration instance concurrently. One +/// task for [`poll_read_ready`] and one task for [`poll_write_ready`]. While +/// violating this requirement is "safe" from a Rust memory safety point of +/// view, it will result in unexpected behavior in the form of lost +/// notifications and tasks hanging. +/// +/// ## Platform-specific events +/// +/// `Registration` 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. +/// +/// [`register`]: #method.register +/// [`poll_read_ready`]: #method.poll_read_ready`] +/// [`poll_write_ready`]: #method.poll_write_ready`] #[derive(Debug)] pub struct Registration { /// Stores the handle. Once set, the value is not changed. @@ -187,7 +207,36 @@ impl Registration { } } - /// Poll for changes in the I/O resource's read readiness. + /// Poll for events on the I/O resource's read readiness stream. + /// + /// If the I/O resource receives a new read readiness event since the last + /// call to `poll_read_ready`, it is returned. If it has not, the current + /// task is notified once a new event is received. + /// + /// Events are [edge-triggered]. + /// + /// Ensure that [`register`] has been called first. + /// + /// # Return value + /// + /// There are several possible return values: + /// + /// * `Ok(Async::Ready(readiness))` means that the I/O resource has received + /// a new readiness event. The readiness value is included. + /// + /// * `Ok(NotReady)` means that no new readiness events have been received + /// since the last call to `poll_read_ready`. + /// + /// * `Err(err)` means that the registration has encountered an error. This + /// error either represents a permanent internal error **or** the fact + /// that [`register`] was not called first. + /// + /// [`register`]: #method.register + /// [edge-triggered]: https://docs.rs/mio/0.6/mio/struct.Poll.html#edge-triggered-and-level-triggered + /// + /// # Panics + /// + /// This function will panic if called from outside of a task context. pub fn poll_read_ready(&self) -> Poll { self.poll_ready(Direction::Read, true) .map(|v| match v { @@ -196,16 +245,48 @@ impl Registration { }) } - /// Try taking the I/O resource's read readiness. + /// Consume any pending read readiness event. /// - /// Unlike `poll_read_ready`, this does not register the current task for - /// notification. + /// This function is identical to [`poll_read_ready`] **except** that it + /// will not notify the current task when a new event is received. As such, + /// it is safe to call this function from outside of a task context. + /// + /// [`poll_read_ready`]: #method.poll_read_ready pub fn take_read_ready(&self) -> io::Result> { self.poll_ready(Direction::Read, false) } - /// Poll for changes in the I/O resource's write readiness. + /// Poll for events on the I/O resource's write readiness stream. + /// + /// If the I/O resource receives a new write readiness event since the last + /// call to `poll_write_ready`, it is returned. If it has not, the current + /// task is notified once a new event is received. + /// + /// Events are [edge-triggered]. + /// + /// Ensure that [`register`] has been called first. + /// + /// # Return value + /// + /// There are several possible return values: + /// + /// * `Ok(Async::Ready(readiness))` means that the I/O resource has received + /// a new readiness event. The readiness value is included. + /// + /// * `Ok(NotReady)` means that no new readiness events have been received + /// since the last call to `poll_write_ready`. + /// + /// * `Err(err)` means that the registration has encountered an error. This + /// error either represents a permanent internal error **or** the fact + /// that [`register`] was not called first. + /// + /// [`register`]: #method.register + /// [edge-triggered]: https://docs.rs/mio/0.6/mio/struct.Poll.html#edge-triggered-and-level-triggered + /// + /// # Panics + /// + /// This function will panic if called from outside of a task context. pub fn poll_write_ready(&self) -> Poll { self.poll_ready(Direction::Write, true) .map(|v| match v { @@ -214,10 +295,13 @@ impl Registration { }) } - /// Try taking the I/O resource's write readiness. + /// Consume any pending write readiness event. /// - /// Unlike `poll_write_ready`, this does not register the current task for - /// notification. + /// This function is identical to [`poll_write_ready`] **except** that it + /// will not notify the current task when a new event is received. As such, + /// it is safe to call this function from outside of a task context. + /// + /// [`poll_write_ready`]: #method.poll_write_ready pub fn take_write_ready(&self) -> io::Result> { self.poll_ready(Direction::Write, false) } diff --git a/tokio-io/src/async_read.rs b/tokio-io/src/async_read.rs index fb72c14b4..08d851ef8 100644 --- a/tokio-io/src/async_read.rs +++ b/tokio-io/src/async_read.rs @@ -6,23 +6,25 @@ use {framed, split, AsyncWrite}; use codec::{Decoder, Encoder, Framed}; use split::{ReadHalf, WriteHalf}; -/// A trait for readable objects which operated in an asynchronous and -/// futures-aware fashion. +/// Read bytes asynchronously. /// -/// This trait inherits from `io::Read` and indicates as a marker that an I/O -/// object is **nonblocking**, meaning that it will return an error instead of -/// blocking when bytes are unavailable, but the stream hasn't reached EOF. -/// Specifically this means that the `read` function for types that implement -/// this trait can have a few return values: +/// This trait inherits from `std::io::Read` and indicates that an I/O object is +/// **non-blocking**. All non-blocking I/O objects must return an error when +/// bytes are unavailable instead of blocking the current thread. +/// +/// Specifically, this means that the `read` function will return one of the +/// following: /// /// * `Ok(n)` means that `n` bytes of data was immediately read and placed into /// the output buffer, where `n` == 0 implies that EOF has been reached. +/// /// * `Err(e) if e.kind() == ErrorKind::WouldBlock` means that no data was read /// into the buffer provided. The I/O object is not currently readable but may /// become readable in the future. Most importantly, **the current future's /// task is scheduled to get unparked when the object is readable**. This /// means that like `Future::poll` you'll receive a notification when the I/O /// object is readable again. +/// /// * `Err(e)` for other errors are standard I/O errors coming from the /// underlying object. /// diff --git a/tokio-io/src/async_write.rs b/tokio-io/src/async_write.rs index e152cd5a7..8c27ee12e 100644 --- a/tokio-io/src/async_write.rs +++ b/tokio-io/src/async_write.rs @@ -5,22 +5,24 @@ use futures::{Async, Poll}; use AsyncRead; -/// A trait for writable objects which operated in an asynchronous and -/// futures-aware fashion. +/// Writes bytes asynchronously. /// -/// This trait inherits from `io::Write` and indicates that an I/O object is -/// **nonblocking**, meaning that it will return an error instead of blocking -/// when bytes cannot currently be written, but hasn't closed. Specifically -/// this means that the `write` function for types that implement this trait -/// can have a few return values: +/// The trait inherits from `std::io::Write` and indicates that an I/O object is +/// **nonblocking**. All non-blocking I/O objects must return an error when +/// bytes cannot be written instead of blocking the current thread. +/// +/// Specifically, this means that the `write` function will return one of the +/// following: /// /// * `Ok(n)` means that `n` bytes of data was immediately written . +/// /// * `Err(e) if e.kind() == ErrorKind::WouldBlock` means that no data was /// written from the buffer provided. The I/O object is not currently /// writable but may become writable in the future. Most importantly, **the /// current future's task is scheduled to get unparked when the object is /// readable**. This means that like `Future::poll` you'll receive a /// notification when the I/O object is writable again. +/// /// * `Err(e)` for other errors are standard I/O errors coming from the /// underlying object. ///