Commit 96d91166 authored by Dominik Charousset's avatar Dominik Charousset

Implement fallback to alternative routes

If BASP loses connection to another BASP broker, it must check whether
an alternative connection exists before purging state.
parent 83c43935
...@@ -37,6 +37,9 @@ namespace basp { ...@@ -37,6 +37,9 @@ namespace basp {
/// BASP peer and provides both direct and indirect paths. /// BASP peer and provides both direct and indirect paths.
class routing_table { class routing_table {
public: public:
using handle_to_node_map = std::unordered_map<connection_handle, node_id>;
using node_to_handle_map = std::unordered_map<node_id, connection_handle>;
explicit routing_table(abstract_broker* parent); explicit routing_table(abstract_broker* parent);
...@@ -96,8 +99,8 @@ public: ...@@ -96,8 +99,8 @@ public:
abstract_broker* parent_; abstract_broker* parent_;
mutable std::mutex mtx_; mutable std::mutex mtx_;
std::unordered_map<connection_handle, node_id> direct_by_hdl_; handle_to_node_map direct_by_hdl_;
std::unordered_map<node_id, connection_handle> direct_by_nid_; node_to_handle_map direct_by_nid_;
std::unordered_map<node_id, node_id_set> indirect_; std::unordered_map<node_id, node_id_set> indirect_;
}; };
......
...@@ -88,10 +88,33 @@ node_id routing_table::erase_direct(const connection_handle& hdl) { ...@@ -88,10 +88,33 @@ node_id routing_table::erase_direct(const connection_handle& hdl) {
auto i = direct_by_hdl_.find(hdl); auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end()) if (i == direct_by_hdl_.end())
return {}; return {};
direct_by_nid_.erase(i->second); // Check whether this handle was the primary mapping for the node.
node_id result = std::move(i->second); auto& node = i->second;
direct_by_hdl_.erase(i->first); auto j = direct_by_nid_.find(node);
return result; if (j != direct_by_nid_.end()) {
if (j->second == hdl) {
// Try to find an alternative for falling back to a differnt connection.
auto predicate = [&](handle_to_node_map::value_type& kvp) {
return kvp.second == node;
};
auto e = direct_by_hdl_.end();
auto alternative = std::find_if(direct_by_hdl_.begin(), e, predicate);
if (alternative != e) {
// Update node <-> handle mapping and only drop the handle.
j->second = alternative->first;
} else {
// This was the only connection to the node. Drop it.
auto result = std::move(node);
direct_by_nid_.erase(j);
direct_by_hdl_.erase(i);
return result;
}
}
}
// Unless this was the only connection to the node, we only drop the
// connection handle.
direct_by_hdl_.erase(i);
return {};
} }
bool routing_table::erase_indirect(const node_id& dest) { bool routing_table::erase_indirect(const node_id& dest) {
......
...@@ -682,6 +682,19 @@ CAF_TEST(BASP clients select which connection to use) { ...@@ -682,6 +682,19 @@ CAF_TEST(BASP clients select which connection to use) {
CAF_CHECK_EQUAL(tbl().lookup_direct(jupiter().id), conn1); CAF_CHECK_EQUAL(tbl().lookup_direct(jupiter().id), conn1);
CAF_CHECK_EQUAL(tbl().lookup_direct(conn1), jupiter().id); CAF_CHECK_EQUAL(tbl().lookup_direct(conn1), jupiter().id);
CAF_CHECK_EQUAL(tbl().lookup_direct(conn2), jupiter().id); CAF_CHECK_EQUAL(tbl().lookup_direct(conn2), jupiter().id);
CAF_MESSAGE("BASP creates proxies for remote actors");
auto msg = make_message("hello world");
mock(conn2,
{basp::message_type::direct_message, 0, 0, 0, 31337, self()->id()},
std::vector<strong_actor_ptr>{}, msg);
self()->receive([&](const std::string& hello) {
auto& sender = self()->current_sender();
CAF_CHECK_EQUAL(sender->id(), 31337u);
CAF_CHECK_EQUAL(sender->node(), jupiter().id);
CAF_CHECK_EQUAL(hello, "hello world");
});
CAF_REQUIRE_EQUAL(proxies().count_proxies(jupiter().id), 1u);
CAF_REQUIRE_NOT_EQUAL(proxies().get(jupiter().id, 31337), nullptr);
CAF_MESSAGE("Receiving select_connection_flag changes the routing table"); CAF_MESSAGE("Receiving select_connection_flag changes the routing table");
mock(conn2, mock(conn2,
{basp::message_type::client_handshake, {basp::message_type::client_handshake,
...@@ -691,6 +704,79 @@ CAF_TEST(BASP clients select which connection to use) { ...@@ -691,6 +704,79 @@ CAF_TEST(BASP clients select which connection to use) {
CAF_CHECK_EQUAL(tbl().lookup_direct(jupiter().id), conn2); CAF_CHECK_EQUAL(tbl().lookup_direct(jupiter().id), conn2);
CAF_CHECK_EQUAL(tbl().lookup_direct(conn1), jupiter().id); CAF_CHECK_EQUAL(tbl().lookup_direct(conn1), jupiter().id);
CAF_CHECK_EQUAL(tbl().lookup_direct(conn2), jupiter().id); CAF_CHECK_EQUAL(tbl().lookup_direct(conn2), jupiter().id);
CAF_MESSAGE("Dropping one connection does not affect existing proxies");
anon_send(sys.middleman().named_broker<basp_broker>(basp_atom),
connection_closed_msg{conn1});
while (mpx()->try_exec_runnable())
; // repeat
CAF_CHECK_EQUAL(proxies().count_proxies(jupiter().id), 1u);
CAF_CHECK_NOT_EQUAL(proxies().get(jupiter().id, 31337), nullptr);
}
CAF_TEST(BASP falls back to alternative routes) {
// TODO: mostly same as the test above, doctest-style SUBCASE macros could
// elminate the copy & paste here.
CAF_MESSAGE("publish an actor at ports 4001 and 4002");
auto hdl1 = accept_handle::from_int(4001);
mpx()->provide_acceptor(4001, hdl1);
CAF_REQUIRE_EQUAL(sys.middleman().publish(self(), 4001), 4001);
auto hdl2 = accept_handle::from_int(4002);
mpx()->provide_acceptor(4002, hdl2);
CAF_REQUIRE_EQUAL(sys.middleman().publish(self(), 4002), 4002);
mpx()->flush_runnables(); // process publish message in basp_broker
auto mm = sys.middleman().actor_handle();
CAF_MESSAGE("connect Jupiter to both ports");
auto conn1 = jupiter().connection;
auto conn2 = connection_handle::from_int(4002);
mpx()->add_pending_connect(hdl1, conn1);
mpx()->add_pending_connect(hdl2, conn2);
mpx()->accept_connection(hdl1);
mpx()->accept_connection(hdl2);
CAF_MESSAGE("BASP one server handshakes for each incoming connection");
auto published_actor_id = self()->id();
std::set<std::string> published_actor_ifs;
mock().receive(conn1, basp::message_type::server_handshake, no_flags,
any_vals, basp::version, invalid_actor_id, invalid_actor_id,
this_node(), defaults::middleman::app_identifiers,
published_actor_id, published_actor_ifs);
mock().receive(conn2, basp::message_type::server_handshake, no_flags,
any_vals, basp::version, invalid_actor_id, invalid_actor_id,
this_node(), defaults::middleman::app_identifiers,
published_actor_id, published_actor_ifs);
CAF_MESSAGE("After receiving the client handshakes, BASP has two routes");
mock(conn1,
{basp::message_type::client_handshake, 0, 0, 0, invalid_actor_id,
invalid_actor_id},
jupiter().id);
mock(conn2,
{basp::message_type::client_handshake, 0, 0, 0, invalid_actor_id,
invalid_actor_id},
jupiter().id);
CAF_CHECK_EQUAL(tbl().lookup_direct(jupiter().id), conn1);
CAF_CHECK_EQUAL(tbl().lookup_direct(conn1), jupiter().id);
CAF_CHECK_EQUAL(tbl().lookup_direct(conn2), jupiter().id);
CAF_MESSAGE("BASP creates proxies for remote actors");
auto msg = make_message("hello world");
mock(conn2,
{basp::message_type::direct_message, 0, 0, 0, 31337, self()->id()},
std::vector<strong_actor_ptr>{}, msg);
self()->receive([&](const std::string& hello) {
auto& sender = self()->current_sender();
CAF_CHECK_EQUAL(sender->id(), 31337u);
CAF_CHECK_EQUAL(sender->node(), jupiter().id);
CAF_CHECK_EQUAL(hello, "hello world");
});
CAF_REQUIRE_EQUAL(proxies().count_proxies(jupiter().id), 1u);
CAF_REQUIRE_NOT_EQUAL(proxies().get(jupiter().id, 31337), nullptr);
CAF_MESSAGE("Dropping the main connection falls back to the alternative");
anon_send(sys.middleman().named_broker<basp_broker>(basp_atom),
connection_closed_msg{conn1});
while (mpx()->try_exec_runnable())
; // repeat
CAF_CHECK_EQUAL(tbl().lookup_direct(jupiter().id), conn2);
CAF_CHECK_EQUAL(tbl().lookup_direct(conn2), jupiter().id);
CAF_CHECK_EQUAL(proxies().count_proxies(jupiter().id), 1u);
CAF_CHECK_NOT_EQUAL(proxies().get(jupiter().id, 31337), nullptr);
} }
CAF_TEST(actor_serialize_and_deserialize) { CAF_TEST(actor_serialize_and_deserialize) {
......
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