Commit 0b60118b authored by Dominik Charousset's avatar Dominik Charousset

Fix shadowing warnings + coding style nitpicks

parent 057b889a
......@@ -41,30 +41,23 @@ namespace io {
* A broker implementation for the Binary Actor System Protocol (BASP).
*/
class basp_broker : public broker, public actor_namespace::backend {
using super = broker;
public:
using id_type = node_id;
basp_broker();
basp_broker(middleman& parent_ref);
behavior make_behavior() override;
void add_published_actor(accept_handle hdl,
const abstract_actor_ptr& whom,
void add_published_actor(accept_handle hdl, const abstract_actor_ptr& whom,
uint16_t port);
void remove_published_actor(const abstract_actor_ptr& whom, uint16_t port);
actor_proxy_ptr make_proxy(const id_type&, actor_id) override;
actor_proxy_ptr make_proxy(const node_id&, actor_id) override;
// dispatches a message from a local actor to a remote node
void dispatch(const actor_addr& from, const actor_addr& to,
message_id mid, const message& msg);
message_id mid, const message& msg);
struct client_handshake_data {
int64_t request_id;
......@@ -77,8 +70,7 @@ class basp_broker : public broker, public actor_namespace::backend {
}
private:
void erase_proxy(const id_type& nid, actor_id aid);
void erase_proxy(const node_id& nid, actor_id aid);
// dispatches a message from a remote node to a local actor
void dispatch(const basp::header& msg, message&& payload);
......@@ -99,7 +91,7 @@ class basp_broker : public broker, public actor_namespace::backend {
struct connection_context {
connection_state state;
connection_handle hdl;
id_type remote_id;
node_id remote_id;
optional<client_handshake_data> handshake_data;
basp::header hdr;
// keep a reference to the published actor of
......@@ -114,7 +106,7 @@ class basp_broker : public broker, public actor_namespace::backend {
void write(binary_serializer& bs, const basp::header& msg);
void send_kill_proxy_instance(const id_type& nid, actor_id aid,
void send_kill_proxy_instance(const node_id& nid, actor_id aid,
uint32_t reason);
connection_state handle_basp_header(connection_context& ctx,
......@@ -130,14 +122,14 @@ class basp_broker : public broker, public actor_namespace::backend {
void init_handshake_as_client(connection_context& ctx);
void init_handshake_as_server(connection_context& ctx,
actor_addr published_actor);
actor_addr published_actor);
void serialize_msg(const actor_addr& sender, message_id mid,
const message& msg, buffer_type& wr_buf);
bool try_set_default_route(const id_type& nid, connection_handle hdl);
bool try_set_default_route(const node_id& nid, connection_handle hdl);
void add_route(const id_type& nid, connection_handle hdl);
void add_route(const node_id& nid, connection_handle hdl);
struct connection_info {
connection_handle hdl;
......@@ -153,7 +145,7 @@ class basp_broker : public broker, public actor_namespace::backend {
}
};
connection_info get_route(const id_type& dest);
connection_info get_route(const node_id& dest);
struct connection_info_less {
inline bool operator()(const connection_info& lhs,
......@@ -166,7 +158,7 @@ class basp_broker : public broker, public actor_namespace::backend {
}
};
using blacklist_entry = std::pair<id_type, connection_handle>;
using blacklist_entry = std::pair<node_id, connection_handle>;
// (default route, [alternative routes])
using routing_table_entry = std::pair<connection_info,
......@@ -183,7 +175,7 @@ class basp_broker : public broker, public actor_namespace::backend {
};
// dest => hops
using routing_table = std::map<id_type, routing_table_entry>;
using routing_table = std::map<node_id, routing_table_entry>;
// sender => request ID
using pending_request = std::pair<actor_addr, message_id>;
......@@ -194,9 +186,8 @@ class basp_broker : public broker, public actor_namespace::backend {
std::map<uint16_t, accept_handle> m_open_ports;
routing_table m_routes; // stores non-direct routes
std::set<blacklist_entry, blacklist_less> m_blacklist; // stores invalidated
// routes
// routes
std::set<pending_request> m_pending_requests;
//std::map<id_type, connection_handle> m_nodes;
// needed to keep track to which node we are talking to at the moment
connection_context* m_current_context;
......@@ -205,7 +196,6 @@ class basp_broker : public broker, public actor_namespace::backend {
const uniform_type_info* m_meta_hdr;
const uniform_type_info* m_meta_msg;
const uniform_type_info* m_meta_id_type;
};
} // namespace io
......
......@@ -86,7 +86,7 @@ behavior basp_broker::make_behavior() {
m_ctx.erase(j);
}
// purge handle from all routes
std::vector<id_type> lost_connections;
std::vector<node_id> lost_connections;
for (auto& kvp : m_routes) {
auto& entry = kvp.second;
if (entry.first.hdl == msg.handle) {
......@@ -135,27 +135,30 @@ behavior basp_broker::make_behavior() {
CAF_LOGM_TRACE("make_behavior$_Dispatch", "");
dispatch(sender, receiver, mid, msg);
},
on(atom("_DelProxy"), arg_match) >> [=](const id_type& nid, actor_id aid) {
on(atom("_DelProxy"), arg_match) >> [=](const node_id& nid, actor_id aid) {
CAF_LOGM_TRACE("make_behavior$_DelProxy",
CAF_TSARG(nid) << ", "
<< CAF_ARG(aid));
erase_proxy(nid, aid);
},
// received from middleman actor
[=](put_atom, network::native_socket fd, const actor_addr& whom, uint16_t port) {
[=](put_atom, network::native_socket fd,
const actor_addr& whom, uint16_t port) {
auto hdl = add_tcp_doorman(fd);
add_published_actor(hdl, actor_cast<abstract_actor_ptr>(whom), port);
parent().notify<hook::actor_published>(whom, port);
},
[=](get_atom, network::native_socket fd, int64_t request_id, actor client, std::set<std::string>& expected_ifs) {
[=](get_atom, network::native_socket fd, int64_t request_id,
actor client, std::set<std::string>& expected_ifs) {
auto hdl = add_tcp_scribe(fd);
auto& ctx = m_ctx[hdl];
ctx.hdl = hdl;
ctx.handshake_data = client_handshake_data{};
auto& hdata = *ctx.handshake_data;
hdata.request_id = request_id;
hdata.client = client;
hdata.expected_ifs.swap(expected_ifs);
// PODs are not movable, so passing expected_ifs to the ctor would cause
// a copy; we avoid this by calling the ctor with an empty set and
// swap afterwards with expected_ifs
ctx.handshake_data = client_handshake_data{request_id, client,
std::set<std::string>()};
ctx.handshake_data->expected_ifs.swap(expected_ifs);
init_handshake_as_client(ctx);
},
// catch-all error handler
......@@ -539,7 +542,7 @@ basp_broker::handle_basp_header(connection_context& ctx,
return await_header;
}
void basp_broker::send_kill_proxy_instance(const id_type& nid, actor_id aid,
void basp_broker::send_kill_proxy_instance(const node_id& nid, actor_id aid,
uint32_t reason) {
CAF_LOG_TRACE(CAF_TSARG(nid) << ", " << CAF_ARG(aid) << CAF_ARG(reason));
auto route = get_route(nid);
......@@ -556,7 +559,7 @@ void basp_broker::send_kill_proxy_instance(const id_type& nid, actor_id aid,
flush(route.hdl);
}
basp_broker::connection_info basp_broker::get_route(const id_type& dest) {
basp_broker::connection_info basp_broker::get_route(const node_id& dest) {
connection_info res;
auto i = m_routes.find(dest);
if (i != m_routes.end()) {
......@@ -569,7 +572,7 @@ basp_broker::connection_info basp_broker::get_route(const id_type& dest) {
return res;
}
actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) {
actor_proxy_ptr basp_broker::make_proxy(const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_TSARG(nid) << ", "
<< CAF_ARG(aid));
CAF_REQUIRE(m_current_context != nullptr);
......@@ -607,13 +610,13 @@ actor_proxy_ptr basp_broker::make_proxy(const id_type& nid, actor_id aid) {
// tell remote side we are monitoring this actor now
binary_serializer bs(std::back_inserter(wr_buf(route.hdl)), &m_namespace);
write(bs, {node(), nid, invalid_actor_id, aid,
0, basp::announce_proxy_instance, 0});
0, basp::announce_proxy_instance, 0});
// run hooks
parent().notify<hook::new_remote_actor>(res->address());
return res;
}
void basp_broker::erase_proxy(const id_type& nid, actor_id aid) {
void basp_broker::erase_proxy(const node_id& nid, actor_id aid) {
CAF_LOGM_TRACE("make_behavior$_DelProxy",
CAF_TSARG(nid) << ", " << CAF_ARG(aid));
m_namespace.erase(nid, aid);
......@@ -622,14 +625,14 @@ void basp_broker::erase_proxy(const id_type& nid, actor_id aid) {
}
}
void basp_broker::add_route(const id_type& nid, connection_handle hdl) {
void basp_broker::add_route(const node_id& nid, connection_handle hdl) {
if (m_blacklist.count(std::make_pair(nid, hdl)) == 0) {
parent().notify<hook::new_route_added>(m_current_context->remote_id, nid);
m_routes[nid].second.insert({hdl, nid});
}
}
bool basp_broker::try_set_default_route(const id_type& nid,
bool basp_broker::try_set_default_route(const node_id& nid,
connection_handle hdl) {
CAF_REQUIRE(!hdl.invalid());
auto& entry = m_routes[nid];
......@@ -649,7 +652,7 @@ void basp_broker::init_handshake_as_client(connection_context& ctx) {
}
void basp_broker::init_handshake_as_server(connection_context& ctx,
actor_addr addr) {
actor_addr addr) {
CAF_LOG_TRACE(CAF_ARG(this));
CAF_REQUIRE(node() != invalid_node_id);
auto& buf = wr_buf(ctx.hdl);
......
......@@ -191,7 +191,8 @@ class middleman_actor_impl : public middleman_actor_base::base {
~middleman_actor_impl();
using get_op_result = either<ok_atom, actor_addr>::or_else<error_atom, std::string>;
using get_op_result = either<ok_atom, actor_addr>
::or_else<error_atom, std::string>;
using get_op_promise = typed_response_promise<get_op_result>;
......@@ -211,12 +212,12 @@ class middleman_actor_impl : public middleman_actor_base::base {
[=](put_atom, const actor_addr& whom, uint16_t port) {
return put(whom, port);
},
[=](get_atom, const std::string& host, uint16_t port,
[=](get_atom, const std::string& hostname, uint16_t port,
std::set<std::string>& expected_ifs) {
return get(host, port, std::move(expected_ifs));
return get(hostname, port, std::move(expected_ifs));
},
[=](get_atom, const std::string& host, uint16_t port) {
return get(host, port, std::set<std::string>());
[=](get_atom, const std::string& hostname, uint16_t port) {
return get(hostname, port, std::set<std::string>());
},
[=](ok_atom ok, int64_t request_id, actor_addr result) {
auto i = m_pending_requests.find(request_id);
......@@ -264,13 +265,14 @@ class middleman_actor_impl : public middleman_actor_base::base {
return {ok_atom{}, actual_port};
}
get_op_promise get(const std::string& host, uint16_t port,
get_op_promise get(const std::string& hostname, uint16_t port,
std::set<std::string> expected_ifs) {
get_op_promise result = make_response_promise();
try {
auto fd = network::new_ipv4_connection_impl(host, port);
auto fd = network::new_ipv4_connection_impl(hostname, port);
auto req_id = m_next_request_id++;
send(m_broker, get_atom{}, fd, req_id, actor{this}, std::move(expected_ifs));
send(m_broker, get_atom{}, fd, req_id,
actor{this}, std::move(expected_ifs));
m_pending_requests.insert(std::make_pair(req_id, result));
}
catch (network_error& err) {
......
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