Add support for designated multicast replicators to Multicaster::send().

1 files changed, 45 insertions, 3 deletions

diff --git a/node/Multicaster.cpp b/node/Multicaster.cpp
index e0fa0255..c5cf456b 100644
--- a/node/Multicaster.cpp
+++ b/node/Multicaster.cpp
@@ -161,15 +161,57 @@ void Multicaster::send(
 	void *tPtr,
 	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.
+	{
+		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)0x04); // includes source MAC
+					((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(network->id(),mg)];
@@ -214,7 +256,7 @@ void Multicaster::send(
 			unsigned int count = 0;
 
 			for(unsigned int i=0;i<activeBridgeCount;++i) {
-				if (activeBridges[i] != RR->identity.address()) {
+				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;
@@ -224,7 +266,7 @@ void Multicaster::send(
 			unsigned long idx = 0;
 			while ((count < limit)&&(idx < gs.members.size())) {
 				const Address ma(gs.members[indexes[idx++]].address);
-				if (std::find(activeBridges,activeBridges + activeBridgeCount,ma) == (activeBridges + activeBridgeCount)) {
+				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;
 				}