Lab5c 实验解析

tinyCoroLab5c 实验解析

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

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

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

并打开include/coro/comp/channel.hpp并大致浏览代码结构。

📖lab5c 任务参考实现

🧑‍💻Task #1 - 实现 channel

channel 的概念来自于 golang，对于 tinyCoro 本质上是一个多生产者多消费者模型。在借助之前实验完成的 condition_variable 和 mutex 的帮助下，channel 的构建将会轻松很多，唯一需要注意的是 channel 是可被关闭的，其关闭行为与 golang 相同，关闭后生产者不可再生产，但消费者仍旧可以消费剩余的元素。

channel 本身带有两个模板参数分别指明了元素类型和容量，容量大于 1 相当于是 buffer_channel，此时生产者可以持续向 buffer_channel 生产多个元素而不会发生阻塞。

tinyCoro 针对容量为 channel 特化了三个模板类，对于这三个模板类有一个共同的基类 channel_base，代码如下：

class channel_base
{
public:
    ~channel_base() noexcept { assert(part_closed() && "detected channel destruct with no_close state"); }

    // 关闭 channel
    auto close() noexcept -> void
    {
        // 将 channel 状态置为部分关闭
        std::atomic_ref<uint8_t>(m_close_state).store(part_close, std::memory_order_release);
        m_producer_cv.notify_all(); // 唤醒所有生产者协程
        m_consumer_cv.notify_all(); // 唤醒所有消费者协程
    }

protected:
    // 检查 channel 是否处于完全关闭状态，原子操作
    inline auto complete_closed_atomic() noexcept -> bool
    {
        return std::atomic_ref<uint8_t>(m_close_state).load(std::memory_order_acquire) <= complete_close;
    }

    // 检查 channel 是否处于部分关闭状态，原子操作
    inline auto part_closed_atomic() noexcept -> bool
    {
        return std::atomic_ref<uint8_t>(m_close_state).load(std::memory_order_acquire) <= part_close;
    }

    // 检查 channel 是否处于部分关闭状态，非原子操作
    inline auto part_closed() noexcept -> bool { return m_close_state <= part_close; }

protected:
    // 表示 channel 已完全关闭，即调用了 close 且全部元素均被消费
    inline static constexpr uint8_t complete_close = 0;
    // 表示 channel 已部分关闭，即调用了 close 但仍有元素未消费
    inline static constexpr uint8_t part_close     = 1; 
    // 表示 channel 未关闭
    inline static constexpr uint8_t no_close       = 2; 

    mutex              m_mtx;         // 用于条件变量的锁 
    condition_variable m_producer_cv; // 生产者条件变量
    condition_variable m_consumer_cv; // 消费者条件变量

    // 保存当前 channel 状态
    alignas(std::atomic_ref<uint8_t>::required_alignment) uint8_t m_close_state{no_close};
};

对于上述代码添加了详细的注释，至于为何要设计三种关闭状态，稍后读者就能明白。

虽然 channel 的模板类有 3 个，但核心思路一致，无非是容量的不同，我们针对第一个模板类即容量超过 1 的情况讲解：

template<concepts::conventional_type T, size_t capacity = 0>
class channel : public detail::channel_base
{
    using data_type = std::optional<T>;
    // 由于生产和消费行为均涉及协程的状态转换，因此 send 和 recv 函数均为协程函数，
    // 返回值直接复用 tinyCoro 的 task 即可
public:
    template<typename value_type>
        requires(std::is_constructible_v<T, value_type &&>)
    auto send(value_type&& value) noexcept -> task<bool>
    {
        // 如果此时 channel 处于关闭状态（不论是半关闭还是完全关闭），
        // 那么返回停止生产函数返回 false
        if (part_closed_atomic())
        {
            co_return false;
        }

        // 尝试加锁
        auto lock = co_await m_mtx.lock_guard();
        // 利用条件变量决定是否应该继续持有锁
        // 如果 channel 没有满或者处于关闭状态均应接触阻塞态
        co_await m_producer_cv.wait(m_mtx, [this]() -> bool { return !full() || part_closed(); });

        // 如果此时 channel 处于关闭状态（不论是半关闭还是完全关闭），
        // 那么返回停止生产函数返回 false
        // 注意在锁的保护下此处不需要使用原子操作
        if (part_closed())
        {
            co_return false;
        }

        // 将元素添加到环形数组尾部
        m_array[m_tail] = std::forward<value_type>(value);
        m_tail++;
        m_num++;
        if (m_tail >= capacity)
        {
            m_tail = 0;
        }
        // 唤醒一个消费者
        m_consumer_cv.notify_one();
        co_return true;
    }

    auto recv() noexcept -> task<data_type>
    {
        // 如果 channel 完全关闭，那么该函数应该直接返回，
        // 避免后续的无效调用产生的开销
        if (complete_closed_atomic())
        {
            co_return std::nullopt;
        }

        // 尝试获取锁
        auto lock = co_await m_mtx.lock_guard();
        // 如果 channel 没有满或者处于关闭状态均应接触阻塞态
        co_await m_consumer_cv.wait(m_mtx, [this]() -> bool { return !empty() || part_closed(); });
        if (empty() && part_closed())
        {
            // 如果此时 channel 已空且处于半关闭状态，那么将 channel 转换为完全关闭状态
            m_close_state = complete_close;
            co_return std::nullopt;
        }
        // 取出环形数组头部元素
        data_type p = std::make_optional<T>(std::move(m_array[m_head]));
        m_head++;
        m_num--;
        if (m_head >= capacity)
        {
            m_head = 0;
        }

        // 通知一个生产者协程
        m_producer_cv.notify_one();
        co_return p;
    }

private:
    // channel 是否为空
    inline auto empty() noexcept -> bool { return m_num == 0; }
    // channel 是否已满
    inline auto full() noexcept -> bool { return m_num == capacity; }

private:
    // 下列变量用于维持一个环形数组
    size_t                  m_head{0}; // 头指针
    size_t                  m_tail{0}; // 尾指针
    size_t                  m_num{0};  // 元素数量
    std::array<T, capacity> m_array;   // 存储元素的数组
};

对于上述代码读者可以看到，通过引入了半关闭和全关闭状态，可以使消费函数提前返回，而无需进行后续的无效调用。 然后消费函数返回的元素是 std::optional 类型的，这样有助于消费者感知到是否 channel 已关闭且元素全部消费完毕。

另外读者可能会发现，上述代码的生产和消费模型，每次操作均是针对一个元素，然后立刻唤醒下一个生产者或消费者，这样的做法虽然保证各个生产者和消费者均能参与，但实在是低效，不过这里作为演示够用了，读者可以自行实现更高效的 channel，后续 tinyCoro 也会为 channel 添加 batch 操作。

对于容量为 0（也可等效于容量为 1）的特化 channel，本质上就是用一个元素代替环形数组，具体实现这里不再赘述。

而对于容量为 2 的次方的情况，tinyCoro 也特化了一个 channel，读者可能不解容量为 2 的次方有何特殊之处，其实这是一个经典的环形数组优化技巧。在上述代码展示的容量非 2 的次方的 channel 实现中，需要依靠数组和三个变量才能实现环形数组，而容量为 2 的次方时，仅需要数组和两个变量就可实现环形数组且状态判断更加轻松，我们可以看下面的代码对比：

// 环形数组实现（暂不考虑线程安全）
// 容量非 2 的次方
template<typename T, size_t capacity>
class ring_buffer{
public:
  bool push(T value) {
    if(full()) {
      return false;
    }
    m_array[m_tail] = std::forward<value_type>(value);
    // 为环形数组添加元素较为繁琐
    m_tail++;
    m_num++;
    if (m_tail >= capacity)
    {
        m_tail = 0;
    }
    return true;
  }
  std::optional<T> pop() {
    if(empty()) {
      return std::nullopt;
    }
    auto data = std::make_optional<T>(std::move(m_array[m_head]));
     // 为环形数组删除元素较为繁琐
    m_head++;
    m_num--;
    if (m_head >= capacity)
    {
        m_head = 0;
    }
    return data;
  }

private:
  bool full() {return m_num == capacity;}
  bool empty() {return m_num == 0;}

private:
  size_t                  m_head{0}; // 头指针
  size_t                  m_tail{0}; // 尾指针
  size_t                  m_num{0};  // 元素数量
  std::array<T, capacity> m_array;   // 存储元素的数组
};
// 容量为 2 的次方
template<typename T, size_t capacity>
class ring_buffer{
  static constexpr size_t mask = capacity - 1;
public:
  bool push(T value) {
    if(full()) {
      return false;
    }
    // 利用掩码获取真正的尾指针
    m_array[m_tail&mask] = std::forward<value_type>(value);
    // 更改尾指针十分简单
    m_tail++;
    return true;
  }
  std::optional<T> pop() {
    if(empty()) {
      return std::nullopt;
    }
    // 利用掩码获取真正的头指针
    auto data = std::make_optional<T>(std::move(m_array[m_head&mask]));
    // 更改头指针十分简单
    m_head++;
    return data;
  }

private:
  bool full() {return m_tail-m_head>=capacity;}
  bool empty() {return m_head == m_tail;}

private:
  size_t                  m_head{0}; // 头指针
  size_t                  m_tail{0}; // 尾指针
  std::array<T, capacity> m_array;   // 存储元素的数组
};

上述代码不难理解，利用容量为 2 的次方这一特点获得一个掩码标志mask，对环形数组添加或移除元素只需要尾指针或头指针加 1 即可，获取真正的头指针和尾指针只需要与掩码mask进行与操作即可，这样的写法十分简洁，实际上 tinyCoro 所依赖的第三方库 AtomicQueue 便对容量为 2 的次方这种情况做了优化。

💡在函数内使用局部变量 channel，生产者调用 close 后函数结束 channel 自动析构，那么此时被唤醒的消费者对析构的 channel 消费不就导致 core dump 了吗？ 目前的 channel 设计存在这样的弊端，用户可以利用 wait_group 等同步组件确保生产者和消费者均退出后再析构 channel，后续 tinyCoro 也会针对 close 进行优化

实验总结

通过本节实验学会了利用现有协程同步组件封装更简单易用的 channel，同时也了解了一种针对环形数组的常用优化技巧。