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