Unverified Commit 732a8a68 authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #813

Port auto-connect changes to cleanup branch
parents 67b7161c 7bedcd63
...@@ -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();
......
...@@ -80,14 +80,8 @@ public: ...@@ -80,14 +80,8 @@ public:
std::vector<strong_actor_ptr>& forwarding_stack, std::vector<strong_actor_ptr>& forwarding_stack,
message& msg) = 0; message& msg) = 0;
/// Called whenever BASP learns the ID of a remote node /// Called whenever BASP learns the ID of a remote node.
/// to which it does not have a direct connection. virtual void learned_new_node(const node_id& nid) = 0;
virtual void learned_new_node_directly(const node_id& nid,
bool was_known_indirectly) = 0;
/// Called whenever BASP learns the ID of a remote node
/// to which it does not have a direct connection.
virtual void learned_new_node_indirectly(const node_id& nid) = 0;
/// Called if a heartbeat was received from `nid` /// Called if a heartbeat was received from `nid`
virtual void handle_heartbeat(const node_id& nid) = 0; virtual void handle_heartbeat(const node_id& nid) = 0;
...@@ -107,9 +101,19 @@ public: ...@@ -107,9 +101,19 @@ public:
return namespace_.system().config(); return namespace_.system().config();
} }
/// Send messages that were buffered while connectivity establishment
/// was pending using `hdl`.
virtual void send_buffered_messages(execution_unit* ctx, node_id nid,
connection_handle hdl) = 0;
/// Returns a reference to the sent buffer. /// Returns a reference to the sent buffer.
virtual buffer_type& get_buffer(connection_handle hdl) = 0; virtual buffer_type& get_buffer(connection_handle hdl) = 0;
/// Returns a reference to a buffer to be sent to node with `nid`.
/// If communication with the node is esstablished, it picks the first
/// available handle, otherwise a buffer for a pending message is returned.
virtual buffer_type& get_buffer(node_id nid) = 0;
/// Flushes the underlying write buffer of `hdl`. /// Flushes the underlying write buffer of `hdl`.
virtual void flush(connection_handle hdl) = 0; virtual void flush(connection_handle hdl) = 0;
...@@ -135,15 +139,15 @@ public: ...@@ -135,15 +139,15 @@ public:
void handle_heartbeat(execution_unit* ctx); void handle_heartbeat(execution_unit* ctx);
/// Returns a route to `target` or `none` on error. /// Returns a route to `target` or `none` on error.
optional<routing_table::route> lookup(const node_id& target); routing_table::lookup_result lookup(const node_id& target);
/// Flushes the underlying buffer of `path`. /// Flushes the underlying buffer of `path`.
void flush(const routing_table::route& path); void flush(connection_handle hdl);
/// Sends a BASP message and implicitly flushes the output buffer of `r`. /// Sends a BASP message and implicitly flushes the output buffer of `r`.
/// This function will update `hdr.payload_len` if a payload was written. /// This function will update `hdr.payload_len` if a payload was written.
void write(execution_unit* ctx, const routing_table::route& r, void write(execution_unit* ctx, connection_handle hdl, header& hdr,
header& hdr, payload_writer* writer = nullptr); payload_writer* writer = nullptr);
/// Adds a new actor to the map of published actors. /// Adds a new actor to the map of published actors.
void add_published_actor(uint16_t port, void add_published_actor(uint16_t port,
...@@ -200,9 +204,9 @@ public: ...@@ -200,9 +204,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,70 +37,90 @@ namespace basp { ...@@ -36,70 +37,90 @@ 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);
virtual ~routing_table(); virtual ~routing_table();
/// Describes a routing path to a node. /// Result for a lookup of a node.
struct route { struct lookup_result {
const node_id& next_hop; /// Tracks whether the node is already known.
connection_handle hdl; bool known;
/// Servant handle to communicate with the node -- if already created.
optional<connection_handle> hdl;
}; };
/// Describes a function object for erase operations that /// Describes a function object for erase operations that
/// is called for each indirectly lost connection. /// is called for each indirectly lost connection.
using erase_callback = callback<const node_id&>; using erase_callback = callback<const node_id&>;
/// Returns a route to `target` or `none` on error. /// Returns the ID of the peer reachable via `hdl` or
optional<route> lookup(const node_id& target);
/// Returns the ID of the peer connected via `hdl` or
/// `none` if `hdl` is unknown. /// `none` if `hdl` is unknown.
node_id lookup_direct(const connection_handle& hdl) const; node_id lookup(const connection_handle& hdl) const;
/// Returns the handle offering a direct connection to `nid` or
/// `invalid_connection_handle` if no direct connection to `nid` exists.
optional<connection_handle> lookup_direct(const node_id& nid) const;
/// Returns the next hop that would be chosen for `nid` /// Returns the state for communication with `nid` along with a handle
/// or `none` if there's no indirect route to `nid`. /// if communication is established or `none` if `nid` is unknown.
node_id lookup_indirect(const node_id& nid) const; lookup_result lookup(const node_id& nid) const;
/// Adds a new direct route to the table. /// Adds a new endpoint to the table.
/// @pre `hdl != invalid_connection_handle && nid != none` /// @pre `hdl != invalid_connection_handle && nid != none`
void add_direct(const connection_handle& hdl, const node_id& nid); void add(const node_id& nid, const connection_handle& hdl);
/// Adds a new indirect route to the table. /// Add a new endpoint to the table.
bool add_indirect(const node_id& hop, const node_id& dest); /// @pre `origin != none && nid != none`
void add(const node_id& nid, const node_id& origin);
/// Blacklist the route to `dest` via `hop`. /// Adds a new endpoint to the table that has no attached information.
void blacklist(const node_id& hop, const node_id& dest); /// that propagated information about the node.
/// @pre `nid != none`
void add(const node_id& nid);
/// Removes a direct connection and calls `cb` for any node /// Removes a direct connection and calls `cb` for any node
/// that became unreachable as a result of this operation, /// that became unreachable as a result of this operation,
/// including the node that is assigned as direct path for `hdl`. /// including the node that is assigned as direct path for `hdl`.
void erase_direct(const connection_handle& hdl, erase_callback& cb); void erase(const connection_handle& hdl, erase_callback& cb);
/// Removes any entry for indirect connection to `dest` and returns /// Queries whether `dest` is reachable directly.
/// `true` if `dest` had an indirect route, otherwise `false`.
bool erase_indirect(const node_id& dest);
/// Queries whether `dest` is reachable.
bool reachable(const node_id& dest); bool reachable(const node_id& dest);
/// Removes all direct and indirect routes to `dest` and calls
/// `cb` for any node that became unreachable as a result of this
/// operation, including `dest`.
/// @returns the number of removed routes (direct and indirect)
size_t erase(const node_id& dest, erase_callback& cb);
/// Returns the parent broker. /// Returns the parent broker.
inline abstract_broker* parent() { inline abstract_broker* parent() {
return parent_; return parent_;
} }
/// Set the forwarding node that first mentioned `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<node_id> origin(const node_id& nid);
/// Set the handle for communication with `nid`.
bool handle(const node_id& nid, const connection_handle& hdl);
/// Get the handle for communication with `nid`
/// or `none` if the node is unknown.
optional<connection_handle> handle(const node_id& nid);
/// Get the addresses to reach `nid` or `none` if the node is unknown.
optional<const endpoint&> address(const node_id& nid);
/// Returns the local autoconnect endpoint.
const endpoint& autoconnect_endpoint();
/// Set the local autoconenct endpoint.
void autoconnect_endpoint(uint16_t, network::address_listing);
public: public:
/// Entry to bundle information for a remote endpoint.
struct node_info {
/// Handle for the node if communication is established.
optional<connection_handle> hdl;
/// The endpoint who told us about the node.
optional<node_id> origin;
};
template <class Map, class Fallback> template <class Map, class Fallback>
typename Map::mapped_type typename Map::mapped_type
get_opt(const Map& m, const typename Map::key_type& k, Fallback&& x) const { get_opt(const Map& m, const typename Map::key_type& k, Fallback&& x) const {
...@@ -109,16 +130,10 @@ public: ...@@ -109,16 +130,10 @@ public:
return std::forward<Fallback>(x); return std::forward<Fallback>(x);
} }
using node_id_set = std::unordered_set<node_id>;
using indirect_entries = std::unordered_map<node_id, // dest
node_id_set>; // hop
abstract_broker* parent_; abstract_broker* parent_;
std::unordered_map<connection_handle, node_id> direct_by_hdl_; std::unordered_map<connection_handle, node_id> nid_by_hdl_;
std::unordered_map<node_id, connection_handle> direct_by_nid_; std::unordered_map<node_id, node_info> node_information_base_;
indirect_entries indirect_; endpoint autoconnect_endpoint_;
indirect_entries blacklist_;
}; };
/// @} /// @}
......
...@@ -77,17 +77,17 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -77,17 +77,17 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
std::vector<strong_actor_ptr>& stages, message& msg); std::vector<strong_actor_ptr>& stages, message& msg);
// performs bookkeeping such as managing `spawn_servers` // performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid); void learned_new_node(const node_id& nid) override;
// inherited from basp::instance::callee // inherited from basp::instance::callee
void learned_new_node_directly(const node_id& nid, void send_buffered_messages(execution_unit* ctx, node_id nid,
bool was_indirectly_before) override; connection_handle hdl) override;
// inherited from basp::instance::callee // inherited from basp::instance::callee
void learned_new_node_indirectly(const node_id& nid) override; buffer_type& get_buffer(connection_handle hdl) override;
// inherited from basp::instance::callee // inherited from basp::instance::callee
buffer_type& get_buffer(connection_handle hdl) override; buffer_type& get_buffer(node_id nid) override;
// inherited from basp::instance::callee // inherited from basp::instance::callee
void flush(connection_handle hdl) override; void flush(connection_handle hdl) override;
...@@ -102,6 +102,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -102,6 +102,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
/// Cleans up any state for `hdl`. /// Cleans up any state for `hdl`.
void cleanup(connection_handle hdl); void cleanup(connection_handle hdl);
/// Try to establish a connection to node with `nid`.
void connect(const node_id& nid);
// pointer to ourselves // pointer to ourselves
broker* self; broker* self;
...@@ -146,6 +149,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -146,6 +149,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// actor // actor
void handle_down_msg(down_msg&); void handle_down_msg(down_msg&);
// buffer messages for nodes while connectivity is established
std::unordered_map<node_id, std::vector<buffer_type>> pending_connectivity;
static const char* name; static const char* name;
}; };
......
...@@ -46,13 +46,8 @@ struct connection_helper_state { ...@@ -46,13 +46,8 @@ struct connection_helper_state {
static const char* name; static const char* name;
}; };
behavior datagram_connection_broker(broker* self,
uint16_t port,
network::address_listing addresses,
actor system_broker);
behavior connection_helper(stateful_actor<connection_helper_state>* self, behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b); actor system_broker);
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -64,25 +64,34 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -64,25 +64,34 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
CAF_ASSERT(nid != this_node()); CAF_ASSERT(nid != this_node());
if (nid == none || aid == invalid_actor_id) if (nid == none || aid == invalid_actor_id)
return nullptr; return nullptr;
// this member function is being called whenever we deserialize a // This member function is being called whenever we deserialize a
// payload received from a remote node; if a remote node A sends // payload received from a remote node; if a remote node A sends
// us a handle to a third node B, then we assume that A offers a route to B // us a handle to a third node B, then we assume that A can tell us
if (nid != this_context->id // how to contact B.
&& !instance.tbl().lookup_direct(nid) // TODO: This should probably happen somewhere else, but is usually only
&& instance.tbl().add_indirect(this_context->id, nid)) // performed in `finalize_handshake` which is only called on receipt
learned_new_node_indirectly(nid); // of server handshakes.
// we need to tell remote side we are watching this actor now; if (this_context->id == none)
// use a direct route if possible, i.e., when talking to a third node this_context->id = instance.tbl().lookup(this_context->hdl);
auto path = instance.tbl().lookup(nid); auto lr = instance.tbl().lookup(nid);
if (!path) { if (nid != this_context->id && !lr.known) {
// this happens if and only if we don't have a path to `nid` instance.tbl().add(nid, this_context->id);
// and current_context_->hdl has been blacklisted connect(nid);
CAF_LOG_DEBUG("cannot create a proxy instance for an actor " }
"running on a node we don't have a route to"); // We need to tell remote side we are watching this actor now;
// use a direct route if possible, i.e., when talking to a third node.
// TODO: Communication setup might still be in progress.
/*
if (lr.known && !lr.hdl) {
// This happens if and only if we don't have a path to `nid`
// and current_context_->hdl has been blacklisted.
CAF_LOG_INFO("cannot create a proxy instance for an actor "
"running on a node we don't have a route to");
return nullptr; return nullptr;
} }
// create proxy and add functor that will be called if we */
// receive a kill_proxy_instance message // Create proxy and add functor that will be called if we
// receive a kill_proxy_instance message.
auto mm = &system().middleman(); auto mm = &system().middleman();
actor_config cfg; actor_config cfg;
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>( auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>(
...@@ -90,22 +99,27 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -90,22 +99,27 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
strong_actor_ptr selfptr{self->ctrl()}; strong_actor_ptr selfptr{self->ctrl()};
res->get()->attach_functor([=](const error& rsn) { res->get()->attach_functor([=](const error& rsn) {
mm->backend().post([=] { mm->backend().post([=] {
// using res->id() instead of aid keeps this actor instance alive // Using res->id() instead of aid keeps this actor instance alive
// until the original instance terminates, thus preventing subtle // until the original instance terminates, thus preventing subtle
// bugs with attachables // bugs with attachables.
auto bptr = static_cast<basp_broker*>(selfptr->get()); auto bptr = static_cast<basp_broker*>(selfptr->get());
if (!bptr->getf(abstract_actor::is_terminated_flag)) if (!bptr->getf(abstract_actor::is_terminated_flag))
bptr->state.proxies().erase(nid, res->id(), rsn); bptr->state.proxies().erase(nid, res->id(), rsn);
}); });
}); });
CAF_LOG_DEBUG("successfully created proxy instance, " CAF_LOG_INFO("successfully created proxy instance, "
"write announce_proxy_instance:" "write announce_proxy_instance:"
<< CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(nid) << CAF_ARG(aid));
<< CAF_ARG2("hdl", this_context->hdl)); // TODO: Can it happen that things have changed here?
// tell remote side we are monitoring this actor now lr = instance.tbl().lookup(nid);
instance.write_announce_proxy(self->context(), get_buffer(this_context->hdl), if (lr.hdl) {
nid, aid); auto hdl = std::move(*lr.hdl);
instance.flush(*path); // Tell remote side we are monitoring this actor now.
instance.write_announce_proxy(self->context(), get_buffer(hdl), nid, aid);
flush(hdl);
} else {
instance.write_announce_proxy(self->context(), get_buffer(nid), nid, aid);
}
mm->notify<hook::new_remote_actor>(res); mm->notify<hook::new_remote_actor>(res);
return res; return res;
} }
...@@ -150,15 +164,21 @@ void basp_broker_state::purge_state(const node_id& nid) { ...@@ -150,15 +164,21 @@ void basp_broker_state::purge_state(const node_id& nid) {
void basp_broker_state::send_kill_proxy_instance(const node_id& nid, void basp_broker_state::send_kill_proxy_instance(const node_id& nid,
actor_id aid, error rsn) { actor_id aid, error rsn) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn));
auto path = instance.tbl().lookup(nid); auto res = instance.tbl().lookup(nid);
if (!path) { if (!res.known) {
CAF_LOG_INFO("cannot send exit message for proxy, no route to host:" CAF_LOG_INFO("cannot send exit message for proxy, no route to host:"
<< CAF_ARG(nid)); << CAF_ARG(nid));
return; return;
} }
instance.write_kill_proxy(self->context(), get_buffer(path->hdl), nid, aid, if (res.hdl) {
rsn); auto hdl = std::move(*res.hdl);
instance.flush(*path); instance.write_kill_proxy(self->context(), get_buffer(hdl), nid, aid,
rsn);
instance.flush(hdl);
} else {
instance.write_kill_proxy(self->context(), get_buffer(nid), nid, aid,
rsn);
}
} }
void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) { void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
...@@ -331,8 +351,8 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -331,8 +351,8 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
auto msg = make_message(sys_atom::value, get_atom::value, "info"); auto msg = make_message(sys_atom::value, get_atom::value, "info");
return sink(name_atm, stages, msg); return sink(name_atm, stages, msg);
}); });
auto path = instance.tbl().lookup(nid); auto res = instance.tbl().lookup(nid);
if (!path) { if (!res.known) {
CAF_LOG_ERROR("learned_new_node called, but no route to nid"); CAF_LOG_ERROR("learned_new_node called, but no route to nid");
return; return;
} }
...@@ -343,46 +363,50 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -343,46 +363,50 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
invalid_actor_id}; invalid_actor_id};
// writing std::numeric_limits<actor_id>::max() is a hack to get // writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases // this send-to-named-actor feature working with older CAF releases
instance.write(self->context(), get_buffer(path->hdl), if (res.hdl) {
hdr, &writer); auto hdl = std::move(*res.hdl);
instance.flush(*path); instance.write(self->context(), get_buffer(hdl), hdr, &writer);
} instance.flush(hdl);
} else {
void basp_broker_state::learned_new_node_directly(const node_id& nid, instance.write(self->context(), get_buffer(nid), hdr, &writer);
bool was_indirectly_before) { }
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid));
if (!was_indirectly_before)
learned_new_node(nid);
} }
void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { void basp_broker_state::connect(const node_id& nid) {
// TODO: Split this by functionality, address query & connecting?
CAF_ASSERT(this_context != nullptr); CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid)); CAF_LOG_TRACE(CAF_ARG(nid));
learned_new_node(nid);
if (!automatic_connections) if (!automatic_connections)
return; return;
// this member function gets only called once, after adding a new // this member function gets only called once, after adding a new
// indirect connection to the routing table; hence, spawning // indirect connection to the routing table; hence, spawning
// our helper here exactly once and there is no need to track // our helper here exactly once and there is no need to track
// in-flight connection requests // in-flight connection requests
auto path = instance.tbl().lookup(nid); if (instance.tbl().lookup(nid).hdl) {
if (!path) { CAF_LOG_ERROR("establish_communication called with established connection");
CAF_LOG_ERROR("learned_new_node_indirectly called, but no route to nid");
return; return;
} }
if (path->next_hop == nid) { auto origin = instance.tbl().origin(nid);
CAF_LOG_ERROR("learned_new_node_indirectly called with direct connection"); if (!origin) {
CAF_LOG_ERROR("establish_communication called, but no node known "
"to ask for contact information");
return; return;
} }
auto ehdl = instance.tbl().handle(*origin);
if (!ehdl) {
CAF_LOG_ERROR("establish_communication called, but node with contact "
"information is no longer reachable");
return;
}
auto hdl = std::move(*ehdl);
using namespace detail; using namespace detail;
auto try_connect = [&](std::string item) { auto try_connect = [&](std::string item) {
auto tmp = get_or(config(), "middleman.attach-utility-actors", false) auto tmp = get_or(config(), "middleman.attach-utility-actors", false)
? system().spawn<hidden>(connection_helper, self) ? system().spawn<hidden>(connection_helper, self)
: system().spawn<detached + hidden>(connection_helper, self); : system().spawn<detached + hidden>(connection_helper, self);
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp)); system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([&item](serializer& sink) -> error { auto writer = make_callback([&item](serializer& sink) -> error {
auto name_atm = atom("ConfigServ"); auto name_atm = atom("PeerServ");
std::vector<actor_id> stages; std::vector<actor_id> stages;
auto msg = make_message(get_atom::value, std::move(item)); auto msg = make_message(get_atom::value, std::move(item));
return sink(name_atm, stages, msg); return sink(name_atm, stages, msg);
...@@ -391,11 +415,11 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -391,11 +415,11 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
basp::header::named_receiver_flag, 0, basp::header::named_receiver_flag, 0,
make_message_id().integer_value(), tmp.id(), make_message_id().integer_value(), tmp.id(),
invalid_actor_id}; invalid_actor_id};
instance.write(self->context(), get_buffer(path->hdl), instance.write(self->context(), get_buffer(hdl),
hdr, &writer); hdr, &writer);
instance.flush(*path); flush(hdl);
}; };
try_connect("basp.default-connectivity-tcp"); try_connect(to_string(nid));
} }
void basp_broker_state::set_context(connection_handle hdl) { void basp_broker_state::set_context(connection_handle hdl) {
...@@ -421,7 +445,7 @@ void basp_broker_state::cleanup(connection_handle hdl) { ...@@ -421,7 +445,7 @@ void basp_broker_state::cleanup(connection_handle hdl) {
purge_state(nid); purge_state(nid);
return none; return none;
}); });
instance.tbl().erase_direct(hdl, cb); instance.tbl().erase(hdl, cb);
// Remove the context for `hdl`, making sure clients receive an error in case // Remove the context for `hdl`, making sure clients receive an error in case
// this connection was closed during handshake. // this connection was closed during handshake.
auto i = ctx.find(hdl); auto i = ctx.find(hdl);
...@@ -441,10 +465,32 @@ basp_broker_state::get_buffer(connection_handle hdl) { ...@@ -441,10 +465,32 @@ basp_broker_state::get_buffer(connection_handle hdl) {
return self->wr_buf(hdl); return self->wr_buf(hdl);
} }
basp_broker_state::buffer_type&
basp_broker_state::get_buffer(node_id nid) {
auto res = instance.tbl().lookup(nid);
if (res.known && res.hdl)
return get_buffer(*res.hdl);
auto& msgs = pending_connectivity[nid];
msgs.emplace_back();
return msgs.back();
}
void basp_broker_state::flush(connection_handle hdl) { void basp_broker_state::flush(connection_handle hdl) {
self->flush(hdl); self->flush(hdl);
} }
void basp_broker_state::send_buffered_messages(execution_unit*,
node_id nid,
connection_handle hdl) {
if (pending_connectivity.count(nid) > 0) {
for (auto& msg : pending_connectivity[nid]) {
auto& buf = get_buffer(hdl);
buf.insert(buf.end(), msg.begin(), msg.end());
}
}
flush(hdl);
}
/****************************************************************************** /******************************************************************************
* basp_broker * * basp_broker *
******************************************************************************/ ******************************************************************************/
...@@ -466,6 +512,18 @@ behavior basp_broker::make_behavior() { ...@@ -466,6 +512,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",
...@@ -533,8 +591,8 @@ behavior basp_broker::make_behavior() { ...@@ -533,8 +591,8 @@ behavior basp_broker::make_behavior() {
<< ", " << CAF_ARG(msg)); << ", " << CAF_ARG(msg));
if (!src) if (!src)
return sec::invalid_argument; return sec::invalid_argument;
auto path = this->state.instance.tbl().lookup(dest_node); auto lr = this->state.instance.tbl().lookup(dest_node);
if (!path) { if (!lr.known) {
CAF_LOG_ERROR("no route to receiving node"); CAF_LOG_ERROR("no route to receiving node");
return sec::no_route_to_receiving_node; return sec::no_route_to_receiving_node;
} }
...@@ -546,9 +604,14 @@ behavior basp_broker::make_behavior() { ...@@ -546,9 +604,14 @@ behavior basp_broker::make_behavior() {
basp::header hdr{basp::message_type::dispatch_message, basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag, 0, basp::header::named_receiver_flag, 0,
cme->mid.integer_value(), src->id(), invalid_actor_id}; cme->mid.integer_value(), src->id(), invalid_actor_id};
state.instance.write(context(), state.get_buffer(path->hdl), if (lr.hdl) {
hdr, &writer); auto hdl = std::move(*lr.hdl);
state.instance.flush(*path); state.instance.write(context(), state.get_buffer(hdl), hdr, &writer);
state.instance.flush(hdl);
} else {
state.instance.write(context(), state.get_buffer(dest_node), hdr,
&writer);
}
return delegated<message>(); return delegated<message>();
}, },
// received from underlying broker implementation // received from underlying broker implementation
...@@ -628,10 +691,10 @@ behavior basp_broker::make_behavior() { ...@@ -628,10 +691,10 @@ behavior basp_broker::make_behavior() {
-> std::tuple<node_id, std::string, uint16_t> { -> std::tuple<node_id, std::string, uint16_t> {
std::string addr; std::string addr;
uint16_t port = 0; uint16_t port = 0;
auto hdl = state.instance.tbl().lookup_direct(x); auto lr = state.instance.tbl().lookup(x);
if (hdl) { if (lr.known && lr.hdl) {
addr = remote_addr(*hdl); addr = remote_addr(*lr.hdl);
port = remote_port(*hdl); port = remote_port(*lr.hdl);
} }
return std::make_tuple(x, std::move(addr), port); return std::make_tuple(x, std::move(addr), port);
}, },
......
...@@ -34,94 +34,45 @@ auto autoconnect_timeout = std::chrono::minutes(10); ...@@ -34,94 +34,45 @@ auto autoconnect_timeout = std::chrono::minutes(10);
const char* connection_helper_state::name = "connection_helper"; const char* connection_helper_state::name = "connection_helper";
behavior datagram_connection_broker(broker* self, uint16_t port,
network::address_listing addresses,
actor system_broker) {
auto& mx = self->system().middleman().backend();
auto& this_node = self->system().node();
auto app_id = get_or(self->config(), "middleman.app-identifier",
defaults::middleman::app_identifier);
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto eptr = mx.new_remote_udp_endpoint(addr, port);
if (eptr) {
auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr));
basp::instance::write_client_handshake(self->context(),
self->wr_buf(hdl), this_node,
app_id);
}
}
}
return {
[=](new_datagram_msg& msg) {
auto hdl = msg.handle;
self->send(system_broker, std::move(msg), self->take(hdl), port);
self->quit();
},
after(autoconnect_timeout) >> [=]() {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min");
self->quit(exit_reason::user_shutdown);
}
};
}
behavior connection_helper(stateful_actor<connection_helper_state>* self, behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b) { actor system_broker) {
CAF_LOG_TRACE(CAF_ARG(b)); CAF_LOG_TRACE(CAF_ARG(system_broker));
self->monitor(b); self->monitor(system_broker);
self->set_down_handler([=](down_msg& dm) { self->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm)); CAF_LOG_TRACE(CAF_ARG(dm));
self->quit(std::move(dm.reason)); self->quit(std::move(dm.reason));
}); });
return { return {
// this config is send from the remote `ConfigServ` // this config is send from the remote `PeerServ`
[=](const std::string& item, message& msg) { [=](const std::string& item, message& msg) {
CAF_LOG_TRACE(CAF_ARG(item) << CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(item) << CAF_ARG(msg));
CAF_LOG_DEBUG("received requested config:" << CAF_ARG(msg)); CAF_LOG_DEBUG("received requested address:" << CAF_ARG(msg));
// whatever happens, we are done afterwards // whatever happens, we are done afterwards
self->quit(); self->quit();
msg.apply({ msg.apply({
[&](uint16_t port, network::address_listing& addresses) { [&](basp::routing_table::endpoint& ep) {
if (item == "basp.default-connectivity-tcp") { auto port = ep.first;
auto& mx = self->system().middleman().backend(); auto& addrs = ep.second;
for (auto& kvp : addresses) { auto& mx = self->system().middleman().backend();
for (auto& addr : kvp.second) { for (auto& kvp : addrs) {
auto hdl = mx.new_tcp_scribe(addr, port); for (auto& addr : kvp.second) {
if (hdl) { auto hdl = mx.new_tcp_scribe(addr, port);
// gotcha! send scribe to our BASP broker if (hdl) {
// to initiate handshake etc. // Gotcha! Send scribe to our broker to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr)); CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom::value, *hdl, port); self->send(system_broker, connect_atom::value, *hdl, port);
return; return;
}
} }
} }
CAF_LOG_INFO("could not connect to node directly");
} else if (item == "basp.default-connectivity-udp") {
auto& sys = self->system();
// create new broker to try addresses for communication via UDP
if (get_or(sys.config(), "middleman.attach-utility-actors", false))
self->system().middleman().spawn_broker<hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
else
self->system().middleman().spawn_broker<detached + hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
} }
CAF_LOG_ERROR("could not connect to node ");
} }
}); });
}, },
after(autoconnect_timeout) >> [=] { after(autoconnect_timeout) >> [=] {
CAF_LOG_TRACE(CAF_ARG("")); CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then // nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min"); CAF_LOG_INFO("aborted connection attempt after 10min");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
} }
}; };
......
...@@ -55,7 +55,7 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -55,7 +55,7 @@ connection_state instance::handle(execution_unit* ctx,
callee_.purge_state(nid); callee_.purge_state(nid);
return none; return none;
}); });
tbl_.erase_direct(dm.handle, cb); tbl_.erase(dm.handle, cb);
return close_connection; return close_connection;
}; };
std::vector<char>* payload = nullptr; std::vector<char>* payload = nullptr;
...@@ -86,27 +86,27 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -86,27 +86,27 @@ connection_state instance::handle(execution_unit* ctx,
void instance::handle_heartbeat(execution_unit* ctx) { void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) { for (auto& kvp: tbl_.nid_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second)); CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
write_heartbeat(ctx, callee_.get_buffer(kvp.first)); write_heartbeat(ctx, callee_.get_buffer(kvp.first));
callee_.flush(kvp.first); callee_.flush(kvp.first);
} }
} }
optional<routing_table::route> instance::lookup(const node_id& target) { routing_table::lookup_result instance::lookup(const node_id& target) {
return tbl_.lookup(target); return tbl_.lookup(target);
} }
void instance::flush(const routing_table::route& path) { void instance::flush(connection_handle hdl) {
callee_.flush(path.hdl); callee_.flush(hdl);
} }
void instance::write(execution_unit* ctx, const routing_table::route& r, void instance::write(execution_unit* ctx, connection_handle hdl,
header& hdr, payload_writer* writer) { header& hdr, payload_writer* writer) {
CAF_LOG_TRACE(CAF_ARG(hdr)); CAF_LOG_TRACE(CAF_ARG(hdr));
CAF_ASSERT(hdr.payload_len == 0 || writer != nullptr); CAF_ASSERT(hdr.payload_len == 0 || writer != nullptr);
write(ctx, callee_.get_buffer(r.hdl), hdr, writer); write(ctx, callee_.get_buffer(hdl), hdr, writer);
flush(r); flush(hdl);
} }
void instance::add_published_actor(uint16_t port, void instance::add_published_actor(uint16_t port,
...@@ -169,8 +169,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -169,8 +169,8 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(receiver) CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(receiver)
<< CAF_ARG(mid) << CAF_ARG(msg)); << CAF_ARG(mid) << CAF_ARG(msg));
CAF_ASSERT(receiver && system().node() != receiver->node()); CAF_ASSERT(receiver && system().node() != receiver->node());
auto path = lookup(receiver->node()); auto lr = lookup(receiver->node());
if (!path) { if (!lr.known) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg); notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false; return false;
} }
...@@ -180,9 +180,14 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender, ...@@ -180,9 +180,14 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
}); });
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(), header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(),
sender ? sender->id() : invalid_actor_id, receiver->id()}; sender ? sender->id() : invalid_actor_id, receiver->id()};
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer); if (lr.hdl) {
flush(*path); auto hdl = std::move(*lr.hdl);
notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg); write(ctx, callee_.get_buffer(hdl), hdr, &writer);
flush(hdl);
} else {
write(ctx, callee_.get_buffer(receiver->node()), hdr, &writer);
}
notify<hook::message_sent>(sender, receiver->node(), receiver, mid, msg);
return true; return true;
} }
...@@ -224,7 +229,7 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf, ...@@ -224,7 +229,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 +252,8 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -247,7 +252,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};
...@@ -295,6 +301,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -295,6 +301,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
switch (hdr.operation) { switch (hdr.operation) {
case message_type::server_handshake: { case message_type::server_handshake: {
node_id source_node; node_id source_node;
basp::routing_table::endpoint autoconn_addr;
actor_id aid = invalid_actor_id; actor_id aid = invalid_actor_id;
std::set<std::string> sigs; std::set<std::string> sigs;
if (!payload_valid()) { if (!payload_valid()) {
...@@ -303,7 +310,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -303,7 +310,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
} else { } else {
binary_deserializer bd{ctx, *payload}; binary_deserializer bd{ctx, *payload};
std::string remote_appid; std::string remote_appid;
if (bd(source_node, remote_appid)) if (bd(source_node, remote_appid, autoconn_addr))
return false; return false;
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);
...@@ -320,25 +327,28 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -320,25 +327,28 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
callee_.finalize_handshake(source_node, aid, sigs); callee_.finalize_handshake(source_node, aid, sigs);
return false; return false;
} }
auto lr = tbl_.lookup(source_node);
// Close redundant connections. // Close redundant connections.
if (tbl_.lookup_direct(source_node)) { if (lr.hdl) {
CAF_LOG_DEBUG("close redundant connection:" << CAF_ARG(source_node)); CAF_LOG_DEBUG("close redundant connection:" << CAF_ARG(source_node));
callee_.finalize_handshake(source_node, aid, sigs); callee_.finalize_handshake(source_node, aid, sigs);
return false; return false;
} }
// 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); if (lr.known)
// write handshake as client in response tbl_.handle(source_node, hdl);
auto path = tbl_.lookup(source_node); else
if (!path) { tbl_.add(source_node, hdl);
CAF_LOG_ERROR("no route to host after server handshake"); // Store autoconnect address.
return false; auto peer_server = system().registry().get(atom("PeerServ"));
} anon_send(actor_cast<actor>(peer_server), put_atom::value,
write_client_handshake(ctx, callee_.get_buffer(path->hdl)); to_string(source_node), make_message(std::move(autoconn_addr)));
callee_.learned_new_node_directly(source_node, false); write_client_handshake(ctx, callee_.get_buffer(hdl));
flush(hdl);
callee_.learned_new_node(source_node);
callee_.finalize_handshake(source_node, aid, sigs); callee_.finalize_handshake(source_node, aid, sigs);
flush(*path); callee_.send_buffered_messages(ctx, source_node, hdl);
break; break;
} }
case message_type::client_handshake: { case message_type::client_handshake: {
...@@ -349,7 +359,9 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -349,7 +359,9 @@ 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;
// TODO: Read addrs separately.
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");
...@@ -358,22 +370,32 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -358,22 +370,32 @@ 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 (tbl_.lookup_direct(source_node)) { auto lr = tbl_.lookup(source_node);
if (lr.hdl) {
CAF_LOG_DEBUG("received second client handshake:" CAF_LOG_DEBUG("received second client handshake:"
<< CAF_ARG(source_node)); << CAF_ARG(source_node));
break; break;
} }
// Add route to this node. // Add this node to our contacts.
CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node)); CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node); // Either add a new node or add the handle to a known one.
callee_.learned_new_node_directly(source_node, false); if (lr.known)
tbl_.handle(source_node, hdl);
else
tbl_.add(source_node, hdl);
callee_.learned_new_node(source_node);
callee_.send_buffered_messages(ctx, source_node, hdl);
// 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: {
// Sanity checks. // Sanity checks.
if (!payload_valid()) if (!payload_valid())
return false; return false;
auto source_node = tbl_.lookup_direct(hdl); auto source_node = tbl_.lookup(hdl);
if (source_node == none) { if (source_node == none) {
CAF_LOG_ERROR("received dispatch_message before handshake"); CAF_LOG_ERROR("received dispatch_message before handshake");
return false; return false;
...@@ -402,7 +424,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -402,7 +424,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
break; break;
} }
case message_type::announce_proxy: { case message_type::announce_proxy: {
auto source_node = tbl_.lookup_direct(hdl); auto source_node = tbl_.lookup(hdl);
if (source_node == none) { if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake"); CAF_LOG_ERROR("received announce_proxy before handshake");
return false; return false;
...@@ -414,7 +436,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -414,7 +436,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
// Sanity checks. // Sanity checks.
if (!payload_valid()) if (!payload_valid())
return false; return false;
auto source_node = tbl_.lookup_direct(hdl); auto source_node = tbl_.lookup(hdl);
if (source_node == none) { if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake"); CAF_LOG_ERROR("received announce_proxy before handshake");
return false; return false;
...@@ -430,7 +452,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, ...@@ -430,7 +452,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
break; break;
} }
case message_type::heartbeat: { case message_type::heartbeat: {
auto source_node = tbl_.lookup_direct(hdl); auto source_node = tbl_.lookup(hdl);
if (source_node == none) { if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake"); CAF_LOG_ERROR("received announce_proxy before handshake");
return false; return false;
......
...@@ -389,7 +389,8 @@ void middleman::init(actor_system_config& cfg) { ...@@ -389,7 +389,8 @@ void middleman::init(actor_system_config& cfg) {
.add_message_type<accept_handle>("@accept_handle") .add_message_type<accept_handle>("@accept_handle")
.add_message_type<connection_handle>("@connection_handle") .add_message_type<connection_handle>("@connection_handle")
.add_message_type<connection_passivated_msg>("@connection_passivated_msg") .add_message_type<connection_passivated_msg>("@connection_passivated_msg")
.add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg"); .add_message_type<acceptor_passivated_msg>("@acceptor_passivated_msg")
.add_message_type<basp::routing_table::endpoint>("@autoconnect_endpoint");
// compute and set ID for this network node // compute and set ID for this network node
node_id this_node{node_id::data::create_singleton()}; node_id this_node{node_id::data::create_singleton()};
system().node_.swap(this_node); system().node_.swap(this_node);
......
...@@ -33,123 +33,96 @@ routing_table::~routing_table() { ...@@ -33,123 +33,96 @@ routing_table::~routing_table() {
// nop // nop
} }
optional<routing_table::route> routing_table::lookup(const node_id& target) { node_id routing_table::lookup(const connection_handle& hdl) const {
auto hdl = lookup_direct(target); return get_opt(nid_by_hdl_, hdl, none);
if (hdl)
return route{target, *hdl};
// pick first available indirect route
auto i = indirect_.find(target);
if (i != indirect_.end()) {
auto& hops = i->second;
while (!hops.empty()) {
auto& hop = *hops.begin();
hdl = lookup_direct(hop);
if (hdl)
return route{hop, *hdl};
hops.erase(hops.begin());
}
}
return none;
} }
node_id routing_table::lookup_direct(const connection_handle& hdl) const { routing_table::lookup_result routing_table::lookup(const node_id& nid) const {
return get_opt(direct_by_hdl_, hdl, none); auto i = node_information_base_.find(nid);
if (i != node_information_base_.end())
return {true, i->second.hdl};
return {false, none};
} }
optional<connection_handle> void routing_table::erase(const connection_handle& hdl, erase_callback& cb) {
routing_table::lookup_direct(const node_id& nid) const { auto i = nid_by_hdl_.find(hdl);
auto i = direct_by_nid_.find(nid); if (i == nid_by_hdl_.end())
if (i != direct_by_nid_.end()) return;
return i->second; cb(i->second);
return none; parent_->parent().notify<hook::connection_lost>(i->second);
node_information_base_.erase(i->second);
nid_by_hdl_.erase(i->first);
// TODO: Look through other nodes and remove this one as an origin?
} }
node_id routing_table::lookup_indirect(const node_id& nid) const { void routing_table::add(const node_id& nid, const connection_handle& hdl) {
auto i = indirect_.find(nid); CAF_ASSERT(nid_by_hdl_.count(hdl) == 0);
if (i == indirect_.end()) CAF_ASSERT(node_information_base_.count(nid) == 0);
return none; nid_by_hdl_.emplace(hdl, nid);
if (i->second.empty()) node_information_base_[nid] = node_info{hdl, none};
return none; parent_->parent().notify<hook::new_connection_established>(nid);
return *i->second.begin();
} }
void routing_table::blacklist(const node_id& hop, const node_id& dest) { void routing_table::add(const node_id& nid, const node_id& origin) {
blacklist_[dest].emplace(hop); CAF_ASSERT(node_information_base_.count(nid) == 0);
auto i = indirect_.find(dest); node_information_base_[nid] = node_info{none, origin};
if (i == indirect_.end()) // TODO: Some new related hook?
return; //parent_->parent().notify<hook::new_connection_established>(nid);
i->second.erase(hop);
if (i->second.empty())
indirect_.erase(i);
} }
void routing_table::erase_direct(const connection_handle& hdl, void routing_table::add(const node_id& nid) {
erase_callback& cb) { //CAF_ASSERT(hdl_by_nid_.count(nid) == 0);
auto i = direct_by_hdl_.find(hdl); CAF_ASSERT(node_information_base_.count(nid) == 0);
if (i == direct_by_hdl_.end()) node_information_base_[nid] = node_info{none, none};
return; // TODO: Some new related hook?
cb(i->second); //parent_->parent().notify<hook::new_connection_established>(nid);
parent_->parent().notify<hook::connection_lost>(i->second);
direct_by_nid_.erase(i->second);
direct_by_hdl_.erase(i->first);
} }
bool routing_table::erase_indirect(const node_id& dest) { bool routing_table::reachable(const node_id& dest) {
auto i = indirect_.find(dest); auto i = node_information_base_.find(dest);
if (i == indirect_.end()) if (i != node_information_base_.end())
return i->second.hdl != none;
return false;
}
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; return false;
if (parent_->parent().has_hook()) i->second.origin = origin;
for (auto& nid : i->second)
parent_->parent().notify<hook::route_lost>(nid, dest);
indirect_.erase(i);
return true; return true;
} }
void routing_table::add_direct(const connection_handle& hdl, optional<node_id> routing_table::origin(const node_id& nid) {
const node_id& nid) { auto i = node_information_base_.find(nid);
CAF_ASSERT(direct_by_hdl_.count(hdl) == 0); if (i == node_information_base_.end())
CAF_ASSERT(direct_by_nid_.count(nid) == 0); return none;
direct_by_hdl_.emplace(hdl, nid); return i->second.origin;
direct_by_nid_.emplace(nid, hdl); }
parent_->parent().notify<hook::new_connection_established>(nid);
bool routing_table::handle(const node_id& nid, const connection_handle& hdl) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.hdl = hdl;
nid_by_hdl_.emplace(hdl, nid);
return true;
} }
bool routing_table::add_indirect(const node_id& hop, const node_id& dest) { optional<connection_handle> routing_table::handle(const node_id& nid) {
auto i = blacklist_.find(dest); auto i = node_information_base_.find(nid);
if (i == blacklist_.end() || i->second.count(hop) == 0) { if (i == node_information_base_.end())
auto& hops = indirect_[dest]; return none;
auto added_first = hops.empty(); return i->second.hdl;
hops.emplace(hop);
parent_->parent().notify<hook::new_route_added>(hop, dest);
return added_first;
}
return false; // blacklisted
} }
bool routing_table::reachable(const node_id& dest) { const routing_table::endpoint& routing_table::autoconnect_endpoint() {
return direct_by_nid_.count(dest) > 0 || indirect_.count(dest) > 0; return autoconnect_endpoint_;
} }
size_t routing_table::erase(const node_id& dest, erase_callback& cb) { void routing_table::autoconnect_endpoint(uint16_t port,
cb(dest); network::address_listing addrs) {
size_t res = 0; autoconnect_endpoint_ = {port, std::move(addrs)};
auto i = indirect_.find(dest);
if (i != indirect_.end()) {
res = i->second.size();
for (auto& nid : i->second) {
cb(nid);
parent_->parent().notify<hook::route_lost>(nid, dest);
}
indirect_.erase(i);
}
auto hdl = lookup_direct(dest);
if (hdl) {
direct_by_hdl_.erase(*hdl);
direct_by_nid_.erase(dest);
parent_->parent().notify<hook::connection_lost>(dest);
++res;
}
return res;
} }
} // namespace basp } // namespace basp
......
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
#include <thread> #include <thread>
#include <vector> #include <vector>
#include "caf/test/io_dsl.hpp"
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/io/all.hpp" #include "caf/io/all.hpp"
...@@ -39,6 +41,14 @@ using std::string; ...@@ -39,6 +41,14 @@ using std::string;
using ping_atom = atom_constant<atom("ping")>; using ping_atom = atom_constant<atom("ping")>;
using pong_atom = atom_constant<atom("pong")>; using pong_atom = atom_constant<atom("pong")>;
using set_atom = atom_constant<atom("set")>;
using begin_atom = atom_constant<atom("begin")>;
using middle_atom = atom_constant<atom("middle")>;
using end_atom = atom_constant<atom("end")>;
using msg_atom = atom_constant<atom("msg")>;
using done_atom = atom_constant<atom("shutdown")>;
/* /*
This test checks whether automatic connections work as expected This test checks whether automatic connections work as expected
...@@ -65,223 +75,282 @@ using pong_atom = atom_constant<atom("pong")>; ...@@ -65,223 +75,282 @@ using pong_atom = atom_constant<atom("pong")>;
*/ */
/* namespace {
std::thread run_prog(const char* arg, uint16_t port, bool use_asio) {
return detail::run_sub_unit_test(invalid_actor, constexpr uint16_t port_earth = 12340;
test::engine::path(), constexpr uint16_t port_mars = 12341;
test::engine::max_runtime(), constexpr uint16_t port_jupiter = 12342;
CAF_XSTR(CAF_SUITE),
use_asio,
{"--port=" + std::to_string(port), arg});
}
// we run the same code on all three nodes, a simple ping-pong client // Used for the tests with the test backend.
struct testee_state { class config : public actor_system_config {
std::set<actor> buddies; public:
uint16_t port = 0; config() {
const char* name = "testee"; load<caf::io::middleman, io::network::test_multiplexer>();
set("scheduler.policy", caf::atom("testing"));
set("middleman.detach-utility-actors", false);
set("middleman.enable-automatic-connections", true);
set("middleman.enable-tcp", true);
set("middleman.enable-udp", false);
}
}; };
behavior testee(stateful_actor<testee_state>* self) { // Used for the tests with the default multiplexer backend.
class simple_config : public actor_system_config {
public:
simple_config() {
load<caf::io::middleman>();
set("middleman.enable-automatic-connections", true);
set("middleman.enable-tcp", true);
set("middleman.enable-udp", false);
}
};
class fixture {
public:
fixture() : earth(cfg_earth), mars(cfg_mars), jupiter(cfg_jupiter) {
CAF_MESSAGE("Earth : " << to_string(earth.node()));
CAF_MESSAGE("Mars : " << to_string(mars.node()));
CAF_MESSAGE("Jupiter: " << to_string(jupiter.node()));
}
simple_config cfg_earth;
simple_config cfg_mars;
simple_config cfg_jupiter;
actor_system earth{cfg_earth};
actor_system mars{cfg_mars};
actor_system jupiter{cfg_jupiter};
};
struct cache {
actor tmp;
};
behavior test_actor(stateful_actor<cache>* self, std::string location,
bool quit_directly) {
return { return {
[self](ping_atom, actor buddy, bool please_broadcast) -> message { [=](set_atom, actor val) {
if (please_broadcast) self->state.tmp = val;
for (auto& x : self->state.buddies)
if (x != buddy)
send_as(buddy, x, ping_atom::value, buddy, false);
self->state.buddies.emplace(std::move(buddy));
return make_message(pong_atom::value, self);
}, },
[self](pong_atom, actor buddy) { [=](begin_atom) {
self->state.buddies.emplace(std::move(buddy)); CAF_REQUIRE(self->state.tmp);
CAF_MESSAGE("starting messaging on " << location);
self->send(self->state.tmp, middle_atom::value, self);
}, },
[self](put_atom, uint16_t new_port) { [=](middle_atom, actor start) {
self->state.port = new_port; CAF_REQUIRE(self->state.tmp);
CAF_MESSAGE("forwaring message on " << location);
self->send(self->state.tmp, end_atom::value, start, self);
}, },
[self](get_atom) { [=](end_atom, actor start, actor middle) {
return self->state.port; CAF_MESSAGE("message arrived on " << location);
if (quit_directly) {
CAF_MESSAGE("telling other nodes to quit from " << location);
self->send(start, done_atom::value);
self->send(middle, done_atom::value);
self->send(self, done_atom::value);
} else {
CAF_MESSAGE("telling intermediate node to quit from " << location);
self->state.tmp = start;
self->send(middle, done_atom::value);
}
},
[=](msg_atom) {
CAF_REQUIRE(self->state.tmp);
CAF_MESSAGE("telling tmp actor to quit from " << location);
self->send(self->state.tmp, done_atom::value);
self->send(self, done_atom::value);
},
[=](done_atom) {
CAF_MESSAGE("actor on " << location << " is quitting");
self->quit();
} }
}; };
} }
void run_earth(bool use_asio, bool as_server, uint16_t pub_port) { } // namespace <anonymous>
scoped_actor self{system};
struct captain : hook {
public:
captain(actor parent) : parent_(std::move(parent)) {
// nop
}
void new_connection_established_cb(const node_id& node) override {
anon_send(parent_, put_atom::value, node);
call_next<hook::new_connection_established>(node);
}
void new_remote_actor_cb(const actor_addr& addr) override { CAF_TEST_FIXTURE_SCOPE(autoconn_tcp_simple_test, fixture)
anon_send(parent_, put_atom::value, addr);
call_next<hook::new_remote_actor>(addr);
}
void connection_lost_cb(const node_id& dest) override { CAF_TEST(build_triangle_simple_tcp) {
anon_send(parent_, delete_atom::value, dest); CAF_MESSAGE("setting up Earth");
} auto on_earth = earth.spawn(test_actor, "Earth", true);
auto earth_port = earth.middleman().publish(on_earth, 0);
CAF_REQUIRE(earth_port);
CAF_MESSAGE("Earth reachable via " << *earth_port);
CAF_MESSAGE("setting up Mars");
auto from_earth = mars.middleman().remote_actor("localhost", *earth_port);
CAF_REQUIRE(from_earth);
auto on_mars = mars.spawn(test_actor, "Mars", true);
anon_send(on_mars, set_atom::value, *from_earth);
auto mars_port = mars.middleman().publish(on_mars, 0);
CAF_REQUIRE(mars_port);
CAF_MESSAGE("Mars reachable via " << *mars_port);
CAF_MESSAGE("setting up Jupiter");
auto from_mars = jupiter.middleman().remote_actor("localhost", *mars_port);
CAF_REQUIRE(from_mars);
auto on_jupiter = jupiter.spawn(test_actor, "Jupiter", true);
anon_send(on_jupiter, set_atom::value, *from_mars);
CAF_MESSAGE("forwarding an actor from Jupiter to Earth via Mars");
anon_send(on_jupiter, begin_atom::value);
jupiter.await_all_actors_done();
mars.await_all_actors_done();
earth.await_all_actors_done();
}
private: CAF_TEST(break_triangle_simple_tcp) {
actor parent_; actor on_earth;
}; actor on_jupiter;
middleman::instance()->add_hook<captain>(self); {
auto aut = system.spawn(testee); simple_config conf;
auto port = publish(aut, pub_port); actor_system mars(conf);
CAF_MESSAGE("published testee at port " << port); // Earth.
std::thread mars_process; CAF_MESSAGE("setting up Earth");
std::thread jupiter_process; on_earth = earth.spawn(test_actor, "Earth", false);
// launch process for Mars auto earth_port = earth.middleman().publish(on_earth, 0);
if (!as_server) { CAF_REQUIRE(earth_port);
CAF_MESSAGE("launch process for Mars"); CAF_MESSAGE("Earth reachable via " << *earth_port);
mars_process = run_prog("--mars", port, use_asio); // Mars.
CAF_MESSAGE("setting up Mars");
auto from_earth = mars.middleman().remote_actor("localhost", *earth_port);
CAF_REQUIRE(from_earth);
auto on_mars = mars.spawn(test_actor, "Mars", false);
anon_send(on_mars, set_atom::value, *from_earth);
auto mars_port = mars.middleman().publish(on_mars, 0);
CAF_REQUIRE(mars_port);
CAF_MESSAGE("Mars reachable via " << *mars_port);
// Jupiter.
CAF_MESSAGE("setting up Jupiter");
auto from_mars = jupiter.middleman().remote_actor("localhost", *mars_port);
CAF_REQUIRE(from_mars);
on_jupiter = jupiter.spawn(test_actor, "Jupiter", false);
anon_send(on_jupiter, set_atom::value, *from_mars);
// Trigger the connection setup.
CAF_MESSAGE("forwarding an actor from Jupiter to Earth via Mars");
anon_send(on_jupiter, begin_atom::value);
mars.await_all_actors_done();
// Leaving the scope will shutdown Mars.
} }
CAF_MESSAGE("wait for Mars to connect"); // Let the remaining nodes communicate.
node_id mars; anon_send(on_earth, msg_atom::value);
self->receive( jupiter.await_all_actors_done();
[&](put_atom, const node_id& nid) { earth.await_all_actors_done();
mars = nid;
CAF_MESSAGE(CAF_ARG(mars));
}
);
actor_addr mars_addr;
uint16_t mars_port;
self->receive_while([&] { return mars_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->request(hdl, sys_atom::value, get_atom::value, "info").then(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
mars_addr = addr;
CAF_MESSAGE(CAF_ARG(mars_addr));
self->request(actor_cast<actor>(mars_addr), get_atom::value).then(
[&](uint16_t mp) {
CAF_MESSAGE("mars published its actor at port " << mp);
mars_port = mp;
}
);
}
);
}
);
// launch process for Jupiter
if (!as_server) {
CAF_MESSAGE("launch process for Jupiter");
jupiter_process = run_prog("--jupiter", mars_port, use_asio);
}
CAF_MESSAGE("wait for Jupiter to connect");
self->receive(
[](put_atom, const node_id& jupiter) {
CAF_MESSAGE(CAF_ARG(jupiter));
}
);
actor_addr jupiter_addr;
self->receive_while([&] { return jupiter_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr);
self->request(hdl, sys_atom::value, get_atom::value, "info").then(
[&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee")
return;
jupiter_addr = addr;
CAF_MESSAGE(CAF_ARG(jupiter_addr));
}
);
}
);
CAF_MESSAGE("shutdown Mars");
anon_send_exit(mars_addr, exit_reason::kill);
if (mars_process.joinable())
mars_process.join();
self->receive(
[&](delete_atom, const node_id& nid) {
CAF_CHECK(nid == mars);
}
);
CAF_MESSAGE("check whether we still can talk to Jupiter");
self->send(aut, ping_atom::value, self, true);
std::set<actor_addr> found;
int i = 0;
self->receive_for(i, 2)(
[&](pong_atom, const actor&) {
found.emplace(self->current_sender());
}
);
std::set<actor_addr> expected{aut.address(), jupiter_addr};
CAF_CHECK(found == expected);
CAF_MESSAGE("shutdown Jupiter");
anon_send_exit(jupiter_addr, exit_reason::kill);
if (jupiter_process.joinable())
jupiter_process.join();
anon_send_exit(aut, exit_reason::kill);
} }
void run_mars(uint16_t port_to_earth, uint16_t pub_port) { CAF_TEST_FIXTURE_SCOPE_END()
auto aut = system.spawn(testee);
auto port = publish(aut, pub_port);
anon_send(aut, put_atom::value, port);
CAF_MESSAGE("published testee at port " << port);
auto earth = remote_actor("localhost", port_to_earth);
send_as(aut, earth, ping_atom::value, aut, false);
}
void run_jupiter(uint16_t port_to_mars) { CAF_TEST_FIXTURE_SCOPE(autoconn_tcp_test,
auto aut = system.spawn(testee); belt_fixture<test_coordinator_fixture<config>>)
auto mars = remote_actor("localhost", port_to_mars);
send_as(aut, mars, ping_atom::value, aut, true);
}
*/
CAF_TEST(triangle_setup) { CAF_TEST(build_triangle_tcp) {
// this unit test is temporarily disabled until problems CAF_MESSAGE("Earth : " << to_string(earth.sys.node()));
// with OBS are sorted out or new actor_system API is in place CAF_MESSAGE("Mars : " << to_string(mars.sys.node()));
CAF_MESSAGE("Jupiter: " << to_string(jupiter.sys.node()));
// Earth.
CAF_MESSAGE("setting up Earth");
auto on_earth = earth.sys.spawn(test_actor, "Earth", true);
CAF_MESSAGE("run initialization code");
exec_all();
CAF_MESSAGE("prepare connection");
prepare_connection(earth, mars, "earth", port_earth);
CAF_MESSAGE("publish dummy on earth");
earth.publish(on_earth, port_earth);
// Mars.
CAF_MESSAGE("setting up Mars");
auto from_earth = mars.remote_actor("earth", port_earth);
CAF_REQUIRE(from_earth);
auto on_mars = mars.sys.spawn(test_actor, "Mars", true);
anon_send(on_mars, set_atom::value, from_earth);
CAF_MESSAGE("run initialization code");
exec_all();
CAF_MESSAGE("prepare connection");
prepare_connection(mars, jupiter, "mars", port_mars);
CAF_MESSAGE("publish dummy on mars");
mars.publish(on_mars, port_mars);
// Jupiter
CAF_MESSAGE("setting up Jupiter");
auto from_mars = jupiter.remote_actor("mars", port_mars);
CAF_REQUIRE(from_mars);
auto on_jupiter = jupiter.sys.spawn(test_actor, "Jupiter", true);
anon_send(on_jupiter, set_atom::value, from_mars);
exec_all();
// This handle will be created by the test multiplexer for the automatically
// opened socket when automatic connections are enabled.
auto hdl_jupiter = accept_handle::from_int(std::numeric_limits<int64_t>::max());
// Prepare automatic connection between Jupiter and Earth.
prepare_connection(jupiter, earth, "jupiter", port_jupiter, hdl_jupiter);
// Add the address information for this test to the config server on Mars.
auto mars_config_server = mars.sys.registry().get(atom("PeerServ"));
network::address_listing interfaces{
{network::protocol::ipv4, std::vector<std::string>{"jupiter"}}
};
basp::routing_table::endpoint addrs{port_jupiter, interfaces};
anon_send(actor_cast<actor>(mars_config_server), put_atom::value,
to_string(jupiter.sys.node()), make_message(addrs));
// Trigger the automatic connection setup.
CAF_MESSAGE("forwarding an actor from Jupiter to Earth via Mars");
anon_send(on_jupiter, begin_atom::value);
exec_all();
} }
/* CAF_TEST(break_triangle_tcp) {
CAF_TEST(triangle_setup) { CAF_MESSAGE("Earth : " << to_string(earth.sys.node()));
uint16_t port = 0; CAF_MESSAGE("Mars : " << to_string(mars.sys.node()));
uint16_t publish_port = 0; CAF_MESSAGE("Jupiter: " << to_string(jupiter.sys.node()));
auto argv = test::engine::argv(); // Earth.
auto argc = test::engine::argc(); CAF_MESSAGE("setting up Earth");
auto r = message_builder(argv, argv + argc).extract_opts({ auto on_earth = earth.sys.spawn(test_actor, "Earth", false);
{"port,p", "port of remote side (when running mars or jupiter)", port}, CAF_MESSAGE("run initialization code");
{"mars", "run mars"}, exec_all();
{"jupiter", "run jupiter"}, CAF_MESSAGE("prepare connection");
{"use-asio", "use ASIO network backend (if available)"}, prepare_connection(earth, mars, "earth", port_earth);
{"server,s", "run in server mode (don't run clients)", publish_port} CAF_MESSAGE("publish dummy on earth");
}); earth.publish(on_earth, port_earth);
// check arguments // Mars.
bool is_mars = r.opts.count("mars") > 0; CAF_MESSAGE("setting up Mars");
bool is_jupiter = r.opts.count("jupiter") > 0; auto from_earth = mars.remote_actor("earth", port_earth);
bool has_port = r.opts.count("port") > 0; CAF_REQUIRE(from_earth);
if (((is_mars || is_jupiter) && !has_port) || (is_mars && is_jupiter)) { auto on_mars = mars.sys.spawn(test_actor, "Mars", false);
CAF_ERROR("need a port when running Mars or Jupiter and cannot " anon_send(on_mars, set_atom::value, from_earth);
"both at the same time"); CAF_MESSAGE("run initialization code");
return; exec_all();
} CAF_MESSAGE("prepare connection");
// enable automatic connections prepare_connection(mars, jupiter, "mars", port_mars);
anon_send(whereis(atom("ConfigServ")), put_atom::value, CAF_MESSAGE("publish dummy on mars");
"middleman.enable-automatic-connections", make_message(true)); mars.publish(on_mars, port_mars);
auto use_asio = r.opts.count("use-asio") > 0; // Jupiter.
# ifdef CAF_USE_ASIO CAF_MESSAGE("setting up Jupiter");
if (use_asio) { auto from_mars = jupiter.remote_actor("mars", port_mars);
CAF_MESSAGE("enable ASIO backend"); CAF_REQUIRE(from_mars);
set_middleman<network::asio_multiplexer>(); auto on_jupiter = jupiter.sys.spawn(test_actor, "Jupiter", false);
} anon_send(on_jupiter, set_atom::value, from_mars);
# endif // CAF_USE_ASIO exec_all();
auto as_server = r.opts.count("server") > 0; // This handle will be created by the test multiplexer for the automatically
if (is_mars) // opened socket when automatic connections are enabled.
run_mars(port, publish_port); auto hdl_jupiter = accept_handle::from_int(std::numeric_limits<int64_t>::max());
else if (is_jupiter) // Prepare automatic connection between Jupiter and Earth.
run_jupiter(port); prepare_connection(jupiter, earth, "jupiter", port_jupiter, hdl_jupiter);
else // Add the address information for this test to the config server on Mars.
run_earth(use_asio, as_server, publish_port); auto mars_config_server = mars.sys.registry().get(atom("PeerServ"));
await_all_actors_done(); network::address_listing interfaces{
shutdown(); {network::protocol::ipv4, std::vector<std::string>{"jupiter"}}
};
basp::routing_table::endpoint addrs{port_jupiter, interfaces};
anon_send(actor_cast<actor>(mars_config_server), put_atom::value,
to_string(jupiter.sys.node()), make_message(addrs));
// Trigger the automatic connection setup between the edge nodes.
CAF_MESSAGE("forwarding an actor from Jupiter to Earth via Mars");
anon_send(on_jupiter, begin_atom::value);
exec_all();
// Shutdown the basp broker of the intermediate node.
using caf::io::basp_broker;
auto bhdl = mars.mm.named_broker<basp_broker>(caf::atom("BASP"));
auto bactor =
dynamic_cast<basp_broker*>(caf::actor_cast<caf::abstract_actor*>(bhdl));
anon_send_exit(bactor, exit_reason::kill);
exec_all();
// Let the remaining nodes communicate.
anon_send(on_earth, msg_atom::value);
exec_all();
} }
*/
CAF_TEST_FIXTURE_SCOPE_END()
...@@ -75,7 +75,6 @@ constexpr uint64_t default_operation_data = make_message_id().integer_value(); ...@@ -75,7 +75,6 @@ constexpr uint64_t default_operation_data = make_message_id().integer_value();
constexpr auto basp_atom = caf::atom("BASP"); constexpr auto basp_atom = caf::atom("BASP");
constexpr auto spawn_serv_atom = caf::atom("SpawnServ"); constexpr auto spawn_serv_atom = caf::atom("SpawnServ");
constexpr auto config_serv_atom = caf::atom("ConfigServ");
} // namespace <anonymous> } // namespace <anonymous>
...@@ -263,7 +262,8 @@ public: ...@@ -263,7 +262,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 +276,10 @@ public: ...@@ -276,10 +276,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,
...@@ -289,9 +289,9 @@ public: ...@@ -289,9 +289,9 @@ public:
make_message(sys_atom::value, get_atom::value, "info")); make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the // test whether basp instance correctly updates the
// routing table upon receiving client handshakes // routing table upon receiving client handshakes
auto path = unbox(tbl().lookup(n.id)); auto lr = tbl().lookup(n.id);
CAF_CHECK_EQUAL(path.hdl, n.connection); CAF_REQUIRE(lr.hdl);
CAF_CHECK_EQUAL(path.next_hop, n.id); CAF_CHECK_EQUAL(*lr.hdl, n.connection);
} }
std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) { std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) {
...@@ -480,7 +480,8 @@ CAF_TEST(non_empty_server_handshake) { ...@@ -480,7 +480,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 +567,12 @@ CAF_TEST(remote_actor_and_send) { ...@@ -566,10 +567,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{}, basp::routing_table::endpoint{},
jupiter().dummy_actor->id(), uint32_t{0})
.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,
...@@ -656,17 +659,10 @@ CAF_TEST_FIXTURE_SCOPE_END() ...@@ -656,17 +659,10 @@ CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture) CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
CAF_TEST_DISABLED(automatic_connection) { CAF_TEST(read_address_after_handshake) {
// this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value,
// "middleman.enable-automatic-connections", make_message(true));
//mpx()->exec_runnable(); // process publish message in basp_broker
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// (this node receives a message from jupiter via mars and responds via mars,
// but then also establishes a connection to jupiter directly)
auto check_node_in_tbl = [&](node& n) { auto check_node_in_tbl = [&](node& n) {
auto hdl = tbl().lookup_direct(n.id); auto lr = tbl().lookup(n.id);
CAF_REQUIRE(hdl); CAF_REQUIRE(lr.hdl);
}; };
mpx()->provide_scribe("jupiter", 8080, jupiter().connection); mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080)); CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
...@@ -676,80 +672,32 @@ CAF_TEST_DISABLED(automatic_connection) { ...@@ -676,80 +672,32 @@ CAF_TEST_DISABLED(automatic_connection) {
publish(self(), 4242); publish(self(), 4242);
mpx()->flush_runnables(); // process publish message in basp_broker mpx()->flush_runnables(); // process publish message in basp_broker
CAF_MESSAGE("connect to mars"); CAF_MESSAGE("connect to mars");
connect_node(mars(), ax, self()->id()); auto ep = instance().tbl().autoconnect_endpoint();
//CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id()); connect_node(mars(), ax, self()->id(), set<string>{}, ep);
check_node_in_tbl(mars()); check_node_in_tbl(mars());
// TODO: this use case is no longer possible. Nodes are required to open a CAF_MESSAGE("Look for mars address information in our peer server");
// connection first, before sending messages (forwarding has been auto peer_server = sys.registry().get(atom("PeerServ"));
// removed). CAF_MESSAGE("Send request");
CAF_MESSAGE("simulate that an actor from jupiter " self()->send(actor_cast<actor>(peer_server), get_atom::value,
"sends a message to us via mars"); to_string(mars().id));
mock(mars().connection, // process get request and send answer
{basp::message_type::dispatch_message, 0, 0, 0, do {
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{}, make_message("hello from jupiter!"))
.receive(mars().connection, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, invalid_actor_id,
spawn_serv_atom, std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(mars().connection, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data,
any_vals, // actor ID of an actor spawned by the BASP broker
invalid_actor_id, config_serv_atom, std::vector<actor_id>{},
make_message(get_atom::value, "basp.default-connectivity-tcp"))
.receive(mars().connection, basp::message_type::announce_proxy, no_flags,
no_payload, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id());
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), mars().id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
auto connection_helper_actor = sys.latest_actor_id();
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
// create a dummy config server and respond to the name lookup
CAF_MESSAGE("receive ConfigServ of jupiter");
network::address_listing res;
res[network::protocol::ipv4].emplace_back("jupiter");
mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0, invalid_actor_id,
connection_helper_actor},
std::vector<actor_id>{},
make_message("basp.default-connectivity-tcp",
make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and
// send the scribe handle over to the BASP broker
while (mpx()->has_pending_scribe("jupiter", 8080)) {
sched.run(); sched.run();
mpx()->flush_runnables(); mpx()->flush_runnables();
} } while (self()->mailbox().empty());
CAF_REQUIRE(mpx()->output_buffer(mars().connection).empty()); CAF_MESSAGE("Process reply");
// send handshake from jupiter
mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().dummy_actor->id(), invalid_actor_id},
std::string{}, jupiter().dummy_actor->id(), uint32_t{0})
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, 1u, no_operation_data, invalid_actor_id,
invalid_actor_id, std::string{});
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
check_node_in_tbl(jupiter());
check_node_in_tbl(mars());
CAF_MESSAGE("receive message from jupiter");
self()->receive( self()->receive(
[](const std::string& str) -> std::string { [&](const std::string& item, message& msg) {
CAF_CHECK_EQUAL(str, "hello from jupiter!"); // Check that we got an entry under the name of our peer.
return "hello from earth!"; 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());
}
);
} }
); );
mpx()->exec_runnable(); // process forwarded message in basp_broker
CAF_MESSAGE("response message must take direct route now");
mock().receive(jupiter().connection, basp::message_type::dispatch_message,
no_flags, any_vals, default_operation_data, self()->id(),
jupiter().dummy_actor->id(), std::vector<actor_id>{},
make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
} }
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