//! [![crates.io version](https://img.shields.io/crates/v/safina-threadpool.svg)](https://crates.io/crates/safina-threadpool)
//! [![license: Apache 2.0](https://gitlab.com/leonhard-llc/safina-rs/-/raw/main/license-apache-2.0.svg)](http://www.apache.org/licenses/LICENSE-2.0)
//! [![unsafe forbidden](https://gitlab.com/leonhard-llc/safina-rs/-/raw/main/unsafe-forbidden-success.svg)](https://github.com/rust-secure-code/safety-dance/)
//! [![pipeline status](https://gitlab.com/leonhard-llc/safina-rs/badges/main/pipeline.svg)](https://gitlab.com/leonhard-llc/safina-rs/-/pipelines)
//!
//! A threadpool.
//!
//! You can use it alone or with [`safina`](https://crates.io/crates/safina),
//! a safe async runtime.
//!
//! # Features
//! - Add closures or `FnOnce` to the pool
//! - One of the pool's threads will execute it
//! - Automatically rstarts panicked threads
//! - `forbid(unsafe_code)`
//! - Depends only on `std`
//! - 100% test coverage
//!
//! # Limitations
//! - Allocates memory
//! - Not optimized
//!
//! # Documentation
//! <https://docs.rs/safina-threadpool>
//!
//! # Examples
//! ```rust
//! # type ProcessResult = ();
//! # fn process_data(data: (), sender: std::sync::mpsc::Sender<ProcessResult>) -> ProcessResult {
//! #    sender.send(()).unwrap();
//! # }
//! # fn f() {
//! # let data_source = vec![(),()];
//! let pool =
//!     safina_threadpool::ThreadPool::new("worker", 2);
//! let receiver = {
//!     let (sender, receiver) =
//!         std::sync::mpsc::channel();
//!     for data in data_source {
//!         let sender_clone = sender.clone();
//!         pool.schedule(
//!             move || process_data(data, sender_clone));
//!     }
//!     receiver
//! };
//! let results: Vec<ProcessResult> =
//!     receiver.iter().collect();
//! // ...
//! # }
//! ```
//!
//! # Alternatives
//! - [`blocking`](https://crates.io/crates/blocking)
//!   - Popular
//!   - A little `unsafe` code
//! - [`threadpool`](https://crates.io/crates/threadpool)
//!   - Popular
//!   - Well maintained
//!   - Dependencies have `unsafe` code
//! - [`futures-executor`](https://crates.io/crates/futures-executor)
//!   - Very popular
//!   - Full of `unsafe`
//! - [`scoped_threadpool`](https://crates.io/crates/scoped_threadpool)
//!   - Popular
//!   - Contains `unsafe` code
//! - [`scheduled-thread-pool`](https://crates.io/crates/scheduled-thread-pool)
//!   - Used by a popular connection pool library
//!   - Dependencies have `unsafe` code
//! - [`workerpool`](https://crates.io/crates/workerpool)
//!   - Dependencies have `unsafe` code
//! - [`threads_pool`](https://crates.io/crates/threads_pool)
//!   - Full of `unsafe`
//! - [`thread-pool`](https://crates.io/crates/thread-pool)
//!   - Old
//!   - Dependencies have `unsafe` code
//! - [`tasque`](https://crates.io/crates/tasque)
//!   - Dependencies have `unsafe` code
//! - [`fast-threadpool`](https://crates.io/crates/fast-threadpool)
//!   - Dependencies have `unsafe` code
//! - [`blocking-permit`](https://crates.io/crates/blocking-permit)
//!   - Full of `unsafe`
//! - [`rayon-core`](https://crates.io/crates/rayon-core)
//!   - Full of `unsafe`
//!
//! # Changelog
//! - v0.1.3 - Support stable Rust!  Needs 1.51+.
//! - v0.1.2 - Add another example
//! - v0.1.1 - Simplified internals and improved documentation.
//! - v0.1.0 - First release
//!
//! # TO DO
//! - DONE - Add `schedule` and `try_schedule`
//! - DONE - Add tests
//! - DONE - Add docs
//! - DONE - Publish on crates.io
//! - Add a stress test
//! - Add a benchmark.  See benchmarks in <https://crates.io/crates/executors>
//! - Add a way for a job to schedule another job on the same thread, with stealing.
//!
//! # Release Process
//! 1. Edit `Cargo.toml` and bump version number.
//! 1. Run `./release.sh`
#![forbid(unsafe_code)]

#[cfg(test)]
mod tests;

use core::fmt::{Display, Formatter};
use core::sync::atomic::{AtomicUsize, Ordering};
use core::time::Duration;
use std::error::Error;
use std::sync::mpsc::{Receiver, RecvTimeoutError, SyncSender, TrySendError};
use std::sync::{Arc, Mutex};

struct AtomicCounter {
    next_value: AtomicUsize,
}
impl AtomicCounter {
    pub fn new() -> Self {
        Self {
            next_value: AtomicUsize::new(0),
        }
    }
    pub fn next(&self) -> usize {
        self.next_value.fetch_add(1, Ordering::AcqRel)
    }
}

/// Returned by [`try_schedule`](struct.ThreadPool.html#method.try_schedule)
/// when the queue is full.
/// This can happen when the program schedules many closures at one time.
/// It can also happen when closures panic their threads.  The pool's
/// throughput goes down when it must create new threads.
#[derive(Debug)]
pub struct QueueFull {}
impl Display for QueueFull {
    fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> {
        std::fmt::Debug::fmt(self, f)
    }
}
impl Error for QueueFull {}

struct Inner {
    name: &'static str,
    next_name_num: AtomicCounter,
    size: usize,
    receiver: Mutex<Receiver<Box<dyn FnOnce() + Send>>>,
}
impl Inner {
    pub fn num_live_threads(self: &Arc<Inner>) -> usize {
        Arc::strong_count(self) - 1
    }

    fn work(self: &Arc<Inner>) {
        loop {
            let recv_result = self
                .receiver
                .lock()
                .unwrap()
                .recv_timeout(Duration::from_millis(500));
            self.start_threads();
            match recv_result {
                Ok(f) => f(),
                Err(RecvTimeoutError::Timeout) => {}
                // ThreadPool was dropped.
                Err(RecvTimeoutError::Disconnected) => return,
            };
            self.start_threads();
        }
    }

    fn start_thread(self: &Arc<Inner>) {
        let self_clone = self.clone();
        if self.num_live_threads() <= self.size {
            std::thread::Builder::new()
                .name(format!("{}{}", self.name, self.next_name_num.next()))
                .spawn(move || self_clone.work())
                .unwrap();
        }
    }

    fn start_threads(self: &Arc<Inner>) {
        while self.num_live_threads() < self.size {
            self.start_thread();
        }
    }
}

/// A collection of threads and a queue for jobs (`FnOnce` structs) they execute.
///
/// Threads die when they execute a job that panics.
/// If one thread survives, it will recreate all the threads.
/// The next call to [`schedule`](#method.schedule) or [`try_schedule`](#method.try_schedule)
/// also recreates threads.
///
/// If your threadpool load is bursty and you want to automatically recover
/// from an all-threads-panicked state, you could use
/// [`safina_timer`](https://crates.io/crates/safina-timer) to periodically call
/// [`schedule`](#method.schedule) or [`try_schedule`](#method.try_schedule).
///
/// # Example
/// ```rust
/// # type ProcessResult = ();
/// # fn process_data(data: (), sender: std::sync::mpsc::Sender<ProcessResult>) -> ProcessResult {
/// #    sender.send(()).unwrap();
/// # }
/// # fn f() {
/// # let data_source = vec![(),()];
/// let pool =
///     safina_threadpool::ThreadPool::new("worker", 2);
/// let receiver = {
///     let (sender, receiver) =
///         std::sync::mpsc::channel();
///     for data in data_source {
///         let sender_clone = sender.clone();
///         pool.schedule(
///             move || process_data(data, sender_clone));
///     }
///     receiver
/// };
/// let results: Vec<ProcessResult> =
///     receiver.iter().collect();
/// // ...
/// # }
/// ```
///
/// ```rust
/// # use core::time::Duration;
/// # use std::sync::Arc;
/// let pool =
///     Arc::new(safina_threadpool::ThreadPool::new("worker", 2));
/// let executor = safina_executor::Executor::default();
/// safina_timer::start_timer_thread();
/// let pool_clone = pool.clone();
/// executor.spawn(async move {
///     loop {
///         safina_timer::sleep_for(Duration::from_millis(500)).await;
///         pool_clone.schedule(|| {});
///     }
/// });
/// # assert_eq!(2, pool.num_live_threads());
/// # for _ in 0..2 {
/// #     pool.schedule(|| {
/// #         std::thread::sleep(Duration::from_millis(100));
/// #         panic!("ignore this panic")
/// #     });
/// # }
/// # std::thread::sleep(Duration::from_millis(200));
/// # assert_eq!(0, pool.num_live_threads());
/// # std::thread::sleep(Duration::from_millis(500));
/// # assert_eq!(2, pool.num_live_threads());
/// ```
pub struct ThreadPool {
    inner: Arc<Inner>,
    sender: SyncSender<Box<dyn FnOnce() + Send>>,
}

impl ThreadPool {
    /// Creates a new thread pool containing `size` threads.
    /// The threads all start immediately.
    ///
    /// Threads are named with `name` with a number.
    /// For example, `ThreadPool::new("worker", 2)`
    /// creates threads named "worker-1" and "worker-2".
    /// If one of those threads panics, the pool creates "worker-3".
    ///
    /// After the `ThreadPool` struct drops, the threads continue processing
    /// jobs and stop when the queue is empty.
    ///
    /// # Panics
    /// Panics if `name` is empty or `size` is zero.
    #[must_use]
    pub fn new(name: &'static str, size: usize) -> Self {
        if name.is_empty() {
            panic!("ThreadPool::new called with empty name")
        }
        if size < 1 {
            panic!("ThreadPool::new called with invalid size value: {:?}", size)
        }
        // Use a channel with bounded size.
        // If the channel was unbounded, the process could OOM when throughput goes down.
        let (sender, receiver) = std::sync::mpsc::sync_channel(size * 200);
        let pool = ThreadPool {
            inner: Arc::new(Inner {
                name,
                next_name_num: AtomicCounter::new(),
                size,
                receiver: Mutex::new(receiver),
            }),
            sender,
        };
        pool.inner.start_threads();
        pool
    }

    /// Returns the number of threads in the pool.
    #[must_use]
    pub fn size(&self) -> usize {
        self.inner.size
    }

    /// Returns the number of threads currently alive.
    #[must_use]
    pub fn num_live_threads(&self) -> usize {
        self.inner.num_live_threads()
    }

    /// Adds a job to the queue.  The next idle thread will execute it.
    /// Jobs are started in FIFO order.
    ///
    /// Blocks when the queue is full.
    /// See [`try_schedule`](#method.try_schedule).
    ///
    /// Recreates any threads that panicked.
    ///
    /// Puts `f` in a [`Box`](https://doc.rust-lang.org/stable/std/boxed/struct.Box.html) before
    /// adding it to the queue.
    #[allow(clippy::missing_panics_doc)]
    pub fn schedule(&self, f: impl FnOnce() + Send + 'static) {
        // If all workers panicked and the queue is full, adding to the queue will
        // block forever.  So we start threads first.
        self.inner.start_threads();
        self.sender.send(Box::new(f)).unwrap();
    }

    /// Adds a job to the queue.  The next idle thread will execute it.
    /// Jobs are started in FIFO order.
    ///
    /// Recreates any threads that panicked.
    ///
    /// Puts `f` in a [`Box`](https://doc.rust-lang.org/stable/std/boxed/struct.Box.html) before
    /// adding it to the queue.
    ///
    /// # Errors
    /// Returns `Err(QueueFull)` when the queue is full.
    pub fn try_schedule(&self, f: impl FnOnce() + Send + 'static) -> Result<(), QueueFull> {
        let result = match self.sender.try_send(Box::new(f)) {
            Ok(_) => Ok(()),
            Err(TrySendError::Disconnected(_)) => unreachable!(),
            Err(TrySendError::Full(_)) => Err(QueueFull {}),
        };
        self.inner.start_threads();
        result
    }
}