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

rtic-sync: always wake wait_queue when attempting to return an item

Browse source 
to freeq
This commit is contained in:
datdenkikniet 2025-03-15 14:35:58 +01:00 • committed by Emil Fresk
parent 4d58d2bcd5
commit e6bd03051f
1 changed files with 33 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -108,6 +108,30 @@ impl<T, const N: usize> Channel<T, N> {
}
}
}
/// Return free slot `slot` to the channel.
///
/// This will do one of two things:
/// 1. If there are any waiting `send`-ers, wake the longest-waiting one and hand it `slot`.
/// 2. else, insert `slot` into `self.freeq`.
///
/// SAFETY: `slot` must be a `u8` that is obtained by dequeueing from [`Self::readyq`].
unsafe fn return_free_slot(&self, slot: u8) {
critical_section::with(|cs| {
fence(Ordering::SeqCst);
// If someone is waiting in the `wait_queue`, wake the first one up & hand it the free slot.
if let Some((wait_head, mut freeq_slot)) = self.wait_queue.pop() {
// SAFETY: `freeq_slot` is valid for writes: we are in a critical
// section & the `SlotPtr` lives for at least the duration of the wait queue link.
unsafe { freeq_slot.replace(Some(slot), cs) };
wait_head.wake();
} else {
assert!(!self.access(cs).freeq.is_full());
unsafe { self.access(cs).freeq.push_back_unchecked(slot) }
}
})
}
}
/// Creates a split channel with `'static` lifetime.
@ -313,18 +337,16 @@ impl<T, const N: usize> Sender<'_, T, N> {
link.remove_from_list(&self.0.wait_queue);
}
// Return our potentially-unused free slot.
// Potentially unnecessary c-s because our link was already popped, so there
// is no way for anything else to access the free slot ptr. Gotta think
// about this a bit more...
//
// SAFETY(replace): `free_slot_ptr2` is valid for writes.
critical_section::with(|cs| {
if let Some(freed_slot) = unsafe { free_slot_ptr2.replace(None, cs) } {
debug_assert!(!self.0.access(cs).freeq.is_full());
// SAFETY: freeq is not full.
unsafe {
self.0.access(cs).freeq.push_back_unchecked(freed_slot);
}
// SAFETY: freed slot is passed to us from `return_free_slot`, which either
// directly (through `try_recv`), or indirectly (through another `return_free_slot`)
// comes from `readyq`.
unsafe { self.0.return_free_slot(freed_slot) };
}
});
});
@ -350,7 +372,7 @@ impl<T, const N: usize> Sender<'_, T, N> {
let slot = unsafe { free_slot_ptr.replace(None, cs) };
// If our link is popped, then:
// 1. We were popped by `try_recv` and it provided us with a slot.
// 1. We were popped by `return_free_lot` and 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.
if let Some(slot) = slot {
Poll::Ready(Ok(slot))
@ -482,22 +504,10 @@ impl<T, const N: usize> Receiver<'_, T, N> {
let r = unsafe { ptr::read(self.0.slots.get_unchecked(rs as usize).get() as *const T) };
// Return the index to the free queue after we've read the value.
critical_section::with(|cs| {
fence(Ordering::SeqCst);
// SAFETY: `rs` comes directly from `readyq`.
unsafe { self.0.return_free_slot(rs) };
// If someone is waiting in the WaiterQueue, wake the first one up & hand it the free slot.
if let Some((wait_head, mut freeq_slot)) = self.0.wait_queue.pop() {
// SAFETY: `freeq_slot` is valid for writes: we are in a critical
// section & the `SlotPtr` lives for at least the duration of the wait queue link.
unsafe { freeq_slot.replace(Some(rs), cs) };
wait_head.wake();
} else {
assert!(!self.0.access(cs).freeq.is_full());
unsafe { self.0.access(cs).freeq.push_back_unchecked(rs) }
}
Ok(r)
})
Ok(r)
} else if self.is_closed() {
Err(ReceiveError::NoSender)
} else {