test/stress/pqueue/pop.cpp

   1 /*
   2     This file is a part of libcds - Concurrent Data Structures library
   3
   4     (C) Copyright Maxim Khizhinsky (libcds.dev@gmail.com) 2006-2016
   5
   6     Source code repo: http://github.com/khizmax/libcds/
   7     Download: http://sourceforge.net/projects/libcds/files/
   8
   9     Redistribution and use in source and binary forms, with or without
  10     modification, are permitted provided that the following conditions are met:
  11
  12     * Redistributions of source code must retain the above copyright notice, this
  13       list of conditions and the following disclaimer.
  14
  15     * Redistributions in binary form must reproduce the above copyright notice,
  16       this list of conditions and the following disclaimer in the documentation
  17       and/or other materials provided with the distribution.
  18
  19     THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  20     AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  21     IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  22     DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  23     FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  24     DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  25     SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  26     CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  27     OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  28     OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29 */
  30
  31 #include "pqueue_type.h"
  32 #include "item.h"
  33
  34 namespace {
  35     static size_t s_nThreadCount = 8;
  36     static size_t s_nQueueSize = 2000000;
  37
  38     class pqueue_pop: public cds_test::stress_fixture
  39     {
  40         typedef cds_test::stress_fixture base_class;
  41
  42     protected:
  43         template <class PQueue>
  44         class Producer: public cds_test::thread
  45         {
  46             typedef cds_test::thread base_class;
  47
  48         public:
  49             Producer( cds_test::thread_pool& pool, PQueue& queue )
  50                 : base_class( pool )
  51                 , m_Queue( queue )
  52             {}
  53
  54             Producer( Producer& src )
  55                 : base_class( src )
  56                 , m_Queue( src.m_Queue )
  57             {}
  58
  59             virtual thread * clone()
  60             {
  61                 return new Producer( *this );
  62             }
  63
  64             virtual void test()
  65             {
  66                 typedef typename PQueue::value_type value_type;
  67                 for ( array_type::const_iterator it = m_arr.begin(); it != m_arr.end(); ++it ) {
  68                     if ( !m_Queue.push( value_type( *it ) ))
  69                         ++m_nPushError;
  70                 }
  71             }
  72
  73             void prepare( size_t nStart, size_t nEnd )
  74             {
  75                 m_arr.reserve( nEnd - nStart );
  76                 for ( size_t i = nStart; i < nEnd; ++i )
  77                     m_arr.push_back( i );
  78                 shuffle( m_arr.begin(), m_arr.end() );
  79             }
  80
  81         public:
  82             PQueue&             m_Queue;
  83             size_t              m_nPushError = 0;
  84
  85             typedef std::vector<size_t> array_type;
  86             array_type          m_arr;
  87         };
  88
  89         template <class PQueue>
  90         class Consumer: public cds_test::thread
  91         {
  92             typedef cds_test::thread base_class;
  93
  94         public:
  95             Consumer( cds_test::thread_pool& pool, PQueue& queue )
  96                 : base_class( pool )
  97                 , m_Queue( queue )
  98             {}
  99
 100             Consumer( Consumer& src )
 101                 : base_class( src )
 102                 , m_Queue( src.m_Queue )
 103             {}
 104
 105             virtual thread * clone()
 106             {
 107                 return new Consumer( *this );
 108             }
 109
 110             virtual void test()
 111             {
 112                 typedef typename PQueue::value_type value_type;
 113                 size_t nPrevKey;
 114                 value_type val;
 115                 if ( m_Queue.pop( val )) {
 116                     ++m_nPopSuccess;
 117                     nPrevKey = val.key;
 118
 119                     while ( !m_Queue.empty() ) {
 120                         if ( m_Queue.pop( val )) {
 121                             ++m_nPopSuccess;
 122                             if ( val.key > nPrevKey )
 123                                 ++m_nPopError;
 124                             else if ( val.key == nPrevKey )
 125                                 ++m_nPopErrorEq;
 126                             nPrevKey = val.key;
 127                         }
 128                         else
 129                             ++m_nPopFailed;
 130                     }
 131                 }
 132                 else
 133                     ++m_nPopFailed;
 134             }
 135
 136         public:
 137             PQueue&             m_Queue;
 138             size_t              m_nPopError = 0;
 139             size_t              m_nPopErrorEq = 0;
 140             size_t              m_nPopSuccess = 0;
 141             size_t              m_nPopFailed = 0;
 142         };
 143
 144     protected:
 145
 146         template <class PQueue>
 147         void test( PQueue& q )
 148         {
 149             size_t const nThreadItemCount = s_nQueueSize / s_nThreadCount;
 150             s_nQueueSize = nThreadItemCount * s_nThreadCount;
 151
 152             cds_test::thread_pool& pool = get_pool();
 153
 154             propout() << std::make_pair( "thread_count", s_nThreadCount )
 155                 << std::make_pair( "push_count", s_nQueueSize );
 156
 157             // push
 158             {
 159                 pool.add( new Producer<PQueue>( pool, q ), s_nThreadCount );
 160
 161                 size_t nStart = 0;
 162                 for ( size_t i = 0; i < pool.size(); ++i ) {
 163                     static_cast<Producer<PQueue>&>( pool.get(i) ).prepare( nStart, nStart + nThreadItemCount );
 164                     nStart += nThreadItemCount;
 165                 }
 166
 167                 std::chrono::milliseconds duration = pool.run();
 168                 propout() << std::make_pair( "producer_duration", duration );
 169             }
 170
 171             // pop
 172             {
 173                 pool.clear();
 174                 pool.add( new Consumer<PQueue>( pool, q ), s_nThreadCount );
 175
 176                 std::chrono::milliseconds duration = pool.run();
 177                 propout() << std::make_pair( "consumer_duration", duration );
 178
 179                 // Analyze result
 180                 size_t nTotalPopped = 0;
 181                 size_t nTotalError = 0;
 182                 size_t nTotalErrorEq = 0;
 183                 size_t nTotalFailed = 0;
 184                 for ( size_t i = 0; i < pool.size(); ++i ) {
 185                     Consumer<PQueue>& cons = static_cast<Consumer<PQueue>&>( pool.get(i));
 186
 187                     nTotalPopped  += cons.m_nPopSuccess;
 188                     nTotalError   += cons.m_nPopError;
 189                     nTotalErrorEq += cons.m_nPopErrorEq;
 190                     nTotalFailed  += cons.m_nPopFailed;
 191                 }
 192
 193                 propout()
 194                     << std::make_pair( "total_popped", nTotalPopped )
 195                     << std::make_pair( "error_pop_double", nTotalErrorEq )
 196                     << std::make_pair( "error_priority_violation", nTotalError );
 197
 198                 EXPECT_EQ( nTotalPopped, s_nQueueSize );
 199                 EXPECT_EQ( nTotalError, 0 );
 200                 EXPECT_EQ( nTotalErrorEq, 0 );
 201             }
 202
 203             propout() << q.statistics();
 204         }
 205
 206     public:
 207         static void SetUpTestCase()\r
 208         {\r
 209             cds_test::config const& cfg = get_config( "pqueue_pop" );\r
 210 \r
 211             s_nThreadCount = cfg.get_size_t( "ThreadCount", s_nThreadCount );
 212             s_nQueueSize = cfg.get_size_t( "QueueSize", s_nQueueSize );
 213
 214             if ( s_nThreadCount == 0 )
 215                 s_nThreadCount = 1;
 216             if ( s_nQueueSize == 0 )
 217                 s_nQueueSize = 1000;\r
 218         }
 219
 220         //static void TearDownTestCase();
 221     };
 222
 223 #define CDSSTRESS_MSPriorityQueue( fixture_t, pqueue_t ) \
 224     TEST_F( fixture_t, pqueue_t ) \
 225     { \
 226         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 227         pqueue_type pq( s_nQueueSize ); \
 228         test( pq ); \
 229     }
 230     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_less )
 231     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_less_stat )
 232     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_cmp )
 233     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_mutex )
 234
 235 #define CDSSTRESS_MSPriorityQueue_static( fixture_t, pqueue_t ) \
 236     TEST_F( fixture_t, pqueue_t ) \
 237     { \
 238         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 239         std::unique_ptr< pqueue_type > pq( new pqueue_type ); \
 240         test( *pq.get() ); \
 241     }
 242     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_less )
 243     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_less_stat )
 244     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_cmp )
 245     //CDSSTRESS_MSPriorityQueue( pqueue_pop, 1MSPriorityQueue_static_mutex )
 246
 247
 248 #define CDSSTRESS_PriorityQueue( fixture_t, pqueue_t ) \
 249     TEST_F( fixture_t, pqueue_t ) \
 250     { \
 251         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 252         pqueue_type pq; \
 253         test( pq ); \
 254     }
 255     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_vector )
 256     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_vector_stat )
 257     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_deque )
 258     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_deque_stat )
 259     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_boost_deque )
 260     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_boost_deque_stat )
 261     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_boost_stable_vector )
 262     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_boost_stable_vector_stat )
 263
 264     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_max )
 265     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_max_stat )
 266     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_min )
 267     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_min_stat )
 268     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_max )
 269     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_max_stat )
 270     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_min )
 271     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_min_stat )
 272     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_max )
 273     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_max_stat )
 274     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_min )
 275     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_min_stat )
 276     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_max )
 277     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_max_stat )
 278     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_min )
 279     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_min_stat )
 280     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_max )
 281     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_max_stat )
 282     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_min )
 283     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_min_stat )
 284 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
 285     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_max )
 286     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_max_stat )
 287     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_min )
 288     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_min_stat )
 289     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_sht_max )
 290     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_sht_max_stat )
 291     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_sht_min )
 292     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_sht_min_stat )
 293 #endif
 294
 295     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_HP_max )
 296     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_HP_max_stat )
 297     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_HP_min )
 298     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_HP_min_stat )
 299     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_DHP_max )
 300     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_DHP_max_stat )
 301     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_DHP_min )
 302     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_DHP_min_stat )
 303     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpi_max )
 304     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpi_min )
 305     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpb_max )
 306     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpb_min )
 307     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpt_max )
 308     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpt_min )
 309 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
 310     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_shb_max )
 311     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_shb_min )
 312     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_sht_max )
 313     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_sht_min )
 314 #endif
 315
 316     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_vector_spin )
 317     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_vector_mutex )
 318     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_deque_spin )
 319     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_deque_mutex )
 320
 321 } // namespace