阻塞 I/O worker

何时使用这条路径

当一个组件拥有长期循环、循环可以阻塞且必须有序停止时，使用 BlockingIoExecutor。它不是任务队列、实时控制循环、协议 adapter 或设备接入框架。

有限且可排队的工作使用线程池；固定周期控制使用专用实时线程。worker 的协议、输入、输出、重试策略和安全行为仍由使用该库的项目负责。

Worker 契约

实现 IBlockingIoWorker::run(stop_token) 和 wakeup()：

• run() 可以等待，但 stop 请求可见后必须返回。
• wakeup() 必须解除当前等待，可重复调用且不得抛异常。
• stop token 本身不能中断任意外部等待；若等待原语不能直接唤醒，使用有限 timeout，并在每次返回后检查 token。

注册后 Facade 拥有 worker：它创建专属线程，在停止时调用 wakeup()，并在释放 worker 前完成 join。

可运行的 mock worker

该教程只用 condition variable 演示生命周期，不引入协议或硬件依赖：

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <iostream>
#include <memory>
#include <mutex>

#include <executor/executor.hpp>

namespace {

class MockBlockingWorker final : public executor::IBlockingIoWorker {
public:
    void run(executor::StopToken stop_token) override {
        std::unique_lock<std::mutex> lock(mutex_);
        started_.store(true, std::memory_order_release);
        condition_.notify_all();
        condition_.wait(lock, [this, stop_token] {
            return woken_ || stop_token.stop_requested();
        });
        stopped_.store(true, std::memory_order_release);
    }

    void wakeup() noexcept override {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            woken_ = true;
        }
        condition_.notify_all();
    }

    bool wait_until_started() {
        std::unique_lock<std::mutex> lock(mutex_);
        return condition_.wait_for(lock, std::chrono::seconds(1), [this] {
            return started_.load(std::memory_order_acquire);
        });
    }

    bool stopped() const {
        return stopped_.load(std::memory_order_acquire);
    }

private:
    std::atomic<bool> started_{false};
    std::atomic<bool> stopped_{false};
    std::mutex mutex_;
    std::condition_variable condition_;
    bool woken_ = false;
};

} // namespace

int main() {
    executor::Executor executor;
    executor::BlockingIoConfig config;
    config.thread_name = "tutorial_io";

    auto worker = std::make_unique<MockBlockingWorker>();
    MockBlockingWorker* worker_view = worker.get();
    executor::BlockingWorkerSpec spec{
        "tutorial_io", config, std::move(worker)};
    auto handle = executor.start_worker(std::move(spec));
    if (!handle.started() || !worker_view->wait_until_started()) {
        std::cerr << "blocking I/O worker start failed\n";
        executor.shutdown();
        return 1;
    }

    const auto running = handle.status();
    handle.stop();
    const auto stopped = handle.status();
    const bool worker_stopped = worker_view->stopped();

    const bool passed = running.is_running && worker_stopped &&
                        !stopped.is_running &&
                        stopped.stop_reason == executor::BlockingIoStopReason::Requested &&
                        stopped.wakeup_count == 1;
    std::cout << "blocking worker started=" << (running.is_running ? "yes" : "no")
              << ", stopped=" << (worker_stopped ? "yes" : "no")
              << ", wakeups=" << stopped.wakeup_count << '\n';
    executor.shutdown();
    return passed ? 0 : 1;
}

./build/examples/tutorial/tutorial_12_blocking_io_worker

blocking worker started=yes, stopped=yes, wakeups=1

生命周期与状态

1. 设置非空的 BlockingIoConfig::thread_name，将 std::unique_ptr<IBlockingIoWorker> 放入 BlockingWorkerSpec，并传给 start_worker()。
2. 需要启动诊断时读取 WorkerHandle::start_result()；WorkerHandle::started() 是其便捷成功判断。显式的注册/启动 API 仍保留，便于渐进迁移。
3. 通过 WorkerHandle::status()（或 get_blocking_io_worker_status(name)）读取状态。ready 只表示 executor 线程设置完成，不代表协议、设备或第一条输入已就绪。
4. 调用 WorkerHandle::request_stop() 可唤醒阻塞 worker 但不 join；调用 WorkerHandle::stop() 则请求停止、唤醒并 join。重复调用安全。

Executor::shutdown() 对所有已注册 I/O worker 采用同样的 stop/wake/join 规则，包括 shutdown(false)。不要 detach worker，也不要在 shutdown 后保留它的引用。

Executor 不负责的部分

本库刻意不决定消息所有权、队列策略、数据新鲜度、重连、设备安全动作或部署调优。这些由实现 worker 的应用自行定义和验证。

完整签名与状态字段见 API 参考。如果工作改为有固定周期预算的控制，请回到专用实时控制循环。