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 Consumer: public cds_test::thread
  45         {
  46             typedef cds_test::thread base_class;
  47
  48         public:
  49             Consumer( cds_test::thread_pool& pool, PQueue& queue )
  50                 : base_class( pool )
  51                 , m_Queue( queue )
  52             {}
  53
  54             Consumer( Consumer& src )
  55                 : base_class( src )
  56                 , m_Queue( src.m_Queue )
  57             {}
  58
  59             virtual thread * clone()
  60             {
  61                 return new Consumer( *this );
  62             }
  63
  64             virtual void test()
  65             {
  66                 typedef typename PQueue::value_type value_type;
  67                 size_t nPrevKey;
  68                 value_type val;
  69                 if ( m_Queue.pop( val )) {
  70                     ++m_nPopSuccess;
  71                     nPrevKey = val.key;
  72
  73                     while ( !m_Queue.empty() ) {
  74                         if ( m_Queue.pop( val )) {
  75                             ++m_nPopSuccess;
  76                             if ( val.key > nPrevKey ) {
  77                                 ++m_nPopError;
  78                                 m_arrFailedPops.emplace_back( failed_pops{ nPrevKey, val.key } );
  79                             }
  80                             else if ( val.key == nPrevKey ) {
  81                                 ++m_nPopErrorEq;
  82                                 m_arrFailedPops.emplace_back( failed_pops{ nPrevKey, val.key } );
  83                             }
  84                             nPrevKey = val.key;
  85                         }
  86                         else
  87                             ++m_nPopFailed;
  88                     }
  89                 }
  90                 else
  91                     ++m_nPopFailed;
  92             }
  93
  94         public:
  95             PQueue&             m_Queue;
  96             size_t              m_nPopError = 0;
  97             size_t              m_nPopErrorEq = 0;
  98             size_t              m_nPopSuccess = 0;
  99             size_t              m_nPopFailed = 0;
 100
 101             struct failed_pops {
 102                 size_t prev_key;
 103                 size_t popped_key;
 104             };
 105             std::vector< failed_pops > m_arrFailedPops;
 106         };
 107
 108     protected:
 109
 110         template <class PQueue>
 111         void test( PQueue& q )
 112         {
 113             cds_test::thread_pool& pool = get_pool();
 114
 115             propout() << std::make_pair( "thread_count", s_nThreadCount )
 116                 << std::make_pair( "push_count", s_nQueueSize );
 117
 118             // push
 119             {
 120                 std::vector< size_t > arr;
 121                 arr.reserve( s_nQueueSize );
 122                 for ( size_t i = 0; i < s_nQueueSize; ++i )
 123                     arr.push_back( i );
 124                 shuffle( arr.begin(), arr.end() );
 125
 126                 size_t nPushError = 0;
 127                 typedef typename PQueue::value_type value_type;
 128                 for ( auto it = arr.begin(); it != arr.end(); ++it ) {
 129                     if ( !q.push( value_type( *it ) ))
 130                         ++nPushError;
 131                 }
 132                 s_nQueueSize -= nPushError;
 133             }
 134
 135             // pop
 136             {
 137                 pool.add( new Consumer<PQueue>( pool, q ), s_nThreadCount );
 138
 139                 std::chrono::milliseconds duration = pool.run();
 140                 propout() << std::make_pair( "consumer_duration", duration );
 141
 142                 // Analyze result
 143                 size_t nTotalPopped = 0;
 144                 size_t nTotalError = 0;
 145                 size_t nTotalErrorEq = 0;
 146                 size_t nTotalFailed = 0;
 147                 for ( size_t i = 0; i < pool.size(); ++i ) {
 148                     Consumer<PQueue>& cons = static_cast<Consumer<PQueue>&>( pool.get(i));
 149
 150                     nTotalPopped  += cons.m_nPopSuccess;
 151                     nTotalError   += cons.m_nPopError;
 152                     nTotalErrorEq += cons.m_nPopErrorEq;
 153                     nTotalFailed  += cons.m_nPopFailed;
 154
 155                     if ( !cons.m_arrFailedPops.empty() ) {
 156                         std::cerr << "Priority violations, thread " << i;
 157                         for ( size_t k = 0; k < cons.m_arrFailedPops.size(); ++k ) {
 158                             std::cerr << "\n    " << "prev_key=" << cons.m_arrFailedPops[k].prev_key << " popped_key=" << cons.m_arrFailedPops[k].popped_key;
 159                         }
 160                         std::cerr << std::endl;
 161                     }
 162                 }
 163
 164                 propout()
 165                     << std::make_pair( "total_popped", nTotalPopped )
 166                     << std::make_pair( "error_pop_double", nTotalErrorEq )
 167                     << std::make_pair( "error_priority_violation", nTotalError );
 168
 169                 EXPECT_EQ( nTotalPopped, s_nQueueSize );
 170                 EXPECT_EQ( nTotalError, 0 ) << "priority violations";
 171                 EXPECT_EQ( nTotalErrorEq, 0 ) << "double key";
 172             }
 173
 174             propout() << q.statistics();
 175         }
 176
 177     public:
 178         static void SetUpTestCase()\r
 179         {\r
 180             cds_test::config const& cfg = get_config( "pqueue_pop" );\r
 181 \r
 182             s_nThreadCount = cfg.get_size_t( "ThreadCount", s_nThreadCount );
 183             s_nQueueSize = cfg.get_size_t( "QueueSize", s_nQueueSize );
 184
 185             if ( s_nThreadCount == 0 )
 186                 s_nThreadCount = 1;
 187             if ( s_nQueueSize == 0 )
 188                 s_nQueueSize = 1000;\r
 189         }
 190
 191         //static void TearDownTestCase();
 192     };
 193
 194 #define CDSSTRESS_MSPriorityQueue( fixture_t, pqueue_t ) \
 195     TEST_F( fixture_t, pqueue_t ) \
 196     { \
 197         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 198         pqueue_type pq( s_nQueueSize + 1 ); \
 199         test( pq ); \
 200     }
 201     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_less )
 202     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_less_stat )
 203     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_cmp )
 204     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_mutex ) // too slow
 205
 206 #define CDSSTRESS_MSPriorityQueue_static( fixture_t, pqueue_t ) \
 207     TEST_F( fixture_t, pqueue_t ) \
 208     { \
 209         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 210         std::unique_ptr< pqueue_type > pq( new pqueue_type ); \
 211         test( *pq.get() ); \
 212     }
 213     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_less )
 214     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_less_stat )
 215     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_cmp )
 216     //CDSSTRESS_MSPriorityQueue( pqueue_pop, 1MSPriorityQueue_static_mutex )
 217
 218
 219 #define CDSSTRESS_PriorityQueue( fixture_t, pqueue_t ) \
 220     TEST_F( fixture_t, pqueue_t ) \
 221     { \
 222         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 223         pqueue_type pq; \
 224         test( pq ); \
 225     }
 226     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_vector )
 227     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_vector_stat )
 228     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_deque )
 229     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_deque_stat )
 230     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_boost_deque )
 231     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_boost_deque_stat )
 232     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_boost_stable_vector )
 233     CDSSTRESS_PriorityQueue( pqueue_pop, FCPQueue_boost_stable_vector_stat )
 234
 235     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_max )
 236     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_max_stat )
 237     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_min )
 238     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_min_stat )
 239     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_max )
 240     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_max_stat )
 241     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_min )
 242     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_min_stat )
 243     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_max )
 244     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_max_stat )
 245     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_min )
 246     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_min_stat )
 247     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_max )
 248     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_max_stat )
 249     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_min )
 250     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_min_stat )
 251     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_max )
 252     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_max_stat )
 253     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_min )
 254     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_min_stat )
 255 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
 256     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_max )
 257     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_max_stat )
 258     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_min )
 259     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_min_stat )
 260     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_sht_max )
 261     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_sht_max_stat )
 262     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_sht_min )
 263     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_sht_min_stat )
 264 #endif
 265
 266     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_HP_max )
 267     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_HP_max_stat )
 268     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_HP_min )
 269     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_HP_min_stat )
 270     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_DHP_max )
 271     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_DHP_max_stat )
 272     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_DHP_min )
 273     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_DHP_min_stat )
 274     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpi_max )
 275     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpi_min )
 276     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpb_max )
 277     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpb_min )
 278     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpt_max )
 279     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_gpt_min )
 280 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
 281     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_shb_max )
 282     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_shb_min )
 283     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_sht_max )
 284     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList_RCU_sht_min )
 285 #endif
 286
 287     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_vector_spin )
 288     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_vector_mutex )
 289     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_deque_spin )
 290     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_deque_mutex )
 291
 292 } // namespace