Commit 45fc6641 authored by Joseph Noir's avatar Joseph Noir

Add a separate store for peer addresses

parent 103b08ef
......@@ -128,12 +128,15 @@ public:
friend class abstract_actor;
/// 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.
/// @pre x in {'SpawnServ', 'ConfigServ', 'StreamServ'}
/// @pre x in {'SpawnServ', 'ConfigServ', 'StreamServ', 'PeerServ'}
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.
......@@ -147,6 +150,11 @@ public:
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(const actor_system&) = delete;
actor_system& operator=(const actor_system&) = delete;
......@@ -547,6 +555,11 @@ private:
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);
}
std::atomic<size_t> ids_;
uniform_type_info_map types_;
node_id node_;
......
......@@ -170,6 +170,97 @@ behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) {
};
}
// -- peer server --------------------------------------------------------------
// A peer server keepy track of the addresses to reach its peers. All addresses
// for a given node are stored under the string representation of iits node id.
// When an entry is requested that does not exist, the requester is subscribed
// to the key and sent a message as soon as an entry is set and then 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;
template <class Processor>
friend void serialize(Processor& proc, peer_state& x, unsigned int) {
proc & x.data;
proc & x.subscribers;
}
};
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 subscriber = actor_cast<strong_actor_ptr>(self->current_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(subscriber, std::move(key), d->second.first);
return;
}
// ... or sub if it is not available.
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(subscriber));
if (subscriber) {
self->state.data[key].second.insert(subscriber);
auto& subscribers = self->state.subscribers;
auto s = subscribers.find(subscriber);
if (s != subscribers.end()) {
s->second.insert(key);
} else {
self->monitor(subscriber);
subscribers.emplace(subscriber, peer_state::topic_set{key});
}
}
}
};
}
// -- stream server ------------------------------------------------------------
// The stream server acts as a man-in-the-middle for all streams that cross the
......@@ -275,14 +366,17 @@ actor_system::actor_system(actor_system_config& cfg)
if (mod)
mod->init(cfg);
groups_.init(cfg);
// spawn config and spawn servers (lazily to not access the scheduler yet)
// spawn config, spawn, and peer servers
// (lazily to not access the scheduler yet)
static constexpr auto Flags = hidden + lazy_init;
spawn_serv(actor_cast<strong_actor_ptr>(spawn<Flags>(spawn_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
registry_.start();
registry_.put(atom("SpawnServ"), spawn_serv());
registry_.put(atom("ConfigServ"), config_serv());
registry_.put(atom("PeerServ"), peer_serv());
for (auto& mod : modules_)
if (mod)
mod->start();
......@@ -303,6 +397,7 @@ actor_system::~actor_system() {
}
registry_.erase(atom("SpawnServ"));
registry_.erase(atom("ConfigServ"));
registry_.erase(atom("PeerServ"));
registry_.erase(atom("StreamServ"));
// group module is the first one, relies on MM
groups_.stop();
......
......@@ -350,8 +350,8 @@ public:
tbl_.handle(hdr.source_node, hdl);
else
tbl_.add(hdr.source_node, hdl);
auto config_server = system().registry().get(atom("ConfigServ"));
anon_send(actor_cast<actor>(config_server), put_atom::value,
auto peer_server = system().registry().get(atom("PeerServ"));
anon_send(actor_cast<actor>(peer_server), put_atom::value,
to_string(hdr.source_node), make_message(addrs));
// Write handshake as client in response.
if (tcp_based)
......@@ -399,8 +399,8 @@ public:
tbl_.handle(hdr.source_node, hdl);
else
tbl_.add(hdr.source_node, hdl);
auto config_server = system().registry().get(atom("ConfigServ"));
anon_send(actor_cast<actor>(config_server), put_atom::value,
auto peer_server = system().registry().get(atom("PeerServ"));
anon_send(actor_cast<actor>(peer_server), put_atom::value,
to_string(hdr.source_node), make_message(addrs));
}
// Since udp is unreliable we answer, maybe our message was lost.
......
......@@ -449,7 +449,7 @@ void basp_broker_state::establish_communication(const node_id& nid) {
: system().spawn<hidden>(connection_helper, self, &instance);
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([&item](serializer& sink) -> error {
auto name_atm = atom("ConfigServ");
auto name_atm = atom("PeerServ");
std::vector<actor_id> stages;
auto msg = make_message(get_atom::value, std::move(item));
return sink(name_atm, stages, msg);
......@@ -876,6 +876,7 @@ behavior basp_broker::make_behavior() {
},
// Received from underlying broker implementation.
[=](const new_connection_msg& msg) {
auto res = state.instance.tbl().lookup(msg.handle);
CAF_LOG_TRACE(CAF_ARG(msg.handle));
auto& bi = state.instance;
bi.write_server_handshake(context(), state.get_buffer(msg.handle),
......
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