From d3d58ba8a7999d52bbccb7c29d86c12c4f879e84 Mon Sep 17 00:00:00 2001
From: Adam Ierymenko <adam.ierymenko@gmail.com>
Date: Fri, 26 Jan 2018 21:19:51 -0500
Subject: [PATCH] Add support for designated multicast replicators to
 Multicaster::send().

---
 node/Multicaster.cpp   | 48 +++++++++++++++++++++++++++++++++++++++---
 node/Multicaster.hpp   |  2 ++
 node/NetworkConfig.hpp | 10 +++++++++
 node/Switch.cpp        |  2 +-
 4 files changed, 58 insertions(+), 4 deletions(-)

diff --git a/node/Multicaster.cpp b/node/Multicaster.cpp
index e0fa02552..c5cf456b7 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;
 				}
diff --git a/node/Multicaster.hpp b/node/Multicaster.hpp
index 91d6eace1..e57f81fea 100644
--- a/node/Multicaster.hpp
+++ b/node/Multicaster.hpp
@@ -132,6 +132,7 @@ public:
 	 * @param tPtr Thread pointer to be handed through to any callbacks called as a result of this call
 	 * @param now Current time
 	 * @param network Network
+	 * @param origin Origin of multicast (to not return to sender) or NULL if none
 	 * @param mg Multicast group
 	 * @param src Source Ethernet MAC address or NULL to skip in packet and compute from ZT address (non-bridged mode)
 	 * @param etherType Ethernet frame type
@@ -142,6 +143,7 @@ public:
 		void *tPtr,
 		int64_t now,
 		const SharedPtr<Network> &network,
+		const Address &origin,
 		const MulticastGroup &mg,
 		const MAC &src,
 		unsigned int etherType,
diff --git a/node/NetworkConfig.hpp b/node/NetworkConfig.hpp
index 156b85616..30b8d0f36 100644
--- a/node/NetworkConfig.hpp
+++ b/node/NetworkConfig.hpp
@@ -322,6 +322,16 @@ public:
 		return r;
 	}
 
+	inline unsigned int multicastReplicators(Address mr[ZT_MAX_NETWORK_SPECIALISTS]) const
+	{
+		unsigned int c = 0;
+		for(unsigned int i=0;i<specialistCount;++i) {
+			if ((specialists[i] & ZT_NETWORKCONFIG_SPECIALIST_TYPE_MULTICAST_REPLICATOR) != 0)
+				mr[c++] = specialists[i];
+		}
+		return c;
+	}
+
 	inline std::vector<Address> alwaysContactAddresses() const
 	{
 		std::vector<Address> r;
diff --git a/node/Switch.cpp b/node/Switch.cpp
index d1510e6c3..5c5130f4b 100644
--- a/node/Switch.cpp
+++ b/node/Switch.cpp
@@ -391,7 +391,7 @@ void Switch::onLocalEthernet(void *tPtr,const SharedPtr<Network> &network,const
 			tPtr,
 			RR->node->now(),
 			network,
-//			network->config().activeBridges(),
+			Address(),
 			multicastGroup,
 			(fromBridged) ? from : MAC(),
 			etherType,