Commit 58369e7d authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Implement BASP v3

BASP v3 streamlines direct node-to-node communication by dropping the
node IDs from the message header. Additionally, BASP now supports
multiple application identifiers and allows connections between nodes as
long as one identifier matches.
parent 68c979ce
......@@ -20,6 +20,8 @@
#include <chrono>
#include <cstddef>
#include <string>
#include <vector>
#include "caf/atom.hpp"
#include "caf/string_view.hpp"
......@@ -74,7 +76,7 @@ extern const atom_value file_verbosity;
namespace middleman {
extern string_view app_identifier;
extern std::vector<std::string> app_identifiers;
extern const atom_value network_backend;
extern const size_t max_consecutive_reads;
extern const size_t heartbeat_interval;
......
......@@ -109,7 +109,8 @@ actor_system_config::actor_system_config()
opt_group{custom_options_, "middleman"}
.add<atom_value>("network-backend",
"either 'default' or 'asio' (if available)")
.add<string>("app-identifier", "application identifier of this node")
.add<std::vector<string>>("app-identifiers",
"valid application identifiers of this node")
.add<bool>("enable-automatic-connections",
"enables automatic connection management")
.add<size_t>("max-consecutive-reads",
......
......@@ -86,7 +86,7 @@ const atom_value file_verbosity = atom("trace");
namespace middleman {
string_view app_identifier = "";
std::vector<std::string> app_identifiers{"generic-caf-app"};
const atom_value network_backend = atom("default");
const size_t max_consecutive_reads = 50;
const size_t heartbeat_interval = 0;
......
......@@ -547,19 +547,18 @@ scheduled_actor::categorize(mailbox_element& x) {
case make_type_token<atom_value, atom_value, std::string>():
if (content.get_as<atom_value>(0) == sys_atom::value
&& content.get_as<atom_value>(1) == get_atom::value) {
auto rp = make_response_promise();
if (!rp.pending()) {
CAF_LOG_WARNING("received anonymous ('get', 'sys', $key) message");
return message_category::internal;
}
auto& what = content.get_as<std::string>(2);
if (what == "info") {
CAF_LOG_DEBUG("reply to 'info' message");
x.sender->enqueue(
make_mailbox_element(ctrl(), x.mid.response_id(),
{}, ok_atom::value, std::move(what),
strong_actor_ptr{ctrl()}, name()),
context());
rp.deliver(ok_atom::value, std::move(what), strong_actor_ptr{ctrl()},
name());
} else {
x.sender->enqueue(
make_mailbox_element(ctrl(), x.mid.response_id(),
{}, sec::unsupported_sys_key),
context());
rp.deliver(make_error(sec::unsupported_sys_key));
}
return message_category::internal;
}
......
......@@ -46,20 +46,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
......@@ -85,7 +81,6 @@ typename Inspector::result_type inspect(Inspector& f, header& hdr) {
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);
}
......@@ -113,8 +108,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);
......
......@@ -90,7 +90,7 @@ public:
virtual void learned_new_node_indirectly(const node_id& nid) = 0;
/// Called if a heartbeat was received from `nid`
virtual void handle_heartbeat(const node_id& nid) = 0;
virtual void handle_heartbeat() = 0;
/// Returns the actor namespace associated to this BASP protocol instance.
proxy_registry& proxies() {
......@@ -162,7 +162,7 @@ public:
/// Returns `true` if a path to destination existed, `false` otherwise.
bool dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
const std::vector<strong_actor_ptr>& forwarding_stack,
const strong_actor_ptr& receiver,
const node_id& dest_node, uint64_t dest_actor, uint8_t flags,
message_id mid, const message& msg);
/// Returns the actor namespace associated to this BASP protocol instance.
......@@ -200,28 +200,19 @@ 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,
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,
void write_monitor_message(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid);
/// Writes a `kill_proxy` to `buf`.
void write_kill_proxy(execution_unit* ctx, buffer_type& buf,
void write_down_message(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid,
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_;
......@@ -241,6 +232,9 @@ public:
std::vector<char>* payload);
private:
void forward(execution_unit* ctx, const node_id& dest_node, const header& hdr,
std::vector<char>& payload);
routing_table tbl_;
published_actor_map published_actors_;
node_id this_node_;
......@@ -252,4 +246,3 @@ private:
} // namespace basp
} // namespace io
} // namespace caf
......@@ -42,31 +42,36 @@ enum class message_type : uint8_t {
/// ![](client_handshake.png)
client_handshake = 0x01,
/// Transmits a direct message from source to destination.
///
/// ![](direct_message.png)
direct_message = 0x02,
/// Transmits a message from `source_node:source_actor` to
/// `dest_node:dest_actor`.
///
/// ![](dispatch_message.png)
dispatch_message = 0x02,
/// ![](routed_message.png)
routed_message = 0x03,
/// Informs the receiving node that the sending node has created a proxy
/// instance for one of its actors. Causes the receiving node to attach
/// a functor to the actor that triggers a kill_proxy_instance
/// message on termination.
///
/// ![](announce_proxy_instance.png)
announce_proxy = 0x03,
/// ![](monitor_message.png)
monitor_message = 0x04,
/// Informs the receiving node that it has a proxy for an actor
/// that has been terminated.
///
/// ![](kill_proxy_instance.png)
kill_proxy = 0x04,
/// ![](down_message.png)
down_message = 0x05,
/// Used to generate periodic traffic between two nodes
/// in order to detect disconnects.
///
/// ![](heartbeat.png)
heartbeat = 0x05,
heartbeat = 0x06,
};
/// @relates message_type
......@@ -77,5 +82,3 @@ std::string to_string(message_type);
} // namespace basp
} // namespace io
} // namespace caf
......@@ -26,14 +26,11 @@ namespace basp {
/// @addtogroup BASP
/// The current BASP version. Different BASP versions will not
/// be able to exchange messages.
constexpr uint64_t version = 2;
/// The current BASP version. Different BASP versions cannot exchange messages.
constexpr uint64_t version = 3;
/// @}
} // namespace basp
} // namespace io
} // namespace caf
......@@ -92,9 +92,8 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::callee
void flush(connection_handle hdl) override;
void handle_heartbeat(const node_id&) override {
// nop
}
// inherited from basp::instance::callee
void handle_heartbeat() override;
/// Sets `this_context` by either creating or accessing state for `hdl`.
void set_context(connection_handle hdl);
......
......@@ -46,13 +46,7 @@ struct connection_helper_state {
static const char* name;
};
behavior datagram_connection_broker(broker* self,
uint16_t port,
network::address_listing addresses,
actor system_broker);
behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b);
} // namespace io
} // namespace caf
......@@ -103,7 +103,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
<< CAF_ARG(nid) << CAF_ARG(aid)
<< CAF_ARG2("hdl", this_context->hdl));
// tell remote side we are monitoring this actor now
instance.write_announce_proxy(self->context(), get_buffer(this_context->hdl),
instance.write_monitor_message(self->context(), get_buffer(this_context->hdl),
nid, aid);
instance.flush(*path);
mm->notify<hook::new_remote_actor>(res);
......@@ -156,7 +156,7 @@ void basp_broker_state::send_kill_proxy_instance(const node_id& nid,
<< CAF_ARG(nid));
return;
}
instance.write_kill_proxy(self->context(), get_buffer(path->hdl), nid, aid,
instance.write_down_message(self->context(), get_buffer(path->hdl), nid, aid,
rsn);
instance.flush(*path);
}
......@@ -324,28 +324,16 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
});
spawn_servers.emplace(nid, tmp);
using namespace detail;
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([](serializer& sink) -> error {
auto name_atm = atom("SpawnServ");
std::vector<actor_id> stages;
auto msg = make_message(sys_atom::value, get_atom::value, "info");
return sink(name_atm, stages, msg);
});
auto path = instance.tbl().lookup(nid);
if (!path) {
CAF_LOG_ERROR("learned_new_node called, but no route to nid");
return;
auto tmp_ptr = actor_cast<strong_actor_ptr>(tmp);
system().registry().put(tmp.id(), tmp_ptr);
std::vector<strong_actor_ptr> stages;
if (!instance.dispatch(self->context(), tmp_ptr, stages, nid,
static_cast<uint64_t>(atom("SpawnServ")),
basp::header::named_receiver_flag, make_message_id(),
make_message(sys_atom::value, get_atom::value, "info"))) {
CAF_LOG_ERROR("learned_new_node called, but no route to remote node"
<< CAF_ARG(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};
// 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),
hdr, &writer);
instance.flush(*path);
}
void basp_broker_state::learned_new_node_directly(const node_id& nid,
......@@ -362,40 +350,23 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
learned_new_node(nid);
if (!automatic_connections)
return;
// 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 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;
}
// 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.
using namespace detail;
auto try_connect = [&](std::string item) {
auto tmp = get_or(config(), "middleman.attach-utility-actors", false)
? system().spawn<hidden>(connection_helper, self)
: system().spawn<detached + hidden>(connection_helper, self);
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([&item](serializer& sink) -> error {
auto name_atm = atom("ConfigServ");
std::vector<actor_id> stages;
auto msg = make_message(get_atom::value, std::move(item));
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};
instance.write(self->context(), get_buffer(path->hdl),
hdr, &writer);
instance.flush(*path);
};
try_connect("basp.default-connectivity-tcp");
auto sender = actor_cast<strong_actor_ptr>(tmp);
system().registry().put(sender->id(), sender);
std::vector<strong_actor_ptr> fwd_stack;
if (!instance.dispatch(self->context(), sender, fwd_stack, nid,
static_cast<uint64_t>(atom("ConfigServ")),
basp::header::named_receiver_flag, make_message_id(),
make_message(get_atom::value,
"basp.default-connectivity-tcp"))) {
CAF_LOG_ERROR("learned_new_node_indirectly called, but no route to nid");
}
}
void basp_broker_state::set_context(connection_handle hdl) {
......@@ -403,7 +374,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,
......@@ -445,6 +416,10 @@ void basp_broker_state::flush(connection_handle hdl) {
self->flush(hdl);
}
void basp_broker_state::handle_heartbeat() {
// nop
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......@@ -513,8 +488,8 @@ behavior basp_broker::make_behavior() {
}
if (src && system().node() == src->node())
system().registry().put(src->id(), src);
if (!state.instance.dispatch(context(), src, fwd_stack,
dest, mid, msg)
if (!state.instance.dispatch(context(), src, fwd_stack, dest->node(),
dest->id(), 0, mid, msg)
&& mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable};
srb(src, mid);
......@@ -524,32 +499,22 @@ behavior basp_broker::make_behavior() {
[=](forward_atom, const node_id& dest_node, atom_value dest_name,
const message& msg) -> result<message> {
auto cme = current_mailbox_element();
if (cme == nullptr)
if (cme == nullptr || cme->sender == nullptr)
return sec::invalid_argument;
auto& src = cme->sender;
CAF_LOG_TRACE(CAF_ARG(src)
CAF_LOG_TRACE(CAF_ARG2("sender", cme->sender)
<< ", " << CAF_ARG(dest_node)
<< ", " << CAF_ARG(dest_name)
<< ", " << CAF_ARG(msg));
if (!src)
return sec::invalid_argument;
auto path = this->state.instance.tbl().lookup(dest_node);
if (!path) {
CAF_LOG_ERROR("no route to receiving node");
return sec::no_route_to_receiving_node;
auto& sender = cme->sender;
if (system().node() == sender->node())
system().registry().put(sender->id(), sender);
if (!state.instance.dispatch(context(), sender, cme->stages,
dest_node, static_cast<uint64_t>(dest_name),
basp::header::named_receiver_flag, cme->mid,
msg)) {
detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable};
srb(sender, cme->mid);
}
if (system().node() == src->node())
system().registry().put(src->id(), src);
auto writer = make_callback([&](serializer& sink) -> error {
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};
state.instance.write(context(), state.get_buffer(path->hdl),
hdr, &writer);
state.instance.flush(*path);
return delegated<message>();
},
// received from underlying broker implementation
......
......@@ -34,41 +34,6 @@ auto autoconnect_timeout = std::chrono::minutes(10);
const char* connection_helper_state::name = "connection_helper";
behavior datagram_connection_broker(broker* self, uint16_t port,
network::address_listing addresses,
actor system_broker) {
auto& mx = self->system().middleman().backend();
auto& this_node = self->system().node();
auto app_id = get_or(self->config(), "middleman.app-identifier",
defaults::middleman::app_identifier);
for (auto& kvp : addresses) {
for (auto& addr : kvp.second) {
auto eptr = mx.new_remote_udp_endpoint(addr, port);
if (eptr) {
auto hdl = (*eptr)->hdl();
self->add_datagram_servant(std::move(*eptr));
basp::instance::write_client_handshake(self->context(),
self->wr_buf(hdl),
none, this_node,
app_id);
}
}
}
return {
[=](new_datagram_msg& msg) {
auto hdl = msg.handle;
self->send(system_broker, std::move(msg), self->take(hdl), port);
self->quit();
},
after(autoconnect_timeout) >> [=]() {
CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then
CAF_LOG_INFO("aborted direct connection attempt after 10min");
self->quit(exit_reason::user_shutdown);
}
};
}
behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b) {
CAF_LOG_TRACE(CAF_ARG(b));
......@@ -101,17 +66,6 @@ behavior connection_helper(stateful_actor<connection_helper_state>* self,
}
}
CAF_LOG_INFO("could not connect to node directly");
} else if (item == "basp.default-connectivity-udp") {
auto& sys = self->system();
// create new broker to try addresses for communication via UDP
if (get_or(sys.config(), "middleman.attach-utility-actors", false))
self->system().middleman().spawn_broker<hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
else
self->system().middleman().spawn_broker<detached + hidden>(
datagram_connection_broker, port, std::move(addresses), b
);
} else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item));
......
......@@ -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,69 +51,44 @@ 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);
bool direct_message_valid(const header& hdr) {
return !zero(hdr.dest_actor) && !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);
bool routed_message_valid(const header& hdr) {
return !zero(hdr.dest_actor) && !zero(hdr.payload_len);
}
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);
bool monitor_message_valid(const header& hdr) {
return !zero(hdr.payload_len) && zero(hdr.operation_data);
}
bool down_message_valid(const header& hdr) {
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)
return zero(hdr.source_actor) && zero(hdr.dest_actor) && zero(hdr.payload_len)
&& zero(hdr.operation_data);
}
......@@ -129,12 +102,14 @@ bool valid(const header& hdr) {
return server_handshake_valid(hdr);
case message_type::client_handshake:
return client_handshake_valid(hdr);
case message_type::dispatch_message:
return dispatch_message_valid(hdr);
case message_type::announce_proxy:
return announce_proxy_instance_valid(hdr);
case message_type::kill_proxy:
return kill_proxy_instance_valid(hdr);
case message_type::direct_message:
return direct_message_valid(hdr);
case message_type::routed_message:
return routed_message_valid(hdr);
case message_type::monitor_message:
return monitor_message_valid(hdr);
case message_type::down_message:
return down_message_valid(hdr);
case message_type::heartbeat:
return heartbeat_valid(hdr);
}
......
......@@ -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);
}
}
......@@ -194,26 +164,34 @@ size_t instance::remove_published_actor(const actor_addr& whom,
bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
const std::vector<strong_actor_ptr>& forwarding_stack,
const strong_actor_ptr& receiver, message_id mid,
const message& msg) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(receiver)
<< CAF_ARG(mid) << CAF_ARG(msg));
CAF_ASSERT(receiver && system().node() != receiver->node());
auto path = lookup(receiver->node());
const node_id& dest_node, uint64_t dest_actor,
uint8_t flags, message_id mid, const message& msg) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(dest_node) << CAF_ARG(mid)
<< CAF_ARG(msg));
CAF_ASSERT(dest_node && this_node_ != dest_node);
auto path = lookup(dest_node);
if (!path) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg);
//notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false;
}
auto& source_node = sender ? sender->node() : this_node_;
if (dest_node == path->next_hop && source_node == this_node_) {
header hdr{message_type::direct_message, flags, 0, mid.integer_value(),
sender ? sender->id() : invalid_actor_id, dest_actor};
auto writer = make_callback([&](serializer& sink) -> error {
return sink(forwarding_stack, msg);
});
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer);
} else {
header hdr{message_type::routed_message, flags, 0, mid.integer_value(),
sender ? sender->id() : invalid_actor_id, dest_actor};
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<std::vector<strong_actor_ptr>&>(forwarding_stack),
const_cast<message&>(msg));
return sink(source_node, dest_node, forwarding_stack, 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);
notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg);
//notify<hook::message_sent>(sender, path->next_hop, receiver, mid, msg);
return true;
}
......@@ -253,230 +231,278 @@ 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",
defaults::middleman::app_identifier);
auto e = sink(id);
if (e)
return e;
if (pa != nullptr) {
auto i = pa->first ? pa->first->id() : invalid_actor_id;
return sink(i, pa->second);
}
auto app_ids = get_or(callee_.config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers);
auto aid = invalid_actor_id;
std::set<std::string> tmp;
return sink(aid, tmp);
auto iface = std::set<std::string>{};
if (pa != nullptr && pa->first != nullptr) {
aid = pa->first->id();
iface = pa->second;
}
return sink(this_node_, app_ids, aid, iface);
});
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};
invalid_actor_id, invalid_actor_id};
write(ctx, out_buf, hdr, &writer);
}
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));
void instance::write_client_handshake(execution_unit* ctx, buffer_type& buf) {
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<std::string&>(app_identifier));
return sink(this_node_);
});
header hdr{message_type::client_handshake, 0, 0, 0,
this_node, remote_side, invalid_actor_id, invalid_actor_id};
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_,
get_or(callee_.config(), "middleman.app-identifier",
defaults::middleman::app_identifier));
}
void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf,
void instance::write_monitor_message(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};
write(ctx, buf, hdr);
auto writer = make_callback([&](serializer& sink) -> error {
return sink(this_node_, dest_node);
});
header hdr{message_type::monitor_message, 0, 0, 0, invalid_actor_id, aid};
write(ctx, buf, hdr, &writer);
}
void instance::write_kill_proxy(execution_unit* ctx, buffer_type& buf,
void instance::write_down_message(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid,
const error& rsn) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid) << CAF_ARG(rsn));
auto writer = make_callback([&](serializer& sink) -> error {
return sink(const_cast<error&>(rsn));
return sink(this_node_, dest_node, rsn);
});
header hdr{message_type::kill_proxy, 0, 0, 0,
this_node_, dest_node, aid, invalid_actor_id};
header hdr{message_type::down_message, 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);
}
bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
std::vector<char>* payload) {
// function object for checking payload validity
auto payload_valid = [&]() -> bool {
return payload != nullptr && payload->size() == hdr.payload_len;
};
// handle message to ourselves
CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(hdr));
// Check payload validity.
if (payload == nullptr) {
if (hdr.payload_len != 0) {
CAF_LOG_WARNING("invalid payload");
return false;
}
} else if (hdr.payload_len != payload->size()) {
CAF_LOG_WARNING("invalid payload");
return false;
}
// Dispatch by message type.
switch (hdr.operation) {
case message_type::server_handshake: {
// Deserialize payload.
binary_deserializer bd{ctx, *payload};
node_id source_node;
std::vector<std::string> app_ids;
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
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");
if (auto err = bd(source_node, app_ids, aid, sigs)) {
CAF_LOG_WARNING("unable to deserialize payload of server handshake:"
<< ctx->system().render(err));
return false;
}
e = bd(aid, sigs);
if (e)
// Check the application ID.
auto whitelist = get_or(callee_.config(), "middleman.app-identifiers",
defaults::middleman::app_identifiers);
auto i = std::find_first_of(app_ids.begin(), app_ids.end(),
whitelist.begin(), whitelist.end());
if (i == app_ids.end()) {
CAF_LOG_WARNING("refuse to connect to server due to app ID mismatch:"
<< CAF_ARG(app_ids) << CAF_ARG(whitelist));
return false;
}
// close self connection after handshake is done
if (hdr.source_node == this_node_) {
// Close connection to ourselves immediately after sending client HS.
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 this connection if we already have a direct connection.
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG("close redundant direct 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 direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node);
auto was_indirect = tbl_.erase_indirect(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, was_indirect);
callee_.finalize_handshake(source_node, aid, sigs);
flush(*path);
break;
}
case message_type::client_handshake: {
if (!payload_valid()) {
CAF_LOG_ERROR("fail to receive the app identifier");
return false;
} else {
// Deserialize payload.
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
node_id source_node;
if (auto err = bd(source_node)) {
CAF_LOG_WARNING("unable to deserialize payload of client handshake:"
<< ctx->system().render(err));
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)) {
CAF_LOG_DEBUG("received second client handshake:"
<< CAF_ARG(hdr.source_node));
// Drop repeated handshakes.
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG("received repeated client handshake:"
<< 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 direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node);
auto was_indirect = tbl_.erase_indirect(source_node);
callee_.learned_new_node_directly(source_node, was_indirect);
break;
}
case message_type::dispatch_message: {
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);
case message_type::direct_message: {
// 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 (auto err = bd(forwarding_stack, msg)) {
CAF_LOG_WARNING("unable to deserialize payload of direct message:"
<< ctx->system().render(err));
return false;
}
auto e = bd(forwarding_stack, msg);
if (e)
return false;
CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg));
// Dispatch message to callee_.
auto source_node = tbl_.lookup_direct(hdl);
if (hdr.has(header::named_receiver_flag))
callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name,
make_message_id(hdr.operation_data),
forwarding_stack, msg);
callee_.deliver(source_node, hdr.source_actor,
static_cast<atom_value>(hdr.dest_actor),
make_message_id(hdr.operation_data), forwarding_stack,
msg);
else
callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor,
make_message_id(hdr.operation_data),
forwarding_stack, msg);
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::routed_message: {
// Deserialize payload.
binary_deserializer bd{ctx, *payload};
node_id source_node;
node_id dest_node;
if (auto err = bd(source_node, dest_node)) {
CAF_LOG_WARNING(
"unable to deserialize source and destination for routed message:"
<< ctx->system().render(err));
return false;
}
if (dest_node != this_node_) {
forward(ctx, dest_node, hdr, *payload);
return true;
}
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
if (auto err = bd(forwarding_stack, msg)) {
CAF_LOG_WARNING("unable to deserialize payload of routed message:"
<< ctx->system().render(err));
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 (source_node != none && source_node != this_node_
&& last_hop != source_node && !tbl_.lookup_direct(source_node)
&& tbl_.add_indirect(last_hop, source_node))
callee_.learned_new_node_indirectly(source_node);
if (hdr.has(header::named_receiver_flag))
callee_.deliver(source_node, hdr.source_actor,
static_cast<atom_value>(hdr.dest_actor),
make_message_id(hdr.operation_data), forwarding_stack,
msg);
else
callee_.deliver(source_node, hdr.source_actor, hdr.dest_actor,
make_message_id(hdr.operation_data), forwarding_stack,
msg);
break;
case message_type::kill_proxy: {
if (!payload_valid())
}
case message_type::monitor_message: {
// Deserialize payload.
binary_deserializer bd{ctx, *payload};
node_id source_node;
node_id dest_node;
if (auto err = bd(source_node, dest_node)) {
CAF_LOG_WARNING("unable to deserialize payload of monitor message:"
<< ctx->system().render(err));
return false;
}
if (dest_node == this_node_)
callee_.proxy_announced(source_node, hdr.dest_actor);
else
forward(ctx, dest_node, hdr, *payload);
break;
}
case message_type::down_message: {
// Deserialize payload.
binary_deserializer bd{ctx, *payload};
node_id source_node;
node_id dest_node;
error fail_state;
auto e = bd(fail_state);
if (e)
if (auto err = bd(source_node, dest_node, fail_state)) {
CAF_LOG_WARNING("unable to deserialize payload of down message:"
<< ctx->system().render(err));
return false;
callee_.proxies().erase(hdr.source_node, hdr.source_actor,
}
if (dest_node == this_node_)
callee_.proxies().erase(source_node, hdr.source_actor,
std::move(fail_state));
else
forward(ctx, dest_node, hdr, *payload);
break;
}
case message_type::heartbeat: {
CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node));
callee_.handle_heartbeat(hdr.source_node);
CAF_LOG_TRACE("received heartbeat");
callee_.handle_heartbeat();
break;
}
default:
default: {
CAF_LOG_ERROR("invalid operation");
return false;
}
}
return true;
}
void instance::forward(execution_unit* ctx, const node_id& dest_node,
const header& hdr, std::vector<char>& payload) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(hdr) << CAF_ARG(payload));
auto path = lookup(dest_node);
if (path) {
binary_serializer bs{ctx, callee_.get_buffer(path->hdl)};
if (auto err = bs(hdr)) {
CAF_LOG_ERROR("unable to serialize BASP header");
return;
}
if (auto err = bs.apply_raw(payload.size(), payload.data())) {
CAF_LOG_ERROR("unable to serialize raw payload");
return;
}
flush(*path);
notify<hook::message_forwarded>(hdr, &payload);
} else {
CAF_LOG_WARNING("cannot forward message, no route to destination");
notify<hook::message_forwarding_failed>(hdr, &payload);
}
}
} // namespace basp
} // namespace io
} // namespace caf
......@@ -29,9 +29,10 @@ namespace {
const char* message_type_strings[] = {
"server_handshake",
"client_handshake",
"dispatch_message",
"announce_proxy_instance",
"kill_proxy_instance",
"direct_message",
"routed_message",
"proxy_creation",
"proxy_destruction",
"heartbeat"
};
......
......@@ -274,25 +274,20 @@ 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,
{basp::message_type::client_handshake, 0, 0, 0, invalid_actor_id,
invalid_actor_id},
n.id)
.receive(hdl, basp::message_type::server_handshake, no_flags, any_vals,
basp::version, invalid_actor_id, invalid_actor_id, this_node(),
defaults::middleman::app_identifiers, 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,
.receive(hdl, basp::message_type::direct_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>{},
default_operation_data, any_vals,
static_cast<uint64_t>(spawn_serv_atom),
std::vector<strong_actor_ptr>{},
make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the
// routing table upon receiving client handshakes
......@@ -318,7 +313,7 @@ public:
buffer buf;
std::tie(hdr, buf) = read_from_out_buf(hdl);
CAF_MESSAGE("dispatch output buffer for connection " << hdl.id());
CAF_REQUIRE(hdr.operation == basp::message_type::dispatch_message);
CAF_REQUIRE(hdr.operation == basp::message_type::direct_message);
binary_deserializer source{mpx_, buf};
std::vector<strong_actor_ptr> stages;
message msg;
......@@ -352,8 +347,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) {
......@@ -383,11 +376,15 @@ public:
ob.erase(ob.begin(), ob.begin() + basp::header_size);
}
CAF_CHECK_EQUAL(operation, hdr.operation);
if (hdr.operation == basp::message_type::direct_message) {
binary_deserializer source{this_->mpx(), payload};
std::vector<strong_actor_ptr> fwd_stack;
message msg;
source(fwd_stack, msg);
}
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());
......@@ -475,9 +472,7 @@ CAF_TEST(empty_server_handshake) {
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,
invalid_actor_id, invalid_actor_id};
basp::version, invalid_actor_id, invalid_actor_id};
CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK_EQUAL(to_string(hdr), to_string(expected));
......@@ -490,13 +485,20 @@ CAF_TEST(non_empty_server_handshake) {
instance().add_published_actor(4242, actor_cast<strong_actor_ptr>(self()),
{"caf::replies_to<@u16>::with<@u16>"});
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{},
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"});
CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf));
basp::header hdr;
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, invalid_actor_id, invalid_actor_id};
CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK_EQUAL(to_string(hdr), to_string(expected));
buffer expected_payload;
binary_serializer bd{nullptr, expected_payload};
bd(instance().this_node(), defaults::middleman::app_identifiers, self()->id(),
set<string>{"caf::replies_to<@u16>::with<@u16>"});
CAF_CHECK_EQUAL(hexstr(payload), hexstr(expected_payload));
}
CAF_TEST(remote_address_and_port) {
......@@ -524,14 +526,12 @@ CAF_TEST(client_handshake_and_dispatch) {
connect_node(jupiter());
// 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());
{basp::message_type::direct_message, 0, 0, 0,
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_addr>{}, make_message(1, 2, 3))
.receive(jupiter().connection, basp::message_type::monitor_message,
no_flags, any_vals, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id(), this_node(), jupiter().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
......@@ -562,15 +562,13 @@ CAF_TEST(message_forwarding) {
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,
{basp::message_type::routed_message, 0, 0, default_operation_data,
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);
jupiter().id, mars().id, std::vector<strong_actor_ptr>{}, msg)
.receive(mars().connection, basp::message_type::routed_message, no_flags,
any_vals, default_operation_data, invalid_actor_id,
mars().dummy_actor->id(), jupiter().id, mars().id,
std::vector<strong_actor_ptr>{}, msg);
}
CAF_TEST(publish_and_connect) {
......@@ -600,27 +598,21 @@ 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,
invalid_actor_id, invalid_actor_id},
jupiter().id, defaults::middleman::app_identifiers,
jupiter().dummy_actor->id(), std::set<std::string>{})
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, any_vals, no_operation_data, invalid_actor_id,
invalid_actor_id, this_node())
.receive(jupiter().connection, basp::message_type::direct_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>{},
default_operation_data, any_vals,
static_cast<uint64_t>(spawn_serv_atom),
std::vector<strong_actor_ptr>{},
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());
.receive(jupiter().connection, basp::message_type::monitor_message,
no_flags, any_vals, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id(), this_node(), jupiter().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,21 +637,16 @@ 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>{},
mock().receive(jupiter().connection, basp::message_type::direct_message,
no_flags, any_vals, default_operation_data, invalid_actor_id,
jupiter().dummy_actor->id(), std::vector<strong_actor_ptr>{},
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(),
{basp::message_type::direct_message, 0, 0, 0,
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{},
make_message("hi there!"));
std::vector<strong_actor_ptr>{}, make_message("hi there!"));
self()->receive(
[&](const string& str) {
CAF_CHECK_EQUAL(to_string(self()->current_sender()), to_string(result));
......@@ -680,11 +667,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::monitor_message,
no_flags, any_vals, no_operation_data, invalid_actor_id,
prx->id(), this_node(), prx->node());
CAF_CHECK_EQUAL(prx->node(), jupiter().id);
CAF_CHECK_EQUAL(prx->id(), jupiter().dummy_actor->id());
auto testee = sys.spawn(testee_impl);
......@@ -692,20 +677,16 @@ 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::direct_message, 0, 0, 0, prx->id(), testee->id()},
std::vector<strong_actor_ptr>{}, 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);
mock().receive(jupiter().connection, basp::message_type::direct_message,
no_flags, any_vals, default_operation_data, testee->id(),
prx->id(), std::vector<strong_actor_ptr>{}, msg);
}
CAF_TEST(indirect_connections) {
......@@ -721,26 +702,22 @@ CAF_TEST(indirect_connections) {
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(),
{basp::message_type::routed_message, 0, 0, 0,
jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{},
jupiter().id, this_node(), 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,
mx.receive(mars().connection, basp::message_type::routed_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>{},
default_operation_data, any_vals,
static_cast<uint64_t>(spawn_serv_atom), this_node(), jupiter().id,
std::vector<strong_actor_ptr>{},
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());
mx.receive(mars().connection, basp::message_type::monitor_message, no_flags,
any_vals, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id(), this_node(), jupiter().id);
CAF_MESSAGE("receive message from jupiter");
self()->receive(
[](const std::string& str) -> std::string {
......@@ -749,12 +726,10 @@ CAF_TEST(indirect_connections) {
}
);
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>{},
mock().receive(mars().connection, basp::message_type::routed_message,
no_flags, any_vals, default_operation_data, self()->id(),
jupiter().dummy_actor->id(), this_node(), jupiter().id,
std::vector<strong_actor_ptr>{},
make_message("hello from earth!"));
}
......@@ -763,91 +738,78 @@ CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
CAF_TEST(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));
//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)
// Utility helper for verifying routing tables.
auto check_node_in_tbl = [&](node& n) {
auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl);
};
// Setup.
mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax);
publish(self(), 4242);
mpx()->flush_runnables(); // process publish message in basp_broker
// Process publish message in basp_broker.
mpx()->flush_runnables();
CAF_MESSAGE("connect to mars");
connect_node(mars(), ax, self()->id());
//CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id());
check_node_in_tbl(mars());
CAF_MESSAGE("simulate that an actor from jupiter "
"sends a message to us via mars");
CAF_MESSAGE("simulate that a message from jupiter travels over 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>{},
{basp::message_type::routed_message, 0, 0,
make_message_id().integer_value(), jupiter().dummy_actor->id(),
self()->id()},
jupiter().id, this_node(), std::vector<strong_actor_ptr>{},
make_message("hello from jupiter!"))
.receive(mars().connection,
basp::message_type::dispatch_message,
.receive(mars().connection, basp::message_type::routed_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>{},
default_operation_data, any_vals,
static_cast<uint64_t>(spawn_serv_atom), this_node(), jupiter().id,
std::vector<strong_actor_ptr>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(mars().connection,
basp::message_type::dispatch_message,
.receive(mars().connection, basp::message_type::routed_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, this_node(), jupiter().id,
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>{},
static_cast<uint64_t>(config_serv_atom), this_node(), jupiter().id,
std::vector<strong_actor_ptr>{},
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());
.receive(mars().connection, basp::message_type::monitor_message, no_flags,
any_vals, no_operation_data, invalid_actor_id,
jupiter().dummy_actor->id(), this_node(), jupiter().id);
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), mars().id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
auto connection_helper_actor = sys.latest_actor_id();
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
// create a dummy config server and respond to the name lookup
// Create a dummy config server and respond to the name lookup.
CAF_MESSAGE("receive ConfigServ of jupiter");
network::address_listing res;
res[network::protocol::ipv4].emplace_back("jupiter");
mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0,
this_node(), this_node(),
invalid_actor_id, connection_helper_actor},
std::vector<actor_id>{},
{basp::message_type::routed_message, 0, 0,
make_message_id().integer_value(), invalid_actor_id,
connection_helper_actor},
this_node(), this_node(), std::vector<strong_actor_ptr>{},
make_message("basp.default-connectivity-tcp",
make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and
// send the scribe handle over to the BASP broker
// Our connection helper should now connect to jupiter and send the scribe
// handle over to the BASP broker.
while (mpx()->has_pending_scribe("jupiter", 8080)) {
sched.run();
mpx()->flush_runnables();
}
CAF_REQUIRE(mpx()->output_buffer(mars().connection).empty());
// send handshake from jupiter
// 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{});
{basp::message_type::server_handshake, no_flags, 0, basp::version,
invalid_actor_id, invalid_actor_id},
jupiter().id, defaults::middleman::app_identifiers,
jupiter().dummy_actor->id(), std::set<std::string>{})
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, any_vals, no_operation_data, invalid_actor_id,
invalid_actor_id, this_node());
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
check_node_in_tbl(jupiter());
......@@ -861,12 +823,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,
mock().receive(jupiter().connection, basp::message_type::direct_message,
no_flags, any_vals, make_message_id().integer_value(),
self()->id(), jupiter().dummy_actor->id(),
std::vector<actor_id>{},
std::vector<strong_actor_ptr>{},
make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
}
......
......@@ -18,7 +18,7 @@
#include "caf/config.hpp"
#define CAF_SUITE io_dynamic_remote_actor_tcp
#define CAF_SUITE io_dynamic_remote_actor
#include "caf/test/dsl.hpp"
#include <vector>
......@@ -142,7 +142,7 @@ behavior linking_actor(event_based_actor* self, const actor& buddy) {
CAF_TEST_FIXTURE_SCOPE(dynamic_remote_actor_tests, fixture)
CAF_TEST(identity_semantics_tcp) {
CAF_TEST(identity_semantics) {
// server side
auto server = server_side.spawn(make_pong_behavior);
auto port1 = unbox(server_side_mm.publish(server, 0, local_host));
......@@ -158,7 +158,7 @@ CAF_TEST(identity_semantics_tcp) {
anon_send_exit(server, exit_reason::user_shutdown);
}
CAF_TEST(ping_pong_tcp) {
CAF_TEST(ping_pong) {
// server side
auto port = unbox(server_side_mm.publish(
server_side.spawn(make_pong_behavior), 0, local_host));
......@@ -167,7 +167,7 @@ CAF_TEST(ping_pong_tcp) {
client_side.spawn(make_ping_behavior, pong);
}
CAF_TEST(custom_message_type_tcp) {
CAF_TEST(custom_message_type) {
// server side
auto port = unbox(server_side_mm.publish(
server_side.spawn(make_sort_behavior), 0, local_host));
......@@ -176,7 +176,7 @@ CAF_TEST(custom_message_type_tcp) {
client_side.spawn(make_sort_requester_behavior, sorter);
}
CAF_TEST(remote_link_tcp) {
CAF_TEST(remote_link) {
// server side
auto port = unbox(
server_side_mm.publish(server_side.spawn(fragile_mirror), 0, local_host));
......
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