Bond.cpp 66 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730
  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, "resultant _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. // Omitted
  982. }
  983. void Bond::dequeueNextActiveBackupPath(const uint64_t now)
  984. {
  985. //fprintf(stderr, "dequeueNextActiveBackupPath\n");
  986. if (_abFailoverQueue.empty()) {
  987. return;
  988. }
  989. _abPath = _abFailoverQueue.front();
  990. _abFailoverQueue.pop_front();
  991. _lastActiveBackupPathChange = now;
  992. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  993. if (_paths[i]) {
  994. _paths[i]->resetPacketCounts();
  995. }
  996. }
  997. }
  998. void Bond::processActiveBackupTasks(const int64_t now)
  999. {
  1000. //fprintf(stderr, "%llu processActiveBackupTasks\n", (now - RR->bc->getBondStartTime()));
  1001. char pathStr[128]; char prevPathStr[128]; char curPathStr[128];
  1002. SharedPtr<Path> prevActiveBackupPath = _abPath;
  1003. SharedPtr<Path> nonPreferredPath;
  1004. bool bFoundPrimarySlave = false;
  1005. /**
  1006. * Select initial "active" active-backup slave
  1007. */
  1008. if (!_abPath) {
  1009. fprintf(stderr, "%llu no active backup path yet...\n", ((now - RR->bc->getBondStartTime())));
  1010. /**
  1011. * [Automatic mode]
  1012. * The user has not explicitly specified slaves or their failover schedule,
  1013. * the bonding policy will now select the first eligible path and set it as
  1014. * its active backup path, if a substantially better path is detected the bonding
  1015. * policy will assign it as the new active backup path. If the path fails it will
  1016. * simply find the next eligible path.
  1017. */
  1018. if (!userHasSpecifiedSlaves()) {
  1019. fprintf(stderr, "%llu AB: (auto) user did not specify any slaves. waiting until we know more\n", ((now - RR->bc->getBondStartTime())));
  1020. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1021. if (_paths[i] && _paths[i]->eligible(now,_ackSendInterval)) {
  1022. _paths[i]->address().toString(curPathStr);
  1023. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1024. if (slave) {
  1025. fprintf(stderr, "%llu AB: (initial) [%d] found eligible path %s on: %s\n", ((now - RR->bc->getBondStartTime())), i, curPathStr, slave->ifname().c_str());
  1026. }
  1027. _abPath = _paths[i];
  1028. break;
  1029. }
  1030. }
  1031. }
  1032. /**
  1033. * [Manual mode]
  1034. * The user has specified slaves or failover rules that the bonding policy should adhere to.
  1035. */
  1036. else if (userHasSpecifiedSlaves()) {
  1037. fprintf(stderr, "%llu AB: (manual) no active backup slave, checking local.conf\n", ((now - RR->bc->getBondStartTime())));
  1038. if (userHasSpecifiedPrimarySlave()) {
  1039. fprintf(stderr, "%llu AB: (manual) user has specified primary slave, looking for it.\n", ((now - RR->bc->getBondStartTime())));
  1040. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1041. if (!_paths[i]) {
  1042. continue;
  1043. }
  1044. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1045. if (_paths[i]->eligible(now,_ackSendInterval) && slave->primary()) {
  1046. if (!_paths[i]->preferred()) {
  1047. _paths[i]->address().toString(curPathStr);
  1048. 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);
  1049. nonPreferredPath = _paths[i];
  1050. bFoundPrimarySlave = true;
  1051. }
  1052. if (_paths[i]->preferred()) {
  1053. _abPath = _paths[i];
  1054. _abPath->address().toString(curPathStr);
  1055. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1056. if (slave) {
  1057. 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());
  1058. }
  1059. bFoundPrimarySlave = true;
  1060. break;
  1061. }
  1062. }
  1063. }
  1064. if (_abPath) {
  1065. _abPath->address().toString(curPathStr);
  1066. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _abPath->localSocket());
  1067. if (slave) {
  1068. fprintf(stderr, "%llu AB: (initial) found preferred primary path: %s on %s\n", ((now - RR->bc->getBondStartTime())), curPathStr, slave->ifname().c_str());
  1069. }
  1070. }
  1071. else {
  1072. if (bFoundPrimarySlave && nonPreferredPath) {
  1073. fprintf(stderr, "%llu AB: (initial) found a non-preferred primary path\n", ((now - RR->bc->getBondStartTime())));
  1074. _abPath = nonPreferredPath;
  1075. }
  1076. }
  1077. if (!_abPath) {
  1078. fprintf(stderr, "%llu AB: (initial) designated primary slave is not yet ready\n", ((now - RR->bc->getBondStartTime())));
  1079. // TODO: Should fail-over to specified backup or just wait?
  1080. }
  1081. }
  1082. else if (!userHasSpecifiedPrimarySlave()) {
  1083. int _abIdx = ZT_MAX_PEER_NETWORK_PATHS;
  1084. fprintf(stderr, "%llu AB: (initial) user did not specify primary slave, just picking something\n", ((now - RR->bc->getBondStartTime())));
  1085. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1086. if (_paths[i] && _paths[i]->eligible(now,_ackSendInterval)) {
  1087. _abIdx = i;
  1088. break;
  1089. }
  1090. }
  1091. if (_abIdx == ZT_MAX_PEER_NETWORK_PATHS) {
  1092. fprintf(stderr, "%llu AB: (initial) unable to find a candidate next-best, no change\n", ((now - RR->bc->getBondStartTime())));
  1093. }
  1094. else {
  1095. _abPath = _paths[_abIdx];
  1096. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _abPath->localSocket());
  1097. if (slave) {
  1098. 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());
  1099. }
  1100. }
  1101. }
  1102. }
  1103. }
  1104. /**
  1105. * Update and maintain the active-backup failover queue
  1106. */
  1107. if (_abPath) {
  1108. // Don't worry about the failover queue until we have an active slave
  1109. // Remove ineligible paths from the failover slave queue
  1110. for (std::list<SharedPtr<Path> >::iterator it(_abFailoverQueue.begin()); it!=_abFailoverQueue.end();) {
  1111. if ((*it) && !(*it)->eligible(now,_ackSendInterval)) {
  1112. (*it)->address().toString(curPathStr);
  1113. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, (*it)->localSocket());
  1114. if (slave) {
  1115. 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());
  1116. }
  1117. it = _abFailoverQueue.erase(it);
  1118. } else {
  1119. ++it;
  1120. }
  1121. }
  1122. /**
  1123. * Failover instructions were provided by user, build queue according those as well as IPv
  1124. * preference, disregarding performance.
  1125. */
  1126. if (userHasSpecifiedFailoverInstructions()) {
  1127. /**
  1128. * Clear failover scores
  1129. */
  1130. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1131. if (_paths[i]) {
  1132. _paths[i]->_failoverScore = 0;
  1133. }
  1134. }
  1135. //fprintf(stderr, "AB: (fq) user has specified specific failover instructions, will follow them.\n");
  1136. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1137. if (!_paths[i] || !_paths[i]->allowed() || !_paths[i]->eligible(now,_ackSendInterval)) {
  1138. continue;
  1139. }
  1140. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1141. _paths[i]->address().toString(pathStr);
  1142. int failoverScoreHandicap = _paths[i]->_failoverScore;
  1143. if (_paths[i]->preferred())
  1144. {
  1145. failoverScoreHandicap += ZT_MULTIPATH_FAILOVER_HANDICAP_PREFERRED;
  1146. //fprintf(stderr, "%s on %s ----> %d for preferred\n", pathStr, _paths[i]->ifname().c_str(), failoverScoreHandicap);
  1147. }
  1148. if (slave->primary()) {
  1149. // If using "optimize" primary reselect mode, ignore user slave designations
  1150. failoverScoreHandicap += ZT_MULTIPATH_FAILOVER_HANDICAP_PRIMARY;
  1151. //fprintf(stderr, "%s on %s ----> %d for primary\n", pathStr, _paths[i]->ifname().c_str(), failoverScoreHandicap);
  1152. }
  1153. if (!_paths[i]->_failoverScore) {
  1154. // If we didn't inherit a failover score from a "parent" that wants to use this path as a failover
  1155. int newHandicap = failoverScoreHandicap ? failoverScoreHandicap : _paths[i]->_allocation;
  1156. _paths[i]->_failoverScore = newHandicap;
  1157. //fprintf(stderr, "%s on %s ----> %d for allocation\n", pathStr, _paths[i]->ifname().c_str(), newHandicap);
  1158. }
  1159. SharedPtr<Slave> failoverSlave;
  1160. if (slave->failoverToSlave().length()) {
  1161. failoverSlave = RR->bc->getSlaveByName(_policyAlias, slave->failoverToSlave());
  1162. }
  1163. if (failoverSlave) {
  1164. for (int j=0; j<ZT_MAX_PEER_NETWORK_PATHS; j++) {
  1165. if (_paths[j] && getSlave(_paths[j]) == failoverSlave.ptr()) {
  1166. _paths[j]->address().toString(pathStr);
  1167. int inheritedHandicap = failoverScoreHandicap - 10;
  1168. int newHandicap = _paths[j]->_failoverScore > inheritedHandicap ? _paths[j]->_failoverScore : inheritedHandicap;
  1169. //fprintf(stderr, "\thanding down %s on %s ----> %d\n", pathStr, getSlave(_paths[j])->ifname().c_str(), newHandicap);
  1170. if (!_paths[j]->preferred()) {
  1171. newHandicap--;
  1172. }
  1173. _paths[j]->_failoverScore = newHandicap;
  1174. }
  1175. }
  1176. }
  1177. if (_paths[i].ptr() != _abPath.ptr()) {
  1178. bool bFoundPathInQueue = false;
  1179. for (std::list<SharedPtr<Path> >::iterator it(_abFailoverQueue.begin()); it!=_abFailoverQueue.end();++it) {
  1180. if (_paths[i].ptr() == (*it).ptr()) {
  1181. bFoundPathInQueue = true;
  1182. }
  1183. }
  1184. if (!bFoundPathInQueue) {
  1185. _paths[i]->address().toString(curPathStr);
  1186. 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());
  1187. _abFailoverQueue.push_front(_paths[i]);
  1188. }
  1189. }
  1190. }
  1191. }
  1192. /**
  1193. * No failover instructions provided by user, build queue according to performance
  1194. * and IPv preference.
  1195. */
  1196. else if (!userHasSpecifiedFailoverInstructions()) {
  1197. for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1198. if (!_paths[i]
  1199. || !_paths[i]->allowed()
  1200. || !_paths[i]->eligible(now,_ackSendInterval)) {
  1201. continue;
  1202. }
  1203. int failoverScoreHandicap = 0;
  1204. if (_paths[i]->preferred()) {
  1205. failoverScoreHandicap = ZT_MULTIPATH_FAILOVER_HANDICAP_PREFERRED;
  1206. }
  1207. bool includeRefractoryPeriod = true;
  1208. if (!_paths[i]->eligible(now,includeRefractoryPeriod)) {
  1209. failoverScoreHandicap = -10000;
  1210. }
  1211. if (getSlave(_paths[i])->primary() && _abSlaveSelectMethod != ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE) {
  1212. // If using "optimize" primary reselect mode, ignore user slave designations
  1213. failoverScoreHandicap = ZT_MULTIPATH_FAILOVER_HANDICAP_PRIMARY;
  1214. }
  1215. if (_paths[i].ptr() == negotiatedPath.ptr()) {
  1216. _paths[i]->_negotiated = true;
  1217. failoverScoreHandicap = ZT_MULTIPATH_FAILOVER_HANDICAP_NEGOTIATED;
  1218. } else {
  1219. _paths[i]->_negotiated = false;
  1220. }
  1221. _paths[i]->_failoverScore = _paths[i]->_allocation + failoverScoreHandicap;
  1222. if (_paths[i].ptr() != _abPath.ptr()) {
  1223. bool bFoundPathInQueue = false;
  1224. for (std::list<SharedPtr<Path> >::iterator it(_abFailoverQueue.begin()); it!=_abFailoverQueue.end();++it) {
  1225. if (_paths[i].ptr() == (*it).ptr()) {
  1226. bFoundPathInQueue = true;
  1227. }
  1228. }
  1229. if (!bFoundPathInQueue) {
  1230. _paths[i]->address().toString(curPathStr);
  1231. 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());
  1232. _abFailoverQueue.push_front(_paths[i]);
  1233. }
  1234. }
  1235. }
  1236. }
  1237. _abFailoverQueue.sort(PathQualityComparator());
  1238. if (_abFailoverQueue.empty()) {
  1239. fprintf(stderr, "%llu AB: (fq) the failover queue is empty, the active-backup bond is no longer fault-tolerant\n", ((now - RR->bc->getBondStartTime())));
  1240. }
  1241. }
  1242. /**
  1243. * Short-circuit if we have no queued paths
  1244. */
  1245. if (_abFailoverQueue.empty()) {
  1246. return;
  1247. }
  1248. /**
  1249. * Fulfill primary reselect obligations
  1250. */
  1251. if (_abPath && !_abPath->eligible(now,_ackSendInterval)) { // Implicit ZT_MULTIPATH_RESELECTION_POLICY_FAILURE
  1252. _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);
  1253. if (!_abFailoverQueue.empty()) {
  1254. fprintf(stderr, "%llu AB: (failure) there are (%lu) slaves in queue to choose from...\n", ((now - RR->bc->getBondStartTime())), _abFailoverQueue.size());
  1255. dequeueNextActiveBackupPath(now);
  1256. _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());
  1257. } else {
  1258. fprintf(stderr, "%llu AB: (failure) nothing available in the slave queue, doing nothing.\n", ((now - RR->bc->getBondStartTime())));
  1259. }
  1260. }
  1261. /**
  1262. * Detect change to prevent flopping during later optimization step.
  1263. */
  1264. if (prevActiveBackupPath != _abPath) {
  1265. _lastActiveBackupPathChange = now;
  1266. }
  1267. if (_abSlaveSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_ALWAYS) {
  1268. if (_abPath && !getSlave(_abPath)->primary()
  1269. && getSlave(_abFailoverQueue.front())->primary()) {
  1270. fprintf(stderr, "%llu AB: (always) switching to available primary\n", ((now - RR->bc->getBondStartTime())));
  1271. dequeueNextActiveBackupPath(now);
  1272. }
  1273. }
  1274. if (_abSlaveSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_BETTER) {
  1275. if (_abPath && !getSlave(_abPath)->primary()) {
  1276. fprintf(stderr, "%llu AB: (better) active backup has switched to \"better\" primary slave according to re-select policy.\n", ((now - RR->bc->getBondStartTime())));
  1277. if (getSlave(_abFailoverQueue.front())->primary()
  1278. && (_abFailoverQueue.front()->_failoverScore > _abPath->_failoverScore)) {
  1279. dequeueNextActiveBackupPath(now);
  1280. fprintf(stderr, "%llu AB: (better) switched back to user-defined primary\n", ((now - RR->bc->getBondStartTime())));
  1281. }
  1282. }
  1283. }
  1284. if (_abSlaveSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE && !_abFailoverQueue.empty()) {
  1285. /**
  1286. * Implement link negotiation that was previously-decided
  1287. */
  1288. if (_abFailoverQueue.front()->_negotiated) {
  1289. dequeueNextActiveBackupPath(now);
  1290. _abPath->address().toString(prevPathStr);
  1291. fprintf(stderr, "%llu AB: (optimize) switched to negotiated path %s on %s\n", ((now - RR->bc->getBondStartTime())), prevPathStr, getSlave(_abPath)->ifname().c_str());
  1292. _lastPathNegotiationCheck = now;
  1293. }
  1294. else {
  1295. // Try to find a better path and automatically switch to it -- not too often, though.
  1296. if ((now - _lastActiveBackupPathChange) > ZT_MULTIPATH_MIN_ACTIVE_BACKUP_AUTOFLOP_INTERVAL) {
  1297. if (!_abFailoverQueue.empty()) {
  1298. //fprintf(stderr, "AB: (optimize) there are (%d) slaves in queue to choose from...\n", _abFailoverQueue.size());
  1299. int newFScore = _abFailoverQueue.front()->_failoverScore;
  1300. int prevFScore = _abPath->_failoverScore;
  1301. // Establish a minimum switch threshold to prevent flapping
  1302. int failoverScoreDifference = _abFailoverQueue.front()->_failoverScore - _abPath->_failoverScore;
  1303. int thresholdQuantity = (ZT_MULTIPATH_ACTIVE_BACKUP_OPTIMIZE_MIN_THRESHOLD * (float)_abPath->_allocation);
  1304. if ((failoverScoreDifference > 0) && (failoverScoreDifference > thresholdQuantity)) {
  1305. SharedPtr<Path> oldPath = _abPath;
  1306. _abPath->address().toString(prevPathStr);
  1307. dequeueNextActiveBackupPath(now);
  1308. _abPath->address().toString(curPathStr);
  1309. 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);
  1310. }
  1311. }
  1312. }
  1313. }
  1314. }
  1315. }
  1316. void Bond::setReasonableDefaults(int policy)
  1317. {
  1318. // If invalid bonding policy, try default
  1319. int _defaultBondingPolicy = BondController::defaultBondingPolicy();
  1320. if (policy <= ZT_BONDING_POLICY_NONE || policy > ZT_BONDING_POLICY_BALANCE_AWARE) {
  1321. // If no default set, use NONE (effectively disabling this bond)
  1322. if (_defaultBondingPolicy < ZT_BONDING_POLICY_NONE || _defaultBondingPolicy > ZT_BONDING_POLICY_BALANCE_AWARE) {
  1323. _bondingPolicy= ZT_BONDING_POLICY_NONE;
  1324. }
  1325. _bondingPolicy= _defaultBondingPolicy;
  1326. } else {
  1327. _bondingPolicy= policy;
  1328. }
  1329. _downDelay = 0;
  1330. _upDelay = 0;
  1331. _allowFlowHashing=false;
  1332. _bondMonitorInterval=0;
  1333. _allowPathNegotiation=false;
  1334. _shouldCollectPathStatistics=false;
  1335. _lastPathNegotiationReceived=0;
  1336. _lastBackgroundTaskCheck=0;
  1337. _lastPathNegotiationCheck=0;
  1338. _lastFlowStatReset=0;
  1339. _lastFlowExpirationCheck=0;
  1340. _localUtility=0;
  1341. _numBondedPaths=0;
  1342. _rrPacketsSentOnCurrSlave=0;
  1343. _rrIdx=0;
  1344. _lastPathNegotiationReceived=0;
  1345. _pathNegotiationCutoffCount=0;
  1346. _lastFlowRebalance=0;
  1347. _totalBondUnderload = 0;
  1348. //_maxAcceptableLatency
  1349. _maxAcceptablePacketDelayVariance = 50;
  1350. _maxAcceptablePacketLossRatio = 0.10;
  1351. _maxAcceptablePacketErrorRatio = 0.10;
  1352. _userHasSpecifiedSlaveSpeeds=0;
  1353. _lastFrame=0;
  1354. // TODO: Remove
  1355. _header=false;
  1356. _lastLogTS = 0;
  1357. _lastPrintTS = 0;
  1358. /**
  1359. * Paths are actively monitored to provide a real-time quality/preference-ordered rapid failover queue.
  1360. */
  1361. switch (policy) {
  1362. case ZT_BONDING_POLICY_ACTIVE_BACKUP:
  1363. _failoverInterval = 5000;
  1364. _abSlaveSelectMethod = ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE;
  1365. _slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
  1366. _qualityWeights[ZT_QOS_LAT_IDX] = 0.2f;
  1367. _qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;
  1368. _qualityWeights[ZT_QOS_PDV_IDX] = 0.2f;
  1369. _qualityWeights[ZT_QOS_PLR_IDX] = 0.2f;
  1370. _qualityWeights[ZT_QOS_PER_IDX] = 0.2f;
  1371. _qualityWeights[ZT_QOS_THR_IDX] = 0.2f;
  1372. _qualityWeights[ZT_QOS_THM_IDX] = 0.0f;
  1373. _qualityWeights[ZT_QOS_THV_IDX] = 0.0f;
  1374. _qualityWeights[ZT_QOS_SCP_IDX] = 0.0f;
  1375. break;
  1376. /**
  1377. * All seemingly-alive paths are used. Paths are not actively monitored.
  1378. */
  1379. case ZT_BONDING_POLICY_BROADCAST:
  1380. _downDelay = 30000;
  1381. _upDelay = 0;
  1382. break;
  1383. /**
  1384. * Paths are monitored to determine when/if one needs to be added or removed from the rotation
  1385. */
  1386. case ZT_BONDING_POLICY_BALANCE_RR:
  1387. _failoverInterval = 5000;
  1388. _allowFlowHashing = false;
  1389. _packetsPerSlave = 8;
  1390. _slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
  1391. _qualityWeights[ZT_QOS_LAT_IDX] = 0.4f;
  1392. _qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;
  1393. _qualityWeights[ZT_QOS_PDV_IDX] = 0.2f;
  1394. _qualityWeights[ZT_QOS_PLR_IDX] = 0.1f;
  1395. _qualityWeights[ZT_QOS_PER_IDX] = 0.1f;
  1396. _qualityWeights[ZT_QOS_THR_IDX] = 0.1f;
  1397. _qualityWeights[ZT_QOS_THM_IDX] = 0.0f;
  1398. _qualityWeights[ZT_QOS_THV_IDX] = 0.0f;
  1399. _qualityWeights[ZT_QOS_SCP_IDX] = 0.0f;
  1400. break;
  1401. /**
  1402. * Path monitoring is used to determine the capacity of each
  1403. * path and where to place the next flow.
  1404. */
  1405. case ZT_BONDING_POLICY_BALANCE_XOR:
  1406. _failoverInterval = 5000;;
  1407. _upDelay=_bondMonitorInterval*2;
  1408. _allowFlowHashing = true;
  1409. _slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
  1410. _qualityWeights[ZT_QOS_LAT_IDX] = 0.4f;
  1411. _qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;
  1412. _qualityWeights[ZT_QOS_PDV_IDX] = 0.2f;
  1413. _qualityWeights[ZT_QOS_PLR_IDX] = 0.1f;
  1414. _qualityWeights[ZT_QOS_PER_IDX] = 0.1f;
  1415. _qualityWeights[ZT_QOS_THR_IDX] = 0.1f;
  1416. _qualityWeights[ZT_QOS_THM_IDX] = 0.0f;
  1417. _qualityWeights[ZT_QOS_THV_IDX] = 0.0f;
  1418. _qualityWeights[ZT_QOS_SCP_IDX] = 0.0f;
  1419. break;
  1420. /**
  1421. * Path monitoring is used to determine the capacity of each
  1422. * path and where to place the next flow. Additionally, re-shuffling
  1423. * of flows may take place.
  1424. */
  1425. case ZT_BONDING_POLICY_BALANCE_AWARE:
  1426. _failoverInterval = 3000;
  1427. _allowFlowHashing = true;
  1428. _slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
  1429. _qualityWeights[ZT_QOS_LAT_IDX] = 0.3f;
  1430. _qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;
  1431. _qualityWeights[ZT_QOS_PDV_IDX] = 0.1f;
  1432. _qualityWeights[ZT_QOS_PLR_IDX] = 0.1f;
  1433. _qualityWeights[ZT_QOS_PER_IDX] = 0.1f;
  1434. _qualityWeights[ZT_QOS_THR_IDX] = 0.0f;
  1435. _qualityWeights[ZT_QOS_THM_IDX] = 0.4f;
  1436. _qualityWeights[ZT_QOS_THV_IDX] = 0.0f;
  1437. _qualityWeights[ZT_QOS_SCP_IDX] = 0.0f;
  1438. break;
  1439. default:
  1440. break;
  1441. }
  1442. /**
  1443. * Timer geometries and counters
  1444. */
  1445. _bondMonitorInterval = _failoverInterval / 3;
  1446. _ackSendInterval = _failoverInterval;
  1447. _qualityEstimationInterval = _failoverInterval * 2;
  1448. _dynamicPathMonitorInterval = 0;
  1449. _downDelay=0;
  1450. _upDelay=0;
  1451. _ackCutoffCount = 0;
  1452. _lastAckRateCheck = 0;
  1453. _qosSendInterval = _bondMonitorInterval * 4;
  1454. _qosCutoffCount = 0;
  1455. _lastQoSRateCheck = 0;
  1456. throughputMeasurementInterval = _ackSendInterval * 2;
  1457. BondController::setMinReqPathMonitorInterval(_bondMonitorInterval);
  1458. _defaultPathRefractoryPeriod = 8000;
  1459. fprintf(stderr, "TIMERS: strat=%d, fi= %d, bmi= %d, qos= %d, ack= %d, estimateInt= %d, refractory= %d, ud= %d, dd= %d\n",
  1460. _slaveMonitorStrategy,
  1461. _failoverInterval,
  1462. _bondMonitorInterval,
  1463. _qosSendInterval,
  1464. _ackSendInterval,
  1465. _qualityEstimationInterval,
  1466. _defaultPathRefractoryPeriod,
  1467. _upDelay,
  1468. _downDelay);
  1469. _lastQualityEstimation=0;
  1470. }
  1471. void Bond::setUserQualityWeights(float weights[], int len)
  1472. {
  1473. if (len == ZT_QOS_WEIGHT_SIZE) {
  1474. float weightTotal = 0.0;
  1475. for (unsigned int i=0; i<ZT_QOS_WEIGHT_SIZE; ++i) {
  1476. weightTotal += weights[i];
  1477. }
  1478. if (weightTotal > 0.99 && weightTotal < 1.01) {
  1479. memcpy(_qualityWeights, weights, len * sizeof(float));
  1480. }
  1481. }
  1482. }
  1483. bool Bond::relevant() {
  1484. return _peer->identity().address().toInt() == 0x16a03a3d03
  1485. || _peer->identity().address().toInt() == 0x4410300d03
  1486. || _peer->identity().address().toInt() == 0x795cbf86fa;
  1487. }
  1488. SharedPtr<Slave> Bond::getSlave(const SharedPtr<Path>& path)
  1489. {
  1490. return RR->bc->getSlaveBySocket(_policyAlias, path->localSocket());
  1491. }
  1492. void Bond::dumpInfo(const int64_t now)
  1493. {
  1494. char pathStr[128];
  1495. //char oldPathStr[128];
  1496. char currPathStr[128];
  1497. if (!relevant()) {
  1498. return;
  1499. }
  1500. /*
  1501. fprintf(stderr, "---[ bp=%d, id=%llx, dd=%d, up=%d, pmi=%d, specifiedSlaves=%d, _specifiedPrimarySlave=%d, _specifiedFailInst=%d ]\n",
  1502. _policy, _peer->identity().address().toInt(), _downDelay, _upDelay, _monitorInterval, _userHasSpecifiedSlaves, _userHasSpecifiedPrimarySlave, _userHasSpecifiedFailoverInstructions);
  1503. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  1504. fprintf(stderr, "Paths (bp=%d, stats=%d, primaryReselect=%d) :\n",
  1505. _policy, _shouldCollectPathStatistics, _abSlaveSelectMethod);
  1506. }
  1507. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR
  1508. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR
  1509. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE) {
  1510. fprintf(stderr, "Paths (bp=%d, stats=%d, fh=%d) :\n",
  1511. _policy, _shouldCollectPathStatistics, _allowFlowHashing);
  1512. }*/
  1513. if ((now - _lastLogTS) < 1000) {
  1514. return;
  1515. }
  1516. _lastPrintTS = now;
  1517. _lastLogTS = now;
  1518. fprintf(stderr, "\n\n");
  1519. for(int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {
  1520. if (_paths[i]) {
  1521. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[i]->localSocket());
  1522. _paths[i]->address().toString(pathStr);
  1523. fprintf(stderr, " %2d: lat=%8.3f, ac=%3d, fail%5s, fscore=%6d, in=%7d, out=%7d, age=%7ld, ack=%7ld, ref=%6d, ls=%llx",
  1524. i,
  1525. _paths[i]->_latencyMean,
  1526. _paths[i]->_allocation,
  1527. slave->failoverToSlave().c_str(),
  1528. _paths[i]->_failoverScore,
  1529. _paths[i]->_packetsIn,
  1530. _paths[i]->_packetsOut,
  1531. (long)_paths[i]->age(now),
  1532. (long)_paths[i]->ackAge(now),
  1533. _paths[i]->_refractoryPeriod,
  1534. _paths[i]->localSocket()
  1535. );
  1536. if (slave->spare()) {
  1537. fprintf(stderr, " SPR.");
  1538. } else {
  1539. fprintf(stderr, " ");
  1540. }
  1541. if (slave->primary()) {
  1542. fprintf(stderr, " PRIM.");
  1543. } else {
  1544. fprintf(stderr, " ");
  1545. }
  1546. if (_paths[i]->allowed()) {
  1547. fprintf(stderr, " ALL.");
  1548. } else {
  1549. fprintf(stderr, " ");
  1550. }
  1551. if (_paths[i]->eligible(now,_ackSendInterval)) {
  1552. fprintf(stderr, " ELI.");
  1553. } else {
  1554. fprintf(stderr, " ");
  1555. }
  1556. if (_paths[i]->preferred()) {
  1557. fprintf(stderr, " PREF.");
  1558. } else {
  1559. fprintf(stderr, " ");
  1560. }
  1561. if (_paths[i]->_negotiated) {
  1562. fprintf(stderr, " NEG.");
  1563. } else {
  1564. fprintf(stderr, " ");
  1565. }
  1566. if (_paths[i]->bonded()) {
  1567. fprintf(stderr, " BOND ");
  1568. } else {
  1569. fprintf(stderr, " ");
  1570. }
  1571. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP && _abPath && (_abPath == _paths[i].ptr())) {
  1572. fprintf(stderr, " ACTIVE ");
  1573. } else if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  1574. fprintf(stderr, " ");
  1575. }
  1576. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP && _abFailoverQueue.size() && (_abFailoverQueue.front().ptr() == _paths[i].ptr())) {
  1577. fprintf(stderr, " NEXT ");
  1578. } else if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  1579. fprintf(stderr, " ");
  1580. }
  1581. fprintf(stderr, "%5s %s\n", slave->ifname().c_str(), pathStr);
  1582. }
  1583. }
  1584. if (_bondingPolicy== ZT_BONDING_POLICY_ACTIVE_BACKUP) {
  1585. if (!_abFailoverQueue.empty()) {
  1586. fprintf(stderr, "\nFailover Queue:\n");
  1587. for (std::list<SharedPtr<Path> >::iterator it(_abFailoverQueue.begin()); it!=_abFailoverQueue.end();++it) {
  1588. (*it)->address().toString(currPathStr);
  1589. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, (*it)->localSocket());
  1590. fprintf(stderr, "\t%8s\tspeed=%7d\trelSpeed=%3d\tipvPref=%3d\tfscore=%9d\t\t%s\n",
  1591. slave->ifname().c_str(),
  1592. slave->speed(),
  1593. slave->relativeSpeed(),
  1594. slave->ipvPref(),
  1595. (*it)->_failoverScore,
  1596. currPathStr);
  1597. }
  1598. }
  1599. else
  1600. {
  1601. fprintf(stderr, "\nFailover Queue size = %lu\n", _abFailoverQueue.size());
  1602. }
  1603. }
  1604. if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR
  1605. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR
  1606. || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE) {
  1607. /*
  1608. if (_numBondedPaths) {
  1609. fprintf(stderr, "\nBonded Paths:\n");
  1610. for (int i=0; i<_numBondedPaths; ++i) {
  1611. _paths[_bondedIdx[i]].p->address().toString(currPathStr);
  1612. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, _paths[_bondedIdx[i]].p->localSocket());
  1613. fprintf(stderr, " [%d]\t%8s\tflows=%3d\tspeed=%7d\trelSpeed=%3d\tipvPref=%3d\tfscore=%9d\t\t%s\n", i,
  1614. //fprintf(stderr, " [%d]\t%8s\tspeed=%7d\trelSpeed=%3d\tflowCount=%2d\tipvPref=%3d\tfscore=%9d\t\t%s\n", i,
  1615. slave->ifname().c_str(),
  1616. numberOfAssignedFlows(_paths[_bondedIdx[i]].p),
  1617. slave->speed(),
  1618. slave->relativeSpeed(),
  1619. //_paths[_bondedIdx[i]].p->assignedFlows.size(),
  1620. slave->ipvPref(),
  1621. _paths[_bondedIdx[i]].p->failoverScore(),
  1622. currPathStr);
  1623. }
  1624. }
  1625. */
  1626. /*
  1627. if (_allowFlowHashing) {
  1628. //Mutex::Lock _l(_flows_m);
  1629. if (_flows.size()) {
  1630. fprintf(stderr, "\nFlows:\n");
  1631. std::map<int32_t,SharedPtr<Flow> >::iterator it = _flows.begin();
  1632. while (it != _flows.end()) {
  1633. it->second->assignedPath()->address().toString(currPathStr);
  1634. SharedPtr<Slave> slave =RR->bc->getSlaveBySocket(_policyAlias, it->second->assignedPath()->localSocket());
  1635. fprintf(stderr, " [%4x] in=%16llu, out=%16llu, bytes=%16llu, last=%16llu, if=%8s\t\t%s\n",
  1636. it->second->id(),
  1637. it->second->bytesInPerUnitTime(),
  1638. it->second->bytesOutPerUnitTime(),
  1639. it->second->totalBytes(),
  1640. it->second->age(now),
  1641. slave->ifname().c_str(),
  1642. currPathStr);
  1643. ++it;
  1644. }
  1645. }
  1646. }
  1647. */
  1648. }
  1649. //fprintf(stderr, "\n\n\n\n\n");
  1650. }
  1651. } // namespace ZeroTier