RR(renv),

_peer(peer),

_qosCutoffCount(0),

_ackCutoffCount(0),

_lastAckRateCheck(0),

_lastQoSRateCheck(0),

_lastQualityEstimation(0),

_lastCheckUserPreferences(0),

_lastBackgroundTaskCheck(0),

_lastBondStatusLog(0),

_lastPathNegotiationReceived(0),

_lastPathNegotiationCheck(0),

_lastSentPathNegotiationRequest(0),

_lastFlowStatReset(0),

_lastFlowExpirationCheck(0),

_lastFlowRebalance(0),

_lastFrame(0),

_lastActiveBackupPathChange(0)

{

setReasonableDefaults(policy, SharedPtr<Bond>(), false);

_policyAlias = BondController::getPolicyStrByCode(policy);

RR(renv),

_policyAlias(policyAlias),

_peer(peer)

{

setReasonableDefaults(BondController::getPolicyCodeByStr(basePolicy), SharedPtr<Bond>(), false);

RR(renv),

_peer(peer),

_lastAckRateCheck(0),

_lastQoSRateCheck(0),

_lastQualityEstimation(0),

_lastCheckUserPreferences(0),

_lastBackgroundTaskCheck(0),

_lastBondStatusLog(0),

_lastPathNegotiationReceived(0),

_lastPathNegotiationCheck(0),

_lastFlowStatReset(0),

_lastFlowExpirationCheck(0),

_lastFlowRebalance(0),

_lastFrame(0)

{

setReasonableDefaults(originalBond->_bondingPolicy, originalBond, true);

{

char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

Mutex::Lock _l(_paths_m);

if (!RR->bc->linkAllowed(_policyAlias, getLink(path))) {

return;

}

bool alreadyPresent = false;

for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {

if (path.ptr() == _paths[i].ptr()) {

// Previously encountered path, not notifying bond

alreadyPresent = true;

break;

}

}

if (!alreadyPresent) {

for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {

if (!_paths[i]) {

_paths[i] = path;

sprintf(traceMsg, "%s (bond) Nominating link %s/%s to peer %llx. It has now entered its trial period",

OSUtils::humanReadableTimestamp().c_str(), getLink(path)->ifname().c_str(), pathStr, _peer->_id.address().toInt());

RR->t->bondStateMessage(NULL, traceMsg);

_paths[i]->startTrial(now);

break;

}

}

}

curateBond(now, true);

estimatePathQuality(now);

{

Mutex::Lock _l(_paths_m);

/**

if (_bondingPolicy == ZT_BONDING_POLICY_ACTIVE_BACKUP) {

if (_abPath) {

return _abPath;

}

}

/**

if (_bondingPolicy == ZT_BONDING_POLICY_BROADCAST) {

return SharedPtr<Path>(); // Handled in Switch::_trySend()

}

if (!_numBondedPaths) {

return SharedPtr<Path>(); // No paths assigned to bond yet, cannot balance traffic

}

/**

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_RR) {

if (!_allowFlowHashing) {

if (_packetsPerLink == 0) {

// Randomly select a path

return _paths[_bondedIdx[_freeRandomByte % _numBondedPaths]]; // TODO: Optimize

}

if (_rrPacketsSentOnCurrLink < _packetsPerLink) {

// Continue to use this link

++_rrPacketsSentOnCurrLink;

return _paths[_bondedIdx[_rrIdx]];

}

// Reset striping counter

_rrPacketsSentOnCurrLink = 0;

if (_numBondedPaths == 1) {

_rrIdx = 0;

}

else {

int _tempIdx = _rrIdx;

for (int searchCount = 0; searchCount < (_numBondedPaths-1); searchCount++) {

_tempIdx = (_tempIdx == (_numBondedPaths-1)) ? 0 : _tempIdx+1;

if (_bondedIdx[_tempIdx] != ZT_MAX_PEER_NETWORK_PATHS) {

if (_paths[_bondedIdx[_tempIdx]] && _paths[_bondedIdx[_tempIdx]]->eligible(now,_ackSendInterval)) {

_rrIdx = _tempIdx;

break;

}

}

}

}

if (_paths[_bondedIdx[_rrIdx]]) {

return _paths[_bondedIdx[_rrIdx]];

}

}

}

/**

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR || _bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE) {

if (!_allowFlowHashing || flowId == -1) {

// No specific path required for unclassified traffic, send on anything

return _paths[_bondedIdx[_freeRandomByte % _numBondedPaths]]; // TODO: Optimize

}

else if (_allowFlowHashing) {

// TODO: Optimize

Mutex::Lock _l(_flows_m);

SharedPtr<Flow> flow;

if (_flows.count(flowId)) {

flow = _flows[flowId];

flow->updateActivity(now);

}

else {

unsigned char entropy;

Utils::getSecureRandom(&entropy, 1);

flow = createFlow(SharedPtr<Path>(), flowId, entropy, now);

}

if (flow) {

return flow->assignedPath();

}

}

}

return SharedPtr<Path>();

{

//char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

//sprintf(traceMsg, "%s (qos) Invalid packet on link %s/%s from peer %llx",

// OSUtils::humanReadableTimestamp().c_str(), getLink(path)->ifname().c_str(), pathStr, _peer->_id.address().toInt());

//RR->t->bondStateMessage(NULL, traceMsg);

Mutex::Lock _l(_paths_m);

for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {

if (_paths[i] == path) {

_paths[i]->packetValiditySamples.push(false);

}

}

uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)

{

//char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

//sprintf(traceMsg, "%s (bond) Outgoing packet on link %s/%s to peer %llx",

// OSUtils::humanReadableTimestamp().c_str(), getLink(path)->ifname().c_str(), pathStr, _peer->_id.address().toInt());

//RR->t->bondStateMessage(NULL, traceMsg);

_freeRandomByte += (unsigned char)(packetId >> 8); // Grab entropy to use in path selection logic

if (!_shouldCollectPathStatistics) {

return;

}

bool isFrame = (verb == Packet::VERB_FRAME || verb == Packet::VERB_EXT_FRAME);

bool shouldRecord = (packetId & (ZT_QOS_ACK_DIVISOR - 1)

&& (verb != Packet::VERB_ACK)

&& (verb != Packet::VERB_QOS_MEASUREMENT));

if (isFrame || shouldRecord) {

Mutex::Lock _l(_paths_m);

if (isFrame) {

++(path->_packetsOut);

_lastFrame=now;

}

if (shouldRecord) {

path->_unackedBytes += payloadLength;

// Take note that we're expecting a VERB_ACK on this path as of a specific time

if (path->qosStatsOut.size() < ZT_QOS_MAX_OUTSTANDING_RECORDS) {

path->qosStatsOut[packetId] = now;

}

}

}

if (_allowFlowHashing && (flowId != ZT_QOS_NO_FLOW)) {

Mutex::Lock _l(_flows_m);

if (_flows.count(flowId)) {

_flows[flowId]->recordOutgoingBytes(payloadLength);

}

}

Packet::Verb verb, int32_t flowId, int64_t now)

{

//char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

//sprintf(traceMsg, "%s (bond) Incoming packet on link %s/%s from peer %llx [id=%llx, len=%d, verb=%d, flowId=%x]",

// OSUtils::humanReadableTimestamp().c_str(), getLink(path)->ifname().c_str(), pathStr, _peer->_id.address().toInt(), packetId, payloadLength, verb, flowId);

//RR->t->bondStateMessage(NULL, traceMsg);

bool isFrame = (verb == Packet::VERB_FRAME || verb == Packet::VERB_EXT_FRAME);

bool shouldRecord = (packetId & (ZT_QOS_ACK_DIVISOR - 1)

&& (verb != Packet::VERB_ACK)

&& (verb != Packet::VERB_QOS_MEASUREMENT));

if (isFrame || shouldRecord) {

Mutex::Lock _l(_paths_m);

if (isFrame) {

++(path->_packetsIn);

_lastFrame=now;

}

if (shouldRecord) {

path->ackStatsIn[packetId] = payloadLength;

++(path->_packetsReceivedSinceLastAck);

path->qosStatsIn[packetId] = now;

++(path->_packetsReceivedSinceLastQoS);

path->packetValiditySamples.push(true);

}

}

/**

if ((flowId != ZT_QOS_NO_FLOW)

&& (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_RR

|| _bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR

|| _bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE)) {

Mutex::Lock _l(_flows_m);

SharedPtr<Flow> flow;

if (!_flows.count(flowId)) {

flow = createFlow(path, flowId, 0, now);

} else {

flow = _flows[flowId];

}

if (flow) {

flow->recordIncomingBytes(payloadLength);

}

}

{

Mutex::Lock _l(_paths_m);

//char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

//sprintf(traceMsg, "%s (qos) Received QoS packet sampling %d frames from peer %llx via %s/%s",

// OSUtils::humanReadableTimestamp().c_str(), count, _peer->_id.address().toInt(), getLink(path)->ifname().c_str(), pathStr);

//RR->t->bondStateMessage(NULL, traceMsg);

// Look up egress times and compute latency values for each record

std::map<uint64_t,uint64_t>::iterator it;

for (int j=0; j<count; j++) {

it = path->qosStatsOut.find(rx_id[j]);

if (it != path->qosStatsOut.end()) {

path->latencySamples.push(((uint16_t)(now - it->second) - rx_ts[j]) / 2);

path->qosStatsOut.erase(it);

}

}

path->qosRecordSize.push(count);

{

Mutex::Lock _l(_paths_m);

//char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

//sprintf(traceMsg, "%s (qos) Received ACK packet for %d bytes from peer %llx via %s/%s",

// OSUtils::humanReadableTimestamp().c_str(), ackedBytes, _peer->_id.address().toInt(), getLink(path)->ifname().c_str(), pathStr);

//RR->t->bondStateMessage(NULL, traceMsg);

path->_lastAckReceived = now;

path->_unackedBytes = (ackedBytes > path->_unackedBytes) ? 0 : path->_unackedBytes - ackedBytes;

int64_t timeSinceThroughputEstimate = (now - path->_lastThroughputEstimation);

if (timeSinceThroughputEstimate >= throughputMeasurementInterval) {

// TODO: See if this floating point math can be reduced

uint64_t throughput = (uint64_t)((float)(path->_bytesAckedSinceLastThroughputEstimation) / ((float)timeSinceThroughputEstimate / (float)1000));

throughput /= 1000;

if (throughput > 0.0) {

path->throughputSamples.push(throughput);

path->_throughputMax = throughput > path->_throughputMax ? throughput : path->_throughputMax;

}

path->_lastThroughputEstimation = now;

path->_bytesAckedSinceLastThroughputEstimation = 0;

} else {

path->_bytesAckedSinceLastThroughputEstimation += ackedBytes;

}

{

int32_t len = 0;

std::map<uint64_t,uint64_t>::iterator it = path->qosStatsIn.begin();

int i=0;

int numRecords = std::min(path->_packetsReceivedSinceLastQoS,ZT_QOS_TABLE_SIZE);

while (i<numRecords && it != path->qosStatsIn.end()) {

uint64_t id = it->first;

memcpy(qosBuffer, &id, sizeof(uint64_t));

qosBuffer+=sizeof(uint64_t);

uint16_t holdingTime = (uint16_t)(now - it->second);

memcpy(qosBuffer, &holdingTime, sizeof(uint16_t));

qosBuffer+=sizeof(uint16_t);

len+=sizeof(uint64_t)+sizeof(uint16_t);

path->qosStatsIn.erase(it++);

++i;

}

return len;

{

char traceMsg[256];

char curPathStr[128];

unsigned int idx = ZT_MAX_PEER_NETWORK_PATHS;

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR) {

idx = abs((int)(flow->id() % (_numBondedPaths)));

SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, _paths[_bondedIdx[idx]]->localSocket());

_paths[_bondedIdx[idx]]->address().toString(curPathStr);

sprintf(traceMsg, "%s (balance-xor) Assigned outgoing flow %x to peer %llx to link %s/%s, %lu active flow(s)",

OSUtils::humanReadableTimestamp().c_str(), flow->id(), _peer->_id.address().toInt(), link->ifname().c_str(), curPathStr, _flows.size());

RR->t->bondStateMessage(NULL, traceMsg);

flow->assignPath(_paths[_bondedIdx[idx]],now);

++(_paths[_bondedIdx[idx]]->_assignedFlowCount);

}

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE) {

unsigned char entropy;

Utils::getSecureRandom(&entropy, 1);

if (_totalBondUnderload) {

entropy %= _totalBondUnderload;

}

if (!_numBondedPaths) {

sprintf(traceMsg, "%s (balance-aware) There are no bonded paths, cannot assign flow %x\n",

OSUtils::humanReadableTimestamp().c_str(), flow->id());

RR->t->bondStateMessage(NULL, traceMsg);

return false;

}

/* Since there may be scenarios where a path is removed before we can re-estimate

int totalIncompleteAllocation = 0;

for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {

if (_paths[i] && _paths[i]->bonded()) {

totalIncompleteAllocation += _paths[i]->_allocation;

}

}

entropy %= totalIncompleteAllocation;

for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {

if (_paths[i] && _paths[i]->bonded()) {

SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, _paths[i]->localSocket());

_paths[i]->address().toString(curPathStr);

uint8_t probabilitySegment = (_totalBondUnderload > 0) ? _paths[i]->_affinity : _paths[i]->_allocation;

if (entropy <= probabilitySegment) {

idx = i;

break;

}

entropy -= probabilitySegment;

}

}

if (idx < ZT_MAX_PEER_NETWORK_PATHS) {

if (flow->_assignedPath) {

flow->_previouslyAssignedPath = flow->_assignedPath;

}

flow->assignPath(_paths[idx],now);

++(_paths[idx]->_assignedFlowCount);

}

else {

fprintf(stderr, "could not assign flow?\n"); exit(0); // TODO: Remove for production

return false;

}

}

if (_bondingPolicy == ZT_BONDING_POLICY_ACTIVE_BACKUP) {

if (_abOverflowEnabled) {

flow->assignPath(_abPath, now);

} else {

sprintf(traceMsg, "%s (bond) Unable to assign outgoing flow %x to peer %llx, no active overflow link",

OSUtils::humanReadableTimestamp().c_str(), flow->id(), _peer->_id.address().toInt());

RR->t->bondStateMessage(NULL, traceMsg);

return false;

}

}

flow->assignedPath()->address().toString(curPathStr);

SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, flow->assignedPath()->localSocket());

sprintf(traceMsg, "%s (bond) Assigned outgoing flow %x to peer %llx to link %s/%s, %lu active flow(s)",

OSUtils::humanReadableTimestamp().c_str(), flow->id(), _peer->_id.address().toInt(), link->ifname().c_str(), curPathStr, _flows.size());

RR->t->bondStateMessage(NULL, traceMsg);

return true;

{

char traceMsg[256];

char curPathStr[128];

// ---

if (!_numBondedPaths) {

sprintf(traceMsg, "%s (bond) There are no bonded paths to peer %llx, cannot assign flow %x\n",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), flowId);

RR->t->bondStateMessage(NULL, traceMsg);

return SharedPtr<Flow>();

}

if (_flows.size() >= ZT_FLOW_MAX_COUNT) {

sprintf(traceMsg, "%s (bond) Maximum number of flows on bond to peer %llx reached (%d), forcibly forgetting oldest flow\n",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), ZT_FLOW_MAX_COUNT);

RR->t->bondStateMessage(NULL, traceMsg);

forgetFlowsWhenNecessary(0, true, now);

}

SharedPtr<Flow> flow = new Flow(flowId, now);

_flows[flowId] = flow;

/**

if (path) {

flow->assignPath(path,now);

path->address().toString(curPathStr);

path->_assignedFlowCount++;

SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, flow->assignedPath()->localSocket());

sprintf(traceMsg, "%s (bond) Assigned incoming flow %x from peer %llx to link %s/%s, %lu active flow(s)",

OSUtils::humanReadableTimestamp().c_str(), flow->id(), _peer->_id.address().toInt(), link->ifname().c_str(), curPathStr, _flows.size());

RR->t->bondStateMessage(NULL, traceMsg);

}

/**

else if (!path) {

assignFlowToBondedPath(flow, now);

}

return flow;

{

char traceMsg[256];

std::map<int32_t,SharedPtr<Flow> >::iterator it = _flows.begin();

std::map<int32_t,SharedPtr<Flow> >::iterator oldestFlow = _flows.end();

SharedPtr<Flow> expiredFlow;

if (age) { // Remove by specific age

while (it != _flows.end()) {

if (it->second->age(now) > age) {

sprintf(traceMsg, "%s (bond) Forgetting flow %x between this node and peer %llx, %lu active flow(s)",

OSUtils::humanReadableTimestamp().c_str(), it->first, _peer->_id.address().toInt(), (_flows.size()-1));

RR->t->bondStateMessage(NULL, traceMsg);

it->second->assignedPath()->_assignedFlowCount--;

it = _flows.erase(it);

} else {

++it;

}

}

}

else if (oldest) { // Remove single oldest by natural expiration

uint64_t maxAge = 0;

while (it != _flows.end()) {

if (it->second->age(now) > maxAge) {

maxAge = (now - it->second->age(now));

oldestFlow = it;

}

++it;

}

if (oldestFlow != _flows.end()) {

sprintf(traceMsg, "%s (bond) Forgetting oldest flow %x (of age %llu) between this node and peer %llx, %lu active flow(s)",

OSUtils::humanReadableTimestamp().c_str(), oldestFlow->first, oldestFlow->second->age(now), _peer->_id.address().toInt(), (_flows.size()-1));

RR->t->bondStateMessage(NULL, traceMsg);

oldestFlow->second->assignedPath()->_assignedFlowCount--;

_flows.erase(oldestFlow);

}

}

{

char traceMsg[256];

if (_abLinkSelectMethod != ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE) {

return;

}

Mutex::Lock _l(_paths_m);

char pathStr[128];

path->address().toString(pathStr);

if (!_lastPathNegotiationCheck) {

return;

}

SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, path->localSocket());

if (remoteUtility > _localUtility) {

char pathStr[128]; path->address().toString(pathStr);

sprintf(traceMsg, "%s (bond) Peer %llx suggests using alternate link %s/%s. Remote utility (%d) is GREATER than local utility (%d), switching to said link\n",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), link->ifname().c_str(), pathStr, remoteUtility, _localUtility);

RR->t->bondStateMessage(NULL, traceMsg);

negotiatedPath = path;

}

if (remoteUtility < _localUtility) {

sprintf(traceMsg, "%s (bond) Peer %llx suggests using alternate link %s/%s. Remote utility (%d) is LESS than local utility (%d), not switching\n",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), link->ifname().c_str(), pathStr, remoteUtility, _localUtility);

RR->t->bondStateMessage(NULL, traceMsg);

}

if (remoteUtility == _localUtility) {

sprintf(traceMsg, "%s (bond) Peer %llx suggests using alternate link %s/%s. Remote utility (%d) is equal to local utility (%d)\n",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), link->ifname().c_str(), pathStr, remoteUtility, _localUtility);

RR->t->bondStateMessage(NULL, traceMsg);

if (_peer->_id.address().toInt() > RR->node->identity().address().toInt()) {

sprintf(traceMsg, "%s (bond) Agreeing with peer %llx to use alternate link %s/%s\n",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), link->ifname().c_str(), pathStr);

RR->t->bondStateMessage(NULL, traceMsg);

negotiatedPath = path;

} else {

sprintf(traceMsg, "%s (bond) Ignoring petition from peer %llx to use alternate link %s/%s\n",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), link->ifname().c_str(), pathStr);

RR->t->bondStateMessage(NULL, traceMsg);

}

}

{

char pathStr[128];

int maxInPathIdx = ZT_MAX_PEER_NETWORK_PATHS;

int maxOutPathIdx = ZT_MAX_PEER_NETWORK_PATHS;

uint64_t maxInCount = 0;

uint64_t maxOutCount = 0;

for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {

if (!_paths[i]) {

continue;

}

if (_paths[i]->_packetsIn > maxInCount) {

maxInCount = _paths[i]->_packetsIn;

maxInPathIdx = i;

}

if (_paths[i]->_packetsOut > maxOutCount) {

maxOutCount = _paths[i]->_packetsOut;

maxOutPathIdx = i;

}

_paths[i]->resetPacketCounts();

}

bool _peerLinksSynchronized = ((maxInPathIdx != ZT_MAX_PEER_NETWORK_PATHS)

&& (maxOutPathIdx != ZT_MAX_PEER_NETWORK_PATHS)

&& (maxInPathIdx != maxOutPathIdx)) ? false : true;

/**

if (!_peerLinksSynchronized) {

_localUtility = _paths[maxOutPathIdx]->_failoverScore - _paths[maxInPathIdx]->_failoverScore;

if (_paths[maxOutPathIdx]->_negotiated) {

_localUtility -= ZT_MULTIPATH_FAILOVER_HANDICAP_NEGOTIATED;

}

if ((now - _lastSentPathNegotiationRequest) > ZT_PATH_NEGOTIATION_CUTOFF_TIME) {

//fprintf(stderr, "BT: (sync) it's been long enough, sending more requests.\n");

_numSentPathNegotiationRequests = 0;

}

if (_numSentPathNegotiationRequests < ZT_PATH_NEGOTIATION_TRY_COUNT) {

if (_localUtility >= 0) {

//fprintf(stderr, "BT: (sync) paths appear to be out of sync (utility=%d)\n", _localUtility);

sendPATH_NEGOTIATION_REQUEST(tPtr, _paths[maxOutPathIdx]);

++_numSentPathNegotiationRequests;

_lastSentPathNegotiationRequest = now;

_paths[maxOutPathIdx]->address().toString(pathStr);

SharedPtr<Link> link =RR->bc->getLinkBySocket(_policyAlias, _paths[maxOutPathIdx]->localSocket());

//fprintf(stderr, "sending request to use %s on %s, ls=%llx, utility=%d\n", pathStr, link->ifname().c_str(), _paths[maxOutPathIdx]->localSocket(), _localUtility);

}

}

/**

else if ((now - _lastSentPathNegotiationRequest) > (2 * ZT_PATH_NEGOTIATION_CHECK_INTERVAL)) {

if (_localUtility == 0) {

// There's no loss to us, just switch without sending a another request

//fprintf(stderr, "BT: (sync) giving up, switching to remote peer's path.\n");

negotiatedPath = _paths[maxInPathIdx];

}

}

}

{

char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

sprintf(traceMsg, "%s (bond) Sending link negotiation request to peer %llx via link %s/%s, local utility is %d",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), getLink(path)->ifname().c_str(), pathStr, _localUtility);

RR->t->bondStateMessage(NULL, traceMsg);

if (_abLinkSelectMethod != ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE) {

return;

}

Packet outp(_peer->_id.address(),RR->identity.address(),Packet::VERB_PATH_NEGOTIATION_REQUEST);

outp.append<int16_t>(_localUtility);

if (path->address()) {

outp.armor(_peer->key(),false,_peer->aesKeysIfSupported());

RR->node->putPacket(tPtr,path->localSocket(),path->address(),outp.data(),outp.size());

}

const InetAddress &atAddress,int64_t now)

{

Packet outp(_peer->_id.address(),RR->identity.address(),Packet::VERB_ACK);

int32_t bytesToAck = 0;

std::map<uint64_t,uint16_t>::iterator it = path->ackStatsIn.begin();

while (it != path->ackStatsIn.end()) {

bytesToAck += it->second;

++it;

}

//char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

//sprintf(traceMsg, "%s (qos) Sending ACK packet for %d bytes to peer %llx via link %s/%s",

// OSUtils::humanReadableTimestamp().c_str(), bytesToAck, _peer->_id.address().toInt(), getLink(path)->ifname().c_str(), pathStr);

//RR->t->bondStateMessage(NULL, traceMsg);

outp.append<uint32_t>(bytesToAck);

if (atAddress) {

outp.armor(_peer->key(),false,_peer->aesKeysIfSupported());

RR->node->putPacket(tPtr,localSocket,atAddress,outp.data(),outp.size());

} else {

RR->sw->send(tPtr,outp,false);

}

path->ackStatsIn.clear();

path->_packetsReceivedSinceLastAck = 0;

path->_lastAckSent = now;

const InetAddress &atAddress,int64_t now)

{

//char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);

//sprintf(traceMsg, "%s (qos) Sending QoS packet to peer %llx via link %s/%s",

// OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), getLink(path)->ifname().c_str(), pathStr);

//RR->t->bondStateMessage(NULL, traceMsg);

const int64_t _now = RR->node->now();

Packet outp(_peer->_id.address(),RR->identity.address(),Packet::VERB_QOS_MEASUREMENT);

char qosData[ZT_QOS_MAX_PACKET_SIZE];

int16_t len = generateQoSPacket(path, _now,qosData);

outp.append(qosData,len);

if (atAddress) {

outp.armor(_peer->key(),false,_peer->aesKeysIfSupported());

RR->node->putPacket(tPtr,localSocket,atAddress,outp.data(),outp.size());

} else {

RR->sw->send(tPtr,outp,false);

}

// Account for the fact that a VERB_QOS_MEASUREMENT was just sent. Reset timers.

path->_packetsReceivedSinceLastQoS = 0;

path->_lastQoSMeasurement = now;

{

Mutex::Lock _l(_paths_m);

if (!_peer->_canUseMultipath || (now - _lastBackgroundTaskCheck) < ZT_BOND_BACKGROUND_TASK_MIN_INTERVAL) {

return;

}

_lastBackgroundTaskCheck = now;

// Compute dynamic path monitor timer interval

if (_linkMonitorStrategy == ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC) {

int suggestedMonitorInterval = (now - _lastFrame) / 100;

_dynamicPathMonitorInterval = std::min(ZT_PATH_HEARTBEAT_PERIOD, ((suggestedMonitorInterval > _bondMonitorInterval) ? suggestedMonitorInterval : _bondMonitorInterval));

}

// TODO: Clarify and generalize this logic

if (_linkMonitorStrategy == ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_DYNAMIC) {

_shouldCollectPathStatistics = true;

}

// Memoize oft-used properties in the packet ingress/egress logic path

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE) {

// Required for real-time balancing

_shouldCollectPathStatistics = true;

}

if (_bondingPolicy == ZT_BONDING_POLICY_ACTIVE_BACKUP) {

if (_abLinkSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_BETTER) {

// Required for judging suitability of primary link after recovery

_shouldCollectPathStatistics = true;

}

if (_abLinkSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_OPTIMIZE) {

// Required for judging suitability of new candidate primary

_shouldCollectPathStatistics = true;

}

}

if ((now - _lastCheckUserPreferences) > 1000) {

_lastCheckUserPreferences = now;

applyUserPrefs();

}

curateBond(now,false);

if ((now - _lastQualityEstimation) > _qualityEstimationInterval) {

_lastQualityEstimation = now;

estimatePathQuality(now);

}

dumpInfo(now);

// Send QOS/ACK packets as needed

if (_shouldCollectPathStatistics) {

for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {

if (_paths[i] && _paths[i]->allowed()) {

if (_paths[i]->needsToSendQoS(now,_qosSendInterval)) {

sendQOS_MEASUREMENT(tPtr, _paths[i], _paths[i]->localSocket(), _paths[i]->address(), now);

}

if (_paths[i]->needsToSendAck(now,_ackSendInterval)) {

sendACK(tPtr, _paths[i], _paths[i]->localSocket(), _paths[i]->address(), now);

}

}

}

}

// Perform periodic background tasks unique to each bonding policy

switch (_bondingPolicy)

{

case ZT_BONDING_POLICY_ACTIVE_BACKUP:

processActiveBackupTasks(tPtr, now);

break;

case ZT_BONDING_POLICY_BROADCAST:

break;

case ZT_BONDING_POLICY_BALANCE_RR:

case ZT_BONDING_POLICY_BALANCE_XOR:

case ZT_BONDING_POLICY_BALANCE_AWARE:

processBalanceTasks(now);

break;

default:

break;

}

// Check whether or not a path negotiation needs to be performed

if (((now - _lastPathNegotiationCheck) > ZT_PATH_NEGOTIATION_CHECK_INTERVAL) && _allowPathNegotiation) {

_lastPathNegotiationCheck = now;

pathNegotiationCheck(tPtr, now);

}

{

for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {

if (!_paths[i]) {

continue;

}

SharedPtr<Link> sl = getLink(_paths[i]);

if (sl) {

if (sl->monitorInterval() == 0) { // If no interval was specified for this link, use more generic bond-wide interval

sl->setMonitorInterval(_bondMonitorInterval);

}

RR->bc->setMinReqPathMonitorInterval((sl->monitorInterval() < RR->bc->minReqPathMonitorInterval()) ? sl->monitorInterval() : RR->bc->minReqPathMonitorInterval());

bool bFoundCommonLink = false;

SharedPtr<Link> commonLink =RR->bc->getLinkBySocket(_policyAlias, _paths[i]->localSocket());

for(unsigned int j=0;j<ZT_MAX_PEER_NETWORK_PATHS;++j) {

if (_paths[j] && _paths[j].ptr() != _paths[i].ptr()) {

if (RR->bc->getLinkBySocket(_policyAlias, _paths[j]->localSocket()) == commonLink) {

bFoundCommonLink = true;

}

}

}

_paths[i]->_monitorInterval = sl->monitorInterval();

_paths[i]->_upDelay = sl->upDelay() ? sl->upDelay() : _upDelay;

_paths[i]->_downDelay = sl->downDelay() ? sl->downDelay() : _downDelay;

_paths[i]->_ipvPref = sl->ipvPref();

_paths[i]->_mode = sl->mode();

_paths[i]->_enabled = sl->enabled();

_paths[i]->_onlyPathOnLink = !bFoundCommonLink;

}

}

if (_peer) {

_peer->_shouldCollectPathStatistics = _shouldCollectPathStatistics;

_peer->_bondingPolicy = _bondingPolicy;

}

{

char traceMsg[256];

char pathStr[128];

uint8_t tmpNumAliveLinks = 0;

uint8_t tmpNumTotalLinks = 0;

/**

for(unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {

if (!_paths[i]) {

continue;

}

tmpNumTotalLinks++;

if (_paths[i]->alive(now, true)) {

tmpNumAliveLinks++;

}

bool currEligibility = _paths[i]->eligible(now,_ackSendInterval);

if (currEligibility != _paths[i]->_lastEligibilityState) {

_paths[i]->address().toString(pathStr);

char traceMsg[256]; _paths[i]->address().toString(pathStr);

sprintf(traceMsg, "%s (bond) Eligibility of link %s/%s to peer %llx has changed from %d to %d",

OSUtils::humanReadableTimestamp().c_str(), getLink(_paths[i])->ifname().c_str(), pathStr, _peer->_id.address().toInt(), _paths[i]->_lastEligibilityState, currEligibility);

RR->t->bondStateMessage(NULL, traceMsg);

if (currEligibility) {

rebuildBond = true;

}

if (!currEligibility) {

_paths[i]->adjustRefractoryPeriod(now, _defaultPathRefractoryPeriod, !currEligibility);

if (_paths[i]->bonded()) {

char pathStr[128]; _paths[i]->address().toString(pathStr);

sprintf(traceMsg, "%s (bond) Link %s/%s to peer %llx was bonded, reallocation of its flows will occur soon",

OSUtils::humanReadableTimestamp().c_str(), getLink(_paths[i])->ifname().c_str(), pathStr, _peer->_id.address().toInt());

RR->t->bondStateMessage(NULL, traceMsg);

rebuildBond = true;

_paths[i]->_shouldReallocateFlows = _paths[i]->bonded();

_paths[i]->setBonded(false);

} else {

sprintf(traceMsg, "%s (bond) Link %s/%s to peer %llx was not bonded, no allocation consequences",

OSUtils::humanReadableTimestamp().c_str(), getLink(_paths[i])->ifname().c_str(), pathStr, _peer->_id.address().toInt());

RR->t->bondStateMessage(NULL, traceMsg);

}

}

}

if (currEligibility) {

_paths[i]->adjustRefractoryPeriod(now, _defaultPathRefractoryPeriod, false);

}

_paths[i]->_lastEligibilityState = currEligibility;

}

_numAliveLinks = tmpNumAliveLinks;

_numTotalLinks = tmpNumTotalLinks;

/* Determine health status to report to user */

bool tmpHealthStatus = true;

if (_bondingPolicy == ZT_BONDING_POLICY_ACTIVE_BACKUP) {

if (_numAliveLinks < 2) {

// Considered healthy if there is at least one failover link

tmpHealthStatus = false;

}

}

if (_bondingPolicy == ZT_BONDING_POLICY_BROADCAST) {

if (_numAliveLinks < 1) {

// Considerd healthy if we're able to send frames at all

tmpHealthStatus = false;

}

}

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_RR) {

if (_numAliveLinks < _numTotalLinks) {

// Considerd healthy if all known paths are alive, this should be refined to account for user bond config settings

tmpHealthStatus = false;

}

}

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR) {

if (_numAliveLinks < _numTotalLinks) {

// Considerd healthy if all known paths are alive, this should be refined to account for user bond config settings

tmpHealthStatus = false;

}

}

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE) {

if (_numAliveLinks < _numTotalLinks) {

// Considerd healthy if all known paths are alive, this should be refined to account for user bond config settings

tmpHealthStatus = false;

}

}

if (tmpHealthStatus != _isHealthy) {

std::string healthStatusStr;

if (tmpHealthStatus == true) {

healthStatusStr = "HEALTHY";

} else {

healthStatusStr = "DEGRADED";

}

sprintf(traceMsg, "%s (bond) Bond to peer %llx is in a %s state (%d/%d links)",

OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), healthStatusStr.c_str(), _numAliveLinks, _numTotalLinks);

RR->t->bondStateMessage(NULL, traceMsg);

}

_isHealthy = tmpHealthStatus;

/**

if ((_bondingPolicy == ZT_BONDING_POLICY_BALANCE_RR)

|| (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR)

|| (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE)) {

if (!_numBondedPaths) {

rebuildBond = true;

}

// TODO: Optimize

if (rebuildBond) {

int updatedBondedPathCount = 0;

std::map<SharedPtr<Link>,int> linkMap;

for (int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {

if (_paths[i] && _paths[i]->allowed() && (_paths[i]->eligible(now,_ackSendInterval) || !_numBondedPaths)) {

SharedPtr<Link> link =RR->bc->getLinkBySocket(_policyAlias, _paths[i]->localSocket());

if (!linkMap.count(link)) {

linkMap[link] = i;

}

else {

bool overriden = false;

_paths[i]->address().toString(pathStr);

//fprintf(stderr, " link representative path already exists! (%s %s)\n", getLink(_paths[i])->ifname().c_str(), pathStr);

if (_paths[i]->preferred() && !_paths[linkMap[link]]->preferred()) {

// Override previous choice if preferred

if (_paths[linkMap[link]]->_assignedFlowCount) {

_paths[linkMap[link]]->_deprecated = true;

}

else {

_paths[linkMap[link]]->_deprecated = true;

_paths[linkMap[link]]->setBonded(false);

}

linkMap[link] = i;

overriden = true;

}

if ((_paths[i]->preferred() && _paths[linkMap[link]]->preferred())

|| (!_paths[i]->preferred() && !_paths[linkMap[link]]->preferred())) {

if (_paths[i]->preferenceRank() > _paths[linkMap[link]]->preferenceRank()) {

// Override if higher preference

if (_paths[linkMap[link]]->_assignedFlowCount) {

_paths[linkMap[link]]->_deprecated = true;

}

else {

_paths[linkMap[link]]->_deprecated = true;

_paths[linkMap[link]]->setBonded(false);

}

linkMap[link] = i;

}

}

}

}

}

std::map<SharedPtr<Link>,int>::iterator it = linkMap.begin();

for (int i=0; i<ZT_MAX_PEER_NETWORK_PATHS; ++i) {

if (!_paths[i]) {

continue;

}

_bondedIdx[i] = ZT_MAX_PEER_NETWORK_PATHS;

if (it != linkMap.end()) {

_bondedIdx[i] = it->second;

_paths[_bondedIdx[i]]->setBonded(true);

++it;

++updatedBondedPathCount;

_paths[_bondedIdx[i]]->address().toString(pathStr);

//fprintf(stderr, "setting i=%d, _bondedIdx[%d]=%d to bonded (%s %s)\n", i, i, _bondedIdx[i], getLink(_paths[_bondedIdx[i]])->ifname().c_str(), pathStr);

}

}

_numBondedPaths = updatedBondedPathCount;

if (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_RR) {

// Cause a RR reset since the currently used index might no longer be valid

_rrPacketsSentOnCurrLink = _packetsPerLink;

}

}

}

{

char pathStr[128];

uint32_t totUserSpecifiedLinkSpeed = 0;

if (_numBondedPaths) { // Compute relative user-specified speeds of links

for(unsigned int i=0;i<_numBondedPaths;++i) {

SharedPtr<Link> link =RR->bc->getLinkBySocket(_policyAlias, _paths[i]->localSocket());

if (_paths[i] && _paths[i]->allowed()) {

totUserSpecifiedLinkSpeed += link->speed();

}

}

for(unsigned int i=0;i<_numBondedPaths;++i) {

SharedPtr<Link> link =RR->bc->getLinkBySocket(_policyAlias, _paths[i]->localSocket());

if (_paths[i] && _paths[i]->allowed()) {

link->setRelativeSpeed(round( ((float)link->speed() / (float)totUserSpecifiedLinkSpeed) * 255));

}

}

}

float lat[ZT_MAX_PEER_NETWORK_PATHS];

float pdv[ZT_MAX_PEER_NETWORK_PATHS];

float plr[ZT_MAX_PEER_NETWORK_PATHS];

float per[ZT_MAX_PEER_NETWORK_PATHS];