std::jthread 停线程：自动 join 不等于能打断阻塞

传感器节点退出时偶尔卡住，最后只能让进程管理器发 SIGKILL。线程主体并不复杂：从队列取包、解码、发布。问题通常出在“没有数据”这条路径——工作线程睡在条件变量或阻塞 I/O 里，主线程设置了一个布尔量，却没有办法把它叫醒。

C++20 的 std::jthread 解决了线程所有权的一半问题。一个仍然 joinable() 的 jthread 析构时会先调用 request_stop()，再调用 join()。但停止请求是协作式的：工作函数必须观察 std::stop_token，等待点也必须能因停止请求返回。否则自动 join() 只是自动地一直等下去。

让空队列等待也能取消

std::condition_variable_any 提供接受 stop_token 的等待重载，适合这种队列：

#include <condition_variable>
#include <deque>
#include <mutex>
#include <stop_token>
#include <thread>

class Decoder {
 public:
  Decoder() : worker_([this](std::stop_token stop) { run(stop); }) {}

  void push(Packet packet) {
    {
      std::lock_guard lock(mutex_);
      queue_.push_back(std::move(packet));
    }
    ready_.notify_one();
  }

 private:
  void run(std::stop_token stop) {
    for (;;) {
      Packet packet;
      {
        std::unique_lock lock(mutex_);
        ready_.wait(lock, stop, [this] { return !queue_.empty(); });
        if (stop.stop_requested() && queue_.empty()) {
          return;
        }
        packet = std::move(queue_.front());
        queue_.pop_front();
      }
      decode(packet);  // 不要拿着 queue mutex 做重活
    }
  }

  std::mutex mutex_;
  std::condition_variable_any ready_;
  std::deque<Packet> queue_;
  std::jthread worker_;  // 最后声明，析构时最先停止并 join
};

成员声明顺序不是排版偏好。C++ 按声明的逆序析构成员；这里 worker_ 先析构并等待线程退出，队列、条件变量和 mutex 此时仍然活着。若把线程声明在最前面，状态对象会先被销毁，工作线程还可能访问它们，结果就是 use-after-free。

上面的退出策略是“停止后把已取出的包做完，空队列就退出”。如果业务要求排空队列，应把条件改为“收到停止且队列为空才退出”，并明确关闭后 push() 是否仍被允许。两种策略都可以，不能靠析构时机碰运气。

request_stop() 不是线程中断

下面这些操作不会因为 token 变成 stop 状态就自动返回：

• 没有超时的 recv()、设备 SDK read()；
• 普通 std::condition_variable::wait()；
• 第三方库内部的长时间推理或解码；
• 一个从不检查 token 的计算循环。

网络采集可用 poll()/select() 的有限超时周期性检查 token，或者用专门的唤醒文件描述符。也可以注册 std::stop_callback 去调用线程安全的取消 API，但回调会在成功发出停止请求的那个线程上同步执行；回调里做阻塞工作，会反过来卡住关闭线程。对 socket 使用 shutdown() 前还要处理文件描述符复用和并发 close 的生命周期问题。

显式关闭比“等析构”更容易验收

我倾向给组件留一个幂等的 stop()，让测试能观察关闭边界：

void stop() {
  if (worker_.joinable()) {
    worker_.request_stop();
    worker_.join();
  }
}

重复 request_stop() 是安全的；只有第一次真正发出请求的调用返回 true。join() 则不能重复调用，所以仍要检查 joinable()，并保证多个控制线程不会同时操作同一个 jthread 对象。

最小验证不是“程序最后能退出”，而是让 worker 确实停在空队列等待处，再调用 stop()：

Decoder decoder;
decoder.wait_until_worker_is_idle_for_test();
decoder.stop();
assert(decoder.processed_count() == 0);

测试可用 std::latch 确认工作线程已进入等待路径，避免用 sleep_for 猜时序。ThreadSanitizer 再覆盖“push 与 stop 同时发生”的竞争：

g++ -std=c++20 -pthread -fsanitize=thread -g decoder_test.cpp
./a.out

jthread 把“忘记 join”变成了 RAII，但它不会替第三方 I/O 设计取消协议。真正需要审查的是每一个可能无限等待的点，以及线程所访问对象的生命周期。