mirror
/
devilutionX-ZeroTierOne
mirror of https://github.com/diasurgical/ZeroTierOne.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Simplify locking semantics some more to address a deadlock.

pull/1/head
Adam Ierymenko 12 years ago
parent
7c0f5e97e1
commit
0e47f13f14
4 changed files with 161 additions and 204 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 30
      node/IncomingPacket.cpp
  2. 70
      node/Path.hpp
  3. 159
      node/Peer.cpp
  4. 106
      node/Peer.hpp

30
node/IncomingPacket.cpp
Unescape View File

@ -70,7 +70,7 @@ bool IncomingPacket::tryDecode(const RuntimeEnvironment *RR)
switch(verb()) {
//case Packet::VERB_NOP:
default: // ignore unknown verbs, but if they pass auth check they are "received"
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),verb(),0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),verb(),0,Packet::VERB_NOP,Utils::now());
return true;
case Packet::VERB_HELLO: return _doHELLO(RR);
case Packet::VERB_ERROR: return _doERROR(RR,peer);
@ -152,7 +152,7 @@ bool IncomingPacket::_doERROR(const RuntimeEnvironment *RR,const SharedPtr<Peer>
default: break;
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_ERROR,inRePacketId,inReVerb,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_ERROR,inRePacketId,inReVerb,Utils::now());
} catch (std::exception &ex) {
TRACE("dropped ERROR from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
@ -214,7 +214,7 @@ bool IncomingPacket::_doHELLO(const RuntimeEnvironment *RR)
peer = RR->topology->addPeer(newPeer);
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_HELLO,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_HELLO,0,Packet::VERB_NOP,Utils::now());
peer->setRemoteVersion(protoVersion,vMajor,vMinor,vRevision);
// If a supernode has a version higher than ours, this causes a software
@ -334,7 +334,7 @@ bool IncomingPacket::_doOK(const RuntimeEnvironment *RR,const SharedPtr<Peer> &p
default: break;
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_OK,inRePacketId,inReVerb,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_OK,inRePacketId,inReVerb,Utils::now());
} catch (std::exception &ex) {
TRACE("dropped OK from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
@ -367,7 +367,7 @@ bool IncomingPacket::_doWHOIS(const RuntimeEnvironment *RR,const SharedPtr<Peer>
} else {
TRACE("dropped WHOIS from %s(%s): missing or invalid address",source().toString().c_str(),_remoteAddress.toString().c_str());
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_WHOIS,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_WHOIS,0,Packet::VERB_NOP,Utils::now());
} catch ( ... ) {
TRACE("dropped WHOIS from %s(%s): unexpected exception",source().toString().c_str(),_remoteAddress.toString().c_str());
}
@ -399,7 +399,7 @@ bool IncomingPacket::_doRENDEZVOUS(const RuntimeEnvironment *RR,const SharedPtr<
if ((port > 0)&&((addrlen == 4)||(addrlen == 16))) {
InetAddress atAddr(field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ADDRESS,addrlen),addrlen,port);
TRACE("RENDEZVOUS from %s says %s might be at %s, starting NAT-t",source().toString().c_str(),with.toString().c_str(),atAddr.toString().c_str());
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_RENDEZVOUS,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_RENDEZVOUS,0,Packet::VERB_NOP,Utils::now());
RR->sw->contact(withPeer,atAddr);
} else {
TRACE("dropped corrupt RENDEZVOUS from %s(%s) (bad address or port)",source().toString().c_str(),_remoteAddress.toString().c_str());
@ -440,7 +440,7 @@ bool IncomingPacket::_doFRAME(const RuntimeEnvironment *RR,const SharedPtr<Peer>
network->tapPut(MAC(peer->address(),network->id()),network->mac(),etherType,field(ZT_PROTO_VERB_FRAME_IDX_PAYLOAD,payloadLen),payloadLen);
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_FRAME,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_FRAME,0,Packet::VERB_NOP,Utils::now());
} else {
TRACE("dropped FRAME from %s(%s): we are not connected to network %.16llx",source().toString().c_str(),_remoteAddress.toString().c_str(),at<uint64_t>(ZT_PROTO_VERB_FRAME_IDX_NETWORK_ID));
}
@ -517,7 +517,7 @@ bool IncomingPacket::_doEXT_FRAME(const RuntimeEnvironment *RR,const SharedPtr<P
network->tapPut(from,to,etherType,field(comLen + ZT_PROTO_VERB_EXT_FRAME_IDX_PAYLOAD,payloadLen),payloadLen);
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_EXT_FRAME,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_EXT_FRAME,0,Packet::VERB_NOP,Utils::now());
} else {
TRACE("dropped EXT_FRAME from %s(%s): we are not connected to network %.16llx",source().toString().c_str(),_remoteAddress.toString().c_str(),at<uint64_t>(ZT_PROTO_VERB_FRAME_IDX_NETWORK_ID));
}
@ -594,7 +594,7 @@ bool IncomingPacket::_doP5_MULTICAST_FRAME(const RuntimeEnvironment *RR,const Sh
}
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_P5_MULTICAST_FRAME,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_P5_MULTICAST_FRAME,0,Packet::VERB_NOP,Utils::now());
if (RR->topology->amSupernode()) {
// To support legacy peers, old fashioned "P5" multicasts are propagated manually by supernodes.
@ -652,7 +652,7 @@ bool IncomingPacket::_doMULTICAST_LIKE(const RuntimeEnvironment *RR,const Shared
for(unsigned int ptr=ZT_PACKET_IDX_PAYLOAD;ptr<size();ptr+=18)
RR->mc->add(now,at<uint64_t>(ptr),MulticastGroup(MAC(field(ptr + 8,6),6),at<uint32_t>(ptr + 14)),Address(),peer->address());
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_MULTICAST_LIKE,0,Packet::VERB_NOP,now);
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_MULTICAST_LIKE,0,Packet::VERB_NOP,now);
} catch (std::exception &ex) {
TRACE("dropped MULTICAST_LIKE from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
@ -676,7 +676,7 @@ bool IncomingPacket::_doNETWORK_MEMBERSHIP_CERTIFICATE(const RuntimeEnvironment
}
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_NETWORK_MEMBERSHIP_CERTIFICATE,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_NETWORK_MEMBERSHIP_CERTIFICATE,0,Packet::VERB_NOP,Utils::now());
} catch (std::exception &ex) {
TRACE("dropped NETWORK_MEMBERSHIP_CERTIFICATE from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());
} catch ( ... ) {
@ -726,7 +726,7 @@ bool IncomingPacket::_doNETWORK_CONFIG_REQUEST(const RuntimeEnvironment *RR,cons
}
#endif // !__WINDOWS__
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_NETWORK_CONFIG_REQUEST,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_NETWORK_CONFIG_REQUEST,0,Packet::VERB_NOP,Utils::now());
} catch (std::exception &exc) {
TRACE("dropped NETWORK_CONFIG_REQUEST from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),exc.what());
} catch ( ... ) {
@ -747,7 +747,7 @@ bool IncomingPacket::_doNETWORK_CONFIG_REFRESH(const RuntimeEnvironment *RR,cons
nw->requestConfiguration();
}
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_NETWORK_CONFIG_REFRESH,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_NETWORK_CONFIG_REFRESH,0,Packet::VERB_NOP,Utils::now());
} catch (std::exception &exc) {
TRACE("dropped NETWORK_CONFIG_REFRESH from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),exc.what());
} catch ( ... ) {
@ -778,7 +778,7 @@ bool IncomingPacket::_doMULTICAST_GATHER(const RuntimeEnvironment *RR,const Shar
}
}
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_MULTICAST_GATHER,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_MULTICAST_GATHER,0,Packet::VERB_NOP,Utils::now());
} catch (std::exception &exc) {
TRACE("dropped MULTICAST_GATHER from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),exc.what());
} catch ( ... ) {
@ -870,7 +870,7 @@ bool IncomingPacket::_doMULTICAST_FRAME(const RuntimeEnvironment *RR,const Share
}
} // else ignore -- not a member of this network
peer->receive(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_MULTICAST_FRAME,0,Packet::VERB_NOP,Utils::now());
peer->received(RR,_fromSock,_remoteAddress,hops(),packetId(),Packet::VERB_MULTICAST_FRAME,0,Packet::VERB_NOP,Utils::now());
} catch (std::exception &exc) {
TRACE("dropped MULTICAST_FRAME from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),exc.what());
} catch ( ... ) {

70
node/Path.hpp
Unescape View File

@ -37,14 +37,14 @@
#include "Constants.hpp"
#include "InetAddress.hpp"
#include "Utils.hpp"
#include "Buffer.hpp"
#define ZT_PATH_SERIALIZATION_VERSION 3
namespace ZeroTier {
/**
* WAN address and protocol for reaching a peer
*
* This structure is volatile and memcpy-able, and depends on
* InetAddress being similarly safe.
*/
class Path
{
@ -67,7 +67,6 @@ public:
Path(const Path &p)
{
// InetAddress is memcpy'able
memcpy(this,&p,sizeof(Path));
}
@ -79,6 +78,16 @@ public:
_type(t),
_fixed(fixed) {}
inline void init(const InetAddress &addr,Type t,bool fixed = false)
{
_lastSend = 0;
_lastReceived = 0;
_lastPing = 0;
_addr = addr;
_type = t;
_fixed = fixed;
}
inline Path &operator=(const Path &p)
{
if (this != &p)
@ -150,59 +159,6 @@ 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(_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((unsigned char)_type);
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);
_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;
}
_type = (Type)b[p++];
_fixed = (b[p++] != 0);
return (p - startAt);
}
private:
volatile uint64_t _lastSend;
volatile uint64_t _lastReceived;

159
node/Peer.cpp
Unescape View File

@ -37,20 +37,8 @@
namespace ZeroTier {
Peer::Peer() :
_lastUsed(0),
_lastReceive(0),
_lastUnicastFrame(0),
_lastMulticastFrame(0),
_lastAnnouncedTo(0),
_vMajor(0),
_vMinor(0),
_vRevision(0),
_latency(0) {}
Peer::Peer(const Identity &myIdentity,const Identity &peerIdentity)
throw(std::runtime_error) :
_id(peerIdentity),
_lastUsed(0),
_lastReceive(0),
_lastUnicastFrame(0),
@ -59,13 +47,15 @@ Peer::Peer(const Identity &myIdentity,const Identity &peerIdentity)
_vMajor(0),
_vMinor(0),
_vRevision(0),
_latency(0)
_numPaths(0),
_latency(0),
_id(peerIdentity)
{
if (!myIdentity.agree(peerIdentity,_key,ZT_PEER_SECRET_KEY_LENGTH))
throw std::runtime_error("new peer identity key agreement failed");
}
void Peer::receive(
void Peer::received(
const RuntimeEnvironment *RR,
const SharedPtr<Socket> &fromSock,
const InetAddress &remoteAddr,
@ -79,8 +69,6 @@ void Peer::receive(
// Update system-wide last packet receive time
*((const_cast<uint64_t *>(&(RR->timeOfLastPacketReceived)))) = now;
Mutex::Lock _l(_lock);
// Global last receive time regardless of path
_lastReceive = now;
@ -88,28 +76,35 @@ void Peer::receive(
// Learn paths from direct packets (hops == 0)
{
bool havePath = false;
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
if ((p->address() == remoteAddr)&&(p->tcp() == fromSock->tcp())) {
p->received(now);
for(unsigned int p=0,np=_numPaths;p<np;++p) {
if ((_paths[p].address() == remoteAddr)&&(_paths[p].tcp() == fromSock->tcp())) {
_paths[p].received(now);
havePath = true;
break;
}
}
if (!havePath) {
Path::Type pt = Path::PATH_TYPE_UDP;
switch(fromSock->type()) {
case Socket::ZT_SOCKET_TYPE_TCP_IN:
pt = Path::PATH_TYPE_TCP_IN;
break;
case Socket::ZT_SOCKET_TYPE_TCP_OUT:
pt = Path::PATH_TYPE_TCP_OUT;
break;
default:
break;
unsigned int np = _numPaths;
if (np >= ZT_PEER_MAX_PATHS)
clean(now);
np = _numPaths;
if (np < ZT_PEER_MAX_PATHS) {
Path::Type pt = Path::PATH_TYPE_UDP;
switch(fromSock->type()) {
case Socket::ZT_SOCKET_TYPE_TCP_IN:
pt = Path::PATH_TYPE_TCP_IN;
break;
case Socket::ZT_SOCKET_TYPE_TCP_OUT:
pt = Path::PATH_TYPE_TCP_OUT;
break;
default:
break;
}
_paths[np].init(remoteAddr,pt,false);
_paths[np].received(now);
_numPaths = ++np;
}
_paths.push_back(Path(remoteAddr,pt,false));
_paths.back().received(now);
}
}
@ -121,10 +116,12 @@ void Peer::receive(
if ((now - _lastAnnouncedTo) >= ((ZT_MULTICAST_LIKE_EXPIRE / 2) - 1000)) {
_lastAnnouncedTo = now;
bool isSupernode = RR->topology->isSupernode(_id.address());
Packet outp(_id.address(),RR->identity.address(),Packet::VERB_MULTICAST_LIKE);
std::vector< SharedPtr<Network> > networks(RR->nc->networks());
for(std::vector< SharedPtr<Network> >::iterator n(networks.begin());n!=networks.end();++n) {
if ( ((*n)->isAllowed(_id.address())) || ((*n)->controller() == _id.address()) || (RR->topology->isSupernode(_id.address())) ) {
if ( ((*n)->isAllowed(_id.address())) || (isSupernode) ) {
std::set<MulticastGroup> mgs((*n)->multicastGroups());
for(std::set<MulticastGroup>::iterator mg(mgs.begin());mg!=mgs.end();++mg) {
if ((outp.size() + 18) > ZT_UDP_DEFAULT_PAYLOAD_MTU) {
@ -155,8 +152,6 @@ void Peer::receive(
Path::Type Peer::send(const RuntimeEnvironment *RR,const void *data,unsigned int len,uint64_t now)
{
Mutex::Lock _l(_lock);
/* For sending ordinary packets, paths are divided into two categories:
* "normal" and "TCP out." Normal includes UDP and incoming TCP. We want
* to treat outbound TCP differently since if we use it it may end up
@ -166,17 +161,17 @@ Path::Type Peer::send(const RuntimeEnvironment *RR,const void *data,unsigned int
Path *bestTcpOutPath = (Path *)0;
uint64_t bestNormalPathLastReceived = 0;
uint64_t bestTcpOutPathLastReceived = 0;
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
uint64_t lr = p->lastReceived();
if (p->type() == Path::PATH_TYPE_TCP_OUT) {
for(unsigned int p=0,np=_numPaths;p<np;++p) {
uint64_t lr = _paths[p].lastReceived();
if (_paths[p].type() == Path::PATH_TYPE_TCP_OUT) {
if (lr >= bestTcpOutPathLastReceived) {
bestTcpOutPathLastReceived = lr;
bestTcpOutPath = &(*p);
bestTcpOutPath = &(_paths[p]);
}
} else {
if (lr >= bestNormalPathLastReceived) {
bestNormalPathLastReceived = lr;
bestNormalPath = &(*p);
bestNormalPath = &(_paths[p]);
}
}
}
@ -214,7 +209,6 @@ bool Peer::sendPing(const RuntimeEnvironment *RR,uint64_t now)
{
bool sent = false;
SharedPtr<Peer> self(this);
Mutex::Lock _l(_lock);
/* Ping (and thus open) outbound TCP connections if we have no other options
* or if the TCP tunneling master switch is enabled and pings have been
@ -222,22 +216,22 @@ bool Peer::sendPing(const RuntimeEnvironment *RR,uint64_t now)
uint64_t lastNormalPingSent = 0;
uint64_t lastNormalReceive = 0;
bool haveNormal = false;
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
if (p->type() != Path::PATH_TYPE_TCP_OUT) {
lastNormalPingSent = std::max(lastNormalPingSent,p->lastPing());
lastNormalReceive = std::max(lastNormalReceive,p->lastReceived());
for(unsigned int p=0,np=_numPaths;p<np;++p) {
if (_paths[p].type() != Path::PATH_TYPE_TCP_OUT) {
lastNormalPingSent = std::max(lastNormalPingSent,_paths[p].lastPing());
lastNormalReceive = std::max(lastNormalReceive,_paths[p].lastReceived());
haveNormal = true;
}
}
const bool useTcpOut = ( (!haveNormal) || ( (RR->tcpTunnelingEnabled) && (lastNormalPingSent > RR->timeOfLastResynchronize) && (lastNormalPingSent > lastNormalReceive) && ((lastNormalPingSent - lastNormalReceive) >= ZT_TCP_TUNNEL_FAILOVER_TIMEOUT) ) );
const bool useTcpOut = ( (!haveNormal) || ( (RR->tcpTunnelingEnabled) && (lastNormalPingSent > RR->timeOfLastResynchronize) && (lastNormalPingSent > lastNormalReceive) && ((lastNormalPingSent - lastNormalReceive) >= ZT_TCP_TUNNEL_FAILOVER_TIMEOUT) ) );
TRACE("PING %s (useTcpOut==%d)",_id.address().toString().c_str(),(int)useTcpOut);
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
if ((useTcpOut)||(p->type() != Path::PATH_TYPE_TCP_OUT)) {
p->pinged(now); // attempts to ping are logged whether they look successful or not
if (RR->sw->sendHELLO(self,*p)) {
p->sent(now);
for(unsigned int p=0,np=_numPaths;p<np;++p) {
if ((useTcpOut)||(_paths[p].type() != Path::PATH_TYPE_TCP_OUT)) {
_paths[p].pinged(now); // attempts to ping are logged whether they look successful or not
if (RR->sw->sendHELLO(self,_paths[p])) {
_paths[p].sent(now);
sent = true;
}
}
@ -248,33 +242,68 @@ bool Peer::sendPing(const RuntimeEnvironment *RR,uint64_t now)
void Peer::clean(uint64_t now)
{
Mutex::Lock _l(_lock);
unsigned long i = 0,o = 0,l = (unsigned long)_paths.size();
while (i != l) {
if (_paths[i].active(now)) // active includes fixed
_paths[o++] = _paths[i];
++i;
unsigned int np = _numPaths;
unsigned int x = 0;
unsigned int y = 0;
while (x < np) {
if (_paths[x].active(now))
_paths[y++] = _paths[x];
++x;
}
_numPaths = y;
}
void Peer::addPath(const Path &newp)
{
unsigned int np = _numPaths;
for(unsigned int p=0;p<np;++p) {
if (_paths[p] == newp) {
_paths[p].setFixed(newp.fixed());
return;
}
}
if (np >= ZT_PEER_MAX_PATHS)
clean(Utils::now());
np = _numPaths;
if (np < ZT_PEER_MAX_PATHS) {
_paths[np] = newp;
_numPaths = ++np;
}
}
void Peer::clearPaths(bool fixedToo)
{
if (fixedToo) {
_numPaths = 0;
} else {
unsigned int np = _numPaths;
unsigned int x = 0;
unsigned int y = 0;
while (x < np) {
if (_paths[x].fixed())
_paths[y++] = _paths[x];
++x;
}
_numPaths = y;
}
_paths.resize(o);
}
void Peer::getBestActiveUdpPathAddresses(uint64_t now,InetAddress &v4,InetAddress &v6) const
{
uint64_t bestV4 = 0,bestV6 = 0;
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
if ((p->type() == Path::PATH_TYPE_UDP)&&(p->active(now))) {
uint64_t lr = p->lastReceived();
for(unsigned int p=0,np=_numPaths;p<np;++p) {
if ((_paths[p].type() == Path::PATH_TYPE_UDP)&&(_paths[p].active(now))) {
uint64_t lr = _paths[p].lastReceived();
if (lr) {
if (p->address().isV4()) {
if (_paths[p].address().isV4()) {
if (lr >= bestV4) {
bestV4 = lr;
v4 = p->address();
v4 = _paths[p].address();
}
} else if (p->address().isV6()) {
} else if (_paths[p].address().isV6()) {
if (lr >= bestV6) {
bestV6 = lr;
v6 = p->address();
v6 = _paths[p].address();
}
}
}

106
node/Peer.hpp
Unescape View File

@ -30,9 +30,9 @@
#include <stdint.h>
#include <vector>
#include <algorithm>
#include <utility>
#include <vector>
#include <stdexcept>
#include "Constants.hpp"
@ -48,23 +48,29 @@
#include "Socket.hpp"
#include "AtomicCounter.hpp"
#include "NonCopyable.hpp"
#include "Mutex.hpp"
/**
* Maximum number of paths a peer can have
*/
#define ZT_PEER_MAX_PATHS 8
namespace ZeroTier {
/**
* Peer on P2P Network
*
* This struture is not locked, volatile, and memcpy-able. NonCopyable
* semantics are just there to prevent bugs, not because it isn't safe
* to copy.
*/
class Peer : NonCopyable
{
friend class SharedPtr<Peer>;
public:
/**
* Construct an uninitialized peer (used with deserialize())
*/
Peer();
private:
Peer() {} // disabled to prevent bugs -- should not be constructed uninitialized
public:
~Peer() { Utils::burn(_key,sizeof(_key)); }
/**
@ -115,7 +121,7 @@ public:
* @param inReVerb Verb in reply to (for OK/ERROR, VERB_NOP otherwise)
* @param now Current time
*/
void receive(
void received(
const RuntimeEnvironment *RR,
const SharedPtr<Socket> &fromSock,
const InetAddress &remoteAddr,
@ -162,8 +168,10 @@ public:
*/
std::vector<Path> paths() const
{
Mutex::Lock _l(_lock);
return _paths;
std::vector<Path> pp;
for(unsigned int p=0,np=_numPaths;p<np;++p)
pp.push_back(_paths[p]);
return pp;
}
/**
@ -172,9 +180,8 @@ public:
*/
inline bool haveUdpPath(const InetAddress &addr) const
{
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
if ((p->type() == Path::PATH_TYPE_UDP)&&(p->address() == addr))
for(unsigned int p=0,np=_numPaths;p<np;++p) {
if ((_paths[p].type() == Path::PATH_TYPE_UDP)&&(_paths[p].address() == addr))
return true;
}
return false;
@ -187,9 +194,8 @@ public:
throw()
{
uint64_t x = 0;
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p)
x = std::max(x,p->lastReceived());
for(unsigned int p=0,np=_numPaths;p<np;++p)
x = std::max(x,_paths[p].lastReceived());
return x;
}
@ -200,9 +206,8 @@ public:
throw()
{
uint64_t x = 0;
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p)
x = std::max(x,p->lastSend());
for(unsigned int p=0,np=_numPaths;p<np;++p)
x = std::max(x,_paths[p].lastSend());
return x;
}
@ -215,10 +220,9 @@ public:
inline void lastPingAndDirectReceive(uint64_t &lp,uint64_t &lr)
throw()
{
Mutex::Lock _l(_lock);
for(std::vector<Path>::const_iterator p(_paths.begin());p!=_paths.end();++p) {
lp = std::max(lp,p->lastPing());
lr = std::max(lr,p->lastReceived());
for(unsigned int p=0,np=_numPaths;p<np;++p) {
lp = std::max(lp,_paths[p].lastPing());
lr = std::max(lr,_paths[p].lastReceived());
}
}
@ -251,11 +255,7 @@ public:
* @param now Current time
* @return True if peer has received something within ZT_PEER_ACTIVITY_TIMEOUT ms
*/
inline bool alive(uint64_t now) const
throw()
{
return ((now - _lastReceive) < ZT_PEER_ACTIVITY_TIMEOUT);
}
inline bool alive(uint64_t now) const throw() { return ((now - _lastReceive) < ZT_PEER_ACTIVITY_TIMEOUT); }
/**
* @return Current latency or 0 if unknown (max: 65535)
@ -284,12 +284,7 @@ public:
/**
* @return True if this peer has at least one direct IP address path
*/
inline bool hasDirectPath() const
throw()
{
Mutex::Lock _l(_lock);
return (!_paths.empty());
}
inline bool hasDirectPath() const throw() { return (_numPaths != 0); }
/**
* @param now Current time
@ -298,9 +293,8 @@ 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))
for(unsigned int p=0,np=_numPaths;p<np;++p) {
if (_paths[p].active(now))
return true;
}
return false;
@ -311,35 +305,14 @@ public:
*
* @param p New path to add
*/
inline void addPath(const Path &newp)
{
Mutex::Lock _l(_lock);
for(std::vector<Path>::iterator p(_paths.begin());p!=_paths.end();++p) {
if (*p == newp) {
p->setFixed(newp.fixed());
return;
}
}
_paths.push_back(newp);
}
void addPath(const Path &newp);
/**
* Clear paths
*
* @param fixedToo If true, clear fixed paths as well as learned ones
*/
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;
}
void clearPaths(bool fixedToo);
/**
* @return 256-bit secret symmetric encryption key
@ -405,11 +378,6 @@ public:
private:
void _announceMulticastGroups(const RuntimeEnvironment *RR,uint64_t now);
unsigned char _key[ZT_PEER_SECRET_KEY_LENGTH];
Identity _id;
std::vector<Path> _paths;
volatile uint64_t _lastUsed;
volatile uint64_t _lastReceive; // direct or indirect
volatile uint64_t _lastUnicastFrame;
@ -419,9 +387,13 @@ private:
volatile uint16_t _vMajor;
volatile uint16_t _vMinor;
volatile uint16_t _vRevision;
volatile unsigned int _latency;
Mutex _lock;
Path _paths[ZT_PEER_MAX_PATHS];
volatile unsigned int _numPaths;
volatile unsigned int _latency;
unsigned char _key[ZT_PEER_SECRET_KEY_LENGTH];
Identity _id;
AtomicCounter __refCount;
};

Write Preview
Loading…
Cancel
Save