Merge branch 'dev' into edge

controller/DB.hpp

@@ -40,6 +40,7 @@
 #include <unordered_set>
 #include <vector>
 #include <atomic>
+#include <mutex>
 
 #include "../ext/json/json.hpp"
 

controller/FileDB.cpp

@@ -32,7 +32,9 @@ namespace ZeroTier
 FileDB::FileDB(EmbeddedNetworkController *const nc,const Identity &myId,const char *path) :
 	DB(nc,myId,path),
 	_networksPath(_path + ZT_PATH_SEPARATOR_S + "network"),
-	_tracePath(_path + ZT_PATH_SEPARATOR_S + "trace")
+	_tracePath(_path + ZT_PATH_SEPARATOR_S + "trace"),
+	_onlineChanged(false),
+	_running(true)
 {
 	OSUtils::mkdir(_path.c_str());
 	OSUtils::lockDownFile(_path.c_str(),true);
@@ -69,9 +71,65 @@ FileDB::FileDB(EmbeddedNetworkController *const nc,const Identity &myId,const ch
 			} catch ( ... ) {}
 		}
 	}
+
+	_onlineUpdateThread = std::thread([this]() {
+		unsigned int cnt = 0;
+		while (this->_running) {
+			usleep(250);
+			if ((++cnt % 20) == 0) { // 5 seconds
+				std::lock_guard<std::mutex> l(this->_online_l);
+				if (!this->_running) return;
+				if (this->_onlineChanged) {
+					char p[4096],atmp[64];
+					for(auto nw=this->_online.begin();nw!=this->_online.end();++nw) {
+						OSUtils::ztsnprintf(p,sizeof(p),"%s" ZT_PATH_SEPARATOR_S "%.16llx-online.json",_networksPath.c_str(),(unsigned long long)nw->first);
+						FILE *f = fopen(p,"wb");
+						if (f) {
+							fprintf(f,"{");
+							const char *memberPrefix = "";
+							for(auto m=nw->second.begin();m!=nw->second.end();++m) {
+								fprintf(f,"%s\"%.10llx\":{" ZT_EOL_S,memberPrefix,(unsigned long long)m->first);
+								memberPrefix = ",";
+								InetAddress lastAddr;
+								const char *timestampPrefix = " ";
+								int cnt = 0;
+								for(auto ts=m->second.rbegin();ts!=m->second.rend();) {
+									if (cnt < 25) {
+										if (lastAddr != ts->second) {
+											lastAddr = ts->second;
+											fprintf(f,"%s\"%lld\":\"%s\"" ZT_EOL_S,timestampPrefix,(long long)ts->first,ts->second.toString(atmp));
+											timestampPrefix = ",";
+											++cnt;
+											++ts;
+										} else {
+											ts = std::map<int64_t,InetAddress>::reverse_iterator(m->second.erase(std::next(ts).base()));
+										}
+									} else {
+										ts = std::map<int64_t,InetAddress>::reverse_iterator(m->second.erase(std::next(ts).base()));
+									}
+								}
+								fprintf(f,"}");
+							}
+							fprintf(f,"}" ZT_EOL_S);
+							fclose(f);
+						}
+					}
+					this->_onlineChanged = false;
+				}
+			}
+		}
+	});
 }
 
-FileDB::~FileDB() {}
+FileDB::~FileDB()
+{
+	try {
+		_online_l.lock();
+		_running = false;
+		_online_l.unlock();
+		_onlineUpdateThread.join();
+	} catch ( ... ) {}
+}
 
 bool FileDB::waitForReady() { return true; }
 bool FileDB::isReady() { return true; }
@@ -94,14 +152,10 @@ void FileDB::save(nlohmann::json *orig,nlohmann::json &record)
 			if (nwid) {
 				nlohmann::json old;
 				get(nwid,old);
-
 				if ((!old.is_object())||(old != record)) {
-					OSUtils::ztsnprintf(p1,sizeof(p1),"%s" ZT_PATH_SEPARATOR_S "%.16llx.json.new",_networksPath.c_str(),nwid);
+					OSUtils::ztsnprintf(p1,sizeof(p1),"%s" ZT_PATH_SEPARATOR_S "%.16llx.json",_networksPath.c_str(),nwid);
-					OSUtils::ztsnprintf(p2,sizeof(p2),"%s" ZT_PATH_SEPARATOR_S "%.16llx.json",_networksPath.c_str(),nwid);
 					if (!OSUtils::writeFile(p1,OSUtils::jsonDump(record,-1)))
 						fprintf(stderr,"WARNING: controller unable to write to path: %s" ZT_EOL_S,p1);
-					OSUtils::rename(p1,p2);
-
 					_networkChanged(old,record,true);
 				}
 			}
@@ -111,10 +165,9 @@ void FileDB::save(nlohmann::json *orig,nlohmann::json &record)
 			if ((id)&&(nwid)) {
 				nlohmann::json network,old;
 				get(nwid,network,id,old);
-
 				if ((!old.is_object())||(old != record)) {
 					OSUtils::ztsnprintf(pb,sizeof(pb),"%s" ZT_PATH_SEPARATOR_S "%.16llx" ZT_PATH_SEPARATOR_S "member",_networksPath.c_str(),(unsigned long long)nwid);
-					OSUtils::ztsnprintf(p1,sizeof(p1),"%s" ZT_PATH_SEPARATOR_S "%.10llx.json.new",pb,(unsigned long long)id);
+					OSUtils::ztsnprintf(p1,sizeof(p1),"%s" ZT_PATH_SEPARATOR_S "%.10llx.json",pb,(unsigned long long)id);
 					if (!OSUtils::writeFile(p1,OSUtils::jsonDump(record,-1))) {
 						OSUtils::ztsnprintf(p2,sizeof(p2),"%s" ZT_PATH_SEPARATOR_S "%.16llx",_networksPath.c_str(),(unsigned long long)nwid);
 						OSUtils::mkdir(p2);
@@ -122,9 +175,6 @@ void FileDB::save(nlohmann::json *orig,nlohmann::json &record)
 						if (!OSUtils::writeFile(p1,OSUtils::jsonDump(record,-1)))
 							fprintf(stderr,"WARNING: controller unable to write to path: %s" ZT_EOL_S,p1);
 					}
-					OSUtils::ztsnprintf(p2,sizeof(p2),"%s" ZT_PATH_SEPARATOR_S "%.10llx.json",pb,(unsigned long long)id);
-					OSUtils::rename(p1,p2);
-
 					_memberChanged(old,record,true);
 				}
 			}
@@ -144,29 +194,39 @@ void FileDB::eraseNetwork(const uint64_t networkId)
 	get(networkId,network);
 	char p[16384];
 	OSUtils::ztsnprintf(p,sizeof(p),"%s" ZT_PATH_SEPARATOR_S "%.16llx.json",_networksPath.c_str(),networkId);
-
+	OSUtils::rm(p);
-	if (OSUtils::fileExists(p,false)){
+	OSUtils::ztsnprintf(p,sizeof(p),"%s" ZT_PATH_SEPARATOR_S "%.16llx-online.json",_networksPath.c_str(),networkId);
-	  OSUtils::rm(p);
+	OSUtils::rm(p);
-        }
+	OSUtils::ztsnprintf(p,sizeof(p),"%s" ZT_PATH_SEPARATOR_S "%.16llx" ZT_PATH_SEPARATOR_S "member",_networksPath.c_str(),(unsigned long long)networkId);
+	OSUtils::rmDashRf(p);
 	_networkChanged(network,nullJson,true);
+	std::lock_guard<std::mutex> l(this->_online_l);
+	this->_online.erase(networkId);
+	this->_onlineChanged = true;
 }
 
 void FileDB::eraseMember(const uint64_t networkId,const uint64_t memberId)
 {
 	nlohmann::json network,member,nullJson;
 	get(networkId,network);
-        get(memberId,member);
+	get(memberId,member);
-	char p[16384];
+	char p[4096];
 	OSUtils::ztsnprintf(p,sizeof(p),"%s" ZT_PATH_SEPARATOR_S "%.16llx" ZT_PATH_SEPARATOR_S "member" ZT_PATH_SEPARATOR_S "%.10llx.json",_networksPath.c_str(),networkId,memberId);
-	if (OSUtils::fileExists(p,false)){
+	OSUtils::rm(p);
-	    OSUtils::rm(p);
-	}
 	_memberChanged(member,nullJson,true);
+	std::lock_guard<std::mutex> l(this->_online_l);
+	this->_online[networkId].erase(memberId);
+	this->_onlineChanged = true;
 }
 
 void FileDB::nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress)
 {
-	// Nothing to do here right now in the filesystem store mode since we can just get this from the peer list
+	char mid[32],atmp[64];
+	OSUtils::ztsnprintf(mid,sizeof(mid),"%.10llx",(unsigned long long)memberId);
+	physicalAddress.toString(atmp);
+	std::lock_guard<std::mutex> l(this->_online_l);
+	this->_online[networkId][memberId][OSUtils::now()] = physicalAddress;
+	this->_onlineChanged = true;
 }
 
 } // namespace ZeroTier

controller/FileDB.hpp

@@ -48,6 +48,11 @@ public:
 protected:
 	std::string _networksPath;
 	std::string _tracePath;
+	std::thread _onlineUpdateThread;
+	std::map< uint64_t,std::map<uint64_t,std::map<int64_t,InetAddress> > > _online;
+	std::mutex _online_l;
+	bool _onlineChanged;
+	bool _running;
 };
 
 } // namespace ZeroTier

node/IncomingPacket.cpp

@@ -502,8 +502,8 @@ bool IncomingPacket::_doOK(const RuntimeEnvironment *RR,void *tPtr,const SharedP
 				}
 			}
 
-			if (!hops() && (RR->node->getMultipathMode() != ZT_MULTIPATH_NONE)) {
+			if (!hops()) {
-				_path->updateLatency((unsigned int)latency, RR->node->now());
+				_path->updateLatency((unsigned int)latency,RR->node->now());
 			}
 
 			peer->setRemoteVersion(vProto,vMajor,vMinor,vRevision);
@@ -614,7 +614,7 @@ bool IncomingPacket::_doRENDEZVOUS(const RuntimeEnvironment *RR,void *tPtr,const
 			const unsigned int port = at<uint16_t>(ZT_PROTO_VERB_RENDEZVOUS_IDX_PORT);
 			const unsigned int addrlen = (*this)[ZT_PROTO_VERB_RENDEZVOUS_IDX_ADDRLEN];
 			if ((port > 0)&&((addrlen == 4)||(addrlen == 16))) {
-				const InetAddress atAddr(field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ADDRESS,addrlen),addrlen,port);
+				InetAddress atAddr(field(ZT_PROTO_VERB_RENDEZVOUS_IDX_ADDRESS,addrlen),addrlen,port);
 				if (RR->node->shouldUsePathForZeroTierTraffic(tPtr,with,_path->localSocket(),atAddr)) {
 					const uint64_t junk = RR->node->prng();
 					RR->node->putPacket(tPtr,_path->localSocket(),atAddr,&junk,4,2); // send low-TTL junk packet to 'open' local NAT(s) and stateful firewalls