Commit 4658d45a authored by Joseph Noir's avatar Joseph Noir

Exchange reachable interfaces in handshake

parent 4d2ac56c
......@@ -309,6 +309,7 @@ public:
case message_type::server_handshake: {
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
basp::routing_table::address_map addrs;
if (!payload_valid()) {
CAF_LOG_ERROR("fail to receive the app identifier");
return false;
......@@ -322,7 +323,7 @@ public:
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
e = bd(aid, sigs);
e = bd(aid, sigs, addrs);
if (e)
return false;
}
......@@ -343,7 +344,8 @@ public:
}
// Add this node to our contacts.
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.add(hdl, hdr.source_node);
tbl_.add(hdr.source_node, hdl);
tbl_.addresses(hdr.source_node, addrs);
// TODO: Add addresses to share with other nodes?
// Write handshake as client in response.
if (tcp_based)
......@@ -356,6 +358,7 @@ public:
break;
}
case message_type::client_handshake: {
basp::routing_table::address_map addrs;
if (!payload_valid()) {
CAF_LOG_ERROR("fail to receive the app identifier");
return false;
......@@ -369,6 +372,9 @@ public:
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
e = bd(addrs);
if (e)
return false;
}
auto new_node = (this_node() != hdr.source_node
&& !tbl_.lookup(hdr.source_node).known);
......@@ -381,8 +387,8 @@ public:
} else {
// Add this node to our contacts.
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.add(hdl, hdr.source_node);
// TODO: Add addresses for future sharing of contact info.
tbl_.add(hdr.source_node, hdl);
tbl_.addresses(hdr.source_node, addrs);
}
// Since udp is unreliable we answer, maybe our message was lost.
if (!tcp_based) {
......
......@@ -55,6 +55,9 @@ namespace basp {
class routing_table {
public:
using endpoint_handle = variant<connection_handle, datagram_handle>;
using address_endpoint = std::pair<uint16_t, network::address_listing>;
using address_map = std::map<network::protocol::transport,
address_endpoint>;
explicit routing_table(abstract_broker* parent);
......@@ -82,9 +85,14 @@ public:
/// Adds a new endpoint to the table.
/// @pre `hdl != invalid_connection_handle && nid != none`
void add(const endpoint_handle& hdl, const node_id& nid);
void add(const node_id& nid, const endpoint_handle& hdl);
/// Adds a new node that is not reachable yet. It's state is set to `pending`.
/// Add a new endpoint to the table.
/// @pre `origin != none && nid != none`
void add(const node_id& nid, const node_id& origin);
/// Adds a new endpoint to the table that has no attached information.
/// that propagated information about the node.
/// @pre `nid != none`
void add(const node_id& nid);
......@@ -93,7 +101,7 @@ public:
/// including the node that is assigned as direct path for `hdl`.
void erase(const endpoint_handle& hdl, erase_callback& cb);
/// Queries whether `dest` is reachable.
/// Queries whether `dest` is reachable directly.
bool reachable(const node_id& dest);
/// Returns the parent broker.
......@@ -102,17 +110,35 @@ public:
}
/// Set the forwarding node that first mentioned `hdl`.
bool forwarder(const node_id& nid, endpoint_handle hdl);
bool origin(const node_id& nid, const node_id& origin);
/// Get the forwarding node that first mentioned `hdl`
/// or `none` if the node is unknown.
optional<endpoint_handle> forwarder(const node_id& nid);
optional<node_id> origin(const node_id& nid);
/// Set the handle for communication with `nid`.
bool handle(const node_id& nid, const endpoint_handle& hdl);
/// Get the handle for communication with `nid`
/// or `none` if the node is unknown.
optional<endpoint_handle> handle(const node_id& nid);
/// Add `addrs` to the addresses to reach `nid`.
bool addresses(const node_id& nid, network::address_listing addrs);
/// Set `addrs` as the addresses to reach `nid`.
bool addresses(const node_id& nid, address_map addrs);
/// Set `addrs` as the addresses to reach `nid` with `proto`.
bool addresses(const node_id& nid, network::protocol::transport proto,
address_endpoint addrs);
/// Get the addresses to reach `nid` or `none` if the node is unknown.
optional<const network::address_listing&> addresses(const node_id& nid);
optional<const address_map&> addresses(const node_id& nid);
/// Add a port, address pair under a key to the local addresses.
void local_addresses(network::protocol::transport proto,
address_endpoint addrs);
/// Get a reference to an address map for the local node.
const address_map& local_addresses();
public:
/// Entry to bundle information for a remote endpoint.
......@@ -120,9 +146,9 @@ public:
/// Handle for the node if communication is established.
optional<endpoint_handle> hdl;
/// Interfaces of the nodes for sharing with neighbors.
network::address_listing addrs;
address_map addrs;
/// The endpoint who told us about the node.
optional<endpoint_handle> origin;
optional<node_id> origin;
};
template <class Map, class Fallback>
......@@ -137,6 +163,7 @@ public:
abstract_broker* parent_;
std::unordered_map<endpoint_handle, node_id> nid_by_hdl_;
std::unordered_map<node_id, node_info> node_information_base_;
address_map local_addrs_;
};
/// @}
......
......@@ -325,11 +325,11 @@ void instance::write_server_handshake(execution_unit* ctx,
return e;
if (pa != nullptr) {
auto i = pa->first ? pa->first->id() : invalid_actor_id;
return sink(i, pa->second);
return sink(i, pa->second, tbl_.local_addresses());
}
auto aid = invalid_actor_id;
std::set<std::string> tmp;
return sink(aid, tmp);
return sink(aid, tmp, tbl_.local_addresses());
});
header hdr{message_type::server_handshake, 0, 0, version,
this_node_, none,
......@@ -346,7 +346,8 @@ void instance::write_client_handshake(execution_unit* ctx,
uint16_t sequence_number) {
CAF_LOG_TRACE(CAF_ARG(remote_side));
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<std::string&>(app_identifier));
return sink(const_cast<std::string&>(app_identifier),
tbl_.local_addresses());
});
header hdr{message_type::client_handshake, 0, 0, 0,
this_node, remote_side, invalid_actor_id, invalid_actor_id,
......
......@@ -397,6 +397,8 @@ void middleman::init(actor_system_config& cfg) {
cfg.add_message_type<network::protocol>("@protocol")
.add_message_type<network::address_listing>("@address_listing")
.add_message_type<network::receive_buffer>("@receive_buffer")
.add_message_type<basp::routing_table::address_endpoint>("@address_endpoint")
.add_message_type<basp::routing_table::address_map>("@address_map")
.add_message_type<new_data_msg>("@new_data_msg")
.add_message_type<new_connection_msg>("@new_connection_msg")
.add_message_type<acceptor_closed_msg>("@acceptor_closed_msg")
......
......@@ -50,22 +50,26 @@ void routing_table::erase(const endpoint_handle& hdl, erase_callback& cb) {
return;
cb(i->second);
parent_->parent().notify<hook::connection_lost>(i->second);
// TODO: Should we keep address information and set the state to 'none'?
node_information_base_.erase(i->second);
//hdl_by_nid_.erase(i->second);
nid_by_hdl_.erase(i->first);
// TODO: Look through other nodes and remove this one as an origin?
}
void routing_table::add(const endpoint_handle& hdl, const node_id& nid) {
void routing_table::add(const node_id& nid, const endpoint_handle& hdl) {
CAF_ASSERT(nid_by_hdl_.count(hdl) == 0);
//CAF_ASSERT(hdl_by_nid_.count(nid) == 0);
CAF_ASSERT(node_information_base_.count(nid) == 0);
nid_by_hdl_.emplace(hdl, nid);
//hdl_by_nid_.emplace(nid, hdl);
node_information_base_[nid] = node_info{hdl, {}, none};
parent_->parent().notify<hook::new_connection_established>(nid);
}
void routing_table::add(const node_id& nid, const node_id& origin) {
CAF_ASSERT(node_information_base_.count(nid) == 0);
node_information_base_[nid] = node_info{none, {}, origin};
// TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid);
}
void routing_table::add(const node_id& nid) {
//CAF_ASSERT(hdl_by_nid_.count(nid) == 0);
CAF_ASSERT(node_information_base_.count(nid) == 0);
......@@ -81,8 +85,8 @@ bool routing_table::reachable(const node_id& dest) {
return false;
}
bool routing_table::forwarder(const node_id& nid,
routing_table::endpoint_handle origin) {
bool routing_table::origin(const node_id& nid,
const node_id& origin) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
......@@ -90,27 +94,51 @@ bool routing_table::forwarder(const node_id& nid,
return true;
}
optional<routing_table::endpoint_handle>
routing_table::forwarder(const node_id& nid) {
optional<node_id>
routing_table::origin(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.origin;
}
bool routing_table::handle(const node_id& nid,
const routing_table::endpoint_handle& hdl) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.hdl = hdl;
return true;
}
optional<routing_table::endpoint_handle>
routing_table::handle(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return none;
return i->second.hdl;
}
bool routing_table::addresses(const node_id& nid,
network::address_listing addrs) {
routing_table::address_map addrs) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
for (auto& p : addrs) {
auto& existing = i->second.addrs[p.first];
existing.insert(existing.end(), p.second.begin(), p.second.end());
}
i->second.addrs = addrs;
return true;
}
optional<const network::address_listing&>
bool routing_table::addresses(const node_id& nid,
network::protocol::transport proto,
routing_table::address_endpoint addrs) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.addrs[proto] = addrs;
return true;
}
optional<const routing_table::address_map&>
routing_table::addresses(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
......@@ -118,6 +146,15 @@ routing_table::addresses(const node_id& nid) {
return i->second.addrs;
}
void routing_table::local_addresses(network::protocol::transport key,
routing_table::address_endpoint addrs) {
local_addrs_[key] = addrs;
}
const routing_table::address_map& routing_table::local_addresses() {
return local_addrs_;
}
} // namespace basp
} // namespace io
} // namespace caf
......@@ -291,13 +291,15 @@ public:
mock(hdl,
{basp::message_type::client_handshake, 0, 0, 0,
n.id, this_node(),
invalid_actor_id, invalid_actor_id}, std::string{})
invalid_actor_id, invalid_actor_id}, std::string{},
basp::routing_table::address_map{})
.receive(hdl,
basp::message_type::server_handshake, no_flags,
any_vals, basp::version, this_node(), node_id{none},
published_actor_id, invalid_actor_id, std::string{},
published_actor_id,
published_actor_ifs)
published_actor_ifs,
basp::routing_table::address_map{})
// upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node
.receive(hdl,
......@@ -509,7 +511,8 @@ CAF_TEST(non_empty_server_handshake) {
basp::version, this_node(), none,
self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr, std::string{},
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"});
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"},
basp::routing_table::address_map{});
CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf));
}
......@@ -600,11 +603,13 @@ CAF_TEST(remote_actor_and_send) {
jupiter().dummy_actor->id(), invalid_actor_id},
std::string{},
jupiter().dummy_actor->id(),
uint32_t{0})
uint32_t{0},
basp::routing_table::address_map{})
.receive(jupiter().connection,
basp::message_type::client_handshake, no_flags, 1u,
basp::message_type::client_handshake, no_flags, 2u,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, invalid_actor_id, std::string{})
invalid_actor_id, invalid_actor_id, std::string{},
basp::routing_table::address_map{})
.receive(jupiter().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
......
......@@ -289,14 +289,16 @@ public:
mock(src, hdl,
{basp::message_type::client_handshake, 0, 0, 0,
n.id, this_node(),
invalid_actor_id, invalid_actor_id}, std::string{})
invalid_actor_id, invalid_actor_id}, std::string{},
basp::routing_table::address_map{})
// upon receiving the client handshake, the server should answer
// with the server handshake and send the dispatch_message blow
.receive(hdl,
basp::message_type::server_handshake, no_flags,
any_vals, basp::version, this_node(), node_id{none},
published_actor_id, invalid_actor_id, std::string{},
published_actor_id, published_actor_ifs)
published_actor_id, published_actor_ifs,
basp::routing_table::address_map{})
// upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node
.receive(hdl,
......@@ -606,7 +608,8 @@ CAF_TEST(non_empty_server_handshake_udp) {
basp::version, this_node(), none,
self()->id(), invalid_actor_id, 0};
to_buf(expected_buf, expected, nullptr, std::string{},
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"});
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"},
basp::routing_table::address_map{});
CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf));
}
......@@ -695,9 +698,10 @@ CAF_TEST(remote_actor_and_send_udp) {
auto na = registry()->named_actors();
mock()
.receive(jupiter().endpoint,
basp::message_type::client_handshake, no_flags, 1u,
basp::message_type::client_handshake, no_flags, 2u,
no_operation_data, this_node(), node_id(),
invalid_actor_id, invalid_actor_id, std::string{});
invalid_actor_id, invalid_actor_id, std::string{},
basp::routing_table::address_map{});
mock(jupiter().endpoint,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().id, none,
......@@ -705,7 +709,8 @@ CAF_TEST(remote_actor_and_send_udp) {
0}, // sequence number, first message
std::string{},
jupiter().dummy_actor->id(),
uint32_t{0})
uint32_t{0},
basp::routing_table::address_map{})
.receive(jupiter().endpoint,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
......@@ -888,9 +893,10 @@ CAF_TEST(out_of_order_delivery_udp) {
auto na = registry()->named_actors();
mock()
.receive(jupiter().endpoint,
basp::message_type::client_handshake, no_flags, 1u,
basp::message_type::client_handshake, no_flags, 2u,
no_operation_data, this_node(), node_id(),
invalid_actor_id, invalid_actor_id, std::string{});
invalid_actor_id, invalid_actor_id, std::string{},
basp::routing_table::address_map{});
mock(jupiter().endpoint, jupiter().endpoint,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().id, none,
......@@ -898,7 +904,8 @@ CAF_TEST(out_of_order_delivery_udp) {
0}, // sequence number, first message
std::string{},
jupiter().dummy_actor->id(),
uint32_t{0})
uint32_t{0},
basp::routing_table::address_map{})
.receive(jupiter().endpoint,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
......
......@@ -54,7 +54,11 @@ struct fixture {
io::middleman& client_side_mm = client_side.middleman();
};
behavior make_pong_behavior() {
behavior make_pong_behavior(event_based_actor* self) {
self->set_exit_handler([=](exit_msg& m) {
CAF_MESSAGE("Pong received exit message.");
self->quit(m.reason);
});
return {
[](int val) -> int {
++val;
......
......@@ -55,7 +55,11 @@ struct fixture {
io::middleman& client_side_mm = client_side.middleman();
};
behavior make_pong_behavior() {
behavior make_pong_behavior(event_based_actor* self) {
self->set_exit_handler([=](exit_msg& m) {
CAF_MESSAGE("Pong received exit message.");
self->quit(m.reason);
});
return {
[](int val) -> int {
++val;
......
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