feat(registry): named pid registry as a bimap
register/whereis/unregister plus the inverse lookup name_of. Erlang semantics: one name per pid, one pid per name, registering over a live binding errors; same-binding re-register is an idempotent Ok. Bimap = two HashMaps held in exact inverse under one Leaf RawMutex in RuntimeInner; the invariant is debug-asserted on every mutation. The inverse direction costs one extra String per binding and buys O(1) pid->name for supervisors/tracing/diagnostics. Cleanup is lazy: no finalize_actor hook, no Slot field (which would buy into the reset-in-three-places invariant). Liveness rides the generation-checked slot word - pids are never reused, so a stale binding is detectable, never misdirected - and bindings to dead actors behave as absent, pruned on contact by whereis/name_of/register/unregister. Liveness checks under the registry lock read only the atomic slot word, so the leaf rule holds trivially. send_by_name is deliberately absent: smarm has no per-pid mailbox, so there is nothing generic to send *to* - the registry yields a Pid, usable with monitor/link/request_stop and as routing for user channels. Tests: roundtrip both directions, idempotence, NameTaken on live holder, one-name-per-pid, NoProc, death-evaporates-binding (via lookup pruning and via register's own eviction path), unregister, cross-actor lookup wired into request_stop, and a 4-scheduler name-contention churn test.
This commit is contained in:
smarm
@@ -0,0 +1,221 @@
|
||||
//! Named registry tests. Run under the scheduler: registration requires a
|
||||
//! live runtime and live actors.
|
||||
|
||||
use smarm::{channel, name_of, register, run, spawn, unregister, whereis, RegisterError};
|
||||
|
||||
#[test]
|
||||
fn register_whereis_roundtrip() {
|
||||
run(|| {
|
||||
let (tx, rx) = channel::<()>();
|
||||
let h = spawn(move || {
|
||||
rx.recv().unwrap();
|
||||
});
|
||||
register("worker", h.pid()).unwrap();
|
||||
assert_eq!(whereis("worker"), Some(h.pid()));
|
||||
assert_eq!(name_of(h.pid()).as_deref(), Some("worker"));
|
||||
tx.send(()).unwrap();
|
||||
h.join().unwrap();
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn register_is_idempotent_for_same_binding() {
|
||||
run(|| {
|
||||
let (tx, rx) = channel::<()>();
|
||||
let h = spawn(move || {
|
||||
rx.recv().unwrap();
|
||||
});
|
||||
register("svc", h.pid()).unwrap();
|
||||
// Same name, same pid: a no-op Ok, not NameTaken.
|
||||
register("svc", h.pid()).unwrap();
|
||||
tx.send(()).unwrap();
|
||||
h.join().unwrap();
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn duplicate_name_on_live_holder_is_rejected() {
|
||||
run(|| {
|
||||
let (tx, rx) = channel::<()>();
|
||||
let (tx2, rx2) = channel::<()>();
|
||||
let a = spawn(move || {
|
||||
rx.recv().unwrap();
|
||||
});
|
||||
let b = spawn(move || {
|
||||
rx2.recv().unwrap();
|
||||
});
|
||||
register("svc", a.pid()).unwrap();
|
||||
assert_eq!(
|
||||
register("svc", b.pid()),
|
||||
Err(RegisterError::NameTaken { holder: a.pid() })
|
||||
);
|
||||
tx.send(()).unwrap();
|
||||
tx2.send(()).unwrap();
|
||||
a.join().unwrap();
|
||||
b.join().unwrap();
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn one_name_per_pid() {
|
||||
run(|| {
|
||||
let (tx, rx) = channel::<()>();
|
||||
let h = spawn(move || {
|
||||
rx.recv().unwrap();
|
||||
});
|
||||
register("first", h.pid()).unwrap();
|
||||
assert_eq!(
|
||||
register("second", h.pid()),
|
||||
Err(RegisterError::PidAlreadyRegistered { name: "first".into() })
|
||||
);
|
||||
tx.send(()).unwrap();
|
||||
h.join().unwrap();
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn registering_a_dead_pid_is_noproc() {
|
||||
run(|| {
|
||||
let h = spawn(|| {});
|
||||
let pid = h.pid();
|
||||
h.join().unwrap();
|
||||
assert_eq!(register("ghost", pid), Err(RegisterError::NoProc));
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn binding_evaporates_on_death_and_name_is_reusable() {
|
||||
run(|| {
|
||||
let a = spawn(|| {});
|
||||
let pid_a = a.pid();
|
||||
register("svc", pid_a).unwrap();
|
||||
a.join().unwrap();
|
||||
|
||||
// Dead holder: both lookup directions report unbound.
|
||||
assert_eq!(whereis("svc"), None);
|
||||
assert_eq!(name_of(pid_a), None);
|
||||
|
||||
// And the name is free for a successor.
|
||||
let (tx, rx) = channel::<()>();
|
||||
let b = spawn(move || {
|
||||
rx.recv().unwrap();
|
||||
});
|
||||
register("svc", b.pid()).unwrap();
|
||||
assert_eq!(whereis("svc"), Some(b.pid()));
|
||||
tx.send(()).unwrap();
|
||||
b.join().unwrap();
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn register_over_a_dead_holder_succeeds_without_lookup_in_between() {
|
||||
// The eviction path inside register() itself (not via whereis pruning).
|
||||
run(|| {
|
||||
let a = spawn(|| {});
|
||||
let pid_a = a.pid();
|
||||
register("svc", pid_a).unwrap();
|
||||
a.join().unwrap();
|
||||
|
||||
let (tx, rx) = channel::<()>();
|
||||
let b = spawn(move || {
|
||||
rx.recv().unwrap();
|
||||
});
|
||||
register("svc", b.pid()).unwrap();
|
||||
assert_eq!(whereis("svc"), Some(b.pid()));
|
||||
assert_eq!(name_of(pid_a), None);
|
||||
tx.send(()).unwrap();
|
||||
b.join().unwrap();
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unregister_frees_both_directions() {
|
||||
run(|| {
|
||||
let (tx, rx) = channel::<()>();
|
||||
let h = spawn(move || {
|
||||
rx.recv().unwrap();
|
||||
});
|
||||
register("svc", h.pid()).unwrap();
|
||||
assert_eq!(unregister("svc"), Some(h.pid()));
|
||||
assert_eq!(whereis("svc"), None);
|
||||
assert_eq!(name_of(h.pid()), None);
|
||||
assert_eq!(unregister("svc"), None);
|
||||
|
||||
// The pid may take a new name afterwards.
|
||||
register("svc2", h.pid()).unwrap();
|
||||
assert_eq!(name_of(h.pid()).as_deref(), Some("svc2"));
|
||||
tx.send(()).unwrap();
|
||||
h.join().unwrap();
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn whereis_from_another_actor_and_usable_with_runtime_apis() {
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
let stopped = Arc::new(AtomicBool::new(false));
|
||||
let stopped2 = stopped.clone();
|
||||
run(move || {
|
||||
let (tx, rx) = channel::<()>();
|
||||
let svc = spawn(move || {
|
||||
// Parks forever; only a request_stop ends it.
|
||||
let _ = rx.recv();
|
||||
});
|
||||
register("stoppable", svc.pid()).unwrap();
|
||||
|
||||
let stopped3 = stopped2.clone();
|
||||
let client = spawn(move || {
|
||||
let pid = whereis("stoppable").expect("name must resolve cross-actor");
|
||||
// The registry's pids plug into the rest of the runtime API.
|
||||
smarm::request_stop(pid);
|
||||
stopped3.store(true, Ordering::Relaxed);
|
||||
});
|
||||
client.join().unwrap();
|
||||
svc.join().unwrap(); // a stopped actor joins Ok
|
||||
drop(tx);
|
||||
});
|
||||
assert!(stopped.load(Ordering::Relaxed));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn registry_under_concurrent_churn_multi_thread() {
|
||||
// Many actors racing to claim the same names while holders die: the
|
||||
// bimap invariant (debug-asserted internally) and Erlang semantics must
|
||||
// hold under real parallelism.
|
||||
use std::sync::atomic::{AtomicU32, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
let wins = Arc::new(AtomicU32::new(0));
|
||||
let wins2 = wins.clone();
|
||||
smarm::init(smarm::Config::exact(4)).run(move || {
|
||||
let mut handles = Vec::new();
|
||||
for round in 0..8 {
|
||||
let name = format!("contested-{}", round % 2);
|
||||
for _ in 0..8 {
|
||||
let name = name.clone();
|
||||
let wins = wins2.clone();
|
||||
handles.push(spawn(move || {
|
||||
let me = smarm::self_pid();
|
||||
match register(&name, me) {
|
||||
Ok(()) => {
|
||||
wins.fetch_add(1, Ordering::Relaxed);
|
||||
smarm::yield_now();
|
||||
// May have been pruned-by-contact never; we are
|
||||
// alive, so our binding must still resolve to us.
|
||||
assert_eq!(whereis(&name), Some(me));
|
||||
assert_eq!(unregister(&name), Some(me));
|
||||
}
|
||||
Err(RegisterError::NameTaken { .. }) => {}
|
||||
Err(e) => panic!("unexpected register error: {e}"),
|
||||
}
|
||||
}));
|
||||
}
|
||||
}
|
||||
for h in handles {
|
||||
h.join().unwrap();
|
||||
}
|
||||
});
|
||||
// At least one registration per name must have succeeded.
|
||||
assert!(wins.load(Ordering::Relaxed) >= 2);
|
||||
}
|
||||
Reference in New Issue
Block a user