Commit 61068dc7 authored by Dominik Charousset's avatar Dominik Charousset

Coding style nitpicks and cleanup

parent efa04d70
...@@ -65,9 +65,8 @@ class abstract_coordinator { ...@@ -65,9 +65,8 @@ class abstract_coordinator {
void delayed_send(Duration rel_time, actor_addr from, channel to, void delayed_send(Duration rel_time, actor_addr from, channel to,
message_id mid, message data) { message_id mid, message data) {
m_timer->enqueue(invalid_actor_addr, invalid_message_id, m_timer->enqueue(invalid_actor_addr, invalid_message_id,
make_message(atom("_Send"), duration{rel_time}, make_message(duration{rel_time}, std::move(from),
std::move(from), std::move(to), mid, std::move(to), mid, std::move(data)),
std::move(data)),
nullptr); nullptr);
} }
......
...@@ -52,8 +52,6 @@ namespace { ...@@ -52,8 +52,6 @@ namespace {
using hrc = std::chrono::high_resolution_clock; using hrc = std::chrono::high_resolution_clock;
using time_point = hrc::time_point;
using timer_actor_policies = policy::actor_policies<policy::no_scheduling, using timer_actor_policies = policy::actor_policies<policy::no_scheduling,
policy::not_prioritizing, policy::not_prioritizing,
policy::no_resume, policy::no_resume,
...@@ -87,7 +85,7 @@ class timer_actor final : public detail::proper_actor<blocking_actor, ...@@ -87,7 +85,7 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
return next_message(); return next_message();
} }
inline unique_mailbox_element_pointer try_dequeue(const time_point& tp) { inline unique_mailbox_element_pointer try_dequeue(const hrc::time_point& tp) {
if (scheduling_policy().await_data(this, tp)) { if (scheduling_policy().await_data(this, tp)) {
return next_message(); return next_message();
} }
...@@ -95,25 +93,23 @@ class timer_actor final : public detail::proper_actor<blocking_actor, ...@@ -95,25 +93,23 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
} }
void act() override { void act() override {
trap_exit(true);
// setup & local variables // setup & local variables
bool done = false; bool done = false;
unique_mailbox_element_pointer msg_ptr; unique_mailbox_element_pointer msg_ptr;
auto tout = hrc::now(); std::multimap<hrc::time_point, delayed_msg> messages;
std::multimap<decltype(tout), delayed_msg> messages;
// message handling rules // message handling rules
message_handler mfun{ message_handler mfun{
on(atom("_Send"), arg_match) >> [&](const duration& d, [&](const duration& d, actor_addr& from, channel& to,
actor_addr& from, channel& to, message_id mid, message& msg) {
message_id mid, message& tup) {
insert_dmsg(messages, d, std::move(from), insert_dmsg(messages, d, std::move(from),
std::move(to), mid, std::move(tup)); std::move(to), mid, std::move(msg));
}, },
[&](const exit_msg&) { [&](const exit_msg&) {
done = true; done = true;
}, },
others() >> [&] { others() >> [&] {
std::cerr << "coordinator::timer_loop: UNKNOWN MESSAGE: " CAF_LOG_ERROR("unexpected: " << to_string(msg_ptr->msg));
<< to_string(msg_ptr->msg) << std::endl;
} }
}; };
// loop // loop
...@@ -122,7 +118,7 @@ class timer_actor final : public detail::proper_actor<blocking_actor, ...@@ -122,7 +118,7 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
if (messages.empty()) if (messages.empty())
msg_ptr = dequeue(); msg_ptr = dequeue();
else { else {
tout = hrc::now(); auto tout = hrc::now();
// handle timeouts (send messages) // 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) {
...@@ -165,19 +161,18 @@ void printer_loop(blocking_actor* self) { ...@@ -165,19 +161,18 @@ void printer_loop(blocking_actor* self) {
self->receive_while([&] { return running; })( self->receive_while([&] { return running; })(
on(atom("add"), arg_match) >> [&](std::string& str) { on(atom("add"), arg_match) >> [&](std::string& str) {
auto s = self->last_sender(); auto s = self->last_sender();
if (!str.empty() && s) { if (str.empty() || s == invalid_actor_addr) {
auto i = out.find(s); return;
if (i == out.end()) { }
i = out.insert(make_pair(s, std::move(str))).first; auto i = out.find(s);
// monitor actor to flush its output on exit if (i == out.end()) {
self->monitor(s); i = out.insert(make_pair(s, std::move(str))).first;
flush_if_needed(i->second); // monitor actor to flush its output on exit
} else { self->monitor(s);
auto& ref = i->second; } else {
ref += std::move(str); i->second += std::move(str);
flush_if_needed(ref);
}
} }
flush_if_needed(i->second);
}, },
on(atom("flush")) >> [&] { on(atom("flush")) >> [&] {
flush_output(self->last_sender()); flush_output(self->last_sender());
...@@ -189,10 +184,9 @@ void printer_loop(blocking_actor* self) { ...@@ -189,10 +184,9 @@ void printer_loop(blocking_actor* self) {
[&](const exit_msg&) { [&](const exit_msg&) {
running = false; running = false;
}, },
others() >> [self] { others() >> [&] {
std::cerr << "*** unexpected: " std::cerr << "*** unexpected: " << to_string(self->last_dequeued())
<< to_string(self->last_dequeued()) << std::endl;
<< std::endl;
} }
); );
} }
......
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