Commit cd1fc630 authored by Joseph Noir's avatar Joseph Noir

WIP Removing indirection from routing table

parent 77350c81
......@@ -27,6 +27,7 @@ set (LIBCAF_IO_SRCS
src/dgram_doorman.cpp
src/dgram_acceptor_manager.cpp
src/dgram_communicator_manager.cpp
src/visitors.cpp
# BASP files
src/header.cpp
src/message_type.cpp
......
......@@ -82,12 +82,12 @@ public:
/// 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,
bool was_known_indirectly) = 0;
virtual void learned_new_node_directly(const node_id& nid) = 0;
// bool was_known_indirectly) = 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_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`
virtual void handle_heartbeat(const node_id& nid) = 0;
......@@ -230,6 +230,8 @@ private:
published_actor_map published_actors_;
node_id this_node_;
callee& callee_;
flush_visitor flush_;
wr_buf_visitor wr_buf_;
};
/// @}
......
......@@ -86,11 +86,11 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
void learned_new_node(const node_id& nid);
// inherited from basp::instance::listener
void learned_new_node_directly(const node_id& nid,
bool was_known_indirectly_before) override;
void learned_new_node_directly(const node_id& nid) override;
// bool was_known_indirectly_before) override;
// inherited from basp::instance::listener
void learned_new_node_indirectly(const node_id& nid) override;
// void learned_new_node_indirectly(const node_id& nid) override;
void handle_heartbeat(const node_id&) override {
// nop
......@@ -112,13 +112,14 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
optional<response_promise> callback;
// TODO: ordering
// TODO: reliability
};
};
void set_context(connection_handle hdl);
void set_context(dgram_scribe_handle hdl);
// visitors to handle variants
wr_buf_visitor wr_buf_of_hdl;
purge_visitor purge_state_vis;
wr_buf_visitor wr_buf_vis;
// pointer to ourselves
broker* self;
......@@ -161,6 +162,11 @@ public:
behavior make_behavior() override;
proxy_registry* proxy_registry_ptr() override;
resume_result resume(execution_unit*, size_t) override;
private:
// visitors
addr_visitor addr_;
port_visitor port_;
};
} // namespace io
......
......@@ -28,10 +28,12 @@ class broker;
class doorman;
class middleman;
class basp_broker;
class dgram_scribe;
class dgram_doorman;
class receive_policy;
class abstract_broker;
class dgram_scribe;
struct basp_broker_state;
template <class... Sigs>
class typed_broker;
......
......@@ -20,27 +20,53 @@
#ifndef CAF_IO_VISITORS_HPP
#define CAF_IO_VISITORS_HPP
#include "caf/io/fwd.hpp"
#include "caf/io/abstract_broker.hpp"
namespace caf {
namespace io {
struct wr_buf_visitor {
wr_buf_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
using result_type = std::vector<char>&;
wr_buf_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->wr_buf(hdl); }
abstract_broker* ptr;
};
struct flush_visitor {
flush_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
using result_type = std::vector<char>&;
flush_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->flush(hdl); }
abstract_broker* ptr;
};
struct addr_visitor {
using result_type = std::string;
addr_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->remote_addr(hdl); }
abstract_broker* ptr;
};
struct port_visitor {
using result_type = uint16_t;
port_visitor(abstract_broker* ptr) : ptr{ptr} { /* nop */ }
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->remote_port(hdl); }
abstract_broker* ptr;
};
struct purge_visitor {
using result_type = void;
purge_visitor(basp_broker_state* ptr) : state{ptr} { /* nop */ }
result_type operator()(const connection_handle& hdl);
result_type operator()(const dgram_scribe_handle& hdl);
basp_broker_state* state;
};
} // namespace io
} // namespace caf
......
......@@ -50,7 +50,8 @@ const char* basp_broker_state::name = "basp_broker";
basp_broker_state::basp_broker_state(broker* selfptr)
: basp::instance::callee(selfptr->system(),
static_cast<proxy_registry::backend&>(*this)),
wr_buf_of_hdl(selfptr),
purge_state_vis(this),
wr_buf_vis(selfptr),
self(selfptr),
instance(selfptr, *this) {
CAF_ASSERT(this_node() != none);
......@@ -70,10 +71,13 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
// this member function is being called whenever we deserialize a
// payload received from a remote node; if a remote node A sends
// us a handle to a third node B, then we assume that A offers a route to B
/*
* TODO: Some other notification?
if (nid != this_context->id
&& instance.tbl().lookup_direct(nid) == invalid_connection_handle
&& instance.tbl().lookup_hdl(nid)
&& instance.tbl().add_indirect(this_context->id, nid))
learned_new_node_indirectly(nid);
*/
// we need to tell remote side we are watching this actor now;
// use a direct route if possible, i.e., when talking to a third node
auto path = instance.tbl().lookup(nid);
......@@ -106,8 +110,7 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
<< CAF_ARG(nid) << CAF_ARG(aid));
// tell remote side we are monitoring this actor now
instance.write_announce_proxy(self->context(),
apply_visitor(wr_buf_of_hdl,
this_context->hdl),
apply_visitor(wr_buf_vis, this_context->hdl),
nid, aid);
instance.tbl().flush(*path);
mm->notify<hook::new_remote_actor>(res);
......@@ -149,9 +152,10 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
void basp_broker_state::purge_state(const node_id& nid) {
CAF_LOG_TRACE(CAF_ARG(nid));
auto hdl = instance.tbl().lookup_direct(nid);
if (hdl == invalid_connection_handle)
auto hdl = instance.tbl().lookup_hdl(nid);
if (!hdl)
return;
/*
auto i = tcp_ctx.find(hdl);
if (i != tcp_ctx.end()) {
auto& ref = i->second;
......@@ -161,6 +165,8 @@ void basp_broker_state::purge_state(const node_id& nid) {
}
tcp_ctx.erase(i);
}
*/
apply_visitor(purge_state_vis, *hdl);
proxies().erase(nid);
}
......@@ -363,14 +369,16 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
instance.flush(*path);
}
void basp_broker_state::learned_new_node_directly(const node_id& nid,
bool was_indirectly_before) {
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);
// if (!was_indirectly_before)
learned_new_node(nid);
}
/*
* TODO: Can I just remove this?
void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
CAF_ASSERT(this_context != nullptr);
CAF_LOG_TRACE(CAF_ARG(nid));
......@@ -448,6 +456,7 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
instance.write(self->context(), path->wr_buf, hdr, &writer);
instance.flush(*path);
}
*/
void basp_broker_state::set_context(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
......@@ -486,7 +495,6 @@ void basp_broker_state::set_context(dgram_scribe_handle hdl) {
).first;
}
this_context = &i->second;
}
/******************************************************************************
......@@ -494,7 +502,9 @@ void basp_broker_state::set_context(dgram_scribe_handle hdl) {
******************************************************************************/
basp_broker::basp_broker(actor_config& cfg)
: stateful_actor<basp_broker_state, broker>(cfg) {
: stateful_actor<basp_broker_state, broker>(cfg),
addr_(this),
port_(this) {
// nop
}
......@@ -632,7 +642,7 @@ behavior basp_broker::make_behavior() {
// a connection, we also could try to reach this node via other
// hops to be resilient to (rare) network failures or if a
// node is reachable via several interfaces and only one fails
auto nid = state.instance.tbl().lookup_direct(msg.handle);
auto nid = state.instance.tbl().lookup_node(msg.handle);
// tell BASP instance we've lost connection
state.instance.handle_node_shutdown(nid);
CAF_ASSERT(nid == none
......@@ -766,10 +776,10 @@ behavior basp_broker::make_behavior() {
-> std::tuple<node_id, std::string, uint16_t> {
std::string addr;
uint16_t port = 0;
auto hdl = state.instance.tbl().lookup_direct(x);
if (hdl != invalid_connection_handle) {
addr = remote_addr(hdl);
port = remote_port(hdl);
auto hdl = state.instance.tbl().lookup_hdl(x);
if (hdl) {
addr = apply_visitor(addr_, *hdl);
port = apply_visitor(port_, *hdl);
}
return std::make_tuple(x, std::move(addr), port);
},
......
......@@ -42,7 +42,9 @@ instance::callee::~callee() {
instance::instance(abstract_broker* parent, callee& lstnr)
: tbl_(parent),
this_node_(parent->system().node()),
callee_(lstnr) {
callee_(lstnr),
flush_(parent),
wr_buf_(parent) {
CAF_ASSERT(this_node_ != none);
}
......@@ -142,7 +144,7 @@ connection_state instance::handle(execution_unit* ctx,
return err();
}
// close this connection if we already have a direct connection
if (tbl_.lookup_direct(hdr.source_node) != invalid_connection_handle) {
if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("close connection since we already have a "
"direct connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
......@@ -150,8 +152,8 @@ connection_state instance::handle(execution_unit* ctx,
}
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add_direct(dm.handle, hdr.source_node);
auto was_indirect = tbl_.erase_indirect(hdr.source_node);
tbl_.add(dm.handle, hdr.source_node);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
// write handshake as client in response
auto path = tbl_.lookup(hdr.source_node);
if (!path) {
......@@ -159,13 +161,13 @@ connection_state instance::handle(execution_unit* ctx,
return err();
}
write_client_handshake(ctx, path->wr_buf, hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
callee_.learned_new_node_directly(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs);
flush(*path);
break;
}
case message_type::client_handshake: {
if (tbl_.lookup_direct(hdr.source_node) != invalid_connection_handle) {
if (tbl_.lookup_hdl(hdr.source_node)) {
CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node));
break;
......@@ -186,9 +188,9 @@ connection_state instance::handle(execution_unit* ctx,
}
// add direct route to this node and remove any indirect entry
CAF_LOG_INFO("new direct connection:" << CAF_ARG(hdr.source_node));
tbl_.add_direct(dm.handle, hdr.source_node);
auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node, was_indirect);
tbl_.add(dm.handle, hdr.source_node);
// auto was_indirect = tbl_.erase_indirect(hdr.source_node);
callee_.learned_new_node_directly(hdr.source_node);
break;
}
case message_type::dispatch_message: {
......@@ -196,13 +198,16 @@ connection_state instance::handle(execution_unit* ctx,
return err();
// 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(dm.handle);
auto last_hop = tbl_.lookup_node(dm.handle);
/*
* TODO: if we don't need this anymore, what does the rest do here?
if (hdr.source_node != none
&& hdr.source_node != this_node_
&& last_hop != hdr.source_node
&& tbl_.lookup_direct(hdr.source_node) == invalid_connection_handle
&& tbl_.add_indirect(last_hop, hdr.source_node))
&& tbl_.lookup_hdl(hdr.source_node))
&& 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;
......@@ -263,8 +268,8 @@ void instance::handle_heartbeat(execution_unit* ctx) {
CAF_LOG_TRACE("");
for (auto& kvp: tbl_.direct_by_hdl_) {
CAF_LOG_TRACE(CAF_ARG(kvp.first) << CAF_ARG(kvp.second));
write_heartbeat(ctx, tbl_.parent_->wr_buf(kvp.first), kvp.second);
tbl_.parent_->flush(kvp.first);
write_heartbeat(ctx, apply_visitor(wr_buf_, kvp.first), kvp.second);
apply_visitor(flush_, kvp.first);
}
}
......@@ -280,15 +285,15 @@ void instance::handle_node_shutdown(const node_id& affected_node) {
tbl_.erase(affected_node, cb);
}
optional<routing_table::route> instance::lookup(const node_id& target) {
optional<routing_table::endpoint> instance::lookup(const node_id& target) {
return tbl_.lookup(target);
}
void instance::flush(const routing_table::route& path) {
void instance::flush(const routing_table::endpoint& path) {
tbl_.flush(path);
}
void instance::write(execution_unit* ctx, const routing_table::route& r,
void instance::write(execution_unit* ctx, const routing_table::endpoint& r,
header& hdr, payload_writer* writer) {
CAF_LOG_TRACE(CAF_ARG(hdr));
CAF_ASSERT(hdr.payload_len == 0 || writer != nullptr);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2016 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* 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. *
******************************************************************************/
#include "caf/io/visitors.hpp"
namespace caf {
namespace io {
purge_visitor::result_type purge_visitor::operator()(const connection_handle& h) {
auto i = state->tcp_ctx.find(h);
if (i != state->tcp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state->tcp_ctx.erase(i);
}
}
purge_visitor::result_type
purge_visitor::operator()(const dgram_scribe_handle& h) {
auto i = state->udp_ctx.find(h);
if (i != state->udp_ctx.end()) {
auto& ref = i->second;
if (ref.callback) {
CAF_LOG_DEBUG("connection closed during handshake");
ref.callback->deliver(sec::disconnect_during_handshake);
}
state->udp_ctx.erase(i);
}
}
} // namespace io
} // namespace caf
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