perf(cuda): reuse file mappings

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.
  *

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::<CudaSession>();
     register_cuda_layout(session);

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;

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;

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,15 +137,137 @@ unsafe extern "C" {
     fn getpagesize() -> std::ffi::c_int;
 }
 
+fn page_size() -> VortexResult<usize> {
+    // 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<Mutex<FileMappingCache>> =
+    LazyLock::new(|| Mutex::new(VecDeque::with_capacity(FILE_MAPPINGS_CACHED_MAX + 1)));
+
+/// The entries of [`FILE_MAPPINGS`].
+type FileMappingCache = VecDeque<(FileVersion, Arc<FileMapping>)>;
+
+/// 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<Self> {
+        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<Mmap>,
+}
+
+impl FileMapping {
+    /// The cached mapping of `path`'s current version, mapping it on first use.
+    fn open(path: &Path) -> VortexResult<Arc<Self>> {
+        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
 /// warm read. Devices that read pageable memory through the host page tables can instead copy
 /// 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<Mmap>,
+    /// Present until drop; released only after the copy stream has drained.
+    file: Option<Arc<FileMapping>>,
     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::<u8>(pad + length)?;
+        let output = mapped.stream.device_alloc::<u8>(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