Commit 4b358e63 authored by Dominik Charousset's avatar Dominik Charousset

Fix state cleanup in BASP broker

BASP needs to release all state kept for connection handles, because the
OS re-uses socket IDs as soon as they are closed. Since the connection
handle is the socket ID this leads to inconsitent state.
parent b89c7db8
...@@ -117,10 +117,6 @@ public: ...@@ -117,10 +117,6 @@ public:
/// Sends heartbeat messages to all valid nodes those are directly connected. /// Sends heartbeat messages to all valid nodes those are directly connected.
void handle_heartbeat(execution_unit* ctx); void handle_heartbeat(execution_unit* ctx);
/// Handles failure or shutdown of a single node. This function purges
/// all routes to `affected_node` from the routing table.
void handle_node_shutdown(const node_id& affected_node);
/// Returns a route to `target` or `none` on error. /// Returns a route to `target` or `none` on error.
optional<routing_table::route> lookup(const node_id& target); optional<routing_table::route> lookup(const node_id& target);
......
...@@ -107,8 +107,12 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -107,8 +107,12 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
optional<response_promise> callback; optional<response_promise> callback;
}; };
/// Sets `this_context` by either creating or accessing state for `hdl`.
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
/// Cleans up any state for `hdl`.
void cleanup(connection_handle hdl);
// pointer to ourselves // pointer to ourselves
broker* self; broker* self;
......
...@@ -121,52 +121,31 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid, ...@@ -121,52 +121,31 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
CAF_ASSERT(this_context != nullptr); CAF_ASSERT(this_context != nullptr);
this_context->id = nid; this_context->id = nid;
auto& cb = this_context->callback; auto& cb = this_context->callback;
if (!cb) if (cb == none)
return; return;
auto cleanup = detail::make_scope_guard([&] {
cb = none;
});
strong_actor_ptr ptr; strong_actor_ptr ptr;
if (aid == invalid_actor_id) { // aid can be invalid when connecting to the default port of a node
// can occur when connecting to the default port of a node if (aid != invalid_actor_id) {
cb->deliver(nid, ptr, std::move(sigs)); if (nid == this_node()) {
return; // connected to self
} ptr = actor_cast<strong_actor_ptr>(system().registry().get(aid));
if (nid == this_node()) { CAF_LOG_INFO_IF(!ptr, "actor not found:" << CAF_ARG(aid));
// connected to self } else {
ptr = actor_cast<strong_actor_ptr>(system().registry().get(aid)); ptr = namespace_.get_or_put(nid, aid);
CAF_LOG_INFO_IF(!ptr, "actor not found:" << CAF_ARG(aid)); CAF_LOG_ERROR_IF(!ptr, "creating actor in finalize_handshake failed");
} else { }
ptr = namespace_.get_or_put(nid, aid);
CAF_LOG_ERROR_IF(!ptr, "creating actor in finalize_handshake failed");
} }
cb->deliver(make_message(nid, ptr, std::move(sigs))); cb->deliver(nid, std::move(ptr), std::move(sigs));
this_context->callback = none; cb = none;
} }
void basp_broker_state::purge_state(const node_id& nid) { void basp_broker_state::purge_state(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(nid)); CAF_LOG_TRACE(CAF_ARG(nid));
// We can only be sure a node is down when we had a direct connection.
auto hdl = instance.tbl().lookup_direct(nid);
if (hdl == invalid_connection_handle)
return;
// Cleanup state in case we lost connection during handshake.
auto i = ctx.find(hdl);
if (i != ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
ctx.erase(i);
}
// Destroy all proxies of the lost node. // Destroy all proxies of the lost node.
namespace_.erase(nid); namespace_.erase(nid);
// Cleanup all possible remaining references to the lost node. // Cleanup all remaining references to the lost node.
for (auto& kvp : monitored_actors) for (auto& kvp : monitored_actors)
kvp.second.erase(nid); kvp.second.erase(nid);
// Close network connection.
self->close(hdl);
} }
void basp_broker_state::send_kill_proxy_instance(const node_id& nid, void basp_broker_state::send_kill_proxy_instance(const node_id& nid,
...@@ -486,6 +465,29 @@ void basp_broker_state::set_context(connection_handle hdl) { ...@@ -486,6 +465,29 @@ void basp_broker_state::set_context(connection_handle hdl) {
this_context = &i->second; this_context = &i->second;
} }
void basp_broker_state::cleanup(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
// Remove handle from the routing table and clean up any node-specific state
// we might still have.
auto cb = make_callback([&](const node_id& nid) -> error {
purge_state(nid);
return none;
});
instance.tbl().erase_direct(hdl, cb);
// Remove the context for `hdl`, making sure clients receive an error in case
// this connection was closed during handshake.
auto i = ctx.find(hdl);
if (i != ctx.end()) {
auto& ref = i->second;
CAF_ASSERT(i->first == ref.hdl);
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
ctx.erase(i);
}
}
/****************************************************************************** /******************************************************************************
* basp_broker * * basp_broker *
******************************************************************************/ ******************************************************************************/
...@@ -529,9 +531,7 @@ behavior basp_broker::make_behavior() { ...@@ -529,9 +531,7 @@ behavior basp_broker::make_behavior() {
auto next = state.instance.handle(context(), msg, ctx.hdr, auto next = state.instance.handle(context(), msg, ctx.hdr,
ctx.cstate == basp::await_payload); ctx.cstate == basp::await_payload);
if (next == basp::close_connection) { if (next == basp::close_connection) {
// The function `instance::handle` calls purge_state. All state has state.cleanup(msg.handle);
// been cleaned up at this stage. In particular, accessing this_context
// results in a heap-use-after-free error.
close(msg.handle); close(msg.handle);
return; return;
} }
...@@ -606,30 +606,7 @@ behavior basp_broker::make_behavior() { ...@@ -606,30 +606,7 @@ behavior basp_broker::make_behavior() {
// received from underlying broker implementation // received from underlying broker implementation
[=](const connection_closed_msg& msg) { [=](const connection_closed_msg& msg) {
CAF_LOG_TRACE(CAF_ARG(msg.handle)); CAF_LOG_TRACE(CAF_ARG(msg.handle));
// TODO: currently we assume a node has gone offline once we lose state.cleanup(msg.handle);
// a connection, we also could try to reach this node via other
// hops to be resilient to (rare) network failures or if a
// node is reachable via several interfaces and only one fails
auto nid = state.instance.tbl().lookup_direct(msg.handle);
if (nid != none) {
// tell BASP instance we've lost connection
state.instance.handle_node_shutdown(nid);
} else {
// check whether the connection failed during handshake
auto pred = [&](const basp_broker_state::ctx_map::value_type& x) {
return x.second.hdl == msg.handle;
};
auto e = state.ctx.end();
auto i = std::find_if(state.ctx.begin(), e, pred);
if (i != e) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state.ctx.erase(i);
}
}
}, },
// received from underlying broker implementation // received from underlying broker implementation
[=](const acceptor_closed_msg& msg) { [=](const acceptor_closed_msg& msg) {
......
...@@ -902,8 +902,10 @@ event_handler::event_handler(default_multiplexer& dm, native_socket sockfd) ...@@ -902,8 +902,10 @@ event_handler::event_handler(default_multiplexer& dm, native_socket sockfd)
} }
event_handler::~event_handler() { event_handler::~event_handler() {
if (fd_ != invalid_native_socket) if (fd_ != invalid_native_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
closesocket(fd_); closesocket(fd_);
}
} }
void event_handler::close_read_channel() { void event_handler::close_read_channel() {
...@@ -1024,6 +1026,7 @@ void stream::stop_reading() { ...@@ -1024,6 +1026,7 @@ void stream::stop_reading() {
} }
void stream::removed_from_loop(operation op) { void stream::removed_from_loop(operation op) {
CAF_LOG_TRACE(CAF_ARG(op));
switch (op) { switch (op) {
case operation::read: reader_.reset(); break; case operation::read: reader_.reset(); break;
case operation::write: writer_.reset(); break; case operation::write: writer_.reset(); break;
...@@ -1120,6 +1123,7 @@ public: ...@@ -1120,6 +1123,7 @@ public:
void close() { void close() {
if (fd_ != invalid_native_socket) { if (fd_ != invalid_native_socket) {
CAF_LOG_DEBUG("close socket" << CAF_ARG(fd_));
closesocket(fd_); closesocket(fd_);
fd_ = invalid_native_socket; fd_ = invalid_native_socket;
} }
......
...@@ -63,7 +63,8 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -63,7 +63,8 @@ connection_state instance::handle(execution_unit* ctx,
if (is_payload) { if (is_payload) {
payload = &dm.buf; payload = &dm.buf;
if (payload->size() != hdr.payload_len) { if (payload->size() != hdr.payload_len) {
CAF_LOG_WARNING("received invalid payload"); CAF_LOG_WARNING("received invalid payload, expected"
<< hdr.payload_len << "bytes, got" << payload->size());
return err(); return err();
} }
} else { } else {
...@@ -139,14 +140,14 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -139,14 +140,14 @@ connection_state instance::handle(execution_unit* ctx,
if (hdr.source_node == this_node_) { if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately"); CAF_LOG_INFO("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err(); return close_connection;
} }
// close this connection if we already have a direct connection // close this connection if we already have a direct connection
if (tbl_.lookup_direct(hdr.source_node) != invalid_connection_handle) { if (tbl_.lookup_direct(hdr.source_node) != invalid_connection_handle) {
CAF_LOG_INFO("close connection since we already have a " CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node)); "direct connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err(); return close_connection;
} }
// add direct route to this node and remove any indirect entry // add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node)); CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
...@@ -262,18 +263,6 @@ void instance::handle_heartbeat(execution_unit* ctx) { ...@@ -262,18 +263,6 @@ void instance::handle_heartbeat(execution_unit* ctx) {
} }
} }
void instance::handle_node_shutdown(const node_id& affected_node) {
CAF_LOG_TRACE(CAF_ARG(affected_node));
if (affected_node == none)
return;
CAF_LOG_INFO("lost direct connection:" << CAF_ARG(affected_node));
auto cb = make_callback([&](const node_id& nid) -> error {
callee_.purge_state(nid);
return none;
});
tbl_.erase(affected_node, cb);
}
optional<routing_table::route> instance::lookup(const node_id& target) { optional<routing_table::route> instance::lookup(const node_id& target) {
return tbl_.lookup(target); return tbl_.lookup(target);
} }
......
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