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

namespace density
{
    template <
      typename CALLABLE,
      typename ALLOCATOR_TYPE                      = default_allocator,
      function_type_erasure   ERASURE              = function_standard_erasure,
      concurrency_cardinality PROD_CARDINALITY     = concurrency_multiple,
      concurrency_cardinality CONSUMER_CARDINALITY = concurrency_multiple,
      typename BUSY_WAIT_FUNC                      = default_busy_wait>
    class sp_function_queue;

#ifndef DOXYGEN_DOC_GENERATION
    template <
      typename RET_VAL,
      typename... PARAMS,
      typename ALLOCATOR_TYPE,
      function_type_erasure   ERASURE,
      concurrency_cardinality PROD_CARDINALITY,
      concurrency_cardinality CONSUMER_CARDINALITY,
      typename BUSY_WAIT_FUNC>
    class sp_function_queue<
      RET_VAL(PARAMS...),
      ALLOCATOR_TYPE,
      ERASURE,
      PROD_CARDINALITY,
      CONSUMER_CARDINALITY,
      BUSY_WAIT_FUNC>
#else
    template <
      typename CALLABLE,
      typename ALLOCATOR_TYPE                      = default_allocator,
      function_type_erasure   ERASURE              = function_standard_erasure,
      concurrency_cardinality PROD_CARDINALITY     = concurrency_multiple,
      concurrency_cardinality CONSUMER_CARDINALITY = concurrency_multiple,
      typename BUSY_WAIT_FUNC                      = default_busy_wait>
    class sp_function_queue
#endif
    {
      private:
        using UnderlyingQueue = sp_heter_queue<
          detail::FunctionRuntimeType<ERASURE, RET_VAL(PARAMS...)>,
          ALLOCATOR_TYPE,
          PROD_CARDINALITY,
          CONSUMER_CARDINALITY,
          BUSY_WAIT_FUNC>;
        UnderlyingQueue m_queue;

      public:
        static constexpr bool concurrent_puts = PROD_CARDINALITY == concurrency_multiple;

        static constexpr bool concurrent_consumes = CONSUMER_CARDINALITY == concurrency_multiple;

        static constexpr bool concurrent_put_consumes = true;

        static constexpr bool is_seq_cst = true;

        sp_function_queue() noexcept = default;

        sp_function_queue(sp_function_queue && i_source) noexcept = default;

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

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

        ~sp_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<CONSTRUCTION_PARAMS>(i_construction_params)...);
        }

        template <typename ELEMENT_COMPLETE_TYPE>
        bool try_push(
          progress_guarantee i_progress_guarantee,
          ELEMENT_COMPLETE_TYPE &&
            i_source) noexcept(noexcept(m_queue
                                          .try_push(
                                            i_progress_guarantee,
                                            std::forward<ELEMENT_COMPLETE_TYPE>(i_source))))
        {
            return m_queue.try_push(
              i_progress_guarantee, std::forward<ELEMENT_COMPLETE_TYPE>(i_source));
        }

        template <typename ELEMENT_COMPLETE_TYPE, typename... CONSTRUCTION_PARAMS>
        bool
          try_emplace(progress_guarantee i_progress_guarantee, CONSTRUCTION_PARAMS &&... i_construction_params) noexcept(
            noexcept(m_queue.template try_emplace<ELEMENT_COMPLETE_TYPE>(
              i_progress_guarantee, std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...)))
        {
            return m_queue.template try_emplace<ELEMENT_COMPLETE_TYPE>(
              i_progress_guarantee, std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...);
        }

        template <typename ELEMENT_TYPE>
        put_transaction<typename std::decay<ELEMENT_TYPE>::type> try_start_push(
          progress_guarantee i_progress_guarantee,
          ELEMENT_TYPE &&    i_source) noexcept(noexcept(m_queue
                                                        .try_start_push(
                                                          i_progress_guarantee,
                                                          std::forward<ELEMENT_TYPE>(i_source))))
        {
            return m_queue.try_start_push(
              i_progress_guarantee, std::forward<ELEMENT_TYPE>(i_source));
        }


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

        template <typename ELEMENT_COMPLETE_TYPE>
        bool try_reentrant_push(
          progress_guarantee i_progress_guarantee,
          ELEMENT_COMPLETE_TYPE &&
            i_source) noexcept(noexcept(m_queue
                                          .try_reentrant_push(
                                            i_progress_guarantee,
                                            std::forward<ELEMENT_COMPLETE_TYPE>(i_source))))
        {
            return m_queue.try_reentrant_push(
              i_progress_guarantee, std::forward<ELEMENT_COMPLETE_TYPE>(i_source));
        }

        template <typename ELEMENT_COMPLETE_TYPE, typename... CONSTRUCTION_PARAMS>
        bool try_reentrant_emplace(
          progress_guarantee i_progress_guarantee,
          CONSTRUCTION_PARAMS &&... i_construction_params) noexcept(noexcept(m_queue
                                                                               .template try_reentrant_emplace<
                                                                                 ELEMENT_COMPLETE_TYPE>(
                                                                                 i_progress_guarantee,
                                                                                 std::forward<
                                                                                   CONSTRUCTION_PARAMS>(
                                                                                   i_construction_params)...)))
        {
            return m_queue.template try_reentrant_emplace<ELEMENT_COMPLETE_TYPE>(
              i_progress_guarantee, std::forward<CONSTRUCTION_PARAMS>(i_construction_params)...);
        }

        template <typename ELEMENT_TYPE>
        reentrant_put_transaction<typename std::decay<ELEMENT_TYPE>::type> try_start_reentrant_push(
          progress_guarantee i_progress_guarantee,
          ELEMENT_TYPE &&    i_source) noexcept(noexcept(m_queue
                                                        .try_start_reentrant_push(
                                                          i_progress_guarantee,
                                                          std::forward<ELEMENT_TYPE>(i_source))))
        {
            return m_queue.try_start_reentrant_push(
              i_progress_guarantee, std::forward<ELEMENT_TYPE>(i_source));
        }


        template <typename ELEMENT_TYPE, typename... CONSTRUCTION_PARAMS>
        reentrant_put_transaction<ELEMENT_TYPE> try_start_reentrant_emplace(
          progress_guarantee i_progress_guarantee,
          CONSTRUCTION_PARAMS &&... i_construction_params) noexcept(noexcept(m_queue
                                                                               .template try_start_reentrant_emplace<
                                                                                 ELEMENT_TYPE>(
                                                                                 i_progress_guarantee,
                                                                                 std::forward<
                                                                                   ELEMENT_TYPE>(
                                                                                   i_construction_params)...)))
        {
            return m_queue.template try_start_reentrant_emplace<ELEMENT_TYPE>(
              i_progress_guarantee, std::forward<ELEMENT_TYPE>(i_construction_params)...);
        }

        typename std::conditional<std::is_void<RET_VAL>::value, bool, density::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, density::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