async_input_buffer.hpp

Go to the documentation of this file.

// -----------------------------------------------------------------------------------------------------
// Copyright (c) 2006-2023, Knut Reinert & Freie Universität Berlin
// Copyright (c) 2016-2023, 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 <concepts>
#include <iterator>
#include <memory>
#include <ranges>
#include <thread>
 
#include <seqan3/contrib/parallel/buffer_queue.hpp>
#include <seqan3/core/range/detail/adaptor_from_functor.hpp>
 
//-----------------------------------------------------------------------------
// This is the path a value takes when using this views::
//   urange
// → async_input_buffer_view.buffer [size n]
// → iterator.cached_value [size 1]
// → user
//-----------------------------------------------------------------------------
 
namespace seqan3::detail
{
 
template <std::ranges::range urng_t>
class async_input_buffer_view : public std::ranges::view_interface<async_input_buffer_view<urng_t>>
{
private:
    static_assert(std::ranges::input_range<urng_t>,
                  "The range parameter to async_input_buffer_view must be at least a std::ranges::input_range.");
    static_assert(std::ranges::view<urng_t>,
                  "The range parameter to async_input_buffer_view must model std::ranges::view.");
    static_assert(std::movable<std::ranges::range_value_t<urng_t>>,
                  "The range parameter to async_input_buffer_view must have a value_type that is std::movable.");
    static_assert(
        std::constructible_from<std::ranges::range_value_t<urng_t>,
                                std::remove_reference_t<std::ranges::range_reference_t<urng_t>> &&>,
        "The range parameter to async_input_buffer_view must have a value_type that is constructible by a moved "
        "value of its reference type.");
 
    using urng_iterator_type = std::ranges::iterator_t<urng_t>;
 
    struct state
    {
        urng_t urange;
 
        contrib::fixed_buffer_queue<std::ranges::range_value_t<urng_t>> buffer;
 
        std::thread producer;
    };
 
    std::shared_ptr<state> state_ptr = nullptr;
 
    class iterator;
 
public:
    async_input_buffer_view() = default;                                            
    async_input_buffer_view(async_input_buffer_view const &) = default;             
    async_input_buffer_view(async_input_buffer_view &&) = default;                  
    async_input_buffer_view & operator=(async_input_buffer_view const &) = default; 
    async_input_buffer_view & operator=(async_input_buffer_view &&) = default;      
    ~async_input_buffer_view() = default;                                           
 
    async_input_buffer_view(urng_t _urng, size_t const buffer_size)
    {
        auto deleter = [](state * p)
        {
            if (p != nullptr)
            {
                p->buffer.close();
                p->producer.join();
                delete p;
            }
        };
 
        state_ptr = std::shared_ptr<state>(
            new state{std::move(_urng),
                      contrib::fixed_buffer_queue<std::ranges::range_value_t<urng_t>>{buffer_size},
                      std::thread{}}, // thread is set/started below, needs rest of state
            deleter);
 
        auto runner = [&state = *state_ptr]()
        {
            for (auto && val : state.urange)
                if (state.buffer.wait_push(std::move(val)) == contrib::queue_op_status::closed)
                    break;
 
            state.buffer.close();
        };
 
        state_ptr->producer = std::thread{runner};
    }
 
    template <typename other_urng_t>
        requires (!std::same_as<std::remove_cvref_t<other_urng_t>, async_input_buffer_view>)
              && // prevent recursive instantiation
                 std::ranges::viewable_range<other_urng_t>
              && std::constructible_from<urng_t, std::ranges::ref_view<std::remove_reference_t<other_urng_t>>>
    async_input_buffer_view(other_urng_t && _urng, size_t const buffer_size) :
        async_input_buffer_view{std::views::all(_urng), buffer_size}
    {}
 
    iterator begin()
    {
        assert(state_ptr != nullptr);
        return {state_ptr->buffer};
    }
 
    iterator begin() const = delete;
 
    std::default_sentinel_t end()
    {
        return std::default_sentinel;
    }
 
    std::default_sentinel_t end() const = delete;
};
 
template <std::ranges::range urng_t>
class async_input_buffer_view<urng_t>::iterator
{
    using sentinel_type = std::default_sentinel_t;
 
    contrib::fixed_buffer_queue<std::ranges::range_value_t<urng_t>> * buffer_ptr = nullptr;
 
    mutable std::ranges::range_value_t<urng_t> cached_value;
 
    bool at_end = false;
 
public:
    using difference_type = std::iter_difference_t<urng_iterator_type>;
    using value_type = std::iter_value_t<urng_iterator_type>;
    using pointer = detail::iter_pointer_t<urng_iterator_type>;
    using reference = std::iter_reference_t<urng_iterator_type>;
    using iterator_category = std::input_iterator_tag;
    using iterator_concept = iterator_category;
 
    iterator() = default; 
    //TODO: delete:
    iterator(iterator const & rhs) = default; 
    iterator(iterator && rhs) = default;      
    //TODO: delete:
    iterator & operator=(iterator const & rhs) = default; 
    iterator & operator=(iterator && rhs) = default;      
    ~iterator() noexcept = default;                       
 
    iterator(contrib::fixed_buffer_queue<std::ranges::range_value_t<urng_t>> & buffer) noexcept : buffer_ptr{&buffer}
    {
        ++(*this); // cache first value
    }
 
    reference operator*() const noexcept
    {
        return cached_value;
    }
 
    pointer operator->() const noexcept
    {
        return std::addressof(cached_value);
    }
 
    iterator & operator++() noexcept
    {
        if (at_end) // TODO unlikely
            return *this;
 
        assert(buffer_ptr != nullptr);
 
        if (buffer_ptr->wait_pop(cached_value) == contrib::queue_op_status::closed)
            at_end = true;
 
        return *this;
    }
 
    void operator++(int) noexcept
    {
        ++(*this);
    }
 
    friend constexpr bool operator==(iterator const & lhs, std::default_sentinel_t const &) noexcept
    {
        return lhs.at_end;
    }
 
    friend constexpr bool operator==(std::default_sentinel_t const &, iterator const & rhs) noexcept
    {
        return rhs == std::default_sentinel_t{};
    }
 
    friend constexpr bool operator!=(iterator const & lhs, std::default_sentinel_t const &) noexcept
    {
        return !(lhs == std::default_sentinel_t{});
    }
 
    friend constexpr bool operator!=(std::default_sentinel_t const &, iterator const & rhs) noexcept
    {
        return rhs != std::default_sentinel_t{};
    }
};
 
template <std::ranges::viewable_range urng_t>
async_input_buffer_view(urng_t &&, size_t const buffer_size) -> async_input_buffer_view<std::views::all_t<urng_t>>;
 
// ============================================================================
//  async_input_buffer_fn (adaptor definition
// ============================================================================
 
struct async_input_buffer_fn
{
    constexpr auto operator()(size_t const buffer_size) const
    {
        return detail::adaptor_from_functor{*this, buffer_size};
    }
 
    template <std::ranges::range urng_t>
    constexpr auto operator()(urng_t && urange, size_t const buffer_size) const
    {
        static_assert(std::ranges::input_range<urng_t>,
                      "The range parameter to views::async_input_buffer must be at least a std::ranges::input_range.");
        static_assert(std::ranges::viewable_range<urng_t>,
                      "The range parameter to views::async_input_buffer cannot be a temporary of a non-view range.");
        static_assert(std::movable<std::ranges::range_value_t<urng_t>>,
                      "The range parameter to views::async_input_buffer must have a value_type that is std::movable.");
        static_assert(
            std::constructible_from<std::ranges::range_value_t<urng_t>,
                                    std::remove_reference_t<std::ranges::range_reference_t<urng_t>> &&>,
            "The range parameter to views::async_input_buffer must have a value_type that is constructible by a moved "
            "value of its reference type.");
 
        if (buffer_size == 0)
            throw std::invalid_argument{"The buffer_size parameter to views::async_input_buffer must be > 0."};
 
        return detail::async_input_buffer_view{std::forward<urng_t>(urange), buffer_size};
    }
};
 
} // namespace seqan3::detail
 
//-----------------------------------------------------------------------------
// View shortcut for functor.
//-----------------------------------------------------------------------------
 
namespace seqan3::views
{
inline constexpr auto async_input_buffer = detail::async_input_buffer_fn{};
} // namespace seqan3::views