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-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 "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_VyukovQueue( queue_pop )
 189
 190 #undef CDSSTRESS_Queue_F
 191
 192
 193     // ********************************************************************
 194     // SegmentedQueue test
 195
 196     class segmented_queue_pop
 197         : public queue_pop
 198         , public ::testing::WithParamInterface< size_t >
 199     {
 200         typedef queue_pop base_class;
 201
 202     protected:
 203         template <typename Queue>
 204         void test()
 205         {
 206             size_t quasi_factor = GetParam();
 207
 208             Queue q( quasi_factor );
 209             propout() << std::make_pair( "quasi_factor", quasi_factor );
 210             base_class::test( q );
 211         }
 212
 213     public:
 214         static std::vector< size_t > get_test_parameters()
 215         {
 216             cds_test::config const& cfg = cds_test::stress_fixture::get_config( "queue_pop" );
 217             bool bIterative = cfg.get_bool( "SegmentedQueue_Iterate", false );
 218             size_t quasi_factor = cfg.get_size_t( "SegmentedQueue_SegmentSize", 256 );
 219
 220             std::vector<size_t> args;
 221             if ( bIterative && quasi_factor > 4 ) {
 222                 for ( size_t qf = 4; qf <= quasi_factor; qf *= 2 )
 223                     args.push_back( qf );
 224             }
 225             else {
 226                 if ( quasi_factor > 2 )
 227                     args.push_back( quasi_factor );
 228                 else
 229                     args.push_back( 2 );
 230             }
 231
 232             return args;
 233         }
 234     };
 235
 236 #define CDSSTRESS_Queue_F( test_fixture, type_name, level ) \
 237     TEST_P( test_fixture, type_name ) \
 238     { \
 239         if ( !check_detail_level( level )) return; \
 240         typedef typename queue::Types<value_type>::type_name queue_type; \
 241         test< queue_type >(); \
 242     }
 243
 244     CDSSTRESS_SegmentedQueue( segmented_queue_pop )
 245
 246     INSTANTIATE_TEST_CASE_P( SQ,
 247         segmented_queue_pop,
 248         ::testing::ValuesIn( segmented_queue_pop::get_test_parameters()));
 249
 250 } // namespace