mirror/rtic
1
0
Fork 0
mirror of https://github.com/rtic-rs/rtic.git synced 2025-12-18 22:05:37 +01:00
Code Issues Projects Releases Packages Wiki Activity

rtic-sync: Channel: Sender: rewriter send logic to be easier to validate

Browse source
This commit is contained in:
datdenkikniet 2025-03-13 22:41:16 +01:00 • committed by Emil Fresk
parent daf977dcff
commit 4fa3f5ddba
1 changed files with 36 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

41
rtic-sync/src/channel.rs
View file

@ -333,20 +333,33 @@ impl<T, const N: usize> Sender<'_, T, N> {
// Do all this in one critical section, else there can be race conditions // Do all this in one critical section, else there can be race conditions
critical_section::with(|cs| { critical_section::with(|cs| {
let wq_empty = self.0.wait_queue.is_empty(); let wq_empty = self.0.wait_queue.is_empty();
let fq_empty = self.0.access(cs).freeq.is_empty(); let freeq_empty = self.0.access(cs).freeq.is_empty();
if !wq_empty || fq_empty {
// SAFETY: This pointer is only dereferenced here and on drop of the future // SAFETY: This pointer is only dereferenced here and on drop of the future
// which happens outside this `poll_fn`'s stack frame. // which happens outside this `poll_fn`'s stack frame.
let link = unsafe { link_ptr.get() }; let link = unsafe { link_ptr.get() };
// We are already in the wait queue.
if let Some(link) = link { if let Some(link) = link {
if !link.is_popped() { if link.is_popped() {
return Poll::Pending; // If our link is popped, then:
// 1. We were popped by `try_recv` and it provided us with a slot.
// 2. We were popped by `Receiver::drop` and it did not provide us with a slot, and the channel is closed.
let slot = unsafe { free_slot_ptr.replace(None, cs) };
if let Some(slot) = slot {
Poll::Ready(Ok(slot))
} else { } else {
// Fall through to dequeue Poll::Ready(Err(()))
} }
} else { } else {
// Place the link in the wait queue on first run. Poll::Pending
}
}
// We are not in the wait queue, but others are, or there is currently no free
// slot available.
else if !wq_empty || freeq_empty {
// Place the link in the wait queue.
let link_ref = let link_ref =
link.insert(Link::new((cx.waker().clone(), free_slot_ptr.clone()))); link.insert(Link::new((cx.waker().clone(), free_slot_ptr.clone())));
@ -358,19 +371,13 @@ impl<T, const N: usize> Sender<'_, T, N> {
// `link_ptr` lives until the end of the stack frame. // `link_ptr` lives until the end of the stack frame.
unsafe { self.0.wait_queue.push(Pin::new_unchecked(link_ref)) }; unsafe { self.0.wait_queue.push(Pin::new_unchecked(link_ref)) };
return Poll::Pending; Poll::Pending
} }
} // We are not in the wait queue, no one else is waiting, and there is a free slot available.
else {
// SAFETY: `free_slot_ptr` is valid for writes, as `free_slot_ptr` is still alive. assert!(!self.0.access(cs).freeq.is_empty());
let slot = unsafe { free_slot_ptr.replace(None, cs) } let slot = unsafe { self.0.access(cs).freeq.pop_back_unchecked() };
.or_else(|| self.0.access(cs).freeq.pop_back());
if let Some(slot) = slot {
Poll::Ready(Ok(slot)) Poll::Ready(Ok(slot))
} else {
debug_assert!(self.is_closed());
Poll::Ready(Err(()))
} }
}) })
}) })