dspk_view/pyramid/
storage.rs

1//! The bytes behind a pyramid: an anonymous mapping (in memory; the OS commits only the pages
2//! that are filled) or a file next to the recording (kept between sessions).
3//!
4//! A file is `<dir>/cache/<file name>/<source>.minmax` (see [`pyramid_path`]): a
5//! [`HEADER_BYTES`] header (magic, version, a complete flag, the [`PyramidIdentity`] and `base`)
6//! then the levels. It is valid only when complete and built from the same identity.
7
8use std::fs::{self, OpenOptions};
9use std::mem::ManuallyDrop;
10use std::path::{Path, PathBuf};
11use std::time::UNIX_EPOCH;
12
13use dspk_core::{DspError, DspResult, RecordingSource};
14use memmap2::{MmapMut, MmapRaw};
15
16const MAGIC: &[u8; 8] = b"DKMINMAX";
17const VERSION: u32 = 1;
18/// Byte offset of the complete flag in the header.
19const COMPLETE_AT: usize = 12;
20/// Header bytes of a file; level data starts page-aligned after it.
21pub const HEADER_BYTES: usize = 4096;
22const MAX_ID_BYTES: usize = 1024;
23
24/// What a pyramid file was built from; a mismatch means the recording changed and the file is
25/// rebuilt.
26#[derive(Debug, Clone, PartialEq)]
27pub struct PyramidIdentity {
28    /// Source id within the recording (several signals can share one file).
29    pub source_id: String,
30    /// Size in bytes of the recording file (0 for folders).
31    pub len: u64,
32    /// Modification time of the recording (nanoseconds since the Unix epoch).
33    pub mtime_ns: u64,
34    pub channels: usize,
35    pub samples: u64,
36    pub sample_rate_hz: f64,
37}
38
39impl PyramidIdentity {
40    /// Identity of source `source_id` of the recording at `path` with `source`'s shape. For a
41    /// folder store the modification time of its `zarr.json` is used when present.
42    pub fn of(path: &Path, source_id: &str, source: &dyn RecordingSource) -> DspResult<Self> {
43        let stamp = if path.is_dir() && path.join("zarr.json").exists() { path.join("zarr.json") } else { path.to_path_buf() };
44        let meta = fs::metadata(&stamp).map_err(|e| DspError::Io(format!("{}: {e}", stamp.display())))?;
45        let mtime_ns = meta.modified().ok().and_then(|t| t.duration_since(UNIX_EPOCH).ok()).map_or(0, |d| d.as_nanos() as u64);
46        let info = source.info();
47        Ok(Self {
48            source_id: source_id.to_string(),
49            len: if path.is_dir() { 0 } else { meta.len() },
50            mtime_ns,
51            channels: info.channel_count(),
52            samples: info.samples,
53            sample_rate_hz: info.sample_rate_hz(),
54        })
55    }
56
57    /// Header with the complete flag cleared: magic, version, flag, then identity and `base`.
58    fn header(&self, base: u64) -> Vec<u8> {
59        let mut h = Vec::with_capacity(HEADER_BYTES);
60        h.extend_from_slice(MAGIC);
61        h.extend_from_slice(&VERSION.to_le_bytes());
62        h.extend_from_slice(&0u32.to_le_bytes());
63        for v in [self.channels as u64, self.samples, base, self.len, self.mtime_ns, self.sample_rate_hz.to_bits()] {
64            h.extend_from_slice(&v.to_le_bytes());
65        }
66        h.extend_from_slice(&(self.source_id.len() as u32).to_le_bytes());
67        h.extend_from_slice(self.source_id.as_bytes());
68        h.resize(HEADER_BYTES, 0);
69        h
70    }
71}
72
73/// `<dir>/cache/<file name>/<source id>.minmax` for the recording at `path`.
74pub fn pyramid_path(path: &Path, source_id: &str) -> PathBuf {
75    let dir = path.parent().unwrap_or(Path::new("."));
76    let name = path.file_name().map_or_else(|| "recording".into(), |n| n.to_string_lossy().into_owned());
77    let id: String = source_id.chars().map(|c| if c.is_ascii_alphanumeric() || c == '-' || c == '_' { c } else { '_' }).collect();
78    let id = if id.is_empty() { "main".to_string() } else { id };
79    dir.join("cache").join(name).join(format!("{id}.minmax"))
80}
81
82pub(crate) struct Storage {
83    map: ManuallyDrop<MmapRaw>,
84    /// The file, for file-backed storage.
85    path: Option<PathBuf>,
86}
87
88impl Storage {
89    /// `bytes` of zeroed anonymous memory.
90    pub fn memory(bytes: usize) -> DspResult<Self> {
91        let map = MmapMut::map_anon(bytes.max(1))?;
92        Ok(Self { map: ManuallyDrop::new(MmapRaw::from(map)), path: None })
93    }
94
95    /// The complete file at `path` built from `identity` with level-0 bucket `base` and `bytes`
96    /// long; `None` when it is missing, incomplete, of another version or built from something
97    /// else.
98    pub fn open_file(path: &Path, identity: &PyramidIdentity, base: u64, bytes: usize) -> DspResult<Option<Self>> {
99        let Ok(file) = OpenOptions::new().read(true).write(true).open(path) else { return Ok(None) };
100        if file.metadata().map(|m| m.len()).unwrap_or(0) != bytes as u64 {
101            return Ok(None);
102        }
103        let map = MmapRaw::map_raw(&file)?;
104        // SAFETY: the mapping is `bytes ≥ HEADER_BYTES` long (checked above) and not written yet.
105        let header = unsafe { std::slice::from_raw_parts(map.as_ptr(), HEADER_BYTES) };
106        let expected = identity.header(base);
107        let complete = u32::from_le_bytes(header[COMPLETE_AT..COMPLETE_AT + 4].try_into().expect("4 bytes")) == 1;
108        if header[..COMPLETE_AT] != expected[..COMPLETE_AT] || !complete || header[COMPLETE_AT + 4..] != expected[COMPLETE_AT + 4..] {
109            return Ok(None);
110        }
111        Ok(Some(Self { map: ManuallyDrop::new(map), path: Some(path.to_path_buf()) }))
112    }
113
114    /// A new, empty file at `path` (replacing any other), `bytes` long, headed by `identity`.
115    pub fn create_file(path: &Path, identity: &PyramidIdentity, base: u64, bytes: usize) -> DspResult<Self> {
116        if identity.source_id.len() > MAX_ID_BYTES {
117            return Err(DspError::InvalidConfig(format!("source id longer than {MAX_ID_BYTES} bytes")));
118        }
119        if let Some(dir) = path.parent() {
120            fs::create_dir_all(dir).map_err(|e| DspError::Io(format!("{}: {e}", dir.display())))?;
121        }
122        let file = OpenOptions::new().read(true).write(true).create(true).truncate(true).open(path)
123            .map_err(|e| DspError::Io(format!("{}: {e}", path.display())))?;
124        file.set_len(bytes as u64)?;
125        let map = MmapRaw::map_raw(&file)?;
126        let header = identity.header(base);
127        // SAFETY: the mapping is `bytes ≥ HEADER_BYTES` long and nothing else uses it yet.
128        unsafe { std::ptr::copy_nonoverlapping(header.as_ptr(), map.as_mut_ptr(), HEADER_BYTES) };
129        Ok(Self { map: ManuallyDrop::new(map), path: Some(path.to_path_buf()) })
130    }
131
132    pub fn path(&self) -> Option<&Path> {
133        self.path.as_deref()
134    }
135
136    pub fn ptr(&self) -> *mut u8 {
137        self.map.as_mut_ptr()
138    }
139
140    /// Flushes a complete file and only then sets its complete flag (nothing for memory).
141    pub fn mark_complete(&self) -> DspResult<()> {
142        if self.path.is_none() {
143            return Ok(());
144        }
145        self.map.flush()?;
146        // SAFETY: the flag lies in the header, which only this call writes after creation.
147        unsafe { self.map.as_mut_ptr().add(COMPLETE_AT).cast::<[u8; 4]>().write_unaligned(1u32.to_le_bytes()) };
148        self.map.flush_range(0, HEADER_BYTES)?;
149        Ok(())
150    }
151}
152
153impl Drop for Storage {
154    fn drop(&mut self) {
155        // SAFETY: not used after this point.
156        unsafe { ManuallyDrop::drop(&mut self.map) };
157    }
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163
164    #[test]
165    fn pyramid_path_sits_next_to_the_recording() {
166        assert_eq!(pyramid_path(Path::new("/data/a.ap.bin"), "ElectricalSeries/x"), PathBuf::from("/data/cache/a.ap.bin/ElectricalSeries_x.minmax"));
167    }
168}
dspk_view/pyramid/storage.rs

dspk_view/pyramid/
storage.rs
