如何确保Boost.Asio协程在Serial类中按顺序无交错执行？

我试图确保Serial类中的协程调用按顺序完整执行，无交错情况。尽管已使用strand，但不同调用的操作仍出现交错。

我的代码如下：

#include <fmt/core.h>
#include <boost/asio.hpp>

namespace asio = boost::asio;
using namespace std::chrono_literals;

class Serial {
   public:
    Serial(asio::any_io_executor ex)
        : ex_(ex), timer_1_{ex}, timer_2_{ex} {};
    asio::awaitable<void> Read(int i) {
        co_await asio::dispatch(bind_executor(ex_, asio::deferred));
        fmt::println("in read {}", i);
        co_await Send(i);
        timer_2_.expires_after(50ms);
        co_await timer_2_.async_wait(bind_executor(ex_, asio::deferred));
        int j = co_await Receive();
        fmt::println("out read {}", j);
        co_return;
    }

   private:
    asio::awaitable<void> Send(int i) {
        co_await asio::dispatch(bind_executor(ex_, asio::deferred));
        fmt::println("in send {}", i);
        i_ = i;
        timer_1_.expires_after(50ms);
        co_await timer_1_.async_wait(bind_executor(ex_, asio::deferred));
        fmt::println("out send {}", i);
        co_return;
    }

    asio::awaitable<int> Receive() {
        co_await asio::dispatch(bind_executor(ex_, asio::deferred));
        fmt::println("in receive {}", i_);
        timer_1_.expires_after(50ms);
        co_await timer_1_.async_wait(bind_executor(ex_, asio::deferred));
        fmt::println("out receive {}", i_);
        co_return i_;
    }

    asio::any_io_executor ex_;
    asio::steady_timer timer_1_;
    asio::steady_timer timer_2_;
    int i_ = 0;
};

asio::awaitable<void> TestRead(Serial& s, int i) {
    co_await s.Read(i);
    co_return;
}

int main() {
    asio::io_context io;

    auto strand = asio::make_strand(io);

    Serial s{strand};
    co_spawn(io, TestRead(s, 0), asio::detached);
    co_spawn(io, TestRead(s, 1), asio::detached);

    io.run();
}

Program returned: 0
in read 0
in send 0
in read 1
in send 1
out send 1
in receive 1
out receive 1
out read 1

我希望操作按顺序完整执行，无交错：

in read 0
in send 0
out send 0
in receive 0
out receive 0
out read 0
in read 1
in send 1
out send 1
in receive 1
out receive 1
out read 1

• 已使用strand尝试序列化执行
• 尝试用post替代dispatch
• 尝试为每个操作显式绑定executor

如何修改代码，确保一次完整的Read()调用（包括其内部的Send()和Receive()调用）执行完成后，再开始处理下一次Read()调用？

我希望尽可能保留协程使用，而非重写为回调或完全不同的架构。我知道mutex可以解决，但有没有符合Asio风格的惯用方案？

要实现整个Read()调用的串行执行，Asio中有更贴合其设计风格的方案——通过维护串行任务链来确保前一个任务完全结束后再启动下一个，无需使用mutex。

修改方案 #

在Serial类中添加一个用于追踪待处理任务的协程成员，每次调用Read()时，将当前任务追加到任务链末尾，确保只有前一个任务完成后，当前任务才会执行：

#include <fmt/core.h>
#include <boost/asio.hpp>

namespace asio = boost::asio;
using namespace std::chrono_literals;

class Serial {
   public:
    Serial(asio::any_io_executor ex)
        : ex_(ex), timer_1_{ex}, timer_2_{ex}, pending_(asio::make_ready_awaitable<void>()) {};

    asio::awaitable<void> Read(int i) {
        // 保存当前待处理任务，将新任务追加到链中
        auto prev_task = std::move(pending_);
        pending_ = [this, i, prev = std::move(prev_task)]() -> asio::awaitable<void> {
            // 等待前一个任务完成
            co_await prev;
            
            // 执行当前Read的完整逻辑
            co_await asio::dispatch(bind_executor(ex_, asio::deferred));
            fmt::println("in read {}", i);
            co_await Send(i);
            timer_2_.expires_after(50ms);
            co_await timer_2_.async_wait(bind_executor(ex_, asio::deferred));
            int j = co_await Receive();
            fmt::println("out read {}", j);
        }();
        
        // 等待当前任务完成
        co_await pending_;
    }

   private:
    asio::awaitable<void> Send(int i) {
        co_await asio::dispatch(bind_executor(ex_, asio::deferred));
        fmt::println("in send {}", i);
        i_ = i;
        timer_1_.expires_after(50ms);
        co_await timer_1_.async_wait(bind_executor(ex_, asio::deferred));
        fmt::println("out send {}", i);
        co_return;
    }

    asio::awaitable<int> Receive() {
        co_await asio::dispatch(bind_executor(ex_, asio::deferred));
        fmt::println("in receive {}", i_);
        timer_1_.expires_after(50ms);
        co_await timer_1_.async_wait(bind_executor(ex_, asio::deferred));
        fmt::println("out receive {}", i_);
        co_return i_;
    }

    asio::any_io_executor ex_;
    asio::steady_timer timer_1_;
    asio::steady_timer timer_2_;
    int i_ = 0;
    asio::awaitable<void> pending_; // 追踪串行任务链
};

asio::awaitable<void> TestRead(Serial& s, int i) {
    co_await s.Read(i);
    co_return;
}

int main() {
    asio::io_context io;

    auto strand = asio::make_strand(io);

    Serial s{strand};
    co_spawn(io, TestRead(s, 0), asio::detached);
    co_spawn(io, TestRead(s, 1), asio::detached);

    io.run();
}

原理说明 #

• pending_成员保存了当前正在等待执行的任务链，初始状态为已就绪的空任务。
• 每次调用Read()时，会创建一个新的协程任务，该任务首先等待前一个任务prev完成，再执行自身的Read逻辑。
• 新任务会替换pending_，后续的Read()调用会自动等待这个新任务完成，从而形成串行执行的任务链。

这种方式完全基于Asio的协程机制，符合其异步编程的惯用风格，同时保留了协程的简洁性，无需切换到回调架构。

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