Unverified Commit f503a0f4 authored by Joseph Noir's avatar Joseph Noir Committed by GitHub

Merge pull request #1070

Make sure BASP nodes agree on a single connection
parents a7a8dc50 a17e558a
......@@ -63,9 +63,13 @@ struct header {
header() = default;
/// Identifies a receiver by name rather than ID.
/// Identifies a receiver by name rather than ID in a `direct_message`.
static const uint8_t named_receiver_flag = 0x01;
/// Forces the server to use this connection to a client in a
/// `client_handshake`.
static const uint8_t select_connection_flag = 0x02;
/// Queries whether this header has the given flag.
bool has(uint8_t flag) const {
return (flags & flag) != 0;
......@@ -117,4 +121,3 @@ constexpr size_t header_size = sizeof(actor_id) * 2
} // namespace basp
} // namespace io
} // namespace caf
......@@ -191,7 +191,8 @@ public:
buffer_type& out_buf, optional<uint16_t> port);
/// Writes the client handshake to `buf`.
void write_client_handshake(execution_unit* ctx, buffer_type& buf);
void write_client_handshake(execution_unit* ctx, buffer_type& buf,
uint8_t flags = 0);
/// Writes an `announce_proxy` to `buf`.
void write_monitor_message(execution_unit* ctx, buffer_type& buf,
......
......@@ -37,6 +37,9 @@ namespace basp {
/// BASP peer and provides both direct and indirect paths.
class routing_table {
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);
......@@ -67,6 +70,14 @@ public:
/// @pre `hdl != invalid_connection_handle && nid != none`
void add_direct(const connection_handle& hdl, const node_id& nid);
/// When two CAF nodes connect to each other, multiple connections might spin
/// up simultaneously until both sides agree to a single connection.
/// @pre `lookup_direct(hdl == nid)`
void add_alternative(const connection_handle& hdl, const node_id& nid);
/// Forces `lookup_direct` to always resolve `nid` to `hdl`.
void select_alternative(const connection_handle& hdl, const node_id& nid);
/// Adds a new indirect route to the table.
bool add_indirect(const node_id& hop, const node_id& dest);
......@@ -88,8 +99,8 @@ public:
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_;
handle_to_node_map direct_by_hdl_;
node_to_handle_map direct_by_nid_;
std::unordered_map<node_id, node_id_set> indirect_;
};
......
......@@ -245,11 +245,12 @@ void instance::write_server_handshake(execution_unit* ctx, buffer_type& out_buf,
write(ctx, out_buf, hdr, &writer);
}
void instance::write_client_handshake(execution_unit* ctx, buffer_type& buf) {
void instance::write_client_handshake(execution_unit* ctx, buffer_type& buf,
uint8_t flags) {
auto writer = make_callback(
[&](serializer& sink) -> error { return sink(this_node_); });
header hdr{message_type::client_handshake,
0,
flags,
0,
0,
invalid_actor_id,
......@@ -338,13 +339,17 @@ connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
tbl_.add_direct(hdl, source_node);
auto was_indirect = tbl_.erase_indirect(source_node);
// write handshake as client in response
// Make sure the correct path is registered in the routing table.
auto path = tbl_.lookup(source_node);
if (!path) {
CAF_LOG_ERROR("no route to host after server handshake");
return no_route_to_receiving_node;
}
write_client_handshake(ctx, callee_.get_buffer(path->hdl));
// Repeat client handshake with select_connection_flag to make sure the
// server uses this connection in its routing table.
write_client_handshake(ctx, callee_.get_buffer(path->hdl),
header::select_connection_flag);
// Inform the callee.
callee_.learned_new_node_directly(source_node, was_indirect);
callee_.finalize_handshake(source_node, aid, sigs);
flush(*path);
......@@ -359,14 +364,22 @@ connection_state instance::handle(execution_unit* ctx, connection_handle hdl,
<< ctx->system().render(err));
return serializing_basp_payload_failed;
}
// Drop repeated handshakes.
// Handle repeated handshakes by updateing the routing table as necessary.
if (tbl_.lookup_direct(source_node)) {
CAF_LOG_DEBUG(
"received repeated client handshake:" << CAF_ARG(source_node));
if (hdr.has(header::select_connection_flag)) {
CAF_LOG_DEBUG("client selected this connection:"
<< CAF_ARG(source_node) << CAF_ARG(hdl));
tbl_.select_alternative(hdl, source_node);
} else {
CAF_LOG_DEBUG("new alternative route:" << CAF_ARG(source_node)
<< CAF_ARG(hdl));
tbl_.add_alternative(hdl, source_node);
}
break;
}
// Add direct route to this node and remove any indirect entry.
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node));
CAF_LOG_DEBUG("new direct connection:" << CAF_ARG(source_node)
<< CAF_ARG(hdl));
tbl_.add_direct(hdl, source_node);
auto was_indirect = tbl_.erase_indirect(source_node);
callee_.learned_new_node_directly(source_node, was_indirect);
......
......@@ -85,13 +85,34 @@ node_id routing_table::lookup_indirect(const node_id& nid) const {
node_id routing_table::erase_direct(const connection_handle& hdl) {
std::unique_lock<std::mutex> guard{mtx_};
// Sanity check: do nothing if the handle is not mapped to a node.
auto i = direct_by_hdl_.find(hdl);
if (i == direct_by_hdl_.end())
return {};
direct_by_nid_.erase(i->second);
node_id result = std::move(i->second);
direct_by_hdl_.erase(i->first);
return result;
// We always remove i from direct_by_hdl_.
auto node = std::move(i->second);
direct_by_hdl_.erase(i);
// Sanity check: direct_by_nid_ should contain a reverse mapping.
auto j = direct_by_nid_.find(node);
if (j == direct_by_nid_.end()) {
CAF_LOG_WARNING("no reverse mapping exists for the connection handle");
return node;
}
// Try to find an alternative connection for communicating with the node.
auto predicate = [&](const 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.
j->second = alternative->first;
return {};
} else {
// Drop the node after losing the last connection to it.
direct_by_nid_.erase(j);
return node;
}
}
bool routing_table::erase_indirect(const node_id& dest) {
......@@ -113,6 +134,22 @@ void routing_table::add_direct(const connection_handle& hdl,
CAF_IGNORE_UNUSED(nid_added);
}
void routing_table::add_alternative(const connection_handle& hdl,
const node_id& nid) {
std::unique_lock<std::mutex> guard{mtx_};
CAF_ASSERT(direct_by_nid_.count(nid) != 0);
// This member function is safe to call repeatedly. Hence, we ignore the
// result of emplace on purpose.
direct_by_hdl_.emplace(hdl, nid);
}
void routing_table::select_alternative(const connection_handle& hdl,
const node_id& nid) {
std::unique_lock<std::mutex> guard{mtx_};
CAF_ASSERT(direct_by_hdl_[hdl] == nid);
direct_by_nid_[nid] = hdl;
}
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.
......
......@@ -293,6 +293,9 @@ behavior basp_broker::make_behavior() {
ctx.callback = rp;
// await server handshake
configure_read(hdl, receive_policy::exactly(basp::header_size));
// send client handshake
instance.write_client_handshake(context(), get_buffer(hdl));
flush(hdl);
},
[=](delete_atom, const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(nid) << ", " << CAF_ARG(aid));
......
......@@ -591,6 +591,9 @@ CAF_TEST(remote_actor_and_send) {
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, any_vals, no_operation_data, invalid_actor_id,
invalid_actor_id, this_node())
.receive(jupiter().connection, basp::message_type::client_handshake,
basp::header::select_connection_flag, any_vals, no_operation_data,
invalid_actor_id, invalid_actor_id, this_node())
.receive(jupiter().connection, basp::message_type::direct_message,
basp::header::named_receiver_flag, any_vals,
default_operation_data, any_vals,
......@@ -638,6 +641,143 @@ CAF_TEST(remote_actor_and_send) {
});
}
CAF_TEST(BASP clients select which connection to use) {
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("Receiving select_connection_flag changes the routing table");
mock(conn2,
{basp::message_type::client_handshake,
basp::header::select_connection_flag, 0, 0, invalid_actor_id,
invalid_actor_id},
jupiter().id);
CAF_CHECK_EQUAL(tbl().lookup_direct(jupiter().id), conn2);
CAF_CHECK_EQUAL(tbl().lookup_direct(conn1), 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) {
auto testee_impl = [](event_based_actor* testee_self) -> behavior {
testee_self->set_default_handler(reflect_and_quit);
......@@ -787,7 +927,10 @@ CAF_TEST(automatic_connection) {
jupiter().dummy_actor->id(), std::set<std::string>{})
.receive(jupiter().connection, basp::message_type::client_handshake,
no_flags, any_vals, no_operation_data, invalid_actor_id,
invalid_actor_id, this_node());
invalid_actor_id, this_node())
.receive(jupiter().connection, basp::message_type::client_handshake,
basp::header::select_connection_flag, any_vals, no_operation_data,
invalid_actor_id, invalid_actor_id, this_node());
CAF_CHECK_EQUAL(tbl().lookup_indirect(jupiter().id), none);
CAF_CHECK_EQUAL(tbl().lookup_indirect(mars().id), none);
check_node_in_tbl(jupiter());
......
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