void/ZeroTierOne
2
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2025-06-05 20:13:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Match formatting of Bond-related sources to ZeroTier standard (no functional changes)

Browse source
This commit is contained in:
Joseph Henry 2021-05-03 17:59:31 -07:00
parent 63fd2cbaeb
commit 29e5880d8b
No known key found for this signature in database
GPG key ID: C45B33FF5EBC9344
6 changed files with 971 additions and 654 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

515
node/Bond.cpp
View file

@ -11,72 +11,68 @@
*/
/****/
#include <cmath>
#include "Bond.hpp"
#include "../osdep/OSUtils.hpp"
#include "Peer.hpp"
#include "Bond.hpp"
#include "Switch.hpp"
#include "Flow.hpp"
#include "Path.hpp"
#include <cmath>
namespace ZeroTier {
Bond::Bond(const RuntimeEnvironment *renv, int policy, const SharedPtr<Peer>& peer) :
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)
Bond::Bond(const RuntimeEnvironment* renv, int policy, const SharedPtr<Peer>& peer)
: 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);
}
Bond::Bond(const RuntimeEnvironment *renv, std::string& basePolicy, std::string& policyAlias, const SharedPtr<Peer>& peer) :
RR(renv),
_policyAlias(policyAlias),
_peer(peer)
Bond::Bond(const RuntimeEnvironment* renv, std::string& basePolicy, std::string& policyAlias, const SharedPtr<Peer>& peer) : RR(renv), _policyAlias(policyAlias), _peer(peer)
{
setReasonableDefaults(BondController::getPolicyCodeByStr(basePolicy), SharedPtr<Bond>(), false);
}
Bond::Bond(const RuntimeEnvironment *renv, SharedPtr<Bond> originalBond, const SharedPtr<Peer>& peer) :
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)
Bond::Bond(const RuntimeEnvironment* renv, SharedPtr<Bond> originalBond, const SharedPtr<Peer>& peer)
: 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);
}
void Bond::nominatePath(const SharedPtr<Path>& path, int64_t now)
{
char traceMsg[256]; char pathStr[128]; path->address().toString(pathStr);
char traceMsg[256];
char pathStr[128];
path->address().toString(pathStr);
Mutex::Lock _l(_paths_m);
if (! RR->bc->linkAllowed(_policyAlias, getLink(path))) {
return;
@ -93,8 +89,7 @@ void Bond::nominatePath(const SharedPtr<Path>& path, int64_t now)
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());
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;
@ -204,8 +199,7 @@ void Bond::recordIncomingInvalidPacket(const SharedPtr<Path>& path)
}
}
void Bond::recordOutgoingPacket(const SharedPtr<Path> &path, const uint64_t packetId,
uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)
void Bond::recordOutgoingPacket(const SharedPtr<Path>& path, const uint64_t packetId, 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",
@ -216,9 +210,7 @@ void Bond::recordOutgoingPacket(const SharedPtr<Path> &path, const uint64_t pack
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));
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) {
@ -241,17 +233,14 @@ void Bond::recordOutgoingPacket(const SharedPtr<Path> &path, const uint64_t pack
}
}
void Bond::recordIncomingPacket(const SharedPtr<Path>& path, uint64_t packetId, uint16_t payloadLength,
Packet::Verb verb, int32_t flowId, int64_t now)
void Bond::recordIncomingPacket(const SharedPtr<Path>& path, uint64_t packetId, uint16_t 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));
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) {
@ -271,15 +260,13 @@ void Bond::recordIncomingPacket(const SharedPtr<Path>& path, uint64_t packetId,
* that the next time we send a packet out that is part of a flow we know
* which path to use.
*/
if ((flowId != ZT_QOS_NO_FLOW)
&& (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_RR
|| _bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR
|| _bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE)) {
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 {
}
else {
flow = _flows[flowId];
}
if (flow) {
@ -327,7 +314,8 @@ void Bond::receivedAck(const SharedPtr<Path>& path, int64_t now, int32_t ackedBy
}
path->_lastThroughputEstimation = now;
path->_bytesAckedSinceLastThroughputEstimation = 0;
} else {
}
else {
path->_bytesAckedSinceLastThroughputEstimation += ackedBytes;
}
}
@ -361,8 +349,15 @@ bool Bond::assignFlowToBondedPath(SharedPtr<Flow> &flow, int64_t now)
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());
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);
@ -374,8 +369,7 @@ bool Bond::assignFlowToBondedPath(SharedPtr<Flow> &flow, int64_t now)
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());
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;
}
@ -410,24 +404,32 @@ bool Bond::assignFlowToBondedPath(SharedPtr<Flow> &flow, int64_t now)
++(_paths[idx]->_assignedFlowCount);
}
else {
fprintf(stderr, "could not assign flow?\n"); exit(0); // TODO: Remove for production
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());
}
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());
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;
}
@ -438,14 +440,12 @@ SharedPtr<Flow> Bond::createFlow(const SharedPtr<Path> &path, int32_t flowId, un
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);
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);
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);
}
@ -461,8 +461,15 @@ SharedPtr<Flow> Bond::createFlow(const SharedPtr<Path> &path, int32_t flowId, un
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());
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);
}
/**
@ -484,12 +491,12 @@ void Bond::forgetFlowsWhenNecessary(uint64_t age, bool oldest, int64_t now)
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));
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 {
}
else {
++it;
}
}
@ -504,8 +511,14 @@ void Bond::forgetFlowsWhenNecessary(uint64_t age, bool oldest, int64_t now)
++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));
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);
@ -527,29 +540,50 @@ void Bond::processIncomingPathNegotiationRequest(uint64_t now, SharedPtr<Path> &
}
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);
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);
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);
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);
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);
}
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);
}
}
@ -576,9 +610,7 @@ void Bond::pathNegotiationCheck(void *tPtr, const int64_t now)
}
_paths[i]->resetPacketCounts();
}
bool _peerLinksSynchronized = ((maxInPathIdx != ZT_MAX_PEER_NETWORK_PATHS)
&& (maxOutPathIdx != ZT_MAX_PEER_NETWORK_PATHS)
&& (maxInPathIdx != maxOutPathIdx)) ? false : true;
bool _peerLinksSynchronized = ((maxInPathIdx != ZT_MAX_PEER_NETWORK_PATHS) && (maxOutPathIdx != ZT_MAX_PEER_NETWORK_PATHS) && (maxInPathIdx != maxOutPathIdx)) ? false : true;
/**
* Determine utility and attempt to petition remote peer to switch to our chosen path
*/
@ -617,9 +649,17 @@ void Bond::pathNegotiationCheck(void *tPtr, const int64_t now)
void Bond::sendPATH_NEGOTIATION_REQUEST(void* tPtr, const SharedPtr<Path>& path)
{
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);
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;
@ -632,8 +672,7 @@ void Bond::sendPATH_NEGOTIATION_REQUEST(void *tPtr, const SharedPtr<Path> &path)
}
}
void Bond::sendACK(void *tPtr, const SharedPtr<Path> &path,const int64_t localSocket,
const InetAddress &atAddress,int64_t now)
void Bond::sendACK(void* tPtr, const SharedPtr<Path>& path, const int64_t localSocket, const InetAddress& atAddress, int64_t now)
{
Packet outp(_peer->_id.address(), RR->identity.address(), Packet::VERB_ACK);
int32_t bytesToAck = 0;
@ -650,7 +689,8 @@ void Bond::sendACK(void *tPtr, const SharedPtr<Path> &path,const int64_t localSo
if (atAddress) {
outp.armor(_peer->key(), false, _peer->aesKeysIfSupported());
RR->node->putPacket(tPtr, localSocket, atAddress, outp.data(), outp.size());
} else {
}
else {
RR->sw->send(tPtr, outp, false);
}
path->ackStatsIn.clear();
@ -658,8 +698,7 @@ void Bond::sendACK(void *tPtr, const SharedPtr<Path> &path,const int64_t localSo
path->_lastAckSent = now;
}
void Bond::sendQOS_MEASUREMENT(void *tPtr,const SharedPtr<Path> &path,const int64_t localSocket,
const InetAddress &atAddress,int64_t now)
void Bond::sendQOS_MEASUREMENT(void* tPtr, const SharedPtr<Path>& path, const int64_t localSocket, 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",
@ -673,7 +712,8 @@ void Bond::sendQOS_MEASUREMENT(void *tPtr,const SharedPtr<Path> &path,const int6
if (atAddress) {
outp.armor(_peer->key(), false, _peer->aesKeysIfSupported());
RR->node->putPacket(tPtr, localSocket, atAddress, outp.data(), outp.size());
} else {
}
else {
RR->sw->send(tPtr, outp, false);
}
// Account for the fact that a VERB_QOS_MEASUREMENT was just sent. Reset timers.
@ -740,8 +780,7 @@ void Bond::processBackgroundTasks(void *tPtr, const int64_t now)
}
}
// Perform periodic background tasks unique to each bonding policy
switch (_bondingPolicy)
{
switch (_bondingPolicy) {
case ZT_BONDING_POLICY_ACTIVE_BACKUP:
processActiveBackupTasks(tPtr, now);
break;
@ -818,9 +857,17 @@ void Bond::curateBond(const int64_t now, bool rebuildBond)
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);
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;
@ -828,16 +875,22 @@ void Bond::curateBond(const int64_t now, bool rebuildBond)
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());
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());
}
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);
}
}
@ -862,25 +915,25 @@ void Bond::curateBond(const int64_t now, bool rebuildBond)
}
if (_bondingPolicy == ZT_BONDING_POLICY_BROADCAST) {
if (_numAliveLinks < 1) {
// Considerd healthy if we're able to send frames at all
// Considered 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
// Considered 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
// Considered 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
// Considered healthy if all known paths are alive, this should be refined to account for user bond config settings
tmpHealthStatus = false;
}
}
@ -888,11 +941,11 @@ void Bond::curateBond(const int64_t now, bool rebuildBond)
std::string healthStatusStr;
if (tmpHealthStatus == true) {
healthStatusStr = "HEALTHY";
} else {
}
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);
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);
}
@ -902,9 +955,7 @@ void Bond::curateBond(const int64_t now, bool rebuildBond)
* Curate the set of paths that are part of the bond proper. Selects a single path
* per logical link according to eligibility and user-specified constraints.
*/
if ((_bondingPolicy == ZT_BONDING_POLICY_BALANCE_RR)
|| (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR)
|| (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE)) {
if ((_bondingPolicy == ZT_BONDING_POLICY_BALANCE_RR) || (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_XOR) || (_bondingPolicy == ZT_BONDING_POLICY_BALANCE_AWARE)) {
if (! _numBondedPaths) {
rebuildBond = true;
}
@ -934,8 +985,7 @@ void Bond::curateBond(const int64_t now, bool rebuildBond)
linkMap[link] = i;
overriden = true;
}
if ((_paths[i]->preferred() && _paths[linkMap[link]]->preferred())
|| (!_paths[i]->preferred() && !_paths[linkMap[link]]->preferred())) {
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) {
@ -1043,7 +1093,10 @@ void Bond::estimatePathQuality(const int64_t now)
// Packet was lost
it = _paths[i]->qosStatsOut.erase(it);
++currentLostRecords;
} else { ++it; }
}
else {
++it;
}
}
quality[i] = 0;
@ -1121,9 +1174,16 @@ void Bond::processBalanceTasks(const int64_t now)
continue;
}
if (! _paths[i]->eligible(now, _ackSendInterval) && _paths[i]->_shouldReallocateFlows) {
char traceMsg[256]; char pathStr[128]; _paths[i]->address().toString(pathStr);
sprintf(traceMsg, "%s (balance-*) Reallocating flows to peer %llx from dead link %s/%s to surviving links",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), getLink(_paths[i])->ifname().c_str(), pathStr);
char traceMsg[256];
char pathStr[128];
_paths[i]->address().toString(pathStr);
sprintf(
traceMsg,
"%s (balance-*) Reallocating flows to peer %llx from dead link %s/%s to surviving links",
OSUtils::humanReadableTimestamp().c_str(),
(unsigned long long)(_peer->_id.address().toInt()),
getLink(_paths[i])->ifname().c_str(),
pathStr);
RR->t->bondStateMessage(NULL, traceMsg);
std::map<int32_t, SharedPtr<Flow> >::iterator flow_it = _flows.begin();
while (flow_it != _flows.end()) {
@ -1150,9 +1210,16 @@ void Bond::processBalanceTasks(const int64_t now)
}
if (_paths[i] && _paths[i]->bonded() && _paths[i]->eligible(now, _ackSendInterval) && (_paths[i]->_allocation < minimumAllocationValue) && _paths[i]->_assignedFlowCount) {
_paths[i]->address().toString(curPathStr);
char traceMsg[256]; char pathStr[128]; _paths[i]->address().toString(pathStr);
sprintf(traceMsg, "%s (balance-*) Reallocating flows to peer %llx from under-performing link %s/%s\n",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), getLink(_paths[i])->ifname().c_str(), pathStr);
char traceMsg[256];
char pathStr[128];
_paths[i]->address().toString(pathStr);
sprintf(
traceMsg,
"%s (balance-*) Reallocating flows to peer %llx from under-performing link %s/%s\n",
OSUtils::humanReadableTimestamp().c_str(),
(unsigned long long)(_peer->_id.address().toInt()),
getLink(_paths[i])->ifname().c_str(),
pathStr);
RR->t->bondStateMessage(NULL, traceMsg);
std::map<int32_t, SharedPtr<Flow> >::iterator flow_it = _flows.begin();
while (flow_it != _flows.end()) {
@ -1199,8 +1266,7 @@ void Bond::processBalanceTasks(const int64_t now)
if ((now - _lastFlowRebalance) > ZT_FLOW_REBALANCE_INTERVAL) {
std::map<int32_t, SharedPtr<Flow> >::iterator flow_it = _flows.begin();
while (flow_it != _flows.end()) {
if (flow_it->second->_previouslyAssignedPath && flow_it->second->_previouslyAssignedPath->eligible(now, _ackSendInterval)
&& (flow_it->second->_previouslyAssignedPath->_allocation >= (minimumAllocationValue * 2))) {
if (flow_it->second->_previouslyAssignedPath && flow_it->second->_previouslyAssignedPath->eligible(now, _ackSendInterval) && (flow_it->second->_previouslyAssignedPath->_allocation >= (minimumAllocationValue * 2))) {
// fprintf(stderr, "moving flow back onto its previous path assignment (based on eligibility)\n");
(flow_it->second->_assignedPath->_assignedFlowCount)--;
flow_it->second->assignPath(flow_it->second->_previouslyAssignedPath, now);
@ -1216,8 +1282,7 @@ void Bond::processBalanceTasks(const int64_t now)
if ((now - _lastFlowRebalance) > ZT_FLOW_REBALANCE_INTERVAL) {
std::map<int32_t, SharedPtr<Flow> >::iterator flow_it = _flows.begin();
while (flow_it != _flows.end()) {
if (flow_it->second->_previouslyAssignedPath && flow_it->second->_previouslyAssignedPath->eligible(now, _ackSendInterval)
&& (flow_it->second->_previouslyAssignedPath->_allocation >= (minimumAllocationValue * 2))) {
if (flow_it->second->_previouslyAssignedPath && flow_it->second->_previouslyAssignedPath->eligible(now, _ackSendInterval) && (flow_it->second->_previouslyAssignedPath->_allocation >= (minimumAllocationValue * 2))) {
// fprintf(stderr, "moving flow back onto its previous path assignment (based on performance)\n");
(flow_it->second->_assignedPath->_assignedFlowCount)--;
flow_it->second->assignPath(flow_it->second->_previouslyAssignedPath, now);
@ -1260,9 +1325,11 @@ bool Bond::abForciblyRotateLink()
_abPath->address().toString(prevPathStr);
dequeueNextActiveBackupPath(RR->node->now());
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Forcibly rotating peer %llx link from %s/%s to %s/%s",
sprintf(
traceMsg,
"%s (active-backup) Forcibly rotating peer %llx link from %s/%s to %s/%s",
OSUtils::humanReadableTimestamp().c_str(),
_peer->_id.address().toInt(),
(unsigned long long)(_peer->_id.address().toInt()),
getLink(prevPath)->ifname().c_str(),
prevPathStr,
getLink(_abPath)->ifname().c_str(),
@ -1285,23 +1352,28 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
bool bFoundPrimaryLink = false;
/**
* Generate periodic statuc report
* Generate periodic status report
*/
if ((now - _lastBondStatusLog) > ZT_MULTIPATH_BOND_STATUS_INTERVAL) {
_lastBondStatusLog = now;
if (_abPath) {
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Active link to peer %llx is %s/%s, failover queue size is %zu",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), getLink(_abPath)->ifname().c_str(), curPathStr, _abFailoverQueue.size());
sprintf(
traceMsg,
"%s (active-backup) Active link to peer %llx is %s/%s, failover queue size is %zu",
OSUtils::humanReadableTimestamp().c_str(),
(unsigned long long)(_peer->_id.address().toInt()),
getLink(_abPath)->ifname().c_str(),
curPathStr,
_abFailoverQueue.size());
RR->t->bondStateMessage(NULL, traceMsg);
} else {
sprintf(traceMsg, "%s (active-backup) No active link to peer %llx",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
}
else {
sprintf(traceMsg, "%s (active-backup) No active link to peer %llx", OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
if (_abFailoverQueue.empty()) {
sprintf(traceMsg, "%s (active-backup) Failover queue is empty, bond to peer %llx is NOT currently fault-tolerant",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
sprintf(traceMsg, "%s (active-backup) Failover queue is empty, bond to peer %llx is NOT currently fault-tolerant", OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
@ -1318,15 +1390,14 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
* simply find the next eligible path.
*/
if (! userHasSpecifiedLinks()) {
sprintf(traceMsg, "%s (active-backup) No links to peer %llx specified. Searching...",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt()); RR->t->bondStateMessage(NULL, traceMsg);
sprintf(traceMsg, "%s (active-backup) No links to peer %llx specified. Searching...", OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
for (int i = 0; i < ZT_MAX_PEER_NETWORK_PATHS; ++i) {
if (_paths[i] && _paths[i]->eligible(now, _ackSendInterval)) {
_paths[i]->address().toString(curPathStr);
SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, _paths[i]->localSocket());
if (link) {
sprintf(traceMsg, "%s (active-backup) Found eligible link %s/%s to peer %llx",
OSUtils::humanReadableTimestamp().c_str(), getLink(_paths[i])->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
sprintf(traceMsg, "%s (active-backup) Found eligible link %s/%s to peer %llx", OSUtils::humanReadableTimestamp().c_str(), getLink(_paths[i])->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
_abPath = _paths[i];
@ -1366,30 +1437,26 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
_abPath->address().toString(curPathStr);
SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, _abPath->localSocket());
if (link) {
sprintf(traceMsg, "%s (active-backup) Found preferred primary link %s/%s to peer %llx",
OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
sprintf(traceMsg, "%s (active-backup) Found preferred primary link %s/%s to peer %llx", OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
else {
if (bFoundPrimaryLink && nonPreferredPath) {
sprintf(traceMsg, "%s (active-backup) Found non-preferred primary link to peer %llx",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
sprintf(traceMsg, "%s (active-backup) Found non-preferred primary link to peer %llx", OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
_abPath = nonPreferredPath;
}
}
if (! _abPath) {
sprintf(traceMsg, "%s (active-backup) Designated primary link to peer %llx is not yet ready",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
sprintf(traceMsg, "%s (active-backup) Designated primary link to peer %llx is not yet ready", OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
// TODO: Should wait for some time (failover interval?) and then swtich to spare link
// TODO: Should wait for some time (failover interval?) and then switch to spare link
}
}
else if (! userHasSpecifiedPrimaryLink()) {
int _abIdx = ZT_MAX_PEER_NETWORK_PATHS;
sprintf(traceMsg, "%s (active-backup) User did not specify a primary link to peer %llx, selecting first available link",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
sprintf(traceMsg, "%s (active-backup) User did not specify a primary link to peer %llx, selecting first available link", OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
for (int i = 0; i < ZT_MAX_PEER_NETWORK_PATHS; ++i) {
if (_paths[i] && _paths[i]->eligible(now, _ackSendInterval)) {
@ -1405,8 +1472,7 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, _abPath->localSocket());
if (link) {
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Selected non-primary link %s/%s to peer %llx",
OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
sprintf(traceMsg, "%s (active-backup) Selected non-primary link %s/%s to peer %llx", OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
@ -1425,11 +1491,18 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, (*it)->localSocket());
it = _abFailoverQueue.erase(it);
if (link) {
sprintf(traceMsg, "%s (active-backup) Link %s/%s to peer %llx is now ineligible, removing from failover queue, there are %zu links in the queue",
OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt(), _abFailoverQueue.size());
sprintf(
traceMsg,
"%s (active-backup) Link %s/%s to peer %llx is now ineligible, removing from failover queue, there are %zu links in the queue",
OSUtils::humanReadableTimestamp().c_str(),
getLink(_abPath)->ifname().c_str(),
curPathStr,
_peer->_id.address().toInt(),
_abFailoverQueue.size());
RR->t->bondStateMessage(NULL, traceMsg);
}
} else {
}
else {
++it;
}
}
@ -1493,8 +1566,15 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
}
if (! bFoundPathInQueue) {
_abFailoverQueue.push_front(_paths[i]);
_paths[i]->address().toString(curPathStr); sprintf(traceMsg, "%s (active-backup) Added link %s/%s to peer %llx to failover queue, there are %zu links in the queue",
OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt(), _abFailoverQueue.size());
_paths[i]->address().toString(curPathStr);
sprintf(
traceMsg,
"%s (active-backup) Added link %s/%s to peer %llx to failover queue, there are %zu links in the queue",
OSUtils::humanReadableTimestamp().c_str(),
getLink(_abPath)->ifname().c_str(),
curPathStr,
_peer->_id.address().toInt(),
_abFailoverQueue.size());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
@ -1506,9 +1586,7 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
*/
else if (! userHasSpecifiedFailoverInstructions()) {
for (int i = 0; i < ZT_MAX_PEER_NETWORK_PATHS; ++i) {
if (!_paths[i]
|| !_paths[i]->allowed()
|| !_paths[i]->eligible(now,_ackSendInterval)) {
if (! _paths[i] || ! _paths[i]->allowed() || ! _paths[i]->eligible(now, _ackSendInterval)) {
continue;
}
int failoverScoreHandicap = 0;
@ -1526,7 +1604,8 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
if (_paths[i].ptr() == negotiatedPath.ptr()) {
_paths[i]->_negotiated = true;
failoverScoreHandicap = ZT_MULTIPATH_FAILOVER_HANDICAP_NEGOTIATED;
} else {
}
else {
_paths[i]->_negotiated = false;
}
_paths[i]->_failoverScore = _paths[i]->_allocation + failoverScoreHandicap;
@ -1540,8 +1619,14 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
if (! bFoundPathInQueue) {
_abFailoverQueue.push_front(_paths[i]);
_paths[i]->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Added link %s/%s to peer %llx to failover queue, there are %zu links in the queue",
OSUtils::humanReadableTimestamp().c_str(), getLink(_paths[i])->ifname().c_str(), curPathStr, _peer->_id.address().toInt(), _abFailoverQueue.size());
sprintf(
traceMsg,
"%s (active-backup) Added link %s/%s to peer %llx to failover queue, there are %zu links in the queue",
OSUtils::humanReadableTimestamp().c_str(),
getLink(_paths[i])->ifname().c_str(),
curPathStr,
_peer->_id.address().toInt(),
_abFailoverQueue.size());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
@ -1560,18 +1645,23 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
*/
if (_abPath && ! _abPath->eligible(now, _ackSendInterval)) { // Implicit ZT_MULTIPATH_RESELECTION_POLICY_FAILURE
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Link %s/%s to peer %llx has failed. Selecting new link from failover queue, there are %zu links in the queue",
OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt(), _abFailoverQueue.size());
sprintf(
traceMsg,
"%s (active-backup) Link %s/%s to peer %llx has failed. Selecting new link from failover queue, there are %zu links in the queue",
OSUtils::humanReadableTimestamp().c_str(),
getLink(_abPath)->ifname().c_str(),
curPathStr,
_peer->_id.address().toInt(),
_abFailoverQueue.size());
RR->t->bondStateMessage(NULL, traceMsg);
if (! _abFailoverQueue.empty()) {
dequeueNextActiveBackupPath(now);
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Active link to peer %llx has been switched to %s/%s",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), getLink(_abPath)->ifname().c_str(), curPathStr);
sprintf(traceMsg, "%s (active-backup) Active link to peer %llx has been switched to %s/%s", OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt(), getLink(_abPath)->ifname().c_str(), curPathStr);
RR->t->bondStateMessage(NULL, traceMsg);
} else {
sprintf(traceMsg, "%s (active-backup) Failover queue is empty. No links to peer %llx to choose from",
OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
}
else {
sprintf(traceMsg, "%s (active-backup) Failover queue is empty. No links to peer %llx to choose from", OSUtils::humanReadableTimestamp().c_str(), _peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
@ -1582,24 +1672,32 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
_lastActiveBackupPathChange = now;
}
if (_abLinkSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_ALWAYS) {
if (_abPath && !getLink(_abPath)->primary()
&& getLink(_abFailoverQueue.front())->primary()) {
if (_abPath && ! getLink(_abPath)->primary() && getLink(_abFailoverQueue.front())->primary()) {
dequeueNextActiveBackupPath(now);
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Switching back to available primary link %s/%s to peer %llx [linkSelectionMethod = always]",
OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
sprintf(
traceMsg,
"%s (active-backup) Switching back to available primary link %s/%s to peer %llx [linkSelectionMethod = always]",
OSUtils::humanReadableTimestamp().c_str(),
getLink(_abPath)->ifname().c_str(),
curPathStr,
_peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
if (_abLinkSelectMethod == ZT_MULTIPATH_RESELECTION_POLICY_BETTER) {
if (_abPath && ! getLink(_abPath)->primary()) {
// Active backup has switched to "better" primary link according to re-select policy.
if (getLink(_abFailoverQueue.front())->primary()
&& (_abFailoverQueue.front()->_failoverScore > _abPath->_failoverScore)) {
if (getLink(_abFailoverQueue.front())->primary() && (_abFailoverQueue.front()->_failoverScore > _abPath->_failoverScore)) {
dequeueNextActiveBackupPath(now);
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Switching back to user-defined primary link %s/%s to peer %llx [linkSelectionMethod = better]",
OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
sprintf(
traceMsg,
"%s (active-backup) Switching back to user-defined primary link %s/%s to peer %llx [linkSelectionMethod = better]",
OSUtils::humanReadableTimestamp().c_str(),
getLink(_abPath)->ifname().c_str(),
curPathStr,
_peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
@ -1613,8 +1711,13 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
_abPath->address().toString(prevPathStr);
_lastPathNegotiationCheck = now;
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Switching negotiated link %s/%s to peer %llx [linkSelectionMethod = optimize]",
OSUtils::humanReadableTimestamp().c_str(), getLink(_abPath)->ifname().c_str(), curPathStr, _peer->_id.address().toInt());
sprintf(
traceMsg,
"%s (active-backup) Switching negotiated link %s/%s to peer %llx [linkSelectionMethod = optimize]",
OSUtils::humanReadableTimestamp().c_str(),
getLink(_abPath)->ifname().c_str(),
curPathStr,
_peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
else {
@ -1631,8 +1734,17 @@ void Bond::processActiveBackupTasks(void *tPtr, const int64_t now)
_abPath->address().toString(prevPathStr);
dequeueNextActiveBackupPath(now);
_abPath->address().toString(curPathStr);
sprintf(traceMsg, "%s (active-backup) Switching from %s/%s (fscore=%d) to better link %s/%s (fscore=%d) for peer %llx [linkSelectionMethod = optimize]",
OSUtils::humanReadableTimestamp().c_str(), getLink(oldPath)->ifname().c_str(), prevPathStr, prevFScore, getLink(_abPath)->ifname().c_str(), curPathStr, newFScore, _peer->_id.address().toInt());
sprintf(
traceMsg,
"%s (active-backup) Switching from %s/%s (fscore=%d) to better link %s/%s (fscore=%d) for peer %llx [linkSelectionMethod = optimize]",
OSUtils::humanReadableTimestamp().c_str(),
getLink(oldPath)->ifname().c_str(),
prevPathStr,
prevFScore,
getLink(_abPath)->ifname().c_str(),
curPathStr,
newFScore,
_peer->_id.address().toInt());
RR->t->bondStateMessage(NULL, traceMsg);
}
}
@ -1651,7 +1763,8 @@ void Bond::setReasonableDefaults(int policy, SharedPtr<Bond> templateBond, bool
_bondingPolicy = ZT_BONDING_POLICY_NONE;
}
_bondingPolicy = _defaultBondingPolicy;
} else {
}
else {
_bondingPolicy = policy;
}
@ -1783,8 +1896,7 @@ void Bond::setReasonableDefaults(int policy, SharedPtr<Bond> templateBond, bool
_failoverInterval = templateBond->_failoverInterval >= 250 ? templateBond->_failoverInterval : _failoverInterval;
_downDelay = templateBond->_downDelay;
_upDelay = templateBond->_upDelay;
if (templateBond->_linkMonitorStrategy == ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_PASSIVE
&& templateBond->_failoverInterval != 0) {
if (templateBond->_linkMonitorStrategy == ZT_MULTIPATH_SLAVE_MONITOR_STRATEGY_PASSIVE && templateBond->_failoverInterval != 0) {
// fprintf(stderr, "warning: passive path monitoring was specified, this will prevent failovers from happening in a timely manner.\n");
}
_abLinkSelectMethod = templateBond->_abLinkSelectMethod;
@ -1816,7 +1928,8 @@ void Bond::setUserQualityWeights(float weights[], int len)
}
}
bool Bond::relevant() {
bool Bond::relevant()
{
return false;
}

211
node/Bond.hpp
View file

@ -14,26 +14,26 @@
#ifndef ZT_BOND_HPP
#define ZT_BOND_HPP
#include <map>
#include "Flow.hpp"
#include "Path.hpp"
#include "Peer.hpp"
#include "../osdep/Link.hpp"
#include "Flow.hpp"
#include "Packet.hpp"
#include <list>
#include <map>
namespace ZeroTier {
class RuntimeEnvironment;
class Link;
class Peer;
class Bond
{
class Bond {
friend class SharedPtr<Bond>;
friend class Peer;
friend class BondController;
struct PathQualityComparator
{
struct PathQualityComparator {
bool operator()(const SharedPtr<Path>& a, const SharedPtr<Path>& b)
{
if (a->_failoverScore == b->_failoverScore) {
@ -44,7 +44,6 @@ class Bond
};
public:
// TODO: Remove
bool _header;
int64_t _lastLogTS;
@ -84,7 +83,10 @@ public:
/**
* @return The human-readable name of the bonding policy
*/
std::string policyAlias() { return _policyAlias; }
std::string policyAlias()
{
return _policyAlias;
}
/**
* Inform the bond about the path that its peer (owning object) just learned about.
@ -135,8 +137,7 @@ public:
* @param flowId Flow ID
* @param now Current time
*/
void recordOutgoingPacket(const SharedPtr<Path> &path, uint64_t packetId,
uint16_t payloadLength, Packet::Verb verb, int32_t flowId, int64_t now);
void recordOutgoingPacket(const SharedPtr<Path>& path, uint64_t packetId, uint16_t payloadLength, Packet::Verb verb, int32_t flowId, int64_t now);
/**
* Process the contents of an inbound VERB_QOS_MEASUREMENT to gather path quality observations.
@ -176,8 +177,7 @@ public:
* @param flowId Flow ID
* @param now Current time
*/
void recordIncomingPacket(const SharedPtr<Path>& path, uint64_t packetId, uint16_t payloadLength,
Packet::Verb verb, int32_t flowId, int64_t now);
void recordIncomingPacket(const SharedPtr<Path>& path, uint64_t packetId, uint16_t payloadLength, Packet::Verb verb, int32_t flowId, int64_t now);
/**
* Determines the most appropriate path for packet and flow egress. This decision is made by
@ -246,8 +246,7 @@ public:
* @param atAddress
* @param now Current time
*/
void sendACK(void *tPtr, const SharedPtr<Path> &path,int64_t localSocket,
const InetAddress &atAddress,int64_t now);
void sendACK(void* tPtr, const SharedPtr<Path>& path, int64_t localSocket, const InetAddress& atAddress, int64_t now);
/**
* Sends a VERB_QOS_MEASUREMENT to the remote peer.
@ -258,8 +257,7 @@ public:
* @param atAddress
* @param now Current time
*/
void sendQOS_MEASUREMENT(void *tPtr,const SharedPtr<Path> &path,int64_t localSocket,
const InetAddress &atAddress,int64_t now);
void sendQOS_MEASUREMENT(void* tPtr, const SharedPtr<Path>& path, int64_t localSocket, const InetAddress& atAddress, int64_t now);
/**
* Sends a VERB_PATH_NEGOTIATION_REQUEST to the remote peer.
@ -311,64 +309,82 @@ public:
/**
* @param latencyInMilliseconds Maximum acceptable latency.
*/
void setMaxAcceptableLatency(int16_t latencyInMilliseconds) {
void setMaxAcceptableLatency(int16_t latencyInMilliseconds)
{
_maxAcceptableLatency = latencyInMilliseconds;
}
/**
* @param latencyInMilliseconds Maximum acceptable (mean) latency.
*/
void setMaxAcceptableMeanLatency(int16_t latencyInMilliseconds) {
void setMaxAcceptableMeanLatency(int16_t latencyInMilliseconds)
{
_maxAcceptableMeanLatency = latencyInMilliseconds;
}
/**
* @param latencyVarianceInMilliseconds Maximum acceptable packet delay variance (jitter).
*/
void setMaxAcceptablePacketDelayVariance(int16_t latencyVarianceInMilliseconds) {
void setMaxAcceptablePacketDelayVariance(int16_t latencyVarianceInMilliseconds)
{
_maxAcceptablePacketDelayVariance = latencyVarianceInMilliseconds;
}
/**
* @param lossRatio Maximum acceptable packet loss ratio (PLR).
*/
void setMaxAcceptablePacketLossRatio(float lossRatio) {
void setMaxAcceptablePacketLossRatio(float lossRatio)
{
_maxAcceptablePacketLossRatio = lossRatio;
}
/**
* @param errorRatio Maximum acceptable packet error ratio (PER).
*/
void setMaxAcceptablePacketErrorRatio(float errorRatio) {
void setMaxAcceptablePacketErrorRatio(float errorRatio)
{
_maxAcceptablePacketErrorRatio = errorRatio;
}
/**
* @param errorRatio Maximum acceptable packet error ratio (PER).
*/
void setMinAcceptableAllocation(float minAlloc) {
void setMinAcceptableAllocation(float minAlloc)
{
_minAcceptableAllocation = minAlloc * 255;
}
/**
* @return Whether the user has defined links for use on this bond
*/
inline bool userHasSpecifiedLinks() { return _userHasSpecifiedLinks; }
inline bool userHasSpecifiedLinks()
{
return _userHasSpecifiedLinks;
}
/**
* @return Whether the user has defined a set of failover link(s) for this bond
*/
inline bool userHasSpecifiedFailoverInstructions() { return _userHasSpecifiedFailoverInstructions; };
inline bool userHasSpecifiedFailoverInstructions()
{
return _userHasSpecifiedFailoverInstructions;
};
/**
* @return Whether the user has specified a primary link
*/
inline bool userHasSpecifiedPrimaryLink() { return _userHasSpecifiedPrimaryLink; }
inline bool userHasSpecifiedPrimaryLink()
{
return _userHasSpecifiedPrimaryLink;
}
/**
* @return Whether the user has specified link speeds
*/
inline bool userHasSpecifiedLinkSpeeds() { return _userHasSpecifiedLinkSpeeds; }
inline bool userHasSpecifiedLinkSpeeds()
{
return _userHasSpecifiedLinkSpeeds;
}
/**
* Periodically perform maintenance tasks for each active bond.
@ -391,7 +407,8 @@ public:
_lastAckRateCheck = now;
if (_ackCutoffCount > numToDrain) {
_ackCutoffCount -= numToDrain;
} else {
}
else {
_ackCutoffCount = 0;
}
return (_ackCutoffCount < ZT_ACK_CUTOFF_LIMIT);
@ -410,7 +427,8 @@ public:
_lastQoSRateCheck = now;
if (_qosCutoffCount > numToDrain) {
_qosCutoffCount -= numToDrain;
} else {
}
else {
_qosCutoffCount = 0;
}
return (_qosCutoffCount < ZT_QOS_CUTOFF_LIMIT);
@ -426,7 +444,8 @@ public:
{
if ((now - _lastPathNegotiationReceived) <= ZT_PATH_NEGOTIATION_CUTOFF_TIME)
++_pathNegotiationCutoffCount;
else _pathNegotiationCutoffCount = 0;
else
_pathNegotiationCutoffCount = 0;
_lastPathNegotiationReceived = now;
return (_pathNegotiationCutoffCount < ZT_PATH_NEGOTIATION_CUTOFF_LIMIT);
}
@ -434,130 +453,206 @@ public:
/**
* @param interval Maximum amount of time user expects a failover to take on this bond.
*/
inline void setFailoverInterval(uint32_t interval) { _failoverInterval = interval; }
inline void setFailoverInterval(uint32_t interval)
{
_failoverInterval = interval;
}
/**
* @param interval Maximum amount of time user expects a failover to take on this bond.
*/
inline uint32_t getFailoverInterval() { return _failoverInterval; }
inline uint32_t getFailoverInterval()
{
return _failoverInterval;
}
/**
* @param strategy Strategy that the bond uses to re-assign protocol flows.
*/
inline void setFlowRebalanceStrategy(uint32_t strategy) { _flowRebalanceStrategy = strategy; }
inline void setFlowRebalanceStrategy(uint32_t strategy)
{
_flowRebalanceStrategy = strategy;
}
/**
* @param strategy Strategy that the bond uses to prob for path aliveness and quality
*/
inline void setLinkMonitorStrategy(uint8_t strategy) { _linkMonitorStrategy = strategy; }
inline void setLinkMonitorStrategy(uint8_t strategy)
{
_linkMonitorStrategy = strategy;
}
/**
* @param abOverflowEnabled Whether "overflow" mode is enabled for this active-backup bond
*/
inline void setOverflowMode(bool abOverflowEnabled) { _abOverflowEnabled = abOverflowEnabled; }
inline void setOverflowMode(bool abOverflowEnabled)
{
_abOverflowEnabled = abOverflowEnabled;
}
/**
* @return the current up delay parameter
*/
inline uint16_t getUpDelay() { return _upDelay; }
inline uint16_t getUpDelay()
{
return _upDelay;
}
/**
* @param upDelay Length of time before a newly-discovered path is admitted to the bond
*/
inline void setUpDelay(int upDelay) { if (upDelay >= 0) { _upDelay = upDelay; } }
inline void setUpDelay(int upDelay)
{
if (upDelay >= 0) {
_upDelay = upDelay;
}
}
/**
* @return Length of time before a newly-failed path is removed from the bond
*/
inline uint16_t getDownDelay() { return _downDelay; }
inline uint16_t getDownDelay()
{
return _downDelay;
}
/**
* @param downDelay Length of time before a newly-failed path is removed from the bond
*/
inline void setDownDelay(int downDelay) { if (downDelay >= 0) { _downDelay = downDelay; } }
inline void setDownDelay(int downDelay)
{
if (downDelay >= 0) {
_downDelay = downDelay;
}
}
/**
* @return the current monitoring interval for the bond (can be overridden with intervals specific to certain links.)
*/
inline uint16_t getBondMonitorInterval() { return _bondMonitorInterval; }
inline uint16_t getBondMonitorInterval()
{
return _bondMonitorInterval;
}
/**
* Set the current monitoring interval for the bond (can be overridden with intervals specific to certain links.)
*
* @param monitorInterval How often gratuitous VERB_HELLO(s) are sent to remote peer.
*/
inline void setBondMonitorInterval(uint16_t interval) { _bondMonitorInterval = interval; }
inline void setBondMonitorInterval(uint16_t interval)
{
_bondMonitorInterval = interval;
}
/**
* @param policy Bonding policy for this bond
*/
inline void setPolicy(uint8_t policy) { _bondingPolicy = policy; }
inline void setPolicy(uint8_t policy)
{
_bondingPolicy = policy;
}
/**
* @return the current bonding policy
*/
inline uint8_t getPolicy() { return _bondingPolicy; }
inline uint8_t getPolicy()
{
return _bondingPolicy;
}
/**
* @return the health status of the bond
*/
inline bool isHealthy() { return _isHealthy; }
inline bool isHealthy()
{
return _isHealthy;
}
/**
* @return the number of links comprising this bond which are considered alive
*/
inline uint8_t getNumAliveLinks() { return _numAliveLinks; };
inline uint8_t getNumAliveLinks()
{
return _numAliveLinks;
};
/**
* @return the number of links comprising this bond
*/
inline uint8_t getNumTotalLinks() { return _numTotalLinks; }
inline uint8_t getNumTotalLinks()
{
return _numTotalLinks;
}
/**
*
* @param allowFlowHashing
*/
inline void setFlowHashing(bool allowFlowHashing) { _allowFlowHashing = allowFlowHashing; }
inline void setFlowHashing(bool allowFlowHashing)
{
_allowFlowHashing = allowFlowHashing;
}
/**
* @return Whether flow-hashing is currently enabled for this bond.
*/
bool flowHashingEnabled() { return _allowFlowHashing; }
bool flowHashingEnabled()
{
return _allowFlowHashing;
}
/**
*
* @param packetsPerLink
*/
inline void setPacketsPerLink(int packetsPerLink) { _packetsPerLink = packetsPerLink; }
inline void setPacketsPerLink(int packetsPerLink)
{
_packetsPerLink = packetsPerLink;
}
/**
* @return Number of packets to be sent on each interface in a balance-rr bond
*/
inline int getPacketsPerLink() { return _packetsPerLink; }
inline int getPacketsPerLink()
{
return _packetsPerLink;
}
/**
*
* @param linkSelectMethod
*/
inline void setLinkSelectMethod(uint8_t method) { _abLinkSelectMethod = method; }
inline void setLinkSelectMethod(uint8_t method)
{
_abLinkSelectMethod = method;
}
/**
*
* @return
*/
inline uint8_t getLinkSelectMethod() { return _abLinkSelectMethod; }
inline uint8_t getLinkSelectMethod()
{
return _abLinkSelectMethod;
}
/**
*
* @param allowPathNegotiation
*/
inline void setAllowPathNegotiation(bool allowPathNegotiation) { _allowPathNegotiation = allowPathNegotiation; }
inline void setAllowPathNegotiation(bool allowPathNegotiation)
{
_allowPathNegotiation = allowPathNegotiation;
}
/**
*
* @return
*/
inline bool allowPathNegotiation() { return _allowPathNegotiation; }
inline bool allowPathNegotiation()
{
return _allowPathNegotiation;
}
/**
* Forcibly rotates the currently active link used in an active-backup bond to the next link in the failover queue
@ -566,10 +661,12 @@ public:
*/
bool abForciblyRotateLink();
SharedPtr<Peer> getPeer() { return _peer; }
SharedPtr<Peer> getPeer()
{
return _peer;
}
private:
const RuntimeEnvironment* RR;
AtomicCounter __refCount;

31
node/BondController.cpp
View file

@ -11,19 +11,19 @@
*/
/****/
#include "../osdep/OSUtils.hpp"
#include "Constants.hpp"
#include "BondController.hpp"
#include "Peer.hpp"
#include "../osdep/OSUtils.hpp"
#include "Bond.hpp"
#include "Node.hpp"
#include "RuntimeEnvironment.hpp"
namespace ZeroTier {
int BondController::_minReqPathMonitorInterval;
uint8_t BondController::_defaultBondingPolicy;
BondController::BondController(const RuntimeEnvironment *renv) :
RR(renv)
BondController::BondController(const RuntimeEnvironment* renv) : RR(renv)
{
bondStartTime = RR->node->now();
_defaultBondingPolicy = ZT_BONDING_POLICY_NONE;
@ -92,27 +92,30 @@ SharedPtr<Bond> BondController::createTransportTriggeredBond(const RuntimeEnviro
std::string policyAlias;
if (! _policyTemplateAssignments.count(identity)) {
if (_defaultBondingPolicy) {
sprintf(traceMsg, "%s (bond) Creating new default %s bond to peer %llx",
OSUtils::humanReadableTimestamp().c_str(), getPolicyStrByCode(_defaultBondingPolicy).c_str(), identity); RR->t->bondStateMessage(NULL, traceMsg);
sprintf(traceMsg, "%s (bond) Creating new default %s bond to peer %llx", OSUtils::humanReadableTimestamp().c_str(), getPolicyStrByCode(_defaultBondingPolicy).c_str(), (unsigned long long)identity);
RR->t->bondStateMessage(NULL, traceMsg);
bond = new Bond(renv, _defaultBondingPolicy, peer);
}
if (! _defaultBondingPolicy && _defaultBondingPolicyStr.length()) {
sprintf(traceMsg, "%s (bond) Creating new default custom %s bond to peer %llx",
OSUtils::humanReadableTimestamp().c_str(), _defaultBondingPolicyStr.c_str(), identity);
sprintf(traceMsg, "%s (bond) Creating new default custom %s bond to peer %llx", OSUtils::humanReadableTimestamp().c_str(), _defaultBondingPolicyStr.c_str(), (unsigned long long)identity);
RR->t->bondStateMessage(NULL, traceMsg);
bond = new Bond(renv, _bondPolicyTemplates[_defaultBondingPolicyStr].ptr(), peer);
}
}
else {
if (! _bondPolicyTemplates[_policyTemplateAssignments[identity]]) {
sprintf(traceMsg, "%s (bond) Creating new bond. Assignment for peer %llx was specified as %s but the bond definition was not found. Using default %s",
OSUtils::humanReadableTimestamp().c_str(), identity, _policyTemplateAssignments[identity].c_str(), getPolicyStrByCode(_defaultBondingPolicy).c_str());
sprintf(
traceMsg,
"%s (bond) Creating new bond. Assignment for peer %llx was specified as %s but the bond definition was not found. Using default %s",
OSUtils::humanReadableTimestamp().c_str(),
(unsigned long long)identity,
_policyTemplateAssignments[identity].c_str(),
getPolicyStrByCode(_defaultBondingPolicy).c_str());
RR->t->bondStateMessage(NULL, traceMsg);
bond = new Bond(renv, _defaultBondingPolicy, peer);
}
else {
sprintf(traceMsg, "%s (bond) Creating new default bond %s to peer %llx",
OSUtils::humanReadableTimestamp().c_str(), _defaultBondingPolicyStr.c_str(), identity);
sprintf(traceMsg, "%s (bond) Creating new default bond %s to peer %llx", OSUtils::humanReadableTimestamp().c_str(), _defaultBondingPolicyStr.c_str(), (unsigned long long)identity);
RR->t->bondStateMessage(NULL, traceMsg);
bond = new Bond(renv, _bondPolicyTemplates[_policyTemplateAssignments[identity]].ptr(), peer);
}

89
node/BondController.hpp
View file

@ -14,25 +14,24 @@
#ifndef ZT_BONDCONTROLLER_HPP
#define ZT_BONDCONTROLLER_HPP
#include "../osdep/Link.hpp"
#include "../osdep/Phy.hpp"
#include "SharedPtr.hpp"
#include <map>
#include <vector>
#include "SharedPtr.hpp"
#include "../osdep/Phy.hpp"
#include "../osdep/Link.hpp"
namespace ZeroTier {
class RuntimeEnvironment;
class Bond;
class Peer;
class Mutex;
class BondController
{
class BondController {
friend class Bond;
public:
BondController(const RuntimeEnvironment* renv);
/**
@ -43,17 +42,26 @@ public:
/**
* @return The minimum interval required to poll the active bonds to fulfill all active monitoring timing requirements.
*/
int minReqPathMonitorInterval() { return _minReqPathMonitorInterval; }
int minReqPathMonitorInterval()
{
return _minReqPathMonitorInterval;
}
/**
* @param minReqPathMonitorInterval The minimum interval required to poll the active bonds to fulfill all active monitoring timing requirements.
*/
static void setMinReqPathMonitorInterval(int minReqPathMonitorInterval) { _minReqPathMonitorInterval = minReqPathMonitorInterval; }
static void setMinReqPathMonitorInterval(int minReqPathMonitorInterval)
{
_minReqPathMonitorInterval = minReqPathMonitorInterval;
}
/**
* @return Whether the bonding layer is currently set up to be used.
*/
bool inUse() { return !_bondPolicyTemplates.empty() || _defaultBondingPolicy; }
bool inUse()
{
return ! _bondPolicyTemplates.empty() || _defaultBondingPolicy;
}
/**
* @param basePolicyName Bonding policy name (See ZeroTierOne.h)
@ -61,11 +69,21 @@ public:
*/
static int getPolicyCodeByStr(const std::string& basePolicyName)
{
if (basePolicyName == "active-backup") { return 1; }
if (basePolicyName == "broadcast") { return 2; }
if (basePolicyName == "balance-rr") { return 3; }
if (basePolicyName == "balance-xor") { return 4; }
if (basePolicyName == "balance-aware") { return 5; }
if (basePolicyName == "active-backup") {
return 1;
}
if (basePolicyName == "broadcast") {
return 2;
}
if (basePolicyName == "balance-rr") {
return 3;
}
if (basePolicyName == "balance-xor") {
return 4;
}
if (basePolicyName == "balance-aware") {
return 5;
}
return 0; // "none"
}
@ -75,11 +93,21 @@ public:
*/
static std::string getPolicyStrByCode(int policy)
{
if (policy == 1) { return "active-backup"; }
if (policy == 2) { return "broadcast"; }
if (policy == 3) { return "balance-rr"; }
if (policy == 4) { return "balance-xor"; }
if (policy == 5) { return "balance-aware"; }
if (policy == 1) {
return "active-backup";
}
if (policy == 2) {
return "broadcast";
}
if (policy == 3) {
return "balance-rr";
}
if (policy == 4) {
return "balance-xor";
}
if (policy == 5) {
return "balance-aware";
}
return "none";
}
@ -88,19 +116,28 @@ public:
*
* @param bp Bonding policy
*/
void setBondingLayerDefaultPolicy(uint8_t bp) { _defaultBondingPolicy = bp; }
void setBondingLayerDefaultPolicy(uint8_t bp)
{
_defaultBondingPolicy = bp;
}
/**
* Sets the default (custom) bonding policy for new or undefined bonds.
*
* @param alias Human-readable string alias for bonding policy
*/
void setBondingLayerDefaultPolicyStr(std::string alias) { _defaultBondingPolicyStr = alias; }
void setBondingLayerDefaultPolicyStr(std::string alias)
{
_defaultBondingPolicyStr = alias;
}
/**
* @return The default bonding policy
*/
static int defaultBondingPolicy() { return _defaultBondingPolicy; }
static int defaultBondingPolicy()
{
return _defaultBondingPolicy;
}
/**
* Add a user-defined link to a given bonding policy.
@ -175,10 +212,12 @@ public:
*/
bool allowedToBind(const std::string& ifname);
uint64_t getBondStartTime() { return bondStartTime; }
uint64_t getBondStartTime()
{
return bondStartTime;
}
private:
Phy<BondController*>* _phy;
const RuntimeEnvironment* RR;

62
node/Flow.hpp
View file

@ -22,20 +22,14 @@ namespace ZeroTier {
/**
* A protocol flow that is identified by the origin and destination port.
*/
struct Flow
{
struct Flow {
/**
* @param flowId Given flow ID
* @param now Current time
*/
Flow(int32_t flowId, int64_t now) :
_flowId(flowId),
_bytesInPerUnitTime(0),
_bytesOutPerUnitTime(0),
_lastActivity(now),
_lastPathReassignment(0),
_assignedPath(SharedPtr<Path>())
{}
Flow(int32_t flowId, int64_t now) : _flowId(flowId), _bytesInPerUnitTime(0), _bytesOutPerUnitTime(0), _lastActivity(now), _lastPathReassignment(0), _assignedPath(SharedPtr<Path>())
{
}
/**
* Reset flow statistics
@ -49,36 +43,54 @@ struct Flow
/**
* @return The Flow's ID
*/
int32_t id() { return _flowId; }
int32_t id()
{
return _flowId;
}
/**
* @return Number of incoming bytes processed on this flow per unit time
*/
int64_t bytesInPerUnitTime() { return _bytesInPerUnitTime; }
int64_t bytesInPerUnitTime()
{
return _bytesInPerUnitTime;
}
/**
* Record number of incoming bytes on this flow
*
* @param bytes Number of incoming bytes
*/
void recordIncomingBytes(uint64_t bytes) { _bytesInPerUnitTime += bytes; }
void recordIncomingBytes(uint64_t bytes)
{
_bytesInPerUnitTime += bytes;
}
/**
* @return Number of outgoing bytes processed on this flow per unit time
*/
int64_t bytesOutPerUnitTime() { return _bytesOutPerUnitTime; }
int64_t bytesOutPerUnitTime()
{
return _bytesOutPerUnitTime;
}
/**
* Record number of outgoing bytes on this flow
*
* @param bytes
*/
void recordOutgoingBytes(uint64_t bytes) { _bytesOutPerUnitTime += bytes; }
void recordOutgoingBytes(uint64_t bytes)
{
_bytesOutPerUnitTime += bytes;
}
/**
* @return The total number of bytes processed on this flow
*/
uint64_t totalBytes() { return _bytesInPerUnitTime + _bytesOutPerUnitTime; }
uint64_t totalBytes()
{
return _bytesInPerUnitTime + _bytesOutPerUnitTime;
}
/**
* How long since a packet was sent or received in this flow
@ -86,24 +98,34 @@ struct Flow
* @param now Current time
* @return The age of the flow in terms of last recorded activity
*/
int64_t age(int64_t now) { return now - _lastActivity; }
int64_t age(int64_t now)
{
return now - _lastActivity;
}
/**
* Record that traffic was processed on this flow at the given time.
*
* @param now Current time
*/
void updateActivity(int64_t now) { _lastActivity = now; }
void updateActivity(int64_t now)
{
_lastActivity = now;
}
/**
* @return Path assigned to this flow
*/
SharedPtr<Path> assignedPath() { return _assignedPath; }
SharedPtr<Path> assignedPath()
{
return _assignedPath;
}
/**
* @param path Assigned path over which this flow should be handled
*/
void assignPath(const SharedPtr<Path> &path, int64_t now) {
void assignPath(const SharedPtr<Path>& path, int64_t now)
{
_assignedPath = path;
_lastPathReassignment = now;
}

137
osdep/Link.hpp
View file

@ -14,19 +14,20 @@
#ifndef ZT_LINK_HPP
#define ZT_LINK_HPP
#include <string>
#include "../node/AtomicCounter.hpp"
#include "../node/SharedPtr.hpp"
#include <string>
namespace ZeroTier {
class Link
{
class Link {
friend class SharedPtr<Link>;
public:
Link() {}
Link()
{
}
/**
*
@ -38,123 +39,165 @@ public:
* @param failoverToLinkStr
* @param userSpecifiedAlloc
*/
Link(std::string& ifnameStr,
uint8_t ipvPref,
uint32_t speed,
uint32_t linkMonitorInterval,
uint32_t upDelay,
uint32_t downDelay,
bool enabled,
uint8_t mode,
std::string failoverToLinkStr,
float userSpecifiedAlloc) :
_ifnameStr(ifnameStr),
_ipvPref(ipvPref),
_speed(speed),
_relativeSpeed(0),
_linkMonitorInterval(linkMonitorInterval),
_upDelay(upDelay),
_downDelay(downDelay),
_enabled(enabled),
_mode(mode),
_failoverToLinkStr(failoverToLinkStr),
_userSpecifiedAlloc(userSpecifiedAlloc),
_isUserSpecified(false)
{}
Link(std::string& ifnameStr, uint8_t ipvPref, uint32_t speed, uint32_t linkMonitorInterval, uint32_t upDelay, uint32_t downDelay, bool enabled, uint8_t mode, std::string failoverToLinkStr, float userSpecifiedAlloc)
: _ifnameStr(ifnameStr)
, _ipvPref(ipvPref)
, _speed(speed)
, _relativeSpeed(0)
, _linkMonitorInterval(linkMonitorInterval)
, _upDelay(upDelay)
, _downDelay(downDelay)
, _enabled(enabled)
, _mode(mode)
, _failoverToLinkStr(failoverToLinkStr)
, _userSpecifiedAlloc(userSpecifiedAlloc)
, _isUserSpecified(false)
{
}
/**
* @return The string representation of this link's underlying interface's system name.
*/
inline std::string ifname() { return _ifnameStr; }
inline std::string ifname()
{
return _ifnameStr;
}
/**
* @return Whether this link is designated as a primary.
*/
inline bool primary() { return _mode == ZT_MULTIPATH_SLAVE_MODE_PRIMARY; }
inline bool primary()
{
return _mode == ZT_MULTIPATH_SLAVE_MODE_PRIMARY;
}
/**
* @return Whether this link is designated as a spare.
*/
inline bool spare() { return _mode == ZT_MULTIPATH_SLAVE_MODE_SPARE; }
inline bool spare()
{
return _mode == ZT_MULTIPATH_SLAVE_MODE_SPARE;
}
/**
* @return The name of the link interface that should be used in the event of a failure.
*/
inline std::string failoverToLink() { return _failoverToLinkStr; }
inline std::string failoverToLink()
{
return _failoverToLinkStr;
}
/**
* @return Whether this link interface was specified by the user or auto-detected.
*/
inline bool isUserSpecified() { return _isUserSpecified; }
inline bool isUserSpecified()
{
return _isUserSpecified;
}
/**
* Signify that this link was specified by the user and not the result of auto-detection.
*
* @param isUserSpecified
*/
inline void setAsUserSpecified(bool isUserSpecified) { _isUserSpecified = isUserSpecified; }
inline void setAsUserSpecified(bool isUserSpecified)
{
_isUserSpecified = isUserSpecified;
}
/**
* @return Whether or not the user has specified failover instructions.
*/
inline bool userHasSpecifiedFailoverInstructions() { return _failoverToLinkStr.length(); }
inline bool userHasSpecifiedFailoverInstructions()
{
return _failoverToLinkStr.length();
}
/**
* @return The speed of the link relative to others in the bond.
*/
inline uint8_t relativeSpeed() { return _relativeSpeed; }
inline uint8_t relativeSpeed()
{
return _relativeSpeed;
}
/**
* Sets the speed of the link relative to others in the bond.
*
* @param relativeSpeed The speed relative to the rest of the link.
*/
inline void setRelativeSpeed(uint8_t relativeSpeed) { _relativeSpeed = relativeSpeed; }
inline void setRelativeSpeed(uint8_t relativeSpeed)
{
_relativeSpeed = relativeSpeed;
}
/**
* Sets the speed of the link relative to others in the bond.
*
* @param relativeSpeed
*/
inline void setMonitorInterval(uint32_t interval) { _linkMonitorInterval = interval; }
inline void setMonitorInterval(uint32_t interval)
{
_linkMonitorInterval = interval;
}
/**
* @return The absolute speed of the link (as specified by the user.)
*/
inline uint32_t monitorInterval() { return _linkMonitorInterval; }
inline uint32_t monitorInterval()
{
return _linkMonitorInterval;
}
/**
* @return The absolute speed of the link (as specified by the user.)
*/
inline uint32_t speed() { return _speed; }
inline uint32_t speed()
{
return _speed;
}
/**
* @return The address preference for this link (as specified by the user.)
*/
inline uint8_t ipvPref() { return _ipvPref; }
inline uint8_t ipvPref()
{
return _ipvPref;
}
/**
* @return The mode (e.g. primary/spare) for this link (as specified by the user.)
*/
inline uint8_t mode() { return _mode; }
inline uint8_t mode()
{
return _mode;
}
/**
* @return The upDelay parameter for all paths on this link.
*/
inline uint32_t upDelay() { return _upDelay; }
inline uint32_t upDelay()
{
return _upDelay;
}
/**
* @return The downDelay parameter for all paths on this link.
*/
inline uint32_t downDelay() { return _downDelay; }
inline uint32_t downDelay()
{
return _downDelay;
}
/**
* @return Whether this link is enabled or disabled
*/
inline uint8_t enabled() { return _enabled; }
inline uint8_t enabled()
{
return _enabled;
}
private:
/**
* String representation of underlying interface's system name
*/