diff --git a/Cargo.toml b/Cargo.toml index d4b2f936f..d34347991 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,7 +20,6 @@ appveyor = { repository = "alexcrichton/tokio" } bytes = "0.4" log = "0.3" mio = "0.6.10" -scoped-tls = "0.1.0" slab = "0.4" iovec = "0.1" tokio-io = "0.1" diff --git a/src/lib.rs b/src/lib.rs index 3c6f1cc6b..fff2c0616 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -86,6 +86,7 @@ #![doc(html_root_url = "https://docs.rs/tokio-core/0.1")] #![deny(missing_docs)] #![deny(warnings)] +#![allow(unused_macros)] extern crate bytes; #[macro_use] @@ -96,9 +97,6 @@ extern crate slab; #[macro_use] extern crate tokio_io; -#[macro_use] -extern crate scoped_tls; - #[macro_use] extern crate log; diff --git a/src/reactor/io_token.rs b/src/reactor/io_token.rs index 285a50772..4fa1a0ab6 100644 --- a/src/reactor/io_token.rs +++ b/src/reactor/io_token.rs @@ -1,17 +1,14 @@ -use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::Ordering; use std::io; -use futures::task; use mio::event::Evented; -use reactor::{Message, Remote, Handle, Direction}; +use reactor::{Remote, Handle, Direction}; /// A token that identifies an active I/O resource. pub struct IoToken { token: usize, - // TODO: can we avoid this allocation? It's kind of a bummer... - readiness: Arc, + handle: Remote, } impl IoToken { @@ -32,32 +29,45 @@ impl IoToken { /// associated with has gone away, or if there is an error communicating /// with the event loop. pub fn new(source: &Evented, handle: &Handle) -> io::Result { - match handle.inner.upgrade() { + match handle.remote.inner.upgrade() { Some(inner) => { - let (ready, token) = try!(inner.borrow_mut().add_source(source)); - Ok(IoToken { token: token, readiness: ready }) + let token = try!(inner.add_source(source)); + let handle = handle.remote().clone(); + + Ok(IoToken { token, handle }) } None => Err(io::Error::new(io::ErrorKind::Other, "event loop gone")), } } - /// Consumes the last readiness notification the token this source is for + /// Returns a reference to the remote handle + pub fn remote(&self) -> &Remote { + &self.handle + } + + /// 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. + /// + /// 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) + let inner = match self.handle.inner.upgrade() { + Some(inner) => inner, + None => return 0, + }; + + let io_dispatch = inner.io_dispatch.read().unwrap(); + io_dispatch[self.token].readiness.swap(0, Ordering::SeqCst) } /// Schedule the current future task to receive a notification when the @@ -82,8 +92,13 @@ impl IoToken { /// /// This function will also panic if there is not a currently running future /// task. - pub fn schedule_read(&self, handle: &Remote) { - handle.send(Message::Schedule(self.token, task::current(), Direction::Read)); + pub fn schedule_read(&self) { + let inner = match self.handle.inner.upgrade() { + Some(inner) => inner, + None => return, + }; + + inner.schedule(self.token, Direction::Read); } /// Schedule the current future task to receive a notification when the @@ -109,8 +124,13 @@ impl IoToken { /// /// This function will also panic if there is not a currently running future /// task. - pub fn schedule_write(&self, handle: &Remote) { - handle.send(Message::Schedule(self.token, task::current(), Direction::Write)); + pub fn schedule_write(&self) { + let inner = match self.handle.inner.upgrade() { + Some(inner) => inner, + None => return, + }; + + inner.schedule(self.token, Direction::Write); } /// Unregister all information associated with a token on an event loop, @@ -135,7 +155,12 @@ impl IoToken { /// 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: &Remote) { - handle.send(Message::DropSource(self.token)); + pub fn drop_source(&self) { + let inner = match self.handle.inner.upgrade() { + Some(inner) => inner, + None => return, + }; + + inner.drop_source(self.token) } } diff --git a/src/reactor/mod.rs b/src/reactor/mod.rs index aba1bc9f9..44f0afc16 100644 --- a/src/reactor/mod.rs +++ b/src/reactor/mod.rs @@ -4,18 +4,15 @@ //! happening in `tokio-core`. This reactor (or event loop) is used to drive I/O //! resources. -use std::cell::RefCell; use std::fmt; use std::io::{self, ErrorKind}; -use std::rc::{Rc, Weak}; -use std::sync::Arc; +use std::sync::{Arc, Weak, RwLock}; use std::sync::atomic::{AtomicUsize, ATOMIC_USIZE_INIT, Ordering}; -use std::time::{Instant, Duration}; +use std::time::{Duration}; use futures::{Future, Async}; -use futures::executor::{self, Spawn, Notify}; -use futures::sync::mpsc; -use futures::task::Task; +use futures::executor::{self, Notify}; +use futures::task::{AtomicTask}; use mio; use mio::event::Evented; use slab::Slab; @@ -25,8 +22,8 @@ mod io_token; mod poll_evented; pub use self::poll_evented::PollEvented; +/// Global counter used to assign unique IDs to reactor instances. static NEXT_LOOP_ID: AtomicUsize = ATOMIC_USIZE_INIT; -scoped_thread_local!(static CURRENT_LOOP: Core); /// An event loop. /// @@ -34,29 +31,29 @@ scoped_thread_local!(static CURRENT_LOOP: Core); /// 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 { + /// Reuse the `mio::Events` value across calls to poll. events: mio::Events, - tx: mpsc::UnboundedSender, - rx: RefCell>>, - _rx_registration: mio::Registration, - rx_readiness: Arc, - inner: Rc>, + /// State shared between the reactor and the handles. + inner: Arc, - // 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. + /// 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, } struct Inner { + /// Unique identifier referencing this reactor. id: usize, + + /// The underlying system event queue. io: mio::Poll, - // Dispatch slabs for I/O and futures events - io_dispatch: Slab, + /// Dispatch slabs for I/O and futures events + io_dispatch: RwLock>, } /// An unique ID for a Core @@ -76,7 +73,7 @@ pub struct CoreId(usize); #[derive(Clone)] pub struct Remote { id: usize, - tx: mpsc::UnboundedSender, + inner: Weak, } /// A non-sendable handle to an event loop, useful for manufacturing instances @@ -84,13 +81,12 @@ pub struct Remote { #[derive(Clone)] pub struct Handle { remote: Remote, - inner: Weak>, } struct ScheduledIo { - readiness: Arc, - reader: Option, - writer: Option, + readiness: AtomicUsize, + reader: AtomicTask, + writer: AtomicTask, } enum Direction { @@ -98,50 +94,40 @@ enum Direction { Write, } -enum Message { - DropSource(usize), - Schedule(usize, Task, Direction), - Run(Box), -} - -const TOKEN_MESSAGES: mio::Token = mio::Token(0); const TOKEN_FUTURE: mio::Token = mio::Token(1); const TOKEN_START: usize = 2; +fn _assert_kinds() { + fn _assert() {} + + _assert::(); + _assert::(); +} + impl Core { /// Creates a new event loop, returning any error that happened during the /// creation. pub fn new() -> io::Result { + // Create the I/O poller let io = try!(mio::Poll::new()); + + // Create a registration for unblocking the reactor when the "run" + // future becomes ready. let future_pair = mio::Registration::new2(); try!(io.register(&future_pair.0, TOKEN_FUTURE, mio::Ready::readable(), mio::PollOpt::level())); - let (tx, rx) = mpsc::unbounded(); - let channel_pair = mio::Registration::new2(); - try!(io.register(&channel_pair.0, - TOKEN_MESSAGES, - mio::Ready::readable(), - mio::PollOpt::level())); - let rx_readiness = Arc::new(MySetReadiness(channel_pair.1)); - rx_readiness.notify(0); Ok(Core { events: mio::Events::with_capacity(1024), - tx: tx, - rx: RefCell::new(executor::spawn(rx)), - _rx_registration: channel_pair.0, - rx_readiness: rx_readiness, - _future_registration: future_pair.0, future_readiness: Arc::new(MySetReadiness(future_pair.1)), - - inner: Rc::new(RefCell::new(Inner { + inner: Arc::new(Inner { id: NEXT_LOOP_ID.fetch_add(1, Ordering::Relaxed), io: io, - io_dispatch: Slab::with_capacity(1), - })), + io_dispatch: RwLock::new(Slab::with_capacity(1)), + }), }) } @@ -152,18 +138,16 @@ impl Core { /// 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 { - remote: self.remote(), - inner: Rc::downgrade(&self.inner), - } + let remote = self.remote(); + Handle { remote } } /// Generates a remote handle to this event loop which can be used to spawn /// tasks from other threads into this event loop. pub fn remote(&self) -> Remote { Remote { - id: self.inner.borrow().id, - tx: self.tx.clone(), + id: self.inner.id, + inner: Arc::downgrade(&self.inner), } } @@ -193,9 +177,7 @@ impl Core { loop { if future_fired { - let res = try!(CURRENT_LOOP.set(self, || { - task.poll_future_notify(&self.future_readiness, 0) - })); + let res = task.poll_future_notify(&self.future_readiness, 0)?; if let Async::Ready(e) = res { return Ok(e) } @@ -217,19 +199,14 @@ impl Core { } fn poll(&mut self, max_wait: Option) -> bool { - let start = Instant::now(); - // Block waiting for an event to happen, peeling out how many events // happened. - let amt = match self.inner.borrow_mut().io.poll(&mut self.events, max_wait) { - Ok(a) => a, + match self.inner.io.poll(&mut self.events, max_wait) { + Ok(_) => {} Err(ref e) if e.kind() == ErrorKind::Interrupted => return false, + // TODO: This should return an io::Result instead of panic. Err(e) => panic!("error in poll: {}", e), - }; - - let after_poll = Instant::now(); - debug!("loop poll - {:?}", after_poll - start); - debug!("loop time - {:?}", after_poll); + } // Process all the events that came in, dispatching appropriately let mut fired = false; @@ -238,86 +215,35 @@ impl Core { let token = event.token(); trace!("event {:?} {:?}", event.readiness(), event.token()); - if token == TOKEN_MESSAGES { - self.rx_readiness.0.set_readiness(mio::Ready::empty()).unwrap(); - CURRENT_LOOP.set(&self, || self.consume_queue()); - } else if token == TOKEN_FUTURE { + if token == TOKEN_FUTURE { self.future_readiness.0.set_readiness(mio::Ready::empty()).unwrap(); fired = true; } else { self.dispatch(token, event.readiness()); } } - debug!("loop process - {} events, {:?}", amt, after_poll.elapsed()); + return fired } fn dispatch(&mut self, token: mio::Token, ready: mio::Ready) { let token = usize::from(token) - TOKEN_START; - self.dispatch_io(token, ready) - } + let io_dispatch = self.inner.io_dispatch.read().unwrap(); - fn dispatch_io(&mut self, token: usize, ready: mio::Ready) { - let mut reader = None; - let mut writer = None; - let mut inner = self.inner.borrow_mut(); - if let Some(io) = inner.io_dispatch.get_mut(token) { + if let Some(io) = io_dispatch.get(token) { io.readiness.fetch_or(ready2usize(ready), Ordering::Relaxed); if ready.is_writable() { - writer = io.writer.take(); + io.writer.notify(); } if !(ready & (!mio::Ready::writable())).is_empty() { - reader = io.reader.take(); + io.reader.notify(); } } - drop(inner); - // 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); - } - } - - /// Method used to notify a task handle. - /// - /// Note that this should be used instead of `handle.notify()` 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.notify()); - } - - fn consume_queue(&self) { - debug!("consuming notification queue"); - // TODO: can we do better than `.unwrap()` here? - loop { - let msg = self.rx.borrow_mut().poll_stream_notify(&self.rx_readiness, 0).unwrap(); - match msg { - Async::Ready(Some(msg)) => self.notify(msg), - Async::NotReady | - Async::Ready(None) => break, - } - } - } - - fn notify(&self, msg: Message) { - match msg { - Message::DropSource(tok) => self.inner.borrow_mut().drop_source(tok), - Message::Schedule(tok, wake, dir) => { - let task = self.inner.borrow_mut().schedule(tok, wake, dir); - if let Some(task) = task { - self.notify_handle(task); - } - } - Message::Run(r) => r.call_box(self), - } } /// Get the ID of this loop pub fn id(&self) -> CoreId { - CoreId(self.inner.borrow().id) + CoreId(self.inner.id) } } @@ -330,111 +256,79 @@ impl fmt::Debug for Core { } impl Inner { - fn add_source(&mut self, source: &Evented) - -> io::Result<(Arc, usize)> { - debug!("adding a new I/O source"); - let sched = ScheduledIo { - readiness: Arc::new(AtomicUsize::new(0)), - reader: None, - writer: None, - }; - if self.io_dispatch.len() == self.io_dispatch.capacity() { - let amt = self.io_dispatch.len(); - self.io_dispatch.reserve_exact(amt); - } - let entry = self.io_dispatch.vacant_entry(); - let key = entry.key(); + /// Register an I/O resource with the reactor. + /// + /// The registration token is returned. + fn add_source(&self, source: &Evented) + -> io::Result + { + // Acquire a write lock + let key = self.io_dispatch.write().unwrap() + .insert(ScheduledIo { + readiness: AtomicUsize::new(0), + reader: AtomicTask::new(), + writer: AtomicTask::new(), + }); + try!(self.io.register(source, mio::Token(TOKEN_START + key), mio::Ready::readable() | mio::Ready::writable() | platform::all(), mio::PollOpt::edge())); - let sched = entry.insert(sched); - Ok((sched.readiness.clone(), key)) + + Ok(key) } - fn deregister_source(&mut self, source: &Evented) -> io::Result<()> { + fn deregister_source(&self, source: &Evented) -> io::Result<()> { self.io.deregister(source) } - fn drop_source(&mut self, token: usize) { + fn drop_source(&self, token: usize) { debug!("dropping I/O source: {}", token); - self.io_dispatch.remove(token); + self.io_dispatch.write().unwrap().remove(token); } - fn schedule(&mut self, token: usize, wake: Task, dir: Direction) - -> Option { + /// Registers interest in the I/O resource associated with `token`. + fn schedule(&self, token: usize, dir: Direction) { debug!("scheduling direction for: {}", token); - let sched = self.io_dispatch.get_mut(token).unwrap(); - let (slot, ready) = match dir { - Direction::Read => (&mut sched.reader, !mio::Ready::writable()), - Direction::Write => (&mut sched.writer, mio::Ready::writable()), + let io_dispatch = self.io_dispatch.read().unwrap(); + let sched = io_dispatch.get(token).unwrap(); + + let (task, ready) = match dir { + Direction::Read => (&sched.reader, !mio::Ready::writable()), + Direction::Write => (&sched.writer, mio::Ready::writable()), }; + + task.register(); + if sched.readiness.load(Ordering::SeqCst) & ready2usize(ready) != 0 { - debug!("cancelling block"); - *slot = None; - Some(wake) - } else { - debug!("blocking"); - *slot = Some(wake); - None + task.notify(); } } } impl Remote { - fn send(&self, msg: Message) { - self.with_loop(|lp| { - match lp { - Some(lp) => { - // We want to make sure that all messages are received in - // order, so we need to consume pending messages before - // delivering this message to the core. The actually - // `consume_queue` function, however, can be somewhat slow - // right now where receiving on a channel will acquire a - // lock and block the current task. - // - // To speed this up check the message queue's readiness as a - // sort of preflight check to see if we've actually got any - // messages. This should just involve some atomics and if it - // comes back false then we know for sure there are no - // pending messages, so we can immediately deliver our - // message. - if lp.rx_readiness.0.readiness().is_readable() { - lp.consume_queue(); - } - lp.notify(msg); - } - None => { - match self.tx.unbounded_send(msg) { - Ok(()) => {} - - // TODO: this error should punt upwards and we should - // notify the caller that the message wasn't - // received. This is tokio-core#17 - Err(e) => drop(e), - } - } - } - }) + /// Return the ID of the represented Core + pub fn id(&self) -> CoreId { + CoreId(self.id) } - fn with_loop(&self, f: F) -> R - where F: FnOnce(Option<&Core>) -> R - { - if CURRENT_LOOP.is_set() { - CURRENT_LOOP.with(|lp| { - let same = lp.inner.borrow().id == self.id; - if same { - f(Some(lp)) - } else { - f(None) - } - }) - } else { - f(None) - } + /// Attempts to "promote" this remote to a handle, if possible. + /// + /// This function is intended for structures which typically work through a + /// `Remote` but want to optimize runtime when the remote doesn't actually + /// leave the thread of the original reactor. This will attempt to return a + /// handle if the `Remote` is on the same thread as the event loop and the + /// event loop is running. + /// + /// If this `Remote` has moved to a different thread or if the event loop is + /// running, then `None` may be returned. If you need to guarantee access to + /// a `Handle`, then you can call this function and fall back to using + /// `spawn` above if it returns `None`. + pub fn handle(&self) -> Option { + let remote = self.clone(); + Some(Handle { remote } ) } /// Spawns a new future into the event loop this remote is associated with. @@ -454,41 +348,8 @@ impl Remote { pub(crate) fn run(&self, f: F) where F: FnOnce(&Handle) + Send + 'static, { - self.send(Message::Run(Box::new(|lp: &Core| { - f(&lp.handle()); - }))); - } - - /// Return the ID of the represented Core - pub fn id(&self) -> CoreId { - CoreId(self.id) - } - - /// Attempts to "promote" this remote to a handle, if possible. - /// - /// This function is intended for structures which typically work through a - /// `Remote` but want to optimize runtime when the remote doesn't actually - /// leave the thread of the original reactor. This will attempt to return a - /// handle if the `Remote` is on the same thread as the event loop and the - /// event loop is running. - /// - /// If this `Remote` has moved to a different thread or if the event loop is - /// running, then `None` may be returned. If you need to guarantee access to - /// a `Handle`, then you can call this function and fall back to using - /// `spawn` above if it returns `None`. - pub fn handle(&self) -> Option { - if CURRENT_LOOP.is_set() { - CURRENT_LOOP.with(|lp| { - let same = lp.inner.borrow().id == self.id; - if same { - Some(lp.handle()) - } else { - None - } - }) - } else { - None - } + let handle = self.handle().unwrap(); + f(&handle); } } diff --git a/src/reactor/poll_evented.rs b/src/reactor/poll_evented.rs index 560808ca0..aa19803f0 100644 --- a/src/reactor/poll_evented.rs +++ b/src/reactor/poll_evented.rs @@ -65,7 +65,6 @@ use reactor::io_token::IoToken; /// otherwise probably avoid using two tasks on the same `PollEvented`. pub struct PollEvented { token: IoToken, - handle: Remote, readiness: AtomicUsize, io: E, } @@ -85,9 +84,10 @@ impl PollEvented { /// This method returns a future which will resolve to the readiness stream /// when it's ready. pub fn new(io: E, handle: &Handle) -> io::Result> { + let token = IoToken::new(&io, handle)?; + Ok(PollEvented { - token: try!(IoToken::new(&io, handle)), - handle: handle.remote().clone(), + token, readiness: AtomicUsize::new(0), io: io, }) @@ -106,11 +106,12 @@ impl PollEvented { /// method is called, and will likely return an error if this `PollEvented` /// was created on a separate event loop from the `handle` specified. pub fn deregister(self, handle: &Handle) -> io::Result<()> { - let inner = match handle.inner.upgrade() { + let inner = match handle.remote.inner.upgrade() { Some(inner) => inner, None => return Ok(()), }; - let ret = inner.borrow_mut().deregister_source(&self.io); + + let ret = inner.deregister_source(&self.io); return ret } } @@ -223,7 +224,7 @@ impl PollEvented { pub fn need_read(&self) { let bits = super::ready2usize(super::read_ready()); self.readiness.fetch_and(!bits, Ordering::SeqCst); - self.token.schedule_read(&self.handle) + self.token.schedule_read(); } /// Indicates to this source of events that the corresponding I/O object is @@ -249,13 +250,13 @@ impl PollEvented { pub fn need_write(&self) { let bits = super::ready2usize(Ready::writable()); self.readiness.fetch_and(!bits, Ordering::SeqCst); - self.token.schedule_write(&self.handle) + self.token.schedule_write(); } /// Returns a reference to the event loop handle that this readiness stream /// is associated with. pub fn remote(&self) -> &Remote { - &self.handle + self.token.remote() } /// Returns a shared reference to the underlying I/O object this readiness @@ -380,6 +381,6 @@ fn is_wouldblock(r: &io::Result) -> bool { impl Drop for PollEvented { fn drop(&mut self) { - self.token.drop_source(&self.handle); + self.token.drop_source(); } }