Commit 5d10f7c7 authored by Dominik Charousset's avatar Dominik Charousset

Fix lifetime management of brokers

Solve an issue that caused brokers to live indefinitely, i.e., the destructor
of brokers was never called. Also, the runtime no longer emits
`connection_closed_msg` messages if a broker closes one of its sockets.
parent 6bd8dd63
...@@ -56,12 +56,8 @@ using broker_ptr = intrusive_ptr<broker>; ...@@ -56,12 +56,8 @@ using broker_ptr = intrusive_ptr<broker>;
class broker : public extend<local_actor>:: class broker : public extend<local_actor>::
with<mixin::behavior_stack_based<behavior>::impl>, with<mixin::behavior_stack_based<behavior>::impl>,
public spawn_as_is { public spawn_as_is {
friend class policy::sequential_invoke;
using super = combined_type;
public: public:
using super = combined_type;
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
...@@ -80,18 +76,18 @@ class broker : public extend<local_actor>:: ...@@ -80,18 +76,18 @@ class broker : public extend<local_actor>::
virtual message disconnect_message() = 0; virtual message disconnect_message() = 0;
inline broker* parent() { inline broker* parent() {
return m_broker; return m_broker.get();
} }
servant(broker* ptr); servant(broker* ptr);
void set_broker(broker* ptr); void set_broker(broker* ptr);
void disconnect(); void disconnect(bool invoke_disconnect_message);
bool m_disconnected; bool m_disconnected;
broker* m_broker; intrusive_ptr<broker> m_broker;
}; };
/** /**
...@@ -265,90 +261,6 @@ class broker : public extend<local_actor>:: ...@@ -265,90 +261,6 @@ class broker : public extend<local_actor>::
void invoke_message(const actor_addr& sender, message_id mid, message& msg); void invoke_message(const actor_addr& sender, message_id mid, message& msg);
/*
template <class Socket>
connection_handle add_connection(Socket sock) {
CAF_LOG_TRACE("");
class impl : public scribe {
public:
impl(broker* parent, Socket&& s)
: scribe(parent, network::conn_hdl_from_socket(s)),
m_launched(false),
m_stream(parent->backend()) {
m_stream.init(std::move(s));
}
void configure_read(receive_policy::config config) override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.configure_read(config);
if (!m_launched) launch();
}
buffer_type& wr_buf() override {
return m_stream.wr_buf();
}
buffer_type& rd_buf() override {
return m_stream.rd_buf();
}
void stop_reading() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.stop_reading();
disconnect();
}
void flush() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.flush(this);
}
void launch() {
CAF_LOGM_TRACE("caf::io::broker::scribe", "");
CAF_REQUIRE(!m_launched);
m_launched = true;
m_stream.start(this);
}
private:
bool m_launched;
network::stream<Socket> m_stream;
};
intrusive_ptr<impl> ptr{new impl{this, std::move(sock)}};
m_scribes.insert(std::make_pair(ptr->hdl(), ptr));
return ptr->hdl();
}
template <class SocketAcceptor>
accept_handle add_acceptor(SocketAcceptor sock) {
CAF_LOG_TRACE("sock.fd = " << sock.fd());
CAF_REQUIRE(sock.fd() != network::invalid_native_socket);
class impl : public doorman {
public:
impl(broker* parent, SocketAcceptor&& s)
: doorman(parent, network::accept_hdl_from_socket(s)),
m_acceptor(parent->backend()) {
m_acceptor.init(std::move(s));
}
void new_connection() override {
accept_msg().handle = m_broker->add_connection(
std::move(m_acceptor.accepted_socket()));
m_broker->invoke_message(invalid_actor_addr,
message_id::invalid,
m_accept_msg);
}
void stop_reading() override {
m_acceptor.stop_reading();
disconnect();
}
void launch() override {
m_acceptor.start(this);
}
private:
network::acceptor<SocketAcceptor> m_acceptor;
};
intrusive_ptr<impl> ptr{new impl{this, std::move(sock)}};
m_doormen.insert(std::make_pair(ptr->hdl(), ptr));
if (is_initialized()) {
ptr->launch();
}
return ptr->hdl();
}
*/
void enqueue(const actor_addr&, message_id, message, void enqueue(const actor_addr&, message_id, message,
execution_unit*) override; execution_unit*) override;
......
...@@ -49,6 +49,7 @@ behavior basp_broker::make_behavior() { ...@@ -49,6 +49,7 @@ behavior basp_broker::make_behavior() {
[=](new_data_msg& msg) { [=](new_data_msg& msg) {
CAF_LOGM_TRACE("make_behavior$new_data_msg", CAF_LOGM_TRACE("make_behavior$new_data_msg",
"handle = " << msg.handle.id()); "handle = " << msg.handle.id());
CAF_REQUIRE(m_ctx.count(msg.handle) > 0);
new_data(m_ctx[msg.handle], msg.buf); new_data(m_ctx[msg.handle], msg.buf);
}, },
// received from underlying broker implementation // received from underlying broker implementation
...@@ -66,6 +67,7 @@ behavior basp_broker::make_behavior() { ...@@ -66,6 +67,7 @@ behavior basp_broker::make_behavior() {
[=](const connection_closed_msg& msg) { [=](const connection_closed_msg& msg) {
CAF_LOGM_TRACE("make_behavior$connection_closed_msg", CAF_LOGM_TRACE("make_behavior$connection_closed_msg",
CAF_MARG(msg.handle, id)); CAF_MARG(msg.handle, id));
CAF_REQUIRE(m_ctx.count(msg.handle) > 0);
// purge handle from all routes // purge handle from all routes
std::vector<id_type> lost_connections; std::vector<id_type> lost_connections;
for (auto& kvp : m_routes) { for (auto& kvp : m_routes) {
...@@ -123,7 +125,8 @@ behavior basp_broker::make_behavior() { ...@@ -123,7 +125,8 @@ behavior basp_broker::make_behavior() {
} }
void basp_broker::new_data(connection_context& ctx, buffer_type& buf) { void basp_broker::new_data(connection_context& ctx, buffer_type& buf) {
CAF_LOG_TRACE(CAF_TARG(ctx.state, static_cast<int>)); CAF_LOG_TRACE(CAF_TARG(ctx.state, static_cast<int>) << ", "
CAF_MARG(ctx.hdl, id));
m_current_context = &ctx; m_current_context = &ctx;
connection_state next_state; connection_state next_state;
switch (ctx.state) { switch (ctx.state) {
...@@ -133,6 +136,7 @@ void basp_broker::new_data(connection_context& ctx, buffer_type& buf) { ...@@ -133,6 +136,7 @@ void basp_broker::new_data(connection_context& ctx, buffer_type& buf) {
if (!basp::valid(ctx.hdr)) { if (!basp::valid(ctx.hdr)) {
CAF_LOG_INFO("invalid broker message received"); CAF_LOG_INFO("invalid broker message received");
close(ctx.hdl); close(ctx.hdl);
m_ctx.erase(ctx.hdl);
return; return;
} }
next_state = handle_basp_header(ctx); next_state = handle_basp_header(ctx);
...@@ -146,6 +150,7 @@ void basp_broker::new_data(connection_context& ctx, buffer_type& buf) { ...@@ -146,6 +150,7 @@ void basp_broker::new_data(connection_context& ctx, buffer_type& buf) {
CAF_LOG_DEBUG("transition: " << ctx.state << " -> " << next_state); CAF_LOG_DEBUG("transition: " << ctx.state << " -> " << next_state);
if (next_state == close_connection) { if (next_state == close_connection) {
close(ctx.hdl); close(ctx.hdl);
m_ctx.erase(ctx.hdl);
return; return;
} }
ctx.state = next_state; ctx.state = next_state;
...@@ -352,7 +357,7 @@ basp_broker::handle_basp_header(connection_context& ctx, ...@@ -352,7 +357,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
} else { } else {
auto mm = middleman::instance(); auto mm = middleman::instance();
entry.first->attach_functor([=](uint32_t reason) { entry.first->attach_functor([=](uint32_t reason) {
mm->run_later([=] { mm->backend().dispatch([=] {
CAF_LOGM_TRACE("handle_basp_header$proxy_functor", CAF_LOGM_TRACE("handle_basp_header$proxy_functor",
CAF_ARG(reason)); CAF_ARG(reason));
auto bro = mm->get_named_broker<basp_broker>(atom("_BASP")); auto bro = mm->get_named_broker<basp_broker>(atom("_BASP"));
...@@ -548,7 +553,7 @@ actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) { ...@@ -548,7 +553,7 @@ actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) {
auto mm = middleman::instance(); auto mm = middleman::instance();
auto res = make_counted<remote_actor_proxy>(aid, nid, self); auto res = make_counted<remote_actor_proxy>(aid, nid, self);
res->attach_functor([=](uint32_t) { res->attach_functor([=](uint32_t) {
mm->run_later([=] { mm->backend().dispatch([=] {
// 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
......
...@@ -44,31 +44,20 @@ broker::servant::servant(broker* ptr) : m_disconnected(false), m_broker(ptr) { ...@@ -44,31 +44,20 @@ broker::servant::servant(broker* ptr) : m_disconnected(false), m_broker(ptr) {
} }
void broker::servant::set_broker(broker* new_broker) { void broker::servant::set_broker(broker* new_broker) {
if (!m_disconnected) m_broker = new_broker; if (!m_disconnected) {
m_broker = new_broker;
}
} }
void broker::servant::disconnect() { void broker::servant::disconnect(bool invoke_disconnect_message) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (!m_disconnected) { if (!m_disconnected) {
CAF_LOG_DEBUG("disconnect servant from broker"); CAF_LOG_DEBUG("disconnect servant from broker");
m_disconnected = true; m_disconnected = true;
remove_from_broker(); remove_from_broker();
if (m_broker->exit_reason() == exit_reason::not_exited) { if (invoke_disconnect_message) {
if (m_broker->is_running()) { auto msg = disconnect_message();
CAF_LOG_DEBUG("broker is running, push message to cache"); m_broker->invoke_message(m_broker->address(),message_id::invalid, msg);
// push this message to the cache to make sure we
// don't have interleaved message handlers
auto e = mailbox_element::create(m_broker->address(),
message_id::invalid,
disconnect_message());
m_broker->m_priority_policy
.push_to_cache(unique_mailbox_element_pointer{e});
}
else {
CAF_LOG_DEBUG("broker is not running, invoke handler");
m_broker->enqueue(m_broker->address(), message_id::invalid,
disconnect_message(), nullptr);
}
} }
} }
} }
...@@ -108,7 +97,7 @@ void broker::scribe::io_failure(network::operation op) { ...@@ -108,7 +97,7 @@ void broker::scribe::io_failure(network::operation op) {
<< ", " << CAF_TARG(op, static_cast<int>)); << ", " << CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging // keep compiler happy when compiling w/o logging
static_cast<void>(op); static_cast<void>(op);
disconnect(); disconnect(true);
} }
broker::doorman::doorman(broker* parent, accept_handle hdl) broker::doorman::doorman(broker* parent, accept_handle hdl)
...@@ -135,7 +124,7 @@ void broker::doorman::io_failure(network::operation op) { ...@@ -135,7 +124,7 @@ void broker::doorman::io_failure(network::operation op) {
<< CAF_TARG(op, static_cast<int>)); << CAF_TARG(op, static_cast<int>));
// keep compiler happy when compiling w/o logging // keep compiler happy when compiling w/o logging
static_cast<void>(op); static_cast<void>(op);
disconnect(); disconnect(true);
} }
class broker::continuation { class broker::continuation {
...@@ -164,8 +153,6 @@ class broker::continuation { ...@@ -164,8 +153,6 @@ class broker::continuation {
void broker::invoke_message(const actor_addr& sender, message_id mid, void broker::invoke_message(const actor_addr& sender, message_id mid,
message& msg) { message& msg) {
CAF_LOG_TRACE(CAF_TARG(msg, to_string)); CAF_LOG_TRACE(CAF_TARG(msg, to_string));
is_running(true);
auto sg = detail::make_scope_guard([=] { is_running(false); });
if (planned_exit_reason() != exit_reason::not_exited if (planned_exit_reason() != exit_reason::not_exited
|| bhvr_stack().empty()) { || bhvr_stack().empty()) {
CAF_LOG_DEBUG("actor already finished execution" CAF_LOG_DEBUG("actor already finished execution"
...@@ -264,7 +251,7 @@ void broker::write(connection_handle hdl, size_t bs, const void* buf) { ...@@ -264,7 +251,7 @@ void broker::write(connection_handle hdl, size_t bs, const void* buf) {
void broker::enqueue(const actor_addr& sender, message_id mid, message msg, void broker::enqueue(const actor_addr& sender, message_id mid, message msg,
execution_unit*) { execution_unit*) {
middleman::instance()->run_later(continuation{this, sender, parent().backend().post(continuation{this, sender,
mid, std::move(msg)}); mid, std::move(msg)});
} }
...@@ -276,32 +263,34 @@ void broker::cleanup(uint32_t reason) { ...@@ -276,32 +263,34 @@ void broker::cleanup(uint32_t reason) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
close_all(); close_all();
super::cleanup(reason); super::cleanup(reason);
deref(); // release implicit reference count from middleman
} }
void broker::launch(bool is_hidden, execution_unit*) { void broker::launch(bool is_hidden, execution_unit*) {
// add implicit reference count held by the middleman
ref();
is_registered(!is_hidden); is_registered(!is_hidden);
CAF_PUSH_AID(id()); CAF_PUSH_AID(id());
CAF_LOGF_TRACE("init and launch broker with id " << id()); CAF_LOGF_TRACE("init and launch broker with ID " << id());
// we want to make sure initialization is executed in MM context // we want to make sure initialization is executed in MM context
broker_ptr self = this; become(
self->become( on(atom("INITMSG")) >> [=] {
on(atom("INITMSG")) >> [self] { CAF_LOGF_TRACE("ID " << id());
CAF_LOGF_TRACE(CAF_MARG(self, get)); unbecome();
self->unbecome();
// launch backends now, because user-defined initialization // launch backends now, because user-defined initialization
// might call functions like add_connection // might call functions like add_connection
for (auto& kvp : self->m_doormen) { for (auto& kvp : m_doormen) {
kvp.second->launch(); kvp.second->launch();
} }
self->is_initialized(true); is_initialized(true);
// run user-defined initialization code // run user-defined initialization code
auto bhvr = self->make_behavior(); auto bhvr = make_behavior();
if (bhvr) { if (bhvr) {
self->become(std::move(bhvr)); become(std::move(bhvr));
} }
} }
); );
self->enqueue(invalid_actor_addr, message_id::invalid, enqueue(invalid_actor_addr, message_id::invalid,
make_message(atom("INITMSG")), nullptr); make_message(atom("INITMSG")), nullptr);
} }
......
...@@ -686,7 +686,7 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self, ...@@ -686,7 +686,7 @@ connection_handle default_multiplexer::add_tcp_scribe(broker* self,
void stop_reading() override { void stop_reading() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", ""); CAF_LOGM_TRACE("caf::io::broker::scribe", "");
m_stream.stop_reading(); m_stream.stop_reading();
disconnect(); disconnect(false);
} }
void flush() override { void flush() override {
CAF_LOGM_TRACE("caf::io::broker::scribe", ""); CAF_LOGM_TRACE("caf::io::broker::scribe", "");
...@@ -728,7 +728,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self, ...@@ -728,7 +728,7 @@ accept_handle default_multiplexer::add_tcp_doorman(broker* self,
} }
void stop_reading() override { void stop_reading() override {
m_acceptor.stop_reading(); m_acceptor.stop_reading();
disconnect(); disconnect(false);
} }
void launch() override { void launch() override {
m_acceptor.start(this); m_acceptor.start(this);
......
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