Loading...
Searching...
No Matches
async_input_buffer.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 <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
adaptor_from_functor.hpp
Provides seqan3::detail::adaptor_from_functor.
std::addressof
T addressof(T... args)
std::forward
T forward(T... args)
seqan3::views::async_input_buffer
constexpr auto async_input_buffer
A view adapter that returns a concurrent-queue-like view over the underlying range.
Definition async_input_buffer.hpp:478
iterator
memory
seqan3::views
The SeqAn namespace for views.
Definition char_strictly_to.hpp:19
std::rel_ops::operator!=
T operator!=(T... args)
thread
Hide me