micromegas_tracing/event/
block.rs

1use crate::dyn_deps::DynDeps;
2use crate::prelude::*;
3
4/// Blocks are a batch of events sent as a unit
5#[derive(Debug)]
6pub struct EventBlock<Q: ExtractDeps> {
7    pub process_id: uuid::Uuid,
8    pub stream_id: uuid::Uuid,
9    pub begin: DualTime,
10    pub events: Q,
11    pub end: Option<DualTime>,
12    pub event_offset: usize,
13    /// Dependency records for the dynamic property sets and dynamic strings that `events` refer
14    /// to by id, inserted by log, metrics and thread dispatch.
15    pub dyn_deps: DynDeps<Q::DepsQueue>,
16}
17
18impl<Q> EventBlock<Q>
19where
20    Q: micromegas_transit::HeterogeneousQueue + ExtractDeps,
21{
22    pub fn close(&mut self) {
23        self.end = Some(DualTime::now());
24    }
25
26    /// Close this block with an already-stamped `DualTime`, so that it can be
27    /// the same value used as the replacement block's `begin` — making
28    /// `block[k].end == block[k+1].begin` true by construction. Used by the
29    /// four `dispatch.rs` flush paths; `close()` is retained for standalone
30    /// callers (benches/tests) that swap no stream and so have no shared
31    /// timestamp to pass.
32    pub fn close_at(&mut self, end: DualTime) {
33        self.end = Some(end);
34    }
35}
36
37/// Serialize the dependencies of the events in the primary queue
38pub trait ExtractDeps {
39    type DepsQueue: micromegas_transit::HeterogeneousQueue;
40    /// `dyn_deps` holds the records of the dynamic ids that the events refer to. They are
41    /// appended after the static dependencies, which they may reference.
42    fn extract(&self, dyn_deps: &DynDeps<Self::DepsQueue>) -> Self::DepsQueue;
43}
44
45/// Event blocks must adhere to TracingBlock in order to be managed by the stream
46pub trait TracingBlock {
47    type Queue: ExtractDeps;
48
49    fn new(
50        buffer_size: usize,
51        process_id: uuid::Uuid,
52        stream_id: uuid::Uuid,
53        object_offset: usize,
54    ) -> Self;
55    fn len_bytes(&self) -> usize;
56    fn capacity_bytes(&self) -> usize;
57    fn nb_objects(&self) -> usize;
58    fn object_offset(&self) -> usize;
59    fn events_mut(&mut self) -> &mut Self::Queue;
60    fn hint_max_obj_size(&self) -> usize {
61        // blocks with less than this amount of available memory will be considered full
62        128
63    }
64}
65
66impl<Q> TracingBlock for EventBlock<Q>
67where
68    Q: micromegas_transit::HeterogeneousQueue + ExtractDeps,
69{
70    type Queue = Q;
71    fn new(
72        buffer_size: usize,
73        process_id: uuid::Uuid,
74        stream_id: uuid::Uuid,
75        event_offset: usize,
76    ) -> Self {
77        Self {
78            process_id,
79            stream_id,
80            begin: DualTime::now(),
81            events: Q::new(buffer_size),
82            end: None,
83            event_offset,
84            dyn_deps: DynDeps::new(),
85        }
86    }
87
88    fn len_bytes(&self) -> usize {
89        self.events.len_bytes() + self.dyn_deps.len_bytes()
90    }
91
92    fn capacity_bytes(&self) -> usize {
93        self.events.capacity_bytes()
94    }
95
96    fn nb_objects(&self) -> usize {
97        self.events.nb_objects()
98    }
99
100    fn object_offset(&self) -> usize {
101        self.event_offset
102    }
103
104    fn events_mut(&mut self) -> &mut Self::Queue {
105        &mut self.events
106    }
107}
micromegas_tracing/event/block.rs

micromegas_tracing/event/
block.rs
