From cc1b275ad97bf186f21b487aa57d7893bee3c956 Mon Sep 17 00:00:00 2001
From: Adam Ierymenko <adam.ierymenko@gmail.com>
Date: Tue, 27 Oct 2015 16:47:13 -0700
Subject: [PATCH] Replicate peer endpoints and forget paths if we have them --
this allows two clusters to talk to each other, whereas forgetting all paths
does not.
---
node/Cluster.cpp | 45 +++++++++++++++++++++++++++------------------
node/Cluster.hpp | 7 ++++++-
node/Constants.hpp | 4 ++--
node/Peer.cpp | 11 ++++-------
node/Peer.hpp | 19 +++++++++++++++++++
5 files changed, 58 insertions(+), 28 deletions(-)
diff --git a/node/Cluster.cpp b/node/Cluster.cpp
index b2f3d5854..0797d83d0 100644
--- a/node/Cluster.cpp
+++ b/node/Cluster.cpp
@@ -210,22 +210,30 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len)
} break;
case STATE_MESSAGE_HAVE_PEER: {
- try {
- Identity id;
- ptr += id.deserialize(dmsg,ptr);
- if (id) {
- RR->topology->saveIdentity(id);
- {
- Mutex::Lock _l2(_peerAffinities_m);
- _PA &pa = _peerAffinities[id.address()];
- pa.ts = RR->node->now();
- pa.mid = fromMemberId;
- }
- TRACE("[%u] has %s",(unsigned int)fromMemberId,id.address().toString().c_str());
- }
- } catch ( ... ) {
- // ignore invalid identities
- }
+ Identity id;
+ InetAddress physicalAddress;
+ ptr += id.deserialize(dmsg,ptr);
+ ptr += physicalAddress.deserialize(dmsg,ptr);
+ if (id) {
+ // Forget any paths that we have to this peer at its address
+ if (physicalAddress) {
+ SharedPtr<Peer> myPeerRecord(RR->topology->getPeer(id.address()));
+ if (myPeerRecord)
+ myPeerRecord->removePathByAddress(physicalAddress);
+ }
+
+ // Always save identity to update file time
+ RR->topology->saveIdentity(id);
+
+ // Set peer affinity to its new home
+ {
+ Mutex::Lock _l2(_peerAffinities_m);
+ _PA &pa = _peerAffinities[id.address()];
+ pa.ts = RR->node->now();
+ pa.mid = fromMemberId;
+ }
+ TRACE("[%u] has %s @ %s",(unsigned int)fromMemberId,id.address().toString().c_str(),physicalAddress.toString().c_str());
+ }
} break;
case STATE_MESSAGE_MULTICAST_LIKE: {
@@ -396,7 +404,7 @@ bool Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPee
return true;
}
-void Cluster::replicateHavePeer(const Identity &peerId)
+void Cluster::replicateHavePeer(const Identity &peerId,const InetAddress &physicalAddress)
{
const uint64_t now = RR->node->now();
{ // Use peer affinity table to track our own last announce time for peers
@@ -405,7 +413,7 @@ void Cluster::replicateHavePeer(const Identity &peerId)
if (pa.mid != _id) {
pa.ts = now;
pa.mid = _id;
- } else if ((now - pa.ts) >= ZT_CLUSTER_HAVE_PEER_ANNOUNCE_PERIOD) {
+ } else if ((now - pa.ts) < ZT_CLUSTER_HAVE_PEER_ANNOUNCE_PERIOD) {
return;
} else {
pa.ts = now;
@@ -415,6 +423,7 @@ void Cluster::replicateHavePeer(const Identity &peerId)
// announcement
Buffer<4096> buf;
peerId.serialize(buf,false);
+ physicalAddress.serialize(buf);
{
Mutex::Lock _l(_memberIds_m);
for(std::vector<uint16_t>::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) {
diff --git a/node/Cluster.hpp b/node/Cluster.hpp
index 42a26c7ff..c3367d574 100644
--- a/node/Cluster.hpp
+++ b/node/Cluster.hpp
@@ -117,6 +117,10 @@ public:
/**
* Cluster member has this peer:
* <[...] binary serialized peer identity>
+ * <[...] binary serialized peer remote physical address>
+ *
+ * Clusters send this message when they learn a path to a peer. The
+ * replicated physical address is the one learned.
*/
STATE_MESSAGE_HAVE_PEER = 2,
@@ -225,8 +229,9 @@ public:
* Advertise to the cluster that we have this peer
*
* @param peerId Identity of peer that we have
+ * @param physicalAddress Physical address of peer (from our POV)
*/
- void replicateHavePeer(const Identity &peerId);
+ void replicateHavePeer(const Identity &peerId,const InetAddress &physicalAddress);
/**
* Advertise a multicast LIKE to the cluster
diff --git a/node/Constants.hpp b/node/Constants.hpp
index bef1183a8..4b06db449 100644
--- a/node/Constants.hpp
+++ b/node/Constants.hpp
@@ -317,7 +317,7 @@
/**
* Minimum delay between attempts to confirm new paths to peers (to avoid HELLO flooding)
*/
-#define ZT_MIN_PATH_CONFIRMATION_INTERVAL 5000
+#define ZT_MIN_PATH_CONFIRMATION_INTERVAL 1000
/**
* Interval between direct path pushes in milliseconds
@@ -350,7 +350,7 @@
/**
* Maximum number of endpoints to contact per address type (to limit pushes like GitHub issue #235)
*/
-#define ZT_PUSH_DIRECT_PATHS_MAX_ENDPOINTS_PER_TYPE 2
+#define ZT_PUSH_DIRECT_PATHS_MAX_ENDPOINTS_PER_TYPE 4
/**
* A test pseudo-network-ID that can be joined
diff --git a/node/Peer.cpp b/node/Peer.cpp
index 99e2156e5..99eb32c71 100644
--- a/node/Peer.cpp
+++ b/node/Peer.cpp
@@ -140,13 +140,10 @@ void Peer::received(
_lastMulticastFrame = now;
#ifdef ZT_ENABLE_CLUSTER
- // If we're in cluster mode and there's a better endpoint, stop here and don't
- // learn or confirm paths. Also reset any existing paths, since they should
- // go there and no longer talk to us here.
- if (redirectTo) {
- _numPaths = 0;
+ // If we think this peer belongs elsewhere, don't learn this path or
+ // do other connection init stuff.
+ if (redirectTo)
return;
- }
#endif
if ((now - _lastAnnouncedTo) >= ((ZT_MULTICAST_LIKE_EXPIRE / 2) - 1000)) {
@@ -206,7 +203,7 @@ void Peer::received(
#ifdef ZT_ENABLE_CLUSTER
if ((RR->cluster)&&(pathIsConfirmed))
- RR->cluster->replicateHavePeer(_id);
+ RR->cluster->replicateHavePeer(_id,remoteAddr);
#endif
if (needMulticastGroupAnnounce) {
diff --git a/node/Peer.hpp b/node/Peer.hpp
index aa75b3f48..69343f20c 100644
--- a/node/Peer.hpp
+++ b/node/Peer.hpp
@@ -412,6 +412,25 @@ public:
*/
void clean(const RuntimeEnvironment *RR,uint64_t now);
+ /**
+ * Remove all paths with this remote address
+ *
+ * @param addr Remote address to remove
+ */
+ inline void removePathByAddress(const InetAddress &addr)
+ {
+ Mutex::Lock _l(_lock);
+ unsigned int np = _numPaths;
+ unsigned int x = 0;
+ unsigned int y = 0;
+ while (x < np) {
+ if (_paths[x].address() != addr)
+ _paths[y++] = _paths[x];
+ ++x;
+ }
+ _numPaths = y;
+ }
+
/**
* Find a common set of addresses by which two peers can link, if any
*