test/stress/queue/spsc_queue.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 #include <vector>
  34 #include <algorithm>
  35 #include <type_traits>
  36
  37 // Single producer/single consumer queue push/pop test
  38 namespace {
  39
  40     static size_t s_nQueueSize = 1024;
  41     static size_t s_nPassCount = 1024;
  42
  43     static std::atomic<size_t> s_nProducerDone( 0 );
  44
  45     class spsc_queue: public cds_test::stress_fixture
  46     {
  47     protected:
  48        typedef size_t value_type;
  49
  50         enum {
  51             producer_thread,
  52             consumer_thread
  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, producer_thread )
  63                 , m_Queue( queue )
  64             {}
  65
  66             Producer( Producer& src )
  67                 : base_class( src )
  68                 , m_Queue( src.m_Queue )
  69             {}
  70
  71             virtual thread * clone()
  72             {
  73                 return new Producer( *this );
  74             }
  75
  76             virtual void test()
  77             {
  78                 size_t const nPushCount = s_nQueueSize * s_nPassCount;
  79                 m_nPushFailed = 0;
  80
  81                 for ( value_type v = 0; v < nPushCount; ++v ) {
  82                     if ( !m_Queue.push( v )) {
  83                         ++m_nPushFailed;
  84                         --v;
  85                     }
  86                 }
  87
  88                 s_nProducerDone.fetch_add( 1 );
  89             }
  90
  91         public:
  92             Queue&              m_Queue;
  93             size_t              m_nPushFailed = 0;
  94         };
  95
  96         template <class Queue>
  97         class Consumer: public cds_test::thread
  98         {
  99             typedef cds_test::thread base_class;
 100
 101         public:
 102             Queue&              m_Queue;
 103             size_t              m_nPopEmpty = 0;
 104             size_t              m_nPopped = 0;
 105             size_t              m_nBadValue = 0;
 106
 107         public:
 108             Consumer( cds_test::thread_pool& pool, Queue& queue )
 109                 : base_class( pool, consumer_thread )
 110                 , m_Queue( queue )
 111             {}
 112             Consumer( Consumer& src )
 113                 : base_class( src )
 114                 , m_Queue( src.m_Queue )
 115             {}
 116
 117             virtual thread * clone()
 118             {
 119                 return new Consumer( *this );
 120             }
 121
 122             virtual void test()
 123             {
 124                 value_type v;
 125                 value_type prev = 0 - 1;
 126
 127                 while ( true ) {
 128                     if ( m_Queue.pop( v )) {
 129                         ++m_nPopped;
 130                         if ( v != prev + 1 )
 131                             ++m_nBadValue;
 132                         prev = v;
 133                     }
 134                     else {
 135                         ++m_nPopEmpty;
 136
 137                         if ( s_nProducerDone.load() != 0 ) {
 138                             if ( m_Queue.empty())
 139                                 break;
 140                         }
 141                     }
 142                 }
 143             }
 144         };
 145
 146     protected:
 147         size_t m_nThreadPushCount;
 148
 149     protected:
 150         template <class Queue>
 151         void test_queue( Queue& q )
 152         {
 153             cds_test::thread_pool& pool = get_pool();
 154             auto producer = new Producer<Queue>( pool, q );
 155             auto consumer = new Consumer<Queue>( pool, q );
 156
 157             pool.add( producer, 1 );
 158             pool.add( consumer, 1 );
 159
 160             s_nProducerDone.store( 0 );
 161
 162             propout() << std::make_pair( "queue_size", s_nQueueSize )
 163                       << std::make_pair( "pass_count", s_nPassCount );
 164
 165             std::chrono::milliseconds duration = pool.run();
 166
 167             propout() << std::make_pair( "duration", duration );
 168
 169             // analyze result
 170             EXPECT_EQ( consumer->m_nBadValue, 0u );
 171             EXPECT_EQ( consumer->m_nPopped, s_nQueueSize * s_nPassCount );
 172
 173             propout()
 174                 << std::make_pair( "producer_push_count", s_nQueueSize * s_nPassCount )
 175                 << std::make_pair( "producer_push_failed", producer->m_nPushFailed )
 176                 << std::make_pair( "consumer_pop_count", consumer->m_nPopped )
 177                 << std::make_pair( "consumer_pop_empty", consumer->m_nPopEmpty )
 178                 << std::make_pair( "consumer_bad_value", consumer->m_nBadValue );
 179         }
 180
 181         template <class Queue>
 182         void test( Queue& q )
 183         {
 184             test_queue( q );
 185             propout() << q.statistics();
 186         }
 187
 188     public:
 189         static void SetUpTestCase()
 190         {
 191             cds_test::config const& cfg = get_config( "spsc_queue" );
 192
 193             s_nQueueSize = cfg.get_size_t( "QueueSize", s_nQueueSize );
 194             s_nPassCount = cfg.get_size_t( "PassCount", s_nPassCount );
 195
 196             if ( s_nQueueSize == 0u )
 197                 s_nQueueSize = 1024;
 198             if ( s_nPassCount == 0u )
 199                 s_nPassCount = 1024;
 200         }
 201     };
 202
 203     CDSSTRESS_MSQueue( spsc_queue )
 204     CDSSTRESS_MoirQueue( spsc_queue )
 205     CDSSTRESS_BasketQueue( spsc_queue )
 206     CDSSTRESS_OptimsticQueue( spsc_queue )
 207     //CDSSTRESS_FCQueue( spsc_queue )
 208     //CDSSTRESS_FCDeque( spsc_queue )
 209     CDSSTRESS_RWQueue( spsc_queue )
 210     //CDSSTRESS_StdQueue( spsc_queue )
 211
 212 #undef CDSSTRESS_Queue_F
 213 #define CDSSTRESS_Queue_F( test_fixture, type_name ) \
 214     TEST_F( test_fixture, type_name ) \
 215     { \
 216         typedef queue::Types< value_type >::type_name queue_type; \
 217         queue_type queue( s_nQueueSize ); \
 218         test( queue ); \
 219     }
 220
 221     CDSSTRESS_WeakRingBuffer( spsc_queue )
 222     CDSSTRESS_VyukovQueue( spsc_queue )
 223     CDSSTRESS_VyukovSingleConsumerQueue( spsc_queue )
 224
 225 #undef CDSSTRESS_Queue_F
 226
 227 } // namespace