Commit c57316d1 authored by Dominik Charousset's avatar Dominik Charousset

Add streaming utility to test_coordinator_fixture

parent 6c35258b
...@@ -135,16 +135,7 @@ TESTEE(stream_multiplexer) { ...@@ -135,16 +135,7 @@ TESTEE(stream_multiplexer) {
}; };
} }
struct fixture : test_coordinator_fixture<> { using fixture = test_coordinator_fixture<>;
std::chrono::microseconds cycle;
fixture() : cycle(cfg.streaming_credit_round_interval_us) {
// Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(atom("batch"), timespan{1000});
// Make sure the current time isn't invalid.
sched.clock().current_time += cycle;
}
};
} // namespace <anonymous> } // namespace <anonymous>
...@@ -171,7 +162,7 @@ CAF_TEST(depth_3_pipeline_with_fork) { ...@@ -171,7 +162,7 @@ CAF_TEST(depth_3_pipeline_with_fork) {
auto predicate = [&] { auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean(); return st.stage->inbound_paths().empty() && st.stage->out().clean();
}; };
sched.run_dispatch_loop(predicate, cycle); sched.run_dispatch_loop(predicate, streaming_cycle);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, 1275); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, 1275);
...@@ -198,7 +189,7 @@ CAF_TEST(depth_3_pipeline_with_join) { ...@@ -198,7 +189,7 @@ CAF_TEST(depth_3_pipeline_with_join) {
auto predicate = [&] { auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean(); return st.stage->inbound_paths().empty() && st.stage->out().clean();
}; };
sched.run_dispatch_loop(predicate, cycle); sched.run_dispatch_loop(predicate, streaming_cycle);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 2550); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 2550);
......
...@@ -248,16 +248,7 @@ TESTEE(stream_multiplexer) { ...@@ -248,16 +248,7 @@ TESTEE(stream_multiplexer) {
}; };
} }
struct fixture : test_coordinator_fixture<> { using fixture = test_coordinator_fixture<>;
std::chrono::microseconds cycle;
fixture() : cycle(cfg.streaming_credit_round_interval_us) {
// Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(atom("batch"), timespan{1000});
// Make sure the current time isn't invalid.
sched.clock().current_time += cycle;
}
};
} // namespace <anonymous> } // namespace <anonymous>
...@@ -287,7 +278,7 @@ CAF_TEST(depth_3_pipeline_with_fork) { ...@@ -287,7 +278,7 @@ CAF_TEST(depth_3_pipeline_with_fork) {
auto predicate = [&] { auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean(); return st.stage->inbound_paths().empty() && st.stage->out().clean();
}; };
sched.run_dispatch_loop(predicate, cycle); sched.run_dispatch_loop(predicate, streaming_cycle);
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 0u);
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, 1275); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk1).state.x, 1275);
......
...@@ -211,16 +211,7 @@ TESTEE(doubler) { ...@@ -211,16 +211,7 @@ TESTEE(doubler) {
}; };
} }
struct fixture : test_coordinator_fixture<> { using fixture = test_coordinator_fixture<>;
std::chrono::microseconds cycle;
fixture() : cycle(cfg.streaming_credit_round_interval_us) {
// Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(atom("batch"), timespan{1000});
// Make sure the current time isn't invalid.
sched.clock().current_time += cycle;
}
};
} // namespace <anonymous> } // namespace <anonymous>
...@@ -239,7 +230,7 @@ CAF_TEST(depth_2_pipeline_50_items) { ...@@ -239,7 +230,7 @@ CAF_TEST(depth_2_pipeline_50_items) {
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
sched.clock().current_time += cycle; sched.clock().current_time += credit_round_interval;
sched.dispatch(); sched.dispatch();
expect((timeout_msg), from(snk).to(snk)); expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src)); expect((timeout_msg), from(src).to(src));
...@@ -260,7 +251,7 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) { ...@@ -260,7 +251,7 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) {
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
sched.clock().current_time += cycle; sched.clock().current_time += credit_round_interval;
sched.dispatch(); sched.dispatch();
expect((timeout_msg), from(snk).to(snk)); expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src)); expect((timeout_msg), from(src).to(src));
...@@ -287,7 +278,7 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) { ...@@ -287,7 +278,7 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) {
expect((upstream_msg::ack_open), from(snk).to(src)); expect((upstream_msg::ack_open), from(snk).to(src));
CAF_MESSAGE("start data transmission (a single batch)"); CAF_MESSAGE("start data transmission (a single batch)");
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
sched.clock().current_time += cycle; sched.clock().current_time += credit_round_interval;
sched.dispatch(); sched.dispatch();
expect((timeout_msg), from(snk).to(snk)); expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src)); expect((timeout_msg), from(src).to(src));
...@@ -313,7 +304,7 @@ CAF_TEST(depth_2_pipeline_500_items) { ...@@ -313,7 +304,7 @@ CAF_TEST(depth_2_pipeline_500_items) {
expect((downstream_msg::batch), from(src).to(snk)); expect((downstream_msg::batch), from(src).to(snk));
} }
CAF_MESSAGE("trigger timeouts"); CAF_MESSAGE("trigger timeouts");
sched.clock().current_time += cycle; sched.clock().current_time += credit_round_interval;
sched.dispatch(); sched.dispatch();
expect((timeout_msg), from(snk).to(snk)); expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(src).to(src)); expect((timeout_msg), from(src).to(src));
...@@ -377,7 +368,7 @@ CAF_TEST(depth_3_pipeline_50_items) { ...@@ -377,7 +368,7 @@ CAF_TEST(depth_3_pipeline_50_items) {
auto stg = sys.spawn(filter); auto stg = sys.spawn(filter);
auto snk = sys.spawn(sum_up); auto snk = sys.spawn(sum_up);
auto next_cycle = [&] { auto next_cycle = [&] {
sched.clock().current_time += cycle; sched.clock().current_time += credit_round_interval;
sched.dispatch(); sched.dispatch();
expect((timeout_msg), from(snk).to(snk)); expect((timeout_msg), from(snk).to(snk));
expect((timeout_msg), from(stg).to(stg)); expect((timeout_msg), from(stg).to(stg));
...@@ -424,7 +415,7 @@ CAF_TEST(depth_4_pipeline_500_items) { ...@@ -424,7 +415,7 @@ CAF_TEST(depth_4_pipeline_500_items) {
expect((upstream_msg::ack_open), from(stg2).to(stg1)); expect((upstream_msg::ack_open), from(stg2).to(stg1));
expect((upstream_msg::ack_open), from(stg1).to(src)); expect((upstream_msg::ack_open), from(stg1).to(src));
CAF_MESSAGE("start data transmission"); CAF_MESSAGE("start data transmission");
sched.run_dispatch_loop(cycle); sched.run_dispatch_loop(credit_round_interval);
CAF_MESSAGE("check sink result"); CAF_MESSAGE("check sink result");
CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125000); CAF_CHECK_EQUAL(deref<sum_up_actor>(snk).state.x, 125000);
} }
......
...@@ -173,16 +173,7 @@ TESTEE(log_consumer) { ...@@ -173,16 +173,7 @@ TESTEE(log_consumer) {
}; };
} }
struct fixture : test_coordinator_fixture<> { using fixture = test_coordinator_fixture<>;
std::chrono::microseconds cycle;
fixture() : cycle(cfg.streaming_credit_round_interval_us) {
// Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(atom("batch"), timespan{1000});
// Make sure the current time isn't invalid.
sched.clock().current_time += cycle;
}
};
} // namespace <anonymous> } // namespace <anonymous>
...@@ -196,7 +187,7 @@ CAF_TEST(select_all) { ...@@ -196,7 +187,7 @@ CAF_TEST(select_all) {
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk)); CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, level::all); self->send(snk * src, level::all);
sched.run_dispatch_loop(cycle); sched.run_dispatch_loop(streaming_cycle);
CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk).state.log, CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk).state.log,
make_log(level::all)); make_log(level::all));
} }
...@@ -207,7 +198,7 @@ CAF_TEST(select_trace) { ...@@ -207,7 +198,7 @@ CAF_TEST(select_trace) {
CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk)); CAF_MESSAGE(CAF_ARG(self) << CAF_ARG(src) << CAF_ARG(snk));
CAF_MESSAGE("initiate stream handshake"); CAF_MESSAGE("initiate stream handshake");
self->send(snk * src, level::trace); self->send(snk * src, level::trace);
sched.run_dispatch_loop(cycle); sched.run_dispatch_loop(streaming_cycle);
CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk).state.log, CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk).state.log,
make_log(level::trace)); make_log(level::trace));
} }
...@@ -226,7 +217,7 @@ CAF_TEST(forking) { ...@@ -226,7 +217,7 @@ CAF_TEST(forking) {
auto predicate = [&] { auto predicate = [&] {
return st.stage->inbound_paths().empty() && st.stage->out().clean(); return st.stage->inbound_paths().empty() && st.stage->out().clean();
}; };
sched.run_dispatch_loop(predicate, cycle); sched.run_dispatch_loop(predicate, streaming_cycle);
CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk1).state.log, CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk1).state.log,
make_log(level::trace)); make_log(level::trace));
CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk2).state.log, CAF_CHECK_EQUAL(deref<log_consumer_actor>(snk2).state.log,
......
...@@ -20,6 +20,8 @@ ...@@ -20,6 +20,8 @@
#include "caf/meta/annotation.hpp" #include "caf/meta/annotation.hpp"
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include "caf/detail/gcd.hpp"
namespace { namespace {
struct wildcard { }; struct wildcard { };
...@@ -421,24 +423,57 @@ protected: ...@@ -421,24 +423,57 @@ protected:
std::function<void ()> check_; std::function<void ()> check_;
}; };
/// A fixture with a deterministic scheduler setup.
template <class Config = caf::actor_system_config> template <class Config = caf::actor_system_config>
struct test_coordinator_fixture { struct test_coordinator_fixture {
/// A deterministic scheduler type.
using scheduler_type = caf::scheduler::test_coordinator; using scheduler_type = caf::scheduler::test_coordinator;
/// Convenience alias for std::chrono::microseconds.
using us_t = std::chrono::microseconds;
/// The user-generated system config.
Config cfg; Config cfg;
/// Host system for (scheduled) actors.
caf::actor_system sys; caf::actor_system sys;
/// A scoped actor for conveniently sending and receiving messages.
caf::scoped_actor self; caf::scoped_actor self;
/// Deterministic scheduler.
scheduler_type& sched; scheduler_type& sched;
/// Duration between two credit rounds.
caf::timespan credit_round_interval;
/// Max send delay for stream batches.
caf::timespan max_batch_delay;
/// Duration a single cycle, computed as GCD of credit-round-interval and
/// max-batch-delay. Using this duration for `sched.run_dispatch_loop()`
/// advances the clock in ideal steps.
caf::timespan streaming_cycle;
test_coordinator_fixture() test_coordinator_fixture()
: sys(cfg.parse(caf::test::engine::argc(), caf::test::engine::argv()) : sys(cfg.parse(caf::test::engine::argc(), caf::test::engine::argv())
.set("scheduler.policy", caf::atom("testing"))), .set("scheduler.policy", caf::atom("testing"))),
self(sys), self(sys, true),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())) { sched(dynamic_cast<scheduler_type&>(sys.scheduler())),
// nop credit_round_interval(us_t{cfg.streaming_credit_round_interval_us}),
} max_batch_delay(us_t{cfg.streaming_max_batch_delay_us}) {
// Configure the clock to measure each batch item with 1us.
~test_coordinator_fixture() { sched.clock().time_per_unit.emplace(caf::atom("batch"),
caf::timespan{1000});
// Compute reasonable step size.
auto cycle_count = caf::detail::gcd(credit_round_interval.count(),
max_batch_delay.count());
streaming_cycle = caf::timespan{cycle_count};
// Make sure the current time isn't 0.
sched.clock().current_time += streaming_cycle;
}
virtual ~test_coordinator_fixture() {
sched.clock().cancel_all(); sched.clock().cancel_all();
sched.run(); sched.run();
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment