|
|
|
|
@ -2,13 +2,16 @@ use std::fmt;
|
|
|
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
|
use std::time::Duration;
|
|
|
|
|
|
|
|
|
|
use mio::{self, Evented};
|
|
|
|
|
use mio::{self, event::Source, Interest};
|
|
|
|
|
use slab::Slab;
|
|
|
|
|
|
|
|
|
|
use crate::io;
|
|
|
|
|
use crate::rt::RUNTIME;
|
|
|
|
|
use crate::task::{Context, Poll, Waker};
|
|
|
|
|
|
|
|
|
|
// TODO: ADD AIO and LIO?
|
|
|
|
|
const INTEREST_ALL: Interest = Interest::READABLE.add(Interest::WRITABLE);
|
|
|
|
|
|
|
|
|
|
/// Data associated with a registered I/O handle.
|
|
|
|
|
#[derive(Debug)]
|
|
|
|
|
struct Entry {
|
|
|
|
|
@ -25,7 +28,7 @@ struct Entry {
|
|
|
|
|
/// The state of a networking driver.
|
|
|
|
|
pub struct Reactor {
|
|
|
|
|
/// A mio instance that polls for new events.
|
|
|
|
|
poller: mio::Poll,
|
|
|
|
|
poller: Mutex<mio::Poll>,
|
|
|
|
|
|
|
|
|
|
/// A list into which mio stores events.
|
|
|
|
|
events: Mutex<mio::Events>,
|
|
|
|
|
@ -33,8 +36,8 @@ pub struct Reactor {
|
|
|
|
|
/// A collection of registered I/O handles.
|
|
|
|
|
entries: Mutex<Slab<Arc<Entry>>>,
|
|
|
|
|
|
|
|
|
|
/// Dummy I/O handle that is only used to wake up the polling thread.
|
|
|
|
|
notify_reg: (mio::Registration, mio::SetReadiness),
|
|
|
|
|
/// Mio waker that is only used to wake up the polling thread.
|
|
|
|
|
notify_waker: mio::Waker,
|
|
|
|
|
|
|
|
|
|
/// An identifier for the notification handle.
|
|
|
|
|
notify_token: mio::Token,
|
|
|
|
|
@ -64,25 +67,38 @@ impl Reactor {
|
|
|
|
|
/// Creates a new reactor for polling I/O events.
|
|
|
|
|
pub fn new() -> io::Result<Reactor> {
|
|
|
|
|
let poller = mio::Poll::new()?;
|
|
|
|
|
let notify_reg = mio::Registration::new2();
|
|
|
|
|
let mut entries = Slab::new();
|
|
|
|
|
|
|
|
|
|
let mut reactor = Reactor {
|
|
|
|
|
poller,
|
|
|
|
|
// Register a waker for waking up the polling thread.
|
|
|
|
|
let vacant = entries.vacant_entry();
|
|
|
|
|
let notify_token = mio::Token(vacant.key());
|
|
|
|
|
let notify_waker = mio::Waker::new(poller.registry(), notify_token)?;
|
|
|
|
|
// dumy entry to avoid reusing the same token
|
|
|
|
|
vacant.insert(Arc::new(Entry {
|
|
|
|
|
token: notify_token.clone(),
|
|
|
|
|
readers: Mutex::new(Readers {
|
|
|
|
|
ready: false,
|
|
|
|
|
wakers: Vec::new(),
|
|
|
|
|
}),
|
|
|
|
|
writers: Mutex::new(Writers {
|
|
|
|
|
ready: false,
|
|
|
|
|
wakers: Vec::new(),
|
|
|
|
|
}),
|
|
|
|
|
}));
|
|
|
|
|
|
|
|
|
|
let reactor = Reactor {
|
|
|
|
|
poller: Mutex::new(poller),
|
|
|
|
|
events: Mutex::new(mio::Events::with_capacity(1000)),
|
|
|
|
|
entries: Mutex::new(Slab::new()),
|
|
|
|
|
notify_reg,
|
|
|
|
|
notify_token: mio::Token(0),
|
|
|
|
|
entries: Mutex::new(entries),
|
|
|
|
|
notify_waker,
|
|
|
|
|
notify_token,
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
// Register a dummy I/O handle for waking up the polling thread.
|
|
|
|
|
let entry = reactor.register(&reactor.notify_reg.0)?;
|
|
|
|
|
reactor.notify_token = entry.token;
|
|
|
|
|
|
|
|
|
|
Ok(reactor)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Registers an I/O event source and returns its associated entry.
|
|
|
|
|
fn register(&self, source: &dyn Evented) -> io::Result<Arc<Entry>> {
|
|
|
|
|
fn register(&self, source: &mut dyn Source) -> io::Result<Arc<Entry>> {
|
|
|
|
|
let mut entries = self.entries.lock().unwrap();
|
|
|
|
|
|
|
|
|
|
// Reserve a vacant spot in the slab and use its key as the token value.
|
|
|
|
|
@ -104,17 +120,19 @@ impl Reactor {
|
|
|
|
|
vacant.insert(entry.clone());
|
|
|
|
|
|
|
|
|
|
// Register the I/O event source in the poller.
|
|
|
|
|
let interest = mio::Ready::all();
|
|
|
|
|
let opts = mio::PollOpt::edge();
|
|
|
|
|
self.poller.register(source, token, interest, opts)?;
|
|
|
|
|
self.poller
|
|
|
|
|
.lock()
|
|
|
|
|
.unwrap()
|
|
|
|
|
.registry()
|
|
|
|
|
.register(source, token, INTEREST_ALL)?;
|
|
|
|
|
|
|
|
|
|
Ok(entry)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Deregisters an I/O event source associated with an entry.
|
|
|
|
|
fn deregister(&self, source: &dyn Evented, entry: &Entry) -> io::Result<()> {
|
|
|
|
|
fn deregister(&self, source: &mut dyn Source, entry: &Entry) -> io::Result<()> {
|
|
|
|
|
// Deregister the I/O object from the mio instance.
|
|
|
|
|
self.poller.deregister(source)?;
|
|
|
|
|
self.poller.lock().unwrap().registry().deregister(source)?;
|
|
|
|
|
|
|
|
|
|
// Remove the entry associated with the I/O object.
|
|
|
|
|
self.entries.lock().unwrap().remove(entry.token.0);
|
|
|
|
|
@ -124,7 +142,7 @@ impl Reactor {
|
|
|
|
|
|
|
|
|
|
/// Notifies the reactor so that polling stops blocking.
|
|
|
|
|
pub fn notify(&self) -> io::Result<()> {
|
|
|
|
|
self.notify_reg.1.set_readiness(mio::Ready::readable())
|
|
|
|
|
self.notify_waker.wake()
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Waits on the poller for new events and wakes up tasks blocked on I/O handles.
|
|
|
|
|
@ -134,7 +152,7 @@ impl Reactor {
|
|
|
|
|
let mut events = self.events.lock().unwrap();
|
|
|
|
|
|
|
|
|
|
// Block on the poller until at least one new event comes in.
|
|
|
|
|
self.poller.poll(&mut events, timeout)?;
|
|
|
|
|
self.poller.lock().unwrap().poll(&mut events, timeout)?;
|
|
|
|
|
|
|
|
|
|
// Lock the entire entry table while we're processing new events.
|
|
|
|
|
let entries = self.entries.lock().unwrap();
|
|
|
|
|
@ -147,16 +165,12 @@ impl Reactor {
|
|
|
|
|
|
|
|
|
|
if token == self.notify_token {
|
|
|
|
|
// If this is the notification token, we just need the notification state.
|
|
|
|
|
self.notify_reg.1.set_readiness(mio::Ready::empty())?;
|
|
|
|
|
self.notify_waker.wake()?;
|
|
|
|
|
} else {
|
|
|
|
|
// Otherwise, look for the entry associated with this token.
|
|
|
|
|
if let Some(entry) = entries.get(token.0) {
|
|
|
|
|
// Set the readiness flags from this I/O event.
|
|
|
|
|
let readiness = event.readiness();
|
|
|
|
|
|
|
|
|
|
// Wake up reader tasks blocked on this I/O handle.
|
|
|
|
|
let reader_interests = mio::Ready::all() - mio::Ready::writable();
|
|
|
|
|
if !(readiness & reader_interests).is_empty() {
|
|
|
|
|
if event.is_readable() {
|
|
|
|
|
let mut readers = entry.readers.lock().unwrap();
|
|
|
|
|
readers.ready = true;
|
|
|
|
|
for w in readers.wakers.drain(..) {
|
|
|
|
|
@ -166,8 +180,7 @@ impl Reactor {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// Wake up writer tasks blocked on this I/O handle.
|
|
|
|
|
let writer_interests = mio::Ready::all() - mio::Ready::readable();
|
|
|
|
|
if !(readiness & writer_interests).is_empty() {
|
|
|
|
|
if event.is_writable() {
|
|
|
|
|
let mut writers = entry.writers.lock().unwrap();
|
|
|
|
|
writers.ready = true;
|
|
|
|
|
for w in writers.wakers.drain(..) {
|
|
|
|
|
@ -187,7 +200,7 @@ impl Reactor {
|
|
|
|
|
///
|
|
|
|
|
/// This handle wraps an I/O event source and exposes a "futurized" interface on top of it,
|
|
|
|
|
/// implementing traits `AsyncRead` and `AsyncWrite`.
|
|
|
|
|
pub struct Watcher<T: Evented> {
|
|
|
|
|
pub struct Watcher<T: Source> {
|
|
|
|
|
/// Data associated with the I/O handle.
|
|
|
|
|
entry: Arc<Entry>,
|
|
|
|
|
|
|
|
|
|
@ -195,16 +208,16 @@ pub struct Watcher<T: Evented> {
|
|
|
|
|
source: Option<T>,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl<T: Evented> Watcher<T> {
|
|
|
|
|
impl<T: Source> Watcher<T> {
|
|
|
|
|
/// Creates a new I/O handle.
|
|
|
|
|
///
|
|
|
|
|
/// The provided I/O event source will be kept registered inside the reactor's poller for the
|
|
|
|
|
/// lifetime of the returned I/O handle.
|
|
|
|
|
pub fn new(source: T) -> Watcher<T> {
|
|
|
|
|
pub fn new(mut source: T) -> Watcher<T> {
|
|
|
|
|
Watcher {
|
|
|
|
|
entry: RUNTIME
|
|
|
|
|
.reactor()
|
|
|
|
|
.register(&source)
|
|
|
|
|
.register(&mut source)
|
|
|
|
|
.expect("cannot register an I/O event source"),
|
|
|
|
|
source: Some(source),
|
|
|
|
|
}
|
|
|
|
|
@ -324,18 +337,18 @@ impl<T: Evented> Watcher<T> {
|
|
|
|
|
/// This method is typically used to convert `Watcher`s to raw file descriptors/handles.
|
|
|
|
|
#[allow(dead_code)]
|
|
|
|
|
pub fn into_inner(mut self) -> T {
|
|
|
|
|
let source = self.source.take().unwrap();
|
|
|
|
|
let mut source = self.source.take().unwrap();
|
|
|
|
|
RUNTIME
|
|
|
|
|
.reactor()
|
|
|
|
|
.deregister(&source, &self.entry)
|
|
|
|
|
.deregister(&mut source, &self.entry)
|
|
|
|
|
.expect("cannot deregister I/O event source");
|
|
|
|
|
source
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl<T: Evented> Drop for Watcher<T> {
|
|
|
|
|
impl<T: Source> Drop for Watcher<T> {
|
|
|
|
|
fn drop(&mut self) {
|
|
|
|
|
if let Some(ref source) = self.source {
|
|
|
|
|
if let Some(ref mut source) = self.source {
|
|
|
|
|
RUNTIME
|
|
|
|
|
.reactor()
|
|
|
|
|
.deregister(source, &self.entry)
|
|
|
|
|
@ -344,7 +357,7 @@ impl<T: Evented> Drop for Watcher<T> {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
impl<T: Evented + fmt::Debug> fmt::Debug for Watcher<T> {
|
|
|
|
|
impl<T: Source + fmt::Debug> fmt::Debug for Watcher<T> {
|
|
|
|
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
|
|
|
|
f.debug_struct("Watcher")
|
|
|
|
|
.field("entry", &self.entry)
|
|
|
|
|
|
dignifiedquire