RethinkDB direct connectivity integration.

controller/EmbeddedNetworkController.cpp

@@ -36,32 +36,27 @@
 
 #include "../include/ZeroTierOne.h"
 #include "../version.h"
-#include "../node/Constants.hpp"
 
 #include "EmbeddedNetworkController.hpp"
 
 #include "../node/Node.hpp"
-#include "../node/Utils.hpp"
 #include "../node/CertificateOfMembership.hpp"
 #include "../node/NetworkConfig.hpp"
 #include "../node/Dictionary.hpp"
-#include "../node/InetAddress.hpp"
 #include "../node/MAC.hpp"
-#include "../node/Address.hpp"
 
 using json = nlohmann::json;
 
 // API version reported via JSON control plane
 #define ZT_NETCONF_CONTROLLER_API_VERSION 3
 
-// Number of requests to remember in member history
-#define ZT_NETCONF_DB_MEMBER_HISTORY_LENGTH 2
-
 // Min duration between requests for an address/nwid combo to prevent floods
 #define ZT_NETCONF_MIN_REQUEST_PERIOD 1000
 
 namespace ZeroTier {
 
+namespace {
+
 static json _renderRule(ZT_VirtualNetworkRule &rule)
 {
 	char tmp[128];
@@ -456,38 +451,32 @@ static bool _parseRule(json &r,ZT_VirtualNetworkRule &rule)
 	return false;
 }
 
+} // anonymous namespace
+
 EmbeddedNetworkController::EmbeddedNetworkController(Node *node,const char *dbPath) :
 	_startTime(OSUtils::now()),
-	_running(true),
-	_lastDumpedStatus(0),
-	_db(dbPath,this),
-	_node(node)
+	_node(node),
+	_path(dbPath),
+	_sender((NetworkController::Sender *)0),
+	_db(this,_signingId.address(),dbPath)
 {
-	if ((dbPath[0] == '-')&&(dbPath[1] == 0))
-		_startThreads(); // start threads now in Central harnessed mode
 }
 
 EmbeddedNetworkController::~EmbeddedNetworkController()
 {
-	std::vector<Thread> t;
-	{
-		Mutex::Lock _l(_threads_m);
-		_running = false;
-		t = _threads;
-	}
-	if (t.size() > 0) {
-		_queue.stop();
-		for(std::vector<Thread>::iterator i(t.begin());i!=t.end();++i)
-			Thread::join(*i);
-	}
+	std::lock_guard<std::mutex> l(_threads_l);
+	_queue.stop();
+	for(auto t=_threads.begin();t!=_threads.end();++t)
+		t->join();
 }
 
 void EmbeddedNetworkController::init(const Identity &signingId,Sender *sender)
 {
-	this->_sender = sender;
-	this->_signingId = signingId;
 	char tmp[64];
-	this->_signingIdAddressString = signingId.address().toString(tmp);
+	_signingId = signingId;
+	_sender = sender;
+	_signingIdAddressString = signingId.address().toString(tmp);
+	_db.waitForReady();
 }
 
 void EmbeddedNetworkController::request(
@@ -507,7 +496,7 @@ void EmbeddedNetworkController::request(
 	qe->identity = identity;
 	qe->metaData = metaData;
 	qe->type = _RQEntry::RQENTRY_TYPE_REQUEST;
-	_queue.post(qe);
+	_queue.post(std::unique_ptr<_RQEntry>(qe));
 }
 
 unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
@@ -523,7 +512,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
 		if ((path.size() >= 2)&&(path[1].length() == 16)) {
 			const uint64_t nwid = Utils::hexStrToU64(path[1].c_str());
 			json network;
-			if (!_db.getNetwork(nwid,network))
+			if (!_db.get(nwid,network))
 				return 404;
 
 			if (path.size() >= 3) {
@@ -535,7 +524,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
 
 						const uint64_t address = Utils::hexStrToU64(path[3].c_str());
 						json member;
-						if (!_db.getNetworkMember(nwid,address,member))
+						if (!_db.get(nwid,network,address,member))
 							return 404;
 						_addMemberNonPersistedFields(nwid,address,member,OSUtils::now());
 						responseBody = OSUtils::jsonDump(member);
@@ -545,14 +534,17 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
 						// List members and their revisions
 
 						responseBody = "{";
-						responseBody.reserve((_db.memberCount(nwid) + 1) * 32);
-						_db.eachMember(nwid,[&responseBody](uint64_t networkId,uint64_t nodeId,const json &member) {
-							if ((member.is_object())&&(member.size() > 0)) {
+						std::vector<json> members;
+						if (_db.get(nwid,network,members)) {
+							responseBody.reserve((members.size() + 2) * 32);
+							std::string mid;
+							for(auto member=members.begin();member!=members.end();++member) {
+								mid = (*member)["id"];
 								char tmp[128];
-								OSUtils::ztsnprintf(tmp,sizeof(tmp),"%s%.10llx\":%llu",(responseBody.length() > 1) ? ",\"" : "\"",(unsigned long long)nodeId,(unsigned long long)OSUtils::jsonInt(member["revision"],0));
+								OSUtils::ztsnprintf(tmp,sizeof(tmp),"%s\"%s\":%llu",(responseBody.length() > 1) ? ",\"" : "\"",mid.c_str(),(unsigned long long)OSUtils::jsonInt((*member)["revision"],0));
 								responseBody.append(tmp);
 							}
-						});
+						}
 						responseBody.push_back('}');
 						responseContentType = "application/json";
 
@@ -565,9 +557,9 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
 				// Get network
 
 				const int64_t now = OSUtils::now();
-				JSONDB::NetworkSummaryInfo ns;
-				_db.getNetworkSummaryInfo(nwid,ns);
-				_addNetworkNonPersistedFields(network,now,ns);
+				ControllerDB::NetworkSummaryInfo ns;
+				_db.summary(nwid,ns);
+				_addNetworkNonPersistedFields(nwid,network,now,ns);
 				responseBody = OSUtils::jsonDump(network);
 				responseContentType = "application/json";
 				return 200;
@@ -576,7 +568,8 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
 		} else if (path.size() == 1) {
 			// List networks
 
-			std::vector<uint64_t> networkIds(_db.networkIds());
+			std::vector<uint64_t> networkIds;
+			_db.networks(networkIds);
 			char tmp[64];
 			responseBody = "[";
 			responseBody.reserve((networkIds.size() + 1) * 24);
@@ -647,8 +640,8 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
 					char addrs[24];
 					OSUtils::ztsnprintf(addrs,sizeof(addrs),"%.10llx",(unsigned long long)address);
 
-					json member;
-					_db.getNetworkMember(nwid,address,member);
+					json member,network;
+					_db.get(nwid,network,address,member);
 					json origMember(member); // for detecting changes
 					_initMember(member);
 
@@ -674,10 +667,6 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
 									member["lastAuthorizedCredentialType"] = "api";
 									member["lastAuthorizedCredential"] = json();
 								}
-
-								// Member is being de-authorized, so spray Revocation objects to all online members
-								if (!newAuth)
-									onNetworkMemberDeauthorize(nwid,address);
 							}
 						}
 
@@ -743,8 +732,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
 					if (member != origMember) {
 						json &revj = member["revision"];
 						member["revision"] = (revj.is_number() ? ((uint64_t)revj + 1ULL) : 1ULL);
-						_db.saveNetworkMember(nwid,address,member);
-						onNetworkMemberUpdate(nwid,address);
+						_db.save(member);
 					}
 
 					_addMemberNonPersistedFields(nwid,address,member,now);
@@ -777,7 +765,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
 				OSUtils::ztsnprintf(nwids,sizeof(nwids),"%.16llx",(unsigned long long)nwid);
 
 				json network;
-				_db.getNetwork(nwid,network);
+				_db.get(nwid,network);
 				json origNetwork(network); // for detecting changes
 				_initNetwork(network);
 
@@ -996,13 +984,12 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
 				if (network != origNetwork) {
 					json &revj = network["revision"];
 					network["revision"] = (revj.is_number() ? ((uint64_t)revj + 1ULL) : 1ULL);
-					_db.saveNetwork(nwid,network);
-					onNetworkUpdate(nwid);
+					_db.save(network);
 				}
 
-				JSONDB::NetworkSummaryInfo ns;
-				_db.getNetworkSummaryInfo(nwid,ns);
-				_addNetworkNonPersistedFields(network,now,ns);
+				ControllerDB::NetworkSummaryInfo ns;
+				_db.summary(nwid,ns);
+				_addNetworkNonPersistedFields(nwid,network,now,ns);
 
 				responseBody = OSUtils::jsonDump(network);
 				responseContentType = "application/json";
@@ -1034,10 +1021,11 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpDELETE(
 				if ((path.size() == 4)&&(path[2] == "member")&&(path[3].length() == 10)) {
 					const uint64_t address = Utils::hexStrToU64(path[3].c_str());
 
-					json member = _db.eraseNetworkMember(nwid,address);
+					json network,member;
+					_db.get(nwid,network,address,member);
 
 					{
-						Mutex::Lock _l(_memberStatus_m);
+						std::lock_guard<std::mutex> l(_memberStatus_l);
 						_memberStatus.erase(_MemberStatusKey(nwid,address));
 					}
 
@@ -1048,10 +1036,12 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpDELETE(
 					return 200;
 				}
 			} else {
-				json network = _db.eraseNetwork(nwid);
+				json network;
+				_db.get(nwid,network);
+				_db.eraseNetwork(nwid);
 
 				{
-					Mutex::Lock _l(_memberStatus_m);
+					std::lock_guard<std::mutex> l(_memberStatus_l);
 					for(auto i=_memberStatus.begin();i!=_memberStatus.end();) {
 						if (i->first.networkId == nwid)
 							_memberStatus.erase(i++);
@@ -1074,17 +1064,12 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpDELETE(
 
 void EmbeddedNetworkController::handleRemoteTrace(const ZT_RemoteTrace &rt)
 {
+	/*
 	static volatile unsigned long idCounter = 0;
 	char id[128],tmp[128];
 	std::string k,v;
 
 	try {
-		std::vector<uint64_t> nw4m(_db.networksForMember(rt.origin));
-
-		// Ignore remote traces from members we don't know about
-		if (nw4m.empty())
-			return;
-
 		// Convert Dictionary into JSON object
 		json d;
 		char *saveptr = (char *)0;
@@ -1122,44 +1107,17 @@ void EmbeddedNetworkController::handleRemoteTrace(const ZT_RemoteTrace &rt)
 		d["objtype"] = "trace";
 		d["ts"] = now;
 		d["nodeId"] = Utils::hex10(rt.origin,tmp);
-
-		bool accept = true;
-		/*
-		for(std::vector<uint64_t>::const_iterator nwid(nw4m.begin());nwid!=nw4m.end();++nwid) {
-			json nconf;
-			if (_db.getNetwork(*nwid,nconf)) {
-				try {
-					if (OSUtils::jsonString(nconf["remoteTraceTarget"],"") == _signingIdAddressString) {
-						accept = true;
-						break;
-					}
-				} catch ( ... ) {} // ignore missing fields or other errors, drop trace message
-			}
-			if (_db.getNetworkMember(*nwid,rt.origin,nconf)) {
-				try {
-					if (OSUtils::jsonString(nconf["remoteTraceTarget"],"") == _signingIdAddressString) {
-						accept = true;
-						break;
-					}
-				} catch ( ... ) {} // ignore missing fields or other errors, drop trace message
-			}
-		}
-		*/
-		if (accept) {
-			char p[128];
-			OSUtils::ztsnprintf(p,sizeof(p),"trace/%s",id);
-			_db.writeRaw(p,OSUtils::jsonDump(d,-1));
-		}
 	} catch ( ... ) {
 		// drop invalid trace messages if an error occurs
 	}
+	*/
 }
 
 void EmbeddedNetworkController::onNetworkUpdate(const uint64_t networkId)
 {
 	// Send an update to all members of the network that are online
 	const int64_t now = OSUtils::now();
-	Mutex::Lock _l(_memberStatus_m);
+	std::lock_guard<std::mutex> l(_memberStatus_l);
 	for(auto i=_memberStatus.begin();i!=_memberStatus.end();++i) {
 		if ((i->first.networkId == networkId)&&(i->second.online(now))&&(i->second.lastRequestMetaData))
 			request(networkId,InetAddress(),0,i->second.identity,i->second.lastRequestMetaData);
@@ -1170,7 +1128,7 @@ void EmbeddedNetworkController::onNetworkMemberUpdate(const uint64_t networkId,c
 {
 	// Push update to member if online
 	try {
-		Mutex::Lock _l(_memberStatus_m);
+		std::lock_guard<std::mutex> l(_memberStatus_l);
 		_MemberStatus &ms = _memberStatus[_MemberStatusKey(networkId,memberId)];
 		if ((ms.online(OSUtils::now()))&&(ms.lastRequestMetaData))
 			request(networkId,InetAddress(),0,ms.identity,ms.lastRequestMetaData);
@@ -1183,7 +1141,7 @@ void EmbeddedNetworkController::onNetworkMemberDeauthorize(const uint64_t networ
 	Revocation rev((uint32_t)_node->prng(),networkId,0,now,ZT_REVOCATION_FLAG_FAST_PROPAGATE,Address(memberId),Revocation::CREDENTIAL_TYPE_COM);
 	rev.sign(_signingId);
 	{
-		Mutex::Lock _l(_memberStatus_m);
+		std::lock_guard<std::mutex> l(_memberStatus_l);
 		for(auto i=_memberStatus.begin();i!=_memberStatus.end();++i) {
 			if ((i->first.networkId == networkId)&&(i->second.online(now)))
 				_node->ncSendRevocation(Address(i->first.nodeId),rev);
@@ -1191,54 +1149,6 @@ void EmbeddedNetworkController::onNetworkMemberDeauthorize(const uint64_t networ
 	}
 }
 
-void EmbeddedNetworkController::threadMain()
-	throw()
-{
-	char tmp[256];
-	_RQEntry *qe = (_RQEntry *)0;
-	while (_running) {
-		const BlockingQueue<_RQEntry *>::TimedWaitResult wr = _queue.get(qe,1000);
-		if ((wr == BlockingQueue<_RQEntry *>::STOP)||(!_running))
-			break;
-
-		try {
-			if ((wr == BlockingQueue<_RQEntry *>::OK)&&(qe->type == _RQEntry::RQENTRY_TYPE_REQUEST)) {
-				_request(qe->nwid,qe->fromAddr,qe->requestPacketId,qe->identity,qe->metaData);
-				delete qe;
-			}
-
-			// Every 10s we update a 'status' containing member online state, etc.
-			const uint64_t now = OSUtils::now();
-			if ((now - _lastDumpedStatus) >= 10000) {
-				_lastDumpedStatus = now;
-				bool first = true;
-				OSUtils::ztsnprintf(tmp,sizeof(tmp),"{\"id\":\"%.10llx-status\",\"objtype\":\"status\",\"memberStatus\":[",_signingId.address().toInt());
-				std::string st(tmp);
-				{
-					Mutex::Lock _l(_memberStatus_m);
-					st.reserve(48 * (_memberStatus.size() + 1));
-					_db.eachId([this,&st,&now,&first,&tmp](uint64_t networkId,uint64_t nodeId) {
-						uint64_t lrt = 0ULL;
-						auto ms = this->_memberStatus.find(_MemberStatusKey(networkId,nodeId));
-						if (ms != _memberStatus.end())
-							lrt = ms->second.lastRequestTime;
-						OSUtils::ztsnprintf(tmp,sizeof(tmp),"%s\"%.16llx\",\"%.10llx\",%llu",
-							(first) ? "" : ",",
-							(unsigned long long)networkId,
-							(unsigned long long)nodeId,
-							(unsigned long long)lrt);
-						st.append(tmp);
-						first = false;
-					});
-				}
-				OSUtils::ztsnprintf(tmp,sizeof(tmp),"],\"clock\":%llu,\"startTime\":%llu,\"uptime\":%llu,\"vMajor\":%d,\"vMinor\":%d,\"vRev\":%d}",(unsigned long long)now,(unsigned long long)_startTime,(unsigned long long)(now - _startTime),ZEROTIER_ONE_VERSION_MAJOR,ZEROTIER_ONE_VERSION_MINOR,ZEROTIER_ONE_VERSION_REVISION);
-				st.append(tmp);
-				_db.writeRaw("status",st);
-			}
-		} catch ( ... ) {}
-	}
-}
-
 void EmbeddedNetworkController::_request(
 	uint64_t nwid,
 	const InetAddress &fromAddr,
@@ -1247,7 +1157,7 @@ void EmbeddedNetworkController::_request(
 	const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData)
 {
 	char nwids[24];
-	JSONDB::NetworkSummaryInfo ns;
+	ControllerDB::NetworkSummaryInfo ns;
 	json network,member,origMember;
 
 	if (((!_signingId)||(!_signingId.hasPrivate()))||(_signingId.address().toInt() != (nwid >> 24))||(!_sender))
@@ -1256,7 +1166,7 @@ void EmbeddedNetworkController::_request(
 	const int64_t now = OSUtils::now();
 
 	if (requestPacketId) {
-		Mutex::Lock _l(_memberStatus_m);
+		std::lock_guard<std::mutex> l(_memberStatus_l);
 		_MemberStatus &ms = _memberStatus[_MemberStatusKey(nwid,identity.address().toInt())];
 		if ((now - ms.lastRequestTime) <= ZT_NETCONF_MIN_REQUEST_PERIOD)
 			return;
@@ -1264,7 +1174,7 @@ void EmbeddedNetworkController::_request(
 	}
 
 	OSUtils::ztsnprintf(nwids,sizeof(nwids),"%.16llx",nwid);
-	if (!_db.getNetworkAndMember(nwid,identity.address().toInt(),network,member,ns)) {
+	if (!_db.get(nwid,network,identity.address().toInt(),member,ns)) {
 		_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_OBJECT_NOT_FOUND);
 		return;
 	}
@@ -1355,7 +1265,7 @@ void EmbeddedNetworkController::_request(
 			member["vProto"] = vProto;
 
 			{
-				Mutex::Lock _l(_memberStatus_m);
+				std::lock_guard<std::mutex> l(_memberStatus_l);
 				_MemberStatus &ms = _memberStatus[_MemberStatusKey(nwid,identity.address().toInt())];
 
 				ms.vMajor = (int)vMajor;
@@ -1379,7 +1289,7 @@ void EmbeddedNetworkController::_request(
 		if (origMember != member) {
 			json &revj = member["revision"];
 			member["revision"] = (revj.is_number() ? ((uint64_t)revj + 1ULL) : 1ULL);
-			_db.saveNetworkMember(nwid,identity.address().toInt(),member);
+			_db.save(member);
 		}
 		_sender->ncSendError(nwid,requestPacketId,identity.address(),NetworkController::NC_ERROR_ACCESS_DENIED);
 		return;
@@ -1746,23 +1656,84 @@ void EmbeddedNetworkController::_request(
 	if (member != origMember) {
 		json &revj = member["revision"];
 		member["revision"] = (revj.is_number() ? ((uint64_t)revj + 1ULL) : 1ULL);
-		_db.saveNetworkMember(nwid,identity.address().toInt(),member);
+		_db.save(member);
 	}
 
 	_sender->ncSendConfig(nwid,requestPacketId,identity.address(),*(nc.get()),metaData.getUI(ZT_NETWORKCONFIG_REQUEST_METADATA_KEY_VERSION,0) < 6);
 }
 
+/*
+void EmbeddedNetworkController::threadMain()
+	throw()
+{
+	char tmp[256];
+	_RQEntry *qe = (_RQEntry *)0;
+	while (_running) {
+		const BlockingQueue<_RQEntry *>::TimedWaitResult wr = _queue.get(qe,1000);
+		if ((wr == BlockingQueue<_RQEntry *>::STOP)||(!_running))
+			break;
+
+		try {
+			if ((wr == BlockingQueue<_RQEntry *>::OK)&&(qe->type == _RQEntry::RQENTRY_TYPE_REQUEST)) {
+				_request(qe->nwid,qe->fromAddr,qe->requestPacketId,qe->identity,qe->metaData);
+				delete qe;
+			}
+
+			// Every 10s we update a 'status' containing member online state, etc.
+			const uint64_t now = OSUtils::now();
+			if ((now - _lastDumpedStatus) >= 10000) {
+				_lastDumpedStatus = now;
+				bool first = true;
+				OSUtils::ztsnprintf(tmp,sizeof(tmp),"{\"id\":\"%.10llx-status\",\"objtype\":\"status\",\"memberStatus\":[",_signingId.address().toInt());
+				std::string st(tmp);
+				{
+					Mutex::Lock _l(_memberStatus_m);
+					st.reserve(48 * (_memberStatus.size() + 1));
+					_db.eachId([this,&st,&now,&first,&tmp](uint64_t networkId,uint64_t nodeId) {
+						uint64_t lrt = 0ULL;
+						auto ms = this->_memberStatus.find(_MemberStatusKey(networkId,nodeId));
+						if (ms != _memberStatus.end())
+							lrt = ms->second.lastRequestTime;
+						OSUtils::ztsnprintf(tmp,sizeof(tmp),"%s\"%.16llx\",\"%.10llx\",%llu",
+							(first) ? "" : ",",
+							(unsigned long long)networkId,
+							(unsigned long long)nodeId,
+							(unsigned long long)lrt);
+						st.append(tmp);
+						first = false;
+					});
+				}
+				OSUtils::ztsnprintf(tmp,sizeof(tmp),"],\"clock\":%llu,\"startTime\":%llu,\"uptime\":%llu,\"vMajor\":%d,\"vMinor\":%d,\"vRev\":%d}",(unsigned long long)now,(unsigned long long)_startTime,(unsigned long long)(now - _startTime),ZEROTIER_ONE_VERSION_MAJOR,ZEROTIER_ONE_VERSION_MINOR,ZEROTIER_ONE_VERSION_REVISION);
+				st.append(tmp);
+				_db.writeRaw("status",st);
+			}
+		} catch ( ... ) {}
+	}
+}
+*/
+
 void EmbeddedNetworkController::_startThreads()
 {
-	Mutex::Lock _l(_threads_m);
-	if (_threads.size() == 0) {
-		long hwc = (long)std::thread::hardware_concurrency();
-		if (hwc < 1)
-			hwc = 1;
-		else if (hwc > 16)
-			hwc = 16;
-		for(long i=0;i<hwc;++i)
-			_threads.push_back(Thread::start(this));
+	std::lock_guard<std::mutex> l(_threads_l);
+	if (!_threads.empty())
+		return;
+	const long hwc = std::max((long)std::thread::hardware_concurrency(),(long)1);
+	for(long t=0;t<hwc;++t) {
+		_threads.emplace_back([this]() {
+			for(;;) {
+				std::unique_ptr<_RQEntry> qe;
+				if (_queue.get(qe))
+					break;
+				try {
+					if (qe)
+						_request(qe->nwid,qe->fromAddr,qe->requestPacketId,qe->identity,qe->metaData);
+				} catch (std::exception &e) {
+					fprintf(stderr,"ERROR: exception in controller request handling thread: %s" ZT_EOL_S,e.what());
+				} catch ( ... ) {
+					fprintf(stderr,"ERROR: exception in controller request handling thread: unknown exception" ZT_EOL_S);
+				}
+			}
+		});
 	}
 }
 

controller/EmbeddedNetworkController.hpp

@@ -19,6 +19,8 @@
 #ifndef ZT_SQLITENETWORKCONTROLLER_HPP
 #define ZT_SQLITENETWORKCONTROLLER_HPP
 
+#define ZT_CONTROLLER_USE_RETHINKDB
+
 #include <stdint.h>
 
 #include <string>
@@ -30,9 +32,7 @@
 #include <unordered_map>
 
 #include "../node/Constants.hpp"
-
 #include "../node/NetworkController.hpp"
-#include "../node/Mutex.hpp"
 #include "../node/Utils.hpp"
 #include "../node/Address.hpp"
 #include "../node/InetAddress.hpp"
@@ -44,18 +44,23 @@
 
 #include "../ext/json/json.hpp"
 
-#include "JSONDB.hpp"
+#ifdef ZT_CONTROLLER_USE_RETHINKDB
+#include "RethinkDB.hpp"
+#endif
 
 namespace ZeroTier {
 
+#ifdef ZT_CONTROLLER_USE_RETHINKDB
+typedef RethinkDB ControllerDB;
+#endif
 class Node;
 
-class EmbeddedNetworkController : public NetworkController,NonCopyable
+class EmbeddedNetworkController : public NetworkController
 {
 public:
 	/**
 	 * @param node Parent node
-	 * @param dbPath Path to store data
+	 * @param dbPath Database path (file path or database credentials)
 	 */
 	EmbeddedNetworkController(Node *node,const char *dbPath);
 	virtual ~EmbeddedNetworkController();
@@ -98,22 +103,7 @@ public:
 	void onNetworkMemberUpdate(const uint64_t networkId,const uint64_t memberId);
 	void onNetworkMemberDeauthorize(const uint64_t networkId,const uint64_t memberId);
 
-	void threadMain()
-		throw();
-
 private:
-	struct _RQEntry
-	{
-		uint64_t nwid;
-		uint64_t requestPacketId;
-		InetAddress fromAddr;
-		Identity identity;
-		Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> metaData;
-		enum {
-			RQENTRY_TYPE_REQUEST = 0
-		} type;
-	};
-
 	void _request(uint64_t nwid,const InetAddress &fromAddr,uint64_t requestPacketId,const Identity &identity,const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData);
 	void _startThreads();
 
@@ -166,12 +156,20 @@ private:
 		}
 		network["objtype"] = "network";
 	}
-	inline void _addNetworkNonPersistedFields(nlohmann::json &network,int64_t now,const JSONDB::NetworkSummaryInfo &ns)
+	inline void _addNetworkNonPersistedFields(const uint64_t nwid,nlohmann::json &network,int64_t now,const ControllerDB::NetworkSummaryInfo &ns)
 	{
 		network["clock"] = now;
 		network["authorizedMemberCount"] = ns.authorizedMemberCount;
-		network["activeMemberCount"] = ns.activeMemberCount;
 		network["totalMemberCount"] = ns.totalMemberCount;
+		{
+			std::lock_guard<std::mutex> l(_memberStatus_l);
+			unsigned long ac = 0;
+			for(auto ms=_memberStatus.begin();ms!=_memberStatus.end();++ms) {
+				if ((ms->first.networkId == nwid)&&(ms->second.online(now)))
+					++ac;
+			}
+			network["activeMemberCount"] = ac;
+		}
 	}
 	inline void _removeNetworkNonPersistedFields(nlohmann::json &network)
 	{
@@ -185,8 +183,10 @@ private:
 	inline void _addMemberNonPersistedFields(uint64_t nwid,uint64_t nodeId,nlohmann::json &member,int64_t now)
 	{
 		member["clock"] = now;
-		Mutex::Lock _l(_memberStatus_m);
-		member["online"] = _memberStatus[_MemberStatusKey(nwid,nodeId)].online(now);
+		{
+			std::lock_guard<std::mutex> l(_memberStatus_l);
+			member["online"] = _memberStatus[_MemberStatusKey(nwid,nodeId)].online(now);
+		}
 	}
 	inline void _removeMemberNonPersistedFields(nlohmann::json &member)
 	{
@@ -197,23 +197,17 @@ private:
 		member.erase("lastRequestMetaData");
 	}
 
-	const int64_t _startTime;
-
-	volatile bool _running;
-	BlockingQueue<_RQEntry *> _queue;
-	std::vector<Thread> _threads;
-	volatile uint64_t _lastDumpedStatus;
-	Mutex _threads_m;
-
-	JSONDB _db;
-
-	Node *const _node;
-	std::string _path;
-
-	NetworkController::Sender *_sender;
-	Identity _signingId;
-	std::string _signingIdAddressString;
-
+	struct _RQEntry
+	{
+		uint64_t nwid;
+		uint64_t requestPacketId;
+		InetAddress fromAddr;
+		Identity identity;
+		Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> metaData;
+		enum {
+			RQENTRY_TYPE_REQUEST = 0
+		} type;
+	};
 	struct _MemberStatusKey
 	{
 		_MemberStatusKey() : networkId(0),nodeId(0) {}
@@ -239,8 +233,19 @@ private:
 			return (std::size_t)(networkIdNodeId.networkId + networkIdNodeId.nodeId);
 		}
 	};
+
+	const int64_t _startTime;
+	Node *const _node;
+	std::string _path;
+	Identity _signingId;
+	std::string _signingIdAddressString;
+	NetworkController::Sender *_sender;
+	ControllerDB _db;
+	BlockingQueue< std::unique_ptr<_RQEntry> > _queue;
+	std::vector<std::thread> _threads;
+	std::mutex _threads_l;
 	std::unordered_map< _MemberStatusKey,_MemberStatus,_MemberStatusHash > _memberStatus;
-	Mutex _memberStatus_m;
+	std::mutex _memberStatus_l;
 };
 
 } // namespace ZeroTier

controller/RethinkDB.cpp

@@ -1,4 +1,25 @@
+/*
+ * ZeroTier One - Network Virtualization Everywhere
+ * Copyright (C) 2011-2015  ZeroTier, Inc.
+ *
+ * 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/>.
+ */
+
+#ifdef ZT_CONTROLLER_USE_RETHINKDB
+
 #include "RethinkDB.hpp"
+#include "EmbeddedNetworkController.hpp"
 
 #include <chrono>
 #include <algorithm>
@@ -7,19 +28,25 @@
 #include "../ext/librethinkdbxx/build/include/rethinkdb.h"
 
 namespace R = RethinkDB;
-using nlohmann::json;
+using json = nlohmann::json;
 
 namespace ZeroTier {
 
-RethinkDB::RethinkDB(const Address &myAddress,const char *host,const int port,const char *db,const char *auth) :
+RethinkDB::RethinkDB(EmbeddedNetworkController *const nc,const Address &myAddress,const char *path) :
+	_controller(nc),
 	_myAddress(myAddress),
-	_host(host ? host : "127.0.0.1"),
-	_db(db),
-	_auth(auth ? auth : ""),
-	_port((port > 0) ? port : 28015),
-	_ready(2), // two tables need to be synchronized before we're ready
-	_run(1)
+	_ready(2), // two tables need to be synchronized before we're ready, so this is ready when it reaches 0
+	_run(1),
+	_waitNoticePrinted(false)
 {
+	std::vector<std::string> ps(OSUtils::split(path,":","",""));
+	if ((ps.size() != 5)||(ps[0] != "rethinkdb"))
+		throw std::runtime_error("invalid rethinkdb database url");
+	_host = ps[1];
+	_db = ps[2];
+	_auth = ps[3];
+	_port = Utils::strToInt(ps[4].c_str());
+
 	_readyLock.lock();
 
 	{
@@ -31,29 +58,35 @@ RethinkDB::RethinkDB(const Address &myAddress,const char *host,const int port,co
 	_membersDbWatcher = std::thread([this]() {
 		while (_run == 1) {
 			try {
-				auto rdb = R::connect(this->_host,this->_port,this->_auth);
+				std::unique_ptr<R::Connection> rdb(R::connect(this->_host,this->_port,this->_auth));
 				if (rdb) {
 					_membersDbWatcherConnection = (void *)rdb.get();
-					auto cur = R::db(this->_db).table("Member").get_all(this->_myAddressStr,R::optargs("index","controllerId")).changes(R::optargs("squash",0.1,"include_initial",true,"include_types",true,"include_states",true)).run(*rdb);
+					auto cur = R::db(this->_db).table("Member",R::optargs("read_mode","outdated")).get_all(this->_myAddressStr,R::optargs("index","controllerId")).changes(R::optargs("squash",0.05,"include_initial",true,"include_types",true,"include_states",true)).run(*rdb);
 					while (cur.has_next()) {
 						if (_run != 1) break;
 						json tmp(json::parse(cur.next().as_json()));
 						if ((tmp["type"] == "state")&&(tmp["state"] == "ready")) {
-							if (--this->_ready == 0)
+							if (--this->_ready == 0) {
+								if (_waitNoticePrinted)
+									fprintf(stderr,"NOTICE: controller RethinkDB data download complete." ZT_EOL_S);
 								this->_readyLock.unlock();
+							}
 						} else {
 							try {
-								this->_memberChanged(tmp["old_val"],tmp["new_val"]);
+								json &ov = tmp["old_val"];
+								json &nv = tmp["new_val"];
+								if (ov.is_object()||nv.is_object())
+									this->_memberChanged(ov,nv);
 							} catch ( ... ) {} // ignore bad records
 						}
 					}
 				}
 			} catch (std::exception &e) {
-				fprintf(stderr,"ERROR: controller RethinkDB: %s" ZT_EOL_S,e.what());
+				fprintf(stderr,"ERROR: controller RethinkDB (member change stream): %s" ZT_EOL_S,e.what());
 			} catch (R::Error &e) {
-				fprintf(stderr,"ERROR: controller RethinkDB: %s" ZT_EOL_S,e.message.c_str());
+				fprintf(stderr,"ERROR: controller RethinkDB (member change stream): %s" ZT_EOL_S,e.message.c_str());
 			} catch ( ... ) {
-				fprintf(stderr,"ERROR: controller RethinkDB: unknown exception" ZT_EOL_S);
+				fprintf(stderr,"ERROR: controller RethinkDB (member change stream): unknown exception" ZT_EOL_S);
 			}
 			std::this_thread::sleep_for(std::chrono::milliseconds(250));
 		}
@@ -62,50 +95,131 @@ RethinkDB::RethinkDB(const Address &myAddress,const char *host,const int port,co
 	_networksDbWatcher = std::thread([this]() {
 		while (_run == 1) {
 			try {
-				auto rdb = R::connect(this->_host,this->_port,this->_auth);
+				std::unique_ptr<R::Connection> rdb(R::connect(this->_host,this->_port,this->_auth));
 				if (rdb) {
 					_membersDbWatcherConnection = (void *)rdb.get();
-					auto cur = R::db(this->_db).table("Network").get_all(this->_myAddressStr,R::optargs("index","controllerId")).changes(R::optargs("squash",0.1,"include_initial",true,"include_types",true,"include_states",true)).run(*rdb);
+					auto cur = R::db(this->_db).table("Network",R::optargs("read_mode","outdated")).get_all(this->_myAddressStr,R::optargs("index","controllerId")).changes(R::optargs("squash",0.05,"include_initial",true,"include_types",true,"include_states",true)).run(*rdb);
 					while (cur.has_next()) {
 						if (_run != 1) break;
 						json tmp(json::parse(cur.next().as_json()));
 						if ((tmp["type"] == "state")&&(tmp["state"] == "ready")) {
-							if (--this->_ready == 0)
+							if (--this->_ready == 0) {
+								if (_waitNoticePrinted)
+									fprintf(stderr,"NOTICE: controller RethinkDB data download complete." ZT_EOL_S);
 								this->_readyLock.unlock();
+							}
 						} else {
 							try {
-								this->_networkChanged(tmp["old_val"],tmp["new_val"]);
+								json &ov = tmp["old_val"];
+								json &nv = tmp["new_val"];
+								if (ov.is_object()||nv.is_object())
+									this->_networkChanged(ov,nv);
 							} catch ( ... ) {} // ignore bad records
 						}
 					}
 				}
 			} catch (std::exception &e) {
-				fprintf(stderr,"ERROR: controller RethinkDB: %s" ZT_EOL_S,e.what());
+				fprintf(stderr,"ERROR: controller RethinkDB (network change stream): %s" ZT_EOL_S,e.what());
 			} catch (R::Error &e) {
-				fprintf(stderr,"ERROR: controller RethinkDB: %s" ZT_EOL_S,e.message.c_str());
+				fprintf(stderr,"ERROR: controller RethinkDB (network change stream): %s" ZT_EOL_S,e.message.c_str());
 			} catch ( ... ) {
-				fprintf(stderr,"ERROR: controller RethinkDB: unknown exception" ZT_EOL_S);
+				fprintf(stderr,"ERROR: controller RethinkDB (network change stream): unknown exception" ZT_EOL_S);
 			}
 			std::this_thread::sleep_for(std::chrono::milliseconds(250));
 		}
 	});
+
+	for(int t=0;t<ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS;++t) {
+		_commitThread[t] = std::thread([this]() {
+			std::unique_ptr<R::Connection> rdb;
+			std::unique_ptr<nlohmann::json> config;
+			while ((this->_commitQueue.get(config))&&(_run == 1)) {
+				if (!config)
+					continue;
+				json record;
+				const std::string objtype = (*config)["objtype"];
+				const char *table;
+				std::string deleteId;
+				if (objtype == "member") {
+					const std::string nwid = (*config)["nwid"];
+					const std::string id = (*config)["id"];
+					record["id"] = nwid + "-" + id;
+					record["controllerId"] = this->_myAddressStr;
+					record["networkId"] = nwid;
+					record["nodeId"] = id;
+					record["config"] = *config;
+					table = "Member";
+				} else if (objtype == "network") {
+					const std::string id = (*config)["id"];
+					record["id"] = id;
+					record["controllerId"] = this->_myAddressStr;
+					record["config"] = *config;
+					table = "Network";
+				} else if (objtype == "delete_network") {
+					deleteId = (*config)["id"];
+					table = "Network";
+				} else if (objtype == "delete_member") {
+					deleteId = (*config)["nwid"];
+					deleteId.push_back('-');
+					const std::string tmp = (*config)["id"];
+					deleteId.append(tmp);
+					table = "Member";
+				} else {
+					continue;
+				}
+
+				while (_run == 1) {
+					try {
+						if (!rdb)
+							rdb = R::connect(this->_host,this->_port,this->_auth);
+						if (rdb) {
+							if (deleteId.length() > 0) {
+								R::db(this->_db).table(table).get(deleteId).delete_().run(*rdb);
+							} else {
+								R::db(this->_db).table(table).insert(record.dump(),R::optargs("conflict","update","return_changes",false)).run(*rdb);
+							}
+							break;
+						} else {
+							fprintf(stderr,"ERROR: controller RethinkDB (insert/update): connect failed (will retry)" ZT_EOL_S);
+						}
+					} catch (std::exception &e) {
+						fprintf(stderr,"ERROR: controller RethinkDB (insert/update): %s" ZT_EOL_S,e.what());
+						rdb.reset();
+					} catch (R::Error &e) {
+						fprintf(stderr,"ERROR: controller RethinkDB (insert/update): %s" ZT_EOL_S,e.message.c_str());
+						rdb.reset();
+					} catch ( ... ) {
+						fprintf(stderr,"ERROR: controller RethinkDB (insert/update): unknown exception" ZT_EOL_S);
+						rdb.reset();
+					}
+					std::this_thread::sleep_for(std::chrono::milliseconds(250));
+				}
+			}
+		});
+	}
 }
 
 RethinkDB::~RethinkDB()
 {
-	// FIXME: not totally safe but will generally work, and only happens on shutdown anyway
+	// FIXME: not totally safe but will generally work, and only happens on shutdown anyway.
+	// Would need to add some kind of 'whack it' support to librethinkdbxx to do better.
 	_run = 0;
-	std::this_thread::sleep_for(std::chrono::milliseconds(10));
+	std::this_thread::sleep_for(std::chrono::milliseconds(100));
 	if (_membersDbWatcherConnection)
 		((R::Connection *)_membersDbWatcherConnection)->close();
 	if (_networksDbWatcherConnection)
 		((R::Connection *)_networksDbWatcherConnection)->close();
+	_commitQueue.stop();
+	for(int t=0;t<ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS;++t)
+		_commitThread[t].join();
 	_membersDbWatcher.join();
 	_networksDbWatcher.join();
 }
 
 inline bool RethinkDB::get(const uint64_t networkId,nlohmann::json &network)
 {
+	waitForReady();
+
 	std::shared_ptr<_Network> nw;
 	{
 		std::lock_guard<std::mutex> l(_networks_l);
@@ -121,8 +235,33 @@ inline bool RethinkDB::get(const uint64_t networkId,nlohmann::json &network)
 	return true;
 }
 
+inline bool RethinkDB::get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member)
+{
+	waitForReady();
+
+	std::shared_ptr<_Network> nw;
+	{
+		std::lock_guard<std::mutex> l(_networks_l);
+		auto nwi = _networks.find(networkId);
+		if (nwi == _networks.end())
+			return false;
+		nw = nwi->second;
+	}
+
+	std::lock_guard<std::mutex> l2(nw->lock);
+	auto m = nw->members.find(memberId);
+	if (m == nw->members.end())
+		return false;
+	network = nw->config;
+	member = m->second;
+
+	return true;
+}
+
 inline bool RethinkDB::get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member,NetworkSummaryInfo &info)
 {
+	waitForReady();
+
 	std::shared_ptr<_Network> nw;
 	{
 		std::lock_guard<std::mutex> l(_networks_l);
@@ -133,7 +272,7 @@ inline bool RethinkDB::get(const uint64_t networkId,nlohmann::json &network,cons
 	}
 
 	std::lock_guard<std::mutex> l2(nw->lock);
-		auto m = nw->members.find(memberId);
+	auto m = nw->members.find(memberId);
 	if (m == nw->members.end())
 		return false;
 	network = nw->config;
@@ -145,6 +284,8 @@ inline bool RethinkDB::get(const uint64_t networkId,nlohmann::json &network,cons
 
 inline bool RethinkDB::get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohmann::json> &members)
 {
+	waitForReady();
+
 	std::shared_ptr<_Network> nw;
 	{
 		std::lock_guard<std::mutex> l(_networks_l);
@@ -164,6 +305,8 @@ inline bool RethinkDB::get(const uint64_t networkId,nlohmann::json &network,std:
 
 inline bool RethinkDB::summary(const uint64_t networkId,NetworkSummaryInfo &info)
 {
+	waitForReady();
+
 	std::shared_ptr<_Network> nw;
 	{
 		std::lock_guard<std::mutex> l(_networks_l);
@@ -179,10 +322,51 @@ inline bool RethinkDB::summary(const uint64_t networkId,NetworkSummaryInfo &info
 	return true;
 }
 
+void RethinkDB::networks(std::vector<uint64_t> &networks)
+{
+	waitForReady();
+	std::lock_guard<std::mutex> l(_networks_l);
+	networks.reserve(_networks.size() + 1);
+	for(auto n=_networks.begin();n!=_networks.end();++n)
+		networks.push_back(n->first);
+}
+
+void RethinkDB::save(const nlohmann::json &record)
+{
+	waitForReady();
+	_commitQueue.post(std::unique_ptr<nlohmann::json>(new nlohmann::json(record)));
+}
+
+void RethinkDB::eraseNetwork(const uint64_t networkId)
+{
+	char tmp2[24];
+	waitForReady();
+	Utils::hex(networkId,tmp2);
+	json tmp;
+	tmp["id"] = tmp2;
+	tmp["objtype"] = "delete_network"; // pseudo-type, tells thread to delete network
+	_commitQueue.post(std::unique_ptr<nlohmann::json>(new nlohmann::json(tmp)));
+}
+
+void RethinkDB::eraseMember(const uint64_t networkId,const uint64_t memberId)
+{
+	char tmp2[24];
+	json tmp;
+	waitForReady();
+	Utils::hex(networkId,tmp2);
+	tmp["nwid"] = tmp2;
+	Utils::hex10(memberId,tmp2);
+	tmp["id"] = tmp2;
+	tmp["objtype"] = "delete_member"; // pseudo-type, tells thread to delete network
+	_commitQueue.post(std::unique_ptr<nlohmann::json>(new nlohmann::json(tmp)));
+}
+
 void RethinkDB::_memberChanged(nlohmann::json &old,nlohmann::json &member)
 {
 	uint64_t memberId = 0;
 	uint64_t networkId = 0;
+	bool isAuth = false;
+	bool wasAuth = false;
 	std::shared_ptr<_Network> nw;
 
 	if (old.is_object()) {
@@ -201,7 +385,8 @@ void RethinkDB::_memberChanged(nlohmann::json &old,nlohmann::json &member)
 					std::lock_guard<std::mutex> l(nw->lock);
 					if (OSUtils::jsonBool(config["activeBridge"],false))
 						nw->activeBridgeMembers.erase(memberId);
-					if (OSUtils::jsonBool(config["authorized"],false))
+					wasAuth = OSUtils::jsonBool(config["authorized"],false);
+					if (wasAuth)
 						nw->authorizedMembers.erase(memberId);
 					json &ips = config["ipAssignments"];
 					if (ips.is_array()) {
@@ -234,35 +419,43 @@ void RethinkDB::_memberChanged(nlohmann::json &old,nlohmann::json &member)
 					nw2.reset(new _Network);
 				nw = nw2;
 			}
-			std::lock_guard<std::mutex> l(nw->lock);
-
-			nw->members[memberId] = config;
-
-			if (OSUtils::jsonBool(config["activeBridge"],false))
-				nw->activeBridgeMembers.insert(memberId);
-			const bool isAuth = OSUtils::jsonBool(config["authorized"],false);
-			if (isAuth)
-				nw->authorizedMembers.insert(memberId);
-			json &ips = config["ipAssignments"];
-			if (ips.is_array()) {
-				for(unsigned long i=0;i<ips.size();++i) {
-					json &ipj = ips[i];
-					if (ipj.is_string()) {
-						const std::string ips = ipj;
-						InetAddress ipa(ips.c_str());
-						ipa.setPort(0);
-						nw->allocatedIps.insert(ipa);
+
+			{
+				std::lock_guard<std::mutex> l(nw->lock);
+
+				nw->members[memberId] = config;
+
+				if (OSUtils::jsonBool(config["activeBridge"],false))
+					nw->activeBridgeMembers.insert(memberId);
+				isAuth = OSUtils::jsonBool(config["authorized"],false);
+				if (isAuth)
+					nw->authorizedMembers.insert(memberId);
+				json &ips = config["ipAssignments"];
+				if (ips.is_array()) {
+					for(unsigned long i=0;i<ips.size();++i) {
+						json &ipj = ips[i];
+						if (ipj.is_string()) {
+							const std::string ips = ipj;
+							InetAddress ipa(ips.c_str());
+							ipa.setPort(0);
+							nw->allocatedIps.insert(ipa);
+						}
 					}
 				}
-			}
 
-			if (!isAuth) {
-				const int64_t ldt = (int64_t)OSUtils::jsonInt(config["lastDeauthorizedTime"],0ULL);
-				if (ldt > nw->mostRecentDeauthTime)
-					nw->mostRecentDeauthTime = ldt;
+				if (!isAuth) {
+					const int64_t ldt = (int64_t)OSUtils::jsonInt(config["lastDeauthorizedTime"],0ULL);
+					if (ldt > nw->mostRecentDeauthTime)
+						nw->mostRecentDeauthTime = ldt;
+				}
 			}
+
+			_controller->onNetworkMemberUpdate(networkId,memberId);
 		}
 	}
+
+	if ((wasAuth)&&(!isAuth)&&(networkId)&&(memberId))
+		_controller->onNetworkMemberDeauthorize(networkId,memberId);
 }
 
 void RethinkDB::_networkChanged(nlohmann::json &old,nlohmann::json &network)
@@ -281,8 +474,11 @@ void RethinkDB::_networkChanged(nlohmann::json &old,nlohmann::json &network)
 						nw2.reset(new _Network);
 					nw = nw2;
 				}
-				std::lock_guard<std::mutex> l2(nw->lock);
-				nw->config = config;
+				{
+					std::lock_guard<std::mutex> l2(nw->lock);
+					nw->config = config;
+				}
+				_controller->onNetworkUpdate(id);
 			}
 		}
 	} else if (old.is_object()) {
@@ -306,3 +502,5 @@ int main(int argc,char **argv)
 	pause();
 }
 */
+
+#endif // ZT_CONTROLLER_USE_RETHINKDB

controller/RethinkDB.hpp

@@ -1,3 +1,23 @@
+/*
+ * ZeroTier One - Network Virtualization Everywhere
+ * Copyright (C) 2011-2015  ZeroTier, Inc.
+ *
+ * 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/>.
+ */
+
+#ifdef ZT_CONTROLLER_USE_RETHINKDB
+
 #ifndef ZT_CONTROLLER_RETHINKDB_HPP
 #define ZT_CONTROLLER_RETHINKDB_HPP
 
@@ -5,6 +25,7 @@
 #include "../node/Address.hpp"
 #include "../node/InetAddress.hpp"
 #include "../osdep/OSUtils.hpp"
+#include "../osdep/BlockingQueue.hpp"
 
 #include <memory>
 #include <string>
@@ -15,9 +36,13 @@
 
 #include "../ext/json/json.hpp"
 
+#define ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS 2
+
 namespace ZeroTier
 {
 
+class EmbeddedNetworkController;
+
 class RethinkDB
 {
 public:
@@ -31,24 +56,41 @@ public:
 		int64_t mostRecentDeauthTime;
 	};
 
-	RethinkDB(const Address &myAddress,const char *host,const int port,const char *db,const char *auth);
+	RethinkDB(EmbeddedNetworkController *const nc,const Address &myAddress,const char *path);
 	~RethinkDB();
 
-	inline bool ready() const { return (_ready <= 0); }
-
 	inline void waitForReady() const
 	{
 		while (_ready > 0) {
+			if (!_waitNoticePrinted) {
+				_waitNoticePrinted = true;
+				fprintf(stderr,"NOTICE: controller RethinkDB waiting for initial data download..." ZT_EOL_S);
+			}
 			_readyLock.lock();
 			_readyLock.unlock();
 		}
 	}
 
+	inline bool hasNetwork(const uint64_t networkId) const
+	{
+		std::lock_guard<std::mutex> l(_networks_l);
+		return (_networks.find(networkId) != _networks.end());
+	}
+
 	bool get(const uint64_t networkId,nlohmann::json &network);
+	bool get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member);
 	bool get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member,NetworkSummaryInfo &info);
 	bool get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohmann::json> &members);
+
 	bool summary(const uint64_t networkId,NetworkSummaryInfo &info);
 
+	void networks(std::vector<uint64_t> &networks);
+
+	void save(const nlohmann::json &record);
+
+	void eraseNetwork(const uint64_t networkId);
+	void eraseMember(const uint64_t networkId,const uint64_t memberId);
+
 private:
 	struct _Network
 	{
@@ -76,12 +118,13 @@ private:
 		info.mostRecentDeauthTime = nw->mostRecentDeauthTime;
 	}
 
+	EmbeddedNetworkController *const _controller;
 	const Address _myAddress;
 	std::string _myAddressStr;
 	std::string _host;
 	std::string _db;
 	std::string _auth;
-	const int _port;
+	int _port;
 
 	void *_networksDbWatcherConnection;
 	void *_membersDbWatcherConnection;
@@ -89,13 +132,19 @@ private:
 	std::thread _membersDbWatcher;
 
 	std::unordered_map< uint64_t,std::shared_ptr<_Network> > _networks;
-	std::mutex _networks_l;
+	mutable std::mutex _networks_l;
+
+	BlockingQueue< std::unique_ptr<nlohmann::json> > _commitQueue;
+	std::thread _commitThread[ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS];
 
 	mutable std::mutex _readyLock; // locked until ready
 	std::atomic<int> _ready;
 	std::atomic<int> _run;
+	mutable volatile bool _waitNoticePrinted;
 };
 
 } // namespace ZeroTier
 
 #endif
+
+#endif // ZT_CONTROLLER_USE_RETHINKDB