tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-20 00:00:08 +02:00
Code Issues Packages Projects Releases Wiki Activity

io: rewrite slab to support compaction (#2757)

Browse Source 
The I/O driver uses a slab to store per-resource state. Doing this
provides two benefits. First, allocating state is streamlined. Second,
resources may be safely indexed using a `usize` type. The `usize` is
used passed to the OS's selector when registering for receiving events.

The original slab implementation used a `Vec` backed by `RwLock`. This
primarily caused contention when reading state. This implementation also
only **grew** the slab capacity but never shrank. In #1625, the slab was
rewritten to use a lock-free strategy. The lock contention was removed
but this implementation was still grow-only.

This change adds the ability to release memory. Similar to the previous
implementation, it structures the slab to use a vector of pages. This
enables growing the slab without having to move any previous entries. It
also adds the ability to release pages. This is done by introducing a
lock when allocating/releasing slab entries. This does not impact
benchmarks, primarily due to the existing implementation not being
"done" and also having a lock around allocating and releasing.

A `Slab::compact()` function is added. Pages are iterated. When a page
is found with no slots in use, the page is freed. The `compact()`
function is called occasionally by the I/O driver.

Fixes #2505
This commit is contained in:
Carl Lerche
2020-08-11 22:28:43 -07:00
committed by GitHub
parent 674985d9fb
commit 8feebab7cd
17 changed files with 928 additions and 1344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/io/driver/mod.rs
+72 -139
View File
@@ -3,23 +3,30 @@ 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::park::{Park, Unpark};
use crate::runtime::context;
use crate::util::slab::{Address, Slab};
use crate::util::bit;
use crate::util::slab::{self, Slab};
use mio::event::Evented;
use std::fmt;
use std::io;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::{Arc, Weak};
use std::task::Waker;
use std::time::Duration;
/// I/O driver, backed by Mio
pub(crate) struct Driver {
/// Tracks the number of times `turn` is called. It is safe for this to wrap
/// as it is mostly used to determine when to call `compact()`
tick: u16,
/// Reuse the `mio::Events` value across calls to poll.
events: mio::Events,
events: Option<mio::Events>,
/// Primary slab handle containing the state for each resource registered
/// with this driver.
resources: Slab<ScheduledIo>,
/// State shared between the reactor and the handles.
inner: Arc<Inner>,
@@ -37,11 +44,8 @@ pub(super) struct Inner {
/// The underlying system event queue.
io: mio::Poll,
/// Dispatch slabs for I/O and futures events
pub(super) io_dispatch: Slab<ScheduledIo>,
/// The number of sources in `io_dispatch`.
n_sources: AtomicUsize,
/// Allocates `ScheduledIo` handles when creating new resources.
pub(super) io_dispatch: slab::Allocator<ScheduledIo>,
/// Used to wake up the reactor from a call to `turn`
wakeup: mio::SetReadiness,
@@ -53,7 +57,19 @@ pub(super) enum Direction {
Write,
}
const TOKEN_WAKEUP: mio::Token = mio::Token(Address::NULL);
// TODO: Don't use a fake token. Instead, reserve a slot entry for the wakeup
// token.
const TOKEN_WAKEUP: mio::Token = mio::Token(1 << 31);
const ADDRESS: bit::Pack = bit::Pack::least_significant(24);
// Packs the generation value in the `readiness` field.
//
// The generation prevents a race condition where a slab slot is reused for a
// new socket while the I/O driver is about to apply a readiness event. The
// generaton value is checked when setting new readiness. If the generation do
// not match, then the readiness event is discarded.
const GENERATION: bit::Pack = ADDRESS.then(7);
fn _assert_kinds() {
fn _assert<T: Send + Sync>() {}
@@ -69,6 +85,8 @@ impl Driver {
pub(crate) fn new() -> io::Result<Driver> {
let io = mio::Poll::new()?;
let wakeup_pair = mio::Registration::new2();
let slab = Slab::new();
let allocator = slab.allocator();
io.register(
&wakeup_pair.0,
@@ -78,12 +96,13 @@ impl Driver {
)?;
Ok(Driver {
events: mio::Events::with_capacity(1024),
tick: 0,
events: Some(mio::Events::with_capacity(1024)),
resources: slab,
_wakeup_registration: wakeup_pair.0,
inner: Arc::new(Inner {
io,
io_dispatch: Slab::new(),
n_sources: AtomicUsize::new(0),
io_dispatch: allocator,
wakeup: wakeup_pair.1,
}),
})
@@ -102,16 +121,27 @@ impl Driver {
}
fn turn(&mut self, max_wait: Option<Duration>) -> io::Result<()> {
// How often to call `compact()` on the resource slab
const COMPACT_INTERVAL: u16 = 256;
self.tick = self.tick.wrapping_add(1);
if self.tick % COMPACT_INTERVAL == 0 {
self.resources.compact();
}
let mut events = self.events.take().expect("i/o driver event store missing");
// Block waiting for an event to happen, peeling out how many events
// happened.
match self.inner.io.poll(&mut self.events, max_wait) {
match self.inner.io.poll(&mut events, max_wait) {
Ok(_) => {}
Err(e) => return Err(e),
}
// Process all the events that came in, dispatching appropriately
for event in self.events.iter() {
for event in events.iter() {
let token = event.token();
if token == TOKEN_WAKEUP {
@@ -124,22 +154,24 @@ impl Driver {
}
}
self.events = Some(events);
Ok(())
}
fn dispatch(&self, token: mio::Token, ready: mio::Ready) {
fn dispatch(&mut self, token: mio::Token, ready: mio::Ready) {
let mut rd = None;
let mut wr = None;
let address = Address::from_usize(token.0);
let addr = slab::Address::from_usize(ADDRESS.unpack(token.0));
let io = match self.inner.io_dispatch.get(address) {
let io = match self.resources.get(addr) {
Some(io) => io,
None => return,
};
if io
.set_readiness(address, |curr| curr | ready.as_usize())
.set_readiness(Some(token.0), |curr| curr | ready.as_usize())
.is_err()
{
// token no longer valid!
@@ -164,6 +196,18 @@ impl Driver {
}
}
impl Drop for Driver {
fn drop(&mut self) {
self.resources.for_each(|io| {
// If a task is waiting on the I/O resource, notify it. The task
// will then attempt to use the I/O resource and fail due to the
// driver being shutdown.
io.reader.wake();
io.writer.wake();
})
}
}
impl Park for Driver {
type Unpark = Handle;
type Error = io::Error;
@@ -246,24 +290,20 @@ impl Inner {
&self,
source: &dyn Evented,
ready: mio::Ready,
) -> io::Result<Address> {
let address = self.io_dispatch.alloc().ok_or_else(|| {
) -> io::Result<slab::Ref<ScheduledIo>> {
let (address, shared) = self.io_dispatch.allocate().ok_or_else(|| {
io::Error::new(
io::ErrorKind::Other,
"reactor at max registered I/O resources",
)
})?;
self.n_sources.fetch_add(1, SeqCst);
let token = GENERATION.pack(shared.generation(), ADDRESS.pack(address.as_usize(), 0));
self.io.register(
source,
mio::Token(address.to_usize()),
ready,
mio::PollOpt::edge(),
)?;
self.io
.register(source, mio::Token(token), ready, mio::PollOpt::edge())?;
Ok(address)
Ok(shared)
}
/// Deregisters an I/O resource from the reactor.
@@ -271,21 +311,11 @@ impl Inner {
self.io.deregister(source)
}
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: Address, dir: Direction, w: Waker) {
let sched = self
.io_dispatch
.get(token)
.unwrap_or_else(|| panic!("IO resource for token {:?} does not exist!", token));
pub(super) fn register(&self, io: &slab::Ref<ScheduledIo>, dir: Direction, w: Waker) {
let waker = match dir {
Direction::Read => &sched.reader,
Direction::Write => &sched.writer,
Direction::Read => &io.reader,
Direction::Write => &io.writer,
};
waker.register(w);
@@ -303,100 +333,3 @@ impl Direction {
}
}
}
#[cfg(all(test, loom))]
mod tests {
use super::*;
use loom::thread;
// No-op `Evented` impl just so we can have something to pass to `add_source`.
struct NotEvented;
impl Evented for NotEvented {
fn register(
&self,
_: &mio::Poll,
_: mio::Token,
_: mio::Ready,
_: mio::PollOpt,
) -> io::Result<()> {
Ok(())
}
fn reregister(
&self,
_: &mio::Poll,
_: mio::Token,
_: mio::Ready,
_: mio::PollOpt,
) -> io::Result<()> {
Ok(())
}
fn deregister(&self, _: &mio::Poll) -> io::Result<()> {
Ok(())
}
}
#[test]
fn tokens_unique_when_dropped() {
loom::model(|| {
let reactor = Driver::new().unwrap();
let inner = reactor.inner;
let inner2 = inner.clone();
let token_1 = inner.add_source(&NotEvented, mio::Ready::all()).unwrap();
let thread = thread::spawn(move || {
inner2.drop_source(token_1);
});
let token_2 = inner.add_source(&NotEvented, mio::Ready::all()).unwrap();
thread.join().unwrap();
assert!(token_1 != token_2);
})
}
#[test]
fn tokens_unique_when_dropped_on_full_page() {
loom::model(|| {
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
// may be reused.
for _ in 0..31 {
inner.add_source(&NotEvented, mio::Ready::all()).unwrap();
}
let token_1 = inner.add_source(&NotEvented, mio::Ready::all()).unwrap();
let thread = thread::spawn(move || {
inner2.drop_source(token_1);
});
let token_2 = inner.add_source(&NotEvented, mio::Ready::all()).unwrap();
thread.join().unwrap();
assert!(token_1 != token_2);
})
}
#[test]
fn tokens_unique_concurrent_add() {
loom::model(|| {
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, mio::Ready::all()).unwrap();
token_2
});
let token_1 = inner.add_source(&NotEvented, mio::Ready::all()).unwrap();
let token_2 = thread.join().unwrap();
assert!(token_1 != token_2);
})
}
}
tokio/src/io/driver/scheduled_io.rs
+27 -61
View File
@@ -1,47 +1,30 @@
use crate::loom::future::AtomicWaker;
use crate::loom::sync::atomic::AtomicUsize;
use crate::util::bit;
use crate::util::slab::{Address, Entry, Generation};
use crate::util::slab::Entry;
use std::sync::atomic::Ordering::{AcqRel, Acquire, SeqCst};
use std::sync::atomic::Ordering::{AcqRel, Acquire, Release};
/// Stored in the I/O driver resource slab.
#[derive(Debug)]
pub(crate) struct ScheduledIo {
/// Packs the resource's readiness with the resource's generation.
readiness: AtomicUsize,
/// Task waiting on read readiness
pub(crate) reader: AtomicWaker,
/// Task waiting on write readiness
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) {
let state = self.readiness.load(Acquire);
fn reset(&self, generation: Generation) -> bool {
let mut current = self.readiness.load(Acquire);
let generation = super::GENERATION.unpack(state);
let next = super::GENERATION.pack_lossy(generation + 1, 0);
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
self.readiness.store(next, Release);
}
}
@@ -56,24 +39,8 @@ impl Default for ScheduledIo {
}
impl ScheduledIo {
#[cfg(all(test, loom))]
/// 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())
pub(crate) fn generation(&self) -> usize {
super::GENERATION.unpack(self.readiness.load(Acquire))
}
/// Sets the readiness on this `ScheduledIo` by invoking the given closure on
@@ -92,32 +59,35 @@ impl ScheduledIo {
/// Otherwise, this returns the previous readiness.
pub(crate) fn set_readiness(
&self,
address: Address,
token: Option<usize>,
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(());
let current_generation = super::GENERATION.unpack(current);
if let Some(token) = token {
// Check that the generation for this access is still the
// current one.
if super::GENERATION.unpack(token) != 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 <= super::ADDRESS.max_value(),
"new readiness value would overwrite generation bits!"
);
match self.readiness.compare_exchange(
current,
PACK.pack(generation.to_usize(), new),
super::GENERATION.pack(current_generation, new),
AcqRel,
Acquire,
) {
@@ -135,7 +105,3 @@ impl Drop for ScheduledIo {
self.reader.wake();
}
}
fn unpack_generation(src: usize) -> Generation {
Generation::new(PACK.unpack(src))
}
tokio/src/io/registration.rs
+24 -24
View File
@@ -1,5 +1,5 @@
use crate::io::driver::{platform, Direction, Handle};
use crate::util::slab::Address;
use crate::io::driver::{platform, Direction, Handle, ScheduledIo};
use crate::util::slab;
use mio::{self, Evented};
use std::io;
@@ -39,11 +39,17 @@ cfg_io_driver! {
/// [`poll_write_ready`]: method@Self::poll_write_ready`
#[derive(Debug)]
pub struct Registration {
/// Handle to the associated driver.
handle: Handle,
address: Address,
/// Reference to state stored by the driver.
shared: slab::Ref<ScheduledIo>,
}
}
unsafe impl Send for Registration {}
unsafe impl Sync for Registration {}
// ===== impl Registration =====
impl Registration {
@@ -104,7 +110,7 @@ impl Registration {
T: Evented,
{
let handle = Handle::current();
let address = if let Some(inner) = handle.inner() {
let shared = if let Some(inner) = handle.inner() {
inner.add_source(io, ready)?
} else {
return Err(io::Error::new(
@@ -113,7 +119,7 @@ impl Registration {
));
};
Ok(Registration { handle, address })
Ok(Registration { handle, shared })
}
/// Deregisters the I/O resource from the reactor it is associated with.
@@ -272,14 +278,12 @@ 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.address, direction, cx.waker().clone())
inner.register(&self.shared, direction, cx.waker().clone())
}
let mask = direction.mask();
let mask_no_hup = (mask - platform::hup() - platform::error()).as_usize();
let sched = inner.io_dispatch.get(self.address).unwrap();
// This consumes the current readiness state **except** for HUP and
// error. HUP and error are excluded because a) they are final states
// and never transitition out and b) both the read AND the write
@@ -296,9 +300,10 @@ impl Registration {
// AND write. A specific case that `EPOLLERR` occurs is when the read
// end of a pipe is closed. When this occurs, a peer blocked by
// writing to the pipe should be notified.
let curr_ready = sched
.set_readiness(self.address, |curr| curr & (!mask_no_hup))
.unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address));
let curr_ready = self
.shared
.set_readiness(None, |curr| curr & (!mask_no_hup))
.unwrap_or_else(|_| unreachable!());
let mut ready = mask & mio::Ready::from_usize(curr_ready);
@@ -306,14 +311,15 @@ impl Registration {
if let Some(cx) = cx {
// Update the task info
match direction {
Direction::Read => sched.reader.register_by_ref(cx.waker()),
Direction::Write => sched.writer.register_by_ref(cx.waker()),
Direction::Read => self.shared.reader.register_by_ref(cx.waker()),
Direction::Write => self.shared.writer.register_by_ref(cx.waker()),
}
// Try again
let curr_ready = sched
.set_readiness(self.address, |curr| curr & (!mask_no_hup))
.unwrap_or_else(|_| panic!("address {:?} no longer valid!", self.address));
let curr_ready = self
.shared
.set_readiness(None, |curr| curr & (!mask_no_hup))
.unwrap();
ready = mask & mio::Ready::from_usize(curr_ready);
}
}
@@ -326,15 +332,9 @@ impl Registration {
}
}
unsafe impl Send for Registration {}
unsafe impl Sync for Registration {}
impl Drop for Registration {
fn drop(&mut self) {
let inner = match self.handle.inner() {
Some(inner) => inner,
None => return,
};
inner.drop_source(self.address);
drop(self.shared.reader.take_waker());
drop(self.shared.writer.take_waker());
}
}
tokio/src/loom/std/atomic_ptr.rs
+7 -1
View File
@@ -1,5 +1,5 @@
use std::fmt;
use std::ops::Deref;
use std::ops::{Deref, DerefMut};
/// `AtomicPtr` providing an additional `load_unsync` function.
pub(crate) struct AtomicPtr<T> {
@@ -21,6 +21,12 @@ impl<T> Deref for AtomicPtr<T> {
}
}
impl<T> DerefMut for AtomicPtr<T> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.inner
}
}
impl<T> fmt::Debug for AtomicPtr<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
self.deref().fmt(fmt)
tokio/src/util/bit.rs
+8 -10
View File
@@ -7,16 +7,6 @@ pub(crate) struct Pack {
}
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);
@@ -53,6 +43,14 @@ impl Pack {
(base & !self.mask) | (value << self.shift)
}
/// Packs the value with `base`, losing any bits of `value` that fit.
///
/// If `value` is larger than the max value that can be represented by the
/// allotted width, the most significant bits are truncated.
pub(crate) fn pack_lossy(&self, value: usize, base: usize) -> usize {
self.pack(value & self.max_value(), base)
}
pub(crate) fn unpack(&self, src: usize) -> usize {
unpack(src, self.mask, self.shift)
}
tokio/src/util/slab.rs
+790
View File
@@ -0,0 +1,790 @@
use crate::loom::cell::UnsafeCell;
use crate::loom::sync::atomic::{AtomicBool, AtomicUsize};
use crate::loom::sync::{Arc, Mutex};
use crate::util::bit;
use std::fmt;
use std::mem;
use std::ops;
use std::ptr;
use std::sync::atomic::Ordering::Relaxed;
/// Amortized allocation for homogeneous data types.
///
/// The slab pre-allocates chunks of memory to store values. It uses a similar
/// growing strategy as `Vec`. When new capacity is needed, the slab grows by
/// 2x.
///
/// # Pages
///
/// Unlike `Vec`, growing does not require moving existing elements. Instead of
/// being a continuous chunk of memory for all elements, `Slab` is an array of
/// arrays. The top-level array is an array of pages. Each page is 2x bigger
/// than the previous one. When the slab grows, a new page is allocated.
///
/// Pages are lazily initialized.
///
/// # Allocating
///
/// When allocating an object, first previously used slots are reused. If no
/// previously used slot is available, a new slot is initialized in an existing
/// page. If all pages are full, then a new page is allocated.
///
/// When an allocated object is released, it is pushed into it's page's free
/// list. Allocating scans all pages for a free slot.
///
/// # Indexing
///
/// The slab is able to index values using an address. Even when the indexed
/// object has been released, it is still safe to index. This is a key ability
/// for using the slab with the I/O driver. Addresses are registered with the
/// OS's selector and I/O resources can be released without synchronizing with
/// the OS.
///
/// # Compaction
///
/// `Slab::compact` will release pages that have been allocated but are no
/// longer used. This is done by scanning the pages and finding pages with no
/// allocated objects. These pages are then freed.
///
/// # Synchronization
///
/// The `Slab` structure is able to provide (mostly) unsynchronized reads to
/// values stored in the slab. Insertions and removals are synchronized. Reading
/// objects via `Ref` is fully unsynchronized. Indexing objects uses amortized
/// synchronization.
///
pub(crate) struct Slab<T> {
/// Array of pages. Each page is synchronized.
pages: [Arc<Page<T>>; NUM_PAGES],
/// Caches the array pointer & number of initialized slots.
cached: [CachedPage<T>; NUM_PAGES],
}
/// Allocate values in the associated slab.
pub(crate) struct Allocator<T> {
/// Pages in the slab. The first page has a capacity of 16 elements. Each
/// following page has double the capacity of the previous page.
///
/// Each returned `Ref` holds a reference count to this `Arc`.
pages: [Arc<Page<T>>; NUM_PAGES],
}
/// References a slot in the slab. Indexing a slot using an `Address` is memory
/// safe even if the slot has been released or the page has been deallocated.
/// However, it is not guaranteed that the slot has not been reused and is now
/// represents a different value.
///
/// The I/O driver uses a counter to track the slot's generation. Once accessing
/// the slot, the generations are compared. If they match, the value matches the
/// address.
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
pub(crate) struct Address(usize);
/// An entry in the slab.
pub(crate) trait Entry: Default {
/// Reset the entry's value and track the generation.
fn reset(&self);
}
/// A reference to a value stored in the slab
pub(crate) struct Ref<T> {
value: *const Value<T>,
}
/// Maximum number of pages a slab can contain.
const NUM_PAGES: usize = 19;
/// Minimum number of slots a page can contain.
const PAGE_INITIAL_SIZE: usize = 32;
const PAGE_INDEX_SHIFT: u32 = PAGE_INITIAL_SIZE.trailing_zeros() + 1;
/// A page in the slab
struct Page<T> {
/// Slots
slots: Mutex<Slots<T>>,
// Number of slots currently being used. This is not guaranteed to be up to
// date and should only be used as a hint.
used: AtomicUsize,
// Set to `true` when the page has been allocated.
allocated: AtomicBool,
// The number of slots the page can hold.
len: usize,
// Length of all previous pages combined
prev_len: usize,
}
struct CachedPage<T> {
/// Pointer to the page's slots.
slots: *const Slot<T>,
/// Number of initialized slots.
init: usize,
}
/// Page state
struct Slots<T> {
/// Slots
slots: Vec<Slot<T>>,
head: usize,
/// Number of slots currently in use.
used: usize,
}
unsafe impl<T: Sync> Sync for Page<T> {}
unsafe impl<T: Sync> Send for Page<T> {}
unsafe impl<T: Sync> Sync for CachedPage<T> {}
unsafe impl<T: Sync> Send for CachedPage<T> {}
/// A slot in the slab. Contains slot-specific metadata.
///
/// `#[repr(C)]` guarantees that the struct starts w/ `value`. We use pointer
/// math to map a value pointer to an index in the page.
#[repr(C)]
struct Slot<T> {
/// Pointed to by `Ref`.
value: UnsafeCell<Value<T>>,
/// Next entry in the free list.
next: u32,
}
/// Value paired with a reference to the page
struct Value<T> {
/// Value stored in the value
value: T,
/// Pointer to the page containing the slot.
///
/// A raw pointer is used as this creates a ref cycle.
page: *const Page<T>,
}
impl<T> Slab<T> {
/// Create a new, empty, slab
pub(crate) fn new() -> Slab<T> {
// Initializing arrays is a bit annoying. Instead of manually writing
// out an array and every single entry, `Default::default()` is used to
// initialize the array, then the array is iterated and each value is
// initialized.
let mut slab = Slab {
pages: Default::default(),
cached: Default::default(),
};
let mut len = PAGE_INITIAL_SIZE;
let mut prev_len: usize = 0;
for page in &mut slab.pages {
let page = Arc::get_mut(page).unwrap();
page.len = len;
page.prev_len = prev_len;
len *= 2;
prev_len += page.len;
// Ensure we don't exceed the max address space.
debug_assert!(
page.len - 1 + page.prev_len < (1 << 24),
"max = {:b}",
page.len - 1 + page.prev_len
);
}
slab
}
/// Returns a new `Allocator`.
///
/// The `Allocator` supports concurrent allocation of objects.
pub(crate) fn allocator(&self) -> Allocator<T> {
Allocator {
pages: self.pages.clone(),
}
}
/// Returns a reference to the value stored at the given address.
///
/// `&mut self` is used as the call may update internal cached state.
pub(crate) fn get(&mut self, addr: Address) -> Option<&T> {
let page_idx = addr.page();
let slot_idx = self.pages[page_idx].slot(addr);
// If the address references a slot that was last seen as uninitialized,
// the `CachedPage` is updated. This requires acquiring the page lock
// and updating the slot pointer and initialized offset.
if self.cached[page_idx].init <= slot_idx {
self.cached[page_idx].refresh(&self.pages[page_idx]);
}
// If the address **still** references an uninitialized slot, then the
// address is invalid and `None` is returned.
if self.cached[page_idx].init <= slot_idx {
return None;
}
// Get a reference to the value. The lifetime of the returned reference
// is bound to `&self`. The only way to invalidate the underlying memory
// is to call `compact()`. The lifetimes prevent calling `compact()`
// while references to values are outstanding.
//
// The referenced data is never mutated. Only `&self` references are
// used and the data is `Sync`.
Some(self.cached[page_idx].get(slot_idx))
}
/// Calls the given function with a reference to each slot in the slab. The
/// slot may not be in-use.
///
/// This is used by the I/O driver during the shutdown process to notify
/// each pending task.
pub(crate) fn for_each(&mut self, mut f: impl FnMut(&T)) {
for page_idx in 0..self.pages.len() {
// It is required to avoid holding the lock when calling the
// provided function. The function may attempt to acquire the lock
// itself. If we hold the lock here while calling `f`, a deadlock
// situation is possible.
//
// Instead of iterating the slots directly in `page`, which would
// require holding the lock, the cache is updated and the slots are
// iterated from the cache.
self.cached[page_idx].refresh(&self.pages[page_idx]);
for slot_idx in 0..self.cached[page_idx].init {
f(self.cached[page_idx].get(slot_idx));
}
}
}
// Release memory back to the allocator.
//
// If pages are empty, the underlying memory is released back to the
// allocator.
pub(crate) fn compact(&mut self) {
// Iterate each page except the very first one. The very first page is
// never freed.
for (idx, page) in (&self.pages[1..]).iter().enumerate() {
if page.used.load(Relaxed) != 0 || !page.allocated.load(Relaxed) {
// If the page has slots in use or the memory has not been
// allocated then it cannot be compacted.
continue;
}
let mut slots = match page.slots.try_lock() {
Ok(slots) => slots,
// If the lock cannot be acquired due to being held by another
// thread, don't try to compact the page.
_ => continue,
};
if slots.used > 0 || slots.slots.capacity() == 0 {
// The page is in use or it has not yet been allocated. Either
// way, there is no more work to do.
continue;
}
page.allocated.store(false, Relaxed);
// Remove the slots vector from the page. This is done so that the
// freeing process is done outside of the lock's critical section.
let vec = mem::replace(&mut slots.slots, vec![]);
slots.head = 0;
// Drop the lock so we can drop the vector outside the lock below.
drop(slots);
// Clear cache
self.cached[idx].slots = ptr::null();
self.cached[idx].init = 0;
drop(vec);
}
}
}
impl<T> fmt::Debug for Slab<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
debug(fmt, "Slab", &self.pages[..])
}
}
impl<T: Entry> Allocator<T> {
/// Allocate a new entry and return a handle to the entry.
///
/// Scans pages from smallest to biggest, stopping when a slot is found.
/// Pages are allocated if necessary.
///
/// Returns `None` if the slab is full.
pub(crate) fn allocate(&self) -> Option<(Address, Ref<T>)> {
// Find the first available slot.
for page in &self.pages[..] {
if let Some((addr, val)) = Page::allocate(page) {
return Some((addr, val));
}
}
None
}
}
impl<T> fmt::Debug for Allocator<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
debug(fmt, "slab::Allocator", &self.pages[..])
}
}
impl<T> ops::Deref for Ref<T> {
type Target = T;
fn deref(&self) -> &T {
// Safety: `&mut` is never handed out to the underlying value. The page
// is not freed until all `Ref` values are dropped.
unsafe { &(*self.value).value }
}
}
impl<T> Drop for Ref<T> {
fn drop(&mut self) {
// Safety: `&mut` is never handed out to the underlying value. The page
// is not freed until all `Ref` values are dropped.
let _ = unsafe { (*self.value).release() };
}
}
impl<T: fmt::Debug> fmt::Debug for Ref<T> {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
(**self).fmt(fmt)
}
}
impl<T: Entry> Page<T> {
// Allocates an object, returns the ref and address.
//
// `self: &Arc<Page<T>>` is avoided here as this would not work with the
// loom `Arc`.
fn allocate(me: &Arc<Page<T>>) -> Option<(Address, Ref<T>)> {
// Before acquiring the lock, use the `used` hint.
if me.used.load(Relaxed) == me.len {
return None;
}
// Allocating objects requires synchronization
let mut locked = me.slots.lock().unwrap();
if locked.head < locked.slots.len() {
// Re-use an already initialized slot.
//
// Help out the borrow checker
let locked = &mut *locked;
// Get the index of the slot at the head of the free stack. This is
// the slot that will be reused.
let idx = locked.head;
let slot = &locked.slots[idx];
// Update the free stack head to point to the next slot.
locked.head = slot.next as usize;
// Increment the number of used slots
locked.used += 1;
me.used.store(locked.used, Relaxed);
// Reset the slot
slot.value.with(|ptr| unsafe { (*ptr).value.reset() });
// Return a reference to the slot
Some((me.addr(idx), slot.gen_ref(me)))
} else if me.len == locked.slots.len() {
// The page is full
None
} else {
// No initialized slots are available, but the page has more
// capacity. Initialize a new slot.
let idx = locked.slots.len();
if idx == 0 {
// The page has not yet been allocated. Allocate the storage for
// all page slots.
locked.slots.reserve_exact(me.len);
}
// Initialize a new slot
locked.slots.push(Slot {
value: UnsafeCell::new(Value {
value: Default::default(),
page: &**me as *const _,
}),
next: 0,
});
// Increment the head to indicate the free stack is empty
locked.head += 1;
// Increment the number of used slots
locked.used += 1;
me.used.store(locked.used, Relaxed);
me.allocated.store(true, Relaxed);
debug_assert_eq!(locked.slots.len(), locked.head);
Some((me.addr(idx), locked.slots[idx].gen_ref(me)))
}
}
}
impl<T> Page<T> {
/// Returns the slot index within the current page referenced by the given
/// address.
fn slot(&self, addr: Address) -> usize {
addr.0 - self.prev_len
}
/// Returns the address for the given slot
fn addr(&self, slot: usize) -> Address {
Address(slot + self.prev_len)
}
}
impl<T> Default for Page<T> {
fn default() -> Page<T> {
Page {
used: AtomicUsize::new(0),
allocated: AtomicBool::new(false),
slots: Mutex::new(Slots {
slots: Vec::new(),
head: 0,
used: 0,
}),
len: 0,
prev_len: 0,
}
}
}
impl<T> Page<T> {
/// Release a slot into the page's free list
fn release(&self, value: *const Value<T>) {
let mut locked = self.slots.lock().unwrap();
let idx = locked.index_for(value);
locked.slots[idx].next = locked.head as u32;
locked.head = idx;
locked.used -= 1;
self.used.store(locked.used, Relaxed);
}
}
impl<T> CachedPage<T> {
/// Refresh the cache
fn refresh(&mut self, page: &Page<T>) {
let slots = page.slots.lock().unwrap();
self.slots = slots.slots.as_ptr();
self.init = slots.slots.len();
}
// Get a value by index
fn get(&self, idx: usize) -> &T {
assert!(idx < self.init);
// Safety: Pages are allocated concurrently, but are only ever
// **deallocated** by `Slab`. `Slab` will always have a more
// conservative view on the state of the slot array. Once `CachedPage`
// sees a slot pointer and initialized offset, it will remain valid
// until `compact()` is called. The `compact()` function also updates
// `CachedPage`.
unsafe {
let slot = self.slots.add(idx);
let value = slot as *const Value<T>;
&(*value).value
}
}
}
impl<T> Default for CachedPage<T> {
fn default() -> CachedPage<T> {
CachedPage {
slots: ptr::null(),
init: 0,
}
}
}
impl<T> Slots<T> {
/// Maps a slot pointer to an offset within the current page.
///
/// The pointer math removes the `usize` index from the `Ref` struct,
/// shrinking the struct to a single pointer size. The contents of the
/// function is safe, the resulting `usize` is bounds checked before being
/// used.
///
/// # Panics
///
/// panics if the provided slot pointer is not contained by the page.
fn index_for(&self, slot: *const Value<T>) -> usize {
use std::mem;
let base = &self.slots[0] as *const _ as usize;
assert!(base != 0, "page is unallocated");
let slot = slot as usize;
let width = mem::size_of::<Slot<T>>();
assert!(slot >= base, "unexpected pointer");
let idx = (slot - base) / width;
assert!(idx < self.slots.len() as usize);
idx
}
}
impl<T: Entry> Slot<T> {
/// Generates a `Ref` for the slot. This involves bumping the page's ref count.
fn gen_ref(&self, page: &Arc<Page<T>>) -> Ref<T> {
// The ref holds a ref on the page. The `Arc` is forgotten here and is
// resurrected in `release` when the `Ref` is dropped. By avoiding to
// hold on to an explicit `Arc` value, the struct size of `Ref` is
// reduced.
mem::forget(page.clone());
let slot = self as *const Slot<T>;
let value = slot as *const Value<T>;
Ref { value }
}
}
impl<T> Value<T> {
// Release the slot, returning the `Arc<Page<T>>` logically owned by the ref.
fn release(&self) -> Arc<Page<T>> {
// Safety: called by `Ref`, which owns an `Arc<Page<T>>` instance.
let page = unsafe { Arc::from_raw(self.page) };
page.release(self as *const _);
page
}
}
impl Address {
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.0 + PAGE_INITIAL_SIZE) >> PAGE_INDEX_SHIFT;
(bit::pointer_width() - slot_shifted.leading_zeros()) as usize
}
pub(crate) const fn as_usize(self) -> usize {
self.0
}
pub(crate) fn from_usize(src: usize) -> Address {
Address(src)
}
}
fn debug<T>(fmt: &mut fmt::Formatter<'_>, name: &str, pages: &[Arc<Page<T>>]) -> fmt::Result {
let mut capacity = 0;
let mut len = 0;
for page in pages {
if page.allocated.load(Relaxed) {
capacity += page.len;
len += page.used.load(Relaxed);
}
}
fmt.debug_struct(name)
.field("len", &len)
.field("capacity", &capacity)
.finish()
}
#[cfg(all(test, not(loom)))]
mod test {
use super::*;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering::SeqCst;
struct Foo {
cnt: AtomicUsize,
id: AtomicUsize,
}
impl Default for Foo {
fn default() -> Foo {
Foo {
cnt: AtomicUsize::new(0),
id: AtomicUsize::new(0),
}
}
}
impl Entry for Foo {
fn reset(&self) {
self.cnt.fetch_add(1, SeqCst);
}
}
#[test]
fn insert_remove() {
let mut slab = Slab::<Foo>::new();
let alloc = slab.allocator();
let (addr1, foo1) = alloc.allocate().unwrap();
foo1.id.store(1, SeqCst);
assert_eq!(0, foo1.cnt.load(SeqCst));
let (addr2, foo2) = alloc.allocate().unwrap();
foo2.id.store(2, SeqCst);
assert_eq!(0, foo2.cnt.load(SeqCst));
assert_eq!(1, slab.get(addr1).unwrap().id.load(SeqCst));
assert_eq!(2, slab.get(addr2).unwrap().id.load(SeqCst));
drop(foo1);
assert_eq!(1, slab.get(addr1).unwrap().id.load(SeqCst));
let (addr3, foo3) = alloc.allocate().unwrap();
assert_eq!(addr3, addr1);
assert_eq!(1, foo3.cnt.load(SeqCst));
foo3.id.store(3, SeqCst);
assert_eq!(3, slab.get(addr3).unwrap().id.load(SeqCst));
drop(foo2);
drop(foo3);
slab.compact();
// The first page is never released
assert!(slab.get(addr1).is_some());
assert!(slab.get(addr2).is_some());
assert!(slab.get(addr3).is_some());
}
#[test]
fn insert_many() {
let mut slab = Slab::<Foo>::new();
let alloc = slab.allocator();
let mut entries = vec![];
for i in 0..10_000 {
let (addr, val) = alloc.allocate().unwrap();
val.id.store(i, SeqCst);
entries.push((addr, val));
}
for (i, (addr, v)) in entries.iter().enumerate() {
assert_eq!(i, v.id.load(SeqCst));
assert_eq!(i, slab.get(*addr).unwrap().id.load(SeqCst));
}
entries.clear();
for i in 0..10_000 {
let (addr, val) = alloc.allocate().unwrap();
val.id.store(10_000 - i, SeqCst);
entries.push((addr, val));
}
for (i, (addr, v)) in entries.iter().enumerate() {
assert_eq!(10_000 - i, v.id.load(SeqCst));
assert_eq!(10_000 - i, slab.get(*addr).unwrap().id.load(SeqCst));
}
}
#[test]
fn insert_drop_reverse() {
let mut slab = Slab::<Foo>::new();
let alloc = slab.allocator();
let mut entries = vec![];
for i in 0..10_000 {
let (addr, val) = alloc.allocate().unwrap();
val.id.store(i, SeqCst);
entries.push((addr, val));
}
for _ in 0..10 {
// Drop 1000 in reverse
for _ in 0..1_000 {
entries.pop();
}
// Check remaining
for (i, (addr, v)) in entries.iter().enumerate() {
assert_eq!(i, v.id.load(SeqCst));
assert_eq!(i, slab.get(*addr).unwrap().id.load(SeqCst));
}
}
}
#[test]
fn no_compaction_if_page_still_in_use() {
let mut slab = Slab::<Foo>::new();
let alloc = slab.allocator();
let mut entries1 = vec![];
let mut entries2 = vec![];
for i in 0..10_000 {
let (addr, val) = alloc.allocate().unwrap();
val.id.store(i, SeqCst);
if i % 2 == 0 {
entries1.push((addr, val, i));
} else {
entries2.push(val);
}
}
drop(entries2);
for (addr, _, i) in &entries1 {
assert_eq!(*i, slab.get(*addr).unwrap().id.load(SeqCst));
}
}
#[test]
fn compact_all() {
let mut slab = Slab::<Foo>::new();
let alloc = slab.allocator();
let mut entries = vec![];
for _ in 0..2 {
entries.clear();
for i in 0..10_000 {
let (addr, val) = alloc.allocate().unwrap();
val.id.store(i, SeqCst);
entries.push((addr, val));
}
let mut addrs = vec![];
for (addr, _) in entries.drain(..) {
addrs.push(addr);
}
slab.compact();
// The first page is never freed
for addr in &addrs[PAGE_INITIAL_SIZE..] {
assert!(slab.get(*addr).is_none());
}
}
}
}
tokio/src/util/slab/addr.rs
-154
View File
@@ -1,154 +0,0 @@
//! 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, INITIAL_PAGE_SIZE, MAX_PAGES, MAX_THREADS};
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);
}
}
}
tokio/src/util/slab/entry.rs
-7
View File
@@ -1,7 +0,0 @@
use crate::util::slab::Generation;
pub(crate) trait Entry: Default {
fn generation(&self) -> Generation;
fn reset(&self, generation: Generation) -> bool;
}
tokio/src/util/slab/generation.rs
-32
View File
@@ -1,32 +0,0 @@
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
}
}
tokio/src/util/slab/mod.rs
-107
View File
@@ -1,107 +0,0 @@
//! 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()
}
}
tokio/src/util/slab/page.rs
-187
View File
@@ -1,187 +0,0 @@
use crate::loom::cell::UnsafeCell;
use crate::util::slab::{Address, Entry, Slot, TransferStack, INITIAL_PAGE_SIZE};
use std::fmt;
/// Data accessed only by the thread that owns the shard.
pub(crate) struct Local {
head: UnsafeCell<usize>,
}
/// Data accessed by any thread.
pub(crate) struct Shared<T> {
remote: TransferStack,
size: usize,
prev_sz: usize,
slab: UnsafeCell<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 {
pub(crate) fn new() -> Self {
Self {
head: UnsafeCell::new(0),
}
}
fn head(&self) -> usize {
self.head.with(|head| unsafe { *head })
}
fn set_head(&self, new_head: usize) {
self.head.with_mut(|head| unsafe {
*head = new_head;
})
}
}
impl<T: Entry> Shared<T> {
pub(crate) fn new(size: usize, prev_sz: usize) -> Shared<T> {
Self {
prev_sz,
size,
remote: TransferStack::new(),
slab: UnsafeCell::new(None),
}
}
/// Allocates storage for this page if it does not allready exist.
///
/// This requires unique access to the page (e.g. it is called from the
/// thread that owns the page, or, in the case of `SingleShard`, while the
/// lock is held). In order to indicate this, a reference to the page's
/// `Local` data is taken by this function; the `Local` argument is not
/// actually used, but requiring it ensures that this is only called when
/// local access is held.
#[cold]
fn alloc_page(&self, _: &Local) {
debug_assert!(self.slab.with(|s| unsafe { (*s).is_none() }));
let mut slab = Vec::with_capacity(self.size);
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
// page has not yet been allocated, other threads will not try
// to access it yet.
unsafe {
*s = Some(slab.into_boxed_slice());
}
});
}
pub(crate) fn alloc(&self, local: &Local) -> Option<Address> {
let head = local.head();
// are there any items on the local free list? (fast path)
let head = if head < self.size {
head
} else {
// if the local free list is empty, pop all the items on the remote
// free list onto the local free list.
self.remote.pop_all()?
};
// 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 == Address::NULL {
return None;
}
// do we need to allocate storage for this page?
let page_needs_alloc = self.slab.with(|s| unsafe { (*s).is_none() });
if page_needs_alloc {
self.alloc_page(local);
}
let gen = self.slab.with(|slab| {
let slab = unsafe { &*(slab) }
.as_ref()
.expect("page must have been allocated to alloc!");
let slot = &slab[head];
local.set_head(slot.next());
slot.generation()
});
let index = head + self.prev_sz;
Some(Address::new(index, gen))
}
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: 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(addr.generation()) {
slot.set_next(local.head());
local.set_head(offset);
}
})
}
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(addr.generation()) {
return;
}
self.remote.push(offset, |next| slot.set_next(next));
})
}
}
impl fmt::Debug for Local {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.head.with(|head| {
let head = unsafe { *head };
f.debug_struct("Local")
.field("head", &format_args!("{:#0x}", head))
.finish()
})
}
}
impl<T> fmt::Debug for Shared<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("Shared")
.field("remote", &self.remote)
.field("prev_sz", &self.prev_sz)
.field("size", &self.size)
// .field("slab", &self.slab)
.finish()
}
}
tokio/src/util/slab/shard.rs
-105
View File
@@ -1,105 +0,0 @@
use crate::util::slab::{page, Address, Entry, 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()
}
}
tokio/src/util/slab/slot.rs
-42
View File
@@ -1,42 +0,0 @@
use crate::loom::cell::UnsafeCell;
use crate::util::slab::{Entry, Generation};
/// Stores an entry in the slab.
pub(super) struct Slot<T> {
next: UnsafeCell<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: UnsafeCell::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;
})
}
}
tokio/src/util/slab/stack.rs
-58
View File
@@ -1,58 +0,0 @@
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()
}
}
tokio/src/util/slab/tests/loom_slab.rs
-327
View File
@@ -1,327 +0,0 @@
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
}
tokio/src/util/slab/tests/loom_stack.rs
-88
View File
@@ -1,88 +0,0 @@
use crate::util::slab::TransferStack;
use loom::cell::UnsafeCell;
use loom::sync::Arc;
use loom::thread;
#[test]
fn transfer_stack() {
loom::model(|| {
let causalities = [UnsafeCell::new(None), UnsafeCell::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(),
"UnsafeCell 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(),
"UnsafeCell 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(),
"UnsafeCell write must happen-before index is pushed to the stack!",
);
});
}
t1.join().unwrap();
t2.join().unwrap();
});
}
tokio/src/util/slab/tests/mod.rs
-2
View File
@@ -1,2 +0,0 @@
mod loom_slab;
mod loom_stack;