Loading...
Searching...
No Matches
execution_handler_parallel.hpp
Go to the documentation of this file.
// SPDX-FileCopyrightText: 2006-2025 Knut Reinert & Freie Universität Berlin
// SPDX-FileCopyrightText: 2016-2025 Knut Reinert & MPI für molekulare Genetik
// SPDX-License-Identifier: BSD-3-Clause
#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 &&) = delete;
internal_state & operator=(internal_state const &) = delete;
internal_state & operator=(internal_state &&) = delete;
~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
basic.hpp
Provides various type traits on generic types.
std::forward
T forward(T... args)
functional
std::is_lvalue_reference_v
T is_lvalue_reference_v
std::make_unique
T make_unique(T... args)
std::experimental::filesystem::status
T status(T... args)
thread
type_traits
vector
Hide me