use std::future::Future;
use std::sync::Arc;
use std::task::{Context, Poll, Wake, Waker};
use std::time::Duration;
use tokio::runtime::RuntimeFlavor;
pub use tokio::runtime::Handle as RuntimeHandle;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NotBlockable {
CurrentThreadRuntime,
BackgroundWorker,
}
impl std::fmt::Display for NotBlockable {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
NotBlockable::CurrentThreadRuntime => {
f.write_str("cannot block a current-thread runtime")
}
NotBlockable::BackgroundWorker => {
f.write_str("cannot block a background-facility worker thread")
}
}
}
}
impl std::error::Error for NotBlockable {}
pub(crate) struct ThreadWaker(std::thread::Thread);
impl ThreadWaker {
pub(crate) fn for_current_thread() -> Waker {
Waker::from(Arc::new(ThreadWaker(std::thread::current())))
}
}
impl Wake for ThreadWaker {
fn wake(self: Arc<Self>) {
self.0.unpark();
}
fn wake_by_ref(self: &Arc<Self>) {
self.0.unpark();
}
}
pub(crate) fn park_on_interruptible<F: Future>(
fut: F,
mut should_cancel: impl FnMut() -> bool,
) -> Option<F::Output> {
let mut fut = std::pin::pin!(fut);
let waker = ThreadWaker::for_current_thread();
let mut cx = Context::from_waker(&waker);
loop {
if should_cancel() {
return None;
}
if let Poll::Ready(value) = fut.as_mut().poll(&mut cx) {
return Some(value);
}
std::thread::park();
}
}
fn park_on<F: Future>(fut: F) -> F::Output {
park_on_interruptible(fut, || false).expect("uncancellable driver returned None")
}
pub fn block_on_sync<F: Future>(fut: F) -> Result<F::Output, NotBlockable> {
if crate::runtime::background::facility::on_facility_thread() {
return Err(NotBlockable::BackgroundWorker);
}
match RuntimeHandle::try_current() {
Ok(handle) => match handle.runtime_flavor() {
RuntimeFlavor::CurrentThread => Err(NotBlockable::CurrentThreadRuntime),
_ => Ok(tokio::task::block_in_place(|| handle.block_on(fut))),
},
Err(_) => Ok(park_on(fut)),
}
}
#[cfg(tokio_backend)]
#[derive(Clone)]
pub struct BlockingBridge {
handle: tokio::runtime::Handle,
}
#[cfg(exec_backend)]
#[derive(Clone)]
pub struct BlockingBridge;
#[cfg(tokio_backend)]
impl BlockingBridge {
pub fn capture() -> Self {
Self {
handle: tokio::runtime::Handle::current(),
}
}
pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
assert!(
RuntimeHandle::try_current().is_err(),
"BlockingBridge::block_on must not be called from a runtime thread"
);
self.handle.block_on(fut)
}
pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
self.handle.spawn(future)
}
}
#[cfg(exec_backend)]
impl BlockingBridge {
pub fn capture() -> Self {
Self
}
pub fn block_on<F: Future>(&self, fut: F) -> F::Output {
park_on(fut)
}
pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
spawn(future)
}
}
#[cfg(tokio_backend)]
pub fn test_block_on<F: Future>(fut: F) -> F::Output {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("failed to build tokio test runtime")
.block_on(fut)
}
#[cfg(exec_backend)]
pub fn test_block_on<F: Future>(fut: F) -> F::Output {
park_on(fut)
}
pub const HAS_TOKIO_REACTOR: bool = cfg!(tokio_backend);
#[cfg(tokio_backend)]
pub type TaskHandle<T> = tokio::task::JoinHandle<T>;
#[cfg(tokio_backend)]
pub type TaskAbortHandle = tokio::task::AbortHandle;
#[cfg(tokio_backend)]
pub type TaskJoinError = tokio::task::JoinError;
#[cfg(exec_backend)]
pub type TaskHandle<T> = crate::runtime::background::future_exec::JoinFuture<T>;
#[cfg(exec_backend)]
pub type TaskAbortHandle = crate::runtime::background::future_exec::AbortHandle;
#[cfg(exec_backend)]
pub type TaskJoinError = crate::runtime::background::future_exec::JoinError;
#[cfg(any(exec_backend, test))]
static BACKGROUND: std::sync::OnceLock<crate::runtime::background::BackgroundExecutor> =
std::sync::OnceLock::new();
#[cfg(any(exec_backend, test))]
fn background() -> &'static crate::runtime::background::BackgroundExecutor {
BACKGROUND.get_or_init(crate::runtime::background::BackgroundExecutor::new)
}
#[cfg(any(exec_backend, test))]
pub fn background_init() {
let _ = background();
}
#[cfg(tokio_backend)]
pub fn spawn<F>(future: F) -> TaskHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
tokio::spawn(future)
}
#[cfg(exec_backend)]
pub fn spawn<F>(future: F) -> TaskHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_future};
spawn_future(
&background().callbacks().handle(),
DEFAULT_SPAWN_PRIORITY,
future,
)
}
#[cfg(tokio_backend)]
pub async fn yield_now() {
tokio::task::yield_now().await;
}
#[cfg(exec_backend)]
pub async fn yield_now() {
struct YieldNow(bool);
impl Future for YieldNow {
type Output = ();
fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
if self.0 {
Poll::Ready(())
} else {
self.0 = true;
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
YieldNow(false).await;
}
#[cfg(tokio_backend)]
pub fn spawn_blocking<F, R>(f: F) -> TaskHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
tokio::task::spawn_blocking(f)
}
#[cfg(exec_backend)]
pub fn spawn_blocking<F, R>(f: F) -> TaskHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_blocking_on};
spawn_blocking_on(
&background().callbacks().handle(),
DEFAULT_SPAWN_PRIORITY,
f,
)
}
pub struct TaskSet<T> {
tasks: Vec<TaskHandle<T>>,
}
impl<T> Default for TaskSet<T> {
fn default() -> Self {
Self::new()
}
}
impl<T> TaskSet<T> {
pub fn new() -> Self {
Self { tasks: Vec::new() }
}
pub fn len(&self) -> usize {
self.tasks.len()
}
pub fn is_empty(&self) -> bool {
self.tasks.is_empty()
}
}
impl<T: Send + 'static> TaskSet<T> {
pub fn spawn<F>(&mut self, future: F)
where
F: Future<Output = T> + Send + 'static,
{
self.tasks.push(spawn(future));
}
pub async fn join_next(&mut self) -> Option<Result<T, TaskJoinError>> {
if self.tasks.is_empty() {
return None;
}
std::future::poll_fn(|cx| {
for i in 0..self.tasks.len() {
if let std::task::Poll::Ready(result) =
std::pin::Pin::new(&mut self.tasks[i]).poll(cx)
{
self.tasks.swap_remove(i);
return std::task::Poll::Ready(Some(result));
}
}
std::task::Poll::Pending
})
.await
}
}
impl<T> Drop for TaskSet<T> {
fn drop(&mut self) {
for task in &self.tasks {
task.abort();
}
}
}
#[cfg(tokio_backend)]
pub async fn sleep(duration: Duration) {
tokio::time::sleep(duration).await;
}
#[cfg(exec_backend)]
pub async fn sleep(duration: Duration) {
crate::runtime::background::timer_sleep::sleep(&background().timer().handle(), duration).await;
}
#[cfg(tokio_backend)]
pub type Instant = tokio::time::Instant;
#[cfg(exec_backend)]
pub type Instant = std::time::Instant;
#[cfg(tokio_backend)]
pub async fn sleep_until(deadline: Instant) {
tokio::time::sleep_until(deadline).await;
}
#[cfg(exec_backend)]
pub async fn sleep_until(deadline: Instant) {
crate::runtime::background::timer_sleep::sleep_until(&background().timer().handle(), deadline)
.await;
}
#[cfg(tokio_backend)]
pub struct Interval {
inner: tokio::time::Interval,
}
#[cfg(tokio_backend)]
impl Interval {
pub async fn tick(&mut self) {
self.inner.tick().await;
}
}
#[cfg(exec_backend)]
pub type Interval = crate::runtime::background::timer_sleep::TimerInterval;
#[cfg(tokio_backend)]
pub fn interval(period: Duration) -> Interval {
Interval {
inner: tokio::time::interval(period),
}
}
#[cfg(exec_backend)]
pub fn interval(period: Duration) -> Interval {
crate::runtime::background::timer_sleep::interval(&background().timer().handle(), period)
}
#[cfg(tokio_backend)]
pub use tokio::time::error::Elapsed;
#[cfg(exec_backend)]
#[derive(Debug)]
pub struct Elapsed(());
#[cfg(exec_backend)]
impl std::fmt::Display for Elapsed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "deadline has elapsed")
}
}
#[cfg(exec_backend)]
impl std::error::Error for Elapsed {}
#[cfg(tokio_backend)]
pub async fn timeout<F: Future>(duration: Duration, fut: F) -> Result<F::Output, Elapsed> {
tokio::time::timeout(duration, fut).await
}
#[cfg(exec_backend)]
pub async fn timeout<F: Future>(duration: Duration, fut: F) -> Result<F::Output, Elapsed> {
let mut sleep = std::pin::pin!(sleep(duration));
let mut fut = std::pin::pin!(fut);
std::future::poll_fn(move |cx| {
if let Poll::Ready(v) = fut.as_mut().poll(cx) {
return Poll::Ready(Ok(v));
}
if sleep.as_mut().poll(cx).is_ready() {
return Poll::Ready(Err(Elapsed(())));
}
Poll::Pending
})
.await
}
#[cfg(tokio_backend)]
pub async fn timeout_at<F: Future>(deadline: Instant, fut: F) -> Result<F::Output, Elapsed> {
tokio::time::timeout_at(deadline, fut).await
}
#[cfg(exec_backend)]
pub async fn timeout_at<F: Future>(deadline: Instant, fut: F) -> Result<F::Output, Elapsed> {
let mut sleep = std::pin::pin!(sleep_until(deadline));
let mut fut = std::pin::pin!(fut);
std::future::poll_fn(move |cx| {
if let Poll::Ready(v) = fut.as_mut().poll(cx) {
return Poll::Ready(Ok(v));
}
if sleep.as_mut().poll(cx).is_ready() {
return Poll::Ready(Err(Elapsed(())));
}
Poll::Pending
})
.await
}
pub fn runtime_handle() -> tokio::runtime::Handle {
tokio::runtime::Handle::current()
}
pub const PRIORITY_MIN: u8 = 0;
pub const PRIORITY_MAX: u8 = 99;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum ThreadPriority {
Low,
CaServerLow,
CaServerHigh,
Medium,
ScanLow,
ScanHigh,
High,
Iocsh,
Custom(u8),
}
impl ThreadPriority {
pub const fn value(self) -> u8 {
let v = match self {
ThreadPriority::Low => 10,
ThreadPriority::CaServerLow => 20,
ThreadPriority::CaServerHigh => 40,
ThreadPriority::Medium => 50,
ThreadPriority::ScanLow => 60,
ThreadPriority::ScanHigh => 70,
ThreadPriority::High => 90,
ThreadPriority::Iocsh => 91,
ThreadPriority::Custom(v) => v,
};
if v > PRIORITY_MAX { PRIORITY_MAX } else { v }
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StackSizeClass {
Small,
Medium,
Big,
}
impl StackSizeClass {
pub fn bytes(self) -> usize {
let unit = 0x10000usize * std::mem::size_of::<usize>();
match self {
StackSizeClass::Small => unit,
StackSizeClass::Medium => 2 * unit,
StackSizeClass::Big => 4 * unit,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PriorityApplied {
Realtime,
Disabled,
Unsupported,
BestEffortFailed,
}
impl PriorityApplied {
pub fn is_realtime(self) -> bool {
matches!(self, PriorityApplied::Realtime)
}
}
pub const RT_PRIORITY_ENV: &str = "EPICS_RS_ALLOW_RT_PRIORITY";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RtPolicy {
Disabled,
AllowRealtime,
}
pub const DEFAULT_POLICY: RtPolicy = default_policy(cfg!(epics_embedded_target));
const fn default_policy(on_rtems: bool) -> RtPolicy {
if on_rtems {
RtPolicy::AllowRealtime
} else {
RtPolicy::Disabled
}
}
impl RtPolicy {
pub fn from_env_value(raw: Option<&str>) -> RtPolicy {
Self::resolve(raw, DEFAULT_POLICY)
}
pub fn resolve(raw: Option<&str>, default: RtPolicy) -> RtPolicy {
let Some(raw) = raw else {
return default;
};
let v = raw.trim();
let on = v.eq_ignore_ascii_case("yes")
|| v.eq_ignore_ascii_case("true")
|| v.eq_ignore_ascii_case("on")
|| v == "1";
if on {
RtPolicy::AllowRealtime
} else {
RtPolicy::Disabled
}
}
pub fn current() -> RtPolicy {
static POLICY: std::sync::OnceLock<RtPolicy> = std::sync::OnceLock::new();
*POLICY.get_or_init(|| {
RtPolicy::from_env_value(std::env::var(RT_PRIORITY_ENV).ok().as_deref())
})
}
}
pub fn apply_to_current_thread(priority: ThreadPriority) -> PriorityApplied {
apply_to_current_thread_under(RtPolicy::current(), priority)
}
pub fn apply_to_current_thread_under(
policy: RtPolicy,
priority: ThreadPriority,
) -> PriorityApplied {
match policy {
RtPolicy::Disabled => PriorityApplied::Disabled,
RtPolicy::AllowRealtime => apply_priority_impl(priority.value()),
}
}
pub fn enter_ioc_thread(priority: ThreadPriority) -> PriorityApplied {
name_current_thread();
#[cfg(target_os = "vxworks")]
epics_rtems_boot::stats::register_task();
apply_to_current_thread(priority)
}
#[cfg(not(target_os = "rtems"))]
pub fn name_current_thread() {}
#[cfg(target_os = "rtems")]
pub fn name_current_thread() {
let current = std::thread::current();
let Some(name) = current.name() else {
return;
};
let Ok(c_name) = std::ffi::CString::new(truncate_thread_name(name)) else {
return;
};
let rc =
unsafe { rtems_sched::pthread_setname_np(rtems_sched::pthread_self(), c_name.as_ptr()) };
if rc != 0 {
tracing::debug!(
target: "epics_base_rs::runtime",
thread = name,
errno = rc,
"pthread_setname_np failed; thread stays unnamed in the task listing"
);
}
}
#[cfg(any(target_os = "rtems", test))]
const RTEMS_MAX_THREAD_NAME_BYTES: usize = 15;
#[cfg(any(target_os = "rtems", test))]
fn truncate_thread_name(name: &str) -> &str {
let mut end = name.len().min(RTEMS_MAX_THREAD_NAME_BYTES);
while end > 0 && !name.is_char_boundary(end) {
end -= 1;
}
&name[..end]
}
#[cfg(target_os = "linux")]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum RtRange {
Unsupported,
Denied,
Available { min: i32, max: i32 },
}
#[cfg(target_os = "linux")]
fn permitted_fifo_range() -> RtRange {
static RANGE: std::sync::OnceLock<RtRange> = std::sync::OnceLock::new();
*RANGE.get_or_init(probe_fifo_range)
}
#[cfg(target_os = "linux")]
fn probe_fifo_range() -> RtRange {
let (min, max) = unsafe {
(
libc::sched_get_priority_min(libc::SCHED_FIFO),
libc::sched_get_priority_max(libc::SCHED_FIFO),
)
};
if min < 0 || max < 0 || max < min {
return RtRange::Unsupported;
}
let probe = std::thread::Builder::new()
.name("cbRtProbe".to_string())
.stack_size(StackSizeClass::Small.bytes())
.spawn(move || {
if set_fifo_priority(min) != 0 {
return RtRange::Denied;
}
let (mut low, mut high) = (min, max);
while low < high {
let mid = (high + low) / 2;
if set_fifo_priority(mid) != 0 {
high = mid;
} else {
low = mid + 1;
}
}
let top = if set_fifo_priority(high) != 0 {
high - 1
} else {
high
};
RtRange::Available { min, max: top }
});
match probe.map(std::thread::JoinHandle::join) {
Ok(Ok(range)) => range,
_ => RtRange::Denied,
}
}
#[cfg(any(target_os = "linux", test))]
fn map_epics_priority(epics_priority: u8, min: i32, max: i32) -> i32 {
if max == min {
return max;
}
let slope = (max - min) as f64 / 100.0;
let oss = epics_priority as f64 * slope + min as f64;
(oss as i32).clamp(min, max)
}
#[cfg(any(target_os = "rtems", test))]
const RTEMS_MAXIMUM_PRIORITY: i32 = 255;
#[cfg(any(target_os = "rtems", test))]
const fn rtems_core_priority(epics_priority: u8) -> i32 {
if epics_priority > 99 {
100
} else {
199 - epics_priority as i32
}
}
#[cfg(any(target_os = "rtems", test))]
fn map_epics_priority_rtems(epics_priority: u8) -> i32 {
RTEMS_MAXIMUM_PRIORITY - rtems_core_priority(epics_priority)
}
#[cfg(any(target_os = "vxworks", test))]
fn map_epics_priority_vxworks(epics_priority: u8) -> i32 {
56 + epics_priority.min(99) as i32
}
#[cfg(any(target_os = "linux", epics_embedded_target))]
fn set_fifo_priority(oss: i32) -> i32 {
SCHED_CALLS_MADE.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
#[cfg(test)]
SCHED_CALLS.with(|c| c.set(c.get() + 1));
set_fifo_priority_raw(oss)
}
#[cfg(target_os = "linux")]
fn set_fifo_priority_raw(oss: i32) -> i32 {
let param = libc::sched_param {
sched_priority: oss,
};
unsafe { libc::pthread_setschedparam(libc::pthread_self(), libc::SCHED_FIFO, ¶m) }
}
#[cfg(target_os = "rtems")]
mod rtems_sched {
use std::ffi::c_int;
pub const SCHED_FIFO: c_int = 1;
#[repr(C, align(8))]
pub struct SchedParam {
pub sched_priority: c_int,
pub sporadic_tail: [u8; 44],
}
const _: () = {
assert!(core::mem::size_of::<SchedParam>() == 48);
assert!(core::mem::align_of::<SchedParam>() == 8);
};
unsafe extern "C" {
pub fn pthread_self() -> libc::pthread_t;
pub fn pthread_setschedparam(
thread: libc::pthread_t,
policy: c_int,
param: *const SchedParam,
) -> c_int;
pub fn pthread_setname_np(thread: libc::pthread_t, name: *const std::ffi::c_char) -> c_int;
}
}
#[cfg(target_os = "rtems")]
fn set_fifo_priority_raw(oss: i32) -> i32 {
let param = rtems_sched::SchedParam {
sched_priority: oss,
sporadic_tail: [0u8; 44],
};
unsafe {
rtems_sched::pthread_setschedparam(
rtems_sched::pthread_self(),
rtems_sched::SCHED_FIFO,
¶m,
)
}
}
#[cfg(target_os = "vxworks")]
fn set_fifo_priority_raw(oss: i32) -> i32 {
let param = libc::sched_param {
sched_priority: oss,
sched_ss_low_priority: 0,
sched_ss_repl_period: libc::timespec {
tv_sec: 0,
tv_nsec: 0,
},
sched_ss_init_budget: libc::timespec {
tv_sec: 0,
tv_nsec: 0,
},
sched_ss_max_repl: 0,
};
unsafe { libc::pthread_setschedparam(libc::pthread_self(), libc::SCHED_FIFO, ¶m) }
}
#[cfg(target_os = "linux")]
fn warn_rt_denied_once() {
static WARNED: std::sync::Once = std::sync::Once::new();
WARNED.call_once(|| {
#[cfg(test)]
DENIED_WARNINGS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
tracing::warn!(
target: "epics_base_rs::runtime",
switch = RT_PRIORITY_ENV,
"{RT_PRIORITY_ENV} asked for real-time scheduling, but this process may not \
use SCHED_FIFO (needs CAP_SYS_NICE or a non-zero RLIMIT_RTPRIO). Every IOC \
thread stays at the default scheduling policy; timing is not real-time. \
Logged once per process."
);
});
}
#[cfg(target_os = "linux")]
fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
let (min, max) = match permitted_fifo_range() {
RtRange::Unsupported => return PriorityApplied::Unsupported,
RtRange::Denied => {
warn_rt_denied_once();
return PriorityApplied::BestEffortFailed;
}
RtRange::Available { min, max } => (min, max),
};
let oss = map_epics_priority(epics_priority, min, max);
let rc = set_fifo_priority(oss);
if rc == 0 {
PriorityApplied::Realtime
} else {
tracing::debug!(
target: "epics_base_rs::runtime",
epics_priority,
oss,
errno = rc,
"SCHED_FIFO priority not applied; thread stays at default policy"
);
PriorityApplied::BestEffortFailed
}
}
#[cfg(target_os = "rtems")]
fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
let oss = map_epics_priority_rtems(epics_priority);
let rc = set_fifo_priority(oss);
if rc == 0 {
PriorityApplied::Realtime
} else {
tracing::debug!(
target: "epics_base_rs::runtime",
epics_priority,
oss,
errno = rc,
"SCHED_FIFO priority not applied; thread stays at default policy"
);
PriorityApplied::BestEffortFailed
}
}
#[cfg(target_os = "vxworks")]
fn apply_priority_impl(epics_priority: u8) -> PriorityApplied {
let oss = map_epics_priority_vxworks(epics_priority);
let rc = set_fifo_priority(oss);
if rc == 0 {
PriorityApplied::Realtime
} else {
tracing::debug!(
target: "epics_base_rs::runtime",
epics_priority,
oss,
errno = rc,
"SCHED_FIFO priority not applied; thread stays at default policy"
);
PriorityApplied::BestEffortFailed
}
}
#[cfg(not(any(target_os = "linux", epics_embedded_target)))]
fn apply_priority_impl(_epics_priority: u8) -> PriorityApplied {
PriorityApplied::Unsupported
}
pub fn sched_calls_made() -> usize {
SCHED_CALLS_MADE.load(std::sync::atomic::Ordering::Relaxed)
}
static SCHED_CALLS_MADE: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
#[cfg(all(test, any(target_os = "linux", epics_embedded_target)))]
thread_local! {
static SCHED_CALLS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
}
#[cfg(all(test, target_os = "linux"))]
static DENIED_WARNINGS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
#[cfg(tokio_backend)]
pub fn spawn_blocking_with_priority<F, R>(priority: ThreadPriority, f: F) -> TaskHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
tokio::task::spawn_blocking(move || {
let _ = enter_ioc_thread(priority);
f()
})
}
#[cfg(exec_backend)]
pub fn spawn_blocking_with_priority<F, R>(_priority: ThreadPriority, f: F) -> TaskHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_blocking_on};
spawn_blocking_on(
&background().callbacks().handle(),
DEFAULT_SPAWN_PRIORITY,
f,
)
}
#[cfg(tokio_backend)]
pub fn spawn_dedicated_thread<F>(
name: String,
priority: ThreadPriority,
stack: StackSizeClass,
f: F,
) -> std::io::Result<std::thread::JoinHandle<()>>
where
F: FnOnce() + Send + 'static,
{
let ambient = InheritedRuntime::capture();
std::thread::Builder::new()
.name(name)
.stack_size(stack.bytes())
.spawn(move || {
ambient.run(move || {
let _ = enter_ioc_thread(priority);
f()
})
})
}
#[cfg(tokio_backend)]
pub(crate) struct InheritedRuntime(Option<tokio::runtime::Handle>);
#[cfg(tokio_backend)]
impl InheritedRuntime {
pub(crate) fn capture() -> Self {
Self(
tokio::runtime::Handle::try_current()
.ok()
.filter(|h| h.runtime_flavor() != RuntimeFlavor::CurrentThread),
)
}
pub(crate) fn run<R>(&self, f: impl FnOnce() -> R) -> R {
let _entered = self.0.as_ref().map(|h| h.enter());
f()
}
}
#[cfg(exec_backend)]
pub(crate) struct InheritedRuntime;
#[cfg(exec_backend)]
impl InheritedRuntime {
pub(crate) fn capture() -> Self {
Self
}
pub(crate) fn run<R>(&self, f: impl FnOnce() -> R) -> R {
f()
}
}
#[cfg(exec_backend)]
pub fn spawn_dedicated_thread<F>(
name: String,
priority: ThreadPriority,
stack: StackSizeClass,
f: F,
) -> std::io::Result<std::thread::JoinHandle<()>>
where
F: FnOnce() + Send + 'static,
{
std::thread::Builder::new()
.name(name)
.stack_size(stack.bytes())
.spawn(move || {
let _ = enter_ioc_thread(priority);
f()
})
}
pub struct MandatoryThread {
name: String,
priority: ThreadPriority,
stack: StackSizeClass,
}
impl MandatoryThread {
pub fn new(name: impl Into<String>, priority: ThreadPriority, stack: StackSizeClass) -> Self {
Self {
name: name.into(),
priority,
stack,
}
}
pub fn spawn<F>(self, f: F) -> std::thread::JoinHandle<()>
where
F: FnOnce() + Send + 'static,
{
let name = self.name.clone();
match self.try_spawn(f) {
Ok(handle) => handle,
Err(e) => mandatory_thread_unavailable(&name, &e),
}
}
pub fn try_spawn<F>(self, f: F) -> std::io::Result<std::thread::JoinHandle<()>>
where
F: FnOnce() + Send + 'static,
{
let priority = self.priority;
std::thread::Builder::new()
.name(self.name)
.stack_size(self.stack.bytes())
.spawn(move || {
let _ = enter_ioc_thread(priority);
f()
})
}
}
fn mandatory_thread_failure_message(name: &str, err: &std::io::Error) -> String {
format!(
"FATAL: the IOC could not create its mandatory `{name}` thread: {err}. \
Continuing would leave this IOC answering clients while the work that \
thread owns never runs, so the process is aborting instead \
(C dbScan.c:943-959 wedges iocInit for the same reason)."
)
}
fn mandatory_thread_unavailable(name: &str, err: &std::io::Error) -> ! {
eprintln!("{}", mandatory_thread_failure_message(name, err));
std::process::abort()
}
#[cfg(test)]
mod tests {
use super::*;
fn production_scope(src: &str) -> &str {
match src.find("\n#[cfg(test)]") {
Some(i) => &src[..i],
None => src,
}
}
fn censused_files() -> [(&'static str, &'static str); 5] {
[
("runtime/task.rs", include_str!("task.rs")),
(
"runtime/background/delayed_timer.rs",
include_str!("background/delayed_timer.rs"),
),
(
"runtime/background/scan_once.rs",
include_str!("background/scan_once.rs"),
),
(
"runtime/background/callback_executor.rs",
include_str!("background/callback_executor.rs"),
),
("runtime/worker_pool.rs", include_str!("worker_pool.rs")),
]
}
#[test]
fn every_thread_in_this_crate_states_a_stack_size() {
let mut unclassified = Vec::new();
let mut checked = 0usize;
for (label, src) in censused_files() {
let prod = production_scope(src);
for (n, after) in prod.split("thread::Builder::new()").skip(1).enumerate() {
checked += 1;
let chain = after.split(".spawn(").next().unwrap_or("");
if !chain.contains(".stack_size(") {
unclassified.push(format!("{label} (Builder #{})", n + 1));
}
}
let bare = concat!("thread", "::spawn(");
for (n, line) in prod.lines().enumerate() {
let t = line.trim_start();
if t.starts_with("//") {
continue;
}
if t.contains(bare) && !t.contains("Builder") {
unclassified.push(format!("{label}:{} (bare spawn)", n + 1));
}
}
}
assert!(
checked >= 5,
"expected to find the crate's Builder sites, found {checked} — \
did a file move? update this guard's file list"
);
assert!(
unclassified.is_empty(),
"these threads inherit std's 2 MiB default on RTEMS: {unclassified:?}"
);
}
#[test]
fn every_background_facility_thread_is_mandatory() {
let bare = concat!("thread", "::spawn(");
let mut strays = Vec::new();
let mut owned = 0usize;
for (label, src) in censused_files() {
if !label.contains("/background/") {
continue;
}
for (n, line) in production_scope(src).lines().enumerate() {
let t = line.trim_start();
if t.starts_with("//") {
continue;
}
if t.contains("MandatoryThread::new(") {
owned += 1;
}
if t.contains("thread::Builder::new()") {
strays.push(format!("{label}:{} (raw Builder)", n + 1));
}
if t.contains(bare) && !t.contains("Builder") {
strays.push(format!("{label}:{} (bare spawn)", n + 1));
}
}
}
assert!(
strays.is_empty(),
"a facility thread created outside `MandatoryThread` resolves its \
own spawn failure, and the only resolution that keeps this IOC \
honest is not serving: {strays:?}"
);
assert!(
owned >= 3,
"expected the callback pool, `cbTimer` and `scanOnce`, found {owned} \
`MandatoryThread` sites — did a file move? update the census list"
);
}
#[epics_macros_rs::epics_test]
async fn test_spawn() {
let handle = spawn(async { 42 });
assert_eq!(handle.await.unwrap(), 42);
}
#[epics_macros_rs::epics_test]
async fn test_spawn_blocking() {
let handle = spawn_blocking(|| 123);
assert_eq!(handle.await.unwrap(), 123);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_dedicated_thread_carries_the_ambient_runtime() {
let (tx, rx) = std::sync::mpsc::channel();
let joined = spawn_dedicated_thread(
"dedicated-with-runtime".into(),
ThreadPriority::CaServerLow,
StackSizeClass::Small,
move || {
let outcome = block_on_sync(async {
let inner = spawn(async { 7u32 }).await.expect("inner task");
sleep(Duration::from_millis(1)).await;
inner
});
let _ = tx.send((
std::thread::current().name().map(str::to_string),
outcome.ok(),
));
},
)
.expect("dedicated thread spawned");
let (name, value) = rx
.recv_timeout(Duration::from_secs(5))
.expect("the dedicated thread must complete, not panic");
assert_eq!(name.as_deref(), Some("dedicated-with-runtime"));
assert_eq!(
value,
Some(7),
"a spawn and a timer must both work on the dedicated thread"
);
joined.join().expect("dedicated thread joined");
}
#[epics_macros_rs::epics_test]
async fn a_dedicated_thread_can_block_under_a_current_thread_ambient() {
let (tx, rx) = std::sync::mpsc::channel();
let joined = spawn_dedicated_thread(
"dedicated-current-thread-ambient".into(),
ThreadPriority::Low,
StackSizeClass::Small,
move || {
let (ctx, crx) = tokio::sync::mpsc::channel::<u32>(1);
let outcome = block_on_sync(async move {
ctx.send(9u32).await.expect("send into a depth-1 channel");
let mut crx = crx;
crx.recv().await
});
let _ = tx.send(outcome.ok().flatten());
},
)
.expect("dedicated thread spawned");
assert_eq!(
rx.recv_timeout(Duration::from_secs(5))
.expect("the dedicated thread must complete, not panic"),
Some(9),
"a dedicated thread must be able to park under a current-thread \
ambient runtime; inheriting that handle makes block_on_sync \
refuse and every pump built on it exit at once"
);
joined.join().expect("dedicated thread joined");
}
#[test]
fn a_dedicated_thread_runs_without_an_ambient_runtime() {
let (tx, rx) = std::sync::mpsc::channel();
let joined = spawn_dedicated_thread(
"dedicated-no-runtime".into(),
ThreadPriority::Low,
StackSizeClass::Small,
move || {
let _ = tx.send((
std::thread::current().name().map(str::to_string),
block_on_sync(async { 5u32 }).ok(),
));
},
)
.expect("dedicated thread spawned");
let (name, value) = rx
.recv_timeout(Duration::from_secs(5))
.expect("the dedicated thread must run with no runtime to capture");
assert_eq!(name.as_deref(), Some("dedicated-no-runtime"));
assert_eq!(value, Some(5));
joined.join().expect("dedicated thread joined");
}
#[test]
fn a_future_on_a_callback_band_is_refused_a_blocking_bridge() {
use crate::runtime::background::callback_executor::CallbackPool;
use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_future};
let pool = CallbackPool::new();
let (release, mut park_here) = tokio::sync::mpsc::channel::<()>(1);
let (report, outcome) = std::sync::mpsc::channel();
let _handle = spawn_future(&pool.handle(), DEFAULT_SPAWN_PRIORITY, async move {
let _ = report.send(block_on_sync(async move { park_here.recv().await }));
});
let got = outcome.recv_timeout(Duration::from_secs(5));
let _ = release.try_send(());
match got {
Ok(result) => assert_eq!(
result.map(|v| v.is_some()),
Err(NotBlockable::BackgroundWorker),
"a band worker must be refused the blocking bridge, not given one"
),
Err(_) => panic!(
"the band worker parked inside block_on_sync instead of being \
refused — the band has one worker, so this is the deadlock the \
invariant exists to prevent"
),
}
}
#[test]
fn a_blocking_closure_on_a_callback_band_is_refused_too() {
use crate::runtime::background::callback_executor::CallbackPool;
use crate::runtime::background::future_exec::{DEFAULT_SPAWN_PRIORITY, spawn_blocking_on};
let pool = CallbackPool::new();
let (release, mut park_here) = tokio::sync::mpsc::channel::<()>(1);
let (report, outcome) = std::sync::mpsc::channel();
let _handle = spawn_blocking_on(&pool.handle(), DEFAULT_SPAWN_PRIORITY, move || {
let _ = report.send(block_on_sync(async move { park_here.recv().await }));
});
let got = outcome.recv_timeout(Duration::from_secs(5));
let _ = release.try_send(());
match got {
Ok(result) => assert_eq!(
result.map(|v| v.is_some()),
Err(NotBlockable::BackgroundWorker),
"the refusal keys on the thread, not on how work reached it"
),
Err(_) => panic!("the band worker parked instead of being refused"),
}
}
#[test]
fn a_thread_that_only_submits_to_a_band_still_blocks() {
use crate::runtime::background::callback_executor::{CallbackPool, CallbackPriority};
let pool = CallbackPool::new();
let (tx, rx) = std::sync::mpsc::channel();
pool.request(
CallbackPriority::Medium,
Box::new(move || tx.send(1u32).unwrap()),
)
.expect("the band accepts the callback");
assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), 1);
assert_eq!(
block_on_sync(async { 5u32 }),
Ok(5),
"the submitting thread runs no facility loop, so it may still park"
);
}
#[epics_macros_rs::epics_test]
async fn test_sleep() {
let start = std::time::Instant::now();
sleep(Duration::from_millis(10)).await;
assert!(start.elapsed() >= Duration::from_millis(10));
}
#[epics_macros_rs::epics_test]
async fn timeout_yields_the_value_when_the_future_finishes_first() {
let r = timeout(Duration::from_secs(30), async { 42 }).await;
assert_eq!(r.unwrap(), 42);
}
#[epics_macros_rs::epics_test]
async fn timeout_elapses_on_a_future_that_never_finishes() {
let r = timeout(Duration::from_millis(10), std::future::pending::<()>()).await;
assert!(r.is_err());
}
#[test]
fn priority_named_levels_match_epics_thread_h() {
assert_eq!(ThreadPriority::Low.value(), 10);
assert_eq!(ThreadPriority::CaServerLow.value(), 20);
assert_eq!(ThreadPriority::CaServerHigh.value(), 40);
assert_eq!(ThreadPriority::Medium.value(), 50);
assert_eq!(ThreadPriority::ScanLow.value(), 60);
assert_eq!(ThreadPriority::ScanHigh.value(), 70);
assert_eq!(ThreadPriority::High.value(), 90);
assert_eq!(ThreadPriority::Iocsh.value(), 91);
}
#[test]
fn priority_ordering_ca_server_below_scan() {
assert!(ThreadPriority::CaServerHigh.value() < ThreadPriority::ScanLow.value());
assert!(ThreadPriority::CaServerLow.value() < ThreadPriority::ScanLow.value());
}
#[test]
fn priority_custom_clamps_to_max() {
assert_eq!(ThreadPriority::Custom(200).value(), PRIORITY_MAX);
assert_eq!(ThreadPriority::Custom(99).value(), 99);
assert_eq!(ThreadPriority::Custom(0).value(), PRIORITY_MIN);
}
#[test]
fn stack_size_classes_ordered() {
assert!(StackSizeClass::Small.bytes() < StackSizeClass::Medium.bytes());
assert!(StackSizeClass::Medium.bytes() < StackSizeClass::Big.bytes());
assert_eq!(
StackSizeClass::Small.bytes(),
0x10000 * std::mem::size_of::<usize>()
);
}
#[test]
fn the_classes_are_the_c_posix_table_factor_for_factor() {
let unit = 0x10000 * std::mem::size_of::<usize>();
assert_eq!(StackSizeClass::Small.bytes(), unit);
assert_eq!(StackSizeClass::Medium.bytes(), 2 * unit);
assert_eq!(StackSizeClass::Big.bytes(), 4 * unit);
const TARGET_UNIT: usize = 0x10000 * 4;
assert_eq!(
[TARGET_UNIT, 2 * TARGET_UNIT, 4 * TARGET_UNIT],
[256 * 1024, 512 * 1024, 1024 * 1024],
"armv7-rtems-eabihf: Small / Medium / Big in bytes"
);
}
#[cfg(all(target_os = "linux", target_env = "gnu"))]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_dedicated_thread_reports_the_stack_it_was_asked_for() {
fn reported_stack_bytes() -> usize {
unsafe {
let mut attr: libc::pthread_attr_t = std::mem::zeroed();
assert_eq!(
libc::pthread_getattr_np(libc::pthread_self(), &mut attr),
0,
"pthread_getattr_np"
);
let mut addr: *mut libc::c_void = std::ptr::null_mut();
let mut size: libc::size_t = 0;
assert_eq!(
libc::pthread_attr_getstack(&attr, &mut addr, &mut size),
0,
"pthread_attr_getstack"
);
libc::pthread_attr_destroy(&mut attr);
size
}
}
for class in [
StackSizeClass::Small,
StackSizeClass::Medium,
StackSizeClass::Big,
] {
let (tx, rx) = std::sync::mpsc::channel();
let joined = spawn_dedicated_thread(
format!("stack-readback-{class:?}"),
ThreadPriority::CaServerLow,
class,
move || {
let _ = tx.send(reported_stack_bytes());
},
)
.expect("dedicated thread spawned");
let got = rx.recv().expect("the thread reported its stack");
joined.join().expect("thread joined");
let asked = class.bytes();
assert!(
got >= asked && got < asked + 64 * 1024,
"{class:?}: asked for {asked} bytes, thread reports {got}"
);
}
}
#[cfg(all(target_os = "linux", target_env = "gnu"))]
#[test]
fn the_small_classes_are_below_the_default_that_would_mask_a_failure() {
const STD_DEFAULT_MIN_STACK_SIZE: usize = 2 * 1024 * 1024;
assert!(StackSizeClass::Small.bytes() < STD_DEFAULT_MIN_STACK_SIZE);
assert!(StackSizeClass::Medium.bytes() < STD_DEFAULT_MIN_STACK_SIZE);
}
#[test]
fn apply_priority_returns_a_defined_outcome() {
let outcome = apply_to_current_thread(ThreadPriority::ScanHigh);
assert!(matches!(
outcome,
PriorityApplied::Realtime
| PriorityApplied::Disabled
| PriorityApplied::Unsupported
| PriorityApplied::BestEffortFailed
));
}
#[test]
fn the_rt_default_is_on_for_rtems_and_off_for_hosted() {
assert_eq!(
default_policy(true),
RtPolicy::AllowRealtime,
"RTEMS honours its priorities by default, as base does \
(EPICS_ALLOW_POSIX_THREAD_PRIORITY_SCHEDULING=YES, CONFIG_ENV:57)"
);
assert_eq!(
default_policy(false),
RtPolicy::Disabled,
"hosted stays opt-in: RLIMIT_RTPRIO makes the request fail on a \
desktop, and where it succeeds a runaway band wedges the machine"
);
assert_eq!(DEFAULT_POLICY, default_policy(cfg!(epics_embedded_target)));
assert_eq!(RtPolicy::from_env_value(None), DEFAULT_POLICY);
for default in [RtPolicy::AllowRealtime, RtPolicy::Disabled] {
assert_eq!(
RtPolicy::resolve(Some("NO"), default),
RtPolicy::Disabled,
"explicit NO must turn it off even where the default is {default:?}"
);
assert_eq!(
RtPolicy::resolve(Some("YES"), default),
RtPolicy::AllowRealtime,
"explicit YES must turn it on even where the default is {default:?}"
);
assert_eq!(
RtPolicy::resolve(None, default),
default,
"unset must resolve to the default and nothing else"
);
}
}
#[test]
fn rt_switch_explicit_values_win_over_the_default() {
assert_eq!(RtPolicy::from_env_value(None), DEFAULT_POLICY);
for on in ["YES", "yes", "Yes", "true", "TRUE", "on", "1", " yes "] {
assert_eq!(
RtPolicy::from_env_value(Some(on)),
RtPolicy::AllowRealtime,
"{on:?} should turn the switch on"
);
}
for off in ["", "NO", "no", "false", "off", "0", "y", "yes please", "2"] {
assert_eq!(
RtPolicy::from_env_value(Some(off)),
RtPolicy::Disabled,
"{off:?} should leave the switch off"
);
}
}
#[cfg(target_os = "linux")]
#[test]
fn switch_off_makes_no_scheduler_calls() {
let before = SCHED_CALLS.with(std::cell::Cell::get);
for p in [
ThreadPriority::Low,
ThreadPriority::CaServerLow,
ThreadPriority::ScanHigh,
ThreadPriority::Iocsh,
ThreadPriority::Custom(0),
ThreadPriority::Custom(99),
] {
assert_eq!(
apply_to_current_thread_under(RtPolicy::Disabled, p),
PriorityApplied::Disabled
);
}
assert_eq!(
SCHED_CALLS.with(std::cell::Cell::get),
before,
"the switch is off, so nothing may reach pthread_setschedparam"
);
}
#[cfg(target_os = "linux")]
#[test]
fn switch_on_either_sticks_or_falls_back_without_killing_the_thread() {
let outcome = std::thread::spawn(|| {
let range = permitted_fifo_range();
let before = SCHED_CALLS.with(std::cell::Cell::get);
let outcome =
apply_to_current_thread_under(RtPolicy::AllowRealtime, ThreadPriority::ScanHigh);
let calls = SCHED_CALLS.with(std::cell::Cell::get) - before;
assert_eq!(2 + 2, 4);
let mut policy = 0i32;
let mut param = libc::sched_param { sched_priority: 0 };
let rc = unsafe {
libc::pthread_getschedparam(libc::pthread_self(), &mut policy, &mut param)
};
assert_eq!(rc, 0, "pthread_getschedparam failed");
match (range, outcome) {
(RtRange::Available { min, max }, PriorityApplied::Realtime) => {
assert_eq!(calls, 1, "one apply must be one scheduler call");
assert_eq!(policy, libc::SCHED_FIFO, "SCHED_FIFO did not stick");
assert_eq!(
param.sched_priority,
map_epics_priority(ThreadPriority::ScanHigh.value(), min, max),
"wrong OS priority for epicsThreadPriorityScanHigh"
);
}
(RtRange::Denied, PriorityApplied::BestEffortFailed) => {
assert_eq!(calls, 0, "a settled denial must not re-ask the OS");
assert_ne!(
policy,
libc::SCHED_FIFO,
"fallback reported but the thread is real-time scheduled"
);
}
(RtRange::Unsupported, PriorityApplied::Unsupported) => {
assert_eq!(calls, 0, "no SCHED_FIFO range means no scheduler call");
}
(range, outcome) => {
panic!("range {range:?} and outcome {outcome:?} disagree")
}
}
outcome
})
.join()
.expect("probe thread panicked");
eprintln!("host RT outcome: {outcome:?}");
}
#[cfg(target_os = "linux")]
#[test]
fn denial_is_reported_once_not_per_thread() {
let threads: Vec<_> = (0..8)
.map(|_| {
std::thread::spawn(|| {
for _ in 0..8 {
let _ = apply_to_current_thread_under(
RtPolicy::AllowRealtime,
ThreadPriority::Low,
);
}
})
})
.collect();
for t in threads {
t.join().expect("worker panicked");
}
assert!(
DENIED_WARNINGS.load(std::sync::atomic::Ordering::Relaxed) <= 1,
"64 denied requests across 8 threads must not produce more than one warning"
);
}
#[cfg(target_os = "linux")]
#[test]
fn epics_priority_maps_onto_the_permitted_fifo_range() {
let (min, max) = (1, 99);
assert_eq!(map_epics_priority(0, min, max), 1);
assert_eq!(map_epics_priority(20, min, max), 1 + (20.0 * 0.98) as i32);
assert_eq!(map_epics_priority(99, min, max), 1 + (99.0 * 0.98) as i32);
assert!(
map_epics_priority(ThreadPriority::CaServerHigh.value(), min, max)
< map_epics_priority(ThreadPriority::ScanLow.value(), min, max)
);
assert_eq!(map_epics_priority(0, 1, 10), 1);
assert_eq!(map_epics_priority(99, 1, 10), 1 + (99.0 * 0.09) as i32);
assert_eq!(map_epics_priority(50, 7, 7), 7);
}
#[test]
fn rtems_priority_map_stays_below_the_libbsd_network_band() {
const LIBBSD_IRQS_CORE: i32 = 96;
const LIBBSD_DEFAULT_CORE: i32 = 100;
const POSIX_MIN: i32 = 1;
const POSIX_MAX: i32 = 254;
for epics in 0..=u8::MAX {
let posix = map_epics_priority_rtems(epics);
let core = RTEMS_MAXIMUM_PRIORITY - posix;
assert!(
(POSIX_MIN..=POSIX_MAX).contains(&posix),
"EPICS {epics} maps to posix {posix}, outside the settable \
[{POSIX_MIN}, {POSIX_MAX}]"
);
assert!(
core >= LIBBSD_DEFAULT_CORE,
"EPICS {epics} maps to core {core}, more urgent than libbsd's \
default band ({LIBBSD_DEFAULT_CORE}) and its IRQS \
({LIBBSD_IRQS_CORE})"
);
assert!(
core <= RTEMS_MAXIMUM_PRIORITY - 56,
"EPICS {epics} maps to core {core}, less urgent than the \
EPICS band's own floor of 199"
);
}
assert_eq!(map_epics_priority_rtems(0), 56);
assert_eq!(map_epics_priority_rtems(99), 155);
assert_eq!(map_epics_priority_rtems(100), 155);
assert_eq!(map_epics_priority_rtems(u8::MAX), 155);
assert!(
RTEMS_MAXIMUM_PRIORITY - map_epics_priority_rtems(ThreadPriority::ScanLow.value())
< RTEMS_MAXIMUM_PRIORITY
- map_epics_priority_rtems(ThreadPriority::CaServerHigh.value())
);
}
#[test]
fn vxworks_priority_map_matches_the_rtems_posix_values() {
for epics in 0..=u8::MAX {
assert_eq!(
map_epics_priority_vxworks(epics),
map_epics_priority_rtems(epics),
"EPICS {epics}: VxWorks and RTEMS must set the identical POSIX \
SCHED_FIFO value"
);
}
assert_eq!(map_epics_priority_vxworks(0), 56);
assert_eq!(map_epics_priority_vxworks(99), 155);
assert_eq!(map_epics_priority_vxworks(100), 155);
assert_eq!(map_epics_priority_vxworks(u8::MAX), 155);
}
#[test]
fn rtems_priority_map_is_not_the_hosted_linear_map() {
const LIBBSD_IRQS_CORE: i32 = 96;
let (min, max) = (1, 254);
let hosted_core = |epics: u8| RTEMS_MAXIMUM_PRIORITY - map_epics_priority(epics, min, max);
let rtems_core = |epics: u8| RTEMS_MAXIMUM_PRIORITY - map_epics_priority_rtems(epics);
let hosted_above_irqs: Vec<u8> = (0..=99)
.filter(|&e| hosted_core(e) < LIBBSD_IRQS_CORE)
.collect();
let rtems_above_irqs: Vec<u8> = (0..=99)
.filter(|&e| rtems_core(e) < LIBBSD_IRQS_CORE)
.collect();
assert_eq!(
rtems_above_irqs,
Vec::<u8>::new(),
"the RTEMS map must place no EPICS priority above libbsd's IRQS"
);
assert_eq!(
hosted_above_irqs.first().copied(),
Some(63),
"base-on-RTEMS-6's posix map crosses IRQS at EPICS 63; that number \
is the recorded reason for this deviation"
);
assert_eq!(
hosted_core(91),
24,
"upstream posix map: EPICS 91 -> core 24"
);
assert_eq!(rtems_core(91), 108, "this port: EPICS 91 -> core 108");
assert_eq!(
map_epics_priority_rtems(99) - map_epics_priority_rtems(0),
99
);
assert_eq!(
map_epics_priority(99, min, max) - map_epics_priority(0, min, max),
250
);
}
#[test]
fn thread_names_are_cut_to_the_rtems_budget_on_a_char_boundary() {
assert_eq!(RTEMS_MAX_THREAD_NAME_BYTES, 15);
assert_eq!(truncate_thread_name("CAS-event"), "CAS-event");
assert_eq!(truncate_thread_name("123456789012345"), "123456789012345");
assert_eq!(
truncate_thread_name("CAS-client-blocking 10.0.0.1:5064"),
"CAS-client-blo"[..14].to_owned() + "c"
);
assert!(truncate_thread_name("CAS-client-blocking 10.0.0.1:5064").len() <= 15);
let mixed = "aaaaaaaaaaaaaaé";
assert_eq!(mixed.len(), 16);
assert_eq!(truncate_thread_name(mixed), "aaaaaaaaaaaaaa");
assert_eq!(truncate_thread_name(""), "");
}
#[test]
fn every_thread_in_this_crate_publishes_its_name() {
let files = [
("runtime/task.rs", include_str!("task.rs")),
(
"runtime/background/delayed_timer.rs",
include_str!("background/delayed_timer.rs"),
),
(
"runtime/background/scan_once.rs",
include_str!("background/scan_once.rs"),
),
(
"runtime/background/callback_executor.rs",
include_str!("background/callback_executor.rs"),
),
];
const EXEMPT: &str = ".name(\"cbRtProbe\".to_string())";
let mut anonymous = Vec::new();
let mut checked = 0usize;
for (label, src) in files {
for (n, after) in production_scope(src)
.split("thread::Builder::new()")
.skip(1)
.enumerate()
{
let (chain, body) = after.split_once(".spawn(").unwrap_or((after, ""));
if chain.contains(EXEMPT) {
continue;
}
checked += 1;
if !body.contains("enter_ioc_thread(") && !body.contains("name_current_thread()") {
anonymous.push(format!("{label} (Builder #{})", n + 1));
}
}
}
assert!(
checked >= 3,
"expected to find the crate's Builder sites, found {checked} — \
did a file move? update this guard's file list"
);
assert!(
anonymous.is_empty(),
"these threads are invisible in an RTEMS task listing: {anonymous:?}"
);
}
#[test]
fn only_the_prologue_reaches_the_banding_call() {
let files = [
("runtime/task.rs", include_str!("task.rs")),
(
"runtime/background/delayed_timer.rs",
include_str!("background/delayed_timer.rs"),
),
(
"runtime/background/scan_once.rs",
include_str!("background/scan_once.rs"),
),
(
"runtime/background/callback_executor.rs",
include_str!("background/callback_executor.rs"),
),
];
let allowed = [
"pub fn apply_to_current_thread(priority: ThreadPriority) -> PriorityApplied {",
"apply_to_current_thread(priority)",
];
let mut seen_definition = false;
for (label, src) in files {
let callers: Vec<&str> = production_scope(src)
.lines()
.map(str::trim)
.filter(|l| l.contains("apply_to_current_thread("))
.filter(|l| !l.starts_with("//"))
.collect();
seen_definition |= callers.contains(&allowed[0]);
let strays: Vec<&&str> = callers.iter().filter(|l| !allowed.contains(l)).collect();
assert!(
strays.is_empty(),
"{label}: only `enter_ioc_thread` may band a thread; \
everything else would band an OS-anonymous one — {strays:?}"
);
}
assert!(
seen_definition,
"the banding function moved out of this file list; update the guard"
);
}
#[test]
fn the_prologue_registers_the_thread_for_the_vxworks_census() {
let body = production_scope(include_str!("task.rs"))
.split_once("pub fn enter_ioc_thread(")
.expect("the prologue is still in this file")
.1
.split_once("\n}\n")
.expect("the prologue's body is terminated")
.0;
assert!(
body.contains("#[cfg(target_os = \"vxworks\")]"),
"the census registration must stay gated to the one OS whose \
backend needs it; `epics-rtems-boot` is a dependency of this \
package on that target only"
);
assert!(
body.contains("epics_rtems_boot::stats::register_task();"),
"VxWorks gives an RTP no task enumerator, so `dump_tasks` and \
`stack_report` list exactly what announced itself here"
);
}
#[test]
fn a_mandatory_thread_runs_under_its_declared_name() {
let (tx, rx) = std::sync::mpsc::channel();
let join = MandatoryThread::new(
"cbTestOwner",
ThreadPriority::ScanLow,
StackSizeClass::Small,
)
.spawn(move || {
let _ = tx.send(
std::thread::current()
.name()
.map(str::to_owned)
.unwrap_or_default(),
);
});
assert_eq!(rx.recv().expect("the body ran"), "cbTestOwner");
join.join().expect("the thread exited cleanly");
}
#[test]
fn try_spawn_hands_back_a_handle_on_success() {
let (tx, rx) = std::sync::mpsc::channel();
let join =
MandatoryThread::new("cbTestTry", ThreadPriority::ScanLow, StackSizeClass::Small)
.try_spawn(move || {
let _ = tx.send(());
})
.expect("a thread is creatable in the test environment");
rx.recv().expect("the body ran");
join.join().expect("the thread exited cleanly");
}
#[test]
fn the_fatal_message_names_the_thread_and_the_error() {
let msg = mandatory_thread_failure_message(
"scan-0.1",
&std::io::Error::from(std::io::ErrorKind::WouldBlock),
);
assert!(msg.contains("scan-0.1"), "{msg}");
assert!(msg.contains("FATAL"), "{msg}");
assert!(
msg.contains(&std::io::Error::from(std::io::ErrorKind::WouldBlock).to_string()),
"{msg}"
);
}
#[cfg(all(unix, not(target_os = "rtems"), not(target_os = "vxworks")))]
#[test]
fn a_mandatory_thread_that_cannot_be_created_aborts_the_process() {
use std::os::unix::process::ExitStatusExt;
const CHILD: &str = "EPICS_RS_MANDATORY_THREAD_ABORT_CHILD";
const TEST: &str =
"runtime::task::tests::a_mandatory_thread_that_cannot_be_created_aborts_the_process";
if std::env::var_os(CHILD).is_some() {
mandatory_thread_unavailable(
"scan-0.1",
&std::io::Error::from(std::io::ErrorKind::WouldBlock),
);
}
let out =
std::process::Command::new(std::env::current_exe().expect("the test binary path"))
.args(["--exact", TEST, "--nocapture"])
.env(CHILD, "1")
.output()
.expect("re-exec the test binary");
assert_eq!(
out.status.signal(),
Some(libc::SIGABRT),
"a mandatory thread's failure must abort the process, not unwind \
one thread — child exited {:?}, stderr: {}",
out.status,
String::from_utf8_lossy(&out.stderr)
);
let stderr = String::from_utf8_lossy(&out.stderr);
assert!(
stderr.contains("scan-0.1"),
"the console must name the thread; got: {stderr}"
);
}
#[epics_macros_rs::epics_test]
async fn spawn_blocking_with_priority_runs_closure() {
let handle = spawn_blocking_with_priority(ThreadPriority::CaServerHigh, || 7);
assert_eq!(handle.await.unwrap(), 7);
}
#[test]
fn background_global_inits_and_runs_work() {
background_init();
let exec = background();
let (tx, rx) = std::sync::mpsc::channel();
exec.callbacks()
.handle()
.request(
crate::runtime::background::CallbackPriority::Medium,
Box::new(move || tx.send(1u8).unwrap()),
)
.unwrap();
assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), 1);
}
}