dd: simplify signal handling by removing Alarm timer thread

Author: ChrisDryden

Cargo.lock

@@ -32,8 +32,7 @@ thiserror = { workspace = true }
 fluent = { workspace = true }
 
 [target.'cfg(any(target_os = "linux", target_os = "android"))'.dependencies]
-nix = { workspace = true, features = ["fs"] }
-signal-hook = { workspace = true }
+nix = { workspace = true, features = ["fs", "signal"] }
 
 [[bin]]
 name = "dd"

src/uu/dd/Cargo.toml

@@ -23,6 +23,8 @@ use nix::fcntl::OFlag;
 use parseargs::Parser;
 use progress::ProgUpdateType;
 use progress::{ProgUpdate, ReadStat, StatusLevel, WriteStat, gen_prog_updater};
+#[cfg(target_os = "linux")]
+use progress::{check_and_reset_sigusr1, install_sigusr1_handler};
 use uucore::io::OwnedFileDescriptorOrHandle;
 use uucore::translate;
 
@@ -45,8 +47,7 @@ use std::os::unix::{
 #[cfg(windows)]
 use std::os::windows::{fs::MetadataExt, io::AsHandle};
 use std::path::Path;
-use std::sync::atomic::AtomicU8;
-use std::sync::{Arc, atomic::Ordering::Relaxed, mpsc};
+use std::sync::mpsc;
 use std::thread;
 use std::time::{Duration, Instant};
 
@@ -86,72 +87,6 @@ struct Settings {
     buffered: bool,
 }
 
-/// A timer which triggers on a given interval
-///
-/// After being constructed with [`Alarm::with_interval`], [`Alarm::get_trigger`]
-/// will return [`ALARM_TRIGGER_TIMER`] once per the given [`Duration`].
-/// Alarm can be manually triggered with closure returned by [`Alarm::manual_trigger_fn`].
-/// [`Alarm::get_trigger`] will return [`ALARM_TRIGGER_SIGNAL`] in this case.
-///
-/// Can be cloned, but the trigger status is shared across all instances so only
-/// the first caller each interval will yield true.
-///
-/// When all instances are dropped the background thread will exit on the next interval.
-pub struct Alarm {
-    interval: Duration,
-    trigger: Arc<AtomicU8>,
-}
-
-pub const ALARM_TRIGGER_NONE: u8 = 0;
-pub const ALARM_TRIGGER_TIMER: u8 = 1;
-pub const ALARM_TRIGGER_SIGNAL: u8 = 2;
-
-impl Alarm {
-    /// use to construct alarm timer with duration
-    pub fn with_interval(interval: Duration) -> Self {
-        let trigger = Arc::new(AtomicU8::default());
-
-        let weak_trigger = Arc::downgrade(&trigger);
-        thread::spawn(move || {
-            while let Some(trigger) = weak_trigger.upgrade() {
-                thread::sleep(interval);
-                trigger.store(ALARM_TRIGGER_TIMER, Relaxed);
-            }
-        });
-
-        Self { interval, trigger }
-    }
-
-    /// Returns a closure that allows to manually trigger the alarm
-    ///
-    /// This is useful for cases where more than one alarm even source exists
-    /// In case of `dd` there is the SIGUSR1/SIGINFO case where we want to
-    /// trigger an manual progress report.
-    pub fn manual_trigger_fn(&self) -> Box<dyn Send + Sync + Fn()> {
-        let weak_trigger = Arc::downgrade(&self.trigger);
-        Box::new(move || {
-            if let Some(trigger) = weak_trigger.upgrade() {
-                trigger.store(ALARM_TRIGGER_SIGNAL, Relaxed);
-            }
-        })
-    }
-
-    /// Use this function to poll for any pending alarm event
-    ///
-    /// Returns `ALARM_TRIGGER_NONE` for no pending event.
-    /// Returns `ALARM_TRIGGER_TIMER` if the event was triggered by timer
-    /// Returns `ALARM_TRIGGER_SIGNAL` if the event was triggered manually
-    /// by the closure returned from `manual_trigger_fn`
-    pub fn get_trigger(&self) -> u8 {
-        self.trigger.swap(ALARM_TRIGGER_NONE, Relaxed)
-    }
-
-    // Getter function for the configured interval duration
-    pub fn get_interval(&self) -> Duration {
-        self.interval
-    }
-}
-
 /// A number in blocks or bytes
 ///
 /// Some values (seek, skip, iseek, oseek) can have values either in blocks or in bytes.
@@ -1177,20 +1112,11 @@ fn dd_copy(mut i: Input, o: Output) -> io::Result<()> {
     // This is the max size needed.
     let mut buf = vec![BUF_INIT_BYTE; bsize];
 
-    // Spawn a timer thread to provide a scheduled signal indicating when we
-    // should send an update of our progress to the reporting thread.
-    //
-    // This avoids the need to query the OS monotonic clock for every block.
-    let alarm = Alarm::with_interval(Duration::from_secs(1));
+    let mut next_progress_time = Instant::now() + Duration::from_secs(1);
 
-    // The signal handler spawns an own thread that waits for signals.
-    // When the signal is received, it calls a handler function.
-    // We inject a handler function that manually triggers the alarm.
     #[cfg(target_os = "linux")]
-    let signal_handler = progress::SignalHandler::install_signal_handler(alarm.manual_trigger_fn());
-    #[cfg(target_os = "linux")]
-    if let Err(e) = &signal_handler {
-        if Some(StatusLevel::None) != i.settings.status {
+    if let Err(e) = install_sigusr1_handler() {
+        if i.settings.status != Some(StatusLevel::None) {
             eprintln!("{}\n\t{e}", translate!("dd-warning-signal-handler"));
         }
     }
@@ -1270,17 +1196,18 @@ fn dd_copy(mut i: Input, o: Output) -> io::Result<()> {
         // error.
         rstat += rstat_update;
         wstat += wstat_update;
-        match alarm.get_trigger() {
-            ALARM_TRIGGER_NONE => {}
-            t @ (ALARM_TRIGGER_TIMER | ALARM_TRIGGER_SIGNAL) => {
-                let tp = match t {
-                    ALARM_TRIGGER_TIMER => ProgUpdateType::Periodic,
-                    _ => ProgUpdateType::Signal,
-                };
-                let prog_update = ProgUpdate::new(rstat, wstat, start.elapsed(), tp);
-                prog_tx.send(prog_update).unwrap_or(());
-            }
-            _ => {}
+
+        #[cfg(target_os = "linux")]
+        if check_and_reset_sigusr1() {
+            let prog_update =
+                ProgUpdate::new(rstat, wstat, start.elapsed(), ProgUpdateType::Signal);
+            prog_tx.send(prog_update).unwrap_or(());
+        }
+        if Instant::now() >= next_progress_time {
+            let prog_update =
+                ProgUpdate::new(rstat, wstat, start.elapsed(), ProgUpdateType::Periodic);
+            prog_tx.send(prog_update).unwrap_or(());
+            next_progress_time += Duration::from_secs(1);
         }
     }
 

src/uu/dd/src/dd.rs

@@ -10,13 +10,10 @@
 //! [`gen_prog_updater`] function can be used to implement a progress
 //! updater that runs in its own thread.
 use std::io::Write;
-use std::sync::mpsc;
 #[cfg(target_os = "linux")]
-use std::thread::JoinHandle;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::mpsc;
 use std::time::Duration;
-
-#[cfg(target_os = "linux")]
-use signal_hook::iterator::Handle;
 use uucore::{
     error::{UResult, set_exit_code},
     format::num_format::{FloatVariant, Formatter},
@@ -448,47 +445,22 @@ pub(crate) fn gen_prog_updater(
     }
 }
 
-/// signal handler listens for SIGUSR1 signal and runs provided closure.
 #[cfg(target_os = "linux")]
-pub(crate) struct SignalHandler {
-    handle: Handle,
-    thread: Option<JoinHandle<()>>,
-}
+static SIGUSR1_RECEIVED: AtomicBool = AtomicBool::new(false);
 
 #[cfg(target_os = "linux")]
-impl SignalHandler {
-    pub(crate) fn install_signal_handler(
-        f: Box<dyn Send + Sync + Fn()>,
-    ) -> Result<Self, std::io::Error> {
-        use signal_hook::consts::signal::SIGUSR1;
-        use signal_hook::iterator::Signals;
-
-        let mut signals = Signals::new([SIGUSR1])?;
-        let handle = signals.handle();
-        let thread = std::thread::spawn(move || {
-            for signal in &mut signals {
-                match signal {
-                    SIGUSR1 => (*f)(),
-                    _ => unreachable!(),
-                }
-            }
-        });
+pub(crate) fn check_and_reset_sigusr1() -> bool {
+    SIGUSR1_RECEIVED.swap(false, Ordering::Relaxed)
+}
 
-        Ok(Self {
-            handle,
-            thread: Some(thread),
-        })
-    }
+#[cfg(target_os = "linux")]
+extern "C" fn sigusr1_handler(_: std::os::raw::c_int) {
+    SIGUSR1_RECEIVED.store(true, Ordering::Relaxed);
 }
 
 #[cfg(target_os = "linux")]
-impl Drop for SignalHandler {
-    fn drop(&mut self) {
-        self.handle.close();
-        if let Some(thread) = std::mem::take(&mut self.thread) {
-            thread.join().unwrap();
-        }
-    }
+pub(crate) fn install_sigusr1_handler() -> Result<(), nix::errno::Errno> {
+    uucore::signals::install_signal_handler(nix::sys::signal::Signal::SIGUSR1, sigusr1_handler)
 }
 
 /// Return a closure that can be used in its own thread to print progress info.

src/uu/dd/src/progress.rs

@@ -14,7 +14,8 @@
 use nix::errno::Errno;
 #[cfg(unix)]
 use nix::sys::signal::{
-    SigHandler::SigDfl, SigHandler::SigIgn, Signal::SIGINT, Signal::SIGPIPE, signal,
+    SaFlags, SigAction, SigHandler, SigHandler::SigDfl, SigHandler::SigIgn, SigSet, Signal,
+    Signal::SIGINT, Signal::SIGPIPE, sigaction, signal,
 };
 
 /// The default signal value.
@@ -435,6 +436,21 @@ pub fn ignore_interrupts() -> Result<(), Errno> {
     unsafe { signal(SIGINT, SigIgn) }.map(|_| ())
 }
 
+/// Installs a signal handler. The handler must be async-signal-safe.
+#[cfg(unix)]
+pub fn install_signal_handler(
+    sig: Signal,
+    handler: extern "C" fn(std::os::raw::c_int),
+) -> Result<(), Errno> {
+    let action = SigAction::new(
+        SigHandler::Handler(handler),
+        SaFlags::SA_RESTART,
+        SigSet::empty(),
+    );
+    unsafe { sigaction(sig, &action) }?;
+    Ok(())
+}
+
 // Detect closed stdin/stdout before Rust reopens them as /dev/null (see issue #2873)
 #[cfg(unix)]
 use std::sync::atomic::{AtomicBool, Ordering};