#![cfg(all(feature = "shm", target_os = "linux"))]
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use std::io::{BufRead, BufReader};
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
struct Scratch(PathBuf);
impl Scratch {
fn new(tag: &str) -> Scratch {
let p = std::env::temp_dir().join(format!("tf_tree_rv-{}-{tag}", std::process::id()));
let _ = std::fs::remove_dir_all(&p);
std::fs::create_dir_all(&p).unwrap();
std::env::set_var("TF_TREE_RUNTIME_DIR", &p);
Scratch(p)
}
}
impl Drop for Scratch {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.0);
}
}
struct Kid(Child, Option<BufReader<std::process::ChildStdout>>);
impl Kid {
fn spawn(dir: &PathBuf, args: &[&str]) -> Kid {
let exe = env!("CARGO_BIN_EXE_tf_tree_rendezvous_child");
let child = Command::new(exe)
.args(args)
.env("TF_TREE_RUNTIME_DIR", dir)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::inherit())
.spawn()
.expect("spawn the rendezvous child helper");
Kid(child, None)
}
fn line(&mut self) -> String {
let reader = self
.1
.get_or_insert_with(|| BufReader::new(self.0.stdout.take().expect("piped stdout")));
let mut line = String::new();
reader.read_line(&mut line).expect("read child line");
line.trim_end().to_string()
}
fn poke(&mut self) {
use std::io::Write;
if let Some(mut stdin) = self.0.stdin.take() {
let _ = writeln!(stdin, "go");
}
}
fn kill(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
impl Drop for Kid {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
#[test]
fn a_foreign_process_joins_and_reads_the_same_transform() {
let scratch = Scratch::new("join");
let mut owner = Kid::spawn(&scratch.0, &["own"]);
let published = owner.line();
assert!(published.starts_with("owning "), "got {published}");
let owner_value = published.strip_prefix("owning ").unwrap().to_string();
let mut joiner = Kid::spawn(&scratch.0, &["join"]);
let joined = joiner.line();
assert!(
joined.starts_with("joined "),
"the joiner did not attach: {joined}"
);
let joiner_value = joined.strip_prefix("joined ").unwrap().to_string();
assert_eq!(
joiner_value, owner_value,
"the joiner read a different transform than the owner published"
);
}
#[test]
fn a_consumer_that_will_not_create_fails_fast_on_an_empty_machine() {
let scratch = Scratch::new("never");
let mut kid = Kid::spawn(&scratch.0, &["join"]);
let line = kid.line();
assert!(
line.starts_with("error"),
"expected a fast failure, got {line}"
);
assert!(
line.contains("no arena"),
"the error should name the absent arena: {line}"
);
}
#[test]
fn the_free_open_joins_a_served_arena() {
let scratch = Scratch::new("free-open");
let mut owner = Kid::spawn(&scratch.0, &["own"]);
let published = owner.line();
assert!(published.starts_with("owning "), "got {published}");
let owner_value = published.strip_prefix("owning ").unwrap().to_string();
let mut joiner = Kid::spawn(&scratch.0, &["open-free"]);
let joined = joiner.line();
assert!(
joined.starts_with("joined "),
"tf_tree::open() did not join a served arena: {joined}"
);
assert_eq!(
joined.strip_prefix("joined ").unwrap(),
owner_value,
"tf_tree::open() read a different transform than the owner published"
);
}
#[test]
fn a_read_only_attach_refuses_to_create() {
use tf_tree::{AttachMode, Capacity, CreatePolicy, EdgeCfg, InterpPolicy, TreeBuilder};
let _scratch = Scratch::new("ro-create");
let layout = || {
TreeBuilder::new()
.default_interp(InterpPolicy::LerpSlerp)
.dynamic_edge("map", "base", EdgeCfg::new(Capacity::slots(64)))
};
for policy in [CreatePolicy::IfAbsent, CreatePolicy::Always] {
let err = tf_tree::Open::new()
.mode(AttachMode::ReadOnly)
.create(policy)
.layout_if_creating(layout())
.open()
.err()
.expect("a read-only creator must be refused");
assert!(
matches!(err, tf_tree::OpenError::ReadOnlyCannotCreate),
"expected ReadOnlyCannotCreate for {policy:?}, got {err:?}"
);
}
let err = tf_tree::Open::new()
.create(CreatePolicy::Never)
.open()
.err()
.expect("nothing should have been created");
assert!(
matches!(
err,
tf_tree::OpenError::Rendezvous(tf_tree_ipc::IpcError::ArenaAbsent)
),
"the refused open left an arena behind: {err:?}"
);
}
#[test]
fn require_create_refuses_a_live_arena_and_releases_its_slot() {
use tf_tree::{AttachMode, Capacity, CreatePolicy, EdgeCfg, InterpPolicy, TreeBuilder};
let scratch = Scratch::new("require-create");
let mut owner = Kid::spawn(&scratch.0, &["own"]);
assert!(owner.line().starts_with("owning "));
let err = tf_tree::Open::new()
.mode(AttachMode::ReadWrite)
.create(CreatePolicy::IfAbsent)
.require_create(true)
.layout_if_creating(
TreeBuilder::new()
.default_interp(InterpPolicy::LerpSlerp)
.dynamic_edge("map", "base", EdgeCfg::new(Capacity::slots(64))),
)
.open()
.err()
.expect("a second owner must not silently join");
assert!(
matches!(err, tf_tree::OpenError::ArenaAlreadyLive),
"expected ArenaAlreadyLive, got {err:?}"
);
let lock = tf_tree_ipc::LockFile::open(&scratch.0.join("0/default.lock")).unwrap();
assert!(
!lock.probe_participant(1).unwrap().held,
"the refused attach kept its participant lock byte"
);
let mut joiner = Kid::spawn(&scratch.0, &["join"]);
assert!(
joiner.line().starts_with("joined "),
"the refusal disturbed the arena"
);
}
#[test]
fn a_consumer_waits_for_an_arena_that_starts_late() {
use std::sync::mpsc;
use std::time::{Duration, Instant};
let scratch = Scratch::new("late-start");
let dir = scratch.0.clone();
let (tx, rx) = mpsc::channel::<Kid>();
let spawner = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(200));
let mut owner = Kid::spawn(&dir, &["own"]);
assert!(
owner.line().starts_with("owning "),
"the owner did not start"
);
let _ = tx.send(owner);
});
let started = Instant::now();
let tree = tf_tree::Open::new()
.await_open(Duration::from_secs(20))
.expect("the wait should have outlasted a publisher 200 ms late");
let elapsed = started.elapsed();
assert!(
elapsed >= Duration::from_millis(200),
"the wait returned before the publisher could have started ({elapsed:?}) — \
it did not actually wait"
);
assert!(
elapsed < Duration::from_secs(10),
"the wait took far longer than the publisher's 200 ms delay: {elapsed:?}"
);
assert!(!tree.is_writable(), "the default attach is read-only (D18)");
let owner = rx
.recv_timeout(Duration::from_secs(20))
.expect("the spawner thread never produced an owner");
drop(tree);
drop(owner);
spawner.join().expect("spawner thread");
}
#[test]
fn a_wait_for_an_arena_that_never_starts_gives_up() {
use std::sync::mpsc;
use std::time::{Duration, Instant};
let _scratch = Scratch::new("never-starts");
let budget = Duration::from_millis(300);
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
let started = Instant::now();
let outcome = tf_tree::Open::new().await_open(budget);
let _ = tx.send((outcome.err(), started.elapsed()));
});
let (err, elapsed) = rx.recv_timeout(Duration::from_secs(30)).expect(
"await_open never returned: it ignored its deadline. There is no \
.config/nextest.toml in this repository, so nothing else would have \
bounded this",
);
let err = err.expect("an empty machine has no arena to open");
assert!(
matches!(
err,
tf_tree::OpenError::Rendezvous(tf_tree_ipc::IpcError::ArenaAbsent)
| tf_tree::OpenError::Rendezvous(
tf_tree_ipc::IpcError::ArenaHeldButUnreachable { .. }
)
),
"expected the last retryable rendezvous error, got {err:?}"
);
assert!(elapsed >= budget, "it gave up early: {elapsed:?}");
assert!(
elapsed < budget * 20,
"it overran its budget by more than the backoff can explain: {elapsed:?}"
);
}
#[test]
fn a_consumer_waits_for_a_frame_interned_after_the_arena_exists() {
use std::io::Write;
use std::time::{Duration, Instant};
let scratch = Scratch::new("late-frame");
let mut owner = Kid::spawn(&scratch.0, &["own-headroom"]);
assert_eq!(owner.line(), "owning");
let consumer = tf_tree::Open::new()
.open()
.expect("join the arena the owner already created");
assert!(!consumer.is_writable(), "the default attach is read-only");
assert!(
consumer.frames().unwrap().iter().all(|n| n != "late_frame"),
"the frame under test was already interned before the wait began"
);
let mut stdin = owner.0.stdin.take().expect("piped stdin");
let poker = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(200));
let _ = writeln!(stdin, "go");
});
let started = Instant::now();
let [late] = consumer
.await_frames(["late_frame"], Duration::from_secs(20))
.expect("the frame was interned well inside the budget");
let elapsed = started.elapsed();
poker.join().expect("poker thread");
let interned = owner.line();
let owner_id: u32 = interned
.strip_prefix("interned ")
.expect(&interned)
.parse()
.unwrap();
assert_eq!(
late.get(),
owner_id,
"the waiter resolved to a different id than the owner interned"
);
assert!(
elapsed >= Duration::from_millis(200),
"the wait returned before the owner could have interned ({elapsed:?})"
);
assert!(
elapsed < Duration::from_secs(10),
"the wait far outlasted the intern it was waiting for: {elapsed:?}"
);
}
#[test]
fn a_frames_wait_for_a_name_nobody_will_intern_gives_up() {
use std::sync::mpsc;
use std::time::{Duration, Instant};
use tf_tree::AwaitError;
use tf_tree_core::frame::blake3_64;
const MISSING: &str = "no_publisher_will_ever_declare_this";
let scratch = Scratch::new("frames-timeout");
let mut owner = Kid::spawn(&scratch.0, &["own"]);
assert!(
owner.line().starts_with("owning "),
"the owner did not start"
);
let budget = Duration::from_millis(300);
let (tx, rx) = mpsc::channel();
std::thread::spawn(move || {
let consumer = tf_tree::Open::new()
.open()
.expect("join the arena the owner is serving");
let writable = consumer.is_writable();
let empty = consumer.await_frames([], Duration::from_millis(0));
let started = Instant::now();
let outcome = consumer.await_frames(["map", MISSING], budget);
let _ = tx.send((writable, empty, outcome, started.elapsed()));
});
let (writable, empty, outcome, elapsed) = rx.recv_timeout(Duration::from_secs(30)).expect(
"await_frames never returned: it ignored its deadline. There is no \
.config/nextest.toml in this repository, so nothing else would have \
bounded this",
);
assert!(!writable, "the default attach must be read-only (D18)");
assert_eq!(
empty,
Ok([]),
"a zero-length request on a waitable tree must be answered without \
touching the arena"
);
let err = outcome.expect_err("a name nobody interned must not resolve");
assert_eq!(
err,
AwaitError::Timeout {
hash: blake3_64(MISSING)
},
"the timeout named the wrong frame, or was not a timeout at all"
);
assert!(elapsed >= budget, "it gave up early: {elapsed:?}");
assert!(
elapsed < budget * 20,
"it overran its budget by more than the backoff can explain: {elapsed:?}"
);
}
#[test]
fn a_stopped_peer_is_alive_and_a_killed_one_is_not() {
let scratch = Scratch::new("liveness");
let mut owner = Kid::spawn(&scratch.0, &["own"]);
assert!(owner.line().starts_with("owning "));
let mut peer = Kid::spawn(&scratch.0, &["join-rw"]);
assert!(peer.line().starts_with("joined "), "peer did not join");
let observer_alive = |scratch: &PathBuf| {
let mut k = Kid::spawn(scratch, &["peer-alive", "1"]);
let line = k.line();
k.kill();
line
};
let pid = peer.0.id();
assert!(
std::process::Command::new("kill")
.args(["-STOP", &pid.to_string()])
.status()
.is_ok_and(|s| s.success()),
"could not SIGSTOP the peer"
);
assert_eq!(
observer_alive(&scratch.0),
"alive true",
"a SIGSTOPped participant was reported dead — a slow publisher must \
never be mistaken for a hung one (D17)"
);
let _ = std::process::Command::new("kill")
.args(["-CONT", &pid.to_string()])
.status();
peer.kill();
assert_eq!(
observer_alive(&scratch.0),
"alive false",
"a SIGKILLed participant was still reported alive"
);
}
#[test]
fn a_read_only_peer_holds_a_byte_without_an_arena_record() {
let scratch = Scratch::new("ro-slot");
let mut owner = Kid::spawn(&scratch.0, &["own"]);
assert!(owner.line().starts_with("owning "));
let mut ro = Kid::spawn(&scratch.0, &["join"]);
assert!(
ro.line().starts_with("joined "),
"read-only peer did not join"
);
let mut rw = Kid::spawn(&scratch.0, &["join-rw"]);
assert!(rw.line().starts_with("joined "), "second peer did not join");
let mut probe = Kid::spawn(&scratch.0, &["peer-alive", "1"]);
let slot1 = probe.line();
probe.kill();
let mut probe2 = Kid::spawn(&scratch.0, &["peer-alive", "2"]);
let slot2 = probe2.line();
probe2.kill();
assert_eq!(
(slot1.as_str(), slot2.as_str()),
("alive false", "alive true"),
"the byte/record asymmetry changed: slot 1 should hold a read-only \
peer's lock byte with no arena record, slot 2 a registered one"
);
}
#[test]
fn a_claim_takes_a_lease_and_a_dead_holder_releases_it() {
let scratch = Scratch::new("claim-lease-e2e");
let mut owner = Kid::spawn(&scratch.0, &["own-claiming"]);
let line = owner.line();
assert!(line.starts_with("claimed "), "got {line}");
let edge: u32 = line.strip_prefix("claimed ").unwrap().parse().unwrap();
let lock = tf_tree_ipc::LockFile::open(&scratch.0.join("0/default.lock")).unwrap();
assert!(
lock.probe_claim(edge).unwrap().held,
"claiming through open() did not take the edge's lease"
);
owner.kill();
assert!(
!lock.probe_claim(edge).unwrap().held,
"the lease outlived its holder: a dead writer would leak its edge"
);
}
#[test]
fn a_reaper_does_not_reap_its_own_live_claim() {
let scratch = Scratch::new("self-reap");
let mut kid = Kid::spawn(&scratch.0, &["own-reap"]);
assert_eq!(kid.line(), "claimed");
kid.poke();
assert_eq!(
kid.line(),
"reaped 0 still_ours true",
"the reaper revoked its own live claim — F_OFD_GETLK does not report a \
description's own locks, so every edge this process holds reads free"
);
}
#[test]
fn a_killed_writers_edge_is_reaped_and_can_be_reclaimed() {
let scratch = Scratch::new("reap-dead");
let mut owner = Kid::spawn(&scratch.0, &["own-reap"]);
assert_eq!(owner.line(), "claimed");
let mut peer = Kid::spawn(&scratch.0, &["join-claiming"]);
let claimed = peer.line();
assert!(
claimed.starts_with("claimed "),
"peer did not claim: {claimed}"
);
peer.kill();
owner.poke();
let line = owner.line();
assert!(
line.starts_with("reaped 1 ") && line.ends_with("still_ours true"),
"expected exactly the dead peer's edge to be reaped, got {line}"
);
}
#[cfg(feature = "test-hooks")]
static REAPER: std::sync::OnceLock<tf_tree::Tree> = std::sync::OnceLock::new();
#[cfg(feature = "test-hooks")]
static ARMED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true);
#[cfg(feature = "test-hooks")]
static REAPED: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
#[cfg(feature = "test-hooks")]
fn reap_from_inside_the_window() {
use std::sync::atomic::Ordering;
if !ARMED.swap(false, Ordering::Relaxed) {
return;
}
if let Some(t) = REAPER.get() {
REAPED.fetch_add(t.reap_dead(), Ordering::Relaxed);
}
}
#[test]
#[cfg(feature = "test-hooks")]
fn the_acquire_window_backs_out() {
use std::sync::atomic::Ordering;
use tf_tree::{AttachMode, Capacity, CreatePolicy, EdgeCfg, InterpPolicy, TreeBuilder};
let _scratch = Scratch::new("acquire-window");
let claimer = tf_tree::Open::new()
.mode(AttachMode::ReadWrite)
.create(CreatePolicy::IfAbsent)
.layout_if_creating(
TreeBuilder::new()
.default_interp(InterpPolicy::LerpSlerp)
.dynamic_edge("map", "base", EdgeCfg::new(Capacity::slots(64))),
)
.open()
.expect("create");
let reaper = tf_tree::Open::new()
.mode(AttachMode::ReadWrite)
.create(CreatePolicy::Never)
.open()
.expect("join as a second read-write participant");
assert_ne!(
claimer.participant_slot(),
reaper.participant_slot(),
"both handles took the same slot, so the reaper would skip the claim as its own"
);
REAPER.set(reaper).ok().expect("set reaper");
tf_tree::CLAIM_WINDOW_HOOK
.set(reap_from_inside_the_window as fn())
.ok()
.expect("install hook");
let child = claimer.frame("base").unwrap();
let parent = claimer.frame("map").unwrap();
let err = claimer
.claim(child, parent)
.err()
.expect("a claim reaped inside its own acquire window must not succeed");
assert!(
matches!(err, tf_tree::ClaimApiError::ReapedDuringClaim { .. }),
"expected ReapedDuringClaim, got {err:?}"
);
assert_eq!(
REAPED.load(Ordering::Relaxed),
1,
"the hook did not actually reap anything, so the guard was never exercised"
);
let writer = claimer
.claim(child, parent)
.expect("the retry after ReapedDuringClaim must succeed");
writer
.push(
1_000,
&tf_tree_math::exp_se3([0.0, 0.0, 0.1, 1.0, 0.0, 0.0]),
)
.expect("and the reclaimed edge must be publishable");
}