C++异步：libunifex中的concepts详解！

Posted 2022-07-09 腾讯云开发者

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定义。

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实现