Commit 762a926a authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'issue/1065' into topic/novaquark

parents 9906b7a5 96d91166
...@@ -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