micromegas_tracing/event/
stream.rs

1use serde::{Deserialize, Serialize};
2use std::collections::HashMap;
3use std::sync::Arc;
4use std::sync::atomic::{AtomicUsize, Ordering};
5
6use crate::dyn_deps::DynDeps;
7use crate::event::{EventBlock, ExtractDeps, TracingBlock};
8
9/// StreamDesc is the metadata associated with an event stream
10#[derive(Debug, Serialize, Deserialize)]
11pub struct StreamDesc {
12    pub stream_id: uuid::Uuid,
13    pub process_id: uuid::Uuid,
14    pub tags: Vec<String>,
15    pub properties: HashMap<String, String>,
16}
17
18impl StreamDesc {
19    pub fn new(
20        process_id: uuid::Uuid,
21        tags: &[String],
22        properties: HashMap<String, String>,
23    ) -> Self {
24        let stream_id = uuid::Uuid::new_v4();
25        Self {
26            stream_id,
27            process_id,
28            tags: tags.to_vec(),
29            properties,
30        }
31    }
32}
33
34/// EventStream are a sequence of blocks sent (or dropped) as they become full
35#[derive(Debug)]
36pub struct EventStream<Block> {
37    stream_desc: Arc<StreamDesc>,
38    current_block: Arc<Block>,
39    full_threshold: AtomicUsize,
40}
41
42impl<Block> EventStream<Block>
43where
44    Block: TracingBlock,
45{
46    pub fn new(
47        buffer_size: usize,
48        process_id: uuid::Uuid,
49        tags: &[String],
50        properties: HashMap<String, String>,
51    ) -> Self {
52        let stream_desc = Arc::new(StreamDesc::new(process_id, tags, properties));
53        let block = Arc::new(Block::new(
54            buffer_size,
55            process_id,
56            stream_desc.stream_id,
57            0,
58        ));
59        let max_obj_size = block.hint_max_obj_size();
60        Self {
61            stream_desc,
62            current_block: block,
63            full_threshold: AtomicUsize::new(buffer_size - max_obj_size),
64        }
65    }
66
67    pub fn desc(&self) -> Arc<StreamDesc> {
68        self.stream_desc.clone()
69    }
70
71    pub fn stream_id(&self) -> uuid::Uuid {
72        self.stream_desc.stream_id
73    }
74
75    pub fn set_full(&mut self) {
76        self.full_threshold.store(0, Ordering::Relaxed);
77    }
78
79    pub fn replace_block(&mut self, new_block: Arc<Block>) -> Arc<Block> {
80        let old_block = self.current_block.clone();
81        let max_obj_size = new_block.hint_max_obj_size();
82        self.full_threshold
83            .store(new_block.capacity_bytes() - max_obj_size, Ordering::Relaxed);
84        self.current_block = new_block;
85        old_block
86    }
87
88    pub fn is_full(&self) -> bool {
89        let full_size = self.full_threshold.load(Ordering::Relaxed);
90        self.current_block.len_bytes() > full_size
91    }
92
93    pub fn is_empty(&self) -> bool {
94        self.current_block.len_bytes() == 0
95    }
96
97    pub fn get_block_ref(&self) -> &Block {
98        &self.current_block
99    }
100
101    pub fn get_events_mut(&mut self) -> &mut Block::Queue {
102        //get_mut_unchecked should be faster
103        Arc::get_mut(&mut self.current_block).unwrap().events_mut()
104    }
105
106    pub fn process_id(&self) -> uuid::Uuid {
107        self.stream_desc.process_id
108    }
109
110    pub fn tags(&self) -> &[String] {
111        &self.stream_desc.tags
112    }
113
114    pub fn properties(&self) -> &HashMap<String, String> {
115        &self.stream_desc.properties
116    }
117}
118
119impl<Q> EventStream<EventBlock<Q>>
120where
121    Q: ExtractDeps,
122    EventBlock<Q>: TracingBlock,
123{
124    /// The event queue together with the records of the dynamic ids its events refer to.
125    pub fn get_block_mut_parts(&mut self) -> (&mut Q, &mut DynDeps<Q::DepsQueue>) {
126        let block = Arc::get_mut(&mut self.current_block).unwrap();
127        (&mut block.events, &mut block.dyn_deps)
128    }
129}
micromegas_tracing/event/stream.rs

micromegas_tracing/event/
stream.rs
