1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054 |
- // lock-free single-producer/single-consumer ringbuffer
- // this algorithm is implemented in various projects (linux kernel)
- //
- // Copyright (C) 2009-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_SPSC_QUEUE_HPP_INCLUDED
- #define BOOST_LOCKFREE_SPSC_QUEUE_HPP_INCLUDED
- #include <algorithm>
- #include <memory>
- #include <boost/aligned_storage.hpp>
- #include <boost/assert.hpp>
- #include <boost/static_assert.hpp>
- #include <boost/core/allocator_access.hpp>
- #include <boost/utility.hpp>
- #include <boost/next_prior.hpp>
- #include <boost/utility/enable_if.hpp>
- #include <boost/config.hpp> // for BOOST_LIKELY
- #include <boost/type_traits/has_trivial_destructor.hpp>
- #include <boost/type_traits/is_convertible.hpp>
- #include <boost/lockfree/detail/atomic.hpp>
- #include <boost/lockfree/detail/copy_payload.hpp>
- #include <boost/lockfree/detail/parameter.hpp>
- #include <boost/lockfree/detail/prefix.hpp>
- #include <boost/lockfree/lockfree_forward.hpp>
- #ifdef BOOST_HAS_PRAGMA_ONCE
- #pragma once
- #endif
- namespace boost {
- namespace lockfree {
- namespace detail {
- typedef parameter::parameters<boost::parameter::optional<tag::capacity>,
- boost::parameter::optional<tag::allocator>
- > ringbuffer_signature;
- template <typename T>
- class ringbuffer_base
- {
- #ifndef BOOST_DOXYGEN_INVOKED
- protected:
- typedef std::size_t size_t;
- static const int padding_size = BOOST_LOCKFREE_CACHELINE_BYTES - sizeof(size_t);
- atomic<size_t> write_index_;
- char padding1[padding_size]; /* force read_index and write_index to different cache lines */
- atomic<size_t> read_index_;
- BOOST_DELETED_FUNCTION(ringbuffer_base(ringbuffer_base const&))
- BOOST_DELETED_FUNCTION(ringbuffer_base& operator= (ringbuffer_base const&))
- protected:
- ringbuffer_base(void):
- write_index_(0), read_index_(0)
- {}
- static size_t next_index(size_t arg, size_t max_size)
- {
- size_t ret = arg + 1;
- while (BOOST_UNLIKELY(ret >= max_size))
- ret -= max_size;
- return ret;
- }
- static size_t read_available(size_t write_index, size_t read_index, size_t max_size)
- {
- if (write_index >= read_index)
- return write_index - read_index;
- const size_t ret = write_index + max_size - read_index;
- return ret;
- }
- static size_t write_available(size_t write_index, size_t read_index, size_t max_size)
- {
- size_t ret = read_index - write_index - 1;
- if (write_index >= read_index)
- ret += max_size;
- return ret;
- }
- size_t read_available(size_t max_size) const
- {
- size_t write_index = write_index_.load(memory_order_acquire);
- const size_t read_index = read_index_.load(memory_order_relaxed);
- return read_available(write_index, read_index, max_size);
- }
- size_t write_available(size_t max_size) const
- {
- size_t write_index = write_index_.load(memory_order_relaxed);
- const size_t read_index = read_index_.load(memory_order_acquire);
- return write_available(write_index, read_index, max_size);
- }
- bool push(T const & t, T * buffer, size_t max_size)
- {
- const size_t write_index = write_index_.load(memory_order_relaxed); // only written from push thread
- const size_t next = next_index(write_index, max_size);
- if (next == read_index_.load(memory_order_acquire))
- return false; /* ringbuffer is full */
- new (buffer + write_index) T(t); // copy-construct
- write_index_.store(next, memory_order_release);
- return true;
- }
- size_t push(const T * input_buffer, size_t input_count, T * internal_buffer, size_t max_size)
- {
- return push(input_buffer, input_buffer + input_count, internal_buffer, max_size) - input_buffer;
- }
- template <typename ConstIterator>
- ConstIterator push(ConstIterator begin, ConstIterator end, T * internal_buffer, size_t max_size)
- {
- // FIXME: avoid std::distance
- const size_t write_index = write_index_.load(memory_order_relaxed); // only written from push thread
- const size_t read_index = read_index_.load(memory_order_acquire);
- const size_t avail = write_available(write_index, read_index, max_size);
- if (avail == 0)
- return begin;
- size_t input_count = std::distance(begin, end);
- input_count = (std::min)(input_count, avail);
- size_t new_write_index = write_index + input_count;
- const ConstIterator last = boost::next(begin, input_count);
- if (write_index + input_count > max_size) {
- /* copy data in two sections */
- const size_t count0 = max_size - write_index;
- const ConstIterator midpoint = boost::next(begin, count0);
- std::uninitialized_copy(begin, midpoint, internal_buffer + write_index);
- std::uninitialized_copy(midpoint, last, internal_buffer);
- new_write_index -= max_size;
- } else {
- std::uninitialized_copy(begin, last, internal_buffer + write_index);
- if (new_write_index == max_size)
- new_write_index = 0;
- }
- write_index_.store(new_write_index, memory_order_release);
- return last;
- }
- template <typename Functor>
- bool consume_one(Functor & functor, T * buffer, size_t max_size)
- {
- const size_t write_index = write_index_.load(memory_order_acquire);
- const size_t read_index = read_index_.load(memory_order_relaxed); // only written from pop thread
- if ( empty(write_index, read_index) )
- return false;
- T & object_to_consume = buffer[read_index];
- functor( object_to_consume );
- object_to_consume.~T();
- size_t next = next_index(read_index, max_size);
- read_index_.store(next, memory_order_release);
- return true;
- }
- template <typename Functor>
- bool consume_one(Functor const & functor, T * buffer, size_t max_size)
- {
- const size_t write_index = write_index_.load(memory_order_acquire);
- const size_t read_index = read_index_.load(memory_order_relaxed); // only written from pop thread
- if ( empty(write_index, read_index) )
- return false;
- T & object_to_consume = buffer[read_index];
- functor( object_to_consume );
- object_to_consume.~T();
- size_t next = next_index(read_index, max_size);
- read_index_.store(next, memory_order_release);
- return true;
- }
- template <typename Functor>
- size_t consume_all (Functor const & functor, T * internal_buffer, size_t max_size)
- {
- const size_t write_index = write_index_.load(memory_order_acquire);
- const size_t read_index = read_index_.load(memory_order_relaxed); // only written from pop thread
- const size_t avail = read_available(write_index, read_index, max_size);
- if (avail == 0)
- return 0;
- const size_t output_count = avail;
- size_t new_read_index = read_index + output_count;
- if (read_index + output_count > max_size) {
- /* copy data in two sections */
- const size_t count0 = max_size - read_index;
- const size_t count1 = output_count - count0;
- run_functor_and_delete(internal_buffer + read_index, internal_buffer + max_size, functor);
- run_functor_and_delete(internal_buffer, internal_buffer + count1, functor);
- new_read_index -= max_size;
- } else {
- run_functor_and_delete(internal_buffer + read_index, internal_buffer + read_index + output_count, functor);
- if (new_read_index == max_size)
- new_read_index = 0;
- }
- read_index_.store(new_read_index, memory_order_release);
- return output_count;
- }
- template <typename Functor>
- size_t consume_all (Functor & functor, T * internal_buffer, size_t max_size)
- {
- const size_t write_index = write_index_.load(memory_order_acquire);
- const size_t read_index = read_index_.load(memory_order_relaxed); // only written from pop thread
- const size_t avail = read_available(write_index, read_index, max_size);
- if (avail == 0)
- return 0;
- const size_t output_count = avail;
- size_t new_read_index = read_index + output_count;
- if (read_index + output_count > max_size) {
- /* copy data in two sections */
- const size_t count0 = max_size - read_index;
- const size_t count1 = output_count - count0;
- run_functor_and_delete(internal_buffer + read_index, internal_buffer + max_size, functor);
- run_functor_and_delete(internal_buffer, internal_buffer + count1, functor);
- new_read_index -= max_size;
- } else {
- run_functor_and_delete(internal_buffer + read_index, internal_buffer + read_index + output_count, functor);
- if (new_read_index == max_size)
- new_read_index = 0;
- }
- read_index_.store(new_read_index, memory_order_release);
- return output_count;
- }
- size_t pop (T * output_buffer, size_t output_count, T * internal_buffer, size_t max_size)
- {
- const size_t write_index = write_index_.load(memory_order_acquire);
- const size_t read_index = read_index_.load(memory_order_relaxed); // only written from pop thread
- const size_t avail = read_available(write_index, read_index, max_size);
- if (avail == 0)
- return 0;
- output_count = (std::min)(output_count, avail);
- size_t new_read_index = read_index + output_count;
- if (read_index + output_count > max_size) {
- /* copy data in two sections */
- const size_t count0 = max_size - read_index;
- const size_t count1 = output_count - count0;
- copy_and_delete(internal_buffer + read_index, internal_buffer + max_size, output_buffer);
- copy_and_delete(internal_buffer, internal_buffer + count1, output_buffer + count0);
- new_read_index -= max_size;
- } else {
- copy_and_delete(internal_buffer + read_index, internal_buffer + read_index + output_count, output_buffer);
- if (new_read_index == max_size)
- new_read_index = 0;
- }
- read_index_.store(new_read_index, memory_order_release);
- return output_count;
- }
- template <typename OutputIterator>
- size_t pop_to_output_iterator (OutputIterator it, T * internal_buffer, size_t max_size)
- {
- const size_t write_index = write_index_.load(memory_order_acquire);
- const size_t read_index = read_index_.load(memory_order_relaxed); // only written from pop thread
- const size_t avail = read_available(write_index, read_index, max_size);
- if (avail == 0)
- return 0;
- size_t new_read_index = read_index + avail;
- if (read_index + avail > max_size) {
- /* copy data in two sections */
- const size_t count0 = max_size - read_index;
- const size_t count1 = avail - count0;
- it = copy_and_delete(internal_buffer + read_index, internal_buffer + max_size, it);
- copy_and_delete(internal_buffer, internal_buffer + count1, it);
- new_read_index -= max_size;
- } else {
- copy_and_delete(internal_buffer + read_index, internal_buffer + read_index + avail, it);
- if (new_read_index == max_size)
- new_read_index = 0;
- }
- read_index_.store(new_read_index, memory_order_release);
- return avail;
- }
- const T& front(const T * internal_buffer) const
- {
- const size_t read_index = read_index_.load(memory_order_relaxed); // only written from pop thread
- return *(internal_buffer + read_index);
- }
- T& front(T * internal_buffer)
- {
- const size_t read_index = read_index_.load(memory_order_relaxed); // only written from pop thread
- return *(internal_buffer + read_index);
- }
- #endif
- public:
- /** reset the ringbuffer
- *
- * \note Not thread-safe
- * */
- void reset(void)
- {
- if ( !boost::has_trivial_destructor<T>::value ) {
- // make sure to call all destructors!
- detail::consume_noop consume_functor;
- (void)consume_all( consume_functor );
- } else {
- write_index_.store(0, memory_order_relaxed);
- read_index_.store(0, memory_order_release);
- }
- }
- /** Check if the ringbuffer is empty
- *
- * \return true, if the ringbuffer is empty, false otherwise
- * \note Due to the concurrent nature of the ringbuffer the result may be inaccurate.
- * */
- bool empty(void)
- {
- return empty(write_index_.load(memory_order_relaxed), read_index_.load(memory_order_relaxed));
- }
- /**
- * \return true, if implementation is lock-free.
- *
- * */
- bool is_lock_free(void) const
- {
- return write_index_.is_lock_free() && read_index_.is_lock_free();
- }
- private:
- bool empty(size_t write_index, size_t read_index)
- {
- return write_index == read_index;
- }
- template< class OutputIterator >
- OutputIterator copy_and_delete( T * first, T * last, OutputIterator out )
- {
- if (boost::has_trivial_destructor<T>::value) {
- return std::copy(first, last, out); // will use memcpy if possible
- } else {
- for (; first != last; ++first, ++out) {
- *out = *first;
- first->~T();
- }
- return out;
- }
- }
- template< class Functor >
- void run_functor_and_delete( T * first, T * last, Functor & functor )
- {
- for (; first != last; ++first) {
- functor(*first);
- first->~T();
- }
- }
- template< class Functor >
- void run_functor_and_delete( T * first, T * last, Functor const & functor )
- {
- for (; first != last; ++first) {
- functor(*first);
- first->~T();
- }
- }
- };
- template <typename T, std::size_t MaxSize>
- class compile_time_sized_ringbuffer:
- public ringbuffer_base<T>
- {
- typedef std::size_t size_type;
- static const std::size_t max_size = MaxSize + 1;
- typedef typename boost::aligned_storage<max_size * sizeof(T),
- boost::alignment_of<T>::value
- >::type storage_type;
- storage_type storage_;
- T * data()
- {
- return static_cast<T*>(storage_.address());
- }
- const T * data() const
- {
- return static_cast<const T*>(storage_.address());
- }
- protected:
- size_type max_number_of_elements() const
- {
- return max_size;
- }
- ~compile_time_sized_ringbuffer(void)
- {
- // destroy all remaining items
- detail::consume_noop consume_functor;
- (void)consume_all(consume_functor);
- }
- public:
- bool push(T const & t)
- {
- return ringbuffer_base<T>::push(t, data(), max_size);
- }
- template <typename Functor>
- bool consume_one(Functor & f)
- {
- return ringbuffer_base<T>::consume_one(f, data(), max_size);
- }
- template <typename Functor>
- bool consume_one(Functor const & f)
- {
- return ringbuffer_base<T>::consume_one(f, data(), max_size);
- }
- template <typename Functor>
- size_type consume_all(Functor & f)
- {
- return ringbuffer_base<T>::consume_all(f, data(), max_size);
- }
- template <typename Functor>
- size_type consume_all(Functor const & f)
- {
- return ringbuffer_base<T>::consume_all(f, data(), max_size);
- }
- size_type push(T const * t, size_type size)
- {
- return ringbuffer_base<T>::push(t, size, data(), max_size);
- }
- template <size_type size>
- size_type push(T const (&t)[size])
- {
- return push(t, size);
- }
- template <typename ConstIterator>
- ConstIterator push(ConstIterator begin, ConstIterator end)
- {
- return ringbuffer_base<T>::push(begin, end, data(), max_size);
- }
- size_type pop(T * ret, size_type size)
- {
- return ringbuffer_base<T>::pop(ret, size, data(), max_size);
- }
- template <typename OutputIterator>
- size_type pop_to_output_iterator(OutputIterator it)
- {
- return ringbuffer_base<T>::pop_to_output_iterator(it, data(), max_size);
- }
- const T& front(void) const
- {
- return ringbuffer_base<T>::front(data());
- }
- T& front(void)
- {
- return ringbuffer_base<T>::front(data());
- }
- };
- template <typename T, typename Alloc>
- class runtime_sized_ringbuffer:
- public ringbuffer_base<T>,
- private Alloc
- {
- typedef std::size_t size_type;
- size_type max_elements_;
- #ifdef BOOST_NO_CXX11_ALLOCATOR
- typedef typename Alloc::pointer pointer;
- #else
- typedef std::allocator_traits<Alloc> allocator_traits;
- typedef typename allocator_traits::pointer pointer;
- #endif
- pointer array_;
- protected:
- size_type max_number_of_elements() const
- {
- return max_elements_;
- }
- public:
- explicit runtime_sized_ringbuffer(size_type max_elements):
- max_elements_(max_elements + 1)
- {
- #ifdef BOOST_NO_CXX11_ALLOCATOR
- array_ = Alloc::allocate(max_elements_);
- #else
- Alloc& alloc = *this;
- array_ = allocator_traits::allocate(alloc, max_elements_);
- #endif
- }
- template <typename U>
- runtime_sized_ringbuffer(typename boost::allocator_rebind<Alloc, U>::type const & alloc, size_type max_elements):
- Alloc(alloc), max_elements_(max_elements + 1)
- {
- #ifdef BOOST_NO_CXX11_ALLOCATOR
- array_ = Alloc::allocate(max_elements_);
- #else
- Alloc& allocator = *this;
- array_ = allocator_traits::allocate(allocator, max_elements_);
- #endif
- }
- runtime_sized_ringbuffer(Alloc const & alloc, size_type max_elements):
- Alloc(alloc), max_elements_(max_elements + 1)
- {
- #ifdef BOOST_NO_CXX11_ALLOCATOR
- array_ = Alloc::allocate(max_elements_);
- #else
- Alloc& allocator = *this;
- array_ = allocator_traits::allocate(allocator, max_elements_);
- #endif
- }
- ~runtime_sized_ringbuffer(void)
- {
- // destroy all remaining items
- detail::consume_noop consume_functor;
- (void)consume_all(consume_functor);
- #ifdef BOOST_NO_CXX11_ALLOCATOR
- Alloc::deallocate(array_, max_elements_);
- #else
- Alloc& allocator = *this;
- allocator_traits::deallocate(allocator, array_, max_elements_);
- #endif
- }
- bool push(T const & t)
- {
- return ringbuffer_base<T>::push(t, &*array_, max_elements_);
- }
- template <typename Functor>
- bool consume_one(Functor & f)
- {
- return ringbuffer_base<T>::consume_one(f, &*array_, max_elements_);
- }
- template <typename Functor>
- bool consume_one(Functor const & f)
- {
- return ringbuffer_base<T>::consume_one(f, &*array_, max_elements_);
- }
- template <typename Functor>
- size_type consume_all(Functor & f)
- {
- return ringbuffer_base<T>::consume_all(f, &*array_, max_elements_);
- }
- template <typename Functor>
- size_type consume_all(Functor const & f)
- {
- return ringbuffer_base<T>::consume_all(f, &*array_, max_elements_);
- }
- size_type push(T const * t, size_type size)
- {
- return ringbuffer_base<T>::push(t, size, &*array_, max_elements_);
- }
- template <size_type size>
- size_type push(T const (&t)[size])
- {
- return push(t, size);
- }
- template <typename ConstIterator>
- ConstIterator push(ConstIterator begin, ConstIterator end)
- {
- return ringbuffer_base<T>::push(begin, end, &*array_, max_elements_);
- }
- size_type pop(T * ret, size_type size)
- {
- return ringbuffer_base<T>::pop(ret, size, &*array_, max_elements_);
- }
- template <typename OutputIterator>
- size_type pop_to_output_iterator(OutputIterator it)
- {
- return ringbuffer_base<T>::pop_to_output_iterator(it, &*array_, max_elements_);
- }
- const T& front(void) const
- {
- return ringbuffer_base<T>::front(&*array_);
- }
- T& front(void)
- {
- return ringbuffer_base<T>::front(&*array_);
- }
- };
- #ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
- template <typename T, typename A0, typename A1>
- #else
- template <typename T, typename ...Options>
- #endif
- struct make_ringbuffer
- {
- #ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
- typedef typename ringbuffer_signature::bind<A0, A1>::type bound_args;
- #else
- typedef typename ringbuffer_signature::bind<Options...>::type bound_args;
- #endif
- typedef extract_capacity<bound_args> extract_capacity_t;
- static const bool runtime_sized = !extract_capacity_t::has_capacity;
- static const size_t capacity = extract_capacity_t::capacity;
- typedef extract_allocator<bound_args, T> extract_allocator_t;
- typedef typename extract_allocator_t::type allocator;
- // allocator argument is only sane, for run-time sized ringbuffers
- BOOST_STATIC_ASSERT((mpl::if_<mpl::bool_<!runtime_sized>,
- mpl::bool_<!extract_allocator_t::has_allocator>,
- mpl::true_
- >::type::value));
- typedef typename mpl::if_c<runtime_sized,
- runtime_sized_ringbuffer<T, allocator>,
- compile_time_sized_ringbuffer<T, capacity>
- >::type ringbuffer_type;
- };
- } /* namespace detail */
- /** The spsc_queue class provides a single-writer/single-reader fifo queue, pushing and popping is wait-free.
- *
- * \b Policies:
- * - \c boost::lockfree::capacity<>, optional <br>
- * If this template argument is passed to the options, the size of the ringbuffer is set at compile-time.
- *
- * - \c boost::lockfree::allocator<>, defaults to \c boost::lockfree::allocator<std::allocator<T>> <br>
- * Specifies the allocator that is used to allocate the ringbuffer. This option is only valid, if the ringbuffer is configured
- * to be sized at run-time
- *
- * \b Requirements:
- * - T must have a default constructor
- * - T must be copyable
- * */
- #ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
- template <typename T, class A0, class A1>
- #else
- template <typename T, typename ...Options>
- #endif
- class spsc_queue:
- #ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
- public detail::make_ringbuffer<T, A0, A1>::ringbuffer_type
- #else
- public detail::make_ringbuffer<T, Options...>::ringbuffer_type
- #endif
- {
- private:
- #ifndef BOOST_DOXYGEN_INVOKED
- #ifdef BOOST_NO_CXX11_VARIADIC_TEMPLATES
- typedef typename detail::make_ringbuffer<T, A0, A1>::ringbuffer_type base_type;
- static const bool runtime_sized = detail::make_ringbuffer<T, A0, A1>::runtime_sized;
- typedef typename detail::make_ringbuffer<T, A0, A1>::allocator allocator_arg;
- #else
- typedef typename detail::make_ringbuffer<T, Options...>::ringbuffer_type base_type;
- static const bool runtime_sized = detail::make_ringbuffer<T, Options...>::runtime_sized;
- typedef typename detail::make_ringbuffer<T, Options...>::allocator allocator_arg;
- #endif
- struct implementation_defined
- {
- typedef allocator_arg allocator;
- typedef std::size_t size_type;
- };
- #endif
- public:
- typedef T value_type;
- typedef typename implementation_defined::allocator allocator;
- typedef typename implementation_defined::size_type size_type;
- /** Constructs a spsc_queue
- *
- * \pre spsc_queue must be configured to be sized at compile-time
- */
- spsc_queue(void)
- {
- // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
- // this function and this function may be compiled even when it isn't being used.
- BOOST_ASSERT(!runtime_sized);
- }
- /** Constructs a spsc_queue with a custom allocator
- *
- * \pre spsc_queue must be configured to be sized at compile-time
- *
- * \note This is just for API compatibility: an allocator isn't actually needed
- */
- template <typename U>
- explicit spsc_queue(typename boost::allocator_rebind<allocator, U>::type const &)
- {
- BOOST_STATIC_ASSERT(!runtime_sized);
- }
- /** Constructs a spsc_queue with a custom allocator
- *
- * \pre spsc_queue must be configured to be sized at compile-time
- *
- * \note This is just for API compatibility: an allocator isn't actually needed
- */
- explicit spsc_queue(allocator const &)
- {
- // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
- // this function and this function may be compiled even when it isn't being used.
- BOOST_ASSERT(!runtime_sized);
- }
- /** Constructs a spsc_queue for element_count elements
- *
- * \pre spsc_queue must be configured to be sized at run-time
- */
- explicit spsc_queue(size_type element_count):
- base_type(element_count)
- {
- // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
- // this function and this function may be compiled even when it isn't being used.
- BOOST_ASSERT(runtime_sized);
- }
- /** Constructs a spsc_queue for element_count elements with a custom allocator
- *
- * \pre spsc_queue must be configured to be sized at run-time
- */
- template <typename U>
- spsc_queue(size_type element_count, typename boost::allocator_rebind<allocator, U>::type const & alloc):
- base_type(alloc, element_count)
- {
- BOOST_STATIC_ASSERT(runtime_sized);
- }
- /** Constructs a spsc_queue for element_count elements with a custom allocator
- *
- * \pre spsc_queue must be configured to be sized at run-time
- */
- spsc_queue(size_type element_count, allocator_arg const & alloc):
- base_type(alloc, element_count)
- {
- // Don't use BOOST_STATIC_ASSERT() here since it will be evaluated when compiling
- // this function and this function may be compiled even when it isn't being used.
- BOOST_ASSERT(runtime_sized);
- }
- /** Pushes object t to the ringbuffer.
- *
- * \pre only one thread is allowed to push data to the spsc_queue
- * \post object will be pushed to the spsc_queue, unless it is full.
- * \return true, if the push operation is successful.
- *
- * \note Thread-safe and wait-free
- * */
- bool push(T const & t)
- {
- return base_type::push(t);
- }
- /** Pops one object from ringbuffer.
- *
- * \pre only one thread is allowed to pop data to the spsc_queue
- * \post if ringbuffer is not empty, object will be discarded.
- * \return true, if the pop operation is successful, false if ringbuffer was empty.
- *
- * \note Thread-safe and wait-free
- */
- bool pop ()
- {
- detail::consume_noop consume_functor;
- return consume_one( consume_functor );
- }
- /** Pops one object from ringbuffer.
- *
- * \pre only one thread is allowed to pop data to the spsc_queue
- * \post if ringbuffer is not empty, object will be copied to ret.
- * \return true, if the pop operation is successful, false if ringbuffer was empty.
- *
- * \note Thread-safe and wait-free
- */
- template <typename U>
- typename boost::enable_if<typename is_convertible<T, U>::type, bool>::type
- pop (U & ret)
- {
- detail::consume_via_copy<U> consume_functor(ret);
- return consume_one( consume_functor );
- }
- /** Pushes as many objects from the array t as there is space.
- *
- * \pre only one thread is allowed to push data to the spsc_queue
- * \return number of pushed items
- *
- * \note Thread-safe and wait-free
- */
- size_type push(T const * t, size_type size)
- {
- return base_type::push(t, size);
- }
- /** Pushes as many objects from the array t as there is space available.
- *
- * \pre only one thread is allowed to push data to the spsc_queue
- * \return number of pushed items
- *
- * \note Thread-safe and wait-free
- */
- template <size_type size>
- size_type push(T const (&t)[size])
- {
- return push(t, size);
- }
- /** Pushes as many objects from the range [begin, end) as there is space .
- *
- * \pre only one thread is allowed to push data to the spsc_queue
- * \return iterator to the first element, which has not been pushed
- *
- * \note Thread-safe and wait-free
- */
- template <typename ConstIterator>
- ConstIterator push(ConstIterator begin, ConstIterator end)
- {
- return base_type::push(begin, end);
- }
- /** Pops a maximum of size objects from ringbuffer.
- *
- * \pre only one thread is allowed to pop data to the spsc_queue
- * \return number of popped items
- *
- * \note Thread-safe and wait-free
- * */
- size_type pop(T * ret, size_type size)
- {
- return base_type::pop(ret, size);
- }
- /** Pops a maximum of size objects from spsc_queue.
- *
- * \pre only one thread is allowed to pop data to the spsc_queue
- * \return number of popped items
- *
- * \note Thread-safe and wait-free
- * */
- template <size_type size>
- size_type pop(T (&ret)[size])
- {
- return pop(ret, size);
- }
- /** Pops objects to the output iterator it
- *
- * \pre only one thread is allowed to pop data to the spsc_queue
- * \return number of popped items
- *
- * \note Thread-safe and wait-free
- * */
- template <typename OutputIterator>
- typename boost::disable_if<typename is_convertible<T, OutputIterator>::type, size_type>::type
- pop(OutputIterator it)
- {
- return base_type::pop_to_output_iterator(it);
- }
- /** 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)
- {
- return base_type::consume_one(f);
- }
- /// \copydoc boost::lockfree::spsc_queue::consume_one(Functor & rhs)
- template <typename Functor>
- bool consume_one(Functor const & f)
- {
- return base_type::consume_one(f);
- }
- /** 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_type consume_all(Functor & f)
- {
- return base_type::consume_all(f);
- }
- /// \copydoc boost::lockfree::spsc_queue::consume_all(Functor & rhs)
- template <typename Functor>
- size_type consume_all(Functor const & f)
- {
- return base_type::consume_all(f);
- }
- /** get number of elements that are available for read
- *
- * \return number of available elements that can be popped from the spsc_queue
- *
- * \note Thread-safe and wait-free, should only be called from the consumer thread
- * */
- size_type read_available() const
- {
- return base_type::read_available(base_type::max_number_of_elements());
- }
- /** get write space to write elements
- *
- * \return number of elements that can be pushed to the spsc_queue
- *
- * \note Thread-safe and wait-free, should only be called from the producer thread
- * */
- size_type write_available() const
- {
- return base_type::write_available(base_type::max_number_of_elements());
- }
- /** get reference to element in the front of the queue
- *
- * Availability of front element can be checked using read_available().
- *
- * \pre only a consuming thread is allowed to check front element
- * \pre read_available() > 0. If ringbuffer is empty, it's undefined behaviour to invoke this method.
- * \return reference to the first element in the queue
- *
- * \note Thread-safe and wait-free
- */
- const T& front() const
- {
- BOOST_ASSERT(read_available() > 0);
- return base_type::front();
- }
- /// \copydoc boost::lockfree::spsc_queue::front() const
- T& front()
- {
- BOOST_ASSERT(read_available() > 0);
- return base_type::front();
- }
- /** reset the ringbuffer
- *
- * \note Not thread-safe
- * */
- void reset(void)
- {
- if ( !boost::has_trivial_destructor<T>::value ) {
- // make sure to call all destructors!
- detail::consume_noop consume_functor;
- (void)consume_all(consume_functor);
- } else {
- base_type::write_index_.store(0, memory_order_relaxed);
- base_type::read_index_.store(0, memory_order_release);
- }
- }
- };
- } /* namespace lockfree */
- } /* namespace boost */
- #endif /* BOOST_LOCKFREE_SPSC_QUEUE_HPP_INCLUDED */
|