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