wip

Author: divi255

Cargo.toml

@@ -15,19 +15,16 @@ object-id = "0.1"
 thiserror = "1.0"
 rtsc-derive = "0.1"
 pin-project = "1.1"
+parking_lot_rt = "0.12"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 lock_api = "0.4.12"
 libc = "0.2.155"
 linux-futex = "1.0.0"
 
-[target.'cfg(not(target_os = "linux"))'.dependencies]
-parking_lot_rt = "0.12"
-
 [dev-dependencies]
 insta = "1.39"
 tokio = { version = "1.36", features = ["rt", "macros", "time"] }
-parking_lot_rt = "0.12"
 
 [features]
 serde = ["lock_api/serde"]

src/buf.rs

@@ -69,7 +69,7 @@ mod test {
 
     #[test]
     fn test_data_buffer() {
-        let buf: DataBuffer<i32> = DataBuffer::bounded(3);
+        let buf: DataBuffer<_> = DataBuffer::bounded(3);
         assert_eq!(buf.len(), 0);
         buf.try_push(1);
         assert_eq!(buf.len(), 1);
@@ -85,7 +85,7 @@ mod test {
 
     #[test]
     fn test_data_buffer_other_mutex() {
-        let buf: DataBuffer<i32> = DataBuffer::bounded(3);
+        let buf: DataBuffer<i32, parking_lot_rt::RawMutex> = DataBuffer::bounded(3);
         assert_eq!(buf.len(), 0);
     }
 }

src/cell/datacell.rs

@@ -1,47 +1,69 @@
-use crate::{Error, Result};
+use crate::{
+    condvar_api::RawCondvar,
+    locking::{Condvar, RawMutex},
+    Error, Result,
+};
+use lock_api::RawMutex as RawMutexTrait;
 use std::{sync::Arc, time::Duration};
 
-use crate::locking::{Condvar, Mutex};
-
 struct CellValue<P> {
     current: Option<P>,
     closed: bool,
 }
 
-struct DataCellInner<P> {
-    value: Mutex<CellValue<P>>,
-    data_available: Condvar,
+impl<P> Default for CellValue<P> {
+    fn default() -> Self {
+        Self {
+            current: None,
+            closed: false,
+        }
+    }
+}
+
+struct DataCellInner<P, M, CV> {
+    value: lock_api::Mutex<M, CellValue<P>>,
+    data_available: CV,
 }
 
 /// A simple data cell that can be used to pass data between threads. Acts similarly to a
 /// ring-buffer channel with a capacity of 1.
-pub struct DataCell<P> {
-    inner: Arc<DataCellInner<P>>,
+pub struct DataCell<P, M = RawMutex, CV = Condvar>
+    {
+    inner: Arc<DataCellInner<P, M, CV>>,
 }
 
-impl<P> Default for DataCell<P> {
+impl<P, M, CV> Default for DataCell<P, M, CV>
+where
+    M: RawMutexTrait,
+    CV: RawCondvar,
+{
     fn default() -> Self {
         Self {
             inner: Arc::new(DataCellInner {
-                value: Mutex::new(CellValue {
-                    current: None,
-                    closed: false,
-                }),
-                data_available: Condvar::new(),
+                value: <_>::default(),
+                data_available: CV::new(),
             }),
         }
     }
 }
 
-impl<P> Clone for DataCell<P> {
+impl<P, M, CV> Clone for DataCell<P, M, CV>
+where
+    M: RawMutexTrait,
+    CV: RawCondvar,
+{
     fn clone(&self) -> Self {
         Self {
             inner: self.inner.clone(),
         }
     }
 }
 
-impl<P> DataCell<P> {
+impl<P, M, CV> DataCell<P, M, CV>
+where
+    M: RawMutexTrait,
+    CV: RawCondvar + RawCondvar<RawMutex = M>,
+{
     /// Creates a new data cell
     pub fn new() -> Self {
         Self::default()
@@ -80,7 +102,9 @@ impl<P> DataCell<P> {
             if let Some(current) = value.current.take() {
                 return Ok(current);
             }
-            self.inner.data_available.wait(&mut value);
+            self.inner
+                .data_available
+                .wait::<CellValue<P>, M>(&mut value);
         }
     }
     /// Retrieves the data from the cell with the given timeout
@@ -96,7 +120,7 @@ impl<P> DataCell<P> {
             if self
                 .inner
                 .data_available
-                .wait_for(&mut value, timeout)
+                .wait_for::<CellValue<P>, M>(&mut value, timeout)
                 .timed_out()
             {
                 return Err(Error::Timeout);
@@ -113,7 +137,11 @@ impl<P> DataCell<P> {
     }
 }
 
-impl<P> Iterator for DataCell<P> {
+impl<P, M, CV> Iterator for DataCell<P, M, CV>
+where
+    M: RawMutexTrait,
+    CV: RawCondvar + RawCondvar<RawMutex = M>,
+{
     type Item = P;
     fn next(&mut self) -> Option<Self::Item> {
         self.get().ok()
@@ -131,7 +159,20 @@ mod test {
 
     #[test]
     fn test_datacell() {
-        let cell = DataCell::new();
+        let cell: DataCell<_> = DataCell::new();
+        cell.set(32);
+        let cell2 = cell.clone();
+        let handle = thread::spawn(move || {
+            thread::sleep(Duration::from_millis(100));
+            cell2.set(42);
+        });
+        assert_eq!(cell.get().unwrap(), 42);
+        handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_datacell_other_mutex() {
+        let cell: DataCell<_, parking_lot_rt::RawMutex, parking_lot_rt::Condvar> = DataCell::new();
         let cell2 = cell.clone();
         let handle = thread::spawn(move || {
             thread::sleep(Duration::from_millis(100));
@@ -143,7 +184,7 @@ mod test {
 
     #[test]
     fn test_datacell_close() {
-        let cell = DataCell::new();
+        let cell: DataCell<_> = DataCell::new();
         let cell2 = cell.clone();
         let handle = thread::spawn(move || {
             cell2.set(42);
@@ -155,7 +196,7 @@ mod test {
 
     #[test]
     fn test_datacell_try_get() {
-        let cell = DataCell::new();
+        let cell: DataCell<_> = DataCell::new();
         assert_eq!(cell.try_get().unwrap_err(), Error::ChannelEmpty);
         let cell2 = cell.clone();
         let handle = thread::spawn(move || {

src/condvar_api.rs

@@ -0,0 +1,69 @@
+use std::time::Duration;
+
+use lock_api::{MutexGuard, RawMutex};
+
+/// The `Condvar` type trait
+pub trait RawCondvar {
+    /// The mutex type
+    type RawMutex: RawMutex;
+
+    /// Create a new `Condvar`
+    fn new() -> Self;
+    /// Wait on the `Condvar`
+    fn wait<T, M>(&self, mutex_guard: &mut MutexGuard<'_, Self::RawMutex, T>);
+    /// Wait on the `Condvar` with a timeout
+    fn wait_for<T, M>(
+        &self,
+        mutex_guard: &mut MutexGuard<'_, Self::RawMutex, T>,
+        timeout: Duration,
+    ) -> WaitTimeoutResult;
+    /// Notify one waiter
+    fn notify_one(&self);
+    /// Notify all waiters
+    fn notify_all(&self);
+}
+
+/// Result, returned by [`Condvar::wait_for`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct WaitTimeoutResult {
+    timed_out: bool,
+}
+
+impl WaitTimeoutResult {
+    /// Create a new `WaitTimeoutResult`
+    pub fn new(timed_out: bool) -> Self {
+        Self { timed_out }
+    }
+    /// Returns `true` if the wait timed out.
+    pub fn timed_out(&self) -> bool {
+        self.timed_out
+    }
+}
+
+impl RawCondvar for parking_lot_rt::Condvar {
+    type RawMutex = parking_lot_rt::RawMutex;
+
+    fn new() -> Self {
+        parking_lot_rt::Condvar::new()
+    }
+
+    fn wait<T, M>(&self, mutex_guard: &mut MutexGuard<'_, Self::RawMutex, T>) {
+        self.wait(mutex_guard)
+    }
+
+    fn wait_for<T, M>(
+        &self,
+        mutex_guard: &mut MutexGuard<'_, Self::RawMutex, T>,
+        timeout: Duration,
+    ) -> WaitTimeoutResult {
+        WaitTimeoutResult::new(self.wait_for(mutex_guard, timeout).timed_out())
+    }
+
+    fn notify_one(&self) {
+        self.notify_one();
+    }
+
+    fn notify_all(&self) {
+        self.notify_all();
+    }
+}

src/lib.rs

@@ -31,6 +31,8 @@ pub use bma_ts;
 pub mod base_channel;
 /// Base async channel type, allows to build async channels with a custom storage
 pub mod base_channel_async;
+/// Conditional traits
+pub mod condvar_api;
 /// Time-limited operations
 pub mod ops;
 /// Policy-based deque

src/pi.rs

@@ -7,6 +7,8 @@ use std::{
 use linux_futex::{AsFutex as _, Futex, PiFutex, Private, TimedWaitError, WaitError};
 use lock_api::{GuardSend, RawMutex as RawMutexTrait, RawMutexTimed};
 
+use crate::condvar_api::{RawCondvar, WaitTimeoutResult};
+
 thread_local! {
     #[allow(clippy::cast_possible_truncation)]
     static TID: libc::pid_t = unsafe { libc::syscall(libc::SYS_gettid) as i32 }
@@ -17,19 +19,6 @@ fn tid() -> libc::pid_t {
     TID.with(|it| *it)
 }
 
-/// Result, returned by [`Condvar::wait_for`].
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub struct WaitTimeoutResult {
-    timed_out: bool,
-}
-
-impl WaitTimeoutResult {
-    /// Returns `true` if the wait timed out.
-    pub fn timed_out(&self) -> bool {
-        self.timed_out
-    }
-}
-
 /// Priority-inheritance based Condvar implementation for the priority-inheritance [`Mutex`].
 #[derive(Default)]
 pub struct Condvar {
@@ -86,12 +75,12 @@ impl Condvar {
             let now = Instant::now();
             loop {
                 let Some(remaining) = timeout.checked_sub(now.elapsed()) else {
-                    break WaitTimeoutResult { timed_out: true };
+                    break WaitTimeoutResult::new(true)
                 };
                 unlock!();
                 match fx.wait_for(0, remaining) {
-                    Ok(()) => break WaitTimeoutResult { timed_out: false },
-                    Err(TimedWaitError::TimedOut) => break WaitTimeoutResult { timed_out: true },
+                    Ok(()) => break WaitTimeoutResult::new(false),
+                    Err(TimedWaitError::TimedOut) => break WaitTimeoutResult::new(true),
                     Err(TimedWaitError::Interrupted) => continue,
                     Err(TimedWaitError::WrongValue) => unreachable!(),
                 }
@@ -100,7 +89,7 @@ impl Condvar {
             loop {
                 unlock!();
                 match fx.wait(0) {
-                    Ok(()) => break WaitTimeoutResult { timed_out: false },
+                    Ok(()) => break WaitTimeoutResult::new(false),
                     Err(WaitError::Interrupted) => continue,
                     Err(WaitError::WrongValue) => unreachable!(),
                 }
@@ -295,6 +284,30 @@ pub type MutexGuard<'a, T> = lock_api::MutexGuard<'a, PiLock, T>;
 /// Compatibility name
 pub type RawMutex = PiLock;
 
+impl RawCondvar for Condvar {
+    type RawMutex = PiLock;
+
+    fn new() -> Self {
+        Self::new()
+    }
+
+    fn wait<T, M>(&self, guard: &mut MutexGuard<T>) {
+        self.wait(guard)
+    }
+
+    fn wait_for<T, M>(&self, guard: &mut MutexGuard<T>, timeout: Duration) -> WaitTimeoutResult {
+        self.wait_for(guard, timeout)
+    }
+
+    fn notify_one(&self) {
+        self.notify_one()
+    }
+
+    fn notify_all(&self) {
+        self.notify_all()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::{sync::Arc, thread, time::Duration};