async_input_buffer.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 <thread>
 
#include <seqan3/core/type_traits/all.hpp>
#include <seqan3/contrib/parallel/buffer_queue.hpp>
#include <seqan3/range/shortcuts.hpp>
#include <seqan3/range/views/detail.hpp>
#include <seqan3/std/concepts>
#include <seqan3/std/iterator>
#include <seqan3/std/ranges>
 
//-----------------------------------------------------------------------------
// This is the path a value takes when using this views::
//   urange
// → async_input_buffer_view.buffer               [size n]
// → async_input_buffer_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 an std::ranges::InputRange.");
    static_assert(std::ranges::view<urng_t>,
        "The range parameter to async_input_buffer_view must model std::ranges::View.");
    static_assert(std::movable<value_type_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<value_type_t<urng_t>, std::remove_reference_t<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<value_type_t<urng_t>> buffer;
 
        std::thread producer;
    };
 
    std::shared_ptr<state> state_ptr = nullptr;
 
    class async_input_buffer_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<value_type_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<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, 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}
    {}
 
    async_input_buffer_iterator begin()
    {
        assert(state_ptr != nullptr);
        return {state_ptr->buffer};
    }
 
    async_input_buffer_iterator begin() const = delete;
 
    async_input_buffer_iterator cbegin() const = delete;
 
    std::ranges::default_sentinel_t end()
    {
        return std::ranges::default_sentinel;
    }
 
    std::ranges::default_sentinel_t end() const = delete;
 
    std::ranges::default_sentinel_t cend() const = delete;
};
 
template <typename urng_t>
class async_input_buffer_view<urng_t>::async_input_buffer_iterator
{
    using sentinel_type = std::ranges::default_sentinel_t;
 
    contrib::fixed_buffer_queue<value_type_t<urng_t>> * buffer_ptr = nullptr;
 
    mutable value_type_t<urng_t> cached_value;
 
    bool at_end = false;
 
public:
 
    using difference_type   = difference_type_t<urng_iterator_type>;
    using value_type        = value_type_t<urng_iterator_type>;
    using pointer           = value_type *;
    using reference         = value_type &;
    using iterator_category = void;
    using iterator_concept  = std::input_iterator_tag;
 
    async_input_buffer_iterator()                                                    = default; 
    //TODO: delete:
    async_input_buffer_iterator(async_input_buffer_iterator const & rhs)             = default; 
    async_input_buffer_iterator(async_input_buffer_iterator && rhs)                  = default; 
    //TODO: delete:
    async_input_buffer_iterator & operator=(async_input_buffer_iterator const & rhs) = default; 
    async_input_buffer_iterator & operator=(async_input_buffer_iterator && rhs)      = default; 
    ~async_input_buffer_iterator()                                          noexcept = default; 
 
    async_input_buffer_iterator(contrib::fixed_buffer_queue<value_type_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);
    }
 
    async_input_buffer_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==(async_input_buffer_iterator const & lhs,
                                     std::ranges::default_sentinel_t const &) noexcept
    {
        return lhs.at_end;
    }
 
    friend constexpr bool operator==(std::ranges::default_sentinel_t const &,
                                     async_input_buffer_iterator const & rhs) noexcept
    {
        return rhs == std::ranges::default_sentinel_t{};
    }
 
    friend constexpr bool operator!=(async_input_buffer_iterator const & lhs,
                                     std::ranges::default_sentinel_t const &) noexcept
    {
        return !(lhs == std::ranges::default_sentinel_t{});
    }
 
    friend constexpr bool operator!=(std::ranges::default_sentinel_t const &,
                                     async_input_buffer_iterator const & rhs) noexcept
    {
        return rhs != std::ranges::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::ranges::all_view<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 an std::ranges::InputRange.");
        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<value_type_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<value_type_t<urng_t>, std::remove_reference_t<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};
    }
};
 
}  // seqan3::detail
 
//-----------------------------------------------------------------------------
// View shortcut for functor.
//-----------------------------------------------------------------------------
 
namespace seqan3::views
{
inline constexpr auto async_input_buffer = detail::async_input_buffer_fn{};
 
} // namespace seqan3::views