Skip to main content

windowed_stats/experimental/series/
mod.rs

1// Copyright 2024 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! Round-robin multi-resolution time series.
6
7mod interval;
8
9pub(crate) mod buffer;
10
11pub mod interpolation;
12pub mod metadata;
13pub mod statistic;
14
15use derivative::Derivative;
16use std::fmt::{Debug, Display};
17use std::io;
18use std::marker::PhantomData;
19use std::num::NonZeroUsize;
20
21use crate::experimental::Vec1;
22use crate::experimental::clock::{ObservationTime, Tick, Timed, Timestamp, TimestampExt};
23use crate::experimental::series::buffer::{
24    BufferStrategy, DeltaSimple8bRle, DeltaZigzagSimple8bRle, RingBuffer, Simple8bRle,
25    Uncompressed, ZigzagSimple8bRle, encoding,
26};
27use crate::experimental::series::interpolation::{
28    ConstantSample, Interpolation, InterpolationKind, LastSample,
29};
30use crate::experimental::series::metadata::{BitsetIndex, Metadata};
31use crate::experimental::series::statistic::{
32    FoldError, PostAggregation, SerialStatistic, Statistic,
33};
34
35pub use crate::experimental::series::buffer::{Capacity, decode};
36pub use crate::experimental::series::interval::{SamplingInterval, SamplingProfile};
37
38/// A [`TimeMatrix`] type that can be advanced forward in time.
39///
40/// This trait provides tick operations for `[TimeMatrix`] types. Ticking a [`TimeMatrix`] causes
41/// sample interpolation within and aggregation propagation across [`SamplingInterval`]s.
42///
43/// Importantly, this trait is `dyn` compatible and type erased; it can be used to tick a
44/// [`TimeMatrix`] regardless of its input type parameters (sample type, interpolation type, etc.).
45///
46/// See also the [`TimeMatrixFold`] subtrait.
47pub trait TimeMatrixTick {
48    fn tick(&mut self, timestamp: Timestamp) -> Result<(), FoldError>;
49
50    fn tick_and_get_buffers(&mut self, timestamp: Timestamp)
51    -> Result<SerializedBuffer, FoldError>;
52}
53
54/// A [`TimeMatrix`] type that can sample data.
55///
56/// This trait provides fold operations for `TimeMatrix` types. Folding samples updates
57/// aggregations and advances a [`TimeMatrix`] forward in time.
58///
59/// See also the [`TimeMatrixTick`] supertrait. This trait supports both ticking and sampling, but
60/// is not completely type erased: the sample input type parameter `T` is needed.
61pub trait TimeMatrixFold<T>: TimeMatrixTick {
62    fn fold(&mut self, sample: Timed<T>) -> Result<(), FoldError>;
63}
64
65/// A type that describes the semantics of data folded by `Sampler`s.
66///
67/// Data semantics determine how statistics are interpreted and time series are aggregated and
68/// buffered.
69pub trait DataSemantic {
70    type Metadata: Metadata;
71
72    fn display() -> impl Display;
73}
74
75/// A continually increasing value.
76///
77/// Counters are analogous to an odometer in a vehicle.
78#[derive(Debug)]
79pub enum Counter {}
80
81impl BufferStrategy<u64, LastSample> for Counter {
82    type Buffer = DeltaSimple8bRle;
83}
84
85impl DataSemantic for Counter {
86    type Metadata = ();
87
88    fn display() -> impl Display {
89        "counter"
90    }
91}
92
93/// A fluctuating value.
94///
95/// Gauges are analogous to a speedometer in a vehicle.
96#[derive(Debug)]
97pub enum Gauge {}
98
99impl<P> BufferStrategy<f32, P> for Gauge
100where
101    P: InterpolationKind,
102{
103    type Buffer = Uncompressed<f32>;
104}
105
106impl BufferStrategy<i64, ConstantSample> for Gauge {
107    type Buffer = ZigzagSimple8bRle;
108}
109
110impl BufferStrategy<i64, LastSample> for Gauge {
111    type Buffer = DeltaZigzagSimple8bRle<i64>;
112}
113
114impl BufferStrategy<u64, ConstantSample> for Gauge {
115    type Buffer = Simple8bRle;
116}
117
118impl BufferStrategy<u64, LastSample> for Gauge {
119    type Buffer = DeltaZigzagSimple8bRle<u64>;
120}
121
122impl DataSemantic for Gauge {
123    type Metadata = ();
124
125    fn display() -> impl Display {
126        "gauge"
127    }
128}
129
130/// A semantic like `Gauge` that avoids [`DeltaZigzagSimple8bRle`] until we fix
131/// some other issues.
132///
133/// TODO(https://fxbug.dev/436253782): Delete this type when the viewer can
134/// decode `DeltaZigzagSimple8bRle`.
135///
136/// OR
137///
138/// TODO(https://fxbug.dev/457443158): Delete this type when
139/// `ConstantAggregation` is introduced and netstack's time series is changed to
140/// `TimeMatrix<Diff<u64>, ConstantAggregation>`.
141///
142/// Whichever happens first.
143pub enum GaugeForceSimple8bRle {}
144
145impl<T: Into<u64>> BufferStrategy<T, LastSample> for GaugeForceSimple8bRle {
146    type Buffer = Simple8bRle;
147}
148
149impl DataSemantic for GaugeForceSimple8bRle {
150    type Metadata = ();
151
152    fn display() -> impl Display {
153        "gauge"
154    }
155}
156
157/// A set of Boolean values.
158///
159/// Bitsets are analogous to indicator lamps in a vehicle.
160#[derive(Debug)]
161pub enum Bitset {}
162
163impl<A, P> BufferStrategy<A, P> for Bitset
164where
165    Simple8bRle: RingBuffer<A>,
166    P: InterpolationKind,
167{
168    type Buffer = Simple8bRle;
169}
170
171impl DataSemantic for Bitset {
172    type Metadata = BitsetIndex;
173
174    fn display() -> impl Display {
175        "bitset"
176    }
177}
178
179/// A buffer of serialized data from a time series.
180#[derive(Clone, Debug)]
181struct SerializedTimeSeries {
182    interval: SamplingInterval,
183    data: Vec<u8>,
184}
185
186impl SerializedTimeSeries {
187    /// Gets the sampling interval for the aggregations in the buffer.
188    pub fn interval(&self) -> &SamplingInterval {
189        &self.interval
190    }
191}
192
193/// An unbuffered statistical time series specification.
194///
195/// This type samples and interpolates timed data and produces aggregations per its statistic and
196/// sampling interval. It is a specification insofar that it does **not** buffer the series of
197/// aggregations.
198#[derive(Clone, Debug)]
199struct TimeSeries<F>
200where
201    F: Statistic,
202{
203    interval: SamplingInterval,
204    statistic: F,
205}
206
207impl<F> TimeSeries<F>
208where
209    F: Statistic,
210{
211    pub fn new(interval: SamplingInterval) -> Self
212    where
213        F: Default,
214    {
215        TimeSeries { interval, statistic: F::default() }
216    }
217
218    pub const fn with_statistic(interval: SamplingInterval, statistic: F) -> Self {
219        TimeSeries { interval, statistic }
220    }
221
222    /// Folds interpolations for intervals intersected by the given [`Tick`] and gets the
223    /// aggregations.
224    ///
225    /// The returned iterator performs the computation and so it must be consumed to change the
226    /// state of the statistic.
227    ///
228    /// [`Tick`]: crate::experimental::clock::Tick
229    #[must_use]
230    fn interpolate_and_get_aggregations<'i, P>(
231        &'i mut self,
232        interpolation: &'i mut P,
233        tick: Tick,
234    ) -> impl 'i + Iterator<Item = Result<(NonZeroUsize, F::Aggregation), FoldError>>
235    where
236        P: Interpolation<F::Sample>,
237    {
238        self.interval.fold_and_get_expirations(tick, PhantomData::<F::Sample>).flat_map(
239            move |expiration| {
240                expiration
241                    .interpolate_and_get_aggregation(&mut self.statistic, interpolation)
242                    .transpose()
243            },
244        )
245    }
246
247    /// Folds the given sample and interpolations for intervals intersected by the given [`Tick`]
248    /// and gets the aggregations.
249    ///
250    /// The returned iterator performs the computation and so it must be consumed to change the
251    /// state of the statistic.
252    ///
253    /// [`Tick`]: crate::experimental::clock::Tick
254    #[must_use]
255    fn fold_and_get_aggregations<'i, P>(
256        &'i mut self,
257        interpolation: &'i mut P,
258        tick: Tick,
259        sample: F::Sample,
260    ) -> impl 'i + Iterator<Item = Result<(NonZeroUsize, F::Aggregation), FoldError>>
261    where
262        P: Interpolation<F::Sample>,
263    {
264        self.interval.fold_and_get_expirations(tick, sample).flat_map(move |expiration| {
265            expiration.fold_and_get_aggregation(&mut self.statistic, interpolation).transpose()
266        })
267    }
268
269    /// Gets the sampling interval of the series.
270    pub fn interval(&self) -> &SamplingInterval {
271        &self.interval
272    }
273}
274
275impl<F, R, A> TimeSeries<PostAggregation<F, R>>
276where
277    F: Default + Statistic,
278    R: Clone + Fn(F::Aggregation) -> A,
279    A: Clone,
280{
281    pub fn with_transform(interval: SamplingInterval, transform: R) -> Self {
282        TimeSeries { interval, statistic: PostAggregation::from_transform(transform) }
283    }
284}
285
286/// A buffered round-robin statistical time series.
287///
288/// This type composes a [`TimeSeries`] with a round-robin buffer of aggregations and interpolation
289/// state. Aggregations produced by the time series when sampling or interpolating are pushed into
290/// the buffer.
291#[derive(Derivative)]
292#[derivative(
293    Clone(bound = "F: Clone, F::Buffer: Clone, P::Output<F::Sample>: Clone,"),
294    Debug(bound = "F: Debug,
295                   F::Buffer: Debug,
296                   P::Output<F::Sample>: Debug,")
297)]
298struct BufferedTimeSeries<F, P>
299where
300    F: SerialStatistic<P>,
301    P: InterpolationKind,
302{
303    buffer: F::Buffer,
304    interpolation: P::Output<F::Sample>,
305    series: TimeSeries<F>,
306}
307
308impl<F, P> BufferedTimeSeries<F, P>
309where
310    F: SerialStatistic<P>,
311    P: InterpolationKind,
312{
313    pub fn new(interpolation: P::Output<F::Sample>, series: TimeSeries<F>) -> Self {
314        let buffer = F::buffer(&series.interval);
315        BufferedTimeSeries { buffer, interpolation, series }
316    }
317
318    /// Folds interpolations for intervals intersected by the given [`Tick`] and buffers the
319    /// aggregations.
320    ///
321    /// # Errors
322    ///
323    /// Returns an error if sampling fails.
324    ///
325    /// [`Tick`]: crate::experimental::clock::Tick
326    fn interpolate(&mut self, tick: Tick) -> Result<(), FoldError> {
327        for aggregation in
328            self.series.interpolate_and_get_aggregations(&mut self.interpolation, tick)
329        {
330            let (count, aggregation) = aggregation?;
331            if count.get() == 1 {
332                self.buffer.push(aggregation);
333            } else {
334                self.buffer.fill(aggregation, count);
335            }
336        }
337        Ok(())
338    }
339
340    /// Folds the given sample and interpolations for intervals intersected by the given [`Tick`]
341    /// and buffers the aggregations.
342    ///
343    /// # Errors
344    ///
345    /// Returns an error if sampling fails.
346    ///
347    /// [`Tick`]: crate::experimental::clock::Tick
348    fn fold(&mut self, tick: Tick, sample: F::Sample) -> Result<(), FoldError> {
349        for aggregation in
350            self.series.fold_and_get_aggregations(&mut self.interpolation, tick, sample)
351        {
352            let (count, aggregation) = aggregation?;
353            if count.get() == 1 {
354                self.buffer.push(aggregation);
355            } else {
356                self.buffer.fill(aggregation, count);
357            }
358        }
359        Ok(())
360    }
361
362    pub fn serialize_and_get_buffer(&self) -> io::Result<SerializedTimeSeries> {
363        let mut data = vec![];
364        self.buffer.serialize(&mut data)?;
365        Ok(SerializedTimeSeries { interval: *self.series.interval(), data })
366    }
367}
368
369/// A buffer of data from time matrix.
370#[derive(Clone, Debug, PartialEq)]
371pub struct SerializedBuffer {
372    pub data_semantic: String,
373    pub data: Vec<u8>,
374}
375
376impl SerializedBuffer {
377    /// Records the current state of this `TimeMatrix` into `node`.
378    pub fn write_to_inspect(self, node: &fuchsia_inspect::Node) {
379        let Self { data_semantic, data } = self;
380        node.record_string("type", data_semantic);
381        node.record_bytes("data", data);
382    }
383
384    /// Records an attempt at retrieving a serialized buffer to inspect.
385    pub fn write_to_inspect_or_error<E: Debug>(
386        result: Result<Self, E>,
387        node: &fuchsia_inspect::Node,
388    ) {
389        match result {
390            Ok(b) => b.write_to_inspect(node),
391            Err(e) => node.record_string("type", format!("error: {:?}", e)),
392        }
393    }
394}
395
396/// One or more statistical round-robin time series.
397///
398/// A time matrix is a round-robin multi-resolution time series that samples and interpolates timed
399/// data, computes statistical aggregations for elapsed [sampling intervals][`SamplingInterval`],
400/// and buffers those aggregations. The sample data, statistic, and interpolation of series in a
401/// time matrix must be the same, but the sampling intervals can and should differ.
402#[derive(Derivative)]
403#[derivative(
404    Clone(bound = "F: Clone, F::Buffer: Clone, P::Output<F::Sample>: Clone,"),
405    Debug(bound = "F: Debug,
406                   F::Buffer: Debug,
407                   P::Output<F::Sample>: Debug,")
408)]
409pub struct TimeMatrix<F, P>
410where
411    F: SerialStatistic<P>,
412    P: InterpolationKind,
413{
414    created: Timestamp,
415    last: ObservationTime,
416    buffers: Vec1<BufferedTimeSeries<F, P>>,
417}
418
419impl<F, P> TimeMatrix<F, P>
420where
421    F: SerialStatistic<P>,
422    P: InterpolationKind,
423{
424    fn from_series_with<Q>(
425        created: Timestamp,
426        series: impl Into<Vec1<TimeSeries<F>>>,
427        mut interpolation: Q,
428    ) -> Self
429    where
430        Q: FnMut() -> P::Output<F::Sample>,
431    {
432        let buffers =
433            series.into().map_into(|series| BufferedTimeSeries::new((interpolation)(), series));
434        TimeMatrix { created, last: ObservationTime::at(created), buffers }
435    }
436
437    /// Constructs a time matrix with the given sampling profile and interpolation.
438    ///
439    /// Statistics are default initialized.
440    pub fn new(profile: impl Into<SamplingProfile>, interpolation: P::Output<F::Sample>) -> Self
441    where
442        F: Default,
443    {
444        Self::new_at(Timestamp::now(), profile, interpolation)
445    }
446
447    pub(crate) fn new_at(
448        timestamp: Timestamp,
449        profile: impl Into<SamplingProfile>,
450        interpolation: P::Output<F::Sample>,
451    ) -> Self
452    where
453        F: Default,
454    {
455        let sampling_intervals = profile.into().into_sampling_intervals();
456        TimeMatrix::from_series_with(
457            timestamp,
458            sampling_intervals.map_into(TimeSeries::new),
459            || interpolation.clone(),
460        )
461    }
462
463    /// Constructs a time matrix with the given statistic.
464    pub fn with_statistic(
465        profile: impl Into<SamplingProfile>,
466        interpolation: P::Output<F::Sample>,
467        statistic: F,
468    ) -> Self {
469        let sampling_intervals = profile.into().into_sampling_intervals();
470        TimeMatrix::from_series_with(
471            Timestamp::now(),
472            sampling_intervals
473                .map_into(|window| TimeSeries::with_statistic(window, statistic.clone())),
474            || interpolation.clone(),
475        )
476    }
477
478    /// Folds the given sample and interpolations and gets the aggregation buffers.
479    ///
480    /// To fold a sample without serializing buffers, use [`Sampler::fold`].
481    ///
482    /// [`Sampler::fold`]: crate::experimental::series::Sampler::fold
483    pub fn fold_and_get_buffers(
484        &mut self,
485        sample: Timed<F::Sample>,
486    ) -> Result<SerializedBuffer, FoldError> {
487        self.fold(sample)?;
488        let series_buffers = self
489            .buffers
490            .try_map_ref(BufferedTimeSeries::serialize_and_get_buffer)
491            .map_err::<FoldError, _>(From::from)?;
492        self.serialize(series_buffers).map_err(From::from)
493    }
494
495    fn serialize(
496        &self,
497        series_buffers: Vec1<SerializedTimeSeries>,
498    ) -> io::Result<SerializedBuffer> {
499        use crate::experimental::clock::DurationExt;
500        use byteorder::{LittleEndian, WriteBytesExt};
501        use std::io::Write;
502
503        let created_timestamp = u32::try_from(self.created.quantize()).unwrap_or(u32::MAX);
504        let end_timestamp =
505            u32::try_from(self.last.last_update_timestamp.quantize()).unwrap_or(u32::MAX);
506
507        let mut buffer = vec![];
508        buffer.write_u8(1)?; // Version number.
509        buffer.write_u32::<LittleEndian>(created_timestamp)?; // Matrix creation time.
510        buffer.write_u32::<LittleEndian>(end_timestamp)?; // Last observed or interpolated sample
511        // time.
512        encoding::serialize_buffer_type_descriptors::<F, P>(&mut buffer)?; // Buffer descriptors.
513
514        for series in series_buffers {
515            const GRANULARITY_FIELD_LEN: usize = 2;
516            let len = u16::try_from(series.data.len() + GRANULARITY_FIELD_LEN).unwrap_or(u16::MAX);
517            let granularity =
518                u16::try_from(series.interval().duration().into_quanta()).unwrap_or(u16::MAX);
519
520            buffer.write_u16::<LittleEndian>(len)?;
521            buffer.write_u16::<LittleEndian>(granularity)?;
522            buffer.write_all(&series.data[..len as usize - GRANULARITY_FIELD_LEN])?;
523        }
524        Ok(SerializedBuffer {
525            data_semantic: format!("{}", <F as Statistic>::Semantic::display()),
526            data: buffer,
527        })
528    }
529}
530
531impl<F, R, P, A> TimeMatrix<PostAggregation<F, R>, P>
532where
533    PostAggregation<F, R>: SerialStatistic<P, Aggregation = A>,
534    F: Default + SerialStatistic<P>,
535    R: Clone + Fn(F::Aggregation) -> A,
536    P: InterpolationKind,
537    A: Clone,
538{
539    /// Constructs a time matrix with the default statistic and given transform for
540    /// post-aggregation.
541    pub fn with_transform(
542        profile: impl Into<SamplingProfile>,
543        interpolation: P::Output<<PostAggregation<F, R> as Statistic>::Sample>,
544        transform: R,
545    ) -> Self
546    where
547        R: Clone,
548    {
549        let sampling_intervals = profile.into().into_sampling_intervals();
550        TimeMatrix::from_series_with(
551            Timestamp::now(),
552            sampling_intervals
553                .map_into(|window| TimeSeries::with_transform(window, transform.clone())),
554            || interpolation.clone(),
555        )
556    }
557}
558
559impl<F, P> Default for TimeMatrix<F, P>
560where
561    F: Default + SerialStatistic<P>,
562    P: InterpolationKind,
563    P::Output<F::Sample>: Default,
564{
565    fn default() -> Self {
566        TimeMatrix::new(SamplingProfile::default(), P::Output::default())
567    }
568}
569
570impl<F, P> TimeMatrixFold<F::Sample> for TimeMatrix<F, P>
571where
572    F: SerialStatistic<P>,
573    P: InterpolationKind,
574{
575    fn fold(&mut self, sample: Timed<F::Sample>) -> Result<(), FoldError> {
576        let (timestamp, sample) = sample.into();
577        let tick = self.last.tick(timestamp, true)?;
578        Ok(for buffer in self.buffers.iter_mut() {
579            buffer.fold(tick, sample.clone())?;
580        })
581    }
582}
583
584impl<F, P> TimeMatrixTick for TimeMatrix<F, P>
585where
586    F: SerialStatistic<P>,
587    P: InterpolationKind,
588{
589    fn tick(&mut self, timestamp: Timestamp) -> Result<(), FoldError> {
590        let tick = self.last.tick(timestamp.into(), false)?;
591        Ok(for buffer in self.buffers.iter_mut() {
592            buffer.interpolate(tick)?;
593        })
594    }
595
596    fn tick_and_get_buffers(
597        &mut self,
598        timestamp: Timestamp,
599    ) -> Result<SerializedBuffer, FoldError> {
600        self.tick(timestamp)?;
601        let series_buffers = self
602            .buffers
603            .try_map_ref(BufferedTimeSeries::serialize_and_get_buffer)
604            .map_err::<FoldError, _>(From::from)?;
605        self.serialize(series_buffers).map_err(From::from)
606    }
607}
608
609#[cfg(test)]
610mod tests {
611    use fuchsia_async as fasync;
612
613    use crate::experimental::clock::{Timed, Timestamp};
614    use crate::experimental::series::interpolation::{ConstantSample, LastSample};
615    use crate::experimental::series::statistic::{
616        ArithmeticMean, LatchMax, Max, PostAggregation, Sum, Transform, Union,
617    };
618    use crate::experimental::series::{
619        SamplingProfile, TimeMatrix, TimeMatrixFold, TimeMatrixTick,
620    };
621
622    fn fold_and_interpolate_f32(matrix: &mut impl TimeMatrixFold<f32>) {
623        matrix.fold(Timed::now(0.0)).unwrap();
624        matrix.fold(Timed::now(1.0)).unwrap();
625        matrix.fold(Timed::now(2.0)).unwrap();
626        matrix.tick(Timestamp::now()).unwrap();
627    }
628
629    // TODO(https://fxbug.dev/356218503): Replace this with meaningful unit tests that assert the
630    //                                    outputs of a `TimeMatrix`.
631    // This "test" is considered successful as long as it builds.
632    #[test]
633    fn static_test_define_time_matrix() {
634        type Mean<T> = ArithmeticMean<T>;
635        type MeanTransform<T, F> = Transform<Mean<T>, F>;
636
637        let _exec = fasync::TestExecutor::new_with_fake_time();
638
639        // Arithmetic mean time matrices.
640        let _ = TimeMatrix::<Mean<f32>, ConstantSample>::default();
641        let _ = TimeMatrix::<Mean<f32>, LastSample>::new(
642            SamplingProfile::balanced(),
643            LastSample::or(0.0f32),
644        );
645        let _ = TimeMatrix::<_, ConstantSample>::with_statistic(
646            SamplingProfile::granular(),
647            ConstantSample::default(),
648            Mean::<f32>::default(),
649        );
650
651        // Discrete arithmetic mean time matrices.
652        let mut matrix = TimeMatrix::<MeanTransform<f32, i64>, LastSample>::with_transform(
653            SamplingProfile::highly_granular(),
654            LastSample::or(0.0f32),
655            |aggregation| aggregation.ceil() as i64,
656        );
657        fold_and_interpolate_f32(&mut matrix);
658        // This time matrix is constructed verbosely with no ad-hoc type definitions nor ergonomic
659        // constructors. This is as raw as it gets.
660        let mut matrix = TimeMatrix::<_, ConstantSample>::with_statistic(
661            SamplingProfile::default(),
662            ConstantSample::default(),
663            PostAggregation::<ArithmeticMean<f32>, _>::from_transform(|aggregation: f32| {
664                aggregation.ceil() as i64
665            }),
666        );
667        fold_and_interpolate_f32(&mut matrix);
668    }
669
670    // TODO(https://fxbug.dev/356218503): Replace this with meaningful unit tests that assert the
671    //                                    outputs of a `TimeMatrix`.
672    // This "test" is considered successful as long as it builds.
673    #[test]
674    fn static_test_supported_statistic_and_interpolation_combinations() {
675        let _exec = fasync::TestExecutor::new_with_fake_time();
676
677        let _ = TimeMatrix::<ArithmeticMean<f32>, ConstantSample>::default();
678        let _ = TimeMatrix::<ArithmeticMean<f32>, LastSample>::default();
679        let _ = TimeMatrix::<LatchMax<u64>, LastSample>::default();
680        let _ = TimeMatrix::<Max<u64>, ConstantSample>::default();
681        let _ = TimeMatrix::<Max<u64>, LastSample>::default();
682        let _ = TimeMatrix::<Sum<u64>, ConstantSample>::default();
683        let _ = TimeMatrix::<Sum<u64>, LastSample>::default();
684        let _ = TimeMatrix::<Union<u64>, ConstantSample>::default();
685        let _ = TimeMatrix::<Union<u64>, LastSample>::default();
686    }
687
688    #[test]
689    fn time_matrix_with_uncompressed_buffer() {
690        let exec = fasync::TestExecutor::new_with_fake_time();
691        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(3_000_000_000));
692        let mut time_matrix = TimeMatrix::<ArithmeticMean<f32>, ConstantSample>::new(
693            SamplingProfile::highly_granular(),
694            ConstantSample::default(),
695        );
696        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
697        assert_eq!(
698            buffer.data,
699            vec![
700                1, // version number
701                3, 0, 0, 0, // created timestamp
702                3, 0, 0, 0, // last timestamp
703                0, 0, // type: uncompressed; subtype: f32
704                4, 0, // series 1: length in bytes
705                10, 0, // series 1 granularity: 10s
706                0, 0, // number of elements
707                4, 0, // series 2: length in bytes
708                60, 0, // series 2 granularity: 60s
709                0, 0, // number of elements
710            ]
711        );
712
713        time_matrix.fold(Timed::now(f32::from_bits(42u32))).unwrap();
714        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(10_000_000_000));
715        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
716        assert_eq!(
717            buffer.data,
718            vec![
719                1, // version number
720                3, 0, 0, 0, // created timestamp
721                10, 0, 0, 0, // last timestamp
722                0, 0, // type: uncompressed; subtype: f32
723                8, 0, // series 1: length in bytes
724                10, 0, // series 1 granularity: 10s
725                1, 0, // number of elements
726                42, 0, 0, 0, // item 1
727                4, 0, // series 2: length in bytes
728                60, 0, // series 2 granularity: 60s
729                0, 0, // number of elements
730            ]
731        );
732
733        // Advance several time steps to test ring buffer's `fill`
734        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(50_000_000_000));
735        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
736        assert_eq!(
737            buffer.data,
738            vec![
739                1, // version number
740                3, 0, 0, 0, // created timestamp
741                50, 0, 0, 0, // last timestamp
742                0, 0, // type: uncompressed; subtype: f32
743                24, 0, // series 1: length in bytes
744                10, 0, // series 1 granularity: 10s
745                5, 0, // number of elements
746                42, 0, 0, 0, // item 1
747                0, 0, 0, 0, // item 2
748                0, 0, 0, 0, // item 3
749                0, 0, 0, 0, // item 4
750                0, 0, 0, 0, // item 5
751                4, 0, // series 2: length in bytes
752                60, 0, // series 2 granularity: 60s
753                0, 0, // number of elements
754            ]
755        );
756    }
757
758    #[test]
759    fn time_matrix_with_simple8b_rle_buffer() {
760        let exec = fasync::TestExecutor::new_with_fake_time();
761        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(3_000_000_000));
762        let mut time_matrix = TimeMatrix::<Max<u64>, ConstantSample>::new(
763            SamplingProfile::highly_granular(),
764            ConstantSample::default(),
765        );
766        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
767        assert_eq!(
768            buffer.data,
769            vec![
770                1, // version number
771                3, 0, 0, 0, // created timestamp
772                3, 0, 0, 0, // last timestamp
773                1, 0, // type: simple8b RLE; subtype: unsigned
774                7, 0, // series 1: length in bytes
775                10, 0, // series 1 granularity: 10s
776                0, 0, // number of selector elements and value blocks
777                0, 0, // head selector index
778                0, // number of values in last block
779                7, 0, // series 2: length in bytes
780                60, 0, // series 2 granularity: 60s
781                0, 0, // number of selector elements and value blocks
782                0, 0, // head selector index
783                0, // number of values in last block
784            ]
785        );
786
787        time_matrix.fold(Timed::now(15)).unwrap();
788        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(10_000_000_000));
789        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
790        assert_eq!(
791            buffer.data,
792            vec![
793                1, // version number
794                3, 0, 0, 0, // created timestamp
795                10, 0, 0, 0, // last timestamp
796                1, 0, // type: simple8b RLE; subtype: unsigned
797                16, 0, // series 1: length in bytes
798                10, 0, // series 1 granularity: 10s
799                1, 0, // number of selector elements and value blocks
800                0, 0,    // head selector index
801                1,    // number of values in last block
802                0x0f, // RLE selector
803                15, 0, 0, 0, 0, 0, 1, 0, // value 15 appears 1 time
804                7, 0, // series 2: length in bytes
805                60, 0, // series 2 granularity: 60s
806                0, 0, // number of selector elements and value blocks
807                0, 0, // head selector index
808                0, // number of values in last block
809            ]
810        );
811
812        // Advance several time steps to test ring buffer's `fill`
813        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(50_000_000_000));
814        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
815        assert_eq!(
816            buffer.data,
817            vec![
818                1, // version number
819                3, 0, 0, 0, // created timestamp
820                50, 0, 0, 0, // last timestamp
821                1, 0, // type: simple8b RLE; subtype: unsigned
822                16, 0, // series 1: length in bytes
823                10, 0, // series 1 granularity: 10s
824                1, 0, // number of selector elements and value blocks
825                0, 0,    // head selector index
826                5,    // number of values in last block
827                0x03, // 4-bit selector
828                0x0f, 0, 0, 0, 0, 0, 0, 0, // values 15, 0, 0, 0, 0
829                7, 0, // series 2: length in bytes
830                60, 0, // series 2 granularity: 60s
831                0, 0, // number of selector elements and value blocks
832                0, 0, // head selector index
833                0, // number of values in last block
834            ]
835        );
836    }
837
838    #[test]
839    fn time_matrix_with_zigzag_simple8b_rle_buffer() {
840        let exec = fasync::TestExecutor::new_with_fake_time();
841        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(3_000_000_000));
842        let mut time_matrix = TimeMatrix::<Max<i64>, ConstantSample>::new(
843            SamplingProfile::highly_granular(),
844            ConstantSample::default(),
845        );
846        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
847        assert_eq!(
848            buffer.data,
849            vec![
850                1, // version number
851                3, 0, 0, 0, // created timestamp
852                3, 0, 0, 0, // last timestamp
853                1, 1, // type: simple8b RLE; subtype: signed (zigzag encoded)
854                7, 0, // series 1: length in bytes
855                10, 0, // series 1 granularity: 10s
856                0, 0, // number of selector elements and value blocks
857                0, 0, // head selector index
858                0, // number of values in last block
859                7, 0, // series 2: length in bytes
860                60, 0, // series 2 granularity: 60s
861                0, 0, // number of selector elements and value blocks
862                0, 0, // head selector index
863                0, // number of values in last block
864            ]
865        );
866
867        time_matrix.fold(Timed::now(-8)).unwrap();
868        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(10_000_000_000));
869        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
870        assert_eq!(
871            buffer.data,
872            vec![
873                1, // version number
874                3, 0, 0, 0, // created timestamp
875                10, 0, 0, 0, // last timestamp
876                1, 1, // type: simple8b RLE; subtype: signed (zigzag encoded)
877                16, 0, // series 1: length in bytes
878                10, 0, // series 1 granularity: 10s
879                1, 0, // number of selector elements and value blocks
880                0, 0,    // head selector index
881                1,    // number of values in last block
882                0x0f, // RLE selector
883                15, 0, 0, 0, 0, 0, 1, 0, // value -8 (encoded as 15) appears 1 time
884                7, 0, // series 2: length in bytes
885                60, 0, // series 2 granularity: 60s
886                0, 0, // number of selector elements and value blocks
887                0, 0, // head selector index
888                0, // number of values in last block
889            ]
890        );
891
892        // Advance several time steps to test ring buffer's `fill`
893        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(50_000_000_000));
894        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
895        assert_eq!(
896            buffer.data,
897            vec![
898                1, // version number
899                3, 0, 0, 0, // created timestamp
900                50, 0, 0, 0, // last timestamp
901                1, 1, // type: simple8b RLE; subtype: signed (zigzag encoded)
902                16, 0, // series 1: length in bytes
903                10, 0, // series 1 granularity: 10s
904                1, 0, // number of selector elements and value blocks
905                0, 0,    // head selector index
906                5,    // number of values in last block
907                0x03, // 4-bit selector
908                0x0f, 0, 0, 0, 0, 0, 0, 0, // values -8 (encoded as 15), 0, 0, 0, 0
909                7, 0, // series 2: length in bytes
910                60, 0, // series 2 granularity: 60s
911                0, 0, // number of selector elements and value blocks
912                0, 0, // head selector index
913                0, // number of values in last block
914            ]
915        );
916    }
917
918    #[test]
919    fn time_matrix_with_delta_simple8b_rle_buffer() {
920        let exec = fasync::TestExecutor::new_with_fake_time();
921        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(3_000_000_000));
922        let mut time_matrix = TimeMatrix::<LatchMax<u64>, LastSample>::new(
923            SamplingProfile::highly_granular(),
924            LastSample::or(0),
925        );
926        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
927        assert_eq!(
928            buffer.data,
929            vec![
930                1, // version number
931                3, 0, 0, 0, // created timestamp
932                3, 0, 0, 0, // last timestamp
933                2, 0, // type: delta simple8b RLE; subtype: unsigned
934                7, 0, // series 1: length in bytes
935                10, 0, // series 1 granularity: 10s
936                0, 0, // number of base value + selector elements or value blocks
937                0, 0, // head selector index
938                0, // number of values in last block
939                7, 0, // series 2: length in bytes
940                60, 0, // series 2 granularity: 60s
941                0, 0, // number of base value + selector elements or value blocks
942                0, 0, // head selector index
943                0, // number of values in last block
944            ]
945        );
946
947        time_matrix.fold(Timed::now(42)).unwrap();
948        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(10_000_000_000));
949        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
950        assert_eq!(
951            buffer.data,
952            vec![
953                1, // version number
954                3, 0, 0, 0, // created timestamp
955                10, 0, 0, 0, // last timestamp
956                2, 0, // type: delta simple8b RLE; subtype: unsigned
957                15, 0, // series 1: length in bytes
958                10, 0, // series 1 granularity: 10s
959                1, 0, // number of base value + selector elements or value blocks
960                0, 0, // head selector index
961                0, // number of values in last block
962                42, 0, 0, 0, 0, 0, 0, 0, // base value
963                7, 0, // series 2: length in bytes
964                60, 0, // series 2 granularity: 60s
965                0, 0, // number of base value + selector elements or value blocks
966                0, 0, // head selector index
967                0, // number of values in last block
968            ]
969        );
970
971        time_matrix.fold(Timed::now(57)).unwrap();
972        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(20_000_000_000));
973        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
974        assert_eq!(
975            buffer.data,
976            vec![
977                1, // version number
978                3, 0, 0, 0, // created timestamp
979                20, 0, 0, 0, // last timestamp
980                2, 0, // type: delta simple8b RLE; subtype: unsigned
981                24, 0, // series 1: length in bytes
982                10, 0, // series 1 granularity: 10s
983                2, 0, // number of base value + selector elements or value blocks
984                0, 0, // head selector index
985                1, // number of values in last block
986                42, 0, 0, 0, 0, 0, 0, 0,    // base value
987                0x0f, // RLE selector
988                15, 0, 0, 0, 0, 0, 1, 0, // value 15 (delta) appears 1 time
989                7, 0, // series 2: length in bytes
990                60, 0, // series 2 granularity: 60s
991                0, 0, // number of base value + selector elements or value blocks
992                0, 0, // head selector index
993                0, // number of values in last block
994            ]
995        );
996
997        // Advance several time steps to test ring buffer's `fill`
998        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(50_000_000_000));
999        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1000        assert_eq!(
1001            buffer.data,
1002            vec![
1003                1, // version number
1004                3, 0, 0, 0, // created timestamp
1005                50, 0, 0, 0, // last timestamp
1006                2, 0, // type: delta simple8b RLE; subtype: unsigned
1007                24, 0, // series 1: length in bytes
1008                10, 0, // series 1 granularity: 10s
1009                2, 0, // number of base value + selector elements or value blocks
1010                0, 0, // head selector index
1011                4, // number of values in last block
1012                42, 0, 0, 0, 0, 0, 0, 0,    // base value
1013                0x03, // 4-bit selector
1014                0x0f, 0, 0, 0, 0, 0, 0, 0, // values 15, 0, 0, 0
1015                7, 0, // series 2: length in bytes
1016                60, 0, // series 2 granularity: 60s
1017                0, 0, // number of base value + selector elements or value blocks
1018                0, 0, // head selector index
1019                0, // number of values in last block
1020            ]
1021        );
1022    }
1023
1024    #[test]
1025    fn time_matrix_with_delta_zigzag_simple8b_rle_buffer_i64() {
1026        let exec = fasync::TestExecutor::new_with_fake_time();
1027        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(3_000_000_000));
1028        let mut time_matrix = TimeMatrix::<Max<i64>, LastSample>::new(
1029            SamplingProfile::highly_granular(),
1030            LastSample::or(0),
1031        );
1032        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1033        assert_eq!(
1034            buffer.data,
1035            vec![
1036                1, // version number
1037                3, 0, 0, 0, // created timestamp
1038                3, 0, 0, 0, // last timestamp
1039                2, 1, // type: delta simple8b RLE; subtype: signed
1040                7, 0, // series 1: length in bytes
1041                10, 0, // series 1 granularity: 10s
1042                0, 0, // number of base value + selector elements or value blocks
1043                0, 0, // head selector index
1044                0, // number of values in last block
1045                7, 0, // series 2: length in bytes
1046                60, 0, // series 2 granularity: 60s
1047                0, 0, // number of base value + selector elements or value blocks
1048                0, 0, // head selector index
1049                0, // number of values in last block
1050            ]
1051        );
1052
1053        time_matrix.fold(Timed::now(42)).unwrap();
1054        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(10_000_000_000));
1055        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1056        assert_eq!(
1057            buffer.data,
1058            vec![
1059                1, // version number
1060                3, 0, 0, 0, // created timestamp
1061                10, 0, 0, 0, // last timestamp
1062                2, 1, // type: delta simple8b RLE; subtype: signed
1063                15, 0, // series 1: length in bytes
1064                10, 0, // series 1 granularity: 10s
1065                1, 0, // number of base value + selector elements or value blocks
1066                0, 0, // head selector index
1067                0, // number of values in last block
1068                42, 0, 0, 0, 0, 0, 0, 0, // base value
1069                7, 0, // series 2: length in bytes
1070                60, 0, // series 2 granularity: 60s
1071                0, 0, // number of base value + selector elements or value blocks
1072                0, 0, // head selector index
1073                0, // number of values in last block
1074            ]
1075        );
1076
1077        time_matrix.fold(Timed::now(34)).unwrap();
1078        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(20_000_000_000));
1079        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1080        assert_eq!(
1081            buffer.data,
1082            vec![
1083                1, // version number
1084                3, 0, 0, 0, // created timestamp
1085                20, 0, 0, 0, // last timestamp
1086                2, 1, // type: delta simple8b RLE; subtype: signed
1087                24, 0, // series 1: length in bytes
1088                10, 0, // series 1 granularity: 10s
1089                2, 0, // number of base value + selector elements or value blocks
1090                0, 0, // head selector index
1091                1, // number of values in last block
1092                42, 0, 0, 0, 0, 0, 0, 0,    // base value
1093                0x0f, // RLE selector
1094                15, 0, 0, 0, 0, 0, 1, 0, // value -8 (delta) encoded as 15, appearing 1 time
1095                7, 0, // series 2: length in bytes
1096                60, 0, // series 2 granularity: 60s
1097                0, 0, // number of base value + selector elements or value blocks
1098                0, 0, // head selector index
1099                0, // number of values in last block
1100            ]
1101        );
1102
1103        // Advance several time steps to test ring buffer's `fill`
1104        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(50_000_000_000));
1105        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1106        assert_eq!(
1107            buffer.data,
1108            vec![
1109                1, // version number
1110                3, 0, 0, 0, // created timestamp
1111                50, 0, 0, 0, // last timestamp
1112                2, 1, // type: delta simple8b RLE; subtype: signed
1113                24, 0, // series 1: length in bytes
1114                10, 0, // series 1 granularity: 10s
1115                2, 0, // number of base value + selector elements or value blocks
1116                0, 0, // head selector index
1117                4, // number of values in last block
1118                42, 0, 0, 0, 0, 0, 0, 0,    // base value
1119                0x03, // 4-bit selector
1120                0x0f, 0, 0, 0, 0, 0, 0, 0, // diff values -8 (encoded as 15), 0, 0, 0
1121                7, 0, // series 2: length in bytes
1122                60, 0, // series 2 granularity: 60s
1123                0, 0, // number of base value + selector elements or value blocks
1124                0, 0, // head selector index
1125                0, // number of values in last block
1126            ]
1127        );
1128    }
1129
1130    #[test]
1131    fn time_matrix_with_delta_zigzag_simple8b_rle_buffer_u64() {
1132        let exec = fasync::TestExecutor::new_with_fake_time();
1133        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(3_000_000_000));
1134        let mut time_matrix = TimeMatrix::<Max<u64>, LastSample>::new(
1135            SamplingProfile::highly_granular(),
1136            LastSample::or(0),
1137        );
1138        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1139        assert_eq!(
1140            buffer.data,
1141            vec![
1142                1, // version number
1143                3, 0, 0, 0, // created timestamp
1144                3, 0, 0, 0, // last timestamp
1145                2, 2, // type: delta simple8b RLE; subtype: unsigned with signed diff
1146                7, 0, // series 1: length in bytes
1147                10, 0, // series 1 granularity: 10s
1148                0, 0, // number of base value + selector elements or value blocks
1149                0, 0, // head selector index
1150                0, // number of values in last block
1151                7, 0, // series 2: length in bytes
1152                60, 0, // series 2 granularity: 60s
1153                0, 0, // number of base value + selector elements or value blocks
1154                0, 0, // head selector index
1155                0, // number of values in last block
1156            ]
1157        );
1158
1159        time_matrix.fold(Timed::now(1)).unwrap();
1160        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(10_000_000_000));
1161        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1162        assert_eq!(
1163            buffer.data,
1164            vec![
1165                1, // version number
1166                3, 0, 0, 0, // created timestamp
1167                10, 0, 0, 0, // last timestamp
1168                2, 2, // type: delta simple8b RLE; subtype: unsigned with signed diff
1169                15, 0, // series 1: length in bytes
1170                10, 0, // series 1 granularity: 10s
1171                1, 0, // number of base value + selector elements or value blocks
1172                0, 0, // head selector index
1173                0, // number of values in last block
1174                1, 0, 0, 0, 0, 0, 0, 0, // base value
1175                7, 0, // series 2: length in bytes
1176                60, 0, // series 2 granularity: 60s
1177                0, 0, // number of base value + selector elements or value blocks
1178                0, 0, // head selector index
1179                0, // number of values in last block
1180            ]
1181        );
1182
1183        time_matrix.fold(Timed::now(u64::MAX)).unwrap();
1184        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(20_000_000_000));
1185        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1186        assert_eq!(
1187            buffer.data,
1188            vec![
1189                1, // version number
1190                3, 0, 0, 0, // created timestamp
1191                20, 0, 0, 0, // last timestamp
1192                2, 2, // type: delta simple8b RLE; subtype: unsigned with signed diff
1193                24, 0, // series 1: length in bytes
1194                10, 0, // series 1 granularity: 10s
1195                2, 0, // number of base value + selector elements or value blocks
1196                0, 0, // head selector index
1197                1, // number of values in last block
1198                1, 0, 0, 0, 0, 0, 0, 0,    // base value
1199                0x0f, // RLE selector
1200                3, 0, 0, 0, 0, 0, 1, 0, // value -2 (delta) encoded as 3, appearing 1 time
1201                7, 0, // series 2: length in bytes
1202                60, 0, // series 2 granularity: 60s
1203                0, 0, // number of base value + selector elements or value blocks
1204                0, 0, // head selector index
1205                0, // number of values in last block
1206            ]
1207        );
1208
1209        // Advance several time steps to test ring buffer's `fill`
1210        exec.set_fake_time(fasync::MonotonicInstant::from_nanos(50_000_000_000));
1211        let buffer = time_matrix.tick_and_get_buffers(Timestamp::now()).unwrap();
1212        assert_eq!(
1213            buffer.data,
1214            vec![
1215                1, // version number
1216                3, 0, 0, 0, // created timestamp
1217                50, 0, 0, 0, // last timestamp
1218                2, 2, // type: delta simple8b RLE; subtype: unsigned with signed diff
1219                24, 0, // series 1: length in bytes
1220                10, 0, // series 1 granularity: 10s
1221                2, 0, // number of base value + selector elements or value blocks
1222                0, 0, // head selector index
1223                4, // number of values in last block
1224                1, 0, 0, 0, 0, 0, 0, 0,    // base value
1225                0x01, // 2-bit selector
1226                3, 0, 0, 0, 0, 0, 0, 0, // diff values -2 (encoded as 3), 0, 0, 0
1227                7, 0, // series 2: length in bytes
1228                60, 0, // series 2 granularity: 60s
1229                0, 0, // number of base value + selector elements or value blocks
1230                0, 0, // head selector index
1231                0, // number of values in last block
1232            ]
1233        );
1234    }
1235}