test/stress/pqueue/push.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 pqueue {
  35
  36     static size_t s_nThreadCount = 8;
  37     static size_t s_nQueueSize = 2000000;
  38
  39     class pqueue_push: public cds_test::stress_fixture
  40     {
  41         typedef cds_test::stress_fixture base_class;
  42
  43     protected:
  44         template <class PQueue>
  45         class Producer: public cds_test::thread
  46         {
  47             typedef cds_test::thread base_class;
  48
  49         public:
  50             Producer( cds_test::thread_pool& pool, PQueue& queue )
  51                 : base_class( pool )
  52                 , m_Queue( queue )
  53             {}
  54
  55             Producer( Producer& src )
  56                 : base_class( src )
  57                 , m_Queue( src.m_Queue )
  58             {}
  59
  60             virtual thread * clone()
  61             {
  62                 return new Producer( *this );
  63             }
  64
  65             virtual void test()
  66             {
  67                 typedef typename PQueue::value_type value_type;
  68
  69                 for ( auto it = m_arr.begin(); it != m_arr.end(); ++it ) {
  70                     if ( !m_Queue.push( value_type( *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                 shuffle( m_arr.begin(), m_arr.end());
  81             }
  82
  83         public:
  84             PQueue&             m_Queue;
  85             size_t              m_nPushError = 0;
  86             std::vector<size_t> m_arr;
  87         };
  88
  89     protected:
  90         template <class PQueue>
  91         void analyze( PQueue& q )
  92         {
  93             cds_test::thread_pool& pool = get_pool();
  94
  95             for ( size_t i = 0; i < pool.size(); ++i ) {
  96                 Producer<PQueue>& prod = static_cast<Producer<PQueue>&>(pool.get( i ));
  97                 EXPECT_EQ( prod.m_nPushError, 0u ) << "producer=" << i;
  98             }
  99             EXPECT_FALSE( q.empty());
 100
 101             typedef std::vector<size_t> vector_type;
 102             vector_type arr;
 103             arr.reserve( s_nQueueSize );
 104
 105             size_t nPopped = 0;
 106             typename PQueue::value_type val;
 107             while ( q.pop( val )) {
 108                 nPopped++;
 109                 arr.push_back( val.key );
 110             }
 111
 112             EXPECT_EQ( arr.size(), s_nQueueSize );
 113             auto it = arr.begin();
 114             size_t nPrev = *it;
 115             ++it;
 116             for ( auto itEnd = arr.end(); it != itEnd; ++it ) {
 117                 EXPECT_EQ( nPrev - 1, *it );
 118                 nPrev = *it;
 119             }
 120         }
 121
 122         template <class PQueue>
 123         void test( PQueue& q )
 124         {
 125             cds_test::thread_pool& pool = get_pool();
 126             pool.add( new Producer<PQueue>( pool, q ), s_nThreadCount );
 127
 128             size_t nStart = 0;
 129             size_t nThreadItemCount = s_nQueueSize / s_nThreadCount;
 130             s_nQueueSize = nThreadItemCount * s_nThreadCount;
 131
 132             for ( size_t i = 0; i < pool.size(); ++i ) {
 133                 static_cast<Producer<PQueue>&>(pool.get( i )).prepare( nStart, nStart + nThreadItemCount );
 134                 nStart += nThreadItemCount;
 135             }
 136
 137             propout() << std::make_pair( "thread_count", s_nThreadCount )
 138                 << std::make_pair( "push_count", s_nQueueSize );
 139
 140             std::chrono::milliseconds duration = pool.run();
 141             propout() << std::make_pair( "duration", duration );
 142
 143             analyze( q );
 144
 145             propout() << q.statistics();
 146         }
 147
 148     public:
 149         static void SetUpTestCase()
 150         {
 151             cds_test::config const& cfg = get_config( "pqueue_push" );
 152
 153             s_nThreadCount = cfg.get_size_t( "ThreadCount", s_nThreadCount );
 154             s_nQueueSize = cfg.get_size_t( "QueueSize", s_nQueueSize );
 155
 156             if ( s_nThreadCount == 0u )
 157                 s_nThreadCount = 1;
 158             if ( s_nQueueSize == 0u )
 159                 s_nQueueSize = 1000;
 160         }
 161
 162         //static void TearDownTestCase();
 163     };
 164
 165 #define CDSSTRESS_MSPriorityQueue( fixture_t, pqueue_t ) \
 166     TEST_F( fixture_t, pqueue_t ) \
 167     { \
 168         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 169         pqueue_type pq( s_nQueueSize ); \
 170         test( pq ); \
 171     }
 172     CDSSTRESS_MSPriorityQueue( pqueue_push, MSPriorityQueue_dyn_less )
 173     CDSSTRESS_MSPriorityQueue( pqueue_push, MSPriorityQueue_dyn_less_stat )
 174     CDSSTRESS_MSPriorityQueue( pqueue_push, MSPriorityQueue_dyn_cmp )
 175     //CDSSTRESS_MSPriorityQueue( pqueue_push, MSPriorityQueue_dyn_mutex ) // too slow
 176
 177 #define CDSSTRESS_MSPriorityQueue_static( fixture_t, pqueue_t ) \
 178     TEST_F( fixture_t, pqueue_t ) \
 179     { \
 180         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 181         std::unique_ptr< pqueue_type > pq( new pqueue_type ); \
 182         test( *pq.get()); \
 183     }
 184     //CDSSTRESS_MSPriorityQueue( pqueue_push, MSPriorityQueue_static_less )
 185     //CDSSTRESS_MSPriorityQueue( pqueue_push, MSPriorityQueue_static_less_stat )
 186     //CDSSTRESS_MSPriorityQueue( pqueue_push, MSPriorityQueue_static_cmp )
 187     //CDSSTRESS_MSPriorityQueue( pqueue_push, MSPriorityQueue_static_mutex )
 188
 189
 190 #define CDSSTRESS_PriorityQueue( fixture_t, pqueue_t ) \
 191     TEST_F( fixture_t, pqueue_t ) \
 192     { \
 193         typedef pqueue::Types<pqueue::simple_value>::pqueue_t pqueue_type; \
 194         pqueue_type pq; \
 195         test( pq ); \
 196     }
 197     CDSSTRESS_PriorityQueue( pqueue_push, FCPQueue_vector )
 198     CDSSTRESS_PriorityQueue( pqueue_push, FCPQueue_vector_stat )
 199     CDSSTRESS_PriorityQueue( pqueue_push, FCPQueue_deque )
 200     CDSSTRESS_PriorityQueue( pqueue_push, FCPQueue_deque_stat )
 201     CDSSTRESS_PriorityQueue( pqueue_push, FCPQueue_boost_deque )
 202     CDSSTRESS_PriorityQueue( pqueue_push, FCPQueue_boost_deque_stat )
 203     CDSSTRESS_PriorityQueue( pqueue_push, FCPQueue_boost_stable_vector )
 204     CDSSTRESS_PriorityQueue( pqueue_push, FCPQueue_boost_stable_vector_stat )
 205
 206     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_HP_max )
 207     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_HP_max_stat )
 208     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_HP_min )
 209     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_HP_min_stat )
 210     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_DHP_max )
 211     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_DHP_max_stat )
 212     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_DHP_min )
 213     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_DHP_min_stat )
 214     // CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpi_max )
 215     // CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpi_max_stat )
 216     // CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpi_min )
 217     // CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpi_min_stat )
 218     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpb_max )
 219     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpb_max_stat )
 220     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpb_min )
 221     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpb_min_stat )
 222     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpt_max )
 223     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpt_max_stat )
 224     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpt_min )
 225     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_gpt_min_stat )
 226 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
 227     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_shb_max )
 228     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_shb_max_stat )
 229     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_shb_min )
 230     CDSSTRESS_PriorityQueue( pqueue_push, EllenBinTree_RCU_shb_min_stat )
 231 #endif
 232
 233     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_HP_max )
 234     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_HP_max_stat )
 235     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_HP_min )
 236     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_HP_min_stat )
 237     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_DHP_max )
 238     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_DHP_max_stat )
 239     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_DHP_min )
 240     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_DHP_min_stat )
 241     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_RCU_gpi_max )
 242     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_RCU_gpi_min )
 243     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_RCU_gpb_max )
 244     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_RCU_gpb_min )
 245     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_RCU_gpt_max )
 246     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_RCU_gpt_min )
 247 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
 248     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_RCU_shb_max )
 249     CDSSTRESS_PriorityQueue( pqueue_push, SkipList_RCU_shb_min )
 250 #endif
 251
 252     CDSSTRESS_PriorityQueue( pqueue_push, StdPQueue_vector_spin )
 253     CDSSTRESS_PriorityQueue( pqueue_push, StdPQueue_vector_mutex )
 254     CDSSTRESS_PriorityQueue( pqueue_push, StdPQueue_deque_spin )
 255     CDSSTRESS_PriorityQueue( pqueue_push, StdPQueue_deque_mutex )
 256
 257 } // namespace queue
 258