diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index a4cdc41fd9a..8d516e82d93 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -24,7 +24,7 @@ repos: hooks: - id: check-added-large-files # Retain the exact lossless cross-platform performance evidence archives. - exclude: ^docs/benchmarks/((executor-copy-audit|bounded-read-ahead)-20261101|(directory-task-slots|packed-visit-snapshots|bounded-match-results|useful-match-workers|direct-evaluator-workers)-20261102)\.json\.gz$ + exclude: ^docs/benchmarks/((executor-copy-audit|bounded-read-ahead)-20261101|(directory-task-slots|packed-visit-snapshots|bounded-match-results|useful-match-workers|direct-evaluator-workers|derived-content-handoff)-20261102)\.json\.gz$ - id: check-merge-conflict - id: check-yaml # BINARY_TYPES below: archive containers are binary but can START with text (a phar begins with diff --git a/CHANGELOG.md b/CHANGELOG.md index 71ee552ce57..ae745662d03 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,8 @@ # 0.11.0 +- Reuse requested hashes and line counts from content already loaded by workers, without retaining file bytes between stages. + - Limit warmed matcher participation to available work chunks while retaining useful read concurrency and persistent workers. - Reuse bounded matcher output buffers between batches while clearing stale selection/error state and releasing oversized old output. diff --git a/TODO.md b/TODO.md index 20b1db41f7f..6f5a87c2fcb 100644 --- a/TODO.md +++ b/TODO.md @@ -412,7 +412,11 @@ EP/JC/P work; candidates are not speedup claims until measured on both platforms retain real allocation savings alongside unacceptable Linux comparison regressions. A deeper typed-plan representation remains unqualified; EP07 boundaries still apply. Preserve existing stage policy while isolating task storage and lifecycle costs. -- [ ] O13-O22: Reconcile remaining derived-value handoff, static catalogs, invariant operands, +- [x] O13: Qualify requested derived-value handoff from matching workers to ordered consumers. + [Cross-platform evidence](docs/benchmarks/derived-content-handoff-20261102.md) retains + read-count, lifetime/error/limit tests, large field-workload gains and slower controls. + Transfer requested hashes/line values only; keep raw bytes confined to each evaluator. +- [ ] O14-O22: Reconcile remaining static catalogs, invariant operands, paired streaming compare, reusable stage pools, byte-bounded overlap, archive indexing/admission, optional bulk metadata, similarity representations and output churn with EP/JC/P acceptance. Keep macrobenchmark, kernel and diagnostic contracts separate. diff --git a/docs/benchmarks/derived-content-handoff-20261102.json.gz b/docs/benchmarks/derived-content-handoff-20261102.json.gz new file mode 100644 index 00000000000..031f5c5f93e Binary files /dev/null and b/docs/benchmarks/derived-content-handoff-20261102.json.gz differ diff --git a/docs/benchmarks/derived-content-handoff-20261102.md b/docs/benchmarks/derived-content-handoff-20261102.md new file mode 100644 index 00000000000..880ae4060cc --- /dev/null +++ b/docs/benchmarks/derived-content-handoff-20261102.md @@ -0,0 +1,72 @@ +# Reuse requested content values between stages (2026-11-02) + +## Decision + +Retain the request-driven derived-cache handoff. Compiled field renderers identify required hashes and line counts once per run, including defaults and transformed fields. Histogram enums add line demand. Matched/selected workers derive only those values from content already loaded by their evaluator. Move requested digest map nodes and line values into an optional result-indexed snapshot vector; keep raw content in the worker. Ordinary filters retain their result layout and allocate no derived-cache vector. + +The existing per-entry snapshot already reuses reads within one evaluator. The gap was the worker-to-coordinator boundary: the coordinator could reread content for fields and reductions after filtering. Gains combine reuse of those bytes and moving requested digest/line computation onto existing workers. They must not be attributed entirely to fewer copies. No thread, clock, stat, shared mutable cache, generic wrapper or user flag is added. + +Keep streaming sources without loaded snapshots on the lazy coordinator path. Preserve result-limit laziness, selected/error states, empty/unmatched/regrowing batches, binary line handling, transforms/defaults, archive serial admission and effect invalidation. Cached missing values remain distinct from values not requested. A returned derived snapshot has no raw content; later unrequested consumers can still read lazily. + +## Measurement contract + +Measured parent: `bcfef58c06979254b2081b6f02b1040821464b5b` (#1052, documentation only). Baseline and `derived_values` use clean `clang_release`/ThinLTO executables copied before timing. Runtime source identities, binary hashes, fixture data, scripts and all raw samples are retained. Later support-test refinements do not alter the timed runtime. Compilation, lint and sanitizers do not overlap timing on a host. + +Each host has 50 native cases in three sessions: warmup discarded per case/variant/session, nine rotating rounds and fastest-seven means. Independent output oracles validate every invocation; hash/line expectations are computed from actual fixture bytes before timing. Workloads use 10/1,000/10,000 files plus 272/288/320-entry fixtures with 256 short entries followed by expensive 128 KiB entries. Requested worker counts are one/three/ten. Linux uses tmpfs and CPU affinity; macOS uses warm APFS without affinity. + +Twenty-six whole-engine in-memory cells run three sessions and nine repetitions/fastest seven, with separate processes and rotating participant order. Content fields, output/filter/traversal and comparison controls use one/four workers. Four native cases receive five-session confirmation and five kernels receive three-session confirmation, each adding a byte-identical baseline participant. Confirmation never replaces initial slower samples. + +## Native results + +Median paired session changes; negative means faster. + +| Fixture / task | Workers | Linux | macOS | +| :----------------------- | ------: | ------: | ------: | +| flat10000 / derived-hash | 1 | +0.60% | -0.10% | +| flat10000 / derived-hash | 3 | -36.22% | -48.53% | +| flat10000 / derived-hash | 10 | -43.18% | -49.17% | +| heavy320 / derived-lines | 1 | -0.35% | +0.78% | +| heavy320 / derived-lines | 3 | -26.57% | -35.04% | +| heavy320 / derived-lines | 10 | -19.76% | -34.55% | +| heavy320 / derived-hash | 3 | -57.77% | -56.91% | +| heavy320 / derived-hash | 10 | -72.61% | -73.62% | +| flat10000 / files | 1 | +1.55% | +0.12% | +| compare-large / compare | 3 | -2.55% | +0.69% | + +One-worker cases stay near baseline; large parallel hash cases improve substantially on both hosts. These are selected workloads, not a claim that every search becomes faster. The largest initial native median costs are +1.55% Linux and +0.78% macOS, with session reversals retained. + +## Kernel results and confirmation + +| Kernel | Initial Linux | Confirm Linux | Baseline repeat Linux | Initial macOS | Confirm macOS | Baseline repeat macOS | +| :------------------------- | ------------: | ------------: | --------------------: | ------------: | ------------: | --------------------: | +| RE2/content-fields/1000/4 | -67.84% | -66.75% | -0.07% | -71.82% | -72.19% | -0.14% | +| RE2/content-fields/10000/1 | +0.09% | +0.08% | +0.05% | +0.94% | -1.67% | +0.04% | +| RE2/content-fields/10000/4 | -70.14% | -70.00% | +0.14% | -73.01% | -72.80% | +0.19% | +| RE2/large-output/10000/1 | -0.54% | +0.26% | +0.14% | +6.63% | -0.52% | +2.66% | +| RE2/rg-files/10000/4 | +0.25% | +2.03% | +1.58% | +1.47% | -2.31% | -3.37% | + +Retain the initial macOS 10,000-file/one-worker large-output control (+6.63%) and its targeted confirmation above. That output path requests no derived values, so the exact cause must not be assigned to saved hashing work. The byte-identical baseline participant helps expose noise but does not eliminate it. All slower samples and unrelated shifts remain in the archive. + +## Native confirmation + +| Fixture / task | Workers | Linux candidate | Linux baseline repeat | macOS candidate | macOS baseline repeat | +| :----------------------- | ------: | --------------: | --------------------: | --------------: | --------------------: | +| heavy320 / derived-hash | 10 | -72.38% | +0.48% | -73.52% | +0.24% | +| flat10000 / files | 1 | +2.07% | +0.35% | +0.28% | +0.02% | +| flat10000 / derived-hash | 1 | +0.70% | +0.55% | +0.51% | -0.00% | +| heavy320 / derived-lines | 1 | +0.32% | +0.03% | +0.82% | +0.13% | + +## Correctness, ownership and footprint + +The native field test processes 128 files, including unselected entries, and observes exactly 128 VFS reads with serial and parallel execution while repeated digest/line fields produce identical output. A result limit preserves the lazy path and its single consumed output read. Snapshot tests prove that returned caches retain no bytes, do not initiate reads for unloaded sources, move only requested values and retain binary/read-failure null results. Pool tests cover unmatched, empty and inline regrowth without stale values. Dedicated parallel column, summary, line-bucket histogram, line-sum histogram and count-only histogram cases match serial output and observe one read per file. Rg summary tests exercise rejected content matches and read-error equivalence. + +Derived storage is bounded by the current batch population and requested cache values, rather than total file bytes. It is not a new total-process memory cap. Hash strings/map nodes and snapshot metadata still use memory; ordinary searches requesting no such values allocate no side vector. No first-output or total-RSS improvement is inferred from elapsed time alone. + +- linux: unstripped CLI output 12,292,800 -> 12,334,504 bytes (+41,704). +- macos: unstripped CLI output 4,051,424 -> 4,069,744 bytes (+18,320). + +Both complete core/extension release suites pass 306/306, with required Linux FUSE mounts. Focused ASan passes five suites per host, and Linux TSan passes three worker/run suites. Header-aware local clang-tidy covers 42 affected translation units; initial test const-qualification findings are fixed and rechecked separately. Final normal hooks and current pushed-head CI remain required before merge. + +## Lossless evidence + +[derived-content-handoff-20261102.json.gz](derived-content-handoff-20261102.json.gz): 1,176,635 bytes; SHA-256 `ad5f377bacbac56bf45aa504038db70f40893324d648e55c67580a1e1a478550`. Resolve fixtures through `fixtures[report.fixtures_ref]`; timed and final support sources are separate, and initial failure/confirmation logs remain distinct. diff --git a/docs/design-parallel.md b/docs/design-parallel.md index f30423b9d26..f8103ebdec0 100644 --- a/docs/design-parallel.md +++ b/docs/design-parallel.md @@ -68,7 +68,11 @@ A bounded directory-read worker pool with a single coordinator: does not disable host-file parallelism: mounted/archive-member entries flush pending host results and use the serial evaluator. Both paths keep the same VFS and safety preflight. - Workers own each entry's content snapshot and rendered output. Predicates and match output reuse - that snapshot; source bytes are released when evaluation finishes. The batch retains rendered + that snapshot; source bytes are released when evaluation finishes. When templates or histograms + request hashes or line counts, selected workers derive those values from bytes already loaded and + move only requested caches to the ordered coordinator. A separate result-indexed vector exists + only when values are requested; raw content never crosses this boundary. Streaming sources without + loaded snapshots and result-limited listings keep lazy coordinator reads. The batch retains rendered records for ordered emission. Full line/context output buffers each input; filename/count/quiet selection streams where the VFS supports it, retaining the longest line and draining the source for binary/error checks. Neither the entry-count batch bound nor streaming limits record size. diff --git a/docs/performance-analysis.md b/docs/performance-analysis.md index 850fc3b16e1..dfc34068feb 100644 --- a/docs/performance-analysis.md +++ b/docs/performance-analysis.md @@ -3323,3 +3323,21 @@ virtual calls alone do not establish an acceptable elapsed improvement. Preserve the rejected source rather than adding this representation to the shipping runtime. A deeper typed plan needs a separate lifetime-safe design and fresh profiles; O13 and the remaining candidates continue independently. + +## Requested content values across the stage boundary + +[The O13 report and raw evidence](benchmarks/derived-content-handoff-20261102.md) qualify requested +derived caches from content already loaded by matching workers. Existing snapshots already cache +within one evaluator; this removes the later coordinator reread and computes requested digests/line +counts using existing workers. Compiled fields and histogram enums define demand once per run. +Only requested derived values move to ordered consumers, not raw file bytes or a shared cache. +Ordinary filters allocate no side vector and keep their result layout. Result limits and streaming +sources without loaded snapshots retain the lazy path. + +Four-worker content-field kernels confirm roughly 67-70% lower elapsed time on Linux and 72-73% +on macOS for the larger populations. Heavy native ten-worker hash cases improve about 73-74%. +One-worker cases stay near baseline. These gains combine avoided reads/copies with parallel derived +work, rather than proving that copies alone accounted for that time. The report keeps the initial +macOS output control (+6.63%) and its confirmation (-0.52%), identical-baseline participants, all +slower samples, source identities and validation. This does not replace O14/O15 invariant/catalog +work, O17/O18 scheduling or O22 output construction. diff --git a/release-site.json b/release-site.json index 2f54a16c003..5b4daf96501 100644 --- a/release-site.json +++ b/release-site.json @@ -70,7 +70,8 @@ "docs/benchmarks/packed-visit-snapshots-20261102.md": "docs/benchmarks/packed-visit-snapshots-20261102.html", "docs/benchmarks/bounded-match-results-20261102.md": "docs/benchmarks/bounded-match-results-20261102.html", "docs/benchmarks/useful-match-workers-20261102.md": "docs/benchmarks/useful-match-workers-20261102.html", - "docs/benchmarks/direct-evaluator-workers-20261102.md": "docs/benchmarks/direct-evaluator-workers-20261102.html" + "docs/benchmarks/direct-evaluator-workers-20261102.md": "docs/benchmarks/direct-evaluator-workers-20261102.html", + "docs/benchmarks/derived-content-handoff-20261102.md": "docs/benchmarks/derived-content-handoff-20261102.html" }, "generated_html": { "XFF.md": { diff --git a/xff/content/snapshot.cc b/xff/content/snapshot.cc index 8f1a4cf8aa1..233095f0b13 100644 --- a/xff/content/snapshot.cc +++ b/xff/content/snapshot.cc @@ -3,10 +3,22 @@ #include "xff/content/snapshot.h" +#include +#include + #include "xff/content/line_match.h" namespace xff::content { +void Derivation::Include(Derivation other) { + lines |= other.lines; + for (auto& digest : other.digests) { + if (!std::ranges::contains(digests, digest)) { + digests.push_back(std::move(digest)); + } + } +} + absl::StatusOr Snapshot::Read(const vfs::FileSystem& fs, std::string_view path) const { if (!bytes_) { bytes_ = fs.ReadContent(path); @@ -21,6 +33,13 @@ std::optional Snapshot::Digest( const vfs::FileSystem& fs, std::string_view path, const hash::AlgoEncoding& spec) const { + return DigestValue(fs, path, spec); +} + +const std::optional& Snapshot::DigestValue( + const vfs::FileSystem& fs, + std::string_view path, + const hash::AlgoEncoding& spec) const { auto [entry, inserted] = digests_.try_emplace(std::pair{std::string(spec.algo), spec.encoding}); if (inserted) { const auto bytes = Read(fs, path); @@ -39,6 +58,23 @@ std::optional Snapshot::Lines(const vfs::FileSystem& fs, std::strin return *lines_; } +Snapshot Snapshot::TakeDerived(const Derivation& request, const vfs::FileSystem& fs, std::string_view path) && { + Snapshot derived; + if (!Loaded()) { + return derived; + } + if (request.lines) { + static_cast(Lines(fs, path)); + derived.lines_ = std::exchange(lines_, std::nullopt); + } + for (const auto& digest : request.digests) { + static_cast(DigestValue(fs, path, {.algo = digest.algorithm, .encoding = digest.encoding})); + auto node = digests_.extract(std::pair{digest.algorithm, digest.encoding}); + derived.digests_.insert(std::move(node)); + } + return derived; +} + void Snapshot::Invalidate() { bytes_.reset(); digests_.clear(); diff --git a/xff/content/snapshot.h b/xff/content/snapshot.h index 2fc2d82ad27..6f52456f925 100644 --- a/xff/content/snapshot.h +++ b/xff/content/snapshot.h @@ -10,6 +10,7 @@ #include #include #include +#include #include "absl/status/statusor.h" #include "xff/hash/hash.h" @@ -17,6 +18,22 @@ namespace xff::content { +// Immutable after run preparation. Algorithm names own their bytes independently of templates. +struct DigestRequest { + std::string algorithm; + hash::Encoding encoding = hash::Encoding::kHex; + bool operator==(const DigestRequest&) const = default; +}; + +struct Derivation { + bool lines = false; + std::vector digests; + + bool Empty() const { return !lines && digests.empty(); } + + void Include(Derivation other); +}; + // Lazily owned bytes and requested derived values for exactly one entry. A snapshot belongs // to one evaluator; it has no locks and is never shared between workers. Read errors are cached. // Invalidate at effect boundaries and before using a different entry. Views expire on invalidation. @@ -28,9 +45,18 @@ class Snapshot final { std::optional Lines(const vfs::FileSystem& fs, std::string_view path) const; void Invalidate(); + // Extract only requested derived caches from already loaded entry bytes. This never initiates + // a VFS read and the returned snapshot owns no raw content, including on a cached read error. + Snapshot TakeDerived(const Derivation& request, const vfs::FileSystem& fs, std::string_view path) &&; + bool Loaded() const { return bytes_.has_value(); } private: + const std::optional& DigestValue( + const vfs::FileSystem& fs, + std::string_view path, + const hash::AlgoEncoding& spec) const; + mutable std::optional> bytes_; mutable std::map, std::optional> digests_; mutable std::optional> lines_; diff --git a/xff/content/snapshot_test.cc b/xff/content/snapshot_test.cc index 51b00d216ca..0dc423fb09f 100644 --- a/xff/content/snapshot_test.cc +++ b/xff/content/snapshot_test.cc @@ -20,6 +20,7 @@ namespace { using ::mbo::testing::EqualsText; using ::mbo::testing::IsOkAndHolds; using ::mbo::testing::StatusIs; +using ::testing::ElementsAre; using ::testing::Eq; using ::testing::IsFalse; using ::testing::IsTrue; @@ -59,6 +60,64 @@ struct SnapshotTest : ::testing::Test { Snapshot snapshot; }; +TEST_F(SnapshotTest, DerivedTransferRetainsRequestedValuesWithoutRetainingBytes) { + EXPECT_THAT(snapshot.Read(fs, "file"), IsOkAndHolds(EqualsText(fs.bytes))); + const auto digest = hash::HashData("md5", fs.bytes).value_or("unsupported"); + Derivation request{.lines = true, .digests = {{.algorithm = "md5"}}}; + request.Include({.digests = {{.algorithm = "md5"}}}); + EXPECT_THAT(request.digests, ElementsAre(DigestRequest{.algorithm = "md5"})); + auto derived = std::move(snapshot).TakeDerived(request, fs, "file"); + EXPECT_THAT(derived.Loaded(), IsFalse()); + fs.bytes = "replacement"; + EXPECT_THAT(derived.Lines(fs, "file"), Optional(2)); + EXPECT_THAT(derived.Digest(fs, "file", {.algo = "md5"}), Optional(EqualsText(digest))); + EXPECT_THAT(fs.reads, Eq(1)); + // Unrequested byte consumers still read lazily; no old raw content crossed the boundary. + EXPECT_THAT(derived.Read(fs, "file"), IsOkAndHolds(EqualsText("replacement"))); + EXPECT_THAT(fs.reads, Eq(2)); + derived.Invalidate(); + EXPECT_THAT(derived.Lines(fs, "file"), Optional(1)); + EXPECT_THAT(fs.reads, Eq(3)); +} + +TEST_F(SnapshotTest, UnrequestedCachesAreNotTransferredAndBinaryLineFailuresRemainCached) { + EXPECT_THAT( + snapshot.Digest(fs, "file", {.algo = "sha256"}), + Optional(EqualsText(hash::HashData("sha256", fs.bytes).value_or("unsupported")))); + const auto derived = std::move(snapshot).TakeDerived({.digests = {{.algorithm = "md5"}}}, fs, "file"); + EXPECT_THAT(derived.Loaded(), IsFalse()); + fs.bytes = "replacement"; + EXPECT_THAT( + derived.Digest(fs, "file", {.algo = "sha256"}), + Optional(EqualsText(hash::HashData("sha256", fs.bytes).value_or("unsupported")))); + EXPECT_THAT(fs.reads, Eq(2)); + Snapshot binary; + fs.bytes = std::string("one\0two", 7); + EXPECT_THAT(binary.Read(fs, "file"), IsOkAndHolds(EqualsText(fs.bytes))); + const auto binary_derived = std::move(binary).TakeDerived({.lines = true}, fs, "file"); + EXPECT_THAT(binary_derived.Loaded(), IsFalse()); + fs.bytes = "now text"; + EXPECT_THAT(binary_derived.Lines(fs, "file"), Eq(std::nullopt)); + EXPECT_THAT(fs.reads, Eq(3)); +} + +TEST_F(SnapshotTest, DerivationWithoutLoadedBytesDoesNotReadAndCachedFailuresStayFailures) { + const auto unloaded = std::move(snapshot).TakeDerived({.lines = true}, fs, "file"); + EXPECT_THAT(unloaded.Loaded(), IsFalse()); + EXPECT_THAT(fs.reads, Eq(0)); + EXPECT_THAT(unloaded.Lines(fs, "file"), Optional(2)); + EXPECT_THAT(fs.reads, Eq(1)); + Snapshot failing; + fs.fail = true; + EXPECT_THAT(failing.Read(fs, "file"), StatusIs(absl::StatusCode::kPermissionDenied)); + const auto failed = std::move(failing).TakeDerived({.lines = true, .digests = {{.algorithm = "md5"}}}, fs, "file"); + fs.fail = false; + EXPECT_THAT(failed.Loaded(), IsFalse()); + EXPECT_THAT(failed.Lines(fs, "file"), Eq(std::nullopt)); + EXPECT_THAT(failed.Digest(fs, "file", {.algo = "md5"}), Eq(std::nullopt)); + EXPECT_THAT(fs.reads, Eq(2)); +} + TEST_F(SnapshotTest, ReusesBytesAndRequestedDerivedValuesUntilInvalidated) { EXPECT_THAT(snapshot.Loaded(), IsFalse()); for (int repeat = 0; repeat < 2; ++repeat) { diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel index 3c376a39ecb..0fc597ab718 100644 --- a/xff/engine/BUILD.bazel +++ b/xff/engine/BUILD.bazel @@ -372,6 +372,7 @@ cc_library( ":evaluate_cc", ":expression_execution_cc", ":owned_visit_cc", + "//xff/content:snapshot_cc", "//xff/fuzzy:fuzzy_cc", "//xff/parser:parser_cc", "//xff/registry:registry_cc", diff --git a/xff/engine/parallel_match.cc b/xff/engine/parallel_match.cc index bd375ef4732..69775c08182 100644 --- a/xff/engine/parallel_match.cc +++ b/xff/engine/parallel_match.cc @@ -36,12 +36,14 @@ ParallelMatch::ParallelMatch( std::size_t workers, bool scores, std::optional output, - mbo::types::OptionalRef execution) + mbo::types::OptionalRef execution, + content::Derivation derivation) : expression_(expression), workers_(std::max(workers, std::size_t{1})), scores_(scores), output_(std::move(output)), - execution_(execution) {} + execution_(execution), + derivation_(std::move(derivation)) {} ParallelMatch::~ParallelMatch() { { @@ -53,6 +55,10 @@ ParallelMatch::~ParallelMatch() { } } +ContentSnapshot ParallelMatch::TakeDerived(std::size_t index) { + return derivation_.Empty() ? ContentSnapshot{} : std::move(derived_.at(index)); +} + bool ParallelMatch::Ready(std::size_t generation, std::size_t worker) const { return stop_ || (generation_ != generation && worker < active_workers_); } @@ -65,6 +71,10 @@ const std::vector& ParallelMatch::Match(std::vector const absl::MutexLock lock(mutex_); entries_ = std::move(entries); results_.resize(entries_.size()); + if (!derivation_.Empty()) { + derived_.clear(); + derived_.resize(entries_.size()); + } if (output_ && !results_.empty()) { const auto limit = std::min(kReusableRecordBytes, kReusableOutputBytes / results_.size()); // Release oversized old records before evaluating any new entry. Current output can still @@ -204,6 +214,10 @@ void ParallelMatch::EvaluateEntry( } else { result.content.reset(); } + if (!derivation_.Empty() && result.evaluation.matched + && (!result.content || (result.content->selected.ok() && *result.content->selected))) { + derived_.at(index) = std::move(context.content).TakeDerived(derivation_, *visit.fs, visit.path); + } } } // namespace xff::engine diff --git a/xff/engine/parallel_match.h b/xff/engine/parallel_match.h index cc2f720fe0a..d3e2165ab71 100644 --- a/xff/engine/parallel_match.h +++ b/xff/engine/parallel_match.h @@ -17,6 +17,7 @@ #include "absl/status/statusor.h" #include "absl/synchronization/mutex.h" #include "mbo/types/optional_ref.h" +#include "xff/content/snapshot.h" #include "xff/engine/evaluate.h" #include "xff/engine/expression_execution.h" #include "xff/engine/owned_visit.h" @@ -63,7 +64,8 @@ class ParallelMatch final { std::size_t workers, bool scores, std::optional output = std::nullopt, - mbo::types::OptionalRef execution = {}); + mbo::types::OptionalRef execution = {}, + content::Derivation derivation = {}); ~ParallelMatch(); ParallelMatch(const ParallelMatch&) = delete; ParallelMatch& operator=(const ParallelMatch&) = delete; @@ -76,6 +78,9 @@ class ParallelMatch final { const std::vector& Entries() const { return entries_; } + // Consume one ordered entry's derived caches. No snapshot storage exists without a request. + ContentSnapshot TakeDerived(std::size_t index); + private: static constexpr std::size_t kChunkSize = 4; void Run(std::size_t worker); @@ -99,6 +104,7 @@ class ParallelMatch final { const bool scores_; const std::optional output_; const mbo::types::OptionalRef execution_; + const content::Derivation derivation_; std::optional coordinator_; fuzzy::ScoreWorkspace coordinator_fuzzy_workspace_; std::vector threads_; @@ -111,6 +117,7 @@ class ParallelMatch final { // Published under mutex_, immutable until all workers have completed the batch. std::vector entries_; std::vector results_; + std::vector derived_; std::atomic next_ = 0; }; diff --git a/xff/engine/rg_test.cc b/xff/engine/rg_test.cc index dba92ace8c0..fb63ad2cbc9 100644 --- a/xff/engine/rg_test.cc +++ b/xff/engine/rg_test.cc @@ -549,6 +549,103 @@ TEST_F(RgEngineTest, OrdinaryRgSearchUsesContentWorkersWithoutNativeExpression) EXPECT_THAT(output, HasSubstr("tree/299:hit 299\n")); } +TEST_F(RgEngineTest, ParallelFieldRenderingReusesOnlyRequestedDerivedValues) { + fs.files.clear(); + for (std::size_t index = 0; index < 128; ++index) { + fs.files.emplace("tree/" + std::to_string(index), index % 4 == 0 ? "miss\n" : "hit\nsecond\n"); + } + const std::string field = "--template={path}:{hash:md5}:{hash:md5}:{lines}:{lines}"; + EXPECT_THAT(Run({"--jobs=1", field, "tree", "-type", "f", "-content", "hit"}, false).errors, Eq(0)); + const auto serial = output; + EXPECT_THAT(fs.reads.load(), Eq(128)); + fs.reads = 0; + fs.track_threads = true; + fs.expected_threads = 4; + EXPECT_THAT(Run({"--jobs=4", field, "tree", "-type", "f", "-content", "hit"}, false).errors, Eq(0)); + EXPECT_THAT(output, EqualsText(serial)); + EXPECT_THAT(fs.reads.load(), Eq(128)); + EXPECT_THAT(fs.read_threads.size(), Gt(1)); + fs.reads = 0; + EXPECT_THAT( + Run({"--jobs=4", "--max-results=1", field, "tree", "-type", "f", "-content", "hit"}, false).errors, Eq(0)); + // A limit retains coordinator laziness instead of deriving hashes for unrendered matches. + EXPECT_THAT(fs.reads.load(), Eq(129)); +} + +TEST_F(RgEngineTest, ParallelColumnsAndLineHistogramsReuseLoadedContent) { + fs.files.clear(); + for (std::size_t index = 0; index < 128; ++index) { + fs.files.emplace("tree/" + std::to_string(index), index % 2 == 0 ? "hit\nsecond\n" : "miss\n"); + } + const std::vector> controls{ + {"--columns=path,lines", "--format=csv"}, {"--summary={lines}", "--format=jsonl"}, + {"--histogram=lines", "--format=jsonl"}, {"--histogram=ext:sum(lines)", "--format=jsonl"}, + {"--histogram=ext", "--format=jsonl"}, + }; + for (const auto& output_controls : controls) { + auto arguments = output_controls; + arguments.insert(arguments.end(), {"--jobs=1", "tree", "-type", "f", "-content", "hit"}); + fs.reads = 0; + EXPECT_THAT(Run(arguments, false).errors, Eq(0)); + const auto serial = output; + EXPECT_THAT(fs.reads.load(), Eq(128)); + arguments.at(output_controls.size()) = "--jobs=4"; + fs.reads = 0; + EXPECT_THAT(Run(arguments, false).errors, Eq(0)); + EXPECT_THAT(output, EqualsText(serial)); + EXPECT_THAT(fs.reads.load(), Eq(128)); + } +} + +TEST_F(RgEngineTest, DerivedSummariesRespectRgSelectionAndReadFailures) { + fs.files.clear(); + for (std::size_t index = 0; index < 128; ++index) { + fs.files.emplace("tree/" + std::to_string(index), index % 2 == 0 ? "hit needle\nsecond\n" : "hit other\n"); + } + EXPECT_THAT( + Run({"--jobs=1", "--summary={lines}", "--format=jsonl", "needle", "tree", "--xff", "-content", "hit"}).errors, + Eq(0)); + const auto serial = output; + EXPECT_THAT( + Run({"--jobs=4", "--summary={lines}", "--format=jsonl", "needle", "tree", "--xff", "-content", "hit"}).errors, + Eq(0)); + EXPECT_THAT(output, EqualsText(serial)); + fs.read_error_path = "tree/0"; + // Rg encounters the failure itself when no native content predicate reads first. + const auto serial_failure = Run({"--jobs=1", "--summary={lines}", "--format=jsonl", "needle", "tree"}); + const auto failed_output = output; + const auto serial_errors = errors; + EXPECT_THAT(serial_failure.errors, Gt(0)); + const auto parallel_failure = Run({"--jobs=4", "--summary={lines}", "--format=jsonl", "needle", "tree"}); + EXPECT_THAT(parallel_failure.errors, Eq(serial_failure.errors)); + EXPECT_THAT(output, EqualsText(failed_output)); + EXPECT_THAT(errors, Eq(serial_errors)); +} + +TEST_F(RgEngineTest, DerivedPoolSlotsClearAcrossUnmatchedEmptyAndInlineBatches) { + fs.files["tree/a"] = "hit\nsecond\nthird\n"; + MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"tree", "-content", "hit"})); + ParallelMatch matcher(*command.expression, 4, false, std::nullopt, {}, {.lines = true}); + EXPECT_THAT(matcher.Match(RepeatedEntries(128)), SizeIs(128)); + const auto first = matcher.TakeDerived(0); + EXPECT_THAT(first.Loaded(), IsFalse()); + EXPECT_THAT(first.Lines(fs, "tree/a"), Optional(3)); + EXPECT_THAT(fs.reads.load(), Eq(128)); + fs.files["tree/a"] = "miss\n"; + const auto& unmatched = matcher.Match(RepeatedEntries(128)); + EXPECT_THAT(unmatched.front().evaluation.matched, IsFalse()); + const auto second = matcher.TakeDerived(0); + EXPECT_THAT(second.Lines(fs, "tree/a"), Optional(1)); + EXPECT_THAT(fs.reads.load(), Eq(257)); + EXPECT_THAT(matcher.Match({}), IsEmpty()); + fs.files["tree/a"] = "hit\nlast\n"; + EXPECT_THAT(matcher.Match(RepeatedEntries(1)), SizeIs(1)); + const auto last = matcher.TakeDerived(0); + EXPECT_THAT(last.Loaded(), IsFalse()); + EXPECT_THAT(last.Lines(fs, "tree/a"), Optional(2)); + EXPECT_THAT(fs.reads.load(), Eq(258)); +} + TEST_F(RgEngineTest, MatcherPoolGrowsForLaterBatchesAndRunsShortTailsInline) { fs.files.clear(); for (std::size_t index = 0; index < 256; ++index) { diff --git a/xff/engine/run.cc b/xff/engine/run.cc index 6c3389c8c3d..01b10741e32 100644 --- a/xff/engine/run.cc +++ b/xff/engine/run.cc @@ -64,6 +64,7 @@ #include "xff/cli/globals.h" #include "xff/config/safety.h" #include "xff/content/line_match.h" +#include "xff/content/snapshot.h" #include "xff/datetime/datetime.h" #include "xff/engine/collect.h" #include "xff/engine/consumers.h" @@ -6046,9 +6047,33 @@ RunResult RunFindCore( content_output->grep.quiet = true; } } + content::Derivation derivation; + // Preserve lazy work for outputs that may never be consumed. A worker only derives from + // bytes it already loaded; streaming and serial/archive paths keep their existing reads. + if (!max_results->has_value()) { + const auto encoding = hash::ParseEncoding(hash_encoding).value_or(hash::Encoding::kHex); + if (any_reduction && !shards.enabled && !collections.Active()) { + for (const auto& field : summary_templates) { + if (field) { + derivation.Include(field->ContentDerivation(hash_algorithm, encoding)); + } + } + derivation.lines |= std::ranges::any_of(histograms, [](const HistogramSpec& spec) { + return spec.bucket == HistBucket::kLinesRange + || (spec.agg != HistAgg::kCount && spec.metric == HistMetric::kLines); + }); + } else if (implicit_print && !is_tree && !match_output && !(rg_output && grep_options.match_output)) { + if (column_templates.empty() && compiled_tmpl) { + derivation.Include(compiled_tmpl->ContentDerivation(hash_algorithm, encoding)); + } + for (const auto& field : column_templates) { + derivation.Include(field.ContentDerivation(hash_algorithm, encoding)); + } + } + } parallel_match.emplace( parallel_expression, options.workers, full_metadata || rank_by_score, std::move(content_output), - executions->Parallel()); + executions->Parallel(), std::move(derivation)); } std::vector pending_matches; // Decision-only batches retain no file bytes or rendered line records. Give them more work @@ -6083,7 +6108,7 @@ RunResult RunFindCore( } std::map outputs; finish_entry( - visit, outputs, evaluated.matched, std::nullopt, {}, + visit, outputs, evaluated.matched, std::nullopt, parallel_match->TakeDerived(index), result.content ? mbo::types::OptionalRef{*result.content} : mbo::types::OptionalRef{}); } diff --git a/xff/presentation/fields/BUILD.bazel b/xff/presentation/fields/BUILD.bazel index 5a49c422903..8b540a2c938 100644 --- a/xff/presentation/fields/BUILD.bazel +++ b/xff/presentation/fields/BUILD.bazel @@ -49,6 +49,7 @@ cc_test( srcs = ["fields_test.cc"], deps = [ ":fields_cc", + "//xff/content:snapshot_cc", "//xff/env:env_cc", "//xff/vfs:vfs_cc", "@abseil-cpp//absl/time", diff --git a/xff/presentation/fields/fields.cc b/xff/presentation/fields/fields.cc index cf7fe9b46e4..482e0d8c023 100644 --- a/xff/presentation/fields/fields.cc +++ b/xff/presentation/fields/fields.cc @@ -1275,6 +1275,23 @@ hash::DefaultUsage Template::HashDefaultsUsed() const { return usage; } +content::Derivation Template::ContentDerivation(std::string_view algorithm, hash::Encoding encoding) const { + content::Derivation request; + for (const Segment& segment : segments_) { + if (segment.fn == &LinesField) { + request.lines = true; + } else if (segment.fn == &HashField) { + const std::string_view qualifier = + segment.post == Segment::PostProcess::kNone ? segment.qualifier : std::string_view{}; + const auto spec = hash::ParseSpec(qualifier, algorithm.empty() ? "sha256" : algorithm, encoding); + if (spec) { + request.Include({.digests = {{.algorithm = std::string(spec->algo), .encoding = spec->encoding}}}); + } + } + } + return request; +} + std::size_t Template::ContentFieldCount() const { return static_cast(absl::c_count_if( segments_, [](const Segment& segment) { return segment.fn == &HashField || segment.fn == &LinesField; })); diff --git a/xff/presentation/fields/fields.h b/xff/presentation/fields/fields.h index 76ec7150878..1d2df309308 100644 --- a/xff/presentation/fields/fields.h +++ b/xff/presentation/fields/fields.h @@ -160,6 +160,9 @@ class Template { // This describes template structure, not observed reads or bytes. std::size_t ContentFieldCount() const; + // Resolve requested derived values once, including hash defaults and transformed fields. + content::Derivation ContentDerivation(std::string_view algorithm, hash::Encoding encoding) const; + // Conservative object/record/text capacity budget. String capacity is counted even // when inline; copied templates each count shared transforms. Excludes regex backend, // diagnostic status payload, allocator and shared-ownership bookkeeping allocations. diff --git a/xff/presentation/fields/fields_test.cc b/xff/presentation/fields/fields_test.cc index 9bdf817b1e9..7ae47f64a96 100644 --- a/xff/presentation/fields/fields_test.cc +++ b/xff/presentation/fields/fields_test.cc @@ -30,6 +30,7 @@ #include "gtest/gtest.h" #include "mbo/testing/matchers.h" #include "mbo/testing/status.h" +#include "xff/content/snapshot.h" #include "xff/env/env.h" #include "xff/vfs/entry.h" @@ -155,6 +156,27 @@ TEST_F(FieldsTest, TimeZoneDemandIncludesTransformedTimeFields) { EXPECT_THAT(Template::Compile("{{mtime}} {name} {size} {text}").NeedsTimeZone(), IsFalse()); } +TEST_F(FieldsTest, ContentDerivationUsesCompiledFieldsDefaultsAndTransforms) { + const auto request = + Template::Compile("{{hash}} {name} {hash} {hash:sha256} {hash:md5/base64} {lines:s/2/two/} {hash:s/a/b/}") + .ContentDerivation("sha256", hash::Encoding::kHex); + EXPECT_THAT(request.lines, IsTrue()); + EXPECT_THAT( + request.digests, ElementsAre( + content::DigestRequest{.algorithm = "sha256"}, + content::DigestRequest{.algorithm = "md5", .encoding = hash::Encoding::kBase64})); + EXPECT_THAT( + Template::Compile("{{lines}} {name} {text}").ContentDerivation("", hash::Encoding::kHex).Empty(), IsTrue()); +} + +TEST_F(FieldsTest, ContentDerivationKeepsDefaultEncodingAndSkipsInvalidHashFields) { + EXPECT_THAT( + Template::Compile("{hash}").ContentDerivation("", hash::Encoding::kBase64).digests, + ElementsAre(content::DigestRequest{.algorithm = "sha256", .encoding = hash::Encoding::kBase64})); + EXPECT_THAT( + Template::Compile("{hash:not-an-algorithm}").ContentDerivation("", hash::Encoding::kHex).Empty(), IsTrue()); +} + TEST_F(FieldsTest, ContentFieldCountInspectsCompiledFieldsWithoutRendering) { EXPECT_THAT(Template::Compile("{{hash}} {name} {text}").ContentFieldCount(), Eq(0U)); EXPECT_THAT(Template::Compile("{hash} {hash:sha256/base64} {lines}").ContentFieldCount(), Eq(3U));