Yet more WIP on mulitcast algo...

node/Constants.hpp

@@ -191,7 +191,7 @@
 #define ZT_PEER_SECRET_KEY_LENGTH 32
 
 /**
- * How often Topology::clean() and Network::clean() are called in ms
+ * How often Topology::clean() and Network::clean() and similar are called, in ms
  */
 #define ZT_DB_CLEAN_PERIOD 300000
 
@@ -238,9 +238,32 @@
 #define ZT_MULTICAST_LOCAL_POLL_PERIOD 10000
 
 /**
- * Minimum delay between attempts to gather multicast topology info if members > 0
+ * Minimum delay between multicast endpoint gathering attempts
+ *
+ * Actual delay will vary between MIN and MAX research rate depending on
+ * how many endpoints we have -- MIN for 0, MAX for one less than limit.
+ * If we have the limit of known multicast endpoints, no further attempts
+ * to gather them are made.
  */
-#define ZT_MULTICAST_TOPOLOGY_RESEARCH_RATE_THROTTLE 120000
+#define ZT_MULTICAST_TOPOLOGY_GATHER_DELAY_MIN (ZT_MULTICAST_LIKE_EXPIRE / 50)
+
+/**
+ * Maximum delay between multicast endpoint gathering attempts
+ */
+#define ZT_MULTICAST_TOPOLOGY_GATHER_DELAY_MAX (ZT_MULTICAST_LIKE_EXPIRE / 2)
+
+/**
+ * Timeout for outgoing multicasts
+ *
+ * Attempts will be made to gather recipients and send until we reach
+ * the limit or sending times out.
+ */
+#define ZT_MULTICAST_TRANSMIT_TIMEOUT (ZT_MULTICAST_TOPOLOGY_GATHER_DELAY_MIN * 3)
+
+/**
+ * Default maximum number of peers to address with a single multicast (if unspecified in network)
+ */
+#define ZT_DEFAULT_MULTICAST_LIMIT 64
 
 /**
  * Delay between scans of the topology active peer DB for peers that need ping

node/MulticastTopology.cpp

@@ -41,19 +41,64 @@ MulticastTopology::~MulticastTopology()
 {
 }
 
-void MulticastTopology::clean(const Topology &topology)
+void MulticastTopology::add(const MulticastGroup &mg,const Address &member,const Address &learnedFrom)
 {
-	uint64_t now = Utils::now();
+	std::vector<MulticastGroupMember> &mv = _groups[mg].members;
+	for(std::vector<MulticastGroupMember>::iterator m(mv.begin());m!=mv.end();++m) {
+		if (m->address == member) {
+			if (m->learnedFrom) // once a member has been seen directly, we keep its status as direct
+				m->learnedFrom = learnedFrom;
+			m->timestamp = Utils::now();
+			return;
+		}
+	}
+	mv.push_back(MulticastGroupMember(member,learnedFrom,Utils::now()));
+}
 
-	for(std::map< MulticastGroup,std::vector<MulticastGroupMember> >::iterator mm(_members.begin());mm!=_members.end();) {
-		std::vector<MulticastGroupMember>::iterator reader(mm->second.begin());
-		std::vector<MulticastGroupMember>::iterator writer(mm->second.begin());
-		unsigned long count = 0;
-		while (reader != mm->second.end()) {
+void MulticastTopology::erase(const MulticastGroup &mg,const Address &member)
+{
+	std::map< MulticastGroup,MulticastGroupStatus >::iterator r(_groups.find(mg));
+	if (r != _groups.end()) {
+		for(std::vector<MulticastGroupMember>::iterator m(r->second.members.begin());m!=r->second.members.end();++m) {
+			if (m->address == member) {
+				r->second.members.erase(m);
+				if (r->second.members.empty())
+					_groups.erase(r);
+				return;
+			}
+		}
+	}
+}
+
+unsigned int MulticastTopology::want(const MulticastGroup &mg,uint64_t now,unsigned int limit,bool updateLastGatheredTimeOnNonzeroReturn)
+{
+	MulticastGroupStatus &gs = _groups[mg];
+	if ((unsigned int)gs.members.size() >= limit) {
+		// We already caught our limit, don't need to go fishing any more.
+		return 0;
+	} else {
+		// Compute the delay between fishing expeditions from the fraction of the limit that we already have.
+		const uint64_t rateDelay = (uint64_t)ZT_MULTICAST_TOPOLOGY_GATHER_DELAY_MIN + (uint64_t)(((double)gs.members.size() / (double)limit) * (double)(ZT_MULTICAST_TOPOLOGY_GATHER_DELAY_MAX - ZT_MULTICAST_TOPOLOGY_GATHER_DELAY_MIN));
+
+		if ((now - gs.lastGatheredMembers) >= rateDelay) {
+			if (updateLastGatheredTimeOnNonzeroReturn)
+				gs.lastGatheredMembers = now;
+			return (limit - (unsigned int)gs.members.size());
+		} else return 0;
+	}
+}
+
+void MulticastTopology::clean(uint64_t now,const Topology &topology)
+{
+	for(std::map< MulticastGroup,MulticastGroupStatus >::iterator mm(_groups.begin());mm!=_groups.end();) {
+		std::vector<MulticastGroupMember>::iterator reader(mm->second.members.begin());
+		std::vector<MulticastGroupMember>::iterator writer(mm->second.members.begin());
+		unsigned int count = 0;
+		while (reader != mm->second.members.end()) {
 			if ((now - reader->timestamp) < ZT_MULTICAST_LIKE_EXPIRE) {
 				*writer = *reader;
 
-				/* We sort in ascending order of most recent relevant activity. For peers we've learned
+				/* We rank in ascending order of most recent relevant activity. For peers we've learned
 				 * about by direct LIKEs, we do this in order of their own activity. For indirectly
 				 * acquired peers we do this minus a constant to place these categorically below directly
 				 * learned peers. For peers with no active Peer record, we use the time we last learned
@@ -78,10 +123,10 @@ void MulticastTopology::clean(const Topology &topology)
 		}
 
 		if (count) {
-			mm->second.resize(count);
-			std::sort(mm->second.begin(),mm->second.end()); // sorts in ascending order of rank
+			std::sort(mm->second.members.begin(),writer); // sorts in ascending order of rank
+			mm->second.members.resize(count); // trim off the ones we cut, after writer
 			++mm;
-		} else _members.erase(mm++);
+		} else _groups.erase(mm++);
 	}
 }
 

node/MulticastTopology.hpp

@@ -37,7 +37,6 @@
 #include "Constants.hpp"
 #include "Address.hpp"
 #include "MulticastGroup.hpp"
-#include "Mutex.hpp"
 #include "Utils.hpp"
 
 namespace ZeroTier {
@@ -46,6 +45,8 @@ class Topology;
 
 /**
  * Database of known multicast peers within a network
+ *
+ * This structure is not guarded by a mutex; the caller must synchronize access.
  */
 class MulticastTopology
 {
@@ -64,6 +65,14 @@ private:
 		inline bool operator<(const MulticastGroupMember &m) const throw() { return (rank < m.rank); }
 	};
 
+	struct MulticastGroupStatus
+	{
+		MulticastGroupStatus() : lastGatheredMembers(0) {}
+
+		uint64_t lastGatheredMembers; // time we last gathered members
+		std::vector<MulticastGroupMember> members; // members of this group
+	};
+
 public:
 	MulticastTopology();
 	~MulticastTopology();
@@ -75,20 +84,7 @@ public:
 	 * @param member Member to add/update
 	 * @param learnedFrom Address from which we learned this member or NULL/0 Address if direct
 	 */
-	inline void add(const MulticastGroup &mg,const Address &member,const Address &learnedFrom)
-	{
-		Mutex::Lock _l(_members_m);
-		std::vector<MulticastGroupMember> &mv = _members[mg];
-		for(std::vector<MulticastGroupMember>::iterator m(mv.begin());m!=mv.end();++m) {
-			if (m->address == member) {
-				if (m->learnedFrom) // once a member has been seen directly, we keep its status as direct
-					m->learnedFrom = learnedFrom;
-				m->timestamp = Utils::now();
-				return;
-			}
-		}
-		mv.push_back(MulticastGroupMember(member,learnedFrom,Utils::now()));
-	}
+	void add(const MulticastGroup &mg,const Address &member,const Address &learnedFrom);
 
 	/**
 	 * Erase a member from a multicast group (if present)
@@ -96,65 +92,50 @@ public:
 	 * @param mg Multicast group
 	 * @param member Member to erase
 	 */
-	inline void erase(const MulticastGroup &mg,const Address &member)
-	{
-		Mutex::Lock _l(_members_m);
-		std::map< MulticastGroup,std::vector<MulticastGroupMember> >::iterator r(_members.find(mg));
-		if (r != _members.end()) {
-			for(std::vector<MulticastGroupMember>::iterator m(r->second.begin());m!=r->second.end();++m) {
-				if (m->address == member) {
-					r->second.erase(m);
-					return;
-				}
-			}
-		}
-	}
+	void erase(const MulticastGroup &mg,const Address &member);
 
 	/**
 	 * @param mg Multicast group
-	 * @return Number of known peers in group
+	 * @return Tuple of: time we last gathered members (or 0 for never) and number of known members
 	 */
-	inline unsigned int memberCount(const MulticastGroup &mg) const
+	inline std::pair<uint64_t,unsigned int> groupStatus(const MulticastGroup &mg) const
 	{
-		Mutex::Lock _l(_members_m);
-		std::map< MulticastGroup,std::vector<MulticastGroupMember> >::const_iterator r(_members.find(mg));
-		return ((r != _members.end()) ? (unsigned int)r->second.size() : (unsigned int)0);
+		std::map< MulticastGroup,MulticastGroupStatus >::const_iterator r(_groups.find(mg));
+		return ((r != _groups.end()) ? std::pair<uint64_t,unsigned int>(r->second.lastGatheredMembers,r->second.members.size()) : std::pair<uint64_t,unsigned int>(0,0));
 	}
 
 	/**
-	 * Iterate over the known members of a multicast group
+	 * Return the number of new members we should want to gather or 0 for none
 	 *
 	 * @param mg Multicast group
-	 * @param func Function to be called with multicast group and address of member
-	 * @tparam F Function type (explicitly template on "FuncObj &" if reference instead of copy should be passed)
-	 * @return Number of members in multicast group for which function was called
+	 * @param now Current time
+	 * @param limit The maximum number we want per multicast group on this network
+	 * @param updateLastGatheredTimeOnNonzeroReturn If true, reset group's last gathered time to 'now' on non-zero return
 	 */
-	template<typename F>
-	inline unsigned int eachMember(const MulticastGroup &mg,F func) const
+	unsigned int want(const MulticastGroup &mg,uint64_t now,unsigned int limit,bool updateLastGatheredTimeOnNonzeroReturn);
+
+	/**
+	 * Update last gathered members time for a group
+	 *
+	 * @param mg Multicast group
+	 * @param now Current time
+	 */
+	inline void gatheringMembersNow(const MulticastGroup &mg,uint64_t now)
 	{
-		Mutex::Lock _l(_members_m);
-		std::map< MulticastGroup,std::vector<MulticastGroupMember> >::const_iterator r(_members.find(mg));
-		if (r != _members.end()) {
-			// We go in reverse order because most recently learned members are pushed to the end
-			// of the vector. The priority resort algorithm in clean() sorts in ascending order
-			// of propagation priority too.
-			for(std::vector<MulticastGroupMember>::const_reverse_iterator m(r->second.rbegin());m!=r->second.rend();++m) {
-				func(mg,m->address);
-			}
-			return (unsigned int)r->second.size();
-		} else return 0;
+		_groups[mg].lastGatheredMembers = now;
 	}
 
 	/**
 	 * Clean up and resort database
 	 *
+	 * @param now Current time
 	 * @param topology Global peer topology
+	 * @param trim Trim lists to a maximum of this many members per multicast group
 	 */
-	void clean(const Topology &topology);
+	void clean(uint64_t now,const Topology &topology);
 
 private:
-	std::map< MulticastGroup,std::vector<MulticastGroupMember> > _members;
-	Mutex _members_m;
+	std::map< MulticastGroup,MulticastGroupStatus > _groups;
 };
 
 } // namespace ZeroTier

node/Network.cpp

@@ -316,6 +316,8 @@ void Network::clean()
 			_multicastGroupsBehindMe.erase(mg++);
 		else ++mg;
 	}
+
+	_multicastTopology.clean(now,*(_r->topology),(_config) ? _config->multicastLimit() : (unsigned int)ZT_DEFAULT_MULTICAST_LIMIT);
 }
 
 Network::Status Network::status() const

node/Network.hpp

@@ -51,6 +51,7 @@
 #include "Identity.hpp"
 #include "InetAddress.hpp"
 #include "BandwidthAccount.hpp"
+#include "MulticastTopology.hpp"
 #include "NetworkConfig.hpp"
 #include "CertificateOfMembership.hpp"
 #include "Thread.hpp"
@@ -445,16 +446,13 @@ private:
 	EthernetTap *volatile _tap; // tap device or NULL if not initialized yet
 	volatile bool _enabled;
 
-	std::set<MulticastGroup> _myMulticastGroups; // multicast groups that we belong to including those behind us (updated periodically)
-	std::map<MulticastGroup,uint64_t> _multicastGroupsBehindMe; // multicast groups bridged to us and when we last saw activity on each
+	std::set< MulticastGroup > _myMulticastGroups; // multicast groups that we belong to including those behind us (updated periodically)
+	std::map< MulticastGroup,uint64_t > _multicastGroupsBehindMe; // multicast groups bridged to us and when we last saw activity on each
+	std::map< MulticastGroup,BandwidthAccount > _multicastRateAccounts;
+	MulticastTopology _multicastTopology;
 
 	std::map<MAC,Address> _remoteBridgeRoutes; // remote addresses where given MACs are reachable
 
-	// Deprecated, but will be kept around until P5_MULTICAST_FRAME is gone -- but the
-	// entry for us is still used by both. Eventually there will only be one BandwidthAccount,
-	// namely ours.
-	std::map< std::pair<Address,MulticastGroup>,BandwidthAccount > _multicastRateAccounts;
-
 	std::map<Address,CertificateOfMembership> _membershipCertificates; // Other members' certificates of membership
 	std::map<Address,uint64_t> _lastPushedMembershipCertificate; // When did we last push our certificate to each remote member?
 

node/OutboundMulticast.hpp

@@ -0,0 +1,125 @@
+/*
+ * ZeroTier One - Global Peer to Peer Ethernet
+ * Copyright (C) 2011-2014  ZeroTier Networks LLC
+ *
+ * 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
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * --
+ *
+ * ZeroTier may be used and distributed under the terms of the GPLv3, which
+ * are available at: http://www.gnu.org/licenses/gpl-3.0.html
+ *
+ * If you would like to embed ZeroTier into a commercial application or
+ * redistribute it in a modified binary form, please contact ZeroTier Networks
+ * LLC. Start here: http://www.zerotier.com/
+ */
+
+#ifndef ZT_OUTBOUNDMULTICAST_HPP
+#define ZT_OUTBOUNDMULTICAST_HPP
+
+#include <stdint.h>
+
+#include <vector>
+#include <algorithm>
+
+#include "Constants.hpp"
+#include "MAC.hpp"
+#include "MulticastGroup.hpp"
+#include "Address.hpp"
+#include "Packet.hpp"
+#include "Switch.hpp"
+
+namespace ZeroTier {
+
+/**
+ * An outbound multicast packet
+ *
+ * This object isn't guarded by a mutex; caller must synchronize access.
+ */
+class OutboundMulticast
+{
+public:
+	/**
+	 * Initialize outbound multicast
+	 *
+	 * @param timestamp Creation time
+	 * @param self My ZeroTier address
+	 * @param nwid Network ID
+	 * @param src Source MAC address of frame
+	 * @param dest Destination multicast group (MAC + ADI)
+	 * @param etherType 16-bit Ethernet type ID
+	 * @param payload Data
+	 * @param len Length of data
+	 * @throws std::out_of_range Data too large to fit in a MULTICAST_FRAME
+	 */
+	OutboundMulticast(uint64_t timestamp,const Address &self,uint64_t nwid,const MAC &src,const MulticastGroup &dest,unsigned int etherType,const void *payload,unsigned int len) :
+		_timestamp(timestamp),
+		_nwid(nwid),
+		_source(src),
+		_destination(dest),
+		_etherType(etherType)
+	{
+		_packet.setSource(self);
+		_packet.setVerb(Packet::VERB_MULTICAST_FRAME);
+		_packet.append((char)0);
+		_packet.append((uint32_t)dest.adi());
+		dest.mac().appendTo(_packet);
+		src.appendTo(_packet);
+		_packet.append((uint16_t)etherType);
+		_packet.append(payload,len);
+		_packet.compress();
+	}
+
+	/**
+	 * @return Multicast creation time
+	 */
+	inline uint64_t timestamp() const throw() { return _timestamp; }
+
+	/**
+	 * @return Number of unique recipients to which this packet has already been sent
+	 */
+	inline unsigned int sendCount() const throw() { return (unsigned int)_alreadySentTo.size(); }
+
+	/**
+	 * Try to send this to a given peer if it hasn't been sent to them already
+	 *
+	 * @param sw Switch instance to send packets
+	 * @param toAddr Destination address
+	 * @return True if address is new and packet was sent to switch, false if duplicate
+	 */
+	inline bool send(Switch &sw,const Address &toAddr)
+	{
+		// If things get really big, we can go to a sorted vector or a flat_set implementation
+		for(std::vector<Address>::iterator a(_alreadySentTo.begin());a!=_alreadySentTo.end();++a) {
+			if (*a == toAddr)
+				return false;
+		}
+		_alreadySentTo.push_back(toAddr);
+		sw.send(Packet(_packet,toAddr),true);
+		return true;
+	}
+
+private:
+	uint64_t _timestamp;
+	uint64_t _nwid;
+	MAC _source;
+	MulticastGroup _destination;
+	unsigned int _etherType;
+	Packet _packet; // packet contains basic structure of MULTICAST_FRAME and payload, is re-used with new IV and addressing each time
+	std::vector<Address> _alreadySentTo;
+};
+
+} // namespace ZeroTier
+
+#endif

node/Packet.cpp

@@ -48,7 +48,7 @@ const char *Packet::verbString(Verb v)
 		case VERB_NETWORK_MEMBERSHIP_CERTIFICATE: return "NETWORK_MEMBERSHIP_CERTIFICATE";
 		case VERB_NETWORK_CONFIG_REQUEST: return "NETWORK_CONFIG_REQUEST";
 		case VERB_NETWORK_CONFIG_REFRESH: return "NETWORK_CONFIG_REFRESH";
-		case VERB_MULTICAST_LONELY: return "MULTICAST_LONELY";
+		case VERB_MULTICAST_GATHER: return "MULTICAST_GATHER";
 		case VERB_MULTICAST_FRAME: return "MULTICAST_FRAME";
 	}
 	return "(unknown)";

node/Packet.hpp

@@ -664,7 +664,7 @@ public:
 		 *   <[8] 64-bit network ID>
 		 *   <[6] MAC address of multicast group being queried>
 		 *   <[4] 32-bit ADI for multicast group being queried>
-		 *   <[2] 16-bit (suggested) max number of multicast peers desired>
+		 *   <[4] 32-bit (suggested) max number of multicast peers desired or 0 for no limit>
 		 *  [<[...] network membership certificate (optional)>]
 		 *
 		 * Flags are:
@@ -673,13 +673,13 @@ public:
 		 * This message asks a peer for additional known endpoints that have
 		 * LIKEd a given multicast group. It's sent when the sender wishes
 		 * to send multicast but does not have the desired number of recipient
-		 * peers. (Hence it is "lonely." :)
+		 * peers.
 		 *
 		 * OK response payload:
 		 *   <[8] 64-bit network ID>
 		 *   <[6] MAC address of multicast group being queried>
 		 *   <[4] 32-bit ADI for multicast group being queried>
-		 *   <[2] 16-bit total number of known members in this multicast group>
+		 *   <[4] 32-bit total number of known members in this multicast group>
 		 *   <[2] 16-bit number of members enumerated in this packet>
 		 *   <[...] series of 5-byte ZeroTier addresses of enumerated members>
 		 *
@@ -691,7 +691,7 @@ public:
 		 * ERRORs are optional and are only generated if permission is denied,
 		 * certificate of membership is out of date, etc.
 		 */
-		VERB_MULTICAST_LONELY = 13,
+		VERB_MULTICAST_GATHER = 13,
 
 		/* Multicast frame:
 		 *   <[1] flags (currently unused, must be 0)>
@@ -709,6 +709,9 @@ public:
 		 * ERROR response payload:
 		 *   <[6] multicast group MAC>
 		 *   <[4] 32-bit multicast group ADI>
+		 *
+		 * ERRORs are optional and can be generated if a certificate is needed or if
+		 * multicasts for this multicast group are no longer wanted.
 		 */
 		VERB_MULTICAST_FRAME = 14
 	};
@@ -781,6 +784,22 @@ public:
 		(*this)[ZT_PACKET_IDX_FLAGS] = 0; // zero flags and hops
 	}
 
+	/**
+	 * Make a copy of a packet with a new initialization vector and destination address
+	 *
+	 * This can be used to take one draft prototype packet and quickly make copies to
+	 * encrypt for different destinations.
+	 *
+	 * @param prototype Prototype packet
+	 * @param dest Destination ZeroTier address for new packet
+	 */
+	Packet(const Packet &prototype,const Address &dest) :
+		Buffer<ZT_PROTO_MAX_PACKET_LENGTH>(prototype)
+	{
+		Utils::getSecureRandom(field(ZT_PACKET_IDX_IV,8),8);
+		setDestination(dest);
+	}
+
 	/**
 	 * Construct a new empty packet with a unique random packet ID
 	 * 

node/PacketDecoder.cpp

@@ -99,8 +99,8 @@ bool PacketDecoder::tryDecode(const RuntimeEnvironment *_r)
 				return _doFRAME(_r,peer);
 			case Packet::VERB_EXT_FRAME:
 				return _doEXT_FRAME(_r,peer);
-			case Packet::VERB_MULTICAST_FRAME:
-				return _doMULTICAST_FRAME(_r,peer);
+			case Packet::VERB_P5_MULTICAST_FRAME:
+				return _doP5_MULTICAST_FRAME(_r,peer);
 			case Packet::VERB_MULTICAST_LIKE:
 				return _doMULTICAST_LIKE(_r,peer);
 			case Packet::VERB_NETWORK_MEMBERSHIP_CERTIFICATE:

node/PacketDecoder.hpp

@@ -119,7 +119,7 @@ private:
 	bool _doRENDEZVOUS(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
 	bool _doFRAME(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
 	bool _doEXT_FRAME(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
-	bool _doMULTICAST_FRAME(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
+	bool _doP5_MULTICAST_FRAME(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
 	bool _doMULTICAST_LIKE(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
 	bool _doNETWORK_MEMBERSHIP_CERTIFICATE(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);
 	bool _doNETWORK_CONFIG_REQUEST(const RuntimeEnvironment *_r,const SharedPtr<Peer> &peer);

objects.mk

@@ -12,7 +12,7 @@ OBJS=\
 	node/Identity.o \
 	node/InetAddress.o \
 	node/Logger.o \
-	node/Multicaster.o \
+	node/MulticastTopology.o \
 	node/Network.o \
 	node/NetworkConfig.o \
 	node/Node.o \