algorithm_executor_blocking.hpp

Go to the documentation of this file.

// -----------------------------------------------------------------------------------------------------
// Copyright (c) 2006-2021, Knut Reinert & Freie Universität Berlin
// Copyright (c) 2016-2021, 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 <optional>
#include <seqan3/std/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);
 
        if (status == fill_status::end_of_resource)
            return {std::nullopt};
 
        assert(status == fill_status::non_empty_buffer);
        assert(bucket_it != buffer_it->end());
 
        std::optional<algorithm_result_t> 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