diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 1c9714707e8..dc9ff527579 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -24,7 +24,7 @@ repos: hooks: - id: check-added-large-files # Retain the exact lossless cross-platform performance evidence archives. - exclude: ^docs/benchmarks/(executor-copy-audit|bounded-read-ahead)-20261101\.json\.gz$ + exclude: ^docs/benchmarks/((executor-copy-audit|bounded-read-ahead)-20261101|directory-task-slots-20261102)\.json\.gz$ - id: check-merge-conflict - id: check-yaml # BINARY_TYPES below: archive containers are binary but can START with text (a phar begins with diff --git a/CHANGELOG.md b/CHANGELOG.md index aab3febbe73..de0d1b7f89a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,8 @@ # 0.11.0 +- Reuse typed directory-task records and publish completion without the shared queue lock. + - Reuse budgeted native comparison buffers across chunks while preserving legacy decoder reads. - Avoid redundant zeroing in native bounded range reads while preserving short-read, EOF and error handling. diff --git a/TODO.md b/TODO.md index f65a6d6a413..37527092ee9 100644 --- a/TODO.md +++ b/TODO.md @@ -392,8 +392,11 @@ EP/JC/P work; candidates are not speedup claims until measured on both platforms - [x] O07: Qualify cursor-owned native buffers, preserving budget lifetimes and legacy streams. [Native/fallback evidence](docs/benchmarks/cursor-owned-read-buffers-20261102.md) selects one initial clear plus reuse; reject uninitialized Linux regressions and the first shared fallback loop. -- [ ] O08-O12: Profile and qualify typed reusable directory task slots, batch ownership, - bounded result reuse, active-worker dispatch and direct production evaluation. +- [x] O08: Qualify typed reusable directory-task slots without changing worker/admission policy. + [Initial and refined evidence](docs/benchmarks/directory-task-slots-20261102.md) selects + atomic ownership handoff, rejecting global completion locking and retaining small regressions. +- [ ] O09-O12: Profile and qualify batch ownership, bounded result reuse, active-worker dispatch + and direct production evaluation. Preserve existing stage policy while isolating task storage and lifecycle costs. - [ ] O13-O22: Reconcile remaining derived-value handoff, static catalogs, invariant operands, paired streaming compare, reusable stage pools, byte-bounded overlap, archive indexing/admission, diff --git a/docs/benchmarks/directory-task-slots-20261102.json.gz b/docs/benchmarks/directory-task-slots-20261102.json.gz new file mode 100644 index 00000000000..85ef299ccce Binary files /dev/null and b/docs/benchmarks/directory-task-slots-20261102.json.gz differ diff --git a/docs/benchmarks/directory-task-slots-20261102.md b/docs/benchmarks/directory-task-slots-20261102.md new file mode 100644 index 00000000000..a8748608ff3 --- /dev/null +++ b/docs/benchmarks/directory-task-slots-20261102.md @@ -0,0 +1,92 @@ +# Typed reusable directory jobs (2026-11-02) + +## Decision + +Retain typed reusable task records and per-task atomic completion on the existing directory worker pool. Replace per-job shared packaged tasks, futures and erased queue callbacks with move-only typed requests to stable slots. No new executor, worker policy, clock gate, CLI flag or dependency is added. The former futures already used persistent workers; they did not create a thread per file. + +The initial typed implementation completed each result under the global queue mutex. That implementation is rejected despite lower allocation counts: Linux directory-heavy controls regressed about 11-14% with some worker/task combinations. The refined implementation publishes each result through a three-state atomic handoff and uses the queue mutex only for admission, queue operations and recycling. Standard atomic wait supplies blocking for the sole consumer. + +Keep the original two-directory admission rule, 128-entry stat chunks, 512-entry eager-stat gate, source/archive restrictions, four-listings-per-worker frontier and requested worker counts. This does not reintroduce the rejected compact frontier ring, narrower window, remaining-work planner or timed admission policy. Serial/file-only scans allocate no queue storage. + +## Ownership and synchronization + +A request exclusively owns the consumer lease for one deque-stable task record. Moves disengage the old lease. Required storage/pool observers use references; optional task observers use OptionalRef. The producer alone reads inputs and fills its result. A release/acquire atomic publication hands that payload to its consumer. The second participant discards an abandoned result: either abandonment sees Ready, or completion sees Abandoned. Exactly one participant recycles the slot. + +A late notification after recycling is harmless: atomic wait rechecks the current value, and a slot cannot be reused while its old consumer still waits. Slot addresses remain valid until all pool workers join. The queue drains already submitted jobs when walking stops, preserving existing speculative-read behavior; abandoned errors are not published as visited errors. Move discarded payloads into a local value so their allocation cleanup runs after the queue mutex unlocks. + +Directory paths remain owned by their tasks. Stat slices borrow their listing only until all stat results are consumed. Jobs remain leaves: no worker submits work to its own pool or waits for a sibling. Pool destruction joins workers before the filesystem owner or walker state dies. No shared ownership is introduced for task records. + +Slot metadata retains its high-water capacity until that walker ends, while strings/result payloads are released promptly. The frontier bound is per active recursion level and counts requests; it is not a global byte bound, and abandoned queued jobs still drain. Do not claim constant retained memory for arbitrary tree depth or huge listings. + +## Measurement contract + +Both phases use exact parent `d2ff56bc444be383de44b3e3b5dd349700abac87`, including O01-O07 runtime changes. After #1047 merged, the publication branch was synchronized onto main; the two base trees are identical. Preserve the original measurement identity rather than relabel old data. Release binaries use LLVM 23.1.4 + ThinLTO; all sources, source/executable hashes, build flags, raw samples and scripts are in the archive below. + +Initial variants are baseline, lazily allocated old queue, and typed slots with global-lock completion. Refined variants are fresh baseline, the global-lock candidate and atomic completion. The lazy-only control isolates constructor allocation from repeated task handoff. Do not pool absolute times between phases. + +The 80 native cases cover ten files; flat 1,000/10,000-file directories; 8/1,024 one-file directories; 8/128 directories of 512 files; an 8-by-8 nested tree of 64 files; long collection names; and small/large two-root comparison controls. Files contain `x` plus newline. Requested 1/3/10 workers cover listing, size-formatting and literal-content tasks, with the established largest-content and collection selections preserved. + +Each case has three sessions, a discarded warmup per variant/session, nine interleaved repetitions and the fastest seven retained. Every native invocation validates an independent path/size digest or complete collection/comparison records. Both phases together contain 28,800 fresh invocations across the hosts, including warmups. Linux uses tmpfs with affinity restricted to requested CPUs; macOS uses warm APFS and worker allowances without affinity. Builds, sanitizers and lint finish before timing on each host. + +Separate in-memory RE2 engine controls retain the established 36 workloads, 1/4 requested workers, nine repetitions/fastest seven and two sessions, without CPU affinity. Two Linux rg-files cells vary sharply between those sessions (+26.47/-3.25% and +20.41/-2.14%); retain them unchanged. Supplementary single-case processes interleave all variants across three sessions and eight cases. Linux supplements use explicit 1/4-CPU affinity, a distinct contract; macOS remains unpinned. Neither kernel population has the native driver's independent per-invocation output oracle. + +## Refined native results + +Changes are medians of paired session ratios; negative means faster. These are selected representative cases, with every case retained in the archive. + +| Fixture / task | Workers | Linux change | macOS change | +| :---------------------- | ------: | -----------: | -----------: | +| broad1024x1 / files | 3 | -48.27% | -1.93% | +| broad1024x1 / files | 10 | -8.03% | +0.29% | +| broad1024x1 / metadata | 3 | -23.30% | -0.28% | +| broad1024x1 / metadata | 10 | -11.76% | +0.17% | +| broad1024x1 / content | 10 | -5.69% | +0.48% | +| broad128x512 / files | 3 | -4.09% | +0.32% | +| broad128x512 / metadata | 3 | -0.39% | -0.64% | +| nested8x8x64 / files | 3 | -0.26% | +0.11% | +| flat10000 / metadata | 3 | +1.12% | -0.23% | +| flat10000 / metadata | 10 | +1.40% | +0.48% | +| flat10 / files | 1 | -0.33% | +0.52% | + +The large Linux many-directory gains are not universal scan speedups. macOS is mostly near baseline and its syscall/storage cost differs. The global-lock Linux control again regresses 10.50% for many-directory ten-worker listing and 13.63% for its three-worker metadata case; atomic completion changes those to -8.03% and -23.30%. This establishes the synchronization refinement, rather than attributing all elapsed changes to allocation counts. + +## Allocation traffic and footprint + +ASan diagnostics never provide performance timings. Hooks accept only the initializing coordinator; worker callbacks are rejected before accessing global POD counters. This avoids the macOS lazy-TLS bootstrap recursion exposed by the first diagnostic that started actual worker threads. A committed worker-start regression covers it. Fixture setup, warmup and assertions occur outside measurement. Cross-thread frees are excluded, so allocation/free differences are not live-byte or leak accounting. + +| Host | 1,024-directory coordinator allocations: old | Atomic slots | Requested bytes: old | Atomic slots | +| :---- | -------------------------------------------: | -----------: | -------------------: | -----------: | +| macos | 4,145 | 1,056 | 617,136 | 332,656 | +| linux | 4,137 | 1,056 | 625,104 | 332,656 | + +The lazy-only queue control does not remove per-job allocation traffic; both typed mechanisms reduce it about 75% in this diagnostic. Counts include unavoidable coordinator traversal work and exclude worker result allocations. This is not an all-process allocation claim or a peak-RSS reduction of the same percentage. + +- macos: worst native paired median +1.16% at ('broad8x1', 10, 'files'); all small cases and slower samples remain. +- macos: build-output executable 4,070,368 -> 4,051,984 bytes (-18,384), before release stripping/compression. +- linux: worst native paired median +3.40% at ('broad8x1', 10, 'metadata'); all small cases and slower samples remain. +- linux: build-output executable 12,309,616 -> 12,295,424 bytes (-14,192), before release stripping/compression. + +## Supplementary in-memory controls + +The focused Linux rg-files medians are +2.79% (1,000/four workers) and +1.40% (10,000/four). Those remain small costs, not erased regressions. No claim is made that every kernel improves. + +| Kernel control | Linux change | macOS change | +| :---------------------- | -----------: | -----------: | +| RE2/filter/10/1 | +1.03% | -0.17% | +| RE2/native-lines/1000/1 | -0.09% | -2.22% | +| RE2/rg-files/10/4 | +2.44% | -0.91% | +| RE2/rg-files/1000/4 | +2.79% | +0.29% | +| RE2/rg-files/10000/4 | +1.40% | -0.06% | +| RE2/traversal/10/4 | +0.65% | +0.02% | +| RE2/traversal/1000/4 | +1.79% | +1.96% | +| RE2/traversal/10000/4 | +1.05% | +0.38% | + +## Validation and limits + +Both full core/extension release suites pass 305/305 for both implementations; Linux requires working FUSE integration. Focused macOS ASan, Linux TSan and the full allocation diagnostics pass. Scoped local clang-tidy passes all three modified translation units. Tests cover mixed stat/directory slot reuse above the actual 512-entry stat gate, partial chunk tails, repeated windows, moved/default requests, coordinator visit order, ordered read errors, early-stop drain bounds and abandoned-error suppression. Existing archive/owner, metadata-demand, symlink, prune/depth and readdir-race tests remain in the full suites. + +These measurements qualify cached local storage and in-memory engine controls, not new network-storage concurrency defaults or archive parallel admission. There is no default worker-count change. Further cache-line layout, batching and shared-executor experiments need their own evidence; do not fold their unmeasured hypotheses into this result. Normal commit/push hooks and current-head CI are required before merge. + +## Lossless evidence + +[directory-task-slots-20261102.json.gz](directory-task-slots-20261102.json.gz): 1,418,765 bytes; SHA-256 `b3810d7d8e2553f11416c4307886a6959e1b4d910777c921a51f35ccdc1766af`. Restore a native report's fixtures from `fixtures[report.fixtures_ref]`. All initial, refined, targeted, rejected/lazy controls and source identities remain available. diff --git a/docs/performance-analysis.md b/docs/performance-analysis.md index 523483a44af..6e3fbd5f0c3 100644 --- a/docs/performance-analysis.md +++ b/docs/performance-analysis.md @@ -3207,3 +3207,36 @@ comparisons drop from 43 to 13 allocations, with total requested bytes about 2.1 this is allocation traffic, not peak RSS. Both fully enabled 305-test suites and focused sanitizers pass. Initial and refined raw source/binary identities, all 10,800 oracle-checked invocations, memory fallback kernels and rejected variants are retained in the report artifacts. + +## O08: typed task leases and atomic completion + +The [directory-task report](benchmarks/directory-task-slots-20261102.md) replaces the existing +pool's shared packaged tasks, futures and erased callbacks with reusable stable records and +move-only consumer leases. This does not add worker threads or change the current admission, +stat chunks, source restrictions, frontier width or ordering. Futures previously ran on the +same persistent workers; do not describe this as removing one thread per file. + +The first typed candidate completed under the queue mutex and regressed important Linux +controls by roughly 11-14%. Keep that implementation and its full matrix as a rejected control. +The selected implementation uses a release/acquire Pending/Ready/Abandoned handshake for +producer/consumer ownership and standard atomic wait. Only the second participant releases an +abandoned result; slot addresses remain stable until all workers join. Queue admission and +recycling keep the existing mutex, and discarded payload values die outside that lock. + +In refined native measurements, Linux's 1,024-directory listing improves 48.27% with three +workers and 8.03% with ten; its three-worker metadata case improves 23.30%. macOS is mostly +near baseline. Worst native paired medians remain +1.16% macOS / +3.40% Linux. Two initially +noisy unpinned Linux rg-files kernels receive separate paired, CPU-pinned confirmation; the +1,000/10,000-file four-worker medians are +2.79% / +1.40%, and those costs are retained. + +Coordinator-only ASan counts fall about 75% at 1,024 directories (4,145 to 1,056 macOS; +4,137 to 1,056 Linux). This excludes worker allocations/frees and is neither release timing +nor net live-memory accounting. The new actual-worker diagnostic exposed a macOS TLS-bootstrap +recursion in the allocator hook; reject foreign-thread callbacks before touching coordinator-owned +POD state, and retain a worker-start regression. No instrumented timings enter the decision. + +Both full all-extension suites pass 305/305, with FUSE required on Linux; focused ASan/TSan and +scoped lint pass. The two three-way native phases preserve all 28,800 invocations, exact output +oracles, fixture definitions, source/binary identities and slower cases. Slot metadata retains +its high-water capacity until walker destruction; the per-level request frontier is not a global +byte bound. Cache-line layout and wider scheduling changes remain separate, unmeasured follow-ups. diff --git a/release-site.json b/release-site.json index 5d584777d51..e6f5cf5421f 100644 --- a/release-site.json +++ b/release-site.json @@ -65,7 +65,8 @@ "docs/benchmarks/lexical-field-views-20261102.md": "docs/benchmarks/lexical-field-views-20261102.html", "docs/benchmarks/field-lookup-controls-20261102.md": "docs/benchmarks/field-lookup-controls-20261102.html", "docs/benchmarks/host-range-overwrite-20261102.md": "docs/benchmarks/host-range-overwrite-20261102.html", - "docs/benchmarks/cursor-owned-read-buffers-20261102.md": "docs/benchmarks/cursor-owned-read-buffers-20261102.html" + "docs/benchmarks/cursor-owned-read-buffers-20261102.md": "docs/benchmarks/cursor-owned-read-buffers-20261102.html", + "docs/benchmarks/directory-task-slots-20261102.md": "docs/benchmarks/directory-task-slots-20261102.html" }, "generated_html": { "XFF.md": { diff --git a/xff/engine/expression_allocations_test.cc b/xff/engine/expression_allocations_test.cc index 06a2e58f80c..a128b48faba 100644 --- a/xff/engine/expression_allocations_test.cc +++ b/xff/engine/expression_allocations_test.cc @@ -8,6 +8,7 @@ #include #include #include +#include #include #include #include @@ -39,6 +40,7 @@ namespace xff::engine { namespace { +using ::mbo::testing::IsOk; using ::mbo::testing::IsOkAndHolds; using ::testing::Eq; using ::testing::Field; @@ -60,16 +62,25 @@ struct AllocationRecording { AllocationCounts counts; }; -// Only this diagnostic executable installs hooks. Thread-local POD counters neither allocate -// nor lock, and ignore allocations made by the test runner outside a measured operation. +// Only this diagnostic executable installs hooks. One coordinator owns these POD counters; +// worker callbacks are rejected before accessing state, avoiding lazy macOS TLS allocation +// recursively entering an allocation hook. No recording path allocates or locks. AllocationRecording& Recording() { - static constinit thread_local AllocationRecording recording; + static constinit AllocationRecording recording; return recording; } #if defined(ABSL_HAVE_ADDRESS_SANITIZER) +std::thread::id RecordingThread() { + static const auto kThread = std::this_thread::get_id(); + return kThread; +} + // XFF_ABI_POINTER: sanitizer allocator-hook callback signature. void RecordAllocation(const volatile void*, std::size_t bytes) { + if (std::this_thread::get_id() != RecordingThread()) { + return; + } auto& recording = Recording(); if (recording.enabled) { ++recording.counts.allocations; @@ -79,6 +90,9 @@ void RecordAllocation(const volatile void*, std::size_t bytes) { // XFF_ABI_POINTER: sanitizer allocator-hook callback signature. void RecordFree(const volatile void*) { + if (std::this_thread::get_id() != RecordingThread()) { + return; + } auto& recording = Recording(); if (recording.enabled) { ++recording.counts.frees; @@ -88,7 +102,11 @@ void RecordFree(const volatile void*) { bool InstallHooks() { #if defined(ABSL_HAVE_ADDRESS_SANITIZER) - static const int kInstalled = __sanitizer_install_malloc_and_free_hooks(RecordAllocation, RecordFree); + static const int kInstalled = [] { + // Initialize the immutable owner identity before registration can invoke either callback. + (void)RecordingThread(); + return __sanitizer_install_malloc_and_free_hooks(RecordAllocation, RecordFree); + }(); return kInstalled > 0; #else return false; @@ -137,7 +155,7 @@ auto MeasureAllocations(Action&& action) { return result; } -class AllocationFs final : public vfs::FileSystem { +class AllocationFs : public vfs::FileSystem { public: absl::StatusOr> ReadDir(std::string_view) const override { return absl::FailedPreconditionError("allocation fixture must not traverse"); @@ -166,6 +184,33 @@ class AllocationFs final : public vfs::FileSystem { absl::StatusOr IsCaseSensitive(std::string_view) const override { return true; } }; +class DirectoryAllocationFs final : public AllocationFs { + public: + explicit DirectoryAllocationFs(std::size_t directories) : directories_(directories) {} + + absl::StatusOr> ReadDir(std::string_view path) const override { + std::vector entries; + if (path == "root") { + entries.reserve(directories_); + for (std::size_t index = 0; index < directories_; ++index) { + const auto name = "d" + std::to_string(index); + entries.push_back({.path = "root/" + name, .name = name, .type = vfs::FileType::kDirectory}); + } + } + return entries; + } + + absl::StatusOr Stat(std::string_view path, bool) const override { + return vfs::Metadata{ + .type = vfs::FileType::kDirectory, + .ino = path == "root" ? 1UZ : 2UZ, + }; + } + + private: + std::size_t directories_; +}; + struct AllocationTest : ::testing::Test { void SetUp() override { const bool installed = InstallHooks(); @@ -198,6 +243,21 @@ TEST_F(AllocationTest, CounterObservesAllocationAndFreeWithoutCountingAssertions EXPECT_THAT(released.counts.frees, Ge(1)); } +TEST_F(AllocationTest, WorkerAllocationsDoNotInitializeOrAccessRecorderState) { + const auto measured = MeasureAllocations([] { + std::size_t size = 0; + std::thread worker([&] { + const std::vector bytes(4'096, 'x'); + size = bytes.size(); + }); + worker.join(); + return size; + }); + EXPECT_THAT(measured.value, Eq(4'096)); + // Thread creation itself may allocate on the coordinator; only the worker payload is excluded. + RecordCounts("coordinator_thread_start", measured.counts); +} + TEST_F(AllocationTest, NativeComparisonAllocationsDoNotGrowWithChunkCount) { const vfs::LocalFs fs; const auto measure = @@ -234,6 +294,39 @@ TEST_F(AllocationTest, NativeComparisonAllocationsDoNotGrowWithChunkCount) { RecordCounts("sixteen_chunks", long_read.counts); } +TEST_F(AllocationTest, RecordCoordinatorDirectoryTaskAllocationTraffic) { + struct Outcome { + absl::Status status; + std::size_t visits = 0; + std::size_t errors = 0; + }; + + static constexpr auto kDirectoryCounts = std::to_array({16, 64, 256, 1'024}); + const auto roots = std::to_array({"root"}); + for (const auto directories : kDirectoryCounts) { + const DirectoryAllocationFs fs(directories); + const auto run = [&] { + Outcome outcome; + outcome.status = Walk( + fs, roots, {.workers = 3, .metadata_fields = vfs::MetadataFields::kBasic}, + [&](const Visit&) { + ++outcome.visits; + return WalkAction::kContinue; + }, + [&](std::string_view, absl::Status) { ++outcome.errors; }); + return outcome; + }; + EXPECT_THAT(run().status, IsOk()); + const auto measured = MeasureAllocations(run); + EXPECT_THAT(measured.value.status, IsOk()); + EXPECT_THAT(measured.value.visits, Eq(directories + 1)); + EXPECT_THAT(measured.value.errors, Eq(0)); + RecordCounts(absl::StrCat(directories, "_directories"), measured.counts); + } + // Hooks accept only the coordinator: worker-side listing/stat allocations are excluded. + // This diagnostic reports coordinator traffic; sanitizer timing is never performance evidence. +} + TEST_F(AllocationTest, ReusedFuzzyWorkspaceAvoidsHotScoreAllocations) { const fuzzy::FzfQuery query("fl"); fuzzy::ScoreWorkspace workspace; diff --git a/xff/engine/walk.cc b/xff/engine/walk.cc index 318dd0c7fc9..1f7377b13d9 100644 --- a/xff/engine/walk.cc +++ b/xff/engine/walk.cc @@ -16,11 +16,11 @@ #include "xff/engine/walk.h" #include +#include #include #include #include #include -#include #include #include #include @@ -73,85 +73,258 @@ struct Stated { // The result of reading one directory: its children, or a ReadDir error. using Listing = absl::StatusOr>; -// 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) {} +class Walker { + // Typed leaf jobs share the existing persistent workers. A request owns a slot until its + // result is consumed, or marks it abandoned so the worker can release it after completion. + class ReadPool { + public: + enum class Kind { kDirectory, kStat }; + + private: + enum class TaskState { kPending, kReady, kAbandoned }; + + struct Storage; + + struct Task { + Task(Storage& storage, std::size_t index) : storage(storage), index(index) {} + + Storage& storage; + std::size_t index = 0; + Kind kind = Kind::kDirectory; + std::string directory; + absl::Span entries; + Listing listing; + std::vector stated; + std::atomic state{TaskState::kPending}; + }; + + struct Storage { + explicit Storage(ReadPool& pool) : pool(pool) {} + + ReadPool& pool; + std::deque tasks; + std::queue pending; + std::vector free; + }; + + public: + template + class Request { + public: + using Result = std::conditional_t>; + Request() = default; + + explicit Request(Task& task) : task_(task) {} + + ~Request() { Reset(); } + + Request(const Request&) = delete; + Request& operator=(const Request&) = delete; + + Request(Request&& other) noexcept : task_(other.task_) { other.task_.reset(); } + + Request& operator=(Request&& other) noexcept { + if (this != &other) { + Reset(); + if (other.task_) { + task_.set_ref(*other.task_); + } + other.task_.reset(); + } + return *this; + } + + bool Valid() const { return task_.has_value(); } + + Result Take() { + auto result = task_->storage.pool.template Take(*task_); + task_.reset(); + return result; + } + + private: + void Reset() { + if (task_) { + task_->storage.pool.Abandon(*task_); + task_.reset(); + } + } + + mbo::types::OptionalRef task_; + }; + + ReadPool(Walker& walker, std::size_t workers) : walker_(walker), 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 ReadPool&) = delete; + ReadPool& operator=(const ReadPool&) = delete; + ReadPool(ReadPool&&) = delete; + ReadPool& operator=(ReadPool&&) = delete; + + 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 + ~ReadPool() { + { + const absl::MutexLock lock(mutex_); + stop_ = true; + } + for (auto& thread : threads_) { + thread.join(); + } } - for (std::thread& thread : threads_) { - thread.join(); + + Request SubmitDirectory(std::string_view directory) { + return Request(Submit(Kind::kDirectory, directory, {})); } - } - 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; - } + Request SubmitStats(absl::Span entries) { + return Request(Submit(Kind::kStat, {}, entries)); + } - 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; + private: + Task& Submit(Kind kind, std::string_view directory, absl::Span entries) { + Task& task = [&] -> Task& { + const absl::MutexLock lock(mutex_); + if (!storage_) { + storage_.emplace(*this); } - job = std::move(queue_.front()); - queue_.pop(); + std::size_t index = 0; + if (storage_->free.empty()) { + index = storage_->tasks.size(); + storage_->tasks.emplace_back(*storage_, index); + } else { + index = storage_->free.back(); + storage_->free.pop_back(); + } + auto& selected = storage_->tasks.at(index); + selected.kind = kind; + selected.directory = directory; + selected.entries = entries; + selected.state.store(TaskState::kPending, std::memory_order_relaxed); + if (!threads_.empty()) { + storage_->pending.push(index); + } + return selected; + }(); + if (threads_.empty()) { + Execute(task); + } + return task; + } + + // NOLINTNEXTLINE(readability-convert-member-functions-to-static): lock contract uses this pool's mutex. + void Recycle(Task& task) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) { + std::string{}.swap(task.directory); + task.entries = {}; + task.listing = absl::UnknownError(""); + std::vector{}.swap(task.stated); + task.storage.free.push_back(task.index); + } + + template + Request::Result Take(Task& task) { + // Exactly one consumer owns this lease. Ready publishes the completed payload; + // only its consumer can take it, so an abandoned state cannot reach this wait. + task.state.wait(TaskState::kPending, std::memory_order_acquire); + auto result = [&] { + if constexpr (kKind == Kind::kDirectory) { + return std::move(task.listing); + } else { + return std::move(task.stated); + } + }(); + { + const absl::MutexLock lock(mutex_); + Recycle(task); } - job(); + return result; + } + + void ReleaseCompleted(Task& task) { + struct Payload { + std::string directory; + Listing listing; + std::vector stated; + }; + + // Declare the moved payload before the lock so its allocations die after unlocking. + const Payload released{ + .directory = std::move(task.directory), + .listing = std::move(task.listing), + .stated = std::move(task.stated), + }; + const absl::MutexLock lock(mutex_); + Recycle(task); } - } - 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 -}; + void Abandon(Task& task) { + // The second participant releases an unused payload: either this consumer observes + // Ready, or the producer observes Abandoned when publishing its result. + if (task.state.exchange(TaskState::kAbandoned, std::memory_order_acq_rel) == TaskState::kReady) { + ReleaseCompleted(task); + } + } + + template + void Complete(Task& task, Result result) { + if constexpr (std::is_same_v) { + task.listing = std::move(result); + } else { + task.stated = std::move(result); + } + if (task.state.exchange(TaskState::kReady, std::memory_order_acq_rel) == TaskState::kAbandoned) { + ReleaseCompleted(task); + } else { + // Storage remains stable until every worker joins. A late notification after + // recycling is harmless: wait rechecks the new generation's atomic value. + task.state.notify_one(); + } + } + + void Execute(Task& task) { + if (task.kind == Kind::kDirectory) { + Complete(task, walker_.ReadDir(task.directory)); + } else { + Complete(task, walker_.StatEntries(task.entries)); + } + } + + bool Pending() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) { + return stop_ || (storage_ && !storage_->pending.empty()); + } + + mbo::types::OptionalRef Next() { + const absl::MutexLock lock(mutex_, absl::Condition(this, &ReadPool::Pending)); + if (!storage_ || storage_->pending.empty()) { + return std::nullopt; + } + const auto index = storage_->pending.front(); + storage_->pending.pop(); + return storage_->tasks.at(index); + } + + void Run() { + while (auto task = Next()) { + Execute(*task); + } + } + + Walker& walker_; + const std::size_t max_workers_; + mutable absl::Mutex mutex_; + std::optional storage_ ABSL_GUARDED_BY(mutex_); + bool stop_ ABSL_GUARDED_BY(mutex_) = false; + std::vector threads_; + }; + + using ReadRequest = ReadPool::Request; + using StatRequest = ReadPool::Request; -class Walker { public: Walker( const vfs::FileSystem& fs, @@ -165,7 +338,7 @@ class Walker { on_error_(on_error), mount_container_(mount_container), follow_children_(options.symlinks == SymlinkMode::kAll), - pool_(options.workers > 1 ? options.workers : std::size_t{0}) {} + pool_(*this, options.workers > 1 ? options.workers : std::size_t{0}) {} // 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. @@ -343,9 +516,9 @@ class Walker { 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 pool_.SubmitStats(chunk); }; - std::vector>> reads; + std::vector reads; reads.reserve(pending); for (std::size_t index = 0; index < pending; ++index) { reads.push_back(submit(index)); @@ -354,7 +527,7 @@ class Walker { 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); + children.append_range(read.Take() | std::views::as_rvalue); if (index + pending < chunks) { read = submit(index + pending); } @@ -362,9 +535,7 @@ class Walker { return children; } - std::future SubmitRead(const std::string& dir) { - return pool_.Submit([this, dir] { return ReadDir(dir); }); - } + ReadRequest SubmitRead(const std::string& dir) { return pool_.SubmitDirectory(dir); } // Bound speculative listings without changing the existing leaf-worker pool. // Each active recursion level owns at most four requests per available worker; @@ -398,7 +569,7 @@ class Walker { ReadFrontier(ReadFrontier&&) = delete; ReadFrontier& operator=(ReadFrontier&&) = delete; - std::future Take(std::size_t index) { + ReadRequest Take(std::size_t index) { if (!requests_) { return {}; } @@ -428,7 +599,7 @@ class Walker { private: struct Request { std::size_t index; - std::future read; + ReadRequest read; }; bool Eligible(std::size_t index) const { @@ -452,7 +623,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 request completes. 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) { @@ -507,7 +678,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; } @@ -547,15 +718,15 @@ class Walker { } } - // Reads `dir` (from its prefetched future, or now) and recurses its children, + // Reads `dir` (from its prefetched request, 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->Take() : 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()))) { @@ -587,7 +758,7 @@ class Walker { auto read = reads.Take(i); VisitSubtree( children.at(i), depth, - read.valid() ? mbo::types::OptionalRef{read} : mbo::types::OptionalRef>{}); + read.Valid() ? mbo::types::OptionalRef{read} : mbo::types::OptionalRef{}); } return; } @@ -612,7 +783,7 @@ class Walker { auto read = reads.Take(i); VisitSubtree( children.at(i), depth, - read.valid() ? mbo::types::OptionalRef{read} : 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. @@ -644,8 +815,8 @@ class Walker { auto read = reads.Take(i); Descend( children.at(i), depth, - read.valid() ? mbo::types::OptionalRef{read} - : 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. diff --git a/xff/engine/walk_parallel_test.cc b/xff/engine/walk_parallel_test.cc index 0dc85f5b57f..08aeb179938 100644 --- a/xff/engine/walk_parallel_test.cc +++ b/xff/engine/walk_parallel_test.cc @@ -41,7 +41,7 @@ struct StatTrace { bool await_peer = false; }; -struct WideFs final : vfs::FileSystem { +struct WideFs : vfs::FileSystem { explicit WideFs(std::shared_ptr trace = std::make_shared()) : trace(std::move(trace)) {} std::shared_ptr trace; @@ -101,13 +101,49 @@ struct WideFs final : vfs::FileSystem { absl::StatusOr IsCaseSensitive(std::string_view) const override { return true; } }; +struct DirectoryFs final : WideFs { + DirectoryFs() { count = 4; } + + std::size_t directories = 513; + std::set unreadable; + mutable std::map reads; + + absl::StatusOr> ReadDir(std::string_view path) const override { + { + const std::scoped_lock lock(trace->mutex); + ++reads[std::string(path)]; + } + if (unreadable.contains(std::string(path))) { + return absl::PermissionDeniedError(path); + } + if (path != "root") { + return WideFs::ReadDir(path); + } + std::vector entries; + entries.reserve(directories); + for (std::size_t index = 0; index < directories; ++index) { + const auto name = "d" + std::to_string(index); + entries.push_back({.path = "root/" + name, .name = name, .type = vfs::FileType::kDirectory}); + } + return entries; + } + + absl::StatusOr StatFields(std::string_view path, bool follow, vfs::MetadataFields fields) + const override { + if (path != "root" && path.find('/', 5) == std::string_view::npos) { + return vfs::Metadata{.type = vfs::FileType::kDirectory, .ino = 2}; + } + return WideFs::StatFields(path, follow, fields); + } +}; + struct WalkLog { std::vector> visits; std::vector> errors; std::set visitors; }; -WalkLog Record(const WideFs& fs, const WalkOptions& options) { +WalkLog Record(const WideFs& fs, const WalkOptions& options, std::size_t stop_after = 0) { WalkLog log; const auto roots = std::to_array({"root"}); EXPECT_THAT( @@ -116,7 +152,7 @@ WalkLog Record(const WideFs& fs, const WalkOptions& options) { [&](const Visit& visit) { log.visits.emplace_back(visit.path, visit.metadata.size); log.visitors.insert(std::this_thread::get_id()); - return WalkAction::kContinue; + return stop_after != 0 && log.visits.size() >= stop_after ? WalkAction::kStop : WalkAction::kContinue; }, [&](std::string_view path, const absl::Status& status) { log.errors.emplace_back(path, status.code()); }), IsOk()); @@ -200,5 +236,58 @@ TEST_F(WalkParallelTest, OwnedMountRemainsSerialWithoutArchiveSourceHints) { EXPECT_THAT(trace->threads, ElementsAre(std::this_thread::get_id())); } +TEST_F(WalkParallelTest, ReusedDirectoryAndStatRequestsKeepExactResultsAndReadCounts) { + static constexpr auto kOrders = std::to_array({ + SortOrder::kNone, + SortOrder::kDir, + SortOrder::kTree, + SortOrder::kGlobal, + SortOrder::kSubtree, + }); + for (const auto order : kOrders) { + SCOPED_TRACE(static_cast(order)); + const DirectoryFs serial; + const DirectoryFs parallel; + const auto expected = Record(serial, {.sort = order, .workers = 1}); + const auto actual = Record(parallel, {.sort = order, .workers = 3}); + EXPECT_THAT(actual.visits, Eq(expected.visits)); + EXPECT_THAT(actual.errors, IsEmpty()); + EXPECT_THAT(actual.visitors, ElementsAre(std::this_thread::get_id())); + EXPECT_THAT(parallel.reads, SizeIs(parallel.directories + 1)); + EXPECT_THAT(parallel.trace->children, Eq(parallel.directories * parallel.count)); + for (const auto& [path, count] : parallel.reads) { + SCOPED_TRACE(path); + EXPECT_THAT(count, Eq(1)); + } + } +} + +TEST_F(WalkParallelTest, ReusedRequestsPreserveDirectoryErrorPublicationOrder) { + DirectoryFs serial; + DirectoryFs parallel; + serial.unreadable = parallel.unreadable = {"root/d100", "root/d17", "root/d3"}; + const auto expected = Record(serial, {.sort = SortOrder::kTree, .workers = 1}); + const auto actual = Record(parallel, {.sort = SortOrder::kTree, .workers = 4}); + EXPECT_THAT(actual.visits, Eq(expected.visits)); + EXPECT_THAT(actual.errors, Eq(expected.errors)); + EXPECT_THAT(actual.errors, SizeIs(3)); + EXPECT_THAT(parallel.reads, SizeIs(parallel.directories + 1)); +} + +TEST_F(WalkParallelTest, AbandonedRequestsDrainWithoutPublishingUnvisitedErrors) { + DirectoryFs serial; + DirectoryFs parallel; + serial.unreadable = parallel.unreadable = {"root/d100"}; + // Under directory-first order, all root directories are visited before the first leaf. + // Stop within the second child's leaves, leaving prefetched siblings without consumers. + const std::size_t stop_after = 1 + parallel.directories + parallel.count + 1; + const auto expected = Record(serial, {.sort = SortOrder::kDir, .workers = 1}, stop_after); + const auto actual = Record(parallel, {.sort = SortOrder::kDir, .workers = 3}, stop_after); + EXPECT_THAT(actual.visits, Eq(expected.visits)); + EXPECT_THAT(actual.errors, IsEmpty()); + // Walk joins the workers before returning, including already queued abandoned reads. + EXPECT_THAT(parallel.reads.at("root/d100"), Eq(1)); +} + } // namespace } // namespace xff::engine