conc_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/heter_queue.h>
#include <mutex>

namespace density
{
    template <typename RUNTIME_TYPE = runtime_type<>, typename ALLOCATOR_TYPE = default_allocator>
    class conc_heter_queue
    {
        using InnerQueue = heter_queue<RUNTIME_TYPE, ALLOCATOR_TYPE>;

        enum class PrivateType
        {
        };

      public:
        static constexpr bool concurrent_puts = true;

        static constexpr bool concurrent_consumes = true;

        static constexpr bool concurrent_put_consumes = true;

        static constexpr bool is_seq_cst = true;

        constexpr static size_t min_alignment = InnerQueue::min_alignment;

        using runtime_type    = RUNTIME_TYPE;
        using value_type      = std::pair<const runtime_type &, void * const>;
        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 iterator;
        class const_iterator;

        constexpr conc_heter_queue() noexcept = default;

        constexpr explicit conc_heter_queue(const ALLOCATOR_TYPE & i_source_allocator) noexcept
            : m_queue(i_source_allocator)
        {
        }

        constexpr explicit conc_heter_queue(ALLOCATOR_TYPE && i_source_allocator) noexcept
            : m_queue(std::move(i_source_allocator))
        {
        }

        conc_heter_queue(conc_heter_queue && i_source) noexcept
            : m_queue(std::move(i_source.m_queue))
        {
        }

        conc_heter_queue & operator=(conc_heter_queue && i_source) noexcept
        {
            m_queue = std::move(i_source.m_queue);
            return *this;
        }

        allocator_type
          get_allocator() noexcept(std::is_nothrow_copy_constructible<allocator_type>::value)
        {
            return m_queue.get_allocator();
        }

        allocator_type & get_allocator_ref() noexcept { return m_queue.get_allocator_ref(); }

        const allocator_type & get_allocator_ref() const noexcept
        {
            return m_queue.get_allocator_ref();
        }

        friend void swap(
          conc_heter_queue<RUNTIME_TYPE, ALLOCATOR_TYPE> & i_first,
          conc_heter_queue<RUNTIME_TYPE, ALLOCATOR_TYPE> & i_second) noexcept
        {
            swap(i_first.m_queue, i_second.m_queue);
        }

        ~conc_heter_queue() = default;

        bool empty() const noexcept
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            return m_queue.empty();
        }

        void clear() noexcept
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            m_queue.clear();
        }

        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 = default;

            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_lock(std::move(i_source.m_lock)),
                  m_put_transaction(std::move(i_source.m_put_transaction))
            {
            }

            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();

                    put_transaction source(std::move(i_source));

                    using namespace density;
                    swap(m_put_transaction, source.m_put_transaction);
                    swap(m_lock, source.m_lock);
                }
                return *this;
            }

            friend void swap(put_transaction & i_first, put_transaction & i_second) noexcept
            {
                using namespace std;
                swap(i_first.m_put_transaction, i_second.m_put_transaction);
                swap(i_first.m_lock, i_second.m_lock);
            }

            void * raw_allocate(size_t i_size, size_t i_alignment)
            {
                return m_put_transaction.raw_allocate(i_size, i_alignment);
            }

            template <typename INPUT_ITERATOR>
            typename std::iterator_traits<INPUT_ITERATOR>::value_type *
              raw_allocate_copy(INPUT_ITERATOR i_begin, INPUT_ITERATOR i_end)
            {
                return m_put_transaction.raw_allocate_copy(i_begin, i_end);
            }

            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 m_put_transaction.raw_allocate_copy(
                  std::begin(i_source_range), std::end(i_source_range));
            }

            void commit() noexcept
            {
                DENSITY_ASSERT(m_put_transaction && m_lock.owns_lock());
                m_put_transaction.commit();
                m_lock.unlock();
            }

            void cancel() noexcept
            {
                DENSITY_ASSERT(m_put_transaction && m_lock.owns_lock());
                m_put_transaction.cancel();
                m_lock.unlock();
            }

            bool empty() const noexcept { return m_put_transaction.empty(); }

            explicit operator bool() const noexcept { return m_put_transaction.operator bool(); }


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

#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 *>(m_put_transaction.element_ptr());
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                return m_put_transaction.complete_type();
            }

            ~put_transaction()
            {
                if (m_lock.owns_lock())
                {
                    m_put_transaction.cancel();
                }
            }

            put_transaction(
              PrivateType,
              std::unique_lock<std::mutex> && i_lock,
              typename InnerQueue::template put_transaction<ELEMENT_COMPLETE_TYPE> &&
                i_put_transaction) noexcept
                : m_lock(std::move(i_lock)), m_put_transaction(std::move(i_put_transaction))
            {
                if (!m_put_transaction)
                    m_lock.unlock();
            }

          private: // data members
            std::unique_lock<std::mutex>                                         m_lock;
            typename InnerQueue::template put_transaction<ELEMENT_COMPLETE_TYPE> m_put_transaction;
            template <typename OTHERTYPE> friend class put_transaction;
        };

        class consume_operation
        {
          public:
            consume_operation() noexcept = default;

            consume_operation(const consume_operation &) = delete;

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

            consume_operation(consume_operation && i_source) noexcept = default;

            consume_operation & operator=(consume_operation && i_source) noexcept
            {
                m_lock              = std::move(i_source.m_lock);
                m_consume_operation = std::move(i_source.m_consume_operation);
                return *this;
            }

            ~consume_operation()
            {
                if (m_lock.owns_lock())
                    cancel();
            }

            friend void swap(consume_operation & i_first, consume_operation & i_second) noexcept
            {
                std::swap(i_first.m_lock, i_second.m_lock);
                std::swap(i_first.m_consume_operation, i_second.m_consume_operation);
            }

            bool empty() const noexcept { return m_consume_operation.empty(); }

            explicit operator bool() const noexcept { return m_consume_operation.operator bool(); }

            void commit() noexcept
            {
                DENSITY_ASSERT(!empty());
                m_consume_operation.commit();
                m_lock.unlock();
            }

            void commit_nodestroy() noexcept
            {
                DENSITY_ASSERT(!empty());
                m_consume_operation.commit_nodestroy();
                m_lock.unlock();
            }

            void cancel() noexcept
            {
                DENSITY_ASSERT(!empty());
                m_consume_operation.cancel();
                m_lock.unlock();
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                return m_consume_operation.complete_type();
            }

            void * unaligned_element_ptr() const noexcept
            {
                return m_consume_operation.unaligned_element_ptr();
            }

            void * element_ptr() const noexcept { return m_consume_operation.element_ptr(); }

            template <typename COMPLETE_ELEMENT_TYPE>
            COMPLETE_ELEMENT_TYPE & element() const noexcept
            {
                return m_consume_operation.template element<COMPLETE_ELEMENT_TYPE>();
            }

            consume_operation(
              PrivateType,
              std::unique_lock<std::mutex> &&           i_lock,
              typename InnerQueue::consume_operation && i_consume_operation) noexcept
                : m_lock(std::move(i_lock)), m_consume_operation(std::move(i_consume_operation))
            {
            }

            bool start_consume_impl(PrivateType, conc_heter_queue * i_queue)
            {
                if (m_lock.owns_lock())
                    cancel();

                m_lock = std::unique_lock<std::mutex>(i_queue->m_mutex);

                bool const result = i_queue->m_queue.try_start_consume(m_consume_operation);
                if (!result)
                    m_lock.unlock();
                return result;
            }

          private:
            std::unique_lock<std::mutex>           m_lock;
            typename InnerQueue::consume_operation m_consume_operation;
        };

        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)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            return put_transaction<ELEMENT_TYPE>(
              PrivateType(),
              std::move(lock),
              m_queue.template start_emplace<ELEMENT_TYPE>(
                std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...));
        }

        put_transaction<> start_dyn_push(const runtime_type & i_type)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            return put_transaction<>(
              PrivateType(), std::move(lock), m_queue.start_dyn_push(i_type));
        }


        put_transaction<> start_dyn_push_copy(const runtime_type & i_type, const void * i_source)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            return put_transaction<>(
              PrivateType(), std::move(lock), m_queue.start_dyn_push_copy(i_type, i_source));
        }

        put_transaction<> start_dyn_push_move(const runtime_type & i_type, void * i_source)
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            return put_transaction<>(
              PrivateType(), std::move(lock), m_queue.start_dyn_push_move(i_type, i_source));
        }


        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
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            auto                         consume = m_queue.try_start_consume();
            if (!consume)
                lock.unlock();
            return consume_operation(PrivateType(), std::move(lock), std::move(consume));
        }

        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 = default;

            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_transaction(std::move(i_source.m_put_transaction))
            {
                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
            {
                m_queue           = i_source.m_queue;
                m_put_transaction = std::move(i_source.m_put_transaction);
                i_source.m_queue  = nullptr;
                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_transaction, i_second.m_put_transaction);
            }

            void * raw_allocate(size_t i_size, size_t i_alignment)
            {
                DENSITY_ASSERT(!empty());
                std::lock_guard<std::mutex> lock(m_queue->m_mutex);
                return m_put_transaction.raw_allocate(i_size, i_alignment);
            }

            template <typename INPUT_ITERATOR>
            typename std::iterator_traits<INPUT_ITERATOR>::value_type *
              raw_allocate_copy(INPUT_ITERATOR i_begin, INPUT_ITERATOR i_end)
            {
                DENSITY_ASSERT(!empty());
                std::lock_guard<std::mutex> lock(m_queue->m_mutex);
                return m_put_transaction.raw_allocate_copy(i_begin, i_end);
            }

            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)))
            {
                DENSITY_ASSERT(!empty());
                std::lock_guard<std::mutex> lock(m_queue->m_mutex);
                return m_put_transaction.raw_allocate_copy(
                  std::begin(i_source_range), std::end(i_source_range));
            }

            void commit() noexcept
            {
                DENSITY_ASSERT(!empty());
                std::lock_guard<std::mutex> lock(m_queue->m_mutex);
                m_put_transaction.commit();
                m_queue = nullptr;
            }


            void cancel() noexcept
            {
                DENSITY_ASSERT(!empty());
                std::lock_guard<std::mutex> lock(m_queue->m_mutex);
                m_put_transaction.cancel();
                m_queue = nullptr;
            }

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

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

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

            void * element_ptr() const noexcept { return m_put_transaction.element_ptr(); }

#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
            {
                return m_put_transaction.complete_type();
            }

            ~reentrant_put_transaction()
            {
                if (m_put_transaction)
                {
                    cancel();
                }
            }

            reentrant_put_transaction(
              PrivateType,
              conc_heter_queue * i_queue,
              typename InnerQueue::template reentrant_put_transaction<ELEMENT_COMPLETE_TYPE> &&
                i_put_transaction) noexcept
                : m_queue(i_queue), m_put_transaction(std::move(i_put_transaction))
            {
            }

          private:
            conc_heter_queue * m_queue = nullptr;
            typename InnerQueue::template reentrant_put_transaction<ELEMENT_COMPLETE_TYPE>
              m_put_transaction;
            template <typename OTHERTYPE> friend class reentrant_put_transaction;
        };


        class reentrant_consume_operation
        {
          public:
            reentrant_consume_operation() noexcept = default;

            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 = default;

            reentrant_consume_operation &
              operator=(reentrant_consume_operation && i_source) noexcept = default;

            ~reentrant_consume_operation()
            {
                if (!empty())
                {
                    cancel();
                }
            }

            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_consume_operation, i_second.m_consume_operation);
            }

            bool empty() const noexcept { return m_consume_operation.empty(); }

            explicit operator bool() const noexcept { return m_consume_operation.operator bool(); }

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

            void commit() noexcept
            {
                DENSITY_ASSERT(!empty());
                std::lock_guard<std::mutex> lock(m_queue->m_mutex);
                m_consume_operation.commit();
            }

            void commit_nodestroy() noexcept
            {
                DENSITY_ASSERT(!empty());
                std::lock_guard<std::mutex> lock(m_queue->m_mutex);
                m_consume_operation.commit_nodestroy();
            }

            void cancel() noexcept
            {
                DENSITY_ASSERT(!empty());
                std::lock_guard<std::mutex> lock(m_queue->m_mutex);
                m_consume_operation.cancel();
            }

            const RUNTIME_TYPE & complete_type() const noexcept
            {
                return m_consume_operation.complete_type();
            }

            void * unaligned_element_ptr() const noexcept
            {
                return m_consume_operation.unaligned_element_ptr();
            }

            void * element_ptr() const noexcept { return m_consume_operation.element_ptr(); }

            template <typename COMPLETE_ELEMENT_TYPE>
            COMPLETE_ELEMENT_TYPE & element() const noexcept
            {
                return m_consume_operation.template element<COMPLETE_ELEMENT_TYPE>();
            }

            reentrant_consume_operation(
              PrivateType,
              conc_heter_queue *                                  i_queue,
              typename InnerQueue::reentrant_consume_operation && i_consume_operation) noexcept
                : m_queue(i_queue), m_consume_operation(std::move(i_consume_operation))
            {
            }

            bool start_consume_impl(PrivateType, conc_heter_queue * i_queue)
            {
                DENSITY_ASSUME(i_queue != nullptr);

                // first we take the locks, because they may throw
                std::unique_lock<std::mutex> lock;
                if (m_queue != nullptr)
                {
                    lock = std::unique_lock<std::mutex>(m_queue->m_mutex);
                }
                std::unique_lock<std::mutex> new_lock;
                if (m_queue != i_queue && i_queue != nullptr)
                {
                    new_lock = std::unique_lock<std::mutex>(i_queue->m_mutex);
                }

                // nothing can throw from now on
                if (m_consume_operation)
                    m_consume_operation.cancel();
                lock    = std::move(new_lock);
                m_queue = i_queue;
                return m_queue->m_queue.try_start_reentrant_consume(m_consume_operation);
            }

          private:
            conc_heter_queue *                               m_queue = nullptr;
            typename InnerQueue::reentrant_consume_operation m_consume_operation;
        };

        template <typename ELEMENT_TYPE> void reentrant_push(ELEMENT_TYPE && i_source)
        {
            return reentrant_emplace<ELEMENT_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<typename std::decay<ELEMENT_TYPE>::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)
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            auto put_transaction = m_queue.template start_reentrant_emplace<ELEMENT_TYPE>(
              std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...);
            return reentrant_put_transaction<ELEMENT_TYPE>(
              PrivateType(), this, std::move(put_transaction));
        }

        reentrant_put_transaction<> start_reentrant_dyn_push(const runtime_type & i_type)
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            auto                        put_transaction = m_queue.start_reentrant_dyn_push(i_type);
            return reentrant_put_transaction<>(PrivateType(), this, std::move(put_transaction));
        }


        reentrant_put_transaction<>
          start_reentrant_dyn_push_copy(const runtime_type & i_type, const void * i_source)
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            auto put_transaction = m_queue.start_reentrant_dyn_push_copy(i_type, i_source);
            return reentrant_put_transaction<>(PrivateType(), this, std::move(put_transaction));
        }

        reentrant_put_transaction<>
          start_reentrant_dyn_push_move(const runtime_type & i_type, void * i_source)
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            auto put_transaction = m_queue.start_reentrant_dyn_push_move(i_type, i_source);
            return reentrant_put_transaction<>(PrivateType(), this, std::move(put_transaction));
        }

        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
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            return reentrant_consume_operation(
              PrivateType(), this, m_queue.try_start_reentrant_consume());
        }

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

      private:
        mutable std::mutex m_mutex;
        InnerQueue         m_queue;
    };


} // namespace density