From 0ab3e49be91ed7a8723c8b58750aef77c01e8d08 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Fri, 4 Sep 2015 14:44:22 -0700 Subject: Starting in on Switch... kill map in defrag queue, which will probably improve performance pretty decently under high load with lots of peers. --- node/Switch.cpp | 49 ++++++++++++++++++++++++++----------------------- 1 file changed, 26 insertions(+), 23 deletions(-) (limited to 'node/Switch.cpp') diff --git a/node/Switch.cpp b/node/Switch.cpp index 989f497a..3dcb7002 100644 --- a/node/Switch.cpp +++ b/node/Switch.cpp @@ -531,11 +531,14 @@ unsigned long Switch::doTimerTasks(uint64_t now) { // Time out packets that didn't get all their fragments. Mutex::Lock _l(_defragQueue_m); - for(std::map< uint64_t,DefragQueueEntry >::iterator i(_defragQueue.begin());i!=_defragQueue.end();) { - if ((now - i->second.creationTime) > ZT_FRAGMENTED_PACKET_RECEIVE_TIMEOUT) { + Hashtable< uint64_t,DefragQueueEntry >::Iterator i(_defragQueue); + uint64_t *packetId = (uint64_t *)0; + DefragQueueEntry *qe = (DefragQueueEntry *)0; + while (i.next(packetId,qe)) { + if ((now - qe->creationTime) > ZT_FRAGMENTED_PACKET_RECEIVE_TIMEOUT) { TRACE("incomplete fragmented packet %.16llx timed out, fragments discarded",i->first); - _defragQueue.erase(i++); - } else ++i; + _defragQueue.erase(*packetId); + } } } @@ -577,32 +580,31 @@ void Switch::_handleRemotePacketFragment(const InetAddress &fromAddr,const void // seeing a Packet::Fragment? Mutex::Lock _l(_defragQueue_m); - std::map< uint64_t,DefragQueueEntry >::iterator dqe(_defragQueue.find(pid)); + DefragQueueEntry &dq = _defragQueue[pid]; - if (dqe == _defragQueue.end()) { + if (!dq.creationTime) { // We received a Packet::Fragment without its head, so queue it and wait - DefragQueueEntry &dq = _defragQueue[pid]; dq.creationTime = RR->node->now(); dq.frags[fno - 1] = fragment; dq.totalFragments = tf; // total fragment count is known dq.haveFragments = 1 << fno; // we have only this fragment //TRACE("fragment (%u/%u) of %.16llx from %s",fno + 1,tf,pid,fromAddr.toString().c_str()); - } else if (!(dqe->second.haveFragments & (1 << fno))) { + } else if (!(dq.haveFragments & (1 << fno))) { // We have other fragments and maybe the head, so add this one and check - dqe->second.frags[fno - 1] = fragment; - dqe->second.totalFragments = tf; + dq.frags[fno - 1] = fragment; + dq.totalFragments = tf; //TRACE("fragment (%u/%u) of %.16llx from %s",fno + 1,tf,pid,fromAddr.toString().c_str()); - if (Utils::countBits(dqe->second.haveFragments |= (1 << fno)) == tf) { + if (Utils::countBits(dq.haveFragments |= (1 << fno)) == tf) { // We have all fragments -- assemble and process full Packet //TRACE("packet %.16llx is complete, assembling and processing...",pid); - SharedPtr packet(dqe->second.frag0); + SharedPtr packet(dq.frag0); for(unsigned int f=1;fappend(dqe->second.frags[f - 1].payload(),dqe->second.frags[f - 1].payloadLength()); - _defragQueue.erase(dqe); + packet->append(dq.frags[f - 1].payload(),dq.frags[f - 1].payloadLength()); + _defragQueue.erase(pid); // dq no longer valid after this if (!packet->tryDecode(RR)) { Mutex::Lock _l(_rxQueue_m); @@ -645,26 +647,27 @@ void Switch::_handleRemotePacketHead(const InetAddress &fromAddr,const void *dat uint64_t pid = packet->packetId(); Mutex::Lock _l(_defragQueue_m); - std::map< uint64_t,DefragQueueEntry >::iterator dqe(_defragQueue.find(pid)); + DefragQueueEntry &dq = _defragQueue[pid]; - if (dqe == _defragQueue.end()) { + if (!dq.creationTime) { // If we have no other fragments yet, create an entry and save the head - DefragQueueEntry &dq = _defragQueue[pid]; + dq.creationTime = RR->node->now(); dq.frag0 = packet; dq.totalFragments = 0; // 0 == unknown, waiting for Packet::Fragment dq.haveFragments = 1; // head is first bit (left to right) //TRACE("fragment (0/?) of %.16llx from %s",pid,fromAddr.toString().c_str()); - } else if (!(dqe->second.haveFragments & 1)) { + } else if (!(dq.haveFragments & 1)) { // If we have other fragments but no head, see if we are complete with the head - if ((dqe->second.totalFragments)&&(Utils::countBits(dqe->second.haveFragments |= 1) == dqe->second.totalFragments)) { + + if ((dq.totalFragments)&&(Utils::countBits(dq.haveFragments |= 1) == dq.totalFragments)) { // We have all fragments -- assemble and process full Packet //TRACE("packet %.16llx is complete, assembling and processing...",pid); // packet already contains head, so append fragments - for(unsigned int f=1;fsecond.totalFragments;++f) - packet->append(dqe->second.frags[f - 1].payload(),dqe->second.frags[f - 1].payloadLength()); - _defragQueue.erase(dqe); + for(unsigned int f=1;fappend(dq.frags[f - 1].payload(),dq.frags[f - 1].payloadLength()); + _defragQueue.erase(pid); // dq no longer valid after this if (!packet->tryDecode(RR)) { Mutex::Lock _l(_rxQueue_m); @@ -672,7 +675,7 @@ void Switch::_handleRemotePacketHead(const InetAddress &fromAddr,const void *dat } } else { // Still waiting on more fragments, so queue the head - dqe->second.frag0 = packet; + dq.frag0 = packet; } } // else this is a duplicate head, ignore } else { -- cgit v1.2.3 From db0369e9b8e6bf49ce8ec8f7fe706004314d4548 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Fri, 4 Sep 2015 14:56:39 -0700 Subject: Remove way-overkill multimap from Switch. --- node/Switch.cpp | 27 ++++++++++++++------------- node/Switch.hpp | 8 +++++--- 2 files changed, 19 insertions(+), 16 deletions(-) (limited to 'node/Switch.cpp') diff --git a/node/Switch.cpp b/node/Switch.cpp index 3dcb7002..7d6f8094 100644 --- a/node/Switch.cpp +++ b/node/Switch.cpp @@ -291,7 +291,7 @@ void Switch::send(const Packet &packet,bool encrypt,uint64_t nwid) if (!_trySend(packet,encrypt,nwid)) { Mutex::Lock _l(_txQueue_m); - _txQueue.insert(std::pair< Address,TXQueueEntry >(packet.destination(),TXQueueEntry(RR->node->now(),packet,encrypt,nwid))); + _txQueue.push_back(TXQueueEntry(packet.destination(),RR->node->now(),packet,encrypt,nwid)); } } @@ -435,11 +435,12 @@ void Switch::doAnythingWaitingForPeer(const SharedPtr &peer) { // finish sending any packets waiting on peer's public key / identity Mutex::Lock _l(_txQueue_m); - std::pair< std::multimap< Address,TXQueueEntry >::iterator,std::multimap< Address,TXQueueEntry >::iterator > waitingTxQueueItems(_txQueue.equal_range(peer->address())); - for(std::multimap< Address,TXQueueEntry >::iterator txi(waitingTxQueueItems.first);txi!=waitingTxQueueItems.second;) { - if (_trySend(txi->second.packet,txi->second.encrypt,txi->second.nwid)) - _txQueue.erase(txi++); - else ++txi; + for(std::list< TXQueueEntry >::iterator txi(_txQueue.begin());txi!=_txQueue.end();) { + if (txi->dest == peer->address()) { + if (_trySend(txi->packet,txi->encrypt,txi->nwid)) + _txQueue.erase(txi++); + else ++txi; + } else ++txi; } } } @@ -509,13 +510,13 @@ unsigned long Switch::doTimerTasks(uint64_t now) { // Time out TX queue packets that never got WHOIS lookups or other info. Mutex::Lock _l(_txQueue_m); - for(std::multimap< Address,TXQueueEntry >::iterator i(_txQueue.begin());i!=_txQueue.end();) { - if (_trySend(i->second.packet,i->second.encrypt,i->second.nwid)) - _txQueue.erase(i++); - else if ((now - i->second.creationTime) > ZT_TRANSMIT_QUEUE_TIMEOUT) { - TRACE("TX %s -> %s timed out",i->second.packet.source().toString().c_str(),i->second.packet.destination().toString().c_str()); - _txQueue.erase(i++); - } else ++i; + for(std::list< TXQueueEntry >::iterator txi(_txQueue.begin());txi!=_txQueue.end();) { + if (_trySend(txi->packet,txi->encrypt,txi->nwid)) + _txQueue.erase(txi++); + else if ((now - txi->creationTime) > ZT_TRANSMIT_QUEUE_TIMEOUT) { + TRACE("TX %s -> %s timed out",txi->packet.source().toString().c_str(),txi->packet.destination().toString().c_str()); + _txQueue.erase(txi++); + } else ++txi; } } diff --git a/node/Switch.hpp b/node/Switch.hpp index a1b36014..0791681f 100644 --- a/node/Switch.hpp +++ b/node/Switch.hpp @@ -214,22 +214,24 @@ private: std::list< SharedPtr > _rxQueue; Mutex _rxQueue_m; - // ZeroTier-layer TX queue by destination ZeroTier address + // ZeroTier-layer TX queue entry struct TXQueueEntry { TXQueueEntry() {} - TXQueueEntry(uint64_t ct,const Packet &p,bool enc,uint64_t nw) : + TXQueueEntry(Address d,uint64_t ct,const Packet &p,bool enc,uint64_t nw) : + dest(d), creationTime(ct), nwid(nw), packet(p), encrypt(enc) {} + Address dest; uint64_t creationTime; uint64_t nwid; Packet packet; // unencrypted/unMAC'd packet -- this is done at send time bool encrypt; }; - std::multimap< Address,TXQueueEntry > _txQueue; + std::list< TXQueueEntry > _txQueue; Mutex _txQueue_m; // Tracks sending of VERB_RENDEZVOUS to relaying peers -- cgit v1.2.3 From 3dba016a9354d9c50743877988c8d928d25f7a2b Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Fri, 4 Sep 2015 15:21:22 -0700 Subject: Almost done... very few std::map<>s remaining in any spot that matters. --- node/Hashtable.hpp | 5 +++-- node/Switch.cpp | 38 ++++++++++++++++++-------------------- node/Switch.hpp | 19 ++++++++++++++++++- 3 files changed, 39 insertions(+), 23 deletions(-) (limited to 'node/Switch.cpp') diff --git a/node/Hashtable.hpp b/node/Hashtable.hpp index 29c54838..bcc111e3 100644 --- a/node/Hashtable.hpp +++ b/node/Hashtable.hpp @@ -39,8 +39,9 @@ namespace ZeroTier { * A minimal hash table implementation for the ZeroTier core * * This is not a drop-in replacement for STL containers, and has several - * limitations. It's designed to be small and fast for use in the - * ZeroTier core. + * limitations. Keys can be uint64_t or an object, and if the latter they + * must implement a method called hashCode() that returns an unsigned long + * value that is evenly distributed. */ template class Hashtable diff --git a/node/Switch.cpp b/node/Switch.cpp index 7d6f8094..d5ee3e23 100644 --- a/node/Switch.cpp +++ b/node/Switch.cpp @@ -309,31 +309,18 @@ bool Switch::unite(const Address &p1,const Address &p2,bool force) const uint64_t now = RR->node->now(); - std::pair cg(Peer::findCommonGround(*p1p,*p2p,now)); - if (!(cg.first)) - return false; - - if (cg.first.ipScope() != cg.second.ipScope()) - return false; - - // Addresses are sorted in key for last unite attempt map for order - // invariant lookup: (p1,p2) == (p2,p1) - Array uniteKey; - if (p1 >= p2) { - uniteKey[0] = p2; - uniteKey[1] = p1; - } else { - uniteKey[0] = p1; - uniteKey[1] = p2; - } { Mutex::Lock _l(_lastUniteAttempt_m); - std::map< Array< Address,2 >,uint64_t >::const_iterator e(_lastUniteAttempt.find(uniteKey)); - if ((!force)&&(e != _lastUniteAttempt.end())&&((now - e->second) < ZT_MIN_UNITE_INTERVAL)) + uint64_t &luts = _lastUniteAttempt[_LastUniteKey(p1,p2)]; + if (((now - luts) < ZT_MIN_UNITE_INTERVAL)&&(!force)) return false; - else _lastUniteAttempt[uniteKey] = now; + luts = now; } + std::pair cg(Peer::findCommonGround(*p1p,*p2p,now)); + if ((!(cg.first))||(cg.first.ipScope() != cg.second.ipScope())) + return false; + TRACE("unite: %s(%s) <> %s(%s)",p1.toString().c_str(),cg.second.toString().c_str(),p2.toString().c_str(),cg.first.toString().c_str()); /* Tell P1 where to find P2 and vice versa, sending the packets to P1 and @@ -543,6 +530,17 @@ unsigned long Switch::doTimerTasks(uint64_t now) } } + { // Remove really old last unite attempt entries to keep table size controlled + Mutex::Lock _l(_lastUniteAttempt_m); + Hashtable< _LastUniteKey,uint64_t >::Iterator i(_lastUniteAttempt); + _LastUniteKey *k = (_LastUniteKey *)0; + uint64_t *v = (uint64_t *)0; + while (i.next(k,v)) { + if ((now - *v) >= (ZT_MIN_UNITE_INTERVAL * 16)) + _lastUniteAttempt.erase(*k); + } + } + return nextDelay; } diff --git a/node/Switch.hpp b/node/Switch.hpp index 0791681f..2d83b70c 100644 --- a/node/Switch.hpp +++ b/node/Switch.hpp @@ -235,7 +235,24 @@ private: Mutex _txQueue_m; // Tracks sending of VERB_RENDEZVOUS to relaying peers - std::map< Array< Address,2 >,uint64_t > _lastUniteAttempt; // key is always sorted in ascending order, for set-like behavior + struct _LastUniteKey + { + _LastUniteKey() : x(0),y(0) {} + _LastUniteKey(const Address &a1,const Address &a2) + { + if (a1 > a2) { + x = a2.toInt(); + y = a1.toInt(); + } else { + x = a1.toInt(); + y = a2.toInt(); + } + } + inline unsigned long hashCode() const throw() { return ((unsigned long)x ^ (unsigned long)y); } + inline bool operator==(const _LastUniteKey &k) const throw() { return ((x == k.x)&&(y == k.y)); } + uint64_t x,y; + }; + Hashtable< _LastUniteKey,uint64_t > _lastUniteAttempt; // key is always sorted in ascending order, for set-like behavior Mutex _lastUniteAttempt_m; // Active attempts to contact remote peers, including state of multi-phase NAT traversal -- cgit v1.2.3 From 85b90f122a826510491e15a1bfd6c1fe75dfe069 Mon Sep 17 00:00:00 2001 From: Adam Ierymenko Date: Fri, 4 Sep 2015 15:35:43 -0700 Subject: Final std::map<> from Switch, and add some smallish default values for hash size. --- node/Switch.cpp | 41 ++++++++++++++++++++++++----------------- node/Switch.hpp | 3 ++- 2 files changed, 26 insertions(+), 18 deletions(-) (limited to 'node/Switch.cpp') diff --git a/node/Switch.cpp b/node/Switch.cpp index d5ee3e23..995abef4 100644 --- a/node/Switch.cpp +++ b/node/Switch.cpp @@ -67,7 +67,10 @@ static const char *etherTypeName(const unsigned int etherType) Switch::Switch(const RuntimeEnvironment *renv) : RR(renv), - _lastBeaconResponse(0) + _lastBeaconResponse(0), + _outstandingWhoisRequests(32), + _defragQueue(32), + _lastUniteAttempt(8) // only really used on root servers and upstreams, and it'll grow there just fine { } @@ -389,10 +392,13 @@ void Switch::requestWhois(const Address &addr) bool inserted = false; { Mutex::Lock _l(_outstandingWhoisRequests_m); - std::pair< std::map< Address,WhoisRequest >::iterator,bool > entry(_outstandingWhoisRequests.insert(std::pair(addr,WhoisRequest()))); - if ((inserted = entry.second)) - entry.first->second.lastSent = RR->node->now(); - entry.first->second.retries = 0; // reset retry count if entry already existed + WhoisRequest &r = _outstandingWhoisRequests[addr]; + if (r.lastSent) { + r.retries = 0; // reset retry count if entry already existed, but keep waiting and retry again after normal timeout + } else { + r.lastSent = RR->node->now(); + inserted = true; + } } if (inserted) _sendWhoisRequest(addr,(const Address *)0,0); @@ -474,24 +480,25 @@ unsigned long Switch::doTimerTasks(uint64_t now) { // Retry outstanding WHOIS requests Mutex::Lock _l(_outstandingWhoisRequests_m); - for(std::map< Address,WhoisRequest >::iterator i(_outstandingWhoisRequests.begin());i!=_outstandingWhoisRequests.end();) { - unsigned long since = (unsigned long)(now - i->second.lastSent); + Hashtable< Address,WhoisRequest >::Iterator i(_outstandingWhoisRequests); + Address *a = (Address *)0; + WhoisRequest *r = (WhoisRequest *)0; + while (i.next(a,r)) { + const unsigned long since = (unsigned long)(now - r->lastSent); if (since >= ZT_WHOIS_RETRY_DELAY) { - if (i->second.retries >= ZT_MAX_WHOIS_RETRIES) { - TRACE("WHOIS %s timed out",i->first.toString().c_str()); - _outstandingWhoisRequests.erase(i++); - continue; + if (r->retries >= ZT_MAX_WHOIS_RETRIES) { + TRACE("WHOIS %s timed out",a->toString().c_str()); + _outstandingWhoisRequests.erase(*a); } else { - i->second.lastSent = now; - i->second.peersConsulted[i->second.retries] = _sendWhoisRequest(i->first,i->second.peersConsulted,i->second.retries); - ++i->second.retries; - TRACE("WHOIS %s (retry %u)",i->first.toString().c_str(),i->second.retries); + r->lastSent = now; + r->peersConsulted[r->retries] = _sendWhoisRequest(*a,r->peersConsulted,r->retries); + ++r->retries; + TRACE("WHOIS %s (retry %u)",a->toString().c_str(),r->retries); nextDelay = std::min(nextDelay,(unsigned long)ZT_WHOIS_RETRY_DELAY); } } else { nextDelay = std::min(nextDelay,ZT_WHOIS_RETRY_DELAY - since); } - ++i; } } @@ -524,7 +531,7 @@ unsigned long Switch::doTimerTasks(uint64_t now) DefragQueueEntry *qe = (DefragQueueEntry *)0; while (i.next(packetId,qe)) { if ((now - qe->creationTime) > ZT_FRAGMENTED_PACKET_RECEIVE_TIMEOUT) { - TRACE("incomplete fragmented packet %.16llx timed out, fragments discarded",i->first); + TRACE("incomplete fragmented packet %.16llx timed out, fragments discarded",*packetId); _defragQueue.erase(*packetId); } } diff --git a/node/Switch.hpp b/node/Switch.hpp index 2d83b70c..55e2c362 100644 --- a/node/Switch.hpp +++ b/node/Switch.hpp @@ -190,11 +190,12 @@ private: // Outsanding WHOIS requests and how many retries they've undergone struct WhoisRequest { + WhoisRequest() : lastSent(0),retries(0) {} uint64_t lastSent; Address peersConsulted[ZT_MAX_WHOIS_RETRIES]; // by retry unsigned int retries; // 0..ZT_MAX_WHOIS_RETRIES }; - std::map< Address,WhoisRequest > _outstandingWhoisRequests; + Hashtable< Address,WhoisRequest > _outstandingWhoisRequests; Mutex _outstandingWhoisRequests_m; // Packet defragmentation queue -- comes before RX queue in path -- cgit v1.2.3