Skip to content

Commit

Permalink
Cover all possible schedulables in main thread scheduler qt (#550)
Browse files Browse the repository at this point in the history 
* fix main thread scheduler

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
  • Loading branch information
@victimsnino @pre-commit-ci
victimsnino and pre-commit-ci[bot] authored Apr 1, 2024
1 parent cf14a07 commit 6def295
Show file tree
Hide file tree
Showing 2 changed files with 61 additions and 4 deletions.
34 changes: 31 additions & 3 deletions src/rppqt/rppqt/schedulers/main_thread.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
#include <rppqt/schedulers/fwd.hpp> // own forwarding
#include <rppqt/utils/exceptions.hpp>

#include "rpp/schedulers/fwd.hpp"

#include <QCoreApplication>
#include <QTimer>
#include <chrono>
Expand All @@ -37,17 +39,43 @@ namespace rppqt::schedulers
{
const auto application = QCoreApplication::instance();
if (!application)
throw utils::no_active_qapplication{"Pointer to application is null. Create QApplication before using main_thread_scheduler!"};
{
handler.on_error(std::make_exception_ptr(utils::no_active_qapplication{"Pointer to application is null. Create QApplication before using main_thread_scheduler!"}));
return;
}

QTimer::singleShot(std::chrono::duration_cast<std::chrono::milliseconds>(duration), application, [fn = std::forward<Fn>(fn), handler = std::forward<Handler>(handler), ... args = std::forward<Args>(args)]() mutable {
if (const auto new_duration = fn(handler, args...))
defer_for(new_duration->value, std::move(fn), std::move(handler), std::move(args)...);
if (!handler.is_disposed())
invoke(std::move(fn), std::move(handler), std::move(args)...);
});
}

static constexpr rpp::schedulers::details::none_disposable get_disposable() { return {}; }

static rpp::schedulers::time_point now() { return rpp::schedulers::clock_type::now(); }

private:
template<rpp::schedulers::constraint::schedulable_handler Handler, typename... Args, rpp::schedulers::constraint::schedulable_delay_from_now_fn<Handler, Args...> Fn>
static void invoke(Fn&& fn, Handler&& handler, Args&&... args)
{
if (const auto new_duration = fn(handler, args...))
defer_for(new_duration->value, std::forward<Fn>(fn), std::forward<Handler>(handler), std::forward<Args>(args)...);
}

template<rpp::schedulers::constraint::schedulable_handler Handler, typename... Args, rpp::schedulers::constraint::schedulable_delay_from_this_timepoint_fn<Handler, Args...> Fn>
static void invoke(Fn&& fn, Handler&& handler, Args&&... args)
{
const auto now = rpp::schedulers::clock_type::now();
if (const auto new_duration = fn(handler, args...))
defer_for(now + new_duration->value - rpp::schedulers::clock_type::now(), std::forward<Fn>(fn), std::forward<Handler>(handler), std::forward<Args>(args)...);
}

template<rpp::schedulers::constraint::schedulable_handler Handler, typename... Args, rpp::schedulers::constraint::schedulable_delay_to_fn<Handler, Args...> Fn>
static void invoke(Fn&& fn, Handler&& handler, Args&&... args)
{
if (const auto new_tp = fn(handler, args...))
defer_for(new_tp->value - rpp::schedulers::clock_type::now(), std::forward<Fn>(fn), std::forward<Handler>(handler), std::forward<Args>(args)...);
}
};

public:
Expand Down
31 changes: 30 additions & 1 deletion src/tests/rppqt/test_main_thread_scheduler.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,17 @@
#include <rppqt/schedulers/main_thread.hpp>

#include "mock_observer.hpp"
#include "rpp/disposables/fwd.hpp"
#include "rpp/schedulers/fwd.hpp"

#include <QApplication>
#include <future>
#include <optional>

TEST_CASE("main_thread_scheduler schedules actions to main thread")
{
auto observer = mock_observer_strategy<int>{}.get_observer().as_dynamic();
auto d = rpp::composite_disposable_wrapper::make();
auto observer = mock_observer_strategy<int>{}.get_observer(d).as_dynamic();

int argc{};
QCoreApplication application{argc, nullptr};
Expand All @@ -47,6 +51,24 @@ TEST_CASE("main_thread_scheduler schedules actions to main thread")
}
}

SECTION("nothing happens for disposed handler")
{
std::promise<std::thread::id> execution_thread{};
std::thread{[&] {
rppqt::schedulers::main_thread_scheduler::create_worker().schedule([&](const auto&) -> rpp::schedulers::optional_delay_from_now {
execution_thread.set_value(std::this_thread::get_id());
return {};
},
observer);
}}.join();

d.dispose();

application.exec();
auto future = execution_thread.get_future();
REQUIRE(future.wait_for(std::chrono::seconds{1}) == std::future_status::timeout);
}

SECTION("recursive scheduling to main thread")
{
std::string execution{};
Expand All @@ -68,4 +90,11 @@ TEST_CASE("main_thread_scheduler schedules actions to main thread")
application.exec();
CHECK(execution == "outer inner outer inner ");
}

SECTION("scheduler can be applied for all types of schedulables")
{
rppqt::schedulers::main_thread_scheduler::create_worker().schedule([&](const auto&) -> rpp::schedulers::optional_delay_from_now { return std::nullopt; }, observer);
rppqt::schedulers::main_thread_scheduler::create_worker().schedule([&](const auto&) -> rpp::schedulers::optional_delay_from_this_timepoint { return std::nullopt; }, observer);
rppqt::schedulers::main_thread_scheduler::create_worker().schedule([&](const auto&) -> rpp::schedulers::optional_delay_to { return std::nullopt; }, observer);
}
}

1 comment on commit 6def295

@github-actions
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

BENCHMARK RESULTS (AUTOGENERATED)

ci-ubuntu-gcc

General
name rxcpp rpp prev rpp ratio
Subscribe empty callbacks to empty observable 312.22 ns 2.16 ns 2.16 ns 1.00
Subscribe empty callbacks to empty observable via pipe operator 313.65 ns 2.16 ns 2.16 ns 1.00

Sources
name rxcpp rpp prev rpp ratio
from array of 1 - create + subscribe + immediate 721.21 ns 0.31 ns 0.31 ns 1.00
from array of 1 - create + subscribe + current_thread 1065.71 ns 5.25 ns 5.25 ns 1.00
concat_as_source of just(1 immediate) create + subscribe 2205.21 ns 114.36 ns 114.24 ns 1.00
defer from array of 1 - defer + create + subscribe + immediate 758.84 ns 0.31 ns 0.31 ns 1.00
interval - interval + take(3) + subscribe + immediate 2154.29 ns 59.23 ns 59.23 ns 1.00
interval - interval + take(3) + subscribe + current_thread 3048.68 ns 32.73 ns 32.73 ns 1.00

Filtering Operators
name rxcpp rpp prev rpp ratio
immediate_just+take(1)+subscribe 1117.68 ns 0.31 ns 0.31 ns 1.00
immediate_just+filter(true)+subscribe 868.27 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+skip(1)+subscribe 1032.43 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe 883.58 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+first()+subscribe 1255.60 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+last()+subscribe 930.22 ns 0.31 ns 0.31 ns 1.00
immediate_just+take_last(1)+subscribe 1152.36 ns 18.82 ns 18.52 ns 1.02

Schedulers
name rxcpp rpp prev rpp ratio
immediate scheduler create worker + schedule 280.35 ns 2.16 ns 2.16 ns 1.00
current_thread scheduler create worker + schedule 381.40 ns 7.10 ns 7.15 ns 0.99
current_thread scheduler create worker + schedule + recursive schedule 836.08 ns 65.96 ns 64.58 ns 1.02

Transforming Operators
name rxcpp rpp prev rpp ratio
immediate_just+map(v*2)+subscribe 878.87 ns 0.31 ns 0.31 ns 1.00
immediate_just+scan(10, std::plus)+subscribe 914.43 ns 0.31 ns 0.31 ns 1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe 2350.80 ns 159.85 ns 169.14 ns 0.95
immediate_just+buffer(2)+subscribe 1571.06 ns 14.52 ns 14.20 ns 1.02
immediate_just+window(2)+subscribe + subscsribe inner 2341.97 ns 1060.98 ns 1037.45 ns 1.02

Conditional Operators
name rxcpp rpp prev rpp ratio
immediate_just+take_while(false)+subscribe 851.98 ns - - 0.00
immediate_just+take_while(true)+subscribe 847.07 ns 0.31 ns 0.31 ns 1.00

Utility Operators
name rxcpp rpp prev rpp ratio
immediate_just(1)+subscribe_on(immediate)+subscribe 2050.92 ns 0.31 ns 0.31 ns 1.00

Combining Operators
name rxcpp rpp prev rpp ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe 3600.06 ns 179.65 ns 175.99 ns 1.02
immediate_just(1) + merge_with(immediate_just(2)) + subscribe 3672.68 ns 177.68 ns 165.10 ns 1.08
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe - 137.86 ns 141.57 ns 0.97
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe 3585.15 ns 1152.28 ns 1126.43 ns 1.02
immediate_just(1) + zip(immediate_just(2)) + subscribe 2112.64 ns 218.61 ns 214.10 ns 1.02

Subjects
name rxcpp rpp prev rpp ratio
publish_subject with 1 observer - on_next 34.49 ns 11.72 ns 12.02 ns 0.98

Scenarios
name rxcpp rpp prev rpp ratio
basic sample 1412.84 ns 14.51 ns 14.55 ns 1.00
basic sample with immediate scheduler 1355.05 ns 5.55 ns 5.55 ns 1.00

Aggregating Operators
name rxcpp rpp prev rpp ratio
immediate_just+reduce(10, std::plus)+subscribe 927.04 ns 0.31 ns 0.31 ns 1.00

Error Handling Operators
name rxcpp rpp prev rpp ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe 1056.32 ns 123.76 ns 122.03 ns 1.01

ci-macos

General
name rxcpp rpp prev rpp ratio
Subscribe empty callbacks to empty observable 1149.29 ns 4.49 ns 3.85 ns 1.17
Subscribe empty callbacks to empty observable via pipe operator 1158.63 ns 4.60 ns 3.87 ns 1.19

Sources
name rxcpp rpp prev rpp ratio
from array of 1 - create + subscribe + immediate 2265.80 ns 0.27 ns 0.23 ns 1.17
from array of 1 - create + subscribe + current_thread 2809.93 ns 29.45 ns 25.38 ns 1.16
concat_as_source of just(1 immediate) create + subscribe 6146.40 ns 353.52 ns 328.44 ns 1.08
defer from array of 1 - defer + create + subscribe + immediate 2500.04 ns 0.27 ns 0.23 ns 1.14
interval - interval + take(3) + subscribe + immediate 5834.71 ns 129.58 ns 114.40 ns 1.13
interval - interval + take(3) + subscribe + current_thread 7282.58 ns 142.19 ns 107.70 ns 1.32

Filtering Operators
name rxcpp rpp prev rpp ratio
immediate_just+take(1)+subscribe 3405.09 ns 0.34 ns 0.23 ns 1.47
immediate_just+filter(true)+subscribe 2493.81 ns 0.27 ns 0.23 ns 1.18
immediate_just(1,2)+skip(1)+subscribe 4167.15 ns 0.55 ns 0.23 ns 2.34
immediate_just(1,1,2)+distinct_until_changed()+subscribe 3092.32 ns 0.65 ns 0.47 ns 1.38
immediate_just(1,2)+first()+subscribe 4494.68 ns 0.29 ns 0.23 ns 1.23
immediate_just(1,2)+last()+subscribe 3205.16 ns 0.35 ns 0.23 ns 1.49
immediate_just+take_last(1)+subscribe 4147.51 ns 83.15 ns 71.45 ns 1.16

Schedulers
name rxcpp rpp prev rpp ratio
immediate scheduler create worker + schedule 1150.65 ns 4.84 ns 3.93 ns 1.23
current_thread scheduler create worker + schedule 1413.47 ns 44.39 ns 39.53 ns 1.12
current_thread scheduler create worker + schedule + recursive schedule 2377.02 ns 254.83 ns 215.83 ns 1.18

Transforming Operators
name rxcpp rpp prev rpp ratio
immediate_just+map(v*2)+subscribe 3261.42 ns 0.28 ns 0.23 ns 1.22
immediate_just+scan(10, std::plus)+subscribe 2798.57 ns 0.58 ns 0.47 ns 1.25
immediate_just+flat_map(immediate_just(v*2))+subscribe 7575.46 ns 517.47 ns 419.63 ns 1.23
immediate_just+buffer(2)+subscribe 2937.36 ns 105.30 ns 69.89 ns 1.51
immediate_just+window(2)+subscribe + subscsribe inner 9023.34 ns 2720.89 ns 2275.36 ns 1.20

Conditional Operators
name rxcpp rpp prev rpp ratio
immediate_just+take_while(false)+subscribe 2509.30 ns - - 0.00
immediate_just+take_while(true)+subscribe 3701.12 ns 0.29 ns 0.23 ns 1.26

Utility Operators
name rxcpp rpp prev rpp ratio
immediate_just(1)+subscribe_on(immediate)+subscribe 5858.19 ns 0.30 ns 0.23 ns 1.27

Combining Operators
name rxcpp rpp prev rpp ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe 9420.73 ns 560.80 ns 446.27 ns 1.26
immediate_just(1) + merge_with(immediate_just(2)) + subscribe 10327.77 ns 552.64 ns 446.31 ns 1.24
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe - 570.97 ns 474.71 ns 1.20
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe 9479.97 ns 2253.79 ns 1898.15 ns 1.19
immediate_just(1) + zip(immediate_just(2)) + subscribe 8529.28 ns 1015.14 ns 889.58 ns 1.14

Subjects
name rxcpp rpp prev rpp ratio
publish_subject with 1 observer - on_next 86.44 ns 56.52 ns 47.28 ns 1.20

Scenarios
name rxcpp rpp prev rpp ratio
basic sample 3634.30 ns 127.76 ns 116.81 ns 1.09
basic sample with immediate scheduler 4002.03 ns 20.44 ns 14.07 ns 1.45

Aggregating Operators
name rxcpp rpp prev rpp ratio
immediate_just+reduce(10, std::plus)+subscribe 2821.36 ns 0.27 ns 0.23 ns 1.17

Error Handling Operators
name rxcpp rpp prev rpp ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe 13175.91 ns 5638.36 ns 4084.45 ns 1.38

ci-ubuntu-clang

General
name rxcpp rpp prev rpp ratio
Subscribe empty callbacks to empty observable 275.85 ns 0.88 ns 0.88 ns 1.00
Subscribe empty callbacks to empty observable via pipe operator 281.34 ns 0.88 ns 0.88 ns 1.00

Sources
name rxcpp rpp prev rpp ratio
from array of 1 - create + subscribe + immediate 572.35 ns 0.31 ns 0.31 ns 1.00
from array of 1 - create + subscribe + current_thread 796.49 ns 5.55 ns 5.56 ns 1.00
concat_as_source of just(1 immediate) create + subscribe 1910.57 ns 115.37 ns 113.10 ns 1.02
defer from array of 1 - defer + create + subscribe + immediate 589.16 ns 0.31 ns 0.31 ns 1.00
interval - interval + take(3) + subscribe + immediate 1511.44 ns 58.31 ns 58.31 ns 1.00
interval - interval + take(3) + subscribe + current_thread 2072.89 ns 31.47 ns 31.55 ns 1.00

Filtering Operators
name rxcpp rpp prev rpp ratio
immediate_just+take(1)+subscribe 896.10 ns 0.31 ns 0.31 ns 1.00
immediate_just+filter(true)+subscribe 663.88 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+skip(1)+subscribe 866.00 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe 699.08 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+first()+subscribe 1088.35 ns 0.31 ns 0.31 ns 1.00
immediate_just(1,2)+last()+subscribe 749.00 ns 0.31 ns 0.31 ns 1.00
immediate_just+take_last(1)+subscribe 956.75 ns 0.31 ns 0.31 ns 1.00

Schedulers
name rxcpp rpp prev rpp ratio
immediate scheduler create worker + schedule 202.21 ns 0.88 ns 0.88 ns 1.00
current_thread scheduler create worker + schedule 306.99 ns 5.89 ns 5.58 ns 1.05
current_thread scheduler create worker + schedule + recursive schedule 621.88 ns 59.17 ns 60.81 ns 0.97

Transforming Operators
name rxcpp rpp prev rpp ratio
immediate_just+map(v*2)+subscribe 649.40 ns 0.31 ns 0.31 ns 1.00
immediate_just+scan(10, std::plus)+subscribe 724.34 ns 0.31 ns 0.31 ns 1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe 1776.70 ns 119.26 ns 119.41 ns 1.00
immediate_just+buffer(2)+subscribe 1362.56 ns 13.88 ns 14.51 ns 0.96
immediate_just+window(2)+subscribe + subscsribe inner 2143.32 ns 790.68 ns 811.35 ns 0.97

Conditional Operators
name rxcpp rpp prev rpp ratio
immediate_just+take_while(false)+subscribe 648.14 ns - - 0.00
immediate_just+take_while(true)+subscribe 661.72 ns 0.31 ns 0.31 ns 1.00

Utility Operators
name rxcpp rpp prev rpp ratio
immediate_just(1)+subscribe_on(immediate)+subscribe 1610.92 ns 0.31 ns 0.31 ns 1.00

Combining Operators
name rxcpp rpp prev rpp ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe 2538.04 ns 125.84 ns 126.13 ns 1.00
immediate_just(1) + merge_with(immediate_just(2)) + subscribe 2999.84 ns 121.50 ns 121.43 ns 1.00
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe - 110.66 ns 109.94 ns 1.01
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe 2619.36 ns 727.74 ns 727.96 ns 1.00
immediate_just(1) + zip(immediate_just(2)) + subscribe 1764.02 ns 164.83 ns 167.87 ns 0.98

Subjects
name rxcpp rpp prev rpp ratio
publish_subject with 1 observer - on_next 26.22 ns 13.87 ns 13.88 ns 1.00

Scenarios
name rxcpp rpp prev rpp ratio
basic sample 1041.23 ns 13.28 ns 13.28 ns 1.00
basic sample with immediate scheduler 1013.81 ns 6.17 ns 6.17 ns 1.00

Aggregating Operators
name rxcpp rpp prev rpp ratio
immediate_just+reduce(10, std::plus)+subscribe 758.08 ns 0.31 ns 0.31 ns 1.00

Error Handling Operators
name rxcpp rpp prev rpp ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe 861.88 ns 126.38 ns 126.27 ns 1.00

ci-windows

General
name rxcpp rpp prev rpp ratio
Subscribe empty callbacks to empty observable 582.14 ns 4.02 ns 4.39 ns 0.91
Subscribe empty callbacks to empty observable via pipe operator 590.68 ns 4.01 ns 4.02 ns 1.00

Sources
name rxcpp rpp prev rpp ratio
from array of 1 - create + subscribe + immediate 1177.34 ns 5.55 ns 5.55 ns 1.00
from array of 1 - create + subscribe + current_thread 1473.65 ns 19.77 ns 19.14 ns 1.03
concat_as_source of just(1 immediate) create + subscribe 4658.13 ns 171.29 ns 190.21 ns 0.90
defer from array of 1 - defer + create + subscribe + immediate 1209.11 ns 5.55 ns 5.86 ns 0.95
interval - interval + take(3) + subscribe + immediate 3132.91 ns 134.13 ns 134.17 ns 1.00
interval - interval + take(3) + subscribe + current_thread 3468.73 ns 58.32 ns 65.53 ns 0.89

Filtering Operators
name rxcpp rpp prev rpp ratio
immediate_just+take(1)+subscribe 1857.07 ns 12.85 ns 12.86 ns 1.00
immediate_just+filter(true)+subscribe 1697.63 ns 12.36 ns 12.35 ns 1.00
immediate_just(1,2)+skip(1)+subscribe 1738.42 ns 13.12 ns 13.12 ns 1.00
immediate_just(1,1,2)+distinct_until_changed()+subscribe 1612.70 ns 15.96 ns 15.94 ns 1.00
immediate_just(1,2)+first()+subscribe 2090.66 ns 12.64 ns 12.64 ns 1.00
immediate_just(1,2)+last()+subscribe 1477.45 ns 14.05 ns 14.04 ns 1.00
immediate_just+take_last(1)+subscribe 2070.76 ns 61.45 ns 59.40 ns 1.03

Schedulers
name rxcpp rpp prev rpp ratio
immediate scheduler create worker + schedule 488.42 ns 6.18 ns 6.17 ns 1.00
current_thread scheduler create worker + schedule 664.75 ns 19.14 ns 24.83 ns 0.77
current_thread scheduler create worker + schedule + recursive schedule 1109.02 ns 115.13 ns 129.67 ns 0.89

Transforming Operators
name rxcpp rpp prev rpp ratio
immediate_just+map(v*2)+subscribe 1336.20 ns 12.31 ns 12.36 ns 1.00
immediate_just+scan(10, std::plus)+subscribe 1431.93 ns 21.29 ns 21.27 ns 1.00
immediate_just+flat_map(immediate_just(v*2))+subscribe 3935.74 ns 229.26 ns 229.32 ns 1.00
immediate_just+buffer(2)+subscribe 2372.03 ns 60.73 ns 59.42 ns 1.02
immediate_just+window(2)+subscribe + subscsribe inner 4142.34 ns 1563.74 ns 1534.45 ns 1.02

Conditional Operators
name rxcpp rpp prev rpp ratio
immediate_just+take_while(false)+subscribe 1346.70 ns 11.45 ns 11.95 ns 0.96
immediate_just+take_while(true)+subscribe 1344.76 ns 12.37 ns 15.12 ns 0.82

Utility Operators
name rxcpp rpp prev rpp ratio
immediate_just(1)+subscribe_on(immediate)+subscribe 3904.44 ns 7.71 ns 7.71 ns 1.00

Combining Operators
name rxcpp rpp prev rpp ratio
immediate_just(immediate_just(1), immediate_just(1)) + merge() + subscribe 5185.42 ns 250.85 ns 254.20 ns 0.99
immediate_just(1) + merge_with(immediate_just(2)) + subscribe 6433.52 ns 235.92 ns 235.58 ns 1.00
immediate_just(1) + with_latest_from(immediate_just(2)) + subscribe - 230.54 ns 441.76 ns 0.52
immediate_just(immediate_just(1),immediate_just(1)) + switch_on_next() + subscribe 6205.41 ns 957.64 ns 1479.16 ns 0.65
immediate_just(1) + zip(immediate_just(2)) + subscribe 3880.07 ns 540.33 ns 821.52 ns 0.66

Subjects
name rxcpp rpp prev rpp ratio
publish_subject with 1 observer - on_next 36.11 ns 26.53 ns 26.51 ns 1.00

Scenarios
name rxcpp rpp prev rpp ratio
basic sample 1883.09 ns 59.12 ns 59.35 ns 1.00
basic sample with immediate scheduler 1891.44 ns 35.50 ns 35.48 ns 1.00

Aggregating Operators
name rxcpp rpp prev rpp ratio
immediate_just+reduce(10, std::plus)+subscribe 1475.72 ns 17.82 ns 17.82 ns 1.00

Error Handling Operators
name rxcpp rpp prev rpp ratio
create(on_next(1), on_error())+on_error_resume_next(immediate_just(2)))+subscribe 1967.70 ns 339.11 ns 333.40 ns 1.02

Please sign in to comment.