library/std/src/sys/itron/condvar.rs

   1 //! POSIX conditional variable implementation based on user-space wait queues.
   2 use super::{abi, error::expect_success_aborting, spin::SpinMutex, task, time::with_tmos_strong};
   3 use crate::{mem::replace, ptr::NonNull, sys::locks::Mutex, time::Duration};
   4
   5 // The implementation is inspired by the queue-based implementation shown in
   6 // Andrew D. Birrell's paper "Implementing Condition Variables with Semaphores"
   7
   8 pub struct Condvar {
   9     waiters: SpinMutex<waiter_queue::WaiterQueue>,
  10 }
  11
  12 unsafe impl Send for Condvar {}
  13 unsafe impl Sync for Condvar {}
  14
  15 impl Condvar {
  16     #[inline]
  17     pub const fn new() -> Condvar {
  18         Condvar { waiters: SpinMutex::new(waiter_queue::WaiterQueue::new()) }
  19     }
  20
  21     pub fn notify_one(&self) {
  22         self.waiters.with_locked(|waiters| {
  23             if let Some(task) = waiters.pop_front() {
  24                 // Unpark the task
  25                 match unsafe { abi::wup_tsk(task) } {
  26                     // The task already has a token.
  27                     abi::E_QOVR => {}
  28                     // Can't undo the effect; abort the program on failure
  29                     er => {
  30                         expect_success_aborting(er, &"wup_tsk");
  31                     }
  32                 }
  33             }
  34         });
  35     }
  36
  37     pub fn notify_all(&self) {
  38         self.waiters.with_locked(|waiters| {
  39             while let Some(task) = waiters.pop_front() {
  40                 // Unpark the task
  41                 match unsafe { abi::wup_tsk(task) } {
  42                     // The task already has a token.
  43                     abi::E_QOVR => {}
  44                     // Can't undo the effect; abort the program on failure
  45                     er => {
  46                         expect_success_aborting(er, &"wup_tsk");
  47                     }
  48                 }
  49             }
  50         });
  51     }
  52
  53     pub unsafe fn wait(&self, mutex: &Mutex) {
  54         // Construct `Waiter`.
  55         let mut waiter = waiter_queue::Waiter::new();
  56         let waiter = NonNull::from(&mut waiter);
  57
  58         self.waiters.with_locked(|waiters| unsafe {
  59             waiters.insert(waiter);
  60         });
  61
  62         unsafe { mutex.unlock() };
  63
  64         // Wait until `waiter` is removed from the queue
  65         loop {
  66             // Park the current task
  67             expect_success_aborting(unsafe { abi::slp_tsk() }, &"slp_tsk");
  68
  69             if !self.waiters.with_locked(|waiters| unsafe { waiters.is_queued(waiter) }) {
  70                 break;
  71             }
  72         }
  73
  74         mutex.lock();
  75     }
  76
  77     pub unsafe fn wait_timeout(&self, mutex: &Mutex, dur: Duration) -> bool {
  78         // Construct and pin `Waiter`
  79         let mut waiter = waiter_queue::Waiter::new();
  80         let waiter = NonNull::from(&mut waiter);
  81
  82         self.waiters.with_locked(|waiters| unsafe {
  83             waiters.insert(waiter);
  84         });
  85
  86         unsafe { mutex.unlock() };
  87
  88         // Park the current task and do not wake up until the timeout elapses
  89         // or the task gets woken up by `notify_*`
  90         match with_tmos_strong(dur, |tmo| {
  91             let er = unsafe { abi::tslp_tsk(tmo) };
  92             if er == 0 {
  93                 // We were unparked. Are we really dequeued?
  94                 if self.waiters.with_locked(|waiters| unsafe { waiters.is_queued(waiter) }) {
  95                     // No we are not. Continue waiting.
  96                     return abi::E_TMOUT;
  97                 }
  98             }
  99             er
 100         }) {
 101             abi::E_TMOUT => {}
 102             er => {
 103                 expect_success_aborting(er, &"tslp_tsk");
 104             }
 105         }
 106
 107         // Remove `waiter` from `self.waiters`. If `waiter` is still in
 108         // `waiters`, it means we woke up because of a timeout. Otherwise,
 109         // we woke up because of `notify_*`.
 110         let success = self.waiters.with_locked(|waiters| unsafe { !waiters.remove(waiter) });
 111
 112         mutex.lock();
 113         success
 114     }
 115 }
 116
 117 mod waiter_queue {
 118     use super::*;
 119
 120     pub struct WaiterQueue {
 121         head: Option<ListHead>,
 122     }
 123
 124     #[derive(Copy, Clone)]
 125     struct ListHead {
 126         first: NonNull<Waiter>,
 127         last: NonNull<Waiter>,
 128     }
 129
 130     unsafe impl Send for ListHead {}
 131     unsafe impl Sync for ListHead {}
 132
 133     pub struct Waiter {
 134         // These fields are only accessed through `&[mut] WaiterQueue`.
 135         /// The waiting task's ID. Will be zeroed when the task is woken up
 136         /// and removed from a queue.
 137         task: abi::ID,
 138         priority: abi::PRI,
 139         prev: Option<NonNull<Waiter>>,
 140         next: Option<NonNull<Waiter>>,
 141     }
 142
 143     unsafe impl Send for Waiter {}
 144     unsafe impl Sync for Waiter {}
 145
 146     impl Waiter {
 147         #[inline]
 148         pub fn new() -> Self {
 149             let task = task::current_task_id();
 150             let priority = task::task_priority(abi::TSK_SELF);
 151
 152             // Zeroness of `Waiter::task` indicates whether the `Waiter` is
 153             // linked to a queue or not. This invariant is important for
 154             // the correctness.
 155             debug_assert_ne!(task, 0);
 156
 157             Self { task, priority, prev: None, next: None }
 158         }
 159     }
 160
 161     impl WaiterQueue {
 162         #[inline]
 163         pub const fn new() -> Self {
 164             Self { head: None }
 165         }
 166
 167         /// # Safety
 168         ///
 169         ///  - The caller must own `*waiter_ptr`. The caller will lose the
 170         ///    ownership until `*waiter_ptr` is removed from `self`.
 171         ///
 172         ///  - `*waiter_ptr` must be valid until it's removed from the queue.
 173         ///
 174         ///  - `*waiter_ptr` must not have been previously inserted to a `WaiterQueue`.
 175         ///
 176         pub unsafe fn insert(&mut self, mut waiter_ptr: NonNull<Waiter>) {
 177             unsafe {
 178                 let waiter = waiter_ptr.as_mut();
 179
 180                 debug_assert!(waiter.prev.is_none());
 181                 debug_assert!(waiter.next.is_none());
 182
 183                 if let Some(head) = &mut self.head {
 184                     // Find the insertion position and insert `waiter`
 185                     let insert_after = {
 186                         let mut cursor = head.last;
 187                         loop {
 188                             if waiter.priority >= cursor.as_ref().priority {
 189                                 // `cursor` and all previous waiters have the same or higher
 190                                 // priority than `current_task_priority`. Insert the new
 191                                 // waiter right after `cursor`.
 192                                 break Some(cursor);
 193                             }
 194                             cursor = if let Some(prev) = cursor.as_ref().prev {
 195                                 prev
 196                             } else {
 197                                 break None;
 198                             };
 199                         }
 200                     };
 201
 202                     if let Some(mut insert_after) = insert_after {
 203                         // Insert `waiter` after `insert_after`
 204                         let insert_before = insert_after.as_ref().next;
 205
 206                         waiter.prev = Some(insert_after);
 207                         insert_after.as_mut().next = Some(waiter_ptr);
 208
 209                         waiter.next = insert_before;
 210                         if let Some(mut insert_before) = insert_before {
 211                             insert_before.as_mut().prev = Some(waiter_ptr);
 212                         } else {
 213                             head.last = waiter_ptr;
 214                         }
 215                     } else {
 216                         // Insert `waiter` to the front
 217                         waiter.next = Some(head.first);
 218                         head.first.as_mut().prev = Some(waiter_ptr);
 219                         head.first = waiter_ptr;
 220                     }
 221                 } else {
 222                     // `waiter` is the only element
 223                     self.head = Some(ListHead { first: waiter_ptr, last: waiter_ptr });
 224                 }
 225             }
 226         }
 227
 228         /// Given a `Waiter` that was previously inserted to `self`, remove
 229         /// it from `self` if it's still there.
 230         #[inline]
 231         pub unsafe fn remove(&mut self, mut waiter_ptr: NonNull<Waiter>) -> bool {
 232             unsafe {
 233                 let waiter = waiter_ptr.as_mut();
 234                 if waiter.task != 0 {
 235                     let head = self.head.as_mut().unwrap();
 236
 237                     match (waiter.prev, waiter.next) {
 238                         (Some(mut prev), Some(mut next)) => {
 239                             prev.as_mut().next = Some(next);
 240                             next.as_mut().prev = Some(prev);
 241                         }
 242                         (None, Some(mut next)) => {
 243                             head.first = next;
 244                             next.as_mut().prev = None;
 245                         }
 246                         (Some(mut prev), None) => {
 247                             prev.as_mut().next = None;
 248                             head.last = prev;
 249                         }
 250                         (None, None) => {
 251                             self.head = None;
 252                         }
 253                     }
 254
 255                     waiter.task = 0;
 256
 257                     true
 258                 } else {
 259                     false
 260                 }
 261             }
 262         }
 263
 264         /// Given a `Waiter` that was previously inserted to `self`, return a
 265         /// flag indicating whether it's still in `self`.
 266         #[inline]
 267         pub unsafe fn is_queued(&self, waiter: NonNull<Waiter>) -> bool {
 268             unsafe { waiter.as_ref().task != 0 }
 269         }
 270
 271         #[inline]
 272         pub fn pop_front(&mut self) -> Option<abi::ID> {
 273             unsafe {
 274                 let head = self.head.as_mut()?;
 275                 let waiter = head.first.as_mut();
 276
 277                 // Get the ID
 278                 let id = replace(&mut waiter.task, 0);
 279
 280                 // Unlink the waiter
 281                 if let Some(mut next) = waiter.next {
 282                     head.first = next;
 283                     next.as_mut().prev = None;
 284                 } else {
 285                     self.head = None;
 286                 }
 287
 288                 Some(id)
 289             }
 290         }
 291     }
 292 }