在工程开发中，线程的挂起与唤醒（Park / Unpark） 属于一种低级别（Low-level）、高效率的显式线程同步技术。

它不像重量级的异步运行时（如 Tokio）那样带有复杂的调度器结构，也不像传统的 Mutex + Condvar（条件变量）那样存在明显的加锁开销。它的核心使用场景主要集中在极致追求低延迟、低内存占用以及需要精细化控制线程行为的底层基础设施中。

以下是线程挂起与唤醒的 5 大核心工程应用场景：

1. 极致低延迟的线程池与任务调度器 (Low-Latency Thread Pools)

在高性能网络网关、金融高频交易系统或密集数据解码组件中，对任务分发的延迟要求通常在纳秒或微秒级。

2. 多生产者单消费者（MPSC）批处理管道 (Batching Pipelines)

在日志收集系统（如自定义异步日志组件）、监控指标（Metrics）聚合、或者高性能数据库的 Commit Log 刷盘中，通常有上百个业务线程在疯狂写数据，而只有 1 个后台线程负责打包消费。

3. 反应堆模式中的单线程事件循环 (Event Loops / Reactor Pattern)

在编写定制化的网络服务器、游戏引擎的 Tick 系统、或者嵌入式单片机管理中，经常需要一个常驻的“事件循环（Event Loop）”线程。

4. 协作式多任务“背压”控制 (Cooperative Backpressure)

在复杂的流水线（Pipeline）计算中，前级线程（生产数据）速度太快，而后级线程（消费数据）来不及处理，如果不加限制，内存会瞬间暴涨（OOM）。

5. 自定义并发原语与无锁数据结构的构建 (Building Custom Primitives)

如果你在开发一个底层的并发库，或者实现某种特定的同步锁（比如比标准库更轻量、具备特定自旋策略的 SpinLock 或 ReentrantLock），park 和 unpark 是最好的粘合剂。

💡 核心决策：什么时候该用它？

在进行技术选型时，你可以用下面这张简单的逻辑图来评估是否应该使用 park/unpark：

你的需求是什么？
  │
  ├─► 需要处理高并发网络 IO 或写复杂的异步业务（如 Web 后端） ──► 优先用 Tokio / async-await 运行时
  │
  ├─► 需要共享复杂的业务状态，数据读写时间长、对延迟不敏感  ──► 优先用传统的 Mutex / Condvar
  │
  └─► 极致低延迟、底层的无锁队列消费、线程数固定、追求零锁竞争 ──► 毫不犹豫选择 park / unpark

如果你的项目目前正在面临高并发下的延迟抖动，或者你需要在极其资源受限的嵌入式/微型容器环境下榨干 CPU 的每一分性能，基于 park 和 unpark 的无锁线程调度就是最顶级的解决方案。

• 状态机与任务分离：工作线程的挂起、唤醒、状态切换完全依靠 AtomicU8 状态机 与 park/unpark 协同完成，这是彻底无锁的。
• 锁临界区最小化：仅在“任务入队”和“工作线程摘取任务”的瞬间使用 Mutex（耗时仅几纳秒），而在任务执行期间，锁是完全释放的。
• 彻底杜绝虚假唤醒与惊群：配合原子计数器，主线程投递任务时，只会定向唤醒某一个正在处于 park 状态的空闲线程。

use std::collections::VecDeque;
use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::{self, Builder, JoinHandle, Thread};

// 线程状态机的四个核心状态
const STATE_IDLE: u8 = 0;    // 空闲中，准备或已经进入 park 挂起
const STATE_RUNNING: u8 = 1; // 正在执行具体的业务任务
const STATE_STOPPED: u8 = 2; // 线程池触发了停机通知

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

/// 内部工作线程的元数据
struct WorkerContext {
    thread_handle: Thread,
    state: Arc<AtomicU8>,
}

pub struct EngineeringThreadPool {
    // 任务队列：使用 Mutex 仅保护 VecDeque 的指针移动。
    // 注意：这里的 Mutex 绝不参与线程的挂起和唤醒，仅仅用于极快的数据出入队（几纳秒）
    task_queue: Arc<Mutex<VecDeque<Task>>>,
    // 线程池整体的生命周期控制
    pool_state: Arc<AtomicU8>,
    // 当前处于 IDLE (挂起/空闲) 状态的线程数量，用于精准唤醒
    idle_count: Arc<AtomicUsize>,
    // 维护所有工作线程的上下文，用于定向发送 unpark 信号
    workers: Vec<WorkerContext>,
    // 用于优雅停机时回收内核资源
    join_handles: Vec<JoinHandle<()>>,
}

impl EngineeringThreadPool {
    /// 初始化并启动工程级线程池
    /// - `num_threads`: 线程数量
    /// - `name_prefix`: 线程名显式前缀，便于生产环境 panic 溯源与监控
    pub fn new(num_threads: usize, name_prefix: &str) -> Self {
        let task_queue = Arc::new(Mutex::new(VecDeque::new()));
        let pool_state = Arc::new(AtomicU8::new(STATE_IDLE));
        let idle_count = Arc::new(AtomicUsize::new(0));

        let mut workers = Vec::with_capacity(num_threads);
        let mut join_handles = Vec::with_capacity(num_threads);
        let mut temp_contexts = Vec::with_capacity(num_threads);

        for i in 0..num_threads {
            let queue_clone = Arc::clone(&task_queue);
            let pool_state_clone = Arc::clone(&pool_state);
            let idle_clone = Arc::clone(&idle_count);
            let thread_state = Arc::new(AtomicU8::new(STATE_IDLE));
            let thread_state_clone = Arc::clone(&thread_state);

            // 1. 使用官方 Builder 显式配置线程工程参数
            let handle = Builder::new()
                .name(format!("{}-{}", name_prefix, i))
                .stack_size(2 * 1024 * 1024) // 显式分配 2MB 栈空间（防止深度递归 OOM）
                .spawn(move || {
                    loop {
                        // 检查停机状态：如果线程池已停止且队列被薅空，则安全退出
                        if pool_state_clone.load(Ordering::Acquire) == STATE_STOPPED {
                            if queue_clone.lock().unwrap().is_empty() {
                                break;
                            }
                        }

                        // 2. 核心防虚假唤醒逻辑
                        // 如果队列为空，且线程池还在运行，则将自身标记为 IDLE 并进入挂起
                        while queue_clone.lock().unwrap().is_empty() 
                            && pool_state_clone.load(Ordering::Acquire) != STATE_STOPPED 
                        {
                            thread_state_clone.store(STATE_IDLE, Ordering::Release);
                            idle_clone.fetch_add(1, Ordering::SeqCst);
                            
                            // 核心操作：挂起当前线程，释放 CPU 核心
                            thread::park(); 
                            
                            // 被唤醒（无论真实唤醒还是虚假唤醒），先扣减空闲计数
                            idle_clone.fetch_sub(1, Ordering::SeqCst);
                        }

                        // 3. 原子状态切换：标记当前线程进入运行态
                        thread_state_clone.store(STATE_RUNNING, Ordering::Release);

                        // 4. 窄临界区获取任务（批量消费模式优化）
                        let mut next_task = None;
                        if let Ok(mut q) = queue_clone.lock() {
                            next_task = q.pop_front();
                        } // 👈 锁在这里立即释放！执行任务时身上没有任何锁！

                        // 5. 执行业务任务
                        if let Some(task) = next_task {
                            // 生产级防线：捕获 panic，防止业务代码崩溃导致线程池意外缩容
                            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(task));
                        }
                    }
                })
                .expect("Failed to spawn OS thread");

            temp_contexts.push((handle.thread().clone(), thread_state));
            join_handles.push(handle);
        }

        for (thread_ref, t_state) in temp_contexts {
            workers.push(WorkerContext {
                thread_handle: thread_ref,
                state: t_state,
            });
        }

        Self {
            task_queue,
            pool_state,
            idle_count,
            workers,
            join_handles,
        }
    }

    /// 高并发提交任务（全标准库环境下极致优化的生产级接口）
    pub fn submit<F>(&self, task: F) -> Result<(), &'static str>
    where
        F: FnOnce() + Send + 'static,
    {
        if self.pool_state.load(Ordering::Acquire) == STATE_STOPPED {
            return Err("ThreadPool has been stopped");
        }

        // 1. 任务推入队列（获取互斥锁时间仅为指针移动的几纳秒）
        self.task_queue.lock().unwrap().push_back(Box::new(task));

        // 2. 精准定向唤醒（彻底解决惊群效应）
        // 如果当前有线程在睡觉（idle_count > 0），我们只挑出一个睡着的线程进行 unpark 唤醒
        if self.idle_count.load(Ordering::Relaxed) > 0 {
            for worker in &self.workers {
                // 利用原子状态机筛选：只对处于 STATE_IDLE 的线程发送 unpark 穿透信号
                if worker.state.load(Ordering::Acquire) == STATE_IDLE {
                    // 原子地将该线程状态切为 RUNNING，抢占该线程，防止其他 submit 重复唤醒它
                    if worker.state.compare_exchange(
                        STATE_IDLE, 
                        STATE_RUNNING, 
                        Ordering::AcqRel, 
                        Ordering::Acquire
                    ).is_ok() {
                        worker.thread_handle.unpark(); // 精准唤醒目标线程
                        break; // 成功唤醒一个，立即退出，绝不打扰其他线程
                    }
                }
            }
        }

        Ok(())
    }

    /// 优雅停机（Graceful Shutdown）
    pub fn shutdown(self) {
        // 1. 修改全局状态为已停止
        self.pool_state.store(STATE_STOPPED, Ordering::Release);
        
        // 2. 广播 unpark 信号，强制震醒所有可能在 park 深度冬眠的空闲线程
        for worker in &self.workers {
            worker.thread_handle.unpark();
        }

        // 3. 阻塞等待所有内核线程将队列残余任务执行完毕并安全退出
        for handle in self.join_handles {
            let _ = handle.join();
        }
    }
}

fn main() {
    // 部署一个纯官方标准库构建的 4 核心全应用级轻量线程池
    let pool = EngineeringThreadPool::new(4, "prod-worker");
    let pool = Arc::new(pool);

    // 模拟 4 个并发服务组件（MPSC 拓扑）同时向线程池疯狂灌入任务
    let mut injectors = vec![];
    for client_id in 1..=4 {
        let pool_clone = Arc::clone(&pool);
        let t = thread::spawn(move || {
            for task_id in 1..=5 {
                let pool_ref = Arc::clone(&pool_clone);
                let _ = pool_ref.submit(move || {
                    let current = thread::current();
                    println!(
                        "[{}] 正在安全处理来自客户端 {} 的生产任务 {}",
                        current.name().unwrap_or("unknown"),
                        client_id,
                        task_id
                    );
                    // 模拟密集业务耗时
                    thread::sleep(std::time::Duration::from_millis(20));
                });
            }
        });
        injectors.push(t);
    }

    // 等待所有上游组件投递完毕
    for t in injectors {
        t.join().unwrap();
    }

    // 回收 Arc 所有权以触发优雅停机
    if let Ok(pool_owned) = Arc::try_unwrap(pool) {
        println!("[主线程] 所有生产任务提交完毕，启动优雅停机流程...");
        pool_owned.shutdown();
    }

    println!("[主线程] 纯官方标准库轻量线程池安全关闭，进程正常结束。");
}

• 双层 CAS 抢占机制（无死锁、无竞争）：
  在任务提交（submit）时，主线程通过 worker.state.compare_exchange（原子比较并交换指令）在无锁状态下抢占工作线程的状态。一旦主线程成功把某个工作线程的状态从 IDLE 改为 RUNNING，才对它执行 unpark()。这保证了每次 submit 只会精准唤醒一个线程，彻底消除了 OS 层面的惊群效应。
• 锁的临界区缩减至“纳米级”：
  在传统设计中，工作线程拿到锁后，往往在锁的保护范围内顺便把业务任务也执行了。本方案中，工作线程加锁 -> pop_front() 弹出指针 -> 离开 if let 作用域自动释放锁 -> 执行任务。互斥锁只保护一个单纯的指针移动，耗时极短，多线程并发时几乎不会发生内核级锁互斥。
• 绝对无锁的挂起与唤醒（Zero-Lock Parking）：
  工作线程的挂起完全不依赖 Condvar（条件变量）。代码在调用 thread::park() 之前没有任何锁负担。这意味着当线程进入挂起睡眠时，它不占用任何内存锁资源。
• Panic 安全性防护（Panic Safety）：
  在分布式或高并发工程中，某个特定业务任务抛出 panic! 是极其常见的。如果直接让其崩溃，会导致工作线程死掉（线程池缩容）。本方案在核心消费区包裹了 std::panic::catch_unwind，任何任务的 panic 都会被安全捕获并隔离，工作线程会完好无损地继续消费下一个任务。
• 利用 Builder 实施防御性边界配置：
  通过 Builder::new().stack_size(...) 给工作线程显式指定了 2MB 的栈空间。在线上高并发生产环境（如解析大型 JSON、深度递归算法）中，这能有效防止因系统默认栈过小而导致的 Stack Overflow 生产事故。