This is an automated email from the ASF dual-hosted git repository.
airborne12 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f7396d7229a [improvement](be) Remove per-term contention from SNII
index build memory accounting (#70007)
f7396d7229a is described below
commit f7396d7229a26b421b07dde19f3be1dede8a7256
Author: Jack <[email protected]>
AuthorDate: Fri Nov 6 16:03:50 2026 +0800
[improvement](be) Remove per-term contention from SNII index build memory
accounting (#70007)
### What problem does this PR solve?
Issue Number: None
Related PR: #66052, #67657
Problem Summary: Stream load into a SNII table stops scaling once flush
threads compete. On a log table with ten inverted indexes and mostly
unique terms, SNII loaded at 314 MB/s against 854 MB/s for V2, with 3.0
times the CPU and 9.8 times the kernel CPU. The adaptive flush pool is
on by default and allows up to 4 threads per core, so this is the normal
case on a busy BE. With the pool fixed at 6 threads SNII was 23% slower
than V2, and at 30 threads 60% slower.
Two process-wide hot spots cause it, and both are reached per term:
1. `SpimiTermBuffer::report_arena_delta()` calls
`GlobalMemoryLimiter::report()` whenever the resident total changes,
which is once per new term. `report()` takes a process-wide mutex, looks
the writer up and reads the memory signals. The limiter keeps only the
arena size, which grows one 32 KiB block at a time, so nearly every call
repeats the same number. At 30 threads the flush threads blocked about
476k times per second against 21k for V2, and every futex wait sampled
was on one address. That was 13.6 cores of kernel CPU against 3.5.
2. The Reservations that #67657 added to the term encoders change
several times per term, and each change updates two process-wide atomics
through `snii_build_consume_release()`. That lambda took 54% of the user
CPU of the flush threads.
Both scale with new terms and with threads, which is why duplicate data
and a small pool look mild.
- The limiter is now told only when the arena size changes, or when the
resident total has moved by 1 MiB since its last decision. Nothing it
stores is lost. The step catches capacity doublings that move the
resident total by MiBs inside one arena block, and it is the step in
which the decision's input moves anyway. On 2M tokens with 600k new
terms a buffer now makes 321 decisions; before, every new term was one.
- A reporter now holds its changes back from the two process-wide
counters until they add up to 1 MiB, and publishes the remainder
whenever its live bytes are back at zero, and when it is destroyed. One
term swings tens of KiB plus a ZSTD context, so per-term reservations no
longer reach the shared cache lines. The zero rule matters because a
finished text writer's reporter is not destroyed with the writer: its
`IndexFileWriter` keeps it until the rowset writer closes.
**What this costs.** The reporter itself stays exact, so per-writer
budgets and spill thresholds do not change. The process-wide counters
can lag each writer that is still building by less than 1 MiB:
- `SniiIndexBuild` and `snii_index_build_mem_consumption` no longer show
a writer that stays below 1 MiB while it builds. They are exact again as
soon as a writer is empty, so a finished segment leaves nothing behind.
- The forced-spill decision reads the sum, so it can start late by the
lag of the writers still building: at most 1 GiB with 100 flush threads
on a 10-index table, against a share of at least 4 *
`inverted_index_ram_buffer_size` (2 GiB by default). A writer below 1
MiB could not be a victim anyway, since forced spills only pick arenas
above `snii_forced_spill_min_arena_bytes` (64 MiB by default), and the
process-level backstops read real memory.
The index bytes do not change.
| Throughput, MB/s, default adaptive flush pool | V2 | SNII | SNII vs V2
| SNII CPU / V2 CPU |
|---|---|---|---|---|
| Log data, before | 854 | 314 | -62.9% | 3.01 |
| Log data, after | 852 | **770** | -7.7% | 1.11 |
| Short logs, before | 997 | 389 | -61.0% | 2.84 |
| Short logs, after | 1003 | **865** | -13.7% | 1.20 |
Both binaries were built from master and alternated in one window, 4
rounds each with the order flipped, and the table shows medians. One BE
pinned to 32 cores, local disk, stream load of JSON lines with 16
concurrent 100 MB batches. Log data is expanded from a real sample of
service logs, where 1% of the rows carry base64 blobs and make up 68% of
the bytes. Short logs leave those rows out. V2 is unchanged, as it
should be, and kernel CPU falls from 516 s to 43 s on the log data.
An earlier prototype with the same two changes also held up with other
shapes and pools: fixed 30 threads -59.6% to -8.7%, fixed 6 threads
-23.1% to -14.8%, 1 GB batches -43.8% to -13.7%, and data with few
unique terms -20.2% to -11.7%.
Forced spills were checked on a real load: with
`snii_index_build_max_memory_limit_percent = 1`,
`snii_forced_spill_min_arena_bytes = 1 MiB`, and 16 concurrent 1 GB
batches, the spill directory held up to 224 files at once and the
limiter logged its shortfall warning. It stayed empty with the default
share. Eight MATCH, phrase and equality queries returned the same counts
as V2 on the 11M rows, and the tracker went back to 0.
---
.../snii/writer/snii_build_memory_tracker.cpp | 75 +++++++--
.../snii/writer/snii_build_memory_tracker.h | 25 ++-
.../inverted/snii/writer/spimi_term_buffer.cpp | 24 ++-
.../index/inverted/snii/writer/spimi_term_buffer.h | 9 +-
.../index/inverted/snii/snii_bkd_adapter_test.cpp | 5 +-
.../snii/writer/snii_build_memory_tracker_test.cpp | 187 ++++++++++++++++++---
.../writer/spimi_global_memory_limiter_test.cpp | 106 +++++++++++-
7 files changed, 377 insertions(+), 54 deletions(-)
diff --git
a/be/src/storage/index/inverted/snii/writer/snii_build_memory_tracker.cpp
b/be/src/storage/index/inverted/snii/writer/snii_build_memory_tracker.cpp
index 3c634339db9..8df493cbdba 100644
--- a/be/src/storage/index/inverted/snii/writer/snii_build_memory_tracker.cpp
+++ b/be/src/storage/index/inverted/snii/writer/snii_build_memory_tracker.cpp
@@ -18,6 +18,9 @@
#include "storage/index/inverted/snii/writer/snii_build_memory_tracker.h"
#include <atomic>
+#include <cstdlib>
+#include <memory>
+#include <utility>
#include "common/metrics/doris_metrics.h"
#include "common/metrics/metrics.h"
@@ -53,34 +56,76 @@ namespace {
// maintained directly instead of being derived by subtraction, and why it is a
// plain atomic rather than a second MemTracker.
std::atomic<int64_t> g_registered_build_bytes {0};
-} // namespace
-int64_t snii_registered_build_bytes() {
- return g_registered_build_bytes.load(std::memory_order_relaxed);
-}
+// One reporter's changes that have not reached the process-wide counters yet.
+// Every change is either in pending_ or published, never both, so the counters
+// add up to the reporter's live bytes minus pending_.
+//
+// Changes publish once pending_ reaches a step, and also whenever the
reporter's live
+// bytes are back at zero. A finished text writer's reporter is not destroyed
with the
+// writer: IndexFileWriter keeps it until the rowset writer closes, so
whatever it held
+// back would stay in the spill decision's input for every later segment of
the load.
+// For example +1.5 MiB publishes, -1.25 MiB publishes, and the last -0.25 MiB
takes the
+// live bytes to zero and publishes as well. A writer holds its arena for the
whole
+// build, so per-term swings never reach zero and this costs one publish per
finished
+// writer.
+class HeldBackBuildBytes {
+public:
+ HeldBackBuildBytes(doris::MemTracker* tracker, bool registered)
+ : tracker_(tracker), registered_(registered) {}
+ ~HeldBackBuildBytes() { publish(pending_.exchange(0,
std::memory_order_relaxed)); }
-MemoryReporter::ConsumeReleaseFn
snii_build_consume_release(BuildMemoryPopulation population) {
- auto* tracker = snii_build_mem_tracker();
- // MemTracker and the subset counter are both thread-safe atomics, which is
- // what MemoryReporter requires of this callback: it is invoked
- // concurrently, from Reservation destructors, and must not throw.
+ void add(int64_t delta) {
+ const int64_t pending = pending_.fetch_add(delta,
std::memory_order_relaxed) + delta;
+ // acq_rel, so that whoever brings live_ to zero sees every change
counted in
+ // live_ before it already in pending_ when it exchanges.
+ const int64_t live = live_.fetch_add(delta, std::memory_order_acq_rel)
+ delta;
+ // At zero only a non-empty remainder is published: a scratch
reservation that is
+ // taken and given back per term leaves pending_ at zero too, and must
not touch
+ // the shared counters.
+ if (std::abs(pending) >= kBuildBytesPublishStep || (live == 0 &&
pending != 0)) {
+ publish(pending_.exchange(0, std::memory_order_relaxed));
+ }
+ }
+
+private:
+ // MemTracker and the subset counter are both thread-safe atomics.
//
// The registered path updates two of them, which is NOT atomic as a pair
--
// and does not need to be. The decision layer reads only
// g_registered_build_bytes; the tracker is observation. Because no
consumer
// combines the two, there is no window in which they can disagree with
each
// other in a way anything can observe.
- const bool registered = population == BuildMemoryPopulation::kRegistered;
- return [tracker, registered](int64_t delta) {
+ void publish(int64_t delta) {
if (delta >= 0) {
- tracker->consume(delta);
+ tracker_->consume(delta);
} else {
- tracker->release(-delta);
+ tracker_->release(-delta);
}
- if (registered) {
+ if (registered_) {
g_registered_build_bytes.fetch_add(delta,
std::memory_order_relaxed);
}
- };
+ }
+
+ doris::MemTracker* const tracker_;
+ const bool registered_;
+ std::atomic<int64_t> pending_ {0};
+ std::atomic<int64_t> live_ {0};
+};
+} // namespace
+
+int64_t snii_registered_build_bytes() {
+ return g_registered_build_bytes.load(std::memory_order_relaxed);
+}
+
+MemoryReporter::ConsumeReleaseFn
snii_build_consume_release(BuildMemoryPopulation population) {
+ // pending_ is atomic because MemoryReporter requires this callback to be
safe
+ // to invoke concurrently; it is also invoked from Reservation destructors
and
+ // must not throw. It is shared because std::function must be copyable: the
+ // last copy to go publishes the remainder.
+ auto held_back = std::make_shared<HeldBackBuildBytes>(
+ snii_build_mem_tracker(), population ==
BuildMemoryPopulation::kRegistered);
+ return [held_back = std::move(held_back)](int64_t delta) {
held_back->add(delta); };
}
} // namespace doris::snii::writer
diff --git
a/be/src/storage/index/inverted/snii/writer/snii_build_memory_tracker.h
b/be/src/storage/index/inverted/snii/writer/snii_build_memory_tracker.h
index 4d80565ce04..f6e6443bc57 100644
--- a/be/src/storage/index/inverted/snii/writer/snii_build_memory_tracker.h
+++ b/be/src/storage/index/inverted/snii/writer/snii_build_memory_tracker.h
@@ -17,6 +17,8 @@
#pragma once
+#include <cstdint>
+
#include "storage/index/inverted/snii/writer/memory_reporter.h"
namespace doris {
@@ -37,7 +39,28 @@ enum class BuildMemoryPopulation {
kUnregistered,
};
-// Mirrors a reporter's live bytes into the process tracker.
+// A reporter holds its changes back from the two process-wide counters until
they
+// add up to this many bytes. Reservations move several times per term, so
writing
+// every change would have all flush threads bouncing the same cache lines: a
log
+// column of mostly unique terms spent over half of its flush CPU there. The
step has
+// to exceed what one term swings, for example 32 KiB of window scratch plus a
ZSTD
+// context.
+//
+// The price is that the counters can lag a reporter by less than one step
while its live
+// bytes are above zero. They are exact again whenever the live bytes are
zero, so a
+// finished writer leaves nothing behind even though IndexFileWriter keeps its
reporter
+// until the rowset writer closes. The reporter itself stays exact, so its
budget and
+// spill threshold do not change. The forced-spill decision reads the sum, so
it can start
+// late by the lag of the writers still building: at most 1 GiB with 100 flush
threads on
+// a 10-index table, against a share of at least 4 *
inverted_index_ram_buffer_size. A
+// writer below the step could not be a victim anyway, because forced spills
only pick
+// arenas above snii_forced_spill_min_arena_bytes (default 64 MiB), and the
process-level
+// backstops read real memory.
+inline constexpr int64_t kBuildBytesPublishStep = 1LL << 20;
+
+// Mirrors a reporter's live bytes into the process tracker, to within one
publish step
+// while they are above zero, and exactly whenever they are zero or the
reporter is
+// destroyed.
MemoryReporter::ConsumeReleaseFn
snii_build_consume_release(BuildMemoryPopulation population);
// Returns reclaimable build bytes from registered writers.
diff --git a/be/src/storage/index/inverted/snii/writer/spimi_term_buffer.cpp
b/be/src/storage/index/inverted/snii/writer/spimi_term_buffer.cpp
index 394adcfb5de..6cea37c7109 100644
--- a/be/src/storage/index/inverted/snii/writer/spimi_term_buffer.cpp
+++ b/be/src/storage/index/inverted/snii/writer/spimi_term_buffer.cpp
@@ -41,6 +41,7 @@
#include "storage/index/inverted/snii/writer/encoded_spill_run.h"
#include "storage/index/inverted/snii/writer/global_memory_limiter.h"
#include "storage/index/inverted/snii/writer/posting_external_sort.h"
+#include "storage/index/inverted/snii/writer/snii_build_memory_tracker.h"
#include "storage/index/inverted/snii/writer/spill_run_codec.h"
#include "storage/index/inverted/snii/writer/temp_dir.h"
@@ -346,8 +347,10 @@ void
SpimiTermBuffer::attach_global_limiter(GlobalMemoryLimiter* limiter) {
// moment. The buffer's total memory needs no reporting here: it already
// reaches the limiter through the SniiIndexBuild observation tracker that
// mem_reporter_ feeds.
+ reported_arena_ = pool_.arena_bytes();
+ decided_resident_ = static_cast<int64_t>(resident_bytes());
global_limiter_->register_buffer(&global_spill_requested_,
-
static_cast<int64_t>(pool_.arena_bytes()));
+ static_cast<int64_t>(reported_arena_));
}
void SpimiTermBuffer::report_arena_delta() {
@@ -375,9 +378,22 @@ void SpimiTermBuffer::report_arena_delta() {
}
// Report reclaimable arena bytes to the process-wide limiter. It may flag
// this buffer for a spill, which the owner thread handles after this call.
- if (global_limiter_ != nullptr) {
- global_limiter_->report(&global_spill_requested_,
- static_cast<int64_t>(pool_.arena_bytes()));
+ //
+ // The call takes a process-wide mutex, so it is made only when the
decision can
+ // change. The resident total above moves on every new term while the
arena moves
+ // one 32 KiB block at a time, so a log column that interns 4000 new terms
into one
+ // block would otherwise take that mutex 4000 times to repeat one number,
from every
+ // flush thread at once. The limiter keeps the arena, so a moved arena is
reported.
+ // Its decision also reads the build bytes, which reporters publish one
+ // kBuildBytesPublishStep at a time, so a resident move of that size is
reported too
+ // even inside one arena block: the term slots doubling from 2^17 to 2^18
entries add
+ // 3.5 MiB in one token, which has to be judged then and not at the next
block.
+ if (global_limiter_ != nullptr &&
+ (pool_.arena_bytes() != reported_arena_ ||
+ std::abs(now - decided_resident_) >= kBuildBytesPublishStep)) {
+ reported_arena_ = pool_.arena_bytes();
+ decided_resident_ = now;
+ global_limiter_->report(&global_spill_requested_,
static_cast<int64_t>(reported_arena_));
}
reported_resident_ = now;
}
diff --git a/be/src/storage/index/inverted/snii/writer/spimi_term_buffer.h
b/be/src/storage/index/inverted/snii/writer/spimi_term_buffer.h
index 4e1f20c1635..d891741ecc1 100644
--- a/be/src/storage/index/inverted/snii/writer/spimi_term_buffer.h
+++ b/be/src/storage/index/inverted/snii/writer/spimi_term_buffer.h
@@ -287,8 +287,9 @@ private:
// Reports the signed change in REAL resident bytes (resident_bytes()) to
// mem_reporter_ since the previous call, then caches the new total.
// Single-source diff: every grow/reset/free emits EXACTLY ONE delta
- // (self-balancing -> impossible to double-count or miss a negative).
No-op when
- // mem_reporter_ is null.
+ // (self-balancing -> impossible to double-count or miss a negative). An
attached
+ // global limiter is told only when the arena or the resident total moved
enough
+ // to change its decision (see the definition). No-op when both are null.
void report_arena_delta();
Status merge_runs_streamed(const StreamedTermConsumer& fn);
Status prepare_run_merge();
@@ -389,6 +390,10 @@ private:
// doubles as the buffer's registry identity.
GlobalMemoryLimiter* global_limiter_ = nullptr;
std::atomic<bool> global_spill_requested_ {false};
+ // The arena size and resident total of the limiter's last decision;
+ // report_arena_delta() stays off the registry mutex until one of them
moves.
+ uint64_t reported_arena_ = 0;
+ int64_t decided_resident_ = 0;
// Forced-spill floor / run-file cap (see the public setters above).
uint64_t forced_spill_min_arena_bytes_ = kDefaultForcedSpillMinArenaBytes;
size_t max_run_files_ = kDefaultMaxRunFilesPerBuffer;
diff --git a/be/test/storage/index/inverted/snii/snii_bkd_adapter_test.cpp
b/be/test/storage/index/inverted/snii/snii_bkd_adapter_test.cpp
index 6b8cbaac9b7..dffdfe30045 100644
--- a/be/test/storage/index/inverted/snii/snii_bkd_adapter_test.cpp
+++ b/be/test/storage/index/inverted/snii/snii_bkd_adapter_test.cpp
@@ -340,12 +340,13 @@ TEST_F(SniiBkdAdapterTest,
BuildMemoryIsChargedAndGivenBackAtFinish) {
SniiBkdIndexColumnWriter writer(&index_file_writer, &_meta, kFieldType);
assert_ok(writer.init());
- constexpr uint32_t kNulls = 200000;
+ constexpr uint32_t kNulls = 600000;
assert_ok(writer.add_nulls(kNulls));
std::vector<int64_t> values(1000);
std::iota(values.begin(), values.end(), 0);
assert_ok(writer.add_values(reinterpret_cast<const
uint8_t*>(values.data()), values.size()));
- // At least the 4 bytes of every NULL docid, and the builder's point
buffer on top.
+ // At least the 4 bytes of every NULL docid, and the builder's point
buffer on top. The
+ // NULL docids are one change larger than a publish step, so none of it is
held back.
EXPECT_GE(tracker->consumption() - baseline, static_cast<int64_t>(kNulls)
* 4);
// A BKD build cannot be reclaimed by a forced spill, so it is not offered
to the spill decision.
EXPECT_EQ(::doris::snii::writer::snii_registered_build_bytes(),
registered_baseline);
diff --git
a/be/test/storage/index/inverted/snii/writer/snii_build_memory_tracker_test.cpp
b/be/test/storage/index/inverted/snii/writer/snii_build_memory_tracker_test.cpp
index 35ebac5be49..ce265efa7ee 100644
---
a/be/test/storage/index/inverted/snii/writer/snii_build_memory_tracker_test.cpp
+++
b/be/test/storage/index/inverted/snii/writer/snii_build_memory_tracker_test.cpp
@@ -23,6 +23,8 @@
#include <cstdint>
#include <memory>
+#include <thread>
+#include <vector>
#include "common/status.h"
#include "runtime/memory/mem_tracker.h"
@@ -31,6 +33,10 @@
namespace doris::snii::writer {
namespace {
+// Changes of at least one step reach the shared counters at once, so tests
that are
+// about what those counters mean use whole steps.
+constexpr int64_t kStep = kBuildBytesPublishStep;
+
TEST(SniiBuildMemoryTracker, IsALabelledProcessSingleton) {
doris::MemTracker* tracker = snii_build_mem_tracker();
ASSERT_NE(tracker, nullptr);
@@ -46,16 +52,16 @@ TEST(SniiBuildMemoryTracker,
ReservationsMoveTheTrackerAndReturnToBaseline) {
MemoryReporter
reporter(snii_build_consume_release(BuildMemoryPopulation::kRegistered),
/*cap_bytes=*/0,
MemoryReporter::CapPolicy::kSpillThreshold);
MemoryReporter::Reservation reservation = reporter.make_reservation();
- ASSERT_TRUE(reservation.set_bytes(4096).ok());
- EXPECT_EQ(reporter.current_bytes(), 4096);
+ ASSERT_TRUE(reservation.set_bytes(kStep).ok());
+ EXPECT_EQ(reporter.current_bytes(), kStep);
EXPECT_EQ(tracker->consumption() - baseline, reporter.current_bytes());
- ASSERT_TRUE(reservation.set_bytes(64 * 1024).ok());
+ ASSERT_TRUE(reservation.set_bytes(4 * kStep).ok());
EXPECT_EQ(tracker->consumption() - baseline, reporter.current_bytes())
<< "the tracker must track the reporter's LIVE bytes, not its
peak";
// Shrinking is the half a leak would silently skip.
- ASSERT_TRUE(reservation.set_bytes(1024).ok());
+ ASSERT_TRUE(reservation.set_bytes(2 * kStep).ok());
EXPECT_EQ(tracker->consumption() - baseline, reporter.current_bytes());
}
EXPECT_EQ(tracker->consumption(), baseline)
@@ -66,11 +72,11 @@ TEST(SniiBuildMemoryTracker,
LegacyReportPathAlsoMovesTheTracker) {
doris::MemTracker* tracker = snii_build_mem_tracker();
const int64_t baseline = tracker->consumption();
MemoryReporter
reporter(snii_build_consume_release(BuildMemoryPopulation::kRegistered));
- reporter.report(1500);
- EXPECT_EQ(tracker->consumption() - baseline, 1500);
- reporter.report(500);
- EXPECT_EQ(tracker->consumption() - baseline, 2000);
- reporter.report(-2000);
+ reporter.report(3 * kStep);
+ EXPECT_EQ(tracker->consumption() - baseline, 3 * kStep);
+ reporter.report(kStep);
+ EXPECT_EQ(tracker->consumption() - baseline, 4 * kStep);
+ reporter.report(-4 * kStep);
EXPECT_EQ(tracker->consumption(), baseline);
}
@@ -84,13 +90,13 @@ TEST(SniiBuildMemoryTracker,
SeparateReportersAccumulateIntoOneLine) {
snii_build_consume_release(BuildMemoryPopulation::kRegistered));
auto merge = std::make_unique<MemoryReporter>(
snii_build_consume_release(BuildMemoryPopulation::kRegistered));
- ingestion->report(700);
- merge->report(300);
- EXPECT_EQ(tracker->consumption() - baseline, 1000);
+ ingestion->report(7 * kStep);
+ merge->report(3 * kStep);
+ EXPECT_EQ(tracker->consumption() - baseline, 10 * kStep);
- ingestion->report(-700);
- EXPECT_EQ(tracker->consumption() - baseline, 300);
- merge->report(-300);
+ ingestion->report(-7 * kStep);
+ EXPECT_EQ(tracker->consumption() - baseline, 3 * kStep);
+ merge->report(-3 * kStep);
EXPECT_EQ(tracker->consumption(), baseline);
}
@@ -105,15 +111,15 @@ TEST(SniiBuildMemoryTracker,
RegisteredPopulationIsTrackedSeparatelyFromTheObser
{
MemoryReporter
ingestion(snii_build_consume_release(BuildMemoryPopulation::kRegistered));
MemoryReporter
merge(snii_build_consume_release(BuildMemoryPopulation::kUnregistered));
- ingestion.report(700);
- merge.report(300);
+ ingestion.report(7 * kStep);
+ merge.report(3 * kStep);
// ONE observation line covers both: that is Part 1's goal.
- EXPECT_EQ(tracker->consumption() - baseline, 1000);
+ EXPECT_EQ(tracker->consumption() - baseline, 10 * kStep);
// What the decision layer judges, read as a single value: the
- // reclaimable population alone, with the merge's 300 never included.
- EXPECT_EQ(snii_registered_build_bytes() - registered_baseline, 700);
- ingestion.report(-700);
- merge.report(-300);
+ // reclaimable population alone, with the merge's bytes never included.
+ EXPECT_EQ(snii_registered_build_bytes() - registered_baseline, 7 *
kStep);
+ ingestion.report(-7 * kStep);
+ merge.report(-3 * kStep);
}
EXPECT_EQ(tracker->consumption(), baseline);
EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
@@ -129,8 +135,8 @@ TEST(SniiBuildMemoryTracker,
ReporterDestructorDrainsUnbalancedBytes) {
const int64_t baseline = tracker->consumption();
{
MemoryReporter
leaky(snii_build_consume_release(BuildMemoryPopulation::kRegistered));
- leaky.report(4096); // no matching report(-4096): the bug being
contained
- EXPECT_EQ(tracker->consumption() - baseline, 4096);
+ leaky.report(4 * kStep); // no matching report(-4 * kStep): the bug
being contained
+ EXPECT_EQ(tracker->consumption() - baseline, 4 * kStep);
}
EXPECT_EQ(tracker->consumption(), baseline)
<< "an unbalanced reporter must not leave residue in a
process-wide counter";
@@ -142,8 +148,8 @@ TEST(SniiBuildMemoryTracker,
ReporterDestructorDrainsTheRegisteredSubsetToo) {
const int64_t registered_baseline = snii_registered_build_bytes();
{
MemoryReporter
leaky(snii_build_consume_release(BuildMemoryPopulation::kRegistered));
- leaky.report(4096);
- EXPECT_EQ(snii_registered_build_bytes() - registered_baseline, 4096);
+ leaky.report(4 * kStep);
+ EXPECT_EQ(snii_registered_build_bytes() - registered_baseline, 4 *
kStep);
}
EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
}
@@ -153,11 +159,138 @@ TEST(SniiBuildMemoryTracker,
UnregisteredResidueNeverReachesTheDecisionInput) {
const int64_t registered_baseline = snii_registered_build_bytes();
{
MemoryReporter
leaky(snii_build_consume_release(BuildMemoryPopulation::kUnregistered));
- leaky.report(4096);
+ leaky.report(4 * kStep);
EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
}
EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
}
+// This is what keeps flush threads off each other's cache lines: a term
encoder takes
+// scratch and gives it back for every term, and that must never reach the
shared
+// counters. The reporter stays exact, because the hard budget of compaction
reads it.
+// Both populations: BKD writers and compaction merges run the same per-term
reservations
+// on the same flush threads as the text writers.
+TEST(SniiBuildMemoryTracker, PerTermSwingsStayOffTheSharedCounters) {
+ doris::MemTracker* tracker = snii_build_mem_tracker();
+ const int64_t baseline = tracker->consumption();
+ const int64_t registered_baseline = snii_registered_build_bytes();
+ constexpr int64_t kTermScratch = 32 * 1024;
+ for (BuildMemoryPopulation population :
+ {BuildMemoryPopulation::kRegistered,
BuildMemoryPopulation::kUnregistered}) {
+ MemoryReporter reporter(snii_build_consume_release(population));
+ for (int term = 0; term < 1000; ++term) {
+ MemoryReporter::Reservation scratch =
reporter.make_postings_reservation();
+ ASSERT_TRUE(scratch.set_bytes(kTermScratch).ok());
+ ASSERT_EQ(reporter.current_bytes(), kTermScratch);
+ ASSERT_EQ(tracker->consumption(), baseline);
+ ASSERT_EQ(snii_registered_build_bytes(), registered_baseline);
+ }
+ EXPECT_EQ(reporter.current_bytes(), 0);
+ }
+ EXPECT_EQ(tracker->consumption(), baseline);
+ EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
+}
+
+TEST(SniiBuildMemoryTracker, HeldBackBytesArePublishedTogetherAtTheStep) {
+ doris::MemTracker* tracker = snii_build_mem_tracker();
+ const int64_t baseline = tracker->consumption();
+ const int64_t registered_baseline = snii_registered_build_bytes();
+ MemoryReporter
reporter(snii_build_consume_release(BuildMemoryPopulation::kRegistered));
+
+ reporter.report(kStep - 1);
+ EXPECT_EQ(tracker->consumption(), baseline);
+ EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
+
+ reporter.report(1);
+ EXPECT_EQ(tracker->consumption() - baseline, kStep);
+ EXPECT_EQ(snii_registered_build_bytes() - registered_baseline, kStep);
+
+ // Publishing restarts the count, so one byte back must not bounce off the
step.
+ reporter.report(-1);
+ EXPECT_EQ(tracker->consumption() - baseline, kStep);
+
+ reporter.report(-(kStep - 1));
+ EXPECT_EQ(tracker->consumption(), baseline);
+ EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
+}
+
+// +1.5 steps and -1.25 steps both publish, and the last -0.25 steps never
reaches a step.
+// The reporter must not wait for its destructor to give that back:
IndexFileWriter keeps
+// a finished text writer's reporter until the rowset writer closes, so every
segment of a
+// load would leave its remainder in the spill decision's input. An empty
reporter
+// publishes what it holds back.
+TEST(SniiBuildMemoryTracker, AnEmptyReporterHoldsNothingBack) {
+ doris::MemTracker* tracker = snii_build_mem_tracker();
+ const int64_t baseline = tracker->consumption();
+ const int64_t registered_baseline = snii_registered_build_bytes();
+ auto retained = std::make_shared<MemoryReporter>(
+ snii_build_consume_release(BuildMemoryPopulation::kRegistered));
+ retained->report(kStep + kStep / 2);
+ retained->report(-(kStep + kStep / 4));
+ EXPECT_EQ(tracker->consumption() - baseline, kStep / 4);
+ retained->report(-kStep / 4);
+ EXPECT_EQ(retained->current_bytes(), 0);
+ EXPECT_EQ(tracker->consumption(), baseline)
+ << "a finished writer whose reporter is still retained left bytes
behind";
+ EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
+
+ // The same holds when the reporter goes back to zero through a
reservation: both steps
+ // publish, and only the zero rule can return the last quarter step.
+ {
+ MemoryReporter::Reservation reservation = retained->make_reservation();
+ ASSERT_TRUE(reservation.set_bytes(kStep + kStep / 2).ok());
+ ASSERT_TRUE(reservation.set_bytes(kStep / 4).ok());
+ EXPECT_EQ(tracker->consumption() - baseline, kStep / 4);
+ }
+ EXPECT_EQ(retained->current_bytes(), 0);
+ EXPECT_EQ(tracker->consumption(), baseline);
+ EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
+}
+
+// Unbalanced bytes still go back when the reporter is destroyed.
+TEST(SniiBuildMemoryTracker, DestroyingAReporterPublishesWhatItHeldBack) {
+ doris::MemTracker* tracker = snii_build_mem_tracker();
+ const int64_t baseline = tracker->consumption();
+ {
+ MemoryReporter
reporter(snii_build_consume_release(BuildMemoryPopulation::kRegistered));
+ reporter.report(kStep + kStep / 2);
+ reporter.report(-(kStep + kStep / 4));
+ reporter.report(kStep / 8);
+ EXPECT_EQ(tracker->consumption() - baseline, kStep / 4);
+ }
+ EXPECT_EQ(tracker->consumption(), baseline);
+}
+
+TEST(SniiBuildMemoryTracker, ConcurrentChangesOnOneReporterLoseNothing) {
+ doris::MemTracker* tracker = snii_build_mem_tracker();
+ const int64_t baseline = tracker->consumption();
+ const int64_t registered_baseline = snii_registered_build_bytes();
+ constexpr int kThreads = 8;
+ constexpr int64_t kChange = kStep / 3 + 7;
+ {
+ MemoryReporter
reporter(snii_build_consume_release(BuildMemoryPopulation::kRegistered));
+ std::vector<std::thread> threads;
+ for (int i = 0; i < kThreads; ++i) {
+ // Each thread leaves its own net on the reporter, so a lost
change would
+ // show as a wrong total and not just cancel out.
+ threads.emplace_back([&reporter] {
+ for (int n = 0; n < 2000; ++n) {
+ reporter.report(kChange);
+ reporter.report(n == 0 ? 0 : -kChange);
+ }
+ });
+ }
+ for (std::thread& thread : threads) {
+ thread.join();
+ }
+ EXPECT_EQ(reporter.current_bytes(), kThreads * kChange);
+ const int64_t published = tracker->consumption() - baseline;
+ EXPECT_LT(reporter.current_bytes() - published, kStep);
+ EXPECT_GT(reporter.current_bytes() - published, -kStep);
+ }
+ EXPECT_EQ(tracker->consumption(), baseline);
+ EXPECT_EQ(snii_registered_build_bytes(), registered_baseline);
+}
+
} // namespace
} // namespace doris::snii::writer
diff --git
a/be/test/storage/index/inverted/snii/writer/spimi_global_memory_limiter_test.cpp
b/be/test/storage/index/inverted/snii/writer/spimi_global_memory_limiter_test.cpp
index c9849c2a7c9..d47def95290 100644
---
a/be/test/storage/index/inverted/snii/writer/spimi_global_memory_limiter_test.cpp
+++
b/be/test/storage/index/inverted/snii/writer/spimi_global_memory_limiter_test.cpp
@@ -272,7 +272,9 @@ TEST(SniiGlobalMemoryLimiter,
UnregisteredChurnNeverPerturbsTheDecisionInput) {
const int64_t registered_baseline =
doris::snii::writer::snii_registered_build_bytes();
MemoryReporter ingestion(doris::snii::writer::snii_build_consume_release(
doris::snii::writer::BuildMemoryPopulation::kRegistered));
- ingestion.report(700); // the only registered bytes in flight
+ // The only registered bytes in flight, enough to reach the shared
counters at once.
+ constexpr int64_t kRegistered =
doris::snii::writer::kBuildBytesPublishStep;
+ ingestion.report(kRegistered);
std::atomic<bool> stop {false};
std::thread churn([&stop] {
@@ -300,7 +302,7 @@ TEST(SniiGlobalMemoryLimiter,
UnregisteredChurnNeverPerturbsTheDecisionInput) {
for (int i = 0; i < 200000 || !saw_churn; ++i) {
const int64_t observed =
doris::snii::writer::snii_registered_build_bytes() -
registered_baseline;
- ASSERT_EQ(observed, 700) << "unregistered churn leaked into the
decision's input";
+ ASSERT_EQ(observed, kRegistered) << "unregistered churn leaked into
the decision's input";
ASSERT_GE(doris::snii::writer::snii_registered_build_bytes(), 0)
<< "decision input went negative";
if (tracker->consumption() != tracker_baseline) {
@@ -315,7 +317,7 @@ TEST(SniiGlobalMemoryLimiter,
UnregisteredChurnNeverPerturbsTheDecisionInput) {
EXPECT_TRUE(saw_churn) << "churn thread never moved the observation
tracker; the invariant "
"above was never actually exercised";
- ingestion.report(-700);
+ ingestion.report(-kRegistered);
EXPECT_EQ(doris::snii::writer::snii_registered_build_bytes(),
registered_baseline);
}
@@ -749,6 +751,104 @@ TEST(SniiSpimiGlobalSpill,
AttachRegistersReportsTrackArenaAndDtorUnregisters) {
EXPECT_EQ(lim.eligible_arena_bytes(), 0);
}
+// A new term moves the resident total on every token, but the limiter keeps
only the
+// arena size, so terms that fit in the current block have nothing to tell it.
Taking
+// its mutex for them is what stalled every flush thread on a column of unique
terms.
+TEST(SniiSpimiGlobalSpill, NewTermsThatKeepTheArenaStayOffTheLimiter) {
+ std::atomic<int> decisions {0};
+ GlobalMemoryLimiter lim;
+ lim.set_signals_provider([&decisions] {
+ decisions.fetch_add(1, std::memory_order_relaxed);
+ return BuildMemorySignals {};
+ });
+ SpimiTermBuffer buf(/*has_positions=*/true, /*spill_threshold_bytes=*/0);
+ buf.attach_global_limiter(&lim);
+
+ buf.add_token(unigram(0), /*docid=*/0, /*pos=*/0); // allocates the first
block
+ ASSERT_TRUE(buf.status().ok());
+ const uint64_t arena = buf.arena_bytes_for_test();
+ const uint64_t resident = buf.resident_bytes_for_test();
+ const int decisions_for_first_block = decisions.load();
+ ASSERT_GT(decisions_for_first_block, 0) << "a grown arena must reach the
limiter";
+
+ for (uint32_t i = 1; i < 300; ++i) {
+ buf.add_token(unigram(i), /*docid=*/i, /*pos=*/0);
+ }
+ ASSERT_TRUE(buf.status().ok());
+ ASSERT_EQ(buf.arena_bytes_for_test(), arena) << "every term has to fit in
the first block";
+ ASSERT_GT(buf.resident_bytes_for_test(), resident)
+ << "without a moving resident total this would pass for the wrong
reason";
+ EXPECT_EQ(decisions.load(), decisions_for_first_block);
+}
+
+// The decision also reads the build bytes, which move one publish step at a
time, so a
+// resident move of a step reaches the limiter at once even inside one arena
block.
+// Capacity doublings of the term tables are such moves: one new term adds
MiBs while its
+// first posting still fits in the current block.
+TEST(SniiSpimiGlobalSpill, ResidentStepInsideOneArenaBlockReachesTheLimiter) {
+ std::atomic<int> decisions {0};
+ GlobalMemoryLimiter lim;
+ lim.set_signals_provider([&decisions] {
+ decisions.fetch_add(1, std::memory_order_relaxed);
+ return BuildMemorySignals {};
+ });
+ SpimiTermBuffer buf(/*has_positions=*/true, /*spill_threshold_bytes=*/0);
+ buf.attach_global_limiter(&lim);
+
+ bool checked = false;
+ for (uint32_t i = 0; i < (1U << 21) && !checked; ++i) {
+ const uint64_t arena = buf.arena_bytes_for_test();
+ const uint64_t resident = buf.resident_bytes_for_test();
+ const int before = decisions.load();
+ buf.add_token("t" + std::to_string(i), /*docid=*/i, /*pos=*/0);
+ if (buf.arena_bytes_for_test() == arena &&
+ buf.resident_bytes_for_test() >=
+ resident + doris::snii::writer::kBuildBytesPublishStep) {
+ // One add_token can report twice (interning and accumulating), so
this
+ // asks only that the step was decided at this term.
+ EXPECT_GT(decisions.load(), before)
+ << "the resident step at term " << i << " was not decided";
+ checked = true;
+ }
+ }
+ ASSERT_TRUE(buf.status().ok());
+ ASSERT_TRUE(checked) << "no term grew the resident total by a step inside
one arena block";
+}
+
+// Skipping unchanged arenas must not lose the reaction to pressure: pressure
that
+// arrives while a block is filling is decided as soon as the arena grows, and
the
+// flag is then honored like any other request.
+TEST(SniiSpimiGlobalSpill, PressureIsDecidedWhenTheArenaNextGrows) {
+ snii_testing::reset_global_forced_spills();
+ FakeSignals signals;
+ GlobalMemoryLimiter lim;
+ lim.set_signals_provider(signals.provider());
+ lim.set_min_victim_arena_bytes(1);
+ SpimiTermBuffer buf(/*has_positions=*/true, /*spill_threshold_bytes=*/0);
+ buf.set_forced_spill_min_arena_bytes(1);
+ buf.attach_global_limiter(&lim);
+
+ buf.add_token("hot", /*docid=*/0, /*pos=*/0);
+ ASSERT_TRUE(buf.status().ok());
+ const uint64_t first_block = buf.arena_bytes_for_test();
+ ASSERT_FALSE(buf.global_spill_requested_for_test());
+
+ signals.set_overage(/*overage=*/1);
+ uint32_t docid = 1;
+ while (buf.arena_bytes_for_test() == first_block &&
buf.run_count_for_test() == 0) {
+ buf.add_token("hot", docid, /*pos=*/0);
+ ++docid;
+ ASSERT_LT(docid, 1000000U) << "the arena never grew";
+ }
+ ASSERT_TRUE(buf.status().ok());
+ // report_arena_delta() runs before the gate reads its request, and a
request it
+ // flags for this token is honored by the next one.
+ buf.add_token("hot", docid, /*pos=*/0);
+ ASSERT_TRUE(buf.status().ok());
+ EXPECT_EQ(buf.run_count_for_test(), 1U) << "the limiter never saw the
pressure";
+ EXPECT_EQ(snii_testing::global_forced_spills(), 1U);
+}
+
// End-to-end: two attached buffers, one small and one that grows. The limiter
// must flag the larger-ARENA grower once its arena clears the victim floor
(the
// small buffer's single arena block never does); the grower's own next token
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
