From 57e29857cf79019af03f6a3dfe0bf6fd36e2fab2 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 15:27:53 -0700 Subject: Cluster work -- integrating with the rest of the code. --- node/Cluster.hpp | 118 ++++++++++++++++++++++++------------------------------- 1 file changed, 51 insertions(+), 67 deletions(-) (limited to 'node/Cluster.hpp') diff --git a/node/Cluster.hpp b/node/Cluster.hpp index 016730e3..f253e6f6 100644 --- a/node/Cluster.hpp +++ b/node/Cluster.hpp @@ -34,43 +34,32 @@ #include #include "Constants.hpp" +#include "../include/ZeroTierOne.h" #include "Address.hpp" #include "InetAddress.hpp" #include "SHA512.hpp" #include "Utils.hpp" #include "Buffer.hpp" #include "Mutex.hpp" +#include "SharedPtr.hpp" /** * Timeout for cluster members being considered "alive" */ -#define ZT_CLUSTER_TIMEOUT ZT_PEER_ACTIVITY_TIMEOUT +#define ZT_CLUSTER_TIMEOUT 30000 /** - * Maximum cluster message length in bytes - * - * Cluster nodes speak via TCP, with data encapsulated into individually - * encrypted and authenticated messages. The maximum message size is - * 65535 (0xffff) since the TCP stream uses 16-bit message size headers - * (and this is a reasonable chunk size anyway). - */ -#define ZT_CLUSTER_MAX_MESSAGE_LENGTH 65535 - -/** - * Maximum number of physical addresses we will cache for a cluster member - */ -#define ZT_CLUSTER_MEMBER_MAX_PHYSICAL_ADDRS 8 - -/** - * How frequently should doPeriodicTasks() be ideally called? (ms) + * Desired period between doPeriodicTasks() in milliseconds */ -#define ZT_CLUSTER_PERIODIC_TASK_DEADLINE 10 +#define ZT_CLUSTER_PERIODIC_TASK_PERIOD 50 namespace ZeroTier { class RuntimeEnvironment; class CertificateOfMembership; class MulticastGroup; +class Peer; +class Identity; /** * Multi-homing cluster state replication and packet relaying @@ -95,22 +84,6 @@ class MulticastGroup; class Cluster { public: - /** - * Which distance algorithm is this cluster using? - */ - enum DistanceAlgorithm - { - /** - * Simple linear distance in three dimensions - */ - DISTANCE_SIMPLE = 0, - - /** - * Haversine formula using X,Y as lat,long and ignoring Z - */ - DISTANCE_HAVERSINE = 1 - }; - /** * State message types */ @@ -184,25 +157,18 @@ public: /** * Construct a new cluster - * - * @param renv Runtime environment - * @param id This member's ID in the cluster - * @param da Distance algorithm this cluster uses to compute distance and hand off peers - * @param x My X - * @param y My Y - * @param z My Z - * @param sendFunction Function to call to send messages to other cluster members - * @param arg First argument to sendFunction */ Cluster( const RuntimeEnvironment *renv, uint16_t id, - DistanceAlgorithm da, + const std::vector &zeroTierPhysicalEndpoints, int32_t x, int32_t y, int32_t z, - void (*sendFunction)(void *,uint16_t,const void *,unsigned int), - void *arg); + void (*sendFunction)(void *,unsigned int,const void *,unsigned int), + void *sendFunctionArg, + int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *), + void *addressToLocationFunctionArg); ~Cluster(); @@ -222,9 +188,9 @@ public: /** * Advertise to the cluster that we have this peer * - * @param peerAddress Peer address that we have + * @param peerId Identity of peer that we have */ - void replicateHavePeer(const Address &peerAddress); + void replicateHavePeer(const Identity &peerId); /** * Advertise a multicast LIKE to the cluster @@ -243,7 +209,7 @@ public: void replicateCertificateOfNetworkMembership(const CertificateOfMembership &com); /** - * Call every ~ZT_CLUSTER_PERIODIC_TASK_DEADLINE milliseconds. + * Call every ~ZT_CLUSTER_PERIODIC_TASK_PERIOD milliseconds. */ void doPeriodicTasks(); @@ -254,6 +220,23 @@ public: */ void addMember(uint16_t memberId); + /** + * Remove a member ID from this cluster + * + * @param memberId Member ID to remove + */ + void removeMember(uint16_t memberId); + + /** + * Redirect this peer to a better cluster member if needed + * + * @param peer Peer to (possibly) redirect + * @param peerPhysicalAddress Physical address of peer's current best path (where packet was most recently received or getBestPath()->address()) + * @param offload Always redirect if possible -- can be used to offload peers during shutdown + * @return True if peer was redirected + */ + bool redirectPeer(const SharedPtr &peer,const InetAddress &peerPhysicalAddress,bool offload); + private: void _send(uint16_t memberId,const void *msg,unsigned int len); void _flush(uint16_t memberId); @@ -262,44 +245,45 @@ private: uint16_t _masterSecret[ZT_SHA512_DIGEST_LEN / sizeof(uint16_t)]; unsigned char _key[ZT_PEER_SECRET_KEY_LENGTH]; const RuntimeEnvironment *RR; - void (*_sendFunction)(void *,uint16_t,const void *,unsigned int); - void *_arg; + void (*_sendFunction)(void *,unsigned int,const void *,unsigned int); + void *_sendFunctionArg; + int (*_addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *); + void *_addressToLocationFunctionArg; const int32_t _x; const int32_t _y; const int32_t _z; - const DistanceAlgorithm _da; const uint16_t _id; + const std::vector _zeroTierPhysicalEndpoints; struct _Member { unsigned char key[ZT_PEER_SECRET_KEY_LENGTH]; - uint64_t lastReceivedFrom; uint64_t lastReceivedAliveAnnouncement; - uint64_t lastSentTo; uint64_t lastAnnouncedAliveTo; uint64_t load; int32_t x,y,z; - InetAddress physicalAddresses[ZT_CLUSTER_MEMBER_MAX_PHYSICAL_ADDRS]; - unsigned int physicalAddressCount; + std::vector zeroTierPhysicalEndpoints; Buffer q; Mutex lock; - _Member() : - lastReceivedFrom(0), - lastReceivedAliveAnnouncement(0), - lastSentTo(0), - lastAnnouncedAliveTo(0), - load(0), - x(0), - y(0), - z(0), - physicalAddressCount(0) {} - + inline void clear() + { + lastReceivedAliveAnnouncement = 0; + lastAnnouncedAliveTo = 0; + load = 0; + x = 0; + y = 0; + z = 0; + zeroTierPhysicalEndpoints.clear(); + q.clear(); + } + + _Member() { this->clear(); } ~_Member() { Utils::burn(key,sizeof(key)); } }; -- cgit v1.2.3 From eb79d4a2f34b34c49cd2d69efac22d9bc8ac27cb Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 16:24:21 -0700 Subject: Wire up peer announcement in cluster. --- node/Cluster.cpp | 72 +++++++++++++++++++++++++++++++++++++++---------- node/Cluster.hpp | 10 +++++-- node/IncomingPacket.cpp | 1 - node/Peer.cpp | 7 ++++- node/Topology.cpp | 27 ++++++++++--------- 5 files changed, 87 insertions(+), 30 deletions(-) (limited to 'node/Cluster.hpp') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index 5b76d1f0..f1dc45b9 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -263,8 +263,6 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) remotePeerAddress.appendTo(rendezvousForDest); Buffer<2048> rendezvousForOtherEnd; - rendezvousForOtherEnd.addSize(2); // leave room for payload size - rendezvousForOtherEnd.append((uint8_t)STATE_MESSAGE_PROXY_SEND); remotePeerAddress.appendTo(rendezvousForOtherEnd); rendezvousForOtherEnd.append((uint8_t)Packet::VERB_RENDEZVOUS); const unsigned int rendezvousForOtherEndPayloadSizePtr = rendezvousForOtherEnd.size(); @@ -298,9 +296,8 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) } if (haveMatch) { + _send(fromMemberId,STATE_MESSAGE_PROXY_SEND,rendezvousForOtherEnd.data(),rendezvousForOtherEnd.size()); RR->sw->send(rendezvousForDest,true,0); - rendezvousForOtherEnd.setAt(0,(uint16_t)(rendezvousForOtherEnd.size() - 2)); - _send(fromMemberId,rendezvousForOtherEnd.data(),rendezvousForOtherEnd.size()); } } } @@ -331,14 +328,64 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) void Cluster::replicateHavePeer(const Identity &peerId) { + { // Use peer affinity table to track our own last announce time for peers + _PeerAffinity pa(peerId.address(),_id,RR->node->now()); + Mutex::Lock _l2(_peerAffinities_m); + std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),pa)); // O(log(n)) + if ((i != _peerAffinities.end())&&(i->key == pa.key)) { + if ((pa.timestamp - i->timestamp) >= ZT_CLUSTER_HAVE_PEER_ANNOUNCE_PERIOD) { + i->timestamp = pa.timestamp; + // continue to announcement + } else { + // we've already announced this peer recently, so skip + return; + } + } else { + _peerAffinities.push_back(pa); + std::sort(_peerAffinities.begin(),_peerAffinities.end()); // probably a more efficient way to insert but okay for now + // continue to announcement + } + } + + // announcement + Buffer<4096> buf; + peerId.serialize(buf,false); + { + Mutex::Lock _l(_memberIds_m); + for(std::vector::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) { + Mutex::Lock _l2(_members[*mid].lock); + _send(*mid,STATE_MESSAGE_HAVE_PEER,buf.data(),buf.size()); + } + } } void Cluster::replicateMulticastLike(uint64_t nwid,const Address &peerAddress,const MulticastGroup &group) { + Buffer<4096> buf; + buf.append((uint64_t)nwid); + peerAddress.appendTo(buf); + group.mac().appendTo(buf); + buf.append((uint32_t)group.adi()); + { + Mutex::Lock _l(_memberIds_m); + for(std::vector::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) { + Mutex::Lock _l2(_members[*mid].lock); + _send(*mid,STATE_MESSAGE_MULTICAST_LIKE,buf.data(),buf.size()); + } + } } void Cluster::replicateCertificateOfNetworkMembership(const CertificateOfMembership &com) { + Buffer<4096> buf; + com.serialize(buf); + { + Mutex::Lock _l(_memberIds_m); + for(std::vector::const_iterator mid(_memberIds.begin());mid!=_memberIds.end();++mid) { + Mutex::Lock _l2(_members[*mid].lock); + _send(*mid,STATE_MESSAGE_COM,buf.data(),buf.size()); + } + } } void Cluster::doPeriodicTasks() @@ -371,7 +418,7 @@ void Cluster::doPeriodicTasks() alive.append((uint8_t)_zeroTierPhysicalEndpoints.size()); for(std::vector::const_iterator pe(_zeroTierPhysicalEndpoints.begin());pe!=_zeroTierPhysicalEndpoints.end();++pe) pe->serialize(alive); - _send(*mid,alive.data(),alive.size()); + _send(*mid,STATE_MESSAGE_ALIVE,alive.data(),alive.size()); _members[*mid].lastAnnouncedAliveTo = now; } @@ -498,18 +545,15 @@ bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPh } } -void Cluster::_send(uint16_t memberId,const void *msg,unsigned int len) +void Cluster::_send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len) { _Member &m = _members[memberId]; // assumes m.lock is locked! - for(;;) { - if ((m.q.size() + len) > ZT_CLUSTER_MAX_MESSAGE_LENGTH) - _flush(memberId); - else { - m.q.append(msg,len); - break; - } - } + if ((m.q.size() + len + 3) > ZT_CLUSTER_MAX_MESSAGE_LENGTH) + _flush(memberId); + m.q.append((uint16_t)(len + 1)); + m.q.append((uint8_t)type); + m.q.append(msg,len); } void Cluster::_flush(uint16_t memberId) diff --git a/node/Cluster.hpp b/node/Cluster.hpp index f253e6f6..df061c60 100644 --- a/node/Cluster.hpp +++ b/node/Cluster.hpp @@ -42,12 +42,18 @@ #include "Buffer.hpp" #include "Mutex.hpp" #include "SharedPtr.hpp" +#include "Hashtable.hpp" /** * Timeout for cluster members being considered "alive" */ #define ZT_CLUSTER_TIMEOUT 30000 +/** + * How often should we announce that we have a peer? + */ +#define ZT_CLUSTER_HAVE_PEER_ANNOUNCE_PERIOD 60000 + /** * Desired period between doPeriodicTasks() in milliseconds */ @@ -238,7 +244,7 @@ public: bool redirectPeer(const SharedPtr &peer,const InetAddress &peerPhysicalAddress,bool offload); private: - void _send(uint16_t memberId,const void *msg,unsigned int len); + void _send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len); void _flush(uint16_t memberId); // These are initialized in the constructor and remain static @@ -292,7 +298,7 @@ private: std::vector _memberIds; Mutex _memberIds_m; - // Record tracking which members have which peers and how recently they claimed this + // Record tracking which members have which peers and how recently they claimed this -- also used to track our last claimed time struct _PeerAffinity { _PeerAffinity(const Address &a,uint16_t mid,uint64_t ts) : diff --git a/node/IncomingPacket.cpp b/node/IncomingPacket.cpp index a4d45068..7a47c0c6 100644 --- a/node/IncomingPacket.cpp +++ b/node/IncomingPacket.cpp @@ -272,7 +272,6 @@ bool IncomingPacket::_doHELLO(const RuntimeEnvironment *RR) TRACE("rejected HELLO from %s(%s): packet failed authentication",id.address().toString().c_str(),_remoteAddress.toString().c_str()); return true; } - peer = RR->topology->addPeer(newPeer); // Continue at // VALID diff --git a/node/Peer.cpp b/node/Peer.cpp index 0ba379c6..45e2fedd 100644 --- a/node/Peer.cpp +++ b/node/Peer.cpp @@ -34,6 +34,7 @@ #include "Network.hpp" #include "AntiRecursion.hpp" #include "SelfAwareness.hpp" +#include "Cluster.hpp" #include @@ -107,7 +108,6 @@ void Peer::received( // Learn paths if they've been confirmed via a HELLO or an ECHO RemotePath *slot = (RemotePath *)0; if (np < ZT_MAX_PEER_NETWORK_PATHS) { - // Add new path slot = &(_paths[np++]); } else { uint64_t slotLRmin = 0xffffffffffffffffULL; @@ -141,6 +141,11 @@ void Peer::received( } } } + +#ifdef ZT_ENABLE_CLUSTER + if ((pathIsConfirmed)&&(RR->cluster)) + RR->cluster->replicateHavePeer(_id); +#endif } if ((now - _lastAnnouncedTo) >= ((ZT_MULTICAST_LIKE_EXPIRE / 2) - 1000)) { diff --git a/node/Topology.cpp b/node/Topology.cpp index 56ca47c8..88c8856c 100644 --- a/node/Topology.cpp +++ b/node/Topology.cpp @@ -122,18 +122,22 @@ Topology::~Topology() SharedPtr Topology::addPeer(const SharedPtr &peer) { if (peer->address() == RR->identity.address()) { - TRACE("BUG: addNewPeer() caught and ignored attempt to add peer for self"); + TRACE("BUG: addPeer() caught and ignored attempt to add peer for self"); throw std::logic_error("cannot add peer for self"); } - const uint64_t now = RR->node->now(); - Mutex::Lock _l(_lock); - - SharedPtr &p = _peers.set(peer->address(),peer); - p->use(now); - saveIdentity(p->identity()); + SharedPtr np; + { + Mutex::Lock _l(_lock); + SharedPtr &hp = _peers[peer->address()]; + if (!hp) + hp = peer; + np = hp; + } + np->use(RR->node->now()); + saveIdentity(np->identity()); - return p; + return np; } SharedPtr Topology::getPeer(const Address &zta) @@ -143,13 +147,12 @@ SharedPtr Topology::getPeer(const Address &zta) return SharedPtr(); } - const uint64_t now = RR->node->now(); Mutex::Lock _l(_lock); SharedPtr &ap = _peers[zta]; if (ap) { - ap->use(now); + ap->use(RR->node->now()); return ap; } @@ -157,13 +160,13 @@ SharedPtr Topology::getPeer(const Address &zta) if (id) { try { ap = SharedPtr(new Peer(RR->identity,id)); - ap->use(now); + ap->use(RR->node->now()); return ap; } catch ( ... ) {} // invalid identity? } + // If we get here it means we read an invalid cache identity or had some other error _peers.erase(zta); - return SharedPtr(); } -- cgit v1.2.3 From d6dee7bb5ccb4ae1fcf818afc246befc9133be72 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 17:22:53 -0700 Subject: Clustered handling of relaying. --- node/Cluster.cpp | 47 +++++++++++++++++++++++++++++++++++++++++++++++ node/Cluster.hpp | 11 +++++++++++ node/Switch.cpp | 12 +++++++++++- 3 files changed, 69 insertions(+), 1 deletion(-) (limited to 'node/Cluster.hpp') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index 41685a07..7f4a21b0 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -352,6 +352,53 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) } } +bool Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPeerAddress,const void *data,unsigned int len) +{ + if (len > 16384) // sanity check + return false; + + uint64_t mostRecentTimestamp = 0; + uint16_t canHasPeer = 0; + + { // Anyone got this peer? + Mutex::Lock _l2(_peerAffinities_m); + std::vector<_PeerAffinity>::iterator i(std::lower_bound(_peerAffinities.begin(),_peerAffinities.end(),_PeerAffinity(toPeerAddress,0,0))); // O(log(n)) + while ((i != _peerAffinities.end())&&(i->address() == toPeerAddress)) { + uint16_t mid = i->clusterMemberId(); + if ((mid != _id)&&(i->timestamp > mostRecentTimestamp)) { + mostRecentTimestamp = i->timestamp; + canHasPeer = mid; + } + } + } + + const uint64_t now = RR->node->now(); + if ((now - mostRecentTimestamp) < ZT_PEER_ACTIVITY_TIMEOUT) { + Buffer<16384> buf; + + InetAddress v4,v6; + if (fromPeerAddress) { + SharedPtr fromPeer(RR->topology->getPeer(fromPeerAddress)); + if (fromPeer) + fromPeer->getBestActiveAddresses(now,v4,v6); + } + buf.append((uint8_t)( (v4) ? ((v6) ? 2 : 1) : ((v6) ? 1 : 0) )); + if (v4) + v4.serialize(buf); + if (v6) + v6.serialize(buf); + buf.append((uint16_t)len); + buf.append(data,len); + + { + Mutex::Lock _l2(_members[canHasPeer].lock); + _send(canHasPeer,STATE_MESSAGE_RELAY,buf.data(),buf.size()); + } + } + + return false; +} + void Cluster::replicateHavePeer(const Identity &peerId) { { // Use peer affinity table to track our own last announce time for peers diff --git a/node/Cluster.hpp b/node/Cluster.hpp index df061c60..60157b15 100644 --- a/node/Cluster.hpp +++ b/node/Cluster.hpp @@ -191,6 +191,17 @@ public: */ void handleIncomingStateMessage(const void *msg,unsigned int len); + /** + * Send this packet via another node in this cluster if another node has this peer + * + * @param fromPeerAddress Source peer address (if known, should be NULL for fragments) + * @param toPeerAddress Destination peer address + * @param data Packet or packet fragment data + * @param len Length of packet or fragment + * @return True if this data was sent via another cluster member, false if none have this peer + */ + bool sendViaCluster(const Address &fromPeerAddress,const Address &toPeerAddress,const void *data,unsigned int len); + /** * Advertise to the cluster that we have this peer * diff --git a/node/Switch.cpp b/node/Switch.cpp index 249a21d5..0edaa96d 100644 --- a/node/Switch.cpp +++ b/node/Switch.cpp @@ -45,6 +45,7 @@ #include "AntiRecursion.hpp" #include "SelfAwareness.hpp" #include "Packet.hpp" +#include "Cluster.hpp" namespace ZeroTier { @@ -567,6 +568,11 @@ void Switch::_handleRemotePacketFragment(const InetAddress &localAddr,const Inet // It wouldn't hurt anything, just redundant and unnecessary. SharedPtr relayTo = RR->topology->getPeer(destination); if ((!relayTo)||(!relayTo->send(RR,fragment.data(),fragment.size(),RR->node->now()))) { +#ifdef ZT_ENABLE_CLUSTER + if ((RR->cluster)&&(RR->cluster->sendViaCluster(Address(),destination,fragment.data(),fragment.size()))) + return; // sent by way of another member of this cluster +#endif + // Don't know peer or no direct path -- so relay via root server relayTo = RR->topology->getBestRoot(); if (relayTo) @@ -642,7 +648,11 @@ void Switch::_handleRemotePacketHead(const InetAddress &localAddr,const InetAddr if ((relayTo)&&((relayTo->send(RR,packet->data(),packet->size(),RR->node->now())))) { unite(source,destination,false); } else { - // Don't know peer or no direct path -- so relay via root server +#ifdef ZT_ENABLE_CLUSTER + if ((RR->cluster)&&(RR->cluster->sendViaCluster(source,destination,packet->data(),packet->size()))) + return; // sent by way of another member of this cluster +#endif + relayTo = RR->topology->getBestRoot(&source,1,true); if (relayTo) relayTo->send(RR,packet->data(),packet->size(),RR->node->now()); -- cgit v1.2.3 From 978b056a0134edd9598a3edb219b31c61405c3b4 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Tue, 20 Oct 2015 17:36:10 -0700 Subject: Wire in redirectPeer(), now about ready to test clustering! --- node/Cluster.cpp | 8 ++++---- node/Cluster.hpp | 4 ++-- node/Peer.cpp | 16 +++++++++------- 3 files changed, 15 insertions(+), 13 deletions(-) (limited to 'node/Cluster.hpp') diff --git a/node/Cluster.cpp b/node/Cluster.cpp index abe970e0..d9514db5 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -549,7 +549,7 @@ void Cluster::removeMember(uint16_t memberId) _memberIds = newMemberIds; } -bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPhysicalAddress,bool offload) +bool Cluster::redirectPeer(const Address &peerAddress,const InetAddress &peerPhysicalAddress,bool offload) { if (!peerPhysicalAddress) // sanity check return false; @@ -585,7 +585,7 @@ bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPh } if (best.size() > 0) { - TRACE("peer %s is at [%d,%d,%d], distance to us is %f, sending to %u instead for better distance %f",peer->address().toString().c_str(),px,py,pz,currentDistance,bestMember,bestDistance); + TRACE("peer %s is at [%d,%d,%d], distance to us is %f, sending to %u instead for better distance %f",peerAddress.toString().c_str(),px,py,pz,currentDistance,bestMember,bestDistance); /* if (peer->remoteVersionProtocol() >= 5) { // If it's a newer peer send VERB_PUSH_DIRECT_PATHS which is more idiomatic @@ -593,7 +593,7 @@ bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPh // Otherwise send VERB_RENDEZVOUS for ourselves, which will trick peers into trying other endpoints for us even if they're too old for PUSH_DIRECT_PATHS for(std::vector::const_iterator a(best.begin());a!=best.end();++a) { if ((a->ss_family == AF_INET)||(a->ss_family == AF_INET6)) { - Packet outp(peer->address(),RR->identity.address(),Packet::VERB_RENDEZVOUS); + Packet outp(peerAddress,RR->identity.address(),Packet::VERB_RENDEZVOUS); outp.append((uint8_t)0); // no flags RR->identity.address().appendTo(outp); // HACK: rendezvous with ourselves! with really old peers this will only work if I'm a root server! outp.append((uint16_t)a->port()); @@ -611,7 +611,7 @@ bool Cluster::redirectPeer(const SharedPtr &peer,const InetAddress &peerPh return true; } else { - TRACE("peer %s is at [%d,%d,%d], distance to us is %f and this seems to be the best",peer->address().toString().c_str(),px,py,pz,currentDistance); + TRACE("peer %s is at [%d,%d,%d], distance to us is %f and this seems to be the best",peerAddress.toString().c_str(),px,py,pz,currentDistance); return false; } } else { diff --git a/node/Cluster.hpp b/node/Cluster.hpp index 60157b15..2e60fd6b 100644 --- a/node/Cluster.hpp +++ b/node/Cluster.hpp @@ -247,12 +247,12 @@ public: /** * Redirect this peer to a better cluster member if needed * - * @param peer Peer to (possibly) redirect + * @param peerAddress Peer to (possibly) redirect * @param peerPhysicalAddress Physical address of peer's current best path (where packet was most recently received or getBestPath()->address()) * @param offload Always redirect if possible -- can be used to offload peers during shutdown * @return True if peer was redirected */ - bool redirectPeer(const SharedPtr &peer,const InetAddress &peerPhysicalAddress,bool offload); + bool redirectPeer(const Address &peerAddress,const InetAddress &peerPhysicalAddress,bool offload); private: void _send(uint16_t memberId,StateMessageType type,const void *msg,unsigned int len); diff --git a/node/Peer.cpp b/node/Peer.cpp index 45e2fedd..fa7b3aa4 100644 --- a/node/Peer.cpp +++ b/node/Peer.cpp @@ -82,6 +82,7 @@ void Peer::received( { const uint64_t now = RR->node->now(); bool needMulticastGroupAnnounce = false; + bool pathIsConfirmed = false; { Mutex::Lock _l(_lock); @@ -89,8 +90,6 @@ void Peer::received( _lastReceive = now; if (!hops) { - bool pathIsConfirmed = false; - /* Learn new paths from direct (hops == 0) packets */ { unsigned int np = _numPaths; @@ -141,11 +140,6 @@ void Peer::received( } } } - -#ifdef ZT_ENABLE_CLUSTER - if ((pathIsConfirmed)&&(RR->cluster)) - RR->cluster->replicateHavePeer(_id); -#endif } if ((now - _lastAnnouncedTo) >= ((ZT_MULTICAST_LIKE_EXPIRE / 2) - 1000)) { @@ -159,6 +153,14 @@ void Peer::received( _lastMulticastFrame = now; } +#ifdef ZT_ENABLE_CLUSTER + if ((pathIsConfirmed)&&(RR->cluster)) { + // Either shuttle this peer off somewhere else or report to other members that we have it + if (!RR->cluster->redirectPeer(_id.address(),remoteAddr,false)) + RR->cluster->replicateHavePeer(_id); + } +#endif + if (needMulticastGroupAnnounce) { const std::vector< SharedPtr > networks(RR->node->allNetworks()); for(std::vector< SharedPtr >::const_iterator n(networks.begin());n!=networks.end();++n) -- cgit v1.2.3