Return self in GATHER requests if self is a member of multicast group, and reinstate legacy support.

node/IncomingPacket.cpp

@@ -79,7 +79,7 @@ bool IncomingPacket::tryDecode(const RuntimeEnvironment *RR)
 				case Packet::VERB_RENDEZVOUS:                     return _doRENDEZVOUS(RR,peer);
 				case Packet::VERB_FRAME:                          return _doFRAME(RR,peer);
 				case Packet::VERB_EXT_FRAME:                      return _doEXT_FRAME(RR,peer);
-				//case Packet::VERB_P5_MULTICAST_FRAME:             return _doP5_MULTICAST_FRAME(RR,peer);
+				case Packet::VERB_P5_MULTICAST_FRAME:             return _doP5_MULTICAST_FRAME(RR,peer);
 				case Packet::VERB_MULTICAST_LIKE:                 return _doMULTICAST_LIKE(RR,peer);
 				case Packet::VERB_NETWORK_MEMBERSHIP_CERTIFICATE: return _doNETWORK_MEMBERSHIP_CERTIFICATE(RR,peer);
 				case Packet::VERB_NETWORK_CONFIG_REQUEST:         return _doNETWORK_CONFIG_REQUEST(RR,peer);

node/Multicaster.cpp

@@ -37,6 +37,7 @@
 #include "Peer.hpp"
 #include "CMWC4096.hpp"
 #include "C25519.hpp"
+#include "NodeConfig.hpp"
 #include "CertificateOfMembership.hpp"
 #include "Logger.hpp"
 
@@ -54,66 +55,67 @@ Multicaster::~Multicaster()
 unsigned int Multicaster::gather(const Address &queryingPeer,uint64_t nwid,const MulticastGroup &mg,Packet &appendTo,unsigned int limit) const
 {
 	unsigned char *p;
-	unsigned int n = 0,i,rptr,skipped = 0;
-	uint64_t a,done[(ZT_PROTO_MAX_PACKET_LENGTH / 5) + 1];
+	unsigned int added = 0,i,k,rptr,totalKnown = 0;
+	uint64_t a,picked[(ZT_PROTO_MAX_PACKET_LENGTH / 5) + 1];
 
-	Mutex::Lock _l(_groups_m);
-
-	std::map< std::pair<uint64_t,MulticastGroup>,MulticastGroupStatus >::const_iterator gs(_groups.find(std::pair<uint64_t,MulticastGroup>(nwid,mg)));
-	if ((gs == _groups.end())||(gs->second.members.empty())) {
-		appendTo.append((uint32_t)0);
-		appendTo.append((uint16_t)0);
-		//TRACE("..MC Multicaster::gather() attached 0 of 0 peers for %.16llx/%s (1)",nwid,mg.toString().c_str());
+	if (!limit)
 		return 0;
+	if (limit > 0xffff) // TODO: multiple return packets not yet supported
+		limit = 0xffff;
+
+	{ // Return myself if I am a member of this group
+		SharedPtr<Network> network(RR->nc->network(nwid));
+		if ((network)&&(network->subscribedToMulticastGroup(mg))) {
+			RR->identity.address().appendTo(appendTo);
+			++totalKnown;
+			++added;
+		}
 	}
 
-	if (limit > gs->second.members.size())
-		limit = (unsigned int)gs->second.members.size();
-	if (limit > ((ZT_PROTO_MAX_PACKET_LENGTH / ZT_ADDRESS_LENGTH) + 1))
-		limit = (ZT_PROTO_MAX_PACKET_LENGTH / ZT_ADDRESS_LENGTH) + 1;
+	Mutex::Lock _l(_groups_m);
 
-	unsigned int totalAt = appendTo.size();
+	const unsigned int totalAt = appendTo.size();
 	appendTo.addSize(4); // sizeof(uint32_t)
-	unsigned int nAt = appendTo.size();
+	const unsigned int addedAt = appendTo.size();
 	appendTo.addSize(2); // sizeof(uint16_t)
 
-	// Members are returned in random order so that repeated gather queries
-	// will return different subsets of a large multicast group.
-	while ((n < limit)&&((appendTo.size() + ZT_ADDRESS_LENGTH) <= ZT_PROTO_MAX_PACKET_LENGTH)) {
-		rptr = (unsigned int)RR->prng->next32();
+	std::map< std::pair<uint64_t,MulticastGroup>,MulticastGroupStatus >::const_iterator gs(_groups.find(std::pair<uint64_t,MulticastGroup>(nwid,mg)));
+	if ((gs != _groups.end())&&(!gs->second.members.empty())) {
+		totalKnown += gs->second.members.size();
+
+		// 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 < gs->second.members.size())&&((appendTo.size() + ZT_ADDRESS_LENGTH) <= ZT_PROTO_MAX_PACKET_LENGTH)) {
+			rptr = (unsigned int)RR->prng->next32();
 restart_member_scan:
-		a = gs->second.members[rptr % (unsigned int)gs->second.members.size()].address.toInt();
-		for(i=0;i<n;++i) {
-			if (done[i] == a) {
-				++rptr;
-				goto restart_member_scan;
+			a = gs->second.members[rptr % (unsigned int)gs->second.members.size()].address.toInt();
+			for(i=0;i<k;++i) {
+				if (picked[i] == a) {
+					++rptr;
+					goto restart_member_scan;
+				}
+			}
+			picked[k++] = a;
+
+			if (queryingPeer.toInt() != a) { // do not return the peer that is making the request as a result
+				p = (unsigned char *)appendTo.appendField(ZT_ADDRESS_LENGTH);
+				*(p++) = (unsigned char)((a >> 32) & 0xff);
+				*(p++) = (unsigned char)((a >> 24) & 0xff);
+				*(p++) = (unsigned char)((a >> 16) & 0xff);
+				*(p++) = (unsigned char)((a >> 8) & 0xff);
+				*p = (unsigned char)(a & 0xff);
+				++added;
 			}
-		}
-
-		// Log that we've picked this one
-		done[n++] = a;
-
-		if (queryingPeer.toInt() == a) {
-			++skipped;
-		} else {
-			// Append to packet
-			p = (unsigned char *)appendTo.appendField(ZT_ADDRESS_LENGTH);
-			*(p++) = (unsigned char)((a >> 32) & 0xff);
-			*(p++) = (unsigned char)((a >> 24) & 0xff);
-			*(p++) = (unsigned char)((a >> 16) & 0xff);
-			*(p++) = (unsigned char)((a >> 8) & 0xff);
-			*p = (unsigned char)(a & 0xff);
 		}
 	}
 
-	n -= skipped;
-
-	appendTo.setAt(totalAt,(uint32_t)(gs->second.members.size() - skipped));
-	appendTo.setAt(nAt,(uint16_t)n);
+	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 n;
+	return added;
 }
 
 std::vector<Address> Multicaster::getLegacySubscribers(uint64_t nwid,const MulticastGroup &mg) const
@@ -383,7 +385,7 @@ void Multicaster::_add(uint64_t now,uint64_t nwid,const MulticastGroup &mg,Multi
 	// this somewhere else but we'll try this for now.
 	gs.members.push_back(MulticastGroupMember(member,learnedFrom,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)"));
+	//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)"));
 
 	// Try to send to any outgoing multicasts that are waiting for more recipients
 	for(std::list<OutboundMulticast>::iterator tx(gs.txQueue.begin());tx!=gs.txQueue.end();) {

node/Network.hpp

@@ -163,6 +163,16 @@ public:
 		return _myMulticastGroups;
 	}
 
+	/**
+	 * @param mg Multicast group
+	 * @return True if this network endpoint / peer is a member
+	 */
+	bool subscribedToMulticastGroup(const MulticastGroup &mg) const
+	{
+		Mutex::Lock _l(_lock);
+		return (_myMulticastGroups.find(mg) != _myMulticastGroups.end());
+	}
+
 	/**
 	 * Apply a NetworkConfig to this network
 	 *

node/OutboundMulticast.cpp

@@ -58,6 +58,7 @@ void OutboundMulticast::init(
 	if (gatherLimit) flags |= 0x02;
 	if (src) flags |= 0x04;
 
+	/*
 	TRACE(">>MC %.16llx INIT %.16llx/%s limit %u gatherLimit %u from %s to %s length %u com==%d",
 		(unsigned long long)this,
 		nwid,
@@ -68,6 +69,7 @@ void OutboundMulticast::init(
 		dest.toString().c_str(),
 		len,
 		(com) ? 1 : 0);
+	*/
 
 	_packetNoCom.setSource(RR->identity.address());
 	_packetNoCom.setVerb(Packet::VERB_MULTICAST_FRAME);
@@ -107,12 +109,12 @@ void OutboundMulticast::sendOnly(const RuntimeEnvironment *RR,const Address &toA
 		if (network->peerNeedsOurMembershipCertificate(toAddr,Utils::now())) {
 			_packetWithCom.newInitializationVector();
 			_packetWithCom.setDestination(toAddr);
-			TRACE(">>MC %.16llx -> %s (with COM)",(unsigned long long)this,toAddr.toString().c_str());
+			//TRACE(">>MC %.16llx -> %s (with COM)",(unsigned long long)this,toAddr.toString().c_str());
 			RR->sw->send(_packetWithCom,true);
 			return;
 		}
 	}
-	TRACE(">>MC %.16llx -> %s (without COM)",(unsigned long long)this,toAddr.toString().c_str());
+	//TRACE(">>MC %.16llx -> %s (without COM)",(unsigned long long)this,toAddr.toString().c_str());
 	_packetNoCom.newInitializationVector();
 	_packetNoCom.setDestination(toAddr);
 	RR->sw->send(_packetNoCom,true);