github/ZeroTierOne
1
0
Fork 0
mirror of https://github.com/ZeroTier/ZeroTierOne synced 2025-08-14 02:27:38 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Consolidation of multipath logic. Better system separation

Browse source
This commit is contained in:
Joseph Henry 2021-09-01 21:37:49 -07:00
commit e1af003e4f
No known key found for this signature in database
GPG key ID: C45B33FF5EBC9344
23 changed files with 2004 additions and 3162 deletions
Download patch file Download diff file Expand all files Collapse all files

2180
node/Bond.cpp
View file

File diff suppressed because it is too large Load diff

1011
node/Bond.hpp
View file

File diff suppressed because it is too large Load diff

215
node/BondController.cpp
View file

@ -1,215 +0,0 @@
/*
* Copyright (c)2013-2020 ZeroTier, Inc.
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file in the project's root directory.
*
* Change Date: 2025-01-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2.0 of the Apache License.
*/
/****/
#include "BondController.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)
{
bondStartTime = RR->node->now();
_defaultBondingPolicy = ZT_BONDING_POLICY_NONE;
}
bool BondController::linkAllowed(std::string& policyAlias, SharedPtr<Link> link)
{
bool foundInDefinitions = false;
if (_linkDefinitions.count(policyAlias)) {
auto it = _linkDefinitions[policyAlias].begin();
while (it != _linkDefinitions[policyAlias].end()) {
if (link->ifname() == (*it)->ifname()) {
foundInDefinitions = true;
break;
}
++it;
}
}
return _linkDefinitions[policyAlias].empty() || foundInDefinitions;
}
void BondController::addCustomLink(std::string& policyAlias, SharedPtr<Link> link)
{
Mutex::Lock _l(_links_m);
_linkDefinitions[policyAlias].push_back(link);
auto search = _interfaceToLinkMap[policyAlias].find(link->ifname());
if (search == _interfaceToLinkMap[policyAlias].end()) {
link->setAsUserSpecified(true);
_interfaceToLinkMap[policyAlias].insert(std::pair<std::string, SharedPtr<Link> >(link->ifname(), link));
}
}
bool BondController::addCustomPolicy(const SharedPtr<Bond>& newBond)
{
Mutex::Lock _l(_bonds_m);
if (! _bondPolicyTemplates.count(newBond->policyAlias())) {
_bondPolicyTemplates[newBond->policyAlias()] = newBond;
return true;
}
return false;
}
bool BondController::assignBondingPolicyToPeer(int64_t identity, const std::string& policyAlias)
{
Mutex::Lock _l(_bonds_m);
if (! _policyTemplateAssignments.count(identity)) {
_policyTemplateAssignments[identity] = policyAlias;
return true;
}
return false;
}
SharedPtr<Bond> BondController::getBondByPeerId(int64_t identity)
{
Mutex::Lock _l(_bonds_m);
return _bonds.count(identity) ? _bonds[identity] : SharedPtr<Bond>();
}
SharedPtr<Bond> BondController::createTransportTriggeredBond(const RuntimeEnvironment* renv, const SharedPtr<Peer>& peer)
{
Mutex::Lock _l(_bonds_m);
int64_t identity = peer->identity().address().toInt();
Bond* bond = nullptr;
char traceMsg[128];
if (! _bonds.count(identity)) {
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(), (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(), (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(),
(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(), (unsigned long long)identity);
RR->t->bondStateMessage(NULL, traceMsg);
bond = new Bond(renv, _bondPolicyTemplates[_policyTemplateAssignments[identity]].ptr(), peer);
}
}
}
if (bond) {
_bonds[identity] = bond;
/**
* Determine if user has specified anything that could affect the bonding policy's decisions
*/
if (_interfaceToLinkMap.count(bond->policyAlias())) {
std::map<std::string, SharedPtr<Link> >::iterator it = _interfaceToLinkMap[bond->policyAlias()].begin();
while (it != _interfaceToLinkMap[bond->policyAlias()].end()) {
if (it->second->isUserSpecified()) {
bond->_userHasSpecifiedLinks = true;
}
if (it->second->isUserSpecified() && it->second->primary()) {
bond->_userHasSpecifiedPrimaryLink = true;
}
if (it->second->isUserSpecified() && it->second->userHasSpecifiedFailoverInstructions()) {
bond->_userHasSpecifiedFailoverInstructions = true;
}
if (it->second->isUserSpecified() && (it->second->speed() > 0)) {
bond->_userHasSpecifiedLinkSpeeds = true;
}
++it;
}
}
return bond;
}
return SharedPtr<Bond>();
}
SharedPtr<Link> BondController::getLinkBySocket(const std::string& policyAlias, uint64_t localSocket)
{
Mutex::Lock _l(_links_m);
char ifname[16];
_phy->getIfName((PhySocket*)((uintptr_t)localSocket), ifname, 16);
std::string ifnameStr(ifname);
auto search = _interfaceToLinkMap[policyAlias].find(ifnameStr);
if (search == _interfaceToLinkMap[policyAlias].end()) {
SharedPtr<Link> s = new Link(ifnameStr, 0, 0, 0, 0, 0, true, ZT_MULTIPATH_SLAVE_MODE_SPARE, "", 0.0);
_interfaceToLinkMap[policyAlias].insert(std::pair<std::string, SharedPtr<Link> >(ifnameStr, s));
return s;
}
else {
return search->second;
}
}
SharedPtr<Link> BondController::getLinkByName(const std::string& policyAlias, const std::string& ifname)
{
Mutex::Lock _l(_links_m);
auto search = _interfaceToLinkMap[policyAlias].find(ifname);
if (search != _interfaceToLinkMap[policyAlias].end()) {
return search->second;
}
return SharedPtr<Link>();
}
bool BondController::allowedToBind(const std::string& ifname)
{
return true;
/*
if (!_defaultBondingPolicy) {
return true; // no restrictions
}
Mutex::Lock _l(_links_m);
if (_interfaceToLinkMap.empty()) {
return true; // no restrictions
}
std::map<std::string, std::map<std::string, SharedPtr<Link> > >::iterator policyItr = _interfaceToLinkMap.begin();
while (policyItr != _interfaceToLinkMap.end()) {
std::map<std::string, SharedPtr<Link> >::iterator linkItr = policyItr->second.begin();
while (linkItr != policyItr->second.end()) {
if (linkItr->first == ifname) {
return true;
}
++linkItr;
}
++policyItr;
}
return false;
*/
}
void BondController::processBackgroundTasks(void* tPtr, const int64_t now)
{
Mutex::Lock _l(_bonds_m);
std::map<int64_t, SharedPtr<Bond> >::iterator bondItr = _bonds.begin();
while (bondItr != _bonds.end()) {
bondItr->second->processBackgroundTasks(tPtr, now);
++bondItr;
}
}
} // namespace ZeroTier

278
node/BondController.hpp
View file

@ -1,278 +0,0 @@
/*
* Copyright (c)2013-2020 ZeroTier, Inc.
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file in the project's root directory.
*
* Change Date: 2025-01-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2.0 of the Apache License.
*/
/****/
#ifndef ZT_BONDCONTROLLER_HPP
#define ZT_BONDCONTROLLER_HPP
#include "../osdep/Link.hpp"
#include "../osdep/Phy.hpp"
#include "SharedPtr.hpp"
#include <map>
#include <vector>
namespace ZeroTier {
class RuntimeEnvironment;
class Bond;
class Peer;
class Mutex;
class BondController {
friend class Bond;
public:
BondController(const RuntimeEnvironment* renv);
/**
* @return Whether this link is permitted to become a member of a bond.
*/
bool linkAllowed(std::string& policyAlias, SharedPtr<Link> link);
/**
* @return The minimum interval required to poll the active bonds to fulfill all active monitoring timing requirements.
*/
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;
}
/**
* @return Whether the bonding layer is currently set up to be used.
*/
bool inUse()
{
return ! _bondPolicyTemplates.empty() || _defaultBondingPolicy;
}
/**
* @param basePolicyName Bonding policy name (See ZeroTierOne.h)
* @return The bonding policy code for a given human-readable bonding policy name
*/
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;
}
return 0; // "none"
}
/**
* @param policy Bonding policy code (See ZeroTierOne.h)
* @return The human-readable name for the given bonding policy code
*/
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";
}
return "none";
}
/**
* Sets the default bonding policy for new or undefined bonds.
*
* @param bp Bonding policy
*/
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;
}
/**
* @return The default bonding policy
*/
static int defaultBondingPolicy()
{
return _defaultBondingPolicy;
}
/**
* Add a user-defined link to a given bonding policy.
*
* @param policyAlias User-defined custom name for variant of bonding policy
* @param link Pointer to new link definition
*/
void addCustomLink(std::string& policyAlias, SharedPtr<Link> link);
/**
* Add a user-defined bonding policy that is based on one of the standard types.
*
* @param newBond Pointer to custom Bond object
* @return Whether a uniquely-named custom policy was successfully added
*/
bool addCustomPolicy(const SharedPtr<Bond>& newBond);
/**
* Assigns a specific bonding policy
*
* @param identity
* @param policyAlias
* @return
*/
bool assignBondingPolicyToPeer(int64_t identity, const std::string& policyAlias);
/**
* Get pointer to bond by a given peer ID
*
* @param peer Remote peer ID
* @return A pointer to the Bond
*/
SharedPtr<Bond> getBondByPeerId(int64_t identity);
/**
* Add a new bond to the bond controller.
*
* @param renv Runtime environment
* @param peer Remote peer that this bond services
* @return A pointer to the newly created Bond
*/
SharedPtr<Bond> createTransportTriggeredBond(const RuntimeEnvironment* renv, const SharedPtr<Peer>& peer);
/**
* Periodically perform maintenance tasks for the bonding layer.
*
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
* @param now Current time
*/
void processBackgroundTasks(void* tPtr, int64_t now);
/**
* Gets a reference to a physical link definition given a policy alias and a local socket.
*
* @param policyAlias Policy in use
* @param localSocket Local source socket
* @return Physical link definition
*/
SharedPtr<Link> getLinkBySocket(const std::string& policyAlias, uint64_t localSocket);
/**
* Gets a reference to a physical link definition given its human-readable system name.
*
* @param policyAlias Policy in use
* @param ifname Alphanumeric human-readable name
* @return Physical link definition
*/
SharedPtr<Link> getLinkByName(const std::string& policyAlias, const std::string& ifname);
/**
* @param ifname Name of interface that we want to know if we can bind to
*/
bool allowedToBind(const std::string& ifname);
uint64_t getBondStartTime()
{
return bondStartTime;
}
private:
Phy<BondController*>* _phy;
const RuntimeEnvironment* RR;
Mutex _bonds_m;
Mutex _links_m;
/**
* The last time that the bond controller updated the set of bonds.
*/
uint64_t _lastBackgroundBondControlTaskCheck;
/**
* The minimum monitoring interval among all paths in this bond.
*/
static int _minReqPathMonitorInterval;
/**
* The default bonding policy used for new bonds unless otherwise specified.
*/
static uint8_t _defaultBondingPolicy;
/**
* The default bonding policy used for new bonds unless otherwise specified.
*/
std::string _defaultBondingPolicyStr;
/**
* All currently active bonds.
*/
std::map<int64_t, SharedPtr<Bond> > _bonds;
/**
* Map of peers to custom bonding policies
*/
std::map<int64_t, std::string> _policyTemplateAssignments;
/**
* User-defined bonding policies (can be assigned to a peer)
*/
std::map<std::string, SharedPtr<Bond> > _bondPolicyTemplates;
/**
* Set of links defined for a given bonding policy
*/
std::map<std::string, std::vector<SharedPtr<Link> > > _linkDefinitions;
/**
* Set of link objects mapped to their physical interfaces
*/
std::map<std::string, std::map<std::string, SharedPtr<Link> > > _interfaceToLinkMap;
// TODO: Remove
uint64_t bondStartTime;
};
} // namespace ZeroTier
#endif

200
node/Constants.hpp
View file

@ -361,7 +361,7 @@
/**
* Maximum number of outgoing packets we monitor for QoS information
*/
#define ZT_QOS_MAX_OUTSTANDING_RECORDS (1024*16)
#define ZT_QOS_MAX_OUTSTANDING_RECORDS (1024 * 16)
/**
* Interval used for rate-limiting the computation of path quality estimates.
@ -403,117 +403,11 @@
/**
* All unspecified traffic is put in this bucket. Anything in a bucket with a
* smaller value is deprioritized. Anything in a bucket with a higher value is
* smaller value is de-prioritized. Anything in a bucket with a higher value is
prioritized over other traffic.
*/
#define ZT_AQM_DEFAULT_BUCKET 0
/**
* How often we emit a one-liner bond summary for each peer
*/
#define ZT_MULTIPATH_BOND_STATUS_INTERVAL 60000
/**
* How long before we consider a path to be dead in the general sense. This is
* used while searching for default or alternative paths to try in the absence
* of direct guidance from the user or a selection policy.
*/
#define ZT_MULTIPATH_DEFAULT_FAILOVER_INTERVAL 10000
/**
* How often flows are evaluated
*/
#define ZT_MULTIPATH_FLOW_CHECK_INTERVAL 10000
/**
* How long before we consider a flow to be dead and remove it from the
* policy's list.
*/
#define ZT_MULTIPATH_FLOW_EXPIRATION_INTERVAL (60000 * 5)
/**
* How often a flow's statistical counters are reset
*/
#define ZT_FLOW_STATS_RESET_INTERVAL ZT_MULTIPATH_FLOW_EXPIRATION_INTERVAL
/**
* Maximum number of flows allowed before we start forcibly forgetting old ones
*/
#define ZT_FLOW_MAX_COUNT (1024*64)
/**
* How often flows are rebalanced across link (if at all)
*/
#define ZT_FLOW_MIN_REBALANCE_INTERVAL 5000
/**
* How often flows are rebalanced across link (if at all)
*/
#define ZT_FLOW_REBALANCE_INTERVAL 5000
/**
* A defensive timer to prevent path quality metrics from being
* processed too often.
*/
#define ZT_BOND_BACKGROUND_TASK_MIN_INTERVAL ZT_CORE_TIMER_TASK_GRANULARITY
/**
* How often a bonding policy's background tasks are processed,
* some need more frequent attention than others.
*/
#define ZT_MULTIPATH_ACTIVE_BACKUP_CHECK_INTERVAL ZT_CORE_TIMER_TASK_GRANULARITY
/**
* Minimum amount of time (since a previous transition) before the active-backup bonding
* policy is allowed to transition to a different link. Only valid for active-backup.
*/
#define ZT_MULTIPATH_MIN_ACTIVE_BACKUP_AUTOFLOP_INTERVAL 10000
/**
* How often a peer checks that incoming (and outgoing) traffic on a bonded link is
* appropriately paired.
*/
#define ZT_PATH_NEGOTIATION_CHECK_INTERVAL 15000
/**
* Time horizon for path negotiation paths cutoff
*/
#define ZT_PATH_NEGOTIATION_CUTOFF_TIME 60000
/**
* Maximum number of path negotiations within cutoff time
*
* This limits response to PATH_NEGOTIATION to CUTOFF_LIMIT responses
* per CUTOFF_TIME milliseconds per peer to prevent this from being
* useful for DOS amplification attacks.
*/
#define ZT_PATH_NEGOTIATION_CUTOFF_LIMIT 8
/**
* How many times a peer will attempt to petition another peer to synchronize its
* traffic to the same path before giving up and surrendering to the other peer's preference.
*/
#define ZT_PATH_NEGOTIATION_TRY_COUNT 3
/**
* How much greater the quality of a path should be before an
* optimization procedure triggers a switch.
*/
#define ZT_MULTIPATH_ACTIVE_BACKUP_OPTIMIZE_MIN_THRESHOLD 0.10
/**
* Artificially inflates the failover score for paths which meet
* certain non-performance-related policy ranking criteria.
*/
#define ZT_MULTIPATH_FAILOVER_HANDICAP_PREFERRED 500
#define ZT_MULTIPATH_FAILOVER_HANDICAP_PRIMARY 1000
#define ZT_MULTIPATH_FAILOVER_HANDICAP_NEGOTIATED 5000
/**
* An indicator that no flow is to be associated with the given packet
*/
#define ZT_QOS_NO_FLOW -1
/**
* Timeout for overall peer activity (measured from last receive)
*/
@ -604,8 +498,8 @@
#define ZT_ACK_CUTOFF_LIMIT 128
#define ZT_ACK_DRAINAGE_DIVISOR (1000 / ZT_ACK_CUTOFF_LIMIT)
#define ZT_MULTIPATH_DEFAULT_REFRCTORY_PERIOD 8000
#define ZT_MULTIPATH_MAX_REFRACTORY_PERIOD 600000
#define ZT_BOND_DEFAULT_REFRCTORY_PERIOD 8000
#define ZT_BOND_MAX_REFRACTORY_PERIOD 600000
/**
* Maximum number of direct path pushes within cutoff time
@ -641,6 +535,92 @@
*/
#define ZT_PEER_GENERAL_RATE_LIMIT 1000
/**
* Minimum allowed amount of time between flow/path optimizations (anti-flapping)
*/
#define ZT_BOND_OPTIMIZE_INTERVAL 15000
/**
* Maximum number of flows allowed before we start forcibly forgetting old ones
*/
#define ZT_FLOW_MAX_COUNT (1024 * 64)
/**
* How often we emit a bond summary for each bond
*/
#define ZT_BOND_STATUS_INTERVAL 3000
/**
* How long before we consider a path to be dead in the general sense. This is
* used while searching for default or alternative paths to try in the absence
* of direct guidance from the user or a selection policy.
*/
#define ZT_BOND_FAILOVER_DEFAULT_INTERVAL 5000
/**
* Anything below this value gets into thrashing territory since we divide
* this value by ZT_BOND_ECHOS_PER_FAILOVER_INTERVAL to send ECHOs often.
*/
#define ZT_BOND_FAILOVER_MIN_INTERVAL 250
/**
* How many times per failover interval that an ECHO is sent. This should be
* at least 2. Anything more then 4 starts to increase overhead significantly.
*/
#define ZT_BOND_ECHOS_PER_FAILOVER_INTERVAL 4
/**
* A defensive timer to prevent path quality metrics from being
* processed too often.
*/
#define ZT_BOND_BACKGROUND_TASK_MIN_INTERVAL ZT_CORE_TIMER_TASK_GRANULARITY
/**
* How often a bonding policy's background tasks are processed,
* some need more frequent attention than others.
*/
#define ZT_BOND_ACTIVE_BACKUP_CHECK_INTERVAL ZT_CORE_TIMER_TASK_GRANULARITY
/**
* Time horizon for path negotiation paths cutoff
*/
#define ZT_PATH_NEGOTIATION_CUTOFF_TIME 60000
/**
* Maximum number of path negotiations within cutoff time
*
* This limits response to PATH_NEGOTIATION to CUTOFF_LIMIT responses
* per CUTOFF_TIME milliseconds per peer to prevent this from being
* useful for DOS amplification attacks.
*/
#define ZT_PATH_NEGOTIATION_CUTOFF_LIMIT 8
/**
* How many times a peer will attempt to petition another peer to synchronize its
* traffic to the same path before giving up and surrendering to the other peer's preference.
*/
#define ZT_PATH_NEGOTIATION_TRY_COUNT 3
/**
* How much greater the quality of a path should be before an
* optimization procedure triggers a switch.
*/
#define ZT_BOND_ACTIVE_BACKUP_OPTIMIZE_MIN_THRESHOLD 0.10
/**
* Artificially inflates the failover score for paths which meet
* certain non-performance-related policy ranking criteria.
*/
#define ZT_BOND_FAILOVER_HANDICAP_PREFERRED 500
#define ZT_BOND_FAILOVER_HANDICAP_PRIMARY 1000
#define ZT_BOND_FAILOVER_HANDICAP_NEGOTIATED 5000
/**
* An indicator that no flow is to be associated with the given packet
*/
#define ZT_QOS_NO_FLOW -1
/**
* Don't do expensive identity validation more often than this
*

146
node/Flow.hpp
View file

@ -1,146 +0,0 @@
/*
* Copyright (c)2013-2020 ZeroTier, Inc.
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file in the project's root directory.
*
* Change Date: 2025-01-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2.0 of the Apache License.
*/
/****/
#ifndef ZT_FLOW_HPP
#define ZT_FLOW_HPP
#include "Path.hpp"
#include "SharedPtr.hpp"
namespace ZeroTier {
/**
* A protocol flow that is identified by the origin and destination port.
*/
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>())
{
}
/**
* Reset flow statistics
*/
void resetByteCounts()
{
_bytesInPerUnitTime = 0;
_bytesOutPerUnitTime = 0;
}
/**
* @return The Flow's ID
*/
int32_t id()
{
return _flowId;
}
/**
* @return Number of incoming bytes processed on this flow per unit time
*/
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;
}
/**
* @return Number of outgoing bytes processed on this flow per unit time
*/
int64_t bytesOutPerUnitTime()
{
return _bytesOutPerUnitTime;
}
/**
* Record number of outgoing bytes on this flow
*
* @param bytes
*/
void recordOutgoingBytes(uint64_t bytes)
{
_bytesOutPerUnitTime += bytes;
}
/**
* @return The total number of bytes processed on this flow
*/
uint64_t totalBytes()
{
return _bytesInPerUnitTime + _bytesOutPerUnitTime;
}
/**
* How long since a packet was sent or received in this 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;
}
/**
* Record that traffic was processed on this flow at the given time.
*
* @param now Current time
*/
void updateActivity(int64_t now)
{
_lastActivity = now;
}
/**
* @return Path assigned to this flow
*/
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)
{
_assignedPath = path;
_lastPathReassignment = now;
}
AtomicCounter __refCount;
int32_t _flowId;
uint64_t _bytesInPerUnitTime;
uint64_t _bytesOutPerUnitTime;
int64_t _lastActivity;
int64_t _lastPathReassignment;
SharedPtr<Path> _assignedPath;
SharedPtr<Path> _previouslyAssignedPath;
};
} // namespace ZeroTier
#endif

28
node/IncomingPacket.cpp
View file

@ -88,7 +88,6 @@ bool IncomingPacket::tryDecode(const RuntimeEnvironment *RR,void *tPtr,int32_t f
peer->received(tPtr,_path,hops(),packetId(),payloadLength(),v,0,Packet::VERB_NOP,false,0,ZT_QOS_NO_FLOW);
break;
case Packet::VERB_HELLO: r = _doHELLO(RR,tPtr,true); break;
case Packet::VERB_ACK: r = _doACK(RR,tPtr,peer); break;
case Packet::VERB_QOS_MEASUREMENT: r = _doQOS_MEASUREMENT(RR,tPtr,peer); break;
case Packet::VERB_ERROR: r = _doERROR(RR,tPtr,peer); break;
case Packet::VERB_OK: r = _doOK(RR,tPtr,peer); break;
@ -222,35 +221,12 @@ bool IncomingPacket::_doERROR(const RuntimeEnvironment *RR,void *tPtr,const Shar
return true;
}
bool IncomingPacket::_doACK(const RuntimeEnvironment *RR,void *tPtr,const SharedPtr<Peer> &peer)
{
SharedPtr<Bond> bond = peer->bond();
if (!bond || !bond->rateGateACK(RR->node->now())) {
return true;
}
/* Dissect incoming ACK packet. From this we can estimate current throughput of the path, establish known
* maximums and detect packet loss. */
int32_t ackedBytes;
if (payloadLength() != sizeof(ackedBytes)) {
return true; // ignore
}
memcpy(&ackedBytes, payload(), sizeof(ackedBytes));
if (bond) {
bond->receivedAck(_path, RR->node->now(), Utils::ntoh(ackedBytes));
}
return true;
}
bool IncomingPacket::_doQOS_MEASUREMENT(const RuntimeEnvironment *RR,void *tPtr,const SharedPtr<Peer> &peer)
{
SharedPtr<Bond> bond = peer->bond();
/* TODO: Fix rate gate issue
if (!bond || !bond->rateGateQoS(RR->node->now())) {
if (!bond || !bond->rateGateQoS(RR->node->now(), _path)) {
return true;
}
*/
/* Dissect incoming QoS packet. From this we can compute latency values and their variance.
* The latency variance is used as a measure of "jitter". */
if (payloadLength() > ZT_QOS_MAX_PACKET_SIZE || payloadLength() < ZT_QOS_MIN_PACKET_SIZE) {
return true; // ignore
}
@ -1329,7 +1305,7 @@ bool IncomingPacket::_doPATH_NEGOTIATION_REQUEST(const RuntimeEnvironment *RR,vo
{
uint64_t now = RR->node->now();
SharedPtr<Bond> bond = peer->bond();
if (!bond || !bond->rateGatePathNegotiation(now)) {
if (!bond || !bond->rateGatePathNegotiation(now, _path)) {
return true;
}
if (payloadLength() != sizeof(int16_t)) {

1
node/IncomingPacket.hpp
View file

@ -112,7 +112,6 @@ private:
// been authenticated, decrypted, decompressed, and classified.
bool _doERROR(const RuntimeEnvironment *RR,void *tPtr,const SharedPtr<Peer> &peer);
bool _doHELLO(const RuntimeEnvironment *RR,void *tPtr,const bool alreadyAuthenticated);
bool _doACK(const RuntimeEnvironment *RR,void *tPtr,const SharedPtr<Peer> &peer);
bool _doQOS_MEASUREMENT(const RuntimeEnvironment *RR,void *tPtr,const SharedPtr<Peer> &peer);
bool _doOK(const RuntimeEnvironment *RR,void *tPtr,const SharedPtr<Peer> &peer);
bool _doWHOIS(const RuntimeEnvironment *RR,void *tPtr,const SharedPtr<Peer> &peer);

24
node/Node.cpp
View file

@ -103,7 +103,7 @@ Node::Node(void *uptr,void *tptr,const struct ZT_Node_Callbacks *callbacks,int64
const unsigned long mcs = sizeof(Multicaster) + (((sizeof(Multicaster) & 0xf) != 0) ? (16 - (sizeof(Multicaster) & 0xf)) : 0);
const unsigned long topologys = sizeof(Topology) + (((sizeof(Topology) & 0xf) != 0) ? (16 - (sizeof(Topology) & 0xf)) : 0);
const unsigned long sas = sizeof(SelfAwareness) + (((sizeof(SelfAwareness) & 0xf) != 0) ? (16 - (sizeof(SelfAwareness) & 0xf)) : 0);
const unsigned long bc = sizeof(BondController) + (((sizeof(BondController) & 0xf) != 0) ? (16 - (sizeof(BondController) & 0xf)) : 0);
const unsigned long bc = sizeof(Bond) + (((sizeof(Bond) & 0xf) != 0) ? (16 - (sizeof(Bond) & 0xf)) : 0);
m = reinterpret_cast<char *>(::malloc(16 + ts + sws + mcs + topologys + sas + bc));
if (!m)
@ -121,14 +121,14 @@ Node::Node(void *uptr,void *tptr,const struct ZT_Node_Callbacks *callbacks,int64
m += topologys;
RR->sa = new (m) SelfAwareness(RR);
m += sas;
RR->bc = new (m) BondController(RR);
RR->bc = new (m) Bond(RR);
} catch ( ... ) {
if (RR->sa) RR->sa->~SelfAwareness();
if (RR->topology) RR->topology->~Topology();
if (RR->mc) RR->mc->~Multicaster();
if (RR->sw) RR->sw->~Switch();
if (RR->t) RR->t->~Trace();
if (RR->bc) RR->bc->~BondController();
if (RR->bc) RR->bc->~Bond();
::free(m);
throw;
}
@ -147,7 +147,7 @@ Node::~Node()
if (RR->mc) RR->mc->~Multicaster();
if (RR->sw) RR->sw->~Switch();
if (RR->t) RR->t->~Trace();
if (RR->bc) RR->bc->~BondController();
if (RR->bc) RR->bc->~Bond();
::free(RR->rtmem);
}
@ -252,18 +252,14 @@ ZT_ResultCode Node::processBackgroundTasks(void *tptr,int64_t now,volatile int64
_now = now;
Mutex::Lock bl(_backgroundTasksLock);
unsigned long bondCheckInterval = ZT_CORE_TIMER_TASK_GRANULARITY;
// Process background bond tasks
unsigned long bondCheckInterval = ZT_PING_CHECK_INVERVAL;
if (RR->bc->inUse()) {
// Gratuitously ping active peers so that QoS metrics have enough data to work with (if active path monitoring is enabled)
bondCheckInterval = std::min(std::max(RR->bc->minReqPathMonitorInterval(), ZT_CORE_TIMER_TASK_GRANULARITY), ZT_PING_CHECK_INVERVAL);
if ((now - _lastGratuitousPingCheck) >= bondCheckInterval) {
Hashtable< Address,std::vector<InetAddress> > alwaysContact;
_PingPeersThatNeedPing pfunc(RR,tptr,alwaysContact,now);
RR->topology->eachPeer<_PingPeersThatNeedPing &>(pfunc);
bondCheckInterval = std::max(RR->bc->minReqMonitorInterval(), ZT_CORE_TIMER_TASK_GRANULARITY);
if ((now - _lastGratuitousPingCheck) >= ZT_CORE_TIMER_TASK_GRANULARITY) {
_lastGratuitousPingCheck = now;
RR->bc->processBackgroundTasks(tptr, now);
}
RR->bc->processBackgroundTasks(tptr, now);
}
unsigned long timeUntilNextPingCheck = ZT_PING_CHECK_INVERVAL;
@ -512,7 +508,7 @@ ZT_PeerList *Node::peers() const
}
if (pi->second->bond()) {
p->isBonded = pi->second->bond();
p->bondingPolicy = pi->second->bond()->getPolicy();
p->bondingPolicy = pi->second->bond()->policy();
p->isHealthy = pi->second->bond()->isHealthy();
p->numAliveLinks = pi->second->bond()->getNumAliveLinks();
p->numTotalLinks = pi->second->bond()->getNumTotalLinks();

4
node/Node.hpp
View file

@ -34,7 +34,7 @@
#include "Salsa20.hpp"
#include "NetworkController.hpp"
#include "Hashtable.hpp"
#include "BondController.hpp"
#include "Bond.hpp"
// Bit mask for "expecting reply" hash
#define ZT_EXPECTING_REPLIES_BUCKET_MASK1 255
@ -187,7 +187,7 @@ public:
inline const Identity &identity() const { return _RR.identity; }
inline BondController *bondController() const { return _RR.bc; }
inline Bond *bondController() const { return _RR.bc; }
/**
* Register that we are expecting a reply to a packet ID

441
node/Path.hpp
View file

@ -29,8 +29,6 @@
#include "Packet.hpp"
#include "RingBuffer.hpp"
#include "../osdep/Link.hpp"
/**
* Maximum return value of preferenceRank()
*/
@ -88,46 +86,7 @@ public:
_localSocket(-1),
_latency(0xffff),
_addr(),
_ipScope(InetAddress::IP_SCOPE_NONE),
_lastAckReceived(0),
_lastAckSent(0),
_lastQoSMeasurement(0),
_lastThroughputEstimation(0),
_lastRefractoryUpdate(0),
_lastAliveToggle(0),
_lastEligibilityState(false),
_lastTrialBegin(0),
_refractoryPeriod(0),
_monitorInterval(0),
_upDelay(0),
_downDelay(0),
_ipvPref(0),
_mode(0),
_onlyPathOnLink(false),
_enabled(false),
_bonded(false),
_negotiated(false),
_deprecated(false),
_shouldReallocateFlows(false),
_assignedFlowCount(0),
_latencyMean(0),
_latencyVariance(0),
_packetLossRatio(0),
_packetErrorRatio(0),
_throughputMean(0),
_throughputMax(0),
_throughputVariance(0),
_allocation(0),
_byteLoad(0),
_relativeByteLoad(0),
_affinity(0),
_failoverScore(0),
_unackedBytes(0),
_packetsReceivedSinceLastAck(0),
_packetsReceivedSinceLastQoS(0),
_bytesAckedSinceLastThroughputEstimation(0),
_packetsIn(0),
_packetsOut(0)
_ipScope(InetAddress::IP_SCOPE_NONE)
{}
Path(const int64_t localSocket,const InetAddress &addr) :
@ -137,46 +96,7 @@ public:
_localSocket(localSocket),
_latency(0xffff),
_addr(addr),
_ipScope(addr.ipScope()),
_lastAckReceived(0),
_lastAckSent(0),
_lastQoSMeasurement(0),
_lastThroughputEstimation(0),
_lastRefractoryUpdate(0),
_lastAliveToggle(0),
_lastEligibilityState(false),
_lastTrialBegin(0),
_refractoryPeriod(0),
_monitorInterval(0),
_upDelay(0),
_downDelay(0),
_ipvPref(0),
_mode(0),
_onlyPathOnLink(false),
_enabled(false),
_bonded(false),
_negotiated(false),
_deprecated(false),
_shouldReallocateFlows(false),
_assignedFlowCount(0),
_latencyMean(0),
_latencyVariance(0),
_packetLossRatio(0),
_packetErrorRatio(0),
_throughputMean(0),
_throughputMax(0),
_throughputVariance(0),
_allocation(0),
_byteLoad(0),
_relativeByteLoad(0),
_affinity(0),
_failoverScore(0),
_unackedBytes(0),
_packetsReceivedSinceLastAck(0),
_packetsReceivedSinceLastQoS(0),
_bytesAckedSinceLastThroughputEstimation(0),
_packetsIn(0),
_packetsOut(0)
_ipScope(addr.ipScope())
{}
/**
@ -186,9 +106,6 @@ public:
*/
inline void received(const uint64_t t)
{
if (!alive(t,_bonded)) {
_lastAliveToggle = _lastIn;
}
_lastIn = t;
}
@ -317,21 +234,11 @@ public:
return (((age < (ZT_PATH_HEARTBEAT_PERIOD + 5000)) ? l : (l + 0xffff + age)) * (long)((ZT_INETADDRESS_MAX_SCOPE - _ipScope) + 1));
}
/**
* @param bonded Whether this path is part of a bond.
*/
inline void setBonded(bool bonded) { _bonded = bonded; }
/**
* @return True if this path is currently part of a bond.
*/
inline bool bonded() { return _bonded; }
/**
* @return True if this path is alive (receiving heartbeats)
*/
inline bool alive(const int64_t now, bool bondingEnabled = false) const {
return (bondingEnabled && _monitorInterval) ? ((now - _lastIn) < (_monitorInterval * 3)) : ((now - _lastIn) < (ZT_PATH_HEARTBEAT_PERIOD + 5000));
inline bool alive(const int64_t now) const {
return (now - _lastIn) < (ZT_PATH_HEARTBEAT_PERIOD + 5000);
}
/**
@ -339,11 +246,6 @@ public:
*/
inline bool needsHeartbeat(const int64_t now) const { return ((now - _lastOut) >= ZT_PATH_HEARTBEAT_PERIOD); }
/**
* @return True if this path needs a heartbeat in accordance to the user-specified path monitor frequency
*/
inline bool needsGratuitousHeartbeat(const int64_t now) { return allowed() && (_monitorInterval > 0) && ((now - _lastOut) >= _monitorInterval); }
/**
* @return Last time we sent something
*/
@ -364,134 +266,7 @@ public:
*/
inline int64_t lastTrustEstablishedPacketReceived() const { return _lastTrustEstablishedPacketReceived; }
/**
* @return Time since last VERB_ACK was received
*/
inline int64_t ackAge(int64_t now) { return _lastAckReceived ? now - _lastAckReceived : 0; }
/**
* Set or update a refractory period for the path.
*
* @param punishment How much a path should be punished
* @param pathFailure Whether this call is the result of a recent path failure
*/
inline void adjustRefractoryPeriod(int64_t now, uint32_t punishment, bool pathFailure) {
if (pathFailure) {
unsigned int suggestedRefractoryPeriod = _refractoryPeriod ? punishment + (_refractoryPeriod * 2) : punishment;
_refractoryPeriod = std::min(suggestedRefractoryPeriod, (unsigned int)ZT_MULTIPATH_MAX_REFRACTORY_PERIOD);
_lastRefractoryUpdate = 0;
} else {
uint32_t drainRefractory = 0;
if (_lastRefractoryUpdate) {
drainRefractory = (now - _lastRefractoryUpdate);
} else {
drainRefractory = (now - _lastAliveToggle);
}
_lastRefractoryUpdate = now;
if (_refractoryPeriod > drainRefractory) {
_refractoryPeriod -= drainRefractory;
} else {
_refractoryPeriod = 0;
_lastRefractoryUpdate = 0;
}
}
}
/**
* Determine the current state of eligibility of the path.
*
* @param includeRefractoryPeriod Whether current punishment should be taken into consideration
* @return True if this path can be used in a bond at the current time
*/
inline bool eligible(uint64_t now, int ackSendInterval, bool includeRefractoryPeriod = false) {
if (includeRefractoryPeriod && _refractoryPeriod) {
return false;
}
bool acceptableAge = age(now) < ((_monitorInterval * 4) + _downDelay); // Simple RX age (driven by packets of any type and gratuitous VERB_HELLOs)
bool acceptableAckAge = ackAge(now) < (ackSendInterval); // Whether the remote peer is actually responding to our outgoing traffic or simply sending stuff to us
bool notTooEarly = (now - _lastAliveToggle) >= _upDelay; // Whether we've waited long enough since the link last came online
bool inTrial = (now - _lastTrialBegin) < _upDelay; // Whether this path is still in its trial period
bool currEligibility = allowed() && (((acceptableAge || acceptableAckAge) && notTooEarly) || inTrial);
return currEligibility;
}
/**
* Record when this path first entered the bond. Each path is given a trial period where it is admitted
* to the bond without requiring observations to prove its performance or reliability.
*/
inline void startTrial(uint64_t now) { _lastTrialBegin = now; }
/**
* @return True if a path is permitted to be used in a bond (according to user pref.)
*/
inline bool allowed() {
return _enabled
&& (!_ipvPref
|| ((_addr.isV4() && (_ipvPref == 4 || _ipvPref == 46 || _ipvPref == 64))
|| ((_addr.isV6() && (_ipvPref == 6 || _ipvPref == 46 || _ipvPref == 64)))));
}
/**
* @return True if a path is preferred over another on the same physical link (according to user pref.)
*/
inline bool preferred() {
return _onlyPathOnLink
|| (_addr.isV4() && (_ipvPref == 4 || _ipvPref == 46))
|| (_addr.isV6() && (_ipvPref == 6 || _ipvPref == 64));
}
/**
* @param now Current time
* @return Whether an ACK (VERB_ACK) packet needs to be emitted at this time
*/
inline bool needsToSendAck(int64_t now, int ackSendInterval) {
return ((now - _lastAckSent) >= ackSendInterval ||
(_packetsReceivedSinceLastAck == ZT_QOS_TABLE_SIZE)) && _packetsReceivedSinceLastAck;
}
/**
* @param now Current time
* @return Whether a QoS (VERB_QOS_MEASUREMENT) packet needs to be emitted at this time
*/
inline bool needsToSendQoS(int64_t now, int qosSendInterval) {
return ((_packetsReceivedSinceLastQoS >= ZT_QOS_TABLE_SIZE) ||
((now - _lastQoSMeasurement) > qosSendInterval)) && _packetsReceivedSinceLastQoS;
}
/**
* Reset packet counters
*/
inline void resetPacketCounts()
{
_packetsIn = 0;
_packetsOut = 0;
}
/**
* The mean latency (computed from a sliding window.)
*/
float latencyMean() { return _latencyMean; }
/**
* Packet delay variance (computed from a sliding window.)
*/
float latencyVariance() { return _latencyVariance; }
/**
* The ratio of lost packets to received packets.
*/
float packetLossRatio() { return _packetLossRatio; }
/**
* The ratio of packets that failed their MAC/CRC checks to those that did not.
*/
float packetErrorRatio() { return _packetErrorRatio; }
/**
*
*/
uint8_t allocation() { return _allocation; }
void *_bondingMetricPtr;
private:
@ -503,212 +278,6 @@ private:
InetAddress _addr;
InetAddress::IpScope _ipScope; // memoize this since it's a computed value checked often
AtomicCounter __refCount;
std::map<uint64_t,uint64_t> qosStatsOut; // id:egress_time
std::map<uint64_t,uint64_t> qosStatsIn; // id:now
std::map<uint64_t,uint16_t> ackStatsIn; // id:len
RingBuffer<int,ZT_QOS_SHORTTERM_SAMPLE_WIN_SIZE> qosRecordSize;
RingBuffer<float,ZT_QOS_SHORTTERM_SAMPLE_WIN_SIZE> qosRecordLossSamples;
RingBuffer<uint64_t,ZT_QOS_SHORTTERM_SAMPLE_WIN_SIZE> throughputSamples;
RingBuffer<bool,ZT_QOS_SHORTTERM_SAMPLE_WIN_SIZE> packetValiditySamples;
RingBuffer<float,ZT_QOS_SHORTTERM_SAMPLE_WIN_SIZE> _throughputVarianceSamples;
RingBuffer<uint16_t,ZT_QOS_SHORTTERM_SAMPLE_WIN_SIZE> latencySamples;
/**
* Last time that a VERB_ACK was received on this path.
*/
uint64_t _lastAckReceived;
/**
* Last time that a VERB_ACK was sent out on this path.
*/
uint64_t _lastAckSent;
/**
* Last time that a VERB_QOS_MEASUREMENT was sent out on this path.
*/
uint64_t _lastQoSMeasurement;
/**
* Last time that the path's throughput was estimated.
*/
uint64_t _lastThroughputEstimation;
/**
* The last time that the refractory period was updated.
*/
uint64_t _lastRefractoryUpdate;
/**
* The last time that the path was marked as "alive".
*/
uint64_t _lastAliveToggle;
/**
* State of eligibility at last check. Used for determining state changes.
*/
bool _lastEligibilityState;
/**
* Timestamp indicating when this path's trial period began.
*/
uint64_t _lastTrialBegin;
/**
* Amount of time that this path will be prevented from becoming a member of a bond.
*/
uint32_t _refractoryPeriod;
/**
* Monitor interval specific to this path or that was inherited from the bond controller.
*/
int32_t _monitorInterval;
/**
* Up delay interval specific to this path or that was inherited from the bond controller.
*/
uint32_t _upDelay;
/**
* Down delay interval specific to this path or that was inherited from the bond controller.
*/
uint32_t _downDelay;
/**
* IP version preference inherited from the physical link.
*/
uint8_t _ipvPref;
/**
* Mode inherited from the physical link.
*/
uint8_t _mode;
/**
* IP version preference inherited from the physical link.
*/
bool _onlyPathOnLink;
/**
* Enabled state inherited from the physical link.
*/
bool _enabled;
/**
* Whether this path is currently part of a bond.
*/
bool _bonded;
/**
* Whether this path was intentionally negotiated by either peer.
*/
bool _negotiated;
/**
* Whether this path has been deprecated due to performance issues. Current traffic flows
* will be re-allocated to other paths in the most non-disruptive manner (if possible),
* and new traffic will not be allocated to this path.
*/
bool _deprecated;
/**
* Whether flows should be moved from this path. Current traffic flows will be re-allocated
* immediately.
*/
bool _shouldReallocateFlows;
/**
* The number of flows currently assigned to this path.
*/
uint16_t _assignedFlowCount;
/**
* The mean latency (computed from a sliding window.)
*/
float _latencyMean;
/**
* Packet delay variance (computed from a sliding window.)
*/
float _latencyVariance;
/**
* The ratio of lost packets to received packets.
*/
float _packetLossRatio;
/**
* The ratio of packets that failed their MAC/CRC checks to those that did not.
*/
float _packetErrorRatio;
/**
* The estimated mean throughput of this path.
*/
uint64_t _throughputMean;
/**
* The maximum observed throughput of this path.
*/
uint64_t _throughputMax;
/**
* The variance in the estimated throughput of this path.
*/
float _throughputVariance;
/**
* The relative quality of this path to all others in the bond, [0-255].
*/
uint8_t _allocation;
/**
* How much load this path is under.
*/
uint64_t _byteLoad;
/**
* How much load this path is under (relative to other paths in the bond.)
*/
uint8_t _relativeByteLoad;
/**
* Relative value expressing how "deserving" this path is of new traffic.
*/
uint8_t _affinity;
/**
* Score that indicates to what degree this path is preferred over others that
* are available to the bonding policy. (specifically for active-backup)
*/
uint32_t _failoverScore;
/**
* Number of bytes thus far sent that have not been acknowledged by the remote peer.
*/
int64_t _unackedBytes;
/**
* Number of packets received since the last VERB_ACK was sent to the remote peer.
*/
int32_t _packetsReceivedSinceLastAck;
/**
* Number of packets received since the last VERB_QOS_MEASUREMENT was sent to the remote peer.
*/
int32_t _packetsReceivedSinceLastQoS;
/**
* Bytes acknowledged via incoming VERB_ACK since the last estimation of throughput.
*/
uint64_t _bytesAckedSinceLastThroughputEstimation;
/**
* Counters used for tracking path load.
*/
int _packetsIn;
int _packetsOut;
};
} // namespace ZeroTier

52
node/Peer.cpp
View file

@ -50,12 +50,7 @@ Peer::Peer(const RuntimeEnvironment *renv,const Identity &myIdentity,const Ident
_directPathPushCutoffCount(0),
_credentialsCutoffCount(0),
_echoRequestCutoffCount(0),
_uniqueAlivePathCount(0),
_localMultipathSupported(false),
_remoteMultipathSupported(false),
_canUseMultipath(false),
_shouldCollectPathStatistics(0),
_bondingPolicy(0),
_lastComputedAggregateMeanLatency(0)
{
if (!myIdentity.agree(peerIdentity,_key))
@ -229,7 +224,8 @@ void Peer::received(
SharedPtr<Path> Peer::getAppropriatePath(int64_t now, bool includeExpired, int32_t flowId)
{
if (!_bondToPeer) {
Mutex::Lock _l(_bond_m);
if (!_bond) {
Mutex::Lock _l(_paths_m);
unsigned int bestPath = ZT_MAX_PEER_NETWORK_PATHS;
/**
@ -253,7 +249,7 @@ SharedPtr<Path> Peer::getAppropriatePath(int64_t now, bool includeExpired, int32
}
return SharedPtr<Path>();
}
return _bondToPeer->getAppropriatePath(now, flowId);
return _bond->getAppropriatePath(now, flowId);
}
void Peer::introduce(void *const tPtr,const int64_t now,const SharedPtr<Peer> &other) const
@ -444,39 +440,22 @@ void Peer::tryMemorizedPath(void *tPtr,int64_t now)
void Peer::performMultipathStateCheck(void *tPtr, int64_t now)
{
Mutex::Lock _l(_bond_m);
/**
* Check for conditions required for multipath bonding and create a bond
* if allowed.
*/
_localMultipathSupported = ((RR->bc->inUse()) && (ZT_PROTO_VERSION > 9));
if (_localMultipathSupported) {
int currAlivePathCount = 0;
int duplicatePathsFound = 0;
for (unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
if (_paths[i].p) {
currAlivePathCount++;
for (unsigned int j=0;j<ZT_MAX_PEER_NETWORK_PATHS;++j) {
if (_paths[i].p && _paths[j].p && _paths[i].p->address().ipsEqual2(_paths[j].p->address()) && i != j) {
duplicatePathsFound+=1;
break;
}
}
}
}
_uniqueAlivePathCount = (currAlivePathCount - (duplicatePathsFound / 2));
_remoteMultipathSupported = _vProto > 9;
_canUseMultipath = _localMultipathSupported && _remoteMultipathSupported && (_uniqueAlivePathCount > 1);
}
if (_canUseMultipath && !_bondToPeer) {
if (_localMultipathSupported && !_bond) {
if (RR->bc) {
_bondToPeer = RR->bc->createTransportTriggeredBond(RR, this);
_bond = RR->bc->createTransportTriggeredBond(RR, this);
/**
* Allow new bond to retroactively learn all paths known to this peer
*/
if (_bondToPeer) {
if (_bond) {
for (unsigned int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
if (_paths[i].p) {
_bondToPeer->nominatePath(_paths[i].p, now);
_bond->nominatePathToBond(_paths[i].p, now);
}
}
}
@ -510,8 +489,7 @@ unsigned int Peer::doPingAndKeepalive(void *tPtr,int64_t now)
if (_paths[i].p) {
// Clean expired and reduced priority paths
if ( ((now - _paths[i].lr) < ZT_PEER_PATH_EXPIRATION) && (_paths[i].priority == maxPriority) ) {
if ((sendFullHello)||(_paths[i].p->needsHeartbeat(now))
|| (_canUseMultipath && _paths[i].p->needsGratuitousHeartbeat(now))) {
if ((sendFullHello)||(_paths[i].p->needsHeartbeat(now))) {
attemptToContactAt(tPtr,_paths[i].p->localSocket(),_paths[i].p->address(),now,sendFullHello);
_paths[i].p->sent(now);
sent |= (_paths[i].p->address().ss_family == AF_INET) ? 0x1 : 0x2;
@ -591,27 +569,27 @@ void Peer::resetWithinScope(void *tPtr,InetAddress::IpScope scope,int inetAddres
void Peer::recordOutgoingPacket(const SharedPtr<Path> &path, const uint64_t packetId,
uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)
{
if (!_shouldCollectPathStatistics || !_bondToPeer) {
if (!_shouldCollectPathStatistics || !_bond) {
return;
}
_bondToPeer->recordOutgoingPacket(path, packetId, payloadLength, verb, flowId, now);
_bond->recordOutgoingPacket(path, packetId, payloadLength, verb, flowId, now);
}
void Peer::recordIncomingInvalidPacket(const SharedPtr<Path>& path)
{
if (!_shouldCollectPathStatistics || !_bondToPeer) {
if (!_shouldCollectPathStatistics || !_bond) {
return;
}
_bondToPeer->recordIncomingInvalidPacket(path);
_bond->recordIncomingInvalidPacket(path);
}
void Peer::recordIncomingPacket(const SharedPtr<Path> &path, const uint64_t packetId,
uint16_t payloadLength, const Packet::Verb verb, const int32_t flowId, int64_t now)
{
if (!_shouldCollectPathStatistics || !_bondToPeer) {
if (!_shouldCollectPathStatistics || !_bond) {
return;
}
_bondToPeer->recordIncomingPacket(path, packetId, payloadLength, verb, flowId, now);
_bond->recordIncomingPacket(path, packetId, payloadLength, verb, flowId, now);
}
} // namespace ZeroTier

30
node/Peer.hpp
View file

@ -33,7 +33,6 @@
#include "Hashtable.hpp"
#include "Mutex.hpp"
#include "Bond.hpp"
#include "BondController.hpp"
#include "AES.hpp"
#define ZT_PEER_MAX_SERIALIZED_STATE_SIZE (sizeof(Peer) + 32 + (sizeof(Path) * 2))
@ -305,12 +304,13 @@ public:
*/
inline unsigned int latency(const int64_t now)
{
if (_canUseMultipath) {
if (_localMultipathSupported) {
return (int)_lastComputedAggregateMeanLatency;
} else {
SharedPtr<Path> bp(getAppropriatePath(now,false));
if (bp)
if (bp) {
return bp->latency();
}
return 0xffff;
}
}
@ -503,16 +503,20 @@ public:
}
/**
*
* @return
* @return The bonding policy used to reach this peer
*/
SharedPtr<Bond> bond() { return _bondToPeer; }
SharedPtr<Bond> bond() { return _bond; }
/**
*
* @return
* @return The bonding policy used to reach this peer
*/
inline int8_t bondingPolicy() { return _bondingPolicy; }
inline int8_t bondingPolicy() {
Mutex::Lock _l(_paths_m);
if (_bond) {
return _bond->policy();
}
return ZT_BOND_POLICY_NONE;
}
//inline const AES *aesKeysIfSupported() const
//{ return (const AES *)0; }
@ -562,6 +566,7 @@ private:
_PeerPath _paths[ZT_MAX_PEER_NETWORK_PATHS];
Mutex _paths_m;
Mutex _bond_m;
Identity _id;
@ -571,18 +576,13 @@ private:
AtomicCounter __refCount;
bool _remotePeerMultipathEnabled;
int _uniqueAlivePathCount;
bool _localMultipathSupported;
bool _remoteMultipathSupported;
bool _canUseMultipath;
volatile bool _shouldCollectPathStatistics;
volatile int8_t _bondingPolicy;
int32_t _lastComputedAggregateMeanLatency;
SharedPtr<Bond> _bondToPeer;
SharedPtr<Bond> _bond;
};
} // namespace ZeroTier

4
node/RuntimeEnvironment.hpp
View file

@ -30,7 +30,7 @@ class Multicaster;
class NetworkController;
class SelfAwareness;
class Trace;
class BondController;
class Bond;
/**
* Holds global state for an instance of ZeroTier::Node
@ -76,7 +76,7 @@ public:
Multicaster *mc;
Topology *topology;
SelfAwareness *sa;
BondController *bc;
Bond *bc;
// This node's identity and string representations thereof
Identity identity;

4
node/Switch.cpp
View file

@ -1003,14 +1003,12 @@ bool Switch::_trySend(void *tPtr,Packet &packet,bool encrypt,int32_t flowId)
const SharedPtr<Peer> peer(RR->topology->getPeer(tPtr,destination));
if (peer) {
if ((peer->bondingPolicy() == ZT_BONDING_POLICY_BROADCAST)
if ((peer->bondingPolicy() == ZT_BOND_POLICY_BROADCAST)
&& (packet.verb() == Packet::VERB_FRAME || packet.verb() == Packet::VERB_EXT_FRAME)) {
const SharedPtr<Peer> relay(RR->topology->getUpstreamPeer());
Mutex::Lock _l(peer->_paths_m);
for(int i=0;i<ZT_MAX_PEER_NETWORK_PATHS;++i) {
if (peer->_paths[i].p && peer->_paths[i].p->alive(now)) {
char pathStr[128];
peer->_paths[i].p->address().toString(pathStr);
_sendViaSpecificPath(tPtr,peer,peer->_paths[i].p,now,packet,encrypt,flowId);
}
}