ExternalVendorCode/ZeroTierOne
1
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2025-02-20 09:46:13 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Improved multipath link monitoring

Browse Source
This commit is contained in:
Joseph Henry 2022-09-20 14:27:34 -07:00
parent 0797adf223
commit bc521504ca
No known key found for this signature in database
GPG Key ID: C45B33FF5EBC9344
9 changed files with 232 additions and 255 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
include/ZeroTierOne.h
View File

@ -86,6 +86,11 @@ extern "C" {
*/
#define ZT_MIN_PHYSMTU 1400
/**
* Maximum physical interface name length. This number is gigantic because of Windows.
*/
#define ZT_MAX_PHYSIFNAME 256
/**
* Default UDP payload size (physical path MTU) not including UDP and IP overhead
*
@ -1317,35 +1322,20 @@ typedef struct
*/
float packetErrorRatio;
/**
* Mean throughput
*/
uint64_t throughputMean;
/**
* Maximum observed throughput
*/
float throughputMax;
/**
* Throughput variance
*/
float throughputVariance;
/**
* Address scope
*/
uint8_t scope;
/**
* Percentage of traffic allocated to this path
* Percentage of traffic allocated to this path (0-255)
*/
float allocation;
uint8_t allocation;
/**
* Name of physical interface (for monitoring)
* Name of physical interface this path resides on
*/
char ifname[32];
char ifname[ZT_MAX_PHYSIFNAME];
uint64_t localSocket;
@ -1354,6 +1344,21 @@ typedef struct
*/
int expired;
/**
* Whether this path is currently included in the bond
*/
uint8_t bonded;
/**
* Whether this path is currently eligible to be used in a bond
*/
uint8_t eligible;
/**
* The speed of this link (as given to bonding layer)
*/
uint32_t linkSpeed;
/**
* Is path preferred?
*/

32
node/Bond.cpp
View File

@ -28,6 +28,8 @@ uint8_t Bond::_defaultPolicy = ZT_BOND_POLICY_NONE;
Phy<Bond*>* Bond::_phy;
Binder* Bond::_binder;
Mutex Bond::_bonds_m;
Mutex Bond::_links_m;
@ -158,13 +160,13 @@ void Bond::destroyBond(uint64_t peerId)
SharedPtr<Link> Bond::getLinkBySocket(const std::string& policyAlias, uint64_t localSocket, bool createIfNeeded = false)
{
Mutex::Lock _l(_links_m);
char ifname[64] = { 0 };
_phy->getIfName((PhySocket*)((uintptr_t)localSocket), ifname, sizeof(ifname) - 1);
char ifname[ZT_MAX_PHYSIFNAME] = {};
_binder->getIfName((PhySocket*)((uintptr_t)localSocket), ifname, sizeof(ifname) - 1);
std::string ifnameStr(ifname);
auto search = _interfaceToLinkMap[policyAlias].find(ifnameStr);
if (search == _interfaceToLinkMap[policyAlias].end()) {
if (createIfNeeded) {
SharedPtr<Link> s = new Link(ifnameStr, 0, 0, true, ZT_BOND_SLAVE_MODE_PRIMARY, "", 0.0);
SharedPtr<Link> s = new Link(ifnameStr, 0, 0, true, ZT_BOND_SLAVE_MODE_PRIMARY, "");
_interfaceToLinkMap[policyAlias].insert(std::pair<std::string, SharedPtr<Link> >(ifnameStr, s));
return s;
}
@ -250,6 +252,12 @@ void Bond::nominatePathToBond(const SharedPtr<Path>& path, int64_t now)
}
}
if (! alreadyPresent) {
SharedPtr<Link> link = getLink(path);
if (link) {
std::string ifnameStr = std::string(link->ifname());
memset(path->_ifname, 0x0, ZT_MAX_PHYSIFNAME);
memcpy(path->_ifname, ifnameStr.c_str(), std::min((int)ifnameStr.length(), ZT_MAX_PHYSIFNAME));
}
/**
* Find somewhere to stick it
*/
@ -523,6 +531,7 @@ int32_t Bond::generateQoSPacket(int pathIdx, int64_t now, char* qosBuffer)
std::map<uint64_t, uint64_t>::iterator it = _paths[pathIdx].qosStatsIn.begin();
int i = 0;
int numRecords = std::min(_paths[pathIdx].packetsReceivedSinceLastQoS, ZT_QOS_TABLE_SIZE);
debug("numRecords=%3d, packetsReceivedSinceLastQoS=%3d, _paths[pathIdx].qosStatsIn.size()=%3lu", numRecords, _paths[pathIdx].packetsReceivedSinceLastQoS, _paths[pathIdx].qosStatsIn.size());
while (i < numRecords && it != _paths[pathIdx].qosStatsIn.end()) {
uint64_t id = it->first;
memcpy(qosBuffer, &id, sizeof(uint64_t));
@ -800,8 +809,8 @@ void Bond::sendQOS_MEASUREMENT(void* tPtr, int pathIdx, int64_t localSocket, con
Packet outp(_peer->_id.address(), RR->identity.address(), Packet::VERB_QOS_MEASUREMENT);
char qosData[ZT_QOS_MAX_PACKET_SIZE];
int16_t len = generateQoSPacket(pathIdx, _now, qosData);
debug("sending QOS via link %s (len=%d)", pathToStr(_paths[pathIdx].p).c_str(), len);
if (len) {
debug("sending QOS via link %s (len=%d)", pathToStr(_paths[pathIdx].p).c_str(), len);
outp.append(qosData, len);
if (atAddress) {
outp.armor(_peer->key(), false, _peer->aesKeysIfSupported());
@ -905,6 +914,7 @@ void Bond::curateBond(int64_t now, bool rebuildBond)
SharedPtr<Link> link = getLink(_paths[i].p);
if (! link) {
log("link is no longer valid, removing from bond");
_paths[i].p->_valid = false;
_paths[i] = NominatedPath();
_paths[i].p = SharedPtr<Path>();
continue;
@ -1109,6 +1119,7 @@ void Bond::curateBond(int64_t now, bool rebuildBond)
if (_policy == ZT_BOND_POLICY_BALANCE_RR) {
// Cause a RR reset since the current index might no longer be valid
_rrPacketsSentOnCurrLink = _packetsPerLink;
_rrIdx = 0;
}
}
}
@ -1166,9 +1177,13 @@ void Bond::estimatePathQuality(int64_t now)
_paths[i].p->_packetLossRatio = _paths[i].packetLossRatio;
_paths[i].p->_packetErrorRatio = _paths[i].packetErrorRatio;
_paths[i].p->_bonded = _paths[i].bonded;
_paths[i].p->_givenLinkSpeed = 0;//_paths[i].givenLinkSpeed;
_paths[i].p->_eligible = _paths[i].eligible;
// _valid is written elsewhere
_paths[i].p->_allocation = _paths[i].allocation;
SharedPtr<Link> link = RR->bc->getLinkBySocket(_policyAlias, _paths[i].p->localSocket());
if (link) {
_paths[i].p->_givenLinkSpeed = link->speed();
}
//_paths[i].packetErrorRatio = 1.0 - (_paths[i].packetValiditySamples.count() ? _paths[i].packetValiditySamples.mean() : 1.0);
// Drain unacknowledged QoS records
@ -1725,10 +1740,11 @@ void Bond::setBondParameters(int policy, SharedPtr<Bond> templateBond, bool useT
_policy = (policy <= ZT_BOND_POLICY_NONE || policy > ZT_BOND_POLICY_BALANCE_AWARE) ? _defaultPolicy : policy;
// Check if non-leaf to prevent spamming infrastructure
ZT_PeerRole role;
if (_peer) {
ZT_PeerRole role = RR->topology->role(_peer->address());
_isLeaf = (role != ZT_PEER_ROLE_PLANET && role != ZT_PEER_ROLE_MOON);
role = RR->topology->role(_peer->address());
}
_isLeaf = _peer ? (role != ZT_PEER_ROLE_PLANET && role != ZT_PEER_ROLE_MOON) : false;
// Flows

59
node/Bond.hpp
View File

@ -14,6 +14,7 @@
#ifndef ZT_BOND_HPP
#define ZT_BOND_HPP
#include "../osdep/Binder.hpp"
#include "../osdep/Phy.hpp"
#include "Packet.hpp"
#include "Path.hpp"
@ -122,7 +123,7 @@ class Link {
* @param failoverToLinkStr
* @param userSpecifiedAlloc
*/
Link(std::string ifnameStr, uint8_t ipvPref, uint32_t speed, bool enabled, uint8_t mode, std::string failoverToLinkStr, float userSpecifiedAlloc)
Link(std::string ifnameStr, uint8_t ipvPref, uint32_t speed, bool enabled, uint8_t mode, std::string failoverToLinkStr)
: _ifnameStr(ifnameStr)
, _ipvPref(ipvPref)
, _speed(speed)
@ -130,7 +131,6 @@ class Link {
, _enabled(enabled)
, _mode(mode)
, _failoverToLinkStr(failoverToLinkStr)
, _userSpecifiedAlloc(userSpecifiedAlloc)
, _isUserSpecified(false)
{
}
@ -287,11 +287,6 @@ class Link {
*/
std::string _failoverToLinkStr;
/**
* User-specified allocation
*/
float _userSpecifiedAlloc;
/**
* Whether or not this link was created as a result of manual user specification. This is
* important to know because certain policy decisions are dependent on whether the user
@ -328,6 +323,14 @@ class Bond {
return ! _bondPolicyTemplates.empty() || _defaultPolicy;
}
/**
* Sets a pointer to an instance of _binder used by the Bond to get interface data
*/
static void setBinder(Binder* b)
{
_binder = b;
}
/**
* @param basePolicyName Bonding policy name (See ZeroTierOne.h)
* @return The bonding policy code for a given human-readable bonding policy name
@ -461,7 +464,7 @@ class Bond {
* @param createIfNeeded Whether a Link object is created if the name wasn't previously in the link map
* @return Physical link definition
*/
static SharedPtr<Link> getLinkBySocket(const std::string& policyAlias, uint64_t localSocket, bool createIfNeeded);
SharedPtr<Link> getLinkBySocket(const std::string& policyAlias, uint64_t localSocket, bool createIfNeeded);
/**
* Gets a reference to a physical link definition given its human-readable system name.
@ -840,14 +843,6 @@ class Bond {
_maxAcceptablePacketErrorRatio = errorRatio;
}
/**
* @param errorRatio Maximum acceptable packet error ratio (PER).
*/
void setMinAcceptableAllocation(float minAlloc)
{
_minAcceptableAllocation = (uint8_t)(minAlloc * 255);
}
/**
* @return Whether the user has defined links for use on this bond
*/
@ -970,14 +965,6 @@ class Bond {
return _failoverInterval;
}
/**
* @param strategy Strategy that the bond uses to re-assign protocol flows.
*/
inline void setFlowRebalanceStrategy(uint32_t strategy)
{
_flowRebalanceStrategy = strategy;
}
/**
* @param strategy Strategy that the bond uses to prob for path aliveness and quality
*/
@ -1150,26 +1137,8 @@ class Bond {
*/
bool abForciblyRotateLink();
/**
* @param now Current time
* @return All known paths to this peer
*/
inline std::vector<SharedPtr<Path> > paths(const int64_t now) const
{
std::vector<SharedPtr<Path> > pp;
Mutex::Lock _l(_paths_m);
for (unsigned int i = 0; i < ZT_MAX_PEER_NETWORK_PATHS; ++i) {
if (! _paths[i].p) {
continue;
}
pp.push_back(_paths[i].p);
}
return pp;
}
/**
* Emit message to tracing system but with added timestamp and subsystem info
*
*/
void log(const char* fmt, ...)
#ifdef __GNUC__
@ -1201,7 +1170,6 @@ class Bond {
/**
* Emit message to tracing system but with added timestamp and subsystem info
*
*/
void debug(const char* fmt, ...)
#ifdef __GNUC__
@ -1412,7 +1380,6 @@ class Bond {
{
p = path;
whenNominated = now;
p->_bondingMetricPtr = (void*)this;
}
};
@ -1487,6 +1454,8 @@ class Bond {
std::string _policyAlias; // Custom name given by the user to this bond type.
static Binder* _binder;
/**
* Set of indices corresponding to paths currently included in the bond proper. This
* may only be updated during a call to curateBond(). The reason for this is so that
@ -1518,7 +1487,6 @@ class Bond {
// balance-aware
uint64_t _totalBondUnderload;
uint8_t _flowRebalanceStrategy;
// dynamic link monitoring
uint8_t _linkMonitorStrategy;
@ -1546,7 +1514,6 @@ class Bond {
uint16_t _maxAcceptableLatency;
uint16_t _maxAcceptableMeanLatency;
uint16_t _maxAcceptablePacketDelayVariance;
uint8_t _minAcceptableAllocation;
/**
* Link state reporting

33
node/Node.cpp
View File

@ -496,15 +496,30 @@ ZT_PeerList *Node::peers() const
SharedPtr<Path> bestp(pi->second->getAppropriatePath(_now,false));
p->pathCount = 0;
for(std::vector< SharedPtr<Path> >::iterator path(paths.begin());path!=paths.end();++path) {
memcpy(&(p->paths[p->pathCount].address),&((*path)->address()),sizeof(struct sockaddr_storage));
p->paths[p->pathCount].localSocket = (*path)->localSocket();
p->paths[p->pathCount].lastSend = (*path)->lastOut();
p->paths[p->pathCount].lastReceive = (*path)->lastIn();
p->paths[p->pathCount].trustedPathId = RR->topology->getOutboundPathTrust((*path)->address());
p->paths[p->pathCount].expired = 0;
p->paths[p->pathCount].preferred = ((*path) == bestp) ? 1 : 0;
p->paths[p->pathCount].scope = (*path)->ipScope();
++p->pathCount;
if((*path)->valid()) {
memcpy(&(p->paths[p->pathCount].address),&((*path)->address()),sizeof(struct sockaddr_storage));
p->paths[p->pathCount].localSocket = (*path)->localSocket();
p->paths[p->pathCount].lastSend = (*path)->lastOut();
p->paths[p->pathCount].lastReceive = (*path)->lastIn();
p->paths[p->pathCount].trustedPathId = RR->topology->getOutboundPathTrust((*path)->address());
p->paths[p->pathCount].expired = 0;
p->paths[p->pathCount].preferred = ((*path) == bestp) ? 1 : 0;
p->paths[p->pathCount].scope = (*path)->ipScope();
if (pi->second->bond()) {
p->paths[p->pathCount].latencyMean = (*path)->latencyMean();
p->paths[p->pathCount].latencyVariance = (*path)->latencyVariance();
p->paths[p->pathCount].packetLossRatio = (*path)->packetLossRatio();
p->paths[p->pathCount].packetErrorRatio = (*path)->packetErrorRatio();
p->paths[p->pathCount].allocation = (*path)->allocation();
p->paths[p->pathCount].linkSpeed = (*path)->givenLinkSpeed();
p->paths[p->pathCount].bonded = (*path)->bonded();
p->paths[p->pathCount].eligible = (*path)->eligible();
std::string ifname = std::string((*path)->ifname());
memset(p->paths[p->pathCount].ifname, 0x0, std::min((int)ifname.length() + 1, ZT_MAX_PHYSIFNAME));
memcpy(p->paths[p->pathCount].ifname, ifname.c_str(), std::min((int)ifname.length(), ZT_MAX_PHYSIFNAME));
}
++p->pathCount;
}
}
if (pi->second->bond()) {
p->isBonded = pi->second->bond();

54
node/Path.hpp
View File

@ -85,6 +85,15 @@ public:
_lastTrustEstablishedPacketReceived(0),
_lastEchoRequestReceived(0),
_localSocket(-1),
_latencyMean(0.0),
_latencyVariance(0.0),
_packetLossRatio(0.0),
_packetErrorRatio(0.0),
_valid(true),
_eligible(false),
_bonded(false),
_givenLinkSpeed(0),
_allocation(0),
_latency(0xffff),
_addr(),
_ipScope(InetAddress::IP_SCOPE_NONE)
@ -96,6 +105,15 @@ public:
_lastTrustEstablishedPacketReceived(0),
_lastEchoRequestReceived(0),
_localSocket(localSocket),
_latencyMean(0.0),
_latencyVariance(0.0),
_packetLossRatio(0.0),
_packetErrorRatio(0.0),
_valid(true),
_eligible(false),
_bonded(false),
_givenLinkSpeed(0),
_allocation(0),
_latency(0xffff),
_addr(addr),
_ipScope(addr.ipScope())
@ -300,6 +318,17 @@ public:
*/
inline unsigned int packetErrorRatio() const { return _packetErrorRatio; }
/**
* @return Whether this path is valid as reported by the bonding layer. The bonding layer
* actually checks with Phy to see if the interface is still up
*/
inline unsigned int valid() const { return _valid; }
/**
* @return Whether this path is eligible for use in a bond as reported by the bonding layer
*/
inline unsigned int eligible() const { return _eligible; }
/**
* @return Whether this path is bonded as reported by the bonding layer
*/
@ -313,27 +342,36 @@ public:
/**
* @return Traffic allocation as reported by the bonding layer
*/
inline unsigned int allocation() const { return _allocation; }
inline unsigned char allocation() const { return _allocation; }
void *_bondingMetricPtr;
/**
* @return Physical interface name that this path lives on
*/
char *ifname() {
return _ifname;
}
private:
char _ifname[ZT_MAX_PHYSIFNAME] = { };
volatile int64_t _lastOut;
volatile int64_t _lastIn;
volatile int64_t _lastTrustEstablishedPacketReceived;
int64_t _lastEchoRequestReceived;
int64_t _localSocket;
volatile float _latencyMean;
volatile float _latencyVariance;
volatile float _packetLossRatio;
volatile float _packetErrorRatio;
volatile bool _valid;
volatile bool _eligible;
volatile bool _bonded;
volatile int64_t _givenLinkSpeed;
volatile int8_t _allocation;
int64_t _lastEchoRequestReceived;
int64_t _localSocket;
volatile uint32_t _givenLinkSpeed;
volatile uint8_t _allocation;
volatile unsigned int _latency;
InetAddress _addr;

44
one.cpp
View File

@ -611,41 +611,49 @@ static int cli(int argc,char **argv)
} else {
int numAliveLinks = OSUtils::jsonInt(j["numAliveLinks"],0);
int numTotalLinks = OSUtils::jsonInt(j["numTotalLinks"],0);
printf("Peer : %s\n", arg1.c_str());
printf("Bond : %s\n", OSUtils::jsonString(j["bondingPolicy"],"-").c_str());
printf("Link Select Method : %d\n", (int)OSUtils::jsonInt(j["linkSelectMethod"],0));
printf("Links : %d/%d\n", numAliveLinks, numTotalLinks);
printf("Failover Interval : %d (ms)\n", (int)OSUtils::jsonInt(j["failoverInterval"],0));
printf("Up Delay : %d (ms)\n", (int)OSUtils::jsonInt(j["upDelay"],0));
printf("Down Delay : %d (ms)\n", (int)OSUtils::jsonInt(j["downDelay"],0));
printf("Packets Per Link : %d (ms)\n", (int)OSUtils::jsonInt(j["packetsPerLink"],0));
nlohmann::json &p = j["links"];
printf("Peer : %s\n", arg1.c_str());
printf("Bond : %s\n", OSUtils::jsonString(j["bondingPolicyStr"],"-").c_str());
printf("Link Select Method : %d\n", (int)OSUtils::jsonInt(j["linkSelectMethod"],0));
printf("Links : %d/%d\n", numAliveLinks, numTotalLinks);
printf("Failover Interval (ms) : %d\n", (int)OSUtils::jsonInt(j["failoverInterval"],0));
printf("Up Delay (ms) : %d\n", (int)OSUtils::jsonInt(j["upDelay"],0));
printf("Down Delay (ms) : %d\n", (int)OSUtils::jsonInt(j["downDelay"],0));
printf("Packets Per Link : %d\n", (int)OSUtils::jsonInt(j["packetsPerLink"],0));
nlohmann::json &p = j["paths"];
if (p.is_array()) {
printf("\n interface\t\t\t\t\t path\n");
for(int i=0; i<80; i++) { printf("-"); }
printf("\nidx"
" interface"
" "
"path socket\n");
for(int i=0; i<100; i++) { printf("-"); }
printf("\n");
for (int i=0; i<p.size(); i++)
{
printf("[%3d] %21s %50s\n",
printf("%2d: %26s %51s %.16llx\n",
i,
OSUtils::jsonString(p[i]["ifname"],"-").c_str(),
OSUtils::jsonString(p[i]["path"],"-").c_str()
OSUtils::jsonString(p[i]["address"],"-").c_str(),
(unsigned long long)OSUtils::jsonInt(p[i]["localSocket"],0)
);
}
printf("\n lat pdv plr per speed alloc alive bonded\n");
for(int i=0; i<80; i++) { printf("-"); }
printf("\nidx lat pdv "
"plr per speed alloc "
"rx_age tx_age eligible bonded\n");
for(int i=0; i<100; i++) { printf("-"); }
printf("\n");
for (int i=0; i<p.size(); i++)
{
printf("[%3d] %7.2f %7.2f %6.2f %6.2f %8d %9d %8d %8d\n",
printf("%2d: %8.2f %8.2f %7.4f %7.4f %7d %6.2f %11d %11d %9d %7d\n",
i,
OSUtils::jsonDouble(p[i]["latencyMean"], 0),
OSUtils::jsonDouble(p[i]["latencyVariance"], 0),
OSUtils::jsonDouble(p[i]["packetLossRatio"], 0),
OSUtils::jsonDouble(p[i]["packetErrorRatio"], 0),
(int)OSUtils::jsonInt(p[i]["givenLinkSpeed"], 0),
(int)OSUtils::jsonInt(p[i]["allocation"], 0),
(int)OSUtils::jsonInt(p[i]["alive"],0),
OSUtils::jsonDouble(p[i]["allocation"], 0),
(int)OSUtils::jsonInt(p[i]["lastInAge"], 0),
(int)OSUtils::jsonInt(p[i]["lastOutAge"], 0),
(int)OSUtils::jsonInt(p[i]["eligible"],0),
(int)OSUtils::jsonInt(p[i]["bonded"],0));
}
}

22
osdep/Binder.hpp
View File

@ -66,6 +66,9 @@
// Max number of bindings
#define ZT_BINDER_MAX_BINDINGS 256
// Maximum physical interface name length. This number is gigantic because of Windows.
#define ZT_MAX_PHYSIFNAME 256
namespace ZeroTier {
/**
@ -88,6 +91,7 @@ class Binder {
PhySocket* udpSock;
PhySocket* tcpListenSock;
InetAddress address;
char ifname[256] = {};
};
public:
@ -443,7 +447,7 @@ class Binder {
_bindings[_bindingCount].udpSock = udps;
_bindings[_bindingCount].tcpListenSock = tcps;
_bindings[_bindingCount].address = ii->first;
phy.setIfName(udps, (char*)ii->second.c_str(), (int)ii->second.length());
memcpy(_bindings[_bindingCount].ifname, (char*)ii->second.c_str(), (int)ii->second.length());
++_bindingCount;
}
}
@ -514,6 +518,22 @@ class Binder {
return false;
}
/**
* @param s Socket object
* @param nameBuf Buffer to store name of interface which this Socket object is bound to
* @param buflen Length of buffer to copy name into
*/
void getIfName(PhySocket* s, char* nameBuf, int buflen) const
{
Mutex::Lock _l(_lock);
for (unsigned int b = 0, c = _bindingCount; b < c; ++b) {
if (_bindings[b].udpSock == s) {
memcpy(nameBuf, _bindings[b].ifname, buflen);
break;
}
}
}
private:
_Binding _bindings[ZT_BINDER_MAX_BINDINGS];
std::atomic<unsigned int> _bindingCount;

35
osdep/Phy.hpp
View File

@ -140,12 +140,11 @@ private:
};
struct PhySocketImpl {
PhySocketImpl() { memset(ifname, 0, sizeof(ifname)); }
PhySocketImpl() {}
PhySocketType type;
ZT_PHY_SOCKFD_TYPE sock;
void *uptr; // user-settable pointer
ZT_PHY_SOCKADDR_STORAGE_TYPE saddr; // remote for TCP_OUT and TCP_IN, local for TCP_LISTEN, RAW, and UDP
char ifname[256 + 4];
};
std::list<PhySocketImpl> _socks;
@ -243,38 +242,6 @@ public:
return &(reinterpret_cast<PhySocketImpl*>(s)->uptr);
}
/**
* @param s Socket object
* @param nameBuf Buffer to store name of interface which this Socket object is bound to
* @param buflen Length of buffer to copy name into
*/
static inline void getIfName(PhySocket* s, char* nameBuf, int buflen)
{
PhySocketImpl& sws = *(reinterpret_cast<PhySocketImpl*>(s));
if (sws.type == ZT_PHY_SOCKET_CLOSED) {
return;
}
if (s) {
memcpy(nameBuf, reinterpret_cast<PhySocketImpl*>(s)->ifname, buflen);
}
}
/**
* @param s Socket object
* @param ifname Buffer containing name of interface that this Socket object is bound to
* @param len Length of name of interface
*/
static inline void setIfName(PhySocket* s, char* ifname, int len)
{
PhySocketImpl& sws = *(reinterpret_cast<PhySocketImpl*>(s));
if (sws.type == ZT_PHY_SOCKET_CLOSED) {
return;
}
if (s) {
memcpy(&(reinterpret_cast<PhySocketImpl*>(s)->ifname), ifname, len);
}
}
/**
* Cause poll() to stop waiting immediately
*

165
service/OneService.cpp
View File

@ -520,7 +520,7 @@ static void _networkToJson(nlohmann::json &nj,NetworkState &ns)
}
}
static void _peerToJson(nlohmann::json &pj,const ZT_Peer *peer)
static void _peerToJson(nlohmann::json &pj,const ZT_Peer *peer, SharedPtr<Bond> &bond)
{
char tmp[256];
@ -541,10 +541,15 @@ static void _peerToJson(nlohmann::json &pj,const ZT_Peer *peer)
pj["latency"] = peer->latency;
pj["role"] = prole;
pj["isBonded"] = peer->isBonded;
if (peer->isBonded) {
pj["bondingPolicy"] = peer->bondingPolicy;
if (bond && peer->isBonded) {
pj["bondingPolicyCode"] = peer->bondingPolicy;
pj["bondingPolicyStr"] = Bond::getPolicyStrByCode(peer->bondingPolicy);
pj["numAliveLinks"] = peer->numAliveLinks;
pj["numTotalLinks"] = peer->numTotalLinks;
pj["failoverInterval"] = bond->getFailoverInterval();
pj["downDelay"] = bond->getDownDelay();
pj["upDelay"] = bond->getUpDelay();
pj["packetsPerLink"] = bond->getPacketsPerLink();
}
nlohmann::json pa = nlohmann::json::array();
@ -560,56 +565,25 @@ static void _peerToJson(nlohmann::json &pj,const ZT_Peer *peer)
j["expired"] = (bool)(peer->paths[i].expired != 0);
j["preferred"] = (bool)(peer->paths[i].preferred != 0);
j["localSocket"] = peer->paths[i].localSocket;
if (bond && peer->isBonded) {
uint64_t now = OSUtils::now();
j["ifname"] = std::string(peer->paths[i].ifname);
j["latencyMean"] = peer->paths[i].latencyMean;
j["latencyVariance"] = peer->paths[i].latencyVariance;
j["packetLossRatio"] = peer->paths[i].packetLossRatio;
j["packetErrorRatio"] = peer->paths[i].packetErrorRatio;
j["lastInAge"] = (now - lastReceive);
j["lastOutAge"] = (now - lastSend);
j["bonded"] = peer->paths[i].bonded;
j["eligible"] = peer->paths[i].eligible;
j["givenLinkSpeed"] = peer->paths[i].linkSpeed;
j["allocation"] = std::round(((float)(peer->paths[i].allocation) / 255.0) * 1000.0) / 1000.0;
}
pa.push_back(j);
}
pj["paths"] = pa;
}
static void _bondToJson(nlohmann::json &pj, SharedPtr<Bond> &bond)
{
uint64_t now = OSUtils::now();
int bondingPolicy = bond->policy();
pj["bondingPolicy"] = Bond::getPolicyStrByCode(bondingPolicy);
if (bondingPolicy == ZT_BOND_POLICY_NONE) {
return;
}
pj["numAliveLinks"] = bond->getNumAliveLinks();
pj["numTotalLinks"] = bond->getNumTotalLinks();
pj["failoverInterval"] = bond->getFailoverInterval();
pj["downDelay"] = bond->getDownDelay();
pj["upDelay"] = bond->getUpDelay();
if (bondingPolicy == ZT_BOND_POLICY_BALANCE_RR) {
pj["packetsPerLink"] = bond->getPacketsPerLink();
}
if (bondingPolicy == ZT_BOND_POLICY_ACTIVE_BACKUP) {
pj["linkSelectMethod"] = bond->getLinkSelectMethod();
}
nlohmann::json pa = nlohmann::json::array();
std::vector< SharedPtr<Path> > paths = bond->paths(now);
for(unsigned int i=0;i<paths.size();++i) {
char pathStr[128];
paths[i]->address().toString(pathStr);
nlohmann::json j;
j["ifname"] = bond->getLink(paths[i])->ifname();
j["path"] = pathStr;
j["latencyMean"] = paths[i]->latencyMean();
j["latencyVariance"] = paths[i]->latencyVariance();
j["packetLossRatio"] = paths[i]->packetLossRatio();
j["packetErrorRatio"] = paths[i]->packetErrorRatio();
j["alive"] = paths[i]->alive(now);
j["bonded"] = paths[i]->bonded();
j["givenLinkSpeed"] = paths[i]->givenLinkSpeed();
j["allocation"] = paths[i]->allocation();
pa.push_back(j);
}
pj["links"] = pa;
}
static void _moonToJson(nlohmann::json &mj,const World &world)
{
char tmp[4096];
@ -1496,23 +1470,28 @@ public:
if (ps[0] == "bond") {
if (_node->bondController()->inUse()) {
if (ps.size() == 3) {
//fprintf(stderr, "ps[0]=%s\nps[1]=%s\nps[2]=%s\n", ps[0].c_str(), ps[1].c_str(), ps[2].c_str());
if (ps[2].length() == 10) {
// check if hex string
const uint64_t id = Utils::hexStrToU64(ps[2].c_str());
if (ps[1] == "show") {
SharedPtr<Bond> bond = _node->bondController()->getBondByPeerId(id);
if (bond) {
_bondToJson(res,bond);
scode = 200;
} else {
fprintf(stderr, "unable to find bond to peer %llx\n", (unsigned long long)id);
scode = 400;
ZT_PeerList *pl = _node->peers();
if (pl) {
uint64_t wantp = Utils::hexStrToU64(ps[2].c_str());
for(unsigned long i=0;i<pl->peerCount;++i) {
if (pl->peers[i].address == wantp) {
if (ps[1] == "show") {
SharedPtr<Bond> bond = _node->bondController()->getBondByPeerId(wantp);
if (bond) {
_peerToJson(res,&(pl->peers[i]),bond);
scode = 200;
} else {
fprintf(stderr, "unable to find bond to peer %llx\n", (unsigned long long)wantp);
scode = 400;
}
}
}
}
}
if (ps[1] == "flows") {
fprintf(stderr, "displaying flows\n");
}
_node->freeQueryResult((void *)pl);
}
}
} else {
@ -1635,36 +1614,12 @@ public:
res = nlohmann::json::array();
for(unsigned long i=0;i<pl->peerCount;++i) {
nlohmann::json pj;
_peerToJson(pj,&(pl->peers[i]));
res.push_back(pj);
}
scode = 200;
} else if (ps.size() == 2) {
// Return a single peer by ID or 404 if not found
uint64_t wantp = Utils::hexStrToU64(ps[1].c_str());
for(unsigned long i=0;i<pl->peerCount;++i) {
if (pl->peers[i].address == wantp) {
_peerToJson(res,&(pl->peers[i]));
scode = 200;
break;
SharedPtr<Bond> bond = SharedPtr<Bond>();
if (pl->peers[i].isBonded) {
const uint64_t id = pl->peers[i].address;
bond = _node->bondController()->getBondByPeerId(id);
}
}
} else scode = 404;
_node->freeQueryResult((void *)pl);
} else scode = 500;
} else if (ps[0] == "bonds") {
ZT_PeerList *pl = _node->peers();
if (pl) {
if (ps.size() == 1) {
// Return [array] of all peers
res = nlohmann::json::array();
for(unsigned long i=0;i<pl->peerCount;++i) {
nlohmann::json pj;
_peerToJson(pj,&(pl->peers[i]));
_peerToJson(pj,&(pl->peers[i]),bond);
res.push_back(pj);
}
@ -1675,7 +1630,11 @@ public:
uint64_t wantp = Utils::hexStrToU64(ps[1].c_str());
for(unsigned long i=0;i<pl->peerCount;++i) {
if (pl->peers[i].address == wantp) {
_peerToJson(res,&(pl->peers[i]));
SharedPtr<Bond> bond = SharedPtr<Bond>();
if (pl->peers[i].isBonded) {
bond = _node->bondController()->getBondByPeerId(wantp);
}
_peerToJson(res,&(pl->peers[i]),bond);
scode = 200;
break;
}
@ -1769,11 +1728,11 @@ public:
if (ps[0] == "bond") {
if (_node->bondController()->inUse()) {
if (ps.size() == 3) {
//fprintf(stderr, "ps[0]=%s\nps[1]=%s\nps[2]=%s\n", ps[0].c_str(), ps[1].c_str(), ps[2].c_str());
if (ps[2].length() == 10) {
// check if hex string
const uint64_t id = Utils::hexStrToU64(ps[2].c_str());
if (ps[1] == "rotate") {
exit(0);
SharedPtr<Bond> bond = _node->bondController()->getBondByPeerId(id);
if (bond) {
scode = bond->abForciblyRotateLink() ? 200 : 400;
@ -1782,9 +1741,6 @@ public:
scode = 400;
}
}
if (ps[1] == "enable") {
fprintf(stderr, "enabling bond\n");
}
}
}
} else {
@ -2039,6 +1995,7 @@ public:
json &settings = lc["settings"];
if (!_node->bondController()->inUse()) {
_node->bondController()->setBinder(&_binder);
// defaultBondingPolicy
std::string defaultBondingPolicyStr(OSUtils::jsonString(settings["defaultBondingPolicy"],""));
int defaultBondingPolicy = _node->bondController()->getPolicyCodeByStr(defaultBondingPolicyStr);
@ -2073,7 +2030,6 @@ public:
newTemplateBond->setMaxAcceptablePacketDelayVariance(OSUtils::jsonInt(customPolicy["maxAcceptablePacketDelayVariance"],-1));
newTemplateBond->setMaxAcceptablePacketLossRatio((float)OSUtils::jsonDouble(customPolicy["maxAcceptablePacketLossRatio"],-1));
newTemplateBond->setMaxAcceptablePacketErrorRatio((float)OSUtils::jsonDouble(customPolicy["maxAcceptablePacketErrorRatio"],-1));
newTemplateBond->setMinAcceptableAllocation((float)OSUtils::jsonDouble(customPolicy["minAcceptableAllocation"],0));
// Quality weights
json &qualityWeights = customPolicy["qualityWeights"];
if (qualityWeights.size() == ZT_QOS_WEIGHT_SIZE) {
@ -2088,7 +2044,6 @@ public:
// Bond-specific properties
newTemplateBond->setUpDelay(OSUtils::jsonInt(customPolicy["upDelay"],-1));
newTemplateBond->setDownDelay(OSUtils::jsonInt(customPolicy["downDelay"],-1));
newTemplateBond->setFlowRebalanceStrategy(OSUtils::jsonInt(customPolicy["flowRebalanceStrategy"],(uint64_t)0));
newTemplateBond->setFailoverInterval(OSUtils::jsonInt(customPolicy["failoverInterval"],ZT_BOND_FAILOVER_DEFAULT_INTERVAL));
newTemplateBond->setPacketsPerLink(OSUtils::jsonInt(customPolicy["packetsPerLink"],-1));
@ -2097,16 +2052,8 @@ public:
for (json::iterator linkItr = links.begin(); linkItr != links.end();++linkItr) {
std::string linkNameStr(linkItr.key());
json &link = linkItr.value();
bool enabled = OSUtils::jsonInt(link["enabled"],true);
uint32_t speed = OSUtils::jsonInt(link["speed"],0);
float alloc = (float)OSUtils::jsonDouble(link["alloc"],0);
if (speed && alloc) {
fprintf(stderr, "error: cannot specify both speed (%d) and alloc (%f) for link (%s), pick one, link disabled.\n",
speed, alloc, linkNameStr.c_str());
enabled = false;
}
uint8_t ipvPref = OSUtils::jsonInt(link["ipvPref"],0);
std::string failoverToStr(OSUtils::jsonString(link["failoverTo"],""));
// Mode
@ -2124,7 +2071,7 @@ public:
failoverToStr = "";
enabled = false;
}
_node->bondController()->addCustomLink(customPolicyStr, new Link(linkNameStr,ipvPref,speed,enabled,linkMode,failoverToStr,alloc));
_node->bondController()->addCustomLink(customPolicyStr, new Link(linkNameStr,ipvPref,speed,enabled,linkMode,failoverToStr));
}
std::string linkSelectMethodStr(OSUtils::jsonString(customPolicy["activeReselect"],"optimize"));
if (linkSelectMethodStr == "always") {
@ -2142,12 +2089,6 @@ public:
if (newTemplateBond->getLinkSelectMethod() < 0 || newTemplateBond->getLinkSelectMethod() > 3) {
fprintf(stderr, "warning: invalid value (%s) for linkSelectMethod, assuming mode: always\n", linkSelectMethodStr.c_str());
}
/*
newBond->setPolicy(_node->bondController()->getPolicyCodeByStr(basePolicyStr));
newBond->setFlowHashing((bool)OSUtils::jsonInt(userSpecifiedBondingPolicies[i]["allowFlowHashing"],(bool)allowFlowHashing));
newBond->setBondMonitorInterval((unsigned int)OSUtils::jsonInt(userSpecifiedBondingPolicies[i]["monitorInterval"],(uint64_t)0));
newBond->setAllowPathNegotiation((bool)OSUtils::jsonInt(userSpecifiedBondingPolicies[i]["allowPathNegotiation"],(bool)false));
*/
if (!_node->bondController()->addCustomPolicy(newTemplateBond)) {
fprintf(stderr, "error: a custom policy of this name (%s) already exists.\n", customPolicyStr.c_str());
}