Skip to main content

epics_libcom_rs/runtime/
task.rs

1// RTEMS-EXEC-MODEL-ALLOW(2): the two multi-thread-flavored tests prove a
2// dedicated thread carries the ambient tokio runtime / requested stack; the
3// tokio flavor is the property under test. Both run and pass in the
4// exec-backend suite.
5
6use std::future::Future;
7use std::sync::Arc;
8use std::task::{Context, Poll, Wake, Waker};
9use std::time::Duration;
10use tokio::runtime::RuntimeFlavor;
11
12pub use tokio::runtime::Handle as RuntimeHandle;
13
14pub use crate::runtime::background::CallbackPriority;
15
16/// A synchronous caller asked to block on an async operation from a thread
17/// where blocking cannot be made sound.
18///
19/// Both variants are the same defect seen through two executors: the calling
20/// thread is one the awaited future needs in order to make progress, so parking
21/// it parks the thing that would wake it. No blocking mechanism can fix that;
22/// the caller has to `await` the async operation instead of blocking on it.
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24pub enum NotBlockable {
25    /// A **current-thread** tokio runtime is entered on this thread. Parking it
26    /// stops every task on that runtime, including whichever one holds the
27    /// state the awaited future is waiting for.
28    CurrentThreadRuntime,
29    /// This thread is a background-facility worker — a callback band, the
30    /// delayed-callback timer, or the scanOnce worker
31    /// ([`crate::runtime::background`]). Each facility has a bounded worker set
32    /// and every unit of work it carries is enqueued for those workers, so a
33    /// parked worker is waiting for work only it could have run. On RTEMS
34    /// [`Reactor::spawn`] routes here, which makes the callback bands the one
35    /// other place where parking is unsound.
36    BackgroundWorker,
37}
38
39impl std::fmt::Display for NotBlockable {
40    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41        match self {
42            NotBlockable::CurrentThreadRuntime => {
43                f.write_str("cannot block a current-thread runtime")
44            }
45            NotBlockable::BackgroundWorker => {
46                f.write_str("cannot block a background-facility worker thread")
47            }
48        }
49    }
50}
51
52impl std::error::Error for NotBlockable {}
53
54/// A [`Waker`] that unparks the thread that built it. The single owner of the
55/// "poll-then-park" wake mechanism in this crate: both [`park_on`] (the sync
56/// bridge) and the RTEMS future executor
57/// ([`crate::runtime::background::future_exec`]) drive a future by polling on a
58/// thread and parking it between polls, so both build one of these on their own
59/// thread and rely on the future's cross-thread waker to unpark them.
60pub(crate) struct ThreadWaker(std::thread::Thread);
61
62impl ThreadWaker {
63    /// A waker over the *current* thread — call this on the thread that will
64    /// park.
65    pub(crate) fn for_current_thread() -> Waker {
66        Waker::from(Arc::new(ThreadWaker(std::thread::current())))
67    }
68}
69
70impl Wake for ThreadWaker {
71    fn wake(self: Arc<Self>) {
72        self.0.unpark();
73    }
74    fn wake_by_ref(self: &Arc<Self>) {
75        self.0.unpark();
76    }
77}
78
79/// Drive `fut` to completion on this thread, parking between polls, and stop
80/// early when `should_cancel` returns `true`.
81///
82/// Returns `Some(output)` when the future completed, or `None` when it was
83/// cancelled before completing (the future is dropped in place on cancel,
84/// running its destructors — the same "drop at the next suspension point"
85/// semantics a cancelled tokio task has).
86///
87/// The future must only await runtime-agnostic primitives (`tokio::sync`
88/// locks/channels/notifies): nothing here drives a reactor or a timer wheel, so
89/// whoever wakes us must be running on some other thread. A cancel is observed
90/// on the next wake — the caller that flips `should_cancel` must also
91/// [`unpark`](std::thread::Thread::unpark) this thread so a *parked* driver
92/// re-checks promptly rather than sleeping until the future's own waker fires.
93pub(crate) fn park_on_interruptible<F: Future>(
94    fut: F,
95    mut should_cancel: impl FnMut() -> bool,
96) -> Option<F::Output> {
97    let mut fut = std::pin::pin!(fut);
98    let waker = ThreadWaker::for_current_thread();
99    let mut cx = Context::from_waker(&waker);
100    loop {
101        if should_cancel() {
102            return None;
103        }
104        if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) {
105            return Some(value);
106        }
107        std::thread::park();
108    }
109}
110
111/// Drive `fut` to completion on this thread, parking between polls. Thin
112/// uncancellable wrapper over [`park_on_interruptible`].
113///
114/// The future must only await runtime-agnostic primitives (`tokio::sync`
115/// locks/channels/notifies): nothing here drives a reactor or a timer wheel, so
116/// whoever wakes us must be running on some other thread.
117fn park_on<F: Future>(fut: F) -> F::Output {
118    // Never cancels, so `park_on_interruptible` always returns `Some`.
119    park_on_interruptible(fut, || false).expect("uncancellable driver returned None")
120}
121
122/// Block the calling thread on `fut`, picking the mechanism that is sound for
123/// the thread we are actually on.
124///
125/// This is the single owner of "sync call over async state" in this crate; the
126/// four caller contexts are not interchangeable and picking one mechanism for
127/// all of them is what makes such bridges panic:
128///
129/// - **A background-facility worker** —
130///   [`Err(BackgroundWorker)`](NotBlockable::BackgroundWorker), checked first,
131///   because it is a property of the *thread* and holds whatever runtime is or
132///   is not entered on it. See
133///   [`background::facility::on_facility_thread`](crate::runtime::background)
134///   for why parking one is unsound.
135/// - **No runtime entered** (a plain `std::thread`, an iocsh thread) — park the
136///   thread. Nothing else runs here, so there is nothing to starve; the tasks
137///   that will wake us live on some other runtime's threads.
138/// - **Multi-thread runtime worker** — [`tokio::task::block_in_place`], which
139///   hands this worker's remaining tasks to a sibling before it is parked.
140/// - **Current-thread runtime** —
141///   [`Err(CurrentThreadRuntime)`](NotBlockable::CurrentThreadRuntime). Parking
142///   the only thread of that runtime halts every task on it, including the one
143///   that would wake us.
144///
145/// The two refusals are reported to the caller rather than panicked on (today)
146/// or deadlocked on (the worse alternative) — an illegal blocking bridge is a
147/// value the caller must handle, not a review item.
148pub fn block_on_sync<F: Future>(fut: F) -> Result<F::Output, NotBlockable> {
149    if crate::runtime::background::facility::on_facility_thread() {
150        return Err(NotBlockable::BackgroundWorker);
151    }
152    match RuntimeHandle::try_current() {
153        Ok(handle) => match handle.runtime_flavor() {
154            RuntimeFlavor::CurrentThread => Err(NotBlockable::CurrentThreadRuntime),
155            _ => Ok(tokio::task::block_in_place(|| handle.block_on(fut))),
156        },
157        Err(_) => Ok(park_on(fut)),
158    }
159}
160
161/// A capability, captured where the backend's executor is reachable, to run
162/// async work from a plain blocking thread (iocsh, a REPL, a script thread).
163///
164/// [`block_on_sync`] answers "may I block *here*, now?" per call and can only
165/// use whatever runtime is visible on the calling thread. This type answers
166/// the reachability question once, at [`capture`](Self::capture) time, and
167/// carries the answer to a thread the runtime is otherwise invisible from: a
168/// tokio handle is thread-local state, so a blocking thread spawned *before*
169/// it exists has no way to find it. The exec backend's executor is
170/// process-global, so there is nothing to carry and the bridge is a ZST —
171/// which is what makes an API taking a `BlockingBridge` compile and work on
172/// both backends, where one taking `tokio::runtime::Handle` pinned every
173/// caller to tokio.
174#[cfg(tokio_backend)]
175#[derive(Clone)]
176pub struct BlockingBridge {
177    handle: tokio::runtime::Handle,
178}
179
180/// See the `tokio_backend` definition. The executor here is the
181/// process-global background executor, reachable from any thread, so there is
182/// no state to capture.
183#[cfg(exec_backend)]
184#[derive(Clone)]
185pub struct BlockingBridge;
186
187#[cfg(tokio_backend)]
188impl BlockingBridge {
189    /// Capture the current tokio runtime.
190    ///
191    /// # Panics
192    /// Panics when no runtime is entered on this thread — call it on the
193    /// async setup path (where the runtime is known), not on the blocking
194    /// thread the bridge is being made for.
195    pub fn capture() -> Self {
196        Self {
197            handle: tokio::runtime::Handle::current(),
198        }
199    }
200
201    /// [`capture`](Self::capture) for a caller that has somewhere else to be
202    /// if no runtime is entered — `None` instead of a panic.
203    pub fn try_capture() -> Option<Self> {
204        RuntimeHandle::try_current()
205            .ok()
206            .map(|handle| Self { handle })
207    }
208
209    /// Drive `fut` to completion on this thread, with the captured runtime
210    /// entered so the future may spawn and use the reactor.
211    ///
212    /// # Panics
213    /// Panics on a runtime worker thread: blocking one parks tasks that may
214    /// include the future's own wakers (the same refusal `block_on_sync`
215    /// reports as a value).
216    pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
217        assert!(
218            RuntimeHandle::try_current().is_err(),
219            "BlockingBridge::block_on must not be called from a runtime thread"
220        );
221        self.handle.block_on(fut)
222    }
223
224    /// Spawn `future` onto the captured runtime — [`Reactor::spawn`] for a
225    /// thread the runtime is not entered on.
226    pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
227    where
228        F: Future + Send + 'static,
229        F::Output: Send + 'static,
230    {
231        self.handle.spawn(future)
232    }
233
234    /// The captured runtime as a [`Reactor`], for handing to a reactor-bound
235    /// task the bridge starts.
236    ///
237    /// The bridge already answered "is the executor reachable from here?" at
238    /// capture time; this is the same answer in the shape a spawn site takes.
239    pub fn reactor(&self) -> Reactor {
240        Reactor {
241            handle: self.handle.clone(),
242        }
243    }
244}
245
246#[cfg(exec_backend)]
247impl BlockingBridge {
248    /// The exec backend's executor is process-global; capturing is a no-op
249    /// and never panics.
250    pub fn capture() -> Self {
251        Self
252    }
253
254    /// See the `tokio_backend` definition; capturing never fails here.
255    pub fn try_capture() -> Option<Self> {
256        Some(Self)
257    }
258
259    /// Drive `fut` on this thread via `park_on`; whatever it spawns or
260    /// sleeps on lands on the background executor.
261    pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
262        park_on(fut)
263    }
264
265    /// [`Reactor::spawn`] — the global executor needs no captured state.
266    ///
267    /// The reactor seam mirrors `tokio::spawn`, whose callers are servers,
268    /// clients and iocsh, never record support: there is no record here and so
269    /// no `PRIO` to read. It takes the middle band, C's `callbackRequest`
270    /// default for general deferred work (`callback.h:42`).
271    pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
272    where
273        F: Future + Send + 'static,
274        F::Output: Send + 'static,
275    {
276        spawn_background(CallbackPriority::Medium, future)
277    }
278
279    /// See the `tokio_backend` definition; this carries no runtime, only the
280    /// default (callback-pool) spawn target.
281    pub fn reactor(&self) -> Reactor {
282        Reactor::default()
283    }
284}
285
286/// Drive an async test body to completion — the driver behind
287/// `#[epics_test]` (`epics-macros-rs`).
288///
289/// The point of the indirection is that the *backend* picks the driver, not
290/// the test. On `tokio_backend` this builds exactly what `#[tokio::test]`
291/// builds: a fresh current-thread runtime with IO and time enabled. On
292/// `exec_backend` (the RTEMS target, or a host run with
293/// `EPICS_RS_BUILD_EXEC_BACKEND=thread`) no tokio runtime exists to build, so
294/// the
295/// test thread itself drives the future via `park_on`, and everything the
296/// body spawns or sleeps on lands on the process-global background executor
297/// (lazily initialised on first use) — the same seam the RTEMS boot path
298/// exercises. A test written with `#[epics_test]` therefore needs no
299/// per-backend gating and no `RTEMS-EXEC-MODEL-ALLOW` census entry.
300#[cfg(tokio_backend)]
301pub fn test_block_on<F: Future>(fut: F) -> F::Output {
302    tokio::runtime::Builder::new_current_thread()
303        .enable_all()
304        .build()
305        .expect("failed to build tokio test runtime")
306        .block_on(fut)
307}
308
309/// `exec_backend` twin of [`test_block_on`]: see the `tokio_backend` copy for
310/// the contract. The body's awaits must reach only runtime-agnostic
311/// primitives (`park_on`'s rule) — a body that touches `tokio::net` or
312/// `tokio::time` directly belongs under `#[tokio::test]` with a backend gate
313/// instead.
314#[cfg(exec_backend)]
315pub fn test_block_on<F: Future>(fut: F) -> F::Output {
316    park_on(fut)
317}
318
319// ---------------------------------------------------------------------------
320// Platform-selected task handle types (decision A2 / B)
321//
322// The seam hands back one of these aliases from every spawn; call sites in this
323// crate name only the alias, never a tokio handle. Hosted = the tokio handle
324// types. RTEMS = the always-compiled, host-tested mirrors in
325// `background::future_exec` (`JoinFuture`/`AbortHandle`/`JoinError`), which
326// reproduce exactly the subset of the tokio surface the call sites use.
327// ---------------------------------------------------------------------------
328
329/// `true` when [`Reactor::spawn`] lands the future on the tokio runtime,
330/// `false` when it lands on the reactor-free background executor
331/// (`exec_backend` — the RTEMS target, or a host build with
332/// `EPICS_RS_BUILD_EXEC_BACKEND=thread`).
333///
334/// # What this is for
335///
336/// It is the *exported* form of `build.rs`'s backend decision, and the reason
337/// it is exported is that a spawned future's access to a tokio **reactor** is
338/// decided here and consumed in other crates. A future handed to
339/// [`Reactor::spawn`] on `exec_backend` runs on a callback-pool worker with no
340/// reactor entered, so
341/// every `tokio::net` socket it opens panics — *even in a process that has a
342/// tokio runtime somewhere else*, because the runtime is not entered on that
343/// worker.
344///
345/// `epics-ca-rs` and `epics-pva-rs` therefore have to make the same decision
346/// this crate makes, for their own compilation, and they make it in their own
347/// `build.rs` from the same two inputs (target OS, `EPICS_RS_BUILD_EXEC_BACKEND`).
348/// That is three copies of one rule, so each of them pins the copy against this
349/// constant with a `const` assertion — a build where the two disagree, which
350/// now means one of the scripts did not see the variable, fails to compile
351/// instead of panicking at boot.
352pub const HAS_TOKIO_REACTOR: bool = cfg!(tokio_backend);
353
354/// Handle to a spawned task — `await` for its result, `abort()` to cancel.
355#[cfg(tokio_backend)]
356pub type TaskHandle<T> = tokio::task::JoinHandle<T>;
357/// Detached cancellation handle for a spawned task.
358#[cfg(tokio_backend)]
359pub type TaskAbortHandle = tokio::task::AbortHandle;
360/// Error from awaiting a [`TaskHandle`] (cancelled or panicked).
361#[cfg(tokio_backend)]
362pub type TaskJoinError = tokio::task::JoinError;
363
364#[cfg(exec_backend)]
365pub type TaskHandle<T> = crate::runtime::background::future_exec::JoinFuture<T>;
366#[cfg(exec_backend)]
367pub type TaskAbortHandle = crate::runtime::background::future_exec::AbortHandle;
368#[cfg(exec_backend)]
369pub type TaskJoinError = crate::runtime::background::future_exec::JoinError;
370
371// ---------------------------------------------------------------------------
372// Process-global background executor (C `callbackInit` facilities)
373//
374// One process-global `BackgroundExecutor` — callback pool + delayed timer +
375// scanOnce worker — on *every* backend, because it is the only executor whose
376// existence does not depend on an ambient runtime. Two init paths, both
377// landing on the same `OnceLock`:
378//
379//   * Explicit — `background_init()` from `IocApplication::run`, mirroring C's
380//     `callbackInit` running early in `iocInit` (callback.c:286) so the
381//     facilities exist before any record processing can defer a tail.
382//   * Lazy fallback — the first `spawn_background`/`sleep_background` on a path
383//     that never went through `run` (a unit test, an embedded harness)
384//     initialises it on demand via the same `get_or_init`.
385//
386// `exec_backend` additionally routes the *ambient* seam (`spawn`, `sleep`,
387// `interval`) here, because on that backend there is nothing else to route to.
388// ---------------------------------------------------------------------------
389
390static BACKGROUND: std::sync::OnceLock<crate::runtime::background::BackgroundExecutor> =
391    std::sync::OnceLock::new();
392
393/// The process-global background executor, initialised on first use.
394fn background() -> &'static crate::runtime::background::BackgroundExecutor {
395    BACKGROUND.get_or_init(crate::runtime::background::BackgroundExecutor::new)
396}
397
398/// Eagerly start the process-global background executor — C `callbackInit`
399/// parity (callback.c:286), called once from `IocApplication::run`. Idempotent:
400/// a second call (or a prior lazy init) is a no-op, matching `callbackInit`'s
401/// own re-entry guard (callback.c:292-295).
402pub fn background_init() {
403    let _ = background();
404}
405
406/// Has the process-global background executor been built yet?
407///
408/// C's twin is `epicsAtomicGetIntT(&cbState) != cbInit` — the guard
409/// `callbackSetQueueSize` and `callbackParallelThreads` refuse on
410/// (`callback.c:107`, `:162`). Sizing knobs are read once, when the pool
411/// is constructed, so a write after this returns `true` changes nothing
412/// and has to be reported rather than silently accepted.
413pub fn background_started() -> bool {
414    BACKGROUND.get().is_some()
415}
416
417/// Queue statistics for all three callback bands, or `None` when the
418/// background executor has not been built yet — C `callbackQueueStatus`
419/// (`callback.c:115-141`), whose `-1` return is exactly this `None` and
420/// is what `callbackQueueShow` turns into its "not initialized, yet"
421/// diagnostic. `reset` clears every band's high-water mark, as C's does
422/// for the whole array regardless of which band is being read.
423///
424/// Reads `BACKGROUND` rather than calling `background()`, because a
425/// report must not be the thing that starts the facility it reports on.
426pub fn background_callback_stats(
427    reset: bool,
428) -> Option<
429    [crate::runtime::background::CallbackQueueStats;
430        crate::runtime::background::NUM_CALLBACK_PRIORITIES],
431> {
432    let exec = BACKGROUND.get()?;
433    Some(
434        crate::runtime::background::CallbackPriority::ALL.map(|p| exec.callbacks().stats(p, reset)),
435    )
436}
437
438/// Create the process-global `scanOnce` worker thread now — C `initOnce`
439/// (`dbScan.c:768-780`), called from the port's `scanInit` equivalent so the
440/// thread exists from IOC init rather than from the first one-shot. Idempotent.
441pub fn background_scan_once_start() {
442    background().scan_once().start();
443}
444
445/// Ring statistics for the `scanOnce` facility, or `None` before the
446/// background executor exists — C `scanOnceQueueStatus` (`dbScan.c:734-757`)
447/// and its `if (!onceQ) return -1` guard.
448pub fn background_scan_once_stats(
449    reset: bool,
450) -> Option<crate::runtime::background::ScanOnceQueueStats> {
451    Some(BACKGROUND.get()?.scan_once().stats(reset))
452}
453
454/// Handle to a task spawned on the process-global background executor —
455/// `await` for its result, `abort()` to cancel. Distinct from [`TaskHandle`]
456/// because that one is the *ambient* executor's handle and is
457/// `tokio::task::JoinHandle` on a hosted build.
458pub type BackgroundTaskHandle<T> = crate::runtime::background::JoinFuture<T>;
459
460/// Spawn a deferred tail on the process-global background executor.
461///
462/// The counterpart to [`Reactor::spawn`], and the difference is the whole
463/// point of having both: that one needs a [`Reactor`], which on a hosted build
464/// is the tokio runtime with its I/O and time drivers, while this one always
465/// lands on the same executor no matter who calls it and needs no capability
466/// at all. Record processing is reached from a plain `std::thread` — every
467/// blocking CA/PVA connection thread drives it through [`block_on_sync`] →
468/// `park_on` — so a tail it defers must not depend on the caller's thread
469/// having a runtime, and must not be handed a `Reactor` either.
470///
471/// Anything awaited inside `future` is subject to the same rule: use
472/// [`sleep_background`], [`interval_background`] and [`spawn_blocking_background`]
473/// rather than their ambient counterparts, and no `tokio::net` socket, whose
474/// reactor this executor deliberately does not have.
475///
476/// # The band is the caller's to name
477///
478/// `priority` picks which of the three callback queues runs the tail — C
479/// `callbackRequest` dispatches on `CALLBACK.priority` (`callback.c:355-365`)
480/// and record support sets that from the record it is deferring:
481/// `callbackSetPriority(prec->prio, &pcb->callback)` at the top of
482/// `seqRecord.c:146` `process()`, re-read every cycle. There is deliberately
483/// no defaulted spelling of this function: a record tail that silently took
484/// one fixed band would make every record's `PRIO` field select nothing, so
485/// the band is a parameter and a site with no record to name must say which
486/// band it means and why. Use
487/// [`CallbackPriority::from_record_prio`] wherever a `PRIO` is in hand.
488pub fn spawn_background<F>(priority: CallbackPriority, future: F) -> BackgroundTaskHandle<F::Output>
489where
490    F: Future + Send + 'static,
491    F::Output: Send + 'static,
492{
493    use crate::runtime::background::spawn_future;
494    spawn_future(&background().callbacks().handle(), priority, future)
495}
496
497/// [`spawn_background`] for a blocking closure — runs it on a callback-pool
498/// worker of the named band.
499pub fn spawn_blocking_background<F, R>(priority: CallbackPriority, f: F) -> BackgroundTaskHandle<R>
500where
501    F: FnOnce() -> R + Send + 'static,
502    R: Send + 'static,
503{
504    use crate::runtime::background::spawn_blocking_on;
505    spawn_blocking_on(&background().callbacks().handle(), priority, f)
506}
507
508/// Sleep on the process-global delayed-callback timer — C
509/// `callbackRequestDelayed` (callback.c:410) — measured on `std::time`.
510///
511/// The delay counterpart to [`spawn_background`]: a tail deferred there must
512/// wait on a timer that exists without a runtime, which `tokio::time` does not.
513pub async fn sleep_background(duration: Duration) {
514    crate::runtime::background::timer_sleep::sleep(&background().timer().handle(), duration).await;
515}
516
517/// Periodic ticker on the process-global delayed-callback timer — the
518/// [`interval`] counterpart for work spawned by [`spawn_background`].
519pub fn interval_background(period: Duration) -> crate::runtime::background::TimerInterval {
520    crate::runtime::background::timer_sleep::interval(&background().timer().handle(), period)
521}
522
523// ---------------------------------------------------------------------------
524// The reactor capability
525//
526// Spawning used to go through an *ambient* free function. On a hosted build
527// `spawn` was `tokio::spawn`, which reads a thread-local and panics with
528// "there is no reactor running" on any thread that is not inside a runtime; on
529// `exec_backend` it was `spawn_background`, which lands on the process-global
530// background executor and has no reactor at all. One name, two executors with
531// opposite capabilities, and which one you got was decided by a thread-local
532// the call site could not see. Nothing but a comment and a textual census kept
533// a reactor-bound future off the reactor-free executor, or the reverse.
534//
535// [`Reactor`] replaces that thread-local with a value. A site that needs the
536// I/O and time drivers takes one, so the requirement is in the signature of
537// whatever owns the site instead of in a comment beside it, and a record
538// callback — which is handed no `Reactor` — cannot reach the reactor-bound
539// spawn at all without [`Reactor::current`], which is one named call a census
540// can ban outright in the scopes where it must never appear.
541// ---------------------------------------------------------------------------
542
543/// The executor that owns the reactor — the capability a reactor-bound task is
544/// spawned through.
545///
546/// Hold one and you may spawn a future that opens a `tokio::net` socket, waits
547/// on `tokio::time`, or installs a signal handler. The counterpart is
548/// [`spawn_background`], which needs no capability because the executor it
549/// lands on is process-global — and, for the same reason, has no reactor.
550///
551/// # Where one comes from
552///
553/// [`Reactor::current`] mints one from the runtime entered on the calling
554/// thread, and returns `None` rather than panicking when there is none.
555/// Everything else receives one: the IOC and the CA/PVA servers and clients
556/// capture it once on their own setup path and hand `&Reactor` down to the
557/// tasks they start. [`BlockingBridge::reactor`] is the third source, for a
558/// blocking thread the runtime is not entered on.
559#[cfg(tokio_backend)]
560#[derive(Clone, Debug)]
561pub struct Reactor {
562    handle: tokio::runtime::Handle,
563}
564
565/// See the `tokio_backend` definition.
566///
567/// **There is no tokio reactor on this backend.** A `Reactor` here is the
568/// process-global background executor, so it is a ZST and
569/// [`Reactor::current`] always succeeds — which is what lets one signature
570/// compile on both backends. What it does *not* do is make `tokio::net` work
571/// on RTEMS: that build reaches the network through the blocking drivers, and
572/// [`HAS_TOKIO_REACTOR`] is the constant that says which world a call site is
573/// compiled into.
574///
575/// Deliberately **not** `Copy`, though a ZST could be: a capability that moves
576/// on one backend and copies on the other makes every call site's `clone()`
577/// correct under `tokio_backend` and a `clippy::clone_on_copy` error here. One
578/// shape on both backends is what lets the same source compile clean under
579/// either.
580#[cfg(exec_backend)]
581#[derive(Clone, Debug, Default)]
582pub struct Reactor {
583    /// Where [`Reactor::spawn`] puts the future. `None` is the process-global
584    /// callback pool's middle band — see [`Reactor::with_executor`].
585    exec: Option<Arc<crate::runtime::background::DedicatedExecutor>>,
586}
587
588#[cfg(tokio_backend)]
589impl Reactor {
590    /// The runtime entered on this thread, or `None` when there is none.
591    ///
592    /// Deliberately not a panicking constructor: the ambient `spawn` this
593    /// replaces panicked from inside itself, which is exactly the failure the
594    /// capability exists to move into the type. A caller that genuinely cannot
595    /// continue without one writes its own `expect`, which names the reason and
596    /// can be found by a census.
597    pub fn current() -> Option<Self> {
598        RuntimeHandle::try_current()
599            .ok()
600            .map(|handle| Self { handle })
601    }
602
603    /// Spawn a reactor-bound future.
604    ///
605    /// Works from any thread, including one the runtime is not entered on —
606    /// the handle carries the runtime, where `tokio::spawn` read a
607    /// thread-local.
608    pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
609    where
610        F: Future + Send + 'static,
611        F::Output: Send + 'static,
612    {
613        self.handle.spawn(future)
614    }
615
616    /// Run a blocking closure on the runtime's blocking pool.
617    pub fn spawn_blocking<F, R>(&self, f: F) -> TaskHandle<R>
618    where
619        F: FnOnce() -> R + Send + 'static,
620        R: Send + 'static,
621    {
622        self.handle.spawn_blocking(f)
623    }
624
625    /// The `exec_backend` twin's no-op: here the runtime *is* the executor and
626    /// its band is the runtime's to set, so `exec` is ignored. It exists so a
627    /// server that chooses its own band compiles unchanged on both backends —
628    /// see the `exec_backend` copy for what it does there.
629    pub fn with_executor(
630        &self,
631        _exec: &Arc<crate::runtime::background::DedicatedExecutor>,
632    ) -> Self {
633        self.clone()
634    }
635}
636
637#[cfg(exec_backend)]
638impl Reactor {
639    /// The background executor is process-global, so this never fails — see
640    /// the type's own docs for what that does and does not promise.
641    pub fn current() -> Option<Self> {
642        Some(Self::default())
643    }
644
645    /// This reactor, with spawned futures routed to `exec` instead of the
646    /// callback pool.
647    ///
648    /// The band a server's connection work runs in is the server's to choose,
649    /// and the reactor is the capability it is already handed — so binding the
650    /// executor here is what moves every downstream `spawn`, including the ones
651    /// in futures this reactor is cloned into, without a second handle to
652    /// thread beside it.
653    ///
654    /// The `tokio_backend` twin ignores `exec`: there the runtime *is* the
655    /// executor, and its band is the runtime's to set.
656    pub fn with_executor(&self, exec: &Arc<crate::runtime::background::DedicatedExecutor>) -> Self {
657        Self {
658            exec: Some(Arc::clone(exec)),
659        }
660    }
661
662    /// Spawn on this reactor's executor, or — with none bound —
663    /// [`spawn_background`] verbatim on the middle band, for the same reason
664    /// as the `tokio_backend` copy: the reactor seam carries no record.
665    pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
666    where
667        F: Future + Send + 'static,
668        F::Output: Send + 'static,
669    {
670        match &self.exec {
671            Some(exec) => crate::runtime::background::spawn_future(
672                &exec.handle(),
673                CallbackPriority::Medium,
674                future,
675            ),
676            None => spawn_background(CallbackPriority::Medium, future),
677        }
678    }
679
680    /// [`Reactor::spawn`] for a blocking closure. A bound executor takes it on
681    /// one of its own workers; with none bound this is
682    /// [`spawn_blocking_background`] verbatim, middle band.
683    pub fn spawn_blocking<F, R>(&self, f: F) -> TaskHandle<R>
684    where
685        F: FnOnce() -> R + Send + 'static,
686        R: Send + 'static,
687    {
688        match &self.exec {
689            Some(exec) => crate::runtime::background::spawn_blocking_on(
690                &exec.handle(),
691                CallbackPriority::Medium,
692                f,
693            ),
694            None => spawn_blocking_background(CallbackPriority::Medium, f),
695        }
696    }
697}
698
699/// Yield the current task once — the seam replacement for
700/// `tokio::task::yield_now`, so no call site names `tokio::task` directly.
701#[cfg(tokio_backend)]
702pub async fn yield_now() {
703    tokio::task::yield_now().await;
704}
705
706/// Exec-backend yield: return `Pending` once with the waker already woken.
707/// On the cooperative background executor that re-enqueues the task behind
708/// whatever else is runnable; under a `park_on` driver the wake sets the
709/// park token, so the driver re-polls immediately — both give one fair
710/// scheduling point, which is all `yield_now` promises.
711#[cfg(exec_backend)]
712pub async fn yield_now() {
713    struct YieldNow(bool);
714    impl Future for YieldNow {
715        type Output = ();
716        fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
717            if self.0 {
718                Poll::Ready(())
719            } else {
720                self.0 = true;
721                cx.waker().wake_by_ref();
722                Poll::Pending
723            }
724        }
725    }
726    YieldNow(false).await;
727}
728
729/// Run a blocking closure on a worker thread.
730///
731/// **Not the reactor capability.** What this needs is a pool of threads that
732/// may block, not the I/O and time drivers, and the two are different
733/// requirements even though tokio happens to hang both off one runtime: on a
734/// hosted build this is `tokio::task::spawn_blocking`, which does require a
735/// runtime entered on the calling thread, and on `exec_backend` it is the
736/// callback pool, which requires nothing. That residual dual meaning is why
737/// [`Reactor::spawn_blocking`] exists beside it — a site that already holds a
738/// [`Reactor`] should use that one and say so in its signature. This free
739/// function stays for the callers whose work is reactor-free by nature
740/// (`runtime::fs`, `waitpid`, a name lookup) and for whom a process-global
741/// blocking pool, which this crate does not yet have, would be the right
742/// destination.
743#[cfg(tokio_backend)]
744pub fn spawn_blocking<F, R>(f: F) -> TaskHandle<R>
745where
746    F: FnOnce() -> R + Send + 'static,
747    R: Send + 'static,
748{
749    tokio::task::spawn_blocking(f)
750}
751
752/// RTEMS: the callback pool, which needs no runtime — see the `tokio_backend`
753/// copy for why this one is not the reactor capability. Middle band: this is
754/// the ambient `spawn_blocking`, which no record reaches.
755#[cfg(exec_backend)]
756pub fn spawn_blocking<F, R>(f: F) -> TaskHandle<R>
757where
758    F: FnOnce() -> R + Send + 'static,
759    R: Send + 'static,
760{
761    spawn_blocking_background(CallbackPriority::Medium, f)
762}
763
764/// A set of spawned tasks, joined as they complete — the seam replacement for
765/// `tokio::task::JoinSet`.
766///
767/// `JoinSet` is a **fourth spelling of `tokio::spawn`**, and the one no seam
768/// guard caught: `JoinSet::spawn` calls `tokio::spawn` internally, so it panics
769/// with *"there is no reactor running"* on any thread that is not inside a
770/// tokio runtime — which on RTEMS is every callback-band worker. Measured on
771/// target: the CA client's transport manager died on `cbMedium` at its first
772/// connect. Naming it here means a call site can express "spawn a set of tasks
773/// and reap them as they finish" without reaching past the seam.
774///
775/// The three properties call sites depend on, all preserved:
776///
777/// * **Concurrency** — every member runs independently; joining one does not
778///   block the others.
779/// * **Pair-by-value** — [`Self::join_next`] yields whichever member finished
780///   first, so a task that returns its own key can be matched to its state.
781/// * **Abort on drop** — dropping the set cancels every member that has not
782///   finished, which is the property that distinguishes a `JoinSet` from a bag
783///   of detached `JoinHandle`s.
784pub struct TaskSet<T> {
785    tasks: Vec<TaskHandle<T>>,
786}
787
788impl<T> Default for TaskSet<T> {
789    fn default() -> Self {
790        Self::new()
791    }
792}
793
794impl<T> TaskSet<T> {
795    /// An empty set.
796    pub fn new() -> Self {
797        Self { tasks: Vec::new() }
798    }
799
800    /// Number of members that have not yet been joined.
801    pub fn len(&self) -> usize {
802        self.tasks.len()
803    }
804
805    /// `true` when no member is outstanding.
806    pub fn is_empty(&self) -> bool {
807        self.tasks.is_empty()
808    }
809}
810
811impl<T: Send + 'static> TaskSet<T> {
812    /// Spawn `future` into the set — through [`Reactor::spawn`], so the set is
813    /// as reactor-bound as its members and says so in this signature.
814    pub fn spawn<F>(&mut self, reactor: &Reactor, future: F)
815    where
816        F: Future<Output = T> + Send + 'static,
817    {
818        self.tasks.push(reactor.spawn(future));
819    }
820
821    /// Wait for the next member to finish and return its result, removing it
822    /// from the set. `None` when the set is empty — matching
823    /// `JoinSet::join_next`, so a `select!` arm on it goes quiet rather than
824    /// spinning once every task has been reaped.
825    ///
826    /// Cancel-safe: the returned future holds no state of its own, so a
827    /// `select!` that drops it loses nothing.
828    pub async fn join_next(&mut self) -> Option<Result<T, TaskJoinError>> {
829        if self.tasks.is_empty() {
830            return None;
831        }
832        std::future::poll_fn(|cx| {
833            for i in 0..self.tasks.len() {
834                // Both backends' handles are `Unpin` (tokio's `JoinHandle`,
835                // and `JoinFuture`, whose only field is an `Arc`), so this
836                // needs no pin projection. Polling every pending member
837                // re-registers this waker with each — the shape both handles
838                // document.
839                if let std::task::Poll::Ready(result) =
840                    std::pin::Pin::new(&mut self.tasks[i]).poll(cx)
841                {
842                    self.tasks.swap_remove(i);
843                    return std::task::Poll::Ready(Some(result));
844                }
845            }
846            std::task::Poll::Pending
847        })
848        .await
849    }
850}
851
852impl<T> Drop for TaskSet<T> {
853    /// Cancel every outstanding member — `JoinSet`'s drop behaviour, and the
854    /// reason a call site reaches for a set rather than a `Vec` of handles.
855    fn drop(&mut self) {
856        for task in &self.tasks {
857            task.abort();
858        }
859    }
860}
861
862/// The timer the *calling thread* can actually reach.
863///
864/// `tokio::time` reads a thread-local time driver and panics when there is
865/// none, but this process always has a second timer that needs no runtime at
866/// all: the delayed-callback timer [`sleep_background`] waits on (C
867/// `callbackRequestDelayed`). Which of the two to use is a property of the
868/// calling thread, not of the build, so it cannot be a `cfg`.
869///
870/// It is deliberately not "always the background timer". Under
871/// `#[tokio::test(start_paused = true)]` tokio's clock is virtual and only
872/// tokio's own timer advances with it, so a wall-clock sleep on a runtime
873/// thread would wait out a deadline the test means to skip. A thread with no
874/// runtime has no virtual clock to honour, so there the wall-clock timer is
875/// both the correct one and the only one — which is why this arms a timer
876/// where the seam used to panic instead.
877#[cfg(tokio_backend)]
878async fn sleep_on_reachable_timer(duration: Duration) {
879    if tokio::runtime::Handle::try_current().is_ok() {
880        tokio::time::sleep(duration).await;
881    } else {
882        sleep_background(duration).await;
883    }
884}
885
886#[cfg(tokio_backend)]
887pub async fn sleep(duration: Duration) {
888    sleep_on_reachable_timer(duration).await;
889}
890
891/// RTEMS: see [`Reactor::spawn`] — this backend's only executor is the
892/// background one.
893#[cfg(exec_backend)]
894pub async fn sleep(duration: Duration) {
895    sleep_background(duration).await;
896}
897
898/// The instant [`sleep_until`] measures deadlines against — **the backend's own
899/// clock**, which is the whole point of naming it here.
900///
901/// A deadline is only meaningful in the clock the timer that waits on it runs
902/// on. The hosted timer is tokio's, and under `#[tokio::test(start_paused =
903/// true)]` tokio's clock is virtual and advances on `sleep`, not with the wall
904/// — so a `std::time::Instant` deadline handed to a tokio timer is a deadline
905/// in a *different* timeline, and the wait is wrong by however far the two have
906/// diverged. The RTEMS timer runs on `std::time::Instant` (1-second-quantized
907/// on target).
908///
909/// Taking the alias rather than a concrete instant type is what keeps a caller
910/// from mixing them: `Instant::now() + timeout` is the deadline `sleep_until`
911/// will actually honour, on both backends.
912#[cfg(tokio_backend)]
913pub type Instant = tokio::time::Instant;
914/// See the hosted definition.
915#[cfg(exec_backend)]
916pub type Instant = std::time::Instant;
917
918/// `base + d` on the runtime's own [`Instant`], saturating where the bare
919/// `+` would panic — [`crate::runtime::time::deadline_after`]'s rule for
920/// the alias above.
921///
922/// The alias exists so a deadline cannot be built in the wrong timeline;
923/// this exists so it cannot be built by a panicking add. Any `Duration`
924/// that came from a `double` — a record field, an `EPICS_*` env var, a
925/// caller-supplied timeout — can be [`Duration::MAX`] by the rule in
926/// [`crate::runtime::time::duration_from_secs`], and `Instant + MAX`
927/// aborts the task rather than never firing. Both concrete `Instant`
928/// types offer `checked_add`, so one body covers both backends.
929pub fn deadline_after(base: Instant, d: Duration) -> Instant {
930    base.checked_add(d)
931        .unwrap_or_else(|| Instant::now() + crate::runtime::time::FAR_FUTURE)
932}
933
934/// `base` = now. See [`deadline_after`].
935pub fn deadline_from_now(d: Duration) -> Instant {
936    deadline_after(Instant::now(), d)
937}
938
939#[cfg(tokio_backend)]
940pub async fn sleep_until(deadline: Instant) {
941    if tokio::runtime::Handle::try_current().is_ok() {
942        tokio::time::sleep_until(deadline).await;
943    } else {
944        // No runtime means no virtual clock, so `Instant::now()` here reads the
945        // same wall clock the background timer runs on and the subtraction
946        // loses nothing. See [`sleep_on_reachable_timer`].
947        sleep_on_reachable_timer(deadline.saturating_duration_since(Instant::now())).await;
948    }
949}
950
951/// RTEMS: sleep-until on the delayed-callback timer via the host-tested `Sleep`.
952#[cfg(exec_backend)]
953pub async fn sleep_until(deadline: Instant) {
954    crate::runtime::background::timer_sleep::sleep_until(&background().timer().handle(), deadline)
955        .await;
956}
957
958/// Periodic ticker — the seam replacement for `tokio::time::interval`, so no
959/// production site names `tokio::time` directly (decision A2). The hosted build
960/// wraps `tokio::time::Interval`, preserving its default
961/// `MissedTickBehavior::Burst` catch-up and immediate first tick; the RTEMS
962/// build substitutes the runtime-free
963/// [`crate::runtime::background::timer_sleep::TimerInterval`], which reproduces
964/// the same semantics over the delayed-callback timer.
965#[cfg(tokio_backend)]
966pub struct Interval {
967    inner: tokio::time::Interval,
968}
969
970#[cfg(tokio_backend)]
971impl Interval {
972    /// Complete at the next tick. The first tick is immediate (tokio parity);
973    /// callers that want to skip it await `tick()` once up front.
974    pub async fn tick(&mut self) {
975        self.inner.tick().await;
976    }
977}
978
979/// RTEMS: the periodic ticker is the runtime-free `TimerInterval` (same
980/// immediate-first-tick + Burst catch-up semantics, same `tick()` surface).
981#[cfg(exec_backend)]
982pub type Interval = crate::runtime::background::timer_sleep::TimerInterval;
983
984/// Build a periodic ticker firing every `period` — the seam replacement for
985/// `tokio::time::interval`.
986#[cfg(tokio_backend)]
987pub fn interval(period: Duration) -> Interval {
988    Interval {
989        inner: tokio::time::interval(period),
990    }
991}
992
993/// RTEMS: build the periodic ticker on the delayed-callback timer.
994#[cfg(exec_backend)]
995pub fn interval(period: Duration) -> Interval {
996    crate::runtime::background::timer_sleep::interval(&background().timer().handle(), period)
997}
998
999/// The timeout's error — "the deadline elapsed before the future completed".
1000///
1001/// The seam's own type on **both** backends, where the hosted half used to
1002/// re-export `tokio::time::error::Elapsed`. It has to be: [`timeout`] returns
1003/// this error without arming a timer when the budget is already spent, and
1004/// tokio's `Elapsed` has no public constructor (`pub(crate) fn new`), so an
1005/// alias makes the expired branch unwritable. No call site in this workspace
1006/// names the type — every one of the 81 discards or maps it — so the change is
1007/// invisible above the seam.
1008#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1009pub struct Elapsed(());
1010
1011impl std::fmt::Display for Elapsed {
1012    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1013        write!(f, "deadline has elapsed")
1014    }
1015}
1016
1017impl std::error::Error for Elapsed {}
1018
1019/// C's floor for "the remaining budget is worth blocking on":
1020/// `CAC_SIGNIFICANT_DELAY`, 1 µs
1021/// (`epics-base/modules/ca/src/client/iocinf.h:37`; the `1.0/CLOCKS_PER_SEC`
1022/// spelling at `:35` is the other platform's arm of the same `#if`).
1023pub const SIGNIFICANT_DELAY: Duration = Duration::from_micros(1);
1024
1025/// Poll `fut` once and give up — the shape C's `pendIO` has when the budget is
1026/// already spent.
1027///
1028/// `ca_client_context::pendIO` (`ca_client_context.cpp:490-499`) tests the
1029/// outstanding-work counter first and *then* compares `remaining <
1030/// CAC_SIGNIFICANT_DELAY`, returning `ECA_TIMEOUT` before it ever blocks. So an
1031/// expired budget in C decides on state alone; nothing races it. One poll of
1032/// `fut` is that same "is the work already done" test, and the `Pending` arm is
1033/// C's immediate `ECA_TIMEOUT`.
1034async fn poll_once_then_expire<F: Future>(fut: F) -> Result<F::Output, Elapsed> {
1035    let mut fut = std::pin::pin!(fut);
1036    std::future::poll_fn(move |cx| {
1037        Poll::Ready(match fut.as_mut().poll(cx) {
1038            Poll::Ready(v) => Ok(v),
1039            Poll::Pending => Err(Elapsed(())),
1040        })
1041    })
1042    .await
1043}
1044
1045/// Await `fut`, giving up after `duration` — the seam's only bounded wait.
1046///
1047/// It belongs here rather than at the call sites for the same reason [`sleep`]
1048/// does: a deadline needs a timer, and which timer that is depends on the
1049/// backend and on the calling thread. Call sites that reach for
1050/// `tokio::time::timeout` directly pin themselves to the tokio timer wheel,
1051/// which exists on one backend and only inside a runtime.
1052///
1053/// A budget under [`SIGNIFICANT_DELAY`] never reaches a timer at all. It cannot:
1054/// a timeout polls the inner future first and only then arms its sleep, and a
1055/// zero-length sleep is not reported elapsed until the time driver next runs — so on a loaded machine the inner future wins a race that
1056/// C does not have, and an already-expired `-w` returns success. Measured as
1057/// `caget -w -1` exiting 0 where C exits 1
1058/// (`tool_lib.c:628-638` via `ca_client_context.cpp:490-499`).
1059///
1060/// One body for both backends, raced against [`sleep`], because the backend was
1061/// never the question a bounded wait had to ask. The hosted half called
1062/// `tokio::time::timeout`, which panics on a thread with no runtime, so the
1063/// seam had a deadline the exec backend honoured for every caller and the
1064/// hosted one honoured only inside a runtime. Callers had no way to express
1065/// that difference and did not try to: `asyn-rs`'s `PortHandle::await_reply`
1066/// simply lost its `queue_timeout` on every plain-thread `submit_blocking`.
1067/// [`sleep`] now picks a timer the calling thread can reach, so the race below
1068/// fires wherever it is polled.
1069pub async fn timeout<F: Future>(duration: Duration, fut: F) -> Result<F::Output, Elapsed> {
1070    if duration < SIGNIFICANT_DELAY {
1071        return poll_once_then_expire(fut).await;
1072    }
1073    let mut sleep = std::pin::pin!(sleep(duration));
1074    let mut fut = std::pin::pin!(fut);
1075    std::future::poll_fn(move |cx| {
1076        if let Poll::Ready(v) = fut.as_mut().poll(cx) {
1077            return Poll::Ready(Ok(v));
1078        }
1079        if sleep.as_mut().poll(cx).is_ready() {
1080            return Poll::Ready(Err(Elapsed(())));
1081        }
1082        Poll::Pending
1083    })
1084    .await
1085}
1086
1087/// [`timeout`] against an absolute [`Instant`] instead of a duration — the
1088/// seam's `tokio::time::timeout_at`.
1089///
1090/// It exists because one deadline shared across several sequential awaits is a
1091/// different bound from a fresh duration per await: `caget_many` gives its
1092/// whole batch one deadline, so a slow first PV eats the budget the rest would
1093/// otherwise each get in full. Expressing that with `timeout` would need the
1094/// caller to do the subtraction, which is the arithmetic that drifts.
1095///
1096/// Raced against [`sleep_until`], the absolute-deadline twin of what
1097/// [`timeout`] races against, and one body for the same reason.
1098pub async fn timeout_at<F: Future>(deadline: Instant, fut: F) -> Result<F::Output, Elapsed> {
1099    if deadline.saturating_duration_since(Instant::now()) < SIGNIFICANT_DELAY {
1100        return poll_once_then_expire(fut).await;
1101    }
1102    let mut sleep = std::pin::pin!(sleep_until(deadline));
1103    let mut fut = std::pin::pin!(fut);
1104    std::future::poll_fn(move |cx| {
1105        if let Poll::Ready(v) = fut.as_mut().poll(cx) {
1106            return Poll::Ready(Ok(v));
1107        }
1108        if sleep.as_mut().poll(cx).is_ready() {
1109            return Poll::Ready(Err(Elapsed(())));
1110        }
1111        Poll::Pending
1112    })
1113    .await
1114}
1115
1116pub fn runtime_handle() -> tokio::runtime::Handle {
1117    tokio::runtime::Handle::current()
1118}
1119
1120// ---------------------------------------------------------------------------
1121// EPICS thread priority abstraction
1122//
1123// C parity: `modules/libcom/src/osi/epicsThread.h:73-92` defines an
1124// integer priority space `0..=99` (`epicsThreadPriorityMin/Max`) with a
1125// set of named levels, plus three stack-size classes.
1126// `osi/os/posix/osdThread.c` maps an EPICS priority `p` onto the OS
1127// SCHED_FIFO range with `oss = p * (max-min)/100 + min` and falls back
1128// to a non-RT (default-policy) thread when the process lacks permission
1129// to use SCHED_FIFO.
1130//
1131// The Rust port runs work as tokio tasks on a shared pool, so there is
1132// no per-task OS thread to re-prioritise for `spawn`. What is portably
1133// achievable is: (a) the priority enum + named levels as a first-class
1134// type, (b) a stack-size class with the C size table, and (c) a
1135// best-effort OS-scheduler priority applied to the *current* OS thread
1136// (used by dedicated `spawn_blocking` threads and the runtime's worker
1137// threads). `apply_to_current_thread` reports whether the OS actually
1138// honoured the request.
1139//
1140// (c) is opt-in and off by default — see `RT_PRIORITY_ENV`. C's switch
1141// (`EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING`) defaults to YES because
1142// a C IOC is deployed onto a machine chosen for it; this crate is just as
1143// often linked into a desktop tool, where a silent SCHED_FIFO request is
1144// either a guaranteed failure or a way to starve the box.
1145// ---------------------------------------------------------------------------
1146
1147/// Minimum EPICS thread priority (`epicsThreadPriorityMin`).
1148pub const PRIORITY_MIN: u8 = 0;
1149/// Maximum EPICS thread priority (`epicsThreadPriorityMax`).
1150pub const PRIORITY_MAX: u8 = 99;
1151
1152/// EPICS thread priority — an integer `0..=99` with the named levels
1153/// from `epicsThreadPriority*` (`epicsThread.h:73-83`). Lower values
1154/// are lower priority; the CA server bands sit below the scan bands so
1155/// scan threads preempt CA-server threads on a loaded IOC.
1156#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
1157pub enum ThreadPriority {
1158    /// `epicsThreadPriorityLow` = 10.
1159    Low,
1160    /// `epicsThreadPriorityCAServerLow` = 20.
1161    CaServerLow,
1162    /// `epicsThreadPriorityCAServerHigh` = 40.
1163    CaServerHigh,
1164    /// `epicsThreadPriorityMedium` = 50.
1165    Medium,
1166    /// `epicsThreadPriorityScanLow` = 60.
1167    ScanLow,
1168    /// `epicsThreadPriorityScanHigh` = 70.
1169    ScanHigh,
1170    /// `epicsThreadPriorityHigh` = 90.
1171    High,
1172    /// `epicsThreadPriorityIocsh` = 91.
1173    Iocsh,
1174    /// An explicit priority value, clamped to `0..=99` on use.
1175    Custom(u8),
1176}
1177
1178impl ThreadPriority {
1179    /// The raw EPICS priority value `0..=99`, matching the
1180    /// `epicsThreadPriority*` constants in `epicsThread.h`.
1181    ///
1182    /// `const` so a server can *derive* its band from the named one C derives
1183    /// it from — `CaServerLow - 2` rather than a bare `18` with a comment
1184    /// asserting the two are the same number. C builds exactly that ladder at
1185    /// `caservertask.c:563-575`; restating its output as a literal is how the
1186    /// ladder and its constants come to disagree.
1187    pub const fn value(self) -> u8 {
1188        let v = match self {
1189            ThreadPriority::Low => 10,
1190            ThreadPriority::CaServerLow => 20,
1191            ThreadPriority::CaServerHigh => 40,
1192            ThreadPriority::Medium => 50,
1193            ThreadPriority::ScanLow => 60,
1194            ThreadPriority::ScanHigh => 70,
1195            ThreadPriority::High => 90,
1196            ThreadPriority::Iocsh => 91,
1197            ThreadPriority::Custom(v) => v,
1198        };
1199        // `Ord::min` is not `const`; the clamp is the same one.
1200        if v > PRIORITY_MAX { PRIORITY_MAX } else { v }
1201    }
1202}
1203
1204/// Stack-size class — `epicsThreadStackSizeClass` (`epicsThread.h:91`).
1205///
1206/// The byte size is implementation-dependent in C. These mirror the POSIX
1207/// table `STACK_SIZE(f) = f * 0x10000 * sizeof(void*)`
1208/// (`libcom/src/osi/os/posix/osdThread.c:506-509`), pointer-width
1209/// parameterised exactly as the C macro is: Small = 1, Medium = 2, Big = 4
1210/// units of `0x10000 * sizeof(void*)`. On a 64-bit host that is
1211/// 512 KiB / 1 MiB / 2 MiB; on `armv7-rtems-eabihf` it is
1212/// **256 KiB / 512 KiB / 1 MiB**.
1213///
1214/// # This is the table a C IOC on RTEMS 6 uses too
1215///
1216/// Base has a second, much smaller table — 5000 / 8000 / 11000 bytes, floored
1217/// at `RTEMS_MINIMUM_STACK_SIZE` — in `os/RTEMS-score/osdThread.c:136-150`.
1218/// It does not apply here. `configure/toolchain.c:29-35` selects
1219/// `OS_API = posix` for `__RTEMS_MAJOR__ >= 5`, so an RTEMS 6 build searches
1220/// `os/RTEMS-posix` then `os/RTEMS`, neither of which contains an
1221/// `osdThread.c`, and lands on `os/posix/osdThread.c` — the file above. The
1222/// score table is what RTEMS **4/5** used.
1223///
1224/// Do not "align" these constants with 5000/8000/11000. That would be a
1225/// regression against the C IOC we are matching, not a correction: on RTEMS 6
1226/// the C IOC asks `pthread_attr_setstacksize` for the POSIX number
1227/// (`os/posix/osdThread.c:212-215`), which is the same call `std` makes for
1228/// us, with the same argument.
1229#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1230pub enum StackSizeClass {
1231    Small,
1232    Medium,
1233    Big,
1234}
1235
1236impl StackSizeClass {
1237    /// Stack size in bytes for this class, matching the POSIX
1238    /// `stackSizeTable` in `osdThread.c` on **every** target.
1239    ///
1240    /// C's `STACK_SIZE(f) = f * 0x10000 * sizeof(void*)` is parameterised by
1241    /// pointer width and so is this, so the two agree by construction rather
1242    /// than on one word size: 512 KiB / 1 MiB / 2 MiB on a 64-bit host, and
1243    /// 256 KiB / 512 KiB / 1 MiB on `armv7-rtems-eabihf` — which is exactly
1244    /// what a C IOC asks `pthread_attr_setstacksize` for on that target.
1245    ///
1246    /// Read "on a 64-bit target" here before: it was wrong in the direction
1247    /// that matters, because it invited a reader to assume the RTEMS numbers
1248    /// were unverified. This crate is portable to 64-bit embedded targets
1249    /// too — `x86_64-wrs-vxworks` — and pays for it: a 64-bit pointer doubles
1250    /// every class in this table, so an `x86_64-wrs-vxworks` CA client thread
1251    /// costs exactly 2× what the same thread costs on `armv7-rtems-eabihf`,
1252    /// pointer width for pointer width, not a difference in the formula.
1253    pub fn bytes(self) -> usize {
1254        // STACK_SIZE(f) = f * 0x10000 * sizeof(void*)
1255        let unit = 0x10000usize * std::mem::size_of::<usize>();
1256        match self {
1257            StackSizeClass::Small => unit,
1258            StackSizeClass::Medium => 2 * unit,
1259            StackSizeClass::Big => 4 * unit,
1260        }
1261    }
1262}
1263
1264/// Outcome of a best-effort OS-scheduler priority change.
1265#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1266pub enum PriorityApplied {
1267    /// The OS scheduler honoured the requested priority (real-time
1268    /// SCHED_FIFO band applied).
1269    Realtime,
1270    /// Real-time scheduling was never requested: the opt-in switch
1271    /// [`RT_PRIORITY_ENV`] is off, so **no scheduler call was made at
1272    /// all** and the thread keeps the process default policy.
1273    Disabled,
1274    /// The platform does not expose a portable scheduler priority API
1275    /// (e.g. Windows here, or a non-Unix target) — no change applied.
1276    Unsupported,
1277    /// The platform exposes the API but rejected the request (typically
1278    /// the process lacks `CAP_SYS_NICE`/root for SCHED_FIFO). C's
1279    /// `osdThread.c` makes the same best-effort fall back to a non-RT
1280    /// thread in this case (`osdThread.c:647` "Try again without
1281    /// SCHED_FIFO").
1282    BestEffortFailed,
1283}
1284
1285impl PriorityApplied {
1286    /// `true` only when the OS actually applied a real-time priority.
1287    pub fn is_realtime(self) -> bool {
1288        matches!(self, PriorityApplied::Realtime)
1289    }
1290}
1291
1292/// Environment switch that opts this process in to real-time (SCHED_FIFO)
1293/// scheduling for the IOC threads that carry an EPICS priority.
1294///
1295/// The switch is read in both directions on every target; what differs is
1296/// what it defaults to when unset — see [`DEFAULT_POLICY`].
1297///
1298/// Accepted "on" values, case-insensitive: `YES`, `TRUE`, `ON`, `1`.
1299/// Any other *explicit* value is off.
1300///
1301/// # Relationship to the C switch
1302///
1303/// C base has the same concept under
1304/// `EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING` (`envDefs.h:80`, read at
1305/// `osdThread.c:389`), and `envGetBoolConfigParam` (`envSubr.c:331`) accepts
1306/// only case-insensitive `yes`. We deliberately do **not** reuse that name:
1307/// its base default is `YES` on every target (`configure/CONFIG_ENV:57`)
1308/// while ours is `YES` only on RTEMS, so one name would carry two different
1309/// defaults on a hosted build depending on which implementation read it.
1310pub const RT_PRIORITY_ENV: &str = "EPICS_RS_ALLOW_RT_PRIORITY";
1311
1312/// Whether this process may ask the OS for real-time scheduling.
1313///
1314/// Resolved from [`RT_PRIORITY_ENV`] exactly once per process by
1315/// [`RtPolicy::current`]. It is a *parameter* of
1316/// [`apply_to_current_thread_under`] rather than a check buried inside the
1317/// syscall wrapper, so "switch off ⟹ no scheduler call" is a property of
1318/// the call graph and not of a runtime branch some future caller can skip.
1319#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1320pub enum RtPolicy {
1321    /// Never touch the OS scheduler.
1322    Disabled,
1323    /// Best-effort SCHED_FIFO, falling back to default scheduling.
1324    AllowRealtime,
1325}
1326
1327/// What [`RT_PRIORITY_ENV`] means when it is **unset**, for the target this
1328/// was compiled for.
1329///
1330/// `AllowRealtime` on RTEMS, `Disabled` everywhere else. The asymmetry is
1331/// deliberate and is a property of the default itself — not a `setenv` the
1332/// boot shim performs before `main()`. A `setenv` is a runtime side effect any
1333/// later caller can undo or reorder, and a variable one component writes for
1334/// another to read is the dual-meaning shape this code keeps removing.
1335///
1336/// Why the embedded targets differ from hosted:
1337///
1338///   - Base's own equivalent switch,
1339///     `EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING`, defaults to `YES`
1340///     (`configure/CONFIG_ENV:57`). An IOC that honours its priorities is
1341///     upstream's default posture, not an opt-in.
1342///   - The opt-in gate exists for RT-Linux, where asking for SCHED_FIFO needs
1343///     `CAP_SYS_NICE` or a non-zero `RLIMIT_RTPRIO` (so on a desktop the
1344///     request merely fails), and where a runaway RT band on a box that
1345///     *grants* it can wedge a developer's machine. Neither failure mode
1346///     exists on RTEMS or VxWorks: there is no RLIMIT_RTPRIO, no
1347///     `CAP_SYS_NICE` gate, and no desktop to wedge on either.
1348///   - The band invariant is now a test rather than a hope —
1349///     `rtems_priority_map_stays_below_the_libbsd_network_band` proves every
1350///     u8 input lands in core 100..199, at or below libbsd's default band and
1351///     strictly less urgent than IRQS(96)/TIME(98).
1352///   - **VxWorks is measurement-backed, not assumed.** On the bring-up box
1353///     (VxWorks 7, `x86_64-wrs-vxworks`), 11 of 11 measured threads landed
1354///     `PriorityApplied::Realtime` via `SCHED_FIFO`, exactly one scheduler
1355///     call each, at `posix = 56 + epics` — the same POSIX value RTEMS gets
1356///     (see `map_epics_priority_rtems`) — which VxWorks's own POSIX layer
1357///     then inverts into its native task-priority space at `vx = 199 -
1358///     epics`, exact: EPICS base's own vxWorks-port formula
1359///     (`vxWorks/osdThread.c:99`), reached by construction rather than by
1360///     restating it (see `map_epics_priority_vxworks`).
1361///
1362/// An explicit value still wins in **both** directions on every target, so
1363/// `EPICS_RS_ALLOW_RT_PRIORITY=NO` turns it off on RTEMS or VxWorks.
1364pub const DEFAULT_POLICY: RtPolicy = default_policy(cfg!(epics_embedded_target));
1365
1366/// [`DEFAULT_POLICY`] as a pure function of the one target fact it depends
1367/// on, so both arms are reachable from a host test run. A host CI will never
1368/// execute the RTEMS arm otherwise, and an untested default is exactly the
1369/// kind that drifts.
1370const fn default_policy(on_rtems: bool) -> RtPolicy {
1371    if on_rtems {
1372        RtPolicy::AllowRealtime
1373    } else {
1374        RtPolicy::Disabled
1375    }
1376}
1377
1378impl RtPolicy {
1379    /// Parse a raw switch value (`None` = unset ⇒ [`DEFAULT_POLICY`]).
1380    pub fn from_env_value(raw: Option<&str>) -> RtPolicy {
1381        Self::resolve(raw, DEFAULT_POLICY)
1382    }
1383
1384    /// [`Self::from_env_value`] with the unset-default injected, so a host
1385    /// test can ask what an RTEMS process would do with the same input.
1386    pub fn resolve(raw: Option<&str>, default: RtPolicy) -> RtPolicy {
1387        let Some(raw) = raw else {
1388            return default;
1389        };
1390        let v = raw.trim();
1391        let on = v.eq_ignore_ascii_case("yes")
1392            || v.eq_ignore_ascii_case("true")
1393            || v.eq_ignore_ascii_case("on")
1394            || v == "1";
1395        if on {
1396            RtPolicy::AllowRealtime
1397        } else {
1398            RtPolicy::Disabled
1399        }
1400    }
1401
1402    /// The process-wide policy, read from [`RT_PRIORITY_ENV`] on first use
1403    /// and cached. Caching matches C, which resolves its switch once in
1404    /// `epicsThreadInit` (`osdThread.c:389`), and keeps later `set_var`
1405    /// calls from changing the scheduling of threads already running.
1406    pub fn current() -> RtPolicy {
1407        static POLICY: std::sync::OnceLock<RtPolicy> = std::sync::OnceLock::new();
1408        *POLICY.get_or_init(|| {
1409            RtPolicy::from_env_value(std::env::var(RT_PRIORITY_ENV).ok().as_deref())
1410        })
1411    }
1412}
1413
1414/// Apply an EPICS [`ThreadPriority`] to the **current OS thread**, best
1415/// effort.
1416///
1417/// C parity: mirrors `osdThread.c`'s SCHED_FIFO mapping
1418/// `oss = p * (max-min)/100 + min` over the kernel's
1419/// `sched_get_priority_min/max(SCHED_FIFO)` range, and the
1420/// EPERM-fallback to a non-RT thread.
1421///
1422/// Returns [`PriorityApplied`] describing what the platform allowed —
1423/// callers running in environments without RT permission still get a
1424/// running thread, just at the default policy, exactly as a C IOC does.
1425///
1426/// Note: tokio tasks spawned via [`Reactor::spawn`] share worker threads, so
1427/// this is meaningful for [`Reactor::spawn_blocking`] closures and for tuning
1428/// the runtime's worker threads at startup — not for individual async
1429/// tasks.
1430///
1431/// Platform support: the OS-scheduler change is wired on Linux, via the
1432/// range-probed linear map of `os/posix/osdThread.c`, and on RTEMS, via the
1433/// fixed map of `os/RTEMS-score/osdThread.c` inverted into POSIX space (see
1434/// `map_epics_priority_rtems` — the two maps differ in shape, deliberately).
1435/// On other targets the priority enum + API surface still exist but `apply`
1436/// reports [`PriorityApplied::Unsupported`] — no band has been measured there.
1437///
1438/// Opt-in: real-time scheduling is only ever requested when
1439/// [`RT_PRIORITY_ENV`] is set (see [`RtPolicy`]). With the switch off this
1440/// returns [`PriorityApplied::Disabled`] without calling the OS at all.
1441pub fn apply_to_current_thread(priority: ThreadPriority) -> PriorityApplied {
1442    apply_to_current_thread_under(RtPolicy::current(), priority)
1443}
1444
1445/// [`apply_to_current_thread`] with the real-time policy supplied by the
1446/// caller instead of read from the environment.
1447///
1448/// The single gate: [`RtPolicy::Disabled`] returns before any scheduler
1449/// call is reachable. Exposed so a caller that already owns its RT policy
1450/// (and the tests that must exercise both states in one process, since
1451/// [`RtPolicy::current`] is cached) does not have to mutate the environment.
1452pub fn apply_to_current_thread_under(
1453    policy: RtPolicy,
1454    priority: ThreadPriority,
1455) -> PriorityApplied {
1456    match policy {
1457        RtPolicy::Disabled => PriorityApplied::Disabled,
1458        RtPolicy::AllowRealtime => apply_priority_impl(priority.value()),
1459    }
1460}
1461
1462thread_local! {
1463    /// C `epicsThreadOSD::isOkToBlock` (`osdThread.c`).
1464    ///
1465    /// `true` here and `false` set by [`enter_ioc_thread`] reproduces C's two
1466    /// defaults without a second registry: `create_threadInfo` `calloc`s the
1467    /// field to 0 for every `epicsThreadCreate` thread, and `createImplicit`
1468    /// (`osdThread.c:710`) sets it to 1 for every thread that reaches the
1469    /// epicsThread API without having been created by it. Our EPICS threads are
1470    /// exactly the ones that pass the prologue.
1471    static OK_TO_BLOCK: std::cell::Cell<bool> = const { std::cell::Cell::new(true) };
1472}
1473
1474/// C `epicsThreadIsOkToBlock` (`osdThread.c:1145-1150`).
1475///
1476/// "May this thread wait on something that is not bounded by its own work?" —
1477/// asked by facilities that would otherwise stall a scan or a serving loop to
1478/// keep themselves in step. `runtime::log`'s errlog is the caller C has
1479/// (`errlog.c:186`); the answer is what stops a real-time thread from waiting
1480/// on the log drain and what stops the log's own worker from waiting on itself.
1481pub fn thread_is_ok_to_block() -> bool {
1482    OK_TO_BLOCK.with(std::cell::Cell::get)
1483}
1484
1485/// C `epicsThreadSetOkToBlock` (`osdThread.c:1152-1157`).
1486///
1487/// C's callers are `iocsh` around a script (`iocsh.cpp:1122`, restored at
1488/// `:1327`) and `iocInit` (`iocInit.c:126`): both raise a thread that WAS
1489/// created by `epicsThreadCreate` back to blocking for the duration of a boot,
1490/// because a boot is not real-time work. A thread that never passed
1491/// [`enter_ioc_thread`] is already blocking and needs no call.
1492pub fn set_thread_ok_to_block(ok: bool) {
1493    OK_TO_BLOCK.with(|c| c.set(ok));
1494}
1495
1496/// The prologue an IOC thread runs as its first statement, when it takes on
1497/// its role: publish its name to the OS, then request its scheduling band.
1498///
1499/// Two things a thread owes the operator, and they have different gates.
1500/// The band is opt-in ([`RT_PRIORITY_ENV`]) and best effort. The **name** is
1501/// unconditional: a thread that cannot be identified in a task listing
1502/// cannot be diagnosed, and on RTEMS that listing is often the only
1503/// instrument there is — bring-up had to measure libbsd's priority band by
1504/// other means precisely because none of our threads carried a name the
1505/// kernel could show.
1506///
1507/// Use this rather than [`apply_to_current_thread`] at a thread's entry, so
1508/// naming cannot be forgotten by the next thread somebody adds. Call
1509/// [`apply_to_current_thread`] directly only when re-banding a thread that
1510/// is already named and running. A thread that deliberately takes no EPICS
1511/// band — the iocsh script runners — calls [`name_current_thread`] alone
1512/// rather than inventing a priority just to be visible.
1513///
1514/// The band this asks the OS for is also what orders the blocking locks in
1515/// `server::database::record_lock` and its siblings: they are
1516/// priority-inheritance mutexes, so the wait queue is the *kernel's* and it
1517/// is ranked by the scheduling priority requested here. With
1518/// [`RtPolicy::Disabled`] no scheduler call happens and there is no ordering
1519/// to have — the hosted default, where the locks still exclude but do not
1520/// prioritise ([`crate::runtime::sync::is_pi_mutex_active`]).
1521/// A third thing on VxWorks, for the same reason as the name: an RTP cannot
1522/// enumerate its own tasks, so the statistics funnel's thread census is built
1523/// from what announces itself here. This is the seam because it is already the
1524/// one every IOC thread passes through to take its band — "every thread that
1525/// bands itself registers itself" adds a consequence to that invariant rather
1526/// than a rule to remember at each spawn. A thread that starts outside it is
1527/// invisible to that census, and the census output says so in its own header.
1528pub fn enter_ioc_thread(priority: ThreadPriority) -> PriorityApplied {
1529    name_current_thread();
1530    // C `create_threadInfo` leaves `isOkToBlock` at 0 for every thread
1531    // `epicsThreadCreate` makes, and `iocsh` raises it back to 1 for the
1532    // thread running a shell (`iocsh.cpp:1122`, restored at `:1327`) — as
1533    // does `iocInit` for the thread that boots (`iocInit.c:126`), which here
1534    // is that same thread running the script's `iocInit` line. The band is
1535    // the identity: `ThreadPriority::Iocsh` is taken by the three iocsh
1536    // threads and by nothing else, which `ioc_app.rs`'s
1537    // `iocsh_threads_take_the_iocsh_band` pins. Deciding it here rather than
1538    // at those three spawns keeps one owner for the flag, so a fourth shell
1539    // cannot start without it.
1540    set_thread_ok_to_block(priority == ThreadPriority::Iocsh);
1541    #[cfg(target_os = "vxworks")]
1542    epics_rtems_boot::stats::register_task();
1543    let applied = apply_to_current_thread(priority);
1544    thread_registry::register_current(priority, applied);
1545    applied
1546}
1547
1548/// Put the calling thread on the thread list as C's `_main_`.
1549///
1550/// C's `epicsThreadInit` runs `once()` under a `pthread_once`
1551/// (`osdThread.c:406-412`): it builds a `threadInfo` for the first thread to
1552/// touch the epicsThread API, names it `_main_`, gives it EPICS priority 0 and
1553/// `ellAdd`s it to `pthreadList` — without touching that thread's scheduling.
1554/// In a C IOC that thread is the one running `iocsh()`, which is why
1555/// `epicsThreadShowAll` there always lists the shell's own thread.
1556///
1557/// [`enter_ioc_thread`] cannot stand in for this. It bands the thread to the
1558/// priority it is handed, which C never does to `_main_`, and it takes the
1559/// row's name from `std::thread::current().name()` — `None`, and so `noname`,
1560/// for a shell started with a bare `std::thread::spawn`.
1561///
1562/// Idempotent for the reason C's guard is a `pthread_once`: a process has one
1563/// `_main_`. The row leaves the list when the registering thread ends, which
1564/// is where C's thread-specific-data destructor removes it.
1565pub fn register_main_thread() {
1566    thread_registry::register_main();
1567}
1568
1569/// Push `std::thread::current().name()` down to the OS thread object.
1570///
1571/// No-op off RTEMS: `std` already calls the platform's `pthread_setname_np`
1572/// from `Builder::spawn` on every hosted target it supports. RTEMS is not in
1573/// that list, so a name set with `Builder::name` lives only in Rust's own
1574/// `Thread` struct and never reaches the kernel — which is what makes our
1575/// threads invisible to an RTEMS task listing.
1576#[cfg(not(target_os = "rtems"))]
1577pub fn name_current_thread() {}
1578
1579/// RTEMS: `pthread_setname_np` (`cpukit/posix/src/pthreadsetnamenp.c`) into
1580/// `_Thread_Set_name`, which `strlcpy`s into `_Thread_Maximum_name_size`.
1581///
1582/// That size is `CONFIGURE_MAXIMUM_THREAD_NAME_SIZE`, default **16**
1583/// including the NUL (`rtems/score/thread.h:1079` (`rtems_6`),
1584/// `rtems/confdefs/threads.h:92-93` (`rtems_6`)), and the boot shim does not override
1585/// it — so 15 usable bytes, the same budget `std` truncates to on Linux
1586/// (`TASK_COMM_LEN`). Truncating here rather than letting the kernel do it
1587/// keeps that existing rule and keeps the call's success unambiguous:
1588/// `_Thread_Set_name` still *sets* an over-long name, it just also returns
1589/// `STATUS_RESULT_TOO_LARGE` → `ERANGE`, so an untruncated call would report
1590/// failure for a name it had in fact applied.
1591#[cfg(target_os = "rtems")]
1592pub fn name_current_thread() {
1593    let current = std::thread::current();
1594    let Some(name) = current.name() else {
1595        return;
1596    };
1597    let Ok(c_name) = std::ffi::CString::new(truncate_thread_name(name)) else {
1598        // An interior NUL cannot come from `Builder::name`, which takes a
1599        // `String`; nothing to publish if one ever did.
1600        return;
1601    };
1602    // SAFETY: `pthread_setname_np` acts on the calling thread and reads a
1603    // NUL-terminated string that outlives the call.
1604    let rc =
1605        unsafe { rtems_sched::pthread_setname_np(rtems_sched::pthread_self(), c_name.as_ptr()) };
1606    if rc != 0 {
1607        tracing::debug!(
1608            target: "epics_base_rs::runtime",
1609            thread = name,
1610            errno = rc,
1611            "pthread_setname_np failed; thread stays unnamed in the task listing"
1612        );
1613    }
1614}
1615
1616/// `CONFIGURE_MAXIMUM_THREAD_NAME_SIZE` (default 16) minus the NUL.
1617#[cfg(any(target_os = "rtems", test))]
1618const RTEMS_MAX_THREAD_NAME_BYTES: usize = 15;
1619
1620/// Cut a thread name to what an RTEMS thread object can hold, on a UTF-8
1621/// boundary.
1622///
1623/// Byte budget, not character count — `_Thread_Set_name` `strlcpy`s bytes —
1624/// but never mid-codepoint, or the task listing shows invalid UTF-8. Kept as
1625/// a pure function so the rule is testable on the host, where the caller
1626/// that applies it does not exist.
1627#[cfg(any(target_os = "rtems", test))]
1628fn truncate_thread_name(name: &str) -> &str {
1629    let mut end = name.len().min(RTEMS_MAX_THREAD_NAME_BYTES);
1630    while end > 0 && !name.is_char_boundary(end) {
1631        end -= 1;
1632    }
1633    &name[..end]
1634}
1635
1636/// The process's list of live EPICS threads — C's `pthreadList`
1637/// (`os/posix/osdThread.c:94`).
1638///
1639/// C keeps a list because no OS gives a process a portable thread
1640/// enumerator, and `epicsThreadShowAll` has to print one. Entries go on in
1641/// `start_routine` (`:436-440`), executed by the new thread itself, and come
1642/// off in `free_threadInfo` (`:232`), which is the thread-specific-data
1643/// destructor `epicsThreadInit` registered — so a dead thread leaves the list
1644/// at the moment it dies, not when somebody next reads it.
1645///
1646/// Both moments are the same here. [`enter_ioc_thread`] inserts, and the
1647/// `Registration` it parks in thread-local storage removes on the way out.
1648/// That makes the prologue the single owner of membership: a thread cannot be
1649/// listed without having been named and banded, and cannot stay listed once
1650/// its stack is gone. It also makes a second [`enter_ioc_thread`] on one
1651/// thread — a re-band — replace that thread's row rather than add a second
1652/// one, because storing the new `Registration` drops the old.
1653///
1654/// One row C has that this cannot: C's `_main_` (`osdThread.c:406-412`),
1655/// added by `epicsThreadInit`'s `pthread_once` for whichever thread first
1656/// called into `epicsThread`. This port has no process-init hook that runs on
1657/// the main thread, so the main thread is listed only if it runs the prologue
1658/// itself.
1659/// C `epicsThreadOSD`'s `isSuspended` flag and its `suspendEvent`
1660/// (`osdThread.c:179`, `:793-801`), as one cell instead of two.
1661///
1662/// A thread is suspended *exactly while* it is blocked in
1663/// [`suspend_self`]. The bit the `STATE` column prints, the bit
1664/// `epicsThreadResume` tests, and the condition the parked thread waits on
1665/// are one `bool` under one mutex, so a `SUSPEND` row and a resume that
1666/// refuses cannot disagree: there is no second flag to fall out of step.
1667///
1668/// One `Suspension` exists per thread, owned by that thread's
1669/// thread-local; its registry row holds a clone of the same `Arc`, which
1670/// is why reading a [`ThreadInfo`] out of a snapshot still sees the live
1671/// state — as C's listing walk reads `pthreadInfo->isSuspended` live.
1672#[derive(Debug, Default)]
1673struct Suspension {
1674    suspended: std::sync::Mutex<bool>,
1675    cv: std::sync::Condvar,
1676}
1677
1678impl Suspension {
1679    /// A poisoned cell is still readable: a panic elsewhere must not make a
1680    /// thread permanently unresumable, nor wedge the listing.
1681    fn lock(&self) -> std::sync::MutexGuard<'_, bool> {
1682        self.suspended.lock().unwrap_or_else(|e| e.into_inner())
1683    }
1684
1685    /// C `epicsThreadSuspendSelf` (`osdThread.c:785-795`): raise the flag,
1686    /// then wait. C does the two under no lock and relies on its event
1687    /// latching if a resume lands between them; holding the mutex across
1688    /// both closes that window instead of depending on the latch.
1689    fn suspend(&self) {
1690        let mut suspended = self.lock();
1691        *suspended = true;
1692        while *suspended {
1693            suspended = self.cv.wait(suspended).unwrap_or_else(|e| e.into_inner());
1694        }
1695    }
1696
1697    /// C `epicsThreadResume` (`osdThread.c:797-802`) with C's iocsh
1698    /// `if (!epicsThreadIsSuspended(tid))` arm (`libComRegister.c:445-449`)
1699    /// folded into it, so the test and the act cannot race apart.
1700    ///
1701    /// `false` means the thread was not suspended and nothing was done —
1702    /// C's event would have latched a token there, and deliberately does
1703    /// not here: a `dbc` issued while a lock set is running must not bank a
1704    /// resume that skips the next breakpoint.
1705    fn resume(&self) -> bool {
1706        let mut suspended = self.lock();
1707        if !*suspended {
1708            return false;
1709        }
1710        *suspended = false;
1711        self.cv.notify_all();
1712        true
1713    }
1714
1715    fn is_suspended(&self) -> bool {
1716        *self.lock()
1717    }
1718}
1719
1720thread_local! {
1721    /// The calling thread's cell, made on first use so a thread that never
1722    /// registered can still park itself and be seen once it does.
1723    static SUSPENSION: Arc<Suspension> = Arc::new(Suspension::default());
1724}
1725
1726fn current_suspension() -> Arc<Suspension> {
1727    SUSPENSION.with(Arc::clone)
1728}
1729
1730/// C `epicsThreadSuspendSelf` (`osdThread.c:785-795`) — park the calling
1731/// thread until something resumes it.
1732///
1733/// Only the calling thread can put itself here, which is C's rule and the
1734/// reason the state has a single writer: `epicsThreadShowAll` reports what
1735/// this call is doing, it does not learn it from a flag someone else set.
1736pub fn suspend_self() {
1737    // C `epicsThreadSuspendSelf` reaches `createImplicit` for a thread with no
1738    // row (`osdThread.c:790-793`), so parking always leaves something for
1739    // `epicsThreadShowAll` to mark `SUSPEND` and for `epicsThreadResume` to
1740    // name. Without it a thread could be suspended and unreachable.
1741    thread_registry::register_current_implicit();
1742    current_suspension().suspend();
1743}
1744
1745/// C `epicsThreadResume(id)` (`osdThread.c:797-802`) on a handle a
1746/// subsystem stored for itself — `dbBkpt.c`'s `pnode->taskid`, resumed by
1747/// `dbc`/`dbs` (`dbBkpt.c:518`, `:547`).
1748///
1749/// Matches the `EPICS ID` only, never the OS id that [`thread_by_id`] also
1750/// accepts: this is a handle the port issued to itself, not a token a
1751/// shell user typed, and the two number spaces can collide.
1752///
1753/// `false` means no such thread, or it was not suspended.
1754pub fn resume_thread(id: u64) -> bool {
1755    thread_report()
1756        .into_iter()
1757        .find(|t| t.id == id)
1758        .is_some_and(|t| t.resume())
1759}
1760
1761mod thread_registry {
1762    use super::{PriorityApplied, ThreadPriority};
1763    use std::sync::Mutex;
1764    use std::sync::atomic::{AtomicU64, Ordering};
1765
1766    /// Creation order, as C's `ellAdd` appends.
1767    static THREADS: Mutex<Vec<super::ThreadInfo>> = Mutex::new(Vec::new());
1768
1769    /// C hands out the address of the thread's `epicsThreadOSD` as the
1770    /// `EPICS ID`; a counter serves the same purpose — an opaque handle
1771    /// `epicsThreadShow` can be given back — without publishing an address.
1772    static NEXT_ID: AtomicU64 = AtomicU64::new(1);
1773
1774    /// Removes this thread's row when the thread ends. Held in thread-local
1775    /// storage because that is the only destructor Rust runs at exactly the
1776    /// point C's TSD destructor runs, and because [`super::enter_ioc_thread`]
1777    /// returns [`PriorityApplied`] to callers that discard it — a guard
1778    /// handed back there would be dropped immediately.
1779    struct Registration(u64);
1780
1781    impl Drop for Registration {
1782        fn drop(&mut self) {
1783            let id = self.0;
1784            lock().retain(|t| t.id != id);
1785        }
1786    }
1787
1788    thread_local! {
1789        static REGISTRATION: std::cell::RefCell<Option<Registration>> =
1790            const { std::cell::RefCell::new(None) };
1791    }
1792
1793    /// A poisoned list is still a readable list: a panic while formatting one
1794    /// row must not make the IOC's thread listing permanently unavailable.
1795    fn lock() -> std::sync::MutexGuard<'static, Vec<super::ThreadInfo>> {
1796        THREADS.lock().unwrap_or_else(|e| e.into_inner())
1797    }
1798
1799    /// Add the calling thread. Called only from [`super::enter_ioc_thread`].
1800    pub(super) fn register_current(priority: ThreadPriority, applied: PriorityApplied) {
1801        push(
1802            std::thread::current()
1803                .name()
1804                .unwrap_or("noname")
1805                .to_string(),
1806            priority,
1807            applied,
1808        );
1809    }
1810
1811    /// C's `once()` row (`osdThread.c:406-412`): `_main_`, EPICS priority 0,
1812    /// on the list without a scheduling change. Guarded like C's
1813    /// `pthread_once` because a process has one `_main_`.
1814    pub(super) fn register_main() {
1815        static ONCE: std::sync::Once = std::sync::Once::new();
1816        ONCE.call_once(|| {
1817            push(
1818                "_main_".to_string(),
1819                ThreadPriority::Custom(0),
1820                PriorityApplied::Disabled,
1821            );
1822        });
1823    }
1824
1825    /// The one place a row is built, so which entry point made a row cannot
1826    /// change the row's shape or how it is reaped.
1827    fn push(name: String, priority: ThreadPriority, applied: PriorityApplied) {
1828        let info = super::ThreadInfo {
1829            id: NEXT_ID.fetch_add(1, Ordering::Relaxed),
1830            name,
1831            os_id: os_thread_id(),
1832            priority,
1833            applied,
1834            // The same cell the thread's own `suspend_self` blocks on, so
1835            // the row cannot report a state the thread is not in.
1836            suspension: super::current_suspension(),
1837        };
1838        let id = info.id;
1839        lock().push(info);
1840        // Drop the previous guard outside the `RefCell` borrow: its `Drop`
1841        // takes the list lock, and a re-band would otherwise re-enter the
1842        // borrow through a path that is easy to add later and hard to see.
1843        let previous = REGISTRATION.with(|r| r.borrow_mut().replace(Registration(id)));
1844        drop(previous);
1845    }
1846
1847    /// The calling thread's row id, from the guard the prologue parked in
1848    /// thread-local storage — no list walk, and no dependence on the row
1849    /// still being there.
1850    pub(super) fn current_id() -> Option<u64> {
1851        REGISTRATION.with(|r| r.borrow().as_ref().map(|reg| reg.0))
1852    }
1853
1854    /// C `createImplicit` (`osdThread.c:697-735`): the row a thread that never
1855    /// went through `epicsThreadCreate` gets the first time libCom needs one
1856    /// for it. C makes one rather than answering NULL in both
1857    /// `epicsThreadGetIdSelf` (`:942`) and `epicsThreadSuspendSelf` (`:792`),
1858    /// which is why a thread can park itself and still be listed and resumed.
1859    ///
1860    /// C names it `non-EPICS_%ld` of the `pthread_t` and gives it EPICS
1861    /// priority 0 without touching its scheduling. The number here is the
1862    /// row's own OS id — the one its `LWP ID`/`PTHREAD ID` column prints — so
1863    /// the name and the column agree, which on Linux C's do not.
1864    pub(super) fn register_current_implicit() {
1865        if REGISTRATION.with(|r| r.borrow().is_some()) {
1866            return;
1867        }
1868        push(
1869            format!("non-EPICS_{}", os_thread_id()),
1870            ThreadPriority::Custom(0),
1871            PriorityApplied::Disabled,
1872        );
1873    }
1874
1875    /// A snapshot of the list, in creation order.
1876    ///
1877    /// A copy rather than a borrow because C walks the live list under
1878    /// `listLock` while calling `epicsThreadShowInfo`, which does I/O; taking
1879    /// the rows out first keeps the lock off the write path of whatever
1880    /// stream the shell is printing to.
1881    pub(super) fn snapshot() -> Vec<super::ThreadInfo> {
1882        lock().clone()
1883    }
1884
1885    /// Linux prints `LWP ID` because that is the number `top`, `ps -L` and
1886    /// `/proc` show; every other target prints the `pthread_t`, which is all
1887    /// POSIX defines. C makes the same split, in two `osdThreadExtra.c`
1888    /// files.
1889    #[cfg(target_os = "linux")]
1890    fn os_thread_id() -> u64 {
1891        // SAFETY: `gettid` takes no arguments and cannot fail.
1892        unsafe { libc::syscall(libc::SYS_gettid) as u64 }
1893    }
1894
1895    #[cfg(target_os = "rtems")]
1896    fn os_thread_id() -> u64 {
1897        // SAFETY: `pthread_self` takes no arguments and cannot fail.
1898        unsafe { super::rtems_sched::pthread_self() as u64 }
1899    }
1900
1901    #[cfg(all(unix, not(any(target_os = "linux", target_os = "rtems"))))]
1902    fn os_thread_id() -> u64 {
1903        // SAFETY: `pthread_self` takes no arguments and cannot fail.
1904        unsafe { libc::pthread_self() as u64 }
1905    }
1906
1907    /// Windows prints a `WIN32-ID` column, which C fills from
1908    /// `GetCurrentThreadId()` (`os/WIN32/osdThread.c:564`, printed at
1909    /// `:1043`). The zero this used to return is not a thread id there and
1910    /// was the same one for every row, so `thread_by_id` matched the token
1911    /// `0` against whichever thread the snapshot listed first — enough for
1912    /// `epicsThreadShow 0` to print a stranger's row and for
1913    /// `epicsThreadResume 0` to report on it instead of rejecting the id.
1914    #[cfg(windows)]
1915    fn os_thread_id() -> u64 {
1916        // SAFETY: `GetCurrentThreadId` takes no arguments and cannot fail.
1917        unsafe { GetCurrentThreadId() }.into()
1918    }
1919
1920    #[cfg(windows)]
1921    unsafe extern "system" {
1922        fn GetCurrentThreadId() -> u32;
1923    }
1924}
1925
1926/// One row of [`thread_report`] — what C's `epicsThreadShowInfo` prints about
1927/// one thread (`os/Linux/osdThreadExtra.c:31-55`).
1928#[derive(Clone, Debug)]
1929pub struct ThreadInfo {
1930    id: u64,
1931    name: String,
1932    os_id: u64,
1933    priority: ThreadPriority,
1934    applied: PriorityApplied,
1935    suspension: Arc<Suspension>,
1936}
1937
1938impl ThreadInfo {
1939    /// The `EPICS ID` column, and the handle `epicsThreadShow <id>` accepts.
1940    pub fn id(&self) -> u64 {
1941        self.id
1942    }
1943
1944    /// The `NAME` column: `std::thread::current().name()`, or `noname` for a
1945    /// thread spawned without one — C's own placeholder for the same case
1946    /// (`osdThread.c:601`).
1947    pub fn name(&self) -> &str {
1948        &self.name
1949    }
1950
1951    /// The `LWP ID` column on Linux, `PTHREAD ID` elsewhere.
1952    pub fn os_id(&self) -> u64 {
1953        self.os_id
1954    }
1955
1956    /// The `OSIPRI` column: the EPICS priority, 0..=99.
1957    pub fn epics_priority(&self) -> u8 {
1958        self.priority.value()
1959    }
1960
1961    /// The `OSSPRI` column: the OS scheduling priority the thread runs at.
1962    ///
1963    /// C reads this live with `pthread_getschedparam` (`osdThreadExtra.c:39`).
1964    /// This recomputes it from the outcome the prologue recorded, through the
1965    /// same map that produced it, which is the same number: nothing in this
1966    /// workspace re-bands a thread after the prologue
1967    /// (`only_the_prologue_reaches_the_banding_call`), and the map's other
1968    /// input — the probed range — is fixed for the life of the process.
1969    ///
1970    /// Any outcome but [`PriorityApplied::Realtime`] left the thread on the
1971    /// default policy, whose `sched_priority` is 0 by POSIX. That is what C's
1972    /// live read returns for those threads too, and what a non-privileged
1973    /// Linux IOC shows in this column for every row.
1974    pub fn os_priority(&self) -> i32 {
1975        if self.applied != PriorityApplied::Realtime {
1976            return 0;
1977        }
1978        self.mapped_os_priority()
1979    }
1980
1981    #[cfg(target_os = "linux")]
1982    fn mapped_os_priority(&self) -> i32 {
1983        match permitted_fifo_range() {
1984            RtRange::Available { min, max } => map_epics_priority(self.priority.value(), min, max),
1985            // Not reachable while the recorded outcome is `Realtime`: the
1986            // other two range states return before the map.
1987            RtRange::Unsupported | RtRange::Denied => 0,
1988        }
1989    }
1990
1991    #[cfg(target_os = "rtems")]
1992    fn mapped_os_priority(&self) -> i32 {
1993        map_epics_priority_rtems(self.priority.value())
1994    }
1995
1996    #[cfg(target_os = "vxworks")]
1997    fn mapped_os_priority(&self) -> i32 {
1998        map_epics_priority_vxworks(self.priority.value())
1999    }
2000
2001    #[cfg(not(any(target_os = "linux", epics_embedded_target)))]
2002    fn mapped_os_priority(&self) -> i32 {
2003        // No band is applied on these targets, so `Realtime` never occurs.
2004        0
2005    }
2006
2007    /// C `epicsThreadIsSuspended` (`osdThread.c:910-914`) — the `STATE`
2008    /// column's source, and what C's iocsh `epicsThreadResume` tests before
2009    /// it acts (`libComRegister.c:445`).
2010    pub fn is_suspended(&self) -> bool {
2011        self.suspension.is_suspended()
2012    }
2013
2014    /// C `epicsThreadResume` (`osdThread.c:797-802`) on this thread.
2015    ///
2016    /// `false` means it was not suspended, which is the arm C's iocsh
2017    /// wrapper reports as `Thread %s is not suspended`
2018    /// (`libComRegister.c:445-449`).
2019    pub fn resume(&self) -> bool {
2020        self.suspension.resume()
2021    }
2022
2023    /// The row `epicsThreadShowInfo` prints, without its newline.
2024    ///
2025    /// `%16.16s %14p %8lu    %3d%8d %8.8s%s` on Linux, `%12lu` for the OS id
2026    /// elsewhere. The trailing `%s` is C's ` ZOMBIE` marker, for a thread
2027    /// whose function has returned but whose `epicsThreadOSD` is still
2028    /// referenced; it is always empty here because the row and the thread end
2029    /// together.
2030    ///
2031    /// The `%8.8s` state column is C's `isSuspended ? "SUSPEND" : "OK"`
2032    /// (`os/Linux/osdThreadExtra.c:48-52`), read live from the cell the
2033    /// thread's own [`suspend_self`] blocks on.
2034    pub fn show_line(&self) -> String {
2035        format!(
2036            "{:>16.16} {:>14} {:>os_id_width$}    {:3}{:8} {:>8.8}",
2037            self.name,
2038            format!("{:#x}", self.id),
2039            self.os_id,
2040            self.epics_priority(),
2041            self.os_priority(),
2042            if self.is_suspended() { "SUSPEND" } else { "OK" },
2043            os_id_width = OS_ID_WIDTH,
2044        )
2045    }
2046}
2047
2048/// The header `epicsThreadShowInfo(0, level)` prints
2049/// (`os/Linux/osdThreadExtra.c:34-35`).
2050#[cfg(target_os = "linux")]
2051pub const THREAD_SHOW_HEADER: &str =
2052    "            NAME       EPICS ID   LWP ID   OSIPRI  OSSPRI  STATE";
2053
2054/// `os/posix/osdThreadExtra.c:27-28` — the generic POSIX header, which names
2055/// and widens the OS id column differently.
2056#[cfg(not(target_os = "linux"))]
2057pub const THREAD_SHOW_HEADER: &str =
2058    "            NAME       EPICS ID   PTHREAD ID   OSIPRI  OSSPRI  STATE";
2059
2060/// Field width of the OS id column, matching the header above.
2061#[cfg(target_os = "linux")]
2062const OS_ID_WIDTH: usize = 8;
2063
2064#[cfg(not(target_os = "linux"))]
2065const OS_ID_WIDTH: usize = 12;
2066
2067/// Every live EPICS thread, in creation order — C's `epicsThreadMap` /
2068/// `ellFirst(&pthreadList)` walk (`osdThread.c:1000-1004`).
2069pub fn thread_report() -> Vec<ThreadInfo> {
2070    thread_registry::snapshot()
2071}
2072
2073/// The calling thread's `EPICS ID` — C `epicsThreadGetIdSelf`
2074/// (`osdThread.c:936-945`).
2075///
2076/// A thread that never ran the prologue is given a row here, as C's own
2077/// `createImplicit` does at `:942`, so this cannot answer "no id": a thread
2078/// that can be asked for its handle is a thread the listing can show and
2079/// `epicsThreadResume` can name.
2080///
2081/// This is the id a subsystem stores when C stores an `epicsThreadId` of its
2082/// own — `dbBkpt.c`'s `pnode->taskid`, printed by `dbstat` as `T:` — so the
2083/// handle in that column and the `EPICS ID` column of `epicsThreadShowAll` are
2084/// one value rather than two namings of one thread.
2085pub fn current_thread_id() -> u64 {
2086    thread_registry::register_current_implicit();
2087    thread_registry::current_id().expect("register_current_implicit installed a row")
2088}
2089
2090/// The thread whose `EPICS ID` or OS id is `id` — C's match in
2091/// `epicsThreadShow` (`osdThread.c:1051-1053`).
2092///
2093/// C compares against the `epicsThreadOSD *` and the `pthread_t`. The second
2094/// is deliberately the *printed* OS id here instead: on Linux C prints
2095/// `lwpId` and matches `tid`, so the number on screen is not one C accepts —
2096/// a shell user can only ever act on what the listing showed them.
2097pub fn thread_by_id(id: u64) -> Option<ThreadInfo> {
2098    thread_report()
2099        .into_iter()
2100        .find(|t| t.id == id || t.os_id == id)
2101}
2102
2103/// The first live thread named `name` — C's `epicsThreadGetId`
2104/// (`osdThread.c:1039-1060`), which `epicsThreadShow`'s iocsh wrapper falls
2105/// back to when the argument does not parse as a number.
2106pub fn thread_by_name(name: &str) -> Option<ThreadInfo> {
2107    thread_report().into_iter().find(|t| t.name == name)
2108}
2109
2110/// C's `epicsThreadShowAll` trailer, printed to stderr
2111/// (`osdThread.c:1027-1031`).
2112///
2113/// The range is the one this process may actually enter, not the policy's
2114/// nominal range — the distinction `find_pri_range` exists for. With the
2115/// real-time switch off there is no range to report and no probe may be run
2116/// to find one (that probe is itself a scheduler call, and the switch's
2117/// guarantee is that none happens); `0 0` is then the literal truth, the
2118/// SCHED_OTHER priority every thread in the process holds.
2119///
2120/// Memory is never locked: nothing in this workspace calls `mlockall`, so
2121/// C's `epicsThreadRealtimeLock` has no counterpart to report.
2122pub fn osd_priority_range_line() -> String {
2123    let (min, max) = osd_priority_range();
2124    format!("OSD priority range min: {min} max {max}, memory not locked")
2125}
2126
2127#[cfg(target_os = "linux")]
2128fn osd_priority_range() -> (i32, i32) {
2129    match RtPolicy::current() {
2130        RtPolicy::Disabled => (0, 0),
2131        RtPolicy::AllowRealtime => match permitted_fifo_range() {
2132            RtRange::Available { min, max } => (min, max),
2133            // C's `find_pri_range` reports the failed probe the same way:
2134            // `min_pri = max_pri = min` when the policy is refused, `-1` when
2135            // the kernel has no range at all (`osdThread.c:275-289`).
2136            RtRange::Denied => {
2137                // SAFETY: `sched_get_priority_min` takes only a policy
2138                // constant and has no preconditions.
2139                let min = unsafe { libc::sched_get_priority_min(libc::SCHED_FIFO) };
2140                (min, min)
2141            }
2142            RtRange::Unsupported => (-1, -1),
2143        },
2144    }
2145}
2146
2147#[cfg(target_os = "rtems")]
2148fn osd_priority_range() -> (i32, i32) {
2149    match RtPolicy::current() {
2150        RtPolicy::Disabled => (0, 0),
2151        // No probe on this target, by construction: the map's image is fixed
2152        // and inside the settable range. Report that image.
2153        RtPolicy::AllowRealtime => (map_epics_priority_rtems(0), map_epics_priority_rtems(99)),
2154    }
2155}
2156
2157#[cfg(target_os = "vxworks")]
2158fn osd_priority_range() -> (i32, i32) {
2159    match RtPolicy::current() {
2160        RtPolicy::Disabled => (0, 0),
2161        RtPolicy::AllowRealtime => (
2162            map_epics_priority_vxworks(0),
2163            map_epics_priority_vxworks(99),
2164        ),
2165    }
2166}
2167
2168#[cfg(not(any(target_os = "linux", epics_embedded_target)))]
2169fn osd_priority_range() -> (i32, i32) {
2170    // No OS-scheduler priority API is wired here, so there is no band and no
2171    // range — the same `Unsupported` this target reports from `apply`.
2172    (0, 0)
2173}
2174
2175/// The SCHED_FIFO priority range this process may actually enter.
2176///
2177/// C parity: `find_pri_range` (`osdThread.c:259-314`). The kernel's
2178/// `sched_get_priority_max` reports the *policy's* range and ignores
2179/// `RLIMIT_RTPRIO`, so on an RT box with a restricted limit the nominal
2180/// range is wider than the usable one; C binary-searches for the real
2181/// ceiling and so do we.
2182#[cfg(target_os = "linux")]
2183#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2184enum RtRange {
2185    /// The kernel does not report a SCHED_FIFO range at all.
2186    Unsupported,
2187    /// The range exists but this process may not enter it — no
2188    /// `CAP_SYS_NICE` and `RLIMIT_RTPRIO` is 0. C's equivalent is
2189    /// `usePolicy == 0` (`osdThread.c:279-285`, `:334`), which makes it
2190    /// stop asking for SCHED_FIFO for the life of the process.
2191    Denied,
2192    /// Priorities `min..=max` are settable.
2193    Available { min: i32, max: i32 },
2194}
2195
2196/// Probe once per process and cache. Only reachable on the
2197/// [`RtPolicy::AllowRealtime`] path, so a default (switch-off) process
2198/// never runs the probe and never makes a scheduler call.
2199#[cfg(target_os = "linux")]
2200fn permitted_fifo_range() -> RtRange {
2201    static RANGE: std::sync::OnceLock<RtRange> = std::sync::OnceLock::new();
2202    *RANGE.get_or_init(probe_fifo_range)
2203}
2204
2205#[cfg(target_os = "linux")]
2206fn probe_fifo_range() -> RtRange {
2207    // SAFETY: sched_get_priority_min/max take only an int policy and have
2208    // no preconditions.
2209    let (min, max) = unsafe {
2210        (
2211            libc::sched_get_priority_min(libc::SCHED_FIFO),
2212            libc::sched_get_priority_max(libc::SCHED_FIFO),
2213        )
2214    };
2215    if min < 0 || max < 0 || max < min {
2216        return RtRange::Unsupported;
2217    }
2218
2219    // The probe *changes the scheduling of the thread that runs it*, so it
2220    // runs on a throwaway thread — exactly why C hands `find_pri_range` to
2221    // its own `pthread_create`/`pthread_join` pair (`osdThread.c:316-334`).
2222    let charge = crate::runtime::worker_pool::ThreadCharge::fixed(StackSizeClass::Small);
2223    let probe = std::thread::Builder::new()
2224        .name("cbRtProbe".to_string())
2225        // Two `sched_get_priority_*` calls and a `sched_setscheduler`; nothing
2226        // recurses. Linux-only, so this is not the RTEMS ceiling — but there is
2227        // no reason for a throwaway probe to reserve 2 MiB on any target.
2228        .stack_size(StackSizeClass::Small.bytes())
2229        .spawn(move || {
2230            let _charge = charge;
2231            // `osdThread.c:277-287`: failing at the minimum means no
2232            // permission for SCHED_FIFO at all.
2233            if set_fifo_priority(min) != 0 {
2234                return RtRange::Denied;
2235            }
2236            // `osdThread.c:296-307`: binary-search the real ceiling.
2237            let (mut low, mut high) = (min, max);
2238            while low < high {
2239                let mid = (high + low) / 2;
2240                if set_fifo_priority(mid) != 0 {
2241                    high = mid;
2242                } else {
2243                    low = mid + 1;
2244                }
2245            }
2246            // `osdThread.c:310`: `max_pri = try_pri(max) ? max-1 : max`.
2247            let top = if set_fifo_priority(high) != 0 {
2248                high - 1
2249            } else {
2250                high
2251            };
2252            RtRange::Available { min, max: top }
2253        });
2254    match probe.map(std::thread::JoinHandle::join) {
2255        Ok(Ok(range)) => range,
2256        // Cannot spawn, or the probe died: treat as no RT rather than
2257        // guessing a range we have not shown to be settable.
2258        _ => RtRange::Denied,
2259    }
2260}
2261
2262/// Map an EPICS priority `0..=99` onto the permitted SCHED_FIFO range.
2263///
2264/// C parity: `epicsThreadGetPosixPriority` (`osdThread.c:129-144`) — the
2265/// POSIX counterpart of the `epicsThreadGetOssPriorityValue` used on
2266/// RTEMS/vxWorks (`RTEMS-score/osdThread.c:94`, `vxWorks/osdThread.c:99`).
2267///
2268/// **Hosted only.** RTEMS deliberately does not use this map — see
2269/// `map_epics_priority_rtems` for the shape and the reason. The `test`
2270/// arm of the cfg exists so the two maps can be compared in one process
2271/// on the host; without it the divergence test would silently vanish.
2272#[cfg(any(target_os = "linux", test))]
2273fn map_epics_priority(epics_priority: u8, min: i32, max: i32) -> i32 {
2274    // `osdThread.c:133-134`: a degenerate range collapses to one level.
2275    if max == min {
2276        return max;
2277    }
2278    let slope = (max - min) as f64 / 100.0;
2279    let oss = epics_priority as f64 * slope + min as f64;
2280    // `ThreadPriority::value` caps at 99 and the slope is over 100, so this
2281    // cannot exceed `max`; the clamp guards the probed bounds, which are
2282    // runtime values rather than compile-time constants.
2283    (oss as i32).clamp(min, max)
2284}
2285
2286/// The highest RTEMS *core* priority number, i.e. the least urgent level.
2287///
2288/// Measured on the bring-up guest (RTEMS 6 + libbsd, QEMU
2289/// `xilinx_zynq_a9`): `RTEMS_MAXIMUM_PRIORITY == 255`, the idle thread runs
2290/// at core 255, and `sched_get_priority_min/max(SCHED_FIFO)` report `1`/`254`.
2291/// The POSIX-to-core inversion `core = 255 - posix` was verified in both
2292/// directions on that guest.
2293#[cfg(any(target_os = "rtems", test))]
2294const RTEMS_MAXIMUM_PRIORITY: i32 = 255;
2295
2296/// The RTEMS *core* priority an EPICS priority must land on.
2297///
2298/// Verbatim `epicsThreadGetOssPriorityValue` from EPICS's own RTEMS port,
2299/// `libcom/src/osi/os/RTEMS-score/osdThread.c:94-102`:
2300///
2301/// ```c
2302/// int epicsThreadGetOssPriorityValue(unsigned int osiPriority)
2303/// {
2304///     if (osiPriority > 99) { return 100; }
2305///     else { return (199 - (signed int)osiPriority); }
2306/// }
2307/// ```
2308///
2309/// Fixed offsets, not a range-scaled slope. The whole EPICS space therefore
2310/// occupies core `100..=199` and nothing else can be reached — which is the
2311/// property [`map_epics_priority_rtems`] is chosen for.
2312#[cfg(any(target_os = "rtems", test))]
2313const fn rtems_core_priority(epics_priority: u8) -> i32 {
2314    if epics_priority > 99 {
2315        100
2316    } else {
2317        199 - epics_priority as i32
2318    }
2319}
2320
2321/// Map an EPICS priority onto an RTEMS **POSIX** SCHED_FIFO priority.
2322///
2323/// A distinct function from `map_epics_priority` on purpose: the two have
2324/// different *shapes*, not different endpoints. Expressing this one as the
2325/// hosted linear map with `min`/`max` retuned would re-introduce the linear
2326/// map the moment somebody adjusted a constant, and the linear map is the
2327/// thing this arm exists to avoid.
2328///
2329/// **Deliberate deviation from base-on-RTEMS-6.** EPICS base compiles
2330/// `os/posix/osdThread.c` on RTEMS 6 — `configure/toolchain.c:31-36` sets
2331/// `OS_API = posix` for `__RTEMS_MAJOR__ >= 5`, and `os/RTEMS-posix/` ships
2332/// no `osdThread.c` — so upstream applies the *linear* map
2333/// `oss = epics*(max-min)/100 + min` over `find_pri_range`'s result, which
2334/// on this guest is `min=1`/`max=254`, with
2335/// `EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING` defaulting to `YES`
2336/// (`configure/CONFIG_ENV:57`). That places EPICS 91 (the CA server band) at
2337/// posix 231, i.e. **core 24** — far above libbsd's network threads. The
2338/// crossover is EPICS **63** (posix 160, core 95): every EPICS priority at or
2339/// above it outranks the interrupt server. Reproducing that would reproduce
2340/// the hazard, so this port takes EPICS's *own* RTEMS answer instead —
2341/// [`rtems_core_priority`] — and inverts it into the POSIX space we actually
2342/// set:
2343///
2344/// ```text
2345/// core = RTEMS_MAXIMUM_PRIORITY - posix   (measured)
2346/// core = 199 - epics                      (RTEMS-score/osdThread.c:94-102)
2347/// ⟹ posix = 255 - (199 - epics) = 56 + epics
2348/// ```
2349///
2350/// So EPICS 0 → posix 56 → core 199, EPICS 99 → posix 155 → core 100, and
2351/// anything above 99 clamps to posix 155. Every value is inside the guest's
2352/// settable `[1, 254]`. **Measured on target**, core 100 is also where
2353/// libbsd's own twelve default-band worker threads sit, so the map's most
2354/// urgent reachable value *ties* libbsd's default band there rather than
2355/// staying strictly below it — a boundary tie by construction, not a
2356/// collision-free image. It is still strictly below `IRQS`(96)/`TIME`(98);
2357/// see `rtems_priority_map_stays_below_the_libbsd_network_band`, which
2358/// asserts the non-strict `core >= 100` this tie actually produces.
2359#[cfg(any(target_os = "rtems", test))]
2360pub(crate) fn map_epics_priority_rtems(epics_priority: u8) -> i32 {
2361    RTEMS_MAXIMUM_PRIORITY - rtems_core_priority(epics_priority)
2362}
2363
2364/// Map an EPICS priority onto a VxWorks **POSIX** SCHED_FIFO priority.
2365///
2366/// **Measurement-backed**, not derived: on the bring-up box (VxWorks 7,
2367/// `x86_64-wrs-vxworks`), setting `posix = 56 + epics` — the identical POSIX
2368/// value `map_epics_priority_rtems` computes for RTEMS — landed 11 of 11
2369/// measured threads at `PriorityApplied::Realtime`, one scheduler call each.
2370/// VxWorks's own POSIX layer then inverts that POSIX value into its native
2371/// task-priority space, and the result observed there was `vx = 199 -
2372/// epics`, exact: EPICS base's own vxWorks-port formula
2373/// (`vxWorks/osdThread.c:99`, `oss = 199 - osiPriority`) — reached by a
2374/// different route (we set the POSIX value; VxWorks inverts it, rather than
2375/// us computing the native value directly as C's own port does).
2376///
2377/// Deliberately **not** implemented by calling `rtems_core_priority` /
2378/// `map_epics_priority_rtems`: those compute an RTEMS **core** priority
2379/// through `RTEMS_MAXIMUM_PRIORITY`, an RTEMS kernel constant measured on the
2380/// RTEMS bring-up guest — machinery VxWorks has no equivalent of. The two
2381/// happen to land on the same POSIX number; this restates the `56 + epics`
2382/// arithmetic directly so this function cites no RTEMS-specific fact and a
2383/// change to the RTEMS core-priority mechanism cannot silently move the
2384/// VxWorks value with it.
2385#[cfg(any(target_os = "vxworks", test))]
2386pub(crate) fn map_epics_priority_vxworks(epics_priority: u8) -> i32 {
2387    56 + epics_priority.min(99) as i32
2388}
2389
2390/// Ask the OS for SCHED_FIFO at `oss` on the **calling** thread. The single
2391/// place this crate touches the scheduler; returns the raw `pthread_*`
2392/// status (0 on success). Counting lives here so the "switch off ⟹ no
2393/// scheduler call" guarantee is observable on every target that has one.
2394#[cfg(any(target_os = "linux", epics_embedded_target))]
2395fn set_fifo_priority(oss: i32) -> i32 {
2396    SCHED_CALLS_MADE.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2397    #[cfg(test)]
2398    SCHED_CALLS.with(|c| c.set(c.get() + 1));
2399    set_fifo_priority_raw(oss)
2400}
2401
2402#[cfg(target_os = "linux")]
2403fn set_fifo_priority_raw(oss: i32) -> i32 {
2404    let param = libc::sched_param {
2405        sched_priority: oss,
2406    };
2407    // SAFETY: pthread_setschedparam operates on the calling thread with a
2408    // stack-local sched_param and a valid policy constant.
2409    unsafe { libc::pthread_setschedparam(libc::pthread_self(), libc::SCHED_FIFO, &param) }
2410}
2411
2412/// The RTEMS scheduler surface, declared here rather than taken from `libc`.
2413///
2414/// `libc`'s `newlib/rtems` module (0.2.188) declares neither `sched_param`,
2415/// `SCHED_FIFO`, `pthread_setschedparam` nor `pthread_self` — its sibling
2416/// newlib targets `vita` and `horizon` declare all of them, RTEMS does not.
2417/// The functions exist in the RTEMS 6 kernel
2418/// (`cpukit/posix/src/pthreadsetschedparam.c`) and in the toolchain headers;
2419/// only the Rust binding is missing.
2420///
2421/// `libc::timespec` is deliberately NOT used to describe the sporadic-server
2422/// tail: `libc` types `time_t` as `i32` for every newlib target except
2423/// `horizon`/`espidf` (`src/unix/newlib/mod.rs:55-64`), while the arm-rtems6
2424/// toolchain has `sizeof(time_t) == 8`. Its `timespec` is therefore half the
2425/// real width on this target, so the tail is carried as opaque bytes sized
2426/// from the target compiler instead.
2427///
2428/// **RTEMS-only, deliberately not widened to VxWorks.** VxWorks is not
2429/// newlib, and `libc` *does* declare `sched_param`/`SCHED_FIFO`/
2430/// `pthread_setschedparam`/`pthread_self` for it — with a different
2431/// `sched_param` layout (48 bytes, but `sched_priority: c_int` followed by a
2432/// *typed* `sched_ss_low_priority`/two `timespec`s/`sched_ss_max_repl` tail,
2433/// not this module's opaque bytes). Reusing this RTEMS-shaped struct for
2434/// VxWorks was measured to "work" only because `SCHED_FIFO` never reads past
2435/// `sched_priority` at offset 0 — the tail's true shape never mattered for
2436/// that policy — which is exactly the kind of coincidence a struct-layout
2437/// mismatch should not be allowed to depend on. VxWorks's `set_fifo_priority_raw`
2438/// arm below therefore uses `libc::sched_param` directly.
2439#[cfg(target_os = "rtems")]
2440mod rtems_sched {
2441    use std::ffi::c_int;
2442
2443    /// `sys/sched.h`: `#define SCHED_FIFO 1`.
2444    pub const SCHED_FIFO: c_int = 1;
2445
2446    /// `struct sched_param` as arm-rtems6 lays it out.
2447    ///
2448    /// `sys/features.h:404-405` defines both `_POSIX_SPORADIC_SERVER` and
2449    /// `_POSIX_THREAD_SPORADIC_SERVER`, so `sys/sched.h` compiles the
2450    /// sporadic-server tail in. Measured with the target compiler
2451    /// (`arm-rtems6-gcc`, `sizeof`/`offsetof` via array-length symbols):
2452    ///
2453    /// | field | offset | size |
2454    /// |-------|--------|------|
2455    /// | `sched_priority`        |  0 |  4 |
2456    /// | `sched_ss_low_priority` |  4 |  4 |
2457    /// | `sched_ss_repl_period`  |  8 | 16 |
2458    /// | `sched_ss_init_budget`  | 24 | 16 |
2459    /// | `sched_ss_max_repl`     | 40 |  4 |
2460    ///
2461    /// total 48, align 8. `SCHED_FIFO` makes the kernel read only
2462    /// `sched_priority` (`_POSIX_Thread_Translate_sched_param` takes the
2463    /// sporadic branch for `SCHED_SPORADIC` alone), but the struct is
2464    /// declared at full width anyway so the kernel is never handed a pointer
2465    /// to less memory than its own header describes.
2466    #[repr(C, align(8))]
2467    pub struct SchedParam {
2468        pub sched_priority: c_int,
2469        /// Offsets 4..48 — the sporadic-server fields, unused under
2470        /// `SCHED_FIFO` and always zeroed.
2471        pub sporadic_tail: [u8; 44],
2472    }
2473
2474    // The whole point of the opaque tail is that the width is right. If a
2475    // future edit reaches for `libc::timespec` here, this stops the build
2476    // instead of silently handing the kernel a short buffer.
2477    const _: () = {
2478        assert!(core::mem::size_of::<SchedParam>() == 48);
2479        assert!(core::mem::align_of::<SchedParam>() == 8);
2480    };
2481
2482    unsafe extern "C" {
2483        pub fn pthread_self() -> libc::pthread_t;
2484        pub fn pthread_setschedparam(
2485            thread: libc::pthread_t,
2486            policy: c_int,
2487            param: *const SchedParam,
2488        ) -> c_int;
2489        /// `cpukit/posix/src/pthreadsetnamenp.c`. Also absent from `libc`'s
2490        /// `newlib/rtems` module.
2491        pub fn pthread_setname_np(thread: libc::pthread_t, name: *const std::ffi::c_char) -> c_int;
2492    }
2493}
2494
2495#[cfg(target_os = "rtems")]
2496fn set_fifo_priority_raw(oss: i32) -> i32 {
2497    let param = rtems_sched::SchedParam {
2498        sched_priority: oss,
2499        sporadic_tail: [0u8; 44],
2500    };
2501    // SAFETY: `pthread_setschedparam` acts on the calling thread, is handed a
2502    // stack-local `sched_param` of the target's own width (asserted above),
2503    // and a policy constant taken from `sys/sched.h`.
2504    unsafe {
2505        rtems_sched::pthread_setschedparam(
2506            rtems_sched::pthread_self(),
2507            rtems_sched::SCHED_FIFO,
2508            &param,
2509        )
2510    }
2511}
2512
2513/// VxWorks: `libc::sched_param` directly, not the RTEMS-shaped struct above.
2514///
2515/// Unlike RTEMS, `libc` declares this target's own `sched_param` — a
2516/// `sched_priority: c_int` followed by a *typed* sporadic-server tail
2517/// (`sched_ss_low_priority: c_int`, two `libc::timespec` fields,
2518/// `sched_ss_max_repl: c_int`) — so there is nothing to hand-lay: the tail is
2519/// zeroed rather than omitted because `SCHED_FIFO` never reads it, matching
2520/// the RTEMS arm's own reasoning, but the fields are the platform's real
2521/// fields at the platform's real offsets rather than opaque bytes sized by
2522/// guesswork.
2523#[cfg(target_os = "vxworks")]
2524fn set_fifo_priority_raw(oss: i32) -> i32 {
2525    let param = libc::sched_param {
2526        sched_priority: oss,
2527        sched_ss_low_priority: 0,
2528        sched_ss_repl_period: libc::timespec {
2529            tv_sec: 0,
2530            tv_nsec: 0,
2531        },
2532        sched_ss_init_budget: libc::timespec {
2533            tv_sec: 0,
2534            tv_nsec: 0,
2535        },
2536        sched_ss_max_repl: 0,
2537    };
2538    // SAFETY: pthread_setschedparam operates on the calling thread with a
2539    // stack-local sched_param of libc's own VxWorks width and a valid policy
2540    // constant.
2541    unsafe { libc::pthread_setschedparam(libc::pthread_self(), libc::SCHED_FIFO, &param) }
2542}
2543
2544/// The unprivileged-fallback message. Emitted **once per process**: the
2545/// denial is a property of the process, not of the thread that happened to
2546/// notice it first, and an IOC creates a thread per CA client.
2547#[cfg(target_os = "linux")]
2548fn warn_rt_denied_once() {
2549    static WARNED: std::sync::Once = std::sync::Once::new();
2550    WARNED.call_once(|| {
2551        #[cfg(test)]
2552        DENIED_WARNINGS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2553        tracing::warn!(
2554            target: "epics_base_rs::runtime",
2555            switch = RT_PRIORITY_ENV,
2556            "{RT_PRIORITY_ENV} asked for real-time scheduling, but this process may not \
2557             use SCHED_FIFO (needs CAP_SYS_NICE or a non-zero RLIMIT_RTPRIO). Every IOC \
2558             thread stays at the default scheduling policy; timing is not real-time. \
2559             Logged once per process."
2560        );
2561    });
2562}
2563
2564#[cfg(target_os = "linux")]
2565fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
2566    let (min, max) = match permitted_fifo_range() {
2567        RtRange::Unsupported => return PriorityApplied::Unsupported,
2568        RtRange::Denied => {
2569            warn_rt_denied_once();
2570            return PriorityApplied::BestEffortFailed;
2571        }
2572        RtRange::Available { min, max } => (min, max),
2573    };
2574    let oss = map_epics_priority(epics_priority, min, max);
2575    let rc = set_fifo_priority(oss);
2576    if rc == 0 {
2577        PriorityApplied::Realtime
2578    } else {
2579        // The probe proved this range settable, so a failure here is not
2580        // the permission case the warning covers — keep it at debug.
2581        tracing::debug!(
2582            target: "epics_base_rs::runtime",
2583            epics_priority,
2584            oss,
2585            errno = rc,
2586            "SCHED_FIFO priority not applied; thread stays at default policy"
2587        );
2588        PriorityApplied::BestEffortFailed
2589    }
2590}
2591
2592#[cfg(target_os = "rtems")]
2593fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
2594    // Without this, every IOC thread runs at one level just above idle:
2595    // `cpukit/posix/src/pthreadattrdefault.c:49-58` (both `rtems` pins) sets
2596    // `inheritsched = PTHREAD_INHERIT_SCHED` in the default attribute set and
2597    // `std` never calls `pthread_attr_setinheritsched`, so a thread inherits
2598    // its creator's parameters — and every IOC thread descends from
2599    // `POSIX_Init`, which the boot shim deliberately lowers to
2600    // `RTEMS_MAXIMUM_PRIORITY - 1`. The CA receiver/sender ordering that stops
2601    // a stalled client starving command dispatch does not hold at one level.
2602    //
2603    // No range probe, unlike Linux. The probe exists there because
2604    // `sched_get_priority_max` reports the *policy's* range while
2605    // `RLIMIT_RTPRIO`/`CAP_SYS_NICE` decide the usable one, so the settable
2606    // ceiling has to be searched for. RTEMS has no such permission gate —
2607    // `pthread_setschedparam` (`cpukit/posix/src/pthreadsetschedparam.c`)
2608    // performs no privilege check — and this map's image is a fixed
2609    // `[56, 155]`, inside the measured settable `[1, 254]` by construction.
2610    // There is nothing to discover, and a probe thread would itself need a
2611    // band to run in.
2612    let oss = map_epics_priority_rtems(epics_priority);
2613    let rc = set_fifo_priority(oss);
2614    if rc == 0 {
2615        PriorityApplied::Realtime
2616    } else {
2617        tracing::debug!(
2618            target: "epics_base_rs::runtime",
2619            epics_priority,
2620            oss,
2621            errno = rc,
2622            "SCHED_FIFO priority not applied; thread stays at default policy"
2623        );
2624        PriorityApplied::BestEffortFailed
2625    }
2626}
2627
2628/// **Measurement-backed** (VxWorks 7, `x86_64-wrs-vxworks` bring-up box): no
2629/// range probe here either, and for the same reason as RTEMS —
2630/// `pthread_setschedparam` performed no privilege check there, 11 of 11
2631/// measured threads landed `PriorityApplied::Realtime`, and
2632/// [`map_epics_priority_vxworks`]'s fixed image is inside the settable range
2633/// by construction. There is nothing to discover on this target either.
2634#[cfg(target_os = "vxworks")]
2635fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
2636    let oss = map_epics_priority_vxworks(epics_priority);
2637    let rc = set_fifo_priority(oss);
2638    if rc == 0 {
2639        PriorityApplied::Realtime
2640    } else {
2641        tracing::debug!(
2642            target: "epics_base_rs::runtime",
2643            epics_priority,
2644            oss,
2645            errno = rc,
2646            "SCHED_FIFO priority not applied; thread stays at default policy"
2647        );
2648        PriorityApplied::BestEffortFailed
2649    }
2650}
2651
2652#[cfg(not(any(target_os = "linux", epics_embedded_target)))]
2653fn apply_priority_impl(_epics_priority: u8) -> PriorityApplied {
2654    // No OS-scheduler priority API is wired on other targets. The three that
2655    // are wired each needed a *measured* target band before they could be:
2656    // Linux probes for its settable ceiling at runtime, RTEMS's map is
2657    // pinned against libbsd's network-thread band measured on the bring-up
2658    // guest, and VxWorks's map is the RTEMS one's POSIX value, confirmed by
2659    // measurement on its own bring-up box. No number here is guessable, so a
2660    // new target gets `Unsupported` until somebody measures it rather than a
2661    // plausible-looking range.
2662    PriorityApplied::Unsupported
2663}
2664
2665/// How many times this process has asked the OS scheduler for SCHED_FIFO,
2666/// across every thread — including the one-off range probe.
2667///
2668/// The observable form of the opt-in guarantee: with [`RT_PRIORITY_ENV`]
2669/// unset, a process can run its whole life and this stays `0`. Also answers
2670/// "did this IOC ever actually try to go real-time?" from a log line.
2671///
2672/// Always `0` off Linux, where no scheduler call is wired at all.
2673pub fn sched_calls_made() -> usize {
2674    SCHED_CALLS_MADE.load(std::sync::atomic::Ordering::Relaxed)
2675}
2676
2677static SCHED_CALLS_MADE: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
2678
2679#[cfg(all(test, any(target_os = "linux", epics_embedded_target)))]
2680thread_local! {
2681    /// Every scheduler call this crate makes passes through
2682    /// [`set_fifo_priority`], which bumps this. Tests assert the delta is
2683    /// zero with the switch off — the property "switch off ⟹ no sched
2684    /// calls" observed directly rather than inferred from a return value.
2685    ///
2686    /// Per-thread, not global: `pthread_setschedparam` acts on the calling
2687    /// thread, so a per-thread count is the exact quantity, and a test
2688    /// cannot be perturbed by a concurrent one (the unit tests share a
2689    /// process under plain `cargo test`).
2690    static SCHED_CALLS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
2691}
2692
2693/// How many times the once-per-process denial warning was emitted.
2694#[cfg(all(test, target_os = "linux"))]
2695static DENIED_WARNINGS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
2696
2697/// Spawn a blocking closure on a dedicated thread and apply the given
2698/// EPICS [`ThreadPriority`] to that thread before running `f`.
2699///
2700/// The priority application is best effort (see
2701/// [`apply_to_current_thread`]); `f` runs regardless of whether the OS
2702/// honoured the request. This is the priority-aware counterpart of
2703/// [`Reactor::spawn_blocking`] for IOC threads (CA server, scan) that a C IOC
2704/// would run in a distinct SCHED band.
2705#[cfg(tokio_backend)]
2706pub fn spawn_blocking_with_priority<F, R>(priority: ThreadPriority, f: F) -> TaskHandle<R>
2707where
2708    F: FnOnce() -> R + Send + 'static,
2709    R: Send + 'static,
2710{
2711    tokio::task::spawn_blocking(move || {
2712        let _ = enter_ioc_thread(priority);
2713        f()
2714    })
2715}
2716
2717/// RTEMS: run the blocking closure on a callback-pool worker.
2718///
2719/// The requested EPICS [`ThreadPriority`] is **not** yet mapped onto a callback
2720/// band here: the pool workers are long-lived and shared, so re-prioritising the
2721/// running worker per task would leak that priority into the next callback it
2722/// drains. The closure runs at the pool's default Medium band; mapping
2723/// `ThreadPriority` to a `CallbackPriority` band is deferred to RTEMS bring-up.
2724#[cfg(exec_backend)]
2725pub fn spawn_blocking_with_priority<F, R>(_priority: ThreadPriority, f: F) -> TaskHandle<R>
2726where
2727    F: FnOnce() -> R + Send + 'static,
2728    R: Send + 'static,
2729{
2730    use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_blocking_on};
2731    spawn_blocking_on(
2732        &background().callbacks().handle(),
2733        DEFAULT_SPAWN_PRIORITY,
2734        f,
2735    )
2736}
2737
2738/// Spawn a **dedicated OS thread** that runs `f` at `priority` with a `stack`
2739/// of [`StackSizeClass`], plus whatever ambient async context
2740/// [`block_on_sync`] needs on this target.
2741///
2742/// # Why the stack class is a parameter and not a default
2743///
2744/// C creates an IOC thread with `epicsThreadCreate(name, priority, stackSize,
2745/// fn, arg)` — three attributes. This seam carried the first two and let the
2746/// third fall through to whatever `std` picks, which is **2 MiB on RTEMS**:
2747/// `std/src/sys/thread/unix.rs` gates its `DEFAULT_MIN_STACK_SIZE` on
2748/// `not(any(l4re, vxworks, espidf, nuttx))`, and vxWorks got a 256 KiB
2749/// carve-out where RTEMS did not.
2750///
2751/// That is invisible on the host, where a thread stack is lazily-committed
2752/// virtual address space, and decisive on the target, where it is carved
2753/// eagerly out of a fixed pool. Making it a parameter is what stops a new
2754/// per-connection thread from silently costing 2 MiB: there is no default to
2755/// inherit, so every caller states what the thread is for.
2756///
2757/// Not [`spawn_blocking_with_priority`], and the difference is the point.
2758/// That one hands the closure to a *pool*: tokio's blocking pool on the host,
2759/// and on RTEMS a shared callback-pool worker that also drops the priority
2760/// (see its `exec_backend` arm). Both are right for work that finishes. A
2761/// server thread that lives as long as its connection would occupy a pool
2762/// worker for that whole time, so the pool is the wrong home for it — an IOC
2763/// thread that a C IOC would create with `epicsThreadCreate` wants a thread of
2764/// its own, and the priority a C IOC gives it.
2765///
2766/// # Why the ambient context is part of this, and not the caller's problem
2767///
2768/// `block_on_sync` picks its mechanism from the thread it is called on, but it
2769/// cannot *create* the context a future needs. On the host a fresh
2770/// `std::thread` has no runtime, so a future that spawns tasks or arms timers
2771/// panics with "there is no reactor running" the moment it is polled — even
2772/// though `block_on_sync` itself was perfectly happy to park. On RTEMS the
2773/// exec backend is process-global (`background_init`), so a bare thread is
2774/// already complete. That asymmetry is a property of the two backends, so it
2775/// is resolved here, at the seam, rather than by a `cfg` in every server that
2776/// wants a thread.
2777///
2778/// The captured context is whatever the *calling* thread is running under, so
2779/// call this from the runtime the work should belong to. When there is none
2780/// (RTEMS always; on the host a caller that is itself outside a runtime) the
2781/// thread simply runs without one, which is exactly right for a future whose
2782/// awaits are all runtime-agnostic.
2783///
2784/// # A current-thread ambient is not inherited, and that is the rule
2785///
2786/// [`RuntimeHandle::try_current`] answers two different questions with one
2787/// value: *"am I running on this runtime's thread"* and *"has this thread
2788/// merely entered this handle"*. [`block_on_sync`] cannot distinguish them, so
2789/// it must assume the first and refuse to park under a `CurrentThread` flavor —
2790/// correct on that runtime's own thread, where parking halts the task that
2791/// would wake you, and wrong on a dedicated thread, where it halts nothing.
2792///
2793/// So the dual meaning is removed here, at the one place a dedicated thread's
2794/// context is decided, rather than left for `block_on_sync` to guess: a
2795/// `CurrentThread` ambient is **not** inherited, and the thread runs with no
2796/// runtime — the `park_on` arm, which is sound for it and is the only arm RTEMS
2797/// ever takes.
2798///
2799/// Nothing is lost by declining it. What inheriting buys is stated above —
2800/// `spawn` and the timer inside `block_on_sync` — and under a `CurrentThread`
2801/// ambient `block_on_sync` returns
2802/// [`Err(CurrentThreadRuntime)`](NotBlockable::CurrentThreadRuntime), so every one of those
2803/// powers is unreachable anyway. Inheriting it can only convert a thread that
2804/// would have worked into one that cannot block at all. Measured as exactly
2805/// that: the PVA client's blocking byte pumps
2806/// (`runtime::blocking_io::spawn_pump`) are dedicated threads whose bodies are
2807/// pure `tokio::sync` channel traffic, and every `#[tokio::test]` that drives
2808/// one is `CurrentThread` by default — inheritance made the reader pump exit on
2809/// its first chunk and the connection read as "server closed during handshake".
2810#[cfg(tokio_backend)]
2811pub fn spawn_dedicated_thread<F>(
2812    name: String,
2813    priority: ThreadPriority,
2814    stack: StackSizeClass,
2815    f: F,
2816) -> std::io::Result<std::thread::JoinHandle<()>>
2817where
2818    F: FnOnce() + Send + 'static,
2819{
2820    let ambient = InheritedRuntime::capture();
2821    let charge = crate::runtime::worker_pool::ThreadCharge::fixed(stack);
2822    std::thread::Builder::new()
2823        .name(name)
2824        .stack_size(stack.bytes())
2825        .spawn(move || {
2826            // Dies with the thread, so the account tracks threads that exist.
2827            let _charge = charge;
2828            // Held for the whole body: it is what makes `tokio::spawn` and the
2829            // timer reachable from this thread, and therefore what lets a future
2830            // written for the hosted driver run unchanged under `block_on_sync`.
2831            ambient.run(move || {
2832                let _ = enter_ioc_thread(priority);
2833                f()
2834            })
2835        })
2836}
2837
2838/// The ambient async context a worker body should run under — captured on the
2839/// thread that *submitted* the work, applied on the thread that runs it.
2840///
2841/// **One owner for the question `spawn_dedicated_thread`'s docs above answer at
2842/// length.** Two callers need it and they differ in *when* they capture:
2843/// `spawn_dedicated_thread` captures once, at spawn, because the thread it
2844/// creates serves exactly one body; `runtime::worker_pool` captures per **job**,
2845/// because a pooled worker outlives the runtime that first used it. A pooled
2846/// worker that inherited its ambient at creation would hold a `Handle` to a
2847/// runtime that has since been dropped — every `#[tokio::test]` builds and drops
2848/// its own — and enter it for every later connection.
2849///
2850/// The `CurrentThread` filter is the rule stated above and must not be
2851/// re-derived: a current-thread ambient is *not* inherited, because
2852/// `block_on_sync` cannot distinguish "I am that runtime's thread" from "I have
2853/// merely entered its handle" and must refuse to park under it.
2854#[cfg(tokio_backend)]
2855pub(crate) struct InheritedRuntime(Option<tokio::runtime::Handle>);
2856
2857#[cfg(tokio_backend)]
2858impl InheritedRuntime {
2859    /// Capture the calling thread's runtime, if it is one a dedicated thread
2860    /// may enter.
2861    pub(crate) fn capture() -> Self {
2862        Self(
2863            tokio::runtime::Handle::try_current()
2864                .ok()
2865                .filter(|h| h.runtime_flavor() != RuntimeFlavor::CurrentThread),
2866        )
2867    }
2868
2869    /// Run `f` with the captured context entered for its whole duration.
2870    pub(crate) fn run<R>(&self, f: impl FnOnce() -> R) -> R {
2871        let _entered = self.0.as_ref().map(|h| h.enter());
2872        f()
2873    }
2874}
2875
2876/// RTEMS: the exec backend's spawn pool and timer are process-global, so there
2877/// is no per-thread context to capture or enter. Same shape so the callers need
2878/// no `cfg` of their own.
2879#[cfg(exec_backend)]
2880pub(crate) struct InheritedRuntime;
2881
2882#[cfg(exec_backend)]
2883impl InheritedRuntime {
2884    pub(crate) fn capture() -> Self {
2885        Self
2886    }
2887
2888    pub(crate) fn run<R>(&self, f: impl FnOnce() -> R) -> R {
2889        f()
2890    }
2891}
2892
2893/// RTEMS: a plain thread is already complete — the exec backend's spawn pool
2894/// and timer are process-global, so there is no per-thread context to enter.
2895#[cfg(exec_backend)]
2896pub fn spawn_dedicated_thread<F>(
2897    name: String,
2898    priority: ThreadPriority,
2899    stack: StackSizeClass,
2900    f: F,
2901) -> std::io::Result<std::thread::JoinHandle<()>>
2902where
2903    F: FnOnce() + Send + 'static,
2904{
2905    let charge = crate::runtime::worker_pool::ThreadCharge::fixed(stack);
2906    std::thread::Builder::new()
2907        .name(name)
2908        .stack_size(stack.bytes())
2909        .spawn(move || {
2910            let _charge = charge;
2911            let _ = enter_ioc_thread(priority);
2912            f()
2913        })
2914}
2915
2916/// A thread the IOC **cannot correctly run without** — the scan rates, the
2917/// callback bands, the delayed-callback timer, the boot script.
2918///
2919/// # Invariant
2920///
2921/// **An IOC that fails to start a mandatory thread MUST NOT continue serving.**
2922/// A thread-local panic is not that: on a `panic = "unwind"` target — and RTEMS
2923/// and VxWorks both default to unwind — `Builder::spawn(..).expect(..)` kills
2924/// only the thread that called it. Measured on a VxWorks 7 RTP on a 1 GB guest:
2925/// `EAGAIN` from the periodic-scan spawn panicked the `scan-owner` thread, the
2926/// stop guard it held unwound and stopped the rates that *had* started, and the
2927/// process went on answering CA with zero periodic scanning — a half-IOC whose
2928/// records simply never process.
2929///
2930/// C has no such state. `spawnPeriodic` (`dbScan.c:939-955`) calls
2931/// `epicsThreadCreateOpt` and then `epicsEventWait(startStopEvent)`; the event
2932/// is posted by `periodicTask` itself, so when the thread was never created
2933/// nobody posts it and `iocInit` wedges. C never reaches "serving".
2934///
2935/// # Why there is no `Result` on [`spawn`](Self::spawn)
2936///
2937/// Because there is nothing a caller could do with one that satisfies the
2938/// invariant. Every caller that is *not* inside a fallible boot step would have
2939/// to re-derive "this must be fatal" locally, and that is precisely the `.expect`
2940/// the type exists to remove. The one shape that *can* satisfy it without
2941/// aborting — a caller still inside a boot step that returns its error to the
2942/// owner that decides whether to serve — is [`try_spawn`](Self::try_spawn), and
2943/// that obligation is stated on it.
2944///
2945/// Name, band and stack class are constructor parameters for the same reason
2946/// they are on [`spawn_dedicated_thread`]: a caller cannot omit what it must
2947/// pass, so the RTEMS thread census (2 MiB default stacks, OS-anonymous
2948/// threads) is closed by signature rather than by a source sweep.
2949pub struct MandatoryThread {
2950    name: String,
2951    priority: ThreadPriority,
2952    stack: StackSizeClass,
2953}
2954
2955impl MandatoryThread {
2956    /// Declare a mandatory thread: its C thread name, the EPICS band it holds,
2957    /// and the stack class the C IOC gives it.
2958    pub fn new(name: impl Into<String>, priority: ThreadPriority, stack: StackSizeClass) -> Self {
2959        Self {
2960            name: name.into(),
2961            priority,
2962            stack,
2963        }
2964    }
2965
2966    /// Start it, or take the process down.
2967    ///
2968    /// For every caller with no error path back to whoever decides that this
2969    /// IOC serves — a constructor returning `Self`, a `OnceLock` initialiser, a
2970    /// future that parks forever. See the type docs for why this returns no
2971    /// `Result`.
2972    pub fn spawn<F>(self, f: F) -> std::thread::JoinHandle<()>
2973    where
2974        F: FnOnce() + Send + 'static,
2975    {
2976        let name = self.name.clone();
2977        match self.try_spawn(f) {
2978            Ok(handle) => handle,
2979            Err(e) => mandatory_thread_unavailable(&name, &e),
2980        }
2981    }
2982
2983    /// Start it, handing the failure to a caller that is **still inside a
2984    /// fallible boot step**.
2985    ///
2986    /// The obligation this carries: the returned error MUST reach the owner
2987    /// that decides whether the IOC serves, and that owner MUST refuse. It must
2988    /// not be unwrapped, logged-and-ignored, or turned into a warning — any of
2989    /// those re-opens exactly the half-IOC the type docs describe. Use
2990    /// [`spawn`](Self::spawn) when no such path exists.
2991    pub fn try_spawn<F>(self, f: F) -> std::io::Result<std::thread::JoinHandle<()>>
2992    where
2993        F: FnOnce() + Send + 'static,
2994    {
2995        let priority = self.priority;
2996        let charge = crate::runtime::worker_pool::ThreadCharge::fixed(self.stack);
2997        std::thread::Builder::new()
2998            .name(self.name)
2999            .stack_size(self.stack.bytes())
3000            .spawn(move || {
3001                let _charge = charge;
3002                let _ = enter_ioc_thread(priority);
3003                f()
3004            })
3005    }
3006}
3007
3008/// What the operator reads on the console when a mandatory thread could not be
3009/// created. Split out from [`mandatory_thread_unavailable`] so the wording is
3010/// testable without a process that aborts.
3011fn mandatory_thread_failure_message(name: &str, err: &std::io::Error) -> String {
3012    format!(
3013        "FATAL: the IOC could not create its mandatory `{name}` thread: {err}. \
3014         Continuing would leave this IOC answering clients while the work that \
3015         thread owns never runs, so the process is aborting instead \
3016         (C dbScan.c:939-955 wedges iocInit for the same reason)."
3017    )
3018}
3019
3020/// The single fatal exit for a mandatory thread that could not be created.
3021///
3022/// `eprintln!` and not `tracing`/`errlog`: on the RTEMS and VxWorks targets no
3023/// subscriber is installed, so a `tracing` event at this point is discarded and
3024/// the operator sees an IOC that simply went quiet. Only `eprintln!` and panic
3025/// output reach the console there.
3026///
3027/// `abort` and not `exit`: unwinding would run every other thread's destructors
3028/// against a half-built IOC, and the boot state that made the spawn fail is not
3029/// one to tear down tidily.
3030fn mandatory_thread_unavailable(name: &str, err: &std::io::Error) -> ! {
3031    eprintln!("{}", mandatory_thread_failure_message(name, err));
3032    std::process::abort()
3033}
3034
3035#[cfg(test)]
3036mod tests {
3037    use super::*;
3038
3039    /// The band binding, end to end: a reactor with an executor bound puts its
3040    /// spawned future on that executor's worker, and one without still uses the
3041    /// process-global callback pool.
3042    ///
3043    /// This is the property every PVA connection future depends on — the seam
3044    /// is one `with_executor` at the top of `serve`, and everything downstream
3045    /// inherits it through its clone of the reactor.
3046    #[cfg(exec_backend)]
3047    #[test]
3048    fn a_bound_executor_takes_the_spawned_future_off_the_callback_pool() {
3049        use crate::runtime::background::DedicatedExecutor;
3050        use std::sync::mpsc;
3051        use std::time::Duration;
3052
3053        fn worker_name_of(reactor: &Reactor) -> String {
3054            let (tx, rx) = mpsc::channel();
3055            reactor.spawn(async move {
3056                let name = std::thread::current()
3057                    .name()
3058                    .unwrap_or_default()
3059                    .to_string();
3060                tx.send(name).expect("send");
3061            });
3062            rx.recv_timeout(Duration::from_secs(5))
3063                .expect("the future ran")
3064        }
3065
3066        let plain = Reactor::default();
3067        assert!(
3068            worker_name_of(&plain).starts_with("cbMedium"),
3069            "an unbound reactor still lands on C's middle callback band"
3070        );
3071
3072        let exec = Arc::new(
3073            DedicatedExecutor::new("BOUNDEXEC", ThreadPriority::Custom(18), 1)
3074                .expect("executor starts"),
3075        );
3076        let bound = plain.with_executor(&exec);
3077        assert_eq!(worker_name_of(&bound), "BOUNDEXEC");
3078        // The binding is per-reactor, not global: the one it was derived from
3079        // is unchanged.
3080        assert!(worker_name_of(&plain).starts_with("cbMedium"));
3081    }
3082
3083    /// The workspace's one production-slice rule, under this file's old name.
3084    ///
3085    /// `Keep` because the guards below strip comments themselves. What they
3086    /// gain from the shared rule is the two properties the truncating one did
3087    /// not hold: the seven `#[cfg(any(target_os = "rtems", test))]` items here
3088    /// ship on RTEMS and stay in the slice, and the line numbers two of these
3089    /// guards report still name lines of the file a reader will open.
3090    fn production_scope(src: &'static str) -> &'static str {
3091        source_guard::production(src, source_guard::Comments::Keep)
3092    }
3093
3094    /// Every source in this crate, labelled the way the messages below name
3095    /// files.
3096    ///
3097    /// Read from the crate directory rather than listed. The list was written
3098    /// out three times here and two copies had drifted: both
3099    /// `every_thread_in_this_crate_publishes_its_name` and
3100    /// `only_the_prologue_reaches_the_banding_call` named four files and
3101    /// omitted `runtime/worker_pool.rs`, which spawns a named, banded,
3102    /// stack-classed thread — so neither guard had ever looked at it, while
3103    /// the stack-size guard had. A duplicated covered set is not a covered
3104    /// set; it is three answers to one question.
3105    ///
3106    /// Rooted at `src`, not at `src/runtime`, because the claim these guards
3107    /// make is about this crate. `epics-base-rs`'s two thread-creating files
3108    /// (`server/ioc_app.rs`, `server/scan.rs`) are swept by the same
3109    /// assertions in that crate's own `tests/thread_census.rs`:
3110    /// `include_str!` must not cross a crate boundary — a path outside the
3111    /// package directory does not survive `cargo publish` — so the guard is
3112    /// split by subject, not weakened.
3113    fn crate_sources() -> Vec<(String, &'static str)> {
3114        let files = source_guard::sweep(source_guard::module_dir!("src"), &[]);
3115        assert!(
3116            files.iter().any(|(label, _)| *label == "runtime/task.rs"),
3117            "the crate sweep did not find runtime/task.rs, so it is reading \
3118             the wrong directory and every guard below would pass vacuously"
3119        );
3120        files.into_iter().map(|(l, s)| (l.to_string(), s)).collect()
3121    }
3122
3123    /// The subset that creates an OS thread.
3124    ///
3125    /// Derived from the source rather than declared, so a file joins the
3126    /// census the day it spawns — which is the one property a hand-written
3127    /// list cannot have. `Strip` so that prose naming a thread API does not
3128    /// enrol a file that creates no thread.
3129    fn censused_files() -> Vec<(String, &'static str)> {
3130        let bare = concat!("thread", "::spawn(");
3131        let files: Vec<(String, &'static str)> = crate_sources()
3132            .into_iter()
3133            .filter(|(_, src)| {
3134                let code = source_guard::production(src, source_guard::Comments::Strip);
3135                code.contains("thread::Builder::new()") || code.contains(bare)
3136            })
3137            .collect();
3138        assert!(
3139            files.len() >= 2,
3140            "expected at least `runtime/task.rs` and `runtime/worker_pool.rs` \
3141             to create threads, found {files:?} — the derivation broke, and \
3142             every guard over it would pass vacuously"
3143        );
3144        files
3145    }
3146
3147    /// Every thread this crate creates states a stack size.
3148    ///
3149    /// `std` gives RTEMS the generic 2 MiB `DEFAULT_MIN_STACK_SIZE`
3150    /// (`std/src/sys/thread/unix.rs`: the carve-out list names vxworks, l4re,
3151    /// espidf and nuttx — not rtems). On the host that is lazily-committed
3152    /// address space and costs nothing measurable; on the target it is carved
3153    /// eagerly out of a fixed pool, which is why an unset stack size is the
3154    /// first ceiling the IOC hits rather than a rounding error.
3155    ///
3156    /// `spawn_dedicated_thread` and [`MandatoryThread`] are enforced by their
3157    /// signatures — the class is a parameter, so a caller cannot omit it. This
3158    /// covers the threads that still build a `std::thread::Builder` directly.
3159    ///
3160    /// It also bans the API that has no class to state:
3161    /// `std::thread::spawn` cannot express a stack size at all, so a site
3162    /// using it does not fail the `Builder` check above — it is invisible to
3163    /// it. Same defect, different anchor. (The bare `thread::spawn` sites
3164    /// elsewhere in the workspace — `ca::repeater`, `ca::calink`,
3165    /// `ca::server::ca_server`, `pva::server::pva_server`, `bridge::pvalink` —
3166    /// are distinct twice over: none is a mandatory IOC thread, and all but the
3167    /// per-command link helpers sit behind `#[cfg(not(target_os = "rtems"))]`
3168    /// module gates, so they are not in the RTEMS closure at all. Every file
3169    /// listed here is.)
3170    ///
3171    /// Fails today, on Linux, with no cross toolchain.
3172    #[test]
3173    fn every_thread_in_this_crate_states_a_stack_size() {
3174        let mut unclassified = Vec::new();
3175        let mut checked = 0usize;
3176        for (label, src) in censused_files() {
3177            let prod = production_scope(src);
3178            for (n, after) in prod.split("thread::Builder::new()").skip(1).enumerate() {
3179                checked += 1;
3180                // The class must be set before the closure is handed over;
3181                // `.spawn(` ends the builder chain.
3182                let chain = after.split(".spawn(").next().unwrap_or("");
3183                if !chain.contains(".stack_size(") {
3184                    unclassified.push(format!("{label} (Builder #{})", n + 1));
3185                }
3186            }
3187            // The classless API. Split so this guard does not match its own
3188            // needle in the file it is written in.
3189            let bare = concat!("thread", "::spawn(");
3190            for (n, line) in prod.lines().enumerate() {
3191                let t = line.trim_start();
3192                if t.starts_with("//") {
3193                    continue;
3194                }
3195                if t.contains(bare) && !t.contains("Builder") {
3196                    unclassified.push(format!("{label}:{} (bare spawn)", n + 1));
3197                }
3198            }
3199        }
3200
3201        // Five: `spawn_dedicated_thread`'s two `cfg` arms, `MandatoryThread`,
3202        // the RT-policy probe, and `worker_pool`'s pooled worker. The floor was
3203        // seven until the three background facilities moved onto
3204        // `MandatoryThread`, which states the class in its constructor —
3205        // `every_background_facility_thread_is_mandatory` is what keeps that
3206        // move from being a hole rather than a hand-off.
3207        assert!(
3208            checked >= 5,
3209            "expected to find the crate's Builder sites, found {checked} — \
3210             did a file move? update this guard's file list"
3211        );
3212        assert!(
3213            unclassified.is_empty(),
3214            "these threads inherit std's 2 MiB default on RTEMS: {unclassified:?}"
3215        );
3216    }
3217
3218    /// The three background facilities create their threads through
3219    /// [`MandatoryThread`], and nothing else.
3220    ///
3221    /// Each of them — the callback bands, `cbTimer`, `scanOnce` — is a thread
3222    /// the IOC cannot correctly run without, and each is created from a
3223    /// constructor reached through a `OnceLock` initialiser, so there is no
3224    /// error path back to whoever decided this IOC serves. They used to resolve
3225    /// the spawn `Result` with `.expect`, which on a `panic = "unwind"` target
3226    /// (RTEMS and VxWorks both default to unwind) killed only the thread that
3227    /// happened to touch the facility first and left the IOC serving without
3228    /// the band, the timer or the `scanOnce` worker.
3229    ///
3230    /// The ban is the structural half: with no raw `Builder` and no bare
3231    /// `thread::spawn` in these files, "mandatory" is not a property a new
3232    /// thread here can forget to declare.
3233    #[test]
3234    fn every_background_facility_thread_is_mandatory() {
3235        let bare = concat!("thread", "::spawn(");
3236        let mut strays = Vec::new();
3237        let mut owned = 0usize;
3238        // The background module, swept whole. Not `censused_files()`: these
3239        // files hold no `Builder` of their own precisely because they went
3240        // through `MandatoryThread`, so deriving their set from thread
3241        // creation would empty it and pass.
3242        let files = source_guard::sweep(source_guard::module_dir!("src/runtime/background"), &[]);
3243        for (label, src) in files {
3244            for (n, line) in production_scope(src).lines().enumerate() {
3245                let t = line.trim_start();
3246                if t.starts_with("//") {
3247                    continue;
3248                }
3249                if t.contains("MandatoryThread::new(") {
3250                    owned += 1;
3251                }
3252                if t.contains("thread::Builder::new()") {
3253                    strays.push(format!("{label}:{} (raw Builder)", n + 1));
3254                }
3255                if t.contains(bare) && !t.contains("Builder") {
3256                    strays.push(format!("{label}:{} (bare spawn)", n + 1));
3257                }
3258            }
3259        }
3260        assert!(
3261            strays.is_empty(),
3262            "a facility thread created outside `MandatoryThread` resolves its \
3263             own spawn failure, and the only resolution that keeps this IOC \
3264             honest is not serving: {strays:?}"
3265        );
3266        assert!(
3267            owned >= 3,
3268            "expected the callback pool, `cbTimer` and `scanOnce`, found {owned} \
3269             `MandatoryThread` sites — a facility moved out of this module"
3270        );
3271    }
3272
3273    #[epics_macros_rs::epics_test]
3274    async fn test_spawn() {
3275        let reactor = Reactor::current().expect("the test driver enters an executor");
3276        let handle = reactor.spawn(async { 42 });
3277        assert_eq!(handle.await.unwrap(), 42);
3278    }
3279
3280    #[epics_macros_rs::epics_test]
3281    async fn test_spawn_blocking() {
3282        let handle = spawn_blocking(|| 123);
3283        assert_eq!(handle.await.unwrap(), 123);
3284    }
3285
3286    /// The property `spawn_dedicated_thread` exists for. A future written for
3287    /// the hosted driver — one that spawns a task and arms a timer — must run
3288    /// unchanged on the thread this hands back. On a plain `std::thread` it
3289    /// does not: it panics with "there is no reactor running" as soon as it is
3290    /// polled, however willing `block_on_sync` was to park.
3291    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
3292    async fn a_dedicated_thread_carries_the_ambient_runtime() {
3293        let (tx, rx) = std::sync::mpsc::channel();
3294        let joined = spawn_dedicated_thread(
3295            "dedicated-with-runtime".into(),
3296            ThreadPriority::CaServerLow,
3297            StackSizeClass::Small,
3298            move || {
3299                let outcome = block_on_sync(async {
3300                    let reactor = Reactor::current()
3301                        .expect("the dedicated thread carries the ambient runtime");
3302                    let inner = reactor.spawn(async { 7u32 }).await.expect("inner task");
3303                    sleep(Duration::from_millis(1)).await;
3304                    inner
3305                });
3306                let _ = tx.send((
3307                    std::thread::current().name().map(str::to_string),
3308                    outcome.ok(),
3309                ));
3310            },
3311        )
3312        .expect("dedicated thread spawned");
3313
3314        let (name, value) = rx
3315            .recv_timeout(Duration::from_secs(5))
3316            .expect("the dedicated thread must complete, not panic");
3317        assert_eq!(name.as_deref(), Some("dedicated-with-runtime"));
3318        assert_eq!(
3319            value,
3320            Some(7),
3321            "a spawn and a timer must both work on the dedicated thread"
3322        );
3323        joined.join().expect("dedicated thread joined");
3324    }
3325
3326    /// The third boundary, and the one that was missing: a **current-thread**
3327    /// ambient runtime.
3328    ///
3329    /// The two neighbours below and above cover "multi-thread ambient" and "no
3330    /// ambient". This is the case between them, and inheriting the handle there
3331    /// is what made `block_on_sync` return `NotBlockable` on a thread that was
3332    /// perfectly able to park — silently, since every caller reads the refusal
3333    /// as "the connection ended". `#[tokio::test]` is `CurrentThread` by
3334    /// default, so this is also the flavor most of the workspace's tests hand a
3335    /// dedicated thread.
3336    ///
3337    /// The assertion is on `block_on_sync` succeeding, not on the absence of a
3338    /// handle, because being able to block is the property the thread is spawned
3339    /// for; how that is arranged is this function's business.
3340    #[epics_macros_rs::epics_test]
3341    async fn a_dedicated_thread_can_block_under_a_current_thread_ambient() {
3342        let (tx, rx) = std::sync::mpsc::channel();
3343        let joined = spawn_dedicated_thread(
3344            "dedicated-current-thread-ambient".into(),
3345            ThreadPriority::Low,
3346            StackSizeClass::Small,
3347            move || {
3348                // A runtime-agnostic await, the only kind a parking thread may
3349                // use — and the exact shape both blocking-io pumps run.
3350                let (ctx, crx) = tokio::sync::mpsc::channel::<u32>(1);
3351                let outcome = block_on_sync(async move {
3352                    ctx.send(9u32).await.expect("send into a depth-1 channel");
3353                    let mut crx = crx;
3354                    crx.recv().await
3355                });
3356                let _ = tx.send(outcome.ok().flatten());
3357            },
3358        )
3359        .expect("dedicated thread spawned");
3360
3361        assert_eq!(
3362            rx.recv_timeout(Duration::from_secs(5))
3363                .expect("the dedicated thread must complete, not panic"),
3364            Some(9),
3365            "a dedicated thread must be able to park under a current-thread \
3366             ambient runtime; inheriting that handle makes block_on_sync \
3367             refuse and every pump built on it exit at once"
3368        );
3369        joined.join().expect("dedicated thread joined");
3370    }
3371
3372    /// The other boundary: no runtime to capture. The thread still runs, and a
3373    /// runtime-agnostic await still completes — that is `park_on`, and it is
3374    /// the only arm RTEMS ever takes.
3375    #[test]
3376    fn a_dedicated_thread_runs_without_an_ambient_runtime() {
3377        let (tx, rx) = std::sync::mpsc::channel();
3378        let joined = spawn_dedicated_thread(
3379            "dedicated-no-runtime".into(),
3380            ThreadPriority::Low,
3381            StackSizeClass::Small,
3382            move || {
3383                let _ = tx.send((
3384                    std::thread::current().name().map(str::to_string),
3385                    block_on_sync(async { 5u32 }).ok(),
3386                ));
3387            },
3388        )
3389        .expect("dedicated thread spawned");
3390
3391        let (name, value) = rx
3392            .recv_timeout(Duration::from_secs(5))
3393            .expect("the dedicated thread must run with no runtime to capture");
3394        assert_eq!(name.as_deref(), Some("dedicated-no-runtime"));
3395        assert_eq!(value, Some(5));
3396        joined.join().expect("dedicated thread joined");
3397    }
3398
3399    // --- The band-blocking invariant -----------------------------------------
3400    //
3401    // MUST NOT: work running on a background-facility worker thread — a
3402    // callback band, the delayed timer, the scanOnce worker — block that
3403    // thread on async progress. The gate is `block_on_sync`; the mark is set
3404    // by `background::facility::run_facility_loop`, the one function every
3405    // worker loop goes through.
3406    //
3407    // Each of the three cases below is written so that a *broken* gate fails
3408    // the test instead of hanging it: the awaited future is completable from
3409    // the test thread, so a worker that parked can always be released before
3410    // the assertion runs and the pool's `Drop` can still join it.
3411
3412    /// The case the invariant exists for: a future spawned onto a callback
3413    /// band. On RTEMS this is exactly what [`Reactor::spawn`] produces, and the
3414    /// band has
3415    /// one worker — parking it stops every deferred callback, every FLNK tail
3416    /// and every other monitor on that band.
3417    #[test]
3418    fn a_future_on_a_callback_band_is_refused_a_blocking_bridge() {
3419        use crate::runtime::background::callback_executor::CallbackPool;
3420        use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_future};
3421
3422        let pool = CallbackPool::new();
3423        // Held by the test: `recv()` never completes until we send, so a gate
3424        // that does not refuse leaves the worker parked here.
3425        let (release, mut park_here) = tokio::sync::mpsc::channel::<()>(1);
3426        let (report, outcome) = std::sync::mpsc::channel();
3427
3428        let _handle = spawn_future(&pool.handle(), DEFAULT_SPAWN_PRIORITY, async move {
3429            let _ = report.send(block_on_sync(async move { park_here.recv().await }));
3430        });
3431
3432        let got = outcome.recv_timeout(Duration::from_secs(5));
3433        // Release a worker the gate failed to protect, so the assertions below
3434        // report a failure instead of hanging `CallbackPool::drop`'s join.
3435        let _ = release.try_send(());
3436
3437        match got {
3438            Ok(result) => assert_eq!(
3439                result.map(|v| v.is_some()),
3440                Err(NotBlockable::BackgroundWorker),
3441                "a band worker must be refused the blocking bridge, not given one"
3442            ),
3443            Err(_) => panic!(
3444                "the band worker parked inside block_on_sync instead of being \
3445                 refused — the band has one worker, so this is the deadlock the \
3446                 invariant exists to prevent"
3447            ),
3448        }
3449    }
3450
3451    /// The same thread, reached the other way: `spawn_blocking` also lands on a
3452    /// band worker under the exec backend, and a blocking closure holds that
3453    /// worker for its whole run. The rule is a property of the thread, so it
3454    /// must not depend on which spawn put the work there.
3455    #[test]
3456    fn a_blocking_closure_on_a_callback_band_is_refused_too() {
3457        use crate::runtime::background::callback_executor::CallbackPool;
3458        use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_blocking_on};
3459
3460        let pool = CallbackPool::new();
3461        let (release, mut park_here) = tokio::sync::mpsc::channel::<()>(1);
3462        let (report, outcome) = std::sync::mpsc::channel();
3463
3464        let _handle = spawn_blocking_on(&pool.handle(), DEFAULT_SPAWN_PRIORITY, move || {
3465            let _ = report.send(block_on_sync(async move { park_here.recv().await }));
3466        });
3467
3468        let got = outcome.recv_timeout(Duration::from_secs(5));
3469        let _ = release.try_send(());
3470
3471        match got {
3472            Ok(result) => assert_eq!(
3473                result.map(|v| v.is_some()),
3474                Err(NotBlockable::BackgroundWorker),
3475                "the refusal keys on the thread, not on how work reached it"
3476            ),
3477            Err(_) => panic!("the band worker parked instead of being refused"),
3478        }
3479    }
3480
3481    /// The other side of the boundary, so the gate cannot be satisfied by
3482    /// refusing everything: an ordinary thread that merely *submits* to the
3483    /// pool still blocks. The mark covers the worker loop's own thread and
3484    /// nothing else.
3485    #[test]
3486    fn a_thread_that_only_submits_to_a_band_still_blocks() {
3487        use crate::runtime::background::callback_executor::{CallbackPool, CallbackPriority};
3488
3489        let pool = CallbackPool::new();
3490        let (tx, rx) = std::sync::mpsc::channel();
3491        pool.request(
3492            CallbackPriority::Medium,
3493            Box::new(move || tx.send(1u32).unwrap()),
3494        )
3495        .expect("the band accepts the callback");
3496        assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), 1);
3497        assert_eq!(
3498            block_on_sync(async { 5u32 }),
3499            Ok(5),
3500            "the submitting thread runs no facility loop, so it may still park"
3501        );
3502    }
3503
3504    #[epics_macros_rs::epics_test]
3505    async fn test_sleep() {
3506        let start = std::time::Instant::now();
3507        sleep(Duration::from_millis(10)).await;
3508        assert!(start.elapsed() >= Duration::from_millis(10));
3509    }
3510
3511    // The two halves of `timeout`'s contract. They read as trivial against a
3512    // tokio delegation, and that is the point: they are what a later backend
3513    // swap has to keep true, on a seam whose whole purpose is to be
3514    // reimplemented.
3515    #[epics_macros_rs::epics_test]
3516    async fn timeout_yields_the_value_when_the_future_finishes_first() {
3517        let r = timeout(Duration::from_secs(30), async { 42 }).await;
3518        assert_eq!(r.unwrap(), 42);
3519    }
3520
3521    #[epics_macros_rs::epics_test]
3522    async fn timeout_elapses_on_a_future_that_never_finishes() {
3523        let r = timeout(Duration::from_millis(10), std::future::pending::<()>()).await;
3524        assert!(r.is_err());
3525    }
3526
3527    // The boundary the two tests above cannot reach, because
3528    // `#[epics_test]` always gives them a driver: a thread with no runtime
3529    // at all. `tokio::time` panics there, so the hosted half of this seam
3530    // used to as well, and every caller that could not prove a runtime
3531    // simply stopped passing a deadline — see `asyn-rs`'s
3532    // `PortHandle::await_reply`. `park_on` is the driver a blocking bridge
3533    // really uses, and the background timer thread is what wakes it.
3534    #[test]
3535    fn a_bounded_wait_fires_on_a_thread_with_no_runtime() {
3536        assert!(
3537            tokio::runtime::Handle::try_current().is_err(),
3538            "the premise of this test"
3539        );
3540        let started = std::time::Instant::now();
3541        let r = park_on(timeout(
3542            Duration::from_millis(30),
3543            std::future::pending::<()>(),
3544        ));
3545        assert!(r.is_err(), "the deadline is the only thing that can end it");
3546        let waited = started.elapsed();
3547        assert!(
3548            waited >= Duration::from_millis(30) && waited < Duration::from_secs(2),
3549            "woke at its own deadline (waited {waited:?})"
3550        );
3551    }
3552
3553    #[test]
3554    fn a_deadline_wait_fires_on_a_thread_with_no_runtime() {
3555        let started = std::time::Instant::now();
3556        let r = park_on(timeout_at(
3557            deadline_from_now(Duration::from_millis(30)),
3558            std::future::pending::<()>(),
3559        ));
3560        assert!(r.is_err());
3561        let waited = started.elapsed();
3562        assert!(
3563            waited >= Duration::from_millis(30) && waited < Duration::from_secs(2),
3564            "woke at its own deadline (waited {waited:?})"
3565        );
3566    }
3567
3568    #[test]
3569    fn a_sleep_on_a_thread_with_no_runtime_returns() {
3570        let started = std::time::Instant::now();
3571        park_on(sleep(Duration::from_millis(10)));
3572        assert!(started.elapsed() >= Duration::from_millis(10));
3573    }
3574
3575    /// The other half of the same boundary: a runtime thread must still take
3576    /// tokio's timer, because that is the only one that moves with tokio's
3577    /// clock — which `#[tokio::test(start_paused = true)]` makes virtual. The
3578    /// discriminator is the background executor's `OnceLock`: it is untouched
3579    /// unless something reached for the wall-clock timer. Relies on nextest's
3580    /// process-per-test isolation for the "untouched" half.
3581    #[cfg(tokio_backend)]
3582    #[tokio::test]
3583    async fn a_bounded_wait_on_a_runtime_thread_keeps_the_runtime_s_timer() {
3584        assert!(
3585            BACKGROUND.get().is_none(),
3586            "the premise: nothing has started the background facility yet"
3587        );
3588        let r = timeout(Duration::from_millis(10), std::future::pending::<()>()).await;
3589        assert!(r.is_err());
3590        assert!(
3591            BACKGROUND.get().is_none(),
3592            "a runtime thread armed tokio's timer, not the background one"
3593        );
3594    }
3595
3596    #[test]
3597    fn priority_named_levels_match_epics_thread_h() {
3598        // epicsThread.h:73-83 named-level constants.
3599        assert_eq!(ThreadPriority::Low.value(), 10);
3600        assert_eq!(ThreadPriority::CaServerLow.value(), 20);
3601        assert_eq!(ThreadPriority::CaServerHigh.value(), 40);
3602        assert_eq!(ThreadPriority::Medium.value(), 50);
3603        assert_eq!(ThreadPriority::ScanLow.value(), 60);
3604        assert_eq!(ThreadPriority::ScanHigh.value(), 70);
3605        assert_eq!(ThreadPriority::High.value(), 90);
3606        assert_eq!(ThreadPriority::Iocsh.value(), 91);
3607    }
3608
3609    #[test]
3610    fn priority_ordering_ca_server_below_scan() {
3611        // Real-time invariant: scan threads must outrank CA-server
3612        // threads so scans preempt the CA server on a loaded IOC.
3613        assert!(ThreadPriority::CaServerHigh.value() < ThreadPriority::ScanLow.value());
3614        assert!(ThreadPriority::CaServerLow.value() < ThreadPriority::ScanLow.value());
3615    }
3616
3617    #[test]
3618    fn priority_custom_clamps_to_max() {
3619        assert_eq!(ThreadPriority::Custom(200).value(), PRIORITY_MAX);
3620        assert_eq!(ThreadPriority::Custom(99).value(), 99);
3621        assert_eq!(ThreadPriority::Custom(0).value(), PRIORITY_MIN);
3622    }
3623
3624    #[test]
3625    fn stack_size_classes_ordered() {
3626        // STACK_SIZE table is strictly increasing Small < Medium < Big.
3627        assert!(StackSizeClass::Small.bytes() < StackSizeClass::Medium.bytes());
3628        assert!(StackSizeClass::Medium.bytes() < StackSizeClass::Big.bytes());
3629        // Small = 0x10000 * sizeof(usize).
3630        assert_eq!(
3631            StackSizeClass::Small.bytes(),
3632            0x10000 * std::mem::size_of::<usize>()
3633        );
3634    }
3635
3636    /// The three classes against the C table, factor by factor.
3637    ///
3638    /// `STACK_SIZE(f) = f * 0x10000 * sizeof(void*)` with factors 1, 2, 4
3639    /// (`libcom/src/osi/os/posix/osdThread.c:506-509`) — the file a C IOC on
3640    /// RTEMS 6 compiles, because `configure/toolchain.c:29-35` picks
3641    /// `OS_API = posix` for `__RTEMS_MAJOR__ >= 5`. Pinning the factors
3642    /// separately from the unit is what makes a silent edit of one of them
3643    /// fail: `stack_size_classes_ordered` above is satisfied by any
3644    /// increasing triple.
3645    #[test]
3646    fn the_classes_are_the_c_posix_table_factor_for_factor() {
3647        let unit = 0x10000 * std::mem::size_of::<usize>();
3648        assert_eq!(StackSizeClass::Small.bytes(), unit);
3649        assert_eq!(StackSizeClass::Medium.bytes(), 2 * unit);
3650        assert_eq!(StackSizeClass::Big.bytes(), 4 * unit);
3651        // And what the same table yields on the target the RTEMS port builds
3652        // for (`sizeof(void*) == 4`), spelled out so a reader on a 64-bit
3653        // host does not have to re-derive it.
3654        const TARGET_UNIT: usize = 0x10000 * 4;
3655        assert_eq!(
3656            [TARGET_UNIT, 2 * TARGET_UNIT, 4 * TARGET_UNIT],
3657            [256 * 1024, 512 * 1024, 1024 * 1024],
3658            "armv7-rtems-eabihf: Small / Medium / Big in bytes"
3659        );
3660    }
3661
3662    /// The stack a caller *states* is the stack the thread *reports*.
3663    ///
3664    /// The source guard above only proves a number reached the builder. This
3665    /// asks the running thread what it actually got, through
3666    /// `pthread_getattr_np`, and that is the property the RTEMS ceiling
3667    /// depends on: `std` gates its 2 MiB `DEFAULT_MIN_STACK_SIZE` on a
3668    /// carve-out list that omits rtems, so a size that fails to arrive is
3669    /// silently 2 MiB rather than an error.
3670    ///
3671    /// The mechanism this exercises is not host-specific: `std`'s
3672    /// `Thread::new` calls `pthread_attr_setstacksize(attr, max(stack,
3673    /// PTHREAD_STACK_MIN))` on every non-espidf/nuttx unix
3674    /// (`std/src/sys/thread/unix.rs`), and `libc` gives rtems
3675    /// `PTHREAD_STACK_MIN = 0`, so the `max` cannot raise our request there
3676    /// either. Glibc-only because `pthread_getattr_np` is the readback API.
3677    #[cfg(all(target_os = "linux", target_env = "gnu"))]
3678    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
3679    async fn a_dedicated_thread_reports_the_stack_it_was_asked_for() {
3680        fn reported_stack_bytes() -> usize {
3681            unsafe {
3682                let mut attr: libc::pthread_attr_t = std::mem::zeroed();
3683                assert_eq!(
3684                    libc::pthread_getattr_np(libc::pthread_self(), &mut attr),
3685                    0,
3686                    "pthread_getattr_np"
3687                );
3688                let mut addr: *mut libc::c_void = std::ptr::null_mut();
3689                let mut size: libc::size_t = 0;
3690                assert_eq!(
3691                    libc::pthread_attr_getstack(&attr, &mut addr, &mut size),
3692                    0,
3693                    "pthread_attr_getstack"
3694                );
3695                libc::pthread_attr_destroy(&mut attr);
3696                size
3697            }
3698        }
3699
3700        for class in [
3701            StackSizeClass::Small,
3702            StackSizeClass::Medium,
3703            StackSizeClass::Big,
3704        ] {
3705            let (tx, rx) = std::sync::mpsc::channel();
3706            let joined = spawn_dedicated_thread(
3707                format!("stack-readback-{class:?}"),
3708                ThreadPriority::CaServerLow,
3709                class,
3710                move || {
3711                    let _ = tx.send(reported_stack_bytes());
3712                },
3713            )
3714            .expect("dedicated thread spawned");
3715            let got = rx.recv().expect("the thread reported its stack");
3716            joined.join().expect("thread joined");
3717
3718            let asked = class.bytes();
3719            // The kernel rounds up to a page; it must never round *down*, and
3720            // it must not silently substitute something of a different order.
3721            assert!(
3722                got >= asked && got < asked + 64 * 1024,
3723                "{class:?}: asked for {asked} bytes, thread reports {got}"
3724            );
3725        }
3726    }
3727
3728    /// The distinguishing half of the readback: a class below `std`'s default
3729    /// must land below it. Without this the test above would pass on a
3730    /// platform that ignored every request and handed out 2 MiB, for the two
3731    /// classes that happen to be smaller than that on a 64-bit host.
3732    #[cfg(all(target_os = "linux", target_env = "gnu"))]
3733    #[test]
3734    fn the_small_classes_are_below_the_default_that_would_mask_a_failure() {
3735        const STD_DEFAULT_MIN_STACK_SIZE: usize = 2 * 1024 * 1024;
3736        assert!(StackSizeClass::Small.bytes() < STD_DEFAULT_MIN_STACK_SIZE);
3737        assert!(StackSizeClass::Medium.bytes() < STD_DEFAULT_MIN_STACK_SIZE);
3738    }
3739
3740    #[test]
3741    fn apply_priority_returns_a_defined_outcome() {
3742        // The result depends on the platform + permissions of the test
3743        // host; we only assert it is one of the defined outcomes and
3744        // does not panic. On a CI box without CAP_SYS_NICE this is
3745        // typically BestEffortFailed — which is C-parity behaviour.
3746        let outcome = apply_to_current_thread(ThreadPriority::ScanHigh);
3747        assert!(matches!(
3748            outcome,
3749            PriorityApplied::Realtime
3750                | PriorityApplied::Disabled
3751                | PriorityApplied::Unsupported
3752                | PriorityApplied::BestEffortFailed
3753        ));
3754    }
3755
3756    /// Both defaults, and both override directions against each default.
3757    ///
3758    /// The RTEMS arm is unreachable from a host test run unless the default
3759    /// is a function of the target rather than a `cfg` block, which is why
3760    /// `default_policy`/`resolve` take their input explicitly.
3761    #[test]
3762    fn the_rt_default_is_on_for_rtems_and_off_for_hosted() {
3763        // (1) the two defaults themselves
3764        assert_eq!(
3765            default_policy(true),
3766            RtPolicy::AllowRealtime,
3767            "RTEMS honours its priorities by default, as base does \
3768             (EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING=YES, CONFIG_ENV:57)"
3769        );
3770        assert_eq!(
3771            default_policy(false),
3772            RtPolicy::Disabled,
3773            "hosted stays opt-in: RLIMIT_RTPRIO makes the request fail on a \
3774             desktop, and where it succeeds a runaway band wedges the machine"
3775        );
3776
3777        // (2) the compiled-in default is wired to the target, not to a guess
3778        assert_eq!(DEFAULT_POLICY, default_policy(cfg!(epics_embedded_target)));
3779        assert_eq!(RtPolicy::from_env_value(None), DEFAULT_POLICY);
3780
3781        // (3) an explicit value wins over EITHER default, in BOTH directions.
3782        //     The RTEMS-off case is the one an operator needs: turning RT
3783        //     scheduling off on a target that defaults to on.
3784        for default in [RtPolicy::AllowRealtime, RtPolicy::Disabled] {
3785            assert_eq!(
3786                RtPolicy::resolve(Some("NO"), default),
3787                RtPolicy::Disabled,
3788                "explicit NO must turn it off even where the default is {default:?}"
3789            );
3790            assert_eq!(
3791                RtPolicy::resolve(Some("YES"), default),
3792                RtPolicy::AllowRealtime,
3793                "explicit YES must turn it on even where the default is {default:?}"
3794            );
3795            assert_eq!(
3796                RtPolicy::resolve(None, default),
3797                default,
3798                "unset must resolve to the default and nothing else"
3799            );
3800        }
3801    }
3802
3803    #[test]
3804    fn rt_switch_explicit_values_win_over_the_default() {
3805        // Unset takes the target's default; that is
3806        // `the_rt_default_is_on_for_rtems_and_off_for_hosted`'s subject.
3807        assert_eq!(RtPolicy::from_env_value(None), DEFAULT_POLICY);
3808        // C's `envGetBoolConfigParam` (envSubr.c:331) accepts only
3809        // case-insensitive "yes"; we also take the spellings a hand-written
3810        // startup script is likely to use.
3811        for on in ["YES", "yes", "Yes", "true", "TRUE", "on", "1", " yes "] {
3812            assert_eq!(
3813                RtPolicy::from_env_value(Some(on)),
3814                RtPolicy::AllowRealtime,
3815                "{on:?} should turn the switch on"
3816            );
3817        }
3818        // Everything else is off. Silence is the safe direction: a
3819        // misspelling must never grant a process RT scheduling.
3820        for off in ["", "NO", "no", "false", "off", "0", "y", "yes please", "2"] {
3821            assert_eq!(
3822                RtPolicy::from_env_value(Some(off)),
3823                RtPolicy::Disabled,
3824                "{off:?} should leave the switch off"
3825            );
3826        }
3827    }
3828
3829    /// Switch off ⟹ the OS scheduler is never called.
3830    ///
3831    /// Mutation check: deleting the `RtPolicy::Disabled` arm in
3832    /// `apply_to_current_thread_under` (so it always calls
3833    /// `apply_priority_impl`) makes the `SCHED_CALLS` assertion fail.
3834    #[cfg(target_os = "linux")]
3835    #[test]
3836    fn switch_off_makes_no_scheduler_calls() {
3837        let before = SCHED_CALLS.with(std::cell::Cell::get);
3838        for p in [
3839            ThreadPriority::Low,
3840            ThreadPriority::CaServerLow,
3841            ThreadPriority::ScanHigh,
3842            ThreadPriority::Iocsh,
3843            ThreadPriority::Custom(0),
3844            ThreadPriority::Custom(99),
3845        ] {
3846            assert_eq!(
3847                apply_to_current_thread_under(RtPolicy::Disabled, p),
3848                PriorityApplied::Disabled
3849            );
3850        }
3851        assert_eq!(
3852            SCHED_CALLS.with(std::cell::Cell::get),
3853            before,
3854            "the switch is off, so nothing may reach pthread_setschedparam"
3855        );
3856    }
3857
3858    /// Switch on: either the host grants SCHED_FIFO — and then the policy
3859    /// must actually be in force on the thread, at the mapped priority — or
3860    /// it does not, and the thread keeps running under the default policy.
3861    ///
3862    /// Runs on its own thread: on a host that *does* grant RT, leaving the
3863    /// test-harness thread in a real-time band would outlive the test.
3864    #[cfg(target_os = "linux")]
3865    #[test]
3866    fn switch_on_either_sticks_or_falls_back_without_killing_the_thread() {
3867        let outcome = std::thread::spawn(|| {
3868            // Resolve (and cache) the probe first, so what the call below is
3869            // expected to do is known rather than order-dependent.
3870            let range = permitted_fifo_range();
3871            let before = SCHED_CALLS.with(std::cell::Cell::get);
3872            let outcome =
3873                apply_to_current_thread_under(RtPolicy::AllowRealtime, ThreadPriority::ScanHigh);
3874            let calls = SCHED_CALLS.with(std::cell::Cell::get) - before;
3875
3876            // Whatever the host allowed, this thread is still running.
3877            assert_eq!(2 + 2, 4);
3878
3879            let mut policy = 0i32;
3880            let mut param = libc::sched_param { sched_priority: 0 };
3881            // SAFETY: reads the calling thread's own scheduling into
3882            // stack-local outputs.
3883            let rc = unsafe {
3884                libc::pthread_getschedparam(libc::pthread_self(), &mut policy, &mut param)
3885            };
3886            assert_eq!(rc, 0, "pthread_getschedparam failed");
3887
3888            match (range, outcome) {
3889                (RtRange::Available { min, max }, PriorityApplied::Realtime) => {
3890                    // The host permits FIFO — assert the policy stuck, at
3891                    // the C-mapped priority, off exactly one scheduler call.
3892                    assert_eq!(calls, 1, "one apply must be one scheduler call");
3893                    assert_eq!(policy, libc::SCHED_FIFO, "SCHED_FIFO did not stick");
3894                    assert_eq!(
3895                        param.sched_priority,
3896                        map_epics_priority(ThreadPriority::ScanHigh.value(), min, max),
3897                        "wrong OS priority for epicsThreadPriorityScanHigh"
3898                    );
3899                }
3900                (RtRange::Denied, PriorityApplied::BestEffortFailed) => {
3901                    // Unprivileged: the fallback leaves the thread at the
3902                    // default policy rather than failing the caller, and —
3903                    // the anti-spam property — asks the OS nothing further
3904                    // now that the probe has settled the question once.
3905                    assert_eq!(calls, 0, "a settled denial must not re-ask the OS");
3906                    assert_ne!(
3907                        policy,
3908                        libc::SCHED_FIFO,
3909                        "fallback reported but the thread is real-time scheduled"
3910                    );
3911                }
3912                (RtRange::Unsupported, PriorityApplied::Unsupported) => {
3913                    assert_eq!(calls, 0, "no SCHED_FIFO range means no scheduler call");
3914                }
3915                (range, outcome) => {
3916                    panic!("range {range:?} and outcome {outcome:?} disagree")
3917                }
3918            }
3919            outcome
3920        })
3921        .join()
3922        .expect("probe thread panicked");
3923        eprintln!("host RT outcome: {outcome:?}");
3924    }
3925
3926    /// The unprivileged fallback is logged once, not once per thread.
3927    #[cfg(target_os = "linux")]
3928    #[test]
3929    fn denial_is_reported_once_not_per_thread() {
3930        let threads: Vec<_> = (0..8)
3931            .map(|_| {
3932                std::thread::spawn(|| {
3933                    for _ in 0..8 {
3934                        let _ = apply_to_current_thread_under(
3935                            RtPolicy::AllowRealtime,
3936                            ThreadPriority::Low,
3937                        );
3938                    }
3939                })
3940            })
3941            .collect();
3942        for t in threads {
3943            t.join().expect("worker panicked");
3944        }
3945        assert!(
3946            DENIED_WARNINGS.load(std::sync::atomic::Ordering::Relaxed) <= 1,
3947            "64 denied requests across 8 threads must not produce more than one warning"
3948        );
3949    }
3950
3951    /// The mapping itself, against `epicsThreadGetPosixPriority`
3952    /// (`osdThread.c:129-144`).
3953    #[cfg(target_os = "linux")]
3954    #[test]
3955    fn epics_priority_maps_onto_the_permitted_fifo_range() {
3956        // Linux's nominal SCHED_FIFO range.
3957        let (min, max) = (1, 99);
3958        // oss = p * (max-min)/100 + min
3959        assert_eq!(map_epics_priority(0, min, max), 1);
3960        assert_eq!(map_epics_priority(20, min, max), 1 + (20.0 * 0.98) as i32);
3961        assert_eq!(map_epics_priority(99, min, max), 1 + (99.0 * 0.98) as i32);
3962        // Ordering is preserved: the CA server sits below the scan bands.
3963        assert!(
3964            map_epics_priority(ThreadPriority::CaServerHigh.value(), min, max)
3965                < map_epics_priority(ThreadPriority::ScanLow.value(), min, max)
3966        );
3967        // A range restricted by RLIMIT_RTPRIO still spans the whole EPICS
3968        // space rather than saturating at the top.
3969        assert_eq!(map_epics_priority(0, 1, 10), 1);
3970        assert_eq!(map_epics_priority(99, 1, 10), 1 + (99.0 * 0.09) as i32);
3971        // Degenerate range collapses (osdThread.c:133).
3972        assert_eq!(map_epics_priority(50, 7, 7), 7);
3973    }
3974
3975    /// The RTEMS map's *image* is the whole point of choosing it, so assert
3976    /// the image, not sampled points: for **every** `u8` input the resulting
3977    /// RTEMS core priority lands in `100..=199`, and therefore below libbsd's
3978    /// network threads.
3979    ///
3980    /// Provenance of the band, measured on the bring-up guest (RTEMS 6 +
3981    /// libbsd, QEMU `xilinx_zynq_a9`) — lower core number is *more* urgent:
3982    ///
3983    /// | core | thread |
3984    /// |------|--------|
3985    /// | 96   | libbsd `IRQS` (interrupt server) |
3986    /// | 98   | libbsd `TIME` |
3987    /// | 100  | libbsd default — twelve further network threads |
3988    /// | 254  | DHCP, outside the band |
3989    /// | 255  | idle, `RTEMS_MAXIMUM_PRIORITY` |
3990    ///
3991    /// So `core >= 100` means: never more urgent than any libbsd network
3992    /// thread, and strictly less urgent than `IRQS`/`TIME`. That is a
3993    /// property of the map's construction (fixed offsets over a 100-wide
3994    /// EPICS space), not of the endpoints, which is why no input — including
3995    /// the out-of-range ones `ThreadPriority::value` cannot currently produce
3996    /// — can escape it.
3997    #[test]
3998    fn rtems_priority_map_stays_below_the_libbsd_network_band() {
3999        /// libbsd's most urgent network thread on the measured guest.
4000        const LIBBSD_IRQS_CORE: i32 = 96;
4001        /// libbsd's default band; twelve of its threads sit here.
4002        const LIBBSD_DEFAULT_CORE: i32 = 100;
4003        // The guest's settable SCHED_FIFO range.
4004        const POSIX_MIN: i32 = 1;
4005        const POSIX_MAX: i32 = 254;
4006
4007        for epics in 0..=u8::MAX {
4008            let posix = map_epics_priority_rtems(epics);
4009            let core = RTEMS_MAXIMUM_PRIORITY - posix;
4010            assert!(
4011                (POSIX_MIN..=POSIX_MAX).contains(&posix),
4012                "EPICS {epics} maps to posix {posix}, outside the settable \
4013                 [{POSIX_MIN}, {POSIX_MAX}]"
4014            );
4015            assert!(
4016                core >= LIBBSD_DEFAULT_CORE,
4017                "EPICS {epics} maps to core {core}, more urgent than libbsd's \
4018                 default band ({LIBBSD_DEFAULT_CORE}) and its IRQS \
4019                 ({LIBBSD_IRQS_CORE})"
4020            );
4021            assert!(
4022                core <= RTEMS_MAXIMUM_PRIORITY - 56,
4023                "EPICS {epics} maps to core {core}, less urgent than the \
4024                 EPICS band's own floor of 199"
4025            );
4026        }
4027        // The two ends of the EPICS space, as `RTEMS-score/osdThread.c:94-102`
4028        // defines them, and the clamp above it.
4029        assert_eq!(map_epics_priority_rtems(0), 56);
4030        assert_eq!(map_epics_priority_rtems(99), 155);
4031        assert_eq!(map_epics_priority_rtems(100), 155);
4032        assert_eq!(map_epics_priority_rtems(u8::MAX), 155);
4033        // Ordering still holds: a higher EPICS priority is a more urgent core.
4034        assert!(
4035            RTEMS_MAXIMUM_PRIORITY - map_epics_priority_rtems(ThreadPriority::ScanLow.value())
4036                < RTEMS_MAXIMUM_PRIORITY
4037                    - map_epics_priority_rtems(ThreadPriority::CaServerHigh.value())
4038        );
4039    }
4040
4041    /// [`map_epics_priority_vxworks`] must land on the exact same POSIX
4042    /// values as [`map_epics_priority_rtems`] — that equality is the
4043    /// measured fact [`DEFAULT_POLICY`]'s doc cites, and this function
4044    /// deliberately does not call into the RTEMS one (see its own doc), so
4045    /// nothing else pins the two together if one of them drifts.
4046    #[test]
4047    fn vxworks_priority_map_matches_the_rtems_posix_values() {
4048        for epics in 0..=u8::MAX {
4049            assert_eq!(
4050                map_epics_priority_vxworks(epics),
4051                map_epics_priority_rtems(epics),
4052                "EPICS {epics}: VxWorks and RTEMS must set the identical POSIX \
4053                 SCHED_FIFO value"
4054            );
4055        }
4056        // The measured endpoints, restated directly per this function's own
4057        // doc rather than only via the equality above.
4058        assert_eq!(map_epics_priority_vxworks(0), 56);
4059        assert_eq!(map_epics_priority_vxworks(99), 155);
4060        assert_eq!(map_epics_priority_vxworks(100), 155);
4061        assert_eq!(map_epics_priority_vxworks(u8::MAX), 155);
4062    }
4063
4064    /// The RTEMS map is not the hosted map with retuned endpoints, and the
4065    /// difference is exactly the reason the RTEMS arm exists.
4066    ///
4067    /// Stated as the hazard rather than as `assert_ne!` on a sample: over the
4068    /// EPICS space, feeding the *hosted linear* map the guest's own probed
4069    /// range (`min=1`, `max=254`) puts some priorities above libbsd's `IRQS`
4070    /// at core 96, and the fixed RTEMS map puts none there. The crossover is
4071    /// pinned at EPICS 63 because that number is the justification recorded in
4072    /// the commit message; if base's map or the measured band ever moves, this
4073    /// fails rather than the deviation quietly losing its reason.
4074    #[test]
4075    fn rtems_priority_map_is_not_the_hosted_linear_map() {
4076        const LIBBSD_IRQS_CORE: i32 = 96;
4077        // What `find_pri_range` yields on the guest (osdThread.c:295-311).
4078        let (min, max) = (1, 254);
4079
4080        let hosted_core = |epics: u8| RTEMS_MAXIMUM_PRIORITY - map_epics_priority(epics, min, max);
4081        let rtems_core = |epics: u8| RTEMS_MAXIMUM_PRIORITY - map_epics_priority_rtems(epics);
4082
4083        let hosted_above_irqs: Vec<u8> = (0..=99)
4084            .filter(|&e| hosted_core(e) < LIBBSD_IRQS_CORE)
4085            .collect();
4086        let rtems_above_irqs: Vec<u8> = (0..=99)
4087            .filter(|&e| rtems_core(e) < LIBBSD_IRQS_CORE)
4088            .collect();
4089
4090        assert_eq!(
4091            rtems_above_irqs,
4092            Vec::<u8>::new(),
4093            "the RTEMS map must place no EPICS priority above libbsd's IRQS"
4094        );
4095        assert_eq!(
4096            hosted_above_irqs.first().copied(),
4097            Some(63),
4098            "base-on-RTEMS-6's posix map crosses IRQS at EPICS 63; that number \
4099             is the recorded reason for this deviation"
4100        );
4101        // And concretely at the CA server band the audit cares about.
4102        assert_eq!(
4103            hosted_core(91),
4104            24,
4105            "upstream posix map: EPICS 91 -> core 24"
4106        );
4107        assert_eq!(rtems_core(91), 108, "this port: EPICS 91 -> core 108");
4108        // Shapes, not endpoints: the hosted map spans the whole probed range,
4109        // this one spans exactly 100 levels wherever it is placed.
4110        assert_eq!(
4111            map_epics_priority_rtems(99) - map_epics_priority_rtems(0),
4112            99
4113        );
4114        assert_eq!(
4115            map_epics_priority(99, min, max) - map_epics_priority(0, min, max),
4116            250
4117        );
4118    }
4119
4120    /// The name budget an RTEMS thread object actually has:
4121    /// `CONFIGURE_MAXIMUM_THREAD_NAME_SIZE` defaults to 16 *including* the
4122    /// NUL (`rtems/score/thread.h:1079` (`rtems_6`), `rtems/confdefs/threads.h:92-93` (`rtems_6`))
4123    /// and the boot shim does not override it, so 15 bytes — the same budget
4124    /// `std` truncates to on Linux.
4125    ///
4126    /// Truncating here rather than letting `_Thread_Set_name` do it is what
4127    /// keeps the call's result meaningful: that function `strlcpy`s and
4128    /// *still applies* the truncated name, but returns
4129    /// `STATUS_RESULT_TOO_LARGE` → `ERANGE`, so an untruncated call would log
4130    /// a failure for a name it had in fact set.
4131    #[test]
4132    fn thread_names_are_cut_to_the_rtems_budget_on_a_char_boundary() {
4133        assert_eq!(RTEMS_MAX_THREAD_NAME_BYTES, 15);
4134        // Short names pass through untouched.
4135        assert_eq!(truncate_thread_name("CAS-event"), "CAS-event");
4136        // Exactly at the budget.
4137        assert_eq!(truncate_thread_name("123456789012345"), "123456789012345");
4138        // Over it — a real per-client CA thread name.
4139        assert_eq!(
4140            truncate_thread_name("CAS-client-blocking 10.0.0.1:5064"),
4141            "CAS-client-blo"[..14].to_owned() + "c"
4142        );
4143        assert!(truncate_thread_name("CAS-client-blocking 10.0.0.1:5064").len() <= 15);
4144        // Never mid-codepoint: 'é' is two bytes, so a cut landing inside it
4145        // must step back rather than produce invalid UTF-8. 14 ASCII bytes
4146        // plus 'é' is 16 bytes; the budget cuts at 15, inside the 'é'.
4147        let mixed = "aaaaaaaaaaaaaaé";
4148        assert_eq!(mixed.len(), 16);
4149        assert_eq!(truncate_thread_name(mixed), "aaaaaaaaaaaaaa");
4150        // Empty stays empty rather than underflowing the boundary walk.
4151        assert_eq!(truncate_thread_name(""), "");
4152    }
4153
4154    /// Every thread this crate starts publishes its name to the OS.
4155    ///
4156    /// `std` calls the platform `pthread_setname_np` from `Builder::spawn`
4157    /// on the hosted targets it supports, and RTEMS is not one of them — so
4158    /// there, a name set with `Builder::name` lives only in Rust's `Thread`
4159    /// struct and the kernel shows nothing. Bring-up had to measure libbsd's
4160    /// priority band by other means for exactly that reason.
4161    ///
4162    /// The defect is a call that is *absent*, so this is source inspection
4163    /// over every production `Builder` site in the crate — the same sweep
4164    /// shape as `every_thread_in_this_crate_states_a_stack_size`, and it
4165    /// fails the same way when a new thread forgets. Either prologue counts:
4166    /// `enter_ioc_thread` for a thread with an EPICS band, bare
4167    /// `name_current_thread` for one that deliberately has none.
4168    #[test]
4169    fn every_thread_in_this_crate_publishes_its_name() {
4170        // The one exemption, named rather than pattern-matched: the
4171        // SCHED_FIFO range probe is `#[cfg(target_os = "linux")]`, exists for
4172        // two `sched_*` calls and a join, and never runs on the target whose
4173        // task listing this guard is about.
4174        const EXEMPT: &str = ".name(\"cbRtProbe\".to_string())";
4175
4176        let mut anonymous = Vec::new();
4177        let mut checked = 0usize;
4178        for (label, src) in censused_files() {
4179            for (n, after) in production_scope(src)
4180                .split("thread::Builder::new()")
4181                .skip(1)
4182                .enumerate()
4183            {
4184                let (chain, body) = after.split_once(".spawn(").unwrap_or((after, ""));
4185                if chain.contains(EXEMPT) {
4186                    continue;
4187                }
4188                checked += 1;
4189                // The prologue is the closure's first work, so look at the
4190                // closure, not the builder chain.
4191                if !body.contains("enter_ioc_thread(") && !body.contains("name_current_thread()") {
4192                    anonymous.push(format!("{label} (Builder #{})", n + 1));
4193                }
4194            }
4195        }
4196
4197        // Three: `spawn_dedicated_thread`'s two `cfg` arms and
4198        // `MandatoryThread::try_spawn`, all of which run the prologue for the
4199        // caller. The floor was five until the three background facilities
4200        // moved onto `MandatoryThread`, whose constructor takes the band —
4201        // `every_background_facility_thread_is_mandatory` covers those files.
4202        assert!(
4203            checked >= 3,
4204            "expected to find the crate's Builder sites, found {checked} — \
4205             did a file move? update this guard's file list"
4206        );
4207        assert!(
4208            anonymous.is_empty(),
4209            "these threads are invisible in an RTEMS task listing: {anonymous:?}"
4210        );
4211    }
4212
4213    /// The banding half of the prologue is not reachable without the naming
4214    /// half: nothing in this crate's production scope calls
4215    /// [`apply_to_current_thread`] except [`enter_ioc_thread`] itself.
4216    ///
4217    /// Separate from the sweep above because it catches the other direction —
4218    /// a thread that is named by `Builder` but takes its band directly, which
4219    /// the closure-body sweep would pass if the naming call happened to be
4220    /// somewhere else in the file.
4221    #[test]
4222    fn only_the_prologue_reaches_the_banding_call() {
4223        // Only the definition and the prologue's own delegation, both in
4224        // task.rs. Anywhere else is a thread banded without being named.
4225        let allowed = [
4226            "pub fn apply_to_current_thread(priority: ThreadPriority) -> PriorityApplied {",
4227            "let applied = apply_to_current_thread(priority);",
4228        ];
4229        let mut seen_definition = false;
4230        // Every file in the crate, not the thread-creating ones: a call that
4231        // bands a thread is a defect wherever it is written, and this guard's
4232        // subject is the absence of such a call.
4233        for (label, src) in crate_sources() {
4234            let callers: Vec<&str> = production_scope(src)
4235                .lines()
4236                .map(str::trim)
4237                .filter(|l| l.contains("apply_to_current_thread("))
4238                .filter(|l| !l.starts_with("//"))
4239                .collect();
4240            seen_definition |= callers.contains(&allowed[0]);
4241            let strays: Vec<&&str> = callers.iter().filter(|l| !allowed.contains(l)).collect();
4242            assert!(
4243                strays.is_empty(),
4244                "{label}: only `enter_ioc_thread` may band a thread; \
4245                 everything else would band an OS-anonymous one — {strays:?}"
4246            );
4247        }
4248        assert!(
4249            seen_definition,
4250            "the banding function moved out of this file list; update the guard"
4251        );
4252    }
4253
4254    /// The prologue must also announce the thread to the statistics funnel's
4255    /// census, and that call has to be checked as text because nothing else can
4256    /// check it: it is `#[cfg]`ed to VxWorks, so on the host and on RTEMS it
4257    /// compiles away and deleting it breaks no build and no test. What it would
4258    /// break is one target's task census, which would come back empty — an IOC
4259    /// that reads as having no threads rather than as having a missing call.
4260    ///
4261    /// Located inside the prologue's own body rather than anywhere in the file,
4262    /// because a registration that drifted out of the single thread-transition
4263    /// owner is the same defect as no registration: threads would start without
4264    /// passing it.
4265    #[test]
4266    fn the_prologue_registers_the_thread_for_the_vxworks_census() {
4267        let body = production_scope(include_str!("task.rs"))
4268            .split_once("pub fn enter_ioc_thread(")
4269            .expect("the prologue is still in this file")
4270            .1
4271            .split_once("\n}\n")
4272            .expect("the prologue's body is terminated")
4273            .0;
4274        assert!(
4275            body.contains("#[cfg(target_os = \"vxworks\")]"),
4276            "the census registration must stay gated to the one OS whose \
4277             backend needs it; `epics-rtems-boot` is a dependency of this \
4278             package on that target only"
4279        );
4280        assert!(
4281            body.contains("epics_rtems_boot::stats::register_task();"),
4282            "VxWorks gives an RTP no task enumerator, so `dump_tasks` and \
4283             `stack_report` list exactly what announced itself here"
4284        );
4285        assert!(
4286            body.contains("thread_registry::register_current(priority, applied);"),
4287            "`epicsThreadShowAll` prints exactly what the prologue registered; \
4288             a registration outside the prologue is a thread that can start \
4289             without one"
4290        );
4291    }
4292
4293    /// Both of C's `isOkToBlock` defaults, and they differ by exactly one
4294    /// thing: whether the thread passed the prologue.
4295    ///
4296    /// C sets 1 in `createImplicit` (`osdThread.c:710`) for a thread that
4297    /// reaches the epicsThread API without having been created by it, and
4298    /// leaves the `calloc`ed 0 for every `epicsThreadCreate` thread. A test
4299    /// thread is the first kind; anything spawned through this module is the
4300    /// second. `runtime::log`'s errlog reads it to decide whether a producer
4301    /// may wait for the log drain, so getting this backwards either stalls a
4302    /// scan thread on the log or lets the log's own worker wait for itself.
4303    #[test]
4304    fn only_a_thread_that_ran_the_prologue_is_not_ok_to_block() {
4305        assert!(
4306            thread_is_ok_to_block(),
4307            "a thread this module did not create is C's implicit context"
4308        );
4309        let banded = std::thread::spawn(|| {
4310            let _ = enter_ioc_thread(ThreadPriority::ScanLow);
4311            let after_prologue = thread_is_ok_to_block();
4312            set_thread_ok_to_block(true);
4313            (after_prologue, thread_is_ok_to_block())
4314        })
4315        .join()
4316        .expect("the banded thread");
4317        assert_eq!(banded, (false, true));
4318
4319        // The one band that is a shell, so the prologue does what C's
4320        // `iocsh`/`iocInit` do to their own thread rather than leaving it at
4321        // the `epicsThreadCreate` default.
4322        let shell = std::thread::spawn(|| {
4323            let _ = enter_ioc_thread(ThreadPriority::Iocsh);
4324            thread_is_ok_to_block()
4325        })
4326        .join()
4327        .expect("the shell thread");
4328        assert!(shell, "an iocsh thread boots a database and may block");
4329        assert!(
4330            thread_is_ok_to_block(),
4331            "the flag is per-thread: the child's prologue must not clear ours"
4332        );
4333    }
4334
4335    /// The list holds a thread for exactly as long as the thread exists: C's
4336    /// `ellAdd` in `start_routine` and `ellDelete` in the thread-specific-data
4337    /// destructor. A row that outlived its thread would be a `epicsThreadShow`
4338    /// listing of stacks that are gone.
4339    #[test]
4340    fn a_row_lasts_exactly_as_long_as_its_thread() {
4341        let name = "cbRegLife";
4342        assert!(thread_by_name(name).is_none(), "name must start unused");
4343
4344        let (started_tx, started_rx) = std::sync::mpsc::channel();
4345        let (finish_tx, finish_rx) = std::sync::mpsc::channel();
4346        let join = std::thread::Builder::new()
4347            .name(name.to_string())
4348            .spawn(move || {
4349                let _ = enter_ioc_thread(ThreadPriority::ScanLow);
4350                started_tx.send(()).unwrap();
4351                finish_rx.recv().unwrap();
4352            })
4353            .unwrap();
4354
4355        started_rx.recv().unwrap();
4356        let listed = thread_by_name(name).expect("listed while running");
4357        assert_eq!(listed.epics_priority(), ThreadPriority::ScanLow.value());
4358        assert_eq!(listed.name(), name);
4359        assert_ne!(listed.id(), 0, "every row carries a usable EPICS ID");
4360
4361        finish_tx.send(()).unwrap();
4362        join.join().unwrap();
4363        assert!(
4364            thread_by_name(name).is_none(),
4365            "the row must be reaped with the thread, not left for the next reader"
4366        );
4367    }
4368
4369    /// C's `once()` row, all four of its boundaries in one test because they
4370    /// share one process-wide guard: `pthread_once` fires for the first caller
4371    /// only, so a second test calling [`register_main_thread`] would be
4372    /// asserting against the first test's row rather than its own.
4373    ///
4374    /// Run on a bare `std::thread::spawn` because that is the shape the shell
4375    /// threads have — no `Builder::name`, so `std::thread::current().name()`
4376    /// is `None` and the row's name can only come from this call.
4377    #[test]
4378    fn the_main_row_is_cs_once_row() {
4379        let main_rows = || {
4380            thread_report()
4381                .into_iter()
4382                .filter(|t| t.name() == "_main_")
4383                .count()
4384        };
4385        assert_eq!(main_rows(), 0, "nothing registers `_main_` on its own");
4386
4387        let before = sched_calls_made();
4388        let (registered_tx, registered_rx) = std::sync::mpsc::channel();
4389        let (finish_tx, finish_rx) = std::sync::mpsc::channel();
4390        let join = std::thread::spawn(move || {
4391            register_main_thread();
4392            registered_tx.send(()).unwrap();
4393            finish_rx.recv().unwrap();
4394            // C's guard is per process, not per thread: the second call adds
4395            // nothing.
4396            register_main_thread();
4397            registered_tx.send(()).unwrap();
4398            finish_rx.recv().unwrap();
4399        });
4400
4401        registered_rx.recv().unwrap();
4402        let row = thread_by_name("_main_").expect("listed while the shell runs");
4403        assert_eq!(row.epics_priority(), 0, "C's `_main_` is EPICS priority 0");
4404        assert_eq!(row.os_priority(), 0, "C changes no scheduling for `_main_`");
4405        assert_eq!(
4406            sched_calls_made(),
4407            before,
4408            "registering `_main_` must ask the scheduler for nothing"
4409        );
4410        assert_eq!(main_rows(), 1);
4411
4412        finish_tx.send(()).unwrap();
4413        registered_rx.recv().unwrap();
4414        assert_eq!(main_rows(), 1, "one process, one `_main_`");
4415
4416        finish_tx.send(()).unwrap();
4417        join.join().unwrap();
4418        assert_eq!(
4419            main_rows(),
4420            0,
4421            "the row is reaped where C's thread-specific-data destructor \
4422             removes it — when the thread holding it ends"
4423        );
4424    }
4425
4426    /// Re-banding a thread is one thread, so it is one row. The boundary the
4427    /// `thread_local` guard exists for: storing the new registration drops the
4428    /// old one, which is what removes the superseded row.
4429    #[test]
4430    fn re_entering_the_prologue_replaces_the_row_rather_than_adding_one() {
4431        let name = "cbRegReband";
4432        let (started_tx, started_rx) = std::sync::mpsc::channel();
4433        let (finish_tx, finish_rx) = std::sync::mpsc::channel();
4434        let join = std::thread::Builder::new()
4435            .name(name.to_string())
4436            .spawn(move || {
4437                let _ = enter_ioc_thread(ThreadPriority::Low);
4438                let _ = enter_ioc_thread(ThreadPriority::ScanHigh);
4439                started_tx.send(()).unwrap();
4440                finish_rx.recv().unwrap();
4441            })
4442            .unwrap();
4443
4444        started_rx.recv().unwrap();
4445        let rows: Vec<ThreadInfo> = thread_report()
4446            .into_iter()
4447            .filter(|t| t.name() == name)
4448            .collect();
4449        assert_eq!(rows.len(), 1, "one thread is one row: {rows:?}");
4450        assert_eq!(rows[0].epics_priority(), ThreadPriority::ScanHigh.value());
4451
4452        finish_tx.send(()).unwrap();
4453        join.join().unwrap();
4454    }
4455
4456    /// Both handles a shell user can read off the listing resolve, and a
4457    /// number that is neither resolves to nothing rather than to the first row.
4458    #[test]
4459    fn a_row_resolves_by_epics_id_and_by_os_id() {
4460        let name = "cbRegLookup";
4461        let (started_tx, started_rx) = std::sync::mpsc::channel();
4462        let (finish_tx, finish_rx) = std::sync::mpsc::channel();
4463        let join = std::thread::Builder::new()
4464            .name(name.to_string())
4465            .spawn(move || {
4466                let _ = enter_ioc_thread(ThreadPriority::Medium);
4467                started_tx.send(()).unwrap();
4468                finish_rx.recv().unwrap();
4469            })
4470            .unwrap();
4471
4472        started_rx.recv().unwrap();
4473        let listed = thread_by_name(name).expect("listed while running");
4474        assert_eq!(thread_by_id(listed.id()).map(|t| t.id()), Some(listed.id()));
4475        assert_eq!(
4476            thread_by_id(listed.os_id()).map(|t| t.id()),
4477            Some(listed.id())
4478        );
4479        assert!(thread_by_id(u64::MAX).is_none());
4480
4481        finish_tx.send(()).unwrap();
4482        join.join().unwrap();
4483    }
4484
4485    /// The `OSSPRI` boundary: a thread that never entered the real-time band
4486    /// reads 0, which is the `sched_priority` a SCHED_OTHER thread has and
4487    /// what C's live `pthread_getschedparam` returns for it. Every row on a
4488    /// default hosted IOC is this case.
4489    #[test]
4490    fn os_priority_is_zero_for_every_outcome_but_realtime() {
4491        for applied in [
4492            PriorityApplied::Disabled,
4493            PriorityApplied::Unsupported,
4494            PriorityApplied::BestEffortFailed,
4495        ] {
4496            let row = ThreadInfo {
4497                id: 1,
4498                name: "cbOss".to_string(),
4499                os_id: 7,
4500                priority: ThreadPriority::High,
4501                applied,
4502                suspension: Arc::default(),
4503            };
4504            assert_eq!(row.os_priority(), 0, "{applied:?}");
4505            assert_eq!(row.epics_priority(), ThreadPriority::High.value());
4506        }
4507    }
4508
4509    /// C's `%16.16s %14p %8lu    %3d%8d %8.8s%s`, against two rows copied from
4510    /// a running `softIoc` built from the pinned C tree. The EPICS ID column
4511    /// holds C's `epicsThreadOSD` addresses so the widths are the ones C
4512    /// actually produced, not ones chosen to fit.
4513    #[cfg(target_os = "linux")]
4514    #[test]
4515    fn show_line_matches_c_s_columns_byte_for_byte() {
4516        let main = ThreadInfo {
4517            id: 0x604f_304e_f4f0,
4518            name: "_main_".to_string(),
4519            os_id: 3_650_070,
4520            priority: ThreadPriority::Custom(0),
4521            applied: PriorityApplied::Disabled,
4522            suspension: Arc::default(),
4523        };
4524        assert_eq!(
4525            main.show_line(),
4526            "          _main_ 0x604f304ef4f0  3650070      0       0       OK"
4527        );
4528
4529        let errlog = ThreadInfo {
4530            id: 0x604f_304f_8a90,
4531            name: "errlog".to_string(),
4532            os_id: 3_650_072,
4533            priority: ThreadPriority::Low,
4534            applied: PriorityApplied::Disabled,
4535            suspension: Arc::default(),
4536        };
4537        assert_eq!(
4538            errlog.show_line(),
4539            "          errlog 0x604f304f8a90  3650072     10       0       OK"
4540        );
4541
4542        assert_eq!(
4543            THREAD_SHOW_HEADER,
4544            "            NAME       EPICS ID   LWP ID   OSIPRI  OSSPRI  STATE"
4545        );
4546    }
4547
4548    /// The other `osdThreadExtra.c`: a wider OS id column under a different
4549    /// name. Written out rather than derived so the two layouts cannot drift
4550    /// into each other.
4551    #[cfg(not(target_os = "linux"))]
4552    #[test]
4553    fn show_line_matches_c_s_generic_posix_columns() {
4554        let row = ThreadInfo {
4555            id: 0x604f_304e_f4f0,
4556            name: "_main_".to_string(),
4557            os_id: 3_650_070,
4558            priority: ThreadPriority::Custom(0),
4559            applied: PriorityApplied::Disabled,
4560            suspension: Arc::default(),
4561        };
4562        assert_eq!(
4563            row.show_line(),
4564            "          _main_ 0x604f304ef4f0      3650070      0       0       OK"
4565        );
4566        assert_eq!(
4567            THREAD_SHOW_HEADER,
4568            "            NAME       EPICS ID   PTHREAD ID   OSIPRI  OSSPRI  STATE"
4569        );
4570    }
4571
4572    /// The suspended state end to end, and the boundary a naive latch gets
4573    /// wrong.
4574    ///
4575    /// C's `epicsThreadResume` signals a latching event, so a resume with
4576    /// nobody suspended banks a token that would skip the next park. C's own
4577    /// iocsh refuses before it can (`libComRegister.c:445-449`) and `dbc`
4578    /// only reaches it for a lock set that is stopped; this port makes the
4579    /// refusal the primitive, so the two resumes below must leave nothing
4580    /// behind and the park after them must still hold.
4581    #[test]
4582    fn a_suspended_thread_reads_suspend_and_only_a_resume_wakes_it() {
4583        let (id_tx, id_rx) = std::sync::mpsc::channel();
4584        let (go_tx, go_rx) = std::sync::mpsc::channel::<()>();
4585        let (woke_tx, woke_rx) = std::sync::mpsc::channel();
4586        let thread = std::thread::Builder::new()
4587            .name("suspendTarget".to_string())
4588            .spawn(move || {
4589                let _ = enter_ioc_thread(ThreadPriority::ScanLow);
4590                id_tx.send(current_thread_id()).unwrap();
4591                go_rx.recv().unwrap();
4592                suspend_self();
4593                woke_tx.send(()).unwrap();
4594            })
4595            .unwrap();
4596        let id = id_rx.recv().unwrap();
4597
4598        // Nothing is suspended yet: both resumes report the refusal and,
4599        // crucially, bank nothing.
4600        assert!(!resume_thread(id));
4601        assert!(!resume_thread(id));
4602        assert!(!thread_by_id(id).unwrap().is_suspended());
4603        assert!(thread_by_id(id).unwrap().show_line().ends_with("      OK"));
4604
4605        go_tx.send(()).unwrap();
4606        let row = (0..500)
4607            .find_map(|_| {
4608                let row = thread_by_id(id).expect("row");
4609                row.is_suspended().then_some(row).or_else(|| {
4610                    std::thread::sleep(Duration::from_millis(10));
4611                    None
4612                })
4613            })
4614            .expect("the thread must reach suspend_self");
4615        assert!(
4616            row.show_line().ends_with(" SUSPEND"),
4617            "C prints SUSPEND for isSuspended (osdThreadExtra.c:53), got {:?}",
4618            row.show_line()
4619        );
4620        assert_eq!(
4621            woke_rx.recv_timeout(Duration::from_millis(250)),
4622            Err(std::sync::mpsc::RecvTimeoutError::Timeout),
4623            "the two earlier resumes must not have banked a wake-up"
4624        );
4625
4626        assert!(
4627            resume_thread(id),
4628            "C epicsThreadResume on a suspended thread"
4629        );
4630        woke_rx
4631            .recv_timeout(Duration::from_secs(5))
4632            .expect("resumed");
4633        thread.join().unwrap();
4634    }
4635
4636    /// C `createImplicit` (`osdThread.c:697-735`): a thread that never went
4637    /// through `epicsThreadCreate` is given a `non-EPICS_` row the first time
4638    /// its handle is asked for, so it can be listed and resumed rather than
4639    /// being suspended somewhere nothing can name.
4640    #[test]
4641    fn a_thread_that_never_registered_gets_c_s_implicit_row() {
4642        let (id, name, listed) = std::thread::spawn(|| {
4643            let id = current_thread_id();
4644            let row = thread_by_id(id).expect("createImplicit must put it on the list");
4645            (id, row.name().to_string(), row.epics_priority())
4646        })
4647        .join()
4648        .unwrap();
4649        assert!(name.starts_with("non-EPICS_"), "C's name, got {name:?}");
4650        assert_eq!(listed, 0, "C gives the implicit row osiPriority 0");
4651        assert!(
4652            thread_by_id(id).is_none(),
4653            "the implicit row is reaped with the thread, as C frees it"
4654        );
4655    }
4656
4657    /// A name longer than the column is cut, not allowed to shift every field
4658    /// after it — C's `%16.16s` precision, which a plain width would lose.
4659    #[test]
4660    fn a_long_name_is_truncated_rather_than_widening_the_row() {
4661        let row = ThreadInfo {
4662            id: 1,
4663            name: "aVeryLongThreadNameIndeed".to_string(),
4664            os_id: 7,
4665            priority: ThreadPriority::Low,
4666            applied: PriorityApplied::Disabled,
4667            suspension: Arc::default(),
4668        };
4669        let short = ThreadInfo {
4670            name: "s".to_string(),
4671            ..row.clone()
4672        };
4673        let line = row.show_line();
4674        assert!(line.starts_with("aVeryLongThreadN "), "{line}");
4675        assert_eq!(
4676            line.len(),
4677            short.show_line().len(),
4678            "a row's width must not depend on its name"
4679        );
4680    }
4681
4682    /// The stderr trailer C prints after the listing. With the real-time
4683    /// switch off the range is genuinely a single point, and no probe may run
4684    /// to discover a wider one.
4685    #[test]
4686    fn the_show_all_trailer_reports_the_range_this_process_may_enter() {
4687        let line = osd_priority_range_line();
4688        assert!(line.ends_with(", memory not locked"), "{line}");
4689        if RtPolicy::current() == RtPolicy::Disabled {
4690            assert_eq!(line, "OSD priority range min: 0 max 0, memory not locked");
4691        }
4692    }
4693
4694    /// The owner path: a mandatory thread that *can* be created runs its body
4695    /// under the name and band it was declared with.
4696    #[test]
4697    fn a_mandatory_thread_runs_under_its_declared_name() {
4698        let (tx, rx) = std::sync::mpsc::channel();
4699        let join = MandatoryThread::new(
4700            "cbTestOwner",
4701            ThreadPriority::ScanLow,
4702            StackSizeClass::Small,
4703        )
4704        .spawn(move || {
4705            let _ = tx.send(
4706                std::thread::current()
4707                    .name()
4708                    .map(str::to_owned)
4709                    .unwrap_or_default(),
4710            );
4711        });
4712        assert_eq!(rx.recv().expect("the body ran"), "cbTestOwner");
4713        join.join().expect("the thread exited cleanly");
4714    }
4715
4716    /// `try_spawn` is the same construction with the failure handed back, so a
4717    /// caller inside a fallible boot step can refuse to serve.
4718    #[test]
4719    fn try_spawn_hands_back_a_handle_on_success() {
4720        let (tx, rx) = std::sync::mpsc::channel();
4721        let join =
4722            MandatoryThread::new("cbTestTry", ThreadPriority::ScanLow, StackSizeClass::Small)
4723                .try_spawn(move || {
4724                    let _ = tx.send(());
4725                })
4726                .expect("a thread is creatable in the test environment");
4727        rx.recv().expect("the body ran");
4728        join.join().expect("the thread exited cleanly");
4729    }
4730
4731    /// The console line names the thread and what the OS said, so an operator
4732    /// reading a target console can tell *which* thread the IOC died for.
4733    ///
4734    /// `EAGAIN` cannot be forced portably — the failure shape is the subject
4735    /// here, not the syscall.
4736    #[test]
4737    fn the_fatal_message_names_the_thread_and_the_error() {
4738        let msg = mandatory_thread_failure_message(
4739            "scan-0.1",
4740            &std::io::Error::from(std::io::ErrorKind::WouldBlock),
4741        );
4742        assert!(msg.contains("scan-0.1"), "{msg}");
4743        assert!(msg.contains("FATAL"), "{msg}");
4744        assert!(
4745            msg.contains(&std::io::Error::from(std::io::ErrorKind::WouldBlock).to_string()),
4746            "{msg}"
4747        );
4748    }
4749
4750    /// The bypass regression: a mandatory thread that cannot be created must
4751    /// take the **process** down, not the calling thread.
4752    ///
4753    /// The defect this closes was measured on a VxWorks 7 RTP: `EAGAIN` from
4754    /// the periodic-scan spawn panicked the `scan-owner` thread, and because
4755    /// both RTEMS and VxWorks default to `panic = "unwind"`, the process
4756    /// survived and went on serving CA with no periodic scanning at all. A test
4757    /// that only asserted "it panics" would have passed against that defect —
4758    /// so this one re-executes itself and asserts the *process* died.
4759    ///
4760    /// Gated off the embedded targets: they have no process to spawn.
4761    #[cfg(all(unix, not(target_os = "rtems"), not(target_os = "vxworks")))]
4762    #[test]
4763    fn a_mandatory_thread_that_cannot_be_created_aborts_the_process() {
4764        use std::os::unix::process::ExitStatusExt;
4765
4766        const CHILD: &str = "EPICS_RS_MANDATORY_THREAD_ABORT_CHILD";
4767        const TEST: &str =
4768            "runtime::task::tests::a_mandatory_thread_that_cannot_be_created_aborts_the_process";
4769
4770        if std::env::var_os(CHILD).is_some() {
4771            mandatory_thread_unavailable(
4772                "scan-0.1",
4773                &std::io::Error::from(std::io::ErrorKind::WouldBlock),
4774            );
4775        }
4776
4777        let out =
4778            std::process::Command::new(std::env::current_exe().expect("the test binary path"))
4779                .args(["--exact", TEST, "--nocapture"])
4780                .env(CHILD, "1")
4781                .output()
4782                .expect("re-exec the test binary");
4783
4784        assert_eq!(
4785            out.status.signal(),
4786            Some(libc::SIGABRT),
4787            "a mandatory thread's failure must abort the process, not unwind \
4788             one thread — child exited {:?}, stderr: {}",
4789            out.status,
4790            String::from_utf8_lossy(&out.stderr)
4791        );
4792        let stderr = String::from_utf8_lossy(&out.stderr);
4793        assert!(
4794            stderr.contains("scan-0.1"),
4795            "the console must name the thread; got: {stderr}"
4796        );
4797    }
4798
4799    #[epics_macros_rs::epics_test]
4800    async fn spawn_blocking_with_priority_runs_closure() {
4801        let handle = spawn_blocking_with_priority(ThreadPriority::CaServerHigh, || 7);
4802        assert_eq!(handle.await.unwrap(), 7);
4803    }
4804
4805    /// C `ca_client_context::pendIO` compares the remaining budget *before* it
4806    /// blocks (`ca_client_context.cpp:490-499`: `remaining <
4807    /// CAC_SIGNIFICANT_DELAY` → `ECA_TIMEOUT`, `break`), so an already-spent
4808    /// budget cannot be beaten by work that lands a moment later.
4809    /// `tokio::time::timeout` could: it polls the inner future, arms a
4810    /// zero-length `Sleep`, and that `Sleep` is not reported elapsed until the
4811    /// time driver next runs — one `yield_now` is enough for the inner future
4812    /// to win, which is how `caget -w -1` exited 0 where C exits 1.
4813    ///
4814    /// A future that is `Pending` on its first poll and `Ready` on its second
4815    /// is the whole race in one value, and it needs no load to show it.
4816    #[epics_macros_rs::epics_test]
4817    async fn an_expired_budget_is_not_beaten_by_work_that_needs_a_second_poll() {
4818        let r = timeout(Duration::ZERO, async {
4819            yield_now().await;
4820            7u8
4821        })
4822        .await;
4823        assert!(
4824            r.is_err(),
4825            "a spent budget must expire before work that was not already done"
4826        );
4827    }
4828
4829    /// The other half of C's order: `pendIO` tests `pndRecvCnt > 0` first, so a
4830    /// request with nothing outstanding returns `ECA_NORMAL` even on an expired
4831    /// budget. One poll of the future is that same test.
4832    #[epics_macros_rs::epics_test]
4833    async fn an_expired_budget_still_takes_work_that_is_already_done() {
4834        let r = timeout(Duration::ZERO, std::future::ready(7u8)).await;
4835        assert_eq!(r.ok(), Some(7));
4836    }
4837
4838    /// [`timeout_at`] carries the same rule against an absolute deadline: a
4839    /// loop that re-uses one deadline reaches it already past, which is the
4840    /// `pvlist-rs` receive loop's shape.
4841    #[epics_macros_rs::epics_test]
4842    async fn a_deadline_already_in_the_past_expires_without_racing_the_work() {
4843        let past = Instant::now() - Duration::from_secs(1);
4844        let r = timeout_at(past, async {
4845            yield_now().await;
4846            7u8
4847        })
4848        .await;
4849        assert!(r.is_err(), "a past deadline must not be raced either");
4850    }
4851
4852    #[test]
4853    fn background_global_inits_and_runs_work() {
4854        // Host-exercises the OnceLock init path the RTEMS spawn/sleep/interval
4855        // arms rely on: background_init() forces creation, background() hands
4856        // back a usable executor whose callback pool runs submitted work.
4857        background_init();
4858        let exec = background();
4859        let (tx, rx) = std::sync::mpsc::channel();
4860        exec.callbacks()
4861            .handle()
4862            .request(
4863                crate::runtime::background::CallbackPriority::Medium,
4864                Box::new(move || tx.send(1u8).unwrap()),
4865            )
4866            .unwrap();
4867        assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), 1);
4868    }
4869}