mirror of
https://github.com/tokio-rs/tokio.git
synced 2026-08-28 00:00:11 +02:00
Reorganize the entire crate:
Renamed APIs * Loop => reactor::Core * LoopHandle => reactor::Handle * LoopPin => reactor::Pinned * TcpStream => net::TcpStream * TcpListener => net::TcpListener * UdpSocket => net::UdpSocket * Sender => channel::Sender * Receiver => channel::Receiver * Timeout => reactor::Timeout * ReadinessStream => reactor::PollEvented * All `LoopHandle` methods to construct objects are now free functions on the associated types, e.g. `LoopHandle::tcp_listen` is now `TcpListener::bind` * All APIs taking a `Handle` now take a `Handle` as the last argument * All future-returning APIs now return concrete types instead of trait objects Added APIs * io::Io trait -- Read + Write + ability to poll Removed without replacement: * AddSource * AddTimeout * IoToken * TimeoutToken Closes #3 Closes #6
This commit is contained in:
Alex Crichton
@@ -0,0 +1,118 @@
|
||||
//! A thin wrapper around a mpsc queue and mio-based channel information
|
||||
//!
|
||||
//! Normally the standard library's channels would suffice but we unfortunately
|
||||
//! need the `Sender<T>` half to be `Sync`, so to accomplish this for now we
|
||||
//! just vendor the same mpsc queue as the one in the standard library and then
|
||||
//! we pair that with the `mio::channel` module's Ctl pairs to control the
|
||||
//! readiness notifications on the channel.
|
||||
|
||||
use std::cell::Cell;
|
||||
use std::io;
|
||||
use std::marker;
|
||||
use std::sync::Arc;
|
||||
|
||||
use mio;
|
||||
use mio::channel::{ctl_pair, SenderCtl, ReceiverCtl};
|
||||
|
||||
use mpsc_queue::{Queue, PopResult};
|
||||
|
||||
pub struct Sender<T> {
|
||||
ctl: SenderCtl,
|
||||
inner: Arc<Queue<T>>,
|
||||
}
|
||||
|
||||
pub struct Receiver<T> {
|
||||
ctl: ReceiverCtl,
|
||||
inner: Arc<Queue<T>>,
|
||||
_marker: marker::PhantomData<Cell<()>>, // this type is not Sync
|
||||
}
|
||||
|
||||
pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
|
||||
let inner = Arc::new(Queue::new());
|
||||
let (tx, rx) = ctl_pair();
|
||||
|
||||
let tx = Sender {
|
||||
ctl: tx,
|
||||
inner: inner.clone(),
|
||||
};
|
||||
let rx = Receiver {
|
||||
ctl: rx,
|
||||
inner: inner.clone(),
|
||||
_marker: marker::PhantomData,
|
||||
};
|
||||
(tx, rx)
|
||||
}
|
||||
|
||||
impl<T> Sender<T> {
|
||||
pub fn send(&self, data: T) -> io::Result<()> {
|
||||
self.inner.push(data);
|
||||
self.ctl.inc()
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Receiver<T> {
|
||||
pub fn recv(&self) -> io::Result<Option<T>> {
|
||||
// Note that the underlying method is `unsafe` because it's only safe
|
||||
// if one thread accesses it at a time.
|
||||
//
|
||||
// We, however, are the only thread with a `Receiver<T>` because this
|
||||
// type is not `Sync`. and we never handed out another instance.
|
||||
match unsafe { self.inner.pop() } {
|
||||
PopResult::Data(t) => {
|
||||
try!(self.ctl.dec());
|
||||
Ok(Some(t))
|
||||
}
|
||||
|
||||
// If the queue is either in an inconsistent or empty state, then
|
||||
// we return `None` for both instances. Note that the standard
|
||||
// library performs a yield loop in the event of `Inconsistent`,
|
||||
// which means that there's data in the queue but a sender hasn't
|
||||
// finished their operation yet.
|
||||
//
|
||||
// We do this because the queue will continue to be readable as
|
||||
// the thread performing the push will eventually call `inc`, so
|
||||
// if we return `None` and the event loop just loops aruond calling
|
||||
// this method then we'll eventually get back to the same spot
|
||||
// and due the retry.
|
||||
//
|
||||
// Basically, the inconsistent state doesn't mean we need to busy
|
||||
// wait, but instead we can forge ahead and assume by the time we
|
||||
// go to the kernel and come back we'll no longer be in an
|
||||
// inconsistent state.
|
||||
PopResult::Empty |
|
||||
PopResult::Inconsistent => Ok(None),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Just delegate everything to `self.ctl`
|
||||
impl<T> mio::Evented for Receiver<T> {
|
||||
fn register(&self,
|
||||
poll: &mio::Poll,
|
||||
token: mio::Token,
|
||||
interest: mio::Ready,
|
||||
opts: mio::PollOpt) -> io::Result<()> {
|
||||
self.ctl.register(poll, token, interest, opts)
|
||||
}
|
||||
|
||||
fn reregister(&self,
|
||||
poll: &mio::Poll,
|
||||
token: mio::Token,
|
||||
interest: mio::Ready,
|
||||
opts: mio::PollOpt) -> io::Result<()> {
|
||||
self.ctl.reregister(poll, token, interest, opts)
|
||||
}
|
||||
|
||||
fn deregister(&self, poll: &mio::Poll) -> io::Result<()> {
|
||||
self.ctl.deregister(poll)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Clone for Sender<T> {
|
||||
fn clone(&self) -> Sender<T> {
|
||||
Sender {
|
||||
ctl: self.ctl.clone(),
|
||||
inner: self.inner.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,175 @@
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::io;
|
||||
|
||||
use futures::{Future, Poll};
|
||||
use futures::task;
|
||||
use mio;
|
||||
|
||||
use reactor::{Message, Handle, CoreFuture, Direction, Core};
|
||||
|
||||
/// A future which will resolve a unique `tok` token for an I/O object.
|
||||
///
|
||||
/// Created through the `Handle::add_source` method, this future can also
|
||||
/// resolve to an error if there's an issue communicating with the event loop.
|
||||
pub struct IoTokenNew<E> {
|
||||
inner: CoreFuture<(E, (Arc<AtomicUsize>, usize)), E>,
|
||||
}
|
||||
|
||||
/// A token that identifies an active timeout.
|
||||
pub struct IoToken {
|
||||
token: usize,
|
||||
// TODO: can we avoid this allocation? It's kind of a bummer...
|
||||
readiness: Arc<AtomicUsize>,
|
||||
}
|
||||
|
||||
impl IoToken {
|
||||
/// Add a new source to an event loop, returning a future which will resolve
|
||||
/// to the token that can be used to identify this source.
|
||||
///
|
||||
/// When a new I/O object is created it needs to be communicated to the
|
||||
/// event loop to ensure that it's registered and ready to receive
|
||||
/// notifications. The event loop with then respond back with the I/O object
|
||||
/// and a token which can be used to send more messages to the event loop.
|
||||
///
|
||||
/// The token returned is then passed in turn to each of the methods below
|
||||
/// to interact with notifications on the I/O object itself.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// The returned future will panic if the event loop this handle is
|
||||
/// associated with has gone away, or if there is an error communicating
|
||||
/// with the event loop.
|
||||
pub fn new<E>(source: E, handle: &Handle) -> IoTokenNew<E>
|
||||
where E: mio::Evented + Send + 'static,
|
||||
{
|
||||
IoTokenNew {
|
||||
inner: CoreFuture {
|
||||
handle: handle.clone(),
|
||||
data: Some(source),
|
||||
result: None,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Consumes the last readiness notification the token this source is for
|
||||
/// registered.
|
||||
///
|
||||
/// Currently sources receive readiness notifications on an edge-basis. That
|
||||
/// is, once you receive a notification that an object can be read, you
|
||||
/// won't receive any more notifications until all of that data has been
|
||||
/// read.
|
||||
///
|
||||
/// The event loop will fill in this information and then inform futures
|
||||
/// that they're ready to go with the `schedule` method, and then the `poll`
|
||||
/// method can use this to figure out what happened.
|
||||
///
|
||||
/// > **Note**: This method should generally not be used directly, but
|
||||
/// > rather the `ReadinessStream` type should be used instead.
|
||||
// TODO: this should really return a proper newtype/enum, not a usize
|
||||
pub fn take_readiness(&self) -> usize {
|
||||
self.readiness.swap(0, Ordering::SeqCst)
|
||||
}
|
||||
|
||||
/// Schedule the current future task to receive a notification when the
|
||||
/// corresponding I/O object is readable.
|
||||
///
|
||||
/// Once an I/O object has been registered with the event loop through the
|
||||
/// `add_source` method, this method can be used with the assigned token to
|
||||
/// notify the current future task when the next read notification comes in.
|
||||
///
|
||||
/// The current task will only receive a notification **once** and to
|
||||
/// receive further notifications it will need to call `schedule_read`
|
||||
/// again.
|
||||
///
|
||||
/// > **Note**: This method should generally not be used directly, but
|
||||
/// > rather the `ReadinessStream` type should be used instead.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function will panic if the event loop this handle is associated
|
||||
/// with has gone away, or if there is an error communicating with the event
|
||||
/// loop.
|
||||
///
|
||||
/// This function will also panic if there is not a currently running future
|
||||
/// task.
|
||||
pub fn schedule_read(&self, handle: &Handle) {
|
||||
handle.send(Message::Schedule(self.token, task::park(), Direction::Read));
|
||||
}
|
||||
|
||||
/// Schedule the current future task to receive a notification when the
|
||||
/// corresponding I/O object is writable.
|
||||
///
|
||||
/// Once an I/O object has been registered with the event loop through the
|
||||
/// `add_source` method, this method can be used with the assigned token to
|
||||
/// notify the current future task when the next write notification comes
|
||||
/// in.
|
||||
///
|
||||
/// The current task will only receive a notification **once** and to
|
||||
/// receive further notifications it will need to call `schedule_write`
|
||||
/// again.
|
||||
///
|
||||
/// > **Note**: This method should generally not be used directly, but
|
||||
/// > rather the `ReadinessStream` type should be used instead.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function will panic if the event loop this handle is associated
|
||||
/// with has gone away, or if there is an error communicating with the event
|
||||
/// loop.
|
||||
///
|
||||
/// This function will also panic if there is not a currently running future
|
||||
/// task.
|
||||
pub fn schedule_write(&self, handle: &Handle) {
|
||||
handle.send(Message::Schedule(self.token, task::park(), Direction::Write));
|
||||
}
|
||||
|
||||
/// Unregister all information associated with a token on an event loop,
|
||||
/// deallocating all internal resources assigned to the given token.
|
||||
///
|
||||
/// This method should be called whenever a source of events is being
|
||||
/// destroyed. This will ensure that the event loop can reuse `tok` for
|
||||
/// another I/O object if necessary and also remove it from any poll
|
||||
/// notifications and callbacks.
|
||||
///
|
||||
/// Note that wake callbacks may still be invoked after this method is
|
||||
/// called as it may take some time for the message to drop a source to
|
||||
/// reach the event loop. Despite this fact, this method will attempt to
|
||||
/// ensure that the callbacks are **not** invoked, so pending scheduled
|
||||
/// callbacks cannot be relied upon to get called.
|
||||
///
|
||||
/// > **Note**: This method should generally not be used directly, but
|
||||
/// > rather the `ReadinessStream` type should be used instead.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This function will panic if the event loop this handle is associated
|
||||
/// with has gone away, or if there is an error communicating with the event
|
||||
/// loop.
|
||||
pub fn drop_source(&self, handle: &Handle) {
|
||||
handle.send(Message::DropSource(self.token));
|
||||
}
|
||||
}
|
||||
|
||||
impl<E> Future for IoTokenNew<E>
|
||||
where E: mio::Evented + Send + 'static,
|
||||
{
|
||||
type Item = (E, IoToken);
|
||||
type Error = io::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<(E, IoToken), io::Error> {
|
||||
let res = try_ready!(self.inner.poll(|lp, io| {
|
||||
let pair = try!(lp.add_source(&io));
|
||||
Ok((io, pair))
|
||||
}, |io, slot| {
|
||||
Message::Run(Box::new(move |lp: &Core| {
|
||||
let res = lp.add_source(&io).map(|p| (io, p));
|
||||
slot.try_produce(res).ok()
|
||||
.expect("add source try_produce intereference");
|
||||
}))
|
||||
}));
|
||||
|
||||
let (io, (ready, token)) = res;
|
||||
Ok((io, IoToken { token: token, readiness: ready }).into())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,681 @@
|
||||
//! The core reactor driving all I/O
|
||||
//!
|
||||
//! This module contains the `Core` type which is the reactor for all I/O
|
||||
//! happening in `tokio-core`. This reactor (or event loop) is used to run
|
||||
//! futures, schedule tasks, issue I/O requests, etc.
|
||||
|
||||
use std::cell::RefCell;
|
||||
use std::io::{self, ErrorKind};
|
||||
use std::mem;
|
||||
use std::rc::{Rc, Weak};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering};
|
||||
use std::time::{Instant, Duration};
|
||||
|
||||
use futures::{Future, Poll, IntoFuture, Async};
|
||||
use futures::task::{self, Unpark, Task, Spawn};
|
||||
use mio;
|
||||
use slab::Slab;
|
||||
|
||||
use slot::{self, Slot};
|
||||
use timer_wheel::{TimerWheel, Timeout as WheelTimeout};
|
||||
|
||||
mod channel;
|
||||
mod io_token;
|
||||
mod timeout_token;
|
||||
use self::channel::{Sender, Receiver, channel};
|
||||
|
||||
mod poll_evented;
|
||||
mod timeout;
|
||||
pub use self::poll_evented::{PollEvented, PollEventedNew};
|
||||
pub use self::timeout::{Timeout, TimeoutNew};
|
||||
|
||||
static NEXT_LOOP_ID: AtomicUsize = ATOMIC_USIZE_INIT;
|
||||
scoped_thread_local!(static CURRENT_LOOP: Core);
|
||||
|
||||
const SLAB_CAPACITY: usize = 1024 * 64;
|
||||
|
||||
/// An 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.
|
||||
// TODO: expand this
|
||||
pub struct Core {
|
||||
id: usize,
|
||||
io: mio::Poll,
|
||||
events: mio::Events,
|
||||
tx: Sender<Message>,
|
||||
rx: Receiver<Message>,
|
||||
io_dispatch: RefCell<Slab<ScheduledIo, usize>>,
|
||||
task_dispatch: RefCell<Slab<ScheduledTask, usize>>,
|
||||
|
||||
// Incoming queue of newly spawned futures
|
||||
new_futures: Rc<NewFutures>,
|
||||
_new_futures_registration: mio::Registration,
|
||||
|
||||
// Used for determining when the future passed to `run` is ready. Once the
|
||||
// registration is passed to `io` above we never touch it again, just keep
|
||||
// it alive.
|
||||
_future_registration: mio::Registration,
|
||||
future_readiness: Arc<MySetReadiness>,
|
||||
|
||||
// Timer wheel keeping track of all timeouts. The `usize` stored in the
|
||||
// timer wheel is an index into the slab below.
|
||||
//
|
||||
// The slab below keeps track of the timeouts themselves as well as the
|
||||
// state of the timeout itself. The `TimeoutToken` type is an index into the
|
||||
// `timeouts` slab.
|
||||
timer_wheel: RefCell<TimerWheel<usize>>,
|
||||
timeouts: RefCell<Slab<(WheelTimeout, TimeoutState), usize>>,
|
||||
}
|
||||
|
||||
/// Handle to an event loop, used to construct I/O objects, send messages, and
|
||||
/// otherwise interact indirectly with the event loop itself.
|
||||
///
|
||||
/// Handles can be cloned, and when cloned they will still refer to the
|
||||
/// same underlying event loop.
|
||||
#[derive(Clone)]
|
||||
pub struct Handle {
|
||||
id: usize,
|
||||
tx: Sender<Message>,
|
||||
}
|
||||
|
||||
/// A non-sendable handle to an event loop, useful for manufacturing instances
|
||||
/// of `LoopData`.
|
||||
#[derive(Clone)]
|
||||
pub struct Pinned {
|
||||
handle: Handle,
|
||||
futures: Weak<NewFutures>,
|
||||
}
|
||||
|
||||
struct ScheduledIo {
|
||||
readiness: Arc<AtomicUsize>,
|
||||
reader: Option<Task>,
|
||||
writer: Option<Task>,
|
||||
}
|
||||
|
||||
struct ScheduledTask {
|
||||
_registration: mio::Registration,
|
||||
spawn: Option<Spawn<Box<Future<Item=(), Error=()>>>>,
|
||||
wake: Arc<MySetReadiness>,
|
||||
}
|
||||
|
||||
struct NewFutures {
|
||||
queue: RefCell<Vec<Box<Future<Item=(), Error=()>>>>,
|
||||
ready: mio::SetReadiness,
|
||||
}
|
||||
|
||||
enum TimeoutState {
|
||||
NotFired,
|
||||
Fired,
|
||||
Waiting(Task),
|
||||
}
|
||||
|
||||
enum Direction {
|
||||
Read,
|
||||
Write,
|
||||
}
|
||||
|
||||
enum Message {
|
||||
DropSource(usize),
|
||||
Schedule(usize, Task, Direction),
|
||||
AddTimeout(Instant, Arc<Slot<io::Result<(usize, Instant)>>>),
|
||||
UpdateTimeout(usize, Task),
|
||||
CancelTimeout(usize),
|
||||
Run(Box<FnBox>),
|
||||
}
|
||||
|
||||
const TOKEN_MESSAGES: mio::Token = mio::Token(0);
|
||||
const TOKEN_FUTURE: mio::Token = mio::Token(1);
|
||||
const TOKEN_NEW_FUTURES: mio::Token = mio::Token(2);
|
||||
const TOKEN_START: usize = 3;
|
||||
|
||||
impl Core {
|
||||
/// Creates a new event loop, returning any error that happened during the
|
||||
/// creation.
|
||||
pub fn new() -> io::Result<Core> {
|
||||
let (tx, rx) = channel();
|
||||
let io = try!(mio::Poll::new());
|
||||
try!(io.register(&rx,
|
||||
TOKEN_MESSAGES,
|
||||
mio::Ready::readable(),
|
||||
mio::PollOpt::edge()));
|
||||
let future_pair = mio::Registration::new(&io,
|
||||
TOKEN_FUTURE,
|
||||
mio::Ready::readable(),
|
||||
mio::PollOpt::level());
|
||||
let new_future_pair = mio::Registration::new(&io,
|
||||
TOKEN_NEW_FUTURES,
|
||||
mio::Ready::readable(),
|
||||
mio::PollOpt::level());
|
||||
Ok(Core {
|
||||
id: NEXT_LOOP_ID.fetch_add(1, Ordering::Relaxed),
|
||||
io: io,
|
||||
events: mio::Events::with_capacity(1024),
|
||||
tx: tx,
|
||||
rx: rx,
|
||||
io_dispatch: RefCell::new(Slab::with_capacity(SLAB_CAPACITY)),
|
||||
task_dispatch: RefCell::new(Slab::with_capacity(SLAB_CAPACITY)),
|
||||
timeouts: RefCell::new(Slab::with_capacity(SLAB_CAPACITY)),
|
||||
timer_wheel: RefCell::new(TimerWheel::new()),
|
||||
_future_registration: future_pair.0,
|
||||
future_readiness: Arc::new(MySetReadiness(future_pair.1)),
|
||||
_new_futures_registration: new_future_pair.0,
|
||||
new_futures: Rc::new(NewFutures {
|
||||
queue: RefCell::new(Vec::new()),
|
||||
ready: new_future_pair.1,
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
/// Generates a handle to this event loop used to construct I/O objects and
|
||||
/// send messages.
|
||||
///
|
||||
/// Handles to an event loop are cloneable as well and clones will always
|
||||
/// refer to the same event loop.
|
||||
pub fn handle(&self) -> Handle {
|
||||
Handle {
|
||||
id: self.id,
|
||||
tx: self.tx.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a "pin" of this event loop which cannot be sent across threads
|
||||
/// but can be used as a proxy to the event loop itself.
|
||||
///
|
||||
/// Currently the primary use for this is to use as a handle to add data
|
||||
/// to the event loop directly. The `Pinned::add_loop_data` method can
|
||||
/// be used to immediately create instances of `LoopData` structures.
|
||||
pub fn pin(&self) -> Pinned {
|
||||
Pinned {
|
||||
handle: self.handle(),
|
||||
futures: Rc::downgrade(&self.new_futures),
|
||||
}
|
||||
}
|
||||
|
||||
/// Runs a future until completion, driving the event loop while we're
|
||||
/// otherwise waiting for the future to complete.
|
||||
///
|
||||
/// This function will begin executing the event loop and will finish once
|
||||
/// the provided future is resolve. Note that the future argument here
|
||||
/// crucially does not require the `'static` nor `Send` bounds. As a result
|
||||
/// the future will be "pinned" to not only this thread but also this stack
|
||||
/// frame.
|
||||
///
|
||||
/// This function will returns the value that the future resolves to once
|
||||
/// the future has finished. If the future never resolves then this function
|
||||
/// will never return.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This method will **not** catch panics from polling the future `f`. If
|
||||
/// the future panics then it's the responsibility of the caller to catch
|
||||
/// that panic and handle it as appropriate.
|
||||
///
|
||||
/// Similarly, because the provided future will be pinned not only to this
|
||||
/// thread but also to this task, any attempt to poll the future on a
|
||||
/// separate thread will result in a panic. That is, calls to
|
||||
/// `task::poll_on` must be avoided.
|
||||
pub fn run<F>(&mut self, f: F) -> Result<F::Item, F::Error>
|
||||
where F: Future,
|
||||
{
|
||||
let mut task = task::spawn(f);
|
||||
let ready = self.future_readiness.clone();
|
||||
|
||||
// Next, move all that data into a dynamically dispatched closure to cut
|
||||
// down on monomorphization costs. Inside this closure we unset the
|
||||
// readiness of the future (as we're about to poll it) and then we check
|
||||
// to see if it's done. If it's not then the event loop will turn again.
|
||||
let mut res = None;
|
||||
self._run(&mut || {
|
||||
assert!(res.is_none());
|
||||
match task.poll_future(ready.clone()) {
|
||||
Ok(Async::NotReady) => {}
|
||||
Ok(Async::Ready(e)) => res = Some(Ok(e)),
|
||||
Err(e) => res = Some(Err(e)),
|
||||
}
|
||||
res.is_some()
|
||||
});
|
||||
res.expect("run should not return until future is done")
|
||||
}
|
||||
|
||||
fn _run(&mut self, done: &mut FnMut() -> bool) {
|
||||
// Check to see if we're done immediately, if so we shouldn't do any
|
||||
// work.
|
||||
if CURRENT_LOOP.set(self, || done()) {
|
||||
return
|
||||
}
|
||||
|
||||
let mut finished = false;
|
||||
while !finished {
|
||||
let amt;
|
||||
// On Linux, Poll::poll is epoll_wait, which may return EINTR if a
|
||||
// ptracer attaches. This retry loop prevents crashing when
|
||||
// attaching strace, or similar.
|
||||
let start = Instant::now();
|
||||
loop {
|
||||
let timeout = self.timer_wheel.borrow().next_timeout().map(|t| {
|
||||
if t < start {
|
||||
Duration::new(0, 0)
|
||||
} else {
|
||||
t - start
|
||||
}
|
||||
});
|
||||
match self.io.poll(&mut self.events, timeout) {
|
||||
Ok(a) => {
|
||||
amt = a;
|
||||
break;
|
||||
}
|
||||
Err(ref e) if e.kind() == ErrorKind::Interrupted => {}
|
||||
err @ Err(_) => {
|
||||
err.unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
debug!("loop poll - {:?}", start.elapsed());
|
||||
debug!("loop time - {:?}", Instant::now());
|
||||
|
||||
// First up, process all timeouts that may have just occurred.
|
||||
let start = Instant::now();
|
||||
self.consume_timeouts(start);
|
||||
|
||||
// Next, process all the events that came in.
|
||||
for i in 0..self.events.len() {
|
||||
let event = self.events.get(i).unwrap();
|
||||
let token = event.token();
|
||||
trace!("event {:?} {:?}", event.kind(), event.token());
|
||||
|
||||
if token == TOKEN_MESSAGES {
|
||||
CURRENT_LOOP.set(&self, || self.consume_queue());
|
||||
} else if token == TOKEN_FUTURE {
|
||||
self.future_readiness.0.set_readiness(mio::Ready::none()).unwrap();
|
||||
if !finished && CURRENT_LOOP.set(self, || done()) {
|
||||
finished = true;
|
||||
}
|
||||
} else if token == TOKEN_NEW_FUTURES {
|
||||
self.new_futures.ready.set_readiness(mio::Ready::none()).unwrap();
|
||||
let mut new_futures = self.new_futures.queue.borrow_mut();
|
||||
for future in new_futures.drain(..) {
|
||||
self.spawn(future);
|
||||
}
|
||||
} else {
|
||||
self.dispatch(token, event.kind());
|
||||
}
|
||||
}
|
||||
|
||||
debug!("loop process - {} events, {:?}", amt, start.elapsed());
|
||||
}
|
||||
}
|
||||
|
||||
fn dispatch(&self, token: mio::Token, ready: mio::Ready) {
|
||||
let token = usize::from(token) - TOKEN_START;
|
||||
if token % 2 == 0 {
|
||||
self.dispatch_io(token / 2, ready)
|
||||
} else {
|
||||
self.dispatch_task(token / 2)
|
||||
}
|
||||
}
|
||||
|
||||
fn dispatch_io(&self, token: usize, ready: mio::Ready) {
|
||||
let mut reader = None;
|
||||
let mut writer = None;
|
||||
if let Some(io) = self.io_dispatch.borrow_mut().get_mut(token) {
|
||||
if ready.is_readable() {
|
||||
reader = io.reader.take();
|
||||
io.readiness.fetch_or(1, Ordering::Relaxed);
|
||||
}
|
||||
if ready.is_writable() {
|
||||
writer = io.writer.take();
|
||||
io.readiness.fetch_or(2, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
// TODO: don't notify the same task twice
|
||||
if let Some(reader) = reader {
|
||||
self.notify_handle(reader);
|
||||
}
|
||||
if let Some(writer) = writer {
|
||||
self.notify_handle(writer);
|
||||
}
|
||||
}
|
||||
|
||||
fn dispatch_task(&self, token: usize) {
|
||||
let (task, wake) = match self.task_dispatch.borrow_mut().get_mut(token) {
|
||||
Some(slot) => (slot.spawn.take(), slot.wake.clone()),
|
||||
None => return,
|
||||
};
|
||||
wake.0.set_readiness(mio::Ready::none()).unwrap();
|
||||
let mut task = match task {
|
||||
Some(task) => task,
|
||||
None => return,
|
||||
};
|
||||
let res = CURRENT_LOOP.set(self, || task.poll_future(wake));
|
||||
let mut dispatch = self.task_dispatch.borrow_mut();
|
||||
match res {
|
||||
Ok(Async::NotReady) => {
|
||||
assert!(dispatch[token].spawn.is_none());
|
||||
dispatch[token].spawn = Some(task);
|
||||
}
|
||||
Ok(Async::Ready(())) |
|
||||
Err(()) => {
|
||||
dispatch.remove(token).unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn consume_timeouts(&mut self, now: Instant) {
|
||||
while let Some(idx) = self.timer_wheel.borrow_mut().poll(now) {
|
||||
trace!("firing timeout: {}", idx);
|
||||
let handle = self.timeouts.borrow_mut()[idx].1.fire();
|
||||
if let Some(handle) = handle {
|
||||
self.notify_handle(handle);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Method used to notify a task handle.
|
||||
///
|
||||
/// Note that this should be used instead fo `handle.unpark()` to ensure
|
||||
/// that the `CURRENT_LOOP` variable is set appropriately.
|
||||
fn notify_handle(&self, handle: Task) {
|
||||
debug!("notifying a task handle");
|
||||
CURRENT_LOOP.set(&self, || handle.unpark());
|
||||
}
|
||||
|
||||
fn add_source(&self, source: &mio::Evented)
|
||||
-> io::Result<(Arc<AtomicUsize>, usize)> {
|
||||
debug!("adding a new I/O source");
|
||||
let sched = ScheduledIo {
|
||||
readiness: Arc::new(AtomicUsize::new(0)),
|
||||
reader: None,
|
||||
writer: None,
|
||||
};
|
||||
let mut dispatch = self.io_dispatch.borrow_mut();
|
||||
if dispatch.vacant_entry().is_none() {
|
||||
let amt = dispatch.len();
|
||||
dispatch.reserve_exact(amt);
|
||||
}
|
||||
let entry = dispatch.vacant_entry().unwrap();
|
||||
try!(self.io.register(source,
|
||||
mio::Token(TOKEN_START + entry.index() * 2),
|
||||
mio::Ready::readable() | mio::Ready::writable(),
|
||||
mio::PollOpt::edge()));
|
||||
Ok((sched.readiness.clone(), entry.insert(sched).index()))
|
||||
}
|
||||
|
||||
fn drop_source(&self, token: usize) {
|
||||
debug!("dropping I/O source: {}", token);
|
||||
self.io_dispatch.borrow_mut().remove(token).unwrap();
|
||||
}
|
||||
|
||||
fn schedule(&self, token: usize, wake: Task, dir: Direction) {
|
||||
debug!("scheduling direction for: {}", token);
|
||||
let to_call = {
|
||||
let mut dispatch = self.io_dispatch.borrow_mut();
|
||||
let sched = dispatch.get_mut(token).unwrap();
|
||||
let (slot, bit) = match dir {
|
||||
Direction::Read => (&mut sched.reader, 1),
|
||||
Direction::Write => (&mut sched.writer, 2),
|
||||
};
|
||||
if sched.readiness.load(Ordering::SeqCst) & bit != 0 {
|
||||
*slot = None;
|
||||
Some(wake)
|
||||
} else {
|
||||
*slot = Some(wake);
|
||||
None
|
||||
}
|
||||
};
|
||||
if let Some(to_call) = to_call {
|
||||
debug!("schedule immediately done");
|
||||
self.notify_handle(to_call);
|
||||
}
|
||||
}
|
||||
|
||||
fn add_timeout(&self, at: Instant) -> io::Result<(usize, Instant)> {
|
||||
let mut timeouts = self.timeouts.borrow_mut();
|
||||
if timeouts.vacant_entry().is_none() {
|
||||
let len = timeouts.len();
|
||||
timeouts.reserve_exact(len);
|
||||
}
|
||||
let entry = timeouts.vacant_entry().unwrap();
|
||||
let timeout = self.timer_wheel.borrow_mut().insert(at, entry.index());
|
||||
let when = *timeout.when();
|
||||
let entry = entry.insert((timeout, TimeoutState::NotFired));
|
||||
debug!("added a timeout: {}", entry.index());
|
||||
Ok((entry.index(), when))
|
||||
}
|
||||
|
||||
fn update_timeout(&self, token: usize, handle: Task) {
|
||||
debug!("updating a timeout: {}", token);
|
||||
let to_wake = self.timeouts.borrow_mut()[token].1.block(handle);
|
||||
if let Some(to_wake) = to_wake {
|
||||
self.notify_handle(to_wake);
|
||||
}
|
||||
}
|
||||
|
||||
fn cancel_timeout(&self, token: usize) {
|
||||
debug!("cancel a timeout: {}", token);
|
||||
let pair = self.timeouts.borrow_mut().remove(token);
|
||||
if let Some((timeout, _state)) = pair {
|
||||
self.timer_wheel.borrow_mut().cancel(&timeout);
|
||||
}
|
||||
}
|
||||
|
||||
fn spawn(&self, future: Box<Future<Item=(), Error=()>>) {
|
||||
let unpark = {
|
||||
let mut dispatch = self.task_dispatch.borrow_mut();
|
||||
if dispatch.vacant_entry().is_none() {
|
||||
let len = dispatch.len();
|
||||
dispatch.reserve_exact(len);
|
||||
}
|
||||
let entry = dispatch.vacant_entry().unwrap();
|
||||
let token = TOKEN_START + 2 * entry.index() + 1;
|
||||
let pair = mio::Registration::new(&self.io,
|
||||
mio::Token(token),
|
||||
mio::Ready::readable(),
|
||||
mio::PollOpt::level());
|
||||
let unpark = Arc::new(MySetReadiness(pair.1));
|
||||
let entry = entry.insert(ScheduledTask {
|
||||
spawn: Some(task::spawn(future)),
|
||||
wake: unpark,
|
||||
_registration: pair.0,
|
||||
});
|
||||
entry.get().wake.clone()
|
||||
};
|
||||
unpark.unpark();
|
||||
}
|
||||
|
||||
fn consume_queue(&self) {
|
||||
debug!("consuming notification queue");
|
||||
// TODO: can we do better than `.unwrap()` here?
|
||||
while let Some(msg) = self.rx.recv().unwrap() {
|
||||
self.notify(msg);
|
||||
}
|
||||
}
|
||||
|
||||
fn notify(&self, msg: Message) {
|
||||
match msg {
|
||||
Message::DropSource(tok) => self.drop_source(tok),
|
||||
Message::Schedule(tok, wake, dir) => self.schedule(tok, wake, dir),
|
||||
|
||||
Message::AddTimeout(at, slot) => {
|
||||
slot.try_produce(self.add_timeout(at))
|
||||
.expect("interference with try_produce on timeout");
|
||||
}
|
||||
Message::UpdateTimeout(t, handle) => self.update_timeout(t, handle),
|
||||
Message::CancelTimeout(t) => self.cancel_timeout(t),
|
||||
Message::Run(r) => r.call_box(self),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Handle {
|
||||
fn send(&self, msg: Message) {
|
||||
self.with_loop(|lp| {
|
||||
match lp {
|
||||
Some(lp) => {
|
||||
// Need to execute all existing requests first, to ensure
|
||||
// that our message is processed "in order"
|
||||
lp.consume_queue();
|
||||
lp.notify(msg);
|
||||
}
|
||||
None => {
|
||||
match self.tx.send(msg) {
|
||||
Ok(()) => {}
|
||||
|
||||
// This should only happen when there was an error
|
||||
// writing to the pipe to wake up the event loop,
|
||||
// hopefully that never happens
|
||||
Err(e) => {
|
||||
panic!("error sending message to event loop: {}", e)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn with_loop<F, R>(&self, f: F) -> R
|
||||
where F: FnOnce(Option<&Core>) -> R
|
||||
{
|
||||
if CURRENT_LOOP.is_set() {
|
||||
CURRENT_LOOP.with(|lp| {
|
||||
if lp.id == self.id {
|
||||
f(Some(lp))
|
||||
} else {
|
||||
f(None)
|
||||
}
|
||||
})
|
||||
} else {
|
||||
f(None)
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawns a new future into the event loop this handle is associated this.
|
||||
///
|
||||
/// This function takes a closure which is executed within the context of
|
||||
/// the I/O loop itself. The future returned by the closure will be
|
||||
/// scheduled on the event loop an run to completion.
|
||||
///
|
||||
/// Note that while the closure, `F`, requires the `Send` bound as it might
|
||||
/// cross threads, the future `R` does not.
|
||||
pub fn spawn<F, R>(&self, f: F)
|
||||
where F: FnOnce(&Pinned) -> R + Send + 'static,
|
||||
R: IntoFuture<Item=(), Error=()>,
|
||||
R::Future: 'static,
|
||||
{
|
||||
self.send(Message::Run(Box::new(|lp: &Core| {
|
||||
let f = f(&lp.pin());
|
||||
lp.spawn(Box::new(f.into_future()));
|
||||
})));
|
||||
}
|
||||
}
|
||||
|
||||
impl Pinned {
|
||||
/// Returns a reference to the underlying handle to the event loop.
|
||||
pub fn handle(&self) -> &Handle {
|
||||
&self.handle
|
||||
}
|
||||
|
||||
/// Spawns a new future on the event loop this pin is associated this.
|
||||
pub fn spawn<F>(&self, f: F)
|
||||
where F: Future<Item=(), Error=()> + 'static,
|
||||
{
|
||||
let inner = match self.futures.upgrade() {
|
||||
Some(inner) => inner,
|
||||
None => return,
|
||||
};
|
||||
inner.queue.borrow_mut().push(Box::new(f));
|
||||
inner.ready.set_readiness(mio::Ready::readable()).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
struct CoreFuture<T, U> {
|
||||
handle: Handle,
|
||||
data: Option<U>,
|
||||
result: Option<(Arc<Slot<io::Result<T>>>, slot::Token)>,
|
||||
}
|
||||
|
||||
impl<T, U> CoreFuture<T, U>
|
||||
where T: 'static,
|
||||
{
|
||||
fn poll<F, G>(&mut self, f: F, g: G) -> Poll<T, io::Error>
|
||||
where F: FnOnce(&Core, U) -> io::Result<T>,
|
||||
G: FnOnce(U, Arc<Slot<io::Result<T>>>) -> Message,
|
||||
{
|
||||
match self.result {
|
||||
Some((ref result, ref mut token)) => {
|
||||
result.cancel(*token);
|
||||
match result.try_consume() {
|
||||
Ok(Ok(t)) => return Ok(t.into()),
|
||||
Ok(Err(e)) => return Err(e),
|
||||
Err(_) => {}
|
||||
}
|
||||
let task = task::park();
|
||||
*token = result.on_full(move |_| {
|
||||
task.unpark();
|
||||
});
|
||||
Ok(Async::NotReady)
|
||||
}
|
||||
None => {
|
||||
let data = &mut self.data;
|
||||
let ret = self.handle.with_loop(|lp| {
|
||||
lp.map(|lp| f(lp, data.take().unwrap()))
|
||||
});
|
||||
if let Some(ret) = ret {
|
||||
debug!("loop future done immediately on event loop");
|
||||
return ret.map(|e| e.into())
|
||||
}
|
||||
debug!("loop future needs to send info to event loop");
|
||||
|
||||
let task = task::park();
|
||||
let result = Arc::new(Slot::new(None));
|
||||
let token = result.on_full(move |_| {
|
||||
task.unpark();
|
||||
});
|
||||
self.result = Some((result.clone(), token));
|
||||
self.handle.send(g(data.take().unwrap(), result));
|
||||
Ok(Async::NotReady)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl TimeoutState {
|
||||
fn block(&mut self, handle: Task) -> Option<Task> {
|
||||
match *self {
|
||||
TimeoutState::Fired => return Some(handle),
|
||||
_ => {}
|
||||
}
|
||||
*self = TimeoutState::Waiting(handle);
|
||||
None
|
||||
}
|
||||
|
||||
fn fire(&mut self) -> Option<Task> {
|
||||
match mem::replace(self, TimeoutState::Fired) {
|
||||
TimeoutState::NotFired => None,
|
||||
TimeoutState::Fired => panic!("fired twice?"),
|
||||
TimeoutState::Waiting(handle) => Some(handle),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct MySetReadiness(mio::SetReadiness);
|
||||
|
||||
impl Unpark for MySetReadiness {
|
||||
fn unpark(&self) {
|
||||
self.0.set_readiness(mio::Ready::readable())
|
||||
.expect("failed to set readiness");
|
||||
}
|
||||
}
|
||||
|
||||
trait FnBox: Send + 'static {
|
||||
fn call_box(self: Box<Self>, lp: &Core);
|
||||
}
|
||||
|
||||
impl<F: FnOnce(&Core) + Send + 'static> FnBox for F {
|
||||
fn call_box(self: Box<Self>, lp: &Core) {
|
||||
(*self)(lp)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,285 @@
|
||||
//! Readiness tracking streams, backing I/O objects.
|
||||
//!
|
||||
//! This module contains the core type which is used to back all I/O on object
|
||||
//! in `tokio-core`. The `PollEvented` type is the implementation detail of
|
||||
//! all I/O. Each `PollEvented` manages registration with a reactor,
|
||||
//! acquisition of a token, and tracking of the readiness state on the
|
||||
//! underlying I/O primitive.
|
||||
|
||||
use std::io::{self, Read, Write};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
use futures::{Future, Poll, Async};
|
||||
use mio;
|
||||
|
||||
use io::Io;
|
||||
use reactor::Handle;
|
||||
use reactor::io_token::{IoToken, IoTokenNew};
|
||||
|
||||
/// 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.
|
||||
///
|
||||
/// 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.
|
||||
pub struct PollEvented<E> {
|
||||
token: IoToken,
|
||||
handle: Handle,
|
||||
readiness: AtomicUsize,
|
||||
io: E,
|
||||
}
|
||||
|
||||
/// Future returned from `PollEvented::new` which will resolve to a
|
||||
/// `PollEvented`.
|
||||
pub struct PollEventedNew<E> {
|
||||
inner: IoTokenNew<E>,
|
||||
handle: Handle,
|
||||
}
|
||||
|
||||
impl<E> PollEvented<E>
|
||||
where E: mio::Evented + Send + 'static,
|
||||
{
|
||||
/// Creates a new readiness stream associated with the provided
|
||||
/// `loop_handle` and for the given `source`.
|
||||
///
|
||||
/// This method returns a future which will resolve to the readiness stream
|
||||
/// when it's ready.
|
||||
pub fn new(source: E, handle: &Handle) -> PollEventedNew<E> {
|
||||
PollEventedNew {
|
||||
inner: IoToken::new(source, handle),
|
||||
handle: handle.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<E> PollEvented<E> {
|
||||
/// Tests to see if this source is ready to be read from or not.
|
||||
///
|
||||
/// If this stream is not ready for a read then `NotReady` will be returned
|
||||
/// and the current task will be scheduled to receive a notification when
|
||||
/// the stream is readable again. In other words, this method is only safe
|
||||
/// to call from within the context of a future's task, typically done in a
|
||||
/// `Future::poll` method.
|
||||
pub fn poll_read(&self) -> Async<()> {
|
||||
if self.readiness.load(Ordering::SeqCst) & 1 != 0 {
|
||||
return Async::Ready(())
|
||||
}
|
||||
self.readiness.fetch_or(self.token.take_readiness(), Ordering::SeqCst);
|
||||
if self.readiness.load(Ordering::SeqCst) & 1 != 0 {
|
||||
Async::Ready(())
|
||||
} else {
|
||||
self.token.schedule_read(&self.handle);
|
||||
Async::NotReady
|
||||
}
|
||||
}
|
||||
|
||||
/// Tests to see if this source is ready to be written to or not.
|
||||
///
|
||||
/// If this stream is not ready for a write then `NotReady` will be returned
|
||||
/// and the current task will be scheduled to receive a notification when
|
||||
/// the stream is writable again. In other words, this method is only safe
|
||||
/// to call from within the context of a future's task, typically done in a
|
||||
/// `Future::poll` method.
|
||||
pub fn poll_write(&self) -> Async<()> {
|
||||
if self.readiness.load(Ordering::SeqCst) & 2 != 0 {
|
||||
return Async::Ready(())
|
||||
}
|
||||
self.readiness.fetch_or(self.token.take_readiness(), Ordering::SeqCst);
|
||||
if self.readiness.load(Ordering::SeqCst) & 2 != 0 {
|
||||
Async::Ready(())
|
||||
} else {
|
||||
self.token.schedule_write(&self.handle);
|
||||
Async::NotReady
|
||||
}
|
||||
}
|
||||
|
||||
/// Indicates to this source of events that the corresponding I/O object is
|
||||
/// no longer readable, but it needs to be.
|
||||
///
|
||||
/// This function, like `poll_read`, is only safe to call from the context
|
||||
/// of a future's task (typically in a `Future::poll` implementation). It
|
||||
/// informs this readiness stream that the underlying object is no longer
|
||||
/// readable, typically because a "would block" error was seen.
|
||||
///
|
||||
/// The flag indicating that this stream is readable is unset and the
|
||||
/// current task is scheduled to receive a notification when the stream is
|
||||
/// then again readable.
|
||||
pub fn need_read(&self) {
|
||||
self.readiness.fetch_and(!1, Ordering::SeqCst);
|
||||
self.token.schedule_read(&self.handle)
|
||||
}
|
||||
|
||||
/// Indicates to this source of events that the corresponding I/O object is
|
||||
/// no longer writable, but it needs to be.
|
||||
///
|
||||
/// This function, like `poll_write`, is only safe to call from the context
|
||||
/// of a future's task (typically in a `Future::poll` implementation). It
|
||||
/// informs this readiness stream that the underlying object is no longer
|
||||
/// writable, typically because a "would block" error was seen.
|
||||
///
|
||||
/// The flag indicating that this stream is writable is unset and the
|
||||
/// current task is scheduled to receive a notification when the stream is
|
||||
/// then again writable.
|
||||
pub fn need_write(&self) {
|
||||
self.readiness.fetch_and(!2, Ordering::SeqCst);
|
||||
self.token.schedule_write(&self.handle)
|
||||
}
|
||||
|
||||
/// Returns a reference to the event loop handle that this readiness stream
|
||||
/// is associated with.
|
||||
pub fn handle(&self) -> &Handle {
|
||||
&self.handle
|
||||
}
|
||||
|
||||
/// 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
|
||||
}
|
||||
}
|
||||
|
||||
impl<E: Read> Read for PollEvented<E> {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
if let Async::NotReady = self.poll_read() {
|
||||
return Err(mio::would_block())
|
||||
}
|
||||
let r = self.get_mut().read(buf);
|
||||
if is_wouldblock(&r) {
|
||||
self.need_read();
|
||||
}
|
||||
return r
|
||||
}
|
||||
}
|
||||
|
||||
impl<E: Write> Write for PollEvented<E> {
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
if let Async::NotReady = self.poll_write() {
|
||||
return Err(mio::would_block())
|
||||
}
|
||||
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() {
|
||||
return Err(mio::would_block())
|
||||
}
|
||||
let r = self.get_mut().flush();
|
||||
if is_wouldblock(&r) {
|
||||
self.need_write();
|
||||
}
|
||||
return r
|
||||
}
|
||||
}
|
||||
|
||||
impl<E: Read + Write> Io for PollEvented<E> {
|
||||
fn poll_read(&mut self) -> Async<()> {
|
||||
<PollEvented<E>>::poll_read(self)
|
||||
}
|
||||
|
||||
fn poll_write(&mut self) -> Async<()> {
|
||||
<PollEvented<E>>::poll_write(self)
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, E> Read for &'a PollEvented<E>
|
||||
where &'a E: Read,
|
||||
{
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
if let Async::NotReady = self.poll_read() {
|
||||
return Err(mio::would_block())
|
||||
}
|
||||
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 &'a E: Write,
|
||||
{
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
if let Async::NotReady = self.poll_write() {
|
||||
return Err(mio::would_block())
|
||||
}
|
||||
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() {
|
||||
return Err(mio::would_block())
|
||||
}
|
||||
let r = self.get_ref().flush();
|
||||
if is_wouldblock(&r) {
|
||||
self.need_write();
|
||||
}
|
||||
return r
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, E> Io for &'a PollEvented<E>
|
||||
where &'a E: Read + Write,
|
||||
{
|
||||
fn poll_read(&mut self) -> Async<()> {
|
||||
<PollEvented<E>>::poll_read(self)
|
||||
}
|
||||
|
||||
fn poll_write(&mut self) -> Async<()> {
|
||||
<PollEvented<E>>::poll_write(self)
|
||||
}
|
||||
}
|
||||
|
||||
fn is_wouldblock<T>(r: &io::Result<T>) -> bool {
|
||||
match *r {
|
||||
Ok(_) => false,
|
||||
Err(ref e) => e.kind() == io::ErrorKind::WouldBlock,
|
||||
}
|
||||
}
|
||||
|
||||
impl<E> Drop for PollEvented<E> {
|
||||
fn drop(&mut self) {
|
||||
self.token.drop_source(&self.handle);
|
||||
}
|
||||
}
|
||||
|
||||
impl<E> Future for PollEventedNew<E>
|
||||
where E: mio::Evented + Send + 'static,
|
||||
{
|
||||
type Item = PollEvented<E>;
|
||||
type Error = io::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<PollEvented<E>, io::Error> {
|
||||
let (io, token) = try_ready!(self.inner.poll());
|
||||
Ok(PollEvented {
|
||||
token: token,
|
||||
handle: self.handle.clone(),
|
||||
io: io,
|
||||
readiness: AtomicUsize::new(0),
|
||||
}.into())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
//! Support for creating futures that represent timeouts.
|
||||
//!
|
||||
//! This module contains the `Timeout` type which is a future that will resolve
|
||||
//! at a particular point in the future.
|
||||
|
||||
use std::io;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use futures::{Future, Poll, Async};
|
||||
|
||||
use reactor::Handle;
|
||||
use reactor::timeout_token::TimeoutToken;
|
||||
use io::IoFuture;
|
||||
|
||||
/// A future representing the notification that a timeout has occurred.
|
||||
///
|
||||
/// Timeouts are created through the `LoopHandle::timeout` or
|
||||
/// `LoopHandle::timeout_at` methods indicating when a timeout should fire at.
|
||||
/// Note that timeouts are not intended for high resolution timers, but rather
|
||||
/// they will likely fire some granularity after the exact instant that they're
|
||||
/// otherwise indicated to fire at.
|
||||
pub struct Timeout {
|
||||
token: TimeoutToken,
|
||||
handle: Handle,
|
||||
}
|
||||
|
||||
/// Future returned from `Timeout::new` and `Timeout::new_at` which will resolve
|
||||
/// to the actual `Timeout` itself.
|
||||
pub struct TimeoutNew {
|
||||
inner: IoFuture<Timeout>,
|
||||
}
|
||||
|
||||
impl Timeout {
|
||||
/// Creates a new timeout which will fire at `dur` time into the future.
|
||||
///
|
||||
/// This function will return a future that will resolve to the actual
|
||||
/// timeout object. The timeout object itself is then a future which will be
|
||||
/// set to fire at the specified point in the future.
|
||||
pub fn new(dur: Duration, handle: &Handle) -> TimeoutNew {
|
||||
Timeout::new_at(Instant::now() + dur, handle)
|
||||
}
|
||||
|
||||
/// Creates a new timeout which will fire at the time specified by `at`.
|
||||
///
|
||||
/// This function will return a future that will resolve to the actual
|
||||
/// timeout object. The timeout object itself is then a future which will be
|
||||
/// set to fire at the specified point in the future.
|
||||
pub fn new_at(at: Instant, handle: &Handle) -> TimeoutNew {
|
||||
let handle = handle.clone();
|
||||
TimeoutNew {
|
||||
inner: TimeoutToken::new(at, &handle).map(move |token| {
|
||||
Timeout {
|
||||
token: token,
|
||||
handle: handle,
|
||||
}
|
||||
}).boxed(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Future for Timeout {
|
||||
type Item = ();
|
||||
type Error = io::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<(), io::Error> {
|
||||
// TODO: is this fast enough?
|
||||
let now = Instant::now();
|
||||
if *self.token.when() <= now {
|
||||
Ok(Async::Ready(()))
|
||||
} else {
|
||||
self.token.update_timeout(&self.handle);
|
||||
Ok(Async::NotReady)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Future for TimeoutNew {
|
||||
type Item = Timeout;
|
||||
type Error = io::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<Timeout, io::Error> {
|
||||
self.inner.poll()
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Timeout {
|
||||
fn drop(&mut self) {
|
||||
self.token.cancel_timeout(&self.handle);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
use std::io;
|
||||
use std::time::Instant;
|
||||
|
||||
use futures::{Future, Poll};
|
||||
use futures::task;
|
||||
|
||||
use reactor::{Message, Core, Handle, CoreFuture};
|
||||
|
||||
/// Return value from the `Handle::add_timeout` method, a future that will
|
||||
/// resolve to a `TimeoutToken` to configure the behavior of that timeout.
|
||||
pub struct TimeoutTokenNew {
|
||||
inner: CoreFuture<(usize, Instant), Instant>,
|
||||
}
|
||||
|
||||
/// A token that identifies an active timeout.
|
||||
pub struct TimeoutToken {
|
||||
token: usize,
|
||||
when: Instant,
|
||||
}
|
||||
|
||||
impl TimeoutToken {
|
||||
/// Adds a new timeout to get fired at the specified instant, notifying the
|
||||
/// specified task.
|
||||
pub fn new(at: Instant, handle: &Handle) -> TimeoutTokenNew {
|
||||
TimeoutTokenNew {
|
||||
inner: CoreFuture {
|
||||
handle: handle.clone(),
|
||||
data: Some(at),
|
||||
result: None,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the instant in time when this timeout token will "fire".
|
||||
///
|
||||
/// Note that this instant may *not* be the instant that was passed in when
|
||||
/// the timeout was created. The event loop does not support high resolution
|
||||
/// timers, so the exact resolution of when a timeout may fire may be
|
||||
/// slightly fudged.
|
||||
pub fn when(&self) -> &Instant {
|
||||
&self.when
|
||||
}
|
||||
|
||||
/// Updates a previously added timeout to notify a new task instead.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This method will panic if the timeout specified was not created by this
|
||||
/// loop handle's `add_timeout` method.
|
||||
pub fn update_timeout(&self, handle: &Handle) {
|
||||
handle.send(Message::UpdateTimeout(self.token, task::park()))
|
||||
}
|
||||
|
||||
/// Cancel a previously added timeout.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// This method will panic if the timeout specified was not created by this
|
||||
/// loop handle's `add_timeout` method.
|
||||
pub fn cancel_timeout(&self, handle: &Handle) {
|
||||
debug!("cancel timeout {}", self.token);
|
||||
handle.send(Message::CancelTimeout(self.token))
|
||||
}
|
||||
}
|
||||
|
||||
impl Future for TimeoutTokenNew {
|
||||
type Item = TimeoutToken;
|
||||
type Error = io::Error;
|
||||
|
||||
fn poll(&mut self) -> Poll<TimeoutToken, io::Error> {
|
||||
let (t, i) = try_ready!(self.inner.poll(Core::add_timeout,
|
||||
Message::AddTimeout));
|
||||
Ok(TimeoutToken {
|
||||
token: t,
|
||||
when: i,
|
||||
}.into())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user