Method ReadCache

src/data/sparse_page_source.h:284–347

  common::Monitor monitor_;

  [[nodiscard]] bool ReadCache() {
    if (!cache_info_->written) {
      return false;
    }
    auto n_batches = this->cache_info_->Size();
    if (ring_->empty()) {
      ring_->resize(n_batches);
    }

    std::int32_t n_prefetches =
        std::min(this->workers_.NumWorkers(), this->param_.n_prefetch_batches);
    n_prefetches = std::max(n_prefetches, 1);
    std::int32_t n_prefetch_batches = std::min(static_cast<bst_idx_t>(n_prefetches), n_batches);
    CHECK_GT(n_prefetch_batches, 0);
    CHECK_LE(n_prefetch_batches, this->param_.n_prefetch_batches);
    std::size_t fetch_it = this->count_;

    exce_.Rethrow();
    // Clear out the existing page before loading new ones. This helps reduce memory usage
    // when page is not loaded with mmap. The destruction policy handles any necessary
    // synchronizations (e.g., CUDA stream sync for Ellpack pages).
    this->DestroyPage(&page_);

    for (std::int32_t i = 0; i < n_prefetch_batches; ++i, ++fetch_it) {
      bool restart = fetch_it == n_batches;
      fetch_it %= n_batches;  // ring
      if (ring_->at(fetch_it).valid()) {
        continue;
      }
      auto const* self = this;  // make sure it's const
      CHECK_LT(fetch_it, cache_info_->offset.size());
      // Make sure the new iteration starts with a copy to avoid spilling configuration.
      if (restart) {
        this->param_.prefetch_copy = true;
      }
      auto p = this->param_;
      ring_->at(fetch_it) = this->workers_.Submit([fetch_it, self, p, this] {
        auto page = std::make_shared<S>();
        this->exce_.Run([&] {
          std::unique_ptr<typename FormatStreamPolicy::FormatT> fmt{self->CreatePageFormat(p)};
          auto name = self->cache_info_->ShardName();
          auto [offset, length] = self->cache_info_->View(fetch_it);
          std::unique_ptr<typename FormatStreamPolicy::ReaderT> fi{
              self->CreateReader(name, offset, length)};
          CHECK(fmt->Read(page.get(), fi.get()));
        });
        return page;
      });
      this->fetch_cnt_++;
    }

    CHECK_EQ(std::count_if(ring_->cbegin(), ring_->cend(), [](auto const& f) { return f.valid(); }),
             n_prefetch_batches)
        << "Sparse DMatrix assumes forward iteration.";

    monitor_.Start("Wait-" + std::to_string(count_));
    CHECK((*ring_)[count_].valid());
    page_ = (*ring_)[count_].get();