catlog/stdlib/analyses/
reachability.rs

1//! Reachability analyses of models.
2
3use itertools::Itertools;
4use std::collections::HashMap;
5
6use crate::dbl::modal::model::{ModalDblModel, ModalOb};
7use crate::dbl::theory::Unital;
8use crate::one::category::FgCategory;
9use crate::stdlib::analyses::petri::transition_interface;
10use crate::zero::QualifiedName;
11
12#[cfg(feature = "serde")]
13use serde::{Deserialize, Serialize};
14#[cfg(feature = "serde-wasm")]
15use tsify::Tsify;
16
17/// Data defining a reachability problem for a Petri net.
18#[derive(Clone)]
19#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
20#[cfg_attr(feature = "serde-wasm", derive(Tsify))]
21#[cfg_attr(
22    feature = "serde-wasm",
23    tsify(into_wasm_abi, from_wasm_abi, hashmap_as_object)
24)]
25pub struct ReachabilityProblemData {
26    /// Map from place IDs to number of initial token of that type.
27    pub tokens: HashMap<QualifiedName, i32>,
28
29    /// Map from place IDs to number of forbidden tokens of that type.
30    pub forbidden: HashMap<QualifiedName, i32>,
31}
32
33/// The "Region Algebra for Petri Nets" algorithm from Ch 31 of
34/// ([Clarke et al 2018](crate::refs::HandbookModelChecking)):
35/// "Symbolic Model Checking in Non Boolean Domains".
36pub fn subreachability(m: &ModalDblModel<Unital>, data: ReachabilityProblemData) -> bool {
37    // Convert model into a pair of matrices
38    //--------------------------------------
39
40    // Get a canonical ordering of the objects
41    let ob_vec: Vec<_> = m.ob_generators().sorted().collect();
42    let ob_inv: HashMap<_, _> = ob_vec.iter().enumerate().map(|(x, y)| (y.clone(), x)).collect();
43    let n_p = ob_vec.len();
44    if n_p == 0 {
45        return true;
46    }
47
48    // Get a canonical ordering of the homs
49    let hom_vec: Vec<_> = m.mor_generators().sorted().collect();
50
51    let hom_inv: HashMap<_, _> = hom_vec.iter().enumerate().map(|(x, y)| (y.clone(), x)).collect();
52    let n_t = hom_vec.len();
53
54    // Populate the I/O matrices from the hom src/tgt data
55    let mut i_mat = vec![vec![0; n_t]; n_p];
56    let mut o_mat = vec![vec![0; n_t]; n_p];
57
58    for e in m.mor_generators() {
59        let e_idx = *hom_inv.get(&e).unwrap();
60        let (inputs, outputs) = transition_interface(m, &e);
61        for ob in inputs {
62            if let ModalOb::Generator(u) = ob {
63                i_mat[*ob_inv.get(&u).unwrap()][e_idx] += 1;
64            }
65        }
66        for ob in outputs {
67            if let ModalOb::Generator(u) = ob {
68                o_mat[*ob_inv.get(&u).unwrap()][e_idx] += 1;
69            }
70        }
71    }
72    let (i_mat_, o_mat_) = (&i_mat, &o_mat);
73
74    // Parse input data
75    //-----------------
76    let mut f: Vec<Vec<_>> =
77        vec![ob_vec.iter().map(|u| *data.forbidden.get(u).unwrap_or(&0)).collect()];
78    let init: Vec<_> = ob_vec.iter().map(|u| *data.tokens.get(u).unwrap_or(&0)).collect();
79
80    // Apply recursive algorithm until fix point
81    //------------------------------------------
82    loop {
83        // For each transition + region (in `f`) pair `(t,v)`, compute the
84        // region that accesses `v` via firing `t`.
85        let pre: Vec<Vec<_>> = (0..n_t)
86            .flat_map(|t| {
87                f.iter().map(move |v| {
88                    (0..n_p).map(move |p| {
89                        std::cmp::max(i_mat_[p][t], v[p] - (o_mat_[p][t] - i_mat_[p][t]))
90                    })
91                })
92            })
93            .map(|z| z.collect())
94            .collect();
95
96        // Filter `pre` for regions which are not already in `f`.
97        let newstuff: Vec<Vec<_>> = pre
98            .into_iter()
99            .filter(|v| f.iter().all(|old| (0..n_p).any(|p| v[p] < old[p])))
100            .unique()
101            .collect();
102
103        // We have terminated when there is nothing new generated by `pre`
104        if newstuff.is_empty() {
105            break;
106        }
107
108        // Update f with new stuff and remove extraneous old stuff
109        f.retain(|v| newstuff.iter().all(|n| (0..n_p).any(|p| v[p] < n[p])));
110        f.extend(newstuff);
111    }
112
113    // Check whether input tokening lies within the region which can access
114    // the forbidden state, `init`.
115    let init_in_forbbiden = f.iter().any(|v| (0..n_p).all(|p| v[p] <= init[p]));
116    !init_in_forbbiden
117}
118
119#[cfg(test)]
120mod tests {
121    use super::*;
122    use crate::dbl::{model::*, theory::*};
123    use crate::stdlib::th_sym_monoidal_category;
124    use crate::zero::name;
125    use std::rc::Rc;
126
127    /// The example Petri net has the following structure:
128    ///
129    /// ```text
130    ///                      t1 t2  t3   
131    /// let i_mat = vec![vec![0, 1, 0],  p1
132    ///                  vec![0, 1, 0],  p2
133    ///                  vec![0, 0, 1]]; p3
134    ///
135    /// let o_mat = vec![vec![1, 0, 0],  p1
136    ///                  vec![0, 0, 1],  p2
137    ///                  vec![0, 1, 0]]; p3
138    /// ```
139    ///
140    /// **Warning**: the Petri net is drawn incorrectly in Handbook of Model
141    /// Checking.
142    ///
143    /// Let the forbidden state be (0,0,2).
144    ///
145    /// The algorithm terminates in four steps:
146    /// [(0,0,2)] -> [(0,0,2),(1,1,1)] -> [(0,0,2),(0,1,1),(2,2,0)]
147    /// -> [(0,0,2),(0,1,1),(0,2,0)].
148    ///
149    /// So the three ways one can reach the forbidden state are:
150    /// 1. starting in the forbidden state (or any superset)
151    /// 2. having two tokens in p2
152    /// 3. having one token in each p2 and p3
153    ///
154    /// Consider using algorithm from "Minimal Coverability Tree Construction
155    /// Made Complete and Efficient" for a more efficient algorithm which allows
156    /// "inf" as a possible specification of an invalid state.
157    #[test]
158    fn validate_subreachability() {
159        // Define a petri net
160        let th = Rc::new(th_sym_monoidal_category());
161        let (ob_type, op) = (ModalObType::new(name("Object")), name("tensor"));
162        let mut model = ModalDblModel::new(th);
163        let (p1, p2, p3, t1, t2, t3) =
164            (name("p1"), name("p2"), name("p3"), name("t1"), name("t2"), name("t3"));
165        model.add_ob(p1.clone(), ob_type.clone());
166        model.add_ob(p2.clone(), ob_type.clone());
167        model.add_ob(p3.clone(), ob_type.clone());
168        let [x, y, z] = [p1.clone(), p2.clone(), p3.clone()].map(ModalOb::from);
169        model.add_mor(
170            t1,
171            ModalOb::App(ModalOb::List(List::Symmetric, vec![]).into(), op.clone()),
172            ModalOb::App(ModalOb::List(List::Symmetric, vec![x.clone()]).into(), op.clone()),
173            ModalMorType::Zero(ob_type.clone()),
174        );
175        model.add_mor(
176            t2,
177            ModalOb::App(
178                ModalOb::List(List::Symmetric, vec![x.clone(), y.clone()]).into(),
179                op.clone(),
180            ),
181            ModalOb::App(ModalOb::List(List::Symmetric, vec![z.clone()]).into(), op.clone()),
182            ModalMorType::Zero(ob_type.clone()),
183        );
184        model.add_mor(
185            t3,
186            ModalOb::App(ModalOb::List(List::Symmetric, vec![z.clone()]).into(), op.clone()),
187            ModalOb::App(ModalOb::List(List::Symmetric, vec![y.clone()]).into(), op.clone()),
188            ModalMorType::Zero(ob_type),
189        );
190
191        // Test starting configurations, see if (0,0,2) is reachable subtokening
192        fn test_input(m: &ModalDblModel<Unital>, x1: i32, x2: i32, x3: i32, expect: bool) {
193            let (p1, p2, p3) = (name("p1"), name("p2"), name("p3"));
194            let forbidden = HashMap::from_iter([(p1.clone(), 0), (p2.clone(), 0), (p3.clone(), 2)]);
195
196            let data = ReachabilityProblemData {
197                tokens: HashMap::from_iter([(p1, x1), (p2, x2), (p3, x3)]),
198                forbidden: forbidden.clone(),
199            };
200            assert_eq!(subreachability(m, data), expect);
201        }
202
203        test_input(&model, 0, 0, 2, false);
204        test_input(&model, 0, 1, 1, false);
205        test_input(&model, 0, 2, 0, false);
206        test_input(&model, 1, 0, 1, true);
207        test_input(&model, 1, 1, 0, true);
208    }
209}
catlog/stdlib/analyses/reachability.rs

catlog/stdlib/analyses/
reachability.rs
