Commit 42f463c9 authored by Dominik Charousset's avatar Dominik Charousset

Port test coordinator to new clock API

parent 2f308ccb
......@@ -30,6 +30,15 @@ public:
time_point now() const noexcept override;
/// Tries to dispatch the next timeout or delayed message regardless of its
/// timestamp. Returns `false` if `schedule().empty()`, otherwise `true`.
bool dispatch_once();
/// Dispatches all timeouts and delayed messages regardless of their
/// timestamp. Returns the number of dispatched events.
size_t dispatch();
/// Advances the time by `x` and dispatches timeouts and delayed messages.
void advance_time(duration_type x);
};
......
......@@ -45,20 +45,9 @@ public:
/// A double-ended queue representing our current job queue.
std::deque<resumable*> jobs;
/// A scheduled message or timeout.
struct delayed_msg {
strong_actor_ptr from;
strong_actor_ptr to;
message_id mid;
message msg;
};
/// A clock type using the highest available precision.
using hrc = std::chrono::high_resolution_clock;
/// A map type for storing scheduled messages and timeouts.
std::multimap<hrc::time_point, delayed_msg> delayed_messages;
/// Returns whether at least one job is in the queue.
inline bool has_job() const {
return !jobs.empty();
......
......@@ -25,6 +25,27 @@ test_actor_clock::time_point test_actor_clock::now() const noexcept {
return current_time;
}
bool test_actor_clock::dispatch_once() {
if (schedule_.empty())
return false;
visitor f{this};
auto i = schedule_.begin();
visit(f, i->second);
schedule_.erase(i);
return true;
}
size_t test_actor_clock::dispatch() {
if (schedule_.empty())
return 0u;
visitor f{this};
auto result = schedule_.size();
for (auto& kvp : schedule_)
visit(f, kvp.second);
schedule_.clear();
return result;
}
void test_actor_clock::advance_time(duration_type x) {
visitor f{this};
current_time += x;
......
......@@ -62,33 +62,6 @@ private:
message_handler mh_;
};
class dummy_timer : public monitorable_actor {
public:
dummy_timer(actor_config& cfg, test_coordinator* parent)
: monitorable_actor(cfg),
parent_(parent) {
mh_.assign(
[&](const duration& d, strong_actor_ptr& from,
strong_actor_ptr& to, message_id mid, message& msg) {
auto tout = test_coordinator::hrc::now();
tout += d;
using delayed_msg = test_coordinator::delayed_msg;
parent_->delayed_messages.emplace(tout, delayed_msg{std::move(from),
std::move(to), mid,
std::move(msg)});
}
);
}
void enqueue(mailbox_element_ptr what, execution_unit*) override {
mh_(what->content());
}
private:
test_coordinator* parent_;
message_handler mh_;
};
} // namespace <anonymous>
test_coordinator::test_coordinator(actor_system& sys) : super(sys) {
......@@ -173,20 +146,11 @@ size_t test_coordinator::run(size_t max_count) {
}
bool test_coordinator::dispatch_once() {
auto i = delayed_messages.begin();
if (i == delayed_messages.end())
return false;
auto& dm = i->second;
dm.to->enqueue(dm.from, dm.mid, std::move(dm.msg), nullptr);
delayed_messages.erase(i);
return true;
return clock().dispatch_once();
}
size_t test_coordinator::dispatch() {
size_t res = 0;
while (dispatch_once())
++res;
return res;
return clock().dispatch();
}
std::pair<size_t, size_t> test_coordinator::run_dispatch_loop() {
......
......@@ -280,7 +280,6 @@ struct fixture {
self(system),
sched(dynamic_cast<scheduler::test_coordinator&>(system.scheduler())) {
CAF_REQUIRE(sched.jobs.empty());
CAF_REQUIRE(sched.delayed_messages.empty());
}
~fixture() {
......@@ -335,8 +334,7 @@ CAF_TEST(nested_timeout) {
// not respond to the message yet, i.e., timeout arrives before response
sched.run();
// dispatch second timeout
CAF_REQUIRE(!sched.delayed_messages.empty());
sched.dispatch();
CAF_REQUIRE_EQUAL(sched.dispatch(), true);
CAF_REQUIRE_EQUAL(sched.next_job<local_actor>().name(), string{"ping"});
CAF_CHECK(!had_timeout);
CAF_CHECK(sched.next_job<ping_actor>().state.had_first_timeout);
......
......@@ -87,7 +87,6 @@ struct fixture {
self(system),
sched(dynamic_cast<scheduler::test_coordinator&>(system.scheduler())) {
CAF_REQUIRE(sched.jobs.empty());
CAF_REQUIRE(sched.delayed_messages.empty());
}
~fixture() {
......
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