diff options
Diffstat (limited to 'boost/lockfree/queue.hpp')
| -rw-r--r-- | boost/lockfree/queue.hpp | 550 |
1 files changed, 550 insertions, 0 deletions
diff --git a/boost/lockfree/queue.hpp b/boost/lockfree/queue.hpp new file mode 100644 index 0000000000..74001fba9d --- /dev/null +++ b/boost/lockfree/queue.hpp @@ -0,0 +1,550 @@ +// lock-free queue from +// Michael, M. M. and Scott, M. L., +// "simple, fast and practical non-blocking and blocking concurrent queue algorithms" +// +// Copyright (C) 2008-2013 Tim Blechmann +// +// Distributed under the Boost Software License, Version 1.0. (See +// accompanying file LICENSE_1_0.txt or copy at +// http://www.boost.org/LICENSE_1_0.txt) + +#ifndef BOOST_LOCKFREE_FIFO_HPP_INCLUDED +#define BOOST_LOCKFREE_FIFO_HPP_INCLUDED + +#include <boost/assert.hpp> +#include <boost/static_assert.hpp> +#include <boost/type_traits/has_trivial_assign.hpp> +#include <boost/type_traits/has_trivial_destructor.hpp> + +#include <boost/lockfree/detail/atomic.hpp> +#include <boost/lockfree/detail/copy_payload.hpp> +#include <boost/lockfree/detail/freelist.hpp> +#include <boost/lockfree/detail/parameter.hpp> +#include <boost/lockfree/detail/tagged_ptr.hpp> + +#ifdef BOOST_HAS_PRAGMA_ONCE +#pragma once +#endif + + +#if defined(_MSC_VER) +#pragma warning(push) +#pragma warning(disable: 4324) // structure was padded due to __declspec(align()) +#endif + + +namespace boost { +namespace lockfree { +namespace detail { + +typedef parameter::parameters<boost::parameter::optional<tag::allocator>, + boost::parameter::optional<tag::capacity> + > queue_signature; + +} /* namespace detail */ + + +/** The queue class provides a multi-writer/multi-reader queue, pushing and popping is lock-free, + * construction/destruction has to be synchronized. It uses a freelist for memory management, + * freed nodes are pushed to the freelist and not returned to the OS before the queue is destroyed. + * + * \b Policies: + * - \ref boost::lockfree::fixed_sized, defaults to \c boost::lockfree::fixed_sized<false> \n + * Can be used to completely disable dynamic memory allocations during push in order to ensure lockfree behavior. \n + * If the data structure is configured as fixed-sized, the internal nodes are stored inside an array and they are addressed + * by array indexing. This limits the possible size of the queue to the number of elements that can be addressed by the index + * type (usually 2**16-2), but on platforms that lack double-width compare-and-exchange instructions, this is the best way + * to achieve lock-freedom. + * + * - \ref boost::lockfree::capacity, optional \n + * If this template argument is passed to the options, the size of the queue is set at compile-time.\n + * It this option implies \c fixed_sized<true> + * + * - \ref boost::lockfree::allocator, defaults to \c boost::lockfree::allocator<std::allocator<void>> \n + * Specifies the allocator that is used for the internal freelist + * + * \b Requirements: + * - T must have a copy constructor + * - T must have a trivial assignment operator + * - T must have a trivial destructor + * + * */ +#ifndef BOOST_DOXYGEN_INVOKED +template <typename T, + class A0 = boost::parameter::void_, + class A1 = boost::parameter::void_, + class A2 = boost::parameter::void_> +#else +template <typename T, ...Options> +#endif +class queue +{ +private: +#ifndef BOOST_DOXYGEN_INVOKED + +#ifdef BOOST_HAS_TRIVIAL_DESTRUCTOR + BOOST_STATIC_ASSERT((boost::has_trivial_destructor<T>::value)); +#endif + +#ifdef BOOST_HAS_TRIVIAL_ASSIGN + BOOST_STATIC_ASSERT((boost::has_trivial_assign<T>::value)); +#endif + + typedef typename detail::queue_signature::bind<A0, A1, A2>::type bound_args; + + static const bool has_capacity = detail::extract_capacity<bound_args>::has_capacity; + static const size_t capacity = detail::extract_capacity<bound_args>::capacity + 1; // the queue uses one dummy node + static const bool fixed_sized = detail::extract_fixed_sized<bound_args>::value; + static const bool node_based = !(has_capacity || fixed_sized); + static const bool compile_time_sized = has_capacity; + + struct BOOST_LOCKFREE_CACHELINE_ALIGNMENT node + { + typedef typename detail::select_tagged_handle<node, node_based>::tagged_handle_type tagged_node_handle; + typedef typename detail::select_tagged_handle<node, node_based>::handle_type handle_type; + + node(T const & v, handle_type null_handle): + data(v)//, next(tagged_node_handle(0, 0)) + { + /* increment tag to avoid ABA problem */ + tagged_node_handle old_next = next.load(memory_order_relaxed); + tagged_node_handle new_next (null_handle, old_next.get_next_tag()); + next.store(new_next, memory_order_release); + } + + node (handle_type null_handle): + next(tagged_node_handle(null_handle, 0)) + {} + + node(void) + {} + + atomic<tagged_node_handle> next; + T data; + }; + + typedef typename detail::extract_allocator<bound_args, node>::type node_allocator; + typedef typename detail::select_freelist<node, node_allocator, compile_time_sized, fixed_sized, capacity>::type pool_t; + typedef typename pool_t::tagged_node_handle tagged_node_handle; + typedef typename detail::select_tagged_handle<node, node_based>::handle_type handle_type; + + void initialize(void) + { + node * n = pool.template construct<true, false>(pool.null_handle()); + tagged_node_handle dummy_node(pool.get_handle(n), 0); + head_.store(dummy_node, memory_order_relaxed); + tail_.store(dummy_node, memory_order_release); + } + + struct implementation_defined + { + typedef node_allocator allocator; + typedef std::size_t size_type; + }; + +#endif + + BOOST_DELETED_FUNCTION(queue(queue const&)) + BOOST_DELETED_FUNCTION(queue& operator= (queue const&)) + +public: + typedef T value_type; + typedef typename implementation_defined::allocator allocator; + typedef typename implementation_defined::size_type size_type; + + /** + * \return true, if implementation is lock-free. + * + * \warning It only checks, if the queue head and tail nodes and the freelist can be modified in a lock-free manner. + * On most platforms, the whole implementation is lock-free, if this is true. Using c++0x-style atomics, there is + * no possibility to provide a completely accurate implementation, because one would need to test every internal + * node, which is impossible if further nodes will be allocated from the operating system. + * */ + bool is_lock_free (void) const + { + return head_.is_lock_free() && tail_.is_lock_free() && pool.is_lock_free(); + } + + //! Construct queue + // @{ + queue(void): + head_(tagged_node_handle(0, 0)), + tail_(tagged_node_handle(0, 0)), + pool(node_allocator(), capacity) + { + BOOST_ASSERT(has_capacity); + initialize(); + } + + template <typename U> + explicit queue(typename node_allocator::template rebind<U>::other const & alloc): + head_(tagged_node_handle(0, 0)), + tail_(tagged_node_handle(0, 0)), + pool(alloc, capacity) + { + BOOST_STATIC_ASSERT(has_capacity); + initialize(); + } + + explicit queue(allocator const & alloc): + head_(tagged_node_handle(0, 0)), + tail_(tagged_node_handle(0, 0)), + pool(alloc, capacity) + { + BOOST_ASSERT(has_capacity); + initialize(); + } + // @} + + //! Construct queue, allocate n nodes for the freelist. + // @{ + explicit queue(size_type n): + head_(tagged_node_handle(0, 0)), + tail_(tagged_node_handle(0, 0)), + pool(node_allocator(), n + 1) + { + BOOST_ASSERT(!has_capacity); + initialize(); + } + + template <typename U> + queue(size_type n, typename node_allocator::template rebind<U>::other const & alloc): + head_(tagged_node_handle(0, 0)), + tail_(tagged_node_handle(0, 0)), + pool(alloc, n + 1) + { + BOOST_STATIC_ASSERT(!has_capacity); + initialize(); + } + // @} + + /** \copydoc boost::lockfree::stack::reserve + * */ + void reserve(size_type n) + { + pool.template reserve<true>(n); + } + + /** \copydoc boost::lockfree::stack::reserve_unsafe + * */ + void reserve_unsafe(size_type n) + { + pool.template reserve<false>(n); + } + + /** Destroys queue, free all nodes from freelist. + * */ + ~queue(void) + { + T dummy; + while(unsynchronized_pop(dummy)) + {} + + pool.template destruct<false>(head_.load(memory_order_relaxed)); + } + + /** Check if the queue is empty + * + * \return true, if the queue is empty, false otherwise + * \note The result is only accurate, if no other thread modifies the queue. Therefore it is rarely practical to use this + * value in program logic. + * */ + bool empty(void) const + { + return pool.get_handle(head_.load()) == pool.get_handle(tail_.load()); + } + + /** Pushes object t to the queue. + * + * \post object will be pushed to the queue, if internal node can be allocated + * \returns true, if the push operation is successful. + * + * \note Thread-safe. If internal memory pool is exhausted and the memory pool is not fixed-sized, a new node will be allocated + * from the OS. This may not be lock-free. + * */ + bool push(T const & t) + { + return do_push<false>(t); + } + + /** Pushes object t to the queue. + * + * \post object will be pushed to the queue, if internal node can be allocated + * \returns true, if the push operation is successful. + * + * \note Thread-safe and non-blocking. If internal memory pool is exhausted, operation will fail + * \throws if memory allocator throws + * */ + bool bounded_push(T const & t) + { + return do_push<true>(t); + } + + +private: +#ifndef BOOST_DOXYGEN_INVOKED + template <bool Bounded> + bool do_push(T const & t) + { + using detail::likely; + + node * n = pool.template construct<true, Bounded>(t, pool.null_handle()); + handle_type node_handle = pool.get_handle(n); + + if (n == NULL) + return false; + + for (;;) { + tagged_node_handle tail = tail_.load(memory_order_acquire); + node * tail_node = pool.get_pointer(tail); + tagged_node_handle next = tail_node->next.load(memory_order_acquire); + node * next_ptr = pool.get_pointer(next); + + tagged_node_handle tail2 = tail_.load(memory_order_acquire); + if (likely(tail == tail2)) { + if (next_ptr == 0) { + tagged_node_handle new_tail_next(node_handle, next.get_next_tag()); + if ( tail_node->next.compare_exchange_weak(next, new_tail_next) ) { + tagged_node_handle new_tail(node_handle, tail.get_next_tag()); + tail_.compare_exchange_strong(tail, new_tail); + return true; + } + } + else { + tagged_node_handle new_tail(pool.get_handle(next_ptr), tail.get_next_tag()); + tail_.compare_exchange_strong(tail, new_tail); + } + } + } + } +#endif + +public: + + /** Pushes object t to the queue. + * + * \post object will be pushed to the queue, if internal node can be allocated + * \returns true, if the push operation is successful. + * + * \note Not Thread-safe. If internal memory pool is exhausted and the memory pool is not fixed-sized, a new node will be allocated + * from the OS. This may not be lock-free. + * \throws if memory allocator throws + * */ + bool unsynchronized_push(T const & t) + { + node * n = pool.template construct<false, false>(t, pool.null_handle()); + + if (n == NULL) + return false; + + for (;;) { + tagged_node_handle tail = tail_.load(memory_order_relaxed); + tagged_node_handle next = tail->next.load(memory_order_relaxed); + node * next_ptr = next.get_ptr(); + + if (next_ptr == 0) { + tail->next.store(tagged_node_handle(n, next.get_next_tag()), memory_order_relaxed); + tail_.store(tagged_node_handle(n, tail.get_next_tag()), memory_order_relaxed); + return true; + } + else + tail_.store(tagged_node_handle(next_ptr, tail.get_next_tag()), memory_order_relaxed); + } + } + + /** Pops object from queue. + * + * \post if pop operation is successful, object will be copied to ret. + * \returns true, if the pop operation is successful, false if queue was empty. + * + * \note Thread-safe and non-blocking + * */ + bool pop (T & ret) + { + return pop<T>(ret); + } + + /** Pops object from queue. + * + * \pre type U must be constructible by T and copyable, or T must be convertible to U + * \post if pop operation is successful, object will be copied to ret. + * \returns true, if the pop operation is successful, false if queue was empty. + * + * \note Thread-safe and non-blocking + * */ + template <typename U> + bool pop (U & ret) + { + using detail::likely; + for (;;) { + tagged_node_handle head = head_.load(memory_order_acquire); + node * head_ptr = pool.get_pointer(head); + + tagged_node_handle tail = tail_.load(memory_order_acquire); + tagged_node_handle next = head_ptr->next.load(memory_order_acquire); + node * next_ptr = pool.get_pointer(next); + + tagged_node_handle head2 = head_.load(memory_order_acquire); + if (likely(head == head2)) { + if (pool.get_handle(head) == pool.get_handle(tail)) { + if (next_ptr == 0) + return false; + + tagged_node_handle new_tail(pool.get_handle(next), tail.get_next_tag()); + tail_.compare_exchange_strong(tail, new_tail); + + } else { + if (next_ptr == 0) + /* this check is not part of the original algorithm as published by michael and scott + * + * however we reuse the tagged_ptr part for the freelist and clear the next part during node + * allocation. we can observe a null-pointer here. + * */ + continue; + detail::copy_payload(next_ptr->data, ret); + + tagged_node_handle new_head(pool.get_handle(next), head.get_next_tag()); + if (head_.compare_exchange_weak(head, new_head)) { + pool.template destruct<true>(head); + return true; + } + } + } + } + } + + /** Pops object from queue. + * + * \post if pop operation is successful, object will be copied to ret. + * \returns true, if the pop operation is successful, false if queue was empty. + * + * \note Not thread-safe, but non-blocking + * + * */ + bool unsynchronized_pop (T & ret) + { + return unsynchronized_pop<T>(ret); + } + + /** Pops object from queue. + * + * \pre type U must be constructible by T and copyable, or T must be convertible to U + * \post if pop operation is successful, object will be copied to ret. + * \returns true, if the pop operation is successful, false if queue was empty. + * + * \note Not thread-safe, but non-blocking + * + * */ + template <typename U> + bool unsynchronized_pop (U & ret) + { + for (;;) { + tagged_node_handle head = head_.load(memory_order_relaxed); + node * head_ptr = pool.get_pointer(head); + tagged_node_handle tail = tail_.load(memory_order_relaxed); + tagged_node_handle next = head_ptr->next.load(memory_order_relaxed); + node * next_ptr = pool.get_pointer(next); + + if (pool.get_handle(head) == pool.get_handle(tail)) { + if (next_ptr == 0) + return false; + + tagged_node_handle new_tail(pool.get_handle(next), tail.get_next_tag()); + tail_.store(new_tail); + } else { + if (next_ptr == 0) + /* this check is not part of the original algorithm as published by michael and scott + * + * however we reuse the tagged_ptr part for the freelist and clear the next part during node + * allocation. we can observe a null-pointer here. + * */ + continue; + detail::copy_payload(next_ptr->data, ret); + tagged_node_handle new_head(pool.get_handle(next), head.get_next_tag()); + head_.store(new_head); + pool.template destruct<false>(head); + return true; + } + } + } + + /** consumes one element via a functor + * + * pops one element from the queue and applies the functor on this object + * + * \returns true, if one element was consumed + * + * \note Thread-safe and non-blocking, if functor is thread-safe and non-blocking + * */ + template <typename Functor> + bool consume_one(Functor & f) + { + T element; + bool success = pop(element); + if (success) + f(element); + + return success; + } + + /// \copydoc boost::lockfree::queue::consume_one(Functor & rhs) + template <typename Functor> + bool consume_one(Functor const & f) + { + T element; + bool success = pop(element); + if (success) + f(element); + + return success; + } + + /** consumes all elements via a functor + * + * sequentially pops all elements from the queue and applies the functor on each object + * + * \returns number of elements that are consumed + * + * \note Thread-safe and non-blocking, if functor is thread-safe and non-blocking + * */ + template <typename Functor> + size_t consume_all(Functor & f) + { + size_t element_count = 0; + while (consume_one(f)) + element_count += 1; + + return element_count; + } + + /// \copydoc boost::lockfree::queue::consume_all(Functor & rhs) + template <typename Functor> + size_t consume_all(Functor const & f) + { + size_t element_count = 0; + while (consume_one(f)) + element_count += 1; + + return element_count; + } + +private: +#ifndef BOOST_DOXYGEN_INVOKED + atomic<tagged_node_handle> head_; + static const int padding_size = BOOST_LOCKFREE_CACHELINE_BYTES - sizeof(tagged_node_handle); + char padding1[padding_size]; + atomic<tagged_node_handle> tail_; + char padding2[padding_size]; + + pool_t pool; +#endif +}; + +} /* namespace lockfree */ +} /* namespace boost */ + +#if defined(_MSC_VER) +#pragma warning(pop) +#endif + +#endif /* BOOST_LOCKFREE_FIFO_HPP_INCLUDED */ |