Rust自定义Worker Pool编译失败，求助实现指定接口

问题描述 #

尝试实现一个Worker Pool，设计逻辑为：

• 结构体可添加作为任务的函数（支持捕获外部作用域数据）
• 生成与机器最大线程数一致的工作线程
• 每个线程从任务队列取任务执行，完成后继续获取下一个任务直至所有任务执行完毕

原实现代码：

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

pub struct ThreadPool<'a>
{
    task_queue: Vec<Task<'a>>,
    task_count: Arc<Mutex<usize>>,
}

pub struct Task<'a>(Box<dyn FnMut() + 'a>);
unsafe impl<'a> Send for Task<'a> {}

impl<'a> Task<'a>
{
    fn call(&mut self) { self.0() }
}

impl<'a> ThreadPool<'a>
{
    fn add_task<T>(&mut self, task: T)
    where
        T: 'a + FnMut() -> (),
    {
        self.task_queue.push(Task(Box::new(task)));
    }

    fn run(&mut self)
    {
        let thread_count = thread::available_parallelism().unwrap().get();
        println!("{}", thread_count);

        let mut handlers = Vec::with_capacity(thread_count);
        for _ in 0..thread_count
        {
            unsafe {
                let queue = &mut self.task_queue as *mut Vec<Task<'a>>;
                let task_count = Arc::clone(&self.task_count);
                handlers.push(thread::spawn(move || {
                    let index = task_count.lock().unwrap().overflowing_add(1).0 - 1;
                    (*queue)[index].call();
                }));
            }
        }
    }
}

期望使用接口：

let mut thread_pool = ThreadPool {
            task_queue: Vec::new(),
            task_count: Arc::new(Mutex::new(0)),
        };

        for i in 0..100
        {
            thread_pool.add_task(move || println!(r"ran {i} task"));
        }

        thread_pool.run();

编译错误信息 #

error[E0277]: *mut std::vec::Vec<Task<'a>> cannot be sent between threads safely
--> examples/06_fluid/thread_pool.rs:38:45
|
38 | handlers.push(thread::spawn(move || {
| ------------- ^------
| | |
| _______________________________|_____________within this [closure@examples/06_fluid/thread_pool.rs:38:45: 38:52]
| | |
| | required by a bound introduced by this call
39 | | let index = task_count.lock().unwrap().overflowing_add(1).0 - 1;
40 | | (*queue)[index].call();
41 | | }));
| |_________________^ *mut std::vec::Vec<Task<'a>> cannot be sent between threads safely
|
= help: within [closure@examples/06_fluid/thread_pool.rs:38:45: 38:52], the trait Send is not implemented for *mut std::vec::Vec<Task<'a>>
note: required because it's used within this closure
--> examples/06_fluid/thread_pool.rs:38:45
|
38 | handlers.push(thread::spawn(move || {
| ^^^^^^^
note: required by a bound in spawn
--> /home/makogan/.rustup/toolchains/nightly-2022-10-29-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:705:8
|
705 | F: Send + 'static,
| ^^^^ required by this bound in spawn

For more information about this error, try rustc --explain E0277.
error: could not compile neverengine due to previous error

错误分析 #

1. 裸指针的线程安全问题：用unsafe将task_queue转为*mut Vec<Task<'a>>，但裸指针默认不实现Send trait，而thread::spawn要求闭包必须满足Send + 'static约束，导致编译失败。
2. 任务分配逻辑缺陷：当前每个线程仅执行一个任务，未实现“循环取任务”的设计；用overflowing_add生成索引，会导致多线程重复获取索引或超出队列长度的问题。

修复方案 #

1. 用Arc<Mutex<VecDeque<Task<'a>>>>包装任务队列，通过Rust安全并发机制实现多线程共享与修改，替代不安全的裸指针。
2. 调整线程逻辑：每个线程循环从队列头部弹出任务，直到队列为空时退出。
3. 优化ThreadPool构造函数，避免用户手动初始化内部字段。
4. 为Task添加Send约束，确保任务可安全跨线程传递。

修复后的完整代码 #

use std::sync::{Arc, Mutex};
use std::thread;
use std::collections::VecDeque;

pub struct ThreadPool<'a> {
    task_queue: Arc<Mutex<VecDeque<Task<'a>>>>,
}

pub struct Task<'a>(Box<dyn FnMut() + 'a + Send>);

impl<'a> Task<'a> {
    fn call(&mut self) {
        self.0()
    }
}

impl<'a> ThreadPool<'a> {
    // 提供构造函数，简化初始化
    pub fn new() -> Self {
        Self {
            task_queue: Arc::new(Mutex::new(VecDeque::new())),
        }
    }

    pub fn add_task<T>(&mut self, task: T)
    where
        T: 'a + FnMut() + Send,
    {
        self.task_queue.lock().unwrap().push_back(Task(Box::new(task)));
    }

    pub fn run(&self) {
        let thread_count = thread::available_parallelism().unwrap().get();
        println!("Using {} threads", thread_count);

        let mut handlers = Vec::with_capacity(thread_count);
        let queue = Arc::clone(&self.task_queue);

        for _ in 0..thread_count {
            let queue_clone = Arc::clone(&queue);
            handlers.push(thread::spawn(move || {
                loop {
                    // 尝试从队列取出任务，队列为空则退出线程
                    let mut task = match queue_clone.lock().unwrap().pop_front() {
                        Some(task) => task,
                        None => break,
                    };
                    task.call();
                }
            }));
        }

        // 等待所有线程完成任务
        for handler in handlers {
            handler.join().unwrap();
        }
    }
}

验证示例 #

fn main() {
    let mut thread_pool = ThreadPool::new();

    for i in 0..100 {
        thread_pool.add_task(move || println!("ran {i} task"));
    }

    thread_pool.run();
}

内容的提问来源于stack exchange，提问作者Makogan