cu29_unifiedlog/
lib.rs

1#![cfg_attr(not(feature = "std"), no_std)]
2
3#[cfg(not(feature = "std"))]
4extern crate alloc;
5extern crate core;
6
7#[cfg(feature = "std")]
8pub mod memmap;
9
10#[cfg(feature = "std")]
11mod compat {
12    // backward compatibility for the std implementation
13    pub use crate::memmap::MmapUnifiedLogger as UnifiedLogger;
14    pub use crate::memmap::MmapUnifiedLoggerBuilder as UnifiedLoggerBuilder;
15    pub use crate::memmap::MmapUnifiedLoggerRead as UnifiedLoggerRead;
16    pub use crate::memmap::MmapUnifiedLoggerWrite as UnifiedLoggerWrite;
17    pub use crate::memmap::UnifiedLoggerIOReader;
18}
19
20#[cfg(feature = "std")]
21pub use compat::*;
22
23#[cfg(not(feature = "std"))]
24mod imp {
25    pub use alloc::string::ToString;
26    pub use alloc::sync::Arc;
27    pub use alloc::vec::Vec;
28    pub use core::fmt::Debug;
29    pub use core::fmt::Display;
30    pub use core::fmt::Formatter;
31    pub use core::fmt::Result as FmtResult;
32    pub use spin::Mutex;
33}
34
35#[cfg(feature = "std")]
36mod imp {
37    pub use std::fmt::Debug;
38    pub use std::fmt::Display;
39    pub use std::fmt::Formatter;
40    pub use std::fmt::Result as FmtResult;
41    pub use std::sync::Arc;
42    pub use std::sync::Mutex;
43}
44
45use imp::*;
46
47use bincode::error::EncodeError;
48use bincode::{Decode, Encode};
49use cu29_traits::{CuError, CuResult, UnifiedLogType, WriteStream};
50
51/// ID to spot the beginning of a Copper Log
52#[allow(dead_code)]
53pub const MAIN_MAGIC: [u8; 4] = [0xB4, 0xA5, 0x50, 0xFF]; // BRASS OFF
54
55/// ID to spot a section of Copper Log
56pub const SECTION_MAGIC: [u8; 2] = [0xFA, 0x57]; // FAST
57
58pub const SECTION_HEADER_COMPACT_SIZE: u16 = 512; // Usual minimum size for a disk sector.
59
60/// The main file header of the datalogger.
61#[derive(Encode, Decode, Debug)]
62pub struct MainHeader {
63    pub magic: [u8; 4],            // Magic number to identify the file.
64    pub first_section_offset: u16, // This is to align with a page at write time.
65    pub page_size: u16,
66}
67
68impl Display for MainHeader {
69    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
70        writeln!(
71            f,
72            "  Magic -> {:2x}{:2x}{:2x}{:2x}",
73            self.magic[0], self.magic[1], self.magic[2], self.magic[3]
74        )?;
75        writeln!(f, "  first_section_offset -> {}", self.first_section_offset)?;
76        writeln!(f, "  page_size -> {}", self.page_size)
77    }
78}
79
80/// Each concurrent sublogger is tracked through a section header.
81/// They form a linked list of sections.
82/// The entry type is used to identify the type of data in the section.
83#[derive(Encode, Decode, Debug)]
84pub struct SectionHeader {
85    pub magic: [u8; 2],  // Magic number to identify the section.
86    pub block_size: u16, // IMPORTANT: we assume this header fits in this block size.
87    pub entry_type: UnifiedLogType,
88    pub offset_to_next_section: u32, // offset from the first byte of this header to the first byte of the next header (MAGIC to MAGIC).
89    pub used: u32,                   // how much of the section is filled.
90}
91
92impl Display for SectionHeader {
93    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
94        writeln!(f, "    Magic -> {:2x}{:2x}", self.magic[0], self.magic[1])?;
95        writeln!(f, "    type -> {:?}", self.entry_type)?;
96        write!(
97            f,
98            "    use  -> {} / {}",
99            self.used, self.offset_to_next_section
100        )
101    }
102}
103
104impl Default for SectionHeader {
105    fn default() -> Self {
106        Self {
107            magic: SECTION_MAGIC,
108            block_size: 512,
109            entry_type: UnifiedLogType::Empty,
110            offset_to_next_section: 0,
111            used: 0,
112        }
113    }
114}
115
116pub enum AllocatedSection<S: SectionStorage> {
117    NoMoreSpace,
118    Section(SectionHandle<S>),
119}
120
121/// A Storage is an append-only structure that can update a header section.
122pub trait SectionStorage: Send + Sync {
123    /// This rewinds the storage, serialize the header and jumps to the beginning of the user data storage.
124    fn initialize<E: Encode>(&mut self, header: &E) -> Result<usize, EncodeError>;
125    /// This updates the header leaving the position to the end of the user data storage.
126    fn post_update_header<E: Encode>(&mut self, header: &E) -> Result<usize, EncodeError>;
127    /// Appends the entry to the user data storage.
128    fn append<E: Encode>(&mut self, entry: &E) -> Result<usize, EncodeError>;
129    /// Flushes the section to the underlying storage
130    fn flush(&mut self) -> CuResult<usize>;
131}
132
133/// A SectionHandle is a handle to a section in the datalogger.
134/// It allows tracking the lifecycle of the section.
135#[derive(Default)]
136pub struct SectionHandle<S: SectionStorage> {
137    header: SectionHeader, // keep a copy of the header as metadata
138    storage: S,
139}
140
141impl<S: SectionStorage> SectionHandle<S> {
142    pub fn create(header: SectionHeader, mut storage: S) -> CuResult<Self> {
143        // Write the first version of the header.
144        let _ = storage.initialize(&header).map_err(|e| e.to_string())?;
145        Ok(Self { header, storage })
146    }
147    pub fn append<E: Encode>(&mut self, entry: E) -> Result<usize, EncodeError> {
148        self.storage.append(&entry)
149    }
150
151    pub fn get_storage(&self) -> &S {
152        &self.storage
153    }
154
155    pub fn get_storage_mut(&mut self) -> &mut S {
156        &mut self.storage
157    }
158
159    pub fn post_update_header(&mut self) -> Result<usize, EncodeError> {
160        self.storage.post_update_header(&self.header)
161    }
162}
163
164/// Basic statistics for the unified logger.
165/// Note: the total_allocated_space might grow for the std implementation
166pub struct UnifiedLogStatus {
167    pub total_used_space: usize,
168    pub total_allocated_space: usize,
169}
170
171/// The writing interface to the unified logger.
172/// Writing is "almost" linear as various streams can allocate sections and track them until
173/// they drop them.
174pub trait UnifiedLogWrite<S: SectionStorage>: Send + Sync {
175    /// A section is a contiguous chunk of memory that can be used to write data.
176    /// It can store various types of data as specified by the entry_type.
177    /// The requested_section_size is the size of the section to allocate.
178    /// It returns a handle to the section that can be used to write data until
179    /// it is flushed with flush_section, it is then considered unmutable.
180    fn add_section(
181        &mut self,
182        entry_type: UnifiedLogType,
183        requested_section_size: usize,
184    ) -> CuResult<SectionHandle<S>>;
185
186    /// Flush the given section to the underlying storage.
187    fn flush_section(&mut self, section: &mut SectionHandle<S>);
188
189    /// Returns the current status of the unified logger.
190    fn status(&self) -> UnifiedLogStatus;
191}
192
193/// Read back a unified log linearly.
194pub trait UnifiedLogRead {
195    /// Read through the unified logger until it reaches the UnifiedLogType given in datalogtype.
196    /// It will return the byte array of the section if found.
197    fn read_next_section_type(&mut self, datalogtype: UnifiedLogType) -> CuResult<Option<Vec<u8>>>;
198
199    /// Read through the next section entry regardless of its type.
200    /// It will return the header and the byte array of the section.
201    /// Note the last Entry should be of UnifiedLogType::LastEntry if the log is not corrupted.
202    fn raw_read_section(&mut self) -> CuResult<(SectionHeader, Vec<u8>)>;
203}
204
205/// Create a new stream to write to the unifiedlogger.
206pub fn stream_write<E: Encode, S: SectionStorage>(
207    logger: Arc<Mutex<impl UnifiedLogWrite<S>>>,
208    entry_type: UnifiedLogType,
209    minimum_allocation_amount: usize,
210) -> CuResult<impl WriteStream<E>> {
211    LogStream::new(entry_type, logger, minimum_allocation_amount)
212}
213
214/// A wrapper around the unifiedlogger that implements the Write trait.
215struct LogStream<S: SectionStorage, L: UnifiedLogWrite<S>> {
216    entry_type: UnifiedLogType,
217    parent_logger: Arc<Mutex<L>>,
218    current_section: SectionHandle<S>,
219    current_position: usize,
220    minimum_allocation_amount: usize,
221}
222
223impl<S: SectionStorage, L: UnifiedLogWrite<S>> LogStream<S, L> {
224    fn new(
225        entry_type: UnifiedLogType,
226        parent_logger: Arc<Mutex<L>>,
227        minimum_allocation_amount: usize,
228    ) -> CuResult<Self> {
229        #[cfg(feature = "std")]
230        let section = parent_logger
231            .lock()
232            .expect("Could not lock a section at MmapStream creation")
233            .add_section(entry_type, minimum_allocation_amount)?;
234
235        #[cfg(not(feature = "std"))]
236        let section = parent_logger
237            .lock()
238            .add_section(entry_type, minimum_allocation_amount)?;
239
240        Ok(Self {
241            entry_type,
242            parent_logger,
243            current_section: section,
244            current_position: 0,
245            minimum_allocation_amount,
246        })
247    }
248}
249
250impl<S: SectionStorage, L: UnifiedLogWrite<S>> Debug for LogStream<S, L> {
251    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
252        write!(f, "MmapStream {{ entry_type: {:?}, current_position: {}, minimum_allocation_amount: {} }}", self.entry_type, self.current_position, self.minimum_allocation_amount)
253    }
254}
255
256impl<E: Encode, S: SectionStorage, L: UnifiedLogWrite<S>> WriteStream<E> for LogStream<S, L> {
257    fn log(&mut self, obj: &E) -> CuResult<()> {
258        //let dst = self.current_section.get_user_buffer();
259        // let result = encode_into_slice(obj, dst, standard());
260        let result = self.current_section.append(obj);
261        match result {
262            Ok(nb_bytes) => {
263                self.current_position += nb_bytes;
264                self.current_section.header.used += nb_bytes as u32;
265                Ok(())
266            }
267            Err(e) => match e {
268                EncodeError::UnexpectedEnd => {
269                    #[cfg(feature = "std")]
270                    let logger_guard = self.parent_logger.lock();
271
272                    #[cfg(not(feature = "std"))]
273                    let mut logger_guard = self.parent_logger.lock();
274
275                    #[cfg(feature = "std")]
276                    let mut logger_guard =
277                        match logger_guard {
278                            Ok(g) => g,
279                            Err(_) => return Err(
280                                "Logger mutex poisoned while reporting EncodeError::UnexpectedEnd"
281                                    .into(),
282                            ), // It will retry but at least not completely crash.
283                        };
284
285                    logger_guard.flush_section(&mut self.current_section);
286                    self.current_section = logger_guard
287                        .add_section(self.entry_type, self.minimum_allocation_amount)?;
288
289                    let result = self.current_section.append(obj).expect(
290                        "Failed to encode object in a newly minted section. Unrecoverable failure.",
291                    ); // If we fail just after creating a section, there is not much we can do, we need to bail.
292
293                    self.current_position += result;
294                    self.current_section.header.used += result as u32;
295                    Ok(())
296                }
297                _ => {
298                    let err =
299                        <&str as Into<CuError>>::into("Unexpected error while encoding object.")
300                            .add_cause(e.to_string().as_str());
301                    Err(err)
302                }
303            },
304        }
305    }
306}
307
308impl<S: SectionStorage, L: UnifiedLogWrite<S>> Drop for LogStream<S, L> {
309    fn drop(&mut self) {
310        #[cfg(feature = "std")]
311        let logger_guard = self.parent_logger.lock();
312
313        #[cfg(not(feature = "std"))]
314        let mut logger_guard = self.parent_logger.lock();
315
316        #[cfg(feature = "std")]
317        let mut logger_guard = match logger_guard {
318            Ok(g) => g,
319            Err(_) => return,
320        };
321        logger_guard.flush_section(&mut self.current_section);
322        #[cfg(feature = "std")]
323        if !std::thread::panicking() {
324            eprintln!("⚠️ MmapStream::drop: logger mutex poisoned");
325        }
326    }
327}
cu29_unifiedlog/lib.rs

cu29_unifiedlog/
lib.rs
