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