Struct rtlola_frontend::mir::TimeDrivenStream

source ·

pub struct TimeDrivenStream {
    pub reference: StreamReference,
    pub frequency: UOM_Frequency,
}

Expand description

Wrapper for output streams providing additional information specific to time-driven streams.

Fields§

§reference: StreamReference

A reference to the stream that is specified.

§frequency: UOM_Frequency

The evaluation frequency of the stream.

Implementations§

source §

impl TimeDrivenStream

source

pub fn period(&self) -> UOM_Time

Returns the evaluation period, i.e., the multiplicative inverse of TimeDrivenStream::frequency.

Examples found in repository ?

src/mir.rs (line 347)

    pub fn period_in_duration(&self) -> Duration {
        Duration::from_nanos(
            self.period()
                .get::<nanosecond>()
                .to_integer()
                .try_into()
                .expect("Period [ns] too large for u64!"),
        )
    }

More examples

Hide additional examples

src/mir/schedule.rs (line 74)

    pub(crate) fn from(ir: &RtLolaMir) -> Result<Schedule, String> {
        let stream_periods = ir
            .time_driven
            .iter()
            .filter_map(|tds| ir.output(tds.reference).is_spawned().not().then(|| tds.period()));
        let spawn_periods = ir.outputs.iter().filter_map(|o| {
            if let PacingType::Periodic(freq) = &o.spawn.pacing {
                Some(UOM_Time::new::<second>(freq.get::<uom::si::frequency::hertz>().inv()))
            } else {
                None
            }
        });
        let close_periods = ir.outputs.iter().filter_map(|o| {
            if let PacingType::Periodic(freq) = &o.close.pacing {
                o.close
                    .has_self_reference
                    .not()
                    .then(|| UOM_Time::new::<second>(freq.get::<uom::si::frequency::hertz>().inv()))
            } else {
                None
            }
        });
        let periods: Vec<UOM_Time> = stream_periods.chain(spawn_periods).chain(close_periods).collect();
        if periods.is_empty() {
            // Nothing to schedule here
            return Ok(Schedule {
                hyper_period: None,
                deadlines: vec![],
            });
        }
        let gcd = Self::find_extend_period(&periods);
        let hyper_period = Self::find_hyper_period(&periods);

        let extend_steps = Self::build_extend_steps(ir, gcd, hyper_period)?;
        let extend_steps = Self::apply_periodicity(&extend_steps);
        let mut deadlines = Self::condense_deadlines(gcd, extend_steps);
        Self::sort_deadlines(ir, &mut deadlines);

        let hyper_period = Duration::from_nanos(hyper_period.get::<nanosecond>().to_integer().to_u64().unwrap());
        Ok(Schedule {
            hyper_period: Some(hyper_period),
            deadlines,
        })
    }

    /// Determines the maximal amount of time the process can wait between successive checks for
    /// due deadlines without missing one.
    fn find_extend_period(rates: &[UOM_Time]) -> UOM_Time {
        assert!(!rates.is_empty());
        let rates: Vec<Rational> = rates.iter().map(|r| r.get::<nanosecond>()).collect();
        let gcd = math::rational_gcd_all(&rates);
        UOM_Time::new::<nanosecond>(gcd)
    }

    /// Determines the hyper period of the given `rates`.
    fn find_hyper_period(rates: &[UOM_Time]) -> UOM_Time {
        assert!(!rates.is_empty());
        let rates: Vec<Rational> = rates.iter().map(|r| r.get::<nanosecond>()).collect();
        let lcm = math::rational_lcm_all(&rates);
        let lcm = math::rational_lcm(lcm, Rational::one()); // needs to be multiple of 1 ns
        UOM_Time::new::<nanosecond>(lcm)
    }

    /// Takes a vec of gcd-sized intervals. In each interval, there are streams that need
    /// to be scheduled periodically at this point in time.
    /// Example:
    /// Hyper-period: 2 seconds, gcd: 500ms, streams: (c @ .5Hz), (b @ 1Hz), (a @ 2Hz)
    /// Input:  `[[a] [b]   []  [c]]`
    /// Output: `[[a] [a,b] [a] [a,b,c]]`
    fn apply_periodicity(steps: &[Vec<Task>]) -> Vec<Vec<Task>> {
        // Whenever there are streams in a cell at index `i`,
        // add them to every cell with index k*i within bounds, where k > 1.
        // k = 0 would always schedule them initially, so this must be skipped.
        // TODO: Skip last half of the array.
        let mut res = vec![Vec::new(); steps.len()];
        for (ix, streams) in steps.iter().enumerate() {
            if !streams.is_empty() {
                let mut k = 1;
                while let Some(target) = res.get_mut(k * (ix + 1) - 1) {
                    target.extend(streams);
                    k += 1;
                }
            }
        }
        res
    }

    /// Build extend steps for each gcd-sized time interval up to the hyper period.
    /// Example:
    /// Hyper-period: 2 seconds, gcd: 500ms, streams: (c @ .5Hz), (b @ 1Hz), (a @ 2Hz)
    /// Result: `[[a] [b] [] [c]]`
    /// Meaning: `a` starts being scheduled after one gcd, `b` after two gcds, `c` after 4 gcds.
    fn build_extend_steps(ir: &RtLolaMir, gcd: UOM_Time, hyper_period: UOM_Time) -> Result<Vec<Vec<Task>>, String> {
        let num_steps = hyper_period.get::<second>() / gcd.get::<second>();
        assert!(num_steps.is_integer());
        let num_steps = num_steps.to_integer() as usize;
        if num_steps >= 10_000_000 {
            return Err("stream frequencies are too incompatible to generate schedule".to_string());
        }
        let mut extend_steps = vec![Vec::new(); num_steps];
        for s in ir
            .time_driven
            .iter()
            .filter(|tds| !ir.output(tds.reference).is_spawned())
        {
            let ix = s.period().get::<second>() / gcd.get::<second>();
            // Period must be integer multiple of gcd by def of gcd
            assert!(ix.is_integer());
            let ix = ix.to_integer() as usize;
            let ix = ix - 1;
            extend_steps[ix].push(Task::Evaluate(s.reference.out_ix()));
        }
        let periodic_spawns = ir.outputs.iter().filter_map(|o| {
            match &o.spawn.pacing {
                PacingType::Periodic(freq) => {
                    Some((
                        o.reference.out_ix(),
                        UOM_Time::new::<second>(freq.get::<uom::si::frequency::hertz>().inv()),
                    ))
                },
                _ => None,
            }
        });
        for (out_ix, period) in periodic_spawns {
            let ix = period.get::<second>() / gcd.get::<second>();
            // Period must be integer multiple of gcd by def of gcd
            assert!(ix.is_integer());
            let ix = ix.to_integer() as usize;
            let ix = ix - 1;
            extend_steps[ix].push(Task::Spawn(out_ix));
        }

        let periodic_close = ir.outputs.iter().filter_map(|o| {
            if let PacingType::Periodic(freq) = &o.close.pacing {
                o.close.has_self_reference.not().then(|| {
                    (
                        o.reference.out_ix(),
                        UOM_Time::new::<second>(freq.get::<uom::si::frequency::hertz>().inv()),
                    )
                })
            } else {
                None
            }
        });
        for (out_ix, period) in periodic_close {
            let ix = period.get::<second>() / gcd.get::<second>();
            // Period must be integer multiple of gcd by def of gcd
            assert!(ix.is_integer());
            let ix = ix.to_integer() as usize;
            let ix = ix - 1;
            extend_steps[ix].push(Task::Close(out_ix));
        }
        Ok(extend_steps)
    }