Version 0.4.1 - RateLimiter for multicast, bug fixes.

This version adds a draft of the multicast rate limiting architecture. A
few minor bugs are also fixed. The Linux version builds in debug mode for
now.

Makefile.linux

@@ -6,12 +6,12 @@ ARCH=$(shell uname -m)
 DEFS=-DZT_ARCH="$(ARCH)" -DZT_OSNAME="linux" -DZT_TRACE
 
 # Uncomment for a release optimized build
-CFLAGS=-Wall -O3 -fno-unroll-loops -fstack-protector -pthread $(INCLUDES) -DNDEBUG $(DEFS)
-STRIP=strip --strip-all
+#CFLAGS=-Wall -O3 -fno-unroll-loops -fstack-protector -pthread $(INCLUDES) -DNDEBUG $(DEFS)
+#STRIP=strip --strip-all
 
 # Uncomment for a debug build
-#CFLAGS=-Wall -g -pthread $(INCLUDES) -DZT_TRACE -DZT_LOG_STDOUT $(DEFS)
-#STRIP=echo
+CFLAGS=-Wall -g -pthread $(INCLUDES) -DZT_TRACE $(DEFS)
+STRIP=echo
 
 CXXFLAGS=$(CFLAGS) -fno-rtti
 

node/Constants.hpp

@@ -226,9 +226,16 @@ error_no_ZT_ARCH_defined;
  * the sum of BREADTH^i where I is from 1 to DEPTH. This ignores the effect
  * of the rate limiting algorithm or bloom filter collisions.
  *
- * 7 results in a max of 21844 recipients for a given multicast.
+ * 5 results in a max of 1364 recipients for a given multicast. With a limit
+ * of 50 bytes/sec (average) for multicast, this results in a worst case of
+ * around 68kb/sec of multicast traffic. FYI the average multicast traffic
+ * from a Mac seems to be about ~25bytes/sec. Windows measurements are TBD.
+ * Linux is quieter than Mac.
+ *
+ * This are eventually going to become per-network tunable parameters, along
+ * with per-network peer multicast rate limits.
  */
-#define ZT_MULTICAST_PROPAGATION_DEPTH 7
+#define ZT_MULTICAST_PROPAGATION_DEPTH 5
 
 /**
  * Length of ring buffer history of recent multicast packets
@@ -259,6 +266,26 @@ error_no_ZT_ARCH_defined;
  */
 #define ZT_MULTICAST_LOCAL_POLL_PERIOD 10000
 
+/**
+ * Default bytes per second limit for multicasts per peer on a network
+ */
+#define ZT_MULTICAST_DEFAULT_BYTES_PER_SECOND 50.0
+
+/**
+ * Default balance preload for multicast rate limiters on a network
+ */
+#define ZT_MULTICAST_DEFAULT_RATE_PRELOAD 20000.0
+
+/**
+ * Default maximum balance for multicast rate limiters
+ */
+#define ZT_MULTICAST_DEFAULT_RATE_MAX_BALANCE 20000.0
+
+/**
+ * Default minimum balance for multicast rate limiters (max debt)
+ */
+#define ZT_MULTICAST_DEFAULT_RATE_MIN_BALANCE -10000.0
+
 /**
  * Delay between scans of the topology active peer DB for peers that need ping
  */

node/EthernetTap.cpp

@@ -187,7 +187,7 @@ EthernetTap::EthernetTap(
 
 	TRACE("tap %s created",_dev);
 
-	_thread = Thread<EthernetTap>::start(this);
+	_thread = Thread::start(this);
 }
 #endif // __LINUX__
 
@@ -271,14 +271,14 @@ EthernetTap::EthernetTap(
 
 	::pipe(_shutdownSignalPipe);
 
-	_thread = Thread<EthernetTap>::start(this);
+	_thread = Thread::start(this);
 }
 #endif // __APPLE__
 
 EthernetTap::~EthernetTap()
 {
 	::write(_shutdownSignalPipe[1],"\0",1); // causes thread to exit
-	Thread<EthernetTap>::join(_thread);
+	Thread::join(_thread);
 	::close(_fd);
 }
 
@@ -557,6 +557,10 @@ void EthernetTap::threadMain()
 	char getBuf[4096 + 14];
 	Buffer<4096> data;
 
+	// Wait for a moment after startup -- wait for Network to finish
+	// constructing itself.
+	Thread::sleep(500);
+
 	FD_ZERO(&readfds);
 	FD_ZERO(&nullfds);
 	int nfds = (int)std::max(_shutdownSignalPipe[0],_fd) + 1;

node/EthernetTap.hpp

@@ -180,7 +180,7 @@ private:
 	const unsigned int _mtu;
 
 	const RuntimeEnvironment *_r;
-	Thread<EthernetTap> _thread;
+	Thread _thread;
 
 	std::set<InetAddress> _ips;
 	Mutex _ips_m;

node/Multicaster.hpp

@@ -63,6 +63,9 @@ namespace ZeroTier {
  * This is written as a generic class so that it can be mocked and tested
  * in simulation. It also always takes 'now' as an argument, permitting
  * running in simulated time.
+ *
+ * This does not handle network permission or rate limiting, only the
+ * propagation algorithm.
  */
 class Multicaster
 {
@@ -328,6 +331,7 @@ private:
 	// Address and time of last LIKE
 	typedef std::pair<Address,uint64_t> MulticastMembership;
 
+	// Network : MulticastGroup -> vector<Address : time of last LIKE>
 	std::map< MulticastChannel,std::vector<MulticastMembership> > _multicastMemberships;
 	Mutex _multicastMemberships_m;
 };

node/Network.cpp

@@ -104,6 +104,17 @@ bool Network::Certificate::qualifyMembership(const Network::Certificate &mc) con
 	return true;
 }
 
+const char *Network::statusString(const Status s)
+	throw()
+{
+	switch(s) {
+		case NETWORK_WAITING_FOR_FIRST_AUTOCONF: return "WAITING_FOR_FIRST_AUTOCONF";
+		case NETWORK_OK: return "OK";
+		case NETWORK_ACCESS_DENIED: return "ACCESS_DENIED";
+	}
+	return "(invalid)";
+}
+
 Network::~Network()
 {
 	delete _tap;
@@ -111,8 +122,8 @@ Network::~Network()
 	if (_destroyOnDelete) {
 		std::string confPath(_r->homePath + ZT_PATH_SEPARATOR_S + "networks.d" + ZT_PATH_SEPARATOR_S + toString() + ".conf");
 		std::string mcdbPath(_r->homePath + ZT_PATH_SEPARATOR_S + "networks.d" + ZT_PATH_SEPARATOR_S + toString() + ".mcerts");
-		unlink(confPath.c_str());
-		unlink(mcdbPath.c_str());
+		Utils::rm(confPath);
+		Utils::rm(mcdbPath);
 	} else {
 		// Causes flush of membership certs to disk
 		clean();
@@ -128,7 +139,11 @@ SharedPtr<Network> Network::newInstance(const RuntimeEnvironment *renv,uint64_t
 	// that then causes the Network instance to be deleted before it is finished
 	// being constructed. C++ edge cases, how I love thee.
 	SharedPtr<Network> nw(new Network());
+	nw->_ready = false; // disable handling of Ethernet frames during construct
 	nw->_r = renv;
+	nw->_rlLimit.bytesPerSecond = ZT_MULTICAST_DEFAULT_BYTES_PER_SECOND;
+	nw->_rlLimit.maxBalance = ZT_MULTICAST_DEFAULT_RATE_MAX_BALANCE;
+	nw->_rlLimit.minBalance = ZT_MULTICAST_DEFAULT_RATE_MIN_BALANCE;
 	nw->_tap = new EthernetTap(renv,renv->identity.address().toMAC(),ZT_IF_MTU,&_CBhandleTapData,nw.ptr());
 	nw->_id = id;
 	nw->_lastConfigUpdate = 0;
@@ -136,6 +151,7 @@ SharedPtr<Network> Network::newInstance(const RuntimeEnvironment *renv,uint64_t
 	if (nw->controller() == renv->identity.address()) // sanity check, this isn't supported for now
 		throw std::runtime_error("cannot add a network for which I am the netconf master");
 	nw->_restoreState();
+	nw->_ready = true; // enable handling of Ethernet frames
 	nw->requestConfiguration();
 	return nw;
 }
@@ -202,9 +218,10 @@ void Network::clean()
 	std::string mcdbPath(_r->homePath + ZT_PATH_SEPARATOR_S + "networks.d" + ZT_PATH_SEPARATOR_S + toString() + ".mcerts");
 
 	Mutex::Lock _l(_lock);
+
 	if (_configuration.isOpen()) {
 		_membershipCertificates.clear();
-		unlink(mcdbPath.c_str());
+		Utils::rm(mcdbPath);
 	} else {
 		FILE *mcdb = fopen(mcdbPath.c_str(),"wb");
 		bool writeError = false;
@@ -236,14 +253,24 @@ void Network::clean()
 		if (mcdb)
 			fclose(mcdb);
 		if (writeError) {
-			unlink(mcdbPath.c_str());
+			Utils::rm(mcdbPath);
 			LOG("error: unable to write to membership cert database at: %s",mcdbPath.c_str());
 		}
 	}
 }
 
+Network::Status Network::status() const
+{
+	Mutex::Lock _l(_lock);
+	if (_configuration.containsAllFields())
+		return NETWORK_OK;
+	return NETWORK_WAITING_FOR_FIRST_AUTOCONF;
+}
+
 void Network::_CBhandleTapData(void *arg,const MAC &from,const MAC &to,unsigned int etherType,const Buffer<4096> &data)
 {
+	if (!((Network *)arg)->_ready)
+		return;
 	const RuntimeEnvironment *_r = ((Network *)arg)->_r;
 	try {
 		_r->sw->onLocalEthernet(SharedPtr<Network>((Network *)arg),from,to,etherType,data);

node/Network.hpp

@@ -47,6 +47,7 @@
 #include "Dictionary.hpp"
 #include "Identity.hpp"
 #include "InetAddress.hpp"
+#include "RateLimiter.hpp"
 
 namespace ZeroTier {
 
@@ -265,6 +266,23 @@ public:
 		}
 	};
 
+	/**
+	 * Status for networks
+	 */
+	enum Status
+	{
+		NETWORK_WAITING_FOR_FIRST_AUTOCONF,
+		NETWORK_OK,
+		NETWORK_ACCESS_DENIED
+	};
+
+	/**
+	 * @param s Status
+	 * @return String description
+	 */
+	static const char *statusString(const Status s)
+		throw();
+
 private:
 	// Only NodeConfig can create, only SharedPtr can delete
 
@@ -404,12 +422,39 @@ public:
 		return _lastConfigUpdate;
 	}
 
+	/**
+	 * @return Status of this network
+	 */
+	Status status() const;
+
+	/**
+	 * Invoke multicast rate limiter gate for a given address
+	 *
+	 * @param addr Address to check
+	 * @param bytes Bytes address wishes to send us / propagate
+	 * @return True if allowed, false if overshot rate limit
+	 */
+	inline bool multicastRateGate(const Address &addr,unsigned int bytes)
+	{
+		Mutex::Lock _l(_lock);
+		std::map<Address,RateLimiter>::iterator rl(_multicastRateLimiters.find(addr));
+		if (rl == _multicastRateLimiters.end()) {
+			RateLimiter &newrl = _multicastRateLimiters[addr];
+			newrl.init(ZT_MULTICAST_DEFAULT_RATE_PRELOAD);
+			return newrl.gate(_rlLimit,(double)bytes);
+		}
+		return rl->second.gate(_rlLimit,(double)bytes);
+	}
+
 private:
 	static void _CBhandleTapData(void *arg,const MAC &from,const MAC &to,unsigned int etherType,const Buffer<4096> &data);
 	void _restoreState();
 
 	const RuntimeEnvironment *_r;
 
+	// Rate limits for this network
+	RateLimiter::Limit _rlLimit;
+
 	// Tap and tap multicast memberships
 	EthernetTap *_tap;
 	std::set<MulticastGroup> _multicastGroups;
@@ -417,6 +462,9 @@ private:
 	// Membership certificates supplied by peers
 	std::map<Address,Certificate> _membershipCertificates;
 
+	// Rate limiters for each multicasting peer
+	std::map<Address,RateLimiter> _multicastRateLimiters;
+
 	// Configuration from network master node
 	Config _configuration;
 	Certificate _myCertificate;
@@ -424,6 +472,7 @@ private:
 	uint64_t _id;
 	volatile uint64_t _lastConfigUpdate;
 	volatile bool _destroyOnDelete;
+	volatile bool _ready;
 
 	Mutex _lock;
 

node/Node.cpp

@@ -339,8 +339,8 @@ Node::ReasonForTermination Node::run()
 		Utils::lockDownFile(identitySecretPath.c_str(),false);
 
 		// Clean up some obsolete files if present -- this will be removed later
-		unlink((_r->homePath + ZT_PATH_SEPARATOR_S + "status").c_str());
-		unlink((_r->homePath + ZT_PATH_SEPARATOR_S + "thisdeviceismine").c_str());
+		Utils::rm((_r->homePath + ZT_PATH_SEPARATOR_S + "status"));
+		Utils::rm((_r->homePath + ZT_PATH_SEPARATOR_S + "thisdeviceismine"));
 
 		// Make sure networks.d exists
 		mkdir((_r->homePath + ZT_PATH_SEPARATOR_S + "networks.d").c_str(),0700);
@@ -431,7 +431,7 @@ Node::ReasonForTermination Node::run()
 			if (lastDelayDelta >= ZT_SLEEP_WAKE_DETECTION_THRESHOLD) {
 				resynchronize = true;
 				LOG("probable suspend/resume detected, pausing a moment for things to settle...");
-				Thread<Node>::sleep(ZT_SLEEP_WAKE_SETTLE_TIME);
+				Thread::sleep(ZT_SLEEP_WAKE_SETTLE_TIME);
 			}
 
 			// Periodically check our network environment, sending pings out to all

node/NodeConfig.cpp

@@ -175,11 +175,12 @@ std::vector<std::string> NodeConfig::execute(const char *command)
 			}
 			// TODO: display network status, such as "permission denied to closed
 			// network" or "waiting".
-			_P("200 listnetworks %.16llx OK %s %s %s",
+			_P("200 listnetworks %.16llx %s %s %s %s",
 				(unsigned long long)nw->first,
-				(nw->second->isOpen() ? "public" : "private"),
+				Network::statusString(nw->second->status()),
+				(nw->second->isOpen() ? "open" : "private"),
 				nw->second->tap().deviceName().c_str(),
-				tmp.c_str());
+				((tmp.length() > 0) ? tmp.c_str() : "-"));
 		}
 	} else if (cmd[0] == "join") {
 		if (cmd.size() > 1) {

node/PacketDecoder.cpp

@@ -494,106 +494,106 @@ bool PacketDecoder::_doMULTICAST_FRAME(const RuntimeEnvironment *_r,const Shared
 {
 	try {
 		SharedPtr<Network> network(_r->nc->network(at<uint64_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_NETWORK_ID)));
-		if (network) {
-			if (network->isAllowed(source())) {
-				if (size() > ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD) {
+		if ((network)&&(network->isAllowed(source()))) {
+			Address originalSubmitterAddress(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SUBMITTER_ADDRESS,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH);
+			MAC fromMac(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SOURCE_MAC,6));
+			MulticastGroup mg(MAC(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_DESTINATION_MAC,6)),at<uint32_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_ADI));
+			unsigned int hops = (*this)[ZT_PROTO_VERB_MULTICAST_FRAME_IDX_HOP_COUNT];
+			unsigned int etherType = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_ETHERTYPE);
+			unsigned int datalen = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD_LENGTH);
+			unsigned int signaturelen = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SIGNATURE_LENGTH);
+			unsigned char *dataAndSignature = field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD,datalen + signaturelen);
 
-					Address originalSubmitterAddress(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SUBMITTER_ADDRESS,ZT_ADDRESS_LENGTH),ZT_ADDRESS_LENGTH);
-					MAC fromMac(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SOURCE_MAC,6));
-					MulticastGroup mg(MAC(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_DESTINATION_MAC,6)),at<uint32_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_ADI));
-					unsigned int hops = (*this)[ZT_PROTO_VERB_MULTICAST_FRAME_IDX_HOP_COUNT];
-					unsigned int etherType = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_ETHERTYPE);
-					unsigned int datalen = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD_LENGTH);
-					unsigned int signaturelen = at<uint16_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_SIGNATURE_LENGTH);
-					unsigned char *dataAndSignature = field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_PAYLOAD,datalen + signaturelen);
+			uint64_t mccrc = Multicaster::computeMulticastDedupCrc(network->id(),fromMac,mg,etherType,dataAndSignature,datalen);
+			uint64_t now = Utils::now();
+			bool isDuplicate = _r->multicaster->checkDuplicate(mccrc,now);
 
-					uint64_t mccrc = Multicaster::computeMulticastDedupCrc(network->id(),fromMac,mg,etherType,dataAndSignature,datalen);
-					uint64_t now = Utils::now();
-					bool isDuplicate = _r->multicaster->checkDuplicate(mccrc,now);
+			if (originalSubmitterAddress == _r->identity.address()) {
+				// Technically should not happen, since the original submitter is
+				// excluded from consideration as a propagation recipient.
+				TRACE("dropped boomerang MULTICAST_FRAME received from %s(%s)",source().toString().c_str(),_remoteAddress.toString().c_str());
+			} else if ((!isDuplicate)||(_r->topology->amSupernode())) {
+				//
+				// If I am a supernode, I will repeatedly propagate duplicates. That's
+				// because supernodes are used to bridge sparse multicast groups. Non-
+				// supernodes will ignore duplicates completely.
+				//
+				// TODO: supernodes should keep a local bloom filter too and OR it with
+				// the bloom from the packet in order to pick different recipients each
+				// time a multicast returns to them for repropagation.
+				//
 
-					if (originalSubmitterAddress == _r->identity.address()) {
-						// Technically should not happen, since the original submitter is
-						// excluded from consideration as a propagation recipient.
-						TRACE("dropped boomerang MULTICAST_FRAME received from %s(%s)",source().toString().c_str(),_remoteAddress.toString().c_str());
-					} else if ((!isDuplicate)||(_r->topology->amSupernode())) {
-						//
-						// If I am a supernode, I will repeatedly propagate duplicates. That's
-						// because supernodes are used to bridge sparse multicast groups. Non-
-						// supernodes will ignore duplicates completely.
-						//
-						// TODO: supernodes should keep a local bloom filter too and OR it with
-						// the bloom from the packet in order to pick different recipients each
-						// time a multicast returns to them for repropagation.
-						//
+				SharedPtr<Peer> originalSubmitter(_r->topology->getPeer(originalSubmitterAddress));
+				if (!originalSubmitter) {
+					TRACE("requesting WHOIS on original multicast frame submitter %s",originalSubmitterAddress.toString().c_str());
+					_r->sw->requestWhois(originalSubmitterAddress);
+					_step = DECODE_WAITING_FOR_MULTICAST_FRAME_ORIGINAL_SENDER_LOOKUP;
+					return false; // try again if/when we get OK(WHOIS)
+				} else if (Multicaster::verifyMulticastPacket(originalSubmitter->identity(),network->id(),fromMac,mg,etherType,dataAndSignature,datalen,dataAndSignature + datalen,signaturelen)) {
+					// In checking the multicast rate, we don't re-check if this is
+					// a duplicate. That's because if isDuplicate is true it means
+					// we're a supernode and it's a second pass relay.
+					if ((isDuplicate)||(network->multicastRateGate(originalSubmitter->address(),datalen))) {
+						_r->multicaster->addToDedupHistory(mccrc,now);
 
-						SharedPtr<Peer> originalSubmitter(_r->topology->getPeer(originalSubmitterAddress));
-						if (!originalSubmitter) {
-							TRACE("requesting WHOIS on original multicast frame submitter %s",originalSubmitterAddress.toString().c_str());
-							_r->sw->requestWhois(originalSubmitterAddress);
-							_step = DECODE_WAITING_FOR_MULTICAST_FRAME_ORIGINAL_SENDER_LOOKUP;
-							return false; // try again if/when we get OK(WHOIS)
-						} else if (Multicaster::verifyMulticastPacket(originalSubmitter->identity(),network->id(),fromMac,mg,etherType,dataAndSignature,datalen,dataAndSignature + datalen,signaturelen)) {
-							_r->multicaster->addToDedupHistory(mccrc,now);
+						// Even if we are a supernode, we still don't repeatedly inject
+						// duplicates into our own tap.
+						if (!isDuplicate)
+							network->tap().put(fromMac,mg.mac(),etherType,dataAndSignature,datalen);
 
-							// Even if we are a supernode, we still don't repeatedly inject
-							// duplicates into our own tap.
-							if (!isDuplicate)
-								network->tap().put(fromMac,mg.mac(),etherType,dataAndSignature,datalen);
+						if (++hops < ZT_MULTICAST_PROPAGATION_DEPTH) {
+							Address upstream(source()); // save this since we mangle it
 
-							if (++hops < ZT_MULTICAST_PROPAGATION_DEPTH) {
-								Address upstream(source()); // save this since we mangle it
+							Multicaster::MulticastBloomFilter bloom(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_BLOOM_FILTER,ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE_BYTES));
+							SharedPtr<Peer> propPeers[ZT_MULTICAST_PROPAGATION_BREADTH];
+							unsigned int np = _r->multicaster->pickNextPropagationPeers(
+								*(_r->prng),
+								*(_r->topology),
+								network->id(),
+								mg,
+								originalSubmitterAddress,
+								upstream,
+								bloom,
+								ZT_MULTICAST_PROPAGATION_BREADTH,
+								propPeers,
+								now);
 
-								Multicaster::MulticastBloomFilter bloom(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_BLOOM_FILTER,ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE_BYTES));
-								SharedPtr<Peer> propPeers[ZT_MULTICAST_PROPAGATION_BREADTH];
-								unsigned int np = _r->multicaster->pickNextPropagationPeers(
-									*(_r->prng),
-									*(_r->topology),
-									network->id(),
-									mg,
-									originalSubmitterAddress,
-									upstream,
-									bloom,
-									ZT_MULTICAST_PROPAGATION_BREADTH,
-									propPeers,
-									now);
+							// In a bit of a hack, we re-use this packet to repeat it
+							// to our multicast propagation recipients. Afterwords we
+							// return true just to be sure this is the end of this
+							// packet's life cycle, since it is now mangled.
 
-								// In a bit of a hack, we re-use this packet to repeat it
-								// to our multicast propagation recipients. Afterwords we
-								// return true just to be sure this is the end of this
-								// packet's life cycle, since it is now mangled.
+							setSource(_r->identity.address());
+							(*this)[ZT_PROTO_VERB_MULTICAST_FRAME_IDX_HOP_COUNT] = hops;
+							memcpy(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_BLOOM_FILTER,ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE_BYTES),bloom.data(),ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE_BYTES);
+							compress();
 
-								setSource(_r->identity.address());
-								(*this)[ZT_PROTO_VERB_MULTICAST_FRAME_IDX_HOP_COUNT] = hops;
-								memcpy(field(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_BLOOM_FILTER,ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE_BYTES),bloom.data(),ZT_PROTO_VERB_MULTICAST_FRAME_BLOOM_FILTER_SIZE_BYTES);
-								compress();
-
-								for(unsigned int i=0;i<np;++i) {
-									//TRACE("propagating multicast from original node %s: %s -> %s",originalSubmitterAddress.toString().c_str(),upstream.toString().c_str(),propPeers[i]->address().toString().c_str());
-									// Re-use this packet to re-send multicast frame to everyone
-									// downstream from us.
-									newInitializationVector();
-									setDestination(propPeers[i]->address());
-									_r->sw->send(*this,true);
-								}
-
-								return true;
-							} else {
-								//TRACE("terminating MULTICAST_FRAME propagation from %s(%s): max depth reached",source().toString().c_str(),_remoteAddress.toString().c_str());
+							for(unsigned int i=0;i<np;++i) {
+								//TRACE("propagating multicast from original node %s: %s -> %s",originalSubmitterAddress.toString().c_str(),upstream.toString().c_str(),propPeers[i]->address().toString().c_str());
+								// Re-use this packet to re-send multicast frame to everyone
+								// downstream from us.
+								newInitializationVector();
+								setDestination(propPeers[i]->address());
+								_r->sw->send(*this,true);
 							}
+
+							// Return here just to be safe, since this packet's state is no
+							// longer valid.
+							return true;
 						} else {
-							LOG("rejected MULTICAST_FRAME from %s(%s) due to failed signature check (falsely claims origin %s)",source().toString().c_str(),_remoteAddress.toString().c_str(),originalSubmitterAddress.toString().c_str());
+							//TRACE("terminating MULTICAST_FRAME propagation from %s(%s): max depth reached",source().toString().c_str(),_remoteAddress.toString().c_str());
 						}
 					} else {
-						TRACE("dropped redundant MULTICAST_FRAME from %s(%s)",source().toString().c_str(),_remoteAddress.toString().c_str());
+						LOG("dropped MULTICAST_FRAME from original sender %s: rate limit overrun",originalSubmitter->address().toString().c_str());
 					}
 				} else {
-					TRACE("dropped MULTICAST_FRAME from %s(%s): invalid short packet",source().toString().c_str(),_remoteAddress.toString().c_str());
+					TRACE("rejected MULTICAST_FRAME forwarded by %s(%s): failed signature check (falsely claims origin %s)",source().toString().c_str(),_remoteAddress.toString().c_str(),originalSubmitterAddress.toString().c_str());
 				}
 			} else {
-				TRACE("dropped MULTICAST_FRAME from %s(%s): not a member of closed network %llu",source().toString().c_str(),_remoteAddress.toString().c_str(),network->id());
+				TRACE("dropped duplicate MULTICAST_FRAME from %s(%s)",source().toString().c_str(),_remoteAddress.toString().c_str());
 			}
 		} else {
-			TRACE("dropped MULTICAST_FRAME from %s(%s): network %llu unknown or we are not a member",source().toString().c_str(),_remoteAddress.toString().c_str(),at<uint64_t>(ZT_PROTO_VERB_MULTICAST_FRAME_IDX_NETWORK_ID));
+			TRACE("dropped MULTICAST_FRAME from %s(%s): network %.16llx unknown or sender not allowed",source().toString().c_str(),_remoteAddress.toString().c_str(),(unsigned long long)network->id());
 		}
 	} catch (std::exception &ex) {
 		TRACE("dropped MULTICAST_FRAME from %s(%s): unexpected exception: %s",source().toString().c_str(),_remoteAddress.toString().c_str(),ex.what());

node/RateLimiter.hpp

@@ -0,0 +1,138 @@
+/*
+ * ZeroTier One - Global Peer to Peer Ethernet
+ * Copyright (C) 2012-2013  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_RATELIMITER_HPP
+#define _ZT_RATELIMITER_HPP
+
+#include <math.h>
+#include "Utils.hpp"
+
+namespace ZeroTier {
+
+/**
+ * Burstable rate limiter
+ *
+ * This limits a transfer rate to a maximum bytes per second using an
+ * accounting method based on a balance rather than accumulating an
+ * average rate. The result is a burstable rate limit rather than a
+ * continuous rate limit; the link being limited may use all its balance
+ * at once or slowly over time. Balance constantly replenishes over time
+ * up to a configurable maximum balance.
+ */
+class RateLimiter
+{
+public:
+	/**
+	 * Limits to apply to a rate limiter
+	 *
+	 * Since many rate limiters may share the same fixed limit values,
+	 * save memory by breaking this out into a struct parameter that
+	 * can be passed into RateLimiter's methods.
+	 */
+	struct Limit
+	{
+		/**
+		 * Speed in bytes per second, or rate of balance accrual
+		 */
+		double bytesPerSecond;
+
+		/**
+		 * Maximum balance that can ever be accrued (should be > 0.0)
+		 */
+		double maxBalance;
+
+		/**
+		 * Minimum balance, or maximum allowable "debt" (should be <= 0.0)
+		 */
+		double minBalance;
+	};
+
+	/**
+	 * Create an uninitialized rate limiter
+	 *
+	 * init() must be called before this is used.
+	 */
+	RateLimiter() throw() {}
+
+	/**
+	 * @param preload Initial balance to place in account
+	 */
+	RateLimiter(double preload)
+		throw()
+	{
+		init(preload);
+	}
+
+	/**
+	 * Initialize or re-initialize rate limiter
+	 *
+	 * @param preload Initial balance to place in account
+	 */
+	inline void init(double preload)
+		throw()
+	{
+		_lastTime = Utils::nowf();
+		_balance = preload;
+	}
+
+	/**
+	 * Update balance based on current clock and supplied Limits bytesPerSecond and maxBalance
+	 *
+	 * @param lim Current limits in effect
+	 * @return New balance
+	 */
+	inline double updateBalance(const Limit &lim)
+		throw()
+	{
+		double lt = _lastTime;
+		double now = _lastTime = Utils::nowf();
+		return (_balance = fmin(lim.maxBalance,_balance + (lim.bytesPerSecond * (now - lt))));
+	}
+
+	/**
+	 * Update balance and test if a block of 'bytes' should be permitted to be transferred
+	 *
+	 * @param lim Current limits in effect
+	 * @param bytes Number of bytes that we wish to transfer
+	 * @return True if balance was sufficient
+	 */
+	inline bool gate(const Limit &lim,double bytes)
+		throw()
+	{
+		bool allow = (updateBalance(lim) >= bytes);
+		_balance = fmax(lim.minBalance,_balance - bytes);
+		return allow;
+	}
+
+private:
+	double _lastTime;
+	double _balance;
+};
+
+} // namespace ZeroTier
+
+#endif

node/Service.cpp

@@ -62,7 +62,7 @@ Service::Service(const RuntimeEnvironment *renv,const char *name,const char *pat
 	_childStderr(0),
 	_run(true)
 {
-	_thread = Thread<Service>::start(this);
+	_thread = Thread::start(this);
 }
 
 Service::~Service()
@@ -77,14 +77,14 @@ Service::~Service()
 				pid = 0;
 				break;
 			}
-			Thread<Service>::sleep(100);
+			Thread::sleep(100);
 		}
 		if (pid > 0) {
 			::kill(pid,SIGKILL);
 			waitpid(pid,&st,0);
 		}
 	}
-	Thread<Service>::join(_thread);
+	Thread::join(_thread);
 }
 
 bool Service::send(const Dictionary &msg)
@@ -136,7 +136,7 @@ void Service::threadMain()
 				close(in[0]);
 				close(out[1]);
 				close(err[1]);
-				Thread<Service>::sleep(500); // give child time to start
+				Thread::sleep(500); // give child time to start
 				_childStdin = in[1];
 				_childStdout = out[0];
 				_childStderr = err[0];
@@ -168,7 +168,7 @@ void Service::threadMain()
 
 				LOG("service %s exited with exit code: %d, delaying 1s to attempt relaunch",_name.c_str(),st);
 
-				Thread<Service>::sleep(1000); // wait to relaunch
+				Thread::sleep(1000); // wait to relaunch
 				continue;
 			}
 		}

node/Service.hpp

@@ -114,7 +114,7 @@ public:
 
 private:
 	const RuntimeEnvironment *_r;
-	Thread<Service> _thread;
+	Thread _thread;
 	std::string _path;
 	std::string _name;
 	void *_arg;

node/Thread.hpp

@@ -57,11 +57,8 @@ static void *___zt_threadMain(void *instance)
 }
 
 /**
- * A thread of a given class type
- *
- * @tparam C Class using Thread
+ * A thread identifier, and static methods to start and join threads
  */
-template<typename C>
 class Thread
 {
 public:
@@ -90,7 +87,9 @@ public:
 	 * @param instance Instance whose threadMain() method gets called by new thread
 	 * @return Thread identifier
 	 * @throws std::runtime_error Unable to create thread
+	 * @tparam C Class containing threadMain()
 	 */
+	template<typename C>
 	static inline Thread start(C *instance)
 		throw(std::runtime_error)
 	{

node/Topology.cpp

@@ -54,7 +54,7 @@ Topology::Topology(const RuntimeEnvironment *renv,const char *dbpath)
 
 	Utils::lockDownFile(dbpath,false); // node.db caches secrets
 
-	_thread = Thread<Topology>::start(this);
+	_thread = Thread::start(this);
 }
 
 Topology::~Topology()
@@ -67,7 +67,7 @@ Topology::~Topology()
 		_peerDeepVerifyJobs.back().type = _PeerDeepVerifyJob::EXIT_THREAD;
 	}
 	_peerDeepVerifyJobs_c.signal();
-	Thread<Topology>::join(_thread);
+	Thread::join(_thread);
 	KISSDB_close(&_dbm);
 }
 

node/Topology.hpp

@@ -299,7 +299,7 @@ private:
 	};
 
 	const RuntimeEnvironment *const _r;
-	Thread<Topology> _thread;
+	Thread _thread;
 
 	std::map< Address,SharedPtr<Peer> > _activePeers;
 	Mutex _activePeers_m;

node/UdpSocket.cpp

@@ -120,7 +120,7 @@ UdpSocket::UdpSocket(
 		}
 	}
 
-	_thread = Thread<UdpSocket>::start(this);
+	_thread = Thread::start(this);
 }
 
 UdpSocket::~UdpSocket()
@@ -131,7 +131,7 @@ UdpSocket::~UdpSocket()
 		::shutdown(s,SHUT_RDWR);
 		::close(s);
 	}
-	Thread<UdpSocket>::join(_thread);
+	Thread::join(_thread);
 }
 
 bool UdpSocket::send(const InetAddress &to,const void *data,unsigned int len,int hopLimit)

node/UdpSocket.hpp

@@ -94,7 +94,7 @@ public:
 		throw();
 
 private:
-	Thread<UdpSocket> _thread;
+	Thread _thread;
 	void (*_packetHandler)(UdpSocket *,void *,const InetAddress &,const void *,unsigned int);
 	void *_arg;
 	int _localPort;

node/Utils.hpp

@@ -43,6 +43,12 @@
 #include "../ext/lz4/lz4.h"
 #include "../ext/lz4/lz4hc.h"
 
+#ifdef __WINDOWS__
+#include <windows.h>
+#else
+#include <unistd.h>
+#endif
+
 #include "Constants.hpp"
 
 /**
@@ -58,6 +64,27 @@ namespace ZeroTier {
 class Utils
 {
 public:
+	/**
+	 * Delete a file
+	 *
+	 * @param path Path to delete
+	 * @return True if delete was successful
+	 */
+	static inline bool rm(const char *path)
+		throw()
+	{
+#ifdef __WINDOWS__
+		foo;
+#else
+		return (unlink(path) == 0);
+#endif
+	}
+	static inline bool rm(const std::string &path)
+		throw()
+	{
+		return rm(path.c_str());
+	}
+
 	/**
 	 * List a directory's contents
 	 *
@@ -320,6 +347,17 @@ public:
 		return ( (1000ULL * (uint64_t)tv.tv_sec) + (uint64_t)(tv.tv_usec / 1000) );
 	};
 
+	/**
+	 * @return Current time in seconds since epoch, to the highest available resolution
+	 */
+	static inline double nowf()
+		throw()
+	{
+		struct timeval tv;
+		gettimeofday(&tv,(struct timezone *)0);
+		return ( ((double)tv.tv_sec) + (((double)tv.tv_usec) / 1000000.0) );
+	}
+
 	/**
 	 * Read the full contents of a file into a string buffer
 	 *

version.h

@@ -41,6 +41,6 @@
 /**
  * Revision: 16-bit (0-65535)
  */
-#define ZEROTIER_ONE_VERSION_REVISION 0
+#define ZEROTIER_ONE_VERSION_REVISION 1
 
 #endif