pub use std::sync::Arc;
pub use tokio::sync::{Mutex, Notify, RwLock, broadcast, mpsc, oneshot};
#[cfg(all(target_os = "linux", feature = "linux-rt"))]
pub type PriorityInheritanceMutex<T> = pi_mutex::PiMutex<T>;
#[cfg(target_os = "rtems")]
pub type PriorityInheritanceMutex<T> = pi_mutex::PiMutex<T>;
#[cfg(not(any(all(target_os = "linux", feature = "linux-rt"), target_os = "rtems")))]
pub type PriorityInheritanceMutex<T> = parking_lot::Mutex<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),
}
}
#[cfg(test)]
pub fn raw_addr(&self) -> usize {
self.inner.get() as usize
}
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());
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[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);
}
}