Unverified Commit e9946674 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #816

Implement BASP v3
parents 792f39c2 06b57043
doc/png/basp_header.png

219 KB | W: | H:

doc/png/basp_header.png

27.7 KB | W: | H:

doc/png/basp_header.png
doc/png/basp_header.png
doc/png/basp_header.png
doc/png/basp_header.png
  • 2-up
  • Swipe
  • Onion skin
doc/png/client_handshake.png

213 KB | W: | H:

doc/png/client_handshake.png

25.7 KB | W: | H:

doc/png/client_handshake.png
doc/png/client_handshake.png
doc/png/client_handshake.png
doc/png/client_handshake.png
  • 2-up
  • Swipe
  • Onion skin
doc/png/heartbeat.png

198 KB | W: | H:

doc/png/heartbeat.png

22.3 KB | W: | H:

doc/png/heartbeat.png
doc/png/heartbeat.png
doc/png/heartbeat.png
doc/png/heartbeat.png
  • 2-up
  • Swipe
  • Onion skin
doc/png/server_handshake.png

243 KB | W: | H:

doc/png/server_handshake.png

36.2 KB | W: | H:

doc/png/server_handshake.png
doc/png/server_handshake.png
doc/png/server_handshake.png
doc/png/server_handshake.png
  • 2-up
  • Swipe
  • Onion skin
...@@ -339,6 +339,10 @@ private: ...@@ -339,6 +339,10 @@ private:
static std::string render_pec(uint8_t, atom_value, const message&); static std::string render_pec(uint8_t, atom_value, const message&);
void extract_config_file_path(string_list& args); void extract_config_file_path(string_list& args);
/// Adjusts the content of the configuration, e.g., for ensuring backwards
/// compatibility with older options.
error adjust_content();
}; };
/// @private /// @private
......
...@@ -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,9 @@ actor_system_config::actor_system_config() ...@@ -109,7 +109,9 @@ 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<string>("app-identifier", "DEPRECATED: use app-identifiers instead")
.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",
...@@ -275,7 +277,7 @@ error actor_system_config::parse(string_list args, std::istream& ini) { ...@@ -275,7 +277,7 @@ error actor_system_config::parse(string_list args, std::istream& ini) {
std::cout << std::flush; std::cout << std::flush;
cli_helptext_printed = true; cli_helptext_printed = true;
} }
return none; return adjust_content();
} }
error actor_system_config::parse(string_list args, const char* ini_file_cstr) { error actor_system_config::parse(string_list args, const char* ini_file_cstr) {
...@@ -302,6 +304,11 @@ actor_system_config::add_error_category(atom_value x, error_renderer y) { ...@@ -302,6 +304,11 @@ actor_system_config::add_error_category(atom_value x, error_renderer y) {
actor_system_config& actor_system_config::set_impl(string_view name, actor_system_config& actor_system_config::set_impl(string_view name,
config_value value) { config_value value) {
if (name == "middleman.app-identifier") {
// TODO: Print a warning with 0.18 and remove this code with 0.19.
value.convert_to_list();
return set_impl("middleman.app-identifiers", std::move(value));
}
auto opt = custom_options_.qualified_name_lookup(name); auto opt = custom_options_.qualified_name_lookup(name);
if (opt != nullptr && opt->check(value) == none) { if (opt != nullptr && opt->check(value) == none) {
opt->store(value); opt->store(value);
...@@ -377,6 +384,25 @@ void actor_system_config::extract_config_file_path(string_list& args) { ...@@ -377,6 +384,25 @@ void actor_system_config::extract_config_file_path(string_list& args) {
args.erase(i); args.erase(i);
} }
error actor_system_config::adjust_content() {
// TODO: Print a warning to STDERR if 'app-identifier' is present with 0.18
// and remove this code with 0.19.
auto i = content.find("middleman");
if (i != content.end()) {
if (auto mm = get_if<settings>(&i->second)) {
auto j = mm->find("app-identifier");
if (j != mm->end()) {
if (!mm->contains("app-identifiers")) {
j->second.convert_to_list();
mm->emplace("app-identifiers", std::move(j->second));
}
mm->container().erase(j);
}
}
}
return none;
}
const settings& content(const actor_system_config& cfg) { const settings& content(const actor_system_config& cfg) {
return cfg.content; return cfg.content;
} }
......
...@@ -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;
} }
......
...@@ -67,7 +67,7 @@ ...@@ -67,7 +67,7 @@
/// Whenever a node learns the address of a remotely running actor, it /// Whenever a node learns the address of a remotely running actor, it
/// creates Ma local proxy instance representing this actor and sends an /// creates Ma local proxy instance representing this actor and sends an
/// `announce_proxy_instance` to the node hosting the actor. Whenever an actor /// `announce_proxy_instance` to the node hosting the actor. Whenever an actor
/// terminates, the hosting node sends `kill_proxy_instance` messages to all /// terminates, the hosting node sends `down_message` messages to all
/// nodes that have a proxy for this actor. This enables network-transparent /// nodes that have a proxy for this actor. This enables network-transparent
/// actor monitoring. There are two possible ways addresses can be learned: /// actor monitoring. There are two possible ways addresses can be learned:
/// ///
......
...@@ -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,35 @@ enum class message_type : uint8_t { ...@@ -42,31 +42,35 @@ 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 down_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 +81,3 @@ std::string to_string(message_type); ...@@ -77,5 +81,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. Note: BASP is not backwards compatible.
/// 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);
...@@ -139,10 +138,10 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -139,10 +138,10 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// keeps a list of nodes that monitor a particular local actor // keeps a list of nodes that monitor a particular local actor
monitored_actor_map monitored_actors; monitored_actor_map monitored_actors;
// sends a kill_proxy message to a remote node // sends a basp::down_message message to a remote node
void send_kill_proxy_instance(const node_id& nid, actor_id aid, error err); void send_basp_down_message(const node_id& nid, actor_id aid, error err);
// sends kill_proxy_instance message to all nodes monitoring the terminated // sends basp::down_message to all nodes monitoring the terminated
// actor // actor
void handle_down_msg(down_msg&); void handle_down_msg(down_msg&);
......
...@@ -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
...@@ -82,7 +82,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -82,7 +82,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
return nullptr; return nullptr;
} }
// create proxy and add functor that will be called if we // create proxy and add functor that will be called if we
// receive a kill_proxy_instance message // receive a basp::down_message
auto mm = &system().middleman(); auto mm = &system().middleman();
actor_config cfg; actor_config cfg;
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>( auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>(
...@@ -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;
...@@ -147,8 +147,8 @@ void basp_broker_state::purge_state(const node_id& nid) { ...@@ -147,8 +147,8 @@ void basp_broker_state::purge_state(const node_id& nid) {
kvp.second.erase(nid); kvp.second.erase(nid);
} }
void basp_broker_state::send_kill_proxy_instance(const node_id& nid, void basp_broker_state::send_basp_down_message(const node_id& nid, actor_id aid,
actor_id aid, error rsn) { error rsn) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn));
auto path = instance.tbl().lookup(nid); auto path = instance.tbl().lookup(nid);
if (!path) { if (!path) {
...@@ -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);
} }
...@@ -168,7 +168,7 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) { ...@@ -168,7 +168,7 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
if (ptr == nullptr) { if (ptr == nullptr) {
CAF_LOG_DEBUG("kill proxy immediately"); CAF_LOG_DEBUG("kill proxy immediately");
// kill immediately if actor has already terminated // kill immediately if actor has already terminated
send_kill_proxy_instance(nid, aid, exit_reason::unknown); send_basp_down_message(nid, aid, exit_reason::unknown);
} else { } else {
auto entry = ptr->address(); auto entry = ptr->address();
auto i = monitored_actors.find(entry); auto i = monitored_actors.find(entry);
...@@ -187,7 +187,7 @@ void basp_broker_state::handle_down_msg(down_msg& dm) { ...@@ -187,7 +187,7 @@ void basp_broker_state::handle_down_msg(down_msg& dm) {
if (i == monitored_actors.end()) if (i == monitored_actors.end())
return; return;
for (auto& nid : i->second) for (auto& nid : i->second)
send_kill_proxy_instance(nid, dm.source.id(), dm.reason); send_basp_down_message(nid, dm.source.id(), dm.reason);
monitored_actors.erase(i); monitored_actors.erase(i);
} }
...@@ -324,28 +324,17 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -324,28 +324,17 @@ 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,
auto path = instance.tbl().lookup(nid); "info"))) {
if (!path) { CAF_LOG_ERROR("learned_new_node called, but no route to remote node"
CAF_LOG_ERROR("learned_new_node called, but no route to nid"); << CAF_ARG(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 +351,23 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -362,40 +351,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 +375,7 @@ void basp_broker_state::set_context(connection_handle hdl) { ...@@ -403,7 +375,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 +417,10 @@ void basp_broker_state::flush(connection_handle hdl) { ...@@ -445,6 +417,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 +489,8 @@ behavior basp_broker::make_behavior() { ...@@ -513,8 +489,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 +500,22 @@ behavior basp_broker::make_behavior() { ...@@ -524,32 +500,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);
} }
......
...@@ -79,36 +79,6 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -79,36 +79,6 @@ connection_state instance::handle(execution_unit* ctx,
} }
} }
CAF_LOG_DEBUG(CAF_ARG(hdr)); 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)) if (!handle(ctx, dm.handle, hdr, payload))
return err(); return err();
return await_header; return await_header;
...@@ -118,7 +88,7 @@ void instance::handle_heartbeat(execution_unit* ctx) { ...@@ -118,7 +88,7 @@ void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) { for (auto& kvp: tbl_.direct_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second)); 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); callee_.flush(kvp.first);
} }
} }
...@@ -194,26 +164,34 @@ size_t instance::remove_published_actor(const actor_addr& whom, ...@@ -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, bool instance::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, message_id mid, const node_id& dest_node, uint64_t dest_actor,
const message& msg) { uint8_t flags, message_id mid, const message& msg) {
CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(receiver) CAF_LOG_TRACE(CAF_ARG(sender) << CAF_ARG(dest_node) << CAF_ARG(mid)
<< CAF_ARG(mid) << CAF_ARG(msg)); << CAF_ARG(msg));
CAF_ASSERT(receiver && system().node() != receiver->node()); CAF_ASSERT(dest_node && this_node_ != dest_node);
auto path = lookup(receiver->node()); auto path = lookup(dest_node);
if (!path) { if (!path) {
notify<hook::message_sending_failed>(sender, receiver, mid, msg); //notify<hook::message_sending_failed>(sender, receiver, mid, msg);
return false; return false;
} }
auto writer = make_callback([&](serializer& sink) -> error { auto& source_node = sender ? sender->node() : this_node_;
return sink(const_cast<std::vector<strong_actor_ptr>&>(forwarding_stack), if (dest_node == path->next_hop && source_node == this_node_) {
const_cast<message&>(msg)); header hdr{message_type::direct_message, flags, 0, mid.integer_value(),
}); sender ? sender->id() : invalid_actor_id, dest_actor};
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(), auto writer = make_callback([&](serializer& sink) -> error {
sender ? sender->node() : this_node(), receiver->node(), return sink(forwarding_stack, msg);
sender ? sender->id() : invalid_actor_id, receiver->id()}; });
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer); 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(source_node, dest_node, forwarding_stack, msg);
});
write(ctx, callee_.get_buffer(path->hdl), hdr, &writer);
}
flush(*path); 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; return true;
} }
...@@ -253,230 +231,278 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf, ...@@ -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"); CAF_LOG_DEBUG_IF(!pa && port, "no actor published");
auto writer = make_callback([&](serializer& sink) -> error { auto writer = make_callback([&](serializer& sink) -> error {
auto id = get_or(callee_.config(), "middleman.app-identifier", auto app_ids = get_or(callee_.config(), "middleman.app-identifiers",
defaults::middleman::app_identifier); defaults::middleman::app_identifiers);
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 aid = invalid_actor_id; auto aid = invalid_actor_id;
std::set<std::string> tmp; auto iface = std::set<std::string>{};
return sink(aid, tmp); 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, header hdr{message_type::server_handshake, 0, 0, version,
this_node_, none, invalid_actor_id, invalid_actor_id};
(pa != nullptr) && pa->first ? pa->first->id() : invalid_actor_id,
invalid_actor_id};
write(ctx, out_buf, hdr, &writer); write(ctx, out_buf, hdr, &writer);
} }
void instance::write_client_handshake(execution_unit* ctx, void instance::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) {
CAF_LOG_TRACE(CAF_ARG(remote_side));
auto writer = make_callback([&](serializer& sink) -> error { 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, 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); write(ctx, buf, hdr, &writer);
} }
void instance::write_client_handshake(execution_unit* ctx, void instance::write_monitor_message(execution_unit* ctx, buffer_type& buf,
buffer_type& buf, const node_id& dest_node, actor_id aid) {
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,
const node_id& dest_node, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid));
header hdr{message_type::announce_proxy, 0, 0, 0, auto writer = make_callback([&](serializer& sink) -> error {
this_node_, dest_node, invalid_actor_id, aid}; return sink(this_node_, dest_node);
write(ctx, buf, hdr); });
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 node_id& dest_node, actor_id aid,
const error& rsn) { const error& rsn) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid) << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid) << CAF_ARG(rsn));
auto writer = make_callback([&](serializer& sink) -> error { 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, header hdr{message_type::down_message, 0, 0, 0, aid, invalid_actor_id};
this_node_, dest_node, aid, invalid_actor_id};
write(ctx, buf, hdr, &writer); write(ctx, buf, hdr, &writer);
} }
void instance::write_heartbeat(execution_unit* ctx, void instance::write_heartbeat(execution_unit* ctx, buffer_type& buf) {
buffer_type& buf, CAF_LOG_TRACE("");
const node_id& remote_side) { header hdr{message_type::heartbeat, 0, 0, 0, invalid_actor_id,
CAF_LOG_TRACE(CAF_ARG(remote_side)); invalid_actor_id};
header hdr{message_type::heartbeat, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr); write(ctx, buf, hdr);
} }
bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr, bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
std::vector<char>* payload) { std::vector<char>* payload) {
// function object for checking payload validity CAF_LOG_TRACE(CAF_ARG(hdl) << CAF_ARG(hdr));
auto payload_valid = [&]() -> bool { // Check payload validity.
return payload != nullptr && payload->size() == hdr.payload_len; if (payload == nullptr) {
}; if (hdr.payload_len != 0) {
// handle message to ourselves 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) { switch (hdr.operation) {
case message_type::server_handshake: { 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; actor_id aid = invalid_actor_id;
std::set<std::string> sigs; std::set<std::string> sigs;
if (!payload_valid()) { if (auto err = bd(source_node, app_ids, aid, sigs)) {
CAF_LOG_ERROR("fail to receive the app identifier"); CAF_LOG_WARNING("unable to deserialize payload of server handshake:"
<< ctx->system().render(err));
return false;
}
// 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; return false;
} else {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return false;
auto appid = get_if<std::string>(&callee_.config(),
"middleman.app-identifier");
if ((appid && *appid != remote_appid)
|| (!appid && !remote_appid.empty())) {
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
e = bd(aid, sigs);
if (e)
return false;
} }
// close self connection after handshake is done // Close connection to ourselves immediately after sending client HS.
if (hdr.source_node == this_node_) { if (source_node == this_node_) {
CAF_LOG_DEBUG("close connection to self immediately"); 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; return false;
} }
// close this connection if we already have a direct connection // Close this connection if we already have a direct connection.
if (tbl_.lookup_direct(hdr.source_node)) { if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG("close connection since we already have a " CAF_LOG_DEBUG("close redundant direct connection:"
"direct connection: " << CAF_ARG(hdr.source_node)); << CAF_ARG(source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(source_node, aid, sigs);
return false; return false;
} }
// add direct route to this node and remove any indirect entry // Add direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(hdr.source_node)); CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, hdr.source_node); tbl_.add_direct(hdl, source_node);
auto was_indirect = tbl_.erase_indirect(hdr.source_node); auto was_indirect = tbl_.erase_indirect(source_node);
// write handshake as client in response // write handshake as client in response
auto path = tbl_.lookup(hdr.source_node); auto path = tbl_.lookup(source_node);
if (!path) { if (!path) {
CAF_LOG_ERROR("no route to host after server handshake"); CAF_LOG_ERROR("no route to host after server handshake");
return false; return false;
} }
write_client_handshake(ctx, callee_.get_buffer(path->hdl), write_client_handshake(ctx, callee_.get_buffer(path->hdl));
hdr.source_node); callee_.learned_new_node_directly(source_node, was_indirect);
callee_.learned_new_node_directly(hdr.source_node, was_indirect); callee_.finalize_handshake(source_node, aid, sigs);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path); flush(*path);
break; break;
} }
case message_type::client_handshake: { case message_type::client_handshake: {
if (!payload_valid()) { // Deserialize payload.
CAF_LOG_ERROR("fail to receive the app identifier"); binary_deserializer bd{ctx, *payload};
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; return false;
} else {
binary_deserializer bd{ctx, *payload};
std::string remote_appid;
auto e = bd(remote_appid);
if (e)
return false;
auto appid = get_if<std::string>(&callee_.config(),
"middleman.app-identifier");
if ((appid && *appid != remote_appid)
|| (!appid && !remote_appid.empty())) {
CAF_LOG_ERROR("app identifier mismatch");
return false;
}
} }
if (tbl_.lookup_direct(hdr.source_node)) { // Drop repeated handshakes.
CAF_LOG_DEBUG("received second client handshake:" if (tbl_.lookup_direct(source_node)) {
<< CAF_ARG(hdr.source_node)); CAF_LOG_DEBUG("received repeated client handshake:"
<< CAF_ARG(source_node));
break; break;
} }
// add direct route to this node and remove any indirect entry // Add direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(hdr.source_node)); CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, hdr.source_node); tbl_.add_direct(hdl, source_node);
auto was_indirect = tbl_.erase_indirect(hdr.source_node); auto was_indirect = tbl_.erase_indirect(source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect); callee_.learned_new_node_directly(source_node, was_indirect);
break; break;
} }
case message_type::dispatch_message: { case message_type::direct_message: {
if (!payload_valid()) // Deserialize payload.
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);
binary_deserializer bd{ctx, *payload}; binary_deserializer bd{ctx, *payload};
auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack; std::vector<strong_actor_ptr> forwarding_stack;
message msg; message msg;
if (hdr.has(header::named_receiver_flag)) { if (auto err = bd(forwarding_stack, msg)) {
auto e = bd(receiver_name); CAF_LOG_WARNING("unable to deserialize payload of direct message:"
if (e) << ctx->system().render(err));
return false;
}
auto e = bd(forwarding_stack, msg);
if (e)
return false; 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)) if (hdr.has(header::named_receiver_flag))
callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name, callee_.deliver(source_node, hdr.source_actor,
make_message_id(hdr.operation_data), static_cast<atom_value>(hdr.dest_actor),
forwarding_stack, msg); make_message_id(hdr.operation_data), forwarding_stack,
msg);
else else
callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor, callee_.deliver(source_node, hdr.source_actor, hdr.dest_actor,
make_message_id(hdr.operation_data), make_message_id(hdr.operation_data), forwarding_stack,
forwarding_stack, msg); msg);
break; break;
} }
case message_type::announce_proxy: case message_type::routed_message: {
callee_.proxy_announced(hdr.source_node, hdr.dest_actor); // 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; 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; 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}; binary_deserializer bd{ctx, *payload};
node_id source_node;
node_id dest_node;
error fail_state; error fail_state;
auto e = bd(fail_state); if (auto err = bd(source_node, dest_node, fail_state)) {
if (e) CAF_LOG_WARNING("unable to deserialize payload of down message:"
<< ctx->system().render(err));
return false; return false;
callee_.proxies().erase(hdr.source_node, hdr.source_actor, }
std::move(fail_state)); 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; break;
} }
case message_type::heartbeat: { case message_type::heartbeat: {
CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node)); CAF_LOG_TRACE("received heartbeat");
callee_.handle_heartbeat(hdr.source_node); callee_.handle_heartbeat();
break; break;
} }
default: default: {
CAF_LOG_ERROR("invalid operation"); CAF_LOG_ERROR("invalid operation");
return false; return false;
}
} }
return true; 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 basp
} // namespace io } // namespace io
} // namespace caf } // namespace caf
...@@ -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"
}; };
......
...@@ -274,26 +274,21 @@ public: ...@@ -274,26 +274,21 @@ public:
// technically, the server handshake arrives // technically, the server handshake arrives
// before we send the client handshake // before we send the client handshake
mock(hdl, mock(hdl,
{basp::message_type::client_handshake, 0, 0, 0, {basp::message_type::client_handshake, 0, 0, 0, invalid_actor_id,
n.id, this_node(), invalid_actor_id},
invalid_actor_id, invalid_actor_id}, std::string{}) n.id)
.receive(hdl, .receive(hdl, basp::message_type::server_handshake, no_flags, any_vals,
basp::message_type::server_handshake, no_flags, basp::version, invalid_actor_id, invalid_actor_id, this_node(),
any_vals, basp::version, this_node(), node_id{none}, defaults::middleman::app_identifiers, published_actor_id,
published_actor_id, invalid_actor_id, std::string{}, published_actor_ifs)
published_actor_id, // upon receiving our client handshake, BASP will check
published_actor_ifs) // whether there is a SpawnServ actor on this node
// upon receiving our client handshake, BASP will check .receive(hdl, basp::message_type::direct_message,
// whether there is a SpawnServ actor on this node basp::header::named_receiver_flag, any_vals,
.receive(hdl, default_operation_data, any_vals,
basp::message_type::dispatch_message, static_cast<uint64_t>(spawn_serv_atom),
basp::header::named_receiver_flag, any_vals, std::vector<strong_actor_ptr>{},
default_operation_data, make_message(sys_atom::value, get_atom::value, "info"));
this_node(), n.id,
any_vals, invalid_actor_id,
spawn_serv_atom,
std::vector<actor_addr>{},
make_message(sys_atom::value, get_atom::value, "info"));
// test whether basp instance correctly updates the // test whether basp instance correctly updates the
// routing table upon receiving client handshakes // routing table upon receiving client handshakes
auto path = tbl().lookup(n.id); auto path = tbl().lookup(n.id);
...@@ -318,7 +313,7 @@ public: ...@@ -318,7 +313,7 @@ public:
buffer buf; buffer buf;
std::tie(hdr, buf) = read_from_out_buf(hdl); std::tie(hdr, buf) = read_from_out_buf(hdl);
CAF_MESSAGE("dispatch output buffer for connection " << hdl.id()); 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}; binary_deserializer source{mpx_, buf};
std::vector<strong_actor_ptr> stages; std::vector<strong_actor_ptr> stages;
message msg; message msg;
...@@ -352,8 +347,6 @@ public: ...@@ -352,8 +347,6 @@ public:
maybe<uint8_t> flags, maybe<uint8_t> flags,
maybe<uint32_t> payload_len, maybe<uint32_t> payload_len,
maybe<uint64_t> operation_data, maybe<uint64_t> operation_data,
maybe<node_id> source_node,
maybe<node_id> dest_node,
maybe<actor_id> source_actor, maybe<actor_id> source_actor,
maybe<actor_id> dest_actor, maybe<actor_id> dest_actor,
const Ts&... xs) { const Ts&... xs) {
...@@ -383,11 +376,15 @@ public: ...@@ -383,11 +376,15 @@ public:
ob.erase(ob.begin(), ob.begin() + basp::header_size); ob.erase(ob.begin(), ob.begin() + basp::header_size);
} }
CAF_CHECK_EQUAL(operation, hdr.operation); 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(flags, static_cast<uint8_t>(hdr.flags));
CAF_CHECK_EQUAL(payload_len, hdr.payload_len); CAF_CHECK_EQUAL(payload_len, hdr.payload_len);
CAF_CHECK_EQUAL(operation_data, hdr.operation_data); 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(source_actor, hdr.source_actor);
CAF_CHECK_EQUAL(dest_actor, hdr.dest_actor); CAF_CHECK_EQUAL(dest_actor, hdr.dest_actor);
CAF_REQUIRE_EQUAL(buf.size(), payload.size()); CAF_REQUIRE_EQUAL(buf.size(), payload.size());
...@@ -475,9 +472,7 @@ CAF_TEST(empty_server_handshake) { ...@@ -475,9 +472,7 @@ CAF_TEST(empty_server_handshake) {
std::tie(hdr, payload) = from_buf(buf); std::tie(hdr, payload) = from_buf(buf);
basp::header expected{basp::message_type::server_handshake, 0, basp::header expected{basp::message_type::server_handshake, 0,
static_cast<uint32_t>(payload.size()), static_cast<uint32_t>(payload.size()),
basp::version, basp::version, invalid_actor_id, invalid_actor_id};
this_node(), none,
invalid_actor_id, invalid_actor_id};
CAF_CHECK(basp::valid(hdr)); CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr)); CAF_CHECK(basp::is_handshake(hdr));
CAF_CHECK_EQUAL(to_string(hdr), to_string(expected)); CAF_CHECK_EQUAL(to_string(hdr), to_string(expected));
...@@ -490,13 +485,20 @@ CAF_TEST(non_empty_server_handshake) { ...@@ -490,13 +485,20 @@ CAF_TEST(non_empty_server_handshake) {
instance().add_published_actor(4242, actor_cast<strong_actor_ptr>(self()), instance().add_published_actor(4242, actor_cast<strong_actor_ptr>(self()),
{"caf::replies_to<@u16>::with<@u16>"}); {"caf::replies_to<@u16>::with<@u16>"});
instance().write_server_handshake(mpx(), buf, uint16_t{4242}); instance().write_server_handshake(mpx(), buf, uint16_t{4242});
buffer expected_buf; basp::header hdr;
basp::header expected{basp::message_type::server_handshake, 0, 0, buffer payload;
basp::version, this_node(), none, std::tie(hdr, payload) = from_buf(buf);
self()->id(), invalid_actor_id}; basp::header expected{basp::message_type::server_handshake, 0,
to_buf(expected_buf, expected, nullptr, std::string{}, static_cast<uint32_t>(payload.size()),
self()->id(), set<string>{"caf::replies_to<@u16>::with<@u16>"}); basp::version, invalid_actor_id, invalid_actor_id};
CAF_CHECK_EQUAL(hexstr(buf), hexstr(expected_buf)); 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) { CAF_TEST(remote_address_and_port) {
...@@ -524,14 +526,12 @@ CAF_TEST(client_handshake_and_dispatch) { ...@@ -524,14 +526,12 @@ CAF_TEST(client_handshake_and_dispatch) {
connect_node(jupiter()); connect_node(jupiter());
// send a message via `dispatch` from node 0 // send a message via `dispatch` from node 0
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::direct_message, 0, 0, 0,
jupiter().id, this_node(), jupiter().dummy_actor->id(), self()->id()}, jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_addr>{}, std::vector<actor_addr>{}, make_message(1, 2, 3))
make_message(1, 2, 3)) .receive(jupiter().connection, basp::message_type::monitor_message,
.receive(jupiter().connection, no_flags, any_vals, no_operation_data, invalid_actor_id,
basp::message_type::announce_proxy, no_flags, no_payload, jupiter().dummy_actor->id(), this_node(), jupiter().id);
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
// must've created a proxy for our remote actor // must've created a proxy for our remote actor
CAF_REQUIRE(proxies().count_proxies(jupiter().id) == 1); CAF_REQUIRE(proxies().count_proxies(jupiter().id) == 1);
// must've send remote node a message that this proxy is monitored now // must've send remote node a message that this proxy is monitored now
...@@ -562,15 +562,13 @@ CAF_TEST(message_forwarding) { ...@@ -562,15 +562,13 @@ CAF_TEST(message_forwarding) {
auto msg = make_message(1, 2, 3); auto msg = make_message(1, 2, 3);
// send a message from node 0 to node 1, forwarded by this node // send a message from node 0 to node 1, forwarded by this node
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, default_operation_data, {basp::message_type::routed_message, 0, 0, default_operation_data,
jupiter().id, mars().id,
invalid_actor_id, mars().dummy_actor->id()}, invalid_actor_id, mars().dummy_actor->id()},
msg) jupiter().id, mars().id, std::vector<strong_actor_ptr>{}, msg)
.receive(mars().connection, .receive(mars().connection, basp::message_type::routed_message, no_flags,
basp::message_type::dispatch_message, no_flags, any_vals, any_vals, default_operation_data, invalid_actor_id,
default_operation_data, jupiter().id, mars().id, mars().dummy_actor->id(), jupiter().id, mars().id,
invalid_actor_id, mars().dummy_actor->id(), std::vector<strong_actor_ptr>{}, msg);
msg);
} }
CAF_TEST(publish_and_connect) { CAF_TEST(publish_and_connect) {
...@@ -600,27 +598,21 @@ CAF_TEST(remote_actor_and_send) { ...@@ -600,27 +598,21 @@ CAF_TEST(remote_actor_and_send) {
auto na = registry()->named_actors(); auto na = registry()->named_actors();
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version, {basp::message_type::server_handshake, 0, 0, basp::version,
jupiter().id, none, invalid_actor_id, invalid_actor_id},
jupiter().dummy_actor->id(), invalid_actor_id}, jupiter().id, defaults::middleman::app_identifiers,
std::string{}, jupiter().dummy_actor->id(), std::set<std::string>{})
jupiter().dummy_actor->id(), .receive(jupiter().connection, basp::message_type::client_handshake,
uint32_t{0}) no_flags, any_vals, no_operation_data, invalid_actor_id,
.receive(jupiter().connection, invalid_actor_id, this_node())
basp::message_type::client_handshake, no_flags, 1u, .receive(jupiter().connection, basp::message_type::direct_message,
no_operation_data, this_node(), jupiter().id, basp::header::named_receiver_flag, any_vals,
invalid_actor_id, invalid_actor_id, std::string{}) default_operation_data, any_vals,
.receive(jupiter().connection, static_cast<uint64_t>(spawn_serv_atom),
basp::message_type::dispatch_message, std::vector<strong_actor_ptr>{},
basp::header::named_receiver_flag, any_vals, make_message(sys_atom::value, get_atom::value, "info"))
default_operation_data, this_node(), jupiter().id, .receive(jupiter().connection, basp::message_type::monitor_message,
any_vals, invalid_actor_id, no_flags, any_vals, no_operation_data, invalid_actor_id,
spawn_serv_atom, jupiter().dummy_actor->id(), this_node(), jupiter().id);
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info"))
.receive(jupiter().connection,
basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
CAF_MESSAGE("BASP broker should've send the proxy"); CAF_MESSAGE("BASP broker should've send the proxy");
f.receive( f.receive(
[&](node_id nid, strong_actor_ptr res, std::set<std::string> ifs) { [&](node_id nid, strong_actor_ptr res, std::set<std::string> ifs) {
...@@ -645,21 +637,16 @@ CAF_TEST(remote_actor_and_send) { ...@@ -645,21 +637,16 @@ CAF_TEST(remote_actor_and_send) {
anon_send(actor_cast<actor>(result), 42); anon_send(actor_cast<actor>(result), 42);
mpx()->flush_runnables(); mpx()->flush_runnables();
// mpx()->exec_runnable(); // process forwarded message in basp_broker // mpx()->exec_runnable(); // process forwarded message in basp_broker
mock() mock().receive(jupiter().connection, basp::message_type::direct_message,
.receive(jupiter().connection, no_flags, any_vals, default_operation_data, invalid_actor_id,
basp::message_type::dispatch_message, no_flags, any_vals, jupiter().dummy_actor->id(), std::vector<strong_actor_ptr>{},
default_operation_data, this_node(), jupiter().id, make_message(42));
invalid_actor_id, jupiter().dummy_actor->id(),
std::vector<actor_id>{},
make_message(42));
auto msg = make_message("hi there!"); auto msg = make_message("hi there!");
CAF_MESSAGE("send message via BASP (from proxy)"); CAF_MESSAGE("send message via BASP (from proxy)");
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::direct_message, 0, 0, 0,
jupiter().id, this_node(),
jupiter().dummy_actor->id(), self()->id()}, jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{}, std::vector<strong_actor_ptr>{}, make_message("hi there!"));
make_message("hi there!"));
self()->receive( self()->receive(
[&](const string& str) { [&](const string& str) {
CAF_CHECK_EQUAL(to_string(self()->current_sender()), to_string(result)); CAF_CHECK_EQUAL(to_string(self()->current_sender()), to_string(result));
...@@ -680,11 +667,9 @@ CAF_TEST(actor_serialize_and_deserialize) { ...@@ -680,11 +667,9 @@ CAF_TEST(actor_serialize_and_deserialize) {
}; };
connect_node(jupiter()); connect_node(jupiter());
auto prx = proxies().get_or_put(jupiter().id, jupiter().dummy_actor->id()); auto prx = proxies().get_or_put(jupiter().id, jupiter().dummy_actor->id());
mock() mock().receive(jupiter().connection, basp::message_type::monitor_message,
.receive(jupiter().connection, no_flags, any_vals, no_operation_data, invalid_actor_id,
basp::message_type::announce_proxy, no_flags, no_payload, prx->id(), this_node(), prx->node());
no_operation_data, this_node(), prx->node(),
invalid_actor_id, prx->id());
CAF_CHECK_EQUAL(prx->node(), jupiter().id); CAF_CHECK_EQUAL(prx->node(), jupiter().id);
CAF_CHECK_EQUAL(prx->id(), jupiter().dummy_actor->id()); CAF_CHECK_EQUAL(prx->id(), jupiter().dummy_actor->id());
auto testee = sys.spawn(testee_impl); auto testee = sys.spawn(testee_impl);
...@@ -692,20 +677,16 @@ CAF_TEST(actor_serialize_and_deserialize) { ...@@ -692,20 +677,16 @@ CAF_TEST(actor_serialize_and_deserialize) {
CAF_MESSAGE("send message via BASP (from proxy)"); CAF_MESSAGE("send message via BASP (from proxy)");
auto msg = make_message(actor_cast<actor_addr>(prx)); auto msg = make_message(actor_cast<actor_addr>(prx));
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::direct_message, 0, 0, 0, prx->id(), testee->id()},
prx->node(), this_node(), std::vector<strong_actor_ptr>{}, msg);
prx->id(), testee->id()},
std::vector<actor_id>{},
msg);
// testee must've responded (process forwarded message in BASP broker) // testee must've responded (process forwarded message in BASP broker)
CAF_MESSAGE("wait until BASP broker writes to its output buffer"); CAF_MESSAGE("wait until BASP broker writes to its output buffer");
while (mpx()->output_buffer(jupiter().connection).empty()) while (mpx()->output_buffer(jupiter().connection).empty())
mpx()->exec_runnable(); // process forwarded message in basp_broker mpx()->exec_runnable(); // process forwarded message in basp_broker
// output buffer must contain the reflected message // output buffer must contain the reflected message
mock().receive(jupiter().connection, basp::message_type::dispatch_message, mock().receive(jupiter().connection, basp::message_type::direct_message,
no_flags, any_vals, default_operation_data, this_node(), no_flags, any_vals, default_operation_data, testee->id(),
prx->node(), testee->id(), prx->id(), std::vector<actor_id>{}, prx->id(), std::vector<strong_actor_ptr>{}, msg);
msg);
} }
CAF_TEST(indirect_connections) { CAF_TEST(indirect_connections) {
...@@ -721,26 +702,22 @@ CAF_TEST(indirect_connections) { ...@@ -721,26 +702,22 @@ CAF_TEST(indirect_connections) {
connect_node(mars(), ax, self()->id()); connect_node(mars(), ax, self()->id());
CAF_MESSAGE("actor from Jupiter sends a message to us via Mars"); CAF_MESSAGE("actor from Jupiter sends a message to us via Mars");
auto mx = mock(mars().connection, auto mx = mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::routed_message, 0, 0, 0,
jupiter().id, this_node(),
jupiter().dummy_actor->id(), self()->id()}, jupiter().dummy_actor->id(), self()->id()},
std::vector<actor_id>{}, jupiter().id, this_node(), std::vector<actor_id>{},
make_message("hello from jupiter!")); make_message("hello from jupiter!"));
CAF_MESSAGE("expect ('sys', 'get', \"info\") from Earth to Jupiter at Mars"); CAF_MESSAGE("expect ('sys', 'get', \"info\") from Earth to Jupiter at Mars");
// this asks Jupiter if it has a 'SpawnServ' // this asks Jupiter if it has a 'SpawnServ'
mx.receive(mars().connection, mx.receive(mars().connection, basp::message_type::routed_message,
basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
default_operation_data, this_node(), jupiter().id, default_operation_data, any_vals,
any_vals, invalid_actor_id, static_cast<uint64_t>(spawn_serv_atom), this_node(), jupiter().id,
spawn_serv_atom, std::vector<strong_actor_ptr>{},
std::vector<actor_id>{},
make_message(sys_atom::value, get_atom::value, "info")); make_message(sys_atom::value, get_atom::value, "info"));
CAF_MESSAGE("expect announce_proxy message at Mars from Earth to Jupiter"); CAF_MESSAGE("expect announce_proxy message at Mars from Earth to Jupiter");
mx.receive(mars().connection, mx.receive(mars().connection, basp::message_type::monitor_message, no_flags,
basp::message_type::announce_proxy, no_flags, no_payload, any_vals, no_operation_data, invalid_actor_id,
no_operation_data, this_node(), jupiter().id, jupiter().dummy_actor->id(), this_node(), jupiter().id);
invalid_actor_id, jupiter().dummy_actor->id());
CAF_MESSAGE("receive message from jupiter"); CAF_MESSAGE("receive message from jupiter");
self()->receive( self()->receive(
[](const std::string& str) -> std::string { [](const std::string& str) -> std::string {
...@@ -749,13 +726,11 @@ CAF_TEST(indirect_connections) { ...@@ -749,13 +726,11 @@ CAF_TEST(indirect_connections) {
} }
); );
mpx()->exec_runnable(); // process forwarded message in basp_broker mpx()->exec_runnable(); // process forwarded message in basp_broker
mock() mock().receive(mars().connection, basp::message_type::routed_message,
.receive(mars().connection, no_flags, any_vals, default_operation_data, self()->id(),
basp::message_type::dispatch_message, no_flags, any_vals, jupiter().dummy_actor->id(), this_node(), jupiter().id,
default_operation_data, this_node(), jupiter().id, std::vector<strong_actor_ptr>{},
self()->id(), jupiter().dummy_actor->id(), make_message("hello from earth!"));
std::vector<actor_id>{},
make_message("hello from earth!"));
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -763,91 +738,78 @@ CAF_TEST_FIXTURE_SCOPE_END() ...@@ -763,91 +738,78 @@ CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture) CAF_TEST_FIXTURE_SCOPE(basp_tests_with_autoconn, autoconn_enabled_fixture)
CAF_TEST(automatic_connection) { CAF_TEST(automatic_connection) {
// this tells our BASP broker to enable the automatic connection feature // Utility helper for verifying routing tables.
//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)
auto check_node_in_tbl = [&](node& n) { auto check_node_in_tbl = [&](node& n) {
auto hdl = tbl().lookup_direct(n.id); auto hdl = tbl().lookup_direct(n.id);
CAF_REQUIRE(hdl); CAF_REQUIRE(hdl);
}; };
// Setup.
mpx()->provide_scribe("jupiter", 8080, jupiter().connection); mpx()->provide_scribe("jupiter", 8080, jupiter().connection);
CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080)); CAF_CHECK(mpx()->has_pending_scribe("jupiter", 8080));
CAF_MESSAGE("self: " << to_string(self()->address())); CAF_MESSAGE("self: " << to_string(self()->address()));
auto ax = accept_handle::from_int(4242); auto ax = accept_handle::from_int(4242);
mpx()->provide_acceptor(4242, ax); mpx()->provide_acceptor(4242, ax);
publish(self(), 4242); 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"); CAF_MESSAGE("connect to mars");
connect_node(mars(), ax, self()->id()); connect_node(mars(), ax, self()->id());
//CAF_CHECK_EQUAL(tbl().lookup_direct(mars().id).id(), mars().connection.id());
check_node_in_tbl(mars()); check_node_in_tbl(mars());
CAF_MESSAGE("simulate that an actor from jupiter " CAF_MESSAGE("simulate that a message from jupiter travels over mars");
"sends a message to us via mars");
mock(mars().connection, mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::routed_message, 0, 0,
jupiter().id, this_node(), make_message_id().integer_value(), jupiter().dummy_actor->id(),
jupiter().dummy_actor->id(), self()->id()}, self()->id()},
std::vector<actor_id>{}, jupiter().id, this_node(), std::vector<strong_actor_ptr>{},
make_message("hello from jupiter!")) make_message("hello from jupiter!"))
.receive(mars().connection, .receive(mars().connection, basp::message_type::routed_message,
basp::message_type::dispatch_message, basp::header::named_receiver_flag, any_vals,
basp::header::named_receiver_flag, any_vals, default_operation_data, any_vals,
default_operation_data, this_node(), jupiter().id, static_cast<uint64_t>(spawn_serv_atom), this_node(), jupiter().id,
any_vals, invalid_actor_id, spawn_serv_atom, std::vector<strong_actor_ptr>{},
std::vector<actor_id>{}, make_message(sys_atom::value, get_atom::value, "info"))
make_message(sys_atom::value, get_atom::value, "info")) .receive(mars().connection, basp::message_type::routed_message,
.receive(mars().connection, basp::header::named_receiver_flag, any_vals,
basp::message_type::dispatch_message, default_operation_data,
basp::header::named_receiver_flag, any_vals, any_vals, // actor ID of an actor spawned by the BASP broker
default_operation_data, this_node(), jupiter().id, static_cast<uint64_t>(config_serv_atom), this_node(), jupiter().id,
any_vals, // actor ID of an actor spawned by the BASP broker std::vector<strong_actor_ptr>{},
invalid_actor_id, make_message(get_atom::value, "basp.default-connectivity-tcp"))
config_serv_atom, .receive(mars().connection, basp::message_type::monitor_message, no_flags,
std::vector<actor_id>{}, any_vals, no_operation_data, invalid_actor_id,
make_message(get_atom::value, "basp.default-connectivity-tcp")) jupiter().dummy_actor->id(), this_node(), jupiter().id);
.receive(mars().connection,
basp::message_type::announce_proxy, no_flags, no_payload,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, jupiter().dummy_actor->id());
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); 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(jupiter().id), mars().id);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none); CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
auto connection_helper_actor = sys.latest_actor_id(); auto connection_helper_actor = sys.latest_actor_id();
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); 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"); CAF_MESSAGE("receive ConfigServ of jupiter");
network::address_listing res; network::address_listing res;
res[network::protocol::ipv4].emplace_back("jupiter"); res[network::protocol::ipv4].emplace_back("jupiter");
mock(mars().connection, mock(mars().connection,
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::routed_message, 0, 0,
this_node(), this_node(), make_message_id().integer_value(), invalid_actor_id,
invalid_actor_id, connection_helper_actor}, connection_helper_actor},
std::vector<actor_id>{}, this_node(), this_node(), std::vector<strong_actor_ptr>{},
make_message("basp.default-connectivity-tcp", make_message("basp.default-connectivity-tcp",
make_message(uint16_t{8080}, std::move(res)))); make_message(uint16_t{8080}, std::move(res))));
// our connection helper should now connect to jupiter and // Our connection helper should now connect to jupiter and send the scribe
// send the scribe handle over to the BASP broker // handle over to the BASP broker.
while (mpx()->has_pending_scribe("jupiter", 8080)) { while (mpx()->has_pending_scribe("jupiter", 8080)) {
sched.run(); sched.run();
mpx()->flush_runnables(); mpx()->flush_runnables();
} }
CAF_REQUIRE(mpx()->output_buffer(mars().connection).empty()); CAF_REQUIRE(mpx()->output_buffer(mars().connection).empty());
// send handshake from jupiter // Send handshake from jupiter.
mock(jupiter().connection, mock(jupiter().connection,
{basp::message_type::server_handshake, 0, 0, basp::version, {basp::message_type::server_handshake, no_flags, 0, basp::version,
jupiter().id, none, invalid_actor_id, invalid_actor_id},
jupiter().dummy_actor->id(), invalid_actor_id}, jupiter().id, defaults::middleman::app_identifiers,
std::string{}, jupiter().dummy_actor->id(), std::set<std::string>{})
jupiter().dummy_actor->id(), .receive(jupiter().connection, basp::message_type::client_handshake,
uint32_t{0}) no_flags, any_vals, no_operation_data, invalid_actor_id,
.receive(jupiter().connection, invalid_actor_id, this_node());
basp::message_type::client_handshake, no_flags, 1u,
no_operation_data, this_node(), jupiter().id,
invalid_actor_id, invalid_actor_id, std::string{});
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none); CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none); CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
check_node_in_tbl(jupiter()); check_node_in_tbl(jupiter());
...@@ -861,13 +823,11 @@ CAF_TEST(automatic_connection) { ...@@ -861,13 +823,11 @@ CAF_TEST(automatic_connection) {
); );
mpx()->exec_runnable(); // process forwarded message in basp_broker mpx()->exec_runnable(); // process forwarded message in basp_broker
CAF_MESSAGE("response message must take direct route now"); CAF_MESSAGE("response message must take direct route now");
mock() mock().receive(jupiter().connection, basp::message_type::direct_message,
.receive(jupiter().connection, no_flags, any_vals, make_message_id().integer_value(),
basp::message_type::dispatch_message, no_flags, any_vals, self()->id(), jupiter().dummy_actor->id(),
default_operation_data, this_node(), jupiter().id, std::vector<strong_actor_ptr>{},
self()->id(), jupiter().dummy_actor->id(), make_message("hello from earth!"));
std::vector<actor_id>{},
make_message("hello from earth!"));
CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u); CAF_CHECK_EQUAL(mpx()->output_buffer(mars().connection).size(), 0u);
} }
......
...@@ -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