summaryrefslogtreecommitdiff
path: root/node/Multicaster.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'node/Multicaster.cpp')
-rw-r--r--node/Multicaster.cpp157
1 files changed, 115 insertions, 42 deletions
diff --git a/node/Multicaster.cpp b/node/Multicaster.cpp
index 8e534b5e..753e4ee0 100644
--- a/node/Multicaster.cpp
+++ b/node/Multicaster.cpp
@@ -1,6 +1,6 @@
/*
* ZeroTier One - Network Virtualization Everywhere
- * Copyright (C) 2011-2016 ZeroTier, Inc. https://www.zerotier.com/
+ * Copyright (C) 2011-2018 ZeroTier, Inc. https://www.zerotier.com/
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
@@ -14,13 +14,20 @@
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * --
+ *
+ * You can be released from the requirements of the license by purchasing
+ * a commercial license. Buying such a license is mandatory as soon as you
+ * develop commercial closed-source software that incorporates or links
+ * directly against ZeroTier software without disclosing the source code
+ * of your own application.
*/
#include <algorithm>
#include "Constants.hpp"
#include "RuntimeEnvironment.hpp"
-#include "SharedPtr.hpp"
#include "Multicaster.hpp"
#include "Topology.hpp"
#include "Switch.hpp"
@@ -29,6 +36,7 @@
#include "C25519.hpp"
#include "CertificateOfMembership.hpp"
#include "Node.hpp"
+#include "Network.hpp"
namespace ZeroTier {
@@ -43,7 +51,7 @@ Multicaster::~Multicaster()
{
}
-void Multicaster::addMultiple(void *tPtr,uint64_t now,uint64_t nwid,const MulticastGroup &mg,const void *addresses,unsigned int count,unsigned int totalKnown)
+void Multicaster::addMultiple(void *tPtr,int64_t now,uint64_t nwid,const MulticastGroup &mg,const void *addresses,unsigned int count,unsigned int totalKnown)
{
const unsigned char *p = (const unsigned char *)addresses;
const unsigned char *e = p + (5 * count);
@@ -103,7 +111,7 @@ unsigned int Multicaster::gather(const Address &queryingPeer,uint64_t nwid,const
// Members are returned in random order so that repeated gather queries
// will return different subsets of a large multicast group.
k = 0;
- while ((added < limit)&&(k < s->members.size())&&((appendTo.size() + ZT_ADDRESS_LENGTH) <= ZT_UDP_DEFAULT_PAYLOAD_MTU)) {
+ while ((added < limit)&&(k < s->members.size())&&((appendTo.size() + ZT_ADDRESS_LENGTH) <= ZT_PROTO_MAX_PACKET_LENGTH)) {
rptr = (unsigned int)RR->node->prng();
restart_member_scan:
@@ -131,8 +139,6 @@ restart_member_scan:
appendTo.setAt(totalAt,(uint32_t)totalKnown);
appendTo.setAt(addedAt,(uint16_t)added);
- //TRACE("..MC Multicaster::gather() attached %u of %u peers for %.16llx/%s (2)",n,(unsigned int)(gs->second.members.size() - skipped),nwid,mg.toString().c_str());
-
return added;
}
@@ -153,23 +159,66 @@ std::vector<Address> Multicaster::getMembers(uint64_t nwid,const MulticastGroup
void Multicaster::send(
void *tPtr,
- unsigned int limit,
- uint64_t now,
- uint64_t nwid,
- bool disableCompression,
- const std::vector<Address> &alwaysSendTo,
+ int64_t now,
+ const SharedPtr<Network> &network,
+ const Address &origin,
const MulticastGroup &mg,
const MAC &src,
unsigned int etherType,
const void *data,
unsigned int len)
{
- unsigned long idxbuf[8194];
+ unsigned long idxbuf[4096];
unsigned long *indexes = idxbuf;
+ // If we're in hub-and-spoke designated multicast replication mode, see if we
+ // have a multicast replicator active. If so, pick the best and send it
+ // there. If we are a multicast replicator or if none are alive, fall back
+ // to sender replication. Note that bridges do not do this since this would
+ // break bridge route learning. This is sort of an edge case limitation of
+ // the current protocol and could be fixed, but fixing it would add more
+ // complexity than the fix is probably worth. Bridges are generally high
+ // bandwidth nodes.
+ if (!network->config().isActiveBridge(RR->identity.address())) {
+ Address multicastReplicators[ZT_MAX_NETWORK_SPECIALISTS];
+ const unsigned int multicastReplicatorCount = network->config().multicastReplicators(multicastReplicators);
+ if (multicastReplicatorCount) {
+ if (std::find(multicastReplicators,multicastReplicators + multicastReplicatorCount,RR->identity.address()) == (multicastReplicators + multicastReplicatorCount)) {
+ SharedPtr<Peer> bestMulticastReplicator;
+ SharedPtr<Path> bestMulticastReplicatorPath;
+ unsigned int bestMulticastReplicatorLatency = 0xffff;
+ for(unsigned int i=0;i<multicastReplicatorCount;++i) {
+ const SharedPtr<Peer> p(RR->topology->getPeerNoCache(multicastReplicators[i]));
+ if ((p)&&(p->isAlive(now))) {
+ const SharedPtr<Path> pp(p->getBestPath(now,false));
+ if ((pp)&&(pp->latency() < bestMulticastReplicatorLatency)) {
+ bestMulticastReplicatorLatency = pp->latency();
+ bestMulticastReplicatorPath = pp;
+ bestMulticastReplicator = p;
+ }
+ }
+ }
+ if (bestMulticastReplicator) {
+ Packet outp(bestMulticastReplicator->address(),RR->identity.address(),Packet::VERB_MULTICAST_FRAME);
+ outp.append((uint64_t)network->id());
+ outp.append((uint8_t)0x0c); // includes source MAC | please replicate
+ ((src) ? src : MAC(RR->identity.address(),network->id())).appendTo(outp);
+ mg.mac().appendTo(outp);
+ outp.append((uint32_t)mg.adi());
+ outp.append((uint16_t)etherType);
+ outp.append(data,len);
+ if (!network->config().disableCompression()) outp.compress();
+ outp.armor(bestMulticastReplicator->key(),true);
+ bestMulticastReplicatorPath->send(RR,tPtr,outp.data(),outp.size(),now);
+ return;
+ }
+ }
+ }
+ }
+
try {
Mutex::Lock _l(_groups_m);
- MulticastGroupStatus &gs = _groups[Multicaster::Key(nwid,mg)];
+ MulticastGroupStatus &gs = _groups[Multicaster::Key(network->id(),mg)];
if (!gs.members.empty()) {
// Allocate a memory buffer if group is monstrous
@@ -187,6 +236,10 @@ void Multicaster::send(
}
}
+ Address activeBridges[ZT_MAX_NETWORK_SPECIALISTS];
+ const unsigned int activeBridgeCount = network->config().activeBridges(activeBridges);
+ const unsigned int limit = network->config().multicastLimit;
+
if (gs.members.size() >= limit) {
// Skip queue if we already have enough members to complete the send operation
OutboundMulticast out;
@@ -194,8 +247,8 @@ void Multicaster::send(
out.init(
RR,
now,
- nwid,
- disableCompression,
+ network->id(),
+ network->config().disableCompression(),
limit,
1, // we'll still gather a little from peers to keep multicast list fresh
src,
@@ -206,9 +259,9 @@ void Multicaster::send(
unsigned int count = 0;
- for(std::vector<Address>::const_iterator ast(alwaysSendTo.begin());ast!=alwaysSendTo.end();++ast) {
- if (*ast != RR->identity.address()) {
- out.sendOnly(RR,tPtr,*ast); // optimization: don't use dedup log if it's a one-pass send
+ for(unsigned int i=0;i<activeBridgeCount;++i) {
+ if ((activeBridges[i] != RR->identity.address())&&(activeBridges[i] != origin)) {
+ out.sendOnly(RR,tPtr,activeBridges[i]); // optimization: don't use dedup log if it's a one-pass send
if (++count >= limit)
break;
}
@@ -216,40 +269,59 @@ void Multicaster::send(
unsigned long idx = 0;
while ((count < limit)&&(idx < gs.members.size())) {
- Address ma(gs.members[indexes[idx++]].address);
- if (std::find(alwaysSendTo.begin(),alwaysSendTo.end(),ma) == alwaysSendTo.end()) {
+ const Address ma(gs.members[indexes[idx++]].address);
+ if ((std::find(activeBridges,activeBridges + activeBridgeCount,ma) == (activeBridges + activeBridgeCount))&&(ma != origin)) {
out.sendOnly(RR,tPtr,ma); // optimization: don't use dedup log if it's a one-pass send
++count;
}
}
} else {
- unsigned int gatherLimit = (limit - (unsigned int)gs.members.size()) + 1;
+ const unsigned int gatherLimit = (limit - (unsigned int)gs.members.size()) + 1;
if ((gs.members.empty())||((now - gs.lastExplicitGather) >= ZT_MULTICAST_EXPLICIT_GATHER_DELAY)) {
gs.lastExplicitGather = now;
Address explicitGatherPeers[16];
unsigned int numExplicitGatherPeers = 0;
+
SharedPtr<Peer> bestRoot(RR->topology->getUpstreamPeer());
if (bestRoot)
explicitGatherPeers[numExplicitGatherPeers++] = bestRoot->address();
- explicitGatherPeers[numExplicitGatherPeers++] = Network::controllerFor(nwid);
- SharedPtr<Network> network(RR->node->network(nwid));
- if (network) {
- std::vector<Address> anchors(network->config().anchors());
- for(std::vector<Address>::const_iterator a(anchors.begin());a!=anchors.end();++a) {
- if (*a != RR->identity.address()) {
- explicitGatherPeers[numExplicitGatherPeers++] = *a;
- if (numExplicitGatherPeers == 16)
- break;
- }
+
+ explicitGatherPeers[numExplicitGatherPeers++] = network->controller();
+
+ Address ac[ZT_MAX_NETWORK_SPECIALISTS];
+ const unsigned int accnt = network->config().alwaysContactAddresses(ac);
+ unsigned int shuffled[ZT_MAX_NETWORK_SPECIALISTS];
+ for(unsigned int i=0;i<accnt;++i)
+ shuffled[i] = i;
+ for(unsigned int i=0,k=accnt>>1;i<k;++i) {
+ const uint64_t x = RR->node->prng();
+ const unsigned int x1 = shuffled[(unsigned int)x % accnt];
+ const unsigned int x2 = shuffled[(unsigned int)(x >> 32) % accnt];
+ const unsigned int tmp = shuffled[x1];
+ shuffled[x1] = shuffled[x2];
+ shuffled[x2] = tmp;
+ }
+ for(unsigned int i=0;i<accnt;++i) {
+ explicitGatherPeers[numExplicitGatherPeers++] = ac[shuffled[i]];
+ if (numExplicitGatherPeers == 16)
+ break;
+ }
+
+ std::vector<Address> anchors(network->config().anchors());
+ for(std::vector<Address>::const_iterator a(anchors.begin());a!=anchors.end();++a) {
+ if (*a != RR->identity.address()) {
+ explicitGatherPeers[numExplicitGatherPeers++] = *a;
+ if (numExplicitGatherPeers == 16)
+ break;
}
}
for(unsigned int k=0;k<numExplicitGatherPeers;++k) {
const CertificateOfMembership *com = (network) ? ((network->config().com) ? &(network->config().com) : (const CertificateOfMembership *)0) : (const CertificateOfMembership *)0;
Packet outp(explicitGatherPeers[k],RR->identity.address(),Packet::VERB_MULTICAST_GATHER);
- outp.append(nwid);
+ outp.append(network->id());
outp.append((uint8_t)((com) ? 0x01 : 0x00));
mg.mac().appendTo(outp);
outp.append((uint32_t)mg.adi());
@@ -267,8 +339,8 @@ void Multicaster::send(
out.init(
RR,
now,
- nwid,
- disableCompression,
+ network->id(),
+ network->config().disableCompression(),
limit,
gatherLimit,
src,
@@ -277,11 +349,14 @@ void Multicaster::send(
data,
len);
+ if (origin)
+ out.logAsSent(origin);
+
unsigned int count = 0;
- for(std::vector<Address>::const_iterator ast(alwaysSendTo.begin());ast!=alwaysSendTo.end();++ast) {
- if (*ast != RR->identity.address()) {
- out.sendAndLog(RR,tPtr,*ast);
+ for(unsigned int i=0;i<activeBridgeCount;++i) {
+ if (activeBridges[i] != RR->identity.address()) {
+ out.sendAndLog(RR,tPtr,activeBridges[i]);
if (++count >= limit)
break;
}
@@ -290,7 +365,7 @@ void Multicaster::send(
unsigned long idx = 0;
while ((count < limit)&&(idx < gs.members.size())) {
Address ma(gs.members[indexes[idx++]].address);
- if (std::find(alwaysSendTo.begin(),alwaysSendTo.end(),ma) == alwaysSendTo.end()) {
+ if (std::find(activeBridges,activeBridges + activeBridgeCount,ma) == (activeBridges + activeBridgeCount)) {
out.sendAndLog(RR,tPtr,ma);
++count;
}
@@ -303,7 +378,7 @@ void Multicaster::send(
delete [] indexes;
}
-void Multicaster::clean(uint64_t now)
+void Multicaster::clean(int64_t now)
{
{
Mutex::Lock _l(_groups_m);
@@ -361,7 +436,7 @@ void Multicaster::addCredential(void *tPtr,const CertificateOfMembership &com,bo
}
}
-void Multicaster::_add(void *tPtr,uint64_t now,uint64_t nwid,const MulticastGroup &mg,MulticastGroupStatus &gs,const Address &member)
+void Multicaster::_add(void *tPtr,int64_t now,uint64_t nwid,const MulticastGroup &mg,MulticastGroupStatus &gs,const Address &member)
{
// assumes _groups_m is locked
@@ -378,8 +453,6 @@ void Multicaster::_add(void *tPtr,uint64_t now,uint64_t nwid,const MulticastGrou
gs.members.push_back(MulticastGroupMember(member,now));
- //TRACE("..MC %s joined multicast group %.16llx/%s via %s",member.toString().c_str(),nwid,mg.toString().c_str(),((learnedFrom) ? learnedFrom.toString().c_str() : "(direct)"));
-
for(std::list<OutboundMulticast>::iterator tx(gs.txQueue.begin());tx!=gs.txQueue.end();) {
if (tx->atLimit())
gs.txQueue.erase(tx++);
generated by cgit v1.2.3 (git 2.39.1) at 2025-07-03 08:18:13 +0000