Commit 328abb98 authored by Joseph Noir's avatar Joseph Noir

Resolve simultaneous handshakes via UDP

parent 9616bd0c
...@@ -348,6 +348,12 @@ public: ...@@ -348,6 +348,12 @@ public:
"connection: " << CAF_ARG(hdr.source_node)); "connection: " << CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs); callee_.finalize_handshake(hdr.source_node, aid, sigs);
return false; return false;
} else if (!tcp_based && (tbl_.received_client_handshake(hdr.source_node)
&& this_node() < hdr.source_node)) {
CAF_LOG_INFO("simultaneous handshake, let the other node proceed: "
<< CAF_ARG(hdr.source_node));
callee_.finalize_handshake(hdr.source_node, aid, sigs);
return false;
} }
// 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));
...@@ -411,6 +417,7 @@ public: ...@@ -411,6 +417,7 @@ public:
} else { } else {
if (!lr.known) if (!lr.known)
tbl_.add(hdr.source_node); tbl_.add(hdr.source_node);
tbl_.received_client_handshake(hdr.source_node, true);
uint16_t seq = ep->requires_ordering ? ep->seq_outgoing++ : 0; uint16_t seq = ep->requires_ordering ? ep->seq_outgoing++ : 0;
write_server_handshake(ctx, callee_.get_buffer(hdl), port, seq); write_server_handshake(ctx, callee_.get_buffer(hdl), port, seq);
callee_.flush(hdl); callee_.flush(hdl);
...@@ -437,6 +444,7 @@ public: ...@@ -437,6 +444,7 @@ public:
} }
CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node)); CAF_LOG_INFO("new endpoint:" << CAF_ARG(hdr.source_node));
tbl_.handle(hdr.source_node, hdl); tbl_.handle(hdr.source_node, hdl);
tbl_.received_client_handshake(hdr.source_node, false);
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;
......
...@@ -133,6 +133,12 @@ public: ...@@ -133,6 +133,12 @@ public:
/// Get a reference to an address map for the local node. /// Get a reference to an address map for the local node.
const address_map& local_addresses(); const address_map& local_addresses();
// Set the received client handshake flag for `nid`.
bool received_client_handshake(const node_id& nid, bool flag);
// Get the received client handshake flag for `nid`.
bool received_client_handshake(const node_id& nid);
public: public:
/// Entry to bundle information for a remote endpoint. /// Entry to bundle information for a remote endpoint.
struct node_info { struct node_info {
...@@ -140,6 +146,9 @@ public: ...@@ -140,6 +146,9 @@ public:
optional<endpoint_handle> hdl; optional<endpoint_handle> hdl;
/// The endpoint who told us about the node. /// The endpoint who told us about the node.
optional<node_id> origin; optional<node_id> origin;
/// Track if we received a client handshake to solve simultaneous
/// handshake with UDP.
bool received_client_handshake;
}; };
template <class Map, class Fallback> template <class Map, class Fallback>
......
...@@ -59,13 +59,13 @@ void routing_table::add(const node_id& nid, const endpoint_handle& hdl) { ...@@ -59,13 +59,13 @@ void routing_table::add(const node_id& nid, const endpoint_handle& hdl) {
CAF_ASSERT(nid_by_hdl_.count(hdl) == 0); CAF_ASSERT(nid_by_hdl_.count(hdl) == 0);
CAF_ASSERT(node_information_base_.count(nid) == 0); CAF_ASSERT(node_information_base_.count(nid) == 0);
nid_by_hdl_.emplace(hdl, nid); nid_by_hdl_.emplace(hdl, nid);
node_information_base_[nid] = node_info{hdl, none}; node_information_base_[nid] = node_info{hdl, none, false};
parent_->parent().notify<hook::new_connection_established>(nid); parent_->parent().notify<hook::new_connection_established>(nid);
} }
void routing_table::add(const node_id& nid, const node_id& origin) { void routing_table::add(const node_id& nid, const node_id& origin) {
CAF_ASSERT(node_information_base_.count(nid) == 0); CAF_ASSERT(node_information_base_.count(nid) == 0);
node_information_base_[nid] = node_info{none, origin}; node_information_base_[nid] = node_info{none, origin, false};
// TODO: Some new related hook? // TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid); //parent_->parent().notify<hook::new_connection_established>(nid);
} }
...@@ -73,7 +73,7 @@ void routing_table::add(const node_id& nid, const node_id& origin) { ...@@ -73,7 +73,7 @@ void routing_table::add(const node_id& nid, const node_id& origin) {
void routing_table::add(const node_id& nid) { void routing_table::add(const node_id& nid) {
//CAF_ASSERT(hdl_by_nid_.count(nid) == 0); //CAF_ASSERT(hdl_by_nid_.count(nid) == 0);
CAF_ASSERT(node_information_base_.count(nid) == 0); CAF_ASSERT(node_information_base_.count(nid) == 0);
node_information_base_[nid] = node_info{none, none}; node_information_base_[nid] = node_info{none, none, false};
// TODO: Some new related hook? // TODO: Some new related hook?
//parent_->parent().notify<hook::new_connection_established>(nid); //parent_->parent().notify<hook::new_connection_established>(nid);
} }
...@@ -129,6 +129,21 @@ const routing_table::address_map& routing_table::local_addresses() { ...@@ -129,6 +129,21 @@ const routing_table::address_map& routing_table::local_addresses() {
return local_addrs_; return local_addrs_;
} }
bool routing_table::received_client_handshake(const node_id& nid, bool flag) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
i->second.received_client_handshake = flag;
return true;
}
bool routing_table::received_client_handshake(const node_id& nid) {
auto i = node_information_base_.find(nid);
if (i == node_information_base_.end())
return false;
return i->second.received_client_handshake;
}
} // namespace basp } // namespace basp
} // namespace io } // namespace io
} // namespace caf } // namespace caf
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