20#include <seqan3/utility/parallel/detail/reader_writer_manager.hpp>
23namespace seqan3::detail
51class execution_handler_parallel
70 execution_handler_parallel(
size_t const thread_count) : state{
std::
make_unique<internal_state>()}
72 auto * q = &(state->queue);
73 for (
size_t i = 0; i < thread_count; ++i)
75 state->thread_pool.emplace_back(
81 if (q->wait_pop(task) == contrib::queue_op_status::closed)
104 execution_handler_parallel() : execution_handler_parallel{1u}
107 execution_handler_parallel(execution_handler_parallel
const &) =
delete;
108 execution_handler_parallel(execution_handler_parallel &&) =
default;
109 execution_handler_parallel & operator=(execution_handler_parallel
const &) =
delete;
110 execution_handler_parallel & operator=(execution_handler_parallel &&) =
default;
111 ~execution_handler_parallel() =
default;
133 template <std::copy_constructible algorithm_t,
typename algorithm_input_t, std::copy_constructible callback_t>
134 requires std::invocable<algorithm_t, algorithm_input_t, callback_t>
135 && (std::is_lvalue_reference_v<algorithm_input_t> || std::move_constructible<algorithm_input_t>)
136 void execute(algorithm_t && algorithm, algorithm_input_t && input, callback_t && callback)
138 assert(state !=
nullptr);
156 using forward_input_t = std::tuple_element_t<0,
decltype(input_tpl)>;
157 algorithm(std::forward<forward_input_t>(std::get<0>(input_tpl)), std::move(callback));
160 [[maybe_unused]] contrib::queue_op_status
status = state->queue.wait_push(std::move(task));
161 assert(status == contrib::queue_op_status::success);
180 template <std::copy_constructible algorithm_t,
181 std::ranges::input_range algorithm_input_range_t,
182 std::copy_constructible callback_t>
183 requires std::invocable<algorithm_t, std::ranges::range_reference_t<algorithm_input_range_t>, callback_t>
184 void bulk_execute(algorithm_t && algorithm, algorithm_input_range_t && input_range, callback_t && callback)
186 for (
auto && input : input_range)
187 execute(algorithm,
std::
forward<decltype(input)>(input), callback);
195 assert(state !=
nullptr);
197 state->stop_and_wait();
216 internal_state() =
default;
217 internal_state(internal_state
const &) =
delete;
218 internal_state(internal_state &&) =
delete;
219 internal_state & operator=(internal_state
const &) =
delete;
220 internal_state & operator=(internal_state &&) =
delete;
241 for (
auto & t : thread_pool)
251 contrib::fixed_buffer_queue<task_type> queue{10000};
Provides various type traits on generic types.
Provides seqan3::buffer_queue.
SeqAn specific customisations in the standard namespace.