From 69975fb9601bbb21659db283d888470733bae660 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Wed, 20 Nov 2019 00:05:14 -0800 Subject: [PATCH] Refactor the I/O driver, extracting slab to `tokio::util`. (#1792) The I/O driver is made private and moved to `tokio::io::driver`. `Registration` is moved to `tokio::io::Registration` and `PollEvented` is moved to `tokio::io::PollEvented`. Additionally, the concurrent slab used by the I/O driver is cleaned up and extracted to `tokio::util::slab`, allowing it to eventually be used by other types. --- .../{net/driver/reactor => io/driver}/mod.rs | 177 ++----- tokio/src/{net => io}/driver/platform.rs | 0 tokio/src/io/driver/scheduled_io.rs | 142 ++++++ tokio/src/io/mod.rs | 10 + tokio/src/{net/util => io}/poll_evented.rs | 6 +- tokio/src/{net/driver => io}/registration.rs | 30 +- tokio/src/lib.rs | 4 +- tokio/src/net/driver/mod.rs | 141 ------ tokio/src/net/driver/reactor/dispatch/iter.rs | 53 -- tokio/src/net/driver/reactor/dispatch/mod.rs | 36 -- tokio/src/net/driver/reactor/dispatch/pack.rs | 89 ---- .../reactor/dispatch/page/scheduled_io.rs | 169 ------- .../net/driver/reactor/dispatch/page/stack.rs | 151 ------ .../driver/reactor/dispatch/sharded_slab.rs | 274 ---------- .../driver/reactor/dispatch/tests/loom/mod.rs | 204 -------- .../dispatch/tests/loom/single_shard.rs | 181 ------- .../reactor/dispatch/tests/loom/small_slab.rs | 473 ------------------ .../net/driver/reactor/dispatch/tests/mod.rs | 30 -- tokio/src/net/driver/reactor/dispatch/tid.rs | 168 ------- .../reactor/reactor/sharded_slab/page/slot.rs | 102 ---- tokio/src/net/mod.rs | 5 - tokio/src/net/tcp/listener.rs | 2 +- tokio/src/net/tcp/stream.rs | 3 +- tokio/src/net/udp/socket.rs | 2 +- tokio/src/net/unix/datagram.rs | 2 +- tokio/src/net/unix/listener.rs | 2 +- tokio/src/net/unix/stream.rs | 3 +- tokio/src/net/util/mod.rs | 4 - tokio/src/process/unix/mod.rs | 2 +- tokio/src/process/windows.rs | 2 +- tokio/src/runtime/io.rs | 6 +- tokio/src/signal/unix.rs | 3 +- tokio/src/sync/task/atomic_waker.rs | 2 + tokio/src/util/bit.rs | 86 ++++ tokio/src/util/mod.rs | 15 +- tokio/src/util/slab/addr.rs | 155 ++++++ tokio/src/util/slab/entry.rs | 7 + tokio/src/util/slab/generation.rs | 32 ++ tokio/src/util/slab/mod.rs | 109 ++++ .../page/mod.rs => util/slab/page.rs} | 150 ++---- tokio/src/util/slab/shard.rs | 108 ++++ tokio/src/util/slab/slot.rs | 42 ++ tokio/src/util/slab/stack.rs | 58 +++ tokio/src/util/slab/tests/loom_slab.rs | 327 ++++++++++++ tokio/src/util/slab/tests/loom_stack.rs | 88 ++++ tokio/src/util/slab/tests/mod.rs | 2 + tokio/tests/{net_driver.rs => io_driver.rs} | 18 +- tokio/tests/io_driver_drop.rs | 44 ++ tokio/tests/net_driver_drop.rs | 47 -- 49 files changed, 1351 insertions(+), 2415 deletions(-) rename tokio/src/{net/driver/reactor => io/driver}/mod.rs (66%) rename tokio/src/{net => io}/driver/platform.rs (100%) create mode 100644 tokio/src/io/driver/scheduled_io.rs rename tokio/src/{net/util => io}/poll_evented.rs (99%) rename tokio/src/{net/driver => io}/registration.rs (92%) delete mode 100644 tokio/src/net/driver/mod.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/iter.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/mod.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/pack.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/page/scheduled_io.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/page/stack.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/sharded_slab.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/tests/loom/mod.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/tests/loom/single_shard.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/tests/loom/small_slab.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/tests/mod.rs delete mode 100644 tokio/src/net/driver/reactor/dispatch/tid.rs delete mode 100644 tokio/src/net/driver/reactor/reactor/sharded_slab/page/slot.rs delete mode 100644 tokio/src/net/util/mod.rs create mode 100644 tokio/src/util/bit.rs create mode 100644 tokio/src/util/slab/addr.rs create mode 100644 tokio/src/util/slab/entry.rs create mode 100644 tokio/src/util/slab/generation.rs create mode 100644 tokio/src/util/slab/mod.rs rename tokio/src/{net/driver/reactor/dispatch/page/mod.rs => util/slab/page.rs} (54%) create mode 100644 tokio/src/util/slab/shard.rs create mode 100644 tokio/src/util/slab/slot.rs create mode 100644 tokio/src/util/slab/stack.rs create mode 100644 tokio/src/util/slab/tests/loom_slab.rs create mode 100644 tokio/src/util/slab/tests/loom_stack.rs create mode 100644 tokio/src/util/slab/tests/mod.rs rename tokio/tests/{net_driver.rs => io_driver.rs} (83%) create mode 100644 tokio/tests/io_driver_drop.rs delete mode 100644 tokio/tests/net_driver_drop.rs diff --git a/tokio/src/net/driver/reactor/mod.rs b/tokio/src/io/driver/mod.rs similarity index 66% rename from tokio/src/net/driver/reactor/mod.rs rename to tokio/src/io/driver/mod.rs index 8bcded411..4f47dc389 100644 --- a/tokio/src/net/driver/reactor/mod.rs +++ b/tokio/src/io/driver/mod.rs @@ -1,31 +1,24 @@ +pub(crate) mod platform; + +mod scheduled_io; +pub(crate) use scheduled_io::ScheduledIo; // pub(crate) for tests + use crate::loom::sync::atomic::AtomicUsize; -use crate::net::driver::platform; use crate::runtime::{Park, Unpark}; - -use std::sync::atomic::Ordering::SeqCst; - -mod dispatch; -use dispatch::SingleShard; -pub(crate) use dispatch::MAX_SOURCES; +use crate::util::slab::{Address, Slab}; use mio::event::Evented; use std::cell::RefCell; +use std::fmt; use std::io; use std::marker::PhantomData; -#[cfg(all(unix, not(target_os = "fuchsia")))] -use std::os::unix::io::{AsRawFd, RawFd}; use std::sync::{Arc, Weak}; +use std::sync::atomic::Ordering::SeqCst; use std::task::Waker; use std::time::Duration; -use std::{fmt, usize}; -/// 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 { +/// I/O driver, backed by Mio +pub(crate) struct Driver { /// Reuse the `mio::Events` value across calls to poll. events: mio::Events, @@ -35,33 +28,18 @@ pub struct Reactor { _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. +/// A reference to an I/O driver #[derive(Clone)] -pub struct Handle { +pub(crate) struct Handle { inner: Weak, } -/// 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: (), -} - pub(super) struct Inner { /// The underlying system event queue. io: mio::Poll, /// Dispatch slabs for I/O and futures events - // TODO(eliza): once worker threads are available, replace this with a - // properly sharded slab. - pub(super) io_dispatch: SingleShard, + pub(super) io_dispatch: Slab, /// The number of sources in `io_dispatch`. n_sources: AtomicUsize, @@ -81,7 +59,7 @@ thread_local! { static CURRENT_REACTOR: RefCell> = RefCell::new(None) } -const TOKEN_WAKEUP: mio::Token = mio::Token(MAX_SOURCES); +const TOKEN_WAKEUP: mio::Token = mio::Token(Address::NULL); fn _assert_kinds() { fn _assert() {} @@ -89,11 +67,11 @@ fn _assert_kinds() { _assert::(); } -// ===== impl Reactor ===== +// ===== impl Driver ===== #[derive(Debug)] /// Guard that resets current reactor on drop. -pub struct DefaultGuard<'a> { +pub(crate) struct DefaultGuard<'a> { _lifetime: PhantomData<&'a u8>, } @@ -107,7 +85,7 @@ impl Drop for DefaultGuard<'_> { } /// Sets handle for a default reactor, returning guard that unsets it on drop. -pub fn set_default(handle: &Handle) -> DefaultGuard<'_> { +pub(crate) fn set_default(handle: &Handle) -> DefaultGuard<'_> { CURRENT_REACTOR.with(|current| { let mut current = current.borrow_mut(); @@ -125,10 +103,10 @@ pub fn set_default(handle: &Handle) -> DefaultGuard<'_> { } } -impl Reactor { +impl Driver { /// Creates a new event loop, returning any error that happened during the /// creation. - pub fn new() -> io::Result { + pub(crate) fn new() -> io::Result { let io = mio::Poll::new()?; let wakeup_pair = mio::Registration::new2(); @@ -139,12 +117,12 @@ impl Reactor { mio::PollOpt::level(), )?; - Ok(Reactor { + Ok(Driver { events: mio::Events::with_capacity(1024), _wakeup_registration: wakeup_pair.0, inner: Arc::new(Inner { io, - io_dispatch: SingleShard::new(), + io_dispatch: Slab::new(), n_sources: AtomicUsize::new(0), wakeup: wakeup_pair.1, }), @@ -157,52 +135,13 @@ impl Reactor { /// 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 { + pub(crate) fn handle(&self) -> Handle { Handle { 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) -> io::Result { - 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.n_sources.load(SeqCst) == 0 - } - - fn poll(&mut self, max_wait: Option) -> io::Result<()> { + fn turn(&mut self, max_wait: Option) -> 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) { @@ -232,13 +171,15 @@ impl Reactor { let mut rd = None; let mut wr = None; - let io = match self.inner.io_dispatch.get(token.0) { + let address = Address::from_usize(token.0); + + let io = match self.inner.io_dispatch.get(address) { Some(io) => io, None => return, }; if io - .set_readiness(token.0, |curr| curr | ready.as_usize()) + .set_readiness(address, |curr| curr | ready.as_usize()) .is_err() { // token no longer valid! @@ -263,14 +204,7 @@ impl Reactor { } } -#[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 { +impl Park for Driver { type Unpark = Handle; type Error = io::Error; @@ -289,9 +223,9 @@ impl Park for Reactor { } } -impl fmt::Debug for Reactor { +impl fmt::Debug for Driver { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "Reactor") + write!(f, "Driver") } } @@ -348,22 +282,24 @@ impl Inner { /// Register an I/O resource with the reactor. /// /// The registration token is returned. - pub(super) fn add_source(&self, source: &dyn Evented) -> io::Result { - let token = self.io_dispatch.alloc().ok_or_else(|| { + pub(super) fn add_source(&self, source: &dyn Evented) -> io::Result
{ + let address = self.io_dispatch.alloc().ok_or_else(|| { io::Error::new( io::ErrorKind::Other, "reactor at max registered I/O resources", ) })?; + self.n_sources.fetch_add(1, SeqCst); + self.io.register( source, - mio::Token(token), + mio::Token(address.to_usize()), mio::Ready::all(), mio::PollOpt::edge(), )?; - Ok(token) + Ok(address) } /// Deregisters an I/O resource from the reactor. @@ -371,20 +307,21 @@ impl Inner { self.io.deregister(source) } - pub(super) fn drop_source(&self, token: usize) { - self.io_dispatch.remove(token); + pub(super) fn drop_source(&self, address: Address) { + self.io_dispatch.remove(address); self.n_sources.fetch_sub(1, SeqCst); } /// Registers interest in the I/O resource associated with `token`. - pub(super) fn register(&self, token: usize, dir: Direction, w: Waker) { + pub(super) fn register(&self, token: Address, dir: Direction, w: Waker) { let sched = self .io_dispatch .get(token) - .unwrap_or_else(|| panic!("IO resource for token {} does not exist!", token)); + .unwrap_or_else(|| panic!("IO resource for token {:?} does not exist!", token)); + let readiness = sched .get_readiness(token) - .unwrap_or_else(|| panic!("token {} no longer valid!", token)); + .unwrap_or_else(|| panic!("token {:?} no longer valid!", token)); let (waker, ready) = match dir { Direction::Read => (&sched.reader, !mio::Ready::writable()), @@ -392,24 +329,13 @@ impl Inner { }; waker.register(w); + if readiness & 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. - for io in self.io_dispatch.unique_iter() { - io.writer.wake(); - io.reader.wake(); - } - } -} - impl Direction { pub(super) fn mask(self) -> mio::Ready { match self { @@ -459,20 +385,16 @@ mod tests { #[test] fn tokens_unique_when_dropped() { loom::model(|| { - println!("\n--- iteration ---\n"); - let reactor = Reactor::new().unwrap(); + let reactor = Driver::new().unwrap(); let inner = reactor.inner; let inner2 = inner.clone(); let token_1 = inner.add_source(&NotEvented).unwrap(); - println!("token 1: {:#x}", token_1); let thread = thread::spawn(move || { inner2.drop_source(token_1); - println!("dropped: {:#x}", token_1); }); let token_2 = inner.add_source(&NotEvented).unwrap(); - println!("token 2: {:#x}", token_2); thread.join().unwrap(); assert!(token_1 != token_2); @@ -482,8 +404,7 @@ mod tests { #[test] fn tokens_unique_when_dropped_on_full_page() { loom::model(|| { - println!("\n--- iteration ---\n"); - let reactor = Reactor::new().unwrap(); + let reactor = Driver::new().unwrap(); let inner = reactor.inner; let inner2 = inner.clone(); // add sources to fill up the first page so that the dropped index @@ -493,14 +414,11 @@ mod tests { } let token_1 = inner.add_source(&NotEvented).unwrap(); - println!("token 1: {:#x}", token_1); let thread = thread::spawn(move || { inner2.drop_source(token_1); - println!("dropped: {:#x}", token_1); }); let token_2 = inner.add_source(&NotEvented).unwrap(); - println!("token 2: {:#x}", token_2); thread.join().unwrap(); assert!(token_1 != token_2); @@ -510,19 +428,16 @@ mod tests { #[test] fn tokens_unique_concurrent_add() { loom::model(|| { - println!("\n--- iteration ---\n"); - let reactor = Reactor::new().unwrap(); + let reactor = Driver::new().unwrap(); let inner = reactor.inner; let inner2 = inner.clone(); let thread = thread::spawn(move || { let token_2 = inner2.add_source(&NotEvented).unwrap(); - println!("token 2: {:#x}", token_2); token_2 }); let token_1 = inner.add_source(&NotEvented).unwrap(); - println!("token 1: {:#x}", token_1); let token_2 = thread.join().unwrap(); assert!(token_1 != token_2); diff --git a/tokio/src/net/driver/platform.rs b/tokio/src/io/driver/platform.rs similarity index 100% rename from tokio/src/net/driver/platform.rs rename to tokio/src/io/driver/platform.rs diff --git a/tokio/src/io/driver/scheduled_io.rs b/tokio/src/io/driver/scheduled_io.rs new file mode 100644 index 000000000..1eb6624c6 --- /dev/null +++ b/tokio/src/io/driver/scheduled_io.rs @@ -0,0 +1,142 @@ +use crate::loom::future::AtomicWaker; +use crate::loom::sync::atomic::AtomicUsize; +use crate::util::bit; +use crate::util::slab::{Address, Entry, Generation}; + +use std::sync::atomic::Ordering::{Acquire, AcqRel, SeqCst}; + +#[derive(Debug)] +pub(crate) struct ScheduledIo { + readiness: AtomicUsize, + pub(crate) reader: AtomicWaker, + pub(crate) writer: AtomicWaker, +} + +const PACK: bit::Pack = bit::Pack::most_significant(Generation::WIDTH); + +impl Entry for ScheduledIo { + fn generation(&self) -> Generation { + unpack_generation(self.readiness.load(SeqCst)) + } + + fn reset(&self, generation: Generation) -> bool { + let mut current = self.readiness.load(Acquire); + + loop { + if unpack_generation(current) != generation { + return false; + } + + let next = PACK.pack(generation.next().to_usize(), 0); + + match self.readiness.compare_exchange( + current, + next, + AcqRel, + Acquire, + ) { + Ok(_) => break, + Err(actual) => current = actual, + } + } + + drop(self.reader.take_waker()); + drop(self.writer.take_waker()); + + true + } +} + +impl Default for ScheduledIo { + fn default() -> ScheduledIo { + ScheduledIo { + readiness: AtomicUsize::new(0), + reader: AtomicWaker::new(), + writer: AtomicWaker::new(), + } + } +} + +impl ScheduledIo { + /// Returns the current readiness value of this `ScheduledIo`, if the + /// provided `token` is still a valid access. + /// + /// # Returns + /// + /// If the given token's generation no longer matches the `ScheduledIo`'s + /// generation, then the corresponding IO resource has been removed and + /// replaced with a new resource. In that case, this method returns `None`. + /// Otherwise, this returns the current readiness. + pub(crate) fn get_readiness(&self, address: Address) -> Option { + let ready = self.readiness.load(Acquire); + + if unpack_generation(ready) != address.generation() { + return None; + } + + Some(ready & !PACK.mask()) + } + + /// Sets the readiness on this `ScheduledIo` by invoking the given closure on + /// the current value, returning the previous readiness value. + /// + /// # Arguments + /// - `token`: the token for this `ScheduledIo`. + /// - `f`: a closure returning a new readiness value given the previous + /// readiness. + /// + /// # Returns + /// + /// If the given token's generation no longer matches the `ScheduledIo`'s + /// generation, then the corresponding IO resource has been removed and + /// replaced with a new resource. In that case, this method returns `Err`. + /// Otherwise, this returns the previous readiness. + pub(crate) fn set_readiness( + &self, + address: Address, + f: impl Fn(usize) -> usize, + ) -> Result { + let generation = address.generation(); + + let mut current = self.readiness.load(Acquire); + + loop { + // Check that the generation for this access is still the current + // one. + if unpack_generation(current) != generation { + return Err(()); + } + // Mask out the generation bits so that the modifying function + // doesn't see them. + let current_readiness = current & mio::Ready::all().as_usize(); + let new = f(current_readiness); + + debug_assert!( + new <= !PACK.max_value(), + "new readiness value would overwrite generation bits!" + ); + + match self.readiness.compare_exchange( + current, + PACK.pack(generation.to_usize(), new), + AcqRel, + Acquire, + ) { + Ok(_) => return Ok(current), + // we lost the race, retry! + Err(actual) => current = actual, + } + } + } +} + +impl Drop for ScheduledIo { + fn drop(&mut self) { + self.writer.wake(); + self.reader.wake(); + } +} + +fn unpack_generation(src: usize) -> Generation { + Generation::new(PACK.unpack(src)) +} diff --git a/tokio/src/io/mod.rs b/tokio/src/io/mod.rs index df1888cea..6fa8a17fc 100644 --- a/tokio/src/io/mod.rs +++ b/tokio/src/io/mod.rs @@ -49,6 +49,16 @@ pub use self::async_read::AsyncRead; mod async_write; pub use self::async_write::AsyncWrite; +cfg_io_driver! { + pub(crate) mod driver; + + mod poll_evented; + pub use poll_evented::PollEvented; + + mod registration; + pub use registration::Registration; +} + cfg_io_std! { mod stderr; pub use stderr::{stderr, Stderr}; diff --git a/tokio/src/net/util/poll_evented.rs b/tokio/src/io/poll_evented.rs similarity index 99% rename from tokio/src/net/util/poll_evented.rs rename to tokio/src/io/poll_evented.rs index 08dea3f38..d1644ca20 100644 --- a/tokio/src/net/util/poll_evented.rs +++ b/tokio/src/io/poll_evented.rs @@ -1,5 +1,5 @@ -use crate::io::{AsyncRead, AsyncWrite}; -use crate::net::driver::{platform, Registration}; +use crate::io::{AsyncRead, AsyncWrite, Registration}; +use crate::io::driver::{platform}; use mio::event::Evented; use std::fmt; @@ -52,7 +52,7 @@ use std::task::{Context, Poll}; /// [`clear_read_ready`]. /// /// ```rust -/// use tokio::net::util::PollEvented; +/// use tokio::io::PollEvented; /// /// use futures::ready; /// use mio::Ready; diff --git a/tokio/src/net/driver/registration.rs b/tokio/src/io/registration.rs similarity index 92% rename from tokio/src/net/driver/registration.rs rename to tokio/src/io/registration.rs index 8e946d5a9..1a8db6d79 100644 --- a/tokio/src/net/driver/registration.rs +++ b/tokio/src/io/registration.rs @@ -1,9 +1,9 @@ -use super::platform; -use super::reactor::{Direction, Handle}; +use crate::io::driver::{Direction, Handle, platform}; +use crate::util::slab::Address; use mio::{self, Evented}; use std::task::{Context, Poll}; -use std::{io, usize}; +use std::io; /// Associates an I/O resource with the reactor instance that drives it. /// @@ -38,7 +38,7 @@ use std::{io, usize}; #[derive(Debug)] pub struct Registration { handle: Handle, - token: usize, + address: Address, } // ===== impl Registration ===== @@ -50,12 +50,12 @@ impl Registration { /// /// - `Ok` if the registration happened successfully /// - `Err` if an error was encountered during registration - pub fn new(io: &T) -> io::Result + pub fn new(io: &T) -> io::Result where T: Evented, { let handle = Handle::current(); - let token = if let Some(inner) = handle.inner() { + let address = if let Some(inner) = handle.inner() { inner.add_source(io)? } else { return Err(io::Error::new( @@ -63,7 +63,8 @@ impl Registration { "failed to find event loop", )); }; - Ok(Self { handle, token }) + + Ok(Registration { handle, address }) } /// Deregister the I/O resource from the reactor it is associated with. @@ -212,13 +213,13 @@ impl Registration { // If the task should be notified about new events, ensure that it has // been registered if let Some(ref cx) = cx { - inner.register(self.token, direction, cx.waker().clone()) + inner.register(self.address, direction, cx.waker().clone()) } let mask = direction.mask(); let mask_no_hup = (mask - platform::hup()).as_usize(); - let sched = inner.io_dispatch.get(self.token).unwrap(); + let sched = inner.io_dispatch.get(self.address).unwrap(); // This consumes the current readiness state **except** for HUP. HUP is // excluded because a) it is a final state and never transitions out of @@ -229,8 +230,9 @@ impl Registration { // `poll_ready` is called again with a _`direction` of `Write`, the HUP // state would not be visible. let curr_ready = sched - .set_readiness(self.token, |curr| curr & (!mask_no_hup)) - .unwrap_or_else(|_| panic!("token {} no longer valid!", self.token)); + .set_readiness(self.address, |curr| curr & (!mask_no_hup)) + .unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address)); + let mut ready = mask & mio::Ready::from_usize(curr_ready); if ready.is_empty() { @@ -243,8 +245,8 @@ impl Registration { // Try again let curr_ready = sched - .set_readiness(self.token, |curr| curr & (!mask_no_hup)) - .unwrap_or_else(|_| panic!("token {} no longer valid!", self.token)); + .set_readiness(self.address, |curr| curr & (!mask_no_hup)) + .unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address)); ready = mask & mio::Ready::from_usize(curr_ready); } } @@ -266,6 +268,6 @@ impl Drop for Registration { Some(inner) => inner, None => return, }; - inner.drop_source(self.token); + inner.drop_source(self.address); } } diff --git a/tokio/src/lib.rs b/tokio/src/lib.rs index d82d58d8a..d1c4397ca 100644 --- a/tokio/src/lib.rs +++ b/tokio/src/lib.rs @@ -117,9 +117,7 @@ cfg_time! { pub mod time; } -cfg_rt_threaded! { - mod util; -} +mod util; cfg_macros! { #[cfg(not(test))] // Work around for rust-lang/rust#62127 diff --git a/tokio/src/net/driver/mod.rs b/tokio/src/net/driver/mod.rs deleted file mode 100644 index 2b3afc7d8..000000000 --- a/tokio/src/net/driver/mod.rs +++ /dev/null @@ -1,141 +0,0 @@ -//! Event loop that drives Tokio I/O resources. -//! -//! This module contains [`Reactor`], which is the event loop that drives all -//! Tokio I/O resources. It is the reactor's job to receive events from the -//! operating system ([epoll], [kqueue], [IOCP], etc...) and forward them to -//! waiting tasks. It is the bridge between operating system and the futures -//! model. -//! -//! # Overview -//! -//! When using Tokio, all operations are asynchronous and represented by -//! futures. These futures, representing the application logic, are scheduled by -//! an executor (see [runtime model] for more details). Executors wait for -//! notifications before scheduling the future for execution time, i.e., nothing -//! happens until an event is received indicating that the task can make -//! progress. -//! -//! The reactor receives events from the operating system and notifies the -//! executor. -//! -//! Let's start with a basic example, establishing a TCP connection. -//! -//! ``` -//! use tokio::net::TcpStream; -//! -//! # async fn process(_t: T) {} -//! -//! # #[tokio::main] -//! # async fn dox() -> Result<(), Box> { -//! let stream = TcpStream::connect("93.184.216.34:9243").await?; -//! -//! println!("successfully connected"); -//! -//! process(stream).await; -//! # Ok(()) -//! # } -//! ``` -//! -//! Establishing a TCP connection usually cannot be completed immediately. -//! [`TcpStream::connect`] does not block the current thread. Instead, it -//! returns a [future][connect-future] that resolves once the TCP connection has -//! been established. The connect future itself has no way of knowing when the -//! TCP connection has been established. -//! -//! Before returning the future, [`TcpStream::connect`] registers the socket -//! with a reactor. This registration process, handled by [`Registration`], is -//! what links the [`TcpStream`] with the [`Reactor`] instance. At this point, -//! the reactor starts listening for connection events from the operating system -//! for that socket. -//! -//! Once the connect future is passed to [`tokio::run`], it is spawned onto a -//! thread pool. The thread pool waits until it is notified that the connection -//! has completed. -//! -//! When the TCP connection is established, the reactor receives an event from -//! the operating system. It then notifies the thread pool, telling it that the -//! connect future can complete. At this point, the thread pool will schedule -//! the task to run on one of its worker threads. This results in the `and_then` -//! closure to get executed. -//! -//! ## Eager registration -//! -//! Notice how the snippet does not explicitly reference a reactor. When -//! [`TcpStream::connect`] is called, it registers the socket with the current -//! reactor, but no reactor is specified. This works because a reactor -//! instance is automatically made available when using the Tokio [runtime], -//! which is done using [`tokio::main`]. The Tokio runtime's executor sets a -//! thread-local variable referencing the associated [`Reactor`] instance and -//! [`Handle::current`] (used by [`Registration`]) returns the reference. -//! -//! ## Implementation -//! -//! The reactor implementation uses [`mio`] to interface with the operating -//! system's event queue. A call to [`Reactor::poll`] results in a single -//! call to [`Poll::poll`] which in turn results in a single call to the -//! operating system's selector. -//! -//! The reactor maintains state for each registered I/O resource. This tracks -//! the executor task to notify when events are provided by the operating -//! system's selector. This state is stored in a `Sync` data structure and -//! referenced by [`Registration`]. When the [`Registration`] instance is -//! dropped, this state is cleaned up. Because the state is stored in a `Sync` -//! data structure, the [`Registration`] instance is able to be moved to other -//! threads. -//! -//! By default, a runtime's default reactor runs on a background thread. This -//! ensures that application code cannot significantly impact the reactor's -//! responsiveness. -//! -//! ## Integrating with the reactor -//! -//! Tokio comes with a number of I/O resources, like TCP and UDP sockets, that -//! automatically integrate with the reactor. However, library authors or -//! applications may wish to implement their own resources that are also backed -//! by the reactor. -//! -//! There are a couple of ways to do this. -//! -//! If the custom I/O resource implements [`mio::Evented`] and implements -//! [`std::io::Read`] and / or [`std::io::Write`], then [`PollEvented`] is the -//! most suited. -//! -//! Otherwise, [`Registration`] can be used directly. This provides the lowest -//! level primitive needed for integrating with the reactor: a stream of -//! readiness events. -//! -//! [`Reactor`]: struct.Reactor.html -//! [`Registration`]: struct.Registration.html -//! [runtime model]: https://tokio.rs/docs/internals/runtime-model/ -//! [epoll]: http://man7.org/linux/man-pages/man7/epoll.7.html -//! [kqueue]: https://www.freebsd.org/cgi/man.cgi?query=kqueue&sektion=2 -//! [IOCP]: https://msdn.microsoft.com/en-us/library/windows/desktop/aa365198(v=vs.85).aspx -//! [`TcpStream::connect`]: ../net/struct.TcpStream.html#method.connect -//! [`connect`]: ../net/struct.TcpStream.html#method.connect -//! [connect-future]: ../net/struct.ConnectFuture.html -//! [`tokio::run`]: ../runtime/fn.run.html -//! [`TcpStream`]: ../net/struct.TcpStream.html -//! [runtime]: ../runtime -//! [`Handle::current`]: struct.Handle.html#method.current -//! [`mio`]: https://github.com/carllerche/mio -//! [`Reactor::poll`]: struct.Reactor.html#method.poll -//! [`Poll::poll`]: https://docs.rs/mio/0.6/mio/struct.Poll.html#method.poll -//! [`mio::Evented`]: https://docs.rs/mio/0.6/mio/trait.Evented.html -//! [`PollEvented`]: struct.PollEvented.html -//! [`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 -#[cfg(all(loom, test))] -macro_rules! loom_thread_local { - ($($tts:tt)+) => { loom::thread_local!{ $($tts)+ } } -} - -#[cfg(any(not(loom), not(test)))] -macro_rules! loom_thread_local { - ($($tts:tt)+) => { std::thread_local!{ $($tts)+ } } -} -pub(crate) mod platform; -mod reactor; -mod registration; - -pub use self::reactor::{set_default, DefaultGuard, Handle, Reactor}; -pub use self::registration::Registration; diff --git a/tokio/src/net/driver/reactor/dispatch/iter.rs b/tokio/src/net/driver/reactor/dispatch/iter.rs deleted file mode 100644 index f785f5d48..000000000 --- a/tokio/src/net/driver/reactor/dispatch/iter.rs +++ /dev/null @@ -1,53 +0,0 @@ -use super::{ - page::{self, ScheduledIo}, - Shard, -}; -use std::slice; - -pub(in crate::net::driver::reactor) struct UniqueIter<'a> { - pub(super) shards: slice::IterMut<'a, Shard>, - pub(super) pages: slice::Iter<'a, page::Shared>, - pub(super) slots: Option>, -} - -impl<'a> Iterator for UniqueIter<'a> { - type Item = &'a ScheduledIo; - fn next(&mut self) -> Option { - loop { - if let Some(item) = self.slots.as_mut().and_then(|slots| slots.next()) { - return Some(item); - } - - if let Some(page) = self.pages.next() { - self.slots = page.iter(); - } - - if let Some(shard) = self.shards.next() { - self.pages = shard.iter(); - } else { - return None; - } - } - } -} - -pub(in crate::net::driver::reactor) struct ShardIter<'a> { - pub(super) pages: slice::IterMut<'a, page::Shared>, - pub(super) slots: Option>, -} - -impl<'a> Iterator for ShardIter<'a> { - type Item = &'a ScheduledIo; - fn next(&mut self) -> Option { - loop { - if let Some(item) = self.slots.as_mut().and_then(|slots| slots.next()) { - return Some(item); - } - if let Some(page) = self.pages.next() { - self.slots = page.iter(); - } else { - return None; - } - } - } -} diff --git a/tokio/src/net/driver/reactor/dispatch/mod.rs b/tokio/src/net/driver/reactor/dispatch/mod.rs deleted file mode 100644 index d7262a354..000000000 --- a/tokio/src/net/driver/reactor/dispatch/mod.rs +++ /dev/null @@ -1,36 +0,0 @@ -//! A lock-free concurrent slab. - -#[cfg(all(test, loom))] -macro_rules! test_println { - ($($arg:tt)*) => { - println!("{:?} {}", crate::net::driver::reactor::dispatch::Tid::current(), format_args!($($arg)*)) - } -} - -mod iter; -mod pack; -mod page; -mod sharded_slab; -mod tid; - -#[cfg(all(test, loom))] -// this is used by sub-modules -use self::tests::test_util; -use pack::{Pack, WIDTH}; -use sharded_slab::Shard; -#[cfg(all(test, loom))] -pub(crate) use sharded_slab::Slab; -pub(crate) use sharded_slab::{SingleShard, MAX_SOURCES}; -use tid::Tid; - -#[cfg(target_pointer_width = "64")] -const MAX_THREADS: usize = 4096; -#[cfg(target_pointer_width = "32")] -const MAX_THREADS: usize = 2048; -const INITIAL_PAGE_SIZE: usize = 32; -const MAX_PAGES: usize = WIDTH / 4; -// Chosen arbitrarily. -const RESERVED_BITS: usize = 5; - -#[cfg(test)] -mod tests; diff --git a/tokio/src/net/driver/reactor/dispatch/pack.rs b/tokio/src/net/driver/reactor/dispatch/pack.rs deleted file mode 100644 index 0be44a48a..000000000 --- a/tokio/src/net/driver/reactor/dispatch/pack.rs +++ /dev/null @@ -1,89 +0,0 @@ -pub(super) const WIDTH: usize = std::mem::size_of::() * 8; - -/// Trait encapsulating the calculations required for bit-packing slab indices. -/// -/// This allows us to avoid manually repeating some calculations when packing -/// and unpacking indices. -pub(crate) trait Pack: Sized { - // ====== provided by each implementation ================================= - - /// The number of bits occupied by this type when packed into a usize. - /// - /// This must be provided to determine the number of bits into which to pack - /// the type. - const LEN: usize; - /// The type packed on the less significant side of this type. - /// - /// If this type is packed into the least significant bit of a usize, this - /// should be `()`, which occupies no bytes. - /// - /// This is used to calculate the shift amount for packing this value. - type Prev: Pack; - - // ====== calculated automatically ======================================== - - /// A number consisting of `Self::LEN` 1 bits, starting at the least - /// significant bit. - /// - /// This is the higest value this type can represent. This number is shifted - /// left by `Self::SHIFT` bits to calculate this type's `MASK`. - /// - /// This is computed automatically based on `Self::LEN`. - const BITS: usize = { - let shift = 1 << (Self::LEN - 1); - shift | (shift - 1) - }; - /// The number of bits to shift a number to pack it into a usize with other - /// values. - /// - /// This is caculated automatically based on the `LEN` and `SHIFT` constants - /// of the previous value. - const SHIFT: usize = Self::Prev::SHIFT + Self::Prev::LEN; - - /// The mask to extract only this type from a packed `usize`. - /// - /// This is calculated by shifting `Self::BITS` left by `Self::SHIFT`. - const MASK: usize = Self::BITS << Self::SHIFT; - - fn as_usize(&self) -> usize; - fn from_usize(val: usize) -> Self; - - #[inline(always)] - fn pack(&self, to: usize) -> usize { - let value = self.as_usize(); - debug_assert!(value <= Self::BITS); - - (to & !Self::MASK) | (value << Self::SHIFT) - } - - #[inline(always)] - fn from_packed(from: usize) -> Self { - let value = (from & Self::MASK) >> Self::SHIFT; - debug_assert!(value <= Self::BITS); - Self::from_usize(value) - } -} - -impl Pack for () { - const BITS: usize = 0; - const LEN: usize = 0; - const SHIFT: usize = 0; - const MASK: usize = 0; - - type Prev = (); - - fn as_usize(&self) -> usize { - unreachable!() - } - fn from_usize(_val: usize) -> Self { - unreachable!() - } - - fn pack(&self, _to: usize) -> usize { - unreachable!() - } - - fn from_packed(_from: usize) -> Self { - unreachable!() - } -} diff --git a/tokio/src/net/driver/reactor/dispatch/page/scheduled_io.rs b/tokio/src/net/driver/reactor/dispatch/page/scheduled_io.rs deleted file mode 100644 index aa0441220..000000000 --- a/tokio/src/net/driver/reactor/dispatch/page/scheduled_io.rs +++ /dev/null @@ -1,169 +0,0 @@ -use super::super::{Pack, Tid, RESERVED_BITS, WIDTH}; -use crate::loom::{cell::CausalCell, sync::atomic::AtomicUsize}; -use crate::sync::AtomicWaker; - -use std::sync::atomic::Ordering; - -#[derive(Debug)] -pub(crate) struct ScheduledIo { - /// The offset of the next item on the free list. - next: CausalCell, - readiness: AtomicUsize, - pub(in crate::net::driver) reader: AtomicWaker, - pub(in crate::net::driver) writer: AtomicWaker, -} - -#[repr(transparent)] -#[derive(Copy, Clone, Debug, PartialEq, Eq, Ord, PartialOrd)] -pub(crate) struct Generation { - value: usize, -} - -impl Pack for Generation { - /// Use all the remaining bits in the word for the generation counter, minus - /// any bits reserved by the user. - const LEN: usize = (WIDTH - RESERVED_BITS) - Self::SHIFT; - - type Prev = Tid; - - #[inline(always)] - fn from_usize(u: usize) -> Self { - debug_assert!(u <= Self::BITS); - Self::new(u) - } - - #[inline(always)] - fn as_usize(&self) -> usize { - self.value - } -} - -impl Generation { - const ONE: usize = 1 << Self::SHIFT; - - fn new(value: usize) -> Self { - Self { value } - } - - fn next(self) -> Self { - Self::from_usize((self.value + 1) % Self::BITS) - } -} - -impl ScheduledIo { - pub(super) fn new(next: usize) -> Self { - Self { - next: CausalCell::new(next), - readiness: AtomicUsize::new(0), - reader: AtomicWaker::new(), - writer: AtomicWaker::new(), - } - } - - #[inline] - pub(super) fn alloc(&self) -> Generation { - Generation::from_packed(self.readiness.load(Ordering::SeqCst)) - } - - #[inline(always)] - pub(super) fn next(&self) -> usize { - self.next.with(|next| unsafe { *next }) - } - - #[inline] - pub(super) fn reset(&self, gen: Generation) -> bool { - let mut current = self.readiness.load(Ordering::Acquire); - loop { - if Generation::from_packed(current) != gen { - return false; - } - let next_gen = gen.next().pack(0); - match self.readiness.compare_exchange( - current, - next_gen, - Ordering::AcqRel, - Ordering::Acquire, - ) { - Ok(_) => break, - Err(actual) => current = actual, - } - } - drop(self.reader.take_waker()); - drop(self.writer.take_waker()); - true - } - - #[inline(always)] - pub(super) fn set_next(&self, next: usize) { - self.next.with_mut(|n| unsafe { - (*n) = next; - }) - } - - /// Returns the current readiness value of this `ScheduledIo`, if the - /// provided `token` is still a valid access. - /// - /// # Returns - /// - /// If the given token's generation no longer matches the `ScheduledIo`'s - /// generation, then the corresponding IO resource has been removed and - /// replaced with a new resource. In that case, this method returns `None`. - /// Otherwise, this returns the current readiness. - pub(in crate::net::driver) fn get_readiness(&self, token: usize) -> Option { - let gen = token & Generation::MASK; - let ready = self.readiness.load(Ordering::Acquire); - if ready & Generation::MASK != gen { - return None; - } - Some(ready & (!Generation::MASK)) - } - - /// Sets the readiness on this `ScheduledIo` by invoking the given closure on - /// the current value, returning the previous readiness value. - /// - /// # Arguments - /// - `token`: the token for this `ScheduledIo`. - /// - `f`: a closure returning a new readiness value given the previous - /// readiness. - /// - /// # Returns - /// - /// If the given token's generation no longer matches the `ScheduledIo`'s - /// generation, then the corresponding IO resource has been removed and - /// replaced with a new resource. In that case, this method returns `Err`. - /// Otherwise, this returns the previous readiness. - pub(in crate::net::driver) fn set_readiness( - &self, - token: usize, - f: impl Fn(usize) -> usize, - ) -> Result { - let gen = token & Generation::MASK; - let mut current = self.readiness.load(Ordering::Acquire); - loop { - // Check that the generation for this access is still the current - // one. - if current & Generation::MASK != gen { - return Err(()); - } - // Mask out the generation bits so that the modifying function - // doesn't see them. - let current_readiness = current & mio::Ready::all().as_usize(); - let new = f(current_readiness); - debug_assert!( - new < Generation::ONE, - "new readiness value would overwrite generation bits!" - ); - - match self.readiness.compare_exchange( - current, - new | gen, - Ordering::AcqRel, - Ordering::Acquire, - ) { - Ok(_) => return Ok(current), - // we lost the race, retry! - Err(actual) => current = actual, - } - } - } -} diff --git a/tokio/src/net/driver/reactor/dispatch/page/stack.rs b/tokio/src/net/driver/reactor/dispatch/page/stack.rs deleted file mode 100644 index d98f124f5..000000000 --- a/tokio/src/net/driver/reactor/dispatch/page/stack.rs +++ /dev/null @@ -1,151 +0,0 @@ -use crate::loom::sync::atomic::AtomicUsize; - -use std::fmt; -use std::sync::atomic::Ordering; - -pub(super) struct TransferStack { - head: AtomicUsize, -} - -impl TransferStack { - pub(super) fn new() -> Self { - Self { - head: AtomicUsize::new(super::Addr::NULL), - } - } - - pub(super) fn pop_all(&self) -> Option { - let val = self.head.swap(super::Addr::NULL, Ordering::Acquire); - if val == super::Addr::NULL { - None - } else { - Some(val) - } - } - - pub(super) fn push(&self, value: usize, before: impl Fn(usize)) { - let mut next = self.head.load(Ordering::Relaxed); - loop { - before(next); - - match self - .head - .compare_exchange(next, value, Ordering::AcqRel, Ordering::Acquire) - { - // lost the race! - Err(actual) => next = actual, - Ok(_) => return, - } - } - } -} - -impl fmt::Debug for TransferStack { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - // Loom likes to dump all its internal state in `fmt::Debug` impls, so - // we override this to just print the current value in tests. - f.debug_struct("TransferStack") - .field( - "head", - &format_args!("{:#x}", self.head.load(Ordering::Relaxed)), - ) - .finish() - } -} - -#[cfg(all(test, loom))] -mod test { - use super::super::super::test_util; - use super::*; - use loom::cell::CausalCell; - use loom::thread; - use std::sync::Arc; - - #[test] - fn transfer_stack() { - test_util::run_model("transfer_stack", || { - let causalities = [CausalCell::new(None), CausalCell::new(None)]; - let shared = Arc::new((causalities, TransferStack::new())); - let shared1 = shared.clone(); - let shared2 = shared.clone(); - - // Spawn two threads that both try to push to the stack. - let t1 = thread::spawn(move || { - let (causalities, stack) = &*shared1; - stack.push(0, |prev| { - causalities[0].with_mut(|c| unsafe { - *c = Some(prev); - }); - test_println!("prev={:#x}", prev) - }); - }); - - let t2 = thread::spawn(move || { - let (causalities, stack) = &*shared2; - stack.push(1, |prev| { - causalities[1].with_mut(|c| unsafe { - *c = Some(prev); - }); - test_println!("prev={:#x}", prev) - }); - }); - - let (causalities, stack) = &*shared; - - // Try to pop from the stack... - let mut idx = stack.pop_all(); - while idx == None { - idx = stack.pop_all(); - thread::yield_now(); - } - let idx = idx.unwrap(); - test_println!("popped {:#x}", idx); - - let saw_both = causalities[idx].with(|val| { - let val = unsafe { *val }; - assert!( - val.is_some(), - "CausalCell write must happen-before index is pushed to the stack!", - ); - // were there two entries in the stack? if so, check that - // both saw a write. - if let Some(c) = causalities.get(val.unwrap()) { - test_println!("saw both entries!"); - c.with(|val| { - let val = unsafe { *val }; - assert!( - val.is_some(), - "CausalCell write must happen-before index is pushed to the stack!", - ); - }); - true - } else { - false - } - }); - - // We only saw one push. Ensure that the other push happens too. - if !saw_both { - // Try to pop from the stack... - let mut idx = stack.pop_all(); - while idx == None { - idx = stack.pop_all(); - thread::yield_now(); - } - let idx = idx.unwrap(); - - test_println!("popped {:#x}", idx); - causalities[idx].with(|val| { - let val = unsafe { *val }; - assert!( - val.is_some(), - "CausalCell write must happen-before index is pushed to the stack!", - ); - }); - } - - t1.join().unwrap(); - t2.join().unwrap(); - }); - } -} diff --git a/tokio/src/net/driver/reactor/dispatch/sharded_slab.rs b/tokio/src/net/driver/reactor/dispatch/sharded_slab.rs deleted file mode 100644 index b85f1bddf..000000000 --- a/tokio/src/net/driver/reactor/dispatch/sharded_slab.rs +++ /dev/null @@ -1,274 +0,0 @@ -use super::*; -use std::fmt; - -use crate::loom::sync::Mutex; - -/// A sharded slab. -pub(crate) struct Slab { - shards: Box<[Shard]>, -} - -/// A slab implemented with a single shard. -// TODO(eliza): once worker threads are available, this type will be -// unnecessary and can be removed. -#[derive(Debug)] -pub(crate) struct SingleShard { - shard: Shard, - local: Mutex<()>, -} - -// ┌─────────────┐ ┌────────┐ -// │ page 1 │ │ │ -// ├─────────────┤ ┌───▶│ next──┼─┐ -// │ page 2 │ │ ├────────┤ │ -// │ │ │ │XXXXXXXX│ │ -// │ local_free──┼─┘ ├────────┤ │ -// │ global_free─┼─┐ │ │◀┘ -// ├─────────────┤ └───▶│ next──┼─┐ -// │ page 3 │ ├────────┤ │ -// └─────────────┘ │XXXXXXXX│ │ -// ... ├────────┤ │ -// ┌─────────────┐ │XXXXXXXX│ │ -// │ page n │ ├────────┤ │ -// └─────────────┘ │ │◀┘ -// │ next──┼───▶ -// ├────────┤ -// │XXXXXXXX│ -// └────────┘ -// ... -pub(super) struct Shard { - #[cfg(debug_assertions)] - tid: usize, - /// The local free list for each page. - /// - /// These are only ever accessed from this shard's thread, so they are - /// stored separately from the shared state for the page that can be - /// accessed concurrently, to minimize false sharing. - local: Box<[page::Local]>, - /// The shared state for each page in this shard. - /// - /// This consists of the page's metadata (size, previous size), remote free - /// list, and a pointer to the actual array backing that page. - shared: Box<[page::Shared]>, -} - -pub(crate) const TOKEN_SHIFT: usize = Tid::SHIFT + Tid::LEN; -pub(crate) const MAX_SOURCES: usize = (1 << TOKEN_SHIFT) - 1; - -#[allow(dead_code)] // coming back soon! -impl Slab { - /// Returns a new slab with the default configuration parameters. - pub(crate) fn new() -> Self { - Self::with_max_threads(MAX_THREADS) - } - - pub(crate) fn with_max_threads(max_threads: usize) -> Self { - // Round the max number of threads to the next power of two and clamp to - // the maximum representable number. - let max = max_threads.next_power_of_two().min(MAX_THREADS); - let shards = (0..max).map(Shard::new).collect(); - Self { shards } - } - - /// allocs a value into the slab, returning a key that can be used to - /// access it. - /// - /// If this function returns `None`, then the shard for the current thread - /// is full and no items can be added until some are removed, or the maximum - /// number of shards has been reached. - pub(crate) fn alloc(&self) -> Option { - let tid = Tid::current(); - self.shards[tid.as_usize()].alloc().map(|idx| tid.pack(idx)) - } - - /// Removes the value associated with the given key from the slab. - pub(crate) fn remove(&self, idx: usize) { - let tid = Tid::from_packed(idx); - let shard = &self.shards[tid.as_usize()]; - if tid.is_current() { - shard.remove_local(idx) - } else { - shard.remove_remote(idx) - } - } - - /// Return a reference to the value associated with the given key. - /// - /// If the slab does not contain a value for the given key, `None` is - /// returned instead. - pub(in crate::net::driver) fn get(&self, token: usize) -> Option<&page::ScheduledIo> { - let tid = Tid::from_packed(token); - self.shards.get(tid.as_usize())?.get(token) - } - - /// Returns an iterator over all the items in the slab. - pub(in crate::net::driver::reactor) fn unique_iter(&mut self) -> iter::UniqueIter<'_> { - let mut shards = self.shards.iter_mut(); - let shard = shards.next().expect("must be at least 1 shard"); - let mut pages = shard.iter(); - let slots = pages.next().and_then(page::Shared::iter); - iter::UniqueIter { - shards, - slots, - pages, - } - } -} - -impl SingleShard { - /// Returns a new slab with the default configuration parameters. - pub(crate) fn new() -> Self { - Self { - shard: Shard::new(0), - local: Mutex::new(()), - } - } - - /// allocs a value into the slab, returning a key that can be used to - /// access it. - /// - /// If this function returns `None`, then the shard for the current thread - /// is full and no items can be added until some are removed, or the maximum - /// number of shards has been reached. - pub(crate) fn alloc(&self) -> Option { - // we must lock the slab to alloc an item. - let _local = self.local.lock().unwrap(); - self.shard.alloc() - } - - /// Removes the value associated with the given key from the slab. - pub(crate) fn remove(&self, idx: usize) { - // try to lock the slab so that we can use `remove_local`. - let lock = self.local.try_lock(); - // if we were able to lock the slab, we are "local" and can use the fast - // path; otherwise, we will use `remove_remote`. - if lock.is_ok() { - self.shard.remove_local(idx) - } else { - self.shard.remove_remote(idx) - } - } - - /// Return a reference to the value associated with the given key. - /// - /// If the slab does not contain a value for the given key, `None` is - /// returned instead. - pub(in crate::net::driver) fn get(&self, token: usize) -> Option<&page::ScheduledIo> { - self.shard.get(token) - } - - /// Returns an iterator over all the items in the slab. - pub(in crate::net::driver::reactor) fn unique_iter(&mut self) -> iter::ShardIter<'_> { - let mut pages = self.shard.iter_mut(); - let slots = pages.next().and_then(|pg| pg.iter()); - iter::ShardIter { slots, pages } - } -} - -impl Shard { - fn new(_idx: usize) -> Self { - let mut total_sz = 0; - let shared = (0..MAX_PAGES) - .map(|page_num| { - let sz = page::size(page_num); - let prev_sz = total_sz; - total_sz += sz; - page::Shared::new(sz, prev_sz) - }) - .collect(); - let local = (0..MAX_PAGES).map(|_| page::Local::new()).collect(); - Self { - #[cfg(debug_assertions)] - tid: _idx, - local, - shared, - } - } - - fn alloc(&self) -> Option { - // Can we fit the value into an existing page? - for (page_idx, page) in self.shared.iter().enumerate() { - let local = self.local(page_idx); - if let Some(page_offset) = page.alloc(local) { - return Some(page_offset); - } - } - - None - } - - #[inline(always)] - fn get(&self, idx: usize) -> Option<&page::ScheduledIo> { - #[cfg(debug_assertions)] - debug_assert_eq!(Tid::from_packed(idx).as_usize(), self.tid); - - let addr = page::Addr::from_packed(idx); - let i = addr.index(); - - if i > self.shared.len() { - return None; - } - self.shared[i].get(addr) - } - - /// Remove an item on the shard's local thread. - fn remove_local(&self, idx: usize) { - #[cfg(debug_assertions)] - debug_assert_eq!(Tid::from_packed(idx).as_usize(), self.tid); - let addr = page::Addr::from_packed(idx); - let page_idx = addr.index(); - - if let Some(page) = self.shared.get(page_idx) { - page.remove_local(self.local(page_idx), addr, idx); - } - } - - /// Remove an item, while on a different thread from the shard's local thread. - fn remove_remote(&self, idx: usize) { - #[cfg(debug_assertions)] - debug_assert_eq!(Tid::from_packed(idx).as_usize(), self.tid); - let addr = page::Addr::from_packed(idx); - let page_idx = addr.index(); - - if let Some(page) = self.shared.get(page_idx) { - page.remove_remote(addr, idx); - } - } - - #[inline(always)] - fn local(&self, i: usize) -> &page::Local { - &self.local[i] - } - - pub(super) fn iter(&self) -> std::slice::Iter<'_, page::Shared> { - self.shared.iter() - } - - fn iter_mut(&mut self) -> std::slice::IterMut<'_, page::Shared> { - self.shared.iter_mut() - } -} - -impl fmt::Debug for Slab { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("Slab") - .field("shards", &self.shards) - .finish() - } -} - -unsafe impl Send for Slab {} -unsafe impl Sync for Slab {} - -unsafe impl Send for SingleShard {} -unsafe impl Sync for SingleShard {} - -impl fmt::Debug for Shard { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let mut d = f.debug_struct("Shard"); - - #[cfg(debug_assertions)] - d.field("tid", &self.tid); - d.field("shared", &self.shared).finish() - } -} diff --git a/tokio/src/net/driver/reactor/dispatch/tests/loom/mod.rs b/tokio/src/net/driver/reactor/dispatch/tests/loom/mod.rs deleted file mode 100644 index 495cac1c5..000000000 --- a/tokio/src/net/driver/reactor/dispatch/tests/loom/mod.rs +++ /dev/null @@ -1,204 +0,0 @@ -use self::test_util::*; -use super::super::Slab; -use loom::sync::{Arc, Condvar, Mutex}; -use loom::thread; - -pub(crate) mod test_util { - use std::sync::atomic::{AtomicUsize, Ordering}; - - pub(crate) fn run_model(name: &'static str, f: impl Fn() + Sync + Send + 'static) { - run_builder(name, loom::model::Builder::new(), f) - } - - pub(crate) fn run_builder( - name: &'static str, - builder: loom::model::Builder, - f: impl Fn() + Sync + Send + 'static, - ) { - let iters = AtomicUsize::new(1); - builder.check(move || { - println!( - "\n------------ running test {}; iteration {} ------------\n", - name, - iters.fetch_add(1, Ordering::SeqCst) - ); - f() - }); - } -} - -fn store_val(slab: &Arc, readiness: usize) -> usize { - println!("store: {}", readiness); - let key = slab.alloc().expect("allocate slot"); - if let Some(slot) = slab.get(key) { - slot.set_readiness(key, |_| readiness) - .expect("generation should still be valid!"); - } else { - panic!("slab did not contain a value for {:#x}", key); - } - key -} - -fn get_val(slab: &Arc, key: usize) -> Option { - slab.get(key).and_then(|s| s.get_readiness(key)) -} - -mod single_shard; -mod small_slab; - -#[test] -fn local_remove() { - run_model("local_remove", || { - let slab = Arc::new(Slab::new()); - - let s = slab.clone(); - let t1 = thread::spawn(move || { - let idx = store_val(&s, 1); - assert_eq!(get_val(&s, idx), Some(1)); - s.remove(idx); - assert_eq!(get_val(&s, idx), None); - let idx = store_val(&s, 2); - assert_eq!(get_val(&s, idx), Some(2)); - s.remove(idx); - assert_eq!(get_val(&s, idx), None); - }); - - let s = slab.clone(); - let t2 = thread::spawn(move || { - let idx = store_val(&s, 3); - assert_eq!(get_val(&s, idx), Some(3)); - s.remove(idx); - assert_eq!(get_val(&s, idx), None); - let idx = store_val(&s, 4); - s.remove(idx); - assert_eq!(get_val(&s, idx), None); - }); - - let s = slab; - let idx1 = store_val(&s, 5); - assert_eq!(get_val(&s, idx1), Some(5)); - let idx2 = store_val(&s, 6); - assert_eq!(get_val(&s, idx2), Some(6)); - s.remove(idx1); - assert_eq!(get_val(&s, idx1), None); - assert_eq!(get_val(&s, idx2), Some(6)); - s.remove(idx2); - assert_eq!(get_val(&s, idx2), None); - - t1.join().expect("thread 1 should not panic"); - t2.join().expect("thread 2 should not panic"); - }); -} - -#[test] -fn remove_remote() { - run_model("remove_remote", || { - let slab = Arc::new(Slab::new()); - - let idx1 = store_val(&slab, 1); - assert_eq!(get_val(&slab, idx1), Some(1)); - - let idx2 = store_val(&slab, 2); - assert_eq!(get_val(&slab, idx2), Some(2)); - - let idx3 = store_val(&slab, 3); - assert_eq!(get_val(&slab, idx3), Some(3)); - - let s = slab.clone(); - let t1 = thread::spawn(move || { - assert_eq!(get_val(&s, idx2), Some(2)); - s.remove(idx2); - assert_eq!(get_val(&s, idx2), None); - }); - - let s = slab.clone(); - let t2 = thread::spawn(move || { - assert_eq!(get_val(&s, idx3), Some(3)); - s.remove(idx3); - assert_eq!(get_val(&s, idx3), None); - }); - - t1.join().expect("thread 1 should not panic"); - t2.join().expect("thread 2 should not panic"); - - assert_eq!(get_val(&slab, idx1), Some(1)); - assert_eq!(get_val(&slab, idx2), None); - assert_eq!(get_val(&slab, idx3), None); - }); -} - -#[test] -fn concurrent_alloc_remove() { - run_model("concurrent_alloc_remove", || { - let slab = Arc::new(Slab::new()); - let pair = Arc::new((Mutex::new(None), Condvar::new())); - - let slab2 = slab.clone(); - let pair2 = pair.clone(); - let remover = thread::spawn(move || { - let (lock, cvar) = &*pair2; - for i in 0..2 { - test_println!("--- remover i={} ---", i); - let mut next = lock.lock().unwrap(); - while next.is_none() { - next = cvar.wait(next).unwrap(); - } - let key = next.take().unwrap(); - slab2.remove(key); - assert_eq!(get_val(&slab2, key), None); - cvar.notify_one(); - } - }); - - let (lock, cvar) = &*pair; - for i in 0..2 { - test_println!("--- allocator i={} ---", i); - let key = store_val(&slab, i); - - let mut next = lock.lock().unwrap(); - *next = Some(key); - cvar.notify_one(); - - // Wait for the item to be removed. - while next.is_some() { - next = cvar.wait(next).unwrap(); - } - - assert_eq!(get_val(&slab, key), None); - } - - remover.join().unwrap(); - }) -} - -// #[test] -// fn unique_iter() { -// run_model("unique_iter", || { -// let mut slab = Arc::new(Slab::new()); - -// let s = slab.clone(); -// let t1 = thread::spawn(move || { -// store_val(&s, 1); -// store_val(&s, 2); -// }); - -// let s = slab.clone(); -// let t2 = thread::spawn(move || { -// store_val(&s, 3); -// store_val(&s, 4); -// }); - -// t1.join().expect("thread 1 should not panic"); -// t2.join().expect("thread 2 should not panic"); - -// let slab = Arc::get_mut(&mut slab).expect("other arcs should be dropped"); -// let items: Vec<_> = slab -// .unique_iter() -// .map(|i| i.readiness.load(Ordering::Acquire)) -// .collect(); -// assert!(items.contains(&1), "items: {:?}", items); -// assert!(items.contains(&2), "items: {:?}", items); -// assert!(items.contains(&3), "items: {:?}", items); -// assert!(items.contains(&4), "items: {:?}", items); -// }); -// } diff --git a/tokio/src/net/driver/reactor/dispatch/tests/loom/single_shard.rs b/tokio/src/net/driver/reactor/dispatch/tests/loom/single_shard.rs deleted file mode 100644 index cbc974caf..000000000 --- a/tokio/src/net/driver/reactor/dispatch/tests/loom/single_shard.rs +++ /dev/null @@ -1,181 +0,0 @@ -use super::super::super::SingleShard; -use super::test_util; -use loom::sync::{Arc, Condvar, Mutex}; -use loom::thread; - -fn store_val(slab: &Arc, readiness: usize) -> usize { - println!("store: {}", readiness); - let key = slab.alloc().expect("allocate slot"); - if let Some(slot) = slab.get(key) { - slot.set_readiness(key, |_| readiness) - .expect("generation should still be valid!"); - } else { - panic!("slab did not contain a value for {:#x}", key); - } - key -} - -fn get_val(slab: &Arc, key: usize) -> Option { - slab.get(key).and_then(|s| { - let rdy = s.get_readiness(key); - test_println!("--> got readiness {:?} with key {:#x}", rdy, key); - rdy - }) -} - -#[test] -fn local_remove() { - test_util::run_model("single_shard::local_remove", || { - let slab = Arc::new(SingleShard::new()); - - let s = slab.clone(); - let t1 = thread::spawn(move || { - let idx = store_val(&s, 1); - assert_eq!(get_val(&s, idx), Some(1)); - s.remove(idx); - assert_eq!(get_val(&s, idx), None); - let idx = store_val(&s, 2); - assert_eq!(get_val(&s, idx), Some(2)); - s.remove(idx); - assert_eq!(get_val(&s, idx), None); - }); - - let s = slab.clone(); - let t2 = thread::spawn(move || { - let idx = store_val(&s, 3); - assert_eq!(get_val(&s, idx), Some(3)); - s.remove(idx); - assert_eq!(get_val(&s, idx), None); - let idx = store_val(&s, 4); - s.remove(idx); - assert_eq!(get_val(&s, idx), None); - }); - - let s = slab; - let idx1 = store_val(&s, 5); - assert_eq!(get_val(&s, idx1), Some(5)); - let idx2 = store_val(&s, 6); - assert_eq!(get_val(&s, idx2), Some(6)); - s.remove(idx1); - assert_eq!(get_val(&s, idx1), None); - assert_eq!(get_val(&s, idx2), Some(6)); - s.remove(idx2); - assert_eq!(get_val(&s, idx2), None); - - t1.join().expect("thread 1 should not panic"); - t2.join().expect("thread 2 should not panic"); - }); -} - -#[test] -fn remove_remote() { - test_util::run_model("single_shard::remove_remote", || { - let slab = Arc::new(SingleShard::new()); - - let idx1 = store_val(&slab, 1); - assert_eq!(get_val(&slab, idx1), Some(1)); - - let idx2 = store_val(&slab, 2); - assert_eq!(get_val(&slab, idx2), Some(2)); - - let idx3 = store_val(&slab, 3); - assert_eq!(get_val(&slab, idx3), Some(3)); - - let s = slab.clone(); - let t1 = thread::spawn(move || { - assert_eq!(get_val(&s, idx2), Some(2)); - s.remove(idx2); - assert_eq!(get_val(&s, idx2), None); - }); - - let s = slab.clone(); - let t2 = thread::spawn(move || { - assert_eq!(get_val(&s, idx3), Some(3)); - s.remove(idx3); - assert_eq!(get_val(&s, idx3), None); - }); - - t1.join().expect("thread 1 should not panic"); - t2.join().expect("thread 2 should not panic"); - - assert_eq!(get_val(&slab, idx1), Some(1)); - assert_eq!(get_val(&slab, idx2), None); - assert_eq!(get_val(&slab, idx3), None); - }); -} - -#[test] -fn concurrent_alloc_remove() { - test_util::run_model("single_shard::concurrent_alloc_remove", || { - let slab = Arc::new(SingleShard::new()); - let pair = Arc::new((Mutex::new(None), Condvar::new())); - - let slab2 = slab.clone(); - let pair2 = pair.clone(); - let remover = thread::spawn(move || { - let (lock, cvar) = &*pair2; - for i in 0..2 { - test_println!("--- remover i={} ---", i); - let mut next = lock.lock().unwrap(); - while next.is_none() { - next = cvar.wait(next).unwrap(); - } - let key = next.take().unwrap(); - slab2.remove(key); - assert_eq!(get_val(&slab2, key), None); - cvar.notify_one(); - } - }); - - let (lock, cvar) = &*pair; - for i in 0..2 { - test_println!("--- allocator i={} ---", i); - let key = store_val(&slab, i); - - let mut next = lock.lock().unwrap(); - *next = Some(key); - cvar.notify_one(); - - // Wait for the item to be removed. - while next.is_some() { - next = cvar.wait(next).unwrap(); - } - - assert_eq!(get_val(&slab, key), None); - } - - remover.join().unwrap(); - }) -} - -// #[test] -// fn unique_iter() { -// test_util::run_model("single_shard::unique_iter", || { -// let mut slab = Arc::new(SingleShard::new()); - -// let s = slab.clone(); -// let t1 = thread::spawn(move || { -// store_val(&s, 1); -// store_val(&s, 2); -// }); - -// let s = slab.clone(); -// let t2 = thread::spawn(move || { -// store_val(&s, 3); -// store_val(&s, 4); -// }); - -// t1.join().expect("thread 1 should not panic"); -// t2.join().expect("thread 2 should not panic"); - -// let slab = Arc::get_mut(&mut slab).expect("other arcs should be dropped"); -// let items: Vec<_> = slab -// .unique_iter() -// .map(|i| i.readiness.load(Ordering::Acquire)) -// .collect(); -// assert!(items.contains(&1), "items: {:?}", items); -// assert!(items.contains(&2), "items: {:?}", items); -// assert!(items.contains(&3), "items: {:?}", items); -// assert!(items.contains(&4), "items: {:?}", items); -// }); -// } diff --git a/tokio/src/net/driver/reactor/dispatch/tests/loom/small_slab.rs b/tokio/src/net/driver/reactor/dispatch/tests/loom/small_slab.rs deleted file mode 100644 index 108f3b54b..000000000 --- a/tokio/src/net/driver/reactor/dispatch/tests/loom/small_slab.rs +++ /dev/null @@ -1,473 +0,0 @@ -use super::test_util; -use loom::sync::{Arc, Condvar, Mutex}; -use loom::thread; - -use pack::{Pack, WIDTH}; -use sharded_slab::Shard; -use sharded_slab::Slab; -use tid::Tid; - -// Overridden for tests -const INITIAL_PAGE_SIZE: usize = 2; -const MAX_PAGES: usize = 1; - -// Constants not overridden -#[cfg(target_pointer_width = "64")] -const MAX_THREADS: usize = 4096; -#[cfg(target_pointer_width = "32")] -const MAX_THREADS: usize = 2048; -const RESERVED_BITS: usize = 5; - -#[path = "../../page/mod.rs"] -#[allow(dead_code)] -mod page; - -#[path = "../../pack.rs"] -#[allow(dead_code)] -mod pack; - -#[path = "../../iter.rs"] -#[allow(dead_code)] -mod iter; - -#[path = "../../sharded_slab.rs"] -#[allow(dead_code)] -mod sharded_slab; - -#[path = "../../tid.rs"] -#[allow(dead_code)] -mod tid; - -fn store_val(slab: &Arc, readiness: usize) -> usize { - println!("store: {}", readiness); - let key = slab.alloc().expect("allocate slot"); - if let Some(slot) = slab.get(key) { - slot.set_readiness(key, |_| readiness) - .expect("generation should still be valid!"); - } else { - panic!("slab did not contain a value for {:#x}", key); - } - key -} - -fn get_val(slab: &Arc, key: usize) -> Option { - slab.get(key).and_then(|s| { - let rdy = s.get_readiness(key); - test_println!("--> got readiness {:?} with key {:#x}", rdy, key); - rdy - }) -} - -fn store_when_free(slab: &Arc, readiness: usize) -> usize { - test_println!("store: {}", readiness); - let key = loop { - if let Some(key) = slab.alloc() { - break key; - } - test_println!("-> full; retry"); - thread::yield_now(); - }; - if let Some(slot) = slab.get(key) { - slot.set_readiness(key, |_| readiness) - .expect("generation should still be valid!"); - } else { - panic!("slab did not contain a value for {:#x}", key); - } - key -} - -#[test] -fn remove_remote_and_reuse() { - let mut model = loom::model::Builder::new(); - model.max_branches = 100000; - test_util::run_builder("remove_remote_and_reuse", model, || { - let slab = Arc::new(Slab::new()); - - let idx1 = store_val(&slab, 1); - let idx2 = store_val(&slab, 2); - - assert_eq!(get_val(&slab, idx1), Some(1)); - assert_eq!(get_val(&slab, idx2), Some(2)); - - let s = slab.clone(); - let t1 = thread::spawn(move || { - s.remove(idx1); - let value = get_val(&s, idx1); - - // We may or may not see the new value yet, depending on when - // this occurs, but we must either see the new value or `None`; - // the old value has been removed! - assert!(value == None || value == Some(3)); - }); - - let idx3 = store_when_free(&slab, 3); - t1.join().expect("thread 1 should not panic"); - - assert_eq!(get_val(&slab, idx3), Some(3)); - assert_eq!(get_val(&slab, idx2), Some(2)); - }); -} - -#[test] -fn concurrent_remove_remote_and_reuse() { - let mut model = loom::model::Builder::new(); - model.max_branches = 100000; - // set a preemption bound, or else this will run for a *really* long time. - model.preemption_bound = Some(2); // chosen arbitrarily. - test_util::run_builder("concurrent_remove_remote_and_reuse", model, || { - let slab = Arc::new(Slab::new()); - - let idx1 = store_val(&slab, 1); - let idx2 = store_val(&slab, 2); - - assert_eq!(get_val(&slab, idx1), Some(1)); - assert_eq!(get_val(&slab, idx2), Some(2)); - - let s = slab.clone(); - let s2 = slab.clone(); - let t1 = thread::spawn(move || { - s.remove(idx1); - }); - - let t2 = thread::spawn(move || { - s2.remove(idx2); - }); - - let idx3 = store_when_free(&slab, 3); - t1.join().expect("thread 1 should not panic"); - t2.join().expect("thread 1 should not panic"); - - assert!(get_val(&slab, idx1).is_none()); - assert!(get_val(&slab, idx2).is_none()); - assert_eq!(get_val(&slab, idx3), Some(3)); - }); -} - -mod single_shard { - use super::sharded_slab::SingleShard; - use super::*; - - fn store_val(slab: &Arc, readiness: usize) -> usize { - println!("store: {}", readiness); - let key = slab.alloc().expect("allocate slot"); - if let Some(slot) = slab.get(key) { - slot.set_readiness(key, |_| readiness) - .expect("generation should still be valid!"); - } else { - panic!("slab did not contain a value for {:#x}", key); - } - key - } - - fn get_val(slab: &Arc, key: usize) -> Option { - slab.get(key).and_then(|s| { - let rdy = s.get_readiness(key); - test_println!("--> got readiness {:?} with key {:#x}", rdy, key); - rdy - }) - } - - fn store_when_free(slab: &Arc, readiness: usize) -> usize { - test_println!("store: {}", readiness); - let key = loop { - if let Some(key) = slab.alloc() { - break key; - } - test_println!("-> full; retry"); - thread::yield_now(); - }; - if let Some(slot) = slab.get(key) { - slot.set_readiness(key, |_| readiness) - .expect("generation should still be valid!"); - } else { - panic!("slab did not contain a value for {:#x}", key); - } - key - } - - #[test] - fn remove_remote_and_reuse() { - let mut model = loom::model::Builder::new(); - model.max_branches = 100000; - test_util::run_builder("single_shard::remove_remote_and_reuse", model, || { - let slab = Arc::new(SingleShard::new()); - - let idx1 = store_val(&slab, 1); - let idx2 = store_val(&slab, 2); - - assert_eq!(get_val(&slab, idx1), Some(1)); - assert_eq!(get_val(&slab, idx2), Some(2)); - - let s = slab.clone(); - let t1 = thread::spawn(move || { - s.remove(idx1); - let value = get_val(&s, idx1); - - // We may or may not see the new value yet, depending on when - // this occurs, but we must either see the new value or `None`; - // the old value has been removed! - assert!(value == None || value == Some(3)); - }); - - let idx3 = store_when_free(&slab, 3); - t1.join().expect("thread 1 should not panic"); - - assert_eq!(get_val(&slab, idx3), Some(3)); - assert_eq!(get_val(&slab, idx2), Some(2)); - }); - } - - #[test] - fn concurrent_remove_remote_and_reuse() { - let mut model = loom::model::Builder::new(); - model.max_branches = 100000; - // set a preemption bound, or else this will run for a *really* long time. - model.preemption_bound = Some(2); // chosen arbitrarily. - test_util::run_builder("single_shard::remove_remote_and_reuse", model, || { - let slab = Arc::new(SingleShard::new()); - - let idx1 = store_val(&slab, 1); - let idx2 = store_val(&slab, 2); - - assert_eq!(get_val(&slab, idx1), Some(1)); - assert_eq!(get_val(&slab, idx2), Some(2)); - - let s = slab.clone(); - let s2 = slab.clone(); - let t1 = thread::spawn(move || { - s.remove(idx1); - }); - - let t2 = thread::spawn(move || { - s2.remove(idx2); - }); - - let idx3 = store_when_free(&slab, 3); - t1.join().expect("thread 1 should not panic"); - t2.join().expect("thread 1 should not panic"); - - assert!(get_val(&slab, idx1).is_none()); - assert!(get_val(&slab, idx2).is_none()); - assert_eq!(get_val(&slab, idx3), Some(3)); - }); - } - - #[test] - fn alloc_remove_get() { - test_util::run_model("single_shard::alloc_remove_get", || { - let slab = Arc::new(SingleShard::new()); - let pair = Arc::new((Mutex::new(None), Condvar::new())); - - let slab2 = slab.clone(); - let pair2 = pair.clone(); - let t1 = thread::spawn(move || { - let slab = slab2; - let (lock, cvar) = &*pair2; - // allocate one entry just so that we have to use the final one for - // all future allocations. - let _key0 = store_val(&slab, 0); - let key = store_val(&slab, 1); - - let mut next = lock.lock().unwrap(); - *next = Some(key); - cvar.notify_one(); - // remove the second entry - slab.remove(key); - // store a new readiness at the same location (since the slab - // already has an entry in slot 0) - store_val(&slab, 2); - }); - - let (lock, cvar) = &*pair; - // wait for the second entry to be stored... - let mut next = lock.lock().unwrap(); - while next.is_none() { - next = cvar.wait(next).unwrap(); - } - let key = next.unwrap(); - - // our generation will be stale when the second store occurs at that - // index, we must not see the value of that store. - let val = get_val(&slab, key); - assert_ne!(val, Some(2), "generation must have advanced!"); - - t1.join().unwrap(); - }) - } - - #[test] - fn alloc_remove_set() { - test_util::run_model("single_shard::alloc_remove_set", || { - let slab = Arc::new(SingleShard::new()); - let pair = Arc::new((Mutex::new(None), Condvar::new())); - - let slab2 = slab.clone(); - let pair2 = pair.clone(); - let t1 = thread::spawn(move || { - let slab = slab2; - let (lock, cvar) = &*pair2; - // allocate one entry just so that we have to use the final one for - // all future allocations. - let _key0 = store_val(&slab, 0); - let key = store_val(&slab, 1); - - let mut next = lock.lock().unwrap(); - *next = Some(key); - cvar.notify_one(); - - slab.remove(key); - // remove the old entry and insert a new one, with a new generation. - let key2 = slab.alloc().expect("store key 2"); - // after the remove, we must not see the value written with the - // stale index. - assert_eq!( - get_val(&slab, key), - None, - "stale set must no longer be visible" - ); - assert_eq!(get_val(&slab, key2), Some(0)); - key2 - }); - - let (lock, cvar) = &*pair; - - // wait for the second entry to be stored. the index we get from the - // other thread may become stale after a write. - let mut next = lock.lock().unwrap(); - while next.is_none() { - next = cvar.wait(next).unwrap(); - } - let key = next.unwrap(); - - // try to write to the index with our generation - slab.get(key).map(|val| val.set_readiness(key, |_| 2)); - - let key2 = t1.join().unwrap(); - // after the remove, we must not see the value written with the - // stale index either. - assert_eq!( - get_val(&slab, key), - None, - "stale set must no longer be visible" - ); - assert_eq!(get_val(&slab, key2), Some(0)); - }) - } -} - -#[test] -fn alloc_remove_get() { - test_util::run_model("alloc_remove_get", || { - let slab = Arc::new(Slab::new()); - let pair = Arc::new((Mutex::new(None), Condvar::new())); - - let slab2 = slab.clone(); - let pair2 = pair.clone(); - let t1 = thread::spawn(move || { - let slab = slab2; - let (lock, cvar) = &*pair2; - // allocate one entry just so that we have to use the final one for - // all future allocations. - let _key0 = store_val(&slab, 0); - let key = store_val(&slab, 1); - - let mut next = lock.lock().unwrap(); - *next = Some(key); - cvar.notify_one(); - // remove the second entry - slab.remove(key); - // store a new readiness at the same location (since the slab - // already has an entry in slot 0) - store_val(&slab, 2); - }); - - let (lock, cvar) = &*pair; - // wait for the second entry to be stored... - let mut next = lock.lock().unwrap(); - while next.is_none() { - next = cvar.wait(next).unwrap(); - } - let key = next.unwrap(); - - // our generation will be stale when the second store occurs at that - // index, we must not see the value of that store. - let val = get_val(&slab, key); - assert_ne!(val, Some(2), "generation must have advanced!"); - - t1.join().unwrap(); - }) -} - -#[test] -fn alloc_remove_set() { - test_util::run_model("alloc_remove_set", || { - let slab = Arc::new(Slab::new()); - let pair = Arc::new((Mutex::new(None), Condvar::new())); - - let slab2 = slab.clone(); - let pair2 = pair.clone(); - let t1 = thread::spawn(move || { - let slab = slab2; - let (lock, cvar) = &*pair2; - // allocate one entry just so that we have to use the final one for - // all future allocations. - let _key0 = store_val(&slab, 0); - let key = store_val(&slab, 1); - - let mut next = lock.lock().unwrap(); - *next = Some(key); - cvar.notify_one(); - - slab.remove(key); - // remove the old entry and insert a new one, with a new generation. - let key2 = slab.alloc().expect("store key 2"); - // after the remove, we must not see the value written with the - // stale index. - assert_eq!( - get_val(&slab, key), - None, - "stale set must no longer be visible" - ); - assert_eq!(get_val(&slab, key2), Some(0)); - key2 - }); - - let (lock, cvar) = &*pair; - - // wait for the second entry to be stored. the index we get from the - // other thread may become stale after a write. - let mut next = lock.lock().unwrap(); - while next.is_none() { - next = cvar.wait(next).unwrap(); - } - let key = next.unwrap(); - - // try to write to the index with our generation - slab.get(key).map(|val| val.set_readiness(key, |_| 2)); - - let key2 = t1.join().unwrap(); - // after the remove, we must not see the value written with the - // stale index either. - assert_eq!( - get_val(&slab, key), - None, - "stale set must no longer be visible" - ); - assert_eq!(get_val(&slab, key2), Some(0)); - }) -} - -// #[test] -// fn custom_page_sz() { -// let mut model = loom::model::Builder::new(); -// model.max_branches = 100000; -// model.check(|| { -// let slab = Arc::new(Slab::new()); - -// for i in 0..1024 { -// test_println!("{}", i); -// let k = store_val(&slab, i); -// assert_eq!(get_val(&slab, k), Some(i)); -// } -// }); -// } diff --git a/tokio/src/net/driver/reactor/dispatch/tests/mod.rs b/tokio/src/net/driver/reactor/dispatch/tests/mod.rs deleted file mode 100644 index 65ead8ef2..000000000 --- a/tokio/src/net/driver/reactor/dispatch/tests/mod.rs +++ /dev/null @@ -1,30 +0,0 @@ -mod idx { - use super::super::{page, Pack, Tid}; - use proptest::prelude::*; - - proptest! { - #[test] - fn tid_roundtrips(tid in 0usize..Tid::BITS) { - let tid = Tid::from_usize(tid); - let packed = tid.pack(0); - assert_eq!(tid, Tid::from_packed(packed)); - } - - #[test] - fn idx_roundtrips( - tid in 0usize..Tid::BITS, - addr in 0usize..page::Addr::BITS, - ) { - let tid = Tid::from_usize(tid); - let addr = page::Addr::from_usize(addr); - let packed = tid.pack(addr.pack(0)); - assert_eq!(addr, page::Addr::from_packed(packed)); - assert_eq!(tid, Tid::from_packed(packed)); - } - } -} - -#[cfg(loom)] -mod loom; -#[cfg(loom)] -pub(super) use self::loom::test_util; diff --git a/tokio/src/net/driver/reactor/dispatch/tid.rs b/tokio/src/net/driver/reactor/dispatch/tid.rs deleted file mode 100644 index 6d0e1c7a8..000000000 --- a/tokio/src/net/driver/reactor/dispatch/tid.rs +++ /dev/null @@ -1,168 +0,0 @@ -use super::{page, Pack}; -use std::{ - cell::{Cell, UnsafeCell}, - collections::VecDeque, - fmt, - marker::PhantomData, - sync::{ - atomic::{AtomicUsize, Ordering}, - Mutex, - }, -}; - -use lazy_static::lazy_static; - -/// Uniquely identifies a thread. -#[derive(PartialEq, Eq, Copy, Clone)] -pub(crate) struct Tid { - id: usize, - _not_send: PhantomData>, -} - -/// Registers that a thread is currently using a thread ID. -/// -/// This is stored in a thread local on each thread that has been assigned an -/// ID. When the thread terminates, the thread local is dropped, indicating that -/// that thread's ID number may be reused. This is to avoid exhausting the -/// available bits for thread IDs in scenarios where threads are spawned and -/// terminated very frequently. -#[derive(Debug)] -struct Registration(Cell>); - -/// Tracks any thread IDs that can be reused, and a monotonic counter for -/// generating new thread IDs. -struct Registry { - /// The next thread ID number; used when there are no free IDs. - next: AtomicUsize, - /// A queue of thread IDs whose threads have terminated. These will be - /// reused if possible. - free: Mutex>, -} - -lazy_static! { - static ref REGISTRY: Registry = Registry { - next: AtomicUsize::new(0), - free: Mutex::new(VecDeque::new()), - }; -} -loom_thread_local! { - static REGISTRATION: Registration = Registration::new(); -} - -// === impl Tid === - -impl Pack for Tid { - const LEN: usize = super::MAX_THREADS.trailing_zeros() as usize + 1; - - type Prev = page::Addr; - - #[inline(always)] - fn as_usize(&self) -> usize { - self.id - } - - #[inline(always)] - fn from_usize(id: usize) -> Self { - debug_assert!(id <= Self::BITS); - Self { - id, - _not_send: PhantomData, - } - } -} - -impl Tid { - #[inline] - pub(crate) fn current() -> Self { - REGISTRATION - .try_with(Registration::current) - .unwrap_or_else(|_| Self::poisoned()) - } - - pub(crate) fn is_current(self) -> bool { - REGISTRATION - .try_with(|r| self == r.current()) - .unwrap_or(false) - } - - #[inline(always)] - pub(crate) fn new(id: usize) -> Self { - Self { - id, - _not_send: PhantomData, - } - } - - #[cold] - fn poisoned() -> Self { - Self { - id: std::usize::MAX, - _not_send: PhantomData, - } - } - - /// Returns true if the local thread ID was accessed while unwinding. - pub(crate) fn is_poisoned(self) -> bool { - self.id == std::usize::MAX - } -} - -impl fmt::Debug for Tid { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - if self.is_poisoned() { - f.debug_tuple("Tid") - .field(&format_args!("")) - .finish() - } else { - f.debug_tuple("Tid") - .field(&format_args!("{:#x}", self.id)) - .finish() - } - } -} - -// === impl Registration === - -impl Registration { - fn new() -> Self { - Self(Cell::new(None)) - } - - #[inline(always)] - fn current(&self) -> Tid { - if let Some(tid) = self.0.get().map(Tid::new) { - tid - } else { - self.register() - } - } - - #[cold] - fn register(&self) -> Tid { - let id = REGISTRY - .free - .lock() - .ok() - .and_then(|mut free| { - if free.len() > 1 { - free.pop_front() - } else { - None - } - }) - .unwrap_or_else(|| REGISTRY.next.fetch_add(1, Ordering::AcqRel)); - debug_assert!(id <= Tid::BITS, "thread ID overflow!"); - self.0.set(Some(id)); - Tid::new(id) - } -} - -impl Drop for Registration { - fn drop(&mut self) { - if let Some(id) = self.0.get() { - if let Ok(mut free) = REGISTRY.free.lock() { - free.push_back(id); - } - } - } -} diff --git a/tokio/src/net/driver/reactor/reactor/sharded_slab/page/slot.rs b/tokio/src/net/driver/reactor/reactor/sharded_slab/page/slot.rs deleted file mode 100644 index 8ef4aa2d1..000000000 --- a/tokio/src/net/driver/reactor/reactor/sharded_slab/page/slot.rs +++ /dev/null @@ -1,102 +0,0 @@ -use super::super::{RESERVED_BITS, WIDTH}; -use super::ScheduledIo; -use crate::sync::{ - atomic::{AtomicBool, Ordering}, - CausalCell, -}; - -#[derive(Debug)] -pub(crate) struct Slot { - empty: AtomicBool, - /// The offset of the next item on the free list. - next: CausalCell, - /// The data stored in the slot. - item: ScheduledIo, -} - -#[repr(transparent)] -#[derive(Copy, Clone, Debug, PartialEq, Eq, Ord, PartialOrd)] -pub(crate) struct Generation { - value: usize, -} - -impl Pack for Generation { - /// Use all the remaining bits in the word for the generation counter, minus - /// any bits reserved by the user. - const LEN: usize = (WIDTH - RESERVED_BITS) - Self::SHIFT; - - type Prev = Tid; - - #[inline(always)] - fn from_usize(u: usize) -> Self { - debug_assert!(u <= Self::BITS); - Self::new(u) - } - - #[inline(always)] - fn as_usize(&self) -> usize { - self.value - } -} - -impl Generation { - fn new(value: usize) -> Self { - Self { value } - } -} - -impl Slot { - pub(super) fn new(next: usize) -> Self { - Self { - empty: AtomicBool::new(true), - item: ScheduledIo::default(), - next: CausalCell::new(next), - } - } - - #[inline(always)] - pub(super) fn get(&self, gen: Generation) -> Option<&T> { - let current = self.gen.load(Ordering::Acquire); - test_println!("-> get {:?}; current={:?}", gen, current); - - // Is the index's generation the same as the current generation? If not, - // the item that index referred to was removed, so return `None`. - if gen.value != current { - return None; - } - - Some(&self.item) - } - - #[inline] - pub(super) fn insert(&self) -> Generation { - Generation::from_usize(self.gen.load(Ordering::Acquire)) - } - - #[inline(always)] - pub(super) fn next(&self) -> usize { - self.next.with(|next| unsafe { *next }) - } - - #[inline] - pub(super) fn reset(&self, gen: Generation) -> bool { - let next = (gen.value + 1) % Generation::BITS; - let actual = self - .generation - .compare_and_swap(gen.value, next, Ordering::AcqRel); - test_println!("-> remove {:?}; next={:?}; actual={:?}", gen, next, actual); - if actual != gen { - return false; - }; - - self.item.reset(); - true - } - - #[inline(always)] - pub(super) fn set_next(&self, next: usize) { - self.next.with_mut(|n| unsafe { - (*n) = next; - }) - } -} diff --git a/tokio/src/net/mod.rs b/tokio/src/net/mod.rs index 50021e37b..f02b6259d 100644 --- a/tokio/src/net/mod.rs +++ b/tokio/src/net/mod.rs @@ -24,11 +24,6 @@ mod addr; pub use addr::ToSocketAddrs; -cfg_io_driver! { - pub mod driver; - pub mod util; -} - cfg_tcp! { pub mod tcp; pub use tcp::{TcpListener, TcpStream}; diff --git a/tokio/src/net/tcp/listener.rs b/tokio/src/net/tcp/listener.rs index c0e7f1b34..33316bad0 100644 --- a/tokio/src/net/tcp/listener.rs +++ b/tokio/src/net/tcp/listener.rs @@ -1,6 +1,6 @@ use crate::future::poll_fn; +use crate::io::PollEvented; use crate::net::tcp::{Incoming, TcpStream}; -use crate::net::util::PollEvented; use crate::net::ToSocketAddrs; use std::convert::TryFrom; diff --git a/tokio/src/net/tcp/stream.rs b/tokio/src/net/tcp/stream.rs index 1a844e86c..f1905d879 100644 --- a/tokio/src/net/tcp/stream.rs +++ b/tokio/src/net/tcp/stream.rs @@ -1,7 +1,6 @@ use crate::future::poll_fn; -use crate::io::{AsyncRead, AsyncWrite}; +use crate::io::{AsyncRead, AsyncWrite, PollEvented}; use crate::net::tcp::split::{split, ReadHalf, WriteHalf}; -use crate::net::util::PollEvented; use crate::net::ToSocketAddrs; use bytes::{Buf, BufMut}; diff --git a/tokio/src/net/udp/socket.rs b/tokio/src/net/udp/socket.rs index b1d1e5cc4..085090147 100644 --- a/tokio/src/net/udp/socket.rs +++ b/tokio/src/net/udp/socket.rs @@ -1,6 +1,6 @@ use crate::future::poll_fn; +use crate::io::PollEvented; use crate::net::udp::split::{split, UdpSocketRecvHalf, UdpSocketSendHalf}; -use crate::net::util::PollEvented; use crate::net::ToSocketAddrs; use std::convert::TryFrom; diff --git a/tokio/src/net/unix/datagram.rs b/tokio/src/net/unix/datagram.rs index 55eebd6c1..a26480cb1 100644 --- a/tokio/src/net/unix/datagram.rs +++ b/tokio/src/net/unix/datagram.rs @@ -1,5 +1,5 @@ use crate::future::poll_fn; -use crate::net::util::PollEvented; +use crate::io::PollEvented; use std::convert::TryFrom; use std::fmt; diff --git a/tokio/src/net/unix/listener.rs b/tokio/src/net/unix/listener.rs index c7de68b13..c7c853350 100644 --- a/tokio/src/net/unix/listener.rs +++ b/tokio/src/net/unix/listener.rs @@ -1,6 +1,6 @@ use crate::future::poll_fn; +use crate::io::PollEvented; use crate::net::unix::{Incoming, UnixStream}; -use crate::net::util::PollEvented; use mio::Ready; use mio_uds; diff --git a/tokio/src/net/unix/stream.rs b/tokio/src/net/unix/stream.rs index 774d20f99..83d3c8d60 100644 --- a/tokio/src/net/unix/stream.rs +++ b/tokio/src/net/unix/stream.rs @@ -1,8 +1,7 @@ use crate::future::poll_fn; -use crate::io::{AsyncRead, AsyncWrite}; +use crate::io::{AsyncRead, AsyncWrite, PollEvented}; use crate::net::unix::split::{split, ReadHalf, WriteHalf}; use crate::net::unix::ucred::{self, UCred}; -use crate::net::util::PollEvented; use bytes::{Buf, BufMut}; use iovec::IoVec; diff --git a/tokio/src/net/util/mod.rs b/tokio/src/net/util/mod.rs deleted file mode 100644 index 1f46a5366..000000000 --- a/tokio/src/net/util/mod.rs +++ /dev/null @@ -1,4 +0,0 @@ -//! Utilities for implementing networking types. - -mod poll_evented; -pub use self::poll_evented::PollEvented; diff --git a/tokio/src/process/unix/mod.rs b/tokio/src/process/unix/mod.rs index 72f6f0bfd..c25d98974 100644 --- a/tokio/src/process/unix/mod.rs +++ b/tokio/src/process/unix/mod.rs @@ -27,7 +27,7 @@ use orphan::{OrphanQueue, OrphanQueueImpl, Wait}; mod reap; use reap::Reaper; -use crate::net::util::PollEvented; +use crate::io::PollEvented; use crate::process::kill::Kill; use crate::process::SpawnedChild; use crate::signal::unix::{signal, Signal, SignalKind}; diff --git a/tokio/src/process/windows.rs b/tokio/src/process/windows.rs index d25807d63..54330e36d 100644 --- a/tokio/src/process/windows.rs +++ b/tokio/src/process/windows.rs @@ -15,7 +15,7 @@ //! `RegisterWaitForSingleObject` and then wait on the other end of the oneshot //! from then on out. -use crate::net::util::PollEvented; +use crate::io::PollEvented; use crate::process::kill::Kill; use crate::process::SpawnedChild; use crate::sync::oneshot; diff --git a/tokio/src/runtime/io.rs b/tokio/src/runtime/io.rs index 564e51a65..014d39de2 100644 --- a/tokio/src/runtime/io.rs +++ b/tokio/src/runtime/io.rs @@ -7,7 +7,7 @@ pub(crate) use std::io::Result; cfg_io_driver! { - use crate::net::driver; + use crate::io::driver; use std::io; @@ -16,7 +16,7 @@ cfg_io_driver! { /// When the `io-driver` feature is enabled, this is the "real" I/O driver /// backed by Mio. Without the `io-driver` feature, this is a thread parker /// backed by a condition variable. - pub(crate) type Driver = driver::Reactor; + pub(crate) type Driver = driver::Driver; /// The handle the runtime stores for future use. /// @@ -24,7 +24,7 @@ cfg_io_driver! { pub(crate) type Handle = driver::Handle; pub(crate) fn create_driver() -> io::Result<(Driver, Handle)> { - let driver = driver::Reactor::new()?; + let driver = driver::Driver::new()?; let handle = driver.handle(); Ok((driver, handle)) diff --git a/tokio/src/signal/unix.rs b/tokio/src/signal/unix.rs index 8551e85ce..bb0afffea 100644 --- a/tokio/src/signal/unix.rs +++ b/tokio/src/signal/unix.rs @@ -5,8 +5,7 @@ #![cfg(unix)] -use crate::io::AsyncRead; -use crate::net::util::PollEvented; +use crate::io::{AsyncRead, PollEvented}; use crate::signal::registry::{globals, EventId, EventInfo, Globals, Init, Storage}; use crate::sync::mpsc::{channel, Receiver}; diff --git a/tokio/src/sync/task/atomic_waker.rs b/tokio/src/sync/task/atomic_waker.rs index 49a0ac048..127eed357 100644 --- a/tokio/src/sync/task/atomic_waker.rs +++ b/tokio/src/sync/task/atomic_waker.rs @@ -1,3 +1,5 @@ +#![cfg_attr(any(loom, not(feature = "sync")), allow(dead_code, unreachable_pub))] + use crate::loom::cell::CausalCell; use crate::loom::sync::atomic::{self, AtomicUsize}; diff --git a/tokio/src/util/bit.rs b/tokio/src/util/bit.rs new file mode 100644 index 000000000..03fb0cfbe --- /dev/null +++ b/tokio/src/util/bit.rs @@ -0,0 +1,86 @@ +use std::fmt; + +#[derive(Clone, Copy)] +pub(crate) struct Pack { + mask: usize, + shift: u32, +} + +impl Pack { + /// Value is packed in the `width` most-significant bits. + pub(crate) const fn most_significant(width: u32) -> Pack { + let mask = mask_for(width).reverse_bits(); + + Pack { + mask, + shift: mask.trailing_zeros(), + } + } + + /// Value is packed in the `width` least-significant bits. + pub(crate) const fn least_significant(width: u32) -> Pack { + let mask = mask_for(width); + + Pack { + mask, + shift: 0, + } + } + + /// Value is packed in the `width` more-significant bits. + pub(crate) const fn then(&self, width: u32) -> Pack { + let shift = pointer_width() - self.mask.leading_zeros(); + let mask = mask_for(width) << shift; + + Pack { + mask, + shift, + } + } + + /// Mask used to unpack value + pub(crate) const fn mask(&self) -> usize { + self.mask + } + + /// Width, in bits, dedicated to storing the value. + pub(crate) const fn width(&self) -> u32 { + pointer_width() - (self.mask >> self.shift).leading_zeros() + } + + /// Max representable value + pub(crate) const fn max_value(&self) -> usize { + (1 << self.width()) - 1 + } + + pub(crate) fn pack(&self, value: usize, base: usize) -> usize { + assert!(value <= self.max_value()); + (base & !self.mask) | (value << self.shift) + } + + pub(crate) fn unpack(&self, src: usize) -> usize { + unpack(src, self.mask, self.shift) + } +} + +impl fmt::Debug for Pack { + fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(fmt, "Pack {{ mask: {:b}, shift: {} }}", self.mask, self.shift) + } +} + +/// Returns the width of a pointer in bits +pub(crate) const fn pointer_width() -> u32 { + std::mem::size_of::() as u32 * 8 +} + +/// Returns a `usize` with the right-most `n` bits set. +pub(crate) const fn mask_for(n: u32) -> usize { + let shift = 1usize.wrapping_shl(n - 1); + shift | (shift - 1) +} + +/// Unpack a value using a mask & shift +pub(crate) const fn unpack(src: usize, mask: usize, shift: u32) -> usize { + (src & mask) >> shift +} diff --git a/tokio/src/util/mod.rs b/tokio/src/util/mod.rs index 44377fcf9..a50b33ac3 100644 --- a/tokio/src/util/mod.rs +++ b/tokio/src/util/mod.rs @@ -1,5 +1,12 @@ -mod pad; -pub(crate) use self::pad::CachePadded; +cfg_io_driver! { + pub(crate) mod bit; + pub(crate) mod slab; +} -mod rand; -pub(crate) use self::rand::FastRand; +cfg_rt_threaded! { + mod pad; + pub(crate) use pad::CachePadded; + + mod rand; + pub(crate) use rand::FastRand; +} diff --git a/tokio/src/util/slab/addr.rs b/tokio/src/util/slab/addr.rs new file mode 100644 index 000000000..2efe93efd --- /dev/null +++ b/tokio/src/util/slab/addr.rs @@ -0,0 +1,155 @@ +//! Tracks the location of an entry in a slab. +//! +//! # Index packing +//! +//! A slab index consists of multiple indices packed into a single `usize` value +//! that correspond to different parts of the slab. +//! +//! The least significant `MAX_PAGES + INITIAL_PAGE_SIZE.trailing_zeros() + 1` +//! bits store the address within a shard, starting at 0 for the first slot on +//! the first page. To index a slot within a shard, we first find the index of +//! the page that the address falls on, and then the offset of the slot within +//! that page. +//! +//! Since every page is twice as large as the previous page, and all page sizes +//! are powers of two, we can determine the page index that contains a given +//! address by shifting the address down by the smallest page size and looking +//! at how many twos places necessary to represent that number, telling us what +//! power of two page size it fits inside of. We can determine the number of +//! twos places by counting the number of leading zeros (unused twos places) in +//! the number's binary representation, and subtracting that count from the +//! total number of bits in a word. +//! +//! Once we know what page contains an address, we can subtract the size of all +//! previous pages from the address to determine the offset within the page. +//! +//! After the page address, the next `MAX_THREADS.trailing_zeros() + 1` least +//! significant bits are the thread ID. These are used to index the array of +//! shards to find which shard a slot belongs to. If an entry is being removed +//! and the thread ID of its index matches that of the current thread, we can +//! use the `remove_local` fast path; otherwise, we have to use the synchronized +//! `remove_remote` path. +//! +//! Finally, a generation value is packed into the index. The `RESERVED_BITS` +//! most significant bits are left unused, and the remaining bits between the +//! last bit of the thread ID and the first reserved bit are used to store the +//! generation. The generation is used as part of an atomic read-modify-write +//! loop every time a `ScheduledIo`'s readiness is modified, or when the +//! resource is removed, to guard against the ABA problem. +//! +//! Visualized: +//! +//! ```text +//! ┌──────────┬───────────────┬──────────────────┬──────────────────────────┐ +//! │ reserved │ generation │ thread ID │ address │ +//! └▲─────────┴▲──────────────┴▲─────────────────┴▲────────────────────────▲┘ +//! │ │ │ │ │ +//! bits(usize) │ bits(MAX_THREADS) │ 0 +//! │ │ +//! bits(usize) - RESERVED MAX_PAGES + bits(INITIAL_PAGE_SIZE) +//! ``` + +use crate::util::bit; +use crate::util::slab::{Generation, MAX_PAGES, MAX_THREADS, INITIAL_PAGE_SIZE}; + +use std::usize; + +/// References the location at which an entry is stored in a slab. +#[derive(Debug, Copy, Clone, Eq, PartialEq)] +pub(crate) struct Address(usize); + +const PAGE_INDEX_SHIFT: u32 = INITIAL_PAGE_SIZE.trailing_zeros() + 1; + +/// Address in the shard +const SLOT: bit::Pack = bit::Pack::least_significant( + MAX_PAGES as u32 + PAGE_INDEX_SHIFT); + +/// Masks the thread identifier +const THREAD: bit::Pack = SLOT.then(MAX_THREADS.trailing_zeros() + 1); + +/// Masks the generation +const GENERATION: bit::Pack = THREAD.then( + bit::pointer_width().wrapping_sub(RESERVED.width() + THREAD.width() + SLOT.width())); + +// Chosen arbitrarily +const RESERVED: bit::Pack = bit::Pack::most_significant(5); + +impl Address { + /// Represents no entry, picked to avoid collision with Mio's internals. + /// This value should not be passed to mio. + pub(crate) const NULL: usize = usize::MAX >> 1; + + /// Re-exported by `Generation`. + pub(super) const GENERATION_WIDTH: u32 = GENERATION.width(); + + pub(super) fn new(shard_index: usize, generation: Generation) -> Address { + let mut repr = 0; + + repr = SLOT.pack(shard_index, repr); + repr = GENERATION.pack(generation.to_usize(), repr); + + Address(repr) + } + + /// Convert from a `usize` representation. + pub(crate) fn from_usize(src: usize) -> Address { + assert_ne!(src, Self::NULL); + + Address(src) + } + + /// Convert to a `usize` representation + pub(crate) fn to_usize(self) -> usize { + self.0 + } + + pub(crate) fn generation(self) -> Generation { + Generation::new(GENERATION.unpack(self.0)) + } + + /// Returns the page index + pub(super) fn page(self) -> usize { + // Since every page is twice as large as the previous page, and all page + // sizes are powers of two, we can determine the page index that + // contains a given address by shifting the address down by the smallest + // page size and looking at how many twos places necessary to represent + // that number, telling us what power of two page size it fits inside + // of. We can determine the number of twos places by counting the number + // of leading zeros (unused twos places) in the number's binary + // representation, and subtracting that count from the total number of + // bits in a word. + let slot_shifted = (self.slot() + INITIAL_PAGE_SIZE) >> PAGE_INDEX_SHIFT; + (bit::pointer_width() - slot_shifted.leading_zeros()) as usize + } + + /// Returns the slot index + pub(super) fn slot(self) -> usize { + SLOT.unpack(self.0) + } +} + +#[cfg(test)] +cfg_not_loom! { + use proptest::proptest; + + #[test] + fn test_pack_format() { + assert_eq!(5, RESERVED.width()); + assert_eq!(0b11111, RESERVED.max_value()); + } + + proptest! { + #[test] + fn address_roundtrips( + slot in 0usize..SLOT.max_value(), + generation in 0usize..Generation::MAX, + ) { + let address = Address::new(slot, Generation::new(generation)); + // Round trip + let address = Address::from_usize(address.to_usize()); + + assert_eq!(address.slot(), slot); + assert_eq!(address.generation().to_usize(), generation); + } + } +} diff --git a/tokio/src/util/slab/entry.rs b/tokio/src/util/slab/entry.rs new file mode 100644 index 000000000..2e0b10b0f --- /dev/null +++ b/tokio/src/util/slab/entry.rs @@ -0,0 +1,7 @@ +use crate::util::slab::Generation; + +pub(crate) trait Entry: Default { + fn generation(&self) -> Generation; + + fn reset(&self, generation: Generation) -> bool; +} diff --git a/tokio/src/util/slab/generation.rs b/tokio/src/util/slab/generation.rs new file mode 100644 index 000000000..4b16b2caf --- /dev/null +++ b/tokio/src/util/slab/generation.rs @@ -0,0 +1,32 @@ +use crate::util::bit; +use crate::util::slab::Address; + +/// An mutation identifier for a slot in the slab. The generation helps prevent +/// accessing an entry with an outdated token. +#[derive(Copy, Clone, Debug, PartialEq, Eq, Ord, PartialOrd)] +pub(crate) struct Generation(usize); + +impl Generation { + pub(crate) const WIDTH: u32 = Address::GENERATION_WIDTH; + + pub(super) const MAX: usize = bit::mask_for(Address::GENERATION_WIDTH); + + /// Create a new generation + /// + /// # Panics + /// + /// Panics if `value` is greater than max generation. + pub(crate) fn new(value: usize) -> Generation { + assert!(value <= Self::MAX); + Generation(value) + } + + /// Returns the next generation value + pub(crate) fn next(self) -> Generation { + Generation((self.0 + 1) & Self::MAX) + } + + pub(crate) fn to_usize(self) -> usize { + self.0 + } +} diff --git a/tokio/src/util/slab/mod.rs b/tokio/src/util/slab/mod.rs new file mode 100644 index 000000000..e9b53c536 --- /dev/null +++ b/tokio/src/util/slab/mod.rs @@ -0,0 +1,109 @@ +//! A lock-free concurrent slab. + +mod addr; +pub(crate) use addr::Address; + +mod entry; +pub(crate) use entry::Entry; + +mod generation; +pub(crate) use generation::Generation; + +mod page; + +mod shard; +use shard::Shard; + +mod slot; +use slot::Slot; + +mod stack; +use stack::TransferStack; + +#[cfg(all(loom, test))] +mod tests; + +use crate::loom::sync::Mutex; +use crate::util::bit; + +use std::fmt; + +#[cfg(target_pointer_width = "64")] +const MAX_THREADS: usize = 4096; + +#[cfg(target_pointer_width = "32")] +const MAX_THREADS: usize = 2048; + +/// Max number of pages per slab +const MAX_PAGES: usize = bit::pointer_width() as usize / 4; + +cfg_not_loom! { + /// Size of first page + const INITIAL_PAGE_SIZE: usize = 32; +} + +cfg_loom! { + const INITIAL_PAGE_SIZE: usize = 2; +} + +/// A sharded slab. +pub(crate) struct Slab { + // Signal shard for now. Eventually there will be more. + shard: Shard, + local: Mutex<()>, +} + +unsafe impl Send for Slab {} +unsafe impl Sync for Slab {} + +impl Slab { + /// Returns a new slab with the default configuration parameters. + pub(crate) fn new() -> Slab { + Slab { + shard: Shard::new(), + local: Mutex::new(()), + } + } + + /// allocs a value into the slab, returning a key that can be used to + /// access it. + /// + /// If this function returns `None`, then the shard for the current thread + /// is full and no items can be added until some are removed, or the maximum + /// number of shards has been reached. + pub(crate) fn alloc(&self) -> Option
{ + // we must lock the slab to alloc an item. + let _local = self.local.lock().unwrap(); + self.shard.alloc() + } + + /// Removes the value associated with the given key from the slab. + pub(crate) fn remove(&self, idx: Address) { + // try to lock the slab so that we can use `remove_local`. + let lock = self.local.try_lock(); + + // if we were able to lock the slab, we are "local" and can use the fast + // path; otherwise, we will use `remove_remote`. + if lock.is_ok() { + self.shard.remove_local(idx) + } else { + self.shard.remove_remote(idx) + } + } + + /// Return a reference to the value associated with the given key. + /// + /// If the slab does not contain a value for the given key, `None` is + /// returned instead. + pub(crate) fn get(&self, token: Address) -> Option<&T> { + self.shard.get(token) + } +} + +impl fmt::Debug for Slab { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Slab") + .field("shard", &self.shard) + .finish() + } +} diff --git a/tokio/src/net/driver/reactor/dispatch/page/mod.rs b/tokio/src/util/slab/page.rs similarity index 54% rename from tokio/src/net/driver/reactor/dispatch/page/mod.rs rename to tokio/src/util/slab/page.rs index 9bef94f31..6b1c64c05 100644 --- a/tokio/src/net/driver/reactor/dispatch/page/mod.rs +++ b/tokio/src/util/slab/page.rs @@ -1,71 +1,24 @@ -use super::{Pack, INITIAL_PAGE_SIZE, WIDTH}; use crate::loom::cell::CausalCell; +use crate::util::slab::{Address, Entry, Slot, TransferStack, INITIAL_PAGE_SIZE}; -pub(crate) mod scheduled_io; -mod stack; -pub(crate) use self::scheduled_io::ScheduledIo; -use self::stack::TransferStack; use std::fmt; -/// A page address encodes the location of a slot within a shard (the page -/// number and offset within that page) as a single linear value. -#[repr(transparent)] -#[derive(Copy, Clone, Eq, PartialEq, PartialOrd, Ord)] -pub(crate) struct Addr { - addr: usize, -} - -impl Addr { - const NULL: usize = Self::BITS + 1; - const INDEX_SHIFT: usize = INITIAL_PAGE_SIZE.trailing_zeros() as usize + 1; - - pub(crate) fn index(self) -> usize { - // Since every page is twice as large as the previous page, and all page sizes - // are powers of two, we can determine the page index that contains a given - // address by shifting the address down by the smallest page size and - // looking at how many twos places necessary to represent that number, - // telling us what power of two page size it fits inside of. We can - // determine the number of twos places by counting the number of leading - // zeros (unused twos places) in the number's binary representation, and - // subtracting that count from the total number of bits in a word. - WIDTH - ((self.addr + INITIAL_PAGE_SIZE) >> Self::INDEX_SHIFT).leading_zeros() as usize - } - - pub(crate) fn offset(self) -> usize { - self.addr - } -} - -pub(super) fn size(n: usize) -> usize { - INITIAL_PAGE_SIZE * 2usize.pow(n as _) -} - -impl Pack for Addr { - const LEN: usize = super::MAX_PAGES + Self::INDEX_SHIFT; - - type Prev = (); - - fn as_usize(&self) -> usize { - self.addr - } - - fn from_usize(addr: usize) -> Self { - debug_assert!(addr <= Self::BITS); - Self { addr } - } -} - -pub(in crate::net::driver) type Iter<'a> = std::slice::Iter<'a, ScheduledIo>; - +/// Data accessed only by the thread that owns the shard. pub(crate) struct Local { head: CausalCell, } -pub(crate) struct Shared { +/// Data accessed by any thread. +pub(crate) struct Shared { remote: TransferStack, size: usize, prev_sz: usize, - slab: CausalCell>>, + slab: CausalCell]>>>, +} + +/// Returns the size of the page at index `n` +pub(super) fn size(n: usize) -> usize { + INITIAL_PAGE_SIZE << n } impl Local { @@ -75,12 +28,10 @@ impl Local { } } - #[inline(always)] fn head(&self) -> usize { self.head.with(|head| unsafe { *head }) } - #[inline(always)] fn set_head(&self, new_head: usize) { self.head.with_mut(|head| unsafe { *head = new_head; @@ -88,10 +39,8 @@ impl Local { } } -impl Shared { - const NULL: usize = Addr::NULL; - - pub(crate) fn new(size: usize, prev_sz: usize) -> Self { +impl Shared { + pub(crate) fn new(size: usize, prev_sz: usize) -> Shared { Self { prev_sz, size, @@ -113,8 +62,9 @@ impl Shared { debug_assert!(self.slab.with(|s| unsafe { (*s).is_none() })); let mut slab = Vec::with_capacity(self.size); - slab.extend((1..self.size).map(ScheduledIo::new)); - slab.push(ScheduledIo::new(Self::NULL)); + slab.extend((1..self.size).map(Slot::new)); + slab.push(Slot::new(Address::NULL)); + self.slab.with_mut(|s| { // this mut access is safe — it only occurs to initially // allocate the page, which only happens on this thread; if the @@ -126,8 +76,7 @@ impl Shared { }); } - #[inline] - pub(crate) fn alloc(&self, local: &Local) -> Option { + pub(crate) fn alloc(&self, local: &Local) -> Option
{ let head = local.head(); // are there any items on the local free list? (fast path) @@ -141,7 +90,7 @@ impl Shared { // if the head is still null, both the local and remote free lists are // empty --- we can't fit any more items on this page. - if head == Self::NULL { + if head == Address::NULL { return None; } @@ -155,58 +104,64 @@ impl Shared { let slab = unsafe { &*(slab) } .as_ref() .expect("page must have been allocated to alloc!"); + let slot = &slab[head]; + local.set_head(slot.next()); - slot.alloc() + slot.generation() }); let index = head + self.prev_sz; - Some(gen.pack(index)) + + Some(Address::new(index, gen)) } - #[inline] - pub(in crate::net::driver) fn get(&self, addr: Addr) -> Option<&ScheduledIo> { - let page_offset = addr.offset() - self.prev_sz; + pub(crate) fn get(&self, addr: Address) -> Option<&T> { + let page_offset = addr.slot() - self.prev_sz; + self.slab .with(|slab| unsafe { &*slab }.as_ref()?.get(page_offset)) + .map(|slot| slot.get()) } - pub(crate) fn remove_local(&self, local: &Local, addr: Addr, idx: usize) { - let offset = addr.offset() - self.prev_sz; + pub(crate) fn remove_local(&self, local: &Local, addr: Address) { + let offset = addr.slot() - self.prev_sz; + self.slab.with(|slab| { let slab = unsafe { &*slab }.as_ref(); + let slot = if let Some(slot) = slab.and_then(|slab| slab.get(offset)) { slot } else { return; }; - if slot.reset(scheduled_io::Generation::from_packed(idx)) { + + if slot.reset(addr.generation()) { slot.set_next(local.head()); local.set_head(offset); } }) } - pub(crate) fn remove_remote(&self, addr: Addr, idx: usize) { - let offset = addr.offset() - self.prev_sz; + pub(crate) fn remove_remote(&self, addr: Address) { + let offset = addr.slot() - self.prev_sz; + self.slab.with(|slab| { let slab = unsafe { &*slab }.as_ref(); + let slot = if let Some(slot) = slab.and_then(|slab| slab.get(offset)) { slot } else { return; }; - if !slot.reset(scheduled_io::Generation::from_packed(idx)) { + + if !slot.reset(addr.generation()) { return; } + self.remote.push(offset, |next| slot.set_next(next)); }) } - - pub(in crate::net::driver) fn iter(&self) -> Option> { - let slab = self.slab.with(|slab| unsafe { (&*slab).as_ref() }); - slab.map(|slab| slab.iter()) - } } impl fmt::Debug for Local { @@ -220,7 +175,7 @@ impl fmt::Debug for Local { } } -impl fmt::Debug for Shared { +impl fmt::Debug for Shared { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Shared") .field("remote", &self.remote) @@ -230,28 +185,3 @@ impl fmt::Debug for Shared { .finish() } } - -impl fmt::Debug for Addr { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("Addr") - .field("addr", &format_args!("{:#0x}", &self.addr)) - .field("index", &self.index()) - .field("offset", &self.offset()) - .finish() - } -} - -#[cfg(all(test, not(loom)))] -mod test { - use super::*; - use proptest::prelude::*; - - proptest! { - #[test] - fn addr_roundtrips(pidx in 0usize..Addr::BITS) { - let addr = Addr::from_usize(pidx); - let packed = addr.pack(0); - assert_eq!(addr, Addr::from_packed(packed)); - } - } -} diff --git a/tokio/src/util/slab/shard.rs b/tokio/src/util/slab/shard.rs new file mode 100644 index 000000000..711e36d3d --- /dev/null +++ b/tokio/src/util/slab/shard.rs @@ -0,0 +1,108 @@ +use crate::util::slab::{Address, Entry, page, MAX_PAGES}; + +use std::fmt; + +// ┌─────────────┐ ┌────────┐ +// │ page 1 │ │ │ +// ├─────────────┤ ┌───▶│ next──┼─┐ +// │ page 2 │ │ ├────────┤ │ +// │ │ │ │XXXXXXXX│ │ +// │ local_free──┼─┘ ├────────┤ │ +// │ global_free─┼─┐ │ │◀┘ +// ├─────────────┤ └───▶│ next──┼─┐ +// │ page 3 │ ├────────┤ │ +// └─────────────┘ │XXXXXXXX│ │ +// ... ├────────┤ │ +// ┌─────────────┐ │XXXXXXXX│ │ +// │ page n │ ├────────┤ │ +// └─────────────┘ │ │◀┘ +// │ next──┼───▶ +// ├────────┤ +// │XXXXXXXX│ +// └────────┘ +// ... +pub(super) struct Shard { + /// The local free list for each page. + /// + /// These are only ever accessed from this shard's thread, so they are + /// stored separately from the shared state for the page that can be + /// accessed concurrently, to minimize false sharing. + local: Box<[page::Local]>, + /// The shared state for each page in this shard. + /// + /// This consists of the page's metadata (size, previous size), remote free + /// list, and a pointer to the actual array backing that page. + shared: Box<[page::Shared]>, +} + +impl Shard { + pub(super) fn new() -> Shard { + let mut total_sz = 0; + let shared = (0..MAX_PAGES) + .map(|page_num| { + let sz = page::size(page_num); + let prev_sz = total_sz; + total_sz += sz; + page::Shared::new(sz, prev_sz) + }) + .collect(); + + let local = (0..MAX_PAGES).map(|_| page::Local::new()).collect(); + + Shard { + local, + shared, + } + } + + pub(super) fn alloc(&self) -> Option
{ + // Can we fit the value into an existing page? + for (page_idx, page) in self.shared.iter().enumerate() { + let local = self.local(page_idx); + + if let Some(page_offset) = page.alloc(local) { + return Some(page_offset); + } + } + + None + } + + pub(super) fn get(&self, addr: Address) -> Option<&T> { + let page_idx = addr.page(); + + if page_idx > self.shared.len() { + return None; + } + + self.shared[page_idx].get(addr) + } + + /// Remove an item on the shard's local thread. + pub(super) fn remove_local(&self, addr: Address) { + let page_idx = addr.page(); + + if let Some(page) = self.shared.get(page_idx) { + page.remove_local(self.local(page_idx), addr); + } + } + + /// Remove an item, while on a different thread from the shard's local thread. + pub(super) fn remove_remote(&self, addr: Address) { + if let Some(page) = self.shared.get(addr.page()) { + page.remove_remote(addr); + } + } + + fn local(&self, i: usize) -> &page::Local { + &self.local[i] + } +} + +impl fmt::Debug for Shard { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Shard") + .field("shared", &self.shared) + .finish() + } +} diff --git a/tokio/src/util/slab/slot.rs b/tokio/src/util/slab/slot.rs new file mode 100644 index 000000000..59baacf5b --- /dev/null +++ b/tokio/src/util/slab/slot.rs @@ -0,0 +1,42 @@ +use crate::loom::cell::CausalCell; +use crate::util::slab::{Generation, Entry}; + +/// Stores an entry in the slab. +pub(super) struct Slot { + next: CausalCell, + entry: T, +} + +impl Slot { + /// Initialize a new `Slot` linked to `next`. + /// + /// The entry is initialized to a default value. + pub(super) fn new(next: usize) -> Slot { + Slot { + next: CausalCell::new(next), + entry: T::default(), + } + } + + pub(super) fn get(&self) -> &T { + &self.entry + } + + pub(super) fn generation(&self) -> Generation { + self.entry.generation() + } + + pub(super) fn reset(&self, generation: Generation) -> bool { + self.entry.reset(generation) + } + + pub(super) fn next(&self) -> usize { + self.next.with(|next| unsafe { *next }) + } + + pub(super) fn set_next(&self, next: usize) { + self.next.with_mut(|n| unsafe { + (*n) = next; + }) + } +} diff --git a/tokio/src/util/slab/stack.rs b/tokio/src/util/slab/stack.rs new file mode 100644 index 000000000..0ae0d7100 --- /dev/null +++ b/tokio/src/util/slab/stack.rs @@ -0,0 +1,58 @@ +use crate::loom::sync::atomic::AtomicUsize; +use crate::util::slab::Address; + +use std::fmt; +use std::sync::atomic::Ordering; +use std::usize; + +pub(super) struct TransferStack { + head: AtomicUsize, +} + +impl TransferStack { + pub(super) fn new() -> Self { + Self { + head: AtomicUsize::new(Address::NULL), + } + } + + pub(super) fn pop_all(&self) -> Option { + let val = self.head.swap(Address::NULL, Ordering::Acquire); + + if val == Address::NULL { + None + } else { + Some(val) + } + } + + pub(super) fn push(&self, value: usize, before: impl Fn(usize)) { + let mut next = self.head.load(Ordering::Relaxed); + + loop { + before(next); + + match self + .head + .compare_exchange(next, value, Ordering::AcqRel, Ordering::Acquire) + { + // lost the race! + Err(actual) => next = actual, + Ok(_) => return, + } + } + } +} + +impl fmt::Debug for TransferStack { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // Loom likes to dump all its internal state in `fmt::Debug` impls, so + // we override this to just print the current value in tests. + f.debug_struct("TransferStack") + .field( + "head", + &format_args!("{:#x}", self.head.load(Ordering::Relaxed)), + ) + .finish() + } +} diff --git a/tokio/src/util/slab/tests/loom_slab.rs b/tokio/src/util/slab/tests/loom_slab.rs new file mode 100644 index 000000000..8a96a736e --- /dev/null +++ b/tokio/src/util/slab/tests/loom_slab.rs @@ -0,0 +1,327 @@ +use crate::io::driver::ScheduledIo; +use crate::util::slab::{Address, Slab}; + +use loom::sync::{Arc, Condvar, Mutex}; +use loom::thread; + +#[test] +fn local_remove() { + loom::model(|| { + let slab = Arc::new(Slab::new()); + + let s = slab.clone(); + let t1 = thread::spawn(move || { + let idx = store_val(&s, 1); + assert_eq!(get_val(&s, idx), Some(1)); + s.remove(idx); + assert_eq!(get_val(&s, idx), None); + let idx = store_val(&s, 2); + assert_eq!(get_val(&s, idx), Some(2)); + s.remove(idx); + assert_eq!(get_val(&s, idx), None); + }); + + let s = slab.clone(); + let t2 = thread::spawn(move || { + let idx = store_val(&s, 3); + assert_eq!(get_val(&s, idx), Some(3)); + s.remove(idx); + assert_eq!(get_val(&s, idx), None); + let idx = store_val(&s, 4); + s.remove(idx); + assert_eq!(get_val(&s, idx), None); + }); + + let s = slab; + let idx1 = store_val(&s, 5); + assert_eq!(get_val(&s, idx1), Some(5)); + let idx2 = store_val(&s, 6); + assert_eq!(get_val(&s, idx2), Some(6)); + s.remove(idx1); + assert_eq!(get_val(&s, idx1), None); + assert_eq!(get_val(&s, idx2), Some(6)); + s.remove(idx2); + assert_eq!(get_val(&s, idx2), None); + + t1.join().expect("thread 1 should not panic"); + t2.join().expect("thread 2 should not panic"); + }); +} + +#[test] +fn remove_remote() { + loom::model(|| { + let slab = Arc::new(Slab::new()); + + let idx1 = store_val(&slab, 1); + assert_eq!(get_val(&slab, idx1), Some(1)); + + let idx2 = store_val(&slab, 2); + assert_eq!(get_val(&slab, idx2), Some(2)); + + let idx3 = store_val(&slab, 3); + assert_eq!(get_val(&slab, idx3), Some(3)); + + let s = slab.clone(); + let t1 = thread::spawn(move || { + assert_eq!(get_val(&s, idx2), Some(2)); + s.remove(idx2); + assert_eq!(get_val(&s, idx2), None); + }); + + let s = slab.clone(); + let t2 = thread::spawn(move || { + assert_eq!(get_val(&s, idx3), Some(3)); + s.remove(idx3); + assert_eq!(get_val(&s, idx3), None); + }); + + t1.join().expect("thread 1 should not panic"); + t2.join().expect("thread 2 should not panic"); + + assert_eq!(get_val(&slab, idx1), Some(1)); + assert_eq!(get_val(&slab, idx2), None); + assert_eq!(get_val(&slab, idx3), None); + }); +} + +#[test] +fn remove_remote_and_reuse() { + loom::model(|| { + let slab = Arc::new(Slab::new()); + + let idx1 = store_val(&slab, 1); + let idx2 = store_val(&slab, 2); + + assert_eq!(get_val(&slab, idx1), Some(1)); + assert_eq!(get_val(&slab, idx2), Some(2)); + + let s = slab.clone(); + let t1 = thread::spawn(move || { + s.remove(idx1); + let value = get_val(&s, idx1); + + // We may or may not see the new value yet, depending on when + // this occurs, but we must either see the new value or `None`; + // the old value has been removed! + assert!(value == None || value == Some(3)); + }); + + let idx3 = store_when_free(&slab, 3); + t1.join().expect("thread 1 should not panic"); + + assert_eq!(get_val(&slab, idx3), Some(3)); + assert_eq!(get_val(&slab, idx2), Some(2)); + }); +} + +#[test] +fn concurrent_alloc_remove() { + loom::model(|| { + let slab = Arc::new(Slab::new()); + let pair = Arc::new((Mutex::new(None), Condvar::new())); + + let slab2 = slab.clone(); + let pair2 = pair.clone(); + let remover = thread::spawn(move || { + let (lock, cvar) = &*pair2; + for _ in 0..2 { + let mut next = lock.lock().unwrap(); + while next.is_none() { + next = cvar.wait(next).unwrap(); + } + let key = next.take().unwrap(); + slab2.remove(key); + assert_eq!(get_val(&slab2, key), None); + cvar.notify_one(); + } + }); + + let (lock, cvar) = &*pair; + for i in 0..2 { + let key = store_val(&slab, i); + + let mut next = lock.lock().unwrap(); + *next = Some(key); + cvar.notify_one(); + + // Wait for the item to be removed. + while next.is_some() { + next = cvar.wait(next).unwrap(); + } + + assert_eq!(get_val(&slab, key), None); + } + + remover.join().unwrap(); + }) +} + +#[test] +fn concurrent_remove_remote_and_reuse() { + loom::model(|| { + let slab = Arc::new(Slab::new()); + + let idx1 = store_val(&slab, 1); + let idx2 = store_val(&slab, 2); + + assert_eq!(get_val(&slab, idx1), Some(1)); + assert_eq!(get_val(&slab, idx2), Some(2)); + + let s = slab.clone(); + let s2 = slab.clone(); + let t1 = thread::spawn(move || { + s.remove(idx1); + }); + + let t2 = thread::spawn(move || { + s2.remove(idx2); + }); + + let idx3 = store_when_free(&slab, 3); + t1.join().expect("thread 1 should not panic"); + t2.join().expect("thread 1 should not panic"); + + assert!(get_val(&slab, idx1).is_none()); + assert!(get_val(&slab, idx2).is_none()); + assert_eq!(get_val(&slab, idx3), Some(3)); + }); +} + +#[test] +fn alloc_remove_get() { + loom::model(|| { + let slab = Arc::new(Slab::new()); + let pair = Arc::new((Mutex::new(None), Condvar::new())); + + let slab2 = slab.clone(); + let pair2 = pair.clone(); + let t1 = thread::spawn(move || { + let slab = slab2; + let (lock, cvar) = &*pair2; + // allocate one entry just so that we have to use the final one for + // all future allocations. + let _key0 = store_val(&slab, 0); + let key = store_val(&slab, 1); + + let mut next = lock.lock().unwrap(); + *next = Some(key); + cvar.notify_one(); + // remove the second entry + slab.remove(key); + // store a new readiness at the same location (since the slab + // already has an entry in slot 0) + store_val(&slab, 2); + }); + + let (lock, cvar) = &*pair; + // wait for the second entry to be stored... + let mut next = lock.lock().unwrap(); + while next.is_none() { + next = cvar.wait(next).unwrap(); + } + let key = next.unwrap(); + + // our generation will be stale when the second store occurs at that + // index, we must not see the value of that store. + let val = get_val(&slab, key); + assert_ne!(val, Some(2), "generation must have advanced!"); + + t1.join().unwrap(); + }) +} + +#[test] +fn alloc_remove_set() { + loom::model(|| { + let slab = Arc::new(Slab::new()); + let pair = Arc::new((Mutex::new(None), Condvar::new())); + + let slab2 = slab.clone(); + let pair2 = pair.clone(); + let t1 = thread::spawn(move || { + let slab = slab2; + let (lock, cvar) = &*pair2; + // allocate one entry just so that we have to use the final one for + // all future allocations. + let _key0 = store_val(&slab, 0); + let key = store_val(&slab, 1); + + let mut next = lock.lock().unwrap(); + *next = Some(key); + cvar.notify_one(); + + slab.remove(key); + // remove the old entry and insert a new one, with a new generation. + let key2 = slab.alloc().expect("store key 2"); + // after the remove, we must not see the value written with the + // stale index. + assert_eq!( + get_val(&slab, key), + None, + "stale set must no longer be visible" + ); + assert_eq!(get_val(&slab, key2), Some(0)); + key2 + }); + + let (lock, cvar) = &*pair; + + // wait for the second entry to be stored. the index we get from the + // other thread may become stale after a write. + let mut next = lock.lock().unwrap(); + while next.is_none() { + next = cvar.wait(next).unwrap(); + } + let key = next.unwrap(); + + // try to write to the index with our generation + slab.get(key).map(|val| val.set_readiness(key, |_| 2)); + + let key2 = t1.join().unwrap(); + // after the remove, we must not see the value written with the + // stale index either. + assert_eq!( + get_val(&slab, key), + None, + "stale set must no longer be visible" + ); + assert_eq!(get_val(&slab, key2), Some(0)); + }); +} + +fn get_val(slab: &Arc>, address: Address) -> Option { + slab.get(address).and_then(|s| s.get_readiness(address)) +} + +fn store_val(slab: &Arc>, readiness: usize) -> Address { + let key = slab.alloc().expect("allocate slot"); + + if let Some(slot) = slab.get(key) { + slot.set_readiness(key, |_| readiness) + .expect("generation should still be valid!"); + } else { + panic!("slab did not contain a value for {:?}", key); + } + + key +} + +fn store_when_free(slab: &Arc>, readiness: usize) -> Address { + let key = loop { + if let Some(key) = slab.alloc() { + break key; + } + + thread::yield_now(); + }; + + if let Some(slot) = slab.get(key) { + slot.set_readiness(key, |_| readiness) + .expect("generation should still be valid!"); + } else { + panic!("slab did not contain a value for {:?}", key); + } + + key +} diff --git a/tokio/src/util/slab/tests/loom_stack.rs b/tokio/src/util/slab/tests/loom_stack.rs new file mode 100644 index 000000000..d1121bee3 --- /dev/null +++ b/tokio/src/util/slab/tests/loom_stack.rs @@ -0,0 +1,88 @@ +use crate::util::slab::TransferStack; + +use loom::cell::CausalCell; +use loom::sync::Arc; +use loom::thread; + +#[test] +fn transfer_stack() { + loom::model(|| { + let causalities = [CausalCell::new(None), CausalCell::new(None)]; + let shared = Arc::new((causalities, TransferStack::new())); + let shared1 = shared.clone(); + let shared2 = shared.clone(); + + // Spawn two threads that both try to push to the stack. + let t1 = thread::spawn(move || { + let (causalities, stack) = &*shared1; + stack.push(0, |prev| { + causalities[0].with_mut(|c| unsafe { + *c = Some(prev); + }); + }); + }); + + let t2 = thread::spawn(move || { + let (causalities, stack) = &*shared2; + stack.push(1, |prev| { + causalities[1].with_mut(|c| unsafe { + *c = Some(prev); + }); + }); + }); + + let (causalities, stack) = &*shared; + + // Try to pop from the stack... + let mut idx = stack.pop_all(); + while idx == None { + idx = stack.pop_all(); + thread::yield_now(); + } + let idx = idx.unwrap(); + + let saw_both = causalities[idx].with(|val| { + let val = unsafe { *val }; + assert!( + val.is_some(), + "CausalCell write must happen-before index is pushed to the stack!", + ); + // were there two entries in the stack? if so, check that + // both saw a write. + if let Some(c) = causalities.get(val.unwrap()) { + c.with(|val| { + let val = unsafe { *val }; + assert!( + val.is_some(), + "CausalCell write must happen-before index is pushed to the stack!", + ); + }); + true + } else { + false + } + }); + + // We only saw one push. Ensure that the other push happens too. + if !saw_both { + // Try to pop from the stack... + let mut idx = stack.pop_all(); + while idx == None { + idx = stack.pop_all(); + thread::yield_now(); + } + let idx = idx.unwrap(); + + causalities[idx].with(|val| { + let val = unsafe { *val }; + assert!( + val.is_some(), + "CausalCell write must happen-before index is pushed to the stack!", + ); + }); + } + + t1.join().unwrap(); + t2.join().unwrap(); + }); +} diff --git a/tokio/src/util/slab/tests/mod.rs b/tokio/src/util/slab/tests/mod.rs new file mode 100644 index 000000000..7f7935446 --- /dev/null +++ b/tokio/src/util/slab/tests/mod.rs @@ -0,0 +1,2 @@ +mod loom_slab; +mod loom_stack; diff --git a/tokio/tests/net_driver.rs b/tokio/tests/io_driver.rs similarity index 83% rename from tokio/tests/net_driver.rs rename to tokio/tests/io_driver.rs index 5baa4eda8..a97de8d2a 100644 --- a/tokio/tests/net_driver.rs +++ b/tokio/tests/io_driver.rs @@ -1,7 +1,7 @@ #![warn(rust_2018_idioms)] -use tokio::net::driver::Reactor; use tokio::net::TcpListener; +use tokio::runtime; use tokio_test::{assert_ok, assert_pending}; use futures::task::{waker_ref, ArcWake}; @@ -44,7 +44,8 @@ fn test_drop_on_notify() { // shutting down. Then, when the task handle is dropped, the task itself is // dropped. - let mut reactor = assert_ok!(Reactor::new()); + let mut rt = runtime::Builder::new().basic_scheduler().build().unwrap(); + let (addr_tx, addr_rx) = mpsc::channel(); // Define a task that just drains the listener @@ -62,11 +63,11 @@ fn test_drop_on_notify() { })); { - let handle = reactor.handle(); - let _reactor = tokio::net::driver::set_default(&handle); - let waker = waker_ref(&task); - let mut cx = Context::from_waker(&waker); - assert_pending!(task.future.lock().unwrap().as_mut().poll(&mut cx)); + rt.enter(|| { + let waker = waker_ref(&task); + let mut cx = Context::from_waker(&waker); + assert_pending!(task.future.lock().unwrap().as_mut().poll(&mut cx)); + }); } // Get the address @@ -77,5 +78,6 @@ fn test_drop_on_notify() { // Establish a connection to the acceptor let _s = TcpStream::connect(&addr).unwrap(); - reactor.turn(None).unwrap(); + // Force the reactor to turn + rt.block_on(async {}); } diff --git a/tokio/tests/io_driver_drop.rs b/tokio/tests/io_driver_drop.rs new file mode 100644 index 000000000..c75631b73 --- /dev/null +++ b/tokio/tests/io_driver_drop.rs @@ -0,0 +1,44 @@ +#![warn(rust_2018_idioms)] + +use tokio::net::TcpListener; +use tokio::runtime; +use tokio_test::{assert_err, assert_pending, assert_ready, task}; + +#[test] +fn tcp_doesnt_block() { + let rt = runtime::Builder::new().basic_scheduler().build().unwrap(); + + let mut listener = rt.enter(|| { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + TcpListener::from_std(listener).unwrap() + }); + + drop(rt); + + let mut task = task::spawn(async move { + assert_err!(listener.accept().await); + }); + + assert_ready!(task.poll()); +} + +#[test] +fn drop_wakes() { + let rt = runtime::Builder::new().basic_scheduler().build().unwrap(); + + let mut listener = rt.enter(|| { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + TcpListener::from_std(listener).unwrap() + }); + + let mut task = task::spawn(async move { + assert_err!(listener.accept().await); + }); + + assert_pending!(task.poll()); + + drop(rt); + + assert!(task.is_woken()); + assert_ready!(task.poll()); +} diff --git a/tokio/tests/net_driver_drop.rs b/tokio/tests/net_driver_drop.rs deleted file mode 100644 index d52c04d15..000000000 --- a/tokio/tests/net_driver_drop.rs +++ /dev/null @@ -1,47 +0,0 @@ -#![warn(rust_2018_idioms)] - -use tokio::net::driver::{self, Reactor}; -use tokio::net::TcpListener; -use tokio_test::{assert_err, assert_pending, assert_ready, task}; - -#[test] -fn tcp_doesnt_block() { - let reactor = Reactor::new().unwrap(); - let handle = reactor.handle(); - - // Set the current reactor for this thread - let _reactor_guard = driver::set_default(&handle); - - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let mut listener = TcpListener::from_std(listener).unwrap(); - drop(reactor); - - let mut task = task::spawn(async move { - assert_err!(listener.accept().await); - }); - - assert_ready!(task.poll()); -} - -#[test] -fn drop_wakes() { - let reactor = Reactor::new().unwrap(); - let handle = reactor.handle(); - - // Set the current reactor for this thread - let _reactor_guard = driver::set_default(&handle); - - let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); - let mut listener = TcpListener::from_std(listener).unwrap(); - - let mut task = task::spawn(async move { - assert_err!(listener.accept().await); - }); - - assert_pending!(task.poll()); - - drop(reactor); - - assert!(task.is_woken()); - assert_ready!(task.poll()); -}