From b577d476d3392ed1522fbcbe7760de2cd774e138 Mon Sep 17 00:00:00 2001 From: "kevin.larke" Date: Sun, 19 Apr 2020 13:01:21 -0400 Subject: [PATCH] cwSpscQueueTmpl.h/cpp : Initial commit. --- cwSpScQueueTmpl.cpp | 14 ++++++++++++-- cwSpScQueueTmpl.h | 29 +++++++++++------------------ 2 files changed, 23 insertions(+), 20 deletions(-) diff --git a/cwSpScQueueTmpl.cpp b/cwSpScQueueTmpl.cpp index 112c697..9c51e2f 100644 --- a/cwSpScQueueTmpl.cpp +++ b/cwSpScQueueTmpl.cpp @@ -34,6 +34,7 @@ namespace cw unsigned iter; // execution counter unsigned msgN; // count of msg's processed shared_t* share; // shared variables + uint8_t prevId; } ctx_t; @@ -57,7 +58,7 @@ namespace cw m->checksum += m->data[i]; } - c->share->q->push(m); + c->share->q->publish(); c->msgN++; @@ -84,7 +85,16 @@ namespace cw if( curCheckSum != m->checksum ) cwLogError(kOpFailRC,"Checksum mismatch.0x%x != 0x%x ",curCheckSum,m->checksum); - + else + { + uint8_t id = c->prevId + 1; + if( id != m->data[0] ) + cwLogInfo("drop "); + + c->prevId = m->data[0]; + } + + c->msgN++; } c->iter++; diff --git a/cwSpScQueueTmpl.h b/cwSpScQueueTmpl.h index f555e38..418ed79 100644 --- a/cwSpScQueueTmpl.h +++ b/cwSpScQueueTmpl.h @@ -10,31 +10,27 @@ namespace cw public: spScQueueTmpl( unsigned eleN ) { - _aV = mem::allocZ(eleN); - _mem = mem::allocZ(eleN); - _wi.load(std::memory_order_release); - _ri.load(std::memory_order_release); - - for(unsigned i=0; i(eleN); + _wi.store(0,std::memory_order_release); + _ri.store(0,std::memory_order_release); } virtual ~spScQueueTmpl() { mem::release(_aV); - mem::release(_mem); } T* get() { unsigned wi = _wi.load( std::memory_order_relaxed ); - return _aV[wi]; + return _aV + wi; } - rc_t push( T* v ) + rc_t publish() { unsigned ri = _ri.load( std::memory_order_acquire ); - unsigned wi = _wi.load( std::memory_order_relaxed ); + unsigned wi = _wi.load( std::memory_order_relaxed ); // _wi is written in this thread // calc. the count of full elements unsigned n = wi >= ri ? wi-ri : (_aN-ri) + wi; @@ -44,11 +40,9 @@ namespace cw if( n >= _aN-1 ) return kBufTooSmallRC; - // store the new element - _aV[wi] = v; - wi = (wi+1) % _aN; - + + // advance the write position _wi.store( wi, std::memory_order_release ); return kOkRC; @@ -64,7 +58,7 @@ namespace cw if( n == 0 ) return nullptr; - T* v = _aV[ri]; + T* v = _aV + ri; ri = (ri+1) % _aN; @@ -78,8 +72,7 @@ namespace cw private: unsigned _aN = 0; - T** _aV = nullptr; - T* _mem; + T* _aV = nullptr; std::atomic _wi; std::atomic _ri;