Commit 07dcd1e6 authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Remove test_coordinator::run_cycle, refine DSL API

parent 0935d0e3
...@@ -34,6 +34,11 @@ public: ...@@ -34,6 +34,11 @@ 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;
/// Returns whether the actor clock has at least one pending timeout.
bool has_pending_timeout() const {
return !schedule_.empty();
}
/// Triggers the next pending timeout regardless of its timestamp. Sets /// Triggers the next pending timeout regardless of its timestamp. Sets
/// `current_time` to the time point of the triggered timeout unless /// `current_time` to the time point of the triggered timeout unless
/// `current_time` is already set to a later time. /// `current_time` is already set to a later time.
......
...@@ -105,6 +105,11 @@ public: ...@@ -105,6 +105,11 @@ 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());
/// Returns whether at least one pending timeout exists.
bool has_pending_timeout() const {
return clock_.has_pending_timeout();
}
/// Tries to trigger a single timeout. /// Tries to trigger a single timeout.
bool trigger_timeout() { bool trigger_timeout() {
return clock_.trigger_timeout(); return clock_.trigger_timeout();
...@@ -120,23 +125,6 @@ public: ...@@ -120,23 +125,6 @@ public:
return clock_.advance_time(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
/// to `dispatch()` or the default tick-duration when passing 0ns.
std::pair<size_t, size_t> run_cycle_until(bool_predicate predicate,
timespan cycle_duration);
/// Loops until no job or delayed message remains. Returns the total number
/// of events (first) and dispatched delayed messages (second). Advances time
/// by `cycle` nanoseconds between to calls to `dispatch()` or the default
/// tick-duration when passing 0ns.
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) {
after_next_enqueue_ = f; after_next_enqueue_ = f;
......
...@@ -87,7 +87,8 @@ void test_coordinator::start() { ...@@ -87,7 +87,8 @@ void test_coordinator::start() {
} }
void test_coordinator::stop() { void test_coordinator::stop() {
run_cycle(); while (run() > 0)
trigger_timeouts();
} }
void test_coordinator::enqueue(resumable* ptr) { void test_coordinator::enqueue(resumable* ptr) {
...@@ -147,55 +148,6 @@ size_t test_coordinator::run(size_t max_count) { ...@@ -147,55 +148,6 @@ size_t test_coordinator::run(size_t max_count) {
return res; return res;
} }
std::pair<size_t, size_t>
test_coordinator::run_cycle_until(bool_predicate predicate,
timespan cycle_duration) {
CAF_LOG_TRACE(CAF_ARG(cycle_duration)
<< CAF_ARG2("pending_jobs", jobs.size())
<< CAF_ARG2("pending_timeouts", clock_.schedule().size()));
// Bookkeeping.
size_t events = 0;
size_t timeouts = 0;
// Loop until no activity remains.
while (!jobs.empty() || !clock_.schedule().empty()) {
while (try_run_once()) {
++events;
if (predicate()) {
CAF_LOG_DEBUG("stop due to predicate;"
<< CAF_ARG(events) << CAF_ARG(timeouts));
return {events, timeouts};
}
}
if (!clock_.schedule().empty()) {
size_t triggered = 0;
auto next_tout = clock_.schedule().begin()->first;
if (next_tout <= clock_.now()) {
// Trigger expired timeout first without advancing time.
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;
}
}
CAF_LOG_DEBUG("no activity left;" << CAF_ARG(events) << CAF_ARG(timeouts));
return {events, timeouts};
}
std::pair<size_t, size_t> test_coordinator::run_cycle(timespan cycle_duration) {
return run_cycle_until([] { return false; }, cycle_duration);
}
void test_coordinator::inline_next_enqueue() { void test_coordinator::inline_next_enqueue() {
after_next_enqueue([=] { run_once_lifo(); }); after_next_enqueue([=] { run_once_lifo(); });
} }
...@@ -211,7 +163,13 @@ void test_coordinator::inline_all_enqueues_helper() { ...@@ -211,7 +163,13 @@ void test_coordinator::inline_all_enqueues_helper() {
std::pair<size_t, size_t> std::pair<size_t, size_t>
test_coordinator::run_dispatch_loop(timespan cycle_duration) { test_coordinator::run_dispatch_loop(timespan cycle_duration) {
return run_cycle(cycle_duration); size_t messages = 0;
size_t timeouts = 0;
while (has_job() || has_pending_timeout()) {
messages += run();
timeouts += advance_time(cycle_duration);
}
return {messages, timeouts};
} }
} // namespace caf } // namespace caf
......
...@@ -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_ARG(interval_) << CAF_ARG(start_)); CAF_LOG_TRACE(CAF_ARG(now) << CAF_ARG(periods)
<< CAF_ARG(interval_) << CAF_ARG(start_));
CAF_ASSERT(now >= start_); CAF_ASSERT(now >= start_);
size_t result = 0; size_t result = 0;
auto f = [&](size_t tick) { auto f = [&](size_t tick) {
......
...@@ -221,7 +221,7 @@ CAF_TEST(closing_downstreams_before_end_of_stream) { ...@@ -221,7 +221,7 @@ CAF_TEST(closing_downstreams_before_end_of_stream) {
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");
tick(); trigger_timeouts();
consume_messages(); consume_messages();
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());
......
...@@ -212,8 +212,8 @@ TESTEE(doubler) { ...@@ -212,8 +212,8 @@ TESTEE(doubler) {
} }
struct fixture : test_coordinator_fixture<> { struct fixture : test_coordinator_fixture<> {
timespan tick_duration() const override { void tick() {
return credit_round_interval; advance_time(cfg.streaming_credit_round_interval());
} }
}; };
......
...@@ -551,9 +551,6 @@ public: ...@@ -551,9 +551,6 @@ public:
/// A deterministic scheduler type. /// A deterministic scheduler type.
using scheduler_type = caf::scheduler::test_coordinator; using scheduler_type = caf::scheduler::test_coordinator;
/// Callback for boolean predicates.
using bool_predicate = std::function<bool()>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
template <class... Ts> template <class... Ts>
...@@ -569,7 +566,7 @@ public: ...@@ -569,7 +566,7 @@ public:
sched.clock().time_per_unit.emplace(caf::atom("batch"), sched.clock().time_per_unit.emplace(caf::atom("batch"),
caf::timespan{1000}); caf::timespan{1000});
// Make sure the current time isn't 0. // Make sure the current time isn't 0.
sched.clock().current_time += tick_duration(); sched.clock().current_time += std::chrono::hours(1);
credit_round_interval = cfg.streaming_credit_round_interval(); credit_round_interval = cfg.streaming_credit_round_interval();
} }
...@@ -579,33 +576,26 @@ public: ...@@ -579,33 +576,26 @@ public:
// -- DSL functions ---------------------------------------------------------- // -- 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. /// Advances the clock by a single tick duration.
size_t advance_time(caf::timespan interval) { size_t advance_time(caf::timespan interval) {
return sched.clock().advance_time(interval); return sched.clock().advance_time(interval);
} }
/// Allows the next actor to consume one message from its mailbox. /// Allows the next actor to consume one message from its mailbox. Fails the
/// @returns Whether a message was consumed. /// test if no message was consumed.
bool consume_message() { void consume_message() {
return sched.try_run_once(); if (!sched.try_run_once())
CAF_FAIL("no message to consume");
} }
/// Allows each actors to consume all messages from its mailbox. /// Allows each actors to consume all messages from its mailbox. Fails the
/// test if no message was consumed.
/// @returns The number of consumed messages. /// @returns The number of consumed messages.
size_t consume_messages() { size_t consume_messages() {
return sched.run(); auto result = sched.run();
} if (result == 0)
CAF_FAIL("no message to consume");
/// Advances the clock by `tick_duration()` and tries dispatching all pending return result;
/// timeouts.
/// @returns The number of triggered timeouts.
size_t tick() {
return advance_time(tick_duration());
} }
/// Consume messages and trigger timeouts until no activity remains. /// Consume messages and trigger timeouts until no activity remains.
...@@ -618,10 +608,26 @@ public: ...@@ -618,10 +608,26 @@ public:
/// Consume messages and trigger timeouts until `pred` becomes `true` or /// Consume messages and trigger timeouts until `pred` becomes `true` or
/// until no activity remains. /// until no activity remains.
/// @returns The total number of events, i.e., messages consumed and /// @returns The total number of events, i.e., messages consumed and
/// timeouts triggerd. /// timeouts triggered.
size_t run_until(bool_predicate pred) { template <class BoolPredicate>
auto res = sched.run_cycle_until(pred, tick_duration()); size_t run_until(BoolPredicate predicate) {
return res.first + res.second; CAF_LOG_TRACE("");
// Bookkeeping.
size_t events = 0;
// Loop until no activity remains.
while (sched.has_job() || sched.has_pending_timeout()) {
while (sched.try_run_once()) {
++events;
if (predicate()) {
CAF_LOG_DEBUG("stop due to predicate:" << CAF_ARG(events));
return events;
}
}
if (trigger_timeout())
++events;
}
CAF_LOG_DEBUG("no activity left:" << CAF_ARG(events));
return events;
} }
/// Call `run()` when the next scheduled actor becomes ready. /// Call `run()` when the next scheduled actor becomes ready.
...@@ -630,7 +636,8 @@ public: ...@@ -630,7 +636,8 @@ public:
} }
/// Call `run_until(predicate)` when the next scheduled actor becomes ready. /// Call `run_until(predicate)` when the next scheduled actor becomes ready.
void run_until_after_next_ready_event(bool_predicate predicate) { template <class BoolPredicate>
void run_until_after_next_ready_event(BoolPredicate predicate) {
sched.after_next_enqueue([=] { run_until(predicate); }); sched.after_next_enqueue([=] { run_until(predicate); });
} }
...@@ -659,19 +666,19 @@ public: ...@@ -659,19 +666,19 @@ public:
return dynamic_cast<T&>(*ptr); return dynamic_cast<T&>(*ptr);
} }
/// Tries to advance the simulation, e.g., by handling the next message or /// Triggers the next pending timeout.
/// mocking some network activity.
/// @private
virtual bool advance() {
return sched.try_run_once();
}
/// Tries to trigger a timeout.
/// @private
virtual bool trigger_timeout() { virtual bool trigger_timeout() {
return sched.trigger_timeout(); return sched.trigger_timeout();
} }
/// Triggers all pending timeouts.
size_t trigger_timeouts() {
size_t timeouts = 0;
while (trigger_timeout())
++timeouts;
return timeouts;
}
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// The user-generated system config. /// The user-generated system config.
...@@ -686,24 +693,34 @@ public: ...@@ -686,24 +693,34 @@ public:
/// Deterministic scheduler. /// Deterministic scheduler.
scheduler_type& sched; scheduler_type& sched;
// -- deprecated functions --------------------------------------------------- // -- deprecated functionality -----------------------------------------------
void run_exhaustively() CAF_DEPRECATED_MSG("use run() instead");
void run_exhaustively_until(std::function<bool()> f)
CAF_DEPRECATED_MSG("use run_until() instead");
void loop_after_next_enqueue()
CAF_DEPRECATED_MSG("use run_after_next_ready_event() instead");
caf::timespan credit_round_interval CAF_DEPRECATED; caf::timespan credit_round_interval CAF_DEPRECATED;
};
void run_exhaustively() CAF_DEPRECATED_MSG("use run() instead") { template <class Config>
run_exhaustively_until([] { return false; }); void test_coordinator_fixture<Config>::run_exhaustively() {
} run();
}
void run_exhaustively_until(bool_predicate predicate) template <class Config>
CAF_DEPRECATED_MSG("use run_until() instead") { void test_coordinator_fixture<Config>::run_exhaustively_until(
sched.run_cycle_until(predicate, credit_round_interval); std::function<bool()> f) {
} run_until(std::move(f));
}
void loop_after_next_enqueue() template <class Config>
CAF_DEPRECATED_MSG("use run_after_next_ready_event() instead") { void test_coordinator_fixture<Config>::loop_after_next_enqueue() {
sched.after_next_enqueue([=] { run(); }); 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.
template <class T> template <class T>
......
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