积累沉淀

待山花烂漫，化茧成蝶

Coroutine-协程

Posted on Edited on

前言：协程（Coroutine）是一种计算机程序组件，它与子例程（即通常所说的函数或过程）类似，但其执行方式更为灵活。不同于传统的线程和进程，协程允许在特定的地方暂停执行，并能在之后从暂停的地方恢复执行。这种特性使得协程在处理异步编程、并发操作以及I/O密集型任务时特别有用。

什么是协程

  • 核心：是能够暂停和恢复的函数。
  • 能被多次调用的函数，每次只运行协程代码的一部分
  • 每次调用之间本地变量和参数都会被保存，生命周期与调用者无关
  • 与线程的区别：线程会互相争抢，（同一个线程上运行的）协程只会在主动让出控制权时切换，不同线程上得协程也会相会争抢

为什么需要协程

  1. 异步操作
  2. 生成器

普通函数和协程得区别

协程有什么用

  • 使得异步IO代码的实现变得简单，容易维护
  • 例子：boost不同版本的异步IO echo server

C++20协程机制－关键对象

  1. promise_type
  • 对协程行为的配置 (比如是否在协程入口处挂起)
  • 用于调用者与协程之间的数据传递
  • 可以对协程对象增加一些自定义数据结构(用于实现调度器)
  1. std:coroutine_handle<promise_type>
  • 外部操作协程的接口 (本质上是一个指针)
  1. coroutine_interface
  • coroutine的返回对象
  • 对coroutine_handle的封装
  • 在coroutine_handle的基础上再实现一些用户自定义的接口，比如获取返回值等
  1. awaitor
  • 控制协程在挂起时的行为
  • 决定是否挂起
  • 决把控制权交给谁
  • 配置在控制器返回时做什么

协程定义方法

  • 定义一个函数，只要里面出现co_await,co_yield,co_return中的任意一个，就是定义了一个协程
  • 协程的返回值必须是一个coroutine_interface对象

C++20协程机制-关键字

  1. co_await
    co_await
  2. co_yield expr
  • 往promise中写值
  • 等价于co_await promise.yield_value(expr)
  1. co_return expr
  • 结束coroutine
  • 如果expr为空，或者为void,并且promise定义了return_void(),则调用 return_void()
  • 如果expr不为空，且promise定义了return_value(value)，则调用 return_value(expr)
  • return_void和return_value只能定义其中一个

用C++协程实现异步操作

用C++协程实现异步操作：理解 co_await 和 Awaitable.
在C++中，只要函数体内出现了 co_await, co_yield 或 co_return 这三个关键字之一，这个函数就自动变成了协程。
我们先聚焦在co_await上，弄清楚它是怎么工作的，以及如何让它真正跑起来。

  1. co_await是什么?
    co_await的作用是让协程暂停一下,等待某个操作（比如网络请求或文件读取）完成后，再继续执行。co_await就是这个等一等的动作，暂停协程，干别的事，等条件满足再回来。

但问题来了:如果你直接对一个自定义类型用co_await，比如co_await IntReader{}，编译器会一脸懵逼。它不知道这个类型啥时候算“完成”，也不知道结果在哪儿。为了让
co_await能用，我们需要让自定义类型遵守一个规则，这个规则叫 Awaitable.

  1. Awaitable:
    Awaitable 就像一份“协程使用说明书”，告诉编译器怎么处理暂停和恢复。它要求你的类型实现三个关键函数：
  • await_ready（）：告诉协程“现在能不能直接执行？”
  • await_suspend（）：如果要暂停，接下来该干啥?
  • await_resume（）：恢复时，返回什么结果？
    这三个函数一起合作，让co_await知道如何暂停、等待和继续。下面我们一步步拆解。
  1. Awaitable的三个函数详解
    await_ready（）：操作完成了吗？
  • 返回类型：bool
  • 作用：在执行co_await时，协程先问问这个函数：“操作是不是已经好了？”如果返回true，协程就不用暂停，直接往下跑；如果返回false，就得暂停等着。

为什么要有这个函数？
异步操作的时间不确定。假设你在co_await之前已经偷偷启动了一个任务(比如线程或网络请求)，到co_await时可能已经完成了。如果完成了，就没必要暂停，直接用结果多好！await_ready()就是用来检查这种情况的。

await_suspend（）：暂停时做什么？

  • 参数：std::coroutine_handle<>(协程的“遥控器")
  • 返回类型：通常是void，也可以是bool
  • 作用：如果await_ready（）返回false，协程要暂停，这时会调用 await_suspend()。
    这个函数拿到一个协程句柄（std::coroutine_handle<>），可以用它在未来某个时候“唤醒”协程。

什么是协程句柄？
它就像协程的身份证，指向当前暂停的协程实例。调用handle.resume() 就能让协程从暂停的地方继续跑。
返回值的妙用：

  • 返回void：正常暂停，没啥说的。
  • 返回bool：如果返回false，协程不会暂停(相当于给了第二次取消暂停的机会); 返回true(或没返回值时默认)，就真的暂停了。注意这里和await_ready（）的逻辑是反的。

await_resume（）"：恢复后给我什么？

  • 返回类型：可以是void，也可以是具体类型
  • 作用：当协程恢复执行（或者压根没暂停）时，这个函数被调用。它的返回值就是 co_await表达式的结果。

co_return 与 promise_type

  1. 协程的返回类型要求
    C++对协程的返回类型只有一个硬性规定：它必须包含一个名为promise_type的内嵌类型。
  2. 当你调用一个协程函数时：
  • 编译器会在堆上分配空间，保存协程的状态。
  • 同时创建一个promise_type对象，嵌在返回类型里。
  • 通过promise_type定义的函数，你可以控制协程的行为，或者与调用者交换数据。
  1. 协程执行时序
    coroutine
  2. task应用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class IntReader
{
public:
    bool await_ready() { return false;}
    void await_suspend(std::coroutine_handle<> handle)
    {
        std::thread thr([this, handle] {
            std::this_thread::sleep_for(1000ms);
            this->value_ = 1;
            handle.resume();
        });
        thr.detach();
    }
    int await_resume() const
    {
        return value_;
    }
private:
    int value_{0};
};

class task;
class task
{
public:
    struct promise_type
    {
        promise_type() : value_(std::make_shared<int>()) {}
        task get_return_object() { return  task{value_};  }
        void return_value(int value)
        {
            *value_ = value;
        }
        std::suspend_never initial_suspend() { return {}; }
        std::suspend_never final_suspend() noexcept { return {}; }
        void unhandled_exception() {}
    private:
        std::shared_ptr<int> value_;
    };

    explicit task(std::shared_ptr<int> value) : value_(std::move(value)) {}
    [[nodiscard("?")]]
    int get_value() const { return *value_; }

private:
    std::shared_ptr<int> value_;
};

task print_int()
{
    IntReader reader1;
    int total = co_await reader1;

    IntReader reader2;
    total += co_await reader2;

    IntReader reader3;
    total += co_await reader3;

    co_return total ;
}

int main()
{
    auto t = print_int();
    std::string str;
    while (std::cin >> str)
    {
        std::cout << t.get_value() << std::endl;
    }
    return 0;
}

co_yield

  1. 用co_yield打造生成器
    在C++的协程（coroutine）,co_yield能让你的函数变成一个“生成器”，每次生成一个值，暂停一下，等调用者需要时再继续生成下一个值。
  2. co_yield是什么?
  • 生成一个值：把某个值“扔”给调用者。
  • 暂停执行：生成完值后，协程停下来，等调用者喊“继续”。
  1. generator应用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
struct generator
{
    struct promise_type
    {
        // 一开始不暂停
        std::suspend_never initial_suspend() { return {}; }
        // 结束时暂停
        std::suspend_always final_suspend() noexcept { return {}; }
        void unhandled_exception() {}
        // 无返回值
        void return_void() {}
        generator get_return_object()
        {
            return generator( std::coroutine_handle<promise_type>::from_promise(*this) );
        }
        std::suspend_always yield_value(int value)
        {
            current_value = value;
            return {};
        }
        int current_value{0};
    };

    explicit generator(std::coroutine_handle<promise_type> h) : handle(h) {}
    // 清理协程
    ~generator() { if (handle) handle.destroy(); }

    // 获取下一个值
    int next()
    {
        // 恢复协程执行
        handle.resume();
        // 返回生成值
        return handle.promise().current_value;
    }
    // 检查是否结束
    bool done()
    {
        return handle.done();
    }
private:
    std::coroutine_handle<promise_type> handle;
};

generator fib(int n)
{
    int a = 0, b = 1;
    for (int i = 0; i < n; ++i)
    {
        co_yield a;
        int next = a + b;
        a = b;
        b = next;
    }
}

int main()
{
    auto g = fib(10);
    while (!g.done())
    {
        std::cout << g.next() << std::endl;
        std::this_thread::sleep_for(1000ms);
    }
    return 0;
}

Some Slight Improvements

  1. 使用 std::optional 来做容器，自动调用正确的构造/析构函数，支持移动语义。(可以进一步改用 std::variant)。
  2. 定义了 Generator 的几个构造和赋值函数，增强了安全性(safety)。
  3. 在 unhandled_exception() 内存放了 std::exception_ptr。
  4. 用例改为稍微复杂一些的字符串生成，里面有一些性能问题，读者有兴趣的可以找找茬。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
#include <iostream>
#include <optional>
#include <exception>
#include <string>
#include <random>
#include <coroutine>


template<typename T>
class generator
{
public:
    struct promise_type
    {
        generator get_return_object() noexcept
        {
            return generator{ std::coroutine_handle<promise_type>::from_promise(*this) };
        }
        void return_void() const noexcept {}
        std::suspend_always initial_suspend() const noexcept { return {}; }
        std::suspend_always final_suspend() const noexcept { return {}; }

        std::suspend_always yield_value(const T &value) noexcept
        {
            value_ = value;
            return {};
        }
        std::suspend_always yield_value(T &&value) noexcept
        {
            value_.emplace(std::move(value));
            return {};
        }
        void unhandled_exception() { exception_ = std::current_exception(); }
        T &result() noexcept { return *value_; }
        const T &result() const noexcept { return *value_; }
    private:
        std::optional<T> value_;
        std::exception_ptr exception_;
    };

    generator(std::coroutine_handle<promise_type> handle) : handle_(handle) {}
    generator(const generator &) = delete;
    generator &operator=(const generator &) = delete;
    generator &operator=(generator &&other) noexcept
    {
        if (this != std::addressof(other))
        {
            handle_ = other.handle_;
            other.handle_ = nullptr;
        }
        return *this;
    }
    ~generator() noexcept
    {
        if (handle_)
            handle_.destroy();
    }
    void next() { handle_.resume(); }
    T &result() noexcept { return handle_.promise().result(); }
    const T &result() const noexcept { return handle_.promise().result(); }

private:
    std::coroutine_handle<promise_type> handle_;
};

generator<std::string> lottery(size_t size, unsigned int mod)
{
    std::mt19937 rgn(std::random_device{}());
    std::string winning_nums;
    co_yield winning_nums += std::to_string( rgn() % mod );
    for (size_t i = 1; i < size; ++i)
    {
        winning_nums += " ";
        co_yield winning_nums += std::to_string(rgn() % mod);
    }

    co_return;
}

generator<int> fibonacci()
{
    int a = 0, b = 1;
    while (true)
    {
        co_yield a;
        int c = a + b;
        a = b;
        b = c;
    }
    co_return;
}

int main()
{
    auto g = lottery(10, 256);
    for (int i = 0; i < 10; ++i)
    {
        g.next();
        std::cout << g.result() << std::endl;
    }
    auto fib = fibonacci();
    for (int i = 0; i < 10; ++i)
    {
        fib.next();
        std::cout << fib.result() << " ";
    }
    std::cout << std::endl;
    return 0;
}
Buy me a coffee please.