reactor_rt/
actions.rs

1/*
2 * Copyright (c) 2021, TU Dresden.
3 *
4 * Redistribution and use in source and binary forms, with or without modification,
5 * are permitted provided that the following conditions are met:
6 *
7 * 1. Redistributions of source code must retain the above copyright notice,
8 *    this list of conditions and the following disclaimer.
9 *
10 * 2. Redistributions in binary form must reproduce the above copyright notice,
11 *    this list of conditions and the following disclaimer in the documentation
12 *    and/or other materials provided with the distribution.
13 *
14 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
15 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
16 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
17 * THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
18 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
19 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
20 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
21 * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
22 * THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
23 */
24
25use std::cmp::Reverse;
26use std::marker::PhantomData;
27use std::ops::{Deref, DerefMut};
28use std::sync::{Arc, Mutex};
29use std::time::{Duration, Instant};
30
31use crate::assembly::{TriggerId, TriggerLike};
32use crate::*;
33
34use vecmap::{Entry, VecMap};
35
36/// A logical action.
37pub struct LogicalAction<T: Sync>(pub(crate) Action<Logical, T>);
38
39/// A physical action. Physical actions may only be used with
40/// the API of [AsyncCtx](crate::AsyncCtx).
41/// See [ReactionCtx::spawn_physical_thread](crate::ReactionCtx::spawn_physical_thread).
42pub struct PhysicalAction<T: Sync>(pub(crate) Action<Physical, T>);
43
44pub(crate) struct Logical;
45pub(crate) struct Physical;
46
47pub(crate) struct Action<Kind, T: Sync> {
48    pub(crate) min_delay: Duration,
49    id: TriggerId,
50    // is_logical: bool,
51    _logical: PhantomData<Kind>,
52
53    /// Stores values of an action for future scheduled events.
54    /// We rely strongly on the fact that any value put in there by [Action.schedule_future_value]
55    /// will be cleaned up after that tag. Otherwise the map will
56    /// blow up the heap.
57    map: VecMap<Reverse<EventTag>, Option<T>>,
58}
59
60impl<K, T: Sync> Action<K, T> {
61    /// Record a future value that can be queried at a future logical time.
62    /// Note that we don't check that the given time is in the future. If it's
63    /// in the past, the value will never be reclaimed.
64    ///
65    ///
66    #[inline]
67    pub(crate) fn schedule_future_value(&mut self, time: EventTag, value: Option<T>) {
68        match self.map.entry(Reverse(time)) {
69            Entry::Vacant(e) => e.insert(value),
70            Entry::Occupied(ref mut e) => {
71                trace!("Value overwritten in an action for tag {}", time);
72                trace!("This means an action was scheduled several times for the same tag.");
73                e.replace(value)
74            }
75        }
76    }
77
78    #[inline]
79    pub(crate) fn forget_value(&mut self, time: &EventTag) -> Option<T> {
80        self.map.remove(&Reverse(*time)).flatten()
81    }
82
83    fn new_impl(id: TriggerId, min_delay: Option<Duration>, _is_logical: bool) -> Self {
84        Action {
85            min_delay: min_delay.unwrap_or(Duration::ZERO),
86            // is_logical,
87            id,
88            _logical: PhantomData,
89            map: VecMap::new(),
90        }
91    }
92}
93
94impl<T: Sync, K> ReactionTrigger<T> for Action<K, T> {
95    #[inline]
96    fn is_present(&self, now: &EventTag, _start: &Instant) -> bool {
97        self.map.contains_key(&Reverse(*now))
98    }
99
100    #[inline]
101    fn get_value(&self, now: &EventTag, _start: &Instant) -> Option<T>
102    where
103        T: Copy,
104    {
105        self.map.get(&Reverse(*now)).cloned().flatten()
106    }
107
108    #[inline]
109    fn use_value_ref<O>(&self, now: &EventTag, _start: &Instant, action: impl FnOnce(Option<&T>) -> O) -> O {
110        let inmap: Option<&Option<T>> = self.map.get(&Reverse(*now));
111        let v = inmap.and_then(|i| i.as_ref());
112        action(v)
113    }
114}
115
116#[cfg(not(feature = "no-unsafe"))]
117impl<T: Sync, K> triggers::ReactionTriggerWithRefAccess<T> for Action<K, T> {
118    fn get_value_ref(&self, now: &EventTag, _start: &Instant) -> Option<&T> {
119        self.map.get(&Reverse(*now)).and_then(|a| a.as_ref())
120    }
121}
122
123impl<T: Sync> ReactionTrigger<T> for LogicalAction<T> {
124    #[inline]
125    fn is_present(&self, now: &EventTag, start: &Instant) -> bool {
126        self.0.is_present(now, start)
127    }
128
129    #[inline]
130    fn get_value(&self, now: &EventTag, start: &Instant) -> Option<T>
131    where
132        T: Copy,
133    {
134        self.0.get_value(now, start)
135    }
136
137    #[inline]
138    fn use_value_ref<O>(&self, now: &EventTag, start: &Instant, action: impl FnOnce(Option<&T>) -> O) -> O {
139        self.0.use_value_ref(now, start, action)
140    }
141}
142
143#[cfg(not(feature = "no-unsafe"))]
144impl<T: Sync> triggers::ReactionTriggerWithRefAccess<T> for LogicalAction<T> {
145    fn get_value_ref(&self, now: &EventTag, start: &Instant) -> Option<&T> {
146        self.0.get_value_ref(now, start)
147    }
148}
149
150impl<T: Sync> LogicalAction<T> {
151    pub(crate) fn new(id: TriggerId, min_delay: Option<Duration>) -> Self {
152        Self(Action::new_impl(id, min_delay, true))
153    }
154}
155
156impl<T: Sync> PhysicalAction<T> {
157    fn new(id: TriggerId, min_delay: Option<Duration>) -> Self {
158        Self(Action::new_impl(id, min_delay, false))
159    }
160}
161
162impl<T: Sync> TriggerLike for PhysicalAction<T> {
163    fn get_id(&self) -> TriggerId {
164        self.0.id
165    }
166}
167
168impl<T: Sync> TriggerLike for LogicalAction<T> {
169    fn get_id(&self) -> TriggerId {
170        self.0.id
171    }
172}
173/*#[cfg(test)] //fixme
174mod test {
175    use ActionPresence::{NotPresent, Present};
176
177    use crate::*;
178
179    #[test]
180    fn a_value_map_should_be_able_to_store_a_value() {
181        let mut vmap = Action::<i64>::default();
182        let fut = LogicalInstant::now() + Duration::from_millis(500);
183        assert_eq!(NotPresent, vmap.get_value(fut));
184        vmap.schedule(fut, Some(2555));
185        assert_eq!(Present(Some(2555)), vmap.get_value(fut));
186        assert_eq!(Present(Some(2555)), vmap.get_value(fut)); // not deleted
187        vmap.schedule(fut, Some(16));
188        assert_eq!(Present(Some(16)), vmap.get_value(fut));
189        vmap.schedule(fut, None);
190        assert_eq!(Present(None), vmap.get_value(fut));
191        vmap.forget(fut);
192        assert_eq!(NotPresent, vmap.get_value(fut));
193    }
194
195    #[test]
196    fn a_value_map_should_be_able_to_forget_a_value() {
197        let mut vmap = ValueMap::<i64>::default();
198        let fut = LogicalInstant::now() + Duration::from_millis(500);
199        vmap.schedule(fut, Some(2555));
200        assert_eq!(Present(Some(2555)), vmap.get_value(fut));
201        vmap.forget(fut);
202        assert_eq!(NotPresent, vmap.get_value(fut));
203    }
204
205    #[test]
206    fn a_value_map_should_be_able_to_store_more_values() {
207        let mut vmap = ValueMap::<i64>::default();
208        let fut = LogicalInstant::now() + Duration::from_millis(500);
209        let fut2 = LogicalInstant::now() + Duration::from_millis(540);
210        let fut3 = LogicalInstant::now() + Duration::from_millis(560);
211
212        vmap.schedule(fut, Some(1));
213        // order is reversed on purpose
214        vmap.schedule(fut3, Some(3));
215        vmap.schedule(fut2, Some(2));
216
217        assert_eq!(Present(Some(1)), vmap.get_value(fut));
218        assert_eq!(Present(Some(2)), vmap.get_value(fut2));
219        assert_eq!(Present(Some(3)), vmap.get_value(fut3));
220    }
221}
222*/
223
224/// A reference to a physical action. This thing is cloneable
225/// and can be sent to async threads. The contained action
226/// reference is unique and protected by a lock. All operations
227/// on the action are
228///
229/// See [crate::ReactionCtx::spawn_physical_thread].
230#[derive(Clone)]
231pub struct PhysicalActionRef<T: Sync>(Arc<Mutex<PhysicalAction<T>>>);
232
233impl<T: Sync> PhysicalActionRef<T> {
234    pub(crate) fn new(id: TriggerId, min_delay: Option<Duration>) -> Self {
235        Self(Arc::new(Mutex::new(PhysicalAction::new(id, min_delay))))
236    }
237
238    pub(crate) fn use_mut<O>(&self, f: impl FnOnce(&mut PhysicalAction<T>) -> O) -> Result<O, ()> {
239        let mut refmut = self.0.deref().lock().map_err(|_| ())?;
240
241        Ok(f(refmut.deref_mut()))
242    }
243
244    pub(crate) fn use_mut_p<O, P>(&self, p: P, f: impl FnOnce(&mut PhysicalAction<T>, P) -> O) -> Result<O, P> {
245        match self.0.deref().lock() {
246            Ok(mut refmut) => Ok(f(refmut.deref_mut(), p)),
247            Err(_) => Err(p),
248        }
249    }
250
251    pub(crate) fn use_value<O>(&self, f: impl FnOnce(&PhysicalAction<T>) -> O) -> Result<O, ()> {
252        let r#ref = self.0.deref().lock().map_err(|_| ())?;
253
254        Ok(f(r#ref.deref()))
255    }
256}
257
258impl<T: Sync> TriggerLike for PhysicalActionRef<T> {
259    fn get_id(&self) -> TriggerId {
260        self.use_value(|a| a.get_id()).unwrap()
261    }
262}
263
264impl<T: Sync> ReactionTrigger<T> for PhysicalActionRef<T> {
265    fn is_present(&self, now: &EventTag, start: &Instant) -> bool {
266        self.use_value(|a| a.0.is_present(now, start)).unwrap()
267    }
268
269    fn get_value(&self, now: &EventTag, start: &Instant) -> Option<T>
270    where
271        T: Copy,
272    {
273        self.use_value(|a| a.0.get_value(now, start)).unwrap()
274    }
275
276    fn use_value_ref<O>(&self, now: &EventTag, start: &Instant, action: impl FnOnce(Option<&T>) -> O) -> O {
277        self.use_value(|a| a.0.use_value_ref(now, start, action)).unwrap()
278    }
279}
reactor_rt/actions.rs

reactor_rt/
actions.rs
