Commit 9e8037e8 authored by Dominik Charousset's avatar Dominik Charousset

Clean up proxy killing in BASP

parent adc11fec
......@@ -63,11 +63,6 @@ public:
/// for one of our local actors.
virtual void proxy_announced(const node_id& nid, actor_id aid) = 0;
/// Called whenever a remote actor died to destroy
/// the proxy instance on our end.
virtual void kill_proxy(const node_id& nid, actor_id aid,
const error& rsn) = 0;
/// Called for each `dispatch_message` without `named_receiver_flag`.
virtual void deliver(const node_id& source_node, actor_id source_actor,
actor_id dest_actor, message_id mid,
......
......@@ -62,9 +62,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::listener
void proxy_announced(const node_id& nid, actor_id aid) override;
// inherited from basp::instance::listener
void kill_proxy(const node_id& nid, actor_id aid, const error& rsn) override;
// inherited from basp::instance::listener
void deliver(const node_id& src_nid, actor_id src_aid,
actor_id dest_aid, message_id mid,
......
......@@ -159,7 +159,7 @@ void basp_broker_state::purge_state(const node_id& nid) {
}
ctx.erase(i);
}
proxies().erase(nid);
self->close(hdl);
}
void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
......@@ -200,12 +200,6 @@ 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,
const error& rsn) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn));
proxies().erase(nid, aid, rsn);
}
void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
actor_id dest_aid, message_id mid,
std::vector<strong_actor_ptr>& stages,
......@@ -520,14 +514,13 @@ behavior basp_broker::make_behavior() {
auto next = state.instance.handle(context(), msg, ctx.hdr,
ctx.cstate == basp::await_payload);
if (next == basp::close_connection) {
if (ctx.callback) {
CAF_LOG_WARNING("failed to handshake with remote node"
<< CAF_ARG(msg.handle));
ctx.callback->deliver(make_error(sec::disconnect_during_handshake));
}
// The function `instance::handle` calls purge_state. All state has
// been cleaned up at this stage. In particular, accessing this_context
// results in a heap-use-after-free error.
close(msg.handle);
state.ctx.erase(msg.handle);
} else if (next != ctx.cstate) {
return;
}
if (next != ctx.cstate) {
auto rd_size = next == basp::await_payload
? ctx.hdr.payload_len
: basp::header_size;
......
......@@ -237,7 +237,8 @@ connection_state instance::handle(execution_unit* ctx,
auto e = bd(fail_state);
if (e)
return err();
callee_.kill_proxy(hdr.source_node, hdr.source_actor, fail_state);
callee_.proxies().erase(hdr.source_node, hdr.source_actor,
std::move(fail_state));
break;
}
case message_type::heartbeat: {
......
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