RawBlock: switch to Cell or AtomicPtr instead of UnsafeCell
This is the first step towards enabling multithreading on AtomTable. For now RawBlock will default to using Cell, which yields a byte- equivalent compiled output. Also adds an `atomic` feature, which, when enabled, will make RawBlock use an AtomicPtr instead, ensuring that it implements `Sync`.
This commit is contained in:
@@ -12,8 +12,8 @@ use parking_lot::{Mutex, RwLock};
|
|||||||
|
|
||||||
use crate::machine::heap::AllocError;
|
use crate::machine::heap::AllocError;
|
||||||
use crate::machine::machine_indices::IndexPtr;
|
use crate::machine::machine_indices::IndexPtr;
|
||||||
use crate::raw_block::RawBlock;
|
|
||||||
use crate::raw_block::RawBlockTraits;
|
use crate::raw_block::RawBlockTraits;
|
||||||
|
use crate::raw_block::{RawBlock, RawBlockConcurrent};
|
||||||
|
|
||||||
use ordered_float::OrderedFloat;
|
use ordered_float::OrderedFloat;
|
||||||
|
|
||||||
@@ -93,8 +93,9 @@ impl<T: fmt::Debug + RawBlockTraits> OffsetTableImpl<T> {
|
|||||||
// this shouldn't be able to fail
|
// this shouldn't be able to fail
|
||||||
let raw_block =
|
let raw_block =
|
||||||
Arc::try_unwrap(table.block.replace(RawBlock::empty_block())).unwrap();
|
Arc::try_unwrap(table.block.replace(RawBlock::empty_block())).unwrap();
|
||||||
self.0 =
|
self.0 = InnerOffsetTableImpl::Serial(SerialOffsetTable {
|
||||||
InnerOffsetTableImpl::Serial(SerialOffsetTable { block: raw_block });
|
block: raw_block.into(),
|
||||||
|
});
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
Err(table_arc) => {
|
Err(table_arc) => {
|
||||||
@@ -130,7 +131,7 @@ struct SerialOffsetTable<T: RawBlockTraits> {
|
|||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct ConcurrentOffsetTable<T: RawBlockTraits> {
|
pub struct ConcurrentOffsetTable<T: RawBlockTraits> {
|
||||||
block: Arcu<RawBlock<T>, GlobalEpochCounterPool>,
|
block: Arcu<RawBlock<T, RawBlockConcurrent>, GlobalEpochCounterPool>,
|
||||||
growth_lock: RwLock<()>,
|
growth_lock: RwLock<()>,
|
||||||
offset_locks: RwLock<Vec<RwLock<()>>>,
|
offset_locks: RwLock<Vec<RwLock<()>>>,
|
||||||
}
|
}
|
||||||
@@ -250,7 +251,8 @@ impl<T: RawBlockTraits> SerialOffsetTable<T> {
|
|||||||
|
|
||||||
let serial_tbl = mem::replace(self, empty_serial_tbl);
|
let serial_tbl = mem::replace(self, empty_serial_tbl);
|
||||||
let num_tbl_entries = serial_tbl.block.used_bytes() / size_of::<T>();
|
let num_tbl_entries = serial_tbl.block.used_bytes() / size_of::<T>();
|
||||||
let block = Arcu::new(serial_tbl.block, GlobalEpochCounterPool);
|
let raw_block: RawBlock<T, RawBlockConcurrent> = serial_tbl.block.into();
|
||||||
|
let block = Arcu::new(raw_block, GlobalEpochCounterPool);
|
||||||
|
|
||||||
let offset_locks: Vec<RwLock<()>> = (0..num_tbl_entries).map(|_| RwLock::new(())).collect();
|
let offset_locks: Vec<RwLock<()>> = (0..num_tbl_entries).map(|_| RwLock::new(())).collect();
|
||||||
|
|
||||||
@@ -403,6 +405,8 @@ impl F64Table {
|
|||||||
// which breaks the invariant indirection_tbl is meant to enforce.
|
// which breaks the invariant indirection_tbl is meant to enforce.
|
||||||
// Since this branch is never invoked, it does no harm, but that
|
// Since this branch is never invoked, it does no harm, but that
|
||||||
// that will eventually change.
|
// that will eventually change.
|
||||||
|
//
|
||||||
|
// Note: may be indirectly fixed by the use of the new RawBlockConcurrency trait.
|
||||||
{
|
{
|
||||||
let indirection_tbl = concurrent_tbl.indirection_tbl.lock();
|
let indirection_tbl = concurrent_tbl.indirection_tbl.lock();
|
||||||
|
|
||||||
@@ -473,7 +477,9 @@ impl F64Table {
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
*self = Self::Serial(SerialF64Table {
|
*self = Self::Serial(SerialF64Table {
|
||||||
indirection_tbl: indirection_tbl.into_inner(),
|
indirection_tbl: indirection_tbl.into_inner(),
|
||||||
offset_tbl: SerialOffsetTable { block: raw_block },
|
offset_tbl: SerialOffsetTable {
|
||||||
|
block: raw_block.into(),
|
||||||
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|||||||
199
src/raw_block.rs
199
src/raw_block.rs
@@ -1,8 +1,109 @@
|
|||||||
use core::marker::PhantomData;
|
use core::marker::PhantomData;
|
||||||
|
|
||||||
use std::alloc;
|
use std::alloc;
|
||||||
use std::cell::UnsafeCell;
|
use std::cell::Cell;
|
||||||
use std::ptr;
|
use std::ptr;
|
||||||
|
use std::sync::atomic::{AtomicPtr, Ordering};
|
||||||
|
|
||||||
|
trait PtrCellTrait: std::fmt::Debug {
|
||||||
|
fn new(val: *mut u8) -> Self;
|
||||||
|
|
||||||
|
fn get(&self) -> *mut u8;
|
||||||
|
|
||||||
|
/// Modifies the wrapped value.
|
||||||
|
fn set(&mut self, val: *mut u8);
|
||||||
|
|
||||||
|
/// Performs an atomic compare-and-swap on the wrapped value.
|
||||||
|
///
|
||||||
|
/// If the compare succeeded and `cb` returns `Some(new_ptr)`, stored `new_ptr` and returns `Ok(old_ptr).
|
||||||
|
///
|
||||||
|
/// If `cb` returns `None`, returns `Err(old_ptr)`.
|
||||||
|
///
|
||||||
|
/// May retry multiple times if the comparison fails.
|
||||||
|
fn try_update(&self, cb: impl Fn(*mut u8) -> Option<*mut u8>) -> Result<*mut u8, *mut u8>;
|
||||||
|
}
|
||||||
|
|
||||||
|
impl PtrCellTrait for Cell<*mut u8> {
|
||||||
|
fn new(val: *mut u8) -> Self {
|
||||||
|
Cell::new(val)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline(always)]
|
||||||
|
fn get(&self) -> *mut u8 {
|
||||||
|
Cell::get(self)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline(always)]
|
||||||
|
fn set(&mut self, val: *mut u8) {
|
||||||
|
Cell::set(self, val)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline(always)]
|
||||||
|
fn try_update(&self, cb: impl Fn(*mut u8) -> Option<*mut u8>) -> Result<*mut u8, *mut u8> {
|
||||||
|
let val = Cell::get(self);
|
||||||
|
if let Some(new_val) = cb(val) {
|
||||||
|
Cell::set(self, new_val);
|
||||||
|
Ok(val)
|
||||||
|
} else {
|
||||||
|
Err(val)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl PtrCellTrait for AtomicPtr<u8> {
|
||||||
|
fn new(val: *mut u8) -> Self {
|
||||||
|
AtomicPtr::new(val)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline(always)]
|
||||||
|
fn get(&self) -> *mut u8 {
|
||||||
|
self.load(Ordering::Acquire)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline(always)]
|
||||||
|
fn set(&mut self, val: *mut u8) {
|
||||||
|
*self.get_mut() = val;
|
||||||
|
}
|
||||||
|
|
||||||
|
#[inline]
|
||||||
|
fn try_update(&self, cb: impl Fn(*mut u8) -> Option<*mut u8>) -> Result<*mut u8, *mut u8> {
|
||||||
|
let mut prev = PtrCellTrait::get(self);
|
||||||
|
|
||||||
|
while let Some(next) = cb(prev) {
|
||||||
|
match self.compare_exchange_weak(prev, next, Ordering::Relaxed, Ordering::Acquire) {
|
||||||
|
x @ Ok(_) => return x,
|
||||||
|
Err(next_prev) => prev = next_prev,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Err(prev)
|
||||||
|
|
||||||
|
// TODO: replace with the following once 1.95 is the msrv:
|
||||||
|
// AtomicPtr::try_update(self, Ordering::Relaxed, Ordering::Acquire, cb)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Allows the choice of implementation for the mutable pointer in [`RawBlock`].
|
||||||
|
/// Can be one of:
|
||||||
|
/// - [`RawBlockSerial`] (using a [`Cell<*mut u8>`])
|
||||||
|
/// - [`RawBlockConcurrent`] (using a [`AtomicPtr<u8>`])
|
||||||
|
pub(crate) trait RawBlockConcurrency {
|
||||||
|
#[allow(private_bounds)]
|
||||||
|
type PtrCell: PtrCellTrait;
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Copy)]
|
||||||
|
pub struct RawBlockSerial();
|
||||||
|
#[derive(Debug, Clone, Copy)]
|
||||||
|
pub struct RawBlockConcurrent();
|
||||||
|
|
||||||
|
impl RawBlockConcurrency for RawBlockSerial {
|
||||||
|
type PtrCell = Cell<*mut u8>;
|
||||||
|
}
|
||||||
|
|
||||||
|
impl RawBlockConcurrency for RawBlockConcurrent {
|
||||||
|
type PtrCell = AtomicPtr<u8>;
|
||||||
|
}
|
||||||
|
|
||||||
use crate::machine::heap::AllocError;
|
use crate::machine::heap::AllocError;
|
||||||
|
|
||||||
@@ -13,22 +114,24 @@ pub trait RawBlockTraits {
|
|||||||
|
|
||||||
/// A block of memory with fast, lock-free appends.
|
/// A block of memory with fast, lock-free appends.
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct RawBlock<T: RawBlockTraits> {
|
pub struct RawBlock<T: RawBlockTraits, C: RawBlockConcurrency = RawBlockSerial> {
|
||||||
base: *const u8,
|
base: *const u8,
|
||||||
capacity: usize,
|
capacity: usize,
|
||||||
|
|
||||||
ptr: UnsafeCell<*mut u8>,
|
ptr: C::PtrCell,
|
||||||
_marker: PhantomData<T>,
|
_marker: PhantomData<T>,
|
||||||
|
_c_marker: PhantomData<C>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T: RawBlockTraits> RawBlock<T> {
|
impl<T: RawBlockTraits, C: RawBlockConcurrency> RawBlock<T, C> {
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn empty_block() -> Self {
|
pub fn empty_block() -> Self {
|
||||||
RawBlock {
|
RawBlock {
|
||||||
base: ptr::null(),
|
base: ptr::null(),
|
||||||
capacity: 0,
|
capacity: 0,
|
||||||
ptr: UnsafeCell::new(ptr::null_mut()),
|
ptr: C::PtrCell::new(ptr::null_mut()),
|
||||||
_marker: PhantomData,
|
_marker: PhantomData,
|
||||||
|
_c_marker: PhantomData,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -51,7 +154,7 @@ impl<T: RawBlockTraits> RawBlock<T> {
|
|||||||
}
|
}
|
||||||
self.base = new_base;
|
self.base = new_base;
|
||||||
self.capacity = cap;
|
self.capacity = cap;
|
||||||
*self.ptr.get_mut() = self.base.cast_mut();
|
self.ptr.set(self.base.cast_mut());
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -71,7 +174,7 @@ impl<T: RawBlockTraits> RawBlock<T> {
|
|||||||
} else {
|
} else {
|
||||||
self.base = new_base;
|
self.base = new_base;
|
||||||
self.capacity = size * 2;
|
self.capacity = size * 2;
|
||||||
*self.ptr.get_mut() = (self.base as usize + size) as *mut _;
|
self.ptr.set(self.base.add(size).cast_mut());
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -86,7 +189,7 @@ impl<T: RawBlockTraits> RawBlock<T> {
|
|||||||
new_block.init_at_size(self.capacity() * 2)?;
|
new_block.init_at_size(self.capacity() * 2)?;
|
||||||
let allocated = self.used_bytes();
|
let allocated = self.used_bytes();
|
||||||
self.base.copy_to(new_block.base.cast_mut(), allocated);
|
self.base.copy_to(new_block.base.cast_mut(), allocated);
|
||||||
*new_block.ptr.get_mut() = new_block.base.add(allocated).cast_mut();
|
new_block.ptr.set(new_block.base.add(allocated).cast_mut());
|
||||||
|
|
||||||
new_block.debug_check_invariants();
|
new_block.debug_check_invariants();
|
||||||
|
|
||||||
@@ -97,14 +200,12 @@ impl<T: RawBlockTraits> RawBlock<T> {
|
|||||||
#[inline(always)]
|
#[inline(always)]
|
||||||
fn debug_check_invariants(&self) {
|
fn debug_check_invariants(&self) {
|
||||||
if cfg!(debug_assertions) {
|
if cfg!(debug_assertions) {
|
||||||
unsafe {
|
assert!(
|
||||||
assert!(
|
self.ptr.get().cast_const() >= self.base,
|
||||||
*self.ptr.get() as *const _ >= self.base,
|
"self.ptr = {:?} < {:?} = self.base",
|
||||||
"self.ptr = {:?} < {:?} = self.base",
|
self.ptr.get(),
|
||||||
*self.ptr.get(),
|
self.base
|
||||||
self.base
|
);
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
assert!(self.used_bytes() <= self.capacity());
|
assert!(self.used_bytes() <= self.capacity());
|
||||||
}
|
}
|
||||||
@@ -117,27 +218,30 @@ impl<T: RawBlockTraits> RawBlock<T> {
|
|||||||
|
|
||||||
#[inline]
|
#[inline]
|
||||||
pub fn used_bytes(&self) -> usize {
|
pub fn used_bytes(&self) -> usize {
|
||||||
// TODO: safety: UnsafeCell.get()
|
// SAFETY:
|
||||||
// TODO: safety: prove that ∀Γ: reachable, Γ |- (ptr, base): same alloc
|
// - Invariant: `ptr` is in the same allocation as `base`
|
||||||
unsafe { (*self.ptr.get()).offset_from(self.base) as usize }
|
unsafe { self.ptr.get().offset_from(self.base) as usize }
|
||||||
}
|
|
||||||
|
|
||||||
#[inline(always)]
|
|
||||||
unsafe fn free_bytes(&self) -> usize {
|
|
||||||
self.capacity() - self.used_bytes()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub unsafe fn alloc(&self, size: usize) -> *mut u8 {
|
pub unsafe fn alloc(&self, size: usize) -> *mut u8 {
|
||||||
self.debug_check_invariants();
|
self.debug_check_invariants();
|
||||||
|
|
||||||
let aligned_size = size.next_multiple_of(T::align());
|
let aligned_size = size.next_multiple_of(T::align());
|
||||||
if self.free_bytes() >= aligned_size {
|
|
||||||
// TODO: make this an atomic add
|
match self.ptr.try_update(|ptr| {
|
||||||
let ptr = *self.ptr.get();
|
// SAFETY:
|
||||||
*self.ptr.get() = ptr.add(aligned_size) as *mut _;
|
// - Invariant: `ptr` is in the same allocation as `base`
|
||||||
ptr
|
let free_bytes = unsafe { self.capacity() - ptr.offset_from(self.base) as usize };
|
||||||
} else {
|
|
||||||
ptr::null_mut()
|
if free_bytes >= aligned_size {
|
||||||
|
Some(unsafe { ptr.add(aligned_size) })
|
||||||
|
} else {
|
||||||
|
// Not enough space: don't allocate and return a null pointer
|
||||||
|
None
|
||||||
|
}
|
||||||
|
}) {
|
||||||
|
Ok(ptr) => ptr,
|
||||||
|
Err(_) => ptr::null_mut(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -160,9 +264,9 @@ impl<T: RawBlockTraits> RawBlock<T> {
|
|||||||
// - Definition: self.base := alloc(self.capacity)
|
// - Definition: self.base := alloc(self.capacity)
|
||||||
let new_ptr = unsafe { self.base.add(new_size) };
|
let new_ptr = unsafe { self.base.add(new_size) };
|
||||||
|
|
||||||
debug_assert!(new_ptr as usize <= (*self.ptr.get_mut()) as usize,);
|
debug_assert!(new_ptr as usize <= self.ptr.get() as usize,);
|
||||||
|
|
||||||
*self.ptr.get_mut() = new_ptr as *mut u8;
|
self.ptr.set(new_ptr.cast_mut());
|
||||||
|
|
||||||
self.debug_check_invariants();
|
self.debug_check_invariants();
|
||||||
}
|
}
|
||||||
@@ -208,16 +312,37 @@ impl<T: RawBlockTraits> RawBlock<T> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T: RawBlockTraits> Drop for RawBlock<T> {
|
impl<T: RawBlockTraits, C: RawBlockConcurrency> Drop for RawBlock<T, C> {
|
||||||
fn drop(&mut self) {
|
fn drop(&mut self) {
|
||||||
if !self.base.is_null() {
|
if !self.base.is_null() {
|
||||||
unsafe {
|
unsafe {
|
||||||
let layout = alloc::Layout::from_size_align_unchecked(self.capacity(), T::align());
|
let layout = alloc::Layout::from_size_align_unchecked(self.capacity(), T::align());
|
||||||
alloc::dealloc(self.base as *mut _, layout);
|
alloc::dealloc(self.base as *mut _, layout);
|
||||||
}
|
}
|
||||||
|
}
|
||||||
self.base = ptr::null();
|
}
|
||||||
*self.ptr.get_mut() = ptr::null_mut();
|
}
|
||||||
|
|
||||||
|
impl<T: RawBlockTraits> From<RawBlock<T, RawBlockConcurrent>> for RawBlock<T, RawBlockSerial> {
|
||||||
|
fn from(other: RawBlock<T, RawBlockConcurrent>) -> Self {
|
||||||
|
Self {
|
||||||
|
base: other.base,
|
||||||
|
capacity: other.capacity,
|
||||||
|
ptr: PtrCellTrait::new(other.ptr.get()),
|
||||||
|
_marker: PhantomData,
|
||||||
|
_c_marker: PhantomData,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T: RawBlockTraits> From<RawBlock<T, RawBlockSerial>> for RawBlock<T, RawBlockConcurrent> {
|
||||||
|
fn from(other: RawBlock<T, RawBlockSerial>) -> Self {
|
||||||
|
Self {
|
||||||
|
base: other.base,
|
||||||
|
capacity: other.capacity,
|
||||||
|
ptr: PtrCellTrait::new(other.ptr.get()),
|
||||||
|
_marker: PhantomData,
|
||||||
|
_c_marker: PhantomData,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user