test/stress/queue/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 "queue_type.h"
  32
  33 // Multi-threaded queue test for pop operation
  34 namespace {
  35
  36     static size_t s_nThreadCount = 8;
  37     static size_t s_nQueueSize = 20000000 ;   // no more than 20 million records
  38
  39         struct SimpleValue {
  40             size_t    nNo;
  41
  42             SimpleValue(): nNo(0) {}
  43             SimpleValue( size_t n ): nNo(n) {}
  44             size_t getNo() const { return  nNo; }
  45         };
  46
  47     class queue_pop: public cds_test::stress_fixture
  48     {
  49     protected:
  50         struct value_type
  51         {
  52             size_t      nNo;
  53
  54             value_type()
  55                 : nNo( 0 )
  56             {}
  57
  58             value_type( size_t n )
  59                 : nNo( n )
  60             {}
  61         };
  62
  63         template <class Queue>
  64         class Consumer: public cds_test::thread
  65         {
  66             typedef cds_test::thread base_class;
  67
  68         public:
  69             Consumer( cds_test::thread_pool& pool, Queue& queue )
  70                 : base_class( pool )
  71                 , m_Queue( queue )
  72                 , m_arr( new uint8_t[ s_nQueueSize ])
  73                 , m_nPopCount( 0 )
  74             {}
  75
  76             Consumer( Consumer& src )
  77                 : base_class( src )
  78                 , m_Queue( src.m_Queue )
  79                 , m_arr( new uint8_t[ s_nQueueSize ])
  80                 , m_nPopCount( 0 )
  81             {}
  82
  83             virtual thread * clone()
  84             {
  85                 return new Consumer( *this );
  86             }
  87
  88             virtual void test()
  89             {
  90                 memset( m_arr.get(), 0, sizeof( m_arr[0] ) * s_nQueueSize );
  91                 typedef typename Queue::value_type value_type;
  92                 value_type value;
  93                 size_t nPopCount = 0;
  94                 while ( m_Queue.pop( value )) {
  95                     ++m_arr[ value.nNo ];
  96                     ++nPopCount;
  97                 }
  98                 m_nPopCount = nPopCount;
  99             }
 100
 101         public:
 102             Queue&              m_Queue;
 103             std::unique_ptr< uint8_t[] > m_arr;
 104             size_t              m_nPopCount;
 105         };
 106
 107     public:
 108         static void SetUpTestCase()
 109         {
 110             cds_test::config const& cfg = get_config( "queue_pop" );
 111
 112             s_nThreadCount = cfg.get_size_t( "ThreadCount", s_nThreadCount );
 113             s_nQueueSize = cfg.get_size_t( "QueueSize", s_nQueueSize );
 114
 115             if ( s_nThreadCount == 0 )
 116                 s_nThreadCount = 1;
 117             if ( s_nQueueSize == 0 )
 118                 s_nQueueSize = 1000;
 119         }
 120
 121         //static void TearDownTestCase();
 122
 123     protected:
 124         template <class Queue>
 125         void analyze( Queue& q )
 126         {
 127             cds_test::thread_pool& pool = get_pool();
 128             std::unique_ptr< uint8_t[] > arr( new uint8_t[s_nQueueSize] );
 129             memset(arr.get(), 0, sizeof(arr[0]) * s_nQueueSize );
 130
 131             size_t nTotalPops = 0;
 132             for ( size_t i = 0; i < pool.size(); ++i ) {
 133                 Consumer<Queue>& thread = static_cast<Consumer<Queue>&>(pool.get( i ));
 134                 for ( size_t i = 0; i < s_nQueueSize; ++i )
 135                     arr[i] += thread.m_arr[i];
 136                 nTotalPops += thread.m_nPopCount;
 137             }
 138             EXPECT_EQ( nTotalPops, s_nQueueSize );
 139             EXPECT_TRUE( q.empty());
 140
 141             for ( size_t i = 0; i < s_nQueueSize; ++i ) {
 142                 EXPECT_EQ( arr[i], 1 ) << "i=" << i;
 143             }
 144         }
 145
 146         template <class Queue>
 147         void test( Queue& q )
 148         {
 149             cds_test::thread_pool& pool = get_pool();
 150
 151             pool.add( new Consumer<Queue>( pool, q ), s_nThreadCount );
 152
 153             for ( size_t i = 0; i < s_nQueueSize; ++i )
 154                 q.push( i );
 155
 156             propout() << std::make_pair( "thread_count", s_nThreadCount )
 157                 << std::make_pair( "push_count", s_nQueueSize );
 158
 159             std::chrono::milliseconds duration = pool.run();
 160
 161             propout() << std::make_pair( "duration", duration );
 162
 163             analyze( q );
 164
 165             propout() << q.statistics();
 166         }
 167     };
 168
 169     CDSSTRESS_MSQueue( queue_pop )
 170     CDSSTRESS_MoirQueue( queue_pop )
 171     CDSSTRESS_BasketQueue( queue_pop )
 172     CDSSTRESS_OptimsticQueue( queue_pop )
 173     CDSSTRESS_FCQueue( queue_pop )
 174     CDSSTRESS_FCDeque( queue_pop )
 175     CDSSTRESS_RWQueue( queue_pop )
 176     CDSSTRESS_StdQueue( queue_pop )
 177
 178 #undef CDSSTRESS_Queue_F
 179 #define CDSSTRESS_Queue_F( test_fixture, type_name, level ) \
 180     TEST_F( test_fixture, type_name ) \
 181     { \
 182         if ( !check_detail_level( level )) return; \
 183         typedef queue::Types< value_type >::type_name queue_type; \
 184         queue_type queue( s_nQueueSize ); \
 185         test( queue ); \
 186     }
 187
 188     CDSSTRESS_TsigasQueue( queue_pop )
 189     CDSSTRESS_VyukovQueue( queue_pop )
 190
 191 #undef CDSSTRESS_Queue_F
 192
 193
 194     // ********************************************************************
 195     // SegmentedQueue test
 196
 197     class segmented_queue_pop
 198         : public queue_pop
 199         , public ::testing::WithParamInterface< size_t >
 200     {
 201         typedef queue_pop base_class;
 202
 203     protected:
 204         template <typename Queue>
 205         void test()
 206         {
 207             size_t quasi_factor = GetParam();
 208
 209             Queue q( quasi_factor );
 210             propout() << std::make_pair( "quasi_factor", quasi_factor );
 211             base_class::test( q );
 212         }
 213
 214     public:
 215         static std::vector< size_t > get_test_parameters()
 216         {
 217             cds_test::config const& cfg = cds_test::stress_fixture::get_config( "queue_pop" );
 218             bool bIterative = cfg.get_bool( "SegmentedQueue_Iterate", false );
 219             size_t quasi_factor = cfg.get_size_t( "SegmentedQueue_SegmentSize", 256 );
 220
 221             std::vector<size_t> args;
 222             if ( bIterative && quasi_factor > 4 ) {
 223                 for ( size_t qf = 4; qf <= quasi_factor; qf *= 2 )
 224                     args.push_back( qf );
 225             }
 226             else {
 227                 if ( quasi_factor > 2 )
 228                     args.push_back( quasi_factor );
 229                 else
 230                     args.push_back( 2 );
 231             }
 232
 233             return args;
 234         }
 235     };
 236
 237 #define CDSSTRESS_Queue_F( test_fixture, type_name, level ) \
 238     TEST_P( test_fixture, type_name ) \
 239     { \
 240         if ( !check_detail_level( level )) return; \
 241         typedef typename queue::Types<value_type>::type_name queue_type; \
 242         test< queue_type >(); \
 243     }
 244
 245     CDSSTRESS_SegmentedQueue( segmented_queue_pop )
 246
 247     INSTANTIATE_TEST_CASE_P( SQ,
 248         segmented_queue_pop,
 249         ::testing::ValuesIn( segmented_queue_pop::get_test_parameters()));
 250
 251 } // namespace