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.
This commit is contained in:
Carl Lerche
2019-11-20 00:05:14 -08:00
committed by GitHub
parent 7c8b8877d4
commit 69975fb960
49 changed files with 1351 additions and 2415 deletions
@@ -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<Inner>,
}
/// Return value from the `turn` method on `Reactor`.
///
/// Currently this value doesn't actually provide any functionality, but it may
/// in the future give insight into what happened during `turn`.
#[derive(Debug)]
pub struct Turn {
_priv: (),
}
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<ScheduledIo>,
/// The number of sources in `io_dispatch`.
n_sources: AtomicUsize,
@@ -81,7 +59,7 @@ thread_local! {
static CURRENT_REACTOR: RefCell<Option<Handle>> = 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<T: Send + Sync>() {}
@@ -89,11 +67,11 @@ fn _assert_kinds() {
_assert::<Handle>();
}
// ===== 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<Reactor> {
pub(crate) fn new() -> io::Result<Driver> {
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<Duration>) -> io::Result<Turn> {
self.poll(max_wait)?;
Ok(Turn { _priv: () })
}
/// Returns true if the reactor is currently idle.
///
/// Idle is defined as all tasks that have been spawned have completed,
/// either successfully or with an error.
pub fn is_idle(&self) -> bool {
self.inner.n_sources.load(SeqCst) == 0
}
fn poll(&mut self, max_wait: Option<Duration>) -> io::Result<()> {
fn turn(&mut self, max_wait: Option<Duration>) -> io::Result<()> {
// Block waiting for an event to happen, peeling out how many events
// happened.
match self.inner.io.poll(&mut self.events, max_wait) {
@@ -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<usize> {
let token = self.io_dispatch.alloc().ok_or_else(|| {
pub(super) fn add_source(&self, source: &dyn Evented) -> io::Result<Address> {
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);
+142
View File
@@ -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<usize> {
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<usize, ()> {
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))
}
+10
View File
@@ -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};
@@ -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;
@@ -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<T>(io: &T) -> io::Result<Self>
pub fn new<T>(io: &T) -> io::Result<Registration>
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);
}
}
+1 -3
View File
@@ -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
-141
View File
@@ -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: T) {}
//!
//! # #[tokio::main]
//! # async fn dox() -> Result<(), Box<dyn std::error::Error>> {
//! 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;
@@ -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<page::Iter<'a>>,
}
impl<'a> Iterator for UniqueIter<'a> {
type Item = &'a ScheduledIo;
fn next(&mut self) -> Option<Self::Item> {
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<page::Iter<'a>>,
}
impl<'a> Iterator for ShardIter<'a> {
type Item = &'a ScheduledIo;
fn next(&mut self) -> Option<Self::Item> {
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;
}
}
}
}
@@ -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;
@@ -1,89 +0,0 @@
pub(super) const WIDTH: usize = std::mem::size_of::<usize>() * 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!()
}
}
@@ -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<usize>,
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<usize> {
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<usize, ()> {
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,
}
}
}
}
@@ -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<usize> {
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();
});
}
}
@@ -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<usize> {
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<usize> {
// 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<usize> {
// 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()
}
}
@@ -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<Slab>, 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<Slab>, key: usize) -> Option<usize> {
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);
// });
// }
@@ -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<SingleShard>, 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<SingleShard>, key: usize) -> Option<usize> {
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);
// });
// }
@@ -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<Slab>, 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<Slab>, key: usize) -> Option<usize> {
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<Slab>, 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<SingleShard>, 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<SingleShard>, key: usize) -> Option<usize> {
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<SingleShard>, 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));
// }
// });
// }
@@ -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;
@@ -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<UnsafeCell<()>>,
}
/// 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<Option<usize>>);
/// 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<VecDeque<usize>>,
}
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!("<poisoned>"))
.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);
}
}
}
}
@@ -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<usize>,
/// 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;
})
}
}
-5
View File
@@ -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};
+1 -1
View File
@@ -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;
+1 -2
View File
@@ -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};
+1 -1
View File
@@ -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;
+1 -1
View File
@@ -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;
+1 -1
View File
@@ -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;
+1 -2
View File
@@ -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;
-4
View File
@@ -1,4 +0,0 @@
//! Utilities for implementing networking types.
mod poll_evented;
pub use self::poll_evented::PollEvented;
+1 -1
View File
@@ -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};
+1 -1
View File
@@ -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;
+3 -3
View File
@@ -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))
+1 -2
View File
@@ -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};
+2
View File
@@ -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};
+86
View File
@@ -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::<usize>() 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
}
+11 -4
View File
@@ -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;
}
+155
View File
@@ -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);
}
}
}
+7
View File
@@ -0,0 +1,7 @@
use crate::util::slab::Generation;
pub(crate) trait Entry: Default {
fn generation(&self) -> Generation;
fn reset(&self, generation: Generation) -> bool;
}
+32
View File
@@ -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
}
}
+109
View File
@@ -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<T> {
// Signal shard for now. Eventually there will be more.
shard: Shard<T>,
local: Mutex<()>,
}
unsafe impl<T: Send> Send for Slab<T> {}
unsafe impl<T: Sync> Sync for Slab<T> {}
impl<T: Entry> Slab<T> {
/// Returns a new slab with the default configuration parameters.
pub(crate) fn new() -> Slab<T> {
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<Address> {
// 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<T> fmt::Debug for Slab<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Slab")
.field("shard", &self.shard)
.finish()
}
}
@@ -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<usize>,
}
pub(crate) struct Shared {
/// Data accessed by any thread.
pub(crate) struct Shared<T> {
remote: TransferStack,
size: usize,
prev_sz: usize,
slab: CausalCell<Option<Box<[ScheduledIo]>>>,
slab: CausalCell<Option<Box<[Slot<T>]>>>,
}
/// 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<T: Entry> Shared<T> {
pub(crate) fn new(size: usize, prev_sz: usize) -> Shared<T> {
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<usize> {
pub(crate) fn alloc(&self, local: &Local) -> Option<Address> {
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<Iter<'_>> {
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<T> fmt::Debug for Shared<T> {
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));
}
}
}
+108
View File
@@ -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<T> {
/// 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<T>]>,
}
impl<T: Entry> Shard<T> {
pub(super) fn new() -> Shard<T> {
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<Address> {
// 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<T> fmt::Debug for Shard<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Shard")
.field("shared", &self.shared)
.finish()
}
}
+42
View File
@@ -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<T> {
next: CausalCell<usize>,
entry: T,
}
impl<T: Entry> Slot<T> {
/// Initialize a new `Slot` linked to `next`.
///
/// The entry is initialized to a default value.
pub(super) fn new(next: usize) -> Slot<T> {
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;
})
}
}
+58
View File
@@ -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<usize> {
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()
}
}
+327
View File
@@ -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<Slab<ScheduledIo>>, address: Address) -> Option<usize> {
slab.get(address).and_then(|s| s.get_readiness(address))
}
fn store_val(slab: &Arc<Slab<ScheduledIo>>, 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<Slab<ScheduledIo>>, 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
}
+88
View File
@@ -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();
});
}
+2
View File
@@ -0,0 +1,2 @@
mod loom_slab;
mod loom_stack;
@@ -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 {});
}
+44
View File
@@ -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());
}
-47
View File
@@ -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());
}