harmos_storage/
file_batch.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Native File Batches Share Their Original Allocation Through Every View.
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::borrow::Borrow;
20use std::num::NonZeroUsize;
21use std::ops::Range;
22use std::sync::Arc;
23
24// Third Party Imports
25
26// Local Imports
27use crate::{Error, FileBatch, MemorySize, ReadBatch, Result};
28use harmos_domain::model::storage::files::FileEvent;
29
30/* === Constants === */
31
32/* === Macros === */
33
34/* === Type Aliases === */
35
36/* === Structs === */
37
38#[derive(Clone)]
39pub struct FileReadBatch {
40    owner: ReadBatch<FileBatch>,
41}
42
43#[derive(Clone)]
44pub struct FileReadSlice {
45    owner: FileReadBatch,
46    range: Range<usize>,
47}
48
49pub type FileReadEvent = ReadBatch<FileEvent>;
50
51/* === Enums === */
52
53/* === Traits === */
54
55/* === Implementations === */
56
57impl FileReadEvent {
58    pub fn event(&self) -> &FileEvent {
59        self.as_ref()
60    }
61
62    pub fn data(&self) -> Option<&[u8]> {
63        match self.event() {
64            FileEvent::Data(batch) => Some(&batch.bytes),
65            _ => None,
66        }
67    }
68
69    pub fn offset(&self) -> Option<u64> {
70        match self.event() {
71            FileEvent::Data(batch) => Some(batch.offset),
72            _ => None,
73        }
74    }
75}
76
77impl FileReadBatch {
78    pub(crate) fn new(batch: FileBatch) -> Self {
79        Self { owner: ReadBatch::new(batch) }
80    }
81
82    /// Moves a unique file batch through a fallible projection.
83    /// Shared batches or slices refuse, as with `ReadBatch::try_map`.
84    #[tracing::instrument(name = "storage.file_batch.try_map", level = "trace", skip_all, fields(outcome = "cancelled"))]
85    pub fn try_map<U, E>(self, map: impl FnOnce(FileBatch) -> std::result::Result<U, E>) -> std::result::Result<ReadBatch<U>, E>
86    where
87        E: From<Error>,
88    {
89        let result: std::result::Result<ReadBatch<U>, E> = self.owner.try_map(map);
90        tracing::Span::current().record("outcome", if result.is_ok() { "completed" } else { "failed" });
91        result
92    }
93
94    pub fn offset(&self) -> u64 {
95        self.owner.offset
96    }
97
98    pub fn bytes(&self) -> &[u8] {
99        &self.owner.bytes
100    }
101
102    pub fn owner(&self) -> Arc<impl Borrow<FileBatch> + Send + Sync + 'static> {
103        self.owner.owner()
104    }
105
106    pub fn slice(&self, range: Range<usize>) -> FileReadSlice {
107        assert!(range.start <= range.end && range.end <= self.bytes().len(), "file slice is outside its batch");
108        FileReadSlice { owner: self.clone(), range }
109    }
110}
111
112impl Borrow<FileBatch> for FileReadBatch {
113    fn borrow(&self) -> &FileBatch {
114        self.owner.as_ref()
115    }
116}
117
118impl FileReadSlice {
119    pub fn offset(&self) -> u64 {
120        self.owner.offset() + self.range.start as u64
121    }
122
123    pub fn bytes(&self) -> &[u8] {
124        &self.owner.bytes()[self.range.clone()]
125    }
126}
127
128impl MemorySize for FileReadBatch {
129    fn memory_size(&self) -> usize {
130        self.owner.memory_size().saturating_add(std::mem::size_of::<Self>())
131    }
132}
133
134impl std::fmt::Debug for FileReadBatch {
135    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
136        self.owner.fmt(formatter)
137    }
138}
139
140/* === Public Functions === */
141
142pub(crate) fn batch_limit(ceiling: usize, offset: u64, remaining: u64, event: bool) -> Result<NonZeroUsize> {
143    let mut count = ceiling.min(usize::try_from(remaining).unwrap_or(usize::MAX));
144    loop {
145        let batch = 2usize.saturating_add(varint(offset)).saturating_add(varint(count as u64)).saturating_add(count);
146        let framed = if event { batch.saturating_add(1 + varint(batch as u64)) } else { batch };
147        if framed <= ceiling {
148            return NonZeroUsize::new(count).ok_or(Error::ResourceExhausted { resource: "file response batch bytes" });
149        }
150        count = count
151            .checked_sub(framed - ceiling)
152            .filter(|count| *count != 0)
153            .ok_or(Error::ResourceExhausted { resource: "file response batch bytes" })?;
154    }
155}
156
157/* === Private Functions === */
158
159fn varint(mut value: u64) -> usize {
160    let mut bytes = 1;
161    while value >= 128 {
162        value >>= 7;
163        bytes += 1;
164    }
165    bytes
166}
167
168/* === Tests === */
harmos_storage/file_batch.rs

harmos_storage/
file_batch.rs
