C++异步:libunifex中的concepts详解!

Posted 腾讯云开发者

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了C++异步:libunifex中的concepts详解!相关的知识,希望对你有一定的参考价值。

导语 | 本篇我们将深入libunifex的concepts设计的方方面面,结合libunifex中的各种concepts实现,更加深入地了解整个库的实现。希望对此方面感兴趣的开发者们给予一点经验和帮助。

前言

在前文《C++异步变化:libunifex实现!》中我们对libunifex的整体实现做了概要性的介绍,本篇我们将深入libunifex的concepts设计的方方面面,结合libunifex中的各种concepts实现,更加深入了解整个库的实现。

一、Sender/Receiver机制概述

上一篇中我们也提到过,与普通函数通过return来返回值相比,libunifex中的Sender和Receiver所表达的是这样一种关系:一个作为生产者的Sender对象通过:

  • set_value。

  • set_done。

  • set_error这三个cpo的其中一个来向作为消费者的Receiver对象传递值的。

如上图所示,类同于一个function或者一个lambda通过return value来返回调用结果,Sender其实就是一个异步任务中的function/lambda,只不过对比普通函数来说,Sender的对应的值传递的时机更复杂,同时也存在三个不同的分支来代表不同的异步处理状态。

当然,实际我们要做到为异步任务提供标准接口的同时,我们还需要让我们的异步任务能够自由的被各种通用异步算法组合,同时我们还希望相关的实现是具备lazy性质的。


(一)Structured Concurrency与Lazy Execution

比较多Execution的文章,都会聊到Lazy Execution,但Lazy Execution感觉更多是结果,这种特性更多的是来源于Structured Concurrency本身的设计。

这里我们只给出基本的概念,Structed Concurrency会在专门的一篇文章中进行展开。

我们先来看一下整个execution的概览图:

execution相关的设计可以认为是一种从DSL->Compiler->Execute过程都很完备的小型专用语言。对于一个使用structured concurrency来表达的异步操作来说:

  • DSL(BNF组成)-Concurrency Pipeline::=Sender Factory '|' Sender Adapter '|' Receiver。

  • Compiler-通常情况我们可以将connect()看成是编译过程,借由Compiler Time的特性支持,我们可以通过connect()产生runtime所需的OperationState。

  • Execute-这个阶段就很自然了,OperationState的start() 就是DSL本身执行的入口点,当然,执行结果最后是通过:set_value,set_error,set_done这几个 receiver cpos来传递的。

所以对于lazy evaluate这个特性,因为设计本身就是有明显的DSL->Compiler->Execute过程的,必然就会具备lazy性质了(直到最后一步才执行Execute操作)。

structured concurrency本身的完备设计和系统性,为我们定制,或者更进一步支持异构计算打下了良好的基础。

接下来我们来具体的看一下整个connect-start的细节:

(二)发起一个异步操作

libunifex用的方式其实并不复杂,通过connect()操作,Sender和Receiver会生成一个中间态的OperationState对象,这个对象负责持有异步任务需要的所有状态和封装所有需要的逻辑。

对于pipeline组织的异步任务来说,整个OperationState是一步一步按层组织到组织到一起的,每个OperationState负责自己的状态管理,并负责生成每一步的操作结果,整个connect的过程我们可以简单理解为一个树状的OperationState的组织过程。connect()返回的结果则是这个树的根节点,也是一个OperationState。

到了真正执行OperationState的start()阶段,整个过程就简单了,我们以整个OperationState树的根节点作为入口,依次触发作为子节点OperationState的执行,直到整个异步任务执行完成。

所以从这里我们其实也能很明显的看到,通过connect-start机制,状态的保存和生命周期控制都交由OperationState来负责了,connect的时候产生相应的OperationState,在start()执行结束后,相应的OperationState即可释放了,很多时候我们甚至可以很方便的在栈上分配一个OperationState,在start结束后再释放它,这样也避免了过多的堆分配导致的性能下降。

这种构造,首先是很容易产生lazy evaluation的作用,利用生成的OperationState,我们可以将这个OperationState放在任何需要的地方执行,另外,整个OperationState这个时候也形成了一个类似AST(Abstract Syntax Tree)的存在,我们发现lambda post过程中丢失的节点与节点之间的值类型约束,清晰的上下文关系,因为这颗树的存在都解决了,是不是比大家想的要简单? 其实并不简单了,execution提案发展到这一步,大家看看中间各个版本的差异就能感知到了。只不过一些复杂问题的解决,往往到最后大家回头看,相关的方案总是简洁明了,理解起来相对顺畅的。

相较于lambda post依赖lambda对异步执行的中间状态进行保存的即时方式,connect-start机制中整个异步任务中的状态管理和生命周期控制都借由OperationState的存在具像化了,一方面代码具备了可复用性,另外整个机制的约束性明显也从自由使用,自行保证正确性,向带约束的使用,误用的地方会有明显的compiler time报错转移了,避免大家踩一些不容易发现又容易写错的坑。另外对于lambda post下我们经常使用智能指针来对状态进行保存和传递的方式比,connect-start机制也能允许我们更多的使用性能更高的栈对象来处理异步逻辑。


(三)Object-state的生命周期

通过前面的讲述,不难理解,对于connect-start模型来说,一个OperationState的生命周期,其实是在connect()后,到start()调用set_value,set_done,set_error这三者之间的任何一个的时刻,再已经返回异步调用的结果后,我们都应该保证receiver此时是知道对应的OpreationState已经被销毁,任何对OperationState的操作都是非法的了。

另外需要注意的一点是,OperationState对象应该是不存在move和copy语义的,正常来说,一个OperationState的构建都应该是作为connect()的返回值,作为值类型即时构建并返回的。


(四)一个异步操作的完成

libunifex中要完成一个异步任务,们以connect传入的receiver作为首个参数调用set_value,set_done,set_error这三者中的任何一个,相关的异步任务就执行完成了,三者的区别如下:

  • set_value用来表示一个成功执行的异步任务,同时后续会追加0..N个任务执行成功的返回值。

  • set_error毫无疑问用来表示一个执行失败的异步任务了。

  • set_done比较特殊,用来表示set_value和set_error之外的情况,很多时候是表示相关的结果不再需要了,比如A, B两个节点同时执行,我们只需要其中一个节点的结果,A成功执行,我们希望通过取消机制取消B的执行,这种情况如果B本身是一个可取消的节点,那就可以直接返回set_done了。很多时候你可以将set_done当成无返回值,或者返回值为空的情况。

这里我们需要特别注意一下 set_value(receiver) 和set_done(receiver)之间的区别,前者表达的意思很明确,就是一次成功的无任何返回值的异步调用,而后者则可能表达的是异步任务并未成功执行的情况。


二、Receiver Concept

简单来说,任何能够使用set_value,set_error,set_done这三个receiver cpos中的一个的类型都可以作为一个合法的Receiver。

我们也能将Receiver作为异步任务中的继续点来考虑,通过多层receiver cpos的级联,最终组成了我们的整个异步任务。

libunifex中并没有单独存在的通用receiver concept,只有几类针对不同receiver cpos的receiver concept 定义:

  • value_receiver<Values...>-一个能够接受set_value(receiver,Values...)的receiver concept定义。

  • error_receiver<Error>-一个能够接受set_error(receiver,Error)的receiver concept定义。

  • done_receiver-一个能够接受set_done(receiver)receiver concept定义。

相关的代码如下:

namespace unifex

  // CPOs
  inline constexpr unspecified set_value = unspecified;
  inline constexpr unspecified set_error = unspecified;
  inline constexpr unspecified set_done = unspecified;


  template<typename R>
  concept __receiver_common =
    std::move_constructible<R> &&
    std::destructible<R>;


  template<typename R>
  concept done_receiver =
    __receiver_common<R> &&
    requires(R&& r) 
        set_done((R&&)r);
    ;


  template<typename R, typename... Values>
  concept value_receiver =
    __receiver_common<R> &&
    requires(R&& r, Values&&... values) 
      set_value((R&&)r, (Values&&)values...);
    ;


  template<typename R, typename Error>
  concept error_receiver =
    __receiver_common<R> &&
    requires(R&& r, Error&& error) 
      set_error((R&&)r, (Error&&)error);
    ;

这部分代码还是比较好理解的,receiver首先要同时满足可move构造和析构,然后就是done_receiver,value_receiver,error_receiver这个concept的定义了。

小技巧: 注意上面的:inline constexpr unspecified set_value= unspecified;

c++20后,很多cpo的定义都会使用这种方式在文档中出现,这基本都是代表对应的声明是一个cpo对象,因为cpo的声明其实并不简单,直接省略掉具体的cpo实现,会方便我们更好的阅读代码,比如上面的set_value, set_error,set_done这三个receiver cpos。

不同的Sender类型一般有不同的完成通知,所以对于connect()传入的Receiver会有不一样的约束。很多时候我们会根据相关的Sender实现来组合相关的Receiver约束,例如一个Sender如果在不同的分支处理下同时存在set_value,set_done和set_error调用的话,我们需要同时复合三种类型的receiver约束来组合出最终的receiver约束。


Receiver Query

同时,Receiver还有一种特殊的用法,用于作为caller和callee传递上下文信息的存在,比如对于get_stop_token(receiver)这个cpo来说,它可以用来查询当前receiver支持的stop_token类型,从而在相应的sender处调用合适的代码来处理相关的逻辑。


三、Sender Concept

对比明确的Receiver定义,libunifex中并没有一个通用的Sender约束,这是因为我们编译期没法知道Sender是否触发了三个receiver cpos中的一个,libunifex通过一种间接的方式来对Sender进行约束:

namespace unifex

  // Sender CPOs
  inline constexpr unspecified connect = unspecified;


  // Test whether a given sender and receiver can been connected.
  template<typename S, typename R>
  concept sender_to =
    requires(S&& sender, R&& receiver) 
      connect((S&&)sender, (R&&)receiver);
    ;

libunifex通过对connect接受的sender和receiver来间接的约束sender,也就是满足connect()调用的sender即是合法的sender。

虽然libunifex官方想通过加sender_traits和is_sender<T>的方式来改善相关实现,但在泛型加上不定时机两者的作用下,相关的实现不可能太智能, 可能依然还是库作者向如TypedSender对应的类型traits实现,也就是约束是由对应Sender的实现者来手动维护的。

四、TypedSender Concept

TypedSender其实并不是一个标准的concept实现,它仅仅是提供了一种编译期的traits,让我们有能力在编译期查询对应TypedSender能够支持的set_value和set_error的参数类型,使我们可以利用这些信息更好的约束数据在异步节点间的传递,当类型不匹配的情况下,compiler会直接报错。我们来看一个实际的例子:

struct some_typed_sender 
 template<template<typename...> class Variant,
          template<typename...> class Tuple>
 using value_types = Variant<Tuple<int>,
                             Tuple<std::string, int>,
                             Tuple<>>;


 template<template<typename...> class Variant>
 using error_types = Variant<std::exception_ptr>;


 static constexpr bool sends_done = true;
 ...
;

对于一个TypedSender实现,我们可以通过它内嵌的类型:

  • value_types-编译期获取其支持的set_value参数类型。

  • error_types-编译期获取其支持的set_error参数类型。

  • sends_done-该bool值用于判断对应sender是否支持以set_done调用结束异步操作。

对于上面的例子来说,我们容易知道,对应的TypedSender定义可能会对关联的Receiver调用以下函数重载:

  • set_value(R&&,int)

  • set_value(R&&,std::string,int)

  • set_value(R&&)

  • set_error(R&&,std::exception_ptr)

  • set_done(R&&)

不过我们的代码实现中,一般很少直接使用这些内嵌类型,而是通过sender_traits<Sender>来使用它们。

比如:typename unifex::sender_traits<Sender>::template value_types<std::variant, std::tuple>


五、OperationState Concept

一个OperationState对象包含了一个异步任务的所有内部状态。它同样也不是一个直接的类型约束。

我们知道一个OperationState是由connect(Sender,Receiver)返回,同时一个OperationState也是不能被move和copy的。对于一个OperationState对象,它只支持两种操作:

  • start()

  • 析构

一个OperationState的析构只有在未调用start()前,或者在已经发出任务结束通知的情况下才是合法的。

namespace unifex

  // CPO for starting an async operation
  inline constexpr unspecified start = unspecified;


  // CPO for an operation-state object.
  template<typename T>
  concept operation_state =
    std::destructible<T> &&
    requires(T& operation) 
      start(operation);
    ;

从上面的代码中可以看到, 对于OperationState的约束, 主要就两点:

  • 支持析构

  • 支持start()操作


六、Scheduler Concept

之前的execution概览图中,我们其实也能看到,真正负责异常操作执行的,就是Scheduler和更底层的ExecutionContext。libunifex中的Scheduler其实就是一个轻量的Wrapper,真正负责异步任务执行的是底层的Execution Context实现。对于非异构的实现,这里的Execution Context一般代表一个Cpu线程或者一组Cpu线程(线程池),最简单的情况是相关的任务被投递到一个线程上来执行。

libunifex有两种方式挂接相关的异步任务到Scheduler对象上:

  • 调用schedule(Scheduler) 这个cpo生成一个sender,然后再利用pipeline关联异步操作到这个sender上。

  • via(sender,Scheduler) 和typed_via(sender,Scheduler)这两个sender adapter来中间切换后续任务执行的Execution Context。

schedule()的实现机制并不复杂,Execution Context是cpu线程来举例说明,底层会保证在关联的线程上执行一个参数类型为void的set_value(),这样因为sender和sender adapter以及receiver的组合机制,在没有via()或者typed_via()等特殊节点存在的情况下,整个pipeline的相关内容都会在这个线程上被执行。这样也就间接的通过一个set_value()的调用位置决定了整个pipeline执行所归属的Execution Context。

一个Scheduler Concept的定义如下:

namespace unifex

  // The schedule() CPO
  inline constexpr unspecified schedule = ;


  // The scheduler concept.
  template<typename T>
  concept scheduler =
    std::is_nothrow_copy_constructible_v<T> &&
    std::is_nothrow_move_constructible_v<T> &&
    std::destructible<T> &&
    std::equality_comparable<T> &&
    requires(const T cs, T s) 
      schedule(cs); // TODO: Constraint this returns a sender of void.
      schedule(s);
    ;

(一)TimeScheduler

TimeScheduler是Scheduler的扩展实现,除了Scheduler本身的支持外,TimeScheduler还支持在指定的时间执行异步任务的特性。扩展的能力包括以下:

  • typename TimeScheduler::time_point

  • now(ts) -> time_point

  • schedule_at(ts,time_point) -> sender_of<void>

  • schedule_after(ts,duration) -> sender_of<void>

其中:

  • now(ts)-用来返回当前时刻点的cpo

  • schedule_at(ts,time_point)-用于产生一个在指定时刻点调度的sender

  • schedule_after(ts,duration)-用于产生一个在duration指定的时间段后调度的sender

另外,通过time_point类型,用户可以按需定制实现自己的Scheduler来支持自定义的时间单位,以适配复杂的业务需要。

具体的TimerScheduler的Concept定义如下:

namespace unifex

  // TimeScheduler CPOs
  inline constexpr unspecified now = unspecified;
  inline constexpr unspecified schedule_at = unspecified;
  inline constexpr unspecified schedule_after = unspecified;


  template<typename T>
  concept time_scheduler =
    scheduler<T> &&
    requires(const T scheduler) 
      now(scheduler);
      schedule_at(scheduler, now(scheduler));
      schedule_after(scheduler, now(scheduler) - now(scheduler));
    ;

七、其他Concepts实现

(一)StopToken

cancellation机制相关的concept,又是一个比较难讲清的话题,本篇暂时不具体展开了,有时间会单独阐述这部分的具体实现。


(二)ManySender/ManyReceiver Concept

正常来说,一个Sender只会触发一次完成信号,ManySender允许多次触发完成信号,与ManySender对应,也会有ManyReceiver,这部分更多是对Sender/Receiver的一种扩展,本系列的讲述中也不会实际用到它们,这里先直接略过,感兴趣的同学可以自行查阅相关文档。


(三)Stream Concept

可以理解为一种lazy性质的ManySender,当消费者主动调用next()的时候,才会产生值,本系列的讲述中也不会直接使用到,这里直接略过了,感兴趣的读者可以自行查阅相关文档。


八、总结

可以说几个关键性的cpo:

  • receiver cpos:

    set_value()、set_done()、set_error()

  • start()

  • connect()

再加上一系列的Concepts定义:

  • Receiver Concept

  • Sender Concept

  • OperationState Concept

  • Scheduler Concept

  • ...

将整个execution的执行框架勾勒出来了,使整个异步操作有了一个通用的基础框架,而Execution Context则是框架运行的基础,各种Algorithm实现则是细节,后续我们会分别介绍,带大家深入框架运行的基础以及实现的细节,更进一步的了解libunifex的实现。


参考资料:

1.libunifex源码库

2.Madokakaroto-浅谈The C++ Executors

3.Madokakaroto -The C++ Executors设计与演化的简史

 作者简介

沈芳

腾讯后台开发工程师

IEG研发效能部开发人员,毕业于华中科技大学。目前负责CrossEngine Server的开发工作,对GamePlay技术比较感兴趣。

 推荐阅读

万卷共知,一书一页总关情,TVP读书会带你突围阅读迷障!

C++异步变化:libunifex实现!

硬核!Redis知识总结,建议收藏

C++特殊定制:揭秘cpo与tag_invoke!‍

温馨提示:因公众号平台更改了推送规则,公众号推送的文章文末需要点一下“赞”和“在看”,新的文章才会第一时间出现在你的订阅列表里噢~

以上是关于C++异步:libunifex中的concepts详解!的主要内容,如果未能解决你的问题,请参考以下文章

C++异步:structured concurrency实现解析!

C++异步变化:libunifex实现!

c++异步:asio的scheduler实现!

c++异步:asio的scheduler实现!

如何使用 C++ 概念(“concepts lite”)支持构建 gcc?

Concept check