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

vfsi_core/api/
read.rs
