mirror of
https://github.com/qdrant/qdrant.git
synced 2026-09-21 13:37:46 -05:00
[UIO] renames + enforce LiveReload::live_preload (#10351)
* make LiveReload::live_preload required * rename `schedule_prefetch`->`schedule_open` * rename `reschedule_prefetch`->`reschedule_open`
This commit is contained in:
@@ -77,7 +77,7 @@ impl<S: UniversalRead> Pages<S> {
|
||||
if !page_files.contains(&page_path) {
|
||||
break;
|
||||
}
|
||||
fs.schedule_prefetch(&page_path, Some(page_open_options(populate, false)), None)?;
|
||||
fs.schedule_open(&page_path, Some(page_open_options(populate, false)), None)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -465,7 +465,7 @@ impl<S: UniversalRead> Pages<S> {
|
||||
let page_path = self.page_path(page_id);
|
||||
|
||||
// Re-schedule so that unchanged files don't re-fetch.
|
||||
fs.reschedule_prefetch(&page_path, Some(page_open_options(populate, false)), None)?;
|
||||
fs.reschedule_open(&page_path, Some(page_open_options(populate, false)), None)?;
|
||||
}
|
||||
|
||||
for page_id in next_page_id.. {
|
||||
@@ -473,7 +473,7 @@ impl<S: UniversalRead> Pages<S> {
|
||||
if !fs.exists(&page_path)? {
|
||||
break;
|
||||
}
|
||||
fs.schedule_prefetch(&page_path, Some(page_open_options(populate, false)), None)?;
|
||||
fs.schedule_open(&page_path, Some(page_open_options(populate, false)), None)?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
|
||||
@@ -60,7 +60,7 @@ impl<S: UniversalRead> AppendOnlyPages<S> {
|
||||
if !page_files.contains(&path) {
|
||||
break;
|
||||
}
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&path,
|
||||
Some(AppendOnlyPage::<S>::open_options(populate, false)),
|
||||
None,
|
||||
@@ -288,7 +288,7 @@ impl<S: UniversalRead> AppendOnlyPages<S> {
|
||||
if !page_list.contains_key(&path) {
|
||||
break;
|
||||
}
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&path,
|
||||
Some(AppendOnlyPage::<S>::open_options(populate, false)),
|
||||
None,
|
||||
|
||||
@@ -45,7 +45,7 @@ impl<V: Blob, S: UniversalRead> BlobstoreReader<V, S> {
|
||||
) -> Result<()> {
|
||||
// schedule config file, so the config read in `open` is served from the prefetch pool
|
||||
let config_path = base_path.join(CONFIG_FILENAME);
|
||||
fs.schedule_prefetch(&config_path, None, None)?;
|
||||
fs.schedule_open(&config_path, None, None)?;
|
||||
|
||||
// Don't read config now; instead, probe all modes and ignore not-found errors
|
||||
for mode in Mode::iter() {
|
||||
|
||||
@@ -74,7 +74,7 @@ impl<S: UniversalRead> AppendOnlyTracker<S> {
|
||||
dir: &Path,
|
||||
populate: Populate,
|
||||
) -> Result<()> {
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&Self::tracker_file_name(dir),
|
||||
Some(Self::open_options(populate, false)),
|
||||
None,
|
||||
|
||||
@@ -330,7 +330,7 @@ impl<S: UniversalRead> Tracker<S> {
|
||||
) -> Result<()> {
|
||||
// Default a lazy open to partially populating the header.
|
||||
let populate = populate.or_partial(0..size_of::<TrackerHeader>() as u64);
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
tracker_path,
|
||||
Some(tracker_open_options(populate, false)),
|
||||
None,
|
||||
|
||||
@@ -56,7 +56,7 @@ where
|
||||
// the perfect-hash table, so the map can be opened without a full read.
|
||||
options.populate = options.populate.or_partial(0..HEADER_AND_BASIC_PHF_SIZE);
|
||||
|
||||
fs.schedule_prefetch(path.as_ref(), Some(options), None)
|
||||
fs.schedule_open(path.as_ref(), Some(options), None)
|
||||
}
|
||||
|
||||
/// Load the hash map from file.
|
||||
|
||||
@@ -210,7 +210,7 @@ impl<Fs: UniversalReadFs> CachedReadFs for CachedFs<Fs> {
|
||||
self.files_prefetched.lock().clear();
|
||||
}
|
||||
|
||||
fn schedule_prefetch(
|
||||
fn schedule_open(
|
||||
&self,
|
||||
path: &Path,
|
||||
open_arguments: Option<OpenOptions>,
|
||||
@@ -240,7 +240,7 @@ impl<Fs: UniversalReadFs> CachedReadFs for CachedFs<Fs> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn reschedule_prefetch(
|
||||
fn reschedule_open(
|
||||
&self,
|
||||
path: &Path,
|
||||
open_arguments: Option<OpenOptions>,
|
||||
@@ -265,7 +265,7 @@ impl<Fs: UniversalReadFs> CachedReadFs for CachedFs<Fs> {
|
||||
}
|
||||
|
||||
// Otherwise schedule normally
|
||||
self.schedule_prefetch(path, open_arguments, open_extra)
|
||||
self.schedule_open(path, open_arguments, open_extra)
|
||||
}
|
||||
|
||||
fn cached_file_info(&self, path: &Path) -> Option<FileInfo> {
|
||||
|
||||
@@ -36,7 +36,7 @@ impl<S: UniversalRead> OneshotFile<S> {
|
||||
/// A one-shot file is always read in full, so the prefetch populates it —
|
||||
/// the fetch overlaps whatever runs between the schedule and the open.
|
||||
pub fn preopen<Fs: CachedReadFs<File = S>>(fs: &Fs, path: impl AsRef<Path>) -> UioResult<()> {
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
path.as_ref(),
|
||||
Some(Self::open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
|
||||
@@ -166,7 +166,7 @@ pub trait CachedReadFs: UniversalReadFs {
|
||||
/// Open `path` in the background and park the handle in the prefetch
|
||||
/// pool, to be consumed by a later [`UniversalReadFs::open`] of the same
|
||||
/// path. Idempotent per path while the handle is unconsumed.
|
||||
fn schedule_prefetch(
|
||||
fn schedule_open(
|
||||
&self,
|
||||
path: &Path,
|
||||
open_arguments: Option<OpenOptions>,
|
||||
@@ -177,7 +177,7 @@ pub trait CachedReadFs: UniversalReadFs {
|
||||
///
|
||||
/// This will force `Self::open` to return `UnchangedOpen` error if the file
|
||||
/// did not change its `FileInfo` in between snapshots.
|
||||
fn reschedule_prefetch(
|
||||
fn reschedule_open(
|
||||
&self,
|
||||
path: &Path,
|
||||
open_arguments: Option<OpenOptions>,
|
||||
|
||||
@@ -68,21 +68,21 @@ impl InMemoryBitvecFlags {
|
||||
match FlagsMode::detect(fs, directory)?.unwrap_or(FlagsMode::Dynamic) {
|
||||
FlagsMode::Dynamic => {
|
||||
// Status file
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&status_file(directory),
|
||||
Some(bitslice_open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
)?;
|
||||
|
||||
// Bitslice
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&directory.join(FLAGS_FILE),
|
||||
Some(bitslice_open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
)?;
|
||||
}
|
||||
FlagsMode::Compact => {
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&directory.join(COMPACT_FLAGS_FILE),
|
||||
Some(compact_open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
@@ -101,19 +101,19 @@ impl InMemoryBitvecFlags {
|
||||
|
||||
match mode {
|
||||
FlagsMode::Dynamic => {
|
||||
fs.reschedule_prefetch(
|
||||
fs.reschedule_open(
|
||||
&status_file(directory),
|
||||
Some(bitslice_open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
)?;
|
||||
fs.reschedule_prefetch(
|
||||
fs.reschedule_open(
|
||||
&directory.join(FLAGS_FILE),
|
||||
Some(bitslice_open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
)?;
|
||||
}
|
||||
FlagsMode::Compact => {
|
||||
fs.reschedule_prefetch(
|
||||
fs.reschedule_open(
|
||||
&directory.join(COMPACT_FLAGS_FILE),
|
||||
Some(compact_open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
|
||||
@@ -73,7 +73,7 @@ impl<S: UniversalRead> ReadOnlyCompactFlags<S> {
|
||||
};
|
||||
|
||||
Ok(fs
|
||||
.schedule_prefetch(
|
||||
.schedule_open(
|
||||
&directory.join(COMPACT_FLAGS_FILE),
|
||||
Some(open_options(populate)),
|
||||
None,
|
||||
@@ -136,7 +136,7 @@ impl<S: UniversalRead> ReadOnlyCompactFlags<S> {
|
||||
} else {
|
||||
Populate::Partial(ReadRange::new(0, stored_bitmask::HEADER_SIZE as u64))
|
||||
};
|
||||
cached_fs.reschedule_prefetch(
|
||||
cached_fs.reschedule_open(
|
||||
&directory.join(COMPACT_FLAGS_FILE),
|
||||
Some(open_options(populate)),
|
||||
None,
|
||||
|
||||
@@ -85,7 +85,7 @@ impl<S: UniversalRead> ReadOnlyRoaringFlags<S> {
|
||||
) -> OperationResult<bool> {
|
||||
// Status file.
|
||||
if fs
|
||||
.schedule_prefetch(
|
||||
.schedule_open(
|
||||
&status_file(directory),
|
||||
Some(open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
@@ -97,7 +97,7 @@ impl<S: UniversalRead> ReadOnlyRoaringFlags<S> {
|
||||
}
|
||||
|
||||
// Bitslice
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&directory.join(FLAGS_FILE),
|
||||
Some(open_options(populate)),
|
||||
None,
|
||||
@@ -170,14 +170,14 @@ impl<S: UniversalRead> ReadOnlyRoaringFlags<S> {
|
||||
Populate::No
|
||||
};
|
||||
|
||||
cached_fs.reschedule_prefetch(
|
||||
cached_fs.reschedule_open(
|
||||
&directory.join(FLAGS_FILE),
|
||||
Some(open_options(populate)),
|
||||
None,
|
||||
)?;
|
||||
|
||||
// Status file.
|
||||
cached_fs.reschedule_prefetch(
|
||||
cached_fs.reschedule_open(
|
||||
&status_file(directory),
|
||||
Some(open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
|
||||
@@ -31,14 +31,13 @@ use crate::common::operation_error::OperationResult;
|
||||
pub(crate) trait LiveReload {
|
||||
type File: UniversalRead;
|
||||
|
||||
/// Stage everything the next [`Self::live_reload`] needs: schedule
|
||||
/// reopens on kept handles, (re)schedule prefetches for swapped and new
|
||||
/// files. Shared access; must not wait on any fetch.
|
||||
fn live_preload<Fs: CachedReadFs<File = Self::File>>(
|
||||
&self,
|
||||
cached_fs: &Fs,
|
||||
) -> OperationResult<()> {
|
||||
let _ = cached_fs;
|
||||
// todo(uio): don't provide default implementation
|
||||
Ok(())
|
||||
}
|
||||
) -> OperationResult<()>;
|
||||
|
||||
fn live_reload<Fs: UniversalReadFs<File = Self::File>>(
|
||||
&mut self,
|
||||
|
||||
@@ -47,8 +47,8 @@ impl<S: UniversalRead> ReadOnlyDiskIdTracker<S> {
|
||||
}
|
||||
|
||||
let options = Self::open_options();
|
||||
fs.schedule_prefetch(&version_mapping_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(&version_mapping_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_open(
|
||||
&deleted_path(segment_path),
|
||||
Some(Self::deleted_open_options()),
|
||||
None,
|
||||
|
||||
@@ -14,7 +14,7 @@ impl<S: UniversalRead> ReadOnlyDiskIdTracker<S> {
|
||||
/// Stage the fresh deleted-bitslice handle [`live_reload`](Self::live_reload) swaps in.
|
||||
pub fn live_preload(&self, fs: &impl CachedReadFs<File = S>) -> OperationResult<()> {
|
||||
// The reload reads the whole bitslice
|
||||
fs.reschedule_prefetch(
|
||||
fs.reschedule_open(
|
||||
&deleted_path(&self.path),
|
||||
Some(Self::deleted_open_options()),
|
||||
None,
|
||||
|
||||
@@ -58,9 +58,9 @@ impl<S: UniversalRead> DiskMappingReader<S> {
|
||||
|
||||
let options = Self::open_options();
|
||||
|
||||
fs.schedule_prefetch(&i2e_path, Some(options), None)?;
|
||||
fs.schedule_prefetch(&e2i_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(&i2e_path, Some(options), None)?;
|
||||
fs.schedule_open(&e2i_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_open(
|
||||
&is_uuid_path(segment_path),
|
||||
Some(OpenOptions {
|
||||
// Prefetch must not stall on population; only the consuming
|
||||
|
||||
@@ -43,9 +43,9 @@ impl<S: UniversalRead> ReadOnlyImmutableIdTracker<S> {
|
||||
|
||||
let options = Self::open_options();
|
||||
|
||||
fs.schedule_prefetch(&deleted_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_prefetch(&version_mapping_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_prefetch(&mappings_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_open(&deleted_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_open(&version_mapping_path(segment_path), Some(options), None)?;
|
||||
fs.schedule_open(&mappings_path(segment_path), Some(options), None)?;
|
||||
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
@@ -10,7 +10,7 @@ use crate::id_tracker::mutable_id_tracker::read_only::LiveReloadResult;
|
||||
impl<S: UniversalRead> ReadOnlyImmutableIdTracker<S> {
|
||||
/// Stage the fresh deleted-bitslice handle [`live_reload`](Self::live_reload) swaps in.
|
||||
pub fn live_preload(&self, fs: &impl CachedReadFs<File = S>) -> OperationResult<()> {
|
||||
fs.reschedule_prefetch(&deleted_path(&self.path), Some(Self::open_options()), None)?;
|
||||
fs.reschedule_open(&deleted_path(&self.path), Some(Self::open_options()), None)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -33,9 +33,9 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
|
||||
pub fn preopen(fs: &impl CachedReadFs<File = S>, segment_path: &Path) -> OperationResult<()> {
|
||||
let options = Self::open_options();
|
||||
|
||||
fs.schedule_prefetch(&mappings_path(segment_path), Some(options), None)
|
||||
fs.schedule_open(&mappings_path(segment_path), Some(options), None)
|
||||
.ok_not_found()?;
|
||||
fs.schedule_prefetch(&versions_path(segment_path), Some(options), None)
|
||||
fs.schedule_open(&versions_path(segment_path), Some(options), None)
|
||||
.ok_not_found()?;
|
||||
|
||||
Ok(())
|
||||
|
||||
@@ -73,7 +73,7 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
|
||||
.schedule_reopen(|p| fs.cached_file_info(p))
|
||||
.ok_not_found()?,
|
||||
None => fs
|
||||
.schedule_prefetch(&path, Some(options), None)
|
||||
.schedule_open(&path, Some(options), None)
|
||||
.ok_not_found()?,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -177,13 +177,13 @@ where
|
||||
S: UniversalRead,
|
||||
{
|
||||
if fs
|
||||
.schedule_prefetch(&deleted_mask_path(dir), Some(options), None)
|
||||
.schedule_open(&deleted_mask_path(dir), Some(options), None)
|
||||
.ok_not_found()?
|
||||
.is_some()
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
fs.schedule_prefetch(&dir.join(legacy_file), Some(options), None)?;
|
||||
fs.schedule_open(&dir.join(legacy_file), Some(options), None)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
+2
-2
@@ -157,7 +157,7 @@ impl<S: UniversalRead> OnDiskInvertedIndex<S> {
|
||||
// Postings.
|
||||
let postings_path = path.join(POSTINGS_FILE);
|
||||
if fs
|
||||
.schedule_prefetch(
|
||||
.schedule_open(
|
||||
&postings_path,
|
||||
Some(Self::open_options(
|
||||
populate,
|
||||
@@ -180,7 +180,7 @@ impl<S: UniversalRead> OnDiskInvertedIndex<S> {
|
||||
)?;
|
||||
|
||||
// Point to tokens count
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&path.join(POINT_TO_TOKENS_COUNT_FILE),
|
||||
Some(Self::open_options(populate, AdviceSetting::Global)),
|
||||
None,
|
||||
|
||||
@@ -166,7 +166,7 @@ impl<S: UniversalRead> OnDiskGeoIndex<S> {
|
||||
// Stats
|
||||
let stats_path = path.join(STATS_PATH);
|
||||
if fs
|
||||
.schedule_prefetch(&stats_path, None, None)
|
||||
.schedule_open(&stats_path, None, None)
|
||||
.ok_not_found()?
|
||||
.is_none()
|
||||
{
|
||||
@@ -176,17 +176,17 @@ impl<S: UniversalRead> OnDiskGeoIndex<S> {
|
||||
|
||||
// Geohash counts, points map, and point-id list
|
||||
let options = Self::open_options(populate);
|
||||
fs.schedule_prefetch(&path.join(COUNTS_PER_HASH), Some(options), None)?;
|
||||
fs.schedule_prefetch(&path.join(POINTS_MAP), Some(options), None)?;
|
||||
fs.schedule_prefetch(&path.join(POINTS_MAP_IDS), Some(options), None)?;
|
||||
fs.schedule_open(&path.join(COUNTS_PER_HASH), Some(options), None)?;
|
||||
fs.schedule_open(&path.join(POINTS_MAP), Some(options), None)?;
|
||||
fs.schedule_open(&path.join(POINTS_MAP_IDS), Some(options), None)?;
|
||||
|
||||
// Block indexes over the two sorted arrays; optional, absent on old
|
||||
// segments
|
||||
let _ = fs
|
||||
.schedule_prefetch(&path.join(COUNTS_PER_HASH_BLOCK_INDEX), None, None)
|
||||
.schedule_open(&path.join(COUNTS_PER_HASH_BLOCK_INDEX), None, None)
|
||||
.ok_not_found()?;
|
||||
let _ = fs
|
||||
.schedule_prefetch(&path.join(POINTS_MAP_BLOCK_INDEX), None, None)
|
||||
.schedule_open(&path.join(POINTS_MAP_BLOCK_INDEX), None, None)
|
||||
.ok_not_found()?;
|
||||
|
||||
// Point to values
|
||||
|
||||
@@ -54,8 +54,8 @@ impl<T: Numericable + Serialize + DeserializeOwned> Histogram<T> {
|
||||
|
||||
/// Schedule background prefetch of the two files [`open`](Self::open) reads.
|
||||
pub fn preopen(fs: &impl CachedReadFs, path: &Path) -> OperationResult<()> {
|
||||
fs.schedule_prefetch(&path.join(CONFIG_PATH), None, None)?;
|
||||
fs.schedule_prefetch(&path.join(BORDERS_PATH), None, None)?;
|
||||
fs.schedule_open(&path.join(CONFIG_PATH), None, None)?;
|
||||
fs.schedule_open(&path.join(BORDERS_PATH), None, None)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -47,7 +47,7 @@ where
|
||||
// Config
|
||||
let config_path = path.join(CONFIG_PATH);
|
||||
if fs
|
||||
.schedule_prefetch(&config_path, None, None)
|
||||
.schedule_open(&config_path, None, None)
|
||||
.ok_not_found()?
|
||||
.is_none()
|
||||
{
|
||||
|
||||
@@ -78,7 +78,7 @@ impl<S: UniversalRead> PrefixIndex<S> {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
fs.schedule_prefetch(&file_path, Some(Self::prefix_open_options(populate)), None)?;
|
||||
fs.schedule_open(&file_path, Some(Self::prefix_open_options(populate)), None)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -120,7 +120,7 @@ where
|
||||
// Config
|
||||
let config_path = path.join(CONFIG_PATH);
|
||||
if fs
|
||||
.schedule_prefetch(&config_path, None, None)
|
||||
.schedule_open(&config_path, None, None)
|
||||
.ok_not_found()?
|
||||
.is_none()
|
||||
{
|
||||
@@ -133,11 +133,11 @@ where
|
||||
|
||||
// Value pairs
|
||||
let pairs_path = path.join(PAIRS_PATH);
|
||||
fs.schedule_prefetch(&pairs_path, Some(Self::open_options(populate)), None)?;
|
||||
fs.schedule_open(&pairs_path, Some(Self::open_options(populate)), None)?;
|
||||
|
||||
// Block index over the value pairs; optional, absent on old segments
|
||||
let _ = fs
|
||||
.schedule_prefetch(&path.join(PAIRS_BLOCK_INDEX_PATH), None, None)
|
||||
.schedule_open(&path.join(PAIRS_BLOCK_INDEX_PATH), None, None)
|
||||
.ok_not_found()?;
|
||||
|
||||
// Point to values
|
||||
|
||||
@@ -184,7 +184,7 @@ where
|
||||
) -> OperationResult<()> {
|
||||
let file_name = dir.join(POINT_TO_VALUES_PATH);
|
||||
|
||||
fs.schedule_prefetch(&file_name, Some(Self::open_options(populate)), None)?;
|
||||
fs.schedule_open(&file_name, Some(Self::open_options(populate)), None)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -116,7 +116,7 @@ impl<S: UniversalRead> HnswGraph<S> {
|
||||
dir: &Path,
|
||||
residency: GraphLinksResidency,
|
||||
) -> OperationResult<()> {
|
||||
fs.schedule_prefetch(&GraphLayers::get_path(dir), None, None)?;
|
||||
fs.schedule_open(&GraphLayers::get_path(dir), None, None)?;
|
||||
let Some(format) = GraphLayers::probe_links_format(fs, dir)? else {
|
||||
return Ok(());
|
||||
};
|
||||
@@ -125,7 +125,7 @@ impl<S: UniversalRead> HnswGraph<S> {
|
||||
} else {
|
||||
GraphLinks::preopen_options(residency)
|
||||
};
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&GraphLayers::get_links_path(dir, format),
|
||||
Some(options),
|
||||
None,
|
||||
|
||||
@@ -102,7 +102,7 @@ impl<S: UniversalReadExt> ReadOnlyHNSWIndex<S> {
|
||||
populate_override: Option<Populate>,
|
||||
) -> OperationResult<()> {
|
||||
// Graph config; may legitimately be absent (`open` derives defaults).
|
||||
fs.schedule_prefetch(&HnswGraphConfig::get_config_path(path), None, None)
|
||||
fs.schedule_open(&HnswGraphConfig::get_config_path(path), None, None)
|
||||
.ok_not_found()?;
|
||||
|
||||
// Graph data and links
|
||||
|
||||
@@ -135,7 +135,7 @@ impl<S: UniversalReadExt + 'static> VectorIndexReadEnum<S> {
|
||||
// Sparse index config; `open_sparse` reads it off the parked handle.
|
||||
let config_path = SparseIndexConfig::get_config_path(path);
|
||||
if fs
|
||||
.schedule_prefetch(&config_path, None, None)
|
||||
.schedule_open(&config_path, None, None)
|
||||
.ok_not_found()?
|
||||
.is_none()
|
||||
{
|
||||
@@ -165,10 +165,10 @@ impl<S: UniversalReadExt + 'static> VectorIndexReadEnum<S> {
|
||||
inverted_index_compressed_mmap::preopen(fs, path, populate)?;
|
||||
|
||||
// Version check
|
||||
fs.schedule_prefetch(&path.join(VERSION_FILE), None, None)?;
|
||||
fs.schedule_open(&path.join(VERSION_FILE), None, None)?;
|
||||
|
||||
// Indices tracker
|
||||
fs.schedule_prefetch(&IndicesTracker::file_path(path), None, None)?;
|
||||
fs.schedule_open(&IndicesTracker::file_path(path), None, None)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -46,13 +46,13 @@ fn build_cached_fs<Fs: UniversalReadFs>(
|
||||
// Absence is tolerated here: the subsequent read reports it gracefully.
|
||||
for file_name in [VERSION_FILE, SEGMENT_STATE_FILE] {
|
||||
cached_fs
|
||||
.schedule_prefetch(&segment_path.join(file_name), None, None)
|
||||
.schedule_open(&segment_path.join(file_name), None, None)
|
||||
.ok_not_found()?;
|
||||
}
|
||||
|
||||
// Payload index config
|
||||
cached_fs
|
||||
.schedule_prefetch(
|
||||
.schedule_open(
|
||||
&PayloadConfig::get_config_path(&get_payload_index_path(segment_path)),
|
||||
None,
|
||||
None,
|
||||
|
||||
@@ -47,7 +47,7 @@ fn build_cached_fs<Fs: UniversalReadFs>(
|
||||
// Absence is tolerated here: the subsequent read reports it gracefully.
|
||||
for file_name in [VERSION_FILE, SEGMENT_STATE_FILE] {
|
||||
cached_fs
|
||||
.schedule_prefetch(&segment_path.join(file_name), None, None)
|
||||
.schedule_open(&segment_path.join(file_name), None, None)
|
||||
.ok_not_found()?;
|
||||
}
|
||||
|
||||
|
||||
@@ -21,10 +21,10 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> ReadOnlyChunkedVectors<T, S> {
|
||||
populate: Populate,
|
||||
) -> OperationResult<()> {
|
||||
// Config file
|
||||
fs.schedule_prefetch(&config_file(directory), None, None)?;
|
||||
fs.schedule_open(&config_file(directory), None, None)?;
|
||||
|
||||
// Status file
|
||||
fs.schedule_prefetch(&status_file(directory), None, None)?;
|
||||
fs.schedule_open(&status_file(directory), None, None)?;
|
||||
|
||||
// Chunks
|
||||
preopen_chunks(fs, directory, advice, populate)?;
|
||||
@@ -123,7 +123,7 @@ fn preopen_chunks(
|
||||
.is_some();
|
||||
|
||||
if is_chunk {
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&listed.path,
|
||||
Some(chunk_open_options(advice, populate, false)),
|
||||
None,
|
||||
|
||||
@@ -16,7 +16,7 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> LiveReload for ReadOnlyChunkedVe
|
||||
|
||||
fn live_preload<Fs: CachedReadFs<File = S>>(&self, fs: &Fs) -> OperationResult<()> {
|
||||
// Status is the change signal, let reload skip reloading if this didn't change.
|
||||
fs.reschedule_prefetch(&status_file(&self.directory), None, None)?;
|
||||
fs.reschedule_open(&status_file(&self.directory), None, None)?;
|
||||
|
||||
let num_files = list_chunk_files(fs, &self.directory)?.len();
|
||||
|
||||
@@ -25,7 +25,7 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> LiveReload for ReadOnlyChunkedVe
|
||||
let last_chunk = self.config.get_chunk_index(self.len);
|
||||
|
||||
let fresh_from = if last_chunk < self.chunks.len().min(num_files) {
|
||||
fs.reschedule_prefetch(
|
||||
fs.reschedule_open(
|
||||
&chunk_name(&self.directory, last_chunk),
|
||||
Some(chunk_open_options(self.advice, self.populate, false)),
|
||||
None,
|
||||
@@ -37,7 +37,7 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> LiveReload for ReadOnlyChunkedVe
|
||||
|
||||
// Prefetch the rest of the chunks the reload may open.
|
||||
for chunk_id in fresh_from..num_files {
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&chunk_name(&self.directory, chunk_id),
|
||||
Some(chunk_open_options(self.advice, self.populate, false)),
|
||||
None,
|
||||
|
||||
@@ -58,7 +58,7 @@ impl<T: PrimitiveVectorElement, S: UniversalRead> ImmutableDenseVectorData<T, S>
|
||||
populate: Populate,
|
||||
) -> OperationResult<()> {
|
||||
// Vector data
|
||||
fs.schedule_prefetch(vectors_path, Some(Self::open_options(populate)), None)?;
|
||||
fs.schedule_open(vectors_path, Some(Self::open_options(populate)), None)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -39,7 +39,7 @@ impl<T: PrimitiveVectorElement, S: UniversalRead> ReadOnlyImmutableDenseVectorSt
|
||||
ImmutableDenseVectorData::<T, S>::preopen(fs, &path.join(VECTORS_PATH), populate)?;
|
||||
|
||||
// Deleted flags
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&path.join(DELETED_PATH),
|
||||
Some(bitslice_open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
|
||||
@@ -60,7 +60,7 @@ impl MultivectorOffsetsStorageRam {
|
||||
///
|
||||
/// The open reads the whole file, so the prefetch populates it.
|
||||
pub fn preopen(fs: &impl CachedReadFs, path: &Path) -> OperationResult<()> {
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
path,
|
||||
Some(Self::open_options(Populate::PreferBackground)),
|
||||
None,
|
||||
@@ -181,7 +181,7 @@ impl<S: UniversalRead> MultivectorOffsetsStorageMmap<S> {
|
||||
path: &Path,
|
||||
populate: Populate,
|
||||
) -> OperationResult<()> {
|
||||
fs.schedule_prefetch(path, Some(Self::open_options(populate)), None)?;
|
||||
fs.schedule_open(path, Some(Self::open_options(populate)), None)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -183,7 +183,7 @@ impl<S: UniversalRead> QuantizedStorage<S> {
|
||||
path: &Path,
|
||||
populate: Populate,
|
||||
) -> OperationResult<()> {
|
||||
fs.schedule_prefetch(path, Some(Self::open_options(populate)), None)?;
|
||||
fs.schedule_open(path, Some(Self::open_options(populate)), None)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -65,7 +65,7 @@ impl<S: UniversalRead> ReadOnlyQuantizedVectors<S> {
|
||||
// Config; `open` reads it off the parked handle.
|
||||
let config_path = QuantizedVectors::get_config_path(path);
|
||||
if fs
|
||||
.schedule_prefetch(&config_path, None, None)
|
||||
.schedule_open(&config_path, None, None)
|
||||
.ok_not_found()?
|
||||
.is_none()
|
||||
{
|
||||
@@ -73,7 +73,7 @@ impl<S: UniversalRead> ReadOnlyQuantizedVectors<S> {
|
||||
}
|
||||
|
||||
// Per-method metadata
|
||||
fs.schedule_prefetch(&QuantizedVectors::get_meta_path(path), None, None)?;
|
||||
fs.schedule_open(&QuantizedVectors::get_meta_path(path), None, None)?;
|
||||
|
||||
let placement = QuantizedVectors::memory_placement(
|
||||
quantization_config.memory_placement(),
|
||||
|
||||
@@ -61,10 +61,10 @@ fn index_open_options(populate: Populate) -> OpenOptions {
|
||||
/// (mmap).
|
||||
pub fn preopen(fs: &impl CachedReadFs, path: &Path, populate: Populate) -> UioResult<()> {
|
||||
// Config
|
||||
fs.schedule_prefetch(&index_config_file_path(path), None, None)?;
|
||||
fs.schedule_open(&index_config_file_path(path), None, None)?;
|
||||
|
||||
// Index data
|
||||
fs.schedule_prefetch(
|
||||
fs.schedule_open(
|
||||
&index_file_path(path),
|
||||
Some(index_open_options(populate)),
|
||||
None,
|
||||
|
||||
Reference in New Issue
Block a user