Lab2a 实验解析

tinyCoroLab2a 实验解析

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

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

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

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

📖lab2a 任务参考实现

🧑‍💻Task #1 - 完善 engine 初始化以及异步 I/O 支持

对于异步 IO 支持部分，tinyCoro 为 engine 添加了多个成员变量用来记录当前待提交的 IO 数以及正在运行且未完成的 IO 数：

class engine{
  size_t m_num_io_wait_submit{0};
  size_t m_num_io_running{0};
};

在 lab2a 的任务书中解释了协程发起 IO 的过程是线程安全的，因此 IO 状态不使用原子变量记录，对于add_io_submit实现如下：

auto add_io_submit() noexcept -> void
{
    // 表示有一个新增的 IO 等待提交
    m_num_io_wait_submit += 1;
}

对于 engine 的初始化以及析构化，核心逻辑主要是对成员变量的处理及析构：

auto engine::init() noexcept -> void
{
    linfo.egn            = this;
    m_num_io_wait_submit = 0;
    m_num_io_running     = 0;
    m_upxy.init(config::kEntryLength);
}
auto engine::deinit() noexcept -> void
{
    m_upxy.deinit();
    m_num_io_wait_submit = 0;
    m_num_io_running     = 0;
    mpmc_queue<coroutine_handle<>> task_queue;
    m_task_queue.swap(task_queue);
}

get_free_urs用于获取空闲的 sqe，直接使用 uring_proxy 的方法即可：

auto engine::get_free_urs() noexcept -> ursptr { return m_upxy.get_free_sqe(); }

🧑‍💻Task #2 - 完善 engine 任务执行能力

有了成员变量来记录 IO 执行状态，那么empty_io的实现就很简单了：

auto engine::empty_io() noexcept -> bool
{
    return m_num_io_wait_submit == 0 && m_num_io_running == 0;
}

engine 使用第三方库 AtomicQueue 存储协程句柄，因此关于非 IO 部分的任务处理只需要调用 AtomicQueue 提供的方法即可：

auto engine::submit_task(coroutine_handle<> handle) noexcept -> void
{
    assert(handle != nullptr && "engine get nullptr task handle");
    m_task_queue.push(handle);
    wake_up();
}
auto engine::ready() noexcept -> bool { return !m_task_queue.was_empty(); }
auto engine::num_task_schedule() noexcept -> size_t { return m_task_queue.was_size(); }
auto engine::schedule() noexcept -> coroutine_handle<>
{
    auto coro = m_task_queue.pop();
    assert(bool(coro));
    return coro;
}

然后是 IO 的提交和执行部分，engine 额外添加了一个成员函数用来提交 IO：

auto engine::do_io_submit() noexcept -> void
{
    // 利用条件判断来决定是否需要调用 io_uring 的提交操作
    if (m_num_io_wait_submit > 0)
    {
        [[CORO_MAYBE_UNUSED]] auto _ = m_upxy.submit();
        m_num_io_running += m_num_io_wait_submit; 
        m_num_io_wait_submit = 0; // io_uring 会一次提交所有 IO
    }
}

do_io_submit的逻辑很简单，如果有待提交的任务则调用 uring_proxy 的submit即可，然后对m_num_io_running和m_num_io_wait_submit进行更改。

对于核心的 poll_submit，主要分为提交 IO、等待 IO 执行、取出 IO 和处理已完成 IO 四个步骤，代码如下：

auto engine::poll_submit() noexcept -> void
{
    do_io_submit(); // 提交 IO

    auto cnt = m_upxy.wait_eventfd(); // 等待 IO 执行
    if (!wake_by_cqe(cnt))
    {
        return;
    }

    // 取出 IO
    auto num = m_upxy.peek_batch_cqe(m_urc.data(), m_num_io_running.load(std::memory_order_acquire));

    if (num != 0)
    {
        // 处理 IO
        for (int i = 0; i < num; i++)
        {
            handle_cqe_entry(m_urc[i]);
        }
        m_upxy.cq_advance(num);
        m_num_io_running.fetch_sub(num, std::memory_order_acq_rel);
    }
}

上述代码存在一个问题，实验者应该不难看出，那就是工作线程在无任何任务的情况下利用阻塞在 eventfd 读操作上来让出执行权防止 CPU 空转。 如果 io_uring 产生了 cqe 那么会向 eventfd 写值，此时工作线程被唤醒，继续执行任务，这没什么问题。

但是在长期运行工作模式下工作线程阻塞在了读 eventfd，也没有 IO 任务完成，但 scheduler 向其派发了一个任务，那么问题来了，如何通知工作线程？在 eventfd 的帮助下这个问题的解决方案很简单，engine 添加了一个函数wake_up专门用来唤醒工作线程：

auto engine::wake_up(uint64_t val) noexcept -> void
{
    m_upxy.write_eventfd(val);
}

本质上是向 eventfd 写一个值，那么工作线程读取 eventfd 便可以立刻返回，那什么情况下需要调用 wake_up 呢？第一种当然是利用submit_task提交任务时，第二种是准备发起 IO 时，不过因为发起 IO 不再跨线程，所以只有第一种情况需要调用 wake_up。

tinyCoro 中的 io_uring 并没有开启 polling 模式，每次调用 submit 都会涉及到系统调用，因此为了优化此部分，仅在 m_num_io_wait_submit 大于 0 时才会提交。

综上工作线程被唤醒有三种情况并且可以同时发生：

⚠️ 在之前的版本中协程发起 IO 会主动用 wake_up 唤醒工作线程，现在发起 IO 的过程不再跨线程，因此协程发起 IO 无需再调用 wake_up，即 case2 已废弃

• case1： 有新任务提交

• case2： 有新 IO 提交

• case3： 有 IO 已完成

为了区分这三种情况，engine 定义了如下标志位：

static constexpr uint64_t task_mask = (0xFFFFF00000000000);
static constexpr uint64_t io_mask   = (0x00000FFFFF000000);
static constexpr uint64_t cqe_mask  = (0x0000000000FFFFFF);

static constexpr uint64_t task_flag = (((uint64_t)1) << 44);
static constexpr uint64_t io_flag   = (((uint64_t)1) << 24);

即将 64 位整数的高 20 位分给 case1，中间 20 位分给 case2，其余 24 位分给 case3，然后使用下述宏对从 eventfd 读取的值判断属于哪种 case：

#define wake_by_task(val) (((val) & engine::task_mask) > 0)
#define wake_by_io(val)   (((val) & engine::io_mask) > 0)
#define wake_by_cqe(val)  (((val) & engine::cqe_mask) > 0)

因此poll_submit包含下列判断语句来避免后续对 uring_proxy 的无效调用产生的开销：

if (!wake_by_cqe(cnt))
{
    return;
}

💡如果频繁提交 task 那么会通过 wake_up 频繁向 eventfd 写值，岂不是开销很高？ 读者不坊自行测试一下 eventfd 的性能，我个人的测试结果是一次 eventfd 读写不超过 1us，所以 eventfd 真的非常轻量高效，这点开销目前绝不会造成性能瓶颈

综上，engine 的实现难点主要在于如何正确且高效的的避免工作线程一直陷入阻塞状态。

实验总结

• 本节通过完成 engine 学习了如何利用 eventfd 搭建一种高效的事件通知机制