use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Barrier, Mutex};
const PAGE: usize = 4096;
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct WorkspaceReq {
pub a_bytes: usize,
pub tile_bytes: usize,
pub worker_scatter: usize,
pub b_bytes: usize,
pub team_scatter: usize,
pub barriers: usize,
}
impl WorkspaceReq {
pub fn bytes(&self) -> usize {
self.a_bytes
+ self.tile_bytes
+ self.b_bytes
+ (self.worker_scatter + self.team_scatter) * core::mem::size_of::<i64>()
+ self.barriers * core::mem::size_of::<Barrier>()
}
pub fn covered_by(&self, have: &WorkspaceReq) -> bool {
self.a_bytes <= have.a_bytes
&& self.tile_bytes <= have.tile_bytes
&& self.worker_scatter <= have.worker_scatter
&& self.b_bytes <= have.b_bytes
&& self.team_scatter <= have.team_scatter
&& self.barriers <= have.barriers
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct WorkspaceStats {
pub retained_bytes: usize,
pub leased_bytes: usize,
}
#[derive(Default, Debug)]
struct PageBuf {
ptr: *mut u8,
cap: usize,
traced: u64,
}
unsafe impl Send for PageBuf {}
impl PageBuf {
fn traced(owner: u64) -> Self {
Self {
traced: owner,
..Self::default()
}
}
fn ensure(&mut self, bytes: usize) -> *mut u8 {
if bytes == 0 {
return std::ptr::NonNull::<u8>::dangling().as_ptr();
}
if bytes > self.cap {
let want = bytes.next_power_of_two();
let layout = std::alloc::Layout::from_size_align(want, PAGE).expect("workspace layout");
let new = unsafe {
match self.ptr.is_null() {
true => std::alloc::alloc(layout),
false => {
let old = std::alloc::Layout::from_size_align_unchecked(self.cap, PAGE);
std::alloc::realloc(self.ptr, old, want)
}
}
};
assert!(
!new.is_null(),
"workspace allocation of {want} bytes failed"
);
self.ptr = new;
self.cap = want;
if self.traced != 0 {
record(self.traced, want);
}
}
self.ptr
}
fn as_ptr(&self) -> *mut u8 {
if self.ptr.is_null() {
std::ptr::NonNull::<u8>::dangling().as_ptr()
} else {
self.ptr
}
}
fn cap_bytes(&self) -> usize {
self.cap
}
fn trim(&mut self) {
if !self.ptr.is_null() {
unsafe {
std::alloc::dealloc(
self.ptr,
std::alloc::Layout::from_size_align_unchecked(self.cap, PAGE),
)
};
self.ptr = std::ptr::null_mut();
self.cap = 0;
}
}
}
impl Drop for PageBuf {
fn drop(&mut self) {
self.trim();
}
}
#[derive(Debug)]
struct WorkerSlot {
a: PageBuf,
tile: PageBuf,
scratch: Vec<i64>,
}
impl WorkerSlot {
fn new(owner: u64) -> Self {
Self {
a: PageBuf::traced(owner),
tile: PageBuf::traced(owner),
scratch: Vec::new(),
}
}
fn bytes(&self) -> usize {
self.a.cap_bytes()
+ self.tile.cap_bytes()
+ self.scratch.capacity() * core::mem::size_of::<i64>()
}
fn trim(&mut self) {
self.a.trim();
self.tile.trim();
self.scratch = Vec::new();
}
}
#[derive(Debug, Default)]
struct TeamSet {
b: PageBuf,
scatter: Vec<i64>,
barriers: Vec<Barrier>,
pm: usize,
pn: usize,
}
impl TeamSet {
fn prepare(&mut self, req: &WorkspaceReq, pm: usize, pn: usize) {
self.scatter.clear();
self.scatter.reserve(req.team_scatter);
if (self.pm, self.pn) != (pm, pn) {
self.barriers = (0..req.barriers).map(|_| Barrier::new(pm)).collect();
self.pm = pm;
self.pn = pn;
}
}
fn bytes(&self) -> usize {
self.b.cap_bytes()
+ self.scatter.capacity() * core::mem::size_of::<i64>()
+ self.barriers.capacity() * core::mem::size_of::<Barrier>()
}
}
#[derive(Debug)]
pub struct TeamLease<'a> {
owner: &'a ArenaProvider,
set: Option<TeamSet>,
billed: usize,
}
impl<'a> TeamLease<'a> {
pub fn panel(&mut self, bytes: usize) -> *mut u8 {
let set = self.set.as_mut().expect("team lease holds its set");
let before = set.b.cap_bytes();
let ptr = set.b.ensure(bytes);
let after = set.b.cap_bytes();
if after != before {
let delta = after as isize - before as isize;
let mut st = lock(&self.owner.state);
st.owned = add_signed(st.owned, delta, "retained bytes");
st.leased = add_signed(st.leased, delta, "leased bytes");
st.panel = add_signed(st.panel, delta, "panel bytes");
self.billed = add_signed(self.billed, delta, "billed bytes");
}
ptr
}
pub fn scatter_mut(&mut self) -> &mut Vec<i64> {
&mut self.set.as_mut().expect("team lease holds its set").scatter
}
pub fn scatter(&self) -> &[i64] {
&self.set.as_ref().expect("team lease holds its set").scatter
}
pub fn barriers(&self) -> &[Barrier] {
&self
.set
.as_ref()
.expect("team lease holds its set")
.barriers
}
}
impl Drop for TeamLease<'_> {
fn drop(&mut self) {
if let Some(set) = self.set.take() {
let mut st = lock(&self.owner.state);
let delta = set.bytes() as isize - self.billed as isize;
st.owned = add_signed(st.owned, delta, "retained bytes");
st.leased = sub(st.leased, self.billed, "leased bytes");
st.teams.push(set);
}
}
}
mod sealed {
pub trait Sealed {}
impl Sealed for super::ArenaProvider {}
}
pub trait WorkspaceProvider: sealed::Sealed + Sync {
fn with_worker(&self, req: &WorkspaceReq, f: &mut dyn FnMut(*mut u8, *mut u8, &mut Vec<i64>));
fn take_team(&self, req: &WorkspaceReq, pm: usize, pn: usize) -> TeamLease<'_>;
fn trim(&self);
}
#[derive(Debug, Default)]
struct State {
threads: HashMap<std::thread::ThreadId, u64>,
workers: HashMap<u64, WorkerSlot>,
teams: Vec<TeamSet>,
owned: usize,
leased: usize,
panel: usize,
}
#[derive(Debug)]
pub struct ArenaProvider {
key: u64,
state: Mutex<State>,
next_worker: AtomicU64,
traced: u64,
}
impl Default for ArenaProvider {
fn default() -> Self {
Self::new()
}
}
static NEXT_PROVIDER: AtomicU64 = AtomicU64::new(1);
static TRACE: Mutex<Vec<(u64, std::thread::ThreadId, usize)>> = Mutex::new(Vec::new());
fn lock<T>(m: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
m.lock().unwrap_or_else(|e| e.into_inner())
}
fn add_signed(value: usize, delta: isize, what: &str) -> usize {
value
.checked_add_signed(delta)
.unwrap_or_else(|| panic!("workspace {what} went out of range"))
}
fn sub(value: usize, amount: usize, what: &str) -> usize {
value
.checked_sub(amount)
.unwrap_or_else(|| panic!("workspace {what} went out of range"))
}
fn record(owner: u64, bytes: usize) {
lock(&TRACE).push((owner, std::thread::current().id(), bytes));
}
impl ArenaProvider {
pub fn new() -> Self {
Self {
key: NEXT_PROVIDER.fetch_add(1, Ordering::Relaxed),
state: Mutex::new(State::default()),
next_worker: AtomicU64::new(0),
traced: 0,
}
}
pub fn traced() -> Self {
let mut arena = Self::new();
arena.traced = arena.key;
arena
}
pub fn trace_take(&self) -> Vec<(std::thread::ThreadId, usize)> {
let mut mine = Vec::new();
lock(&TRACE).retain(|(owner, thread, bytes)| {
if *owner == self.key {
mine.push((*thread, *bytes));
false
} else {
true
}
});
mine
}
pub fn stats(&self) -> WorkspaceStats {
let st = lock(&self.state);
WorkspaceStats {
retained_bytes: st.owned,
leased_bytes: st.leased,
}
}
pub fn retained_bytes(&self) -> usize {
self.stats().retained_bytes
}
pub fn retained_panel_bytes(&self) -> usize {
lock(&self.state).panel
}
fn checkout(&self) -> Option<(u64, WorkerSlot, usize)> {
let thread = std::thread::current().id();
let mut st = lock(&self.state);
let index = match st.threads.get(&thread) {
Some(&index) => index,
None => {
let index = self.next_worker.fetch_add(1, Ordering::Relaxed);
st.threads.insert(thread, index);
return Some((index, WorkerSlot::new(self.traced), 0));
}
};
let slot = st.workers.remove(&index)?;
let billed = slot.bytes();
st.leased += billed;
Some((index, slot, billed))
}
fn give_back(&self, index: u64, slot: WorkerSlot, billed: usize) {
let delta = slot.bytes() as isize - billed as isize;
let mut st = lock(&self.state);
st.owned = add_signed(st.owned, delta, "retained bytes");
st.leased = sub(st.leased, billed, "leased bytes");
st.workers.insert(index, slot);
}
}
impl Drop for ArenaProvider {
fn drop(&mut self) {
if self.traced != 0 {
lock(&TRACE).retain(|(owner, _, _)| *owner != self.key);
}
}
}
struct SlotGuard<'a> {
owner: &'a ArenaProvider,
index: u64,
slot: Option<WorkerSlot>,
billed: usize,
}
impl Drop for SlotGuard<'_> {
fn drop(&mut self) {
if let Some(slot) = self.slot.take() {
self.owner.give_back(self.index, slot, self.billed);
}
}
}
impl WorkspaceProvider for ArenaProvider {
fn with_worker(&self, req: &WorkspaceReq, f: &mut dyn FnMut(*mut u8, *mut u8, &mut Vec<i64>)) {
let Some((index, slot, billed)) = self.checkout() else {
return fresh_worker(req, f);
};
let mut guard = SlotGuard {
owner: self,
index,
slot: Some(slot),
billed,
};
let slot = guard.slot.as_mut().expect("guard holds its slot");
slot.a.ensure(req.a_bytes);
slot.tile.ensure(req.tile_bytes);
slot.scratch.clear();
slot.scratch.reserve(req.worker_scatter);
let (a, tile) = (slot.a.as_ptr(), slot.tile.as_ptr());
f(a, tile, &mut slot.scratch);
}
fn take_team(&self, req: &WorkspaceReq, pm: usize, pn: usize) -> TeamLease<'_> {
let (mut set, before, before_panel) = {
let mut st = lock(&self.state);
let set = st.teams.pop().unwrap_or_default();
let (bytes, panel) = (set.bytes(), set.b.cap_bytes());
(set, bytes, panel)
};
let mut teardown = Teardown {
owner: self,
before,
before_panel,
armed: true,
};
set.prepare(req, pm, pn);
let after = set.bytes();
teardown.armed = false;
let mut st = lock(&self.state);
st.owned = add_signed(st.owned, after as isize - before as isize, "retained bytes");
st.leased += after;
drop(st);
TeamLease {
owner: self,
set: Some(set),
billed: after,
}
}
fn trim(&self) {
let mut st = lock(&self.state);
let idle_slots: usize = st.workers.values().map(|s| s.bytes()).sum();
let idle_teams: usize = st.teams.iter().map(|t| t.bytes()).sum();
let idle_panels: usize = st.teams.iter().map(|t| t.b.cap_bytes()).sum();
for slot in st.workers.values_mut() {
slot.trim();
}
st.teams.clear();
st.owned = sub(st.owned, idle_slots + idle_teams, "retained bytes");
st.panel = sub(st.panel, idle_panels, "panel bytes");
}
}
struct Teardown<'a> {
owner: &'a ArenaProvider,
before: usize,
before_panel: usize,
armed: bool,
}
impl Drop for Teardown<'_> {
fn drop(&mut self) {
if !self.armed {
return;
}
let mut st = lock(&self.owner.state);
st.owned = sub(st.owned, self.before, "retained bytes");
st.panel = sub(st.panel, self.before_panel, "panel bytes");
}
}
fn fresh_worker(req: &WorkspaceReq, f: &mut dyn FnMut(*mut u8, *mut u8, &mut Vec<i64>)) {
let mut a = PageBuf::default();
let mut tile = PageBuf::default();
let mut scratch = Vec::new();
a.ensure(req.a_bytes);
tile.ensure(req.tile_bytes);
scratch.reserve(req.worker_scatter);
f(a.as_ptr(), tile.as_ptr(), &mut scratch);
}
#[derive(Debug, Default)]
pub struct Workspace {
arena: ArenaProvider,
req: WorkspaceReq,
slots: usize,
}
impl Workspace {
pub fn with_req(req: &WorkspaceReq) -> Self {
let mut ws = Self {
arena: ArenaProvider::new(),
req: *req,
slots: 0,
};
ws.warm();
ws.slots = 1;
ws
}
pub fn reserve(&mut self, req: &WorkspaceReq) {
let before = self.req;
self.req.a_bytes = self.req.a_bytes.max(req.a_bytes);
self.req.tile_bytes = self.req.tile_bytes.max(req.tile_bytes);
self.req.worker_scatter = self.req.worker_scatter.max(req.worker_scatter);
self.req.b_bytes = self.req.b_bytes.max(req.b_bytes);
self.req.team_scatter = self.req.team_scatter.max(req.team_scatter);
self.req.barriers = self.req.barriers.max(req.barriers);
if self.req != before {
self.warm();
}
}
pub fn have(&self) -> usize {
self.req.bytes()
}
pub fn req(&self) -> &WorkspaceReq {
&self.req
}
pub fn slots(&self) -> usize {
self.slots
}
pub fn prepare_for(&mut self, exec: &crate::Exec<'_>) {
let slots = exec.budget().max(1);
if slots <= self.slots {
return;
}
let (arena, req) = (&self.arena, &self.req);
let warmed = exec.broadcast(slots, &|_| arena.with_worker(req, &mut |_, _, _| {}));
if warmed.is_ok() {
self.slots = slots;
}
}
pub fn as_provider(&self) -> &dyn WorkspaceProvider {
&self.arena
}
pub fn retained_bytes(&self) -> usize {
self.arena.retained_bytes()
}
fn warm(&self) {
self.arena.with_worker(&self.req, &mut |_, _, _| {});
let mut lease = self.arena.take_team(&self.req, 1, 1);
if self.req.b_bytes > 0 {
lease.panel(self.req.b_bytes);
}
drop(lease);
}
}
#[cfg(test)]
mod thread_entries {
use super::*;
fn thread_slots(provider: &ArenaProvider) -> usize {
lock(&provider.state).threads.len()
}
fn trace_records(owner: u64) -> usize {
lock(&TRACE)
.iter()
.filter(|(key, _, _)| *key == owner)
.count()
}
#[test]
fn one_entry_per_thread_and_none_carried_over() {
let req = WorkspaceReq::default();
let provider = ArenaProvider::default();
assert_eq!(thread_slots(&provider), 0);
provider.with_worker(&req, &mut |_, _, _| {});
assert_eq!(thread_slots(&provider), 1);
provider.with_worker(&req, &mut |_, _, _| {});
assert_eq!(thread_slots(&provider), 1);
let fresh = ArenaProvider::default();
assert_eq!(thread_slots(&fresh), 0);
fresh.with_worker(&req, &mut |_, _, _| {});
assert_eq!(thread_slots(&fresh), 1);
assert_eq!(thread_slots(&provider), 1);
}
#[test]
fn two_threads_get_two_entries() {
let provider = ArenaProvider::default();
let barrier = std::sync::Barrier::new(2);
std::thread::scope(|scope| {
for _ in 0..2 {
scope.spawn(|| {
provider.with_worker(&WorkspaceReq::default(), &mut |_, _, _| {});
barrier.wait();
});
}
});
assert_eq!(thread_slots(&provider), 2);
}
#[test]
fn dropping_a_traced_provider_drops_its_trace_records() {
let provider = ArenaProvider::traced();
let key = provider.key;
provider.with_worker(
&WorkspaceReq {
a_bytes: 4096,
..Default::default()
},
&mut |_, _, _| {},
);
assert!(trace_records(key) > 0);
drop(provider);
assert_eq!(
trace_records(key),
0,
"a dropped traced provider must not keep records in the process-global log"
);
}
}