Rxcpp 中的调度程序

Question

我正在尝试找出 Rx 的 C++ version 中的调度模型。

了解 C# 版本，其中有一个带有一个 Schedule 方法的简单接口； C++ 版本似乎相当复杂，包含调度程序、工作程序和协调等内容。

对我来说，一个主要的缺失部分是线程池调度程序的实现，它是否存在其他名称？我将如何实现它自己？我应该把它写在PPL（Windows）上面吗？如果我需要在它上面有一个序列化的（Actor like）观察者，我应该使用什么？窥视here 和here 可以表明这不是一项简单的任务。

这确实有助于获得有关该主题的某种概述，因为official 文档是自动生成的并且仍然非常稀疏。

Answer 1

是的，生成的文档是新的，并且尚未记录日程安排。

rxcpp v2 中的调度器基于 RxJava 使用的调度器和工作器构造（涉及 Eric Meijer） RxJava 的文档将对调度程序和工作程序进行解释。 rxcpp增加了schedulable、coordination和coordinator。

scheduler 拥有一个由now() 方法公开的时间线。 scheduler 也是该时间线中workers 的工厂。由于调度程序拥有时间线，因此可以构建时间旅行的调度程序。 virtual-scheduler 是 test-scheduler 的基础，它使用它在毫秒内完成多秒测试。

worker 拥有一个等待时间线的队列 @9​​87654327@s 并且有一个生命周期。当达到schedulable 的时间时，运行schedulable。队列维护插入顺序，以便当 N 个schedulables 具有相同的目标时间时，它们会按照它们插入队列的顺序运行。 worker 保证每个schedulable 在下一个schedulable 开始之前完成。当worker 的生命周期被取消订阅时，所有待处理的schedulables 都将被丢弃。

schedulable 拥有一个函数，拥有一个worker和一个lifetime。当schedulable 的生命周期被取消订阅时，schedulable 函数将不会被调用。 schedulable 被传递给函数，并允许函数重新调度自己或在同一个工作人员上安排其他事情。

新概念是协调和协调者。我添加了这些以简化运营商实现并在运营商实现中引入按使用付费。具体来说，在 Rx.NET 和 RxJava 中，操作员使用原子操作和同步原语来协调来自多个流的消息，即使所有流都在同一个线程上（如 UI 事件）。 rxcpp 中的identity_. . . 协调默认使用并且没有开销。 syncronize_. . . 和 observe_on_. . . 协调分别使用 mutex 和 queue-onto-a-worker，以安全地交错多个流。

coordination 是 coordinators 的工厂，有一个 scheduler。

coordinator 有一个worker，是协调observables、subscribers 和schedulable 函数的工厂。

所有采用多个流或及时处理的运算符（甚至 subscribe_on 和 observe_on）都采用协调参数，而不是调度程序。

以下是一些提供的函数，它们将使用特定的调度程序产生协调。

• identity_immediate()
• identity_current_thread()
• identity_same_worker(worker w)
• serialize_event_loop()
• serialize_new_thread()
• serialize_same_worker(worker w)
• observe_on_event_loop()
• observe_on_new_thread()

目前还没有线程池调度程序。线程池调度程序需要依赖于线程池实现，因为我不想编写线程池。我的计划是为 windows 线程池、apple 线程池和 boost asio 执行器池创建一个调度程序。要回答的一个问题是这些特定于平台的构造是否应该存在于 rxcpp 存储库中或具有特定于平台的存储库。

欢迎贡献、意见和想法！