concurrent_queue.h

Go to the documentation of this file.

/***
* ==++==
*
* Copyright (c) Microsoft Corporation.  All rights reserved.
* Microsoft would like to acknowledge that this concurrency data structure implementation
* is based on Intel's implementation in its Threading Building Blocks ("Intel Material").
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* concurrent_queue.h
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/

/*
    Intel Material Copyright 2005-2008 Intel Corporation.  All Rights Reserved.
*/

#pragma once


#include <crtdefs.h>
#include <memory>
#include <cstddef>
#include <crtdbg.h>
#include <concrt.h>
#include <utility>

#define _PPL_CONTAINER

#if !(defined (_M_X64) || defined (_M_IX86) || defined (_M_ARM))
    #error ERROR: Concurrency Runtime is supported only on X64, X86, and ARM architectures.
#endif  /* !(defined (_M_X64) || defined (_M_IX86) || defined (_M_ARM)) */

#if defined (_M_CEE)
    #error ERROR: Concurrency Runtime is not supported when compiling /clr.
#endif  /* defined (_M_CEE) */

#pragma pack(push,_CRT_PACKING)
#pragma warning(push)
#pragma warning (disable: 4510 4512 4610)  // disable warnings for compiler unable to generate constructor



namespace Concurrency
{

template<typename _Ty, class _Ax = std::allocator<_Ty> >
class concurrent_queue;

namespace details
{

    class _Concurrent_queue_rep;

    typedef size_t _Ticket;

    class _Concurrent_queue_iterator_rep;
    class _Concurrent_queue_iterator_base_v4;
    template<typename _Container, typename _Value> class _Concurrent_queue_iterator;

    //  Type-independent portion of concurrent_queue.
    class _Concurrent_queue_base_v4
    {
        // Internal representation
        _Concurrent_queue_rep* _My_rep;

        friend class _Concurrent_queue_rep;
        friend struct _Micro_queue;
        friend class _Micro_queue_pop_finalizer;
        friend class _Concurrent_queue_iterator_rep;
        friend class _Concurrent_queue_iterator_base_v4;
    protected:
        // Prefix on a page
        struct _Page
        {
            _Page* _Next;
            size_t _Mask;
        };

        // Always a power of 2
        size_t _Items_per_page;

        // Size of an item
        size_t _Item_size;

    private:
        virtual void _Move_item( _Page& _Dst, size_t _Index, void* _Src ) = 0;
        virtual void _Copy_item( _Page& _Dst, size_t _Index, const void* _Src ) = 0;
        virtual void _Assign_and_destroy_item( void* _Dst, _Page& _Src, size_t _Index ) = 0;
    protected:
        _CRTIMP2 _Concurrent_queue_base_v4( size_t _Item_size );
        _CRTIMP2 virtual ~_Concurrent_queue_base_v4();

        // Enqueue item at tail of queue
        _CRTIMP2 void _Internal_push( const void* _Src );

        // Enqueue item at tail of queue by move
        _CRTIMP2 void _Internal_move_push( void* _Src );

        // swap the internal representation
        _CRTIMP2 void _Concurrent_queue_base_v4::_Internal_swap( _Concurrent_queue_base_v4& other );

        // Attempt to dequeue item from queue.
        _CRTIMP2 bool _Internal_pop_if_present( void* _Dst );

        // Get size of queue
        _CRTIMP2 size_t _Internal_size() const;

        // Test instantaneous queue empty
        _CRTIMP2 bool _Internal_empty() const;

        // custom allocator
        virtual _Page *_Allocate_page() = 0;

        // custom de-allocator
        virtual void _Deallocate_page( _Page *p ) = 0;

        // free any remaining pages
        _CRTIMP2 void _Internal_finish_clear() ;

        // throw an exception
        _CRTIMP2 void _Internal_throw_exception() const;

    private:
        // Deny copy construction
        _Concurrent_queue_base_v4( const _Concurrent_queue_base_v4& );

        // Deny assignment
        void operator=( const _Concurrent_queue_base_v4& );
    };

    typedef _Concurrent_queue_base_v4 _Concurrent_queue_base ;


    // A queue using simple locking.
    struct _Micro_queue
    {
        class _Pop_finalizer;
        class _Push_finalizer;

        _Subatomic<_Concurrent_queue_base::_Page*> _Head_page;
        _Subatomic<_Ticket> _Head_counter;

        _Subatomic<_Concurrent_queue_base::_Page*> _Tail_page;
        _Subatomic<_Ticket> _Tail_counter;

        volatile long _Page_mutex_flag;

        void _Push( void* _Item, _Ticket _K, _Concurrent_queue_base& _Base, void (_Concurrent_queue_base::*moveOp)(_Concurrent_queue_base_v4::_Page&, size_t, void*));

        bool _Pop( void* _Dest, _Ticket _K, _Concurrent_queue_base& _Base );
    };

    // Disable warning C4324: structure was padded due to __declspec(align())
    // This padding is expected and necessary.
    #pragma warning(push)
    #pragma warning(disable: 4324)


    // Internal representation of a ConcurrentQueue.
    class _Concurrent_queue_rep
    {
    private:
        friend struct _Micro_queue;

        // Approximately n_queue/golden ratio
        static const size_t _Phi = 3;

    public:
        // Must be power of 2
        static const size_t _N_queue = 8;

        // Map ticket to an array index
        static size_t _Index( _Ticket _K )
        {
            return _K*_Phi%_N_queue;
        }

        __declspec(align(64))
        _Subatomic<_Ticket> _Head_counter;

        __declspec(align(64))
        _Subatomic<_Ticket> _Tail_counter;

        __declspec(align(64))
        _Micro_queue _Array[_N_queue];

        _Micro_queue& _Choose( _Ticket _K )
        {
            // The formula here approximates LRU in a cache-oblivious way.
            return _Array[_Index(_K)];
        }
    };

    #pragma warning(pop)


    // Type-independent portion of _Concurrent_queue_iterator.
    class _Concurrent_queue_iterator_base_v4 {
        // Concurrentconcurrent_queue over which we are iterating.
        _Concurrent_queue_iterator_rep* _My_rep;

        template<typename _C, typename _Ty, typename _U>
        friend bool operator==( const _Concurrent_queue_iterator<_C,_Ty>&, const _Concurrent_queue_iterator<_C,_U>& );

        template<typename _C, typename _Ty, typename _U>
        friend bool operator!=( const _Concurrent_queue_iterator<_C,_Ty>&, const _Concurrent_queue_iterator<_C,_U>& );
    protected:
        // Pointer to current item
        mutable void* _My_item;

        // Default constructor
        _Concurrent_queue_iterator_base_v4()
            : _My_rep(NULL), _My_item(NULL)
        {
        }

        // Copy constructor
        _Concurrent_queue_iterator_base_v4( const _Concurrent_queue_iterator_base_v4& _I )
            : _My_rep(NULL), _My_item(NULL)
        {
            _Assign(_I);
        }

        // Construct iterator pointing to head of queue.
        _CRTIMP2 _Concurrent_queue_iterator_base_v4( const _Concurrent_queue_base&  );

        // Assignment
        _CRTIMP2 void _Assign( const _Concurrent_queue_iterator_base_v4& );

        // Advance iterator one step towards tail of queue.
        _CRTIMP2 void _Advance();

        // Destructor
        _CRTIMP2 ~_Concurrent_queue_iterator_base_v4();
    };

    typedef _Concurrent_queue_iterator_base_v4 concurrent_queue_iterator_base;

    // Meets requirements of a forward iterator for STL.
    template<typename _Container, typename _Value>
    class _Concurrent_queue_iterator: public _Concurrent_queue_iterator_base_v4, public std::iterator<std::forward_iterator_tag, _Value>
    {
        template<typename _Ty, class _Ax> friend class ::Concurrency::concurrent_queue;

        // Construct iterator pointing to head of queue.
        _Concurrent_queue_iterator( const _Concurrent_queue_base& _Queue )
            : _Concurrent_queue_iterator_base_v4(_Queue)
        {
        }
    public:
        _Concurrent_queue_iterator()
        {
        }

        _Concurrent_queue_iterator( const _Concurrent_queue_iterator<_Container,typename _Container::value_type>& _Other )
            : _Concurrent_queue_iterator_base_v4(_Other)
        {
        }

        // Iterator assignment
        _Concurrent_queue_iterator& operator=( const _Concurrent_queue_iterator& _Other )
        {
            _Assign(_Other);
            return *this;
        }

        // Reference to current item
        _Value& operator*() const
        {
            return *static_cast<_Value*>(_My_item);
        }

        _Value* operator->() const
        {
            return &operator*();
        }

        // Advance to next item in queue
        _Concurrent_queue_iterator& operator++()
        {
            _Advance();
            return *this;
        }

        // Post increment
        _Concurrent_queue_iterator operator++(int)
        {
            _Concurrent_queue_iterator _Result = *this;
            _Advance();
            return _Result;
        }
    }; // _Concurrent_queue_iterator

    template<typename _Container, typename _Value>
    struct std::_Is_checked_helper<_Concurrent_queue_iterator<_Container, _Value> >
        : public true_type
    {   // mark _Concurrent_queue_iterator as checked. This suppresses warning C4996
    };

    template<typename _C, typename _Ty, typename _U>
    bool operator==( const _Concurrent_queue_iterator<_C,_Ty>& _I, const _Concurrent_queue_iterator<_C,_U>& _J )
    {
        return _I._My_item==_J._My_item;
    }

    template<typename _C, typename _Ty, typename _U>
    bool operator!=( const _Concurrent_queue_iterator<_C,_Ty>& _I, const _Concurrent_queue_iterator<_C,_U>& _J )
    {
        return _I._My_item!=_J._My_item;
    }

} // namespace details;



template<typename _Ty, class _Ax>
class concurrent_queue: public ::Concurrency::details::_Concurrent_queue_base_v4
{
    template<typename _Container, typename _Value> friend class ::Concurrency::details::_Concurrent_queue_iterator;

    // allocator type
    typedef typename _Ax::template rebind<char>::other _Page_allocator_type;
    _Page_allocator_type _My_allocator;

    // Class used to ensure exception-safety of method "pop"
    class _Destroyer
    {
    private:
        _Ty& _My_value;

        void operator=(const _Destroyer&); // prevent warning: assign operator can't be generated
    public:
        _Destroyer( _Ty& _Value )
            : _My_value(_Value)
        {
        }

        ~_Destroyer()
        {
            _My_value.~_Ty();
        }
    };

    _Ty& _Get_ref( _Page& _Pg, size_t _Index )
    {
        _CONCRT_ASSERT( _Index<_Items_per_page );
        return static_cast<_Ty*>(static_cast<void*>(&_Pg+1))[_Index];
    }

    /*override*/ virtual void _Copy_item( _Page& _Dst, size_t _Index, const void* _Src )
    {
        new( &_Get_ref(_Dst,_Index) ) _Ty(*static_cast<const _Ty*>(_Src));
    }

    /*override*/ virtual void _Move_item( _Page& _Dst, size_t _Index, void* _Src )
    {
        new( &_Get_ref(_Dst,_Index) ) _Ty(std::move(*static_cast<_Ty*>(_Src)));
    }

    /*override*/ virtual void _Assign_and_destroy_item( void* _Dst, _Page& _Src, size_t _Index )
    {
        _Ty& _From = _Get_ref(_Src,_Index);
        _Destroyer _D(_From);
        if (_Dst != NULL)
        {
            *static_cast<_Ty*>(_Dst) = std::move(_From);
        }
    }

    /*overide*/ virtual _Page *_Allocate_page()
    {
        size_t _N = sizeof(_Page) + _Items_per_page*_Item_size;
        _Page *_Pg = reinterpret_cast<_Page*>(_My_allocator.allocate( _N ));
        if( !_Pg )
            _Internal_throw_exception();
        return _Pg;
    }

    /*override*/ virtual void _Deallocate_page( _Page *_Pg )
    {
        size_t _N = sizeof(_Page) + _Items_per_page*_Item_size;
        _My_allocator.deallocate( reinterpret_cast<char*>(_Pg), _N );
    }

public:
    
    typedef _Ty value_type;

    
    typedef _Ax allocator_type;

    
    typedef _Ty& reference;

    
    typedef const _Ty& const_reference;

    
    typedef std::size_t size_type;

    
    typedef std::ptrdiff_t difference_type;

    
    explicit concurrent_queue(const allocator_type  &_Al = allocator_type())
        : _Concurrent_queue_base_v4( sizeof(_Ty) ), _My_allocator( _Al )
    {
    }

    
    concurrent_queue(const concurrent_queue& _OtherQ, const allocator_type &_Al = allocator_type());

    
    concurrent_queue(concurrent_queue&& _OtherQ, const allocator_type &_Al = allocator_type());

    
    template<typename _InputIterator>
    concurrent_queue(_InputIterator _Begin, _InputIterator _End)
        : _Concurrent_queue_base_v4( sizeof(_Ty) ), _My_allocator( allocator_type() )
    {
        while (_Begin != _End)
        {
            this->push(*_Begin);
            ++_Begin;
        }
    }

    
    ~concurrent_queue();

    
    void push( const _Ty& _Src )
    {
        _Internal_push( &_Src );
    }

    
    void push( _Ty&& _Src )
    {
        _Internal_move_push( &_Src );
    }

    
    bool try_pop( _Ty& _Dest )
    {
        return _Internal_pop_if_present( &_Dest );
    }

    
    size_type unsafe_size() const
    {
        return _Internal_size();
    }

    
    bool empty() const
    {
        return _Internal_empty();
    }

    
    allocator_type get_allocator() const
    {
        return this->_My_allocator;
    }

    
    void clear();

    
    typedef details::_Concurrent_queue_iterator<concurrent_queue,_Ty> iterator;

    
    typedef details::_Concurrent_queue_iterator<concurrent_queue,const _Ty> const_iterator;

    
    iterator unsafe_begin()
    {
        return iterator(*this);
    }

    
    iterator unsafe_end()
    {
        return iterator();
    }

    
    const_iterator unsafe_begin() const
    {
        return const_iterator(*this);
    }

    
    const_iterator unsafe_end() const
    {
        return const_iterator();
    }
};



template<typename _Ty, class _Ax>
concurrent_queue<_Ty,_Ax>::concurrent_queue(const concurrent_queue& _Queue, const allocator_type& _Al = allocator_type())
    : _Concurrent_queue_base_v4( sizeof(_Ty) ), _My_allocator(_Al)
{
    concurrent_queue::const_iterator _QEnd = _Queue.unsafe_end();
    for (concurrent_queue::const_iterator _It = _Queue.unsafe_begin(); _It != _QEnd; ++_It)
        this->push(*_It);
}


template<typename _Ty, class _Ax>
concurrent_queue<_Ty,_Ax>::concurrent_queue(concurrent_queue&& _Queue, const allocator_type& _Al = allocator_type())
    : _Concurrent_queue_base_v4( sizeof(_Ty) ), _My_allocator(_Al)
{
    _Internal_swap(_Queue);
}


template<typename _Ty, class _Ax>
concurrent_queue<_Ty,_Ax>::~concurrent_queue()
{
    clear();
    _Internal_finish_clear();
}


template<typename _Ty, class _Ax>
void concurrent_queue<_Ty,_Ax>::clear()
{
    while( !empty() )
    {
        if (!_Internal_pop_if_present(NULL))
        {
            _CONCRT_ASSERT(empty());
            break;
        }
    }
}

} // namespace Concurrency

namespace concurrency = Concurrency;

#pragma warning(pop)
#pragma pack(pop)