cuddl::detail::fastx_load_pool Class Reference

#include <fastx_sequence_file.hpp>

Public Member Functions
	fastx_load_pool (std::span< std::filesystem::path const > paths, size_t workers, decompression_source source={}, size_t window=0, bool defer_fasta=false, bool completion_order=false, uint32_t pack_k=0)
	~fastx_load_pool ()
	fastx_load_pool (fastx_load_pool const &)=delete
fastx_load_pool &	operator= (fastx_load_pool const &)=delete
Result< std::unique_ptr< fastx_sequence_file > >	take (size_t id)
Result< std::pair< size_t, std::unique_ptr< fastx_sequence_file > > >	take_ready ()
Result< std::optional< std::pair< size_t, std::unique_ptr< fastx_sequence_file > > > >	take_ready_until_limit ()
	take_ready, or nothing once every id below the cap limit is taken.
size_t	ready ()
	Completion-order results waiting to be taken.
size_t	cap (size_t limit)
	Stops the loaders before id limit, or after the ids they already claimed.

Detailed Description

Definition at line 1796 of file fastx_sequence_file.hpp.

Constructor & Destructor Documentation

fastx_load_pool() [1/2]

cuddl::detail::fastx_load_pool::fastx_load_pool ( std::span< std::filesystem::path const >  paths, size_t  workers, decompression_source  source = {}, size_t  window = 0, bool  defer_fasta = false, bool  completion_order = false, uint32_t  pack_k = 0  )

inline

Parameters

paths	Files the pool loads in order.
workers	Loader threads.
source	Decompression buffer source; the default detects gzip by magic bytes.
window	Files the loaders may hold ahead of the consumer, at least one per worker. With ordered collection, the window hides a slow file: with a window of one worker-worth there is no slack, and the consumer waits on the straggler while the other loaders sit idle.
defer_fasta	Leave eligible FASTA bytes for a GPU normaliser.
completion_order	Collect through take_ready instead of take(id).
pack_k	Nonzero packs each loaded file into two-bit runs for k-mers this long.

Definition at line 1808 of file fastx_sequence_file.hpp.

                                      {},
        size_t window = 0,
        bool defer_fasta = false,
        bool completion_order = false,
        uint32_t pack_k = 0
    )
        : paths_(paths),
          workers_(std::max(size_t{1}, workers)),
          window_(std::max({size_t{1}, window, workers_})),
          source_(source),
          defer_fasta_(defer_fasta),
          pack_k_(pack_k),
          results_(paths.size()),
          errors_(paths.size()),
          ready_(completion_order ? window_ : 0),
          retired_(defer_fasta ? window_ : 0) {
        threads_.reserve(workers_);
        if (defer_fasta_) retirer_ = std::thread([this] { retire_files(); });
        // A reader spends most of its time waiting on storage, so there are as many as the reads
        // a parallel file system needs in flight, however few the loaders.
        auto const readers =
            ablated("read-depth") ? workers_
                                  : std::min(window_, std::max(workers_, parallel_read_depth));
        readers_.reserve(readers);
        for (size_t i = 0; i < readers; ++i) {
            readers_.emplace_back([this] { read_files(); });
        }
        for (size_t i = 0; i < workers_; ++i) threads_.emplace_back([this] { work(); });
    }

~fastx_load_pool()

cuddl::detail::fastx_load_pool::~fastx_load_pool ( )

inline

Definition at line 1840 of file fastx_sequence_file.hpp.

                       {
        {
            std::lock_guard lock(mutex_);
            stop_ = true;
        }
        assign_.notify_all();
        mapped_ready_.notify_all();
        for (auto& thread : readers_) thread.join();
        for (auto& thread : threads_) thread.join();
        // Loaders may be waiting for retirement space, so stop the reaper only after joining them.
        if (retirer_.joinable()) {
            {
                std::lock_guard lock(retired_mutex_);
                retire_stop_ = true;
            }
            retired_ready_.notify_one();
            retirer_.join();
        }
    }

fastx_load_pool() [2/2]

cuddl::detail::fastx_load_pool::fastx_load_pool ( fastx_load_pool const &  )

delete

Member Function Documentation

cap()

size_t cuddl::detail::fastx_load_pool::cap ( size_t  limit )

inline

Stops the loaders before id limit, or after the ids they already claimed.

A second consumer can own the ids past the returned limit without racing the loaders.

Returns

How many ids the pool loads in total.

Definition at line 1910 of file fastx_sequence_file.hpp.

                             {
        {
            std::lock_guard lock(mutex_);
            limit_ = std::clamp(limit, claimed_end_, limit_);
            limit = limit_;
        }
        assign_.notify_all();
        filled_.notify_all();
        mapped_ready_.notify_all();
        return limit;
    }

operator=()

fastx_load_pool & cuddl::detail::fastx_load_pool::operator= ( fastx_load_pool const &  )

delete

ready()

size_t cuddl::detail::fastx_load_pool::ready ( )

inline

Completion-order results waiting to be taken.

Definition at line 1901 of file fastx_sequence_file.hpp.

                                 {
        std::lock_guard lock(mutex_);
        return ready_count_;
    }

take()

Result< std::unique_ptr< fastx_sequence_file > > cuddl::detail::fastx_load_pool::take ( size_t  id )

inline

Definition at line 1862 of file fastx_sequence_file.hpp.

                                                                             {
        std::unique_lock lock(mutex_);
        filled_.wait(lock, [&] { return results_[id].has_value() || errors_[id] != nullptr; });
        ++taken_;
        // One freed window slot needs one loader, not all of them: waking the rest is a syscall
        // and a context switch each, once per genome, at 100,000 genomes.
        assign_.notify_one();
        lock.unlock();
        if (errors_[id] != nullptr) std::rethrow_exception(errors_[id]);
        return std::move(*results_[id]);
    }

take_ready()

Result< std::pair< size_t, std::unique_ptr< fastx_sequence_file > > > cuddl::detail::fastx_load_pool::take_ready ( )

inline

Definition at line 1874 of file fastx_sequence_file.hpp.

                                                                                           {
        if (ready_.empty())
            return Err(Error::invalid_argument("completion-order loading is disabled"));
        std::unique_lock lock(mutex_);
        filled_.wait(lock, [&] { return ready_count_ != 0; });
        auto const id = ready_[ready_head_];
        ready_head_ = (ready_head_ + 1) % ready_.size();
        --ready_count_;
        ++taken_;
        assign_.notify_one();
        lock.unlock();
        if (errors_[id] != nullptr) std::rethrow_exception(errors_[id]);
        return std::pair{id, CUDDL_TRY(std::move(*results_[id]))};
    }

Here is the call graph for this function:

take_ready_until_limit()

Result< std::optional< std::pair< size_t, std::unique_ptr< fastx_sequence_file > > > > cuddl::detail::fastx_load_pool::take_ready_until_limit ( )

inline

take_ready, or nothing once every id below the cap limit is taken.

Definition at line 1891 of file fastx_sequence_file.hpp.

                             {
        {
            std::unique_lock lock(mutex_);
            filled_.wait(lock, [&] { return ready_count_ != 0 || taken_ >= limit_; });
            if (ready_count_ == 0) return std::nullopt;
        }
        return CUDDL_TRY(take_ready());
    }

---

The documentation for this class was generated from the following file:

• fastx_sequence_file.hpp