44#[ cfg( target_os = "linux" ) ]
55mod direct;
66
7+ use std:: collections:: VecDeque ;
78use std:: fs:: File ;
89use std:: mem:: ManuallyDrop ;
910use std:: ops:: Range ;
11+ use std:: os:: unix:: fs:: MetadataExt ;
1012use std:: path:: Path ;
1113use std:: sync:: Arc ;
14+ use std:: sync:: LazyLock ;
15+ use std:: time:: SystemTime ;
1216
1317use cudarc:: driver:: CudaFunction ;
1418use cudarc:: driver:: CudaSlice ;
@@ -56,8 +60,7 @@ const FILE_READ_CHUNK_BYTES: usize = 4 << 20;
5660const MAPPED_COPY_BLOCKS_MAX : u64 = 1024 ;
5761const MAPPED_COPY_THREADS : u32 = 256 ;
5862
59- /// Filling page tables from several threads contends on the process memory map and measured
60- /// 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.
6164static MAPPED_FILL : Mutex < ( ) > = Mutex :: new ( ( ) ) ;
6265
6366/// Options controlling how [`PooledFileReadAt`] opens and reads a local file.
@@ -132,15 +135,137 @@ unsafe extern "C" {
132135 fn getpagesize ( ) -> std:: ffi:: c_int ;
133136}
134137
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+
135260/// A buffered file whose page-cache pages the device copies directly.
136261///
137262/// A buffered `pread` copies cached pages into pinned staging on the CPU, the slowest step of a
138263/// warm read. Devices that read pageable memory through the host page tables can instead copy
139264/// from a mapping of those pages with a kernel. The CPU then only fills page tables, which is
140265/// several times faster than copying the data.
141266struct MappedFile {
142- /// Present until drop; unmapped only after the copy stream has drained.
143- mapping : Option < Mmap > ,
267+ /// Present until drop; released only after the copy stream has drained.
268+ file : Option < Arc < FileMapping > > ,
144269 copy : CudaFunction ,
145270 stream : VortexCudaStream ,
146271}
@@ -160,22 +285,19 @@ impl MappedFile {
160285 if uses_host_page_tables != 1 {
161286 return Ok ( None ) ;
162287 }
163- let file = File :: open ( path) ?;
164- // SAFETY: Vortex files are immutable while open. As with any file mapping, truncation
165- // by another process would fault on access.
166- let mapping = unsafe { Mmap :: map ( & file) ? } ;
167288 let copy = session. load_function_with_suffixes ( "mapped_copy" , & [ ] ) ?;
168289 Ok ( Some ( Self {
169- mapping : Some ( mapping ) ,
290+ file : Some ( FileMapping :: open ( path ) ? ) ,
170291 copy,
171292 stream : stream. clone ( ) ,
172293 } ) )
173294 }
174295
175296 fn mapping ( & self ) -> & Mmap {
176- self . mapping
297+ self . file
177298 . as_ref ( )
178- . vortex_expect ( "mapping is present until drop" )
299+ . vortex_expect ( "file is present until drop" )
300+ . mapping ( )
179301 }
180302
181303 /// Return the in-bounds file range if every overlapping page is in the page cache.
@@ -191,11 +313,7 @@ impl MappedFile {
191313 if bytes. is_empty ( ) {
192314 return Some ( range) ;
193315 }
194- // SAFETY: getpagesize has no preconditions.
195- let page = usize:: try_from ( unsafe { getpagesize ( ) } ) . ok ( ) ?;
196- if !page. is_power_of_two ( ) {
197- return None ;
198- }
316+ let page = page_size ( ) . ok ( ) ?;
199317 let begin = ( bytes. as_ptr ( ) as usize ) & !( page - 1 ) ;
200318 let end = bytes. as_ptr ( ) as usize + bytes. len ( ) ;
201319 let mut resident = vec ! [ 0u8 ; ( end - begin) . div_ceil( page) ] ;
@@ -262,22 +380,14 @@ impl MappedFile {
262380
263381impl Drop for MappedFile {
264382 fn drop ( & mut self ) {
265- let Some ( mapping ) = self . mapping . take ( ) else {
383+ let Some ( file ) = self . file . take ( ) else {
266384 return ;
267385 } ;
268- // Queued copies may still read the mapping.
386+ // Queued copies may still read the mapping, which the last release unmaps .
269387 if self . stream . synchronize ( ) . is_err ( ) {
270388 // Without proof the copies finished, never unmap memory the device may read.
271- let _leaked = ManuallyDrop :: new ( mapping) ;
272- return ;
389+ let _leaked = ManuallyDrop :: new ( file) ;
273390 }
274- // Unmapping filled page tables costs milliseconds per GiB; keep it off the caller. If
275- // the thread cannot start, the returned closure unmaps here.
276- drop (
277- std:: thread:: Builder :: new ( )
278- . name ( "vortex-unmap" . to_string ( ) )
279- . spawn ( move || drop ( mapping) ) ,
280- ) ;
281391 }
282392}
283393
@@ -366,7 +476,7 @@ impl PooledFileReadAt {
366476 // Start the target at the source's offset within 16 bytes so the copy moves aligned
367477 // words. The slice keeps the file's alignment, which the caller's alignment follows.
368478 let pad = range. start % 16 ;
369- let output = self . stream . device_alloc :: < u8 > ( pad + length) ?;
479+ let output = mapped . stream . device_alloc :: < u8 > ( pad + length) ?;
370480 // The job owns the mapping and output until its copies are queued, even if cancelled.
371481 let mapped = Arc :: clone ( mapped) ;
372482 let output = self
0 commit comments