pub use std::sync::Arc;
pub use tokio::sync::{Mutex, Notify, RwLock, broadcast, mpsc, oneshot};
#[cfg(all(target_os = "linux", feature = "linux-rt"))]
type MutexBackend<T> = pi_mutex::PiMutex<T>;
#[cfg(target_os = "rtems")]
type MutexBackend<T> = pi_mutex::PiMutex<T>;
#[cfg(not(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems")))]
type MutexBackend<T> = parking_lot::Mutex<T>;
pub type PriorityInheritanceMutex<T> = epics_mutex::EpicsMutex<T>;
#[cfg(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems"))]
pub type PriorityInheritanceMutexGuard<'a, T> = pi_mutex::PiMutexGuard<'a, T>;
#[cfg(not(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems")))]
pub type PriorityInheritanceMutexGuard<'a, T> = parking_lot::MutexGuard<'a, T>;
pub fn is_pi_mutex_active() -> bool {
#[cfg(target_os = "rtems")]
{
pi_mutex::protocol() == rtems_pi::PTHREAD_PRIO_INHERIT
}
#[cfg(not(target_os = "rtems"))]
{
cfg!(all(target_os = "linux", feature = "linux-rt"))
}
}
pub fn report_lock_protocol() {
let pi = if is_pi_mutex_active() {
"PI is enabled"
} else {
"PI is not enabled"
};
eprintln!(
"epics-rs: lock protocol: {pi}, RT scheduling {:?}",
crate::runtime::task::RtPolicy::current()
);
}
#[cfg(target_os = "rtems")]
mod rtems_pi {
use std::ffi::c_int;
pub const PTHREAD_PRIO_NONE: c_int = 0;
pub const PTHREAD_PRIO_INHERIT: c_int = 1;
unsafe extern "C" {
pub fn pthread_mutexattr_setprotocol(
attr: *mut libc::pthread_mutexattr_t,
protocol: c_int,
) -> c_int;
}
}
#[cfg(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems"))]
mod pi_mutex {
use std::cell::UnsafeCell;
use std::ffi::c_int;
use std::ops::{Deref, DerefMut};
#[cfg(target_os = "rtems")]
use super::rtems_pi;
#[cfg(target_os = "rtems")]
use super::rtems_pi::pthread_mutexattr_setprotocol;
#[cfg(not(target_os = "rtems"))]
use libc::pthread_mutexattr_setprotocol;
#[cfg(not(target_os = "rtems"))]
pub fn protocol() -> c_int {
libc::PTHREAD_PRIO_INHERIT
}
#[cfg(target_os = "rtems")]
pub fn protocol() -> c_int {
static PROTOCOL: std::sync::OnceLock<c_int> = std::sync::OnceLock::new();
*PROTOCOL.get_or_init(|| {
unsafe {
let mut attr: libc::pthread_mutexattr_t = std::mem::zeroed();
if libc::pthread_mutexattr_init(&mut attr) != 0 {
return rtems_pi::PTHREAD_PRIO_NONE;
}
let mut obtained = rtems_pi::PTHREAD_PRIO_NONE;
if pthread_mutexattr_setprotocol(&mut attr, rtems_pi::PTHREAD_PRIO_INHERIT) == 0 {
let mut probe: libc::pthread_mutex_t = std::mem::zeroed();
if libc::pthread_mutex_init(&mut probe, &attr) == 0 {
libc::pthread_mutex_destroy(&mut probe);
obtained = rtems_pi::PTHREAD_PRIO_INHERIT;
}
}
libc::pthread_mutexattr_destroy(&mut attr);
obtained
}
})
}
pub struct PiMutex<T> {
inner: Box<UnsafeCell<libc::pthread_mutex_t>>,
data: UnsafeCell<T>,
}
unsafe impl<T: Send> Send for PiMutex<T> {}
unsafe impl<T: Send> Sync for PiMutex<T> {}
impl<T> PiMutex<T> {
pub fn new(value: T) -> Self {
let mutex: Box<UnsafeCell<libc::pthread_mutex_t>> =
Box::new(UnsafeCell::new(unsafe { std::mem::zeroed() }));
unsafe {
let mut attr: libc::pthread_mutexattr_t = std::mem::zeroed();
let r = libc::pthread_mutexattr_init(&mut attr);
assert_eq!(r, 0, "pthread_mutexattr_init failed");
let protocol = protocol();
let r = pthread_mutexattr_setprotocol(&mut attr, protocol);
assert_eq!(r, 0, "pthread_mutexattr_setprotocol({protocol}) failed");
let r = libc::pthread_mutex_init(mutex.get(), &attr);
assert_eq!(r, 0, "pthread_mutex_init failed");
libc::pthread_mutexattr_destroy(&mut attr);
}
Self {
inner: mutex,
data: UnsafeCell::new(value),
}
}
pub fn raw_addr(&self) -> usize {
self.inner.get() as usize
}
pub fn try_lock(&self) -> Option<PiMutexGuard<'_, T>> {
if unsafe { libc::pthread_mutex_trylock(self.inner.get()) } != 0 {
return None;
}
Some(PiMutexGuard {
mutex: self,
_not_send: std::marker::PhantomData,
})
}
pub fn lock(&self) -> PiMutexGuard<'_, T> {
unsafe {
let r = libc::pthread_mutex_lock(self.inner.get());
assert_eq!(r, 0, "pthread_mutex_lock failed");
}
PiMutexGuard {
mutex: self,
_not_send: std::marker::PhantomData,
}
}
}
impl<T> Drop for PiMutex<T> {
fn drop(&mut self) {
unsafe {
libc::pthread_mutex_destroy(self.inner.get());
}
}
}
impl<T: std::fmt::Debug> std::fmt::Debug for PiMutex<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let acquired = unsafe { libc::pthread_mutex_trylock(self.inner.get()) } == 0;
if !acquired {
return f
.debug_struct("PiMutex")
.field("data", &"<locked>")
.finish();
}
let out = f
.debug_struct("PiMutex")
.field("data", unsafe { &*self.data.get() })
.finish();
unsafe {
libc::pthread_mutex_unlock(self.inner.get());
}
out
}
}
pub struct PiMutexGuard<'a, T> {
mutex: &'a PiMutex<T>,
_not_send: std::marker::PhantomData<*const ()>,
}
impl<T> Deref for PiMutexGuard<'_, T> {
type Target = T;
fn deref(&self) -> &T {
unsafe { &*self.mutex.data.get() }
}
}
impl<T> DerefMut for PiMutexGuard<'_, T> {
fn deref_mut(&mut self) -> &mut T {
unsafe { &mut *self.mutex.data.get() }
}
}
impl<T> Drop for PiMutexGuard<'_, T> {
fn drop(&mut self) {
unsafe {
libc::pthread_mutex_unlock(self.mutex.inner.get());
}
}
}
}
mod epics_mutex {
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use super::MutexBackend;
struct Entry {
id: u64,
file: &'static str,
line: u32,
addr: usize,
osd_addr: usize,
probe: unsafe fn(usize) -> bool,
}
static MUTEXES: Mutex<Vec<Entry>> = Mutex::new(Vec::new());
static NEXT_ID: AtomicU64 = AtomicU64::new(1);
fn lock() -> std::sync::MutexGuard<'static, Vec<Entry>> {
MUTEXES.lock().unwrap_or_else(|e| e.into_inner())
}
pub struct EpicsMutex<T> {
inner: Box<MutexBackend<T>>,
id: u64,
}
impl<T> EpicsMutex<T> {
#[track_caller]
pub fn new(value: T) -> Self {
let inner = Box::new(MutexBackend::new(value));
let caller = std::panic::Location::caller();
let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
lock().push(Entry {
id,
file: caller.file(),
line: caller.line(),
addr: &*inner as *const MutexBackend<T> as usize,
osd_addr: osd_addr(&inner),
probe: probe_locked::<T>,
});
Self { inner, id }
}
pub fn lock(&self) -> super::PriorityInheritanceMutexGuard<'_, T> {
self.inner.lock()
}
pub fn try_lock(&self) -> Option<super::PriorityInheritanceMutexGuard<'_, T>> {
self.inner.try_lock()
}
#[cfg(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems"))]
pub fn raw_addr(&self) -> usize {
self.inner.raw_addr()
}
}
impl<T> Drop for EpicsMutex<T> {
fn drop(&mut self) {
let id = self.id;
lock().retain(|entry| entry.id != id);
}
}
impl<T: std::fmt::Debug> std::fmt::Debug for EpicsMutex<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.inner.fmt(f)
}
}
#[cfg(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems"))]
fn osd_addr<T>(inner: &MutexBackend<T>) -> usize {
inner.raw_addr()
}
#[cfg(not(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems")))]
fn osd_addr<T>(inner: &MutexBackend<T>) -> usize {
inner as *const MutexBackend<T> as usize
}
unsafe fn probe_locked<T>(addr: usize) -> bool {
let backend = unsafe { &*(addr as *const MutexBackend<T>) };
match backend.try_lock() {
Some(guard) => {
drop(guard);
false
}
None => true,
}
}
#[derive(Clone, Debug)]
pub struct MutexInfo {
addr: usize,
osd_addr: usize,
file: &'static str,
line: u32,
}
impl MutexInfo {
pub fn addr(&self) -> usize {
self.addr
}
pub fn file(&self) -> &'static str {
self.file
}
pub fn line(&self) -> u32 {
self.line
}
pub fn show_lines(&self, level: u32) -> Vec<String> {
let mut lines = vec![format!(
"epicsMutexId {:#x} source {} line {}",
self.addr, self.file, self.line
)];
if level > 0 {
lines.push(format!(
" {} uaddr={:#x}",
super::MUTEX_OSD_LABEL,
self.osd_addr
));
}
lines
}
}
pub struct MutexReport {
pub total: usize,
pub shown: Vec<MutexInfo>,
}
pub fn report(only_locked: bool) -> MutexReport {
let entries = lock();
let mut shown = Vec::new();
for entry in entries.iter() {
if only_locked {
if !unsafe { (entry.probe)(entry.addr) } {
continue;
}
}
shown.push(MutexInfo {
addr: entry.addr,
osd_addr: entry.osd_addr,
file: entry.file,
line: entry.line,
});
}
MutexReport {
total: entries.len(),
shown,
}
}
}
pub use epics_mutex::{MutexInfo, MutexReport};
#[cfg(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems"))]
pub const MUTEX_OSD_LABEL: &str = "pthread_mutex_t*";
#[cfg(not(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems")))]
pub const MUTEX_OSD_LABEL: &str = "parking_lot::Mutex*";
pub fn mutex_report(only_locked: bool) -> MutexReport {
epics_mutex::report(only_locked)
}
pub fn osd_show_all_line() -> &'static str {
if is_pi_mutex_active() {
"PI is enabled"
} else {
"PI is not enabled"
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn an_entry_lasts_exactly_as_long_as_its_mutex() {
let expected_line = line!() + 1;
let m: PriorityInheritanceMutex<i32> = PriorityInheritanceMutex::new(5);
let addr = find_entry(&m).expect("registered at construction").addr();
let entry = find_entry(&m).unwrap();
assert_eq!(entry.file(), file!(), "C's `pFileName` is the caller's");
assert_eq!(entry.line(), expected_line, "C's `lineno` is the caller's");
drop(m);
assert!(
!mutex_report(false).shown.iter().any(|e| e.addr() == addr),
"the entry must come off the list before the mutex is freed"
);
}
fn find_entry<T>(m: &PriorityInheritanceMutex<T>) -> Option<MutexInfo> {
let want = mutex_addr(m);
mutex_report(false)
.shown
.into_iter()
.find(|e| e.addr() == want)
}
fn mutex_addr<T>(m: &PriorityInheritanceMutex<T>) -> usize {
let guard = m.try_lock();
let held = guard.is_none();
drop(guard);
assert!(!held, "helper must not be called on a held mutex");
let before: Vec<usize> = mutex_report(true).shown.iter().map(|e| e.addr()).collect();
let _g = m.lock();
let after: Vec<usize> = mutex_report(true).shown.iter().map(|e| e.addr()).collect();
after.into_iter().find(|a| !before.contains(a)).unwrap()
}
#[test]
fn only_locked_keeps_exactly_the_held_mutexes() {
let m: PriorityInheritanceMutex<i32> = PriorityInheritanceMutex::new(0);
let addr = mutex_addr(&m);
let other: PriorityInheritanceMutex<i32> = PriorityInheritanceMutex::new(0);
let other_addr = mutex_addr(&other);
let free = mutex_report(true);
assert!(
!free.shown.iter().any(|e| e.addr() == addr),
"an unheld mutex must not be listed under onlyLocked"
);
let guard = m.lock();
let held = mutex_report(true);
assert!(
held.shown.iter().any(|e| e.addr() == addr),
"a held mutex must be listed under onlyLocked"
);
assert_eq!(
held.total, free.total,
"the count is the whole list, not the filtered rows — C prints \
`ellCount(&mutexList)` before it filters"
);
assert!(
!held.shown.iter().any(|e| e.addr() == other_addr),
"the filter must exclude the mutex nobody holds"
);
assert!(
held.shown.len() < held.total,
"{} of {}",
held.shown.len(),
held.total
);
drop(guard);
assert!(!mutex_report(true).shown.iter().any(|e| e.addr() == addr));
}
#[test]
fn the_registered_address_survives_moving_the_mutex() {
let m: PriorityInheritanceMutex<i32> = PriorityInheritanceMutex::new(1);
let addr = mutex_addr(&m);
let moved = Box::new(m);
assert_eq!(mutex_addr(&moved), addr);
assert!(mutex_report(false).shown.iter().any(|e| e.addr() == addr));
drop(moved);
assert!(!mutex_report(false).shown.iter().any(|e| e.addr() == addr));
}
#[test]
fn show_lines_adds_the_osd_line_only_above_level_zero() {
let m: PriorityInheritanceMutex<i32> = PriorityInheritanceMutex::new(0);
let entry = find_entry(&m).unwrap();
let plain = entry.show_lines(0);
assert_eq!(plain.len(), 1);
assert_eq!(
plain[0],
format!(
"epicsMutexId {:#x} source {} line {}",
entry.addr(),
entry.file(),
entry.line()
)
);
let detailed = entry.show_lines(1);
assert_eq!(detailed.len(), 2);
assert_eq!(detailed[0], plain[0]);
assert!(
detailed[1].starts_with(&format!(" {MUTEX_OSD_LABEL} uaddr=0x")),
"{}",
detailed[1]
);
}
#[test]
fn the_osd_show_all_line_is_the_pi_report() {
assert_eq!(
osd_show_all_line(),
if is_pi_mutex_active() {
"PI is enabled"
} else {
"PI is not enabled"
}
);
}
#[test]
fn pi_mutex_lock_unlock() {
let m: PriorityInheritanceMutex<i32> = PriorityInheritanceMutex::new(42);
{
let g = m.lock();
assert_eq!(*g, 42);
}
let g = m.lock();
assert_eq!(*g, 42);
}
#[test]
fn is_pi_mutex_active_matches_the_cfg_arm() {
#[cfg(all(target_os = "linux", feature = "linux-rt"))]
assert!(
is_pi_mutex_active(),
"linux-rt selects the pthread PI arm unconditionally"
);
#[cfg(target_os = "rtems")]
assert_eq!(
is_pi_mutex_active(),
pi_mutex::protocol() == rtems_pi::PTHREAD_PRIO_INHERIT,
"the RTEMS report must be the probe result, not the cfg"
);
#[cfg(not(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems")))]
assert!(
!is_pi_mutex_active(),
"the parking_lot fallback arm has no priority inheritance"
);
}
#[cfg(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems"))]
#[test]
fn the_pthread_object_does_not_move_with_the_mutex() {
let m: PriorityInheritanceMutex<i32> = PriorityInheritanceMutex::new(7);
let before = m.raw_addr();
let moved = Box::new(m);
assert_eq!(
moved.raw_addr(),
before,
"the pthread_mutex_t moved with its owner; RTEMS answers EINVAL to \
every lock on a relocated mutex"
);
assert_eq!(*moved.lock(), 7);
}
#[test]
fn pi_mutex_serialises_concurrent_writers() {
const THREADS: u64 = 8;
const PER_THREAD: u64 = 10_000;
let m: Arc<PriorityInheritanceMutex<u64>> = Arc::new(PriorityInheritanceMutex::new(0));
let workers: Vec<_> = (0..THREADS)
.map(|_| {
let m = Arc::clone(&m);
std::thread::spawn(move || {
for _ in 0..PER_THREAD {
*m.lock() += 1;
}
})
})
.collect();
for w in workers {
w.join().expect("worker panicked");
}
assert_eq!(*m.lock(), THREADS * PER_THREAD);
}
}