Commit 0ff9f69e authored by Dominik Charousset's avatar Dominik Charousset

Make routing table thread safe

parent 9d87f63e
......@@ -20,19 +20,18 @@
#include <limits>
#include "caf/error.hpp"
#include "caf/variant.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/binary_deserializer.hpp"
#include "caf/io/hook.hpp"
#include "caf/io/middleman.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/callback.hpp"
#include "caf/error.hpp"
#include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/connection_state.hpp"
#include "caf/io/basp/header.hpp"
#include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp"
#include "caf/io/basp/connection_state.hpp"
#include "caf/io/hook.hpp"
#include "caf/io/middleman.hpp"
#include "caf/variant.hpp"
namespace caf {
namespace io {
......
......@@ -18,10 +18,11 @@
#pragma once
#include <mutex>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include "caf/callback.hpp"
#include "caf/io/abstract_broker.hpp"
#include "caf/io/basp/buffer_type.hpp"
#include "caf/node_id.hpp"
......@@ -47,10 +48,6 @@ public:
connection_handle hdl;
};
/// Describes a function object for erase operations that
/// is called for each indirectly lost connection.
using erase_callback = callback<const node_id&>;
/// Returns a route to `target` or `none` on error.
optional<route> lookup(const node_id& target);
......@@ -73,48 +70,27 @@ public:
/// Adds a new indirect route to the table.
bool add_indirect(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);
/// Removes a direct connection and return the node ID that became
/// unreachable as a result of this operation.
node_id erase_direct(const connection_handle& hdl);
/// 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);
/// Removes all direct and indirect routes to `dest` and calls
/// `cb` for any node that became unreachable as a result of this
/// operation, including `dest`.
/// @returns the number of removed routes (direct and indirect)
size_t erase(const node_id& dest, erase_callback& cb);
/// Returns the parent broker.
inline abstract_broker* parent() {
abstract_broker* parent() {
return parent_;
}
public:
template <class Map, class Fallback>
typename Map::mapped_type
get_opt(const Map& m, const typename Map::key_type& k, Fallback&& x) const {
auto i = m.find(k);
if (i != m.end())
return i->second;
return std::forward<Fallback>(x);
}
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_;
mutable std::mutex mtx_;
std::unordered_map<connection_handle, node_id> direct_by_hdl_;
std::unordered_map<node_id, connection_handle> direct_by_nid_;
indirect_entries indirect_;
std::unordered_map<node_id, node_id_set> indirect_;
};
/// @}
......@@ -122,4 +98,3 @@ public:
} // namespace basp
} // namespace io
} // namespace caf
......@@ -279,11 +279,8 @@ void basp_broker_state::cleanup(connection_handle hdl) {
CAF_LOG_TRACE(CAF_ARG(hdl));
// Remove handle from the routing table and clean up any node-specific state
// we might still have.
auto cb = make_callback([&](const node_id& nid) -> error {
if (auto nid = instance.tbl().erase_direct(hdl))
purge_state(nid);
return none;
});
instance.tbl().erase_direct(hdl, cb);
// Remove the context for `hdl`, making sure clients receive an error in case
// this connection was closed during handshake.
auto i = ctx.find(hdl);
......
......@@ -52,11 +52,8 @@ connection_state instance::handle(execution_unit* ctx,
CAF_LOG_TRACE(CAF_ARG(dm) << CAF_ARG(is_payload));
// function object providing cleanup code on errors
auto err = [&]() -> connection_state {
auto cb = make_callback([&](const node_id& nid) -> error {
if (auto nid = tbl_.erase_direct(dm.handle))
callee_.purge_state(nid);
return none;
});
tbl_.erase_direct(dm.handle, cb);
return close_connection;
};
std::vector<char>* payload = nullptr;
......@@ -383,7 +380,7 @@ bool instance::handle(execution_unit* ctx, connection_handle hdl, header& hdr,
}
auto last_hop = tbl_.lookup_direct(hdl);
if (source_node != none && source_node != this_node_
&& last_hop != source_node && !tbl_.lookup_direct(source_node)
&& last_hop != source_node
&& tbl_.add_indirect(last_hop, source_node))
callee_.learned_new_node_indirectly(source_node);
}
......
......@@ -34,18 +34,23 @@ routing_table::~routing_table() {
}
optional<routing_table::route> routing_table::lookup(const node_id& target) {
auto hdl = lookup_direct(target);
if (hdl)
return route{target, *hdl};
// pick first available indirect route
std::unique_lock<std::mutex> guard{mtx_};
// Check whether we have a direct path first.
{ // Lifetime scope of first iterator.
auto i = direct_by_nid_.find(target);
if (i != direct_by_nid_.end())
return route{target, i->second};
}
// Pick first available indirect route.
auto i = indirect_.find(target);
if (i != indirect_.end()) {
auto& hops = i->second;
while (!hops.empty()) {
auto& hop = *hops.begin();
hdl = lookup_direct(hop);
if (hdl)
return route{hop, *hdl};
auto j = direct_by_nid_.find(hop);
if (j != direct_by_nid_.end())
return route{hop, j->second};
// Erase hops that became invalid.
hops.erase(hops.begin());
}
}
......@@ -53,11 +58,16 @@ optional<routing_table::route> routing_table::lookup(const node_id& target) {
}
node_id routing_table::lookup_direct(const connection_handle& hdl) const {
return get_opt(direct_by_hdl_, hdl, none);
std::unique_lock<std::mutex> guard{mtx_};
auto i = direct_by_hdl_.find(hdl);
if (i != direct_by_hdl_.end())
return i->second;
return none;
}
optional<connection_handle>
routing_table::lookup_direct(const node_id& nid) const {
std::unique_lock<std::mutex> guard{mtx_};
auto i = direct_by_nid_.find(nid);
if (i != direct_by_nid_.end())
return i->second;
......@@ -65,26 +75,28 @@ routing_table::lookup_direct(const node_id& nid) const {
}
node_id routing_table::lookup_indirect(const node_id& nid) const {
std::unique_lock<std::mutex> guard{mtx_};
auto i = indirect_.find(nid);
if (i == indirect_.end())
return none;
if (i->second.empty())
return none;
if (!i->second.empty())
return *i->second.begin();
return none;
}
void routing_table::erase_direct(const connection_handle& hdl,
erase_callback& cb) {
node_id routing_table::erase_direct(const connection_handle& hdl) {
std::unique_lock<std::mutex> guard{mtx_};
auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end())
return;
cb(i->second);
parent_->parent().notify<hook::connection_lost>(i->second);
return none;
direct_by_nid_.erase(i->second);
node_id result = std::move(i->second);
direct_by_hdl_.erase(i->first);
return result;
}
bool routing_table::erase_indirect(const node_id& dest) {
std::unique_lock<std::mutex> guard{mtx_};
auto i = indirect_.find(dest);
if (i == indirect_.end())
return false;
......@@ -97,20 +109,24 @@ bool routing_table::erase_indirect(const node_id& dest) {
void routing_table::add_direct(const connection_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);
direct_by_nid_.emplace(nid, hdl);
std::unique_lock<std::mutex> guard{mtx_};
auto hdl_added = direct_by_hdl_.emplace(hdl, nid).second;
auto nid_added = direct_by_nid_.emplace(nid, hdl).second;
CAF_ASSERT(hdl_added && nid_added);
CAF_IGNORE_UNUSED(hdl_added);
CAF_IGNORE_UNUSED(nid_added);
parent_->parent().notify<hook::new_connection_established>(nid);
}
bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
std::unique_lock<std::mutex> guard{mtx_};
// Never add indirect entries if we already have direct connection.
if (lookup_direct(dest) != none)
if (direct_by_nid_.count(dest) != 0)
return false;
// Never add indirect entries if we don't have a connection to the hop.
if (lookup_direct(hop) == none)
if (direct_by_nid_.count(hop) == 0)
return false;
// Add entry to our node ID set.
auto& hops = indirect_[dest];
auto result = hops.empty();
if (hops.emplace(hop).second)
......@@ -118,32 +134,6 @@ bool routing_table::add_indirect(const node_id& hop, const node_id& dest) {
return result;
}
bool routing_table::reachable(const node_id& dest) {
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();
for (auto& nid : i->second) {
cb(nid);
parent_->parent().notify<hook::route_lost>(nid, dest);
}
indirect_.erase(i);
}
auto hdl = lookup_direct(dest);
if (hdl) {
direct_by_hdl_.erase(*hdl);
direct_by_nid_.erase(dest);
parent_->parent().notify<hook::connection_lost>(dest);
++res;
}
return res;
}
} // namespace basp
} // 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