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

Extract the reactor to a dedicated crate. (#169)

Browse Source 
This allows libraries that require access to reactor related types to
depend on this crate without having to depend on the entirety of Tokio.

For example, libraries that implement their custom I/O resource will
need to access `Registration` or `PollEvented`.
This commit is contained in:
Carl Lerche
2018-03-02 13:51:34 -08:00
committed by GitHub
parent df6e24255b
commit e1b3085153
21 changed files with 1268 additions and 1119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/atomic_task.rs
-191
View File
@@ -1,191 +0,0 @@
use futures::task::{self, Task};
use std::fmt;
use std::cell::UnsafeCell;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::{Acquire, Release};
/// A synchronization primitive for task notification.
///
/// `AtomicTask` will coordinate concurrent notifications with the consumer
/// potentially "updating" the underlying task to notify. This is useful in
/// scenarios where a computation completes in another thread and wants to
/// notify the consumer, but the consumer is in the process of being migrated to
/// a new logical task.
///
/// Consumers should call `register` before checking the result of a computation
/// and producers should call `notify` after producing the computation (this
/// differs from the usual `thread::park` pattern). It is also permitted for
/// `notify` to be called **before** `register`. This results in a no-op.
///
/// A single `AtomicTask` may be reused for any number of calls to `register` or
/// `notify`.
///
/// `AtomicTask` does not provide any memory ordering guarantees, as such the
/// user should use caution and use other synchronization primitives to guard
/// the result of the underlying computation.
pub struct AtomicTask {
state: AtomicUsize,
task: UnsafeCell<Option<Task>>,
}
/// Initial state, the `AtomicTask` is currently not being used.
///
/// The value `2` is picked specifically because it between the write lock &
/// read lock values. Since the read lock is represented by an incrementing
/// counter, this enables an atomic fetch_sub operation to be used for releasing
/// a lock.
const WAITING: usize = 2;
/// The `register` function has determined that the task is no longer current.
/// This implies that `AtomicTask::register` is being called from a different
/// task than is represented by the currently stored task. The write lock is
/// obtained to update the task cell.
const LOCKED_WRITE: usize = 0;
/// At least one call to `notify` happened concurrently to `register` updating
/// the task cell. This state is detected when `register` exits the mutation
/// code and signals to `register` that it is responsible for notifying its own
/// task.
const LOCKED_WRITE_NOTIFIED: usize = 1;
/// The `notify` function has locked access to the task cell for notification.
///
/// The constant is left here mostly for documentation reasons.
#[allow(dead_code)]
const LOCKED_READ: usize = 3;
impl AtomicTask {
/// Create an `AtomicTask` initialized with the given `Task`
pub fn new() -> AtomicTask {
// Make sure that task is Sync
trait AssertSync: Sync {}
impl AssertSync for Task {}
AtomicTask {
state: AtomicUsize::new(WAITING),
task: UnsafeCell::new(None),
}
}
/// Registers the **current** task to be notified on calls to `notify`.
pub fn register(&self) {
self.register_task(task::current());
}
/// Registers the task to be notified on calls to `notify`.
///
/// The new task will take place of any previous tasks that were registered
/// by previous calls to `register`. Any calls to `notify` that happen after
/// a call to `register` (as defined by the memory ordering rules), will
/// notify the `register` caller's task.
///
/// It is safe to call `register` with multiple other threads concurrently
/// calling `notify`. This will result in the `register` caller's current
/// task being notified once.
///
/// This function is safe to call concurrently, but this is generally a bad
/// idea. Concurrent calls to `register` will attempt to register different
/// tasks to be notified. One of the callers will win and have its task set,
/// but there is no guarantee as to which caller will succeed.
pub fn register_task(&self, task: Task) {
match self.state.compare_and_swap(WAITING, LOCKED_WRITE, Acquire) {
WAITING => {
unsafe {
// Locked acquired, update the task cell
*self.task.get() = Some(task);
// Release the lock. If the state transitioned to
// `LOCKED_NOTIFIED`, this means that an notify has been
// signaled, so notify the task.
if LOCKED_WRITE_NOTIFIED == self.state.swap(WAITING, Release) {
(*self.task.get()).as_ref().unwrap().notify();
}
}
}
LOCKED_WRITE | LOCKED_WRITE_NOTIFIED => {
// A thread is concurrently calling `register`. This shouldn't
// happen as it doesn't really make much sense, but it isn't
// unsafe per se. Since two threads are concurrently trying to
// update the task, it's undefined which one "wins" (no ordering
// guarantees), so we can just do nothing.
}
state => {
debug_assert!(state != LOCKED_WRITE, "unexpected state LOCKED_WRITE");
debug_assert!(state != LOCKED_WRITE_NOTIFIED, "unexpected state LOCKED_WRITE_NOTIFIED");
// Currently in a read locked state, this implies that `notify`
// is currently being called on the old task handle. So, we call
// notify on the new task handle
task.notify();
}
}
}
/// Notifies the task that last called `register`.
///
/// If `register` has not been called yet, then this does nothing.
pub fn notify(&self) {
let mut curr = WAITING;
loop {
if curr == LOCKED_WRITE {
// Transition the state to LOCKED_NOTIFIED
let actual = self.state.compare_and_swap(LOCKED_WRITE, LOCKED_WRITE_NOTIFIED, Release);
if curr == actual {
// Success, return
return;
}
// update current state variable and try again
curr = actual;
} else if curr == LOCKED_WRITE_NOTIFIED {
// Currently in `LOCKED_WRITE_NOTIFIED` state, nothing else to do.
return;
} else {
// Currently in a LOCKED_READ state, so attempt to increment the
// lock count.
let actual = self.state.compare_and_swap(curr, curr + 1, Acquire);
// Locked acquired
if actual == curr {
// Notify the task
unsafe {
if let Some(ref task) = *self.task.get() {
task.notify();
}
}
// Release the lock
self.state.fetch_sub(1, Release);
// Done
return;
}
// update current state variable and try again
curr = actual;
}
}
}
}
impl Default for AtomicTask {
fn default() -> Self {
AtomicTask::new()
}
}
impl fmt::Debug for AtomicTask {
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
write!(fmt, "AtomicTask")
}
}
unsafe impl Send for AtomicTask {}
unsafe impl Sync for AtomicTask {}
src/lib.rs
+1 -2
View File
@@ -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.
//!
src/reactor/background.rs
-212
View File
@@ -1,212 +0,0 @@
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.
#[derive(Debug)]
pub struct Background {
/// When `None`, the reactor thread will run until the process terminates.
inner: Option<Inner>,
}
/// Future that resolves when the reactor thread has shutdown.
#[derive(Debug)]
pub struct Shutdown {
inner: Inner,
}
/// Actual Background handle.
#[derive(Debug)]
struct Inner {
/// Handle to the reactor
handle: Handle,
/// Shared state between the background handle and the reactor thread.
shared: Arc<Shared>,
}
#[derive(Debug)]
struct Shared {
/// Signal the reactor thread to shutdown.
shutdown: AtomicUsize,
/// Task to notify when the reactor thread enters a shutdown state.
shutdown_task: AtomicTask,
}
/// Notifies the reactor thread to shutdown once the reactor becomes idle.
const SHUTDOWN_IDLE: usize = 1;
/// Notifies the reactor thread to shutdown immediately.
const SHUTDOWN_NOW: usize = 2;
/// The reactor is currently shutdown.
const SHUTDOWN: usize = 3;
// ===== impl Background =====
impl Background {
/// Launch a reactor in the background and return a handle to the thread.
pub fn new(reactor: Reactor) -> io::Result<Background> {
// Grab a handle to the reactor
let handle = reactor.handle().clone();
// Create the state shared between the background handle and the reactor
// thread.
let shared = Arc::new(Shared {
shutdown: AtomicUsize::new(0),
shutdown_task: AtomicTask::new(),
});
// For the reactor thread
let shared2 = shared.clone();
// Start the reactor thread
thread::Builder::new()
.spawn(move || run(reactor, shared2))?;
Ok(Background {
inner: Some(Inner {
handle,
shared,
}),
})
}
/// Returns a reference to the reactor handle.
pub fn handle(&self) -> &Handle {
&self.inner.as_ref().unwrap().handle
}
/// Shutdown the reactor on idle.
///
/// Returns a future that completes once the reactor thread has shutdown.
pub fn shutdown_on_idle(mut self) -> Shutdown {
let inner = self.inner.take().unwrap();
inner.shutdown_on_idle();
Shutdown { inner }
}
/// Shutdown the reactor immediately
///
/// Returns a future that completes once the reactor thread has shutdown.
pub fn shutdown_now(mut self) -> Shutdown {
let inner = self.inner.take().unwrap();
inner.shutdown_now();
Shutdown { inner }
}
/// Run the reactor on its thread until the process terminates.
pub fn forget(mut self) {
drop(self.inner.take());
}
}
impl Drop for Background {
fn drop(&mut self) {
let inner = match self.inner.take() {
Some(i) => i,
None => return,
};
inner.shutdown_now();
let shutdown = Shutdown { inner };
let _ = shutdown.wait();
}
}
// ===== impl Shutdown =====
impl Future for Shutdown {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<(), ()> {
self.inner.shared.shutdown_task.register();
if !self.inner.is_shutdown() {
return Ok(Async::NotReady);
}
Ok(().into())
}
}
// ===== impl Inner =====
impl Inner {
/// Returns true if the reactor thread is shutdown.
fn is_shutdown(&self) -> bool {
self.shared.shutdown.load(SeqCst) == SHUTDOWN
}
/// Notify the reactor thread to shutdown once the reactor transitions to an
/// idle state.
fn shutdown_on_idle(&self) {
self.shared.shutdown
.compare_and_swap(0, SHUTDOWN_IDLE, SeqCst);
self.handle.wakeup();
}
/// Notify the reactor thread to shutdown immediately.
fn shutdown_now(&self) {
let mut curr = self.shared.shutdown.load(SeqCst);
loop {
if curr >= SHUTDOWN_NOW {
return;
}
let act = self.shared.shutdown
.compare_and_swap(curr, SHUTDOWN_NOW, SeqCst);
if act == curr {
self.handle.wakeup();
return;
}
curr = act;
}
}
}
// ===== impl Reactor thread =====
fn run(mut reactor: Reactor, shared: Arc<Shared>) {
debug!("starting background reactor");
loop {
let shutdown = shared.shutdown.load(SeqCst);
if shutdown == SHUTDOWN_NOW {
debug!("shutting background reactor down NOW");
break;
}
if shutdown == SHUTDOWN_IDLE && reactor.is_idle() {
debug!("shutting background reactor on idle");
break;
}
reactor.turn(None).unwrap();
}
drop(reactor);
// Transition the state to shutdown
shared.shutdown.store(SHUTDOWN, SeqCst);
// Notify any waiters
shared.shutdown_task.notify();
debug!("background reactor has shutdown");
}
src/reactor/mod.rs
+11 -734
View File
@@ -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<Inner>,
_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<Inner>,
}
/// 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<Slab<ScheduledIo>>,
/// 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<Option<Handle>> = 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<T: Send + Sync>() {}
_assert::<Handle>();
}
// ===== 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<F, R>(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<Reactor> {
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<Duration>) -> io::Result<Turn> {
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> {
Background::new(self)
}
fn poll(&mut self, max_wait: Option<Duration>) -> 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<Handle> {
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<Handle> {
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::<Weak<Inner>, usize>(self.inner)
}
}
unsafe fn from_usize(val: usize) -> Handle {
let inner = mem::transmute::<usize, Weak<Inner>>(val);;
Handle { inner }
}
fn inner(&self) -> Option<Arc<Inner>> {
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<usize>
{
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()
}
}
src/reactor/poll_evented.rs
+263 -121
View File
@@ -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<E> {
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<E: fmt::Debug> fmt::Debug for PollEvented<E> {
@@ -91,20 +52,17 @@ impl<E> PollEvented<E> {
pub fn new(io: E, handle: &Handle) -> io::Result<PollEvented<E>>
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<E> PollEvented<E> {
/// 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<Ready> {
// 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<E> PollEvented<E> {
/// 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<E> PollEvented<E> {
/// This function will panic if called outside the context of a future's
/// task.
pub fn poll_ready(&mut self, mask: Ready) -> Async<Ready> {
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<E> PollEvented<E> {
/// 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<E> PollEvented<E> {
/// 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<E> PollEvented<E> {
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<T>(r: &io::Result<T>) -> 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()
}
}
src/reactor/poll_evented2.rs
-415
View File
@@ -1,415 +0,0 @@
#![allow(warnings)]
use reactor::Handle;
use reactor::registration::Registration;
use futures::{task, Async, Poll};
use mio;
use mio::event::Evented;
use tokio_io::{AsyncRead, AsyncWrite};
use std::fmt;
use std::io::{self, Read, Write};
use std::sync::atomic::AtomicUsize;
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
/// 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`
/// type provides [`AsyncRead`] and [`AsyncWrite`] implementations using the
/// underlying I/O resource as well as readiness events provided by the reactor.
///
/// **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.
///
/// ## Readiness events
///
/// 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.
///
/// 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.
///
/// 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.
///
/// This allows the caller to implement additional funcitons. For example,
/// [`TcpListener`] implements accept by using [`poll_read_ready`] and
/// [`need_read`].
///
/// ```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())
/// }
///
/// 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,
}
struct Inner {
registration: Registration,
/// Currently visible read readiness
read_readiness: AtomicUsize,
/// Currently visible write readiness
write_readiness: AtomicUsize,
}
// ===== impl PollEvented =====
impl<E> PollEvented<E>
where E: Evented
{
/// Creates a new `PollEvented` associated with the default reactor.
pub fn new(io: E) -> PollEvented<E> {
PollEvented {
io: io,
inner: Inner {
registration: Registration::new(),
read_readiness: AtomicUsize::new(0),
write_readiness: AtomicUsize::new(0),
}
}
}
/// Creates a new `PollEvented` associated with the specified reactor.
pub fn new_with_handle(io: E, handle: &Handle) -> io::Result<Self> {
let ret = PollEvented::new(io);
ret.inner.registration.register_with(&ret.io, handle)?;
Ok(ret)
}
/// Check the I/O resource's read readiness state.
///
/// 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 from outside of a task context.
pub fn poll_read_ready(&self) -> Poll<mio::Ready, io::Error> {
self.register()?;
// 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()? {
n |= super::ready2usize(ready);
self.inner.read_readiness.store(n, Relaxed);
}
return Ok(super::usize2ready(n).into());
}
}
let ready = try_ready!(self.inner.registration.poll_read_ready());
// Cache the value
self.inner.read_readiness.store(super::ready2usize(ready), Relaxed);
Ok(ready.into())
}
/// Resets the I/O resource's read readiness state and registers the current
/// task to be notified once a read readiness event is received.
///
/// After calling this function, `poll_read_ready` will return `NotReady`
/// until a new read readiness event has been received.
///
/// This function clears **all** readiness state **except** write readiness.
/// This includes any platform-specific readiness bits.
///
/// # Panics
///
/// 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);
if self.poll_read_ready()?.is_ready() {
// Notify the current task
task::current().notify();
}
Ok(())
}
/// Check the I/O resource's write readiness state.
///
/// 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 from outside of a task context.
pub fn poll_write_ready(&self) -> Poll<mio::Ready, io::Error> {
self.register()?;
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()? {
n |= super::ready2usize(ready);
self.inner.write_readiness.store(n, Relaxed);
}
return Ok(super::usize2ready(n).into());
}
}
let ready = try_ready!(self.inner.registration.poll_write_ready());
// Cache the value
self.inner.write_readiness.store(super::ready2usize(ready), Relaxed);
Ok(ready.into())
}
/// Resets the I/O resource's write readiness state and registers the current
/// task to be notified once a write readiness event is received.
///
/// 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 from outside of a task context.
pub fn need_write(&self) -> io::Result<()> {
self.inner.write_readiness.store(0, Relaxed);
if self.poll_write_ready()?.is_ready() {
// Notify the current task
task::current().notify();
}
Ok(())
}
/// Ensure that the I/O resource is registered with the reactor.
fn register(&self) -> io::Result<()> {
self.inner.registration.register(&self.io)?;
Ok(())
}
}
impl<E> PollEvented<E> {
/// Returns a shared reference to the underlying I/O object this readiness
/// stream is wrapping.
pub fn get_ref(&self) -> &E {
&self.io
}
/// Returns a mutable reference to the underlying I/O object this readiness
/// stream is wrapping.
pub fn get_mut(&mut self) -> &mut E {
&mut self.io
}
/// Consumes self, returning the inner I/O object
pub fn into_inner(self) -> E {
self.io
}
}
// ===== Read / Write impls =====
impl<E> Read for PollEvented<E>
where E: Evented + Read,
{
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
if let Async::NotReady = self.poll_read_ready()? {
return Err(io::ErrorKind::WouldBlock.into())
}
let r = self.get_mut().read(buf);
if is_wouldblock(&r) {
self.need_read()?;
}
return r
}
}
impl<E> Write for PollEvented<E>
where E: Evented + Write,
{
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
if let Async::NotReady = self.poll_write_ready()? {
return Err(io::ErrorKind::WouldBlock.into())
}
let r = self.get_mut().write(buf);
if is_wouldblock(&r) {
self.need_write()?;
}
return r
}
fn flush(&mut self) -> io::Result<()> {
if let Async::NotReady = self.poll_write_ready()? {
return Err(io::ErrorKind::WouldBlock.into())
}
let r = self.get_mut().flush();
if is_wouldblock(&r) {
self.need_write()?;
}
return r
}
}
impl<E> AsyncRead for PollEvented<E>
where E: Evented + Read,
{
}
impl<E> AsyncWrite for PollEvented<E>
where E: Evented + Write,
{
fn shutdown(&mut self) -> Poll<(), io::Error> {
Ok(().into())
}
}
// ===== &'a Read / &'a Write impls =====
impl<'a, E> Read for &'a PollEvented<E>
where E: Evented, &'a E: Read,
{
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
if let Async::NotReady = self.poll_read_ready()? {
return Err(io::ErrorKind::WouldBlock.into())
}
let r = self.get_ref().read(buf);
if is_wouldblock(&r) {
self.need_read()?;
}
return r
}
}
impl<'a, E> Write for &'a PollEvented<E>
where E: Evented, &'a E: Write,
{
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
if let Async::NotReady = self.poll_write_ready()? {
return Err(io::ErrorKind::WouldBlock.into())
}
let r = self.get_ref().write(buf);
if is_wouldblock(&r) {
self.need_write()?;
}
return r
}
fn flush(&mut self) -> io::Result<()> {
if let Async::NotReady = self.poll_write_ready()? {
return Err(io::ErrorKind::WouldBlock.into())
}
let r = self.get_ref().flush();
if is_wouldblock(&r) {
self.need_write()?;
}
return r
}
}
impl<'a, E> AsyncRead for &'a PollEvented<E>
where E: Evented, &'a E: Read,
{
}
impl<'a, E> AsyncWrite for &'a PollEvented<E>
where E: Evented, &'a E: Write,
{
fn shutdown(&mut self) -> Poll<(), io::Error> {
Ok(().into())
}
}
fn is_wouldblock<T>(r: &io::Result<T>) -> bool {
match *r {
Ok(_) => false,
Err(ref e) => e.kind() == io::ErrorKind::WouldBlock,
}
}
impl<E: fmt::Debug> fmt::Debug for PollEvented<E> {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("PollEvented")
.field("io", &self.io)
.finish()
}
}
src/reactor/registration.rs
-478
View File
@@ -1,478 +0,0 @@
use reactor::{Handle, Direction};
use futures::{Async, Poll};
use futures::task::{self, Task};
use mio::{self, Evented};
use std::{io, mem, usize};
use std::cell::UnsafeCell;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
/// 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. This is the lowest
/// level API for integrating with a reactor.
///
/// 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.
///
/// 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.
///
/// Setting this requires acquiring the lock from state.
inner: UnsafeCell<Option<Inner>>,
/// Tracks the state of the registration.
///
/// The least significant 2 bits are used to track the lifecycle of the
/// registration. The rest of the `state` variable is a pointer to tasks
/// that must be notified once the lock is released.
state: AtomicUsize,
}
#[derive(Debug)]
struct Inner {
handle: Handle,
token: usize,
}
/// Tasks waiting on readiness notifications.
#[derive(Debug)]
struct Node {
direction: Direction,
task: Task,
next: Option<Box<Node>>,
}
/// Initial state. The handle is not set and the registration is idle.
const INIT: usize = 0;
/// A thread locked the state and will associate a handle.
const LOCKED: usize = 1;
/// A handle has been associated with the registration.
const READY: usize = 2;
/// Masks the lifecycle state
const LIFECYCLE_MASK: usize = 0b11;
/// A fake token used to identify error situations
const ERROR: usize = usize::MAX;
// ===== impl Registration =====
impl Registration {
/// Create a new `Registration`.
///
/// This registration is not associated with a Reactor instance. Call
/// `register` to establish the association.
pub fn new() -> Registration {
Registration {
inner: UnsafeCell::new(None),
state: AtomicUsize::new(INIT),
}
}
/// Register the I/O resource with the default reactor.
///
/// This function is safe to call concurrently and repeatedly. However, only
/// the first call will establish the registration. Subsequent calls will be
/// no-ops.
///
/// If the registration happened successfully, `Ok(true)` is returned.
///
/// If an I/O resource has previously been successfully registered,
/// `Ok(false)` is returned.
///
/// If an error is encountered during registration, `Err` is returned.
pub fn register<T>(&self, io: &T) -> io::Result<bool>
where T: Evented,
{
self.register2(io, || Handle::try_current())
}
/// Register the I/O resource with the specified reactor.
///
/// This function is safe to call concurrently and repeatedly. However, only
/// the first call will establish the registration. Subsequent calls will be
/// no-ops.
///
/// If the registration happened successfully, `Ok(true)` is returned.
///
/// If an I/O resource has previously been successfully registered,
/// `Ok(false)` is returned.
///
/// If an error is encountered during registration, `Err` is returned.
pub fn register_with<T>(&self, io: &T, handle: &Handle) -> io::Result<bool>
where T: Evented,
{
self.register2(io, || Ok(handle.clone()))
}
fn register2<T, F>(&self, io: &T, f: F) -> io::Result<bool>
where T: Evented,
F: Fn() -> io::Result<Handle>,
{
let mut state = self.state.load(SeqCst);
loop {
match state {
INIT => {
// Registration is currently not associated with a handle.
// Get a handle then attempt to lock the state.
let handle = f()?;
let actual = self.state.compare_and_swap(INIT, LOCKED, SeqCst);
if actual != state {
state = actual;
continue;
}
// Create the actual registration
let (inner, res) = Inner::new(io, handle);
unsafe { *self.inner.get() = Some(inner); }
// Transition out of the locked state. This acquires the
// current value, potentially having a list of tasks that
// are pending readiness notifications.
let actual = self.state.swap(READY, SeqCst);
// Consume the stack of nodes.
let ptr = actual & !LIFECYCLE_MASK;
if ptr != 0 {
let mut read = false;
let mut write = false;
let mut curr = unsafe { Box::from_raw(ptr as *mut Node) };
let inner = unsafe { (*self.inner.get()).as_ref().unwrap() };
loop {
let node = *curr;
let Node {
direction,
task,
next,
} = node;
let flag = match direction {
Direction::Read => &mut read,
Direction::Write => &mut write,
};
if !*flag {
*flag = true;
inner.register(direction, task);
}
match next {
Some(next) => curr = next,
None => break,
}
}
}
return res.map(|_| true);
}
_ => return Ok(false),
}
}
}
/// 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 {
Some(v) => Async::Ready(v),
_ => Async::NotReady,
})
}
/// Consume any pending read readiness event.
///
/// 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 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 {
Some(v) => Async::Ready(v),
_ => Async::NotReady,
})
}
/// Consume any pending write readiness event.
///
/// 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)
}
fn poll_ready(&self, direction: Direction, notify: bool)
-> io::Result<Option<mio::Ready>>
{
let mut state = self.state.load(SeqCst);
// Cache the node pointer
let mut node = None;
loop {
match state {
INIT => {
return Err(io::Error::new(io::ErrorKind::Other, "must call `register`
before poll_read_ready"));
}
READY => {
let inner = unsafe { (*self.inner.get()).as_ref().unwrap() };
return inner.poll_ready(direction, notify);
}
_ => {
if !notify {
// Skip the notification tracking junk.
return Ok(None);
}
let ptr = state & !LIFECYCLE_MASK;
// Get the node
let mut n = node.take().unwrap_or_else(|| {
Box::new(Node {
direction,
task: task::current(),
next: None,
})
});
n.next = if ptr == 0 {
None
} else {
// Great care must be taken of the CAS fails
Some(unsafe { Box::from_raw(ptr as *mut Node) })
};
let ptr = Box::into_raw(n);
let next = ptr as usize | (state & LIFECYCLE_MASK);
let actual = self.state.compare_and_swap(state, next, SeqCst);
if actual != state {
// Back out of the node boxing
let mut n = unsafe { Box::from_raw(ptr) };
// We don't really own this
mem::forget(n.next.take());
// Save this for next loop
node = Some(n);
state = actual;
continue;
}
return Ok(None);
}
}
}
}
}
unsafe impl Send for Registration {}
unsafe impl Sync for Registration {}
// ===== impl Inner =====
impl Inner {
fn new<T>(io: &T, handle: Handle) -> (Self, io::Result<()>)
where T: Evented,
{
let mut res = Ok(());
let token = match handle.inner() {
Some(inner) => match inner.add_source(io) {
Ok(token) => token,
Err(e) => {
res = Err(e);
ERROR
}
},
None => {
res = Err(io::Error::new(io::ErrorKind::Other, "event loop gone"));
ERROR
}
};
let inner = Inner {
handle,
token,
};
(inner, res)
}
fn register(&self, direction: Direction, task: Task) {
if self.token == ERROR {
task.notify();
return;
}
let inner = match self.handle.inner() {
Some(inner) => inner,
None => {
task.notify();
return;
}
};
inner.register(self.token, direction, task);
}
fn poll_ready(&self, direction: Direction, notify: bool)
-> io::Result<Option<mio::Ready>>
{
if self.token == ERROR {
return Err(io::Error::new(io::ErrorKind::Other, "failed to associate with reactor"));
}
let inner = match self.handle.inner() {
Some(inner) => inner,
None => return Err(io::Error::new(io::ErrorKind::Other, "reactor gone")),
};
let mask = direction.mask();
let io_dispatch = inner.io_dispatch.read().unwrap();
let sched = &io_dispatch[self.token];
let mut ready = mask & sched.readiness.fetch_and(!mask, SeqCst);
if ready == 0 && notify {
// Update the task info
match direction {
Direction::Read => sched.reader.register(),
Direction::Write => sched.writer.register(),
}
// Try again
ready = mask & sched.readiness.fetch_and(!mask, SeqCst);
}
if ready == 0 {
Ok(None)
} else {
Ok(Some(super::usize2ready(ready)))
}
}
}
impl Drop for Inner {
fn drop(&mut self) {
if self.token == ERROR {
return;
}
let inner = match self.handle.inner() {
Some(inner) => inner,
None => return,
};
inner.drop_source(self.token);
}
}
src/runtime.rs
+2 -3
View File
@@ -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();
});
})