Rebuilt threaded uRCU logic
[libcds.git] / cds / container / vyukov_mpmc_cycle_queue.h
1 //$$CDS-header$$
2
3 #ifndef CDSLIB_CONTAINER_VYUKOV_MPMC_CYCLE_QUEUE_H
4 #define CDSLIB_CONTAINER_VYUKOV_MPMC_CYCLE_QUEUE_H
5
6 #include <cds/container/details/base.h>
7 #include <cds/opt/buffer.h>
8 #include <cds/opt/value_cleaner.h>
9 #include <cds/algo/atomic.h>
10 #include <cds/details/bounded_container.h>
11
12 namespace cds { namespace container {
13
14     /// VyukovMPMCCycleQueue related definitions
15     /** @ingroup cds_nonintrusive_helper
16     */
17     namespace vyukov_queue {
18
19         /// VyukovMPMCCycleQueue default traits
20         struct traits {
21             /// Buffer type for internal array
22             /*
23                 The type of element for the buffer is not important: the queue rebinds
24                 buffer for required type via \p rebind metafunction.
25
26                 For \p VyukovMPMCCycleQueue queue the buffer size should have power-of-2 size.
27             */
28             typedef cds::opt::v::dynamic_buffer< void * > buffer;
29
30             /// A functor to clean item dequeued.
31             /**
32                 The functor  calls the destructor for queue item.
33                 After an item is dequeued, \p value_cleaner cleans the cell that the item has been occupied.
34                 If \p T is a complex type, \p value_cleaner may be the useful feature.
35
36                 Default value is \ref opt::v::destruct_cleaner
37             */
38             typedef cds::opt::v::destruct_cleaner value_cleaner;
39
40             /// Item counting feature; by default, disabled. Use \p cds::atomicity::item_counter to enable item counting
41             typedef cds::atomicity::empty_item_counter item_counter;
42
43             /// C++ memory ordering model
44             /**
45                 Can be \p opt::v::relaxed_ordering (relaxed memory model, the default)
46                 or \p opt::v::sequential_consistent (sequentially consisnent memory model).
47             */
48             typedef opt::v::relaxed_ordering    memory_model;
49
50             /// Padding for internal critical atomic data. Default is \p opt::cache_line_padding
51             enum { padding = opt::cache_line_padding };
52
53             /// Back-off strategy
54             typedef cds::backoff::Default           back_off;
55
56             /// Single-consumer version
57             /**
58                 For single-consumer version of algorithm some additional functions 
59                 (\p front(), \p pop_front()) is available.
60
61                 Default is \p false
62             */
63             static CDS_CONSTEXPR bool const single_consumer = false;
64         };
65
66         /// Metafunction converting option list to \p vyukov_queue::traits
67         /**
68             Supported \p Options are:
69             - \p opt::buffer - the buffer type for internal cyclic array. Possible types are:
70                 \p opt::v::dynamic_buffer (the default), \p opt::v::static_buffer. The type of
71                 element in the buffer is not important: it will be changed via \p rebind metafunction.
72             - \p opt::value_cleaner - a functor to clean item dequeued.
73                 The functor calls the destructor for queue item.
74                 After an item is dequeued, \p value_cleaner cleans the cell that the item has been occupied.
75                 If \p T is a complex type, \p value_cleaner can be an useful feature.
76                 Default value is \ref opt::v::destruct_cleaner
77             - \p opt::back_off - back-off strategy used. If the option is not specified, the \p cds::backoff::Default is used.
78             - \p opt::item_counter - the type of item counting feature. Default is \p cds::atomicity::empty_item_counter (item counting disabled)
79                 To enable item counting use \p cds::atomicity::item_counter
80             - \p opt::padding - padding for internal critical atomic data. Default is \p opt::cache_line_padding
81             - \p opt::memory_model - C++ memory ordering model. Can be \p opt::v::relaxed_ordering (relaxed memory model, the default)
82                 or \p opt::v::sequential_consistent (sequentially consisnent memory model).
83
84             Example: declare \p %VyukovMPMCCycleQueue with item counting and static iternal buffer of size 1024:
85             \code
86             typedef cds::container::VyukovMPMCCycleQueue< Foo,
87                 typename cds::container::vyukov_queue::make_traits<
88                     cds::opt::buffer< cds::opt::v::static_buffer< void *, 1024 >,
89                     cds::opt::item_counte< cds::atomicity::item_counter >
90                 >::type
91             > myQueue;
92             \endcode
93         */
94         template <typename... Options>
95         struct make_traits {
96 #   ifdef CDS_DOXYGEN_INVOKED
97             typedef implementation_defined type;   ///< Metafunction result
98 #   else
99             typedef typename cds::opt::make_options<
100                 typename cds::opt::find_type_traits< traits, Options... >::type
101                 , Options...
102             >::type type;
103 #   endif
104         };
105
106     } //namespace vyukov_queue
107
108     /// Vyukov's MPMC bounded queue
109     /** @ingroup cds_nonintrusive_queue
110         This algorithm is developed by Dmitry Vyukov (see http://www.1024cores.net)
111         It's multi-producer multi-consumer (MPMC), array-based, fails on overflow, does not require GC, w/o priorities, causal FIFO,
112         blocking producers and consumers queue. The algorithm is pretty simple and fast. It's not lock-free in the official meaning,
113         just implemented by means of atomic RMW operations w/o mutexes.
114
115         The cost of enqueue/dequeue is 1 CAS per operation.
116         No dynamic memory allocation/management during operation. Producers and consumers are separated from each other (as in the two-lock queue),
117         i.e. do not touch the same data while queue is not empty.
118
119         There is multiple producer/single consumer version \p cds::container::VyukovMPSCCycleQueue
120         that supports \p front() and \p pop_front() functions.
121
122         Source:
123             - http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
124
125         Template parameters
126         - \p T - type stored in queue.
127         - \p Traits - queue traits, default is \p vykov_queue::traits. You can use \p vykov_queue::make_traits
128             metafunction to make your traits or just derive your traits from \p %vykov_queue::traits:
129             \code
130             struct myTraits: public cds::container::vykov_queue::traits {
131                 typedef cds::atomicity::item_counter    item_counter;
132             };
133             typedef cds::container::VyukovMPMCCycleQueue< Foo, myTraits > myQueue;
134
135             // Equivalent make_traits example:
136             typedef cds::container::VyukovMPMCCycleQueue< cds::gc::HP, Foo,
137                 typename cds::container::vykov_queue::make_traits<
138                     cds::opt::item_counter< cds::atomicity::item_counter >
139                 >::type
140             > myQueue;
141             \endcode
142
143         \par License
144             Simplified BSD license by Dmitry Vyukov (http://www.1024cores.net/site/1024cores/home/code-license)
145     */
146     template <typename T, typename Traits = vyukov_queue::traits >
147     class VyukovMPMCCycleQueue : public cds::bounded_container
148     {
149     public:
150         typedef T value_type;   ///< Value type to be stored in the queue
151         typedef Traits traits;  ///< Queue traits
152         typedef typename traits::item_counter  item_counter;  ///< Item counter type
153         typedef typename traits::memory_model  memory_model;  ///< Memory ordering. See cds::opt::memory_model option
154         typedef typename traits::value_cleaner value_cleaner; ///< Value cleaner, see \p vyukov_queue::traits::value_cleaner
155         typedef typename traits::back_off  back_off;          ///< back-off strategy
156
157         /// \p true for single-consumer version, \p false otherwise
158         static CDS_CONSTEXPR bool const c_single_consumer = traits::single_consumer;
159
160         /// Rebind template arguments
161         template <typename T2, typename Traits2>
162         struct rebind {
163             typedef VyukovMPMCCycleQueue< T2, Traits2 > other   ;   ///< Rebinding result
164         };
165
166     protected:
167         //@cond
168         typedef atomics::atomic<size_t> sequence_type;
169         struct cell_type
170         {
171             sequence_type   sequence;
172             value_type      data;
173
174             cell_type()
175             {}
176         };
177
178         typedef typename traits::buffer::template rebind<cell_type>::other buffer;
179         //@endcond
180
181     protected:
182         //@cond
183         buffer          m_buffer;
184         size_t const    m_nBufferMask;
185         typename opt::details::apply_padding< size_t, traits::padding >::padding_type pad1_;
186         sequence_type   m_posEnqueue;
187         typename opt::details::apply_padding< sequence_type, traits::padding >::padding_type pad2_;
188         sequence_type   m_posDequeue;
189         typename opt::details::apply_padding< sequence_type, traits::padding >::padding_type pad3_;
190         item_counter    m_ItemCounter;
191         //@endcond
192
193     public:
194         /// Constructs the queue of capacity \p nCapacity
195         /**
196             For \p cds::opt::v::static_buffer the \p nCapacity parameter is ignored.
197
198             The buffer capacity must be the power of two.
199         */
200         VyukovMPMCCycleQueue(
201             size_t nCapacity = 0
202             )
203             : m_buffer( nCapacity )
204             , m_nBufferMask( m_buffer.capacity() - 1 )
205         {
206             nCapacity = m_buffer.capacity();
207
208             // Buffer capacity must be power of 2
209             assert( nCapacity >= 2 && (nCapacity & (nCapacity - 1)) == 0 );
210
211             for (size_t i = 0; i != nCapacity; ++i )
212                 m_buffer[i].sequence.store(i, memory_model::memory_order_relaxed);
213
214             m_posEnqueue.store(0, memory_model::memory_order_relaxed);
215             m_posDequeue.store(0, memory_model::memory_order_relaxed);
216         }
217
218         ~VyukovMPMCCycleQueue()
219         {
220             clear();
221         }
222
223         /// Enqueues data to the queue using a functor
224         /**
225             \p Func is a functor called to copy a value to the queue cell.
226             The functor \p f takes one argument - a reference to a empty cell of type \ref value_type :
227             \code
228             cds::container::VyukovMPMCCycleQueue< Foo > myQueue;
229             Bar bar;
230             myQueue.enqueue_with( [&bar]( Foo& dest ) { dest = std::move(bar); } );
231             \endcode
232         */
233         template <typename Func>
234         bool enqueue_with(Func f)
235         {
236             cell_type* cell;
237             back_off bkoff;
238
239             size_t pos = m_posEnqueue.load(memory_model::memory_order_relaxed);
240             for (;;)
241             {
242                 cell = &m_buffer[pos & m_nBufferMask];
243                 size_t seq = cell->sequence.load(memory_model::memory_order_acquire);
244
245                 intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos);
246
247                 if (dif == 0) {
248                     if ( m_posEnqueue.compare_exchange_weak(pos, pos + 1, memory_model::memory_order_relaxed, atomics::memory_order_relaxed ))
249                         break;
250                 }
251                 else if (dif < 0) {
252                     // Queue full?
253                     if ( pos - m_posDequeue.load( memory_model::memory_order_relaxed ) == capacity() )
254                         return false;   // queue full
255                     bkoff();
256                     pos = m_posEnqueue.load( memory_model::memory_order_relaxed );
257                 }
258                 else
259                     pos = m_posEnqueue.load(memory_model::memory_order_relaxed);
260             }
261
262             f( cell->data );
263
264             cell->sequence.store(pos + 1, memory_model::memory_order_release);
265             ++m_ItemCounter;
266
267             return true;
268         }
269
270         /// Enqueues \p val value into the queue.
271         /**
272             The new queue item is created by calling placement new in free cell.
273             Returns \p true if success, \p false if the queue is full.
274         */
275         bool enqueue( value_type const& val )
276         {
277             return enqueue_with( [&val]( value_type& dest ){ new ( &dest ) value_type( val ); });
278         }
279
280         /// Synonym for \p enqueue()
281         bool push( value_type const& data )
282         {
283             return enqueue( data );
284         }
285
286         /// Synonym for \p enqueue_with()
287         template <typename Func>
288         bool push_with( Func f )
289         {
290             return enqueue_with( f );
291         }
292
293         /// Enqueues data of type \ref value_type constructed with <tt>std::forward<Args>(args)...</tt>
294         template <typename... Args>
295         bool emplace( Args&&... args )
296         {
297 #if (CDS_COMPILER == CDS_COMPILER_GCC) && (CDS_COMPILER_VERSION < 40900)
298             //work around unsupported feature in g++ 4.8 for forwarding parameter packs to lambda.
299             return enqueue_with ( std::bind([]( value_type& dest,Args ... args ){ new ( &dest ) value_type( std::forward<Args>(args)... );}, std::placeholders::_1 ,args...));
300 #else
301             return enqueue_with( [&args ...]( value_type& dest ){ new ( &dest ) value_type( std::forward<Args>(args)... ); });
302 #endif
303         }
304
305         /// Dequeues a value using a functor
306         /**
307             \p Func is a functor called to copy dequeued value.
308             The functor takes one argument - a reference to removed node:
309             \code
310             cds:container::VyukovMPMCCycleQueue< Foo > myQueue;
311             Bar bar;
312             myQueue.dequeue_with( [&bar]( Foo& src ) { bar = std::move( src );});
313             \endcode
314             The functor is called only if the queue is not empty.
315         */
316         template <typename Func>
317         bool dequeue_with( Func f )
318         {
319             cell_type * cell;
320             back_off bkoff;
321
322             size_t pos = m_posDequeue.load( memory_model::memory_order_relaxed );
323             for (;;)
324             {
325                 cell = &m_buffer[pos & m_nBufferMask];
326                 size_t seq = cell->sequence.load(memory_model::memory_order_acquire);
327                 intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
328
329                 if (dif == 0) {
330                     if ( m_posDequeue.compare_exchange_weak(pos, pos + 1, memory_model::memory_order_relaxed, atomics::memory_order_relaxed))
331                         break;
332                 }
333                 else if (dif < 0) {
334                     // Queue empty?
335                     if ( pos - m_posEnqueue.load( memory_model::memory_order_relaxed ) == 0 )
336                         return false;   // queue empty
337                     bkoff();
338                     pos = m_posDequeue.load( memory_model::memory_order_relaxed );
339                 }
340                 else
341                     pos = m_posDequeue.load(memory_model::memory_order_relaxed);
342             }
343
344             f( cell->data );
345             value_cleaner()( cell->data );
346             cell->sequence.store( pos + m_nBufferMask + 1, memory_model::memory_order_release );
347             --m_ItemCounter;
348
349             return true;
350         }
351
352         /// Dequeues a value from the queue
353         /**
354             If queue is not empty, the function returns \p true, \p dest contains copy of
355             dequeued value. The assignment operator for type \ref value_type is invoked.
356             If queue is empty, the function returns \p false, \p dest is unchanged.
357         */
358         bool dequeue(value_type & dest )
359         {
360             return dequeue_with( [&dest]( value_type& src ){ dest = src; } );
361         }
362
363         /// Synonym for \p dequeue()
364         bool pop(value_type& data)
365         {
366             return dequeue(data);
367         }
368
369         /// Synonym for \p dequeue_with()
370         template <typename Func>
371         bool pop_with( Func f )
372         {
373             return dequeue_with( f );
374         }
375
376         /// Returns a pointer to top element of the queue or \p nullptr if queue is empty (only for single-consumer version)
377         template <bool SC = c_single_consumer >
378         typename std::enable_if<SC, value_type *>::type front()
379         {
380             static_assert( c_single_consumer, "front() is enabled only if traits::single_consumer is true");
381
382             cell_type * cell;
383             back_off bkoff;
384
385             size_t pos = m_posDequeue.load( memory_model::memory_order_relaxed );
386             for ( ;;)
387             {
388                 cell = &m_buffer[pos & m_nBufferMask];
389                 size_t seq = cell->sequence.load( memory_model::memory_order_acquire );
390                 intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
391
392                 if ( dif == 0 )
393                     return &cell->data;
394                 else if ( dif < 0 ) {
395                     // Queue empty?
396                     if ( pos - m_posEnqueue.load( memory_model::memory_order_relaxed ) == 0 )
397                         return nullptr;   // queue empty
398                     bkoff();
399                     pos = m_posDequeue.load( memory_model::memory_order_relaxed );
400                 }
401                 else
402                     pos = m_posDequeue.load( memory_model::memory_order_relaxed );
403             }
404         }
405
406         /// Pops top element; returns \p true if queue is not empty, \p false otherwise (only for single-consumer version)
407         template <bool SC = c_single_consumer >
408         typename std::enable_if<SC, bool>::type pop_front()
409         {
410             return dequeue_with( []( value_type& ) {} );
411         }
412
413         /// Checks if the queue is empty
414         bool empty() const
415         {
416             const cell_type * cell;
417             back_off bkoff;
418
419             size_t pos = m_posDequeue.load(memory_model::memory_order_relaxed);
420             for (;;)
421             {
422                 cell = &m_buffer[pos & m_nBufferMask];
423                 size_t seq = cell->sequence.load(memory_model::memory_order_acquire);
424                 intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
425
426                 if (dif == 0)
427                     return false;
428                 else if (dif < 0) {
429                     if ( pos - m_posEnqueue.load( memory_model::memory_order_relaxed ) == 0 )
430                         return true;
431                 }
432                 bkoff();
433                 pos = m_posDequeue.load(memory_model::memory_order_relaxed);
434             }
435         }
436
437         /// Clears the queue
438         void clear()
439         {
440             value_type v;
441             while ( pop(v) );
442         }
443
444         /// Returns queue's item count
445         /**
446             The value returned depends on \p vyukov_queue::traits::item_counter option.
447             For \p atomicity::empty_item_counter, the function always returns 0.
448         */
449         size_t size() const
450         {
451             return m_ItemCounter.value();
452         }
453
454         /// Returns capacity of the queue
455         size_t capacity() const
456         {
457             return m_buffer.capacity();
458         }
459     };
460
461     //@cond
462     namespace vyukov_queue { 
463         template <typename Traits>
464         struct single_consumer_traits : public Traits
465         {
466             static CDS_CONSTEXPR bool const single_consumer = true;
467         };
468     } // namespace vyukov_queue
469     //@endcond
470
471     /// Vyukov's queue multiple producer - single consumer version 
472     template <typename T, typename Traits = vyukov_queue::traits >
473     using VyukovMPSCCycleQueue = VyukovMPMCCycleQueue< T, vyukov_queue::single_consumer_traits<Traits> >;
474
475 }}  // namespace cds::container
476
477 #endif // #ifndef CDSLIB_CONTAINER_VYUKOV_MPMC_CYCLE_QUEUE_H