Commit 503bc2f6 authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'topic/improve-autoconnect'

parents b092a645 e781635d
...@@ -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', 'StreamServ', 'PeerServ'}
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;
...@@ -571,6 +579,11 @@ private: ...@@ -571,6 +579,11 @@ private:
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Sets the internal actor for storing the peer addresses.
void peer_serv(strong_actor_ptr x) {
internal_actors_[internal_actor_id(atom("PeerServ"))] = std::move(x);
}
/// Used to generate ascending actor IDs. /// Used to generate ascending actor IDs.
std::atomic<size_t> ids_; std::atomic<size_t> ids_;
......
...@@ -166,6 +166,98 @@ behavior spawn_serv_impl(stateful_actor<spawn_serv_state>* self) { ...@@ -166,6 +166,98 @@ 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 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
...@@ -288,14 +380,17 @@ actor_system::actor_system(actor_system_config& cfg) ...@@ -288,14 +380,17 @@ actor_system::actor_system(actor_system_config& cfg)
if (mod) if (mod)
mod->init(cfg); mod->init(cfg);
groups_.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; 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();
...@@ -316,6 +411,7 @@ actor_system::~actor_system() { ...@@ -316,6 +411,7 @@ actor_system::~actor_system() {
} }
registry_.erase(atom("SpawnServ")); registry_.erase(atom("SpawnServ"));
registry_.erase(atom("ConfigServ")); registry_.erase(atom("ConfigServ"));
registry_.erase(atom("PeerServ"));
registry_.erase(atom("StreamServ")); registry_.erase(atom("StreamServ"));
// group module is the first one, relies on MM // group module is the first one, relies on MM
groups_.stop(); groups_.stop();
......
...@@ -124,7 +124,8 @@ inline bool operator!=(const header& lhs, const header& rhs) { ...@@ -124,7 +124,8 @@ inline bool operator!=(const header& lhs, const header& rhs) {
/// Checks whether given header contains a handshake. /// Checks whether given header contains a handshake.
inline bool is_handshake(const header& hdr) { inline bool is_handshake(const header& hdr) {
return hdr.operation == message_type::server_handshake return hdr.operation == message_type::server_handshake
|| hdr.operation == message_type::client_handshake; || hdr.operation == message_type::client_handshake
|| hdr.operation == message_type::acknowledge_handshake;
} }
/// Checks wheter given header contains a heartbeat. /// Checks wheter given header contains a heartbeat.
......
This diff is collapsed.
...@@ -67,6 +67,10 @@ enum class message_type : uint8_t { ...@@ -67,6 +67,10 @@ enum class message_type : uint8_t {
/// ///
/// ![](heartbeat.png) /// ![](heartbeat.png)
heartbeat = 0x05, heartbeat = 0x05,
/// Last message in a BASP handshake for between two endpoints using UDP
/// as a transport protocol.
acknowledge_handshake = 0x06,
}; };
/// @relates message_type /// @relates message_type
......
...@@ -29,6 +29,8 @@ ...@@ -29,6 +29,8 @@
#include "caf/io/basp/buffer_type.hpp" #include "caf/io/basp/buffer_type.hpp"
#include "caf/io/network/interfaces.hpp"
namespace std { namespace std {
template<> template<>
...@@ -49,73 +51,106 @@ namespace basp { ...@@ -49,73 +51,106 @@ namespace basp {
/// Stores routing information for a single broker participating as /// Stores routing information for a single broker participating as
/// BASP peer and provides both direct and indirect paths. /// BASP peer and provides both direct and indirect paths.
// TODO: Rename `routing_table`.
class routing_table { class routing_table {
public: public:
using endpoint_handle = variant<connection_handle, datagram_handle>; using endpoint_handle = variant<connection_handle, datagram_handle>;
using address_endpoint = std::pair<uint16_t, network::address_listing>;
using address_map = std::map<network::protocol::transport,
address_endpoint>;
explicit routing_table(abstract_broker* parent); 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.
endpoint_handle hdl; bool known;
/// Servant handle to communicate with the node -- if already created.
optional<endpoint_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 endpoint_handle& hdl) const; node_id lookup(const endpoint_handle& hdl) const;
/// Returns the handle offering a direct connection to `nid` or
/// `invalid_connection_handle` if no direct connection to `nid` exists.
optional<endpoint_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 endpoint_handle& hdl, const node_id& nid); void add(const node_id& nid, const endpoint_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 endpoint_handle& hdl, erase_callback& cb); void erase(const endpoint_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 endpoint_handle& hdl);
/// Get the handle for communication with `nid`
/// or `none` if the node is unknown.
optional<endpoint_handle> handle(const node_id& nid);
/// Get the addresses to reach `nid` or `none` if the node is unknown.
optional<const address_map&> addresses(const node_id& nid);
/// Add a port, address pair under a key to the local addresses.
void local_addresses(network::protocol::transport proto,
address_endpoint addrs);
/// Get a reference to an address map for the local node.
const address_map& local_addresses();
// Set the received client handshake flag for `nid`.
bool received_client_handshake(const node_id& nid, bool flag);
// Get the received client handshake flag for `nid`.
bool received_client_handshake(const node_id& nid);
public: public:
/// Entry to bundle information for a remote endpoint.
struct node_info {
/// Handle for the node if communication is established.
optional<endpoint_handle> hdl;
/// The endpoint who told us about the node.
optional<node_id> origin;
/// Track if we received a client handshake to solve simultaneous
/// handshake with UDP.
bool received_client_handshake;
};
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 {
...@@ -125,16 +160,10 @@ public: ...@@ -125,16 +160,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<endpoint_handle, node_id> direct_by_hdl_; std::unordered_map<endpoint_handle, node_id> nid_by_hdl_;
std::unordered_map<node_id, endpoint_handle> direct_by_nid_; std::unordered_map<node_id, node_info> node_information_base_;
indirect_entries indirect_; address_map local_addrs_;
indirect_entries blacklist_;
}; };
/// @} /// @}
......
...@@ -80,14 +80,10 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -80,14 +80,10 @@ 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 // get contact information for `nid` and establish communication
void learned_new_node_directly(const node_id& nid, void establish_communication(const node_id& nid) override;
bool was_indirectly_before) override;
// inherited from basp::instance::callee
void learned_new_node_indirectly(const node_id& nid) override;
// inherited from basp::instance::callee // inherited from basp::instance::callee
uint16_t next_sequence_number(connection_handle hdl) override; uint16_t next_sequence_number(connection_handle hdl) override;
...@@ -107,6 +103,14 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -107,6 +103,14 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::callee // inherited from basp::instance::callee
void drop_pending(basp::endpoint_context& ep, uint16_t seq) override; void drop_pending(basp::endpoint_context& ep, uint16_t seq) override;
// inherited from basp::instance::callee
void send_buffered_messages(execution_unit* ctx, node_id nid,
connection_handle hdl) override;
// inherited from basp::instance::callee
void send_buffered_messages(execution_unit* ctx, node_id nid,
datagram_handle hdl) override;
// inherited from basp::instance::callee // inherited from basp::instance::callee
buffer_type& get_buffer(endpoint_handle hdl) override; buffer_type& get_buffer(endpoint_handle hdl) override;
...@@ -116,6 +120,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -116,6 +120,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::callee // inherited from basp::instance::callee
buffer_type& get_buffer(connection_handle hdl) override; buffer_type& get_buffer(connection_handle hdl) override;
// inherited from basp::instance::callee
buffer_type& get_buffer(node_id nid) override;
// inherited from basp::instance::callee // inherited from basp::instance::callee
buffer_type pop_datagram_buffer(datagram_handle hdl) override; buffer_type pop_datagram_buffer(datagram_handle hdl) override;
...@@ -180,8 +187,12 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -180,8 +187,12 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// maximum queue size for pending messages of endpoints with ordering // maximum queue size for pending messages of endpoints with ordering
const size_t max_pending_messages; const size_t max_pending_messages;
// buffer messages for nodes while connectivity is established
std::unordered_map<node_id, std::vector<buffer_type>> pending_connectivity;
// timeout for delivery of pending messages of endpoints with ordering // timeout for delivery of pending messages of endpoints with ordering
const std::chrono::milliseconds pending_to = std::chrono::milliseconds(100); const std::chrono::milliseconds pending_timeout
= std::chrono::milliseconds(100);
// returns the node identifier of the underlying BASP instance // returns the node identifier of the underlying BASP instance
const node_id& this_node() const { const node_id& this_node() const {
......
...@@ -46,13 +46,13 @@ struct connection_helper_state { ...@@ -46,13 +46,13 @@ struct connection_helper_state {
static const char* name; static const char* name;
}; };
behavior datagram_connection_broker(broker* self, behavior datagram_connection_broker(broker* self, uint16_t port,
uint16_t port,
network::address_listing addresses, network::address_listing addresses,
actor system_broker); actor system_broker,
basp::instance* instance);
behavior connection_helper(stateful_actor<connection_helper_state>* self, behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b); actor b, basp::instance* i);
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -254,7 +254,7 @@ private: ...@@ -254,7 +254,7 @@ private:
std::vector<intrusive_ptr<resumable>> internally_posted_; std::vector<intrusive_ptr<resumable>> internally_posted_;
/// Sequential ids for handles of datagram servants /// Sequential ids for handles of datagram servants
int64_t servant_ids_; std::atomic<int64_t> servant_ids_;
/// Maximum messages per resume run. /// Maximum messages per resume run.
size_t max_throughput_; size_t max_throughput_;
......
...@@ -32,6 +32,7 @@ namespace network { ...@@ -32,6 +32,7 @@ namespace network {
class test_multiplexer : public multiplexer { class test_multiplexer : public multiplexer {
private: private:
struct datagram_endpoint;
struct datagram_data; struct datagram_data;
public: public:
...@@ -96,10 +97,13 @@ public: ...@@ -96,10 +97,13 @@ public:
void provide_acceptor(uint16_t desired_port, accept_handle hdl); void provide_acceptor(uint16_t desired_port, accept_handle hdl);
// Provide a local datagram servant on port with `hdl`.
void provide_datagram_servant(uint16_t desired_port, datagram_handle hdl); void provide_datagram_servant(uint16_t desired_port, datagram_handle hdl);
// Provide a datagram servant for communication a remote endpoint
// with `endpoint_id`.
void provide_datagram_servant(std::string host, uint16_t desired_port, void provide_datagram_servant(std::string host, uint16_t desired_port,
datagram_handle hdl); datagram_handle hdl, intptr_t endpoint_id = 0);
/// Generate an id for a new servant. /// Generate an id for a new servant.
int64_t next_endpoint_id(); int64_t next_endpoint_id();
...@@ -108,10 +112,11 @@ public: ...@@ -108,10 +112,11 @@ public:
using buffer_type = std::vector<char>; using buffer_type = std::vector<char>;
/// Buffers storing bytes for UDP related components. /// Buffers storing bytes for UDP related components.
using endpoint_id_type = std::intptr_t;
using read_buffer_type = network::receive_buffer; using read_buffer_type = network::receive_buffer;
using write_buffer_type = buffer_type; using write_buffer_type = buffer_type;
using read_job_type = std::pair<datagram_handle, read_buffer_type>; using read_job_type = std::pair<endpoint_id_type, read_buffer_type>;
using write_job_type = std::pair<datagram_handle, write_buffer_type>; using write_job_type = std::pair<endpoint_id_type, write_buffer_type>;
using write_job_queue_type = std::deque<write_job_type>; using write_job_queue_type = std::deque<write_job_type>;
using shared_buffer_type = std::shared_ptr<buffer_type>; using shared_buffer_type = std::shared_ptr<buffer_type>;
...@@ -176,8 +181,10 @@ public: ...@@ -176,8 +181,10 @@ public:
datagram_servant_ptr& impl_ptr(datagram_handle hdl); datagram_servant_ptr& impl_ptr(datagram_handle hdl);
/// Returns a map with all servants related to the servant `hdl`. using write_handle_map = std::map<endpoint_id_type, datagram_endpoint>;
std::set<datagram_handle>& servants(datagram_handle hdl);
/// Returns the endpoint for a specific `hdl`.
endpoint_id_type endpoint_id(datagram_handle hdl);
/// Returns `true` if this handle has been closed /// Returns `true` if this handle has been closed
/// for reading, `false` otherwise. /// for reading, `false` otherwise.
...@@ -198,15 +205,27 @@ public: ...@@ -198,15 +205,27 @@ public:
test_multiplexer& peer, std::string host, test_multiplexer& peer, std::string host,
uint16_t port, connection_handle peer_hdl); uint16_t port, connection_handle peer_hdl);
/// Stores `hdl` as a pending endpoint for `src`. void add_pending_endpoint(intptr_t endpoint_id, datagram_handle hdl,
void add_pending_endpoint(datagram_handle src, datagram_handle hdl); shared_job_queue_type write_buffer
= std::make_shared<write_job_queue_type>());
/// Add `hdl` as a pending endpoint to `src` and provide a datagram servant
/// on `peer` that connects the buffers of `hdl` and `peer_hdl`. Calls
/// `add_pending_endpoint(...)` and `peer.provide_endpoint(...)`.
void prepare_endpoints(datagram_handle src, datagram_handle hdl,
test_multiplexer& peer, std::string host,
uint16_t port, datagram_handle peer_hdl);
using pending_connects_map = std::unordered_multimap<accept_handle, using pending_connects_map = std::unordered_multimap<accept_handle,
connection_handle>; connection_handle>;
pending_connects_map& pending_connects(); pending_connects_map& pending_connects();
using pending_endpoints_map = std::unordered_map<int64_t, datagram_handle>; // using pending_endpoints_map = std::unordered_map<int64_t, datagram_handle>;
using pending_endpoints_map = std::map<intptr_t,
std::pair<datagram_handle,
shared_job_queue_type>>;
pending_endpoints_map& pending_endpoints(); pending_endpoints_map& pending_endpoints();
...@@ -219,7 +238,7 @@ public: ...@@ -219,7 +238,7 @@ public:
datagram_handle>; datagram_handle>;
using pending_remote_datagram_endpoints_map using pending_remote_datagram_endpoints_map
= std::map<std::pair<std::string, uint16_t>, datagram_handle>; = std::map<std::pair<std::string, uint16_t>, std::pair<datagram_handle, intptr_t>>;
bool has_pending_scribe(std::string x, uint16_t y); bool has_pending_scribe(std::string x, uint16_t y);
...@@ -237,6 +256,9 @@ public: ...@@ -237,6 +256,9 @@ public:
/// Tries to read data from the external input buffer of `hdl`. /// Tries to read data from the external input buffer of `hdl`.
bool try_read_data(connection_handle hdl); bool try_read_data(connection_handle hdl);
/// Tries to read data from the external input queue of `hdl`.
bool try_read_data(datagram_handle hdl);
/// Poll data on all scribes. /// Poll data on all scribes.
bool read_data(); bool read_data();
...@@ -253,7 +275,7 @@ public: ...@@ -253,7 +275,7 @@ public:
/// Appends `buf` to the virtual network buffer of `hdl` /// Appends `buf` to the virtual network buffer of `hdl`
/// and calls `read_data(hdl)` afterwards. /// and calls `read_data(hdl)` afterwards.
void virtual_send(datagram_handle src, datagram_handle ep, void virtual_send(datagram_handle hdl, endpoint_id_type endpoint,
const buffer_type&); const buffer_type&);
/// Waits until a `runnable` is available and executes it. /// Waits until a `runnable` is available and executes it.
...@@ -323,24 +345,36 @@ private: ...@@ -323,24 +345,36 @@ private:
doorman_data(); doorman_data();
}; };
struct datagram_data { struct datagram_endpoint {
datagram_handle hdl;
shared_job_queue_type vn_buf_ptr; shared_job_queue_type vn_buf_ptr;
shared_job_queue_type wr_buf_ptr; shared_job_queue_type wr_buf_ptr;
write_job_queue_type& vn_buf; write_job_queue_type& vn_buf;
write_job_queue_type& wr_buf; write_job_queue_type& wr_buf;
read_job_type rd_buf;
datagram_endpoint(
datagram_handle hdl = datagram_handle(),
shared_job_queue_type input = std::make_shared<write_job_queue_type>(),
shared_job_queue_type output = std::make_shared<write_job_queue_type>()
);
};
struct datagram_data {
datagram_servant_ptr ptr; datagram_servant_ptr ptr;
datagram_endpoint read_handle;
read_job_type rd_buf;
bool stopped_reading; bool stopped_reading;
bool passive_mode; bool passive_mode;
bool ack_writes; bool ack_writes;
uint16_t port; uint16_t remote_port;
uint16_t local_port; uint16_t local_port;
std::set<datagram_handle> servants; write_handle_map write_handles;
size_t datagram_size; size_t datagram_size;
// Allows creating an entangled scribes where the input of this scribe is // Allows creating an entangled scribes where the input of this scribe is
// the output of another scribe and vice versa. // the output of another scribe and vice versa.
datagram_data( datagram_data(
datagram_handle hdl = datagram_handle{},
shared_job_queue_type input = std::make_shared<write_job_queue_type>(), shared_job_queue_type input = std::make_shared<write_job_queue_type>(),
shared_job_queue_type output = std::make_shared<write_job_queue_type>() shared_job_queue_type output = std::make_shared<write_job_queue_type>()
); );
......
...@@ -211,6 +211,7 @@ void abstract_broker::add_datagram_servant(datagram_servant_ptr ptr) { ...@@ -211,6 +211,7 @@ void abstract_broker::add_datagram_servant(datagram_servant_ptr ptr) {
ptr->set_parent(this); ptr->set_parent(this);
auto hdls = ptr->hdls(); auto hdls = ptr->hdls();
launch_servant(ptr); launch_servant(ptr);
add_hdl_for_datagram_servant(ptr, ptr->hdl());
for (auto& hdl : hdls) for (auto& hdl : hdls)
add_hdl_for_datagram_servant(ptr, hdl); add_hdl_for_datagram_servant(ptr, hdl);
} }
...@@ -275,6 +276,7 @@ void abstract_broker::move_datagram_servant(datagram_servant_ptr ptr) { ...@@ -275,6 +276,7 @@ void abstract_broker::move_datagram_servant(datagram_servant_ptr ptr) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_ASSERT(ptr->parent() != nullptr && ptr->parent() != this); CAF_ASSERT(ptr->parent() != nullptr && ptr->parent() != this);
ptr->set_parent(this); ptr->set_parent(this);
ptr->add_to_loop();
CAF_ASSERT(ptr->parent() == this); CAF_ASSERT(ptr->parent() == this);
auto hdls = ptr->hdls(); auto hdls = ptr->hdls();
for (auto& hdl : hdls) for (auto& hdl : hdls)
......
This diff is collapsed.
...@@ -34,9 +34,11 @@ auto autoconnect_timeout = std::chrono::minutes(10); ...@@ -34,9 +34,11 @@ 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, behavior datagram_connection_broker(broker* self,
uint16_t port,
network::address_listing addresses, network::address_listing addresses,
actor system_broker) { actor system_broker,
basp::instance* instance) {
auto& mx = self->system().middleman().backend(); auto& mx = self->system().middleman().backend();
auto& this_node = self->system().node(); auto& this_node = self->system().node();
auto app_id = get_or(self->config(), "middleman.app-identifier", auto app_id = get_or(self->config(), "middleman.app-identifier",
...@@ -47,81 +49,96 @@ behavior datagram_connection_broker(broker* self, uint16_t port, ...@@ -47,81 +49,96 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
if (eptr) { if (eptr) {
auto hdl = (*eptr)->hdl(); auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr)); self->add_datagram_servant(std::move(*eptr));
basp::instance::write_client_handshake(self->context(), std::vector<char> buf;
self->wr_buf(hdl), instance->write_client_handshake(self->context(), buf,
none, this_node, none, this_node, app_id);
app_id); self->enqueue_datagram(hdl, std::move(buf));
self->flush(hdl);
} }
} }
} }
// We are not interested in attempts that do not work.
self->set_default_handler(drop);
return { return {
[=](new_datagram_msg& msg) { [=](new_datagram_msg& msg) {
auto hdl = msg.handle; auto hdl = msg.handle;
auto cap = msg.buf.capacity();
self->send(system_broker, std::move(msg), self->take(hdl), port); self->send(system_broker, std::move(msg), self->take(hdl), port);
self->quit(); self->quit();
// The buffer returned to the multiplexer will have capacity zero
// if we don't do this. Handling this case here is cheaper than
// checking before each receive in the multiplexer.
msg.buf.reserve(cap);
}, },
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 call it a day, then.
CAF_LOG_INFO("aborted direct connection attempt after 10min"); CAF_LOG_INFO("aborted direct connection attempt after 10 min");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
} }
}; };
} }
bool establish_stream_connection(stateful_actor<connection_helper_state>* self,
const actor& system_broker,
uint16_t port,
network::address_listing& addresses) {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// Gotcha! Send scribe to our BASP broker to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(system_broker, connect_atom::value, *hdl, port);
return true;
}
}
}
return false;
}
behavior connection_helper(stateful_actor<connection_helper_state>* self, behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b) { actor system_broker, basp::instance* i) {
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 config:" << CAF_ARG(msg));
// whatever happens, we are done afterwards if (item.empty() || msg.empty()) {
self->quit(); CAF_LOG_DEBUG("skipping empty info");
msg.apply({
[&](uint16_t port, network::address_listing& addresses) {
if (item == "basp.default-connectivity-tcp") {
auto& mx = self->system().middleman().backend();
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto hdl = mx.new_tcp_scribe(addr, port);
if (hdl) {
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
CAF_LOG_INFO("connected directly:" << CAF_ARG(addr));
self->send(b, connect_atom::value, *hdl, port);
return; return;
} }
// Whatever happens, we are done afterwards.
self->quit();
msg.apply({
[&](basp::routing_table::address_map& addrs) {
if (addrs.count(network::protocol::tcp) > 0) {
auto eps = addrs[network::protocol::tcp];
if (!establish_stream_connection(self, system_broker, eps.first,
eps.second))
CAF_LOG_ERROR("could not connect to node ");
} }
} if (addrs.count(network::protocol::udp) > 0) {
CAF_LOG_INFO("could not connect to node directly"); auto eps = addrs[network::protocol::udp];
} else if (item == "basp.default-connectivity-udp") { // Create new broker to try addresses for communication via UDP.
auto& sys = self->system(); auto b = self->system().middleman().spawn_broker(
// create new broker to try addresses for communication via UDP datagram_connection_broker, eps.first, std::move(eps.second),
if (get_or(sys.config(), "middleman.attach-utility-actors", false)) system_broker, i
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));
} }
} }
}); });
}, },
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 direct connection attempt after 10min");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
} }
......
...@@ -733,7 +733,7 @@ default_multiplexer::new_local_udp_endpoint(uint16_t port, const char* in, ...@@ -733,7 +733,7 @@ default_multiplexer::new_local_udp_endpoint(uint16_t port, const char* in,
} }
int64_t default_multiplexer::next_endpoint_id() { int64_t default_multiplexer::next_endpoint_id() {
return servant_ids_++; return servant_ids_.fetch_add(1);
} }
void default_multiplexer::handle_internal_events() { void default_multiplexer::handle_internal_events() {
......
...@@ -121,6 +121,13 @@ bool heartbeat_valid(const header& hdr) { ...@@ -121,6 +121,13 @@ bool heartbeat_valid(const header& hdr) {
&& zero(hdr.operation_data); && zero(hdr.operation_data);
} }
bool acknowledge_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor);
}
} // namespace <anonymous> } // namespace <anonymous>
bool valid(const header& hdr) { bool valid(const header& hdr) {
...@@ -139,6 +146,8 @@ bool valid(const header& hdr) { ...@@ -139,6 +146,8 @@ bool valid(const header& hdr) {
return kill_proxy_instance_valid(hdr); return kill_proxy_instance_valid(hdr);
case message_type::heartbeat: case message_type::heartbeat:
return heartbeat_valid(hdr); return heartbeat_valid(hdr);
case message_type::acknowledge_handshake:
return acknowledge_handshake_valid(hdr);
} }
} }
......
This diff is collapsed.
...@@ -32,7 +32,8 @@ const char* message_type_strings[] = { ...@@ -32,7 +32,8 @@ const char* message_type_strings[] = {
"dispatch_message", "dispatch_message",
"announce_proxy_instance", "announce_proxy_instance",
"kill_proxy_instance", "kill_proxy_instance",
"heartbeat" "heartbeat",
"acknowledge_handshake"
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -420,6 +420,8 @@ void middleman::init(actor_system_config& cfg) { ...@@ -420,6 +420,8 @@ void middleman::init(actor_system_config& cfg) {
cfg.add_message_type<network::protocol>("@protocol") cfg.add_message_type<network::protocol>("@protocol")
.add_message_type<network::address_listing>("@address_listing") .add_message_type<network::address_listing>("@address_listing")
.add_message_type<network::receive_buffer>("@receive_buffer") .add_message_type<network::receive_buffer>("@receive_buffer")
.add_message_type<basp::routing_table::address_endpoint>("@address_endpoint")
.add_message_type<basp::routing_table::address_map>("@address_map")
.add_message_type<new_data_msg>("@new_data_msg") .add_message_type<new_data_msg>("@new_data_msg")
.add_message_type<new_connection_msg>("@new_connection_msg") .add_message_type<new_connection_msg>("@new_connection_msg")
.add_message_type<acceptor_closed_msg>("@acceptor_closed_msg") .add_message_type<acceptor_closed_msg>("@acceptor_closed_msg")
......
...@@ -33,123 +33,115 @@ routing_table::~routing_table() { ...@@ -33,123 +33,115 @@ routing_table::~routing_table() {
// nop // nop
} }
optional<routing_table::route> routing_table::lookup(const node_id& target) { node_id routing_table::lookup(const endpoint_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 endpoint_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<routing_table::endpoint_handle> void routing_table::erase(const endpoint_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 endpoint_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, false};
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, false};
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 endpoint_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, false};
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; return false;
if (parent_->parent().has_hook()) }
for (auto& nid : i->second)
parent_->parent().notify<hook::route_lost>(nid, dest); bool routing_table::origin(const node_id& nid,
indirect_.erase(i); const node_id& origin) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.origin = origin;
return true; return true;
} }
void routing_table::add_direct(const endpoint_handle& hdl, optional<node_id>
const node_id& nid) { routing_table::origin(const node_id& nid) {
CAF_ASSERT(direct_by_hdl_.count(hdl) == 0); auto i = node_information_base_.find(nid);
CAF_ASSERT(direct_by_nid_.count(nid) == 0); if (i == node_information_base_.end())
direct_by_hdl_.emplace(hdl, nid); return none;
direct_by_nid_.emplace(nid, hdl); return i->second.origin;
parent_->parent().notify<hook::new_connection_established>(nid);
} }
bool routing_table::add_indirect(const node_id& hop, const node_id& dest) { bool routing_table::handle(const node_id& nid,
auto i = blacklist_.find(dest); const routing_table::endpoint_handle& hdl) {
if (i == blacklist_.end() || i->second.count(hop) == 0) { auto i = node_information_base_.find(nid);
auto& hops = indirect_[dest]; if (i == node_information_base_.end())
auto added_first = hops.empty(); return false;
hops.emplace(hop); i->second.hdl = hdl;
parent_->parent().notify<hook::new_route_added>(hop, dest); nid_by_hdl_.emplace(hdl, nid);
return added_first; return true;
}
return false; // blacklisted
} }
bool routing_table::reachable(const node_id& dest) { optional<routing_table::endpoint_handle>
return direct_by_nid_.count(dest) > 0 || indirect_.count(dest) > 0; routing_table::handle(const node_id& nid) {
} auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
size_t routing_table::erase(const node_id& dest, erase_callback& cb) { return none;
cb(dest); return i->second.hdl;
size_t res = 0; }
auto i = indirect_.find(dest);
if (i != indirect_.end()) { void routing_table::local_addresses(network::protocol::transport key,
res = i->second.size(); routing_table::address_endpoint addrs) {
for (auto& nid : i->second) { local_addrs_[key] = addrs;
cb(nid); }
parent_->parent().notify<hook::route_lost>(nid, dest);
} const routing_table::address_map& routing_table::local_addresses() {
indirect_.erase(i); return local_addrs_;
} }
auto hdl = lookup_direct(dest);
if (hdl) { bool routing_table::received_client_handshake(const node_id& nid, bool flag) {
direct_by_hdl_.erase(*hdl); auto i = node_information_base_.find(nid);
direct_by_nid_.erase(dest); if (i == node_information_base_.end())
parent_->parent().notify<hook::connection_lost>(dest); return false;
++res; i->second.received_client_handshake = flag;
} return true;
return res; }
bool routing_table::received_client_handshake(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
return i->second.received_client_handshake;
} }
} // namespace basp } // namespace basp
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -39,6 +39,8 @@ class config : public actor_system_config { ...@@ -39,6 +39,8 @@ class config : public actor_system_config {
public: public:
config() { config() {
load<io::middleman>(); load<io::middleman>();
set("middleman.enable-tcp", true);
set("middleman.enable-udp", false);
add_message_type<std::vector<int>>("std::vector<int>"); add_message_type<std::vector<int>>("std::vector<int>");
if (auto err = parse(test::engine::argc(), test::engine::argv())) if (auto err = parse(test::engine::argc(), test::engine::argv()))
CAF_FAIL("failed to parse config: " << to_string(err)); CAF_FAIL("failed to parse config: " << to_string(err));
...@@ -65,7 +67,11 @@ struct fixture { ...@@ -65,7 +67,11 @@ struct fixture {
} }
}; };
behavior make_pong_behavior() { behavior make_pong_behavior(event_based_actor* self) {
self->set_exit_handler([=](exit_msg& m) {
CAF_MESSAGE("Pong received exit message.");
self->quit(m.reason);
});
return { return {
[](int val) -> int { [](int val) -> int {
++val; ++val;
......
...@@ -39,6 +39,7 @@ class config : public actor_system_config { ...@@ -39,6 +39,7 @@ class config : public actor_system_config {
public: public:
config() { config() {
load<io::middleman>(); load<io::middleman>();
set("middleman.enable-tcp", false);
set("middleman.enable-udp", true); set("middleman.enable-udp", true);
add_message_type<std::vector<int>>("std::vector<int>"); add_message_type<std::vector<int>>("std::vector<int>");
if (auto err = parse(test::engine::argc(), test::engine::argv())) if (auto err = parse(test::engine::argc(), test::engine::argv()))
...@@ -66,7 +67,11 @@ struct fixture { ...@@ -66,7 +67,11 @@ struct fixture {
} }
}; };
behavior make_pong_behavior() { behavior make_pong_behavior(event_based_actor* self) {
self->set_exit_handler([=](exit_msg& m) {
CAF_MESSAGE("Pong received exit message.");
self->quit(m.reason);
});
return { return {
[](int val) -> int { [](int val) -> int {
++val; ++val;
...@@ -204,8 +209,7 @@ CAF_TEST_DISABLED(multiple_endpoints_udp) { ...@@ -204,8 +209,7 @@ CAF_TEST_DISABLED(multiple_endpoints_udp) {
} }
}; };
}); });
auto port = server_sys.middleman().publish_udp(mirror, 0); auto port = unbox(server_sys.middleman().publish_udp(mirror, 0));
CAF_REQUIRE(port);
CAF_MESSAGE("server running on port " << port); CAF_MESSAGE("server running on port " << port);
auto client_fun = [](event_based_actor* self) -> behavior { auto client_fun = [](event_based_actor* self) -> behavior {
return { return {
...@@ -225,7 +229,8 @@ CAF_TEST_DISABLED(multiple_endpoints_udp) { ...@@ -225,7 +229,8 @@ CAF_TEST_DISABLED(multiple_endpoints_udp) {
actor_system client_sys{client_cfg}; actor_system client_sys{client_cfg};
auto client = client_sys.spawn(client_fun); auto client = client_sys.spawn(client_fun);
// Acquire remote actor from the server. // Acquire remote actor from the server.
auto client_srv = client_sys.middleman().remote_actor_udp("localhost", *port); // acquire remote actor from the server
auto client_srv = client_sys.middleman().remote_actor_udp("localhost", port);
CAF_REQUIRE(client_srv); CAF_REQUIRE(client_srv);
// Setup other clients. // Setup other clients.
for (int i = 0; i < 5; ++i) { for (int i = 0; i < 5; ++i) {
...@@ -234,7 +239,7 @@ CAF_TEST_DISABLED(multiple_endpoints_udp) { ...@@ -234,7 +239,7 @@ CAF_TEST_DISABLED(multiple_endpoints_udp) {
CAF_MESSAGE("creating new client"); CAF_MESSAGE("creating new client");
auto other = other_sys.spawn(client_fun); auto other = other_sys.spawn(client_fun);
// Acquire remote actor from the new server. // Acquire remote actor from the new server.
auto other_srv = other_sys.middleman().remote_actor_udp("localhost", *port); auto other_srv = other_sys.middleman().remote_actor_udp("localhost", port);
CAF_REQUIRE(other_srv); CAF_REQUIRE(other_srv);
// Establish communication and exit. // Establish communication and exit.
CAF_MESSAGE("client contacts server and exits"); CAF_MESSAGE("client contacts server and exits");
......
...@@ -82,6 +82,17 @@ public: ...@@ -82,6 +82,17 @@ public:
return *res; return *res;
} }
/// Convenience function for calling `mm.publish` and requiring a valid
/// result.
template <class Handle>
uint16_t publish_udp(Handle whom, uint16_t port, const char* in = nullptr,
bool reuse = false) {
this->sched.inline_next_enqueue();
auto res = mm.publish_udp(whom, port, in, reuse);
CAF_REQUIRE(res);
return *res;
}
/// Convenience function for calling `mm.remote_actor` and requiring a valid /// Convenience function for calling `mm.remote_actor` and requiring a valid
/// result. /// result.
template <class Handle = caf::actor> template <class Handle = caf::actor>
...@@ -93,6 +104,17 @@ public: ...@@ -93,6 +104,17 @@ public:
return *res; return *res;
} }
/// Convenience function for calling `mm.remote_actor` and requiring a valid
/// result.
template <class Handle = caf::actor>
Handle remote_actor_udp(std::string host, uint16_t port) {
this->sched.inline_next_enqueue();
this->sched.after_next_enqueue(run_all_nodes);
auto res = mm.remote_actor_udp<Handle>(std::move(host), port);
CAF_REQUIRE(res);
return *res;
}
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Reference to the node's middleman. /// Reference to the node's middleman.
...@@ -152,6 +174,8 @@ public: ...@@ -152,6 +174,8 @@ public:
using accept_handle = caf::io::accept_handle; using accept_handle = caf::io::accept_handle;
using datagram_handle = caf::io::datagram_handle;
test_network_fixture_base(planets_vector xs) : planets_(std::move(xs)) { test_network_fixture_base(planets_vector xs) : planets_(std::move(xs)) {
// nop // nop
} }
...@@ -166,6 +190,11 @@ public: ...@@ -166,6 +190,11 @@ public:
return connection_handle::from_int(++hdl_id_); return connection_handle::from_int(++hdl_id_);
} }
/// Returns a unique datagram handle.
datagram_handle next_datagram_handle() {
return datagram_handle::from_int(++hdl_id_);
}
/// Prepare a connection from `client` (calls `remote_actor`) to `server` /// Prepare a connection from `client` (calls `remote_actor`) to `server`
/// (calls `publish`). /// (calls `publish`).
/// @returns randomly picked connection handles for the server and the client. /// @returns randomly picked connection handles for the server and the client.
...@@ -190,6 +219,31 @@ public: ...@@ -190,6 +219,31 @@ public:
next_accept_handle()); next_accept_handle());
} }
/// Prepares comminication between `client` (calls `remote_actor_udp`) and
/// `server` (calls `publish_udp`).
/// @returns randomly picked datagram handles for the server and the client.
std::pair<datagram_handle, datagram_handle>
prepare_endpoints(PlanetType& server, PlanetType& client,
std::string host, uint16_t port,
datagram_handle server_endpoint) {
auto server_hdl = next_datagram_handle();
auto client_hdl = next_datagram_handle();
server.mpx.prepare_endpoints(server_endpoint, server_hdl, client.mpx,
std::move(host), port, client_hdl);
return std::make_pair(server_hdl, client_hdl);
}
/// Prepares comminication between `client` (calls `remote_actor_udp`) and
/// `server` (calls `publish_udp`).
/// @returns randomly picked datagram handles for the server and the client.
std::pair<datagram_handle, datagram_handle>
prepare_endpoints(PlanetType& server, PlanetType& client,
std::string host, uint16_t port) {
return prepare_endpoints(server, client, std::move(host), port,
next_datagram_handle());
}
// Convenience function for transmitting all "network" traffic (no new // Convenience function for transmitting all "network" traffic (no new
// connections are accepted). // connections are accepted).
void network_traffic() { void network_traffic() {
...@@ -245,10 +299,15 @@ public: ...@@ -245,10 +299,15 @@ public:
} }
}; };
struct mm_enabled_config : caf::actor_system_config {
mm_enabled_config() {
load<caf::io::middleman>();
}
};
/// A simple fixture that includes three nodes (`earth`, `mars`, and `jupiter`) /// A simple fixture that includes three nodes (`earth`, `mars`, and `jupiter`)
/// that can connect to each other. /// that can connect to each other.
template <class BaseFixture = template <class BaseFixture = test_coordinator_fixture<mm_enabled_config>>
test_coordinator_fixture<caf::actor_system_config>>
class belt_fixture class belt_fixture
: public test_network_fixture_base<test_node_fixture<BaseFixture>> { : public test_network_fixture_base<test_node_fixture<BaseFixture>> {
public: public:
......
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