Formatting

node/Bond.cpp

@@ -140,7 +140,7 @@ SharedPtr<Path> Bond::getAppropriatePath(int64_t now, int32_t flowId)
 					}
 				}
 			}
-			//fprintf(stderr, "resultant _rrIdx=%d\n", _rrIdx);
+			fprintf(stderr, "_rrIdx=%d\n", _rrIdx);
 			if (_paths[_bondedIdx[_rrIdx]]) {
 				return _paths[_bondedIdx[_rrIdx]];
 			}
@@ -1022,7 +1022,144 @@ void Bond::estimatePathQuality(const int64_t now)
 
 void Bond::processBalanceTasks(const int64_t now)
 {
-	// Omitted
+	//fprintf(stderr, "processBalanceTasks\n");
+	char curPathStr[128];
+	if (_allowFlowHashing) {
+		/**
+		 * Clean up and reset flows if necessary
+		 */
+		if ((now - _lastFlowExpirationCheck) > ZT_MULTIPATH_FLOW_CHECK_INTERVAL) {
+			Mutex::Lock _l(_flows_m);
+			forgetFlowsWhenNecessary(ZT_MULTIPATH_FLOW_EXPIRATION_INTERVAL,false,now);
+			_lastFlowExpirationCheck = now;
+		}
+		if ((now - _lastFlowStatReset) > ZT_FLOW_STATS_RESET_INTERVAL) {
+			Mutex::Lock _l(_flows_m);
+			_lastFlowStatReset = now;
+			std::map<int32_t,SharedPtr<Flow> >::iterator it = _flows.begin();
+			while (it != _flows.end()) {
+				it->second->resetByteCounts();
+				++it;
+			}
+		}
+		/**
+		 * Re-allocate flows from dead paths
+		 */
+		if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR || _bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE) {
+			Mutex::Lock _l(_flows_m);
+			for (int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
+				if (!_paths[i]) {
+					continue;
+				}
+				if (!_paths[i]->eligible(now,_ackSendInterval) && _paths[i]->_shouldReallocateFlows) {
+					_paths[i]->address().toString(curPathStr);
+					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());
+					std::map<int32_t,SharedPtr<Flow> >::iterator flow_it = _flows.begin();
+					while (flow_it != _flows.end()) {
+						if (flow_it->second->assignedPath() == _paths[i]) {
+							if(assignFlowToBondedPath(flow_it->second, now)) {
+								_paths[i]->_assignedFlowCount--;
+							}
+						}
+						++flow_it;
+					}
+					_paths[i]->_shouldReallocateFlows = false;
+				}
+			}
+		}
+	}
+	/**
+	 * Tasks specific to (Balance Round Robin)
+	 */
+	if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_RR) {
+		if (_allowFlowHashing) {
+			// TODO: Should ideally failover from (idx) to a random slave, this is so that (idx+1) isn't overloaded
+		}
+		else if (!_allowFlowHashing) {
+			// Nothing
+		}
+	}
+	/**
+	 * Tasks specific to (Balance XOR)
+	 */
+	if (_bondingPolicy== ZT_BONDING_POLICY_BALANCE_XOR) {
+		// Nothing specific for XOR
+	}
+	/**
+	 * Tasks specific to (Balance Aware)
+	 */
+	if ((_bondingPolicy== ZT_BONDING_POLICY_BALANCE_AWARE)) {
+		if (_allowFlowHashing) {
+			Mutex::Lock _l(_flows_m);
+			/**
+			 * Re-balance flows in proportion to slave capacity (or when eligibility changes)
+			 */
+			if ((now - _lastFlowRebalance) > ZT_FLOW_REBALANCE_INTERVAL) {
+				/**
+				 * Determine "load" for bonded paths
+				 */
+				uint64_t totalBytes = 0;
+				for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) { // first pass: compute absolute byte load and total
+					if (_paths[i] && _paths[i]->bonded()) {
+						_paths[i]->_byteLoad = 0;
+						std::map<int32_t,SharedPtr<Flow> >::iterator flow_it = _flows.begin();
+						while (flow_it != _flows.end()) {
+							if (flow_it->second->assignedPath() == _paths[i]) {
+								_paths[i]->_byteLoad += flow_it->second->totalBytes();
+							}
+							++flow_it;
+						}
+						totalBytes += _paths[i]->_byteLoad;
+					}
+				}
+				/**
+				 * Determine "affinity" for bonded path
+				 */
+				//fprintf(stderr, "\n\n");
+				_totalBondUnderload = 0;
+
+				for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) { // second pass: compute relative byte loads and total imbalance
+					if (_paths[i] && _paths[i]->bonded()) {
+						if (totalBytes) {
+							uint8_t relativeByteLoad = std::ceil(((float)_paths[i]->_byteLoad / (float)totalBytes) * (float)255);
+							//fprintf(stderr, "lastComputedAllocation  = %d\n", _paths[i]->allocation);
+							//fprintf(stderr, "  relativeByteLoad      = %d\n", relativeByteLoad);
+							_paths[i]->_relativeByteLoad = relativeByteLoad;
+							uint8_t relativeUnderload = std::max(0, (int)_paths[i]->_allocation - (int)relativeByteLoad);
+							//fprintf(stderr, "    relativeUnderload  = %d\n", relativeUnderload);
+							_totalBondUnderload += relativeUnderload;
+							//fprintf(stderr, "    _totalBondUnderload = %d\n\n", _totalBondUnderload);
+							//_paths[i]->affinity = (relativeUnderload > 0 ? relativeUnderload : _paths[i]->_allocation);
+						}
+						else { // set everything to base values
+							_totalBondUnderload = 0;
+							//_paths[i]->affinity = 0;
+						}
+					}
+				}
+
+				//fprintf(stderr, "_totalBondUnderload=%d (end)\n\n", _totalBondUnderload);
+
+				/**
+				 *
+				 */
+				//fprintf(stderr, "_lastFlowRebalance\n");
+				std::map<int32_t, SharedPtr<Flow> >::iterator it = _flows.begin();
+				while (it != _flows.end()) {
+					int32_t flowId = it->first;
+					SharedPtr<Flow> flow = it->second;
+					if ((now - flow->_lastPathReassignment) > ZT_FLOW_MIN_REBALANCE_INTERVAL) {
+						//fprintf(stdout, "  could move : %x\n", flowId);
+					}
+					++it;
+				}
+				_lastFlowRebalance = now;
+			}
+		}
+		else if (!_allowFlowHashing) {
+			// Nothing
+		}
+	}
 }
 
 void Bond::dequeueNextActiveBackupPath(const uint64_t now)
@@ -1445,7 +1582,7 @@ void Bond::setReasonableDefaults(int policy)
 		case ZT_BONDING_POLICY_BALANCE_RR:
 			_failoverInterval = 5000;
 			_allowFlowHashing = false;
-			_packetsPerSlave = 8;
+			_packetsPerSlave = 512;
 			_slaveMonitorStrategy = ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC;
 			_qualityWeights[ZT_QOS_LAT_IDX] = 0.4f;
 			_qualityWeights[ZT_QOS_LTM_IDX] = 0.0f;

node/Path.hpp

@@ -28,7 +28,6 @@
 #include "Utils.hpp"
 #include "Packet.hpp"
 #include "RingBuffer.hpp"
-//#include "Bond.hpp"
 
 #include "../osdep/Slave.hpp"
 
@@ -48,7 +47,6 @@ class Path
 {
 	friend class SharedPtr<Path>;
 	friend class Bond;
-	//friend class SharedPtr<Bond>;
 
 public:
 	/**

osdep/Slave.hpp

@@ -230,7 +230,6 @@ private:
 	bool _isUserSpecified;
 
 	AtomicCounter __refCount;
-
 };
 
 } // namespace ZeroTier