1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
//! Process-spawn exit ownership helpers for [`RuntimeHandle`].
use beamr::process::ExitReason;
use super::{EngineError, Pid, RuntimeHandle};
use crate::runtime::process_exit::ProcessEnding;
/// The engine's verdict on a wake marker the runtime refused to deliver.
///
/// A wake marker is only ever a nudge: what it announces — a recorded signal,
/// a parked query — is already durable before the marker is enqueued. So the
/// question a refusal raises is not whether the enqueue worked but whether
/// the target ended, and these are the only honest answers to it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum UndeliveredWake {
/// The engine holds a record that the process ended: the exit-cleanup
/// tombstone, or the exit registry's published terminal. The refusal is
/// the completion race and the already-durable record stands.
ProcessEnded,
/// Nothing records an ending and nothing is in flight. Nothing excuses
/// the refusal, and `in_table` carries the liveness the classifier read
/// to decide that, so a caller that must tell a process which is present
/// and simply refusing from one which is gone without a trace does not
/// read the process table a second time and risk a different answer.
ProcessDidNotEnd {
/// Whether the pid was in the scheduler's process table when the
/// classifier looked.
in_table: bool,
},
/// The registry holds the process' record with no terminal published on
/// it while the pid has left the scheduler's table: the exit was in
/// flight, and it had still published nothing when the readiness window
/// the classifier waited out expired.
ExitInFlight,
}
impl RuntimeHandle {
pub(super) fn spawn_with_exit_ownership(
&self,
spawn: impl FnOnce() -> Result<Pid, EngineError>,
) -> Result<Pid, EngineError> {
let reservation = self.process_exits.reserve_spawn()?;
let pid = spawn()?;
if let Err(error) = reservation.register(pid) {
self.scheduler.terminate_process(pid, ExitReason::Kill);
if let Err(cleanup_error) = self.finish_process_monitor_cleanup(pid) {
tracing::error!(pid, %cleanup_error, cause = %error, "spawn rollback cleanup failed after exit ownership setup failed");
return Err(cleanup_error);
}
return Err(error);
}
Ok(pid)
}
/// Ensure `pid` has a runtime-owned, non-consuming exit outcome record.
///
/// A workflow can run to completion on a scheduler thread between its
/// spawn and monitor installation. Registration occurs before the `pid` is
/// returned, so fast exits are read from Aion's permanent cache.
pub(crate) fn ensure_monitorable_pid(&self, pid: Pid) -> Result<(), EngineError> {
if self.process_exits.contains(pid) {
return Ok(());
}
// No record is held. The registry retires a record only after the
// process' terminal outcome has been delivered, so an absence at or
// below the registration watermark is the only shape a reaped process
// can present here; an absence above it is a pid this runtime
// provably never spawned.
if self.process_exits.below_registration_watermark(pid) {
return Err(EngineError::ProcessExitAlreadyTerminal { process_id: pid });
}
Err(super::runtime_error(format!(
"process {pid} was never spawned by this runtime"
)))
}
/// Whether the engine holds a record that `pid` ended.
///
/// Two things record an ending: the exit-cleanup tombstone, which cleanup
/// stamps before it tears any process state down, and the exit registry's
/// published terminal outcome. Absence from the scheduler's process table
/// is neither — beamr retires the process-table row on its own schedule,
/// and a pid this runtime never spawned is absent from the table for a
/// reason that has nothing to do with a workflow ending. Callers that
/// must tell a completed workflow from an unreachable one read this, never
/// liveness.
///
/// # Errors
///
/// Returns [`EngineError::ProcessExitStatePoisoned`] when the exit
/// record's outcome lock is poisoned. The tombstone is read first and
/// settles the question on its own; the registry is consulted only when
/// no tombstone exists, and an unreadable registry is surfaced there
/// rather than silently answered "did not end".
pub(crate) fn process_ending_recorded(&self, pid: Pid) -> Result<bool, EngineError> {
Ok(self.process_cleanup_started(pid) || self.process_exits.has_terminal(pid)?)
}
/// Classify a wake marker the runtime refused to deliver.
///
/// Both wake paths gate on the pid being in the scheduler's process table
/// before they enqueue anything, so the ordinary cause of a refusal is a
/// target that has just finished. beamr drops the process-table row on
/// its own schedule and the exit drainer publishes the terminal on the
/// held record afterwards, so a single read taken between the two sees
/// neither the row nor the terminal. Answering "did not end" there turns
/// an accepted signal into a delivery failure — and what the marker
/// announces is already durable, so a caller that re-sends on that
/// failure appends the signal a second time.
///
/// That gap is waited out rather than guessed: a held record with nothing
/// published on it, for a pid no longer in the table, is an exit in
/// flight, and this parks on that record's own publication for the
/// readiness window the signal-delivery policy already gives a wake. Only
/// an exit that does not publish inside that window is
/// [`UndeliveredWake::ExitInFlight`], and the caller names both facts
/// when it reports it.
///
/// # Errors
///
/// Returns [`EngineError::ProcessExitStatePoisoned`] when the exit
/// record's outcome lock is poisoned, so an unreadable registry is
/// surfaced rather than silently answered "did not end".
pub(crate) fn classify_undelivered_wake(
&self,
pid: Pid,
) -> Result<UndeliveredWake, EngineError> {
// Exit cleanup stamps its tombstone before it tears any process state
// down, and it is the one record of an ending the exit registry never
// holds, so it is asked on its own and first.
if self.process_cleanup_started(pid) {
return Ok(UndeliveredWake::ProcessEnded);
}
// Everything else is decided by ONE read of the registry, which keeps
// whatever evidence that read finds: a terminal the drainer published
// since the tombstone check above is an ending whichever read sees it
// first, and answering "did not end" on it would report an accepted
// signal as undelivered. Only a held record with nothing published on
// it can still be in flight, and only once beamr has dropped the
// process-table row; a pid still in the table has not ended, whatever
// refused the marker; and a registry holding no record for the pid has
// nothing to wait for.
match self.process_exits.ending_knowledge(pid)? {
ProcessEnding::Recorded => return Ok(UndeliveredWake::ProcessEnded),
ProcessEnding::Pending if !self.is_live(pid) => {}
ProcessEnding::Pending => {
return Ok(UndeliveredWake::ProcessDidNotEnd { in_table: true });
}
ProcessEnding::Forgotten | ProcessEnding::NeverRegistered => {
return Ok(UndeliveredWake::ProcessDidNotEnd {
in_table: self.is_live(pid),
});
}
}
let Some(record) = self.process_exits.find(pid) else {
// Retired between the two reads. Retirement always follows the
// terminal it delivered, so a record that was held a moment ago
// and is gone now published its ending.
return Ok(UndeliveredWake::ProcessEnded);
};
// The park answers on the record; the re-read after it also catches an
// exit-cleanup tombstone stamped while this waited, which is a
// recorded ending on its own.
if record.wait_for_terminal(self.signal_delivery().ready_timeout)?
|| self.process_ending_recorded(pid)?
{
return Ok(UndeliveredWake::ProcessEnded);
}
Ok(UndeliveredWake::ExitInFlight)
}
#[cfg(test)]
pub(crate) fn process_exit_for_test(
&self,
pid: Pid,
) -> Result<(ExitReason, beamr::term::Term), EngineError> {
let observed = self.process_exit_outcome(pid)?;
self.release_spawn_heaps(pid);
Ok((observed.reason, observed.result.root()))
}
}