From c9c17eaddd6380724a3b0360e8a2e5ae8f84a69c Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Sun, 11 Nov 2018 22:35:15 -0800 Subject: Retire RethinkDB, simple receive path multithreading. --- attic/RethinkDB.cpp | 497 +++++++++++++++++++++++++++++++ attic/RethinkDB.hpp | 84 ++++++ controller/EmbeddedNetworkController.hpp | 3 + controller/RethinkDB.cpp | 497 ------------------------------- controller/RethinkDB.hpp | 84 ------ service/OneService.cpp | 76 +++++ 6 files changed, 660 insertions(+), 581 deletions(-) create mode 100644 attic/RethinkDB.cpp create mode 100644 attic/RethinkDB.hpp delete mode 100644 controller/RethinkDB.cpp delete mode 100644 controller/RethinkDB.hpp diff --git a/attic/RethinkDB.cpp b/attic/RethinkDB.cpp new file mode 100644 index 00000000..d54b30b6 --- /dev/null +++ b/attic/RethinkDB.cpp @@ -0,0 +1,497 @@ +/* + * ZeroTier One - Network Virtualization Everywhere + * Copyright (C) 2011-2018 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 . + */ + +//#define ZT_CONTROLLER_USE_RETHINKDB + +#ifdef ZT_CONTROLLER_USE_RETHINKDB + +#include +#include +#include +#include + +#include "RethinkDB.hpp" +#include "EmbeddedNetworkController.hpp" + +#include "../version.h" + +#include +#include +#include + +#include "../ext/librethinkdbxx/build/include/rethinkdb.h" + +namespace R = RethinkDB; +using json = nlohmann::json; + +namespace ZeroTier { + +static const char *_timestr() +{ + time_t t = time(0); + char *ts = ctime(&t); + char *p = ts; + if (!p) + return ""; + while (*p) { + if (*p == '\n') { + *p = (char)0; + break; + } + ++p; + } + return ts; +} + +RethinkDB::RethinkDB(EmbeddedNetworkController *const nc,const Identity &myId,const char *path) : + DB(nc,myId,path), + _ready(2), // two tables need to be synchronized before we're ready, so this is ready when it reaches 0 + _run(1), + _waitNoticePrinted(false) +{ + // rethinkdb:host:port:db[:auth] + std::vector ps(OSUtils::split(path,":","","")); + if ((ps.size() < 4)||(ps[0] != "rethinkdb")) + throw std::runtime_error("invalid rethinkdb database url"); + _host = ps[1]; + _port = Utils::strToInt(ps[2].c_str()); + _db = ps[3]; + if (ps.size() > 4) + _auth = ps[4]; + + _readyLock.lock(); + + _membersDbWatcher = std::thread([this]() { + try { + while (_run == 1) { + try { + std::unique_ptr rdb(R::connect(this->_host,this->_port,this->_auth)); + if (rdb) { + _membersDbWatcherConnection = (void *)rdb.get(); + 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 == 2) { + if (_waitNoticePrinted) + fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB data download complete." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + this->_readyLock.unlock(); + } + } else { + try { + json &ov = tmp["old_val"]; + json &nv = tmp["new_val"]; + json oldConfig,newConfig; + if (ov.is_object()) oldConfig = ov["config"]; + if (nv.is_object()) newConfig = nv["config"]; + if (oldConfig.is_object()||newConfig.is_object()) + this->_memberChanged(oldConfig,newConfig,(this->_ready <= 0)); + } catch ( ... ) {} // ignore bad records + } + } + } + } catch (std::exception &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); + } catch (R::Error &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); + } catch ( ... ) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + } + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + } + } catch ( ... ) {} + }); + + _networksDbWatcher = std::thread([this]() { + try { + while (_run == 1) { + try { + std::unique_ptr rdb(R::connect(this->_host,this->_port,this->_auth)); + if (rdb) { + _networksDbWatcherConnection = (void *)rdb.get(); + 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 (_waitNoticePrinted) + fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB data download complete." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + this->_readyLock.unlock(); + } + } else { + try { + json &ov = tmp["old_val"]; + json &nv = tmp["new_val"]; + json oldConfig,newConfig; + if (ov.is_object()) oldConfig = ov["config"]; + if (nv.is_object()) newConfig = nv["config"]; + if (oldConfig.is_object()||newConfig.is_object()) + this->_networkChanged(oldConfig,newConfig,(this->_ready <= 0)); + } catch ( ... ) {} // ignore bad records + } + } + } + } catch (std::exception &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); + } catch (R::Error &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); + } catch ( ... ) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + } + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + } + } catch ( ... ) {} + }); + + for(int t=0;t rdb; + nlohmann::json *config = (nlohmann::json *)0; + while ((this->_commitQueue.get(config))&&(_run == 1)) { + if (!config) + continue; + nlohmann::json record; + const char *table = (const char *)0; + std::string deleteId; + try { + const std::string objtype = (*config)["objtype"]; + 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 == "trace") { + record = *config; + table = "RemoteTrace"; + } 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"; + } + } catch (std::exception &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update record creation): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); + table = (const char *)0; + } catch (R::Error &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update record creation): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); + table = (const char *)0; + } catch ( ... ) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update record creation): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + table = (const char *)0; + } + delete config; + if (!table) + continue; + const std::string jdump(OSUtils::jsonDump(record,-1)); + + while (_run == 1) { + try { + if (!rdb) + rdb = R::connect(this->_host,this->_port,this->_auth); + if (rdb) { + if (deleteId.length() > 0) { + //printf("DELETE: %s" ZT_EOL_S,deleteId.c_str()); + R::db(this->_db).table(table).get(deleteId).delete_().run(*rdb); + } else { + //printf("UPSERT: %s" ZT_EOL_S,record.dump().c_str()); + R::db(this->_db).table(table).insert(R::Datum::from_json(jdump),R::optargs("conflict","update","return_changes",false)).run(*rdb); + } + break; + } else { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): connect failed (will retry)" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + rdb.reset(); + } + } catch (std::exception &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): %s [%s]" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what(),jdump.c_str()); + rdb.reset(); + } catch (R::Error &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): %s [%s]" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str(),jdump.c_str()); + rdb.reset(); + } catch ( ... ) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): unknown exception [%s]" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),jdump.c_str()); + rdb.reset(); + } + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + } + } + } catch (std::exception &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update outer loop): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); + } catch (R::Error &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update outer loop): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); + } catch ( ... ) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update outer loop): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + } + }); + } + + _onlineNotificationThread = std::thread([this]() { + int64_t lastUpdatedNetworkStatus = 0; + std::unordered_map< std::pair,int64_t,_PairHasher > lastOnlineCumulative; + try { + std::unique_ptr rdb; + while (_run == 1) { + try { + if (!rdb) { + _connected = 0; + rdb = R::connect(this->_host,this->_port,this->_auth); + } + + if (rdb) { + _connected = 1; + R::Array batch; + R::Object tmpobj; + + std::unordered_map< std::pair,std::pair,_PairHasher > lastOnline; + { + std::lock_guard l(_lastOnline_l); + lastOnline.swap(_lastOnline); + } + + for(auto i=lastOnline.begin();i!=lastOnline.end();++i) { + lastOnlineCumulative[i->first] = i->second.first; + char tmp[64],tmp2[64]; + OSUtils::ztsnprintf(tmp,sizeof(tmp),"%.16llx-%.10llx",i->first.first,i->first.second); + tmpobj["id"] = tmp; + tmpobj["ts"] = i->second.first; + tmpobj["phy"] = i->second.second.toIpString(tmp2); + batch.emplace_back(tmpobj); + if (batch.size() >= 1024) { + R::db(this->_db).table("MemberStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb); + batch.clear(); + } + } + if (batch.size() > 0) { + R::db(this->_db).table("MemberStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb); + batch.clear(); + } + tmpobj.clear(); + + const int64_t now = OSUtils::now(); + if ((now - lastUpdatedNetworkStatus) > 10000) { + lastUpdatedNetworkStatus = now; + + std::vector< std::pair< uint64_t,std::shared_ptr<_Network> > > networks; + { + std::lock_guard l(_networks_l); + networks.reserve(_networks.size() + 1); + for(auto i=_networks.begin();i!=_networks.end();++i) + networks.push_back(*i); + } + + for(auto i=networks.begin();i!=networks.end();++i) { + char tmp[64]; + Utils::hex(i->first,tmp); + tmpobj["id"] = tmp; + { + std::lock_guard l2(i->second->lock); + tmpobj["authorizedMemberCount"] = i->second->authorizedMembers.size(); + tmpobj["totalMemberCount"] = i->second->members.size(); + unsigned long onlineMemberCount = 0; + for(auto m=i->second->members.begin();m!=i->second->members.end();++m) { + auto lo = lastOnlineCumulative.find(std::pair(i->first,m->first)); + if (lo != lastOnlineCumulative.end()) { + if ((now - lo->second) <= (ZT_NETWORK_AUTOCONF_DELAY * 2)) + ++onlineMemberCount; + else lastOnlineCumulative.erase(lo); + } + } + tmpobj["onlineMemberCount"] = onlineMemberCount; + tmpobj["bridgeCount"] = i->second->activeBridgeMembers.size(); + tmpobj["ts"] = now; + } + batch.emplace_back(tmpobj); + if (batch.size() >= 1024) { + R::db(this->_db).table("NetworkStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb); + batch.clear(); + } + } + if (batch.size() > 0) { + R::db(this->_db).table("NetworkStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb); + batch.clear(); + } + } + } + } catch (std::exception &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); + rdb.reset(); + } catch (R::Error &e) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); + rdb.reset(); + } catch ( ... ) { + fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + rdb.reset(); + } + std::this_thread::sleep_for(std::chrono::milliseconds(250)); + } + } catch ( ... ) {} + }); + + _heartbeatThread = std::thread([this]() { + try { + R::Object controllerRecord; + std::unique_ptr rdb; + + { + char publicId[1024]; + //char secretId[1024]; + char hostname[1024]; + this->_myId.toString(false,publicId); + //this->_myId.toString(true,secretId); + if (gethostname(hostname,sizeof(hostname)) != 0) { + hostname[0] = (char)0; + } else { + for(int i=0;i_myAddressStr.c_str(); + controllerRecord["publicIdentity"] = publicId; + //controllerRecord["secretIdentity"] = secretId; + if (hostname[0]) + controllerRecord["clusterHost"] = hostname; + controllerRecord["vMajor"] = ZEROTIER_ONE_VERSION_MAJOR; + controllerRecord["vMinor"] = ZEROTIER_ONE_VERSION_MINOR; + controllerRecord["vRev"] = ZEROTIER_ONE_VERSION_REVISION; + controllerRecord["vBuild"] = ZEROTIER_ONE_VERSION_BUILD; + } + + while (_run == 1) { + try { + if (!rdb) + rdb = R::connect(this->_host,this->_port,this->_auth); + if (rdb) { + controllerRecord["lastAlive"] = OSUtils::now(); + //printf("HEARTBEAT: %s" ZT_EOL_S,tmp); + R::db(this->_db).table("Controller",R::optargs("read_mode","outdated")).insert(controllerRecord,R::optargs("conflict","update")).run(*rdb); + } + } catch ( ... ) { + rdb.reset(); + } + std::this_thread::sleep_for(std::chrono::milliseconds(1000)); + } + } catch ( ... ) {} + }); +} + +RethinkDB::~RethinkDB() +{ + _run = 0; + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + _commitQueue.stop(); + for(int t=0;tclose(); + if (_networksDbWatcherConnection) + ((R::Connection *)_networksDbWatcherConnection)->close(); + _membersDbWatcher.join(); + _networksDbWatcher.join(); + _heartbeatThread.join(); + _onlineNotificationThread.join(); +} + +bool RethinkDB::waitForReady() +{ + while (_ready > 0) { + if (!_waitNoticePrinted) { + _waitNoticePrinted = true; + fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB waiting for initial data download..." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); + } + _readyLock.lock(); + _readyLock.unlock(); + } + return true; +} + +bool RethinkDB::isReady() +{ + return ((_ready)&&(_connected)); +} + +void RethinkDB::save(nlohmann::json *orig,nlohmann::json &record) +{ + if (!record.is_object()) // sanity check + return; + waitForReady(); + if (orig) { + if (*orig != record) { + record["revision"] = OSUtils::jsonInt(record["revision"],0ULL) + 1; + _commitQueue.post(new nlohmann::json(record)); + } + } else { + record["revision"] = 1; + _commitQueue.post(new nlohmann::json(record)); + } +} + +void RethinkDB::eraseNetwork(const uint64_t networkId) +{ + char tmp2[24]; + waitForReady(); + Utils::hex(networkId,tmp2); + json *tmp = new json(); + (*tmp)["id"] = tmp2; + (*tmp)["objtype"] = "_delete_network"; // pseudo-type, tells thread to delete network + _commitQueue.post(tmp); +} + +void RethinkDB::eraseMember(const uint64_t networkId,const uint64_t memberId) +{ + char tmp2[24]; + json *tmp = new json(); + 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(tmp); +} + +void RethinkDB::nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress) +{ + std::lock_guard l(_lastOnline_l); + std::pair &i = _lastOnline[std::pair(networkId,memberId)]; + i.first = OSUtils::now(); + if (physicalAddress) + i.second = physicalAddress; +} + +} // namespace ZeroTier + +#endif // ZT_CONTROLLER_USE_RETHINKDB diff --git a/attic/RethinkDB.hpp b/attic/RethinkDB.hpp new file mode 100644 index 00000000..60f04c5b --- /dev/null +++ b/attic/RethinkDB.hpp @@ -0,0 +1,84 @@ +/* + * ZeroTier One - Network Virtualization Everywhere + * Copyright (C) 2011-2018 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 . + */ + +#ifdef ZT_CONTROLLER_USE_RETHINKDB + +#ifndef ZT_CONTROLLER_RETHINKDB_HPP +#define ZT_CONTROLLER_RETHINKDB_HPP + +#include "DB.hpp" + +#define ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS 4 + +namespace ZeroTier +{ + +/** + * A controller database driver that talks to RethinkDB + * + * This is for use with ZeroTier Central. Others are free to build and use it + * but be aware that we might change it at any time. + */ +class RethinkDB : public DB +{ +public: + RethinkDB(EmbeddedNetworkController *const nc,const Identity &myId,const char *path); + virtual ~RethinkDB(); + + virtual bool waitForReady(); + virtual bool isReady(); + virtual void save(nlohmann::json *orig,nlohmann::json &record); + virtual void eraseNetwork(const uint64_t networkId); + virtual void eraseMember(const uint64_t networkId,const uint64_t memberId); + virtual void nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress); + +protected: + struct _PairHasher + { + inline std::size_t operator()(const std::pair &p) const { return (std::size_t)(p.first ^ p.second); } + }; + + std::string _host; + std::string _db; + std::string _auth; + int _port; + + void *_networksDbWatcherConnection; + void *_membersDbWatcherConnection; + std::thread _networksDbWatcher; + std::thread _membersDbWatcher; + + BlockingQueue< nlohmann::json * > _commitQueue; + std::thread _commitThread[ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS]; + + std::unordered_map< std::pair,std::pair,_PairHasher > _lastOnline; + mutable std::mutex _lastOnline_l; + std::thread _onlineNotificationThread; + + std::thread _heartbeatThread; + + mutable std::mutex _readyLock; // locked until ready + std::atomic _ready,_connected,_run; + mutable volatile bool _waitNoticePrinted; +}; + +} // namespace ZeroTier + +#endif + +#endif // ZT_CONTROLLER_USE_RETHINKDB diff --git a/controller/EmbeddedNetworkController.hpp b/controller/EmbeddedNetworkController.hpp index df6d4a7b..c3f121c5 100644 --- a/controller/EmbeddedNetworkController.hpp +++ b/controller/EmbeddedNetworkController.hpp @@ -146,10 +146,13 @@ private: Identity _signingId; std::string _signingIdAddressString; NetworkController::Sender *_sender; + std::unique_ptr _db; BlockingQueue< _RQEntry * > _queue; + std::vector _threads; std::mutex _threads_l; + std::unordered_map< _MemberStatusKey,_MemberStatus,_MemberStatusHash > _memberStatus; std::mutex _memberStatus_l; }; diff --git a/controller/RethinkDB.cpp b/controller/RethinkDB.cpp deleted file mode 100644 index d54b30b6..00000000 --- a/controller/RethinkDB.cpp +++ /dev/null @@ -1,497 +0,0 @@ -/* - * ZeroTier One - Network Virtualization Everywhere - * Copyright (C) 2011-2018 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 . - */ - -//#define ZT_CONTROLLER_USE_RETHINKDB - -#ifdef ZT_CONTROLLER_USE_RETHINKDB - -#include -#include -#include -#include - -#include "RethinkDB.hpp" -#include "EmbeddedNetworkController.hpp" - -#include "../version.h" - -#include -#include -#include - -#include "../ext/librethinkdbxx/build/include/rethinkdb.h" - -namespace R = RethinkDB; -using json = nlohmann::json; - -namespace ZeroTier { - -static const char *_timestr() -{ - time_t t = time(0); - char *ts = ctime(&t); - char *p = ts; - if (!p) - return ""; - while (*p) { - if (*p == '\n') { - *p = (char)0; - break; - } - ++p; - } - return ts; -} - -RethinkDB::RethinkDB(EmbeddedNetworkController *const nc,const Identity &myId,const char *path) : - DB(nc,myId,path), - _ready(2), // two tables need to be synchronized before we're ready, so this is ready when it reaches 0 - _run(1), - _waitNoticePrinted(false) -{ - // rethinkdb:host:port:db[:auth] - std::vector ps(OSUtils::split(path,":","","")); - if ((ps.size() < 4)||(ps[0] != "rethinkdb")) - throw std::runtime_error("invalid rethinkdb database url"); - _host = ps[1]; - _port = Utils::strToInt(ps[2].c_str()); - _db = ps[3]; - if (ps.size() > 4) - _auth = ps[4]; - - _readyLock.lock(); - - _membersDbWatcher = std::thread([this]() { - try { - while (_run == 1) { - try { - std::unique_ptr rdb(R::connect(this->_host,this->_port,this->_auth)); - if (rdb) { - _membersDbWatcherConnection = (void *)rdb.get(); - 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 == 2) { - if (_waitNoticePrinted) - fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB data download complete." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - this->_readyLock.unlock(); - } - } else { - try { - json &ov = tmp["old_val"]; - json &nv = tmp["new_val"]; - json oldConfig,newConfig; - if (ov.is_object()) oldConfig = ov["config"]; - if (nv.is_object()) newConfig = nv["config"]; - if (oldConfig.is_object()||newConfig.is_object()) - this->_memberChanged(oldConfig,newConfig,(this->_ready <= 0)); - } catch ( ... ) {} // ignore bad records - } - } - } - } catch (std::exception &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); - } catch (R::Error &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); - } catch ( ... ) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (member change stream): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - } - std::this_thread::sleep_for(std::chrono::milliseconds(250)); - } - } catch ( ... ) {} - }); - - _networksDbWatcher = std::thread([this]() { - try { - while (_run == 1) { - try { - std::unique_ptr rdb(R::connect(this->_host,this->_port,this->_auth)); - if (rdb) { - _networksDbWatcherConnection = (void *)rdb.get(); - 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 (_waitNoticePrinted) - fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB data download complete." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - this->_readyLock.unlock(); - } - } else { - try { - json &ov = tmp["old_val"]; - json &nv = tmp["new_val"]; - json oldConfig,newConfig; - if (ov.is_object()) oldConfig = ov["config"]; - if (nv.is_object()) newConfig = nv["config"]; - if (oldConfig.is_object()||newConfig.is_object()) - this->_networkChanged(oldConfig,newConfig,(this->_ready <= 0)); - } catch ( ... ) {} // ignore bad records - } - } - } - } catch (std::exception &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); - } catch (R::Error &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); - } catch ( ... ) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (network change stream): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - } - std::this_thread::sleep_for(std::chrono::milliseconds(250)); - } - } catch ( ... ) {} - }); - - for(int t=0;t rdb; - nlohmann::json *config = (nlohmann::json *)0; - while ((this->_commitQueue.get(config))&&(_run == 1)) { - if (!config) - continue; - nlohmann::json record; - const char *table = (const char *)0; - std::string deleteId; - try { - const std::string objtype = (*config)["objtype"]; - 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 == "trace") { - record = *config; - table = "RemoteTrace"; - } 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"; - } - } catch (std::exception &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update record creation): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); - table = (const char *)0; - } catch (R::Error &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update record creation): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); - table = (const char *)0; - } catch ( ... ) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update record creation): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - table = (const char *)0; - } - delete config; - if (!table) - continue; - const std::string jdump(OSUtils::jsonDump(record,-1)); - - while (_run == 1) { - try { - if (!rdb) - rdb = R::connect(this->_host,this->_port,this->_auth); - if (rdb) { - if (deleteId.length() > 0) { - //printf("DELETE: %s" ZT_EOL_S,deleteId.c_str()); - R::db(this->_db).table(table).get(deleteId).delete_().run(*rdb); - } else { - //printf("UPSERT: %s" ZT_EOL_S,record.dump().c_str()); - R::db(this->_db).table(table).insert(R::Datum::from_json(jdump),R::optargs("conflict","update","return_changes",false)).run(*rdb); - } - break; - } else { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): connect failed (will retry)" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - rdb.reset(); - } - } catch (std::exception &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): %s [%s]" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what(),jdump.c_str()); - rdb.reset(); - } catch (R::Error &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): %s [%s]" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str(),jdump.c_str()); - rdb.reset(); - } catch ( ... ) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update): unknown exception [%s]" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),jdump.c_str()); - rdb.reset(); - } - std::this_thread::sleep_for(std::chrono::milliseconds(250)); - } - } - } catch (std::exception &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update outer loop): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); - } catch (R::Error &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update outer loop): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); - } catch ( ... ) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (insert/update outer loop): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - } - }); - } - - _onlineNotificationThread = std::thread([this]() { - int64_t lastUpdatedNetworkStatus = 0; - std::unordered_map< std::pair,int64_t,_PairHasher > lastOnlineCumulative; - try { - std::unique_ptr rdb; - while (_run == 1) { - try { - if (!rdb) { - _connected = 0; - rdb = R::connect(this->_host,this->_port,this->_auth); - } - - if (rdb) { - _connected = 1; - R::Array batch; - R::Object tmpobj; - - std::unordered_map< std::pair,std::pair,_PairHasher > lastOnline; - { - std::lock_guard l(_lastOnline_l); - lastOnline.swap(_lastOnline); - } - - for(auto i=lastOnline.begin();i!=lastOnline.end();++i) { - lastOnlineCumulative[i->first] = i->second.first; - char tmp[64],tmp2[64]; - OSUtils::ztsnprintf(tmp,sizeof(tmp),"%.16llx-%.10llx",i->first.first,i->first.second); - tmpobj["id"] = tmp; - tmpobj["ts"] = i->second.first; - tmpobj["phy"] = i->second.second.toIpString(tmp2); - batch.emplace_back(tmpobj); - if (batch.size() >= 1024) { - R::db(this->_db).table("MemberStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb); - batch.clear(); - } - } - if (batch.size() > 0) { - R::db(this->_db).table("MemberStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb); - batch.clear(); - } - tmpobj.clear(); - - const int64_t now = OSUtils::now(); - if ((now - lastUpdatedNetworkStatus) > 10000) { - lastUpdatedNetworkStatus = now; - - std::vector< std::pair< uint64_t,std::shared_ptr<_Network> > > networks; - { - std::lock_guard l(_networks_l); - networks.reserve(_networks.size() + 1); - for(auto i=_networks.begin();i!=_networks.end();++i) - networks.push_back(*i); - } - - for(auto i=networks.begin();i!=networks.end();++i) { - char tmp[64]; - Utils::hex(i->first,tmp); - tmpobj["id"] = tmp; - { - std::lock_guard l2(i->second->lock); - tmpobj["authorizedMemberCount"] = i->second->authorizedMembers.size(); - tmpobj["totalMemberCount"] = i->second->members.size(); - unsigned long onlineMemberCount = 0; - for(auto m=i->second->members.begin();m!=i->second->members.end();++m) { - auto lo = lastOnlineCumulative.find(std::pair(i->first,m->first)); - if (lo != lastOnlineCumulative.end()) { - if ((now - lo->second) <= (ZT_NETWORK_AUTOCONF_DELAY * 2)) - ++onlineMemberCount; - else lastOnlineCumulative.erase(lo); - } - } - tmpobj["onlineMemberCount"] = onlineMemberCount; - tmpobj["bridgeCount"] = i->second->activeBridgeMembers.size(); - tmpobj["ts"] = now; - } - batch.emplace_back(tmpobj); - if (batch.size() >= 1024) { - R::db(this->_db).table("NetworkStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb); - batch.clear(); - } - } - if (batch.size() > 0) { - R::db(this->_db).table("NetworkStatus",R::optargs("read_mode","outdated")).insert(batch,R::optargs("conflict","update")).run(*rdb); - batch.clear(); - } - } - } - } catch (std::exception &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.what()); - rdb.reset(); - } catch (R::Error &e) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): %s" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt(),e.message.c_str()); - rdb.reset(); - } catch ( ... ) { - fprintf(stderr,"[%s] ERROR: %.10llx controller RethinkDB (node status update): unknown exception" ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - rdb.reset(); - } - std::this_thread::sleep_for(std::chrono::milliseconds(250)); - } - } catch ( ... ) {} - }); - - _heartbeatThread = std::thread([this]() { - try { - R::Object controllerRecord; - std::unique_ptr rdb; - - { - char publicId[1024]; - //char secretId[1024]; - char hostname[1024]; - this->_myId.toString(false,publicId); - //this->_myId.toString(true,secretId); - if (gethostname(hostname,sizeof(hostname)) != 0) { - hostname[0] = (char)0; - } else { - for(int i=0;i_myAddressStr.c_str(); - controllerRecord["publicIdentity"] = publicId; - //controllerRecord["secretIdentity"] = secretId; - if (hostname[0]) - controllerRecord["clusterHost"] = hostname; - controllerRecord["vMajor"] = ZEROTIER_ONE_VERSION_MAJOR; - controllerRecord["vMinor"] = ZEROTIER_ONE_VERSION_MINOR; - controllerRecord["vRev"] = ZEROTIER_ONE_VERSION_REVISION; - controllerRecord["vBuild"] = ZEROTIER_ONE_VERSION_BUILD; - } - - while (_run == 1) { - try { - if (!rdb) - rdb = R::connect(this->_host,this->_port,this->_auth); - if (rdb) { - controllerRecord["lastAlive"] = OSUtils::now(); - //printf("HEARTBEAT: %s" ZT_EOL_S,tmp); - R::db(this->_db).table("Controller",R::optargs("read_mode","outdated")).insert(controllerRecord,R::optargs("conflict","update")).run(*rdb); - } - } catch ( ... ) { - rdb.reset(); - } - std::this_thread::sleep_for(std::chrono::milliseconds(1000)); - } - } catch ( ... ) {} - }); -} - -RethinkDB::~RethinkDB() -{ - _run = 0; - std::this_thread::sleep_for(std::chrono::milliseconds(100)); - _commitQueue.stop(); - for(int t=0;tclose(); - if (_networksDbWatcherConnection) - ((R::Connection *)_networksDbWatcherConnection)->close(); - _membersDbWatcher.join(); - _networksDbWatcher.join(); - _heartbeatThread.join(); - _onlineNotificationThread.join(); -} - -bool RethinkDB::waitForReady() -{ - while (_ready > 0) { - if (!_waitNoticePrinted) { - _waitNoticePrinted = true; - fprintf(stderr,"[%s] NOTICE: %.10llx controller RethinkDB waiting for initial data download..." ZT_EOL_S,_timestr(),(unsigned long long)_myAddress.toInt()); - } - _readyLock.lock(); - _readyLock.unlock(); - } - return true; -} - -bool RethinkDB::isReady() -{ - return ((_ready)&&(_connected)); -} - -void RethinkDB::save(nlohmann::json *orig,nlohmann::json &record) -{ - if (!record.is_object()) // sanity check - return; - waitForReady(); - if (orig) { - if (*orig != record) { - record["revision"] = OSUtils::jsonInt(record["revision"],0ULL) + 1; - _commitQueue.post(new nlohmann::json(record)); - } - } else { - record["revision"] = 1; - _commitQueue.post(new nlohmann::json(record)); - } -} - -void RethinkDB::eraseNetwork(const uint64_t networkId) -{ - char tmp2[24]; - waitForReady(); - Utils::hex(networkId,tmp2); - json *tmp = new json(); - (*tmp)["id"] = tmp2; - (*tmp)["objtype"] = "_delete_network"; // pseudo-type, tells thread to delete network - _commitQueue.post(tmp); -} - -void RethinkDB::eraseMember(const uint64_t networkId,const uint64_t memberId) -{ - char tmp2[24]; - json *tmp = new json(); - 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(tmp); -} - -void RethinkDB::nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress) -{ - std::lock_guard l(_lastOnline_l); - std::pair &i = _lastOnline[std::pair(networkId,memberId)]; - i.first = OSUtils::now(); - if (physicalAddress) - i.second = physicalAddress; -} - -} // namespace ZeroTier - -#endif // ZT_CONTROLLER_USE_RETHINKDB diff --git a/controller/RethinkDB.hpp b/controller/RethinkDB.hpp deleted file mode 100644 index 60f04c5b..00000000 --- a/controller/RethinkDB.hpp +++ /dev/null @@ -1,84 +0,0 @@ -/* - * ZeroTier One - Network Virtualization Everywhere - * Copyright (C) 2011-2018 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 . - */ - -#ifdef ZT_CONTROLLER_USE_RETHINKDB - -#ifndef ZT_CONTROLLER_RETHINKDB_HPP -#define ZT_CONTROLLER_RETHINKDB_HPP - -#include "DB.hpp" - -#define ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS 4 - -namespace ZeroTier -{ - -/** - * A controller database driver that talks to RethinkDB - * - * This is for use with ZeroTier Central. Others are free to build and use it - * but be aware that we might change it at any time. - */ -class RethinkDB : public DB -{ -public: - RethinkDB(EmbeddedNetworkController *const nc,const Identity &myId,const char *path); - virtual ~RethinkDB(); - - virtual bool waitForReady(); - virtual bool isReady(); - virtual void save(nlohmann::json *orig,nlohmann::json &record); - virtual void eraseNetwork(const uint64_t networkId); - virtual void eraseMember(const uint64_t networkId,const uint64_t memberId); - virtual void nodeIsOnline(const uint64_t networkId,const uint64_t memberId,const InetAddress &physicalAddress); - -protected: - struct _PairHasher - { - inline std::size_t operator()(const std::pair &p) const { return (std::size_t)(p.first ^ p.second); } - }; - - std::string _host; - std::string _db; - std::string _auth; - int _port; - - void *_networksDbWatcherConnection; - void *_membersDbWatcherConnection; - std::thread _networksDbWatcher; - std::thread _membersDbWatcher; - - BlockingQueue< nlohmann::json * > _commitQueue; - std::thread _commitThread[ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS]; - - std::unordered_map< std::pair,std::pair,_PairHasher > _lastOnline; - mutable std::mutex _lastOnline_l; - std::thread _onlineNotificationThread; - - std::thread _heartbeatThread; - - mutable std::mutex _readyLock; // locked until ready - std::atomic _ready,_connected,_run; - mutable volatile bool _waitNoticePrinted; -}; - -} // namespace ZeroTier - -#endif - -#endif // ZT_CONTROLLER_USE_RETHINKDB diff --git a/service/OneService.cpp b/service/OneService.cpp index b1419234..86bae730 100644 --- a/service/OneService.cpp +++ b/service/OneService.cpp @@ -34,6 +34,9 @@ #include #include #include +#include +#include +#include #include "../version.h" #include "../include/ZeroTierOne.h" @@ -434,6 +437,8 @@ struct TcpConnection Mutex writeq_m; }; +#define ZT_INCOMING_PACKET_THREAD_POOL_SIZE 4 + class OneServiceImpl : public OneService { public: @@ -459,6 +464,18 @@ public: unsigned int _tertiaryPort; volatile unsigned int _udpPortPickerCounter; +#ifdef ZT_INCOMING_PACKET_THREAD_POOL_SIZE + struct { + uint8_t data[2048]; + std::thread thr; + int64_t sock; + struct sockaddr_storage from; + int size; + std::condition_variable cond; + std::mutex lock; + } _incomingPacketWorker[ZT_INCOMING_PACKET_THREAD_POOL_SIZE]; +#endif + // Local configuration and memo-ized information from it json _localConfig; Hashtable< uint64_t,std::vector > _v4Hints; @@ -587,6 +604,39 @@ public: _ports[0] = 0; _ports[1] = 0; _ports[2] = 0; + +#ifdef ZT_INCOMING_PACKET_THREAD_POOL_SIZE + for(unsigned int tn=0;tn l(_incomingPacketWorker[tn].lock); + for(;;) { + _incomingPacketWorker[tn].cond.wait(l); + if (_incomingPacketWorker[tn].size < 0) { + break; + } else if (_incomingPacketWorker[tn].size > 0) { + const ZT_ResultCode rc = _node->processWirePacket( + (void *)0, + OSUtils::now(), + _incomingPacketWorker[tn].sock, + &(_incomingPacketWorker[tn].from), + _incomingPacketWorker[tn].data, + (unsigned int)_incomingPacketWorker[tn].size, + &_nextBackgroundTaskDeadline); + if (ZT_ResultCode_isFatal(rc)) { + char tmp[256]; + OSUtils::ztsnprintf(tmp,sizeof(tmp),"fatal error code from processWirePacket: %d",(int)rc); + Mutex::Lock _l(_termReason_m); + _termReason = ONE_UNRECOVERABLE_ERROR; + _fatalErrorMessage = tmp; + this->terminate(); + break; + } + } + } + }); + } +#endif + #if ZT_VAULT_SUPPORT curl_global_init(CURL_GLOBAL_DEFAULT); #endif @@ -594,6 +644,17 @@ public: virtual ~OneServiceImpl() { +#ifdef ZT_INCOMING_PACKET_THREAD_POOL_SIZE + for(unsigned int tn=0;tn= 16)&&(reinterpret_cast(from)->ipScope() == InetAddress::IP_SCOPE_GLOBAL)) _lastDirectReceiveFromGlobal = OSUtils::now(); +#ifdef ZT_INCOMING_PACKET_THREAD_POOL_SIZE + unsigned long cksum = 0; + for(unsigned int i=0;i(sock); + memcpy(&_incomingPacketWorker[tn].from,from,sizeof(struct sockaddr_storage)); + _incomingPacketWorker[tn].size = (int)len; + _incomingPacketWorker[tn].lock.unlock(); + _incomingPacketWorker[tn].cond.notify_all(); +#else const ZT_ResultCode rc = _node->processWirePacket( (void *)0, OSUtils::now(), @@ -1856,6 +1931,7 @@ public: _fatalErrorMessage = tmp; this->terminate(); } +#endif } inline void phyOnTcpConnect(PhySocket *sock,void **uptr,bool success) -- cgit v1.2.3