execution_handler_parallel.hpp

Go to the documentation of this file.

// -----------------------------------------------------------------------------------------------------
// Copyright (c) 2006-2022, Knut Reinert & Freie Universität Berlin
// Copyright (c) 2016-2022, Knut Reinert & MPI für molekulare Genetik
// This file may be used, modified and/or redistributed under the terms of the 3-clause BSD-License
// shipped with this file and also available at: https://github.com/seqan/seqan3/blob/master/LICENSE.md
// -----------------------------------------------------------------------------------------------------
 
#pragma once
 
#include <concepts>
#include <functional>
#include <ranges>
#include <thread>
#include <type_traits>
#include <vector>
 
#include <seqan3/contrib/parallel/buffer_queue.hpp>
#include <seqan3/utility/parallel/detail/reader_writer_manager.hpp>
#include <seqan3/utility/type_traits/basic.hpp>
 
namespace seqan3::detail
{
 
class execution_handler_parallel
{
private:
    using task_type = std::function<void()>;
 
public:
    execution_handler_parallel(size_t const thread_count) : state{std::make_unique<internal_state>()}
    {
        auto * q = &(state->queue);
        for (size_t i = 0; i < thread_count; ++i)
        {
            state->thread_pool.emplace_back(
                [q]()
                {
                    for (;;)
                    {
                        task_type task;
                        if (q->wait_pop(task) == contrib::queue_op_status::closed)
                            return;
 
                        task();
                    }
                });
        }
    }
 
    execution_handler_parallel() : execution_handler_parallel{1u}
    {}
 
    execution_handler_parallel(execution_handler_parallel const &) = delete;             
    execution_handler_parallel(execution_handler_parallel &&) = default;                 
    execution_handler_parallel & operator=(execution_handler_parallel const &) = delete; 
    execution_handler_parallel & operator=(execution_handler_parallel &&) = default;     
    ~execution_handler_parallel() = default;                                             
 
 
    template <std::copy_constructible algorithm_t, typename algorithm_input_t, std::copy_constructible callback_t>
        requires std::invocable<algorithm_t, algorithm_input_t, callback_t>
              && (std::is_lvalue_reference_v<algorithm_input_t> || std::move_constructible<algorithm_input_t>)
    void execute(algorithm_t && algorithm, algorithm_input_t && input, callback_t && callback)
    {
        assert(state != nullptr);
 
        // Note: Unfortunately, we can't use std::forward_as_tuple here because a std::function object (`task_type`)
        // cannot be constructed if the tuple element type is a rvalue-reference.
        // So we capture the input as a `tuple<algorithm_input_t>` which either is a lvalue reference or has no
        // reference type according to the reference collapsing rules of forwarding references.
        // Then we forward the input into the tuple which either just stores the reference or the input is moved into
        // the tuple. When the task is executed by some thread the stored input will either be forwarded as a
        // lvalue-reference to the algorithm or the input is moved into the algorithm from the tuple. This is valid
        // since the task is executed only once by the parallel execution handler.
        // Here is a discussion about the problem on stackoverflow:
        // https://stackoverflow.com/questions/26831382/capturing-perfectly-forwarded-variable-in-lambda/
 
        // Asynchronously pushes the algorithm job as a task to the queue.
        // Note: that lambda is mutable, s.t. we can move out the content of input_tpl
        task_type task =
            [=, input_tpl = std::tuple<algorithm_input_t>{std::forward<algorithm_input_t>(input)}]() mutable
        {
            using forward_input_t = std::tuple_element_t<0, decltype(input_tpl)>;
            algorithm(std::forward<forward_input_t>(std::get<0>(input_tpl)), std::move(callback));
        };
 
        [[maybe_unused]] contrib::queue_op_status status = state->queue.wait_push(std::move(task));
        assert(status == contrib::queue_op_status::success);
    }
 
    template <std::copy_constructible algorithm_t,
              std::ranges::input_range algorithm_input_range_t,
              std::copy_constructible callback_t>
        requires std::invocable<algorithm_t, std::ranges::range_reference_t<algorithm_input_range_t>, callback_t>
    void bulk_execute(algorithm_t && algorithm, algorithm_input_range_t && input_range, callback_t && callback)
    {
        for (auto && input : input_range)
            execute(algorithm, std::forward<decltype(input)>(input), callback);
 
        wait();
    }
 
    void wait()
    {
        assert(state != nullptr);
 
        state->stop_and_wait();
    }
 
private:
    class internal_state
    {
    public:
        internal_state() = default;                                  
        internal_state(internal_state const &) = delete;             
        internal_state(internal_state &&) = default;                 
        internal_state & operator=(internal_state const &) = delete; 
        internal_state & operator=(internal_state &&) = default;     
 
        ~internal_state()
        {
            stop_and_wait();
        }
 
        void stop_and_wait()
        {
            queue.close();
 
            for (auto & t : thread_pool)
            {
                if (t.joinable())
                    t.join();
            }
        }
 
        std::vector<std::thread> thread_pool{};
        contrib::fixed_buffer_queue<task_type> queue{10000};
    };
 
    std::unique_ptr<internal_state> state{nullptr};
};
 
} // namespace seqan3::detail