Peer.cpp 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695
  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. if (! lowBandwidth) {
  199. std::vector<InetAddress> ma = RR->sa->whoami();
  200. pathsToPush.insert(pathsToPush.end(), ma.begin(), ma.end());
  201. }
  202. if (!pathsToPush.empty()) {
  203. std::vector<InetAddress>::const_iterator p(pathsToPush.begin());
  204. while (p != pathsToPush.end()) {
  205. Packet *const outp = new Packet(_id.address(),RR->identity.address(),Packet::VERB_PUSH_DIRECT_PATHS);
  206. outp->addSize(2); // leave room for count
  207. unsigned int count = 0;
  208. while ((p != pathsToPush.end())&&((outp->size() + 24) < 1200)) {
  209. uint8_t addressType = 4;
  210. switch(p->ss_family) {
  211. case AF_INET:
  212. break;
  213. case AF_INET6:
  214. addressType = 6;
  215. break;
  216. default: // we currently only push IP addresses
  217. ++p;
  218. continue;
  219. }
  220. outp->append((uint8_t)0); // no flags
  221. outp->append((uint16_t)0); // no extensions
  222. outp->append(addressType);
  223. outp->append((uint8_t)((addressType == 4) ? 6 : 18));
  224. outp->append(p->rawIpData(),((addressType == 4) ? 4 : 16));
  225. outp->append((uint16_t)p->port());
  226. ++count;
  227. ++p;
  228. }
  229. if (count) {
  230. Metrics::pkt_push_direct_paths_out++;
  231. outp->setAt(ZT_PACKET_IDX_PAYLOAD,(uint16_t)count);
  232. outp->compress();
  233. outp->armor(_key,true,aesKeysIfSupported());
  234. Metrics::pkt_push_direct_paths_out++;
  235. path->send(RR,tPtr,outp->data(),outp->size(),now);
  236. }
  237. delete outp;
  238. }
  239. }
  240. }
  241. }
  242. }
  243. SharedPtr<Path> Peer::getAppropriatePath(int64_t now, bool includeExpired, int32_t flowId)
  244. {
  245. Mutex::Lock _l(_paths_m);
  246. Mutex::Lock _lb(_bond_m);
  247. if(_bond && _bond->isReady()) {
  248. return _bond->getAppropriatePath(now, flowId);
  249. }
  250. unsigned int bestPath = ZT_MAX_PEER_NETWORK_PATHS;
  251. /**
  252. * Send traffic across the highest quality path only. This algorithm will still
  253. * use the old path quality metric from protocol version 9.
  254. */
  255. long bestPathQuality = 2147483647;
  256. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  257. if (_paths[i].p) {
  258. if ((includeExpired)||((now - _paths[i].lr) < ZT_PEER_PATH_EXPIRATION)) {
  259. const long q = _paths[i].p->quality(now) / _paths[i].priority;
  260. if (q <= bestPathQuality) {
  261. bestPathQuality = q;
  262. bestPath = i;
  263. }
  264. }
  265. } else {
  266. break;
  267. }
  268. }
  269. if (bestPath != ZT_MAX_PEER_NETWORK_PATHS) {
  270. return _paths[bestPath].p;
  271. }
  272. return SharedPtr<Path>();
  273. }
  274. void Peer::introduce(void *const tPtr,const int64_t now,const SharedPtr<Peer> &other) const
  275. {
  276. unsigned int myBestV4ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  277. unsigned int myBestV6ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  278. long myBestV4QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  279. long myBestV6QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  280. unsigned int theirBestV4ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  281. unsigned int theirBestV6ByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  282. long theirBestV4QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  283. long theirBestV6QualityByScope[ZT_INETADDRESS_MAX_SCOPE+1];
  284. for(int i=0;i<=ZT_INETADDRESS_MAX_SCOPE;++i) {
  285. myBestV4ByScope[i] = ZT_MAX_PEER_NETWORK_PATHS;
  286. myBestV6ByScope[i] = ZT_MAX_PEER_NETWORK_PATHS;
  287. myBestV4QualityByScope[i] = 2147483647;
  288. myBestV6QualityByScope[i] = 2147483647;
  289. theirBestV4ByScope[i] = ZT_MAX_PEER_NETWORK_PATHS;
  290. theirBestV6ByScope[i] = ZT_MAX_PEER_NETWORK_PATHS;
  291. theirBestV4QualityByScope[i] = 2147483647;
  292. theirBestV6QualityByScope[i] = 2147483647;
  293. }
  294. Mutex::Lock _l1(_paths_m);
  295. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  296. if (_paths[i].p) {
  297. const long q = _paths[i].p->quality(now) / _paths[i].priority;
  298. const unsigned int s = (unsigned int)_paths[i].p->ipScope();
  299. switch(_paths[i].p->address().ss_family) {
  300. case AF_INET:
  301. if (q <= myBestV4QualityByScope[s]) {
  302. myBestV4QualityByScope[s] = q;
  303. myBestV4ByScope[s] = i;
  304. }
  305. break;
  306. case AF_INET6:
  307. if (q <= myBestV6QualityByScope[s]) {
  308. myBestV6QualityByScope[s] = q;
  309. myBestV6ByScope[s] = i;
  310. }
  311. break;
  312. }
  313. } else {
  314. break;
  315. }
  316. }
  317. Mutex::Lock _l2(other->_paths_m);
  318. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  319. if (other->_paths[i].p) {
  320. const long q = other->_paths[i].p->quality(now) / other->_paths[i].priority;
  321. const unsigned int s = (unsigned int)other->_paths[i].p->ipScope();
  322. switch(other->_paths[i].p->address().ss_family) {
  323. case AF_INET:
  324. if (q <= theirBestV4QualityByScope[s]) {
  325. theirBestV4QualityByScope[s] = q;
  326. theirBestV4ByScope[s] = i;
  327. }
  328. break;
  329. case AF_INET6:
  330. if (q <= theirBestV6QualityByScope[s]) {
  331. theirBestV6QualityByScope[s] = q;
  332. theirBestV6ByScope[s] = i;
  333. }
  334. break;
  335. }
  336. } else {
  337. break;
  338. }
  339. }
  340. unsigned int mine = ZT_MAX_PEER_NETWORK_PATHS;
  341. unsigned int theirs = ZT_MAX_PEER_NETWORK_PATHS;
  342. for(int s=ZT_INETADDRESS_MAX_SCOPE;s>=0;--s) {
  343. if ((myBestV6ByScope[s] != ZT_MAX_PEER_NETWORK_PATHS)&&(theirBestV6ByScope[s] != ZT_MAX_PEER_NETWORK_PATHS)) {
  344. mine = myBestV6ByScope[s];
  345. theirs = theirBestV6ByScope[s];
  346. break;
  347. }
  348. if ((myBestV4ByScope[s] != ZT_MAX_PEER_NETWORK_PATHS)&&(theirBestV4ByScope[s] != ZT_MAX_PEER_NETWORK_PATHS)) {
  349. mine = myBestV4ByScope[s];
  350. theirs = theirBestV4ByScope[s];
  351. break;
  352. }
  353. }
  354. if (mine != ZT_MAX_PEER_NETWORK_PATHS) {
  355. unsigned int alt = (unsigned int)RR->node->prng() & 1; // randomize which hint we send first for black magickal NAT-t reasons
  356. const unsigned int completed = alt + 2;
  357. while (alt != completed) {
  358. if ((alt & 1) == 0) {
  359. Packet outp(_id.address(),RR->identity.address(),Packet::VERB_RENDEZVOUS);
  360. outp.append((uint8_t)0);
  361. other->_id.address().appendTo(outp);
  362. outp.append((uint16_t)other->_paths[theirs].p->address().port());
  363. if (other->_paths[theirs].p->address().ss_family == AF_INET6) {
  364. outp.append((uint8_t)16);
  365. outp.append(other->_paths[theirs].p->address().rawIpData(),16);
  366. } else {
  367. outp.append((uint8_t)4);
  368. outp.append(other->_paths[theirs].p->address().rawIpData(),4);
  369. }
  370. outp.armor(_key,true,aesKeysIfSupported());
  371. Metrics::pkt_rendezvous_out++;
  372. _paths[mine].p->send(RR,tPtr,outp.data(),outp.size(),now);
  373. } else {
  374. Packet outp(other->_id.address(),RR->identity.address(),Packet::VERB_RENDEZVOUS);
  375. outp.append((uint8_t)0);
  376. _id.address().appendTo(outp);
  377. outp.append((uint16_t)_paths[mine].p->address().port());
  378. if (_paths[mine].p->address().ss_family == AF_INET6) {
  379. outp.append((uint8_t)16);
  380. outp.append(_paths[mine].p->address().rawIpData(),16);
  381. } else {
  382. outp.append((uint8_t)4);
  383. outp.append(_paths[mine].p->address().rawIpData(),4);
  384. }
  385. outp.armor(other->_key,true,other->aesKeysIfSupported());
  386. Metrics::pkt_rendezvous_out++;
  387. other->_paths[theirs].p->send(RR,tPtr,outp.data(),outp.size(),now);
  388. }
  389. ++alt;
  390. }
  391. }
  392. }
  393. void Peer::sendHELLO(void *tPtr,const int64_t localSocket,const InetAddress &atAddress,int64_t now)
  394. {
  395. Packet outp(_id.address(),RR->identity.address(),Packet::VERB_HELLO);
  396. outp.append((unsigned char)ZT_PROTO_VERSION);
  397. outp.append((unsigned char)ZEROTIER_ONE_VERSION_MAJOR);
  398. outp.append((unsigned char)ZEROTIER_ONE_VERSION_MINOR);
  399. outp.append((uint16_t)ZEROTIER_ONE_VERSION_REVISION);
  400. outp.append(now);
  401. RR->identity.serialize(outp,false);
  402. atAddress.serialize(outp);
  403. outp.append((uint64_t)RR->topology->planetWorldId());
  404. outp.append((uint64_t)RR->topology->planetWorldTimestamp());
  405. const unsigned int startCryptedPortionAt = outp.size();
  406. std::vector<World> moons(RR->topology->moons());
  407. std::vector<uint64_t> moonsWanted(RR->topology->moonsWanted());
  408. outp.append((uint16_t)(moons.size() + moonsWanted.size()));
  409. for(std::vector<World>::const_iterator m(moons.begin());m!=moons.end();++m) {
  410. outp.append((uint8_t)m->type());
  411. outp.append((uint64_t)m->id());
  412. outp.append((uint64_t)m->timestamp());
  413. }
  414. for(std::vector<uint64_t>::const_iterator m(moonsWanted.begin());m!=moonsWanted.end();++m) {
  415. outp.append((uint8_t)World::TYPE_MOON);
  416. outp.append(*m);
  417. outp.append((uint64_t)0);
  418. }
  419. outp.cryptField(_key,startCryptedPortionAt,outp.size() - startCryptedPortionAt);
  420. Metrics::pkt_hello_out++;
  421. if (atAddress) {
  422. outp.armor(_key,false,nullptr); // false == don't encrypt full payload, but add MAC
  423. RR->node->expectReplyTo(outp.packetId());
  424. RR->node->putPacket(tPtr,RR->node->lowBandwidthModeEnabled() ? localSocket : -1,atAddress,outp.data(),outp.size());
  425. } else {
  426. RR->node->expectReplyTo(outp.packetId());
  427. RR->sw->send(tPtr,outp,false); // false == don't encrypt full payload, but add MAC
  428. }
  429. }
  430. void Peer::attemptToContactAt(void *tPtr,const int64_t localSocket,const InetAddress &atAddress,int64_t now,bool sendFullHello)
  431. {
  432. if ( (!sendFullHello) && (_vProto >= 5) && (!((_vMajor == 1)&&(_vMinor == 1)&&(_vRevision == 0))) ) {
  433. Packet outp(_id.address(),RR->identity.address(),Packet::VERB_ECHO);
  434. outp.armor(_key,true,aesKeysIfSupported());
  435. Metrics::pkt_echo_out++;
  436. RR->node->expectReplyTo(outp.packetId());
  437. RR->node->putPacket(tPtr,localSocket,atAddress,outp.data(),outp.size());
  438. } else {
  439. sendHELLO(tPtr,localSocket,atAddress,now);
  440. }
  441. }
  442. void Peer::tryMemorizedPath(void *tPtr,int64_t now)
  443. {
  444. if ((now - _lastTriedMemorizedPath) >= ZT_TRY_MEMORIZED_PATH_INTERVAL) {
  445. _lastTriedMemorizedPath = now;
  446. InetAddress mp;
  447. if (RR->node->externalPathLookup(tPtr,_id.address(),-1,mp)) {
  448. attemptToContactAt(tPtr,-1,mp,now,true);
  449. }
  450. }
  451. }
  452. void Peer::performMultipathStateCheck(void *tPtr, int64_t now)
  453. {
  454. Mutex::Lock _l(_bond_m);
  455. if (_bond) {
  456. // Once enabled the Bond object persists, no need to update state
  457. return;
  458. }
  459. /**
  460. * Check for conditions required for multipath bonding and create a bond
  461. * if allowed.
  462. */
  463. int numAlivePaths = 0;
  464. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  465. if (_paths[i].p && _paths[i].p->alive(now)) {
  466. numAlivePaths++;
  467. }
  468. }
  469. _localMultipathSupported = ((numAlivePaths >= 1) && (RR->bc->inUse()) && (ZT_PROTO_VERSION > 9));
  470. if (_localMultipathSupported && !_bond) {
  471. if (RR->bc) {
  472. _bond = RR->bc->createBond(RR, this);
  473. /**
  474. * Allow new bond to retroactively learn all paths known to this peer
  475. */
  476. if (_bond) {
  477. for (unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  478. if (_paths[i].p) {
  479. _bond->nominatePathToBond(_paths[i].p, now);
  480. }
  481. }
  482. }
  483. }
  484. }
  485. }
  486. unsigned int Peer::doPingAndKeepalive(void *tPtr,int64_t now)
  487. {
  488. unsigned int sent = 0;
  489. {
  490. Mutex::Lock _l(_paths_m);
  491. performMultipathStateCheck(tPtr, now);
  492. const bool sendFullHello = ((now - _lastSentFullHello) >= ZT_PEER_PING_PERIOD);
  493. if (sendFullHello) {
  494. _lastSentFullHello = now;
  495. }
  496. // Right now we only keep pinging links that have the maximum priority. The
  497. // priority is used to track cluster redirections, meaning that when a cluster
  498. // redirects us its redirect target links override all other links and we
  499. // let those old links expire.
  500. long maxPriority = 0;
  501. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  502. if (_paths[i].p) {
  503. maxPriority = std::max(_paths[i].priority,maxPriority);
  504. } else {
  505. break;
  506. }
  507. }
  508. bool deletionOccurred = false;
  509. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  510. if (_paths[i].p) {
  511. // Clean expired and reduced priority paths
  512. if ( ((now - _paths[i].lr) < ZT_PEER_PATH_EXPIRATION) && (_paths[i].priority == maxPriority) ) {
  513. if ((sendFullHello)||(_paths[i].p->needsHeartbeat(now))) {
  514. attemptToContactAt(tPtr,_paths[i].p->localSocket(),_paths[i].p->address(),now,sendFullHello);
  515. _paths[i].p->sent(now);
  516. sent |= (_paths[i].p->address().ss_family == AF_INET) ? 0x1 : 0x2;
  517. }
  518. } else {
  519. _paths[i] = _PeerPath();
  520. deletionOccurred = true;
  521. }
  522. }
  523. if (!_paths[i].p || deletionOccurred) {
  524. for(unsigned int j=i;j<ZT_MAX_PEER_NETWORK_PATHS;++j) {
  525. if (_paths[j].p && i != j) {
  526. _paths[i] = _paths[j];
  527. _paths[j] = _PeerPath();
  528. break;
  529. }
  530. }
  531. deletionOccurred = false;
  532. }
  533. }
  534. #ifndef ZT_NO_PEER_METRICS
  535. uint16_t alive_path_count_tmp = 0, dead_path_count_tmp = 0;
  536. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  537. if (_paths[i].p) {
  538. if (_paths[i].p->alive(now)) {
  539. alive_path_count_tmp++;
  540. }
  541. else {
  542. dead_path_count_tmp++;
  543. }
  544. }
  545. }
  546. _alive_path_count = alive_path_count_tmp;
  547. _dead_path_count = dead_path_count_tmp;
  548. #endif
  549. }
  550. #ifndef ZT_NO_PEER_METRICS
  551. _peer_latency.Observe(latency(now));
  552. #endif
  553. return sent;
  554. }
  555. void Peer::clusterRedirect(void *tPtr,const SharedPtr<Path> &originatingPath,const InetAddress &remoteAddress,const int64_t now)
  556. {
  557. SharedPtr<Path> np(RR->topology->getPath(originatingPath->localSocket(),remoteAddress));
  558. RR->t->peerRedirected(tPtr,0,*this,np);
  559. attemptToContactAt(tPtr,originatingPath->localSocket(),remoteAddress,now,true);
  560. {
  561. Mutex::Lock _l(_paths_m);
  562. // New priority is higher than the priority of the originating path (if known)
  563. long newPriority = 1;
  564. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  565. if (_paths[i].p) {
  566. if (_paths[i].p == originatingPath) {
  567. newPriority = _paths[i].priority;
  568. break;
  569. }
  570. } else {
  571. break;
  572. }
  573. }
  574. newPriority += 2;
  575. // Erase any paths with lower priority than this one or that are duplicate
  576. // IPs and add this path.
  577. unsigned int j = 0;
  578. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  579. if (_paths[i].p) {
  580. if ((_paths[i].priority >= newPriority)&&(!_paths[i].p->address().ipsEqual2(remoteAddress))) {
  581. if (i != j) {
  582. _paths[j] = _paths[i];
  583. }
  584. ++j;
  585. }
  586. }
  587. }
  588. if (j < ZT_MAX_PEER_NETWORK_PATHS) {
  589. _paths[j].lr = now;
  590. _paths[j].p = np;
  591. _paths[j].priority = newPriority;
  592. ++j;
  593. while (j < ZT_MAX_PEER_NETWORK_PATHS) {
  594. _paths[j].lr = 0;
  595. _paths[j].p.zero();
  596. _paths[j].priority = 1;
  597. ++j;
  598. }
  599. }
  600. }
  601. }
  602. void Peer::resetWithinScope(void *tPtr,InetAddress::IpScope scope,int inetAddressFamily,int64_t now)
  603. {
  604. Mutex::Lock _l(_paths_m);
  605. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  606. if (_paths[i].p) {
  607. if ((_paths[i].p->address().ss_family == inetAddressFamily)&&(_paths[i].p->ipScope() == scope)) {
  608. attemptToContactAt(tPtr,_paths[i].p->localSocket(),_paths[i].p->address(),now,false);
  609. _paths[i].p->sent(now);
  610. _paths[i].lr = 0; // path will not be used unless it speaks again
  611. }
  612. } else {
  613. break;
  614. }
  615. }
  616. }
  617. void Peer::recordOutgoingPacket(const SharedPtr<Path> &path, const uint64_t packetId,
  618. uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)
  619. {
  620. #ifndef ZT_NO_PEER_METRICS
  621. _outgoing_packet++;
  622. #endif
  623. if (_localMultipathSupported && _bond) {
  624. _bond->recordOutgoingPacket(path, packetId, payloadLength, verb, flowId, now);
  625. }
  626. }
  627. void Peer::recordIncomingInvalidPacket(const SharedPtr<Path>& path)
  628. {
  629. #ifndef ZT_NO_PEER_METRICS
  630. _packet_errors++;
  631. #endif
  632. if (_localMultipathSupported && _bond) {
  633. _bond->recordIncomingInvalidPacket(path);
  634. }
  635. }
  636. void Peer::recordIncomingPacket(const SharedPtr<Path> &path, const uint64_t packetId,
  637. uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)
  638. {
  639. if (_localMultipathSupported && _bond) {
  640. _bond->recordIncomingPacket(path, packetId, payloadLength, verb, flowId, now);
  641. }
  642. }
  643. } // namespace ZeroTier