condvar backoffs for thread with large RT priority delta

Author: divi255

src/pi.rs

@@ -1,5 +1,5 @@
 use std::{
-    sync::atomic::{AtomicI32, AtomicU32, Ordering},
+    sync::atomic::{AtomicI32, AtomicU32, Ordering, self},
     thread,
     time::{Duration, Instant},
 };
@@ -74,8 +74,12 @@ impl Condvar {
             "CRITICAL: too many waiters"
         );
         let mutex = unsafe { lock_api::MutexGuard::<'_, PiLock, T>::mutex(mutex_guard).raw() };
-        if mutex.is_locked() {
-            mutex.perform_unlock();
+        macro_rules! unlock {
+            () => {
+                if mutex.is_locked() {
+                    mutex.perform_unlock();
+                }
+            };
         }
         let fx: &Futex<Private> = self.fx.as_futex();
         let result = if let Some(timeout) = timeout {
@@ -84,6 +88,8 @@ impl Condvar {
                 let Some(remaining) = timeout.checked_sub(now.elapsed()) else {
                     break WaitTimeoutResult { timed_out: true };
                 };
+                unlock!();
+                //atomic::fence(Ordering::SeqCst);
                 match fx.wait_for(0, remaining) {
                     Ok(()) => break WaitTimeoutResult { timed_out: false },
                     Err(TimedWaitError::TimedOut) => break WaitTimeoutResult { timed_out: true },
@@ -93,6 +99,8 @@ impl Condvar {
             }
         } else {
             loop {
+                unlock!();
+                //atomic::fence(Ordering::SeqCst);
                 match fx.wait(0) {
                     Ok(()) => break WaitTimeoutResult { timed_out: false },
                     Err(WaitError::Interrupted) => continue,
@@ -108,25 +116,57 @@ impl Condvar {
     /// Notifies one thread waiting on this condvar.
     pub fn notify_one(&self) {
         let fx: &Futex<Private> = self.fx.as_futex();
+        let op_start = Instant::now();
+        let mut backoff = Backoff::new();
         while self.waiters.load(Ordering::SeqCst) > 0 && fx.wake(1) == 0 {
             // there is a chance that some waiter has not been entered into the futex yet, waiting
             // for it in a tiny spin loop
-            thread::yield_now();
+            backoff.backoff();
         }
+        println!("notify_one took {:?} ({})", op_start.elapsed(), backoff.c);
     }
 
     /// Notifies all threads waiting on this condvar.
     pub fn notify_all(&self) {
         let fx: &Futex<Private> = self.fx.as_futex();
+        let mut backoff = Backoff::new();
         loop {
             let to_wake = self.waiters.load(Ordering::SeqCst);
             if to_wake == 0 || fx.wake(to_wake) == to_wake {
                 break;
             }
             // there is a chance that some waiter has not been entered into the futex yet, waiting
             // for it in a tiny spin loop
-            thread::yield_now();
+            backoff.backoff();
+        }
+    }
+}
+
+struct Backoff {
+    n: u32,
+    c: u32,
+}
+
+impl Backoff {
+    fn new() -> Self {
+        Self { n: 0, c: 0 }
+    }
+
+    fn backoff(&mut self) {
+        if self.n == 0 {
+            // first iteration = spin
+            for _ in 0..10 {
+                thread::yield_now();
+            }
+        } else {
+            // otherwise sleep
+            thread::sleep(Duration::from_micros(self.n.into()));
+        }
+        if self.n < 100 {
+            // max sleep time = 100us
+            self.n += 50;
         }
+        self.c += 1;
     }
 }