Commit 9d87f63e authored by Dominik Charousset's avatar Dominik Charousset Committed by Dominik Charousset

Factor dispatching logic into new helper class

parent 6e9f806d
......@@ -68,18 +68,6 @@ public:
/// for one of our local actors.
virtual void proxy_announced(const node_id& nid, actor_id aid) = 0;
/// Called for each `dispatch_message` without `named_receiver_flag`.
virtual void deliver(const node_id& source_node, actor_id source_actor,
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;
/// Called whenever BASP learns the ID of a remote node
/// to which it does not have a direct connection.
virtual void learned_new_node_directly(const node_id& nid,
......@@ -92,6 +80,9 @@ public:
/// Called if a heartbeat was received from `nid`
virtual void handle_heartbeat() = 0;
/// Returns the current CAF scheduler context.
virtual execution_unit* current_execution_unit() = 0;
/// Returns the actor namespace associated to this BASP protocol instance.
proxy_registry& proxies() {
return namespace_;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#pragma once
#include <vector>
#include "caf/actor_control_block.hpp"
#include "caf/actor_proxy.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/config.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/execution_unit.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/logger.hpp"
#include "caf/message.hpp"
#include "caf/message_id.hpp"
#include "caf/node_id.hpp"
namespace caf {
namespace io {
namespace basp {
template <class Subtype>
class remote_message_handler {
public:
void handle_remote_message(execution_unit* ctx) {
CAF_LOG_TRACE("");
// Local variables.
auto& dref = static_cast<Subtype&>(*this);
auto& sys = *dref.system_;
strong_actor_ptr src;
strong_actor_ptr dst;
std::vector<strong_actor_ptr> stages;
message msg;
auto mid = make_message_id(dref.hdr_.operation_data);
binary_deserializer source{ctx, dref.payload_};
// Get the local receiver.
if (dref.hdr_.has(basp::header::named_receiver_flag))
dst = sys.registry().get(static_cast<atom_value>(dref.hdr_.dest_actor));
else
dst = sys.registry().get(dref.hdr_.dest_actor);
// Short circuit it we already know there's nothing to do.
if (dst == nullptr && !mid.is_request()) {
CAF_LOG_INFO("drop asynchronous remote message: unknown destination");
return;
}
// Deserialize source and destination node for routed messages.
if (dref.hdr_.operation == basp::message_type::routed_message) {
node_id src_node;
node_id dst_node;
if (auto err = source(src_node, dst_node)) {
CAF_LOG_ERROR("cannot read source and destination of remote message");
return;
}
CAF_ASSERT(dst_node == sys.node());
if (dref.hdr_.source_actor != 0) {
src = src_node == sys.node()
? sys.registry().get(dref.hdr_.source_actor)
: dref.proxies_->get_or_put(src_node, dref.hdr_.source_actor);
}
} else {
CAF_ASSERT(dref.hdr_.operation == basp::message_type::direct_message);
src = dref.proxies_->get_or_put(dref.last_hop_, dref.hdr_.source_actor);
}
// Send errors for dropped requests.
if (dst == nullptr) {
CAF_ASSERT(mid.is_request());
CAF_LOG_INFO("drop remote request: unknown destination");
detail::sync_request_bouncer srb{exit_reason::remote_link_unreachable};
srb(src, mid);
return;
}
// Get the remainder of the message.
if (auto err = source(stages, msg)) {
CAF_LOG_ERROR("cannot read stages and content of remote message");
return;
}
// Intercept link messages. Forwarding actor proxies signalize linking
// by sending link_atom/unlink_atom message with src == dest.
if (msg.type_token() == make_type_token<atom_value, strong_actor_ptr>()) {
const auto& ptr = msg.get_as<strong_actor_ptr>(1);
switch (static_cast<uint64_t>(msg.get_as<atom_value>(0))) {
default:
break;
case link_atom::value.uint_value(): {
if (ptr != nullptr)
static_cast<actor_proxy*>(ptr->get())->add_link(dst->get());
else
CAF_LOG_WARNING("received link message with invalid target");
return;
}
case unlink_atom::value.uint_value(): {
if (ptr != nullptr)
static_cast<actor_proxy*>(ptr->get())->remove_link(dst->get());
else
CAF_LOG_DEBUG("received unlink message with invalid target");
return;
}
}
}
// Ship the message.
dst->enqueue(make_mailbox_element(std::move(src), mid, std::move(stages),
std::move(msg)),
ctx);
}
};
} // namespace basp
} // namespace io
} // namespace caf
......@@ -61,21 +61,6 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::callee
void proxy_announced(const node_id& nid, actor_id aid) override;
// inherited from basp::instance::callee
void 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) override;
// inherited from basp::instance::callee
void 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) override;
// called from both overriden functions
void 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);
// performs bookkeeping such as managing `spawn_servers`
void learned_new_node(const node_id& nid);
......@@ -95,6 +80,9 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
// inherited from basp::instance::callee
void handle_heartbeat() override;
// inherited from basp::instance::callee
execution_unit* current_execution_unit() override;
/// Sets `this_context` by either creating or accessing state for `hdl`.
void set_context(connection_handle hdl);
......
......@@ -173,96 +173,6 @@ void basp_broker_state::handle_down_msg(down_msg& dm) {
monitored_actors.erase(i);
}
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_aid) << CAF_ARG(msg) << CAF_ARG(mid));
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);
// Intercept link messages. Forwarding actor proxies signalize linking
// by sending link_atom/unlink_atom message with src = dest.
if (msg.type_token() == make_type_token<atom_value, strong_actor_ptr>()) {
switch (static_cast<uint64_t>(msg.get_as<atom_value>(0))) {
default:
break;
case link_atom::value.uint_value(): {
if (src_nid != this_node()) {
CAF_LOG_WARNING("received link message for another node");
return;
}
auto ptr = msg.get_as<strong_actor_ptr>(1);
if (!ptr) {
CAF_LOG_WARNING("received link message with invalid target");
return;
}
if (!src) {
CAF_LOG_DEBUG("received link for invalid actor, report error");
anon_send(actor_cast<actor>(ptr),
make_error(sec::remote_linking_failed));
return;
}
static_cast<actor_proxy*>(ptr->get())->add_link(src->get());
return;
}
case unlink_atom::value.uint_value(): {
if (src_nid != this_node()) {
CAF_LOG_WARNING("received unlink message for an other node");
return;
}
const auto& ptr = msg.get_as<strong_actor_ptr>(1);
if (!ptr) {
CAF_LOG_DEBUG("received unlink message with invalid target");
return;
}
if (!src) {
CAF_LOG_DEBUG("received unlink for invalid actor, report error");
return;
}
static_cast<actor_proxy*>(ptr->get())->remove_link(src->get());
return;
}
}
}
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,
invalid_actor_id,
mid, msg);
if (mid.is_request() && src != nullptr) {
detail::sync_request_bouncer srb{rsn};
srb(src, mid);
}
return;
}
self->parent().notify<hook::message_received>(src_nid, src, dest, mid, msg);
dest->enqueue(make_mailbox_element(std::move(src), mid, std::move(stages),
std::move(msg)),
nullptr);
}
void basp_broker_state::learned_new_node(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(nid));
if (spawn_servers.count(nid) > 0) {
......@@ -321,14 +231,12 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
void basp_broker_state::learned_new_node_directly(const node_id& nid,
bool was_indirectly_before) {
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid));
if (!was_indirectly_before)
learned_new_node(nid);
}
void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid));
learned_new_node(nid);
if (!automatic_connections)
......@@ -403,6 +311,10 @@ void basp_broker_state::handle_heartbeat() {
// nop
}
execution_unit* basp_broker_state::current_execution_unit() {
return self->context();
}
/******************************************************************************
* basp_broker *
******************************************************************************/
......
......@@ -22,6 +22,7 @@
#include "caf/binary_deserializer.hpp"
#include "caf/binary_serializer.hpp"
#include "caf/defaults.hpp"
#include "caf/io/basp/remote_message_handler.hpp"
#include "caf/io/basp/version.hpp"
#include "caf/streambuf.hpp"
......@@ -365,29 +366,6 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
callee_.learned_new_node_directly(source_node, was_indirect);
break;
}
case message_type::direct_message: {
// Deserialize payload.
binary_deserializer bd{ctx, *payload};
std::vector<strong_actor_ptr> forwarding_stack;
message msg;
if (auto err = bd(forwarding_stack, msg)) {
CAF_LOG_WARNING("unable to deserialize payload of direct message:"
<< ctx->system().render(err));
return false;
}
// Dispatch message to callee_.
auto source_node = tbl_.lookup_direct(hdl);
if (hdr.has(header::named_receiver_flag))
callee_.deliver(source_node, hdr.source_actor,
static_cast<atom_value>(hdr.dest_actor),
make_message_id(hdr.operation_data), forwarding_stack,
msg);
else
callee_.deliver(source_node, hdr.source_actor, hdr.dest_actor,
make_message_id(hdr.operation_data), forwarding_stack,
msg);
break;
}
case message_type::routed_message: {
// Deserialize payload.
binary_deserializer bd{ctx, *payload};
......@@ -403,29 +381,32 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
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);
}
// fall through
case message_type::direct_message: {
struct handler : remote_message_handler<handler> {
handler(proxy_registry* proxies, actor_system* system, node_id last_hop,
basp::header& hdr, buffer_type& payload)
: proxies_(proxies),
system_(system),
last_hop_(std::move(last_hop)),
hdr_(hdr),
payload_(payload) {
// nop
}
proxy_registry* proxies_;
actor_system* system_;
node_id last_hop_;
basp::header& hdr_;
buffer_type& payload_;
};
handler f{&proxies(), &system(), tbl_.lookup_direct(hdl), hdr, *payload};
f.handle_remote_message(callee_.current_execution_unit());
break;
}
case message_type::monitor_message: {
......
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