vfsi_core/api/
read.rs

1use super::support::FsReadResult as OwnedReadResult;
2use super::*;
3use std::path::Path;
4
5/// One consuming read operation. Construction performs no I/O or allocation.
6/// Caller buffers are borrowed only until `Vfsi::vread` returns, including on errors.
7pub struct ReadOp<'a, H: crate::api::FileHandle + 'a> {
8    source: OpSource<'a, H>,
9}
10enum OpSource<'a, H: crate::api::FileHandle + 'a> {
11    Whole(&'a Path),
12    Range(&'a H, u64, usize),
13    Into(&'a H, u64, &'a mut [u8]),
14}
15impl<H: crate::api::FileHandle> std::fmt::Debug for ReadOp<'_, H> {
16    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
17        match &self.source {
18            OpSource::Whole(path) => formatter.debug_tuple("ReadOp::whole").field(path).finish(),
19            OpSource::Range(_, _, length) => formatter
20                .debug_struct("ReadOp::range")
21                .field("length", length)
22                .finish_non_exhaustive(),
23            OpSource::Into(_, _, buffer) => formatter
24                .debug_struct("ReadOp::into")
25                .field("length", &buffer.len())
26                .finish_non_exhaustive(),
27        }
28    }
29}
30impl<'a, H: crate::api::FileHandle + 'a> From<&'a str> for ReadOp<'a, H> {
31    fn from(path: &'a str) -> Self {
32        Self::whole(path)
33    }
34}
35impl<'a, H: crate::api::FileHandle + 'a> From<&'a Path> for ReadOp<'a, H> {
36    fn from(path: &'a Path) -> Self {
37        Self::whole(path)
38    }
39}
40impl<'a, H: crate::api::FileHandle + 'a> ReadOp<'a, H> {
41    /// Collect a complete file; fail instead of truncating at the byte budget.
42    pub fn whole<P: AsRef<Path> + ?Sized>(path: &'a P) -> Self {
43        Self {
44            source: OpSource::Whole(path.as_ref()),
45        }
46    }
47    /// Allocate a positional range. Short progress and EOF are reported explicitly.
48    pub fn range(file: &'a H, offset: u64, length: usize) -> Self {
49        Self {
50            source: OpSource::Range(file, offset, length),
51        }
52    }
53    /// Fill caller storage at an absolute offset, without changing the cursor.
54    /// The request length is exactly `buffer.len()`.
55    pub fn into(file: &'a H, offset: u64, buffer: &'a mut [u8]) -> Self {
56        Self {
57            source: OpSource::Into(file, offset, buffer),
58        }
59    }
60    /// Inspect a complete-file path from an external `Vfsi` implementation.
61    pub fn whole_file_path(&self) -> Option<&'a Path> {
62        match &self.source {
63            OpSource::Whole(path) => Some(*path),
64            _ => None,
65        }
66    }
67    /// Inspect the handle, absolute offset, and length of an allocating range.
68    pub fn range_parts(&self) -> Option<(&'a H, u64, usize)> {
69        match &self.source {
70            OpSource::Range(file, offset, length) => Some((*file, *offset, *length)),
71            _ => None,
72        }
73    }
74    /// Borrow the handle, absolute offset, and caller storage for direct dispatch.
75    pub fn buffer_parts_mut(&mut self) -> Option<(&'a H, u64, &mut [u8])> {
76        match &mut self.source {
77            OpSource::Into(file, offset, buffer) => Some((*file, *offset, buffer)),
78            _ => None,
79        }
80    }
81}
82
83/// Progress for one operation, with no borrowed references to caller storage.
84#[derive(Debug, Clone, PartialEq, Eq)]
85pub struct ReadResult {
86    /// Resolved absolute offset; zero for a complete-file operation.
87    pub(crate) offset: u64,
88    /// Number of valid bytes in owned data or the caller's buffer.
89    pub(crate) read: usize,
90    /// True when the backend observed end of file.
91    pub(crate) eof: bool,
92    /// Owned contents for `whole`/`range`; `None` for `into`, including empty reads.
93    pub(crate) data: Option<Vec<u8>>,
94}
95
96impl ReadResult {
97    /// Owned contents. The byte count is derived, never independently supplied.
98    pub fn owned(offset: u64, data: Vec<u8>, eof: bool) -> Self {
99        Self {
100            offset,
101            read: data.len(),
102            eof,
103            data: Some(data),
104        }
105    }
106    /// Progress in caller storage, for implementing `Vfsi`. The implementation
107    /// must ensure `read` does not exceed the corresponding destination length.
108    pub const fn buffered(offset: u64, read: usize, eof: bool) -> Self {
109        Self {
110            offset,
111            read,
112            eof,
113            data: None,
114        }
115    }
116    pub const fn offset(&self) -> u64 {
117        self.offset
118    }
119    pub const fn read(&self) -> usize {
120        self.read
121    }
122    pub const fn eof(&self) -> bool {
123        self.eof
124    }
125    pub fn data(&self) -> Option<&[u8]> {
126        self.data.as_deref()
127    }
128    pub fn into_data(self) -> Option<Vec<u8>> {
129        self.data
130    }
131    pub const fn is_buffered(&self) -> bool {
132        self.data.is_none()
133    }
134}
135
136/// Aggregate logical byte budget, defaulting to the client's policy (16 MiB).
137/// Range request lengths must fit even if EOF returns fewer bytes. Complete
138/// files share the remaining returned-data budget and fail rather than truncate.
139#[derive(Debug, Clone, Copy, PartialEq, Eq)]
140pub struct ReadOptions {
141    max_total_bytes: Option<std::num::NonZeroUsize>,
142}
143impl ReadOptions {
144    /// Inherit the client's aggregate read budget.
145    pub const fn new() -> Self {
146        Self {
147            max_total_bytes: None,
148        }
149    }
150    /// Override the aggregate budget for this call; `None` restores inheritance.
151    /// Only nonzero overrides are representable. Stream larger files instead
152    /// of raising this when bounded memory is required.
153    pub const fn max_total_bytes(mut self, bytes: Option<std::num::NonZeroUsize>) -> Self {
154        self.max_total_bytes = bytes;
155        self
156    }
157    /// None inherits the client's configured read budget.
158    pub const fn total_byte_limit(self) -> Option<std::num::NonZeroUsize> {
159        self.max_total_bytes
160    }
161    /// Resolve an inherited aggregate budget against the client's default.
162    pub fn limit_or(self, default: usize) -> usize {
163        self.max_total_bytes
164            .map_or(default, std::num::NonZeroUsize::get)
165    }
166}
167
168impl Default for ReadOptions {
169    fn default() -> Self {
170        Self::new()
171    }
172}
173pub fn consume_ops<'a, H: crate::api::FileHandle + 'a, R, B>(
174    ops: impl IntoIterator<Item = ReadOp<'a, H>>,
175    budget: usize,
176    range: impl Fn(&'a H, u64, usize) -> R,
177    into: impl Fn(&'a H, u64, &'a mut [u8]) -> B,
178    owned: impl FnOnce(&[ReadRequest<'a, R>], usize) -> Result<Vec<OwnedReadResult>>,
179    buffers: impl FnOnce(&mut [B], usize) -> Result<Vec<crate::api::ReadIntoResult>>,
180) -> Result<Vec<ReadResult>> {
181    let mut owned_ops = Vec::new();
182    let mut owned_indices = Vec::new();
183    let mut buffer_ops = Vec::new();
184    let mut buffer_indices = Vec::new();
185    let mut requested = 0usize;
186    let mut count = 0usize;
187    for (index, op) in ops.into_iter().enumerate() {
188        count += 1;
189        let length = match &op.source {
190            OpSource::Whole(_) => 0,
191            OpSource::Range(_, _, length) => *length,
192            OpSource::Into(_, _, buffer) => buffer.len(),
193        };
194        requested = requested
195            .checked_add(length)
196            .filter(|bytes| *bytes <= budget)
197            .ok_or_else(|| Error::client(index, libc::EFBIG as u32))?;
198        match op.source {
199            OpSource::Whole(path) => {
200                owned_ops.push(ReadRequest::whole_file(path));
201                owned_indices.push(index);
202            }
203            OpSource::Range(file, offset, length) => {
204                owned_ops.push(ReadRequest::range(range(file, offset, length)));
205                owned_indices.push(index);
206            }
207            OpSource::Into(file, offset, buffer) => {
208                buffer_ops.push(into(file, offset, buffer));
209                buffer_indices.push(index);
210            }
211        }
212    }
213    let remap = |error: Error, indices: &[usize]| match error.index() {
214        Some(index) if index < indices.len() => error.with_index(indices[index]),
215        Some(_) => Error::transport(None, "invalid vread_native backend error index"),
216        None => error,
217    };
218    let buffered = if buffer_ops.is_empty() {
219        Vec::new()
220    } else {
221        buffers(&mut buffer_ops, budget).map_err(|e| remap(e, &buffer_indices))?
222    };
223    drop(buffer_ops);
224    if buffered.len() != buffer_indices.len() {
225        return Err(Error::transport(
226            None,
227            "invalid vread_native buffer result count",
228        ));
229    }
230    let mut remaining = budget;
231    for result in &buffered {
232        remaining = remaining
233            .checked_sub(result.read)
234            .ok_or_else(|| Error::transport(None, "vread_native backend exceeded byte budget"))?;
235    }
236    let allocated = if owned_ops.is_empty() {
237        Vec::new()
238    } else {
239        owned(&owned_ops, remaining).map_err(|e| remap(e, &owned_indices))?
240    };
241    if allocated.len() != owned_indices.len() {
242        return Err(Error::transport(
243            None,
244            "invalid vread_native owned result count",
245        ));
246    }
247    let mut output: Vec<Option<ReadResult>> = (0..count).map(|_| None).collect();
248    for (index, result) in buffer_indices.into_iter().zip(buffered) {
249        output[index] = Some(ReadResult {
250            offset: result.offset,
251            read: result.read,
252            eof: result.eof,
253            data: None,
254        });
255    }
256    for (index, result) in owned_indices.into_iter().zip(allocated) {
257        remaining = remaining
258            .checked_sub(result.data.len())
259            .ok_or_else(|| Error::transport(None, "vread_native backend exceeded byte budget"))?;
260        output[index] = Some(ReadResult {
261            offset: result.offset,
262            read: result.data.len(),
263            eof: result.eof,
264            data: Some(result.data),
265        });
266    }
267    output
268        .into_iter()
269        .map(|result| result.ok_or_else(|| Error::transport(None, "vread_native omitted a result")))
270        .collect()
271}
272/// Internal projection for the native owned-result batch path.
273pub struct ReadRequest<'a, R> {
274    pub(crate) source: Source<'a, R>,
275}
276pub(crate) enum Source<'a, R> {
277    Whole(&'a Path),
278    Range(R),
279}
280impl<'a, R> ReadRequest<'a, R> {
281    /// Read a complete path within the aggregate budget, without manual OPEN/CLOSE.
282    pub fn whole_file<P: AsRef<Path> + ?Sized>(path: &'a P) -> Self {
283        Self {
284            source: Source::Whole(path.as_ref()),
285        }
286    }
287    /// Wrap a positional request, preserving short results and EOF.
288    pub fn range(request: R) -> Self {
289        Self {
290            source: Source::Range(request),
291        }
292    }
293    /// Borrow the backend-defined range request, or return `None` for a whole file.
294    /// External [`crate::api::Vfsi`] implementations can use this together with
295    /// a whole-file dispatch path without exposing handles.
296    pub fn range_ref(&self) -> Option<&R> {
297        match &self.source {
298            Source::Range(request) => Some(request),
299            _ => None,
300        }
301    }
302}
303impl<'a, R> From<R> for ReadRequest<'a, R> {
304    fn from(request: R) -> Self {
305        Self::range(request)
306    }
307}
308pub fn read_batch<R>(
309    requests: &[ReadRequest<'_, R>],
310    budget: usize,
311    read_ranges: impl FnOnce(&[&R], usize) -> Result<Vec<OwnedReadResult>>,
312    read_paths: impl FnOnce(&[&Path], usize) -> Result<Vec<Vec<u8>>>,
313) -> Result<Vec<OwnedReadResult>> {
314    if requests.is_empty() {
315        return Ok(Vec::new());
316    }
317    let (mut paths, mut path_indices, mut ranges, mut range_indices) =
318        (Vec::new(), Vec::new(), Vec::new(), Vec::new());
319    for (i, r) in requests.iter().enumerate() {
320        match &r.source {
321            Source::Whole(path) => {
322                paths.push(*path);
323                path_indices.push(i);
324            }
325            Source::Range(range) => {
326                ranges.push(range);
327                range_indices.push(i);
328            }
329        }
330    }
331    let remap = |e: Error, indices: &[usize]| match e.index() {
332        Some(i) if i < indices.len() => e.with_index(indices[i]),
333        Some(_) => Error::transport(None, "invalid vread_native backend error index"),
334        None => e,
335    };
336    // Group ranges before complete paths. Reads do not promise a snapshot.
337    let range_results = if ranges.is_empty() {
338        Vec::new()
339    } else {
340        read_ranges(&ranges, budget).map_err(|e| remap(e, &range_indices))?
341    };
342    if range_results.len() != ranges.len() {
343        return Err(Error::transport(
344            None,
345            "invalid vread_native range result count",
346        ));
347    }
348    let mut remaining = budget;
349    for r in &range_results {
350        remaining = remaining
351            .checked_sub(r.data.len())
352            .ok_or_else(|| Error::transport(None, "vread_native backend exceeded byte budget"))?;
353    }
354    let path_results = if paths.is_empty() {
355        Vec::new()
356    } else {
357        read_paths(&paths, remaining).map_err(|e| remap(e, &path_indices))?
358    };
359    if path_results.len() != paths.len() {
360        return Err(Error::transport(
361            None,
362            "invalid vread_native path result count",
363        ));
364    }
365    for r in &path_results {
366        remaining = remaining
367            .checked_sub(r.len())
368            .ok_or_else(|| Error::transport(None, "vread_native backend exceeded byte budget"))?;
369    }
370    let mut output: Vec<Option<OwnedReadResult>> = (0..requests.len()).map(|_| None).collect();
371    for (i, r) in range_indices.into_iter().zip(range_results) {
372        output[i] = Some(r);
373    }
374    for (i, data) in path_indices.into_iter().zip(path_results) {
375        output[i] = Some(OwnedReadResult {
376            offset: 0,
377            data,
378            eof: true,
379        });
380    }
381    output
382        .into_iter()
383        .map(|r| r.ok_or_else(|| Error::transport(None, "vread_native omitted a result")))
384        .collect()
385}
386
387#[cfg(test)]
388mod option_layout_tests {
389    use super::ReadOptions;
390    use std::num::NonZeroUsize;
391    #[test]
392    fn read_options_are_one_word_and_keep_inheritance_distinct() {
393        assert_eq!(
394            std::mem::size_of::<ReadOptions>(),
395            std::mem::size_of::<usize>()
396        );
397        let mut options = ReadOptions::new();
398        assert_eq!(options.limit_or(7), 7);
399        for bytes in [1, 42, usize::MAX] {
400            let limit = NonZeroUsize::new(bytes);
401            options = options.max_total_bytes(limit);
402            assert_eq!(options.total_byte_limit(), limit);
403            assert_eq!(options.limit_or(7), bytes);
404        }
405        options = options.max_total_bytes(None);
406        assert_eq!(options.total_byte_limit(), None);
407        assert_eq!(options.limit_or(7), 7);
408        assert_eq!(
409            options.limit_or(0),
410            0,
411            "an inherited zero policy must not become unlimited"
412        );
413    }
414}
vfsi_core/api/read.rs

vfsi_core/api/
read.rs
