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