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 1ba52809413..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 { @@ -123,7 +135,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 +148,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()?); } @@ -478,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; @@ -499,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, @@ -532,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, }) } @@ -544,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(()) } @@ -554,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, @@ -573,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); @@ -611,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() @@ -637,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(()) } @@ -662,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 @@ -687,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();