OStack程序员社区-中国程序员成长平台 › 门户 › 编程› C++›C++教程

C++20协程解糖-动手实现协程2-实现co_await和co_return

原作者: [db:作者] 来自: [db:来源] 收藏邀请

在开始之前，我们先修复上一篇文章中的一个bug，SharedState::add_finish_callback中post_all_callbacks应当提前判断settled，否则会在未设置结果的情况下添加callback，callback也会被立即post

template<class T>
class SharedState : public SharedStateBase {
    // ...
    // private
    void add_finish_callback(std::function<void(T&)> callback) {
        finish_callbacks.push_back(std::move(callback));
        if (settled) {
            post_all_callbacks();
        }
    }
};

今天我们要实现的东西包括

给schedular加上timer支持
给Future和Promise补充必要设施以支持C++20协程

如果你看到这行文字，说明这篇文章被无耻的盗用了（或者你正在选中文字），请前往 cnblogs.com/pointer-smq 支持原作者，谢谢

开始动手

首先是Schedular的timer支持，我们这里使用一个简单的优先队列用来管理所有timer，并在poll函数中处理完当帧pending_state后，视情况sleep到最近的timer到期，并处理所有到期的timer

class Schedular {
    // ...
    // public
    using timer_callback = std::function<void()>;
    using timer_item = std::tuple<bool, float, chrono::time_point<chrono::steady_clock, chrono::duration<float>>, timer_callback>;
    using timer = std::chrono::steady_clock;
    struct timer_item_cmp {
        bool operator()(const timer_item& a, const timer_item& b) const {
            return std::get<2>(a) > std::get<2>(b);
        }
    };

    // ...
    // public
    void poll() {
        size_t sz = pending_states.size();
        for (size_t i = 0; i != sz; i++) {
            auto state = std::move(pending_states[i]);
            state->invoke_all_callback();
        }
        pending_states.erase(pending_states.begin(), pending_states.begin() + sz);
        if (timer_queue.empty()) {
            return;
        }
        if (pending_states.empty()) { //如果pending_states为空，则可以sleep较长的时间，等待第一个将要完成的timer
            std::this_thread::sleep_until(std::get<2>(timer_queue.front()));
            auto now = timer::now();
            do {
                deal_one_timer();
            } while (!timer_queue.empty() && std::get<2>(timer_queue.front()) <= now);
        } else { //否则只能处理当帧到期的timer，不能sleep，要及时返回给caller，让caller及时下一次poll处理剩下的pending_states
            auto now = timer::now();
            while (!timer_queue.empty() && std::get<2>(timer_queue.front()) <= now) {
                deal_one_timer();
            }
        }
    }

    void add_timer(bool repeat, float delay, timer_callback callback) {
        auto cur_time = chrono::time_point_cast<chrono::duration<float>>(timer::now());
        auto timeout = cur_time + chrono::duration<float>(delay);
        timer_queue.emplace_back(repeat, delay, timeout, callback);
        std::push_heap(timer_queue.begin(), timer_queue.end(), timer_item_cmp{});
    }

    // ...
    // private
    void deal_one_timer() {
        std::pop_heap(timer_queue.begin(), timer_queue.end(), timer_item_cmp{});
        auto item = std::move(timer_queue.back());
        timer_queue.pop_back();
        std::get<3>(item)();
        if (std::get<0>(item)) {
            add_timer(true, std::get<1>(item), std::move(std::get<3>(item)));
        }
    }

    std::deque<timer_item> timer_queue;
};

这样之后，基于当前调度器的delay函数就可以写出来了

class Schedular {
    // ...
    // public
    Future<float> delay(float second) {
        auto promise = Promise<float>(*this);
        add_timer(false, second, [=]() mutable {
            promise.set_result(second);
        });
        return promise.get_future();
    }
    // ...
};

因为之前我们设计的Future和Promise并不支持void，于是这里简单用Future<float>代替，返回的是等待的秒数。

需要注意的是，这个delay函数虽然返回Future，但并不是协程，协程的判断标准是当且仅当函数中使用了co_await/co_yield/co_return，和返回类型无关。

这个函数同样展示了将回调式API封装为Future的做法，就是把Promise.set_result作为回调传入给API，并返回Promise.get_future，使用者在Future这边等待就好了。

有了这些东西之后，我们可以先把本次的测试代码写出来了

如果你看到这行文字，说明这篇文章被无耻的盗用了（或者你正在选中文字），请前往 cnblogs.com/pointer-smq 支持原作者，谢谢

Future<float> func(Schedular& schedular) {
    std::cout << "start sleep\n";
    auto r = co_await schedular.delay(1.2);
    co_return r;
}

Future<int> func2(Schedular& schedular) {
    auto r = co_await func(schedular);
    std::cout << "slept for " << r << "s\n";
    co_return 42;
}

这里需要注意的是C++协程在编译器实现中，会自动构造一个Promise对象，而我们的Promise并不支持默认构造，必须传入一个Schedular参数。好在C++会替我们自动将协程参数作为作为构造函数参数来构造Promise，因此要在协程参数中指定Schedular，相当于指定Schedular构造了Promise。为一个协程显式指定调度器，是一个很合理的设计，python也是类似的设计。C#将协程调度器隐藏进了Task，因为它有一个全局的默认调度器。如果我们的实现中提供一个全局构造的Schedular，让Promise自动去找他调度，那这里的协程也可以没有参数。

如果你看到这行文字，说明这篇文章被无耻的盗用了（或者你正在选中文字），请前往 cnblogs.com/pointer-smq 支持原作者，谢谢

为了让Future支持协程，代码中还需要补充一系列的内容，列举在下面

template<class T>
class Future {
    // ...
    // public
    // 协程接口
    using promise_type = Promise<T>;

    bool await_ready() { return _state->settled; }

    void await_suspend(exp::coroutine_handle<> handle) {
        add_finish_callback([=](T&) mutable { handle.resume(); });
    }

    T await_resume() { return _state->value; }
    // 协程接口
    // ...
};

promise_type用来指定本future对应的promise，结果的输入端
await_ready检查future是否已经完成
await_suspend用来通知future，协程为了等待它完成，已经暂停，future需要在自己完成的时候，主动恢复协程
await_resume用来通知future，协程已经恢复执行，需要从future中取出结果，用作co_await表达式的结果。这里我们直接返回拷贝，实现中较为合理的是把future持有的对象移动出去，但这样的话被await的future就不能再单独获取结果了。

为了让Promise支持协程，需要补充的内容在下面

// ...
// 在最开头
// 如果你的编译器已经不需要std::experimental了，那就去掉这行，后面使用std而不是exp
namespace exp = std::experimental;

template<class T>
class Promise {
    // ...
    // public
    // 协程接口
    Future<T> get_return_object();

    exp::suspend_never initial_suspend() { return {}; }

    exp::suspend_never final_suspend() noexcept { return {}; }

    void return_value(T v) { set_result(v); }

    void unhandled_exception() { std::terminate(); }
    // 协程接口
    // ...
};

// ...
// 在Future定义后面
template<class T>
Future<T> Promise<T>::get_return_object() {
    return get_future();
}

initial_suspend用来表明协程是否在调用时暂停，异步任务一般返回suepend_never，调用时立即启动
final_suspend用来表明协程是否在co_return后暂停（延迟销毁），我们是使用shared_state的异步任务，因此可以不暂停协程，直接自动销毁协程，让shared_state留在空中靠引用计数清零销毁
return_value用于co_return将结果传入
unhandled_exception用于协程中出现了未处理异常的情况，这里面可以通过std::current_exception来获取当前异常，我们的简化版不可能出现异常，出了就直接terminate
get_return_object就是get_future。大家不要忘记一个协程是先构造promise，后从promise获取future的

有了这些东西之后，编译就不应该再出现错误了，我的编译选项是 clang++-9 test.cpp -stdlib=libc++ -std=c++2a

运行？还差最后一点

为了方便，我们效法python，给Schedular补一个run_until_compete的方法

如果你看到这行文字，说明这篇文章被无耻的盗用了（或者你正在选中文字），请前往 cnblogs.com/pointer-smq 支持原作者，谢谢

class Schedular {
    // ...
    // public
    template<class F, class... Args>
    auto run_until_complete(F&& fn, Args&&... args) -> typename std::invoke_result_t<F&&, Args&&...>::result_type {
        auto future = std::forward<F>(fn)(std::forward<Args>(args)...);
        while (!future.await_ready()) {
            poll();
        }
        return future.await_resume();
    }
};

然后main

int main() {
    Schedular schedular;

    auto r = schedular.run_until_complete(func2, schedular);

    std::cout << "run complete with " << r << "\n";
}

运行结果就有了

start sleep
slept for 1.2s
run complete with 42

怎么样，是不是很简单呢，赶紧自己写一个吧！

如果你看到这行文字，说明这篇文章被无耻的盗用了（或者你正在选中文字），请前往 cnblogs.com/pointer-smq 支持原作者，谢谢

附录 - 全部代码

#include <vector>
#include <deque>
#include <memory>
#include <iostream>
#include <functional>
#include <chrono>
#include <thread>
#include <algorithm>
#include <experimental/coroutine>

namespace exp = std::experimental;

template<class T>
class Future;

template<class T>
class Promise;

class Schedular;

class SharedStateBase : public std::enable_shared_from_this<SharedStateBase> {
    friend class Schedular;
public:
    virtual ~SharedStateBase() = default;
private:
    virtual void invoke_all_callback() = 0;
};

template<class T>
class SharedState : public SharedStateBase {
    friend class Future<T>;
    friend class Promise<T>;
public:
    SharedState(Schedular& schedular)
        : schedular(&schedular)
    {}
    SharedState(const SharedState&) = delete;
    SharedState(SharedState&&) = delete;
    SharedState& operator=(const SharedState&) = delete;
    SharedState& operator=(SharedState&&) = delete;

private:
    template<class U>
    void set(U&& v) {
        if (settled) {
            return;
        }
        settled = true;
        value = std::forward<U>(v);
        post_all_callbacks();
    }

    T& get() { return value; }

    void add_finish_callback(std::function<void(T&)> callback) {
        finish_callbacks.push_back(std::move(callback));
        if (settled) {
            post_all_callbacks();
        }
    }

    void post_all_callbacks();

    virtual void invoke_all_callback() override {
        callback_posted = false;
        size_t sz = finish_callbacks.size();
        for (size_t i = 0; i != sz; i++) {
            auto v = std::move(finish_callbacks[i]);
            v(value);
        }
        finish_callbacks.erase(finish_callbacks.begin(), finish_callbacks.begin()+sz);
    }

    bool settled = false;
    bool callback_posted = false;
    Schedular* schedular = nullptr;
    T value;
    std::vector<std::function<void(T&)>> finish_callbacks;
};

template<class T>
class Promise {
public:
    Promise(Schedular& schedular)
        : _schedular(&schedular)
        , _state(std::make_shared<SharedState<T>>(*_schedular))
    {}

    Future<T> get_future();

    // 协程接口
    Future<T> get_return_object();
    exp::suspend_never initial_suspend() { return {}; }
    exp::suspend_never final_suspend() noexcept { return {}; }
    void return_value(T v) { set_result(v); }
    void unhandled_exception() { std::terminate(); }
    // 协程接口

    template<class U>
    void set_result(U&& value) {
        if (_state->settled) {
            throw std::invalid_argument("already set result");
        }
        _state->set(std::forward<U>(value));
    }
private:
    Schedular* _schedular;
    std::shared_ptr<SharedState<T>> _state;
};

template<class T>
class Future {
public:
    using result_type = T;
    using promise_type = Promise<T>;
    friend class Promise<T>;
private:
    Future(std::shared_ptr<SharedState<T>> state)
        : _state(std::move(state))
    {
    }
public:
    // 协程接口
    bool await_ready() { return _state->settled; }
    void await_suspend(exp::coroutine_handle<> handle) {
        add_finish_callback([=](T&) mutable { handle.resume(); });
    }
    T await_resume() { return _state->value; }
    // 协程接口

    void add_finish_callback(std::function<void(T&)> callback) {
        _state->add_finish_callback(std::move(callback));
    }
private:
    std::shared_ptr<SharedState<T>> _state;
};

template<class T>
Future<T> Promise<T>::get_future() {
    return Future<T>(_state);
}

template<class T>
Future<T> Promise<T>::get_return_object() {
    return get_future();
}

namespace chrono = std::chrono;

class Schedular {
    template<class T>
    friend class SharedState;
public:
    using timer_callback = std::function<void()>;
    using timer_item = std::tuple<bool, float, chrono::time_point<chrono::steady_clock, chrono::duration<float>>, timer_callback>;
    using timer = std::chrono::steady_clock;
    struct timer_item_cmp {
        bool operator()(const timer_item& a, const timer_item& b) const {
            return std::get<2>(a) > std::get<2>(b);
        }
    };

    Schedular() = default;
    Schedular(Schedular&&) = delete;
    Schedular(const Schedular&) = delete;
    Schedular& operator=(Schedular&&) = delete;
    Schedular& operator=(const Schedular&) = delete;

    void poll() {
        size_t sz = pending_states.size();
        for (size_t i = 0; i != sz; i++) {
            auto state = std::move(pending_states[i]);
            state->invoke_all_callback();
        }
        pending_states.erase(pending_states.begin(), pending_states.begin() + sz);
        if (timer_queue.empty()) {
            return;
        }
        if (pending_states.empty()) { //如果pending_states为空，则可以sleep较长的时间，等待第一个将要完成的timer
            std::this_thread::sleep_until(std::get<2>(timer_queue.front()));
            auto now = timer::now();
            do {
                deal_one_timer();
            } while (!timer_queue.empty() && std::get<2>(timer_queue.front()) <= now);
        } else { //否则只能处理当帧到期的timer，不能sleep，要及时返回给caller，让caller及时下一次poll处理剩下的pending_states
            auto now = timer::now();
            while (!timer_queue.empty() && std::get<2>(timer_queue.front()) <= now) {
                deal_one_timer();
            }
        }
    }

    template<class F, class... Args>
    auto run_until_complete(F&& fn, Args&&... args) -> typename std::invoke_result_t<F&&, Args&&...>::result_type {
        auto future = std::forward<F>(fn)(std::forward<Args>(args)...);
        while (!future.await_ready()) {
            poll();
        }
        return future.await_resume();
    }

    void add_timer(bool repeat, float delay, timer_callback callback) {
        auto cur_time = chrono::time_point_cast<chrono::duration<float>>(timer::now());
        auto timeout = cur_time + chrono::duration<float>(delay);
        timer_queue.emplace_back(repeat, delay, timeout, callback);
        std::push_heap(timer_queue.begin(), timer_queue.end(), timer_item_cmp{});
    }

    Future<float> delay(float second) {
        auto promise = Promise<float>(*this);
        add_timer(false, second, [=]() mutable {
            promise.set_result(second);
        });
        return promise.get_future();
    }
private:
    void deal_one_timer() {
        std::pop_heap(timer_queue.begin(), timer_queue.end(), timer_item_cmp{});
        auto item = std::move(timer_queue.back());
        timer_queue.pop_back();
        std::get<3>(item)();
        if (std::get<0>(item)) {
            add_timer(true, std::get<1>(item), std::move(std::get<3>(item)));
        }
    }

    void post_call_state(std::shared_ptr<SharedStateBase> state) {
        pending_states.push_back(std::move(state));
    }
    std::vector<std::shared_ptr<SharedStateBase>> pending_states;
    std::deque<timer_item> timer_queue;
};

template<class T>
void SharedState<T>::post_all_callbacks() {
    if (callback_posted) {
        return;
    }
    callback_posted = true;
    schedular->post_call_state(shared_from_this());
}

Future<float> func(Schedular& schedular) {
    std::cout << "start sleep\n";
    auto r = co_await schedular.delay(1.2);
    co_return r;
}

Future<int> func2(Schedular& schedular) {
    auto r = co_await func(schedular);
    std::cout << "slept for " << r << "s\n";
    co_return 42;
}

int main() {
    Schedular schedular;

    auto r = schedular.run_until_complete(func2, schedular);

    std::cout << "run complete with " << r << "\n";
}

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

C#缓存操作类发布时间：2022-07-13

C++中关于vector的iterator的操作发布时间：2022-07-13

剪的笔顺,诠释剪的笔画,认识剪的部首

1 六六分期app的软件客服如何联系？(六六分期

六六分期app的软件客服如何联系？不知道吗？加qq群【895510560】即可！标题：六六分期

阅读：19234|2023-10-27

2 可心卡盟:win10系统火狐flash插件崩溃怎么

今天小编告诉大家如何处理win10系统火狐flash插件总是崩溃的问题，可能很多用户都不知

阅读：10000|2022-11-06

3 亲亲特价:怎么删除回收站图标

今天小编告诉大家如何对win10系统删除桌面回收站图标进行设置，可能很多用户都不知道

阅读：8331|2022-11-06

4 济南大学虚拟社区:鲁大师节能降温的具体办

今天小编告诉大家如何对win10系统电脑设置节能降温的设置方法，想必大家都遇到过需要

阅读：8701|2022-11-06

5 xlueops.exe:无线网络安装向导

我们在使用xp系统的过程中,经常需要对xp系统无线网络安装向导设置进行设置，可能很多

阅读：8646|2022-11-06

6 女斗合众国:win7系统cf与主机连接不稳定怎

今天小编告诉大家如何处理win7系统玩cf老是与主机连接不稳定的问题，可能很多用户都不

阅读：9668|2022-11-06

7 0xc000022-[cf烟雾头]cf怎么调烟雾头

电脑对日常生活的重要性小编就不多说了，可是一旦碰到win7系统设置cf烟雾头的问题，很

阅读：8632|2022-11-06

8 qizideyouhuo:应用程序无法正常启动0xc0000

我们在日常使用电脑的时候，有的小伙伴们可能在打开应用的时候会遇见提示应用程序无法

阅读：8005|2022-11-06

9 ipz-185:win7系统vcf文件怎么打开

今天小编告诉大家如何对win7系统打开vcf文件进行设置，可能很多用户都不知道怎么对win

阅读：8666|2022-11-06

10 傻哥蹦迪:win10系统s4怎么打开usb调试

今天小编告诉大家如何对win10系统s4开启USB调试模式进行设置，可能很多用户都不知道怎

阅读：7540|2022-11-06

客服电话

电子邮件

C++20协程解糖-动手实现协程2-实现co_await和co_return

开始动手

运行？还差最后一点

附录 - 全部代码

请发表评论

全部评论

上一篇：

下一篇：

bradtraversy/iweather: Ionic 3 mobile we

joaomh/curso-de-matlab

断牙刷新位置时间（断牙属性及刷新位置介绍

bitwiseshiftleft/sjcl: Stanford Javascri

rugk/mastodon-simplified-federation: Sim

剪的笔顺,诠释剪的笔画,认识剪的部首

六六分期app的软件客服如何联系？(六六分期

florent37/ViewAnimator: A fluent Android

florent37/Shrine-MaterialDesign2: implem

CVE-2020-36276

SimpleSoftwareIO/simple-sms: Send and re

关于我们

产品与服务

解决方案

139-2527-9053