tests/unit/pqueue/pop.cpp

   1 //$$CDS-header$$
   2
   3 #include "cppunit/thread.h"
   4 #include "pqueue/pqueue_item.h"
   5 #include "pqueue/pqueue_type.h"
   6
   7 #include <vector>
   8 #include <memory>
   9
  10 namespace pqueue {
  11
  12 #define TEST_CASE( Q ) void Q() { test< Types<pqueue::SimpleValue>::Q >(); }
  13 #define TEST_BOUNDED( Q ) void Q() { test_bounded< Types<pqueue::SimpleValue>::Q >(); }
  14
  15     namespace {
  16         static size_t s_nThreadCount = 8;
  17         static size_t s_nQueueSize = 2000000;
  18     }
  19 } // namespace pqueue
  20
  21 namespace pqueue {
  22
  23     class PQueue_Pop: public CppUnitMini::TestCase
  24     {
  25
  26         template <class PQueue>
  27         class Pusher: public CppUnitMini::TestThread
  28         {
  29             virtual TestThread *    clone()
  30             {
  31                 return new Pusher( *this );
  32             }
  33         public:
  34             PQueue&             m_Queue;
  35             size_t              m_nPushError;
  36
  37             typedef std::vector<size_t> array_type;
  38             array_type          m_arr;
  39
  40         public:
  41             Pusher( CppUnitMini::ThreadPool& pool, PQueue& q )
  42                 : CppUnitMini::TestThread( pool )
  43                 , m_Queue( q )
  44             {}
  45             Pusher( Pusher& src )
  46                 : CppUnitMini::TestThread( src )
  47                 , m_Queue( src.m_Queue )
  48             {}
  49
  50             PQueue_Pop&  getTest()
  51             {
  52                 return static_cast<PQueue_Pop&>( m_Pool.m_Test );
  53             }
  54
  55             virtual void init()
  56             {
  57                 cds::threading::Manager::attachThread();
  58             }
  59             virtual void fini()
  60             {
  61                 cds::threading::Manager::detachThread();
  62             }
  63
  64             virtual void test()
  65             {
  66                 m_nPushError = 0;
  67
  68                 for ( array_type::const_iterator it = m_arr.begin(); it != m_arr.end(); ++it ) {
  69                     if ( !m_Queue.push( SimpleValue( *it ) ))
  70                         ++m_nPushError;
  71                 }
  72             }
  73
  74             void prepare( size_t nStart, size_t nEnd )
  75             {
  76                 m_arr.reserve( nEnd - nStart );
  77                 for ( size_t i = nStart; i < nEnd; ++i )
  78                     m_arr.push_back( i );
  79                 shuffle( m_arr.begin(), m_arr.end() );
  80             }
  81         };
  82
  83         template <class PQueue>
  84         class Popper: public CppUnitMini::TestThread
  85         {
  86             virtual TestThread *    clone()
  87             {
  88                 return new Popper( *this );
  89             }
  90         public:
  91             PQueue&             m_Queue;
  92             size_t              m_nPopError;
  93             size_t              m_nPopErrorEq;
  94             size_t              m_nPopSuccess;
  95             size_t              m_nPopFailed;
  96
  97         public:
  98             Popper( CppUnitMini::ThreadPool& pool, PQueue& q )
  99                 : CppUnitMini::TestThread( pool )
 100                 , m_Queue( q )
 101             {}
 102             Popper( Popper& src )
 103                 : CppUnitMini::TestThread( src )
 104                 , m_Queue( src.m_Queue )
 105             {}
 106
 107             PQueue_Pop&  getTest()
 108             {
 109                 return static_cast<PQueue_Pop&>( m_Pool.m_Test );
 110             }
 111
 112             virtual void init()
 113             {
 114                 cds::threading::Manager::attachThread();
 115             }
 116             virtual void fini()
 117             {
 118                 cds::threading::Manager::detachThread();
 119             }
 120
 121             virtual void test()
 122             {
 123                 m_nPopError = 0;
 124                 m_nPopErrorEq = 0;
 125                 m_nPopSuccess = 0;
 126                 m_nPopFailed = 0;
 127
 128                 size_t nPrevKey;
 129                 SimpleValue val;
 130                 if ( m_Queue.pop( val )) {
 131                     ++m_nPopSuccess;
 132                     nPrevKey = val.key;
 133
 134                     while ( !m_Queue.empty() ) {
 135                         if ( m_Queue.pop( val )) {
 136                             ++m_nPopSuccess;
 137                             if ( val.key > nPrevKey )
 138                                 ++m_nPopError;
 139                             else if ( val.key == nPrevKey )
 140                                 ++m_nPopErrorEq;
 141                             nPrevKey = val.key;
 142                         }
 143                         else
 144                             ++m_nPopFailed;
 145                     }
 146                 }
 147                 else
 148                     ++m_nPopFailed;
 149             }
 150         };
 151
 152     protected:
 153         template <class PQueue>
 154         void test()
 155         {
 156             PQueue testQueue;
 157             test_with( testQueue );
 158         }
 159
 160         template <class PQueue>
 161         void test_bounded()
 162         {
 163             std::unique_ptr<PQueue> pq( new PQueue(s_nQueueSize) );
 164             test_with( *pq.get() );
 165         }
 166
 167         template <class PQueue>
 168         void test_with( PQueue& testQueue )
 169         {
 170             size_t const nThreadItemCount = s_nQueueSize / s_nThreadCount;
 171
 172             // push
 173             {
 174                 CppUnitMini::ThreadPool pool( *this );
 175                 pool.add( new Pusher<PQueue>( pool, testQueue ), s_nThreadCount );
 176
 177                 size_t nStart = 0;
 178                 for ( CppUnitMini::ThreadPool::iterator it = pool.begin(); it != pool.end(); ++it ) {
 179                     Pusher<PQueue> * pThread = static_cast<Pusher<PQueue> *>(*it);
 180                     pThread->prepare( nStart, nStart + nThreadItemCount );
 181                     nStart += nThreadItemCount;
 182                 }
 183
 184                 CPPUNIT_MSG( "   Push, thread count=" << s_nThreadCount << ", item count=" << nThreadItemCount * s_nThreadCount << " ..." );
 185                 pool.run();
 186                 CPPUNIT_MSG( "     Duration=" << pool.avgDuration() );
 187             }
 188
 189             // pop
 190             {
 191                 CppUnitMini::ThreadPool pool( *this );
 192                 pool.add( new Popper<PQueue>( pool, testQueue ), s_nThreadCount );
 193
 194                 CPPUNIT_MSG( "   Pop, thread count=" << s_nThreadCount << ", item count=" << nThreadItemCount * s_nThreadCount << " ..." );
 195                 pool.run();
 196                 CPPUNIT_MSG( "     Duration=" << pool.avgDuration() );
 197
 198                 // Analyze result
 199                 size_t nTotalPopped = 0;
 200                 size_t nTotalError = 0;
 201                 size_t nTotalErrorEq = 0;
 202                 size_t nTotalFailed = 0;
 203                 for ( CppUnitMini::ThreadPool::iterator it = pool.begin(); it != pool.end(); ++it ) {
 204                     Popper<PQueue> * pThread = static_cast<Popper<PQueue> *>(*it);
 205
 206                     nTotalPopped += pThread->m_nPopSuccess;
 207                     nTotalError += pThread->m_nPopError;
 208                     nTotalErrorEq += pThread->m_nPopErrorEq;
 209                     nTotalFailed += pThread->m_nPopFailed;
 210                 }
 211
 212                 CPPUNIT_MSG( "   Total: popped=" << nTotalPopped << ", empty pop=" << nTotalFailed
 213                              << "\n  Errors: pop equal=" << nTotalErrorEq << ", priority violation=" << nTotalError
 214                              );
 215                 CPPUNIT_CHECK( nTotalPopped == nThreadItemCount * s_nThreadCount );
 216                 CPPUNIT_CHECK( nTotalError == 0 );
 217                 CPPUNIT_CHECK( nTotalErrorEq == 0 );
 218             }
 219
 220             CPPUNIT_MSG( testQueue.statistics() );
 221         }
 222
 223         void setUpParams( const CppUnitMini::TestCfg& cfg ) {
 224             s_nThreadCount = cfg.getULong("ThreadCount", (unsigned long) s_nThreadCount );
 225             s_nQueueSize = cfg.getULong("QueueSize", (unsigned long) s_nQueueSize );
 226         }
 227
 228     protected:
 229 #include "pqueue/pqueue_defs.h"
 230         CDSUNIT_DECLARE_MSPriorityQueue
 231         CDSUNIT_DECLARE_EllenBinTree
 232         CDSUNIT_DECLARE_SkipList
 233         CDSUNIT_DECLARE_FCPriorityQueue
 234         CDSUNIT_DECLARE_StdPQueue
 235
 236         CPPUNIT_TEST_SUITE_(PQueue_Pop, "PQueue_Push")
 237             CDSUNIT_TEST_MSPriorityQueue
 238             CDSUNIT_TEST_EllenBinTree
 239             CDSUNIT_TEST_SkipList
 240             CDSUNIT_TEST_FCPriorityQueue
 241             CDUNIT_TEST_StdPQueue
 242         CPPUNIT_TEST_SUITE_END();
 243     };
 244
 245 } // namespace queue
 246
 247 CPPUNIT_TEST_SUITE_REGISTRATION(pqueue::PQueue_Pop);