diff --git a/nsparse/cluster/inverted_list_clusters.cpp b/nsparse/cluster/inverted_list_clusters.cpp index 73ff26b..76bc690 100644 --- a/nsparse/cluster/inverted_list_clusters.cpp +++ b/nsparse/cluster/inverted_list_clusters.cpp @@ -14,6 +14,7 @@ #include #include #include +#include #include #include @@ -118,6 +119,32 @@ InvertedListClusters::InvertedListClusters( docs_.insert(docs_.end(), doc_ids.begin(), doc_ids.end()); offsets_.push_back(docs_.size()); } + sort_cluster_docs(); +} + +// Sort the doc ids within each cluster ascending. Within-cluster iteration +// order does not affect search results (the top-k heap and the visited dedup +// are order-independent), but it dictates the memory access pattern of the +// per-doc gather: indptr[doc_id] and the forward-index rows it points at are +// spread across the multi-GB corpus. Ascending doc ids turn that per-doc random +// gather into a monotonic sweep the hardware prefetcher and TLB can follow. +void InvertedListClusters::sort_cluster_docs() { + if (offsets_.size() <= 1) return; + for (size_t c = 0; c + 1 < offsets_.size(); ++c) { + std::sort(docs_.begin() + offsets_[c], docs_.begin() + offsets_[c + 1]); + } +} + +void InvertedListClusters::validate_doc_ids(size_t num_docs) const { + // docs_ is sorted per cluster but not globally, so scan the flat array. + // One O(nnz) pass at load; negligible against deserialize + the sort above. + for (const idx_t doc_id : docs_) { + if (static_cast(doc_id) >= num_docs) { + throw std::out_of_range( + "InvertedListClusters: doc id out of range for corpus size; " + "index is corrupt or does not match its vectors"); + } + } } InvertedListClusters::InvertedListClusters(const InvertedListClusters& other) = @@ -291,6 +318,10 @@ void InvertedListClusters::deserialize(IOReader* reader) { offsets_.resize(n_offsets); reader->read(offsets_.data(), sizeof(idx_t), n_offsets); } + // Order within each cluster does not affect results but makes the per-doc + // gather monotonic (see sort_cluster_docs). Applied on load so existing + // serialized indexes benefit without a rebuild. + sort_cluster_docs(); reader->read(&n_clusters_, sizeof(size_t), 1); reader->read(&element_size_, sizeof(size_t), 1); diff --git a/nsparse/cluster/inverted_list_clusters.h b/nsparse/cluster/inverted_list_clusters.h index 9b6beb9..aae7357 100644 --- a/nsparse/cluster/inverted_list_clusters.h +++ b/nsparse/cluster/inverted_list_clusters.h @@ -41,6 +41,13 @@ class InvertedListClusters : public Serializable { void serialize(IOWriter* writer) const override; void deserialize(IOReader* reader) override; + // Validate that every stored doc id is within [0, num_docs). The search + // path indexes a VisitedSet sized to num_docs by doc id with no per- + // candidate bounds check (that would cost a branch on the hot path), so an + // out-of-domain id from a corrupt or mismatched serialized index would be + // an out-of-bounds write. Called once at load to fail loudly instead. + void validate_doc_ids(size_t num_docs) const; + // Accumulate per-cluster summary scores for a query into `out` (resized to // the cluster count) using the term-major transpose. The query is given as // its sparse (term, value) pairs; `q_val_bytes` points at the query values @@ -56,6 +63,9 @@ class InvertedListClusters : public Serializable { private: // Build the term-major (CSC) transpose from a per-cluster CSR summary. void build_transpose(const SparseVectors& summaries); + // Sort doc ids ascending within each cluster (result-order invariant) so + // the per-doc gather over the forward index is monotonic, not random. + void sort_cluster_docs(); template void score_summaries_typed(const term_t* q_idx, const T* q_val, size_t q_len, std::vector& out) const; diff --git a/nsparse/seismic_index.cpp b/nsparse/seismic_index.cpp index 332cb8d..e3c57c5 100644 --- a/nsparse/seismic_index.cpp +++ b/nsparse/seismic_index.cpp @@ -16,7 +16,6 @@ #include #include -#include "absl/container/flat_hash_set.h" #include "nsparse/cluster/inverted_list_clusters.h" #include "nsparse/cluster/random_kmeans.h" #include "nsparse/exact_matcher.h" @@ -32,6 +31,7 @@ #include "nsparse/utils/prefetch.h" #include "nsparse/utils/ranker.h" #include "nsparse/utils/vector_process.h" +#include "nsparse/utils/visited_set.h" namespace nsparse { namespace { @@ -41,9 +41,10 @@ constexpr int kElementSize = U32; void query_single_inverted_list( const SparseVectors* vectors, const InvertedListClusters& cluster_invlist, const std::vector& dense, const term_t* q_idx, const float* q_val, - size_t q_len, std::vector& score_scratch, const float heap_factor, + size_t q_len, std::vector& score_scratch, + std::vector& cluster_order, const float heap_factor, const bool first_list, const SearchParameters* search_parameters, - detail::TopKHolder& heap, absl::flat_hash_set& visited) { + detail::TopKHolder& heap, detail::VisitedSet& visited) { // Skip empty clusters size_t csize = cluster_invlist.cluster_size(); if (csize == 0) { @@ -58,39 +59,39 @@ void query_single_inverted_list( cluster_invlist.score_summaries_transposed( q_idx, reinterpret_cast(q_val), q_len, score_scratch); const std::vector& summary_scores = score_scratch; - size_t num_vectors = vectors->num_vectors(); - - std::vector cluster_order = - detail::reorder_clusters(summary_scores, first_list); + const size_t n_clusters = summary_scores.size(); const auto& [indptr, indices, values] = vectors->get_all_data(); - for (const size_t& cluster_id : cluster_order) { - const auto& cluster_score = summary_scores[cluster_id]; + // Process one cluster: prune by summary score, then score its docs. + // Returns false when the early-out fires on the first (sorted) list, which + // means no later cluster can qualify either — the caller stops iterating. + auto process_cluster = [&](size_t cluster_id) -> bool { + const float cluster_score = summary_scores[cluster_id]; if (heap.full() && (cluster_score * heap_factor < heap.peek_score())) { - if (first_list) { - break; - } - continue; + // On the first list clusters are visited in descending score order, + // so once one falls below the threshold every later one does too. + return !first_list; } const auto& docs = cluster_invlist.get_docs(cluster_id); const size_t n_docs = docs.size(); - static constexpr size_t kPrefetchDist1 = 2; // vector data prefetch - static constexpr size_t kPrefetchDist2 = 4; // indptr prefetch for (size_t i = 0; i < n_docs; ++i) { - const auto& doc_id = docs[i]; - if (i + kPrefetchDist2 < n_docs) { - detail::prefetch_indptr(indptr, docs[i + kPrefetchDist2]); - } + const idx_t doc_id = docs[i]; + // Prefetch one doc ahead, only the leading lines of the upcoming + // row; the row is contiguous so the hardware streamer pulls the + // tail, while bounding outstanding software prefetches keeps the + // line-fill buffers from saturating (measured optimum ~4 lines). + static constexpr size_t kPrefetchDist1 = 1; if (i + kPrefetchDist1 < n_docs) { - const idx_t next_doc = docs[i + kPrefetchDist1]; - const idx_t next_start = indptr[next_doc]; - const size_t next_len = indptr[next_doc + 1] - next_start; - detail::prefetch_vector(indices + next_start, - values + next_start, next_len); + const idx_t nd = docs[i + kPrefetchDist1]; + const idx_t next_start = indptr[nd]; + const size_t next_len = indptr[nd + 1] - next_start; + static constexpr size_t kPrefetchHeadLines = 4; + detail::prefetch_vector_head(indices + next_start, + values + next_start, next_len, + kPrefetchHeadLines); } - auto [_, inserted] = visited.insert(doc_id); - if (!inserted) { + if (!visited.insert(static_cast(doc_id))) { continue; } if (id_selector != nullptr && !id_selector->is_member(doc_id)) { @@ -102,6 +103,29 @@ void query_single_inverted_list( indices + start, values + start, len, dense.data()); heap.add(score, doc_id); } + return true; + }; + + if (first_list) { + // Only the first list is score-ordered; sort a per-thread scratch of + // narrow (u32) cluster ids instead of allocating a size_t vector per + // list. Clusters per list stay well under 2^32. + cluster_order.resize(n_clusters); + std::iota(cluster_order.begin(), cluster_order.end(), 0U); + std::ranges::sort(cluster_order, [&](uint32_t a, uint32_t b) { + return summary_scores[a] > summary_scores[b]; + }); + for (const uint32_t cluster_id : cluster_order) { + if (!process_cluster(cluster_id)) { + break; + } + } + } else { + // Later lists are visited in natural cluster order, so iterate directly + // — no order array, iota, or sort needed. + for (size_t cluster_id = 0; cluster_id < n_clusters; ++cluster_id) { + process_cluster(cluster_id); + } } } } // namespace @@ -187,8 +211,14 @@ auto SeismicIndex::search(idx_t n, const idx_t* indptr, const term_t* indices, #pragma omp parallel { std::vector dense(dim, 0.0F); - absl::flat_hash_set visited; - visited.reserve(static_cast(std::max(k, 1)) * 4096); + // Generation-stamped visited set over the doc-id domain: O(1) reset per + // query and a single indexed load per candidate instead of a hashed + // random probe (the doc loop is memory-bound on random gathers). + detail::VisitedSet visited(vectors_->num_vectors()); + // Per-thread scratch reused across queries: the per-cluster summary + // score buffer and the sorted cluster-order buffer (first list only). + std::vector score_scratch; + std::vector cluster_order; #pragma omp for schedule(dynamic, 64) for (idx_t query_idx = 0; query_idx < n; ++query_idx) { @@ -200,7 +230,8 @@ auto SeismicIndex::search(idx_t n, const idx_t* indptr, const term_t* indices, detail::top_k_tokens(q_indices, q_values, len, parameters->cut); auto [distances, labels] = single_query(dense, visited, q_indices, q_values, len, cuts, k, - parameters->heap_factor, search_parameters); + parameters->heap_factor, search_parameters, + score_scratch, cluster_order); result_distances[query_idx] = std::move(distances); result_labels[query_idx] = std::move(labels); } @@ -219,11 +250,13 @@ auto SeismicIndex::search(idx_t n, const idx_t* indptr, const term_t* indices, * @return std::pair, std::vector> */ auto SeismicIndex::single_query(std::vector& dense, - absl::flat_hash_set& visited, + detail::VisitedSet& visited, const term_t* q_indices, const float* q_values, size_t q_len, const std::vector& cuts, int k, float heap_factor, - SearchParameters* search_parameters) + SearchParameters* search_parameters, + std::vector& score_scratch, + std::vector& cluster_order) -> pair_of_score_id_vector_t { size_t num_docs = vectors_->num_vectors(); if (num_docs == 0) { @@ -234,10 +267,9 @@ auto SeismicIndex::single_query(std::vector& dense, for (size_t i = 0; i < q_len; ++i) { dense[q_indices[i]] = q_values[i]; } - visited.clear(); + visited.new_query(); detail::TopKHolder holder(k); - std::vector score_scratch; bool first_list = true; for (const auto& term : cuts) { if (term >= clustered_inverted_lists.size()) [[unlikely]] { @@ -246,8 +278,8 @@ auto SeismicIndex::single_query(std::vector& dense, const auto& cluster_invlist = clustered_inverted_lists[term]; query_single_inverted_list(vectors_.get(), cluster_invlist, dense, q_indices, q_values, q_len, score_scratch, - heap_factor, first_list, search_parameters, - holder, visited); + cluster_order, heap_factor, first_list, + search_parameters, holder, visited); first_list = false; } @@ -287,5 +319,14 @@ void SeismicIndex::read_index(IOReader* io_reader) { SeismicInvertedListsWriter inv_list_writer({}); inv_list_writer.deserialize(io_reader); clustered_inverted_lists = std::move(inv_list_writer.release()); + // The search path indexes a VisitedSet by doc id without a per-candidate + // bounds check; validate the loaded ids fall in the corpus domain once here + // so a corrupt/mismatched index fails loudly instead of writing OOB. + if (vectors_ != nullptr) { + const size_t num_docs = vectors_->num_vectors(); + for (const auto& cluster_invlist : clustered_inverted_lists) { + cluster_invlist.validate_doc_ids(num_docs); + } + } } } // namespace nsparse \ No newline at end of file diff --git a/nsparse/seismic_index.h b/nsparse/seismic_index.h index c9ffc8b..6a89c85 100644 --- a/nsparse/seismic_index.h +++ b/nsparse/seismic_index.h @@ -12,13 +12,13 @@ #include #include -#include "absl/container/flat_hash_set.h" #include "nsparse/cluster/inverted_list_clusters.h" #include "nsparse/index.h" #include "nsparse/io/io.h" #include "nsparse/seismic_common.h" #include "nsparse/sparse_vectors.h" #include "nsparse/types.h" +#include "nsparse/utils/visited_set.h" namespace nsparse { @@ -67,12 +67,15 @@ class SeismicIndex : public Index, public IndexIO { // `dense` and `visited` are per-thread scratch reused across the queries a // thread handles (see search()). `dense` must be all-zero on entry and is // restored to all-zero on exit via a sparse clear over the query's own - // dims (q_indices/q_len); `visited` is cleared on entry. - auto single_query(std::vector& dense, - absl::flat_hash_set& visited, + // dims (q_indices/q_len); `visited` starts a new generation on entry. + // `score_scratch` and `cluster_order` are per-thread scratch reused across + // queries (resized in place), avoiding a per-query/per-list allocation. + auto single_query(std::vector& dense, detail::VisitedSet& visited, const term_t* q_indices, const float* q_values, size_t q_len, const std::vector& cuts, int k, - float heap_factor, SearchParameters* search_parameters) + float heap_factor, SearchParameters* search_parameters, + std::vector& score_scratch, + std::vector& cluster_order) -> pair_of_score_id_vector_t; std::unique_ptr vectors_; SeismicClusterParameters cluster_parameter_; diff --git a/nsparse/seismic_scalar_quantized_index.cpp b/nsparse/seismic_scalar_quantized_index.cpp index 63599fe..e72c146 100644 --- a/nsparse/seismic_scalar_quantized_index.cpp +++ b/nsparse/seismic_scalar_quantized_index.cpp @@ -17,7 +17,6 @@ #include #include -#include "absl/container/flat_hash_set.h" #include "nsparse/cluster/inverted_list_clusters.h" #include "nsparse/cluster/random_kmeans.h" #include "nsparse/exact_matcher.h" @@ -34,6 +33,7 @@ #include "nsparse/utils/prefetch.h" #include "nsparse/utils/scalar_quantizer.h" #include "nsparse/utils/vector_process.h" +#include "nsparse/utils/visited_set.h" namespace nsparse { namespace { @@ -46,7 +46,7 @@ void query_single_inverted_list(const SparseVectors* vectors, float heap_factor, bool first_list, const SearchParameters* search_parameters, detail::TopKHolder& heap, - absl::flat_hash_set& visited) { + detail::VisitedSet& visited) { // Skip empty clusters size_t csize = cluster_invlist.cluster_size(); if (csize == 0) { @@ -80,31 +80,23 @@ void query_single_inverted_list(const SparseVectors* vectors, } const auto& docs = cluster_invlist.get_docs(cluster_id); const size_t n_docs = docs.size(); - // Two-stage prefetch pipeline: - // Stage 1 (distance 2): prefetch indptr[docs[i+2]] so the indptr - // lookup is cached by the time we need it next iteration. - // Stage 2 (distance 1): read indptr[docs[i+1]] (now cached from - // stage 1 issued last iteration), prefetch the actual vector data. - static constexpr size_t kPrefetchDist1 = 2; // vector data prefetch - static constexpr size_t kPrefetchDist2 = 4; // indptr prefetch for (size_t i = 0; i < n_docs; ++i) { - const auto& doc_id = docs[i]; - // Stage 1: prefetch indptr entry for doc at distance 2 - if (i + kPrefetchDist2 < n_docs) { - detail::prefetch_indptr(indptr, docs[i + kPrefetchDist2]); - } - // Stage 2: prefetch vector data for next doc (indptr should - // already be cached from stage 1 issued kPrefetchDist2 - - // kPrefetchDist1 iterations ago) + const idx_t doc_id = docs[i]; + // Prefetch one doc ahead, only the leading lines of the upcoming + // row; the row is contiguous so the hardware streamer pulls the + // tail, while bounding outstanding software prefetches keeps the + // line-fill buffers from saturating (measured optimum ~4 lines). + static constexpr size_t kPrefetchDist1 = 1; if (i + kPrefetchDist1 < n_docs) { const idx_t next_doc = docs[i + kPrefetchDist1]; const idx_t next_start = indptr[next_doc]; const size_t next_len = indptr[next_doc + 1] - next_start; - detail::prefetch_vector(indices + next_start, - values + next_start, next_len); + static constexpr size_t kPrefetchHeadLines = 4; + detail::prefetch_vector_head(indices + next_start, + values + next_start, next_len, + kPrefetchHeadLines); } - auto [_, inserted] = visited.insert(doc_id); - if (!inserted) { + if (!visited.insert(static_cast(doc_id))) { continue; } if (id_selector != nullptr && !id_selector->is_member(doc_id)) { @@ -248,8 +240,7 @@ auto SeismicScalarQuantizedIndex::search(idx_t n, const idx_t* indptr, #pragma omp parallel { std::vector dense(dense_bytes, 0); - absl::flat_hash_set visited; - visited.reserve(static_cast(std::max(k, 1)) * 4096); + detail::VisitedSet visited(vectors_->num_vectors()); #pragma omp for schedule(dynamic, 64) for (idx_t query_idx = 0; query_idx < n; ++query_idx) { @@ -280,7 +271,7 @@ auto SeismicScalarQuantizedIndex::search(idx_t n, const idx_t* indptr, } auto SeismicScalarQuantizedIndex::single_query( - std::vector& dense, absl::flat_hash_set& visited, + std::vector& dense, detail::VisitedSet& visited, const term_t* q_idx, const uint8_t* q_val_bytes, size_t q_len, size_t element_size, const std::vector& cuts, int k, float heap_factor, const ScalarQuantizer& query_sq, @@ -296,7 +287,7 @@ auto SeismicScalarQuantizedIndex::single_query( std::copy_n(q_val_bytes + i * element_size, element_size, dense.data() + static_cast(q_idx[i]) * element_size); } - visited.clear(); + visited.new_query(); detail::TopKHolder holder(k); std::vector score_scratch; @@ -353,6 +344,15 @@ void SeismicScalarQuantizedIndex::read_index(IOReader* io_reader) { SeismicInvertedListsWriter inv_list_writer({}); inv_list_writer.deserialize(io_reader); clustered_inverted_lists = std::move(inv_list_writer.release()); + // The search path indexes a VisitedSet by doc id without a per-candidate + // bounds check; validate the loaded ids fall in the corpus domain once here + // so a corrupt/mismatched index fails loudly instead of writing OOB. + if (vectors_ != nullptr) { + const size_t num_docs = vectors_->num_vectors(); + for (const auto& cluster_invlist : clustered_inverted_lists) { + cluster_invlist.validate_doc_ids(num_docs); + } + } } void SeismicScalarQuantizedIndex::write_header(IOWriter* io_writer) { diff --git a/nsparse/seismic_scalar_quantized_index.h b/nsparse/seismic_scalar_quantized_index.h index 179b226..4cf7da8 100644 --- a/nsparse/seismic_scalar_quantized_index.h +++ b/nsparse/seismic_scalar_quantized_index.h @@ -14,11 +14,11 @@ #include #include -#include "absl/container/flat_hash_set.h" #include "nsparse/cluster/inverted_list_clusters.h" #include "nsparse/index.h" #include "nsparse/seismic_index.h" #include "nsparse/utils/scalar_quantizer.h" +#include "nsparse/utils/visited_set.h" namespace nsparse { @@ -70,9 +70,9 @@ class SeismicScalarQuantizedIndex : public Index, public IndexIO { // thread handles (see search()). `dense` (a dimension-sized quantized-code // buffer, element_size bytes per dim) must be all-zero on entry and is // restored to all-zero on exit via a sparse clear over the query's own dims - // (q_idx/q_len); `visited` is cleared on entry. + // (q_idx/q_len); `visited` starts a new generation on entry. auto single_query(std::vector& dense, - absl::flat_hash_set& visited, const term_t* q_idx, + detail::VisitedSet& visited, const term_t* q_idx, const uint8_t* q_val_bytes, size_t q_len, size_t element_size, const std::vector& cuts, int k, float heap_factor, const ScalarQuantizer& query_sq, diff --git a/nsparse/sparse_vectors.cpp b/nsparse/sparse_vectors.cpp index fe232f0..5558921 100644 --- a/nsparse/sparse_vectors.cpp +++ b/nsparse/sparse_vectors.cpp @@ -15,6 +15,7 @@ #include "nsparse/io/io.h" #include "nsparse/types.h" #include "nsparse/utils/checks.h" +#include "nsparse/utils/hugepage.h" namespace nsparse { SparseVectors::SparseVectors(SparseVectorsConfig config) : config_(config) { @@ -165,6 +166,13 @@ void SparseVectors::deserialize(IOReader* io_reader) { size_t value_size = indptr_[vector_count] * element_size; values_.resize(value_size); io_reader->read(values_.data(), sizeof(uint8_t), value_size); + + // The per-doc dot product gathers randomly across indices_/values_ + // (multi-GB at scale); back them with huge pages to cut TLB/page-walk + // cost. indptr_ is also randomly indexed by doc id. + detail::advise_hugepage(indptr_); + detail::advise_hugepage(indices_); + detail::advise_hugepage(values_); } } } // namespace nsparse \ No newline at end of file diff --git a/nsparse/utils/distance_avx512.h b/nsparse/utils/distance_avx512.h index 1969e23..ae79ce9 100644 --- a/nsparse/utils/distance_avx512.h +++ b/nsparse/utils/distance_avx512.h @@ -333,7 +333,7 @@ inline float dot_product_float_dense(const term_t* indices, // Gather 16 values from dense vector using indices __m512 dense_vals = _mm512_i32gather_ps(idx, dense, sizeof(float)); - // Load 16 weights (aligned load - weights must be 64-byte aligned) + // Load 16 weights __m512 weight_vals = _mm512_loadu_ps(weights + i); // Fused multiply-add: sum += weights * dense_vals diff --git a/nsparse/utils/hugepage.h b/nsparse/utils/hugepage.h new file mode 100644 index 0000000..8077f0e --- /dev/null +++ b/nsparse/utils/hugepage.h @@ -0,0 +1,66 @@ +/** + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +#ifndef NSPARSE_HUGEPAGE_H +#define NSPARSE_HUGEPAGE_H + +#include +#include +#include + +#if defined(__linux__) +#include +#include +#endif + +namespace nsparse::detail { + +// Hint the kernel to back a large, mostly-read-only array with transparent +// huge pages. Random gathers into the multi-GB posting/summary arrays are +// dominated by TLB misses + page-table walks (2 MB pages cover 512x the span +// of 4 KB pages), so this cuts the DTLB/page-walk cost with no change to the +// data layout or results. No-op on non-Linux or for small buffers. +template +inline void advise_hugepage(const std::vector& v) { +#if defined(__linux__) && defined(MADV_HUGEPAGE) + const size_t bytes = v.size() * sizeof(T); + // 2 MB is the x86-64 huge-page size; only bother once the region spans at + // least one huge page so the alignment rounding can't consume the whole + // buffer. + static constexpr size_t kHugePage = size_t{2} << 20; + if (bytes < kHugePage) { + return; + } + const auto base = reinterpret_cast(v.data()); + // madvise() requires a page-aligned start; round the start up and the end + // down to whole huge pages so only fully-covered pages are advised. + const uintptr_t start = (base + kHugePage - 1) & ~(kHugePage - 1); + const uintptr_t end = (base + bytes) & ~(kHugePage - 1); + if (end > start) { + void* p = reinterpret_cast(start); + const size_t len = static_cast(end - start); + // Mark the region so future faults and khugepaged prefer huge pages. + ::madvise(p, len, MADV_HUGEPAGE); +#if defined(MADV_COLLAPSE) + // The arrays are already faulted in (resize + read) as 4 KB pages, so a + // plain MADV_HUGEPAGE only takes effect lazily via khugepaged. + // MADV_COLLAPSE (Linux 6.1+) collapses the existing pages into huge + // pages synchronously, giving the TLB win immediately. Best-effort: + // ignore failure (unsupported kernel, fragmentation). + ::madvise(p, len, MADV_COLLAPSE); +#endif + } +#else + (void)v; +#endif +} + +} // namespace nsparse::detail + +#endif // NSPARSE_HUGEPAGE_H diff --git a/nsparse/utils/prefetch.h b/nsparse/utils/prefetch.h index 72b8784..9b1733d 100644 --- a/nsparse/utils/prefetch.h +++ b/nsparse/utils/prefetch.h @@ -9,6 +9,7 @@ #ifndef PREFETCH_H #define PREFETCH_H +#include #include #include "nsparse/types.h" @@ -45,6 +46,30 @@ inline void prefetch_vector(const term_t* indices, const T* values, } } +// Prefetch only the leading `max_lines` cache lines of each stream. A doc row +// spans ~13 lines here; prefetching all of them for several docs ahead +// overruns the core's ~10-12 line-fill buffers (l1d_pend_miss.fb_full), which +// stalls demand loads. Because the row is contiguous, touching just the first +// line or two lets the hardware stream prefetcher pull the rest while keeping +// the number of outstanding software prefetches bounded. +template +inline void prefetch_vector_head(const term_t* indices, const T* values, + size_t len, size_t max_lines) { + static constexpr size_t kCacheLineSize = 64; // bytes + const char* indices_ptr = reinterpret_cast(indices); + const char* values_ptr = reinterpret_cast(values); + const size_t indices_bytes = + std::min(len * sizeof(term_t), max_lines * kCacheLineSize); + const size_t values_bytes = + std::min(len * sizeof(T), max_lines * kCacheLineSize); + for (size_t offset = 0; offset < indices_bytes; offset += kCacheLineSize) { + NSPARSE_PREFETCH(indices_ptr + offset, 0, 0); + } + for (size_t offset = 0; offset < values_bytes; offset += kCacheLineSize) { + NSPARSE_PREFETCH(values_ptr + offset, 0, 0); + } +} + inline void prefetch_indptr(const idx_t* indptr, idx_t doc_id) { NSPARSE_PREFETCH(&indptr[doc_id], 0, 0); } diff --git a/nsparse/utils/visited_set.h b/nsparse/utils/visited_set.h new file mode 100644 index 0000000..b34ce2b --- /dev/null +++ b/nsparse/utils/visited_set.h @@ -0,0 +1,75 @@ +/** + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + * + * The OpenSearch Contributors require contributions made to + * this file be licensed under the Apache-2.0 license or a + * compatible open source license. + */ + +#ifndef NSPARSE_VISITED_SET_H +#define NSPARSE_VISITED_SET_H + +#include +#include +#include + +namespace nsparse::detail { + +// Membership set over a fixed doc-id domain [0, n), backed by a bit per doc. +// +// The seismic doc loop tests/inserts every candidate doc to dedupe across +// clusters. A hash set hashes and probes a random cache line per candidate and +// must be cleared each query. A full generation-stamped uint32 array (4 B/doc) +// avoids the hashing but at seismic scale (~35 MB) adds a random-access stream +// larger than L2, so it misses cache on every candidate. A bitset is 32x +// smaller (1 bit/doc, ~1.1 MB at 8.8M docs) so far fewer distinct cache lines +// are touched, while the touched-word list lets a new query clear only the +// words it actually dirtied (sparse O(visited) reset, not O(n)). +class VisitedSet { +public: + VisitedSet() = default; + explicit VisitedSet(size_t n) { resize(n); } + + void resize(size_t n) { + bits_.assign((n + 63) / 64, 0); + touched_.clear(); + } + + // Begin a new query: clear only the words dirtied by the previous query. + void new_query() { + for (const size_t w : touched_) { + bits_[w] = 0; + } + touched_.clear(); + } + + // Mark `id` visited; return true if it was newly inserted this query. + // `id` must be in the doc-id domain [0, n) this set was sized for: unlike a + // hash set, an out-of-range id here is an out-of-bounds access on bits_, not + // a safe insert. Doc ids come from the corpus so this invariant holds, but + // it is now load-bearing — assert it in debug builds (compiles out in + // release, keeping the per-candidate hot path branch-free). + bool insert(size_t id) { + assert(id < (bits_.size() << 6) && "VisitedSet id out of domain"); + const size_t w = id >> 6; + const uint64_t mask = uint64_t{1} << (id & 63); + const uint64_t prev = bits_[w]; + if (prev & mask) { + return false; + } + if (prev == 0) { + touched_.push_back(w); + } + bits_[w] = prev | mask; + return true; + } + +private: + std::vector bits_; + std::vector touched_; +}; + +} // namespace nsparse::detail + +#endif // NSPARSE_VISITED_SET_H diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 2d59fa7..0275bdf 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -56,4 +56,9 @@ target_link_libraries(nsparse_test PRIVATE ) include(GoogleTest) -gtest_discover_tests(nsparse_test) +# DISCOVERY_TIMEOUT: gtest_discover_tests runs the built executable with +# --gtest_list_tests as a post-build step. On cold Windows CI runners, loading +# the exe together with its deployed DLLs (e.g. abseil_dll.dll) and enumerating +# every test case can exceed the 5s default, causing a spurious build failure. +# Bump the timeout well above that to keep discovery robust. +gtest_discover_tests(nsparse_test DISCOVERY_TIMEOUT 120)