Commit 7debd139 authored by Dominik Charousset's avatar Dominik Charousset

Simplify implementation of timer_actor

parent c7aae68e
...@@ -29,6 +29,9 @@ ...@@ -29,6 +29,9 @@
#include <condition_variable> #include <condition_variable>
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/after.hpp"
#include "caf/others.hpp"
#include "caf/duration.hpp"
#include "caf/actor_system.hpp" #include "caf/actor_system.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/actor_ostream.hpp" #include "caf/actor_ostream.hpp"
...@@ -53,87 +56,74 @@ namespace { ...@@ -53,87 +56,74 @@ namespace {
using hrc = std::chrono::high_resolution_clock; using hrc = std::chrono::high_resolution_clock;
struct delayed_msg { class timer_actor : public blocking_actor {
public:
explicit timer_actor(actor_config& cfg) : blocking_actor(cfg) {
// nop
}
struct delayed_msg {
strong_actor_ptr from; strong_actor_ptr from;
strong_actor_ptr to; strong_actor_ptr to;
message_id mid; message_id mid;
message msg; message msg;
}; };
inline void deliver(delayed_msg& dm) { void deliver(delayed_msg& dm) {
dm.to->enqueue(dm.from, dm.mid, std::move(dm.msg), nullptr); dm.to->enqueue(dm.from, dm.mid, std::move(dm.msg), nullptr);
} }
template <class Map, class... Ts> template <class Map, class... Ts>
inline void insert_dmsg(Map& storage, const duration& d, Ts&&... xs) { void insert_dmsg(Map& storage, const duration& d, Ts&&... xs) {
auto tout = hrc::now(); auto tout = hrc::now();
tout += d; tout += d;
delayed_msg dmsg{std::forward<Ts>(xs)...}; delayed_msg dmsg{std::forward<Ts>(xs)...};
storage.emplace(std::move(tout), std::move(dmsg)); storage.emplace(std::move(tout), std::move(dmsg));
}
class timer_actor : public blocking_actor {
public:
explicit timer_actor(actor_config& cfg) : blocking_actor(cfg) {
// nop
}
bool await_data(const hrc::time_point& tp) {
if (has_next_message())
return true;
return mailbox().synchronized_await(mtx_, cv_, tp);
}
mailbox_element_ptr try_dequeue() {
blocking_actor::await_data();
return next_message();
}
mailbox_element_ptr try_dequeue(const hrc::time_point& tp) {
if (await_data(tp))
return next_message();
return mailbox_element_ptr{};
} }
void act() override { void act() override {
// setup & local variables // local state
accept_one_cond rc;
bool running = true;
std::multimap<hrc::time_point, delayed_msg> messages; std::multimap<hrc::time_point, delayed_msg> messages;
// message handling rules // our message handler
message_handler mfun{ auto bhvr = detail::make_blocking_behavior(
[&](const duration& d, strong_actor_ptr& from, strong_actor_ptr& to, behavior{
message_id mid, message& msg) { [&](const duration& d, strong_actor_ptr& from,
strong_actor_ptr& to, message_id mid, message& msg) {
insert_dmsg(messages, d, std::move(from), insert_dmsg(messages, d, std::move(from),
std::move(to), mid, std::move(msg)); std::move(to), mid, std::move(msg));
},
[&](const exit_msg& dm) {
if (dm.reason) {
fail_state(dm.reason);
running = false;
} }
}; }
mailbox_element_ptr msg_ptr; },
for (;;) { others >> [&](message_view& x) -> result<message> {
while (! msg_ptr) { std::cerr << "*** unexpected message in timer_actor: "
<< to_string(x.content()) << std::endl;
return sec::unexpected_message;
}
);
// loop until receiving an exit message
while (running) {
if (messages.empty()) { if (messages.empty()) {
msg_ptr = try_dequeue(); // use regular receive as long as we don't have a pending timeout
receive_impl(rc, message_id::make(), bhvr);
} else {
auto tout = messages.begin()->first;
if (await_data(tout)) {
receive_impl(rc, message_id::make(), bhvr);
} else { } else {
auto tout = hrc::now();
// handle timeouts (send messages)
auto it = messages.begin(); auto it = messages.begin();
while (it != messages.end() && (it->first) <= tout) { while (it != messages.end() && (it->first) <= tout) {
deliver(it->second); deliver(it->second);
it = messages.erase(it); it = messages.erase(it);
} }
// wait for next message or next timeout
if (it != messages.end())
msg_ptr = try_dequeue(it->first);
}
}
auto& content = msg_ptr->content();
if (content.type_token() == make_type_token<exit_msg>()) {
auto& em = content.get_as<exit_msg>(0);
if (em.reason) {
fail_state(em.reason);
return;
} }
} }
mfun(content);
msg_ptr.reset();
} }
} }
}; };
......
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