Add Watch sync primitive (similar to tokio::sync::watch)
#2568
Conversation
peterkrull
changed the title
MultiSignal sync primitive (similar to tokio::sync::watch)Watch sync primitive (similar to tokio::sync::watch)
…-types, revised tests.
…ly way it can fail.
|
I am considering removing the peek methods for the receivers. Their use case is dubious, and not marking a message as seen can make stuff hard to reason about for the user. I have also reintroduce predicate methods, which allows the user to await (or try) a change which also meets a certain criteria, defined by a closure |
|
This would be helpful for my work as well. I make a lot of channels with a depth of 1. If there's anything I can do to help push this through let me know. I agree about the peek functions making reasoning hard. I would remove them, though that isn't a super strong preference if someone has a compelling use. |
|
I would be happy to get back to this. I do have some changes to the version I have been using locally. I have abandoned the peek functions since I did not find them useful anyway. I have also put in place (for my own sake) an arbitrary limit on the senders, so there can only be one sender for a Watch channel, but I am okay with reverting this. I also have what I call an |
|
Hmm, I have used depth-1 pubsubs with multiple senders in a couple of cases. First is having something like a disable motors pubsub, that can be published to by either an estop button or i2c task. This could be fairly easily built up from 3 Watch objects and an intermediate task. But as the number of disable sources increase it could get unwieldy. Second is a way to request a sensor reading, many tasks can ask for a new value, and the task responsible for reading from that sensor reads and sends it back on another pubsub. The first part of that can be done with a signal, but I'm not a fan of it's ergonomics. Possibly a Watch-but-backwards would be a useful tool as well. The code may well be identical, but it might be valuable to have a logical separation between the many-to-one and one-to-many primitives. |
|
I mainly intended for this to be a one-to-many channel, since that is what we cannot do with a signal, but there is really no reason why I should impose this hard constraint on others. I will just revert the one-sender limitation.
I actually think it is the other way around, with just a couple of different sources of estop, having shared over-write access is probably fine, but with many more sources I like the approach of having a task specifically with the purpose of aggregating and processing these different sources. I use this for deciding whether it is safe to arm a drone, where I have a whole bunch of different conditions (implemented using bitflags) that must be cleared first. Then the task governing the motors just uses the following to wait for this all-clear. // Ensure all blockers are lowered before proceeding
rcv_arming_blocker.changed_and(|f| f.is_empty()).await; |
|
That's a good point, as the number of blocker/disarm sources rises, the odds of wanting some logic in there rises with it. As an aside, that snippet is from Holsatus, right? Very cool project, thanks for writing and sharing it. I would have thought the lack of an FPU would give the 2040 trouble as a flight controller. The 2350 should be an amazing target though. |
| fn try_changed(&self, id: &mut u64) -> Option<T> { | ||
| self.mutex.lock(|state| { | ||
| let s = state.borrow(); | ||
| match s.current_id > *id { | ||
| true => { | ||
| *id = s.current_id; | ||
| s.data.clone() | ||
| } | ||
| false => None, | ||
| } | ||
| }) | ||
| } | ||
|
|
||
| fn poll_changed_and(&self, id: &mut u64, f: &mut dyn Fn(&T) -> bool, cx: &mut Context<'_>) -> Poll<T> { | ||
| self.mutex.lock(|state| { | ||
| let mut s = state.borrow_mut(); | ||
| match (&s.data, s.current_id > *id) { | ||
| (Some(data), true) if f(data) => { | ||
| *id = s.current_id; | ||
| Poll::Ready(data.clone()) | ||
| } | ||
| _ => { | ||
| s.wakers.register(cx.waker()); | ||
| Poll::Pending | ||
| } | ||
| } | ||
| }) | ||
| } | ||
|
|
||
| fn try_changed_and(&self, id: &mut u64, f: &mut dyn Fn(&T) -> bool) -> Option<T> { | ||
| self.mutex.lock(|state| { | ||
| let s = state.borrow(); | ||
| match (&s.data, s.current_id > *id) { | ||
| (Some(data), true) if f(data) => { | ||
| *id = s.current_id; | ||
| s.data.clone() | ||
| } | ||
| _ => None, | ||
| } | ||
| }) | ||
| } |
There was a problem hiding this comment.
There appears to be a lot of code duplication in these implementations. Could you combine several of these into a single internal one on the Watch type that these could use?
This applies to the different send_ and get_ variants as well.
There was a problem hiding this comment.
I can combine all the poll, try, changed, get, and into a single monolith:
fn receive_inner(
&self,
id: Option<&mut u64>,
f: Option<&mut dyn Fn(&T) -> bool>,
cx: Option<&mut Context<'_>>,
must_change: bool,
) -> Poll<T> {
self.mutex.lock(|state| {
let mut s = state.borrow_mut();
// If we require the value to have changed, but the state ID is not larger
// than the receiver ID, we register the waker and return pending.
if must_change && !id.as_ref().is_some_and(|id| s.current_id > **id) {
cx.map(|cx| s.wakers.register(cx.waker()));
return Poll::Pending;
}
// If the data does not exist, we register the waker and return pending.
let Some(data) = &s.data else {
cx.map(|cx| s.wakers.register(cx.waker()));
return Poll::Pending;
};
// If the predicate exists, but it does not match,
// we register the waker and return pending.
if f.map_or(false, |f| !f(data)) {
cx.map(|cx| s.wakers.register(cx.waker()));
return Poll::Pending;
}
// Update the receiver ID, if any, and return value
id.map(|id| *id = s.current_id);
return Poll::Ready(data.clone());
})
}I guess this would still get optimized well, since user-code can only access this functionality through functions where the Options have a known variant, like:
/// Waits for the `Watch` to change and returns the new value, marking it as seen.
pub async fn changed(&mut self) -> T {
poll_fn(|cx| self.watch.receive_inner(Some(&mut self.at_id), None, Some(cx), true)).await
}
/// Tries to get the new value of the watch without waiting, marking it as seen.
pub fn try_changed(&mut self) -> Option<T> {
match self.watch.receive_inner(Some(&mut self.at_id), None, None, true) {
Poll::Ready(data) => Some(data),
Poll::Pending => None,
}
}There was a problem hiding this comment.
Hmm, yeah. There was more nuance than I saw initially, maybe keep the original then.
There was a problem hiding this comment.
In my quick testing looking at some smaller but similar examples on godbolt, the monolith seems to optimize just fine. So it is mostly a matter of whether you prefer some code duplication, or the monolith. I personally find the duplicated code easier to reason through, and it does not make the code look too crazy elsewhere. I'll leave it up to you. I am fine with either
|
@lulf Do you have any further input on this? |
This PR adds something similar to tokio watch. It is a single-slot channel similar to
Signal, but it can have multiple receivers, all of which get to have a clone of the value. So also similar to aPubSubChannel, but with a single-slot queue. Primary methods of the receivers are:changed: Await changes to the value, marking it as seen.get: Await the value (does not have to change) while marking it as seenAll of the above also have
try_xxxmethods, as well asxxx_andmethods which allow for only receiving when the value matches a predicate function. The doc-test comment and regular tests contain examples.