Some comments on the scheduler system
The scheduler in rxcpp v2 is based on the scheduler and worker constructs that RxJava uses (Eric Meijer was involved) The docs for RxJava will have an explanation for
worker. RxCpp adds
scheduler owns a timeline that is exposed by the
scheduler is also a factory for workers in that timeline. Since a scheduler owns a timeline it is possible to build schedulers that time-travel. The virtual-scheduler is a base for the test-scheduler that uses this to make multi-second tests complete in ms.
worker owns a queue of pending
schedulables for the timeline and has a lifetime. When the time for an
schedulable is reached the
schedulable is run. The queue maintains insertion order so that when N
schedulables have the same target time they are run in the order that they were inserted into the queue. The
worker guarantees that each
schedulable completes before the next
schedulable is started. when the
worker's lifetime is unsubscribed all pending
schedulables are discarded.
schedulable owns a function and has a
worker and a
lifetime. When the
lifetime is unsubscribed the
schedulable function will not be called. The
schedulable is passed to the function and allows the function to reschedule itself or schedule something else on the same worker.
The new concepts are
coordinator. I added these to simplify operator implementations and to introduce pay-for-use in operator implementations. Specifically, in Rx.NET and RxJava, the operators use atomic operations and synchronization primitives to coordinate messages from multiple streams even when all the streams are on the same thread (like UI events). The
identity_... coordinations in RxCpp are used by default and have no overhead. The
observe_on_... coordinations use mutex and queue-onto-a-worker respectively, to interleave multiple streams safely.
coordination is a factory for
coordinators and has a scheduler.
coordinator has a
worker, and is a factory for coordinated observables, subscribers and schedulable functions.
All the operators that take multiple streams or deal in time (even
observe_on) take a coordination parameter, not scheduler.
Here are some supplied functions that will produce a coordination using a particular scheduler.
There is no thread-pool scheduler yet. A thread-pool scheduler requires taking a dependency on a thread-pool implementation since I do not wish to write a thread-pool. My plan is to make a scheduler for the windows thread-pool and the apple thread-pool and the boost asio executor pool.. One question to answer is whether these platform specific constructs should live in the rxcpp repo or have platform specific repos.