@@ -12,6 +12,7 @@ use std::os::unix::fs::MetadataExt;
1212use std:: path:: Path ;
1313use std:: sync:: Arc ;
1414use std:: sync:: LazyLock ;
15+ use std:: sync:: mpsc;
1516use std:: time:: SystemTime ;
1617
1718use cudarc:: driver:: CudaFunction ;
@@ -65,6 +66,24 @@ const MAPPED_COPY_THREADS: u32 = 256;
6566#[ cfg( target_os = "linux" ) ]
6667static MAPPED_FILL : Mutex < ( ) > = Mutex :: new ( ( ) ) ;
6768
69+ /// Releases a closed reader's mapping once its copy stream has drained.
70+ type MappedRelease = Box < dyn FnOnce ( ) + Send > ;
71+
72+ /// Releases of mapped files, each run once its copy stream has drained, off the scan's threads.
73+ static MAPPED_RELEASE_QUEUE : LazyLock < Option < mpsc:: Sender < MappedRelease > > > = LazyLock :: new ( || {
74+ let ( sender, receiver) = mpsc:: channel :: < MappedRelease > ( ) ;
75+ std:: thread:: Builder :: new ( )
76+ . name ( "vortex-mapped-release" . to_string ( ) )
77+ . spawn ( move || {
78+ // Runs until the process exits: the sender is a process-wide static.
79+ for job in receiver {
80+ job ( ) ;
81+ }
82+ } )
83+ . ok ( )
84+ . map ( |_| sender)
85+ } ) ;
86+
6887/// Options controlling how [`PooledFileReadAt`] opens and reads a local file.
6988#[ derive( Clone , Copy , Debug , Default , PartialEq , Eq ) ]
7089pub struct PooledFileReadAtOptions {
@@ -231,6 +250,19 @@ impl FileMapping {
231250/// cold-cache measurement, or to return the address space. Mappings that open readers still use
232251/// stay mapped until those readers close.
233252pub fn release_file_mappings ( ) {
253+ // Closed readers hand their mappings to the release queue; wait until it has dropped them.
254+ if let Some ( queue) = MAPPED_RELEASE_QUEUE . as_ref ( ) {
255+ let ( sender, receiver) = mpsc:: sync_channel :: < ( ) > ( 1 ) ;
256+ if queue
257+ . send ( Box :: new ( move || {
258+ let _ = sender. send ( ( ) ) ;
259+ } ) )
260+ . is_ok ( )
261+ {
262+ // An error means the release thread is gone, leaving nothing to wait for.
263+ let _ = receiver. recv ( ) ;
264+ }
265+ }
234266 let released: Vec < _ > = FILE_MAPPINGS
235267 . lock ( )
236268 . drain ( ..)
@@ -389,10 +421,24 @@ impl Drop for MappedFile {
389421 let Some ( file) = self . file . take ( ) else {
390422 return ;
391423 } ;
392- // Queued copies may still read the mapping, which the last release unmaps.
393- if self . stream . synchronize ( ) . is_err ( ) {
394- // Without proof the copies finished, never unmap memory the device may read.
395- let _leaked = ManuallyDrop :: new ( file) ;
424+ // Queued copies may still read the mapping, which the last release unmaps. The reader
425+ // drops when the scan has requested its last segment, typically on the thread that
426+ // decodes the scan's batches; waiting there for the copy stream to drain stalled every
427+ // decode and selection launch behind the scan's remaining copies.
428+ let stream = self . stream . clone ( ) ;
429+ let release: MappedRelease = Box :: new ( move || {
430+ if stream. synchronize ( ) . is_err ( ) {
431+ // Without proof the copies finished, never unmap memory the device may read.
432+ let _leaked = ManuallyDrop :: new ( file) ;
433+ }
434+ } ) ;
435+ match MAPPED_RELEASE_QUEUE . as_ref ( ) {
436+ Some ( queue) => {
437+ if let Err ( unsent) = queue. send ( release) {
438+ ( unsent. 0 ) ( ) ;
439+ }
440+ }
441+ None => release ( ) ,
396442 }
397443 }
398444}
0 commit comments