比物理线程都好用的 C++20 的协程，你会用吗？

关注

发布于: 2021 年 05 月 17 日

摘要：事件驱动（event driven）是一种常见的代码模型，其通常会有一个主循环（mainloop）不断的从队列中接收事件，然后分发给相应的函数/模块处理。常见使用事件驱动模型的软件包括图形用户界面（GUI），嵌入式设备软件，网络服务端等。

本文分享自华为云社区《C++20的协程在事件驱动代码中的应用》，原文作者：飞得乐。

嵌入式事件驱动代码的难题

事件驱动（event driven）是一种常见的代码模型，其通常会有一个主循环（mainloop）不断的从队列中接收事件，然后分发给相应的函数/模块处理。常见使用事件驱动模型的软件包括图形用户界面（GUI），嵌入式设备软件，网络服务端等。

本文以一个高度简化的嵌入式处理模块做为事件驱动代码的例子：假设该模块需要处理用户命令、外部消息、告警等各种事件，并在主循环中进行分发，那么示例代码如下：

#include <iostream>#include <vector>
enum class EventType {    COMMAND,    MESSAGE,    ALARM};
// 仅用于模拟接收的事件序列std::vector<EventType> g_events{EventType::MESSAGE, EventType::COMMAND, EventType::MESSAGE};
void ProcessCmd(){    std::cout << "Processing Command" << std::endl;}
void ProcessMsg(){    std::cout << "Processing Message" << std::endl;}
void ProcessAlm(){    std::cout << "Processing Alarm" << std::endl;}
int main() {    for (auto event : g_events) {        switch (event) {            case EventType::COMMAND:                ProcessCmd();                break;            case EventType::MESSAGE:                ProcessMsg();                break;            case EventType::ALARM:                ProcessAlm();                break;        }    }    return 0;}

复制代码

这只是一个极简的模型示例，真实的代码要远比它复杂得多，可能还会包含：从特定接口获取事件，解析不同的事件类型，使用表驱动方法进行分发……不过这些和本文关系不大，可暂时先忽略。

用顺序图表示这个模型，大体上是这样：

在实际项目中，常常碰到的一个问题是：有些事件的处理时间很长，比如某个命令可能需要批量的进行上千次硬件操作：

void ProcessCmd(){    for (int i{0}; i < 1000; ++i) {        // 操作硬件接口……    }}

复制代码

这种事件处理函数会长时间的阻塞主循环，导致其他事件一直排队等待。如果所有事件对响应速度都没有要求，那也不会造成问题。但是实际场景中经常会有些事件是需要及时响应的，比如某些告警事件出现后，需要很快的执行业务倒换，否则就会给用户造成损失。这个时候，处理时间很长的事件就会产生问题。

有人会想到额外增加一个线程专用于处理高优先级事件，实践中这确实是个常用方法。然而在嵌入式系统中，事件处理函数会读写很多公共数据结构，还会操作硬件接口，如果并发调用，极容易导致各类数据竞争和硬件操作冲突，而且这些问题常常很难定位和解决。那在多线程的基础上加锁呢？——设计哪些锁，加在哪些地方，也是非常烧脑而且容易出错的工作，如果互斥等待过多，还会影响性能，甚至出现死锁等麻烦的问题。

另一种解决方案是：把处理时间很长的任务切割成很多个小任务，并重新加入到事件队列中。这样就不会长时间的阻塞主循环。这个方案避免了并发编程产生的各种头疼问题，但是却带来另一个难题：如何把一个大流程切割成很多独立小流程？在编码时，这需要程序员解析函数流程的所有上下文信息，设计数据结构单独存储，并建立关联这些数据结构的特殊事件。这往往会带来几倍的额外代码量和工作量。

这个问题几乎在所有事件驱动型软件中都会存在，但在嵌入式软件中尤为突出。这是因为嵌入式环境下的 CPU、线程等资源受限，而实时性要求高，并发编程受限。

C++20 语言给这个问题提供了一种新的解决方案：协程。

C++20 的协程简介

关于协程（coroutine）是什么，在 wikipedia[1]等资料中有很好的介绍，本文就不赘述了。在 C++20 中，协程的关键字只是语法糖：编译器会将函数执行的上下文（包括局部变量等）打包成一个对象，并让未执行完的函数先返回给调用者。之后，调用者使用这个对象，可以让函数从原来的“断点”处继续往下执行。

使用协程，编码时就不再需要费心费力的去把函数“切割”成多个小任务，只用按照习惯的流程写函数内部代码，并在允许暂时中断执行的地方加上 co_yield 语句，编译器就可以将该函数处理为可“分段执行”。

协程用起来的感觉有点像线程切换，因为函数的栈帧（stack frame）被编译器保存成了对象，可以随时恢复出来接着往下运行。但是实际执行时，协程其实还是单线程顺序运行的，并没有物理线程切换，一切都只是编译器的“魔法”。所以用协程可以完全避免多线程切换的性能开销以及资源占用，也不用担心数据竞争等问题。

可惜的是，C++20 标准只提供了协程基础机制，并未提供真正实用的协程库（在 C++23 中可能会改善）。目前要用协程写实际业务的话，可以借助开源库，比如著名的 cppcoro[2]。然而对于本文所述的场景，cppcoro 也没有直接提供对应的工具（generator 经过适当的包装可以解决这个问题，但是不太直观），因此我自己写了一个切割任务的协程工具类用于示例。

自定义的协程工具

下面是我写的 SegmentedTask 工具类的代码。这段代码看起来相当复杂，但是它作为可重用的工具存在，没有必要让程序员都理解它的内部实现，一般只要知道它怎么用就行了。SegmentedTask 的使用很容易：它只有 3 个对外接口：Resume、IsFinished 和 GetReturnValue，其功能可根据接口名字自解释。

#include <optional>#include <coroutine>
template<typename T>class SegmentedTask {public:    struct promise_type {        SegmentedTask<T> get_return_object()         {            return SegmentedTask{Handle::from_promise(*this)};        }
        static std::suspend_never initial_suspend() noexcept { return {}; }        static std::suspend_always final_suspend() noexcept { return {}; }        std::suspend_always yield_value(std::nullopt_t) noexcept { return {}; }
        std::suspend_never return_value(T value) noexcept        {            returnValue = value;            return {};        }
        static void unhandled_exception() { throw; }
        std::optional<T> returnValue;    };     using Handle = std::coroutine_handle<promise_type>;     explicit SegmentedTask(const Handle coroutine) : coroutine{coroutine} {}     ~SegmentedTask()     {         if (coroutine) {            coroutine.destroy();         }    }     SegmentedTask(const SegmentedTask&) = delete;    SegmentedTask& operator=(const SegmentedTask&) = delete;     SegmentedTask(SegmentedTask&& other) noexcept : coroutine(other.coroutine) { other.coroutine = {}; }
    SegmentedTask& operator=(SegmentedTask&& other) noexcept    {        if (this != &other) {            if (coroutine) {                coroutine.destroy();            }            coroutine = other.coroutine;            other.coroutine = {};        }        return *this;    }
    void Resume() const { coroutine.resume(); }    bool IsFinished() const { return coroutine.promise().returnValue.has_value(); }    T GetReturnValue() const { return coroutine.promise().returnValue.value(); } private:    Handle coroutine;};

复制代码

自己编写协程的工具类不光需要深入了解 C++协程机制，而且很容易产生悬空引用等未定义行为。因此强烈建议项目组统一使用编写好的协程类。如果读者想深入学习协程工具的编写方法，可以参考 Rainer Grimm 的博客文章[3]。

接下来，我们使用 SegmentedTask 来改造前面的事件处理代码。当一个 C++函数中使用了 co_await、co_yield、co_return 中的任何一个关键字时，这个函数就变成了协程，其返回值也会变成对应的协程工具类。在示例代码中，需要内层函数提前返回时，使用的是 co_yield。但是 C++20 的 co_yield 后必须跟随一个表达式，这个表达式在示例场景下并没必要，就用了 std::nullopt 让其能编译通过。实际业务环境下，co_yield 可以返回一个数字或者对象用于表示当前任务执行的进度，方便外层查询。

协程不能使用普通 return 语句，必须使用 co_return 来返回值，而且其返回类型也不直接等同于 co_return 后面的表达式类型。

enum class EventType {    COMMAND,    MESSAGE,    ALARM};
std::vector<EventType> g_events{EventType::COMMAND, EventType::ALARM};std::optional<SegmentedTask<int>> suspended;  // 没有执行完的任务保存在这里
SegmentedTask<int> ProcessCmd(){    for (int i{0}; i < 10; ++i) {        std::cout << "Processing step " << i << std::endl;        co_yield std::nullopt;    }    co_return 0;}
void ProcessMsg(){    std::cout << "Processing Message" << std::endl;}
void ProcessAlm(){    std::cout << "Processing Alarm" << std::endl;}
int main(){    for (auto event : g_events) {        switch (event) {            case EventType::COMMAND:                suspended = ProcessCmd();                break;            case EventType::MESSAGE:                ProcessMsg();                break;            case EventType::ALARM:                ProcessAlm();                break;        }    }    while (suspended.has_value() && !suspended->IsFinished()) {        suspended->Resume();    }    if (suspended.has_value()) {        std::cout << "Final return: " << suspended->GetReturnValue() << endl;    }    return 0;}

复制代码

出于让示例简单的目的，事件队列中只放入了一个 COMMAND 和一个 ALARM，COMMAND 是可以分段执行的协程，执行完第一段后，主循环会优先执行队列中剩下的事件，最后再来继续执行 COMMAND 余下的部分。实际场景下，可根据需要灵活选择各种调度策略，比如专门用一个队列存放所有未执行完的分段任务，并在空闲时依次执行。

本文中的代码使用 gcc 10.3 版本编译运行，编译时需要同时加上-std=c++20 和-fcoroutines 两个参数才能支持协程。代码运行结果如下：

Processing step 0Processing AlarmProcessing step 1Processing step 2Processing step 3Processing step 4Processing step 5Processing step 6Processing step 7Processing step 8Processing step 9Final return: 0

复制代码

可以看到 ProcessCmd 函数（协程）的 for 循环语句并没有一次执行完，在中间插入了 ProcessAlm 的执行。如果分析运行线程还会发现，整个过程中并没有物理线程的切换，所有代码都是在同一个线程上顺序执行的。

使用了协程的顺序图变成了这样：

事件处理函数的执行时间长不再是问题，因为可以中途“插入”其他的函数运行，之后再返回断点继续向下运行。

总结

一个较普遍的认识误区是：使用多线程可以提升软件性能。但事实上，只要 CPU 没有空跑，那么当物理线程数超过了 CPU 核数，就不再会提升性能，相反还会由于线程的切换开销而降低性能。大多数开发实践中，并发编程的主要好处并非为了提升性能，而是为了编码的方便，因为现实中的场景模型很多都是并发的，容易直接对应成多线程代码。

协程可以像多线程那样方便直观的编码，但是同时又没有物理线程的开销，更没有互斥、同步等并发编程中令人头大的设计负担，在嵌入式应用等很多场景下，常常是比物理线程更好的选择。

相信随着 C++20 的逐步普及，协程将来会得到越来越广泛的使用。