Peer.cpp 21 KB

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