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// feature-ON 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
14/// A synchronous caller asked to block on an async operation from a thread
15/// where blocking cannot be made sound.
16///
17/// Both variants are the same defect seen through two executors: the calling
18/// thread is one the awaited future needs in order to make progress, so parking
19/// it parks the thing that would wake it. No blocking mechanism can fix that;
20/// the caller has to `await` the async operation instead of blocking on it.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub enum NotBlockable {
23    /// A **current-thread** tokio runtime is entered on this thread. Parking it
24    /// stops every task on that runtime, including whichever one holds the
25    /// state the awaited future is waiting for.
26    CurrentThreadRuntime,
27    /// This thread is a background-facility worker — a callback band, the
28    /// delayed-callback timer, or the scanOnce worker
29    /// ([`crate::runtime::background`]). Each facility has a bounded worker set
30    /// and every unit of work it carries is enqueued for those workers, so a
31    /// parked worker is waiting for work only it could have run. On RTEMS
32    /// [`spawn`] routes here, which makes the callback bands the one other
33    /// place where parking is unsound.
34    BackgroundWorker,
35}
36
37impl std::fmt::Display for NotBlockable {
38    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
39        match self {
40            NotBlockable::CurrentThreadRuntime => {
41                f.write_str("cannot block a current-thread runtime")
42            }
43            NotBlockable::BackgroundWorker => {
44                f.write_str("cannot block a background-facility worker thread")
45            }
46        }
47    }
48}
49
50impl std::error::Error for NotBlockable {}
51
52/// A [`Waker`] that unparks the thread that built it. The single owner of the
53/// "poll-then-park" wake mechanism in this crate: both [`park_on`] (the sync
54/// bridge) and the RTEMS future executor
55/// ([`crate::runtime::background::future_exec`]) drive a future by polling on a
56/// thread and parking it between polls, so both build one of these on their own
57/// thread and rely on the future's cross-thread waker to unpark them.
58pub(crate) struct ThreadWaker(std::thread::Thread);
59
60impl ThreadWaker {
61    /// A waker over the *current* thread — call this on the thread that will
62    /// park.
63    pub(crate) fn for_current_thread() -> Waker {
64        Waker::from(Arc::new(ThreadWaker(std::thread::current())))
65    }
66}
67
68impl Wake for ThreadWaker {
69    fn wake(self: Arc<Self>) {
70        self.0.unpark();
71    }
72    fn wake_by_ref(self: &Arc<Self>) {
73        self.0.unpark();
74    }
75}
76
77/// Drive `fut` to completion on this thread, parking between polls, and stop
78/// early when `should_cancel` returns `true`.
79///
80/// Returns `Some(output)` when the future completed, or `None` when it was
81/// cancelled before completing (the future is dropped in place on cancel,
82/// running its destructors — the same "drop at the next suspension point"
83/// semantics a cancelled tokio task has).
84///
85/// The future must only await runtime-agnostic primitives (`tokio::sync`
86/// locks/channels/notifies): nothing here drives a reactor or a timer wheel, so
87/// whoever wakes us must be running on some other thread. A cancel is observed
88/// on the next wake — the caller that flips `should_cancel` must also
89/// [`unpark`](std::thread::Thread::unpark) this thread so a *parked* driver
90/// re-checks promptly rather than sleeping until the future's own waker fires.
91pub(crate) fn park_on_interruptible<F: Future>(
92    fut: F,
93    mut should_cancel: impl FnMut() -> bool,
94) -> Option<F::Output> {
95    let mut fut = std::pin::pin!(fut);
96    let waker = ThreadWaker::for_current_thread();
97    let mut cx = Context::from_waker(&waker);
98    loop {
99        if should_cancel() {
100            return None;
101        }
102        if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) {
103            return Some(value);
104        }
105        std::thread::park();
106    }
107}
108
109/// Drive `fut` to completion on this thread, parking between polls. Thin
110/// uncancellable wrapper over [`park_on_interruptible`].
111///
112/// The future must only await runtime-agnostic primitives (`tokio::sync`
113/// locks/channels/notifies): nothing here drives a reactor or a timer wheel, so
114/// whoever wakes us must be running on some other thread.
115fn park_on<F: Future>(fut: F) -> F::Output {
116    // Never cancels, so `park_on_interruptible` always returns `Some`.
117    park_on_interruptible(fut, || false).expect("uncancellable driver returned None")
118}
119
120/// Block the calling thread on `fut`, picking the mechanism that is sound for
121/// the thread we are actually on.
122///
123/// This is the single owner of "sync call over async state" in this crate; the
124/// four caller contexts are not interchangeable and picking one mechanism for
125/// all of them is what makes such bridges panic:
126///
127/// - **A background-facility worker** —
128///   [`Err(BackgroundWorker)`](NotBlockable::BackgroundWorker), checked first,
129///   because it is a property of the *thread* and holds whatever runtime is or
130///   is not entered on it. See
131///   [`background::facility::on_facility_thread`](crate::runtime::background)
132///   for why parking one is unsound.
133/// - **No runtime entered** (a plain `std::thread`, an iocsh thread) — park the
134///   thread. Nothing else runs here, so there is nothing to starve; the tasks
135///   that will wake us live on some other runtime's threads.
136/// - **Multi-thread runtime worker** — [`tokio::task::block_in_place`], which
137///   hands this worker's remaining tasks to a sibling before it is parked.
138/// - **Current-thread runtime** —
139///   [`Err(CurrentThreadRuntime)`](NotBlockable::CurrentThreadRuntime). Parking
140///   the only thread of that runtime halts every task on it, including the one
141///   that would wake us.
142///
143/// The two refusals are reported to the caller rather than panicked on (today)
144/// or deadlocked on (the worse alternative) — an illegal blocking bridge is a
145/// value the caller must handle, not a review item.
146pub fn block_on_sync<F: Future>(fut: F) -> Result<F::Output, NotBlockable> {
147    if crate::runtime::background::facility::on_facility_thread() {
148        return Err(NotBlockable::BackgroundWorker);
149    }
150    match RuntimeHandle::try_current() {
151        Ok(handle) => match handle.runtime_flavor() {
152            RuntimeFlavor::CurrentThread => Err(NotBlockable::CurrentThreadRuntime),
153            _ => Ok(tokio::task::block_in_place(|| handle.block_on(fut))),
154        },
155        Err(_) => Ok(park_on(fut)),
156    }
157}
158
159/// A capability, captured where the backend's executor is reachable, to run
160/// async work from a plain blocking thread (iocsh, a REPL, a script thread).
161///
162/// [`block_on_sync`] answers "may I block *here*, now?" per call and can only
163/// use whatever runtime is visible on the calling thread. This type answers
164/// the reachability question once, at [`capture`](Self::capture) time, and
165/// carries the answer to a thread the runtime is otherwise invisible from: a
166/// tokio handle is thread-local state, so a blocking thread spawned *before*
167/// it exists has no way to find it. The exec backend's executor is
168/// process-global, so there is nothing to carry and the bridge is a ZST —
169/// which is what makes an API taking a `BlockingBridge` compile and work on
170/// both backends, where one taking `tokio::runtime::Handle` pinned every
171/// caller to tokio.
172#[cfg(tokio_backend)]
173#[derive(Clone)]
174pub struct BlockingBridge {
175    handle: tokio::runtime::Handle,
176}
177
178/// See the `tokio_backend` definition. The executor here is the
179/// process-global background executor, reachable from any thread, so there is
180/// no state to capture.
181#[cfg(exec_backend)]
182#[derive(Clone)]
183pub struct BlockingBridge;
184
185#[cfg(tokio_backend)]
186impl BlockingBridge {
187    /// Capture the current tokio runtime.
188    ///
189    /// # Panics
190    /// Panics when no runtime is entered on this thread — call it on the
191    /// async setup path (where the runtime is known), not on the blocking
192    /// thread the bridge is being made for.
193    pub fn capture() -> Self {
194        Self {
195            handle: tokio::runtime::Handle::current(),
196        }
197    }
198
199    /// [`capture`](Self::capture) for a caller that has somewhere else to be
200    /// if no runtime is entered — `None` instead of a panic.
201    pub fn try_capture() -> Option<Self> {
202        RuntimeHandle::try_current()
203            .ok()
204            .map(|handle| Self { handle })
205    }
206
207    /// Drive `fut` to completion on this thread, with the captured runtime
208    /// entered so the future may spawn and use the reactor.
209    ///
210    /// # Panics
211    /// Panics on a runtime worker thread: blocking one parks tasks that may
212    /// include the future's own wakers (the same refusal `block_on_sync`
213    /// reports as a value).
214    pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
215        assert!(
216            RuntimeHandle::try_current().is_err(),
217            "BlockingBridge::block_on must not be called from a runtime thread"
218        );
219        self.handle.block_on(fut)
220    }
221
222    /// Spawn `future` onto the captured runtime — [`spawn`] for a thread the
223    /// runtime is not entered on.
224    pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
225    where
226        F: Future + Send + 'static,
227        F::Output: Send + 'static,
228    {
229        self.handle.spawn(future)
230    }
231}
232
233#[cfg(exec_backend)]
234impl BlockingBridge {
235    /// The exec backend's executor is process-global; capturing is a no-op
236    /// and never panics.
237    pub fn capture() -> Self {
238        Self
239    }
240
241    /// See the `tokio_backend` definition; capturing never fails here.
242    pub fn try_capture() -> Option<Self> {
243        Some(Self)
244    }
245
246    /// Drive `fut` on this thread via `park_on`; whatever it spawns or
247    /// sleeps on lands on the background executor.
248    pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
249        park_on(fut)
250    }
251
252    /// [`spawn`] — the global executor needs no captured state.
253    pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
254    where
255        F: Future + Send + 'static,
256        F::Output: Send + 'static,
257    {
258        spawn(future)
259    }
260}
261
262/// Drive an async test body to completion — the driver behind
263/// `#[epics_test]` (`epics-macros-rs`).
264///
265/// The point of the indirection is that the *backend* picks the driver, not
266/// the test. On `tokio_backend` this builds exactly what `#[tokio::test]`
267/// builds: a fresh current-thread runtime with IO and time enabled. On
268/// `exec_backend` (the RTEMS target, or a host run with
269/// `--features rtems-exec-model`) no tokio runtime exists to build, so the
270/// test thread itself drives the future via `park_on`, and everything the
271/// body spawns or sleeps on lands on the process-global background executor
272/// (lazily initialised on first use) — the same seam the RTEMS boot path
273/// exercises. A test written with `#[epics_test]` therefore needs no
274/// per-backend gating and no `RTEMS-EXEC-MODEL-ALLOW` census entry.
275#[cfg(tokio_backend)]
276pub fn test_block_on<F: Future>(fut: F) -> F::Output {
277    tokio::runtime::Builder::new_current_thread()
278        .enable_all()
279        .build()
280        .expect("failed to build tokio test runtime")
281        .block_on(fut)
282}
283
284/// `exec_backend` twin of [`test_block_on`]: see the `tokio_backend` copy for
285/// the contract. The body's awaits must reach only runtime-agnostic
286/// primitives (`park_on`'s rule) — a body that touches `tokio::net` or
287/// `tokio::time` directly belongs under `#[tokio::test]` with a backend gate
288/// instead.
289#[cfg(exec_backend)]
290pub fn test_block_on<F: Future>(fut: F) -> F::Output {
291    park_on(fut)
292}
293
294// ---------------------------------------------------------------------------
295// Platform-selected task handle types (decision A2 / B)
296//
297// The seam hands back one of these aliases from every spawn; call sites in this
298// crate name only the alias, never a tokio handle. Hosted = the tokio handle
299// types. RTEMS = the always-compiled, host-tested mirrors in
300// `background::future_exec` (`JoinFuture`/`AbortHandle`/`JoinError`), which
301// reproduce exactly the subset of the tokio surface the call sites use.
302// ---------------------------------------------------------------------------
303
304/// `true` when [`spawn`] lands the future on the tokio runtime, `false` when it
305/// lands on the reactor-free background executor (`exec_backend` — the RTEMS
306/// target, or a host build with `--features rtems-exec-model`).
307///
308/// # What this is for
309///
310/// It is the *exported* form of `build.rs`'s backend decision, and the reason
311/// it is exported is that a spawned future's access to a tokio **reactor** is
312/// decided here and consumed in other crates. A future handed to [`spawn`] on
313/// `exec_backend` runs on a callback-pool worker with no reactor entered, so
314/// every `tokio::net` socket it opens panics — *even in a process that has a
315/// tokio runtime somewhere else*, because the runtime is not entered on that
316/// worker.
317///
318/// `epics-ca-rs` and `epics-pva-rs` therefore have to make the same decision
319/// this crate makes, for their own compilation, and they make it in their own
320/// `build.rs` from the same two inputs (target OS, `rtems-exec-model` feature).
321/// That is three copies of one rule, so each of them pins the copy against this
322/// constant with a `const` assertion — a build where the two disagree (say,
323/// `epics-base-rs/rtems-exec-model` enabled without `epics-ca-rs`'s) fails to
324/// compile instead of panicking at boot.
325pub const HAS_TOKIO_REACTOR: bool = cfg!(tokio_backend);
326
327/// Handle to a spawned task — `await` for its result, `abort()` to cancel.
328#[cfg(tokio_backend)]
329pub type TaskHandle<T> = tokio::task::JoinHandle<T>;
330/// Detached cancellation handle for a spawned task.
331#[cfg(tokio_backend)]
332pub type TaskAbortHandle = tokio::task::AbortHandle;
333/// Error from awaiting a [`TaskHandle`] (cancelled or panicked).
334#[cfg(tokio_backend)]
335pub type TaskJoinError = tokio::task::JoinError;
336
337#[cfg(exec_backend)]
338pub type TaskHandle<T> = crate::runtime::background::future_exec::JoinFuture<T>;
339#[cfg(exec_backend)]
340pub type TaskAbortHandle = crate::runtime::background::future_exec::AbortHandle;
341#[cfg(exec_backend)]
342pub type TaskJoinError = crate::runtime::background::future_exec::JoinError;
343
344// ---------------------------------------------------------------------------
345// Process-global background executor (C `callbackInit` facilities)
346//
347// One process-global `BackgroundExecutor` — callback pool + delayed timer +
348// scanOnce worker — on *every* backend, because it is the only executor whose
349// existence does not depend on an ambient runtime. Two init paths, both
350// landing on the same `OnceLock`:
351//
352//   * Explicit — `background_init()` from `IocApplication::run`, mirroring C's
353//     `callbackInit` running early in `iocInit` (callback.c:286) so the
354//     facilities exist before any record processing can defer a tail.
355//   * Lazy fallback — the first `spawn_background`/`sleep_background` on a path
356//     that never went through `run` (a unit test, an embedded harness)
357//     initialises it on demand via the same `get_or_init`.
358//
359// `exec_backend` additionally routes the *ambient* seam (`spawn`, `sleep`,
360// `interval`) here, because on that backend there is nothing else to route to.
361// ---------------------------------------------------------------------------
362
363static BACKGROUND: std::sync::OnceLock<crate::runtime::background::BackgroundExecutor> =
364    std::sync::OnceLock::new();
365
366/// The process-global background executor, initialised on first use.
367fn background() -> &'static crate::runtime::background::BackgroundExecutor {
368    BACKGROUND.get_or_init(crate::runtime::background::BackgroundExecutor::new)
369}
370
371/// Eagerly start the process-global background executor — C `callbackInit`
372/// parity (callback.c:286), called once from `IocApplication::run`. Idempotent:
373/// a second call (or a prior lazy init) is a no-op, matching `callbackInit`'s
374/// own re-entry guard (callback.c:292-295).
375pub fn background_init() {
376    let _ = background();
377}
378
379/// Handle to a task spawned on the process-global background executor —
380/// `await` for its result, `abort()` to cancel. Distinct from [`TaskHandle`]
381/// because that one is the *ambient* executor's handle and is
382/// `tokio::task::JoinHandle` on a hosted build.
383pub type BackgroundTaskHandle<T> = crate::runtime::background::JoinFuture<T>;
384
385/// Spawn a deferred tail on the process-global background executor.
386///
387/// The counterpart to [`spawn`], and the difference is the whole point of
388/// having both: [`spawn`] follows the *ambient* execution model, so on a
389/// hosted build it needs a tokio runtime entered on the calling thread, while
390/// this one always lands on the same executor no matter who calls it. Record
391/// processing is reached from a plain `std::thread` — every blocking CA/PVA
392/// connection thread drives it through [`block_on_sync`] → `park_on` — so a
393/// tail it defers must not depend on the caller's thread having a runtime.
394///
395/// Anything awaited inside `future` is subject to the same rule: use
396/// [`sleep_background`], [`interval_background`] and [`spawn_blocking_background`]
397/// rather than their ambient counterparts, and no `tokio::net` socket, whose
398/// reactor this executor deliberately does not have.
399pub fn spawn_background<F>(future: F) -> BackgroundTaskHandle<F::Output>
400where
401    F: Future + Send + 'static,
402    F::Output: Send + 'static,
403{
404    use crate::runtime::background::{DEFAULT_SPAWN_PRIORITY, spawn_future};
405    spawn_future(
406        &background().callbacks().handle(),
407        DEFAULT_SPAWN_PRIORITY,
408        future,
409    )
410}
411
412/// [`spawn_background`] for a blocking closure — runs it on a callback-pool
413/// worker at the default Medium band.
414pub fn spawn_blocking_background<F, R>(f: F) -> BackgroundTaskHandle<R>
415where
416    F: FnOnce() -> R + Send + 'static,
417    R: Send + 'static,
418{
419    use crate::runtime::background::{DEFAULT_SPAWN_PRIORITY, spawn_blocking_on};
420    spawn_blocking_on(
421        &background().callbacks().handle(),
422        DEFAULT_SPAWN_PRIORITY,
423        f,
424    )
425}
426
427/// Sleep on the process-global delayed-callback timer — C
428/// `callbackRequestDelayed` (callback.c:410) — measured on `std::time`.
429///
430/// The delay counterpart to [`spawn_background`]: a tail deferred there must
431/// wait on a timer that exists without a runtime, which `tokio::time` does not.
432pub async fn sleep_background(duration: Duration) {
433    crate::runtime::background::timer_sleep::sleep(&background().timer().handle(), duration).await;
434}
435
436/// Periodic ticker on the process-global delayed-callback timer — the
437/// [`interval`] counterpart for work spawned by [`spawn_background`].
438pub fn interval_background(period: Duration) -> crate::runtime::background::TimerInterval {
439    crate::runtime::background::timer_sleep::interval(&background().timer().handle(), period)
440}
441
442#[cfg(tokio_backend)]
443pub fn spawn<F>(future: F) -> TaskHandle<F::Output>
444where
445    F: Future + Send + 'static,
446    F::Output: Send + 'static,
447{
448    tokio::spawn(future)
449}
450
451/// RTEMS: there is no ambient runtime to follow, so the ambient seam *is* the
452/// background executor — [`spawn_background`] verbatim.
453#[cfg(exec_backend)]
454pub fn spawn<F>(future: F) -> TaskHandle<F::Output>
455where
456    F: Future + Send + 'static,
457    F::Output: Send + 'static,
458{
459    spawn_background(future)
460}
461
462/// Yield the current task once — the seam replacement for
463/// `tokio::task::yield_now`, so no call site names `tokio::task` directly.
464#[cfg(tokio_backend)]
465pub async fn yield_now() {
466    tokio::task::yield_now().await;
467}
468
469/// Exec-backend yield: return `Pending` once with the waker already woken.
470/// On the cooperative background executor that re-enqueues the task behind
471/// whatever else is runnable; under a `park_on` driver the wake sets the
472/// park token, so the driver re-polls immediately — both give one fair
473/// scheduling point, which is all `yield_now` promises.
474#[cfg(exec_backend)]
475pub async fn yield_now() {
476    struct YieldNow(bool);
477    impl Future for YieldNow {
478        type Output = ();
479        fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
480            if self.0 {
481                Poll::Ready(())
482            } else {
483                self.0 = true;
484                cx.waker().wake_by_ref();
485                Poll::Pending
486            }
487        }
488    }
489    YieldNow(false).await;
490}
491
492#[cfg(tokio_backend)]
493pub fn spawn_blocking<F, R>(f: F) -> TaskHandle<R>
494where
495    F: FnOnce() -> R + Send + 'static,
496    R: Send + 'static,
497{
498    tokio::task::spawn_blocking(f)
499}
500
501/// RTEMS: see [`spawn`] — the ambient seam is the background executor.
502#[cfg(exec_backend)]
503pub fn spawn_blocking<F, R>(f: F) -> TaskHandle<R>
504where
505    F: FnOnce() -> R + Send + 'static,
506    R: Send + 'static,
507{
508    spawn_blocking_background(f)
509}
510
511/// A set of spawned tasks, joined as they complete — the seam replacement for
512/// `tokio::task::JoinSet`.
513///
514/// `JoinSet` is a **fourth spelling of `tokio::spawn`**, and the one no seam
515/// guard caught: `JoinSet::spawn` calls `tokio::spawn` internally, so it panics
516/// with *"there is no reactor running"* on any thread that is not inside a
517/// tokio runtime — which on RTEMS is every callback-band worker. Measured on
518/// target: the CA client's transport manager died on `cbMedium` at its first
519/// connect (`doc/calink-rtems-design.md` §11.1). Naming it here means a call
520/// site can express "spawn a set of tasks and reap them as they finish"
521/// without reaching past the seam.
522///
523/// The three properties call sites depend on, all preserved:
524///
525/// * **Concurrency** — every member runs independently; joining one does not
526///   block the others.
527/// * **Pair-by-value** — [`Self::join_next`] yields whichever member finished
528///   first, so a task that returns its own key can be matched to its state.
529/// * **Abort on drop** — dropping the set cancels every member that has not
530///   finished, which is the property that distinguishes a `JoinSet` from a bag
531///   of detached `JoinHandle`s.
532pub struct TaskSet<T> {
533    tasks: Vec<TaskHandle<T>>,
534}
535
536impl<T> Default for TaskSet<T> {
537    fn default() -> Self {
538        Self::new()
539    }
540}
541
542impl<T> TaskSet<T> {
543    /// An empty set.
544    pub fn new() -> Self {
545        Self { tasks: Vec::new() }
546    }
547
548    /// Number of members that have not yet been joined.
549    pub fn len(&self) -> usize {
550        self.tasks.len()
551    }
552
553    /// `true` when no member is outstanding.
554    pub fn is_empty(&self) -> bool {
555        self.tasks.is_empty()
556    }
557}
558
559impl<T: Send + 'static> TaskSet<T> {
560    /// Spawn `future` into the set — through [`spawn`], so the RTEMS build
561    /// lands it on a callback band instead of demanding a tokio runtime.
562    pub fn spawn<F>(&mut self, future: F)
563    where
564        F: Future<Output = T> + Send + 'static,
565    {
566        self.tasks.push(spawn(future));
567    }
568
569    /// Wait for the next member to finish and return its result, removing it
570    /// from the set. `None` when the set is empty — matching
571    /// `JoinSet::join_next`, so a `select!` arm on it goes quiet rather than
572    /// spinning once every task has been reaped.
573    ///
574    /// Cancel-safe: the returned future holds no state of its own, so a
575    /// `select!` that drops it loses nothing.
576    pub async fn join_next(&mut self) -> Option<Result<T, TaskJoinError>> {
577        if self.tasks.is_empty() {
578            return None;
579        }
580        std::future::poll_fn(|cx| {
581            for i in 0..self.tasks.len() {
582                // Both backends' handles are `Unpin` (tokio's `JoinHandle`,
583                // and `JoinFuture`, whose only field is an `Arc`), so this
584                // needs no pin projection. Polling every pending member
585                // re-registers this waker with each — the shape both handles
586                // document.
587                if let std::task::Poll::Ready(result) =
588                    std::pin::Pin::new(&mut self.tasks[i]).poll(cx)
589                {
590                    self.tasks.swap_remove(i);
591                    return std::task::Poll::Ready(Some(result));
592                }
593            }
594            std::task::Poll::Pending
595        })
596        .await
597    }
598}
599
600impl<T> Drop for TaskSet<T> {
601    /// Cancel every outstanding member — `JoinSet`'s drop behaviour, and the
602    /// reason a call site reaches for a set rather than a `Vec` of handles.
603    fn drop(&mut self) {
604        for task in &self.tasks {
605            task.abort();
606        }
607    }
608}
609
610#[cfg(tokio_backend)]
611pub async fn sleep(duration: Duration) {
612    tokio::time::sleep(duration).await;
613}
614
615/// RTEMS: see [`spawn`] — the ambient seam is the background executor.
616#[cfg(exec_backend)]
617pub async fn sleep(duration: Duration) {
618    sleep_background(duration).await;
619}
620
621/// The instant [`sleep_until`] measures deadlines against — **the backend's own
622/// clock**, which is the whole point of naming it here.
623///
624/// A deadline is only meaningful in the clock the timer that waits on it runs
625/// on. The hosted timer is tokio's, and under `#[tokio::test(start_paused =
626/// true)]` tokio's clock is virtual and advances on `sleep`, not with the wall
627/// — so a `std::time::Instant` deadline handed to a tokio timer is a deadline
628/// in a *different* timeline, and the wait is wrong by however far the two have
629/// diverged. The RTEMS timer runs on `std::time::Instant` (1-second-quantized
630/// on target, `doc/calink-rtems-design.md` §5.5).
631///
632/// Taking the alias rather than a concrete instant type is what keeps a caller
633/// from mixing them: `Instant::now() + timeout` is the deadline `sleep_until`
634/// will actually honour, on both backends.
635#[cfg(tokio_backend)]
636pub type Instant = tokio::time::Instant;
637/// See the hosted definition.
638#[cfg(exec_backend)]
639pub type Instant = std::time::Instant;
640
641#[cfg(tokio_backend)]
642pub async fn sleep_until(deadline: Instant) {
643    tokio::time::sleep_until(deadline).await;
644}
645
646/// RTEMS: sleep-until on the delayed-callback timer via the host-tested `Sleep`.
647#[cfg(exec_backend)]
648pub async fn sleep_until(deadline: Instant) {
649    crate::runtime::background::timer_sleep::sleep_until(&background().timer().handle(), deadline)
650        .await;
651}
652
653/// Periodic ticker — the seam replacement for `tokio::time::interval`, so no
654/// production site names `tokio::time` directly (decision A2). The hosted build
655/// wraps `tokio::time::Interval`, preserving its default
656/// `MissedTickBehavior::Burst` catch-up and immediate first tick; the RTEMS
657/// build substitutes the runtime-free
658/// [`crate::runtime::background::timer_sleep::TimerInterval`], which reproduces
659/// the same semantics over the delayed-callback timer.
660#[cfg(tokio_backend)]
661pub struct Interval {
662    inner: tokio::time::Interval,
663}
664
665#[cfg(tokio_backend)]
666impl Interval {
667    /// Complete at the next tick. The first tick is immediate (tokio parity);
668    /// callers that want to skip it await `tick()` once up front.
669    pub async fn tick(&mut self) {
670        self.inner.tick().await;
671    }
672}
673
674/// RTEMS: the periodic ticker is the runtime-free `TimerInterval` (same
675/// immediate-first-tick + Burst catch-up semantics, same `tick()` surface).
676#[cfg(exec_backend)]
677pub type Interval = crate::runtime::background::timer_sleep::TimerInterval;
678
679/// Build a periodic ticker firing every `period` — the seam replacement for
680/// `tokio::time::interval`.
681#[cfg(tokio_backend)]
682pub fn interval(period: Duration) -> Interval {
683    Interval {
684        inner: tokio::time::interval(period),
685    }
686}
687
688/// RTEMS: build the periodic ticker on the delayed-callback timer.
689#[cfg(exec_backend)]
690pub fn interval(period: Duration) -> Interval {
691    crate::runtime::background::timer_sleep::interval(&background().timer().handle(), period)
692}
693
694/// The timeout's error — "the deadline elapsed before the future completed".
695/// Hosted this is tokio's own type so `timeout` composes with tokio-aware
696/// callers; the exec backend substitutes a mirror (same Decision-B alias
697/// pattern as `TaskJoinError` — the consumed surface is Debug/Display only).
698#[cfg(tokio_backend)]
699pub use tokio::time::error::Elapsed;
700
701/// Exec-backend mirror of [`tokio::time::error::Elapsed`] (that type has no
702/// public constructor, so the runtime-free `timeout` below cannot return it).
703#[cfg(exec_backend)]
704#[derive(Debug)]
705pub struct Elapsed(());
706
707#[cfg(exec_backend)]
708impl std::fmt::Display for Elapsed {
709    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
710        write!(f, "deadline has elapsed")
711    }
712}
713
714#[cfg(exec_backend)]
715impl std::error::Error for Elapsed {}
716
717/// Await `fut`, giving up after `duration` — the seam's only bounded wait.
718///
719/// It belongs here rather than at the call sites for the same reason [`sleep`]
720/// does: a deadline needs a timer, and which timer that is depends on the
721/// backend. Call sites that reach for `tokio::time::timeout` directly pin
722/// themselves to the tokio timer wheel.
723#[cfg(tokio_backend)]
724pub async fn timeout<F: Future>(duration: Duration, fut: F) -> Result<F::Output, Elapsed> {
725    tokio::time::timeout(duration, fut).await
726}
727
728/// RTEMS/exec: the same bounded wait raced against the delayed-callback
729/// timer's [`sleep`] — no tokio timer wheel, no runtime.
730#[cfg(exec_backend)]
731pub async fn timeout<F: Future>(duration: Duration, fut: F) -> Result<F::Output, Elapsed> {
732    let mut sleep = std::pin::pin!(sleep(duration));
733    let mut fut = std::pin::pin!(fut);
734    std::future::poll_fn(move |cx| {
735        if let Poll::Ready(v) = fut.as_mut().poll(cx) {
736            return Poll::Ready(Ok(v));
737        }
738        if sleep.as_mut().poll(cx).is_ready() {
739            return Poll::Ready(Err(Elapsed(())));
740        }
741        Poll::Pending
742    })
743    .await
744}
745
746/// [`timeout`] against an absolute [`Instant`] instead of a duration — the
747/// seam's `tokio::time::timeout_at`.
748///
749/// It exists because one deadline shared across several sequential awaits is a
750/// different bound from a fresh duration per await: `caget_many` gives its
751/// whole batch one deadline, so a slow first PV eats the budget the rest would
752/// otherwise each get in full. Expressing that with `timeout` would need the
753/// caller to do the subtraction, which is the arithmetic that drifts.
754#[cfg(tokio_backend)]
755pub async fn timeout_at<F: Future>(deadline: Instant, fut: F) -> Result<F::Output, Elapsed> {
756    tokio::time::timeout_at(deadline, fut).await
757}
758
759/// RTEMS/exec: raced against [`sleep_until`], the absolute-deadline twin of
760/// what [`timeout`] races against.
761#[cfg(exec_backend)]
762pub async fn timeout_at<F: Future>(deadline: Instant, fut: F) -> Result<F::Output, Elapsed> {
763    let mut sleep = std::pin::pin!(sleep_until(deadline));
764    let mut fut = std::pin::pin!(fut);
765    std::future::poll_fn(move |cx| {
766        if let Poll::Ready(v) = fut.as_mut().poll(cx) {
767            return Poll::Ready(Ok(v));
768        }
769        if sleep.as_mut().poll(cx).is_ready() {
770            return Poll::Ready(Err(Elapsed(())));
771        }
772        Poll::Pending
773    })
774    .await
775}
776
777pub fn runtime_handle() -> tokio::runtime::Handle {
778    tokio::runtime::Handle::current()
779}
780
781// ---------------------------------------------------------------------------
782// EPICS thread priority abstraction
783//
784// C parity: `modules/libcom/src/osi/epicsThread.h:73-92` defines an
785// integer priority space `0..=99` (`epicsThreadPriorityMin/Max`) with a
786// set of named levels, plus three stack-size classes.
787// `osi/os/posix/osdThread.c` maps an EPICS priority `p` onto the OS
788// SCHED_FIFO range with `oss = p * (max-min)/100 + min` and falls back
789// to a non-RT (default-policy) thread when the process lacks permission
790// to use SCHED_FIFO.
791//
792// The Rust port runs work as tokio tasks on a shared pool, so there is
793// no per-task OS thread to re-prioritise for `spawn`. What is portably
794// achievable is: (a) the priority enum + named levels as a first-class
795// type, (b) a stack-size class with the C size table, and (c) a
796// best-effort OS-scheduler priority applied to the *current* OS thread
797// (used by dedicated `spawn_blocking` threads and the runtime's worker
798// threads). `apply_to_current_thread` reports whether the OS actually
799// honoured the request.
800//
801// (c) is opt-in and off by default — see `RT_PRIORITY_ENV`. C's switch
802// (`EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING`) defaults to YES because
803// a C IOC is deployed onto a machine chosen for it; this crate is just as
804// often linked into a desktop tool, where a silent SCHED_FIFO request is
805// either a guaranteed failure or a way to starve the box.
806// ---------------------------------------------------------------------------
807
808/// Minimum EPICS thread priority (`epicsThreadPriorityMin`).
809pub const PRIORITY_MIN: u8 = 0;
810/// Maximum EPICS thread priority (`epicsThreadPriorityMax`).
811pub const PRIORITY_MAX: u8 = 99;
812
813/// EPICS thread priority — an integer `0..=99` with the named levels
814/// from `epicsThreadPriority*` (`epicsThread.h:73-83`). Lower values
815/// are lower priority; the CA server bands sit below the scan bands so
816/// scan threads preempt CA-server threads on a loaded IOC.
817#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
818pub enum ThreadPriority {
819    /// `epicsThreadPriorityLow` = 10.
820    Low,
821    /// `epicsThreadPriorityCAServerLow` = 20.
822    CaServerLow,
823    /// `epicsThreadPriorityCAServerHigh` = 40.
824    CaServerHigh,
825    /// `epicsThreadPriorityMedium` = 50.
826    Medium,
827    /// `epicsThreadPriorityScanLow` = 60.
828    ScanLow,
829    /// `epicsThreadPriorityScanHigh` = 70.
830    ScanHigh,
831    /// `epicsThreadPriorityHigh` = 90.
832    High,
833    /// `epicsThreadPriorityIocsh` = 91.
834    Iocsh,
835    /// An explicit priority value, clamped to `0..=99` on use.
836    Custom(u8),
837}
838
839impl ThreadPriority {
840    /// The raw EPICS priority value `0..=99`, matching the
841    /// `epicsThreadPriority*` constants in `epicsThread.h`.
842    ///
843    /// `const` so a server can *derive* its band from the named one C derives
844    /// it from — `CaServerLow - 2` rather than a bare `18` with a comment
845    /// asserting the two are the same number. C builds exactly that ladder at
846    /// `caservertask.c:562-575`; restating its output as a literal is how the
847    /// ladder and its constants come to disagree.
848    pub const fn value(self) -> u8 {
849        let v = match self {
850            ThreadPriority::Low => 10,
851            ThreadPriority::CaServerLow => 20,
852            ThreadPriority::CaServerHigh => 40,
853            ThreadPriority::Medium => 50,
854            ThreadPriority::ScanLow => 60,
855            ThreadPriority::ScanHigh => 70,
856            ThreadPriority::High => 90,
857            ThreadPriority::Iocsh => 91,
858            ThreadPriority::Custom(v) => v,
859        };
860        // `Ord::min` is not `const`; the clamp is the same one.
861        if v > PRIORITY_MAX { PRIORITY_MAX } else { v }
862    }
863}
864
865/// Stack-size class — `epicsThreadStackSizeClass` (`epicsThread.h:91`).
866///
867/// The byte size is implementation-dependent in C. These mirror the POSIX
868/// table `STACK_SIZE(f) = f * 0x10000 * sizeof(void*)`
869/// (`libcom/src/osi/os/posix/osdThread.c:506-509`), pointer-width
870/// parameterised exactly as the C macro is: Small = 1, Medium = 2, Big = 4
871/// units of `0x10000 * sizeof(void*)`. On a 64-bit host that is
872/// 512 KiB / 1 MiB / 2 MiB; on `armv7-rtems-eabihf` it is
873/// **256 KiB / 512 KiB / 1 MiB**.
874///
875/// # This is the table a C IOC on RTEMS 6 uses too
876///
877/// Base has a second, much smaller table — 5000 / 8000 / 11000 bytes, floored
878/// at `RTEMS_MINIMUM_STACK_SIZE` — in `os/RTEMS-score/osdThread.c:136-150`.
879/// It does not apply here. `configure/toolchain.c:29-35` selects
880/// `OS_API = posix` for `__RTEMS_MAJOR__ >= 5`, so an RTEMS 6 build searches
881/// `os/RTEMS-posix` then `os/RTEMS`, neither of which contains an
882/// `osdThread.c`, and lands on `os/posix/osdThread.c` — the file above. The
883/// score table is what RTEMS **4/5** used.
884///
885/// Do not "align" these constants with 5000/8000/11000. That would be a
886/// regression against the C IOC we are matching, not a correction: on RTEMS 6
887/// the C IOC asks `pthread_attr_setstacksize` for the POSIX number
888/// (`os/posix/osdThread.c:212-215`), which is the same call `std` makes for
889/// us, with the same argument.
890#[derive(Debug, Clone, Copy, PartialEq, Eq)]
891pub enum StackSizeClass {
892    Small,
893    Medium,
894    Big,
895}
896
897impl StackSizeClass {
898    /// Stack size in bytes for this class, matching the POSIX
899    /// `stackSizeTable` in `osdThread.c` on **every** target.
900    ///
901    /// C's `STACK_SIZE(f) = f * 0x10000 * sizeof(void*)` is parameterised by
902    /// pointer width and so is this, so the two agree by construction rather
903    /// than on one word size: 512 KiB / 1 MiB / 2 MiB on a 64-bit host, and
904    /// 256 KiB / 512 KiB / 1 MiB on `armv7-rtems-eabihf` — which is exactly
905    /// what a C IOC asks `pthread_attr_setstacksize` for on that target.
906    ///
907    /// Read "on a 64-bit target" here before: it was wrong in the direction
908    /// that matters, because it invited a reader to assume the RTEMS numbers
909    /// were unverified. This crate is portable to 64-bit embedded targets
910    /// too — `x86_64-wrs-vxworks` — and pays for it: a 64-bit pointer doubles
911    /// every class in this table, so an `x86_64-wrs-vxworks` CA client thread
912    /// costs exactly 2× what the same thread costs on `armv7-rtems-eabihf`,
913    /// pointer width for pointer width, not a difference in the formula.
914    pub fn bytes(self) -> usize {
915        // STACK_SIZE(f) = f * 0x10000 * sizeof(void*)
916        let unit = 0x10000usize * std::mem::size_of::<usize>();
917        match self {
918            StackSizeClass::Small => unit,
919            StackSizeClass::Medium => 2 * unit,
920            StackSizeClass::Big => 4 * unit,
921        }
922    }
923}
924
925/// Outcome of a best-effort OS-scheduler priority change.
926#[derive(Debug, Clone, Copy, PartialEq, Eq)]
927pub enum PriorityApplied {
928    /// The OS scheduler honoured the requested priority (real-time
929    /// SCHED_FIFO band applied).
930    Realtime,
931    /// Real-time scheduling was never requested: the opt-in switch
932    /// [`RT_PRIORITY_ENV`] is off, so **no scheduler call was made at
933    /// all** and the thread keeps the process default policy.
934    Disabled,
935    /// The platform does not expose a portable scheduler priority API
936    /// (e.g. Windows here, or a non-Unix target) — no change applied.
937    Unsupported,
938    /// The platform exposes the API but rejected the request (typically
939    /// the process lacks `CAP_SYS_NICE`/root for SCHED_FIFO). C's
940    /// `osdThread.c` makes the same best-effort fall back to a non-RT
941    /// thread in this case (`osdThread.c:647` "Try again without
942    /// SCHED_FIFO").
943    BestEffortFailed,
944}
945
946impl PriorityApplied {
947    /// `true` only when the OS actually applied a real-time priority.
948    pub fn is_realtime(self) -> bool {
949        matches!(self, PriorityApplied::Realtime)
950    }
951}
952
953/// Environment switch that opts this process in to real-time (SCHED_FIFO)
954/// scheduling for the IOC threads that carry an EPICS priority.
955///
956/// The switch is read in both directions on every target; what differs is
957/// what it defaults to when unset — see [`DEFAULT_POLICY`].
958///
959/// Accepted "on" values, case-insensitive: `YES`, `TRUE`, `ON`, `1`.
960/// Any other *explicit* value is off.
961///
962/// # Relationship to the C switch
963///
964/// C base has the same concept under
965/// `EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING` (`envDefs.h:80`, read at
966/// `osdThread.c:389`), and `envGetBoolConfigParam` (`envSubr.c:331`) accepts
967/// only case-insensitive `yes`. We deliberately do **not** reuse that name:
968/// its base default is `YES` on every target (`configure/CONFIG_ENV:57`)
969/// while ours is `YES` only on RTEMS, so one name would carry two different
970/// defaults on a hosted build depending on which implementation read it.
971pub const RT_PRIORITY_ENV: &str = "EPICS_RS_ALLOW_RT_PRIORITY";
972
973/// Whether this process may ask the OS for real-time scheduling.
974///
975/// Resolved from [`RT_PRIORITY_ENV`] exactly once per process by
976/// [`RtPolicy::current`]. It is a *parameter* of
977/// [`apply_to_current_thread_under`] rather than a check buried inside the
978/// syscall wrapper, so "switch off ⟹ no scheduler call" is a property of
979/// the call graph and not of a runtime branch some future caller can skip.
980#[derive(Debug, Clone, Copy, PartialEq, Eq)]
981pub enum RtPolicy {
982    /// Never touch the OS scheduler.
983    Disabled,
984    /// Best-effort SCHED_FIFO, falling back to default scheduling.
985    AllowRealtime,
986}
987
988/// What [`RT_PRIORITY_ENV`] means when it is **unset**, for the target this
989/// was compiled for.
990///
991/// `AllowRealtime` on RTEMS, `Disabled` everywhere else. The asymmetry is
992/// deliberate and is a property of the default itself — not a `setenv` the
993/// boot shim performs before `main()`. A `setenv` is a runtime side effect any
994/// later caller can undo or reorder, and a variable one component writes for
995/// another to read is the dual-meaning shape this code keeps removing.
996///
997/// Why the embedded targets differ from hosted:
998///
999///   - Base's own equivalent switch,
1000///     `EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING`, defaults to `YES`
1001///     (`configure/CONFIG_ENV:57`). An IOC that honours its priorities is
1002///     upstream's default posture, not an opt-in.
1003///   - The opt-in gate exists for RT-Linux, where asking for SCHED_FIFO needs
1004///     `CAP_SYS_NICE` or a non-zero `RLIMIT_RTPRIO` (so on a desktop the
1005///     request merely fails), and where a runaway RT band on a box that
1006///     *grants* it can wedge a developer's machine. Neither failure mode
1007///     exists on RTEMS or VxWorks: there is no RLIMIT_RTPRIO, no
1008///     `CAP_SYS_NICE` gate, and no desktop to wedge on either.
1009///   - The band invariant is now a test rather than a hope —
1010///     `rtems_priority_map_stays_below_the_libbsd_network_band` proves every
1011///     u8 input lands in core 100..199, at or below libbsd's default band and
1012///     strictly less urgent than IRQS(96)/TIME(98).
1013///   - **VxWorks is measurement-backed, not assumed.** On the bring-up box
1014///     (VxWorks 7, `x86_64-wrs-vxworks`), 11 of 11 measured threads landed
1015///     `PriorityApplied::Realtime` via `SCHED_FIFO`, exactly one scheduler
1016///     call each, at `posix = 56 + epics` — the same POSIX value RTEMS gets
1017///     (see `map_epics_priority_rtems`) — which VxWorks's own POSIX layer
1018///     then inverts into its native task-priority space at `vx = 199 -
1019///     epics`, exact: EPICS base's own vxWorks-port formula
1020///     (`vxWorks/osdThread.c:99`), reached by construction rather than by
1021///     restating it (see `map_epics_priority_vxworks`).
1022///
1023/// An explicit value still wins in **both** directions on every target, so
1024/// `EPICS_RS_ALLOW_RT_PRIORITY=NO` turns it off on RTEMS or VxWorks.
1025pub const DEFAULT_POLICY: RtPolicy = default_policy(cfg!(epics_embedded_target));
1026
1027/// [`DEFAULT_POLICY`] as a pure function of the one target fact it depends
1028/// on, so both arms are reachable from a host test run. A host CI will never
1029/// execute the RTEMS arm otherwise, and an untested default is exactly the
1030/// kind that drifts.
1031const fn default_policy(on_rtems: bool) -> RtPolicy {
1032    if on_rtems {
1033        RtPolicy::AllowRealtime
1034    } else {
1035        RtPolicy::Disabled
1036    }
1037}
1038
1039impl RtPolicy {
1040    /// Parse a raw switch value (`None` = unset ⇒ [`DEFAULT_POLICY`]).
1041    pub fn from_env_value(raw: Option<&str>) -> RtPolicy {
1042        Self::resolve(raw, DEFAULT_POLICY)
1043    }
1044
1045    /// [`Self::from_env_value`] with the unset-default injected, so a host
1046    /// test can ask what an RTEMS process would do with the same input.
1047    pub fn resolve(raw: Option<&str>, default: RtPolicy) -> RtPolicy {
1048        let Some(raw) = raw else {
1049            return default;
1050        };
1051        let v = raw.trim();
1052        let on = v.eq_ignore_ascii_case("yes")
1053            || v.eq_ignore_ascii_case("true")
1054            || v.eq_ignore_ascii_case("on")
1055            || v == "1";
1056        if on {
1057            RtPolicy::AllowRealtime
1058        } else {
1059            RtPolicy::Disabled
1060        }
1061    }
1062
1063    /// The process-wide policy, read from [`RT_PRIORITY_ENV`] on first use
1064    /// and cached. Caching matches C, which resolves its switch once in
1065    /// `epicsThreadInit` (`osdThread.c:389`), and keeps later `set_var`
1066    /// calls from changing the scheduling of threads already running.
1067    pub fn current() -> RtPolicy {
1068        static POLICY: std::sync::OnceLock<RtPolicy> = std::sync::OnceLock::new();
1069        *POLICY.get_or_init(|| {
1070            RtPolicy::from_env_value(std::env::var(RT_PRIORITY_ENV).ok().as_deref())
1071        })
1072    }
1073}
1074
1075/// Apply an EPICS [`ThreadPriority`] to the **current OS thread**, best
1076/// effort.
1077///
1078/// C parity: mirrors `osdThread.c`'s SCHED_FIFO mapping
1079/// `oss = p * (max-min)/100 + min` over the kernel's
1080/// `sched_get_priority_min/max(SCHED_FIFO)` range, and the
1081/// EPERM-fallback to a non-RT thread.
1082///
1083/// Returns [`PriorityApplied`] describing what the platform allowed —
1084/// callers running in environments without RT permission still get a
1085/// running thread, just at the default policy, exactly as a C IOC does.
1086///
1087/// Note: tokio tasks spawned via [`spawn`] share worker threads, so
1088/// this is meaningful for [`spawn_blocking`] closures and for tuning
1089/// the runtime's worker threads at startup — not for individual async
1090/// tasks.
1091///
1092/// Platform support: the OS-scheduler change is wired on Linux, via the
1093/// range-probed linear map of `os/posix/osdThread.c`, and on RTEMS, via the
1094/// fixed map of `os/RTEMS-score/osdThread.c` inverted into POSIX space (see
1095/// `map_epics_priority_rtems` — the two maps differ in shape, deliberately).
1096/// On other targets the priority enum + API surface still exist but `apply`
1097/// reports [`PriorityApplied::Unsupported`] — no band has been measured there.
1098///
1099/// Opt-in: real-time scheduling is only ever requested when
1100/// [`RT_PRIORITY_ENV`] is set (see [`RtPolicy`]). With the switch off this
1101/// returns [`PriorityApplied::Disabled`] without calling the OS at all.
1102pub fn apply_to_current_thread(priority: ThreadPriority) -> PriorityApplied {
1103    apply_to_current_thread_under(RtPolicy::current(), priority)
1104}
1105
1106/// [`apply_to_current_thread`] with the real-time policy supplied by the
1107/// caller instead of read from the environment.
1108///
1109/// The single gate: [`RtPolicy::Disabled`] returns before any scheduler
1110/// call is reachable. Exposed so a caller that already owns its RT policy
1111/// (and the tests that must exercise both states in one process, since
1112/// [`RtPolicy::current`] is cached) does not have to mutate the environment.
1113pub fn apply_to_current_thread_under(
1114    policy: RtPolicy,
1115    priority: ThreadPriority,
1116) -> PriorityApplied {
1117    match policy {
1118        RtPolicy::Disabled => PriorityApplied::Disabled,
1119        RtPolicy::AllowRealtime => apply_priority_impl(priority.value()),
1120    }
1121}
1122
1123/// The prologue an IOC thread runs as its first statement, when it takes on
1124/// its role: publish its name to the OS, then request its scheduling band.
1125///
1126/// Two things a thread owes the operator, and they have different gates.
1127/// The band is opt-in ([`RT_PRIORITY_ENV`]) and best effort. The **name** is
1128/// unconditional: a thread that cannot be identified in a task listing
1129/// cannot be diagnosed, and on RTEMS that listing is often the only
1130/// instrument there is — bring-up had to measure libbsd's priority band by
1131/// other means precisely because none of our threads carried a name the
1132/// kernel could show.
1133///
1134/// Use this rather than [`apply_to_current_thread`] at a thread's entry, so
1135/// naming cannot be forgotten by the next thread somebody adds. Call
1136/// [`apply_to_current_thread`] directly only when re-banding a thread that
1137/// is already named and running. A thread that deliberately takes no EPICS
1138/// band — the iocsh script runners — calls [`name_current_thread`] alone
1139/// rather than inventing a priority just to be visible.
1140///
1141/// The band this asks the OS for is also what orders the blocking locks in
1142/// `server::database::record_lock` and its siblings: they are
1143/// priority-inheritance mutexes, so the wait queue is the *kernel's* and it
1144/// is ranked by the scheduling priority requested here. With
1145/// [`RtPolicy::Disabled`] no scheduler call happens and there is no ordering
1146/// to have — the hosted default, where the locks still exclude but do not
1147/// prioritise ([`crate::runtime::sync::is_pi_mutex_active`]).
1148/// A third thing on VxWorks, for the same reason as the name: an RTP cannot
1149/// enumerate its own tasks, so the statistics funnel's thread census is built
1150/// from what announces itself here. This is the seam because it is already the
1151/// one every IOC thread passes through to take its band — "every thread that
1152/// bands itself registers itself" adds a consequence to that invariant rather
1153/// than a rule to remember at each spawn. A thread that starts outside it is
1154/// invisible to that census, and the census output says so in its own header.
1155pub fn enter_ioc_thread(priority: ThreadPriority) -> PriorityApplied {
1156    name_current_thread();
1157    #[cfg(target_os = "vxworks")]
1158    epics_rtems_boot::stats::register_task();
1159    apply_to_current_thread(priority)
1160}
1161
1162/// Push `std::thread::current().name()` down to the OS thread object.
1163///
1164/// No-op off RTEMS: `std` already calls the platform's `pthread_setname_np`
1165/// from `Builder::spawn` on every hosted target it supports. RTEMS is not in
1166/// that list, so a name set with `Builder::name` lives only in Rust's own
1167/// `Thread` struct and never reaches the kernel — which is what makes our
1168/// threads invisible to an RTEMS task listing.
1169#[cfg(not(target_os = "rtems"))]
1170pub fn name_current_thread() {}
1171
1172/// RTEMS: `pthread_setname_np` (`cpukit/posix/src/pthreadsetnamenp.c`) into
1173/// `_Thread_Set_name`, which `strlcpy`s into `_Thread_Maximum_name_size`.
1174///
1175/// That size is `CONFIGURE_MAXIMUM_THREAD_NAME_SIZE`, default **16**
1176/// including the NUL (`rtems/score/thread.h:1079`,
1177/// `rtems/confdefs/threads.h:92-93`), and the boot shim does not override
1178/// it — so 15 usable bytes, the same budget `std` truncates to on Linux
1179/// (`TASK_COMM_LEN`). Truncating here rather than letting the kernel do it
1180/// keeps that existing rule and keeps the call's success unambiguous:
1181/// `_Thread_Set_name` still *sets* an over-long name, it just also returns
1182/// `STATUS_RESULT_TOO_LARGE` → `ERANGE`, so an untruncated call would report
1183/// failure for a name it had in fact applied.
1184#[cfg(target_os = "rtems")]
1185pub fn name_current_thread() {
1186    let current = std::thread::current();
1187    let Some(name) = current.name() else {
1188        return;
1189    };
1190    let Ok(c_name) = std::ffi::CString::new(truncate_thread_name(name)) else {
1191        // An interior NUL cannot come from `Builder::name`, which takes a
1192        // `String`; nothing to publish if one ever did.
1193        return;
1194    };
1195    // SAFETY: `pthread_setname_np` acts on the calling thread and reads a
1196    // NUL-terminated string that outlives the call.
1197    let rc =
1198        unsafe { rtems_sched::pthread_setname_np(rtems_sched::pthread_self(), c_name.as_ptr()) };
1199    if rc != 0 {
1200        tracing::debug!(
1201            target: "epics_base_rs::runtime",
1202            thread = name,
1203            errno = rc,
1204            "pthread_setname_np failed; thread stays unnamed in the task listing"
1205        );
1206    }
1207}
1208
1209/// `CONFIGURE_MAXIMUM_THREAD_NAME_SIZE` (default 16) minus the NUL.
1210#[cfg(any(target_os = "rtems", test))]
1211const RTEMS_MAX_THREAD_NAME_BYTES: usize = 15;
1212
1213/// Cut a thread name to what an RTEMS thread object can hold, on a UTF-8
1214/// boundary.
1215///
1216/// Byte budget, not character count — `_Thread_Set_name` `strlcpy`s bytes —
1217/// but never mid-codepoint, or the task listing shows invalid UTF-8. Kept as
1218/// a pure function so the rule is testable on the host, where the caller
1219/// that applies it does not exist.
1220#[cfg(any(target_os = "rtems", test))]
1221fn truncate_thread_name(name: &str) -> &str {
1222    let mut end = name.len().min(RTEMS_MAX_THREAD_NAME_BYTES);
1223    while end > 0 && !name.is_char_boundary(end) {
1224        end -= 1;
1225    }
1226    &name[..end]
1227}
1228
1229/// The SCHED_FIFO priority range this process may actually enter.
1230///
1231/// C parity: `find_pri_range` (`osdThread.c:259-314`). The kernel's
1232/// `sched_get_priority_max` reports the *policy's* range and ignores
1233/// `RLIMIT_RTPRIO`, so on an RT box with a restricted limit the nominal
1234/// range is wider than the usable one; C binary-searches for the real
1235/// ceiling and so do we.
1236#[cfg(target_os = "linux")]
1237#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1238enum RtRange {
1239    /// The kernel does not report a SCHED_FIFO range at all.
1240    Unsupported,
1241    /// The range exists but this process may not enter it — no
1242    /// `CAP_SYS_NICE` and `RLIMIT_RTPRIO` is 0. C's equivalent is
1243    /// `usePolicy == 0` (`osdThread.c:279-285`, `:331`), which makes it
1244    /// stop asking for SCHED_FIFO for the life of the process.
1245    Denied,
1246    /// Priorities `min..=max` are settable.
1247    Available { min: i32, max: i32 },
1248}
1249
1250/// Probe once per process and cache. Only reachable on the
1251/// [`RtPolicy::AllowRealtime`] path, so a default (switch-off) process
1252/// never runs the probe and never makes a scheduler call.
1253#[cfg(target_os = "linux")]
1254fn permitted_fifo_range() -> RtRange {
1255    static RANGE: std::sync::OnceLock<RtRange> = std::sync::OnceLock::new();
1256    *RANGE.get_or_init(probe_fifo_range)
1257}
1258
1259#[cfg(target_os = "linux")]
1260fn probe_fifo_range() -> RtRange {
1261    // SAFETY: sched_get_priority_min/max take only an int policy and have
1262    // no preconditions.
1263    let (min, max) = unsafe {
1264        (
1265            libc::sched_get_priority_min(libc::SCHED_FIFO),
1266            libc::sched_get_priority_max(libc::SCHED_FIFO),
1267        )
1268    };
1269    if min < 0 || max < 0 || max < min {
1270        return RtRange::Unsupported;
1271    }
1272
1273    // The probe *changes the scheduling of the thread that runs it*, so it
1274    // runs on a throwaway thread — exactly why C hands `find_pri_range` to
1275    // its own `pthread_create`/`pthread_join` pair (`osdThread.c:316-334`).
1276    let charge = crate::runtime::worker_pool::ThreadCharge::fixed(StackSizeClass::Small);
1277    let probe = std::thread::Builder::new()
1278        .name("cbRtProbe".to_string())
1279        // Two `sched_get_priority_*` calls and a `sched_setscheduler`; nothing
1280        // recurses. Linux-only, so this is not the RTEMS ceiling — but there is
1281        // no reason for a throwaway probe to reserve 2 MiB on any target.
1282        .stack_size(StackSizeClass::Small.bytes())
1283        .spawn(move || {
1284            let _charge = charge;
1285            // `osdThread.c:277-287`: failing at the minimum means no
1286            // permission for SCHED_FIFO at all.
1287            if set_fifo_priority(min) != 0 {
1288                return RtRange::Denied;
1289            }
1290            // `osdThread.c:296-307`: binary-search the real ceiling.
1291            let (mut low, mut high) = (min, max);
1292            while low < high {
1293                let mid = (high + low) / 2;
1294                if set_fifo_priority(mid) != 0 {
1295                    high = mid;
1296                } else {
1297                    low = mid + 1;
1298                }
1299            }
1300            // `osdThread.c:310`: `max_pri = try_pri(max) ? max-1 : max`.
1301            let top = if set_fifo_priority(high) != 0 {
1302                high - 1
1303            } else {
1304                high
1305            };
1306            RtRange::Available { min, max: top }
1307        });
1308    match probe.map(std::thread::JoinHandle::join) {
1309        Ok(Ok(range)) => range,
1310        // Cannot spawn, or the probe died: treat as no RT rather than
1311        // guessing a range we have not shown to be settable.
1312        _ => RtRange::Denied,
1313    }
1314}
1315
1316/// Map an EPICS priority `0..=99` onto the permitted SCHED_FIFO range.
1317///
1318/// C parity: `epicsThreadGetPosixPriority` (`osdThread.c:129-144`) — the
1319/// POSIX counterpart of the `epicsThreadGetOssPriorityValue` used on
1320/// RTEMS/vxWorks (`RTEMS-score/osdThread.c:94`, `vxWorks/osdThread.c:99`).
1321///
1322/// **Hosted only.** RTEMS deliberately does not use this map — see
1323/// `map_epics_priority_rtems` for the shape and the reason. The `test`
1324/// arm of the cfg exists so the two maps can be compared in one process
1325/// on the host; without it the divergence test would silently vanish.
1326#[cfg(any(target_os = "linux", test))]
1327fn map_epics_priority(epics_priority: u8, min: i32, max: i32) -> i32 {
1328    // `osdThread.c:133-134`: a degenerate range collapses to one level.
1329    if max == min {
1330        return max;
1331    }
1332    let slope = (max - min) as f64 / 100.0;
1333    let oss = epics_priority as f64 * slope + min as f64;
1334    // `ThreadPriority::value` caps at 99 and the slope is over 100, so this
1335    // cannot exceed `max`; the clamp guards the probed bounds, which are
1336    // runtime values rather than compile-time constants.
1337    (oss as i32).clamp(min, max)
1338}
1339
1340/// The highest RTEMS *core* priority number, i.e. the least urgent level.
1341///
1342/// Measured on the bring-up guest (RTEMS 6 + libbsd, QEMU
1343/// `xilinx_zynq_a9`): `RTEMS_MAXIMUM_PRIORITY == 255`, the idle thread runs
1344/// at core 255, and `sched_get_priority_min/max(SCHED_FIFO)` report `1`/`254`.
1345/// The POSIX-to-core inversion `core = 255 - posix` was verified in both
1346/// directions on that guest.
1347#[cfg(any(target_os = "rtems", test))]
1348const RTEMS_MAXIMUM_PRIORITY: i32 = 255;
1349
1350/// The RTEMS *core* priority an EPICS priority must land on.
1351///
1352/// Verbatim `epicsThreadGetOssPriorityValue` from EPICS's own RTEMS port,
1353/// `libcom/src/osi/os/RTEMS-score/osdThread.c:94-102`:
1354///
1355/// ```c
1356/// int epicsThreadGetOssPriorityValue(unsigned int osiPriority)
1357/// {
1358///     if (osiPriority > 99) { return 100; }
1359///     else { return (199 - (signed int)osiPriority); }
1360/// }
1361/// ```
1362///
1363/// Fixed offsets, not a range-scaled slope. The whole EPICS space therefore
1364/// occupies core `100..=199` and nothing else can be reached — which is the
1365/// property [`map_epics_priority_rtems`] is chosen for.
1366#[cfg(any(target_os = "rtems", test))]
1367const fn rtems_core_priority(epics_priority: u8) -> i32 {
1368    if epics_priority > 99 {
1369        100
1370    } else {
1371        199 - epics_priority as i32
1372    }
1373}
1374
1375/// Map an EPICS priority onto an RTEMS **POSIX** SCHED_FIFO priority.
1376///
1377/// A distinct function from `map_epics_priority` on purpose: the two have
1378/// different *shapes*, not different endpoints. Expressing this one as the
1379/// hosted linear map with `min`/`max` retuned would re-introduce the linear
1380/// map the moment somebody adjusted a constant, and the linear map is the
1381/// thing this arm exists to avoid.
1382///
1383/// **Deliberate deviation from base-on-RTEMS-6.** EPICS base compiles
1384/// `os/posix/osdThread.c` on RTEMS 6 — `configure/toolchain.c:31-36` sets
1385/// `OS_API = posix` for `__RTEMS_MAJOR__ >= 5`, and `os/RTEMS-posix/` ships
1386/// no `osdThread.c` — so upstream applies the *linear* map
1387/// `oss = epics*(max-min)/100 + min` over `find_pri_range`'s result, which
1388/// on this guest is `min=1`/`max=254`, with
1389/// `EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING` defaulting to `YES`
1390/// (`configure/CONFIG_ENV:57`). That places EPICS 91 (the CA server band) at
1391/// posix 231, i.e. **core 24** — far above libbsd's network threads. The
1392/// crossover is EPICS **63** (posix 160, core 95): every EPICS priority at or
1393/// above it outranks the interrupt server. Reproducing that would reproduce
1394/// the hazard, so this port takes EPICS's *own* RTEMS answer instead —
1395/// [`rtems_core_priority`] — and inverts it into the POSIX space we actually
1396/// set:
1397///
1398/// ```text
1399/// core = RTEMS_MAXIMUM_PRIORITY - posix   (measured)
1400/// core = 199 - epics                      (RTEMS-score/osdThread.c:94-102)
1401/// ⟹ posix = 255 - (199 - epics) = 56 + epics
1402/// ```
1403///
1404/// So EPICS 0 → posix 56 → core 199, EPICS 99 → posix 155 → core 100, and
1405/// anything above 99 clamps to posix 155. Every value is inside the guest's
1406/// settable `[1, 254]`. **Measured on target**, core 100 is also where
1407/// libbsd's own twelve default-band worker threads sit, so the map's most
1408/// urgent reachable value *ties* libbsd's default band there rather than
1409/// staying strictly below it — a boundary tie by construction, not a
1410/// collision-free image. It is still strictly below `IRQS`(96)/`TIME`(98);
1411/// see `rtems_priority_map_stays_below_the_libbsd_network_band`, which
1412/// asserts the non-strict `core >= 100` this tie actually produces.
1413#[cfg(any(target_os = "rtems", test))]
1414pub(crate) fn map_epics_priority_rtems(epics_priority: u8) -> i32 {
1415    RTEMS_MAXIMUM_PRIORITY - rtems_core_priority(epics_priority)
1416}
1417
1418/// Map an EPICS priority onto a VxWorks **POSIX** SCHED_FIFO priority.
1419///
1420/// **Measurement-backed**, not derived: on the bring-up box (VxWorks 7,
1421/// `x86_64-wrs-vxworks`), setting `posix = 56 + epics` — the identical POSIX
1422/// value `map_epics_priority_rtems` computes for RTEMS — landed 11 of 11
1423/// measured threads at `PriorityApplied::Realtime`, one scheduler call each.
1424/// VxWorks's own POSIX layer then inverts that POSIX value into its native
1425/// task-priority space, and the result observed there was `vx = 199 -
1426/// epics`, exact: EPICS base's own vxWorks-port formula
1427/// (`vxWorks/osdThread.c:99`, `oss = 199 - osiPriority`) — reached by a
1428/// different route (we set the POSIX value; VxWorks inverts it, rather than
1429/// us computing the native value directly as C's own port does).
1430///
1431/// Deliberately **not** implemented by calling `rtems_core_priority` /
1432/// `map_epics_priority_rtems`: those compute an RTEMS **core** priority
1433/// through `RTEMS_MAXIMUM_PRIORITY`, an RTEMS kernel constant measured on the
1434/// RTEMS bring-up guest — machinery VxWorks has no equivalent of. The two
1435/// happen to land on the same POSIX number; this restates the `56 + epics`
1436/// arithmetic directly so this function cites no RTEMS-specific fact and a
1437/// change to the RTEMS core-priority mechanism cannot silently move the
1438/// VxWorks value with it.
1439#[cfg(any(target_os = "vxworks", test))]
1440pub(crate) fn map_epics_priority_vxworks(epics_priority: u8) -> i32 {
1441    56 + epics_priority.min(99) as i32
1442}
1443
1444/// Ask the OS for SCHED_FIFO at `oss` on the **calling** thread. The single
1445/// place this crate touches the scheduler; returns the raw `pthread_*`
1446/// status (0 on success). Counting lives here so the "switch off ⟹ no
1447/// scheduler call" guarantee is observable on every target that has one.
1448#[cfg(any(target_os = "linux", epics_embedded_target))]
1449fn set_fifo_priority(oss: i32) -> i32 {
1450    SCHED_CALLS_MADE.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1451    #[cfg(test)]
1452    SCHED_CALLS.with(|c| c.set(c.get() + 1));
1453    set_fifo_priority_raw(oss)
1454}
1455
1456#[cfg(target_os = "linux")]
1457fn set_fifo_priority_raw(oss: i32) -> i32 {
1458    let param = libc::sched_param {
1459        sched_priority: oss,
1460    };
1461    // SAFETY: pthread_setschedparam operates on the calling thread with a
1462    // stack-local sched_param and a valid policy constant.
1463    unsafe { libc::pthread_setschedparam(libc::pthread_self(), libc::SCHED_FIFO, &param) }
1464}
1465
1466/// The RTEMS scheduler surface, declared here rather than taken from `libc`.
1467///
1468/// `libc`'s `newlib/rtems` module (0.2.188) declares neither `sched_param`,
1469/// `SCHED_FIFO`, `pthread_setschedparam` nor `pthread_self` — its sibling
1470/// newlib targets `vita` and `horizon` declare all of them, RTEMS does not.
1471/// The functions exist in the RTEMS 6 kernel
1472/// (`cpukit/posix/src/pthreadsetschedparam.c`) and in the toolchain headers;
1473/// only the Rust binding is missing.
1474///
1475/// `libc::timespec` is deliberately NOT used to describe the sporadic-server
1476/// tail: `libc` types `time_t` as `i32` for every newlib target except
1477/// `horizon`/`espidf` (`src/unix/newlib/mod.rs:55-64`), while the arm-rtems6
1478/// toolchain has `sizeof(time_t) == 8`. Its `timespec` is therefore half the
1479/// real width on this target, so the tail is carried as opaque bytes sized
1480/// from the target compiler instead.
1481///
1482/// **RTEMS-only, deliberately not widened to VxWorks.** VxWorks is not
1483/// newlib, and `libc` *does* declare `sched_param`/`SCHED_FIFO`/
1484/// `pthread_setschedparam`/`pthread_self` for it — with a different
1485/// `sched_param` layout (48 bytes, but `sched_priority: c_int` followed by a
1486/// *typed* `sched_ss_low_priority`/two `timespec`s/`sched_ss_max_repl` tail,
1487/// not this module's opaque bytes). Reusing this RTEMS-shaped struct for
1488/// VxWorks was measured to "work" only because `SCHED_FIFO` never reads past
1489/// `sched_priority` at offset 0 — the tail's true shape never mattered for
1490/// that policy — which is exactly the kind of coincidence a struct-layout
1491/// mismatch should not be allowed to depend on. VxWorks's `set_fifo_priority_raw`
1492/// arm below therefore uses `libc::sched_param` directly.
1493#[cfg(target_os = "rtems")]
1494mod rtems_sched {
1495    use std::ffi::c_int;
1496
1497    /// `sys/sched.h`: `#define SCHED_FIFO 1`.
1498    pub const SCHED_FIFO: c_int = 1;
1499
1500    /// `struct sched_param` as arm-rtems6 lays it out.
1501    ///
1502    /// `sys/features.h:404-405` defines both `_POSIX_SPORADIC_SERVER` and
1503    /// `_POSIX_THREAD_SPORADIC_SERVER`, so `sys/sched.h` compiles the
1504    /// sporadic-server tail in. Measured with the target compiler
1505    /// (`arm-rtems6-gcc`, `sizeof`/`offsetof` via array-length symbols):
1506    ///
1507    /// | field | offset | size |
1508    /// |-------|--------|------|
1509    /// | `sched_priority`        |  0 |  4 |
1510    /// | `sched_ss_low_priority` |  4 |  4 |
1511    /// | `sched_ss_repl_period`  |  8 | 16 |
1512    /// | `sched_ss_init_budget`  | 24 | 16 |
1513    /// | `sched_ss_max_repl`     | 40 |  4 |
1514    ///
1515    /// total 48, align 8. `SCHED_FIFO` makes the kernel read only
1516    /// `sched_priority` (`_POSIX_Thread_Translate_sched_param` takes the
1517    /// sporadic branch for `SCHED_SPORADIC` alone), but the struct is
1518    /// declared at full width anyway so the kernel is never handed a pointer
1519    /// to less memory than its own header describes.
1520    #[repr(C, align(8))]
1521    pub struct SchedParam {
1522        pub sched_priority: c_int,
1523        /// Offsets 4..48 — the sporadic-server fields, unused under
1524        /// `SCHED_FIFO` and always zeroed.
1525        pub sporadic_tail: [u8; 44],
1526    }
1527
1528    // The whole point of the opaque tail is that the width is right. If a
1529    // future edit reaches for `libc::timespec` here, this stops the build
1530    // instead of silently handing the kernel a short buffer.
1531    const _: () = {
1532        assert!(core::mem::size_of::<SchedParam>() == 48);
1533        assert!(core::mem::align_of::<SchedParam>() == 8);
1534    };
1535
1536    unsafe extern "C" {
1537        pub fn pthread_self() -> libc::pthread_t;
1538        pub fn pthread_setschedparam(
1539            thread: libc::pthread_t,
1540            policy: c_int,
1541            param: *const SchedParam,
1542        ) -> c_int;
1543        /// `cpukit/posix/src/pthreadsetnamenp.c`. Also absent from `libc`'s
1544        /// `newlib/rtems` module.
1545        pub fn pthread_setname_np(thread: libc::pthread_t, name: *const std::ffi::c_char) -> c_int;
1546    }
1547}
1548
1549#[cfg(target_os = "rtems")]
1550fn set_fifo_priority_raw(oss: i32) -> i32 {
1551    let param = rtems_sched::SchedParam {
1552        sched_priority: oss,
1553        sporadic_tail: [0u8; 44],
1554    };
1555    // SAFETY: `pthread_setschedparam` acts on the calling thread, is handed a
1556    // stack-local `sched_param` of the target's own width (asserted above),
1557    // and a policy constant taken from `sys/sched.h`.
1558    unsafe {
1559        rtems_sched::pthread_setschedparam(
1560            rtems_sched::pthread_self(),
1561            rtems_sched::SCHED_FIFO,
1562            &param,
1563        )
1564    }
1565}
1566
1567/// VxWorks: `libc::sched_param` directly, not the RTEMS-shaped struct above.
1568///
1569/// Unlike RTEMS, `libc` declares this target's own `sched_param` — a
1570/// `sched_priority: c_int` followed by a *typed* sporadic-server tail
1571/// (`sched_ss_low_priority: c_int`, two `libc::timespec` fields,
1572/// `sched_ss_max_repl: c_int`) — so there is nothing to hand-lay: the tail is
1573/// zeroed rather than omitted because `SCHED_FIFO` never reads it, matching
1574/// the RTEMS arm's own reasoning, but the fields are the platform's real
1575/// fields at the platform's real offsets rather than opaque bytes sized by
1576/// guesswork.
1577#[cfg(target_os = "vxworks")]
1578fn set_fifo_priority_raw(oss: i32) -> i32 {
1579    let param = libc::sched_param {
1580        sched_priority: oss,
1581        sched_ss_low_priority: 0,
1582        sched_ss_repl_period: libc::timespec {
1583            tv_sec: 0,
1584            tv_nsec: 0,
1585        },
1586        sched_ss_init_budget: libc::timespec {
1587            tv_sec: 0,
1588            tv_nsec: 0,
1589        },
1590        sched_ss_max_repl: 0,
1591    };
1592    // SAFETY: pthread_setschedparam operates on the calling thread with a
1593    // stack-local sched_param of libc's own VxWorks width and a valid policy
1594    // constant.
1595    unsafe { libc::pthread_setschedparam(libc::pthread_self(), libc::SCHED_FIFO, &param) }
1596}
1597
1598/// The unprivileged-fallback message. Emitted **once per process**: the
1599/// denial is a property of the process, not of the thread that happened to
1600/// notice it first, and an IOC creates a thread per CA client.
1601#[cfg(target_os = "linux")]
1602fn warn_rt_denied_once() {
1603    static WARNED: std::sync::Once = std::sync::Once::new();
1604    WARNED.call_once(|| {
1605        #[cfg(test)]
1606        DENIED_WARNINGS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1607        tracing::warn!(
1608            target: "epics_base_rs::runtime",
1609            switch = RT_PRIORITY_ENV,
1610            "{RT_PRIORITY_ENV} asked for real-time scheduling, but this process may not \
1611             use SCHED_FIFO (needs CAP_SYS_NICE or a non-zero RLIMIT_RTPRIO). Every IOC \
1612             thread stays at the default scheduling policy; timing is not real-time. \
1613             Logged once per process."
1614        );
1615    });
1616}
1617
1618#[cfg(target_os = "linux")]
1619fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
1620    let (min, max) = match permitted_fifo_range() {
1621        RtRange::Unsupported => return PriorityApplied::Unsupported,
1622        RtRange::Denied => {
1623            warn_rt_denied_once();
1624            return PriorityApplied::BestEffortFailed;
1625        }
1626        RtRange::Available { min, max } => (min, max),
1627    };
1628    let oss = map_epics_priority(epics_priority, min, max);
1629    let rc = set_fifo_priority(oss);
1630    if rc == 0 {
1631        PriorityApplied::Realtime
1632    } else {
1633        // The probe proved this range settable, so a failure here is not
1634        // the permission case the warning covers — keep it at debug.
1635        tracing::debug!(
1636            target: "epics_base_rs::runtime",
1637            epics_priority,
1638            oss,
1639            errno = rc,
1640            "SCHED_FIFO priority not applied; thread stays at default policy"
1641        );
1642        PriorityApplied::BestEffortFailed
1643    }
1644}
1645
1646#[cfg(target_os = "rtems")]
1647fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
1648    // Without this, every IOC thread runs at one level just above idle:
1649    // `cpukit/posix/src/pthreadattrdefault.c:49-58` sets
1650    // `inheritsched = PTHREAD_INHERIT_SCHED` in the default attribute set and
1651    // `std` never calls `pthread_attr_setinheritsched`, so a thread inherits
1652    // its creator's parameters — and every IOC thread descends from
1653    // `POSIX_Init`, which the boot shim deliberately lowers to
1654    // `RTEMS_MAXIMUM_PRIORITY - 1`. The CA receiver/sender ordering that stops
1655    // a stalled client starving command dispatch does not hold at one level.
1656    //
1657    // No range probe, unlike Linux. The probe exists there because
1658    // `sched_get_priority_max` reports the *policy's* range while
1659    // `RLIMIT_RTPRIO`/`CAP_SYS_NICE` decide the usable one, so the settable
1660    // ceiling has to be searched for. RTEMS has no such permission gate —
1661    // `pthread_setschedparam` (`cpukit/posix/src/pthreadsetschedparam.c`)
1662    // performs no privilege check — and this map's image is a fixed
1663    // `[56, 155]`, inside the measured settable `[1, 254]` by construction.
1664    // There is nothing to discover, and a probe thread would itself need a
1665    // band to run in.
1666    let oss = map_epics_priority_rtems(epics_priority);
1667    let rc = set_fifo_priority(oss);
1668    if rc == 0 {
1669        PriorityApplied::Realtime
1670    } else {
1671        tracing::debug!(
1672            target: "epics_base_rs::runtime",
1673            epics_priority,
1674            oss,
1675            errno = rc,
1676            "SCHED_FIFO priority not applied; thread stays at default policy"
1677        );
1678        PriorityApplied::BestEffortFailed
1679    }
1680}
1681
1682/// **Measurement-backed** (VxWorks 7, `x86_64-wrs-vxworks` bring-up box): no
1683/// range probe here either, and for the same reason as RTEMS —
1684/// `pthread_setschedparam` performed no privilege check there, 11 of 11
1685/// measured threads landed `PriorityApplied::Realtime`, and
1686/// [`map_epics_priority_vxworks`]'s fixed image is inside the settable range
1687/// by construction. There is nothing to discover on this target either.
1688#[cfg(target_os = "vxworks")]
1689fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
1690    let oss = map_epics_priority_vxworks(epics_priority);
1691    let rc = set_fifo_priority(oss);
1692    if rc == 0 {
1693        PriorityApplied::Realtime
1694    } else {
1695        tracing::debug!(
1696            target: "epics_base_rs::runtime",
1697            epics_priority,
1698            oss,
1699            errno = rc,
1700            "SCHED_FIFO priority not applied; thread stays at default policy"
1701        );
1702        PriorityApplied::BestEffortFailed
1703    }
1704}
1705
1706#[cfg(not(any(target_os = "linux", epics_embedded_target)))]
1707fn apply_priority_impl(_epics_priority: u8) -> PriorityApplied {
1708    // No OS-scheduler priority API is wired on other targets. The three that
1709    // are wired each needed a *measured* target band before they could be:
1710    // Linux probes for its settable ceiling at runtime, RTEMS's map is
1711    // pinned against libbsd's network-thread band measured on the bring-up
1712    // guest, and VxWorks's map is the RTEMS one's POSIX value, confirmed by
1713    // measurement on its own bring-up box. No number here is guessable, so a
1714    // new target gets `Unsupported` until somebody measures it rather than a
1715    // plausible-looking range.
1716    PriorityApplied::Unsupported
1717}
1718
1719/// How many times this process has asked the OS scheduler for SCHED_FIFO,
1720/// across every thread — including the one-off range probe.
1721///
1722/// The observable form of the opt-in guarantee: with [`RT_PRIORITY_ENV`]
1723/// unset, a process can run its whole life and this stays `0`. Also answers
1724/// "did this IOC ever actually try to go real-time?" from a log line.
1725///
1726/// Always `0` off Linux, where no scheduler call is wired at all.
1727pub fn sched_calls_made() -> usize {
1728    SCHED_CALLS_MADE.load(std::sync::atomic::Ordering::Relaxed)
1729}
1730
1731static SCHED_CALLS_MADE: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
1732
1733#[cfg(all(test, any(target_os = "linux", epics_embedded_target)))]
1734thread_local! {
1735    /// Every scheduler call this crate makes passes through
1736    /// [`set_fifo_priority`], which bumps this. Tests assert the delta is
1737    /// zero with the switch off — the property "switch off ⟹ no sched
1738    /// calls" observed directly rather than inferred from a return value.
1739    ///
1740    /// Per-thread, not global: `pthread_setschedparam` acts on the calling
1741    /// thread, so a per-thread count is the exact quantity, and a test
1742    /// cannot be perturbed by a concurrent one (the unit tests share a
1743    /// process under plain `cargo test`).
1744    static SCHED_CALLS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
1745}
1746
1747/// How many times the once-per-process denial warning was emitted.
1748#[cfg(all(test, target_os = "linux"))]
1749static DENIED_WARNINGS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
1750
1751/// Spawn a blocking closure on a dedicated thread and apply the given
1752/// EPICS [`ThreadPriority`] to that thread before running `f`.
1753///
1754/// The priority application is best effort (see
1755/// [`apply_to_current_thread`]); `f` runs regardless of whether the OS
1756/// honoured the request. This is the priority-aware counterpart of
1757/// [`spawn_blocking`] for IOC threads (CA server, scan) that a C IOC
1758/// would run in a distinct SCHED band.
1759#[cfg(tokio_backend)]
1760pub fn spawn_blocking_with_priority<F, R>(priority: ThreadPriority, f: F) -> TaskHandle<R>
1761where
1762    F: FnOnce() -> R + Send + 'static,
1763    R: Send + 'static,
1764{
1765    tokio::task::spawn_blocking(move || {
1766        let _ = enter_ioc_thread(priority);
1767        f()
1768    })
1769}
1770
1771/// RTEMS: run the blocking closure on a callback-pool worker.
1772///
1773/// The requested EPICS [`ThreadPriority`] is **not** yet mapped onto a callback
1774/// band here: the pool workers are long-lived and shared, so re-prioritising the
1775/// running worker per task would leak that priority into the next callback it
1776/// drains. The closure runs at the pool's default Medium band; mapping
1777/// `ThreadPriority` to a `CallbackPriority` band is deferred to RTEMS bring-up.
1778#[cfg(exec_backend)]
1779pub fn spawn_blocking_with_priority<F, R>(_priority: ThreadPriority, f: F) -> TaskHandle<R>
1780where
1781    F: FnOnce() -> R + Send + 'static,
1782    R: Send + 'static,
1783{
1784    use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_blocking_on};
1785    spawn_blocking_on(
1786        &background().callbacks().handle(),
1787        DEFAULT_SPAWN_PRIORITY,
1788        f,
1789    )
1790}
1791
1792/// Spawn a **dedicated OS thread** that runs `f` at `priority` with a `stack`
1793/// of [`StackSizeClass`], plus whatever ambient async context
1794/// [`block_on_sync`] needs on this target.
1795///
1796/// # Why the stack class is a parameter and not a default
1797///
1798/// C creates an IOC thread with `epicsThreadCreate(name, priority, stackSize,
1799/// fn, arg)` — three attributes. This seam carried the first two and let the
1800/// third fall through to whatever `std` picks, which is **2 MiB on RTEMS**:
1801/// `std/src/sys/thread/unix.rs` gates its `DEFAULT_MIN_STACK_SIZE` on
1802/// `not(any(l4re, vxworks, espidf, nuttx))`, and vxWorks got a 256 KiB
1803/// carve-out where RTEMS did not.
1804///
1805/// That is invisible on the host, where a thread stack is lazily-committed
1806/// virtual address space, and decisive on the target, where it is carved
1807/// eagerly out of a fixed pool. Making it a parameter is what stops a new
1808/// per-connection thread from silently costing 2 MiB: there is no default to
1809/// inherit, so every caller states what the thread is for.
1810///
1811/// Not [`spawn_blocking_with_priority`], and the difference is the point.
1812/// That one hands the closure to a *pool*: tokio's blocking pool on the host,
1813/// and on RTEMS a shared callback-pool worker that also drops the priority
1814/// (see its `exec_backend` arm). Both are right for work that finishes. A
1815/// server thread that lives as long as its connection would occupy a pool
1816/// worker for that whole time, so the pool is the wrong home for it — an IOC
1817/// thread that a C IOC would create with `epicsThreadCreate` wants a thread of
1818/// its own, and the priority a C IOC gives it.
1819///
1820/// # Why the ambient context is part of this, and not the caller's problem
1821///
1822/// `block_on_sync` picks its mechanism from the thread it is called on, but it
1823/// cannot *create* the context a future needs. On the host a fresh
1824/// `std::thread` has no runtime, so a future that spawns tasks or arms timers
1825/// panics with "there is no reactor running" the moment it is polled — even
1826/// though `block_on_sync` itself was perfectly happy to park. On RTEMS the
1827/// exec backend is process-global (`background_init`), so a bare thread is
1828/// already complete. That asymmetry is a property of the two backends, so it
1829/// is resolved here, at the seam, rather than by a `cfg` in every server that
1830/// wants a thread.
1831///
1832/// The captured context is whatever the *calling* thread is running under, so
1833/// call this from the runtime the work should belong to. When there is none
1834/// (RTEMS always; on the host a caller that is itself outside a runtime) the
1835/// thread simply runs without one, which is exactly right for a future whose
1836/// awaits are all runtime-agnostic.
1837///
1838/// # A current-thread ambient is not inherited, and that is the rule
1839///
1840/// [`RuntimeHandle::try_current`] answers two different questions with one
1841/// value: *"am I running on this runtime's thread"* and *"has this thread
1842/// merely entered this handle"*. [`block_on_sync`] cannot distinguish them, so
1843/// it must assume the first and refuse to park under a `CurrentThread` flavor —
1844/// correct on that runtime's own thread, where parking halts the task that
1845/// would wake you, and wrong on a dedicated thread, where it halts nothing.
1846///
1847/// So the dual meaning is removed here, at the one place a dedicated thread's
1848/// context is decided, rather than left for `block_on_sync` to guess: a
1849/// `CurrentThread` ambient is **not** inherited, and the thread runs with no
1850/// runtime — the `park_on` arm, which is sound for it and is the only arm RTEMS
1851/// ever takes.
1852///
1853/// Nothing is lost by declining it. What inheriting buys is stated above —
1854/// `spawn` and the timer inside `block_on_sync` — and under a `CurrentThread`
1855/// ambient `block_on_sync` returns
1856/// [`Err(CurrentThreadRuntime)`](NotBlockable::CurrentThreadRuntime), so every one of those
1857/// powers is unreachable anyway. Inheriting it can only convert a thread that
1858/// would have worked into one that cannot block at all. Measured as exactly
1859/// that: the PVA client's blocking byte pumps
1860/// (`runtime::blocking_io::spawn_pump`) are dedicated threads whose bodies are
1861/// pure `tokio::sync` channel traffic, and every `#[tokio::test]` that drives
1862/// one is `CurrentThread` by default — inheritance made the reader pump exit on
1863/// its first chunk and the connection read as "server closed during handshake".
1864#[cfg(tokio_backend)]
1865pub fn spawn_dedicated_thread<F>(
1866    name: String,
1867    priority: ThreadPriority,
1868    stack: StackSizeClass,
1869    f: F,
1870) -> std::io::Result<std::thread::JoinHandle<()>>
1871where
1872    F: FnOnce() + Send + 'static,
1873{
1874    let ambient = InheritedRuntime::capture();
1875    let charge = crate::runtime::worker_pool::ThreadCharge::fixed(stack);
1876    std::thread::Builder::new()
1877        .name(name)
1878        .stack_size(stack.bytes())
1879        .spawn(move || {
1880            // Dies with the thread, so the account tracks threads that exist.
1881            let _charge = charge;
1882            // Held for the whole body: it is what makes `tokio::spawn` and the
1883            // timer reachable from this thread, and therefore what lets a future
1884            // written for the hosted driver run unchanged under `block_on_sync`.
1885            ambient.run(move || {
1886                let _ = enter_ioc_thread(priority);
1887                f()
1888            })
1889        })
1890}
1891
1892/// The ambient async context a worker body should run under — captured on the
1893/// thread that *submitted* the work, applied on the thread that runs it.
1894///
1895/// **One owner for the question `spawn_dedicated_thread`'s docs above answer at
1896/// length.** Two callers need it and they differ in *when* they capture:
1897/// `spawn_dedicated_thread` captures once, at spawn, because the thread it
1898/// creates serves exactly one body; `runtime::worker_pool` captures per **job**,
1899/// because a pooled worker outlives the runtime that first used it. A pooled
1900/// worker that inherited its ambient at creation would hold a `Handle` to a
1901/// runtime that has since been dropped — every `#[tokio::test]` builds and drops
1902/// its own — and enter it for every later connection.
1903///
1904/// The `CurrentThread` filter is the rule stated above and must not be
1905/// re-derived: a current-thread ambient is *not* inherited, because
1906/// `block_on_sync` cannot distinguish "I am that runtime's thread" from "I have
1907/// merely entered its handle" and must refuse to park under it.
1908#[cfg(tokio_backend)]
1909pub(crate) struct InheritedRuntime(Option<tokio::runtime::Handle>);
1910
1911#[cfg(tokio_backend)]
1912impl InheritedRuntime {
1913    /// Capture the calling thread's runtime, if it is one a dedicated thread
1914    /// may enter.
1915    pub(crate) fn capture() -> Self {
1916        Self(
1917            tokio::runtime::Handle::try_current()
1918                .ok()
1919                .filter(|h| h.runtime_flavor() != RuntimeFlavor::CurrentThread),
1920        )
1921    }
1922
1923    /// Run `f` with the captured context entered for its whole duration.
1924    pub(crate) fn run<R>(&self, f: impl FnOnce() -> R) -> R {
1925        let _entered = self.0.as_ref().map(|h| h.enter());
1926        f()
1927    }
1928}
1929
1930/// RTEMS: the exec backend's spawn pool and timer are process-global, so there
1931/// is no per-thread context to capture or enter. Same shape so the callers need
1932/// no `cfg` of their own.
1933#[cfg(exec_backend)]
1934pub(crate) struct InheritedRuntime;
1935
1936#[cfg(exec_backend)]
1937impl InheritedRuntime {
1938    pub(crate) fn capture() -> Self {
1939        Self
1940    }
1941
1942    pub(crate) fn run<R>(&self, f: impl FnOnce() -> R) -> R {
1943        f()
1944    }
1945}
1946
1947/// RTEMS: a plain thread is already complete — the exec backend's spawn pool
1948/// and timer are process-global, so there is no per-thread context to enter.
1949#[cfg(exec_backend)]
1950pub fn spawn_dedicated_thread<F>(
1951    name: String,
1952    priority: ThreadPriority,
1953    stack: StackSizeClass,
1954    f: F,
1955) -> std::io::Result<std::thread::JoinHandle<()>>
1956where
1957    F: FnOnce() + Send + 'static,
1958{
1959    let charge = crate::runtime::worker_pool::ThreadCharge::fixed(stack);
1960    std::thread::Builder::new()
1961        .name(name)
1962        .stack_size(stack.bytes())
1963        .spawn(move || {
1964            let _charge = charge;
1965            let _ = enter_ioc_thread(priority);
1966            f()
1967        })
1968}
1969
1970/// A thread the IOC **cannot correctly run without** — the scan rates, the
1971/// callback bands, the delayed-callback timer, the boot script.
1972///
1973/// # Invariant
1974///
1975/// **An IOC that fails to start a mandatory thread MUST NOT continue serving.**
1976/// A thread-local panic is not that: on a `panic = "unwind"` target — and RTEMS
1977/// and VxWorks both default to unwind — `Builder::spawn(..).expect(..)` kills
1978/// only the thread that called it. Measured on a VxWorks 7 RTP on a 1 GB guest:
1979/// `EAGAIN` from the periodic-scan spawn panicked the `scan-owner` thread, the
1980/// stop guard it held unwound and stopped the rates that *had* started, and the
1981/// process went on answering CA with zero periodic scanning — a half-IOC whose
1982/// records simply never process.
1983///
1984/// C has no such state. `spawnPeriodic` (`dbScan.c:943-959`) calls
1985/// `epicsThreadCreateOpt` and then `epicsEventWait(startStopEvent)`; the event
1986/// is posted by `periodicTask` itself, so when the thread was never created
1987/// nobody posts it and `iocInit` wedges. C never reaches "serving".
1988///
1989/// # Why there is no `Result` on [`spawn`](Self::spawn)
1990///
1991/// Because there is nothing a caller could do with one that satisfies the
1992/// invariant. Every caller that is *not* inside a fallible boot step would have
1993/// to re-derive "this must be fatal" locally, and that is precisely the `.expect`
1994/// the type exists to remove. The one shape that *can* satisfy it without
1995/// aborting — a caller still inside a boot step that returns its error to the
1996/// owner that decides whether to serve — is [`try_spawn`](Self::try_spawn), and
1997/// that obligation is stated on it.
1998///
1999/// Name, band and stack class are constructor parameters for the same reason
2000/// they are on [`spawn_dedicated_thread`]: a caller cannot omit what it must
2001/// pass, so the RTEMS thread census (2 MiB default stacks, OS-anonymous
2002/// threads) is closed by signature rather than by a source sweep.
2003pub struct MandatoryThread {
2004    name: String,
2005    priority: ThreadPriority,
2006    stack: StackSizeClass,
2007}
2008
2009impl MandatoryThread {
2010    /// Declare a mandatory thread: its C thread name, the EPICS band it holds,
2011    /// and the stack class the C IOC gives it.
2012    pub fn new(name: impl Into<String>, priority: ThreadPriority, stack: StackSizeClass) -> Self {
2013        Self {
2014            name: name.into(),
2015            priority,
2016            stack,
2017        }
2018    }
2019
2020    /// Start it, or take the process down.
2021    ///
2022    /// For every caller with no error path back to whoever decides that this
2023    /// IOC serves — a constructor returning `Self`, a `OnceLock` initialiser, a
2024    /// future that parks forever. See the type docs for why this returns no
2025    /// `Result`.
2026    pub fn spawn<F>(self, f: F) -> std::thread::JoinHandle<()>
2027    where
2028        F: FnOnce() + Send + 'static,
2029    {
2030        let name = self.name.clone();
2031        match self.try_spawn(f) {
2032            Ok(handle) => handle,
2033            Err(e) => mandatory_thread_unavailable(&name, &e),
2034        }
2035    }
2036
2037    /// Start it, handing the failure to a caller that is **still inside a
2038    /// fallible boot step**.
2039    ///
2040    /// The obligation this carries: the returned error MUST reach the owner
2041    /// that decides whether the IOC serves, and that owner MUST refuse. It must
2042    /// not be unwrapped, logged-and-ignored, or turned into a warning — any of
2043    /// those re-opens exactly the half-IOC the type docs describe. Use
2044    /// [`spawn`](Self::spawn) when no such path exists.
2045    pub fn try_spawn<F>(self, f: F) -> std::io::Result<std::thread::JoinHandle<()>>
2046    where
2047        F: FnOnce() + Send + 'static,
2048    {
2049        let priority = self.priority;
2050        let charge = crate::runtime::worker_pool::ThreadCharge::fixed(self.stack);
2051        std::thread::Builder::new()
2052            .name(self.name)
2053            .stack_size(self.stack.bytes())
2054            .spawn(move || {
2055                let _charge = charge;
2056                let _ = enter_ioc_thread(priority);
2057                f()
2058            })
2059    }
2060}
2061
2062/// What the operator reads on the console when a mandatory thread could not be
2063/// created. Split out from [`mandatory_thread_unavailable`] so the wording is
2064/// testable without a process that aborts.
2065fn mandatory_thread_failure_message(name: &str, err: &std::io::Error) -> String {
2066    format!(
2067        "FATAL: the IOC could not create its mandatory `{name}` thread: {err}. \
2068         Continuing would leave this IOC answering clients while the work that \
2069         thread owns never runs, so the process is aborting instead \
2070         (C dbScan.c:943-959 wedges iocInit for the same reason)."
2071    )
2072}
2073
2074/// The single fatal exit for a mandatory thread that could not be created.
2075///
2076/// `eprintln!` and not `tracing`/`errlog`: on the RTEMS and VxWorks targets no
2077/// subscriber is installed, so a `tracing` event at this point is discarded and
2078/// the operator sees an IOC that simply went quiet. Only `eprintln!` and panic
2079/// output reach the console there.
2080///
2081/// `abort` and not `exit`: unwinding would run every other thread's destructors
2082/// against a half-built IOC, and the boot state that made the spawn fail is not
2083/// one to tear down tidily.
2084fn mandatory_thread_unavailable(name: &str, err: &std::io::Error) -> ! {
2085    eprintln!("{}", mandatory_thread_failure_message(name, err));
2086    std::process::abort()
2087}
2088
2089#[cfg(test)]
2090mod tests {
2091    use super::*;
2092
2093    /// Everything before the first column-0 `#[cfg(test)]` — the code that
2094    /// actually ships.
2095    fn production_scope(src: &str) -> &str {
2096        match src.find("\n#[cfg(test)]") {
2097            Some(i) => &src[..i],
2098            None => src,
2099        }
2100    }
2101
2102    /// Every file in this crate that creates an OS thread, as (label, source).
2103    ///
2104    /// This crate's files only. `epics-base-rs`'s two thread-creating files
2105    /// (`server/ioc_app.rs`, `server/scan.rs`) are swept by the same assertions
2106    /// in that crate's own `tests/thread_census.rs`: `include_str!` must not
2107    /// cross a crate boundary — a path outside the package directory does not
2108    /// survive `cargo publish` — so the guard was split by subject, not
2109    /// weakened.
2110    fn censused_files() -> [(&'static str, &'static str); 5] {
2111        [
2112            ("runtime/task.rs", include_str!("task.rs")),
2113            (
2114                "runtime/background/delayed_timer.rs",
2115                include_str!("background/delayed_timer.rs"),
2116            ),
2117            (
2118                "runtime/background/scan_once.rs",
2119                include_str!("background/scan_once.rs"),
2120            ),
2121            (
2122                "runtime/background/callback_executor.rs",
2123                include_str!("background/callback_executor.rs"),
2124            ),
2125            ("runtime/worker_pool.rs", include_str!("worker_pool.rs")),
2126        ]
2127    }
2128
2129    /// Every thread this crate creates states a stack size.
2130    ///
2131    /// `std` gives RTEMS the generic 2 MiB `DEFAULT_MIN_STACK_SIZE`
2132    /// (`std/src/sys/thread/unix.rs`: the carve-out list names vxworks, l4re,
2133    /// espidf and nuttx — not rtems). On the host that is lazily-committed
2134    /// address space and costs nothing measurable; on the target it is carved
2135    /// eagerly out of a fixed pool, which is why an unset stack size is the
2136    /// first ceiling the IOC hits rather than a rounding error.
2137    ///
2138    /// `spawn_dedicated_thread` and [`MandatoryThread`] are enforced by their
2139    /// signatures — the class is a parameter, so a caller cannot omit it. This
2140    /// covers the threads that still build a `std::thread::Builder` directly.
2141    ///
2142    /// It also bans the API that has no class to state:
2143    /// `std::thread::spawn` cannot express a stack size at all, so a site
2144    /// using it does not fail the `Builder` check above — it is invisible to
2145    /// it. Same defect, different anchor. (The bare `thread::spawn` sites
2146    /// elsewhere in the workspace — `ca::repeater`, `ca::calink`,
2147    /// `ca::server::ca_server`, `pva::server::pva_server`, `bridge::pvalink` —
2148    /// are distinct twice over: none is a mandatory IOC thread, and all but the
2149    /// per-command link helpers sit behind `#[cfg(not(target_os = "rtems"))]`
2150    /// module gates, so they are not in the RTEMS closure at all. Every file
2151    /// listed here is.)
2152    ///
2153    /// Fails today, on Linux, with no cross toolchain.
2154    #[test]
2155    fn every_thread_in_this_crate_states_a_stack_size() {
2156        let mut unclassified = Vec::new();
2157        let mut checked = 0usize;
2158        for (label, src) in censused_files() {
2159            let prod = production_scope(src);
2160            for (n, after) in prod.split("thread::Builder::new()").skip(1).enumerate() {
2161                checked += 1;
2162                // The class must be set before the closure is handed over;
2163                // `.spawn(` ends the builder chain.
2164                let chain = after.split(".spawn(").next().unwrap_or("");
2165                if !chain.contains(".stack_size(") {
2166                    unclassified.push(format!("{label} (Builder #{})", n + 1));
2167                }
2168            }
2169            // The classless API. Split so this guard does not match its own
2170            // needle in the file it is written in.
2171            let bare = concat!("thread", "::spawn(");
2172            for (n, line) in prod.lines().enumerate() {
2173                let t = line.trim_start();
2174                if t.starts_with("//") {
2175                    continue;
2176                }
2177                if t.contains(bare) && !t.contains("Builder") {
2178                    unclassified.push(format!("{label}:{} (bare spawn)", n + 1));
2179                }
2180            }
2181        }
2182
2183        // Five: `spawn_dedicated_thread`'s two `cfg` arms, `MandatoryThread`,
2184        // the RT-policy probe, and `worker_pool`'s pooled worker. The floor was
2185        // seven until the three background facilities moved onto
2186        // `MandatoryThread`, which states the class in its constructor —
2187        // `every_background_facility_thread_is_mandatory` is what keeps that
2188        // move from being a hole rather than a hand-off.
2189        assert!(
2190            checked >= 5,
2191            "expected to find the crate's Builder sites, found {checked} — \
2192             did a file move? update this guard's file list"
2193        );
2194        assert!(
2195            unclassified.is_empty(),
2196            "these threads inherit std's 2 MiB default on RTEMS: {unclassified:?}"
2197        );
2198    }
2199
2200    /// The three background facilities create their threads through
2201    /// [`MandatoryThread`], and nothing else.
2202    ///
2203    /// Each of them — the callback bands, `cbTimer`, `scanOnce` — is a thread
2204    /// the IOC cannot correctly run without, and each is created from a
2205    /// constructor reached through a `OnceLock` initialiser, so there is no
2206    /// error path back to whoever decided this IOC serves. They used to resolve
2207    /// the spawn `Result` with `.expect`, which on a `panic = "unwind"` target
2208    /// (RTEMS and VxWorks both default to unwind) killed only the thread that
2209    /// happened to touch the facility first and left the IOC serving without
2210    /// the band, the timer or the `scanOnce` worker.
2211    ///
2212    /// The ban is the structural half: with no raw `Builder` and no bare
2213    /// `thread::spawn` in these files, "mandatory" is not a property a new
2214    /// thread here can forget to declare.
2215    #[test]
2216    fn every_background_facility_thread_is_mandatory() {
2217        let bare = concat!("thread", "::spawn(");
2218        let mut strays = Vec::new();
2219        let mut owned = 0usize;
2220        for (label, src) in censused_files() {
2221            if !label.contains("/background/") {
2222                continue;
2223            }
2224            for (n, line) in production_scope(src).lines().enumerate() {
2225                let t = line.trim_start();
2226                if t.starts_with("//") {
2227                    continue;
2228                }
2229                if t.contains("MandatoryThread::new(") {
2230                    owned += 1;
2231                }
2232                if t.contains("thread::Builder::new()") {
2233                    strays.push(format!("{label}:{} (raw Builder)", n + 1));
2234                }
2235                if t.contains(bare) && !t.contains("Builder") {
2236                    strays.push(format!("{label}:{} (bare spawn)", n + 1));
2237                }
2238            }
2239        }
2240        assert!(
2241            strays.is_empty(),
2242            "a facility thread created outside `MandatoryThread` resolves its \
2243             own spawn failure, and the only resolution that keeps this IOC \
2244             honest is not serving: {strays:?}"
2245        );
2246        assert!(
2247            owned >= 3,
2248            "expected the callback pool, `cbTimer` and `scanOnce`, found {owned} \
2249             `MandatoryThread` sites — did a file move? update the census list"
2250        );
2251    }
2252
2253    #[epics_macros_rs::epics_test]
2254    async fn test_spawn() {
2255        let handle = spawn(async { 42 });
2256        assert_eq!(handle.await.unwrap(), 42);
2257    }
2258
2259    #[epics_macros_rs::epics_test]
2260    async fn test_spawn_blocking() {
2261        let handle = spawn_blocking(|| 123);
2262        assert_eq!(handle.await.unwrap(), 123);
2263    }
2264
2265    /// The property `spawn_dedicated_thread` exists for. A future written for
2266    /// the hosted driver — one that spawns a task and arms a timer — must run
2267    /// unchanged on the thread this hands back. On a plain `std::thread` it
2268    /// does not: it panics with "there is no reactor running" as soon as it is
2269    /// polled, however willing `block_on_sync` was to park.
2270    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
2271    async fn a_dedicated_thread_carries_the_ambient_runtime() {
2272        let (tx, rx) = std::sync::mpsc::channel();
2273        let joined = spawn_dedicated_thread(
2274            "dedicated-with-runtime".into(),
2275            ThreadPriority::CaServerLow,
2276            StackSizeClass::Small,
2277            move || {
2278                let outcome = block_on_sync(async {
2279                    let inner = spawn(async { 7u32 }).await.expect("inner task");
2280                    sleep(Duration::from_millis(1)).await;
2281                    inner
2282                });
2283                let _ = tx.send((
2284                    std::thread::current().name().map(str::to_string),
2285                    outcome.ok(),
2286                ));
2287            },
2288        )
2289        .expect("dedicated thread spawned");
2290
2291        let (name, value) = rx
2292            .recv_timeout(Duration::from_secs(5))
2293            .expect("the dedicated thread must complete, not panic");
2294        assert_eq!(name.as_deref(), Some("dedicated-with-runtime"));
2295        assert_eq!(
2296            value,
2297            Some(7),
2298            "a spawn and a timer must both work on the dedicated thread"
2299        );
2300        joined.join().expect("dedicated thread joined");
2301    }
2302
2303    /// The third boundary, and the one that was missing: a **current-thread**
2304    /// ambient runtime.
2305    ///
2306    /// The two neighbours below and above cover "multi-thread ambient" and "no
2307    /// ambient". This is the case between them, and inheriting the handle there
2308    /// is what made `block_on_sync` return `NotBlockable` on a thread that was
2309    /// perfectly able to park — silently, since every caller reads the refusal
2310    /// as "the connection ended". `#[tokio::test]` is `CurrentThread` by
2311    /// default, so this is also the flavor most of the workspace's tests hand a
2312    /// dedicated thread.
2313    ///
2314    /// The assertion is on `block_on_sync` succeeding, not on the absence of a
2315    /// handle, because being able to block is the property the thread is spawned
2316    /// for; how that is arranged is this function's business.
2317    #[epics_macros_rs::epics_test]
2318    async fn a_dedicated_thread_can_block_under_a_current_thread_ambient() {
2319        let (tx, rx) = std::sync::mpsc::channel();
2320        let joined = spawn_dedicated_thread(
2321            "dedicated-current-thread-ambient".into(),
2322            ThreadPriority::Low,
2323            StackSizeClass::Small,
2324            move || {
2325                // A runtime-agnostic await, the only kind a parking thread may
2326                // use — and the exact shape both blocking-io pumps run.
2327                let (ctx, crx) = tokio::sync::mpsc::channel::<u32>(1);
2328                let outcome = block_on_sync(async move {
2329                    ctx.send(9u32).await.expect("send into a depth-1 channel");
2330                    let mut crx = crx;
2331                    crx.recv().await
2332                });
2333                let _ = tx.send(outcome.ok().flatten());
2334            },
2335        )
2336        .expect("dedicated thread spawned");
2337
2338        assert_eq!(
2339            rx.recv_timeout(Duration::from_secs(5))
2340                .expect("the dedicated thread must complete, not panic"),
2341            Some(9),
2342            "a dedicated thread must be able to park under a current-thread \
2343             ambient runtime; inheriting that handle makes block_on_sync \
2344             refuse and every pump built on it exit at once"
2345        );
2346        joined.join().expect("dedicated thread joined");
2347    }
2348
2349    /// The other boundary: no runtime to capture. The thread still runs, and a
2350    /// runtime-agnostic await still completes — that is `park_on`, and it is
2351    /// the only arm RTEMS ever takes.
2352    #[test]
2353    fn a_dedicated_thread_runs_without_an_ambient_runtime() {
2354        let (tx, rx) = std::sync::mpsc::channel();
2355        let joined = spawn_dedicated_thread(
2356            "dedicated-no-runtime".into(),
2357            ThreadPriority::Low,
2358            StackSizeClass::Small,
2359            move || {
2360                let _ = tx.send((
2361                    std::thread::current().name().map(str::to_string),
2362                    block_on_sync(async { 5u32 }).ok(),
2363                ));
2364            },
2365        )
2366        .expect("dedicated thread spawned");
2367
2368        let (name, value) = rx
2369            .recv_timeout(Duration::from_secs(5))
2370            .expect("the dedicated thread must run with no runtime to capture");
2371        assert_eq!(name.as_deref(), Some("dedicated-no-runtime"));
2372        assert_eq!(value, Some(5));
2373        joined.join().expect("dedicated thread joined");
2374    }
2375
2376    // --- The band-blocking invariant (doc/pvalink-rtems-design.md §2.3) ------
2377    //
2378    // MUST NOT: work running on a background-facility worker thread — a
2379    // callback band, the delayed timer, the scanOnce worker — block that
2380    // thread on async progress. The gate is `block_on_sync`; the mark is set
2381    // by `background::facility::run_facility_loop`, the one function every
2382    // worker loop goes through.
2383    //
2384    // Each of the three cases below is written so that a *broken* gate fails
2385    // the test instead of hanging it: the awaited future is completable from
2386    // the test thread, so a worker that parked can always be released before
2387    // the assertion runs and the pool's `Drop` can still join it.
2388
2389    /// The case the invariant exists for: a future spawned onto a callback
2390    /// band. On RTEMS this is exactly what [`spawn`] produces, and the band has
2391    /// one worker — parking it stops every deferred callback, every FLNK tail
2392    /// and every other monitor on that band.
2393    #[test]
2394    fn a_future_on_a_callback_band_is_refused_a_blocking_bridge() {
2395        use crate::runtime::background::callback_executor::CallbackPool;
2396        use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_future};
2397
2398        let pool = CallbackPool::new();
2399        // Held by the test: `recv()` never completes until we send, so a gate
2400        // that does not refuse leaves the worker parked here.
2401        let (release, mut park_here) = tokio::sync::mpsc::channel::<()>(1);
2402        let (report, outcome) = std::sync::mpsc::channel();
2403
2404        let _handle = spawn_future(&pool.handle(), DEFAULT_SPAWN_PRIORITY, async move {
2405            let _ = report.send(block_on_sync(async move { park_here.recv().await }));
2406        });
2407
2408        let got = outcome.recv_timeout(Duration::from_secs(5));
2409        // Release a worker the gate failed to protect, so the assertions below
2410        // report a failure instead of hanging `CallbackPool::drop`'s join.
2411        let _ = release.try_send(());
2412
2413        match got {
2414            Ok(result) => assert_eq!(
2415                result.map(|v| v.is_some()),
2416                Err(NotBlockable::BackgroundWorker),
2417                "a band worker must be refused the blocking bridge, not given one"
2418            ),
2419            Err(_) => panic!(
2420                "the band worker parked inside block_on_sync instead of being \
2421                 refused — the band has one worker, so this is the deadlock the \
2422                 invariant exists to prevent"
2423            ),
2424        }
2425    }
2426
2427    /// The same thread, reached the other way: `spawn_blocking` also lands on a
2428    /// band worker under the exec backend, and a blocking closure holds that
2429    /// worker for its whole run. The rule is a property of the thread, so it
2430    /// must not depend on which spawn put the work there.
2431    #[test]
2432    fn a_blocking_closure_on_a_callback_band_is_refused_too() {
2433        use crate::runtime::background::callback_executor::CallbackPool;
2434        use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_blocking_on};
2435
2436        let pool = CallbackPool::new();
2437        let (release, mut park_here) = tokio::sync::mpsc::channel::<()>(1);
2438        let (report, outcome) = std::sync::mpsc::channel();
2439
2440        let _handle = spawn_blocking_on(&pool.handle(), DEFAULT_SPAWN_PRIORITY, move || {
2441            let _ = report.send(block_on_sync(async move { park_here.recv().await }));
2442        });
2443
2444        let got = outcome.recv_timeout(Duration::from_secs(5));
2445        let _ = release.try_send(());
2446
2447        match got {
2448            Ok(result) => assert_eq!(
2449                result.map(|v| v.is_some()),
2450                Err(NotBlockable::BackgroundWorker),
2451                "the refusal keys on the thread, not on how work reached it"
2452            ),
2453            Err(_) => panic!("the band worker parked instead of being refused"),
2454        }
2455    }
2456
2457    /// The other side of the boundary, so the gate cannot be satisfied by
2458    /// refusing everything: an ordinary thread that merely *submits* to the
2459    /// pool still blocks. The mark covers the worker loop's own thread and
2460    /// nothing else.
2461    #[test]
2462    fn a_thread_that_only_submits_to_a_band_still_blocks() {
2463        use crate::runtime::background::callback_executor::{CallbackPool, CallbackPriority};
2464
2465        let pool = CallbackPool::new();
2466        let (tx, rx) = std::sync::mpsc::channel();
2467        pool.request(
2468            CallbackPriority::Medium,
2469            Box::new(move || tx.send(1u32).unwrap()),
2470        )
2471        .expect("the band accepts the callback");
2472        assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), 1);
2473        assert_eq!(
2474            block_on_sync(async { 5u32 }),
2475            Ok(5),
2476            "the submitting thread runs no facility loop, so it may still park"
2477        );
2478    }
2479
2480    #[epics_macros_rs::epics_test]
2481    async fn test_sleep() {
2482        let start = std::time::Instant::now();
2483        sleep(Duration::from_millis(10)).await;
2484        assert!(start.elapsed() >= Duration::from_millis(10));
2485    }
2486
2487    // The two halves of `timeout`'s contract. They read as trivial against a
2488    // tokio delegation, and that is the point: they are what a later backend
2489    // swap has to keep true, on a seam whose whole purpose is to be
2490    // reimplemented.
2491    #[epics_macros_rs::epics_test]
2492    async fn timeout_yields_the_value_when_the_future_finishes_first() {
2493        let r = timeout(Duration::from_secs(30), async { 42 }).await;
2494        assert_eq!(r.unwrap(), 42);
2495    }
2496
2497    #[epics_macros_rs::epics_test]
2498    async fn timeout_elapses_on_a_future_that_never_finishes() {
2499        let r = timeout(Duration::from_millis(10), std::future::pending::<()>()).await;
2500        assert!(r.is_err());
2501    }
2502
2503    #[test]
2504    fn priority_named_levels_match_epics_thread_h() {
2505        // epicsThread.h:73-83 named-level constants.
2506        assert_eq!(ThreadPriority::Low.value(), 10);
2507        assert_eq!(ThreadPriority::CaServerLow.value(), 20);
2508        assert_eq!(ThreadPriority::CaServerHigh.value(), 40);
2509        assert_eq!(ThreadPriority::Medium.value(), 50);
2510        assert_eq!(ThreadPriority::ScanLow.value(), 60);
2511        assert_eq!(ThreadPriority::ScanHigh.value(), 70);
2512        assert_eq!(ThreadPriority::High.value(), 90);
2513        assert_eq!(ThreadPriority::Iocsh.value(), 91);
2514    }
2515
2516    #[test]
2517    fn priority_ordering_ca_server_below_scan() {
2518        // Real-time invariant: scan threads must outrank CA-server
2519        // threads so scans preempt the CA server on a loaded IOC.
2520        assert!(ThreadPriority::CaServerHigh.value() < ThreadPriority::ScanLow.value());
2521        assert!(ThreadPriority::CaServerLow.value() < ThreadPriority::ScanLow.value());
2522    }
2523
2524    #[test]
2525    fn priority_custom_clamps_to_max() {
2526        assert_eq!(ThreadPriority::Custom(200).value(), PRIORITY_MAX);
2527        assert_eq!(ThreadPriority::Custom(99).value(), 99);
2528        assert_eq!(ThreadPriority::Custom(0).value(), PRIORITY_MIN);
2529    }
2530
2531    #[test]
2532    fn stack_size_classes_ordered() {
2533        // STACK_SIZE table is strictly increasing Small < Medium < Big.
2534        assert!(StackSizeClass::Small.bytes() < StackSizeClass::Medium.bytes());
2535        assert!(StackSizeClass::Medium.bytes() < StackSizeClass::Big.bytes());
2536        // Small = 0x10000 * sizeof(usize).
2537        assert_eq!(
2538            StackSizeClass::Small.bytes(),
2539            0x10000 * std::mem::size_of::<usize>()
2540        );
2541    }
2542
2543    /// The three classes against the C table, factor by factor.
2544    ///
2545    /// `STACK_SIZE(f) = f * 0x10000 * sizeof(void*)` with factors 1, 2, 4
2546    /// (`libcom/src/osi/os/posix/osdThread.c:506-509`) — the file a C IOC on
2547    /// RTEMS 6 compiles, because `configure/toolchain.c:29-35` picks
2548    /// `OS_API = posix` for `__RTEMS_MAJOR__ >= 5`. Pinning the factors
2549    /// separately from the unit is what makes a silent edit of one of them
2550    /// fail: `stack_size_classes_ordered` above is satisfied by any
2551    /// increasing triple.
2552    #[test]
2553    fn the_classes_are_the_c_posix_table_factor_for_factor() {
2554        let unit = 0x10000 * std::mem::size_of::<usize>();
2555        assert_eq!(StackSizeClass::Small.bytes(), unit);
2556        assert_eq!(StackSizeClass::Medium.bytes(), 2 * unit);
2557        assert_eq!(StackSizeClass::Big.bytes(), 4 * unit);
2558        // And what the same table yields on the target the RTEMS port builds
2559        // for (`sizeof(void*) == 4`), spelled out so a reader on a 64-bit
2560        // host does not have to re-derive it.
2561        const TARGET_UNIT: usize = 0x10000 * 4;
2562        assert_eq!(
2563            [TARGET_UNIT, 2 * TARGET_UNIT, 4 * TARGET_UNIT],
2564            [256 * 1024, 512 * 1024, 1024 * 1024],
2565            "armv7-rtems-eabihf: Small / Medium / Big in bytes"
2566        );
2567    }
2568
2569    /// The stack a caller *states* is the stack the thread *reports*.
2570    ///
2571    /// The source guard above only proves a number reached the builder. This
2572    /// asks the running thread what it actually got, through
2573    /// `pthread_getattr_np`, and that is the property the RTEMS ceiling
2574    /// depends on: `std` gates its 2 MiB `DEFAULT_MIN_STACK_SIZE` on a
2575    /// carve-out list that omits rtems, so a size that fails to arrive is
2576    /// silently 2 MiB rather than an error.
2577    ///
2578    /// The mechanism this exercises is not host-specific: `std`'s
2579    /// `Thread::new` calls `pthread_attr_setstacksize(attr, max(stack,
2580    /// PTHREAD_STACK_MIN))` on every non-espidf/nuttx unix
2581    /// (`std/src/sys/thread/unix.rs`), and `libc` gives rtems
2582    /// `PTHREAD_STACK_MIN = 0`, so the `max` cannot raise our request there
2583    /// either. Glibc-only because `pthread_getattr_np` is the readback API.
2584    #[cfg(all(target_os = "linux", target_env = "gnu"))]
2585    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
2586    async fn a_dedicated_thread_reports_the_stack_it_was_asked_for() {
2587        fn reported_stack_bytes() -> usize {
2588            unsafe {
2589                let mut attr: libc::pthread_attr_t = std::mem::zeroed();
2590                assert_eq!(
2591                    libc::pthread_getattr_np(libc::pthread_self(), &mut attr),
2592                    0,
2593                    "pthread_getattr_np"
2594                );
2595                let mut addr: *mut libc::c_void = std::ptr::null_mut();
2596                let mut size: libc::size_t = 0;
2597                assert_eq!(
2598                    libc::pthread_attr_getstack(&attr, &mut addr, &mut size),
2599                    0,
2600                    "pthread_attr_getstack"
2601                );
2602                libc::pthread_attr_destroy(&mut attr);
2603                size
2604            }
2605        }
2606
2607        for class in [
2608            StackSizeClass::Small,
2609            StackSizeClass::Medium,
2610            StackSizeClass::Big,
2611        ] {
2612            let (tx, rx) = std::sync::mpsc::channel();
2613            let joined = spawn_dedicated_thread(
2614                format!("stack-readback-{class:?}"),
2615                ThreadPriority::CaServerLow,
2616                class,
2617                move || {
2618                    let _ = tx.send(reported_stack_bytes());
2619                },
2620            )
2621            .expect("dedicated thread spawned");
2622            let got = rx.recv().expect("the thread reported its stack");
2623            joined.join().expect("thread joined");
2624
2625            let asked = class.bytes();
2626            // The kernel rounds up to a page; it must never round *down*, and
2627            // it must not silently substitute something of a different order.
2628            assert!(
2629                got >= asked && got < asked + 64 * 1024,
2630                "{class:?}: asked for {asked} bytes, thread reports {got}"
2631            );
2632        }
2633    }
2634
2635    /// The distinguishing half of the readback: a class below `std`'s default
2636    /// must land below it. Without this the test above would pass on a
2637    /// platform that ignored every request and handed out 2 MiB, for the two
2638    /// classes that happen to be smaller than that on a 64-bit host.
2639    #[cfg(all(target_os = "linux", target_env = "gnu"))]
2640    #[test]
2641    fn the_small_classes_are_below_the_default_that_would_mask_a_failure() {
2642        const STD_DEFAULT_MIN_STACK_SIZE: usize = 2 * 1024 * 1024;
2643        assert!(StackSizeClass::Small.bytes() < STD_DEFAULT_MIN_STACK_SIZE);
2644        assert!(StackSizeClass::Medium.bytes() < STD_DEFAULT_MIN_STACK_SIZE);
2645    }
2646
2647    #[test]
2648    fn apply_priority_returns_a_defined_outcome() {
2649        // The result depends on the platform + permissions of the test
2650        // host; we only assert it is one of the defined outcomes and
2651        // does not panic. On a CI box without CAP_SYS_NICE this is
2652        // typically BestEffortFailed — which is C-parity behaviour.
2653        let outcome = apply_to_current_thread(ThreadPriority::ScanHigh);
2654        assert!(matches!(
2655            outcome,
2656            PriorityApplied::Realtime
2657                | PriorityApplied::Disabled
2658                | PriorityApplied::Unsupported
2659                | PriorityApplied::BestEffortFailed
2660        ));
2661    }
2662
2663    /// Both defaults, and both override directions against each default.
2664    ///
2665    /// The RTEMS arm is unreachable from a host test run unless the default
2666    /// is a function of the target rather than a `cfg` block, which is why
2667    /// `default_policy`/`resolve` take their input explicitly.
2668    #[test]
2669    fn the_rt_default_is_on_for_rtems_and_off_for_hosted() {
2670        // (1) the two defaults themselves
2671        assert_eq!(
2672            default_policy(true),
2673            RtPolicy::AllowRealtime,
2674            "RTEMS honours its priorities by default, as base does \
2675             (EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING=YES, CONFIG_ENV:57)"
2676        );
2677        assert_eq!(
2678            default_policy(false),
2679            RtPolicy::Disabled,
2680            "hosted stays opt-in: RLIMIT_RTPRIO makes the request fail on a \
2681             desktop, and where it succeeds a runaway band wedges the machine"
2682        );
2683
2684        // (2) the compiled-in default is wired to the target, not to a guess
2685        assert_eq!(DEFAULT_POLICY, default_policy(cfg!(epics_embedded_target)));
2686        assert_eq!(RtPolicy::from_env_value(None), DEFAULT_POLICY);
2687
2688        // (3) an explicit value wins over EITHER default, in BOTH directions.
2689        //     The RTEMS-off case is the one an operator needs: turning RT
2690        //     scheduling off on a target that defaults to on.
2691        for default in [RtPolicy::AllowRealtime, RtPolicy::Disabled] {
2692            assert_eq!(
2693                RtPolicy::resolve(Some("NO"), default),
2694                RtPolicy::Disabled,
2695                "explicit NO must turn it off even where the default is {default:?}"
2696            );
2697            assert_eq!(
2698                RtPolicy::resolve(Some("YES"), default),
2699                RtPolicy::AllowRealtime,
2700                "explicit YES must turn it on even where the default is {default:?}"
2701            );
2702            assert_eq!(
2703                RtPolicy::resolve(None, default),
2704                default,
2705                "unset must resolve to the default and nothing else"
2706            );
2707        }
2708    }
2709
2710    #[test]
2711    fn rt_switch_explicit_values_win_over_the_default() {
2712        // Unset takes the target's default; that is
2713        // `the_rt_default_is_on_for_rtems_and_off_for_hosted`'s subject.
2714        assert_eq!(RtPolicy::from_env_value(None), DEFAULT_POLICY);
2715        // C's `envGetBoolConfigParam` (envSubr.c:331) accepts only
2716        // case-insensitive "yes"; we also take the spellings a hand-written
2717        // startup script is likely to use.
2718        for on in ["YES", "yes", "Yes", "true", "TRUE", "on", "1", " yes "] {
2719            assert_eq!(
2720                RtPolicy::from_env_value(Some(on)),
2721                RtPolicy::AllowRealtime,
2722                "{on:?} should turn the switch on"
2723            );
2724        }
2725        // Everything else is off. Silence is the safe direction: a
2726        // misspelling must never grant a process RT scheduling.
2727        for off in ["", "NO", "no", "false", "off", "0", "y", "yes please", "2"] {
2728            assert_eq!(
2729                RtPolicy::from_env_value(Some(off)),
2730                RtPolicy::Disabled,
2731                "{off:?} should leave the switch off"
2732            );
2733        }
2734    }
2735
2736    /// Switch off ⟹ the OS scheduler is never called.
2737    ///
2738    /// Mutation check: deleting the `RtPolicy::Disabled` arm in
2739    /// `apply_to_current_thread_under` (so it always calls
2740    /// `apply_priority_impl`) makes the `SCHED_CALLS` assertion fail.
2741    #[cfg(target_os = "linux")]
2742    #[test]
2743    fn switch_off_makes_no_scheduler_calls() {
2744        let before = SCHED_CALLS.with(std::cell::Cell::get);
2745        for p in [
2746            ThreadPriority::Low,
2747            ThreadPriority::CaServerLow,
2748            ThreadPriority::ScanHigh,
2749            ThreadPriority::Iocsh,
2750            ThreadPriority::Custom(0),
2751            ThreadPriority::Custom(99),
2752        ] {
2753            assert_eq!(
2754                apply_to_current_thread_under(RtPolicy::Disabled, p),
2755                PriorityApplied::Disabled
2756            );
2757        }
2758        assert_eq!(
2759            SCHED_CALLS.with(std::cell::Cell::get),
2760            before,
2761            "the switch is off, so nothing may reach pthread_setschedparam"
2762        );
2763    }
2764
2765    /// Switch on: either the host grants SCHED_FIFO — and then the policy
2766    /// must actually be in force on the thread, at the mapped priority — or
2767    /// it does not, and the thread keeps running under the default policy.
2768    ///
2769    /// Runs on its own thread: on a host that *does* grant RT, leaving the
2770    /// test-harness thread in a real-time band would outlive the test.
2771    #[cfg(target_os = "linux")]
2772    #[test]
2773    fn switch_on_either_sticks_or_falls_back_without_killing_the_thread() {
2774        let outcome = std::thread::spawn(|| {
2775            // Resolve (and cache) the probe first, so what the call below is
2776            // expected to do is known rather than order-dependent.
2777            let range = permitted_fifo_range();
2778            let before = SCHED_CALLS.with(std::cell::Cell::get);
2779            let outcome =
2780                apply_to_current_thread_under(RtPolicy::AllowRealtime, ThreadPriority::ScanHigh);
2781            let calls = SCHED_CALLS.with(std::cell::Cell::get) - before;
2782
2783            // Whatever the host allowed, this thread is still running.
2784            assert_eq!(2 + 2, 4);
2785
2786            let mut policy = 0i32;
2787            let mut param = libc::sched_param { sched_priority: 0 };
2788            // SAFETY: reads the calling thread's own scheduling into
2789            // stack-local outputs.
2790            let rc = unsafe {
2791                libc::pthread_getschedparam(libc::pthread_self(), &mut policy, &mut param)
2792            };
2793            assert_eq!(rc, 0, "pthread_getschedparam failed");
2794
2795            match (range, outcome) {
2796                (RtRange::Available { min, max }, PriorityApplied::Realtime) => {
2797                    // The host permits FIFO — assert the policy stuck, at
2798                    // the C-mapped priority, off exactly one scheduler call.
2799                    assert_eq!(calls, 1, "one apply must be one scheduler call");
2800                    assert_eq!(policy, libc::SCHED_FIFO, "SCHED_FIFO did not stick");
2801                    assert_eq!(
2802                        param.sched_priority,
2803                        map_epics_priority(ThreadPriority::ScanHigh.value(), min, max),
2804                        "wrong OS priority for epicsThreadPriorityScanHigh"
2805                    );
2806                }
2807                (RtRange::Denied, PriorityApplied::BestEffortFailed) => {
2808                    // Unprivileged: the fallback leaves the thread at the
2809                    // default policy rather than failing the caller, and —
2810                    // the anti-spam property — asks the OS nothing further
2811                    // now that the probe has settled the question once.
2812                    assert_eq!(calls, 0, "a settled denial must not re-ask the OS");
2813                    assert_ne!(
2814                        policy,
2815                        libc::SCHED_FIFO,
2816                        "fallback reported but the thread is real-time scheduled"
2817                    );
2818                }
2819                (RtRange::Unsupported, PriorityApplied::Unsupported) => {
2820                    assert_eq!(calls, 0, "no SCHED_FIFO range means no scheduler call");
2821                }
2822                (range, outcome) => {
2823                    panic!("range {range:?} and outcome {outcome:?} disagree")
2824                }
2825            }
2826            outcome
2827        })
2828        .join()
2829        .expect("probe thread panicked");
2830        eprintln!("host RT outcome: {outcome:?}");
2831    }
2832
2833    /// The unprivileged fallback is logged once, not once per thread.
2834    #[cfg(target_os = "linux")]
2835    #[test]
2836    fn denial_is_reported_once_not_per_thread() {
2837        let threads: Vec<_> = (0..8)
2838            .map(|_| {
2839                std::thread::spawn(|| {
2840                    for _ in 0..8 {
2841                        let _ = apply_to_current_thread_under(
2842                            RtPolicy::AllowRealtime,
2843                            ThreadPriority::Low,
2844                        );
2845                    }
2846                })
2847            })
2848            .collect();
2849        for t in threads {
2850            t.join().expect("worker panicked");
2851        }
2852        assert!(
2853            DENIED_WARNINGS.load(std::sync::atomic::Ordering::Relaxed) <= 1,
2854            "64 denied requests across 8 threads must not produce more than one warning"
2855        );
2856    }
2857
2858    /// The mapping itself, against `epicsThreadGetPosixPriority`
2859    /// (`osdThread.c:129-144`).
2860    #[cfg(target_os = "linux")]
2861    #[test]
2862    fn epics_priority_maps_onto_the_permitted_fifo_range() {
2863        // Linux's nominal SCHED_FIFO range.
2864        let (min, max) = (1, 99);
2865        // oss = p * (max-min)/100 + min
2866        assert_eq!(map_epics_priority(0, min, max), 1);
2867        assert_eq!(map_epics_priority(20, min, max), 1 + (20.0 * 0.98) as i32);
2868        assert_eq!(map_epics_priority(99, min, max), 1 + (99.0 * 0.98) as i32);
2869        // Ordering is preserved: the CA server sits below the scan bands.
2870        assert!(
2871            map_epics_priority(ThreadPriority::CaServerHigh.value(), min, max)
2872                < map_epics_priority(ThreadPriority::ScanLow.value(), min, max)
2873        );
2874        // A range restricted by RLIMIT_RTPRIO still spans the whole EPICS
2875        // space rather than saturating at the top.
2876        assert_eq!(map_epics_priority(0, 1, 10), 1);
2877        assert_eq!(map_epics_priority(99, 1, 10), 1 + (99.0 * 0.09) as i32);
2878        // Degenerate range collapses (osdThread.c:133).
2879        assert_eq!(map_epics_priority(50, 7, 7), 7);
2880    }
2881
2882    /// The RTEMS map's *image* is the whole point of choosing it, so assert
2883    /// the image, not sampled points: for **every** `u8` input the resulting
2884    /// RTEMS core priority lands in `100..=199`, and therefore below libbsd's
2885    /// network threads.
2886    ///
2887    /// Provenance of the band, measured on the bring-up guest (RTEMS 6 +
2888    /// libbsd, QEMU `xilinx_zynq_a9`) — lower core number is *more* urgent:
2889    ///
2890    /// | core | thread |
2891    /// |------|--------|
2892    /// | 96   | libbsd `IRQS` (interrupt server) |
2893    /// | 98   | libbsd `TIME` |
2894    /// | 100  | libbsd default — twelve further network threads |
2895    /// | 254  | DHCP, outside the band |
2896    /// | 255  | idle, `RTEMS_MAXIMUM_PRIORITY` |
2897    ///
2898    /// So `core >= 100` means: never more urgent than any libbsd network
2899    /// thread, and strictly less urgent than `IRQS`/`TIME`. That is a
2900    /// property of the map's construction (fixed offsets over a 100-wide
2901    /// EPICS space), not of the endpoints, which is why no input — including
2902    /// the out-of-range ones `ThreadPriority::value` cannot currently produce
2903    /// — can escape it.
2904    #[test]
2905    fn rtems_priority_map_stays_below_the_libbsd_network_band() {
2906        /// libbsd's most urgent network thread on the measured guest.
2907        const LIBBSD_IRQS_CORE: i32 = 96;
2908        /// libbsd's default band; twelve of its threads sit here.
2909        const LIBBSD_DEFAULT_CORE: i32 = 100;
2910        // The guest's settable SCHED_FIFO range.
2911        const POSIX_MIN: i32 = 1;
2912        const POSIX_MAX: i32 = 254;
2913
2914        for epics in 0..=u8::MAX {
2915            let posix = map_epics_priority_rtems(epics);
2916            let core = RTEMS_MAXIMUM_PRIORITY - posix;
2917            assert!(
2918                (POSIX_MIN..=POSIX_MAX).contains(&posix),
2919                "EPICS {epics} maps to posix {posix}, outside the settable \
2920                 [{POSIX_MIN}, {POSIX_MAX}]"
2921            );
2922            assert!(
2923                core >= LIBBSD_DEFAULT_CORE,
2924                "EPICS {epics} maps to core {core}, more urgent than libbsd's \
2925                 default band ({LIBBSD_DEFAULT_CORE}) and its IRQS \
2926                 ({LIBBSD_IRQS_CORE})"
2927            );
2928            assert!(
2929                core <= RTEMS_MAXIMUM_PRIORITY - 56,
2930                "EPICS {epics} maps to core {core}, less urgent than the \
2931                 EPICS band's own floor of 199"
2932            );
2933        }
2934        // The two ends of the EPICS space, as `RTEMS-score/osdThread.c:94-102`
2935        // defines them, and the clamp above it.
2936        assert_eq!(map_epics_priority_rtems(0), 56);
2937        assert_eq!(map_epics_priority_rtems(99), 155);
2938        assert_eq!(map_epics_priority_rtems(100), 155);
2939        assert_eq!(map_epics_priority_rtems(u8::MAX), 155);
2940        // Ordering still holds: a higher EPICS priority is a more urgent core.
2941        assert!(
2942            RTEMS_MAXIMUM_PRIORITY - map_epics_priority_rtems(ThreadPriority::ScanLow.value())
2943                < RTEMS_MAXIMUM_PRIORITY
2944                    - map_epics_priority_rtems(ThreadPriority::CaServerHigh.value())
2945        );
2946    }
2947
2948    /// [`map_epics_priority_vxworks`] must land on the exact same POSIX
2949    /// values as [`map_epics_priority_rtems`] — that equality is the
2950    /// measured fact [`DEFAULT_POLICY`]'s doc cites, and this function
2951    /// deliberately does not call into the RTEMS one (see its own doc), so
2952    /// nothing else pins the two together if one of them drifts.
2953    #[test]
2954    fn vxworks_priority_map_matches_the_rtems_posix_values() {
2955        for epics in 0..=u8::MAX {
2956            assert_eq!(
2957                map_epics_priority_vxworks(epics),
2958                map_epics_priority_rtems(epics),
2959                "EPICS {epics}: VxWorks and RTEMS must set the identical POSIX \
2960                 SCHED_FIFO value"
2961            );
2962        }
2963        // The measured endpoints, restated directly per this function's own
2964        // doc rather than only via the equality above.
2965        assert_eq!(map_epics_priority_vxworks(0), 56);
2966        assert_eq!(map_epics_priority_vxworks(99), 155);
2967        assert_eq!(map_epics_priority_vxworks(100), 155);
2968        assert_eq!(map_epics_priority_vxworks(u8::MAX), 155);
2969    }
2970
2971    /// The RTEMS map is not the hosted map with retuned endpoints, and the
2972    /// difference is exactly the reason the RTEMS arm exists.
2973    ///
2974    /// Stated as the hazard rather than as `assert_ne!` on a sample: over the
2975    /// EPICS space, feeding the *hosted linear* map the guest's own probed
2976    /// range (`min=1`, `max=254`) puts some priorities above libbsd's `IRQS`
2977    /// at core 96, and the fixed RTEMS map puts none there. The crossover is
2978    /// pinned at EPICS 63 because that number is the justification recorded in
2979    /// the commit message; if base's map or the measured band ever moves, this
2980    /// fails rather than the deviation quietly losing its reason.
2981    #[test]
2982    fn rtems_priority_map_is_not_the_hosted_linear_map() {
2983        const LIBBSD_IRQS_CORE: i32 = 96;
2984        // What `find_pri_range` yields on the guest (osdThread.c:295-311).
2985        let (min, max) = (1, 254);
2986
2987        let hosted_core = |epics: u8| RTEMS_MAXIMUM_PRIORITY - map_epics_priority(epics, min, max);
2988        let rtems_core = |epics: u8| RTEMS_MAXIMUM_PRIORITY - map_epics_priority_rtems(epics);
2989
2990        let hosted_above_irqs: Vec<u8> = (0..=99)
2991            .filter(|&e| hosted_core(e) < LIBBSD_IRQS_CORE)
2992            .collect();
2993        let rtems_above_irqs: Vec<u8> = (0..=99)
2994            .filter(|&e| rtems_core(e) < LIBBSD_IRQS_CORE)
2995            .collect();
2996
2997        assert_eq!(
2998            rtems_above_irqs,
2999            Vec::<u8>::new(),
3000            "the RTEMS map must place no EPICS priority above libbsd's IRQS"
3001        );
3002        assert_eq!(
3003            hosted_above_irqs.first().copied(),
3004            Some(63),
3005            "base-on-RTEMS-6's posix map crosses IRQS at EPICS 63; that number \
3006             is the recorded reason for this deviation"
3007        );
3008        // And concretely at the CA server band the audit cares about.
3009        assert_eq!(
3010            hosted_core(91),
3011            24,
3012            "upstream posix map: EPICS 91 -> core 24"
3013        );
3014        assert_eq!(rtems_core(91), 108, "this port: EPICS 91 -> core 108");
3015        // Shapes, not endpoints: the hosted map spans the whole probed range,
3016        // this one spans exactly 100 levels wherever it is placed.
3017        assert_eq!(
3018            map_epics_priority_rtems(99) - map_epics_priority_rtems(0),
3019            99
3020        );
3021        assert_eq!(
3022            map_epics_priority(99, min, max) - map_epics_priority(0, min, max),
3023            250
3024        );
3025    }
3026
3027    /// The name budget an RTEMS thread object actually has:
3028    /// `CONFIGURE_MAXIMUM_THREAD_NAME_SIZE` defaults to 16 *including* the
3029    /// NUL (`rtems/score/thread.h:1079`, `rtems/confdefs/threads.h:92-93`)
3030    /// and the boot shim does not override it, so 15 bytes — the same budget
3031    /// `std` truncates to on Linux.
3032    ///
3033    /// Truncating here rather than letting `_Thread_Set_name` do it is what
3034    /// keeps the call's result meaningful: that function `strlcpy`s and
3035    /// *still applies* the truncated name, but returns
3036    /// `STATUS_RESULT_TOO_LARGE` → `ERANGE`, so an untruncated call would log
3037    /// a failure for a name it had in fact set.
3038    #[test]
3039    fn thread_names_are_cut_to_the_rtems_budget_on_a_char_boundary() {
3040        assert_eq!(RTEMS_MAX_THREAD_NAME_BYTES, 15);
3041        // Short names pass through untouched.
3042        assert_eq!(truncate_thread_name("CAS-event"), "CAS-event");
3043        // Exactly at the budget.
3044        assert_eq!(truncate_thread_name("123456789012345"), "123456789012345");
3045        // Over it — a real per-client CA thread name.
3046        assert_eq!(
3047            truncate_thread_name("CAS-client-blocking 10.0.0.1:5064"),
3048            "CAS-client-blo"[..14].to_owned() + "c"
3049        );
3050        assert!(truncate_thread_name("CAS-client-blocking 10.0.0.1:5064").len() <= 15);
3051        // Never mid-codepoint: 'é' is two bytes, so a cut landing inside it
3052        // must step back rather than produce invalid UTF-8. 14 ASCII bytes
3053        // plus 'é' is 16 bytes; the budget cuts at 15, inside the 'é'.
3054        let mixed = "aaaaaaaaaaaaaaé";
3055        assert_eq!(mixed.len(), 16);
3056        assert_eq!(truncate_thread_name(mixed), "aaaaaaaaaaaaaa");
3057        // Empty stays empty rather than underflowing the boundary walk.
3058        assert_eq!(truncate_thread_name(""), "");
3059    }
3060
3061    /// Every thread this crate starts publishes its name to the OS.
3062    ///
3063    /// `std` calls the platform `pthread_setname_np` from `Builder::spawn`
3064    /// on the hosted targets it supports, and RTEMS is not one of them — so
3065    /// there, a name set with `Builder::name` lives only in Rust's `Thread`
3066    /// struct and the kernel shows nothing. Bring-up had to measure libbsd's
3067    /// priority band by other means for exactly that reason.
3068    ///
3069    /// The defect is a call that is *absent*, so this is source inspection
3070    /// over every production `Builder` site in the crate — the same sweep
3071    /// shape as `every_thread_in_this_crate_states_a_stack_size`, and it
3072    /// fails the same way when a new thread forgets. Either prologue counts:
3073    /// `enter_ioc_thread` for a thread with an EPICS band, bare
3074    /// `name_current_thread` for one that deliberately has none.
3075    #[test]
3076    fn every_thread_in_this_crate_publishes_its_name() {
3077        // This crate's files only — see the note on the sweep above.
3078        let files = [
3079            ("runtime/task.rs", include_str!("task.rs")),
3080            (
3081                "runtime/background/delayed_timer.rs",
3082                include_str!("background/delayed_timer.rs"),
3083            ),
3084            (
3085                "runtime/background/scan_once.rs",
3086                include_str!("background/scan_once.rs"),
3087            ),
3088            (
3089                "runtime/background/callback_executor.rs",
3090                include_str!("background/callback_executor.rs"),
3091            ),
3092        ];
3093        // The one exemption, named rather than pattern-matched: the
3094        // SCHED_FIFO range probe is `#[cfg(target_os = "linux")]`, exists for
3095        // two `sched_*` calls and a join, and never runs on the target whose
3096        // task listing this guard is about.
3097        const EXEMPT: &str = ".name(\"cbRtProbe\".to_string())";
3098
3099        let mut anonymous = Vec::new();
3100        let mut checked = 0usize;
3101        for (label, src) in files {
3102            for (n, after) in production_scope(src)
3103                .split("thread::Builder::new()")
3104                .skip(1)
3105                .enumerate()
3106            {
3107                let (chain, body) = after.split_once(".spawn(").unwrap_or((after, ""));
3108                if chain.contains(EXEMPT) {
3109                    continue;
3110                }
3111                checked += 1;
3112                // The prologue is the closure's first work, so look at the
3113                // closure, not the builder chain.
3114                if !body.contains("enter_ioc_thread(") && !body.contains("name_current_thread()") {
3115                    anonymous.push(format!("{label} (Builder #{})", n + 1));
3116                }
3117            }
3118        }
3119
3120        // Three: `spawn_dedicated_thread`'s two `cfg` arms and
3121        // `MandatoryThread::try_spawn`, all of which run the prologue for the
3122        // caller. The floor was five until the three background facilities
3123        // moved onto `MandatoryThread`, whose constructor takes the band —
3124        // `every_background_facility_thread_is_mandatory` covers those files.
3125        assert!(
3126            checked >= 3,
3127            "expected to find the crate's Builder sites, found {checked} — \
3128             did a file move? update this guard's file list"
3129        );
3130        assert!(
3131            anonymous.is_empty(),
3132            "these threads are invisible in an RTEMS task listing: {anonymous:?}"
3133        );
3134    }
3135
3136    /// The banding half of the prologue is not reachable without the naming
3137    /// half: nothing in this crate's production scope calls
3138    /// [`apply_to_current_thread`] except [`enter_ioc_thread`] itself.
3139    ///
3140    /// Separate from the sweep above because it catches the other direction —
3141    /// a thread that is named by `Builder` but takes its band directly, which
3142    /// the closure-body sweep would pass if the naming call happened to be
3143    /// somewhere else in the file.
3144    #[test]
3145    fn only_the_prologue_reaches_the_banding_call() {
3146        // This crate's files only — see the note on the sweep above.
3147        let files = [
3148            ("runtime/task.rs", include_str!("task.rs")),
3149            (
3150                "runtime/background/delayed_timer.rs",
3151                include_str!("background/delayed_timer.rs"),
3152            ),
3153            (
3154                "runtime/background/scan_once.rs",
3155                include_str!("background/scan_once.rs"),
3156            ),
3157            (
3158                "runtime/background/callback_executor.rs",
3159                include_str!("background/callback_executor.rs"),
3160            ),
3161        ];
3162        // Only the definition and the prologue's own delegation, both in
3163        // task.rs. Anywhere else is a thread banded without being named.
3164        let allowed = [
3165            "pub fn apply_to_current_thread(priority: ThreadPriority) -> PriorityApplied {",
3166            "apply_to_current_thread(priority)",
3167        ];
3168        let mut seen_definition = false;
3169        for (label, src) in files {
3170            let callers: Vec<&str> = production_scope(src)
3171                .lines()
3172                .map(str::trim)
3173                .filter(|l| l.contains("apply_to_current_thread("))
3174                .filter(|l| !l.starts_with("//"))
3175                .collect();
3176            seen_definition |= callers.contains(&allowed[0]);
3177            let strays: Vec<&&str> = callers.iter().filter(|l| !allowed.contains(l)).collect();
3178            assert!(
3179                strays.is_empty(),
3180                "{label}: only `enter_ioc_thread` may band a thread; \
3181                 everything else would band an OS-anonymous one — {strays:?}"
3182            );
3183        }
3184        assert!(
3185            seen_definition,
3186            "the banding function moved out of this file list; update the guard"
3187        );
3188    }
3189
3190    /// The prologue must also announce the thread to the statistics funnel's
3191    /// census, and that call has to be checked as text because nothing else can
3192    /// check it: it is `#[cfg]`ed to VxWorks, so on the host and on RTEMS it
3193    /// compiles away and deleting it breaks no build and no test. What it would
3194    /// break is one target's task census, which would come back empty — an IOC
3195    /// that reads as having no threads rather than as having a missing call.
3196    ///
3197    /// Located inside the prologue's own body rather than anywhere in the file,
3198    /// because a registration that drifted out of the single thread-transition
3199    /// owner is the same defect as no registration: threads would start without
3200    /// passing it.
3201    #[test]
3202    fn the_prologue_registers_the_thread_for_the_vxworks_census() {
3203        let body = production_scope(include_str!("task.rs"))
3204            .split_once("pub fn enter_ioc_thread(")
3205            .expect("the prologue is still in this file")
3206            .1
3207            .split_once("\n}\n")
3208            .expect("the prologue's body is terminated")
3209            .0;
3210        assert!(
3211            body.contains("#[cfg(target_os = \"vxworks\")]"),
3212            "the census registration must stay gated to the one OS whose \
3213             backend needs it; `epics-rtems-boot` is a dependency of this \
3214             package on that target only"
3215        );
3216        assert!(
3217            body.contains("epics_rtems_boot::stats::register_task();"),
3218            "VxWorks gives an RTP no task enumerator, so `dump_tasks` and \
3219             `stack_report` list exactly what announced itself here"
3220        );
3221    }
3222
3223    /// The owner path: a mandatory thread that *can* be created runs its body
3224    /// under the name and band it was declared with.
3225    #[test]
3226    fn a_mandatory_thread_runs_under_its_declared_name() {
3227        let (tx, rx) = std::sync::mpsc::channel();
3228        let join = MandatoryThread::new(
3229            "cbTestOwner",
3230            ThreadPriority::ScanLow,
3231            StackSizeClass::Small,
3232        )
3233        .spawn(move || {
3234            let _ = tx.send(
3235                std::thread::current()
3236                    .name()
3237                    .map(str::to_owned)
3238                    .unwrap_or_default(),
3239            );
3240        });
3241        assert_eq!(rx.recv().expect("the body ran"), "cbTestOwner");
3242        join.join().expect("the thread exited cleanly");
3243    }
3244
3245    /// `try_spawn` is the same construction with the failure handed back, so a
3246    /// caller inside a fallible boot step can refuse to serve.
3247    #[test]
3248    fn try_spawn_hands_back_a_handle_on_success() {
3249        let (tx, rx) = std::sync::mpsc::channel();
3250        let join =
3251            MandatoryThread::new("cbTestTry", ThreadPriority::ScanLow, StackSizeClass::Small)
3252                .try_spawn(move || {
3253                    let _ = tx.send(());
3254                })
3255                .expect("a thread is creatable in the test environment");
3256        rx.recv().expect("the body ran");
3257        join.join().expect("the thread exited cleanly");
3258    }
3259
3260    /// The console line names the thread and what the OS said, so an operator
3261    /// reading a target console can tell *which* thread the IOC died for.
3262    ///
3263    /// `EAGAIN` cannot be forced portably — the failure shape is the subject
3264    /// here, not the syscall.
3265    #[test]
3266    fn the_fatal_message_names_the_thread_and_the_error() {
3267        let msg = mandatory_thread_failure_message(
3268            "scan-0.1",
3269            &std::io::Error::from(std::io::ErrorKind::WouldBlock),
3270        );
3271        assert!(msg.contains("scan-0.1"), "{msg}");
3272        assert!(msg.contains("FATAL"), "{msg}");
3273        assert!(
3274            msg.contains(&std::io::Error::from(std::io::ErrorKind::WouldBlock).to_string()),
3275            "{msg}"
3276        );
3277    }
3278
3279    /// The bypass regression: a mandatory thread that cannot be created must
3280    /// take the **process** down, not the calling thread.
3281    ///
3282    /// The defect this closes was measured on a VxWorks 7 RTP: `EAGAIN` from
3283    /// the periodic-scan spawn panicked the `scan-owner` thread, and because
3284    /// both RTEMS and VxWorks default to `panic = "unwind"`, the process
3285    /// survived and went on serving CA with no periodic scanning at all. A test
3286    /// that only asserted "it panics" would have passed against that defect —
3287    /// so this one re-executes itself and asserts the *process* died.
3288    ///
3289    /// Gated off the embedded targets: they have no process to spawn.
3290    #[cfg(all(unix, not(target_os = "rtems"), not(target_os = "vxworks")))]
3291    #[test]
3292    fn a_mandatory_thread_that_cannot_be_created_aborts_the_process() {
3293        use std::os::unix::process::ExitStatusExt;
3294
3295        const CHILD: &str = "EPICS_RS_MANDATORY_THREAD_ABORT_CHILD";
3296        const TEST: &str =
3297            "runtime::task::tests::a_mandatory_thread_that_cannot_be_created_aborts_the_process";
3298
3299        if std::env::var_os(CHILD).is_some() {
3300            mandatory_thread_unavailable(
3301                "scan-0.1",
3302                &std::io::Error::from(std::io::ErrorKind::WouldBlock),
3303            );
3304        }
3305
3306        let out =
3307            std::process::Command::new(std::env::current_exe().expect("the test binary path"))
3308                .args(["--exact", TEST, "--nocapture"])
3309                .env(CHILD, "1")
3310                .output()
3311                .expect("re-exec the test binary");
3312
3313        assert_eq!(
3314            out.status.signal(),
3315            Some(libc::SIGABRT),
3316            "a mandatory thread's failure must abort the process, not unwind \
3317             one thread — child exited {:?}, stderr: {}",
3318            out.status,
3319            String::from_utf8_lossy(&out.stderr)
3320        );
3321        let stderr = String::from_utf8_lossy(&out.stderr);
3322        assert!(
3323            stderr.contains("scan-0.1"),
3324            "the console must name the thread; got: {stderr}"
3325        );
3326    }
3327
3328    #[epics_macros_rs::epics_test]
3329    async fn spawn_blocking_with_priority_runs_closure() {
3330        let handle = spawn_blocking_with_priority(ThreadPriority::CaServerHigh, || 7);
3331        assert_eq!(handle.await.unwrap(), 7);
3332    }
3333
3334    #[test]
3335    fn background_global_inits_and_runs_work() {
3336        // Host-exercises the OnceLock init path the RTEMS spawn/sleep/interval
3337        // arms rely on: background_init() forces creation, background() hands
3338        // back a usable executor whose callback pool runs submitted work.
3339        background_init();
3340        let exec = background();
3341        let (tx, rx) = std::sync::mpsc::channel();
3342        exec.callbacks()
3343            .handle()
3344            .request(
3345                crate::runtime::background::CallbackPriority::Medium,
3346                Box::new(move || tx.send(1u8).unwrap()),
3347            )
3348            .unwrap();
3349        assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), 1);
3350    }
3351}