void/ZeroTierOne
2
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2025-06-05 03:53:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Make rxQueue lock-free using an atomic counter ring buffer.

Browse source
This commit is contained in:
Adam Ierymenko 2017-08-08 13:21:10 -07:00
parent 7e6598e9ca
commit e3cf756785
3 changed files with 33 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
node/AtomicCounter.hpp
View file

@ -47,6 +47,15 @@ public:
_v = 0; _v = 0;
} }
inline int load() const
{
#ifdef __GNUC__
return __sync_or_and_fetch(&_v,0);
#else
return _v.load();
#endif
}
inline int operator++() inline int operator++()
{ {
#ifdef __GNUC__ #ifdef __GNUC__

36
node/Switch.cpp
View file

@ -120,10 +120,8 @@ void Switch::onRemotePacket(void *tPtr,const int64_t localSocket,const InetAddre
// Total fragments must be more than 1, otherwise why are we // Total fragments must be more than 1, otherwise why are we
// seeing a Packet::Fragment? // seeing a Packet::Fragment?
Mutex::Lock _l(_rxQueue_m); RXQueueEntry *const rq = _findRXQueueEntry(fragmentPacketId);
RXQueueEntry *const rq = _findRXQueueEntry(now,fragmentPacketId); if (rq->packetId != fragmentPacketId) {
if ((!rq->timestamp)||(rq->packetId != fragmentPacketId)) {
// No packet found, so we received a fragment without its head. // No packet found, so we received a fragment without its head.
rq->timestamp = now; rq->timestamp = now;
@ -250,10 +248,8 @@ void Switch::onRemotePacket(void *tPtr,const int64_t localSocket,const InetAddre
((uint64_t)reinterpret_cast<const uint8_t *>(data)[7]) ((uint64_t)reinterpret_cast<const uint8_t *>(data)[7])
); );
Mutex::Lock _l(_rxQueue_m); RXQueueEntry *const rq = _findRXQueueEntry(packetId);
RXQueueEntry *const rq = _findRXQueueEntry(now,packetId); if (rq->packetId != packetId) {
if ((!rq->timestamp)||(rq->packetId != packetId)) {
// If we have no other fragments yet, create an entry and save the head // If we have no other fragments yet, create an entry and save the head
rq->timestamp = now; rq->timestamp = now;
@ -286,14 +282,7 @@ void Switch::onRemotePacket(void *tPtr,const int64_t localSocket,const InetAddre
// Packet is unfragmented, so just process it // Packet is unfragmented, so just process it
IncomingPacket packet(data,len,path,now); IncomingPacket packet(data,len,path,now);
if (!packet.tryDecode(RR,tPtr)) { if (!packet.tryDecode(RR,tPtr)) {
Mutex::Lock _l(_rxQueue_m); RXQueueEntry *const rq = _nextRXQueueEntry();
RXQueueEntry *rq = &(_rxQueue[ZT_RX_QUEUE_SIZE - 1]);
unsigned long i = ZT_RX_QUEUE_SIZE - 1;
while ((i)&&(rq->timestamp)) {
RXQueueEntry *tmp = &(_rxQueue[--i]);
if (tmp->timestamp < rq->timestamp)
rq = tmp;
}
rq->timestamp = now; rq->timestamp = now;
rq->packetId = packet.packetId(); rq->packetId = packet.packetId();
rq->frag0 = packet; rq->frag0 = packet;
@ -590,15 +579,12 @@ void Switch::doAnythingWaitingForPeer(void *tPtr,const SharedPtr<Peer> &peer)
_outstandingWhoisRequests.erase(peer->address()); _outstandingWhoisRequests.erase(peer->address());
} }
{ // finish processing any packets waiting on peer's public key / identity // finish processing any packets waiting on peer's public key / identity
Mutex::Lock _l(_rxQueue_m); for(unsigned int ptr=0;ptr<ZT_RX_QUEUE_SIZE;++ptr) {
unsigned long i = ZT_RX_QUEUE_SIZE; RXQueueEntry *const rq = &(_rxQueue[ptr]);
while (i) { if ((rq->timestamp)&&(rq->complete)) {
RXQueueEntry *rq = &(_rxQueue[--i]); if (rq->frag0.tryDecode(RR,tPtr))
if ((rq->timestamp)&&(rq->complete)) { rq->timestamp = 0;
if (rq->frag0.tryDecode(RR,tPtr))
rq->timestamp = 0;
}
} }
} }

27
node/Switch.hpp
View file

@ -169,25 +169,24 @@ private:
bool complete; // if true, packet is complete bool complete; // if true, packet is complete
}; };
RXQueueEntry _rxQueue[ZT_RX_QUEUE_SIZE]; RXQueueEntry _rxQueue[ZT_RX_QUEUE_SIZE];
Mutex _rxQueue_m; AtomicCounter _rxQueuePtr;
/* Returns the matching or oldest entry. Caller must check timestamp and // Returns matching or next available RX queue entry
* packet ID to determine which. */ inline RXQueueEntry *_findRXQueueEntry(uint64_t packetId)
inline RXQueueEntry *_findRXQueueEntry(uint64_t now,uint64_t packetId)
{ {
RXQueueEntry *rq; unsigned int ptr = static_cast<unsigned int>(_rxQueuePtr.load());
RXQueueEntry *oldest = &(_rxQueue[ZT_RX_QUEUE_SIZE - 1]); for(unsigned int k=0;k<ZT_RX_QUEUE_SIZE;++k) {
unsigned long i = ZT_RX_QUEUE_SIZE; RXQueueEntry *rq = &(_rxQueue[--ptr % ZT_RX_QUEUE_SIZE]);
while (i) {
rq = &(_rxQueue[--i]);
if ((rq->packetId == packetId)&&(rq->timestamp)) if ((rq->packetId == packetId)&&(rq->timestamp))
return rq; return rq;
if ((now - rq->timestamp) >= ZT_RX_QUEUE_EXPIRE)
rq->timestamp = 0;
if (rq->timestamp < oldest->timestamp)
oldest = rq;
} }
return oldest; return &(_rxQueue[static_cast<unsigned int>(++_rxQueuePtr) % ZT_RX_QUEUE_SIZE]);
}
// Returns next RX queue entry in ring buffer and increments ring counter
inline RXQueueEntry *_nextRXQueueEntry()
{
return &(_rxQueue[static_cast<unsigned int>(++_rxQueuePtr) % ZT_RX_QUEUE_SIZE]);
} }
// ZeroTier-layer TX queue entry // ZeroTier-layer TX queue entry