写点什么

彻底了解 C++ 异步从理论到实践

作者:C++后台开发
  • 2022 年 7 月 16 日
  • 本文字数:7822 字

    阅读完需:约 26 分钟

1. 纠结的开篇

之前设计我们游戏用的 c++框架的时候, 刚好 c++20 的 coroutine 已经发布, 又因为是专门 给 game server 用的 c++ framework, 对多线程的诉求相对有限, 或者本着少并发少奇怪的错误的原则, 除网络和 IO 和日志等少量模块外, 大部分模块主要还是工作在主线程上的, 所以当时设计的重点也就放在了 c++20 coroutine 的包装和使用上, 更多的使用 coroutine 来完善异步的支持. 但如果考虑到 framework 作为前后端公用框架的话, 原来主要针对主线程使用的包装的 coroutine 调度器就显得有些不够用, 以此作为基础, 我们开始了尝试结合比较新的 c++异步思路, 来重新思考应该如何实现一个尽量利用 c++新特性, 业务层简单易用的异步框架了.

本系列的主要内容也是围绕这条主线来铺开, 过程中我们 主要以:

  1. 自有的 framework 异步实现 - 主要落地尝试利用 c++20 的 coroutine 实现一个业务级的调度器.

  2. asio - 这个应该不用多说了, 近年来一直高频迭代, 业界广泛使用的开源第三方库, 中间的异步任务调度, 网络部分的代码实现都非常优质.

  3. libunifex - 最接近当前 sender/receiver 版 execution 提案的可实操版本, c++17/20 兼容, 但不推荐使用 c++17 的版本进行任何尝试, 原因后续文件会展开.这几个库作为基础, 逐步展开我们对 c++异步的探索, 然后再回到落地实践这条主线上, 探讨一个业务侧使用简单, 内部高效的异步库应该如何来实现并落地. 当然, 我们的侧重点主要还是 c++异步的调度和处理上, 网络相关的有部分内容可能会简单提到, 但不会进行深入的展开.   其实整个尝试的过程只能说非常不顺利了, 当然, 随着对相关实现的深入理解和细节的深挖, 收益也是颇多的. 闲话不多说了, 我们直接切入主题, 以对异步的思考来展开这篇总览的内容.

2. 前尘往事 - rstudio framework 实现

rstudio framework 的异步框架由两块比较独立的部分组成:

  1. 一部分是源自 asio 几年前版本的 post 和 strand 部分实现, 另外附加了一些业务侧较常用的像 Fence 等对象;

  2. 另外一部分是主线程的协程调度器实现, 这部分最早是基于 c++17 实现的一版 stackless 协程; 另外一版则是 gcc11.1 正式发布后, 直接用 c++20 重构了整个实现, 直接使用 c++20 的 coroutine 的一个版本.

2.1 asio 部分

  这一部分的内容因为后续有 asio scheduler 实现具体的分析篇章, 这个地方主要以业务侧使用进行展开了.

2.1.1 executor 概述

  • 来源于 1.6X boost 同期的 asio standalone 版本

  • 去除了各平台网络处理相关的代码

  • 仅保留了 post 和相关的功能(新版本有 executor 实现)

  • 早期 c++11 兼容, 无 coroutine 支持

  • 除网络库外, asio 非常有使用价值的一部分代码

2.1.2 一个简单的使用示例

  GJobSystem->Post([]() {        //some calculate task here        //...        GJobSystem->Post(            []() {                //task notify code here                //...            },            rstudio::JobSystemType::kLogicJob);      }, rstudio::JobSystemType::kWorkJob);
复制代码

相关的时序图:

2.1.3 当前框架使用的线程结构

预定义的枚举值:

enum class JobSystemType : int {  kLogicJob = 0,       // logic thread(main thread)  kWorkJob,            // work thread  kSlowJob,            // slow work thread(run io or other slow job)  kNetworkJob,         // add a separate thread for network  kNetworkConnectJob,  // extra connect thread for network  kLogJob,             // log thread  kNotifyExternalJob,  // use external process to report something, 1 thread only~~  kTotalJobTypes,};
复制代码

不同 Job 说明:

  • kLogicJob

  • 主线程(逻辑线程)执行任务

  • kWorkJob

  • Work Thread 线程池执行任务(多个), 一般是计算量可控的小任务

  • kSlowJob

  • IO 专用线程池, IO 相关的任务投递到本线程池

  • kNetworkJob

  • 目前 tbuspp 专用的处理线程

  • kNetworkConnectJob

  • 专用的网络连接线程, tbuspp 模式下不需要

  • kLogJob

  • 日志专用线程, 目前日志模块是自己起的线程, 可以归并到此处管理

  • kNotifyExternalJob

  • 专用的通知线程, 如 lua error 的上报, 使用该类型

【文章福利】另外小编还整理了一些 C++后台开发面试题,教学视频,后端学习路线图免费分享,需要的可以自行添加:学习交流群点击加入~ 群文件共享

小编强力推荐 C++后台开发免费学习地址:C/C++Linux服务器开发高级架构师/C++后台开发架构师​

2.1.4 Timer 任务相关

相关接口:

//NoIgnore versionuint64_t JobSystemModule::AddAlwaysRunJob(JobSystemType jobType,      threads::ThreadJobFunction&& periodJob,       unsigned long periodTimeMs);
uint64_t JobSystemModule::AddTimesRunJob(JobSystemType jobType, threads::ThreadJobFunction&& periodJob, unsigned long periodTimeMs, unsigned int runCount); uint64_t JobSystemModule::AddDelayRunJob(JobSystemType jobType, threads::ThreadJobFunction&& periodJob, unsigned long delayTimeMs); void JobSystemModule::KillTimerJob(uint64_t tid);
复制代码

本部分并未直接使用 asio 原始的 basic_waitable_timer 实现, 而是自己实现的定时任务.

2.1.5 在线程池上关联执行任务 - Strand

  • 特定的情况下, 被派发到 Work 线程池的任务存在依赖关系

  • 需要串联执行的时候, 这个时候我们需要额外的设施 JobStrand

  • 来保证任务是按先后依赖关系来串行执行的

  • 如下图中 part1, part2, part3, part4 串行执行的情况所示

示例代码:

auto strand = GJobSystem->RequestStrand(rstudio::JobSystemType::kWorkJob);starnd.Post([](){     //part1~    // ...});starnd.Post([](){     //part2~    // ...});starnd.Post([](){     //part3~     // ...});starnd.Post([](){     //part4~     // ...});starnd.Post([](){     GJobSystem->Post([](){        //return code here        // ...    }, rstudio::JobSystemType::kLogicJob); });
复制代码

2.1.6 其他辅助设施

JobFence

jobs::JobFencePtr JobSystemModule::RequestFence();
复制代码
  • 字面义, 栅栏, 起到拦截执行的作用.

  • 一般多用于模块的初始化和结束

  • 如 tbuspp 在 kNetworkJob 上的初始化和结束.

示例代码(TcpService 的初始化):

job_system_module_->Post(    [this, workTicket]() {        if (!workTicket || workTicket->IsExpired()) return;
InitInNetworkThread(); }, JobSystemType::kNetworkJob);
period_task_ptr = job_system_module_->AddAlwaysRunJob( JobSystemType::kNetworkJob, [this, workTicket]() { if (!workTicket || workTicket->IsExpired()) return;
LoopInNetworkThread(); }, 10);
fence_->FenceTo((int)JobSystemType::kNetworkJob);fence_->Wait();
复制代码

JobNotify && JobWaiter

jobs::JobWaiterPtr JobSystemModule::RequestWaiter();jobs::JobNotifyPtr JobSystemModule::RequestNotify();
复制代码
  • 批量任务管理使用

  • 等待的方式的区别

  • JobNotify: 执行完成调用额外指定的回调.

  • JobWaiter: 以 Wait 的方式在特定线程等待所有 Job 执行完成.

JobTicket

jobs::JobTicketPtr JobSystemModule::RequestTicket();
复制代码
  • 令牌对象

  • 一般用来处理跨线程的生命周期控制

  • 回调之前先通过 IsExpired()来判断对应对象是否已经释放

示例代码:

GJobSystem->Post(  [this, workTicket]() { if (!workTicket || workTicket->IsExpired()) return;
InitInNetworkThread(); }, JobSystemType::kNetworkJob);
复制代码

2.2 asio 与其他实现的对比

  正好今年的 GDC 上有一个<<One Frame In Halo Infinite>>的分享, 里面主要讲述的是对 Halo Infinite 的引擎升级, 提供新的 JobSystem 和新的动态帧的机制来支撑项目的, 我们直接以它为例子来对比一下 framework 和 Halo 的实现, 并且也借用 Halo Infinite 的例子, 来更好的了解这种 lambda post 模式的缺陷, 以及可以改进的点.   

Halo 引入新的 JobSystem 主要是为了将老的 Tetris 结构的并发模式:

向新的基于 Dependency 的图状结构迁移:

他使用的 JobSystem 的业务 Api 其实很简单, 我们直接来看一下相关的代码:

JobSystem& jobSsytem = JobSystem::Get();JobGraphHandle graphHandle = jobSystem.CreateJobGraph();
JobHandle jobA = jobSystem.AddJob( graphHandle, "JobA", [](){...} );
JobHandle jobB = jobSystem.AddJob( graphHandle, "JobB", [](){...} );
jobSystem.AddJobToJobDependency(jobA, jobB);
jobSystem.SubmitJobGraph(graphHandle);
复制代码

通过这样的机制, 就很容易形成如:

另外还有一个用于同步的 SyncPoint:

JobSystem& jobSystem = JobSystem::Get();JobGraphHandle graphHandle = jobSystem.CreateJobGraph();
SyncPointHandle syncPointX = jobSystem.CreateSyncPoint(graphHandle, "SyncPointX");
JobHandle jobA = jobSystem.AddJob(graphHandle, "JobA", [](){...});JobHandle jobB = jobSystem.AddJob(graphHandle, "JobB", [](){...});
jobSystem.AddJobToSyncPointDependency(jobA, syncPointX);jobSystem.AddSyncPointToJobDependency(syncPointX, jobB);
jobSystem.SubmitJobGraph(graphHandle);
复制代码

大致的作用如下:

这样在 workload 主动触发 SyncPoint 后, 整体执行才会继续往下推进, 这样就能方便的加入一些主动的同步点对整个 Graph 的执行做相关的控制了。

回到 asio, 我们前面也介绍了, 使用 strand 和 post(), 我们也能很方便的构造出 Graph 形的执行情况 , 而 SyncPoint 其实类型 framework 中提供的 Event, 表达上会略有差异, 但很容易看出两套实现其实是相当类同的. 这样的话, Halo 的 JobSystem 有的所有优缺点, framework 基本也同样存在了, 这里简单搬运一下:

对于复杂并发业务的表达以 lambda 内嵌为主, 虽然这种方式尽可能保证所有代码上下文是比较集中的, 对比纯粹使用 callback 的模式有所进步, 但这种自由度过高的方式本身也会存在一些问题, 纯粹靠编码者来维系并发上下文的正确性, 这种情况下状态值在 lambda 之间的传递也需要特别的小心, 容易出错, 并且难以调试。

2.3 coroutine 实现部分

coroutine 部分之前的帖子里已经写得比较详细了, 这里仅给出链接以及简单的代码示例:

  1. 如何在 C++17 中实现 stackless coroutine 以及相关的任务调度器

  2. C++20 Coroutine 实例教学

  3. 另外还有一个 purecpp 大会的演讲视频, 主要内容与上述的两篇文章相关度比较高, 这里也给出相关的链接, 感兴趣的同学可以自行观看:C++20 coroutine 原理与应用

代码示例:

//C++ 20 coroutineauto clientProxy = mRpcClient->CreateServiceProxy("mmo.HeartBeat");mScheduler.CreateTask20([clientProxy]()                         -> rstudio::logic::CoResumingTaskCpp20 {    auto* task = rco_self_task();
printf("step1: task is %llu\n", task->GetId()); co_await rstudio::logic::cotasks::NextFrame{}; printf("step2 after yield!\n"); int c = 0; while (c < 5) { printf("in while loop c=%d\n", c); co_await rstudio::logic::cotasks::Sleep(1000); c++; } for (c = 0; c < 5; c++) { printf("in for loop c=%d\n", c); co_await rstudio::logic::cotasks::NextFrame{}; }
printf("step3 %d\n", c); auto newTaskId = co_await rstudio::logic::cotasks::CreateTask(false, []()-> logic::CoResumingTaskCpp20 { printf("from child coroutine!\n"); co_await rstudio::logic::cotasks::Sleep(2000); printf("after child coroutine sleep\n"); }); printf("new task create in coroutine: %llu\n", newTaskId); printf("Begin wait for task!\n"); co_await rstudio::logic::cotasks::WaitTaskFinish{ newTaskId, 10000 }; printf("After wait for task!\n");
rstudio::logic::cotasks::RpcRequest rpcReq{clientProxy, "DoHeartBeat", rstudio::reflection::Args{ 3 }, 5000}; auto* rpcret = co_await rpcReq; if (rpcret->rpcResultType == rstudio::network::RpcResponseResultType::RequestSuc) { assert(rpcret->totalRet == 1); auto retval = rpcret->retValue.to<int>(); assert(retval == 4); printf("rpc coroutine run suc, val = %d!\n", retval); } else { printf("rpc coroutine run failed! result = %d \n", (int)rpcret->rpcResultType); } co_await rstudio::logic::cotasks::Sleep(5000); printf("step4, after 5s sleep\n"); co_return rstudio::logic::CoNil;} );
复制代码

执行结果:

step1: task is 1step2 after yield!in while loop c=0in while loop c=1in while loop c=2in while loop c=3in while loop c=4in for loop c=0in for loop c=1in for loop c=2in for loop c=3in for loop c=4step3 5new task create in coroutine: 2Begin wait for task!from child coroutine!after child coroutine sleepAfter wait for task!service yield call finish!rpc coroutine run suc, val = 4!step4, after 5s sleep
复制代码

整体来看, 协程的使用还是给异步编程带来了很多便利, 但框架本身的实现其实还是有比较多迭代优化的空间的:

  1. asio 的调度部分与 coroutine 部分的实现是分离的

  2. coroutine 暂时只支持主线程

2.4 小结

上面也结合 halo 的实例说到了一些限制, 那么这些问题有没有好的解决办法了, 答案是肯定的, 虽然 execution 并未完全通过提案, 但整体而言, execution 新的 sender/reciever 模型, 对于解决上面提到的一些缺陷, 应该是提供了非常好的思路, 我们下一章节中继续展开.

3. so easy - execution 就是解?

最开始的想法其实比较简单, 结合原来的 framework, 适当引入提案中的 execution 一些比较可取的思路, 让 framework 的异步编程能更多的吸取 c++新特性和 execution 比较高级的框架抽象能力, 提升整个异步库的实现质量. 所以最开始定的主线思路其实是更多的向 execution 倾斜, 怎么了解掌握 execution, 怎么与现在的 framework 结合成了主线思路.

我们选择的基础参考库是来自冲元宇宙这波改名的 Meta 公司的 libunifex, 客观来说, Meta 公司的 folly 库, 以及 libunifex 库的实现质量, 肯定都是业界前沿的, 对 c++新特性的使用和探索, 也是相当给力的. 这些我们后续在分析 libunifex 具体实现的篇章中也能实际感受到.

但深入了解 libunifex 后, 我们会发现, 它的优点有不少:

  1. 尝试为 c++提供表达异步的框架性结构.

  2. 泛型用得出神入化, ponder 在它前面基本是小弟级别的, 一系列泛用性特别强的 template 编程示例, 比如隐含在 sender/receiver 思路内的 lazy evaluate 表达, 如何在大量使用泛型的情况下提供业务定制点等等.

  3. 结构化的表达并发和异步, 相关代码的编写从自由发挥自主把控走向框架化, 约束化, 能够更有序更可靠的表达复杂异步逻辑

  4. 整个执行 pipeline 的组织, 所有信息是 compile time 和 runtime 完备的, dependencies 不会丢失.

  5. 节点之间的值类型是强制检查的, 有问题的情况 , 大多时候 compiler time 就会报错. 有不少优点的同时, 也有很多缺点:

  6. 整个库的实现严重依赖了 c++20 ranges 采用的一种定制手段 cpo, 并且也使用了类似 ranges 的 pipe 表达方法, 理解相关代码存在一定的门坎.(后续会有具体的篇章展开相关的内容)

  7. 库同时向下兼容了 c++17, 但由于 c++17 本身特性的限制, 引入了大量的宏, 以及 X Macros 展开的方式, 导致相关的代码阅读难度进一步提升. 但实际上 c++17 版本并不具备可维护的价值, 依赖 SIFINAE 的实现, 如果中间任何一环报错, 必然需要在 N 屏的报错中寻找有效信息.

  8. libunifex 对 coroutine 的支持存疑, 虽然让 coroutine 可以作为一种 reciever 存在, 但本质上来说, coroutine 其实更适合拿来做流程控制的胶水, 而不是作为异步中的某个节点存在.

  9. 默认的 scheduler 实现质量离工业级还存在一定的距离, 这一点后续的代码分析中也会具体提到. 诸多问题的存在, 可能也是 execution 提案没有短时间内获得通过的原因吧, 但整体来说, execution 本身的理念还是很有参考价值的, 但以它的现状来说, 离最终的解肯定还是有比较大的距离的.

4. 尝试重新思考 - 要什么, 用什么

事情到这个点就有点尴尬了, 原有的 asio, 架构层面来说, 跟新的 execution 是存在落差的. 而项目实践上来说, asio 相当稳扎稳打, 而以 libunifex 当前的状态来说, 离工业化使用其实是有一定距离的. 但 asio 作者在 21 年时候的两篇演讲(更像 coding show):

  1. Talking Async Ep1: Why C++20 is the Awesomest Language for Network Programming

  2. Talking Async Ep2: Cancellation in depth 第一篇基本整个演示了 asio 从最开始的 callback, 到融入 c++20 coroutine 后的优雅异步表达, 我们可以通过下面的代码片断感受一下:

asio 相关示例代码 1

awaitable<void> listen(tcp::acceptor& acceptor, tcp::endpoint target){  for (;;)  {    auto [e, client] = co_await acceptor.async_accept(use_nothrow_awaitable);    if (e)      break;
auto ex = client.get_executor(); co_spawn(ex, proxy(std::move(client), target), detached); }}
复制代码

asio 相关示例代码 2

  auto [e] = co_await server.async_connect(target, use_nothrow_awaitable);  if (!e)  {    co_await (        (          transfer(client, server, client_to_server_deadline) ||          watchdog(client_to_server_deadline)        )        &&        (          transfer(server, client, server_to_client_deadline) ||          watchdog(server_to_client_deadline)        )      );  }
复制代码

对比原来每个 async_xxx()函数后接 callback 的模式, 整个实现可以说是相当的优雅了, 代码的可读性也得到了极大的提高, 这两段代码都来自于上面的演讲中, 想深入了解的可以直接打开相关的链接观看视频, 很推荐大家去看一下.   能够把复杂的事情用更简洁易懂的方法表达, 这肯定是让人振奋的, 当然, 深入了解相关实现后, 也会发现存在一些问题, 但我们的本意是参考学习, 得出最终想要的可以比较好的支撑并发和异步业务的基础框架, 有这些, 其实已经可以理出一条比较清晰的思路了:

  1. execution 部分主要使用它的 sender/receiver 概念, 和它提供的一些通用的算法. 移除掉所有因为 fallback c++17 引入的大量代码噪声. 抛弃它并不完备的各种 scheduler 实现

  2. 协程借鉴部分 asio 的思路, 首先让协程可以基于 context 上下文, 在跨线程的情况下使用, 另外更多还是使用原有框架有明确的 scheduler 的方式对所有协程进行管理和定制的模式.

  3. 使用 asio 的 scheduler 部分作为 execution 的底层 scheduler 实现, 同时也使用 asio 的 timer 表达, 去除原始 libunifex 依赖不同 scheduler 提供 schedule_at()方法来执行定时器相关逻辑的实现.

  4. 根据业务需要, 定制一些必要的 sender adapter 等简化业务的使用.

  5. 尝试用 execution 框架对接 ISPC 等特殊的并发库, 能够以一个清晰的方式来表达这种混合环境上执行的逻辑.

参考资料

​推荐一个零声教育 C/C++后台开发的免费公开课程,个人觉得老师讲得不错,分享给大家:C/C++后台开发高级架构师,内容包括Linux,Nginx,ZeroMQ,MySQL,Redis,fastdfs,MongoDB,ZK,流媒体,CDN,P2P,K8S,Docker,TCP/IP,协程,DPDK等技术内容,立即学习


原文:C++异步从理论到实践总览篇

用户头像

还未添加个人签名 2022.05.06 加入

还未添加个人简介

评论

发布
暂无评论
彻底了解C++异步从理论到实践_网络编程_C++后台开发_InfoQ写作社区