Commit 86df0d19 authored by Dominik Charousset's avatar Dominik Charousset

Implement automatic connections in BASP

parent 09cd843a
No preview for this file type
......@@ -78,7 +78,8 @@ set (LIBCAF_CORE_SRCS
src/sync_request_bouncer.cpp
src/try_match.cpp
src/uniform_type_info.cpp
src/uniform_type_info_map.cpp)
src/uniform_type_info_map.cpp
src/whereis.cpp)
add_custom_target(libcaf_core)
......
......@@ -156,6 +156,8 @@ protected:
public:
/// @cond PRIVATE
static actor_id latest_actor_id();
enum linking_operation {
establish_link_op,
establish_backlink_op,
......
......@@ -69,6 +69,9 @@ using get_atom = atom_constant<atom("GET")>;
/// Generic 'PUT' atom for request operations.
using put_atom = atom_constant<atom("PUT")>;
/// Generic 'UPDATE' atom, e.g., or signalizing updates in a key-value store.
using update_atom = atom_constant<atom("UPDATE")>;
/// Generic 'DELETE' atom for request operations.
using delete_atom = atom_constant<atom("DELETE")>;
......@@ -111,6 +114,12 @@ using publish_atom = atom_constant<atom("PUBLISH")>;
/// Generic 'UNPUBLISH' atom, e.g., for removing an actor/port mapping.
using unpublish_atom = atom_constant<atom("UNPUBLISH")>;
/// Generic 'PUBLISH' atom, e.g., for publishing actors at a given port.
using subscribe_atom = atom_constant<atom("SUBSCRIBE")>;
/// Generic 'UNPUBLISH' atom, e.g., for removing an actor/port mapping.
using unsubscribe_atom = atom_constant<atom("UNSUBSCRIB")>;
/// Generic 'CONNECT' atom, e.g., for connecting to remote CAF instances.
using connect_atom = atom_constant<atom("CONNECT")>;
......
......@@ -83,6 +83,8 @@ public:
actor get_named(atom_value key) const;
void put_named(atom_value key, actor value);
using named_entries = std::unordered_map<atom_value, actor>;
named_entries named_actors() const;
......@@ -109,6 +111,7 @@ private:
entries entries_;
named_entries named_entries_;
mutable detail::shared_spinlock named_entries_mtx_;
};
} // namespace detail
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_EXPERIMENTAL_WHEREIS_HPP
#define CAF_EXPERIMENTAL_WHEREIS_HPP
#include "caf/fwd.hpp"
#include "caf/atom.hpp"
namespace caf {
namespace experimental {
actor whereis(atom_value registered_name);
} // namespace experimental
} // namespace caf
#endif // CAF_EXPERIMENTAL_WHEREIS_HPP
......@@ -47,6 +47,10 @@ std::atomic<actor_id> ids_;
} // namespace <anonymous>
actor_id abstract_actor::latest_actor_id() {
return ids_.load();
}
// exit_reason_ is guaranteed to be set to 0, i.e., exit_reason::not_exited,
// by std::atomic<> constructor
......
......@@ -22,6 +22,8 @@
#include <mutex>
#include <limits>
#include <stdexcept>
#include <unordered_map>
#include <unordered_set>
#include "caf/spawn.hpp"
#include "caf/locks.hpp"
......@@ -29,7 +31,9 @@
#include "caf/attachable.hpp"
#include "caf/exit_reason.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/experimental/stateful_actor.hpp"
#include "caf/experimental/announce_actor_type.hpp"
#include "caf/scheduler/detached_threads.hpp"
......@@ -131,13 +135,24 @@ void actor_registry::await_running_count_equal(size_t expected) {
}
actor actor_registry::get_named(atom_value key) const {
shared_guard guard{named_entries_mtx_};
auto i = named_entries_.find(key);
if (i == named_entries_.end())
return invalid_actor;
return i->second;
}
void actor_registry::put_named(atom_value key, actor value) {
if (value)
value->attach_functor([=](uint32_t) {
detail::singletons::get_actor_registry()->put_named(key, invalid_actor);
});
exclusive_guard guard{named_entries_mtx_};
named_entries_.emplace(key, std::move(value));
}
auto actor_registry::named_actors() const -> named_entries {
shared_guard guard{named_entries_mtx_};
return named_entries_;
}
......@@ -164,8 +179,72 @@ void actor_registry::stop() {
}
void actor_registry::initialize() {
named_entries_.emplace(atom("spawner"),
experimental::spawn_announce_actor_type_server());
using namespace experimental;
struct kvstate {
using key_type = std::string;
using mapped_type = message;
using subscriber_set = std::unordered_set<actor>;
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<actor,topic_set> subscribers;
};
auto kvstore = [](stateful_actor<kvstate>* self) -> behavior {
return {
[=](put_atom, const std::string& key, message& msg) {
auto& vp = self->state.data[key];
if (vp.first == msg)
return;
vp.first = std::move(msg);
for (auto& subscriber : vp.second)
if (subscriber != self->current_sender())
self->send(subscriber, update_atom::value, key, vp.second);
},
[=](get_atom, std::string& key) -> message {
auto i = self->state.data.find(key);
return make_message(ok_atom::value, std::move(key),
i != self->state.data.end() ? i->second.first
: make_message());
},
[=](subscribe_atom, const std::string& key) {
auto subscriber = actor_cast<actor>(self->current_sender());
if (! subscriber)
return;
self->state.data[key].second.insert(subscriber);
auto& subscribers = self->state.subscribers;
auto i = subscribers.find(subscriber);
if (i != subscribers.end()) {
i->second.insert(key);
} else {
self->monitor(subscriber);
subscribers.emplace(subscriber, kvstate::topic_set{key});
}
},
[=](unsubscribe_atom, const std::string& key) {
auto subscriber = actor_cast<actor>(self->current_sender());
if (! subscriber)
return;
self->state.subscribers[subscriber].erase(key);
self->state.data[key].second.erase(subscriber);
},
[=](const down_msg& dm) {
auto subscriber = actor_cast<actor>(dm.source);
if (! subscriber)
return;
auto& subscribers = self->state.subscribers;
auto i = subscribers.find(subscriber);
if (i == subscribers.end())
return;
for (auto& key : i->second)
self->state.data[key].second.erase(subscriber);
subscribers.erase(i);
},
others >> [] {
return make_message(error_atom::value, "unsupported operation");
}
};
};
named_entries_.emplace(atom("SpawnServ"), spawn_announce_actor_type_server());
named_entries_.emplace(atom("ConfigServ"), spawn<hidden>(kvstore));
}
} // namespace detail
......
......@@ -73,7 +73,7 @@ actor spawn_announce_actor_type_server() {
void announce_actor_type_impl(std::string&& name, spawn_fun f) {
auto registry = detail::singletons::get_actor_registry();
auto server = registry->get_named(atom("spawner"));
auto server = registry->get_named(atom("SpawnServ"));
anon_send(server, add_atom::value, std::move(name), std::move(f));
}
......
......@@ -101,8 +101,9 @@ const char* message::uniform_name_at(size_t pos) const {
}
bool message::equals(const message& other) const {
CAF_ASSERT(vals_);
return vals_->equals(*other.vals());
if (empty())
return other.empty();
return other.empty() ? false : vals_->equals(*other.vals());
}
message message::drop(size_t n) const {
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/experimental/whereis.hpp"
#include "caf/actor.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp"
namespace caf {
namespace experimental {
actor whereis(atom_value registered_name) {
return detail::singletons::get_actor_registry()->get_named(registered_name);
}
} // namespace experimental
} // namespace caf
......@@ -41,7 +41,7 @@ struct fixture {
fixture() {
auto registry = detail::singletons::get_actor_registry();
spawner = registry->get_named(atom("spawner"));
spawner = registry->get_named(atom("SpawnServ"));
}
void set_aut(message args, bool expect_fail = false) {
......
......@@ -203,7 +203,13 @@ enum class message_type : uint32_t {
/// that has been terminated.
///
/// ![](kill_proxy_instance.png)
kill_proxy_instance = 0x04
kill_proxy_instance = 0x04,
/// Looks up a particular named actor in the registry
/// of the receiving node, which replies with a `dispatch_message`.
///
/// ![](name_lookup.png)
name_lookup = 0x05
};
/// @relates message_type
......@@ -308,6 +314,11 @@ public:
/// `invalid_connection_handle` if no direct connection to `nid` exists.
connection_handle lookup_direct(const node_id& nid) const;
/// Returns the next hop that would be chosen for `nid`
/// or `invalid_node_id` if there's no indirect route to `nid`.
node_id lookup_indirect(const node_id& nid) const;
/// Flush output buffer for `r`.
void flush(const route& r);
......@@ -316,7 +327,7 @@ public:
void add_direct(const connection_handle& hdl, const node_id& dest);
/// Adds a new indirect route to the table.
void add_indirect(const node_id& hop, const node_id& dest);
bool add_indirect(const node_id& hop, const node_id& dest);
/// Blacklist the route to `dest` via `hop`.
void blacklist(const node_id& hop, const node_id& dest);
......@@ -326,6 +337,9 @@ public:
/// including the node that is assigned as direct path for `hdl`.
void erase_direct(const connection_handle& hdl, erase_callback& cb);
/// Removes any entry for indirect connection to `dest`.
void erase_indirect(const node_id& dest);
/// Queries whether `dest` is reachable.
bool reachable(const node_id& dest);
......@@ -379,8 +393,12 @@ public:
/// routing table from this callback.
virtual void purge_state(const node_id& nid) = 0;
/// Called whenever a remote node created a proxy
/// for one of our local actors.
virtual void proxy_announced(const node_id& nid, actor_id aid) = 0;
/// Called whenever a remote actor died to destroy
/// the proxy instance on our end.
virtual void kill_proxy(const node_id& nid, actor_id aid, uint32_t rsn) = 0;
/// Called whenever a `dispatch_message` arrived for a local actor.
......@@ -388,6 +406,10 @@ public:
const node_id& dest_node, actor_id dest_actor,
message& msg, message_id mid) = 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;
/// Returns the actor namespace associated to this BASP protocol instance.
actor_namespace& get_namespace() {
return namespace_;
......@@ -508,6 +530,12 @@ public:
actor_id aid,
uint32_t rsn);
/// Writes a name lookup request to `buf`.
void write_name_lookup(buffer_type& buf,
atom_value requested_name,
const node_id& dest_node,
actor_id source_actor);
const node_id& this_node() const {
return this_node_;
}
......
......@@ -64,6 +64,8 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
const node_id& dest_node, actor_id dest_actor,
message& msg, message_id mid) override;
void learned_new_node_indirectly(const node_id& nid) override;
struct connection_context {
basp::connection_state cstate;
basp::header hdr;
......@@ -93,6 +95,11 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
// keeps the associated proxies alive to work around subtle bugs
std::unordered_map<node_id, std::pair<uint16_t, actor_addr>> known_remotes;
// can be enabled by the user to let CAF automatically try
// to establish new connections at runtime to optimize
// routing paths by forming a mesh between all nodes
bool enable_automatic_connections = false;
const node_id& this_node() const {
return instance.this_node();
}
......
......@@ -33,10 +33,11 @@ namespace caf {
namespace io {
namespace network {
// {protocol => address}
using address_listing = std::map<protocol, std::vector<std::string>>;
// {interface_name => {protocol => address}}
using interfaces_map = std::map<std::string,
std::map<protocol,
std::vector<std::string>>>;
using interfaces_map = std::map<std::string, address_listing>;
/// Utility class bundling access to network interface names and addresses.
class interfaces {
......@@ -45,8 +46,7 @@ public:
static interfaces_map list_all(bool include_localhost = true);
/// Returns all addresses for all devices for all protocols.
static std::map<protocol, std::vector<std::string>>
list_addresses(bool include_localhost = true);
static address_listing list_addresses(bool include_localhost = true);
/// Returns all addresses for all devices for given protocol.
static std::vector<std::string> list_addresses(protocol proc,
......
......@@ -39,16 +39,18 @@ namespace basp {
std::string to_string(message_type x) {
switch (x) {
case basp::message_type::server_handshake:
case message_type::server_handshake:
return "server_handshake";
case basp::message_type::client_handshake:
case message_type::client_handshake:
return "client_handshake";
case basp::message_type::dispatch_message:
case message_type::dispatch_message:
return "dispatch_message";
case basp::message_type::announce_proxy_instance:
case message_type::announce_proxy_instance:
return "announce_proxy_instance";
case basp::message_type::kill_proxy_instance:
case message_type::kill_proxy_instance:
return "kill_proxy_instance";
case message_type::name_lookup:
return "name_lookup";
default:
return "???";
}
......@@ -142,6 +144,13 @@ bool kill_proxy_instance_valid(const header& hdr) {
&& ! zero(hdr.operation_data);
}
bool name_lookup_valid(const header& hdr) {
return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
&& ! zero(hdr.source_actor);
}
} // namespace <anonymous>
bool valid(const header& hdr) {
......@@ -158,6 +167,8 @@ bool valid(const header& hdr) {
return announce_proxy_instance_valid(hdr);
case message_type::kill_proxy_instance:
return kill_proxy_instance_valid(hdr);
case message_type::name_lookup:
return name_lookup_valid(hdr);
}
}
......@@ -177,6 +188,19 @@ optional<routing_table::route> routing_table::lookup(const node_id& target) {
auto hdl = lookup_direct(target);
if (hdl != invalid_connection_handle)
return route{parent_->wr_buf(hdl), 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();
auto hdl = lookup_direct(hop);
if (hdl != invalid_connection_handle)
return route{parent_->wr_buf(hdl), hop, hdl};
else
hops.erase(hops.begin());
}
}
return none;
}
......@@ -194,6 +218,15 @@ routing_table::lookup_direct(const node_id& nid) const {
return get_opt(direct_by_nid_, nid, invalid_connection_handle);
}
node_id routing_table::lookup_indirect(const node_id& nid) const {
auto i = indirect_.find(nid);
if (i == indirect_.end())
return invalid_node_id;
if (i->second.empty())
return invalid_node_id;
return *i->second.begin();
}
void routing_table::blacklist(const node_id& hop, const node_id& dest) {
blacklist_[dest].emplace(hop);
auto i = indirect_.find(dest);
......@@ -214,6 +247,10 @@ void routing_table::erase_direct(const connection_handle& hdl,
direct_by_hdl_.erase(i);
}
void routing_table::erase_indirect(const node_id& dest) {
indirect_.erase(dest);
}
void routing_table::add_direct(const connection_handle& hdl,
const node_id& nid) {
CAF_ASSERT(direct_by_hdl_.count(hdl) == 0);
......@@ -222,10 +259,15 @@ void routing_table::add_direct(const connection_handle& hdl,
direct_by_nid_.emplace(nid, hdl);
}
void routing_table::add_indirect(const node_id& hop, const node_id& dest) {
bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
auto i = blacklist_.find(dest);
if (i == blacklist_.end() || i->second.count(hop) == 0)
indirect_[dest].emplace(hop);
if (i == blacklist_.end() || i->second.count(hop) == 0) {
auto& hops = indirect_[dest];
auto added_first = hops.empty();
hops.emplace(hop);
return added_first;
}
return false; // blacklisted
}
bool routing_table::reachable(const node_id& dest) {
......@@ -363,9 +405,10 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return err();
}
// add entry to routing table and call client
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection: " << to_string(hdr.source_node));
tbl_.add_direct(dm.handle, hdr.source_node);
tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
......@@ -386,8 +429,10 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
<< to_string(hdr.source_node) << ", close connection");
return err();
}
// add direct route to this node
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection: " << to_string(hdr.source_node));
tbl_.add_direct(dm.handle, hdr.source_node);
tbl_.erase_indirect(hdr.source_node);
if (payload_valid()) {
binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()};
......@@ -397,6 +442,15 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
case message_type::dispatch_message: {
if (! payload_valid())
return err();
// in case the sender of this message was received via a third node,
// we assume that that node to offers a route to the original source
auto last_hop = tbl_.lookup_direct(dm.handle);
if (hdr.source_node != invalid_node_id
&& hdr.source_node != this_node_
&& last_hop != hdr.source_node
&& tbl_.lookup_direct(hdr.source_node) == invalid_connection_handle
&& tbl_.add_indirect(last_hop, hdr.source_node))
callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()};
message msg;
......@@ -413,6 +467,25 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
callee_.kill_proxy(hdr.source_node, hdr.source_actor,
static_cast<uint32_t>(hdr.operation_data));
break;
case message_type::name_lookup: {
if (hdr.source_node == this_node())
return err();
auto path = tbl_.lookup(hdr.source_node);
if (! path) {
CAF_LOG_INFO("cannot reply to name lookup: no route to source");
return await_header;
}
auto atm = static_cast<atom_value>(hdr.operation_data);
auto res = detail::singletons::get_actor_registry()->get_named(atm);
auto msg = make_message(ok_atom::value, std::move(res));
auto writer = make_callback([&](serializer& sink) {
msg.serialize(sink);
});
write(path->wr_buf, message_type::dispatch_message, nullptr, 0,
this_node(), hdr.source_node, res.id(), hdr.source_actor, &writer);
flush(*path);
break;
}
default:
CAF_LOG_ERROR("invalid operation");
return err();
......@@ -569,6 +642,7 @@ void instance::write(buffer_type& buf, header& hdr, payload_writer* pw) {
void instance::write_server_handshake(buffer_type& out_buf,
optional<uint16_t> port) {
using namespace detail;
published_actor* pa = nullptr;
if (port) {
auto i = published_actors_.find(*port);
......@@ -580,9 +654,8 @@ void instance::write_server_handshake(buffer_type& out_buf,
sink << pa->first.id() << pa->second;
else
sink << invalid_actor_id << std::set<std::string>{};
auto na = named_actors();
for (auto& kvp : na)
sink << kvp.first << kvp.second;
sink << atom("SpawnServ")
<< singletons::get_actor_registry()->get_named(atom("SpawnServ"));
});
header hdr{message_type::server_handshake, 0, version,
this_node_, invalid_node_id,
......@@ -591,10 +664,10 @@ void instance::write_server_handshake(buffer_type& out_buf,
}
void instance::write_client_handshake(buffer_type& buf) {
using namespace detail;
auto writer = make_callback([&](serializer& sink) {
auto na = named_actors();
for (auto& kvp : na)
sink << kvp.first << kvp.second;
sink << atom("SpawnServ")
<< singletons::get_actor_registry()->get_named(atom("SpawnServ"));
});
write(buf,
message_type::client_handshake, nullptr, 0,
......@@ -631,6 +704,15 @@ void instance::write_kill_proxy_instance(buffer_type& buf,
write(buf, hdr);
}
void instance::write_name_lookup(buffer_type& buf,
atom_value requested_name,
const node_id& dest_node,
actor_id source_actor) {
write(buf, message_type::name_lookup, nullptr,
static_cast<uint64_t>(requested_name), this_node_, dest_node,
source_actor, invalid_actor_id);
}
auto instance::named_actors() -> named_actors_map {
return detail::singletons::get_actor_registry()->named_actors();
}
......
......@@ -21,8 +21,11 @@
#include "caf/exception.hpp"
#include "caf/make_counted.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/forwarding_actor_proxy.hpp"
#include "caf/experimental/whereis.hpp"
#include "caf/detail/singletons.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
......@@ -31,7 +34,9 @@
#include "caf/io/middleman.hpp"
#include "caf/io/unpublish.hpp"
using std::string;
#include "caf/io/network/interfaces.hpp"
using namespace caf::experimental;
namespace caf {
namespace io {
......@@ -56,8 +61,10 @@ actor_proxy_ptr basp_broker_state::make_proxy(const node_id& nid,
// this member function is being called whenever we deserialize a
// 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
if (nid != this_context->id)
instance.tbl().add_indirect(this_context->id, nid);
if (nid != this_context->id
&& instance.tbl().lookup_direct(nid) == invalid_connection_handle
&& instance.tbl().add_indirect(this_context->id, nid))
learned_new_node_indirectly(nid);
// 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
auto path = instance.tbl().lookup(nid);
......@@ -216,6 +223,79 @@ void basp_broker_state::deliver(const node_id& source_node,
dest->enqueue(src, mid, std::move(msg), nullptr);
}
void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_TSARG(nid));
if (! enable_automatic_connections)
return;
actor bb = self; // a handle for our helper back to this BASP broker
// this member function gets only called once, after adding a new
// indirect connection to the routing table; hence, spawning
// our helper here exactly once and there is no need to track
// in-flight connection requests
auto connection_helper = [=](event_based_actor* helper, actor s) -> behavior {
helper->monitor(s);
return {
// this is the result of the name lookup we initiate via BASP
[=](ok_atom, actor remote_config_server) {
CAF_LOGF_DEBUG("received remote config server: "
<< to_string(remote_config_server));
helper->send(remote_config_server,
get_atom::value, "basp.default-connectivity");
},
// this is the message we are actually waiting for
[=](ok_atom, const std::string&, message& msg) {
CAF_LOGF_DEBUG("received requested config: " << to_string(msg));
// whatever happens, we are done afterwards
helper->quit();
msg.apply({
[&](uint16_t port, network::address_listing& addresses) {
auto& mx = middleman::instance()->backend();
for (auto& kvp : addresses)
for (auto& addr : kvp.second) {
try {
auto hdl = mx.new_tcp_scribe(addr, port);
// gotcha! send scribe to our BASP broker
// to initiate handshake etc.
helper->send(bb, connect_atom::value, hdl, port);
return;
}
catch (...) {
// simply try next address
}
}
CAF_LOGF_INFO("could not connect to node directly, nid = "
<< to_string(nid));
}
});
},
[=](const down_msg& dm) {
helper->quit(dm.reason);
},
after(std::chrono::minutes(10)) >> [=] {
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOGF_INFO("aborted direct connection attempt after 10min, nid = "
<< to_string(nid));
helper->quit(exit_reason::user_shutdown);
}
};
};
auto path = instance.tbl().lookup(nid);
if (! path) {
CAF_LOG_ERROR("learned_new_node_indirectly called, but no route to nid");
return;
}
if (path->next_hop == nid) {
CAF_LOG_ERROR("learned_new_node_indirectly called with direct connection");
return;
}
using namespace detail;
auto tmp = spawn<detached + hidden + lazy_init>(connection_helper, self);
singletons::get_actor_registry()->put(tmp.id(),
actor_cast<abstract_actor_ptr>(tmp));
instance.write_name_lookup(path->wr_buf, atom("ConfigServ"), nid, tmp.id());
}
void basp_broker_state::set_context(connection_handle hdl) {
CAF_LOG_TRACE(CAF_MARG(hdl, id));
auto i = ctx.find(hdl);
......@@ -255,11 +335,14 @@ bool basp_broker_state::erase_context(connection_handle hdl) {
******************************************************************************/
basp_broker::basp_broker(middleman& mm)
: caf::experimental::stateful_actor<basp_broker_state, broker>(mm) {
: stateful_actor<basp_broker_state, broker>(mm) {
// nop
}
behavior basp_broker::make_behavior() {
// ask the configuration server whether we should open a default port
auto config_server = whereis(atom("ConfigServ"));
send(config_server, get_atom::value, "global.enable-automatic-connections");
return {
// received from underlying broker implementation
[=](new_data_msg& msg) {
......@@ -413,7 +496,7 @@ behavior basp_broker::make_behavior() {
err("no connection to requested node");
return {};
}
auto i = na->find(atom("spawner"));
auto i = na->find(atom("SpawnServ"));
if (i == na->end()) {
err("no spawn server on requested node");
return {};
......@@ -421,6 +504,21 @@ behavior basp_broker::make_behavior() {
delegate(i->second, get_atom::value, std::move(type), std::move(args));
return {};
},
[=](ok_atom, const std::string& key, message& value) {
if (key == "global.enable-automatic-connections") {
value.apply([&](bool enabled) {
if (! enabled)
return;
// open a random port and store a record for others
// how to connect to this port in the configuration server
auto port = add_tcp_doorman(uint16_t{0}).second;
auto addrs = network::interfaces::list_addresses(false);
send(config_server, put_atom::value, "basp.default-connectivity",
make_message(port, std::move(addrs)));
state.enable_automatic_connections = true;
});
}
},
// catch-all error handler
others >>
[=] {
......
......@@ -40,6 +40,8 @@
#include "caf/io/basp_broker.hpp"
#include "caf/io/system_messages.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/detail/logging.hpp"
#include "caf/detail/ripemd_160.hpp"
#include "caf/detail/safe_equal.hpp"
......@@ -58,62 +60,48 @@ namespace io {
namespace {
template <class Subtype, class I>
inline void serialize_impl(const handle<Subtype, I>& hdl, serializer* sink) {
sink->write_value(hdl.id());
void serialize(serializer& sink, std::vector<char>& buf) {
sink << static_cast<uint32_t>(buf.size());
sink.write_raw(buf.size(), buf.data());
}
template <class Subtype, class I>
inline void deserialize_impl(handle<Subtype, I>& hdl, deserializer* source) {
hdl.set_id(source->read<int64_t>());
void serialize(deserializer& source, std::vector<char>& buf) {
auto bs = source.read<uint32_t>();
buf.resize(bs);
source.read_raw(buf.size(), buf.data());
}
inline void serialize_impl(const new_connection_msg& msg, serializer* sink) {
serialize_impl(msg.source, sink);
serialize_impl(msg.handle, sink);
template <class Subtype, class I>
void serialize(serializer& sink, handle<Subtype, I>& hdl) {
sink.write_value(hdl.id());
}
inline void deserialize_impl(new_connection_msg& msg, deserializer* source) {
deserialize_impl(msg.source, source);
deserialize_impl(msg.handle, source);
template <class Subtype, class I>
void serialize(deserializer& source, handle<Subtype, I>& hdl) {
hdl.set_id(source.read<int64_t>());
}
inline void serialize_impl(const new_data_msg& msg, serializer* sink) {
serialize_impl(msg.handle, sink);
auto buf_size = static_cast<uint32_t>(msg.buf.size());
if (buf_size != msg.buf.size()) { // narrowing error
std::ostringstream oss;
oss << "attempted to send more than "
<< std::numeric_limits<uint32_t>::max() << " bytes";
auto errstr = oss.str();
CAF_LOGF_INFO(errstr);
throw std::ios_base::failure(std::move(errstr));
}
sink->write_value(buf_size);
sink->write_raw(msg.buf.size(), msg.buf.data());
template <class Archive>
void serialize(Archive& ar, new_connection_msg& msg) {
serialize(ar, msg.source);
serialize(ar, msg.handle);
}
inline void deserialize_impl(new_data_msg& msg, deserializer* source) {
deserialize_impl(msg.handle, source);
auto buf_size = source->read<uint32_t>();
msg.buf.resize(buf_size);
source->read_raw(msg.buf.size(), msg.buf.data());
template <class Archive>
void serialize(Archive& ar, new_data_msg& msg) {
serialize(ar, msg.handle);
serialize(ar, msg.buf);
}
// connection_closed_msg & acceptor_closed_msg have the same fields
template <class T>
typename std::enable_if<std::is_same<T, connection_closed_msg>::value
|| std::is_same<T, acceptor_closed_msg>::value>::type
serialize_impl(const T& dm, serializer* sink) {
serialize_impl(dm.handle, sink);
}
// connection_closed_msg & acceptor_closed_msg have the same fields
template <class T>
typename std::enable_if<std::is_same<T, connection_closed_msg>::value
|| std::is_same<T, acceptor_closed_msg>::value>::type
deserialize_impl(T& dm, deserializer* source) {
deserialize_impl(dm.handle, source);
template <class Archive, class T>
typename std::enable_if<
std::is_same<T, connection_closed_msg>::value
|| std::is_same<T, acceptor_closed_msg>::value
>::type
serialize(Archive& ar, T& dm) {
serialize(ar, dm.handle);
}
template <class T>
......@@ -126,11 +114,11 @@ public:
}
void serialize(const void* instance, serializer* sink) const {
serialize_impl(super::deref(instance), sink);
io::serialize(*sink, super::deref(const_cast<void*>(instance)));
}
void deserialize(void* instance, deserializer* source) const {
deserialize_impl(super::deref(instance), source);
io::serialize(*source, super::deref(instance));
}
};
......@@ -270,6 +258,7 @@ void middleman::initialize() {
}
// announce io-related types
announce<network::protocol>("caf::io::network::protocol");
announce<network::address_listing>("caf::network::address_listing");
do_announce<new_data_msg>("caf::io::new_data_msg");
do_announce<new_connection_msg>("caf::io::new_connection_msg");
do_announce<acceptor_closed_msg>("caf::io::acceptor_closed_msg");
......
......@@ -61,9 +61,11 @@ void scribe::consume(const void*, size_t num_bytes) {
buf.resize(num_bytes);
read_msg().buf.swap(buf);
parent()->invoke_message(invalid_actor_addr, invalid_message_id, read_msg_);
// swap buffer back to stream and implicitly flush wr_buf()
read_msg().buf.swap(buf);
flush();
if (! read_msg_.empty()) {
// swap buffer back to stream and implicitly flush wr_buf()
read_msg().buf.swap(buf);
flush();
}
}
void scribe::io_failure(network::operation op) {
......
......@@ -31,11 +31,15 @@
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/experimental/whereis.hpp"
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp"
using namespace std;
using namespace caf;
using namespace caf::io;
using namespace caf::experimental;
namespace {
......@@ -74,6 +78,8 @@ public:
self_.reset(new scoped_actor);
// run the initialization message of the BASP broker
mpx_->exec_runnable();
// handle the message from the configuration server
mpx()->exec_runnable();
ahdl_ = accept_handle::from_int(1);
mpx_->assign_tcp_doorman(aut_.get(), ahdl_);
registry_ = detail::singletons::get_actor_registry();
......@@ -234,8 +240,8 @@ public:
published_actor_id, invalid_actor_id},
published_actor_id,
published_actor_ifs,
atom("spawner"),
na[atom("spawner")]);
atom("SpawnServ"),
na[atom("SpawnServ")]);
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto path = tbl().lookup(remote_node(i));
......@@ -379,7 +385,7 @@ CAF_TEST(non_empty_server_handshake) {
self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr,
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"},
atom("spawner"), na[atom("spawner")]);
atom("SpawnServ"), na[atom("SpawnServ")]);
CAF_CHECK(hexstr(buf) == hexstr(expected_buf));
}
......@@ -407,8 +413,8 @@ CAF_TEST(client_handshake_and_dispatch) {
},
THROW_ON_UNEXPECTED(self())
);
// check for message forwarded by `forwarding_actor_proxy`
mpx()->exec_runnable(); // exec the message of our forwarding proxy
CAF_TEST_VERBOSE("exec message of forwarding proxy");
mpx()->exec_runnable();
dispatch_out_buf(remote_hdl(0)); // deserialize and send message from out buf
pseudo_remote(0)->receive(
[](int i) {
......@@ -467,7 +473,7 @@ CAF_TEST(remote_actor_and_send) {
{basp::message_type::client_handshake, 0, 0,
this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id},
atom("spawner"), na[atom("spawner")])
atom("SpawnServ"), registry()->get_named(atom("SpawnServ")))
.expect(remote_hdl(0),
{basp::message_type::announce_proxy_instance, 0, 0,
this_node(), remote_node(0),
......@@ -557,4 +563,140 @@ CAF_TEST(actor_serialize_and_deserialize) {
msg);
}
CAF_TEST(indirect_connections) {
// jupiter [remote hdl 0] -> mars [remote hdl 1] -> earth [this_node]
// (this node receives a message from jupiter via mars and responds via mars)
CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
publish(self(), 4242);
mpx()->exec_runnable(); // process publish message in basp_broker
// connect to mars
connect_node(1, ax, self()->id());
// now, an actor from jupiter sends a message to us via mars
mock(remote_hdl(1),
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), this_node(),
pseudo_remote(0)->id(), self()->id()},
make_message("hello from jupiter!"))
.expect(remote_hdl(1),
{basp::message_type::announce_proxy_instance, 0, 0,
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()});
CAF_MESSAGE("receive message from jupiter");
self()->receive(
[](const std::string& str) -> std::string {
CAF_CHECK(str == "hello from jupiter!");
return "hello from earth!";
},
THROW_ON_UNEXPECTED(self())
);
mpx()->exec_runnable(); // process forwarded message in basp_broker
mock()
.expect(remote_hdl(1),
{basp::message_type::dispatch_message, 0, 0,
this_node(), remote_node(0),
self()->id(), pseudo_remote(0)->id()},
make_message("hello from earth!"));
}
CAF_TEST(automatic_connection) {
// this tells our BASP broker to enable the automatic connection feature
anon_send(aut(), ok_atom::value,
"global.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)
mpx()->provide_scribe("jupiter", 8080, remote_hdl(0));
CAF_CHECK(mpx()->pending_scribes().count(make_pair("jupiter", 8080)) == 1);
CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
publish(self(), 4242);
mpx()->exec_runnable(); // process publish message in basp_broker
// connect to mars
connect_node(1, ax, self()->id());
CAF_CHECK_EQUAL(tbl().lookup_direct(remote_node(1)).id(), remote_hdl(1).id());
// now, an actor from jupiter sends a message to us via mars
mock(remote_hdl(1),
{basp::message_type::dispatch_message, 0, 0,
remote_node(0), this_node(),
pseudo_remote(0)->id(), self()->id()},
make_message("hello from jupiter!"))
.expect(remote_hdl(1),
{basp::message_type::name_lookup, 0,
static_cast<uint64_t>(atom("ConfigServ")),
this_node(), remote_node(0),
abstract_actor::latest_actor_id(), // spawned by BASP broker
invalid_actor_id})
.expect(remote_hdl(1),
{basp::message_type::announce_proxy_instance, 0, 0,
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()});
CAF_CHECK_EQUAL(mpx()->output_buffer(remote_hdl(1)).size(), 0);
CAF_CHECK_EQUAL(tbl().lookup_indirect(remote_node(0)), remote_node(1));
CAF_CHECK_EQUAL(tbl().lookup_indirect(remote_node(1)), invalid_node_id);
auto connection_helper = abstract_actor::latest_actor_id();
auto dummy_config_server = [](event_based_actor* this_actor) -> behavior {
return {
[this_actor](get_atom, std::string& key) {
this_actor->quit();
network::address_listing res;
res[network::protocol::ipv4].push_back("jupiter");
return make_message(ok_atom::value, key,
make_message(uint16_t{8080}, std::move(res)));
}
};
};
CAF_CHECK_EQUAL(mpx()->output_buffer(remote_hdl(1)).size(), 0);
// create a dummy config server and respond to the name lookup
CAF_MESSAGE("receive ConfigServ of jupiter");
auto dummy = spawn(dummy_config_server);
registry()->put(dummy->id(), actor_cast<abstract_actor_ptr>(dummy));
mock(remote_hdl(1),
{basp::message_type::dispatch_message, 0, 0,
this_node(), this_node(),
dummy.id(), connection_helper},
make_message(ok_atom::value, dummy));
// our connection helper should now connect to jupiter and
// send the scribe handle over to the BASP broker
mpx()->exec_runnable();
CAF_CHECK_EQUAL(mpx()->output_buffer(remote_hdl(1)).size(), 0);
CAF_CHECK(mpx()->pending_scribes().count(make_pair("jupiter", 8080)) == 0);
// send handshake from jupiter
mock(remote_hdl(0),
{basp::message_type::server_handshake, 0, basp::version,
remote_node(0), invalid_node_id,
pseudo_remote(0)->id(), invalid_actor_id},
pseudo_remote(0)->id(),
uint32_t{0})
.expect(remote_hdl(0),
{basp::message_type::client_handshake, 0, 0,
this_node(), invalid_node_id,
invalid_actor_id, invalid_actor_id},
atom("SpawnServ"), registry()->get_named(atom("SpawnServ")));
CAF_CHECK_EQUAL(tbl().lookup_indirect(remote_node(0)), invalid_node_id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(remote_node(1)), invalid_node_id);
CAF_CHECK_EQUAL(tbl().lookup_direct(remote_node(0)).id(), remote_hdl(0).id());
CAF_CHECK_EQUAL(tbl().lookup_direct(remote_node(1)).id(), remote_hdl(1).id());
CAF_MESSAGE("receive message from jupiter");
self()->receive(
[](const std::string& str) -> std::string {
CAF_CHECK(str == "hello from jupiter!");
return "hello from earth!";
},
THROW_ON_UNEXPECTED(self())
);
mpx()->exec_runnable(); // process forwarded message in basp_broker
CAF_MESSAGE("response message must take direct route now");
mock()
.expect(remote_hdl(0),
{basp::message_type::dispatch_message, 0, 0,
this_node(), remote_node(0),
self()->id(), pseudo_remote(0)->id()},
make_message("hello from earth!"));
CAF_CHECK(mpx()->output_buffer(remote_hdl(1)).size() == 0);
}
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