mirror of
https://github.com/ZeroTier/ZeroTierOne
synced 2025-08-20 05:13:58 -07:00
.
This commit is contained in:
Adam Ierymenko
parent
baa10c2995
commit
2f20258807
16 changed files with 239 additions and 423 deletions
|
|
@ -216,7 +216,12 @@
|
|||
/**
|
||||
* How often Topology::clean() and Network::clean() and similar are called, in ms
|
||||
*/
|
||||
#define ZT_HOUSEKEEPING_PERIOD 10000
|
||||
#define ZT_HOUSEKEEPING_PERIOD 60000
|
||||
|
||||
/**
|
||||
* How often in ms to write peer state to storage and/or cluster (approximate)
|
||||
*/
|
||||
#define ZT_PEER_STATE_WRITE_PERIOD 10000
|
||||
|
||||
/**
|
||||
* How long to remember peer records in RAM if they haven't been used
|
||||
|
|
@ -322,11 +327,6 @@
|
|||
*/
|
||||
#define ZT_PEER_PATH_EXPIRATION ((ZT_PEER_PING_PERIOD * 4) + 3000)
|
||||
|
||||
/**
|
||||
* Send a full HELLO every this often (ms)
|
||||
*/
|
||||
#define ZT_PEER_SEND_FULL_HELLO_EVERY (ZT_PEER_PING_PERIOD * 2)
|
||||
|
||||
/**
|
||||
* How often to retry expired paths that we're still remembering
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -91,7 +91,10 @@ public:
|
|||
|
||||
~Identity()
|
||||
{
|
||||
delete _privateKey;
|
||||
if (_privateKey) {
|
||||
Utils::burn(_privateKey,sizeof(C25519::Private));
|
||||
delete _privateKey;
|
||||
}
|
||||
}
|
||||
|
||||
inline Identity &operator=(const Identity &id)
|
||||
|
|
|
|||
|
|
@ -585,12 +585,6 @@ bool IncomingPacket::_doWHOIS(const RuntimeEnvironment *RR,void *tPtr,const Shar
|
|||
} else {
|
||||
// Request unknown WHOIS from upstream from us (if we have one)
|
||||
RR->sw->requestWhois(tPtr,addr);
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
// Distribute WHOIS queries across a cluster if we do not know the ID.
|
||||
// This may result in duplicate OKs to the querying peer, which is fine.
|
||||
if (RR->cluster)
|
||||
RR->cluster->sendDistributedQuery(*this);
|
||||
#endif
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1055,12 +1049,6 @@ bool IncomingPacket::_doMULTICAST_GATHER(const RuntimeEnvironment *RR,void *tPtr
|
|||
outp.armor(peer->key(),true,_path->nextOutgoingCounter());
|
||||
_path->send(RR,tPtr,outp.data(),outp.size(),RR->node->now());
|
||||
}
|
||||
|
||||
// If we are a member of a cluster, distribute this GATHER across it
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ((RR->cluster)&&(gatheredLocally < gatherLimit))
|
||||
RR->cluster->sendDistributedQuery(*this);
|
||||
#endif
|
||||
}
|
||||
|
||||
peer->received(tPtr,_path,hops(),packetId(),Packet::VERB_MULTICAST_GATHER,0,Packet::VERB_NOP,trustEstablished);
|
||||
|
|
|
|||
|
|
@ -1067,11 +1067,6 @@ uint64_t Network::handleConfigChunk(void *tPtr,const uint64_t packetId,const Add
|
|||
return 0;
|
||||
}
|
||||
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ((source)&&(RR->cluster))
|
||||
RR->cluster->broadcastNetworkConfigChunk(chunk.field(start,chunk.size() - start),chunk.size() - start);
|
||||
#endif
|
||||
|
||||
// New properly verified chunks can be flooded "virally" through the network
|
||||
if (fastPropagate) {
|
||||
Address *a = (Address *)0;
|
||||
|
|
@ -1099,11 +1094,6 @@ uint64_t Network::handleConfigChunk(void *tPtr,const uint64_t packetId,const Add
|
|||
if ((!c)||(_incomingConfigChunks[i].ts < c->ts))
|
||||
c = &(_incomingConfigChunks[i]);
|
||||
}
|
||||
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ((source)&&(RR->cluster))
|
||||
RR->cluster->broadcastNetworkConfigChunk(chunk.field(start,chunk.size() - start),chunk.size() - start);
|
||||
#endif
|
||||
} else {
|
||||
TRACE("discarded single-chunk unsigned legacy config: this is only allowed if the sender is the controller itself");
|
||||
return 0;
|
||||
|
|
|
|||
151
node/Node.cpp
151
node/Node.cpp
|
|
@ -68,6 +68,7 @@ Node::Node(void *uptr,void *tptr,const struct ZT_Node_Callbacks *callbacks,uint6
|
|||
throw std::runtime_error("callbacks struct version mismatch");
|
||||
memcpy(&_cb,callbacks,sizeof(ZT_Node_Callbacks));
|
||||
|
||||
// Initialize non-cryptographic PRNG from a good random source
|
||||
Utils::getSecureRandom((void *)_prngState,sizeof(_prngState));
|
||||
|
||||
_online = false;
|
||||
|
|
@ -78,33 +79,34 @@ Node::Node(void *uptr,void *tptr,const struct ZT_Node_Callbacks *callbacks,uint6
|
|||
|
||||
uint64_t idtmp[2];
|
||||
idtmp[0] = 0; idtmp[1] = 0;
|
||||
char tmp[512];
|
||||
std::string tmp2;
|
||||
char tmp[1024];
|
||||
int n = stateObjectGet(tptr,ZT_STATE_OBJECT_IDENTITY_SECRET,idtmp,tmp,sizeof(tmp) - 1);
|
||||
if (n > 0) {
|
||||
tmp[n] = (char)0;
|
||||
if (!RR->identity.fromString(tmp))
|
||||
if (RR->identity.fromString(tmp)) {
|
||||
RR->publicIdentityStr = RR->identity.toString(false);
|
||||
RR->secretIdentityStr = RR->identity.toString(true);
|
||||
} else {
|
||||
n = -1;
|
||||
}
|
||||
}
|
||||
|
||||
idtmp[0] = RR->identity.address().toInt(); idtmp[1] = 0;
|
||||
if (n <= 0) {
|
||||
RR->identity.generate();
|
||||
tmp2 = RR->identity.toString(true);
|
||||
stateObjectPut(tptr,ZT_STATE_OBJECT_IDENTITY_SECRET,idtmp,tmp2.data(),(unsigned int)tmp2.length());
|
||||
tmp2 = RR->identity.toString(false);
|
||||
stateObjectPut(tptr,ZT_STATE_OBJECT_IDENTITY_PUBLIC,idtmp,tmp2.data(),(unsigned int)tmp2.length());
|
||||
RR->publicIdentityStr = RR->identity.toString(false);
|
||||
RR->secretIdentityStr = RR->identity.toString(true);
|
||||
stateObjectPut(tptr,ZT_STATE_OBJECT_IDENTITY_SECRET,idtmp,RR->secretIdentityStr.data(),(unsigned int)RR->secretIdentityStr.length());
|
||||
stateObjectPut(tptr,ZT_STATE_OBJECT_IDENTITY_PUBLIC,idtmp,RR->publicIdentityStr.data(),(unsigned int)RR->publicIdentityStr.length());
|
||||
} else {
|
||||
n = stateObjectGet(tptr,ZT_STATE_OBJECT_IDENTITY_PUBLIC,idtmp,tmp,sizeof(tmp) - 1);
|
||||
if (n > 0) {
|
||||
tmp[n] = (char)0;
|
||||
if (RR->identity.toString(false) != tmp)
|
||||
if (RR->publicIdentityStr != tmp)
|
||||
n = -1;
|
||||
}
|
||||
if (n <= 0) {
|
||||
tmp2 = RR->identity.toString(false);
|
||||
stateObjectPut(tptr,ZT_STATE_OBJECT_IDENTITY_PUBLIC,idtmp,tmp2.data(),(unsigned int)tmp2.length());
|
||||
}
|
||||
if (n <= 0)
|
||||
stateObjectPut(tptr,ZT_STATE_OBJECT_IDENTITY_PUBLIC,idtmp,RR->publicIdentityStr.data(),(unsigned int)RR->publicIdentityStr.length());
|
||||
}
|
||||
|
||||
try {
|
||||
|
|
@ -125,24 +127,20 @@ Node::Node(void *uptr,void *tptr,const struct ZT_Node_Callbacks *callbacks,uint6
|
|||
|
||||
Node::~Node()
|
||||
{
|
||||
Mutex::Lock _l(_networks_m);
|
||||
|
||||
_networks.clear(); // destroy all networks before shutdown
|
||||
|
||||
{
|
||||
Mutex::Lock _l(_networks_m);
|
||||
_networks.clear(); // destroy all networks before shutdown
|
||||
}
|
||||
delete RR->sa;
|
||||
delete RR->topology;
|
||||
delete RR->mc;
|
||||
delete RR->sw;
|
||||
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
delete RR->cluster;
|
||||
#endif
|
||||
}
|
||||
|
||||
ZT_ResultCode Node::processStateUpdate(
|
||||
void *tptr,
|
||||
ZT_StateObjectType type,
|
||||
uint64_t id,
|
||||
const uint64_t id[2],
|
||||
const void *data,
|
||||
unsigned int len)
|
||||
{
|
||||
|
|
@ -151,11 +149,12 @@ ZT_ResultCode Node::processStateUpdate(
|
|||
|
||||
case ZT_STATE_OBJECT_PEER_STATE:
|
||||
if (len) {
|
||||
}
|
||||
break;
|
||||
|
||||
case ZT_STATE_OBJECT_PEER_IDENTITY:
|
||||
if (len) {
|
||||
const SharedPtr<Peer> p(RR->topology->getPeer(tptr,Address(id[0])));
|
||||
if (p) {
|
||||
r = (p->applyStateUpdate(data,len)) ? ZT_RESULT_OK : ZT_RESULT_OK_IGNORED;
|
||||
} else {
|
||||
r = (Peer::createFromStateUpdate(RR,tptr,data,len)) ? ZT_RESULT_OK : ZT_RESULT_OK_IGNORED;
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
|
|
@ -163,9 +162,9 @@ ZT_ResultCode Node::processStateUpdate(
|
|||
if (len <= (ZT_NETWORKCONFIG_DICT_CAPACITY - 1)) {
|
||||
if (len < 2) {
|
||||
Mutex::Lock _l(_networks_m);
|
||||
SharedPtr<Network> &nw = _networks[id];
|
||||
SharedPtr<Network> &nw = _networks[id[0]];
|
||||
if (!nw) {
|
||||
nw = SharedPtr<Network>(new Network(RR,tptr,id,(void *)0,(const NetworkConfig *)0));
|
||||
nw = SharedPtr<Network>(new Network(RR,tptr,id[0],(void *)0,(const NetworkConfig *)0));
|
||||
r = ZT_RESULT_OK;
|
||||
}
|
||||
} else {
|
||||
|
|
@ -175,7 +174,7 @@ ZT_ResultCode Node::processStateUpdate(
|
|||
try {
|
||||
if (nconf->fromDictionary(*dict)) {
|
||||
Mutex::Lock _l(_networks_m);
|
||||
SharedPtr<Network> &nw = _networks[id];
|
||||
SharedPtr<Network> &nw = _networks[id[0]];
|
||||
if (nw) {
|
||||
switch (nw->setConfiguration(tptr,*nconf,false)) {
|
||||
default:
|
||||
|
|
@ -189,7 +188,7 @@ ZT_ResultCode Node::processStateUpdate(
|
|||
break;
|
||||
}
|
||||
} else {
|
||||
nw = SharedPtr<Network>(new Network(RR,tptr,id,(void *)0,nconf));
|
||||
nw = SharedPtr<Network>(new Network(RR,tptr,id[0],(void *)0,nconf));
|
||||
}
|
||||
} else {
|
||||
r = ZT_RESULT_ERROR_BAD_PARAMETER;
|
||||
|
|
@ -208,9 +207,14 @@ ZT_ResultCode Node::processStateUpdate(
|
|||
}
|
||||
break;
|
||||
|
||||
case ZT_STATE_OBJECT_NETWORK_MEMBERSHIP:
|
||||
if (len) {
|
||||
}
|
||||
break;
|
||||
|
||||
case ZT_STATE_OBJECT_PLANET:
|
||||
case ZT_STATE_OBJECT_MOON:
|
||||
if (len <= ZT_WORLD_MAX_SERIALIZED_LENGTH) {
|
||||
if ((len)&&(len <= ZT_WORLD_MAX_SERIALIZED_LENGTH)) {
|
||||
World w;
|
||||
try {
|
||||
w.deserialize(Buffer<ZT_WORLD_MAX_SERIALIZED_LENGTH>(data,len));
|
||||
|
|
@ -395,18 +399,7 @@ ZT_ResultCode Node::processBackgroundTasks(void *tptr,uint64_t now,volatile uint
|
|||
}
|
||||
|
||||
try {
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
// If clustering is enabled we have to call cluster->doPeriodicTasks() very often, so we override normal timer deadline behavior
|
||||
if (RR->cluster) {
|
||||
RR->sw->doTimerTasks(tptr,now);
|
||||
RR->cluster->doPeriodicTasks();
|
||||
*nextBackgroundTaskDeadline = now + ZT_CLUSTER_PERIODIC_TASK_PERIOD; // this is really short so just tick at this rate
|
||||
} else {
|
||||
#endif
|
||||
*nextBackgroundTaskDeadline = now + (uint64_t)std::max(std::min(timeUntilNextPingCheck,RR->sw->doTimerTasks(tptr,now)),(unsigned long)ZT_CORE_TIMER_TASK_GRANULARITY);
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
}
|
||||
#endif
|
||||
*nextBackgroundTaskDeadline = now + (uint64_t)std::max(std::min(timeUntilNextPingCheck,RR->sw->doTimerTasks(tptr,now)),(unsigned long)ZT_CORE_TIMER_TASK_GRANULARITY);
|
||||
} catch ( ... ) {
|
||||
return ZT_RESULT_FATAL_ERROR_INTERNAL;
|
||||
}
|
||||
|
|
@ -620,76 +613,6 @@ void Node::setNetconfMaster(void *networkControllerInstance)
|
|||
RR->localNetworkController->init(RR->identity,this);
|
||||
}
|
||||
|
||||
/*
|
||||
ZT_ResultCode Node::clusterInit(
|
||||
unsigned int myId,
|
||||
const struct sockaddr_storage *zeroTierPhysicalEndpoints,
|
||||
unsigned int numZeroTierPhysicalEndpoints,
|
||||
int x,
|
||||
int y,
|
||||
int z,
|
||||
void (*sendFunction)(void *,unsigned int,const void *,unsigned int),
|
||||
void *sendFunctionArg,
|
||||
int (*addressToLocationFunction)(void *,const struct sockaddr_storage *,int *,int *,int *),
|
||||
void *addressToLocationFunctionArg)
|
||||
{
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (RR->cluster)
|
||||
return ZT_RESULT_ERROR_BAD_PARAMETER;
|
||||
|
||||
std::vector<InetAddress> eps;
|
||||
for(unsigned int i=0;i<numZeroTierPhysicalEndpoints;++i)
|
||||
eps.push_back(InetAddress(zeroTierPhysicalEndpoints[i]));
|
||||
std::sort(eps.begin(),eps.end());
|
||||
RR->cluster = new Cluster(RR,myId,eps,x,y,z,sendFunction,sendFunctionArg,addressToLocationFunction,addressToLocationFunctionArg);
|
||||
|
||||
return ZT_RESULT_OK;
|
||||
#else
|
||||
return ZT_RESULT_ERROR_UNSUPPORTED_OPERATION;
|
||||
#endif
|
||||
}
|
||||
|
||||
ZT_ResultCode Node::clusterAddMember(unsigned int memberId)
|
||||
{
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (!RR->cluster)
|
||||
return ZT_RESULT_ERROR_BAD_PARAMETER;
|
||||
RR->cluster->addMember((uint16_t)memberId);
|
||||
return ZT_RESULT_OK;
|
||||
#else
|
||||
return ZT_RESULT_ERROR_UNSUPPORTED_OPERATION;
|
||||
#endif
|
||||
}
|
||||
|
||||
void Node::clusterRemoveMember(unsigned int memberId)
|
||||
{
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (RR->cluster)
|
||||
RR->cluster->removeMember((uint16_t)memberId);
|
||||
#endif
|
||||
}
|
||||
|
||||
void Node::clusterHandleIncomingMessage(const void *msg,unsigned int len)
|
||||
{
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (RR->cluster)
|
||||
RR->cluster->handleIncomingStateMessage(msg,len);
|
||||
#endif
|
||||
}
|
||||
|
||||
void Node::clusterStatus(ZT_ClusterStatus *cs)
|
||||
{
|
||||
if (!cs)
|
||||
return;
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (RR->cluster)
|
||||
RR->cluster->status(*cs);
|
||||
else
|
||||
#endif
|
||||
memset(cs,0,sizeof(ZT_ClusterStatus));
|
||||
}
|
||||
*/
|
||||
|
||||
/****************************************************************************/
|
||||
/* Node methods used only within node/ */
|
||||
/****************************************************************************/
|
||||
|
|
@ -918,7 +841,7 @@ enum ZT_ResultCode ZT_Node_processStateUpdate(
|
|||
ZT_Node *node,
|
||||
void *tptr,
|
||||
ZT_StateObjectType type,
|
||||
uint64_t id,
|
||||
const uint64_t id[2],
|
||||
const void *data,
|
||||
unsigned int len)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -85,7 +85,7 @@ public:
|
|||
ZT_ResultCode processStateUpdate(
|
||||
void *tptr,
|
||||
ZT_StateObjectType type,
|
||||
uint64_t id,
|
||||
const uint64_t id[2],
|
||||
const void *data,
|
||||
unsigned int len);
|
||||
ZT_ResultCode processWirePacket(
|
||||
|
|
|
|||
|
|
@ -46,11 +46,6 @@
|
|||
*/
|
||||
#define ZT_PATH_MAX_PREFERENCE_RANK ((ZT_INETADDRESS_MAX_SCOPE << 1) | 1)
|
||||
|
||||
/**
|
||||
* Maximum distance for a path
|
||||
*/
|
||||
#define ZT_PATH_DISTANCE_MAX 0xffff
|
||||
|
||||
namespace ZeroTier {
|
||||
|
||||
class RuntimeEnvironment;
|
||||
|
|
@ -125,7 +120,6 @@ public:
|
|||
_incomingLinkQualitySlowLogCounter(-64), // discard first fast log
|
||||
_incomingLinkQualityPreviousPacketCounter(0),
|
||||
_outgoingPacketCounter(0),
|
||||
_distance(ZT_PATH_DISTANCE_MAX),
|
||||
_addr(),
|
||||
_localAddress(),
|
||||
_ipScope(InetAddress::IP_SCOPE_NONE)
|
||||
|
|
@ -143,7 +137,6 @@ public:
|
|||
_incomingLinkQualitySlowLogCounter(-64), // discard first fast log
|
||||
_incomingLinkQualityPreviousPacketCounter(0),
|
||||
_outgoingPacketCounter(0),
|
||||
_distance(ZT_PATH_DISTANCE_MAX),
|
||||
_addr(addr),
|
||||
_localAddress(localAddress),
|
||||
_ipScope(addr.ipScope())
|
||||
|
|
@ -311,11 +304,6 @@ public:
|
|||
*/
|
||||
inline uint64_t lastTrustEstablishedPacketReceived() const { return _lastTrustEstablishedPacketReceived; }
|
||||
|
||||
/**
|
||||
* @return Distance (higher is further)
|
||||
*/
|
||||
inline unsigned int distance() const { return _distance; }
|
||||
|
||||
/**
|
||||
* @param lo Last out send
|
||||
* @param li Last in send
|
||||
|
|
@ -344,7 +332,6 @@ private:
|
|||
volatile signed int _incomingLinkQualitySlowLogCounter;
|
||||
volatile unsigned int _incomingLinkQualityPreviousPacketCounter;
|
||||
volatile unsigned int _outgoingPacketCounter;
|
||||
volatile unsigned int _distance;
|
||||
InetAddress _addr;
|
||||
InetAddress _localAddress;
|
||||
InetAddress::IpScope _ipScope; // memoize this since it's a computed value checked often
|
||||
|
|
|
|||
109
node/Peer.cpp
109
node/Peer.cpp
|
|
@ -146,8 +146,8 @@ void Peer::received(
|
|||
path->updateLinkQuality((unsigned int)(packetId & 7));
|
||||
|
||||
if (hops == 0) {
|
||||
// If this is a direct packet (no hops), update existing paths or learn new ones
|
||||
bool pathAlreadyKnown = false;
|
||||
bool newPathLearned = false;
|
||||
|
||||
{
|
||||
Mutex::Lock _l(_paths_m);
|
||||
|
|
@ -188,7 +188,7 @@ void Peer::received(
|
|||
if (verb == Packet::VERB_OK) {
|
||||
potentialNewPeerPath->lr = now;
|
||||
potentialNewPeerPath->p = path;
|
||||
newPathLearned = true;
|
||||
_lastWroteState = 0; // force state write now
|
||||
} else {
|
||||
TRACE("got %s via unknown path %s(%s), confirming...",Packet::verbString(verb),_id.address().toString().c_str(),path->address().toString().c_str());
|
||||
attemptToContactAt(tPtr,path->localAddress(),path->address(),now,true,path->nextOutgoingCounter());
|
||||
|
|
@ -196,9 +196,6 @@ void Peer::received(
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (newPathLearned)
|
||||
writeState(tPtr,now);
|
||||
} else if (this->trustEstablished(now)) {
|
||||
// Send PUSH_DIRECT_PATHS if hops>0 (relayed) and we have a trust relationship (common network membership)
|
||||
if ((now - _lastDirectPathPushSent) >= ZT_DIRECT_PATH_PUSH_INTERVAL) {
|
||||
|
|
@ -270,6 +267,9 @@ void Peer::received(
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
if ((now - _lastWroteState) > ZT_PEER_STATE_WRITE_PERIOD)
|
||||
writeState(tPtr,now);
|
||||
}
|
||||
|
||||
bool Peer::sendDirect(void *tPtr,const void *data,unsigned int len,uint64_t now,bool force)
|
||||
|
|
@ -435,7 +435,7 @@ bool Peer::doPingAndKeepalive(void *tPtr,uint64_t now,int inetAddressFamily)
|
|||
void Peer::writeState(void *tPtr,const uint64_t now)
|
||||
{
|
||||
try {
|
||||
Buffer<sizeof(Peer) + 32 + (sizeof(Path) * 2)> b;
|
||||
Buffer<ZT_PEER_MAX_SERIALIZED_STATE_SIZE> b;
|
||||
|
||||
b.append((uint8_t)1); // version
|
||||
b.append(now);
|
||||
|
|
@ -455,7 +455,6 @@ void Peer::writeState(void *tPtr,const uint64_t now)
|
|||
b.append(_v4Path.p->lastOut());
|
||||
b.append(_v4Path.p->lastIn());
|
||||
b.append(_v4Path.p->lastTrustEstablishedPacketReceived());
|
||||
b.append((uint16_t)_v4Path.p->distance());
|
||||
_v4Path.p->address().serialize(b);
|
||||
_v4Path.p->localAddress().serialize(b);
|
||||
}
|
||||
|
|
@ -464,29 +463,29 @@ void Peer::writeState(void *tPtr,const uint64_t now)
|
|||
b.append(_v6Path.p->lastOut());
|
||||
b.append(_v6Path.p->lastIn());
|
||||
b.append(_v6Path.p->lastTrustEstablishedPacketReceived());
|
||||
b.append((uint16_t)_v6Path.p->distance());
|
||||
_v6Path.p->address().serialize(b);
|
||||
_v6Path.p->localAddress().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);
|
||||
// Save space by sending these as time since now at 100ms resolution
|
||||
b.append((uint16_t)(std::max(now - _lastReceive,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastNontrivialReceive,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastTriedMemorizedPath,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastDirectPathPushSent,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastDirectPathPushReceive,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastCredentialRequestSent,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastWhoisRequestReceived,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastEchoRequestReceived,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastComRequestReceived,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastComRequestSent,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastCredentialsReceived,(uint64_t)6553500) / 100));
|
||||
b.append((uint16_t)(std::max(now - _lastTrustEstablishedPacketReceived,(uint64_t)6553500) / 100));
|
||||
|
||||
b.append(_vProto);
|
||||
b.append(_vMajor);
|
||||
b.append(_vMinor);
|
||||
b.append(_vRevision);
|
||||
b.append((uint8_t)_vProto);
|
||||
b.append((uint8_t)_vMajor);
|
||||
b.append((uint8_t)_vMinor);
|
||||
b.append((uint16_t)_vRevision);
|
||||
|
||||
b.append((uint16_t)0); // length of additional fields
|
||||
|
||||
|
|
@ -501,7 +500,7 @@ void Peer::writeState(void *tPtr,const uint64_t now)
|
|||
bool Peer::applyStateUpdate(const void *data,unsigned int len)
|
||||
{
|
||||
try {
|
||||
Buffer<sizeof(Peer) + 32 + (sizeof(Path) * 2)> b(data,len);
|
||||
Buffer<ZT_PEER_MAX_SERIALIZED_STATE_SIZE> b(data,len);
|
||||
unsigned int ptr = 0;
|
||||
|
||||
if (b[ptr++] != 1)
|
||||
|
|
@ -510,6 +509,11 @@ bool Peer::applyStateUpdate(const void *data,unsigned int len)
|
|||
if (ts <= _lastReceivedStateTimestamp)
|
||||
return false;
|
||||
|
||||
Identity id;
|
||||
ptr += id.deserialize(b,ptr);
|
||||
if (id != _id) // sanity check
|
||||
return false;
|
||||
|
||||
const unsigned int pathCount = (unsigned int)b[ptr++];
|
||||
{
|
||||
Mutex::Lock _l(_paths_m);
|
||||
|
|
@ -518,7 +522,6 @@ bool Peer::applyStateUpdate(const void *data,unsigned int len)
|
|||
const uint64_t lastOut = b.at<uint64_t>(ptr); ptr += 8;
|
||||
const uint64_t lastIn = b.at<uint64_t>(ptr); ptr += 8;
|
||||
const uint64_t lastTrustEstablishedPacketReceived = b.at<uint64_t>(ptr); ptr += 8;
|
||||
const unsigned int distance = b.at<uint16_t>(ptr); ptr += 2;
|
||||
InetAddress addr,localAddr;
|
||||
ptr += addr.deserialize(b,ptr);
|
||||
ptr += localAddr.deserialize(b,ptr);
|
||||
|
|
@ -529,8 +532,9 @@ bool Peer::applyStateUpdate(const void *data,unsigned int len)
|
|||
case AF_INET6: p = &_v6Path; break;
|
||||
}
|
||||
if (p) {
|
||||
if ( ((p->p->address() != addr)||(p->p->localAddress() != localAddr)) && (p->p->distance() > distance) )
|
||||
if ( (!p->p) || ((p->p->address() != addr)||(p->p->localAddress() != localAddr)) ) {
|
||||
p->p = RR->topology->getPath(localAddr,addr);
|
||||
}
|
||||
p->lr = lr;
|
||||
p->p->updateFromRemoteState(lastOut,lastIn,lastTrustEstablishedPacketReceived);
|
||||
}
|
||||
|
|
@ -538,22 +542,22 @@ bool Peer::applyStateUpdate(const void *data,unsigned int len)
|
|||
}
|
||||
}
|
||||
|
||||
_lastReceive = std::max(_lastReceive,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastNontrivialReceive = std::max(_lastNontrivialReceive,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastTriedMemorizedPath = std::max(_lastTriedMemorizedPath,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastDirectPathPushSent = std::max(_lastDirectPathPushSent,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastDirectPathPushReceive = std::max(_lastDirectPathPushReceive,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastCredentialRequestSent = std::max(_lastCredentialRequestSent,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastWhoisRequestReceived = std::max(_lastWhoisRequestReceived,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastEchoRequestReceived = std::max(_lastEchoRequestReceived,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastComRequestReceived = std::max(_lastComRequestReceived,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastComRequestSent = std::max(_lastComRequestSent,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastCredentialsReceived = std::max(_lastCredentialsReceived,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastTrustEstablishedPacketReceived = std::max(_lastTrustEstablishedPacketReceived,b.at<uint64_t>(ptr)); ptr += 8;
|
||||
_lastReceive = std::max(_lastReceive,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastNontrivialReceive = std::max(_lastNontrivialReceive,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastTriedMemorizedPath = std::max(_lastTriedMemorizedPath,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastDirectPathPushSent = std::max(_lastDirectPathPushSent,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastDirectPathPushReceive = std::max(_lastDirectPathPushReceive,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastCredentialRequestSent = std::max(_lastCredentialRequestSent,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastWhoisRequestReceived = std::max(_lastWhoisRequestReceived,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastEchoRequestReceived = std::max(_lastEchoRequestReceived,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastComRequestReceived = std::max(_lastComRequestReceived,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastComRequestSent = std::max(_lastComRequestSent,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastCredentialsReceived = std::max(_lastCredentialsReceived,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
_lastTrustEstablishedPacketReceived = std::max(_lastTrustEstablishedPacketReceived,ts - ((uint64_t)b.at<uint16_t>(ptr) * 100ULL)); ptr += 2;
|
||||
|
||||
_vProto = b.at<uint16_t>(ptr); ptr += 2;
|
||||
_vMajor = b.at<uint16_t>(ptr); ptr += 2;
|
||||
_vMinor = b.at<uint16_t>(ptr); ptr += 2;
|
||||
_vProto = (uint16_t)b[ptr++];
|
||||
_vMajor = (uint16_t)b[ptr++];
|
||||
_vMinor = (uint16_t)b[ptr++];
|
||||
_vRevision = b.at<uint16_t>(ptr); ptr += 2;
|
||||
|
||||
_lastReceivedStateTimestamp = ts;
|
||||
|
|
@ -563,4 +567,25 @@ bool Peer::applyStateUpdate(const void *data,unsigned int len)
|
|||
return false;
|
||||
}
|
||||
|
||||
SharedPtr<Peer> Peer::createFromStateUpdate(const RuntimeEnvironment *renv,void *tPtr,const void *data,unsigned int len)
|
||||
{
|
||||
try {
|
||||
Identity id;
|
||||
{
|
||||
Buffer<ZT_PEER_MAX_SERIALIZED_STATE_SIZE> b(data,len);
|
||||
unsigned int ptr = 0;
|
||||
if (b[ptr++] != 1)
|
||||
return SharedPtr<Peer>();
|
||||
ptr += 8; // skip TS, don't care
|
||||
id.deserialize(b,ptr);
|
||||
}
|
||||
if (id) {
|
||||
const SharedPtr<Peer> p(new Peer(renv,renv->identity,id));
|
||||
if (p->applyStateUpdate(data,len))
|
||||
return renv->topology->addPeer(tPtr,p);
|
||||
}
|
||||
} catch ( ... ) {}
|
||||
return SharedPtr<Peer>();
|
||||
}
|
||||
|
||||
} // namespace ZeroTier
|
||||
|
|
|
|||
|
|
@ -51,6 +51,8 @@
|
|||
#include "Mutex.hpp"
|
||||
#include "NonCopyable.hpp"
|
||||
|
||||
#define ZT_PEER_MAX_SERIALIZED_STATE_SIZE (sizeof(Peer) + 32 + (sizeof(Path) * 2))
|
||||
|
||||
namespace ZeroTier {
|
||||
|
||||
/**
|
||||
|
|
@ -194,9 +196,10 @@ public:
|
|||
bool doPingAndKeepalive(void *tPtr,uint64_t now,int inetAddressFamily);
|
||||
|
||||
/**
|
||||
* Write current peer state to external storage / cluster network
|
||||
* Write object state to external storage and/or cluster network
|
||||
*
|
||||
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
|
||||
* @param now Current time
|
||||
*/
|
||||
void writeState(void *tPtr,const uint64_t now);
|
||||
|
||||
|
|
@ -437,6 +440,17 @@ public:
|
|||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a peer from a remote state update
|
||||
*
|
||||
* @param renv Runtime environment
|
||||
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
|
||||
* @param data State update data
|
||||
* @param len State update length
|
||||
* @return Peer or NULL if data was invalid
|
||||
*/
|
||||
static SharedPtr<Peer> createFromStateUpdate(const RuntimeEnvironment *renv,void *tPtr,const void *data,unsigned int len);
|
||||
|
||||
private:
|
||||
struct _PeerPath
|
||||
{
|
||||
|
|
|
|||
|
|
@ -30,8 +30,8 @@
|
|||
#include <string>
|
||||
|
||||
#include "Constants.hpp"
|
||||
#include "Utils.hpp"
|
||||
#include "Identity.hpp"
|
||||
#include "Mutex.hpp"
|
||||
|
||||
namespace ZeroTier {
|
||||
|
||||
|
|
@ -58,10 +58,13 @@ public:
|
|||
,mc((Multicaster *)0)
|
||||
,topology((Topology *)0)
|
||||
,sa((SelfAwareness *)0)
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
,cluster((Cluster *)0)
|
||||
#endif
|
||||
{
|
||||
Utils::getSecureRandom(&instanceId,sizeof(instanceId));
|
||||
}
|
||||
|
||||
~RuntimeEnvironment()
|
||||
{
|
||||
Utils::burn(reinterpret_cast<void *>(const_cast<char *>(secretIdentityStr.data())),(unsigned int)secretIdentityStr.length());
|
||||
}
|
||||
|
||||
// Node instance that owns this RuntimeEnvironment
|
||||
|
|
@ -87,9 +90,11 @@ public:
|
|||
Multicaster *mc;
|
||||
Topology *topology;
|
||||
SelfAwareness *sa;
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
Cluster *cluster;
|
||||
#endif
|
||||
|
||||
/**
|
||||
* A random integer identifying this run of ZeroTier
|
||||
*/
|
||||
uint32_t instanceId;
|
||||
};
|
||||
|
||||
} // namespace ZeroTier
|
||||
|
|
|
|||
|
|
@ -108,13 +108,7 @@ void Switch::onRemotePacket(void *tPtr,const InetAddress &localAddr,const InetAd
|
|||
const Address destination(fragment.destination());
|
||||
|
||||
if (destination != RR->identity.address()) {
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
const bool isClusterFrontplane = ((RR->cluster)&&(RR->cluster->isClusterPeerFrontplane(fromAddr)));
|
||||
#else
|
||||
const bool isClusterFrontplane = false;
|
||||
#endif
|
||||
|
||||
if ( (!RR->topology->amRoot()) && (!path->trustEstablished(now)) && (!isClusterFrontplane) )
|
||||
if ( (!RR->topology->amRoot()) && (!path->trustEstablished(now)) )
|
||||
return;
|
||||
|
||||
if (fragment.hops() < ZT_RELAY_MAX_HOPS) {
|
||||
|
|
@ -124,13 +118,6 @@ void Switch::onRemotePacket(void *tPtr,const InetAddress &localAddr,const InetAd
|
|||
// It wouldn't hurt anything, just redundant and unnecessary.
|
||||
SharedPtr<Peer> relayTo = RR->topology->getPeer(tPtr,destination);
|
||||
if ((!relayTo)||(!relayTo->sendDirect(tPtr,fragment.data(),fragment.size(),now,false))) {
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ((RR->cluster)&&(!isClusterFrontplane)) {
|
||||
RR->cluster->relayViaCluster(Address(),destination,fragment.data(),fragment.size(),false);
|
||||
return;
|
||||
}
|
||||
#endif
|
||||
|
||||
// Don't know peer or no direct path -- so relay via someone upstream
|
||||
relayTo = RR->topology->getUpstreamPeer();
|
||||
if (relayTo)
|
||||
|
|
@ -197,13 +184,8 @@ void Switch::onRemotePacket(void *tPtr,const InetAddress &localAddr,const InetAd
|
|||
|
||||
//TRACE("<< %.16llx %s -> %s (size: %u)",(unsigned long long)packet->packetId(),source.toString().c_str(),destination.toString().c_str(),packet->size());
|
||||
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ( (source == RR->identity.address()) && ((!RR->cluster)||(!RR->cluster->isClusterPeerFrontplane(fromAddr))) )
|
||||
return;
|
||||
#else
|
||||
if (source == RR->identity.address())
|
||||
return;
|
||||
#endif
|
||||
|
||||
if (destination != RR->identity.address()) {
|
||||
if ( (!RR->topology->amRoot()) && (!path->trustEstablished(now)) && (source != RR->identity.address()) )
|
||||
|
|
@ -212,12 +194,7 @@ void Switch::onRemotePacket(void *tPtr,const InetAddress &localAddr,const InetAd
|
|||
Packet packet(data,len);
|
||||
|
||||
if (packet.hops() < ZT_RELAY_MAX_HOPS) {
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (source != RR->identity.address()) // don't increment hops for cluster frontplane relays
|
||||
packet.incrementHops();
|
||||
#else
|
||||
packet.incrementHops();
|
||||
#endif
|
||||
|
||||
SharedPtr<Peer> relayTo = RR->topology->getPeer(tPtr,destination);
|
||||
if ((relayTo)&&(relayTo->sendDirect(tPtr,packet.data(),packet.size(),now,false))) {
|
||||
|
|
@ -277,12 +254,6 @@ void Switch::onRemotePacket(void *tPtr,const InetAddress &localAddr,const InetAd
|
|||
}
|
||||
}
|
||||
} else {
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ((RR->cluster)&&(source != RR->identity.address())) {
|
||||
RR->cluster->relayViaCluster(source,destination,packet.data(),packet.size(),_shouldUnite(now,source,destination));
|
||||
return;
|
||||
}
|
||||
#endif
|
||||
relayTo = RR->topology->getUpstreamPeer(&source,1,true);
|
||||
if (relayTo)
|
||||
relayTo->sendDirect(tPtr,packet.data(),packet.size(),now,true);
|
||||
|
|
@ -769,14 +740,6 @@ bool Switch::_trySend(void *tPtr,Packet &packet,bool encrypt)
|
|||
const uint64_t now = RR->node->now();
|
||||
const Address destination(packet.destination());
|
||||
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
uint64_t clusterMostRecentTs = 0;
|
||||
int clusterMostRecentMemberId = -1;
|
||||
uint8_t clusterPeerSecret[ZT_PEER_SECRET_KEY_LENGTH];
|
||||
if (RR->cluster)
|
||||
clusterMostRecentMemberId = RR->cluster->checkSendViaCluster(destination,clusterMostRecentTs,clusterPeerSecret);
|
||||
#endif
|
||||
|
||||
const SharedPtr<Peer> peer(RR->topology->getPeer(tPtr,destination));
|
||||
if (peer) {
|
||||
/* First get the best path, and if it's dead (and this is not a root)
|
||||
|
|
@ -788,74 +751,37 @@ bool Switch::_trySend(void *tPtr,Packet &packet,bool encrypt)
|
|||
|
||||
viaPath = peer->getBestPath(now,false);
|
||||
if ( (viaPath) && (!viaPath->alive(now)) && (!RR->topology->isUpstream(peer->identity())) ) {
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ((clusterMostRecentMemberId < 0)||(viaPath->lastIn() > clusterMostRecentTs)) {
|
||||
#endif
|
||||
if ((now - viaPath->lastOut()) > std::max((now - viaPath->lastIn()) * 4,(uint64_t)ZT_PATH_MIN_REACTIVATE_INTERVAL)) {
|
||||
peer->attemptToContactAt(tPtr,viaPath->localAddress(),viaPath->address(),now,false,viaPath->nextOutgoingCounter());
|
||||
viaPath->sent(now);
|
||||
}
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ((now - viaPath->lastOut()) > std::max((now - viaPath->lastIn()) * 4,(uint64_t)ZT_PATH_MIN_REACTIVATE_INTERVAL)) {
|
||||
peer->attemptToContactAt(tPtr,viaPath->localAddress(),viaPath->address(),now,false,viaPath->nextOutgoingCounter());
|
||||
viaPath->sent(now);
|
||||
}
|
||||
#endif
|
||||
viaPath.zero();
|
||||
}
|
||||
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (clusterMostRecentMemberId >= 0) {
|
||||
if ((viaPath)&&(viaPath->lastIn() < clusterMostRecentTs))
|
||||
viaPath.zero();
|
||||
} else if (!viaPath) {
|
||||
#else
|
||||
if (!viaPath) {
|
||||
#endif
|
||||
peer->tryMemorizedPath(tPtr,now); // periodically attempt memorized or statically defined paths, if any are known
|
||||
const SharedPtr<Peer> relay(RR->topology->getUpstreamPeer());
|
||||
if ( (!relay) || (!(viaPath = relay->getBestPath(now,false))) ) {
|
||||
if (!(viaPath = peer->getBestPath(now,true)))
|
||||
return false;
|
||||
}
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
}
|
||||
#else
|
||||
}
|
||||
#endif
|
||||
} else {
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (clusterMostRecentMemberId < 0) {
|
||||
#else
|
||||
requestWhois(tPtr,destination);
|
||||
return false; // if we are not in cluster mode, there is no way we can send without knowing the peer directly
|
||||
#endif
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
}
|
||||
#endif
|
||||
requestWhois(tPtr,destination);
|
||||
return false; // if we are not in cluster mode, there is no way we can send without knowing the peer directly
|
||||
}
|
||||
|
||||
unsigned int chunkSize = std::min(packet.size(),(unsigned int)ZT_UDP_DEFAULT_PAYLOAD_MTU);
|
||||
packet.setFragmented(chunkSize < packet.size());
|
||||
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
const uint64_t trustedPathId = (viaPath) ? RR->topology->getOutboundPathTrust(viaPath->address()) : 0;
|
||||
if (trustedPathId) {
|
||||
packet.setTrusted(trustedPathId);
|
||||
} else {
|
||||
packet.armor((clusterMostRecentMemberId >= 0) ? clusterPeerSecret : peer->key(),encrypt,(viaPath) ? viaPath->nextOutgoingCounter() : 0);
|
||||
}
|
||||
#else
|
||||
const uint64_t trustedPathId = RR->topology->getOutboundPathTrust(viaPath->address());
|
||||
if (trustedPathId) {
|
||||
packet.setTrusted(trustedPathId);
|
||||
} else {
|
||||
packet.armor(peer->key(),encrypt,viaPath->nextOutgoingCounter());
|
||||
}
|
||||
#endif
|
||||
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if ( ((viaPath)&&(viaPath->send(RR,tPtr,packet.data(),chunkSize,now))) || ((clusterMostRecentMemberId >= 0)&&(RR->cluster->sendViaCluster(clusterMostRecentMemberId,destination,packet.data(),chunkSize))) ) {
|
||||
#else
|
||||
if (viaPath->send(RR,tPtr,packet.data(),chunkSize,now)) {
|
||||
#endif
|
||||
if (chunkSize < packet.size()) {
|
||||
// Too big for one packet, fragment the rest
|
||||
unsigned int fragStart = chunkSize;
|
||||
|
|
@ -868,14 +794,7 @@ bool Switch::_trySend(void *tPtr,Packet &packet,bool encrypt)
|
|||
for(unsigned int fno=1;fno<totalFragments;++fno) {
|
||||
chunkSize = std::min(remaining,(unsigned int)(ZT_UDP_DEFAULT_PAYLOAD_MTU - ZT_PROTO_MIN_FRAGMENT_LENGTH));
|
||||
Packet::Fragment frag(packet,fragStart,chunkSize,fno,totalFragments);
|
||||
#ifdef ZT_ENABLE_CLUSTER
|
||||
if (viaPath)
|
||||
viaPath->send(RR,tPtr,frag.data(),frag.size(),now);
|
||||
else if (clusterMostRecentMemberId >= 0)
|
||||
RR->cluster->sendViaCluster(clusterMostRecentMemberId,destination,frag.data(),frag.size());
|
||||
#else
|
||||
viaPath->send(RR,tPtr,frag.data(),frag.size(),now);
|
||||
#endif
|
||||
fragStart += chunkSize;
|
||||
remaining -= chunkSize;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -108,8 +108,6 @@ SharedPtr<Peer> Topology::addPeer(void *tPtr,const SharedPtr<Peer> &peer)
|
|||
np = hp;
|
||||
}
|
||||
|
||||
saveIdentity(tPtr,np->identity());
|
||||
|
||||
return np;
|
||||
}
|
||||
|
||||
|
|
@ -128,18 +126,20 @@ SharedPtr<Peer> Topology::getPeer(void *tPtr,const Address &zta)
|
|||
}
|
||||
|
||||
try {
|
||||
Identity id(_getIdentity(tPtr,zta));
|
||||
if (id) {
|
||||
SharedPtr<Peer> np(new Peer(RR,RR->identity,id));
|
||||
{
|
||||
Mutex::Lock _l(_peers_m);
|
||||
SharedPtr<Peer> &ap = _peers[zta];
|
||||
if (!ap)
|
||||
ap.swap(np);
|
||||
char buf[ZT_PEER_MAX_SERIALIZED_STATE_SIZE];
|
||||
uint64_t idbuf[2]; idbuf[0] = zta.toInt(); idbuf[1] = 0;
|
||||
int len = RR->node->stateObjectGet(tPtr,ZT_STATE_OBJECT_PEER_STATE,idbuf,buf,(unsigned int)sizeof(buf));
|
||||
if (len > 0) {
|
||||
Mutex::Lock _l(_peers_m);
|
||||
SharedPtr<Peer> &ap = _peers[zta];
|
||||
if (ap)
|
||||
return ap;
|
||||
}
|
||||
ap = Peer::createFromStateUpdate(RR,tPtr,buf,len);
|
||||
if (!ap)
|
||||
_peers.erase(zta);
|
||||
return ap;
|
||||
}
|
||||
} catch ( ... ) {} // invalid identity on disk?
|
||||
} catch ( ... ) {} // ignore invalid identities or other strage failures
|
||||
|
||||
return SharedPtr<Peer>();
|
||||
}
|
||||
|
|
@ -154,17 +154,7 @@ Identity Topology::getIdentity(void *tPtr,const Address &zta)
|
|||
if (ap)
|
||||
return (*ap)->identity();
|
||||
}
|
||||
return _getIdentity(tPtr,zta);
|
||||
}
|
||||
|
||||
void Topology::saveIdentity(void *tPtr,const Identity &id)
|
||||
{
|
||||
if (id) {
|
||||
const std::string tmp(id.toString(false));
|
||||
uint64_t idtmp[2];
|
||||
idtmp[0] = id.address().toInt(); idtmp[1] = 0;
|
||||
RR->node->stateObjectPut(tPtr,ZT_STATE_OBJECT_PEER_IDENTITY,idtmp,tmp.data(),(unsigned int)tmp.length());
|
||||
}
|
||||
return Identity();
|
||||
}
|
||||
|
||||
SharedPtr<Peer> Topology::getUpstreamPeer(const Address *avoid,unsigned int avoidCount,bool strictAvoid)
|
||||
|
|
@ -423,21 +413,6 @@ void Topology::doPeriodicTasks(void *tPtr,uint64_t now)
|
|||
}
|
||||
}
|
||||
|
||||
Identity Topology::_getIdentity(void *tPtr,const Address &zta)
|
||||
{
|
||||
char tmp[512];
|
||||
uint64_t idtmp[2];
|
||||
idtmp[0] = zta.toInt(); idtmp[1] = 0;
|
||||
int n = RR->node->stateObjectGet(tPtr,ZT_STATE_OBJECT_PEER_IDENTITY,idtmp,tmp,sizeof(tmp) - 1);
|
||||
if (n > 0) {
|
||||
tmp[n] = (char)0;
|
||||
try {
|
||||
return Identity(tmp);
|
||||
} catch ( ... ) {} // ignore invalid IDs
|
||||
}
|
||||
return Identity();
|
||||
}
|
||||
|
||||
void Topology::_memoizeUpstreams(void *tPtr)
|
||||
{
|
||||
// assumes _upstreams_m and _peers_m are locked
|
||||
|
|
@ -450,10 +425,8 @@ void Topology::_memoizeUpstreams(void *tPtr)
|
|||
} else if (std::find(_upstreamAddresses.begin(),_upstreamAddresses.end(),i->identity.address()) == _upstreamAddresses.end()) {
|
||||
_upstreamAddresses.push_back(i->identity.address());
|
||||
SharedPtr<Peer> &hp = _peers[i->identity.address()];
|
||||
if (!hp) {
|
||||
if (!hp)
|
||||
hp = new Peer(RR,RR->identity,i->identity);
|
||||
saveIdentity(tPtr,i->identity);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -464,10 +437,8 @@ void Topology::_memoizeUpstreams(void *tPtr)
|
|||
} else if (std::find(_upstreamAddresses.begin(),_upstreamAddresses.end(),i->identity.address()) == _upstreamAddresses.end()) {
|
||||
_upstreamAddresses.push_back(i->identity.address());
|
||||
SharedPtr<Peer> &hp = _peers[i->identity.address()];
|
||||
if (!hp) {
|
||||
if (!hp)
|
||||
hp = new Peer(RR,RR->identity,i->identity);
|
||||
saveIdentity(tPtr,i->identity);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -81,6 +81,13 @@ public:
|
|||
*/
|
||||
SharedPtr<Peer> getPeer(void *tPtr,const Address &zta);
|
||||
|
||||
/**
|
||||
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
|
||||
* @param zta ZeroTier address of peer
|
||||
* @return Identity or NULL identity if not found
|
||||
*/
|
||||
Identity getIdentity(void *tPtr,const Address &zta);
|
||||
|
||||
/**
|
||||
* Get a peer only if it is presently in memory (no disk cache)
|
||||
*
|
||||
|
|
@ -116,26 +123,6 @@ public:
|
|||
return p;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the identity of a peer
|
||||
*
|
||||
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
|
||||
* @param zta ZeroTier address of peer
|
||||
* @return Identity or NULL Identity if not found
|
||||
*/
|
||||
Identity getIdentity(void *tPtr,const Address &zta);
|
||||
|
||||
/**
|
||||
* Cache an identity
|
||||
*
|
||||
* This is done automatically on addPeer(), and so is only useful for
|
||||
* cluster identity replication.
|
||||
*
|
||||
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
|
||||
* @param id Identity to cache
|
||||
*/
|
||||
void saveIdentity(void *tPtr,const Identity &id);
|
||||
|
||||
/**
|
||||
* Get the current best upstream peer
|
||||
*
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue