tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-07 00:00:09 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add io facade and update reactor docs (#166)

Browse Source 
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.
This commit is contained in:
Carl Lerche
2018-03-01 21:48:18 -08:00
committed by GitHub
parent 164ee8f106
commit df19119c0a
8 changed files with 508 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+4 -1
View File
@@ -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" }
README.md
+45
View File
@@ -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
src/lib.rs
+118 -38
View File
@@ -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,
};
}
src/reactor/mod.rs
+132 -13
View File
@@ -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<Inner>,
src/reactor/poll_evented2.rs
+92 -96
View File
@@ -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<E> {
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<mio::Ready, io::Error> {
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<mio::Ready, io::Error> {
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);
src/reactor/registration.rs
+99 -15
View File
@@ -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<mio::Ready, io::Error> {
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<Option<mio::Ready>> {
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<mio::Ready, io::Error> {
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<Option<mio::Ready>> {
self.poll_ready(Direction::Write, false)
}
tokio-io/src/async_read.rs
+9 -7
View File
@@ -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.
///
tokio-io/src/async_write.rs
+9 -7
View File
@@ -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.
///