More multicast work...

This commit is contained in:
Adam Ierymenko 2019-09-10 16:20:28 -07:00
parent 592e743349
commit bccb86a401
No known key found for this signature in database
GPG key ID: C8877CF2D7A5D7F3
7 changed files with 410 additions and 230 deletions

View file

@ -258,12 +258,12 @@
/** /**
* Period for multicast LIKE re-announcements to connected nodes * Period for multicast LIKE re-announcements to connected nodes
*/ */
#define ZT_MULTICAST_ANNOUNCE_PERIOD 120000 #define ZT_MULTICAST_ANNOUNCE_PERIOD 60000
/** /**
* Delay between explicit MULTICAST_GATHER requests for a given multicast channel * Period for multicast GATHER on multicast groups
*/ */
#define ZT_MULTICAST_EXPLICIT_GATHER_DELAY (ZT_MULTICAST_LIKE_EXPIRE / 10) #define ZT_MULTICAST_GATHER_PERIOD ZT_MULTICAST_ANNOUNCE_PERIOD
/** /**
* Timeout for outgoing multicasts * Timeout for outgoing multicasts

View file

@ -17,6 +17,7 @@
#include "RuntimeEnvironment.hpp" #include "RuntimeEnvironment.hpp"
#include "Multicaster.hpp" #include "Multicaster.hpp"
#include "Network.hpp" #include "Network.hpp"
#include "Membership.hpp"
#include "Topology.hpp" #include "Topology.hpp"
#include "Switch.hpp" #include "Switch.hpp"
@ -28,7 +29,7 @@ Multicaster::Multicaster(const RuntimeEnvironment *renv) :
Multicaster::~Multicaster() {} Multicaster::~Multicaster() {}
void Multicaster::send( unsigned int Multicaster::send(
void *tPtr, void *tPtr,
int64_t now, int64_t now,
const SharedPtr<Network> &network, const SharedPtr<Network> &network,
@ -40,104 +41,218 @@ void Multicaster::send(
const void *const data, const void *const data,
unsigned int len) unsigned int len)
{ {
static const unsigned int PRIMES[16] = { 2,3,5,7,11,13,17,19,23,29,31,37,41,43,47,53 }; static const unsigned int PRIMES[16] = { 3,5,7,11,13,17,19,23,29,31,37,41,43,47,53,59 }; // 2 is skipped as it's even
if (unlikely(len > ZT_MAX_MTU)) return; // sanity check std::vector< std::pair<int64_t,Address> > recipients;
const NetworkConfig &config = network->config(); const NetworkConfig &config = network->config();
if (config.multicastLimit == 0) return; // multicast disabled if (config.multicastLimit == 0) return; // multicast disabled
Address bridges[ZT_MAX_NETWORK_SPECIALISTS],multicastReplicators[ZT_MAX_NETWORK_SPECIALISTS];
unsigned int bridgeCount = 0,multicastReplicatorCount = 0; Address specialists[ZT_MAX_NETWORK_SPECIALISTS],multicastReplicators[ZT_MAX_NETWORK_SPECIALISTS];
unsigned int specialistCount = 0,multicastReplicatorCount = 0,bridgeCount = 0;
bool amMulticastReplicator = false;
for(unsigned int i=0;i<config.specialistCount;++i) { for(unsigned int i=0;i<config.specialistCount;++i) {
if ((config.specialists[i] & ZT_NETWORKCONFIG_SPECIALIST_TYPE_ACTIVE_BRIDGE) != 0) if (RR->identity.address() == config.specialists[i]) {
bridges[bridgeCount++] = config.specialists[i]; amMulticastReplicator |= ((config.specialists[i] & ZT_NETWORKCONFIG_SPECIALIST_TYPE_MULTICAST_REPLICATOR) != 0);
if ((config.specialists[i] & ZT_NETWORKCONFIG_SPECIALIST_TYPE_MULTICAST_REPLICATOR) != 0) } else {
multicastReplicators[multicastReplicatorCount++] = config.specialists[i]; specialists[specialistCount++] = config.specialists[i];
if ((config.specialists[i] & ZT_NETWORKCONFIG_SPECIALIST_TYPE_ACTIVE_BRIDGE) != 0) {
recipients.push_back(std::pair<int64_t,Address>(0,config.specialists[i]));
++bridgeCount;
} if ((config.specialists[i] & ZT_NETWORKCONFIG_SPECIALIST_TYPE_MULTICAST_REPLICATOR) != 0) {
multicastReplicators[multicastReplicatorCount++] = config.specialists[i];
}
}
} }
std::sort(&(specialists[0]),&(specialists[specialistCount])); // for binary search
std::vector< std::pair<int64_t,Address> > recipients; int64_t lastGather = 0;
bool needMoar = false; _K groupKey(network->id(),mg);
for(unsigned int i=0;i<bridgeCount;++i)
recipients.push_back(std::pair<int64_t,Address>(9223372036854775807LL,bridges[i]));
{ {
Mutex::Lock l2(_groups_l); Mutex::Lock l(_groups_l);
_getMembersByTime(network->id(),mg,recipients); const _G *const g = _groups.get(groupKey);
} if (g) {
std::sort(recipients.begin() + bridgeCount,recipients.end(),std::greater< std::pair<int64_t,Address> >()); lastGather = g->lastGather;
recipients.erase(std::unique(recipients.begin(),recipients.end()),recipients.end()); recipients.reserve(recipients.size() + g->members.size());
if (recipients.size() > config.multicastLimit) { Hashtable< Address,int64_t >::Iterator mi(const_cast<_G *>(g)->members);
recipients.resize(config.multicastLimit); Address *mik = nullptr;
} else if (recipients.size() < config.multicastLimit) { int64_t *miv = nullptr;
needMoar = true; while (mi.next(mik,miv)) {
} if (!std::binary_search(&(specialists[0]),&(specialists[specialistCount]),*mik))
recipients.push_back(std::pair<int64_t,Address>(*miv,*mik));
_txQueue_l.lock();
_OM *om = &(_txQueue[_txQueuePtr++ % ZT_TX_QUEUE_SIZE]);
Mutex::Lock ql(om->lock);
_txQueue_l.unlock();
om->nwid = network->id();
om->src = src;
om->mg = mg;
om->etherType = etherType;
om->dataSize = len;
memcpy(om->data,data,len);
if (existingBloom) {
om->bloomFilterMultiplier = existingBloomMultiplier;
memcpy(om->bloomFilter,existingBloom,sizeof(om->bloomFilter));
} else {
om->bloomFilterMultiplier = 1;
memset(om->bloomFilter,0,sizeof(om->bloomFilter));
if (recipients.size() > 1) {
unsigned int mult = 1;
unsigned int bestMultColl = 0xffffffff;
for(int k=0;k<16;++k) { // 16 == arbitrary limit on iterations for this search, also must be <= size of PRIMES
unsigned int coll = 0;
for(std::vector< std::pair<int64_t,Address> >::const_iterator r(recipients.begin());r!=recipients.end();++r) {
const unsigned int bfi = mult * (unsigned int)r->second.toInt();
const unsigned int byte = (bfi >> 3) % sizeof(om->bloomFilter);
const uint8_t bit = 1 << (bfi & 7);
coll += ((om->bloomFilter[byte] & bit) != 0);
om->bloomFilter[byte] |= bit;
}
memset(om->bloomFilter,0,sizeof(om->bloomFilter));
if (coll <= bestMultColl) {
om->bloomFilterMultiplier = mult;
if (coll == 0) // perfect score, no need to continue searching
break;
bestMultColl = coll;
}
mult = PRIMES[k];
} }
} }
} }
if (multicastReplicatorCount > 0) { // Sort recipients, maintaining bridges first in list
// SEND std::sort(recipients.begin() + bridgeCount,recipients.end(),std::greater< std::pair<int64_t,Address> >());
return;
// Gather new recipients periodically, being more aggressive if we have none.
if ((now - lastGather) > (recipients.empty() ? 5000 : ZT_MULTICAST_GATHER_PERIOD)) {
{
Mutex::Lock l(_groups_l);
_groups[groupKey].lastGather = now;
}
Packet outp(network->controller(),RR->identity.address(),Packet::VERB_MULTICAST_GATHER);
outp.append(network->id());
outp.append((uint8_t)0);
mg.mac().appendTo(outp);
outp.append(mg.adi());
outp.append((uint32_t)0xffffffff);
RR->sw->send(tPtr,outp,true);
for(unsigned int i=0;i<specialistCount;++i) {
outp.newInitializationVector();
outp.setDestination(specialists[i]);
RR->sw->send(tPtr,outp,true);
}
// LEGACY: roots may know about older versions' multicast subscriptions but
// the root's role here is being phased out.
SharedPtr<Peer> root(RR->topology->root(now));
if (root) {
outp.newInitializationVector();
outp.setDestination(root->address());
outp.armor(root->key(),true);
root->sendDirect(tPtr,outp.data(),outp.size(),now,true);
}
} }
SharedPtr<Peer> nextHops[2]; // these by definition are protocol version >= 11 if (recipients.empty())
unsigned int nextHopsBestLatency[2] = { 0xffff,0xffff }; return 0;
for(std::vector< std::pair<int64_t,Address> >::const_iterator r(recipients.begin());r!=recipients.end();++r) {
const unsigned int bfi = om->bloomFilterMultiplier * (unsigned int)r->second.toInt();
const unsigned int bfbyte = (bfi >> 3) % sizeof(om->bloomFilter);
const uint8_t bfbit = 1 << (bfi & 7);
if ((om->bloomFilter[bfbyte] & bfbit) != 0) {
continue;
} else {
SharedPtr<Peer> peer(RR->topology->get(r->second));
if (peer) {
if (peer->remoteVersionProtocol() < 11) {
// SEND
om->bloomFilter[bfbyte] |= bfbit; unsigned int sentCount = 0;
continue;
} else { uint64_t bloomFilter[ZT_MULTICAST_BLOOM_FILTER_SIZE_BITS / 64];
unsigned int bloomMultiplier;
if (existingBloom) {
memcpy(bloomFilter,existingBloom,sizeof(bloomFilter));
bloomMultiplier = existingBloomMultiplier;
} else {
memset(bloomFilter,0,sizeof(bloomFilter));
bloomMultiplier = 1;
// Iteratively search for a bloom multiplier that results in no collisions
// among known recipients. Usually the first iteration is good unless
// the recipient set is quite large.
if (recipients.size() > 1) {
unsigned long bestMultColl = 0xffffffff;
for(int k=0;k<16;++k) { // 16 == arbitrary limit on iterations for this search, also must be <= size of PRIMES
const unsigned int mult = PRIMES[k];
unsigned long coll = 0;
for(std::vector< std::pair<int64_t,Address> >::const_iterator r(recipients.begin());r!=recipients.end();++r) {
const unsigned int bfi = mult * (unsigned int)r->second.toInt();
const unsigned int byte = (bfi >> 3) % sizeof(bloomFilter);
const uint8_t bit = 1 << (bfi & 7);
coll += ((((uint8_t *)bloomFilter)[byte] & bit) != 0);
((uint8_t *)bloomFilter)[byte] |= bit;
}
memset(bloomFilter,0,sizeof(bloomFilter));
if (coll <= bestMultColl) {
bloomMultiplier = mult;
if (coll == 0) // perfect score, no need to continue searching
break;
bestMultColl = coll;
}
}
}
}
// See if there is a multicast replicator, trying to pick the fastest/best one.
Address bestReplicator;
if (multicastReplicatorCount > 0) {
unsigned int bestReplicatorLatency = 0xffff;
for(unsigned int i=0;i<multicastReplicatorCount;++i) {
const unsigned int bfi = bloomMultiplier * (unsigned int)multicastReplicators[i].toInt();
if ((((uint8_t *)bloomFilter)[(bfi >> 3) % sizeof(bloomFilter)] & (1 << (bfi & 7))) == 0) {
SharedPtr<Peer> peer(RR->topology->get(multicastReplicators[i]));
if (peer) {
const unsigned int lat = peer->latency(now);
if (lat <= bestReplicatorLatency) {
bestReplicator = peer->address();
bestReplicatorLatency = lat;
}
} else if (!bestReplicator) {
bestReplicator = multicastReplicators[i];
}
}
}
}
// If this is a multicast replicator, aggressively replicate. Multicast
// replicators are not subject to send count limits.
if (amMulticastReplicator) {
std::vector< std::pair< int,Address > > byLatency;
for(std::vector< std::pair<int64_t,Address> >::const_iterator r(recipients.begin());r!=recipients.end();++r) {
const unsigned int bfi = bloomMultiplier * (unsigned int)r->second.toInt();
if ((((uint8_t *)bloomFilter)[(bfi >> 3) % sizeof(bloomFilter)] & (1 << (bfi & 7))) == 0) {
SharedPtr<Peer> peer(RR->topology->get(r->second));
byLatency.push_back(std::pair< int,Address >((peer) ? (int)peer->latency(now) : 0xffff,r->second));
}
}
std::sort(byLatency.begin(),byLatency.end());
unsigned long cnt = byLatency.size();
if (bestReplicator)
cnt /= 2; // send to only the best half of the latency-sorted population if there are more replicators
for(unsigned long i=0;i<cnt;++i) {
const unsigned int bfi = bloomMultiplier * (unsigned int)byLatency[i].second.toInt();
((uint8_t *)bloomFilter)[(bfi >> 3) % sizeof(bloomFilter)] |= 1 << (bfi & 7);
Packet outp(byLatency[i].second,RR->identity.address(),Packet::VERB_MULTICAST_FRAME);
outp.append(network->id());
outp.append((uint8_t)0x04);
src.appendTo(outp);
mg.mac().appendTo(outp);
outp.append(mg.adi());
outp.append((uint16_t)etherType);
outp.append(data,len);
outp.compress();
RR->sw->send(tPtr,outp,true);
++sentCount;
}
}
// Forward to the next multicast replicator, if any.
if (bestReplicator) {
const unsigned int bfi = bloomMultiplier * (unsigned int)bestReplicator.toInt();
((uint8_t *)bloomFilter)[(bfi >> 3) % sizeof(bloomFilter)] |= 1 << (bfi & 7);
Packet outp(bestReplicator,RR->identity.address(),Packet::VERB_MULTICAST_FRAME);
outp.append((uint8_t)(0x04 | 0x08));
RR->identity.address().appendTo(outp);
outp.append((uint16_t)bloomMultiplier);
outp.append((uint16_t)sizeof(bloomFilter));
outp.append(((uint8_t *)bloomFilter),sizeof(bloomFilter));
src.appendTo(outp);
mg.mac().appendTo(outp);
outp.append(mg.adi());
outp.append((uint16_t)etherType);
outp.append(data,len);
outp.compress();
RR->sw->send(tPtr,outp,true);
++sentCount;
}
// If this is a multicast replicator, we've already replicated.
if (amMulticastReplicator)
return (unsigned int)recipients.size();
// Find the two best next hops (that have never seen this multicast)
// that are newer version nodes.
SharedPtr<Peer> nextHops[2];
unsigned int nextHopsBestLatency[2] = { 0xffff,0xffff };
for(std::vector< std::pair<int64_t,Address> >::iterator r(recipients.begin());r!=recipients.end();++r) {
if (r->first >= 0) {
const unsigned int bfi = bloomMultiplier * (unsigned int)r->second.toInt();
if ((((uint8_t *)bloomFilter)[(bfi >> 3) % sizeof(bloomFilter)] & (1 << (bfi & 7))) == 0) {
const SharedPtr<Peer> peer(RR->topology->get(r->second));
if ((peer)&&(peer->remoteVersionProtocol() >= 11)) {
r->first = -1; // use this field now to flag as non-legacy
const unsigned int lat = peer->latency(now); const unsigned int lat = peer->latency(now);
for(unsigned int nh=0;nh<2;++nh) { for(unsigned int nh=0;nh<2;++nh) {
if (lat <= nextHopsBestLatency[nh]) { if (lat <= nextHopsBestLatency[nh]) {
@ -151,17 +266,57 @@ void Multicaster::send(
} }
} }
// Set bits for next hops in bloom filter
for(unsigned int nh=0;nh<2;++nh) { for(unsigned int nh=0;nh<2;++nh) {
if (nextHops[nh]) { if (nextHops[nh]) {
const unsigned int bfi = om->bloomFilterMultiplier * (unsigned int)nextHops[nh]->address().toInt(); const unsigned int bfi = bloomMultiplier * (unsigned int)nextHops[nh]->address().toInt();
om->bloomFilter[(bfi >> 3) % sizeof(om->bloomFilter)] |= 1 << (bfi & 7); ((uint8_t *)bloomFilter)[(bfi >> 3) % sizeof(bloomFilter)] |= 1 << (bfi & 7);
++sentCount;
} }
} }
for(unsigned int nh=0;nh<2;++nh) { // Send to legacy peers and flag these in bloom filter
if (nextHops[nh]) { const unsigned int limit = config.multicastLimit + bridgeCount;
for(std::vector< std::pair<int64_t,Address> >::const_iterator r(recipients.begin());(r!=recipients.end())&&(sentCount<limit);++r) {
if (r->first >= 0) {
const unsigned int bfi = bloomMultiplier * (unsigned int)r->second.toInt();
((uint8_t *)bloomFilter)[(bfi >> 3) % sizeof(bloomFilter)] |= 1 << (bfi & 7);
Packet outp(r->second,RR->identity.address(),Packet::VERB_MULTICAST_FRAME);
outp.append(network->id());
outp.append((uint8_t)0x04);
src.appendTo(outp);
mg.mac().appendTo(outp);
outp.append(mg.adi());
outp.append((uint16_t)etherType);
outp.append(data,len);
outp.compress();
RR->sw->send(tPtr,outp,true);
++sentCount;
} }
} }
// Send to next hops for P2P propagation
for(unsigned int nh=0;nh<2;++nh) {
if (nextHops[nh]) {
Packet outp(nextHops[nh]->address(),RR->identity.address(),Packet::VERB_MULTICAST_FRAME);
outp.append((uint8_t)(0x04 | 0x08));
RR->identity.address().appendTo(outp);
outp.append((uint16_t)bloomMultiplier);
outp.append((uint16_t)sizeof(bloomFilter));
outp.append(((uint8_t *)bloomFilter),sizeof(bloomFilter));
src.appendTo(outp);
mg.mac().appendTo(outp);
outp.append(mg.adi());
outp.append((uint16_t)etherType);
outp.append(data,len);
outp.compress();
RR->sw->send(tPtr,outp,true);
}
}
return (unsigned int)recipients.size();
} }
void Multicaster::clean(int64_t now) void Multicaster::clean(int64_t now)

View file

@ -30,8 +30,11 @@
#include "SharedPtr.hpp" #include "SharedPtr.hpp"
#include "Packet.hpp" #include "Packet.hpp"
// Size in bits -- this is pretty close to the maximum allowed by the protocol // Size in bits -- do not change as this is about as large as we can support
#define ZT_MULTICAST_BLOOM_FILTER_SIZE_BITS 16384 // This leaves room for up to 10000 MTU data (max supported MTU) and header
// information in a maximum supported size packet. Note that data compression
// will practically reduce this size in transit for sparse or saturated fields.
#define ZT_MULTICAST_BLOOM_FILTER_SIZE_BITS 50048
namespace ZeroTier { namespace ZeroTier {
@ -45,6 +48,44 @@ class Network;
*/ */
class Multicaster class Multicaster
{ {
private:
// Composite key of network ID and multicast group
struct _K
{
uint64_t nwid;
MulticastGroup mg;
ZT_ALWAYS_INLINE _K() : nwid(0),mg() {}
ZT_ALWAYS_INLINE _K(const uint64_t n,const MulticastGroup &g) : nwid(n),mg(g) {}
ZT_ALWAYS_INLINE bool operator==(const _K &k) const { return ((nwid == k.nwid)&&(mg == k.mg)); }
ZT_ALWAYS_INLINE bool operator!=(const _K &k) const { return ((nwid != k.nwid)||(mg != k.mg)); }
ZT_ALWAYS_INLINE unsigned long hashCode() const { return (mg.hashCode() ^ (unsigned long)(nwid ^ (nwid >> 32))); }
};
// Multicast group info
struct _G
{
ZT_ALWAYS_INLINE _G() : lastGather(0),members(16) {}
int64_t lastGather;
Hashtable< Address,int64_t > members;
};
// Outbound multicast
struct _OM
{
uint64_t nwid;
MAC src;
MulticastGroup mg;
unsigned int etherType;
unsigned int dataSize;
unsigned int count;
unsigned int limit;
unsigned int bloomFilterMultiplier;
uint64_t bloomFilter[ZT_MULTICAST_BLOOM_FILTER_SIZE_BITS / 64];
uint8_t data[ZT_MAX_MTU];
Mutex lock;
};
public: public:
Multicaster(const RuntimeEnvironment *renv); Multicaster(const RuntimeEnvironment *renv);
~Multicaster(); ~Multicaster();
@ -60,7 +101,7 @@ public:
ZT_ALWAYS_INLINE void add(const int64_t now,const uint64_t nwid,const MulticastGroup &mg,const Address &member) ZT_ALWAYS_INLINE void add(const int64_t now,const uint64_t nwid,const MulticastGroup &mg,const Address &member)
{ {
Mutex::Lock l(_groups_l); Mutex::Lock l(_groups_l);
_groups[_K(nwid,mg)].set(member,now); _groups[_K(nwid,mg)].members.set(member,now);
} }
/** /**
@ -80,9 +121,9 @@ public:
{ {
Mutex::Lock l(_groups_l); Mutex::Lock l(_groups_l);
const uint8_t *a = (const uint8_t *)addresses; const uint8_t *a = (const uint8_t *)addresses;
Hashtable< Address,int64_t > &members = _groups[_K(nwid,mg)]; _G &g = _groups[_K(nwid,mg)];
while (count--) { while (count--) {
members.set(Address(a,ZT_ADDRESS_LENGTH),now); g.members.set(Address(a,ZT_ADDRESS_LENGTH),now);
a += ZT_ADDRESS_LENGTH; a += ZT_ADDRESS_LENGTH;
} }
} }
@ -98,10 +139,10 @@ public:
{ {
Mutex::Lock l(_groups_l); Mutex::Lock l(_groups_l);
const _K gk(nwid,mg); const _K gk(nwid,mg);
Hashtable< Address,int64_t > *const members = _groups.get(gk); _G *const g = _groups.get(gk);
if (members) { if (g) {
members->erase(member); g->members.erase(member);
if (members->empty()) if (g->members.empty())
_groups.erase(gk); _groups.erase(gk);
} }
} }
@ -121,8 +162,21 @@ public:
ZT_ALWAYS_INLINE unsigned long eachMember(const uint64_t nwid,const MulticastGroup &mg,F func) const ZT_ALWAYS_INLINE unsigned long eachMember(const uint64_t nwid,const MulticastGroup &mg,F func) const
{ {
std::vector< std::pair<int64_t,Address> > sortedByTime; std::vector< std::pair<int64_t,Address> > sortedByTime;
Mutex::Lock l(_groups_l); {
_getMembersByTime(nwid,mg,sortedByTime); Mutex::Lock l(_groups_l);
const _K gk(nwid,mg);
const _G *const g = _groups.get(gk);
if (g) {
sortedByTime.reserve(g->members.size());
{
Hashtable< Address,int64_t >::Iterator mi(const_cast<_G *>(g)->members);
Address *mik = nullptr;
int64_t *miv = nullptr;
while (mi.next(mik,miv))
sortedByTime.push_back(std::pair<int64_t,Address>(*miv,*mik));
}
}
}
std::sort(sortedByTime.begin(),sortedByTime.end()); std::sort(sortedByTime.begin(),sortedByTime.end());
for(std::vector< std::pair<int64_t,Address> >::const_reverse_iterator i(sortedByTime.begin());i!=sortedByTime.end();++i) { for(std::vector< std::pair<int64_t,Address> >::const_reverse_iterator i(sortedByTime.begin());i!=sortedByTime.end();++i) {
if (!func(i->second)) if (!func(i->second))
@ -144,8 +198,9 @@ public:
* @param existingBloom Existing bloom filter or NULL if none * @param existingBloom Existing bloom filter or NULL if none
* @param data Packet data * @param data Packet data
* @param len Length of packet data * @param len Length of packet data
* @return Number of known recipients for multicast (including bridges and replicators)
*/ */
void send( unsigned int send(
void *tPtr, void *tPtr,
int64_t now, int64_t now,
const SharedPtr<Network> &network, const SharedPtr<Network> &network,
@ -166,74 +221,13 @@ public:
void clean(int64_t now); void clean(int64_t now);
private: private:
ZT_ALWAYS_INLINE void _getMembersByTime(const uint64_t nwid,const MulticastGroup &mg,std::vector< std::pair<int64_t,Address> > &byTime)
{
// assumes _groups_l is locked
const _K gk(nwid,mg);
const Hashtable< Address,int64_t > *const members = _groups.get(gk);
if (members) {
byTime.reserve(members->size());
{
Hashtable< Address,int64_t >::Iterator mi(*const_cast<Hashtable< Address,int64_t > *>(members));
Address *mik = nullptr;
int64_t *miv = nullptr;
while (mi.next(mik,miv))
byTime.push_back(std::pair<int64_t,Address>(*miv,*mik));
}
}
}
struct _K
{
uint64_t nwid;
MulticastGroup mg;
ZT_ALWAYS_INLINE _K() : nwid(0),mg() {}
ZT_ALWAYS_INLINE _K(const uint64_t n,const MulticastGroup &g) : nwid(n),mg(g) {}
ZT_ALWAYS_INLINE bool operator==(const _K &k) const { return ((nwid == k.nwid)&&(mg == k.mg)); }
ZT_ALWAYS_INLINE bool operator!=(const _K &k) const { return ((nwid != k.nwid)||(mg != k.mg)); }
ZT_ALWAYS_INLINE unsigned long hashCode() const { return (mg.hashCode() ^ (unsigned long)(nwid ^ (nwid >> 32))); }
};
/*
* Multicast frame:
* <[8] 64-bit network ID>
* <[1] flags>
* [<[...] network certificate of membership (DEPRECATED)>]
* [<[4] 32-bit implicit gather limit (DEPRECATED)>]
* [<[5] ZeroTier address of originating sender (including w/0x08)>]
* [<[2] 16-bit bloom filter multiplier>]
* [<[2] 16-bit length of propagation bloom filter in bytes]
* [<[...] propagation bloom filter>]
* [<[6] source MAC>]
* <[6] destination MAC (multicast address)>
* <[4] 32-bit multicast ADI (multicast address extension)>
* <[2] 16-bit ethertype>
* <[...] ethernet payload>
* [<[2] 16-bit length of signature>]
* [<[...] signature (algorithm depends on sender identity)>]
*/
struct _OM
{
uint64_t nwid;
MAC src;
MulticastGroup mg;
unsigned int etherType;
unsigned int dataSize;
unsigned int bloomFilterMultiplier;
uint8_t bloomFilter[ZT_MULTICAST_BLOOM_FILTER_SIZE_BITS / 8];
uint8_t data[ZT_MAX_MTU];
Mutex lock;
};
const RuntimeEnvironment *const RR; const RuntimeEnvironment *const RR;
_OM _txQueue[ZT_TX_QUEUE_SIZE]; _OM _txQueue[ZT_TX_QUEUE_SIZE];
unsigned int _txQueuePtr; unsigned int _txQueuePtr;
Mutex _txQueue_l; Mutex _txQueue_l;
Hashtable< _K,Hashtable< Address,int64_t > > _groups; Hashtable< _K,_G > _groups;
Mutex _groups_l; Mutex _groups_l;
}; };

View file

@ -1065,6 +1065,43 @@ void Network::doPeriodicTasks(void *tPtr,const int64_t now)
} }
} }
void Network::learnBridgeRoute(const MAC &mac,const Address &addr)
{
Mutex::Lock _l(_remoteBridgeRoutes_l);
_remoteBridgeRoutes[mac] = addr;
// Anti-DOS circuit breaker to prevent nodes from spamming us with absurd numbers of bridge routes
while (_remoteBridgeRoutes.size() > ZT_MAX_BRIDGE_ROUTES) {
Hashtable< Address,unsigned long > counts;
Address maxAddr;
unsigned long maxCount = 0;
MAC *m = (MAC *)0;
Address *a = (Address *)0;
// Find the address responsible for the most entries
{
Hashtable<MAC,Address>::Iterator i(_remoteBridgeRoutes);
while (i.next(m,a)) {
const unsigned long c = ++counts[*a];
if (c > maxCount) {
maxCount = c;
maxAddr = *a;
}
}
}
// Kill this address from our table, since it's most likely spamming us
{
Hashtable<MAC,Address>::Iterator i(_remoteBridgeRoutes);
while (i.next(m,a)) {
if (*a == maxAddr)
_remoteBridgeRoutes.erase(*m);
}
}
}
}
Membership::AddCredentialResult Network::addCredential(void *tPtr,const Address &sentFrom,const Revocation &rev) Membership::AddCredentialResult Network::addCredential(void *tPtr,const Address &sentFrom,const Revocation &rev)
{ {
if (rev.networkId() != _id) if (rev.networkId() != _id)
@ -1300,7 +1337,11 @@ void Network::_externalConfig(ZT_VirtualNetworkConfig *ec) const
ec->type = (_config) ? (_config.isPrivate() ? ZT_NETWORK_TYPE_PRIVATE : ZT_NETWORK_TYPE_PUBLIC) : ZT_NETWORK_TYPE_PRIVATE; ec->type = (_config) ? (_config.isPrivate() ? ZT_NETWORK_TYPE_PRIVATE : ZT_NETWORK_TYPE_PUBLIC) : ZT_NETWORK_TYPE_PRIVATE;
ec->mtu = (_config) ? _config.mtu : ZT_DEFAULT_MTU; ec->mtu = (_config) ? _config.mtu : ZT_DEFAULT_MTU;
ec->dhcp = 0; ec->dhcp = 0;
std::vector<Address> ab(_config.activeBridges()); std::vector<Address> ab;
for(unsigned int i=0;i<_config.specialistCount;++i) {
if ((_config.specialists[i] & ZT_NETWORKCONFIG_SPECIALIST_TYPE_ACTIVE_BRIDGE) != 0)
ab.push_back(Address(_config.specialists[i]));
}
ec->bridge = (std::find(ab.begin(),ab.end(),RR->identity.address()) != ab.end()) ? 1 : 0; ec->bridge = (std::find(ab.begin(),ab.end(),RR->identity.address()) != ab.end()) ? 1 : 0;
ec->broadcastEnabled = (_config) ? (_config.enableBroadcast() ? 1 : 0) : 0; ec->broadcastEnabled = (_config) ? (_config.enableBroadcast() ? 1 : 0) : 0;
ec->portError = _portError; ec->portError = _portError;

View file

@ -80,14 +80,14 @@ public:
~Network(); ~Network();
inline uint64_t id() const { return _id; } ZT_ALWAYS_INLINE uint64_t id() const { return _id; }
inline Address controller() const { return Address(_id >> 24); } ZT_ALWAYS_INLINE Address controller() const { return Address(_id >> 24); }
inline bool multicastEnabled() const { return (_config.multicastLimit > 0); } ZT_ALWAYS_INLINE bool multicastEnabled() const { return (_config.multicastLimit > 0); }
inline bool hasConfig() const { return (_config); } ZT_ALWAYS_INLINE bool hasConfig() const { return (_config); }
inline uint64_t lastConfigUpdate() const { return _lastConfigUpdate; } ZT_ALWAYS_INLINE uint64_t lastConfigUpdate() const { return _lastConfigUpdate; }
inline ZT_VirtualNetworkStatus status() const { return _status(); } ZT_ALWAYS_INLINE ZT_VirtualNetworkStatus status() const { return _status(); }
inline const NetworkConfig &config() const { return _config; } ZT_ALWAYS_INLINE const NetworkConfig &config() const { return _config; }
inline const MAC &mac() const { return _mac; } ZT_ALWAYS_INLINE const MAC &mac() const { return _mac; }
/** /**
* Apply filters to an outgoing packet * Apply filters to an outgoing packet
@ -159,7 +159,7 @@ public:
* @param includeBridgedGroups If true, also check groups we've learned via bridging * @param includeBridgedGroups If true, also check groups we've learned via bridging
* @return True if this network endpoint / peer is a member * @return True if this network endpoint / peer is a member
*/ */
inline bool subscribedToMulticastGroup(const MulticastGroup &mg,const bool includeBridgedGroups) const ZT_ALWAYS_INLINE bool subscribedToMulticastGroup(const MulticastGroup &mg,const bool includeBridgedGroups) const
{ {
Mutex::Lock l(_myMulticastGroups_l); Mutex::Lock l(_myMulticastGroups_l);
if (std::binary_search(_myMulticastGroups.begin(),_myMulticastGroups.end(),mg)) if (std::binary_search(_myMulticastGroups.begin(),_myMulticastGroups.end(),mg))
@ -175,7 +175,7 @@ public:
* @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call * @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
* @param mg New multicast group * @param mg New multicast group
*/ */
inline void multicastSubscribe(void *tPtr,const MulticastGroup &mg) ZT_ALWAYS_INLINE void multicastSubscribe(void *tPtr,const MulticastGroup &mg)
{ {
Mutex::Lock l(_myMulticastGroups_l); Mutex::Lock l(_myMulticastGroups_l);
if (!std::binary_search(_myMulticastGroups.begin(),_myMulticastGroups.end(),mg)) { if (!std::binary_search(_myMulticastGroups.begin(),_myMulticastGroups.end(),mg)) {
@ -190,7 +190,7 @@ public:
* *
* @param mg Multicast group * @param mg Multicast group
*/ */
inline void multicastUnsubscribe(const MulticastGroup &mg) ZT_ALWAYS_INLINE void multicastUnsubscribe(const MulticastGroup &mg)
{ {
Mutex::Lock l(_myMulticastGroups_l); Mutex::Lock l(_myMulticastGroups_l);
std::vector<MulticastGroup>::iterator i(std::lower_bound(_myMulticastGroups.begin(),_myMulticastGroups.end(),mg)); std::vector<MulticastGroup>::iterator i(std::lower_bound(_myMulticastGroups.begin(),_myMulticastGroups.end(),mg));
@ -231,12 +231,12 @@ public:
/** /**
* Set netconf failure to 'access denied' -- called in IncomingPacket when controller reports this * Set netconf failure to 'access denied' -- called in IncomingPacket when controller reports this
*/ */
inline void setAccessDenied() { _netconfFailure = NETCONF_FAILURE_ACCESS_DENIED; } ZT_ALWAYS_INLINE void setAccessDenied() { _netconfFailure = NETCONF_FAILURE_ACCESS_DENIED; }
/** /**
* Set netconf failure to 'not found' -- called by IncomingPacket when controller reports this * Set netconf failure to 'not found' -- called by IncomingPacket when controller reports this
*/ */
inline void setNotFound() { _netconfFailure = NETCONF_FAILURE_NOT_FOUND; } ZT_ALWAYS_INLINE void setNotFound() { _netconfFailure = NETCONF_FAILURE_NOT_FOUND; }
/** /**
* Determine whether this peer is permitted to communicate on this network * Determine whether this peer is permitted to communicate on this network
@ -257,7 +257,7 @@ public:
* @param mac MAC address * @param mac MAC address
* @return ZeroTier address of bridge to this MAC * @return ZeroTier address of bridge to this MAC
*/ */
inline Address findBridgeTo(const MAC &mac) const ZT_ALWAYS_INLINE Address findBridgeTo(const MAC &mac) const
{ {
Mutex::Lock _l(_remoteBridgeRoutes_l); Mutex::Lock _l(_remoteBridgeRoutes_l);
const Address *const br = _remoteBridgeRoutes.get(mac); const Address *const br = _remoteBridgeRoutes.get(mac);
@ -275,42 +275,7 @@ public:
* @param mac MAC address of destination * @param mac MAC address of destination
* @param addr Bridge this MAC is reachable behind * @param addr Bridge this MAC is reachable behind
*/ */
inline void learnBridgeRoute(const MAC &mac,const Address &addr) void learnBridgeRoute(const MAC &mac,const Address &addr);
{
Mutex::Lock _l(_remoteBridgeRoutes_l);
_remoteBridgeRoutes[mac] = addr;
// Anti-DOS circuit breaker to prevent nodes from spamming us with absurd numbers of bridge routes
while (_remoteBridgeRoutes.size() > ZT_MAX_BRIDGE_ROUTES) {
Hashtable< Address,unsigned long > counts;
Address maxAddr;
unsigned long maxCount = 0;
MAC *m = (MAC *)0;
Address *a = (Address *)0;
// Find the address responsible for the most entries
{
Hashtable<MAC,Address>::Iterator i(_remoteBridgeRoutes);
while (i.next(m,a)) {
const unsigned long c = ++counts[*a];
if (c > maxCount) {
maxCount = c;
maxAddr = *a;
}
}
}
// Kill this address from our table, since it's most likely spamming us
{
Hashtable<MAC,Address>::Iterator i(_remoteBridgeRoutes);
while (i.next(m,a)) {
if (*a == maxAddr)
_remoteBridgeRoutes.erase(*m);
}
}
}
}
/** /**
* Learn a multicast group that is bridged to our tap device * Learn a multicast group that is bridged to our tap device
@ -319,7 +284,7 @@ public:
* @param mg Multicast group * @param mg Multicast group
* @param now Current time * @param now Current time
*/ */
inline void learnBridgedMulticastGroup(void *tPtr,const MulticastGroup &mg,int64_t now) ZT_ALWAYS_INLINE void learnBridgedMulticastGroup(void *tPtr,const MulticastGroup &mg,int64_t now)
{ {
Mutex::Lock l(_myMulticastGroups_l); Mutex::Lock l(_myMulticastGroups_l);
_multicastGroupsBehindMe.set(mg,now); _multicastGroupsBehindMe.set(mg,now);
@ -328,7 +293,7 @@ public:
/** /**
* Validate a credential and learn it if it passes certificate and other checks * Validate a credential and learn it if it passes certificate and other checks
*/ */
Membership::AddCredentialResult addCredential(void *tPtr,const CertificateOfMembership &com) ZT_ALWAYS_INLINE Membership::AddCredentialResult addCredential(void *tPtr,const CertificateOfMembership &com)
{ {
if (com.networkId() != _id) if (com.networkId() != _id)
return Membership::ADD_REJECTED; return Membership::ADD_REJECTED;
@ -339,7 +304,7 @@ public:
/** /**
* Validate a credential and learn it if it passes certificate and other checks * Validate a credential and learn it if it passes certificate and other checks
*/ */
inline Membership::AddCredentialResult addCredential(void *tPtr,const Capability &cap) ZT_ALWAYS_INLINE Membership::AddCredentialResult addCredential(void *tPtr,const Capability &cap)
{ {
if (cap.networkId() != _id) if (cap.networkId() != _id)
return Membership::ADD_REJECTED; return Membership::ADD_REJECTED;
@ -350,7 +315,7 @@ public:
/** /**
* Validate a credential and learn it if it passes certificate and other checks * Validate a credential and learn it if it passes certificate and other checks
*/ */
inline Membership::AddCredentialResult addCredential(void *tPtr,const Tag &tag) ZT_ALWAYS_INLINE Membership::AddCredentialResult addCredential(void *tPtr,const Tag &tag)
{ {
if (tag.networkId() != _id) if (tag.networkId() != _id)
return Membership::ADD_REJECTED; return Membership::ADD_REJECTED;
@ -366,7 +331,7 @@ public:
/** /**
* Validate a credential and learn it if it passes certificate and other checks * Validate a credential and learn it if it passes certificate and other checks
*/ */
inline Membership::AddCredentialResult addCredential(void *tPtr,const CertificateOfOwnership &coo) ZT_ALWAYS_INLINE Membership::AddCredentialResult addCredential(void *tPtr,const CertificateOfOwnership &coo)
{ {
if (coo.networkId() != _id) if (coo.networkId() != _id)
return Membership::ADD_REJECTED; return Membership::ADD_REJECTED;
@ -381,7 +346,7 @@ public:
* @param to Destination peer address * @param to Destination peer address
* @param now Current time * @param now Current time
*/ */
inline void pushCredentialsNow(void *tPtr,const Address &to,const int64_t now) ZT_ALWAYS_INLINE void pushCredentialsNow(void *tPtr,const Address &to,const int64_t now)
{ {
Mutex::Lock _l(_memberships_l); Mutex::Lock _l(_memberships_l);
_memberships[to].pushCredentials(RR,tPtr,now,to,_config); _memberships[to].pushCredentials(RR,tPtr,now,to,_config);
@ -394,7 +359,7 @@ public:
* @param to Destination peer address * @param to Destination peer address
* @param now Current time * @param now Current time
*/ */
inline void pushCredentialsIfNeeded(void *tPtr,const Address &to,const int64_t now) ZT_ALWAYS_INLINE void pushCredentialsIfNeeded(void *tPtr,const Address &to,const int64_t now)
{ {
const int64_t tout = std::min(_config.credentialTimeMaxDelta,(int64_t)ZT_PEER_ACTIVITY_TIMEOUT); const int64_t tout = std::min(_config.credentialTimeMaxDelta,(int64_t)ZT_PEER_ACTIVITY_TIMEOUT);
Mutex::Lock _l(_memberships_l); Mutex::Lock _l(_memberships_l);
@ -409,7 +374,7 @@ public:
* This sets the network to completely remove itself on delete. This also prevents the * This sets the network to completely remove itself on delete. This also prevents the
* call of the normal port shutdown event on delete. * call of the normal port shutdown event on delete.
*/ */
inline void destroy() ZT_ALWAYS_INLINE void destroy()
{ {
_memberships_l.lock(); _memberships_l.lock();
_config_l.lock(); _config_l.lock();
@ -423,16 +388,34 @@ public:
* *
* @param ec Buffer to fill with externally-visible network configuration * @param ec Buffer to fill with externally-visible network configuration
*/ */
inline void externalConfig(ZT_VirtualNetworkConfig *ec) const ZT_ALWAYS_INLINE void externalConfig(ZT_VirtualNetworkConfig *ec) const
{ {
Mutex::Lock _l(_config_l); Mutex::Lock _l(_config_l);
_externalConfig(ec); _externalConfig(ec);
} }
/**
* Iterate through memberships
*
* @param f Function of (const Address,const Membership)
*/
template<typename F>
ZT_ALWAYS_INLINE void eachMember(F f)
{
Mutex::Lock ml(_memberships_l);
Hashtable<Address,Membership>::Iterator i(_memberships);
const Address *a = nullptr;
const Membership *m = nullptr;
while (i.next(a,m)) {
if (!f(*a,*m))
break;
}
}
/** /**
* @return Externally usable pointer-to-pointer exported via the core API * @return Externally usable pointer-to-pointer exported via the core API
*/ */
inline void **userPtr() { return &_uPtr; } ZT_ALWAYS_INLINE void **userPtr() { return &_uPtr; }
private: private:
void _requestConfiguration(void *tPtr); void _requestConfiguration(void *tPtr);

View file

@ -683,7 +683,7 @@ public:
/** /**
* Request endpoints for multicast distribution: * Request endpoints for multicast distribution:
* <[8] 64-bit network ID> * <[8] 64-bit network ID>
* <[1] flags> * <[1] flags (unused, must be 0)>
* <[6] MAC address of multicast group being queried> * <[6] MAC address of multicast group being queried>
* <[4] 32-bit ADI for multicast group being queried> * <[4] 32-bit ADI for multicast group being queried>
* <[4] 32-bit requested max number of multicast peers> * <[4] 32-bit requested max number of multicast peers>

View file

@ -280,6 +280,12 @@ public:
return true; return true;
} }
#ifdef __GNUC__
static ZT_ALWAYS_INLINE unsigned int countBits(const uint8_t v) { return (unsigned int)__builtin_popcount((unsigned int)v); }
static ZT_ALWAYS_INLINE unsigned int countBits(const uint16_t v) { return (unsigned int)__builtin_popcount((unsigned int)v); }
static ZT_ALWAYS_INLINE unsigned int countBits(const uint32_t v) { return (unsigned int)__builtin_popcountl((unsigned long)v); }
static ZT_ALWAYS_INLINE unsigned int countBits(const uint64_t v) { return (unsigned int)__builtin_popcountll((unsigned long long)v); }
#else
/** /**
* Count the number of bits set in an integer * Count the number of bits set in an integer
* *
@ -287,15 +293,16 @@ public:
* @return Number of bits set in this integer (0-bits in integer) * @return Number of bits set in this integer (0-bits in integer)
*/ */
template<typename T> template<typename T>
static ZT_ALWAYS_INLINE uint64_t countBits(T v) static ZT_ALWAYS_INLINE unsigned int countBits(T v)
{ {
v = v - ((v >> 1) & (T)~(T)0/3); v = v - ((v >> 1) & (T)~(T)0/3);
v = (v & (T)~(T)0/15*3) + ((v >> 2) & (T)~(T)0/15*3); v = (v & (T)~(T)0/15*3) + ((v >> 2) & (T)~(T)0/15*3);
v = (v + (v >> 4)) & (T)~(T)0/255*15; v = (v + (v >> 4)) & (T)~(T)0/255*15;
return (T)(v * ((~((T)0))/((T)255))) >> ((sizeof(T) - 1) * 8); return (unsigned int)((v * ((~((T)0))/((T)255))) >> ((sizeof(T) - 1) * 8));
} }
#endif
// Byte swappers for big/little endian conversion // Byte swappers for big/little endian conversion
#if __BYTE_ORDER == __LITTLE_ENDIAN #if __BYTE_ORDER == __LITTLE_ENDIAN
static ZT_ALWAYS_INLINE uint8_t hton(uint8_t n) { return n; } static ZT_ALWAYS_INLINE uint8_t hton(uint8_t n) { return n; }
static ZT_ALWAYS_INLINE int8_t hton(int8_t n) { return n; } static ZT_ALWAYS_INLINE int8_t hton(int8_t n) { return n; }