From 37840530d11899f32c847890b9ee2093a7155e04 Mon Sep 17 00:00:00 2001 From: helly25 Date: Fri, 9 Oct 2026 21:35:47 +0100 Subject: [PATCH 1/4] Delay directory workers until useful fan-out --- xff/engine/walk.cc | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/xff/engine/walk.cc b/xff/engine/walk.cc index 011657e3e70..329a8b08e56 100644 --- a/xff/engine/walk.cc +++ b/xff/engine/walk.cc @@ -71,6 +71,12 @@ struct Stated { // The result of reading one directory: its children, or a ReadDir error. using Listing = absl::StatusOr>; +// Creating and synchronizing the directory-read pool has a fixed cost. A small sibling fan-out +// cannot repay it on native memory-backed fixtures; wait for enough independent directory reads +// to keep the workers useful. This is a work-unit threshold rather than a file-count threshold, +// so layout shape remains part of the decision and narrow/deep walks stay serial. +inline constexpr std::size_t kMinParallelDirectoryReads = 64; + // A fixed pool of worker threads running leaf directory or stat jobs. // Workers touch only their job's inputs and the (thread-safe) FileSystem and the // task queue; they never call back into the walk, so no job can wait on another @@ -589,7 +595,7 @@ class Walker { directories.push_back(index); } } - if (directories.size() < 2) { + if (directories.size() < kMinParallelDirectoryReads) { return reads; } pool_.Start(directories.size()); From 5b54236bdaf1470cf957e3184ddefe93a729659a Mon Sep 17 00:00:00 2001 From: helly25 Date: Fri, 9 Oct 2026 23:29:48 +0100 Subject: [PATCH 2/4] Reuse one executor across run phases --- xff/engine/BUILD.bazel | 24 ++++++ xff/engine/parallel_match.cc | 136 +++++++++++++++++------------- xff/engine/parallel_match.h | 19 ++--- xff/engine/parallel_match_test.cc | 32 +++++-- xff/engine/rg_test.cc | 4 +- xff/engine/run.cc | 12 +-- xff/engine/run_executor.cc | 47 +++++++++++ xff/engine/run_executor.h | 66 +++++++++++++++ xff/engine/run_executor_test.cc | 33 ++++++++ xff/engine/walk.cc | 128 ++++++++-------------------- xff/engine/walk.h | 19 +++++ 11 files changed, 345 insertions(+), 175 deletions(-) create mode 100644 xff/engine/run_executor.cc create mode 100644 xff/engine/run_executor.h create mode 100644 xff/engine/run_executor_test.cc diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel index 7b1252abb3b..28dd26f21ac 100644 --- a/xff/engine/BUILD.bazel +++ b/xff/engine/BUILD.bazel @@ -19,6 +19,28 @@ load("//xff:cc.bzl", "cc_binary", "cc_library", "cc_test") package(default_visibility = ["//visibility:private"]) +cc_library( + name = "run_executor_cc", + srcs = ["run_executor.cc"], + hdrs = ["run_executor.h"], + visibility = ["//xff:__subpackages__"], + deps = [ + "@abseil-cpp//absl/base:core_headers", + "@abseil-cpp//absl/synchronization", + ], +) + +cc_test( + name = "run_executor_test", + size = "small", + srcs = ["run_executor_test.cc"], + deps = [ + ":run_executor_cc", + "@googletest//:gtest", + "@googletest//:gtest_main", + ], +) + # Directory traversal over a vfs::FileSystem: a worker pool reads (readdir+lstat) # in parallel while a single coordinator thread visits in --sort order. cc_library( @@ -27,6 +49,7 @@ cc_library( hdrs = ["walk.h"], visibility = ["//xff:__subpackages__"], deps = [ + ":run_executor_cc", "//xff/vfs:vfs_cc", "@abseil-cpp//absl/algorithm:container", "@abseil-cpp//absl/base:core_headers", @@ -248,6 +271,7 @@ cc_library( ":collect_cc", ":evaluate_cc", ":expression_execution_cc", + ":run_executor_cc", "//xff/parser:parser_cc", "//xff/registry:registry_cc", "@abseil-cpp//absl/base:core_headers", diff --git a/xff/engine/parallel_match.cc b/xff/engine/parallel_match.cc index 62afcff0ce5..25dbaa0b046 100644 --- a/xff/engine/parallel_match.cc +++ b/xff/engine/parallel_match.cc @@ -4,6 +4,8 @@ #include "xff/engine/parallel_match.h" #include +#include +#include #include #include #include @@ -12,6 +14,29 @@ #include "xff/registry/descriptor.h" namespace xff::engine { +namespace { + +// A cold executor has a fixed creation and teardown cost. Accumulate enough matcher work to +// repay it; once another run phase has started the executor, ordinary per-batch admission applies. +inline constexpr std::size_t kMinColdParallelEntries = 8'192; +inline constexpr std::size_t kColdProbeEntries = 16; +inline constexpr auto kSlowColdProbe = std::chrono::microseconds(500); + +} // namespace + +struct ParallelMatch::WorkerState { + explicit WorkerState(const ParallelMatch& owner) + : output(owner.output_ ? std::optional(ForkMatchOutput(owner.output_->output)) : std::nullopt), + evaluator(owner.execution_ ? std::optional(owner.execution_->MakeWorker()) : std::nullopt) { + if (owner.expression_ && (!owner.execution_ || !owner.execution_->UsesIndexedMatchers())) { + matchers.Bind(*owner.expression_); + } + } + + std::optional> output; + WorkerMatchers matchers; + std::optional evaluator; +}; bool CanParallelMatch(const parser::Expr& expression) { if (expression.kind == parser::Expr::Kind::kPredicate) { @@ -34,92 +59,89 @@ bool HasContentMatch(const parser::Expr& expression) { ParallelMatch::ParallelMatch( mbo::types::OptionalRef expression, std::size_t workers, + RunExecutor& executor, bool scores, std::optional output, mbo::types::OptionalRef execution) : expression_(expression), workers_(std::max(workers, std::size_t{1})), + executor_(executor), scores_(scores), output_(std::move(output)), execution_(execution) {} -ParallelMatch::~ParallelMatch() { - { - const absl::MutexLock lock(mutex_); - stop_ = true; - } - for (auto& thread : threads_) { - thread.join(); - } -} - -bool ParallelMatch::Ready(std::size_t generation) const { - return stop_ || generation_ != generation; -} - -bool ParallelMatch::Finished() const { - return remaining_ == 0; -} +ParallelMatch::~ParallelMatch() = default; const std::vector& ParallelMatch::Match(std::vector entries) { - const absl::MutexLock lock(mutex_); entries_ = std::move(entries); results_.clear(); results_.resize(entries_.size()); next_.store(0, std::memory_order_relaxed); - // Very small batches cannot amortize waking the pool. - if (workers_ == 1 || entries_.size() < 16 || (threads_.empty() && entries_.size() < 64)) { + const bool cold = executor_.worker_count() == 0; + const auto ensure_coordinator = [&] { if (execution_ && !coordinator_) { coordinator_.emplace(execution_->MakeWorker(ExpressionWorkerRole::kCoordinator)); } + }; + const auto evaluate_coordinator = [&] { EvaluateEntries( {}, {}, coordinator_ ? mbo::types::OptionalRef{*coordinator_} : mbo::types::OptionalRef{}); + }; + // Very small batches cannot amortize waking the executor. + if (workers_ == 1 || entries_.size() < 16) { + ensure_coordinator(); + evaluate_coordinator(); + serial_entries_ += entries_.size(); return results_; } - const std::size_t count = std::min(workers_, (entries_.size() + 15) / 16); - // An early partial batch must not permanently cap the pool for later full batches. - if (threads_.size() < count) { - threads_.reserve(count); - for (std::size_t index = threads_.size(); index < count; ++index) { - threads_.emplace_back([this] { Run(); }); + std::size_t parallel_entries = entries_.size(); + if (cold && serial_entries_ + entries_.size() < kMinColdParallelEntries) { + // A count threshold protects fast local storage, but must not suppress useful concurrency on + // slow filesystems. Measure a small prefix, then either parallelize the rest of this batch or + // finish inline and retain the evidence for later bounded batches. + ensure_coordinator(); + const std::size_t probe = std::min(kColdProbeEntries, entries_.size()); + const auto started = std::chrono::steady_clock::now(); + for (std::size_t index = 0; index < probe; ++index) { + results_.at(index) = EvaluateEntry( + entries_.at(index).AsVisit(), {}, {}, + coordinator_ ? mbo::types::OptionalRef{*coordinator_} + : mbo::types::OptionalRef{}); } + next_.store(probe, std::memory_order_relaxed); + if (std::chrono::steady_clock::now() - started < kSlowColdProbe) { + evaluate_coordinator(); + serial_entries_ += entries_.size(); + return results_; + } + parallel_entries -= probe; + } + const std::size_t count = std::min(workers_, (parallel_entries + 15) / 16); + executor_.Start(count); + while (worker_states_.size() < count) { + worker_states_.push_back(std::make_unique(*this)); + } + std::vector> futures; + futures.reserve(count); + for (std::size_t worker = 0; worker < count; ++worker) { + futures.push_back(executor_.Submit([this, worker] { EvaluateWorker(worker); })); + } + for (std::future& future : futures) { + future.get(); } - remaining_ = threads_.size(); - ++generation_; - mutex_.Await(absl::Condition(this, &ParallelMatch::Finished)); return results_; } -void ParallelMatch::Run() { - const auto local = output_ ? std::optional(ForkMatchOutput(output_->output)) : std::nullopt; - WorkerMatchers matchers; - if (expression_ && (!execution_ || !execution_->UsesIndexedMatchers())) { - matchers.Bind(*expression_); - } - const auto evaluator = execution_ ? std::optional(execution_->MakeWorker()) : std::nullopt; - std::size_t generation = 0; - for (;;) { - { - const auto ready = [&] ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) { return Ready(generation); }; - const absl::MutexLock lock(mutex_, absl::Condition(&ready)); - if (stop_) { - return; - } - generation = generation_; - } - EvaluateEntries( - local ? mbo::types::OptionalRef>{*local} - : mbo::types::OptionalRef>{}, - matchers, - evaluator ? mbo::types::OptionalRef{*evaluator} - : mbo::types::OptionalRef{}); - { - const absl::MutexLock lock(mutex_); - --remaining_; - } - } +void ParallelMatch::EvaluateWorker(std::size_t worker) { + const WorkerState& state = *worker_states_.at(worker); + EvaluateEntries( + state.output ? mbo::types::OptionalRef>{*state.output} + : mbo::types::OptionalRef>{}, + state.matchers, + state.evaluator ? mbo::types::OptionalRef{*state.evaluator} + : mbo::types::OptionalRef{}); } void ParallelMatch::EvaluateEntries( diff --git a/xff/engine/parallel_match.h b/xff/engine/parallel_match.h index 75a4dfaa391..13a3197164c 100644 --- a/xff/engine/parallel_match.h +++ b/xff/engine/parallel_match.h @@ -9,17 +9,15 @@ #include #include #include -#include #include #include -#include "absl/base/thread_annotations.h" #include "absl/status/statusor.h" -#include "absl/synchronization/mutex.h" #include "mbo/types/optional_ref.h" #include "xff/engine/collect.h" #include "xff/engine/evaluate.h" #include "xff/engine/expression_execution.h" +#include "xff/engine/run_executor.h" #include "xff/parser/ast.h" namespace xff::engine { @@ -56,6 +54,7 @@ class ParallelMatch final { ParallelMatch( mbo::types::OptionalRef expression, std::size_t workers, + RunExecutor& executor, bool scores, std::optional output = std::nullopt, mbo::types::OptionalRef execution = {}); @@ -72,9 +71,8 @@ class ParallelMatch final { const std::vector& Entries() const { return entries_; } private: - void Run(); - bool Ready(std::size_t generation) const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - bool Finished() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + struct WorkerState; + void EvaluateWorker(std::size_t worker); void EvaluateEntries( mbo::types::OptionalRef> output = {}, mbo::types::OptionalRef matchers = {}, @@ -87,19 +85,16 @@ class ParallelMatch final { const mbo::types::OptionalRef expression_; const std::size_t workers_; + RunExecutor& executor_; const bool scores_; const std::optional output_; const mbo::types::OptionalRef execution_; std::optional coordinator_; - std::vector threads_; - absl::Mutex mutex_; - bool stop_ ABSL_GUARDED_BY(mutex_) = false; - std::size_t generation_ ABSL_GUARDED_BY(mutex_) = 0; - std::size_t remaining_ ABSL_GUARDED_BY(mutex_) = 0; - // Published under mutex_, immutable until all workers have completed the batch. + std::vector> worker_states_; std::vector entries_; std::vector results_; std::atomic next_ = 0; + std::size_t serial_entries_ = 0; }; } // namespace xff::engine diff --git a/xff/engine/parallel_match_test.cc b/xff/engine/parallel_match_test.cc index 9e270d590b2..106c11337c2 100644 --- a/xff/engine/parallel_match_test.cc +++ b/xff/engine/parallel_match_test.cc @@ -48,7 +48,8 @@ TEST_F(ParallelMatchTest, RepeatedBatchesRetainEntryOrderAndDoNotReuseOldResults MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-type", "f"})); EXPECT_THAT(CanParallelMatch(*command.expression), IsTrue()); EXPECT_THAT(HasContentMatch(*command.expression), IsFalse()); - ParallelMatch matcher(*command.expression, 4, false); + RunExecutor executor(4); + ParallelMatch matcher(*command.expression, 4, executor, false); for (const std::size_t count : {10, 1'000, 100, 10, 10'000}) { const auto& results = matcher.Match(Entries(count)); EXPECT_THAT(results, SizeIs(count)); @@ -62,8 +63,10 @@ TEST_F(ParallelMatchTest, RepeatedBatchesRetainEntryOrderAndDoNotReuseOldResults TEST_F(ParallelMatchTest, ZeroWorkersIsClampedAndUnusedPoolStartsNoWork) { MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-true"})); - const ParallelMatch unused(*command.expression, 4, false); - ParallelMatch matcher(*command.expression, 0, true); + RunExecutor unused_executor(4); + const ParallelMatch unused(*command.expression, 4, unused_executor, false); + RunExecutor executor(0); + ParallelMatch matcher(*command.expression, 0, executor, true); EXPECT_THAT(matcher.Match(Entries(100)), SizeIs(100)); EXPECT_THAT(matcher.Match(Entries(0)), SizeIs(0)); } @@ -74,7 +77,8 @@ TEST_F(ParallelMatchTest, NativeRegexBindingsPreserveOperatorsCaseAndRepeatedBat parser::Parse({"root", "(", "-iregex", "[02468]", "-o", "-regex", "1[0-9]", ")", "!", "-regex", "12"})); parser::BindMatchers(command, regex::Grammar::kRe2, parser::CaseMode::kSensitive); ASSERT_THAT(CanParallelMatch(*command.expression), IsTrue()); - ParallelMatch matcher(*command.expression, 4, false); + RunExecutor executor(4); + ParallelMatch matcher(*command.expression, 4, executor, false); for (const std::size_t count : {10, 1'000, 100, 10}) { const auto& results = matcher.Match(Entries(count)); for (std::size_t index = 0; index < count; ++index) { @@ -87,7 +91,8 @@ TEST_F(ParallelMatchTest, NativeRegexBindingsPreserveOperatorsCaseAndRepeatedBat TEST_F(ParallelMatchTest, UnboundRegexIsANonMatchInWorkers) { MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-regex", "foo"})); - ParallelMatch matcher(*command.expression, 4, false); + RunExecutor executor(4); + ParallelMatch matcher(*command.expression, 4, executor, false); for (const auto& result : matcher.Match(Entries(100))) { EXPECT_THAT(result.evaluation.matched, IsFalse()); } @@ -97,10 +102,11 @@ TEST_F(ParallelMatchTest, PreparedCoordinatorAndWorkersRetainRegexStateAcrossPoo ASSERT_OK_AND_ASSIGN(auto command, parser::Parse({"root", "-regex", "[02468]+"})); parser::BindMatchers(command, regex::Grammar::kRe2, parser::CaseMode::kSensitive); ASSERT_OK_AND_ASSIGN(const auto execution, PrepareExpressionExecution(*command.expression)); - ParallelMatch matcher(*command.expression, 4, false, std::nullopt, execution); + RunExecutor executor_pool(4); + ParallelMatch matcher(*command.expression, 4, executor_pool, false, std::nullopt, execution); // Small batches run on the coordinator, large batches activate private worker state; // returning to the coordinator must not reuse a worker's previous match or scratch. - for (const std::size_t count : {1, 128, 7, 512, 0, 10}) { + for (const std::size_t count : {1, 128, 7, 512, 4'096, 10}) { const auto& results = matcher.Match(Entries(count)); ASSERT_THAT(results, SizeIs(count)); for (std::size_t index = 0; index < count; ++index) { @@ -110,6 +116,18 @@ TEST_F(ParallelMatchTest, PreparedCoordinatorAndWorkersRetainRegexStateAcrossPoo } } +TEST_F(ParallelMatchTest, ColdExecutorWaitsForAccumulatedWork) { + MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-true"})); + RunExecutor executor(4); + ParallelMatch matcher(*command.expression, 4, executor, false); + for (int batch = 0; batch < 7; ++batch) { + EXPECT_THAT(matcher.Match(Entries(1'024)), SizeIs(1'024)); + EXPECT_THAT(executor.worker_count(), Eq(0)); + } + EXPECT_THAT(matcher.Match(Entries(1'024)), SizeIs(1'024)); + EXPECT_THAT(executor.worker_count(), Eq(4)); +} + TEST_F(ParallelMatchTest, StatefulMetadataAndActionExpressionsCannotEnterWorkers) { const std::vector> expressions{ {"root", "-first", "1"}, diff --git a/xff/engine/rg_test.cc b/xff/engine/rg_test.cc index 7a74ee20ace..de5bffe1df8 100644 --- a/xff/engine/rg_test.cc +++ b/xff/engine/rg_test.cc @@ -454,7 +454,9 @@ TEST_F(RgEngineTest, MatcherPoolGrowsForLaterBatchesAndRunsShortTailsInline) { return result; }; MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"tree", "-content", "hit"})); - ParallelMatch matcher(*command.expression, 12, false); + RunExecutor executor(12); + executor.Start(12); + ParallelMatch matcher(*command.expression, 12, executor, false); fs.track_threads = true; for (const auto& [count, workers] : {std::pair{64UZ, 4UZ}, {256UZ, 12UZ}, {8UZ, 1UZ}}) { fs.expected_threads = workers; diff --git a/xff/engine/run.cc b/xff/engine/run.cc index d3409762ff6..592fd29f859 100644 --- a/xff/engine/run.cc +++ b/xff/engine/run.cc @@ -5885,6 +5885,7 @@ RunResult RunFindCore( serial_execution ? std::optional(serial_execution->MakeWorker(ExpressionWorkerRole::kCoordinator)) : std::nullopt; const auto output_evaluator = output_execution ? std::optional(output_execution->MakeWorker(ExpressionWorkerRole::kCoordinator)) : std::nullopt; + RunExecutor run_executor(options.workers > 1 ? options.workers : std::size_t{0}); std::optional parallel_match; if (use_match_pool) { std::optional content_output; @@ -5907,13 +5908,14 @@ RunResult RunFindCore( } } parallel_match.emplace( - parallel_expression, options.workers, full_metadata || rank_by_score, std::move(content_output), + parallel_expression, options.workers, run_executor, full_metadata || rank_by_score, std::move(content_output), executions->Parallel()); } std::vector pending_matches; - // Decision-only batches retain no file bytes or rendered line records. Give them more work - // between coordinator barriers; keep content-output batches smaller because a line is unbounded. - const std::size_t match_batch_size = rg_output || match_output ? 256 : 1'024; + // Decision-only batches retain no file bytes or rendered line records. A cold matcher needs one + // substantial batch to repay executor startup; keep content-output batches smaller because a + // rendered line is unbounded. Slow bounded-output storage is admitted adaptively by ParallelMatch. + const std::size_t match_batch_size = rg_output || match_output ? 256 : 8'192; const auto flush_matches = [&] { if (pending_matches.empty()) { return; @@ -5949,7 +5951,7 @@ RunResult RunFindCore( } }; const absl::Status status = Walk( - walk_fs, roots, options, + walk_fs, roots, options, run_executor, // NOLINTNEXTLINE(readability-function-cognitive-complexity): cohesive dispatch [&](const Visit& visit) { const auto rg_included = [&] { return command.rg && rg_glob_decision(visit) == ignore::Decision::kInclude; }; diff --git a/xff/engine/run_executor.cc b/xff/engine/run_executor.cc new file mode 100644 index 00000000000..70bbaa8b2d7 --- /dev/null +++ b/xff/engine/run_executor.cc @@ -0,0 +1,47 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#include "xff/engine/run_executor.h" + +#include + +namespace xff::engine { + +RunExecutor::~RunExecutor() { + { + const absl::MutexLock lock(mutex_); + stop_ = true; + } + for (std::thread& thread : threads_) { + thread.join(); + } +} + +void RunExecutor::Start(std::size_t useful_workers) { + const std::size_t count = std::min(max_workers_, useful_workers); + threads_.reserve(count); + while (threads_.size() < count) { + threads_.emplace_back([this] { Run(); }); + } +} + +bool RunExecutor::Pending() const { + return stop_ || !queue_.empty(); +} + +void RunExecutor::Run() { + for (;;) { + std::function job; + { + const absl::MutexLock lock(mutex_, absl::Condition(this, &RunExecutor::Pending)); + if (stop_ && queue_.empty()) { + return; + } + job = std::move(queue_.front()); + queue_.pop(); + } + job(); + } +} + +} // namespace xff::engine diff --git a/xff/engine/run_executor.h b/xff/engine/run_executor.h new file mode 100644 index 00000000000..4c003c4d18c --- /dev/null +++ b/xff/engine/run_executor.h @@ -0,0 +1,66 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#ifndef XFF_ENGINE_RUN_EXECUTOR_H_ +#define XFF_ENGINE_RUN_EXECUTOR_H_ + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/synchronization/mutex.h" + +namespace xff::engine { + +// One lazily started worker executor shared by all parallel phases of a run. +// Start grows the pool but never shrinks it; destruction joins the workers once. +class RunExecutor final { + public: + explicit RunExecutor(std::size_t max_workers) : max_workers_(max_workers) {} + + ~RunExecutor(); + + RunExecutor(const RunExecutor&) = delete; + RunExecutor& operator=(const RunExecutor&) = delete; + RunExecutor(RunExecutor&&) = delete; + RunExecutor& operator=(RunExecutor&&) = delete; + + void Start(std::size_t useful_workers); + + template + auto Submit(Job job) ABSL_LOCKS_EXCLUDED(mutex_) { + using Result = std::invoke_result_t; + const auto task = std::make_shared>(std::move(job)); + std::future future = task->get_future(); + if (threads_.empty()) { + (*task)(); + return future; + } + const absl::MutexLock lock(mutex_); + queue_.emplace([task] { (*task)(); }); + return future; + } + + std::size_t worker_count() const { return threads_.size(); } + + private: + bool Pending() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + void Run() ABSL_LOCKS_EXCLUDED(mutex_); + + const std::size_t max_workers_; + mutable absl::Mutex mutex_; + std::queue> queue_ ABSL_GUARDED_BY(mutex_); + bool stop_ ABSL_GUARDED_BY(mutex_) = false; + std::vector threads_; +}; + +} // namespace xff::engine + +#endif // XFF_ENGINE_RUN_EXECUTOR_H_ diff --git a/xff/engine/run_executor_test.cc b/xff/engine/run_executor_test.cc new file mode 100644 index 00000000000..b204b7d8650 --- /dev/null +++ b/xff/engine/run_executor_test.cc @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#include "xff/engine/run_executor.h" + +#include +#include + +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +namespace xff::engine { +namespace { +using ::testing::Eq; +using ::testing::Ne; + +struct RunExecutorTest : ::testing::Test {}; + +TEST_F(RunExecutorTest, RunsInlineUntilStartedAndRetainsWorkers) { + RunExecutor executor(3); + const std::thread::id coordinator = std::this_thread::get_id(); + EXPECT_THAT(executor.Submit([] { return std::this_thread::get_id(); }).get(), Eq(coordinator)); + + executor.Start(2); + EXPECT_THAT(executor.worker_count(), Eq(2)); + EXPECT_THAT(executor.Submit([] { return std::this_thread::get_id(); }).get(), Ne(coordinator)); + + executor.Start(3); + EXPECT_THAT(executor.worker_count(), Eq(3)); +} + +} // namespace +} // namespace xff::engine diff --git a/xff/engine/walk.cc b/xff/engine/walk.cc index 329a8b08e56..87ab1fde68b 100644 --- a/xff/engine/walk.cc +++ b/xff/engine/walk.cc @@ -18,17 +18,13 @@ #include #include #include -#include #include #include #include -#include #include #include #include #include -#include -#include #include #include @@ -40,6 +36,7 @@ #include "absl/types/span.h" #include "mbo/status/status_macros.h" #include "mbo/types/optional_ref.h" +#include "xff/engine/run_executor.h" #include "xff/vfs/entry.h" #include "xff/vfs/filesystem.h" @@ -77,84 +74,6 @@ using Listing = absl::StatusOr>; // so layout shape remains part of the decision and narrow/deep walks stay serial. inline constexpr std::size_t kMinParallelDirectoryReads = 64; -// A fixed pool of worker threads running leaf directory or stat jobs. -// Workers touch only their job's inputs and the (thread-safe) FileSystem and the -// task queue; they never call back into the walk, so no job can wait on another -// and there is no shared walk state to race (the coordinator runs everything -// else on one thread). With zero workers, `Submit` runs the job inline. -class ReadPool { - public: - explicit ReadPool(std::size_t workers) : max_workers_(workers) {} - - // Start only when sibling directories or eager stat chunks provide independent work. - void Start(std::size_t directories) { - const std::size_t count = std::min(max_workers_, directories); - threads_.reserve(count); - while (threads_.size() < count) { - threads_.emplace_back([this] { Run(); }); - } - } - - ~ReadPool() { - { - const absl::MutexLock lock(mutex_); - stop_ = true; // turns Pending() true, so every worker's Await wakes - } - for (std::thread& thread : threads_) { - thread.join(); - } - } - - ReadPool(const ReadPool&) = delete; - ReadPool& operator=(const ReadPool&) = delete; - ReadPool(ReadPool&&) = delete; - ReadPool& operator=(ReadPool&&) = delete; - - // Enqueues a read job (or runs it inline with no workers). Caller must NOT - // hold `mutex_`. - template - auto Submit(Job job) ABSL_LOCKS_EXCLUDED(mutex_) { - using Result = std::invoke_result_t; - const auto task = std::make_shared>(std::move(job)); - std::future future = task->get_future(); - if (threads_.empty()) { - (*task)(); // sequential: run inline - return future; - } - const absl::MutexLock lock(mutex_); - queue_.emplace([task] { (*task)(); }); - return future; - } - - private: - // A job is ready or the pool is stopping. `absl::Mutex::Await` evaluates this - // with `mutex_` held, so it requires the lock. - bool Pending() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) { return stop_ || !queue_.empty(); } - - // Worker loop: drain jobs until stopped and the queue is empty. Caller (the - // worker thread) must NOT hold `mutex_`. - void Run() ABSL_LOCKS_EXCLUDED(mutex_) { - for (;;) { - std::function job; - { - const absl::MutexLock lock(mutex_, absl::Condition(this, &ReadPool::Pending)); - if (stop_ && queue_.empty()) { - return; - } - job = std::move(queue_.front()); - queue_.pop(); - } - job(); - } - } - - const std::size_t max_workers_; - mutable absl::Mutex mutex_; - std::queue> queue_ ABSL_GUARDED_BY(mutex_); - bool stop_ ABSL_GUARDED_BY(mutex_) = false; - std::vector threads_; // started on demand, joined in the dtor; not shared otherwise -}; - class Walker { public: Walker( @@ -162,14 +81,15 @@ class Walker { const WalkOptions& options, Visitor visit, WalkErrorFn on_error, - mbo::types::OptionalRef mount_container) + mbo::types::OptionalRef mount_container, + RunExecutor& executor) : fs_(fs), options_(options), visit_(visit), on_error_(on_error), mount_container_(mount_container), follow_children_(options.symlinks == SymlinkMode::kAll), - pool_(options.workers > 1 ? options.workers : std::size_t{0}) {} + executor_(executor) {} // Whether `stated` is a FILE this walk should try to open as a container. `kRoots` offers only the // paths named on the command line (depth 0), `kAll` offers every file met; `kNone` never asks. @@ -224,7 +144,7 @@ class Walker { // The mounter is passed on, so a container inside a container dives too - bounded by // --archive-depth through `container_depth_`. Under `roots` that bound never binds: a member is // never at depth 0, so the mode itself already says "only the archive I was pointed at". - Walker inner(*mounted, options_, visit_, on_error_, mount_container_); + Walker inner(*mounted, options_, visit_, on_error_, mount_container_, executor_); // The inner walk's entries belong to the CONTAINER's filesystem, so they carry shared // ownership of it: a consumer that outlives the dive (a mount) keeps the reader alive. inner.fs_owner_ = std::move(mounted); @@ -342,12 +262,12 @@ class Walker { } const auto chunks = (entries.size() + kChunk - 1) / kChunk; const auto pending = std::min(chunks, options_.workers); - pool_.Start(pending); + executor_.Start(pending); const auto all_entries = absl::MakeSpan(entries); const auto submit = [&](std::size_t index) { const auto offset = index * kChunk; const auto chunk = all_entries.subspan(offset, std::min(kChunk, entries.size() - offset)); - return pool_.Submit([this, chunk] { return StatEntries(chunk); }); + return executor_.Submit([this, chunk] { return StatEntries(chunk); }); }; std::vector>> reads; reads.reserve(pending); @@ -367,7 +287,7 @@ class Walker { } std::future SubmitRead(const std::string& dir) { - return pool_.Submit([this, dir] { return ReadDir(dir); }); + return executor_.Submit([this, dir] { return ReadDir(dir); }); } bool Descendable(const Stated& stated, int depth) const { @@ -598,7 +518,7 @@ class Walker { if (directories.size() < kMinParallelDirectoryReads) { return reads; } - pool_.Start(directories.size()); + executor_.Start(directories.size()); for (const std::size_t index : directories) { reads[index] = SubmitRead(children[index].path); } @@ -622,8 +542,7 @@ class Walker { std::string_view current_root_; std::size_t current_root_index_ = 0; std::set> ancestors_; - // Join pending reads before destroying the mounted filesystem or other walker state. - ReadPool pool_; + RunExecutor& executor_; }; } // namespace @@ -634,7 +553,18 @@ absl::Status Walk( const WalkOptions& options, Visitor visit, WalkErrorFn on_error) { - Walker walker(fs, options, visit, on_error, /*mount_container=*/{}); + RunExecutor executor(options.workers > 1 ? options.workers : std::size_t{0}); + return Walk(fs, roots, options, executor, visit, on_error); +} + +absl::Status Walk( + const vfs::FileSystem& fs, + absl::Span roots, + const WalkOptions& options, + RunExecutor& executor, + Visitor visit, + WalkErrorFn on_error) { + Walker walker(fs, options, visit, on_error, /*mount_container=*/{}, executor); walker.WalkRoots(roots); return absl::OkStatus(); } @@ -646,7 +576,19 @@ absl::Status Walk( Visitor visit, WalkErrorFn on_error, ContainerMounter mount_container) { - Walker walker(fs, options, visit, on_error, mount_container); + RunExecutor executor(options.workers > 1 ? options.workers : std::size_t{0}); + return Walk(fs, roots, options, executor, visit, on_error, mount_container); +} + +absl::Status Walk( + const vfs::FileSystem& fs, + absl::Span roots, + const WalkOptions& options, + RunExecutor& executor, + Visitor visit, + WalkErrorFn on_error, + ContainerMounter mount_container) { + Walker walker(fs, options, visit, on_error, mount_container, executor); walker.WalkRoots(roots); return absl::OkStatus(); } diff --git a/xff/engine/walk.h b/xff/engine/walk.h index af0ac89edc0..54ba06e1e5c 100644 --- a/xff/engine/walk.h +++ b/xff/engine/walk.h @@ -32,6 +32,8 @@ namespace xff::engine { +class RunExecutor; + // Which symlinks the walk resolves (stats the target) before testing/descending: // none (find `-P`, the default), only command-line operands (`-H`), or all // (`-L`). Following enables filesystem-loop detection. @@ -187,6 +189,14 @@ absl::Status Walk( Visitor visit, WalkErrorFn on_error); +absl::Status Walk( + const vfs::FileSystem& fs, + absl::Span roots, + const WalkOptions& options, + RunExecutor& executor, + Visitor visit, + WalkErrorFn on_error); + // As above, plus archive diving: when `options.archive` allows it, a FILE the walk meets is offered to // `mount_container`, and a container that opens is descended into as though it were a directory - its // members visited at the depth below it, through the mounted filesystem. @@ -203,6 +213,15 @@ absl::Status Walk( WalkErrorFn on_error, ContainerMounter mount_container); +absl::Status Walk( + const vfs::FileSystem& fs, + absl::Span roots, + const WalkOptions& options, + RunExecutor& executor, + Visitor visit, + WalkErrorFn on_error, + ContainerMounter mount_container); + } // namespace xff::engine #endif // XFF_ENGINE_WALK_H_ From 2c56ec76c9317dd38e0fe9389f3324721941dd67 Mon Sep 17 00:00:00 2001 From: helly25 Date: Fri, 9 Oct 2026 23:49:30 +0100 Subject: [PATCH 3/4] Drain coarse work batches through explicit run completions --- XFF.md | 2 +- docs/benchmark-correctness-audit.md | 78 ++++++++++++++- docs/design-parallel.md | 38 +++++-- xff/cli/globals.cc | 8 +- xff/engine/BUILD.bazel | 5 + xff/engine/parallel_match.cc | 11 +-- xff/engine/parallel_match_test.cc | 3 +- xff/engine/run_executor.cc | 2 +- xff/engine/run_executor.h | 101 +++++++++++++++++-- xff/engine/run_executor_test.cc | 52 +++++++++- xff/engine/walk.cc | 147 +++++++++++++++++----------- xff/engine/walk_test.cc | 100 +++++++++++++++++++ 12 files changed, 457 insertions(+), 90 deletions(-) diff --git a/XFF.md b/XFF.md index 1b799728ac4..9ba8849e7a5 100644 --- a/XFF.md +++ b/XFF.md @@ -564,7 +564,7 @@ See also: [Configuration](#topic-config), [Archives](#topic-archive), [Output](# - `-j N, -j=N, --jobs=N|all` - directory-read, eligible content-match, comparison and -exec workers _(global, find, xff, rg)_ - `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories on up to `N` worker threads; large directory listings can split required eager metadata reads across the same pool. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches also match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core. + `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories using one lazily started pool of up to `N` worker threads, reused for directory reads, required eager metadata and eligible content matching. Workers drain batches of independent work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core. See also: [Output](#topic-output), [Statistics](#topic-stats) diff --git a/docs/benchmark-correctness-audit.md b/docs/benchmark-correctness-audit.md index f44c423f3de..60c45fd76bc 100644 --- a/docs/benchmark-correctness-audit.md +++ b/docs/benchmark-correctness-audit.md @@ -112,8 +112,9 @@ retrofit the binary identity of a historical measurement. Regular CI retains report artifacts for 30 days and executable build artifacts for three days; hosted backfill artifacts omit executables. Preserve available original binaries, build manifests -and logs before cleanup or artifact expiry. Linux executable verification remains outstanding; -the 72 checked Mac binaries do not verify the Linux measurements. +and logs before cleanup or artifact expiry. The initial Mac audit left Linux executable +verification outstanding; the native Linux follow-up below now covers the available local +Linux artifacts. The 72 checked Mac binaries do not verify the Linux measurements. On Linux, verify each retained executable against its original report with the PR's `tools/benchmark_provenance.py --report=... --binary=... --retention-reference=...` command. @@ -244,3 +245,76 @@ metadata and slow-storage controls on both platforms. Run affected correctness/c and appropriate sanitizers. Report absolute XFF time, reference time and same-run ratio together. The current evidence supports investigating specific costs; it does not establish that XFF will beat every reference tool on every workload. + +## Native Linux follow-up (2026-10-09) + +The native session uses merged audit driver commit +`4ea895f63106f05df7c90b629d20433ae1e03016`. The original engine baseline remains +`772994ab7905be77915978f06ea90ae5934aaeef`; both source revisions produce the same +retained executable SHA-256, +`f97833a8eb45f49b596665b47911c2df9a2558619778512fd31d1b38f2e56106`. +Baseline builds use `clang_release`, with the historical ICU 70 build-library +path supplied explicitly. Complete prepared-grid runs retain nine raw samples +per cell, use the fastest seven estimator and store fixtures on verified tmpfs. + +### Retained Linux executables + +All 142 completed reports under the local Linux benchmark batch inventory match +their original retained executables and their original build manifests. They +cover 72 distinct commits and 38 distinct executable hashes; no mismatches were +found. Verification uses `benchmark_provenance.verify_retention`, checks the +standalone manifest against the report's embedded build record, and records a +measurement-bound proof for each report. Historical observations are unchanged. + +The original reports and binaries remain under `/home/marcus/xff-benchmarks/batches`. +The native proof inventory is retained at +`/tmp/xff-linux-retention-verification/summary.json`, with per-report +`binary-verification.json` files below that directory. These checks verify the +available local Linux artifacts; they do not verify expired hosted artifacts. + +### Worker scheduling experiment + +Linux `kernel.perf_event_paranoid` is now `0`, and native `perf` hardware counters +are available. Startup-boundary profiles identify worker creation and teardown, +Abseil mutex slow paths and scheduling overhead. Keeping directory reads serial +removes the small-work cliff but loses useful concurrency on large broad trees. +[PR #991](https://github.com/mboworks/xff/pull/991) admits directory reads only +when at least 64 independent sibling directories are available. Its hosted Linux +and macOS tests, sanitizers, all three benchmark shards on each platform and +combined benchmark comparison pass. + +The next experiment shares one run-owned executor across directory read-ahead, +eager stat work and eligible content matching. Each admitted batch submits at +most one drain job per useful worker. Directory jobs repeatedly claim independent +listing reads; stat jobs claim 128-entry chunks; matcher jobs claim entry chunks. +Only the coordinator recurses, evaluates traversal controls and consumes ordered +results. Threads remain bounded and are reused until command completion. + +Correctness checks include nested read-ahead above the admission threshold, +partial stat chunks with missing entries, repeated matcher transitions and +executor reuse. An early-stop test deliberately holds a prefetched sibling read +open while the coordinator quits: `Walk` must wait before destroying the walker, +even though its executor survives for another phase. Explicit Abseil completion +signals replaced the experiment's standard-library futures after expanded TSan +coverage reported a shared-state completion/deallocation race. + +The interrupted `/tmp/xff-perf-shared-executor-v2` run is diagnostic only: +correctness builds overlapped its measurements, and its measured source predates +the final scheduling changes. It supplies no acceptance timing evidence. + +### Memory-accounting finding and remaining acceptance work + +Direct `/usr/bin/time -v` runs show approximately 5.3-5.4 MiB peak RSS for a small +XFF invocation, while the Python comparison worker reports approximately +22-23 MiB for the same command. A trivial native command also reproduces the +Python worker's high floor. The forked launcher's inherited high-water mark must +be corrected or explicitly excluded before using those reported peaks to choose +arena sizes. This is separate from the worker scheduler experiment. + +The complete audit remains open. Final shared-executor timing must be collected +after every build has finished, using the complete prepared grid. Repeated +controls must cover content, eager metadata and slow storage. Native allocation +profiles and the separate 20k/50k/100k single-worker slope investigation still +need completion. Hosted macOS checks provide portability evidence for PR #991; +they do not replace the specified native M5 Pro A/B and sampling controls for +the shared-executor experiment. diff --git a/docs/design-parallel.md b/docs/design-parallel.md index 141e1a55f37..bbddeee63a8 100644 --- a/docs/design-parallel.md +++ b/docs/design-parallel.md @@ -132,13 +132,34 @@ actions still occur in traversal order. ## Parallelism control -A single knob, `-j N` (long form `--jobs`), configures independent limits: up to -`N` directory-read workers, up to `N` eligible content-matcher workers, and at most -`N` outstanding semicolon-form `-exec`/`-execdir` children. These are separate -limits, not one shared `N`-operation budget. Content matching does not run alongside -execution actions in the same expression. `-j 1` makes each part synchronous. -`-j all` (`--jobs=all`) uses every detected core (`hardware_concurrency()`) for -each limit, regardless of the active mode's default. +A single knob, `-j N` (long form `--jobs`), caps one run-owned executor at `N` +workers for directory reads, eager stat batches and eligible content matching. +The coordinator retains traversal and output ownership. The executor starts +lazily, grows only as useful work becomes available, and reuses its threads +across phases and mounted-filesystem walks until the command finishes. +Semicolon-form `-exec`/`-execdir` separately allows at most `N` outstanding child +processes; content matching does not run alongside execution actions in the same +expression. `-j 1` makes each part synchronous. `-j all` (`--jobs=all`) uses every +detected core (`hardware_concurrency()`) for these limits, regardless of the +active mode's default. + +The coordinator admits directory read-ahead when at least 64 independent sibling +reads are available. It submits at most one drain job per worker; each job claims +successive directories from that batch and publishes indexed listing results. +Workers perform leaf reads and never recurse or wait on jobs in their own +executor. Eager stat work similarly uses drain jobs over 128-entry chunks after +at least 512 entries are available. This bounds executor queue operations by +worker count while preserving coordinator visitation order. + +Content matching retains private evaluator and regex state for each logical +worker. Decision-only matching collects up to 8,192 entries before flushing; +rendered-content batches remain bounded at 256 entries because their output can +be large. A cold executor keeps cheap small batches on the coordinator until +8,192 entries have accumulated. A timed 16-entry prefix admits slower work +earlier when that prefix takes at least 500 microseconds. Once traversal or +matching has started the executor, batches of at least 16 entries may reuse it. +Each submitted matcher job drains entry chunks, and results are consumed in +original entry order. Under `-j > 1` the serial `-exec ... ;` / `-execdir ... ;` form launches its child on a bounded runner (at most `N` outstanding) instead of running it synchronously. @@ -178,7 +199,8 @@ design-config.md): entry from the pruned subtree is visited or acted upon. - **`-quit`** - the coordinator stops visiting entries immediately. Workers only perform leaf directory reads and do not inspect the walk's stop flag; already - submitted reads finish while the pool is destroyed. Their results are ignored, + submitted drain jobs finish before their walker is destroyed, even when the + executor will be reused by a later phase. Their results are ignored, and no new expression actions are run. Exit status follows the normal model. - **`-depth` (post-order)** - children before parent. The ordering layer holds a directory's own visit until its subtree has been emitted; under `none` this diff --git a/xff/cli/globals.cc b/xff/cli/globals.cc index c30469d3009..790786ea8a3 100644 --- a/xff/cli/globals.cc +++ b/xff/cli/globals.cc @@ -903,10 +903,10 @@ constexpr std::array kGlobals = std::to_array({ .details = "`N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached " "short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes " "directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories " - "on up to `N` worker threads; large directory listings can split required eager metadata reads " - "across the same pool. Lazy and metadata-free listings do not start stat workers. " - "Independent content predicates and rg searches also match in bounded " - "parallel batches, " + "using one lazily started pool of up to `N` worker threads, reused for directory reads, " + "required eager metadata and eligible content matching. Workers drain batches of independent " + "work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start " + "stat workers. Independent content predicates and rg searches match in bounded parallel batches, " "with output kept in traversal order. Small initial batches, archive members, stateful expressions " "stay on the coordinator. Metadata-consuming output does not itself prevent eligible content " "filters from running in workers; reductions and rendering remain serial. Independent " diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel index 28dd26f21ac..78bf4bbbde9 100644 --- a/xff/engine/BUILD.bazel +++ b/xff/engine/BUILD.bazel @@ -26,6 +26,7 @@ cc_library( visibility = ["//xff:__subpackages__"], deps = [ "@abseil-cpp//absl/base:core_headers", + "@abseil-cpp//absl/functional:any_invocable", "@abseil-cpp//absl/synchronization", ], ) @@ -36,6 +37,7 @@ cc_test( srcs = ["run_executor_test.cc"], deps = [ ":run_executor_cc", + "@abseil-cpp//absl/synchronization", "@googletest//:gtest", "@googletest//:gtest_main", ], @@ -163,12 +165,15 @@ cc_test( size = "small", srcs = ["walk_test.cc"], deps = [ + ":run_executor_cc", ":walk_cc", "//xff/vfs:local_fs_cc", "//xff/vfs:vfs_cc", "@abseil-cpp//absl/functional:function_ref", "@abseil-cpp//absl/status", "@abseil-cpp//absl/status:statusor", + "@abseil-cpp//absl/strings", + "@abseil-cpp//absl/synchronization", "@googletest//:gtest", "@googletest//:gtest_main", "@mboworks_mbo//mbo/testing:status_cc", diff --git a/xff/engine/parallel_match.cc b/xff/engine/parallel_match.cc index 25dbaa0b046..33eada5398c 100644 --- a/xff/engine/parallel_match.cc +++ b/xff/engine/parallel_match.cc @@ -5,7 +5,6 @@ #include #include -#include #include #include #include @@ -123,13 +122,13 @@ const std::vector& ParallelMatch::Match(std::vector(*this)); } - std::vector> futures; - futures.reserve(count); + std::vector> tasks; + tasks.reserve(count); for (std::size_t worker = 0; worker < count; ++worker) { - futures.push_back(executor_.Submit([this, worker] { EvaluateWorker(worker); })); + tasks.push_back(executor_.Submit([this, worker] { EvaluateWorker(worker); })); } - for (std::future& future : futures) { - future.get(); + for (RunTask& task : tasks) { + task.Get(); } return results_; } diff --git a/xff/engine/parallel_match_test.cc b/xff/engine/parallel_match_test.cc index 106c11337c2..4ef6d3b6268 100644 --- a/xff/engine/parallel_match_test.cc +++ b/xff/engine/parallel_match_test.cc @@ -106,7 +106,7 @@ TEST_F(ParallelMatchTest, PreparedCoordinatorAndWorkersRetainRegexStateAcrossPoo ParallelMatch matcher(*command.expression, 4, executor_pool, false, std::nullopt, execution); // Small batches run on the coordinator, large batches activate private worker state; // returning to the coordinator must not reuse a worker's previous match or scratch. - for (const std::size_t count : {1, 128, 7, 512, 4'096, 10}) { + for (const std::size_t count : {1, 128, 7, 512, 8'192, 10}) { const auto& results = matcher.Match(Entries(count)); ASSERT_THAT(results, SizeIs(count)); for (std::size_t index = 0; index < count; ++index) { @@ -114,6 +114,7 @@ TEST_F(ParallelMatchTest, PreparedCoordinatorAndWorkersRetainRegexStateAcrossPoo EXPECT_THAT(results.at(index).evaluation.matched, Eq(path.find_first_of("13579") == std::string::npos)); } } + EXPECT_THAT(executor_pool.worker_count(), Eq(4)); } TEST_F(ParallelMatchTest, ColdExecutorWaitsForAccumulatedWork) { diff --git a/xff/engine/run_executor.cc b/xff/engine/run_executor.cc index 70bbaa8b2d7..50e2170e1e7 100644 --- a/xff/engine/run_executor.cc +++ b/xff/engine/run_executor.cc @@ -31,7 +31,7 @@ bool RunExecutor::Pending() const { void RunExecutor::Run() { for (;;) { - std::function job; + absl::AnyInvocable job; { const absl::MutexLock lock(mutex_, absl::Condition(this, &RunExecutor::Pending)); if (stop_ && queue_.empty()) { diff --git a/xff/engine/run_executor.h b/xff/engine/run_executor.h index 4c003c4d18c..7e8d959849b 100644 --- a/xff/engine/run_executor.h +++ b/xff/engine/run_executor.h @@ -5,22 +5,97 @@ #define XFF_ENGINE_RUN_EXECUTOR_H_ #include -#include -#include #include +#include #include #include #include #include +#include #include #include "absl/base/thread_annotations.h" +#include "absl/functional/any_invocable.h" #include "absl/synchronization/mutex.h" +#include "absl/synchronization/notification.h" namespace xff::engine { +namespace executor_detail { +template +struct TaskState final { + using Stored = std::conditional_t, std::monostate, Result>; + std::optional result; + absl::Notification completed; +}; +} // namespace executor_detail + +template +class RunPromise; + +// A single-consumer result. Wait, Ready and Get require a valid task; Get consumes +// the value once. Completion publishes through an Abseil notification; no worker +// may touch the value after signaling completion. +template +class RunTask final { + public: + RunTask() = default; + RunTask(RunTask&&) = default; + RunTask& operator=(RunTask&&) = default; + RunTask(const RunTask&) = delete; + RunTask& operator=(const RunTask&) = delete; + + bool Valid() const { return state_ != nullptr; } + + bool Ready() const { return state_->completed.HasBeenNotified(); } + + void Wait() const { state_->completed.WaitForNotification(); } + + Result Get() { + Wait(); + if constexpr (!std::is_void_v) { + return std::move(*state_->result); + } + } + + private: + friend class RunPromise; + + explicit RunTask(std::shared_ptr> state) : state_(std::move(state)) {} + + std::shared_ptr> state_; +}; + +// The producer side of a RunTask, used when a drain job produces indexed results. +// Obtain one Task and call SetValue exactly once before releasing the producer. +template +class RunPromise final { + public: + using Stored = typename executor_detail::TaskState::Stored; + + RunTask Task() const { return RunTask(state_); } + + void SetValue(Stored result) + requires(!std::is_void_v) + { + state_->result.emplace(std::move(result)); + state_->completed.Notify(); + } + + void SetValue() + requires(std::is_void_v) + { + state_->completed.Notify(); + } + + private: + std::shared_ptr> state_ = std::make_shared>(); +}; + // One lazily started worker executor shared by all parallel phases of a run. // Start grows the pool but never shrinks it; destruction joins the workers once. +// Start, Submit and worker_count are called only by the coordinator. Jobs stay +// leaves and must not submit or wait on work in this executor. class RunExecutor final { public: explicit RunExecutor(std::size_t max_workers) : max_workers_(max_workers) {} @@ -37,15 +112,23 @@ class RunExecutor final { template auto Submit(Job job) ABSL_LOCKS_EXCLUDED(mutex_) { using Result = std::invoke_result_t; - const auto task = std::make_shared>(std::move(job)); - std::future future = task->get_future(); + RunPromise promise; + auto result = promise.Task(); + absl::AnyInvocable task = [promise = std::move(promise), job = std::move(job)]() mutable { + if constexpr (std::is_void_v) { + job(); + promise.SetValue(); + } else { + promise.SetValue(job()); + } + }; if (threads_.empty()) { - (*task)(); - return future; + task(); + return result; } const absl::MutexLock lock(mutex_); - queue_.emplace([task] { (*task)(); }); - return future; + queue_.emplace(std::move(task)); + return result; } std::size_t worker_count() const { return threads_.size(); } @@ -56,7 +139,7 @@ class RunExecutor final { const std::size_t max_workers_; mutable absl::Mutex mutex_; - std::queue> queue_ ABSL_GUARDED_BY(mutex_); + std::queue> queue_ ABSL_GUARDED_BY(mutex_); bool stop_ ABSL_GUARDED_BY(mutex_) = false; std::vector threads_; }; diff --git a/xff/engine/run_executor_test.cc b/xff/engine/run_executor_test.cc index b204b7d8650..5f25f6f7187 100644 --- a/xff/engine/run_executor_test.cc +++ b/xff/engine/run_executor_test.cc @@ -3,30 +3,76 @@ #include "xff/engine/run_executor.h" -#include +#include +#include #include +#include +#include +#include "absl/synchronization/barrier.h" #include "gmock/gmock.h" #include "gtest/gtest.h" namespace xff::engine { namespace { +using ::testing::Contains; using ::testing::Eq; +using ::testing::IsFalse; using ::testing::Ne; +using ::testing::Not; +using ::testing::SizeIs; struct RunExecutorTest : ::testing::Test {}; TEST_F(RunExecutorTest, RunsInlineUntilStartedAndRetainsWorkers) { RunExecutor executor(3); const std::thread::id coordinator = std::this_thread::get_id(); - EXPECT_THAT(executor.Submit([] { return std::this_thread::get_id(); }).get(), Eq(coordinator)); + EXPECT_THAT(executor.Submit([] { return std::this_thread::get_id(); }).Get(), Eq(coordinator)); executor.Start(2); EXPECT_THAT(executor.worker_count(), Eq(2)); - EXPECT_THAT(executor.Submit([] { return std::this_thread::get_id(); }).get(), Ne(coordinator)); + EXPECT_THAT(executor.Submit([] { return std::this_thread::get_id(); }).Get(), Ne(coordinator)); executor.Start(3); EXPECT_THAT(executor.worker_count(), Eq(3)); + executor.Start(1); + EXPECT_THAT(executor.worker_count(), Eq(3)); + executor.Start(100); + EXPECT_THAT(executor.worker_count(), Eq(3)); +} + +TEST_F(RunExecutorTest, ReusesTheSameBoundedThreadsAcrossPhases) { + RunExecutor executor(3); + executor.Start(3); + const auto phase = [&] { + absl::Barrier started(3); + std::vector> tasks; + for (int worker = 0; worker < 3; ++worker) { + tasks.push_back(executor.Submit([&] { + started.Block(); + return std::this_thread::get_id(); + })); + } + std::set threads; + for (auto& task : tasks) { + threads.insert(task.Get()); + } + return threads; + }; + const auto first = phase(); + EXPECT_THAT(first, SizeIs(3)); + EXPECT_THAT(first, Not(Contains(std::this_thread::get_id()))); + EXPECT_THAT(phase(), Eq(first)); +} + +TEST_F(RunExecutorTest, SupportsMoveOnlyJobsAndResults) { + RunExecutor executor(1); + executor.Start(1); + auto task = executor.Submit([value = std::make_unique(42)]() mutable { return std::move(value); }); + auto result = task.Get(); + EXPECT_THAT(*result, Eq(42)); + EXPECT_THAT(RunTask{}.Valid(), IsFalse()); + executor.Submit([] {}).Get(); } } // namespace diff --git a/xff/engine/walk.cc b/xff/engine/walk.cc index 87ab1fde68b..cec68b8ed2c 100644 --- a/xff/engine/walk.cc +++ b/xff/engine/walk.cc @@ -16,12 +16,11 @@ #include "xff/engine/walk.h" #include +#include #include #include -#include #include #include -#include #include #include #include @@ -91,6 +90,15 @@ class Walker { follow_children_(options.symlinks == SymlinkMode::kAll), executor_(executor) {} + ~Walker() { + // A stop action may leave prefetched siblings unconsumed. Their drain jobs + // still reference this walker, so join those jobs before its state is torn + // down. Normal walks reach already-completed jobs here. + for (RunTask& drain : drain_tasks_) { + drain.Wait(); + } + } + // Whether `stated` is a FILE this walk should try to open as a container. `kRoots` offers only the // paths named on the command line (depth 0), `kAll` offers every file met; `kNone` never asks. bool ShouldTryMount(const Stated& stated, int depth) const { @@ -223,23 +231,26 @@ class Walker { return Stated{.path = std::move(path), .name = std::move(entry_name), .metadata = *metadata, .ok = true}; } + Stated StatEntry(vfs::Entry entry) const { + if (options_.metadata != MetadataDemand::kAlways && entry.type != vfs::FileType::kUnknown + && entry.type != vfs::FileType::kDirectory && (!follow_children_ || entry.type != vfs::FileType::kSymlink)) { + return { + .path = std::move(entry.path), + .name = std::move(entry.name), + .metadata = {.type = entry.type, .source = entry.source}, + .ok = true, + .metadata_loaded = false, + }; + } + return StatNode(std::move(entry.path), follow_children_, std::move(entry.name)); + } + // Resolve exactly the eager metadata required for this owned listing slice. Safe on a worker. std::vector StatEntries(absl::Span entries) const { std::vector children; children.reserve(entries.size()); for (vfs::Entry& entry : entries) { - if (options_.metadata != MetadataDemand::kAlways && entry.type != vfs::FileType::kUnknown - && entry.type != vfs::FileType::kDirectory && (!follow_children_ || entry.type != vfs::FileType::kSymlink)) { - children.push_back({ - .path = std::move(entry.path), - .name = std::move(entry.name), - .metadata = {.type = entry.type, .source = entry.source}, - .ok = true, - .metadata_loaded = false, - }); - } else { - children.push_back(StatNode(std::move(entry.path), follow_children_, std::move(entry.name))); - } + children.push_back(StatEntry(std::move(entry))); } return children; } @@ -263,33 +274,31 @@ class Walker { const auto chunks = (entries.size() + kChunk - 1) / kChunk; const auto pending = std::min(chunks, options_.workers); executor_.Start(pending); - const auto all_entries = absl::MakeSpan(entries); - const auto submit = [&](std::size_t index) { - const auto offset = index * kChunk; - const auto chunk = all_entries.subspan(offset, std::min(kChunk, entries.size() - offset)); - return executor_.Submit([this, chunk] { return StatEntries(chunk); }); - }; - std::vector>> reads; + std::atomic next = 0; + std::vector children(entries.size()); + std::vector> reads; reads.reserve(pending); - for (std::size_t index = 0; index < pending; ++index) { - reads.push_back(submit(index)); + for (std::size_t worker = 0; worker < pending; ++worker) { + reads.push_back(executor_.Submit([&] { + for (;;) { + const std::size_t index = next.fetch_add(1, std::memory_order_relaxed); + if (index >= chunks) { + return; + } + const std::size_t first = index * kChunk; + const std::size_t end = std::min(first + kChunk, entries.size()); + for (std::size_t entry_index = first; entry_index < end; ++entry_index) { + children[entry_index] = StatEntry(std::move(entries[entry_index])); + } + } + })); } - std::vector children; - children.reserve(entries.size()); - for (std::size_t index = 0; index < chunks; ++index) { - auto& read = reads.at(index % pending); - children.append_range(read.get() | std::views::as_rvalue); - if (index + pending < chunks) { - read = submit(index + pending); - } + for (RunTask& read : reads) { + read.Get(); } return children; } - std::future SubmitRead(const std::string& dir) { - return executor_.Submit([this, dir] { return ReadDir(dir); }); - } - bool Descendable(const Stated& stated, int depth) const { const bool is_dir = stated.ok && stated.metadata.type == vfs::FileType::kDirectory; const bool within_depth = options_.max_depth < 0 || depth < options_.max_depth; @@ -297,7 +306,7 @@ class Walker { return is_dir && within_depth && on_root_fs; } - // Only the coordinator accesses this record after its listing future completes. Exclusive + // Only the coordinator accesses this record after its listing completion arrives. Exclusive // ownership makes the lazy cache lock-free; it is never shared with matcher workers. absl::Status LoadMetadata(const Stated& stated) const { if (!stated.metadata_loaded) { @@ -352,7 +361,7 @@ class Walker { // Pre-order by default; post-order (`-depth`) descends first, then visits, and // `-prune` has no effect (matching find). `prefetched` is the directory's // already-submitted listing read (from the parent's batch), or empty to read now. - void VisitSubtree(const Stated& stated, int depth, mbo::types::OptionalRef> prefetched) { + void VisitSubtree(const Stated& stated, int depth, mbo::types::OptionalRef> prefetched) { if (stopped_) { return; } @@ -392,15 +401,15 @@ class Walker { } } - // Reads `dir` (from its prefetched future, or now) and recurses its children, + // Reads `dir` (from its prefetched result, or now) and recurses its children, // guarding against filesystem loops (only possible when following symlinks). - void Descend(const Stated& dir, int depth, mbo::types::OptionalRef> prefetched) { + void Descend(const Stated& dir, int depth, mbo::types::OptionalRef> prefetched) { const std::pair id{dir.metadata.dev, dir.metadata.ino}; if (!ancestors_.insert(id).second) { on_error_(dir.path, absl::FailedPreconditionError("filesystem loop detected")); return; } - Listing listing = prefetched.has_value() ? prefetched->get() : ReadNow(dir.path); + Listing listing = prefetched.has_value() ? prefetched->Get() : ReadNow(dir.path); if (!listing.ok()) { // A directory that vanished before we could read it is the same readdir race. if (!(options_.ignore_readdir_race && absl::IsNotFound(listing.status()))) { @@ -425,14 +434,14 @@ class Walker { // Inline DFS at each entry's position. kTree emits a subtree in its sorted // place; post-order (`-depth`) always uses this shape (descend then visit). if (options_.sort == SortOrder::kTree || options_.sort == SortOrder::kGlobal || options_.post_order) { - std::vector> reads = SubmitSubdirReads(children, depth); + std::vector> reads = SubmitSubdirReads(children, depth); for (std::size_t i = 0; i < children.size(); ++i) { if (stopped_) { return; } VisitSubtree( children[i], depth, - reads[i].valid() ? mbo::types::OptionalRef{reads[i]} : mbo::types::OptionalRef>{}); + reads[i].Valid() ? mbo::types::OptionalRef{reads[i]} : mbo::types::OptionalRef>{}); } return; } @@ -448,7 +457,7 @@ class Walker { VisitOne(children[i], depth, options_.mount_before_visit && dived[i]); } } - std::vector> reads = SubmitSubdirReads(children, depth); + std::vector> reads = SubmitSubdirReads(children, depth); for (std::size_t i = 0; i < children.size(); ++i) { if (stopped_) { return; @@ -456,7 +465,7 @@ class Walker { if (IsDir(children[i])) { VisitSubtree( children[i], depth, - reads[i].valid() ? mbo::types::OptionalRef{reads[i]} : mbo::types::OptionalRef>{}); + reads[i].Valid() ? mbo::types::OptionalRef{reads[i]} : mbo::types::OptionalRef>{}); } else if (dived[i]) { // A container groups its members like a directory, so under kSubtree it belongs in the // subtree block rather than the flat block its own entry was emitted in. @@ -476,7 +485,7 @@ class Walker { dived[i] = WillDive(children[i], depth); pruned[i] = VisitOne(children[i], depth, options_.mount_before_visit && dived[i]) == WalkAction::kPrune; } - std::vector> reads = SubmitSubdirReads(children, depth, pruned); + std::vector> reads = SubmitSubdirReads(children, depth, pruned); for (std::size_t i = 0; i < children.size(); ++i) { if (stopped_) { return; @@ -487,8 +496,8 @@ class Walker { if (Descendable(children[i], depth)) { Descend( children[i], depth, - reads[i].valid() ? mbo::types::OptionalRef{reads[i]} - : mbo::types::OptionalRef>{}); // entry already visited above + reads[i].Valid() ? mbo::types::OptionalRef{reads[i]} + : mbo::types::OptionalRef>{}); // entry already visited above } else if (dived[i]) { // `--archive=all`: a container met mid-walk descends exactly where a directory would, and // after its own visit, so a prune on the container still skips its members. @@ -497,15 +506,17 @@ class Walker { } } - // Submits a read for every descendable subdirectory in `children`, returning a - // vector aligned with `children` (an invalid future where there is no read), so - // the pool overlaps their IO. The sequential walk (no workers) reads lazily at - // descend time instead, so it never pre-reads siblings it might not reach. - std::vector> SubmitSubdirReads( + // Makes every descendable subdirectory in `children` available to a bounded set + // of drain jobs, returning a vector aligned with `children` (an invalid task + // where there is no read). Each drain performs many reads when the batch is + // large, so executor synchronization is paid per worker rather than per + // directory. The sequential walk reads lazily at descend time instead, so it + // never pre-reads siblings it might not reach. + std::vector> SubmitSubdirReads( const std::vector& children, int depth, const std::vector& pruned = {}) { - std::vector> reads(children.size()); + std::vector> reads(children.size()); if (options_.workers <= 1) { return reads; } @@ -518,9 +529,34 @@ class Walker { if (directories.size() < kMinParallelDirectoryReads) { return reads; } + std::erase_if(drain_tasks_, [](RunTask& drain) { return drain.Ready(); }); executor_.Start(directories.size()); - for (const std::size_t index : directories) { - reads[index] = SubmitRead(children[index].path); + + struct ReadBatch final { + std::vector paths; + std::vector> promises; + std::atomic next = 0; + }; + + auto batch = std::make_shared(); + batch->paths.reserve(directories.size()); + batch->promises.resize(directories.size()); + for (std::size_t batch_index = 0; batch_index < directories.size(); ++batch_index) { + const std::size_t child_index = directories[batch_index]; + batch->paths.push_back(children[child_index].path); + reads[child_index] = batch->promises[batch_index].Task(); + } + const std::size_t drains = std::min(executor_.worker_count(), directories.size()); + for (std::size_t worker = 0; worker < drains; ++worker) { + drain_tasks_.push_back(executor_.Submit([this, batch] { + for (;;) { + const std::size_t index = batch->next.fetch_add(1, std::memory_order_relaxed); + if (index >= batch->paths.size()) { + return; + } + batch->promises[index].SetValue(ReadDir(batch->paths[index])); + } + })); } return reads; } @@ -543,6 +579,7 @@ class Walker { std::size_t current_root_index_ = 0; std::set> ancestors_; RunExecutor& executor_; + std::vector> drain_tasks_; }; } // namespace diff --git a/xff/engine/walk_test.cc b/xff/engine/walk_test.cc index fb545bad6f0..80b7032f556 100644 --- a/xff/engine/walk_test.cc +++ b/xff/engine/walk_test.cc @@ -16,10 +16,12 @@ #include "xff/engine/walk.h" #include +#include #include #include #include #include +#include #include #include #include @@ -32,9 +34,11 @@ #include "absl/status/status.h" #include "absl/status/statusor.h" #include "absl/strings/str_cat.h" +#include "absl/synchronization/notification.h" #include "gmock/gmock.h" #include "gtest/gtest.h" #include "mbo/testing/status.h" +#include "xff/engine/run_executor.h" #include "xff/vfs/entry.h" #include "xff/vfs/filesystem.h" #include "xff/vfs/local_fs.h" @@ -52,6 +56,7 @@ using ::testing::IsTrue; namespace fs = ::std::filesystem; using ::mbo::testing::IsOk; using ::testing::ElementsAre; +using ::testing::Eq; using ::testing::IsEmpty; using ::testing::Lt; using ::testing::Not; @@ -385,6 +390,101 @@ struct WalkFakeFsTest : ::testing::Test { int errors_ = 0; }; +TEST_F(WalkFakeFsTest, ParallelNestedBatchesPreserveOrderingAndPostOrder) { + std::vector siblings; + std::vector nested; + for (std::size_t index = 0; index < 65; ++index) { + const std::string name = absl::StrCat("d", index); + const std::string path = absl::StrCat("/r/", name); + siblings.push_back(DirEntry(path, name)); + fs_.AddDir(path, 1, {}); + const std::string inner_path = absl::StrCat("/r/d0/", name); + nested.push_back(DirEntry(inner_path, name)); + fs_.AddDir(inner_path, 1, {FileEntry(absl::StrCat(inner_path, "/f"), "f")}); + fs_.AddFile(absl::StrCat(inner_path, "/f"), 1); + } + fs_.AddDir("/r", 1, std::move(siblings)); + fs_.AddDir("/r/d0", 1, std::move(nested)); + for (const SortOrder sort : {SortOrder::kTree, SortOrder::kDir, SortOrder::kSubtree, SortOrder::kGlobal}) { + for (const bool post_order : {false, true}) { + const auto serial = Seen(WalkOptions{.post_order = post_order, .sort = sort, .workers = 1}); + EXPECT_THAT(errors_, Eq(0)); + EXPECT_THAT(Seen(WalkOptions{.post_order = post_order, .sort = sort, .workers = 3}), Eq(serial)); + EXPECT_THAT(errors_, Eq(0)); + } + } +} + +TEST_F(WalkFakeFsTest, ParallelStatDrainsPreservePartialChunkAndErrorOrder) { + std::vector files; + for (std::size_t index = 0; index < 513; ++index) { + const std::string name = absl::StrCat("f", index); + const std::string path = absl::StrCat("/r/", name); + files.push_back(FileEntry(path, name)); + // Retain a missing entry at the final partial chunk boundary. + if (index < 512) { + fs_.AddFile(path, 1); + } + } + fs_.AddDir("/r", 1, std::move(files)); + const auto serial = Seen(WalkOptions{.sort = SortOrder::kTree, .workers = 1}); + EXPECT_THAT(errors_, Eq(1)); + EXPECT_THAT(Seen(WalkOptions{.sort = SortOrder::kTree, .workers = 3}), Eq(serial)); + EXPECT_THAT(errors_, Eq(1)); +} + +// Hold one sibling read open across the coordinator's stop action. This tests +// walker lifetime independently of executor lifetime: the executor is reused +// after Walk returns, and cannot protect a destroyed walker by joining at exit. +class BlockingReadFs final : public FakeFs { + public: + absl::StatusOr> ReadDir(std::string_view dir) const override { + if (dir == "/r/d0") { + started.Notify(); + release.WaitForNotification(); + } + return FakeFs::ReadDir(dir); + } + + mutable absl::Notification started; + mutable absl::Notification release; +}; + +TEST_F(WalkFakeFsTest, StopWaitsForUnusedPrefetchBeforeReturningToSharedExecutor) { + BlockingReadFs blocking_fs; + std::vector siblings; + for (std::size_t index = 0; index < 65; ++index) { + const std::string name = absl::StrCat("d", index); + const std::string path = absl::StrCat("/r/", name); + siblings.push_back(DirEntry(path, name)); + blocking_fs.AddDir(path, 1, {}); + } + blocking_fs.AddDir("/r", 1, std::move(siblings)); + RunExecutor executor(3); + absl::Notification stopping; + auto walk = std::async(std::launch::async, [&] { + return Walk( + blocking_fs, {"/r"}, WalkOptions{.sort = SortOrder::kTree, .workers = 3}, executor, + [&](const Visit& visit) { + if (visit.depth == 1) { + blocking_fs.started.WaitForNotification(); + stopping.Notify(); + return WalkAction::kStop; + } + return WalkAction::kContinue; + }, + [](std::string_view, absl::Status) {}); + }); + stopping.WaitForNotification(); + const auto while_blocked = walk.wait_for(std::chrono::milliseconds(100)); + // Release before asserting so a failed assertion cannot strand a worker. + blocking_fs.release.Notify(); + EXPECT_THAT(while_blocked, Eq(std::future_status::timeout)); + EXPECT_THAT(walk.get(), IsOk()); + EXPECT_THAT(executor.worker_count(), Eq(3)); + EXPECT_THAT(executor.Submit([] { return 42; }).Get(), Eq(42)); +} + TEST_F(WalkFakeFsTest, XdevStopsAtDeviceBoundary) { // /r (dev 1) holds a.txt (dev 1) and the mount point mnt (dev 2), whose child // x.txt (dev 2) lives on the other filesystem. From 1358d1fc4dcf942581036076ea1dc4fa731f82cc Mon Sep 17 00:00:00 2001 From: helly25 Date: Sat, 10 Oct 2026 11:28:18 +0100 Subject: [PATCH 4/4] Fix shared executor lint and timing-dependent MSan assertions Address all six clang-tidy diagnostics in Walker and RunExecutorTest. Explicitly forbid copying or moving a Walker whose drain jobs capture it, use const observers where possible, and reserve the bounded task vector. The matcher intentionally admits workers early when its timed prefix is slow. Test the four-worker budget and eventual accumulated-work admission without assuming sanitizer or scheduler speed. Keep runtime thresholds and downstream caller-inclusive budgeting unchanged. --- xff/engine/parallel_match_test.cc | 7 +++++-- xff/engine/run_executor_test.cc | 5 +++-- xff/engine/walk.cc | 10 ++++++++-- 3 files changed, 16 insertions(+), 6 deletions(-) diff --git a/xff/engine/parallel_match_test.cc b/xff/engine/parallel_match_test.cc index 4ef6d3b6268..44d6b2506d7 100644 --- a/xff/engine/parallel_match_test.cc +++ b/xff/engine/parallel_match_test.cc @@ -24,6 +24,7 @@ namespace { using ::testing::Eq; using ::testing::IsFalse; using ::testing::IsTrue; +using ::testing::Le; using ::testing::SizeIs; struct ParallelMatchTest : ::testing::Test { @@ -117,13 +118,15 @@ TEST_F(ParallelMatchTest, PreparedCoordinatorAndWorkersRetainRegexStateAcrossPoo EXPECT_THAT(executor_pool.worker_count(), Eq(4)); } -TEST_F(ParallelMatchTest, ColdExecutorWaitsForAccumulatedWork) { +TEST_F(ParallelMatchTest, AccumulatedWorkStartsWorkersWithinBudget) { MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-true"})); RunExecutor executor(4); ParallelMatch matcher(*command.expression, 4, executor, false); for (int batch = 0; batch < 7; ++batch) { EXPECT_THAT(matcher.Match(Entries(1'024)), SizeIs(1'024)); - EXPECT_THAT(executor.worker_count(), Eq(0)); + // A slow prefix can activate workers before the accumulated-entry threshold. + // Sanitizer overhead and scheduling can trigger that path even for -true. + EXPECT_THAT(executor.worker_count(), Le(4)); } EXPECT_THAT(matcher.Match(Entries(1'024)), SizeIs(1'024)); EXPECT_THAT(executor.worker_count(), Eq(4)); diff --git a/xff/engine/run_executor_test.cc b/xff/engine/run_executor_test.cc index 5f25f6f7187..3e0bf0381e1 100644 --- a/xff/engine/run_executor_test.cc +++ b/xff/engine/run_executor_test.cc @@ -47,6 +47,7 @@ TEST_F(RunExecutorTest, ReusesTheSameBoundedThreadsAcrossPhases) { const auto phase = [&] { absl::Barrier started(3); std::vector> tasks; + tasks.reserve(3); for (int worker = 0; worker < 3; ++worker) { tasks.push_back(executor.Submit([&] { started.Block(); @@ -68,8 +69,8 @@ TEST_F(RunExecutorTest, ReusesTheSameBoundedThreadsAcrossPhases) { TEST_F(RunExecutorTest, SupportsMoveOnlyJobsAndResults) { RunExecutor executor(1); executor.Start(1); - auto task = executor.Submit([value = std::make_unique(42)]() mutable { return std::move(value); }); - auto result = task.Get(); + auto task = executor.Submit([value = std::make_unique(42)] mutable { return std::move(value); }); + const auto result = task.Get(); EXPECT_THAT(*result, Eq(42)); EXPECT_THAT(RunTask{}.Valid(), IsFalse()); executor.Submit([] {}).Get(); diff --git a/xff/engine/walk.cc b/xff/engine/walk.cc index cec68b8ed2c..f3560908ce2 100644 --- a/xff/engine/walk.cc +++ b/xff/engine/walk.cc @@ -90,11 +90,17 @@ class Walker { follow_children_(options.symlinks == SymlinkMode::kAll), executor_(executor) {} + // Leaf jobs capture this walker, so its identity must remain stable until drained. + Walker(const Walker&) = delete; + Walker& operator=(const Walker&) = delete; + Walker(Walker&&) = delete; + Walker& operator=(Walker&&) = delete; + ~Walker() { // A stop action may leave prefetched siblings unconsumed. Their drain jobs // still reference this walker, so join those jobs before its state is torn // down. Normal walks reach already-completed jobs here. - for (RunTask& drain : drain_tasks_) { + for (const RunTask& drain : drain_tasks_) { drain.Wait(); } } @@ -538,7 +544,7 @@ class Walker { std::atomic next = 0; }; - auto batch = std::make_shared(); + const auto batch = std::make_shared(); batch->paths.reserve(directories.size()); batch->promises.resize(directories.size()); for (std::size_t batch_index = 0; batch_index < directories.size(); ++batch_index) {