Commit 77350c81 authored by Joseph Noir's avatar Joseph Noir

Removing indirection from routing table

parent e327c64e
......@@ -130,14 +130,14 @@ public:
void handle_node_shutdown(const node_id& affected_node);
/// Returns a route to `target` or `none` on error.
optional<routing_table::route> lookup(const node_id& target);
optional<routing_table::endpoint> lookup(const node_id& target);
/// Flushes the underlying buffer of `path`.
void flush(const routing_table::route& path);
void flush(const routing_table::endpoint& path);
/// Sends a BASP message and implicitly flushes the output buffer of `r`.
/// This function will update `hdr.payload_len` if a payload was written.
void write(execution_unit* ctx, const routing_table::route& r,
void write(execution_unit* ctx, const routing_table::endpoint& r,
header& hdr, payload_writer* writer = nullptr);
/// Adds a new actor to the map of published actors.
......
......@@ -26,6 +26,7 @@
#include "caf/node_id.hpp"
#include "caf/callback.hpp"
#include "caf/io/visitors.hpp"
#include "caf/io/abstract_broker.hpp"
#include "caf/io/basp/buffer_type.hpp"
......@@ -34,21 +35,55 @@ namespace caf {
namespace io {
namespace basp {
struct hash_visitor {
using result_type = size_t;
template <class T>
result_type operator()(const T& hdl) const {
std::hash<T> f;
return f(hdl);
}
};
} // namespace basp
} // namespace io
} // namespace caf
namespace std {
template<>
struct hash<caf::variant<caf::io::connection_handle,
caf::io::dgram_scribe_handle>> {
size_t operator()(const caf::variant<caf::io::connection_handle,
caf::io::dgram_scribe_handle>& hdl) const {
caf::io::basp::hash_visitor vis;
return caf::apply_visitor(vis, hdl);
}
};
} // namespace std
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// Stores routing information for a single broker participating as
/// BASP peer and provides both direct and indirect paths.
class routing_table {
public:
using endpoint_handle = variant<connection_handle, dgram_scribe_handle>;
explicit routing_table(abstract_broker* parent);
virtual ~routing_table();
/// Describes a routing path to a node.
struct route {
struct endpoint {
buffer_type& wr_buf;
const node_id& next_hop;
connection_handle hdl;
endpoint_handle hdl;
};
/// Describes a function object for erase operations that
......@@ -56,41 +91,53 @@ public:
using erase_callback = callback<const node_id&>;
/// Returns a route to `target` or `none` on error.
optional<route> lookup(const node_id& target);
optional<endpoint> lookup(const node_id& target);
/// Returns the ID of the peer connected via `hdl` or
/// `none` if `hdl` is unknown.
node_id lookup_direct(const connection_handle& hdl) const;
//node_id lookup_direct(const endpoint_handle& hdl) const;
node_id lookup_node(const endpoint_handle& hdl) const;
/// Returns the handle offering a direct connection to `nid` or
/// `invalid_connection_handle` if no direct connection to `nid` exists.
connection_handle lookup_direct(const node_id& nid) const;
/// `invalid_endpoint_handle` if no direct connection to `nid` exists.
//endpoint_handle lookup_direct(const node_id& nid) const;
optional<routing_table::endpoint_handle> lookup_hdl(const node_id& nid) const;
/*
/// Returns the next hop that would be chosen for `nid`
/// or `none` if there's no indirect route to `nid`.
node_id lookup_indirect(const node_id& nid) const;
*/
/// Flush output buffer for `r`.
void flush(const route& r);
void flush(const endpoint& r);
/// Adds a new direct route to the table.
/// @pre `hdl != invalid_connection_handle && nid != none`
void add_direct(const connection_handle& hdl, const node_id& dest);
/// @pre `hdl != invalid_endpoint_handle && nid != none`
//void add_direct(const endpoint_handle& hdl, const node_id& dest);
void add(const endpoint_handle& hdl, const node_id& dest);
/*
/// Adds a new indirect route to the table.
bool add_indirect(const node_id& hop, const node_id& dest);
*/
/*
/// Blacklist the route to `dest` via `hop`.
void blacklist(const node_id& hop, const node_id& dest);
*/
/// Removes a direct connection and calls `cb` for any node
/// that became unreachable as a result of this operation,
/// including the node that is assigned as direct path for `hdl`.
void erase_direct(const connection_handle& hdl, erase_callback& cb);
// void erase_direct(const endpoint_handle& hdl, erase_callback& cb);
void erase(const endpoint_handle& hdl, erase_callback& cb);
/*
/// Removes any entry for indirect connection to `dest` and returns
/// `true` if `dest` had an indirect route, otherwise `false`.
bool erase_indirect(const node_id& dest);
*/
/// Queries whether `dest` is reachable.
bool reachable(const node_id& dest);
......@@ -113,14 +160,20 @@ public:
using node_id_set = std::unordered_set<node_id>;
/*
using indirect_entries = std::unordered_map<node_id, // dest
node_id_set>; // hop
*/
abstract_broker* parent_;
std::unordered_map<connection_handle, node_id> direct_by_hdl_;
std::unordered_map<node_id, connection_handle> direct_by_nid_;
std::unordered_map<endpoint_handle, node_id> direct_by_hdl_;
std::unordered_map<node_id, endpoint_handle> direct_by_nid_;
/*
indirect_entries indirect_;
indirect_entries blacklist_;
*/
// visitors for endpoint_handles
wr_buf_visitor wr_buf_;
flush_visitor flush_;
};
/// @}
......
......@@ -36,6 +36,7 @@
#include "caf/io/basp/all.hpp"
#include "caf/io/broker.hpp"
#include "caf/io/visitors.hpp"
#include "caf/io/typed_broker.hpp"
namespace caf {
......@@ -116,13 +117,8 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
void set_context(connection_handle hdl);
void set_context(dgram_scribe_handle hdl);
struct wr_buf_visitor {
wr_buf_visitor(broker* ptr) : ptr{ptr} { /* nop */ }
using result_type = std::vector<char>&;
result_type operator()(connection_handle hdl) { return ptr->wr_buf(hdl); }
result_type operator()(dgram_scribe_handle hdl) { return ptr->wr_buf(hdl); }
broker* ptr;
} wr_buf_of_hdl;
// visitors to handle variants
wr_buf_visitor wr_buf_of_hdl;
// pointer to ourselves
broker* self;
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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. *
******************************************************************************/
#ifndef CAF_IO_VISITORS_HPP
#define CAF_IO_VISITORS_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>&;
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>&;
template <typename Handle>
result_type operator()(const Handle& hdl) { return ptr->flush(hdl); }
abstract_broker* ptr;
};
} // namespace io
} // namespace caf
#endif // CAF_IO_VISITORS_HPP
......@@ -56,7 +56,7 @@ connection_state instance::handle(execution_unit* ctx,
callee_.purge_state(nid);
return none;
});
tbl_.erase_direct(dm.handle, cb);
tbl_.erase(dm.handle, cb);
return close_connection;
};
std::vector<char>* payload = nullptr;
......
......@@ -25,7 +25,10 @@ namespace caf {
namespace io {
namespace basp {
routing_table::routing_table(abstract_broker* parent) : parent_(parent) {
routing_table::routing_table(abstract_broker* parent)
: parent_(parent),
wr_buf_(parent),
flush_(parent) {
// nop
}
......@@ -33,11 +36,12 @@ routing_table::~routing_table() {
// nop
}
optional<routing_table::route> routing_table::lookup(const node_id& target) {
auto hdl = lookup_direct(target);
if (hdl != invalid_connection_handle)
return route{parent_->wr_buf(hdl), target, hdl};
optional<routing_table::endpoint> routing_table::lookup(const node_id& target) {
auto hdl = lookup_hdl(target);
if (hdl)
return endpoint{apply_visitor(wr_buf_, *hdl), target, *hdl};
// pick first available indirect route
/*
auto i = indirect_.find(target);
if (i != indirect_.end()) {
auto& hops = i->second;
......@@ -49,22 +53,24 @@ optional<routing_table::route> routing_table::lookup(const node_id& target) {
else
hops.erase(hops.begin());
}
}
}*/
return none;
}
void routing_table::flush(const route& r) {
parent_->flush(r.hdl);
void routing_table::flush(const endpoint& r) {
apply_visitor(flush_, r.hdl);
}
node_id routing_table::lookup_direct(const connection_handle& hdl) const {
node_id routing_table::lookup_node(const endpoint_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, none);
}
connection_handle routing_table::lookup_direct(const node_id& nid) const {
return get_opt(direct_by_nid_, nid, invalid_connection_handle);
optional<routing_table::endpoint_handle>
routing_table::lookup_hdl(const node_id& nid) const {
return get_opt(direct_by_nid_, nid, none);
}
/*
node_id routing_table::lookup_indirect(const node_id& nid) const {
auto i = indirect_.find(nid);
if (i == indirect_.end())
......@@ -73,7 +79,9 @@ node_id routing_table::lookup_indirect(const node_id& nid) const {
return none;
return *i->second.begin();
}
*/
/*
void routing_table::blacklist(const node_id& hop, const node_id& dest) {
blacklist_[dest].emplace(hop);
auto i = indirect_.find(dest);
......@@ -83,9 +91,10 @@ void routing_table::blacklist(const node_id& hop, const node_id& dest) {
if (i->second.empty())
indirect_.erase(i);
}
*/
void routing_table::erase_direct(const connection_handle& hdl,
erase_callback& cb) {
//void routing_table::erase_direct(const endpoint_handle& hdl,
void routing_table::erase(const endpoint_handle& hdl, erase_callback& cb) {
auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end())
return;
......@@ -95,6 +104,7 @@ void routing_table::erase_direct(const connection_handle& hdl,
direct_by_hdl_.erase(i);
}
/*
bool routing_table::erase_indirect(const node_id& dest) {
auto i = indirect_.find(dest);
if (i == indirect_.end())
......@@ -105,9 +115,9 @@ bool routing_table::erase_indirect(const node_id& dest) {
indirect_.erase(i);
return true;
}
*/
void routing_table::add_direct(const connection_handle& hdl,
const node_id& nid) {
void routing_table::add(const endpoint_handle& hdl, const node_id& nid) {
CAF_ASSERT(direct_by_hdl_.count(hdl) == 0);
CAF_ASSERT(direct_by_nid_.count(nid) == 0);
direct_by_hdl_.emplace(hdl, nid);
......@@ -115,6 +125,7 @@ void routing_table::add_direct(const connection_handle& hdl,
parent_->parent().notify<hook::new_connection_established>(nid);
}
/*
bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
auto i = blacklist_.find(dest);
if (i == blacklist_.end() || i->second.count(hop) == 0) {
......@@ -126,14 +137,16 @@ bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
}
return false; // blacklisted
}
*/
bool routing_table::reachable(const node_id& dest) {
return direct_by_nid_.count(dest) > 0 || indirect_.count(dest) > 0;
return direct_by_nid_.count(dest) > 0; // || indirect_.count(dest) > 0;
}
size_t routing_table::erase(const node_id& dest, erase_callback& cb) {
cb(dest);
size_t res = 0;
/*
auto i = indirect_.find(dest);
if (i != indirect_.end()) {
res = i->second.size();
......@@ -143,8 +156,9 @@ size_t routing_table::erase(const node_id& dest, erase_callback& cb) {
}
indirect_.erase(i);
}
auto hdl = lookup_direct(dest);
if (hdl != invalid_connection_handle) {
*/
auto hdl = lookup_hdl(dest);
if (hdl) {
direct_by_hdl_.erase(hdl);
direct_by_nid_.erase(dest);
parent_->parent().notify<hook::connection_lost>(dest);
......
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