upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-09-21 13:37:46 -05:00
Code Issues Packages Projects Releases Wiki Activity

perf: read whole objects in a single GET instead of HEAD+GET (#9587)

Browse Source 
* perf: read whole objects in a single GET instead of HEAD+GET

* feat: read object tail from offset to EOF in a single GET

Generalize the whole-object single-GET read so a tail read from a known
offset also avoids the separate len()/HEAD round-trip. The backend
primitive becomes `read_from(path, from)`, issuing an open-ended
`GetRange::Offset(from)` GET (or a plain GET for from == 0) and reporting
the object's total size from the response; `read_whole` is now a thin
wrapper over it.

The owned blob pipeline's `schedule_whole` no longer HEADs the remote to
size a tail read. An offset at/past EOF is an unsatisfiable range (HTTP
416) rather than an empty body, so the buffer builder disambiguates with
a single len() only on the error path, yielding an empty read when the
tail is genuinely empty. The disk cache's reopen prefiller is made
tolerant of that empty tail so it no longer truncates the local mirror.

Also split the now-hard-to-follow simple_disk_cache `file.rs` into a
`file/` module (type/state, init state machine, reopen, read surface)
and document the FromScratch vs Prefiller init sources.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: generall <andrey@vasnetsov.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Daniel Boros
2026-08-04 11:16:55 +02:00
committed by generall
co-authored by Claude Opus 4.8 generall
parent d4b06b0ece
commit 9f4a2eda30
13 changed files with 1030 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/common/common/src/universal_io/simple_disk_cache/file.rs
-386
View File
@@ -1,386 +0,0 @@
use std::assert_matches;
use std::fmt::Debug;
use std::io::{self, ErrorKind};
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use parking_lot::Mutex;
use super::BLOCK_SIZE;
use super::fs::DiskCacheFs;
use crate::ext::aligned_vec::ACow;
use crate::generic_consts::{AccessPattern, Sequential};
use crate::mmap::AdviceSetting;
use crate::universal_io::simple_disk_cache::local_state::LocalState;
use crate::universal_io::simple_disk_cache::pipeline::{DiskCachePipeline, OwnedDiskCachePipeline};
use crate::universal_io::simple_disk_cache::{DiskCacheRemote, to_block_range};
use crate::universal_io::{
BorrowedReadPipeline, OpenOptions, OwnedReadPipeline, Populate, ReadRange, Result,
UniversalIoError, UniversalKind, UniversalRead, UniversalReadFs, UserData,
};
/// A lazily-populated local mirror of an immutable remote file.
///
/// The remote is assumed to be immutable for the lifetime of the file;
/// this type implements [`UniversalRead`] only, but not [`UniversalWrite`].
///
/// The local mirror can either be initialized lazily on first read (filling
/// blocks on demand from the remote) or eagerly if populate is set.
///
/// WARN: There should be only a single instance of DiskCache per path.
/// Initializing multiple instances will try to re-read from remote.
pub struct DiskCache<R>
where
R: UniversalRead,
{
/// Clone of the remote filesystem handle, used to lazily open `remote`.
remote_fs: R::Fs,
/// Backend-specific per-open extras for the remote.
remote_extra: <R::Fs as UniversalReadFs>::OpenExtra,
/// Path to the remote file. Used to lazily open `remote`.
remote_path: PathBuf,
/// Open options for when the local mmap is initialized.
pub(super) open_options: OpenOptions,
/// Path to the local mmap file.
pub(super) local_path: PathBuf,
/// Lazily-initialized mirror.
// We could switch to LazyLock, but there is no fallible initialization
pub(super) state: OnceLock<State<R>>,
/// Guards initialization of `local` and carries the source of init.
pub(super) init_lock: Arc<Mutex<InitSource<R>>>,
}
#[derive(Debug)]
pub(super) struct State<R> {
pub remote: R,
pub local: LocalState,
}
impl<R> Debug for DiskCache<R>
where
R: UniversalRead,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DiskCache")
.field("remote_fs", &self.remote_fs)
.field("remote_path", &self.remote_path)
.field("open_options", &self.open_options)
.field("local_path", &self.local_path)
.field("state", &self.state)
.finish_non_exhaustive()
}
}
/// Where the [`LocalState`] comes from on first init.
pub(super) enum InitSource<R: UniversalRead> {
/// Build an empty local mmap and let reads fill blocks on demand.
FromScratch,
/// Wait for the prefill pipeline.
Prefiller(R::OwnedReadPipeline<()>),
/// Wait for the prefill pipeline, but from reopen
PartialPrefiller {
prefiller: R::OwnedReadPipeline<u64>,
local_state: LocalState,
},
}
impl<R> Debug for InitSource<R>
where
R: UniversalRead,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
InitSource::FromScratch => write!(f, "FromScratch"),
InitSource::Prefiller(_) => write!(f, "Prefiller"),
InitSource::PartialPrefiller { .. } => write!(f, "PartialPrefiller"),
}
}
}
impl<R> DiskCache<R>
where
R: DiskCacheRemote,
{
pub(super) fn new(
remote_fs: R::Fs,
remote_extra: <R::Fs as UniversalReadFs>::OpenExtra,
remote_path: impl AsRef<Path>,
local_path: PathBuf,
options: OpenOptions,
init_source: InitSource<R>,
) -> Self {
Self {
remote_fs,
remote_extra,
remote_path: remote_path.as_ref().to_owned(),
open_options: options,
local_path,
state: OnceLock::new(),
init_lock: Arc::new(Mutex::new(init_source)),
}
}
pub(super) fn open_remote(&self) -> Result<R> {
let remote_options = OpenOptions {
writeable: false,
populate: Populate::No,
need_sequential: false,
advice: AdviceSetting::Global,
};
self.remote_fs
.open(&self.remote_path, remote_options, self.remote_extra.clone())
}
/// Return the cached [`State`], initializing it on first call.
pub(super) fn state(&self) -> Result<&State<R>> {
if let Some(state) = self.state.get() {
return Ok(state);
}
let mut init_guard = self.init_lock.lock();
// Try again now that we have the lock, in case another thread initialized it first.
if self.state.get().is_none() {
self.init_state(&mut init_guard, true, None)?;
}
Ok(self.state.get().expect("just initialized"))
}
/// Initialize the local state depending on `InitSource`
///
/// If `allow_from_scratch` is false, this method will avoid initializing if `InitSource::FromScratch` is set.
/// This is helpful for [`Self::reopen`] scenario where we can avoid work if no reads have taken place.
pub(super) fn init_state(
&self,
init_guard: &mut InitSource<R>,
allow_from_scratch: bool,
known_length: Option<u64>,
) -> Result<()> {
let state = match std::mem::replace(init_guard, InitSource::FromScratch) {
InitSource::FromScratch => {
if !allow_from_scratch {
return Ok(());
}
self.new_state_from_scratch(self.open_remote()?, known_length)?
}
InitSource::Prefiller(mut prefiller) => {
match prefiller.wait()? {
Some((_, bytes)) => {
let local = LocalState::new(
&self.local_path,
// bytes length is the length of the remote file
bytes.len() as u64,
self.open_options,
)?;
let blocks_range = to_block_range(0..bytes.len() as u64);
unsafe { local.write_mmap_bytes(&bytes, blocks_range) };
State {
local,
remote: prefiller.into_inner(),
}
}
None => {
debug_assert!(
false,
"Looks like the request for prefill bytes was incorrect"
);
if !allow_from_scratch {
return Ok(());
}
// init from scratch
self.new_state_from_scratch(prefiller.into_inner(), known_length)?
}
}
}
InitSource::PartialPrefiller {
mut prefiller,
mut local_state,
} => {
match prefiller.wait()? {
Some((start, bytes)) => {
let end = start + bytes.len() as u64;
local_state.resize(&self.local_path, end)?;
let blocks_range = to_block_range(start..end);
unsafe { local_state.write_mmap_bytes(&bytes, blocks_range) }
}
None => {
// The remote file didn't grow
// TODO: double check that the remote didn't shrink?
}
};
let remote = prefiller.into_inner();
State {
remote,
local: local_state,
}
}
};
self.state
.set(state)
.expect("OnceLock::set must succeed while holding init_lock");
Ok(())
}
fn new_state_from_scratch(&self, remote: R, known_length: Option<u64>) -> Result<State<R>> {
let len = match known_length {
Some(len) => len,
None => remote.len::<u8>()?,
};
let local = LocalState::new(&self.local_path, len, self.open_options)?;
Ok(State { remote, local })
}
/// Make sure every byte in the range `byte_start..remote_len` is present on the local file
fn populate_from(&self, byte_start: u64) -> std::result::Result<(), UniversalIoError> {
if crate::low_memory::low_memory_mode().skip_populate() {
return Ok(());
}
let remote_len = self.state()?.remote.len::<u8>()?;
if remote_len == 0 {
return Ok(());
}
let one_byte_per_block = (byte_start..remote_len)
.step_by(BLOCK_SIZE)
.map(|byte_offset| ((), ReadRange::one(byte_offset)));
for result in self.read_iter::<Sequential, u8, ()>(one_byte_per_block)? {
result?;
}
Ok(())
}
}
impl<R> UniversalRead for DiskCache<R>
where
R: DiskCacheRemote,
{
type Fs = DiskCacheFs<R>;
type BorrowedReadPipeline<'a, U>
= DiskCachePipeline<'a, R, U>
where
R: 'a,
Self: 'a,
U: UserData;
type OwnedReadPipeline<U>
= OwnedDiskCachePipeline<R, U>
where
U: UserData;
fn reopen(&mut self) -> Result<()> {
// Wait for InitSource::Prefill, if set.
let mut init_guard = self.init_lock.lock();
self.init_state(&mut init_guard, false, None)?;
let Some(state) = self.state.take() else {
// If `self.state` didn't initialize after `init_state`, we are not populating
// and we haven't made any reads.
//
// The first read will take care of initializing to the remote length.
return Ok(());
};
let State {
mut remote,
mut local,
} = state;
// Reopen remote so it reflects current length
remote.reopen()?;
let local_len = local.mmap().len::<u8>()?;
match self.open_options.populate {
Populate::Auto | Populate::No => {
let remote_len = remote.len::<u8>()?;
// The remote is assumed to be append-only; a smaller file is unexpected.
if local_len > remote_len {
return Err(UniversalIoError::Io(io::Error::new(
ErrorKind::UnexpectedEof,
format!(
"Reopen encountered a smaller file than expected; old_len: {local_len}, new_len: {remote_len}"
),
)));
}
// Make the new length visible; new blocks will be filled lazily on read.
local.resize(&self.local_path, remote_len)?;
// return the updated local state and remote
self.state
.set(State { remote, local })
.expect("we just take()'d it");
}
Populate::Blocking | Populate::PreferBackground => {
// Re-fetch from the start of the (possibly partial) tail block so
// we still make an page-aligned read.
let from = local_len.saturating_sub(local_len % BLOCK_SIZE as u64);
let mut remote_pipeline = R::OwnedReadPipeline::new(remote)?;
// FIXME: check can_schedule in a loop?
remote_pipeline.schedule_whole(from, from)?;
assert_matches!(
*init_guard,
InitSource::FromScratch,
"by this point, InitSource must be FromScratch"
);
*init_guard = InitSource::PartialPrefiller {
prefiller: remote_pipeline,
local_state: local,
};
// For blocking, resolve the prefill now instead of on first read.
if matches!(self.open_options.populate, Populate::Blocking) {
self.init_state(&mut init_guard, false, None)?;
}
}
}
Ok(())
}
fn read_bytes<P: AccessPattern>(&self, range: Range<u64>, align: usize) -> Result<ACow<'_>> {
let mut pipeline = DiskCachePipeline::<R, ()>::new()?;
pipeline.schedule::<P>((), self, range, align)?;
let (_, bytes) = pipeline.wait()?.expect("there's exactly one read");
Ok(bytes)
}
fn len<T>(&self) -> Result<u64> {
self.state()?.local.mmap().len::<T>()
}
fn populate(&self) -> Result<()> {
self.populate_from(0)
}
fn populate_auto() -> bool {
false
}
fn clear_ram_cache(&self) -> Result<()> {
if let Some(state) = self.state.get() {
state.local.mmap().clear_ram_cache()?;
}
Ok(())
}
fn kind() -> UniversalKind {
UniversalKind::SimpleDiskCache
}
}
lib/common/common/src/universal_io/simple_disk_cache/file/init.rs
+207
View File
@@ -0,0 +1,207 @@
//! Lazy first-use initialization of a [`DiskCache`]'s [`State`].
//!
//! A freshly opened `DiskCache` has no [`State`] yet — only an [`InitSource`]
//! describing *how* the local mirror should be brought to life. The first
//! operation that actually needs the mirror ([`DiskCache::state`], or an
//! explicit prefill from [`super::reopen`]) takes `init_lock` and runs
//! [`DiskCache::init_state`] exactly once, consuming the `InitSource` and
//! publishing a populated `State` into the `OnceLock`.
use super::{DiskCache, State};
use crate::universal_io::simple_disk_cache::local_state::LocalState;
use crate::universal_io::simple_disk_cache::{DiskCacheRemote, to_block_range};
use crate::universal_io::{OwnedReadPipeline, Result, UniversalRead};
/// Where a [`DiskCache`]'s [`State`] comes from the first time it is needed.
///
/// `FromScratch` is *lazy*: it needs only the remote's **length** and faults
/// blocks in on demand. `Prefiller` is *eager*: it holds an in-flight whole-object
/// read and fills the mirror from those **bytes** at once.
///
/// ```text
/// Populate::No | Auto → FromScratch → remote.len(); empty mmap; blocks faulted in on read
/// Populate::Blocking|Pref → Prefiller(pipeline) → pipeline.wait(); mmap sized to bytes; all written
/// ```
pub(in crate::universal_io::simple_disk_cache) enum InitSource<R: UniversalRead> {
/// Lazy: build an empty local mmap (sized from the remote length) and let
/// reads fill blocks on demand. Chosen for `Populate::No` / `Populate::Auto`.
FromScratch,
/// Eager: an in-flight whole-object read scheduled at open time; init waits
/// on it and writes the whole mirror. For `Populate::Blocking` / `PreferBackground`.
Prefiller(R::OwnedReadPipeline<()>),
/// The reopen-time counterpart of [`Prefiller`](Self::Prefiller): an in-flight
/// read of just the appended tail (block-aligned old length → new EOF). Init
/// resizes the mirror and writes only that suffix. See [`super::reopen`].
PartialPrefiller {
prefiller: R::OwnedReadPipeline<u64>,
local_state: LocalState,
},
}
impl<R> std::fmt::Debug for InitSource<R>
where
R: UniversalRead,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
InitSource::FromScratch => write!(f, "FromScratch"),
InitSource::Prefiller(_) => write!(f, "Prefiller"),
InitSource::PartialPrefiller { .. } => write!(f, "PartialPrefiller"),
}
}
}
impl<R> DiskCache<R>
where
R: DiskCacheRemote,
{
/// Return the cached [`State`], initializing it on first call.
pub(in crate::universal_io::simple_disk_cache) fn state(&self) -> Result<&State<R>> {
if let Some(state) = self.state.get() {
return Ok(state);
}
let mut init_guard = self.init_lock.lock();
// Try again now that we have the lock, in case another thread initialized it first.
if self.state.get().is_none() {
self.init_state(&mut init_guard, true, None)?;
}
Ok(self.state.get().expect("just initialized"))
}
/// Initialize the local state depending on `InitSource`
///
/// If `allow_from_scratch` is false, this method will avoid initializing if `InitSource::FromScratch` is set.
/// This is helpful for [`Self::reopen`] scenario where we can avoid work if no reads have taken place.
pub(in crate::universal_io::simple_disk_cache) fn init_state(
&self,
init_guard: &mut InitSource<R>,
allow_from_scratch: bool,
known_length: Option<u64>,
) -> Result<()> {
let state = match std::mem::replace(init_guard, InitSource::FromScratch) {
InitSource::FromScratch => {
if !allow_from_scratch {
return Ok(());
}
self.new_state_from_scratch(self.open_remote()?, known_length)?
}
InitSource::Prefiller(mut prefiller) => {
match prefiller.wait()? {
Some((_, bytes)) => {
let local = LocalState::new(
&self.local_path,
// bytes length is the length of the remote file
bytes.len() as u64,
self.open_options,
)?;
let blocks_range = to_block_range(0..bytes.len() as u64);
unsafe { local.write_mmap_bytes(&bytes, blocks_range) };
State {
local,
remote: prefiller.into_inner(),
}
}
None => {
debug_assert!(
false,
"Looks like the request for prefill bytes was incorrect"
);
if !allow_from_scratch {
return Ok(());
}
// init from scratch
self.new_state_from_scratch(prefiller.into_inner(), known_length)?
}
}
}
InitSource::PartialPrefiller {
mut prefiller,
mut local_state,
} => {
match prefiller.wait()? {
Some((start, bytes)) if !bytes.is_empty() => {
let end = start + bytes.len() as u64;
local_state.resize(&self.local_path, end)?;
let blocks_range = to_block_range(start..end);
unsafe { local_state.write_mmap_bytes(&bytes, blocks_range) }
}
// `None`: nothing was scheduled. `Some(_, empty)`: the
// open-ended tail read from our block-aligned offset came back
// empty, i.e. the remote didn't grow past it. Either way there
// is nothing to apply — and we must not `resize` down to the
// offset, which would truncate the local mirror.
// TODO: double check that the remote didn't shrink?
Some(_) | None => {}
};
let remote = prefiller.into_inner();
State {
remote,
local: local_state,
}
}
};
self.state
.set(state)
.expect("OnceLock::set must succeed while holding init_lock");
Ok(())
}
fn new_state_from_scratch(&self, remote: R, known_length: Option<u64>) -> Result<State<R>> {
let len = match known_length {
Some(len) => len,
None => remote.len::<u8>()?,
};
let local = LocalState::new(&self.local_path, len, self.open_options)?;
Ok(State { remote, local })
}
/// Fill the local mirror from one whole-object read when the cache is cold
/// (`InitSource::FromScratch`); other init sources fall back to the normal
/// initialization.
pub(super) fn ensure_whole_local(&self) -> Result<()> {
if self.state.get().is_some() {
return Ok(());
}
let mut init_guard = self.init_lock.lock();
if self.state.get().is_some() {
return Ok(());
}
// Only a cold cache (`FromScratch`) gets the dedicated whole-object fill
// below. If a prefill is already in flight (eager populate, or a reopen
// tail read), resolve it through the normal initialization path instead.
let from_scratch = match &*init_guard {
InitSource::FromScratch => true,
InitSource::Prefiller(_) => false,
InitSource::PartialPrefiller { .. } => false,
};
if !from_scratch {
return self.init_state(&mut init_guard, true, None);
}
let remote = self.open_remote()?;
let bytes = remote.read_whole::<u8>()?;
let len = bytes.len() as u64;
let local = LocalState::new(&self.local_path, len, self.open_options)?;
if len > 0 {
// SAFETY: `bytes` covers the whole file `0..len` and the remote is
// immutable, so the mmap is filled exactly once with correct data.
unsafe { local.write_mmap_bytes(&bytes, to_block_range(0..len)) };
}
self.state
.set(State { remote, local })
.expect("OnceLock::set must succeed while holding init_lock");
Ok(())
}
}
lib/common/common/src/universal_io/simple_disk_cache/file/mod.rs
+119
View File
@@ -0,0 +1,119 @@
//! [`DiskCache`]: a lazily-populated local mirror of an immutable remote file.
//!
//! The implementation is split across this module's submodules:
//!
//! - this file — the [`DiskCache`] type, its backing [`State`], and the cheap
//! constructors ([`DiskCache::new`], [`DiskCache::open_remote`]).
//! - [`init`] — how the mirror is brought to life on first use: the
//! [`InitSource`] state machine and [`DiskCache::init_state`]. **Start here**
//! to understand cold-start vs. eager-prefill behavior.
//! - [`reopen`] — refreshing the mirror after the (append-only) remote grew.
//! - [`read`] — the [`UniversalRead`] implementation (the public read surface).
use std::fmt::Debug;
use std::path::{Path, PathBuf};
use std::sync::{Arc, OnceLock};
use parking_lot::Mutex;
use super::DiskCacheRemote;
use super::local_state::LocalState;
use crate::mmap::AdviceSetting;
use crate::universal_io::{OpenOptions, Populate, Result, UniversalRead, UniversalReadFs};
mod init;
mod read;
mod reopen;
pub(in crate::universal_io::simple_disk_cache) use init::InitSource;
/// A lazily-populated local mirror of an immutable remote file.
///
/// The remote is assumed to be immutable for the lifetime of the file;
/// this type implements [`UniversalRead`] only, but not [`UniversalWrite`].
///
/// The local mirror can either be initialized lazily on first read (filling
/// blocks on demand from the remote) or eagerly if populate is set. See
/// [`init`] for the precise lifecycle.
///
/// WARN: There should be only a single instance of DiskCache per path.
/// Initializing multiple instances will try to re-read from remote.
pub struct DiskCache<R>
where
R: UniversalRead,
{
/// Clone of the remote filesystem handle, used to lazily open `remote`.
remote_fs: R::Fs,
/// Backend-specific per-open extras for the remote.
remote_extra: <R::Fs as UniversalReadFs>::OpenExtra,
/// Path to the remote file. Used to lazily open `remote`.
remote_path: PathBuf,
/// Open options for when the local mmap is initialized.
pub(super) open_options: OpenOptions,
/// Path to the local mmap file.
pub(super) local_path: PathBuf,
/// Lazily-initialized mirror.
// We could switch to LazyLock, but there is no fallible initialization
pub(super) state: OnceLock<State<R>>,
/// Guards initialization of `local` and carries the source of init.
pub(super) init_lock: Arc<Mutex<InitSource<R>>>,
}
/// The materialized mirror: the opened `remote` handle paired with its local
/// mmap mirror. Created exactly once, lazily, by [`DiskCache::init_state`].
#[derive(Debug)]
pub(super) struct State<R> {
pub remote: R,
pub local: LocalState,
}
impl<R> Debug for DiskCache<R>
where
R: UniversalRead,
{
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DiskCache")
.field("remote_fs", &self.remote_fs)
.field("remote_path", &self.remote_path)
.field("open_options", &self.open_options)
.field("local_path", &self.local_path)
.field("state", &self.state)
.finish_non_exhaustive()
}
}
impl<R> DiskCache<R>
where
R: DiskCacheRemote,
{
pub(super) fn new(
remote_fs: R::Fs,
remote_extra: <R::Fs as UniversalReadFs>::OpenExtra,
remote_path: impl AsRef<Path>,
local_path: PathBuf,
options: OpenOptions,
init_source: InitSource<R>,
) -> Self {
Self {
remote_fs,
remote_extra,
remote_path: remote_path.as_ref().to_owned(),
open_options: options,
local_path,
state: OnceLock::new(),
init_lock: Arc::new(Mutex::new(init_source)),
}
}
pub(super) fn open_remote(&self) -> Result<R> {
let remote_options = OpenOptions {
writeable: false,
populate: Populate::No,
need_sequential: false,
advice: AdviceSetting::Global,
};
self.remote_fs
.open(&self.remote_path, remote_options, self.remote_extra.clone())
}
}
lib/common/common/src/universal_io/simple_disk_cache/file/read.rs
+106
View File
@@ -0,0 +1,106 @@
//! The [`UniversalRead`] implementation for [`DiskCache`] — the public read
//! surface. The heavy lifting lives elsewhere: first-use init in [`super::init`],
//! growth handling in [`super::reopen`].
use std::borrow::Cow;
use std::ops::Range;
use super::DiskCache;
use crate::ext::aligned_vec::ACow;
use crate::generic_consts::{AccessPattern, Sequential};
use crate::universal_io::simple_disk_cache::fs::DiskCacheFs;
use crate::universal_io::simple_disk_cache::pipeline::{DiskCachePipeline, OwnedDiskCachePipeline};
use crate::universal_io::simple_disk_cache::{BLOCK_SIZE, DiskCacheRemote};
use crate::universal_io::{
BorrowedReadPipeline, Item, ReadRange, Result, UniversalIoError, UniversalKind, UniversalRead,
UserData,
};
impl<R> DiskCache<R>
where
R: DiskCacheRemote,
{
/// Make sure every byte in the range `byte_start..remote_len` is present on the local file
fn populate_from(&self, byte_start: u64) -> std::result::Result<(), UniversalIoError> {
if crate::low_memory::low_memory_mode().skip_populate() {
return Ok(());
}
let remote_len = self.state()?.remote.len::<u8>()?;
if remote_len == 0 {
return Ok(());
}
let one_byte_per_block = (byte_start..remote_len)
.step_by(BLOCK_SIZE)
.map(|byte_offset| ((), ReadRange::one(byte_offset)));
for result in self.read_iter::<Sequential, u8, ()>(one_byte_per_block)? {
result?;
}
Ok(())
}
}
impl<R> UniversalRead for DiskCache<R>
where
R: DiskCacheRemote,
{
type Fs = DiskCacheFs<R>;
type BorrowedReadPipeline<'a, U>
= DiskCachePipeline<'a, R, U>
where
R: 'a,
Self: 'a,
U: UserData;
type OwnedReadPipeline<U>
= OwnedDiskCachePipeline<R, U>
where
U: UserData;
fn reopen(&mut self) -> Result<()> {
self.reopen_impl()
}
fn read_bytes<P: AccessPattern>(&self, range: Range<u64>, align: usize) -> Result<ACow<'_>> {
let mut pipeline = DiskCachePipeline::<R, ()>::new()?;
pipeline.schedule::<P>((), self, range, align)?;
let (_, bytes) = pipeline.wait()?.expect("there's exactly one read");
Ok(bytes)
}
fn read_whole<T: Item>(&self) -> Result<Cow<'_, [T]>> {
self.ensure_whole_local()?;
let length = self.len::<T>()?;
self.read::<Sequential, T>(ReadRange {
byte_offset: 0,
length,
})
}
fn len<T>(&self) -> Result<u64> {
self.state()?.local.mmap().len::<T>()
}
fn populate(&self) -> Result<()> {
self.populate_from(0)
}
fn populate_auto() -> bool {
false
}
fn clear_ram_cache(&self) -> Result<()> {
if let Some(state) = self.state.get() {
state.local.mmap().clear_ram_cache()?;
}
Ok(())
}
fn kind() -> UniversalKind {
UniversalKind::SimpleDiskCache
}
}
lib/common/common/src/universal_io/simple_disk_cache/file/reopen.rs
+96
View File
@@ -0,0 +1,96 @@
//! Refreshing the local mirror after the (append-only) remote has grown.
//!
//! The remote is immutable except that it may be *appended to*. [`reopen`] picks
//! up that growth: for the lazy populate modes it just resizes the mirror and
//! lets later reads fault the new blocks in; for the eager modes it schedules a
//! tail read of the appended suffix and stages it as an
//! [`InitSource::PartialPrefiller`](super::InitSource::PartialPrefiller).
//!
//! [`reopen`]: crate::universal_io::UniversalRead::reopen
use std::assert_matches;
use std::io::{self, ErrorKind};
use super::{DiskCache, InitSource, State};
use crate::universal_io::simple_disk_cache::{BLOCK_SIZE, DiskCacheRemote};
use crate::universal_io::{OwnedReadPipeline, Populate, Result, UniversalIoError, UniversalRead};
impl<R> DiskCache<R>
where
R: DiskCacheRemote,
{
/// Body of [`UniversalRead::reopen`](crate::universal_io::UniversalRead::reopen).
pub(super) fn reopen_impl(&mut self) -> Result<()> {
// Wait for InitSource::Prefill, if set.
let mut init_guard = self.init_lock.lock();
self.init_state(&mut init_guard, false, None)?;
let Some(state) = self.state.take() else {
// If `self.state` didn't initialize after `init_state`, we are not populating
// and we haven't made any reads.
//
// The first read will take care of initializing to the remote length.
return Ok(());
};
let State {
mut remote,
mut local,
} = state;
// Reopen remote so it reflects current length
remote.reopen()?;
let local_len = local.mmap().len::<u8>()?;
match self.open_options.populate {
Populate::Auto | Populate::No => {
let remote_len = remote.len::<u8>()?;
// The remote is assumed to be append-only; a smaller file is unexpected.
if local_len > remote_len {
return Err(UniversalIoError::Io(io::Error::new(
ErrorKind::UnexpectedEof,
format!(
"Reopen encountered a smaller file than expected; old_len: {local_len}, new_len: {remote_len}"
),
)));
}
// Make the new length visible; new blocks will be filled lazily on read.
local.resize(&self.local_path, remote_len)?;
// return the updated local state and remote
self.state
.set(State { remote, local })
.expect("we just take()'d it");
}
Populate::Blocking | Populate::PreferBackground => {
// Re-fetch from the start of the (possibly partial) tail block so
// we still make an page-aligned read.
let from = local_len.saturating_sub(local_len % BLOCK_SIZE as u64);
let mut remote_pipeline = R::OwnedReadPipeline::new(remote)?;
// FIXME: check can_schedule in a loop?
remote_pipeline.schedule_whole(from, from)?;
assert_matches!(
*init_guard,
InitSource::FromScratch,
"by this point, InitSource must be FromScratch"
);
*init_guard = InitSource::PartialPrefiller {
prefiller: remote_pipeline,
local_state: local,
};
// For blocking, resolve the prefill now instead of on first read.
if matches!(self.open_options.populate, Populate::Blocking) {
self.init_state(&mut init_guard, false, None)?;
}
}
}
Ok(())
}
}
lib/common/io_bridge_object_store/src/file.rs
+25 -2
View File
@@ -1,12 +1,15 @@
use std::borrow::Cow;
use std::ops::Range;
use std::path::{Path, PathBuf};
use common::ext::aligned_vec::ACow;
use common::generic_consts::AccessPattern;
use common::universal_io::{Result, UniversalKind, UniversalRead, UserData};
use common::universal_io::{Item, Result, UniversalKind, UniversalRead, UserData};
use crate::fs::BlobFs;
use crate::pipeline::{BorrowedBlobPipeline, OwnedBlobPipeline, read_into_byte_buffer};
use crate::pipeline::{
BorrowedBlobPipeline, OwnedBlobPipeline, read_into_byte_buffer, read_whole_into_byte_buffer,
};
use crate::read::AsyncRead;
use crate::runtime::BridgeRuntime;
@@ -90,6 +93,15 @@ impl<A: AsyncRead + Clone> UniversalRead for BlobFile<A> {
Ok(ACow::Owned(buf))
}
fn read_whole<T: Item>(&self) -> Result<Cow<'_, [T]>> {
let buf = self
.runtime
.block_on(read_whole_into_byte_buffer::<A>(self, align_of::<T>()))?;
Ok(ACow::Owned(buf)
.try_cast_bytemuck()
.expect("aligned whole-object buffer casts to [T]"))
}
fn len<T>(&self) -> Result<u64> {
let item_size = size_of::<T>() as u64;
let len = self.runtime.block_on(self.inner.len(&self.path))?;
@@ -188,6 +200,17 @@ mod tests {
async move { Ok(futures::stream::once(async move { Ok(bytes) }).boxed()) }
}
fn read_from(
&self,
_path: &Path,
from: u64,
) -> impl Future<Output = Result<(u64, BoxStream<'static, Result<Bytes>>)>> + Send + 'static
{
let size = self.data.len() as u64;
let tail = self.data.slice(from as usize..);
async move { Ok((size, futures::stream::once(async move { Ok(tail) }).boxed())) }
}
fn len(&self, _path: &Path) -> impl Future<Output = Result<u64>> + Send + 'static {
let len = self.data.len() as u64;
async move { Ok(len) }
lib/common/io_bridge_object_store/src/pipeline/buffer.rs
+79 -23
View File
@@ -2,8 +2,10 @@ use std::future::Future;
use std::ops::Range;
use aligned_vec::{AVec, RuntimeAlign, avec_rt};
use bytes::Bytes;
use common::universal_io::{Result, UniversalIoError};
use futures::StreamExt as _;
use futures::stream::BoxStream;
use crate::file::BlobFile;
use crate::read::AsyncRead;
@@ -25,28 +27,82 @@ pub(crate) fn read_into_byte_buffer<A: AsyncRead>(
let len = (range.end - range.start) as usize;
let stream_fut = file.inner.read_range(&file.path, range);
async move {
let mut stream = stream_fut.await?;
let mut buf = avec_rt!([align] | 0u8; len);
let mut off = 0;
while let Some(chunk) = stream.next().await {
let chunk = chunk?;
let end = off + chunk.len();
if end > buf.len() {
return Err(UniversalIoError::S3Config {
description: format!(
"over-read: tried to write {end} bytes into a buffer of size {}",
buf.len(),
),
});
}
buf[off..end].copy_from_slice(&chunk);
off = end;
}
if off != buf.len() {
return Err(UniversalIoError::S3Config {
description: format!("short read: expected {} bytes, got {off}", buf.len()),
});
}
Ok(buf)
let stream = stream_fut.await?;
let buf = avec_rt!([align] | 0u8; len);
fold_stream_into_buffer(stream, buf).await
}
}
/// Like [`read_into_byte_buffer`], but fetches the whole object in one GET,
/// sizing the buffer from the response length (no separate `len`/HEAD).
pub(crate) fn read_whole_into_byte_buffer<A: AsyncRead + Clone>(
file: &BlobFile<A>,
align: usize,
) -> impl Future<Output = Result<AVec<u8, RuntimeAlign>>> + Send + 'static {
read_from_into_byte_buffer(file, 0, align)
}
/// Like [`read_into_byte_buffer`], but fetches everything from byte offset
/// `from` to the end of the object in one open-ended GET, sizing the buffer
/// from the object's total length carried in the response — no separate
/// `len`/HEAD round-trip on the happy path. `from == 0` reads the whole object.
///
/// An offset at or past the end has no tail to read. The backend reports that as
/// an unsatisfiable-range error rather than an empty body, so the error path
/// confirms with a single `len`: if `from >= eof` the tail is genuinely empty
/// and we yield a zero-length buffer; otherwise the original read error stands.
pub(crate) fn read_from_into_byte_buffer<A: AsyncRead + Clone>(
file: &BlobFile<A>,
from: u64,
align: usize,
) -> impl Future<Output = Result<AVec<u8, RuntimeAlign>>> + Send + 'static {
let read_fut = file.inner.read_from(&file.path, from);
// Cloned for the cold disambiguation path only; building the `len` future is
// deferred until a read error actually occurs.
let inner = file.inner.clone();
let path = file.path.clone();
async move {
let (size, stream) = match read_fut.await {
Ok(ok) => ok,
Err(err) => {
let eof = inner.len(&path).await?;
if from >= eof {
return Ok(avec_rt!([align] | 0u8; 0));
}
return Err(err);
}
};
let len = size.saturating_sub(from) as usize;
let buf = avec_rt!([align] | 0u8; len);
fold_stream_into_buffer(stream, buf).await
}
}
/// Copy every chunk of `stream` into `buf`, erroring if the streamed bytes
/// don't exactly fill it.
async fn fold_stream_into_buffer(
mut stream: BoxStream<'static, Result<Bytes>>,
mut buf: AVec<u8, RuntimeAlign>,
) -> Result<AVec<u8, RuntimeAlign>> {
let mut off = 0;
while let Some(chunk) = stream.next().await {
let chunk = chunk?;
let end = off + chunk.len();
if end > buf.len() {
return Err(UniversalIoError::S3Config {
description: format!(
"over-read: tried to write {end} bytes into a buffer of size {}",
buf.len(),
),
});
}
buf[off..end].copy_from_slice(&chunk);
off = end;
}
if off != buf.len() {
return Err(UniversalIoError::S3Config {
description: format!("short read: expected {} bytes, got {off}", buf.len()),
});
}
Ok(buf)
}
lib/common/io_bridge_object_store/src/pipeline/mod.rs
+1 -1
View File
@@ -14,7 +14,7 @@ mod inner;
mod owned;
pub use borrowed::BorrowedBlobPipeline;
pub(crate) use buffer::read_into_byte_buffer;
pub(crate) use buffer::{read_into_byte_buffer, read_whole_into_byte_buffer};
pub use owned::OwnedBlobPipeline;
pub(crate) const BLOB_PIPELINE_CAPACITY: usize = 256;
lib/common/io_bridge_object_store/src/pipeline/owned.rs
+9 -9
View File
@@ -1,10 +1,10 @@
use std::ops::Range;
use common::ext::aligned_vec::ACow;
use common::generic_consts::{AccessPattern, Sequential};
use common::universal_io::{OwnedReadPipeline, Result, UniversalRead, UserData};
use common::generic_consts::AccessPattern;
use common::universal_io::{OwnedReadPipeline, Result, UserData};
use super::buffer::read_into_byte_buffer;
use super::buffer::{read_from_into_byte_buffer, read_into_byte_buffer};
use super::inner::PipelineInner;
use crate::file::BlobFile;
use crate::read::AsyncRead;
@@ -50,12 +50,12 @@ where
}
fn schedule_whole(&mut self, user_data: U, from: u64) -> Result<()> {
// TODO(uio): implement schedule_whole in `AsyncRead`
let eof = self.file.len::<u8>()?;
if from >= eof {
return Ok(());
}
self.schedule::<Sequential>(user_data, from..eof, 1)
// One open-ended GET from `from` to EOF, byte-aligned, sized from the
// response — no separate `len`/HEAD round-trip. `from == 0` reads the
// whole object; an offset at or past EOF resolves to an empty read
// inside the future (see `read_from_into_byte_buffer`).
let future = read_from_into_byte_buffer::<A>(&self.file, from, 1);
self.inner.schedule(&self.file.runtime, user_data, future)
}
fn wait(&mut self) -> Result<Option<(U, ACow<'_>)>> {
lib/common/io_bridge_object_store/src/read.rs
+31
View File
@@ -65,6 +65,37 @@ pub trait AsyncRead: Send + Sync + Sized + 'static {
range: Range<u64>,
) -> impl Future<Output = Result<BoxStream<'static, Result<Bytes>>>> + Send + 'static;
/// Fetch the object at `path` from byte offset `from` to its end in one
/// request — no separate `len`/HEAD round-trip. `from == 0` reads the whole
/// object.
///
/// The returned `u64` is the **total size of the whole object, in bytes**
/// (as reported by the GET response, e.g. parsed from `Content-Range`/
/// `Content-Length`). It is *not* the length of the returned tail: the
/// stream yields exactly `total - from` bytes, so the absolute offset of EOF
/// is `total`, and on success `from <= total` always holds. For `from == 0`
/// the two coincide (`total` bytes are streamed).
///
/// If `from` is at or past the end of the object the request may be rejected
/// by the backend as an unsatisfiable range (e.g. HTTP 416). Callers that
/// must tolerate an empty tail should disambiguate with [`len`](Self::len);
/// see `pipeline::read_from_into_byte_buffer`.
fn read_from(
&self,
path: &Path,
from: u64,
) -> impl Future<Output = Result<(u64, BoxStream<'static, Result<Bytes>>)>> + Send + 'static;
/// Fetch the whole object at `path` in one request. Convenience for
/// [`read_from(path, 0)`](Self::read_from).
fn read_whole(
&self,
path: &Path,
) -> impl Future<Output = Result<(u64, BoxStream<'static, Result<Bytes>>)>> + Send + 'static
{
self.read_from(path, 0)
}
fn len(&self, path: &Path) -> impl Future<Output = Result<u64>> + Send + 'static;
fn is_empty(&self, path: &Path) -> impl Future<Output = Result<bool>> + Send + 'static {
lib/common/io_bridge_object_store/src/source.rs
+29
View File
@@ -166,6 +166,35 @@ impl<S: BlobBackend> AsyncRead for Arc<S> {
}
}
fn read_from(
&self,
path: &Path,
from: u64,
) -> impl Future<Output = Result<(u64, BoxStream<'static, Result<Bytes>>)>> + Send + 'static
{
let store = self.clone();
let key = build_key(path);
async move {
// `from == 0` is a plain whole-object GET; a positive offset asks the
// backend for everything from `from` onward in a single open-ended
// GET (`Range: bytes=from-`). Either way the response carries the
// object's total size, so no separate HEAD is needed.
let opts = GetOptions {
range: (from > 0).then_some(GetRange::Offset(from)),
..Default::default()
};
let result = store.get_opts(&key, opts).await.map_err(|err| match err {
object_store::Error::NotFound { .. } => UniversalIoError::NotFound {
path: PathBuf::from(key.to_string()),
},
other => UniversalIoError::s3(other),
})?;
let size = result.meta.size;
let stream = result.into_stream().map_err(UniversalIoError::s3).boxed();
Ok((size, stream))
}
}
fn len(&self, path: &Path) -> impl Future<Output = Result<u64>> + Send + 'static {
let store = self.clone();
let key = build_key(path);
lib/common/io_bridge_object_store/src/tests/mod.rs
+2
View File
@@ -1,3 +1,5 @@
#[cfg(test)]
mod integration;
pub mod rustfs;
#[cfg(test)]
mod whole_read;
lib/common/io_bridge_object_store/src/tests/whole_read.rs
+326
View File
@@ -0,0 +1,326 @@
#![cfg(test)]
use std::future::Future;
use std::ops::Range;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use bytes::Bytes;
use common::generic_consts::Sequential;
use common::universal_io::{
DiskCacheConfig, DiskCacheFs, DiskCacheFsContext, OpenOptions, OwnedReadPipeline, Populate,
ReadRange, Result, UniversalIoError, UniversalKind, UniversalRead, UniversalReadFileOps,
UniversalReadFs,
};
use futures::stream::{BoxStream, StreamExt};
use crate::pipeline::OwnedBlobPipeline;
use crate::read::AsyncRead;
use crate::{BlobFile, BridgeRuntime};
/// Request counters shared with every clone of a [`CountingSource`], so a test
/// can assert how many remote requests of each kind a read path issued.
#[derive(Clone, Default)]
struct Counters {
len: Arc<AtomicUsize>,
whole: Arc<AtomicUsize>,
range: Arc<AtomicUsize>,
/// Open-ended tail GETs: `read_from` with a positive offset.
from: Arc<AtomicUsize>,
}
#[derive(Clone)]
struct CountingConfig {
data: Bytes,
counters: Counters,
}
/// Test [`AsyncRead`] backend serving a fixed blob and counting the remote
/// requests it receives: `len` (HEAD), `read_whole` (single GET) and
/// `read_range` (ranged GET).
#[derive(Clone)]
struct CountingSource {
data: Bytes,
counters: Counters,
}
impl CountingSource {
fn new(data: &'static [u8]) -> Self {
Self {
data: Bytes::from_static(data),
counters: Counters::default(),
}
}
fn config(&self) -> CountingConfig {
CountingConfig {
data: self.data.clone(),
counters: self.counters.clone(),
}
}
}
impl AsyncRead for CountingSource {
type Config = CountingConfig;
fn open(config: &Self::Config) -> Result<Self> {
Ok(Self {
data: config.data.clone(),
counters: config.counters.clone(),
})
}
fn list_files(
&self,
_prefix: &Path,
) -> impl Future<Output = Result<Vec<PathBuf>>> + Send + 'static {
std::future::ready(Ok(vec![]))
}
fn exists(&self, _path: &Path) -> impl Future<Output = Result<bool>> + Send + 'static {
std::future::ready(Ok(true))
}
fn create(&self, _path: &Path) -> impl Future<Output = Result<()>> + Send + 'static {
std::future::ready(Ok(()))
}
fn remove(&self, _path: &Path) -> impl Future<Output = Result<()>> + Send + 'static {
std::future::ready(Ok(()))
}
fn remove_dir(&self, _path: &Path) -> impl Future<Output = Result<()>> + Send + 'static {
std::future::ready(Ok(()))
}
fn atomic_save(
&self,
_path: &Path,
_bytes: Bytes,
) -> impl Future<Output = Result<()>> + Send + 'static {
std::future::ready(Ok(()))
}
fn read_range(
&self,
_path: &Path,
range: Range<u64>,
) -> impl Future<Output = Result<BoxStream<'static, Result<Bytes>>>> + Send + 'static {
self.counters.range.fetch_add(1, Ordering::Relaxed);
let bytes = self.data.slice(range.start as usize..range.end as usize);
async move { Ok(futures::stream::once(async move { Ok(bytes) }).boxed()) }
}
fn read_from(
&self,
_path: &Path,
from: u64,
) -> impl Future<Output = Result<(u64, BoxStream<'static, Result<Bytes>>)>> + Send + 'static
{
if from == 0 {
self.counters.whole.fetch_add(1, Ordering::Relaxed);
} else {
self.counters.from.fetch_add(1, Ordering::Relaxed);
}
let size = self.data.len() as u64;
let data = self.data.clone();
async move {
// A positive offset at or past EOF is an unsatisfiable range; mimic
// the backend's 416 rather than yielding an empty body, so the
// pipeline's empty-tail disambiguation is exercised.
if from > 0 && from >= size {
return Err(UniversalIoError::S3Config {
description: "requested range not satisfiable".into(),
});
}
let tail = data.slice(from as usize..);
Ok((size, futures::stream::once(async move { Ok(tail) }).boxed()))
}
}
fn len(&self, _path: &Path) -> impl Future<Output = Result<u64>> + Send + 'static {
self.counters.len.fetch_add(1, Ordering::Relaxed);
let len = self.data.len() as u64;
async move { Ok(len) }
}
fn kind() -> UniversalKind {
UniversalKind::S3
}
}
const DATA: &[u8] = b"the quick brown fox jumps over the lazy dog";
#[test]
fn blob_file_read_whole_uses_single_get_without_head() {
let source = CountingSource::new(DATA);
let counters = source.counters.clone();
let file = BlobFile::new(source, BridgeRuntime::global(), "obj");
let bytes = file.read_whole::<u8>().expect("read_whole");
assert_eq!(&bytes[..], DATA);
assert_eq!(
counters.whole.load(Ordering::Relaxed),
1,
"read_whole should issue exactly one whole-object GET",
);
assert_eq!(
counters.len.load(Ordering::Relaxed),
0,
"read_whole must not issue a separate len/HEAD request",
);
assert_eq!(counters.range.load(Ordering::Relaxed), 0);
}
#[test]
fn owned_pipeline_tail_read_uses_single_get_without_head() {
let source = CountingSource::new(DATA);
let counters = source.counters.clone();
let file = BlobFile::new(source, BridgeRuntime::global(), "obj");
let mut pipeline =
<OwnedBlobPipeline<CountingSource, ()> as OwnedReadPipeline<()>>::new(file).unwrap();
let from = 10u64;
pipeline.schedule_whole((), from).unwrap();
let (_, bytes) = pipeline.wait().unwrap().expect("exactly one read");
assert_eq!(&bytes[..], &DATA[from as usize..]);
assert_eq!(
counters.from.load(Ordering::Relaxed),
1,
"a tail read should issue exactly one open-ended GET",
);
assert_eq!(
counters.len.load(Ordering::Relaxed),
0,
"a tail read must not issue a separate len/HEAD request",
);
assert_eq!(counters.whole.load(Ordering::Relaxed), 0);
assert_eq!(counters.range.load(Ordering::Relaxed), 0);
}
#[test]
fn owned_pipeline_empty_tail_resolves_to_empty_read() {
let source = CountingSource::new(DATA);
let counters = source.counters.clone();
let file = BlobFile::new(source, BridgeRuntime::global(), "obj");
let mut pipeline =
<OwnedBlobPipeline<CountingSource, ()> as OwnedReadPipeline<()>>::new(file).unwrap();
// Offset exactly at EOF: there is no tail to read.
pipeline.schedule_whole((), DATA.len() as u64).unwrap();
let (_, bytes) = pipeline
.wait()
.unwrap()
.expect("an (empty) read is scheduled");
assert!(bytes.is_empty(), "an offset at EOF yields an empty read");
// The open-ended GET is attempted once, then a single len() confirms the
// offset is past EOF — there is no cheaper way to learn the tail is empty.
assert_eq!(counters.from.load(Ordering::Relaxed), 1);
assert_eq!(counters.len.load(Ordering::Relaxed), 1);
assert_eq!(counters.whole.load(Ordering::Relaxed), 0);
assert_eq!(counters.range.load(Ordering::Relaxed), 0);
}
#[test]
fn disk_cache_read_whole_skips_remote_len() {
let tmp = tempfile::Builder::new()
.prefix("uio_whole_read")
.tempdir()
.unwrap();
let local_dir = tmp.path().to_path_buf();
let source = CountingSource::new(DATA);
let counters = source.counters.clone();
let config = DiskCacheConfig::new(PathBuf::from("bucket"), local_dir).unwrap();
let fs = DiskCacheFs::<BlobFile<CountingSource>>::from_context(DiskCacheFsContext {
config: Arc::new(config),
remote: source.config(),
})
.unwrap();
let file = fs
.open(
Path::new("bucket/data.bin"),
OpenOptions {
writeable: false,
populate: Populate::No,
..OpenOptions::new_for_test()
},
(),
)
.unwrap();
let bytes = file.read_whole::<u8>().expect("read_whole");
assert_eq!(&bytes[..], DATA);
assert_eq!(
counters.whole.load(Ordering::Relaxed),
1,
"disk cache read_whole should issue a single whole-object GET",
);
assert_eq!(
counters.len.load(Ordering::Relaxed),
0,
"disk cache read_whole must not HEAD the remote for its length",
);
let again = file
.read::<Sequential, u8>(ReadRange::new(0, DATA.len() as u64))
.expect("local read");
assert_eq!(&again[..], DATA);
assert_eq!(counters.whole.load(Ordering::Relaxed), 1);
assert_eq!(counters.range.load(Ordering::Relaxed), 0);
assert_eq!(counters.len.load(Ordering::Relaxed), 0);
}
#[test]
fn disk_cache_prefill_open_uses_whole_get_without_head() {
let tmp = tempfile::Builder::new()
.prefix("uio_whole_read")
.tempdir()
.unwrap();
let local_dir = tmp.path().to_path_buf();
let source = CountingSource::new(DATA);
let counters = source.counters.clone();
let config = DiskCacheConfig::new(PathBuf::from("bucket"), local_dir).unwrap();
let fs = DiskCacheFs::<BlobFile<CountingSource>>::from_context(DiskCacheFsContext {
config: Arc::new(config),
remote: source.config(),
})
.unwrap();
let file = fs
.open(
Path::new("bucket/data.bin"),
OpenOptions {
writeable: false,
populate: Populate::Blocking,
..OpenOptions::new_for_test()
},
(),
)
.unwrap();
assert_eq!(
counters.whole.load(Ordering::Relaxed),
1,
"prefill should issue a single whole-object GET",
);
assert_eq!(
counters.len.load(Ordering::Relaxed),
0,
"prefill must not HEAD the remote for its length",
);
assert_eq!(counters.range.load(Ordering::Relaxed), 0);
let bytes = file.read_whole::<u8>().expect("read_whole");
assert_eq!(&bytes[..], DATA);
assert_eq!(counters.whole.load(Ordering::Relaxed), 1);
assert_eq!(counters.len.load(Ordering::Relaxed), 0);
}