Commit d32b78d0 authored by Dominik Charousset's avatar Dominik Charousset

Fix leaks in group and I/O modules, relates #454

parent 827149d9
...@@ -115,7 +115,7 @@ public: ...@@ -115,7 +115,7 @@ public:
// nop // nop
} }
static data* create_singleton(); static intrusive_ptr<data> create_singleton();
int compare(const node_id& other) const; int compare(const node_id& other) const;
......
...@@ -131,6 +131,7 @@ actor_system::actor_system(actor_system_config&& cfg) ...@@ -131,6 +131,7 @@ actor_system::actor_system(actor_system_config&& cfg)
for (auto& mod : modules_) for (auto& mod : modules_)
if (mod) if (mod)
mod->start(); mod->start();
groups_.start();
// store config parameters in ConfigServ // store config parameters in ConfigServ
auto cs = actor_cast<actor>(registry_.get(atom("ConfigServ"))); auto cs = actor_cast<actor>(registry_.get(atom("ConfigServ")));
anon_send(cs, put_atom::value, "middleman.enable-automatic-connections", anon_send(cs, put_atom::value, "middleman.enable-automatic-connections",
...@@ -140,6 +141,7 @@ actor_system::actor_system(actor_system_config&& cfg) ...@@ -140,6 +141,7 @@ actor_system::actor_system(actor_system_config&& cfg)
actor_system::~actor_system() { actor_system::~actor_system() {
if (await_actors_before_shutdown_) if (await_actors_before_shutdown_)
await_all_actors_done(); await_all_actors_done();
groups_.stop();
// stop modules in reverse order // stop modules in reverse order
for (auto i = modules_.rbegin(); i != modules_.rend(); ++i) for (auto i = modules_.rbegin(); i != modules_.rend(); ++i)
if (*i) if (*i)
......
...@@ -99,6 +99,7 @@ local_actor::local_actor(int init_flags) ...@@ -99,6 +99,7 @@ local_actor::local_actor(int init_flags)
} }
local_actor::~local_actor() { local_actor::~local_actor() {
CAF_LOG_TRACE(CAF_ARG(planned_exit_reason()));
if (planned_exit_reason() == exit_reason::not_exited) if (planned_exit_reason() == exit_reason::not_exited)
cleanup(exit_reason::unreachable, nullptr); cleanup(exit_reason::unreachable, nullptr);
} }
......
...@@ -141,7 +141,7 @@ std::atomic<uint8_t> system_id; ...@@ -141,7 +141,7 @@ std::atomic<uint8_t> system_id;
} // <anonymous> } // <anonymous>
// initializes singleton // initializes singleton
node_id::data* node_id::data::create_singleton() { intrusive_ptr<node_id::data> node_id::data::create_singleton() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto ifs = detail::get_mac_addresses(); auto ifs = detail::get_mac_addresses();
std::vector<std::string> macs; std::vector<std::string> macs;
...@@ -157,7 +157,9 @@ node_id::data* node_id::data::create_singleton() { ...@@ -157,7 +157,9 @@ node_id::data* node_id::data::create_singleton() {
// by overriding the last byte in the node ID with the actor system "ID" // by overriding the last byte in the node ID with the actor system "ID"
nid.back() = system_id.fetch_add(1); nid.back() = system_id.fetch_add(1);
// note: pointer has a ref count of 1 -> implicitly held by detail::singletons // note: pointer has a ref count of 1 -> implicitly held by detail::singletons
return new node_id::data(detail::get_process_id(), nid); intrusive_ptr<data> result;
result.reset(new node_id::data(detail::get_process_id(), nid), false);
return result;
} }
uint32_t node_id::process_id() const { uint32_t node_id::process_id() const {
......
...@@ -300,6 +300,7 @@ void asio_multiplexer::exec_later(resumable* rptr) { ...@@ -300,6 +300,7 @@ void asio_multiplexer::exec_later(resumable* rptr) {
exec_later(ptr.get()); exec_later(ptr.get());
break; break;
case resumable::done: case resumable::done:
case resumable::awaiting_message:
intrusive_ptr_release(ptr.get()); intrusive_ptr_release(ptr.get());
break; break;
default: default:
......
...@@ -44,9 +44,7 @@ public: ...@@ -44,9 +44,7 @@ public:
void set_parent(abstract_broker* ptr); void set_parent(abstract_broker* ptr);
/// Returns the parent broker of this manager. /// Returns the parent broker of this manager.
inline abstract_broker* parent() { abstract_broker* parent();
return parent_;
}
/// Returns `true` if this manager has a parent, `false` otherwise. /// Returns `true` if this manager has a parent, `false` otherwise.
inline bool detached() const { inline bool detached() const {
...@@ -78,7 +76,7 @@ protected: ...@@ -78,7 +76,7 @@ protected:
virtual void detach_from(abstract_broker* ptr) = 0; virtual void detach_from(abstract_broker* ptr) = 0;
private: private:
abstract_broker* parent_; strong_actor_ptr parent_;
}; };
} // namespace network } // namespace network
......
...@@ -42,41 +42,19 @@ void abstract_broker::enqueue(strong_actor_ptr src, message_id mid, ...@@ -42,41 +42,19 @@ void abstract_broker::enqueue(strong_actor_ptr src, message_id mid,
void abstract_broker::enqueue(mailbox_element_ptr ptr, execution_unit*) { void abstract_broker::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_PUSH_AID(id()); CAF_PUSH_AID(id());
CAF_LOG_TRACE("enqueue " << CAF_ARG(ptr->msg)); local_actor::enqueue(std::move(ptr), &backend());
auto mid = ptr->mid;
auto sender = ptr->sender;
switch (mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: {
// re-schedule broker
CAF_LOG_DEBUG("unblocked_reader");
backend().exec_later(this);
break;
}
case detail::enqueue_result::queue_closed: {
CAF_LOG_DEBUG("queue_closed");
if (mid.is_request()) {
detail::sync_request_bouncer f{exit_reason()};
f(sender, mid);
}
break;
}
case detail::enqueue_result::success:
// enqueued to a running actors' mailbox; nothing to do
CAF_LOG_DEBUG("success");
break;
}
} }
void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) { void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) {
CAF_ASSERT(eu != nullptr); CAF_ASSERT(eu != nullptr);
CAF_ASSERT(eu == &backend()); CAF_ASSERT(eu == &backend());
// add implicit reference count held by middleman/multiplexer // add implicit reference count held by middleman/multiplexer
intrusive_ptr_add_ref(ctrl());
is_registered(! is_hidden); is_registered(! is_hidden);
CAF_PUSH_AID(id()); CAF_PUSH_AID(id());
CAF_LOG_TRACE("init and launch broker:" << CAF_ARG(id())); CAF_LOG_TRACE("init and launch broker:" << CAF_ARG(id()));
if (is_lazy && mailbox().try_block()) if (is_lazy && mailbox().try_block())
return; return;
intrusive_ptr_add_ref(ctrl());
eu->exec_later(this); eu->exec_later(this);
} }
......
...@@ -978,16 +978,17 @@ void pipe_reader::handle_event(operation op) { ...@@ -978,16 +978,17 @@ void pipe_reader::handle_event(operation op) {
CAF_LOG_TRACE(CAF_ARG(op)); CAF_LOG_TRACE(CAF_ARG(op));
switch (op) { switch (op) {
case operation::read: { case operation::read: {
auto cb = try_read_next(); auto cb = try_read_next();
switch (cb->resume(&backend(), backend().max_throughput())) { switch (cb->resume(&backend(), backend().max_throughput())) {
case resumable::resume_later: case resumable::resume_later:
backend().exec_later(cb); backend().exec_later(cb);
break; break;
case resumable::done: case resumable::done:
case resumable::awaiting_message:
intrusive_ptr_release(cb); intrusive_ptr_release(cb);
break; break;
default: default:
; // ignored break; // ignored
} }
break; break;
} }
......
...@@ -27,7 +27,7 @@ namespace caf { ...@@ -27,7 +27,7 @@ namespace caf {
namespace io { namespace io {
namespace network { namespace network {
manager::manager(abstract_broker* ptr) : parent_(ptr) { manager::manager(abstract_broker* ptr) : parent_(ptr->ctrl()) {
// nop // nop
} }
...@@ -37,7 +37,11 @@ manager::~manager() { ...@@ -37,7 +37,11 @@ manager::~manager() {
void manager::set_parent(abstract_broker* ptr) { void manager::set_parent(abstract_broker* ptr) {
if (! detached()) if (! detached())
parent_ = ptr; parent_ = ptr ? ptr->ctrl() : nullptr;
}
abstract_broker* manager::parent() {
return parent_ ? static_cast<abstract_broker*>(parent_->get()) : nullptr;
} }
void manager::detach(execution_unit* ctx, bool invoke_disconnect_message) { void manager::detach(execution_unit* ctx, bool invoke_disconnect_message) {
......
...@@ -301,7 +301,6 @@ void middleman::stop() { ...@@ -301,7 +301,6 @@ void middleman::stop() {
ptr->context(&backend()); ptr->context(&backend());
ptr->planned_exit_reason(exit_reason::normal); ptr->planned_exit_reason(exit_reason::normal);
ptr->finished(); ptr->finished();
//intrusive_ptr_release(ptr);
} }
} }
}); });
......
...@@ -33,7 +33,9 @@ test_multiplexer::test_multiplexer(actor_system* sys) : multiplexer(sys) { ...@@ -33,7 +33,9 @@ test_multiplexer::test_multiplexer(actor_system* sys) : multiplexer(sys) {
} }
test_multiplexer::~test_multiplexer() { test_multiplexer::~test_multiplexer() {
// nop // get rid of extra ref count
for (auto& ptr : resumables_)
intrusive_ptr_release(ptr.get());
} }
connection_handle connection_handle
...@@ -393,6 +395,7 @@ void test_multiplexer::exec(resumable_ptr& ptr) { ...@@ -393,6 +395,7 @@ void test_multiplexer::exec(resumable_ptr& ptr) {
exec_later(ptr.get()); exec_later(ptr.get());
break; break;
case resumable::done: case resumable::done:
case resumable::awaiting_message:
intrusive_ptr_release(ptr.get()); intrusive_ptr_release(ptr.get());
break; break;
default: default:
......
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