mini_rx/
dag.rs

1use std::fmt::Debug;
2use crate::dag_uid::RxDAGUid;
3use crate::rx_impl::{RxDAGElem, RxImpl, Rx, RxEdgeImpl};
4use crate::rx_ref::{RxRef, Var, CRx};
5use crate::misc::frozen_vec::{FrozenVec, FrozenSlice};
6use crate::misc::assert_variance::assert_is_covariant;
7use crate::misc::slice_split3::SliceSplit3;
8
9/// Returns a slice of [RxDAG] you can read nodes from.
10///
11/// Note that [RxContext] and [MutRxContext] are neither subset nor superset of each other.
12/// You can't read snapshots without recomputing, and you can't write inputs.
13pub trait RxContext<'a, 'c: 'a> {
14    fn sub_dag(self) -> RxSubDAG<'a, 'c>;
15}
16
17/// Returns a slice of [RxDAG] you can write variables in.
18///
19/// Note that [RxContext] and [MutRxContext] are neither subset nor superset of each other.
20/// You can't read snapshots without recomputing, and you can't write inputs.
21pub trait MutRxContext<'a, 'c: 'a> {
22    fn sub_dag(self) -> RxSubDAG<'a, 'c>;
23}
24
25/// The centralized structure which contains all your interconnected reactive values.
26///
27/// This structure is a directed-acyclic-graph (DAG), hence why its called [RxDAG].
28/// We don't support computed values recursively depending on each other, which is why it's acyclic.
29/// At the root/top of the DAG are your variables ([Var]), which you can explicitly set.
30/// All other values are [CRx]s, which are computed.
31/// [CRx] incoming edges are inputs, which are [Var]s and other [CRx]s.
32/// [CRx] outgoing edges are outputs, which are more [CRx]s, but there are also "edges"
33/// which just go nowhere, those are side-effects.
34///
35/// ## Performance notes
36///
37/// Currently no nodes ([Var]s or [CRx]s) are deallocated until the entire DAG is deallocated,
38/// so if you keep creating and discarding nodes you will leak memory (TODO fix this?)
39///
40/// ## Implementation
41///
42/// Internally this is a vector of interspersed nodes and edges.
43/// The edges refer to other nodes relative to their own position.
44/// Later Rxs *must* depend on earlier Rxs.
45/// "edges" can have zero outputs, those are side-effects as mentioned above.
46///
47/// When the DAG recomputes, it simply iterates through each node and edge in order and calls [RxDAGElem::recompute].
48/// If the nodes were changed (directly or as edge output), they set their new value, and mark that they got recomputed.
49/// The edges will recompute and change their output nodes if any of their inputs got recomputed.
50///
51/// The DAG has interior mutability, in that it can add nodes without a mutable borrow.
52/// See [elsa](https://docs.rs/elsa/latest/elsa/) crate for why this is sound (though actually the soundness argument is contested).
53/// Internally we use a modified version of [elsa](https://docs.rs/elsa/latest/elsa/) and [stable-deref-trait](https://docs.rs/stable-deref-trait/latest/stable-deref-trait/).
54/// which lets us return lifetime-parameterized values instead of references among other things.
55///
56/// Setting [Var]s is also interior mutability, and OK because we don't use those values until [RxDAGElem::recompute].
57///
58/// The DAG and refs have an ID so that you can't use one ref on another DAG, however this is checked at runtime.
59/// The lifetimes are checked at compile-time though.
60#[derive(Debug)]
61pub struct RxDAG<'c>(FrozenVec<RxDAGElem<'c>>, RxDAGUid<'c>);
62
63/// Allows you to read from an [RxDAG].
64#[derive(Debug, Clone, Copy)]
65pub struct RxDAGSnapshot<'a, 'c: 'a>(&'a RxDAG<'c>);
66
67/// Slice of an [RxDAG]
68#[doc(hidden)]
69#[derive(Debug, Clone, Copy)]
70pub struct RxSubDAG<'a, 'c: 'a> {
71    pub(crate) before: FrozenSlice<'a, RxDAGElem<'c>>,
72    pub(crate) index: usize,
73    pub(crate) id: RxDAGUid<'c>
74}
75assert_is_covariant!(for['a] (RxSubDAG<'a, 'c>) over 'c);
76
77/// Allows you to read from a slice of an [RxDAG].
78#[derive(Debug, Clone, Copy)]
79pub struct RxInput<'a, 'c: 'a>(pub(crate) RxSubDAG<'a, 'c>);
80
81impl<'c> RxDAG<'c> {
82    /// Create an empty DAG
83    pub fn new() -> Self {
84        Self(FrozenVec::new(), RxDAGUid::next())
85    }
86
87    /// Create a variable ([Var]) in this DAG.
88    pub fn new_var<T: 'c>(&self, init: T) -> Var<'c, T> {
89        let index = self.next_index();
90        let rx = RxImpl::new(init);
91        self.0.push(RxDAGElem::Node(Box::new(rx)));
92        Var::new(RxRef::new(self, index))
93    }
94
95    // region new_crx boilerplate
96
97    /// Run a closure when inputs change, without creating any outputs (for side-effects).
98    pub fn run_crx<F: FnMut(RxInput<'_, 'c>) + 'c>(&self, mut compute: F) {
99        let mut input_backwards_offsets = Vec::new();
100        let () = Self::run_compute(&mut compute, RxInput(self.sub_dag()), &mut input_backwards_offsets);
101        let compute_edge = RxEdgeImpl::<'c, _>::new(input_backwards_offsets, 0, move |mut input_backwards_offsets: &mut Vec<usize>, input: RxInput<'_, 'c>, outputs: &mut dyn Iterator<Item=&Rx<'c>>| {
102            input_backwards_offsets.clear();
103            let () = Self::run_compute(&mut compute, input, &mut input_backwards_offsets);
104            debug_assert!(outputs.next().is_none());
105        });
106        self.0.push(RxDAGElem::Edge(Box::new(compute_edge)));
107    }
108
109    /// Create a computed value ([CRx]) in this DAG.
110    pub fn new_crx<T: 'c, F: FnMut(RxInput<'_, 'c>) -> T + 'c>(&self, mut compute: F) -> CRx<'c, T> {
111        let mut input_backwards_offsets = Vec::new();
112        let init = Self::run_compute(&mut compute, RxInput(self.sub_dag()), &mut input_backwards_offsets);
113        let compute_edge = RxEdgeImpl::<'c, _>::new(input_backwards_offsets, 1, move |mut input_backwards_offsets: &mut Vec<usize>, input: RxInput<'_, 'c>, outputs: &mut dyn Iterator<Item=&Rx<'c>>| {
114            input_backwards_offsets.clear();
115            let output = Self::run_compute(&mut compute, input, &mut input_backwards_offsets);
116            unsafe { outputs.next().unwrap().set_dyn(output); }
117            debug_assert!(outputs.next().is_none());
118        });
119        self.0.push(RxDAGElem::Edge(Box::new(compute_edge)));
120
121        let index = self.next_index();
122        let rx = RxImpl::new(init);
123        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx)));
124        CRx::new(RxRef::new(self, index))
125    }
126
127    /// Create 2 computed values ([CRx]s) in this DAG which are created from the same function.
128    pub fn new_crx2<T1: 'c, T2: 'c, F: FnMut(RxInput<'_, 'c>) -> (T1, T2) + 'c>(&self, mut compute: F) -> (CRx<'c, T1>, CRx<'c, T2>) {
129        let mut input_backwards_offsets = Vec::new();
130        let (init1, init2) = Self::run_compute(&mut compute, RxInput(self.sub_dag()), &mut input_backwards_offsets);
131        let compute_edge = RxEdgeImpl::<'c, _>::new(input_backwards_offsets, 2, move |mut input_backwards_offsets: &mut Vec<usize>, input: RxInput<'_, 'c>, outputs: &mut dyn Iterator<Item=&Rx<'c>>| {
132            input_backwards_offsets.clear();
133            let (output1, output2) = Self::run_compute(&mut compute, input, &mut input_backwards_offsets);
134            unsafe { outputs.next().unwrap().set_dyn(output1); }
135            unsafe { outputs.next().unwrap().set_dyn(output2); }
136            debug_assert!(outputs.next().is_none());
137        });
138        self.0.push(RxDAGElem::Edge(Box::new(compute_edge)));
139
140        let index = self.next_index();
141        let rx1 = RxImpl::new(init1);
142        let rx2 = RxImpl::new(init2);
143        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx1)));
144        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx2)));
145        (CRx::new(RxRef::new(self, index)), CRx::new(RxRef::new(self, index + 1)))
146    }
147
148    /// Create 3 computed values ([CRx]s) in this DAG which are created from the same function.
149    pub fn new_crx3<T1: 'c, T2: 'c, T3: 'c, F: FnMut(RxInput<'_, 'c>) -> (T1, T2, T3) + 'c>(&self, mut compute: F) -> (CRx<'c, T1>, CRx<'c, T2>, CRx<'c, T3>) {
150        let mut input_backwards_offsets = Vec::new();
151        let (init1, init2, init3) = Self::run_compute(&mut compute, RxInput(self.sub_dag()), &mut input_backwards_offsets);
152        let compute_edge = RxEdgeImpl::<'c, _>::new(input_backwards_offsets, 3, move |mut input_backwards_offsets: &mut Vec<usize>, input: RxInput<'_, 'c>, outputs: &mut dyn Iterator<Item=&Rx<'c>>| {
153            input_backwards_offsets.clear();
154            let (output1, output2, output3) = Self::run_compute(&mut compute, input, &mut input_backwards_offsets);
155            unsafe { outputs.next().unwrap().set_dyn(output1); }
156            unsafe { outputs.next().unwrap().set_dyn(output2); }
157            unsafe { outputs.next().unwrap().set_dyn(output3); }
158            debug_assert!(outputs.next().is_none());
159        });
160        self.0.push(RxDAGElem::Edge(Box::new(compute_edge)));
161
162        let index = self.next_index();
163        let rx1 = RxImpl::new(init1);
164        let rx2 = RxImpl::new(init2);
165        let rx3 = RxImpl::new(init3);
166        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx1)));
167        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx2)));
168        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx3)));
169        (CRx::new(RxRef::new(self, index)), CRx::new(RxRef::new(self, index + 1)), CRx::new(RxRef::new(self, index + 2)))
170    }
171
172    /// Create 4 computed values ([CRx]s) in this DAG which are created from the same function.
173    pub fn new_crx4<T1: 'c, T2: 'c, T3: 'c, T4: 'c, F: FnMut(RxInput<'_, 'c>) -> (T1, T2, T3, T4) + 'c>(&self, mut compute: F) -> (CRx<'c, T1>, CRx<'c, T2>, CRx<'c, T3>, CRx<'c, T4>) {
174        let mut input_backwards_offsets = Vec::new();
175        let (init1, init2, init3, init4) = Self::run_compute(&mut compute, RxInput(self.sub_dag()), &mut input_backwards_offsets);
176        let compute_edge = RxEdgeImpl::<'c, _>::new(input_backwards_offsets, 4, move |mut input_backwards_offsets: &mut Vec<usize>, input: RxInput<'_, 'c>, outputs: &mut dyn Iterator<Item=&Rx<'c>>| {
177            input_backwards_offsets.clear();
178            let (output1, output2, output3, output4) = Self::run_compute(&mut compute, input, &mut input_backwards_offsets);
179            unsafe { outputs.next().unwrap().set_dyn(output1); }
180            unsafe { outputs.next().unwrap().set_dyn(output2); }
181            unsafe { outputs.next().unwrap().set_dyn(output3); }
182            unsafe { outputs.next().unwrap().set_dyn(output4); }
183            debug_assert!(outputs.next().is_none());
184        });
185        self.0.push(RxDAGElem::Edge(Box::new(compute_edge)));
186
187        let index = self.next_index();
188        let rx1 = RxImpl::new(init1);
189        let rx2 = RxImpl::new(init2);
190        let rx3 = RxImpl::new(init3);
191        let rx4 = RxImpl::new(init4);
192        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx1)));
193        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx2)));
194        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx3)));
195        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx4)));
196        (CRx::new(RxRef::new(self, index)), CRx::new(RxRef::new(self, index + 1)), CRx::new(RxRef::new(self, index + 2)), CRx::new(RxRef::new(self, index + 3)))
197    }
198
199    /// Create 5 computed values ([CRx]s) in this DAG which are created from the same function.
200    pub fn new_crx5<T1: 'c, T2: 'c, T3: 'c, T4: 'c, T5: 'c, F: FnMut(RxInput<'_, 'c>) -> (T1, T2, T3, T4, T5) + 'c>(&self, mut compute: F) -> (CRx<'c, T1>, CRx<'c, T2>, CRx<'c, T3>, CRx<'c, T4>, CRx<'c, T5>) {
201        let mut input_backwards_offsets = Vec::new();
202        let (init1, init2, init3, init4, init5) = Self::run_compute(&mut compute, RxInput(self.sub_dag()), &mut input_backwards_offsets);
203        let compute_edge = RxEdgeImpl::<'c, _>::new(input_backwards_offsets, 5, move |mut input_backwards_offsets: &mut Vec<usize>, input: RxInput<'_, 'c>, outputs: &mut dyn Iterator<Item=&Rx<'c>>| {
204            input_backwards_offsets.clear();
205            let (output1, output2, output3, output4, output5) = Self::run_compute(&mut compute, input, &mut input_backwards_offsets);
206            unsafe { outputs.next().unwrap().set_dyn(output1); }
207            unsafe { outputs.next().unwrap().set_dyn(output2); }
208            unsafe { outputs.next().unwrap().set_dyn(output3); }
209            unsafe { outputs.next().unwrap().set_dyn(output4); }
210            unsafe { outputs.next().unwrap().set_dyn(output5); }
211            debug_assert!(outputs.next().is_none());
212        });
213        self.0.push(RxDAGElem::Edge(Box::new(compute_edge)));
214
215        let index = self.next_index();
216        let rx1 = RxImpl::new(init1);
217        let rx2 = RxImpl::new(init2);
218        let rx3 = RxImpl::new(init3);
219        let rx4 = RxImpl::new(init4);
220        let rx5 = RxImpl::new(init5);
221        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx1)));
222        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx2)));
223        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx3)));
224        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx4)));
225        self.0.push(RxDAGElem::<'c>::Node(Box::new(rx5)));
226        (CRx::new(RxRef::new(self, index)), CRx::new(RxRef::new(self, index + 1)), CRx::new(RxRef::new(self, index + 2)), CRx::new(RxRef::new(self, index + 3)), CRx::new(RxRef::new(self, index + 4)))
227    }
228
229    // endregion
230
231    fn next_index(&self) -> usize {
232        self.0.len()
233    }
234
235    fn run_compute<T, F: FnMut(RxInput<'_, 'c>) -> T + 'c>(compute: &mut F, input: RxInput<'_, 'c>, input_backwards_offsets: &mut Vec<usize>) -> T {
236        debug_assert!(input_backwards_offsets.is_empty());
237
238        let result = compute(input);
239        let input_indices = input.post_read();
240
241        input_indices
242            .into_iter()
243            .map(|index| input.0.index - index)
244            .collect_into(input_backwards_offsets);
245        result
246    }
247
248    /// Update all [Var]s with their new values and recompute [CRx]s.
249    ///
250    /// This requires a shared reference and actually does the "reactive updates".
251    pub fn recompute(&mut self) {
252        for (index, (before, current, after)) in self.0.as_mut().iter_mut_split3s().enumerate() {
253            current.recompute(index, before, after, self.1);
254        }
255
256        for current in self.0.as_mut().iter_mut() {
257            current.post_recompute();
258        }
259    }
260
261    /// Recomputes if necessary and then returns an [RxContext] you can use to get the current value.
262    pub fn now(&mut self) -> RxDAGSnapshot<'_, 'c> {
263        self.recompute();
264        RxDAGSnapshot(self)
265    }
266
267    /// Returns an [RxContext] you can use to get the current value.
268    /// However any newly-set values or computations will not be returned until [RxDAG::recompute] is called.
269    pub fn stale(&self) -> RxDAGSnapshot<'_, 'c> {
270        RxDAGSnapshot(self)
271    }
272
273    pub(crate) fn id(&self) -> RxDAGUid<'c> {
274        self.1
275    }
276}
277
278impl<'a, 'c: 'a> RxContext<'a, 'c> for RxDAGSnapshot<'a, 'c> {
279    fn sub_dag(self) -> RxSubDAG<'a, 'c> {
280        RxSubDAG {
281            before: FrozenSlice::from(&self.0.0),
282            index: self.0.0.len(),
283            id: self.0.1
284        }
285    }
286}
287
288impl<'a, 'c: 'a> MutRxContext<'a, 'c> for &'a RxDAG<'c> {
289    fn sub_dag(self) -> RxSubDAG<'a, 'c> {
290        RxDAGSnapshot(self).sub_dag()
291    }
292}
293
294impl<'a, 'c: 'a> RxContext<'a, 'c> for RxInput<'a, 'c> {
295    fn sub_dag(self) -> RxSubDAG<'a, 'c> {
296        self.0
297    }
298}
299
300impl<'a, 'c: 'a> RxInput<'a, 'c> {
301    fn post_read(&self) -> Vec<usize> {
302        let mut results = Vec::new();
303        for (index, current) in self.0.before.iter().enumerate() {
304            if current.post_read() {
305                results.push(index)
306            }
307        }
308        results
309    }
310}
mini_rx/dag.rs

mini_rx/
dag.rs
