background_runner/

lib.rs

//! This crate provides the [`BackgroundRunner`] struct, which runs a
//! given task in the background once [`BackgroundRunner::update()`] is called.
//!
//! ```rust
//! use background_runner::BackgroundRunner;
//! use std::thread::sleep;
//! use std::time::Duration;
//! use std::time::Instant;
//!
//! let mut runner_iter = 0;
//!
//! // Set up the runner, giving it a task to run
//! let runner = BackgroundRunner::new(move |iter| {
//!     // Simulate some heavy work
//!     println!("runner_iter = {runner_iter}, iter = {iter}");
//!     runner_iter += 1;
//!     sleep(Duration::from_millis(10));
//! });
//!
//! let start = Instant::now();
//! let mut iter = 0;
//! while start.elapsed().as_millis() < 100 {
//!     // Update the runner with the current loop iteration
//!     runner.update(&iter);
//!     iter += 1;
//! }
//! ```

#[cfg(test)]
mod test;

use std::sync::Arc;
use std::sync::Condvar;
use std::sync::Mutex;
use std::sync::TryLockError;
use std::sync::atomic;
use std::sync::atomic::AtomicBool;

/// Runner for a background task to be invoked periodically.
#[derive(Debug)]
pub struct BackgroundRunner<T> {
    sync: Arc<Sync<T>>,
    background_thread: Option<std::thread::JoinHandle<()>>,
}

impl<T: 'static + Send> BackgroundRunner<T> {
    /// Create a new background runner with the given task
    ///
    /// This spawns a new thread to execute the task periodically, triggered by the [`Self::update()`] method.
    pub fn new(task: impl 'static + Send + FnMut(&T)) -> Self
    where
        T: Default,
    {
        Self::with_init_data(T::default(), task)
    }

    /// Create a new background runner with the given initial data and task
    ///
    /// This spawns a new thread to execute the task periodically, triggered by the [`Self::update()`] method.
    pub fn with_init_data(init_data: T, mut task: impl 'static + Send + FnMut(&T)) -> Self {
        let sync = Arc::new(Sync::new(init_data));
        let cloned_sync = Arc::clone(&sync);
        let background_thread = Some(std::thread::spawn(move || {
            loop {
                let sync_status = cloned_sync.wait_read(|data| task(data));
                if sync_status == SyncStatus::Disconnected {
                    break;
                }
            }
        }));

        Self {
            sync,
            background_thread,
        }
    }

    /// Trigger the runner's task if it's not currently running.
    ///
    /// Also see [`Self::update_with()`] for details.
    ///
    /// `T::clone_from()` is invoked to update the data that is passed to the task.
    pub fn update(&self, new_data: &T)
    where
        T: Clone,
    {
        self.update_with(|data| data.clone_from(new_data));
    }

    /// Trigger the runner's task if it's not currently running.
    ///
    /// This will never block the current thread. If the task is currently running
    /// and thus not able to process this update request, this update is dicarded
    /// and this method returns immediately without any effect (also `f` is not
    /// invoked in this case).
    ///
    /// The given closure is invoked with a mutable reference to the currently
    /// stored data to allow for its modification (e.g. via `clone_from`).
    pub fn update_with(&self, f: impl FnOnce(&mut T)) {
        self.sync.try_write(f);
    }

    /// Trigger the runner's task, blocking until it can be executed.
    ///
    /// Also see [`Self::wait_and_update_with()`] for details.
    ///
    /// `T::clone_from()` is invoked to update the data that is passed to the task.
    pub fn wait_and_update(&self, new_data: &T)
    where
        T: Clone,
    {
        self.wait_and_update_with(|data| data.clone_from(new_data));
    }

    /// Trigger the runner's task, blocking until it can be executed.
    ///
    /// Unlike [`Self::update_with()`], this method will block the current
    /// thread until the `new_data` can be handed over to the background task.
    /// Thus, it is ensured that the task will see this or a later submitted
    /// datum, given that is does not panic and the program is not terminated.
    ///
    /// This function is particularly useful in conjunction with
    /// [`Self::join()`] to ensure that the last update is processed completely.
    ///
    /// The given closure is invoked with a mutable reference to the currently
    /// stored data to allow for its modification.
    pub fn wait_and_update_with(&self, f: impl FnOnce(&mut T)) {
        self.sync.wait_write(f);
    }

    /// Wait for the background thread to finish.
    ///
    /// This function will wait until the background thread has finished executing
    /// it's last task and exited. This is useful to ensure that the last task
    /// has completed, in particular, before terminating the program.
    ///
    /// Notice, if the program is terminating, without calling this method,
    /// the background thread might be terminated abruptly, possibly
    /// in the middle of executing the task.
    pub fn join(mut self) {
        // Signal the background thread to exit
        self.sync.disconnect();

        // Wait for the background thread to finish
        if let Some(thread) = self.background_thread.take() {
            let _ = thread.join();
        };
    }
}

impl<T> Drop for BackgroundRunner<T> {
    fn drop(&mut self) {
        // Notify the runner thread without blocking here
        self.sync.disconnect();
    }
}

#[must_use]
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
enum SyncStatus {
    Connected,
    Disconnected,
}

#[derive(Debug)]
struct Sync<T> {
    mutex: Mutex<State<T>>,
    cvar: Condvar,
    is_disconnected: AtomicBool,
}

impl<T> Sync<T> {
    fn new(value: T) -> Self {
        let state = State::new(value);
        let mutex = Mutex::new(state);
        let cvar = Condvar::new();
        let is_disconnected = AtomicBool::new(false);

        Self {
            mutex,
            cvar,
            is_disconnected,
        }
    }

    /// Used by the runner thread to wait for a data update
    ///
    /// Returns `SyncStatus::Disconnected` if the background runner is disconnected.
    fn wait_read(&self, f: impl FnOnce(&T)) -> SyncStatus {
        let guard = self.mutex.lock().unwrap();
        let mut state = self
            .cvar
            .wait_while(guard, |state| {
                !state.is_dirty && !self.is_disconnected.load(atomic::Ordering::SeqCst)
            })
            .unwrap();

        if state.is_dirty {
            f(state.read());
        }

        if self.is_disconnected.load(atomic::Ordering::SeqCst) {
            return SyncStatus::Disconnected;
        }

        SyncStatus::Connected
    }

    /// Invokes the given closure if the background runner is not currently working.
    fn try_write(&self, f: impl FnOnce(&mut T)) {
        let mut state = match self.mutex.try_lock() {
            Ok(state) => state,
            Err(TryLockError::Poisoned(p)) => panic!("Runner panicked: {p}"),
            Err(TryLockError::WouldBlock) => return,
        };

        f(state.write());
        self.cvar.notify_all();
    }

    /// Invokes the given closure, blocking until the lock is acquired.
    fn wait_write(&self, f: impl FnOnce(&mut T)) {
        let mut state = self.mutex.lock().expect("Runner panicked");
        f(state.write());
        self.cvar.notify_all();
    }

    /// Marks this sync object as disconnected
    ///
    /// This is used to signal the runner thread about the background runner being dropped.
    fn disconnect(&self) {
        self.is_disconnected.store(true, atomic::Ordering::SeqCst);
        self.cvar.notify_all();
    }
}

/// State protected by the mutex
#[derive(Debug)]
struct State<T> {
    /// Actual data stored by calls to [`BackgroundRunner::update()`]
    data: T,

    /// Flag indicating whether the data has been updated since the last read
    is_dirty: bool,
}

impl<T> State<T> {
    fn new(data: T) -> Self {
        Self {
            data,
            is_dirty: false,
        }
    }

    /// Marks this state as not dirty and returns a reference to the data
    fn read(&mut self) -> &T {
        self.is_dirty = false;
        &self.data
    }

    /// Marks this state as dirty and returns a mutable reference to the data
    fn write(&mut self) -> &mut T {
        self.is_dirty = true;
        &mut self.data
    }
}