Commit 83c43935 authored by Dominik Charousset's avatar Dominik Charousset

Make sure BASP nodes agree on a single connection

This change fixes a race on the routing tables when spinning up multiple
connections between two nodes simultaneously. With 2 nodes A and B, this
is the sequence causing an error:

- A publish an actor on port x.
- A publish an actor on port y.
- B connects to A:x with connection handle b1.
- B connects to A:y with connection handle b2.
- A sees a new connection handle a1 for the connection to A:x.
- A writes server handshake A:x.
- A reads client handshake A:x, adds (B, a1) to the routing table.
- A sees a new connection handle a2 for the connection to A:y.
- A writes server handshake A:y.
- A reads client handshake A:y and drops it.
- B reads server handshake on b2 and adds (A, b2) to the routing table.
- B reads server handshake on b1 and closes the redundant connection.

After that point, B uses the connection `a2 <-> b2`. However, the server
believes `a1 <-> b1` is the active connection. When receiving data via
`a2`, the server has no mapping for the handle in its routing table.
Consequently, it cannot attribute incoming data on this connection to
node B.

To fix this dilemma, this change adds additional steps:

- The server now adds *all* connection handles to its routing table. The
  first handle is going to be the default route. However, additional
  handles get added as alternatives. When receiving data on any of these
  sockets, the server is able to attribute it to right node ID.
- The client dictates which connection gets used. After reading the
  server handshake and setting a direct route to the server, the client
  repeats its handshake with the new select connection flag.
- When the server receives a handshake where the select connection flag
  is set, it updates it routing table to have the node ID resolve to the
  right connection handle.

This patch is backwards compatible. Older versions of CAF in the network
simply ignore the unrecognized flag.
parent 8f4e5b99
......@@ -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,
......
......@@ -67,6 +67,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 resolve `nid` always 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);
......
......@@ -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,14 +339,20 @@ 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;
}
// 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);
break;
}
case message_type::client_handshake: {
......@@ -357,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);
......
......@@ -113,6 +113,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.
......
......@@ -592,6 +592,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,
......@@ -639,6 +642,57 @@ 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("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_TEST(actor_serialize_and_deserialize) {
auto testee_impl = [](event_based_actor* testee_self) -> behavior {
testee_self->set_default_handler(reflect_and_quit);
......@@ -788,7 +842,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