ncd 0.1.3

//! General purpose, "Fire 'n Forget", threadpool.

mod pool;
mod queue;

use std::{sync::OnceLock, thread, time::Duration};

use parking_lot::Mutex;
use queue::*;

pub use pool::*;

static POOL: OnceLock<ThreadPool> = OnceLock::new();
static INIT_LOCK: Mutex<()> = Mutex::new(());

/// Initialize the global thread pool.
///
/// # Panics
/// * Panics if `num_threads` is zero.
/// * Panics if a worker thread cannot be created.
#[allow(clippy::result_unit_err)]
pub fn init(name_pattern: &str, num_threads: usize) -> Result<(), ()> {
    let _guard = INIT_LOCK.lock();
    if POOL.get().is_some() {
        return Err(());
    }

    POOL.set(ThreadPool::new(name_pattern, num_threads))
        .map_err(|_| ())
}

/// Spawn a background loop on the global threadpool that repeatedly runs `f()`
/// and sleeps between iterations until `ok()` returns `true`.
///
/// ## Semantics
/// - **Order**: This uses **post-condition** semantics. Each iteration:
///   1) calls `f()`, 2) checks `ok()`, 3) sleeps.
///
///   That means `f()` will run **at least once**, even if `ok()` would have
///   returned `true` before the first iteration.
/// - **Period**: The effective period is roughly `f()` runtime **plus**
///   `sleep_for` (this is not a precise timer).
/// - **Cancellation / Shared State**: Because the task runs on a background
///   thread, any state observed by `f`/`ok` must be `'static` (e.g. use
///   `Arc<AtomicBool>` for a stop flag).
/// - **Pool Capacity**: Each active loop occupies one worker, including while
///   sleeping. Enough long-running loops can prevent other tasks from running.
///
/// ## Panics
/// - An unwinding panic inside `f()` or `ok()` aborts the task. Panics aborting
///   the process cannot be caught.
pub fn loop_and_sleep_until<F, P>(f: F, ok: P, sleep_for: Duration)
where
    F: FnMut() + Send + 'static,
    P: FnMut() -> bool + Send + 'static,
{
    submit(move || {
        let mut f = f;
        let mut ok = ok;

        loop {
            f();

            if ok() {
                break;
            }

            thread::sleep(sleep_for);
        }
    });
}

/// Submit a task to the global threadpool.
///
/// # Notes
/// * If `init` was not called, there will be 4 threads.
///
/// # Panics
/// * Panics if the default worker threads cannot be created.
pub fn submit<F: FnOnce() + Send + 'static>(task: F) {
    global_pool().submit(task);
}

fn global_pool() -> &'static ThreadPool {
    if let Some(pool) = POOL.get() {
        return pool;
    }

    let _guard = INIT_LOCK.lock();
    if let Some(pool) = POOL.get() {
        return pool;
    }

    let result = POOL.set(ThreadPool::new("ncd-dsptch", 4));
    debug_assert!(result.is_ok());
    POOL.get().expect("global pool was just initialized")
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn repeated_init_does_not_construct_another_pool() {
        assert!(init("global", 1).is_ok());

        // An interior NUL would panic during thread construction. Returning
        // Err proves the existing pool is checked first.
        assert!(init("not\0constructed", 1).is_err());
    }
}