Commit d8842778 authored by Dominik Charousset's avatar Dominik Charousset

Implement send-to-named-actor feature in BASP

parent 873e85e1
doc/announce_proxy_instance.png

37.9 KB | W: | H:

doc/announce_proxy_instance.png

207 KB | W: | H:

doc/announce_proxy_instance.png
doc/announce_proxy_instance.png
doc/announce_proxy_instance.png
doc/announce_proxy_instance.png
  • 2-up
  • Swipe
  • Onion skin
No preview for this file type
doc/basp_header.png

76.8 KB | W: | H:

doc/basp_header.png

219 KB | W: | H:

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

35.8 KB | W: | H:

doc/client_handshake.png

213 KB | W: | H:

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

47 KB | W: | H:

doc/dispatch_message.png

285 KB | W: | H:

doc/dispatch_message.png
doc/dispatch_message.png
doc/dispatch_message.png
doc/dispatch_message.png
  • 2-up
  • Swipe
  • Onion skin
doc/kill_proxy_instance.png

39.3 KB | W: | H:

doc/kill_proxy_instance.png

212 KB | W: | H:

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

44.1 KB | W: | H:

doc/server_handshake.png

243 KB | W: | H:

doc/server_handshake.png
doc/server_handshake.png
doc/server_handshake.png
doc/server_handshake.png
  • 2-up
  • Swipe
  • Onion skin
......@@ -456,6 +456,8 @@ private:
opencl::manager* opencl_manager_;
riac::probe* probe_;
bool await_actors_before_shutdown_;
strong_actor_ptr config_serv_;
strong_actor_ptr spawn_serv_;
};
} // namespace caf
......
......@@ -144,151 +144,12 @@ auto actor_registry::named_actors() const -> name_map {
return named_entries_;
}
namespace {
struct kvstate {
using key_type = std::string;
using mapped_type = message;
using subscriber_set = std::unordered_set<strong_actor_ptr>;
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<strong_actor_ptr, topic_set> subscribers;
const char* name = "caf.config_server";
template <class Processor>
friend void serialize(Processor& proc, kvstate& x, const unsigned int) {
proc & x.data;
proc & x.subscribers;
}
};
} // namespace <anonymous>
void actor_registry::start() {
auto kvstore = [](stateful_actor<kvstate>* self) -> behavior {
CAF_LOG_TRACE("");
std::string wildcard = "*";
auto unsubscribe_all = [=](actor subscriber) {
auto& subscribers = self->state.subscribers;
auto ptr = actor_cast<strong_actor_ptr>(subscriber);
auto i = subscribers.find(ptr);
if (i == subscribers.end())
return;
for (auto& key : i->second)
self->state.data[key].second.erase(ptr);
subscribers.erase(i);
};
self->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
auto ptr = actor_cast<strong_actor_ptr>(dm.source);
if (ptr)
unsubscribe_all(actor_cast<actor>(std::move(ptr)));
});
return {
[=](put_atom, const std::string& key, message& msg) {
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(msg));
if (key == "*")
return;
auto& vp = self->state.data[key];
vp.first = std::move(msg);
for (auto& subscriber_ptr : vp.second) {
// we never put a nullptr in our map
auto subscriber = actor_cast<actor>(subscriber_ptr);
if (subscriber != self->current_sender())
self->send(subscriber, update_atom::value, key, vp.second);
}
// also iterate all subscribers for '*'
for (auto& subscriber : self->state.data[wildcard].second)
if (subscriber != self->current_sender())
self->send(actor_cast<actor>(subscriber), update_atom::value,
key, vp.second);
},
[=](get_atom, std::string& key) -> message {
CAF_LOG_TRACE(CAF_ARG(key));
if (key == wildcard) {
std::vector<std::pair<std::string, message>> msgs;
for (auto& kvp : self->state.data)
if (kvp.first != "*")
msgs.emplace_back(kvp.first, kvp.second.first);
return make_message(ok_atom::value, std::move(msgs));
}
auto i = self->state.data.find(key);
return make_message(ok_atom::value, std::move(key),
i != self->state.data.end() ? i->second.first
: make_message());
},
[=](subscribe_atom, const std::string& key) {
auto subscriber = actor_cast<strong_actor_ptr>(self->current_sender());
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(subscriber));
if (! subscriber)
return;
self->state.data[key].second.insert(subscriber);
auto& subscribers = self->state.subscribers;
auto i = subscribers.find(subscriber);
if (i != subscribers.end()) {
i->second.insert(key);
} else {
self->monitor(subscriber);
subscribers.emplace(subscriber, kvstate::topic_set{key});
}
},
[=](unsubscribe_atom, const std::string& key) {
auto subscriber = actor_cast<strong_actor_ptr>(self->current_sender());
if (! subscriber)
return;
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(subscriber));
if (key == wildcard) {
unsubscribe_all(actor_cast<actor>(std::move(subscriber)));
return;
}
self->state.subscribers[subscriber].erase(key);
self->state.data[key].second.erase(subscriber);
}
};
};
auto spawn_serv = [](event_based_actor* self) -> behavior {
CAF_LOG_TRACE("");
return {
[=](get_atom, const std::string& name, message& args)
-> result<ok_atom, strong_actor_ptr, std::set<std::string>> {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args));
actor_config cfg{self->context()};
auto res = self->system().types().make_actor(name, cfg, args);
if (! res.first)
return sec::cannot_spawn_actor_from_arguments;
return {ok_atom::value, res.first, res.second};
}
};
};
// NOTE: all actors that we spawn here *must not* access the scheduler,
// because the scheduler will make sure that the registry is running
// as part of its initialization; hence, all actors have to
// use the lazy_init flag
//named_entries_.emplace(atom("SpawnServ"), system_.spawn_announce_actor_type_server());
auto cs = system_.spawn<hidden+lazy_init>(kvstore);
put(atom("ConfigServ"), actor_cast<strong_actor_ptr>(std::move(cs)));
auto ss = system_.spawn<hidden+lazy_init>(spawn_serv);
put(atom("SpawnServ"), actor_cast<strong_actor_ptr>(std::move(ss)));
// nop
}
void actor_registry::stop() {
class dropping_execution_unit : public execution_unit {
public:
dropping_execution_unit(actor_system* sys) : execution_unit(sys) {
// nop
}
void exec_later(resumable*) override {
// should not happen in the first place
CAF_LOG_ERROR("actor registry actor called exec_later during shutdown");
}
};
// the scheduler is already stopped -> invoke exit messages manually
dropping_execution_unit dummy{&system_};
for (auto& kvp : named_entries_) {
auto mp = mailbox_element::make(nullptr, invalid_message_id, {},
exit_msg{kvp.second->address(),
exit_reason::kill});
auto ptr = static_cast<local_actor*>(actor_cast<abstract_actor*>(kvp.second));
ptr->exec_single_event(&dummy, mp);
}
named_entries_.clear();
}
} // namespace caf
......@@ -19,7 +19,10 @@
#include "caf/actor_system.hpp"
#include <unordered_set>
#include "caf/send.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/policy/work_sharing.hpp"
......@@ -31,6 +34,134 @@
namespace caf {
namespace {
struct kvstate {
using key_type = std::string;
using mapped_type = message;
using subscriber_set = std::unordered_set<strong_actor_ptr>;
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<strong_actor_ptr, topic_set> subscribers;
const char* name = "caf.config_server";
template <class Processor>
friend void serialize(Processor& proc, kvstate& x, const unsigned int) {
proc & x.data;
proc & x.subscribers;
}
};
behavior config_serv_impl(stateful_actor<kvstate>* self) {
CAF_LOG_TRACE("");
std::string wildcard = "*";
auto unsubscribe_all = [=](actor subscriber) {
auto& subscribers = self->state.subscribers;
auto ptr = actor_cast<strong_actor_ptr>(subscriber);
auto i = subscribers.find(ptr);
if (i == subscribers.end())
return;
for (auto& key : i->second)
self->state.data[key].second.erase(ptr);
subscribers.erase(i);
};
self->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
auto ptr = actor_cast<strong_actor_ptr>(dm.source);
if (ptr)
unsubscribe_all(actor_cast<actor>(std::move(ptr)));
});
return {
[=](put_atom, const std::string& key, message& msg) {
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(msg));
if (key == "*")
return;
auto& vp = self->state.data[key];
vp.first = std::move(msg);
for (auto& subscriber_ptr : vp.second) {
// we never put a nullptr in our map
auto subscriber = actor_cast<actor>(subscriber_ptr);
if (subscriber != self->current_sender())
self->send(subscriber, update_atom::value, key, vp.second);
}
// also iterate all subscribers for '*'
for (auto& subscriber : self->state.data[wildcard].second)
if (subscriber != self->current_sender())
self->send(actor_cast<actor>(subscriber), update_atom::value,
key, vp.second);
},
[=](get_atom, std::string& key) -> message {
CAF_LOG_TRACE(CAF_ARG(key));
if (key == wildcard) {
std::vector<std::pair<std::string, message>> msgs;
for (auto& kvp : self->state.data)
if (kvp.first != "*")
msgs.emplace_back(kvp.first, kvp.second.first);
return make_message(ok_atom::value, std::move(msgs));
}
auto i = self->state.data.find(key);
return make_message(ok_atom::value, std::move(key),
i != self->state.data.end() ? i->second.first
: make_message());
},
[=](subscribe_atom, const std::string& key) {
auto subscriber = actor_cast<strong_actor_ptr>(self->current_sender());
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(subscriber));
if (! subscriber)
return;
self->state.data[key].second.insert(subscriber);
auto& subscribers = self->state.subscribers;
auto i = subscribers.find(subscriber);
if (i != subscribers.end()) {
i->second.insert(key);
} else {
self->monitor(subscriber);
subscribers.emplace(subscriber, kvstate::topic_set{key});
}
},
[=](unsubscribe_atom, const std::string& key) {
auto subscriber = actor_cast<strong_actor_ptr>(self->current_sender());
if (! subscriber)
return;
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(subscriber));
if (key == wildcard) {
unsubscribe_all(actor_cast<actor>(std::move(subscriber)));
return;
}
self->state.subscribers[subscriber].erase(key);
self->state.data[key].second.erase(subscriber);
}
};
}
behavior spawn_serv_impl(event_based_actor* self) {
CAF_LOG_TRACE("");
return {
[=](get_atom, const std::string& name, message& args)
-> result<ok_atom, strong_actor_ptr, std::set<std::string>> {
CAF_LOG_TRACE(CAF_ARG(name) << CAF_ARG(args));
actor_config cfg{self->context()};
auto res = self->system().types().make_actor(name, cfg, args);
if (! res.first)
return sec::cannot_spawn_actor_from_arguments;
return {ok_atom::value, res.first, res.second};
}
};
}
class dropping_execution_unit : public execution_unit {
public:
dropping_execution_unit(actor_system* sys) : execution_unit(sys) {
// nop
}
void exec_later(resumable*) override {
// should not happen in the first place
CAF_LOG_ERROR("actor registry actor called exec_later during shutdown");
}
};
} // namespace <anonymous>
actor_system::module::~module() {
// nop
}
......@@ -125,22 +256,31 @@ actor_system::actor_system(actor_system_config&& cfg)
types_.error_renderers_ = std::move(cfg.error_renderers_);
// move remaining config
node_.swap(cfg.network_id);
// spawn config and spawn servers (lazily to not access the scheduler yet)
static constexpr auto Flags = hidden + lazy_init;
spawn_serv_ = actor_cast<strong_actor_ptr>(spawn<Flags>(spawn_serv_impl));
config_serv_ = actor_cast<strong_actor_ptr>(spawn<Flags>(config_serv_impl));
// fire up remaining modules
logger_.start();
registry_.start();
registry_.put(atom("SpawnServ"), spawn_serv_);
registry_.put(atom("ConfigServ"), config_serv_);
for (auto& mod : modules_)
if (mod)
mod->start();
groups_.start();
// store config parameters in ConfigServ
auto cs = actor_cast<actor>(registry_.get(atom("ConfigServ")));
anon_send(cs, put_atom::value, "middleman.enable-automatic-connections",
make_message(cfg.middleman_enable_automatic_connections));
}
actor_system::~actor_system() {
if (await_actors_before_shutdown_)
await_all_actors_done();
// shutdown system-level servers
anon_send_exit(spawn_serv_, exit_reason::user_shutdown);
anon_send_exit(config_serv_, exit_reason::user_shutdown);
// release memory as soon as possible
spawn_serv_ = nullptr;
config_serv_ = nullptr;
// group module is the first one, relies on MM
groups_.stop();
// stop modules in reverse order
for (auto i = modules_.rbegin(); i != modules_.rend(); ++i)
......
......@@ -39,24 +39,56 @@ namespace basp {
/// message types consist of only a header.
struct header {
message_type operation;
uint8_t padding1;
uint8_t padding2;
uint8_t flags;
uint32_t payload_len;
uint64_t operation_data;
node_id source_node;
node_id dest_node;
actor_id source_actor;
actor_id dest_actor;
inline header(message_type m_operation, uint8_t m_flags,
uint32_t m_payload_len, uint64_t m_operation_data,
node_id m_source_node, node_id m_dest_node,
actor_id m_source_actor, actor_id m_dest_actor)
: operation(m_operation),
flags(m_flags),
payload_len(m_payload_len),
operation_data(m_operation_data),
source_node(std::move(m_source_node)),
dest_node(std::move(m_dest_node)),
source_actor(m_source_actor),
dest_actor(m_dest_actor) {
// nop
}
header() = default;
/// Identifies a receiver by name rather than ID.
static const uint8_t named_receiver_flag = 0x01;
/// Queries whether this header has the given flag.
inline bool has(uint8_t flag) const {
return (flags & flag) != 0;
}
};
/// @relates header
template <class Processor>
void serialize(Processor& proc, header& hdr, const unsigned int) {
uint8_t pad = 0;
proc & hdr.operation;
proc & pad;
proc & pad;
proc & hdr.flags;
proc & hdr.payload_len;
proc & hdr.operation_data;
proc & hdr.source_node;
proc & hdr.dest_node;
proc & hdr.source_actor;
proc & hdr.dest_actor;
proc & hdr.payload_len;
proc & hdr.operation;
proc & hdr.operation_data;
}
/// @relates header
......
......@@ -68,10 +68,15 @@ public:
virtual void kill_proxy(const node_id& nid, actor_id aid,
const error& rsn) = 0;
/// Called whenever a `dispatch_message` arrived for a local actor.
/// Called for each `dispatch_message` without `named_receiver_flag`.
virtual void deliver(const node_id& source_node, actor_id source_actor,
const node_id& dest_node, actor_id dest_actor,
message_id mid,
actor_id dest_actor, message_id mid,
std::vector<strong_actor_ptr>& forwarding_stack,
message& msg) = 0;
/// Called for each `dispatch_message` with `named_receiver_flag`.
virtual void deliver(const node_id& source_node, actor_id source_actor,
atom_value dest_actor, message_id mid,
std::vector<strong_actor_ptr>& forwarding_stack,
message& msg) = 0;
......@@ -175,19 +180,6 @@ public:
return published_actors_;
}
/// Writes a header (build from the arguments)
/// followed by its payload to `storage`.
void write(execution_unit* ctx,
buffer_type& storage,
message_type operation,
uint32_t* payload_len,
uint64_t operation_data,
const node_id& source_node,
const node_id& dest_node,
actor_id source_actor,
actor_id dest_actor,
payload_writer* writer = nullptr);
/// Writes a header followed by its payload to `storage`.
void write(execution_unit* ctx, buffer_type& storage, header& hdr,
payload_writer* writer = nullptr);
......@@ -203,11 +195,13 @@ public:
void write_client_handshake(execution_unit* ctx,
buffer_type& buf, const node_id& remote_side);
/// Writes a `kill_proxy_instance` to `buf`.
void write_kill_proxy_instance(execution_unit* ctx,
buffer_type& buf,
const node_id& dest_node,
actor_id aid,
/// Writes an `announce_proxy` to `buf`.
void write_announce_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid);
/// Writes a `kill_proxy` to `buf`.
void write_kill_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid,
const error& fail_state);
/// Writes a `heartbeat` to `buf`.
......
......@@ -31,7 +31,7 @@ namespace basp {
/// Describes the first header field of a BASP message and determines the
/// interpretation of the other header fields.
enum class message_type : uint32_t {
enum class message_type : uint8_t {
/// Send from server, i.e., the node with a published actor, to client,
/// i.e., node that initiates a new connection using remote_actor().
///
......@@ -56,15 +56,16 @@ enum class message_type : uint32_t {
/// message on termination.
///
/// ![](announce_proxy_instance.png)
announce_proxy_instance = 0x03,
announce_proxy = 0x03,
/// Informs the receiving node that it has a proxy for an actor
/// that has been terminated.
///
/// ![](kill_proxy_instance.png)
kill_proxy_instance = 0x04,
kill_proxy = 0x04,
/// Send to remote node which has direct connection.
/// Used to generate periodic traffic between two nodes
/// in order to detect disconnects.
///
/// ![](heartbeat.png)
heartbeat = 0x05,
......
......@@ -29,7 +29,7 @@ namespace basp {
/// The current BASP version. Different BASP versions will not
/// be able to exchange messages.
constexpr uint64_t version = 1;
constexpr uint64_t version = 2;
/// @}
......
......@@ -67,9 +67,18 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::listener
void deliver(const node_id& source_node, actor_id source_actor,
const node_id& dest_node, actor_id dest_actor,
message_id mid, std::vector<strong_actor_ptr>& stages,
message& msg) override;
actor_id dest_actor, message_id mid,
std::vector<strong_actor_ptr>& stages, message& msg) override;
// inherited from basp::instance::listener
void deliver(const node_id& source_node, actor_id source_actor,
atom_value dest_actor, message_id mid,
std::vector<strong_actor_ptr>& stages, message& msg) override;
// called from both overriden functions
void deliver(const node_id& source_node, actor_id source_actor,
strong_actor_ptr dest, message_id mid,
std::vector<strong_actor_ptr>& stages, message& msg);
// performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid);
......
......@@ -293,15 +293,14 @@ void asio_multiplexer::exec_later(resumable* rptr) {
switch (rptr->subtype()) {
case resumable::io_actor:
case resumable::function_object: {
intrusive_ptr<resumable> ptr{rptr};
service().post([=]() {
intrusive_ptr<resumable> ptr{rptr, false};
service().post([=]() mutable {
switch (ptr->resume(this, max_throughput())) {
case resumable::resume_later:
exec_later(ptr.get());
exec_later(ptr.release());
break;
case resumable::done:
case resumable::awaiting_message:
intrusive_ptr_release(ptr.get());
break;
default:
; // ignored
......
......@@ -104,11 +104,8 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
"write announce_proxy_instance:"
<< CAF_ARG(nid) << CAF_ARG(aid));
// tell remote side we are monitoring this actor now
instance.write(self->context(),
self->wr_buf(this_context->hdl),
basp::message_type::announce_proxy_instance, nullptr, 0,
this_node(), nid,
invalid_actor_id, aid);
instance.write_announce_proxy(self->context(),
self->wr_buf(this_context->hdl), nid, aid);
instance.tbl().flush(*path);
mm->notify<hook::new_remote_actor>(res);
return res;
......@@ -177,7 +174,7 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
<< CAF_ARG(nid));
return;
}
instance.write_kill_proxy_instance(self->context(), path->wr_buf,
instance.write_kill_proxy(self->context(), path->wr_buf,
nid, aid, rsn);
instance.tbl().flush(*path);
};
......@@ -208,43 +205,39 @@ void basp_broker_state::kill_proxy(const node_id& nid, actor_id aid,
proxies().erase(nid, aid, rsn);
}
void basp_broker_state::deliver(const node_id& src_nid,
actor_id src_aid,
const node_id& dest_nid,
actor_id dest_aid,
message_id mid,
void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
actor_id dest_aid, message_id mid,
std::vector<strong_actor_ptr>& stages,
message& msg) {
CAF_LOG_TRACE(CAF_ARG(src_nid)
<< CAF_ARG(src_aid) << CAF_ARG(dest_nid)
CAF_LOG_TRACE(CAF_ARG(src_nid) << CAF_ARG(src_aid)
<< CAF_ARG(dest_aid) << CAF_ARG(msg) << CAF_ARG(mid));
strong_actor_ptr src;
if (src_nid == this_node())
src = system().registry().get(src_aid);
else
src = proxies().get_or_put(src_nid, src_aid);
strong_actor_ptr dest;
auto rsn = exit_reason::remote_link_unreachable;
if (dest_nid != this_node()) {
dest = proxies().get_or_put(dest_nid, dest_aid);
} else {
// TODO: replace hack with clean solution
if (dest_aid == std::numeric_limits<actor_id>::max()) {
// this hack allows CAF to talk to older CAF versions; CAF <= 0.14 will
// discard this message silently as the receiver is not found, while
// CAF >= 0.14.1 will use the operation data to identify named actors
auto dest_name = static_cast<atom_value>(mid.integer_value());
CAF_LOG_DEBUG(CAF_ARG(dest_name));
mid = message_id::make(); // override this since the message is async
dest = system().registry().get(dest_name);
} else {
dest = system().registry().get(dest_aid);
}
}
deliver(src_nid, src_aid, system().registry().get(dest_aid),
mid, stages, msg);
}
void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
atom_value dest_name, message_id mid,
std::vector<strong_actor_ptr>& stages,
message& msg) {
CAF_LOG_TRACE(CAF_ARG(src_nid) << CAF_ARG(src_aid)
<< CAF_ARG(dest_name) << CAF_ARG(msg) << CAF_ARG(mid));
deliver(src_nid, src_aid, system().registry().get(dest_name),
mid, stages, msg);
}
void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
strong_actor_ptr dest, message_id mid,
std::vector<strong_actor_ptr>& stages,
message& msg) {
CAF_LOG_TRACE(CAF_ARG(src_nid) << CAF_ARG(src_aid) << CAF_ARG(dest)
<< CAF_ARG(msg) << CAF_ARG(mid));
auto src = src_nid == this_node() ? system().registry().get(src_aid)
: proxies().get_or_put(src_nid, src_aid);
if (! dest) {
auto rsn = exit_reason::remote_link_unreachable;
CAF_LOG_INFO("cannot deliver message, destination not found");
self->parent().notify<hook::invalid_message_received>(src_nid, src,
dest_aid, mid, msg);
0, mid, msg);
if (mid.valid() && src) {
detail::sync_request_bouncer srb{rsn};
srb(src, mid);
......@@ -303,24 +296,30 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
using namespace detail;
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([](serializer& sink) {
auto name = atom("SpawnServ");
std::vector<actor_id> stages;
auto msg = make_message(sys_atom::value, get_atom::value, "info");
sink << stages << msg;
sink << name << stages << msg;
});
auto path = instance.tbl().lookup(nid);
if (! path) {
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, 0, 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(),
path->wr_buf,
instance.write(self->context(), path->wr_buf, hdr, &writer);
/*
basp::message_type::dispatch_message,
nullptr, static_cast<uint64_t>(atom("SpawnServ")),
this_node(), nid,
tmp.id(), std::numeric_limits<actor_id>::max(),
&writer);
*/
instance.flush(*path);
}
......@@ -400,19 +399,15 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
auto tmp = system().spawn<detached + hidden>(connection_helper, self);
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([](serializer& sink) {
auto name = atom("ConfigServ");
std::vector<actor_id> stages;
auto msg = make_message(get_atom::value, "basp.default-connectivity");
sink << stages << msg;
sink << name << stages << msg;
});
// 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(),
path->wr_buf,
basp::message_type::dispatch_message,
nullptr, static_cast<uint64_t>(atom("ConfigServ")),
this_node(), nid,
tmp.id(), std::numeric_limits<actor_id>::max(),
&writer);
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, 0, this_node(), nid, tmp.id(), invalid_actor_id};
instance.write(self->context(), path->wr_buf, hdr, &writer);
instance.flush(*path);
}
......@@ -424,8 +419,8 @@ void basp_broker_state::set_context(connection_handle hdl) {
i = ctx.emplace(hdl,
connection_context{
basp::await_header,
basp::header{basp::message_type::server_handshake, 0, 0,
invalid_node_id, invalid_node_id,
basp::header{basp::message_type::server_handshake, 0,
0, 0, invalid_node_id, invalid_node_id,
invalid_actor_id, invalid_actor_id},
hdl,
invalid_node_id,
......@@ -524,23 +519,18 @@ behavior basp_broker::make_behavior() {
system().registry().put(sender->id(), sender);
auto writer = make_callback([&](serializer& sink) {
std::vector<actor_addr> stages;
sink << stages << msg;
sink << receiver_name << stages << msg;
});
auto path = this->state.instance.tbl().lookup(receiving_node);
if (! path) {
CAF_LOG_ERROR("no route to receiving node");
return sec::no_route_to_receiving_node;
}
// writing std::numeric_limits<actor_id>::max() is a hack to get
// this send-to-named-actor feature working with older CAF releases
this->state.instance.write(context(),
path->wr_buf,
basp::message_type::dispatch_message,
nullptr, static_cast<uint64_t>(receiver_name),
state.this_node(), receiving_node,
sender->id(),
std::numeric_limits<actor_id>::max(),
&writer);
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
0, 0, state.this_node(), receiving_node,
sender->id(), invalid_actor_id};
state.instance.write(context(), path->wr_buf, hdr, &writer);
state.instance.flush(*path);
return unit;
},
......
......@@ -25,10 +25,20 @@ namespace caf {
namespace io {
namespace basp {
const uint8_t header::named_receiver_flag;
std::string to_bin(uint8_t x) {
std::string res;
for (auto offset = 7; offset > 0; --offset)
res += std::to_string((x >> offset) & 0x01);
return res;
}
std::string to_string(const header &hdr) {
std::ostringstream oss;
oss << "{"
<< to_string(hdr.operation) << ", "
<< to_bin(hdr.flags) << ", "
<< hdr.payload_len << ", "
<< hdr.operation_data << ", "
<< to_string(hdr.source_node) << ", "
......@@ -41,6 +51,7 @@ std::string to_string(const header &hdr) {
bool operator==(const header& lhs, const header& rhs) {
return lhs.operation == rhs.operation
&& lhs.flags == rhs.flags
&& lhs.payload_len == rhs.payload_len
&& lhs.operation_data == rhs.operation_data
&& lhs.source_node == rhs.source_node
......@@ -79,7 +90,7 @@ bool client_handshake_valid(const header& hdr) {
bool dispatch_message_valid(const header& hdr) {
return valid(hdr.dest_node)
&& ! zero(hdr.dest_actor)
&& (! zero(hdr.dest_actor) || hdr.has(header::named_receiver_flag))
&& ! zero(hdr.payload_len);
}
......@@ -125,9 +136,9 @@ bool valid(const header& hdr) {
return client_handshake_valid(hdr);
case message_type::dispatch_message:
return dispatch_message_valid(hdr);
case message_type::announce_proxy_instance:
case message_type::announce_proxy:
return announce_proxy_instance_valid(hdr);
case message_type::kill_proxy_instance:
case message_type::kill_proxy:
return kill_proxy_instance_valid(hdr);
case message_type::heartbeat:
return heartbeat_valid(hdr);
......
......@@ -173,20 +173,27 @@ connection_state instance::handle(execution_unit* ctx,
&& tbl_.add_indirect(last_hop, hdr.source_node))
callee_.learned_new_node_indirectly(hdr.source_node);
binary_deserializer bd{ctx, *payload};
auto receiver_name = static_cast<atom_value>(0);
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
if (hdr.has(header::named_receiver_flag))
bd >> receiver_name;
bd >> forwarding_stack >> msg;
CAF_LOG_DEBUG(CAF_ARG(forwarding_stack) << CAF_ARG(msg));
callee_.deliver(hdr.source_node, hdr.source_actor,
hdr.dest_node, hdr.dest_actor,
if (hdr.has(header::named_receiver_flag))
callee_.deliver(hdr.source_node, hdr.source_actor, receiver_name,
message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg);
else
callee_.deliver(hdr.source_node, hdr.source_actor, hdr.dest_actor,
message_id::from_integer_value(hdr.operation_data),
forwarding_stack, msg);
break;
}
case message_type::announce_proxy_instance:
case message_type::announce_proxy:
callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
break;
case message_type::kill_proxy_instance: {
case message_type::kill_proxy: {
if (! payload_valid())
return err();
binary_deserializer bd{ctx, *payload};
......@@ -311,7 +318,7 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
auto writer = make_callback([&](serializer& sink) {
sink << forwarding_stack << msg;
});
header hdr{message_type::dispatch_message, 0, mid.integer_value(),
header hdr{message_type::dispatch_message, 0, 0, mid.integer_value(),
sender ? sender->node() : this_node(), receiver->node(),
sender ? sender->id() : invalid_actor_id, receiver->id()};
write(ctx, path->wr_buf, hdr, &writer);
......@@ -320,65 +327,30 @@ bool instance::dispatch(execution_unit* ctx, const strong_actor_ptr& sender,
return true;
}
void instance::write(execution_unit* ctx,
buffer_type& buf,
message_type operation,
uint32_t* payload_len,
uint64_t operation_data,
const node_id& source_node,
const node_id& dest_node,
actor_id source_actor,
actor_id dest_actor,
payload_writer* pw) {
CAF_LOG_TRACE(CAF_ARG(operation) << CAF_ARG(operation_data)
<< CAF_ARG(source_node) << CAF_ARG(dest_node)
<< CAF_ARG(source_actor) << CAF_ARG(dest_actor));
if (! pw) {
CAF_LOG_DEBUG("send message without payload");
uint32_t zero = 0;
binary_serializer bs{ctx, buf};
bs << source_node
<< dest_node
<< source_actor
<< dest_actor
<< zero
<< operation
<< operation_data;
return;
}
// reserve space in the buffer to write the payload later on
auto wr_pos = buf.size();
void instance::write(execution_unit* ctx, buffer_type& buf,
header& hdr, payload_writer* pw) {
CAF_LOG_TRACE(CAF_ARG(hdr));
try {
if (pw) {
auto pos = buf.size();
// write payload first (skip first 72 bytes and write header later)
char placeholder[basp::header_size];
buf.insert(buf.end(), std::begin(placeholder), std::end(placeholder));
auto pl_pos = buf.size();
try { // lifetime scope of first serializer (write payload)
binary_serializer bs{ctx, buf};
(*pw)(bs);
auto plen = buf.size() - pos - basp::header_size;
CAF_ASSERT(plen <= std::numeric_limits<uint32_t>::max());
hdr.payload_len = static_cast<uint32_t>(plen);
stream_serializer<charbuf> out{ctx, buf.data() + pos, basp::header_size};
out << hdr;
} else {
binary_serializer bs{ctx, buf};
bs << hdr;
}
}
catch (std::exception& e) {
printf("EXCEPTION: %s\n", e.what());
CAF_LOG_ERROR(CAF_ARG(e.what()));
}
// write broker message to the reserved space
stream_serializer<charbuf> bs2{ctx, buf.data() + wr_pos, pl_pos - wr_pos};
auto plen = static_cast<uint32_t>(buf.size() - pl_pos);
bs2 << source_node
<< dest_node
<< source_actor
<< dest_actor
<< plen
<< operation
<< operation_data;
CAF_LOG_DEBUG("wrote payload" << CAF_ARG(plen));
if (payload_len)
*payload_len = plen;
}
void instance::write(execution_unit* ctx, buffer_type& buf,
header& hdr, payload_writer* pw) {
CAF_LOG_TRACE(CAF_ARG(hdr));
write(ctx, buf, hdr.operation, &hdr.payload_len, hdr.operation_data,
hdr.source_node, hdr.dest_node, hdr.source_actor, hdr.dest_actor, pw);
}
void instance::write_server_handshake(execution_unit* ctx,
......@@ -399,7 +371,7 @@ void instance::write_server_handshake(execution_unit* ctx,
sink << i << pa->second;
}
});
header hdr{message_type::server_handshake, 0, version,
header hdr{message_type::server_handshake, 0, 0, version,
this_node_, invalid_node_id,
pa && pa->first ? pa->first->id() : invalid_actor_id,
invalid_actor_id};
......@@ -410,20 +382,27 @@ void instance::write_client_handshake(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side) {
CAF_LOG_TRACE(CAF_ARG(remote_side));
write(ctx, buf, message_type::client_handshake, nullptr, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id);
header hdr{message_type::client_handshake, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr);
}
void instance::write_kill_proxy_instance(execution_unit* ctx,
buffer_type& buf,
const node_id& dest_node,
actor_id aid,
void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid));
header hdr{message_type::announce_proxy, 0, 0, 0,
this_node_, dest_node, invalid_actor_id, aid};
write(ctx, buf, hdr);
}
void instance::write_kill_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid,
const error& rsn) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid) << CAF_ARG(rsn));
auto writer = make_callback([&](serializer& sink) {
sink << rsn;
});
header hdr{message_type::kill_proxy_instance, 0, 0,
header hdr{message_type::kill_proxy, 0, 0, 0,
this_node_, dest_node, aid, invalid_actor_id};
write(ctx, buf, hdr, &writer);
}
......@@ -432,8 +411,9 @@ void instance::write_heartbeat(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side) {
CAF_LOG_TRACE(CAF_ARG(remote_side));
write(ctx, buf, message_type::heartbeat, nullptr, 0,
this_node_, remote_side, invalid_actor_id, 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);
}
} // namespace basp
......
......@@ -31,9 +31,9 @@ std::string to_string(message_type x) {
return "client_handshake";
case message_type::dispatch_message:
return "dispatch_message";
case message_type::announce_proxy_instance:
case message_type::announce_proxy:
return "announce_proxy_instance";
case message_type::kill_proxy_instance:
case message_type::kill_proxy:
return "kill_proxy_instance";
case message_type::heartbeat:
return "heartbeat";
......
This diff is collapsed.
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