mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-09-07 00:00:08 +02:00
reactor: rename tokio-reactor -> tokio-net (#1450)
* reactor: rename tokio-reactor -> tokio-net This is in preparation for #1264
This commit is contained in:
Carl Lerche
@@ -0,0 +1,613 @@
|
||||
#![doc(html_root_url = "https://docs.rs/tokio-net/0.2.0-alpha.1")]
|
||||
#![warn(
|
||||
missing_debug_implementations,
|
||||
missing_docs,
|
||||
rust_2018_idioms,
|
||||
unreachable_pub
|
||||
)]
|
||||
#![doc(test(no_crate_inject, attr(deny(rust_2018_idioms))))]
|
||||
|
||||
//! Event loop that drives Tokio I/O resources.
|
||||
//!
|
||||
//! The reactor is the engine that drives asynchronous I/O resources (like TCP and
|
||||
//! UDP sockets). It is backed by [`mio`] and acts as a bridge between [`mio`] and
|
||||
//! [`futures`].
|
||||
//!
|
||||
//! The crate provides:
|
||||
//!
|
||||
//! * [`Reactor`] is the main type of this crate. It performs the event loop logic.
|
||||
//!
|
||||
//! * [`Handle`] provides a reference to a reactor instance.
|
||||
//!
|
||||
//! * [`Registration`] and [`PollEvented`] allow third parties to implement I/O
|
||||
//! resources that are driven by the reactor.
|
||||
//!
|
||||
//! Application authors will not use this crate directly. Instead, they will use the
|
||||
//! `tokio` crate. Library authors should only depend on `tokio-net` if they
|
||||
//! are building a custom I/O resource.
|
||||
//!
|
||||
//! For more details, see [reactor module] documentation in the Tokio crate.
|
||||
//!
|
||||
//! [`mio`]: http://github.com/carllerche/mio
|
||||
//! [`futures`]: http://github.com/rust-lang-nursery/futures-rs
|
||||
//! [`Reactor`]: struct.Reactor.html
|
||||
//! [`Handle`]: struct.Handle.html
|
||||
//! [`Registration`]: struct.Registration.html
|
||||
//! [`PollEvented`]: struct.PollEvented.html
|
||||
//! [reactor module]: https://docs.rs/tokio/0.1/tokio/reactor/index.html
|
||||
|
||||
mod poll_evented;
|
||||
mod registration;
|
||||
mod sharded_rwlock;
|
||||
|
||||
// ===== Public re-exports =====
|
||||
|
||||
pub use self::poll_evented::PollEvented;
|
||||
pub use self::registration::Registration;
|
||||
|
||||
// ===== Private imports =====
|
||||
|
||||
use crate::sharded_rwlock::RwLock;
|
||||
use log::{debug, log_enabled, trace, Level};
|
||||
use mio::event::Evented;
|
||||
use slab::Slab;
|
||||
use std::cell::RefCell;
|
||||
use std::io;
|
||||
#[cfg(all(unix, not(target_os = "fuchsia")))]
|
||||
use std::os::unix::io::{AsRawFd, RawFd};
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::atomic::Ordering::{Relaxed, SeqCst};
|
||||
use std::sync::{Arc, Weak};
|
||||
use std::task::Waker;
|
||||
use std::time::{Duration, Instant};
|
||||
use std::{fmt, usize};
|
||||
use tokio_executor::park::{Park, Unpark};
|
||||
use tokio_sync::AtomicWaker;
|
||||
|
||||
/// 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.
|
||||
///
|
||||
/// By default, most components bind lazily to reactors.
|
||||
/// To get this behavior when manually passing a `Handle`, use `default()`.
|
||||
#[derive(Clone)]
|
||||
pub struct Handle {
|
||||
inner: Option<HandlePriv>,
|
||||
}
|
||||
|
||||
/// Like `Handle`, but never `None`.
|
||||
#[derive(Clone)]
|
||||
struct HandlePriv {
|
||||
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: (),
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_handle_size() {
|
||||
use std::mem;
|
||||
assert_eq!(mem::size_of::<Handle>(), mem::size_of::<HandlePriv>());
|
||||
}
|
||||
|
||||
struct Inner {
|
||||
/// The underlying system event queue.
|
||||
io: mio::Poll,
|
||||
|
||||
/// ABA guard counter
|
||||
next_aba_guard: AtomicUsize,
|
||||
|
||||
/// 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 {
|
||||
aba_guard: usize,
|
||||
readiness: AtomicUsize,
|
||||
reader: AtomicWaker,
|
||||
writer: AtomicWaker,
|
||||
}
|
||||
|
||||
#[derive(Debug, Eq, PartialEq, Clone, Copy)]
|
||||
pub(crate) enum Direction {
|
||||
Read,
|
||||
Write,
|
||||
}
|
||||
|
||||
thread_local! {
|
||||
/// Tracks the reactor for the current execution context.
|
||||
static CURRENT_REACTOR: RefCell<Option<HandlePriv>> = RefCell::new(None)
|
||||
}
|
||||
|
||||
const TOKEN_SHIFT: usize = 22;
|
||||
|
||||
// Kind of arbitrary, but this reserves some token space for later usage.
|
||||
const MAX_SOURCES: usize = (1 << TOKEN_SHIFT) - 1;
|
||||
const TOKEN_WAKEUP: mio::Token = mio::Token(MAX_SOURCES);
|
||||
|
||||
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 fn with_default<F, R>(handle: &Handle, f: F) -> R
|
||||
where
|
||||
F: FnOnce() -> 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"
|
||||
);
|
||||
|
||||
let handle = match handle.as_priv() {
|
||||
Some(handle) => handle,
|
||||
None => {
|
||||
panic!("`handle` does not reference a reactor");
|
||||
}
|
||||
};
|
||||
|
||||
*current = Some(handle.clone());
|
||||
}
|
||||
|
||||
f()
|
||||
})
|
||||
}
|
||||
|
||||
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,
|
||||
next_aba_guard: AtomicUsize::new(0),
|
||||
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: Some(HandlePriv {
|
||||
inner: Arc::downgrade(&self.inner),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// 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().is_empty()
|
||||
}
|
||||
|
||||
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(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();
|
||||
trace!(
|
||||
"loop process - {} events, {}.{:03}s",
|
||||
events,
|
||||
dur.as_secs(),
|
||||
dur.subsec_millis()
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn dispatch(&self, token: mio::Token, ready: mio::Ready) {
|
||||
let aba_guard = token.0 & !MAX_SOURCES;
|
||||
let token = token.0 & MAX_SOURCES;
|
||||
|
||||
let mut rd = None;
|
||||
let mut wr = None;
|
||||
|
||||
// Create a scope to ensure that notifying the tasks stays out of the
|
||||
// lock's critical section.
|
||||
{
|
||||
let io_dispatch = self.inner.io_dispatch.read();
|
||||
|
||||
let io = match io_dispatch.get(token) {
|
||||
Some(io) => io,
|
||||
None => return,
|
||||
};
|
||||
|
||||
if aba_guard != io.aba_guard {
|
||||
return;
|
||||
}
|
||||
|
||||
io.readiness.fetch_or(ready.as_usize(), Relaxed);
|
||||
|
||||
if ready.is_writable() || platform::is_hup(ready) {
|
||||
wr = io.writer.take_waker();
|
||||
}
|
||||
|
||||
if !(ready & (!mio::Ready::writable())).is_empty() {
|
||||
rd = io.reader.take_waker();
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(w) = rd {
|
||||
w.wake();
|
||||
}
|
||||
|
||||
if let Some(w) = wr {
|
||||
w.wake();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(unix, not(target_os = "fuchsia")))]
|
||||
impl AsRawFd for Reactor {
|
||||
fn as_raw_fd(&self) -> RawFd {
|
||||
self.inner.io.as_raw_fd()
|
||||
}
|
||||
}
|
||||
|
||||
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 {
|
||||
#[doc(hidden)]
|
||||
#[deprecated(note = "semantics were sometimes surprising, use Handle::default()")]
|
||||
pub fn current() -> Handle {
|
||||
// TODO: Should this panic on error?
|
||||
HandlePriv::try_current()
|
||||
.map(|handle| Handle {
|
||||
inner: Some(handle),
|
||||
})
|
||||
.unwrap_or(Handle {
|
||||
inner: Some(HandlePriv { inner: Weak::new() }),
|
||||
})
|
||||
}
|
||||
|
||||
fn as_priv(&self) -> Option<&HandlePriv> {
|
||||
self.inner.as_ref()
|
||||
}
|
||||
}
|
||||
|
||||
impl Unpark for Handle {
|
||||
fn unpark(&self) {
|
||||
if let Some(ref h) = self.inner {
|
||||
h.wakeup();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for Handle {
|
||||
/// Returns a "default" handle, i.e., a handle that lazily binds to a reactor.
|
||||
fn default() -> Handle {
|
||||
Handle { inner: None }
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Debug for Handle {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "Handle")
|
||||
}
|
||||
}
|
||||
|
||||
// ===== impl HandlePriv =====
|
||||
|
||||
impl HandlePriv {
|
||||
/// Try to get a handle to the current reactor.
|
||||
///
|
||||
/// Returns `Err` if no handle is found.
|
||||
pub(crate) fn try_current() -> io::Result<HandlePriv> {
|
||||
CURRENT_REACTOR.with(|current| match *current.borrow() {
|
||||
Some(ref handle) => Ok(handle.clone()),
|
||||
None => Err(io::Error::new(io::ErrorKind::Other, "no current reactor")),
|
||||
})
|
||||
}
|
||||
|
||||
/// 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 inner(&self) -> Option<Arc<Inner>> {
|
||||
self.inner.upgrade()
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Debug for HandlePriv {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "HandlePriv")
|
||||
}
|
||||
}
|
||||
|
||||
// ===== impl Inner =====
|
||||
|
||||
impl Inner {
|
||||
/// Register an I/O resource with the reactor.
|
||||
///
|
||||
/// The registration token is returned.
|
||||
fn add_source(&self, source: &dyn Evented) -> io::Result<usize> {
|
||||
// Get an ABA guard value
|
||||
let aba_guard = self.next_aba_guard.fetch_add(1 << TOKEN_SHIFT, Relaxed);
|
||||
|
||||
let key = {
|
||||
// Block to contain the write lock
|
||||
let mut io_dispatch = self.io_dispatch.write();
|
||||
|
||||
if io_dispatch.len() == MAX_SOURCES {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"reactor at max \
|
||||
registered I/O resources",
|
||||
));
|
||||
}
|
||||
|
||||
io_dispatch.insert(ScheduledIo {
|
||||
aba_guard,
|
||||
readiness: AtomicUsize::new(0),
|
||||
reader: AtomicWaker::new(),
|
||||
writer: AtomicWaker::new(),
|
||||
})
|
||||
};
|
||||
|
||||
let token = aba_guard | key;
|
||||
debug!("adding I/O source: {}", token);
|
||||
|
||||
self.io.register(
|
||||
source,
|
||||
mio::Token(token),
|
||||
mio::Ready::all(),
|
||||
mio::PollOpt::edge(),
|
||||
)?;
|
||||
|
||||
Ok(key)
|
||||
}
|
||||
|
||||
/// Deregisters an I/O resource from the reactor.
|
||||
fn deregister_source(&self, source: &dyn Evented) -> io::Result<()> {
|
||||
self.io.deregister(source)
|
||||
}
|
||||
|
||||
fn drop_source(&self, token: usize) {
|
||||
debug!("dropping I/O source: {}", token);
|
||||
self.io_dispatch.write().remove(token);
|
||||
}
|
||||
|
||||
/// Registers interest in the I/O resource associated with `token`.
|
||||
fn register(&self, token: usize, dir: Direction, w: Waker) {
|
||||
debug!("scheduling {:?} for: {}", dir, token);
|
||||
let io_dispatch = self.io_dispatch.read();
|
||||
let sched = io_dispatch.get(token).unwrap();
|
||||
|
||||
let (waker, ready) = match dir {
|
||||
Direction::Read => (&sched.reader, !mio::Ready::writable()),
|
||||
Direction::Write => (&sched.writer, mio::Ready::writable()),
|
||||
};
|
||||
|
||||
waker.register(w);
|
||||
|
||||
if sched.readiness.load(SeqCst) & ready.as_usize() != 0 {
|
||||
waker.wake();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
for (_, io) in io.iter() {
|
||||
io.writer.wake();
|
||||
io.reader.wake();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Direction {
|
||||
fn mask(self) -> mio::Ready {
|
||||
match self {
|
||||
Direction::Read => {
|
||||
// Everything except writable is signaled through read.
|
||||
mio::Ready::all() - mio::Ready::writable()
|
||||
}
|
||||
Direction::Write => mio::Ready::writable() | platform::hup(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
mod platform {
|
||||
use mio::unix::UnixReady;
|
||||
use mio::Ready;
|
||||
|
||||
pub(crate) fn hup() -> Ready {
|
||||
UnixReady::hup().into()
|
||||
}
|
||||
|
||||
pub(crate) fn is_hup(ready: Ready) -> bool {
|
||||
UnixReady::from(ready).is_hup()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
mod platform {
|
||||
use mio::Ready;
|
||||
|
||||
pub(crate) fn hup() -> Ready {
|
||||
Ready::empty()
|
||||
}
|
||||
|
||||
pub(crate) fn is_hup(_: Ready) -> bool {
|
||||
false
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,440 @@
|
||||
use crate::{Handle, Registration};
|
||||
|
||||
use tokio_io::{AsyncRead, AsyncWrite};
|
||||
|
||||
use futures_core::ready;
|
||||
use mio;
|
||||
use mio::event::Evented;
|
||||
use std::fmt;
|
||||
use std::io::{self, Read, Write};
|
||||
use std::marker::Unpin;
|
||||
use std::pin::Pin;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::atomic::Ordering::Relaxed;
|
||||
use std::task::{Context, Poll};
|
||||
|
||||
/// Associates an I/O resource that implements the [`std::io::Read`] and/or
|
||||
/// [`std::io::Write`] traits with the reactor that drives it.
|
||||
///
|
||||
/// `PollEvented` uses [`Registration`] internally to take a type that
|
||||
/// implements [`mio::Evented`] as well as [`std::io::Read`] and or
|
||||
/// [`std::io::Write`] and associate it with a reactor that will drive it.
|
||||
///
|
||||
/// Once the [`mio::Evented`] type is wrapped by `PollEvented`, it can be
|
||||
/// used from within the future's execution model. As such, the `PollEvented`
|
||||
/// type provides [`AsyncRead`] and [`AsyncWrite`] implementations using the
|
||||
/// underlying I/O resource as well as readiness events provided by the reactor.
|
||||
///
|
||||
/// **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 concurrently. 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 [`clear_read_ready`] or
|
||||
/// [`clear_write_ready`]. This clears the readiness state until a new readiness
|
||||
/// event is received.
|
||||
///
|
||||
/// This allows the caller to implement additional functions. For example,
|
||||
/// [`TcpListener`] implements poll_accept by using [`poll_read_ready`] and
|
||||
/// [`clear_read_ready`].
|
||||
///
|
||||
/// ```rust
|
||||
/// use tokio_net::PollEvented;
|
||||
///
|
||||
/// use futures_core::ready;
|
||||
/// use mio::Ready;
|
||||
/// use mio::net::{TcpStream, TcpListener};
|
||||
/// use std::io;
|
||||
/// use std::task::{Context, Poll};
|
||||
///
|
||||
/// struct MyListener {
|
||||
/// poll_evented: PollEvented<TcpListener>,
|
||||
/// }
|
||||
///
|
||||
/// impl MyListener {
|
||||
/// pub fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<Result<TcpStream, io::Error>> {
|
||||
/// let ready = Ready::readable();
|
||||
///
|
||||
/// ready!(self.poll_evented.poll_read_ready(cx, ready))?;
|
||||
///
|
||||
/// match self.poll_evented.get_ref().accept() {
|
||||
/// Ok((socket, _)) => Poll::Ready(Ok(socket)),
|
||||
/// Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
|
||||
/// self.poll_evented.clear_read_ready(cx, ready);
|
||||
/// Poll::Pending
|
||||
/// }
|
||||
/// Err(e) => Poll::Ready(Err(e)),
|
||||
/// }
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// ## Platform-specific events
|
||||
///
|
||||
/// `PollEvented` also allows receiving platform-specific `mio::Ready` events.
|
||||
/// These events are included as part of the read readiness event stream. The
|
||||
/// write readiness event stream is only for `Ready::writable()` events.
|
||||
///
|
||||
/// [`std::io::Read`]: https://doc.rust-lang.org/std/io/trait.Read.html
|
||||
/// [`std::io::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
|
||||
/// [`clear_read_ready`]: #method.clear_read_ready
|
||||
/// [`clear_write_ready`]: #method.clear_write_ready
|
||||
/// [`poll_read_ready`]: #method.poll_read_ready
|
||||
/// [`poll_write_ready`]: #method.poll_write_ready
|
||||
pub struct PollEvented<E: Evented> {
|
||||
io: Option<E>,
|
||||
inner: Inner,
|
||||
}
|
||||
|
||||
struct Inner {
|
||||
registration: Registration,
|
||||
|
||||
/// Currently visible read readiness
|
||||
read_readiness: AtomicUsize,
|
||||
|
||||
/// Currently visible write readiness
|
||||
write_readiness: AtomicUsize,
|
||||
}
|
||||
|
||||
// ===== impl PollEvented =====
|
||||
|
||||
macro_rules! poll_ready {
|
||||
($me:expr, $mask:expr, $cache:ident, $take:ident, $poll:expr) => {{
|
||||
$me.register()?;
|
||||
|
||||
// Load cached & encoded readiness.
|
||||
let mut cached = $me.inner.$cache.load(Relaxed);
|
||||
let mask = $mask | crate::platform::hup();
|
||||
|
||||
// See if the current readiness matches any bits.
|
||||
let mut ret = mio::Ready::from_usize(cached) & $mask;
|
||||
|
||||
if ret.is_empty() {
|
||||
// Readiness does not match, consume the registration's readiness
|
||||
// stream. This happens in a loop to ensure that the stream gets
|
||||
// drained.
|
||||
loop {
|
||||
let ready = match $poll? {
|
||||
Poll::Ready(v) => v,
|
||||
Poll::Pending => return Poll::Pending,
|
||||
};
|
||||
cached |= ready.as_usize();
|
||||
|
||||
// Update the cache store
|
||||
$me.inner.$cache.store(cached, Relaxed);
|
||||
|
||||
ret |= ready & mask;
|
||||
|
||||
if !ret.is_empty() {
|
||||
return Poll::Ready(Ok(ret));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Check what's new with the registration stream. This will not
|
||||
// request to be notified
|
||||
if let Some(ready) = $me.inner.registration.$take()? {
|
||||
cached |= ready.as_usize();
|
||||
$me.inner.$cache.store(cached, Relaxed);
|
||||
}
|
||||
|
||||
Poll::Ready(Ok(mio::Ready::from_usize(cached)))
|
||||
}
|
||||
}};
|
||||
}
|
||||
|
||||
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: Some(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);
|
||||
|
||||
if let Some(handle) = handle.as_priv() {
|
||||
ret.inner
|
||||
.registration
|
||||
.register_with_priv(ret.io.as_ref().unwrap(), handle)?;
|
||||
}
|
||||
|
||||
Ok(ret)
|
||||
}
|
||||
|
||||
/// Returns a shared reference to the underlying I/O object this readiness
|
||||
/// stream is wrapping.
|
||||
pub fn get_ref(&self) -> &E {
|
||||
self.io.as_ref().unwrap()
|
||||
}
|
||||
|
||||
/// Returns a mutable reference to the underlying I/O object this readiness
|
||||
/// stream is wrapping.
|
||||
pub fn get_mut(&mut self) -> &mut E {
|
||||
self.io.as_mut().unwrap()
|
||||
}
|
||||
|
||||
/// Consumes self, returning the inner I/O object
|
||||
///
|
||||
/// This function will deregister the I/O resource from the reactor before
|
||||
/// returning. If the deregistration operation fails, an error is returned.
|
||||
///
|
||||
/// Note that deregistering does not guarantee that the I/O resource can be
|
||||
/// registered with a different reactor. Some I/O resource types can only be
|
||||
/// associated with a single reactor instance for their lifetime.
|
||||
pub fn into_inner(mut self) -> io::Result<E> {
|
||||
let io = self.io.take().unwrap();
|
||||
self.inner.registration.deregister(&io)?;
|
||||
Ok(io)
|
||||
}
|
||||
|
||||
/// Check the I/O resource's read readiness state.
|
||||
///
|
||||
/// The mask argument allows specifying what readiness to notify on. This
|
||||
/// can be any value, including platform specific readiness, **except**
|
||||
/// `writable`. HUP is always implicitly included on platforms that support
|
||||
/// it.
|
||||
///
|
||||
/// If the resource is not ready for a read then `Poll::Pending` 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 [`clear_read_ready`].
|
||||
///
|
||||
/// [`clear_read_ready`]: #method.clear_read_ready
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if:
|
||||
///
|
||||
/// * `ready` includes writable.
|
||||
/// * called from outside of a task context.
|
||||
pub fn poll_read_ready(
|
||||
&self,
|
||||
cx: &mut Context<'_>,
|
||||
mask: mio::Ready,
|
||||
) -> Poll<io::Result<mio::Ready>> {
|
||||
assert!(!mask.is_writable(), "cannot poll for write readiness");
|
||||
poll_ready!(
|
||||
self,
|
||||
mask,
|
||||
read_readiness,
|
||||
take_read_ready,
|
||||
self.inner.registration.poll_read_ready(cx)
|
||||
)
|
||||
}
|
||||
|
||||
/// Clears 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
|
||||
/// `Poll::Pending` until a new read readiness event has been received.
|
||||
///
|
||||
/// The `mask` argument specifies the readiness bits to clear. This may not
|
||||
/// include `writable` or `hup`.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if:
|
||||
///
|
||||
/// * `ready` includes writable or HUP
|
||||
/// * called from outside of a task context.
|
||||
pub fn clear_read_ready(&self, cx: &mut Context<'_>, ready: mio::Ready) -> io::Result<()> {
|
||||
// Cannot clear write readiness
|
||||
assert!(!ready.is_writable(), "cannot clear write readiness");
|
||||
assert!(
|
||||
!crate::platform::is_hup(ready),
|
||||
"cannot clear HUP readiness"
|
||||
);
|
||||
|
||||
self.inner
|
||||
.read_readiness
|
||||
.fetch_and(!ready.as_usize(), Relaxed);
|
||||
|
||||
if self.poll_read_ready(cx, ready)?.is_ready() {
|
||||
// Notify the current task
|
||||
cx.waker().wake_by_ref();
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Check the I/O resource's write readiness state.
|
||||
///
|
||||
/// This always checks for writable readiness and also checks for HUP
|
||||
/// readiness on platforms that support it.
|
||||
///
|
||||
/// 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 [`clear_write_ready`].
|
||||
///
|
||||
/// [`clear_write_ready`]: #method.clear_write_ready
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function panics if:
|
||||
///
|
||||
/// * `ready` contains bits besides `writable` and `hup`.
|
||||
/// * called from outside of a task context.
|
||||
pub fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
|
||||
poll_ready!(
|
||||
self,
|
||||
mio::Ready::writable(),
|
||||
write_readiness,
|
||||
take_write_ready,
|
||||
self.inner.registration.poll_write_ready(cx)
|
||||
)
|
||||
}
|
||||
|
||||
/// 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 only clears writable readiness. HUP (on platforms that support HUP)
|
||||
/// cannot be cleared as it is a final state.
|
||||
///
|
||||
/// After calling this function, `poll_write_ready(Ready::writable())` will
|
||||
/// return `NotReady` until a new write readiness event has been received.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function will panic if called from outside of a task context.
|
||||
pub fn clear_write_ready(&self, cx: &mut Context<'_>) -> io::Result<()> {
|
||||
let ready = mio::Ready::writable();
|
||||
|
||||
self.inner
|
||||
.write_readiness
|
||||
.fetch_and(!ready.as_usize(), Relaxed);
|
||||
|
||||
if self.poll_write_ready(cx)?.is_ready() {
|
||||
// Notify the current task
|
||||
cx.waker().wake_by_ref();
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Ensure that the I/O resource is registered with the reactor.
|
||||
fn register(&self) -> io::Result<()> {
|
||||
self.inner
|
||||
.registration
|
||||
.register(self.io.as_ref().unwrap())?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
// ===== Read / Write impls =====
|
||||
|
||||
impl<E> AsyncRead for PollEvented<E>
|
||||
where
|
||||
E: Evented + Read + Unpin,
|
||||
{
|
||||
fn poll_read(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut [u8],
|
||||
) -> Poll<io::Result<usize>> {
|
||||
ready!(self.poll_read_ready(cx, mio::Ready::readable()))?;
|
||||
|
||||
let r = (*self).get_mut().read(buf);
|
||||
|
||||
if is_wouldblock(&r) {
|
||||
self.clear_read_ready(cx, mio::Ready::readable())?;
|
||||
return Poll::Pending;
|
||||
}
|
||||
|
||||
Poll::Ready(r)
|
||||
}
|
||||
}
|
||||
|
||||
impl<E> AsyncWrite for PollEvented<E>
|
||||
where
|
||||
E: Evented + Write + Unpin,
|
||||
{
|
||||
fn poll_write(
|
||||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
buf: &[u8],
|
||||
) -> Poll<io::Result<usize>> {
|
||||
ready!(self.poll_write_ready(cx))?;
|
||||
|
||||
let r = (*self).get_mut().write(buf);
|
||||
|
||||
if is_wouldblock(&r) {
|
||||
self.clear_write_ready(cx)?;
|
||||
return Poll::Pending;
|
||||
}
|
||||
|
||||
Poll::Ready(r)
|
||||
}
|
||||
|
||||
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
ready!(self.poll_write_ready(cx))?;
|
||||
|
||||
let r = (*self).get_mut().flush();
|
||||
|
||||
if is_wouldblock(&r) {
|
||||
self.clear_write_ready(cx)?;
|
||||
return Poll::Pending;
|
||||
}
|
||||
|
||||
Poll::Ready(r)
|
||||
}
|
||||
|
||||
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
}
|
||||
|
||||
fn is_wouldblock<T>(r: &io::Result<T>) -> bool {
|
||||
match *r {
|
||||
Ok(_) => false,
|
||||
Err(ref e) => e.kind() == io::ErrorKind::WouldBlock,
|
||||
}
|
||||
}
|
||||
|
||||
impl<E: Evented + 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()
|
||||
}
|
||||
}
|
||||
|
||||
impl<E: Evented> Drop for PollEvented<E> {
|
||||
fn drop(&mut self) {
|
||||
if let Some(io) = self.io.take() {
|
||||
// Ignore errors
|
||||
let _ = self.inner.registration.deregister(&io);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,559 @@
|
||||
use crate::{Direction, Handle, HandlePriv};
|
||||
use log::debug;
|
||||
use mio::{self, Evented};
|
||||
use std::cell::UnsafeCell;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::atomic::Ordering::SeqCst;
|
||||
use std::task::{Context, Poll, Waker};
|
||||
use std::{io, ptr, usize};
|
||||
|
||||
/// 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: HandlePriv,
|
||||
token: usize,
|
||||
}
|
||||
|
||||
/// Tasks waiting on readiness notifications.
|
||||
#[derive(Debug)]
|
||||
struct Node {
|
||||
direction: Direction,
|
||||
waker: Waker,
|
||||
next: *mut 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.
|
||||
///
|
||||
/// # Return
|
||||
///
|
||||
/// 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, HandlePriv::try_current)
|
||||
}
|
||||
|
||||
/// Deregister the I/O resource from the reactor it is associated with.
|
||||
///
|
||||
/// This function must be called before the I/O resource associated with the
|
||||
/// registration is dropped.
|
||||
///
|
||||
/// Note that deregistering does not guarantee that the I/O resource can be
|
||||
/// registered with a different reactor. Some I/O resource types can only be
|
||||
/// associated with a single reactor instance for their lifetime.
|
||||
///
|
||||
/// # Return
|
||||
///
|
||||
/// If the deregistration was successful, `Ok` is returned. Any calls to
|
||||
/// `Reactor::turn` that happen after a successful call to `deregister` will
|
||||
/// no longer result in notifications getting sent for this registration.
|
||||
///
|
||||
/// `Err` is returned if an error is encountered.
|
||||
pub fn deregister<T>(&mut self, io: &T) -> io::Result<()>
|
||||
where
|
||||
T: Evented,
|
||||
{
|
||||
// The state does not need to be checked and coordination is not
|
||||
// necessary as this function takes `&mut self`. This guarantees a
|
||||
// single thread is accessing the instance.
|
||||
if let Some(inner) = unsafe { (*self.inner.get()).as_ref() } {
|
||||
inner.deregister(io)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 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, || match handle.as_priv() {
|
||||
Some(handle) => Ok(handle.clone()),
|
||||
None => HandlePriv::try_current(),
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn register_with_priv<T>(&self, io: &T, handle: &HandlePriv) -> 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<HandlePriv>,
|
||||
{
|
||||
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 mut read = false;
|
||||
let mut write = false;
|
||||
let mut ptr = (actual & !LIFECYCLE_MASK) as *mut Node;
|
||||
|
||||
let inner = unsafe { (*self.inner.get()).as_ref().unwrap() };
|
||||
|
||||
while !ptr.is_null() {
|
||||
let node = unsafe { Box::from_raw(ptr) };
|
||||
let node = *node;
|
||||
let Node {
|
||||
direction,
|
||||
waker,
|
||||
next,
|
||||
} = node;
|
||||
|
||||
let flag = match direction {
|
||||
Direction::Read => &mut read,
|
||||
Direction::Write => &mut write,
|
||||
};
|
||||
|
||||
if !*flag {
|
||||
*flag = true;
|
||||
|
||||
inner.register(direction, waker);
|
||||
}
|
||||
|
||||
ptr = next;
|
||||
}
|
||||
|
||||
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.
|
||||
///
|
||||
/// All events except `HUP` are [edge-triggered]. Once `HUP` is returned,
|
||||
/// the function will always return `Ready(HUP)`. This should be treated as
|
||||
/// the end of the readiness stream.
|
||||
///
|
||||
/// Ensure that [`register`] has been called first.
|
||||
///
|
||||
/// # Return value
|
||||
///
|
||||
/// There are several possible return values:
|
||||
///
|
||||
/// * `Poll::Ready(Ok(readiness))` means that the I/O resource has received
|
||||
/// a new readiness event. The readiness value is included.
|
||||
///
|
||||
/// * `Poll::Pending` means that no new readiness events have been received
|
||||
/// since the last call to `poll_read_ready`.
|
||||
///
|
||||
/// * `Poll::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, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
|
||||
let v = self.poll_ready(Direction::Read, Some(cx))?;
|
||||
match v {
|
||||
Some(v) => Poll::Ready(Ok(v)),
|
||||
None => Poll::Pending,
|
||||
}
|
||||
}
|
||||
|
||||
/// 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, None)
|
||||
}
|
||||
|
||||
/// 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.
|
||||
///
|
||||
/// All events except `HUP` are [edge-triggered]. Once `HUP` is returned,
|
||||
/// the function will always return `Ready(HUP)`. This should be treated as
|
||||
/// the end of the readiness stream.
|
||||
///
|
||||
/// Ensure that [`register`] has been called first.
|
||||
///
|
||||
/// # Return value
|
||||
///
|
||||
/// There are several possible return values:
|
||||
///
|
||||
/// * `Poll::Ready(Ok(readiness))` means that the I/O resource has received
|
||||
/// a new readiness event. The readiness value is included.
|
||||
///
|
||||
/// * `Poll::Pending` means that no new readiness events have been received
|
||||
/// since the last call to `poll_write_ready`.
|
||||
///
|
||||
/// * `Poll::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, cx: &mut Context<'_>) -> Poll<io::Result<mio::Ready>> {
|
||||
let v = self.poll_ready(Direction::Write, Some(cx))?;
|
||||
match v {
|
||||
Some(v) => Poll::Ready(Ok(v)),
|
||||
None => Poll::Pending,
|
||||
}
|
||||
}
|
||||
|
||||
/// 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, None)
|
||||
}
|
||||
|
||||
fn poll_ready(
|
||||
&self,
|
||||
direction: Direction,
|
||||
cx: Option<&mut Context<'_>>,
|
||||
) -> 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, cx);
|
||||
}
|
||||
LOCKED => {
|
||||
let cx = if let Some(ref cx) = cx {
|
||||
cx
|
||||
} else {
|
||||
// Skip the notification tracking junk.
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let next_ptr = (state & !LIFECYCLE_MASK) as *mut Node;
|
||||
|
||||
// Get the node
|
||||
let mut n = node.take().unwrap_or_else(|| {
|
||||
Box::new(Node {
|
||||
direction,
|
||||
waker: cx.waker().clone(),
|
||||
next: ptr::null_mut(),
|
||||
})
|
||||
});
|
||||
|
||||
n.next = next_ptr;
|
||||
|
||||
let node_ptr = Box::into_raw(n);
|
||||
let next = node_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 n = unsafe { Box::from_raw(node_ptr) };
|
||||
|
||||
// Save this for next loop
|
||||
node = Some(n);
|
||||
|
||||
state = actual;
|
||||
continue;
|
||||
}
|
||||
|
||||
return Ok(None);
|
||||
}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for Registration {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
unsafe impl Send for Registration {}
|
||||
unsafe impl Sync for Registration {}
|
||||
|
||||
// ===== impl Inner =====
|
||||
|
||||
impl Inner {
|
||||
fn new<T>(io: &T, handle: HandlePriv) -> (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, waker: Waker) {
|
||||
if self.token == ERROR {
|
||||
waker.wake();
|
||||
return;
|
||||
}
|
||||
|
||||
let inner = match self.handle.inner() {
|
||||
Some(inner) => inner,
|
||||
None => {
|
||||
waker.wake();
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
inner.register(self.token, direction, waker);
|
||||
}
|
||||
|
||||
fn deregister<E: Evented>(&self, io: &E) -> io::Result<()> {
|
||||
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")),
|
||||
};
|
||||
|
||||
inner.deregister_source(io)
|
||||
}
|
||||
|
||||
fn poll_ready(
|
||||
&self,
|
||||
direction: Direction,
|
||||
cx: Option<&mut Context<'_>>,
|
||||
) -> 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 mask_no_hup = (mask - crate::platform::hup()).as_usize();
|
||||
|
||||
let io_dispatch = inner.io_dispatch.read();
|
||||
let sched = &io_dispatch[self.token];
|
||||
|
||||
// This consumes the current readiness state **except** for HUP. HUP is
|
||||
// excluded because a) it is a final state and never transitions out of
|
||||
// HUP and b) both the read AND the write directions need to be able to
|
||||
// observe this state.
|
||||
//
|
||||
// If HUP were to be cleared when `direction` is `Read`, then when
|
||||
// `poll_ready` is called again with a _`direction` of `Write`, the HUP
|
||||
// state would not be visible.
|
||||
let mut ready =
|
||||
mask & mio::Ready::from_usize(sched.readiness.fetch_and(!mask_no_hup, SeqCst));
|
||||
|
||||
if ready.is_empty() {
|
||||
if let Some(cx) = cx {
|
||||
debug!("scheduling {:?} for: {}", direction, self.token);
|
||||
// Update the task info
|
||||
match direction {
|
||||
Direction::Read => sched.reader.register_by_ref(cx.waker()),
|
||||
Direction::Write => sched.writer.register_by_ref(cx.waker()),
|
||||
}
|
||||
|
||||
// Try again
|
||||
ready =
|
||||
mask & mio::Ready::from_usize(sched.readiness.fetch_and(!mask_no_hup, SeqCst));
|
||||
}
|
||||
}
|
||||
|
||||
if ready.is_empty() {
|
||||
Ok(None)
|
||||
} else {
|
||||
Ok(Some(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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,217 @@
|
||||
//! A scalable reader-writer lock.
|
||||
//!
|
||||
//! This implementation makes read operations faster and more scalable due to less contention,
|
||||
//! while making write operations slower. It also incurs much higher memory overhead than
|
||||
//! traditional reader-writer locks.
|
||||
|
||||
use crossbeam_utils::CachePadded;
|
||||
use lazy_static::lazy_static;
|
||||
use num_cpus;
|
||||
use parking_lot;
|
||||
use std::cell::UnsafeCell;
|
||||
use std::collections::HashMap;
|
||||
use std::marker::PhantomData;
|
||||
use std::mem;
|
||||
use std::ops::{Deref, DerefMut};
|
||||
use std::sync::Mutex;
|
||||
use std::thread::{self, ThreadId};
|
||||
|
||||
/// A scalable read-writer lock.
|
||||
///
|
||||
/// This type of lock allows a number of readers or at most one writer at any point in time. The
|
||||
/// write portion of this lock typically allows modification of the underlying data (exclusive
|
||||
/// access) and the read portion of this lock typically allows for read-only access (shared
|
||||
/// access).
|
||||
///
|
||||
/// This reader-writer lock differs from typical implementations in that it internally creates a
|
||||
/// list of reader-writer locks called 'shards'. Shards are aligned and padded to the cache line
|
||||
/// size.
|
||||
///
|
||||
/// Read operations lock only one shard specific to the current thread, while write operations lock
|
||||
/// every shard in succession. This strategy makes concurrent read operations faster due to less
|
||||
/// contention, but write operations are slower due to increased amount of locking.
|
||||
pub(crate) struct RwLock<T> {
|
||||
/// A list of locks protecting the internal data.
|
||||
shards: Vec<CachePadded<parking_lot::RwLock<()>>>,
|
||||
|
||||
/// The internal data.
|
||||
value: UnsafeCell<T>,
|
||||
}
|
||||
|
||||
unsafe impl<T: Send> Send for RwLock<T> {}
|
||||
unsafe impl<T: Send + Sync> Sync for RwLock<T> {}
|
||||
|
||||
impl<T> RwLock<T> {
|
||||
/// Creates a new `RwLock` initialized with `value`.
|
||||
pub(crate) fn new(value: T) -> RwLock<T> {
|
||||
// The number of shards is a power of two so that the modulo operation in `read` becomes a
|
||||
// simple bitwise "and".
|
||||
let num_shards = num_cpus::get().next_power_of_two();
|
||||
|
||||
RwLock {
|
||||
shards: (0..num_shards)
|
||||
.map(|_| CachePadded::new(parking_lot::RwLock::new(())))
|
||||
.collect(),
|
||||
value: UnsafeCell::new(value),
|
||||
}
|
||||
}
|
||||
|
||||
/// Locks this `RwLock` with shared read access, blocking the current thread until it can be
|
||||
/// acquired.
|
||||
///
|
||||
/// The calling thread will be blocked until there are no more writers which hold the lock.
|
||||
/// There may be other readers currently inside the lock when this method returns. This method
|
||||
/// does not provide any guarantees with respect to the ordering of whether contentious readers
|
||||
/// or writers will acquire the lock first.
|
||||
///
|
||||
/// Returns an RAII guard which will release this thread's shared access once it is dropped.
|
||||
pub(crate) fn read(&self) -> RwLockReadGuard<'_, T> {
|
||||
// Take the current thread index and map it to a shard index. Thread indices will tend to
|
||||
// distribute shards among threads equally, thus reducing contention due to read-locking.
|
||||
let shard_index = thread_index() & (self.shards.len() - 1);
|
||||
|
||||
RwLockReadGuard {
|
||||
parent: self,
|
||||
_guard: self.shards[shard_index].read(),
|
||||
_marker: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
/// Locks this rwlock with exclusive write access, blocking the current thread until it can be
|
||||
/// acquired.
|
||||
///
|
||||
/// This function will not return while other writers or other readers currently have access to
|
||||
/// the lock.
|
||||
///
|
||||
/// Returns an RAII guard which will drop the write access of this rwlock when dropped.
|
||||
pub(crate) fn write(&self) -> RwLockWriteGuard<'_, T> {
|
||||
// Write-lock each shard in succession.
|
||||
for shard in &self.shards {
|
||||
// The write guard is forgotten, but the lock will be manually unlocked in `drop`.
|
||||
mem::forget(shard.write());
|
||||
}
|
||||
|
||||
RwLockWriteGuard {
|
||||
parent: self,
|
||||
_marker: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A guard used to release the shared read access of a `RwLock` when dropped.
|
||||
pub(crate) struct RwLockReadGuard<'a, T> {
|
||||
parent: &'a RwLock<T>,
|
||||
_guard: parking_lot::RwLockReadGuard<'a, ()>,
|
||||
_marker: PhantomData<parking_lot::RwLockReadGuard<'a, T>>,
|
||||
}
|
||||
|
||||
unsafe impl<'a, T: Sync> Sync for RwLockReadGuard<'a, T> {}
|
||||
|
||||
impl<'a, T> Deref for RwLockReadGuard<'a, T> {
|
||||
type Target = T;
|
||||
|
||||
fn deref(&self) -> &T {
|
||||
unsafe { &*self.parent.value.get() }
|
||||
}
|
||||
}
|
||||
|
||||
/// A guard used to release the exclusive write access of a `RwLock` when dropped.
|
||||
pub(crate) struct RwLockWriteGuard<'a, T> {
|
||||
parent: &'a RwLock<T>,
|
||||
_marker: PhantomData<parking_lot::RwLockWriteGuard<'a, T>>,
|
||||
}
|
||||
|
||||
unsafe impl<'a, T: Sync> Sync for RwLockWriteGuard<'a, T> {}
|
||||
|
||||
impl<'a, T> Drop for RwLockWriteGuard<'a, T> {
|
||||
fn drop(&mut self) {
|
||||
// Unlock the shards in reverse order of locking.
|
||||
for shard in self.parent.shards.iter().rev() {
|
||||
unsafe {
|
||||
shard.force_unlock_write();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, T> Deref for RwLockWriteGuard<'a, T> {
|
||||
type Target = T;
|
||||
|
||||
fn deref(&self) -> &T {
|
||||
unsafe { &*self.parent.value.get() }
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, T> DerefMut for RwLockWriteGuard<'a, T> {
|
||||
fn deref_mut(&mut self) -> &mut T {
|
||||
unsafe { &mut *self.parent.value.get() }
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a `usize` that identifies the current thread.
|
||||
///
|
||||
/// Each thread is associated with an 'index'. Indices usually tend to be consecutive numbers
|
||||
/// between 0 and the number of running threads, but there are no guarantees. During TLS teardown
|
||||
/// the associated index might change.
|
||||
#[inline]
|
||||
pub(crate) fn thread_index() -> usize {
|
||||
REGISTRATION.try_with(|reg| reg.index).unwrap_or(0)
|
||||
}
|
||||
|
||||
/// The global registry keeping track of registered threads and indices.
|
||||
struct ThreadIndices {
|
||||
/// Mapping from `ThreadId` to thread index.
|
||||
mapping: HashMap<ThreadId, usize>,
|
||||
|
||||
/// A list of free indices.
|
||||
free_list: Vec<usize>,
|
||||
|
||||
/// The next index to allocate if the free list is empty.
|
||||
next_index: usize,
|
||||
}
|
||||
|
||||
lazy_static! {
|
||||
static ref THREAD_INDICES: Mutex<ThreadIndices> = Mutex::new(ThreadIndices {
|
||||
mapping: HashMap::new(),
|
||||
free_list: Vec::new(),
|
||||
next_index: 0,
|
||||
});
|
||||
}
|
||||
|
||||
/// A registration of a thread with an index.
|
||||
///
|
||||
/// When dropped, unregisters the thread and frees the reserved index.
|
||||
struct Registration {
|
||||
index: usize,
|
||||
thread_id: ThreadId,
|
||||
}
|
||||
|
||||
impl Drop for Registration {
|
||||
fn drop(&mut self) {
|
||||
let mut indices = THREAD_INDICES.lock().unwrap();
|
||||
indices.mapping.remove(&self.thread_id);
|
||||
indices.free_list.push(self.index);
|
||||
}
|
||||
}
|
||||
|
||||
thread_local! {
|
||||
static REGISTRATION: Registration = {
|
||||
let thread_id = thread::current().id();
|
||||
let mut indices = THREAD_INDICES.lock().unwrap();
|
||||
|
||||
let index = match indices.free_list.pop() {
|
||||
Some(i) => i,
|
||||
None => {
|
||||
let i = indices.next_index;
|
||||
indices.next_index += 1;
|
||||
i
|
||||
}
|
||||
};
|
||||
indices.mapping.insert(thread_id, index);
|
||||
|
||||
Registration {
|
||||
index,
|
||||
thread_id,
|
||||
}
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user