cuddl::detail::database_stager< K, BucketCount, Layout > Class Template Reference

Encodes genome records into register rows through a device arena. More...

Public Member Functions
	database_stager (cuda::stream_ref stream, staging_plan bounds, reference_build_statistics *statistics=nullptr, bool direct=false, cuda::std::optional< blacklist_view > blacklist=cuda::std::nullopt)
	~database_stager ()
	Frees the arenas only after every copy into them.
	database_stager (database_stager const &)=delete
database_stager &	operator= (database_stager const &)=delete
Result< void >	add_genome (size_t genome, std::span< fastx_sequence_extent const > records, char const *pinned_base=nullptr, size_t pinned_size=0, std::unique_ptr< fastx_sequence_file > holder={})
	Stages one genome.
Result< void >	add_packed_genome (size_t genome, std::unique_ptr< fastx_sequence_file > holder, char const *pinned_base=nullptr, size_t pinned_size=0)
	Stages one genome that pack_fastx_sequence_file packed into two-bit runs.
Result< void >	add_resident (size_t genome, char const *bases, size_t size)
	Stages one genome's bases that already sit on the device, without a copy.
Result< void >	begin_group (size_t genomes)
	Clears the register rows of the next group of genomes.
Result< device_span< uint32_t const > >	end_group (size_t genomes)
	Launches everything staged and returns the group's rows on the device.
Result< void >	finish ()
	Waits for the device and publishes the statistics.
Result< void >	flush ()
	One launch for everything staged so far.
void	wait_hold_slot ()
	Waits until the next genome staged by copy can be held without blocking.

Static Public Attributes
static constexpr size_t	hold_slots = stager_hold_slots

Detailed Description

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>
class cuddl::detail::database_stager< K, BucketCount, Layout >

Encodes genome records into register rows through a device arena.

One copy of the staging machinery, fed by every entry point that builds a reference database from bases. A producer supplies, per genome, the spans of its records: one span per record, bases only, no line breaks, and no k-mer crossing two spans.

The arena, its descriptors, the record runs, the k - 1 overlap a record too large for the arena needs and the batch launches all live here. Rows stay on the device: each group's rows go to the caller's sink, which copies them wherever the build wants them. A producer keeps only what is its own: where the bytes come from, and which of them must outlive the copies the device makes from them.

Definition at line 222 of file database_staging.cuh.

Constructor & Destructor Documentation

database_stager() [1/2]

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

cuddl::detail::database_stager< K, BucketCount, Layout >::database_stager ( cuda::stream_ref  stream, staging_plan  bounds, reference_build_statistics *  statistics = nullptr, bool  direct = false, cuda::std::optional< blacklist_view >  blacklist = cuda::std::nullopt  )

inline

Exceptions

cuda::cuda_error

or std::bad_alloc when the device buffers cannot be allocated.

Parameters

stream	Stream owning the arena, descriptors, and row store.
bounds	Arena sizing the device row store, arena, and descriptors.
statistics	Optional build statistics sink, null to skip.
direct	stages the caller's bytes in place, for a device that reads pageable host memory: no copy, and the arena stays unused.

Definition at line 256 of file database_staging.cuh.

        : stream_(stream),
          bounds_(bounds),
          statistics_(statistics),
          direct_(direct),
          blacklist_(blacklist),
          rows_(
              cuda::make_device_buffer<uint32_t>(
                  stream,
                  stream.device(),
                  bounds.group * BucketCount,
                  cuda::no_init
              )
          ),
          copy_stream_(stream.device()),
          kernel_stream_(stream.device(), greatest_stream_priority()),
          // The copy stream fills one arena while kernels read the others.
          arena_(
              cuda::make_device_buffer<char>(
                  stream,
                  stream.device(),
                  direct ? bounds.staging : stager_arenas * arena_stride(bounds.staging),
                  cuda::no_init
              )
          ),
          descriptors_(
              cuda::make_device_buffer<sequence_batch_chunk>(
                  stream,
                  stream.device(),
                  bounds.max_pieces,
                  cuda::no_init
              )
          ),
          held_(hold_slots),
          sm_(static_cast<size_t>(
              stream.device().attribute(cuda::device_attributes::multiprocessor_count)
          )),
          max_grid_(
              static_cast<size_t>(
                  stream.device().attribute(cuda::device_attributes::max_grid_dim_x)
              )
          ) {
        held_consumed_.reserve(hold_slots);
        for (size_t slot = 0; slot < hold_slots; ++slot) {
            held_consumed_.emplace_back(stream);
        }
        staged_chunks_.reserve(std::min<size_t>(bounds_.max_pieces, size_t{1} << 16));
        copied_chunks_.reserve(staged_chunks_.capacity());
        // The arenas are allocated in order on the main stream.
        copy_stream_.wait(stream_);
    }

~database_stager()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

cuddl::detail::database_stager< K, BucketCount, Layout >::~database_stager ( )

inline

Frees the arenas only after every copy into them.

Definition at line 314 of file database_staging.cuh.

                       {
        CUDDL_CUDA_ABORT(cudaEventRecord(copies_done_.get(), copy_stream_.get()));
        CUDDL_CUDA_ABORT(cudaStreamWaitEvent(stream_.get(), copies_done_.get()));
    }

database_stager() [2/2]

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

cuddl::detail::database_stager< K, BucketCount, Layout >::database_stager ( database_stager< K, BucketCount, Layout > const &  )

delete

Member Function Documentation

add_genome()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

Result< void > cuddl::detail::database_stager< K, BucketCount, Layout >::add_genome ( size_t  genome, std::span< fastx_sequence_extent const >  records, char const *  pinned_base = nullptr, size_t  pinned_size = 0, std::unique_ptr< fastx_sequence_file >  holder = {}  )

inline

Stages one genome.

holder keeps its bytes alive until the device is done.

A genome with no records, or with records shorter than one window, keeps its row and records no descriptor. pinned_base names the page-locked buffer the bytes came from, when the producer has one, so the statistics can say which transfer path was taken.

Definition at line 326 of file database_staging.cuh.

                                                    {}
    ) {
        auto const record_size = [&](size_t index) {
            return static_cast<size_t>(records[index].end - records[index].begin);
        };
        size_t record = 0;
        while (record < records.size()) {
            // Staged in place, the arena is not the limit; the descriptors and the piece size are.
            auto const room =
                direct_ ? std::numeric_limits<size_t>::max() : bounds_.staging - arena_used_;
            size_t run = record;
            while (run < records.size()) {
                auto const run_bytes =
                    static_cast<size_t>(records[run].end - records[record].begin);
                if (run_bytes > room || record_size(run) > bounds_.max_piece ||
                    staged_chunks_.size() + (run - record) >= bounds_.max_pieces) {
                    break;
                }
                ++run;
            }
            if (run == record) {
                if (arena_used_ != 0) {
                    CUDDL_TRY(flush());
                    continue;
                }
                // One record larger than the whole arena: split it, and give each piece the
                // k - 1 bases of overlap its kmers need.
                auto const size = record_size(record);
                size_t offset = 0;
                while (offset < size) {
                    auto const overlap = offset == 0 ? size_t{0} : std::min(offset, size_t{K - 1});
                    auto const bases = std::min(size - offset, bounds_.max_piece);
                    auto const span = overlap + bases;
                    if (span < K) break;
                    if (arena_used_ + span > bounds_.staging ||
                        staged_chunks_.size() >= bounds_.max_pieces) {
                        CUDDL_TRY(flush());
                    }
                    auto const* const source = records[record].begin + offset - overlap;
                    CUDDL_TRY(stage(
                        source,
                        span,
                        genome,
                        static_cast<uint32_t>(span - K + 1),
                        pinned_base,
                        pinned_size
                    ));
                    offset += bases;
                }
                ++record;
                continue;
            }
            // Copy the run in one transfer, then describe each record inside it. Staged in
            // place, the records keep their own addresses and there is nothing to copy.
            auto const* const source = records[record].begin;
            auto const run_bytes = static_cast<size_t>(records[run - 1].end - source);
            account(source, run_bytes, pinned_base, pinned_size);
            if (!direct_) {
                CUDDL_CUDA_TRY(
                    cuda::copy_bytes(
                        copy_stream_,
                        cuda::std::span{source, run_bytes},
                        device_span<char>{arena() + arena_used_, run_bytes}
                    )
                );
            }
            for (size_t index = record; index < run; ++index) {
                auto const size = record_size(index);
                if (size < K) continue;  // no window to record
                auto const* const bases =
                    direct_ ? records[index].begin
                            : arena() + arena_used_ +
                                  static_cast<size_t>(records[index].begin - source);
                describe(bases, static_cast<uint32_t>(size - K + 1), genome);
            }
            if (direct_) {
                // Staged in place the arena never fills, so the bytes held for the device are
                // what bounds a batch. Half the arena, because a batch is still held while the
                // next one fills: the two together then cost what the copies would have.
                held_bytes_ += run_bytes;
                if (held_bytes_ >= bounds_.staging / 2) {
                    CUDDL_TRY(flush());
                }
            } else {
                arena_used_ += run_bytes;
                ++transfers_;
            }
            record = run;
        }
        if (holder) hold(std::move(holder));
        return Ok();
    }

add_packed_genome()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

Result< void > cuddl::detail::database_stager< K, BucketCount, Layout >::add_packed_genome ( size_t  genome, std::unique_ptr< fastx_sequence_file >  holder, char const *  pinned_base = nullptr, size_t  pinned_size = 0  )

inline

Stages one genome that pack_fastx_sequence_file packed into two-bit runs.

Runs keep their words together. One too large for a batch splits into pieces of whole words whose window counts are multiples of sixteen, so every piece starts on a word.

Definition at line 428 of file database_staging.cuh.

      {
        constexpr size_t word_bases = 16;
        auto const& file = *holder;
        auto const& runs = file.packed;
        auto const words = [&](size_t index) {
            return (runs[index].bases + word_bases - 1) / word_bases;
        };
        size_t run = 0;
        while (run < runs.size()) {
            auto const room = direct_ ? std::numeric_limits<size_t>::max() : arena_room_words();
            size_t last = run;
            while (last < runs.size()) {
                auto const span_words = runs[last].word + words(last) - runs[run].word;
                if (span_words > room || words(last) * sizeof(uint32_t) > bounds_.max_piece ||
                    staged_chunks_.size() + (last - run) >= bounds_.max_pieces) {
                    break;
                }
                ++last;
            }
            if (last == run) {
                if (arena_used_ != 0 || !staged_chunks_.empty()) {
                    CUDDL_TRY(flush());
                    continue;
                }
                auto const halo = (size_t{K} - 1 + word_bases - 1) / word_bases;
                auto const capacity =
                    std::min(bounds_.max_piece, bounds_.staging) / sizeof(uint32_t);
                if (capacity <= halo) {
                    return Err(Error::resource("staging arena too small for a packed run"));
                }
                auto const step = (capacity - halo) * word_bases;
                auto const total = runs[run].bases - K + 1;
                for (size_t start = 0; start < total; start += step) {
                    auto const windows = std::min(step, total - start);
                    auto const first = runs[run].word + start / word_bases;
                    auto const count = (start + windows + K - 1 + word_bases - 1) / word_bases -
                                       start / word_bases;
                    if ((!direct_ && count > arena_room_words()) ||
                        staged_chunks_.size() >= bounds_.max_pieces) {
                        CUDDL_TRY(flush());
                    }
                    auto const* const bases =
                        CUDDL_TRY(place_words(file.words + first, count, pinned_base, pinned_size));
                    describe(bases, static_cast<uint32_t>(windows), genome, true);
                }
                ++run;
                continue;
            }
            auto const first = runs[run].word;
            auto const count = runs[last - 1].word + words(last - 1) - first;
            auto const* const base =
                CUDDL_TRY(place_words(file.words + first, count, pinned_base, pinned_size));
            for (size_t index = run; index < last; ++index) {
                describe(
                    base + (runs[index].word - first) * sizeof(uint32_t),
                    static_cast<uint32_t>(runs[index].bases - K + 1),
                    genome,
                    true
                );
            }
            if (direct_ && held_bytes_ >= bounds_.staging / 2) CUDDL_TRY(flush());
            run = last;
        }
        hold(std::move(holder));
        return Ok();
    }

Here is the call graph for this function:

add_resident()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

Result< void > cuddl::detail::database_stager< K, BucketCount, Layout >::add_resident ( size_t  genome, char const *  bases, size_t  size  )

inline

Stages one genome's bases that already sit on the device, without a copy.

The bytes must stay valid until the batch that reads them runs: through the next flush, whose kernel the caller then orders its reuse of them after.

Definition at line 504 of file database_staging.cuh.

                                                                                           {
        if (size < K) return Ok();
        if (staged_chunks_.size() >= bounds_.max_pieces) CUDDL_TRY(flush());
        describe(bases, static_cast<uint32_t>(size - K + 1), genome);
        return Ok();
    }

Here is the call graph for this function:

begin_group()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

Result< void > cuddl::detail::database_stager< K, BucketCount, Layout >::begin_group ( size_t  genomes )

inline

Clears the register rows of the next group of genomes.

Definition at line 512 of file database_staging.cuh.

                                                           {
        auto const resident = CUDDL_TRY(begin_resident(stream_));
        CUDDL_CUDA_TRY(
            cuda::fill_bytes(
                stream_, cuda::std::span{rows_.data(), genomes * BucketCount}, uint32_t{0}
            )
        );
        CUDDL_TRY(end_resident(resident, stream_));
        return Ok();
    }

Here is the call graph for this function:

Here is the caller graph for this function:

end_group()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

Result< device_span< uint32_t const > > cuddl::detail::database_stager< K, BucketCount, Layout >::end_group ( size_t  genomes )

inline

Launches everything staged and returns the group's rows on the device.

Each genome's row holds BucketCount registers, the layout a single sketch allocation has. The rows are complete once the stream reaches this point, and the next begin_group reuses them, so a consumer enqueues its reads on the stager's stream before then.

Definition at line 529 of file database_staging.cuh.

                                                                                {
        CUDDL_TRY(flush());
        return device_span<uint32_t const>{rows_.data(), genomes * BucketCount};
    }

Here is the call graph for this function:

Here is the caller graph for this function:

finish()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

Result< void > cuddl::detail::database_stager< K, BucketCount, Layout >::finish ( )

inline

Waits for the device and publishes the statistics.

Definition at line 535 of file database_staging.cuh.

                                        {
        CUDDL_CUDA_TRY(stream_.sync());
        CUDDL_CUDA_TRY(copy_stream_.sync());
        if (statistics_ != nullptr) {
            statistics_->staging_bytes = bounds_.staging;
            statistics_->batches = batches_;
            statistics_->transfers = transfers_;
            for (auto const& span : resident_spans_) {
                float elapsed_ms = 0;
                CUDDL_CUDA_TRY(cudaEventElapsedTime(&elapsed_ms, span.start, span.stop));
                statistics_->resident_compute_ms += elapsed_ms;
            }
        }
        return Ok();
    }

Here is the call graph for this function:

Here is the caller graph for this function:

flush()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

Result< void > cuddl::detail::database_stager< K, BucketCount, Layout >::flush ( )

inline

One launch for everything staged so far.

Copies queued after it land in the arena only once the device has finished reading it, so no host wait is needed beyond the descriptor copy's own.

Definition at line 669 of file database_staging.cuh.

                                       {
        if (staged_chunks_.empty()) {
            // Records shorter than k stage bytes but record no window, so a flush can arrive
            // with nothing to launch. The offsets still have to reset, or a caller that flushed
            // to make room would ask again for the same room.
            arena_used_ = 0;
            block_end_ = 0;
            return Ok();
        }
        copied_consumed_.sync();
        copied_chunks_.swap(staged_chunks_);
        // The sketch kernel runs ahead of lower-priority work sharing the device, such as
        // nvCOMP inflation: the copies that feed the next batch wait on it.
        queued_.record(stream_);
        kernel_stream_.wait(queued_);
        if (!direct_) {
            copies_done_.record(copy_stream_);
            kernel_stream_.wait(copies_done_);
        }
        CUDDL_CUDA_TRY(
            cuda::copy_bytes(
                kernel_stream_,
                cuda::std::span{copied_chunks_.data(), copied_chunks_.size()},
                device_span<sequence_batch_chunk>{descriptors_.data(), copied_chunks_.size()}
            )
        );
        copied_consumed_.record(kernel_stream_);
        auto const grid = std::min(block_end_, max_grid_);
        if (grid != 0) {
            auto const resident = CUDDL_TRY(begin_resident(kernel_stream_));
            // The kernel keeps the default carveout: a 50% L1 split sped up filtered sketching on
            // an RTX 5090, but made `gpu`-backend builds 2.6 times slower beside nvCOMP's kernels.
            auto launch = [&]<bool Filter, bool Naive>() {
                add_sequence_batch_kernel<BucketCount, Layout, Filter, Naive>
                    <<<static_cast<uint32_t>(grid), 256, 0, kernel_stream_.get()>>>(
                        descriptors_.data(),
                        copied_chunks_.size(),
                        block_end_,
                        K,
                        rows_.data(),
                        blacklist_.value_or(blacklist_view{})
                    );
            };
            auto const encode = [&]<bool Naive>() {
                if (blacklist_) {
                    launch.template operator()<true, Naive>();
                } else {
                    launch.template operator()<false, Naive>();
                }
            };
            // Only ablation builds compile the naive encoder.
            if constexpr (ablations_enabled) {
                if (ablated("kmer-encoding")) {
                    encode.template operator()<true>();
                } else {
                    encode.template operator()<false>();
                }
            } else {
                encode.template operator()<false>();
            }
            CUDDL_CUDA_TRY(cudaGetLastError());
            CUDDL_TRY(end_resident(resident, kernel_stream_));
        }
        kernel_done_.record(kernel_stream_);
        stream_.wait(kernel_done_);
        if (direct_ && !batch_files_.empty()) {
            // The launch above is what reads those buffers, so they wait on this batch's event.
            released_.push_back({cuda::event{stream_}, std::move(batch_files_)});
            batch_files_.clear();
            released_.back().first.record(stream_);
            while (released_.size() > held_batches) {
                released_.front().first.sync();
                released_.pop_front();
            }
        }
        if (!direct_) {
            // The next batch fills the other arena once the kernel that last read it is done.
            arena_done_[arena_index_].record(stream_);
            arena_index_ = (arena_index_ + 1) % stager_arenas;
            copy_stream_.wait(arena_done_[arena_index_]);
        }
        ++batches_;
        staged_chunks_.clear();
        arena_used_ = 0;
        held_bytes_ = 0;
        block_end_ = 0;
        return Ok();
    }

Here is the call graph for this function:

Here is the caller graph for this function:

operator=()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

database_stager & cuddl::detail::database_stager< K, BucketCount, Layout >::operator= ( database_stager< K, BucketCount, Layout > const &  )

delete

wait_hold_slot()

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

void cuddl::detail::database_stager< K, BucketCount, Layout >::wait_hold_slot ( )

inline

Waits until the next genome staged by copy can be held without blocking.

The thread that stages copied genomes may call this outside any lock it shares with other staging threads, so waiting on the transfers does not hold them up.

Definition at line 762 of file database_staging.cuh.

                          {
        if (direct_) return;
        auto const slot = held_count_ % held_.size();
        if (held_[slot]) held_consumed_[slot].sync();
    }

Member Data Documentation

hold_slots

template<uint32_t K, size_t BucketCount, typename Layout = default_register_layout>

constexpr size_t cuddl::detail::database_stager< K, BucketCount, Layout >::hold_slots = stager_hold_slots

staticconstexpr

Definition at line 248 of file database_staging.cuh.

---

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

• database_staging.cuh