Add daemon thread to controller and move network member cache refreshes there.

2 files changed, 107 insertions, 41 deletions

diff --git a/controller/EmbeddedNetworkController.cpp b/controller/EmbeddedNetworkController.cpp
index e98e4d6d..713b7618 100644
--- a/controller/EmbeddedNetworkController.cpp
+++ b/controller/EmbeddedNetworkController.cpp
@@ -60,6 +60,22 @@ using json = nlohmann::json;
 
 namespace ZeroTier {
 
+// JSON blob I/O
+static json _readJson(const std::string &path)
+{
+	std::string buf;
+	if (OSUtils::readFile(path.c_str(),buf)) {
+		try {
+			return json::parse(buf);
+		} catch ( ... ) {}
+	}
+	return json::object();
+}
+static bool _writeJson(const std::string &path,const json &obj)
+{
+	return OSUtils::writeFile(path.c_str(),obj.dump(2));
+}
+
 // Get JSON values as unsigned integers, strings, or booleans, doing type conversion if possible
 static uint64_t _jI(const json &jv,const uint64_t dfl)
 {
@@ -394,16 +410,67 @@ static bool _parseRule(const json &r,ZT_VirtualNetworkRule &rule)
 
 EmbeddedNetworkController::EmbeddedNetworkController(Node *node,const char *dbPath) :
 	_node(node),
-	_path(dbPath)
+	_path(dbPath),
+	_daemonRun(true)
 {
 	OSUtils::mkdir(dbPath);
 	OSUtils::lockDownFile(dbPath,true); // networks might contain auth tokens, etc., so restrict directory permissions
+	_daemon = Thread::start(this);
 }
 
 EmbeddedNetworkController::~EmbeddedNetworkController()
 {
 }
 
+void EmbeddedNetworkController::threadMain()
+	throw()
+{
+	uint64_t lastUpdatedNetworkMemberCache = 0;
+	while (_daemonRun) {
+		// Every 60 seconds we rescan the filesystem for network members and rebuild our cache
+		if ((OSUtils::now() - lastUpdatedNetworkMemberCache) >= 60000) {
+			const std::vector<std::string> networks(OSUtils::listSubdirectories((_path + ZT_PATH_SEPARATOR_S + "network").c_str()));
+			for(auto n=networks.begin();n!=networks.end();++n) {
+				if (n->length() == 16) {
+					const std::vector<std::string> members(OSUtils::listSubdirectories((*n + ZT_PATH_SEPARATOR_S + "member").c_str()));
+					std::map<Address,nlohmann::json> newCache;
+					for(auto m=members.begin();m!=members.end();++m) {
+						if (m->length() == ZT_ADDRESS_LENGTH_HEX) {
+							const Address maddr(*m);
+							try {
+								const json mj(_readJson((_path + ZT_PATH_SEPARATOR_S + "network" + ZT_PATH_SEPARATOR_S + *n + ZT_PATH_SEPARATOR_S + "member" + ZT_PATH_SEPARATOR_S + *m + ZT_PATH_SEPARATOR_S + "config.json")));
+								if ((mj.is_object())&&(mj.size() > 0)) {
+									newCache[maddr] = mj;
+								}
+							} catch ( ... ) {}
+						}
+					}
+					{
+						Mutex::Lock _l(_networkMemberCache_m);
+						_networkMemberCache[Utils::hexStrToU64(n->c_str())] = newCache;
+					}
+				}
+			}
+			lastUpdatedNetworkMemberCache = OSUtils::now();
+		}
+
+		{	// Every 25ms we push up to 50 network refreshes, which amounts to a max of about 300-500kb/sec
+			unsigned int count = 0;
+			Mutex::Lock _l(_refreshQueue_m);
+			while (_refreshQueue.size() > 0) {
+				_Refresh &r = _refreshQueue.front();
+				if (_node)
+					_node->pushNetworkRefresh(r.dest,r.nwid,r.blacklistAddresses,r.blacklistThresholds,r.numBlacklistEntries);
+				_refreshQueue.pop_front();
+				if (++count >= 50)
+					break;
+			}
+		}
+
+		Thread::sleep(25);
+	}
+}
+
 NetworkController::ResultCode EmbeddedNetworkController::doNetworkConfigRequest(const InetAddress &fromAddr,const Identity &signingId,const Identity &identity,uint64_t nwid,const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData,NetworkConfig &nc)
 {
 	if (((!signingId)||(!signingId.hasPrivate()))||(signingId.address().toInt() != (nwid >> 24))) {
@@ -1082,8 +1149,19 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
 
 					_writeJson(_memberJP(nwid,Address(address),true).c_str(),member);
 
-					if (_node)
-						_node->pushNetworkRefresh(address,nwid,(const uint64_t *)0,(const uint64_t *)0,0);
+					{
+						Mutex::Lock _l(_networkMemberCache_m);
+						_networkMemberCache[nwid][Address(address)] = member;
+					}
+
+					{
+						Mutex::Lock _l(_refreshQueue_m);
+						_refreshQueue.push_back(_Refresh());
+						_Refresh &r = _refreshQueue.back();
+						r.dest = Address(address);
+						r.nwid = nwid;
+						r.numBlacklistEntries = 0;
+					}
 
 					// Add non-persisted fields
 					member["clock"] = now;
@@ -1478,24 +1556,9 @@ void EmbeddedNetworkController::_circuitTestCallback(ZT_Node *node,ZT_CircuitTes
 void EmbeddedNetworkController::_getNetworkMemberInfo(uint64_t now,uint64_t nwid,_NetworkMemberInfo &nmi)
 {
 	Mutex::Lock _mcl(_networkMemberCache_m);
-
 	auto memberCacheEntry = _networkMemberCache[nwid];
-	if ((now - memberCacheEntry.second) >= ZT_NETCONF_NETWORK_MEMBER_CACHE_EXPIRE) {
-		const std::string bp(_networkBP(nwid,false) + ZT_PATH_SEPARATOR_S + "member");
-		std::vector<std::string> members(OSUtils::listSubdirectories(bp.c_str()));
-		for(std::vector<std::string>::iterator m(members.begin());m!=members.end();++m) {
-			if (m->length() == ZT_ADDRESS_LENGTH_HEX) {
-				nlohmann::json mj(_readJson(bp + ZT_PATH_SEPARATOR_S + *m + ZT_PATH_SEPARATOR_S + "config.json"));
-				if ((mj.is_object())&&(mj.size() > 0)) {
-					memberCacheEntry.first[Address(*m)] = mj;
-				}
-			}
-		}
-		memberCacheEntry.second = now;
-	}
-
-	nmi.totalMemberCount = memberCacheEntry.first.size();
-	for(std::map< Address,nlohmann::json >::const_iterator nm(memberCacheEntry.first.begin());nm!=memberCacheEntry.first.end();++nm) {
+	nmi.totalMemberCount = memberCacheEntry.size();
+	for(std::map< Address,nlohmann::json >::const_iterator nm(memberCacheEntry.begin());nm!=memberCacheEntry.end();++nm) {
 		if (_jB(nm->second["authorized"],false)) {
 			++nmi.authorizedMemberCount;
 
diff --git a/controller/EmbeddedNetworkController.hpp b/controller/EmbeddedNetworkController.hpp
index 47419f52..b7a40a4c 100644
--- a/controller/EmbeddedNetworkController.hpp
+++ b/controller/EmbeddedNetworkController.hpp
@@ -25,21 +25,21 @@
 #include <map>
 #include <vector>
 #include <set>
+#include <list>
 
 #include "../node/Constants.hpp"
 
 #include "../node/NetworkController.hpp"
 #include "../node/Mutex.hpp"
 #include "../node/Utils.hpp"
+#include "../node/Address.hpp"
 #include "../node/InetAddress.hpp"
 
 #include "../osdep/OSUtils.hpp"
+#include "../osdep/Thread.hpp"
 
 #include "../ext/json/json.hpp"
 
-// Expiration time for network member cache entries in ms
-#define ZT_NETCONF_NETWORK_MEMBER_CACHE_EXPIRE 30000
-
 namespace ZeroTier {
 
 class Node;
@@ -50,6 +50,10 @@ public:
 	EmbeddedNetworkController(Node *node,const char *dbPath);
 	virtual ~EmbeddedNetworkController();
 
+	// Thread main method -- do not call directly
+	void threadMain()
+		throw();
+
 	virtual NetworkController::ResultCode doNetworkConfigRequest(
 		const InetAddress &fromAddr,
 		const Identity &signingId,
@@ -83,22 +87,6 @@ public:
 private:
 	static void _circuitTestCallback(ZT_Node *node,ZT_CircuitTest *test,const ZT_CircuitTestReport *report);
 
-	// JSON blob I/O
-	inline nlohmann::json _readJson(const std::string &path)
-	{
-		std::string buf;
-		if (OSUtils::readFile(path.c_str(),buf)) {
-			try {
-				return nlohmann::json::parse(buf);
-			} catch ( ... ) {}
-		}
-		return nlohmann::json::object();
-	}
-	inline bool _writeJson(const std::string &path,const nlohmann::json &obj)
-	{
-		return OSUtils::writeFile(path.c_str(),obj.dump(2));
-	}
-
 	// Network base path and network JSON path
 	inline std::string _networkBP(const uint64_t nwid,bool create)
 	{
@@ -133,8 +121,8 @@ private:
 		return (_memberBP(nwid,member,create) + ZT_PATH_SEPARATOR + "config.json");
 	}
 
-	// We cache the members of networks in memory to avoid having to scan the filesystem so much
-	std::map< uint64_t,std::pair< std::map< Address,nlohmann::json >,uint64_t > > _networkMemberCache;
+	// In-memory cache of network members
+	std::map< uint64_t,std::map< Address,nlohmann::json > > _networkMemberCache;
 	Mutex _networkMemberCache_m;
 
 	// Gathers a bunch of statistics about members of a network, IP assignments, etc. that we need in various places
@@ -211,6 +199,21 @@ private:
 	// Last request time by address, for rate limitation
 	std::map< std::pair<uint64_t,uint64_t>,uint64_t > _lastRequestTime;
 	Mutex _lastRequestTime_m;
+
+	// Queue of network member refreshes to be pushed
+	struct _Refresh
+	{
+		Address dest;
+		uint64_t nwid;
+		uint64_t blacklistAddresses[64];
+		uint64_t blacklistThresholds[64];
+		unsigned int numBlacklistEntries;
+	};
+	std::list< _Refresh > _refreshQueue;
+	Mutex _refreshQueue_m;
+
+	Thread _daemon;
+	volatile bool _daemonRun;
 };
 
 } // namespace ZeroTier