Commit 681e3f04 authored by Dominik Charousset's avatar Dominik Charousset

Tweak new BASP features to be downward compatible

parent 86df0d19
No preview for this file type
doc/basp_header.png

39.3 KB | W: | H:

doc/basp_header.png

76.8 KB | W: | H:

doc/basp_header.png
doc/basp_header.png
doc/basp_header.png
doc/basp_header.png
  • 2-up
  • Swipe
  • Onion skin
......@@ -51,7 +51,7 @@ template <class Result,
local_actor,
typename std::remove_pointer<FirstArg>::type
>::value>
struct infer_handle_from_fun_ {
struct infer_handle_from_fun_impl {
using type = actor;
using impl = event_based_actor;
using behavior_type = behavior;
......@@ -60,7 +60,7 @@ struct infer_handle_from_fun_ {
// dynamically typed actor returning a behavior
template <class Impl>
struct infer_handle_from_fun_<void, Impl*, true> {
struct infer_handle_from_fun_impl<void, Impl*, true> {
using type = actor;
using impl = Impl;
using behavior_type = behavior;
......@@ -69,7 +69,7 @@ struct infer_handle_from_fun_<void, Impl*, true> {
// dynamically typed actor with self pointer
template <class Impl>
struct infer_handle_from_fun_<behavior, Impl*, true> {
struct infer_handle_from_fun_impl<behavior, Impl*, true> {
using type = actor;
using impl = Impl;
using behavior_type = behavior;
......@@ -78,7 +78,7 @@ struct infer_handle_from_fun_<behavior, Impl*, true> {
// statically typed actor returning a behavior
template <class... Sigs, class FirstArg>
struct infer_handle_from_fun_<typed_behavior<Sigs...>, FirstArg, false> {
struct infer_handle_from_fun_impl<typed_behavior<Sigs...>, FirstArg, false> {
using type = typed_actor<Sigs...>;
using impl = typed_event_based_actor<Sigs...>;
using behavior_type = typed_behavior<Sigs...>;
......@@ -87,7 +87,7 @@ struct infer_handle_from_fun_<typed_behavior<Sigs...>, FirstArg, false> {
// statically typed actor with self pointer
template <class Result, class... Sigs>
struct infer_handle_from_fun_<Result, typed_event_based_actor<Sigs...>*, true> {
struct infer_handle_from_fun_impl<Result, typed_event_based_actor<Sigs...>*, true> {
using type = typed_actor<Sigs...>;
using impl = typed_event_based_actor<Sigs...>;
using behavior_type = typed_behavior<Sigs...>;
......@@ -99,7 +99,7 @@ struct infer_handle_from_fun {
using result_type = typename Trait::result_type;
using arg_types = typename Trait::arg_types;
using first_arg = typename detail::tl_head<arg_types>::type;
using delegate = infer_handle_from_fun_<result_type, first_arg>;
using delegate = infer_handle_from_fun_impl<result_type, first_arg>;
using type = typename delegate::type;
using impl = typename delegate::impl;
using behavior_type = typename delegate::behavior_type;
......
......@@ -222,7 +222,7 @@ struct infer_typed_actor_base<Result, T*, true> {
/// constructor arguments.
template <class C, spawn_options Os = no_spawn_options, class... Ts>
typename actor_handle_from_signature_list<typename C::signatures>::type
spawn_typed(Ts&&... xs) {
CAF_DEPRECATED spawn_typed(Ts&&... xs) {
return spawn_class<C, Os>(nullptr, empty_before_launch_callback{},
std::forward<Ts>(xs)...);
}
......
......@@ -187,6 +187,7 @@ void actor_registry::initialize() {
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;
const char* name = "caf.config_server";
};
auto kvstore = [](stateful_actor<kvstate>* self) -> behavior {
return {
......
......@@ -265,6 +265,23 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
ok_atom::value, self->address()),
self->host());
},
[&](sys_atom, get_atom, std::string& what) {
CAF_LOGF_TRACE(CAF_ARG(what));
if (what == "info") {
CAF_LOGF_DEBUG("reply to 'info' message");
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
ok_atom::value, std::move(what),
self->address(), self->name()),
self->host());
return;
}
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
error_atom::value,
"unknown key: " + std::move(what)),
self->host());
},
others >> [&] {
mismatch = true;
}
......
......@@ -165,7 +165,7 @@ namespace basp {
/// The current BASP version. Different BASP versions will not
/// be able to exchange messages.
constexpr uint64_t version = 2;
constexpr uint64_t version = 1;
/// Storage type for raw bytes.
using buffer_type = std::vector<char>;
......@@ -203,13 +203,7 @@ enum class message_type : uint32_t {
/// that has been terminated.
///
/// ![](kill_proxy_instance.png)
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
kill_proxy_instance = 0x04
};
/// @relates message_type
......@@ -337,8 +331,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);
/// Removes any entry for indirect connection to `dest` and returns
/// `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);
......@@ -406,6 +401,11 @@ 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_directly(const node_id& nid,
bool was_known_indirectly) = 0;
/// Called whenever BASP learns the ID of a remote node
/// to which it does not have a direct connection.
virtual void learned_new_node_indirectly(const node_id& nid) = 0;
......@@ -514,7 +514,7 @@ public:
void write_server_handshake(buffer_type& buf, optional<uint16_t> port);
/// Writes the client handshake to `buf`.
void write_client_handshake(buffer_type& buf);
void write_client_handshake(buffer_type& buf, const node_id& remote_side);
/// Writes a `dispatch_error` to `buf`.
void write_dispatch_error(buffer_type& buf,
......@@ -530,32 +530,15 @@ 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_;
}
using named_actors_map = std::unordered_map<atom_value, actor>;
/// Returns the named actors for this node.
named_actors_map named_actors();
/// Returns the named actors for `nid`.
optional<const named_actors_map&> named_actors(const node_id& nid);
private:
void read_named_actors(binary_deserializer& source, const node_id& nid);
routing_table tbl_;
published_actor_map published_actors_;
node_id this_node_;
callee& callee_;
std::unordered_map<node_id, named_actors_map> named_actors_;
};
/// Checks whether given header contains a handshake.
......
......@@ -60,10 +60,19 @@ struct basp_broker_state : actor_namespace::backend, basp::instance::callee {
// inherited from basp::instance::listener
void kill_proxy(const node_id& nid, actor_id aid, uint32_t rsn) override;
// inherited from basp::instance::listener
void deliver(const node_id& source_node, actor_id source_actor,
const node_id& dest_node, actor_id dest_actor,
message& msg, message_id mid) override;
// performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid);
// inherited from basp::instance::listener
void learned_new_node_directly(const node_id& nid,
bool was_known_indirectly_before) override;
// inherited from basp::instance::listener
void learned_new_node_indirectly(const node_id& nid) override;
struct connection_context {
......@@ -95,6 +104,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;
// stores handles to spawn servers for other nodes; these servers
// are spawned whenever the broker learns a new node ID and try to
// get a 'SpawnServ' instance on the remote side
std::unordered_map<node_id, actor> spawn_servers;
// 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
......
......@@ -49,8 +49,6 @@ std::string to_string(message_type x) {
return "announce_proxy_instance";
case message_type::kill_proxy_instance:
return "kill_proxy_instance";
case message_type::name_lookup:
return "name_lookup";
default:
return "???";
}
......@@ -115,12 +113,21 @@ bool zero(T val) {
bool server_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& ! valid(hdr.dest_node)
&& zero(hdr.dest_actor)
&& ! zero(hdr.operation_data)
&& ! zero(hdr.payload_len);
&& ( (! zero(hdr.source_actor) && ! zero(hdr.payload_len))
|| (zero(hdr.source_actor) && zero(hdr.payload_len)));
}
bool client_handshake_valid(const header& hdr) {
return valid(hdr.source_node);
return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& zero(hdr.operation_data);
}
bool dispatch_message_valid(const header& hdr) {
......@@ -133,7 +140,10 @@ bool announce_proxy_instance_valid(const header& hdr) {
return valid(hdr.source_node)
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
&& ! zero(hdr.dest_actor);
&& zero(hdr.source_actor)
&& ! zero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& zero(hdr.operation_data);
}
bool kill_proxy_instance_valid(const header& hdr) {
......@@ -141,16 +151,11 @@ bool kill_proxy_instance_valid(const header& hdr) {
&& valid(hdr.dest_node)
&& hdr.source_node != hdr.dest_node
&& ! zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& ! 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) {
......@@ -167,8 +172,6 @@ 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);
}
}
......@@ -247,8 +250,8 @@ 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);
bool routing_table::erase_indirect(const node_id& dest) {
return indirect_.erase(dest) > 0;
}
void routing_table::add_direct(const connection_handle& hdl,
......@@ -390,7 +393,7 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
std::set<std::string> sigs;
binary_deserializer bd{payload->data(), payload->size(),
&get_namespace()};
// read first half of payload (ID and interface of published actor)
// read payload (ID and interface of published actor)
bd >> aid >> sigs;
// close self connection after handshake is done
if (hdr.source_node == this_node_) {
......@@ -408,37 +411,32 @@ connection_state instance::handle(const new_data_msg& dm, header& hdr,
// 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);
auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return err();
}
// read remainig payload
read_named_actors(bd, hdr.source_node);
write_client_handshake(path->wr_buf);
// tell client to create proxy etc.
write_client_handshake(path->wr_buf, hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
break;
}
case message_type::client_handshake:
case message_type::client_handshake: {
if (tbl_.lookup_direct(hdr.source_node) != invalid_connection_handle) {
CAF_LOG_INFO("received second client handshake for "
<< to_string(hdr.source_node) << ", close connection");
return err();
<< to_string(hdr.source_node) << " (ignored)");
break;
}
// 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()};
read_named_actors(bd, hdr.source_node);
}
auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
break;
}
case message_type::dispatch_message: {
if (! payload_valid())
return err();
......@@ -467,25 +465,6 @@ 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();
......@@ -652,10 +631,6 @@ void instance::write_server_handshake(buffer_type& out_buf,
auto writer = make_callback([&](serializer& sink) {
if (pa)
sink << pa->first.id() << pa->second;
else
sink << invalid_actor_id << std::set<std::string>{};
sink << atom("SpawnServ")
<< singletons::get_actor_registry()->get_named(atom("SpawnServ"));
});
header hdr{message_type::server_handshake, 0, version,
this_node_, invalid_node_id,
......@@ -663,17 +638,12 @@ void instance::write_server_handshake(buffer_type& out_buf,
write(out_buf, hdr, &writer);
}
void instance::write_client_handshake(buffer_type& buf) {
using namespace detail;
auto writer = make_callback([&](serializer& sink) {
sink << atom("SpawnServ")
<< singletons::get_actor_registry()->get_named(atom("SpawnServ"));
});
void instance::write_client_handshake(buffer_type& buf,
const node_id& remote_side) {
write(buf,
message_type::client_handshake, nullptr, 0,
this_node_, invalid_node_id,
invalid_actor_id, invalid_actor_id,
&writer);
this_node_, remote_side,
invalid_actor_id, invalid_actor_id);
}
void instance::write_dispatch_error(buffer_type& buf,
......@@ -704,39 +674,6 @@ 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();
}
auto instance::named_actors(const node_id& nid)
-> optional<const named_actors_map&> {
auto i = named_actors_.find(nid);
if (i == named_actors_.end())
return none;
return i->second;
}
void instance::read_named_actors(binary_deserializer& bd, const node_id& nid) {
auto& storage = named_actors_[nid];
CAF_ASSERT(storage.empty());
while (! bd.at_end()) {
atom_value key;
actor value;
bd >> key >> value;
storage.emplace(key, std::move(value));
}
}
} // namespace basp
} // namespace io
} // namespace caf
......@@ -19,6 +19,8 @@
#include "caf/io/basp_broker.hpp"
#include <limits>
#include "caf/exception.hpp"
#include "caf/make_counted.hpp"
#include "caf/event_based_actor.hpp"
......@@ -208,10 +210,20 @@ void basp_broker_state::deliver(const node_id& source_node,
}
abstract_actor_ptr dest;
uint32_t rsn = exit_reason::remote_link_unreachable;
if (dest_node == this_node())
std::tie(dest, rsn) = registry->get_entry(dest_actor);
else
if (dest_node != this_node())
dest = get_namespace().get_or_put(dest_node, dest_actor);
else
if (dest_actor == std::numeric_limits<actor_id>::max()) {
// this hack allows CAF to talk to older CAF versions; CAF <= 0.14 will
// discard this message silently as the receiver is not found, while
// CAF >= 0.14.1 will use the operation data to identify named actors
auto dest_name = static_cast<atom_value>(mid.integer_value());
CAF_LOG_DEBUG("dest_name = " << to_string(dest_name));
mid = message_id::make(); // override this since the message is async
dest = actor_cast<abstract_actor_ptr>(registry->get_named(dest_name));
} else {
std::tie(dest, rsn) = registry->get_entry(dest_actor);
}
if (! dest) {
CAF_LOG_INFO("cannot deliver message, destination not found");
if (mid.valid() && src != invalid_actor_addr) {
......@@ -223,9 +235,73 @@ 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(const node_id& nid) {
CAF_LOG_TRACE(CAF_TSARG(nid));
auto& tmp = spawn_servers[nid];
tmp = spawn<hidden>([=](event_based_actor* this_actor) -> behavior {
return {
[=](ok_atom, const std::string& /* key == "info" */,
const actor_addr& config_serv_addr, const std::string& /* name */) {
auto config_serv = actor_cast<actor>(config_serv_addr);
this_actor->monitor(config_serv);
this_actor->become(
[=](spawn_atom, std::string& type, message& args)
-> delegated<either<ok_atom, actor_addr, std::set<std::string>>
::or_else<error_atom, std::string>> {
this_actor->delegate(config_serv, get_atom::value,
std::move(type), std::move(args));
return {};
},
[=](const down_msg& dm) {
this_actor->quit(dm.reason);
},
others >> [=] {
CAF_LOGF_ERROR("spawn server has received unexpected message: "
<< to_string(this_actor->current_message()));
}
);
},
after(std::chrono::minutes(5)) >> [=] {
CAF_LOGF_INFO("no spawn server on node " << to_string(nid));
this_actor->quit();
}
};
});
using namespace detail;
singletons::get_actor_registry()->put(tmp.id(),
actor_cast<abstract_actor_ptr>(tmp));
auto writer = make_callback([](serializer& sink) {
auto msg = make_message(sys_atom::value, get_atom::value, "info");
msg.serialize(sink);
});
auto path = instance.tbl().lookup(nid);
if (! path) {
CAF_LOG_ERROR("learned_new_node called, but no route to nid");
return;
}
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
instance.write(path->wr_buf,
basp::message_type::dispatch_message,
nullptr, static_cast<uint64_t>(atom("SpawnServ")),
this_node(), nid,
tmp.id(), std::numeric_limits<actor_id>::max(),
&writer);
instance.flush(*path);
}
void basp_broker_state::learned_new_node_directly(const node_id& nid,
bool was_indirectly_before) {
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_TSARG(nid));
if (! was_indirectly_before)
learned_new_node(nid);
}
void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_TSARG(nid));
learned_new_node(nid);
if (! enable_automatic_connections)
return;
actor bb = self; // a handle for our helper back to this BASP broker
......@@ -236,14 +312,7 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
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
// this config is send from the remote `ConfigServ`
[=](ok_atom, const std::string&, message& msg) {
CAF_LOGF_DEBUG("received requested config: " << to_string(msg));
// whatever happens, we are done afterwards
......@@ -290,10 +359,22 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
return;
}
using namespace detail;
auto tmp = spawn<detached + hidden + lazy_init>(connection_helper, self);
auto tmp = spawn<detached + hidden>(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());
auto writer = make_callback([](serializer& sink) {
auto msg = make_message(get_atom::value, "basp.default-connectivity");
msg.serialize(sink);
});
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
instance.write(path->wr_buf,
basp::message_type::dispatch_message,
nullptr, static_cast<uint64_t>(atom("ConfigServ")),
this_node(), nid,
tmp.id(), std::numeric_limits<actor_id>::max(),
&writer);
instance.flush(*path);
}
void basp_broker_state::set_context(connection_handle hdl) {
......@@ -491,17 +572,12 @@ behavior basp_broker::make_behavior() {
auto rp = make_response_promise();
rp.deliver(error_atom::value, std::move(errmsg));
};
auto na = state.instance.named_actors(nid);
if (! na) {
auto i = state.spawn_servers.find(nid);
if (i == state.spawn_servers.end()) {
err("no connection to requested node");
return {};
}
auto i = na->find(atom("SpawnServ"));
if (i == na->end()) {
err("no spawn server on requested node");
return {};
}
delegate(i->second, get_atom::value, std::move(type), std::move(args));
delegate(i->second, spawn_atom::value, std::move(type), std::move(args));
return {};
},
[=](ok_atom, const std::string& key, message& value) {
......
......@@ -258,7 +258,7 @@ void middleman::initialize() {
}
// announce io-related types
announce<network::protocol>("caf::io::network::protocol");
announce<network::address_listing>("caf::network::address_listing");
announce<network::address_listing>("caf::io::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");
......
......@@ -25,6 +25,7 @@
#include <array>
#include <mutex>
#include <memory>
#include <limits>
#include <iostream>
#include <condition_variable>
......@@ -36,6 +37,43 @@
#include "caf/io/network/interfaces.hpp"
#include "caf/io/network/test_multiplexer.hpp"
namespace std {
ostream& operator<<(ostream& out, const caf::io::basp::message_type& x) {
return out << to_string(x);
}
template <class T>
ostream& operator<<(ostream& out, const caf::variant<caf::anything, T>& x) {
using std::to_string;
using caf::to_string;
using caf::io::basp::to_string;
if (get<caf::anything>(&x) != nullptr)
return out << "*";
return out << to_string(get<T>(x));
}
} // namespace std
namespace caf {
template <class T>
bool operator==(const variant<anything, T>& x, const T& y) {
return get<anything>(&x) != nullptr || get<T>(x) == y;
}
template <class T>
bool operator==(const T& x, const variant<anything, T>& y) {
return (y == x);
}
template <class T>
std::string to_string(const variant<anything, T>& x) {
return get<anything>(&x) == nullptr ? std::string{"*"} : to_string(get<T>(x));
}
} // namespace caf
using namespace std;
using namespace caf;
using namespace caf::io;
......@@ -75,6 +113,7 @@ public:
auto mm = middleman::instance();
aut_ = mm->get_named_broker<basp_broker>(atom("_BASP"));
this_node_ = detail::singletons::get_node_id();
CAF_TEST_VERBOSE("this node: " << to_string(this_node_));
self_.reset(new scoped_actor);
// run the initialization message of the BASP broker
mpx_->exec_runnable();
......@@ -185,6 +224,22 @@ public:
using payload_writer = basp::instance::payload_writer;
void to_payload(binary_serializer&) {
// end of recursion
}
template <class T, class... Ts>
void to_payload(binary_serializer& bs, const T& x, const Ts&... xs) {
bs << x;
to_payload(bs, xs...);
}
template <class... Ts>
void to_payload(buffer& buf, const Ts&... xs) {
binary_serializer bs{std::back_inserter(buf), &get_namespace()};
to_payload(bs, xs...);
}
void to_buf(buffer& buf, basp::header& hdr, payload_writer* writer) {
instance().write(buf, hdr, writer);
}
......@@ -233,15 +288,26 @@ public:
{basp::message_type::client_handshake, 0, 0,
remote_node(i), this_node(),
invalid_actor_id, invalid_actor_id});
auto na = registry()->named_actors();
if (published_actor_id != invalid_actor_id)
m.expect(hdl,
{basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id,
published_actor_id, invalid_actor_id},
basp::message_type::server_handshake, any_vals, basp::version,
this_node(), node_id{invalid_node_id},
published_actor_id, invalid_actor_id,
published_actor_id,
published_actor_ifs,
atom("SpawnServ"),
na[atom("SpawnServ")]);
published_actor_ifs);
else
m.expect(hdl,
basp::message_type::server_handshake, any_vals, basp::version,
this_node(), node_id{invalid_node_id},
invalid_actor_id, invalid_actor_id);
// upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node
m.expect(hdl,
basp::message_type::dispatch_message, any_vals,
static_cast<uint64_t>(atom("SpawnServ")),
this_node(), remote_node(i),
any_vals, std::numeric_limits<actor_id>::max(),
make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
auto path = tbl().lookup(remote_node(i));
......@@ -292,29 +358,49 @@ public:
}
template <class... Ts>
mock_t& expect(connection_handle hdl, basp::header hdr, const Ts&... xs) {
CAF_MESSAGE("expect " << num++ << ". sent message to be a "
<< to_string(hdr.operation));
mock_t& expect(connection_handle hdl,
variant<anything, basp::message_type> operation,
variant<anything, uint32_t> payload_len,
variant<anything, uint64_t> operation_data,
variant<anything, node_id> source_node,
variant<anything, node_id> dest_node,
variant<anything, actor_id> source_actor,
variant<anything, actor_id> dest_actor,
const Ts&... xs) {
CAF_MESSAGE("expect " << num << ". sent message to be a "
<< operation);
buffer buf;
this_->to_buf(buf, hdr, nullptr, xs...);
this_->to_payload(buf, xs...);
buffer& ob = this_->mpx()->output_buffer(hdl);
CAF_REQUIRE(buf.size() <= ob.size());
auto first = ob.begin();
auto last = ob.begin() + static_cast<ptrdiff_t>(buf.size());
buffer cpy(first, last);
ob.erase(first, last);
basp::header tmp;
CAF_TEST_VERBOSE("output buffer has " << ob.size() << " bytes");
CAF_REQUIRE(ob.size() >= basp::header_size);
basp::header hdr;
{ // lifetime scope of source
auto source = this_->make_deserializer(cpy);
basp::read_hdr(source, tmp);
}
CAF_REQUIRE(to_string(hdr) == to_string(tmp));
CAF_REQUIRE(equal(buf.begin(), buf.begin() + basp::header_size,
cpy.begin()));
buf.erase(buf.begin(), buf.begin() + basp::header_size);
cpy.erase(cpy.begin(), cpy.begin() + basp::header_size);
CAF_REQUIRE(hdr.payload_len == buf.size() && cpy.size() == buf.size());
CAF_REQUIRE(hexstr(buf) == hexstr(cpy));
auto source = this_->make_deserializer(ob);
basp::read_hdr(source, hdr);
}
buffer payload;
if (hdr.payload_len > 0) {
CAF_REQUIRE(ob.size() >= (basp::header_size + hdr.payload_len));
auto first = ob.begin() + basp::header_size;
auto end = first + hdr.payload_len;
payload.assign(first, end);
CAF_TEST_VERBOSE("erase " << std::distance(ob.begin(), end)
<< " bytes from output buffer");
ob.erase(ob.begin(), end);
} else {
ob.erase(ob.begin(), ob.begin() + basp::header_size);
}
CAF_CHECK_EQUAL(operation, hdr.operation);
CAF_CHECK_EQUAL(payload_len, hdr.payload_len);
CAF_CHECK_EQUAL(operation_data, hdr.operation_data);
CAF_CHECK_EQUAL(source_node, hdr.source_node);
CAF_CHECK_EQUAL(dest_node, hdr.dest_node);
CAF_CHECK_EQUAL(source_actor, hdr.source_actor);
CAF_CHECK_EQUAL(dest_actor, hdr.dest_actor);
CAF_REQUIRE(payload.size() == buf.size());
CAF_REQUIRE(hexstr(payload) == hexstr(buf));
++num;
return *this;
}
......@@ -367,8 +453,7 @@ CAF_TEST(empty_server_handshake) {
invalid_actor_id, invalid_actor_id};
CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK(hdr == expected);
CAF_CHECK(! payload.empty());
CAF_CHECK_EQUAL(to_string(hdr), to_string(expected));
}
CAF_TEST(non_empty_server_handshake) {
......@@ -379,13 +464,11 @@ CAF_TEST(non_empty_server_handshake) {
{"caf::replies_to<@u16>::with<@u16>"});
instance().write_server_handshake(buf, 4242);
buffer expected_buf;
auto na = registry()->named_actors();
basp::header expected{basp::message_type::server_handshake, 0, basp::version,
this_node(), invalid_node_id,
self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr,
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"},
atom("SpawnServ"), na[atom("SpawnServ")]);
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"});
CAF_CHECK(hexstr(buf) == hexstr(expected_buf));
}
......@@ -397,9 +480,9 @@ CAF_TEST(client_handshake_and_dispatch) {
remote_node(0), this_node(), pseudo_remote(0)->id(), self()->id()},
make_message(1, 2, 3))
.expect(remote_hdl(0),
{basp::message_type::announce_proxy_instance, 0, 0,
basp::message_type::announce_proxy_instance, uint32_t{0}, uint64_t{0},
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()});
invalid_actor_id, pseudo_remote(0)->id());
// must've created a proxy for our remote actor
CAF_REQUIRE(get_namespace().count_proxies(remote_node(0)) == 1);
// must've send remote node a message that this proxy is monitored now
......@@ -436,9 +519,9 @@ CAF_TEST(message_forwarding) {
invalid_actor_id, pseudo_remote(1)->id()},
msg)
.expect(remote_hdl(1),
{basp::message_type::dispatch_message, 0, 0,
basp::message_type::dispatch_message, any_vals, uint64_t{0},
remote_node(0), remote_node(1),
invalid_actor_id, pseudo_remote(1)->id()},
invalid_actor_id, pseudo_remote(1)->id(),
msg);
}
......@@ -470,14 +553,19 @@ CAF_TEST(remote_actor_and_send) {
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")))
basp::message_type::client_handshake, uint32_t{0}, uint64_t{0},
this_node(), remote_node(0),
invalid_actor_id, invalid_actor_id)
.expect(remote_hdl(0),
{basp::message_type::announce_proxy_instance, 0, 0,
basp::message_type::dispatch_message, any_vals,
static_cast<uint64_t>(atom("SpawnServ")),
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()});
any_vals, std::numeric_limits<actor_id>::max(),
make_message(sys_atom::value, get_atom::value, "info"))
.expect(remote_hdl(0),
basp::message_type::announce_proxy_instance, uint32_t{0}, uint64_t{0},
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id());
// basp broker should've send the proxy
f.await(
[&](ok_atom, node_id nid, actor_addr res, std::set<std::string> ifs) {
......@@ -503,9 +591,9 @@ CAF_TEST(remote_actor_and_send) {
mpx()->exec_runnable(); // process forwarded message in basp_broker
mock()
.expect(remote_hdl(0),
{basp::message_type::dispatch_message, 0, 0,
basp::message_type::dispatch_message, any_vals, uint64_t{0},
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()},
invalid_actor_id, pseudo_remote(0)->id(),
make_message(42));
auto msg = make_message("hi there!");
CAF_MESSAGE("send message via BASP (from proxy)");
......@@ -537,9 +625,9 @@ CAF_TEST(actor_serialize_and_deserialize) {
auto prx = get_namespace().get_or_put(remote_node(0), pseudo_remote(0)->id());
mock()
.expect(remote_hdl(0),
{basp::message_type::announce_proxy_instance, 0, 0,
basp::message_type::announce_proxy_instance, uint32_t{0}, uint64_t{0},
this_node(), prx->node(),
invalid_actor_id, prx->id()});
invalid_actor_id, prx->id());
CAF_CHECK(prx->node() == remote_node(0));
CAF_CHECK(prx->id() == pseudo_remote(0)->id());
auto testee = spawn(testee_impl);
......@@ -557,9 +645,9 @@ CAF_TEST(actor_serialize_and_deserialize) {
// output buffer must contain the reflected message
mock()
.expect(remote_hdl(0),
{basp::message_type::dispatch_message, 0, 0,
basp::message_type::dispatch_message, any_vals, uint64_t{0},
this_node(), prx->node(),
testee->id(), prx->id()},
testee->id(), prx->id(),
msg);
}
......@@ -579,10 +667,18 @@ CAF_TEST(indirect_connections) {
remote_node(0), this_node(),
pseudo_remote(0)->id(), self()->id()},
make_message("hello from jupiter!"))
// this asks Jupiter if it has a 'SpawnServ'
.expect(remote_hdl(1),
{basp::message_type::announce_proxy_instance, 0, 0,
basp::message_type::dispatch_message, any_vals,
static_cast<uint64_t>(atom("SpawnServ")),
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()});
any_vals, std::numeric_limits<actor_id>::max(),
make_message(sys_atom::value, get_atom::value, "info"))
// this tells Jupiter that Earth learned the address of one its actors
.expect(remote_hdl(1),
basp::message_type::announce_proxy_instance, uint32_t{0}, uint64_t{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 {
......@@ -594,9 +690,9 @@ CAF_TEST(indirect_connections) {
mpx()->exec_runnable(); // process forwarded message in basp_broker
mock()
.expect(remote_hdl(1),
{basp::message_type::dispatch_message, 0, 0,
basp::message_type::dispatch_message, any_vals, uint64_t{0},
this_node(), remote_node(0),
self()->id(), pseudo_remote(0)->id()},
self()->id(), pseudo_remote(0)->id(),
make_message("hello from earth!"));
}
......@@ -625,40 +721,37 @@ CAF_TEST(automatic_connection) {
pseudo_remote(0)->id(), self()->id()},
make_message("hello from jupiter!"))
.expect(remote_hdl(1),
{basp::message_type::name_lookup, 0,
basp::message_type::dispatch_message, any_vals,
static_cast<uint64_t>(atom("SpawnServ")),
this_node(), remote_node(0),
any_vals, std::numeric_limits<actor_id>::max(),
make_message(sys_atom::value, get_atom::value, "info"))
.expect(remote_hdl(1),
basp::message_type::dispatch_message, any_vals,
static_cast<uint64_t>(atom("ConfigServ")),
this_node(), remote_node(0),
abstract_actor::latest_actor_id(), // spawned by BASP broker
invalid_actor_id})
any_vals, // actor ID of an actor spawned by the BASP broker
std::numeric_limits<actor_id>::max(),
make_message(get_atom::value, "basp.default-connectivity"))
.expect(remote_hdl(1),
{basp::message_type::announce_proxy_instance, 0, 0,
basp::message_type::announce_proxy_instance, uint32_t{0}, uint64_t{0},
this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()});
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));
network::address_listing res;
res[network::protocol::ipv4].push_back("jupiter");
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));
invalid_actor_id, connection_helper},
make_message(ok_atom::value, "basp.default-connectivity",
make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and
// send the scribe handle over to the BASP broker
mpx()->exec_runnable();
......@@ -672,10 +765,9 @@ CAF_TEST(automatic_connection) {
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")));
basp::message_type::client_handshake, uint32_t{0}, uint64_t{0},
this_node(), remote_node(0),
invalid_actor_id, invalid_actor_id);
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());
......@@ -692,9 +784,9 @@ CAF_TEST(automatic_connection) {
CAF_MESSAGE("response message must take direct route now");
mock()
.expect(remote_hdl(0),
{basp::message_type::dispatch_message, 0, 0,
basp::message_type::dispatch_message, any_vals, uint64_t{0},
this_node(), remote_node(0),
self()->id(), pseudo_remote(0)->id()},
self()->id(), pseudo_remote(0)->id(),
make_message("hello from earth!"));
CAF_CHECK(mpx()->output_buffer(remote_hdl(1)).size() == 0);
}
......
......@@ -111,10 +111,11 @@ CAF_TEST(remote_spawn) {
announce_actor_type("mirror", mirror);
auto argv = caf::test::engine::argv();
auto argc = caf::test::engine::argc();
uint16_t port;
uint16_t port = 0;
auto r = message_builder(argv, argv + argc).extract_opts({
{"server,s", "run as server (don't run client"},
{"client,c", "add client port (two needed)", port},
{"port,p", "force a port in server mode", port},
{"use-asio", "use ASIO network backend (if available)"}
});
if (! r.error.empty() || r.opts.count("help") > 0 || ! r.remainder.empty()) {
......@@ -135,7 +136,7 @@ CAF_TEST(remote_spawn) {
return;
}
auto serv = spawn(server);
port = io::publish(serv, 0);
port = io::publish(serv, port);
CAF_TEST_INFO("published server at port " << port);
if (r.opts.count("server") == 0) {
auto child = detail::run_program(invalid_actor, argv[0], "-n", "-s",
......
......@@ -219,7 +219,8 @@ CAF_TEST(test_uniform_type) {
"caf::io::connection_closed_msg",
"caf::io::network::protocol",
"caf::io::new_connection_msg",
"caf::io::new_data_msg"));
"caf::io::new_data_msg",
"caf::io::network::address_listing"));
CAF_MESSAGE("io types checked");
}
// check whether enums can be announced as members
......
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