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}