use std::collections::HashSet;
use std::fmt;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tf_tree_arena::{Arena, ArenaLayout, HeapArena, LayoutError};
#[cfg(all(feature = "shm", target_os = "linux"))]
use tf_tree_arena::{AttachMode, MappedArena, ShmError};
use tf_tree_core::arena_view::{ArenaBuilder, ArenaView};
use tf_tree_core::edge::{claim, EdgeKind, EdgeRecord, Publisher};
use tf_tree_core::frame::blake3_64;
use tf_tree_core::plan::{compile, Domain, EdgeMeta, Guard, InterpPolicy, Stamp, SystemDomain};
use tf_tree_core::topology::{TopoLockError, TopoLockView};
use tf_tree_core::{
EdgeId, FrameError, FrameId, LookupError, ParticipantError, PushError, TopologyError,
};
use tf_tree_math::Iso3;
use crate::cache;
pub(crate) const MIN_BACKOFF: Duration = Duration::from_micros(200);
pub(crate) const MAX_BACKOFF: Duration = Duration::from_millis(4);
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum AwaitError {
#[error("no frame with name hash {hash:#018x} appeared before the deadline")]
Timeout {
hash: u64,
},
#[error("await_frames refuses a writable tree; use Tree::frame, which interns on demand and cannot fail for absence")]
WritableTree,
#[error("await_frames refuses a frozen .tft tree: it has no writers, so no name can appear")]
FrozenTree,
#[error("{0:?}")]
Frame(FrameError),
#[error("this tree was opened before a fork() and is being used in the child")]
ChildDetached,
}
fn all_interned<const N: usize>(found: &[Option<FrameId>; N]) -> Option<[FrameId; N]> {
let mut out = [FrameId::new(1)?; N];
for (dst, src) in out.iter_mut().zip(found.iter()) {
*dst = (*src)?;
}
Some(out)
}
fn stored_name(bytes: &[u8], len: u8) -> String {
let n = (len as usize).min(bytes.len());
String::from_utf8_lossy(&bytes[..n]).into_owned()
}
fn next_pow2_u32(n: u32) -> u32 {
let mut p: u64 = 1;
let target = u64::from(n);
while p < target {
p <<= 1;
}
if p > u64::from(u32::MAX) {
1u32 << 31
} else {
p as u32
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Capacity(u32);
impl Capacity {
#[must_use]
pub fn slots(n: u32) -> Capacity {
Capacity(next_pow2_u32(n))
}
#[must_use]
pub fn history(rate_hz: f64, secs: f64) -> Capacity {
let needed = (rate_hz * secs).ceil();
let clamped = if needed.is_finite() && needed >= 1.0 {
if needed > f64::from(u32::MAX) {
u32::MAX
} else {
needed as u32
}
} else {
1
};
Capacity(next_pow2_u32(clamped))
}
#[inline]
#[must_use]
pub fn get(self) -> u32 {
self.0
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub struct EdgeCfg {
pub capacity: Capacity,
pub interp: Option<InterpPolicy>,
pub domain: Option<u8>,
pub nominal_rate_mhz: u32,
}
impl EdgeCfg {
#[must_use]
pub fn new(capacity: Capacity) -> EdgeCfg {
EdgeCfg {
capacity,
interp: None,
domain: None,
nominal_rate_mhz: 0,
}
}
#[must_use]
pub fn nominal_rate_hz(mut self, rate_hz: f64) -> EdgeCfg {
let mhz = rate_hz * 1000.0;
self.nominal_rate_mhz = if mhz.is_finite() && mhz >= 1.0 && mhz <= f64::from(u32::MAX) {
mhz.round() as u32
} else {
0
};
self
}
#[must_use]
pub fn interp(mut self, interp: InterpPolicy) -> EdgeCfg {
self.interp = Some(interp);
self
}
#[must_use]
pub fn domain(mut self, domain: u8) -> EdgeCfg {
self.domain = Some(domain);
self
}
}
#[derive(Clone, Copy, Debug)]
enum EdgeDeclKind {
Static(Iso3),
Dynamic(EdgeCfg),
}
#[derive(Clone, Debug)]
struct EdgeDecl {
parent: String,
child: String,
kind: EdgeDeclKind,
}
#[derive(Clone, Debug)]
pub struct TreeBuilder {
default_interp: InterpPolicy,
default_domain: u8,
frames: Vec<String>,
edges: Vec<EdgeDecl>,
frame_headroom: u32,
edge_headroom: u32,
}
impl Default for TreeBuilder {
fn default() -> Self {
TreeBuilder::new()
}
}
impl TreeBuilder {
#[must_use]
pub fn new() -> TreeBuilder {
TreeBuilder {
default_interp: InterpPolicy::ScLerp,
default_domain: SystemDomain::TAG,
frames: Vec::new(),
edges: Vec::new(),
frame_headroom: 0,
edge_headroom: 0,
}
}
#[must_use]
pub fn default_interp(mut self, interp: InterpPolicy) -> TreeBuilder {
self.default_interp = interp;
self
}
#[must_use]
pub fn default_domain(mut self, domain: u8) -> TreeBuilder {
self.default_domain = domain;
self
}
#[must_use]
pub fn frame(mut self, name: &str) -> TreeBuilder {
self.frames.push(name.to_owned());
self
}
#[must_use]
pub fn static_edge(mut self, parent: &str, child: &str, iso: &Iso3) -> TreeBuilder {
self.edges.push(EdgeDecl {
parent: parent.to_owned(),
child: child.to_owned(),
kind: EdgeDeclKind::Static(*iso),
});
self
}
#[must_use]
pub fn dynamic_edge(mut self, parent: &str, child: &str, cfg: EdgeCfg) -> TreeBuilder {
self.edges.push(EdgeDecl {
parent: parent.to_owned(),
child: child.to_owned(),
kind: EdgeDeclKind::Dynamic(cfg),
});
self
}
#[must_use]
pub fn frame_headroom(mut self, n: u32) -> TreeBuilder {
self.frame_headroom = n;
self
}
#[must_use]
pub fn edge_headroom(mut self, n: u32) -> TreeBuilder {
self.edge_headroom = n;
self
}
pub fn build(self) -> Result<Tree, BuildError> {
let arena = self
.build_with(|layout, pid, start, boot| Ok(HeapArena::new(layout, pid, start, boot)))?;
let backing = ArenaBacking::Heap(arena);
let (participant, incarnation) = register_participant(&ArenaView::new(backing.as_dyn()))
.map_err(BuildError::Participant)?;
let liveness = liveness_for(ArenaView::new(backing.as_dyn()).header().boot_id);
#[cfg(all(feature = "shm", target_os = "linux"))]
let fork_gen = fork_gen_for(&backing);
Ok(Tree {
arena: backing,
participant,
incarnation,
liveness,
decl: Mutex::new(()),
#[cfg(all(feature = "shm", target_os = "linux"))]
attachment: None,
#[cfg(all(feature = "shm", target_os = "linux"))]
claim_lock: None,
#[cfg(all(feature = "shm", target_os = "linux"))]
fork_gen,
})
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub fn build_shared(self, name: &str) -> Result<Tree, BuildError> {
let arena = self.build_with(|layout, pid, start, boot| {
MappedArena::create(name, layout, pid, start, boot).map_err(BuildError::Shm)
})?;
arena.populate_hot();
let backing = ArenaBacking::Mapped(arena);
let (participant, incarnation) = register_participant(&ArenaView::new(backing.as_dyn()))
.map_err(BuildError::Participant)?;
let liveness = liveness_for(ArenaView::new(backing.as_dyn()).header().boot_id);
#[cfg(all(feature = "shm", target_os = "linux"))]
let fork_gen = fork_gen_for(&backing);
Ok(Tree {
arena: backing,
participant,
incarnation,
liveness,
decl: Mutex::new(()),
#[cfg(all(feature = "shm", target_os = "linux"))]
attachment: None,
#[cfg(all(feature = "shm", target_os = "linux"))]
claim_lock: None,
#[cfg(all(feature = "shm", target_os = "linux"))]
fork_gen,
})
}
fn build_with<A: Arena>(
self,
make: impl FnOnce(&ArenaLayout, u32, u64, [u8; 16]) -> Result<A, BuildError>,
) -> Result<A, BuildError> {
let mut names: Vec<&str> = Vec::new();
let mut seen: HashSet<&str> = HashSet::new();
for f in &self.frames {
if seen.insert(f.as_str()) {
names.push(f.as_str());
}
}
for e in &self.edges {
if seen.insert(e.parent.as_str()) {
names.push(e.parent.as_str());
}
if seen.insert(e.child.as_str()) {
names.push(e.child.as_str());
}
}
let mut children: HashSet<&str> = HashSet::new();
for e in &self.edges {
if !children.insert(e.child.as_str()) {
return Err(BuildError::DuplicateEdge {
child: blake3_64(&e.child),
});
}
}
let frame_count = names.len() as u64;
let edge_count = self.edges.len() as u64;
let max_frames = frame_count + 1 + u64::from(self.frame_headroom);
let max_edges = edge_count + 1 + u64::from(self.edge_headroom);
let max_frames = u32::try_from(max_frames).map_err(|_| BuildError::TooManyFrames)?;
let max_edges = u32::try_from(max_edges).map_err(|_| BuildError::TooManyEdges)?;
let mut caps = std::vec![0u32; max_edges as usize];
for (i, e) in self.edges.iter().enumerate() {
if let EdgeDeclKind::Dynamic(cfg) = &e.kind {
caps[i + 1] = cfg.capacity.get();
}
}
let layout = ArenaLayout::new(max_frames, max_edges, caps)?;
let boot_id = boot_id();
let mut arena = make(&layout, std::process::id(), process_start_time(), boot_id)?;
{
let mut builder = ArenaBuilder::new(&mut arena);
builder
.view()
.header()
.edge_count
.store(edge_count as u32 + 1, Ordering::Relaxed);
for &n in &names {
builder.view().intern(n).map_err(BuildError::Frame)?;
}
let mut running_off: u32 = 0;
for (i, e) in self.edges.iter().enumerate() {
let edge_id = (i + 1) as u32;
let parent = builder
.view()
.intern(&e.parent)
.map_err(BuildError::Frame)?;
let child = builder.view().intern(&e.child).map_err(BuildError::Frame)?;
let record = match &e.kind {
EdgeDeclKind::Static(iso) => EdgeRecord::static_edge(
parent.get(),
child.get(),
iso.to_bits(),
self.default_domain,
),
EdgeDeclKind::Dynamic(cfg) => {
let capacity = cfg.capacity.get();
let interp = cfg.interp.unwrap_or(self.default_interp);
let domain = cfg.domain.unwrap_or(self.default_domain);
let mut record = EdgeRecord::dynamic(
parent.get(),
child.get(),
capacity,
running_off,
running_off,
interp.as_u8(),
domain,
);
record.nominal_rate_mhz = cfg.nominal_rate_mhz;
running_off += capacity;
record
}
};
builder
.declare_edge(EdgeId(edge_id), record)
.map_err(BuildError::Topology)?;
builder
.view()
.topology()
.set_parent(child, parent.get(), edge_id)
.map_err(BuildError::Topology)?;
}
}
Ok(arena)
}
}
enum ArenaBacking {
Heap(HeapArena),
#[cfg(all(feature = "shm", target_os = "linux"))]
Mapped(MappedArena),
#[cfg(all(feature = "shm", target_os = "linux"))]
Frozen(tf_tree_arena::FrozenArena),
}
impl ArenaBacking {
fn as_dyn(&self) -> &dyn Arena {
match self {
ArenaBacking::Heap(a) => a,
#[cfg(all(feature = "shm", target_os = "linux"))]
ArenaBacking::Mapped(a) => a,
#[cfg(all(feature = "shm", target_os = "linux"))]
ArenaBacking::Frozen(a) => a,
}
}
fn is_writable(&self) -> bool {
match self {
ArenaBacking::Heap(_) => true,
#[cfg(all(feature = "shm", target_os = "linux"))]
ArenaBacking::Mapped(a) => a.is_writable(),
#[cfg(all(feature = "shm", target_os = "linux"))]
ArenaBacking::Frozen(_) => false,
}
}
fn is_shared(&self) -> bool {
match self {
#[cfg(all(feature = "shm", target_os = "linux"))]
ArenaBacking::Mapped(_) => true,
#[cfg(all(feature = "shm", target_os = "linux"))]
ArenaBacking::Frozen(_) => false,
ArenaBacking::Heap(_) => false,
}
}
fn is_frozen(&self) -> bool {
match self {
#[cfg(all(feature = "shm", target_os = "linux"))]
ArenaBacking::Frozen(_) => true,
#[cfg(all(feature = "shm", target_os = "linux"))]
ArenaBacking::Mapped(_) => false,
ArenaBacking::Heap(_) => false,
}
}
}
pub struct EdgeWriter<'a> {
publisher: Publisher<'a>,
#[cfg(all(feature = "shm", target_os = "linux"))]
fork_gen: Option<u64>,
#[cfg(all(feature = "shm", target_os = "linux"))]
_lease: Option<ClaimLease>,
}
impl EdgeWriter<'_> {
#[cfg(all(feature = "shm", target_os = "linux"))]
fn detached(&self) -> bool {
self.fork_gen
.is_some_and(|g| g != tf_tree_ipc::fork::generation())
}
pub fn push(&self, stamp: i64, iso: &Iso3) -> Result<(), PushError> {
#[cfg(all(feature = "shm", target_os = "linux"))]
if self.detached() {
return Err(PushError::ChildDetached);
}
self.publisher.push(stamp, iso)
}
}
impl Drop for EdgeWriter<'_> {
fn drop(&mut self) {
#[cfg(all(feature = "shm", target_os = "linux"))]
if self.detached() {
self.publisher.abandon();
}
}
}
impl<'a> core::ops::Deref for EdgeWriter<'a> {
type Target = Publisher<'a>;
fn deref(&self) -> &Publisher<'a> {
&self.publisher
}
}
pub struct OwnedWriter {
writer: Box<EdgeWriter<'static>>,
#[allow(dead_code)]
tree: Arc<Tree>,
}
impl OwnedWriter {
#[inline]
#[must_use]
pub fn edge(&self) -> EdgeId {
self.writer.edge()
}
#[inline]
pub fn push(&self, stamp: i64, iso: &Iso3) -> Result<(), PushError> {
self.writer.push(stamp, iso)
}
pub fn release(self) {
drop(self);
}
}
#[cfg(all(feature = "test-hooks", feature = "shm", target_os = "linux"))]
#[doc(hidden)]
pub static CLAIM_WINDOW_HOOK: std::sync::OnceLock<fn()> = std::sync::OnceLock::new();
#[cfg(all(feature = "shm", target_os = "linux"))]
pub(crate) struct ClaimLease {
lock: std::sync::Arc<tf_tree_ipc::LockFile>,
edge: u32,
fork_gen: u64,
}
#[cfg(all(feature = "shm", target_os = "linux"))]
impl Drop for ClaimLease {
fn drop(&mut self) {
if self.fork_gen != tf_tree_ipc::fork::generation() {
return;
}
let _ = self.lock.release_claim(self.edge);
}
}
type BoxedLiveness = Box<dyn Fn(u32, &tf_tree_core::ParticipantRecord) -> bool + Send + Sync>;
pub struct Tree {
arena: ArenaBacking,
participant: u32,
incarnation: u64,
liveness: BoxedLiveness,
decl: Mutex<()>,
#[cfg(all(feature = "shm", target_os = "linux"))]
attachment: Option<crate::open::Attachment>,
#[cfg(all(feature = "shm", target_os = "linux"))]
claim_lock: Option<std::sync::Arc<tf_tree_ipc::LockFile>>,
#[cfg(all(feature = "shm", target_os = "linux"))]
fork_gen: Option<u64>,
}
impl Tree {
#[must_use]
pub fn detached(&self) -> bool {
#[cfg(all(feature = "shm", target_os = "linux"))]
{
self.fork_gen
.is_some_and(|g| g != tf_tree_ipc::fork::generation())
}
#[cfg(not(all(feature = "shm", target_os = "linux")))]
{
false
}
}
pub(crate) fn view(&self) -> ArenaView<'_> {
#[cfg(all(feature = "shm", target_os = "linux"))]
if self.detached() {
return ArenaView::new(poison_arena());
}
ArenaView::new(self.arena.as_dyn())
.as_participant(self.participant)
.with_liveness(&*self.liveness)
.writable(self.is_writable())
}
pub fn frame(&self, name: &str) -> Result<FrameId, FrameError> {
if self.detached() {
return Err(FrameError::ChildDetached);
}
if !self.arena.is_writable() {
return self.view().find_frame(name)?.ok_or(FrameError::ReadOnly);
}
self.view().intern(name)
}
pub fn frames(&self) -> Result<Vec<String>, LookupError> {
if self.detached() {
return Err(LookupError::ChildDetached);
}
let view = self.view();
let count = view.header().frame_count.load(Ordering::Relaxed);
let mut out = Vec::with_capacity(count as usize);
for raw in 1..=count {
let Some(id) = FrameId::new(raw) else {
continue;
};
let Some(rec) = view.frame_record(id) else {
continue;
};
if rec.name_hash == 0 {
continue;
}
out.push(stored_name(&rec.name, rec.name_len));
}
Ok(out)
}
pub fn await_frames<const N: usize>(
&self,
names: [&str; N],
timeout: Duration,
) -> Result<[FrameId; N], AwaitError> {
if self.is_writable() {
return Err(AwaitError::WritableTree);
}
if self.arena.is_frozen() {
return Err(AwaitError::FrozenTree);
}
let start = std::time::Instant::now();
let mut found: [Option<FrameId>; N] = [None; N];
let mut backoff = MIN_BACKOFF;
loop {
if self.detached() {
return Err(AwaitError::ChildDetached);
}
let view = self.view();
for (slot, name) in found.iter_mut().zip(names.iter()) {
if slot.is_some() {
continue;
}
match view.find_frame(name) {
Ok(id) => *slot = id,
Err(e) => return Err(AwaitError::Frame(e)),
}
}
if let Some(ids) = all_interned(&found) {
return Ok(ids);
}
if start.elapsed() >= timeout {
let hash = names
.iter()
.zip(found.iter())
.find(|(_, slot)| slot.is_none())
.map_or(0, |(name, _)| blake3_64(name));
return Err(AwaitError::Timeout { hash });
}
let left = timeout.saturating_sub(start.elapsed());
std::thread::sleep(core::cmp::min(backoff, left));
backoff = core::cmp::min(backoff * 2, MAX_BACKOFF);
}
}
pub fn edges(&self) -> Result<Vec<(String, String)>, LookupError> {
if self.detached() {
return Err(LookupError::ChildDetached);
}
let view = self.view();
let count = view.header().edge_count.load(Ordering::Relaxed);
let mut out = Vec::with_capacity(count.saturating_sub(1) as usize);
for raw in 1..=count {
let Some(rec) = view.edge(EdgeId(raw)) else {
continue;
};
let name = |f: u32| -> Option<String> {
let r = view.frame_record(FrameId::new(f)?)?;
Some(stored_name(&r.name, r.name_len))
};
let (Some(parent), Some(child)) = (name(rec.parent), name(rec.child)) else {
continue;
};
out.push((parent, child));
}
Ok(out)
}
pub fn reparent(&self, child: FrameId, new_parent: FrameId) -> Result<(), ReparentError> {
if self.detached() {
return Err(ReparentError::ChildDetached);
}
if !self.arena.is_writable() {
return Err(ReparentError::ReadOnly);
}
let _local = self.decl.lock().unwrap_or_else(|e| e.into_inner());
let view = self.view();
let header = view.header();
let lock = TopoLockView::new(&header.topo_lock.owner, &header.topo_lock.acquired_at_nanos);
let participants = view.participants();
let arena_boot = header.boot_id;
let is_alive = move |slot: u32| participant_is_alive(&participants, slot, &arena_boot);
let _topo = lock.acquire(self.participant, now_nanos(), &is_alive)?;
let (_p, _depth, edge, _gen) =
view.topology()
.read_frame(child)
.ok_or(ReparentError::Topology(TopologyError::UnknownFrame {
frame: child.get(),
}))?;
if edge == 0 {
return Err(ReparentError::NoEdge { child });
}
view.topology().set_parent(child, new_parent.get(), edge)?;
Ok(())
}
pub fn claim(&self, child: FrameId, parent: FrameId) -> Result<EdgeWriter<'_>, ClaimApiError> {
if self.detached() {
return Err(ClaimApiError::ChildDetached);
}
if !self.arena.is_writable() {
return Err(ClaimApiError::ReadOnly);
}
let view = self.view();
let (p, _depth, edge, _gen) = view
.topology()
.read_frame(child)
.ok_or(ClaimApiError::UnknownFrame { child })?;
if edge == 0 {
return Err(ClaimApiError::NoEdge { child });
}
if p != parent.get() {
return Err(ClaimApiError::ParentMismatch {
child,
expected: parent.get(),
actual: p,
});
}
let eid = EdgeId(edge);
let (Some(ring), Some(claim_rec)) = (view.ring(eid), view.claim(eid)) else {
return Err(ClaimApiError::NotDynamic { child, edge: eid });
};
let (epoch, owner) = claim(claim_rec, self.participant)?;
#[cfg(all(feature = "test-hooks", feature = "shm", target_os = "linux"))]
if let Some(hook) = CLAIM_WINDOW_HOOK.get() {
hook();
}
#[cfg(all(feature = "shm", target_os = "linux"))]
let lease = self.take_claim_lease(eid, claim_rec, epoch, owner)?;
#[cfg(all(feature = "shm", target_os = "linux"))]
self.populate_edge_rings(eid);
Ok(EdgeWriter {
publisher: Publisher::new(ring, claim_rec, epoch, owner),
#[cfg(all(feature = "shm", target_os = "linux"))]
fork_gen: self.fork_gen,
#[cfg(all(feature = "shm", target_os = "linux"))]
_lease: lease,
})
}
pub fn claim_owned(
self: &Arc<Tree>,
child: FrameId,
parent: FrameId,
) -> Result<OwnedWriter, ClaimApiError> {
#[allow(unsafe_code)]
let tree: &'static Tree = unsafe { &*Arc::as_ptr(self) };
let writer = tree.claim(child, parent)?;
Ok(OwnedWriter {
writer: Box::new(writer),
tree: Arc::clone(self),
})
}
#[cfg(all(feature = "shm", target_os = "linux"))]
fn take_claim_lease(
&self,
eid: EdgeId,
claim_rec: &tf_tree_core::edge::ClaimRecord,
epoch: u64,
owner: u64,
) -> Result<Option<ClaimLease>, ClaimApiError> {
let Some(lock) = self.claim_lock.as_ref() else {
return Ok(None);
};
match lock.try_take_claim(eid.0) {
Ok(tf_tree_ipc::LockAttempt::Acquired) => {}
Ok(tf_tree_ipc::LockAttempt::Contended) => {
tf_tree_core::edge::release(claim_rec, owner);
return Err(ClaimApiError::LeaseContended { edge: eid });
}
Err(_) => {
tf_tree_core::edge::release(claim_rec, owner);
return Err(ClaimApiError::LeaseUnavailable { edge: eid });
}
}
if claim_rec.epoch.load(Ordering::Acquire) != epoch {
tf_tree_core::edge::release(claim_rec, owner);
let _ = lock.release_claim(eid.0);
return Err(ClaimApiError::ReapedDuringClaim { edge: eid });
}
Ok(Some(ClaimLease {
lock: std::sync::Arc::clone(lock),
edge: eid.0,
fork_gen: tf_tree_ipc::fork::generation(),
}))
}
pub fn plan(
&self,
target: FrameId,
source: FrameId,
) -> Result<tf_tree_core::Plan, LookupError> {
if self.detached() {
return Err(LookupError::ChildDetached);
}
let view = self.view();
let topo = view.topology();
#[cfg(all(feature = "shm", target_os = "linux"))]
let edge_meta = |eid| {
self.populate_edge_rings(eid);
edge_meta(&view, eid)
};
#[cfg(not(all(feature = "shm", target_os = "linux")))]
let edge_meta = |eid| edge_meta(&view, eid);
compile(&topo, edge_meta, target, source)
}
#[cfg(all(feature = "shm", target_os = "linux"))]
fn populate_edge_rings(&self, eid: EdgeId) {
let ArenaBacking::Mapped(arena) = &self.arena else {
return;
};
if let Some(extents) = self.view().ring_extents(eid) {
for (off, len) in extents {
arena.populate(off, len);
}
}
}
#[must_use]
pub fn guard(&self) -> Guard<'_> {
if self.detached() {
return Guard::detached(self.view());
}
let g = Guard::new(self.view());
#[cfg(all(feature = "shm", target_os = "linux"))]
let g = if self.is_shared() {
g.with_fork_check(tf_tree_ipc::fork::generation)
} else {
g
};
g
}
pub fn lookup<D: Domain>(
&self,
target: &str,
source: &str,
stamp: Stamp<D>,
) -> Result<Iso3, LookupError> {
if self.detached() {
return Err(LookupError::ChildDetached);
}
let view = self.view();
let t = find(&view, target)?;
let s = find(&view, source)?;
let generation = view.topology().stable_generation();
let (plan, _hit) = cache::get_or_compile(self, t, s, generation)?;
let g = self.guard();
plan.at(&g, stamp)
}
#[must_use]
#[cfg(feature = "unstable")]
pub fn arena_view(&self) -> ArenaView<'_> {
self.view()
}
#[must_use]
pub fn arena_size_bytes(&self) -> usize {
self.view().header().arena_size as usize
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub fn attach_shared(fd: std::os::fd::OwnedFd, mode: AttachMode) -> Result<Tree, ShmError> {
Tree::attach_shared_inner(fd, mode, None)
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub fn attach_shared_at(
fd: std::os::fd::OwnedFd,
mode: AttachMode,
slot: u32,
) -> Result<Tree, ShmError> {
Tree::attach_shared_inner(fd, mode, Some(slot))
}
#[cfg(all(feature = "shm", target_os = "linux"))]
fn attach_shared_inner(
fd: std::os::fd::OwnedFd,
mode: AttachMode,
slot: Option<u32>,
) -> Result<Tree, ShmError> {
let arena = MappedArena::attach(fd, mode)?;
arena.populate_hot();
let backing = ArenaBacking::Mapped(arena);
let (participant, incarnation) = if backing.is_writable() {
let view = ArenaView::new(backing.as_dyn());
match slot {
Some(s) => (
s,
register_participant_at(&view, s)
.map_err(|_| ShmError::ParticipantTableFull)?,
),
None => register_participant(&view).map_err(|_| ShmError::ParticipantTableFull)?,
}
} else {
(u32::MAX, 0)
};
let liveness = liveness_for(ArenaView::new(backing.as_dyn()).header().boot_id);
#[cfg(all(feature = "shm", target_os = "linux"))]
let fork_gen = fork_gen_for(&backing);
Ok(Tree {
arena: backing,
participant,
incarnation,
liveness,
decl: Mutex::new(()),
#[cfg(all(feature = "shm", target_os = "linux"))]
attachment: None,
#[cfg(all(feature = "shm", target_os = "linux"))]
claim_lock: None,
#[cfg(all(feature = "shm", target_os = "linux"))]
fork_gen,
})
}
#[cfg(all(feature = "shm", target_os = "linux"))]
#[must_use]
pub fn shared_fd(&self) -> Option<std::os::fd::BorrowedFd<'_>> {
match &self.arena {
ArenaBacking::Mapped(a) => Some(a.as_raw_fd()),
ArenaBacking::Frozen(_) | ArenaBacking::Heap(_) => None,
}
}
#[must_use]
pub fn is_shared(&self) -> bool {
self.arena.is_shared()
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub(crate) fn from_frozen(arena: tf_tree_arena::FrozenArena) -> Tree {
let backing = ArenaBacking::Frozen(arena);
let fork_gen = fork_gen_for(&backing);
Tree {
arena: backing,
participant: u32::MAX,
incarnation: 0,
liveness: Box::new(|_, _| false),
decl: Mutex::new(()),
attachment: None,
claim_lock: None,
fork_gen,
}
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub(crate) fn backing(&self) -> &dyn Arena {
self.arena.as_dyn()
}
#[must_use]
pub fn boot_id(&self) -> [u8; 16] {
self.view().header().boot_id
}
#[must_use]
pub fn is_writable(&self) -> bool {
self.arena.is_writable()
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub(crate) fn hold_attachment(
&mut self,
session: crate::open::JoinedSession,
socket: std::os::fd::OwnedFd,
) {
self.attachment = Some(crate::open::Attachment::Joined {
_session: session,
_socket: socket,
});
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub(crate) fn hold_ownership(
&mut self,
session: crate::open::JoinedSession,
server: crate::open::OwnerThread,
) {
self.attachment = Some(crate::open::Attachment::Owner {
_session: session,
server,
});
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub(crate) fn use_ofd_liveness(&mut self, probe: crate::open::LivenessProbe) {
let own_slot = self.participant;
self.liveness = Box::new(move |slot, rec| {
if slot == own_slot {
return true;
}
probe.is_held(slot).unwrap_or_else(|| record_is_alive(rec))
});
}
#[cfg(all(feature = "shm", target_os = "linux"))]
pub(crate) fn use_claim_leases(&mut self, lock: std::sync::Arc<tf_tree_ipc::LockFile>) {
self.claim_lock = Some(lock);
}
#[cfg(all(feature = "shm", target_os = "linux"))]
#[must_use]
pub fn reap_dead(&self) -> usize {
self.reap_inner(None)
}
#[cfg(all(feature = "shm", target_os = "linux"))]
#[must_use]
pub fn reap_participant(&self, slot: u32) -> usize {
self.reap_inner(Some(slot))
}
#[cfg(all(feature = "shm", target_os = "linux"))]
fn reap_inner(&self, only_slot: Option<u32>) -> usize {
let Some(lock) = self.claim_lock.as_ref() else {
return 0; };
if self.participant == u32::MAX || !self.arena.is_writable() {
return 0;
}
let own_slot = self.participant;
let view = self.view();
let max_edges = view.header().max_edges;
let mut reaped = 0;
for edge in 0..max_edges {
let Some(rec) = view.claim(EdgeId(edge)) else {
continue;
};
let owner = rec.owner.load(Ordering::Acquire);
if owner == 0 {
continue;
}
let owner_slot = tf_tree_core::edge::slot_of(owner);
if owner_slot == own_slot {
continue;
}
if only_slot.is_some_and(|s| owner_slot != s) {
continue;
}
if lock.probe_claim(edge).map_or(true, |p| p.held) {
continue;
}
tf_tree_core::edge::reap(rec);
reaped += 1;
}
reaped
}
#[must_use]
pub fn participant_slot(&self) -> u32 {
self.participant
}
#[must_use]
pub fn participant_alive(&self, slot: u32) -> bool {
match self.view().participants().get(slot) {
None => false,
Some(rec) => {
tf_tree_core::participant::state_of(rec.state.load(Ordering::Acquire))
== tf_tree_core::participant::LIVE
&& (self.liveness)(slot, rec)
}
}
}
#[must_use]
pub fn instance_uuid(&self) -> [u8; 16] {
self.view().header().instance_uuid
}
#[must_use]
pub fn describe(&self, err: LookupError) -> Described<'_> {
Described(err, self)
}
fn frame_name(&self, id: FrameId) -> String {
let Some(rec) = self.view().frame_record(id) else {
return std::format!("frame#{}", id.get());
};
let n = rec.name_len as usize;
std::str::from_utf8(&rec.name[..n])
.unwrap_or("<invalid-utf8>")
.to_owned()
}
fn edge_name(&self, id: EdgeId) -> String {
let view = self.view();
let Some(rec) = view.edge(id) else {
return std::format!("edge#{}", id.get());
};
let parent = FrameId::new(rec.parent)
.map(|f| self.frame_name(f))
.unwrap_or_else(|| "<root>".to_owned());
let child = FrameId::new(rec.child)
.map(|f| self.frame_name(f))
.unwrap_or_else(|| "<root>".to_owned());
std::format!("{parent}->{child} (edge#{})", id.get())
}
}
fn find(view: &ArenaView, name: &str) -> Result<FrameId, LookupError> {
match view.find_frame(name) {
Ok(Some(id)) => Ok(id),
Ok(None) | Err(_) => Err(LookupError::UnknownFrame {
hash: blake3_64(name),
}),
}
}
fn edge_meta(view: &ArenaView, eid: EdgeId) -> Option<EdgeMeta> {
let e = view.edge(eid)?;
Some(EdgeMeta {
kind: EdgeKind::from_u8(e.kind),
domain: e.domain,
static_pose: Iso3::from_bits(&e.static_pose),
})
}
impl Drop for Tree {
fn drop(&mut self) {
if self.participant != u32::MAX && self.arena.is_writable() {
self.view()
.participants()
.release(self.participant, self.incarnation);
}
}
}
#[cfg(all(feature = "shm", target_os = "linux"))]
fn fork_gen_for(backing: &ArenaBacking) -> Option<u64> {
match backing {
ArenaBacking::Heap(_) => None,
ArenaBacking::Frozen(_) => None,
ArenaBacking::Mapped(_) => {
tf_tree_ipc::fork::arm();
let _ = poison_arena();
Some(tf_tree_ipc::fork::generation())
}
}
}
#[cfg(all(feature = "shm", target_os = "linux"))]
fn poison_arena() -> &'static HeapArena {
static POISON: std::sync::OnceLock<HeapArena> = std::sync::OnceLock::new();
POISON.get_or_init(|| {
HeapArena::new(&tf_tree_arena::ArenaLayout::minimal(), 0, 0, [0u8; 16])
})
}
fn register_participant(view: &ArenaView) -> Result<(u32, u64), ParticipantError> {
view.participants()
.register(std::process::id(), process_start_time(), now_nanos())
}
#[cfg(all(feature = "shm", target_os = "linux"))]
fn register_participant_at(view: &ArenaView, slot: u32) -> Result<u64, ParticipantError> {
view.participants()
.register_at(slot, std::process::id(), process_start_time(), now_nanos())
}
fn now_nanos() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| i64::try_from(d.as_nanos()).unwrap_or(i64::MAX))
}
fn record_is_alive(rec: &tf_tree_core::ParticipantRecord) -> bool {
use core::sync::atomic::Ordering;
if tf_tree_core::participant::state_of(rec.state.load(Ordering::Acquire))
!= tf_tree_core::participant::LIVE
{
return false;
}
let pid = rec.pid.load(Ordering::Relaxed);
let start_time = rec.start_time.load(Ordering::Relaxed);
match read_start_time(pid) {
ProcStartTime::Known(st) => st == start_time,
ProcStartTime::NoSuchProcess => false,
ProcStartTime::Unreadable => true,
}
}
fn liveness_for(arena_boot: [u8; 16]) -> BoxedLiveness {
let host = *host_boot_id();
if arena_boot != [0u8; 16] && host != [0u8; 16] && arena_boot != host {
return Box::new(|_, _| false);
}
Box::new(|_slot, rec| record_is_alive(rec))
}
fn participant_is_alive(
participants: &tf_tree_core::ParticipantTable<'_>,
slot: u32,
arena_boot: &[u8; 16],
) -> bool {
let host_boot = host_boot_id();
if *arena_boot != [0u8; 16] && *host_boot != [0u8; 16] && arena_boot != host_boot {
return false;
}
let Some((pid, start_time, _incarnation)) = participants.identity(slot) else {
return false;
};
match read_start_time(pid) {
ProcStartTime::Known(st) => st == start_time,
ProcStartTime::NoSuchProcess => false,
ProcStartTime::Unreadable => true,
}
}
enum ProcStartTime {
Known(u64),
NoSuchProcess,
Unreadable,
}
fn read_start_time(pid: u32) -> ProcStartTime {
let stat = match std::fs::read_to_string(std::format!("/proc/{pid}/stat")) {
Ok(s) => s,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return ProcStartTime::NoSuchProcess,
Err(_) => return ProcStartTime::Unreadable,
};
let Some(after_comm) = stat.rfind(')').map(|i| &stat[i + 1..]) else {
return ProcStartTime::Unreadable;
};
after_comm
.split_whitespace()
.nth(19)
.and_then(|v| v.parse().ok())
.map_or(ProcStartTime::Unreadable, ProcStartTime::Known)
}
fn host_boot_id() -> &'static [u8; 16] {
static ID: std::sync::OnceLock<[u8; 16]> = std::sync::OnceLock::new();
ID.get_or_init(boot_id)
}
fn boot_id() -> [u8; 16] {
let Ok(text) = std::fs::read_to_string("/proc/sys/kernel/random/boot_id") else {
return [0u8; 16];
};
let mut out = [0u8; 16];
let mut nibbles = text.trim().bytes().filter_map(|b| match b {
b'0'..=b'9' => Some(b - b'0'),
b'a'..=b'f' => Some(b - b'a' + 10),
b'A'..=b'F' => Some(b - b'A' + 10),
_ => None, });
for byte in &mut out {
let (Some(hi), Some(lo)) = (nibbles.next(), nibbles.next()) else {
return [0u8; 16]; };
*byte = (hi << 4) | lo;
}
out
}
fn process_start_time() -> u64 {
let Ok(stat) = std::fs::read_to_string("/proc/self/stat") else {
return 0;
};
let Some(after_comm) = stat.rfind(')').map(|i| &stat[i + 1..]) else {
return 0;
};
after_comm
.split_whitespace()
.nth(19)
.and_then(|v| v.parse().ok())
.unwrap_or(0)
}
pub struct Described<'a>(LookupError, &'a Tree);
impl fmt::Display for Described<'_> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let tree = self.1;
match self.0 {
LookupError::UnknownFrame { hash } => {
write!(f, "unknown frame (name hash {hash:#018x})")?;
const SHOWN: usize = 8;
match tree.frames() {
Ok(names) if names.is_empty() => write!(
f,
"; this tree has no frames yet, so no publisher has \
declared anything into it. Wait for one with \
Tree::await_frames, or declare the frame on the \
TreeBuilder that creates the arena"
),
Ok(mut names) => {
let total = names.len();
names.sort_unstable();
names.truncate(SHOWN);
f.write_str("; this tree has ")?;
for (i, n) in names.iter().enumerate() {
if i > 0 {
f.write_str(", ")?;
}
f.write_str(n)?;
}
if total > SHOWN {
write!(f, ", … ({total} total)")?;
}
write!(
f,
". If the name is spelled right, its publisher has \
not declared it yet: wait with Tree::await_frames, \
or declare it on the TreeBuilder that creates the \
arena"
)
}
Err(_) => write!(
f,
"; this tree was opened before a fork() and is being \
used in the child, so it can name nothing"
),
}
}
LookupError::Disconnected {
target,
source,
cut_at,
} => write!(
f,
"no path from {} to {}: disconnected at {}",
tree.frame_name(target),
tree.frame_name(source),
tree.frame_name(cut_at),
),
LookupError::TreeTooDeep { depth } => {
write!(f, "path depth {depth} exceeds the maximum of {MAX}", MAX = tf_tree_core::MAX_DEPTH)
}
LookupError::NoData { edge } => {
write!(f, "no samples on {}", tree.edge_name(edge))
}
LookupError::Extrapolation {
edge,
requested,
oldest,
newest,
} => write!(
f,
"lookup on {} would extrapolate: requested {requested} ns, history [{oldest}, {newest}] ns",
tree.edge_name(edge),
),
LookupError::SlotRecycled { edge } => {
write!(f, "the ring on {} lapped the reader mid-read", tree.edge_name(edge))
}
LookupError::SlotContended { edge } => {
write!(f, "a slot on {} stayed contended too long", tree.edge_name(edge))
}
LookupError::TopologyChanged { plan, current } => write!(
f,
"plan is stale: compiled at topology generation {plan}, current is {current} (re-plan)",
),
LookupError::TimeDomainMismatch { expected, got } => write!(
f,
"time-domain mismatch: plan expects domain {expected}, query supplied {got}",
),
LookupError::MixedTimeDomains {
edge,
expected,
got,
} => write!(
f,
"path crosses time domains: {} is in domain {got}, the rest of the path is in domain {expected}",
tree.edge_name(edge),
),
LookupError::UnknownEdge { edge } => {
write!(f, "{} names no usable edge in this tree", tree.edge_name(edge))
}
LookupError::FrameOutOfRange { frame } => write!(
f,
"frame id {} is out of range for this tree",
frame.get(),
),
LookupError::MissingEdge { child } => write!(
f,
"frame {} has a parent but no edge records the link",
tree.frame_name(child),
),
other => write!(f, "{other:?}"),
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum BuildError {
#[error("two edges declare the same child (name hash {child:#018x})")]
DuplicateEdge {
child: u64,
},
#[error("too many frames for the u32 id space")]
TooManyFrames,
#[error("too many edges for the u32 id space")]
TooManyEdges,
#[error("arena layout error: {0:?}")]
Layout(LayoutError),
#[error("frame error: {0:?}")]
Frame(FrameError),
#[error("topology error: {0:?}")]
Topology(TopologyError),
#[cfg(all(feature = "shm", target_os = "linux"))]
#[error("shared memory error: {0:?}")]
Shm(ShmError),
#[error("participant table full: {0:?}")]
Participant(ParticipantError),
}
impl From<LayoutError> for BuildError {
fn from(e: LayoutError) -> BuildError {
BuildError::Layout(e)
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum ReparentError {
#[error("this handle belongs to the pre-fork process; open a new tree in the child")]
ChildDetached,
#[error("frame {} has no edge to re-parent", child.get())]
NoEdge {
child: FrameId,
},
#[error("topology error: {0:?}")]
Topology(TopologyError),
#[error("arena is mapped read-only")]
ReadOnly,
#[error("the topology lock is held by live participant slot {owner_slot}")]
LockContended {
owner_slot: u32,
},
}
impl From<TopologyError> for ReparentError {
fn from(e: TopologyError) -> ReparentError {
ReparentError::Topology(e)
}
}
impl From<TopoLockError> for ReparentError {
fn from(e: TopoLockError) -> ReparentError {
match e {
TopoLockError::Contended { owner_slot } => ReparentError::LockContended { owner_slot },
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum ClaimApiError {
#[error("this handle belongs to the pre-fork process; open a new tree in the child")]
ChildDetached,
#[error("edge {edge:?}: the claim record was free but its lease is held")]
LeaseContended {
edge: EdgeId,
},
#[error("edge {edge:?}: the claim lease could not be taken")]
LeaseUnavailable {
edge: EdgeId,
},
#[error("edge {edge:?}: reaped while being claimed; retry")]
ReapedDuringClaim {
edge: EdgeId,
},
#[error("frame {} is not a frame of this tree", child.get())]
UnknownFrame {
child: FrameId,
},
#[error("no edge attaches child frame {}", child.get())]
NoEdge {
child: FrameId,
},
#[error("edge#{} attaching frame {} is not a dynamic edge", edge.get(), child.get())]
NotDynamic {
child: FrameId,
edge: EdgeId,
},
#[error("child frame {} is attached to {actual}, not the requested {expected}", child.get())]
ParentMismatch {
child: FrameId,
expected: u32,
actual: u32,
},
#[error("edge already claimed by participant slot {}", .0.owner_slot())]
AlreadyClaimed(tf_tree_core::ClaimError),
#[error("arena is mapped read-only")]
ReadOnly,
}
impl From<tf_tree_core::ClaimError> for ClaimApiError {
fn from(e: tf_tree_core::ClaimError) -> ClaimApiError {
ClaimApiError::AlreadyClaimed(e)
}
}
trait ClaimErrorExt {
fn owner_slot(&self) -> u32;
}
impl ClaimErrorExt for tf_tree_core::ClaimError {
fn owner_slot(&self) -> u32 {
match self {
tf_tree_core::ClaimError::EdgeAlreadyClaimed { owner_slot } => *owner_slot,
_ => 0,
}
}
}