You need to sign in or sign up before continuing.
Commit 1664eb97 authored by Dominik Charousset's avatar Dominik Charousset

Remove all mappings for stream managers when done

parent a89cb0ef
...@@ -1111,9 +1111,12 @@ public: ...@@ -1111,9 +1111,12 @@ public:
/// needed. /// needed.
bool add_stream_manager(stream_slot id, stream_manager_ptr ptr); bool add_stream_manager(stream_slot id, stream_manager_ptr ptr);
/// Removes a stream manager. /// Removes the stream manager mapped to `id` in `O(log n)`.
void erase_stream_manager(stream_slot id); void erase_stream_manager(stream_slot id);
/// Removes all entries for `mgr` in `O(n)`.
void erase_stream_manager(const stream_manager_ptr& mgr);
/// @pre `x.content().match_elements<open_stream_msg>()` /// @pre `x.content().match_elements<open_stream_msg>()`
invoke_message_result handle_open_stream_msg(mailbox_element& x); invoke_message_result handle_open_stream_msg(mailbox_element& x);
......
...@@ -294,14 +294,16 @@ struct downstream_msg_visitor { ...@@ -294,14 +294,16 @@ struct downstream_msg_visitor {
inptr.reset(); inptr.reset();
qs_ref.erase_later(dm.slots.receiver); qs_ref.erase_later(dm.slots.receiver);
selfptr->erase_stream_manager(dm.slots.receiver); selfptr->erase_stream_manager(dm.slots.receiver);
if (mgr->done()) if (mgr->done()) {
selfptr->erase_stream_manager(mgr);
mgr->stop(); mgr->stop();
}
return intrusive::task_result::stop; return intrusive::task_result::stop;
} }
else if (mgr->done()) { else if (mgr->done()) {
CAF_LOG_DEBUG("path is done receiving and closes its manager"); CAF_LOG_DEBUG("path is done receiving and closes its manager");
selfptr->erase_stream_manager(mgr);
mgr->stop(); mgr->stop();
selfptr->erase_stream_manager(dm.slots.receiver);
return intrusive::task_result::stop; return intrusive::task_result::stop;
} }
return intrusive::task_result::resume; return intrusive::task_result::resume;
...@@ -933,6 +935,18 @@ void scheduled_actor::erase_stream_manager(stream_slot id) { ...@@ -933,6 +935,18 @@ void scheduled_actor::erase_stream_manager(stream_slot id) {
stream_ticks_.stop(); stream_ticks_.stop();
} }
void scheduled_actor::erase_stream_manager(const stream_manager_ptr& mgr) {
auto i = stream_managers_.begin();
auto e = stream_managers_.end();
while (i != e)
if (i->second == mgr)
i = stream_managers_.erase(i);
else
++i;
if (stream_managers_.empty())
stream_ticks_.stop();
}
invoke_message_result invoke_message_result
scheduled_actor::handle_open_stream_msg(mailbox_element& x) { scheduled_actor::handle_open_stream_msg(mailbox_element& x) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
......
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