diff --git a/rtic-channel/src/lib.rs b/rtic-channel/src/lib.rs index eafa25c3bf..acfa801315 100644 --- a/rtic-channel/src/lib.rs +++ b/rtic-channel/src/lib.rs @@ -532,7 +532,7 @@ mod tests { #[tokio::test] async fn stress_channel() { - const NUM_RUNS: usize = 1_000000; + const NUM_RUNS: usize = 1_000; const QUEUE_SIZE: usize = 10; let (s, mut r) = make_channel!(u32, QUEUE_SIZE); diff --git a/rtic-channel/src/wait_queue.rs b/rtic-channel/src/wait_queue.rs index e6d5a8b97e..2de6311d17 100644 --- a/rtic-channel/src/wait_queue.rs +++ b/rtic-channel/src/wait_queue.rs @@ -105,8 +105,6 @@ pub struct Link { _up: PhantomPinned, } -unsafe impl Send for Link {} - impl Link { const R: Ordering = Ordering::Relaxed; diff --git a/rtic-time/src/lib.rs b/rtic-time/src/lib.rs index 6b23f7653d..44fdbcecfc 100644 --- a/rtic-time/src/lib.rs +++ b/rtic-time/src/lib.rs @@ -73,6 +73,26 @@ pub struct TimerQueue { /// This indicates that there was a timeout. pub struct TimeoutError; +/// This is needed to make the async closure in `delay_until` accept that we "share" +/// the link possible between threads. +struct LinkPtr(*mut Option>>); + +impl Clone for LinkPtr { + fn clone(&self) -> Self { + LinkPtr(self.0) + } +} + +impl LinkPtr { + /// This will dereference the pointer stored within and give out an `&mut`. + unsafe fn get(&mut self) -> &mut Option>> { + &mut *self.0 + } +} + +unsafe impl Send for LinkPtr {} +unsafe impl Sync for LinkPtr {} + impl TimerQueue { /// Make a new queue. pub const fn new() -> Self { @@ -189,7 +209,9 @@ impl TimerQueue { let mut link_ptr: Option>> = None; // Make this future `Drop`-safe, also shadow the original definition so we can't abuse it. - let link_ptr = &mut link_ptr as *mut Option>>; + let mut link_ptr = + LinkPtr(&mut link_ptr as *mut Option>>); + let mut link_ptr2 = link_ptr.clone(); let queue = &self.queue; let marker = &AtomicUsize::new(0); @@ -205,7 +227,7 @@ impl TimerQueue { // SAFETY: This pointer is only dereferenced here and on drop of the future // which happens outside this `poll_fn`'s stack frame. - let link = unsafe { &mut *link_ptr }; + let link = unsafe { link_ptr2.get() }; if link.is_none() { let mut link_ref = link.insert(Link::new(WaitingWaker { waker: cx.waker().clone(), @@ -231,7 +253,7 @@ impl TimerQueue { // SAFETY: We only run this and dereference the pointer if we have // exited the `poll_fn` below in the `drop(dropper)` call. The other dereference // of this pointer is in the `poll_fn`. - if let Some(link) = unsafe { &mut *link_ptr } { + if let Some(link) = unsafe { link_ptr.get() } { if link.val.was_poped.load(Ordering::Relaxed) { // If it was poped from the queue there is no need to run delete dropper.defuse();