execution_handler_parallel.hpp

Go to the documentation of this file.

// -----------------------------------------------------------------------------------------------------
// Copyright (c) 2006-2020, Knut Reinert & Freie Universität Berlin
// Copyright (c) 2016-2020, 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 <functional>
#include <thread>
#include <type_traits>
#include <vector>
 
#include <seqan3/alignment/pairwise/detail/concept.hpp>
#include <seqan3/contrib/parallel/buffer_queue.hpp>
#include <seqan3/core/parallel/detail/reader_writer_manager.hpp>
#include <seqan3/core/platform.hpp>
#include <seqan3/std/concepts>
#include <seqan3/std/ranges>
 
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{std::thread::hardware_concurrency()}
    {}
 
    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()
    {
        if (state != nullptr)
            wait();
    }
 
    template <typename algorithm_t, indexed_sequence_pair_range indexed_sequence_pairs_t, typename delegate_type>
    void execute(algorithm_t && algorithm,
                 indexed_sequence_pairs_t indexed_sequence_pairs,
                 delegate_type && delegate)
    {
        assert(state != nullptr);
 
        // Asynchronously pushes the alignment job as a task to the queue.
        task_type task = [=, indexed_sequence_pairs = std::move(indexed_sequence_pairs)] ()
        {
            delegate(algorithm(std::move(indexed_sequence_pairs)));
        };
 
        [[maybe_unused]] contrib::queue_op_status status = state->queue.wait_push(std::move(task));
        assert(status == contrib::queue_op_status::success);
    }
 
    void wait()
    {
        assert(state != nullptr);
 
        if (!state->is_waiting)
        {
            state->is_waiting = true;
            state->queue.close();
 
            for (auto & t : state->thread_pool)
            {
                if (t.joinable())
                    t.join();
            }
        }
    }
 
private:
    struct internal_state
    {
        std::vector<std::thread>                 thread_pool{};
        contrib::fixed_buffer_queue<task_type>   queue{10000};
        bool                                     is_waiting{false};
    };
 
    std::unique_ptr<internal_state> state{nullptr};
};
 
} // namespace seqan3