rtlola_hir/modes/

ordering.rs

use std::collections::HashMap;

use petgraph::algo::is_cyclic_directed;
use petgraph::Outgoing;
use serde::{Deserialize, Serialize};

use super::{Ordered, OrderedTrait, TypedTrait};
use crate::hir::{Hir, SRef};
use crate::modes::dependencies::ExtendedDepGraph;
use crate::modes::{DepAnaTrait, DependencyGraph, HirMode};

impl OrderedTrait for Ordered {
    fn stream_layers(&self, sr: SRef) -> StreamLayers {
        self.stream_layers[&sr]
    }
}

/// Represents a layer indicating the position when an expression can be evaluated
#[derive(Debug, Clone, Copy, PartialOrd, Ord, PartialEq, Eq, Serialize, Deserialize)]
pub struct Layer(usize);

/// Wrapper to collect the layer when a stream instance is spawned and evaluated
#[derive(Debug, Clone, Copy, PartialOrd, Ord, PartialEq, Eq, Serialize, Deserialize)]
pub struct StreamLayers {
    spawn: Layer,
    #[cfg(feature = "shift_layer")]
    shift: Layer,
    evaluation: Layer,
}

impl StreamLayers {
    #[cfg(not(feature = "shift_layer"))]
    /// Produces the wrapper [StreamLayers] for a given spawn and evaluation layer
    pub(crate) fn new(spawn_layer: Layer, evaluation_layer: Layer) -> StreamLayers {
        StreamLayers {
            spawn: spawn_layer,
            evaluation: evaluation_layer,
        }
    }

    #[cfg(feature = "shift_layer")]
    /// Produces the wrapper [StreamLayers] for a given spawn, shift and evaluation layer
    pub(crate) fn new(
        spawn_layer: Layer,
        shift_layer: Layer,
        evaluation_layer: Layer,
    ) -> StreamLayers {
        StreamLayers {
            spawn: spawn_layer,
            shift: shift_layer,
            evaluation: evaluation_layer,
        }
    }

    /// Returns the layer when a stream is spawned
    pub fn spawn_layer(&self) -> Layer {
        self.spawn
    }

    #[cfg(feature = "shift_layer")]
    /// Returns the layer when a stream is shifted
    pub fn shift_layer(&self) -> Layer {
        self.shift
    }

    /// Returns the layer when a new stream value is produced
    pub fn evaluation_layer(&self) -> Layer {
        self.evaluation
    }
}

impl From<Layer> for usize {
    fn from(layer: Layer) -> usize {
        layer.0
    }
}

impl Layer {
    /// Produces a [Layer]
    pub fn new(layer: usize) -> Self {
        Layer(layer)
    }

    /// Returns the [Layer] as `usize`
    pub fn inner(self) -> usize {
        self.0
    }
}

impl Ordered {
    /// Returns the spawn and evaluation layer of each stream
    ///
    /// This function analyzes the `spec` and returns the spawn and evaluation layer of each stream.
    /// The analysis splits the dependency graph into two subgraphs: one with the event-based streams and one with the periodic streams.
    /// From these two graphs, it computed the spawn and evaluation layer for event-based and periodic streams separately.
    pub(crate) fn analyze<M>(spec: &Hir<M>) -> Ordered
    where
        M: HirMode + DepAnaTrait + TypedTrait,
    {
        // split graph in periodic and event-based
        let stream_layers = Self::compute_layers(spec, spec.graph());
        Ordered { stream_layers }
    }

    /// Returns the spawn and evaluation layer of for either each event-based stream or periodic stream
    ///
    /// This function splits the dependency graph into two subgraphs and returns the spawn and evaluation layer of each stream.
    /// The first graph only contains the lookups occurring in the spawn part of the stream template.
    /// The analysis computes from this graph the spawn layers of each stream.
    /// The function computes from the second graph the evaluation layers of each stream.
    fn compute_layers<M>(spec: &Hir<M>, graph: &DependencyGraph) -> HashMap<SRef, StreamLayers>
    where
        M: HirMode + DepAnaTrait + TypedTrait,
    {
        // Prepare graphs
        let graph = &graph
            .clone()
            .without_negative_offset_edges()
            .without_close()
            .without_different_pacing(spec);
        let spawn_graph = &graph.clone().only_spawn();
        #[cfg(feature = "shift_layer")]
        let shift_graph = &graph.clone().only_filter();

        debug_assert!(
            !is_cyclic_directed(graph),
            "This should be already checked in the dependency analysis."
        );

        // start analysis
        let mut evaluation_layers = spec
            .inputs()
            .map(|i| {
                (
                    i.sr,
                    Layer::new(if cfg!(feature = "shift_layer") { 1 } else { 0 }),
                )
            })
            .collect::<HashMap<SRef, Layer>>();
        let mut spawn_layers = spec
            .inputs()
            .map(|i| (i.sr, Layer::new(0)))
            .collect::<HashMap<SRef, Layer>>();
        #[cfg(feature = "shift_layer")]
        let mut shift_layers = spec
            .inputs()
            .map(|i| (i.sr, Layer::new(0)))
            .collect::<HashMap<SRef, Layer>>();

        while graph.node_count() != evaluation_layers.len() {
            // build spawn layers
            spawn_graph.node_indices().for_each(|node| {
                let sref = spawn_graph.node_weight(node).unwrap();
                // If we dont know the spawn layer yet
                if !spawn_layers.contains_key(sref) {
                    // get evaluation layer of successors
                    let neighbor_layers: Vec<_> = spawn_graph
                        .neighbors_directed(node, Outgoing)
                        .map(|outgoing_neighbor| {
                            evaluation_layers
                                .get(spawn_graph.node_weight(outgoing_neighbor).unwrap())
                                .copied()
                        })
                        .collect();
                    let computed_spawn_layer = if neighbor_layers.is_empty() {
                        Some(Layer::new(0))
                    } else {
                        neighbor_layers
                            .into_iter()
                            .try_fold(Layer::new(0), |cur_res_layer, neighbor_layer| {
                                neighbor_layer.map(|nl| std::cmp::max(cur_res_layer, nl))
                            })
                            .map(|layer| Layer::new(layer.inner() + 1))
                    };
                    if let Some(layer) = computed_spawn_layer {
                        spawn_layers.insert(*sref, layer);
                    }
                }
            });
            #[cfg(feature = "shift_layer")]
            shift_graph.node_indices().for_each(|node| {
                let sref = shift_graph.node_weight(node).unwrap();
                // if we dont know the shift layer and evaluation layer, but the spawn layer is known
                if !shift_layers.contains_key(sref)
                    && !evaluation_layers.contains_key(sref)
                    && spawn_layers.contains_key(sref)
                {
                    // Get evaluation layer of successors
                    let neighbor_layers: Vec<_> = shift_graph
                        .neighbors_directed(node, Outgoing)
                        .flat_map(|outgoing_neighbor| {
                            //ignore selfloops
                            (outgoing_neighbor != node).then_some(outgoing_neighbor)
                        })
                        .map(|outgoing_neighbor| {
                            evaluation_layers
                                .get(shift_graph.node_weight(outgoing_neighbor).unwrap())
                                .copied()
                        })
                        .collect();
                    let computed_shift_layer = if neighbor_layers.is_empty() {
                        // There are no successors
                        Some(Layer::new(1))
                    } else {
                        neighbor_layers
                            .into_iter()
                            .try_fold(Layer::new(0), |cur_res_layer, neighbor_layer| {
                                neighbor_layer.map(|nl| std::cmp::max(cur_res_layer, nl))
                            })
                            .map(|layer| Layer::new(layer.inner() + 1))
                    };
                    if let Some(layer) = computed_shift_layer {
                        // Shift layer has to be greater than spawn layer
                        let layer = if spawn_layers[sref] < layer {
                            layer
                        } else {
                            Layer::new(spawn_layers[sref].inner() + 1)
                        };
                        shift_layers.insert(*sref, layer);
                    }
                }
            });

            #[cfg(feature = "shift_layer")]
            let previous_layers = &shift_layers;
            #[cfg(not(feature = "shift_layer"))]
            let previous_layers = &spawn_layers;

            graph.node_indices().for_each(|node| {
                let sref = graph.node_weight(node).unwrap();
                // if we dont know the evaluation layer, but the spawn layer is known
                if !evaluation_layers.contains_key(sref) && previous_layers.contains_key(sref) {
                    // Get evaluation layer of successors
                    let neighbor_layers: Vec<_> = graph
                        .neighbors_directed(node, Outgoing)
                        .flat_map(|outgoing_neighbor| {
                            //ignore selfloops
                            (outgoing_neighbor != node).then_some(outgoing_neighbor)
                        })
                        .map(|outgoing_neighbor| {
                            evaluation_layers
                                .get(graph.node_weight(outgoing_neighbor).unwrap())
                                .copied()
                        })
                        .collect();
                    let computed_evaluation_layer = if neighbor_layers.is_empty() {
                        // There are no successors
                        Some(Layer::new(if cfg!(feature = "shift_layer") {
                            2
                        } else {
                            1
                        }))
                    } else {
                        // eval_layer = max(successor_eval_layers) + 1
                        neighbor_layers
                            .into_iter()
                            .try_fold(Layer::new(0), |cur_res_layer, neighbor_layer| {
                                neighbor_layer.map(|nl| std::cmp::max(cur_res_layer, nl))
                            })
                            .map(|layer| Layer::new(layer.inner() + 1))
                    };
                    if let Some(layer) = computed_evaluation_layer {
                        // Evaluation layer has to be greater than spawn layer
                        let layer = if previous_layers[sref] < layer {
                            layer
                        } else {
                            Layer::new(previous_layers[sref].inner() + 1)
                        };
                        evaluation_layers.insert(*sref, layer);
                    }
                }
            });
        }
        evaluation_layers
            .into_iter()
            .map(|(key, evaluation_layer)| {
                #[cfg(feature = "shift_layer")]
                let layer = (
                    key,
                    StreamLayers::new(spawn_layers[&key], shift_layers[&key], evaluation_layer),
                );
                #[cfg(not(feature = "shift_layer"))]
                let layer = (key, StreamLayers::new(spawn_layers[&key], evaluation_layer));
                layer
            })
            .collect::<HashMap<SRef, StreamLayers>>()
    }
}

#[cfg(test)]
fn check_eval_order_for_spec(spec: &str, ref_layers: HashMap<SRef, StreamLayers>) {
    use crate::{config::FrontendConfig, BaseMode};
    use rtlola_parser::{parse, ParserConfig};

    let parser_config = ParserConfig::for_string(spec.to_string());
    let frontend_config = FrontendConfig::from(&parser_config);
    let ast = parse(&parser_config).unwrap_or_else(|e| panic!("{:?}", e));
    let hir = Hir::<BaseMode>::from_ast(ast)
        .unwrap()
        .check_types(&frontend_config)
        .unwrap()
        .analyze_dependencies(&frontend_config)
        .unwrap();
    let order = Ordered::analyze(&hir);
    let Ordered { stream_layers } = order;
    assert_eq!(stream_layers.len(), ref_layers.len());
    stream_layers.iter().for_each(|(sr, layers)| {
        let ref_layers = &ref_layers[sr];
        assert_eq!(ref_layers.spawn_layer(), layers.spawn_layer());
        assert_eq!(ref_layers.evaluation_layer(), layers.evaluation_layer());
        #[cfg(feature = "shift_layer")]
        assert_eq!(ref_layers.shift_layer(), layers.shift_layer());
    });
}

#[cfg(test)]
#[cfg(not(feature = "shift_layer"))]
mod tests {
    use super::*;

    #[test]
    fn synchronous_lookup() {
        let spec = "input a: UInt8\noutput b: UInt8 := a\noutput c:UInt8 := b";
        let sname_to_sref = vec![("a", SRef::In(0)), ("b", SRef::Out(0)), ("c", SRef::Out(1))]
            .into_iter()
            .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn hold_lookup() {
        let spec =
            "input a: UInt8\ninput b:UInt8\noutput c: UInt8 := a.hold().defaults(to: 0) + b\noutput d: UInt8 := c.hold().defaults(to: 0) + a";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::In(1)),
            ("c", SRef::Out(0)),
            ("d", SRef::Out(1)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn offset_lookup() {
        let spec = "input a: UInt8\noutput b: UInt8 := a.offset(by: -1).defaults(to: 0)\noutput c: UInt8 := b.offset(by: -1).defaults(to: 0)";
        let sname_to_sref = vec![("a", SRef::In(0)), ("b", SRef::Out(0)), ("c", SRef::Out(1))]
            .into_iter()
            .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn sliding_window_lookup() {
        let spec = "input a: UInt8\noutput b: UInt8 @1Hz := a.aggregate(over: 1s, using: sum)\noutput c: UInt8 := a + 3\noutput d: UInt8 @1Hz := c.aggregate(over: 1s, using: sum)\noutput e: UInt8 := b + d\noutput f: UInt8 @2Hz := e.hold().defaults(to: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
            ("e", SRef::Out(3)),
            ("f", SRef::Out(4)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let ref_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["e"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
            (
                sname_to_sref["f"],
                StreamLayers::new(Layer::new(0), Layer::new(3)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, ref_layers)
    }

    #[test]
    fn discrete_window_lookup() {
        let spec = "input a: UInt8\n\
        output b: UInt8 := a.aggregate(over_discrete: 5, using: sum)\n\
        output c: UInt8 := a + 3\n\
        output d: UInt8 := c.aggregate(over_discrete: 5, using: sum)";

        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let ref_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, ref_layers)
    }

    #[test]
    fn offset_lookups() {
        let spec = "input a: UInt8\noutput b: UInt8 := a.offset(by:-1).defaults(to: 0)\noutput c: UInt8 := a.offset(by:-2).defaults(to: 0)\noutput d: UInt8 := a.offset(by:-3).defaults(to: 0)\noutput e: UInt8 := a.offset(by:-4).defaults(to: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
            ("e", SRef::Out(3)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["e"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }
    #[test]
    fn negative_loop_different_offsets() {
        let spec = "input a: Int8\noutput b: Int8 := a.offset(by: -1).defaults(to: 0) + d.offset(by:-2).defaults(to:0)\noutput c: Int8 := b.offset(by:-3).defaults(to: 0)\noutput d: Int8 := c.offset(by:-4).defaults(to: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn lookup_chain() {
        let spec = "input a: Int8\noutput b: Int8 := a + d.hold().defaults(to:0)\noutput c: Int8 := b\noutput d: Int8 := c.offset(by:-4).defaults(to: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(3)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn multiple_input_stream() {
        let spec = "input a: Int8\ninput b: Int8\noutput c: Int8 := a + b.hold().defaults(to:0)\noutput d: Int8 := a + c.offset(by: -1).defaults(to: 0)\noutput e: Int8 := c + 3\noutput f: Int8 := c + 6\noutput g: Int8 := b + 3\noutput h: Int8 := g + f";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::In(1)),
            ("c", SRef::Out(0)),
            ("d", SRef::Out(1)),
            ("e", SRef::Out(2)),
            ("f", SRef::Out(3)),
            ("g", SRef::Out(4)),
            ("h", SRef::Out(5)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["e"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
            (
                sname_to_sref["f"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
            (
                sname_to_sref["g"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["h"],
                StreamLayers::new(Layer::new(0), Layer::new(3)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn event_and_periodic_stream_mix() {
        let spec =
            "input a : Int8 \ninput b :Int8\noutput c @2Hz := a.hold().defaults(to: 0) + 3\noutput d @1Hz := a.hold().defaults(to: 0) + c\noutput e := a + b";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::In(1)),
            ("c", SRef::Out(0)),
            ("d", SRef::Out(1)),
            ("e", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let ref_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["e"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, ref_layers)
    }

    #[test]
    fn negative_and_postive_lookups_as_loop() {
        let spec = "input a: Int8\noutput b: Int8 := a + d.offset(by:-1).defaults(to:0)\noutput c: Int8 := b\noutput d: Int8 := c";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(3)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }
    #[test]
    fn sliding_windows_chain_and_hold_lookup() {
        let spec = "input a: Int8\noutput b@1Hz := a.aggregate(over: 1s, using: sum) + d.offset(by: -1).defaults(to: 0)\noutput c@2Hz := b.aggregate(over: 1s, using: sum)\noutput d@2Hz := b.hold().defaults(to: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let ref_layers = vec![
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(2)),
            ),
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, ref_layers)
    }

    #[test]
    fn simple_chain_with_parameter() {
        let spec =
            "input a: Int8\noutput b := a + 5\noutput c(para) spawn with b eval with para + a";
        let sname_to_sref = vec![("a", SRef::In(0)), ("b", SRef::Out(0)), ("c", SRef::Out(1))]
            .into_iter()
            .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(2), Layer::new(3)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn lookup_chain_with_parametrization() {
        let spec = "input a: Int8\noutput b(para) spawn with a when a > 6 eval with a + para\noutput c(para) spawn with a when a > 6 eval with a + b(para)\noutput d(para) spawn with a when a > 6 eval with a + c(para)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(1), Layer::new(2)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(1), Layer::new(3)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(1), Layer::new(4)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn parameter_loop_with_lookup_in_close() {
        let spec = "\
        input a: Int8\n\
        input b: Int8\n\
        output c(p) \n\
            spawn with a when a < b\n\
            eval with p + b + g(p).hold().defaults(to: 0)\n\
        output d(p) \n\
            spawn with b when c(4).hold().defaults(to: 0) < 4\n\
            eval with b + 5\n\
        output e(p)\n\
            spawn with b\n\
            eval with d(p).hold().defaults(to: 0) + b\n\
        output f(p)\n\
            spawn with b\n\
            eval when e(p).hold().defaults(to: 0) < 6 with b + 5\n\
        output g(p)\n\
            spawn with b close @true when f(p).hold().defaults(to: 0) < 6\n\
            eval with b + 5";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::In(1)),
            ("c", SRef::Out(0)),
            ("d", SRef::Out(1)),
            ("e", SRef::Out(2)),
            ("f", SRef::Out(3)),
            ("g", SRef::Out(4)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(1), Layer::new(3)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(4), Layer::new(5)),
            ),
            (
                sname_to_sref["e"],
                StreamLayers::new(Layer::new(1), Layer::new(6)),
            ),
            (
                sname_to_sref["f"],
                StreamLayers::new(Layer::new(1), Layer::new(7)),
            ),
            (
                sname_to_sref["g"],
                StreamLayers::new(Layer::new(1), Layer::new(2)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn parameter_nested_lookup_implicit() {
        let spec = "input a: Int8\n input b: Int8\n output c(p) spawn with a eval with p + b\noutput d := c(c(b).hold().defaults(to: 0)).hold().defaults(to: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::In(1)),
            ("c", SRef::Out(0)),
            ("d", SRef::Out(1)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(1), Layer::new(2)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(3)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn parameter_nested_lookup_explicit() {
        let spec = "input a: Int8\n input b: Int8\n output c(p) spawn with a eval with p + b\noutput d := c(b).hold().defaults(to: 0)\noutput e := c(d).hold().defaults(to: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::In(1)),
            ("c", SRef::Out(0)),
            ("d", SRef::Out(1)),
            ("e", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(1), Layer::new(2)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(3)),
            ),
            (
                sname_to_sref["e"],
                StreamLayers::new(Layer::new(0), Layer::new(4)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn test_spawn_eventbased() {
        let spec = "input a: Int32\n\
                  output b(x: Int32) spawn with a eval with x + a";
        let sname_to_sref = vec![("a", SRef::In(0)), ("b", SRef::Out(0))]
            .into_iter()
            .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(1), Layer::new(2)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn test_delay() {
        let spec = "input a: UInt64\n\
                            output a_counter: UInt64 @a := a_counter.offset(by: -1).defaults(to: 0) + 1\n\
                            output b(p: UInt64) spawn with a_counter when a = 1 close when if true then true else b(p) eval @1Hz with a.hold(or: 0) == 2";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("a_counter", SRef::Out(0)),
            ("b", SRef::Out(1)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let ref_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["a_counter"],
                StreamLayers::new(Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(2), Layer::new(3)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, ref_layers)
    }

    #[test]
    fn test_instance_aggregation() {
        let spec = "input a: Int32\n\
        output b (p) spawn with a eval when a > 5 with b(p).offset(by: -1).defaults(to: 0) + a\n\
        output c eval with b.aggregate(over_instances: fresh, using: Σ)\n";
        let sname_to_sref = vec![("a", SRef::In(0)), ("b", SRef::Out(0)), ("c", SRef::Out(1))]
            .into_iter()
            .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(1), Layer::new(2)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(3)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }
}

#[cfg(test)]
#[cfg(feature = "shift_layer")]
mod tests {
    use std::collections::HashMap;

    use crate::hir::SRef;

    use super::{check_eval_order_for_spec, Layer, StreamLayers};

    #[test]
    fn synchronous_lookup() {
        let spec = "input a: UInt8\noutput b: UInt8 := a\noutput c:UInt8 := b";
        let sname_to_sref = vec![("a", SRef::In(0)), ("b", SRef::Out(0)), ("c", SRef::Out(1))]
            .into_iter()
            .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1), Layer::new(2)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(1), Layer::new(3)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn filtered_spec() {
        let spec = "input a: UInt8
        output b := a + 1
        output c eval when b == 0 with a
        output d eval when a == 0 with c.hold(or: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1), Layer::new(2)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(3), Layer::new(4)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(2), Layer::new(5)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }

    #[test]
    fn filtered_spec_offset() {
        let spec = "input a: UInt8
        output b := a + 1
        output c eval when b == 0 with a
        output d eval when b == 0 with c.last(or: 0)";
        let sname_to_sref = vec![
            ("a", SRef::In(0)),
            ("b", SRef::Out(0)),
            ("c", SRef::Out(1)),
            ("d", SRef::Out(2)),
        ]
        .into_iter()
        .collect::<HashMap<&str, SRef>>();
        let event_layers = vec![
            (
                sname_to_sref["a"],
                StreamLayers::new(Layer::new(0), Layer::new(0), Layer::new(1)),
            ),
            (
                sname_to_sref["b"],
                StreamLayers::new(Layer::new(0), Layer::new(1), Layer::new(2)),
            ),
            (
                sname_to_sref["c"],
                StreamLayers::new(Layer::new(0), Layer::new(3), Layer::new(4)),
            ),
            (
                sname_to_sref["d"],
                StreamLayers::new(Layer::new(0), Layer::new(3), Layer::new(4)),
            ),
        ]
        .into_iter()
        .collect();
        check_eval_order_for_spec(spec, event_layers)
    }
}