44#[ cfg( target_os = "linux" ) ]
55mod direct;
66
7+ use std:: collections:: VecDeque ;
78use std:: fs:: File ;
9+ use std:: mem:: ManuallyDrop ;
810use std:: ops:: Range ;
11+ use std:: os:: unix:: fs:: MetadataExt ;
912use std:: path:: Path ;
1013use std:: sync:: Arc ;
14+ use std:: sync:: LazyLock ;
15+ use std:: time:: SystemTime ;
1116
1217use cudarc:: driver:: CudaFunction ;
1318use cudarc:: driver:: CudaSlice ;
@@ -55,8 +60,7 @@ const FILE_READ_CHUNK_BYTES: usize = 4 << 20;
5560const MAPPED_COPY_BLOCKS_MAX : u64 = 1024 ;
5661const MAPPED_COPY_THREADS : u32 = 256 ;
5762
58- /// Filling page tables from several threads contends on the process memory map and measured
59- /// slower than a single thread, so fills are serialized across all mapped files.
63+ /// Serialize page-table fills across all mapped files, releasing the lock after each chunk.
6064static MAPPED_FILL : Mutex < ( ) > = Mutex :: new ( ( ) ) ;
6165
6266/// Options controlling how [`PooledFileReadAt`] opens and reads a local file.
@@ -131,15 +135,137 @@ unsafe extern "C" {
131135 fn getpagesize ( ) -> std:: ffi:: c_int ;
132136}
133137
138+ fn page_size ( ) -> VortexResult < usize > {
139+ // SAFETY: getpagesize has no preconditions.
140+ let page = usize:: try_from ( unsafe { getpagesize ( ) } ) ?;
141+ vortex_ensure ! ( page. is_power_of_two( ) , "invalid page size {page}" ) ;
142+ Ok ( page)
143+ }
144+
145+ /// Number of recently opened file versions whose mappings stay alive between opens.
146+ const FILE_MAPPINGS_CACHED_MAX : usize = 64 ;
147+
148+ /// Mappings of recently opened file versions, least recently opened first.
149+ ///
150+ /// Each open used to map its file and unmap it after the scan. Unmapping holds the process's
151+ /// address-space lock for milliseconds per GiB of filled page tables, and pageable CUDA copies
152+ /// on other threads wait for that lock while holding the driver's: on NDS-H SF10 Q10, every
153+ /// thread stalled for the first ~2 ms of each query. Reusing the mapping also keeps its page
154+ /// tables filled, so hot scans skip most of the fill work.
155+ static FILE_MAPPINGS : LazyLock < Mutex < FileMappingCache > > =
156+ LazyLock :: new ( || Mutex :: new ( VecDeque :: with_capacity ( FILE_MAPPINGS_CACHED_MAX + 1 ) ) ) ;
157+
158+ /// The entries of [`FILE_MAPPINGS`].
159+ type FileMappingCache = VecDeque < ( FileVersion , Arc < FileMapping > ) > ;
160+
161+ /// A file's identity and version: modifying or replacing a file yields a new mapping.
162+ ///
163+ /// Modifying a mapped file in place without changing its size within the timestamp
164+ /// granularity is not detected; Vortex files are immutable once written.
165+ #[ derive( Clone , Copy , Debug , PartialEq , Eq ) ]
166+ struct FileVersion {
167+ device : u64 ,
168+ inode : u64 ,
169+ len : u64 ,
170+ modified : SystemTime ,
171+ }
172+
173+ impl FileVersion {
174+ fn of ( file : & File ) -> VortexResult < Self > {
175+ let metadata = file. metadata ( ) ?;
176+ Ok ( Self {
177+ device : metadata. dev ( ) ,
178+ inode : metadata. ino ( ) ,
179+ len : metadata. len ( ) ,
180+ modified : metadata. modified ( ) ?,
181+ } )
182+ }
183+ }
184+
185+ /// A read-only mapping of one file version, shared by every open of that version.
186+ struct FileMapping {
187+ /// Present until drop; unmapped only after every user's copies have drained.
188+ mapping : Option < Mmap > ,
189+ }
190+
191+ impl FileMapping {
192+ /// The cached mapping of `path`'s current version, mapping it on first use.
193+ fn open ( path : & Path ) -> VortexResult < Arc < Self > > {
194+ let file = File :: open ( path) ?;
195+ let version = FileVersion :: of ( & file) ?;
196+ let mut mappings = FILE_MAPPINGS . lock ( ) ;
197+ if let Some ( index) = mappings. iter ( ) . position ( |( cached, _) | * cached == version) {
198+ let entry = mappings. remove ( index) . vortex_expect ( "index is in bounds" ) ;
199+ let mapping = Arc :: clone ( & entry. 1 ) ;
200+ mappings. push_back ( entry) ;
201+ return Ok ( mapping) ;
202+ }
203+ // SAFETY: Vortex files are immutable while open. As with any file mapping, truncation
204+ // by another process would fault on access.
205+ let mapping = unsafe { Mmap :: map ( & file) ? } ;
206+ let mapping = Arc :: new ( Self {
207+ mapping : Some ( mapping) ,
208+ } ) ;
209+ mappings. push_back ( ( version, Arc :: clone ( & mapping) ) ) ;
210+ if mappings. len ( ) > FILE_MAPPINGS_CACHED_MAX {
211+ // Users still holding the evicted mapping keep it alive until their copies drain.
212+ drop ( mappings. pop_front ( ) ) ;
213+ }
214+ Ok ( mapping)
215+ }
216+
217+ fn mapping ( & self ) -> & Mmap {
218+ self . mapping
219+ . as_ref ( )
220+ . vortex_expect ( "mapping is present until drop" )
221+ }
222+ }
223+
224+ /// Release cached file mappings that no open reader uses, unmapping them before returning.
225+ ///
226+ /// Readers keep files mapped between scans, so later scans of a file reuse its filled page
227+ /// tables. A mapped page cannot be dropped from the page cache, for example by
228+ /// `posix_fadvise(DONTNEED)`, so release the mappings before evicting files, such as before a
229+ /// cold-cache measurement, or to return the address space. Mappings that open readers still use
230+ /// stay mapped until those readers close.
231+ pub fn release_file_mappings ( ) {
232+ let released: Vec < _ > = FILE_MAPPINGS
233+ . lock ( )
234+ . drain ( ..)
235+ . map ( |( _, mapping) | mapping)
236+ . collect ( ) ;
237+ for mapping in released {
238+ if let Ok ( mut mapping) = Arc :: try_unwrap ( mapping) {
239+ // Unmap on this thread, not in the background, so the pages are unmapped on return.
240+ drop ( mapping. mapping . take ( ) ) ;
241+ }
242+ }
243+ }
244+
245+ impl Drop for FileMapping {
246+ fn drop ( & mut self ) {
247+ let Some ( mapping) = self . mapping . take ( ) else {
248+ return ;
249+ } ;
250+ // Unmapping filled page tables costs milliseconds per GiB; keep it off the caller. If
251+ // the thread cannot start, the returned closure unmaps here.
252+ drop (
253+ std:: thread:: Builder :: new ( )
254+ . name ( "vortex-unmap" . to_string ( ) )
255+ . spawn ( move || drop ( mapping) ) ,
256+ ) ;
257+ }
258+ }
259+
134260/// A buffered file whose page-cache pages the device copies directly.
135261///
136262/// A buffered `pread` copies cached pages into pinned staging on the CPU, the slowest step of a
137263/// warm read. Devices that read pageable memory through the host page tables can instead copy
138264/// from a mapping of those pages with a kernel. The CPU then only fills page tables, which is
139265/// several times faster than copying the data.
140266struct MappedFile {
141- /// Present until drop; unmapped only after the copy stream has drained.
142- mapping : Option < Mmap > ,
267+ /// Present until drop; released only after the copy stream has drained.
268+ file : Option < Arc < FileMapping > > ,
143269 copy : CudaFunction ,
144270 stream : VortexCudaStream ,
145271}
@@ -159,22 +285,19 @@ impl MappedFile {
159285 if uses_host_page_tables != 1 {
160286 return Ok ( None ) ;
161287 }
162- let file = File :: open ( path) ?;
163- // SAFETY: Vortex files are immutable while open. As with any file mapping, truncation
164- // by another process would fault on access.
165- let mapping = unsafe { Mmap :: map ( & file) ? } ;
166288 let copy = session. load_function_with_suffixes ( "mapped_copy" , & [ ] ) ?;
167289 Ok ( Some ( Self {
168- mapping : Some ( mapping ) ,
290+ file : Some ( FileMapping :: open ( path ) ? ) ,
169291 copy,
170292 stream : stream. clone ( ) ,
171293 } ) )
172294 }
173295
174296 fn mapping ( & self ) -> & Mmap {
175- self . mapping
297+ self . file
176298 . as_ref ( )
177- . vortex_expect ( "mapping is present until drop" )
299+ . vortex_expect ( "file is present until drop" )
300+ . mapping ( )
178301 }
179302
180303 /// Return the in-bounds file range if every overlapping page is in the page cache.
@@ -190,11 +313,7 @@ impl MappedFile {
190313 if bytes. is_empty ( ) {
191314 return Some ( range) ;
192315 }
193- // SAFETY: getpagesize has no preconditions.
194- let page = usize:: try_from ( unsafe { getpagesize ( ) } ) . ok ( ) ?;
195- if !page. is_power_of_two ( ) {
196- return None ;
197- }
316+ let page = page_size ( ) . ok ( ) ?;
198317 let begin = ( bytes. as_ptr ( ) as usize ) & !( page - 1 ) ;
199318 let end = bytes. as_ptr ( ) as usize + bytes. len ( ) ;
200319 let mut resident = vec ! [ 0u8 ; ( end - begin) . div_ceil( page) ] ;
@@ -261,22 +380,14 @@ impl MappedFile {
261380
262381impl Drop for MappedFile {
263382 fn drop ( & mut self ) {
264- let Some ( mapping ) = self . mapping . take ( ) else {
383+ let Some ( file ) = self . file . take ( ) else {
265384 return ;
266385 } ;
267- // Queued copies may still read the mapping.
386+ // Queued copies may still read the mapping, which the last release unmaps .
268387 if self . stream . synchronize ( ) . is_err ( ) {
269388 // Without proof the copies finished, never unmap memory the device may read.
270- std:: mem:: forget ( mapping) ;
271- return ;
389+ let _leaked = ManuallyDrop :: new ( file) ;
272390 }
273- // Unmapping filled page tables costs milliseconds per GiB; keep it off the caller. If
274- // the thread cannot start, the returned closure unmaps here.
275- drop (
276- std:: thread:: Builder :: new ( )
277- . name ( "vortex-unmap" . to_string ( ) )
278- . spawn ( move || drop ( mapping) ) ,
279- ) ;
280391 }
281392}
282393
@@ -365,7 +476,7 @@ impl PooledFileReadAt {
365476 // Start the target at the source's offset within 16 bytes so the copy moves aligned
366477 // words. The slice keeps the file's alignment, which the caller's alignment follows.
367478 let pad = range. start % 16 ;
368- let output = self . stream . device_alloc :: < u8 > ( pad + length) ?;
479+ let output = mapped . stream . device_alloc :: < u8 > ( pad + length) ?;
369480 // The job owns the mapping and output until its copies are queued, even if cancelled.
370481 let mapped = Arc :: clone ( mapped) ;
371482 let output = self
0 commit comments