test/stress/queue/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 "queue_type.h"
  32
  33 // Multi-threaded queue test for push 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     class queue_push: public cds_test::stress_fixture
  40     {
  41     protected:
  42         struct value_type
  43         {
  44             size_t      nNo;
  45
  46             value_type()
  47                 : nNo( 0 )
  48             {}
  49
  50             value_type( size_t n )
  51                 : nNo( n )
  52             {}
  53         };
  54
  55         template <class Queue>
  56         class Producer: public cds_test::thread
  57         {
  58             typedef cds_test::thread base_class;
  59
  60         public:
  61             Producer( cds_test::thread_pool& pool, Queue& queue )
  62                 : base_class( pool )
  63                 , m_Queue( queue )
  64                 , m_nStartItem( 0 )
  65                 , m_nEndItem( 0 )
  66                 , m_nPushError( 0 )
  67             {}
  68
  69             Producer( Producer& src )
  70                 : base_class( src )
  71                 , m_Queue( src.m_Queue )
  72                 , m_nStartItem( 0 )
  73                 , m_nEndItem( 0 )
  74                 , m_nPushError( 0 )
  75             {}
  76
  77             virtual thread * clone()
  78             {
  79                 return new Producer( *this );
  80             }
  81
  82             virtual void test()
  83             {
  84                 for ( size_t nItem = m_nStartItem; nItem < m_nEndItem; ++nItem ) {
  85                     if ( !m_Queue.push( nItem ))
  86                         ++m_nPushError;
  87                 }
  88             }
  89
  90         public:
  91             Queue&              m_Queue;
  92             size_t              m_nStartItem;
  93             size_t              m_nEndItem;
  94             size_t              m_nPushError;
  95         };
  96
  97     public:
  98         static void SetUpTestCase()
  99         {
 100             cds_test::config const& cfg = get_config( "queue_push" );
 101
 102             s_nThreadCount = cfg.get_size_t( "ThreadCount", s_nThreadCount );
 103             s_nQueueSize = cfg.get_size_t( "QueueSize", s_nQueueSize );
 104
 105             if ( s_nThreadCount == 0u )
 106                 s_nThreadCount = 1;
 107             if ( s_nQueueSize == 0u )
 108                 s_nQueueSize = 1000;
 109         }
 110
 111         //static void TearDownTestCase();
 112
 113     protected:
 114         template <class Queue>
 115         void test( Queue& q )
 116         {
 117             cds_test::thread_pool& pool = get_pool();
 118
 119             pool.add( new Producer<Queue>( pool, q ), s_nThreadCount );
 120
 121             size_t nStart = 0;
 122             size_t nThreadItemCount = s_nQueueSize / s_nThreadCount;
 123             for ( size_t i = 0; i < pool.size(); ++i ) {
 124                 Producer<Queue>& thread = static_cast<Producer<Queue>&>(pool.get( i ));
 125                 thread.m_nStartItem = nStart;
 126                 nStart += nThreadItemCount;
 127                 thread.m_nEndItem = nStart;
 128             }
 129
 130             s_nQueueSize = nThreadItemCount * s_nThreadCount;
 131             propout() << std::make_pair( "thread_count", s_nThreadCount )
 132                 << std::make_pair( "push_count", s_nQueueSize );
 133
 134             std::chrono::milliseconds duration = pool.run();
 135
 136             propout() << std::make_pair( "duration", duration );
 137
 138             analyze( q );
 139
 140             propout() << q.statistics();
 141         }
 142
 143         template <class Queue>
 144         void analyze( Queue& q )
 145         {
 146             size_t nThreadItems = s_nQueueSize / s_nThreadCount;
 147             cds_test::thread_pool& pool = get_pool();
 148
 149             for ( size_t i = 0; i < pool.size(); ++i ) {
 150                 Producer<Queue>& thread = static_cast<Producer<Queue>&>(pool.get( i ));
 151                 EXPECT_EQ( thread.m_nPushError, 0u ) << " producer thread " << i;
 152             }
 153             EXPECT_TRUE( !q.empty());
 154
 155             std::unique_ptr< uint8_t[] > arr( new uint8_t[s_nQueueSize] );
 156             memset( arr.get(), 0, sizeof(arr[0]) * s_nQueueSize );
 157
 158             size_t nPopped = 0;
 159             value_type val;
 160             while ( q.pop( val )) {
 161                 nPopped++;
 162                 ++arr[ val.nNo ];
 163             }
 164
 165             size_t nTotalItems = nThreadItems * s_nThreadCount;
 166             for ( size_t i = 0; i < nTotalItems; ++i ) {
 167                 EXPECT_EQ( arr[i], 1 ) << "i=" << i;
 168             }
 169         }
 170     };
 171
 172     CDSSTRESS_MSQueue( queue_push )
 173     CDSSTRESS_MoirQueue( queue_push )
 174     CDSSTRESS_BasketQueue( queue_push )
 175     CDSSTRESS_OptimsticQueue( queue_push )
 176     CDSSTRESS_RWQueue( queue_push )
 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_push )
 188
 189 #undef CDSSTRESS_Queue_F
 190
 191
 192     // ********************************************************************
 193     // SegmentedQueue test
 194
 195     class segmented_queue_push
 196         : public queue_push
 197         , public ::testing::WithParamInterface< size_t >
 198     {
 199         typedef queue_push 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_push" );
 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_push )
 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_push,
 251         ::testing::ValuesIn( segmented_queue_push::get_test_parameters()), get_test_parameter_name );
 252 #else
 253         INSTANTIATE_TEST_CASE_P( SQ,
 254             segmented_queue_push,
 255             ::testing::ValuesIn( segmented_queue_push::get_test_parameters()));
 256 #endif
 257
 258
 259 } // namespace