Blocking I/O Workers

Use this path for a long-lived wait

Use start_worker(BlockingWorkerSpec) when a component owns one long-lived loop that can block and must still stop cleanly. It returns a WorkerHandle for startup and lifecycle management; it is not a task queue, a real-time control loop, a protocol adapter, or a device-integration framework. The explicit BlockingIoExecutor APIs remain available for incremental migration and diagnostics.

Use the thread pool for finite queueable work. Use a dedicated real-time thread for fixed-period control. The worker's protocol, inputs, outputs, retry policy, and safety behavior remain the library consumer's responsibility.

The worker contract

Implement IBlockingIoWorker::run(stop_token) and wakeup().

• run() may wait, but must return after a stop request becomes observable.
• wakeup() must release the current wait, may be called repeatedly, and must not throw.
• A stop token alone does not interrupt an arbitrary external wait. If the wait primitive cannot be awakened directly, use a bounded timeout and check the token after every return.

The Facade owns the worker after registration. It starts the dedicated thread, calls wakeup() during stopping, and joins before releasing the worker.

Runnable mock worker

The tutorial uses a condition variable only to demonstrate the lifecycle without a protocol or hardware dependency:

#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

Lifecycle and status

1. Configure a nonempty BlockingIoConfig::thread_name and pass a std::unique_ptr<IBlockingIoWorker> in BlockingWorkerSpec to start_worker().
2. Inspect WorkerHandle::start_result() when startup diagnostics are needed; WorkerHandle::started() is its convenience success check. The explicit register/start APIs remain available for incremental migration.
3. Observe WorkerHandle::status() (or get_blocking_io_worker_status(name)). ready describes executor-thread setup only; it does not mean a protocol, device, or first input is ready.
4. Call WorkerHandle::request_stop() to wake a blocked worker without joining, or WorkerHandle::stop() to request stop, wake it, and join. Repeated calls are safe.

Executor::shutdown() applies the same stop/wake/join rule to every registered I/O worker, including shutdown(false). Do not detach a worker or retain references to it after shutdown.

What remains outside Executor

This library deliberately does not decide message ownership, queue policy, data freshness, reconnect behavior, device safety actions, or deployment tuning. Define and test those concerns in the application that implements the worker.

For complete signatures and status fields, see the API reference. Next: return to real-time control when the work instead has a fixed-period budget.