From a305df8f02fce29737f9cabdbacd974bd10dd4c5 Mon Sep 17 00:00:00 2001 From: Jake Abendroth Date: Tue, 6 Oct 2026 18:43:31 -0700 Subject: [PATCH 1/2] sort: create a merge batch's output before opening its inputs When a merge has more inputs than it can open at once, each batch is merged into a temporary file. That file was created after the batch's inputs had been opened. Create it first, so that creating it doesn't compete with them for file descriptors, and take each batch straight from the input iterator instead of collecting it first. --- src/uu/sort/src/merge.rs | 25 +++++-------------------- 1 file changed, 5 insertions(+), 20 deletions(-) diff --git a/src/uu/sort/src/merge.rs b/src/uu/sort/src/merge.rs index 1ba52809413..c1d05431318 100644 --- a/src/uu/sort/src/merge.rs +++ b/src/uu/sort/src/merge.rs @@ -123,7 +123,7 @@ pub fn merge_with_file_limit< F: ExactSizeIterator>, Tmp: WriteableTmpFile + 'static, >( - files: F, + mut files: F, settings: &GlobalSettings, output: Output, tmp_dir: &mut TmpDirWrapper, @@ -136,27 +136,12 @@ pub fn merge_with_file_limit< merger?.write_all(settings, output) } else { let mut temporary_files = vec![]; - let mut batch = Vec::with_capacity(batch_size); - for file in files { - batch.push(file); - if batch.len() >= batch_size { - assert_eq!(batch.len(), batch_size); - let merger = merge_without_limit(batch.into_iter(), settings)?; - batch = Vec::with_capacity(batch_size); - - let mut tmp_file = - Tmp::create(tmp_dir.next_file()?, settings.compress_prog.as_deref())?; - merger.write_all_to(settings, tmp_file.as_write(), || "write failed".into())?; - temporary_files.push(tmp_file.finished_writing()?); - } - } - // Merge any remaining files that didn't get merged in a full batch above. - if !batch.is_empty() { - assert!(batch.len() < batch_size); - let merger = merge_without_limit(batch.into_iter(), settings)?; - + while files.len() > 0 { + // Create the output before opening this batch's inputs, so that + // creating it doesn't compete with them for file descriptors. let mut tmp_file = Tmp::create(tmp_dir.next_file()?, settings.compress_prog.as_deref())?; + let merger = merge_without_limit(files.by_ref().take(batch_size), settings)?; merger.write_all_to(settings, tmp_file.as_write(), || "write failed".into())?; temporary_files.push(tmp_file.finished_writing()?); } From b6d6b10eaa891f5a9c7c658cde4265d991c01bd6 Mon Sep 17 00:00:00 2001 From: Jake Abendroth Date: Tue, 6 Oct 2026 18:43:55 -0700 Subject: [PATCH 2/2] sort: keep to its own temporary directory and files Temporary files were created, reopened and deleted by path, and cleanup emptied the temporary directory by path. Once the directory's path led to another directory, sort read the files there as its own and deleted them. The directory and each file are identified by device and inode when created. Files are created and deleted through a descriptor of the directory that is checked first, and reopened only if they are still the file sort created. Cleanup deletes only the names sort created. Redox and non-Unix systems keep the path-based code. The directory descriptor is closed after each use, so with merges creating their output before opening their inputs, sort needs no more file descriptors than before. --- src/uu/sort/Cargo.toml | 1 + src/uu/sort/src/ext_sort/threaded.rs | 5 +- src/uu/sort/src/merge.rs | 119 ++++++++-- src/uu/sort/src/tmp_dir.rs | 332 +++++++++++++++++++++------ tests/by-util/test_sort.rs | 105 +++++++++ 5 files changed, 467 insertions(+), 95 deletions(-) diff --git a/src/uu/sort/Cargo.toml b/src/uu/sort/Cargo.toml index 8a9f447e18f..2e4feaf77de 100644 --- a/src/uu/sort/Cargo.toml +++ b/src/uu/sort/Cargo.toml @@ -44,6 +44,7 @@ ctrlc = { workspace = true } [target.'cfg(unix)'.dependencies] libc = { workspace = true } +rustix = { workspace = true, features = ["fs"] } [dev-dependencies] divan = { workspace = true } diff --git a/src/uu/sort/src/ext_sort/threaded.rs b/src/uu/sort/src/ext_sort/threaded.rs index 3f6f9271298..48d9498818e 100644 --- a/src/uu/sort/src/ext_sort/threaded.rs +++ b/src/uu/sort/src/ext_sort/threaded.rs @@ -10,7 +10,6 @@ use std::cmp::Ordering; use std::fs::File; use std::io::{Read, Write, stderr}; use std::iter; -use std::path::PathBuf; use std::sync::mpsc::{Receiver, SyncSender}; use std::thread; @@ -22,7 +21,7 @@ use crate::chunks::RecycledChunk; use crate::merge::WriteableCompressedTmpFile; use crate::merge::WriteablePlainTmpFile; use crate::merge::WriteableTmpFile; -use crate::tmp_dir::TmpDirWrapper; +use crate::tmp_dir::{TmpDirWrapper, TmpPath}; use crate::{ GlobalSettings, Line, chunks::{self, Chunk}, @@ -290,7 +289,7 @@ fn read_write_loop( /// `compress_prog` is used to optionally compress file contents. fn write( chunk: &Chunk, - file: (File, PathBuf), + file: (File, TmpPath), compress_prog: Option<&str>, separator: u8, ) -> UResult { diff --git a/src/uu/sort/src/merge.rs b/src/uu/sort/src/merge.rs index c1d05431318..bfbd7fa64f1 100644 --- a/src/uu/sort/src/merge.rs +++ b/src/uu/sort/src/merge.rs @@ -16,10 +16,10 @@ use std::{ cmp::Ordering, collections::BinaryHeap, ffi::{OsStr, OsString}, - fs::{self, File}, + fs::File, io::{self, BufWriter, Read, Write}, iter, - path::{Path, PathBuf}, + path::Path, process::{Child, ChildStdin, ChildStdout, Command, Stdio}, rc::Rc, sync::mpsc::{Receiver, Sender, SyncSender, TryRecvError, channel, sync_channel}, @@ -32,17 +32,17 @@ use crate::{ GlobalSettings, Output, SortError, chunks::{self, Chunk, RecycledChunk}, current_open_fd_count, fd_soft_limit, merge_compare, open, - tmp_dir::TmpDirWrapper, + tmp_dir::{TmpDirWrapper, TmpPath}, }; /// If the output file occurs in the input files as well, copy the contents of the output file -/// and replace its occurrences in the inputs with that copy. +/// and replace its occurrences in the inputs with that copy, which is returned. fn replace_output_file_in_input_files( files: &mut [OsString], output: Option<&OsStr>, tmp_dir: &mut TmpDirWrapper, -) -> UResult<()> { - let mut copy: Option = None; +) -> UResult> { + let mut copy: Option = None; if let Some(Ok(output_path)) = output.map(|path| Path::new(path).canonicalize()) { for file in files { if Path::new(file) @@ -50,7 +50,7 @@ fn replace_output_file_in_input_files( .is_ok_and(|file_path| file_path == output_path) { if let Some(copy) = © { - *file = copy.clone().into_os_string(); + *file = copy.path().into(); } else { // Write through the descriptor `next_file` just opened rather // than `fs::copy`, which would put the source's permission bits @@ -60,13 +60,25 @@ fn replace_output_file_in_input_files( .map_err(|error| SortError::OpenTmpFileFailed { error })?; io::copy(&mut source, &mut copy_file) .map_err(|error| SortError::OpenTmpFileFailed { error })?; - *file = copy_path.clone().into_os_string(); + *file = copy_path.path().into(); copy = Some(copy_path); } } } } - Ok(()) + Ok(copy) +} + +/// Open the merge input `file`. `copy` is the copy of the output file returned by +/// [`replace_output_file_in_input_files`]. +fn open_input(file: &OsStr, copy: Option<&TmpPath>) -> UResult> { + match copy { + Some(copy) if Path::new(file) == copy.path() => Ok(Box::new( + copy.open() + .map_err(|error| SortError::OpenTmpFileFailed { error })?, + )), + _ => open(file), + } } /// Determine the effective merge batch size, enforcing a minimum and respecting the @@ -106,10 +118,10 @@ pub fn merge( output: Output, tmp_dir: &mut TmpDirWrapper, ) -> UResult<()> { - replace_output_file_in_input_files(files, output.as_output_name(), tmp_dir)?; + let copy = replace_output_file_in_input_files(files, output.as_output_name(), tmp_dir)?; let files = files .iter() - .map(|file| open(file).map(|file| PlainMergeInput { inner: file })); + .map(|file| open_input(file, copy.as_ref()).map(|inner| PlainMergeInput { inner })); if settings.compress_prog.is_none() { merge_with_file_limit::<_, _, WriteablePlainTmpFile>(files, settings, output, tmp_dir) } else { @@ -463,7 +475,7 @@ fn check_child_success(mut child: Child, program: &str) -> UResult<()> { pub trait WriteableTmpFile: Sized { type Closed: ClosedTmpFile; type InnerWrite: Write; - fn create(file: (File, PathBuf), compress_prog: Option<&str>) -> UResult; + fn create(file: (File, TmpPath), compress_prog: Option<&str>) -> UResult; /// Closes the temporary file. fn finished_writing(self) -> UResult; fn as_write(&mut self) -> &mut Self::InnerWrite; @@ -484,21 +496,21 @@ pub trait MergeInput: Send { } pub struct WriteablePlainTmpFile { - path: PathBuf, + path: TmpPath, file: BufWriter, } pub struct ClosedPlainTmpFile { - path: PathBuf, + path: TmpPath, } pub struct PlainTmpMergeInput { - path: PathBuf, + path: TmpPath, file: File, } impl WriteableTmpFile for WriteablePlainTmpFile { type Closed = ClosedPlainTmpFile; type InnerWrite = BufWriter; - fn create((file, path): (File, PathBuf), _: Option<&str>) -> UResult { + fn create((file, path): (File, TmpPath), _: Option<&str>) -> UResult { Ok(Self { file: BufWriter::new(file), path, @@ -517,7 +529,10 @@ impl ClosedTmpFile for ClosedPlainTmpFile { type Reopened = PlainTmpMergeInput; fn reopen(self) -> UResult { Ok(PlainTmpMergeInput { - file: File::open(&self.path).map_err(|error| SortError::OpenTmpFileFailed { error })?, + file: self + .path + .open() + .map_err(|error| SortError::OpenTmpFileFailed { error })?, path: self.path, }) } @@ -529,7 +544,9 @@ impl MergeInput for PlainTmpMergeInput { // we ignore failures to delete the temporary file, // because there is a race at the end of the execution and the whole // temporary directory might already be gone. - let _ = fs::remove_file(self.path); + // Close the file first: removing it briefly takes another descriptor. + drop(self.file); + self.path.remove(); Ok(()) } @@ -539,17 +556,17 @@ impl MergeInput for PlainTmpMergeInput { } pub struct WriteableCompressedTmpFile { - path: PathBuf, + path: TmpPath, compress_prog: String, child: Child, child_stdin: BufWriter, } pub struct ClosedCompressedTmpFile { - path: PathBuf, + path: TmpPath, compress_prog: String, } pub struct CompressedTmpMergeInput { - path: PathBuf, + path: TmpPath, compress_prog: String, child: Child, child_stdout: ChildStdout, @@ -558,7 +575,7 @@ impl WriteableTmpFile for WriteableCompressedTmpFile { type Closed = ClosedCompressedTmpFile; type InnerWrite = BufWriter; - fn create((file, path): (File, PathBuf), compress_prog: Option<&str>) -> UResult { + fn create((file, path): (File, TmpPath), compress_prog: Option<&str>) -> UResult { let compress_prog = compress_prog.unwrap(); let mut command = Command::new(compress_prog); command.stdin(Stdio::piped()).stdout(file); @@ -596,8 +613,10 @@ impl ClosedTmpFile for ClosedCompressedTmpFile { fn reopen(self) -> UResult { let mut command = Command::new(&self.compress_prog); // mirroring what is done for ClosedPlainTmpFile - let file = - File::open(&self.path).map_err(|error| SortError::OpenTmpFileFailed { error })?; + let file = self + .path + .open() + .map_err(|error| SortError::OpenTmpFileFailed { error })?; command.stdin(file).stdout(Stdio::piped()).arg("-d"); let mut child = command .spawn() @@ -622,7 +641,7 @@ impl MergeInput for CompressedTmpMergeInput { #[allow(clippy::drop_non_drop)] drop(self.child_stdout); check_child_success(self.child, &self.compress_prog)?; - let _ = fs::remove_file(self.path); + self.path.remove(); Ok(()) } @@ -647,6 +666,7 @@ impl MergeInput for PlainMergeInput { #[cfg(all(test, unix))] mod tests { use super::*; + use std::fs; use std::os::unix::fs::PermissionsExt; /// When the output file is also an input it is copied to a temporary file @@ -672,4 +692,53 @@ mod tests { ); assert_eq!(fs::read(copy).unwrap(), b"a\n"); } + + /// The copy of the output file is only read while it is the file sort + /// created: once the temporary directory's path leads elsewhere, the file + /// found at its name there is not read. + #[test] + #[cfg(not(target_os = "redox"))] + fn output_copy_is_not_read_through_a_replaced_tmp_dir() { + let dir = tempfile::tempdir().unwrap(); + let other = tempfile::tempdir().unwrap(); + fs::write(other.path().join("0"), b"other\n").unwrap(); + let out = dir.path().join("out"); + fs::write(&out, b"a\n").unwrap(); + + let mut tmp_dir = TmpDirWrapper::new(dir.path().to_owned()); + let mut files = vec![out.clone().into_os_string()]; + let copy = + replace_output_file_in_input_files(&mut files, Some(out.as_os_str()), &mut tmp_dir) + .unwrap(); + let tmp = copy.as_ref().unwrap().path().parent().unwrap(); + fs::rename(tmp, dir.path().join("moved")).unwrap(); + std::os::unix::fs::symlink(other.path(), tmp).unwrap(); + + assert!(open_input(&files[0], copy.as_ref()).is_err()); + } + + /// A temporary file that is being merged when the directory's path is made + /// to lead elsewhere: removing it afterwards leaves the file of the same + /// name there alone. + #[test] + #[cfg(not(target_os = "redox"))] + fn finished_input_is_not_removed_through_a_replaced_tmp_dir() { + let dir = tempfile::tempdir().unwrap(); + let other = tempfile::tempdir().unwrap(); + fs::write(other.path().join("0"), b"other\n").unwrap(); + + let mut tmp_dir = TmpDirWrapper::new(dir.path().to_owned()); + let input = WriteablePlainTmpFile::create(tmp_dir.next_file().unwrap(), None) + .unwrap() + .finished_writing() + .unwrap() + .reopen() + .unwrap(); + let tmp = input.path.path().parent().unwrap().to_owned(); + fs::rename(&tmp, dir.path().join("moved")).unwrap(); + std::os::unix::fs::symlink(other.path(), &tmp).unwrap(); + input.finished_reading().unwrap(); + + assert_eq!(fs::read(other.path().join("0")).unwrap(), b"other\n"); + } } diff --git a/src/uu/sort/src/tmp_dir.rs b/src/uu/sort/src/tmp_dir.rs index ee92d18a5c1..d869d0fbe7e 100644 --- a/src/uu/sort/src/tmp_dir.rs +++ b/src/uu/sort/src/tmp_dir.rs @@ -3,56 +3,51 @@ // For the full copyright and license information, please view the LICENSE // file that was distributed with this source code. +// spell-checker:ignore CLOEXEC NONBLOCK RDONLY RUSR unlinkat WRONLY WUSR + #[cfg(not(any(target_os = "redox", target_os = "wasi")))] -use std::path::Path; -#[cfg(not(any(target_os = "redox", target_os = "wasi")))] -use std::sync::atomic::{AtomicBool, Ordering}; -#[cfg(unix)] -use std::{ - fs::Permissions, - os::unix::fs::{OpenOptionsExt, PermissionsExt}, -}; +use std::sync::atomic::AtomicBool; use std::{ - fs::{File, OpenOptions}, + fs::File, path::PathBuf, - sync::{Arc, LazyLock, Mutex}, + sync::{Arc, LazyLock, Mutex, atomic::Ordering}, }; +#[cfg(unix)] +use std::{fs::Permissions, os::unix::fs::PermissionsExt}; -use tempfile::TempDir; use uucore::error::UResult; #[cfg(not(any(target_os = "redox", target_os = "wasi")))] use uucore::{error::USimpleError, show_error, translate}; use crate::SortError; -/// A wrapper around [`TempDir`] that may only exist once in a process. +/// A wrapper around [`tempfile::TempDir`] that may only exist once in a process. /// /// `TmpDirWrapper` handles the allocation of new temporary files in this temporary directory and /// deleting the whole directory when `SIGINT` is received. Creating a second `TmpDirWrapper` will /// fail because `ctrlc::set_handler()` fails when there's already a handler. /// The directory is only created once the first file is requested. pub struct TmpDirWrapper { - temp_dir: Option, + dir: Option>, parent_path: PathBuf, - size: usize, lock: Arc>, } #[derive(Default, Clone)] struct HandlerRegistration { lock: Option>>, - path: Option, + dir: Option>, } // Lazily create the global HandlerRegistration so all TmpDirWrapper instances and the -// SIGINT handler operate on the same lock/path snapshot. +// SIGINT handler operate on the same lock/directory snapshot. static HANDLER_STATE: LazyLock>> = LazyLock::new(|| Arc::new(Mutex::new(HandlerRegistration::default()))); #[cfg(not(any(target_os = "redox", target_os = "wasi")))] fn ensure_signal_handler_installed(state: Arc>) -> UResult<()> { // This shared state must originate from `HANDLER_STATE` so the handler always sees - // the current lock/path pair and can clean up the active temp directory on SIGINT. + // the current lock/directory pair and can clean up the active temp directory on SIGINT. // Install a shared SIGINT handler so the active temp directory is deleted when the user aborts. // Guard to ensure the SIGINT handler is registered once per process and reused. static HANDLER_INSTALLED: AtomicBool = AtomicBool::new(false); @@ -65,16 +60,16 @@ fn ensure_signal_handler_installed(state: Arc>) -> UR } if let Err(e) = ctrlc::set_handler(move || { - // Load the latest lock/path snapshot so the handler cleans the active temp dir. - let (lock, path) = { + // Load the latest lock/directory snapshot so the handler cleans the active temp dir. + let (lock, dir) = { let state = state.lock().unwrap(); - (state.lock.clone(), state.path.clone()) + (state.lock.clone(), state.dir.clone()) }; if let Some(lock) = lock { let _guard = lock.lock().unwrap(); - if let Some(path) = path - && let Err(e) = remove_tmp_dir(&path) + if let Some(dir) = dir + && let Err(e) = dir.remove() { show_error!( "{}", @@ -108,33 +103,32 @@ impl TmpDirWrapper { pub fn new(path: PathBuf) -> Self { Self { parent_path: path, - size: 0, - temp_dir: None, + dir: None, lock: Arc::default(), } } fn init_tmp_dir(&mut self) -> UResult<()> { - assert!(self.temp_dir.is_none()); - assert_eq!(self.size, 0); + assert!(self.dir.is_none()); let mut builder = tempfile::Builder::new(); builder.prefix("uutils_sort"); // The chunks hold a copy of the input, so keep them out of reach of other // users instead of relying on whatever umask the process inherited. #[cfg(unix)] builder.permissions(Permissions::from_mode(0o700)); - self.temp_dir = Some(builder.tempdir_in(&self.parent_path).map_err(|_| { + let dir = SortTmpDir::new(&mut builder, &self.parent_path).map_err(|_| { SortError::TmpFileCreationFailed { path: self.parent_path.clone(), } - })?); + })?; + let dir = Arc::new(dir); + self.dir = Some(Arc::clone(&dir)); - let path = self.temp_dir.as_ref().unwrap().path().to_owned(); let state = HANDLER_STATE.clone(); { let mut guard = state.lock().unwrap(); guard.lock = Some(self.lock.clone()); - guard.path = Some(path); + guard.dir = Some(dir); } // Always attempt to install the signal handler so that Ctrl+C @@ -144,27 +138,18 @@ impl TmpDirWrapper { Ok(()) } - pub fn next_file(&mut self) -> UResult<(File, PathBuf)> { - if self.temp_dir.is_none() { + pub fn next_file(&mut self) -> UResult<(File, TmpPath)> { + if self.dir.is_none() { self.init_tmp_dir()?; } + let dir = self.dir.as_ref().unwrap(); let _lock = self.lock.lock().unwrap(); - let file_name = self.size.to_string(); - self.size += 1; - let path = self.temp_dir.as_ref().unwrap().path().join(file_name); - let mut options = OpenOptions::new(); - options.write(true).create_new(true); - // Restrict the chunks too, so a directory whose mode is later relaxed - // doesn't expose them. - #[cfg(unix)] - options.mode(0o600); - Ok(( - options - .open(&path) - .map_err(|error| SortError::OpenTmpFileFailed { error })?, - path, - )) + // Count the name first, so cleanup removes the file even if creating it + // fails after the file exists. + let index = dir.created.fetch_add(1, Ordering::Relaxed); + dir.create(index) + .map_err(|error| SortError::OpenTmpFileFailed { error }.into()) } /// Function just waits if signal handler was called @@ -184,34 +169,225 @@ impl Drop for TmpDirWrapper { .is_some_and(|current| Arc::ptr_eq(current, &self.lock)) { guard.lock = None; - guard.path = None; + guard.dir = None; } drop(guard); - // Explicitly attempt cleanup before TempDir's Drop runs silently. - // TempDir::drop uses `let _ = remove_dir_all()` which silently - // ignores errors, potentially leaking the directory. #[cfg(not(any(target_os = "redox", target_os = "wasi")))] - if let Some(ref temp_dir) = self.temp_dir { - let _ = remove_tmp_dir(temp_dir.path()); + if let Some(dir) = &self.dir { + let _ = dir.remove(); } } } -/// Remove the directory at `path` by deleting its child files and then itself. -/// Errors while deleting child files are ignored. -#[cfg(not(any(target_os = "redox", target_os = "wasi")))] -fn remove_tmp_dir(path: &Path) -> std::io::Result<()> { - if let Ok(read_dir) = std::fs::read_dir(path) { - for file in read_dir.flatten() { - // if we fail to delete the file here it was probably deleted by another thread - // in the meantime, but that's ok. - let _ = std::fs::remove_file(file.path()); +/// The directory created by [`TmpDirWrapper`], whose files are named `0`, `1`, ... +/// It is only used through a descriptor checked to be this directory, so once its +/// path leads elsewhere, sort neither reads nor removes anything there. +#[cfg(all(unix, not(target_os = "redox")))] +mod imp { + use rustix::fs::{AtFlags, CWD, Mode, OFlags, openat, unlinkat}; + use std::fs::File; + use std::io; + use std::os::unix::fs::MetadataExt; + use std::path::{Path, PathBuf}; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tempfile::{Builder, TempDir}; + + /// The device and inode of a file. + type FileId = (u64, u64); + + fn file_id(file: &File) -> io::Result { + let metadata = file.metadata()?; + Ok((metadata.dev(), metadata.ino())) + } + + fn open_dir(path: &Path) -> io::Result { + let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC; + Ok(openat(CWD, path, flags, Mode::empty())?.into()) + } + + pub struct SortTmpDir { + temp_dir: TempDir, + id: FileId, + pub created: AtomicUsize, + } + + impl SortTmpDir { + /// Create the directory in `parent`. + pub fn new(builder: &mut Builder<'_, '_>, parent: &Path) -> io::Result { + // `remove` cleans up through the descriptor: `TempDir` would remove + // the directory recursively by path, wherever that leads by then. + let temp_dir = builder.disable_cleanup(true).tempdir_in(parent)?; + // The path is resolved once more right after creating the directory; + // there is no way to create and open a directory in one step. + let id = open_dir(temp_dir.path()) + .and_then(|dir| file_id(&dir)) + .inspect_err(|_| { + let _ = std::fs::remove_dir(temp_dir.path()); + })?; + Ok(Self { + temp_dir, + id, + created: AtomicUsize::new(0), + }) + } + + /// Open the directory, if its path still leads to it. The descriptor is + /// only kept for a moment: sort may have none to spare. + fn open_dir(&self) -> io::Result { + let dir = open_dir(self.temp_dir.path())?; + if file_id(&dir)? != self.id { + return Err(io::ErrorKind::NotFound.into()); + } + Ok(dir) + } + + /// Create the file `index`, which must not exist yet. + pub fn create(self: &Arc, index: usize) -> io::Result<(File, TmpPath)> { + let name = index.to_string(); + // Restrict the chunks too, so a directory whose mode is later relaxed + // doesn't expose them. + let file = File::from(openat( + &self.open_dir()?, + name.as_str(), + OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::CLOEXEC, + Mode::RUSR | Mode::WUSR, + )?); + let tmp_path = TmpPath { + dir: Arc::clone(self), + path: self.temp_dir.path().join(name), + index, + id: file_id(&file)?, + }; + Ok((file, tmp_path)) + } + + /// Remove the file `index`, ignoring failures. + fn remove_file(&self, index: usize) { + if let Ok(dir) = self.open_dir() { + let _ = unlinkat(&dir, index.to_string().as_str(), AtFlags::empty()); + } + } + + /// Remove the files created in the directory, then the directory itself. + /// Errors while deleting the files are ignored. + pub fn remove(&self) -> io::Result<()> { + let dir = self.open_dir()?; + for index in 0..self.created.load(Ordering::Relaxed) { + let _ = unlinkat(&dir, index.to_string().as_str(), AtFlags::empty()); + } + // Only removes an empty directory, and doesn't follow a symlink. + std::fs::remove_dir(self.temp_dir.path()) + } + } + + /// A file created by [`super::TmpDirWrapper::next_file`]. + pub struct TmpPath { + dir: Arc, + path: PathBuf, + index: usize, + id: FileId, + } + + impl TmpPath { + pub fn path(&self) -> &Path { + &self.path + } + + /// Open the file for reading, if it is still the file created. This takes + /// no descriptor of the directory, as merges reopen files while others are + /// open. + pub fn open(&self) -> io::Result { + // The path may lead elsewhere by now: don't wait for a FIFO there. + let flags = OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::NONBLOCK | OFlags::CLOEXEC; + let file = File::from(openat(CWD, &self.path, flags, Mode::empty())?); + if file_id(&file)? != self.id { + return Err(io::ErrorKind::NotFound.into()); + } + Ok(file) + } + + /// Delete the file, ignoring failures. + pub fn remove(self) { + self.dir.remove_file(self.index); } } - std::fs::remove_dir(path) } +/// The directory created by [`TmpDirWrapper`], whose files are named `0`, `1`, ... +#[cfg(not(all(unix, not(target_os = "redox"))))] +mod imp { + use std::fs::{File, OpenOptions}; + use std::io; + #[cfg(target_os = "redox")] + use std::os::unix::fs::OpenOptionsExt; + use std::path::{Path, PathBuf}; + use std::sync::atomic::AtomicUsize; + #[cfg(not(any(target_os = "redox", target_os = "wasi")))] + use std::sync::atomic::Ordering; + use tempfile::{Builder, TempDir}; + + pub struct SortTmpDir { + temp_dir: TempDir, + pub created: AtomicUsize, + } + + impl SortTmpDir { + /// Create the directory in `parent`. + pub fn new(builder: &mut Builder<'_, '_>, parent: &Path) -> io::Result { + Ok(Self { + temp_dir: builder.tempdir_in(parent)?, + created: AtomicUsize::new(0), + }) + } + + /// Create the file `index`, which must not exist yet. + pub fn create(&self, index: usize) -> io::Result<(File, TmpPath)> { + let path = self.temp_dir.path().join(index.to_string()); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(target_os = "redox")] + options.mode(0o600); + let file = options.open(&path)?; + Ok((file, TmpPath { path })) + } + + /// Remove the files created in the directory, then the directory itself. + /// Errors while deleting the files are ignored. + #[cfg(not(any(target_os = "redox", target_os = "wasi")))] + pub fn remove(&self) -> io::Result<()> { + for index in 0..self.created.load(Ordering::Relaxed) { + let _ = std::fs::remove_file(self.temp_dir.path().join(index.to_string())); + } + std::fs::remove_dir(self.temp_dir.path()) + } + } + + /// A file created by [`super::TmpDirWrapper::next_file`]. + pub struct TmpPath { + path: PathBuf, + } + + impl TmpPath { + pub fn path(&self) -> &Path { + &self.path + } + + pub fn open(&self) -> io::Result { + File::open(&self.path) + } + + /// Delete the file, ignoring failures. + pub fn remove(self) { + let _ = std::fs::remove_file(&self.path); + } + } +} + +use imp::SortTmpDir; +pub use imp::TmpPath; + #[cfg(all(test, unix))] mod tests { use super::TmpDirWrapper; @@ -242,15 +418,37 @@ mod tests { fn tmp_files_are_private_regardless_of_umask() { // Pin a permissive umask: under 0077 the umask alone would produce 0700 and // 0600, so the assertions would hold for a broken implementation too. The - // guard restores it, and the only other test here that creates files sets - // the modes it cares about explicitly. + // guard restores it, and the other tests here that create files don't + // check modes. let _umask = UmaskGuard::set(0o022); let parent = tempfile::tempdir().unwrap(); let mut wrapper = TmpDirWrapper::new(parent.path().to_owned()); let (_file, path) = wrapper.next_file().unwrap(); - assert_eq!(mode(path.parent().unwrap()), 0o700); - assert_eq!(mode(&path), 0o600); + assert_eq!(mode(path.path().parent().unwrap()), 0o700); + assert_eq!(mode(path.path()), 0o600); + } + + /// Once the directory's path leads to another directory, no file is created, + /// read or removed there, even one with the name of a temporary file. + #[test] + #[cfg(not(target_os = "redox"))] + fn replaced_directory_is_left_alone() { + let parent = tempfile::tempdir().unwrap(); + let mut wrapper = TmpDirWrapper::new(parent.path().to_owned()); + let (_file, path) = wrapper.next_file().unwrap(); + let tmp_dir = path.path().parent().unwrap().to_owned(); + std::fs::rename(&tmp_dir, parent.path().join("moved")).unwrap(); + std::fs::create_dir(&tmp_dir).unwrap(); + std::fs::write(tmp_dir.join("0"), "keep").unwrap(); + + assert!(wrapper.next_file().is_err()); + assert!(path.open().is_err()); + path.remove(); + drop(wrapper); + + assert_eq!(std::fs::read_to_string(tmp_dir.join("0")).unwrap(), "keep"); + assert!(!tmp_dir.join("1").exists()); } } diff --git a/tests/by-util/test_sort.rs b/tests/by-util/test_sort.rs index ee06176e3a5..574c0c01e03 100644 --- a/tests/by-util/test_sort.rs +++ b/tests/by-util/test_sort.rs @@ -1894,6 +1894,47 @@ fn test_more_files_than_fd_limit() { .stdout_only(expected); } +#[test] +#[cfg(target_os = "linux")] +fn test_temporary_files_with_few_file_descriptors() { + use rustix::process::Resource; + let ts = TestScenario::new("sort"); + // Each input is larger than what the merge reads ahead, so it stays open. + let mut names = Vec::new(); + let mut expected = String::new(); + for i in 0..12 { + let name = format!("few_fds_{i:02}"); + let contents = format!("{i:02}\n").repeat(32768); + ts.fixtures.write(&name, &contents); + names.push(name); + expected.push_str(&contents); + } + let unsorted: String = (0..32768) + .flat_map(|_| (0..12).rev().map(|i| format!("{i:02}\n"))) + .collect(); + ts.fixtures.write("unsorted", &unsorted); + + // Stdin, stdout, stderr, two merge inputs and the merge output. + ts.ucmd() + .limit(Resource::Nofile, 6, 6) + .arg("-m") + .args(&names) + .succeeds() + .stdout_only(&expected); + ts.ucmd() + .limit(Resource::Nofile, 6, 6) + .args(&["-S", "64K", "unsorted"]) + .succeeds() + .stdout_only(&expected); + // Starting the compress program can take two more for a moment: on musl and + // glibc before 2.24, a failed exec is reported through a pipe. + ts.ucmd() + .limit(Resource::Nofile, 10, 10) + .args(&["-S", "64K", "--compress-program", "gzip", "unsorted"]) + .succeeds() + .stdout_only(&expected); +} + #[test] fn test_sigpipe_panic() { let mut cmd = new_ucmd!(); @@ -2223,6 +2264,70 @@ fn test_tmp_files_are_private() { child.wait().unwrap().code_is(2); } +#[test] +#[cfg(all(unix, not(target_os = "redox")))] +fn test_tmp_dir_replaced_by_a_symlink() { + use rustix::process::{Pid, Signal, kill_process}; + use std::{ + fs, + io::Write as _, + time::{Duration, Instant}, + }; + + // Once the temporary directory's path leads to another directory, sort must + // neither read nor delete anything there, whether it exits normally or on SIGINT. + for interrupt in [false, true] { + let (at, mut ucmd) = at_and_ucmd!(); + at.mkdir("scratch"); + at.mkdir("other"); + at.write("other/0", "zzz other\n"); + at.write("other/notes", "notes\n"); + at.mkfifo("input"); + let child = ucmd + .args(&["-S", "1K", "-T", "scratch", "input"]) + .run_no_wait(); + // Keep the input open until the directory has been replaced, so that the + // rest of the run happens afterwards. + let mut input = fs::OpenOptions::new() + .write(true) + .open(at.plus("input")) + .unwrap(); + input + .write_all("willow\ncedar\n".repeat(8192).as_bytes()) + .unwrap(); + // Wait for sort's first temporary file, however long the machine takes: + // the deadline is only reached if it never writes one. + let deadline = Instant::now() + Duration::from_secs(60); + let tmp_dir = loop { + let dir = fs::read_dir(at.plus("scratch")).unwrap().next(); + let dir = dir.map(|entry| entry.unwrap().path()); + if let Some(dir) = dir.filter(|dir| dir.join("0").exists()) { + break dir; + } + assert!( + Instant::now() < deadline, + "sort did not write a temporary file" + ); + std::thread::sleep(Duration::from_millis(10)); + }; + fs::rename(&tmp_dir, at.plus("moved")).unwrap(); + std::os::unix::fs::symlink(at.plus("other"), &tmp_dir).unwrap(); + if interrupt { + kill_process(Pid::from_raw(child.id() as i32).unwrap(), Signal::INT).unwrap(); + } + drop(input); + let result = child.wait().unwrap(); + + // sort can't go on without its directory, but must not take any data + // from the other one. + result.code_is(2); + assert!(!result.stdout_str().contains("zzz")); + assert_eq!(at.read("other/0"), "zzz other\n"); + assert_eq!(at.read("other/notes"), "notes\n"); + assert_eq!(fs::read_dir(at.plus("other")).unwrap().count(), 2); + } +} + #[test] fn test_same_sort_mode_twice() { new_ucmd!().args(&["-k", "2n,2n", "empty.txt"]).succeeds();