jtp 0.1.0

use std::{fmt::Display, sync::Arc, time::Duration};

use crossbeam_channel::bounded;

use crate::{task::TaskListeners, RejectedTaskHandler, ThreadPool};

/// An error returned from the [`ThreadPoolBuilder::build`].
#[derive(Debug)]
pub struct TPBuilderError {
    msg: String,
}

impl std::error::Error for TPBuilderError {}

impl Display for TPBuilderError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        writeln!(f, "{}", self.msg)
    }
}

macro_rules! tp_builder_error {
    ($($arg:tt)*) => {
        TPBuilderError { msg: format!($($arg)*) }
    };
}

/// A builder of the [`ThreadPool`].
///
/// # Examples
///
/// ```
/// use jtp::ThreadPoolBuilder;
/// use jtp::RejectedTaskHandler;
/// use std::time::Duration;
///
/// let thread_pool = ThreadPoolBuilder::default()
///     .set_core_pool_size(4)
///     .set_max_pool_size(7)
///     .set_keep_alive_time(Duration::from_secs(2))
///     .set_rejected_handler(RejectedTaskHandler::Discard)
///     .set_lisenter_before_execute(|id| println!("the task {} will be executed.", id))
///     .set_lisenter_before_execute(|id| println!("the task {} has been executed.", id))
///     .build()
///     .unwrap();
/// ```
pub struct ThreadPoolBuilder {
    channel_capacity: usize,
    max_pool_size: usize,
    core_pool_size: usize,
    keep_alive_time: Duration,
    rejected_task_handler: RejectedTaskHandler,
    task_lisenters: TaskListeners,
}

impl Default for ThreadPoolBuilder {
    /// Creates a new builder with the default configuration.
    ///
    /// # Default Configuration
    /// - channel_capacity: 50
    /// - max_pool_size: the number of physical cores of the current
    /// system
    /// - core_pool_size: the half of the `max_pool_size` at least 1
    /// - keep_alive_time: 1 second
    /// - rejected_task_handler: [`RejectedTaskHandler::Abort`]
    /// - before_execute: an empty closure `|| ()`
    /// - after_execute: an empty closure `|| ()`
    fn default() -> Self {
        Self {
            channel_capacity: 50,
            max_pool_size: num_cpus::get_physical(),
            core_pool_size: usize::max(1, num_cpus::get_physical() / 2),
            keep_alive_time: Duration::from_secs(1),
            rejected_task_handler: RejectedTaskHandler::Abort,
            task_lisenters: TaskListeners {
                before_execute: Box::new(|_| {}),
                after_execute: Box::new(|_| {}),
            },
        }
    }
}

impl ThreadPoolBuilder {
    /// Creates the base configuration for the new thread pool.
    ///
    /// See: [`ThreadPoolBuilder::default`]
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Sets the capacity of the task channel.
    #[must_use]
    pub fn set_channel_capacity(mut self, capacity: usize) -> Self {
        self.channel_capacity = capacity;
        self
    }

    /// Sets the maximum allowed number of threads.
    #[must_use]
    pub fn set_max_pool_size(mut self, size: usize) -> Self {
        self.max_pool_size = size;
        self
    }

    /// Sets the number of core threads.
    #[must_use]
    pub fn set_core_pool_size(mut self, size: usize) -> Self {
        self.core_pool_size = size;
        self
    }

    /// Sets the time that non-core threads may remain idle.
    #[must_use]
    pub fn set_keep_alive_time(mut self, time_ms: Duration) -> Self {
        self.keep_alive_time = time_ms;
        self
    }

    /// Sets the policy to handle rejected tasks.
    ///
    /// When you execute a task, the task may be handled by the given
    /// policy if the thread pool and the task channel are full.
    #[must_use]
    pub fn set_rejected_handler(mut self, handler: RejectedTaskHandler) -> Self {
        self.rejected_task_handler = handler;
        self
    }

    /// Sets the listener function that will be invoked before a task
    /// is executed.
    #[must_use]
    pub fn set_lisenter_before_execute<F>(mut self, executor: F) -> Self
    where
        F: Fn(usize) + Send + Sync + 'static,
    {
        self.task_lisenters.before_execute = Box::new(executor);
        self
    }

    /// Sets the listener function that will be invoked after a task
    /// is executed.
    #[must_use]
    pub fn set_lisenter_after_execute<F>(mut self, executor: F) -> Self
    where
        F: Fn(usize) + Send + Sync + 'static,
    {
        self.task_lisenters.after_execute = Box::new(executor);
        self
    }

    /// Creates a thread pool with the arguments.
    ///
    /// # Errors
    ///
    /// Returns an error if you build a thread pool with invalid
    /// arguments.
    ///
    /// # Examples
    ///
    /// ```
    /// use jtp::ThreadPoolBuilder;
    /// let thread_pool = ThreadPoolBuilder::default()
    ///     .set_max_pool_size(5)
    ///     .set_core_pool_size(6)
    ///     .build();
    ///
    /// // core_pool_size must <= max_pool_size
    /// assert!(thread_pool.is_err());
    ///
    /// let thread_pool = ThreadPoolBuilder::default()
    ///     .set_channel_capacity(5)
    ///     .set_max_pool_size(20)
    ///     .build();
    ///
    /// // max_pool_size  must <= channel_capacity
    /// assert!(thread_pool.is_err());
    /// ```
    pub fn build(self) -> Result<ThreadPool, TPBuilderError> {
        self.check_arguments()?;

        let (sender, reciver) = bounded(self.channel_capacity);
        Ok(ThreadPool {
            sender: Some(sender),
            reciver,
            task_lisenters: Arc::new(self.task_lisenters),
            core_workers: Vec::with_capacity(self.core_pool_size),
            workers: Vec::with_capacity(self.max_pool_size - self.core_pool_size),
            core_pool_size: self.core_pool_size,
            max_pool_size: self.max_pool_size,
            keep_alive_time: self.keep_alive_time,
            rejected_task_handler: self.rejected_task_handler,
            next_task_id: 0,
        })
    }

    fn check_arguments(&self) -> Result<(), TPBuilderError> {
        if self.channel_capacity < self.max_pool_size {
            return Err(tp_builder_error!(
                "Invalid arguments: max_tasks({}) < max_pool_size({}).",
                self.channel_capacity,
                self.max_pool_size
            ));
        }

        if self.max_pool_size < self.core_pool_size {
            return Err(tp_builder_error!(
                "Invalid arguments: max_pool_size({}) < core_pool_size({}).",
                self.max_pool_size,
                self.core_pool_size
            ));
        }

        Ok(())
    }
}