heter_queue.h

//   Copyright Giuseppe Campana (giu.campana@gmail.com) 2016-2018.
// 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)

#pragma once
#include <density/default_allocator.h>
#include <density/density_common.h>
#include <density/dynamic_reference.h>
#include <density/runtime_type.h>
#include <iterator>

namespace density
{
    namespace detail
    {
        struct QueueControl
        {
            uintptr_t
              m_next; 
        };

        enum Queue_Flags : uintptr_t
        {
            Queue_Busy = 1, 
            Queue_Dead =
              2, 
            Queue_External =
              4, 
            Queue_AllFlags = Queue_Busy | Queue_Dead | Queue_External
        };
    } // namespace detail

    template <typename RUNTIME_TYPE = runtime_type<>, typename ALLOCATOR_TYPE = default_allocator>
    class heter_queue : private ALLOCATOR_TYPE
    {
        using ControlBlock = detail::QueueControl;

        ControlBlock * m_head;

        ControlBlock * m_tail;

        enum class PrivateType
        {
        };

      public:
        constexpr static size_t min_alignment = detail::size_max(
          detail::Queue_AllFlags + 1, alignof(ControlBlock), alignof(RUNTIME_TYPE));

        using runtime_type    = RUNTIME_TYPE;
        using value_type      = dynamic_reference<RUNTIME_TYPE>;
        using allocator_type  = ALLOCATOR_TYPE;
        using pointer         = value_type *;
        using const_pointer   = const value_type *;
        using reference       = value_type;
        using const_reference = const value_type &;
        using size_type       = std::size_t;
        using difference_type = std::ptrdiff_t;

        class const_iterator;

      private:
        constexpr static auto s_invalid_control_block = ALLOCATOR_TYPE::page_size - 1;

        constexpr static size_t s_sizeof_ControlBlock =
          uint_upper_align(sizeof(ControlBlock), min_alignment);

        constexpr static size_t s_sizeof_RuntimeType =
          uint_upper_align(sizeof(RUNTIME_TYPE), min_alignment);

        constexpr static auto s_max_size_inpage = ALLOCATOR_TYPE::page_size -
                                                  s_sizeof_ControlBlock - s_sizeof_RuntimeType -
                                                  s_sizeof_ControlBlock;

        struct Allocation
        {
            ControlBlock * m_control_block;
            void *         m_user_storage;
        };

      public:
        static constexpr bool concurrent_puts = false;

        static constexpr bool concurrent_consumes = false;

        static constexpr bool concurrent_put_consumes = false;

        static constexpr bool is_seq_cst = true;

        static_assert(
          is_power_of_2(ALLOCATOR_TYPE::page_alignment) &&
            ALLOCATOR_TYPE::page_alignment >= ALLOCATOR_TYPE::page_size &&
            (ALLOCATOR_TYPE::page_alignment % min_alignment) == 0,
          "The alignment of the pages must be a power of 2, greater or equal to the size of the "
          "pages, and a multiple of min_alignment");

        static_assert(
          ALLOCATOR_TYPE::page_size > (min_alignment + alignof(ControlBlock)) * 4,
          "Invalid page size");

        constexpr heter_queue() noexcept
            : m_head(reinterpret_cast<ControlBlock *>(s_invalid_control_block)),
              m_tail(reinterpret_cast<ControlBlock *>(s_invalid_control_block))
        {
            static_assert(std::is_nothrow_default_constructible<ALLOCATOR_TYPE>::value, "");
        }

        constexpr explicit heter_queue(const ALLOCATOR_TYPE & i_source_allocator) noexcept
            : ALLOCATOR_TYPE(i_source_allocator),
              m_head(reinterpret_cast<ControlBlock *>(s_invalid_control_block)),
              m_tail(reinterpret_cast<ControlBlock *>(s_invalid_control_block))
        {
        }

        constexpr explicit heter_queue(ALLOCATOR_TYPE && i_source_allocator) noexcept
            : ALLOCATOR_TYPE(std::move(i_source_allocator)),
              m_head(reinterpret_cast<ControlBlock *>(s_invalid_control_block)),
              m_tail(reinterpret_cast<ControlBlock *>(s_invalid_control_block))
        {
            static_assert(std::is_nothrow_move_constructible<ALLOCATOR_TYPE>::value, "");
        }

        heter_queue(heter_queue && i_source) noexcept
            : ALLOCATOR_TYPE(std::move(static_cast<ALLOCATOR_TYPE &&>(i_source))),
              m_head(i_source.m_head), m_tail(i_source.m_tail)
        {
            static_assert(std::is_nothrow_move_constructible<ALLOCATOR_TYPE>::value, "");

            i_source.m_tail = i_source.m_head =
              reinterpret_cast<ControlBlock *>(s_invalid_control_block);
        }

        heter_queue(const heter_queue & i_source)
            : allocator_type(static_cast<const allocator_type &>(i_source)),
              m_head(reinterpret_cast<ControlBlock *>(s_invalid_control_block)),
              m_tail(reinterpret_cast<ControlBlock *>(s_invalid_control_block))
        {
            for (auto source_it = i_source.cbegin(); source_it != i_source.cend(); source_it++)
            {
                dyn_push_copy(source_it.complete_type(), source_it.element_ptr());
            }
        }

        heter_queue & operator=(heter_queue && i_source) noexcept
        {
            destroy();
            m_tail = m_head = reinterpret_cast<ControlBlock *>(s_invalid_control_block);
            swap(*this, i_source);
            return *this;
        }

        heter_queue & operator=(const heter_queue & i_source)
        {
            auto copy(i_source);
            swap(*this, copy);
            return *this;
        }

        allocator_type
          get_allocator() noexcept(std::is_nothrow_copy_constructible<allocator_type>::value)
        {
            return *this;
        }

        allocator_type & get_allocator_ref() noexcept { return *this; }

        const allocator_type & get_allocator_ref() const noexcept { return *this; }

        friend void swap(
          heter_queue<RUNTIME_TYPE, ALLOCATOR_TYPE> & i_first,
          heter_queue<RUNTIME_TYPE, ALLOCATOR_TYPE> & i_second) noexcept
        {
            using std::swap;
            swap(static_cast<ALLOCATOR_TYPE &>(i_first), static_cast<ALLOCATOR_TYPE &>(i_second));
            swap(i_first.m_head, i_second.m_head);
            swap(i_first.m_tail, i_second.m_tail);
        }

        ~heter_queue() { destroy(); }

        bool empty() const noexcept
        {
            // the queue may contain busy or dead elements, that must be ignored
            for (auto curr = m_head; curr != m_tail;)
            {
                auto const control_bits = curr->m_next & (detail::Queue_Busy | detail::Queue_Dead);
                if (control_bits == 0) // if not busy and not dead
                {
                    return false;
                }
                curr = reinterpret_cast<ControlBlock *>(curr->m_next & ~detail::Queue_AllFlags);
            }
            return true;
        }

        void clear() noexcept
        {
            consume_operation consume;
            while (try_start_consume(consume))
            {
                consume.commit();
            }

            DENSITY_ASSERT_INTERNAL(empty());

            clean_dead_elements();
        }

        template <typename ELEMENT_COMPLETE_TYPE = void> class put_transaction
        {
            static_assert(
              std::is_same<
                ELEMENT_COMPLETE_TYPE,
                typename std::decay<ELEMENT_COMPLETE_TYPE>::type>::value,
              "");

          public:
            put_transaction() noexcept : m_queue(nullptr) {}

            put_transaction(const put_transaction &) = delete;

            put_transaction & operator=(const put_transaction &) = delete;

            template <
              typename OTHERTYPE,
              typename = typename std::enable_if<
                std::is_same<OTHERTYPE, ELEMENT_COMPLETE_TYPE>::value ||
                std::is_void<ELEMENT_COMPLETE_TYPE>::value>::type>
            put_transaction(put_transaction<OTHERTYPE> && i_source) noexcept
                : m_queue(i_source.m_queue), m_put_data(i_source.m_put_data)
            {
                i_source.m_queue = nullptr;
            }

            template <
              typename OTHERTYPE,
              typename = typename std::enable_if<
                std::is_same<OTHERTYPE, ELEMENT_COMPLETE_TYPE>::value ||
                std::is_void<ELEMENT_COMPLETE_TYPE>::value>::type>
            put_transaction & operator=(put_transaction<OTHERTYPE> && i_source) noexcept
            {
                if (
                  this !=
                  static_cast<void *>(
                    &i_source)) // cast to void to allow comparing pointers to unrelated types
                {
                    if (!empty())
                        cancel();

                    std::swap(m_queue, i_source.m_queue);
                    std::swap(m_put_data, i_source.m_put_data);
                }
                return *this;
            }

            friend void swap(put_transaction & i_first, put_transaction & i_second) noexcept
            {
                using std::swap;
                swap(i_first.m_queue, i_second.m_queue);
                swap(i_first.m_put_data, i_second.m_put_data);
            }

            void * raw_allocate(size_t i_size, size_t i_alignment)
            {
                DENSITY_ASSERT(!empty());
                auto push_data =
                  m_queue->inplace_allocate<detail::Queue_Dead, false>(i_size, i_alignment);
                return push_data.m_user_storage;
            }

            template <typename INPUT_ITERATOR>
            typename std::iterator_traits<INPUT_ITERATOR>::value_type *
              raw_allocate_copy(INPUT_ITERATOR i_begin, INPUT_ITERATOR i_end)
            {
                using ValueType = typename std::iterator_traits<INPUT_ITERATOR>::value_type;
                static_assert(
                  std::is_trivially_destructible<ValueType>::value,
                  "raw_allocate_copy provides a raw memory in-place allocation that does not "
                  "invoke "
                  "destructors when deallocating");

                auto const count_s = std::distance(i_begin, i_end);
                auto const count   = static_cast<size_t>(count_s);
                DENSITY_ASSUME(static_cast<decltype(count_s)>(count) == count_s);

                auto const elements = static_cast<ValueType *>(
                  raw_allocate(sizeof(ValueType) * count, alignof(ValueType)));
                for (auto curr = elements; i_begin != i_end; ++i_begin, ++curr)
                    new (curr) ValueType(*i_begin);
                return elements;
            }

            template <typename INPUT_RANGE>
            auto raw_allocate_copy(const INPUT_RANGE & i_source_range)
              -> decltype(std::declval<put_transaction>().raw_allocate_copy(
                std::begin(i_source_range), std::end(i_source_range)))
            {
                return raw_allocate_copy(std::begin(i_source_range), std::end(i_source_range));
            }

            void commit() noexcept
            {
                DENSITY_ASSERT(!empty());
                m_queue = nullptr;
            }

            void cancel() noexcept
            {
                DENSITY_ASSERT(!empty());
                cancel_put_impl(m_put_data.m_control_block);
                m_queue = nullptr;
            }

            bool empty() const noexcept { return m_queue == nullptr; }

            explicit operator bool() const noexcept { return m_queue != nullptr; }

            heter_queue * queue() const noexcept { return m_queue; }

            void * element_ptr() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return m_put_data.m_user_storage;
            }

#ifndef DOXYGEN_DOC_GENERATION
            template <
              typename EL                                               = ELEMENT_COMPLETE_TYPE,
              typename std::enable_if<!std::is_void<EL>::value>::type * = nullptr>
            EL &
#else
            ELEMENT_COMPLETE_TYPE &
#endif
              element() const noexcept
            {
                return *static_cast<ELEMENT_COMPLETE_TYPE *>(element_ptr());
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return *type_after_control(m_put_data.m_control_block);
            }

            ~put_transaction()
            {
                if (m_queue != nullptr)
                {
                    cancel_put_impl(m_put_data.m_control_block);
                }
            }

            put_transaction(PrivateType, heter_queue * i_queue, Allocation i_push_data) noexcept
                : m_queue(i_queue), m_put_data(i_push_data)
            {
            }

          private:
            heter_queue * m_queue; 
            Allocation    m_put_data;

            template <typename OTHERTYPE> friend class put_transaction;
        };

        class consume_operation
        {
          public:
            consume_operation() noexcept : m_control(nullptr) {}

            consume_operation(const consume_operation &) = delete;

            consume_operation & operator=(const consume_operation &) = delete;

            consume_operation(consume_operation && i_source) noexcept
                : m_queue(i_source.m_queue), m_control(i_source.m_control)
            {
                i_source.m_control = nullptr;
            }

            consume_operation & operator=(consume_operation && i_source) noexcept
            {
                if (this != &i_source)
                {
                    if (!empty())
                    {
                        commit();
                    }
                    m_queue            = i_source.m_queue;
                    m_control          = i_source.m_control;
                    i_source.m_control = nullptr;
                }
                return *this;
            }

            ~consume_operation()
            {
                if (m_control != nullptr)
                {
                    m_queue->cancel_consume_impl(m_control);
                }
            }

            friend void swap(consume_operation & i_first, consume_operation & i_second) noexcept
            {
                using std::swap;
                swap(i_first.m_queue, i_second.m_queue);
                swap(i_first.m_control, i_second.m_control);
            }

            bool empty() const noexcept { return m_control == nullptr; }

            explicit operator bool() const noexcept { return m_control != nullptr; }

            heter_queue * queue() const noexcept
            {
                return m_control != nullptr ? m_queue : nullptr;
            }

            void commit() noexcept
            {
                DENSITY_ASSERT(!empty());

                auto const & type    = complete_type();
                auto const   element = element_ptr();
                type.destroy(element);

                type.RUNTIME_TYPE::~RUNTIME_TYPE();

                m_queue->commit_consume_impl(m_control);
                m_control = nullptr;
            }

            void commit_nodestroy() noexcept
            {
                DENSITY_ASSERT(!empty());

                bool destroy_type = !std::is_trivially_destructible<RUNTIME_TYPE>::value;
                if (destroy_type)
                {
                    auto const & type = complete_type();
                    type.RUNTIME_TYPE::~RUNTIME_TYPE();
                }

                m_queue->commit_consume_impl(m_control);
                m_control = nullptr;
            }

            void cancel() noexcept
            {
                DENSITY_ASSERT(!empty());

                m_queue->cancel_consume_impl(m_control);
                m_control = nullptr;
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return *type_after_control(m_control);
            }

            void * unaligned_element_ptr() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return get_unaligned_element(m_control);
            }

            void * element_ptr() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return get_element(m_control);
            }

            template <typename COMPLETE_ELEMENT_TYPE>
            COMPLETE_ELEMENT_TYPE & element() const noexcept
            {
                DENSITY_ASSERT(!empty() && complete_type().template is<COMPLETE_ELEMENT_TYPE>());
                return *static_cast<COMPLETE_ELEMENT_TYPE *>(get_element(m_control));
            }

            consume_operation(PrivateType, heter_queue * i_queue, ControlBlock * i_control) noexcept
                : m_queue(i_queue), m_control(i_control)
            {
            }

            bool start_consume_impl(PrivateType, heter_queue * i_queue)
            {
                if (m_control != nullptr)
                {
                    m_queue->cancel_consume_impl(m_control);
                }
                m_queue   = i_queue;
                m_control = i_queue->start_consume_impl();
                return m_control != nullptr;
            }

          private:
            heter_queue *  m_queue;
            ControlBlock * m_control;
        };

        template <typename ELEMENT_TYPE> void push(ELEMENT_TYPE && i_source)
        {
            return emplace<typename std::decay<ELEMENT_TYPE>::type>(
              std::forward<ELEMENT_TYPE>(i_source));
        }

        template <typename ELEMENT_TYPE, typename... CONSTRUCTION_PARAMS>
        void emplace(CONSTRUCTION_PARAMS &&... i_construction_params)
        {
            start_emplace<ELEMENT_TYPE>(std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...)
              .commit();
        }

        void dyn_push(const RUNTIME_TYPE & i_type) { start_dyn_push(i_type).commit(); }

        void dyn_push_copy(const RUNTIME_TYPE & i_type, const void * i_source)
        {
            start_dyn_push_copy(i_type, i_source).commit();
        }

        void dyn_push_move(const RUNTIME_TYPE & i_type, void * i_source)
        {
            start_dyn_push_move(i_type, i_source).commit();
        }

        template <typename ELEMENT_TYPE>
        put_transaction<typename std::decay<ELEMENT_TYPE>::type>
          start_push(ELEMENT_TYPE && i_source)
        {
            return start_emplace<typename std::decay<ELEMENT_TYPE>::type>(
              std::forward<ELEMENT_TYPE>(i_source));
        }

        template <typename ELEMENT_TYPE, typename... CONSTRUCTION_PARAMS>
        put_transaction<ELEMENT_TYPE> start_emplace(CONSTRUCTION_PARAMS &&... i_construction_params)
        {
            auto push_data = inplace_allocate<
              0,
              true,
              detail::size_of<ELEMENT_TYPE>::value,
              alignof(ELEMENT_TYPE)>();

            RUNTIME_TYPE * type = nullptr;
            try
            {
                auto const type_storage = type_after_control(push_data.m_control_block);
                DENSITY_ASSUME(type_storage != nullptr);
                type = new (type_storage) RUNTIME_TYPE(RUNTIME_TYPE::template make<ELEMENT_TYPE>());

                DENSITY_ASSUME(push_data.m_user_storage != nullptr);
                new (push_data.m_user_storage)
                  ELEMENT_TYPE(std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...);
            }
            catch (...)
            {
                if (type != nullptr)
                    type->RUNTIME_TYPE::~RUNTIME_TYPE();
                DENSITY_ASSERT_INTERNAL(
                  (push_data.m_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
                  0);
                push_data.m_control_block->m_next += detail::Queue_Dead;
                throw;
            }

            return put_transaction<ELEMENT_TYPE>(PrivateType(), this, push_data);
        }

        put_transaction<> start_dyn_push(const RUNTIME_TYPE & i_type)
        {
            auto push_data = inplace_allocate<0, true>(i_type.size(), i_type.alignment());

            RUNTIME_TYPE * type = nullptr;
            try
            {
                auto const type_storage = type_after_control(push_data.m_control_block);
                DENSITY_ASSUME(type_storage != nullptr);
                type = new (type_storage) RUNTIME_TYPE(i_type);

                DENSITY_ASSUME(push_data.m_user_storage != nullptr);
                i_type.default_construct(push_data.m_user_storage);
            }
            catch (...)
            {
                if (type != nullptr)
                    type->RUNTIME_TYPE::~RUNTIME_TYPE();
                DENSITY_ASSERT_INTERNAL(
                  (push_data.m_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
                  0);
                push_data.m_control_block->m_next += detail::Queue_Dead;
                throw;
            }

            return put_transaction<void>(PrivateType(), this, push_data);
        }


        put_transaction<> start_dyn_push_copy(const RUNTIME_TYPE & i_type, const void * i_source)
        {
            auto push_data = inplace_allocate<0, true>(i_type.size(), i_type.alignment());

            RUNTIME_TYPE * type = nullptr;
            try
            {
                auto const type_storage = type_after_control(push_data.m_control_block);
                DENSITY_ASSUME(type_storage != nullptr);
                type = new (type_storage) RUNTIME_TYPE(i_type);

                DENSITY_ASSUME(push_data.m_user_storage != nullptr);
                i_type.copy_construct(push_data.m_user_storage, i_source);
            }
            catch (...)
            {
                if (type != nullptr)
                    type->RUNTIME_TYPE::~RUNTIME_TYPE();
                DENSITY_ASSERT_INTERNAL(
                  (push_data.m_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
                  0);
                push_data.m_control_block->m_next += detail::Queue_Dead;
                throw;
            }

            return put_transaction<void>(PrivateType(), this, push_data);
        }

        put_transaction<> start_dyn_push_move(const RUNTIME_TYPE & i_type, void * i_source)
        {
            auto push_data = inplace_allocate<0, true>(i_type.size(), i_type.alignment());

            RUNTIME_TYPE * type = nullptr;
            try
            {
                auto const type_storage = type_after_control(push_data.m_control_block);
                DENSITY_ASSUME(type_storage != nullptr);
                type = new (type_storage) RUNTIME_TYPE(i_type);

                DENSITY_ASSUME(push_data.m_user_storage != nullptr);
                i_type.move_construct(push_data.m_user_storage, i_source);
            }
            catch (...)
            {
                if (type != nullptr)
                    type->RUNTIME_TYPE::~RUNTIME_TYPE();
                DENSITY_ASSERT_INTERNAL(
                  (push_data.m_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
                  0);
                push_data.m_control_block->m_next += detail::Queue_Dead;
                throw;
            }

            return put_transaction<void>(PrivateType(), this, push_data);
        }


        void pop() noexcept { try_start_consume().commit(); }

        bool try_pop() noexcept
        {
            if (auto operation = try_start_consume())
            {
                operation.commit();
                return true;
            }
            return false;
        }

        consume_operation try_start_consume() noexcept
        {
            return consume_operation(PrivateType(), this, start_consume_impl());
        }

        bool try_start_consume(consume_operation & i_consume) noexcept
        {
            return i_consume.start_consume_impl(PrivateType(), this);
        }


        template <typename ELEMENT_COMPLETE_TYPE = void> class reentrant_put_transaction
        {
            static_assert(
              std::is_same<
                ELEMENT_COMPLETE_TYPE,
                typename std::decay<ELEMENT_COMPLETE_TYPE>::type>::value,
              "");

          public:
            reentrant_put_transaction() noexcept : m_queue(nullptr) {}

            reentrant_put_transaction(const reentrant_put_transaction &) = delete;

            reentrant_put_transaction & operator=(const reentrant_put_transaction &) = delete;

            template <
              typename OTHERTYPE,
              typename = typename std::enable_if<
                std::is_same<OTHERTYPE, ELEMENT_COMPLETE_TYPE>::value ||
                std::is_void<ELEMENT_COMPLETE_TYPE>::value>::type>
            reentrant_put_transaction(reentrant_put_transaction<OTHERTYPE> && i_source) noexcept
                : m_queue(i_source.m_queue), m_put_data(i_source.m_put_data)
            {
                i_source.m_queue = nullptr;
            }

            template <
              typename OTHERTYPE,
              typename = typename std::enable_if<
                std::is_same<OTHERTYPE, ELEMENT_COMPLETE_TYPE>::value ||
                std::is_void<ELEMENT_COMPLETE_TYPE>::value>::type>
            reentrant_put_transaction &
              operator=(reentrant_put_transaction<OTHERTYPE> && i_source) noexcept
            {
                if (
                  this !=
                  static_cast<void *>(
                    &i_source)) // cast to void to allow comparing pointers to unrelated types
                {
                    if (!empty())
                        cancel();

                    std::swap(m_queue, i_source.m_queue);
                    std::swap(m_put_data, i_source.m_put_data);
                }
                return *this;
            }

            friend void swap(
              reentrant_put_transaction & i_first, reentrant_put_transaction & i_second) noexcept
            {
                std::swap(i_first.m_queue, i_second.m_queue);
                std::swap(i_first.m_put_data, i_second.m_put_data);
            }

            void * raw_allocate(size_t i_size, size_t i_alignment)
            {
                DENSITY_ASSERT(!empty());

                auto push_data =
                  m_queue->inplace_allocate<detail::Queue_Dead, false>(i_size, i_alignment);
                return push_data.m_user_storage;
            }

            template <typename INPUT_ITERATOR>
            typename std::iterator_traits<INPUT_ITERATOR>::value_type *
              raw_allocate_copy(INPUT_ITERATOR i_begin, INPUT_ITERATOR i_end)
            {
                using ValueType = typename std::iterator_traits<INPUT_ITERATOR>::value_type;
                static_assert(
                  std::is_trivially_destructible<ValueType>::value,
                  "raw_allocate_copy provides a raw memory in-place allocation that does not "
                  "invoke "
                  "destructors when deallocating");

                auto const count_s = std::distance(i_begin, i_end);
                auto const count   = static_cast<size_t>(count_s);
                DENSITY_ASSUME(static_cast<decltype(count_s)>(count) == count_s);

                auto const elements = static_cast<ValueType *>(
                  raw_allocate(sizeof(ValueType) * count, alignof(ValueType)));
                for (auto curr = elements; i_begin != i_end; ++i_begin, ++curr)
                    new (curr) ValueType(*i_begin);
                return elements;
            }

            template <typename INPUT_RANGE>
            auto raw_allocate_copy(const INPUT_RANGE & i_source_range)
              -> decltype(std::declval<reentrant_put_transaction>().raw_allocate_copy(
                std::begin(i_source_range), std::end(i_source_range)))
            {
                return raw_allocate_copy(std::begin(i_source_range), std::end(i_source_range));
            }

            void commit() noexcept
            {
                DENSITY_ASSERT(!empty());
                commit_reentrant_put_impl(m_put_data.m_control_block);
                m_queue = nullptr;
            }


            void cancel() noexcept
            {
                DENSITY_ASSERT(!empty());
                cancel_reentrant_put_impl(m_put_data.m_control_block);
                m_queue = nullptr;
            }

            bool empty() const noexcept { return m_queue == nullptr; }

            explicit operator bool() const noexcept { return m_queue != nullptr; }

            heter_queue * queue() const noexcept { return m_queue; }

            void * element_ptr() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return m_put_data.m_user_storage;
            }

#ifndef DOXYGEN_DOC_GENERATION
            template <
              typename EL                                               = ELEMENT_COMPLETE_TYPE,
              typename std::enable_if<!std::is_void<EL>::value>::type * = nullptr>
            EL &
#else
            ELEMENT_COMPLETE_TYPE &
#endif
              element() const noexcept
            {
                return *static_cast<ELEMENT_COMPLETE_TYPE *>(element_ptr());
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return *type_after_control(m_put_data.m_control_block);
            }

            ~reentrant_put_transaction()
            {
                if (m_queue != nullptr)
                {
                    cancel_reentrant_put_impl(m_put_data.m_control_block);
                }
            }

            reentrant_put_transaction(
              PrivateType, heter_queue * i_queue, Allocation i_push_data) noexcept
                : m_queue(i_queue), m_put_data(i_push_data)
            {
            }


          private:
            heter_queue * m_queue = nullptr;
            Allocation    m_put_data;

            template <typename OTHERTYPE> friend class reentrant_put_transaction;
        };


        class reentrant_consume_operation
        {
          public:
            reentrant_consume_operation() noexcept : m_control(nullptr) {}

            reentrant_consume_operation(const reentrant_consume_operation &) = delete;

            reentrant_consume_operation & operator=(const reentrant_consume_operation &) = delete;

            reentrant_consume_operation(reentrant_consume_operation && i_source) noexcept
                : m_queue(i_source.m_queue), m_control(i_source.m_control)
            {
                i_source.m_control = nullptr;
            }

            reentrant_consume_operation &
              operator=(reentrant_consume_operation && i_source) noexcept
            {
                if (this != &i_source)
                {
                    if (!empty())
                    {
                        commit();
                    }
                    m_queue            = i_source.m_queue;
                    m_control          = i_source.m_control;
                    i_source.m_control = nullptr;
                }
                return *this;
            }

            ~reentrant_consume_operation()
            {
                if (m_control != nullptr)
                {
                    m_queue->cancel_consume_impl(m_control);
                }
            }

            friend void swap(
              reentrant_consume_operation & i_first,
              reentrant_consume_operation & i_second) noexcept
            {
                std::swap(i_first.m_queue, i_second.m_queue);
                std::swap(i_first.m_control, i_second.m_control);
            }

            bool empty() const noexcept { return m_control == nullptr; }

            explicit operator bool() const noexcept { return m_control != nullptr; }

            heter_queue * queue() const noexcept
            {
                return m_control != nullptr ? m_queue : nullptr;
            }

            void commit() noexcept
            {
                DENSITY_ASSERT(!empty());

                auto const & type    = complete_type();
                auto const   element = element_ptr();
                type.destroy(element);

                type.RUNTIME_TYPE::~RUNTIME_TYPE();

                m_queue->commit_consume_impl(m_control);
                m_control = nullptr;
            }

            void commit_nodestroy() noexcept
            {
                DENSITY_ASSERT(!empty());

                bool destroy_type = !std::is_trivially_destructible<RUNTIME_TYPE>::value;
                if (destroy_type)
                {
                    auto const & type = complete_type();
                    type.RUNTIME_TYPE::~RUNTIME_TYPE();
                }

                m_queue->commit_consume_impl(m_control);
                m_control = nullptr;
            }

            void cancel() noexcept
            {
                DENSITY_ASSERT(!empty());

                m_queue->cancel_consume_impl(m_control);
                m_control = nullptr;
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return *type_after_control(m_control);
            }

            void * unaligned_element_ptr() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return get_unaligned_element(m_control);
            }

            void * element_ptr() const noexcept
            {
                DENSITY_ASSERT(!empty());
                return get_element(m_control);
            }

            template <typename COMPLETE_ELEMENT_TYPE>
            COMPLETE_ELEMENT_TYPE & element() const noexcept
            {
                DENSITY_ASSERT(!empty() && complete_type().template is<COMPLETE_ELEMENT_TYPE>());
                return *static_cast<COMPLETE_ELEMENT_TYPE *>(get_element(m_control));
            }

            reentrant_consume_operation(
              PrivateType, heter_queue * i_queue, ControlBlock * i_control) noexcept
                : m_queue(i_queue), m_control(i_control)
            {
            }

            bool start_consume_impl(PrivateType, heter_queue * i_queue) noexcept
            {
                if (m_control != nullptr)
                {
                    m_queue->cancel_consume_impl(m_control);
                }
                m_queue   = i_queue;
                m_control = i_queue->start_consume_impl();
                return m_control != nullptr;
            }

          private:
            heter_queue *  m_queue;
            ControlBlock * m_control;
        };

        template <typename ELEMENT_TYPE> void reentrant_push(ELEMENT_TYPE && i_source)
        {
            return reentrant_emplace<typename std::decay<ELEMENT_TYPE>::type>(
              std::forward<ELEMENT_TYPE>(i_source));
        }

        template <typename ELEMENT_TYPE, typename... CONSTRUCTION_PARAMS>
        void reentrant_emplace(CONSTRUCTION_PARAMS &&... i_construction_params)
        {
            start_reentrant_emplace<ELEMENT_TYPE>(
              std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...)
              .commit();
        }

        void reentrant_dyn_push(const RUNTIME_TYPE & i_type)
        {
            start_reentrant_dyn_push(i_type).commit();
        }

        void reentrant_dyn_push_copy(const RUNTIME_TYPE & i_type, const void * i_source)
        {
            start_reentrant_dyn_push_copy(i_type, i_source).commit();
        }

        void reentrant_dyn_push_move(const RUNTIME_TYPE & i_type, void * i_source)
        {
            start_reentrant_dyn_push_move(i_type, i_source).commit();
        }

        template <typename ELEMENT_TYPE>
        reentrant_put_transaction<typename std::decay<ELEMENT_TYPE>::type>
          start_reentrant_push(ELEMENT_TYPE && i_source)
        {
            return start_reentrant_emplace<typename std::decay<ELEMENT_TYPE>::type>(
              std::forward<ELEMENT_TYPE>(i_source));
        }

        template <typename ELEMENT_TYPE, typename... CONSTRUCTION_PARAMS>
        reentrant_put_transaction<ELEMENT_TYPE>
          start_reentrant_emplace(CONSTRUCTION_PARAMS &&... i_construction_params)
        {
            auto push_data = inplace_allocate<
              detail::Queue_Busy,
              true,
              detail::size_of<ELEMENT_TYPE>::value,
              alignof(ELEMENT_TYPE)>();

            RUNTIME_TYPE * type = nullptr;
            try
            {
                auto const type_storage = type_after_control(push_data.m_control_block);
                DENSITY_ASSUME(type_storage != nullptr);
                type = new (type_storage) RUNTIME_TYPE(RUNTIME_TYPE::template make<ELEMENT_TYPE>());

                DENSITY_ASSUME(push_data.m_user_storage != nullptr);
                new (push_data.m_user_storage)
                  ELEMENT_TYPE(std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...);
            }
            catch (...)
            {
                if (type != nullptr)
                    type->RUNTIME_TYPE::~RUNTIME_TYPE();
                DENSITY_ASSERT_INTERNAL(
                  (push_data.m_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
                  detail::Queue_Busy);
                push_data.m_control_block->m_next += detail::Queue_Dead - detail::Queue_Busy;
                throw;
            }

            return reentrant_put_transaction<ELEMENT_TYPE>(PrivateType(), this, push_data);
        }

        reentrant_put_transaction<> start_reentrant_dyn_push(const RUNTIME_TYPE & i_type)
        {
            auto push_data =
              inplace_allocate<detail::Queue_Busy, true>(i_type.size(), i_type.alignment());

            RUNTIME_TYPE * type = nullptr;
            try
            {
                auto const type_storage = type_after_control(push_data.m_control_block);
                DENSITY_ASSUME(type_storage != nullptr);
                type = new (type_storage) RUNTIME_TYPE(i_type);

                DENSITY_ASSUME(push_data.m_user_storage != nullptr);
                i_type.default_construct(push_data.m_user_storage);
            }
            catch (...)
            {
                if (type != nullptr)
                    type->RUNTIME_TYPE::~RUNTIME_TYPE();
                DENSITY_ASSERT_INTERNAL(
                  (push_data.m_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
                  detail::Queue_Busy);
                push_data.m_control_block->m_next += detail::Queue_Dead - detail::Queue_Busy;
                throw;
            }

            return reentrant_put_transaction<void>(PrivateType(), this, push_data);
        }


        reentrant_put_transaction<>
          start_reentrant_dyn_push_copy(const RUNTIME_TYPE & i_type, const void * i_source)
        {
            auto push_data =
              inplace_allocate<detail::Queue_Busy, true>(i_type.size(), i_type.alignment());

            RUNTIME_TYPE * type = nullptr;
            try
            {
                auto const type_storage = type_after_control(push_data.m_control_block);
                DENSITY_ASSUME(type_storage != nullptr);
                type = new (type_storage) RUNTIME_TYPE(i_type);

                DENSITY_ASSUME(push_data.m_user_storage != nullptr);
                i_type.copy_construct(push_data.m_user_storage, i_source);
            }
            catch (...)
            {
                if (type != nullptr)
                    type->RUNTIME_TYPE::~RUNTIME_TYPE();
                DENSITY_ASSERT_INTERNAL(
                  (push_data.m_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
                  detail::Queue_Busy);
                push_data.m_control_block->m_next += detail::Queue_Dead - detail::Queue_Busy;
                throw;
            }

            return reentrant_put_transaction<void>(PrivateType(), this, push_data);
        }

        reentrant_put_transaction<>
          start_reentrant_dyn_push_move(const RUNTIME_TYPE & i_type, void * i_source)
        {
            auto push_data =
              inplace_allocate<detail::Queue_Busy, true>(i_type.size(), i_type.alignment());

            RUNTIME_TYPE * type = nullptr;
            try
            {
                auto const type_storage = type_after_control(push_data.m_control_block);
                DENSITY_ASSUME(type_storage != nullptr);
                type = new (type_storage) RUNTIME_TYPE(i_type);

                DENSITY_ASSUME(push_data.m_user_storage != nullptr);
                i_type.move_construct(push_data.m_user_storage, i_source);
            }
            catch (...)
            {
                if (type != nullptr)
                    type->RUNTIME_TYPE::~RUNTIME_TYPE();
                DENSITY_ASSERT_INTERNAL(
                  (push_data.m_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
                  detail::Queue_Busy);
                push_data.m_control_block->m_next += detail::Queue_Dead - detail::Queue_Busy;
                throw;
            }

            return reentrant_put_transaction<void>(PrivateType(), this, push_data);
        }

        void reentrant_pop() noexcept { try_start_reentrant_consume().commit(); }

        bool try_reentrant_pop() noexcept
        {
            if (auto operation = try_start_reentrant_consume())
            {
                operation.commit();
                return true;
            }
            return false;
        }


        reentrant_consume_operation try_start_reentrant_consume() noexcept
        {
            return reentrant_consume_operation(PrivateType(), this, start_consume_impl());
        }

        bool try_start_reentrant_consume(reentrant_consume_operation & i_consume) noexcept
        {
            return i_consume.start_consume_impl(PrivateType(), this);
        }


        // const_iterator

        class const_iterator
        {
          public:
            using iterator_category = std::input_iterator_tag;
            using runtime_type      = RUNTIME_TYPE;
            using value_type        = const_dynamic_reference<RUNTIME_TYPE>;
            using pointer           = value_type *;
            using const_pointer     = const value_type *;
            using reference         = value_type &;
            using const_reference   = const value_type &;
            using size_type         = std::size_t;
            using difference_type   = std::ptrdiff_t;

            const_iterator() noexcept = default;
            const_iterator(const const_iterator & i_source) noexcept
                : m_control(i_source.m_control), m_queue(i_source.m_queue)
            {
                new (&m_value.m_pair) value_type(m_value.m_pair);
            }

            const_iterator & operator=(const const_iterator & i_source) noexcept
            {
                m_control = i_source.m_control;
                m_queue   = i_source.m_queue;
                new (&m_value.m_pair) value_type(m_value.m_pair);
                return *this;
            }

            const_iterator(const heter_queue * i_queue, ControlBlock * i_control) noexcept
                : m_control(i_control), m_queue(i_queue)
            {
                if (m_control != nullptr)
                {
                    const RUNTIME_TYPE & type = *type_after_control(m_control);
                    new (&m_value.m_pair) value_type(type, get_element(m_control));
                }
            }

            const value_type & operator*() const noexcept
            {
                DENSITY_ASSUME(m_control != nullptr);
                return m_value.m_pair;
            }

            const value_type * operator->() const noexcept
            {
                DENSITY_ASSUME(m_control != nullptr);
                return &m_value.m_pair;
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                DENSITY_ASSUME(m_control != nullptr);
                return m_value.m_pair.type();
            }

            const void * element_ptr() const noexcept
            {
                DENSITY_ASSUME(m_control != nullptr);
                return m_value.m_pair.address();
            }

            const_iterator & operator++() noexcept
            {
                DENSITY_ASSUME(m_queue != nullptr);
                m_control = m_queue->next_valid(m_control);
                if (m_control != nullptr)
                {
                    RUNTIME_TYPE & type = *type_after_control(m_control);
                    new (&m_value.m_pair) value_type(type, get_element(m_control));
                }
                return *this;
            }

            const_iterator operator++(int) noexcept
            {
                DENSITY_ASSUME(m_queue != nullptr);
                auto const prev_state = *this;
                ++*this;
                return prev_state;
            }

            bool operator==(const const_iterator & i_other) const noexcept
            {
                return m_control == i_other.m_control;
            }

            bool operator!=(const const_iterator & i_other) const noexcept
            {
                return m_control != i_other.m_control;
            }

          private:
            ControlBlock *      m_control = nullptr;
            const heter_queue * m_queue   = nullptr;
            union Value {
                value_type m_pair;
                Value() noexcept {}
                ~Value() {}
            } m_value;
        };

        // iterator

        class iterator
        {
          public:
            using iterator_category = std::input_iterator_tag;
            using runtime_type      = RUNTIME_TYPE;
            using value_type        = dynamic_reference<RUNTIME_TYPE>;
            using pointer           = value_type *;
            using const_pointer     = const value_type *;
            using reference         = value_type &;
            using const_reference   = const value_type &;
            using size_type         = std::size_t;
            using difference_type   = std::ptrdiff_t;

            iterator() noexcept = default;
            iterator(const iterator & i_source) noexcept
                : m_control(i_source.m_control), m_queue(i_source.m_queue)
            {
                new (&m_value.m_pair) value_type(m_value.m_pair);
            }

            iterator & operator=(const iterator & i_source) noexcept
            {
                m_control = i_source.m_control;
                m_queue   = i_source.m_queue;
                new (&m_value.m_pair) value_type(m_value.m_pair);
                return *this;
            }

            iterator(const heter_queue * i_queue, ControlBlock * i_control) noexcept
                : m_control(i_control), m_queue(i_queue)
            {
                if (m_control != nullptr)
                {
                    const RUNTIME_TYPE & type = *type_after_control(m_control);
                    new (&m_value.m_pair) value_type(type, get_element(m_control));
                }
            }

            const value_type & operator*() const noexcept
            {
                DENSITY_ASSUME(m_control != nullptr);
                return m_value.m_pair;
            }

            const value_type * operator->() const noexcept
            {
                DENSITY_ASSUME(m_control != nullptr);
                return &m_value.m_pair;
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                DENSITY_ASSUME(m_control != nullptr);
                return m_value.m_pair.type();
            }

            void * element_ptr() const noexcept
            {
                DENSITY_ASSUME(m_control != nullptr);
                return m_value.m_pair.address();
            }

            iterator & operator++() noexcept
            {
                DENSITY_ASSUME(m_queue != nullptr);
                m_control = m_queue->next_valid(m_control);
                if (m_control != nullptr)
                {
                    RUNTIME_TYPE & type = *type_after_control(m_control);
                    new (&m_value.m_pair) value_type(type, get_element(m_control));
                }
                return *this;
            }

            iterator operator++(int) noexcept
            {
                DENSITY_ASSUME(m_queue != nullptr);
                auto const prev_state = *this;
                ++*this;
                return prev_state;
            }

            bool operator==(const iterator & i_other) const noexcept
            {
                return m_control == i_other.m_control;
            }

            bool operator!=(const iterator & i_other) const noexcept
            {
                return m_control != i_other.m_control;
            }

          private:
            ControlBlock *      m_control = nullptr;
            const heter_queue * m_queue   = nullptr;
            union Value {
                value_type m_pair;
                Value() noexcept {}
                ~Value() {}
            } m_value;
        };

        iterator begin() noexcept { return iterator(this, first_valid(m_head)); }
        iterator end() noexcept { return iterator(); }

        const_iterator begin() const noexcept { return const_iterator(this, first_valid(m_head)); }
        const_iterator end() const noexcept { return const_iterator(); }

        const_iterator cbegin() const noexcept { return const_iterator(this, first_valid(m_head)); }
        const_iterator cend() const noexcept { return const_iterator(); }

        bool operator==(const heter_queue & i_source) const
        {
            const auto end_1 = cend();
            const auto end_2 = i_source.cend();
            auto       it_2  = i_source.cbegin();
            for (auto it_1 = cbegin(); it_1 != end_1; ++it_1, ++it_2)
            {
                if (
                  it_2 == end_2 || it_1.complete_type() != it_2.complete_type() ||
                  !it_1.complete_type().are_equal(it_1.element_ptr(), it_2.element_ptr()))
                {
                    return false;
                }
            }
            return it_2 == end_2;
        }

        bool operator!=(const heter_queue & i_source) const { return !operator==(i_source); }

      private:
        ControlBlock * first_valid(ControlBlock * i_from) const
        {
            for (auto curr = i_from; curr != m_tail;)
            {
                if ((curr->m_next & (detail::Queue_Busy | detail::Queue_Dead)) == 0)
                {
                    return curr;
                }
                curr = reinterpret_cast<ControlBlock *>(curr->m_next & ~detail::Queue_AllFlags);
            }
            return nullptr;
        }

        ControlBlock * next_valid(ControlBlock * i_from) const
        {
            DENSITY_ASSUME(i_from != m_tail);
            for (auto curr =
                   reinterpret_cast<ControlBlock *>(i_from->m_next & ~static_cast<uintptr_t>(3));
                 curr != m_tail;)
            {
                if ((curr->m_next & (detail::Queue_Busy | detail::Queue_Dead)) == 0)
                {
                    return curr;
                }
                curr = reinterpret_cast<ControlBlock *>(curr->m_next & ~detail::Queue_AllFlags);
            }
            return nullptr;
        }

        static RUNTIME_TYPE * type_after_control(ControlBlock * i_control) noexcept
        {
            return reinterpret_cast<RUNTIME_TYPE *>(address_add(i_control, s_sizeof_ControlBlock));
        }

        static void * get_unaligned_element(ControlBlock * i_control) noexcept
        {
            auto result = address_add(i_control, s_sizeof_ControlBlock + s_sizeof_RuntimeType);
            if (i_control->m_next & detail::Queue_External)
            {
                result = static_cast<ExternalBlock *>(result)->m_element;
            }
            return result;
        }

        static void * get_element(detail::QueueControl * i_control) noexcept
        {
            auto result = address_add(i_control, s_sizeof_ControlBlock + s_sizeof_RuntimeType);
            if (i_control->m_next & detail::Queue_External)
            {
                result = static_cast<ExternalBlock *>(result)->m_element;
            }
            else
            {
                result = address_upper_align(result, type_after_control(i_control)->alignment());
            }
            return result;
        }

        static bool same_page(const void * i_first, const void * i_second) noexcept
        {
            auto const page_mask = ALLOCATOR_TYPE::page_alignment - 1;
            return ((reinterpret_cast<uintptr_t>(i_first) ^ reinterpret_cast<uintptr_t>(i_second)) &
                    ~page_mask) == 0;
        }

        struct ExternalBlock
        {
            void * m_element;
            size_t m_size, m_alignment;
        };

        static void * get_end_of_page(void * i_address) noexcept
        {
            return address_add(
              address_lower_align(i_address, allocator_type::page_alignment),
              allocator_type::page_size - s_sizeof_ControlBlock);
        }


        template <uintptr_t CONTROL_BITS, bool INCLUDE_TYPE>
        Allocation inplace_allocate(size_t i_size, size_t i_alignment)
        {
            DENSITY_ASSERT_INTERNAL(is_power_of_2(i_alignment) && (i_size % i_alignment) == 0);

            if (i_alignment < min_alignment)
            {
                i_alignment = min_alignment;
                i_size      = uint_upper_align(i_size, min_alignment);
            }

            for (;;)
            {
                DENSITY_ASSERT_INTERNAL(
                  address_is_aligned(m_tail, min_alignment) ||
                  m_tail == reinterpret_cast<ControlBlock *>(s_invalid_control_block));

                // allocate space for the control block and the runtime type
                auto const control_block = m_tail;
                void *     new_tail      = address_add(
                  control_block,
                  INCLUDE_TYPE ? (s_sizeof_ControlBlock + s_sizeof_RuntimeType)
                               : s_sizeof_ControlBlock);

                // allocate space for the element
                new_tail                  = address_upper_align(new_tail, i_alignment);
                void * const user_storage = new_tail;
                new_tail                  = address_add(new_tail, i_size);

                // check if a page overflow would occur
                void * end_of_page = get_end_of_page(control_block);
                if (DENSITY_LIKELY(new_tail <= end_of_page))
                {
                    DENSITY_ASSUME(control_block != nullptr);
                    new (control_block) ControlBlock();

                    control_block->m_next = reinterpret_cast<uintptr_t>(new_tail) + CONTROL_BITS;
                    m_tail                = static_cast<ControlBlock *>(new_tail);
                    return Allocation{control_block, user_storage};
                }
                else if (
                  i_size + (i_alignment - min_alignment) <=
                  s_max_size_inpage) // if this allocation may fit in a page
                {
                    // allocate a new page and redo
                    allocate_new_page();
                }
                else
                {
                    // this allocation would never fit in a page, allocate an external block
                    return external_allocate<CONTROL_BITS>(i_size, i_alignment);
                }
            }
        }

        template <uintptr_t CONTROL_BITS, bool INCLUDE_TYPE, size_t SIZE, size_t ALIGNMENT>
        Allocation inplace_allocate()
        {
            static_assert(is_power_of_2(ALIGNMENT) && (SIZE % ALIGNMENT) == 0, "");

            DENSITY_ASSERT_INTERNAL(
              address_is_aligned(m_tail, min_alignment) ||
              m_tail == reinterpret_cast<ControlBlock *>(s_invalid_control_block));

            constexpr size_t alignment = detail::size_max(ALIGNMENT, min_alignment);
            constexpr size_t size      = uint_upper_align(SIZE, alignment);
            constexpr bool   can_fit_in_a_page =
              size + (alignment - min_alignment) <= s_max_size_inpage;
            constexpr bool over_aligned = alignment > min_alignment;

            for (;;)
            {
                // allocate space for the control block and the runtime type
                auto const control_block = m_tail;
                void *     new_tail      = address_add(
                  control_block,
                  INCLUDE_TYPE ? (s_sizeof_ControlBlock + s_sizeof_RuntimeType)
                               : s_sizeof_ControlBlock);

                // allocate space for the element
                if (over_aligned)
                {
                    new_tail = address_upper_align(new_tail, alignment);
                }
                DENSITY_ASSERT_INTERNAL(
                  address_is_aligned(new_tail, min_alignment) ||
                  m_tail == reinterpret_cast<ControlBlock *>(s_invalid_control_block));
                void * new_element = new_tail;
                new_tail           = address_add(new_tail, size);

                // check if a page overflow would occur
                void * end_of_page = get_end_of_page(control_block);
                if (DENSITY_LIKELY(new_tail <= end_of_page))
                {
                    DENSITY_ASSUME(control_block != nullptr);
                    new (control_block) ControlBlock();

                    control_block->m_next = reinterpret_cast<uintptr_t>(new_tail) + CONTROL_BITS;
                    m_tail                = static_cast<ControlBlock *>(new_tail);
                    return Allocation{control_block, new_element};
                }
                else if (can_fit_in_a_page)
                {
                    // allocate a new page and redo
                    allocate_new_page();
                }
                else
                {
                    // this allocation would never fit in a page, allocate an external block
                    return external_allocate<CONTROL_BITS>(size, alignment);
                }
            }
        }

        template <uintptr_t CONTROL_BITS>
        Allocation external_allocate(size_t i_size, size_t i_alignment)
        {
            auto const external_block = ALLOCATOR_TYPE::allocate(i_size, i_alignment);
            try
            {
                /* external blocks always allocate space for the type, because it would be complicated
                    for the consumers to handle both cases*/
                auto const inplace_put = inplace_allocate<CONTROL_BITS, true>(
                  sizeof(ExternalBlock), alignof(ExternalBlock));

                new (inplace_put.m_user_storage) ExternalBlock{external_block, i_size, i_alignment};

                DENSITY_ASSERT((inplace_put.m_control_block->m_next & detail::Queue_External) == 0);
                inplace_put.m_control_block->m_next |= detail::Queue_External;
                return Allocation{inplace_put.m_control_block, external_block};
            }
            catch (...)
            {
                /* if inplace_allocate fails, that means that we were able to allocate the external block,
                    but we were not able to put the struct ExternalBlock in the page (because a new page was
                    necessary, but we could not allocate it). */
                ALLOCATOR_TYPE::deallocate(external_block, i_size, i_alignment);
                throw;
            }
        }

        DENSITY_NO_INLINE void allocate_new_page()
        {
            // page overflow
            if (DENSITY_LIKELY(m_tail != reinterpret_cast<ControlBlock *>(s_invalid_control_block)))
            {
                auto const control_block = m_tail;

                DENSITY_ASSUME(control_block != nullptr);
                new (control_block) ControlBlock();

                auto new_page         = allocator_type::allocate_page();
                control_block->m_next = reinterpret_cast<uintptr_t>(new_page) + detail::Queue_Dead;
                m_tail                = static_cast<ControlBlock *>(new_page);
            }
            else
            {
                // this happens only on a virgin queue
                m_tail = m_head = static_cast<ControlBlock *>(allocator_type::allocate_page());
            }
        }

        DENSITY_NO_INLINE static void cancel_put_impl(ControlBlock * i_control_block)
        {
            const auto type_ptr = type_after_control(i_control_block);
            type_ptr->destroy(get_element(i_control_block));

            type_ptr->RUNTIME_TYPE::~RUNTIME_TYPE();

            DENSITY_ASSERT_INTERNAL(
              (i_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) == 0);
            i_control_block->m_next += detail::Queue_Dead;
        }

        static void commit_reentrant_put_impl(ControlBlock * i_control_block)
        {
            DENSITY_ASSERT_INTERNAL(
              (i_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
              detail::Queue_Busy);
            i_control_block->m_next -= detail::Queue_Busy;
        }

        DENSITY_NO_INLINE static void cancel_reentrant_put_impl(ControlBlock * i_control_block)
        {
            const auto type_ptr = type_after_control(i_control_block);
            type_ptr->destroy(get_element(i_control_block));

            type_ptr->RUNTIME_TYPE::~RUNTIME_TYPE();

            DENSITY_ASSERT_INTERNAL(
              (i_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
              detail::Queue_Busy);
            i_control_block->m_next += (detail::Queue_Dead - detail::Queue_Busy);
        }

        ControlBlock * start_consume_impl() noexcept
        {
            auto       curr = m_head;
            auto const tail = m_tail;
            while (curr != tail)
            {
                if ((curr->m_next & (detail::Queue_Busy | detail::Queue_Dead)) == 0)
                {
                    curr->m_next += detail::Queue_Busy;
                    return curr;
                }

                curr = reinterpret_cast<ControlBlock *>(curr->m_next & ~detail::Queue_AllFlags);
            }

            return nullptr;
        }

        void commit_consume_impl(ControlBlock * i_control_block) noexcept
        {
            DENSITY_ASSERT_INTERNAL(
              (i_control_block->m_next & (detail::Queue_Busy | detail::Queue_Dead)) ==
              detail::Queue_Busy);
            i_control_block->m_next += (detail::Queue_Dead - detail::Queue_Busy);

            clean_dead_elements();
        }

        void clean_dead_elements() noexcept
        {
            auto curr = m_head;
            while (curr != m_tail)
            {
                // break if the current block is busy or is not dead
                if (
                  (curr->m_next & (detail::Queue_Busy | detail::Queue_Dead)) != detail::Queue_Dead)
                {
                    break;
                }

                auto next =
                  reinterpret_cast<ControlBlock *>(curr->m_next & ~detail::Queue_AllFlags);
                if (curr->m_next & detail::Queue_External)
                {
                    auto result = address_add(curr, s_sizeof_ControlBlock + s_sizeof_RuntimeType);
                    const auto & block = *static_cast<ExternalBlock *>(result);
                    ALLOCATOR_TYPE::deallocate(block.m_element, block.m_size, block.m_alignment);
                }

                if (!same_page(next, curr))
                {
                    allocator_type::deallocate_page(curr);
                }

                curr = next;
            }

            DENSITY_ASSERT_INTERNAL(
              curr == m_tail ||
              (curr->m_next & (detail::Queue_Busy | detail::Queue_Dead)) != detail::Queue_Dead);
            m_head = curr;
        }

        void cancel_consume_impl(ControlBlock * i_control_block) noexcept
        {
            DENSITY_ASSERT_INTERNAL(
              (i_control_block->m_next & (detail::Queue_AllFlags - detail::Queue_External)) ==
              detail::Queue_Busy);
            i_control_block->m_next -= detail::Queue_Busy;
        }

        void destroy() noexcept
        {
            clear();
            DENSITY_ASSERT_INTERNAL(m_tail == m_head);
            if (m_head != reinterpret_cast<ControlBlock *>(s_invalid_control_block))
            {
                allocator_type::deallocate_page(m_head);
            }
        }
    };

} // namespace density