monitor: widen stamp eligibility to watchable = named ∪ exported (soak sig 5)
The terminal record existed for watches that raced their target's death, but
e43c673 scoped its stamp to named tenancies — and the pid-identity watch
surface (§4 Slice 3) targets arbitrary actors, including anonymous ones whose
pids cross the boundary in contract replies. The first wild pid-face hit
(width-20 soak, pid_watch_test.exs:47, 1/600 full-suite: a monitor installed
while the child was alive delivered :noproc instead of {:smarm_exit, :panic})
is exactly the residual a0ba9be's commit body deferred.
ever_named becomes `watchable`, with a second set-site: mark_watchable(pid),
which the bridge calls wherever a smarm pid is encoded across the boundary —
BEAM can only watch pids it holds, and can only hold pids that crossed.
Anonymous never-exported churn (holder threads, egress tasks) stays
ineligible, preserving e43c673's LIFO-eviction protection unchanged.
mark_watchable takes the cold lock before the liveness screen: finalize
publishes Done and reads the bit under the same lock, so the mark either
lands before the death stamps or observes the tenancy dead and no-ops —
no lost-stamp window, and marking a corpse cannot invent history (pinned
in the test alongside the mark-while-alive stamp).
This commit is contained in:
+1
-1
@@ -72,7 +72,7 @@ pub use introspect::{StackInfo,
|
|||||||
#[cfg(feature = "observer")]
|
#[cfg(feature = "observer")]
|
||||||
pub use observer::{ObserverReply, ObserverRequest};
|
pub use observer::{ObserverReply, ObserverRequest};
|
||||||
pub use link::{link, trap_exit, unlink, ExitSignal};
|
pub use link::{link, trap_exit, unlink, ExitSignal};
|
||||||
pub use monitor::{demonitor, monitor, terminal_reason, Down, DownReason, Monitor, MonitorId};
|
pub use monitor::{demonitor, mark_watchable, monitor, terminal_reason, Down, DownReason, Monitor, MonitorId};
|
||||||
pub use mutex::{LockTimeout, Mutex, MutexGuard};
|
pub use mutex::{LockTimeout, Mutex, MutexGuard};
|
||||||
pub use pid::{Addressable, Erased, Name, Pid, RawPid};
|
pub use pid::{Addressable, Erased, Name, Pid, RawPid};
|
||||||
pub use pg::{dispatch, join, leave, members, members_as, pick, pick_as, Incarnation, Member, NodeId};
|
pub use pg::{dispatch, join, leave, members, members_as, pick, pick_as, Incarnation, Member, NodeId};
|
||||||
|
|||||||
@@ -177,6 +177,36 @@ pub fn monitor<A>(target: Pid<A>) -> Monitor {
|
|||||||
Monitor { id, target, rx }
|
Monitor { id, target, rx }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Flag `target`'s tenancy as watchable: its death will stamp the slot's
|
||||||
|
/// terminal record (see [`terminal_reason`]), exactly as registering a name
|
||||||
|
/// does. The bridge calls this wherever a smarm pid is *encoded across the
|
||||||
|
/// boundary* — a contract reply, an introspection listing — because BEAM can
|
||||||
|
/// only watch pids it holds, and can only hold pids that crossed. Keeping the
|
||||||
|
/// bit rare is what keeps the record alive: anonymous never-exported churn
|
||||||
|
/// (holder threads, egress tasks) stays ineligible and cannot evict a
|
||||||
|
/// watchable tenancy's record from a LIFO-recycled slot.
|
||||||
|
///
|
||||||
|
/// Generation-checked and live-screened: marking a pid whose tenancy already
|
||||||
|
/// ended is a no-op — its record either exists (it was flagged before dying)
|
||||||
|
/// or is honestly unknowable. Same `Runtime::run()` context contract as
|
||||||
|
/// [`monitor`].
|
||||||
|
pub fn mark_watchable<A>(target: Pid<A>) {
|
||||||
|
let target = target.erase();
|
||||||
|
with_runtime(|inner| {
|
||||||
|
if let Some(slot) = inner.slot_at(target) {
|
||||||
|
// Cold lock FIRST: finalize publishes Done and checks the
|
||||||
|
// watchable bit under this same lock, so the mark either lands
|
||||||
|
// before finalize reads it (the death stamps) or observes the
|
||||||
|
// tenancy already dead (no-op). No lost-stamp window between an
|
||||||
|
// unlocked liveness read and the flag set.
|
||||||
|
let mut cold = slot.cold.lock();
|
||||||
|
if slot.is_live_for(target) {
|
||||||
|
cold.watchable = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
/// The terminal [`DownReason`] of the tenancy `target` names, if that tenancy
|
/// The terminal [`DownReason`] of the tenancy `target` names, if that tenancy
|
||||||
/// ever registered a name and is the *most recent named* death of its slot:
|
/// ever registered a name and is the *most recent named* death of its slot:
|
||||||
/// finalize stamps the slot with `(generation, reason)` for once-registered
|
/// finalize stamps the slot with `(generation, reason)` for once-registered
|
||||||
|
|||||||
+1
-1
@@ -395,7 +395,7 @@ pub(crate) fn register_with<M: Send + 'static>(
|
|||||||
// `me` is screened by the same live() the binding requires below.
|
// `me` is screened by the same live() the binding requires below.
|
||||||
if live(inner, me) {
|
if live(inner, me) {
|
||||||
if let Some(slot) = inner.slot_at(me) {
|
if let Some(slot) = inner.slot_at(me) {
|
||||||
slot.cold.lock().ever_named = true;
|
slot.cold.lock().watchable = true;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
let mut reg = inner.registry.lock();
|
let mut reg = inner.registry.lock();
|
||||||
|
|||||||
+23
-19
@@ -472,24 +472,28 @@ pub(crate) struct SlotCold {
|
|||||||
/// epoch-matched unpark.
|
/// epoch-matched unpark.
|
||||||
pub(crate) waiters: Vec<(Pid, u32)>,
|
pub(crate) waiters: Vec<(Pid, u32)>,
|
||||||
pub(crate) outcome: Option<Outcome>,
|
pub(crate) outcome: Option<Outcome>,
|
||||||
/// The slot's most recent *named-tenancy* death: `(generation, reason)`,
|
/// The slot's most recent *watchable-tenancy* death: `(generation,
|
||||||
/// stamped by `finalize_actor` — but only for a tenancy that ever
|
/// reason)`, stamped by `finalize_actor` — but only for a tenancy whose
|
||||||
/// registered a name (`ever_named`) — and deliberately never cleared: a
|
/// `watchable` bit was set — and deliberately never cleared: a new
|
||||||
/// new tenant's install leaves it standing (it describes the previous
|
/// tenant's install leaves it standing (it describes the previous
|
||||||
/// tenancy), and only the next *named* death overwrites it. Anonymous
|
/// tenancy), and only the next *watchable* death overwrites it.
|
||||||
/// green-thread churn must not evict it: the free list is LIFO, so the
|
/// Anonymous green-thread churn must not evict it: the free list is
|
||||||
/// just-freed slot is the first recycled, and an unconditional stamp
|
/// LIFO, so the just-freed slot is the first recycled, and an
|
||||||
/// made a watchable tenancy's record the shortest-lived data in the
|
/// unconditional stamp made a watchable tenancy's record the
|
||||||
/// runtime. Read generation-matched via
|
/// shortest-lived data in the runtime. Read generation-matched via
|
||||||
/// [`terminal_reason`](crate::monitor::terminal_reason), so a watch that
|
/// [`terminal_reason`](crate::monitor::terminal_reason), so a watch that
|
||||||
/// raced its target's death can recover the real down reason instead of
|
/// raced its target's death can recover the real down reason instead of
|
||||||
/// a blanket `NoProc` (bridge soak signature 4).
|
/// a blanket `NoProc` (bridge soak signatures 4 and 5).
|
||||||
pub(crate) terminal: Option<(u32, DownReason)>,
|
pub(crate) terminal: Option<(u32, DownReason)>,
|
||||||
/// This tenancy registered a name at least once — the stamp-eligibility
|
/// Stamp eligibility for `terminal` above: someone could plausibly hold
|
||||||
/// bit for `terminal` above. Set by `register_with` *before* the binding
|
/// a watch on this tenancy. Two set-sites, both while the tenancy is
|
||||||
/// lands (so no successfully-registered actor can die unflagged; a
|
/// live: `register_with` *before* the binding lands (no successfully
|
||||||
/// failed register's overshoot is harmless), reset at reclaim.
|
/// registered actor can die unflagged; a failed register's overshoot is
|
||||||
pub(crate) ever_named: bool,
|
/// harmless), and [`mark_watchable`](crate::monitor::mark_watchable) —
|
||||||
|
/// the bridge calls it wherever a pid is encoded across the boundary,
|
||||||
|
/// because BEAM can only watch pids it holds and can only hold pids
|
||||||
|
/// that crossed. Reset at reclaim.
|
||||||
|
pub(crate) watchable: bool,
|
||||||
pub(crate) supervisor_channel: Option<Sender<Signal>>,
|
pub(crate) supervisor_channel: Option<Sender<Signal>>,
|
||||||
/// Watchers registered via `monitor()`, each tagged with its
|
/// Watchers registered via `monitor()`, each tagged with its
|
||||||
/// `MonitorId` so `demonitor` can remove exactly one. Each receives one
|
/// `MonitorId` so `demonitor` can remove exactly one. Each receives one
|
||||||
@@ -645,7 +649,7 @@ impl Slot {
|
|||||||
waiters: Vec::new(),
|
waiters: Vec::new(),
|
||||||
outcome: None,
|
outcome: None,
|
||||||
terminal: None,
|
terminal: None,
|
||||||
ever_named: false,
|
watchable: false,
|
||||||
supervisor_channel: None,
|
supervisor_channel: None,
|
||||||
monitors: Vec::new(),
|
monitors: Vec::new(),
|
||||||
links: Vec::new(),
|
links: Vec::new(),
|
||||||
@@ -1648,7 +1652,7 @@ pub(crate) fn reclaim_slot(inner: &RuntimeInner, pid: Pid) {
|
|||||||
cold.waiters.clear();
|
cold.waiters.clear();
|
||||||
cold.monitors.clear();
|
cold.monitors.clear();
|
||||||
cold.links.clear();
|
cold.links.clear();
|
||||||
cold.ever_named = false;
|
cold.watchable = false;
|
||||||
slot.reset_counters();
|
slot.reset_counters();
|
||||||
slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release);
|
slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release);
|
||||||
// The generation bump IS the reclaim: every stale pid is dead from
|
// The generation bump IS the reclaim: every stale pid is dead from
|
||||||
@@ -1696,8 +1700,8 @@ fn finalize_actor(inner: &Arc<RuntimeInner>, pid: Pid, outcome: Outcome) {
|
|||||||
// thread's exit stamped too, the churn behind any real workload
|
// thread's exit stamped too, the churn behind any real workload
|
||||||
// would evict a watchable tenancy's record in well under the race
|
// would evict a watchable tenancy's record in well under the race
|
||||||
// window this exists to cover. Overwritten only by the slot's next
|
// window this exists to cover. Overwritten only by the slot's next
|
||||||
// *named* death.
|
// *watchable* death.
|
||||||
if cold.ever_named {
|
if cold.watchable {
|
||||||
cold.terminal = Some((pid.generation(), down_reason));
|
cold.terminal = Some((pid.generation(), down_reason));
|
||||||
}
|
}
|
||||||
slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release);
|
slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release);
|
||||||
|
|||||||
@@ -17,7 +17,7 @@
|
|||||||
//! caller's deliberate act, not a semantics change.
|
//! caller's deliberate act, not a semantics change.
|
||||||
|
|
||||||
use smarm::{
|
use smarm::{
|
||||||
init, request_stop, resolve_name, terminal_reason, CallError, Config, DownReason, GenServer,
|
init, mark_watchable, request_stop, resolve_name, terminal_reason, CallError, Config, DownReason, GenServer,
|
||||||
GenServerBuilder, GenServerName, NameResolution,
|
GenServerBuilder, GenServerName, NameResolution,
|
||||||
};
|
};
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, Mutex};
|
||||||
@@ -59,6 +59,10 @@ impl GenServer for Filler {
|
|||||||
struct Observed {
|
struct Observed {
|
||||||
exit_reason: Option<DownReason>,
|
exit_reason: Option<DownReason>,
|
||||||
anon_reason: Option<DownReason>,
|
anon_reason: Option<DownReason>,
|
||||||
|
/// Anonymous but export-marked while alive — must stamp (sig 5).
|
||||||
|
marked_reason: Option<DownReason>,
|
||||||
|
/// Marked only after death — must remain unknowable.
|
||||||
|
marked_late_reason: Option<DownReason>,
|
||||||
panic_reason: Option<DownReason>,
|
panic_reason: Option<DownReason>,
|
||||||
stopped_reason: Option<DownReason>,
|
stopped_reason: Option<DownReason>,
|
||||||
live_reason: Option<DownReason>,
|
live_reason: Option<DownReason>,
|
||||||
@@ -126,6 +130,22 @@ fn terminal_record_recovers_the_reason_a_raced_watch_lost() {
|
|||||||
let _ = h.join();
|
let _ = h.join();
|
||||||
let anon_reason = terminal_reason(pid_anon);
|
let anon_reason = terminal_reason(pid_anon);
|
||||||
|
|
||||||
|
// --- mark_watchable: the bridge's export-seam eligibility (sig 5).
|
||||||
|
// An anonymous actor marked while alive stamps like a named one ...
|
||||||
|
let h = smarm::spawn(|| loop {
|
||||||
|
smarm::sleep(Duration::from_millis(2));
|
||||||
|
});
|
||||||
|
let pid_marked = h.pid();
|
||||||
|
mark_watchable(pid_marked);
|
||||||
|
request_stop(pid_marked);
|
||||||
|
let _ = h.join();
|
||||||
|
let marked_reason = terminal_reason(pid_marked);
|
||||||
|
|
||||||
|
// ... while marking a pid whose tenancy already ended is a no-op:
|
||||||
|
// the history is honestly unknowable, not retroactively invented.
|
||||||
|
mark_watchable(pid_anon);
|
||||||
|
let marked_late_reason = terminal_reason(pid_anon);
|
||||||
|
|
||||||
// --- The named target: live readings first. -----------------------
|
// --- The named target: live readings first. -----------------------
|
||||||
let target = GenServerBuilder::new(Target)
|
let target = GenServerBuilder::new(Target)
|
||||||
.named(TARGET)
|
.named(TARGET)
|
||||||
@@ -197,6 +217,8 @@ fn terminal_record_recovers_the_reason_a_raced_watch_lost() {
|
|||||||
unknown_resolution,
|
unknown_resolution,
|
||||||
corpse_resolution_matches,
|
corpse_resolution_matches,
|
||||||
resolution_after_prune,
|
resolution_after_prune,
|
||||||
|
marked_reason,
|
||||||
|
marked_late_reason,
|
||||||
corpse_reason_after_prune,
|
corpse_reason_after_prune,
|
||||||
corpse_reason_after_reuse,
|
corpse_reason_after_reuse,
|
||||||
corpse_reason_after_tenant_death,
|
corpse_reason_after_tenant_death,
|
||||||
@@ -211,6 +233,15 @@ fn terminal_record_recovers_the_reason_a_raced_watch_lost() {
|
|||||||
"anonymous deaths must not stamp: {o:?}"
|
"anonymous deaths must not stamp: {o:?}"
|
||||||
);
|
);
|
||||||
assert_eq!(o.panic_reason, Some(DownReason::Panic), "{o:?}");
|
assert_eq!(o.panic_reason, Some(DownReason::Panic), "{o:?}");
|
||||||
|
assert_eq!(
|
||||||
|
o.marked_reason,
|
||||||
|
Some(DownReason::Stopped),
|
||||||
|
"mark_watchable while alive must make the death stamp: {o:?}"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
o.marked_late_reason, None,
|
||||||
|
"marking a dead tenancy must not invent history: {o:?}"
|
||||||
|
);
|
||||||
assert_eq!(o.stopped_reason, Some(DownReason::Stopped), "{o:?}");
|
assert_eq!(o.stopped_reason, Some(DownReason::Stopped), "{o:?}");
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
o.live_reason, None,
|
o.live_reason, None,
|
||||||
|
|||||||
Reference in New Issue
Block a user