Commit 5658d0e7 authored by Dominik Charousset's avatar Dominik Charousset

Fix refcount of brokers and clear caches properly

parent 51067802
......@@ -91,6 +91,19 @@ class intrusive_partitioned_list {
m_tail.prev = &m_separator;
}
~intrusive_partitioned_list() {
clear();
}
void clear() {
while (!first_empty()) {
erase(first_begin());
}
while (!second_empty()) {
erase(second_begin());
}
}
iterator first_begin() {
return m_head.next;
}
......
......@@ -169,8 +169,8 @@ void deserialize_impl(group& gref, deserializer* source) {
}
void serialize_impl(const channel& chref, serializer* sink) {
// channel is an abstract base class that's either an actor or a group
// to indicate that, we write a flag first, that is
// abstract_channel is an abstract base class that's either an actor
// or a group; we prefix the serialized data using a flag:
// 0 if ptr == nullptr
// 1 if ptr points to an actor
// 2 if ptr points to a group
......
......@@ -42,10 +42,10 @@ namespace io {
*/
class basp_broker : public broker, public actor_namespace::backend {
public:
basp_broker();
basp_broker(middleman& parent_ref);
~basp_broker();
behavior make_behavior() override;
void add_published_actor(accept_handle hdl, const abstract_actor_ptr& whom,
......@@ -90,6 +90,9 @@ class basp_broker : public broker, public actor_namespace::backend {
return m_namespace;
}
protected:
void on_exit() override;
private:
void erase_proxy(const node_id& nid, actor_id aid);
......
......@@ -306,6 +306,8 @@ class broker : public extend<local_actor>::
void launch(bool is_hidden, bool, execution_unit*);
void cleanup(uint32_t reason);
// <backward_compatibility version="0.9">
static constexpr auto at_least = receive_policy_flag::at_least;
......@@ -326,10 +328,14 @@ class broker : public extend<local_actor>::
broker(middleman& parent_ref);
void cleanup(uint32_t reason);
virtual behavior make_behavior() = 0;
/**
* Can be overridden to perform cleanup code before the
* broker closes all its connections.
*/
virtual void on_exit();
/** @endcond */
inline middleman& parent() {
......
......@@ -67,7 +67,9 @@ class middleman : public detail::abstract_singleton {
if (i != m_named_brokers.end()) {
return static_cast<Impl*>(i->second.get());
}
intrusive_ptr<Impl> result{new Impl(*this)};
auto result = detail::make_counted<Impl>(*this);
CAF_REQUIRE(!result->unique());
result->deref(); // local_actor starts with ref count of 1
result->launch(true, false, nullptr);
m_named_brokers.insert(std::make_pair(name, result));
return result;
......
......@@ -68,16 +68,14 @@ functor_payload_writer<F> make_payload_writer(F fun) {
return {fun};
}
basp_broker::basp_broker() : m_namespace(*this) {
basp_broker::basp_broker(middleman& pref) : broker(pref), m_namespace(*this) {
m_meta_msg = uniform_typeid<message>();
m_meta_id_type = uniform_typeid<node_id>();
CAF_LOG_DEBUG("BASP broker started: " << to_string(node()));
}
basp_broker::basp_broker(middleman& pref) : broker(pref), m_namespace(*this) {
m_meta_msg = uniform_typeid<message>();
m_meta_id_type = uniform_typeid<node_id>();
CAF_LOG_DEBUG("BASP broker started: " << to_string(node()));
basp_broker::~basp_broker() {
// nop
}
behavior basp_broker::make_behavior() {
......@@ -178,7 +176,15 @@ behavior basp_broker::make_behavior() {
},
[=](get_atom, connection_handle hdl, int64_t request_id,
actor client, std::set<std::string>& expected_ifs) {
assign_tcp_scribe(hdl);
try {
assign_tcp_scribe(hdl);
}
catch (std::exception&) {
CAF_LOG_DEBUG("failed to assign scribe from handle");
send(client, error_atom{}, request_id,
"failed to assign scribe from handle");
return;
}
auto& ctx = m_ctx[hdl];
ctx.hdl = hdl;
// PODs are not movable, so passing expected_ifs to the ctor would cause
......@@ -676,6 +682,15 @@ actor_proxy_ptr basp_broker::make_proxy(const node_id& nid, actor_id aid) {
return res;
}
void basp_broker::on_exit() {
m_ctx.clear();
m_acceptors.clear();
m_open_ports.clear();
m_routes.clear();
m_blacklist.clear();
m_pending_requests.clear();
}
void basp_broker::erase_proxy(const node_id& nid, actor_id aid) {
CAF_LOGM_TRACE("make_behavior$_DelProxy",
CAF_TSARG(nid) << ", " << CAF_ARG(aid));
......
......@@ -136,6 +136,8 @@ class broker::continuation {
// nop
}
continuation(continuation&&) = default;
inline void operator()() {
CAF_PUSH_AID(m_self->id());
CAF_LOG_TRACE("");
......@@ -155,8 +157,7 @@ policy::invoke_message_result broker::invoke_message(mailbox_element_ptr& msg,
void broker::invoke_message(mailbox_element_ptr& ptr) {
CAF_LOG_TRACE(CAF_TARG(msg, to_string));
if (planned_exit_reason() != exit_reason::not_exited
|| bhvr_stack().empty()) {
if (exit_reason() != exit_reason::not_exited || bhvr_stack().empty()) {
CAF_LOG_DEBUG("actor already finished execution"
<< ", planned_exit_reason = " << planned_exit_reason()
<< ", bhvr_stack().empty() = " << bhvr_stack().empty());
......@@ -206,14 +207,10 @@ void broker::invoke_message(mailbox_element_ptr& ptr) {
// cleanup if needed
if (planned_exit_reason() != exit_reason::not_exited) {
cleanup(planned_exit_reason());
// release implicit reference count held by MM
// deref();
} else if (bhvr_stack().empty()) {
CAF_LOG_DEBUG("bhvr_stack().empty(), quit for normal exit reason");
quit(exit_reason::normal);
cleanup(planned_exit_reason());
// release implicit reference count held by MM
// deref();
}
}
......@@ -262,11 +259,21 @@ broker::broker(middleman& ptr) : m_mm(ptr) {
void broker::cleanup(uint32_t reason) {
CAF_LOG_TRACE(CAF_ARG(reason));
planned_exit_reason(reason);
on_exit();
close_all();
CAF_REQUIRE(m_doormen.empty());
CAF_REQUIRE(m_scribes.empty());
CAF_REQUIRE(current_mailbox_element() == nullptr);
m_cache.clear();
super::cleanup(reason);
deref(); // release implicit reference count from middleman
}
void broker::on_exit() {
// nop
}
void broker::launch(bool is_hidden, bool, execution_unit*) {
// add implicit reference count held by the middleman
ref();
......
......@@ -159,6 +159,11 @@ class middleman_actor_impl : public middleman_actor_base::base {
~middleman_actor_impl();
void on_exit() {
m_pending_requests.clear();
m_broker = invalid_actor;
}
using get_op_result = either<ok_atom, actor_addr>
::or_else<error_atom, std::string>;
......@@ -331,12 +336,10 @@ void middleman::stop() {
CAF_LOGC_TRACE("caf::io::middleman", "stop$lambda", "");
// m_managers will be modified while we are stopping each manager,
// because each manager will call remove(...)
std::vector<broker_ptr> brokers;
for (auto& kvp : m_named_brokers) {
brokers.push_back(kvp.second);
}
for (auto& bro : brokers) {
bro->close_all();
if (kvp.second->exit_reason() == exit_reason::not_exited) {
kvp.second->cleanup(exit_reason::normal);
}
}
});
m_backend_supervisor.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