Commit 9ede3f3a authored by Dominik Charousset's avatar Dominik Charousset

Fix shutdown of scheduler helpers

parent 8d4e2e0a
...@@ -106,19 +106,21 @@ class timer_actor final : public detail::proper_actor<blocking_actor, ...@@ -106,19 +106,21 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
auto tout = hrc::now(); auto tout = hrc::now();
std::multimap<decltype(tout), delayed_msg> messages; std::multimap<decltype(tout), delayed_msg> messages;
// message handling rules // message handling rules
auto mfun = message_handler mfun{
(on(atom("_Send"), arg_match) >> [&](const duration& d, on(atom("_Send"), arg_match) >> [&](const duration& d,
actor_addr& from, channel& to, actor_addr& from, channel& to,
message_id mid, message& tup) { message_id mid, message& tup) {
insert_dmsg(messages, d, std::move(from), std::move(to), mid, insert_dmsg(messages, d, std::move(from),
std::move(tup)); std::move(to), mid, std::move(tup));
}, },
on(atom("DIE")) >> [&] { done = true; }, others() >> [&] { [&](const exit_msg&) {
# ifdef CPPA_DEBUG_MODE done = true;
},
others() >> [&] {
std::cerr << "coordinator::timer_loop: UNKNOWN MESSAGE: " std::cerr << "coordinator::timer_loop: UNKNOWN MESSAGE: "
<< to_string(msg_ptr->msg) << std::endl; << to_string(msg_ptr->msg) << std::endl;
# endif }
}); };
// loop // loop
while (!done) { while (!done) {
while (!msg_ptr) { while (!msg_ptr) {
...@@ -147,6 +149,7 @@ class timer_actor final : public detail::proper_actor<blocking_actor, ...@@ -147,6 +149,7 @@ class timer_actor final : public detail::proper_actor<blocking_actor,
}; };
void printer_loop(blocking_actor* self) { void printer_loop(blocking_actor* self) {
self->trap_exit(true);
std::map<actor_addr, std::string> out; std::map<actor_addr, std::string> out;
auto flush_output = [&out](const actor_addr& s) { auto flush_output = [&out](const actor_addr& s) {
auto i = out.find(s); auto i = out.find(s);
...@@ -189,10 +192,15 @@ void printer_loop(blocking_actor* self) { ...@@ -189,10 +192,15 @@ void printer_loop(blocking_actor* self) {
flush_output(dm.source); flush_output(dm.source);
out.erase(dm.source); out.erase(dm.source);
}, },
on(atom("DIE")) >> [&] { running = false; }, others() >> [self] { [&](const exit_msg&) {
std::cerr << "*** unexpected: " << to_string(self->last_dequeued()) running = false;
},
others() >> [self] {
std::cerr << "*** unexpected: "
<< to_string(self->last_dequeued())
<< std::endl; << std::endl;
}); }
);
} }
} // namespace <anonymous> } // namespace <anonymous>
...@@ -273,7 +281,7 @@ void abstract_coordinator::stop() { ...@@ -273,7 +281,7 @@ void abstract_coordinator::stop() {
alive_workers.erase(i); alive_workers.erase(i);
} }
// shutdown utility actors // shutdown utility actors
CPPA_LOG_DEBUG("send 'DIE' messages to timer & printer"); CPPA_LOG_DEBUG("send exit messages to timer & printer");
anon_send_exit(m_timer->address(), exit_reason::user_shutdown); anon_send_exit(m_timer->address(), exit_reason::user_shutdown);
anon_send_exit(m_printer->address(), exit_reason::user_shutdown); anon_send_exit(m_printer->address(), exit_reason::user_shutdown);
CPPA_LOG_DEBUG("join threads of utility actors"); CPPA_LOG_DEBUG("join threads of utility actors");
......
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