diff --git a/XFF.md b/XFF.md index c258f2b6288..6044130d008 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 using one command-owned, lazily started pool of up to `N-1` background workers plus the calling coordinator, reused for directory reads, required eager metadata, eligible content matching and file comparison. Comparison trees are coordinated sequentially within that same budget. The caller participates in eligible content and comparison work. 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. + `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 command-owned, lazily started pool of up to `N-1` background workers plus the calling coordinator, reused for directory reads, required eager metadata, eligible content matching and file comparison. Comparison trees are coordinated sequentially within that same budget. The caller participates in eligible content and comparison work. Directory read-ahead uses a bounded ordered window and a shared retained-storage allowance; oversized speculative listings are read on demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight filesystem call must finish. Workers reuse the same executor for 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 176a99120d3..42c719c16e8 100644 --- a/docs/benchmark-correctness-audit.md +++ b/docs/benchmark-correctness-audit.md @@ -319,6 +319,39 @@ 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. +### Bounded directory-frontier experiment + +The command-ownership follow-up is draft [PR #996](https://github.com/mboworks/xff/pull/996). +A separate candidate based on that branch replaces whole-sibling directory drain +jobs with an ordered read-ahead window. Its persistent command executor dispatches +one leaf directory read per speculative task and alternates speculative and +foreground queues when both contain work. This is nonpreemptive job-boundary +fairness, not a guarantee against long foreground drains or blocking VFS calls. + +Nested and mounted-filesystem walks share at most twice the started background +worker count in unfinished reads and a provisional 8 MiB charged-storage allowance. +Pending requests charge bookkeeping/path storage before dispatch; completed +speculative listings add vector, string and status charges. Finished listings +release execution slots while keeping byte charges until consumed or destroyed. +This distinction permits deeper reads while older sibling listings remain cached. +Required listings, retained ancestor listings, in-flight VFS storage and allocator +overhead are excluded: the allowance is not a total-memory or RSS bound. + +Per-request and frontier cancellation skip unneeded queued reads after pruning or +termination. Already-running VFS calls finish, and the walker still joins its jobs +before borrowed state is destroyed. Oversized completed speculative listings are +discarded and read on demand; a listing already required by the caller belongs to +ordinary traversal working memory and can be transferred without a duplicate read. + +In-memory controls cover shared slot/byte admission, concurrent retention, +reservation ownership, queue-class fairness, pruning before filesystem access, +early-stop read counts, and nested progress while parent siblings remain cached. +The inherited directory/stat/content admission policies are deliberately unchanged +in this experiment. In particular, accumulated completed content work is still +not valid evidence of sufficient remaining work. The bookkeeping estimate, retained +byte policy and task granularity need allocation profiles and repeated native +measurements. No timing, total-memory improvement or macOS acceptance is claimed. + ### 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 @@ -330,8 +363,11 @@ 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; +controls must cover content, eager metadata and slow storage. Separate evidence +in [PR #995](https://github.com/mboworks/xff/pull/995) now retains the exact-source +Linux baseline, enumeration-only allocation profiles and coarse counters/stacks +for 20k/50k/100k single-worker workloads. Those baseline diagnostics do not complete +precise phase-level attribution or allocation controls for the other workloads, +platforms or new scheduler candidates. 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 b262228f4f4..c4972fd41b7 100644 --- a/docs/design-parallel.md +++ b/docs/design-parallel.md @@ -49,8 +49,9 @@ background workers under `--jobs=N`: result handles in the order required by `--sort`. Directory workers start only when at least 64 sibling directories provide independent work. For a coordinator-read directory with at least 512 children and eagerly required metadata, 128-entry stat chunks use the same executor. The - coordinator also claims stat chunks. At most one drain job per background worker is queued for - each batch; worker jobs never wait on nested jobs. Metadata-free, lazy and + coordinator also claims stat chunks. Directory reads use a bounded ordered window with + command-wide execution and retained-storage limits; stat and content batches use at most one + drain job per participating background worker. Worker jobs never wait on nested jobs. Metadata-free, lazy and owned/archive-source listings keep their serial per-directory path. - The coordinator applies traversal controls and writes the output sink in traversal order. Independent content predicates and rg content searches reuse that executor, with the caller also @@ -77,8 +78,10 @@ background workers under `--jobs=N`: workers share compiled patterns and reuse private match scratch, with bounded retained interpreter heap frames and independent reentrant leases. Entry content and lazy metadata have one owner and need no additional locks. -- A listing future may retain a completed directory read until the coordinator - reaches it. No worker emits a match and no traversal sort collects all matches. +- A listing result may retain a completed directory read until the coordinator + reaches it. Completed speculative results keep their storage charge but release + execution slots, so cached siblings do not by themselves prevent deeper reads. + No worker emits a match and no traversal sort collects all matches. Comparison collects both trees sequentially on the same calling coordinator, reusing the command's executor across the two walks. It then evaluates eligible large regular-file pairs @@ -156,20 +159,47 @@ The caller claims independent stat, content and comparison chunks rather than waiting idle while `N` additional workers run. Sequential comparison inventories remove the second coordinator, but may reduce overlap when each tree has a narrow frontier on slow storage; native A/B controls must quantify that tradeoff. -The executor still uses FIFO batch-drain jobs. A directory drain can occupy a -worker until its entire batch is exhausted, and completed read-ahead listings are -not bounded by a command-wide byte budget. Fair scheduling, bounded read-ahead, -cancellation, and useful-work admission based only on remaining work are still -required; a thread-count limit alone does not solve those problems. The cold +The bounded-frontier follow-up separates foreground stat/content/comparison jobs +from speculative directory reads. When both queues contain work, dispatch +alternates their classes, retaining FIFO order within each class. This is fairness +at job boundaries, not preemption: a large foreground drain or a blocking VFS call +still runs until it returns. Useful-work admission based only on remaining work, +task-size tuning and native timing/allocation validation remain required; the +thread-count and storage limits do not establish performance acceptance. The cold matcher policy below is inherited for isolation, not endorsed by this change. -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. +The coordinator currently admits directory read-ahead when at least 64 independent +sibling reads are available. This inherited count heuristic is provisional, not a +measurement of remaining cost. Each directory job performs one leaf VFS read plus +required metadata, publishes its result and yields before another directory. +Workers never recurse or wait on jobs in their own executor. Eager stat work still +uses drain jobs over 128-entry chunks after at least 512 entries are available. + +Each admitted directory frontier keeps an ordered window of at most twice its +participating background-worker count. All nested and mounted-filesystem walks +also share a command-wide limit of twice the started background-worker count in +unfinished reads. A consumed parent request is not immediately refilled before +descent, allowing the deeper frontier to use available slots. Completion releases +an unfinished-read slot, while cached siblings keep their separate byte charges. + +The experimental shared retained-storage allowance is 8 MiB. Every queued or +completed request first charges 512 bytes of bookkeeping plus its copied path; +completed speculative listings additionally charge vector/string capacities and +status storage. These conservative charges bound represented storage, not actual +allocator use or RSS; the 512-byte estimate and 8 MiB policy require native +allocation and performance controls. An unrepresentable or oversized speculative +listing is discarded and read on demand when visited. If the caller already needs +that result while its read is running, the required listing is transferred without +an additional lookahead charge, avoiding a second read. Required and ancestor +listings, in-flight VFS buffers and allocator overhead are outside this speculative +storage accounting; total traversal memory is not bounded by 8 MiB. + +Reservations are released by their owners on consumption or destruction. +Per-request cancellation handles pruning, and a frontier-wide flag handles early +termination. Workers check both before and after each VFS read. Queued canceled +requests skip filesystem access; already running VFS calls cannot be interrupted. +The walker drains all its submitted jobs before its borrowed filesystem state +disappears, leaving the executor reusable by the next phase. Content matching retains private evaluator and regex state for each logical worker. Decision-only matching collects up to 8,192 entries before flushing; @@ -214,14 +244,16 @@ design-config.md): ## Concurrency correctness - **`-prune`** - the coordinator evaluates the entry and suppresses descent when - the visitor returns prune. Depending on the ordering mode, batched read-ahead - may read the pruned directory's listing before or after that decision, but no - entry from the pruned subtree is visited or acted upon. + the visitor returns prune. Known prune decisions are excluded from admission; + destroying an unused speculative request cancels it. A read already started + before that decision can still finish, but no 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 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. + perform leaf directory reads and poll cancellation flags around each VFS call. + Queued unused reads skip the filesystem; an in-flight call finishes before its + 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 still means a parent waits on its descendants' completion. diff --git a/xff/cli/globals.cc b/xff/cli/globals.cc index ac94e6ca9cf..6b5179e6efb 100644 --- a/xff/cli/globals.cc +++ b/xff/cli/globals.cc @@ -907,8 +907,11 @@ constexpr std::array kGlobals = std::to_array({ "using one command-owned, lazily started pool of up to `N-1` background workers plus the calling " "coordinator, reused for directory reads, required eager metadata, eligible content matching " "and file comparison. Comparison trees are coordinated sequentially within that same budget. " - "The caller participates in eligible content and comparison work. Workers drain batches of independent " - "work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start " + "The caller participates in eligible content and comparison work. Directory read-ahead uses a bounded " + "ordered window and a shared retained-storage allowance; oversized speculative listings are read on " + "demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight " + "filesystem call must finish. Workers reuse the same executor for 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 " diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel index a9dc53f3fdb..f9b3a913751 100644 --- a/xff/engine/BUILD.bazel +++ b/xff/engine/BUILD.bazel @@ -38,6 +38,7 @@ cc_test( deps = [ ":run_executor_cc", "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/time", "@googletest//:gtest", "@googletest//:gtest_main", ], @@ -58,6 +59,7 @@ cc_library( "@abseil-cpp//absl/functional:function_ref", "@abseil-cpp//absl/status", "@abseil-cpp//absl/status:statusor", + "@abseil-cpp//absl/strings:cord", "@abseil-cpp//absl/synchronization", "@abseil-cpp//absl/types:span", "@mboworks_mbo//mbo/status:status_macros_cc", @@ -173,7 +175,9 @@ cc_test( "@abseil-cpp//absl/status", "@abseil-cpp//absl/status:statusor", "@abseil-cpp//absl/strings", + "@abseil-cpp//absl/strings:cord", "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/time", "@googletest//:gtest", "@googletest//:gtest_main", "@mboworks_mbo//mbo/testing:status_cc", diff --git a/xff/engine/run_executor.cc b/xff/engine/run_executor.cc index 50e2170e1e7..ce734f03d9a 100644 --- a/xff/engine/run_executor.cc +++ b/xff/engine/run_executor.cc @@ -4,9 +4,77 @@ #include "xff/engine/run_executor.h" #include +#include +#include +#include namespace xff::engine { +namespace executor_detail { +struct ReadAheadBudget final { + explicit ReadAheadBudget(std::size_t bytes) : limit(bytes) {} + + const std::size_t limit; + // The mutex guards slot and byte accounting, including peaks. A reservation + // exclusively owns its own charge; its shared budget outlives all reservations. + absl::Mutex mutex; + ReadAheadStats usage ABSL_GUARDED_BY(mutex); +}; +} // namespace executor_detail + +struct ReadAheadReservation::Impl final { + Impl(std::shared_ptr state, std::size_t initial_bytes) + : budget(std::move(state)), minimum_bytes(initial_bytes), bytes(initial_bytes) {} + + ~Impl() { + const absl::MutexLock lock(budget->mutex); + if (unfinished) { + --budget->usage.in_use; + } + budget->usage.retained_bytes -= bytes; + } + + Impl(const Impl&) = delete; + Impl& operator=(const Impl&) = delete; + Impl(Impl&&) = delete; + Impl& operator=(Impl&&) = delete; + + const std::shared_ptr budget; + const std::size_t minimum_bytes; + std::size_t bytes = 0; + bool unfinished = true; +}; + +ReadAheadReservation::ReadAheadReservation(std::unique_ptr impl) : impl_(std::move(impl)) {} + +ReadAheadReservation::~ReadAheadReservation() = default; +ReadAheadReservation::ReadAheadReservation(ReadAheadReservation&&) noexcept = default; +ReadAheadReservation& ReadAheadReservation::operator=(ReadAheadReservation&&) noexcept = default; + +bool ReadAheadReservation::Retain(std::size_t bytes) { + const absl::MutexLock lock(impl_->budget->mutex); + auto& usage = impl_->budget->usage; + const std::size_t others = usage.retained_bytes - impl_->bytes; + if (bytes < impl_->minimum_bytes || bytes > impl_->budget->limit - others) { + return false; + } + usage.retained_bytes = others + bytes; + usage.peak_retained_bytes = std::max(usage.peak_retained_bytes, usage.retained_bytes); + impl_->bytes = bytes; + return true; +} + +void ReadAheadReservation::Finish() { + const absl::MutexLock lock(impl_->budget->mutex); + if (impl_->unfinished) { + --impl_->budget->usage.in_use; + impl_->unfinished = false; + } +} + +RunExecutor::RunExecutor(std::size_t max_workers, std::size_t read_ahead_bytes) + : max_workers_(max_workers), read_ahead_bytes_(read_ahead_bytes) {} + RunExecutor::~RunExecutor() { { const absl::MutexLock lock(mutex_); @@ -19,6 +87,13 @@ RunExecutor::~RunExecutor() { void RunExecutor::Start(std::size_t useful_workers) { const std::size_t count = std::min(max_workers_, useful_workers); + if (count == 0) { + return; + } + if (threads_.empty()) { + const absl::MutexLock lock(mutex_); + queues_.emplace(); + } threads_.reserve(count); while (threads_.size() < count) { threads_.emplace_back([this] { Run(); }); @@ -26,7 +101,38 @@ void RunExecutor::Start(std::size_t useful_workers) { } bool RunExecutor::Pending() const { - return stop_ || !queue_.empty(); + return stop_ || (queues_ && (!queues_->foreground.empty() || !queues_->read_ahead.empty())); +} + +std::optional RunExecutor::ReserveReadAhead(std::size_t initial_bytes) { + if (threads_.empty()) { + return std::nullopt; + } + if (!read_ahead_budget_) { + read_ahead_budget_ = std::make_shared(read_ahead_bytes_); + } + const absl::MutexLock lock(read_ahead_budget_->mutex); + const auto capacity = std::min(threads_.size(), std::numeric_limits::max() / 2) * 2; + if (read_ahead_budget_->usage.in_use >= capacity + || initial_bytes > read_ahead_bytes_ - read_ahead_budget_->usage.retained_bytes) { + return std::nullopt; + } + auto impl = std::make_unique(read_ahead_budget_, initial_bytes); + ++read_ahead_budget_->usage.in_use; + read_ahead_budget_->usage.retained_bytes += initial_bytes; + read_ahead_budget_->usage.peak_retained_bytes = + std::max(read_ahead_budget_->usage.peak_retained_bytes, read_ahead_budget_->usage.retained_bytes); + read_ahead_budget_->usage.peak_in_use = + std::max(read_ahead_budget_->usage.peak_in_use, read_ahead_budget_->usage.in_use); + return ReadAheadReservation(std::move(impl)); +} + +ReadAheadStats RunExecutor::ReadAheadUsage() const { + if (!read_ahead_budget_) { + return {}; + } + const absl::MutexLock lock(read_ahead_budget_->mutex); + return read_ahead_budget_->usage; } void RunExecutor::Run() { @@ -34,11 +140,20 @@ void RunExecutor::Run() { absl::AnyInvocable job; { const absl::MutexLock lock(mutex_, absl::Condition(this, &RunExecutor::Pending)); - if (stop_ && queue_.empty()) { + if (!queues_) { + return; // An unstarted executor has no queued work to dispatch. + } + auto& queues = queues_.value(); + if (stop_ && queues.foreground.empty() && queues.read_ahead.empty()) { return; } - job = std::move(queue_.front()); - queue_.pop(); + // Alternating classes prevents an older speculative frontier from sitting + // ahead of all foreground work, without starving speculation in a busy pool. + const bool read_ahead = !queues.read_ahead.empty() && (queues.foreground.empty() || queues.prefer_read_ahead); + auto& queue = read_ahead ? queues.read_ahead : queues.foreground; + job = std::move(queue.front()); + queue.pop(); + queues.prefer_read_ahead = !read_ahead; } job(); } diff --git a/xff/engine/run_executor.h b/xff/engine/run_executor.h index f5116f10f9a..7a5ab66b1fd 100644 --- a/xff/engine/run_executor.h +++ b/xff/engine/run_executor.h @@ -22,6 +22,8 @@ namespace xff::engine { namespace executor_detail { +struct ReadAheadBudget; + template struct TaskState final { using Stored = std::conditional_t, std::monostate, Result>; @@ -30,6 +32,41 @@ struct TaskState final { }; } // namespace executor_detail +enum class RunTaskClass { kForeground, kReadAhead }; + +struct ReadAheadStats { + // Unfinished reads. Completed results keep their byte charge, not an execution slot. + std::size_t in_use = 0; + // Pending/finished request bookkeeping plus retained speculative listing storage. + std::size_t retained_bytes = 0; + std::size_t peak_in_use = 0; + std::size_t peak_retained_bytes = 0; +}; + +// An owned speculative-read reservation. Retain replaces its total byte charge, +// never below the initial bookkeeping charge; a failed charge must discard the +// additional storage instead of publishing it. Finish releases the unfinished-read +// slot without releasing bytes. Destruction releases any remaining slot and bytes. +// The reservation can outlive its executor, but only one owner may mutate it. +class ReadAheadReservation final { + public: + ~ReadAheadReservation(); + ReadAheadReservation(ReadAheadReservation&&) noexcept; + ReadAheadReservation& operator=(ReadAheadReservation&&) noexcept; + ReadAheadReservation(const ReadAheadReservation&) = delete; + ReadAheadReservation& operator=(const ReadAheadReservation&) = delete; + + [[nodiscard]] bool Retain(std::size_t bytes); + // Idempotent. Completed cached results must not occupy execution slots. + void Finish(); + + private: + friend class RunExecutor; + struct Impl; + explicit ReadAheadReservation(std::unique_ptr impl); + std::unique_ptr impl_; +}; + template class RunPromise; @@ -100,7 +137,9 @@ class RunPromise final { // 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) {} + static constexpr std::size_t kReadAheadBytes = 8 * 1'024 * 1'024; + + explicit RunExecutor(std::size_t max_workers, std::size_t read_ahead_bytes = kReadAheadBytes); ~RunExecutor(); @@ -111,8 +150,17 @@ class RunExecutor final { void Start(std::size_t useful_workers); + // Coordinator-only admission and inspection. All walks sharing this executor use + // the same byte allowance and at most twice the started worker count in unfinished reads. + // Initial bytes charge the request's bookkeeping/path before it can enter the queue. + // In-flight VFS calls and the coordinator's required listing are not retained lookahead. + [[nodiscard]] std::optional ReserveReadAhead(std::size_t initial_bytes = 0); + ReadAheadStats ReadAheadUsage() const; + + std::size_t ReadAheadLimit() const { return read_ahead_bytes_; } + template - auto Submit(Job job) ABSL_LOCKS_EXCLUDED(mutex_) { + auto Submit(Job job, RunTaskClass task_class = RunTaskClass::kForeground) ABSL_LOCKS_EXCLUDED(mutex_) { using Result = std::invoke_result_t; RunPromise promise; auto result = promise.Task(); @@ -129,7 +177,12 @@ class RunExecutor final { return result; } const absl::MutexLock lock(mutex_); - queue_.emplace(std::move(task)); + auto& queues = queues_.value(); + if (task_class == RunTaskClass::kReadAhead) { + queues.read_ahead.emplace(std::move(task)); + } else { + queues.foreground.emplace(std::move(task)); + } return result; } @@ -139,9 +192,24 @@ class RunExecutor final { bool Pending() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); void Run() ABSL_LOCKS_EXCLUDED(mutex_); + struct DispatchQueues { + std::queue> foreground; + std::queue> read_ahead; + bool prefer_read_ahead = false; + }; + const std::size_t max_workers_; + const std::size_t read_ahead_bytes_; + // Initialized lazily by the coordinator before publishing the first reservation. + // Workers access only their shared reservation state, never this owning handle. + std::shared_ptr read_ahead_budget_; + // The mutex guards both queues, dispatch preference and shutdown. Jobs run unlocked; + // the coordinator alone grows threads_ and inspects their count. Dispatch and + // shared-budget mutexes are never held simultaneously by this implementation. mutable absl::Mutex mutex_; - std::queue> queue_ ABSL_GUARDED_BY(mutex_); + // std::queue's default deque allocates even while empty. Inline commands never + // construct these queues; Start initializes them before publishing any worker. + std::optional queues_ 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 3e0bf0381e1..10c8cf768f0 100644 --- a/xff/engine/run_executor_test.cc +++ b/xff/engine/run_executor_test.cc @@ -3,6 +3,7 @@ #include "xff/engine/run_executor.h" +#include #include #include #include @@ -10,16 +11,22 @@ #include #include "absl/synchronization/barrier.h" +#include "absl/time/time.h" #include "gmock/gmock.h" #include "gtest/gtest.h" namespace xff::engine { namespace { +using ::testing::_; using ::testing::Contains; +using ::testing::ElementsAre; using ::testing::Eq; using ::testing::IsFalse; +using ::testing::IsTrue; +using ::testing::Le; using ::testing::Ne; using ::testing::Not; +using ::testing::Optional; using ::testing::SizeIs; struct RunExecutorTest : ::testing::Test {}; @@ -76,5 +83,178 @@ TEST_F(RunExecutorTest, SupportsMoveOnlyJobsAndResults) { executor.Submit([] {}).Get(); } +TEST_F(RunExecutorTest, ReadAheadSlotsAndBytesAreSharedAndReleasedByTheirOwner) { + RunExecutor executor(2, 100); + EXPECT_THAT(executor.ReserveReadAhead(), Eq(std::nullopt)); + executor.Start(1); + auto first = executor.ReserveReadAhead(); + auto second = executor.ReserveReadAhead(); + ASSERT_THAT(first, Optional(_)); + ASSERT_THAT(second, Optional(_)); + // The assertions report absence; this guard also proves engagement to static analysis. + if (!first || !second) { + return; + } + EXPECT_THAT(executor.ReserveReadAhead(), Eq(std::nullopt)); + EXPECT_THAT(first.value().Retain(60), IsTrue()); + EXPECT_THAT(second.value().Retain(41), IsFalse()); + EXPECT_THAT(second.value().Retain(40), IsTrue()); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(100)); + EXPECT_THAT(first.value().Retain(std::numeric_limits::max()), IsFalse()); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(100)); + first.reset(); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(40)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(1)); + executor.Start(2); + std::vector additional; + additional.reserve(3); + for (int index = 0; index < 3; ++index) { + auto reservation = executor.ReserveReadAhead(); + ASSERT_THAT(reservation, Optional(_)); + if (!reservation) { + return; + } + additional.push_back(std::move(reservation.value())); + } + EXPECT_THAT(executor.ReserveReadAhead(), Eq(std::nullopt)); + EXPECT_THAT(executor.ReadAheadUsage().peak_in_use, Eq(4)); + EXPECT_THAT(executor.ReadAheadUsage().peak_retained_bytes, Eq(100)); + additional.clear(); + second.reset(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); +} + +TEST_F(RunExecutorTest, ConcurrentReadRetainersCannotExceedTheirSharedByteAllowance) { + RunExecutor executor(3, 100); + executor.Start(3); + absl::Barrier ready(3); + std::vector> tasks; + tasks.reserve(3); + for (int index = 0; index < 3; ++index) { + auto reservation = executor.ReserveReadAhead(); + ASSERT_THAT(reservation, Optional(_)); + tasks.push_back(executor.Submit([&ready, reservation = std::move(reservation)] mutable { + ready.Block(); + const bool retained = reservation.value().Retain(40); + reservation.reset(); + return retained; + })); + } + // Each job releases its charge at completion, so successes are not required to + // overlap. The measured peak must always fit, regardless of worker ordering. + for (auto& task : tasks) { + task.Get(); + } + EXPECT_THAT(executor.ReadAheadUsage().peak_retained_bytes, Le(100)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); +} + +TEST_F(RunExecutorTest, CompletedListingsRetainBytesWithoutBlockingDeeperReadSlots) { + RunExecutor executor(1, 100); + executor.Start(1); + auto first = executor.ReserveReadAhead(20); + auto second = executor.ReserveReadAhead(20); + ASSERT_THAT(first, Optional(_)); + ASSERT_THAT(second, Optional(_)); + if (!first || !second) { + return; + } + EXPECT_THAT(executor.ReserveReadAhead(1), Eq(std::nullopt)); + first.value().Finish(); + second.value().Finish(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(40)); + EXPECT_THAT(executor.ReserveReadAhead(61), Eq(std::nullopt)); + auto deeper = executor.ReserveReadAhead(60); + ASSERT_THAT(deeper, Optional(_)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(1)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(100)); + deeper.reset(); + first.reset(); + second.reset(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); +} + +TEST_F(RunExecutorTest, FinishingIsIdempotentAndCannotDropTheBookkeepingCharge) { + RunExecutor executor(1, 100); + executor.Start(1); + auto reservation = executor.ReserveReadAhead(20); + ASSERT_THAT(reservation, Optional(_)); + if (!reservation) { + return; + } + EXPECT_THAT(reservation.value().Retain(19), IsFalse()); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(20)); + EXPECT_THAT(reservation.value().Retain(100), IsTrue()); + reservation.value().Finish(); + reservation.value().Finish(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(100)); + EXPECT_THAT(reservation.value().Retain(0), IsFalse()); + EXPECT_THAT(reservation.value().Retain(20), IsTrue()); + auto unfinished = executor.ReserveReadAhead(80); + ASSERT_THAT(unfinished, Optional(_)); + reservation.reset(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(1)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(80)); + unfinished.reset(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); +} + +TEST_F(RunExecutorTest, ForegroundAndReadAheadQueuesAlternateWithoutStarvation) { + RunExecutor executor(1); + executor.Start(1); + absl::Notification started; + absl::Notification release; + std::vector order; + auto initial = executor.Submit( + [&] { + started.Notify(); + release.WaitForNotification(); + order.push_back(0); + }, + RunTaskClass::kReadAhead); + const bool running = started.WaitForNotificationWithTimeout(absl::Seconds(5)); + std::vector> tasks; + tasks.reserve(5); + tasks.push_back(executor.Submit([&] { order.push_back(1); }, RunTaskClass::kReadAhead)); + tasks.push_back(executor.Submit([&] { order.push_back(2); }, RunTaskClass::kReadAhead)); + tasks.push_back(executor.Submit([&] { order.push_back(3); })); + tasks.push_back(executor.Submit([&] { order.push_back(4); })); + tasks.push_back(executor.Submit([&] { order.push_back(5); })); + release.Notify(); + initial.Get(); + for (auto& task : tasks) { + task.Get(); + } + EXPECT_THAT(running, IsTrue()); + EXPECT_THAT(order, ElementsAre(0, 3, 1, 4, 2, 5)); +} + +TEST_F(RunExecutorTest, AReservationCanMoveAcrossThreadsAndOutliveItsExecutor) { + std::optional retained; + { + RunExecutor executor(1, 100); + executor.Start(1); + auto reservation = executor.ReserveReadAhead(); + ASSERT_THAT(reservation, Optional(_)); + if (!reservation) { + return; + } + auto task = executor.Submit([reservation = std::move(reservation.value())] mutable { + EXPECT_THAT(reservation.Retain(50), IsTrue()); + return std::move(reservation); + }); + retained.emplace(task.Get()); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(1)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(50)); + } + EXPECT_THAT(retained.value().Retain(100), IsTrue()); + EXPECT_THAT(retained.value().Retain(101), IsFalse()); +} + } // namespace } // namespace xff::engine diff --git a/xff/engine/walk.cc b/xff/engine/walk.cc index 64476713aa3..3c75edd8b50 100644 --- a/xff/engine/walk.cc +++ b/xff/engine/walk.cc @@ -19,8 +19,11 @@ #include #include #include +#include +#include #include #include +#include #include #include #include @@ -31,6 +34,7 @@ #include "absl/base/thread_annotations.h" #include "absl/status/status.h" #include "absl/status/statusor.h" +#include "absl/strings/cord.h" #include "absl/synchronization/mutex.h" #include "absl/types/span.h" #include "mbo/status/status_macros.h" @@ -67,12 +71,91 @@ struct Stated { // The result of reading one directory: its children, or a ReadDir error. using Listing = absl::StatusOr>; +struct PrefetchedListing { + std::optional listing; + ReadAheadReservation reservation; +}; + +struct ReadDemand { + // Only the coordinator changes these flags; workers poll around the VFS call. + // They do not publish data: result handoff uses RunPromise completion instead. + std::atomic required = false; + std::atomic canceled = false; +}; + +class PrefetchedRead final { + public: + PrefetchedRead() = default; + + PrefetchedRead(RunTask task, std::shared_ptr demand) + : task_(std::move(task)), demand_(std::move(demand)) {} + + ~PrefetchedRead() { + if (demand_) { + demand_->canceled.store(true, std::memory_order_relaxed); + } + } + + PrefetchedRead(const PrefetchedRead&) = delete; + PrefetchedRead& operator=(const PrefetchedRead&) = delete; + PrefetchedRead(PrefetchedRead&&) = default; + PrefetchedRead& operator=(PrefetchedRead&&) = delete; + + bool Valid() const { return task_.Valid(); } + + PrefetchedListing Get() { + // Only the coordinator can require a listing, immediately before waiting for + // it. That one result belongs to ordinary traversal working memory, not cache. + demand_->required.store(true, std::memory_order_relaxed); + return task_.Get(); + } + + private: + RunTask task_; + std::shared_ptr demand_; +}; + +std::size_t AddBytes(std::size_t lhs, std::size_t rhs) { + const auto limit = std::numeric_limits::max(); + return rhs > limit - lhs ? limit : lhs + rhs; +} + +std::size_t StatusBytes(absl::Status status) { + std::size_t bytes = status.message().size(); + status.ForEachPayload([&](const std::string_view key, const absl::Cord& payload) { + bytes = AddBytes(bytes, AddBytes(key.size(), payload.EstimatedMemoryUsage())); + }); + return bytes; +} + +// Conservative storage charges, not RSS: inline string capacity and shared status +// payloads can be counted twice. Allocator overhead and the VFS's working memory +// are not represented. Saturation rejects unrepresentable charges safely. +std::size_t ListingBytes(const Listing& listing) { + std::size_t bytes = sizeof(Listing); + if (!listing.ok()) { + return AddBytes(bytes, StatusBytes(listing.status())); + } + bytes = AddBytes(bytes, listing->capacity() * sizeof(Stated)); + for (const Stated& child : *listing) { + bytes = AddBytes(bytes, AddBytes(child.path.capacity(), 1)); + bytes = AddBytes(bytes, AddBytes(child.name.capacity(), 1)); + bytes = AddBytes(bytes, StatusBytes(child.status)); + } + return bytes; +} + // 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 conservative bookkeeping charge for each queued or completed request, excluding +// its path and listing. This is charged before dispatch, even if its result is discarded. +// The allowance is a charged-storage limit, not an allocator/RSS measurement. +inline constexpr std::size_t kReadAheadRequestBytes = 512; + class Walker { public: Walker( @@ -307,6 +390,149 @@ class Walker { return children; } + // A bounded ordered window. Requests share command-wide reservations, including + // across nested walkers. Each executor job performs one leaf directory read, so + // it yields before another directory and can be skipped when this window closes. + class ReadFrontier final { + public: + ReadFrontier( + Walker& walker, + const std::vector& children, + int depth, + mbo::types::OptionalRef> pruned = std::nullopt) + : walker_(walker), children_(children), depth_(depth), pruned_(pruned) { + if (walker_.options_.workers <= 1) { + return; + } + std::size_t directories = 0; + std::size_t minimum_bytes = std::numeric_limits::max(); + // The index aligns the optional prune decisions with their child records. + for (std::size_t index = 0; index < children_.size(); ++index) { + if (Eligible(index)) { + ++directories; + minimum_bytes = std::min(minimum_bytes, RequestBytes(index)); + } + } + if (directories < kMinParallelDirectoryReads) { + return; + } + const auto capacity = walker_.executor_.ReadAheadLimit() / minimum_bytes; + if (capacity == 0) { + return; + } + walker_.executor_.Start(std::min({directories, capacity, walker_.options_.workers - 1})); + window_ = 2 * std::min(walker_.options_.workers - 1, walker_.executor_.worker_count()); + canceled_ = std::make_shared>(false); + requests_.emplace(); + Fill(); + } + + ~ReadFrontier() { + if (canceled_) { + canceled_->store(true, std::memory_order_relaxed); + } + } + + ReadFrontier(const ReadFrontier&) = delete; + ReadFrontier& operator=(const ReadFrontier&) = delete; + ReadFrontier(ReadFrontier&&) = delete; + ReadFrontier& operator=(ReadFrontier&&) = delete; + + PrefetchedRead Take(std::size_t index) { + if (!requests_) { + return {}; + } + auto& requests = requests_.value(); + while (!requests.empty() && requests.front().index < index) { + requests.pop_front(); + } + next_ = std::max(next_, index); + Fill(); + if (requests.empty() || requests.front().index != index) { + return {}; + } + auto task = std::move(requests.front().task); + requests.pop_front(); + // Do not refill before descent. The consumed slot becomes available to the + // deeper frontier; older siblings cannot repeatedly take it ahead of that work. + return task; + } + + private: + struct Request { + std::size_t index; + PrefetchedRead task; + }; + + bool Eligible(std::size_t index) const { + return (!pruned_ || !pruned_->at(index)) && walker_.Descendable(children_.at(index), depth_); + } + + std::size_t RequestBytes(std::size_t index) const { + return AddBytes(kReadAheadRequestBytes, AddBytes(children_.at(index).path.size(), 1)); + } + + void Fill() { + if (!requests_) { + return; + } + auto& requests = requests_.value(); + std::erase_if(walker_.drain_tasks_, [](const RunTask& task) { return task.Ready(); }); + while (requests.size() < window_ && next_ < children_.size()) { + if (!Eligible(next_)) { + ++next_; + continue; + } + const auto request_bytes = RequestBytes(next_); + auto reservation = walker_.executor_.ReserveReadAhead(request_bytes); + if (!reservation) { + return; + } + RunPromise promise; + auto task = promise.Task(); + const auto demand = std::make_shared(); + walker_.drain_tasks_.push_back(walker_.executor_.Submit( + [&walker = walker_, canceled = canceled_, demand, request_bytes, path = children_.at(next_).path, + promise = std::move(promise), reservation = std::move(*reservation)] mutable { + std::optional listing; + if (!canceled->load(std::memory_order_relaxed) && !demand->canceled.load(std::memory_order_relaxed)) { + auto read = walker.ReadDir(path); + if (!canceled->load(std::memory_order_relaxed) && !demand->canceled.load(std::memory_order_relaxed) + && (demand->required.load(std::memory_order_relaxed) + || reservation.Retain(AddBytes(request_bytes, ListingBytes(read))))) { + listing.emplace(std::move(read)); + } + } + reservation.Finish(); + // Release the producer handle before the executor signals this job's + // completion. Otherwise an unobserved result can keep its credit alive + // briefly after Walk has joined the job and inspected the shared budget. + auto completed = std::move(promise); + completed.SetValue({ + .listing = std::move(listing), + .reservation = std::move(reservation), + }); + }, + RunTaskClass::kReadAhead)); + requests.push_back({ + .index = next_, + .task = PrefetchedRead(std::move(task), demand), + }); + ++next_; + } + } + + Walker& walker_; + const std::vector& children_; + const int depth_; + const mbo::types::OptionalRef> pruned_; + std::size_t next_ = 0; + std::size_t window_ = 0; + std::shared_ptr> canceled_; + // No deque allocation in inline walks or below the admission threshold. + std::optional> requests_; + }; + 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; @@ -368,8 +594,8 @@ class Walker { // Visits `stated` and, if it is a descendable directory, descends into it. // 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) { + // already-submitted listing read (from the parent's window), or empty to read now. + void VisitSubtree(const Stated& stated, int depth, mbo::types::OptionalRef prefetched) { if (stopped_) { return; } @@ -411,13 +637,23 @@ class Walker { // 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); + // Required listings and in-flight VFS reads are not speculative storage. An + // oversized speculative result was discarded and is read on demand here. + Listing listing = [&] -> Listing { + if (prefetched) { + auto result = prefetched->Get(); + if (result.listing) { + return std::move(*result.listing); + } + } + return 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()))) { @@ -434,22 +670,23 @@ class Walker { } // Recurses a directory's (already sorted) children. The sort modes differ only - // in how a subdirectory's entry is grouped relative to its subtree. Reads for - // the descendable subdirectories are submitted as a batch up front so the pool - // overlaps their IO while the coordinator visits in order. + // in how a subdirectory's entry is grouped relative to its subtree. Read-ahead + // for a bounded window of descendable subdirectories overlaps their IO while the + // coordinator visits in order. Nested windows share execution and byte limits. // NOLINTNEXTLINE(readability-function-cognitive-complexity): dispatching the ordering modes is cohesive. void HandleChildren(const std::vector& children, int depth) { // 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); + ReadFrontier reads(*this, children, depth); for (std::size_t i = 0; i < children.size(); ++i) { if (stopped_) { return; } + auto read = reads.Take(i); VisitSubtree( children[i], depth, - reads[i].Valid() ? mbo::types::OptionalRef{reads[i]} : mbo::types::OptionalRef>{}); + read.Valid() ? mbo::types::OptionalRef{read} : mbo::types::OptionalRef{}); } return; } @@ -465,15 +702,16 @@ class Walker { VisitOne(children[i], depth, options_.mount_before_visit && dived[i]); } } - std::vector> reads = SubmitSubdirReads(children, depth); + ReadFrontier reads(*this, children, depth); for (std::size_t i = 0; i < children.size(); ++i) { if (stopped_) { return; } if (IsDir(children[i])) { + auto read = reads.Take(i); VisitSubtree( children[i], depth, - reads[i].Valid() ? mbo::types::OptionalRef{reads[i]} : mbo::types::OptionalRef>{}); + read.Valid() ? mbo::types::OptionalRef{read} : 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. @@ -493,7 +731,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); + ReadFrontier reads(*this, children, depth, pruned); for (std::size_t i = 0; i < children.size(); ++i) { if (stopped_) { return; @@ -502,10 +740,11 @@ class Walker { continue; } if (Descendable(children[i], depth)) { + auto read = reads.Take(i); Descend( children[i], depth, - reads[i].Valid() ? mbo::types::OptionalRef{reads[i]} - : mbo::types::OptionalRef>{}); // entry already visited above + read.Valid() ? mbo::types::OptionalRef{read} + : 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. @@ -514,61 +753,6 @@ class Walker { } } - // 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()); - if (options_.workers <= 1) { - return reads; - } - std::vector directories; - for (std::size_t index = 0; index < children.size(); ++index) { - if ((pruned.empty() || !pruned[index]) && Descendable(children[index], depth)) { - directories.push_back(index); - } - } - if (directories.size() < kMinParallelDirectoryReads) { - return reads; - } - std::erase_if(drain_tasks_, [](RunTask& drain) { return drain.Ready(); }); - executor_.Start(std::min(directories.size(), options_.workers - 1)); - - struct ReadBatch final { - std::vector paths; - std::vector> promises; - std::atomic next = 0; - }; - - 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) { - 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(), options_.workers - 1}); - 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; - } - const vfs::FileSystem& fs_; const WalkOptions& options_; Visitor visit_; diff --git a/xff/engine/walk_test.cc b/xff/engine/walk_test.cc index 4c5edea6e14..65146a9af63 100644 --- a/xff/engine/walk_test.cc +++ b/xff/engine/walk_test.cc @@ -16,6 +16,7 @@ #include "xff/engine/walk.h" #include +#include #include #include #include @@ -33,8 +34,10 @@ #include "absl/functional/function_ref.h" #include "absl/status/status.h" #include "absl/status/statusor.h" +#include "absl/strings/cord.h" #include "absl/strings/str_cat.h" #include "absl/synchronization/notification.h" +#include "absl/time/time.h" #include "gmock/gmock.h" #include "gtest/gtest.h" #include "mbo/testing/status.h" @@ -46,7 +49,6 @@ namespace xff::engine { namespace { -using ::testing::IsTrue; // gtest struct-fixture idioms suppressed file-wide: SetUp/TearDown are public overrides of the // base's protected hooks; fixture state carries the private-style `_` suffix; and the data-heavy // TEST_F bodies plus trivial local names (a one-line path lambda) trip cognitive-complexity and @@ -55,9 +57,14 @@ using ::testing::IsTrue; namespace fs = ::std::filesystem; using ::mbo::testing::IsOk; +using ::mbo::testing::StatusIs; using ::testing::ElementsAre; using ::testing::Eq; +using ::testing::Gt; +using ::testing::HasSubstr; using ::testing::IsEmpty; +using ::testing::IsTrue; +using ::testing::Le; using ::testing::Lt; using ::testing::Not; using ::testing::Pair; @@ -485,6 +492,249 @@ TEST_F(WalkFakeFsTest, StopWaitsForUnusedPrefetchBeforeReturningToSharedExecutor EXPECT_THAT(executor.Submit([] { return 42; }).Get(), Eq(42)); } +class CountingReadFs final : public FakeFs { + public: + absl::StatusOr> ReadDir(std::string_view dir) const override { + if (dir != "/r") { + reads.fetch_add(1, std::memory_order_relaxed); + } + if (dir == "/r/d0") { + pruned_reads.fetch_add(1, std::memory_order_relaxed); + } + if (dir.starts_with("/r/d0/")) { + nested_reads.fetch_add(1, std::memory_order_relaxed); + } + return FakeFs::ReadDir(dir); + } + + mutable std::atomic reads = 0; + mutable std::atomic pruned_reads = 0; + mutable std::atomic nested_reads = 0; +}; + +TEST_F(WalkFakeFsTest, PruningCancelsAReadThatHasNotReachedTheFilesystem) { + CountingReadFs 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)); + fs.AddDir(path, 1, {}); + } + fs.AddDir("/r", 1, std::move(siblings)); + RunExecutor executor(1); + executor.Start(1); + absl::Notification started; + absl::Notification release; + absl::Notification passed_pruned; + auto occupied = executor.Submit([&] { + started.Notify(); + release.WaitForNotification(); + }); + const bool running = started.WaitForNotificationWithTimeout(absl::Seconds(5)); + auto walk = std::async(std::launch::async, [&] { + return Walk( + fs, {"/r"}, {.sort = SortOrder::kTree, .workers = 2}, executor, + [&](const Visit& visit) { + if (visit.path == "/r/d0") { + return WalkAction::kPrune; + } + if (visit.path == "/r/d1") { + // The d0 request has been destroyed before the next visit starts. + passed_pruned.Notify(); + } + return WalkAction::kContinue; + }, + [](std::string_view, absl::Status) {}); + }); + const bool pruned = passed_pruned.WaitForNotificationWithTimeout(absl::Seconds(5)); + release.Notify(); + occupied.Get(); + EXPECT_THAT(walk.get(), IsOk()); + EXPECT_THAT(running, IsTrue()); + EXPECT_THAT(pruned, IsTrue()); + EXPECT_THAT(fs.pruned_reads.load(), Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); +} + +TEST_F(WalkFakeFsTest, EarlyStopDoesNotDrainTheWholeSpeculativeSiblingBatch) { + CountingReadFs fs; + std::vector siblings; + for (std::size_t index = 0; index < 1'000; ++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, {}); + } + fs.AddDir("/r", 1, std::move(siblings)); + RunExecutor executor(2); + EXPECT_THAT( + Walk( + fs, {"/r"}, {.sort = SortOrder::kTree, .workers = 3}, executor, + [](const Visit& visit) { return visit.depth == 1 ? WalkAction::kStop : WalkAction::kContinue; }, + [](std::string_view, absl::Status) {}), + IsOk()); + EXPECT_THAT(fs.reads.load(), Le(4)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().peak_in_use, Le(4)); + EXPECT_THAT(executor.Submit([] { return 42; }).Get(), Eq(42)); +} + +TEST_F(WalkFakeFsTest, NestedReadAheadSharesSlotsAndRejectsOversizedSpeculativeResults) { + 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 = absl::StrCat("/r/d0/", name); + nested.push_back(DirEntry(inner, name)); + const auto file = absl::StrCat(inner, "/", std::string(200, 'f')); + fs_.AddDir(inner, 1, {FileEntry(file, std::string(200, 'f'))}); + fs_.AddFile(file, 1); + } + fs_.AddDir("/r", 1, std::move(siblings)); + fs_.AddDir("/r/d0", 1, std::move(nested)); + static constexpr std::array kSortOrders = std::to_array({ + SortOrder::kDir, + SortOrder::kGlobal, + SortOrder::kNone, + SortOrder::kRoots, + SortOrder::kSubtree, + SortOrder::kTree, + }); + for (const SortOrder sort : kSortOrders) { + SCOPED_TRACE(static_cast(sort)); + for (const bool post_order : {false, true}) { + SCOPED_TRACE(post_order); + const auto expected = Seen({.post_order = post_order, .sort = sort, .workers = 1}); + for (const std::size_t budget : {0, 128, 1'024, 8'192, 1'048'576}) { + SCOPED_TRACE(budget); + RunExecutor executor(2, budget); + std::vector visits; + std::vector errors; + EXPECT_THAT( + Walk( + fs_, {"/r"}, {.post_order = post_order, .sort = sort, .workers = 3}, executor, + [&](const Visit& visit) { + visits.emplace_back(visit.path); + return WalkAction::kContinue; + }, + [&](std::string_view, absl::Status status) { errors.push_back(std::move(status)); }), + IsOk()); + EXPECT_THAT(visits, Eq(expected)); + EXPECT_THAT(errors, IsEmpty()); + const auto usage = executor.ReadAheadUsage(); + EXPECT_THAT(usage.peak_in_use, Le(4)); + EXPECT_THAT(usage.peak_retained_bytes, Le(budget)); + EXPECT_THAT(usage.in_use, Eq(0)); + EXPECT_THAT(usage.retained_bytes, Eq(0)); + } + } + } +} + +TEST_F(WalkFakeFsTest, CachedParentSiblingsDoNotStarveNestedReadAhead) { + CountingReadFs fs; + std::vector siblings; + std::vector nested; + siblings.reserve(65); + nested.reserve(65); + for (std::size_t index = 0; index < 65; ++index) { + const std::string name = absl::StrCat("d", index); + const std::string outer = absl::StrCat("/r/", name); + siblings.push_back(DirEntry(outer, name)); + fs.AddDir(outer, 1, {}); + const std::string inner = absl::StrCat("/r/d0/", name); + nested.push_back(DirEntry(inner, name)); + fs.AddDir(inner, 1, {}); + } + fs.AddDir("/r", 1, std::move(siblings)); + fs.AddDir("/r/d0", 1, std::move(nested)); + RunExecutor executor(1); + bool reached_nested = false; + EXPECT_THAT( + Walk( + fs, {"/r"}, {.sort = SortOrder::kTree, .workers = 2}, executor, + [&](const Visit& visit) { + if (visit.path == "/r/d0" || visit.path == "/r/d0/d0") { + // With one worker, this read-ahead FIFO marker proves that all + // previously queued directory jobs have published their results. + executor.Submit([] {}, RunTaskClass::kReadAhead).Get(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Gt(0)); + } + if (visit.path == "/r/d0/d0") { + reached_nested = true; + // Both nested slots are usable although the parent's next sibling + // is still cached. No nested listing has been consumed yet. + EXPECT_THAT(fs.nested_reads.load(), Eq(2)); + return WalkAction::kStop; + } + return WalkAction::kContinue; + }, + [](std::string_view, absl::Status) {}), + IsOk()); + EXPECT_THAT(reached_nested, IsTrue()); + EXPECT_THAT(fs.reads.load(), Eq(4)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); +} + +class FailedListingFs final : public FakeFs { + public: + absl::StatusOr> ReadDir(std::string_view dir) const override { + if (dir == "/r/d1") { + failed_reads.fetch_add(1, std::memory_order_relaxed); + absl::Status failure = absl::PermissionDeniedError("unreadable directory"); + failure.SetPayload("test-detail", absl::Cord(std::string(16'384, 'x'))); + return failure; + } + return FakeFs::ReadDir(dir); + } + + mutable std::atomic failed_reads = 0; +}; + +TEST_F(WalkFakeFsTest, OversizedSpeculativeErrorIsRereadAndReportedOnceWhenRequired) { + FailedListingFs fs; + std::vector siblings; + siblings.reserve(65); + 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, {}); + } + fs.AddDir("/r", 1, std::move(siblings)); + RunExecutor executor(1, 8'192); + std::vector errors; + EXPECT_THAT( + Walk( + fs, {"/r"}, {.sort = SortOrder::kTree, .workers = 2}, executor, + [&](const Visit& visit) { + if (visit.path == "/r/d0") { + // The error result is larger than the entire speculative allowance + // and finishes before the coordinator requires that directory. + executor.Submit([] {}, RunTaskClass::kReadAhead).Get(); + EXPECT_THAT(fs.failed_reads.load(), Eq(1)); + } + return WalkAction::kContinue; + }, + [&](std::string_view path, absl::Status status) { + EXPECT_THAT(path, Eq("/r/d1")); + errors.push_back(std::move(status)); + }), + IsOk()); + EXPECT_THAT(errors, ElementsAre(StatusIs(absl::StatusCode::kPermissionDenied, HasSubstr("unreadable")))); + EXPECT_THAT(fs.failed_reads.load(), Eq(2)); + EXPECT_THAT(executor.ReadAheadUsage().peak_retained_bytes, Le(8'192)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); +} + 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.