如何中断Boost Asio Thread Pool中的所有线程？

问题背景 #

我之前用boost::thread时，靠boost::this_thread::interruption_point()实现线程中断，代码如下：

void do_long_calculations()
{
  for (...)
  {
    boost::this_thread::interruption_point();
    do_some_work();
  }
}

auto t = boost::thread(do_long_calculations);
...
t.interrupt();
t.join();

现在切换到boost::asio::thread_pool，发现它没有标准方法中断运行中的任务，想在销毁线程池前中断所有任务，尝试过直接操作线程ID或底层线程数据，但都不生效：

// 尝试1：通过线程ID中断，找不到对应函数
boost::asio::thread_pool p;
std::set<boost::thread::id> thread_ids;
post(p, []()
{
  thread_ids.insert(boost::this_thread::get_id());
  do_long_calculations();
});
...
for (auto id : thread_ids)
  SOME_INTERRUPTION_FUNCTION(id); // 无对应实现
p.stop();

// 尝试2：操作底层线程数据，无法获取有效指针
boost::asio::thread_pool p;
std::set<boost::detail::thread_data_ptr> threads;
post(p, []()
{
  if (auto d = boost::detail::get_current_thread_data())
    threads.insert(d); // 永远执行不到这里
  do_long_calculations();
});
...
for (auto &d : threads)
  d->interrupt();
p.stop();

该怎么实现需求？是否需要替换boost::this_thread::interruption_point()？

解决方案 #

Boost.Asio线程池的线程属于池内共享资源，官方不提供直接中断线程的接口（避免破坏池的内部状态），因此协作式中断是最安全可行的方案，替代原有的boost中断机制：

1. 用线程安全的中断标志替代interruption_point()

每个任务关联一个线程安全的中断标志，任务循环中主动检查标志，收到中断信号后自行退出。这种方式完全兼容线程池的设计，不会影响池内其他任务。

示例代码：

#include <boost/asio/thread_pool.hpp>
#include <boost/asio/post.hpp>
#include <atomic>
#include <vector>
#include <memory>

void do_long_calculations(std::shared_ptr<std::atomic<bool>> interrupted)
{
    for (int i = 0; i < 1000; ++i)
    {
        // 替代boost::this_thread::interruption_point()的检查逻辑
        if (*interrupted) {
            break; // 主动响应中断
        }
        do_some_work(); // 原有业务逻辑
    }
}

int main()
{
    boost::asio::thread_pool p(4); // 初始化4线程的线程池
    std::vector<std::shared_ptr<std::atomic<bool>>> interrupt_tokens;

    // 提交带中断令牌的任务
    for (int idx = 0; idx < 5; ++idx) {
        auto token = std::make_shared<std::atomic<bool>>(false);
        interrupt_tokens.push_back(token);
        boost::asio::post(p, [token]() {
            do_long_calculations(token);
        });
    }

    // 触发所有任务中断
    for (auto& token : interrupt_tokens) {
        *token = true;
    }

    // 停止线程池并等待所有任务结束
    p.stop();
    p.join();
    return 0;
}

2. 封装可中断任务的提交工具（可选）

如果需要批量管理中断，可以封装一个工具类，统一管理所有任务的中断令牌，简化操作：

class InterruptibleTaskPool {
public:
    InterruptibleTaskPool(size_t thread_count) : pool_(thread_count) {}

    template<typename Func>
    void submit(Func&& func) {
        auto token = std::make_shared<std::atomic<bool>>(false);
        tokens_.push_back(token);
        boost::asio::post(pool_, [token, func = std::forward<Func>(func)]() {
            func(token);
        });
    }

    void interrupt_all() {
        for (auto& token : tokens_) {
            *token = true;
        }
    }

    void stop_and_join() {
        pool_.stop();
        pool_.join();
    }

private:
    boost::asio::thread_pool pool_;
    std::vector<std::shared_ptr<std::atomic<bool>>> tokens_;
};

// 使用示例
void my_task(std::shared_ptr<std::atomic<bool>> interrupted) {
    while (!*interrupted) {
        // 任务逻辑
    }
}

int main() {
    InterruptibleTaskPool pool(4);
    pool.submit(my_task);
    pool.submit(my_task);

    pool.interrupt_all();
    pool.stop_and_join();
    return 0;
}

3. 为什么直接操作线程池线程不可行？

• Boost.Asio线程池的底层线程是内部管理的，boost::detail命名空间的接口属于未公开的实现细节，依赖它会导致代码兼容性差、容易崩溃；
• 线程池的线程会被复用，一个线程可能先后执行多个任务，直接中断线程会导致后续任务被意外终止，破坏线程池的正常工作流程。

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