Commit 238cc81f authored by Joseph Noir's avatar Joseph Noir

Add three-way handshake for UDP-based BASP

A three-way handshake eases the handling of repeated handshakes in BASP
which might happen when a handshake message is lost as well as offering
a defined point to send buffered messages when communication between
nodes is established automatically.
parent e0ffcd6f
...@@ -124,7 +124,8 @@ inline bool operator!=(const header& lhs, const header& rhs) { ...@@ -124,7 +124,8 @@ inline bool operator!=(const header& lhs, const header& rhs) {
/// Checks whether given header contains a handshake. /// Checks whether given header contains a handshake.
inline bool is_handshake(const header& hdr) { inline bool is_handshake(const header& hdr) {
return hdr.operation == message_type::server_handshake return hdr.operation == message_type::server_handshake
|| hdr.operation == message_type::client_handshake; || hdr.operation == message_type::client_handshake
|| hdr.operation == message_type::acknowledge_handshake;
} }
/// Checks wheter given header contains a heartbeat. /// Checks wheter given header contains a heartbeat.
......
...@@ -270,6 +270,11 @@ public: ...@@ -270,6 +270,11 @@ public:
buffer_type& buf, const node_id& remote_side, buffer_type& buf, const node_id& remote_side,
uint16_t sequence_number = 0); uint16_t sequence_number = 0);
/// Writes the acknowledge handshake to finish a UDP handshake to `buf`.
void write_acknowledge_handshake(execution_unit* ctx,
buffer_type& buf, const node_id& remote_side,
uint16_t sequence_number = 0);
/// Writes an `announce_proxy` to `buf`. /// Writes an `announce_proxy` to `buf`.
void write_announce_proxy(execution_unit* ctx, buffer_type& buf, void write_announce_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid, const node_id& dest_node, actor_id aid,
...@@ -310,14 +315,14 @@ public: ...@@ -310,14 +315,14 @@ public:
// Handle message to ourselves. // Handle message to ourselves.
switch (hdr.operation) { switch (hdr.operation) {
case message_type::server_handshake: { case message_type::server_handshake: {
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
basp::routing_table::address_map addrs;
if (!payload_valid()) { if (!payload_valid()) {
CAF_LOG_ERROR("fail to receive the app identifier"); CAF_LOG_ERROR("fail to receive the app identifier");
return false; return false;
} else { }
binary_deserializer bd{ctx, *payload}; binary_deserializer bd{ctx, *payload};
actor_id aid = invalid_actor_id;
std::set<std::string> sigs;
basp::routing_table::address_map addrs;
std::string remote_appid; std::string remote_appid;
auto e = bd(remote_appid); auto e = bd(remote_appid);
if (e) if (e)
...@@ -329,7 +334,6 @@ public: ...@@ -329,7 +334,6 @@ public:
e = bd(aid, sigs, addrs); e = bd(aid, sigs, addrs);
if (e) if (e)
return false; return false;
}
// Close self connection after handshake is done. // Close self connection after handshake is done.
if (hdr.source_node == this_node_) { if (hdr.source_node == this_node_) {
CAF_LOG_INFO("close connection to self immediately"); CAF_LOG_INFO("close connection to self immediately");
...@@ -338,6 +342,7 @@ public: ...@@ -338,6 +342,7 @@ public:
} }
// Close this connection if we already established communication. // Close this connection if we already established communication.
auto lr = tbl_.lookup(hdr.source_node); auto lr = tbl_.lookup(hdr.source_node);
// TODO: Anything additional or different for UDP?
if (lr.hdl) { if (lr.hdl) {
CAF_LOG_INFO("close connection since we already have a " CAF_LOG_INFO("close connection since we already have a "
"connection: " << CAF_ARG(hdr.source_node)); "connection: " << CAF_ARG(hdr.source_node));
...@@ -356,8 +361,12 @@ public: ...@@ -356,8 +361,12 @@ public:
// Write handshake as client in response. // Write handshake as client in response.
if (tcp_based) { if (tcp_based) {
write_client_handshake(ctx, callee_.get_buffer(hdl), hdr.source_node); write_client_handshake(ctx, callee_.get_buffer(hdl), hdr.source_node);
flush(hdl); } else {
auto seq = ep->requires_ordering ? ep->seq_outgoing++ : 0;
write_acknowledge_handshake(ctx, callee_.get_buffer(hdl),
hdr.source_node, seq);
} }
flush(hdl);
callee_.learned_new_node(hdr.source_node); callee_.learned_new_node(hdr.source_node);
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
callee_.send_buffered_messages(ctx, hdr.source_node, hdl); callee_.send_buffered_messages(ctx, hdr.source_node, hdl);
...@@ -368,7 +377,7 @@ public: ...@@ -368,7 +377,7 @@ public:
if (!payload_valid()) { if (!payload_valid()) {
CAF_LOG_ERROR("fail to receive the app identifier"); CAF_LOG_ERROR("fail to receive the app identifier");
return false; return false;
} else { }
binary_deserializer bd{ctx, *payload}; binary_deserializer bd{ctx, *payload};
std::string remote_appid; std::string remote_appid;
auto e = bd(remote_appid); auto e = bd(remote_appid);
...@@ -381,29 +390,15 @@ public: ...@@ -381,29 +390,15 @@ public:
e = bd(addrs); e = bd(addrs);
if (e) if (e)
return false; return false;
}
// Handshakes were only exchanged if `hdl` is set. // Handshakes were only exchanged if `hdl` is set.
auto lr = tbl_.lookup(hdr.source_node); auto lr = tbl_.lookup(hdr.source_node);
auto new_node = (this_node() != hdr.source_node && !lr.hdl); auto new_node = (this_node() != hdr.source_node && !lr.hdl);
auto pending = lr.known && !lr.hdl;
if (pending && hdr.source_node < this_node()) {
CAF_LOG_INFO("simultaneous handshake, let the other node act as "
"the server");
break;
}
if (!new_node) {
if (tcp_based) { if (tcp_based) {
if (!new_node) {
CAF_LOG_INFO("received second client handshake:" CAF_LOG_INFO("received second client handshake:"
<< CAF_ARG(hdr.source_node)); << CAF_ARG(hdr.source_node));
break; break;
} else {
if (lr.hdl && get_if<datagram_handle>(&(*lr.hdl)) == nullptr
&& get_if<datagram_handle>(&(*lr.hdl))->id() != hdl.id()) {
CAF_LOG_INFO("dropping repeated handshake on different handle");
break;
} }
}
} else {
// Add this node to our contacts. // Add this node to our contacts.
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node)); CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
// Either add a new node or add the handle to a known one. // Either add a new node or add the handle to a known one.
...@@ -411,21 +406,37 @@ public: ...@@ -411,21 +406,37 @@ public:
tbl_.handle(hdr.source_node, hdl); tbl_.handle(hdr.source_node, hdl);
else else
tbl_.add(hdr.source_node, hdl); tbl_.add(hdr.source_node, hdl);
callee_.learned_new_node(hdr.source_node);
callee_.send_buffered_messages(ctx, hdr.source_node, hdl);
} else {
if (!lr.known)
tbl_.add(hdr.source_node);
uint16_t seq = ep->requires_ordering ? ep->seq_outgoing++ : 0;
write_server_handshake(ctx, callee_.get_buffer(hdl), port, seq);
callee_.flush(hdl);
}
auto peer_server = system().registry().get(atom("PeerServ")); auto peer_server = system().registry().get(atom("PeerServ"));
anon_send(actor_cast<actor>(peer_server), put_atom::value, anon_send(actor_cast<actor>(peer_server), put_atom::value,
to_string(hdr.source_node), make_message(addrs)); to_string(hdr.source_node), make_message(addrs));
break;
} }
// Since udp is unreliable we answer, maybe our message was lost. case message_type::acknowledge_handshake: {
if (!tcp_based) { if (tcp_based) {
uint16_t seq = ep->requires_ordering ? ep->seq_outgoing++ : 0; CAF_LOG_ERROR("received acknowledge handshake via tcp");
write_server_handshake(ctx, callee_.get_buffer(hdl), port, seq); break;
callee_.flush(hdl); }
auto lr = tbl_.lookup(hdr.source_node);
if (!lr.known) {
CAF_LOG_DEBUG("dropping acknowledge handshake from unknown node");
break;
} }
// We have to call this after `write_server_handshake` because if (lr.hdl) {
// `learned_new_node` expects there to be an entry in the routing table. CAF_LOG_DEBUG("dropping repeated acknowledge handshake");
// TODO: Can we move this in the else block above with the changes to // TODO: Or should we just adopt the new handle?
// the routing table? break;
if (new_node) }
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.handle(hdr.source_node, hdl);
callee_.learned_new_node(hdr.source_node); callee_.learned_new_node(hdr.source_node);
callee_.send_buffered_messages(ctx, hdr.source_node, hdl); callee_.send_buffered_messages(ctx, hdr.source_node, hdl);
break; break;
......
...@@ -67,6 +67,10 @@ enum class message_type : uint8_t { ...@@ -67,6 +67,10 @@ enum class message_type : uint8_t {
/// ///
/// ![](heartbeat.png) /// ![](heartbeat.png)
heartbeat = 0x05, heartbeat = 0x05,
/// Last message in a BASP handshake for between two endpoints using UDP
/// as a transport protocol.
acknowledge_handshake = 0x06,
}; };
/// @relates message_type /// @relates message_type
......
...@@ -121,6 +121,13 @@ bool heartbeat_valid(const header& hdr) { ...@@ -121,6 +121,13 @@ bool heartbeat_valid(const header& hdr) {
&& zero(hdr.operation_data); && zero(hdr.operation_data);
} }
bool acknowledge_handshake_valid(const header& hdr) {
return valid(hdr.source_node)
&& hdr.source_node != hdr.dest_node
&& zero(hdr.source_actor)
&& zero(hdr.dest_actor);
}
} // namespace <anonymous> } // namespace <anonymous>
bool valid(const header& hdr) { bool valid(const header& hdr) {
...@@ -139,6 +146,8 @@ bool valid(const header& hdr) { ...@@ -139,6 +146,8 @@ bool valid(const header& hdr) {
return kill_proxy_instance_valid(hdr); return kill_proxy_instance_valid(hdr);
case message_type::heartbeat: case message_type::heartbeat:
return heartbeat_valid(hdr); return heartbeat_valid(hdr);
case message_type::acknowledge_handshake:
return acknowledge_handshake_valid(hdr);
} }
} }
......
...@@ -364,6 +364,17 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -364,6 +364,17 @@ void instance::write_client_handshake(execution_unit* ctx,
sequence_number); sequence_number);
} }
void instance::write_acknowledge_handshake(execution_unit* ctx,
buffer_type& buf,
const node_id& remote_side,
uint16_t sequence_number) {
header hdr{message_type::acknowledge_handshake, 0, 0, 0,
this_node_, remote_side, invalid_actor_id, invalid_actor_id,
sequence_number};
write(ctx, buf, hdr);
}
void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf, void instance::write_announce_proxy(execution_unit* ctx, buffer_type& buf,
const node_id& dest_node, actor_id aid, const node_id& dest_node, actor_id aid,
uint16_t sequence_number) { uint16_t sequence_number) {
......
...@@ -32,7 +32,8 @@ const char* message_type_strings[] = { ...@@ -32,7 +32,8 @@ const char* message_type_strings[] = {
"dispatch_message", "dispatch_message",
"announce_proxy_instance", "announce_proxy_instance",
"kill_proxy_instance", "kill_proxy_instance",
"heartbeat" "heartbeat",
"acknowledge_handshake"
}; };
} // namespace <anonymous> } // namespace <anonymous>
......
...@@ -294,17 +294,21 @@ public: ...@@ -294,17 +294,21 @@ public:
CAF_MESSAGE("send client handshake"); CAF_MESSAGE("send client handshake");
mock(src, ep, mock(src, ep,
{basp::message_type::client_handshake, 0, 0, 0, {basp::message_type::client_handshake, 0, 0, 0,
n.id, this_node(), n.id, this_node(), invalid_actor_id, invalid_actor_id},
invalid_actor_id, invalid_actor_id}, std::string{}, std::string{}, basp::routing_table::address_map{})
basp::routing_table::address_map{})
// upon receiving the client handshake, the server should answer // upon receiving the client handshake, the server should answer
// with the server handshake and send the dispatch_message blow // with the server handshake and send the dispatch_message blow
.receive(hdl, .receive(hdl,
basp::message_type::server_handshake, no_flags, basp::message_type::server_handshake, no_flags,
any_vals, basp::version, this_node(), node_id{none}, any_vals, basp::version, this_node(), node_id{none},
published_actor_id, invalid_actor_id, std::string{}, published_actor_id, invalid_actor_id, std::string{},
published_actor_id, published_actor_ifs, am) published_actor_id, published_actor_ifs, am);
// upon receiving our client handshake, BASP will check // UDP uses a three-way handshake, so let's answer with the final message.
mock(src, ep,
{basp::message_type::acknowledge_handshake, 0,0,0,
n.id, this_node(), invalid_actor_id, invalid_actor_id,
1})
// upon receiving our acknowledge handshake, BASP will check
// whether there is a SpawnServ actor on this node // whether there is a SpawnServ actor on this node
.receive(hdl, .receive(hdl,
basp::message_type::dispatch_message, basp::message_type::dispatch_message,
...@@ -562,6 +566,26 @@ public: ...@@ -562,6 +566,26 @@ public:
CAF_TEST_FIXTURE_SCOPE(basp_udp_tests, fixture) CAF_TEST_FIXTURE_SCOPE(basp_udp_tests, fixture)
CAF_TEST(empty_client_handshake_udp) {
// test whether basp instance correctly sends a
// client handshake when there's no actor published
buffer buf;
instance().write_client_handshake(mpx(), buf, none);
basp::header hdr;
buffer payload;
std::tie(hdr, payload) = from_buf(buf);
basp::header expected{basp::message_type::client_handshake, 0,
static_cast<uint32_t>(payload.size()), 0,
this_node(), none,
invalid_actor_id, invalid_actor_id,
0};
CAF_CHECK(basp::valid(hdr));
CAF_CHECK(basp::is_handshake(hdr));
CAF_MESSAGE("got : " << to_string(hdr));
CAF_MESSAGE("expecting: " << to_string(expected));
CAF_CHECK_EQUAL(to_string(hdr), to_string(expected));
}
CAF_TEST(empty_server_handshake_udp) { CAF_TEST(empty_server_handshake_udp) {
// test whether basp instance correctly sends a // test whether basp instance correctly sends a
// client handshake when there's no actor published // client handshake when there's no actor published
...@@ -581,15 +605,15 @@ CAF_TEST(empty_server_handshake_udp) { ...@@ -581,15 +605,15 @@ CAF_TEST(empty_server_handshake_udp) {
CAF_CHECK_EQUAL(to_string(hdr), to_string(expected)); CAF_CHECK_EQUAL(to_string(hdr), to_string(expected));
} }
CAF_TEST(empty_client_handshake_udp) { CAF_TEST(empty_acknowledge_handshake_udp) {
// test whether basp instance correctly sends a // test whether a basp instance correctly sends an
// client handshake when there's no actor published // acknowldge handshake
buffer buf; buffer buf;
instance().write_client_handshake(mpx(), buf, none); instance().write_acknowledge_handshake(mpx(), buf, none);
basp::header hdr; basp::header hdr;
buffer payload; buffer payload;
std::tie(hdr, payload) = from_buf(buf); std::tie(hdr, payload) = from_buf(buf);
basp::header expected{basp::message_type::client_handshake, 0, basp::header expected{basp::message_type::acknowledge_handshake, 0,
static_cast<uint32_t>(payload.size()), 0, static_cast<uint32_t>(payload.size()), 0,
this_node(), none, this_node(), none,
invalid_actor_id, invalid_actor_id, invalid_actor_id, invalid_actor_id,
...@@ -646,7 +670,7 @@ CAF_TEST(client_handshake_and_dispatch_udp) { ...@@ -646,7 +670,7 @@ CAF_TEST(client_handshake_and_dispatch_udp) {
mock(jupiter().endpoint, default_sender(), mock(jupiter().endpoint, default_sender(),
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::dispatch_message, 0, 0, 0,
jupiter().id, this_node(), jupiter().dummy_actor->id(), self()->id(), jupiter().id, this_node(), jupiter().dummy_actor->id(), self()->id(),
1}, // increment sequence number 2}, // increment sequence number
std::vector<actor_addr>{}, std::vector<actor_addr>{},
make_message(1, 2, 3)) make_message(1, 2, 3))
.receive(jupiter().endpoint, .receive(jupiter().endpoint,
...@@ -716,6 +740,11 @@ CAF_TEST(remote_actor_and_send_udp) { ...@@ -716,6 +740,11 @@ CAF_TEST(remote_actor_and_send_udp) {
jupiter().dummy_actor->id(), jupiter().dummy_actor->id(),
uint32_t{0}, uint32_t{0},
basp::routing_table::address_map{}) basp::routing_table::address_map{})
.receive(jupiter().endpoint,
basp::message_type::acknowledge_handshake,
no_flags, no_payload, no_operation_data,
this_node(), jupiter().id,
invalid_actor_id, invalid_actor_id)
.receive(jupiter().endpoint, .receive(jupiter().endpoint,
basp::message_type::dispatch_message, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
...@@ -802,7 +831,7 @@ CAF_TEST(actor_serialize_and_deserialize_udp) { ...@@ -802,7 +831,7 @@ CAF_TEST(actor_serialize_and_deserialize_udp) {
{basp::message_type::dispatch_message, 0, 0, 0, {basp::message_type::dispatch_message, 0, 0, 0,
prx->node(), this_node(), prx->node(), this_node(),
prx->id(), testee->id(), prx->id(), testee->id(),
1}, // sequence number, first message after handshake 2}, // sequence number, previous messages: client and ack handshake
std::vector<actor_id>{}, std::vector<actor_id>{},
msg); msg);
// testee must've responded (process forwarded message in BASP broker) // testee must've responded (process forwarded message in BASP broker)
...@@ -856,6 +885,9 @@ CAF_TEST(out_of_order_delivery_udp) { ...@@ -856,6 +885,9 @@ CAF_TEST(out_of_order_delivery_udp) {
jupiter().dummy_actor->id(), jupiter().dummy_actor->id(),
uint32_t{0}, uint32_t{0},
basp::routing_table::address_map{}) basp::routing_table::address_map{})
.receive(jupiter().endpoint, basp::message_type::acknowledge_handshake,
no_flags, no_payload, no_operation_data,
this_node(), jupiter().id, invalid_actor_id, invalid_actor_id)
.receive(jupiter().endpoint, .receive(jupiter().endpoint,
basp::message_type::dispatch_message, basp::message_type::dispatch_message,
basp::header::named_receiver_flag, any_vals, basp::header::named_receiver_flag, any_vals,
......
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