Boost C++ Libraries

boost/lockfree/spsc_queue.hpp

//  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/config.hpp> // for BOOST_LIKELY
#include <boost/core/allocator_access.hpp>
#include <boost/next_prior.hpp>
#include <boost/static_assert.hpp>
#include <boost/utility.hpp>
#include <boost/utility/enable_if.hpp>

#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 from 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 from 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 from 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 from 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 from 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 boost::lockfree


#endif /* BOOST_LOCKFREE_SPSC_QUEUE_HPP_INCLUDED */