Commit 696d995b authored by Dominik Charousset's avatar Dominik Charousset

Fix memory leak in BASP broker

parent 1e23f322
...@@ -27,6 +27,7 @@ ...@@ -27,6 +27,7 @@
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/actor_cast.hpp" #include "caf/actor_cast.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/exit_reason.hpp"
namespace caf { namespace caf {
...@@ -61,8 +62,31 @@ public: ...@@ -61,8 +62,31 @@ public:
/// addresses for remote actors on the fly if needed. /// addresses for remote actors on the fly if needed.
actor_addr read(deserializer* source); actor_addr read(deserializer* source);
/// Ensures that `kill_proxy` is called for each proxy instance.
class proxy_entry {
public:
proxy_entry();
proxy_entry(actor_proxy::anchor_ptr ptr);
proxy_entry(proxy_entry&&) = default;
proxy_entry& operator=(proxy_entry&&) = default;
~proxy_entry();
void reset(uint32_t rsn);
inline explicit operator bool() const {
return static_cast<bool>(ptr_);
}
inline actor_proxy::anchor* operator->() const {
return ptr_.get();
}
private:
actor_proxy::anchor_ptr ptr_;
};
/// A map that stores all proxies for known remote actors. /// A map that stores all proxies for known remote actors.
using proxy_map = std::map<actor_id, actor_proxy::anchor_ptr>; using proxy_map = std::map<actor_id, proxy_entry>;
/// Returns the number of proxies for `node`. /// Returns the number of proxies for `node`.
size_t count_proxies(const key_type& node); size_t count_proxies(const key_type& node);
...@@ -85,7 +109,8 @@ public: ...@@ -85,7 +109,8 @@ public:
void erase(const key_type& node); void erase(const key_type& node);
/// Deletes the proxy with id `aid` for `node`. /// Deletes the proxy with id `aid` for `node`.
void erase(const key_type& node, actor_id aid); void erase(const key_type& node, actor_id aid,
uint32_t rsn = exit_reason::remote_link_unreachable);
/// Queries whether there are any proxies left. /// Queries whether there are any proxies left.
bool empty() const; bool empty() const;
......
...@@ -36,6 +36,28 @@ actor_namespace::backend::~backend() { ...@@ -36,6 +36,28 @@ actor_namespace::backend::~backend() {
// nop // nop
} }
actor_namespace::proxy_entry::proxy_entry() {
// nop
}
actor_namespace::proxy_entry::proxy_entry(actor_proxy::anchor_ptr ptr)
: ptr_(std::move(ptr)) {
// nop
}
actor_namespace::proxy_entry::~proxy_entry() {
reset(exit_reason::remote_link_unreachable);
}
void actor_namespace::proxy_entry::reset(uint32_t rsn) {
if (! ptr_)
return;
auto ptr = ptr_->get();
ptr_.reset();
if (ptr)
ptr->kill_proxy(rsn);
}
actor_namespace::actor_namespace(backend& be) : backend_(be) { actor_namespace::actor_namespace(backend& be) : backend_(be) {
// nop // nop
} }
...@@ -159,15 +181,19 @@ void actor_namespace::erase(const key_type& inf) { ...@@ -159,15 +181,19 @@ void actor_namespace::erase(const key_type& inf) {
proxies_.erase(inf); proxies_.erase(inf);
} }
void actor_namespace::erase(const key_type& inf, actor_id aid) { void actor_namespace::erase(const key_type& inf, actor_id aid, uint32_t rsn) {
CAF_LOG_TRACE(CAF_TARG(inf, to_string) << ", " << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_TARG(inf, to_string) << ", " << CAF_ARG(aid));
auto i = proxies_.find(inf); auto i = proxies_.find(inf);
if (i != proxies_.end()) { if (i != proxies_.end()) {
i->second.erase(aid); auto& submap = i->second;
if (i->second.empty()) { auto j = submap.find(aid);
if (j == submap.end())
return;
j->second.reset(rsn);
submap.erase(j);
if (submap.empty())
proxies_.erase(i); proxies_.erase(i);
} }
}
} }
void actor_namespace::clear() { void actor_namespace::clear() {
......
...@@ -47,8 +47,6 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee { ...@@ -47,8 +47,6 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
// inherited from actor_namespace::backend // inherited from actor_namespace::backend
actor_proxy_ptr make_proxy(const node_id& nid, actor_id aid) override; actor_proxy_ptr make_proxy(const node_id& nid, actor_id aid) override;
void erase_proxy(const node_id& nid, actor_id aid);
// inherited from basp::instance::listener // inherited from basp::instance::listener
void finalize_handshake(const node_id& nid, actor_id aid, void finalize_handshake(const node_id& nid, actor_id aid,
const std::set<std::string>& sigs) override; const std::set<std::string>& sigs) override;
......
...@@ -73,12 +73,13 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid, ...@@ -73,12 +73,13 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid,
intrusive_ptr<basp_broker> ptr = static_cast<basp_broker*>(self); intrusive_ptr<basp_broker> ptr = static_cast<basp_broker*>(self);
auto mm = middleman::instance(); auto mm = middleman::instance();
auto res = make_counted<forwarding_actor_proxy>(aid, nid, ptr); auto res = make_counted<forwarding_actor_proxy>(aid, nid, ptr);
res->attach_functor([=](uint32_t) { res->attach_functor([=](uint32_t rsn) {
mm->backend().dispatch([=] { mm->backend().post([=] {
// using res->id() instead of aid keeps this actor instance alive // using res->id() instead of aid keeps this actor instance alive
// until the original instance terminates, thus preventing subtle // until the original instance terminates, thus preventing subtle
// bugs with attachables // bugs with attachables
ptr->state.erase_proxy(nid, res->id()); if (ptr->exit_reason() == exit_reason::not_exited)
ptr->state.get_namespace().erase(nid, aid, rsn);
}); });
}); });
// tell remote side we are monitoring this actor now // tell remote side we are monitoring this actor now
...@@ -91,11 +92,6 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid, ...@@ -91,11 +92,6 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid,
return res; return res;
} }
void basp_broker_state::erase_proxy(const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid));
get_namespace().erase(nid, aid);
}
void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid, void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
const std::set<std::string>& sigs) { const std::set<std::string>& sigs) {
CAF_LOG_TRACE(CAF_TSARG(nid) CAF_LOG_TRACE(CAF_TSARG(nid)
...@@ -146,10 +142,6 @@ void basp_broker_state::purge_state(const node_id& nid) { ...@@ -146,10 +142,6 @@ void basp_broker_state::purge_state(const node_id& nid) {
auto hdl = instance.tbl().lookup_direct(nid); auto hdl = instance.tbl().lookup_direct(nid);
if (hdl == invalid_connection_handle) if (hdl == invalid_connection_handle)
return; return;
// kill all proxies we have from this node
auto proxies = get_namespace().get_all(nid);
for (auto& p : proxies)
p->kill_proxy(exit_reason::remote_link_unreachable);
get_namespace().erase(nid); get_namespace().erase(nid);
ctx.erase(hdl); ctx.erase(hdl);
known_remotes.erase(nid); known_remotes.erase(nid);
...@@ -191,16 +183,9 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) { ...@@ -191,16 +183,9 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
void basp_broker_state::kill_proxy(const node_id& nid, actor_id aid, void basp_broker_state::kill_proxy(const node_id& nid, actor_id aid,
uint32_t rsn) { uint32_t rsn) {
CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid) << ", " << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid) << ", " << CAF_ARG(rsn));
auto ptr = get_namespace().get(nid, aid); get_namespace().erase(nid, aid, rsn);
if (! ptr) {
CAF_LOG_DEBUG("received kill proxy twice");
return;
}
get_namespace().erase(ptr->node(), ptr->id());
ptr->kill_proxy(static_cast<uint32_t>(rsn));
} }
void basp_broker_state::deliver(const node_id& source_node, void basp_broker_state::deliver(const node_id& source_node,
actor_id source_actor, actor_id source_actor,
const node_id& dest_node, const node_id& dest_node,
......
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