Loading...
Searching...
No Matches
algorithm_executor_blocking.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 <functional>
#include <optional>
#include <ranges>
#include <type_traits>
#include <seqan3/core/algorithm/detail/execution_handler_parallel.hpp>
#include <seqan3/core/algorithm/detail/execution_handler_sequential.hpp>
namespace seqan3::detail
{
template <std::ranges::viewable_range resource_t,
std::semiregular algorithm_t,
std::semiregular algorithm_result_t,
typename execution_handler_t = execution_handler_sequential>
requires std::ranges::forward_range<resource_t>
&& std::invocable<algorithm_t,
std::ranges::range_reference_t<resource_t>,
std::function<void(algorithm_result_t)>>
class algorithm_executor_blocking
{
private:
using resource_type = std::views::all_t<resource_t>;
using resource_iterator_type = std::ranges::iterator_t<resource_type>;
using resource_difference_type = std::iter_difference_t<resource_iterator_type>;
using bucket_type = std::vector<algorithm_result_t>;
using bucket_iterator_type = std::ranges::iterator_t<bucket_type>;
using buffer_type = std::vector<bucket_type>;
using buffer_iterator_type = std::ranges::iterator_t<buffer_type>;
enum fill_status
{
non_empty_buffer,
empty_buffer,
end_of_resource
};
public:
algorithm_executor_blocking() = delete;
algorithm_executor_blocking(algorithm_executor_blocking const &) = delete;
algorithm_executor_blocking(algorithm_executor_blocking && other) noexcept :
algorithm_executor_blocking{std::move(other), other.resource_position()}
{}
algorithm_executor_blocking & operator=(algorithm_executor_blocking const &) = delete;
algorithm_executor_blocking & operator=(algorithm_executor_blocking && other)
{
auto old_resource_position = other.resource_position();
resource = std::move(other.resource);
move_initialise(std::move(other), old_resource_position);
return *this;
}
~algorithm_executor_blocking() = default;
algorithm_executor_blocking(resource_t resource,
algorithm_t algorithm,
algorithm_result_t const SEQAN3_DOXYGEN_ONLY(result) = algorithm_result_t{},
execution_handler_t && exec_handler = execution_handler_t{}) :
exec_handler{std::move(exec_handler)},
resource{std::forward<resource_t>(resource)},
resource_it{std::ranges::begin(this->resource)},
algorithm{std::move(algorithm)}
{
if constexpr (std::same_as<execution_handler_t, execution_handler_parallel>)
buffer_size = static_cast<size_t>(std::ranges::distance(this->resource));
buffer.resize(buffer_size);
buffer_it = buffer.end();
buffer_end_it = buffer_it;
}
std::optional<algorithm_result_t> next_result()
{
fill_status status;
// Each invocation of the algorithm might produce zero results (e.g. a search might not find a query)
// this repeats the algorithm until it produces the first result or the input resource was consumed.
do
{
status = fill_buffer();
}
while (status == fill_status::empty_buffer);
std::optional<algorithm_result_t> result{std::nullopt};
if (status == fill_status::end_of_resource)
return result;
assert(status == fill_status::non_empty_buffer);
assert(bucket_it != buffer_it->end());
result = std::ranges::iter_move(bucket_it);
go_to_next_result(); // Go to next buffered result.
return result;
}
bool is_eof() noexcept
{
return resource_it == std::ranges::end(resource);
}
private:
algorithm_executor_blocking(algorithm_executor_blocking && other,
resource_difference_type old_resource_position) noexcept :
resource{std::move(other.resource)}
{
move_initialise(std::move(other), old_resource_position);
}
resource_difference_type resource_position()
{
// Get the old resource position.
auto position = std::ranges::distance(std::ranges::begin(resource), resource_it);
assert(position >= 0);
return position;
}
fill_status fill_buffer()
{
if (!is_buffer_empty()) // Not everything consumed yet.
return fill_status::non_empty_buffer;
if (is_eof()) // Case: reached end of resource.
return fill_status::end_of_resource;
// Reset the buckets and the buffer iterator.
reset_buffer();
// Execute the algorithm (possibly asynchronous) and fill the buckets in this pre-assigned order.
for (buffer_end_it = buffer_it; buffer_end_it != buffer.end() && !is_eof(); ++buffer_end_it, ++resource_it)
{
exec_handler.execute(algorithm,
*resource_it,
[target_buffer_it = buffer_end_it](auto && algorithm_result)
{
target_buffer_it->push_back(std::move(algorithm_result));
});
}
exec_handler.wait();
// Move the results iterator to the next available result. (This skips empty results of the algorithm)
find_next_non_empty_bucket();
if (is_buffer_empty())
return fill_status::empty_buffer;
return fill_status::non_empty_buffer;
}
bool is_buffer_empty() const
{
return buffer_it == buffer_end_it;
}
void reset_buffer()
{
// Clear all buckets
for (auto & bucket : buffer)
bucket.clear();
// Reset the iterator over the buckets.
buffer_it = buffer.begin();
}
void find_next_non_empty_bucket()
{
assert(buffer_it <= buffer_end_it);
// find first buffered bucket that contains at least one element
buffer_it = std::find_if(buffer_it,
buffer_end_it,
[](auto const & buffer)
{
return !buffer.empty();
});
if (buffer_it != buffer_end_it)
bucket_it = buffer_it->begin();
}
void go_to_next_result()
{
if (++bucket_it == buffer_it->end())
{
++buffer_it;
find_next_non_empty_bucket();
}
}
void move_initialise(algorithm_executor_blocking && other, resource_difference_type old_resource_position) noexcept
{
algorithm = std::move(other.algorithm);
buffer_size = std::move(other.buffer_size);
exec_handler = std::move(other.exec_handler);
// Move the resource and set the iterator state accordingly.
resource_it = std::ranges::next(std::ranges::begin(resource), old_resource_position);
// Get the old buffer and bucket iterator positions.
auto buffer_it_position = other.buffer_it - other.buffer.begin();
auto buffer_end_it_position = other.buffer_end_it - other.buffer.begin();
std::ptrdiff_t bucket_it_position = 0;
if (buffer_it_position != buffer_end_it_position)
bucket_it_position = other.bucket_it - other.buffer_it->begin();
// Move the buffer and set the buffer and bucket iterator accordingly.
buffer = std::move(other.buffer);
buffer_it = buffer.begin() + buffer_it_position;
buffer_end_it = buffer.begin() + buffer_end_it_position;
if (buffer_it_position != buffer_end_it_position)
bucket_it = buffer_it->begin() + bucket_it_position;
}
execution_handler_t exec_handler{};
resource_type resource; // a std::ranges::view
resource_iterator_type resource_it{};
algorithm_t algorithm{};
buffer_type buffer{};
buffer_iterator_type buffer_it{};
buffer_iterator_type buffer_end_it{};
bucket_iterator_type bucket_it{};
size_t buffer_size{1};
};
template <typename resource_rng_t, std::semiregular algorithm_t, std::semiregular algorithm_result_t>
algorithm_executor_blocking(resource_rng_t &&, algorithm_t, algorithm_result_t const &)
-> algorithm_executor_blocking<resource_rng_t, algorithm_t, algorithm_result_t, execution_handler_sequential>;
} // namespace seqan3::detail
std::begin
T begin(T... args)
execution_handler_parallel.hpp
Provides seqan3::detail::execution_handler_parallel.
execution_handler_sequential.hpp
Provides seqan3::detail::execution_handler_sequential.
std::find_if
T find_if(T... args)
std::forward
T forward(T... args)
functional
seqan3::is_eof
constexpr auto is_eof
Checks whether a given letter is equal to the EOF constant defined in <cstdio>.
Definition predicate.hpp:72
optional
std::experimental::filesystem::status
T status(T... args)
type_traits
Hide me