Peer.cpp 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693
  1. /*
  2. * Copyright (c)2013-2020 ZeroTier, Inc.
  3. *
  4. * Use of this software is governed by the Business Source License included
  5. * in the LICENSE.TXT file in the project's root directory.
  6. *
  7. * Change Date: 2025-01-01
  8. *
  9. * On the date above, in accordance with the Business Source License, use
  10. * of this software will be governed by version 2.0 of the Apache License.
  11. */
  12. /****/
  13. #include "../version.h"
  14. #include "Constants.hpp"
  15. #include "Peer.hpp"
  16. #include "Switch.hpp"
  17. #include "Network.hpp"
  18. #include "SelfAwareness.hpp"
  19. #include "Packet.hpp"
  20. #include "Trace.hpp"
  21. #include "InetAddress.hpp"
  22. #include "RingBuffer.hpp"
  23. #include "Utils.hpp"
  24. #include "Metrics.hpp"
  25. namespace ZeroTier {
  26. static unsigned char s_freeRandomByteCounter = 0;
  27. Peer::Peer(const RuntimeEnvironment *renv,const Identity &myIdentity,const Identity &peerIdentity)
  28. : RR(renv)
  29. , _lastReceive(0)
  30. , _lastNontrivialReceive(0)
  31. , _lastTriedMemorizedPath(0)
  32. , _lastDirectPathPushSent(0)
  33. , _lastDirectPathPushReceive(0)
  34. , _lastCredentialRequestSent(0)
  35. , _lastWhoisRequestReceived(0)
  36. , _lastCredentialsReceived(0)
  37. , _lastTrustEstablishedPacketReceived(0)
  38. , _lastSentFullHello(0)
  39. , _lastEchoCheck(0)
  40. , _freeRandomByte((unsigned char)((uintptr_t)this >> 4) ^ ++s_freeRandomByteCounter)
  41. , _vProto(0)
  42. , _vMajor(0)
  43. , _vMinor(0)
  44. , _vRevision(0)
  45. , _id(peerIdentity)
  46. , _directPathPushCutoffCount(0)
  47. , _echoRequestCutoffCount(0)
  48. , _localMultipathSupported(false)
  49. , _lastComputedAggregateMeanLatency(0)
  50. #ifndef ZT_NO_PEER_METRICS
  51. , _peer_latency{Metrics::peer_latency.Add({{"node_id", OSUtils::nodeIDStr(peerIdentity.address().toInt())}}, std::vector<uint64_t>{1,3,6,10,30,60,100,300,600,1000})}
  52. , _alive_path_count{Metrics::peer_path_count.Add({{"node_id", OSUtils::nodeIDStr(peerIdentity.address().toInt())},{"status","alive"}})}
  53. , _dead_path_count{Metrics::peer_path_count.Add({{"node_id", OSUtils::nodeIDStr(peerIdentity.address().toInt())},{"status","dead"}})}
  54. , _incoming_packet{Metrics::peer_packets.Add({{"direction", "rx"},{"node_id", OSUtils::nodeIDStr(peerIdentity.address().toInt())}})}
  55. , _outgoing_packet{Metrics::peer_packets.Add({{"direction", "tx"},{"node_id", OSUtils::nodeIDStr(peerIdentity.address().toInt())}})}
  56. , _packet_errors{Metrics::peer_packet_errors.Add({{"node_id", OSUtils::nodeIDStr(peerIdentity.address().toInt())}})}
  57. #endif
  58. {
  59. if (!myIdentity.agree(peerIdentity,_key)) {
  60. throw ZT_EXCEPTION_INVALID_ARGUMENT;
  61. }
  62. uint8_t ktmp[ZT_SYMMETRIC_KEY_SIZE];
  63. KBKDFHMACSHA384(_key,ZT_KBKDF_LABEL_AES_GMAC_SIV_K0,0,0,ktmp);
  64. _aesKeys[0].init(ktmp);
  65. KBKDFHMACSHA384(_key,ZT_KBKDF_LABEL_AES_GMAC_SIV_K1,0,0,ktmp);
  66. _aesKeys[1].init(ktmp);
  67. Utils::burn(ktmp,ZT_SYMMETRIC_KEY_SIZE);
  68. }
  69. void Peer::received(
  70. void *tPtr,
  71. const SharedPtr<Path> &path,
  72. const unsigned int hops,
  73. const uint64_t packetId,
  74. const unsigned int payloadLength,
  75. const Packet::Verb verb,
  76. const uint64_t inRePacketId,
  77. const Packet::Verb inReVerb,
  78. const bool trustEstablished,
  79. const uint64_t networkId,
  80. const int32_t flowId)
  81. {
  82. const int64_t now = RR->node->now();
  83. _lastReceive = now;
  84. switch (verb) {
  85. case Packet::VERB_FRAME:
  86. case Packet::VERB_EXT_FRAME:
  87. case Packet::VERB_NETWORK_CONFIG_REQUEST:
  88. case Packet::VERB_NETWORK_CONFIG:
  89. case Packet::VERB_MULTICAST_FRAME:
  90. _lastNontrivialReceive = now;
  91. break;
  92. default:
  93. break;
  94. }
  95. #ifndef ZT_NO_PEER_METRICS
  96. _incoming_packet++;
  97. #endif
  98. recordIncomingPacket(path, packetId, payloadLength, verb, flowId, now);
  99. if (trustEstablished) {
  100. _lastTrustEstablishedPacketReceived = now;
  101. path->trustedPacketReceived(now);
  102. }
  103. if (hops == 0) {
  104. // If this is a direct packet (no hops), update existing paths or learn new ones
  105. bool havePath = false;
  106. {
  107. Mutex::Lock _l(_paths_m);
  108. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  109. if (_paths[i].p) {
  110. if (_paths[i].p == path) {
  111. _paths[i].lr = now;
  112. havePath = true;
  113. break;
  114. }
  115. // If same address on same interface then don't learn unless existing path isn't alive (prevents learning loop)
  116. if (_paths[i].p->address().ipsEqual(path->address()) && _paths[i].p->localSocket() == path->localSocket()) {
  117. if (_paths[i].p->alive(now) && !_bond) {
  118. havePath = true;
  119. break;
  120. }
  121. }
  122. } else {
  123. break;
  124. }
  125. }
  126. }
  127. if ( (!havePath) && RR->node->shouldUsePathForZeroTierTraffic(tPtr,_id.address(),path->localSocket(),path->address()) ) {
  128. if (verb == Packet::VERB_OK) {
  129. Mutex::Lock _l(_paths_m);
  130. unsigned int oldestPathIdx = ZT_MAX_PEER_NETWORK_PATHS;
  131. unsigned int oldestPathAge = 0;
  132. unsigned int replacePath = ZT_MAX_PEER_NETWORK_PATHS;
  133. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  134. if (_paths[i].p) {
  135. // Keep track of oldest path as a last resort option
  136. unsigned int currAge = _paths[i].p->age(now);
  137. if (currAge > oldestPathAge) {
  138. oldestPathAge = currAge;
  139. oldestPathIdx = i;
  140. }
  141. if (_paths[i].p->address().ipsEqual(path->address())) {
  142. if (_paths[i].p->localSocket() == path->localSocket()) {
  143. if (!_paths[i].p->alive(now)) {
  144. replacePath = i;
  145. break;
  146. }
  147. }
  148. }
  149. } else {
  150. replacePath = i;
  151. break;
  152. }
  153. }
  154. // If we didn't find a good candidate then resort to replacing oldest path
  155. replacePath = (replacePath == ZT_MAX_PEER_NETWORK_PATHS) ? oldestPathIdx : replacePath;
  156. if (replacePath != ZT_MAX_PEER_NETWORK_PATHS) {
  157. RR->t->peerLearnedNewPath(tPtr, networkId, *this, path, packetId);
  158. _paths[replacePath].lr = now;
  159. _paths[replacePath].p = path;
  160. _paths[replacePath].priority = 1;
  161. Mutex::Lock _l(_bond_m);
  162. if(_bond) {
  163. _bond->nominatePathToBond(_paths[replacePath].p, now);
  164. }
  165. }
  166. } else {
  167. Mutex::Lock ltl(_lastTriedPath_m);
  168. bool triedTooRecently = false;
  169. for(std::list< std::pair< Path *, int64_t > >::iterator i(_lastTriedPath.begin());i!=_lastTriedPath.end();) {
  170. if ((now - i->second) > 1000) {
  171. _lastTriedPath.erase(i++);
  172. } else if (i->first == path.ptr()) {
  173. ++i;
  174. triedTooRecently = true;
  175. } else {
  176. ++i;
  177. }
  178. }
  179. if (!triedTooRecently) {
  180. _lastTriedPath.push_back(std::pair< Path *, int64_t >(path.ptr(), now));
  181. attemptToContactAt(tPtr,path->localSocket(),path->address(),now,true);
  182. path->sent(now);
  183. RR->t->peerConfirmingUnknownPath(tPtr,networkId,*this,path,packetId,verb);
  184. }
  185. }
  186. }
  187. }
  188. // If we have a trust relationship periodically push a message enumerating
  189. // all known external addresses for ourselves. If we already have a path this
  190. // is done less frequently.
  191. if (this->trustEstablished(now)) {
  192. const int64_t sinceLastPush = now - _lastDirectPathPushSent;
  193. bool lowBandwidth = RR->node->lowBandwidthModeEnabled();
  194. int timerScale = lowBandwidth ? 16 : 1;
  195. if (sinceLastPush >= ((hops == 0) ? ZT_DIRECT_PATH_PUSH_INTERVAL_HAVEPATH * timerScale : ZT_DIRECT_PATH_PUSH_INTERVAL)) {
  196. _lastDirectPathPushSent = now;
  197. std::vector<InetAddress> pathsToPush(RR->node->directPaths());
  198. std::vector<InetAddress> ma = RR->sa->whoami();
  199. pathsToPush.insert(pathsToPush.end(), ma.begin(), ma.end());
  200. if (!pathsToPush.empty()) {
  201. std::vector<InetAddress>::const_iterator p(pathsToPush.begin());
  202. while (p != pathsToPush.end()) {
  203. Packet *const outp = new Packet(_id.address(),RR->identity.address(),Packet::VERB_PUSH_DIRECT_PATHS);
  204. outp->addSize(2); // leave room for count
  205. unsigned int count = 0;
  206. while ((p != pathsToPush.end())&&((outp->size() + 24) < 1200)) {
  207. uint8_t addressType = 4;
  208. switch(p->ss_family) {
  209. case AF_INET:
  210. break;
  211. case AF_INET6:
  212. addressType = 6;
  213. break;
  214. default: // we currently only push IP addresses
  215. ++p;
  216. continue;
  217. }
  218. outp->append((uint8_t)0); // no flags
  219. outp->append((uint16_t)0); // no extensions
  220. outp->append(addressType);
  221. outp->append((uint8_t)((addressType == 4) ? 6 : 18));
  222. outp->append(p->rawIpData(),((addressType == 4) ? 4 : 16));
  223. outp->append((uint16_t)p->port());
  224. ++count;
  225. ++p;
  226. }
  227. if (count) {
  228. Metrics::pkt_push_direct_paths_out++;
  229. outp->setAt(ZT_PACKET_IDX_PAYLOAD,(uint16_t)count);
  230. outp->compress();
  231. outp->armor(_key,true,aesKeysIfSupported());
  232. Metrics::pkt_push_direct_paths_out++;
  233. path->send(RR,tPtr,outp->data(),outp->size(),now);
  234. }
  235. delete outp;
  236. }
  237. }
  238. }
  239. }
  240. }
  241. SharedPtr<Path> Peer::getAppropriatePath(int64_t now, bool includeExpired, int32_t flowId)
  242. {
  243. Mutex::Lock _l(_paths_m);
  244. Mutex::Lock _lb(_bond_m);
  245. if(_bond && _bond->isReady()) {
  246. return _bond->getAppropriatePath(now, flowId);
  247. }
  248. unsigned int bestPath = ZT_MAX_PEER_NETWORK_PATHS;
  249. /**
  250. * Send traffic across the highest quality path only. This algorithm will still
  251. * use the old path quality metric from protocol version 9.
  252. */
  253. long bestPathQuality = 2147483647;
  254. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  255. if (_paths[i].p) {
  256. if ((includeExpired)||((now - _paths[i].lr) < ZT_PEER_PATH_EXPIRATION)) {
  257. const long q = _paths[i].p->quality(now) / _paths[i].priority;
  258. if (q <= bestPathQuality) {
  259. bestPathQuality = q;
  260. bestPath = i;
  261. }
  262. }
  263. } else {
  264. break;
  265. }
  266. }
  267. if (bestPath != ZT_MAX_PEER_NETWORK_PATHS) {
  268. return _paths[bestPath].p;
  269. }
  270. return SharedPtr<Path>();
  271. }
  272. void Peer::introduce(void *const tPtr,const int64_t now,const SharedPtr<Peer> &other) const
  273. {
  274. unsigned int myBestV4ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  275. unsigned int myBestV6ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  276. long myBestV4QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  277. long myBestV6QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  278. unsigned int theirBestV4ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  279. unsigned int theirBestV6ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  280. long theirBestV4QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  281. long theirBestV6QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  282. for(int i=0;i<=ZT_INETADDRESS_MAX_SCOPE;++i) {
  283. myBestV4ByScope[i] = ZT_MAX_PEER_NETWORK_PATHS;
  284. myBestV6ByScope[i] = ZT_MAX_PEER_NETWORK_PATHS;
  285. myBestV4QualityByScope[i] = 2147483647;
  286. myBestV6QualityByScope[i] = 2147483647;
  287. theirBestV4ByScope[i] = ZT_MAX_PEER_NETWORK_PATHS;
  288. theirBestV6ByScope[i] = ZT_MAX_PEER_NETWORK_PATHS;
  289. theirBestV4QualityByScope[i] = 2147483647;
  290. theirBestV6QualityByScope[i] = 2147483647;
  291. }
  292. Mutex::Lock _l1(_paths_m);
  293. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  294. if (_paths[i].p) {
  295. const long q = _paths[i].p->quality(now) / _paths[i].priority;
  296. const unsigned int s = (unsigned int)_paths[i].p->ipScope();
  297. switch(_paths[i].p->address().ss_family) {
  298. case AF_INET:
  299. if (q <= myBestV4QualityByScope[s]) {
  300. myBestV4QualityByScope[s] = q;
  301. myBestV4ByScope[s] = i;
  302. }
  303. break;
  304. case AF_INET6:
  305. if (q <= myBestV6QualityByScope[s]) {
  306. myBestV6QualityByScope[s] = q;
  307. myBestV6ByScope[s] = i;
  308. }
  309. break;
  310. }
  311. } else {
  312. break;
  313. }
  314. }
  315. Mutex::Lock _l2(other->_paths_m);
  316. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  317. if (other->_paths[i].p) {
  318. const long q = other->_paths[i].p->quality(now) / other->_paths[i].priority;
  319. const unsigned int s = (unsigned int)other->_paths[i].p->ipScope();
  320. switch(other->_paths[i].p->address().ss_family) {
  321. case AF_INET:
  322. if (q <= theirBestV4QualityByScope[s]) {
  323. theirBestV4QualityByScope[s] = q;
  324. theirBestV4ByScope[s] = i;
  325. }
  326. break;
  327. case AF_INET6:
  328. if (q <= theirBestV6QualityByScope[s]) {
  329. theirBestV6QualityByScope[s] = q;
  330. theirBestV6ByScope[s] = i;
  331. }
  332. break;
  333. }
  334. } else {
  335. break;
  336. }
  337. }
  338. unsigned int mine = ZT_MAX_PEER_NETWORK_PATHS;
  339. unsigned int theirs = ZT_MAX_PEER_NETWORK_PATHS;
  340. for(int s=ZT_INETADDRESS_MAX_SCOPE;s>=0;--s) {
  341. if ((myBestV6ByScope[s] != ZT_MAX_PEER_NETWORK_PATHS)&&(theirBestV6ByScope[s] != ZT_MAX_PEER_NETWORK_PATHS)) {
  342. mine = myBestV6ByScope[s];
  343. theirs = theirBestV6ByScope[s];
  344. break;
  345. }
  346. if ((myBestV4ByScope[s] != ZT_MAX_PEER_NETWORK_PATHS)&&(theirBestV4ByScope[s] != ZT_MAX_PEER_NETWORK_PATHS)) {
  347. mine = myBestV4ByScope[s];
  348. theirs = theirBestV4ByScope[s];
  349. break;
  350. }
  351. }
  352. if (mine != ZT_MAX_PEER_NETWORK_PATHS) {
  353. unsigned int alt = (unsigned int)RR->node->prng() & 1; // randomize which hint we send first for black magickal NAT-t reasons
  354. const unsigned int completed = alt + 2;
  355. while (alt != completed) {
  356. if ((alt & 1) == 0) {
  357. Packet outp(_id.address(),RR->identity.address(),Packet::VERB_RENDEZVOUS);
  358. outp.append((uint8_t)0);
  359. other->_id.address().appendTo(outp);
  360. outp.append((uint16_t)other->_paths[theirs].p->address().port());
  361. if (other->_paths[theirs].p->address().ss_family == AF_INET6) {
  362. outp.append((uint8_t)16);
  363. outp.append(other->_paths[theirs].p->address().rawIpData(),16);
  364. } else {
  365. outp.append((uint8_t)4);
  366. outp.append(other->_paths[theirs].p->address().rawIpData(),4);
  367. }
  368. outp.armor(_key,true,aesKeysIfSupported());
  369. Metrics::pkt_rendezvous_out++;
  370. _paths[mine].p->send(RR,tPtr,outp.data(),outp.size(),now);
  371. } else {
  372. Packet outp(other->_id.address(),RR->identity.address(),Packet::VERB_RENDEZVOUS);
  373. outp.append((uint8_t)0);
  374. _id.address().appendTo(outp);
  375. outp.append((uint16_t)_paths[mine].p->address().port());
  376. if (_paths[mine].p->address().ss_family == AF_INET6) {
  377. outp.append((uint8_t)16);
  378. outp.append(_paths[mine].p->address().rawIpData(),16);
  379. } else {
  380. outp.append((uint8_t)4);
  381. outp.append(_paths[mine].p->address().rawIpData(),4);
  382. }
  383. outp.armor(other->_key,true,other->aesKeysIfSupported());
  384. Metrics::pkt_rendezvous_out++;
  385. other->_paths[theirs].p->send(RR,tPtr,outp.data(),outp.size(),now);
  386. }
  387. ++alt;
  388. }
  389. }
  390. }
  391. void Peer::sendHELLO(void *tPtr,const int64_t localSocket,const InetAddress &atAddress,int64_t now)
  392. {
  393. Packet outp(_id.address(),RR->identity.address(),Packet::VERB_HELLO);
  394. outp.append((unsigned char)ZT_PROTO_VERSION);
  395. outp.append((unsigned char)ZEROTIER_ONE_VERSION_MAJOR);
  396. outp.append((unsigned char)ZEROTIER_ONE_VERSION_MINOR);
  397. outp.append((uint16_t)ZEROTIER_ONE_VERSION_REVISION);
  398. outp.append(now);
  399. RR->identity.serialize(outp,false);
  400. atAddress.serialize(outp);
  401. outp.append((uint64_t)RR->topology->planetWorldId());
  402. outp.append((uint64_t)RR->topology->planetWorldTimestamp());
  403. const unsigned int startCryptedPortionAt = outp.size();
  404. std::vector<World> moons(RR->topology->moons());
  405. std::vector<uint64_t> moonsWanted(RR->topology->moonsWanted());
  406. outp.append((uint16_t)(moons.size() + moonsWanted.size()));
  407. for(std::vector<World>::const_iterator m(moons.begin());m!=moons.end();++m) {
  408. outp.append((uint8_t)m->type());
  409. outp.append((uint64_t)m->id());
  410. outp.append((uint64_t)m->timestamp());
  411. }
  412. for(std::vector<uint64_t>::const_iterator m(moonsWanted.begin());m!=moonsWanted.end();++m) {
  413. outp.append((uint8_t)World::TYPE_MOON);
  414. outp.append(*m);
  415. outp.append((uint64_t)0);
  416. }
  417. outp.cryptField(_key,startCryptedPortionAt,outp.size() - startCryptedPortionAt);
  418. Metrics::pkt_hello_out++;
  419. if (atAddress) {
  420. outp.armor(_key,false,nullptr); // false == don't encrypt full payload, but add MAC
  421. RR->node->expectReplyTo(outp.packetId());
  422. RR->node->putPacket(tPtr,RR->node->lowBandwidthModeEnabled() ? localSocket : -1,atAddress,outp.data(),outp.size());
  423. } else {
  424. RR->node->expectReplyTo(outp.packetId());
  425. RR->sw->send(tPtr,outp,false); // false == don't encrypt full payload, but add MAC
  426. }
  427. }
  428. void Peer::attemptToContactAt(void *tPtr,const int64_t localSocket,const InetAddress &atAddress,int64_t now,bool sendFullHello)
  429. {
  430. if ( (!sendFullHello) && (_vProto >= 5) && (!((_vMajor == 1)&&(_vMinor == 1)&&(_vRevision == 0))) ) {
  431. Packet outp(_id.address(),RR->identity.address(),Packet::VERB_ECHO);
  432. outp.armor(_key,true,aesKeysIfSupported());
  433. Metrics::pkt_echo_out++;
  434. RR->node->expectReplyTo(outp.packetId());
  435. RR->node->putPacket(tPtr,localSocket,atAddress,outp.data(),outp.size());
  436. } else {
  437. sendHELLO(tPtr,localSocket,atAddress,now);
  438. }
  439. }
  440. void Peer::tryMemorizedPath(void *tPtr,int64_t now)
  441. {
  442. if ((now - _lastTriedMemorizedPath) >= ZT_TRY_MEMORIZED_PATH_INTERVAL) {
  443. _lastTriedMemorizedPath = now;
  444. InetAddress mp;
  445. if (RR->node->externalPathLookup(tPtr,_id.address(),-1,mp)) {
  446. attemptToContactAt(tPtr,-1,mp,now,true);
  447. }
  448. }
  449. }
  450. void Peer::performMultipathStateCheck(void *tPtr, int64_t now)
  451. {
  452. Mutex::Lock _l(_bond_m);
  453. if (_bond) {
  454. // Once enabled the Bond object persists, no need to update state
  455. return;
  456. }
  457. /**
  458. * Check for conditions required for multipath bonding and create a bond
  459. * if allowed.
  460. */
  461. int numAlivePaths = 0;
  462. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  463. if (_paths[i].p && _paths[i].p->alive(now)) {
  464. numAlivePaths++;
  465. }
  466. }
  467. _localMultipathSupported = ((numAlivePaths >= 1) && (RR->bc->inUse()) && (ZT_PROTO_VERSION > 9));
  468. if (_localMultipathSupported && !_bond) {
  469. if (RR->bc) {
  470. _bond = RR->bc->createBond(RR, this);
  471. /**
  472. * Allow new bond to retroactively learn all paths known to this peer
  473. */
  474. if (_bond) {
  475. for (unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  476. if (_paths[i].p) {
  477. _bond->nominatePathToBond(_paths[i].p, now);
  478. }
  479. }
  480. }
  481. }
  482. }
  483. }
  484. unsigned int Peer::doPingAndKeepalive(void *tPtr,int64_t now)
  485. {
  486. unsigned int sent = 0;
  487. {
  488. Mutex::Lock _l(_paths_m);
  489. performMultipathStateCheck(tPtr, now);
  490. const bool sendFullHello = ((now - _lastSentFullHello) >= ZT_PEER_PING_PERIOD);
  491. if (sendFullHello) {
  492. _lastSentFullHello = now;
  493. }
  494. // Right now we only keep pinging links that have the maximum priority. The
  495. // priority is used to track cluster redirections, meaning that when a cluster
  496. // redirects us its redirect target links override all other links and we
  497. // let those old links expire.
  498. long maxPriority = 0;
  499. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  500. if (_paths[i].p) {
  501. maxPriority = std::max(_paths[i].priority,maxPriority);
  502. } else {
  503. break;
  504. }
  505. }
  506. bool deletionOccurred = false;
  507. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  508. if (_paths[i].p) {
  509. // Clean expired and reduced priority paths
  510. if ( ((now - _paths[i].lr) < ZT_PEER_PATH_EXPIRATION) && (_paths[i].priority == maxPriority) ) {
  511. if ((sendFullHello)||(_paths[i].p->needsHeartbeat(now))) {
  512. attemptToContactAt(tPtr,_paths[i].p->localSocket(),_paths[i].p->address(),now,sendFullHello);
  513. _paths[i].p->sent(now);
  514. sent |= (_paths[i].p->address().ss_family == AF_INET) ? 0x1 : 0x2;
  515. }
  516. } else {
  517. _paths[i] = _PeerPath();
  518. deletionOccurred = true;
  519. }
  520. }
  521. if (!_paths[i].p || deletionOccurred) {
  522. for(unsigned int j=i;j<ZT_MAX_PEER_NETWORK_PATHS;++j) {
  523. if (_paths[j].p && i != j) {
  524. _paths[i] = _paths[j];
  525. _paths[j] = _PeerPath();
  526. break;
  527. }
  528. }
  529. deletionOccurred = false;
  530. }
  531. }
  532. #ifndef ZT_NO_PEER_METRICS
  533. uint16_t alive_path_count_tmp = 0, dead_path_count_tmp = 0;
  534. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  535. if (_paths[i].p) {
  536. if (_paths[i].p->alive(now)) {
  537. alive_path_count_tmp++;
  538. }
  539. else {
  540. dead_path_count_tmp++;
  541. }
  542. }
  543. }
  544. _alive_path_count = alive_path_count_tmp;
  545. _dead_path_count = dead_path_count_tmp;
  546. #endif
  547. }
  548. #ifndef ZT_NO_PEER_METRICS
  549. _peer_latency.Observe(latency(now));
  550. #endif
  551. return sent;
  552. }
  553. void Peer::clusterRedirect(void *tPtr,const SharedPtr<Path> &originatingPath,const InetAddress &remoteAddress,const int64_t now)
  554. {
  555. SharedPtr<Path> np(RR->topology->getPath(originatingPath->localSocket(),remoteAddress));
  556. RR->t->peerRedirected(tPtr,0,*this,np);
  557. attemptToContactAt(tPtr,originatingPath->localSocket(),remoteAddress,now,true);
  558. {
  559. Mutex::Lock _l(_paths_m);
  560. // New priority is higher than the priority of the originating path (if known)
  561. long newPriority = 1;
  562. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  563. if (_paths[i].p) {
  564. if (_paths[i].p == originatingPath) {
  565. newPriority = _paths[i].priority;
  566. break;
  567. }
  568. } else {
  569. break;
  570. }
  571. }
  572. newPriority += 2;
  573. // Erase any paths with lower priority than this one or that are duplicate
  574. // IPs and add this path.
  575. unsigned int j = 0;
  576. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  577. if (_paths[i].p) {
  578. if ((_paths[i].priority >= newPriority)&&(!_paths[i].p->address().ipsEqual2(remoteAddress))) {
  579. if (i != j) {
  580. _paths[j] = _paths[i];
  581. }
  582. ++j;
  583. }
  584. }
  585. }
  586. if (j < ZT_MAX_PEER_NETWORK_PATHS) {
  587. _paths[j].lr = now;
  588. _paths[j].p = np;
  589. _paths[j].priority = newPriority;
  590. ++j;
  591. while (j < ZT_MAX_PEER_NETWORK_PATHS) {
  592. _paths[j].lr = 0;
  593. _paths[j].p.zero();
  594. _paths[j].priority = 1;
  595. ++j;
  596. }
  597. }
  598. }
  599. }
  600. void Peer::resetWithinScope(void *tPtr,InetAddress::IpScope scope,int inetAddressFamily,int64_t now)
  601. {
  602. Mutex::Lock _l(_paths_m);
  603. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  604. if (_paths[i].p) {
  605. if ((_paths[i].p->address().ss_family == inetAddressFamily)&&(_paths[i].p->ipScope() == scope)) {
  606. attemptToContactAt(tPtr,_paths[i].p->localSocket(),_paths[i].p->address(),now,false);
  607. _paths[i].p->sent(now);
  608. _paths[i].lr = 0; // path will not be used unless it speaks again
  609. }
  610. } else {
  611. break;
  612. }
  613. }
  614. }
  615. void Peer::recordOutgoingPacket(const SharedPtr<Path> &path, const uint64_t packetId,
  616. uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)
  617. {
  618. #ifndef ZT_NO_PEER_METRICS
  619. _outgoing_packet++;
  620. #endif
  621. if (_localMultipathSupported && _bond) {
  622. _bond->recordOutgoingPacket(path, packetId, payloadLength, verb, flowId, now);
  623. }
  624. }
  625. void Peer::recordIncomingInvalidPacket(const SharedPtr<Path>& path)
  626. {
  627. #ifndef ZT_NO_PEER_METRICS
  628. _packet_errors++;
  629. #endif
  630. if (_localMultipathSupported && _bond) {
  631. _bond->recordIncomingInvalidPacket(path);
  632. }
  633. }
  634. void Peer::recordIncomingPacket(const SharedPtr<Path> &path, const uint64_t packetId,
  635. uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)
  636. {
  637. if (_localMultipathSupported && _bond) {
  638. _bond->recordIncomingPacket(path, packetId, payloadLength, verb, flowId, now);
  639. }
  640. }
  641. } // namespace ZeroTier