alignment_executor_two_way.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 <optional>
#include <type_traits>
 
#include <seqan3/alignment/pairwise/alignment_range.hpp>
#include <seqan3/alignment/pairwise/detail/type_traits.hpp>
#include <seqan3/alignment/pairwise/execution/execution_handler_parallel.hpp>
#include <seqan3/alignment/pairwise/execution/execution_handler_sequential.hpp>
#include <seqan3/core/parallel/execution.hpp>
#include <seqan3/core/type_traits/range.hpp>
#include <seqan3/range/shortcuts.hpp>
#include <seqan3/range/views/chunk.hpp>
#include <seqan3/range/views/type_reduce.hpp>
#include <seqan3/range/views/zip.hpp>
#include <seqan3/std/ranges>
 
namespace seqan3::detail
{
 
template <std::ranges::viewable_range resource_t,
          typename alignment_algorithm_t,
          typename execution_handler_t = execution_handler_sequential>
    requires std::ranges::forward_range<resource_t> &&
             std::copy_constructible<alignment_algorithm_t>
class alignment_executor_two_way
{
private:
    using chunked_resource_type = typename chunked_indexed_sequence_pairs<resource_t>::type;
    using chunked_resource_iterator = std::ranges::iterator_t<chunked_resource_type>;
 
    using alignment_result_type = typename alignment_algorithm_t::result_type;
    using buffer_value_type = std::ranges::range_value_t<alignment_result_type>;
    using buffer_type       = std::vector<buffer_value_type>;
    using buffer_pointer    = std::ranges::iterator_t<buffer_type>;
 
public:
 
    using value_type      = buffer_value_type;
    using reference       = std::add_lvalue_reference_t<value_type>;
    using difference_type = typename buffer_type::difference_type;
 
    alignment_executor_two_way() = delete;
    alignment_executor_two_way(alignment_executor_two_way const &) = delete;
 
    alignment_executor_two_way(alignment_executor_two_way && other) noexcept
    {
        move_initialise(std::move(other));
    }
 
    alignment_executor_two_way & operator=(alignment_executor_two_way const &) = delete;
 
    alignment_executor_two_way & operator=(alignment_executor_two_way && other)
    {
        move_initialise(std::move(other));
        return *this;
    }
 
    ~alignment_executor_two_way() = default;
 
    template <typename exec_policy_t = sequenced_policy>
        requires is_execution_policy_v<exec_policy_t>
    alignment_executor_two_way(resource_t resrc,
                               alignment_algorithm_t fn,
                               size_t chunk_size = 1u,
                               exec_policy_t const & SEQAN3_DOXYGEN_ONLY(exec) = seq) :
        kernel{std::move(fn)},
        _chunk_size{chunk_size}
    {
 
        if (chunk_size == 0u)
            throw std::invalid_argument{"The chunk size must be greater than 0."};
 
        chunked_resource = views::zip(std::forward<resource_t>(resrc), std::views::iota(0)) | views::chunk(_chunk_size);
        chunked_resource_it = chunked_resource.begin();
 
        if constexpr (std::same_as<execution_handler_t, execution_handler_parallel>)
            init_buffer(std::ranges::distance(resrc));
        else
            init_buffer(_chunk_size);
    }
 
 
    std::optional<value_type> bump()
    {
        if (underflow() == eof)
            return {std::nullopt};
 
        assert(gptr < egptr);
        return {std::move(*gptr++)};
    }
 
    constexpr size_t in_avail() const noexcept
    {
        return egptr - gptr;
    }
 
    bool is_eof() noexcept
    {
        return chunked_resource_it == std::ranges::end(chunked_resource);
    }
 
    constexpr size_t chunk_size() const noexcept
    {
        return _chunk_size;
    }
 
private:
 
    void setg(buffer_pointer beg, buffer_pointer end)
    {
        gptr = beg;
        egptr = end;
    }
 
    size_t underflow()
    {
        if (gptr < egptr)  // Case: buffer not completely consumed
            return in_avail();
 
        if (is_eof())  // Case: reached end of resource.
            return eof;
 
        // Reset the get pointer.
        setg(std::ranges::begin(buffer), std::ranges::end(buffer));
 
        using std::get;
 
        // Apply the alignment execution.
        size_t count = 0;
        size_t buffer_limit = in_avail();
 
        for (size_t current_chunk_size = std::ranges::distance(*chunked_resource_it);
             count < buffer_limit && !is_eof();
             count += current_chunk_size, gptr += current_chunk_size, ++chunked_resource_it)
        {
            auto && current_chunk = std::ranges::iter_move(chunked_resource_it);
            current_chunk_size = std::ranges::distance(current_chunk);
            assert(in_avail() >= current_chunk_size);
 
            exec_handler.execute(kernel, std::move(current_chunk), [write_to = gptr] (auto && alignment_results)
            {
                std::ranges::move(alignment_results, write_to);
            });
        }
 
        exec_handler.wait();
 
        // Update the available get positions to cover the part of the buffer that was filled.
        setg(std::ranges::begin(buffer), std::ranges::begin(buffer) + count);
 
        return in_avail();
    }
 
    void init_buffer(size_t const size)
    {
        buffer.resize(size);
        setg(std::ranges::end(buffer), std::ranges::end(buffer));
    }
 
    void move_initialise(alignment_executor_two_way && other) noexcept
    {
        kernel = std::move(other.kernel);
        _chunk_size = std::move(other._chunk_size);
        // Get the old resource position.
        std::ptrdiff_t old_resource_pos = std::ranges::distance(other.chunked_resource.begin(),
                                                                other.chunked_resource_it);
        // Move the resource and set the iterator state accordingly.
        chunked_resource = std::move(other.chunked_resource);
        chunked_resource_it = std::ranges::next(chunked_resource.begin(), old_resource_pos);
 
        // Get the old get pointer positions.
        std::ptrdiff_t old_gptr_pos = other.gptr - buffer.begin();
        std::ptrdiff_t old_egptr_pos = other.egptr - buffer.begin();
        // Move the buffer and set the get pointer accordingly.
        buffer = std::move(other.buffer);
        setg(buffer.begin() + old_gptr_pos, buffer.begin() + old_egptr_pos);
    }
 
    static constexpr size_t eof{std::numeric_limits<size_t>::max()};
 
    execution_handler_t exec_handler{};
 
    chunked_resource_type chunked_resource{};
    chunked_resource_iterator chunked_resource_it{};
    alignment_algorithm_t kernel{};
 
    buffer_type buffer{};
    buffer_pointer gptr{};
    buffer_pointer egptr{};
    size_t _chunk_size{};
};
 
template <typename resource_rng_t, typename func_t>
alignment_executor_two_way(resource_rng_t &&, func_t) ->
    alignment_executor_two_way<resource_rng_t, func_t, execution_handler_sequential>;
 
template <typename resource_rng_t, typename func_t, typename exec_policy_t>
    requires is_execution_policy_v<exec_policy_t>
alignment_executor_two_way(resource_rng_t &&, func_t, size_t, exec_policy_t const &) ->
    alignment_executor_two_way<resource_rng_t,
                               func_t,
                               std::conditional_t<std::same_as<exec_policy_t, parallel_policy>,
                                                  execution_handler_parallel,
                                                  execution_handler_sequential>>;
 
} // namespace seqan3::detail