From 34a45d9f59ba39ad566fc5c9acf5f0fe132300d5 Mon Sep 17 00:00:00 2001 From: Alexander Droste Date: Fri, 23 Oct 2026 12:18:15 +0100 Subject: [PATCH] perf(cuda): reuse file mappings without breaking cold scans Cache mappings by file identity and version so repeated scans reuse filled page tables. Add Rust and C APIs to release cached mappings before cold-cache eviction while active readers retain their mappings. Include reuse, versioning, release, and concurrent-read coverage. Original-commit: a31e77427e Signed-off-by: Alexander Droste --- vortex-cuda/ffi/cinclude/vortex_cuda.h | 10 ++ vortex-cuda/ffi/src/lib.rs | 11 ++ vortex-cuda/src/lib.rs | 1 + vortex-cuda/src/pooled_read_at.rs | 1 + vortex-cuda/src/pooled_read_at/file.rs | 167 +++++++++++++++---- vortex-cuda/src/pooled_read_at/file/tests.rs | 93 +++++++++++ 6 files changed, 254 insertions(+), 29 deletions(-) diff --git a/vortex-cuda/ffi/cinclude/vortex_cuda.h b/vortex-cuda/ffi/cinclude/vortex_cuda.h index 2fb8ae8054e..5daccce6663 100644 --- a/vortex-cuda/ffi/cinclude/vortex_cuda.h +++ b/vortex-cuda/ffi/cinclude/vortex_cuda.h @@ -86,6 +86,16 @@ typedef struct vx_cuda_scan_options { extern "C" { #endif // __cplusplus +/** + * Release Vortex's cached file mappings that no open scan uses. + * + * Scans keep their files mapped so later scans of the same file reuse filled page tables. The + * operating system cannot evict mapped pages from the page cache, so call this before evicting + * files, for example before a cold-cache measurement. Mappings still used by open streams stay + * mapped until those streams are released. + */ +void vx_cuda_release_file_mappings(void); + /** * Create a CUDA Vortex session. * diff --git a/vortex-cuda/ffi/src/lib.rs b/vortex-cuda/ffi/src/lib.rs index 7be1c884635..f34e4408ef4 100644 --- a/vortex-cuda/ffi/src/lib.rs +++ b/vortex-cuda/ffi/src/lib.rs @@ -78,6 +78,17 @@ pub struct vx_cuda_scan_options { pub batch_rows: usize, } +/// Release Vortex's cached file mappings that no open scan uses. +/// +/// Scans keep their files mapped so later scans of the same file reuse filled page tables. The +/// operating system cannot evict mapped pages from the page cache, so call this before evicting +/// files, for example before a cold-cache measurement. Mappings still used by open streams stay +/// mapped until those streams are released. +#[unsafe(no_mangle)] +pub extern "C-unwind" fn vx_cuda_release_file_mappings() { + vortex_cuda::release_file_mappings(); +} + fn session_with_cuda(session: &VortexSession) -> &VortexSession { session.get::(); register_cuda_layout(session); diff --git a/vortex-cuda/src/lib.rs b/vortex-cuda/src/lib.rs index 5a13a2f0f54..f2e22433418 100644 --- a/vortex-cuda/src/lib.rs +++ b/vortex-cuda/src/lib.rs @@ -67,6 +67,7 @@ pub use pooled_read_at::PooledByteBufferReadAt; pub use pooled_read_at::PooledFileReadAt; pub use pooled_read_at::PooledFileReadAtOptions; pub use pooled_read_at::PooledObjectStoreReadAt; +pub use pooled_read_at::release_file_mappings; pub use session::CudaSession; pub use session::CudaSessionExt; pub use session::DictionaryExport; diff --git a/vortex-cuda/src/pooled_read_at.rs b/vortex-cuda/src/pooled_read_at.rs index ecaa8a332e2..96b8697090c 100644 --- a/vortex-cuda/src/pooled_read_at.rs +++ b/vortex-cuda/src/pooled_read_at.rs @@ -33,6 +33,7 @@ mod file; pub use file::PooledFileReadAt; pub use file::PooledFileReadAtOptions; +pub use file::release_file_mappings; /// Default number of concurrent requests to allow for object store I/O. pub const DEFAULT_OBJECT_STORE_CONCURRENCY: usize = 192; diff --git a/vortex-cuda/src/pooled_read_at/file.rs b/vortex-cuda/src/pooled_read_at/file.rs index 05f2ffc0471..afa8d736f2e 100644 --- a/vortex-cuda/src/pooled_read_at/file.rs +++ b/vortex-cuda/src/pooled_read_at/file.rs @@ -4,11 +4,15 @@ #[cfg(target_os = "linux")] mod direct; +use std::collections::VecDeque; use std::fs::File; use std::mem::ManuallyDrop; use std::ops::Range; +use std::os::unix::fs::MetadataExt; use std::path::Path; use std::sync::Arc; +use std::sync::LazyLock; +use std::time::SystemTime; use cudarc::driver::CudaFunction; use cudarc::driver::CudaSlice; @@ -25,7 +29,6 @@ use futures::stream; #[cfg(target_os = "linux")] use memmap2::Advice; use memmap2::Mmap; -#[cfg(target_os = "linux")] use parking_lot::Mutex; use tokio::sync::Semaphore; use vortex::array::buffer::BufferHandle; @@ -58,8 +61,7 @@ const FILE_READ_CHUNK_BYTES: usize = 8 << 20; const MAPPED_COPY_BLOCKS_MAX: u64 = 1024; const MAPPED_COPY_THREADS: u32 = 256; -/// Filling page tables from several threads contends on the process memory map and measured -/// slower than a single thread, so fills are serialized across all mapped files. +/// Serialize page-table fills across all mapped files, releasing the lock after each chunk. #[cfg(target_os = "linux")] static MAPPED_FILL: Mutex<()> = Mutex::new(()); @@ -135,6 +137,128 @@ unsafe extern "C" { fn getpagesize() -> std::ffi::c_int; } +fn page_size() -> VortexResult { + // SAFETY: getpagesize has no preconditions. + let page = usize::try_from(unsafe { getpagesize() })?; + vortex_ensure!(page.is_power_of_two(), "invalid page size {page}"); + Ok(page) +} + +/// Number of recently opened file versions whose mappings stay alive between opens. +const FILE_MAPPINGS_CACHED_MAX: usize = 64; + +/// Mappings of recently opened file versions, least recently opened first. +/// +/// Each open used to map its file and unmap it after the scan. Unmapping holds the process's +/// address-space lock for milliseconds per GiB of filled page tables, and pageable CUDA copies +/// on other threads wait for that lock while holding the driver's: on NDS-H SF10 Q10, every +/// thread stalled for the first ~2 ms of each query. Reusing the mapping also keeps its page +/// tables filled, so hot scans skip most of the fill work. +static FILE_MAPPINGS: LazyLock> = + LazyLock::new(|| Mutex::new(VecDeque::with_capacity(FILE_MAPPINGS_CACHED_MAX + 1))); + +/// The entries of [`FILE_MAPPINGS`]. +type FileMappingCache = VecDeque<(FileVersion, Arc)>; + +/// A file's identity and version: modifying or replacing a file yields a new mapping. +/// +/// Modifying a mapped file in place without changing its size within the timestamp +/// granularity is not detected; Vortex files are immutable once written. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct FileVersion { + device: u64, + inode: u64, + len: u64, + modified: SystemTime, +} + +impl FileVersion { + fn of(file: &File) -> VortexResult { + let metadata = file.metadata()?; + Ok(Self { + device: metadata.dev(), + inode: metadata.ino(), + len: metadata.len(), + modified: metadata.modified()?, + }) + } +} + +/// A read-only mapping of one file version, shared by every open of that version. +struct FileMapping { + /// Present until drop; unmapped only after every user's copies have drained. + mapping: Option, +} + +impl FileMapping { + /// The cached mapping of `path`'s current version, mapping it on first use. + fn open(path: &Path) -> VortexResult> { + let file = File::open(path)?; + let version = FileVersion::of(&file)?; + let mut mappings = FILE_MAPPINGS.lock(); + if let Some(index) = mappings.iter().position(|(cached, _)| *cached == version) { + let entry = mappings.remove(index).vortex_expect("index is in bounds"); + let mapping = Arc::clone(&entry.1); + mappings.push_back(entry); + return Ok(mapping); + } + // SAFETY: Vortex files are immutable while open. As with any file mapping, truncation + // by another process would fault on access. + let mapping = unsafe { Mmap::map(&file)? }; + let mapping = Arc::new(Self { + mapping: Some(mapping), + }); + mappings.push_back((version, Arc::clone(&mapping))); + if mappings.len() > FILE_MAPPINGS_CACHED_MAX { + // Users still holding the evicted mapping keep it alive until their copies drain. + drop(mappings.pop_front()); + } + Ok(mapping) + } + + fn mapping(&self) -> &Mmap { + self.mapping + .as_ref() + .vortex_expect("mapping is present until drop") + } +} + +/// Release cached file mappings that no open reader uses, unmapping them before returning. +/// +/// Readers keep files mapped between scans, so later scans of a file reuse its filled page +/// tables. A mapped page cannot be dropped from the page cache, for example by +/// `posix_fadvise(DONTNEED)`, so release the mappings before evicting files, such as before a +/// cold-cache measurement, or to return the address space. Mappings that open readers still use +/// stay mapped until those readers close. +pub fn release_file_mappings() { + let released: Vec<_> = FILE_MAPPINGS + .lock() + .drain(..) + .map(|(_, mapping)| mapping) + .collect(); + for mapping in released { + if let Ok(mut mapping) = Arc::try_unwrap(mapping) { + // Unmap on this thread, not in the background, so the pages are unmapped on return. + drop(mapping.mapping.take()); + } + } +} + +impl Drop for FileMapping { + fn drop(&mut self) { + let Some(mapping) = self.mapping.take() else { + return; + }; + // Unmapping filled page tables costs milliseconds per GiB; keep it off the caller. If + // the thread cannot start, the returned closure unmaps here. + drop( + std::thread::Builder::new() + .name("vortex-unmap".to_string()) + .spawn(move || drop(mapping)), + ); + } +} + /// A buffered file whose page-cache pages the device copies directly. /// /// A buffered `pread` copies cached pages into pinned staging on the CPU, the slowest step of a @@ -142,8 +266,8 @@ unsafe extern "C" { /// from a mapping of those pages with a kernel. The CPU then only fills page tables, which is /// several times faster than copying the data. struct MappedFile { - /// Present until drop; unmapped only after the copy stream has drained. - mapping: Option, + /// Present until drop; released only after the copy stream has drained. + file: Option>, copy: CudaFunction, stream: VortexCudaStream, } @@ -166,22 +290,19 @@ impl MappedFile { if uses_host_page_tables != 1 { return Ok(None); } - let file = File::open(path)?; - // SAFETY: Vortex files are immutable while open. As with any file mapping, truncation - // by another process would fault on access. - let mapping = unsafe { Mmap::map(&file)? }; let copy = session.load_function_with_suffixes("mapped_copy", &[])?; Ok(Some(Self { - mapping: Some(mapping), + file: Some(FileMapping::open(path)?), copy, stream: stream.clone(), })) } fn mapping(&self) -> &Mmap { - self.mapping + self.file .as_ref() - .vortex_expect("mapping is present until drop") + .vortex_expect("file is present until drop") + .mapping() } /// Return the in-bounds file range if every overlapping page is in the page cache. @@ -197,11 +318,7 @@ impl MappedFile { if bytes.is_empty() { return Some(range); } - // SAFETY: getpagesize has no preconditions. - let page = usize::try_from(unsafe { getpagesize() }).ok()?; - if !page.is_power_of_two() { - return None; - } + let page = page_size().ok()?; let begin = (bytes.as_ptr() as usize) & !(page - 1); let end = bytes.as_ptr() as usize + bytes.len(); let mut resident = vec![0u8; (end - begin).div_ceil(page)]; @@ -269,22 +386,14 @@ impl MappedFile { impl Drop for MappedFile { fn drop(&mut self) { - let Some(mapping) = self.mapping.take() else { + let Some(file) = self.file.take() else { return; }; - // Queued copies may still read the mapping. + // Queued copies may still read the mapping, which the last release unmaps. if self.stream.synchronize().is_err() { // Without proof the copies finished, never unmap memory the device may read. - let _leaked = ManuallyDrop::new(mapping); - return; + let _leaked = ManuallyDrop::new(file); } - // Unmapping filled page tables costs milliseconds per GiB; keep it off the caller. If - // the thread cannot start, the returned closure unmaps here. - drop( - std::thread::Builder::new() - .name("vortex-unmap".to_string()) - .spawn(move || drop(mapping)), - ); } } @@ -373,7 +482,7 @@ impl PooledFileReadAt { // Start the target at the source's offset within 16 bytes so the copy moves aligned // words. The slice keeps the file's alignment, which the caller's alignment follows. let pad = range.start % 16; - let output = self.stream.device_alloc::(pad + length)?; + let output = mapped.stream.device_alloc::(pad + length)?; // The job owns the mapping and output until its copies are queued, even if cancelled. let mapped = Arc::clone(mapped); let output = self diff --git a/vortex-cuda/src/pooled_read_at/file/tests.rs b/vortex-cuda/src/pooled_read_at/file/tests.rs index 844f7c44092..c2a5121c216 100644 --- a/vortex-cuda/src/pooled_read_at/file/tests.rs +++ b/vortex-cuda/src/pooled_read_at/file/tests.rs @@ -6,6 +6,7 @@ use std::io; use std::sync::mpsc; use std::time::Duration; +use futures::future::try_join_all; use futures::poll; use parking_lot::Mutex; use rstest::rstest; @@ -24,6 +25,10 @@ use crate::CudaSessionExt; const WAIT: Duration = Duration::from_secs(30); const FILE_OFFSET: u64 = (1 << 32) + 37; +/// Serializes tests that observe or clear the process-wide mapping cache, because releasing it +/// concurrently would make another test's reopened file map again. +static MAPPING_CACHE: Mutex<()> = Mutex::new(()); + #[derive(Default)] struct FakeFileReadBackend { padded: bool, @@ -398,6 +403,94 @@ async fn mapped_ranges_validate_bounds( Ok(()) } +/// Concurrent overlapping mapped reads preserve bytes across chunk boundaries and complete +/// without deadlocking on the shared fill lock. +#[crate::test] +async fn concurrent_mapped_reads_preserve_bytes_and_offsets() -> VortexResult<()> { + let (_runtime, reader, _file) = mapped_reader()?; + if reader.mapped.is_none() { + return Ok(()); + } + // Populate the mapping explicitly: freshly written pages may not all remain resident. + let mapped = reader.mapped.as_ref().unwrap(); + #[cfg(target_os = "linux")] + mapped.mapping().advise(Advice::PopulateRead)?; + assert_eq!( + mapped.cached_range(0, MAPPED_FILE_BYTES), + Some(0..MAPPED_FILE_BYTES) + ); + // A staged fallback must fail rather than silently bypass mapped-copy coverage. + reader.read_slots.close(); + + let reads = [ + (4091, 2 * FILE_READ_CHUNK_BYTES + 3), + (0, MAPPED_FILE_BYTES), + (FILE_READ_CHUNK_BYTES as u64 - 7, 3 * 4096), + ]; + let buffers = wait_for( + "concurrent mapped reads did not finish", + try_join_all( + reads + .iter() + .map(|&(offset, length)| reader.read_at(offset, length, Alignment::of::())), + ), + ) + .await?; + for ((offset, length), buffer) in reads.into_iter().zip(buffers) { + assert_bytes(buffer, offset, length).await?; + } + Ok(()) +} + +/// Opens of one file version share a mapping, so hot scans neither remap nor unmap the file. +/// Changing the file maps its new version. +#[crate::test] +async fn reopened_files_share_mappings_until_modified() -> VortexResult<()> { + let _cache = MAPPING_CACHE.lock(); + let (_runtime, reader, mut file) = mapped_reader()?; + let Some(mapped) = reader.mapped.as_ref() else { + return Ok(()); + }; + let first = Arc::clone( + mapped + .file + .as_ref() + .vortex_expect("open file has a mapping"), + ); + let again = FileMapping::open(file.path())?; + assert!(Arc::ptr_eq(&first, &again)); + + io::Write::write_all(&mut file, &[7; 4096])?; + let modified = FileMapping::open(file.path())?; + assert!(!Arc::ptr_eq(&first, &modified)); + assert_eq!(modified.mapping().len(), MAPPED_FILE_BYTES + 4096); + assert_eq!(modified.mapping()[MAPPED_FILE_BYTES], 7); + Ok(()) +} + +/// Releasing cached mappings unmaps idle ones, so the next open maps the file again, while a +/// mapping an open reader still uses stays alive. +#[crate::test] +async fn released_mappings_are_remapped() -> VortexResult<()> { + let _cache = MAPPING_CACHE.lock(); + let (_runtime, reader, file) = mapped_reader()?; + let Some(mapped) = reader.mapped.as_ref() else { + return Ok(()); + }; + let in_use = Arc::clone( + mapped + .file + .as_ref() + .vortex_expect("open file has a mapping"), + ); + release_file_mappings(); + // The open reader keeps its mapping. + assert_eq!(in_use.mapping()[0], file_byte(0)); + let reopened = FileMapping::open(file.path())?; + assert!(!Arc::ptr_eq(&in_use, &reopened)); + Ok(()) +} + #[crate::test] async fn mapped_read_past_end_fails() -> VortexResult<()> { let (_runtime, reader, _file) = mapped_reader()?;