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