Lab5b 实验解析

tinyCoroLab5b 实验解析

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

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

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

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

📖lab5b 任务参考实现

🧑‍💻Task #1 - 实现 condition_variable

在实现 tinyCoro 的 condition_variable 前我们需要回顾一下 C++ std::condition_variable，比如下面一个经典的生产消费模型：

void producer() {
    std::this_thread::sleep_for(std::chrono::seconds(2));  // 模拟数据生成过程
    std::lock_guard<std::mutex> lock(mtx);
    ready = true;  // 数据已经生成
    cv.notify_one();  // 通知消费者线程
}
void consumer() {
    std::unique_lock<std::mutex> lock(mtx);
    cv.wait(lock, []{return ready;});  // 等待数据生成
    std::cout << "Consumer thread is processing data...\n";
    // 模拟数据处理过程
    std::this_thread::sleep_for(std::chrono::seconds(2));
    std::cout << "Data processed.\n";
}

我相信上述代码对读者一定不难理解，我们可以从中提炼出几个要点：

1. 消费者在调用cv.wait时一定需要借助 lock

2. 一旦消费者因调用cv.wait陷入阻塞，那么必须通过cv.notify才能将其唤醒，否则永久阻塞

对于要点 1 很好理解，条件变量也是确保线程间的正确同步和避免竞态条件的工具，其核心仍然是借助 mutex 对临界区加锁，但条件变量为用户提供了一种可以自定义的加锁顺序，相当于让 mutex 的使用更加灵活。

对于要点 2，只有cv.notify可以唤醒陷入阻塞的消费者，哪怕此时 lock 已处于解锁状态，那么这意味着什么呢？

读者通过 lab4d 实现的 mutex 应该知道，mutex 对协程的阻塞和唤醒是通过其本身维护的 suspend awaiter 链表结构实现的，协程阻塞在 mutex 上，那么其本身就会被挂在到 mutex 的 suspend awaiter 链表中，唤醒协程即从该链表取出 awaiter 并恢复对应的协程即可。因此，要点 2 提供给了我们一个重要信息：条件变量本身也要维护 suspend awaiter 链表，但其与 mutex 的 suspend awaiter 链表是相互独立的。

基于此，我们便可以大致梳理出 tinyCoro 中条件变量的工作流程。我们用cv_suspend_list和mutex_suspend_list分别代指 condition_variable 和 mutex 的 suspend awaiter 链表，当协程调用cv.wait时，参数 mutex 指定了该协程将要获取的锁，如果不存在条件谓词或者条件谓词返回 false，那么该协程将被挂载到cv_suspend_list上，如果存在条件谓词且返回 true，那么协程会尝试对指定的 mutex 加锁，成功则协程恢复执行，否则自身被挂载到mutex_suspend_list上，此时协程的唤醒依赖于mutex.unlock，这部分对 mutex 的操作我们只需要调用我们在已经为 mutex 实现的方法即可。

而cv.notify就十分简单了，如果是notify_one，就从cv_suspend_list中取出一个 awaiter 尝试对 mutex 加锁就可以了，而notify_all则取出全部的 awaiter 分别执行 mutex 加锁即可。

按照惯例我会把上述过程归纳成一张图便于读者理解，如下所示：

cv_show

上图仅展示了复杂一些的状态转换逻辑，注意图中用了两个 mutex 意在告诉读者 condition_variable 的wait是可以指定不同的锁，协程在唤醒后也只会对相应的锁尝试加锁逻辑。

对 condition_variable 的整体工作流程有了大致概念后，我们便可以着手设计了，下面给出 tinyCoro 对于 condition_variable 的定义，代码较长为了方便讲解我会补充详细的注释：

// 条件谓词的类型，入参为空且返回值为 bool
using cond_type = std::function<bool()>;

class condition_variable; // 前向声明
using cond_var = condition_variable; // 简化书写

class condition_variable final
{
public:
    // 注意这里是对 mutex_awaiter 进行了复用，减少冗余代码
    struct cv_awaiter : public mutex::mutex_awaiter
    {
        friend condition_variable;

        // 不支持条件谓词的构造函数
        cv_awaiter(context& ctx, mutex& mtx, cond_var& cv) noexcept
            : mutex_awaiter(ctx, mtx), // 委托构造
              m_cv(cv),
              m_suspend_state(false)
        {
        }

        // 支持条件谓词的构造函数
        cv_awaiter(context& ctx, mutex& mtx, cond_var& cv, cond_type& cond) noexcept
            : mutex_awaiter(ctx, mtx), // 委托构造
              m_cv(cv),
              m_cond(cond),
              m_suspend_state(false)
        {
        }

        // 由于逻辑有变化，需重写原有方法
        auto await_suspend(std::coroutine_handle<> handle) noexcept -> bool;

        // 由于逻辑有变化，需重写原有方法
        auto await_resume() noexcept -> void;

    protected:
        // 实现了 cv.wait 的核心逻辑，这个命名可能有些误导，后续会修复
        auto register_lock() noexcept -> bool;

        // 将自身挂载到 cv 的 suspend awaiter 链表中
        auto register_cv() noexcept -> void;

        // 唤醒 awaiter
        auto wake_up() noexcept -> void;

        // 尝试恢复 awaiter 运行，该函数会被 wake_up 调用
        // 注意这是个虚函数
        auto resume() noexcept -> void override;

        cond_type m_cond; // 条件谓词
        cond_var& m_cv; // 该 awaiter 所依赖的 condition_variable
        bool      m_suspend_state; // 用来确保 context 的引用计数正常，稍后讲解具体作用
        // 这里还需要很多额外的成员变量比如 awaiter 依赖的 context 和 mutex，
        // 但通过继承 mutex_awaiter 就不需要再额外定义
    };

public:
    condition_variable() noexcept = default;
    ~condition_variable() noexcept;

    // 禁止拷贝和移动语义
    CORO_NO_COPY_MOVE(condition_variable);

    // 条件变量的等待方法，该函数为协程函数
    auto wait(mutex& mtx) noexcept -> cv_awaiter;

    // 条件变量的等待方法且支持条件谓词，该函数为协程函数
    auto wait(mutex& mtx, cond_type&& cond) noexcept -> cv_awaiter;

    // 条件变量的等待方法且支持条件谓词，该函数为协程函数
    auto wait(mutex& mtx, cond_type& cond) noexcept -> cv_awaiter;

    // 通知单个 suspepnd_awaiter 恢复运行，该函数为普通函数
    auto notify_one() noexcept -> void;

    // 通知所有 suspepnd_awaiter 恢复运行，该函数为普通函数
    auto notify_all() noexcept -> void;

private:
    detail::spinlock m_lock; // 优化后的自旋锁，也可改用 std::mutex
    alignas(config::kCacheLineSize) cv_awaiter* m_head{nullptr}; // suspend awaiter 链表头指针
    alignas(config::kCacheLineSize) cv_awaiter* m_tail{nullptr}; // suspend awaiter 链表尾指针
};

下面我们针对具体实现讲解，首先是协程调用cv.wait时 condition_variable 的内部函数处理：

auto condition_variable::cv_awaiter::await_suspend(std::coroutine_handle<> handle) noexcept -> bool
{
    m_await_coro = handle; // 保存当前协程句柄便于之后恢复
    return register_lock();
}
// register_lock 返回 true 即表示协程将自身注册到了 suspend awaiter 链表中，即陷入 suspend 状态
// 否则协程继续运行
auto condition_variable::cv_awaiter::register_lock() noexcept -> bool
{
    if (m_cond && m_cond())
    {
        // 如果条件谓词存在且条件谓词返回 true，那么协程继续运行，
        // 这一点与 C++ 中的 std::condition_variable 有所不同，
        // std::condition_variable 在调用 wait 时不会考虑条件谓词，只有被 notify 时才会
        return false;
    }

    // condition_variable 内部必须维持一个变量来表示 awaiter 是否已增加引用计数，
    // 因为协程因 condition_variable 陷入阻塞态且被 notify 时如果条件谓词返回 false，
    // 协程将再次陷入 suspend 状态，如果这里只是调用 m_ctx.register_wait()，那么
    // context 的引用计数便会多次增加，而协程恢复后只会对引用计数减 1，这导致 context
    // 将因引用计数永远无法终止
    // 这个问题的解决方案很多，读者只需要明白其核心问题即可
    m_ctx.register_wait(!m_suspend_state);
    m_suspend_state = true;

    // 此时协程确定要陷入 suspend 状态，那么需要将自身挂载到 condition_variable
    // 的 suspend awaiter 链表中
    register_cv();

    // std::condition_variable 在调用 wait 时其必须持有锁，tinyCoro 同理，
    // 因为协程即将要陷入 suspend 状态，所以释放锁
    m_mtx.unlock();
    return true;
}
auto condition_variable::cv_awaiter::register_cv() noexcept -> void
{
    m_next = nullptr;

    m_cv.m_lock.lock(); // 用自旋锁保护临界区

    // 将当前 awaiter 挂载到链表尾部
    if (m_cv.m_tail == nullptr)
    {
        m_cv.m_head = m_cv.m_tail = this;
    }
    else
    {
        m_cv.m_tail->m_next = this;
        m_cv.m_tail         = this;
    }
    m_cv.m_lock.unlock();
}

然后是协程调用cv.notify_one以及唤醒协程的逻辑：

auto condition_variable::notify_one() noexcept -> void
{
    m_lock.lock(); // 自旋锁保护临界区
    auto cur = m_head;
    if (cur != nullptr)
    {
        // 将链表头作为待唤醒的 awaiter 并更新链表头
        m_head = reinterpret_cast<cv_awaiter*>(m_head->m_next);
        if (m_head == nullptr)
        {
            m_tail = nullptr;
        }
        m_lock.unlock();
        cur->wake_up(); // 尝试唤醒协程
    }
    else
    {
        m_lock.unlock();
    }
}

// 尝试唤醒并不意味着协程会立刻恢复运行
auto condition_variable::cv_awaiter::wake_up() noexcept -> void
{
    // 这里调用 mutex_awaiter::register_lock 是想让 awaiter 尝试获取锁，
    // 如果该函数返回 false，即协程获取到了锁，那么立刻尝试恢复该协程运行
    // 如果该函数返回 true，即协程因对已上锁的锁加锁陷入了 suspend 状态，
    // 那么什么也不做，因为该协程一定会通过 mutex.unlock 被唤醒，即将
    // 唤醒协程的义务转移给了 mutex
    if (!mutex_awaiter::register_lock())
    {
        resume();
    }
    // 读者应该确保理解上述提到的协程的状态转移过程
}
auto condition_variable::cv_awaiter::resume() noexcept -> void
{
    if (m_cond && !m_cond())
    {
        // 存在条件谓词且条件谓词返回 false，那么当前 awaiter 需要再次将自身注册到
        // condition_variable 的 suspend awaiter 链表中并陷入 suspend 状态
        m_ctx.register_wait(!m_suspend_state);
        m_suspend_state = true;

        register_cv();
        m_mtx.unlock();
        return;
    }
    // 不存在条件谓词或条件谓词返回 true，那么协程正式恢复运行，
    // 这里直接复用 mutex_awaiter 的方法即可
    mutex_awaiter::resume(); 
}

上述代码不长，但读者应该注意几个要点，我在注释中多次用了尝试二字，尝试即表示预期的事情不一定发生，例如 awiater 被notify后还要根据条件谓词决定是否会被恢复。然后是代码多次复用了 mutex 的方法，比如在wake_up函数中，如果mutex_awaiter::register_lock返回 true 那么唤醒协程的职责就从 condition_variable 转移到了 mutex，cv_awaiter 将自身挂载到了 mutex 的 suspend awaiter 中，那么 mutex 在唤醒协程时如何区分 mutex_awaiter 和 cv_awaiter 呢？答案是不需要区分，只需要调用awaiter->resume即可，因为是指针调用且resume是虚函数， 所以 mutex 可以正确恢复 cv_awaiter 关联协程的运行。

理解了notify_one的逻辑，那么notify_all的实现就很简单了，代码如下：

auto condition_variable::notify_all() noexcept -> void
{
    cv_awaiter* nxt{nullptr};

    m_lock.lock(); // 自旋锁保护临界区
    // 取出全部链表
    auto cur_head = m_head;
    m_head = m_tail = nullptr;
    m_lock.unlock();

    // 按 FIFO 顺序唤醒
    while (cur_head != nullptr)
    {
        nxt = reinterpret_cast<cv_awaiter*>(cur_head->m_next);
        cur_head->wake_up();
        cur_head = nxt;
    }
}

💡notify_all 按 FIFO 顺序唤醒的过程中有新的 awaiter 挂载进来怎么处理？ 参考 std::condition_variable 的行为，该 awaiter 需要由后续的 notify 处理

最后是协程恢复后的await_resume，仅需要将引用计数减 1 即可：

auto condition_variable::cv_awaiter::await_resume() noexcept -> void
{
    m_ctx.unregister_wait(m_suspend_state);
}

综上，tinyCoro 实现的 condition_variable 比较巧的复用了 mutex 的代码逻辑，用较少的代码量达到了预期的功能，其总体代码设计还是比较优雅的。

实验总结

相信在完成前几个实验后实验者应该已经对 C++ 协程 awaiter 的使用游刃有余了，但通过完成 lab5b 我们有学习了利用 awaiter 实现更为复杂的协程状态转换，这对于编程思维的培养是非常重要的。