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 j = 0; j < s_nQueueSize; ++j )
 135                     arr[j] += thread.m_arr[j];
 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_RWQueue( queue_pop )
 174
 175 #undef CDSSTRESS_Queue_F
 176 #define CDSSTRESS_Queue_F( test_fixture, type_name ) \
 177     TEST_F( test_fixture, type_name ) \
 178     { \
 179         typedef queue::Types< value_type >::type_name queue_type; \
 180         queue_type queue( s_nQueueSize ); \
 181         test( queue ); \
 182     }
 183
 184     CDSSTRESS_VyukovQueue( queue_pop )
 185
 186 #undef CDSSTRESS_Queue_F
 187
 188
 189     // ********************************************************************
 190     // SegmentedQueue test
 191
 192     class segmented_queue_pop
 193         : public queue_pop
 194         , public ::testing::WithParamInterface< size_t >
 195     {
 196         typedef queue_pop base_class;
 197
 198     protected:
 199         template <typename Queue>
 200         void test()
 201         {
 202             size_t quasi_factor = GetParam();
 203
 204             Queue q( quasi_factor );
 205             propout() << std::make_pair( "quasi_factor", quasi_factor );
 206             base_class::test( q );
 207         }
 208
 209     public:
 210         static std::vector< size_t > get_test_parameters()
 211         {
 212             cds_test::config const& cfg = cds_test::stress_fixture::get_config( "queue_pop" );
 213             bool bIterative = cfg.get_bool( "SegmentedQueue_Iterate", false );
 214             size_t quasi_factor = cfg.get_size_t( "SegmentedQueue_SegmentSize", 256 );
 215
 216             std::vector<size_t> args;
 217             if ( bIterative && quasi_factor > 4 ) {
 218                 for ( size_t qf = 4; qf <= quasi_factor; qf *= 2 )
 219                     args.push_back( qf );
 220             }
 221             else {
 222                 if ( quasi_factor > 2 )
 223                     args.push_back( quasi_factor );
 224                 else
 225                     args.push_back( 2 );
 226             }
 227
 228             return args;
 229         }
 230     };
 231
 232 #define CDSSTRESS_Queue_F( test_fixture, type_name ) \
 233     TEST_P( test_fixture, type_name ) \
 234     { \
 235         typedef typename queue::Types<value_type>::type_name queue_type; \
 236         test< queue_type >(); \
 237     }
 238
 239     CDSSTRESS_SegmentedQueue( segmented_queue_pop )
 240
 241 #ifdef CDSTEST_GTEST_INSTANTIATE_TEST_CASE_P_HAS_4TH_ARG
 242     static std::string get_test_parameter_name( testing::TestParamInfo<size_t> const& p )
 243     {
 244         return std::to_string( p.param );
 245     }
 246     INSTANTIATE_TEST_CASE_P( SQ,
 247         segmented_queue_pop,
 248         ::testing::ValuesIn( segmented_queue_pop::get_test_parameters()), get_test_parameter_name );
 249 #else
 250     INSTANTIATE_TEST_CASE_P( SQ,
 251         segmented_queue_pop,
 252         ::testing::ValuesIn( segmented_queue_pop::get_test_parameters()));
 253 #endif
 254
 255
 256 } // namespace