Refactor path stability stuff and add basic multipath support.

This commit is contained in:
Adam Ierymenko 2017-10-25 12:42:14 -07:00
parent 0d8b8d8426
commit 459f1e7bfb
14 changed files with 379 additions and 362 deletions

View File

@ -268,16 +268,6 @@
*/ */
#define ZT_PATH_HEARTBEAT_PERIOD 14000 #define ZT_PATH_HEARTBEAT_PERIOD 14000
/**
* Paths are considered inactive if they have not received traffic in this long
*/
#define ZT_PATH_ALIVE_TIMEOUT 45000
/**
* Minimum time between attempts to check dead paths to see if they can be re-awakened
*/
#define ZT_PATH_MIN_REACTIVATE_INTERVAL 2500
/** /**
* Do not accept HELLOs over a given path more often than this * Do not accept HELLOs over a given path more often than this
*/ */

View File

@ -42,11 +42,6 @@ namespace ZeroTier {
/** /**
* A minimal hash table implementation for the ZeroTier core * A minimal hash table implementation for the ZeroTier core
*
* This is not a drop-in replacement for STL containers, and has several
* limitations. Keys can be uint64_t or an object, and if the latter they
* must implement a method called hashCode() that returns an unsigned long
* value that is evenly distributed.
*/ */
template<typename K,typename V> template<typename K,typename V>
class Hashtable class Hashtable

View File

@ -446,7 +446,8 @@ bool IncomingPacket::_doOK(const RuntimeEnvironment *RR,void *tPtr,const SharedP
} }
if (!hops()) if (!hops())
peer->addDirectLatencyMeasurment((unsigned int)latency); _path->updateLatency((unsigned int)latency);
peer->setRemoteVersion(vProto,vMajor,vMinor,vRevision); peer->setRemoteVersion(vProto,vMajor,vMinor,vRevision);
if ((externalSurfaceAddress)&&(hops() == 0)) if ((externalSurfaceAddress)&&(hops() == 0))
@ -1091,7 +1092,7 @@ bool IncomingPacket::_doPUSH_DIRECT_PATHS(const RuntimeEnvironment *RR,void *tPt
(RR->node->shouldUsePathForZeroTierTraffic(tPtr,peer->address(),_path->localSocket(),a)) ) // should use path (RR->node->shouldUsePathForZeroTierTraffic(tPtr,peer->address(),_path->localSocket(),a)) ) // should use path
{ {
if ((flags & ZT_PUSH_DIRECT_PATHS_FLAG_CLUSTER_REDIRECT) != 0) { if ((flags & ZT_PUSH_DIRECT_PATHS_FLAG_CLUSTER_REDIRECT) != 0) {
peer->redirect(tPtr,_path->localSocket(),a,now); peer->clusterRedirect(tPtr,_path->localSocket(),a,now);
} else if (++countPerScope[(int)a.ipScope()][0] <= ZT_PUSH_DIRECT_PATHS_MAX_PER_SCOPE_AND_FAMILY) { } else if (++countPerScope[(int)a.ipScope()][0] <= ZT_PUSH_DIRECT_PATHS_MAX_PER_SCOPE_AND_FAMILY) {
peer->attemptToContactAt(tPtr,InetAddress(),a,now,false,0); peer->attemptToContactAt(tPtr,InetAddress(),a,now,false,0);
} }
@ -1105,7 +1106,7 @@ bool IncomingPacket::_doPUSH_DIRECT_PATHS(const RuntimeEnvironment *RR,void *tPt
(RR->node->shouldUsePathForZeroTierTraffic(tPtr,peer->address(),_path->localSocket(),a)) ) // should use path (RR->node->shouldUsePathForZeroTierTraffic(tPtr,peer->address(),_path->localSocket(),a)) ) // should use path
{ {
if ((flags & ZT_PUSH_DIRECT_PATHS_FLAG_CLUSTER_REDIRECT) != 0) { if ((flags & ZT_PUSH_DIRECT_PATHS_FLAG_CLUSTER_REDIRECT) != 0) {
peer->redirect(tPtr,_path->localSocket(),a,now); peer->clusterRedirect(tPtr,_path->localSocket(),a,now);
} else if (++countPerScope[(int)a.ipScope()][1] <= ZT_PUSH_DIRECT_PATHS_MAX_PER_SCOPE_AND_FAMILY) { } else if (++countPerScope[(int)a.ipScope()][1] <= ZT_PUSH_DIRECT_PATHS_MAX_PER_SCOPE_AND_FAMILY) {
peer->attemptToContactAt(tPtr,InetAddress(),a,now,false,0); peer->attemptToContactAt(tPtr,InetAddress(),a,now,false,0);
} }

View File

@ -273,9 +273,8 @@ InetAddress InetAddress::network() const
return r; return r;
} }
#ifdef ZT_SDK bool InetAddress::isEqualPrefix(const InetAddress &addr) const
bool InetAddress::isEqualPrefix(const InetAddress &addr) const {
{
if (addr.ss_family == ss_family) { if (addr.ss_family == ss_family) {
switch(ss_family) { switch(ss_family) {
case AF_INET6: { case AF_INET6: {
@ -294,8 +293,7 @@ InetAddress InetAddress::network() const
} }
} }
return false; return false;
} }
#endif
bool InetAddress::containsAddress(const InetAddress &addr) const bool InetAddress::containsAddress(const InetAddress &addr) const
{ {

View File

@ -330,7 +330,6 @@ struct InetAddress : public sockaddr_storage
*/ */
InetAddress network() const; InetAddress network() const;
#ifdef ZT_SDK
/** /**
* Test whether this IPv6 prefix matches the prefix of a given IPv6 address * Test whether this IPv6 prefix matches the prefix of a given IPv6 address
* *
@ -338,7 +337,6 @@ struct InetAddress : public sockaddr_storage
* @return True if this IPv6 prefix matches the prefix of a given IPv6 address * @return True if this IPv6 prefix matches the prefix of a given IPv6 address
*/ */
bool isEqualPrefix(const InetAddress &addr) const; bool isEqualPrefix(const InetAddress &addr) const;
#endif
/** /**
* Test whether this IP/netmask contains this address * Test whether this IP/netmask contains this address

View File

@ -191,12 +191,13 @@ public:
{ {
const std::vector<InetAddress> *const upstreamStableEndpoints = _upstreamsToContact.get(p->address()); const std::vector<InetAddress> *const upstreamStableEndpoints = _upstreamsToContact.get(p->address());
if (upstreamStableEndpoints) { if (upstreamStableEndpoints) {
bool contacted = false;
// Upstreams must be pinged constantly over both IPv4 and IPv6 to allow // Upstreams must be pinged constantly over both IPv4 and IPv6 to allow
// them to perform three way handshake introductions for both stacks. // them to perform three way handshake introductions for both stacks.
if (!p->doPingAndKeepalive(_tPtr,_now,AF_INET)) { const unsigned int sent = p->doPingAndKeepalive(_tPtr,_now);
bool contacted = (sent != 0);
if ((sent & 0x1) == 0) { // bit 0x1 == IPv4 sent
for(unsigned long k=0,ptr=(unsigned long)RR->node->prng();k<(unsigned long)upstreamStableEndpoints->size();++k) { for(unsigned long k=0,ptr=(unsigned long)RR->node->prng();k<(unsigned long)upstreamStableEndpoints->size();++k) {
const InetAddress &addr = (*upstreamStableEndpoints)[ptr++ % upstreamStableEndpoints->size()]; const InetAddress &addr = (*upstreamStableEndpoints)[ptr++ % upstreamStableEndpoints->size()];
if (addr.ss_family == AF_INET) { if (addr.ss_family == AF_INET) {
@ -205,8 +206,9 @@ public:
break; break;
} }
} }
} else contacted = true; }
if (!p->doPingAndKeepalive(_tPtr,_now,AF_INET6)) {
if ((sent & 0x2) == 0) { // bit 0x2 == IPv6 sent
for(unsigned long k=0,ptr=(unsigned long)RR->node->prng();k<(unsigned long)upstreamStableEndpoints->size();++k) { for(unsigned long k=0,ptr=(unsigned long)RR->node->prng();k<(unsigned long)upstreamStableEndpoints->size();++k) {
const InetAddress &addr = (*upstreamStableEndpoints)[ptr++ % upstreamStableEndpoints->size()]; const InetAddress &addr = (*upstreamStableEndpoints)[ptr++ % upstreamStableEndpoints->size()];
if (addr.ss_family == AF_INET6) { if (addr.ss_family == AF_INET6) {
@ -215,8 +217,10 @@ public:
break; break;
} }
} }
} else contacted = true; }
// If we have no memoized addresses for this upstream peer, attempt to contact
// it indirectly so we will be introduced.
if ((!contacted)&&(_bestCurrentUpstream)) { if ((!contacted)&&(_bestCurrentUpstream)) {
const SharedPtr<Path> up(_bestCurrentUpstream->getBestPath(_now,true)); const SharedPtr<Path> up(_bestCurrentUpstream->getBestPath(_now,true));
if (up) if (up)
@ -224,9 +228,11 @@ public:
} }
lastReceiveFromUpstream = std::max(p->lastReceive(),lastReceiveFromUpstream); lastReceiveFromUpstream = std::max(p->lastReceive(),lastReceiveFromUpstream);
_upstreamsToContact.erase(p->address()); // erase from upstreams to contact so that we can WHOIS those that remain
_upstreamsToContact.erase(p->address()); // after this we'll WHOIS all upstreams that remain
} else if (p->isActive(_now)) { } else if (p->isActive(_now)) {
p->doPingAndKeepalive(_tPtr,_now,-1); // Regular non-upstream nodes get pinged if they appear active.
p->doPingAndKeepalive(_tPtr,_now);
} }
} }
@ -420,7 +426,7 @@ ZT_PeerList *Node::peers() const
p->versionMinor = -1; p->versionMinor = -1;
p->versionRev = -1; p->versionRev = -1;
} }
p->latency = pi->second->latency(); p->latency = pi->second->latency(_now);
p->role = RR->topology->role(pi->second->identity().address()); p->role = RR->topology->role(pi->second->identity().address());
std::vector< SharedPtr<Path> > paths(pi->second->paths(_now)); std::vector< SharedPtr<Path> > paths(pi->second->paths(_now));

View File

@ -100,6 +100,7 @@ public:
_incomingLinkQualitySlowLogCounter(-64), // discard first fast log _incomingLinkQualitySlowLogCounter(-64), // discard first fast log
_incomingLinkQualityPreviousPacketCounter(0), _incomingLinkQualityPreviousPacketCounter(0),
_outgoingPacketCounter(0), _outgoingPacketCounter(0),
_latency(0xffff),
_addr(), _addr(),
_ipScope(InetAddress::IP_SCOPE_NONE) _ipScope(InetAddress::IP_SCOPE_NONE)
{ {
@ -117,6 +118,7 @@ public:
_incomingLinkQualitySlowLogCounter(-64), // discard first fast log _incomingLinkQualitySlowLogCounter(-64), // discard first fast log
_incomingLinkQualityPreviousPacketCounter(0), _incomingLinkQualityPreviousPacketCounter(0),
_outgoingPacketCounter(0), _outgoingPacketCounter(0),
_latency(0xffff),
_addr(addr), _addr(addr),
_ipScope(addr.ipScope()) _ipScope(addr.ipScope())
{ {
@ -188,6 +190,19 @@ public:
*/ */
inline void sent(const int64_t t) { _lastOut = t; } inline void sent(const int64_t t) { _lastOut = t; }
/**
* Update path latency with a new measurement
*
* @param l Measured latency
*/
inline void updateLatency(const unsigned int l)
{
unsigned int pl = _latency;
if (pl < 0xffff)
_latency = (pl + l) / 2;
else _latency = l;
}
/** /**
* @return Local socket as specified by external code * @return Local socket as specified by external code
*/ */
@ -259,9 +274,19 @@ public:
} }
/** /**
* @return True if path appears alive * @return Latency or 0xffff if unknown
*/ */
inline bool alive(const int64_t now) const { return ((now - _lastIn) <= ZT_PATH_ALIVE_TIMEOUT); } inline unsigned int latency() const { return _latency; }
/**
* @return Path quality -- lower is better
*/
inline int quality(const int64_t now) const
{
const int l = (int)_latency;
const int age = (int)std::min((now - _lastIn),(int64_t)(ZT_PATH_HEARTBEAT_PERIOD * 10)); // set an upper sanity limit to avoid overflow
return (((age < (ZT_PATH_HEARTBEAT_PERIOD + 5000)) ? l : (l + 0xffff + age)) * (int)((ZT_INETADDRESS_MAX_SCOPE - _ipScope) + 1));
}
/** /**
* @return True if this path needs a heartbeat * @return True if this path needs a heartbeat
@ -300,6 +325,7 @@ private:
volatile signed int _incomingLinkQualitySlowLogCounter; volatile signed int _incomingLinkQualitySlowLogCounter;
volatile unsigned int _incomingLinkQualityPreviousPacketCounter; volatile unsigned int _incomingLinkQualityPreviousPacketCounter;
volatile unsigned int _outgoingPacketCounter; volatile unsigned int _outgoingPacketCounter;
volatile unsigned int _latency;
InetAddress _addr; InetAddress _addr;
InetAddress::IpScope _ipScope; // memoize this since it's a computed value checked often InetAddress::IpScope _ipScope; // memoize this since it's a computed value checked often
volatile uint8_t _incomingLinkQualitySlowLog[32]; volatile uint8_t _incomingLinkQualitySlowLog[32];

View File

@ -52,12 +52,12 @@ Peer::Peer(const RuntimeEnvironment *renv,const Identity &myIdentity,const Ident
_lastComRequestSent(0), _lastComRequestSent(0),
_lastCredentialsReceived(0), _lastCredentialsReceived(0),
_lastTrustEstablishedPacketReceived(0), _lastTrustEstablishedPacketReceived(0),
_lastSentFullHello(0),
_vProto(0), _vProto(0),
_vMajor(0), _vMajor(0),
_vMinor(0), _vMinor(0),
_vRevision(0), _vRevision(0),
_id(peerIdentity), _id(peerIdentity),
_latency(0),
_directPathPushCutoffCount(0), _directPathPushCutoffCount(0),
_credentialsCutoffCount(0) _credentialsCutoffCount(0)
{ {
@ -148,59 +148,47 @@ void Peer::received(
if (hops == 0) { if (hops == 0) {
// If this is a direct packet (no hops), update existing paths or learn new ones // If this is a direct packet (no hops), update existing paths or learn new ones
bool pathAlreadyKnown = false;
{
Mutex::Lock _l(_paths_m);
if ((path->address().ss_family == AF_INET)&&(_v4Path.p)) {
const struct sockaddr_in *const r = reinterpret_cast<const struct sockaddr_in *>(&(path->address()));
const struct sockaddr_in *const l = reinterpret_cast<const struct sockaddr_in *>(&(_v4Path.p->address()));
if ((r->sin_addr.s_addr == l->sin_addr.s_addr)&&(r->sin_port == l->sin_port)&&(path->localSocket() == _v4Path.p->localSocket())) {
_v4Path.lr = now;
pathAlreadyKnown = true;
}
} else if ((path->address().ss_family == AF_INET6)&&(_v6Path.p)) {
const struct sockaddr_in6 *const r = reinterpret_cast<const struct sockaddr_in6 *>(&(path->address()));
const struct sockaddr_in6 *const l = reinterpret_cast<const struct sockaddr_in6 *>(&(_v6Path.p->address()));
if ((!memcmp(r->sin6_addr.s6_addr,l->sin6_addr.s6_addr,16))&&(r->sin6_port == l->sin6_port)&&(path->localSocket() == _v6Path.p->localSocket())) {
_v6Path.lr = now;
pathAlreadyKnown = true;
}
}
}
if ( (!pathAlreadyKnown) && (RR->node->shouldUsePathForZeroTierTraffic(tPtr,_id.address(),path->localSocket(),path->address())) ) {
Mutex::Lock _l(_paths_m); Mutex::Lock _l(_paths_m);
_PeerPath *replacablePath = (_PeerPath *)0; unsigned int worstQualityPath = 0;
if (path->address().ss_family == AF_INET) { int worstQuality = 0;
if ( ( (!_v4Path.p) || (!_v4Path.p->alive(now)) || (path->preferenceRank() >= _v4Path.p->preferenceRank()) ) && ( (now - _v4Path.sticky) > ZT_PEER_PATH_EXPIRATION ) ) { bool havePath = false;
replacablePath = &_v4Path; for(unsigned int p=0;p<ZT_PEER_MAX_PATHS;++p) {
if (_paths[p].p) {
if (_paths[p].p == path) {
_paths[p].lr = now;
havePath = true;
break;
} }
} else if (path->address().ss_family == AF_INET6) { const int q = _paths[p].p->quality(now) / _paths[p].priority;
if ( ( (!_v6Path.p) || (!_v6Path.p->alive(now)) || (path->preferenceRank() >= _v6Path.p->preferenceRank()) ) && ( (now - _v6Path.sticky) > ZT_PEER_PATH_EXPIRATION ) ) { if (q >= worstQuality) {
replacablePath = &_v6Path; worstQuality = q;
worstQualityPath = p;
} }
}
if (replacablePath) {
if (verb == Packet::VERB_OK) {
RR->t->peerLearnedNewPath(tPtr,networkId,*this,replacablePath->p,path,packetId);
replacablePath->lr = now;
replacablePath->p = path;
} else { } else {
RR->t->peerConfirmingUnknownPath(tPtr,networkId,*this,path,packetId,verb); worstQualityPath = p;
break;
}
}
if ((!havePath)&&(RR->node->shouldUsePathForZeroTierTraffic(tPtr,_id.address(),path->localSocket(),path->address()))) {
if (verb == Packet::VERB_OK) {
RR->t->peerLearnedNewPath(tPtr,networkId,*this,_paths[worstQualityPath].p,path,packetId);
_paths[worstQualityPath].lr = now;
_paths[worstQualityPath].p = path;
_paths[worstQualityPath].priority = 1;
} else {
attemptToContactAt(tPtr,path->localSocket(),path->address(),now,true,path->nextOutgoingCounter()); attemptToContactAt(tPtr,path->localSocket(),path->address(),now,true,path->nextOutgoingCounter());
path->sent(now); path->sent(now);
} RR->t->peerConfirmingUnknownPath(tPtr,networkId,*this,path,packetId,verb);
} }
} }
} }
// If we are being relayed or if we're using a global address, send PUSH_DIRECT_PATHS. // If we have a trust relationship periodically push a message enumerating
// In the global address case we push only configured direct paths to accomplish // all known external addresses for ourselves. We now do this even if we
// fall-forward to local backplane networks over e.g. LAN or Amazon VPC. // have a current path since we'll want to use new ones too.
if ( ((hops > 0)||(path->ipScope() == InetAddress::IP_SCOPE_GLOBAL)) && (this->trustEstablished(now)) ) { if (this->trustEstablished(now)) {
if ((now - _lastDirectPathPushSent) >= ZT_DIRECT_PATH_PUSH_INTERVAL) { if ((now - _lastDirectPathPushSent) >= ZT_DIRECT_PATH_PUSH_INTERVAL) {
_lastDirectPathPushSent = now; _lastDirectPathPushSent = now;
@ -210,6 +198,7 @@ void Peer::received(
for(std::vector<InetAddress>::const_iterator i(dps.begin());i!=dps.end();++i) for(std::vector<InetAddress>::const_iterator i(dps.begin());i!=dps.end();++i)
pathsToPush.push_back(*i); pathsToPush.push_back(*i);
// Do symmetric NAT prediction if we are communicating indirectly.
if (hops > 0) { if (hops > 0) {
std::vector<InetAddress> sym(RR->sa->getSymmetricNatPredictions()); std::vector<InetAddress> sym(RR->sa->getSymmetricNatPredictions());
for(unsigned long i=0,added=0;i<sym.size();++i) { for(unsigned long i=0,added=0;i<sym.size();++i) {
@ -264,52 +253,148 @@ void Peer::received(
} }
} }
bool Peer::sendDirect(void *tPtr,const void *data,unsigned int len,int64_t now,bool force) SharedPtr<Path> Peer::getBestPath(int64_t now,bool includeExpired) const
{ {
Mutex::Lock _l(_paths_m); Mutex::Lock _l(_paths_m);
int64_t v6lr = 0; unsigned int bestPath = ZT_PEER_MAX_PATHS;
if ( ((now - _v6Path.lr) < ZT_PEER_PATH_EXPIRATION) && (_v6Path.p) ) int bestPathQuality = 2147483647; // INT_MAX
v6lr = _v6Path.p->lastIn(); for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
int64_t v4lr = 0; if (_paths[i].p) {
if ( ((now - _v4Path.lr) < ZT_PEER_PATH_EXPIRATION) && (_v4Path.p) ) if ((includeExpired)||((now - _paths[i].lr) < ZT_PEER_PATH_EXPIRATION)) {
v4lr = _v4Path.p->lastIn(); const int q = _paths[i].p->quality(now) / _paths[i].priority;
if (q < bestPathQuality) {
if ( (v6lr > v4lr) && ((now - v6lr) < ZT_PATH_ALIVE_TIMEOUT) ) { bestPathQuality = q;
return _v6Path.p->send(RR,tPtr,data,len,now); bestPath = i;
} else if ((now - v4lr) < ZT_PATH_ALIVE_TIMEOUT) {
return _v4Path.p->send(RR,tPtr,data,len,now);
} else if (force) {
if (v6lr > v4lr) {
return _v6Path.p->send(RR,tPtr,data,len,now);
} else if (v4lr) {
return _v4Path.p->send(RR,tPtr,data,len,now);
} }
} }
} else break;
}
return false; if (bestPath != ZT_PEER_MAX_PATHS)
return _paths[bestPath].p;
return SharedPtr<Path>();
} }
SharedPtr<Path> Peer::getBestPath(int64_t now,bool includeExpired) void Peer::introduce(void *const tPtr,const int64_t now,const SharedPtr<Peer> &other) const
{ {
Mutex::Lock _l(_paths_m); unsigned int myBestV4ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
unsigned int myBestV6ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
int64_t v6lr = 0; int myBestV4QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
if ((includeExpired || ((now - _v6Path.lr) < ZT_PEER_PATH_EXPIRATION)) && (_v6Path.p)) { int myBestV6QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
v6lr = _v6Path.p->lastIn(); unsigned int theirBestV4ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
} unsigned int theirBestV6ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
int64_t v4lr = 0; int theirBestV4QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
if ((includeExpired || ((now - _v4Path.lr) < ZT_PEER_PATH_EXPIRATION)) && (_v4Path.p)) { int theirBestV6QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
v4lr = _v4Path.p->lastIn(); for(int i=0;i<=ZT_INETADDRESS_MAX_SCOPE;++i) {
myBestV4ByScope[i] = ZT_PEER_MAX_PATHS;
myBestV6ByScope[i] = ZT_PEER_MAX_PATHS;
myBestV4QualityByScope[i] = 2147483647;
myBestV6QualityByScope[i] = 2147483647;
theirBestV4ByScope[i] = ZT_PEER_MAX_PATHS;
theirBestV6ByScope[i] = ZT_PEER_MAX_PATHS;
theirBestV4QualityByScope[i] = 2147483647;
theirBestV6QualityByScope[i] = 2147483647;
} }
if (v6lr > v4lr) { Mutex::Lock _l1(_paths_m);
return _v6Path.p;
} else if (v4lr) { for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
return _v4Path.p; if (_paths[i].p) {
const int q = _paths[i].p->quality(now) / _paths[i].priority;
const unsigned int s = (unsigned int)_paths[i].p->ipScope();
switch(_paths[i].p->address().ss_family) {
case AF_INET:
if (q < myBestV4QualityByScope[s]) {
myBestV4QualityByScope[s] = q;
myBestV4ByScope[s] = i;
}
break;
case AF_INET6:
if (q < myBestV6QualityByScope[s]) {
myBestV6QualityByScope[s] = q;
myBestV6ByScope[s] = i;
}
break;
}
} else break;
} }
return SharedPtr<Path>(); Mutex::Lock _l2(other->_paths_m);
for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
if (other->_paths[i].p) {
const int q = other->_paths[i].p->quality(now) / other->_paths[i].priority;
const unsigned int s = (unsigned int)other->_paths[i].p->ipScope();
switch(other->_paths[i].p->address().ss_family) {
case AF_INET:
if (q < theirBestV4QualityByScope[s]) {
theirBestV4QualityByScope[s] = q;
theirBestV4ByScope[s] = i;
}
break;
case AF_INET6:
if (q < theirBestV6QualityByScope[s]) {
theirBestV6QualityByScope[s] = q;
theirBestV6ByScope[s] = i;
}
break;
}
} else break;
}
unsigned int mine = ZT_PEER_MAX_PATHS;
unsigned int theirs = ZT_PEER_MAX_PATHS;
for(int s=ZT_INETADDRESS_MAX_SCOPE;s>=0;--s) {
if ((myBestV6ByScope[s] != ZT_PEER_MAX_PATHS)&&(theirBestV6ByScope[s] != ZT_PEER_MAX_PATHS)) {
mine = myBestV6ByScope[s];
theirs = theirBestV6ByScope[s];
break;
}
if ((myBestV4ByScope[s] != ZT_PEER_MAX_PATHS)&&(theirBestV4ByScope[s] != ZT_PEER_MAX_PATHS)) {
mine = myBestV4ByScope[s];
theirs = theirBestV4ByScope[s];
break;
}
}
if (mine != ZT_PEER_MAX_PATHS) {
unsigned int alt = (unsigned int)RR->node->prng() & 1; // randomize which hint we send first for black magickal NAT-t reasons
const unsigned int completed = alt + 2;
while (alt != completed) {
if ((alt & 1) == 0) {
Packet outp(_id.address(),RR->identity.address(),Packet::VERB_RENDEZVOUS);
outp.append((uint8_t)0);
other->_id.address().appendTo(outp);
outp.append((uint16_t)other->_paths[theirs].p->address().port());
if (other->_paths[theirs].p->address().ss_family == AF_INET6) {
outp.append((uint8_t)16);
outp.append(other->_paths[theirs].p->address().rawIpData(),16);
} else {
outp.append((uint8_t)4);
outp.append(other->_paths[theirs].p->address().rawIpData(),4);
}
outp.armor(_key,true,_paths[mine].p->nextOutgoingCounter());
_paths[mine].p->send(RR,tPtr,outp.data(),outp.size(),now);
} else {
Packet outp(other->_id.address(),RR->identity.address(),Packet::VERB_RENDEZVOUS);
outp.append((uint8_t)0);
_id.address().appendTo(outp);
outp.append((uint16_t)_paths[mine].p->address().port());
if (_paths[mine].p->address().ss_family == AF_INET6) {
outp.append((uint8_t)16);
outp.append(_paths[mine].p->address().rawIpData(),16);
} else {
outp.append((uint8_t)4);
outp.append(_paths[mine].p->address().rawIpData(),4);
}
outp.armor(other->_key,true,other->_paths[theirs].p->nextOutgoingCounter());
other->_paths[theirs].p->send(RR,tPtr,outp.data(),outp.size(),now);
}
++alt;
}
}
} }
void Peer::sendHELLO(void *tPtr,const int64_t localSocket,const InetAddress &atAddress,int64_t now,unsigned int counter) void Peer::sendHELLO(void *tPtr,const int64_t localSocket,const InetAddress &atAddress,int64_t now,unsigned int counter)
@ -377,76 +462,83 @@ void Peer::tryMemorizedPath(void *tPtr,int64_t now)
} }
} }
bool Peer::doPingAndKeepalive(void *tPtr,int64_t now,int inetAddressFamily) unsigned int Peer::doPingAndKeepalive(void *tPtr,int64_t now)
{ {
unsigned int sent = 0;
Mutex::Lock _l(_paths_m); Mutex::Lock _l(_paths_m);
if (inetAddressFamily < 0) { const bool sendFullHello = ((now - _lastSentFullHello) >= ZT_PEER_PING_PERIOD);
int64_t v6lr = 0; _lastSentFullHello = now;
if ( ((now - _v6Path.lr) < ZT_PEER_PATH_EXPIRATION) && (_v6Path.p) )
v6lr = _v6Path.p->lastIn();
int64_t v4lr = 0;
if ( ((now - _v4Path.lr) < ZT_PEER_PATH_EXPIRATION) && (_v4Path.p) )
v4lr = _v4Path.p->lastIn();
if (v6lr > v4lr) { unsigned int j = 0;
if ( ((now - _v6Path.lr) >= ZT_PEER_PING_PERIOD) || (_v6Path.p->needsHeartbeat(now)) ) { for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
attemptToContactAt(tPtr,_v6Path.p->localSocket(),_v6Path.p->address(),now,false,_v6Path.p->nextOutgoingCounter()); if (!_paths[i].p) break;
_v6Path.p->sent(now); if ((now - _paths[i].lr) < ZT_PEER_PATH_EXPIRATION) {
return true; if ((sendFullHello)||(_paths[i].p->needsHeartbeat(now))) {
attemptToContactAt(tPtr,_paths[i].p->localSocket(),_paths[i].p->address(),now,sendFullHello,_paths[i].p->nextOutgoingCounter());
_paths[i].p->sent(now);
sent |= (_paths[i].p->address().ss_family == AF_INET) ? 0x1 : 0x2;
} }
} else if (v4lr) { if (i != j)
if ( ((now - _v4Path.lr) >= ZT_PEER_PING_PERIOD) || (_v4Path.p->needsHeartbeat(now)) ) { _paths[j] = _paths[i];
attemptToContactAt(tPtr,_v4Path.p->localSocket(),_v4Path.p->address(),now,false,_v4Path.p->nextOutgoingCounter()); ++j;
_v4Path.p->sent(now);
return true;
}
}
} else {
if ( (inetAddressFamily == AF_INET) && ((now - _v4Path.lr) < ZT_PEER_PATH_EXPIRATION) ) {
if ( ((now - _v4Path.lr) >= ZT_PEER_PING_PERIOD) || (_v4Path.p->needsHeartbeat(now)) ) {
attemptToContactAt(tPtr,_v4Path.p->localSocket(),_v4Path.p->address(),now,false,_v4Path.p->nextOutgoingCounter());
_v4Path.p->sent(now);
return true;
}
} else if ( (inetAddressFamily == AF_INET6) && ((now - _v6Path.lr) < ZT_PEER_PATH_EXPIRATION) ) {
if ( ((now - _v6Path.lr) >= ZT_PEER_PING_PERIOD) || (_v6Path.p->needsHeartbeat(now)) ) {
attemptToContactAt(tPtr,_v6Path.p->localSocket(),_v6Path.p->address(),now,false,_v6Path.p->nextOutgoingCounter());
_v6Path.p->sent(now);
return true;
} }
} }
while(j < ZT_PEER_MAX_PATHS) {
_paths[j].lr = 0;
_paths[j].p.zero();
_paths[j].priority = 1;
++j;
} }
return false; return sent;
} }
void Peer::redirect(void *tPtr,const int64_t localSocket,const InetAddress &remoteAddress,const int64_t now) void Peer::clusterRedirect(void *tPtr,const int64_t localSocket,const InetAddress &remoteAddress,const int64_t now)
{ {
if ((remoteAddress.ss_family != AF_INET)&&(remoteAddress.ss_family != AF_INET6)) // sanity check
return;
SharedPtr<Path> op;
SharedPtr<Path> np(RR->topology->getPath(localSocket,remoteAddress)); SharedPtr<Path> np(RR->topology->getPath(localSocket,remoteAddress));
np->received(now); RR->t->peerRedirected(tPtr,0,*this,np);
attemptToContactAt(tPtr,localSocket,remoteAddress,now,true,np->nextOutgoingCounter()); attemptToContactAt(tPtr,localSocket,remoteAddress,now,true,np->nextOutgoingCounter());
{ {
Mutex::Lock _l(_paths_m); Mutex::Lock _l(_paths_m);
if (remoteAddress.ss_family == AF_INET) { int worstQuality = 0;
op = _v4Path.p; unsigned int worstQualityPath = 0;
_v4Path.lr = now; for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
_v4Path.sticky = now; if (_paths[i].p) {
_v4Path.p = np; if (_paths[i].p == np) { // <-- where's my Fields Medal?
} else if (remoteAddress.ss_family == AF_INET6) { _paths[i].lr = now; // consider this a "receive"
op = _v6Path.p; _paths[i].priority += 5; // kind of arbitrary, bumps way up in best path quality order
_v6Path.lr = now; return;
_v6Path.sticky = now; }
_v6Path.p = np; const int q = _paths[i].p->quality(now) / _paths[i].priority;
if (q >= worstQuality) {
worstQuality = q;
worstQualityPath = i;
}
} else {
worstQualityPath = i;
break;
} }
} }
_paths[worstQualityPath].lr = now;
_paths[worstQualityPath].p = np;
_paths[worstQualityPath].priority = 6; // 1 + 5
}
}
RR->t->peerRedirected(tPtr,0,*this,op,np); void Peer::resetWithinScope(void *tPtr,InetAddress::IpScope scope,int inetAddressFamily,int64_t now)
{
Mutex::Lock _l(_paths_m);
for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
if (_paths[i].p) {
if ((_paths[i].p->address().ss_family == inetAddressFamily)&&(_paths[i].p->ipScope() == scope)) {
attemptToContactAt(tPtr,_paths[i].p->localSocket(),_paths[i].p->address(),now,false,_paths[i].p->nextOutgoingCounter());
_paths[i].p->sent(now);
_paths[i].lr = 0; // path will not be used unless it speaks again
}
} else break;
}
} }
} // namespace ZeroTier } // namespace ZeroTier

View File

@ -53,6 +53,15 @@
#define ZT_PEER_MAX_SERIALIZED_STATE_SIZE (sizeof(Peer) + 32 + (sizeof(Path) * 2)) #define ZT_PEER_MAX_SERIALIZED_STATE_SIZE (sizeof(Peer) + 32 + (sizeof(Path) * 2))
/**
* Maximum number of direct paths to a peer
*
* This can be increased. You'll want about 2X the number of physical links
* you are ever likely to want to bundle/trunk since there is likely to be
* a path for every protocol (IPv4, IPv6, etc.).
*/
#define ZT_PEER_MAX_PATHS 16
namespace ZeroTier { namespace ZeroTier {
/** /**
@ -116,6 +125,8 @@ public:
const uint64_t networkId); const uint64_t networkId);
/** /**
* Check whether we have an active path to this peer via the given address
*
* @param now Current time * @param now Current time
* @param addr Remote address * @param addr Remote address
* @return True if we have an active path to this destination * @return True if we have an active path to this destination
@ -123,7 +134,13 @@ public:
inline bool hasActivePathTo(int64_t now,const InetAddress &addr) const inline bool hasActivePathTo(int64_t now,const InetAddress &addr) const
{ {
Mutex::Lock _l(_paths_m); Mutex::Lock _l(_paths_m);
return ( ((addr.ss_family == AF_INET)&&(_v4Path.p)&&(_v4Path.p->address() == addr)&&(_v4Path.p->alive(now))) || ((addr.ss_family == AF_INET6)&&(_v6Path.p)&&(_v6Path.p->address() == addr)&&(_v6Path.p->alive(now))) ); for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
if (_paths[i].p) {
if (((now - _paths[i].lr) < ZT_PEER_PATH_EXPIRATION)&&(_paths[i].p->address() == addr))
return true;
} else break;
}
return false;
} }
/** /**
@ -136,19 +153,27 @@ public:
* @param force If true, send even if path is not alive * @param force If true, send even if path is not alive
* @return True if we actually sent something * @return True if we actually sent something
*/ */
bool sendDirect(void *tPtr,const void *data,unsigned int len,int64_t now,bool force); inline bool sendDirect(void *tPtr,const void *data,unsigned int len,int64_t now,bool force)
{
SharedPtr<Path> bp(getBestPath(now,force));
if (bp)
return bp->send(RR,tPtr,data,len,now);
return false;
}
/** /**
* Get the best current direct path * Get the best current direct path
* *
* This does not check Path::alive(), but does return the most recently
* active path and does check expiration (which is a longer timeout).
*
* @param now Current time * @param now Current time
* @param includeExpired If true, include even expired paths * @param includeExpired If true, include even expired paths
* @return Best current path or NULL if none * @return Best current path or NULL if none
*/ */
SharedPtr<Path> getBestPath(int64_t now,bool includeExpired); SharedPtr<Path> getBestPath(int64_t now,bool includeExpired) const;
/**
* Send VERB_RENDEZVOUS to this and another peer via the best common IP scope and path
*/
void introduce(void *const tPtr,const int64_t now,const SharedPtr<Peer> &other) const;
/** /**
* Send a HELLO to this peer at a specified physical address * Send a HELLO to this peer at a specified physical address
@ -190,67 +215,39 @@ public:
/** /**
* Send pings or keepalives depending on configured timeouts * Send pings or keepalives depending on configured timeouts
* *
* This also cleans up some internal data structures. It's called periodically from Node.
*
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call * @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
* @param now Current time * @param now Current time
* @param inetAddressFamily Keep this address family alive, or -1 for any * @param inetAddressFamily Keep this address family alive, or -1 for any
* @return True if we have at least one direct path of the given family (or any if family is -1) * @return 0 if nothing sent or bit mask: bit 0x1 if IPv4 sent, bit 0x2 if IPv6 sent (0x3 means both sent)
*/ */
bool doPingAndKeepalive(void *tPtr,int64_t now,int inetAddressFamily); unsigned int doPingAndKeepalive(void *tPtr,int64_t now);
/** /**
* Specify remote path for this peer and forget others * Process a cluster redirect sent by this peer
*
* This overrides normal path learning and tells this peer to be found
* at this address, at least within the address's family. Other address
* families are not modified.
* *
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call * @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
* @param localSocket Local socket as supplied by external code * @param localSocket Local socket as supplied by external code
* @param remoteAddress Remote address * @param remoteAddress Remote address
* @param now Current time * @param now Current time
*/ */
void redirect(void *tPtr,const int64_t localSocket,const InetAddress &remoteAddress,const int64_t now); void clusterRedirect(void *tPtr,const int64_t localSocket,const InetAddress &remoteAddress,const int64_t now);
/** /**
* Reset paths within a given IP scope and address family * Reset paths within a given IP scope and address family
* *
* Resetting a path involves sending an ECHO to it and then deactivating * Resetting a path involves sending an ECHO to it and then deactivating
* it until or unless it responds. * it until or unless it responds. This is done when we detect a change
* to our external IP or another system change that might invalidate
* many or all current paths.
* *
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call * @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
* @param scope IP scope * @param scope IP scope
* @param inetAddressFamily Family e.g. AF_INET * @param inetAddressFamily Family e.g. AF_INET
* @param now Current time * @param now Current time
*/ */
inline void resetWithinScope(void *tPtr,InetAddress::IpScope scope,int inetAddressFamily,int64_t now) void resetWithinScope(void *tPtr,InetAddress::IpScope scope,int inetAddressFamily,int64_t now);
{
Mutex::Lock _l(_paths_m);
if ((inetAddressFamily == AF_INET)&&(_v4Path.lr)&&(_v4Path.p->address().ipScope() == scope)) {
attemptToContactAt(tPtr,_v4Path.p->localSocket(),_v4Path.p->address(),now,false,_v4Path.p->nextOutgoingCounter());
_v4Path.p->sent(now);
_v4Path.lr = 0; // path will not be used unless it speaks again
} else if ((inetAddressFamily == AF_INET6)&&(_v6Path.lr)&&(_v6Path.p->address().ipScope() == scope)) {
attemptToContactAt(tPtr,_v6Path.p->localSocket(),_v6Path.p->address(),now,false,_v6Path.p->nextOutgoingCounter());
_v6Path.p->sent(now);
_v6Path.lr = 0; // path will not be used unless it speaks again
}
}
/**
* Fill parameters with V4 and V6 addresses if known and alive
*
* @param now Current time
* @param v4 Result parameter to receive active IPv4 address, if any
* @param v6 Result parameter to receive active IPv6 address, if any
*/
inline void getRendezvousAddresses(int64_t now,InetAddress &v4,InetAddress &v6) const
{
Mutex::Lock _l(_paths_m);
if (((now - _v4Path.lr) < ZT_PEER_PATH_EXPIRATION)&&(_v4Path.p->alive(now)))
v4 = _v4Path.p->address();
if (((now - _v6Path.lr) < ZT_PEER_PATH_EXPIRATION)&&(_v6Path.p->alive(now)))
v6 = _v6Path.p->address();
}
/** /**
* @param now Current time * @param now Current time
@ -260,10 +257,10 @@ public:
{ {
std::vector< SharedPtr<Path> > pp; std::vector< SharedPtr<Path> > pp;
Mutex::Lock _l(_paths_m); Mutex::Lock _l(_paths_m);
if (((now - _v4Path.lr) < ZT_PEER_PATH_EXPIRATION)&&(_v4Path.p->alive(now))) for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
pp.push_back(_v4Path.p); if (!_paths[i].p) break;
if (((now - _v6Path.lr) < ZT_PEER_PATH_EXPIRATION)&&(_v6Path.p->alive(now))) pp.push_back(_paths[i].p);
pp.push_back(_v6Path.p); }
return pp; return pp;
} }
@ -283,9 +280,13 @@ public:
inline int64_t isActive(int64_t now) const { return ((now - _lastNontrivialReceive) < ZT_PEER_ACTIVITY_TIMEOUT); } inline int64_t isActive(int64_t now) const { return ((now - _lastNontrivialReceive) < ZT_PEER_ACTIVITY_TIMEOUT); }
/** /**
* @return Latency in milliseconds or 0 if unknown * @return Latency in milliseconds of best path or 0xffff if unknown / no paths
*/ */
inline unsigned int latency() const { return _latency; } inline unsigned int latency(const int64_t now) const
{
SharedPtr<Path> bp(getBestPath(now,false));
return ((bp) ? bp->latency() : 0xffff);
}
/** /**
* This computes a quality score for relays and root servers * This computes a quality score for relays and root servers
@ -303,25 +304,12 @@ public:
const uint64_t tsr = now - _lastReceive; const uint64_t tsr = now - _lastReceive;
if (tsr >= ZT_PEER_ACTIVITY_TIMEOUT) if (tsr >= ZT_PEER_ACTIVITY_TIMEOUT)
return (~(unsigned int)0); return (~(unsigned int)0);
unsigned int l = _latency; unsigned int l = latency(now);
if (!l) if (!l)
l = 0xffff; l = 0xffff;
return (l * (((unsigned int)tsr / (ZT_PEER_PING_PERIOD + 1000)) + 1)); return (l * (((unsigned int)tsr / (ZT_PEER_PING_PERIOD + 1000)) + 1));
} }
/**
* Update latency with a new direct measurment
*
* @param l Direct latency measurment in ms
*/
inline void addDirectLatencyMeasurment(unsigned int l)
{
unsigned int ol = _latency;
if ((ol > 0)&&(ol < 10000))
_latency = (ol + std::min(l,(unsigned int)65535)) / 2;
else _latency = std::min(l,(unsigned int)65535);
}
/** /**
* @return 256-bit secret symmetric encryption key * @return 256-bit secret symmetric encryption key
*/ */
@ -442,29 +430,15 @@ public:
/** /**
* Serialize a peer for storage in local cache * Serialize a peer for storage in local cache
* *
* This does not serialize everything, just identity and addresses where the peer * This does not serialize everything, just non-ephemeral information.
* may be reached.
*/ */
template<unsigned int C> template<unsigned int C>
inline void serialize(Buffer<C> &b) const inline void serializeForCache(Buffer<C> &b) const
{ {
b.append((uint8_t)0); b.append((uint8_t)1);
_id.serialize(b); _id.serialize(b);
b.append(_lastReceive);
b.append(_lastNontrivialReceive);
b.append(_lastTriedMemorizedPath);
b.append(_lastDirectPathPushSent);
b.append(_lastDirectPathPushReceive);
b.append(_lastCredentialRequestSent);
b.append(_lastWhoisRequestReceived);
b.append(_lastEchoRequestReceived);
b.append(_lastComRequestReceived);
b.append(_lastComRequestSent);
b.append(_lastCredentialsReceived);
b.append(_lastTrustEstablishedPacketReceived);
b.append((uint16_t)_vProto); b.append((uint16_t)_vProto);
b.append((uint16_t)_vMajor); b.append((uint16_t)_vMajor);
b.append((uint16_t)_vMinor); b.append((uint16_t)_vMinor);
@ -472,15 +446,16 @@ public:
{ {
Mutex::Lock _l(_paths_m); Mutex::Lock _l(_paths_m);
unsigned int pcount = 0; unsigned int pc = 0;
if (_v4Path.p) ++pcount; for(unsigned int i=0;i<ZT_PEER_MAX_PATHS;++i) {
if (_v6Path.p) ++pcount; if (_paths[i].p)
b.append((uint8_t)pcount); ++pc;
if (_v4Path.p) _v4Path.p->address().serialize(b); else break;
if (_v6Path.p) _v6Path.p->address().serialize(b); }
b.append((uint16_t)pc);
for(unsigned int i=0;i<pc;++i)
_paths[i].p->address().serialize(b);
} }
b.append((uint16_t)0);
} }
template<unsigned int C> template<unsigned int C>
@ -488,7 +463,7 @@ public:
{ {
try { try {
unsigned int ptr = 0; unsigned int ptr = 0;
if (b[ptr++] != 0) if (b[ptr++] != 1)
return SharedPtr<Peer>(); return SharedPtr<Peer>();
Identity id; Identity id;
@ -498,15 +473,16 @@ public:
SharedPtr<Peer> p(new Peer(renv,renv->identity,id)); SharedPtr<Peer> p(new Peer(renv,renv->identity,id));
ptr += 12 * 8; // skip deserializing ephemeral state in this case
p->_vProto = b.template at<uint16_t>(ptr); ptr += 2; p->_vProto = b.template at<uint16_t>(ptr); ptr += 2;
p->_vMajor = b.template at<uint16_t>(ptr); ptr += 2; p->_vMajor = b.template at<uint16_t>(ptr); ptr += 2;
p->_vMinor = b.template at<uint16_t>(ptr); ptr += 2; p->_vMinor = b.template at<uint16_t>(ptr); ptr += 2;
p->_vRevision = b.template at<uint16_t>(ptr); ptr += 2; p->_vRevision = b.template at<uint16_t>(ptr); ptr += 2;
const unsigned int pcount = (unsigned int)b[ptr++]; // When we deserialize from the cache we don't actually restore paths. We
for(unsigned int i=0;i<pcount;++i) { // just try them and then re-learn them if they happen to still be up.
// Paths are fairly ephemeral in the real world in most cases.
const unsigned int tryPathCount = b.template at<uint16_t>(ptr); ptr += 2;
for(unsigned int i=0;i<tryPathCount;++i) {
InetAddress inaddr; InetAddress inaddr;
try { try {
ptr += inaddr.deserialize(b,ptr); ptr += inaddr.deserialize(b,ptr);
@ -526,10 +502,10 @@ public:
private: private:
struct _PeerPath struct _PeerPath
{ {
_PeerPath() : lr(0),sticky(0),p() {} _PeerPath() : lr(0),p(),priority(1) {}
int64_t lr; // time of last valid ZeroTier packet int64_t lr; // time of last valid ZeroTier packet
int64_t sticky; // time last set as sticky
SharedPtr<Path> p; SharedPtr<Path> p;
int priority; // >= 1, higher is better
}; };
uint8_t _key[ZT_PEER_SECRET_KEY_LENGTH]; uint8_t _key[ZT_PEER_SECRET_KEY_LENGTH];
@ -548,19 +524,18 @@ private:
int64_t _lastComRequestSent; int64_t _lastComRequestSent;
int64_t _lastCredentialsReceived; int64_t _lastCredentialsReceived;
int64_t _lastTrustEstablishedPacketReceived; int64_t _lastTrustEstablishedPacketReceived;
int64_t _lastSentFullHello;
uint16_t _vProto; uint16_t _vProto;
uint16_t _vMajor; uint16_t _vMajor;
uint16_t _vMinor; uint16_t _vMinor;
uint16_t _vRevision; uint16_t _vRevision;
_PeerPath _v4Path; // IPv4 direct path _PeerPath _paths[ZT_PEER_MAX_PATHS];
_PeerPath _v6Path; // IPv6 direct path
Mutex _paths_m; Mutex _paths_m;
Identity _id; Identity _id;
unsigned int _latency;
unsigned int _directPathPushCutoffCount; unsigned int _directPathPushCutoffCount;
unsigned int _credentialsCutoffCount; unsigned int _credentialsCutoffCount;

View File

@ -169,68 +169,22 @@ void Switch::onRemotePacket(void *tPtr,const int64_t localSocket,const InetAddre
if (packet.hops() < ZT_RELAY_MAX_HOPS) { if (packet.hops() < ZT_RELAY_MAX_HOPS) {
packet.incrementHops(); packet.incrementHops();
SharedPtr<Peer> relayTo = RR->topology->getPeer(tPtr,destination); SharedPtr<Peer> relayTo = RR->topology->getPeer(tPtr,destination);
if ((relayTo)&&(relayTo->sendDirect(tPtr,packet.data(),packet.size(),now,false))) { if ((relayTo)&&(relayTo->sendDirect(tPtr,packet.data(),packet.size(),now,false))) {
if ((source != RR->identity.address())&&(_shouldUnite(now,source,destination))) { // don't send RENDEZVOUS for cluster frontplane relays if ((source != RR->identity.address())&&(_shouldUnite(now,source,destination))) {
const InetAddress *hintToSource = (InetAddress *)0;
const InetAddress *hintToDest = (InetAddress *)0;
InetAddress destV4,destV6;
InetAddress sourceV4,sourceV6;
relayTo->getRendezvousAddresses(now,destV4,destV6);
const SharedPtr<Peer> sourcePeer(RR->topology->getPeer(tPtr,source)); const SharedPtr<Peer> sourcePeer(RR->topology->getPeer(tPtr,source));
if (sourcePeer) { if (sourcePeer)
sourcePeer->getRendezvousAddresses(now,sourceV4,sourceV6); relayTo->introduce(tPtr,now,sourcePeer);
if ((destV6)&&(sourceV6)) {
hintToSource = &destV6;
hintToDest = &sourceV6;
} else if ((destV4)&&(sourceV4)) {
hintToSource = &destV4;
hintToDest = &sourceV4;
}
if ((hintToSource)&&(hintToDest)) {
unsigned int alt = (unsigned int)RR->node->prng() & 1; // randomize which hint we send first for obscure NAT-t reasons
const unsigned int completed = alt + 2;
while (alt != completed) {
if ((alt & 1) == 0) {
Packet outp(source,RR->identity.address(),Packet::VERB_RENDEZVOUS);
outp.append((uint8_t)0);
destination.appendTo(outp);
outp.append((uint16_t)hintToSource->port());
if (hintToSource->ss_family == AF_INET6) {
outp.append((uint8_t)16);
outp.append(hintToSource->rawIpData(),16);
} else {
outp.append((uint8_t)4);
outp.append(hintToSource->rawIpData(),4);
}
send(tPtr,outp,true);
} else {
Packet outp(destination,RR->identity.address(),Packet::VERB_RENDEZVOUS);
outp.append((uint8_t)0);
source.appendTo(outp);
outp.append((uint16_t)hintToDest->port());
if (hintToDest->ss_family == AF_INET6) {
outp.append((uint8_t)16);
outp.append(hintToDest->rawIpData(),16);
} else {
outp.append((uint8_t)4);
outp.append(hintToDest->rawIpData(),4);
}
send(tPtr,outp,true);
}
++alt;
}
}
}
} }
} else { } else {
relayTo = RR->topology->getUpstreamPeer(); relayTo = RR->topology->getUpstreamPeer();
if ((relayTo)&&(relayTo->address() != source)) if ((relayTo)&&(relayTo->address() != source)) {
relayTo->sendDirect(tPtr,packet.data(),packet.size(),now,true); if (relayTo->sendDirect(tPtr,packet.data(),packet.size(),now,true)) {
const SharedPtr<Peer> sourcePeer(RR->topology->getPeer(tPtr,source));
if (sourcePeer)
relayTo->introduce(tPtr,now,sourcePeer);
}
}
} }
} }
} else if ((reinterpret_cast<const uint8_t *>(data)[ZT_PACKET_IDX_FLAGS] & ZT_PROTO_FLAG_FRAGMENTED) != 0) { } else if ((reinterpret_cast<const uint8_t *>(data)[ZT_PACKET_IDX_FLAGS] & ZT_PROTO_FLAG_FRAGMENTED) != 0) {
@ -694,22 +648,7 @@ bool Switch::_trySend(void *tPtr,Packet &packet,bool encrypt)
const SharedPtr<Peer> peer(RR->topology->getPeer(tPtr,destination)); const SharedPtr<Peer> peer(RR->topology->getPeer(tPtr,destination));
if (peer) { if (peer) {
/* First get the best path, and if it's dead (and this is not a root)
* we attempt to re-activate that path but this packet will flow
* upstream. If the path comes back alive, it will be used in the future.
* For roots we don't do the alive check since roots are not required
* to send heartbeats "down" and because we have to at least try to
* go somewhere. */
viaPath = peer->getBestPath(now,false); viaPath = peer->getBestPath(now,false);
if ( (viaPath) && (!viaPath->alive(now)) && (!RR->topology->isUpstream(peer->identity())) ) {
if ((now - viaPath->lastOut()) > std::max((now - viaPath->lastIn()) * 4,(int64_t)ZT_PATH_MIN_REACTIVATE_INTERVAL)) {
peer->attemptToContactAt(tPtr,viaPath->localSocket(),viaPath->address(),now,false,viaPath->nextOutgoingCounter());
viaPath->sent(now);
}
viaPath.zero();
}
if (!viaPath) { if (!viaPath) {
peer->tryMemorizedPath(tPtr,now); // periodically attempt memorized or statically defined paths, if any are known peer->tryMemorizedPath(tPtr,now); // periodically attempt memorized or statically defined paths, if any are known
const SharedPtr<Peer> relay(RR->topology->getUpstreamPeer()); const SharedPtr<Peer> relay(RR->topology->getUpstreamPeer());

View File

@ -431,7 +431,7 @@ void Topology::_savePeer(void *tPtr,const SharedPtr<Peer> &peer)
{ {
try { try {
Buffer<ZT_PEER_MAX_SERIALIZED_STATE_SIZE> buf; Buffer<ZT_PEER_MAX_SERIALIZED_STATE_SIZE> buf;
peer->serialize(buf); peer->serializeForCache(buf);
uint64_t tmpid[2]; tmpid[0] = peer->address().toInt(); tmpid[1] = 0; uint64_t tmpid[2]; tmpid[0] = peer->address().toInt(); tmpid[1] = 0;
RR->node->stateObjectPut(tPtr,ZT_STATE_OBJECT_PEER,tmpid,buf.data(),buf.size()); RR->node->stateObjectPut(tPtr,ZT_STATE_OBJECT_PEER,tmpid,buf.data(),buf.size());
} catch ( ... ) {} // sanity check, discard invalid entries } catch ( ... ) {} // sanity check, discard invalid entries

View File

@ -300,7 +300,7 @@ public:
SharedPtr<Peer> *p = (SharedPtr<Peer> *)0; SharedPtr<Peer> *p = (SharedPtr<Peer> *)0;
while (i.next(a,p)) { while (i.next(a,p)) {
const SharedPtr<Path> pp((*p)->getBestPath(now,false)); const SharedPtr<Path> pp((*p)->getBestPath(now,false));
if ((pp)&&(pp->alive(now))) if (pp)
++cnt; ++cnt;
} }
return cnt; return cnt;

View File

@ -92,16 +92,13 @@ void Trace::peerLearnedNewPath(void *const tPtr,const uint64_t networkId,Peer &p
_send(tPtr,d,networkId); _send(tPtr,d,networkId);
} }
void Trace::peerRedirected(void *const tPtr,const uint64_t networkId,Peer &peer,const SharedPtr<Path> &oldPath,const SharedPtr<Path> &newPath) void Trace::peerRedirected(void *const tPtr,const uint64_t networkId,Peer &peer,const SharedPtr<Path> &newPath)
{ {
char tmp[128]; char tmp[128];
Dictionary<ZT_MAX_REMOTE_TRACE_SIZE> d; Dictionary<ZT_MAX_REMOTE_TRACE_SIZE> d;
d.add(ZT_REMOTE_TRACE_FIELD__EVENT,ZT_REMOTE_TRACE_EVENT__PEER_REDIRECTED_S); d.add(ZT_REMOTE_TRACE_FIELD__EVENT,ZT_REMOTE_TRACE_EVENT__PEER_REDIRECTED_S);
d.add(ZT_REMOTE_TRACE_FIELD__NETWORK_ID,networkId); d.add(ZT_REMOTE_TRACE_FIELD__NETWORK_ID,networkId);
d.add(ZT_REMOTE_TRACE_FIELD__REMOTE_ZTADDR,peer.address()); d.add(ZT_REMOTE_TRACE_FIELD__REMOTE_ZTADDR,peer.address());
if (oldPath) {
d.add(ZT_REMOTE_TRACE_FIELD__OLD_REMOTE_PHYADDR,oldPath->address().toString(tmp));
}
if (newPath) { if (newPath) {
d.add(ZT_REMOTE_TRACE_FIELD__REMOTE_PHYADDR,newPath->address().toString(tmp)); d.add(ZT_REMOTE_TRACE_FIELD__REMOTE_PHYADDR,newPath->address().toString(tmp));
d.add(ZT_REMOTE_TRACE_FIELD__LOCAL_SOCKET,newPath->localSocket()); d.add(ZT_REMOTE_TRACE_FIELD__LOCAL_SOCKET,newPath->localSocket());

View File

@ -105,7 +105,7 @@ public:
void peerConfirmingUnknownPath(void *const tPtr,const uint64_t networkId,Peer &peer,const SharedPtr<Path> &path,const uint64_t packetId,const Packet::Verb verb); void peerConfirmingUnknownPath(void *const tPtr,const uint64_t networkId,Peer &peer,const SharedPtr<Path> &path,const uint64_t packetId,const Packet::Verb verb);
void peerLearnedNewPath(void *const tPtr,const uint64_t networkId,Peer &peer,const SharedPtr<Path> &oldPath,const SharedPtr<Path> &newPath,const uint64_t packetId); void peerLearnedNewPath(void *const tPtr,const uint64_t networkId,Peer &peer,const SharedPtr<Path> &oldPath,const SharedPtr<Path> &newPath,const uint64_t packetId);
void peerRedirected(void *const tPtr,const uint64_t networkId,Peer &peer,const SharedPtr<Path> &oldPath,const SharedPtr<Path> &newPath); void peerRedirected(void *const tPtr,const uint64_t networkId,Peer &peer,const SharedPtr<Path> &newPath);
void incomingPacketMessageAuthenticationFailure(void *const tPtr,const SharedPtr<Path> &path,const uint64_t packetId,const Address &source,const unsigned int hops,const char *reason); void incomingPacketMessageAuthenticationFailure(void *const tPtr,const SharedPtr<Path> &path,const uint64_t packetId,const Address &source,const unsigned int hops,const char *reason);
void incomingPacketInvalid(void *const tPtr,const SharedPtr<Path> &path,const uint64_t packetId,const Address &source,const unsigned int hops,const Packet::Verb verb,const char *reason); void incomingPacketInvalid(void *const tPtr,const SharedPtr<Path> &path,const uint64_t packetId,const Address &source,const unsigned int hops,const Packet::Verb verb,const char *reason);