use std::sync::{mpsc, Arc, Mutex};
use std::thread;

// 1. 术语：使用类型别名（Type Alias）清晰定义一个可在线程间安全传递、仅执行一次的闭包任务
type Job = Box<dyn FnOnce() + Send + 'static>;

// 2. 工作者（Worker）：线程池内部的运行单元
struct Worker {
    id: usize,
    // 保持对 JoinHandle 的所有权，用于后续的优雅停机（Graceful Shutdown）
    thread: Option<thread::JoinHandle<()>>,
}

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Self {
        // 创建常驻子线程
        let thread = thread::spawn(move || loop {
            // 🚀 性能与安全关键点：
            // 1. 先获取 Mutex 锁，确保只有一个 Worker 能访问通道接收端。
            // 2. 使用 recv() 阻塞等待任务。如果通道被关闭（Sender 被 Drop），recv 会返回 Err。
            let message = receiver.lock().unwrap().recv();

            match message {
                Ok(job) => {
                    // println!("[Worker {}] 成功抢到任务，开始执行...", id);
                    job(); // 执行具体的闭包业务逻辑
                }
                Err(_) => {
                    // 术语：通道关闭信号。跳出循环，优雅结束线程
                    println!("[Worker {}] 收到退出信号，正在关闭线程...", id);
                    break;
                }
            }
        });

        Worker {
            id,
            thread: Some(thread),
        }
    }
}

// 3. 线程池核心结构体
pub struct ThreadPool {
    workers: Vec<Worker>,
    // 包装发送端，由于 mpsc::Sender 默认支持 Clone，如果高频写入，也可以使用多生产者发送任务
    sender: Option<mpsc::Sender<Job>>,
}

impl ThreadPool {
    /// 初始化指定线程数量的线程池
    /// # Panics
    /// 如果 size 为 0 则引发恐慌
    pub fn new(size: usize) -> Self {
        assert!(size > 0, "线程池大小必须大于 0");

        let (sender, receiver) = mpsc::channel();
        // 术语：通过 Arc + Mutex 赋予 Receiver 跨线程共享和内部可变性的能力
        let receiver = Arc::new(Mutex::new(receiver));

        let mut workers = Vec::with_capacity(size);
        for id in 0..size {
            // 每个 Worker 共享同一个接收端的引用克隆
            workers.push(Worker::new(id, Arc::clone(&receiver)));
        }

        ThreadPool {
            workers,
            sender: Some(sender),
        }
    }

    /// 向线程池投递任务
    /// 接收任意满足 Send 和 'static 约束的闭包
    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let job = Box::new(f);
        // 将任务推入通道中，等待 Worker 竞争抢夺
        if let Some(ref sender) = self.sender {
            sender.send(job).unwrap();
        }
    }
}

// 4. 优雅停机（Graceful Shutdown）的自动特质实现
impl Drop for ThreadPool {
    fn drop(&mut self) {
        println!("--- 正在触发线程池 Drop 销毁程序 ---");

        // 🚀 核心逻辑 1：首先隐式 Drop 掉发送端（Sender）
        // 这一步会切断通道，导致所有子线程内部的 `recv()` 瞬间收到 Err 并跳出 loop 循环
        drop(self.sender.take());

        // 🚀 核心逻辑 2：显式等待所有子线程彻底执行完当前手头任务并退出
        for worker in &mut self.workers {
            println!("正在汇合（Join）Worker 线程 {}", worker.id);
            if let Some(thread) = worker.thread.take() {
                thread.join().unwrap(); // 术语：Thread Join（线程阻塞汇合）
            }
        }
        println!("--- 线程池已完全安全释放 ---");
    }
}

// ==========================================
// 5. 实战并发测试（Main）
// ==========================================
fn main() {
    // 创建一个包含 4 个常驻工作线程的线程池
    let pool = ThreadPool::new(4);

    // 模拟高并发投放 8 个复杂的计算/IO任务
    for i in 0..8 {
        pool.execute(move || {
            println!("任务 {} 正在被线程 {:?} 执行", i, thread::current().id());
            thread::sleep(std::time::Duration::from_millis(50)); // 模拟任务耗时
        });
    }

    // 故意让主线程等待一下，观察任务并发执行的现象
    thread::sleep(std::time::Duration::from_millis(200));
    
    // 当 main 函数结束时，pool 会离开作用域，自动触发 `drop`，完成优雅停机验证。
}

如果要谈谈这个标准实现的工业级改进空间（或生产环境考量），可以从以下两个硬核方向进行展望，展现高级系统工程师的全局观：

use std::sync::{mpsc, Arc, Mutex, atomic::{AtomicUsize, Ordering}};
use std::time::Duration;
use std::thread;

type Job = Box<dyn FnOnce() + Send + 'static>;

// 1. 线程池共享的内部状态中心（避免大锁，使用原子变量）
struct PoolState {
    sender: mpsc::Sender<Job>,
    current_workers: AtomicUsize, // 当前存活的总线程数
    core_size: usize,             // 核心常驻线程数
    max_size: usize,              // 最大允许线程数
    keep_alive: Duration,         // 临时线程最大闲置存活时间
}

// 2. 动态工作单元
struct Worker;

impl Worker {
    fn spawn(id: usize, state: Arc<PoolState>, receiver: Arc<Mutex<mpsc::Receiver<Job>>>, is_core: bool) {
        state.current_workers.fetch_add(1, Ordering::SeqCst);
        
        thread::spawn(move || {
            loop {
                // 🚀 核心控制流：核心线程无限期阻塞等待；临时线程带超时等待
                let result = if is_core {
                    receiver.lock().unwrap().recv().map_err(|_| mpsc::RecvTimeoutError::Disconnected)
                } else {
                    // 临时线程若在此时间内抢不到锁或拿不到任务，就会触发超时
                    receiver.lock().unwrap().recv_timeout(state.keep_alive)
                };

                match result {
                    Ok(job) => {
                        job(); // 执行任务
                    }
                    Err(mpsc::RecvTimeoutError::Timeout) => {
                        // 术语：动态缩容（Scale Down）
                        // 临时线程超时未拿到任务，主动退出
                        println!("[临时工作者 {}] 闲置超时，开始自动销毁缩容...", id);
                        state.current_workers.fetch_sub(1, Ordering::SeqCst);
                        break;
                    }
                    Err(mpsc::RecvTimeoutError::Disconnected) => {
                        // 线程池被 Drop，通道关闭
                        state.current_workers.fetch_sub(1, Ordering::SeqCst);
                        break;
                    }
                }
            }
        });
    }
}

// 3. 动态线程池核心结构体
pub struct DynamicThreadPool {
    state: Arc<PoolState>,
    receiver: Arc<Mutex<mpsc::Receiver<Job>>>,
    next_worker_id: AtomicUsize,
}

impl DynamicThreadPool {
    pub fn new(core_size: usize, max_size: usize, keep_alive: Duration) -> Self {
        assert!(core_size <= max_size, "核心线程数不能大于最大线程数");
        
        let (sender, receiver) = mpsc::channel();
        let state = Arc::new(PoolState {
            sender,
            current_workers: AtomicUsize::new(0),
            core_size,
            max_size,
            keep_alive,
        });
        let receiver = Arc::new(Mutex::new(receiver));

        // 初始化时，提前拉起核心常驻线程（也可以选择懒加载预热）
        for id in 0..core_size {
            Worker::spawn(id, Arc::clone(&state), Arc::clone(&receiver), true);
        }

        DynamicThreadPool {
            state,
            receiver,
            next_worker_id: AtomicUsize::new(core_size),
        }
    }

    /// 投放任务（包含动态扩容逻辑）
    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let current = self.state.current_workers.load(Ordering::SeqCst);
        
        // 🚀 核心扩容决策（Scale Up）：
        // 如果当前线程数还没有达到最大限制（max_size），
        // 且当前所有存活的线程可能都在忙（标准库 channel 无法直接查看积压数，
        // 工业级通常配合带容量限制的带锁队列判断。此处我们设计为：只要有需求且未达 max_size，就允许激发临时线程扩容）
        if current < self.state.max_size && current >= self.state.core_size {
            let id = self.next_worker_id.fetch_add(1, Ordering::SeqCst);
            println!("[⚡ 触发扩容] 当前线程数 {}, 正在拉起临时工作者 {}...", current, id);
            Worker::spawn(id, Arc::clone(&self.state), Arc::clone(&self.receiver), false);
        }

        let job = Box::new(f);
        self.state.sender.send(job).unwrap();
    }

    pub fn current_thread_count(&self) -> usize {
        self.state.current_workers.load(Ordering::SeqCst)
    }
}

// ==========================================
// 4. 弹性扩缩容实战验证
// ==========================================
fn main() {
    // 创建动态线程池：2个核心常驻，最大允许5个，临时线程最大闲置50毫秒
    let pool = DynamicThreadPool::new(2, 5, Duration::from_millis(50));
    println!("初始常驻线程数: {}", pool.current_thread_count());

    // 1. 并发瞬间投递 10 个阻塞任务，逼迫线程池冲破核心线限制，扩容至 max_size (5个)
    for i in 0..10 {
        pool.execute(move || {
            thread::sleep(Duration::from_millis(30)); // 故意让任务执行慢一点，产生积压
            println!("任务 {} 执行完毕", i);
        });
    }

    // 观察瞬间扩容状态
    thread::sleep(Duration::from_millis(5));
    println!("🔥 洪峰期当前活动线程数: {}", pool.current_thread_count());

    // 2. 等待洪峰过去，任务全部执行完，并进入闲置状态
    thread::sleep(Duration::from_millis(150));

    // 观察自动缩容状态
    println!("🍃 恢复平静后（临时线程已超时释放）当前线程数: {}", pool.current_thread_count());
}