diff options
| -rw-r--r-- | node/Constants.hpp | 2 | ||||
| -rw-r--r-- | node/Node.cpp | 1 | ||||
| -rw-r--r-- | node/Path.hpp | 90 | ||||
| -rw-r--r-- | node/Peer.cpp | 130 | ||||
| -rw-r--r-- | node/Peer.hpp | 127 | ||||
| -rw-r--r-- | node/SocketManager.cpp | 15 | ||||
| -rw-r--r-- | node/SocketManager.hpp | 5 | ||||
| -rw-r--r-- | node/Topology.hpp | 2 |
8 files changed, 268 insertions, 104 deletions
diff --git a/node/Constants.hpp b/node/Constants.hpp index 27d80ed6..cf031e64 100644 --- a/node/Constants.hpp +++ b/node/Constants.hpp @@ -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 diff --git a/node/Node.cpp b/node/Node.cpp index 21d78f54..a4f2c3af 100644 --- a/node/Node.cpp +++ b/node/Node.cpp @@ -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. diff --git a/node/Path.hpp b/node/Path.hpp index 2dd3b560..21ef71bb 100644 --- a/node/Path.hpp +++ b/node/Path.hpp @@ -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; } @@ -82,13 +88,26 @@ public: } /** + * @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 */ inline std::string toString() const { 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; diff --git a/node/Peer.cpp b/node/Peer.cpp index 3a214b6a..ce203890 100644 --- a/node/Peer.cpp +++ b/node/Peer.cpp @@ -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); + + 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; + } } - } - - 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; + } 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 diff --git a/node/Peer.hpp b/node/Peer.hpp index 155053d6..2e19ba14 100644 --- a/node/Peer.hpp +++ b/node/Peer.hpp @@ -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 All known direct paths to this peer + */ + std::vector<Path> paths() const + { + Mutex::Lock _l(_lock); + return _paths; + } /** - * @return Last successfully sent firewall opener + * @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; }; diff --git a/node/SocketManager.cpp b/node/SocketManager.cpp index 7af4b29d..a977a546 100644 --- a/node/SocketManager.cpp +++ b/node/SocketManager.cpp @@ -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 diff --git a/node/SocketManager.hpp b/node/SocketManager.hpp index 20a57285..dbc4e686 100644 --- a/node/SocketManager.hpp +++ b/node/SocketManager.hpp @@ -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) diff --git a/node/Topology.hpp b/node/Topology.hpp index 5c484bb1..e575cb8d 100644 --- a/node/Topology.hpp +++ b/node/Topology.hpp @@ -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); |