reactor_rt/
ports.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::borrow::Borrow;
26#[cfg(not(feature = "no-unsafe"))]
27use std::cell::UnsafeCell;
28use std::collections::HashMap;
29use std::fmt::Debug;
30#[cfg(feature = "no-unsafe")]
31use std::ops::Deref;
32use std::ops::{DerefMut, Index, IndexMut};
33use std::rc::Rc;
34use std::time::Instant;
35
36use atomic_refcell::AtomicRefCell;
37use AssemblyErrorImpl::{CannotBind, CyclicDependency};
38
39use crate::assembly::{AssemblyError, AssemblyErrorImpl, PortId, PortKind, TriggerId, TriggerLike};
40use crate::{EventTag, ReactionTrigger};
41
42/// Represents a port, which carries values of type `T`.
43/// Ports reify the data inputs and outputs of a reactor.
44///
45/// They may be bound to another port, in which case the
46/// upstream port forwards all values to the output port
47/// (logically instantaneously). A port may have only one
48/// upstream binding.
49///
50/// Output ports may also be explicitly [set](super::ReactionCtx::set)
51/// within a reaction, in which case they may not have an
52/// upstream port binding.
53///
54/// Those structural constraints are trusted to have been
55/// verified by the code generator. If necessary we may be
56/// able to add conditional compilation flags that enable
57/// runtime checks.
58///
59pub struct Port<T: Sync> {
60    id: TriggerId,
61    kind: PortKind,
62    bind_status: BindStatus,
63    upstream_binding: Rc<UncheckedCell<Rc<PortCell<T>>>>,
64    //                                 ^^
65    // Note that manipulating this Rc is really unsafe and
66    // requires care to avoid UB.
67    // - Cloning the Rc from different threads concurrently is UB.
68    // But in this framework, that Rc is cloned only during the
69    // assembly phase, which occurs entirely in the main
70    // scheduler thread.
71    // - Modifying the contents concurrently is also UB.
72    // But, by construction of the dependency graph,
73    //   - there is at most one reaction that may set the port,
74    // so we never borrow the PortCell mutably twice concurrently.
75    //   - that reaction, if any, must be executed BEFORE any
76    // reaction that reads the port. That BEFORE is a
77    // synchronization barrier in the parallel runtime
78    // implementation, so there is no simultaneous mutable and immutable borrow.
79    //
80    //
81}
82
83impl<T: Sync> Port<T> {
84    /// Create a new port
85    pub(crate) fn new(id: TriggerId, kind: PortKind) -> Self {
86        Self {
87            id,
88            kind,
89            bind_status: BindStatus::Free,
90            #[cfg(feature = "no-unsafe")]
91            upstream_binding: Rc::new(AtomicRefCell::new(Default::default())),
92            #[cfg(not(feature = "no-unsafe"))]
93            upstream_binding: Rc::new(UnsafeCell::new(Default::default())),
94        }
95    }
96
97    pub(crate) fn is_present_now(&self) -> bool {
98        self.use_ref(|opt| opt.is_some())
99    }
100
101    pub(crate) fn get_kind(&self) -> PortKind {
102        self.kind
103    }
104    #[inline]
105    pub(crate) fn get(&self) -> Option<T>
106    where
107        T: Copy,
108    {
109        self.use_ref(Option::<T>::clone)
110    }
111
112    cfg_if! {
113        if #[cfg(feature = "no-unsafe")] {
114            pub(crate) fn use_ref<R>(&self, f: impl FnOnce(&Option<T>) -> R) -> R {
115                use atomic_refcell::AtomicRef;
116                let cell_ref: AtomicRef<Rc<PortCell<T>>> = AtomicRefCell::borrow(&self.upstream_binding);
117                let binding: &Rc<PortCell<T>> = cell_ref.deref();
118                let class_cell: &PortCell<T> = Rc::borrow(binding);
119                let cell_borrow: &AtomicRef<Option<T>> = &class_cell.value.borrow();
120
121                f(cell_borrow.deref())
122            }
123
124            /// Set the value, see [super::ReactionCtx::set]
125            /// Note: we use a closure to process the dependencies to
126            /// avoid having to clone the dependency list just to return it.
127            pub(crate) fn set_impl(&mut self, new_value: Option<T>) {
128                use atomic_refcell::AtomicRef;
129
130                debug_assert_ne!(self.bind_status, BindStatus::Bound, "Cannot set a bound port ({:?})", self.id);
131
132                let cell_ref: AtomicRef<Rc<PortCell<T>>> = AtomicRefCell::borrow(&self.upstream_binding);
133                let class_cell: &PortCell<T> = Rc::borrow(cell_ref.deref());
134
135                *class_cell.value.borrow_mut() = new_value;
136            }
137
138        } else {
139
140             /// Returns a reference to the value. It is not possible to
141             /// implement this in safe code without reimplementing the same
142             /// kind of logic as AtomicRef, because the ref has to hold
143             /// two borrows at the same time.
144             pub(crate) fn get_ref(&self) -> Option<&T> {
145                 let binding: &UnsafeCell<Rc<PortCell<T>>> = Rc::borrow(&self.upstream_binding);
146                 unsafe {
147                     let cell = &*binding.get();
148                     let opt = &*cell.value.get();
149                    opt.as_ref()
150                 }
151             }
152
153             #[inline]
154             pub(crate) fn use_ref<R>(&self, f: impl FnOnce(&Option<T>) -> R) -> R {
155                let binding: &UnsafeCell<Rc<PortCell<T>>> = Rc::borrow(&self.upstream_binding);
156                let opt: &Option<T> = unsafe {
157                    let cell = &*binding.get();
158                    &*cell.value.get()
159                };
160                f(opt)
161            }
162
163             #[inline]
164             pub(crate) fn set_impl(&mut self, new_value: Option<T>) {
165                debug_assert_ne!(self.bind_status, BindStatus::Bound, "Cannot set a bound port");
166
167                let binding: &UnsafeCell<Rc<PortCell<T>>> = Rc::borrow(&self.upstream_binding);
168
169                unsafe {
170                    let cell: &Rc<PortCell<T>> = &*binding.get();
171                    // note: using write instead of replace would not drop the old value
172                    cell.value.get().replace(new_value);
173                }
174            }
175        }
176    }
177
178    /// Called at the end of a tag.
179    #[inline]
180    pub(crate) fn clear_value(&mut self) {
181        // If this port is bound, then some other port has
182        // a reference to the same cell but is not bound.
183        if self.bind_status != BindStatus::Bound {
184            self.set_impl(None)
185        }
186    }
187
188    pub(crate) fn forward_to(&mut self, downstream: &mut Port<T>) -> Result<(), AssemblyError> {
189        let mut mut_downstream_cell = {
190            cfg_if! {
191                if #[cfg(feature = "no-unsafe")] {
192                    (downstream.upstream_binding).borrow_mut()
193                } else {
194                    unsafe { downstream.upstream_binding.get().as_mut().unwrap() }
195                }
196            }
197        };
198
199        if downstream.bind_status == BindStatus::Bound {
200            return Err(AssemblyError(CannotBind(self.id, downstream.id)));
201        }
202
203        downstream.bind_status = BindStatus::Bound;
204
205        let my_class = {
206            cfg_if! {
207                if #[cfg(feature = "no-unsafe")] {
208                    self.upstream_binding.borrow_mut()
209                } else {
210                    unsafe { self.upstream_binding.get().as_mut().unwrap() }
211                }
212            }
213        };
214
215        my_class
216            .downstreams
217            .borrow_mut()
218            .insert(downstream.id, Rc::clone(&downstream.upstream_binding));
219
220        let new_binding = Rc::clone(&my_class);
221
222        mut_downstream_cell.check_cycle(&self.id, &downstream.id)?;
223
224        mut_downstream_cell.set_upstream(&my_class);
225        *mut_downstream_cell.deref_mut() = new_binding;
226        Ok(())
227    }
228}
229
230impl<T: Sync> ReactionTrigger<T> for Port<T> {
231    #[inline]
232    fn is_present(&self, _now: &EventTag, _start: &Instant) -> bool {
233        self.is_present_now()
234    }
235
236    #[inline]
237    fn get_value(&self, _now: &EventTag, _start: &Instant) -> Option<T>
238    where
239        T: Copy,
240    {
241        self.get()
242    }
243
244    #[inline]
245    fn use_value_ref<O>(&self, _now: &EventTag, _start: &Instant, action: impl FnOnce(Option<&T>) -> O) -> O {
246        self.use_ref(|opt| action(opt.as_ref()))
247    }
248}
249
250#[cfg(not(feature = "no-unsafe"))]
251impl<T: Sync> crate::triggers::ReactionTriggerWithRefAccess<T> for Port<T> {
252    fn get_value_ref(&self, _now: &EventTag, _start: &Instant) -> Option<&T> {
253        self.get_ref()
254    }
255}
256
257impl<T: Sync> TriggerLike for Port<T> {
258    fn get_id(&self) -> TriggerId {
259        self.id
260    }
261}
262
263#[derive(Copy, Clone, Eq, PartialEq, Debug)]
264enum BindStatus {
265    /// A bindable port is also writable explicitly (with set)
266    Free,
267
268    // Means that this port is the upstream of some bound port.
269    // Upstream,
270    /// A bound port cannot be written to explicitly. For a
271    /// set of ports bound together, there is a single cell,
272    /// and a single writable port through which values are
273    /// communicated ([Self::Upstream]).
274    Bound,
275}
276
277cfg_if! {
278    if #[cfg(feature = "no-unsafe")] {
279        type Downstreams<T> = AtomicRefCell<HashMap<PortId, Rc<AtomicRefCell<Rc<PortCell<T>>>>>>;
280        type UncheckedCell<T> = AtomicRefCell<T>;
281    } else {
282        type Downstreams<T> = AtomicRefCell<HashMap<PortId, Rc<UnsafeCell<Rc<PortCell<T>>>>>>;
283        type UncheckedCell<T> = UnsafeCell<T>;
284    }
285}
286
287/// This is the internal cell type that is shared by ports.
288struct PortCell<T: Sync> {
289    /// Cell for the value.
290    value: UncheckedCell<Option<T>>,
291
292    /// This is the set of ports that are "forwarded to".
293    /// When you bind 2 ports A -> B, then the binding of B
294    /// is updated to point to the equiv class of A. The downstream
295    /// field of that equiv class is updated to contain B.
296    ///
297    /// Why?
298    /// When you have bound eg A -> B and *then* bind U -> A,
299    /// then both the equiv class of A and B (the downstream of A)
300    /// need to be updated to point to the equiv class of U
301    ///
302    /// Coincidentally, this means we can track transitive
303    /// cyclic port dependencies:
304    /// - say you have bound A -> B, then B -> C
305    /// - so all three refer to the equiv class of A, whose downstream is now {B, C}
306    /// - if you then try binding C -> A, then we can know
307    ///   that C is in the downstream of A, indicating that there is a cycle.
308    downstreams: Downstreams<T>,
309}
310
311impl<T: Sync> PortCell<T> {
312    fn check_cycle(&self, upstream_id: &PortId, downstream_id: &PortId) -> Result<(), AssemblyError> {
313        if (*self.downstreams.borrow()).contains_key(upstream_id) {
314            Err(AssemblyError(CyclicDependency(*upstream_id, *downstream_id)))
315        } else {
316            Ok(())
317        }
318    }
319
320    /// This updates all downstreams to point to the given equiv class instead of `self`
321    fn set_upstream(&self, new_binding: &Rc<PortCell<T>>) {
322        for cell_rc in (*self.downstreams.borrow()).values() {
323            cfg_if! {
324                if #[cfg(feature = "no-unsafe")] {
325                    let mut ref_mut = cell_rc.borrow_mut();
326                    *ref_mut.deref_mut() = Rc::clone(new_binding);
327                } else {
328                    unsafe {
329                        *cell_rc.get() = Rc::clone(new_binding);
330                    }
331                }
332            }
333        }
334    }
335}
336
337impl<T: Sync> Default for PortCell<T> {
338    fn default() -> Self {
339        PortCell {
340            value: Default::default(),
341            downstreams: Default::default(),
342        }
343    }
344}
345
346/// A multiport is a vector of independent ports (its _channels_)
347/// Multiports have special Lingua Franca syntax, similar to reactor banks.
348pub struct Multiport<T: Sync> {
349    ports: Vec<Port<T>>,
350    id: TriggerId,
351}
352
353impl<T: Sync> Multiport<T> {
354    /// Create a multiport from the given vector of ports.
355    #[inline(always)]
356    pub(crate) fn new(ports: Vec<Port<T>>, id: TriggerId) -> Self {
357        Self { ports, id }
358    }
359
360    /// Returns the number of channels.
361    #[inline(always)]
362    pub fn len(&self) -> usize {
363        self.ports.len()
364    }
365
366    /// Returns true if this multiport is empty.
367    #[inline(always)]
368    pub fn is_empty(&self) -> bool {
369        self.ports.is_empty()
370    }
371
372    /// Iterate over the multiport and return mutable references to individual channels.
373    #[inline(always)]
374    pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut Port<T>> {
375        self.ports.iter_mut()
376    }
377
378    /// Iterate over the channels of this multiport. Returns read-only
379    /// references to individual ports.
380    #[inline(always)]
381    pub fn iter(&self) -> impl Iterator<Item = &Port<T>> {
382        self.into_iter()
383    }
384
385    /// Iterate over only those channels that are set (have a value).
386    /// Returns a tuple with their index (not necessarily contiguous).
387    pub fn enumerate_set(&self) -> impl Iterator<Item = (usize, &Port<T>)> {
388        self.iter().enumerate().filter(|&(_, p)| p.is_present_now())
389    }
390
391    /// Iterate over only those channels that are set (have a value).
392    /// The returned ports are not necessarily contiguous. See
393    /// [Self::enumerate_set] to get access to their index.
394    pub fn iterate_set(&self) -> impl Iterator<Item = &Port<T>> {
395        self.iter().filter(|&p| p.is_present_now())
396    }
397
398    /// Iterate over only those channels that are set (have a value),
399    /// and return a copy of the value.
400    /// The returned ports are not necessarily contiguous. See
401    /// [Self::enumerate_values] to get access to their index.
402    pub fn iterate_values(&self) -> impl Iterator<Item = T> + '_
403    where
404        T: Copy,
405    {
406        self.iter().filter_map(|p| p.get())
407    }
408
409    /// Iterate over only those ports that are set (have a value),
410    /// and return a reference to the value.
411    /// The returned ports are not necessarily contiguous. See
412    /// [Self::enumerate_values] to get access to their index.
413    #[cfg(not(feature = "no-unsafe"))]
414    pub fn iterate_values_ref(&self) -> impl Iterator<Item = &T> + '_ {
415        self.iter().filter_map(|p| p.get_ref())
416    }
417
418    /// Iterate over only those channels that are set (have a value),
419    /// yielding a tuple with their index in the bank and a copy of the value.
420    pub fn enumerate_values(&self) -> impl Iterator<Item = (usize, T)> + '_
421    where
422        T: Copy,
423    {
424        self.iter().enumerate().filter_map(|(i, p)| p.get().map(|v| (i, v)))
425    }
426
427    /// Iterate over only those channels that are set (have a value),
428    /// yielding a tuple with their index in the bank and a reference to the value.
429    #[cfg(not(feature = "no-unsafe"))]
430    pub fn enumerate_values_ref(&self) -> impl Iterator<Item = (usize, &T)> + '_ {
431        self.iter().enumerate().filter_map(|(i, p)| p.get_ref().map(|v| (i, v)))
432    }
433}
434
435impl<T: Sync> TriggerLike for Multiport<T> {
436    fn get_id(&self) -> TriggerId {
437        self.id
438    }
439}
440
441impl<'a, T: Sync> IntoIterator for &'a mut Multiport<T> {
442    type Item = &'a mut Port<T>;
443    type IntoIter = std::slice::IterMut<'a, Port<T>>;
444
445    fn into_iter(self) -> Self::IntoIter {
446        self.ports.iter_mut()
447    }
448}
449
450impl<'a, T: Sync> IntoIterator for &'a Multiport<T> {
451    type Item = &'a Port<T>;
452    type IntoIter = std::slice::Iter<'a, Port<T>>;
453
454    fn into_iter(self) -> Self::IntoIter {
455        self.ports.iter()
456    }
457}
458
459impl<T: Sync> Index<usize> for Multiport<T> {
460    type Output = Port<T>;
461
462    fn index(&self, index: usize) -> &Self::Output {
463        &self.ports[index]
464    }
465}
466
467impl<T: Sync> IndexMut<usize> for Multiport<T> {
468    fn index_mut(&mut self, index: usize) -> &mut Self::Output {
469        &mut self.ports[index]
470    }
471}
reactor_rt/ports.rs

reactor_rt/
ports.rs
