use super::*;
pub fn vread_all_impl_default<F: VectorBackend + ?Sized>(
backend: &mut F,
files: &[FileOperand],
) -> VfResult<Vec<Vec<u8>>> {
backend.vread_all_with_options_impl(files, ReadAllOptions::default())
}
pub fn vread_all_with_options_impl_default<F: VectorBackend + ?Sized>(
backend: &mut F,
files: &[FileOperand],
options: ReadAllOptions,
) -> VfResult<Vec<Vec<u8>>> {
let mut out: Vec<Vec<u8>> = files.iter().map(|_| Vec::new()).collect();
let mut total = 0usize;
let mut limit_error = None;
let stream_budget = options
.max_total_bytes()
.clamp(1, DEFAULT_READ_ALLV_MAX_TOTAL_BYTES);
let chunk_size = stream_budget.min(1024 * 1024);
backend.vstream_impl(
files,
chunk_size,
stream_budget,
&mut |index, _, data, _| {
let Some(next_total) = total.checked_add(data.len()) else {
limit_error = Some(index);
return false;
};
if next_total > options.max_total_bytes() {
limit_error = Some(index);
return false;
}
out[index].extend_from_slice(data);
total = next_total;
true
},
)?;
if let Some(index) = limit_error {
return Err(VfError::failure(index, libc::EFBIG as u32));
}
Ok(out)
}
pub fn vstream_impl_default<F: VectorBackend + ?Sized>(
backend: &mut F,
files: &[FileOperand],
chunk_size: usize,
memory_limit: usize,
cb: &mut ReadStreamCallback<'_>,
) -> VfRes {
use std::collections::VecDeque;
if chunk_size == 0 || memory_limit == 0 {
return Err(VfError::failure(0, ERR_INVAL));
}
let mut pending: VecDeque<usize> = (0..files.len()).collect();
let mut offsets = vec![0u64; files.len()];
while !pending.is_empty() {
let mut budget = memory_limit;
let mut batch_indices = Vec::new();
let mut reads = Vec::new();
while budget > 0 && !pending.is_empty() {
let index = pending.pop_front().expect("pending was non-empty");
let length = chunk_size.min(budget);
reads.push(ReadOp::at(files[index].clone(), offsets[index], length));
batch_indices.push(index);
budget -= length;
}
let remap_error = |error: VfError| {
error.map_index(|index| batch_indices.get(index).copied().unwrap_or(index))
};
let results = backend.vread_impl(&reads).map_err(remap_error)?;
validate_read_results("vstream_impl", &reads, &results).map_err(remap_error)?;
for (batch_index, result) in results.into_iter().enumerate() {
let index = batch_indices[batch_index];
let offset = offsets[index];
let eof = result.eof;
offsets[index] = offset
.checked_add(result.data.len() as u64)
.ok_or_else(|| VfError::failure(index, libc::EOVERFLOW as u32))?;
if !cb(index, offset, &result.data, eof) {
return Ok(());
}
if !eof {
pending.push_back(index);
}
}
}
Ok(())
}
pub fn native_read_file_impl_default<F: VectorBackend + ?Sized>(
backend: &mut F,
file: &FileOperand,
max_bytes: usize,
) -> VfResult<Vec<u8>> {
backend
.vread_all_with_options_impl(
std::slice::from_ref(file),
ReadAllOptions::new().with_max_total_bytes(max_bytes),
)
.and_then(|mut results| {
if results.len() != 1 {
return Err(VfError::transport(
None,
"read_file backend returned an invalid result count",
));
}
let data = results.pop().expect("validated result count");
if data.len() > max_bytes {
return Err(VfError::client(0, libc::EFBIG as u32));
}
Ok(data)
})
}