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