Commit 67b7161c authored by Dominik Charousset's avatar Dominik Charousset

Remove message forwarding

In our BASP redesign, we have decided to remove the (broken) overlay
aspect. Instead, we enable automatic connections per default and require
CAF nodes to establish connections before exchanging messages.
parent fac1cc62
......@@ -22,13 +22,10 @@
#include <cstdint>
#include "caf/error.hpp"
#include "caf/node_id.hpp"
#include "caf/io/basp/message_type.hpp"
#include "caf/meta/omittable.hpp"
#include "caf/meta/type_name.hpp"
#include "caf/io/basp/message_type.hpp"
namespace caf {
namespace io {
namespace basp {
......@@ -46,20 +43,16 @@ struct header {
uint8_t flags;
uint32_t payload_len;
uint64_t operation_data;
node_id source_node;
node_id dest_node;
actor_id source_actor;
actor_id dest_actor;
header(message_type m_operation, uint8_t m_flags, uint32_t m_payload_len,
uint64_t m_operation_data, node_id m_source_node, node_id m_dest_node,
actor_id m_source_actor, actor_id m_dest_actor)
uint64_t m_operation_data, actor_id m_source_actor,
actor_id m_dest_actor)
: operation(m_operation),
flags(m_flags),
payload_len(m_payload_len),
operation_data(m_operation_data),
source_node(std::move(m_source_node)),
dest_node(std::move(m_dest_node)),
source_actor(m_source_actor),
dest_actor(m_dest_actor) {
// nop
......@@ -80,13 +73,9 @@ struct header {
template <class Inspector>
typename Inspector::result_type inspect(Inspector& f, header& hdr) {
uint8_t pad = 0;
return f(meta::type_name("header"),
hdr.operation,
meta::omittable(), pad,
meta::omittable(), pad,
hdr.flags, hdr.payload_len, hdr.operation_data,
hdr.source_node, hdr.dest_node,
hdr.source_actor, hdr.dest_actor);
return f(meta::type_name("header"), hdr.operation, meta::omittable(), pad,
meta::omittable(), pad, hdr.flags, hdr.payload_len,
hdr.operation_data, hdr.source_actor, hdr.dest_actor);
}
/// @relates header
......@@ -113,8 +102,7 @@ inline bool is_heartbeat(const header& hdr) {
bool valid(const header& hdr);
/// Size of a BASP header in serialized form
constexpr size_t header_size = node_id::serialized_size * 2
+ sizeof(actor_id) * 2
constexpr size_t header_size = sizeof(actor_id) * 2
+ sizeof(uint32_t) * 2
+ sizeof(uint64_t);
......
......@@ -200,15 +200,12 @@ public:
buffer_type& out_buf, optional<uint16_t> port);
/// Writes the client handshake to `buf`.
static void write_client_handshake(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side,
static void write_client_handshake(execution_unit* ctx, buffer_type& buf,
const node_id& this_node,
const std::string& app_identifier);
/// Writes the client handshake to `buf`.
void write_client_handshake(execution_unit* ctx,
buffer_type& buf, const node_id& remote_side);
void write_client_handshake(execution_unit* ctx, buffer_type& buf);
/// Writes an `announce_proxy` to `buf`.
void write_announce_proxy(execution_unit* ctx, buffer_type& buf,
......@@ -220,8 +217,7 @@ public:
const error& rsn);
/// Writes a `heartbeat` to `buf`.
void write_heartbeat(execution_unit* ctx, buffer_type& buf,
const node_id& remote_side);
void write_heartbeat(execution_unit* ctx, buffer_type& buf);
const node_id& this_node() const {
return this_node_;
......
......@@ -338,9 +338,9 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
}
// send message to SpawnServ of remote node
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, make_message_id().integer_value(), this_node(), nid,
tmp.id(), invalid_actor_id};
basp::header::named_receiver_flag, 0,
make_message_id().integer_value(), tmp.id(),
invalid_actor_id};
// 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(self->context(), get_buffer(path->hdl),
......@@ -388,9 +388,9 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
return sink(name_atm, stages, msg);
});
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, make_message_id().integer_value(), this_node(), nid,
tmp.id(), invalid_actor_id};
basp::header::named_receiver_flag, 0,
make_message_id().integer_value(), tmp.id(),
invalid_actor_id};
instance.write(self->context(), get_buffer(path->hdl),
hdr, &writer);
instance.flush(*path);
......@@ -403,7 +403,7 @@ void basp_broker_state::set_context(connection_handle hdl) {
auto i = ctx.find(hdl);
if (i == ctx.end()) {
CAF_LOG_DEBUG("create new BASP context:" << CAF_ARG(hdl));
basp::header hdr{basp::message_type::server_handshake, 0, 0, 0, none, none,
basp::header hdr{basp::message_type::server_handshake, 0, 0, 0,
invalid_actor_id, invalid_actor_id};
i = ctx
.emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl,
......@@ -544,9 +544,8 @@ behavior basp_broker::make_behavior() {
return sink(dest_name, cme->stages, const_cast<message&>(msg));
});
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, cme->mid.integer_value(), state.this_node(),
dest_node, src->id(), invalid_actor_id};
basp::header::named_receiver_flag, 0,
cme->mid.integer_value(), src->id(), invalid_actor_id};
state.instance.write(context(), state.get_buffer(path->hdl),
hdr, &writer);
state.instance.flush(*path);
......
......@@ -48,8 +48,7 @@ behavior datagram_connection_broker(broker* self, uint16_t port,
auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr));
basp::instance::write_client_handshake(self->context(),
self->wr_buf(hdl),
none, this_node,
self->wr_buf(hdl), this_node,
app_id);
}
}
......
......@@ -40,8 +40,6 @@ std::string to_string(const header &hdr) {
<< to_bin(hdr.flags) << ", "
<< hdr.payload_len << ", "
<< hdr.operation_data << ", "
<< to_string(hdr.source_node) << ", "
<< to_string(hdr.dest_node) << ", "
<< hdr.source_actor << ", "
<< hdr.dest_actor
<< "}";
......@@ -53,70 +51,43 @@ bool operator==(const header& lhs, const header& rhs) {
&& lhs.flags == rhs.flags
&& lhs.payload_len == rhs.payload_len
&& lhs.operation_data == rhs.operation_data
&& lhs.source_node == rhs.source_node
&& lhs.dest_node == rhs.dest_node
&& lhs.source_actor == rhs.source_actor
&& lhs.dest_actor == rhs.dest_actor;
}
namespace {
bool valid(const node_id& val) {
return val != none;
}
template <class T>
bool zero(T val) {
return val == 0;
}
bool server_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& zero(hdr.dest_actor)
&& !zero(hdr.operation_data);
return !zero(hdr.operation_data);
}
bool client_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor);
return zero(hdr.source_actor) && zero(hdr.dest_actor);
}
bool dispatch_message_valid(const header& hdr) {
return valid(hdr.dest_node)
&& (!zero(hdr.dest_actor) || hdr.has(header::named_receiver_flag))
&& !zero(hdr.payload_len);
return (!zero(hdr.dest_actor) || hdr.has(header::named_receiver_flag))
&& !zero(hdr.payload_len);
}
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.source_actor)
&& !zero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& zero(hdr.operation_data);
return zero(hdr.source_actor) && !zero(hdr.dest_actor)
&& zero(hdr.payload_len) && zero(hdr.operation_data);
}
bool kill_proxy_instance_valid(const header& hdr) {
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);
return !zero(hdr.source_actor) && zero(hdr.dest_actor)
&& !zero(hdr.payload_len) && zero(hdr.operation_data);
}
bool heartbeat_valid(const header& hdr) {
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);
return zero(hdr.source_actor) && zero(hdr.dest_actor) && zero(hdr.payload_len)
&& zero(hdr.operation_data);
}
} // namespace <anonymous>
......
......@@ -79,36 +79,6 @@ connection_state instance::handle(execution_unit* ctx,
}
}
CAF_LOG_DEBUG(CAF_ARG(hdr));
// needs forwarding?
if (!is_handshake(hdr) && !is_heartbeat(hdr) && hdr.dest_node != this_node_) {
CAF_LOG_DEBUG("forward message");
auto path = lookup(hdr.dest_node);
if (path) {
binary_serializer bs{ctx, callee_.get_buffer(path->hdl)};
auto e = bs(hdr);
if (e)
return err();
if (payload != nullptr)
bs.apply_raw(payload->size(), payload->data());
flush(*path);
notify<hook::message_forwarded>(hdr, payload);
} else {
CAF_LOG_INFO("cannot forward message, no route to destination");
if (hdr.source_node != this_node_) {
// TODO: signalize error back to sending node
auto reverse_path = lookup(hdr.source_node);
if (!reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source");
} else {
CAF_LOG_WARNING("not implemented yet: signalize forward failure");
}
} else {
CAF_LOG_WARNING("lost packet with probably spoofed source");
}
notify<hook::message_forwarding_failed>(hdr, payload);
}
return await_header;
}
if (!handle(ctx, dm.handle, hdr, payload))
return err();
return await_header;
......@@ -118,7 +88,7 @@ void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
write_heartbeat(ctx, callee_.get_buffer(kvp.first), kvp.second);
write_heartbeat(ctx, callee_.get_buffer(kvp.first));
callee_.flush(kvp.first);
}
}
......@@ -209,7 +179,6 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
const_cast<message&>(msg));
});
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(),
sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id()};
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer);
flush(*path);
......@@ -253,11 +222,10 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf,
}
CAF_LOG_DEBUG_IF(!pa && port, "no actor published");
auto writer = make_callback([&](serializer& sink) -> error {
auto id = get_or(callee_.config(), "middleman.app-identifier",
auto appid = get_or(callee_.config(), "middleman.app-identifier",
defaults::middleman::app_identifier);
auto e = sink(id);
if (e)
return e;
if (auto err = sink(this_node_, appid))
return err;
if (pa != nullptr) {
auto i = pa->first ? pa->first->id() : invalid_actor_id;
return sink(i, pa->second);
......@@ -267,7 +235,6 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf,
return sink(aid, tmp);
});
header hdr{message_type::server_handshake, 0, 0, version,
this_node_, none,
(pa != nullptr) && pa->first ? pa->first->id() : invalid_actor_id,
invalid_actor_id};
write(ctx, out_buf, hdr, &writer);
......@@ -275,22 +242,20 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf,
void instance::write_client_handshake(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side,
const node_id& this_node,
const std::string& app_identifier) {
CAF_LOG_TRACE(CAF_ARG(remote_side));
CAF_LOG_TRACE(CAF_ARG(this_node) << CAF_ARG(app_identifier));
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<std::string&>(app_identifier));
return sink(const_cast<node_id&>(this_node),
const_cast<std::string&>(app_identifier));
});
header hdr{message_type::client_handshake, 0, 0, 0,
this_node, remote_side, invalid_actor_id, invalid_actor_id};
header hdr{message_type::client_handshake, 0, 0, 0, invalid_actor_id,
invalid_actor_id};
write(ctx, buf, hdr, &writer);
}
void instance::write_client_handshake(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side) {
write_client_handshake(ctx, buf, remote_side, this_node_,
void instance::write_client_handshake(execution_unit* ctx, buffer_type& buf) {
write_client_handshake(ctx, buf, this_node_,
get_or(callee_.config(), "middleman.app-identifier",
defaults::middleman::app_identifier));
}
......@@ -298,8 +263,7 @@ void instance::write_client_handshake(execution_unit* ctx,
void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid));
header hdr{message_type::announce_proxy, 0, 0, 0,
this_node_, dest_node, invalid_actor_id, aid};
header hdr{message_type::announce_proxy, 0, 0, 0, invalid_actor_id, aid};
write(ctx, buf, hdr);
}
......@@ -310,17 +274,14 @@ void instance::write_kill_proxy(execution_unit* ctx, buffer_type& buf,
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<error&>(rsn));
});
header hdr{message_type::kill_proxy, 0, 0, 0,
this_node_, dest_node, aid, invalid_actor_id};
header hdr{message_type::kill_proxy, 0, 0, 0, aid, invalid_actor_id};
write(ctx, buf, hdr, &writer);
}
void instance::write_heartbeat(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side) {
CAF_LOG_TRACE(CAF_ARG(remote_side));
header hdr{message_type::heartbeat, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id};
void instance::write_heartbeat(execution_unit* ctx, buffer_type& buf) {
CAF_LOG_TRACE("");
header hdr{message_type::heartbeat, 0, 0, 0, invalid_actor_id,
invalid_actor_id};
write(ctx, buf, hdr);
}
......@@ -333,6 +294,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
// handle message to ourselves
switch (hdr.operation) {
case message_type::server_handshake: {
node_id source_node;
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
if (!payload_valid()) {
......@@ -341,47 +303,41 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
} else {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
if (bd(source_node, remote_appid))
return false;
auto appid = get_if<std::string>(&callee_.config(),
"middleman.app-identifier");
if ((appid && *appid != remote_appid)
|| (!appid && !remote_appid.empty())) {
auto appid = get_or(callee_.config(), "middleman.app-identifier",
defaults::middleman::app_identifier);
if (appid != remote_appid) {
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
e = bd(aid, sigs);
if (e)
if (bd(aid, sigs))
return false;
}
// close self connection after handshake is done
if (hdr.source_node == this_node_) {
// Close self connection immediately after handshake.
if (source_node == this_node_) {
CAF_LOG_DEBUG("close connection to self immediately");
callee_.finalize_handshake(hdr.source_node, aid, sigs);
callee_.finalize_handshake(source_node, aid, sigs);
return false;
}
// close this connection if we already have a direct connection
if (tbl_.lookup_direct(hdr.source_node)) {
CAF_LOG_DEBUG("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
// Close redundant connections.
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG("close redundant connection:" << CAF_ARG(source_node));
callee_.finalize_handshake(source_node, aid, sigs);
return false;
}
// add direct route to this node and remove any indirect entry
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add_direct(hdl, hdr.source_node);
auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// Add new route to this node.
CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
auto path = tbl_.lookup(source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return false;
}
write_client_handshake(ctx, callee_.get_buffer(path->hdl),
hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
write_client_handshake(ctx, callee_.get_buffer(path->hdl));
callee_.learned_new_node_directly(source_node, false);
callee_.finalize_handshake(source_node, aid, sigs);
flush(*path);
break;
}
......@@ -389,85 +345,97 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
if (!payload_valid()) {
CAF_LOG_ERROR("fail to receive the app identifier");
return false;
} else {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return false;
auto appid = get_if<std::string>(&callee_.config(),
"middleman.app-identifier");
if ((appid && *appid != remote_appid)
|| (!appid && !remote_appid.empty())) {
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
}
if (tbl_.lookup_direct(hdr.source_node)) {
binary_deserializer bd{ctx, *payload};
node_id source_node;
std::string remote_appid;
if (bd(source_node, remote_appid))
return false;
auto appid = get_if<std::string>(&callee_.config(),
"middleman.app-identifier");
if ((appid && *appid != remote_appid)
|| (!appid && !remote_appid.empty())) {
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG("received second client handshake:"
<< CAF_ARG(hdr.source_node));
<< CAF_ARG(source_node));
break;
}
// add direct route to this node and remove any indirect entry
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add_direct(hdl, hdr.source_node);
auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
// Add route to this node.
CAF_LOG_DEBUG("new connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node);
callee_.learned_new_node_directly(source_node, false);
break;
}
case message_type::dispatch_message: {
// Sanity checks.
if (!payload_valid())
return false;
// 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(hdl);
if (hdr.source_node != none
&& hdr.source_node != this_node_
&& last_hop != hdr.source_node
&& !tbl_.lookup_direct(hdr.source_node)
&& tbl_.add_indirect(last_hop, hdr.source_node))
callee_.learned_new_node_indirectly(hdr.source_node);
auto source_node = tbl_.lookup_direct(hdl);
if (source_node == none) {
CAF_LOG_ERROR("received dispatch_message before handshake");
return false;
}
// Deserialize payload.
binary_deserializer bd{ctx, *payload};
auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
if (hdr.has(header::named_receiver_flag)) {
auto e = bd(receiver_name);
if (e)
if (bd(receiver_name))
return false;
}
auto e = bd(forwarding_stack, msg);
if (e)
if (bd(forwarding_stack, msg))
return false;
// Dispatch to callee_.
CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg));
if (hdr.has(header::named_receiver_flag))
callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name,
callee_.deliver(source_node, hdr.source_actor, receiver_name,
make_message_id(hdr.operation_data),
forwarding_stack, msg);
else
callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor,
callee_.deliver(source_node, hdr.source_actor, hdr.dest_actor,
make_message_id(hdr.operation_data),
forwarding_stack, msg);
break;
}
case message_type::announce_proxy:
callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
case message_type::announce_proxy: {
auto source_node = tbl_.lookup_direct(hdl);
if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake");
return false;
}
callee_.proxy_announced(source_node, hdr.dest_actor);
break;
}
case message_type::kill_proxy: {
// Sanity checks.
if (!payload_valid())
return false;
auto source_node = tbl_.lookup_direct(hdl);
if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake");
return false;
}
// Deserialize exit reason.
binary_deserializer bd{ctx, *payload};
error fail_state;
auto e = bd(fail_state);
if (e)
if (bd(fail_state))
return false;
callee_.proxies().erase(hdr.source_node, hdr.source_actor,
// Remove proxy from registry.
callee_.proxies().erase(source_node, hdr.source_actor,
std::move(fail_state));
break;
}
case message_type::heartbeat: {
CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node));
callee_.handle_heartbeat(hdr.source_node);
auto source_node = tbl_.lookup_direct(hdl);
if (source_node == none) {
CAF_LOG_ERROR("received announce_proxy before handshake");
return false;
}
callee_.handle_heartbeat(source_node);
break;
}
default:
......
......@@ -274,32 +274,24 @@ public:
// technically, the server handshake arrives
// before we send the client handshake
mock(hdl,
{basp::message_type::client_handshake, 0, 0, 0,
n.id, this_node(),
invalid_actor_id, invalid_actor_id}, std::string{})
.receive(hdl,
basp::message_type::server_handshake, no_flags,
any_vals, basp::version, this_node(), node_id{none},
published_actor_id, invalid_actor_id, std::string{},
published_actor_id,
published_actor_ifs)
// upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node
.receive(hdl,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data,
this_node(), n.id,
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_addr>{},
make_message(sys_atom::value, get_atom::value, "info"));
{basp::message_type::client_handshake, 0, 0, 0, invalid_actor_id,
invalid_actor_id},
n.id, std::string{})
.receive(hdl, basp::message_type::server_handshake, no_flags, any_vals,
basp::version, published_actor_id, invalid_actor_id, this_node(),
std::string{}, published_actor_id, published_actor_ifs)
// upon receiving our client handshake, BASP will check
// whether there is a SpawnServ actor on this node
.receive(hdl, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, invalid_actor_id,
spawn_serv_atom, std::vector<actor_addr>{},
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(n.id);
CAF_REQUIRE(path);
CAF_CHECK_EQUAL(path->hdl, n.connection);
CAF_CHECK_EQUAL(path->next_hop, n.id);
auto path = unbox(tbl().lookup(n.id));
CAF_CHECK_EQUAL(path.hdl, n.connection);
CAF_CHECK_EQUAL(path.next_hop, n.id);
}
std::pair<basp::header, buffer> read_from_out_buf(connection_handle hdl) {
......@@ -352,8 +344,6 @@ public:
maybe<uint8_t> flags,
maybe<uint32_t> payload_len,
maybe<uint64_t> operation_data,
maybe<node_id> source_node,
maybe<node_id> dest_node,
maybe<actor_id> source_actor,
maybe<actor_id> dest_actor,
const Ts&... xs) {
......@@ -386,8 +376,6 @@ public:
CAF_CHECK_EQUAL(flags, static_cast<uint8_t>(hdr.flags));
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_EQUAL(buf.size(), payload.size());
......@@ -474,9 +462,7 @@ CAF_TEST(empty_server_handshake) {
buffer payload;
std::tie(hdr, payload) = from_buf(buf);
basp::header expected{basp::message_type::server_handshake, 0,
static_cast<uint32_t>(payload.size()),
basp::version,
this_node(), none,
static_cast<uint32_t>(payload.size()), basp::version,
invalid_actor_id, invalid_actor_id};
CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr));
......@@ -492,9 +478,8 @@ CAF_TEST(non_empty_server_handshake) {
instance().write_server_handshake(mpx(), buf, uint16_t{4242});
buffer expected_buf;
basp::header expected{basp::message_type::server_handshake, 0, 0,
basp::version, this_node(), none,
self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr, std::string{},
basp::version, self()->id(), invalid_actor_id};
to_buf(expected_buf, expected, nullptr, this_node(), std::string{},
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"});
CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf));
}
......@@ -525,13 +510,11 @@ CAF_TEST(client_handshake_and_dispatch) {
// send a message via `dispatch` from node 0
mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
jupiter().id, this_node(), jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_addr>{},
make_message(1, 2, 3))
.receive(jupiter().connection,
basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_addr>{}, make_message(1, 2, 3))
.receive(jupiter().connection, basp::message_type::announce_proxy, no_flags,
no_payload, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id());
// must've created a proxy for our remote actor
CAF_REQUIRE(proxies().count_proxies(jupiter().id) == 1);
// must've send remote node a message that this proxy is monitored now
......@@ -555,24 +538,6 @@ CAF_TEST(client_handshake_and_dispatch) {
);
}
CAF_TEST(message_forwarding) {
// connect two remote nodes
connect_node(jupiter());
connect_node(mars());
auto msg = make_message(1, 2, 3);
// send a message from node 0 to node 1, forwarded by this node
mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, default_operation_data,
jupiter().id, mars().id,
invalid_actor_id, mars().dummy_actor->id()},
msg)
.receive(mars().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
default_operation_data, jupiter().id, mars().id,
invalid_actor_id, mars().dummy_actor->id(),
msg);
}
CAF_TEST(publish_and_connect) {
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
......@@ -600,27 +565,19 @@ CAF_TEST(remote_actor_and_send) {
auto na = registry()->named_actors();
mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().id, none,
jupiter().dummy_actor->id(), invalid_actor_id},
std::string{},
jupiter().dummy_actor->id(),
uint32_t{0})
.receive(jupiter().connection,
basp::message_type::client_handshake, no_flags, 1u,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, invalid_actor_id, std::string{})
.receive(jupiter().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, this_node(), jupiter().id,
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(jupiter().connection,
basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
jupiter().id, std::string{}, jupiter().dummy_actor->id(), uint32_t{0})
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, any_vals, no_operation_data, invalid_actor_id,
invalid_actor_id, this_node(), std::string{})
.receive(jupiter().connection, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, invalid_actor_id,
spawn_serv_atom, std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(jupiter().connection, basp::message_type::announce_proxy, no_flags,
no_payload, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id());
CAF_MESSAGE("BASP broker should've send the proxy");
f.receive(
[&](node_id nid, strong_actor_ptr res, std::set<std::string> ifs) {
......@@ -645,28 +602,21 @@ CAF_TEST(remote_actor_and_send) {
anon_send(actor_cast<actor>(result), 42);
mpx()->flush_runnables();
// mpx()->exec_runnable(); // process forwarded message in basp_broker
mock()
.receive(jupiter().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
default_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id(),
std::vector<actor_id>{},
make_message(42));
mock().receive(jupiter().connection, basp::message_type::dispatch_message,
no_flags, any_vals, default_operation_data, invalid_actor_id,
jupiter().dummy_actor->id(), std::vector<actor_id>{},
make_message(42));
auto msg = make_message("hi there!");
CAF_MESSAGE("send message via BASP (from proxy)");
mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
jupiter().id, this_node(),
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{},
make_message("hi there!"));
self()->receive(
[&](const string& str) {
CAF_CHECK_EQUAL(to_string(self()->current_sender()), to_string(result));
CAF_CHECK_EQUAL(self()->current_sender(), result.address());
CAF_CHECK_EQUAL(str, "hi there!");
}
);
std::vector<actor_id>{}, make_message("hi there!"));
self()->receive([&](const string& str) {
CAF_CHECK_EQUAL(to_string(self()->current_sender()), to_string(result));
CAF_CHECK_EQUAL(self()->current_sender(), result.address());
CAF_CHECK_EQUAL(str, "hi there!");
});
}
CAF_TEST(actor_serialize_and_deserialize) {
......@@ -680,11 +630,9 @@ CAF_TEST(actor_serialize_and_deserialize) {
};
connect_node(jupiter());
auto prx = proxies().get_or_put(jupiter().id, jupiter().dummy_actor->id());
mock()
.receive(jupiter().connection,
basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), prx->node(),
invalid_actor_id, prx->id());
mock().receive(jupiter().connection, basp::message_type::announce_proxy,
no_flags, no_payload, no_operation_data, invalid_actor_id,
prx->id());
CAF_CHECK_EQUAL(prx->node(), jupiter().id);
CAF_CHECK_EQUAL(prx->id(), jupiter().dummy_actor->id());
auto testee = sys.spawn(testee_impl);
......@@ -692,77 +640,23 @@ CAF_TEST(actor_serialize_and_deserialize) {
CAF_MESSAGE("send message via BASP (from proxy)");
auto msg = make_message(actor_cast<actor_addr>(prx));
mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
prx->node(), this_node(),
prx->id(), testee->id()},
std::vector<actor_id>{},
msg);
{basp::message_type::dispatch_message, 0, 0, 0, prx->id(), testee->id()},
std::vector<actor_id>{}, msg);
// testee must've responded (process forwarded message in BASP broker)
CAF_MESSAGE("wait until BASP broker writes to its output buffer");
while (mpx()->output_buffer(jupiter().connection).empty())
mpx()->exec_runnable(); // process forwarded message in basp_broker
// output buffer must contain the reflected message
mock().receive(jupiter().connection, basp::message_type::dispatch_message,
no_flags, any_vals, default_operation_data, this_node(),
prx->node(), testee->id(), prx->id(), std::vector<actor_id>{},
msg);
}
CAF_TEST(indirect_connections) {
// this node receives a message from jupiter via mars and responds via mars
// and any ad-hoc automatic connection requests are ignored
CAF_MESSAGE("self: " << to_string(self()->address()));
CAF_MESSAGE("publish self at port 4242");
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
sys.middleman().publish(self(), 4242);
mpx()->flush_runnables(); // process publish message in basp_broker
CAF_MESSAGE("connect to Mars");
connect_node(mars(), ax, self()->id());
CAF_MESSAGE("actor from Jupiter sends a message to us via Mars");
auto mx = mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
jupiter().id, this_node(),
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{},
make_message("hello from jupiter!"));
CAF_MESSAGE("expect ('sys', 'get', \"info\") from Earth to Jupiter at Mars");
// this asks Jupiter if it has a 'SpawnServ'
mx.receive(mars().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, this_node(), jupiter().id,
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"));
CAF_MESSAGE("expect announce_proxy message at Mars from Earth to Jupiter");
mx.receive(mars().connection,
basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
CAF_MESSAGE("receive message from jupiter");
self()->receive(
[](const std::string& str) -> std::string {
CAF_CHECK_EQUAL(str, "hello from jupiter!");
return "hello from earth!";
}
);
mpx()->exec_runnable(); // process forwarded message in basp_broker
mock()
.receive(mars().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
default_operation_data, this_node(), jupiter().id,
self()->id(), jupiter().dummy_actor->id(),
std::vector<actor_id>{},
make_message("hello from earth!"));
no_flags, any_vals, default_operation_data, testee->id(),
prx->id(), std::vector<actor_id>{}, msg);
}
CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
CAF_TEST(automatic_connection) {
CAF_TEST_DISABLED(automatic_connection) {
// this tells our BASP broker to enable the automatic connection feature
//anon_send(aut(), ok_atom::value,
// "middleman.enable-automatic-connections", make_message(true));
......@@ -785,34 +679,29 @@ CAF_TEST(automatic_connection) {
connect_node(mars(), ax, self()->id());
//CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id());
check_node_in_tbl(mars());
// TODO: this use case is no longer possible. Nodes are required to open a
// connection first, before sending messages (forwarding has been
// removed).
CAF_MESSAGE("simulate that an actor from jupiter "
"sends a message to us via mars");
mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
jupiter().id, this_node(),
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{},
make_message("hello from jupiter!"))
.receive(mars().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, this_node(), jupiter().id,
any_vals, invalid_actor_id, spawn_serv_atom,
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(mars().connection,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, this_node(), jupiter().id,
any_vals, // actor ID of an actor spawned by the BASP broker
invalid_actor_id,
config_serv_atom,
std::vector<actor_id>{},
make_message(get_atom::value, "basp.default-connectivity-tcp"))
.receive(mars().connection,
basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
std::vector<actor_id>{}, make_message("hello from jupiter!"))
.receive(mars().connection, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals, invalid_actor_id,
spawn_serv_atom, std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(mars().connection, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data,
any_vals, // actor ID of an actor spawned by the BASP broker
invalid_actor_id, config_serv_atom, std::vector<actor_id>{},
make_message(get_atom::value, "basp.default-connectivity-tcp"))
.receive(mars().connection, basp::message_type::announce_proxy, no_flags,
no_payload, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id());
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), mars().id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
......@@ -823,9 +712,8 @@ CAF_TEST(automatic_connection) {
network::address_listing res;
res[network::protocol::ipv4].emplace_back("jupiter");
mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
this_node(), this_node(),
invalid_actor_id, connection_helper_actor},
{basp::message_type::dispatch_message, 0, 0, 0, invalid_actor_id,
connection_helper_actor},
std::vector<actor_id>{},
make_message("basp.default-connectivity-tcp",
make_message(uint16_t{8080}, std::move(res))));
......@@ -839,15 +727,11 @@ CAF_TEST(automatic_connection) {
// send handshake from jupiter
mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().id, none,
jupiter().dummy_actor->id(), invalid_actor_id},
std::string{},
jupiter().dummy_actor->id(),
uint32_t{0})
.receive(jupiter().connection,
basp::message_type::client_handshake, no_flags, 1u,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, invalid_actor_id, std::string{});
std::string{}, jupiter().dummy_actor->id(), uint32_t{0})
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, 1u, no_operation_data, invalid_actor_id,
invalid_actor_id, std::string{});
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
check_node_in_tbl(jupiter());
......@@ -861,13 +745,10 @@ CAF_TEST(automatic_connection) {
);
mpx()->exec_runnable(); // process forwarded message in basp_broker
CAF_MESSAGE("response message must take direct route now");
mock()
.receive(jupiter().connection,
basp::message_type::dispatch_message, no_flags, any_vals,
default_operation_data, this_node(), jupiter().id,
self()->id(), jupiter().dummy_actor->id(),
std::vector<actor_id>{},
make_message("hello from earth!"));
mock().receive(jupiter().connection, basp::message_type::dispatch_message,
no_flags, any_vals, default_operation_data, self()->id(),
jupiter().dummy_actor->id(), std::vector<actor_id>{},
make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
}
......
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