Commit adfafe67 authored by Dominik Charousset's avatar Dominik Charousset

Fix handling of stream manager references

parent 3d32db6c
...@@ -778,13 +778,15 @@ public: ...@@ -778,13 +778,15 @@ public:
return; return;
} }
CAF_ASSERT(i->second != nullptr); CAF_ASSERT(i->second != nullptr);
i->second->handle(slots, x); auto ptr = i->second;
if (i->second->done()) { ptr->handle(slots, x);
CAF_LOG_INFO("done sending:" << CAF_ARG(slots)); if (ptr->done()) {
i->second->stop(); CAF_LOG_DEBUG("done sending:" << CAF_ARG(slots));
stream_managers_.erase(i); ptr->stop();
if (stream_managers_.empty()) erase_stream_manager(ptr);
stream_ticks_.stop(); } else if (ptr->out().path(slots.receiver) == nullptr) {
CAF_LOG_DEBUG("done sending on path:" << CAF_ARG(slots.receiver));
erase_stream_manager(slots.receiver);
} }
} }
......
...@@ -48,9 +48,10 @@ void monitorable_actor::attach(attachable_ptr ptr) { ...@@ -48,9 +48,10 @@ void monitorable_actor::attach(attachable_ptr ptr) {
attach_impl(ptr); attach_impl(ptr);
return true; return true;
}); });
CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately"); if (!attached) {
if (!attached) CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately");
ptr->actor_exited(fail_state, nullptr); ptr->actor_exited(fail_state, nullptr);
}
} }
size_t monitorable_actor::detach(const attachable::token& what) { size_t monitorable_actor::detach(const attachable::token& what) {
......
...@@ -823,7 +823,6 @@ bool scheduled_actor::finalize() { ...@@ -823,7 +823,6 @@ bool scheduled_actor::finalize() {
// the terminated flag. // the terminated flag.
if (alive()) if (alive())
return false; return false;
setf(is_terminated_flag);
CAF_LOG_DEBUG("actor has no behavior and is ready for cleanup"); CAF_LOG_DEBUG("actor has no behavior and is ready for cleanup");
CAF_ASSERT(!has_behavior()); CAF_ASSERT(!has_behavior());
on_exit(); on_exit();
...@@ -1018,6 +1017,7 @@ void scheduled_actor::erase_stream_manager(stream_slot id) { ...@@ -1018,6 +1017,7 @@ void scheduled_actor::erase_stream_manager(stream_slot id) {
CAF_LOG_TRACE(CAF_ARG(id)); CAF_LOG_TRACE(CAF_ARG(id));
if (stream_managers_.erase(id) != 0 && stream_managers_.empty()) if (stream_managers_.erase(id) != 0 && stream_managers_.empty())
stream_ticks_.stop(); stream_ticks_.stop();
CAF_LOG_DEBUG(CAF_ARG2("stream_managers_.size", stream_managers_.size()));
} }
void scheduled_actor::erase_pending_stream_manager(stream_slot id) { void scheduled_actor::erase_pending_stream_manager(stream_slot id) {
...@@ -1026,7 +1026,8 @@ void scheduled_actor::erase_pending_stream_manager(stream_slot id) { ...@@ -1026,7 +1026,8 @@ void scheduled_actor::erase_pending_stream_manager(stream_slot id) {
} }
void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
{ // Lifetime scope of first iterator pair. CAF_LOG_TRACE("");
if (!stream_managers_.empty()) {
auto i = stream_managers_.begin(); auto i = stream_managers_.begin();
auto e = stream_managers_.end(); auto e = stream_managers_.end();
while (i != e) while (i != e)
...@@ -1034,6 +1035,8 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { ...@@ -1034,6 +1035,8 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
i = stream_managers_.erase(i); i = stream_managers_.erase(i);
else else
++i; ++i;
if (stream_managers_.empty())
stream_ticks_.stop();
} }
{ // Lifetime scope of second iterator pair. { // Lifetime scope of second iterator pair.
auto i = pending_stream_managers_.begin(); auto i = pending_stream_managers_.begin();
...@@ -1044,8 +1047,9 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) { ...@@ -1044,8 +1047,9 @@ void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
else else
++i; ++i;
} }
if (stream_managers_.empty()) CAF_LOG_DEBUG(CAF_ARG2("stream_managers_.size", stream_managers_.size())
stream_ticks_.stop(); << CAF_ARG2("pending_stream_managers_.size",
pending_stream_managers_.size()));
} }
invoke_message_result invoke_message_result
......
...@@ -127,6 +127,9 @@ void stream_manager::shutdown() { ...@@ -127,6 +127,9 @@ void stream_manager::shutdown() {
if (shutting_down()) if (shutting_down())
return; return;
flags_ = is_shutting_down_flag; flags_ = is_shutting_down_flag;
CAF_LOG_DEBUG("emit shutdown messages on" << inbound_paths_.size()
<< "inbound paths;" << CAF_ARG2("out.clean", out().clean())
<< CAF_ARG2("out.paths", out().num_paths()));
for (auto ipath : inbound_paths_) for (auto ipath : inbound_paths_)
ipath->emit_regular_shutdown(self_); ipath->emit_regular_shutdown(self_);
} }
......
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