与外部事件循环互操作

应用可能已经拥有自己的事件循环——典型是 asio 的 io_context 与 strand。 executor 核心库不依赖 asio 或任何第三方事件循环；本指南描述两者如何正确协作、哪些派发是 executor 看不见的、以及让盲区保持安全的纪律。

完整指南见 docs/external_event_loop_interop.md。可编译的伴随示例（用互斥量 + 条件变量复现 strand 语义，不引入任何 SDK 依赖）是 examples/event_loop_interop.cpp：

cpp

// event_loop_interop.cpp
// 《外部事件循环互操作》指南的可编译伴随示例（docs/external_event_loop_interop.md）。
//
// 本示例不依赖 asio：用一个 MiniEventLoop（互斥量 + 双端队列 + 条件变量实现
// 的串行派发循环）等价复现 asio strand/io_context 的互操作模式——
//
//   模式 1（现行合规）：把事件循环托管为 Blocking I/O worker，
//             run(StopToken) 驱动循环，wakeup() 唤醒，start_worker 收尾；
//   模式 2（盲区纪律）：线程池任务通过 post() 把延续派发回串行循环。
//             这些 post 级派发不进入 executor 的 admission/统计/失败事件，
//             因此状态必须由 shared_ptr 拥有、延续只做对象内移交；
//   模式 3（收尾同步）：串行循环每完成一批工作推进 PhaseGate 相位，
//             线程池侧用 wait_for() 等待批次完成（无锁、可超时）。
//
// 对应到 asio：MiniEventLoop ≈ io_context + strand；post() ≈ asio::post(strand,
// ...)；run(StopToken) ≈ io_context::run() + stop 信号；wakeup() ≈
// post 一个空任务或 stop()。

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <deque>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include <executor/comm.hpp>
#include <executor/executor.hpp>

using namespace executor;
using namespace std::chrono_literals;

namespace {

// ---------------------------------------------------------------------------
// MiniEventLoop：asio strand 的最小等价物（所有 post 的任务串行执行）。
// 托管为 IBlockingIoWorker：生命周期、线程命名与 stop 语义全部交给 executor。
// ---------------------------------------------------------------------------
class MiniEventLoopWorker final : public IBlockingIoWorker {
public:
    // 相当于 asio::post(strand, task)：任意线程调用；任务严格串行执行。
    void post(std::function<void()> task) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            queue_.push_back(std::move(task));
        }
        cv_.notify_all();
    }

    // 相当于 io_context::run()：托管线程上串行执行所有已 post 的任务，
    // 直到 stop_token 置位并排空。
    void run(StopToken stop_token) override {
        std::unique_lock<std::mutex> lock(mutex_);
        running_.store(true, std::memory_order_release);
        cv_.notify_all();
        for (;;) {
            cv_.wait(lock, [this, &stop_token] {
                return !queue_.empty() || stop_token.stop_requested();
            });
            while (!queue_.empty()) {
                auto task = std::move(queue_.front());
                queue_.pop_front();
                lock.unlock();
                task();  // 串行执行：此后回调内访问的对象无需额外加锁
                lock.lock();
            }
            if (stop_token.stop_requested()) {
                break;
            }
        }
        running_.store(false, std::memory_order_release);
    }

    // executor 停止路径调用：让 run() 的等待立刻醒来检查 stop_token。
    void wakeup() noexcept override {
        cv_.notify_all();
    }

    bool wait_until_running(std::chrono::milliseconds timeout) {
        std::unique_lock<std::mutex> lock(mutex_);
        return cv_.wait_for(lock, timeout, [this] {
            return running_.load(std::memory_order_acquire);
        });
    }

private:
    std::atomic<bool> running_{false};
    std::mutex mutex_;
    std::condition_variable cv_;
    std::deque<std::function<void()>> queue_;
};

}  // namespace

int main() {
    Executor executor;
    ExecutorConfig config;
    config.min_threads = 2;
    config.max_threads = 4;
    if (!executor.initialize(config)) {
        std::cerr << "failed to initialize executor\n";
        return 1;
    }

    // -----------------------------------------------------------------------
    // 模式 1：把事件循环托管为 Blocking I/O worker（现行合规路线）。
    // -----------------------------------------------------------------------
    auto loop = std::make_unique<MiniEventLoopWorker>();
    MiniEventLoopWorker* loop_view = loop.get();

    BlockingIoConfig worker_config;
    worker_config.thread_name = "event_loop";
    WorkerHandle worker = executor.start_worker(
        BlockingWorkerSpec{"event_loop", worker_config, std::move(loop)});
    if (!worker.status().is_running ||
        !loop_view->wait_until_running(1000ms)) {
        std::cerr << "failed to start hosted event loop\n";
        return 1;
    }
    std::cout << "[1] event loop hosted as blocking I/O worker\n";

    // -----------------------------------------------------------------------
    // 模式 2：pool 任务 -> strand 延续（post 级派发的盲区纪律）。
    //
    // 盲区事实：loop_view->post(...) 的任务不经过 executor 提交路径，
    // 不进入 admission、TaskStatistics 与 failure 事件。纪律：
    //   a) 跨界状态用 shared_ptr 拥有，移交后原线程不再访问；
    //   b) 延续内部异常必须自捕获（盲区里没人替你记录失败）。
    // -----------------------------------------------------------------------
    auto value = std::make_shared<std::atomic<int>>(0);
    auto stage = std::make_shared<std::atomic<int>>(0);

    auto producer = executor.submit([value, stage, loop_view]() noexcept {
        value->store(41, std::memory_order_release);
        stage->store(1, std::memory_order_release);

        // 延续派发回串行循环：此后 value/stage 只在 strand 上访问。
        loop_view->post([value, stage]() noexcept {
            value->fetch_add(1, std::memory_order_acq_rel);
            stage->store(2, std::memory_order_release);
        });
    });
    producer.wait();

    while (stage->load(std::memory_order_acquire) != 2) {
        std::this_thread::yield();
    }
    std::cout << "[2] pool -> strand continuation, value=" << value->load()
              << " (post dispatches are outside executor statistics)\n";

    // -----------------------------------------------------------------------
    // 模式 3：PhaseGate 收尾——串行批次推进相位，pool 侧等待。
    // -----------------------------------------------------------------------
    comm::PhaseGate gate("event_loop_batch");
    constexpr uint64_t kBatchSize = 4;

    auto batch_future = executor.submit([&gate, kBatchSize]() {
        // pool 线程等待串行侧完成整批工作；wait_for 支持超时，不会无限等。
        const comm::CommResult result = gate.wait_for(kBatchSize, 5s);
        return result.ok;
    });

    for (uint64_t i = 0; i < kBatchSize; ++i) {
        loop_view->post([i, &gate]() noexcept {
            // 每个串行任务完成一批次中的一步，然后推进相位。
            gate.advance_to(i + 1);
        });
    }

    const bool batch_ok = batch_future.get();
    std::cout << "[3] phase gate batch finalized, ok=" << batch_ok << "\n";

    // -----------------------------------------------------------------------
    // 收尾：先停止托管循环（run 排空返回），再 shutdown 线程池。
    // -----------------------------------------------------------------------
    worker.request_stop();
    loop_view->wakeup();
    worker.stop();
    std::cout << "[4] hosted loop stopped: "
              << (worker.status().is_running ? "still running" : "stopped")
              << "\n";

    executor.shutdown();
    std::cout << "event loop interop example completed\n";
    return batch_ok ? 0 : 1;
}

模式 1：把循环托管为 Blocking I/O worker

围绕 io_context::run() 实现 IBlockingIoWorker::run(StopToken)，实现 wakeup() 让停止路径能唤醒循环。之后 Executor::start_worker() 拥有线程生命周期、命名与停止顺序，循环也会出现在 BlockingIoExecutorStatus 与 get_snapshot() 中。

cpp

// event_loop_interop.cpp
// 《外部事件循环互操作》指南的可编译伴随示例（docs/external_event_loop_interop.md）。
//
// 本示例不依赖 asio：用一个 MiniEventLoop（互斥量 + 双端队列 + 条件变量实现
// 的串行派发循环）等价复现 asio strand/io_context 的互操作模式——
//
//   模式 1（现行合规）：把事件循环托管为 Blocking I/O worker，
//             run(StopToken) 驱动循环，wakeup() 唤醒，start_worker 收尾；
//   模式 2（盲区纪律）：线程池任务通过 post() 把延续派发回串行循环。
//             这些 post 级派发不进入 executor 的 admission/统计/失败事件，
//             因此状态必须由 shared_ptr 拥有、延续只做对象内移交；
//   模式 3（收尾同步）：串行循环每完成一批工作推进 PhaseGate 相位，
//             线程池侧用 wait_for() 等待批次完成（无锁、可超时）。
//
// 对应到 asio：MiniEventLoop ≈ io_context + strand；post() ≈ asio::post(strand,
// ...)；run(StopToken) ≈ io_context::run() + stop 信号；wakeup() ≈
// post 一个空任务或 stop()。

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <deque>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include <executor/comm.hpp>
#include <executor/executor.hpp>

using namespace executor;
using namespace std::chrono_literals;

namespace {

// ---------------------------------------------------------------------------
// MiniEventLoop：asio strand 的最小等价物（所有 post 的任务串行执行）。
// 托管为 IBlockingIoWorker：生命周期、线程命名与 stop 语义全部交给 executor。
// ---------------------------------------------------------------------------
class MiniEventLoopWorker final : public IBlockingIoWorker {
public:
    // 相当于 asio::post(strand, task)：任意线程调用；任务严格串行执行。
    void post(std::function<void()> task) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            queue_.push_back(std::move(task));
        }
        cv_.notify_all();
    }

    // 相当于 io_context::run()：托管线程上串行执行所有已 post 的任务，
    // 直到 stop_token 置位并排空。
    void run(StopToken stop_token) override {
        std::unique_lock<std::mutex> lock(mutex_);
        running_.store(true, std::memory_order_release);
        cv_.notify_all();
        for (;;) {
            cv_.wait(lock, [this, &stop_token] {
                return !queue_.empty() || stop_token.stop_requested();
            });
            while (!queue_.empty()) {
                auto task = std::move(queue_.front());
                queue_.pop_front();
                lock.unlock();
                task();  // 串行执行：此后回调内访问的对象无需额外加锁
                lock.lock();
            }
            if (stop_token.stop_requested()) {
                break;
            }
        }
        running_.store(false, std::memory_order_release);
    }

    // executor 停止路径调用：让 run() 的等待立刻醒来检查 stop_token。
    void wakeup() noexcept override {
        cv_.notify_all();
    }

    bool wait_until_running(std::chrono::milliseconds timeout) {
        std::unique_lock<std::mutex> lock(mutex_);
        return cv_.wait_for(lock, timeout, [this] {
            return running_.load(std::memory_order_acquire);
        });
    }

private:
    std::atomic<bool> running_{false};
    std::mutex mutex_;
    std::condition_variable cv_;
    std::deque<std::function<void()>> queue_;
};

}  // namespace

int main() {
    Executor executor;
    ExecutorConfig config;
    config.min_threads = 2;
    config.max_threads = 4;
    if (!executor.initialize(config)) {
        std::cerr << "failed to initialize executor\n";
        return 1;
    }

    // -----------------------------------------------------------------------
    // 模式 1：把事件循环托管为 Blocking I/O worker（现行合规路线）。
    // -----------------------------------------------------------------------
    auto loop = std::make_unique<MiniEventLoopWorker>();
    MiniEventLoopWorker* loop_view = loop.get();

    BlockingIoConfig worker_config;
    worker_config.thread_name = "event_loop";
    WorkerHandle worker = executor.start_worker(
        BlockingWorkerSpec{"event_loop", worker_config, std::move(loop)});
    if (!worker.status().is_running ||
        !loop_view->wait_until_running(1000ms)) {
        std::cerr << "failed to start hosted event loop\n";
        return 1;
    }
    std::cout << "[1] event loop hosted as blocking I/O worker\n";

    // -----------------------------------------------------------------------
    // 模式 2：pool 任务 -> strand 延续（post 级派发的盲区纪律）。
    //
    // 盲区事实：loop_view->post(...) 的任务不经过 executor 提交路径，
    // 不进入 admission、TaskStatistics 与 failure 事件。纪律：
    //   a) 跨界状态用 shared_ptr 拥有，移交后原线程不再访问；
    //   b) 延续内部异常必须自捕获（盲区里没人替你记录失败）。
    // -----------------------------------------------------------------------
    auto value = std::make_shared<std::atomic<int>>(0);
    auto stage = std::make_shared<std::atomic<int>>(0);

    auto producer = executor.submit([value, stage, loop_view]() noexcept {
        value->store(41, std::memory_order_release);
        stage->store(1, std::memory_order_release);

        // 延续派发回串行循环：此后 value/stage 只在 strand 上访问。
        loop_view->post([value, stage]() noexcept {
            value->fetch_add(1, std::memory_order_acq_rel);
            stage->store(2, std::memory_order_release);
        });
    });
    producer.wait();

    while (stage->load(std::memory_order_acquire) != 2) {
        std::this_thread::yield();
    }
    std::cout << "[2] pool -> strand continuation, value=" << value->load()
              << " (post dispatches are outside executor statistics)\n";

    // -----------------------------------------------------------------------
    // 模式 3：PhaseGate 收尾——串行批次推进相位，pool 侧等待。
    // -----------------------------------------------------------------------
    comm::PhaseGate gate("event_loop_batch");
    constexpr uint64_t kBatchSize = 4;

    auto batch_future = executor.submit([&gate, kBatchSize]() {
        // pool 线程等待串行侧完成整批工作；wait_for 支持超时，不会无限等。
        const comm::CommResult result = gate.wait_for(kBatchSize, 5s);
        return result.ok;
    });

    for (uint64_t i = 0; i < kBatchSize; ++i) {
        loop_view->post([i, &gate]() noexcept {
            // 每个串行任务完成一批次中的一步，然后推进相位。
            gate.advance_to(i + 1);
        });
    }

    const bool batch_ok = batch_future.get();
    std::cout << "[3] phase gate batch finalized, ok=" << batch_ok << "\n";

    // -----------------------------------------------------------------------
    // 收尾：先停止托管循环（run 排空返回），再 shutdown 线程池。
    // -----------------------------------------------------------------------
    worker.request_stop();
    loop_view->wakeup();
    worker.stop();
    std::cout << "[4] hosted loop stopped: "
              << (worker.status().is_running ? "still running" : "stopped")
              << "\n";

    executor.shutdown();
    std::cout << "event loop interop example completed\n";
    return batch_ok ? 0 : 1;
}

模式 2：strand 延续合法但不可见

线程池任务可以把延续 post 回 strand，但必须认清这意味着什么：

asio::post(strand, ...) 不经过任何 executor 提交路径——没有 admission 判断，也没有排队计数。
被 post 的回调不出现在 TaskStatistics、在途诊断或 ExecutorSnapshot 中。
回调里抛出的异常不进入 executor 失败事件；asio 会吞掉或因此终止。

盲区内的纪律：跨界状态用 shared_ptr 显式移交，移交后原线程不再访问；延续内部自行捕获异常；需要 admission、背压或失败计量的工作必须留在 executor 提交 API 上，只把轻量延续 post 到 strand。

模式 3：用 PhaseGate 收尾批次

不要用轮询加 sleep 检测批次完成。串行侧每完成一步推进 comm::PhaseGate 相位；任意线程都可以带超时地等待：

cpp

// event_loop_interop.cpp
// 《外部事件循环互操作》指南的可编译伴随示例（docs/external_event_loop_interop.md）。
//
// 本示例不依赖 asio：用一个 MiniEventLoop（互斥量 + 双端队列 + 条件变量实现
// 的串行派发循环）等价复现 asio strand/io_context 的互操作模式——
//
//   模式 1（现行合规）：把事件循环托管为 Blocking I/O worker，
//             run(StopToken) 驱动循环，wakeup() 唤醒，start_worker 收尾；
//   模式 2（盲区纪律）：线程池任务通过 post() 把延续派发回串行循环。
//             这些 post 级派发不进入 executor 的 admission/统计/失败事件，
//             因此状态必须由 shared_ptr 拥有、延续只做对象内移交；
//   模式 3（收尾同步）：串行循环每完成一批工作推进 PhaseGate 相位，
//             线程池侧用 wait_for() 等待批次完成（无锁、可超时）。
//
// 对应到 asio：MiniEventLoop ≈ io_context + strand；post() ≈ asio::post(strand,
// ...)；run(StopToken) ≈ io_context::run() + stop 信号；wakeup() ≈
// post 一个空任务或 stop()。

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <deque>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include <executor/comm.hpp>
#include <executor/executor.hpp>

using namespace executor;
using namespace std::chrono_literals;

namespace {

// ---------------------------------------------------------------------------
// MiniEventLoop：asio strand 的最小等价物（所有 post 的任务串行执行）。
// 托管为 IBlockingIoWorker：生命周期、线程命名与 stop 语义全部交给 executor。
// ---------------------------------------------------------------------------
class MiniEventLoopWorker final : public IBlockingIoWorker {
public:
    // 相当于 asio::post(strand, task)：任意线程调用；任务严格串行执行。
    void post(std::function<void()> task) {
        {
            std::lock_guard<std::mutex> lock(mutex_);
            queue_.push_back(std::move(task));
        }
        cv_.notify_all();
    }

    // 相当于 io_context::run()：托管线程上串行执行所有已 post 的任务，
    // 直到 stop_token 置位并排空。
    void run(StopToken stop_token) override {
        std::unique_lock<std::mutex> lock(mutex_);
        running_.store(true, std::memory_order_release);
        cv_.notify_all();
        for (;;) {
            cv_.wait(lock, [this, &stop_token] {
                return !queue_.empty() || stop_token.stop_requested();
            });
            while (!queue_.empty()) {
                auto task = std::move(queue_.front());
                queue_.pop_front();
                lock.unlock();
                task();  // 串行执行：此后回调内访问的对象无需额外加锁
                lock.lock();
            }
            if (stop_token.stop_requested()) {
                break;
            }
        }
        running_.store(false, std::memory_order_release);
    }

    // executor 停止路径调用：让 run() 的等待立刻醒来检查 stop_token。
    void wakeup() noexcept override {
        cv_.notify_all();
    }

    bool wait_until_running(std::chrono::milliseconds timeout) {
        std::unique_lock<std::mutex> lock(mutex_);
        return cv_.wait_for(lock, timeout, [this] {
            return running_.load(std::memory_order_acquire);
        });
    }

private:
    std::atomic<bool> running_{false};
    std::mutex mutex_;
    std::condition_variable cv_;
    std::deque<std::function<void()>> queue_;
};

}  // namespace

int main() {
    Executor executor;
    ExecutorConfig config;
    config.min_threads = 2;
    config.max_threads = 4;
    if (!executor.initialize(config)) {
        std::cerr << "failed to initialize executor\n";
        return 1;
    }

    // -----------------------------------------------------------------------
    // 模式 1：把事件循环托管为 Blocking I/O worker（现行合规路线）。
    // -----------------------------------------------------------------------
    auto loop = std::make_unique<MiniEventLoopWorker>();
    MiniEventLoopWorker* loop_view = loop.get();

    BlockingIoConfig worker_config;
    worker_config.thread_name = "event_loop";
    WorkerHandle worker = executor.start_worker(
        BlockingWorkerSpec{"event_loop", worker_config, std::move(loop)});
    if (!worker.status().is_running ||
        !loop_view->wait_until_running(1000ms)) {
        std::cerr << "failed to start hosted event loop\n";
        return 1;
    }
    std::cout << "[1] event loop hosted as blocking I/O worker\n";

    // -----------------------------------------------------------------------
    // 模式 2：pool 任务 -> strand 延续（post 级派发的盲区纪律）。
    //
    // 盲区事实：loop_view->post(...) 的任务不经过 executor 提交路径，
    // 不进入 admission、TaskStatistics 与 failure 事件。纪律：
    //   a) 跨界状态用 shared_ptr 拥有，移交后原线程不再访问；
    //   b) 延续内部异常必须自捕获（盲区里没人替你记录失败）。
    // -----------------------------------------------------------------------
    auto value = std::make_shared<std::atomic<int>>(0);
    auto stage = std::make_shared<std::atomic<int>>(0);

    auto producer = executor.submit([value, stage, loop_view]() noexcept {
        value->store(41, std::memory_order_release);
        stage->store(1, std::memory_order_release);

        // 延续派发回串行循环：此后 value/stage 只在 strand 上访问。
        loop_view->post([value, stage]() noexcept {
            value->fetch_add(1, std::memory_order_acq_rel);
            stage->store(2, std::memory_order_release);
        });
    });
    producer.wait();

    while (stage->load(std::memory_order_acquire) != 2) {
        std::this_thread::yield();
    }
    std::cout << "[2] pool -> strand continuation, value=" << value->load()
              << " (post dispatches are outside executor statistics)\n";

    // -----------------------------------------------------------------------
    // 模式 3：PhaseGate 收尾——串行批次推进相位，pool 侧等待。
    // -----------------------------------------------------------------------
    comm::PhaseGate gate("event_loop_batch");
    constexpr uint64_t kBatchSize = 4;

    auto batch_future = executor.submit([&gate, kBatchSize]() {
        // pool 线程等待串行侧完成整批工作；wait_for 支持超时，不会无限等。
        const comm::CommResult result = gate.wait_for(kBatchSize, 5s);
        return result.ok;
    });

    for (uint64_t i = 0; i < kBatchSize; ++i) {
        loop_view->post([i, &gate]() noexcept {
            // 每个串行任务完成一批次中的一步，然后推进相位。
            gate.advance_to(i + 1);
        });
    }

    const bool batch_ok = batch_future.get();
    std::cout << "[3] phase gate batch finalized, ok=" << batch_ok << "\n";

    // -----------------------------------------------------------------------
    // 收尾：先停止托管循环（run 排空返回），再 shutdown 线程池。
    // -----------------------------------------------------------------------
    worker.request_stop();
    loop_view->wakeup();
    worker.stop();
    std::cout << "[4] hosted loop stopped: "
              << (worker.status().is_running ? "still running" : "stopped")
              << "\n";

    executor.shutdown();
    std::cout << "event loop interop example completed\n";
    return batch_ok ? 0 : 1;
}

边界上的取消与定时

需要协作取消的任务必须走 executor 提交 API（submit_cancellable + StopToken）。executor 的取消不会伸进 asio 的内部等待。
不依赖 strand 所有权的自建 sleep_until 循环可以迁移到 submit_delayed_with_handle() / submit_periodic_with_handle()（见取消与定时与 docs/MIGRATION.md）。
回调与销毁必须发生在同一 strand 上的 timer，在外部上下文定时器绑定通过评审（T2）之前继续由应用侧管理。SerialExecutionContext 只提供 FIFO 任务纳管，不改变 facade 定时器的执行或销毁上下文。此类 timer 暂不要迁移到 facade 句柄。

与外部事件循环互操作 ​

模式 1：把循环托管为 Blocking I/O worker ​

模式 2：strand 延续合法但不可见 ​

模式 3：用 PhaseGate 收尾批次 ​

边界上的取消与定时 ​

延伸阅读 ​

与外部事件循环互操作

模式 1：把循环托管为 Blocking I/O worker

模式 2：strand 延续合法但不可见

模式 3：用 PhaseGate 收尾批次

边界上的取消与定时

延伸阅读