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-2017
   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                     bool prevPopFailed = false;
  74                     while ( 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, static_cast<size_t>(-1) } );
  79                             prevPopFailed = true;
  80                         }
  81                         else if ( val.key == nPrevKey ) {
  82                             ++m_nPopErrorEq;
  83                             m_arrFailedPops.emplace_back( failed_pops{ nPrevKey, val.key, static_cast<size_t>(-1) } );
  84                         }
  85                         else {
  86                             if ( prevPopFailed )
  87                                 m_arrFailedPops.back().next_key = val.key;
  88                             prevPopFailed = false;
  89                         }
  90                         if ( nPrevKey > val.key )
  91                             nPrevKey = val.key;
  92                     }
  93
  94                 }
  95                 else
  96                     ++m_nPopFailed;
  97             }
  98
  99         public:
 100             PQueue&             m_Queue;
 101             size_t              m_nPopError = 0;
 102             size_t              m_nPopErrorEq = 0;
 103             size_t              m_nPopSuccess = 0;
 104             size_t              m_nPopFailed = 0;
 105
 106             struct failed_pops {
 107                 size_t prev_key;
 108                 size_t popped_key;
 109                 size_t next_key;
 110             };
 111             std::vector< failed_pops > m_arrFailedPops;
 112         };
 113
 114     protected:
 115
 116         template <class PQueue>
 117         void test( PQueue& q )
 118         {
 119             cds_test::thread_pool& pool = get_pool();
 120
 121             // push
 122             {
 123                 std::vector< size_t > arr;
 124                 arr.reserve( s_nQueueSize );
 125                 for ( size_t i = 0; i < s_nQueueSize; ++i )
 126                     arr.push_back( i );
 127                 shuffle( arr.begin(), arr.end());
 128
 129                 size_t nPushError = 0;
 130                 typedef typename PQueue::value_type value_type;
 131                 for ( auto it = arr.begin(); it != arr.end(); ++it ) {
 132                     if ( !q.push( value_type( *it )))
 133                         ++nPushError;
 134                 }
 135                 s_nQueueSize -= nPushError;
 136             }
 137
 138             propout() << std::make_pair( "thread_count", s_nThreadCount )
 139                 << std::make_pair( "push_count", s_nQueueSize );
 140
 141             // pop
 142             {
 143                 pool.add( new Consumer<PQueue>( pool, q ), s_nThreadCount );
 144
 145                 std::chrono::milliseconds duration = pool.run();
 146                 propout() << std::make_pair( "consumer_duration", duration );
 147
 148                 // Analyze result
 149                 size_t nTotalPopped = 0;
 150                 size_t nTotalError = 0;
 151                 size_t nTotalErrorEq = 0;
 152                 size_t nTotalFailed = 0;
 153                 for ( size_t i = 0; i < pool.size(); ++i ) {
 154                     Consumer<PQueue>& cons = static_cast<Consumer<PQueue>&>( pool.get(i));
 155
 156                     nTotalPopped  += cons.m_nPopSuccess;
 157                     nTotalError   += cons.m_nPopError;
 158                     nTotalErrorEq += cons.m_nPopErrorEq;
 159                     nTotalFailed  += cons.m_nPopFailed;
 160
 161                     if ( !cons.m_arrFailedPops.empty()) {
 162                         std::cerr << "Priority violations, thread " << i;
 163                         for ( size_t k = 0; k < cons.m_arrFailedPops.size(); ++k ) {
 164                             std::cerr << "\n    " << "prev_key=" << cons.m_arrFailedPops[k].prev_key << " popped_key=" << cons.m_arrFailedPops[k].popped_key;
 165                             if ( cons.m_arrFailedPops[k].next_key != static_cast<size_t>(-1))
 166                                 std::cerr << " next_key=" << cons.m_arrFailedPops[k].next_key;
 167                             else
 168                                 std::cerr << " next_key unspecified";
 169                         }
 170                         std::cerr << std::endl;
 171                     }
 172                 }
 173
 174                 propout()
 175                     << std::make_pair( "total_popped", nTotalPopped )
 176                     << std::make_pair( "error_pop_double", nTotalErrorEq )
 177                     << std::make_pair( "error_priority_violation", nTotalError );
 178
 179                 EXPECT_EQ( nTotalPopped, s_nQueueSize );
 180                 EXPECT_EQ( nTotalError, 0u ) << "priority violations";
 181                 EXPECT_EQ( nTotalErrorEq, 0u ) << "double key";
 182             }
 183
 184             propout() << q.statistics();
 185         }
 186
 187     public:
 188         static void SetUpTestCase()
 189         {
 190             cds_test::config const& cfg = get_config( "pqueue_pop" );
 191
 192             s_nThreadCount = cfg.get_size_t( "ThreadCount", s_nThreadCount );
 193             s_nQueueSize = cfg.get_size_t( "QueueSize", s_nQueueSize );
 194
 195             if ( s_nThreadCount == 0u )
 196                 s_nThreadCount = 1;
 197             if ( s_nQueueSize == 0u )
 198                 s_nQueueSize = 1000;
 199         }
 200
 201         //static void TearDownTestCase();
 202     };
 203
 204 #define CDSSTRESS_MSPriorityQueue( fixture_t, pqueue_t ) \
 205     TEST_F( fixture_t, pqueue_t ) \
 206     { \
 207         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 208         pqueue_type pq( s_nQueueSize + 1 ); \
 209         test( pq ); \
 210     }
 211     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_less )
 212     CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_cmp )
 213     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_dyn_mutex ) // too slow
 214
 215 #define CDSSTRESS_MSPriorityQueue_static( fixture_t, pqueue_t ) \
 216     TEST_F( fixture_t, pqueue_t ) \
 217     { \
 218         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 219         std::unique_ptr< pqueue_type > pq( new pqueue_type ); \
 220         test( *pq.get()); \
 221     }
 222     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_less )
 223     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_less_stat )
 224     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_cmp )
 225     //CDSSTRESS_MSPriorityQueue( pqueue_pop, MSPriorityQueue_static_mutex )
 226
 227
 228 #define CDSSTRESS_PriorityQueue( fixture_t, pqueue_t ) \
 229     TEST_F( fixture_t, pqueue_t ) \
 230     { \
 231         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 232         pqueue_type pq; \
 233         test( pq ); \
 234     }
 235     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_max )
 236     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_HP_min )
 237     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_max )
 238     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_DHP_min )
 239     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_max )
 240     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_max_stat )
 241     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_min )
 242     // CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpi_min_stat )
 243     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_max )
 244     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpb_min )
 245     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_max )
 246     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_gpt_min )
 247 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
 248     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_max )
 249     CDSSTRESS_PriorityQueue( pqueue_pop, EllenBinTree_RCU_shb_min )
 250 #endif
 251
 252     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_HP_max )
 253     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_HP_min )
 254     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_DHP_max )
 255     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_DHP_min )
 256     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_RCU_gpi_max )
 257     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_RCU_gpi_min )
 258     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_RCU_gpb_max )
 259     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_RCU_gpb_min )
 260     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_RCU_gpt_max )
 261     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_RCU_gpt_min )
 262 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
 263     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_RCU_shb_max )
 264     CDSSTRESS_PriorityQueue( pqueue_pop, SkipList32_RCU_shb_min )
 265 #endif
 266
 267     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_vector_spin )
 268     CDSSTRESS_PriorityQueue( pqueue_pop, StdPQueue_deque_spin )
 269
 270 } // namespace