Lab4a 实验解析

tinyCoroLab4a 实验解析

⚠️tinyCoroLab 的实验强烈推荐实验者独自完成而非直接翻阅实验解析，否则这与读完题直接翻看参考答案无太大区别，实验解析仅供实验者参考。

本节将会以 tinyCoroLab 的官方实现tinyCoro为例，为大家分析并完成 lab4a，请实验者预先下载 tinyCoro 的代码到本地。

git clone https://github.com/sakurs2/tinyCoro

打开include/coro/comp/event.hpp和src/comp/event.cpp并大致浏览代码结构。

📖lab4a 任务参考实现

🧑‍💻Task #1 - 实现 event

因为 event 针对模板参数是否为空具有不同的行为，因此分离出 event_base 来实现共性行为（调度逻辑），其具有唯一一个成员变量用来挂载 suspend awaiter：

std::atomic<awaiter_ptr> event_base::m_state{nullptr};

当m_state等于 event 的 this 指针时表示 event 此时已被 set。

下面给出 event_base 的定义并标明注释：

class event_base
{
public:
    struct awaiter_base
    {
        awaiter_base(context& ctx, event_base& e) noexcept : m_ctx(ctx), m_ev(e) {}

        inline auto next() noexcept -> awaiter_base* { return m_next; }

        auto await_ready() noexcept -> bool;

        auto await_suspend(std::coroutine_handle<> handle) noexcept -> bool;

        auto await_resume() noexcept -> void;

        context&                m_ctx; // 绑定的 context
        event_base&             m_ev; // 绑定的 event
        awaiter_base*           m_next{nullptr}; // 链表的 next 指针
        std::coroutine_handle<> m_await_coro{nullptr}; // 待 resume 的协程句柄
    };

    event_base(bool initial_set = false) noexcept : m_state((initial_set) ? this : nullptr) {}
    ~event_base() noexcept = default;

    event_base(const event_base&)            = delete;
    event_base(event_base&&)                 = delete;
    event_base& operator=(const event_base&) = delete;
    event_base& operator=(event_base&&)      = delete;

    // 判断 event 是否被 set，根据 m_state 是否等于 this
    inline auto is_set() const noexcept -> bool { return m_state.load(std::memory_order_acquire) == this; }

    auto set_state() noexcept -> void; // 设置 event，唤醒所有 suspend awaiter

    auto resume_all_awaiter(awaiter_ptr waiter) noexcept -> void; // 唤醒所有 suspend awaiter

    auto register_awaiter(awaiter_base* waiter) noexcept -> bool; // 挂载 suspend awaiter

private:
    std::atomic<awaiter_ptr> m_state{nullptr};
};

首先针对 awaiter_base 结构体，其核心成员变量主要是用于帮助该 awaiter 在陷入 suspend 状态后正确恢复，另外 next 指针用于指向下一个 suspend awaiter，因为 event 挂载 awaiter 的形式是链表，event_base 中的m_state表示链表头。

下面分析 awaiter_base 成员函数的具体实现并附加注释：

auto event_base::awaiter_base::await_ready() noexcept -> bool
{
    m_ctx.register_wait(); // 增加引用计数
    return m_ev.is_set(); // 如果 event 是否被设置决定协程是否陷入 suspend
}

auto event_base::awaiter_base::await_suspend(std::coroutine_handle<> handle) noexcept -> bool
{
    m_await_coro = handle; // 保存协程句柄

    // 将协程 awaiter 挂载到 event 中，这个过程期间有可能 event 被 set，因此返回 bool 表示是否
    // 挂载成功，挂载失败那么协程恢复运行
    return m_ev.register_awaiter(this); 
}

auto event_base::awaiter_base::await_resume() noexcept -> void
{
    m_ctx.unregister_wait(); // 降低引用计数
}

对于 event_base 的set_state实现如下：

auto event_base::set_state() noexcept -> void
{
    auto flag = m_state.exchange(this, std::memory_order_acq_rel);
    if (flag != this) // event 之前未被 set
    {
        auto waiter = static_cast<awaiter_base*>(flag);
        resume_all_awaiter(waiter); // 取得列表头并恢复所有 suspend awaiter
    }
}

对于 event_base 的resume_all_awaiter实现如下：

auto event_base::resume_all_awaiter(awaiter_ptr waiter) noexcept -> void
{
    // 以循环的形式遍历链表
    while (waiter != nullptr)
    {
        auto cur = static_cast<awaiter_base*>(waiter);

        // 将该 awaiter 绑定的协程投放到其绑定的 context 的任务队列里
        cur->m_ctx.submit_task(cur->m_await_coro);
        waiter = cur->m_next;
    }
}

💡为何 event 挂载 suspend awaiter 采用链表形式？ 挂载的 awaiter 数量不确定，如果 event 使用容器的话可能产生动态内存分配开销，不如直接在 awaiter 中多添加一个 next 指针，这样 event 只需要存储一个链表头就可以了。

对于 event_base 的register_awaiter实现如下：

auto event_base::register_awaiter(awaiter_base* waiter) noexcept -> bool
{
    const auto  set_state = this;
    awaiter_ptr old_value = nullptr;
    // 利用 cas 操作确保挂载操作的原子性
    do
    {
        old_value = m_state.load(std::memory_order_acquire);
        if (old_value == this)
        {
            waiter->m_next = nullptr;
            return false;
        }
        waiter->m_next = static_cast<awaiter_base*>(old_value);
    } while (!m_state.compare_exchange_weak(old_value, waiter, std::memory_order_acquire));

    return true;
}

上述代码是典型的原子变量 cas 的使用场景，当前 awaiter 将 next 指针指向 event_base 的 m_state 并将自身置为新的 m_state，相当于后挂载的 awaiter 会作为最新的链表头，因为整个操作非原子，所以使用compare_exchange_weak判断是否在这个过程中有其他线程修改了状态，因为是 do while 循环，所以使用compare_exchange_weak而不是compare_exchange_strong。

💡后挂载的 awaiter 会作为最新的链表头，恢复协程的时候如果按链表遍历顺序恢复那岂不 LIFO 调度方式？ 是的，如果想实现 FIFO 那么就在取出链表后对其进行反转再按链表遍历顺序恢复。

综上，event_base 实现的协程与恢复机制并不复杂，只是需要注意对原子变量m_state的正确操作。

模板参数为空的 event 只需要继承 event_base 后调用其接口就可以了。

模板参数非空的 event 可以选择继承 event_base 和 container，在存储与取出数据时调用 container 相关的接口即可，需要注意的是此时wait返回的 awaiter 需要在 awaiter_base 的基础上修改await_resume函数并将 container 存储的值返回，代码如下：

auto await_resume() noexcept -> decltype(auto)
{
    detail::event_base::awaiter_base::await_resume(); // 保留降低引用计数的代码
    return static_cast<event&>(m_ev).result();
}

实验总结

• 通过实现 event 的 awaiter 挂载机制了解了链表的一种高效应用场景

• 学会了对多线程编程下对原子变量的正确操作