veecle_os_runtime/datastore/
initialized_reader.rs

1use core::cell::Ref;
2use core::marker::PhantomData;
3
4use crate::datastore::{Storable, slot};
5
6/// Reader for a [`Storable`] type.
7///
8/// Allows [`Actor`]s to read a value of a type written by another actor.
9/// The generic type `T` from the reader specifies the type of the value that is being read.
10///
11/// This reader can be requested directly as an actor input in simple cases, this will mean your actor does not start
12/// running until all `InitializedReader`s it takes have been initialized by their writers.
13/// If you need to do something more complex (e.g. you have interdependencies between actors so one must write an
14/// initial value earlier) then you can take a `Reader` and convert via [`Reader::wait_init`][super::Reader::wait_init]
15/// when ready.
16/// By ensuring the presence of a value for `T` has been written at least once, this reader avoids `Option` when
17/// reading.
18///
19/// # Example
20///
21/// ```rust
22/// # use std::fmt::Debug;
23/// #
24/// # use veecle_os_runtime::{Storable, Reader, InitializedReader};
25/// #
26/// # #[derive(Debug, Default, Storable)]
27/// # pub struct Foo;
28/// #
29/// #[veecle_os_runtime::actor]
30/// async fn foo_reader(mut reader: InitializedReader<'_, Foo>) -> veecle_os_runtime::Never {
31///     loop {
32///         let processed_value = reader.wait_for_update().await.read(|value: &Foo| {
33///             // Do something with the value.
34///         });
35///     }
36/// }
37///
38/// #[veecle_os_runtime::actor]
39/// async fn foo_reader_complex(mut reader: Reader<'_, Foo>) -> veecle_os_runtime::Never {
40///     // Do some initialization that must be completed before waiting for the reader to have an initial value.
41///     let mut reader = reader.wait_init().await;
42///     loop {
43///         let processed_value = reader.wait_for_update().await.read(|value: &Foo| {
44///             // Do something with the value.
45///         });
46///     }
47/// }
48/// ```
49///
50/// [`Actor`]: crate::actor::Actor
51#[derive(Debug)]
52pub struct InitializedReader<'a, T>
53where
54    T: Storable + 'static,
55{
56    waiter: slot::Waiter<'a, T>,
57
58    marker: PhantomData<fn(T)>,
59}
60
61impl<T> InitializedReader<'_, T>
62where
63    T: Storable + 'static,
64{
65    /// Reads the current value of a type.
66    ///
67    /// Can be combined with [`Self::wait_for_update`] to wait for the value to be updated before reading it.
68    ///
69    /// This method takes a closure to ensure the reference is not held across await points.
70    #[veecle_telemetry::instrument]
71    pub fn read<U>(&self, f: impl FnOnce(&T::DataType) -> U) -> U {
72        self.waiter.read(|value| {
73            let value = value
74                .as_ref()
75                .expect("initialized reader should only access initialized values");
76
77            veecle_telemetry::trace!("Slot read", value = format_args!("{value:?}"));
78            f(value)
79        })
80    }
81
82    /// Reads and clones the current value of a type.
83    ///
84    /// This is a wrapper around [`Self::read`] that additionally clones the value.
85    /// You can use it instead of `reader.read(|c| c.clone())`.
86    pub fn read_cloned(&self) -> T::DataType
87    where
88        T::DataType: Clone,
89    {
90        self.read(|t| t.clone())
91    }
92
93    /// Waits for any write to occur.
94    ///
95    /// This future resolving does not imply that `previous_value != new_value`, just that a
96    /// [`Writer`][super::Writer] has written a value of `T` since the last time this future resolved.
97    ///
98    /// This returns `&mut Self` to allow chaining a call to methods accessing the value, for example
99    /// [`read`][Self::read`].
100    #[veecle_telemetry::instrument]
101    pub async fn wait_for_update(&mut self) -> &mut Self {
102        self.waiter.wait().await;
103        self.waiter.update_generation();
104        self
105    }
106}
107
108impl<'a, T> InitializedReader<'a, T>
109where
110    T: Storable + 'static,
111{
112    /// Creates a new `InitializedReader` from a [`Waiter`][slot::Waiter].
113    pub(crate) fn new(waiter: slot::Waiter<'a, T>) -> Self {
114        Self {
115            waiter,
116            marker: Default::default(),
117        }
118    }
119}
120
121impl<T> super::combined_readers::Sealed for InitializedReader<'_, T> where T: Storable {}
122
123impl<T> super::combined_readers::CombinableReader for InitializedReader<'_, T>
124where
125    T: Storable,
126{
127    type ToBeRead = T::DataType;
128
129    fn borrow(&self) -> Ref<'_, Self::ToBeRead> {
130        Ref::map(self.waiter.borrow(), |t| t.as_ref().unwrap())
131    }
132
133    async fn wait_for_update(&mut self) {
134        self.wait_for_update().await;
135    }
136}
137
138#[cfg(test)]
139#[cfg_attr(coverage_nightly, coverage(off))]
140mod tests {
141    use core::pin::pin;
142    use futures::FutureExt;
143
144    use crate::datastore::{Reader, Slot, Storable, Writer, generational};
145
146    #[test]
147    fn read() {
148        #[derive(Eq, PartialEq, Debug, Clone, Storable)]
149        #[storable(crate = crate)]
150        struct Sensor(u8);
151
152        let source = pin!(generational::Source::new());
153        let slot = pin!(Slot::<Sensor>::new());
154
155        let mut writer = Writer::new(source.as_ref().waiter(), slot.as_ref());
156        let reader = Reader::from_slot(slot.as_ref());
157
158        assert!(reader.wait_init().now_or_never().is_none());
159
160        source.as_ref().increment_generation();
161        writer.write(Sensor(5)).now_or_never().unwrap();
162
163        let reader = Reader::from_slot(slot.as_ref())
164            .wait_init()
165            .now_or_never()
166            .unwrap();
167
168        assert_eq!(reader.read(|x: &Sensor| x.clone()), Sensor(5));
169        assert_eq!(reader.read_cloned(), Sensor(5));
170    }
171
172    #[test]
173    fn wait_for_update() {
174        #[derive(Eq, PartialEq, Debug, Clone, Storable)]
175        #[storable(crate = crate)]
176        struct Sensor(u8);
177
178        let source = pin!(generational::Source::new());
179        let slot = pin!(Slot::<Sensor>::new());
180
181        let mut writer = Writer::new(source.as_ref().waiter(), slot.as_ref());
182        let reader = Reader::from_slot(slot.as_ref());
183
184        source.as_ref().increment_generation();
185        writer.write(Sensor(1)).now_or_never().unwrap();
186
187        let mut reader = reader.wait_init().now_or_never().unwrap();
188
189        assert!(reader.wait_for_update().now_or_never().is_some());
190        assert!(reader.wait_for_update().now_or_never().is_none());
191
192        source.as_ref().increment_generation();
193        writer.write(Sensor(1)).now_or_never().unwrap();
194
195        reader
196            .wait_for_update()
197            .now_or_never()
198            .unwrap()
199            .read(|x| assert_eq!(x, &Sensor(1)));
200    }
201
202    #[test]
203    fn wait_init_wait_for_update() {
204        #[derive(Eq, PartialEq, Debug, Clone, Storable)]
205        #[storable(crate = crate)]
206        struct Sensor(u8);
207
208        let source = pin!(generational::Source::new());
209        let slot = pin!(Slot::<Sensor>::new());
210
211        let mut writer = Writer::new(source.as_ref().waiter(), slot.as_ref());
212        let reader = Reader::from_slot(slot.as_ref());
213
214        let mut wait_init_fut = pin!(reader.wait_init());
215        assert!(wait_init_fut.as_mut().now_or_never().is_none());
216        // Increment generation to allow the writer to write.
217        source.as_ref().increment_generation();
218        writer.write(Sensor(1)).now_or_never().unwrap();
219
220        let mut reader = wait_init_fut.now_or_never().unwrap();
221
222        // If `wait_init` does not increment the waiter generation, `now_or_never` must return `Some`.
223        reader
224            .wait_for_update()
225            .now_or_never()
226            .unwrap()
227            .read(|x| assert_eq!(x, &Sensor(1)));
228    }
229}
veecle_os_runtime/datastore/initialized_reader.rs

veecle_os_runtime/datastore/
initialized_reader.rs
