summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--node/Cluster.cpp42
-rw-r--r--node/Cluster.hpp9
-rw-r--r--tests/http/agent.js9
3 files changed, 31 insertions, 29 deletions
diff --git a/node/Cluster.cpp b/node/Cluster.cpp
index a5c30e0f..ec115159 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<Address,unsigned int>,uint64_t >::const_iterator rpe(_remotePeers.lower_bound(std::pair<Address,unsigned int>(fromPeerAddress,0)));
+ std::map< std::pair<Address,unsigned int>,uint64_t >::const_iterator rpe(_remotePeers.lower_bound(std::pair<Address,unsigned int>(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 63196928..4197a14b 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 8dec82d8..8afe517f 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);
});
generated by cgit v1.2.3 (git 2.39.1) at 2025-04-21 10:13:44 +0000