1// lock-free queue from
2// Michael, M. M. and Scott, M. L.,
3// "simple, fast and practical non-blocking and blocking concurrent queue algorithms"
4//
5// Copyright (C) 2008-2013 Tim Blechmann
6//
7// Distributed under the Boost Software License, Version 1.0. (See
8// accompanying file LICENSE_1_0.txt or copy at
9// http://www.boost.org/LICENSE_1_0.txt)
10
11#ifndef BOOST_LOCKFREE_FIFO_HPP_INCLUDED
12#define BOOST_LOCKFREE_FIFO_HPP_INCLUDED
13
14#include <boost/assert.hpp>
15#include <boost/static_assert.hpp>
16#include <boost/core/allocator_access.hpp>
17#include <boost/type_traits/has_trivial_assign.hpp>
18#include <boost/type_traits/has_trivial_destructor.hpp>
19#include <boost/config.hpp> // for BOOST_LIKELY & BOOST_ALIGNMENT
20
21#include <boost/lockfree/detail/atomic.hpp>
22#include <boost/lockfree/detail/copy_payload.hpp>
23#include <boost/lockfree/detail/freelist.hpp>
24#include <boost/lockfree/detail/parameter.hpp>
25#include <boost/lockfree/detail/tagged_ptr.hpp>
26
27#include <boost/lockfree/lockfree_forward.hpp>
28
29#ifdef BOOST_HAS_PRAGMA_ONCE
30#pragma once
31#endif
32
33
34#if defined(_MSC_VER)
35#pragma warning(push)
36#pragma warning(disable: 4324) // structure was padded due to __declspec(align())
37#endif
38
39#if defined(BOOST_INTEL) && (BOOST_INTEL_CXX_VERSION > 1000)
40#pragma warning(push)
41#pragma warning(disable:488) // template parameter unused in declaring parameter types,
42 // gets erronously triggered the queue constructor which
43 // takes an allocator of another type and rebinds it
44#endif
45
46
47
48namespace boost {
49namespace lockfree {
50namespace detail {
51
52typedef parameter::parameters<boost::parameter::optional<tag::allocator>,
53 boost::parameter::optional<tag::capacity>
54 > queue_signature;
55
56} /* namespace detail */
57
58
59/** The queue class provides a multi-writer/multi-reader queue, pushing and popping is lock-free,
60 * construction/destruction has to be synchronized. It uses a freelist for memory management,
61 * freed nodes are pushed to the freelist and not returned to the OS before the queue is destroyed.
62 *
63 * \b Policies:
64 * - \ref boost::lockfree::fixed_sized, defaults to \c boost::lockfree::fixed_sized<false> \n
65 * Can be used to completely disable dynamic memory allocations during push in order to ensure lockfree behavior. \n
66 * If the data structure is configured as fixed-sized, the internal nodes are stored inside an array and they are addressed
67 * by array indexing. This limits the possible size of the queue to the number of elements that can be addressed by the index
68 * type (usually 2**16-2), but on platforms that lack double-width compare-and-exchange instructions, this is the best way
69 * to achieve lock-freedom.
70 *
71 * - \ref boost::lockfree::capacity, optional \n
72 * If this template argument is passed to the options, the size of the queue is set at compile-time.\n
73 * This option implies \c fixed_sized<true>
74 *
75 * - \ref boost::lockfree::allocator, defaults to \c boost::lockfree::allocator<std::allocator<void>> \n
76 * Specifies the allocator that is used for the internal freelist
77 *
78 * \b Requirements:
79 * - T must have a copy constructor
80 * - T must have a trivial assignment operator
81 * - T must have a trivial destructor
82 *
83 * */
84#ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
85template <typename T, class A0, class A1, class A2>
86#else
87template <typename T, typename ...Options>
88#endif
89class queue
90{
91private:
92#ifndef BOOST_DOXYGEN_INVOKED
93
94#ifdef BOOST_HAS_TRIVIAL_DESTRUCTOR
95 BOOST_STATIC_ASSERT((boost::has_trivial_destructor<T>::value));
96#endif
97
98#ifdef BOOST_HAS_TRIVIAL_ASSIGN
99 BOOST_STATIC_ASSERT((boost::has_trivial_assign<T>::value));
100#endif
101
102#ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
103 typedef typename detail::queue_signature::bind<A0, A1, A2>::type bound_args;
104#else
105 typedef typename detail::queue_signature::bind<Options...>::type bound_args;
106#endif
107
108 static const bool has_capacity = detail::extract_capacity<bound_args>::has_capacity;
109 static const size_t capacity = detail::extract_capacity<bound_args>::capacity + 1; // the queue uses one dummy node
110 static const bool fixed_sized = detail::extract_fixed_sized<bound_args>::value;
111 static const bool node_based = !(has_capacity || fixed_sized);
112 static const bool compile_time_sized = has_capacity;
113
114 struct BOOST_ALIGNMENT(BOOST_LOCKFREE_CACHELINE_BYTES) node
115 {
116 typedef typename detail::select_tagged_handle<node, node_based>::tagged_handle_type tagged_node_handle;
117 typedef typename detail::select_tagged_handle<node, node_based>::handle_type handle_type;
118
119 node(T const & v, handle_type null_handle):
120 data(v)
121 {
122 /* increment tag to avoid ABA problem */
123 tagged_node_handle old_next = next.load(memory_order_relaxed);
124 tagged_node_handle new_next (null_handle, old_next.get_next_tag());
125 next.store(new_next, memory_order_release);
126 }
127
128 node (handle_type null_handle):
129 next(tagged_node_handle(null_handle, 0))
130 {}
131
132 node(void)
133 {}
134
135 atomic<tagged_node_handle> next;
136 T data;
137 };
138
139 typedef typename detail::extract_allocator<bound_args, node>::type node_allocator;
140 typedef typename detail::select_freelist<node, node_allocator, compile_time_sized, fixed_sized, capacity>::type pool_t;
141 typedef typename pool_t::tagged_node_handle tagged_node_handle;
142 typedef typename detail::select_tagged_handle<node, node_based>::handle_type handle_type;
143
144 void initialize(void)
145 {
146 node * n = pool.template construct<true, false>(pool.null_handle());
147 tagged_node_handle dummy_node(pool.get_handle(n), 0);
148 head_.store(dummy_node, memory_order_relaxed);
149 tail_.store(dummy_node, memory_order_release);
150 }
151
152 struct implementation_defined
153 {
154 typedef node_allocator allocator;
155 typedef std::size_t size_type;
156 };
157
158#endif
159
160 BOOST_DELETED_FUNCTION(queue(queue const&))
161 BOOST_DELETED_FUNCTION(queue& operator= (queue const&))
162
163public:
164 typedef T value_type;
165 typedef typename implementation_defined::allocator allocator;
166 typedef typename implementation_defined::size_type size_type;
167
168 /**
169 * \return true, if implementation is lock-free.
170 *
171 * \warning It only checks, if the queue head and tail nodes and the freelist can be modified in a lock-free manner.
172 * On most platforms, the whole implementation is lock-free, if this is true. Using c++0x-style atomics, there is
173 * no possibility to provide a completely accurate implementation, because one would need to test every internal
174 * node, which is impossible if further nodes will be allocated from the operating system.
175 * */
176 bool is_lock_free (void) const
177 {
178 return head_.is_lock_free() && tail_.is_lock_free() && pool.is_lock_free();
179 }
180
181 /** Construct a fixed-sized queue
182 *
183 * \pre Must specify a capacity<> argument
184 * */
185 queue(void):
186 head_(tagged_node_handle(0, 0)),
187 tail_(tagged_node_handle(0, 0)),
188 pool(node_allocator(), capacity)
189 {
190 // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
191 // this function and this function may be compiled even when it isn't being used.
192 BOOST_ASSERT(has_capacity);
193 initialize();
194 }
195
196 /** Construct a fixed-sized queue with a custom allocator
197 *
198 * \pre Must specify a capacity<> argument
199 * */
200 template <typename U>
201 explicit queue(typename boost::allocator_rebind<node_allocator, U>::type const & alloc):
202 head_(tagged_node_handle(0, 0)),
203 tail_(tagged_node_handle(0, 0)),
204 pool(alloc, capacity)
205 {
206 BOOST_STATIC_ASSERT(has_capacity);
207 initialize();
208 }
209
210 /** Construct a fixed-sized queue with a custom allocator
211 *
212 * \pre Must specify a capacity<> argument
213 * */
214 explicit queue(allocator const & alloc):
215 head_(tagged_node_handle(0, 0)),
216 tail_(tagged_node_handle(0, 0)),
217 pool(alloc, capacity)
218 {
219 // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
220 // this function and this function may be compiled even when it isn't being used.
221 BOOST_ASSERT(has_capacity);
222 initialize();
223 }
224
225 /** Construct a variable-sized queue
226 *
227 * Allocate n nodes initially for the freelist
228 *
229 * \pre Must \b not specify a capacity<> argument
230 * */
231 explicit queue(size_type n):
232 head_(tagged_node_handle(0, 0)),
233 tail_(tagged_node_handle(0, 0)),
234 pool(node_allocator(), n + 1)
235 {
236 // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
237 // this function and this function may be compiled even when it isn't being used.
238 BOOST_ASSERT(!has_capacity);
239 initialize();
240 }
241
242 /** Construct a variable-sized queue with a custom allocator
243 *
244 * Allocate n nodes initially for the freelist
245 *
246 * \pre Must \b not specify a capacity<> argument
247 * */
248 template <typename U>
249 queue(size_type n, typename boost::allocator_rebind<node_allocator, U>::type const & alloc):
250 head_(tagged_node_handle(0, 0)),
251 tail_(tagged_node_handle(0, 0)),
252 pool(alloc, n + 1)
253 {
254 BOOST_STATIC_ASSERT(!has_capacity);
255 initialize();
256 }
257
258 /** \copydoc boost::lockfree::stack::reserve
259 * */
260 void reserve(size_type n)
261 {
262 pool.template reserve<true>(n);
263 }
264
265 /** \copydoc boost::lockfree::stack::reserve_unsafe
266 * */
267 void reserve_unsafe(size_type n)
268 {
269 pool.template reserve<false>(n);
270 }
271
272 /** Destroys queue, free all nodes from freelist.
273 * */
274 ~queue(void)
275 {
276 T dummy;
277 while(unsynchronized_pop(dummy))
278 {}
279
280 pool.template destruct<false>(head_.load(memory_order_relaxed));
281 }
282
283 /** Check if the queue is empty
284 *
285 * \return true, if the queue is empty, false otherwise
286 * \note The result is only accurate, if no other thread modifies the queue. Therefore it is rarely practical to use this
287 * value in program logic.
288 * */
289 bool empty(void) const
290 {
291 return pool.get_handle(head_.load()) == pool.get_handle(tail_.load());
292 }
293
294 /** Pushes object t to the queue.
295 *
296 * \post object will be pushed to the queue, if internal node can be allocated
297 * \returns true, if the push operation is successful.
298 *
299 * \note Thread-safe. If internal memory pool is exhausted and the memory pool is not fixed-sized, a new node will be allocated
300 * from the OS. This may not be lock-free.
301 * */
302 bool push(T const & t)
303 {
304 return do_push<false>(t);
305 }
306
307 /** Pushes object t to the queue.
308 *
309 * \post object will be pushed to the queue, if internal node can be allocated
310 * \returns true, if the push operation is successful.
311 *
312 * \note Thread-safe and non-blocking. If internal memory pool is exhausted, operation will fail
313 * \throws if memory allocator throws
314 * */
315 bool bounded_push(T const & t)
316 {
317 return do_push<true>(t);
318 }
319
320
321private:
322#ifndef BOOST_DOXYGEN_INVOKED
323 template <bool Bounded>
324 bool do_push(T const & t)
325 {
326 node * n = pool.template construct<true, Bounded>(t, pool.null_handle());
327 handle_type node_handle = pool.get_handle(n);
328
329 if (n == NULL)
330 return false;
331
332 for (;;) {
333 tagged_node_handle tail = tail_.load(memory_order_acquire);
334 node * tail_node = pool.get_pointer(tail);
335 tagged_node_handle next = tail_node->next.load(memory_order_acquire);
336 node * next_ptr = pool.get_pointer(next);
337
338 tagged_node_handle tail2 = tail_.load(memory_order_acquire);
339 if (BOOST_LIKELY(tail == tail2)) {
340 if (next_ptr == 0) {
341 tagged_node_handle new_tail_next(node_handle, next.get_next_tag());
342 if ( tail_node->next.compare_exchange_weak(next, new_tail_next) ) {
343 tagged_node_handle new_tail(node_handle, tail.get_next_tag());
344 tail_.compare_exchange_strong(tail, new_tail);
345 return true;
346 }
347 }
348 else {
349 tagged_node_handle new_tail(pool.get_handle(next_ptr), tail.get_next_tag());
350 tail_.compare_exchange_strong(tail, new_tail);
351 }
352 }
353 }
354 }
355#endif
356
357public:
358
359 /** Pushes object t to the queue.
360 *
361 * \post object will be pushed to the queue, if internal node can be allocated
362 * \returns true, if the push operation is successful.
363 *
364 * \note Not Thread-safe. If internal memory pool is exhausted and the memory pool is not fixed-sized, a new node will be allocated
365 * from the OS. This may not be lock-free.
366 * \throws if memory allocator throws
367 * */
368 bool unsynchronized_push(T const & t)
369 {
370 node * n = pool.template construct<false, false>(t, pool.null_handle());
371
372 if (n == NULL)
373 return false;
374
375 for (;;) {
376 tagged_node_handle tail = tail_.load(memory_order_relaxed);
377 tagged_node_handle next = tail->next.load(memory_order_relaxed);
378 node * next_ptr = next.get_ptr();
379
380 if (next_ptr == 0) {
381 tail->next.store(tagged_node_handle(n, next.get_next_tag()), memory_order_relaxed);
382 tail_.store(tagged_node_handle(n, tail.get_next_tag()), memory_order_relaxed);
383 return true;
384 }
385 else
386 tail_.store(tagged_node_handle(next_ptr, tail.get_next_tag()), memory_order_relaxed);
387 }
388 }
389
390 /** Pops object from queue.
391 *
392 * \post if pop operation is successful, object will be copied to ret.
393 * \returns true, if the pop operation is successful, false if queue was empty.
394 *
395 * \note Thread-safe and non-blocking
396 * */
397 bool pop (T & ret)
398 {
399 return pop<T>(ret);
400 }
401
402 /** Pops object from queue.
403 *
404 * \pre type U must be constructible by T and copyable, or T must be convertible to U
405 * \post if pop operation is successful, object will be copied to ret.
406 * \returns true, if the pop operation is successful, false if queue was empty.
407 *
408 * \note Thread-safe and non-blocking
409 * */
410 template <typename U>
411 bool pop (U & ret)
412 {
413 for (;;) {
414 tagged_node_handle head = head_.load(memory_order_acquire);
415 node * head_ptr = pool.get_pointer(head);
416
417 tagged_node_handle tail = tail_.load(memory_order_acquire);
418 tagged_node_handle next = head_ptr->next.load(memory_order_acquire);
419 node * next_ptr = pool.get_pointer(next);
420
421 tagged_node_handle head2 = head_.load(memory_order_acquire);
422 if (BOOST_LIKELY(head == head2)) {
423 if (pool.get_handle(head) == pool.get_handle(tail)) {
424 if (next_ptr == 0)
425 return false;
426
427 tagged_node_handle new_tail(pool.get_handle(next), tail.get_next_tag());
428 tail_.compare_exchange_strong(tail, new_tail);
429
430 } else {
431 if (next_ptr == 0)
432 /* this check is not part of the original algorithm as published by michael and scott
433 *
434 * however we reuse the tagged_ptr part for the freelist and clear the next part during node
435 * allocation. we can observe a null-pointer here.
436 * */
437 continue;
438 detail::copy_payload(next_ptr->data, ret);
439
440 tagged_node_handle new_head(pool.get_handle(next), head.get_next_tag());
441 if (head_.compare_exchange_weak(head, new_head)) {
442 pool.template destruct<true>(head);
443 return true;
444 }
445 }
446 }
447 }
448 }
449
450 /** Pops object from queue.
451 *
452 * \post if pop operation is successful, object will be copied to ret.
453 * \returns true, if the pop operation is successful, false if queue was empty.
454 *
455 * \note Not thread-safe, but non-blocking
456 *
457 * */
458 bool unsynchronized_pop (T & ret)
459 {
460 return unsynchronized_pop<T>(ret);
461 }
462
463 /** Pops object from queue.
464 *
465 * \pre type U must be constructible by T and copyable, or T must be convertible to U
466 * \post if pop operation is successful, object will be copied to ret.
467 * \returns true, if the pop operation is successful, false if queue was empty.
468 *
469 * \note Not thread-safe, but non-blocking
470 *
471 * */
472 template <typename U>
473 bool unsynchronized_pop (U & ret)
474 {
475 for (;;) {
476 tagged_node_handle head = head_.load(memory_order_relaxed);
477 node * head_ptr = pool.get_pointer(head);
478 tagged_node_handle tail = tail_.load(memory_order_relaxed);
479 tagged_node_handle next = head_ptr->next.load(memory_order_relaxed);
480 node * next_ptr = pool.get_pointer(next);
481
482 if (pool.get_handle(head) == pool.get_handle(tail)) {
483 if (next_ptr == 0)
484 return false;
485
486 tagged_node_handle new_tail(pool.get_handle(next), tail.get_next_tag());
487 tail_.store(new_tail);
488 } else {
489 if (next_ptr == 0)
490 /* this check is not part of the original algorithm as published by michael and scott
491 *
492 * however we reuse the tagged_ptr part for the freelist and clear the next part during node
493 * allocation. we can observe a null-pointer here.
494 * */
495 continue;
496 detail::copy_payload(next_ptr->data, ret);
497 tagged_node_handle new_head(pool.get_handle(next), head.get_next_tag());
498 head_.store(new_head);
499 pool.template destruct<false>(head);
500 return true;
501 }
502 }
503 }
504
505 /** consumes one element via a functor
506 *
507 * pops one element from the queue and applies the functor on this object
508 *
509 * \returns true, if one element was consumed
510 *
511 * \note Thread-safe and non-blocking, if functor is thread-safe and non-blocking
512 * */
513 template <typename Functor>
514 bool consume_one(Functor & f)
515 {
516 T element;
517 bool success = pop(element);
518 if (success)
519 f(element);
520
521 return success;
522 }
523
524 /// \copydoc boost::lockfree::queue::consume_one(Functor & rhs)
525 template <typename Functor>
526 bool consume_one(Functor const & f)
527 {
528 T element;
529 bool success = pop(element);
530 if (success)
531 f(element);
532
533 return success;
534 }
535
536 /** consumes all elements via a functor
537 *
538 * sequentially pops all elements from the queue and applies the functor on each object
539 *
540 * \returns number of elements that are consumed
541 *
542 * \note Thread-safe and non-blocking, if functor is thread-safe and non-blocking
543 * */
544 template <typename Functor>
545 size_t consume_all(Functor & f)
546 {
547 size_t element_count = 0;
548 while (consume_one(f))
549 element_count += 1;
550
551 return element_count;
552 }
553
554 /// \copydoc boost::lockfree::queue::consume_all(Functor & rhs)
555 template <typename Functor>
556 size_t consume_all(Functor const & f)
557 {
558 size_t element_count = 0;
559 while (consume_one(f))
560 element_count += 1;
561
562 return element_count;
563 }
564
565private:
566#ifndef BOOST_DOXYGEN_INVOKED
567 atomic<tagged_node_handle> head_;
568 static const int padding_size = BOOST_LOCKFREE_CACHELINE_BYTES - sizeof(tagged_node_handle);
569 char padding1[padding_size];
570 atomic<tagged_node_handle> tail_;
571 char padding2[padding_size];
572
573 pool_t pool;
574#endif
575};
576
577} /* namespace lockfree */
578} /* namespace boost */
579
580#if defined(BOOST_INTEL) && (BOOST_INTEL_CXX_VERSION > 1000)
581#pragma warning(pop)
582#endif
583
584#if defined(_MSC_VER)
585#pragma warning(pop)
586#endif
587
588#endif /* BOOST_LOCKFREE_FIFO_HPP_INCLUDED */
589

source code of boost/libs/lockfree/include/boost/lockfree/queue.hpp