conc_function_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/conc_heter_queue.h>
#include <density/detail/function_runtime_type.h>

namespace density
{
    template <
      typename CALLABLE,
      typename ALLOCATOR_TYPE       = default_allocator,
      function_type_erasure ERASURE = function_standard_erasure>
    class conc_function_queue;

#ifndef DOXYGEN_DOC_GENERATION
    template <
      typename RET_VAL,
      typename... PARAMS,
      typename ALLOCATOR_TYPE,
      function_type_erasure ERASURE>
    class conc_function_queue<RET_VAL(PARAMS...), ALLOCATOR_TYPE, ERASURE>
#else
    template <
      typename CALLABLE,
      typename ALLOCATOR_TYPE       = default_allocator,
      function_type_erasure ERASURE = function_standard_erasure>
    class conc_function_queue
#endif
    {
      private:
        using UnderlyingQueue = conc_heter_queue<
          detail::FunctionRuntimeType<ERASURE, RET_VAL(PARAMS...)>,
          ALLOCATOR_TYPE>;
        UnderlyingQueue m_queue;

      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;

        conc_function_queue() noexcept = default;

        conc_function_queue(conc_function_queue && i_source) noexcept = default;

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

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

        ~conc_function_queue()
        {
            auto erasure = ERASURE;
            if (erasure == function_manual_clear)
            {
                DENSITY_ASSERT(empty());
            }
        }

        template <typename ELEMENT_COMPLETE_TYPE>
        using put_transaction =
          typename UnderlyingQueue::template put_transaction<ELEMENT_COMPLETE_TYPE>;

        template <typename ELEMENT_COMPLETE_TYPE>
        using reentrant_put_transaction =
          typename UnderlyingQueue::template reentrant_put_transaction<ELEMENT_COMPLETE_TYPE>;

        using consume_operation = typename UnderlyingQueue::consume_operation;

        using reentrant_consume_operation = typename UnderlyingQueue::reentrant_consume_operation;

        template <typename ELEMENT_COMPLETE_TYPE> void push(ELEMENT_COMPLETE_TYPE && i_source)
        {
            m_queue.push(std::forward<ELEMENT_COMPLETE_TYPE>(i_source));
        }

        template <typename ELEMENT_COMPLETE_TYPE, typename... CONSTRUCTION_PARAMS>
        void emplace(CONSTRUCTION_PARAMS &&... i_construction_params)
        {
            m_queue.template emplace<ELEMENT_COMPLETE_TYPE>(
              std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...);
        }

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


        template <typename ELEMENT_TYPE, typename... CONSTRUCTION_PARAMS>
        put_transaction<ELEMENT_TYPE> start_emplace(CONSTRUCTION_PARAMS &&... i_construction_params)
        {
            return m_queue.template start_emplace<ELEMENT_TYPE>(
              std::forward<ELEMENT_TYPE>(i_construction_params)...);
        }

        template <typename ELEMENT_COMPLETE_TYPE>
        void reentrant_push(ELEMENT_COMPLETE_TYPE && i_source)
        {
            m_queue.reentrant_push(std::forward<ELEMENT_COMPLETE_TYPE>(i_source));
        }

        template <typename ELEMENT_COMPLETE_TYPE, typename... CONSTRUCTION_PARAMS>
        void reentrant_emplace(CONSTRUCTION_PARAMS &&... i_construction_params)
        {
            m_queue.template reentrant_emplace<ELEMENT_COMPLETE_TYPE>(
              std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...);
        }

        template <typename ELEMENT_TYPE>
        reentrant_put_transaction<typename std::decay<ELEMENT_TYPE>::type>
          start_reentrant_push(ELEMENT_TYPE && i_source)
        {
            return m_queue.start_reentrant_push(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)
        {
            return m_queue.template start_reentrant_emplace<ELEMENT_TYPE>(
              std::forward<ELEMENT_TYPE>(i_construction_params)...);
        }

        typename std::conditional<std::is_void<RET_VAL>::value, bool, optional<RET_VAL>>::type
          try_consume(PARAMS... i_params)
        {
            return try_consume_impl(std::is_void<RET_VAL>(), std::forward<PARAMS>(i_params)...);
        }

        typename std::conditional<std::is_void<RET_VAL>::value, bool, optional<RET_VAL>>::type
          try_consume(consume_operation & i_consume, PARAMS... i_params)
        {
            return try_consume_impl_cached(
              std::is_void<RET_VAL>(), i_consume, std::forward<PARAMS>(i_params)...);
        }

        typename std::conditional<std::is_void<RET_VAL>::value, bool, optional<RET_VAL>>::type
          try_reentrant_consume(PARAMS... i_params)
        {
            return try_reentrant_consume_impl(
              std::is_void<RET_VAL>(), std::forward<PARAMS>(i_params)...);
        }

        typename std::conditional<std::is_void<RET_VAL>::value, bool, optional<RET_VAL>>::type
          try_reentrant_consume(reentrant_consume_operation & i_consume, PARAMS... i_params)
        {
            return try_reentrant_consume_impl_cached(
              std::is_void<RET_VAL>(), i_consume, std::forward<PARAMS>(i_params)...);
        }

        template <
          function_type_erasure ERASURE_                                     = ERASURE,
          typename std::enable_if<ERASURE_ != function_manual_clear>::type * = nullptr>
        void clear() noexcept
        {
            auto erasure = ERASURE;
            if (erasure == function_manual_clear)
            {
                DENSITY_ASSUME(false);
            }
            else
            {
                m_queue.clear();
            }
        }

        bool empty() noexcept { return m_queue.empty(); }

      private:
        optional<RET_VAL> try_consume_impl(std::false_type, PARAMS... i_params)
        {
            if (auto cons = m_queue.try_start_consume())
            {
                auto && result = cons.complete_type().align_invoke_destroy(
                  cons.unaligned_element_ptr(), std::forward<PARAMS>(i_params)...);
                cons.commit_nodestroy();
                return optional<RET_VAL>(std::move(result));
            }
            else
            {
                return optional<RET_VAL>();
            }
        }

        bool try_consume_impl(std::true_type, PARAMS... i_params)
        {
            if (auto cons = m_queue.try_start_consume())
            {
                cons.complete_type().align_invoke_destroy(
                  cons.unaligned_element_ptr(), std::forward<PARAMS>(i_params)...);
                cons.commit_nodestroy();
                return true;
            }
            else
            {
                return false;
            }
        }

        optional<RET_VAL> try_consume_impl_cached(
          std::false_type, consume_operation & i_consume, PARAMS... i_params)
        {
            if (m_queue.try_start_consume(i_consume))
            {
                auto && result = i_consume.complete_type().align_invoke_destroy(
                  i_consume.unaligned_element_ptr(), std::forward<PARAMS>(i_params)...);
                i_consume.commit_nodestroy();
                return optional<RET_VAL>(std::move(result));
            }
            else
            {
                return optional<RET_VAL>();
            }
        }

        bool
          try_consume_impl_cached(std::true_type, consume_operation & i_consume, PARAMS... i_params)
        {
            if (m_queue.try_start_consume(i_consume))
            {
                i_consume.complete_type().align_invoke_destroy(
                  i_consume.unaligned_element_ptr(), std::forward<PARAMS>(i_params)...);
                i_consume.commit_nodestroy();
                return true;
            }
            else
            {
                return false;
            }
        }

        optional<RET_VAL> try_reentrant_consume_impl(std::false_type, PARAMS... i_params)
        {
            if (auto cons = m_queue.try_start_reentrant_consume())
            {
                auto && result = cons.complete_type().align_invoke_destroy(
                  cons.unaligned_element_ptr(), std::forward<PARAMS>(i_params)...);
                cons.commit_nodestroy();
                return optional<RET_VAL>(std::move(result));
            }
            else
            {
                return optional<RET_VAL>();
            }
        }

        bool try_reentrant_consume_impl(std::true_type, PARAMS... i_params)
        {
            if (auto cons = m_queue.try_start_reentrant_consume())
            {
                cons.complete_type().align_invoke_destroy(
                  cons.unaligned_element_ptr(), std::forward<PARAMS>(i_params)...);
                cons.commit_nodestroy();
                return true;
            }
            else
            {
                return false;
            }
        }

        optional<RET_VAL> try_reentrant_consume_impl_cached(
          std::false_type, reentrant_consume_operation & i_consume, PARAMS... i_params)
        {
            if (m_queue.try_start_reentrant_consume(i_consume))
            {
                auto && result = i_consume.complete_type().align_invoke_destroy(
                  i_consume.unaligned_element_ptr(), std::forward<PARAMS>(i_params)...);
                i_consume.commit_nodestroy();
                return optional<RET_VAL>(std::move(result));
            }
            else
            {
                return optional<RET_VAL>();
            }
        }

        bool try_reentrant_consume_impl_cached(
          std::true_type, reentrant_consume_operation & i_consume, PARAMS... i_params)
        {
            if (m_queue.try_start_reentrant_consume(i_consume))
            {
                i_consume.complete_type().align_invoke_destroy(
                  i_consume.unaligned_element_ptr(), std::forward<PARAMS>(i_params)...);
                i_consume.commit_nodestroy();
                return true;
            }
            else
            {
                return false;
            }
        }
    };

} // namespace density