Commit 78a49dfa authored by Dominik Charousset's avatar Dominik Charousset

Revise testing DSL design

parent 357f8b73
...@@ -34,16 +34,25 @@ public: ...@@ -34,16 +34,25 @@ public:
duration_type difference(atom_value measurement, long units, time_point t0, duration_type difference(atom_value measurement, long units, time_point t0,
time_point t1) const noexcept override; time_point t1) const noexcept override;
/// Tries to dispatch the next timeout or delayed message regardless of its /// Triggers the next pending timeout regardless of its timestamp. Sets
/// timestamp. Returns `false` if `schedule().empty()`, otherwise `true`. /// `current_time` to the time point of the triggered timeout unless
bool dispatch_once(); /// `current_time` is already set to a later time.
/// @returns Whether a timeout was triggered.
bool trigger_timeout();
/// Dispatches all timeouts and delayed messages regardless of their /// Triggers all pending timeouts regardless of their timestamp. Sets
/// timestamp. Returns the number of dispatched events. /// `current_time` to the time point of the latest timeout unless
size_t dispatch(); /// `current_time` is already set to a later time.
/// @returns The number of triggered timeouts.
size_t trigger_timeouts();
/// Triggers all timeouts with timestamp <= now.
/// @returns The number of triggered timeouts.
size_t trigger_expired_timeouts();
/// Advances the time by `x` and dispatches timeouts and delayed messages. /// Advances the time by `x` and dispatches timeouts and delayed messages.
void advance_time(duration_type x); /// @returns The number of triggered timeouts.
size_t advance_time(duration_type x);
/// Configures the returned value for `difference`. For example, inserting /// Configures the returned value for `difference`. For example, inserting
/// `('foo', 120ns)` causes the clock to return `units * 120ns` for any call /// `('foo', 120ns)` causes the clock to return `units * 120ns` for any call
......
...@@ -38,14 +38,14 @@ class test_coordinator : public abstract_coordinator { ...@@ -38,14 +38,14 @@ class test_coordinator : public abstract_coordinator {
public: public:
using super = abstract_coordinator; using super = abstract_coordinator;
/// A type-erased boolean predicate.
using bool_predicate = std::function<bool()>;
test_coordinator(actor_system& sys); test_coordinator(actor_system& sys);
/// A double-ended queue representing our current job queue. /// A double-ended queue representing our current job queue.
std::deque<resumable*> jobs; std::deque<resumable*> jobs;
/// A clock type using the highest available precision.
using hrc = std::chrono::high_resolution_clock;
/// Returns whether at least one job is in the queue. /// Returns whether at least one job is in the queue.
inline bool has_job() const { inline bool has_job() const {
return !jobs.empty(); return !jobs.empty();
...@@ -105,25 +105,37 @@ public: ...@@ -105,25 +105,37 @@ public:
/// left. Returns the number of processed events. /// left. Returns the number of processed events.
size_t run(size_t max_count = std::numeric_limits<size_t>::max()); size_t run(size_t max_count = std::numeric_limits<size_t>::max());
/// Tries to dispatch a single delayed message. /// Tries to trigger a single timeout.
bool dispatch_once(); bool trigger_timeout() {
return clock_.trigger_timeout();
}
/// Dispatches all pending delayed messages. Returns the number of dispatched /// Triggers all pending timeouts.
/// messages. size_t trigger_timeouts() {
size_t dispatch(); return clock_.trigger_timeouts();
}
/// Loops until no job or delayed message remains or `predicate` returns /// Advances simulation time and returns the number of triggered timeouts.
/// `true`. Returns the total number of events (first) and dispatched delayed size_t advance_time(timespan x) {
return clock_.advance_time(x);
}
/// Runs all events, then advances the time by `cycle_duration` and triggers
/// all timeouts, then repeats until no activity remains.
/// @param predicate Stop condition.
/// @param cycle_duration
/// @returns The number of scheduling events (first) and the number of
/// triggered timeouts (second).
/// messages (second). Advances time by `cycle` nanoseconds between to calls /// messages (second). Advances time by `cycle` nanoseconds between to calls
/// to `dispatch()` or the default tick-duration when passing 0ns. /// to `dispatch()` or the default tick-duration when passing 0ns.
std::pair<size_t, size_t> run_dispatch_loop(std::function<bool()> predicate, std::pair<size_t, size_t> run_cycle_until(bool_predicate predicate,
timespan cycle = timespan{0}); timespan cycle_duration);
/// Loops until no job or delayed message remains. Returns the total number /// Loops until no job or delayed message remains. Returns the total number
/// of events (first) and dispatched delayed messages (second). Advances time /// of events (first) and dispatched delayed messages (second). Advances time
/// by `cycle` nanoseconds between to calls to `dispatch()` or the default /// by `cycle` nanoseconds between to calls to `dispatch()` or the default
/// tick-duration when passing 0ns. /// tick-duration when passing 0ns.
std::pair<size_t, size_t> run_dispatch_loop(timespan cycle = timespan{0}); std::pair<size_t, size_t> run_cycle(timespan cycle_duration = timespan{1});
template <class F> template <class F>
void after_next_enqueue(F f) { void after_next_enqueue(F f) {
...@@ -142,6 +154,10 @@ public: ...@@ -142,6 +154,10 @@ public:
detail::test_actor_clock& clock() noexcept override; detail::test_actor_clock& clock() noexcept override;
std::pair<size_t, size_t>
run_dispatch_loop(timespan cycle_duration = timespan{1})
CAF_DEPRECATED_MSG("use run_cycle() instead");
protected: protected:
void start() override; void start() override;
...@@ -161,5 +177,3 @@ private: ...@@ -161,5 +177,3 @@ private:
} // namespace scheduler } // namespace scheduler
} // namespace caf } // namespace caf
...@@ -18,6 +18,8 @@ ...@@ -18,6 +18,8 @@
#include "caf/detail/test_actor_clock.hpp" #include "caf/detail/test_actor_clock.hpp"
#include "caf/logger.hpp"
namespace caf { namespace caf {
namespace detail { namespace detail {
...@@ -40,35 +42,54 @@ test_actor_clock::difference(atom_value measurement, long units, time_point t0, ...@@ -40,35 +42,54 @@ test_actor_clock::difference(atom_value measurement, long units, time_point t0,
return t0 == t1 ? duration_type{1} : t1 - t0; return t0 == t1 ? duration_type{1} : t1 - t0;
} }
bool test_actor_clock::dispatch_once() { bool test_actor_clock::trigger_timeout() {
CAF_LOG_TRACE(CAF_ARG2("schedule.size", schedule_.size()));
if (schedule_.empty()) if (schedule_.empty())
return false; return false;
visitor f{this}; visitor f{this};
auto i = schedule_.begin(); auto i = schedule_.begin();
auto tout = i->first;
if (tout > current_time)
current_time = tout;
visit(f, i->second); visit(f, i->second);
schedule_.erase(i); schedule_.erase(i);
return true; return true;
} }
size_t test_actor_clock::dispatch() { size_t test_actor_clock::trigger_timeouts() {
CAF_LOG_TRACE(CAF_ARG2("schedule.size", schedule_.size()));
if (schedule_.empty()) if (schedule_.empty())
return 0u; return 0u;
visitor f{this}; visitor f{this};
auto result = schedule_.size(); auto result = schedule_.size();
for (auto& kvp : schedule_) for (auto& kvp : schedule_) {
auto tout = kvp.first;
if (tout > current_time)
current_time = tout;
visit(f, kvp.second); visit(f, kvp.second);
}
schedule_.clear(); schedule_.clear();
return result; return result;
} }
void test_actor_clock::advance_time(duration_type x) { size_t test_actor_clock::trigger_expired_timeouts() {
CAF_LOG_TRACE(CAF_ARG2("schedule.size", schedule_.size()));
visitor f{this}; visitor f{this};
current_time += x; size_t result = 0;
auto i = schedule_.begin(); auto i = schedule_.begin();
while (i != schedule_.end() && i->first <= current_time) { while (i != schedule_.end() && i->first <= current_time) {
++result;
visit(f, i->second); visit(f, i->second);
i = schedule_.erase(i); i = schedule_.erase(i);
} }
return result;
}
size_t test_actor_clock::advance_time(duration_type x) {
CAF_LOG_TRACE(CAF_ARG(x) << CAF_ARG2("schedule.size", schedule_.size()));
CAF_ASSERT(x.count() >= 0);
current_time += x;
return trigger_expired_timeouts();
} }
} // namespace detail } // namespace detail
......
...@@ -87,7 +87,7 @@ void test_coordinator::start() { ...@@ -87,7 +87,7 @@ void test_coordinator::start() {
} }
void test_coordinator::stop() { void test_coordinator::stop() {
run_dispatch_loop(); run_cycle();
} }
void test_coordinator::enqueue(resumable* ptr) { void test_coordinator::enqueue(resumable* ptr) {
...@@ -147,44 +147,53 @@ size_t test_coordinator::run(size_t max_count) { ...@@ -147,44 +147,53 @@ size_t test_coordinator::run(size_t max_count) {
return res; return res;
} }
bool test_coordinator::dispatch_once() {
return clock().dispatch_once();
}
size_t test_coordinator::dispatch() {
return clock().dispatch();
}
std::pair<size_t, size_t> std::pair<size_t, size_t>
test_coordinator::run_dispatch_loop(std::function<bool()> predicate, test_coordinator::run_cycle_until(bool_predicate predicate,
timespan cycle) { timespan cycle_duration) {
std::pair<size_t, size_t> res{0, 0}; CAF_LOG_TRACE(CAF_ARG(cycle_duration)
if (cycle.count() == 0) { << CAF_ARG2("pending_jobs", jobs.size())
auto x = system().config().streaming_tick_duration_us(); << CAF_ARG2("pending_timeouts", clock_.schedule().size()));
cycle = std::chrono::microseconds(x); // Bookkeeping.
} size_t events = 0;
for (;;) { size_t timeouts = 0;
size_t progress = 0; // Loop until no activity remains.
while (!jobs.empty() || !clock_.schedule().empty()) {
while (try_run_once()) { while (try_run_once()) {
++progress; ++events;
res.first += 1; if (predicate()) {
if (predicate()) CAF_LOG_DEBUG("stop due to predicate;"
return res; << CAF_ARG(events) << CAF_ARG(timeouts));
return {events, timeouts};
}
} }
clock().current_time += cycle; if (!clock_.schedule().empty()) {
while (dispatch_once()) { size_t triggered = 0;
++progress; auto next_tout = clock_.schedule().begin()->first;
res.second += 1; if (next_tout <= clock_.now()) {
if (predicate()) // Trigger expired timeout first without advancing time.
return res; triggered = clock_.trigger_expired_timeouts();
} else {
// Compute how much cycles we need to trigger at least one timeout.
auto diff = next_tout - clock_.current_time;
auto num_cycles = diff.count() / cycle_duration.count();
auto advancement = num_cycles * cycle_duration;
// `num_cyles` can have one to few cycles since we's always rounding down.
if (clock_.current_time + advancement < next_tout)
advancement += cycle_duration;
// Advance time.
CAF_ASSERT(advancement.count() > 0);
triggered = clock_.advance_time(advancement);
}
CAF_ASSERT(triggered > 0);
timeouts += triggered;
} }
if (progress == 0)
return res;
} }
CAF_LOG_DEBUG("no activity left;" << CAF_ARG(events) << CAF_ARG(timeouts));
return {events, timeouts};
} }
std::pair<size_t, size_t> test_coordinator::run_dispatch_loop(timespan cycle) { std::pair<size_t, size_t> test_coordinator::run_cycle(timespan cycle_duration) {
return run_dispatch_loop([] { return false; }, cycle); return run_cycle_until([] { return false; }, cycle_duration);
} }
void test_coordinator::inline_next_enqueue() { void test_coordinator::inline_next_enqueue() {
...@@ -200,6 +209,11 @@ void test_coordinator::inline_all_enqueues_helper() { ...@@ -200,6 +209,11 @@ void test_coordinator::inline_all_enqueues_helper() {
after_next_enqueue([=] { inline_all_enqueues_helper(); }); after_next_enqueue([=] { inline_all_enqueues_helper(); });
} }
std::pair<size_t, size_t>
test_coordinator::run_dispatch_loop(timespan cycle_duration) {
return run_cycle(cycle_duration);
}
} // namespace caf } // namespace caf
} // namespace scheduler } // namespace scheduler
...@@ -56,7 +56,8 @@ void tick_emitter::interval(duration_type x) { ...@@ -56,7 +56,8 @@ void tick_emitter::interval(duration_type x) {
size_t tick_emitter::timeouts(time_point now, size_t tick_emitter::timeouts(time_point now,
std::initializer_list<size_t> periods) { std::initializer_list<size_t> periods) {
CAF_LOG_TRACE(CAF_ARG(now) << CAF_ARG(periods)); CAF_LOG_TRACE(CAF_ARG(now) << CAF_ARG(periods) << CAF_ARG(interval_) << CAF_ARG(start_));
CAF_ASSERT(now >= start_);
size_t result = 0; size_t result = 0;
auto f = [&](size_t tick) { auto f = [&](size_t tick) {
size_t n = 0; size_t n = 0;
......
...@@ -166,14 +166,14 @@ CAF_TEST(depth_3_pipeline_with_fork) { ...@@ -166,14 +166,14 @@ CAF_TEST(depth_3_pipeline_with_fork) {
CAF_MESSAGE("connect sinks to the stage (fork)"); CAF_MESSAGE("connect sinks to the stage (fork)");
self->send(snk1, join_atom::value, stg); self->send(snk1, join_atom::value, stg);
self->send(snk2, join_atom::value, stg); self->send(snk2, join_atom::value, stg);
sched.run(); consume_messages();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_MESSAGE("connect source to the stage (fork)"); CAF_MESSAGE("connect source to the stage (fork)");
self->send(stg * src, "numbers.txt"); self->send(stg * src, "numbers.txt");
sched.run(); consume_messages();
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(), 1u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
run_exhaustively(); run();
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);
...@@ -189,15 +189,15 @@ CAF_TEST(depth_3_pipeline_with_join) { ...@@ -189,15 +189,15 @@ CAF_TEST(depth_3_pipeline_with_join) {
auto& st = deref<stream_multiplexer_actor>(stg).state; auto& st = deref<stream_multiplexer_actor>(stg).state;
CAF_MESSAGE("connect sink to the stage"); CAF_MESSAGE("connect sink to the stage");
self->send(snk, join_atom::value, stg); self->send(snk, join_atom::value, stg);
sched.run(); consume_messages();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 1u);
CAF_MESSAGE("connect sources to the stage (join)"); CAF_MESSAGE("connect sources to the stage (join)");
self->send(stg * src1, "numbers.txt"); self->send(stg * src1, "numbers.txt");
self->send(stg * src2, "numbers.txt"); self->send(stg * src2, "numbers.txt");
sched.run(); consume_messages();
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(), 2u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 2u);
run_exhaustively(); run();
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);
...@@ -213,17 +213,16 @@ CAF_TEST(closing_downstreams_before_end_of_stream) { ...@@ -213,17 +213,16 @@ CAF_TEST(closing_downstreams_before_end_of_stream) {
CAF_MESSAGE("connect sinks to the stage (fork)"); CAF_MESSAGE("connect sinks to the stage (fork)");
self->send(snk1, join_atom::value, stg); self->send(snk1, join_atom::value, stg);
self->send(snk2, join_atom::value, stg); self->send(snk2, join_atom::value, stg);
sched.run(); consume_messages();
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
CAF_MESSAGE("connect source to the stage (fork)"); CAF_MESSAGE("connect source to the stage (fork)");
self->send(stg * src, "numbers.txt"); self->send(stg * src, "numbers.txt");
sched.run(); consume_messages();
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(), 1u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 1u);
CAF_MESSAGE("do a single round of credit"); CAF_MESSAGE("do a single round of credit");
sched.clock().current_time += streaming_cycle; tick();
sched.dispatch(); consume_messages();
sched.run();
CAF_MESSAGE("make sure the stream isn't done yet"); CAF_MESSAGE("make sure the stream isn't done yet");
CAF_REQUIRE(!deref<file_reader_actor>(src).state.buf.empty()); CAF_REQUIRE(!deref<file_reader_actor>(src).state.buf.empty());
CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u); CAF_CHECK_EQUAL(st.stage->out().num_paths(), 2u);
...@@ -236,7 +235,7 @@ CAF_TEST(closing_downstreams_before_end_of_stream) { ...@@ -236,7 +235,7 @@ CAF_TEST(closing_downstreams_before_end_of_stream) {
self->send(stg, close_atom::value, 0); self->send(stg, close_atom::value, 0);
expect((atom_value, int), from(self).to(stg)); expect((atom_value, int), from(self).to(stg));
CAF_MESSAGE("ship remaining elements"); CAF_MESSAGE("ship remaining elements");
run_exhaustively(); run();
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_LESS(deref<sum_up_actor>(snk1).state.x, sink1_result); CAF_CHECK_LESS(deref<sum_up_actor>(snk1).state.x, sink1_result);
......
...@@ -279,10 +279,9 @@ CAF_TEST(depth_3_pipeline_with_fork) { ...@@ -279,10 +279,9 @@ CAF_TEST(depth_3_pipeline_with_fork) {
sched.run(); sched.run();
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(), 2u); CAF_CHECK_EQUAL(st.stage->inbound_paths().size(), 2u);
auto predicate = [&] { run_until([&] {
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, 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,7 +211,11 @@ TESTEE(doubler) { ...@@ -211,7 +211,11 @@ TESTEE(doubler) {
}; };
} }
using fixture = test_coordinator_fixture<>; struct fixture : test_coordinator_fixture<> {
timespan tick_duration() const override {
return credit_round_interval;
}
};
} // namespace <anonymous> } // namespace <anonymous>
...@@ -230,8 +234,7 @@ CAF_TEST(depth_2_pipeline_50_items) { ...@@ -230,8 +234,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 += credit_round_interval; tick();
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));
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
...@@ -251,8 +254,7 @@ CAF_TEST(depth_2_pipeline_setup2_50_items) { ...@@ -251,8 +254,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 += credit_round_interval; tick();
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));
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
...@@ -278,8 +280,7 @@ CAF_TEST(delayed_depth_2_pipeline_50_items) { ...@@ -278,8 +280,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 += credit_round_interval; tick();
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));
expect((upstream_msg::ack_batch), from(snk).to(src)); expect((upstream_msg::ack_batch), from(snk).to(src));
...@@ -304,8 +305,7 @@ CAF_TEST(depth_2_pipeline_500_items) { ...@@ -304,8 +305,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 += credit_round_interval; tick();
sched.dispatch();
allow((timeout_msg), from(snk).to(snk)); allow((timeout_msg), from(snk).to(snk));
allow((timeout_msg), from(src).to(src)); allow((timeout_msg), from(src).to(src));
CAF_MESSAGE("process ack_batch in source"); CAF_MESSAGE("process ack_batch in source");
...@@ -368,8 +368,7 @@ CAF_TEST(depth_3_pipeline_50_items) { ...@@ -368,8 +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 += credit_round_interval; tick();
sched.dispatch();
allow((timeout_msg), from(snk).to(snk)); allow((timeout_msg), from(snk).to(snk));
allow((timeout_msg), from(stg).to(stg)); allow((timeout_msg), from(stg).to(stg));
allow((timeout_msg), from(src).to(src)); allow((timeout_msg), from(src).to(src));
...@@ -415,7 +414,7 @@ CAF_TEST(depth_4_pipeline_500_items) { ...@@ -415,7 +414,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(credit_round_interval); run();
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);
} }
......
...@@ -16,12 +16,12 @@ ...@@ -16,12 +16,12 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/config.hpp" #include "caf/after.hpp"
#define CAF_SUITE request_timeout #define CAF_SUITE request_timeout
#include "caf/test/unit_test.hpp"
#include <thread> #include "caf/test/dsl.hpp"
#include <chrono> #include <chrono>
#include "caf/all.hpp" #include "caf/all.hpp"
...@@ -70,7 +70,7 @@ behavior ping_single1(ping_actor* self, bool* had_timeout, const actor& buddy) { ...@@ -70,7 +70,7 @@ behavior ping_single1(ping_actor* self, bool* had_timeout, const actor& buddy) {
self->delayed_send(self, std::chrono::seconds(1), timeout_atom::value); self->delayed_send(self, std::chrono::seconds(1), timeout_atom::value);
return { return {
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](timeout_atom) { [=](timeout_atom) {
*had_timeout = true; *had_timeout = true;
...@@ -84,7 +84,7 @@ behavior ping_single2(ping_actor* self, bool* had_timeout, const actor& buddy) { ...@@ -84,7 +84,7 @@ behavior ping_single2(ping_actor* self, bool* had_timeout, const actor& buddy) {
self->send(buddy, ping_atom::value); self->send(buddy, ping_atom::value);
return { return {
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
after(std::chrono::seconds(1)) >> [=] { after(std::chrono::seconds(1)) >> [=] {
*had_timeout = true; *had_timeout = true;
...@@ -97,7 +97,7 @@ behavior ping_single2(ping_actor* self, bool* had_timeout, const actor& buddy) { ...@@ -97,7 +97,7 @@ behavior ping_single2(ping_actor* self, bool* had_timeout, const actor& buddy) {
behavior ping_single3(ping_actor* self, bool* had_timeout, const actor& buddy) { behavior ping_single3(ping_actor* self, bool* had_timeout, const actor& buddy) {
self->request(buddy, milliseconds(100), ping_atom::value).then( self->request(buddy, milliseconds(100), ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](const error& err) { [=](const error& err) {
CAF_REQUIRE(err == sec::request_timeout); CAF_REQUIRE(err == sec::request_timeout);
...@@ -115,7 +115,7 @@ behavior ping_nested1(ping_actor* self, bool* had_timeout, ...@@ -115,7 +115,7 @@ behavior ping_nested1(ping_actor* self, bool* had_timeout,
self->delayed_send(self, std::chrono::seconds(1), timeout_atom::value); self->delayed_send(self, std::chrono::seconds(1), timeout_atom::value);
return { return {
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](timeout_atom) { [=](timeout_atom) {
self->state.had_first_timeout = true; self->state.had_first_timeout = true;
...@@ -136,7 +136,7 @@ behavior ping_nested2(ping_actor* self, bool* had_timeout, const actor& buddy) { ...@@ -136,7 +136,7 @@ behavior ping_nested2(ping_actor* self, bool* had_timeout, const actor& buddy) {
self->send(buddy, ping_atom::value); self->send(buddy, ping_atom::value);
return { return {
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
after(std::chrono::seconds(1)) >> [=] { after(std::chrono::seconds(1)) >> [=] {
self->state.had_first_timeout = true; self->state.had_first_timeout = true;
...@@ -156,7 +156,7 @@ behavior ping_nested2(ping_actor* self, bool* had_timeout, const actor& buddy) { ...@@ -156,7 +156,7 @@ behavior ping_nested2(ping_actor* self, bool* had_timeout, const actor& buddy) {
behavior ping_nested3(ping_actor* self, bool* had_timeout, const actor& buddy) { behavior ping_nested3(ping_actor* self, bool* had_timeout, const actor& buddy) {
self->request(buddy, milliseconds(100), ping_atom::value).then( self->request(buddy, milliseconds(100), ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
self->quit(sec::unexpected_message); self->quit(sec::unexpected_message);
}, },
[=](const error& err) { [=](const error& err) {
...@@ -178,7 +178,7 @@ behavior ping_multiplexed1(ping_actor* self, bool* had_timeout, ...@@ -178,7 +178,7 @@ behavior ping_multiplexed1(ping_actor* self, bool* had_timeout,
const actor& pong_actor) { const actor& pong_actor) {
self->request(pong_actor, milliseconds(100), ping_atom::value).then( self->request(pong_actor, milliseconds(100), ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](const error& err) { [=](const error& err) {
CAF_REQUIRE_EQUAL(err, sec::request_timeout); CAF_REQUIRE_EQUAL(err, sec::request_timeout);
...@@ -190,7 +190,7 @@ behavior ping_multiplexed1(ping_actor* self, bool* had_timeout, ...@@ -190,7 +190,7 @@ behavior ping_multiplexed1(ping_actor* self, bool* had_timeout,
); );
self->request(pong_actor, milliseconds(100), ping_atom::value).then( self->request(pong_actor, milliseconds(100), ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](const error& err) { [=](const error& err) {
CAF_REQUIRE_EQUAL(err, sec::request_timeout); CAF_REQUIRE_EQUAL(err, sec::request_timeout);
...@@ -208,7 +208,7 @@ behavior ping_multiplexed2(ping_actor* self, bool* had_timeout, ...@@ -208,7 +208,7 @@ behavior ping_multiplexed2(ping_actor* self, bool* had_timeout,
const actor& pong_actor) { const actor& pong_actor) {
self->request(pong_actor, milliseconds(100), ping_atom::value).await( self->request(pong_actor, milliseconds(100), ping_atom::value).await(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](const error& err) { [=](const error& err) {
CAF_REQUIRE_EQUAL(err, sec::request_timeout); CAF_REQUIRE_EQUAL(err, sec::request_timeout);
...@@ -220,7 +220,7 @@ behavior ping_multiplexed2(ping_actor* self, bool* had_timeout, ...@@ -220,7 +220,7 @@ behavior ping_multiplexed2(ping_actor* self, bool* had_timeout,
); );
self->request(pong_actor, milliseconds(100), ping_atom::value).await( self->request(pong_actor, milliseconds(100), ping_atom::value).await(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](const error& err) { [=](const error& err) {
CAF_REQUIRE_EQUAL(err, sec::request_timeout); CAF_REQUIRE_EQUAL(err, sec::request_timeout);
...@@ -238,7 +238,7 @@ behavior ping_multiplexed3(ping_actor* self, bool* had_timeout, ...@@ -238,7 +238,7 @@ behavior ping_multiplexed3(ping_actor* self, bool* had_timeout,
const actor& pong_actor) { const actor& pong_actor) {
self->request(pong_actor, milliseconds(100), ping_atom::value).then( self->request(pong_actor, milliseconds(100), ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](const error& err) { [=](const error& err) {
CAF_REQUIRE_EQUAL(err, sec::request_timeout); CAF_REQUIRE_EQUAL(err, sec::request_timeout);
...@@ -250,7 +250,7 @@ behavior ping_multiplexed3(ping_actor* self, bool* had_timeout, ...@@ -250,7 +250,7 @@ behavior ping_multiplexed3(ping_actor* self, bool* had_timeout,
); );
self->request(pong_actor, milliseconds(100), ping_atom::value).await( self->request(pong_actor, milliseconds(100), ping_atom::value).await(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_FAIL("received pong atom");
}, },
[=](const error& err) { [=](const error& err) {
CAF_REQUIRE_EQUAL(err, sec::request_timeout); CAF_REQUIRE_EQUAL(err, sec::request_timeout);
...@@ -263,33 +263,9 @@ behavior ping_multiplexed3(ping_actor* self, bool* had_timeout, ...@@ -263,33 +263,9 @@ behavior ping_multiplexed3(ping_actor* self, bool* had_timeout,
return {}; return {};
} }
struct config : actor_system_config {
config() {
set("scheduler.policy", atom("testing"));
}
};
struct fixture {
config cfg;
actor_system system;
scoped_actor self;
scheduler::test_coordinator& sched;
fixture()
: system(cfg),
self(system),
sched(dynamic_cast<scheduler::test_coordinator&>(system.scheduler())) {
CAF_REQUIRE(sched.jobs.empty());
}
~fixture() {
sched.run_dispatch_loop();
}
};
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(request_timeout_tests, fixture) CAF_TEST_FIXTURE_SCOPE(request_timeout_tests, test_coordinator_fixture<>)
CAF_TEST(single_timeout) { CAF_TEST(single_timeout) {
test_vec fs{{ping_single1, "ping_single1"}, test_vec fs{{ping_single1, "ping_single1"},
...@@ -298,14 +274,14 @@ CAF_TEST(single_timeout) { ...@@ -298,14 +274,14 @@ CAF_TEST(single_timeout) {
for (auto f : fs) { for (auto f : fs) {
bool had_timeout = false; bool had_timeout = false;
CAF_MESSAGE("test implemenation " << f.second); CAF_MESSAGE("test implemenation " << f.second);
auto testee = system.spawn(f.first, &had_timeout, auto testee = sys.spawn(f.first, &had_timeout,
system.spawn<lazy_init>(pong)); sys.spawn<lazy_init>(pong));
CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u);
CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"}); CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"});
sched.run_once(); sched.run_once();
CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u);
CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"pong"}); CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"pong"});
sched.dispatch(); sched.trigger_timeout();
CAF_REQUIRE_EQUAL(sched.jobs.size(), 2u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 2u);
// now, the timeout message is already dispatched, while pong did // now, the timeout message is already dispatched, while pong did
// not respond to the message yet, i.e., timeout arrives before response // not respond to the message yet, i.e., timeout arrives before response
...@@ -321,20 +297,20 @@ CAF_TEST(nested_timeout) { ...@@ -321,20 +297,20 @@ CAF_TEST(nested_timeout) {
for (auto f : fs) { for (auto f : fs) {
bool had_timeout = false; bool had_timeout = false;
CAF_MESSAGE("test implemenation " << f.second); CAF_MESSAGE("test implemenation " << f.second);
auto testee = system.spawn(f.first, &had_timeout, auto testee = sys.spawn(f.first, &had_timeout,
system.spawn<lazy_init>(pong)); sys.spawn<lazy_init>(pong));
CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u);
CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"}); CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"});
sched.run_once(); sched.run_once();
CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u);
CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"pong"}); CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"pong"});
sched.dispatch(); sched.trigger_timeout();
CAF_REQUIRE_EQUAL(sched.jobs.size(), 2u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 2u);
// now, the timeout message is already dispatched, while pong did // now, the timeout message is already dispatched, while pong did
// not respond to the message yet, i.e., timeout arrives before response // not respond to the message yet, i.e., timeout arrives before response
sched.run(); sched.run();
// dispatch second timeout // dispatch second timeout
CAF_REQUIRE_EQUAL(sched.dispatch(), true); CAF_REQUIRE_EQUAL(sched.trigger_timeout(), true);
CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"}); CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"});
CAF_CHECK(!had_timeout); CAF_CHECK(!had_timeout);
CAF_CHECK(sched.next_job<ping_actor>().state.had_first_timeout); CAF_CHECK(sched.next_job<ping_actor>().state.had_first_timeout);
...@@ -350,14 +326,14 @@ CAF_TEST(multiplexed_timeout) { ...@@ -350,14 +326,14 @@ CAF_TEST(multiplexed_timeout) {
for (auto f : fs) { for (auto f : fs) {
bool had_timeout = false; bool had_timeout = false;
CAF_MESSAGE("test implemenation " << f.second); CAF_MESSAGE("test implemenation " << f.second);
auto testee = system.spawn(f.first, &had_timeout, auto testee = sys.spawn(f.first, &had_timeout,
system.spawn<lazy_init>(pong)); sys.spawn<lazy_init>(pong));
CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u);
CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"}); CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"});
sched.run_once(); sched.run_once();
CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 1u);
CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"pong"}); CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"pong"});
sched.dispatch(); sched.trigger_timeouts();
CAF_REQUIRE_EQUAL(sched.jobs.size(), 2u); CAF_REQUIRE_EQUAL(sched.jobs.size(), 2u);
// now, the timeout message is already dispatched, while pong did // now, the timeout message is already dispatched, while pong did
// not respond to the message yet, i.e., timeout arrives before response // not respond to the message yet, i.e., timeout arrives before response
......
...@@ -186,7 +186,7 @@ CAF_TEST(select_all) { ...@@ -186,7 +186,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(streaming_cycle); run();
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));
} }
...@@ -197,7 +197,7 @@ CAF_TEST(select_trace) { ...@@ -197,7 +197,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(streaming_cycle); run();
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));
} }
...@@ -213,10 +213,9 @@ CAF_TEST(forking) { ...@@ -213,10 +213,9 @@ CAF_TEST(forking) {
self->send(snk2 * stg, join_atom::value, level::error); self->send(snk2 * stg, join_atom::value, level::error);
sched.run(); sched.run();
auto& st = deref<log_dispatcher_actor>(stg).state; auto& st = deref<log_dispatcher_actor>(stg).state;
auto predicate = [&] { run_until([&] {
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, 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,
......
...@@ -16,10 +16,11 @@ ...@@ -16,10 +16,11 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/config.hpp" #include "caf/after.hpp"
#define CAF_SUITE simple_timeout #define CAF_SUITE simple_timeout
#include "caf/test/unit_test.hpp"
#include "caf/test/dsl.hpp"
#include <chrono> #include <chrono>
#include <memory> #include <memory>
...@@ -70,33 +71,9 @@ timer::behavior_type timer_impl2(timer::pointer self) { ...@@ -70,33 +71,9 @@ timer::behavior_type timer_impl2(timer::pointer self) {
}; };
} }
struct config : actor_system_config {
config() {
set("scheduler.policy", atom("testing"));
}
};
struct fixture {
config cfg;
actor_system system;
scoped_actor self;
scheduler::test_coordinator& sched;
fixture()
: system(cfg),
self(system),
sched(dynamic_cast<scheduler::test_coordinator&>(system.scheduler())) {
CAF_REQUIRE(sched.jobs.empty());
}
~fixture() {
sched.run_dispatch_loop();
}
};
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(simple_timeout_tests, fixture) CAF_TEST_FIXTURE_SCOPE(simple_timeout_tests, test_coordinator_fixture<>)
CAF_TEST(duration_conversion) { CAF_TEST(duration_conversion) {
duration d1{time_unit::milliseconds, 100}; duration d1{time_unit::milliseconds, 100};
...@@ -107,11 +84,11 @@ CAF_TEST(duration_conversion) { ...@@ -107,11 +84,11 @@ CAF_TEST(duration_conversion) {
} }
CAF_TEST(single_timeout) { CAF_TEST(single_timeout) {
system.spawn(timer_impl); sys.spawn(timer_impl);
} }
CAF_TEST(single_anon_timeout) { CAF_TEST(single_anon_timeout) {
system.spawn(timer_impl2); sys.spawn(timer_impl2);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -1015,7 +1015,7 @@ CAF_TEST_DISABLED(out_of_order_delivery_udp) { ...@@ -1015,7 +1015,7 @@ CAF_TEST_DISABLED(out_of_order_delivery_udp) {
++expected_next; ++expected_next;
} }
); );
sched.dispatch(); sched.trigger_timeouts();
mpx()->flush_runnables(); mpx()->flush_runnables();
CAF_MESSAGE("force delivery via timeout that skips messages"); CAF_MESSAGE("force delivery via timeout that skips messages");
const basp::sequence_type seq_and_payload = 23; const basp::sequence_type seq_and_payload = 23;
...@@ -1023,7 +1023,7 @@ CAF_TEST_DISABLED(out_of_order_delivery_udp) { ...@@ -1023,7 +1023,7 @@ CAF_TEST_DISABLED(out_of_order_delivery_udp) {
.enqueue_back(jupiter().endpoint, header_with_seq(seq_and_payload), .enqueue_back(jupiter().endpoint, header_with_seq(seq_and_payload),
std::vector<actor_id>{}, make_message(seq_and_payload)) std::vector<actor_id>{}, make_message(seq_and_payload))
.deliver(jupiter().endpoint, 1); .deliver(jupiter().endpoint, 1);
sched.dispatch(); sched.trigger_timeouts();
mpx()->exec_runnable(); mpx()->exec_runnable();
self()->receive( self()->receive(
[&](basp::sequence_type val) { [&](basp::sequence_type val) {
......
...@@ -27,12 +27,14 @@ ...@@ -27,12 +27,14 @@
CAF_PUSH_WARNINGS CAF_PUSH_WARNINGS
namespace { /// The type of `_`.
struct wildcard { }; struct wildcard { };
/// Allows ignoring individual messages elements in `expect` clauses, e.g.
/// `expect((int, int), from(foo).to(bar).with(1, _))`.
constexpr wildcard _ = wildcard{}; constexpr wildcard _ = wildcard{};
/// @relates wildcard
constexpr bool operator==(const wildcard&, const wildcard&) { constexpr bool operator==(const wildcard&, const wildcard&) {
return true; return true;
} }
...@@ -56,8 +58,6 @@ msg_cmp_rec(const caf::message& x, const std::tuple<Ts...>& ys) { ...@@ -56,8 +58,6 @@ msg_cmp_rec(const caf::message& x, const std::tuple<Ts...>& ys) {
return cmp_one<I>(x, std::get<I>(ys)) && msg_cmp_rec<I + 1>(x, ys); return cmp_one<I>(x, std::get<I>(ys)) && msg_cmp_rec<I + 1>(x, ys);
} }
} // namespace <anonymous>
// allow comparing arbitrary `T`s to `message` objects for the purpose of the // allow comparing arbitrary `T`s to `message` objects for the purpose of the
// testing DSL // testing DSL
namespace caf { namespace caf {
...@@ -74,8 +74,6 @@ bool operator==(const message& x, const T& y) { ...@@ -74,8 +74,6 @@ bool operator==(const message& x, const T& y) {
} // namespace caf } // namespace caf
namespace {
// dummy function to force ADL later on // dummy function to force ADL later on
//int inspect(int, int); //int inspect(int, int);
...@@ -189,16 +187,16 @@ public: ...@@ -189,16 +187,16 @@ public:
caf_handle& operator=(const caf_handle&) = default; caf_handle& operator=(const caf_handle&) = default;
inline pointer get() const { pointer get() const {
return ptr_; return ptr_;
} }
inline ptrdiff_t compare(const caf_handle& other) const { ptrdiff_t compare(const caf_handle& other) const {
return reinterpret_cast<ptrdiff_t>(ptr_) return reinterpret_cast<ptrdiff_t>(ptr_)
- reinterpret_cast<ptrdiff_t>(other.ptr_); - reinterpret_cast<ptrdiff_t>(other.ptr_);
} }
inline ptrdiff_t compare(std::nullptr_t) const { ptrdiff_t compare(std::nullptr_t) const {
return reinterpret_cast<ptrdiff_t>(ptr_); return reinterpret_cast<ptrdiff_t>(ptr_);
} }
...@@ -546,35 +544,17 @@ struct test_coordinator_fixture_fetch_helper<T> { ...@@ -546,35 +544,17 @@ struct test_coordinator_fixture_fetch_helper<T> {
/// A fixture with a deterministic scheduler setup. /// 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 { class test_coordinator_fixture {
public:
// -- member types -----------------------------------------------------------
/// A deterministic scheduler type. /// A deterministic scheduler type.
using scheduler_type = caf::scheduler::test_coordinator; using scheduler_type = caf::scheduler::test_coordinator;
/// Convenience alias for std::chrono::microseconds. /// Callback for boolean predicates.
using us_t = std::chrono::microseconds; using bool_predicate = std::function<bool()>;
/// The user-generated system config.
Config cfg;
/// Host system for (scheduled) actors.
caf::actor_system sys;
/// A scoped actor for conveniently sending and receiving messages.
caf::scoped_actor self;
/// Deterministic scheduler.
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 // -- constructors, destructors, and assignment operators --------------------
/// max-batch-delay. Using this duration for `sched.run_dispatch_loop()`
/// advances the clock in ideal steps.
caf::timespan streaming_cycle;
template <class... Ts> template <class... Ts>
explicit test_coordinator_fixture(Ts&&... xs) explicit test_coordinator_fixture(Ts&&... xs)
...@@ -584,57 +564,146 @@ struct test_coordinator_fixture { ...@@ -584,57 +564,146 @@ struct test_coordinator_fixture {
.set("logger.inline-output", true) .set("logger.inline-output", true)
.set("middleman.network-backend", caf::atom("testing"))), .set("middleman.network-backend", caf::atom("testing"))),
self(sys, true), self(sys, true),
sched(dynamic_cast<scheduler_type&>(sys.scheduler())), sched(dynamic_cast<scheduler_type&>(sys.scheduler())) {
credit_round_interval(cfg.streaming_credit_round_interval()),
max_batch_delay(cfg.streaming_max_batch_delay()) {
// Configure the clock to measure each batch item with 1us. // Configure the clock to measure each batch item with 1us.
sched.clock().time_per_unit.emplace(caf::atom("batch"), sched.clock().time_per_unit.emplace(caf::atom("batch"),
caf::timespan{1000}); caf::timespan{1000});
// Compute reasonable step size.
auto cycle_us = cfg.streaming_tick_duration_us();
streaming_cycle = caf::timespan{us_t{cycle_us}};
// Make sure the current time isn't 0. // Make sure the current time isn't 0.
sched.clock().current_time += streaming_cycle; sched.clock().current_time += tick_duration();
credit_round_interval = cfg.streaming_credit_round_interval();
} }
virtual ~test_coordinator_fixture() { virtual ~test_coordinator_fixture() {
sched.clock().cancel_all(); run();
sched.run(); }
// -- DSL functions ----------------------------------------------------------
/// Returns the duration of a single clock tick.
virtual caf::timespan tick_duration() const {
return cfg.streaming_tick_duration();
}
/// Advances the clock by a single tick duration.
size_t advance_time(caf::timespan interval) {
return sched.clock().advance_time(interval);
}
/// Allows the next actor to consume one message from its mailbox.
/// @returns Whether a message was consumed.
bool consume_message() {
return sched.try_run_once();
}
/// Allows each actors to consume all messages from its mailbox.
/// @returns The number of consumed messages.
size_t consume_messages() {
return sched.run();
}
/// Advances the clock by `tick_duration()` and tries dispatching all pending
/// timeouts.
/// @returns The number of triggered timeouts.
size_t tick() {
return advance_time(tick_duration());
} }
/// Dispatches messages and timeouts until no activity remains. /// Consume messages and trigger timeouts until no activity remains.
void run_exhaustively() { /// @returns The total number of events, i.e., messages consumed and
sched.run_dispatch_loop(streaming_cycle); /// timeouts triggerd.
size_t run() {
run_until([] { return false; });
} }
/// Dispatches messages and timeouts until no activity remains. /// Consume messages and trigger timeouts until `pred` becomes `true` or
template <class Predicate> /// until no activity remains.
void run_exhaustively_while(Predicate predicate) { /// @returns The total number of events, i.e., messages consumed and
sched.run_dispatch_loop(predicate, streaming_cycle); /// timeouts triggerd.
size_t run_until(bool_predicate pred) {
auto res = sched.run_cycle_until(pred, tick_duration());
return res.first + res.second;
} }
/// Sends a request to `from`, then calls `run_exhaustively`, and finally /// Call `run()` when the next scheduled actor becomes ready.
/// fetches and returns the result. void run_after_next_ready_event() {
sched.after_next_enqueue([=] { run(); });
}
/// Call `run_until(predicate)` when the next scheduled actor becomes ready.
void run_until_after_next_ready_event(bool_predicate predicate) {
sched.after_next_enqueue([=] { run_until(predicate); });
}
/// Sends a request to `hdl`, then calls `run()`, and finally fetches and
/// returns the result.
template <class T, class... Ts, class Handle, class... Us> template <class T, class... Ts, class Handle, class... Us>
typename std::conditional<sizeof...(Ts) == 0, T, std::tuple<T, Ts...>>::type typename std::conditional<sizeof...(Ts) == 0, T, std::tuple<T, Ts...>>::type
request(Handle from, Us... args) { request(Handle hdl, Us... args) {
auto res_hdl = self->request(from, caf::infinite, std::move(args)...); auto res_hdl = self->request(hdl, caf::infinite, std::move(args)...);
run_exhaustively(); run();
test_coordinator_fixture_fetch_helper<T, Ts...> f; test_coordinator_fixture_fetch_helper<T, Ts...> f;
return f(res_hdl); return f(res_hdl);
} }
/// Returns the next message from the next pending actor's mailbox as `T`.
template <class T> template <class T>
const T& peek() { const T& peek() {
return sched.template peek<T>(); return sched.template peek<T>();
} }
/// Dereferences `hdl` and downcasts it to `T`.
template <class T = caf::scheduled_actor, class Handle = caf::actor> template <class T = caf::scheduled_actor, class Handle = caf::actor>
T& deref(const Handle& hdl) { T& deref(const Handle& hdl) {
auto ptr = caf::actor_cast<caf::abstract_actor*>(hdl); auto ptr = caf::actor_cast<caf::abstract_actor*>(hdl);
CAF_REQUIRE(ptr != nullptr); CAF_REQUIRE(ptr != nullptr);
return dynamic_cast<T&>(*ptr); return dynamic_cast<T&>(*ptr);
} }
/// Tries to advance the simulation, e.g., by handling the next message or
/// mocking some network activity.
/// @private
virtual bool advance() {
return sched.try_run_once();
}
/// Tries to trigger a timeout.
/// @private
virtual bool trigger_timeout() {
return sched.trigger_timeout();
}
// -- member variables -------------------------------------------------------
/// The user-generated system config.
Config cfg;
/// Host system for (scheduled) actors.
caf::actor_system sys;
/// A scoped actor for conveniently sending and receiving messages.
caf::scoped_actor self;
/// Deterministic scheduler.
scheduler_type& sched;
// -- deprecated functions ---------------------------------------------------
caf::timespan credit_round_interval CAF_DEPRECATED;
void run_exhaustively() CAF_DEPRECATED_MSG("use run() instead") {
run_exhaustively_until([] { return false; });
}
void run_exhaustively_until(bool_predicate predicate)
CAF_DEPRECATED_MSG("use run_until() instead") {
auto res = sched.run_cycle_until(predicate, credit_round_interval);
return res.first + res.second;
}
void loop_after_next_enqueue()
CAF_DEPRECATED_MSG("use run_after_next_ready_event() instead") {
sched.after_next_enqueue([=] { run(); });
}
}; };
/// Unboxes an expected value or fails the test if it doesn't exist. /// Unboxes an expected value or fails the test if it doesn't exist.
...@@ -653,8 +722,6 @@ T unbox(caf::optional<T> x) { ...@@ -653,8 +722,6 @@ T unbox(caf::optional<T> x) {
return std::move(*x); return std::move(*x);
} }
} // namespace <anonymous>
/// Expands to its argument. /// Expands to its argument.
#define CAF_EXPAND(x) x #define CAF_EXPAND(x) x
......
...@@ -18,41 +18,55 @@ ...@@ -18,41 +18,55 @@
#pragma once #pragma once
#include <functional>
#include "caf/io/all.hpp" #include "caf/io/all.hpp"
#include "caf/io/network/test_multiplexer.hpp" #include "caf/io/network/test_multiplexer.hpp"
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
namespace { /// Ensures that `test_node_fixture` can override `run_exhaustively` even if
/// the base fixture does not declare these member functions virtual.
template <class BaseFixture>
class test_node_fixture_base {
public:
// -- constructors, destructors, and assignment operators --------------------
virtual ~test_node_fixture_base() {
// nop
}
// -- interface functions ----------------------------------------------------
virtual bool advance() = 0;
virtual bool trigger_timeout() = 0;
};
/// A fixture containing all required state to simulate a single CAF node. /// A fixture containing all required state to simulate a single CAF node.
template <class BaseFixture = template <class BaseFixture =
test_coordinator_fixture<caf::actor_system_config>> test_coordinator_fixture<caf::actor_system_config>>
class test_node_fixture : public BaseFixture { class test_node_fixture : public BaseFixture,
test_node_fixture_base<BaseFixture> {
public: public:
using super = BaseFixture; // -- member types -----------------------------------------------------------
using exec_all_nodes_fun = std::function<void ()>; /// Base type.
using super = BaseFixture;
exec_all_nodes_fun exec_all_nodes; /// Callback function type.
caf::io::middleman& mm; using run_all_nodes_fun = std::function<void()>;
caf::io::network::test_multiplexer& mpx;
/// @param fun A function object for delegating to the parent's `exec_all`. /// @param fun A function object for delegating to the parent's `exec_all`.
test_node_fixture(exec_all_nodes_fun fun) test_node_fixture(run_all_nodes_fun fun)
: exec_all_nodes(std::move(fun)), : mm(this->sys.middleman()),
mm(this->sys.middleman()), mpx(dynamic_cast<caf::io::network::test_multiplexer&>(mm.backend())),
mpx(dynamic_cast<caf::io::network::test_multiplexer&>(mm.backend())) { run_all_nodes(std::move(fun)) {
// nop // nop
} }
// Convenience function for transmitting all "network" traffic and running test_node_fixture() : test_node_fixture([=] { this->run(); }) {
// all executables on this node. // nop
void exec_all() {
while (mpx.try_exec_runnable() || mpx.read_data()
|| mpx.try_accept_connection() || this->sched.try_run_once()) {
// rince and repeat
}
} }
/// Convenience function for calling `mm.publish` and requiring a valid /// Convenience function for calling `mm.publish` and requiring a valid
...@@ -71,17 +85,39 @@ public: ...@@ -71,17 +85,39 @@ public:
template <class Handle = caf::actor> template <class Handle = caf::actor>
Handle remote_actor(std::string host, uint16_t port) { Handle remote_actor(std::string host, uint16_t port) {
this->sched.inline_next_enqueue(); this->sched.inline_next_enqueue();
this->sched.after_next_enqueue(exec_all_nodes); this->sched.after_next_enqueue(run_all_nodes);
auto res = mm.remote_actor<Handle>(std::move(host), port); auto res = mm.remote_actor<Handle>(std::move(host), port);
CAF_REQUIRE(res); CAF_REQUIRE(res);
return *res; return *res;
} }
private: // -- member variables -------------------------------------------------------
caf::io::basp_broker* get_basp_broker() {
auto hdl = mm.named_broker<caf::io::basp_broker>(caf::atom("BASP")); /// Reference to the node's middleman.
return dynamic_cast<caf::io::basp_broker*>( caf::io::middleman& mm;
caf::actor_cast<caf::abstract_actor*>(hdl));
/// Reference to the middleman's event multiplexer.
caf::io::network::test_multiplexer& mpx;
/// Callback for triggering all nodes when simulating a network of CAF nodes.
run_all_nodes_fun run_all_nodes;
// -- deprecated functions ---------------------------------------------------
void exec_all() CAF_DEPRECATED_MSG("use run() instead") {
this->run();
}
// -- overriding member functions --------------------------------------------
bool advance() override {
return mpx.try_exec_runnable() || mpx.read_data()
|| mpx.try_accept_connection() || this->sched.try_run_once();
}
bool trigger_timeout() override {
// Same as in dsl.hpp, but we have to provide it here again.
return this->sched.trigger_timeout();
} }
}; };
...@@ -93,9 +129,7 @@ void exec_all_fixtures(Iterator first, Iterator last) { ...@@ -93,9 +129,7 @@ void exec_all_fixtures(Iterator first, Iterator last) {
|| x->mpx.try_exec_runnable() || x->mpx.try_accept_connection(); || x->mpx.try_exec_runnable() || x->mpx.try_accept_connection();
}; };
auto trigger_timeouts = [](fixture_ptr x) { auto trigger_timeouts = [](fixture_ptr x) {
auto& sched = x->sched; x->sched.trigger_timeouts();
sched.clock().current_time += x->credit_round_interval;
sched.dispatch();
}; };
for (;;) { for (;;) {
// Exhaust all messages in the system. // Exhaust all messages in the system.
...@@ -109,14 +143,9 @@ void exec_all_fixtures(Iterator first, Iterator last) { ...@@ -109,14 +143,9 @@ void exec_all_fixtures(Iterator first, Iterator last) {
} }
} }
/// Binds `test_coordinator_fixture<Config>` to `test_node_fixture`.
template <class Config = caf::actor_system_config>
using test_node_fixture_t = test_node_fixture<test_coordinator_fixture<Config>>;
/// Base fixture for simulated network settings with any number of CAF nodes. /// Base fixture for simulated network settings with any number of CAF nodes.
template <class PlanetType> template <class PlanetType>
class fake_network_fixture_base { class test_network_fixture_base {
public: public:
using planets_vector = std::vector<PlanetType*>; using planets_vector = std::vector<PlanetType*>;
...@@ -124,7 +153,7 @@ public: ...@@ -124,7 +153,7 @@ public:
using accept_handle = caf::io::accept_handle; using accept_handle = caf::io::accept_handle;
fake_network_fixture_base(planets_vector xs) : planets_(std::move(xs)) { test_network_fixture_base(planets_vector xs) : planets_(std::move(xs)) {
// nop // nop
} }
...@@ -199,11 +228,11 @@ private: ...@@ -199,11 +228,11 @@ private:
template <class BaseFixture = template <class BaseFixture =
test_coordinator_fixture<caf::actor_system_config>> test_coordinator_fixture<caf::actor_system_config>>
class point_to_point_fixture class point_to_point_fixture
: public fake_network_fixture_base<test_node_fixture<BaseFixture>> { : public test_network_fixture_base<test_node_fixture<BaseFixture>> {
public: public:
using planet_type = test_node_fixture<BaseFixture>; using planet_type = test_node_fixture<BaseFixture>;
using super = fake_network_fixture_base<planet_type>; using super = test_network_fixture_base<planet_type>;
planet_type earth; planet_type earth;
planet_type mars; planet_type mars;
...@@ -217,21 +246,16 @@ public: ...@@ -217,21 +246,16 @@ public:
} }
}; };
/// Binds `test_coordinator_fixture<Config>` to `point_to_point_fixture`.
template <class Config = caf::actor_system_config>
using point_to_point_fixture_t =
point_to_point_fixture<test_coordinator_fixture<Config>>;
/// A simple fixture that includes three nodes (`earth`, `mars`, and `jupiter`) /// A simple fixture that includes three nodes (`earth`, `mars`, and `jupiter`)
/// that can connect to each other. /// that can connect to each other.
template <class BaseFixture = template <class BaseFixture =
test_coordinator_fixture<caf::actor_system_config>> test_coordinator_fixture<caf::actor_system_config>>
class belt_fixture class belt_fixture
: public fake_network_fixture_base<test_node_fixture<BaseFixture>> { : public test_network_fixture_base<test_node_fixture<BaseFixture>> {
public: public:
using planet_type = test_node_fixture<BaseFixture>; using planet_type = test_node_fixture<BaseFixture>;
using super = fake_network_fixture_base<planet_type>; using super = test_network_fixture_base<planet_type>;
planet_type earth; planet_type earth;
planet_type mars; planet_type mars;
...@@ -246,13 +270,6 @@ public: ...@@ -246,13 +270,6 @@ public:
} }
}; };
/// Binds `test_coordinator_fixture<Config>` to `belt_fixture`.
template <class Config = caf::actor_system_config>
using belt_fixture_t =
belt_fixture<test_coordinator_fixture<Config>>;
}// namespace <anonymous>
#define expect_on(where, types, fields) \ #define expect_on(where, types, fields) \
CAF_MESSAGE(#where << ": expect" << #types << "." << #fields); \ CAF_MESSAGE(#where << ": expect" << #types << "." << #fields); \
expect_clause< CAF_EXPAND(CAF_DSL_LIST types) >{where . sched} . fields expect_clause< CAF_EXPAND(CAF_DSL_LIST types) >{where . sched} . fields
......
...@@ -252,7 +252,7 @@ public: ...@@ -252,7 +252,7 @@ public:
return disabled_; return disabled_;
} }
virtual void run() = 0; virtual void run_test_impl() = 0;
private: private:
size_t expected_failures_; size_t expected_failures_;
...@@ -272,9 +272,9 @@ public: ...@@ -272,9 +272,9 @@ public:
// nop // nop
} }
void run() override { void run_test_impl() override {
T impl; T impl;
impl.run(); impl.run_test_impl();
} }
}; };
...@@ -603,12 +603,12 @@ using caf_test_case_auto_fixture = caf::test::dummy_fixture; ...@@ -603,12 +603,12 @@ using caf_test_case_auto_fixture = caf::test::dummy_fixture;
#define CAF_TEST_IMPL(name, disabled_by_default) \ #define CAF_TEST_IMPL(name, disabled_by_default) \
namespace { \ namespace { \
struct CAF_UNIQUE(test) : caf_test_case_auto_fixture { \ struct CAF_UNIQUE(test) : caf_test_case_auto_fixture { \
void run(); \ void run_test_impl(); \
}; \ }; \
::caf::test::detail::adder<::caf::test::test_impl<CAF_UNIQUE(test)>> \ ::caf::test::detail::adder<::caf::test::test_impl<CAF_UNIQUE(test)>> \
CAF_UNIQUE(a){CAF_XSTR(CAF_SUITE), CAF_XSTR(name), disabled_by_default}; \ CAF_UNIQUE(a){CAF_XSTR(CAF_SUITE), CAF_XSTR(name), disabled_by_default}; \
} /* namespace <anonymous> */ \ } /* namespace <anonymous> */ \
void CAF_UNIQUE(test)::run() void CAF_UNIQUE(test)::run_test_impl()
#define CAF_TEST(name) CAF_TEST_IMPL(name, false) #define CAF_TEST(name) CAF_TEST_IMPL(name, false)
......
...@@ -369,7 +369,7 @@ bool engine::run(bool colorize, ...@@ -369,7 +369,7 @@ bool engine::run(bool colorize,
<< '\n'; << '\n';
auto start = std::chrono::high_resolution_clock::now(); auto start = std::chrono::high_resolution_clock::now();
watchdog::start(max_runtime()); watchdog::start(max_runtime());
t->run(); t->run_test_impl();
watchdog::stop(); watchdog::stop();
auto stop = std::chrono::high_resolution_clock::now(); auto stop = std::chrono::high_resolution_clock::now();
auto elapsed = auto elapsed =
......
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