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;
pub use crate::runtime::background::CallbackPriority;
#[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 try_capture() -> Option<Self> {
RuntimeHandle::try_current()
.ok()
.map(|handle| Self { handle })
}
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)
}
pub fn reactor(&self) -> Reactor {
Reactor {
handle: self.handle.clone(),
}
}
}
#[cfg(exec_backend)]
impl BlockingBridge {
pub fn capture() -> Self {
Self
}
pub fn try_capture() -> Option<Self> {
Some(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_background(CallbackPriority::Medium, future)
}
pub fn reactor(&self) -> Reactor {
Reactor
}
}
#[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;
static BACKGROUND: std::sync::OnceLock<crate::runtime::background::BackgroundExecutor> =
std::sync::OnceLock::new();
fn background() -> &'static crate::runtime::background::BackgroundExecutor {
BACKGROUND.get_or_init(crate::runtime::background::BackgroundExecutor::new)
}
pub fn background_init() {
let _ = background();
}
pub fn background_started() -> bool {
BACKGROUND.get().is_some()
}
pub fn background_callback_stats(
reset: bool,
) -> Option<
[crate::runtime::background::CallbackQueueStats;
crate::runtime::background::NUM_CALLBACK_PRIORITIES],
> {
let exec = BACKGROUND.get()?;
Some(
crate::runtime::background::CallbackPriority::ALL.map(|p| exec.callbacks().stats(p, reset)),
)
}
pub fn background_scan_once_start() {
background().scan_once().start();
}
pub fn background_scan_once_stats(
reset: bool,
) -> Option<crate::runtime::background::ScanOnceQueueStats> {
Some(BACKGROUND.get()?.scan_once().stats(reset))
}
pub type BackgroundTaskHandle<T> = crate::runtime::background::JoinFuture<T>;
pub fn spawn_background<F>(priority: CallbackPriority, future: F) -> BackgroundTaskHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
use crate::runtime::background::spawn_future;
spawn_future(&background().callbacks().handle(), priority, future)
}
pub fn spawn_blocking_background<F, R>(priority: CallbackPriority, f: F) -> BackgroundTaskHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
use crate::runtime::background::spawn_blocking_on;
spawn_blocking_on(&background().callbacks().handle(), priority, f)
}
pub async fn sleep_background(duration: Duration) {
crate::runtime::background::timer_sleep::sleep(&background().timer().handle(), duration).await;
}
pub fn interval_background(period: Duration) -> crate::runtime::background::TimerInterval {
crate::runtime::background::timer_sleep::interval(&background().timer().handle(), period)
}
#[cfg(tokio_backend)]
#[derive(Clone, Debug)]
pub struct Reactor {
handle: tokio::runtime::Handle,
}
#[cfg(exec_backend)]
#[derive(Clone, Debug)]
pub struct Reactor;
#[cfg(tokio_backend)]
impl Reactor {
pub fn current() -> Option<Self> {
RuntimeHandle::try_current()
.ok()
.map(|handle| Self { handle })
}
pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
self.handle.spawn(future)
}
pub fn spawn_blocking<F, R>(&self, f: F) -> TaskHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
self.handle.spawn_blocking(f)
}
}
#[cfg(exec_backend)]
impl Reactor {
pub fn current() -> Option<Self> {
Some(Self)
}
pub fn spawn<F>(&self, future: F) -> TaskHandle<F::Output>
where
F: Future + Send + 'static,
F::Output: Send + 'static,
{
spawn_background(CallbackPriority::Medium, future)
}
pub fn spawn_blocking<F, R>(&self, f: F) -> TaskHandle<R>
where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,
{
spawn_blocking_background(CallbackPriority::Medium, f)
}
}
#[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,
{
spawn_blocking_background(CallbackPriority::Medium, 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, reactor: &Reactor, future: F)
where
F: Future<Output = T> + Send + 'static,
{
self.tasks.push(reactor.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)]
async fn sleep_on_reachable_timer(duration: Duration) {
if tokio::runtime::Handle::try_current().is_ok() {
tokio::time::sleep(duration).await;
} else {
sleep_background(duration).await;
}
}
#[cfg(tokio_backend)]
pub async fn sleep(duration: Duration) {
sleep_on_reachable_timer(duration).await;
}
#[cfg(exec_backend)]
pub async fn sleep(duration: Duration) {
sleep_background(duration).await;
}
#[cfg(tokio_backend)]
pub type Instant = tokio::time::Instant;
#[cfg(exec_backend)]
pub type Instant = std::time::Instant;
pub fn deadline_after(base: Instant, d: Duration) -> Instant {
base.checked_add(d)
.unwrap_or_else(|| Instant::now() + crate::runtime::time::FAR_FUTURE)
}
pub fn deadline_from_now(d: Duration) -> Instant {
deadline_after(Instant::now(), d)
}
#[cfg(tokio_backend)]
pub async fn sleep_until(deadline: Instant) {
if tokio::runtime::Handle::try_current().is_ok() {
tokio::time::sleep_until(deadline).await;
} else {
sleep_on_reachable_timer(deadline.saturating_duration_since(Instant::now())).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)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Elapsed(());
impl std::fmt::Display for Elapsed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "deadline has elapsed")
}
}
impl std::error::Error for Elapsed {}
pub const SIGNIFICANT_DELAY: Duration = Duration::from_micros(1);
async fn poll_once_then_expire<F: Future>(fut: F) -> Result<F::Output, Elapsed> {
let mut fut = std::pin::pin!(fut);
std::future::poll_fn(move |cx| {
Poll::Ready(match fut.as_mut().poll(cx) {
Poll::Ready(v) => Ok(v),
Poll::Pending => Err(Elapsed(())),
})
})
.await
}
pub async fn timeout<F: Future>(duration: Duration, fut: F) -> Result<F::Output, Elapsed> {
if duration < SIGNIFICANT_DELAY {
return poll_once_then_expire(fut).await;
}
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
}
pub async fn timeout_at<F: Future>(deadline: Instant, fut: F) -> Result<F::Output, Elapsed> {
if deadline.saturating_duration_since(Instant::now()) < SIGNIFICANT_DELAY {
return poll_once_then_expire(fut).await;
}
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()),
}
}
thread_local! {
static OK_TO_BLOCK: std::cell::Cell<bool> = const { std::cell::Cell::new(true) };
}
pub fn thread_is_ok_to_block() -> bool {
OK_TO_BLOCK.with(std::cell::Cell::get)
}
pub fn set_thread_ok_to_block(ok: bool) {
OK_TO_BLOCK.with(|c| c.set(ok));
}
pub fn enter_ioc_thread(priority: ThreadPriority) -> PriorityApplied {
name_current_thread();
set_thread_ok_to_block(priority == ThreadPriority::Iocsh);
#[cfg(target_os = "vxworks")]
epics_rtems_boot::stats::register_task();
let applied = apply_to_current_thread(priority);
thread_registry::register_current(priority, applied);
applied
}
pub fn register_main_thread() {
thread_registry::register_main();
}
#[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]
}
#[derive(Debug, Default)]
struct Suspension {
suspended: std::sync::Mutex<bool>,
cv: std::sync::Condvar,
}
impl Suspension {
fn lock(&self) -> std::sync::MutexGuard<'_, bool> {
self.suspended.lock().unwrap_or_else(|e| e.into_inner())
}
fn suspend(&self) {
let mut suspended = self.lock();
*suspended = true;
while *suspended {
suspended = self.cv.wait(suspended).unwrap_or_else(|e| e.into_inner());
}
}
fn resume(&self) -> bool {
let mut suspended = self.lock();
if !*suspended {
return false;
}
*suspended = false;
self.cv.notify_all();
true
}
fn is_suspended(&self) -> bool {
*self.lock()
}
}
thread_local! {
static SUSPENSION: Arc<Suspension> = Arc::new(Suspension::default());
}
fn current_suspension() -> Arc<Suspension> {
SUSPENSION.with(Arc::clone)
}
pub fn suspend_self() {
thread_registry::register_current_implicit();
current_suspension().suspend();
}
pub fn resume_thread(id: u64) -> bool {
thread_report()
.into_iter()
.find(|t| t.id == id)
.is_some_and(|t| t.resume())
}
mod thread_registry {
use super::{PriorityApplied, ThreadPriority};
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
static THREADS: Mutex<Vec<super::ThreadInfo>> = Mutex::new(Vec::new());
static NEXT_ID: AtomicU64 = AtomicU64::new(1);
struct Registration(u64);
impl Drop for Registration {
fn drop(&mut self) {
let id = self.0;
lock().retain(|t| t.id != id);
}
}
thread_local! {
static REGISTRATION: std::cell::RefCell<Option<Registration>> =
const { std::cell::RefCell::new(None) };
}
fn lock() -> std::sync::MutexGuard<'static, Vec<super::ThreadInfo>> {
THREADS.lock().unwrap_or_else(|e| e.into_inner())
}
pub(super) fn register_current(priority: ThreadPriority, applied: PriorityApplied) {
push(
std::thread::current()
.name()
.unwrap_or("noname")
.to_string(),
priority,
applied,
);
}
pub(super) fn register_main() {
static ONCE: std::sync::Once = std::sync::Once::new();
ONCE.call_once(|| {
push(
"_main_".to_string(),
ThreadPriority::Custom(0),
PriorityApplied::Disabled,
);
});
}
fn push(name: String, priority: ThreadPriority, applied: PriorityApplied) {
let info = super::ThreadInfo {
id: NEXT_ID.fetch_add(1, Ordering::Relaxed),
name,
os_id: os_thread_id(),
priority,
applied,
suspension: super::current_suspension(),
};
let id = info.id;
lock().push(info);
let previous = REGISTRATION.with(|r| r.borrow_mut().replace(Registration(id)));
drop(previous);
}
pub(super) fn current_id() -> Option<u64> {
REGISTRATION.with(|r| r.borrow().as_ref().map(|reg| reg.0))
}
pub(super) fn register_current_implicit() {
if REGISTRATION.with(|r| r.borrow().is_some()) {
return;
}
push(
format!("non-EPICS_{}", os_thread_id()),
ThreadPriority::Custom(0),
PriorityApplied::Disabled,
);
}
pub(super) fn snapshot() -> Vec<super::ThreadInfo> {
lock().clone()
}
#[cfg(target_os = "linux")]
fn os_thread_id() -> u64 {
unsafe { libc::syscall(libc::SYS_gettid) as u64 }
}
#[cfg(target_os = "rtems")]
fn os_thread_id() -> u64 {
unsafe { super::rtems_sched::pthread_self() as u64 }
}
#[cfg(all(unix, not(any(target_os = "linux", target_os = "rtems"))))]
fn os_thread_id() -> u64 {
unsafe { libc::pthread_self() as u64 }
}
#[cfg(windows)]
fn os_thread_id() -> u64 {
unsafe { GetCurrentThreadId() }.into()
}
#[cfg(windows)]
unsafe extern "system" {
fn GetCurrentThreadId() -> u32;
}
}
#[derive(Clone, Debug)]
pub struct ThreadInfo {
id: u64,
name: String,
os_id: u64,
priority: ThreadPriority,
applied: PriorityApplied,
suspension: Arc<Suspension>,
}
impl ThreadInfo {
pub fn id(&self) -> u64 {
self.id
}
pub fn name(&self) -> &str {
&self.name
}
pub fn os_id(&self) -> u64 {
self.os_id
}
pub fn epics_priority(&self) -> u8 {
self.priority.value()
}
pub fn os_priority(&self) -> i32 {
if self.applied != PriorityApplied::Realtime {
return 0;
}
self.mapped_os_priority()
}
#[cfg(target_os = "linux")]
fn mapped_os_priority(&self) -> i32 {
match permitted_fifo_range() {
RtRange::Available { min, max } => map_epics_priority(self.priority.value(), min, max),
RtRange::Unsupported | RtRange::Denied => 0,
}
}
#[cfg(target_os = "rtems")]
fn mapped_os_priority(&self) -> i32 {
map_epics_priority_rtems(self.priority.value())
}
#[cfg(target_os = "vxworks")]
fn mapped_os_priority(&self) -> i32 {
map_epics_priority_vxworks(self.priority.value())
}
#[cfg(not(any(target_os = "linux", epics_embedded_target)))]
fn mapped_os_priority(&self) -> i32 {
0
}
pub fn is_suspended(&self) -> bool {
self.suspension.is_suspended()
}
pub fn resume(&self) -> bool {
self.suspension.resume()
}
pub fn show_line(&self) -> String {
format!(
"{:>16.16} {:>14} {:>os_id_width$} {:3}{:8} {:>8.8}",
self.name,
format!("{:#x}", self.id),
self.os_id,
self.epics_priority(),
self.os_priority(),
if self.is_suspended() { "SUSPEND" } else { "OK" },
os_id_width = OS_ID_WIDTH,
)
}
}
#[cfg(target_os = "linux")]
pub const THREAD_SHOW_HEADER: &str =
" NAME EPICS ID LWP ID OSIPRI OSSPRI STATE";
#[cfg(not(target_os = "linux"))]
pub const THREAD_SHOW_HEADER: &str =
" NAME EPICS ID PTHREAD ID OSIPRI OSSPRI STATE";
#[cfg(target_os = "linux")]
const OS_ID_WIDTH: usize = 8;
#[cfg(not(target_os = "linux"))]
const OS_ID_WIDTH: usize = 12;
pub fn thread_report() -> Vec<ThreadInfo> {
thread_registry::snapshot()
}
pub fn current_thread_id() -> u64 {
thread_registry::register_current_implicit();
thread_registry::current_id().expect("register_current_implicit installed a row")
}
pub fn thread_by_id(id: u64) -> Option<ThreadInfo> {
thread_report()
.into_iter()
.find(|t| t.id == id || t.os_id == id)
}
pub fn thread_by_name(name: &str) -> Option<ThreadInfo> {
thread_report().into_iter().find(|t| t.name == name)
}
pub fn osd_priority_range_line() -> String {
let (min, max) = osd_priority_range();
format!("OSD priority range min: {min} max {max}, memory not locked")
}
#[cfg(target_os = "linux")]
fn osd_priority_range() -> (i32, i32) {
match RtPolicy::current() {
RtPolicy::Disabled => (0, 0),
RtPolicy::AllowRealtime => match permitted_fifo_range() {
RtRange::Available { min, max } => (min, max),
RtRange::Denied => {
let min = unsafe { libc::sched_get_priority_min(libc::SCHED_FIFO) };
(min, min)
}
RtRange::Unsupported => (-1, -1),
},
}
}
#[cfg(target_os = "rtems")]
fn osd_priority_range() -> (i32, i32) {
match RtPolicy::current() {
RtPolicy::Disabled => (0, 0),
RtPolicy::AllowRealtime => (map_epics_priority_rtems(0), map_epics_priority_rtems(99)),
}
}
#[cfg(target_os = "vxworks")]
fn osd_priority_range() -> (i32, i32) {
match RtPolicy::current() {
RtPolicy::Disabled => (0, 0),
RtPolicy::AllowRealtime => (
map_epics_priority_vxworks(0),
map_epics_priority_vxworks(99),
),
}
}
#[cfg(not(any(target_os = "linux", epics_embedded_target)))]
fn osd_priority_range() -> (i32, i32) {
(0, 0)
}
#[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 charge = crate::runtime::worker_pool::ThreadCharge::fixed(StackSizeClass::Small);
let probe = std::thread::Builder::new()
.name("cbRtProbe".to_string())
.stack_size(StackSizeClass::Small.bytes())
.spawn(move || {
let _charge = charge;
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))]
pub(crate) fn map_epics_priority_rtems(epics_priority: u8) -> i32 {
RTEMS_MAXIMUM_PRIORITY - rtems_core_priority(epics_priority)
}
#[cfg(any(target_os = "vxworks", test))]
pub(crate) 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();
let charge = crate::runtime::worker_pool::ThreadCharge::fixed(stack);
std::thread::Builder::new()
.name(name)
.stack_size(stack.bytes())
.spawn(move || {
let _charge = charge;
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,
{
let charge = crate::runtime::worker_pool::ThreadCharge::fixed(stack);
std::thread::Builder::new()
.name(name)
.stack_size(stack.bytes())
.spawn(move || {
let _charge = charge;
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;
let charge = crate::runtime::worker_pool::ThreadCharge::fixed(self.stack);
std::thread::Builder::new()
.name(self.name)
.stack_size(self.stack.bytes())
.spawn(move || {
let _charge = charge;
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:939-955 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: &'static str) -> &'static str {
source_guard::production(src, source_guard::Comments::Keep)
}
fn crate_sources() -> Vec<(String, &'static str)> {
let files = source_guard::sweep(source_guard::module_dir!("src"), &[]);
assert!(
files.iter().any(|(label, _)| *label == "runtime/task.rs"),
"the crate sweep did not find runtime/task.rs, so it is reading \
the wrong directory and every guard below would pass vacuously"
);
files.into_iter().map(|(l, s)| (l.to_string(), s)).collect()
}
fn censused_files() -> Vec<(String, &'static str)> {
let bare = concat!("thread", "::spawn(");
let files: Vec<(String, &'static str)> = crate_sources()
.into_iter()
.filter(|(_, src)| {
let code = source_guard::production(src, source_guard::Comments::Strip);
code.contains("thread::Builder::new()") || code.contains(bare)
})
.collect();
assert!(
files.len() >= 2,
"expected at least `runtime/task.rs` and `runtime/worker_pool.rs` \
to create threads, found {files:?} — the derivation broke, and \
every guard over it would pass vacuously"
);
files
}
#[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;
let files = source_guard::sweep(source_guard::module_dir!("src/runtime/background"), &[]);
for (label, src) in files {
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 — a facility moved out of this module"
);
}
#[epics_macros_rs::epics_test]
async fn test_spawn() {
let reactor = Reactor::current().expect("the test driver enters an executor");
let handle = reactor.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 reactor = Reactor::current()
.expect("the dedicated thread carries the ambient runtime");
let inner = reactor.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 a_bounded_wait_fires_on_a_thread_with_no_runtime() {
assert!(
tokio::runtime::Handle::try_current().is_err(),
"the premise of this test"
);
let started = std::time::Instant::now();
let r = park_on(timeout(
Duration::from_millis(30),
std::future::pending::<()>(),
));
assert!(r.is_err(), "the deadline is the only thing that can end it");
let waited = started.elapsed();
assert!(
waited >= Duration::from_millis(30) && waited < Duration::from_secs(2),
"woke at its own deadline (waited {waited:?})"
);
}
#[test]
fn a_deadline_wait_fires_on_a_thread_with_no_runtime() {
let started = std::time::Instant::now();
let r = park_on(timeout_at(
deadline_from_now(Duration::from_millis(30)),
std::future::pending::<()>(),
));
assert!(r.is_err());
let waited = started.elapsed();
assert!(
waited >= Duration::from_millis(30) && waited < Duration::from_secs(2),
"woke at its own deadline (waited {waited:?})"
);
}
#[test]
fn a_sleep_on_a_thread_with_no_runtime_returns() {
let started = std::time::Instant::now();
park_on(sleep(Duration::from_millis(10)));
assert!(started.elapsed() >= Duration::from_millis(10));
}
#[cfg(tokio_backend)]
#[tokio::test]
async fn a_bounded_wait_on_a_runtime_thread_keeps_the_runtime_s_timer() {
assert!(
BACKGROUND.get().is_none(),
"the premise: nothing has started the background facility yet"
);
let r = timeout(Duration::from_millis(10), std::future::pending::<()>()).await;
assert!(r.is_err());
assert!(
BACKGROUND.get().is_none(),
"a runtime thread armed tokio's timer, not the background one"
);
}
#[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() {
const EXEMPT: &str = ".name(\"cbRtProbe\".to_string())";
let mut anonymous = Vec::new();
let mut checked = 0usize;
for (label, src) in censused_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 allowed = [
"pub fn apply_to_current_thread(priority: ThreadPriority) -> PriorityApplied {",
"let applied = apply_to_current_thread(priority);",
];
let mut seen_definition = false;
for (label, src) in crate_sources() {
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"
);
assert!(
body.contains("thread_registry::register_current(priority, applied);"),
"`epicsThreadShowAll` prints exactly what the prologue registered; \
a registration outside the prologue is a thread that can start \
without one"
);
}
#[test]
fn only_a_thread_that_ran_the_prologue_is_not_ok_to_block() {
assert!(
thread_is_ok_to_block(),
"a thread this module did not create is C's implicit context"
);
let banded = std::thread::spawn(|| {
let _ = enter_ioc_thread(ThreadPriority::ScanLow);
let after_prologue = thread_is_ok_to_block();
set_thread_ok_to_block(true);
(after_prologue, thread_is_ok_to_block())
})
.join()
.expect("the banded thread");
assert_eq!(banded, (false, true));
let shell = std::thread::spawn(|| {
let _ = enter_ioc_thread(ThreadPriority::Iocsh);
thread_is_ok_to_block()
})
.join()
.expect("the shell thread");
assert!(shell, "an iocsh thread boots a database and may block");
assert!(
thread_is_ok_to_block(),
"the flag is per-thread: the child's prologue must not clear ours"
);
}
#[test]
fn a_row_lasts_exactly_as_long_as_its_thread() {
let name = "cbRegLife";
assert!(thread_by_name(name).is_none(), "name must start unused");
let (started_tx, started_rx) = std::sync::mpsc::channel();
let (finish_tx, finish_rx) = std::sync::mpsc::channel();
let join = std::thread::Builder::new()
.name(name.to_string())
.spawn(move || {
let _ = enter_ioc_thread(ThreadPriority::ScanLow);
started_tx.send(()).unwrap();
finish_rx.recv().unwrap();
})
.unwrap();
started_rx.recv().unwrap();
let listed = thread_by_name(name).expect("listed while running");
assert_eq!(listed.epics_priority(), ThreadPriority::ScanLow.value());
assert_eq!(listed.name(), name);
assert_ne!(listed.id(), 0, "every row carries a usable EPICS ID");
finish_tx.send(()).unwrap();
join.join().unwrap();
assert!(
thread_by_name(name).is_none(),
"the row must be reaped with the thread, not left for the next reader"
);
}
#[test]
fn the_main_row_is_cs_once_row() {
let main_rows = || {
thread_report()
.into_iter()
.filter(|t| t.name() == "_main_")
.count()
};
assert_eq!(main_rows(), 0, "nothing registers `_main_` on its own");
let before = sched_calls_made();
let (registered_tx, registered_rx) = std::sync::mpsc::channel();
let (finish_tx, finish_rx) = std::sync::mpsc::channel();
let join = std::thread::spawn(move || {
register_main_thread();
registered_tx.send(()).unwrap();
finish_rx.recv().unwrap();
register_main_thread();
registered_tx.send(()).unwrap();
finish_rx.recv().unwrap();
});
registered_rx.recv().unwrap();
let row = thread_by_name("_main_").expect("listed while the shell runs");
assert_eq!(row.epics_priority(), 0, "C's `_main_` is EPICS priority 0");
assert_eq!(row.os_priority(), 0, "C changes no scheduling for `_main_`");
assert_eq!(
sched_calls_made(),
before,
"registering `_main_` must ask the scheduler for nothing"
);
assert_eq!(main_rows(), 1);
finish_tx.send(()).unwrap();
registered_rx.recv().unwrap();
assert_eq!(main_rows(), 1, "one process, one `_main_`");
finish_tx.send(()).unwrap();
join.join().unwrap();
assert_eq!(
main_rows(),
0,
"the row is reaped where C's thread-specific-data destructor \
removes it — when the thread holding it ends"
);
}
#[test]
fn re_entering_the_prologue_replaces_the_row_rather_than_adding_one() {
let name = "cbRegReband";
let (started_tx, started_rx) = std::sync::mpsc::channel();
let (finish_tx, finish_rx) = std::sync::mpsc::channel();
let join = std::thread::Builder::new()
.name(name.to_string())
.spawn(move || {
let _ = enter_ioc_thread(ThreadPriority::Low);
let _ = enter_ioc_thread(ThreadPriority::ScanHigh);
started_tx.send(()).unwrap();
finish_rx.recv().unwrap();
})
.unwrap();
started_rx.recv().unwrap();
let rows: Vec<ThreadInfo> = thread_report()
.into_iter()
.filter(|t| t.name() == name)
.collect();
assert_eq!(rows.len(), 1, "one thread is one row: {rows:?}");
assert_eq!(rows[0].epics_priority(), ThreadPriority::ScanHigh.value());
finish_tx.send(()).unwrap();
join.join().unwrap();
}
#[test]
fn a_row_resolves_by_epics_id_and_by_os_id() {
let name = "cbRegLookup";
let (started_tx, started_rx) = std::sync::mpsc::channel();
let (finish_tx, finish_rx) = std::sync::mpsc::channel();
let join = std::thread::Builder::new()
.name(name.to_string())
.spawn(move || {
let _ = enter_ioc_thread(ThreadPriority::Medium);
started_tx.send(()).unwrap();
finish_rx.recv().unwrap();
})
.unwrap();
started_rx.recv().unwrap();
let listed = thread_by_name(name).expect("listed while running");
assert_eq!(thread_by_id(listed.id()).map(|t| t.id()), Some(listed.id()));
assert_eq!(
thread_by_id(listed.os_id()).map(|t| t.id()),
Some(listed.id())
);
assert!(thread_by_id(u64::MAX).is_none());
finish_tx.send(()).unwrap();
join.join().unwrap();
}
#[test]
fn os_priority_is_zero_for_every_outcome_but_realtime() {
for applied in [
PriorityApplied::Disabled,
PriorityApplied::Unsupported,
PriorityApplied::BestEffortFailed,
] {
let row = ThreadInfo {
id: 1,
name: "cbOss".to_string(),
os_id: 7,
priority: ThreadPriority::High,
applied,
suspension: Arc::default(),
};
assert_eq!(row.os_priority(), 0, "{applied:?}");
assert_eq!(row.epics_priority(), ThreadPriority::High.value());
}
}
#[cfg(target_os = "linux")]
#[test]
fn show_line_matches_c_s_columns_byte_for_byte() {
let main = ThreadInfo {
id: 0x604f_304e_f4f0,
name: "_main_".to_string(),
os_id: 3_650_070,
priority: ThreadPriority::Custom(0),
applied: PriorityApplied::Disabled,
suspension: Arc::default(),
};
assert_eq!(
main.show_line(),
" _main_ 0x604f304ef4f0 3650070 0 0 OK"
);
let errlog = ThreadInfo {
id: 0x604f_304f_8a90,
name: "errlog".to_string(),
os_id: 3_650_072,
priority: ThreadPriority::Low,
applied: PriorityApplied::Disabled,
suspension: Arc::default(),
};
assert_eq!(
errlog.show_line(),
" errlog 0x604f304f8a90 3650072 10 0 OK"
);
assert_eq!(
THREAD_SHOW_HEADER,
" NAME EPICS ID LWP ID OSIPRI OSSPRI STATE"
);
}
#[cfg(not(target_os = "linux"))]
#[test]
fn show_line_matches_c_s_generic_posix_columns() {
let row = ThreadInfo {
id: 0x604f_304e_f4f0,
name: "_main_".to_string(),
os_id: 3_650_070,
priority: ThreadPriority::Custom(0),
applied: PriorityApplied::Disabled,
suspension: Arc::default(),
};
assert_eq!(
row.show_line(),
" _main_ 0x604f304ef4f0 3650070 0 0 OK"
);
assert_eq!(
THREAD_SHOW_HEADER,
" NAME EPICS ID PTHREAD ID OSIPRI OSSPRI STATE"
);
}
#[test]
fn a_suspended_thread_reads_suspend_and_only_a_resume_wakes_it() {
let (id_tx, id_rx) = std::sync::mpsc::channel();
let (go_tx, go_rx) = std::sync::mpsc::channel::<()>();
let (woke_tx, woke_rx) = std::sync::mpsc::channel();
let thread = std::thread::Builder::new()
.name("suspendTarget".to_string())
.spawn(move || {
let _ = enter_ioc_thread(ThreadPriority::ScanLow);
id_tx.send(current_thread_id()).unwrap();
go_rx.recv().unwrap();
suspend_self();
woke_tx.send(()).unwrap();
})
.unwrap();
let id = id_rx.recv().unwrap();
assert!(!resume_thread(id));
assert!(!resume_thread(id));
assert!(!thread_by_id(id).unwrap().is_suspended());
assert!(thread_by_id(id).unwrap().show_line().ends_with(" OK"));
go_tx.send(()).unwrap();
let row = (0..500)
.find_map(|_| {
let row = thread_by_id(id).expect("row");
row.is_suspended().then_some(row).or_else(|| {
std::thread::sleep(Duration::from_millis(10));
None
})
})
.expect("the thread must reach suspend_self");
assert!(
row.show_line().ends_with(" SUSPEND"),
"C prints SUSPEND for isSuspended (osdThreadExtra.c:53), got {:?}",
row.show_line()
);
assert_eq!(
woke_rx.recv_timeout(Duration::from_millis(250)),
Err(std::sync::mpsc::RecvTimeoutError::Timeout),
"the two earlier resumes must not have banked a wake-up"
);
assert!(
resume_thread(id),
"C epicsThreadResume on a suspended thread"
);
woke_rx
.recv_timeout(Duration::from_secs(5))
.expect("resumed");
thread.join().unwrap();
}
#[test]
fn a_thread_that_never_registered_gets_c_s_implicit_row() {
let (id, name, listed) = std::thread::spawn(|| {
let id = current_thread_id();
let row = thread_by_id(id).expect("createImplicit must put it on the list");
(id, row.name().to_string(), row.epics_priority())
})
.join()
.unwrap();
assert!(name.starts_with("non-EPICS_"), "C's name, got {name:?}");
assert_eq!(listed, 0, "C gives the implicit row osiPriority 0");
assert!(
thread_by_id(id).is_none(),
"the implicit row is reaped with the thread, as C frees it"
);
}
#[test]
fn a_long_name_is_truncated_rather_than_widening_the_row() {
let row = ThreadInfo {
id: 1,
name: "aVeryLongThreadNameIndeed".to_string(),
os_id: 7,
priority: ThreadPriority::Low,
applied: PriorityApplied::Disabled,
suspension: Arc::default(),
};
let short = ThreadInfo {
name: "s".to_string(),
..row.clone()
};
let line = row.show_line();
assert!(line.starts_with("aVeryLongThreadN "), "{line}");
assert_eq!(
line.len(),
short.show_line().len(),
"a row's width must not depend on its name"
);
}
#[test]
fn the_show_all_trailer_reports_the_range_this_process_may_enter() {
let line = osd_priority_range_line();
assert!(line.ends_with(", memory not locked"), "{line}");
if RtPolicy::current() == RtPolicy::Disabled {
assert_eq!(line, "OSD priority range min: 0 max 0, memory not locked");
}
}
#[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);
}
#[epics_macros_rs::epics_test]
async fn an_expired_budget_is_not_beaten_by_work_that_needs_a_second_poll() {
let r = timeout(Duration::ZERO, async {
yield_now().await;
7u8
})
.await;
assert!(
r.is_err(),
"a spent budget must expire before work that was not already done"
);
}
#[epics_macros_rs::epics_test]
async fn an_expired_budget_still_takes_work_that_is_already_done() {
let r = timeout(Duration::ZERO, std::future::ready(7u8)).await;
assert_eq!(r.ok(), Some(7));
}
#[epics_macros_rs::epics_test]
async fn a_deadline_already_in_the_past_expires_without_racing_the_work() {
let past = Instant::now() - Duration::from_secs(1);
let r = timeout_at(past, async {
yield_now().await;
7u8
})
.await;
assert!(r.is_err(), "a past deadline must not be raced either");
}
#[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);
}
}