gfwleak/zhangyang-libzt
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated core for tptr support

Browse Source
This commit is contained in:
Joseph Henry
2017-03-28 18:56:38 -07:00
parent b1e83a236e
commit bd3b07e00a
68 changed files with 1271 additions and 2033 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
zto/controller/EmbeddedNetworkController.cpp
View File

@@ -61,9 +61,6 @@ using json = nlohmann::json;
// Nodes are considered active if they've queried in less than this long
#define ZT_NETCONF_NODE_ACTIVE_THRESHOLD (ZT_NETWORK_AUTOCONF_DELAY * 2)
// Timeout for disk read cache (ms)
#define ZT_NETCONF_DB_CACHE_TTL 60000
namespace ZeroTier {
static json _renderRule(ZT_VirtualNetworkRule &rule)
@@ -432,6 +429,7 @@ static bool _parseRule(json &r,ZT_VirtualNetworkRule &rule)
}
EmbeddedNetworkController::EmbeddedNetworkController(Node *node,const char *dbPath) :
_startTime(OSUtils::now()),
_threadsStarted(false),
_db(dbPath),
_node(node)
@@ -503,7 +501,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
json network;
{
Mutex::Lock _l(_db_m);
network = _db.get("network",nwids,ZT_NETCONF_DB_CACHE_TTL);
network = _db.get("network",nwids);
}
if (!network.size())
return 404;
@@ -518,7 +516,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
json member;
{
Mutex::Lock _l(_db_m);
member = _db.get("network",nwids,"member",Address(address).toString(),ZT_NETCONF_DB_CACHE_TTL);
member = _db.get("network",nwids,"member",Address(address).toString());
}
if (!member.size())
return 404;
@@ -533,11 +531,10 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
Mutex::Lock _l(_db_m);
responseBody = "{";
std::string pfx(std::string("network/") + nwids + "member/");
_db.filter(pfx,ZT_NETCONF_DB_CACHE_TTL,[&responseBody](const std::string &n,const json &member) {
if (member.size() > 0) {
_db.filter((std::string("network/") + nwids + "/member/"),[&responseBody](const std::string &n,const json &member) {
if ((member.is_object())&&(member.size() > 0)) {
responseBody.append((responseBody.length() == 1) ? "\"" : ",\"");
responseBody.append(OSUtils::jsonString(member["id"],""));
responseBody.append(OSUtils::jsonString(member["id"],"0"));
responseBody.append("\":");
responseBody.append(OSUtils::jsonString(member["revision"],"0"));
}
@@ -567,7 +564,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpGET(
std::set<std::string> networkIds;
{
Mutex::Lock _l(_db_m);
_db.filter("network/",120000,[&networkIds](const std::string &n,const json &obj) {
_db.filter("network/",[&networkIds](const std::string &n,const json &obj) {
if (n.length() == (16 + 8))
networkIds.insert(n.substr(8));
return true; // do not delete
@@ -642,7 +639,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
json member;
{
Mutex::Lock _l(_db_m);
member = _db.get("network",nwids,"member",Address(address).toString(),ZT_NETCONF_DB_CACHE_TTL);
member = _db.get("network",nwids,"member",Address(address).toString());
}
json origMember(member); // for detecting changes
_initMember(member);
@@ -793,7 +790,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
test->timestamp = OSUtils::now();
if (_node) {
_node->circuitTestBegin(test,&(EmbeddedNetworkController::_circuitTestCallback));
_node->circuitTestBegin((void *)0,test,&(EmbeddedNetworkController::_circuitTestCallback));
} else {
_tests.pop_back();
return 500;
@@ -825,7 +822,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
uint64_t tryNwid = nwidPrefix | (nwidPostfix & 0xffffffULL);
if ((tryNwid & 0xffffffULL) == 0ULL) tryNwid |= 1ULL;
Utils::snprintf(nwids,sizeof(nwids),"%.16llx",(unsigned long long)tryNwid);
if (_db.get("network",nwids,ZT_NETCONF_DB_CACHE_TTL).size() <= 0) {
if (_db.get("network",nwids).size() <= 0) {
nwid = tryNwid;
break;
}
@@ -834,7 +831,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
return 503;
}
network = _db.get("network",nwids,ZT_NETCONF_DB_CACHE_TTL);
network = _db.get("network",nwids);
}
json origNetwork(network); // for detecting changes
_initNetwork(network);
@@ -1054,7 +1051,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
}
// Send an update to all members of the network
_db.filter((std::string("network/") + nwids + "/member/"),120000,[this,&now,&nwid](const std::string &n,const json &obj) {
_db.filter((std::string("network/") + nwids + "/member/"),[this,&now,&nwid](const std::string &n,const json &obj) {
_pushMemberUpdate(now,nwid,obj);
return true; // do not delete
});
@@ -1071,7 +1068,15 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpPOST(
} // else 404
} // else 404
} else if (path[0] == "dbtest") {
json testRec;
const uint64_t now = OSUtils::now();
testRec["clock"] = now;
testRec["uptime"] = (now - _startTime);
_db.put("dbtest",testRec);
}
return 404;
}
@@ -1096,7 +1101,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpDELETE(
json network;
{
Mutex::Lock _l(_db_m);
network = _db.get("network",nwids,ZT_NETCONF_DB_CACHE_TTL);
network = _db.get("network",nwids);
}
if (!network.size())
return 404;
@@ -1107,7 +1112,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpDELETE(
Mutex::Lock _l(_db_m);
json member = _db.get("network",nwids,"member",Address(address).toString(),ZT_NETCONF_DB_CACHE_TTL);
json member = _db.get("network",nwids,"member",Address(address).toString());
_db.erase("network",nwids,"member",Address(address).toString());
if (!member.size())
@@ -1120,7 +1125,7 @@ unsigned int EmbeddedNetworkController::handleControlPlaneHttpDELETE(
Mutex::Lock _l(_db_m);
std::string pfx("network/"); pfx.append(nwids);
_db.filter(pfx,120000,[](const std::string &n,const json &obj) {
_db.filter(pfx,[](const std::string &n,const json &obj) {
return false; // delete
});
@@ -1247,8 +1252,8 @@ void EmbeddedNetworkController::_request(
json member;
{
Mutex::Lock _l(_db_m);
network = _db.get("network",nwids,ZT_NETCONF_DB_CACHE_TTL);
member = _db.get("network",nwids,"member",identity.address().toString(),ZT_NETCONF_DB_CACHE_TTL);
network = _db.get("network",nwids);
member = _db.get("network",nwids,"member",identity.address().toString());
}
if (!network.size()) {
@@ -1773,7 +1778,7 @@ void EmbeddedNetworkController::_getNetworkMemberInfo(uint64_t now,uint64_t nwid
{
Mutex::Lock _l(_db_m);
_db.filter(pfx,120000,[&nmi,&now](const std::string &n,const json &member) {
_db.filter(pfx,[&nmi,&now](const std::string &n,const json &member) {
try {
if (OSUtils::jsonBool(member["authorized"],false)) {
++nmi.authorizedMemberCount;

35
zto/controller/EmbeddedNetworkController.hpp
View File

@@ -98,14 +98,6 @@ public:
throw();
private:
static void _circuitTestCallback(ZT_Node *node,ZT_CircuitTest *test,const ZT_CircuitTestReport *report);
void _request(
uint64_t nwid,
const InetAddress &fromAddr,
uint64_t requestPacketId,
const Identity &identity,
const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData);
struct _RQEntry
{
uint64_t nwid;
@@ -114,11 +106,6 @@ private:
Identity identity;
Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> metaData;
};
BlockingQueue<_RQEntry *> _queue;
Thread _threads[ZT_EMBEDDEDNETWORKCONTROLLER_BACKGROUND_THREAD_COUNT];
bool _threadsStarted;
Mutex _threads_m;
// Gathers a bunch of statistics about members of a network, IP assignments, etc. that we need in various places
struct _NetworkMemberInfo
@@ -132,15 +119,11 @@ private:
uint64_t mostRecentDeauthTime;
uint64_t nmiTimestamp; // time this NMI structure was computed
};
std::map<uint64_t,_NetworkMemberInfo> _nmiCache;
Mutex _nmiCache_m;
void _getNetworkMemberInfo(uint64_t now,uint64_t nwid,_NetworkMemberInfo &nmi);
inline void _clearNetworkMemberInfoCache(const uint64_t nwid)
{
Mutex::Lock _l(_nmiCache_m);
_nmiCache.erase(nwid);
}
static void _circuitTestCallback(ZT_Node *node,ZT_CircuitTest *test,const ZT_CircuitTestReport *report);
void _request(uint64_t nwid,const InetAddress &fromAddr,uint64_t requestPacketId,const Identity &identity,const Dictionary<ZT_NETWORKCONFIG_METADATA_DICT_CAPACITY> &metaData);
void _getNetworkMemberInfo(uint64_t now,uint64_t nwid,_NetworkMemberInfo &nmi);
inline void _clearNetworkMemberInfoCache(const uint64_t nwid) { Mutex::Lock _l(_nmiCache_m); _nmiCache.erase(nwid); }
void _pushMemberUpdate(uint64_t now,uint64_t nwid,const nlohmann::json &member);
// These init objects with default and static/informational fields
@@ -196,6 +179,16 @@ private:
member["clock"] = now;
}
const uint64_t _startTime;
BlockingQueue<_RQEntry *> _queue;
Thread _threads[ZT_EMBEDDEDNETWORKCONTROLLER_BACKGROUND_THREAD_COUNT];
bool _threadsStarted;
Mutex _threads_m;
std::map<uint64_t,_NetworkMemberInfo> _nmiCache;
Mutex _nmiCache_m;
JSONDB _db;
Mutex _db_m;

83
zto/controller/JSONDB.cpp
View File

@@ -53,64 +53,35 @@ bool JSONDB::put(const std::string &n,const nlohmann::json &obj)
_E &e = _db[n];
e.obj = obj;
e.lastModifiedOnDisk = OSUtils::getLastModified(path.c_str());
e.lastCheck = OSUtils::now();
return true;
}
const nlohmann::json &JSONDB::get(const std::string &n,unsigned long maxSinceCheck)
const nlohmann::json &JSONDB::get(const std::string &n)
{
if (!_isValidObjectName(n))
return _EMPTY_JSON;
const uint64_t now = OSUtils::now();
std::string buf;
std::map<std::string,_E>::iterator e(_db.find(n));
if (e != _db.end()) {
if ((now - e->second.lastCheck) <= (uint64_t)maxSinceCheck)
return e->second.obj;
const std::string path(_genPath(n,false));
if (!path.length()) // sanity check
return _EMPTY_JSON;
// We are somewhat tolerant to momentary disk failures here. This may
// occur over e.g. EC2's elastic filesystem (NFS).
const uint64_t lm = OSUtils::getLastModified(path.c_str());
if (e->second.lastModifiedOnDisk != lm) {
if (OSUtils::readFile(path.c_str(),buf)) {
try {
e->second.obj = OSUtils::jsonParse(buf);
e->second.lastModifiedOnDisk = lm; // don't update these if there is a parse error -- try again and again ASAP
e->second.lastCheck = now;
} catch ( ... ) {} // parse errors result in "holding pattern" behavior
}
}
if (e != _db.end())
return e->second.obj;
} else {
const std::string path(_genPath(n,false));
if (!path.length())
return _EMPTY_JSON;
if (!OSUtils::readFile(path.c_str(),buf))
return _EMPTY_JSON;
const std::string path(_genPath(n,false));
if (!path.length())
return _EMPTY_JSON;
std::string buf;
if (!OSUtils::readFile(path.c_str(),buf))
return _EMPTY_JSON;
const uint64_t lm = OSUtils::getLastModified(path.c_str());
_E &e2 = _db[n];
try {
e2.obj = OSUtils::jsonParse(buf);
} catch ( ... ) {
e2.obj = _EMPTY_JSON;
buf = "{}";
}
e2.lastModifiedOnDisk = lm;
e2.lastCheck = now;
return e2.obj;
_E &e2 = _db[n];
try {
e2.obj = OSUtils::jsonParse(buf);
} catch ( ... ) {
e2.obj = _EMPTY_JSON;
buf = "{}";
}
return e2.obj;
}
void JSONDB::erase(const std::string &n)
@@ -126,22 +97,14 @@ void JSONDB::erase(const std::string &n)
_db.erase(n);
}
void JSONDB::_reload(const std::string &p)
void JSONDB::_reload(const std::string &p,const std::string &b)
{
std::map<std::string,char> l(OSUtils::listDirectoryFull(p.c_str()));
for(std::map<std::string,char>::iterator li(l.begin());li!=l.end();++li) {
if (li->second == 'f') {
// assume p starts with _basePath, which it always does -- will throw otherwise
std::string n(p.substr(_basePath.length()));
while ((n.length() > 0)&&(n[0] == ZT_PATH_SEPARATOR)) n = n.substr(1);
if (ZT_PATH_SEPARATOR != '/') std::replace(n.begin(),n.end(),ZT_PATH_SEPARATOR,'/');
if ((n.length() > 0)&&(n[n.length() - 1] != '/')) n.push_back('/');
n.append(li->first);
if ((n.length() > 5)&&(n.substr(n.length() - 5) == ".json")) {
this->get(n.substr(0,n.length() - 5),0); // causes load and cache or update
}
} else if (li->second == 'd') {
this->_reload(p + ZT_PATH_SEPARATOR + li->first);
std::vector<std::string> dl(OSUtils::listDirectory(p.c_str()));
for(std::vector<std::string>::const_iterator di(dl.begin());di!=dl.end();++di) {
if ((di->length() > 5)&&(di->substr(di->length() - 5) == ".json")) {
this->get(b + di->substr(0,di->length() - 5));
} else {
this->_reload((p + ZT_PATH_SEPARATOR + *di),(b + *di + ZT_PATH_SEPARATOR));
}
}
}

23
zto/controller/JSONDB.hpp
View File

@@ -45,13 +45,13 @@ public:
JSONDB(const std::string &basePath) :
_basePath(basePath)
{
_reload(_basePath);
_reload(_basePath,std::string());
}
inline void reload()
{
_db.clear();
_reload(_basePath);
_reload(_basePath,std::string());
}
bool writeRaw(const std::string &n,const std::string &obj);
@@ -63,12 +63,12 @@ public:
inline bool put(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const nlohmann::json &obj) { return this->put((n1 + "/" + n2 + "/" + n3 + "/" + n4),obj); }
inline bool put(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const std::string &n5,const nlohmann::json &obj) { return this->put((n1 + "/" + n2 + "/" + n3 + "/" + n4 + "/" + n5),obj); }
const nlohmann::json &get(const std::string &n,unsigned long maxSinceCheck = 0);
const nlohmann::json &get(const std::string &n);
inline const nlohmann::json &get(const std::string &n1,const std::string &n2,unsigned long maxSinceCheck = 0) { return this->get((n1 + "/" + n2),maxSinceCheck); }
inline const nlohmann::json &get(const std::string &n1,const std::string &n2,const std::string &n3,unsigned long maxSinceCheck = 0) { return this->get((n1 + "/" + n2 + "/" + n3),maxSinceCheck); }
inline const nlohmann::json &get(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,unsigned long maxSinceCheck = 0) { return this->get((n1 + "/" + n2 + "/" + n3 + "/" + n4),maxSinceCheck); }
inline const nlohmann::json &get(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const std::string &n5,unsigned long maxSinceCheck = 0) { return this->get((n1 + "/" + n2 + "/" + n3 + "/" + n4 + "/" + n5),maxSinceCheck); }
inline const nlohmann::json &get(const std::string &n1,const std::string &n2) { return this->get((n1 + "/" + n2)); }
inline const nlohmann::json &get(const std::string &n1,const std::string &n2,const std::string &n3) { return this->get((n1 + "/" + n2 + "/" + n3)); }
inline const nlohmann::json &get(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4) { return this->get((n1 + "/" + n2 + "/" + n3 + "/" + n4)); }
inline const nlohmann::json &get(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const std::string &n5) { return this->get((n1 + "/" + n2 + "/" + n3 + "/" + n4 + "/" + n5)); }
void erase(const std::string &n);
@@ -78,11 +78,11 @@ public:
inline void erase(const std::string &n1,const std::string &n2,const std::string &n3,const std::string &n4,const std::string &n5) { this->erase(n1 + "/" + n2 + "/" + n3 + "/" + n4 + "/" + n5); }
template<typename F>
inline void filter(const std::string &prefix,unsigned long maxSinceCheck,F func)
inline void filter(const std::string &prefix,F func)
{
for(std::map<std::string,_E>::iterator i(_db.lower_bound(prefix));i!=_db.end();) {
if ((i->first.length() >= prefix.length())&&(!memcmp(i->first.data(),prefix.data(),prefix.length()))) {
if (!func(i->first,get(i->first,maxSinceCheck))) {
if (!func(i->first,get(i->first))) {
std::map<std::string,_E>::iterator i2(i); ++i2;
this->erase(i->first);
i = i2;
@@ -95,16 +95,13 @@ public:
inline bool operator!=(const JSONDB &db) const { return (!(*this == db)); }
private:
void _reload(const std::string &p);
void _reload(const std::string &p,const std::string &b);
bool _isValidObjectName(const std::string &n);
std::string _genPath(const std::string &n,bool create);
struct _E
{
nlohmann::json obj;
uint64_t lastModifiedOnDisk;
uint64_t lastCheck;
inline bool operator==(const _E &e) const { return (obj == e.obj); }
inline bool operator!=(const _E &e) const { return (obj != e.obj); }
};