ExternalVendorCode/ZeroTierOne
1
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2025-04-11 13:09:55 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Work in progress: refactoring paths, adding TCP fallback.

Browse Source
This commit is contained in:
Adam Ierymenko 2014-03-20 20:07:35 -07:00
parent 45e823d27c
commit ba3f04deed
8 changed files with 268 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
node/Constants.hpp
View File

@ -307,7 +307,7 @@ error_no_byte_order_defined;
/**
* Try TCP tunnels if no response to UDP PINGs in this many milliseconds
*/
#define ZT_TCP_FALLBACK_AFTER 5000
#define ZT_PING_UNANSWERED_AFTER 5000
/**
* Stop relaying via peers that have not responded to direct sends in this long

1
node/Node.cpp
View File

@ -590,6 +590,7 @@ Node::ReasonForTermination Node::run()
* which will trigger a new RENDEZVOUS and a new hole punch. This
* functor excludes supernodes, which are pinged separately above. */
_r->topology->eachPeer(Topology::ResetActivePeers(_r,now));
_r->sm->closeTcpSockets();
} else {
// Periodically check for changes in our local multicast subscriptions
// and broadcast those changes to directly connected peers.

90
node/Path.hpp
View File

@ -30,11 +30,15 @@
#include <stdint.h>
#include <stdexcept>
#include <string>
#include "Constants.hpp"
#include "InetAddress.hpp"
#include "Utils.hpp"
#include "Buffer.hpp"
#include <string>
#define ZT_PATH_SERIALIZATION_VERSION 1
namespace ZeroTier {
@ -45,7 +49,7 @@ class Path
{
public:
Path() :
_lastSent(0),
_lastSend(0),
_lastReceived(0),
_lastFirewallOpener(0),
_lastPing(0),
@ -53,8 +57,8 @@ public:
_tcp(false),
_fixed(false) {}
Path(const InetAddress &addr,bool tcp,bool fixed) :
_lastSent(0),
Path(const InetAddress &addr,bool tcp,bool fixed = false) :
_lastSend(0),
_lastReceived(0),
_lastFirewallOpener(0),
_lastPing(0),
@ -64,13 +68,15 @@ public:
inline const InetAddress &address() const throw() { return _addr; }
inline bool tcp() const throw() { return _tcp; }
inline uint64_t lastSent() const throw() { return _lastSent; }
inline uint64_t lastSend() const throw() { return _lastSend; }
inline uint64_t lastReceived() const throw() { return _lastReceived; }
inline uint64_t lastFirewallOpener() const throw() { return _lastFirewallOpener; }
inline uint64_t lastPing() const throw() { return _lastPing; }
inline bool fixed() const throw() { return _fixed; }
inline void sent(uint64_t t) throw() { _lastSent = t; }
inline void setFixed(bool f) throw() { _fixed = f; }
inline void sent(uint64_t t) throw() { _lastSend = t; }
inline void received(uint64_t t) throw() { _lastReceived = t; }
inline void firewallOpenerSent(uint64_t t) throw() { _lastFirewallOpener = t; }
inline void pinged(uint64_t t) throw() { _lastPing = t; }
@ -81,6 +87,19 @@ public:
return ((_addr)&&((_fixed)||((now - _lastReceived) < ZT_PEER_PATH_ACTIVITY_TIMEOUT)));
}
/**
* @return True if it appears that a ping has gone unanswered
*/
inline bool pingUnanswered(uint64_t now) const
throw()
{
uint64_t lp = _lastPing;
uint64_t lr = _lastReceived;
if (lp)
return ((lr < lp)&&((lp - lr) > ZT_PING_UNANSWERED_AFTER));
return false;
}
/**
* @return Human-readable address and other information about this path, some computed as of current time
*/
@ -88,7 +107,7 @@ public:
{
uint64_t now = Utils::now();
char lsago[32],lrago[32],lfoago[32],lpago[32];
Utils::snprintf(lsago,sizeof(lsago),"%lld",(long long)((_lastSent != 0) ? (now - _lastSent) : -1));
Utils::snprintf(lsago,sizeof(lsago),"%lld",(long long)((_lastSend != 0) ? (now - _lastSend) : -1));
Utils::snprintf(lrago,sizeof(lrago),"%lld",(long long)((_lastReceived != 0) ? (now - _lastReceived) : -1));
Utils::snprintf(lfoago,sizeof(lfoago),"%lld",(long long)((_lastFirewallOpener != 0) ? (now - _lastFirewallOpener) : -1));
Utils::snprintf(lpago,sizeof(lfoago),"%lld",(long long)((_lastPing != 0) ? (now - _lastPing) : -1));
@ -110,8 +129,63 @@ public:
inline bool operator<=(const Path &p) const throw() { return !(p < *this); }
inline bool operator>=(const Path &p) const throw() { return !(*this < p); }
template<unsigned int C>
inline void serialize(Buffer<C> &b) const
{
b.append((unsigned char)ZT_PATH_SERIALIZATION_VERSION);
b.append(_lastSend);
b.append(_lastReceived);
b.append(_lastFirewallOpener);
b.append(_lastPing);
b.append((unsigned char)_addr.type());
switch(_addr.type()) {
case InetAddress::TYPE_NULL:
break;
case InetAddress::TYPE_IPV4:
b.append(_addr.rawIpData(),4);
b.append((uint16_t)_addr.port());
break;
case InetAddress::TYPE_IPV6:
b.append(_addr.rawIpData(),16);
b.append((uint16_t)_addr.port());
break;
}
b.append(_tcp ? (unsigned char)1 : (unsigned char)0);
b.append(_fixed ? (unsigned char)1 : (unsigned char)0);
}
template<unsigned int C>
inline unsigned int deserialize(const Buffer<C> &b,unsigned int startAt = 0)
{
unsigned int p = startAt;
if (b[p++] != ZT_PATH_SERIALIZATION_VERSION)
throw std::invalid_argument("Path: deserialize(): version mismatch");
_lastSend = b.template at<uint64_t>(p); p += sizeof(uint64_t);
_lastReceived = b.template at<uint64_t>(p); p += sizeof(uint64_t);
_lastFirewallOpener = b.template at<uint64_t>(p); p += sizeof(uint64_t);
_lastPing = b.template at<uint64_t>(p); p += sizeof(uint64_t);
switch((InetAddress::AddressType)b[p++]) {
case InetAddress::TYPE_IPV4:
_addr.set(b.field(p,4),4,b.template at<uint16_t>(p + 4));
p += 4 + sizeof(uint16_t);
break;
case InetAddress::TYPE_IPV6:
_addr.set(b.field(p,16),16,b.template at<uint16_t>(p + 16));
p += 16 + sizeof(uint16_t);
break;
default:
_addr.zero();
break;
}
_tcp = (b[p++] != 0);
_fixed = (b[p++] != 0);
return (p - startAt);
}
private:
volatile uint64_t _lastSent;
volatile uint64_t _lastSend;
volatile uint64_t _lastReceived;
volatile uint64_t _lastFirewallOpener;
volatile uint64_t _lastPing;

130
node/Peer.cpp
View File

@ -33,18 +33,14 @@
namespace ZeroTier {
Peer::Peer() :
_id(),
_lastUsed(0),
_lastUnicastFrame(0),
_lastMulticastFrame(0),
_lastAnnouncedTo(0),
_lastPinged(0),
_vMajor(0),
_vMinor(0),
_vRevision(0),
_latency(0)
{
}
_latency(0) {}
Peer::Peer(const Identity &myIdentity,const Identity &peerIdentity)
throw(std::runtime_error) :
@ -53,7 +49,6 @@ Peer::Peer(const Identity &myIdentity,const Identity &peerIdentity)
_lastUnicastFrame(0),
_lastMulticastFrame(0),
_lastAnnouncedTo(0),
_lastPinged(0),
_vMajor(0),
_vMinor(0),
_vRevision(0),
@ -65,6 +60,7 @@ Peer::Peer(const Identity &myIdentity,const Identity &peerIdentity)
void Peer::onReceive(
const RuntimeEnvironment *_r,
const SharedPtr<Socket> &fromSock,
const InetAddress &remoteAddr,
unsigned int hops,
uint64_t packetId,
@ -73,12 +69,24 @@ void Peer::onReceive(
Packet::Verb inReVerb,
uint64_t now)
{
Mutex::Lock _l(_lock);
if (!hops) { // direct packet
// Update last receive info for our direct path
WanPath *const wp = (remoteAddr.isV4() ? &_ipv4p : &_ipv6p);
wp->lastReceive = now;
if (!wp->fixed)
wp->addr = remoteAddr;
// Update receive time on known paths
bool havePath = false;
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
if ((p->address() == remoteAddr)&&(p->tcp() == (fromSock->type() == Socket::ZT_SOCKET_TYPE_TCP))) {
p->received(now);
havePath = true;
break;
}
}
// Learn new UDP paths (learning TCP would require an explicit mechanism)
if ((!havePath)&&(fromSock->type() != Socket::ZT_SOCKET_TYPE_TCP)) {
_paths.push_back(Path(remoteAddr,false,false));
_paths.back().received(now);
}
// Announce multicast LIKEs to peers to whom we have a direct link
if ((now - _lastAnnouncedTo) >= ((ZT_MULTICAST_LIKE_EXPIRE / 2) - 1000)) {
@ -96,18 +104,24 @@ void Peer::onReceive(
bool Peer::send(const RuntimeEnvironment *_r,const void *data,unsigned int len,uint64_t now)
{
if ((_ipv6p.isActive(now))||((!(_ipv4p.addr))&&(_ipv6p.addr))) {
if (_r->sm->send(_ipv6p.addr,false,data,len)) {
_ipv6p.lastSend = now;
return true;
Mutex::Lock _l(_lock);
if (_paths.empty())
return false;
uint64_t bestPathLastReceived = 0;
std::vector<Path>::iterator bestPath;
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
uint64_t lr = p->lastRecevied();
if (lr >= bestPathLastReceived) {
bestPathLastReceived = lr;
bestPath = p;
}
}
if (_ipv4p.addr) {
if (_r->sm->send(_ipv4p.addr,false,data,len)) {
_ipv4p.lastSend = now;
return true;
}
if (_r->sm->send(bestPath->address(),bestPath->tcp(),data,len)) {
bestPath->sent(now);
return true;
}
return false;
@ -116,70 +130,44 @@ bool Peer::send(const RuntimeEnvironment *_r,const void *data,unsigned int len,u
bool Peer::sendFirewallOpener(const RuntimeEnvironment *_r,uint64_t now)
{
bool sent = false;
if (_ipv4p.addr) {
if (_r->sm->sendFirewallOpener(_ipv4p.addr,ZT_FIREWALL_OPENER_HOPS)) {
_ipv4p.lastFirewallOpener = now;
sent = true;
}
}
Mutex::Lock _l(_lock);
if (_ipv6p.addr) {
if (_r->sm->sendFirewallOpener(_ipv6p.addr,ZT_FIREWALL_OPENER_HOPS)) {
_ipv6p.lastFirewallOpener = now;
sent = true;
}
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
if (!p->tcp())
sent |= _r->sm->sendFirewallOpener(p->address(),ZT_FIREWALL_OPENER_HOPS);
}
return sent;
}
bool Peer::sendPing(const RuntimeEnvironment *_r,uint64_t now)
bool Peer::sendPing(const RuntimeEnvironment *_r,uint64_t now,bool firstSinceReset)
{
bool sent = false;
if (_ipv4p.addr) {
TRACE("PING %s(%s)",_id.address().toString().c_str(),_ipv4p.addr.toString().c_str());
if (_r->sw->sendHELLO(SharedPtr<Peer>(this),_ipv4p.addr,false)) {
_ipv4p.lastSend = now;
sent = true;
}
}
SharedPtr<Peer> self(this);
Mutex::Lock _l(_lock);
if (_ipv6p.addr) {
TRACE("PING %s(%s)",_id.address().toString().c_str(),_ipv6p.addr.toString().c_str());
if (_r->sw->sendHELLO(SharedPtr<Peer>(this),_ipv6p.addr,false)) {
_ipv6p.lastSend = now;
sent = true;
bool allPingsUnanswered;
if (!firstSinceReset) {
allPingsUnanswered = true;
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
if (!p->pingUnanswered(now)) {
allPingsUnanswered = false;
break;
}
}
} else allPingsUnanswered = false;
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
if ((allPingsUnanswered)||(!p->tcp())) {
if (_r->sw->sendHELLO(self,p->address(),p->tcp())) {
p->sent(now);
p->pinged(now);
sent = true;
}
}
}
return sent;
}
void Peer::setPathAddress(const InetAddress &addr,bool fixed)
{
if (addr.isV4()) {
_ipv4p.addr = addr;
_ipv4p.fixed = fixed;
} else if (addr.isV6()) {
_ipv6p.addr = addr;
_ipv6p.fixed = fixed;
}
}
void Peer::clearFixedFlag(InetAddress::AddressType t)
{
switch(t) {
case InetAddress::TYPE_NULL:
_ipv4p.fixed = false;
_ipv6p.fixed = false;
break;
case InetAddress::TYPE_IPV4:
_ipv4p.fixed = false;
break;
case InetAddress::TYPE_IPV6:
_ipv6p.fixed = false;
break;
}
}
} // namespace ZeroTier

127
node/Peer.hpp
View File

@ -45,12 +45,12 @@
#include "InetAddress.hpp"
#include "Packet.hpp"
#include "SharedPtr.hpp"
#include "Socket.hpp"
#include "AtomicCounter.hpp"
#include "NonCopyable.hpp"
#include "Mutex.hpp"
// Increment if serialization has changed
#define ZT_PEER_SERIALIZATION_VERSION 7
#define ZT_PEER_SERIALIZATION_VERSION 8
namespace ZeroTier {
@ -61,10 +61,10 @@ class Peer : NonCopyable
{
friend class SharedPtr<Peer>;
private:
~Peer() {}
public:
/**
* Construct an uninitialized peer (used with deserialize())
*/
Peer();
/**
@ -80,12 +80,20 @@ public:
/**
* @return Time peer record was last used in any way
*/
inline uint64_t lastUsed() const throw() { return _lastUsed; }
inline uint64_t lastUsed() const
throw()
{
return _lastUsed;
}
/**
* @param now New time of last use
*/
inline void setLastUsed(uint64_t now) throw() { _lastUsed = now; }
inline void setLastUsed(uint64_t now)
throw()
{
_lastUsed = now;
}
/**
* @return This peer's ZT address (short for identity().address())
@ -101,7 +109,7 @@ public:
* Must be called on authenticated packet receive from this peer
*
* @param _r Runtime environment
* @param localPort Local port on which packet was received
* @param fromSock Socket from which packet was received
* @param remoteAddr Internet address of sender
* @param hops ZeroTier (not IP) hops
* @param packetId Packet ID
@ -112,6 +120,7 @@ public:
*/
void onReceive(
const RuntimeEnvironment *_r,
const SharedPtr<Socket> &fromSock,
const InetAddress &remoteAddr,
unsigned int hops,
uint64_t packetId,
@ -135,7 +144,7 @@ public:
bool send(const RuntimeEnvironment *_r,const void *data,unsigned int len,uint64_t now);
/**
* Send firewall opener to active link
* Send firewall opener to all UDP paths
*
* @param _r Runtime environment
* @param now Current time
@ -144,39 +153,73 @@ public:
bool sendFirewallOpener(const RuntimeEnvironment *_r,uint64_t now);
/**
* Send HELLO to a peer via all active direct paths available
* Send HELLO to a peer via all direct paths available
*
* This begins attempting to use TCP paths if no ping response has been
* received from any UDP path in more than ZT_TCP_FALLBACK_AFTER.
*
* @param _r Runtime environment
* @param now Current time
* @param firstSinceReset If true, this is the first ping sent since a network reset
* @return True if send appears successful for at least one address type
*/
bool sendPing(const RuntimeEnvironment *_r,uint64_t now);
bool sendPing(const RuntimeEnvironment *_r,uint64_t now,bool firstSinceReset);
/**
* @return Last successfully sent firewall opener
* @return All known direct paths to this peer
*/
std::vector<Path> paths() const
{
Mutex::Lock _l(_lock);
return _paths;
}
/**
* @return Last successfully sent firewall opener for any path
*/
inline uint64_t lastFirewallOpener() const
throw()
{
uint64_t x = 0;
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
uint64_t l = p->lastFirewallOpener();
if (l > x)
x = l;
}
return x;
}
/**
* @return Time of last direct packet receive
* @return Time of last direct packet receive for any path
*/
inline uint64_t lastDirectReceive() const
throw()
{
uint64_t x = 0;
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
uint64_t l = p->lastReceive();
if (l > x)
x = l;
}
return x;
}
/**
* @return Time of last direct packet send
* @return Time of last direct packet send for any path
*/
inline uint64_t lastDirectSend() const
throw()
{
uint64_t x = 0;
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
uint64_t l = p->lastSend();
if (l > x)
x = l;
}
return x;
}
/**
@ -246,6 +289,8 @@ public:
inline bool hasDirectPath() const
throw()
{
Mutex::Lock _l(_lock);
return (!_paths.empty());
}
/**
@ -255,6 +300,12 @@ public:
inline bool hasActiveDirectPath(uint64_t now) const
throw()
{
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
if (p->active(now))
return true;
}
return false;
}
/**
@ -262,8 +313,16 @@ public:
*
* @param p New path to add
*/
inline void addPath(const Path &p)
inline void addPath(const Path &newp)
{
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
if (*p == newp) {
p->setFixed(newp.fixed());
return;
}
}
_paths.push_back(newp);
}
/**
@ -273,6 +332,15 @@ public:
*/
inline void clearPaths(bool fixedToo)
{
std::vector<Path> npv;
Mutex::Lock _l(_lock);
if (!fixedToo) {
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
if (p->fixed())
npv.push_back(*p);
}
}
_paths = npv;
}
/**
@ -335,13 +403,13 @@ public:
}
template<unsigned int C>
inline void serialize(Buffer<C> &b)
inline void serialize(Buffer<C> &b) const
{
Mutex::Lock _l(_lock);
b.append((unsigned char)ZT_PEER_SERIALIZATION_VERSION);
b.append(_key,sizeof(_key));
_id.serialize(b,false);
_ipv4p.serialize(b);
_ipv6p.serialize(b);
b.append(_key,sizeof(_key));
b.append(_lastUsed);
b.append(_lastUnicastFrame);
b.append(_lastMulticastFrame);
@ -350,6 +418,9 @@ public:
b.append((uint16_t)_vMinor);
b.append((uint16_t)_vRevision);
b.append((uint16_t)_latency);
b.append((uint16_t)_paths.size());
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p)
p->serialize(b);
}
template<unsigned int C>
inline unsigned int deserialize(const Buffer<C> &b,unsigned int startAt = 0)
@ -359,10 +430,10 @@ public:
if (b[p++] != ZT_PEER_SERIALIZATION_VERSION)
throw std::invalid_argument("Peer: deserialize(): version mismatch");
memcpy(_key,b.field(p,sizeof(_key)),sizeof(_key)); p += sizeof(_key);
Mutex::Lock _l(_lock);
p += _id.deserialize(b,p);
p += _ipv4p.deserialize(b,p);
p += _ipv6p.deserialize(b,p);
memcpy(_key,b.field(p,sizeof(_key)),sizeof(_key)); p += sizeof(_key);
_lastUsed = b.template at<uint64_t>(p); p += sizeof(uint64_t);
_lastUnicastFrame = b.template at<uint64_t>(p); p += sizeof(uint64_t);
_lastMulticastFrame = b.template at<uint64_t>(p); p += sizeof(uint64_t);
@ -371,6 +442,12 @@ public:
_vMinor = b.template at<uint16_t>(p); p += sizeof(uint16_t);
_vRevision = b.template at<uint16_t>(p); p += sizeof(uint16_t);
_latency = b.template at<uint16_t>(p); p += sizeof(uint16_t);
unsigned int npaths = (unsigned int)b.template at<uint16_t>(p); p += sizeof(uint16_t);
_paths.clear();
for(unsigned int i=0;i<npaths;++i) {
_paths.push_back(Path());
p += _paths.back().deserialize(b,p);
}
return (p - startAt);
}
@ -385,9 +462,13 @@ private:
volatile uint64_t _lastUnicastFrame;
volatile uint64_t _lastMulticastFrame;
volatile uint64_t _lastAnnouncedTo;
volatile unsigned int _vMajor,_vMinor,_vRevision;
volatile unsigned int _vMajor;
volatile unsigned int _vMinor;
volatile unsigned int _vRevision;
volatile unsigned int _latency;
Mutex _lock;
AtomicCounter __refCount;
};

15
node/SocketManager.cpp
View File

@ -577,4 +577,19 @@ void SocketManager::whack()
_whackSendPipe_m.unlock();
}
void closeTcpSockets()
{
{
Mutex::Lock _l2(_tcpSockets_m);
_fdSetLock.lock();
for(std::map< InetAddress,SharedPtr<Socket> >::iterator s(_tcpSockets.begin());s!=_tcpSockets.end();++s`) {
FD_CLR((*s)->_sock,&_readfds);
FD_CLR((*s)->_sock,&_writefds);
}
_fdSetLock.unlock();
_tcpSockets.clear();
}
_updateNfds();
}
} // namespace ZeroTier

5
node/SocketManager.hpp
View File

@ -118,6 +118,11 @@ public:
*/
void whack();
/**
* Close TCP sockets
*/
void closeTcpSockets();
private:
// Called by socket implementations when a packet is received
inline void handleReceivedPacket(const SharedPtr<Socket> &sock,const InetAddress &from,Buffer<ZT_SOCKET_MAX_MESSAGE_LEN> &data)

2
node/Topology.hpp
View File

@ -260,7 +260,7 @@ public:
inline void operator()(Topology &t,const SharedPtr<Peer> &p)
{
if (!_supernodeAddresses.count(p->address())) {
p->forgetDirectPaths(false); // false means don't forget 'fixed' paths e.g. supernodes
p->clearPaths(false); // false means don't forget 'fixed' paths e.g. supernodes
if (((_now - p->lastFrame()) < ZT_PEER_LINK_ACTIVITY_TIMEOUT)&&(_supernode)) {
TRACE("sending reset NOP to %s",p->address().toString().c_str());
Packet outp(p->address(),_r->identity.address(),Packet::VERB_NOP);