use super::support::FsReadResult as OwnedReadResult;
use super::*;
use std::path::Path;
#[must_use = "configuration and requests have no effect until used"]
pub struct ReadOp<'a, H: crate::api::FileHandle + 'a> {
source: OpSource<'a, H>,
}
enum OpSource<'a, H: crate::api::FileHandle + 'a> {
Whole(&'a Path),
Range(&'a H, u64, usize),
Into(&'a H, u64, &'a mut [u8]),
}
impl<H: crate::api::FileHandle> std::fmt::Debug for ReadOp<'_, H> {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match &self.source {
OpSource::Whole(path) => formatter.debug_tuple("ReadOp::whole").field(path).finish(),
OpSource::Range(_, _, length) => formatter
.debug_struct("ReadOp::range")
.field("length", length)
.finish_non_exhaustive(),
OpSource::Into(_, _, buffer) => formatter
.debug_struct("ReadOp::into")
.field("length", &buffer.len())
.finish_non_exhaustive(),
}
}
}
impl<'a, H: crate::api::FileHandle + 'a> From<&'a str> for ReadOp<'a, H> {
fn from(path: &'a str) -> Self {
Self::whole(path)
}
}
impl<'a, H: crate::api::FileHandle + 'a> From<&'a Path> for ReadOp<'a, H> {
fn from(path: &'a Path) -> Self {
Self::whole(path)
}
}
impl<'a, H: crate::api::FileHandle + 'a> ReadOp<'a, H> {
pub fn whole<P: AsRef<Path> + ?Sized>(path: &'a P) -> Self {
Self {
source: OpSource::Whole(path.as_ref()),
}
}
pub fn range(file: &'a H, offset: u64, length: usize) -> Self {
Self {
source: OpSource::Range(file, offset, length),
}
}
pub fn into(file: &'a H, offset: u64, buffer: &'a mut [u8]) -> Self {
Self {
source: OpSource::Into(file, offset, buffer),
}
}
pub fn whole_file_path(&self) -> Option<&'a Path> {
match &self.source {
OpSource::Whole(path) => Some(*path),
_ => None,
}
}
pub fn range_parts(&self) -> Option<(&'a H, u64, usize)> {
match &self.source {
OpSource::Range(file, offset, length) => Some((*file, *offset, *length)),
_ => None,
}
}
pub fn buffer_parts_mut(&mut self) -> Option<(&'a H, u64, &mut [u8])> {
match &mut self.source {
OpSource::Into(file, offset, buffer) => Some((*file, *offset, buffer)),
_ => None,
}
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct ReadResult {
offset: u64,
eof: bool,
payload: ReadPayload,
}
#[derive(Clone, PartialEq, Eq)]
enum ReadPayload {
Owned(Vec<u8>),
Buffered(usize),
}
impl std::fmt::Debug for ReadResult {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ReadResult")
.field("offset", &self.offset())
.field("bytes_read", &self.bytes_read())
.field("eof", &self.eof())
.field("data", &self.data())
.finish()
}
}
impl ReadResult {
pub fn owned(offset: u64, data: Vec<u8>, eof: bool) -> Self {
Self {
offset,
eof,
payload: ReadPayload::Owned(data),
}
}
pub const fn buffered(offset: u64, bytes_read: usize, eof: bool) -> Self {
Self {
offset,
eof,
payload: ReadPayload::Buffered(bytes_read),
}
}
pub const fn offset(&self) -> u64 {
self.offset
}
pub const fn bytes_read(&self) -> usize {
match &self.payload {
ReadPayload::Owned(data) => data.len(),
ReadPayload::Buffered(count) => *count,
}
}
pub const fn eof(&self) -> bool {
self.eof
}
pub fn data(&self) -> Option<&[u8]> {
match &self.payload {
ReadPayload::Owned(data) => Some(data.as_slice()),
ReadPayload::Buffered(_) => None,
}
}
pub fn into_data(self) -> Option<Vec<u8>> {
match self.payload {
ReadPayload::Owned(data) => Some(data),
ReadPayload::Buffered(_) => None,
}
}
pub const fn is_buffered(&self) -> bool {
matches!(&self.payload, ReadPayload::Buffered(_))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[must_use = "configuration and requests have no effect until used"]
pub struct ReadOptions {
max_total_bytes: Option<std::num::NonZeroUsize>,
}
impl ReadOptions {
pub const fn new() -> Self {
Self {
max_total_bytes: None,
}
}
pub const fn with_max_total_bytes(mut self, bytes: std::num::NonZeroUsize) -> Self {
self.max_total_bytes = Some(bytes);
self
}
pub const fn max_total_bytes(self) -> Option<std::num::NonZeroUsize> {
self.max_total_bytes
}
pub fn max_total_bytes_or(self, default: usize) -> usize {
self.max_total_bytes
.map_or(default, std::num::NonZeroUsize::get)
}
}
impl Default for ReadOptions {
fn default() -> Self {
Self::new()
}
}
pub fn consume_ops<'a, H: crate::api::FileHandle + 'a, R, B>(
ops: impl IntoIterator<Item = ReadOp<'a, H>>,
budget: usize,
range: impl Fn(&'a H, u64, usize) -> R,
into: impl Fn(&'a H, u64, &'a mut [u8]) -> B,
owned: impl FnOnce(&[ReadRequest<'a, R>], usize) -> Result<Vec<OwnedReadResult>>,
buffers: impl FnOnce(&mut [B], usize) -> Result<Vec<crate::api::ReadIntoResult>>,
) -> Result<Vec<ReadResult>> {
let mut ops = ops.into_iter();
let Some(first) = ops.next() else {
return Ok(Vec::new());
};
let second = ops.next();
if second.is_none() {
let length = match &first.source {
OpSource::Whole(_) => 0,
OpSource::Range(_, _, length) => *length,
OpSource::Into(_, _, buffer) => buffer.len(),
};
if length > budget {
return Err(Error::client(0, libc::EFBIG as u32));
}
let remap = |error: Error| match error.index() {
Some(0) | None => error,
Some(_) => Error::transport(None, "invalid vread_native backend error index"),
};
let request = match first.source {
OpSource::Whole(path) => ReadRequest::whole_file(path),
OpSource::Range(file, offset, length) => {
ReadRequest::range(range(file, offset, length))
}
OpSource::Into(file, offset, buffer) => {
let values = buffers(&mut [into(file, offset, buffer)], budget).map_err(remap)?;
if values.len() != 1 {
return Err(Error::transport(
None,
"invalid vread_native buffer result count",
));
}
if values[0].bytes_read > budget {
return Err(Error::transport(
None,
"vread_native backend exceeded byte budget",
));
}
return Ok(values
.into_iter()
.map(|result| {
ReadResult::buffered(result.offset, result.bytes_read, result.eof)
})
.collect());
}
};
let values = owned(&[request], budget).map_err(remap)?;
if values.len() != 1 {
return Err(Error::transport(
None,
"invalid vread_native owned result count",
));
}
if values[0].data.len() > budget {
return Err(Error::transport(
None,
"vread_native backend exceeded byte budget",
));
}
return Ok(values
.into_iter()
.map(|result| ReadResult::owned(result.offset, result.data, result.eof))
.collect());
}
let ops = std::iter::once(first).chain(second).chain(ops);
let mut owned_ops = Vec::new();
let mut owned_indices = Vec::new();
let mut buffer_ops = Vec::new();
let mut buffer_indices = Vec::new();
let mut requested = 0usize;
let mut count = 0usize;
for (index, op) in ops.enumerate() {
count += 1;
let length = match &op.source {
OpSource::Whole(_) => 0,
OpSource::Range(_, _, length) => *length,
OpSource::Into(_, _, buffer) => buffer.len(),
};
requested = requested
.checked_add(length)
.filter(|bytes| *bytes <= budget)
.ok_or_else(|| Error::client(index, libc::EFBIG as u32))?;
match op.source {
OpSource::Whole(path) => {
owned_ops.push(ReadRequest::whole_file(path));
owned_indices.push(index);
}
OpSource::Range(file, offset, length) => {
owned_ops.push(ReadRequest::range(range(file, offset, length)));
owned_indices.push(index);
}
OpSource::Into(file, offset, buffer) => {
buffer_ops.push(into(file, offset, buffer));
buffer_indices.push(index);
}
}
}
let remap = |error: Error, indices: &[usize]| match error.index() {
Some(index) if index < indices.len() => error.with_index(indices[index]),
Some(_) => Error::transport(None, "invalid vread_native backend error index"),
None => error,
};
let buffered = if buffer_ops.is_empty() {
Vec::new()
} else {
buffers(&mut buffer_ops, budget).map_err(|e| remap(e, &buffer_indices))?
};
drop(buffer_ops);
if buffered.len() != buffer_indices.len() {
return Err(Error::transport(
None,
"invalid vread_native buffer result count",
));
}
let mut remaining = budget;
for result in &buffered {
remaining = remaining
.checked_sub(result.bytes_read)
.ok_or_else(|| Error::transport(None, "vread_native backend exceeded byte budget"))?;
}
let allocated = if owned_ops.is_empty() {
Vec::new()
} else {
owned(&owned_ops, remaining).map_err(|e| remap(e, &owned_indices))?
};
if allocated.len() != owned_indices.len() {
return Err(Error::transport(
None,
"invalid vread_native owned result count",
));
}
for result in &allocated {
remaining = remaining
.checked_sub(result.data.len())
.ok_or_else(|| Error::transport(None, "vread_native backend exceeded byte budget"))?;
}
if buffer_indices.is_empty() {
return Ok(allocated
.into_iter()
.map(|result| ReadResult::owned(result.offset, result.data, result.eof))
.collect());
}
if owned_indices.is_empty() {
return Ok(buffered
.into_iter()
.map(|result| ReadResult::buffered(result.offset, result.bytes_read, result.eof))
.collect());
}
let mut output: Vec<Option<ReadResult>> = (0..count).map(|_| None).collect();
for (index, result) in buffer_indices.into_iter().zip(buffered) {
output[index] = Some(ReadResult::buffered(
result.offset,
result.bytes_read,
result.eof,
));
}
for (index, result) in owned_indices.into_iter().zip(allocated) {
output[index] = Some(ReadResult::owned(result.offset, result.data, result.eof));
}
output
.into_iter()
.map(|result| result.ok_or_else(|| Error::transport(None, "vread_native omitted a result")))
.collect()
}
pub struct ReadRequest<'a, R> {
pub(crate) source: Source<'a, R>,
}
pub(crate) enum Source<'a, R> {
Whole(&'a Path),
Range(R),
}
impl<'a, R> ReadRequest<'a, R> {
pub fn whole_file<P: AsRef<Path> + ?Sized>(path: &'a P) -> Self {
Self {
source: Source::Whole(path.as_ref()),
}
}
pub fn range(request: R) -> Self {
Self {
source: Source::Range(request),
}
}
pub fn range_ref(&self) -> Option<&R> {
match &self.source {
Source::Range(request) => Some(request),
_ => None,
}
}
}
impl<'a, R> From<R> for ReadRequest<'a, R> {
fn from(request: R) -> Self {
Self::range(request)
}
}
pub fn read_batch<R>(
requests: &[ReadRequest<'_, R>],
budget: usize,
read_ranges: impl FnOnce(&[&R], usize) -> Result<Vec<OwnedReadResult>>,
read_paths: impl FnOnce(&[&Path], usize) -> Result<Vec<Vec<u8>>>,
) -> Result<Vec<OwnedReadResult>> {
if requests.is_empty() {
return Ok(Vec::new());
}
let (mut paths, mut path_indices, mut ranges, mut range_indices) =
(Vec::new(), Vec::new(), Vec::new(), Vec::new());
for (i, r) in requests.iter().enumerate() {
match &r.source {
Source::Whole(path) => {
paths.push(*path);
path_indices.push(i);
}
Source::Range(range) => {
ranges.push(range);
range_indices.push(i);
}
}
}
let remap = |e: Error, indices: &[usize]| match e.index() {
Some(i) if i < indices.len() => e.with_index(indices[i]),
Some(_) => Error::transport(None, "invalid vread_native backend error index"),
None => e,
};
let range_results = if ranges.is_empty() {
Vec::new()
} else {
read_ranges(&ranges, budget).map_err(|e| remap(e, &range_indices))?
};
if range_results.len() != ranges.len() {
return Err(Error::transport(
None,
"invalid vread_native range result count",
));
}
let mut remaining = budget;
for r in &range_results {
remaining = remaining
.checked_sub(r.data.len())
.ok_or_else(|| Error::transport(None, "vread_native backend exceeded byte budget"))?;
}
let path_results = if paths.is_empty() {
Vec::new()
} else {
read_paths(&paths, remaining).map_err(|e| remap(e, &path_indices))?
};
if path_results.len() != paths.len() {
return Err(Error::transport(
None,
"invalid vread_native path result count",
));
}
for r in &path_results {
remaining = remaining
.checked_sub(r.len())
.ok_or_else(|| Error::transport(None, "vread_native backend exceeded byte budget"))?;
}
if path_indices.is_empty() {
return Ok(range_results);
}
if range_indices.is_empty() {
return Ok(path_results
.into_iter()
.map(|data| OwnedReadResult {
offset: 0,
data,
eof: true,
})
.collect());
}
let mut output: Vec<Option<OwnedReadResult>> = (0..requests.len()).map(|_| None).collect();
for (i, r) in range_indices.into_iter().zip(range_results) {
output[i] = Some(r);
}
for (i, data) in path_indices.into_iter().zip(path_results) {
output[i] = Some(OwnedReadResult {
offset: 0,
data,
eof: true,
});
}
output
.into_iter()
.map(|r| r.ok_or_else(|| Error::transport(None, "vread_native omitted a result")))
.collect()
}
#[cfg(test)]
mod option_layout_tests {
use super::ReadOptions;
use std::num::NonZeroUsize;
#[test]
fn read_options_are_one_word_and_keep_inheritance_distinct() {
assert_eq!(
std::mem::size_of::<ReadOptions>(),
std::mem::size_of::<usize>()
);
let mut options = ReadOptions::new();
assert_eq!(options.max_total_bytes_or(7), 7);
for bytes in [1, 42, usize::MAX] {
let limit = NonZeroUsize::new(bytes).unwrap();
options = options.with_max_total_bytes(limit);
assert_eq!(options.max_total_bytes(), Some(limit));
assert_eq!(options.max_total_bytes_or(7), bytes);
}
options = ReadOptions::new();
assert_eq!(options.max_total_bytes(), None);
assert_eq!(options.max_total_bytes_or(7), 7);
assert_eq!(
options.max_total_bytes_or(0),
0,
"an inherited zero policy must not become unlimited"
);
}
}
#[cfg(test)]
mod compact_read_tests {
use super::*;
#[test]
fn compact_read_results_preserve_owned_and_buffered_empty_progress() {
if cfg!(target_pointer_width = "64") {
assert_eq!(std::mem::size_of::<ReadResult>(), 40);
}
let owned = ReadResult::owned(7, Vec::new(), true);
const BUFFERED: ReadResult = ReadResult::buffered(7, 0, true);
assert_eq!(owned.bytes_read(), 0);
assert_eq!(owned.offset(), BUFFERED.offset());
assert_eq!(owned.eof(), BUFFERED.eof());
assert_eq!(owned.data(), Some(b"".as_slice()));
assert_eq!(BUFFERED.data(), None);
assert_ne!(owned, BUFFERED);
assert!(!owned.is_buffered());
assert!(BUFFERED.is_buffered());
assert_eq!(owned.clone().into_data(), Some(Vec::new()));
assert_eq!(BUFFERED.clone().into_data(), None);
let data = vec![1, 2, 3];
let pointer = data.as_ptr();
let result = ReadResult::owned(9, data, false);
assert_eq!(result.bytes_read(), 3);
let extracted = result.into_data().unwrap();
assert_eq!(extracted.as_ptr(), pointer);
assert_eq!(
ReadResult::buffered(0, usize::MAX, false).bytes_read(),
usize::MAX
);
}
}