Only runs standard stack test cases
[libcds.git] / 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