Bond.cpp 70 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867
  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: 2024-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 <cmath>
  14. #include "Peer.hpp"
  15. #include "Bond.hpp"
  16. #include "Switch.hpp"
  17. #include "Flow.hpp"
  18. #include "Path.hpp"
  19. namespace ZeroTier {
  20. Bond::Bond(const RuntimeEnvironment *renv, int policy, const SharedPtr<Peer>& peer) :
  21. RR(renv),
  22. _peer(peer)
  23. {
  24. setReasonableDefaults(policy);
  25. _policyAlias = BondController::getPolicyStrByCode(policy);
  26. }
  27. Bond::Bond(std::string& basePolicy, std::string& policyAlias, const SharedPtr<Peer>& peer) :
  28. _policyAlias(policyAlias),
  29. _peer(peer)
  30. {
  31. setReasonableDefaults(BondController::getPolicyCodeByStr(basePolicy));
  32. }
  33. Bond::Bond(const RuntimeEnvironment *renv, const Bond &originalBond, const SharedPtr<Peer>& peer) :
  34. RR(renv),
  35. _peer(peer)
  36. {
  37. // First, set everything to sane defaults
  38. setReasonableDefaults(originalBond._bondingPolicy);
  39. _policyAlias = originalBond._policyAlias;
  40. // Second, apply user specified values (only if they make sense)
  41. _downDelay = originalBond._downDelay;
  42. _upDelay = originalBond._upDelay;
  43. if (originalBond._bondMonitorInterval > 0 && originalBond._bondMonitorInterval < 65535) {
  44. _bondMonitorInterval = originalBond._bondMonitorInterval;
  45. }
  46. else {
  47. fprintf(stderr, "warning: bondMonitorInterval (%d) is out of range, using default (%d)\n", originalBond._bondMonitorInterval, _bondMonitorInterval);
  48. }
  49. if (originalBond._slaveMonitorStrategy == ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_PASSIVE
  50. && originalBond._failoverInterval != 0) {
  51. fprintf(stderr, "warning: passive path monitoring was specified, this will prevent failovers from happening in a timely manner.\n");
  52. }
  53. _abSlaveSelectMethod = originalBond._abSlaveSelectMethod;
  54. memcpy(_qualityWeights, originalBond._qualityWeights, ZT_QOS_WEIGHT_SIZE * sizeof(float));
  55. }
  56. void Bond::nominatePath(const SharedPtr<Path>& path, int64_t now)
  57. {
  58. char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "nominatePath: %s %s\n", getSlave(path)->ifname().c_str(), pathStr);
  59. Mutex::Lock _l(_paths_m);
  60. if (!RR->bc->slaveAllowed(_policyAlias, getSlave(path))) {
  61. return;
  62. }
  63. bool alreadyPresent = false;
  64. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  65. if (path.ptr() == _paths[i].ptr()) {
  66. fprintf(stderr, "previously encountered path, not notifying bond (%s)\n", pathStr);
  67. alreadyPresent = true;
  68. break;
  69. }
  70. }
  71. if (!alreadyPresent) {
  72. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  73. if (!_paths[i]) {
  74. fprintf(stderr, "notifyOfNewPath(): Setting path %s to idx=%d\n", pathStr, i);
  75. _paths[i] = path;
  76. //_paths[i]->slave = RR->bc->getSlaveBySocket(_policyAlias, path->localSocket());
  77. _paths[i]->startTrial(now);
  78. break;
  79. }
  80. }
  81. }
  82. curateBond(now, true);
  83. estimatePathQuality(now);
  84. }
  85. SharedPtr<Path> Bond::getAppropriatePath(int64_t now, int32_t flowId)
  86. {
  87. Mutex::Lock _l(_paths_m);
  88. /**
  89. * active-backup
  90. */
  91. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  92. if (_abPath) {
  93. return _abPath;
  94. }
  95. }
  96. /**
  97. * broadcast
  98. */
  99. if (_bondingPolicy== ZT_BONDING_POLICY_BROADCAST) {
  100. return SharedPtr<Path>(); // Handled in Switch::_trySend()
  101. }
  102. if (!_numBondedPaths) {
  103. return SharedPtr<Path>(); // No paths assigned to bond yet, cannot balance traffic
  104. }
  105. /**
  106. * balance-rr
  107. */
  108. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR) {
  109. if (!_allowFlowHashing) {
  110. //fprintf(stderr, "_rrPacketsSentOnCurrSlave=%d, _numBondedPaths=%d, _rrIdx=%d\n", _rrPacketsSentOnCurrSlave, _numBondedPaths, _rrIdx);
  111. if (_packetsPerSlave == 0) {
  112. // Randomly select a path
  113. return _paths[_bondedIdx[_freeRandomByte % _numBondedPaths]]; // TODO: Optimize
  114. }
  115. if (_rrPacketsSentOnCurrSlave < _packetsPerSlave) {
  116. // Continue to use this slave
  117. ++_rrPacketsSentOnCurrSlave;
  118. return _paths[_bondedIdx[_rrIdx]];
  119. }
  120. // Reset striping counter
  121. _rrPacketsSentOnCurrSlave = 0;
  122. if (_numBondedPaths == 1) {
  123. _rrIdx = 0;
  124. }
  125. else {
  126. int _tempIdx = _rrIdx;
  127. for (int searchCount = 0; searchCount < (_numBondedPaths-1); searchCount++) {
  128. _tempIdx = (_tempIdx == (_numBondedPaths-1)) ? 0 : _tempIdx+1;
  129. if (_paths[_bondedIdx[_tempIdx]] && _paths[_bondedIdx[_tempIdx]]->eligible(now,_ackSendInterval)) {
  130. _rrIdx = _tempIdx;
  131. break;
  132. }
  133. }
  134. }
  135. fprintf(stderr, "_rrIdx=%d\n", _rrIdx);
  136. if (_paths[_bondedIdx[_rrIdx]]) {
  137. return _paths[_bondedIdx[_rrIdx]];
  138. }
  139. }
  140. }
  141. /**
  142. * balance-xor
  143. */
  144. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE) {
  145. if (!_allowFlowHashing || flowId == -1) {
  146. // No specific path required for unclassified traffic, send on anything
  147. return _paths[_bondedIdx[_freeRandomByte % _numBondedPaths]]; // TODO: Optimize
  148. }
  149. else if (_allowFlowHashing) {
  150. // TODO: Optimize
  151. Mutex::Lock _l(_flows_m);
  152. SharedPtr<Flow> flow;
  153. if (_flows.count(flowId)) {
  154. flow = _flows[flowId];
  155. flow->updateActivity(now);
  156. }
  157. else {
  158. unsigned char entropy;
  159. Utils::getSecureRandom(&entropy, 1);
  160. flow = createFlow(SharedPtr<Path>(), flowId, entropy, now);
  161. }
  162. if (flow) {
  163. return flow->assignedPath();
  164. }
  165. }
  166. }
  167. return SharedPtr<Path>();
  168. }
  169. void Bond::recordIncomingInvalidPacket(const SharedPtr<Path>& path)
  170. {
  171. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "recordIncomingInvalidPacket() %s %s\n", getSlave(path)->ifname().c_str(), pathStr);
  172. Mutex::Lock _l(_paths_m);
  173. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  174. if (_paths[i] == path) {
  175. _paths[i]->packetValiditySamples.push(false);
  176. }
  177. }
  178. }
  179. void Bond::recordOutgoingPacket(const SharedPtr<Path> &path, const uint64_t packetId,
  180. uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)
  181. {
  182. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "recordOutgoingPacket() %s %s, packetId=%llx, payloadLength=%d, verb=%x, flowId=%lx\n", getSlave(path)->ifname().c_str(), pathStr, packetId, payloadLength, verb, flowId);
  183. _freeRandomByte += (unsigned char)(packetId >> 8); // Grab entropy to use in path selection logic
  184. if (!_shouldCollectPathStatistics) {
  185. return;
  186. }
  187. bool isFrame = (verb == Packet::VERB_FRAME || verb == Packet::VERB_EXT_FRAME);
  188. bool shouldRecord = (packetId & (ZT_QOS_ACK_DIVISOR - 1)
  189. && (verb != Packet::VERB_ACK)
  190. && (verb != Packet::VERB_QOS_MEASUREMENT));
  191. if (isFrame || shouldRecord) {
  192. Mutex::Lock _l(_paths_m);
  193. if (isFrame) {
  194. ++(path->_packetsOut);
  195. _lastFrame=now;
  196. }
  197. if (shouldRecord) {
  198. path->_unackedBytes += payloadLength;
  199. // Take note that we're expecting a VERB_ACK on this path as of a specific time
  200. if (path->qosStatsOut.size() < ZT_QOS_MAX_OUTSTANDING_RECORDS) {
  201. path->qosStatsOut[packetId] = now;
  202. }
  203. }
  204. }
  205. if (_allowFlowHashing) {
  206. if (_allowFlowHashing && (flowId != ZT_QOS_NO_FLOW)) {
  207. Mutex::Lock _l(_flows_m);
  208. if (_flows.count(flowId)) {
  209. _flows[flowId]->recordOutgoingBytes(payloadLength);
  210. }
  211. }
  212. }
  213. }
  214. void Bond::recordIncomingPacket(const SharedPtr<Path>& path, uint64_t packetId, uint16_t payloadLength,
  215. Packet::Verb verb, int32_t flowId, int64_t now)
  216. {
  217. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "recordIncomingPacket() %s %s, packetId=%llx, payloadLength=%d, verb=%x, flowId=%lx\n", getSlave(path)->ifname().c_str(), pathStr, packetId, payloadLength, verb, flowId);
  218. bool isFrame = (verb == Packet::VERB_FRAME || verb == Packet::VERB_EXT_FRAME);
  219. bool shouldRecord = (packetId & (ZT_QOS_ACK_DIVISOR - 1)
  220. && (verb != Packet::VERB_ACK)
  221. && (verb != Packet::VERB_QOS_MEASUREMENT));
  222. if (isFrame || shouldRecord) {
  223. Mutex::Lock _l(_paths_m);
  224. if (isFrame) {
  225. ++(path->_packetsIn);
  226. _lastFrame=now;
  227. }
  228. if (shouldRecord) {
  229. path->ackStatsIn[packetId] = payloadLength;
  230. ++(path->_packetsReceivedSinceLastAck);
  231. path->qosStatsIn[packetId] = now;
  232. ++(path->_packetsReceivedSinceLastQoS);
  233. path->packetValiditySamples.push(true);
  234. }
  235. }
  236. /**
  237. * Learn new flows and pro-actively create entries for them in the bond so
  238. * that the next time we send a packet out that is part of a flow we know
  239. * which path to use.
  240. */
  241. if ((flowId != ZT_QOS_NO_FLOW)
  242. && (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR
  243. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR
  244. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE)) {
  245. Mutex::Lock _l(_flows_m);
  246. SharedPtr<Flow> flow;
  247. if (!_flows.count(flowId)) {
  248. flow = createFlow(path, flowId, 0, now);
  249. } else {
  250. flow = _flows[flowId];
  251. }
  252. if (flow) {
  253. flow->recordIncomingBytes(payloadLength);
  254. }
  255. }
  256. }
  257. void Bond::receivedQoS(const SharedPtr<Path>& path, int64_t now, int count, uint64_t *rx_id, uint16_t *rx_ts)
  258. {
  259. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "receivedQoS() %s %s\n", getSlave(path)->ifname().c_str(), pathStr);
  260. Mutex::Lock _l(_paths_m);
  261. // Look up egress times and compute latency values for each record
  262. std::map<uint64_t,uint64_t>::iterator it;
  263. for (int j=0; j<count; j++) {
  264. it = path->qosStatsOut.find(rx_id[j]);
  265. if (it != path->qosStatsOut.end()) {
  266. path->latencySamples.push(((uint16_t)(now - it->second) - rx_ts[j]) / 2);
  267. path->qosStatsOut.erase(it);
  268. }
  269. }
  270. path->qosRecordSize.push(count);
  271. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "receivedQoS() on path %s %s, count=%d, successful=%d, qosStatsOut.size()=%d\n", getSlave(path)->ifname().c_str(), pathStr, count, path->aknowledgedQoSRecordCountSinceLastCheck, path->qosStatsOut.size());
  272. }
  273. void Bond::receivedAck(const SharedPtr<Path>& path, int64_t now, int32_t ackedBytes)
  274. {
  275. Mutex::Lock _l(_paths_m);
  276. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "receivedAck() %s %s, (ackedBytes=%d, lastAckReceived=%lld, ackAge=%lld)\n", getSlave(path)->ifname().c_str(), pathStr, ackedBytes, path->lastAckReceived, path->ackAge(now));
  277. path->_lastAckReceived = now;
  278. path->_unackedBytes = (ackedBytes > path->_unackedBytes) ? 0 : path->_unackedBytes - ackedBytes;
  279. int64_t timeSinceThroughputEstimate = (now - path->_lastThroughputEstimation);
  280. if (timeSinceThroughputEstimate >= throughputMeasurementInterval) {
  281. // TODO: See if this floating point math can be reduced
  282. uint64_t throughput = (uint64_t)((float)(path->_bytesAckedSinceLastThroughputEstimation) / ((float)timeSinceThroughputEstimate / (float)1000));
  283. throughput /= 1000;
  284. if (throughput > 0.0) {
  285. path->throughputSamples.push(throughput);
  286. path->_throughputMax = throughput > path->_throughputMax ? throughput : path->_throughputMax;
  287. }
  288. path->_lastThroughputEstimation = now;
  289. path->_bytesAckedSinceLastThroughputEstimation = 0;
  290. } else {
  291. path->_bytesAckedSinceLastThroughputEstimation += ackedBytes;
  292. }
  293. }
  294. int32_t Bond::generateQoSPacket(const SharedPtr<Path>& path, int64_t now, char *qosBuffer)
  295. {
  296. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "generateQoSPacket() %s %s\n", getSlave(path)->ifname().c_str(), pathStr);
  297. int32_t len = 0;
  298. std::map<uint64_t,uint64_t>::iterator it = path->qosStatsIn.begin();
  299. int i=0;
  300. int numRecords = std::min(path->_packetsReceivedSinceLastQoS,ZT_QOS_TABLE_SIZE);
  301. while (i<numRecords && it != path->qosStatsIn.end()) {
  302. uint64_t id = it->first;
  303. memcpy(qosBuffer, &id, sizeof(uint64_t));
  304. qosBuffer+=sizeof(uint64_t);
  305. uint16_t holdingTime = (uint16_t)(now - it->second);
  306. memcpy(qosBuffer, &holdingTime, sizeof(uint16_t));
  307. qosBuffer+=sizeof(uint16_t);
  308. len+=sizeof(uint64_t)+sizeof(uint16_t);
  309. path->qosStatsIn.erase(it++);
  310. ++i;
  311. }
  312. return len;
  313. }
  314. bool Bond::assignFlowToBondedPath(SharedPtr<Flow> &flow, int64_t now)
  315. {
  316. //fprintf(stderr, "assignFlowToBondedPath\n");
  317. char curPathStr[128];
  318. unsigned int idx = ZT_MAX_PEER_NETWORK_PATHS;
  319. if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR) {
  320. idx = abs((int)(flow->id() % (_numBondedPaths)));
  321. flow->assignPath(_paths[_bondedIdx[idx]],now);
  322. }
  323. if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE) {
  324. unsigned char entropy;
  325. Utils::getSecureRandom(&entropy, 1);
  326. if (_totalBondUnderload) {
  327. entropy %= _totalBondUnderload;
  328. }
  329. if (!_numBondedPaths) {
  330. fprintf(stderr, "no bonded paths for flow assignment\n");
  331. return false;
  332. }
  333. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  334. if (_paths[i] && _paths[i]->bonded()) {
  335. SharedPtr<Slave> slave = RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  336. _paths[i]->address().toString(curPathStr);
  337. uint8_t probabilitySegment = (_totalBondUnderload > 0) ? _paths[i]->_affinity : _paths[i]->_allocation;
  338. //fprintf(stderr, "i=%2d, entropy=%3d, alloc=%3d, byteload=%4d, segment=%3d, _totalBondUnderload=%3d, ifname=%s, path=%20s\n", i, entropy, _paths[i]->allocation, _paths[i]->relativeByteLoad, probabilitySegment, _totalBondUnderload, slave->ifname().c_str(), curPathStr);
  339. if (entropy <= probabilitySegment) {
  340. idx = i;
  341. //fprintf(stderr, "\t is best path\n");
  342. break;
  343. }
  344. entropy -= probabilitySegment;
  345. }
  346. }
  347. if (idx < ZT_MAX_PEER_NETWORK_PATHS) {
  348. flow->assignPath(_paths[idx],now);
  349. ++(_paths[idx]->_assignedFlowCount);
  350. }
  351. else {
  352. fprintf(stderr, "could not assign flow?\n"); exit(0); // TODO: Remove
  353. return false;
  354. }
  355. }
  356. flow->assignedPath()->address().toString(curPathStr);
  357. SharedPtr<Slave> slave = RR->bc->getSlaveBySocket(_policyAlias, flow->assignedPath()->localSocket());
  358. fprintf(stderr, "assigned (tx) flow %x with peer %llx to path %s on %s (idx=%d)\n", flow->id(), _peer->_id.address().toInt(), curPathStr, slave->ifname().c_str(), idx);
  359. return true;
  360. }
  361. SharedPtr<Flow> Bond::createFlow(const SharedPtr<Path> &path, int32_t flowId, unsigned char entropy, int64_t now)
  362. {
  363. //fprintf(stderr, "createFlow\n");
  364. char curPathStr[128];
  365. // ---
  366. if (!_numBondedPaths) {
  367. fprintf(stderr, "there are no bonded paths, cannot assign flow\n");
  368. return SharedPtr<Flow>();
  369. }
  370. if (_flows.size() >= ZT_FLOW_MAX_COUNT) {
  371. fprintf(stderr, "max number of flows reached (%d), forcibly forgetting oldest flow\n", ZT_FLOW_MAX_COUNT);
  372. forgetFlowsWhenNecessary(0,true,now);
  373. }
  374. SharedPtr<Flow> flow = new Flow(flowId, now);
  375. _flows[flowId] = flow;
  376. fprintf(stderr, "new flow %x detected with peer %llx, %lu active flow(s)\n", flowId, _peer->_id.address().toInt(), (_flows.size()));
  377. /**
  378. * Add a flow with a given Path already provided. This is the case when a packet
  379. * is received on a path but no flow exists, in this case we simply assign the path
  380. * that the remote peer chose for us.
  381. */
  382. if (path) {
  383. flow->assignPath(path,now);
  384. path->address().toString(curPathStr);
  385. SharedPtr<Slave> slave = RR->bc->getSlaveBySocket(_policyAlias, flow->assignedPath()->localSocket());
  386. fprintf(stderr, "assigned (rx) flow %x with peer %llx to path %s on %s\n", flow->id(), _peer->_id.address().toInt(), curPathStr, slave->ifname().c_str());
  387. }
  388. /**
  389. * Add a flow when no path was provided. This means that it is an outgoing packet
  390. * and that it is up to the local peer to decide how to load-balance its transmission.
  391. */
  392. else if (!path) {
  393. assignFlowToBondedPath(flow, now);
  394. }
  395. return flow;
  396. }
  397. void Bond::forgetFlowsWhenNecessary(uint64_t age, bool oldest, int64_t now)
  398. {
  399. //fprintf(stderr, "forgetFlowsWhenNecessary\n");
  400. std::map<int32_t,SharedPtr<Flow> >::iterator it = _flows.begin();
  401. std::map<int32_t,SharedPtr<Flow> >::iterator oldestFlow = _flows.end();
  402. SharedPtr<Flow> expiredFlow;
  403. if (age) { // Remove by specific age
  404. while (it != _flows.end()) {
  405. if (it->second->age(now) > age) {
  406. fprintf(stderr, "forgetting flow %x between this node and %llx, %lu active flow(s)\n", it->first, _peer->_id.address().toInt(), (_flows.size()-1));
  407. it = _flows.erase(it);
  408. } else {
  409. ++it;
  410. }
  411. }
  412. }
  413. else if (oldest) { // Remove single oldest by natural expiration
  414. uint64_t maxAge = 0;
  415. while (it != _flows.end()) {
  416. if (it->second->age(now) > maxAge) {
  417. maxAge = (now - it->second->age(now));
  418. oldestFlow = it;
  419. }
  420. ++it;
  421. }
  422. if (oldestFlow != _flows.end()) {
  423. fprintf(stderr, "forgetting oldest flow %x (of age %llu) between this node and %llx, %lu active flow(s)\n", oldestFlow->first, oldestFlow->second->age(now), _peer->_id.address().toInt(), (_flows.size()-1));
  424. _flows.erase(oldestFlow);
  425. }
  426. }
  427. fprintf(stderr, "000\n");
  428. }
  429. void Bond::processIncomingPathNegotiationRequest(uint64_t now, SharedPtr<Path> &path, int16_t remoteUtility)
  430. {
  431. //fprintf(stderr, "processIncomingPathNegotiationRequest\n");
  432. if (_abSlaveSelectMethod != ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE) {
  433. return;
  434. }
  435. Mutex::Lock _l(_paths_m);
  436. char pathStr[128];
  437. path->address().toString(pathStr);
  438. if (!_lastPathNegotiationCheck) {
  439. return;
  440. }
  441. SharedPtr<Slave> slave = RR->bc->getSlaveBySocket(_policyAlias, path->localSocket());
  442. if (remoteUtility > _localUtility) {
  443. fprintf(stderr, "peer suggests path, its utility (%d) is greater than ours (%d), we will switch to %s on %s (ls=%llx)\n", remoteUtility, _localUtility, pathStr, slave->ifname().c_str(), path->localSocket());
  444. negotiatedPath = path;
  445. }
  446. if (remoteUtility < _localUtility) {
  447. fprintf(stderr, "peer suggests path, its utility (%d) is less than ours (%d), we will NOT switch to %s on %s (ls=%llx)\n", remoteUtility, _localUtility, pathStr, slave->ifname().c_str(), path->localSocket());
  448. }
  449. if (remoteUtility == _localUtility) {
  450. fprintf(stderr, "peer suggest path, but utility is equal, picking choice made by peer with greater identity.\n");
  451. if (_peer->_id.address().toInt() > RR->node->identity().address().toInt()) {
  452. fprintf(stderr, "peer identity was greater, going with their choice of %s on %s (ls=%llx)\n", pathStr, slave->ifname().c_str(), path->localSocket());
  453. negotiatedPath = path;
  454. } else {
  455. fprintf(stderr, "our identity was greater, no change\n");
  456. }
  457. }
  458. }
  459. void Bond::pathNegotiationCheck(void *tPtr, const int64_t now)
  460. {
  461. //fprintf(stderr, "pathNegotiationCheck\n");
  462. char pathStr[128];
  463. int maxInPathIdx = ZT_MAX_PEER_NETWORK_PATHS;
  464. int maxOutPathIdx = ZT_MAX_PEER_NETWORK_PATHS;
  465. uint64_t maxInCount = 0;
  466. uint64_t maxOutCount = 0;
  467. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  468. if (!_paths[i]) {
  469. continue;
  470. }
  471. if (_paths[i]->_packetsIn > maxInCount) {
  472. maxInCount = _paths[i]->_packetsIn;
  473. maxInPathIdx = i;
  474. }
  475. if (_paths[i]->_packetsOut > maxOutCount) {
  476. maxOutCount = _paths[i]->_packetsOut;
  477. maxOutPathIdx = i;
  478. }
  479. _paths[i]->resetPacketCounts();
  480. }
  481. bool _peerLinksSynchronized = ((maxInPathIdx != ZT_MAX_PEER_NETWORK_PATHS)
  482. && (maxOutPathIdx != ZT_MAX_PEER_NETWORK_PATHS)
  483. && (maxInPathIdx != maxOutPathIdx)) ? false : true;
  484. /**
  485. * Determine utility and attempt to petition remote peer to switch to our chosen path
  486. */
  487. if (!_peerLinksSynchronized) {
  488. _localUtility = _paths[maxOutPathIdx]->_failoverScore - _paths[maxInPathIdx]->_failoverScore;
  489. if (_paths[maxOutPathIdx]->_negotiated) {
  490. _localUtility -= ZT_MULTIPATH_FAILOVER_HANDICAP_NEGOTIATED;
  491. }
  492. if ((now - _lastSentPathNegotiationRequest) > ZT_PATH_NEGOTIATION_CUTOFF_TIME) {
  493. fprintf(stderr, "BT: (sync) it's been long enough, sending more requests.\n");
  494. _numSentPathNegotiationRequests = 0;
  495. }
  496. if (_numSentPathNegotiationRequests < ZT_PATH_NEGOTIATION_TRY_COUNT) {
  497. if (_localUtility >= 0) {
  498. fprintf(stderr, "BT: (sync) paths appear to be out of sync (utility=%d)\n", _localUtility);
  499. sendPATH_NEGOTIATION_REQUEST(tPtr, _paths[maxOutPathIdx]);
  500. ++_numSentPathNegotiationRequests;
  501. _lastSentPathNegotiationRequest = now;
  502. _paths[maxOutPathIdx]->address().toString(pathStr);
  503. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[maxOutPathIdx]->localSocket());
  504. fprintf(stderr, "sending request to use %s on %s, ls=%llx, utility=%d\n", pathStr, slave->ifname().c_str(), _paths[maxOutPathIdx]->localSocket(), _localUtility);
  505. }
  506. }
  507. /**
  508. * Give up negotiating and consider switching
  509. */
  510. else if ((now - _lastSentPathNegotiationRequest) > (2 * ZT_PATH_NEGOTIATION_CHECK_INTERVAL)) {
  511. if (_localUtility == 0) {
  512. // There's no loss to us, just switch without sending a another request
  513. fprintf(stderr, "BT: (sync) giving up, switching to remote peer's path.\n");
  514. negotiatedPath = _paths[maxInPathIdx];
  515. }
  516. }
  517. }
  518. }
  519. void Bond::sendPATH_NEGOTIATION_REQUEST(void *tPtr, const SharedPtr<Path> &path)
  520. {
  521. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "sendPATH_NEGOTIATION_REQUEST() %s %s\n", getSlave(path)->ifname().c_str(), pathStr);
  522. if (_abSlaveSelectMethod != ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE) {
  523. return;
  524. }
  525. Packet outp(_peer->_id.address(),RR->identity.address(),Packet::VERB_PATH_NEGOTIATION_REQUEST);
  526. outp.append<int16_t>(_localUtility);
  527. if (path->address()) {
  528. outp.armor(_peer->key(),false);
  529. RR->node->putPacket(tPtr,path->localSocket(),path->address(),outp.data(),outp.size());
  530. }
  531. }
  532. void Bond::sendACK(void *tPtr,const SharedPtr<Path> &path,const int64_t localSocket,
  533. const InetAddress &atAddress,int64_t now)
  534. {
  535. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "sendACK() %s %s\n", getSlave(path)->ifname().c_str(), pathStr);
  536. Packet outp(_peer->_id.address(),RR->identity.address(),Packet::VERB_ACK);
  537. int32_t bytesToAck = 0;
  538. std::map<uint64_t,uint16_t>::iterator it = path->ackStatsIn.begin();
  539. while (it != path->ackStatsIn.end()) {
  540. bytesToAck += it->second;
  541. ++it;
  542. }
  543. outp.append<uint32_t>(bytesToAck);
  544. if (atAddress) {
  545. outp.armor(_peer->key(),false);
  546. RR->node->putPacket(tPtr,localSocket,atAddress,outp.data(),outp.size());
  547. } else {
  548. RR->sw->send(tPtr,outp,false);
  549. }
  550. path->ackStatsIn.clear();
  551. path->_packetsReceivedSinceLastAck = 0;
  552. path->_lastAckSent = now;
  553. }
  554. void Bond::sendQOS_MEASUREMENT(void *tPtr,const SharedPtr<Path> &path,const int64_t localSocket,
  555. const InetAddress &atAddress,int64_t now)
  556. {
  557. //char pathStr[128];path->address().toString(pathStr);fprintf(stderr, "sendQOS() %s %s\n", getSlave(path)->ifname().c_str(), pathStr);
  558. const int64_t _now = RR->node->now();
  559. Packet outp(_peer->_id.address(),RR->identity.address(),Packet::VERB_QOS_MEASUREMENT);
  560. char qosData[ZT_QOS_MAX_PACKET_SIZE];
  561. int16_t len = generateQoSPacket(path, _now,qosData);
  562. outp.append(qosData,len);
  563. if (atAddress) {
  564. outp.armor(_peer->key(),false);
  565. RR->node->putPacket(tPtr,localSocket,atAddress,outp.data(),outp.size());
  566. } else {
  567. RR->sw->send(tPtr,outp,false);
  568. }
  569. // Account for the fact that a VERB_QOS_MEASUREMENT was just sent. Reset timers.
  570. path->_packetsReceivedSinceLastQoS = 0;
  571. path->_lastQoSMeasurement = now;
  572. }
  573. void Bond::processBackgroundTasks(void *tPtr, const int64_t now)
  574. {
  575. Mutex::Lock _l(_paths_m);
  576. if (!_peer->_canUseMultipath || (now - _lastBackgroundTaskCheck) < ZT_BOND_BACKGROUND_TASK_MIN_INTERVAL) {
  577. return;
  578. }
  579. _lastBackgroundTaskCheck = now;
  580. // Compute dynamic path monitor timer interval
  581. if (_slaveMonitorStrategy == ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC) {
  582. int suggestedMonitorInterval = (now - _lastFrame) / 100;
  583. _dynamicPathMonitorInterval = std::min(ZT_PATH_HEARTBEAT_PERIOD, ((suggestedMonitorInterval > _bondMonitorInterval) ? suggestedMonitorInterval : _bondMonitorInterval));
  584. //fprintf(stderr, "_lastFrame=%llu, suggestedMonitorInterval=%d, _dynamicPathMonitorInterval=%d\n",
  585. // (now-_lastFrame), suggestedMonitorInterval, _dynamicPathMonitorInterval);
  586. }
  587. if (_slaveMonitorStrategy == ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC) {
  588. _shouldCollectPathStatistics = true;
  589. }
  590. // Memoize oft-used properties in the packet ingress/egress logic path
  591. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE) {
  592. // Required for real-time balancing
  593. _shouldCollectPathStatistics = true;
  594. }
  595. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  596. if (_abSlaveSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_BETTER) {
  597. // Required for judging suitability of primary slave after recovery
  598. _shouldCollectPathStatistics = true;
  599. }
  600. if (_abSlaveSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE) {
  601. // Required for judging suitability of new candidate primary
  602. _shouldCollectPathStatistics = true;
  603. }
  604. }
  605. if ((now - _lastCheckUserPreferences) > 1000) {
  606. _lastCheckUserPreferences = now;
  607. applyUserPrefs();
  608. }
  609. curateBond(now,false);
  610. if ((now - _lastQualityEstimation) > _qualityEstimationInterval) {
  611. _lastQualityEstimation = now;
  612. estimatePathQuality(now);
  613. }
  614. dumpInfo(now);
  615. // Send QOS/ACK packets as needed
  616. if (_shouldCollectPathStatistics) {
  617. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  618. if (_paths[i] && _paths[i]->allowed()) {
  619. if (_paths[i]->needsToSendQoS(now,_qosSendInterval)) {
  620. sendQOS_MEASUREMENT(tPtr, _paths[i], _paths[i]->localSocket(), _paths[i]->address(), now);
  621. }
  622. if (_paths[i]->needsToSendAck(now,_ackSendInterval)) {
  623. sendACK(tPtr, _paths[i], _paths[i]->localSocket(), _paths[i]->address(), now);
  624. }
  625. }
  626. }
  627. }
  628. // Perform periodic background tasks unique to each bonding policy
  629. switch (_bondingPolicy)
  630. {
  631. case ZT_BONDING_POLICY_ACTIVE_BACKUP:
  632. processActiveBackupTasks(now);
  633. break;
  634. case ZT_BONDING_POLICY_BROADCAST:
  635. break;
  636. case ZT_BONDING_POLICY_BALANCE_RR:
  637. case ZT_BONDING_POLICY_BALANCE_XOR:
  638. case ZT_BONDING_POLICY_BALANCE_AWARE:
  639. processBalanceTasks(now);
  640. break;
  641. default:
  642. break;
  643. }
  644. // Check whether or not a path negotiation needs to be performed
  645. if (((now - _lastPathNegotiationCheck) > ZT_PATH_NEGOTIATION_CHECK_INTERVAL) && _allowPathNegotiation) {
  646. _lastPathNegotiationCheck = now;
  647. pathNegotiationCheck(tPtr, now);
  648. }
  649. }
  650. void Bond::applyUserPrefs()
  651. {
  652. fprintf(stderr, "applyUserPrefs, _minReqPathMonitorInterval=%d\n", RR->bc->minReqPathMonitorInterval());
  653. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  654. if (!_paths[i]) {
  655. continue;
  656. }
  657. SharedPtr<Slave> sl = getSlave(_paths[i]);
  658. if (sl) {
  659. if (sl->monitorInterval() == 0) { // If no interval was specified for this slave, use more generic bond-wide interval
  660. sl->setMonitorInterval(_bondMonitorInterval);
  661. }
  662. RR->bc->setMinReqPathMonitorInterval((sl->monitorInterval() < RR->bc->minReqPathMonitorInterval()) ? sl->monitorInterval() : RR->bc->minReqPathMonitorInterval());
  663. bool bFoundCommonSlave = false;
  664. SharedPtr<Slave> commonSlave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  665. for(unsigned int j=0;j<ZT_MAX_PEER_NETWORK_PATHS;++j) {
  666. if (_paths[j] && _paths[j].ptr() != _paths[i].ptr()) {
  667. if (RR->bc->getSlaveBySocket(_policyAlias, _paths[j]->localSocket()) == commonSlave) {
  668. bFoundCommonSlave = true;
  669. }
  670. }
  671. }
  672. _paths[i]->_monitorInterval = sl->monitorInterval();
  673. _paths[i]->_upDelay = sl->upDelay() ? sl->upDelay() : _upDelay;
  674. _paths[i]->_downDelay = sl->downDelay() ? sl->downDelay() : _downDelay;
  675. _paths[i]->_ipvPref = sl->ipvPref();
  676. _paths[i]->_mode = sl->mode();
  677. _paths[i]->_enabled = sl->enabled();
  678. _paths[i]->_onlyPathOnSlave = !bFoundCommonSlave;
  679. }
  680. }
  681. if (_peer) {
  682. _peer->_shouldCollectPathStatistics = _shouldCollectPathStatistics;
  683. _peer->_bondingPolicy = _bondingPolicy;
  684. }
  685. }
  686. void Bond::curateBond(const int64_t now, bool rebuildBond)
  687. {
  688. //fprintf(stderr, "%lu curateBond (rebuildBond=%d)\n", ((now - RR->bc->getBondStartTime())), rebuildBond);
  689. char pathStr[128];
  690. /**
  691. * Update path states
  692. */
  693. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  694. if (!_paths[i]) {
  695. continue;
  696. }
  697. bool currEligibility = _paths[i]->eligible(now,_ackSendInterval);
  698. if (currEligibility != _paths[i]->_lastEligibilityState) {
  699. _paths[i]->address().toString(pathStr);
  700. //fprintf(stderr, "\n\n%ld path eligibility (for %s, %s) has changed (from %d to %d)\n", (RR->node->now() - RR->bc->getBondStartTime()), getSlave(_paths[i])->ifname().c_str(), pathStr, _paths[i]->lastCheckedEligibility, _paths[i]->eligible(now,_ackSendInterval));
  701. if (currEligibility) {
  702. rebuildBond = true;
  703. }
  704. if (!currEligibility) {
  705. _paths[i]->adjustRefractoryPeriod(now, _defaultPathRefractoryPeriod, !currEligibility);
  706. if (_paths[i]->bonded()) {
  707. //fprintf(stderr, "the path was bonded, reallocation of its flows will occur soon\n");
  708. rebuildBond = true;
  709. _paths[i]->_shouldReallocateFlows = _paths[i]->bonded();
  710. _paths[i]->setBonded(false);
  711. } else {
  712. //fprintf(stderr, "the path was not bonded, no consequences\n");
  713. }
  714. }
  715. }
  716. if (currEligibility) {
  717. _paths[i]->adjustRefractoryPeriod(now, _defaultPathRefractoryPeriod, false);
  718. }
  719. _paths[i]->_lastEligibilityState = currEligibility;
  720. }
  721. /**
  722. * Curate the set of paths that are part of the bond proper. Selects a single path
  723. * per logical slave according to eligibility and user-specified constraints.
  724. */
  725. if ((_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR)
  726. || (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR)
  727. || (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE)) {
  728. if (!_numBondedPaths) {
  729. rebuildBond = true;
  730. }
  731. // TODO: Optimize
  732. if (rebuildBond) {
  733. int updatedBondedPathCount = 0;
  734. std::map<SharedPtr<Slave>,int> slaveMap;
  735. for (int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  736. if (_paths[i] && _paths[i]->allowed() && (_paths[i]->eligible(now,_ackSendInterval) || !_numBondedPaths)) {
  737. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  738. if (!slaveMap.count(slave)) {
  739. slaveMap[slave] = i;
  740. }
  741. else {
  742. bool overriden = false;
  743. _paths[i]->address().toString(pathStr);
  744. //fprintf(stderr, " slave representative path already exists! (%s %s)\n", getSlave(_paths[i])->ifname().c_str(), pathStr);
  745. if (_paths[i]->preferred() && !_paths[slaveMap[slave]]->preferred()) {
  746. // Override previous choice if preferred
  747. //fprintf(stderr, "overriding since its preferred!\n");
  748. if (_paths[slaveMap[slave]]->_assignedFlowCount) {
  749. _paths[slaveMap[slave]]->_deprecated = true;
  750. }
  751. else {
  752. _paths[slaveMap[slave]]->_deprecated = true;
  753. _paths[slaveMap[slave]]->setBonded(false);
  754. }
  755. slaveMap[slave] = i;
  756. overriden = true;
  757. }
  758. if ((_paths[i]->preferred() && _paths[slaveMap[slave]]->preferred())
  759. || (!_paths[i]->preferred() && !_paths[slaveMap[slave]]->preferred())) {
  760. if (_paths[i]->preferenceRank() > _paths[slaveMap[slave]]->preferenceRank()) {
  761. // Override if higher preference
  762. //fprintf(stderr, "overriding according to preference preferenceRank!\n");
  763. if (_paths[slaveMap[slave]]->_assignedFlowCount) {
  764. _paths[slaveMap[slave]]->_deprecated = true;
  765. }
  766. else {
  767. _paths[slaveMap[slave]]->_deprecated = true;
  768. _paths[slaveMap[slave]]->setBonded(false);
  769. }
  770. slaveMap[slave] = i;
  771. }
  772. }
  773. }
  774. }
  775. }
  776. std::map<SharedPtr<Slave>,int>::iterator it = slaveMap.begin();
  777. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  778. if (!_paths[i]) {
  779. continue;
  780. }
  781. _bondedIdx[i] = ZT_MAX_PEER_NETWORK_PATHS;
  782. if (it != slaveMap.end()) {
  783. _bondedIdx[i] = it->second;
  784. _paths[_bondedIdx[i]]->setBonded(true);
  785. ++it;
  786. ++updatedBondedPathCount;
  787. _paths[_bondedIdx[i]]->address().toString(pathStr);
  788. fprintf(stderr, "setting i=%d, _bondedIdx[%d]=%d to bonded (%s %s)\n", i, i, _bondedIdx[i], getSlave(_paths[_bondedIdx[i]])->ifname().c_str(), pathStr);
  789. }
  790. }
  791. _numBondedPaths = updatedBondedPathCount;
  792. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR) {
  793. // Cause a RR reset since the currently used index might no longer be valid
  794. _rrPacketsSentOnCurrSlave = _packetsPerSlave;
  795. }
  796. }
  797. }
  798. }
  799. void Bond::estimatePathQuality(const int64_t now)
  800. {
  801. char pathStr[128];
  802. //---
  803. uint32_t totUserSpecifiedSlaveSpeed = 0;
  804. if (_numBondedPaths) { // Compute relative user-specified speeds of slaves
  805. for(unsigned int i=0;i<_numBondedPaths;++i) {
  806. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  807. if (_paths[i] && _paths[i]->allowed()) {
  808. totUserSpecifiedSlaveSpeed += slave->speed();
  809. }
  810. }
  811. for(unsigned int i=0;i<_numBondedPaths;++i) {
  812. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  813. if (_paths[i] && _paths[i]->allowed()) {
  814. slave->setRelativeSpeed(round( ((float)slave->speed() / (float)totUserSpecifiedSlaveSpeed) * 255));
  815. }
  816. }
  817. }
  818. float lat[ZT_MAX_PEER_NETWORK_PATHS];
  819. float pdv[ZT_MAX_PEER_NETWORK_PATHS];
  820. float plr[ZT_MAX_PEER_NETWORK_PATHS];
  821. float per[ZT_MAX_PEER_NETWORK_PATHS];
  822. float thr[ZT_MAX_PEER_NETWORK_PATHS];
  823. float thm[ZT_MAX_PEER_NETWORK_PATHS];
  824. float thv[ZT_MAX_PEER_NETWORK_PATHS];
  825. float maxLAT = 0;
  826. float maxPDV = 0;
  827. float maxPLR = 0;
  828. float maxPER = 0;
  829. float maxTHR = 0;
  830. float maxTHM = 0;
  831. float maxTHV = 0;
  832. float quality[ZT_MAX_PEER_NETWORK_PATHS];
  833. uint8_t alloc[ZT_MAX_PEER_NETWORK_PATHS];
  834. float totQuality = 0.0f;
  835. memset(&lat, 0, sizeof(lat));
  836. memset(&pdv, 0, sizeof(pdv));
  837. memset(&plr, 0, sizeof(plr));
  838. memset(&per, 0, sizeof(per));
  839. memset(&thr, 0, sizeof(thr));
  840. memset(&thm, 0, sizeof(thm));
  841. memset(&thv, 0, sizeof(thv));
  842. memset(&quality, 0, sizeof(quality));
  843. memset(&alloc, 0, sizeof(alloc));
  844. // Compute initial summary statistics
  845. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  846. if (!_paths[i] || !_paths[i]->allowed()) {
  847. continue;
  848. }
  849. // Compute/Smooth average of real-world observations
  850. _paths[i]->_latencyMean = _paths[i]->latencySamples.mean();
  851. _paths[i]->_latencyVariance = _paths[i]->latencySamples.stddev();
  852. _paths[i]->_packetErrorRatio = 1.0 - (_paths[i]->packetValiditySamples.count() ? _paths[i]->packetValiditySamples.mean() : 1.0);
  853. if (userHasSpecifiedSlaveSpeeds()) {
  854. // Use user-reported metrics
  855. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  856. if (slave) {
  857. _paths[i]->_throughputMean = slave->speed();
  858. _paths[i]->_throughputVariance = 0;
  859. }
  860. }
  861. /*
  862. else {
  863. // Use estimated metrics
  864. if (_paths[i]->throughputSamples.count()) {
  865. // If we have samples, use them
  866. _paths[i]->throughputMean = (uint64_t)_paths[i]->throughputSamples.mean();
  867. if (_paths[i]->throughputMean > 0) {
  868. _paths[i]->throughputVarianceSamples.push((float)_paths[i]->throughputSamples.stddev() / (float)_paths[i]->throughputMean);
  869. _paths[i]->throughputVariance = _paths[i]->throughputVarianceSamples.mean();
  870. }
  871. }
  872. else {
  873. // No samples have been collected yet, assume best case scenario
  874. _paths[i]->throughputMean = ZT_QOS_THR_NORM_MAX;
  875. _paths[i]->throughputVariance = 0;
  876. }
  877. }
  878. */
  879. // Drain unacknowledged QoS records
  880. std::map<uint64_t,uint64_t>::iterator it = _paths[i]->qosStatsOut.begin();
  881. uint64_t currentLostRecords = 0;
  882. while (it != _paths[i]->qosStatsOut.end()) {
  883. int qosRecordTimeout = 5000; //_paths[i]->monitorInterval() * ZT_MULTIPATH_QOS_ACK_INTERVAL_MULTIPLIER * 8;
  884. if ((now - it->second) >= qosRecordTimeout) {
  885. //fprintf(stderr, "packetId=%llx was lost\n", it->first);
  886. it = _paths[i]->qosStatsOut.erase(it);
  887. ++currentLostRecords;
  888. } else { ++it; }
  889. }
  890. quality[i]=0;
  891. totQuality=0;
  892. // Normalize raw observations according to sane limits and/or user specified values
  893. lat[i] = 1.0 / expf(4*Utils::normalize(_paths[i]->_latencyMean, 0, _maxAcceptableLatency, 0, 1));
  894. pdv[i] = 1.0 / expf(4*Utils::normalize(_paths[i]->_latencyVariance, 0, _maxAcceptablePacketDelayVariance, 0, 1));
  895. plr[i] = 1.0 / expf(4*Utils::normalize(_paths[i]->_packetLossRatio, 0, _maxAcceptablePacketLossRatio, 0, 1));
  896. per[i] = 1.0 / expf(4*Utils::normalize(_paths[i]->_packetErrorRatio, 0, _maxAcceptablePacketErrorRatio, 0, 1));
  897. //thr[i] = 1.0; //Utils::normalize(_paths[i]->throughputMean, 0, ZT_QOS_THR_NORM_MAX, 0, 1);
  898. //thm[i] = 1.0; //Utils::normalize(_paths[i]->throughputMax, 0, ZT_QOS_THM_NORM_MAX, 0, 1);
  899. //thv[i] = 1.0; //1.0 / expf(4*Utils::normalize(_paths[i]->throughputVariance, 0, ZT_QOS_THV_NORM_MAX, 0, 1));
  900. //scp[i] = _paths[i]->ipvPref != 0 ? 1.0 : Utils::normalize(_paths[i]->ipScope(), InetAddress::IP_SCOPE_NONE, InetAddress::IP_SCOPE_PRIVATE, 0, 1);
  901. // Record bond-wide maximums to determine relative values
  902. maxLAT = lat[i] > maxLAT ? lat[i] : maxLAT;
  903. maxPDV = pdv[i] > maxPDV ? pdv[i] : maxPDV;
  904. maxPLR = plr[i] > maxPLR ? plr[i] : maxPLR;
  905. maxPER = per[i] > maxPER ? per[i] : maxPER;
  906. //maxTHR = thr[i] > maxTHR ? thr[i] : maxTHR;
  907. //maxTHM = thm[i] > maxTHM ? thm[i] : maxTHM;
  908. //maxTHV = thv[i] > maxTHV ? thv[i] : maxTHV;
  909. //fprintf(stdout, "EH %d: lat=%8.3f, ltm=%8.3f, pdv=%8.3f, plr=%5.3f, per=%5.3f, thr=%8f, thm=%5.3f, thv=%5.3f, avl=%5.3f, age=%8.2f, scp=%4d, q=%5.3f, qtot=%5.3f, ac=%d if=%s, path=%s\n",
  910. // i, lat[i], ltm[i], pdv[i], plr[i], per[i], thr[i], thm[i], thv[i], avl[i], age[i], scp[i], quality[i], totQuality, alloc[i], getSlave(_paths[i])->ifname().c_str(), pathStr);
  911. }
  912. // Convert metrics to relative quantities and apply contribution weights
  913. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  914. if (_paths[i] && _paths[i]->bonded()) {
  915. quality[i] += ((maxLAT > 0.0f ? lat[i] / maxLAT : 0.0f) * _qualityWeights[ZT_QOS_LAT_IDX]);
  916. quality[i] += ((maxPDV > 0.0f ? pdv[i] / maxPDV : 0.0f) * _qualityWeights[ZT_QOS_PDV_IDX]);
  917. quality[i] += ((maxPLR > 0.0f ? plr[i] / maxPLR : 0.0f) * _qualityWeights[ZT_QOS_PLR_IDX]);
  918. quality[i] += ((maxPER > 0.0f ? per[i] / maxPER : 0.0f) * _qualityWeights[ZT_QOS_PER_IDX]);
  919. //quality[i] += ((maxTHR > 0.0f ? thr[i] / maxTHR : 0.0f) * _qualityWeights[ZT_QOS_THR_IDX]);
  920. //quality[i] += ((maxTHM > 0.0f ? thm[i] / maxTHM : 0.0f) * _qualityWeights[ZT_QOS_THM_IDX]);
  921. //quality[i] += ((maxTHV > 0.0f ? thv[i] / maxTHV : 0.0f) * _qualityWeights[ZT_QOS_THV_IDX]);
  922. //quality[i] += (scp[i] * _qualityWeights[ZT_QOS_SCP_IDX]);
  923. totQuality += quality[i];
  924. }
  925. }
  926. //
  927. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  928. if (_paths[i] && _paths[i]->bonded()) {
  929. alloc[i] = std::ceil((quality[i] / totQuality) * (float)255);
  930. _paths[i]->_allocation = alloc[i];
  931. }
  932. }
  933. /*
  934. if ((now - _lastLogTS) > 500) {
  935. if (!relevant()) {return;}
  936. //fprintf(stderr, "\n");
  937. _lastPrintTS = now;
  938. _lastLogTS = now;
  939. int numPlottablePaths=0;
  940. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  941. if (_paths[i]) {
  942. ++numPlottablePaths;
  943. _paths[i]->address().toString(pathStr);
  944. //fprintf(stderr, "%lu FIN [%d/%d]: pmi=%5d, lat=%4.3f, ltm=%4.3f, pdv=%4.3f, plr=%4.3f, per=%4.3f, thr=%4.3f, thm=%4.3f, thv=%4.3f, age=%4.3f, scp=%4d, q=%4.3f, qtot=%4.3f, ac=%4d, asf=%3d, if=%s, path=%20s, bond=%d, qosout=%d, plrraw=%d\n",
  945. // ((now - RR->bc->getBondStartTime())), i, _numBondedPaths, _paths[i]->monitorInterval,
  946. // lat[i], ltm[i], pdv[i], plr[i], per[i], thr[i], thm[i], thv[i], age[i], scp[i],
  947. // quality[i], totQuality, alloc[i], _paths[i]->assignedFlowCount, getSlave(_paths[i])->ifname().c_str(), pathStr, _paths[i]->bonded(), _paths[i]->qosStatsOut.size(), _paths[i]->packetLossRatio);
  948. }
  949. }
  950. if (numPlottablePaths < 2) {
  951. return;
  952. }
  953. if (!_header) {
  954. fprintf(stdout, "now, bonded, relativeUnderload, flows, ");
  955. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  956. if (_paths[i]) {
  957. _paths[i]->address().toString(pathStr);
  958. std::string label = std::string((pathStr)) + " " + getSlave(_paths[i])->ifname();
  959. for (int i=0; i<19; ++i) {
  960. fprintf(stdout, "%s, ", label.c_str());
  961. }
  962. }
  963. }
  964. _header=true;
  965. }
  966. fprintf(stdout, "%ld, %d, %d, %d, ",((now - RR->bc->getBondStartTime())),_numBondedPaths,_totalBondUnderload, _flows.size());
  967. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  968. if (_paths[i]) {
  969. _paths[i]->address().toString(pathStr);
  970. fprintf(stdout, "%s, %s, %8.3f, %8.3f, %8.3f, %5.3f, %5.3f, %5.3f, %8f, %5.3f, %5.3f, %d, %5.3f, %d, %d, %d, %d, %d, %d, ",
  971. getSlave(_paths[i])->ifname().c_str(), pathStr, _paths[i]->latencyMean, lat[i],pdv[i], _paths[i]->packetLossRatio, plr[i],per[i],thr[i],thm[i],thv[i],(now - _paths[i]->lastIn()),quality[i],alloc[i],
  972. _paths[i]->relativeByteLoad, _paths[i]->assignedFlowCount, _paths[i]->alive(now, true), _paths[i]->eligible(now,_ackSendInterval), _paths[i]->qosStatsOut.size());
  973. }
  974. }
  975. fprintf(stdout, "\n");
  976. }
  977. */
  978. }
  979. void Bond::processBalanceTasks(const int64_t now)
  980. {
  981. //fprintf(stderr, "processBalanceTasks\n");
  982. char curPathStr[128];
  983. if (_allowFlowHashing) {
  984. /**
  985. * Clean up and reset flows if necessary
  986. */
  987. if ((now - _lastFlowExpirationCheck) > ZT_MULTIPATH_FLOW_CHECK_INTERVAL) {
  988. Mutex::Lock _l(_flows_m);
  989. forgetFlowsWhenNecessary(ZT_MULTIPATH_FLOW_EXPIRATION_INTERVAL,false,now);
  990. _lastFlowExpirationCheck = now;
  991. }
  992. if ((now - _lastFlowStatReset) > ZT_FLOW_STATS_RESET_INTERVAL) {
  993. Mutex::Lock _l(_flows_m);
  994. _lastFlowStatReset = now;
  995. std::map<int32_t,SharedPtr<Flow> >::iterator it = _flows.begin();
  996. while (it != _flows.end()) {
  997. it->second->resetByteCounts();
  998. ++it;
  999. }
  1000. }
  1001. /**
  1002. * Re-allocate flows from dead paths
  1003. */
  1004. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE) {
  1005. Mutex::Lock _l(_flows_m);
  1006. for (int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
  1007. if (!_paths[i]) {
  1008. continue;
  1009. }
  1010. if (!_paths[i]->eligible(now,_ackSendInterval) && _paths[i]->_shouldReallocateFlows) {
  1011. _paths[i]->address().toString(curPathStr);
  1012. fprintf(stderr, "%d reallocating flows from dead path %s on %s\n", (RR->node->now() - RR->bc->getBondStartTime()), curPathStr, getSlave(_paths[i])->ifname().c_str());
  1013. std::map<int32_t,SharedPtr<Flow> >::iterator flow_it = _flows.begin();
  1014. while (flow_it != _flows.end()) {
  1015. if (flow_it->second->assignedPath() == _paths[i]) {
  1016. if(assignFlowToBondedPath(flow_it->second, now)) {
  1017. _paths[i]->_assignedFlowCount--;
  1018. }
  1019. }
  1020. ++flow_it;
  1021. }
  1022. _paths[i]->_shouldReallocateFlows = false;
  1023. }
  1024. }
  1025. }
  1026. }
  1027. /**
  1028. * Tasks specific to (Balance Round Robin)
  1029. */
  1030. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR) {
  1031. if (_allowFlowHashing) {
  1032. // TODO: Should ideally failover from (idx) to a random slave, this is so that (idx+1) isn't overloaded
  1033. }
  1034. else if (!_allowFlowHashing) {
  1035. // Nothing
  1036. }
  1037. }
  1038. /**
  1039. * Tasks specific to (Balance XOR)
  1040. */
  1041. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR) {
  1042. // Nothing specific for XOR
  1043. }
  1044. /**
  1045. * Tasks specific to (Balance Aware)
  1046. */
  1047. if ((_bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE)) {
  1048. if (_allowFlowHashing) {
  1049. Mutex::Lock _l(_flows_m);
  1050. /**
  1051. * Re-balance flows in proportion to slave capacity (or when eligibility changes)
  1052. */
  1053. if ((now - _lastFlowRebalance) > ZT_FLOW_REBALANCE_INTERVAL) {
  1054. /**
  1055. * Determine "load" for bonded paths
  1056. */
  1057. uint64_t totalBytes = 0;
  1058. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) { // first pass: compute absolute byte load and total
  1059. if (_paths[i] && _paths[i]->bonded()) {
  1060. _paths[i]->_byteLoad = 0;
  1061. std::map<int32_t,SharedPtr<Flow> >::iterator flow_it = _flows.begin();
  1062. while (flow_it != _flows.end()) {
  1063. if (flow_it->second->assignedPath() == _paths[i]) {
  1064. _paths[i]->_byteLoad += flow_it->second->totalBytes();
  1065. }
  1066. ++flow_it;
  1067. }
  1068. totalBytes += _paths[i]->_byteLoad;
  1069. }
  1070. }
  1071. /**
  1072. * Determine "affinity" for bonded path
  1073. */
  1074. //fprintf(stderr, "\n\n");
  1075. _totalBondUnderload = 0;
  1076. for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) { // second pass: compute relative byte loads and total imbalance
  1077. if (_paths[i] && _paths[i]->bonded()) {
  1078. if (totalBytes) {
  1079. uint8_t relativeByteLoad = std::ceil(((float)_paths[i]->_byteLoad / (float)totalBytes) * (float)255);
  1080. //fprintf(stderr, "lastComputedAllocation = %d\n", _paths[i]->allocation);
  1081. //fprintf(stderr, " relativeByteLoad = %d\n", relativeByteLoad);
  1082. _paths[i]->_relativeByteLoad = relativeByteLoad;
  1083. uint8_t relativeUnderload = std::max(0, (int)_paths[i]->_allocation - (int)relativeByteLoad);
  1084. //fprintf(stderr, " relativeUnderload = %d\n", relativeUnderload);
  1085. _totalBondUnderload += relativeUnderload;
  1086. //fprintf(stderr, " _totalBondUnderload = %d\n\n", _totalBondUnderload);
  1087. //_paths[i]->affinity = (relativeUnderload > 0 ? relativeUnderload : _paths[i]->_allocation);
  1088. }
  1089. else { // set everything to base values
  1090. _totalBondUnderload = 0;
  1091. //_paths[i]->affinity = 0;
  1092. }
  1093. }
  1094. }
  1095. //fprintf(stderr, "_totalBondUnderload=%d (end)\n\n", _totalBondUnderload);
  1096. /**
  1097. *
  1098. */
  1099. //fprintf(stderr, "_lastFlowRebalance\n");
  1100. std::map<int32_t, SharedPtr<Flow> >::iterator it = _flows.begin();
  1101. while (it != _flows.end()) {
  1102. int32_t flowId = it->first;
  1103. SharedPtr<Flow> flow = it->second;
  1104. if ((now - flow->_lastPathReassignment) > ZT_FLOW_MIN_REBALANCE_INTERVAL) {
  1105. //fprintf(stdout, " could move : %x\n", flowId);
  1106. }
  1107. ++it;
  1108. }
  1109. _lastFlowRebalance = now;
  1110. }
  1111. }
  1112. else if (!_allowFlowHashing) {
  1113. // Nothing
  1114. }
  1115. }
  1116. }
  1117. void Bond::dequeueNextActiveBackupPath(const uint64_t now)
  1118. {
  1119. //fprintf(stderr, "dequeueNextActiveBackupPath\n");
  1120. if (_abFailoverQueue.empty()) {
  1121. return;
  1122. }
  1123. _abPath = _abFailoverQueue.front();
  1124. _abFailoverQueue.pop_front();
  1125. _lastActiveBackupPathChange = now;
  1126. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1127. if (_paths[i]) {
  1128. _paths[i]->resetPacketCounts();
  1129. }
  1130. }
  1131. }
  1132. void Bond::processActiveBackupTasks(const int64_t now)
  1133. {
  1134. //fprintf(stderr, "%llu processActiveBackupTasks\n", (now - RR->bc->getBondStartTime()));
  1135. char pathStr[128]; char prevPathStr[128]; char curPathStr[128];
  1136. SharedPtr<Path> prevActiveBackupPath = _abPath;
  1137. SharedPtr<Path> nonPreferredPath;
  1138. bool bFoundPrimarySlave = false;
  1139. /**
  1140. * Select initial "active" active-backup slave
  1141. */
  1142. if (!_abPath) {
  1143. fprintf(stderr, "%llu no active backup path yet...\n", ((now - RR->bc->getBondStartTime())));
  1144. /**
  1145. * [Automatic mode]
  1146. * The user has not explicitly specified slaves or their failover schedule,
  1147. * the bonding policy will now select the first eligible path and set it as
  1148. * its active backup path, if a substantially better path is detected the bonding
  1149. * policy will assign it as the new active backup path. If the path fails it will
  1150. * simply find the next eligible path.
  1151. */
  1152. if (!userHasSpecifiedSlaves()) {
  1153. fprintf(stderr, "%llu AB: (auto) user did not specify any slaves. waiting until we know more\n", ((now - RR->bc->getBondStartTime())));
  1154. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1155. if (_paths[i] && _paths[i]->eligible(now,_ackSendInterval)) {
  1156. _paths[i]->address().toString(curPathStr);
  1157. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1158. if (slave) {
  1159. fprintf(stderr, "%llu AB: (initial) [%d] found eligible path %s on: %s\n", ((now - RR->bc->getBondStartTime())), i, curPathStr, slave->ifname().c_str());
  1160. }
  1161. _abPath = _paths[i];
  1162. break;
  1163. }
  1164. }
  1165. }
  1166. /**
  1167. * [Manual mode]
  1168. * The user has specified slaves or failover rules that the bonding policy should adhere to.
  1169. */
  1170. else if (userHasSpecifiedSlaves()) {
  1171. fprintf(stderr, "%llu AB: (manual) no active backup slave, checking local.conf\n", ((now - RR->bc->getBondStartTime())));
  1172. if (userHasSpecifiedPrimarySlave()) {
  1173. fprintf(stderr, "%llu AB: (manual) user has specified primary slave, looking for it.\n", ((now - RR->bc->getBondStartTime())));
  1174. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1175. if (!_paths[i]) {
  1176. continue;
  1177. }
  1178. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1179. if (_paths[i]->eligible(now,_ackSendInterval) && slave->primary()) {
  1180. if (!_paths[i]->preferred()) {
  1181. _paths[i]->address().toString(curPathStr);
  1182. fprintf(stderr, "%llu AB: (initial) [%d] found path on primary slave, taking note in case we don't find a preferred path\n", ((now - RR->bc->getBondStartTime())), i);
  1183. nonPreferredPath = _paths[i];
  1184. bFoundPrimarySlave = true;
  1185. }
  1186. if (_paths[i]->preferred()) {
  1187. _abPath = _paths[i];
  1188. _abPath->address().toString(curPathStr);
  1189. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1190. if (slave) {
  1191. fprintf(stderr, "%llu AB: (initial) [%d] found preferred path %s on primary slave: %s\n", ((now - RR->bc->getBondStartTime())), i, curPathStr, slave->ifname().c_str());
  1192. }
  1193. bFoundPrimarySlave = true;
  1194. break;
  1195. }
  1196. }
  1197. }
  1198. if (_abPath) {
  1199. _abPath->address().toString(curPathStr);
  1200. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _abPath->localSocket());
  1201. if (slave) {
  1202. fprintf(stderr, "%llu AB: (initial) found preferred primary path: %s on %s\n", ((now - RR->bc->getBondStartTime())), curPathStr, slave->ifname().c_str());
  1203. }
  1204. }
  1205. else {
  1206. if (bFoundPrimarySlave && nonPreferredPath) {
  1207. fprintf(stderr, "%llu AB: (initial) found a non-preferred primary path\n", ((now - RR->bc->getBondStartTime())));
  1208. _abPath = nonPreferredPath;
  1209. }
  1210. }
  1211. if (!_abPath) {
  1212. fprintf(stderr, "%llu AB: (initial) designated primary slave is not yet ready\n", ((now - RR->bc->getBondStartTime())));
  1213. // TODO: Should fail-over to specified backup or just wait?
  1214. }
  1215. }
  1216. else if (!userHasSpecifiedPrimarySlave()) {
  1217. int _abIdx = ZT_MAX_PEER_NETWORK_PATHS;
  1218. fprintf(stderr, "%llu AB: (initial) user did not specify primary slave, just picking something\n", ((now - RR->bc->getBondStartTime())));
  1219. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1220. if (_paths[i] && _paths[i]->eligible(now,_ackSendInterval)) {
  1221. _abIdx = i;
  1222. break;
  1223. }
  1224. }
  1225. if (_abIdx == ZT_MAX_PEER_NETWORK_PATHS) {
  1226. fprintf(stderr, "%llu AB: (initial) unable to find a candidate next-best, no change\n", ((now - RR->bc->getBondStartTime())));
  1227. }
  1228. else {
  1229. _abPath = _paths[_abIdx];
  1230. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _abPath->localSocket());
  1231. if (slave) {
  1232. fprintf(stderr, "%llu AB: (initial) selected non-primary slave idx=%d, %s on %s\n", ((now - RR->bc->getBondStartTime())), _abIdx, pathStr, slave->ifname().c_str());
  1233. }
  1234. }
  1235. }
  1236. }
  1237. }
  1238. /**
  1239. * Update and maintain the active-backup failover queue
  1240. */
  1241. if (_abPath) {
  1242. // Don't worry about the failover queue until we have an active slave
  1243. // Remove ineligible paths from the failover slave queue
  1244. for (std::list<SharedPtr<Path> >::iterator it(_abFailoverQueue.begin()); it!=_abFailoverQueue.end();) {
  1245. if ((*it) && !(*it)->eligible(now,_ackSendInterval)) {
  1246. (*it)->address().toString(curPathStr);
  1247. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, (*it)->localSocket());
  1248. if (slave) {
  1249. fprintf(stderr, "%llu AB: (fq) %s on %s is now ineligible, removing from failover queue\n", ((now - RR->bc->getBondStartTime())), curPathStr, slave->ifname().c_str());
  1250. }
  1251. it = _abFailoverQueue.erase(it);
  1252. } else {
  1253. ++it;
  1254. }
  1255. }
  1256. /**
  1257. * Failover instructions were provided by user, build queue according those as well as IPv
  1258. * preference, disregarding performance.
  1259. */
  1260. if (userHasSpecifiedFailoverInstructions()) {
  1261. /**
  1262. * Clear failover scores
  1263. */
  1264. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1265. if (_paths[i]) {
  1266. _paths[i]->_failoverScore = 0;
  1267. }
  1268. }
  1269. //fprintf(stderr, "AB: (fq) user has specified specific failover instructions, will follow them.\n");
  1270. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1271. if (!_paths[i] || !_paths[i]->allowed() || !_paths[i]->eligible(now,_ackSendInterval)) {
  1272. continue;
  1273. }
  1274. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1275. _paths[i]->address().toString(pathStr);
  1276. int failoverScoreHandicap = _paths[i]->_failoverScore;
  1277. if (_paths[i]->preferred())
  1278. {
  1279. failoverScoreHandicap += ZT_MULTIPATH_FAILOVER_HANDICAP_PREFERRED;
  1280. //fprintf(stderr, "%s on %s ----> %d for preferred\n", pathStr, _paths[i]->ifname().c_str(), failoverScoreHandicap);
  1281. }
  1282. if (slave->primary()) {
  1283. // If using "optimize" primary reselect mode, ignore user slave designations
  1284. failoverScoreHandicap += ZT_MULTIPATH_FAILOVER_HANDICAP_PRIMARY;
  1285. //fprintf(stderr, "%s on %s ----> %d for primary\n", pathStr, _paths[i]->ifname().c_str(), failoverScoreHandicap);
  1286. }
  1287. if (!_paths[i]->_failoverScore) {
  1288. // If we didn't inherit a failover score from a "parent" that wants to use this path as a failover
  1289. int newHandicap = failoverScoreHandicap ? failoverScoreHandicap : _paths[i]->_allocation;
  1290. _paths[i]->_failoverScore = newHandicap;
  1291. //fprintf(stderr, "%s on %s ----> %d for allocation\n", pathStr, _paths[i]->ifname().c_str(), newHandicap);
  1292. }
  1293. SharedPtr<Slave> failoverSlave;
  1294. if (slave->failoverToSlave().length()) {
  1295. failoverSlave = RR->bc->getSlaveByName(_policyAlias, slave->failoverToSlave());
  1296. }
  1297. if (failoverSlave) {
  1298. for (int j=0; j<ZT_MAX_PEER_NETWORK_PATHS; j++) {
  1299. if (_paths[j] && getSlave(_paths[j]) == failoverSlave.ptr()) {
  1300. _paths[j]->address().toString(pathStr);
  1301. int inheritedHandicap = failoverScoreHandicap - 10;
  1302. int newHandicap = _paths[j]->_failoverScore > inheritedHandicap ? _paths[j]->_failoverScore : inheritedHandicap;
  1303. //fprintf(stderr, "\thanding down %s on %s ----> %d\n", pathStr, getSlave(_paths[j])->ifname().c_str(), newHandicap);
  1304. if (!_paths[j]->preferred()) {
  1305. newHandicap--;
  1306. }
  1307. _paths[j]->_failoverScore = newHandicap;
  1308. }
  1309. }
  1310. }
  1311. if (_paths[i].ptr() != _abPath.ptr()) {
  1312. bool bFoundPathInQueue = false;
  1313. for (std::list<SharedPtr<Path> >::iterator it(_abFailoverQueue.begin()); it!=_abFailoverQueue.end();++it) {
  1314. if (_paths[i].ptr() == (*it).ptr()) {
  1315. bFoundPathInQueue = true;
  1316. }
  1317. }
  1318. if (!bFoundPathInQueue) {
  1319. _paths[i]->address().toString(curPathStr);
  1320. fprintf(stderr, "%llu AB: (fq) [%d] added %s on %s to queue\n", ((now - RR->bc->getBondStartTime())), i, curPathStr, getSlave(_paths[i])->ifname().c_str());
  1321. _abFailoverQueue.push_front(_paths[i]);
  1322. }
  1323. }
  1324. }
  1325. }
  1326. /**
  1327. * No failover instructions provided by user, build queue according to performance
  1328. * and IPv preference.
  1329. */
  1330. else if (!userHasSpecifiedFailoverInstructions()) {
  1331. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1332. if (!_paths[i]
  1333. || !_paths[i]->allowed()
  1334. || !_paths[i]->eligible(now,_ackSendInterval)) {
  1335. continue;
  1336. }
  1337. int failoverScoreHandicap = 0;
  1338. if (_paths[i]->preferred()) {
  1339. failoverScoreHandicap = ZT_MULTIPATH_FAILOVER_HANDICAP_PREFERRED;
  1340. }
  1341. bool includeRefractoryPeriod = true;
  1342. if (!_paths[i]->eligible(now,includeRefractoryPeriod)) {
  1343. failoverScoreHandicap = -10000;
  1344. }
  1345. if (getSlave(_paths[i])->primary() && _abSlaveSelectMethod != ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE) {
  1346. // If using "optimize" primary reselect mode, ignore user slave designations
  1347. failoverScoreHandicap = ZT_MULTIPATH_FAILOVER_HANDICAP_PRIMARY;
  1348. }
  1349. if (_paths[i].ptr() == negotiatedPath.ptr()) {
  1350. _paths[i]->_negotiated = true;
  1351. failoverScoreHandicap = ZT_MULTIPATH_FAILOVER_HANDICAP_NEGOTIATED;
  1352. } else {
  1353. _paths[i]->_negotiated = false;
  1354. }
  1355. _paths[i]->_failoverScore = _paths[i]->_allocation + failoverScoreHandicap;
  1356. if (_paths[i].ptr() != _abPath.ptr()) {
  1357. bool bFoundPathInQueue = false;
  1358. for (std::list<SharedPtr<Path> >::iterator it(_abFailoverQueue.begin()); it!=_abFailoverQueue.end();++it) {
  1359. if (_paths[i].ptr() == (*it).ptr()) {
  1360. bFoundPathInQueue = true;
  1361. }
  1362. }
  1363. if (!bFoundPathInQueue) {
  1364. _paths[i]->address().toString(curPathStr);
  1365. fprintf(stderr, "%llu AB: (fq) [%d] added %s on %s to queue\n", ((now - RR->bc->getBondStartTime())), i, curPathStr, getSlave(_paths[i])->ifname().c_str());
  1366. _abFailoverQueue.push_front(_paths[i]);
  1367. }
  1368. }
  1369. }
  1370. }
  1371. _abFailoverQueue.sort(PathQualityComparator());
  1372. if (_abFailoverQueue.empty()) {
  1373. fprintf(stderr, "%llu AB: (fq) the failover queue is empty, the active-backup bond is no longer fault-tolerant\n", ((now - RR->bc->getBondStartTime())));
  1374. }
  1375. }
  1376. /**
  1377. * Short-circuit if we have no queued paths
  1378. */
  1379. if (_abFailoverQueue.empty()) {
  1380. return;
  1381. }
  1382. /**
  1383. * Fulfill primary reselect obligations
  1384. */
  1385. if (_abPath && !_abPath->eligible(now,_ackSendInterval)) { // Implicit ZT_MULTIPATH_RESELECTION_POLICY_FAILURE
  1386. _abPath->address().toString(curPathStr); fprintf(stderr, "%llu AB: (failure) failover event!, active backup path (%s) is no-longer eligible\n", ((now - RR->bc->getBondStartTime())), curPathStr);
  1387. if (!_abFailoverQueue.empty()) {
  1388. fprintf(stderr, "%llu AB: (failure) there are (%lu) slaves in queue to choose from...\n", ((now - RR->bc->getBondStartTime())), _abFailoverQueue.size());
  1389. dequeueNextActiveBackupPath(now);
  1390. _abPath->address().toString(curPathStr); fprintf(stderr, "%llu sAB: (failure) switched to %s on %s\n", ((now - RR->bc->getBondStartTime())), curPathStr, getSlave(_abPath)->ifname().c_str());
  1391. } else {
  1392. fprintf(stderr, "%llu AB: (failure) nothing available in the slave queue, doing nothing.\n", ((now - RR->bc->getBondStartTime())));
  1393. }
  1394. }
  1395. /**
  1396. * Detect change to prevent flopping during later optimization step.
  1397. */
  1398. if (prevActiveBackupPath != _abPath) {
  1399. _lastActiveBackupPathChange = now;
  1400. }
  1401. if (_abSlaveSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_ALWAYS) {
  1402. if (_abPath && !getSlave(_abPath)->primary()
  1403. && getSlave(_abFailoverQueue.front())->primary()) {
  1404. fprintf(stderr, "%llu AB: (always) switching to available primary\n", ((now - RR->bc->getBondStartTime())));
  1405. dequeueNextActiveBackupPath(now);
  1406. }
  1407. }
  1408. if (_abSlaveSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_BETTER) {
  1409. if (_abPath && !getSlave(_abPath)->primary()) {
  1410. fprintf(stderr, "%llu AB: (better) active backup has switched to \"better\" primary slave according to re-select policy.\n", ((now - RR->bc->getBondStartTime())));
  1411. if (getSlave(_abFailoverQueue.front())->primary()
  1412. && (_abFailoverQueue.front()->_failoverScore > _abPath->_failoverScore)) {
  1413. dequeueNextActiveBackupPath(now);
  1414. fprintf(stderr, "%llu AB: (better) switched back to user-defined primary\n", ((now - RR->bc->getBondStartTime())));
  1415. }
  1416. }
  1417. }
  1418. if (_abSlaveSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE && !_abFailoverQueue.empty()) {
  1419. /**
  1420. * Implement link negotiation that was previously-decided
  1421. */
  1422. if (_abFailoverQueue.front()->_negotiated) {
  1423. dequeueNextActiveBackupPath(now);
  1424. _abPath->address().toString(prevPathStr);
  1425. fprintf(stderr, "%llu AB: (optimize) switched to negotiated path %s on %s\n", ((now - RR->bc->getBondStartTime())), prevPathStr, getSlave(_abPath)->ifname().c_str());
  1426. _lastPathNegotiationCheck = now;
  1427. }
  1428. else {
  1429. // Try to find a better path and automatically switch to it -- not too often, though.
  1430. if ((now - _lastActiveBackupPathChange) > ZT_MULTIPATH_MIN_ACTIVE_BACKUP_AUTOFLOP_INTERVAL) {
  1431. if (!_abFailoverQueue.empty()) {
  1432. //fprintf(stderr, "AB: (optimize) there are (%d) slaves in queue to choose from...\n", _abFailoverQueue.size());
  1433. int newFScore = _abFailoverQueue.front()->_failoverScore;
  1434. int prevFScore = _abPath->_failoverScore;
  1435. // Establish a minimum switch threshold to prevent flapping
  1436. int failoverScoreDifference = _abFailoverQueue.front()->_failoverScore - _abPath->_failoverScore;
  1437. int thresholdQuantity = (ZT_MULTIPATH_ACTIVE_BACKUP_OPTIMIZE_MIN_THRESHOLD * (float)_abPath->_allocation);
  1438. if ((failoverScoreDifference > 0) && (failoverScoreDifference > thresholdQuantity)) {
  1439. SharedPtr<Path> oldPath = _abPath;
  1440. _abPath->address().toString(prevPathStr);
  1441. dequeueNextActiveBackupPath(now);
  1442. _abPath->address().toString(curPathStr);
  1443. fprintf(stderr, "%llu AB: (optimize) switched from %s on %s (fs=%d) to %s on %s (fs=%d)\n", ((now - RR->bc->getBondStartTime())), prevPathStr, getSlave(oldPath)->ifname().c_str(), prevFScore, curPathStr, getSlave(_abPath)->ifname().c_str(), newFScore);
  1444. }
  1445. }
  1446. }
  1447. }
  1448. }
  1449. }
  1450. void Bond::setReasonableDefaults(int policy)
  1451. {
  1452. // If invalid bonding policy, try default
  1453. int _defaultBondingPolicy = BondController::defaultBondingPolicy();
  1454. if (policy <= ZT_BONDING_POLICY_NONE || policy > ZT_BONDING_POLICY_BALANCE_AWARE) {
  1455. // If no default set, use NONE (effectively disabling this bond)
  1456. if (_defaultBondingPolicy < ZT_BONDING_POLICY_NONE || _defaultBondingPolicy > ZT_BONDING_POLICY_BALANCE_AWARE) {
  1457. _bondingPolicy= ZT_BONDING_POLICY_NONE;
  1458. }
  1459. _bondingPolicy= _defaultBondingPolicy;
  1460. } else {
  1461. _bondingPolicy= policy;
  1462. }
  1463. _downDelay = 0;
  1464. _upDelay = 0;
  1465. _allowFlowHashing=false;
  1466. _bondMonitorInterval=0;
  1467. _allowPathNegotiation=false;
  1468. _shouldCollectPathStatistics=false;
  1469. _lastPathNegotiationReceived=0;
  1470. _lastBackgroundTaskCheck=0;
  1471. _lastPathNegotiationCheck=0;
  1472. _lastFlowStatReset=0;
  1473. _lastFlowExpirationCheck=0;
  1474. _localUtility=0;
  1475. _numBondedPaths=0;
  1476. _rrPacketsSentOnCurrSlave=0;
  1477. _rrIdx=0;
  1478. _lastPathNegotiationReceived=0;
  1479. _pathNegotiationCutoffCount=0;
  1480. _lastFlowRebalance=0;
  1481. _totalBondUnderload = 0;
  1482. //_maxAcceptableLatency
  1483. _maxAcceptablePacketDelayVariance = 50;
  1484. _maxAcceptablePacketLossRatio = 0.10;
  1485. _maxAcceptablePacketErrorRatio = 0.10;
  1486. _userHasSpecifiedSlaveSpeeds=0;
  1487. _lastFrame=0;
  1488. // TODO: Remove
  1489. _header=false;
  1490. _lastLogTS = 0;
  1491. _lastPrintTS = 0;
  1492. /**
  1493. * Paths are actively monitored to provide a real-time quality/preference-ordered rapid failover queue.
  1494. */
  1495. switch (policy) {
  1496. case ZT_BONDING_POLICY_ACTIVE_BACKUP:
  1497. _failoverInterval = 5000;
  1498. _abSlaveSelectMethod = ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE;
  1499. _slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
  1500. _qualityWeights[ZT_QOS_LAT_IDX] = 0.2f;
  1501. _qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;
  1502. _qualityWeights[ZT_QOS_PDV_IDX] = 0.2f;
  1503. _qualityWeights[ZT_QOS_PLR_IDX] = 0.2f;
  1504. _qualityWeights[ZT_QOS_PER_IDX] = 0.2f;
  1505. _qualityWeights[ZT_QOS_THR_IDX] = 0.2f;
  1506. _qualityWeights[ZT_QOS_THM_IDX] = 0.0f;
  1507. _qualityWeights[ZT_QOS_THV_IDX] = 0.0f;
  1508. _qualityWeights[ZT_QOS_SCP_IDX] = 0.0f;
  1509. break;
  1510. /**
  1511. * All seemingly-alive paths are used. Paths are not actively monitored.
  1512. */
  1513. case ZT_BONDING_POLICY_BROADCAST:
  1514. _downDelay = 30000;
  1515. _upDelay = 0;
  1516. break;
  1517. /**
  1518. * Paths are monitored to determine when/if one needs to be added or removed from the rotation
  1519. */
  1520. case ZT_BONDING_POLICY_BALANCE_RR:
  1521. _failoverInterval = 5000;
  1522. _allowFlowHashing = false;
  1523. _packetsPerSlave = 512;
  1524. _slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
  1525. _qualityWeights[ZT_QOS_LAT_IDX] = 0.4f;
  1526. _qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;
  1527. _qualityWeights[ZT_QOS_PDV_IDX] = 0.2f;
  1528. _qualityWeights[ZT_QOS_PLR_IDX] = 0.1f;
  1529. _qualityWeights[ZT_QOS_PER_IDX] = 0.1f;
  1530. _qualityWeights[ZT_QOS_THR_IDX] = 0.1f;
  1531. _qualityWeights[ZT_QOS_THM_IDX] = 0.0f;
  1532. _qualityWeights[ZT_QOS_THV_IDX] = 0.0f;
  1533. _qualityWeights[ZT_QOS_SCP_IDX] = 0.0f;
  1534. break;
  1535. /**
  1536. * Path monitoring is used to determine the capacity of each
  1537. * path and where to place the next flow.
  1538. */
  1539. case ZT_BONDING_POLICY_BALANCE_XOR:
  1540. _failoverInterval = 5000;;
  1541. _upDelay=_bondMonitorInterval*2;
  1542. _allowFlowHashing = true;
  1543. _slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
  1544. _qualityWeights[ZT_QOS_LAT_IDX] = 0.4f;
  1545. _qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;
  1546. _qualityWeights[ZT_QOS_PDV_IDX] = 0.2f;
  1547. _qualityWeights[ZT_QOS_PLR_IDX] = 0.1f;
  1548. _qualityWeights[ZT_QOS_PER_IDX] = 0.1f;
  1549. _qualityWeights[ZT_QOS_THR_IDX] = 0.1f;
  1550. _qualityWeights[ZT_QOS_THM_IDX] = 0.0f;
  1551. _qualityWeights[ZT_QOS_THV_IDX] = 0.0f;
  1552. _qualityWeights[ZT_QOS_SCP_IDX] = 0.0f;
  1553. break;
  1554. /**
  1555. * Path monitoring is used to determine the capacity of each
  1556. * path and where to place the next flow. Additionally, re-shuffling
  1557. * of flows may take place.
  1558. */
  1559. case ZT_BONDING_POLICY_BALANCE_AWARE:
  1560. _failoverInterval = 3000;
  1561. _allowFlowHashing = true;
  1562. _slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
  1563. _qualityWeights[ZT_QOS_LAT_IDX] = 0.3f;
  1564. _qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;
  1565. _qualityWeights[ZT_QOS_PDV_IDX] = 0.1f;
  1566. _qualityWeights[ZT_QOS_PLR_IDX] = 0.1f;
  1567. _qualityWeights[ZT_QOS_PER_IDX] = 0.1f;
  1568. _qualityWeights[ZT_QOS_THR_IDX] = 0.0f;
  1569. _qualityWeights[ZT_QOS_THM_IDX] = 0.4f;
  1570. _qualityWeights[ZT_QOS_THV_IDX] = 0.0f;
  1571. _qualityWeights[ZT_QOS_SCP_IDX] = 0.0f;
  1572. break;
  1573. default:
  1574. break;
  1575. }
  1576. /**
  1577. * Timer geometries and counters
  1578. */
  1579. _bondMonitorInterval = _failoverInterval / 3;
  1580. _ackSendInterval = _failoverInterval;
  1581. _qualityEstimationInterval = _failoverInterval * 2;
  1582. _dynamicPathMonitorInterval = 0;
  1583. _downDelay=0;
  1584. _upDelay=0;
  1585. _ackCutoffCount = 0;
  1586. _lastAckRateCheck = 0;
  1587. _qosSendInterval = _bondMonitorInterval * 4;
  1588. _qosCutoffCount = 0;
  1589. _lastQoSRateCheck = 0;
  1590. throughputMeasurementInterval = _ackSendInterval * 2;
  1591. BondController::setMinReqPathMonitorInterval(_bondMonitorInterval);
  1592. _defaultPathRefractoryPeriod = 8000;
  1593. fprintf(stderr, "TIMERS: strat=%d, fi= %d, bmi= %d, qos= %d, ack= %d, estimateInt= %d, refractory= %d, ud= %d, dd= %d\n",
  1594. _slaveMonitorStrategy,
  1595. _failoverInterval,
  1596. _bondMonitorInterval,
  1597. _qosSendInterval,
  1598. _ackSendInterval,
  1599. _qualityEstimationInterval,
  1600. _defaultPathRefractoryPeriod,
  1601. _upDelay,
  1602. _downDelay);
  1603. _lastQualityEstimation=0;
  1604. }
  1605. void Bond::setUserQualityWeights(float weights[], int len)
  1606. {
  1607. if (len == ZT_QOS_WEIGHT_SIZE) {
  1608. float weightTotal = 0.0;
  1609. for (unsigned int i=0; i<ZT_QOS_WEIGHT_SIZE; ++i) {
  1610. weightTotal += weights[i];
  1611. }
  1612. if (weightTotal > 0.99 && weightTotal < 1.01) {
  1613. memcpy(_qualityWeights, weights, len * sizeof(float));
  1614. }
  1615. }
  1616. }
  1617. bool Bond::relevant() {
  1618. return _peer->identity().address().toInt() == 0x16a03a3d03
  1619. || _peer->identity().address().toInt() == 0x4410300d03
  1620. || _peer->identity().address().toInt() == 0x795cbf86fa;
  1621. }
  1622. SharedPtr<Slave> Bond::getSlave(const SharedPtr<Path>& path)
  1623. {
  1624. return RR->bc->getSlaveBySocket(_policyAlias, path->localSocket());
  1625. }
  1626. void Bond::dumpInfo(const int64_t now)
  1627. {
  1628. char pathStr[128];
  1629. //char oldPathStr[128];
  1630. char currPathStr[128];
  1631. if (!relevant()) {
  1632. return;
  1633. }
  1634. /*
  1635. fprintf(stderr, "---[ bp=%d, id=%llx, dd=%d, up=%d, pmi=%d, specifiedSlaves=%d, _specifiedPrimarySlave=%d, _specifiedFailInst=%d ]\n",
  1636. _policy, _peer->identity().address().toInt(), _downDelay, _upDelay, _monitorInterval, _userHasSpecifiedSlaves, _userHasSpecifiedPrimarySlave, _userHasSpecifiedFailoverInstructions);
  1637. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  1638. fprintf(stderr, "Paths (bp=%d, stats=%d, primaryReselect=%d) :\n",
  1639. _policy, _shouldCollectPathStatistics, _abSlaveSelectMethod);
  1640. }
  1641. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR
  1642. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR
  1643. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE) {
  1644. fprintf(stderr, "Paths (bp=%d, stats=%d, fh=%d) :\n",
  1645. _policy, _shouldCollectPathStatistics, _allowFlowHashing);
  1646. }*/
  1647. if ((now - _lastLogTS) < 1000) {
  1648. return;
  1649. }
  1650. _lastPrintTS = now;
  1651. _lastLogTS = now;
  1652. fprintf(stderr, "\n\n");
  1653. for(int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1654. if (_paths[i]) {
  1655. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1656. _paths[i]->address().toString(pathStr);
  1657. fprintf(stderr, " %2d: lat=%8.3f, ac=%3d, fail%5s, fscore=%6d, in=%7d, out=%7d, age=%7ld, ack=%7ld, ref=%6d, ls=%llx",
  1658. i,
  1659. _paths[i]->_latencyMean,
  1660. _paths[i]->_allocation,
  1661. slave->failoverToSlave().c_str(),
  1662. _paths[i]->_failoverScore,
  1663. _paths[i]->_packetsIn,
  1664. _paths[i]->_packetsOut,
  1665. (long)_paths[i]->age(now),
  1666. (long)_paths[i]->ackAge(now),
  1667. _paths[i]->_refractoryPeriod,
  1668. _paths[i]->localSocket()
  1669. );
  1670. if (slave->spare()) {
  1671. fprintf(stderr, " SPR.");
  1672. } else {
  1673. fprintf(stderr, " ");
  1674. }
  1675. if (slave->primary()) {
  1676. fprintf(stderr, " PRIM.");
  1677. } else {
  1678. fprintf(stderr, " ");
  1679. }
  1680. if (_paths[i]->allowed()) {
  1681. fprintf(stderr, " ALL.");
  1682. } else {
  1683. fprintf(stderr, " ");
  1684. }
  1685. if (_paths[i]->eligible(now,_ackSendInterval)) {
  1686. fprintf(stderr, " ELI.");
  1687. } else {
  1688. fprintf(stderr, " ");
  1689. }
  1690. if (_paths[i]->preferred()) {
  1691. fprintf(stderr, " PREF.");
  1692. } else {
  1693. fprintf(stderr, " ");
  1694. }
  1695. if (_paths[i]->_negotiated) {
  1696. fprintf(stderr, " NEG.");
  1697. } else {
  1698. fprintf(stderr, " ");
  1699. }
  1700. if (_paths[i]->bonded()) {
  1701. fprintf(stderr, " BOND ");
  1702. } else {
  1703. fprintf(stderr, " ");
  1704. }
  1705. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP && _abPath && (_abPath == _paths[i].ptr())) {
  1706. fprintf(stderr, " ACTIVE ");
  1707. } else if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  1708. fprintf(stderr, " ");
  1709. }
  1710. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP && _abFailoverQueue.size() && (_abFailoverQueue.front().ptr() == _paths[i].ptr())) {
  1711. fprintf(stderr, " NEXT ");
  1712. } else if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  1713. fprintf(stderr, " ");
  1714. }
  1715. fprintf(stderr, "%5s %s\n", slave->ifname().c_str(), pathStr);
  1716. }
  1717. }
  1718. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  1719. if (!_abFailoverQueue.empty()) {
  1720. fprintf(stderr, "\nFailover Queue:\n");
  1721. for (std::list<SharedPtr<Path> >::iterator it(_abFailoverQueue.begin()); it!=_abFailoverQueue.end();++it) {
  1722. (*it)->address().toString(currPathStr);
  1723. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, (*it)->localSocket());
  1724. fprintf(stderr, "\t%8s\tspeed=%7d\trelSpeed=%3d\tipvPref=%3d\tfscore=%9d\t\t%s\n",
  1725. slave->ifname().c_str(),
  1726. slave->speed(),
  1727. slave->relativeSpeed(),
  1728. slave->ipvPref(),
  1729. (*it)->_failoverScore,
  1730. currPathStr);
  1731. }
  1732. }
  1733. else
  1734. {
  1735. fprintf(stderr, "\nFailover Queue size = %lu\n", _abFailoverQueue.size());
  1736. }
  1737. }
  1738. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR
  1739. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR
  1740. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE) {
  1741. /*
  1742. if (_numBondedPaths) {
  1743. fprintf(stderr, "\nBonded Paths:\n");
  1744. for (int i=0; i<_numBondedPaths; ++i) {
  1745. _paths[_bondedIdx[i]].p->address().toString(currPathStr);
  1746. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[_bondedIdx[i]].p->localSocket());
  1747. fprintf(stderr, " [%d]\t%8s\tflows=%3d\tspeed=%7d\trelSpeed=%3d\tipvPref=%3d\tfscore=%9d\t\t%s\n", i,
  1748. //fprintf(stderr, " [%d]\t%8s\tspeed=%7d\trelSpeed=%3d\tflowCount=%2d\tipvPref=%3d\tfscore=%9d\t\t%s\n", i,
  1749. slave->ifname().c_str(),
  1750. numberOfAssignedFlows(_paths[_bondedIdx[i]].p),
  1751. slave->speed(),
  1752. slave->relativeSpeed(),
  1753. //_paths[_bondedIdx[i]].p->assignedFlows.size(),
  1754. slave->ipvPref(),
  1755. _paths[_bondedIdx[i]].p->failoverScore(),
  1756. currPathStr);
  1757. }
  1758. }
  1759. */
  1760. /*
  1761. if (_allowFlowHashing) {
  1762. //Mutex::Lock _l(_flows_m);
  1763. if (_flows.size()) {
  1764. fprintf(stderr, "\nFlows:\n");
  1765. std::map<int32_t,SharedPtr<Flow> >::iterator it = _flows.begin();
  1766. while (it != _flows.end()) {
  1767. it->second->assignedPath()->address().toString(currPathStr);
  1768. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, it->second->assignedPath()->localSocket());
  1769. fprintf(stderr, " [%4x] in=%16llu, out=%16llu, bytes=%16llu, last=%16llu, if=%8s\t\t%s\n",
  1770. it->second->id(),
  1771. it->second->bytesInPerUnitTime(),
  1772. it->second->bytesOutPerUnitTime(),
  1773. it->second->totalBytes(),
  1774. it->second->age(now),
  1775. slave->ifname().c_str(),
  1776. currPathStr);
  1777. ++it;
  1778. }
  1779. }
  1780. }
  1781. */
  1782. }
  1783. //fprintf(stderr, "\n\n\n\n\n");
  1784. }
  1785. } // namespace ZeroTier