Commit adc4041d authored by Joseph Noir's avatar Joseph Noir

Add peer addr cache and initial addr exchange

parent 67b7161c
...@@ -123,12 +123,15 @@ public: ...@@ -123,12 +123,15 @@ public:
friend class abstract_actor; friend class abstract_actor;
/// The number of actors implictly spawned by the actor system on startup. /// The number of actors implictly spawned by the actor system on startup.
static constexpr size_t num_internal_actors = 2; static constexpr size_t num_internal_actors = 3;
/// Returns the ID of an internal actor by its name. /// Returns the ID of an internal actor by its name.
/// @pre x in {'SpawnServ', 'ConfigServ', 'StreamServ'} /// @pre x in {'SpawnServ', 'ConfigServ', 'PeerServ', 'StreamServ'}
static constexpr size_t internal_actor_id(atom_value x) { static constexpr size_t internal_actor_id(atom_value x) {
return x == atom("SpawnServ") ? 0 : (x == atom("ConfigServ") ? 1 : 2); return x == atom("SpawnServ") ? 0
: (x == atom("ConfigServ") ? 1
: (x == atom("PeerServ") ? 2
: 3));
} }
/// Returns the internal actor for dynamic spawn operations. /// Returns the internal actor for dynamic spawn operations.
...@@ -142,6 +145,11 @@ public: ...@@ -142,6 +145,11 @@ public:
return internal_actors_[internal_actor_id(atom("ConfigServ"))]; return internal_actors_[internal_actor_id(atom("ConfigServ"))];
} }
/// Returns the internal actor for storing the addresses of its peers.
inline const strong_actor_ptr& peer_serv() const {
return internal_actors_[internal_actor_id(atom("PeerServ"))];
}
actor_system() = delete; actor_system() = delete;
actor_system(const actor_system&) = delete; actor_system(const actor_system&) = delete;
actor_system& operator=(const actor_system&) = delete; actor_system& operator=(const actor_system&) = delete;
...@@ -569,6 +577,11 @@ private: ...@@ -569,6 +577,11 @@ private:
internal_actors_[internal_actor_id(atom("ConfigServ"))] = std::move(x); internal_actors_[internal_actor_id(atom("ConfigServ"))] = std::move(x);
} }
/// Sets the internal actor for storing the peer addresses.
inline void peer_serv(strong_actor_ptr x) {
internal_actors_[internal_actor_id(atom("PeerServ"))] = std::move(x);
}
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Used to generate ascending actor IDs. /// Used to generate ascending actor IDs.
......
...@@ -166,6 +166,93 @@ behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) { ...@@ -166,6 +166,93 @@ behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) {
}; };
} }
// -- peer server --------------------------------------------------------------
// A peer server keeps track of the addresses to reach its peers. All addresses
// for a given node are stored under the string representation of its node id.
// When an entry is requested that does not exist, the requester is subscribed
// to the key. When the entry is set, it get a copy and is removed from the
// subscribers.
struct peer_state {
using key_type = std::string;
using mapped_type = message;
using subscriber_set = std::unordered_set<strong_actor_ptr>;
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<strong_actor_ptr, topic_set> subscribers;
static const char* name;
};
const char* peer_state::name = "peer_server";
behavior peer_serv_impl(stateful_actor<peer_state>* self) {
CAF_LOG_TRACE("");
std::string wildcard = "*";
auto unsubscribe_all = [=](actor subscriber) {
auto& subscribers = self->state.subscribers;
auto ptr = actor_cast<strong_actor_ptr>(subscriber);
auto i = subscribers.find(ptr);
if (i == subscribers.end())
return;
for (auto& key : i->second)
self->state.data[key].second.erase(ptr);
subscribers.erase(i);
};
self->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
auto ptr = actor_cast<strong_actor_ptr>(dm.source);
if (ptr)
unsubscribe_all(actor_cast<actor>(std::move(ptr)));
});
return {
// Set a key/value pair.
[=](put_atom, const std::string& key, message& msg) {
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(msg));
if (key == wildcard || key.empty())
return;
auto& vp = self->state.data[key];
vp.first = std::move(msg);
for (auto& subscriber_ptr : vp.second) {
// We never put a nullptr in our map.
auto subscriber = actor_cast<actor>(subscriber_ptr);
if (subscriber != self->current_sender()) {
self->send(subscriber, key, vp.first);
self->state.subscribers[subscriber_ptr].erase(key);
}
}
self->state.data[key].second.clear();
},
// Get a key/value pair.
[=](get_atom, std::string& key) {
auto sender = actor_cast<strong_actor_ptr>(self->current_sender());
if (sender) {
CAF_LOG_TRACE(CAF_ARG(key));
// Get the value ...
if (key == wildcard || key.empty())
return;
auto d = self->state.data.find(key);
if (d != self->state.data.end()) {
self->send(actor_cast<actor>(sender), std::move(key),
d->second.first);
return;
}
// ... or sub if it is not available.
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(sender));
self->state.data[key].second.insert(sender);
auto& subscribers = self->state.subscribers;
auto s = subscribers.find(sender);
if (s != subscribers.end()) {
s->second.insert(key);
} else {
self->monitor(sender);
subscribers.emplace(sender, peer_state::topic_set{key});
}
}
}
};
}
// -- stream server ------------------------------------------------------------ // -- stream server ------------------------------------------------------------
// The stream server acts as a man-in-the-middle for all streams that cross the // The stream server acts as a man-in-the-middle for all streams that cross the
...@@ -292,10 +379,12 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -292,10 +379,12 @@ actor_system::actor_system(actor_system_config& cfg)
static constexpr auto Flags = hidden + lazy_init; static constexpr auto Flags = hidden + lazy_init;
spawn_serv(actor_cast<strong_actor_ptr>(spawn<Flags>(spawn_serv_impl))); spawn_serv(actor_cast<strong_actor_ptr>(spawn<Flags>(spawn_serv_impl)));
config_serv(actor_cast<strong_actor_ptr>(spawn<Flags>(config_serv_impl))); config_serv(actor_cast<strong_actor_ptr>(spawn<Flags>(config_serv_impl)));
peer_serv(actor_cast<strong_actor_ptr>(spawn<Flags>(peer_serv_impl)));
// fire up remaining modules // fire up remaining modules
registry_.start(); registry_.start();
registry_.put(atom("SpawnServ"), spawn_serv()); registry_.put(atom("SpawnServ"), spawn_serv());
registry_.put(atom("ConfigServ"), config_serv()); registry_.put(atom("ConfigServ"), config_serv());
registry_.put(atom("PeerServ"), peer_serv());
for (auto& mod : modules_) for (auto& mod : modules_)
if (mod) if (mod)
mod->start(); mod->start();
......
...@@ -200,9 +200,9 @@ public: ...@@ -200,9 +200,9 @@ public:
buffer_type& out_buf, optional<uint16_t> port); buffer_type& out_buf, optional<uint16_t> port);
/// Writes the client handshake to `buf`. /// Writes the client handshake to `buf`.
static void write_client_handshake(execution_unit* ctx, buffer_type& buf, void write_client_handshake(execution_unit* ctx, buffer_type& buf,
const node_id& this_node, const node_id& this_node,
const std::string& app_identifier); const std::string& app_identifier);
/// Writes the client handshake to `buf`. /// Writes the client handshake to `buf`.
void write_client_handshake(execution_unit* ctx, buffer_type& buf); void write_client_handshake(execution_unit* ctx, buffer_type& buf);
......
...@@ -24,6 +24,7 @@ ...@@ -24,6 +24,7 @@
#include "caf/callback.hpp" #include "caf/callback.hpp"
#include "caf/io/abstract_broker.hpp" #include "caf/io/abstract_broker.hpp"
#include "caf/io/basp/buffer_type.hpp" #include "caf/io/basp/buffer_type.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
namespace caf { namespace caf {
...@@ -36,6 +37,7 @@ namespace basp { ...@@ -36,6 +37,7 @@ namespace basp {
/// BASP peer and provides both direct and indirect paths. /// BASP peer and provides both direct and indirect paths.
class routing_table { class routing_table {
public: public:
using endpoint = std::pair<uint16_t, network::address_listing>;
explicit routing_table(abstract_broker* parent); explicit routing_table(abstract_broker* parent);
...@@ -99,6 +101,12 @@ public: ...@@ -99,6 +101,12 @@ public:
return parent_; return parent_;
} }
/// Returns the local autoconnect endpoint.
const endpoint& autoconnect_endpoint();
/// Set the local autoconenct endpoint.
void autoconnect_endpoint(uint16_t, network::address_listing);
public: public:
template <class Map, class Fallback> template <class Map, class Fallback>
typename Map::mapped_type typename Map::mapped_type
...@@ -119,6 +127,7 @@ public: ...@@ -119,6 +127,7 @@ public:
std::unordered_map<node_id, connection_handle> direct_by_nid_; std::unordered_map<node_id, connection_handle> direct_by_nid_;
indirect_entries indirect_; indirect_entries indirect_;
indirect_entries blacklist_; indirect_entries blacklist_;
endpoint autoconnect_endpoint_;
}; };
/// @} /// @}
......
...@@ -466,6 +466,18 @@ behavior basp_broker::make_behavior() { ...@@ -466,6 +466,18 @@ behavior basp_broker::make_behavior() {
if (res) { if (res) {
auto port = res->second; auto port = res->second;
auto addrs = network::interfaces::list_addresses(false); auto addrs = network::interfaces::list_addresses(false);
// Remove link local addresses. These don't work for autoconnects.
for (auto& p : addrs) {
auto& vec = p.second;
vec.erase(std::remove_if(std::begin(vec), std::end(vec),
[](const std::string& str) {
return str.find("fe80") == 0;
}),
vec.end());
}
// Set this as the propagated autoconnect endpoint.
state.instance.tbl().autoconnect_endpoint(port, addrs);
// Add a config serv entry.
auto config_server = system().registry().get(atom("ConfigServ")); auto config_server = system().registry().get(atom("ConfigServ"));
send(actor_cast<actor>(config_server), put_atom::value, send(actor_cast<actor>(config_server), put_atom::value,
"basp.default-connectivity-tcp", "basp.default-connectivity-tcp",
......
...@@ -47,9 +47,12 @@ behavior datagram_connection_broker(broker* self, uint16_t port, ...@@ -47,9 +47,12 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
if (eptr) { if (eptr) {
auto hdl = (*eptr)->hdl(); auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr)); self->add_datagram_servant(std::move(*eptr));
basp::instance::write_client_handshake(self->context(), // TODO: UDP does not work at the moment.
self->wr_buf(hdl), this_node, // basp::instance::write_client_handshake(self->context(),
app_id); // self->wr_buf(hdl), this_node,
// app_id);
static_cast<void>(this_node);
static_cast<void>(hdl);
} }
} }
} }
......
...@@ -224,7 +224,7 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf, ...@@ -224,7 +224,7 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf,
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
auto appid = get_or(callee_.config(), "middleman.app-identifier", auto appid = get_or(callee_.config(), "middleman.app-identifier",
defaults::middleman::app_identifier); defaults::middleman::app_identifier);
if (auto err = sink(this_node_, appid)) if (auto err = sink(this_node_, appid, tbl_.autoconnect_endpoint()))
return err; return err;
if (pa != nullptr) { if (pa != nullptr) {
auto i = pa->first ? pa->first->id() : invalid_actor_id; auto i = pa->first ? pa->first->id() : invalid_actor_id;
...@@ -247,7 +247,8 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -247,7 +247,8 @@ void instance::write_client_handshake(execution_unit* ctx,
CAF_LOG_TRACE(CAF_ARG(this_node) << CAF_ARG(app_identifier)); CAF_LOG_TRACE(CAF_ARG(this_node) << CAF_ARG(app_identifier));
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<node_id&>(this_node), return sink(const_cast<node_id&>(this_node),
const_cast<std::string&>(app_identifier)); const_cast<std::string&>(app_identifier),
tbl_.autoconnect_endpoint());
}); });
header hdr{message_type::client_handshake, 0, 0, 0, invalid_actor_id, header hdr{message_type::client_handshake, 0, 0, 0, invalid_actor_id,
invalid_actor_id}; invalid_actor_id};
...@@ -297,6 +298,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -297,6 +298,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
node_id source_node; node_id source_node;
actor_id aid = invalid_actor_id; actor_id aid = invalid_actor_id;
std::set<std::string> sigs; std::set<std::string> sigs;
basp::routing_table::endpoint autoconn_addr;
if (!payload_valid()) { if (!payload_valid()) {
CAF_LOG_ERROR("fail to receive the app identifier"); CAF_LOG_ERROR("fail to receive the app identifier");
return false; return false;
...@@ -311,7 +313,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -311,7 +313,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
CAF_LOG_ERROR("app identifier mismatch"); CAF_LOG_ERROR("app identifier mismatch");
return false; return false;
} }
if (bd(aid, sigs)) if (bd(aid, sigs, autoconn_addr))
return false; return false;
} }
// Close self connection immediately after handshake. // Close self connection immediately after handshake.
...@@ -329,6 +331,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -329,6 +331,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
// Add new route to this node. // Add new route to this node.
CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node)); CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node); tbl_.add_direct(hdl, source_node);
// Store autoconnect address.
auto peer_server = system().registry().get(atom("PeerServ"));
anon_send(actor_cast<actor>(peer_server), put_atom::value,
to_string(source_node), make_message(std::move(autoconn_addr)));
// write handshake as client in response // write handshake as client in response
auto path = tbl_.lookup(source_node); auto path = tbl_.lookup(source_node);
if (!path) { if (!path) {
...@@ -349,7 +355,8 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -349,7 +355,8 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
binary_deserializer bd{ctx, *payload}; binary_deserializer bd{ctx, *payload};
node_id source_node; node_id source_node;
std::string remote_appid; std::string remote_appid;
if (bd(source_node, remote_appid)) basp::routing_table::endpoint autoconn_addr;
if (bd(source_node, remote_appid, autoconn_addr))
return false; return false;
auto appid = get_if<std::string>(&callee_.config(), auto appid = get_if<std::string>(&callee_.config(),
"middleman.app-identifier"); "middleman.app-identifier");
...@@ -367,6 +374,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -367,6 +374,10 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node)); CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node); tbl_.add_direct(hdl, source_node);
callee_.learned_new_node_directly(source_node, false); callee_.learned_new_node_directly(source_node, false);
// Store autoconnect address.
auto peer_server = system().registry().get(atom("PeerServ"));
anon_send(actor_cast<actor>(peer_server), put_atom::value,
to_string(source_node), make_message(std::move(autoconn_addr)));
break; break;
} }
case message_type::dispatch_message: { case message_type::dispatch_message: {
......
...@@ -152,6 +152,15 @@ size_t routing_table::erase(const node_id& dest, erase_callback& cb) { ...@@ -152,6 +152,15 @@ size_t routing_table::erase(const node_id& dest, erase_callback& cb) {
return res; return res;
} }
const routing_table::endpoint& routing_table::autoconnect_endpoint() {
return autoconnect_endpoint_;
}
void routing_table::autoconnect_endpoint(uint16_t port,
network::address_listing addrs) {
autoconnect_endpoint_ = {port, std::move(addrs)};
}
} // namespace basp } // namespace basp
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -263,7 +263,8 @@ public: ...@@ -263,7 +263,8 @@ public:
void connect_node(node& n, void connect_node(node& n,
optional<accept_handle> ax = none, optional<accept_handle> ax = none,
actor_id published_actor_id = invalid_actor_id, actor_id published_actor_id = invalid_actor_id,
const set<string>& published_actor_ifs = std::set<std::string>{}) { const set<string>& published_actor_ifs = set<string>{},
const basp::routing_table::endpoint& autoconn = {}) {
auto src = ax ? *ax : ahdl_; auto src = ax ? *ax : ahdl_;
CAF_MESSAGE("connect remote node " << n.name CAF_MESSAGE("connect remote node " << n.name
<< ", connection ID = " << n.connection.id() << ", connection ID = " << n.connection.id()
...@@ -276,10 +277,10 @@ public: ...@@ -276,10 +277,10 @@ public:
mock(hdl, mock(hdl,
{basp::message_type::client_handshake, 0, 0, 0, invalid_actor_id, {basp::message_type::client_handshake, 0, 0, 0, invalid_actor_id,
invalid_actor_id}, invalid_actor_id},
n.id, std::string{}) n.id, std::string{}, basp::routing_table::endpoint{})
.receive(hdl, basp::message_type::server_handshake, no_flags, any_vals, .receive(hdl, basp::message_type::server_handshake, no_flags, any_vals,
basp::version, published_actor_id, invalid_actor_id, this_node(), basp::version, published_actor_id, invalid_actor_id, this_node(),
std::string{}, published_actor_id, published_actor_ifs) std::string{}, autoconn, published_actor_id, published_actor_ifs)
// upon receiving our client handshake, BASP will check // upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node // whether there is a SpawnServ actor on this node
.receive(hdl, basp::message_type::dispatch_message, .receive(hdl, basp::message_type::dispatch_message,
...@@ -480,7 +481,8 @@ CAF_TEST(non_empty_server_handshake) { ...@@ -480,7 +481,8 @@ CAF_TEST(non_empty_server_handshake) {
basp::header expected{basp::message_type::server_handshake, 0, 0, basp::header expected{basp::message_type::server_handshake, 0, 0,
basp::version, self()->id(), invalid_actor_id}; basp::version, self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr, this_node(), std::string{}, to_buf(expected_buf, expected, nullptr, this_node(), std::string{},
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"}); basp::routing_table::endpoint{}, self()->id(),
set<string>{"caf::replies_to<@u16>::with<@u16>"});
CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf)); CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf));
} }
...@@ -566,10 +568,12 @@ CAF_TEST(remote_actor_and_send) { ...@@ -566,10 +568,12 @@ CAF_TEST(remote_actor_and_send) {
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version, {basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().dummy_actor->id(), invalid_actor_id}, jupiter().dummy_actor->id(), invalid_actor_id},
jupiter().id, std::string{}, jupiter().dummy_actor->id(), uint32_t{0}) jupiter().id, std::string{}, jupiter().dummy_actor->id(), uint32_t{0},
basp::routing_table::endpoint{})
.receive(jupiter().connection, basp::message_type::client_handshake, .receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, any_vals, no_operation_data, invalid_actor_id, no_flags, any_vals, no_operation_data, invalid_actor_id,
invalid_actor_id, this_node(), std::string{}) invalid_actor_id, this_node(), std::string{},
basp::routing_table::endpoint{})
.receive(jupiter().connection, basp::message_type::dispatch_message, .receive(jupiter().connection, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, invalid_actor_id, default_operation_data, any_vals, invalid_actor_id,
...@@ -752,4 +756,45 @@ CAF_TEST_DISABLED(automatic_connection) { ...@@ -752,4 +756,45 @@ CAF_TEST_DISABLED(automatic_connection) {
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
} }
CAF_TEST(read_address_after_handshake) {
auto check_node_in_tbl = [&](node& n) {
auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl);
};
mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
publish(self(), 4242);
mpx()->flush_runnables(); // process publish message in basp_broker
CAF_MESSAGE("connect to mars");
auto ep = instance().tbl().autoconnect_endpoint();
connect_node(mars(), ax, self()->id(), set<string>{}, ep);
check_node_in_tbl(mars());
CAF_MESSAGE("Look for mars address information in our peer server");
auto peer_server = sys.registry().get(atom("PeerServ"));
CAF_MESSAGE("Send request");
self()->send(actor_cast<actor>(peer_server), get_atom::value,
to_string(mars().id));
// process get request and send answer
do {
sched.run();
mpx()->flush_runnables();
} while (self()->mailbox().empty());
CAF_MESSAGE("Process reply");
self()->receive(
[&](const std::string& item, message& msg) {
// Check that we got an entry under the name of our peer.
CAF_REQUIRE_EQUAL(item, to_string(mars().id));
msg.apply(
[&](basp::routing_table::endpoint& ep) {
// The addresses of our dummy node, thus empty.
CAF_CHECK(ep.second.empty());
}
);
}
);
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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