Commit c771157b authored by Dominik Charousset's avatar Dominik Charousset

More runtime checks in basp_broker plus nitpicks

parent 6cf26532
......@@ -302,6 +302,16 @@ class broker : public extend<local_actor>::
*/
void close(accept_handle handle);
/**
* Checks whether a connection for `handle` exists.
*/
bool valid(connection_handle handle);
/**
* Checks whether an acceptor for `handle` exists.
*/
bool valid(accept_handle handle);
class functor_based;
void launch(bool is_hidden, bool, execution_unit*);
......
......@@ -32,12 +32,12 @@ namespace io {
void unpublish_impl(const actor_addr& whom, uint16_t port, bool block_caller);
/**
* Unpublishes `whom` by closing `port`.
* Unpublishes `whom` by closing `port` or all assigned ports if `port == 0`.
* @param whom Actor that should be unpublished at `port`.
* @param port TCP port.
*/
template <class Handle>
void unpublish(const Handle& whom, uint16_t port) {
void unpublish(const Handle& whom, uint16_t port = 0) {
if (!whom) {
return;
}
......
......@@ -106,7 +106,7 @@ behavior basp_broker::make_behavior() {
if (j != m_ctx.end()) {
auto hd = j->second.handshake_data;
if (hd) {
send(hd->client, error_atom{}, hd->request_id,
send(hd->client, error_atom::value, hd->request_id,
"disconnect during handshake");
}
m_ctx.erase(j);
......@@ -181,7 +181,7 @@ behavior basp_broker::make_behavior() {
}
catch (std::exception&) {
CAF_LOG_DEBUG("failed to assign scribe from handle");
send(client, error_atom{}, request_id,
send(client, error_atom::value, request_id,
"failed to assign scribe from handle");
return;
}
......@@ -198,7 +198,8 @@ behavior basp_broker::make_behavior() {
[=](delete_atom, int64_t request_id, const actor_addr& whom, uint16_t port)
-> message {
if (whom == invalid_actor_addr) {
return make_message(error_atom::value, "whom == invalid_actor_addr");
return make_message(error_atom::value, request_id,
"whom == invalid_actor_addr");
}
auto ptr = actor_cast<abstract_actor_ptr>(whom);
if (port == 0) {
......@@ -584,14 +585,14 @@ basp_broker::handle_basp_header(connection_context& ctx,
+ iface_str;
}
// abort with error
send(hsclient, error_atom{}, hsid, std::move(error_msg));
send(hsclient, error_atom::value, hsid, std::move(error_msg));
return close_connection;
}
auto nid = hdr.source_node;
if (nid == node()) {
CAF_LOG_INFO("incoming connection from self: drop connection");
auto res = detail::singletons::get_actor_registry()->get(remote_aid);
send(hsclient, ok_atom{}, hsid, res->address());
send(hsclient, ok_atom::value, hsid, res->address());
ctx.handshake_data = none;
return close_connection;
}
......@@ -600,7 +601,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
<< " (re-use old one)");
auto proxy = m_namespace.get_or_put(nid, remote_aid);
// discard this peer; there's already an open connection
send(hsclient, ok_atom{}, hsid, proxy->address());
send(hsclient, ok_atom::value, hsid, proxy->address());
ctx.handshake_data = none;
return close_connection;
}
......@@ -610,7 +611,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
// prepare to receive messages
auto proxy = m_namespace.get_or_put(nid, remote_aid);
ctx.published_actor = proxy;
send(hsclient, ok_atom{}, hsid, proxy->address());
send(hsclient, ok_atom::value, hsid, proxy->address());
ctx.handshake_data = none;
parent().notify<hook::new_connection_established>(nid);
break;
......@@ -759,7 +760,6 @@ void basp_broker::add_published_actor(accept_handle hdl,
if (!ptr) {
return;
}
CAF_LOG_TRACE("");
m_acceptors.insert(std::make_pair(hdl, std::make_pair(ptr, port)));
m_open_ports.insert(std::make_pair(port, hdl));
ptr->attach_functor([port](abstract_actor* self, uint32_t) {
......@@ -771,13 +771,19 @@ void basp_broker::add_published_actor(accept_handle hdl,
}
bool basp_broker::remove_published_actor(const abstract_actor_ptr& whom) {
CAF_LOG_TRACE("");
CAF_REQUIRE(whom != nullptr);
size_t erased_elements = 0;
auto last = m_acceptors.end();
auto i = m_acceptors.begin();
while (i != last) {
if (i->second.first == whom) {
auto& kvp = i->second;
if (kvp.first == whom) {
CAF_REQUIRE(valid(i->first));
close(i->first);
m_open_ports.erase(i->second.second);
if (m_open_ports.erase(kvp.second) == 0) {
CAF_LOG_ERROR("inconsistent data: no open port for acceptor!");
}
i = m_acceptors.erase(i);
++erased_elements;
}
......@@ -791,24 +797,26 @@ bool basp_broker::remove_published_actor(const abstract_actor_ptr& whom) {
bool basp_broker::remove_published_actor(const abstract_actor_ptr& whom,
uint16_t port) {
CAF_LOG_TRACE("");
CAF_REQUIRE(whom != nullptr);
CAF_REQUIRE(port != 0);
auto i = m_open_ports.find(port);
if (i == m_open_ports.end()) {
return false;
}
CAF_REQUIRE(valid(i->second));
auto j = m_acceptors.find(i->second);
if (j == m_acceptors.end()) {
CAF_LOG_ERROR("accept handle for port " << port
<< " not found in m_published_actors");
m_open_ports.erase(i);
return false;
}
if (j->second.first != whom) {
CAF_LOG_INFO("port has been bound to a different actor already");
CAF_LOG_INFO("port has been bound to a different actor");
return false;
}
close(j->first);
close(i->second);
m_open_ports.erase(i);
m_acceptors.erase(j);
if (j == m_acceptors.end()) {
CAF_LOG_ERROR("inconsistent data: accept handle for port " << port
<< " not found in m_published_actors");
} else {
m_acceptors.erase(j);
}
return true;
}
......
......@@ -340,6 +340,14 @@ void broker::close_all() {
}
}
bool broker::valid(connection_handle handle) {
return m_scribes.count(handle) > 0;
}
bool broker::valid(accept_handle handle) {
return m_doormen.count(handle) > 0;
}
std::vector<connection_handle> broker::connections() const {
std::vector<connection_handle> result;
for (auto& kvp : m_scribes) {
......
......@@ -160,7 +160,8 @@ class middleman_actor_impl : public middleman_actor_base::base {
~middleman_actor_impl();
void on_exit() {
m_pending_requests.clear();
m_pending_gets.clear();
m_pending_deletes.clear();
m_broker = invalid_actor;
}
......@@ -173,6 +174,8 @@ class middleman_actor_impl : public middleman_actor_base::base {
using del_op_promise = typed_response_promise<del_op_result>;
using map_type = std::map<int64_t, response_promise>;
middleman_actor_base::behavior_type make_behavior() {
return {
[=](put_atom, const actor_addr& whom, uint16_t port,
......@@ -203,31 +206,38 @@ class middleman_actor_impl : public middleman_actor_base::base {
return del(whom, port);
},
[=](ok_atom ok, int64_t request_id) {
auto i = m_pending_requests.find(request_id);
if (i == m_pending_requests.end()) {
// not legal for get results
CAF_REQUIRE(m_pending_gets.count(request_id) == 0);
auto i = m_pending_deletes.find(request_id);
if (i == m_pending_deletes.end()) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(del_op_result{ok}.value);
m_pending_requests.erase(i);
m_pending_deletes.erase(i);
},
[=](ok_atom ok, int64_t request_id, actor_addr result) {
auto i = m_pending_requests.find(request_id);
if (i == m_pending_requests.end()) {
// not legal for delete results
CAF_REQUIRE(m_pending_deletes.count(request_id) == 0);
auto i = m_pending_gets.find(request_id);
if (i == m_pending_gets.end()) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(get_op_result{ok, result}.value);
m_pending_requests.erase(i);
m_pending_gets.erase(i);
},
[=](error_atom error, int64_t request_id, std::string& reason) {
auto i = m_pending_requests.find(request_id);
if (i == m_pending_requests.end()) {
auto fget = [&](response_promise& rp) {
rp.deliver(get_op_result{error, std::move(reason)}.value);
};
auto fdel = [&](response_promise& rp) {
rp.deliver(del_op_result{error, std::move(reason)}.value);
};
if (!finalize_request(m_pending_gets, request_id, fget)
&& !finalize_request(m_pending_deletes, request_id, fdel)) {
CAF_LOG_ERROR("invalid request id: " << request_id);
return;
}
i->second.deliver(get_op_result{error, std::move(reason)}.value);
m_pending_requests.erase(i);
}
};
}
......@@ -248,13 +258,13 @@ class middleman_actor_impl : public middleman_actor_base::base {
actual_port = res.second;
}
catch (bind_failure& err) {
return {error_atom{}, std::string("bind_failure: ") + err.what()};
return {error_atom::value, std::string("bind_failure: ") + err.what()};
}
catch (network_error& err) {
return {error_atom{}, std::string("network_error: ") + err.what()};
return {error_atom::value, std::string("network_error: ") + err.what()};
}
send(m_broker, put_atom{}, hdl, whom, actual_port);
return {ok_atom{}, actual_port};
send(m_broker, put_atom::value, hdl, whom, actual_port);
return {ok_atom::value, actual_port};
}
get_op_promise get(const std::string& hostname, uint16_t port,
......@@ -263,15 +273,15 @@ class middleman_actor_impl : public middleman_actor_base::base {
try {
auto hdl = m_parent.backend().new_tcp_scribe(hostname, port);
auto req_id = m_next_request_id++;
send(m_broker, get_atom{}, hdl, req_id,
send(m_broker, get_atom::value, hdl, req_id,
actor{this}, std::move(expected_ifs));
m_pending_requests.insert(std::make_pair(req_id, result));
m_pending_gets.insert(std::make_pair(req_id, result));
}
catch (network_error& err) {
// fullfil promise immediately
std::string msg = "network_error: ";
msg += err.what();
result.deliver(get_op_result{error_atom{}, std::move(msg)}.value);
result.deliver(get_op_result{error_atom::value, std::move(msg)}.value);
}
return result;
}
......@@ -280,14 +290,26 @@ class middleman_actor_impl : public middleman_actor_base::base {
auto result = make_response_promise();
auto req_id = m_next_request_id++;
send(m_broker, delete_atom::value, req_id, whom, port);
m_pending_requests.insert(std::make_pair(req_id, result));
m_pending_deletes.insert(std::make_pair(req_id, result));
return result;
}
template <class F>
bool finalize_request(map_type& storage, uint64_t req_id, F fun) {
auto i = storage.find(req_id);
if (i == storage.end()) {
return false;
}
fun(i->second);
storage.erase(i);
return true;
}
actor m_broker;
middleman& m_parent;
int64_t m_next_request_id;
std::map<int64_t, response_promise> m_pending_requests;
map_type m_pending_gets;
map_type m_pending_deletes;
};
middleman_actor_impl::~middleman_actor_impl() {
......
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