diff --git a/node/Cluster.cpp b/node/Cluster.cpp index a5c30e0fc..ec1151593 100644 --- a/node/Cluster.cpp +++ b/node/Cluster.cpp @@ -223,10 +223,10 @@ void Cluster::handleIncomingStateMessage(const void *msg,unsigned int len) std::list<_SQE> q; { Mutex::Lock _l(_sendViaClusterQueue_m); - std::map< Address,std::list<_SQE> >::iterator qe(_sendViaClusterQueue.find(id.address())); - if (qe != _sendViaClusterQueue.end()) { - q.swap(qe->second); // just swap ptr instead of copying - _sendViaClusterQueue.erase(qe); + for(std::list<_SQE>::iterator qi(_sendViaClusterQueue.begin());qi!=_sendViaClusterQueue.end();) { + if (qi->toPeerAddress == id.address()) + q.splice(q.end(),_sendViaClusterQueue,qi++); + else ++qi; } } for(std::list<_SQE>::iterator qi(q.begin());qi!=q.end();++qi) @@ -368,16 +368,17 @@ void Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPee if (len > ZT_PROTO_MAX_PACKET_LENGTH) // sanity check return; - _sendViaClusterQueue_m.lock(); - unsigned long queueCount; + unsigned int queueCount = 0; { - std::map< Address,std::list<_SQE> >::const_iterator qe(_sendViaClusterQueue.find(fromPeerAddress)); - queueCount = (qe == _sendViaClusterQueue.end()) ? 0 : (unsigned long)qe->second.size(); - } - _sendViaClusterQueue_m.unlock(); - if (queueCount > ZT_CLUSTER_MAX_QUEUE_PER_SENDER) { - TRACE("dropping sendViaCluster for %s -> %s since queue for sender is full",fromPeerAddress.toString().c_str(),toPeerAddress.toString().c_str()); - return; + Mutex::Lock _l(_sendViaClusterQueue_m); + for(std::list<_SQE>::const_iterator qi(_sendViaClusterQueue.begin());qi!=_sendViaClusterQueue.end();++qi) { + if (qi->fromPeerAddress == fromPeerAddress) { + if (++queueCount > ZT_CLUSTER_MAX_QUEUE_PER_SENDER) { + TRACE("dropping sendViaCluster for %s -> %s since queue for sender is full",fromPeerAddress.toString().c_str(),toPeerAddress.toString().c_str()); + return; + } + } + } } const uint64_t now = RR->node->now(); @@ -386,9 +387,9 @@ void Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPee unsigned int mostRecentMemberId = 0xffffffff; { Mutex::Lock _l2(_remotePeers_m); - std::map< std::pair,uint64_t >::const_iterator rpe(_remotePeers.lower_bound(std::pair(fromPeerAddress,0))); + std::map< std::pair,uint64_t >::const_iterator rpe(_remotePeers.lower_bound(std::pair(toPeerAddress,0))); for(;;) { - if ((rpe == _remotePeers.end())||(rpe->first.first != fromPeerAddress)) + if ((rpe == _remotePeers.end())||(rpe->first.first != toPeerAddress)) break; else if (rpe->second > mostRecentTs) { mostRecentTs = rpe->second; @@ -420,7 +421,7 @@ void Cluster::sendViaCluster(const Address &fromPeerAddress,const Address &toPee if (enqueueAndWait) { TRACE("sendViaCluster %s -> %s enqueueing to wait for HAVE_PEER",fromPeerAddress.toString().c_str(),toPeerAddress.toString().c_str()); Mutex::Lock _l(_sendViaClusterQueue_m); - _sendViaClusterQueue[fromPeerAddress].push_back(_SQE(now,toPeerAddress,data,len,unite)); + _sendViaClusterQueue.push_back(_SQE(now,fromPeerAddress,toPeerAddress,data,len,unite)); return; } } @@ -484,13 +485,8 @@ void Cluster::doPeriodicTasks() { Mutex::Lock _l2(_sendViaClusterQueue_m); - for(std::map< Address,std::list<_SQE> >::iterator qi(_sendViaClusterQueue.begin());qi!=_sendViaClusterQueue.end();) { - for(std::list<_SQE>::iterator qii(qi->second.begin());qii!=qi->second.end();) { - if ((now - qii->timestamp) >= ZT_CLUSTER_QUEUE_EXPIRATION) - qi->second.erase(qii++); - else ++qii; - } - if (qi->second.empty()) + for(std::list<_SQE>::iterator qi(_sendViaClusterQueue.begin());qi!=_sendViaClusterQueue.end();) { + if ((now - qi->timestamp) >= ZT_CLUSTER_QUEUE_EXPIRATION) _sendViaClusterQueue.erase(qi++); else ++qi; } diff --git a/node/Cluster.hpp b/node/Cluster.hpp index 63196928d..4197a14b6 100644 --- a/node/Cluster.hpp +++ b/node/Cluster.hpp @@ -39,6 +39,7 @@ #include "Constants.hpp" #include "../include/ZeroTierOne.h" #include "Address.hpp" +#include "Array.hpp" #include "InetAddress.hpp" #include "SHA512.hpp" #include "Utils.hpp" @@ -74,7 +75,7 @@ /** * Expiration time for send queue entries */ -#define ZT_CLUSTER_QUEUE_EXPIRATION 1500 +#define ZT_CLUSTER_QUEUE_EXPIRATION 500 namespace ZeroTier { @@ -372,18 +373,20 @@ private: struct _SQE { _SQE() : timestamp(0),len(0),unite(false) {} - _SQE(const uint64_t ts,const Address &t,const void *d,const unsigned int l,const bool u) : + _SQE(const uint64_t ts,const Address &f,const Address &t,const void *d,const unsigned int l,const bool u) : timestamp(ts), + fromPeerAddress(f), toPeerAddress(t), len(l), unite(u) { memcpy(data,d,l); } uint64_t timestamp; + Address fromPeerAddress; Address toPeerAddress; unsigned int len; bool unite; unsigned char data[ZT_PROTO_MAX_PACKET_LENGTH]; }; - std::map< Address,std::list<_SQE> > _sendViaClusterQueue; + std::list<_SQE> _sendViaClusterQueue; Mutex _sendViaClusterQueue_m; uint64_t _lastFlushed; diff --git a/tests/http/agent.js b/tests/http/agent.js index 8dec82d8d..8afe517f7 100644 --- a/tests/http/agent.js +++ b/tests/http/agent.js @@ -3,14 +3,17 @@ // --------------------------------------------------------------------------- // Customizable parameters: +// Time between startup and first test attempt +var TEST_STARTUP_LAG = 10000; + // Maximum interval between test attempts (actual timing is random % this) var TEST_INTERVAL_MAX = (60000 * 10); // Test timeout in ms -var TEST_TIMEOUT = 60000; +var TEST_TIMEOUT = 30000; // Where should I get other agents' IDs and POST results? -var SERVER_HOST = '52.32.186.221'; +var SERVER_HOST = '52.26.196.147'; var SERVER_PORT = 18080; // Which port do agents use to serve up test data to each other? @@ -186,5 +189,5 @@ app.get('/',function(req,res) { return res.status(200).send(payload); }); var expressServer = app.listen(AGENT_PORT,function () { // Start timeout-based loop - doTest(); + setTimeout(doTest(),TEST_STARTUP_LAG); });