Commit da0b256f authored by Dominik Charousset's avatar 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 3de83045
...@@ -20,6 +20,8 @@ ...@@ -20,6 +20,8 @@
#include <chrono> #include <chrono>
#include <cstddef> #include <cstddef>
#include <string>
#include <vector>
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/string_view.hpp" #include "caf/string_view.hpp"
...@@ -74,7 +76,7 @@ extern const atom_value file_verbosity; ...@@ -74,7 +76,7 @@ extern const atom_value file_verbosity;
namespace middleman { namespace middleman {
extern string_view app_identifier; extern std::vector<std::string> app_identifiers;
extern const atom_value network_backend; extern const atom_value network_backend;
extern const size_t max_consecutive_reads; extern const size_t max_consecutive_reads;
extern const size_t heartbeat_interval; extern const size_t heartbeat_interval;
......
...@@ -109,7 +109,8 @@ actor_system_config::actor_system_config() ...@@ -109,7 +109,8 @@ actor_system_config::actor_system_config()
opt_group{custom_options_, "middleman"} opt_group{custom_options_, "middleman"}
.add<atom_value>("network-backend", .add<atom_value>("network-backend",
"either 'default' or 'asio' (if available)") "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", .add<bool>("enable-automatic-connections",
"enables automatic connection management") "enables automatic connection management")
.add<size_t>("max-consecutive-reads", .add<size_t>("max-consecutive-reads",
......
...@@ -86,7 +86,7 @@ const atom_value file_verbosity = atom("trace"); ...@@ -86,7 +86,7 @@ const atom_value file_verbosity = atom("trace");
namespace middleman { namespace middleman {
string_view app_identifier = ""; std::vector<std::string> app_identifiers{"generic-caf-app"};
const atom_value network_backend = atom("default"); const atom_value network_backend = atom("default");
const size_t max_consecutive_reads = 50; const size_t max_consecutive_reads = 50;
const size_t heartbeat_interval = 0; const size_t heartbeat_interval = 0;
......
...@@ -547,19 +547,18 @@ scheduled_actor::categorize(mailbox_element& x) { ...@@ -547,19 +547,18 @@ scheduled_actor::categorize(mailbox_element& x) {
case make_type_token<atom_value, atom_value, std::string>(): case make_type_token<atom_value, atom_value, std::string>():
if (content.get_as<atom_value>(0) == sys_atom::value if (content.get_as<atom_value>(0) == sys_atom::value
&& content.get_as<atom_value>(1) == get_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); auto& what = content.get_as<std::string>(2);
if (what == "info") { if (what == "info") {
CAF_LOG_DEBUG("reply to 'info' message"); CAF_LOG_DEBUG("reply to 'info' message");
x.sender->enqueue( rp.deliver(ok_atom::value, std::move(what), strong_actor_ptr{ctrl()},
make_mailbox_element(ctrl(), x.mid.response_id(), name());
{}, ok_atom::value, std::move(what),
strong_actor_ptr{ctrl()}, name()),
context());
} else { } else {
x.sender->enqueue( rp.deliver(make_error(sec::unsupported_sys_key));
make_mailbox_element(ctrl(), x.mid.response_id(),
{}, sec::unsupported_sys_key),
context());
} }
return message_category::internal; return message_category::internal;
} }
......
...@@ -46,22 +46,18 @@ struct header { ...@@ -46,22 +46,18 @@ struct header {
uint8_t flags; uint8_t flags;
uint32_t payload_len; uint32_t payload_len;
uint64_t operation_data; uint64_t operation_data;
node_id source_node;
node_id dest_node;
actor_id source_actor; actor_id source_actor;
actor_id dest_actor; actor_id dest_actor;
header(message_type m_operation, uint8_t m_flags, uint32_t m_payload_len, 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, uint64_t m_operation_data, actor_id m_source_actor,
actor_id m_source_actor, actor_id m_dest_actor) actor_id m_dest_actor)
: operation(m_operation), : operation(m_operation),
flags(m_flags), flags(m_flags),
payload_len(m_payload_len), payload_len(m_payload_len),
operation_data(m_operation_data), operation_data(m_operation_data),
source_node(std::move(m_source_node)), source_actor(m_source_actor),
dest_node(std::move(m_dest_node)), dest_actor(m_dest_actor) {
source_actor(m_source_actor),
dest_actor(m_dest_actor) {
// nop // nop
} }
...@@ -85,7 +81,6 @@ typename Inspector::result_type inspect(Inspector& f, header& hdr) { ...@@ -85,7 +81,6 @@ typename Inspector::result_type inspect(Inspector& f, header& hdr) {
meta::omittable(), pad, meta::omittable(), pad,
meta::omittable(), pad, meta::omittable(), pad,
hdr.flags, hdr.payload_len, hdr.operation_data, hdr.flags, hdr.payload_len, hdr.operation_data,
hdr.source_node, hdr.dest_node,
hdr.source_actor, hdr.dest_actor); hdr.source_actor, hdr.dest_actor);
} }
...@@ -113,8 +108,7 @@ inline bool is_heartbeat(const header& hdr) { ...@@ -113,8 +108,7 @@ inline bool is_heartbeat(const header& hdr) {
bool valid(const header& hdr); bool valid(const header& hdr);
/// Size of a BASP header in serialized form /// Size of a BASP header in serialized form
constexpr size_t header_size = node_id::serialized_size * 2 constexpr size_t header_size = sizeof(actor_id) * 2
+ sizeof(actor_id) * 2
+ sizeof(uint32_t) * 2 + sizeof(uint32_t) * 2
+ sizeof(uint64_t); + sizeof(uint64_t);
......
...@@ -90,7 +90,7 @@ public: ...@@ -90,7 +90,7 @@ public:
virtual void learned_new_node_indirectly(const node_id& nid) = 0; virtual void learned_new_node_indirectly(const node_id& nid) = 0;
/// Called if a heartbeat was received from `nid` /// 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. /// Returns the actor namespace associated to this BASP protocol instance.
proxy_registry& proxies() { proxy_registry& proxies() {
...@@ -162,7 +162,7 @@ public: ...@@ -162,7 +162,7 @@ public:
/// Returns `true` if a path to destination existed, `false` otherwise. /// Returns `true` if a path to destination existed, `false` otherwise.
bool dispatch(execution_unit* ctx, const strong_actor_ptr& sender, bool dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
const std::vector<strong_actor_ptr>& forwarding_stack, 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); message_id mid, const message& msg);
/// Returns the actor namespace associated to this BASP protocol instance. /// Returns the actor namespace associated to this BASP protocol instance.
...@@ -200,28 +200,19 @@ public: ...@@ -200,28 +200,19 @@ public:
buffer_type& out_buf, optional<uint16_t> port); buffer_type& out_buf, optional<uint16_t> port);
/// Writes the client handshake to `buf`. /// Writes the client handshake to `buf`.
static void write_client_handshake(execution_unit* ctx, void write_client_handshake(execution_unit* ctx, buffer_type& buf);
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);
/// Writes an `announce_proxy` to `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); const node_id& dest_node, actor_id aid);
/// Writes a `kill_proxy` to `buf`. /// 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 node_id& dest_node, actor_id aid,
const error& rsn); const error& rsn);
/// Writes a `heartbeat` to `buf`. /// Writes a `heartbeat` to `buf`.
void write_heartbeat(execution_unit* ctx, buffer_type& buf, void write_heartbeat(execution_unit* ctx, buffer_type& buf);
const node_id& remote_side);
const node_id& this_node() const { const node_id& this_node() const {
return this_node_; return this_node_;
...@@ -241,6 +232,9 @@ public: ...@@ -241,6 +232,9 @@ public:
std::vector<char>* payload); std::vector<char>* payload);
private: private:
void forward(execution_unit* ctx, const node_id& dest_node, const header& hdr,
std::vector<char>& payload);
routing_table tbl_; routing_table tbl_;
published_actor_map published_actors_; published_actor_map published_actors_;
node_id this_node_; node_id this_node_;
...@@ -252,4 +246,3 @@ private: ...@@ -252,4 +246,3 @@ private:
} // namespace basp } // namespace basp
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -42,31 +42,36 @@ enum class message_type : uint8_t { ...@@ -42,31 +42,36 @@ enum class message_type : uint8_t {
/// ![](client_handshake.png) /// ![](client_handshake.png)
client_handshake = 0x01, 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 /// Transmits a message from `source_node:source_actor` to
/// `dest_node:dest_actor`. /// `dest_node:dest_actor`.
/// ///
/// ![](dispatch_message.png) /// ![](routed_message.png)
dispatch_message = 0x02, routed_message = 0x03,
/// Informs the receiving node that the sending node has created a proxy /// Informs the receiving node that the sending node has created a proxy
/// instance for one of its actors. Causes the receiving node to attach /// instance for one of its actors. Causes the receiving node to attach
/// a functor to the actor that triggers a kill_proxy_instance /// a functor to the actor that triggers a kill_proxy_instance
/// message on termination. /// message on termination.
/// ///
/// ![](announce_proxy_instance.png) /// ![](monitor_message.png)
announce_proxy = 0x03, monitor_message = 0x04,
/// Informs the receiving node that it has a proxy for an actor /// Informs the receiving node that it has a proxy for an actor
/// that has been terminated. /// that has been terminated.
/// ///
/// ![](kill_proxy_instance.png) /// ![](down_message.png)
kill_proxy = 0x04, down_message = 0x05,
/// Used to generate periodic traffic between two nodes /// Used to generate periodic traffic between two nodes
/// in order to detect disconnects. /// in order to detect disconnects.
/// ///
/// ![](heartbeat.png) /// ![](heartbeat.png)
heartbeat = 0x05, heartbeat = 0x06,
}; };
/// @relates message_type /// @relates message_type
...@@ -77,5 +82,3 @@ std::string to_string(message_type); ...@@ -77,5 +82,3 @@ std::string to_string(message_type);
} // namespace basp } // namespace basp
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -26,14 +26,11 @@ namespace basp { ...@@ -26,14 +26,11 @@ namespace basp {
/// @addtogroup BASP /// @addtogroup BASP
/// The current BASP version. Different BASP versions will not /// The current BASP version. Different BASP versions cannot exchange messages.
/// be able to exchange messages. constexpr uint64_t version = 3;
constexpr uint64_t version = 2;
/// @} /// @}
} // namespace basp } // namespace basp
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -92,9 +92,8 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -92,9 +92,8 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::callee // inherited from basp::instance::callee
void flush(connection_handle hdl) override; void flush(connection_handle hdl) override;
void handle_heartbeat(const node_id&) override { // inherited from basp::instance::callee
// nop void handle_heartbeat() override;
}
/// Sets `this_context` by either creating or accessing state for `hdl`. /// Sets `this_context` by either creating or accessing state for `hdl`.
void set_context(connection_handle hdl); void set_context(connection_handle hdl);
......
...@@ -46,13 +46,7 @@ struct connection_helper_state { ...@@ -46,13 +46,7 @@ struct connection_helper_state {
static const char* name; 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, behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b); actor b);
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -103,8 +103,8 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -103,8 +103,8 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
<< CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(nid) << CAF_ARG(aid)
<< CAF_ARG2("hdl", this_context->hdl)); << CAF_ARG2("hdl", this_context->hdl));
// tell remote side we are monitoring this actor now // 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); nid, aid);
instance.flush(*path); instance.flush(*path);
mm->notify<hook::new_remote_actor>(res); mm->notify<hook::new_remote_actor>(res);
return res; return res;
...@@ -156,8 +156,8 @@ void basp_broker_state::send_kill_proxy_instance(const node_id& nid, ...@@ -156,8 +156,8 @@ void basp_broker_state::send_kill_proxy_instance(const node_id& nid,
<< CAF_ARG(nid)); << CAF_ARG(nid));
return; 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); rsn);
instance.flush(*path); instance.flush(*path);
} }
...@@ -324,28 +324,16 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -324,28 +324,16 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
}); });
spawn_servers.emplace(nid, tmp); spawn_servers.emplace(nid, tmp);
using namespace detail; using namespace detail;
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp)); auto tmp_ptr = actor_cast<strong_actor_ptr>(tmp);
auto writer = make_callback([](serializer& sink) -> error { system().registry().put(tmp.id(), tmp_ptr);
auto name_atm = atom("SpawnServ"); std::vector<strong_actor_ptr> stages;
std::vector<actor_id> stages; if (!instance.dispatch(self->context(), tmp_ptr, stages, nid,
auto msg = make_message(sys_atom::value, get_atom::value, "info"); static_cast<uint64_t>(atom("SpawnServ")),
return sink(name_atm, stages, msg); basp::header::named_receiver_flag, make_message_id(),
}); make_message(sys_atom::value, get_atom::value, "info"))) {
auto path = instance.tbl().lookup(nid); CAF_LOG_ERROR("learned_new_node called, but no route to remote node"
if (!path) { << CAF_ARG(nid));
CAF_LOG_ERROR("learned_new_node called, but no route to nid");
return;
} }
// 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, 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) { ...@@ -362,40 +350,23 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
learned_new_node(nid); learned_new_node(nid);
if (!automatic_connections) if (!automatic_connections)
return; return;
// this member function gets only called once, after adding a new // This member function gets only called once, after adding a new indirect
// indirect connection to the routing table; hence, spawning // connection to the routing table; hence, spawning our helper here exactly
// our helper here exactly once and there is no need to track // once and there is no need to track in-flight connection requests.
// 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;
}
using namespace detail; using namespace detail;
auto try_connect = [&](std::string item) { auto tmp = get_or(config(), "middleman.attach-utility-actors", false)
auto tmp = get_or(config(), "middleman.attach-utility-actors", false)
? system().spawn<hidden>(connection_helper, self) ? system().spawn<hidden>(connection_helper, self)
: system().spawn<detached + hidden>(connection_helper, self); : system().spawn<detached + hidden>(connection_helper, self);
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp)); auto sender = actor_cast<strong_actor_ptr>(tmp);
auto writer = make_callback([&item](serializer& sink) -> error { system().registry().put(sender->id(), sender);
auto name_atm = atom("ConfigServ"); std::vector<strong_actor_ptr> fwd_stack;
std::vector<actor_id> stages; if (!instance.dispatch(self->context(), sender, fwd_stack, nid,
auto msg = make_message(get_atom::value, std::move(item)); static_cast<uint64_t>(atom("ConfigServ")),
return sink(name_atm, stages, msg); basp::header::named_receiver_flag, make_message_id(),
}); make_message(get_atom::value,
basp::header hdr{basp::message_type::dispatch_message, "basp.default-connectivity-tcp"))) {
basp::header::named_receiver_flag, CAF_LOG_ERROR("learned_new_node_indirectly called, but no route to nid");
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");
} }
void basp_broker_state::set_context(connection_handle hdl) { void basp_broker_state::set_context(connection_handle hdl) {
...@@ -403,7 +374,7 @@ 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); auto i = ctx.find(hdl);
if (i == ctx.end()) { if (i == ctx.end()) {
CAF_LOG_DEBUG("create new BASP context:" << CAF_ARG(hdl)); 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}; invalid_actor_id, invalid_actor_id};
i = ctx i = ctx
.emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl, .emplace(hdl, basp::endpoint_context{basp::await_header, hdr, hdl,
...@@ -445,6 +416,10 @@ void basp_broker_state::flush(connection_handle hdl) { ...@@ -445,6 +416,10 @@ void basp_broker_state::flush(connection_handle hdl) {
self->flush(hdl); self->flush(hdl);
} }
void basp_broker_state::handle_heartbeat() {
// nop
}
/****************************************************************************** /******************************************************************************
* basp_broker * * basp_broker *
******************************************************************************/ ******************************************************************************/
...@@ -513,8 +488,8 @@ behavior basp_broker::make_behavior() { ...@@ -513,8 +488,8 @@ behavior basp_broker::make_behavior() {
} }
if (src && system().node() == src->node()) if (src && system().node() == src->node())
system().registry().put(src->id(), src); system().registry().put(src->id(), src);
if (!state.instance.dispatch(context(), src, fwd_stack, if (!state.instance.dispatch(context(), src, fwd_stack, dest->node(),
dest, mid, msg) dest->id(), 0, mid, msg)
&& mid.is_request()) { && mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable}; detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable};
srb(src, mid); srb(src, mid);
...@@ -524,32 +499,22 @@ behavior basp_broker::make_behavior() { ...@@ -524,32 +499,22 @@ behavior basp_broker::make_behavior() {
[=](forward_atom, const node_id& dest_node, atom_value dest_name, [=](forward_atom, const node_id& dest_node, atom_value dest_name,
const message& msg) -> result<message> { const message& msg) -> result<message> {
auto cme = current_mailbox_element(); auto cme = current_mailbox_element();
if (cme == nullptr) if (cme == nullptr || cme->sender == nullptr)
return sec::invalid_argument; return sec::invalid_argument;
auto& src = cme->sender; CAF_LOG_TRACE(CAF_ARG2("sender", cme->sender)
CAF_LOG_TRACE(CAF_ARG(src)
<< ", " << CAF_ARG(dest_node) << ", " << CAF_ARG(dest_node)
<< ", " << CAF_ARG(dest_name) << ", " << CAF_ARG(dest_name)
<< ", " << CAF_ARG(msg)); << ", " << CAF_ARG(msg));
if (!src) auto& sender = cme->sender;
return sec::invalid_argument; if (system().node() == sender->node())
auto path = this->state.instance.tbl().lookup(dest_node); system().registry().put(sender->id(), sender);
if (!path) { if (!state.instance.dispatch(context(), sender, cme->stages,
CAF_LOG_ERROR("no route to receiving node"); dest_node, static_cast<uint64_t>(dest_name),
return sec::no_route_to_receiving_node; 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>(); return delegated<message>();
}, },
// received from underlying broker implementation // received from underlying broker implementation
......
...@@ -34,41 +34,6 @@ auto autoconnect_timeout = std::chrono::minutes(10); ...@@ -34,41 +34,6 @@ auto autoconnect_timeout = std::chrono::minutes(10);
const char* connection_helper_state::name = "connection_helper"; const char* connection_helper_state::name = "connection_helper";
behavior datagram_connection_broker(broker* self, uint16_t port,
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, behavior connection_helper(stateful_actor<connection_helper_state>* self,
actor b) { actor b) {
CAF_LOG_TRACE(CAF_ARG(b)); CAF_LOG_TRACE(CAF_ARG(b));
...@@ -101,17 +66,6 @@ behavior connection_helper(stateful_actor<connection_helper_state>* self, ...@@ -101,17 +66,6 @@ behavior connection_helper(stateful_actor<connection_helper_state>* self,
} }
} }
CAF_LOG_INFO("could not connect to node directly"); 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 { } else {
CAF_LOG_INFO("aborted direct connection attempt, unknown item: " CAF_LOG_INFO("aborted direct connection attempt, unknown item: "
<< CAF_ARG(item)); << CAF_ARG(item));
......
...@@ -40,8 +40,6 @@ std::string to_string(const header &hdr) { ...@@ -40,8 +40,6 @@ std::string to_string(const header &hdr) {
<< to_bin(hdr.flags) << ", " << to_bin(hdr.flags) << ", "
<< hdr.payload_len << ", " << hdr.payload_len << ", "
<< hdr.operation_data << ", " << hdr.operation_data << ", "
<< to_string(hdr.source_node) << ", "
<< to_string(hdr.dest_node) << ", "
<< hdr.source_actor << ", " << hdr.source_actor << ", "
<< hdr.dest_actor << hdr.dest_actor
<< "}"; << "}";
...@@ -53,70 +51,45 @@ bool operator==(const header& lhs, const header& rhs) { ...@@ -53,70 +51,45 @@ bool operator==(const header& lhs, const header& rhs) {
&& lhs.flags == rhs.flags && lhs.flags == rhs.flags
&& lhs.payload_len == rhs.payload_len && lhs.payload_len == rhs.payload_len
&& lhs.operation_data == rhs.operation_data && 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.source_actor == rhs.source_actor
&& lhs.dest_actor == rhs.dest_actor; && lhs.dest_actor == rhs.dest_actor;
} }
namespace { namespace {
bool valid(const node_id& val) {
return val != none;
}
template <class T> template <class T>
bool zero(T val) { bool zero(T val) {
return val == 0; return val == 0;
} }
bool server_handshake_valid(const header& hdr) { bool server_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return !zero(hdr.operation_data);
&& zero(hdr.dest_actor)
&& !zero(hdr.operation_data);
} }
bool client_handshake_valid(const header& hdr) { bool client_handshake_valid(const header& hdr) {
return valid(hdr.source_node) return zero(hdr.source_actor) && zero(hdr.dest_actor);
&& hdr.source_node != hdr.dest_node }
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor); bool direct_message_valid(const header& hdr) {
return !zero(hdr.dest_actor) && !zero(hdr.payload_len);
} }
bool dispatch_message_valid(const header& hdr) { bool routed_message_valid(const header& hdr) {
return valid(hdr.dest_node) return !zero(hdr.dest_actor) && !zero(hdr.payload_len);
&& (!zero(hdr.dest_actor) || hdr.has(header::named_receiver_flag))
&& !zero(hdr.payload_len);
} }
bool announce_proxy_instance_valid(const header& hdr) { bool monitor_message_valid(const header& hdr) {
return valid(hdr.source_node) return !zero(hdr.payload_len) && zero(hdr.operation_data);
&& 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 kill_proxy_instance_valid(const header& hdr) { bool down_message_valid(const header& hdr) {
return valid(hdr.source_node) return !zero(hdr.source_actor) && zero(hdr.dest_actor)
&& valid(hdr.dest_node) && !zero(hdr.payload_len) && zero(hdr.operation_data);
&& hdr.source_node != hdr.dest_node
&& !zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& !zero(hdr.payload_len)
&& zero(hdr.operation_data);
} }
bool heartbeat_valid(const header& hdr) { bool heartbeat_valid(const header& hdr) {
return valid(hdr.source_node) return zero(hdr.source_actor) && zero(hdr.dest_actor) && zero(hdr.payload_len)
&& valid(hdr.dest_node) && zero(hdr.operation_data);
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor)
&& zero(hdr.payload_len)
&& zero(hdr.operation_data);
} }
} // namespace <anonymous> } // namespace <anonymous>
...@@ -129,12 +102,14 @@ bool valid(const header& hdr) { ...@@ -129,12 +102,14 @@ bool valid(const header& hdr) {
return server_handshake_valid(hdr); return server_handshake_valid(hdr);
case message_type::client_handshake: case message_type::client_handshake:
return client_handshake_valid(hdr); return client_handshake_valid(hdr);
case message_type::dispatch_message: case message_type::direct_message:
return dispatch_message_valid(hdr); return direct_message_valid(hdr);
case message_type::announce_proxy: case message_type::routed_message:
return announce_proxy_instance_valid(hdr); return routed_message_valid(hdr);
case message_type::kill_proxy: case message_type::monitor_message:
return kill_proxy_instance_valid(hdr); return monitor_message_valid(hdr);
case message_type::down_message:
return down_message_valid(hdr);
case message_type::heartbeat: case message_type::heartbeat:
return heartbeat_valid(hdr); return heartbeat_valid(hdr);
} }
......
This diff is collapsed.
...@@ -29,9 +29,10 @@ namespace { ...@@ -29,9 +29,10 @@ namespace {
const char* message_type_strings[] = { const char* message_type_strings[] = {
"server_handshake", "server_handshake",
"client_handshake", "client_handshake",
"dispatch_message", "direct_message",
"announce_proxy_instance", "routed_message",
"kill_proxy_instance", "proxy_creation",
"proxy_destruction",
"heartbeat" "heartbeat"
}; };
......
This diff is collapsed.
...@@ -18,7 +18,7 @@ ...@@ -18,7 +18,7 @@
#include "caf/config.hpp" #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 "caf/test/dsl.hpp"
#include <vector> #include <vector>
...@@ -142,7 +142,7 @@ behavior linking_actor(event_based_actor* self, const actor& buddy) { ...@@ -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_FIXTURE_SCOPE(dynamic_remote_actor_tests, fixture)
CAF_TEST(identity_semantics_tcp) { CAF_TEST(identity_semantics) {
// server side // server side
auto server = server_side.spawn(make_pong_behavior); auto server = server_side.spawn(make_pong_behavior);
auto port1 = unbox(server_side_mm.publish(server, 0, local_host)); auto port1 = unbox(server_side_mm.publish(server, 0, local_host));
...@@ -158,7 +158,7 @@ CAF_TEST(identity_semantics_tcp) { ...@@ -158,7 +158,7 @@ CAF_TEST(identity_semantics_tcp) {
anon_send_exit(server, exit_reason::user_shutdown); anon_send_exit(server, exit_reason::user_shutdown);
} }
CAF_TEST(ping_pong_tcp) { CAF_TEST(ping_pong) {
// server side // server side
auto port = unbox(server_side_mm.publish( auto port = unbox(server_side_mm.publish(
server_side.spawn(make_pong_behavior), 0, local_host)); server_side.spawn(make_pong_behavior), 0, local_host));
...@@ -167,7 +167,7 @@ CAF_TEST(ping_pong_tcp) { ...@@ -167,7 +167,7 @@ CAF_TEST(ping_pong_tcp) {
client_side.spawn(make_ping_behavior, pong); client_side.spawn(make_ping_behavior, pong);
} }
CAF_TEST(custom_message_type_tcp) { CAF_TEST(custom_message_type) {
// server side // server side
auto port = unbox(server_side_mm.publish( auto port = unbox(server_side_mm.publish(
server_side.spawn(make_sort_behavior), 0, local_host)); server_side.spawn(make_sort_behavior), 0, local_host));
...@@ -176,7 +176,7 @@ CAF_TEST(custom_message_type_tcp) { ...@@ -176,7 +176,7 @@ CAF_TEST(custom_message_type_tcp) {
client_side.spawn(make_sort_requester_behavior, sorter); client_side.spawn(make_sort_requester_behavior, sorter);
} }
CAF_TEST(remote_link_tcp) { CAF_TEST(remote_link) {
// server side // server side
auto port = unbox( auto port = unbox(
server_side_mm.publish(server_side.spawn(fragile_mirror), 0, local_host)); 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