Lab4d 实验解析

tinyCoroLab4d 验解析

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

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

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

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

📖lab4d 任务参考实现

🧑‍💻Task #1 - 实现 mutex

mutex 的实现难点在于每次unlock只会恢复最多一个协程的运行，对于频繁lock和unlock的场景必须保证协程调度的正确性。

为了讲解方便，我先给出 mutex 实现的函数声明形式并给出注释讲解，这样更直观。

class mutex
{
public:
    struct mutex_awaiter
    {
        mutex_awaiter(context& ctx, mutex& mtx) noexcept : m_ctx(ctx), m_mtx(mtx) {}

        constexpr auto await_ready() noexcept -> bool { return false; }

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

        auto await_resume() noexcept -> void;

        // awaiter 将自身注册到 mutex 的 awaiter 链表中，注册成功即陷入 suspend 状态
        auto register_lock() noexcept -> bool;

        // 对于恢复协程过程的封装，声明为虚函数是因为后续需要
        virtual auto resume() noexcept -> void;

        context&                m_ctx; // 绑定的 context
        mutex&                  m_mtx; // 绑定的 mutex
        mutex_awaiter*          m_next{nullptr}; // mutex 协程链表的 next 指针
        std::coroutine_handle<> m_await_coro{nullptr}; // 待 resume 的协程
    };

    // 用于 mutex.lock_guard 的返回值，其逻辑与 mutex_awaiter 相同因此继承了 mutex_awaiter，
    // 但重写了 await_resume 用于返回 lock_guard
    struct mutex_guard_awaiter : public mutex_awaiter
    {
        using guard_type = detail::lock_guard<mutex>;
        using mutex_awaiter::mutex_awaiter;

        auto await_resume() noexcept -> guard_type
        {
            mutex_awaiter::await_resume(); // 保留原有的逻辑
            return guard_type(m_mtx);
        }
    };

public:
    mutex() noexcept : m_state(nolocked), m_resume_list_head(nullptr) {}
    ~mutex() noexcept { assert(m_state.load(std::memory_order_acquire) == mutex::nolocked); }

    auto try_lock() noexcept -> bool;

    auto lock() noexcept -> mutex_awaiter;

    auto unlock() noexcept -> void;

    auto lock_guard() noexcept -> mutex_guard_awaiter;

private:
    friend mutex_awaiter;

    // 常量 1 表示当前锁处于未加锁状态，常亮 0 表示锁处于加锁但没有协程在等待获取锁
    // 注意 0 和 1 不是随便设置的，原因稍后讲解
    inline static awaiter_ptr nolocked          = reinterpret_cast<awaiter_ptr>(1);
    inline static awaiter_ptr locked_no_waiting = 0; // nullptr
    std::atomic<awaiter_ptr>  m_state; // 与 event 类似，同样也是挂载 suspend awaiter 的链表头
    awaiter_ptr               m_resume_list_head; // 用于存储待恢复的 suspend awaiter，作用稍后讲解
};

然后我们对重点函数的实现进行分析，首先是mutex_awaiter中关于协程调度的函数：

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

auto mutex::mutex_awaiter::await_suspend(std::coroutine_handle<> handle) noexcept -> bool
{
    m_await_coro = handle; // 存储协程句柄
    m_ctx.register_wait(); // 增加引用计数 

    // 挂载自身到 mutex 的 awaiter 链表，挂载成功与否会决定协程
    // 是否陷入 suspend 状态
    return register_lock(); 
}

然后是mutex_awaiter挂载 awaiter 的逻辑：

auto mutex::mutex_awaiter::register_lock() noexcept -> bool
{
    // 下列代码仍然使用了循环+cas 的方法保证操作的原子性
    while (true)
    {
        auto state = m_mtx.m_state.load(std::memory_order_acquire);
        m_next     = nullptr;
        if (state == mutex::nolocked) // 如果未加锁
        {
            if (m_mtx.m_state.compare_exchange_weak(
                    state, mutex::locked_no_waiting, std::memory_order_acq_rel, std::memory_order_relaxed))
            {
                // 如果成功将锁置于 locked_no_waiting 状态，那么当前协程恢复运行
                return false;
            }
        }
        else
        {
            // 此时 mutex 已经被别的协程加锁，
            // state 要么等于指向 awaiter 的指针，要么是 locked_no_waiting
            // 如果 state 是 locked_no_waiting，那么经过下述的操作当前的
            // awaiter 的 next 指针则被置为 locked_no_waiting，而 locked_no_waiting
            // 值为 0，即 nullptr，这样正好方便了链表遍历

            // 将自身的 next 指向 state 并将自身置为 state，
            // 这个操作 lab4 前几个实验也都有用到
            m_next = reinterpret_cast<mutex_awaiter*>(state);
            if (m_mtx.m_state.compare_exchange_weak(
                    state, reinterpret_cast<awaiter_ptr>(this), std::memory_order_acq_rel, std::memory_order_relaxed))
            {
                return true;
            }
        }
    }
}

然后是mutex_awaiter用于恢复协程执行的逻辑：

auto mutex::mutex_awaiter::resume() noexcept -> void
{
    m_ctx.submit_task(m_await_coro); // 将协程句柄到其绑定的 context 的任务队列中
}

对于 mutex 的 lock 部分的接口其实现如下：

auto mutex::try_lock() noexcept -> bool
{
    auto target = nolocked;
    return m_state.compare_exchange_strong(target, locked_no_waiting, std::memory_order_acq_rel, memory_order_relaxed);
}
auto mutex::lock() noexcept -> mutex_awaiter
{
    return mutex_awaiter(local_context(), *this);
}
auto mutex::lock_guard() noexcept -> mutex_guard_awaiter
{
    return mutex_guard_awaiter(local_context(), *this);
}

mutex 的unlock实现相较复杂：

auto mutex::unlock() noexcept -> void
{
    assert(m_state.load(std::memory_order_acquire) != nolocked && "unlock the mutex with unlock state");

    auto to_resume = reinterpret_cast<mutex_awaiter*>(m_resume_list_head);
    // 如果 m_resume_list_head 没空的话那么直接从 m_resume_list_head 取待恢复的协程就好了
    // 这样可以减少对原子变量的操作
    if (to_resume == nullptr)
    {
        // 此时 m_resume_list_head 为空，那么必须对 m_state 操作了
        auto target = locked_no_waiting;
        if (m_state.compare_exchange_strong(target, nolocked, std::memory_order_acq_rel, std::memory_order_relaxed))
        {
            // 此时没有任何 suspend awaiter，那么直接返回
            return;
        }
        // 此时一定存在 suspend awaiter

        // 将 m_state 上挂载的 suspend awaiter 全部取出来
        auto head = m_state.exchange(locked_no_waiting, std::memory_order_acq_rel);
        assert(head != nolocked && head != locked_no_waiting);

        // 下面是一个翻转链表操作，将去除的 suspend awaiter 链表翻转后
        // 挂载到 m_resume_list_head 上
        auto awaiter = reinterpret_cast<mutex_awaiter*>(head);
        do
        {
            auto temp       = awaiter->m_next;
            awaiter->m_next = to_resume;
            to_resume       = awaiter;
            awaiter         = temp;
        } while (awaiter != nullptr);
    }

    assert(to_resume != nullptr && "unexpected to_resume value: nullptr");
    // 取出 m_resume_list_head 的链表头
    m_resume_list_head = to_resume->m_next;
    to_resume->resume();
}

注意上述unlock中对挂载 awaiter 的链表进行翻转操作，因此 mutex 的unlock操作遵从 FIFO 调度逻辑。

综合来看，对 mutex 的实现主要考察实验者对原子变量的使用。

实验总结

• 通过实现 mutex 加深了对原子变量的理解，合理的使用原子变量来构建无锁并发数据结构可以大大提高多线程编程的效率