diff --git a/XFF.md b/XFF.md index 9ba8849e7a5..c258f2b6288 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 lazily started pool of up to `N` worker threads, reused for directory reads, required eager metadata and eligible content matching. Workers drain batches of independent work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core. + `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. See also: [Output](#topic-output), [Statistics](#topic-stats) diff --git a/docs/benchmark-correctness-audit.md b/docs/benchmark-correctness-audit.md index 60c45fd76bc..176a99120d3 100644 --- a/docs/benchmark-correctness-audit.md +++ b/docs/benchmark-correctness-audit.md @@ -275,20 +275,37 @@ available local Linux artifacts; they do not verify expired hosted artifacts. ### Worker scheduling experiment Linux `kernel.perf_event_paranoid` is now `0`, and native `perf` hardware counters -are available. Startup-boundary profiles identify worker creation and teardown, -Abseil mutex slow paths and scheduling overhead. Keeping directory reads serial +are available. Startup-boundary profiles include worker creation and teardown, +Abseil mutex slow paths and scheduling overhead. The later symbolized controls +retained in [PR #995](https://github.com/mboworks/xff/pull/995) identify the +Linux/x86 one-time Abseil TSC frequency calibration (1 ms and 2 ms sleeps) on +the first queue mutex path as the dominant fixed cliff. They do not attribute +those sleeps to each thread's creation. Keeping directory reads serial removes the small-work cliff but loses useful concurrency on large broad trees. [PR #991](https://github.com/mboworks/xff/pull/991) admits directory reads only when at least 64 independent sibling directories are available. Its hosted Linux and macOS tests, sanitizers, all three benchmark shards on each platform and combined benchmark comparison pass. -The next experiment shares one run-owned executor across directory read-ahead, +The initial shared-executor experiment shares one traversal-owned executor across directory read-ahead, eager stat work and eligible content matching. Each admitted batch submits at most one drain job per useful worker. Directory jobs repeatedly claim independent listing reads; stat jobs claim 128-entry chunks; matcher jobs claim entry chunks. Only the coordinator recurses, evaluates traversal controls and consumes ordered -results. Threads remain bounded and are reused until command completion. +results. Its pools are traversal-owned, so two comparison-side coordinators can +still create separate executors before a later comparison-only pool starts. + +A separate command-ownership follow-up removes that gap: `RunFind` owns one +executor, each comparison side walks on the calling coordinator, and eligible +file pairs reuse the same executor after both inventories are collected. The +`--jobs=N` participant budget includes the caller, leaving at most `N-1` +background workers. The caller also claims independent stat, content and +comparison chunks. This is a concurrency experiment, not a performance +acceptance claim. Directory drain fairness, bounded read-ahead, cancellation, +remaining-work admission, slow-storage controls and native M5 Pro validation +remain outstanding. The fresh native accounting and exact-source baseline +evidence live separately in PRs #994 and #995; this branch does not mix their +measurement artifacts into a new engine result. Correctness checks include nested read-ahead above the admission threshold, partial stat chunks with missing entries, repeated matcher transitions and diff --git a/docs/design-parallel.md b/docs/design-parallel.md index bbddeee63a8..b262228f4f4 100644 --- a/docs/design-parallel.md +++ b/docs/design-parallel.md @@ -7,10 +7,11 @@ ## Purpose -The directory walk is the hot path and it is IO-bound: most time is spent -blocked in `readdir`/`lstat`, not in CPU. Running it across several workers -hides that latency and is the single biggest performance lever for xff, -especially on deep trees and high-latency (network) filesystems. +Directory enumeration and metadata lookup are hot paths. Parallel reads can +hide I/O latency, especially on high-latency filesystems. Small or memory-backed +workloads can instead be dominated by startup, dispatch and per-entry CPU work; +admitting workers must be justified by useful remaining work, not file count +alone. Worker reuse removes repeated creation, but does not make dispatch free. Parallelism must not cost correctness. Three things are trivially correct in the current sequential walk and must stay correct under concurrency: deterministic @@ -40,22 +41,24 @@ of the design below; each has a noted follow-up: ## Architecture -A bounded directory-read worker pool with a single coordinator: +A command-owned executor with one calling coordinator and at most `N-1` +background workers under `--jobs=N`: - Workers perform only `ReadDir` plus metadata lookup and return complete listings. The coordinator submits sibling-directory reads ahead, then consumes their - futures in the order required by `--sort`. Directory workers start only when at least - two 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 pool. At most one - chunk per worker is pending; worker jobs never wait on nested jobs. Metadata-free, lazy and + 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 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 can evaluate in a separate bounded worker - pool; all stateful expressions, mutations and execution actions retain the coordinator evaluator. + Independent content predicates and rg content searches reuse that executor, with the caller also + claiming chunks; all stateful expressions, mutations and execution actions retain the coordinator evaluator. - Content matching schedules chunks of four owned entries, so clustered expensive files can use - several workers. Decision-only batches hold at most 1,024 entries; matching-line output retains - the smaller 256-entry bound. Fewer than 64 entries avoid initial thread startup, and tails below - 16 run inline even after a pool exists. Later larger batches can grow the pool up to the worker + several workers. Decision-only batches hold at most 8,192 entries; matching-line output retains + the smaller 256-entry bound. Cold admission is described under Parallelism control; tails below + 16 run inline even after workers exist. Later larger batches can grow the executor up to the background-worker allowance. Completed results return in input order, including a trailing path-only print action. Cheap name/type matching stays inline to avoid scheduling overhead. - Eligibility comes from audited descriptor capabilities, not names: only independent tests @@ -77,8 +80,10 @@ A bounded directory-read worker pool with a single coordinator: - 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. -Comparison evaluates eligible large regular-file pairs in batches of at most 64 after collecting -both trees. Each pair owns persistent read cursors; small patch inputs can be retained up to 64 KiB +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 +in batches of at most 64 on that executor and the caller, without creating a comparison-only +pool or asynchronous side coordinator. Each pair owns persistent read cursors; small patch inputs can be retained up to 64 KiB per side. The coordinator publishes sorted results, summaries and patches. Archive/owned-source pairs remain serial. This is not yet a completed-directory comparison pipeline. @@ -132,8 +137,11 @@ actions still occur in traversal order. ## Parallelism control -A single knob, `-j N` (long form `--jobs`), caps one run-owned executor at `N` -workers for directory reads, eager stat batches and eligible content matching. +A single knob, `-j N` (long form `--jobs`), caps discovery, eager stat, eligible +content and file comparison at `N` participants for the entire command, including +its calling coordinator. The command owns one executor with at most `N-1` +background threads; standalone `Walk` calls likewise reserve one participant for +their caller. The coordinator retains traversal and output ownership. The executor starts lazily, grows only as useful work becomes available, and reuses its threads across phases and mounted-filesystem walks until the command finishes. @@ -143,6 +151,18 @@ expression. `-j 1` makes each part synchronous. `-j all` (`--jobs=all`) uses eve detected core (`hardware_concurrency()`) for these limits, regardless of the active mode's default. +This ownership change remains an experiment, not an accepted performance result. +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 +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. diff --git a/xff/cli/explain_test.sh b/xff/cli/explain_test.sh index 6bba6927651..0e38c906b39 100755 --- a/xff/cli/explain_test.sh +++ b/xff/cli/explain_test.sh @@ -630,11 +630,13 @@ test::explain_resources_use_selected_config_without_traversal() { --buffer=2KiB INI out="$(XFF_TEST_USER_CONFIG="${cfg}" "$(_xff_bin)" "${dir}/absent" --config=resource-view --explain)" - expect_output_contains $'directory-workers-per-walk\t3' "${out}" || return + expect_output_contains $'execution-participants-per-command\t3' "${out}" || return + expect_output_contains $'background-workers-per-command\t2' "${out}" || return expect_output_contains $'content-field-occurrences\t2' "${out}" || return expect_output_contains $'listing-column-buffer\t2048 cell bytes' "${out}" || return out="$(XFF_TEST_USER_CONFIG="${cfg}" "$(_xff_bin)" "${dir}/absent" --config=resource-view --columns=name --jobs=1 --explain)" - expect_output_contains $'directory-workers-per-walk\t1' "${out}" || return + expect_output_contains $'execution-participants-per-command\t1' "${out}" || return + expect_output_contains $'background-workers-per-command\t0' "${out}" || return expect_output_contains $'content-field-occurrences\t0' "${out}" || return } diff --git a/xff/cli/globals.cc b/xff/cli/globals.cc index 790786ea8a3..ac94e6ca9cf 100644 --- a/xff/cli/globals.cc +++ b/xff/cli/globals.cc @@ -900,25 +900,28 @@ constexpr std::array kGlobals = std::to_array({ .group = "scheduling", .header = "Concurrency and ordering", .summary = "directory-read, eligible content-match, comparison and -exec workers", - .details = "`N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached " - "short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes " - "directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories " - "using one lazily started pool of up to `N` worker threads, reused for directory reads, " - "required eager metadata and eligible content matching. Workers drain batches of independent " - "work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start " - "stat workers. Independent content predicates and rg searches match in bounded parallel batches, " - "with output kept in traversal order. Small initial batches, archive members, stateful expressions " - "stay on the coordinator. Metadata-consuming output does not itself prevent eligible content " - "filters from running in workers; reductions and rendering remain serial. Independent " - "regular-file comparison pairs can also use bounded workers, preserving relative-path output " - "order. Batches of only small or metadata-only comparisons stay inline; archive and owned " - "filesystem sources stay serial. " - "Xff may independently keep up to `N` semicolon-form `-exec` / " - "`-execdir` children outstanding. Reads and children can overlap; `N` is not one shared " - "operation budget. The children's truth value is therefore success on launch. The " - "`... +` batch forms still run once after the walk and propagate a failing exit status. With " - "no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and " - "rg modes use every detected core. `all` always means every detected core.", + .details = + "`N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached " + "short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes " + "directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories " + "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.", .see_also = "output,stats", .alias_modes = registry::Modes::kNative, .alias_argument = "N", diff --git a/xff/cli/help_topic_test.sh b/xff/cli/help_topic_test.sh index 8222fb4fa68..6f74b038204 100755 --- a/xff/cli/help_topic_test.sh +++ b/xff/cli/help_topic_test.sh @@ -398,8 +398,11 @@ test::bad_flags_are_hard_errors_even_with_help() { test::meta_spelling_inside_exec_stays_a_child_argument() { # Meta recognition follows parser boundaries. A child command is an opaque # argument run, so its `--help` must not turn xff itself into the usage page. + # Use the POSIX shell builtin: external printf implementations can interpret + # --help specially even after a format argument (including local uutils 0.10.0). local out - out="$("$(_xff_bin)" --allow-exec . -maxdepth 0 -exec printf 'child:%s\n' --help ';' 2>&1)" + # shellcheck disable=SC2016 # The child shell expands its argument, not this test's shell. + out="$("$(_xff_bin)" --allow-exec . -maxdepth 0 -exec sh -c 'printf "child:%s\n" "$1"' sh --help ';' 2>&1)" expect_output_contains 'child:--help' "${out}" expect_output_not_contains 'eXtended File Find' "${out}" } diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel index 78bf4bbbde9..a9dc53f3fdb 100644 --- a/xff/engine/BUILD.bazel +++ b/xff/engine/BUILD.bazel @@ -374,6 +374,26 @@ cc_test( ], ) +cc_test( + name = "command_workers_test", + size = "small", + srcs = ["command_workers_test.cc"], + deps = [ + ":run_cc", + "//xff/parser:parser_cc", + "//xff/vfs:vfs_cc", + "@abseil-cpp//absl/base:core_headers", + "@abseil-cpp//absl/status", + "@abseil-cpp//absl/status:statusor", + "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/time", + "@googletest//:gtest_main", + "@mboworks_mbo//mbo/status:status_macros_cc", + "@mboworks_mbo//mbo/testing:matchers_cc", + "@mboworks_mbo//mbo/testing:status_cc", + ], +) + cc_test( name = "metadata_test", size = "small", @@ -627,6 +647,7 @@ cc_library( srcs = ["parallel_compare.cc"], hdrs = ["parallel_compare.h"], deps = [ + ":run_executor_cc", "//xff/vfs:vfs_cc", "@abseil-cpp//absl/base:core_headers", "@abseil-cpp//absl/status:statusor", diff --git a/xff/engine/command_workers_test.cc b/xff/engine/command_workers_test.cc new file mode 100644 index 00000000000..ac94152fca4 --- /dev/null +++ b/xff/engine/command_workers_test.cc @@ -0,0 +1,231 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/synchronization/mutex.h" +#include "absl/time/time.h" +#include "gmock/gmock.h" +#include "gtest/gtest.h" +#include "mbo/status/status_macros.h" +#include "mbo/testing/matchers.h" +#include "mbo/testing/status.h" +#include "xff/engine/run.h" +#include "xff/parser/parser.h" +#include "xff/vfs/filesystem.h" +#include "xff/vfs/read_source.h" + +namespace xff::engine { +namespace { +using ::mbo::testing::EqualsText; +using ::mbo::testing::IsOk; +using ::testing::Contains; +using ::testing::ElementsAre; +using ::testing::Eq; +using ::testing::Ge; +using ::testing::IsEmpty; +using ::testing::IsTrue; +using ::testing::Le; +using ::testing::SizeIs; + +// Each side has 64 independent sibling directories, enough to exercise read-ahead, +// followed by 64 large files that qualify for the comparison and content workers. +// No host filesystem or timing-based admission assumption enters these controls. +class CommandFs final : public vfs::FileSystem { + public: + explicit CommandFs(std::size_t participants, std::size_t content_participants = 0) + : participants_(participants), + content_participants_(content_participants == 0 ? participants : content_participants) {} + + absl::StatusOr> ReadDir(std::string_view path) const override { + if (!IsDirectory(path)) { + return absl::NotFoundError(path); + } + if (path == "left" || path == "right") { + const absl::MutexLock lock(mutex_); + root_readers_.insert(std::this_thread::get_id()); + std::vector entries; + entries.reserve(64); + for (std::size_t index = 0; index < 64; ++index) { + const auto name = std::to_string(index); + entries.push_back({.path = std::string(path) + "/" + name, .name = name, .type = vfs::FileType::kDirectory}); + } + return entries; + } + Record(path.starts_with("left/") ? 0 : 1, participants_ > 1 ? participants_ - 1 : 1); + return std::vector{ + { + .path = std::string(path) + "/file", + .name = "file", + .type = vfs::FileType::kRegular, + }, + }; + } + + absl::StatusOr Stat(std::string_view path, bool) const override { + const bool file = path.ends_with("/file"); + if (!IsDirectory(file ? path.substr(0, path.size() - 5) : path)) { + return absl::NotFoundError(path); + } + // Only ancestor identities matter for loop detection; left/right and all sibling + // directories have different identities from their own root. + return vfs::Metadata{ + .type = file ? vfs::FileType::kRegular : vfs::FileType::kDirectory, + .size = file ? bytes_.size() : 0, + .ino = path == "left" ? 1U + : path == "right" ? 2U + : 3U, + .dev = 1, + }; + } + + absl::StatusOr ContentSource(std::string_view path) const override { + if (!path.ends_with("/file") || !IsDirectory(path.substr(0, path.size() - 5))) { + return absl::NotFoundError(path); + } + Record(2, content_participants_); + return vfs::MemoryReadSource(bytes_); + } + + absl::StatusOr ReadContent(std::string_view path) const override { + MBO_ASSIGN_OR_RETURN(const auto source, ContentSource(path)); + return vfs::ReadSourceBytes(*source, bytes_.size()); + } + + absl::Status Remove(std::string_view) const override { return absl::PermissionDeniedError("read-only fixture"); } + + bool Access(std::string_view, vfs::AccessMode) const override { return false; } + + absl::StatusOr ReadLink(std::string_view) const override { + return absl::InvalidArgumentError("not a symlink"); + } + + absl::StatusOr FsType(std::string_view) const override { return "memory"; } + + absl::StatusOr IsCaseSensitive(std::string_view) const override { return true; } + + void CheckBudget() const { + const absl::MutexLock lock(mutex_); + EXPECT_THAT(root_readers_, ElementsAre(std::this_thread::get_id())); + std::set all = root_readers_; + for (const auto& threads : phase_readers_) { + all.insert(threads.begin(), threads.end()); + } + EXPECT_THAT(all, SizeIs(Le(participants_))); + EXPECT_THAT(all, Contains(std::this_thread::get_id())); + EXPECT_THAT(synchronized_, IsTrue()); + EXPECT_THAT(phase_readers_.at(2), SizeIs(Ge(content_participants_))); + EXPECT_THAT(phase_readers_.at(2), Contains(coordinator_)); + } + + void CheckWalkReuse() const { + const absl::MutexLock lock(mutex_); + EXPECT_THAT(phase_readers_.at(0), Eq(phase_readers_.at(1))); + EXPECT_THAT(phase_readers_.at(0), SizeIs(participants_ > 1 ? participants_ - 1 : 1)); + } + + private: + static bool IsDirectory(const std::string_view path) { + if (path == "left" || path == "right") { + return true; + } + const auto name = path.starts_with("left/") ? path.substr(5) + : path.starts_with("right/") ? path.substr(6) + : std::string_view{}; + return !name.empty() && std::ranges::all_of(name, [](char digit) { return digit >= '0' && digit <= '9'; }); + } + + void Record(std::size_t phase, std::size_t expected) const ABSL_LOCKS_EXCLUDED(mutex_) { + const absl::MutexLock lock(mutex_); + if (!phase_readers_.at(phase).insert(std::this_thread::get_id()).second) { + return; + } + // First calls rendezvous so one fast worker cannot drain the whole fixture before + // its peers run. Timeout bounds a broken scheduler instead of hanging a test. + const auto ready = [&] ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) { + return phase_readers_.at(phase).size() >= expected + && (phase != 2 || phase_readers_.at(phase).contains(coordinator_)); + }; + synchronized_ = mutex_.AwaitWithTimeout(absl::Condition(&ready), absl::Seconds(5)) && synchronized_; + } + + const std::size_t participants_; + const std::size_t content_participants_; + const std::thread::id coordinator_ = std::this_thread::get_id(); + const std::string bytes_ = std::string(262'144, 'a') + "needle\n"; + // The mutex guards reader identities and rendezvous outcomes. Filesystem data is immutable. + mutable absl::Mutex mutex_; + mutable std::set root_readers_ ABSL_GUARDED_BY(mutex_); + mutable std::array, 3> phase_readers_ ABSL_GUARDED_BY(mutex_); + mutable bool synchronized_ ABSL_GUARDED_BY(mutex_) = true; +}; + +struct CommandWorkersTest : ::testing::Test { + static absl::StatusOr Run(const std::vector& arguments, const CommandFs& fs) { + MBO_ASSIGN_OR_RETURN(auto command, parser::Parse(arguments)); + parser::BindMatchers(command, parser::GrammarFromGlobals(command.globals), parser::CaseMode::kSensitive); + std::string output; + std::vector errors; + const auto result = RunFind( + command, fs, [&](std::string_view text) { output.append(text); }, + [&](std::string_view, absl::Status status) { errors.push_back(std::move(status)); }); + EXPECT_THAT(errors, IsEmpty()); + EXPECT_THAT(result.errors, Eq(0)); + return output; + } +}; + +TEST_F(CommandWorkersTest, SingleParticipantComparisonHasNoSecondCoordinator) { + const CommandFs fs(1); + EXPECT_THAT(Run({"--jobs=1", "--compare", "--compare-select=identical", "left", "right"}, fs), IsOk()); + fs.CheckBudget(); + fs.CheckWalkReuse(); +} + +TEST_F(CommandWorkersTest, ComparisonReusesBoundedWorkersAcrossBothWalksAndPairs) { + const CommandFs serial(1); + ASSERT_OK_AND_ASSIGN( + const auto expected, Run({"--jobs=1", "--compare", "--compare-select=identical", "left", "right"}, serial)); + for (const std::size_t participants : {2, 3, 10}) { + SCOPED_TRACE(participants); + const CommandFs fs(participants); + ASSERT_OK_AND_ASSIGN( + const auto output, + Run({"--jobs=" + std::to_string(participants), "--compare", "--compare-select=identical", "left", "right"}, + fs)); + EXPECT_THAT(output, EqualsText(expected)); + fs.CheckBudget(); + fs.CheckWalkReuse(); + } +} + +TEST_F(CommandWorkersTest, ContentMatchingIncludesTheCallingCoordinatorInTheAllowance) { + const CommandFs serial(1); + ASSERT_OK_AND_ASSIGN(const auto expected, Run({"--jobs=1", "--sort=tree", "left", "-content", "needle"}, serial)); + for (const std::size_t participants : {2, 3, 10}) { + SCOPED_TRACE(participants); + // Matching sees directories as well as files. Force caller/worker overlap without + // assuming that every admitted logical worker reaches a file in this small tail. + const CommandFs fs(participants, std::min(participants, std::size_t{2})); + ASSERT_OK_AND_ASSIGN( + const auto output, + Run({"--jobs=" + std::to_string(participants), "--sort=tree", "left", "-content", "needle"}, fs)); + EXPECT_THAT(output, EqualsText(expected)); + fs.CheckBudget(); + } +} + +} // namespace +} // namespace xff::engine diff --git a/xff/engine/parallel_compare.cc b/xff/engine/parallel_compare.cc index 7d8f058b51e..e201cd763c4 100644 --- a/xff/engine/parallel_compare.cc +++ b/xff/engine/parallel_compare.cc @@ -111,18 +111,10 @@ absl::StatusOr ComparePair(const ComparisonPair& pair, bool re } // namespace -ParallelCompare::ParallelCompare(std::size_t workers, bool retain_patch_inputs) - : workers_(std::max(workers, std::size_t{1})), retain_patch_inputs_(retain_patch_inputs) {} +ParallelCompare::ParallelCompare(std::size_t workers, RunExecutor& executor, bool retain_patch_inputs) + : workers_(std::max(workers, std::size_t{1})), executor_(executor), retain_patch_inputs_(retain_patch_inputs) {} -ParallelCompare::~ParallelCompare() { - { - const absl::MutexLock lock(mutex_); - stop_ = true; - } - for (auto& thread : threads_) { - thread.join(); - } -} +ParallelCompare::~ParallelCompare() = default; bool ParallelCompare::UseWorkers() const { std::size_t reads = 0; @@ -143,16 +135,7 @@ bool ParallelCompare::UseWorkers() const { return workers_ > 1 && reads > 1 && bytes >= 4 * kRetainedFileBytes; } -bool ParallelCompare::Ready(std::size_t generation) const { - return stop_ || generation_ != generation; -} - -bool ParallelCompare::Finished() const { - return remaining_ == 0; -} - const std::vector>& ParallelCompare::Compare(std::vector inputs) { - const absl::MutexLock lock(mutex_); inputs_ = std::move(inputs); results_.clear(); results_.resize(inputs_.size()); @@ -161,34 +144,19 @@ const std::vector>& ParallelCompare::Compare(st EvaluateEntries(); return results_; } - const auto count = std::min(workers_, inputs_.size()); - threads_.reserve(count); - while (threads_.size() < count) { - threads_.emplace_back([this] { Run(); }); + const auto participants = std::min(workers_, inputs_.size()); + executor_.Start(participants - 1); + const auto count = std::min(participants - 1, executor_.worker_count()); + std::vector> tasks; + tasks.reserve(count); + for (std::size_t worker = 0; worker < count; ++worker) { + tasks.push_back(executor_.Submit([this] { EvaluateEntries(); })); } - remaining_ = threads_.size(); - ++generation_; - mutex_.Await(absl::Condition(this, &ParallelCompare::Finished)); - return results_; -} - -void ParallelCompare::Run() { - std::size_t generation = 0; - for (;;) { - { - const auto ready = [&] ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) { return Ready(generation); }; - const absl::MutexLock lock(mutex_, absl::Condition(&ready)); - if (stop_) { - return; - } - generation = generation_; - } - EvaluateEntries(); - { - const absl::MutexLock lock(mutex_); - --remaining_; - } + EvaluateEntries(); + for (RunTask& task : tasks) { + task.Get(); } + return results_; } void ParallelCompare::EvaluateEntries() { diff --git a/xff/engine/parallel_compare.h b/xff/engine/parallel_compare.h index c31e156d701..85477a21453 100644 --- a/xff/engine/parallel_compare.h +++ b/xff/engine/parallel_compare.h @@ -10,13 +10,11 @@ #include #include #include -#include #include -#include "absl/base/thread_annotations.h" #include "absl/status/statusor.h" -#include "absl/synchronization/mutex.h" #include "mbo/types/optional_ref.h" +#include "xff/engine/run_executor.h" #include "xff/vfs/filesystem.h" namespace xff::engine { @@ -49,12 +47,14 @@ struct ComparisonResult { // must outlive Compare and consumption of the returned results/Inputs. Owned archive sources stay serial. // Callers submit at most kBatchSize pairs. // Workers publish values only; errors, patches and summaries are emitted by the coordinator. +// workers includes the calling coordinator. The caller owns and outlives the shared executor; +// Compare completes all its leaf jobs before returning and never creates a separate thread pool. class ParallelCompare final { public: static constexpr std::size_t kBatchSize = 64; static constexpr std::size_t kRetainedFileBytes = 65'536; - explicit ParallelCompare(std::size_t workers, bool retain_patch_inputs = false); + ParallelCompare(std::size_t workers, RunExecutor& executor, bool retain_patch_inputs = false); ~ParallelCompare(); ParallelCompare(const ParallelCompare&) = delete; ParallelCompare& operator=(const ParallelCompare&) = delete; @@ -67,18 +67,13 @@ class ParallelCompare final { private: bool UseWorkers() const; - bool Ready(std::size_t generation) const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - bool Finished() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); - void Run(); void EvaluateEntries(); const std::size_t workers_; + RunExecutor& executor_; const bool retain_patch_inputs_; - std::vector threads_; - absl::Mutex mutex_; - bool stop_ ABSL_GUARDED_BY(mutex_) = false; - std::size_t generation_ ABSL_GUARDED_BY(mutex_) = 0; - std::size_t remaining_ ABSL_GUARDED_BY(mutex_) = 0; + // The coordinator owns batches. Atomic next_ gives each participant distinct result + // elements; all submitted leaf jobs complete before Compare returns or inputs change. std::vector inputs_; std::vector> results_; std::atomic next_ = 0; diff --git a/xff/engine/parallel_compare_test.cc b/xff/engine/parallel_compare_test.cc index 244e0b5024f..45c2887ea3b 100644 --- a/xff/engine/parallel_compare_test.cc +++ b/xff/engine/parallel_compare_test.cc @@ -140,13 +140,14 @@ struct CompareFs final : vfs::FileSystem { }; struct ParallelCompareTest : ::testing::Test { + RunExecutor executor{3}; CompareFs fs; TreeCompareEntry left{.path = "left", .metadata = {.type = vfs::FileType::kRegular, .size = 4}, .fs = fs}; TreeCompareEntry right{.path = "right", .metadata = {.type = vfs::FileType::kRegular, .size = 4}, .fs = fs}; }; TEST_F(ParallelCompareTest, RetainsOnlyCompleteSmallDifferingInputs) { - ParallelCompare compare(1, true); + ParallelCompare compare(1, executor, true); ASSERT_OK_AND_ASSIGN(const auto result, compare.Compare({{.path = "file", .left = left, .right = right}}).front()); EXPECT_THAT(result.same, IsFalse()); EXPECT_THAT(result.bytes, Optional(Field(&ComparisonBytes::left, EqualsText("left")))); @@ -164,7 +165,7 @@ TEST_F(ParallelCompareTest, LargeComparisonsOpenEachSourceOnceAndKeepNoPatchByte fs.files.at("right") = fs.files.at("left"); left.metadata.size = 200'000; right.metadata.size = 200'000; - ParallelCompare compare(1, true); + ParallelCompare compare(1, executor, true); EXPECT_THAT( compare.Compare({{.left = left, .right = right}}).front(), IsOkAndHolds(Field(&ComparisonResult::same, IsTrue()))); @@ -177,7 +178,7 @@ TEST_F(ParallelCompareTest, LargeComparisonsOpenEachSourceOnceAndKeepNoPatchByte } TEST_F(ParallelCompareTest, RepeatedBatchesKeepInputOrderAcrossWorkersAndErrors) { - ParallelCompare compare(4); + ParallelCompare compare(4, executor); fs.track_threads = true; left.metadata.size = right.metadata.size = 262'144; const TreeCompareEntry missing{.path = "missing", .metadata = left.metadata, .fs = fs}; @@ -197,7 +198,7 @@ TEST_F(ParallelCompareTest, ShortReadsAndEarlyEofPreserveRangeSemantics) { const auto source = std::make_shared(); fs.source_override = source; left.metadata.size = right.metadata.size = 100; - ParallelCompare compare(1, true); + ParallelCompare compare(1, executor, true); EXPECT_THAT( compare.Compare({{.left = left, .right = right}}).front(), IsOkAndHolds(Field(&ComparisonResult::same, IsTrue()))); @@ -218,7 +219,7 @@ TEST_F(ParallelCompareTest, CursorFailuresRemainErrors) { fs.source_override = source; source->fail_open = true; left.metadata.size = right.metadata.size = 262'144; - ParallelCompare compare(1, true); + ParallelCompare compare(1, executor, true); EXPECT_THAT(compare.Compare({{.left = left, .right = right}}).front(), StatusIs(absl::StatusCode::kPermissionDenied)); source->fail_open = false; source->fail_at = 2; @@ -227,7 +228,7 @@ TEST_F(ParallelCompareTest, CursorFailuresRemainErrors) { TEST_F(ParallelCompareTest, EmptyFilesNeedNoCursorAndOwnedSourcesStaySerial) { left.metadata.size = right.metadata.size = 0; - ParallelCompare compare(4); + ParallelCompare compare(4, executor); EXPECT_THAT( compare.Compare({{.left = left, .right = right}}).front(), IsOkAndHolds(Field(&ComparisonResult::same, IsTrue()))); @@ -254,7 +255,7 @@ TEST_F(ParallelCompareTest, EmptyFilesNeedNoCursorAndOwnedSourcesStaySerial) { } TEST_F(ParallelCompareTest, CheapKindsAndSizesDoNotReadContent) { - ParallelCompare compare(4); + ParallelCompare compare(4, executor); EXPECT_THAT(compare.Compare({{.left = left}}).front(), IsOkAndHolds(Field(&ComparisonResult::same, IsFalse()))); EXPECT_THAT(compare.Compare({{.right = right}}).front(), IsOkAndHolds(Field(&ComparisonResult::same, IsFalse()))); right.metadata.size = 5; @@ -284,7 +285,7 @@ TEST_F(ParallelCompareTest, LeftFailurePrecedesRightOpenAndReadErrors) { fs.source_override = source; source->fail_at = 0; left.metadata.size = right.metadata.size = 262'144; - ParallelCompare compare(1); + ParallelCompare compare(1, executor); EXPECT_THAT(compare.Compare({{.left = left, .right = right}}).front(), StatusIs(absl::StatusCode::kDataLoss)); EXPECT_THAT(right_fs.sources.load(), Eq(0)); source->fail_at = 100; @@ -304,7 +305,7 @@ TEST_F(ParallelCompareTest, PatchRetentionRequiresSuccessfulEofProbesOnBothSides source->bytes = "abcd"; source->fail_at = 4; fs.source_override = source; - ParallelCompare compare(1, true); + ParallelCompare compare(1, executor, true); EXPECT_THAT(compare.Compare({{.left = left, .right = right}}).front(), StatusIs(absl::StatusCode::kDataLoss)); source->fail_at = 100; const auto right_source = std::make_shared(); @@ -327,7 +328,7 @@ TEST_F(ParallelCompareTest, CursorBuffersRespectAndReleaseSourceBudget) { left.metadata.size = right.metadata.size = 262'144; const auto budget = std::make_shared(ParallelCompare::kRetainedFileBytes); fs.source_override = vfs::MemoryReadSource("short", budget); - ParallelCompare compare(1); + ParallelCompare compare(1, executor); EXPECT_THAT( compare.Compare({{.left = left, .right = right}}).front(), StatusIs(absl::StatusCode::kResourceExhausted)); EXPECT_THAT(budget->MemoryUsed(), Eq(0)); @@ -341,7 +342,7 @@ TEST_F(ParallelCompareTest, CursorBuffersRespectAndReleaseSourceBudget) { TEST_F(ParallelCompareTest, SymlinkErrorsRemainErrors) { left.metadata.type = right.metadata.type = vfs::FileType::kSymlink; - ParallelCompare compare(1); + ParallelCompare compare(1, executor); left.path = "missing"; EXPECT_THAT(compare.Compare({{.left = left, .right = right}}).front(), StatusIs(absl::StatusCode::kNotFound)); left.path = "left"; diff --git a/xff/engine/parallel_match.cc b/xff/engine/parallel_match.cc index 33eada5398c..612af3428f8 100644 --- a/xff/engine/parallel_match.cc +++ b/xff/engine/parallel_match.cc @@ -117,8 +117,14 @@ const std::vector& ParallelMatch::Match(std::vector(*this)); } @@ -127,6 +133,10 @@ const std::vector& ParallelMatch::Match(std::vector& task : tasks) { task.Get(); } diff --git a/xff/engine/parallel_match_test.cc b/xff/engine/parallel_match_test.cc index 4ef6d3b6268..71df891c414 100644 --- a/xff/engine/parallel_match_test.cc +++ b/xff/engine/parallel_match_test.cc @@ -24,6 +24,7 @@ namespace { using ::testing::Eq; using ::testing::IsFalse; using ::testing::IsTrue; +using ::testing::Le; using ::testing::SizeIs; struct ParallelMatchTest : ::testing::Test { @@ -114,7 +115,7 @@ TEST_F(ParallelMatchTest, PreparedCoordinatorAndWorkersRetainRegexStateAcrossPoo EXPECT_THAT(results.at(index).evaluation.matched, Eq(path.find_first_of("13579") == std::string::npos)); } } - EXPECT_THAT(executor_pool.worker_count(), Eq(4)); + EXPECT_THAT(executor_pool.worker_count(), Eq(3)); } TEST_F(ParallelMatchTest, ColdExecutorWaitsForAccumulatedWork) { @@ -123,10 +124,12 @@ TEST_F(ParallelMatchTest, ColdExecutorWaitsForAccumulatedWork) { ParallelMatch matcher(*command.expression, 4, executor, false); for (int batch = 0; batch < 7; ++batch) { EXPECT_THAT(matcher.Match(Entries(1'024)), SizeIs(1'024)); - EXPECT_THAT(executor.worker_count(), Eq(0)); + // Instrumented builds can make a probe expensive enough to admit workers. + // Thread budgeting must hold regardless of that timing-dependent decision. + EXPECT_THAT(executor.worker_count(), Le(3)); } EXPECT_THAT(matcher.Match(Entries(1'024)), SizeIs(1'024)); - EXPECT_THAT(executor.worker_count(), Eq(4)); + EXPECT_THAT(executor.worker_count(), Eq(3)); } TEST_F(ParallelMatchTest, StatefulMetadataAndActionExpressionsCannotEnterWorkers) { diff --git a/xff/engine/resource_inspection_test.cc b/xff/engine/resource_inspection_test.cc index 4865ff6cb03..f40fd554a7b 100644 --- a/xff/engine/resource_inspection_test.cc +++ b/xff/engine/resource_inspection_test.cc @@ -29,8 +29,9 @@ struct ResourceInspectionTest : ::testing::Test { TEST_F(ResourceInspectionTest, InspectsAbsentRootsWithoutTraversal) { ASSERT_OK_AND_ASSIGN(const auto output, Inspect({"--jobs=3", "/absent-resource-fixture", "-type", "f"})); - EXPECT_THAT(output, HasSubstr("directory-workers-per-walk\t3")); - EXPECT_THAT(output, HasSubstr("eligible-command-workers-per-walk\t3 (independent")); + EXPECT_THAT(output, HasSubstr("execution-participants-per-command\t3")); + EXPECT_THAT(output, HasSubstr("background-workers-per-command\t2")); + EXPECT_THAT(output, HasSubstr("eligible-command-children\t3 (independent")); EXPECT_THAT(output, HasSubstr("content-field-occurrences\t0")); EXPECT_THAT(output, Not(HasSubstr("column-buffer\t"))); EXPECT_THAT(output, HasSubstr("static inspection, not measured usage")); @@ -67,7 +68,8 @@ TEST_F(ResourceInspectionTest, ComparisonInventoriesAreIndependentOfColumnBuffer const auto output, Inspect({"/left", "/right", "--compare=summary", "--summary=ext", "--buffer=off", "--jobs=2"})); EXPECT_THAT(output, HasSubstr("walks\t2")); - EXPECT_THAT(output, HasSubstr("directory-workers-per-walk\t2")); + EXPECT_THAT(output, HasSubstr("execution-participants-per-command\t2")); + EXPECT_THAT(output, HasSubstr("background-workers-per-command\t1")); EXPECT_THAT(output, HasSubstr("comparison-state\tboth matched inventories")); EXPECT_THAT(output, HasSubstr("comparison-summary-state\tper-entry contributions")); EXPECT_THAT(output, HasSubstr("--buffer is not a process-memory cap or a comparison-inventory bound")); diff --git a/xff/engine/rg_test.cc b/xff/engine/rg_test.cc index de5bffe1df8..7682f3fc4e8 100644 --- a/xff/engine/rg_test.cc +++ b/xff/engine/rg_test.cc @@ -54,6 +54,7 @@ using ::testing::IsEmpty; using ::testing::IsFalse; using ::testing::IsTrue; using ::testing::Not; +using ::testing::SizeIs; // A restarted cursor reports a failure after an otherwise selected line. class FailingSource final : public vfs::ReadSource { @@ -436,6 +437,31 @@ TEST_F(RgEngineTest, OrdinaryRgSearchUsesContentWorkersWithoutNativeExpression) EXPECT_THAT(output, HasSubstr("tree/299:hit 299\n")); } +TEST_F(RgEngineTest, SlowProbeThatCompletesTheBatchDoesNotStartIdleWorkers) { + fs.files.clear(); + std::vector entries; + for (std::size_t index = 0; index < 16; ++index) { + const auto path = "tree/" + std::to_string(index); + fs.files.emplace(path, "hit\n"); + entries.push_back({.path = path, .metadata = {.type = vfs::FileType::kRegular}, .fs = fs}); + } + // The fixture's first-read rendezvous times out after two seconds without a peer, + // making the coordinator prefix deliberately expensive. All 16 entries belong to + // that prefix: no remaining work may justify starting even one background worker. + fs.track_threads = true; + fs.expected_threads = 2; + ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"tree", "-content", "hit"})); + RunExecutor executor(3); + ParallelMatch matcher(*command.expression, 4, executor, false); + const auto& results = matcher.Match(std::move(entries)); + EXPECT_THAT(results, SizeIs(16)); + for (const auto& result : results) { + EXPECT_THAT(result.evaluation.matched, IsTrue()); + } + EXPECT_THAT(executor.worker_count(), Eq(0)); + EXPECT_THAT(fs.read_threads, ElementsAre(std::this_thread::get_id())); +} + TEST_F(RgEngineTest, MatcherPoolGrowsForLaterBatchesAndRunsShortTailsInline) { fs.files.clear(); for (std::size_t index = 0; index < 256; ++index) { @@ -534,8 +560,13 @@ TEST_F(RgEngineTest, ParallelNativeFilterAndRgSelectionShareEachRead) { } TEST_F(RgEngineTest, MetadataConsumersKeepParallelFilteringAndOrderedResults) { + // This controls filter eligibility with metadata consumers, not cold admission. + // Eager metadata for at least 512 entries warms the shared executor before matching; + // the first-read rendezvous then forces real overlap in the content batch. With a + // smaller cold batch, its serial probe can use up that rendezvous before dispatch. + constexpr std::size_t kFiles = 600; fs.files.clear(); - for (std::size_t index = 0; index < 300; ++index) { + for (std::size_t index = 0; index < kFiles; ++index) { fs.files.emplace("tree/" + std::to_string(index) + ".txt", index % 2 == 0 ? "hit\n" : "miss\n"); } const std::vector> consumers{ @@ -546,7 +577,9 @@ TEST_F(RgEngineTest, MetadataConsumersKeepParallelFilteringAndOrderedResults) { {"--no-match-output", "--color=always"}, }; for (const auto& consumer : consumers) { + SCOPED_TRACE(consumer.front()); for (const bool rg : {false, true}) { + SCOPED_TRACE(rg); auto args = consumer; args.insert(args.begin(), {"--jobs=1", "--archive=none", "--sort=none"}); if (rg) { @@ -563,8 +596,8 @@ TEST_F(RgEngineTest, MetadataConsumersKeepParallelFilteringAndOrderedResults) { EXPECT_THAT(Run(args, rg).errors, Eq(0)); fs.track_threads = false; EXPECT_THAT(output, EqualsText(expected)); - EXPECT_THAT(fs.reads.load(), Eq(300)); - EXPECT_THAT(fs.read_threads.size(), Gt(1)); + EXPECT_THAT(fs.reads.load(), Eq(kFiles)); + EXPECT_THAT(fs.read_threads, SizeIs(Gt(1))); } } } diff --git a/xff/engine/run.cc b/xff/engine/run.cc index 592fd29f859..dddf4419ac7 100644 --- a/xff/engine/run.cc +++ b/xff/engine/run.cc @@ -25,11 +25,9 @@ #include #include #include -#include #include #include #include -#include #include #include #include @@ -256,7 +254,7 @@ bool ResolveRankByScore(const std::vector& globals) { return rank; } -// xff -j N / -j=N / --jobs=N: worker threads for the parallel directory read-ahead (see +// xff -j N / -j=N / --jobs=N: command participants, including the calling coordinator (see // docs/design-parallel.md). When absent, the count is style-scoped (DefaultWorkers). // Last occurrence wins; a non-positive or unparseable value is a usage error. absl::StatusOr ResolveJobs(const std::vector& globals, std::optional style) { @@ -4175,6 +4173,7 @@ RunResult RunFindCore( std::optional style, mbo::types::OptionalRef matched_entry, bool compare_listing, + RunExecutor& run_executor, ExpressionFactory executor, mbo::types::OptionalRef comparison_summaries = std::nullopt, std::size_t root_offset = 0); @@ -4358,6 +4357,7 @@ RunResult RunTreeCompare( WalkErrorFn on_error, std::optional style, std::size_t table_width, + RunExecutor& run_executor, ExpressionFactory executor) { if (command.roots.size() != 2) { on_error("--compare", absl::InvalidArgumentError("requires exactly two roots")); @@ -4417,7 +4417,6 @@ RunResult RunTreeCompare( std::array side_summaries; std::map categories; std::array entries; - std::mutex callback_mutex; std::set reported_errors; const auto run_side = [&](std::size_t side) { const auto collect_callback = [&](const Visit& visit) { @@ -4431,26 +4430,21 @@ RunResult RunTreeCompare( .fs_owner = visit.fs_owner, }); }; - const auto emit_callback = [&](std::string_view text) { - const std::scoped_lock lock(callback_mutex); - emit(text); - }; const auto error_callback = [&](std::string_view path, absl::Status status) { - const std::scoped_lock lock(callback_mutex); if (reported_errors.emplace(absl::StrCat(path, "\n", status.ToString())).second) { on_error(path, std::move(status)); } }; const MatchedEntryFn collect = collect_callback; - const EmitFn synchronized_emit = emit_callback; - const WalkErrorFn synchronized_error = error_callback; + const WalkErrorFn deduplicated_error = error_callback; return RunFindCore( - command, histograms, absl::MakeConstSpan(command.roots).subspan(side, 1), fs, synchronized_emit, - synchronized_error, style, collect, /*compare_listing=*/true, executor, side_summaries.at(side), side); + command, histograms, absl::MakeConstSpan(command.roots).subspan(side, 1), fs, emit, deduplicated_error, style, + collect, /*compare_listing=*/true, run_executor, executor, side_summaries.at(side), side); }; - std::future left_result = std::async(std::launch::async, run_side, 0); + // One coordinator owns both listings. Their read-ahead and matching reuse the command's + // executor; starting a second coordinator would exceed the allowance even under --jobs=1. + const RunResult left_run_result = run_side(0); const RunResult right_result = run_side(1); - const RunResult left_run_result = left_result.get(); if (left_run_result.errors != 0 || right_result.errors != 0) { return RunResult{.errors = std::max(left_run_result.errors, right_result.errors)}; } @@ -4474,7 +4468,8 @@ RunResult RunTreeCompare( } }; ParallelCompare comparisons( - ResolveJobs(command.globals, style).value_or(1), output == TreeCompareOutput::kDiff && selection.different); + ResolveJobs(command.globals, style).value_or(1), run_executor, + output == TreeCompareOutput::kDiff && selection.different); const auto selected = std::array{selection.left_only, selection.right_only, selection.different, selection.identical}; auto left = entries[0].begin(); auto right = entries[1].begin(); @@ -4661,6 +4656,7 @@ RunResult RunFindCore( std::optional style, mbo::types::OptionalRef matched_entry, bool compare_listing, + RunExecutor& run_executor, ExpressionFactory executor, mbo::types::OptionalRef comparison_summaries, std::size_t root_offset) { @@ -5885,7 +5881,6 @@ RunResult RunFindCore( serial_execution ? std::optional(serial_execution->MakeWorker(ExpressionWorkerRole::kCoordinator)) : std::nullopt; const auto output_evaluator = output_execution ? std::optional(output_execution->MakeWorker(ExpressionWorkerRole::kCoordinator)) : std::nullopt; - RunExecutor run_executor(options.workers > 1 ? options.workers : std::size_t{0}); std::optional parallel_match; if (use_match_pool) { std::optional content_output; @@ -7172,9 +7167,10 @@ absl::StatusOr ExplainResources(const parser::Command& command, std }); std::string output = absl::StrCat( "\n# execution resources: static inspection, not measured usage\n", "walks\t", compare ? 2 : 1, "\n", - "directory-workers-per-walk\t", workers, "\n", "eligible-command-workers-per-walk\t", workers, + "execution-participants-per-command\t", workers, "\n", "background-workers-per-command\t", + workers > 1 ? workers - 1 : 0, "\n", "eligible-command-children\t", workers, " (independent semicolon-form child pool)\n", - "evaluation\tordered coordinator; eligible content uses independent workers\n", + "evaluation\tordered coordinator; eligible content and comparison share command workers\n", "traversal-state\tdirectory listings, read-ahead, and traversal stack\n", "content-field-occurrences\t", resources.content_fields, " (hash/lines segments, not predicted read calls)\n", "line-histogram-consumers\t", line_histograms, "\n", "expensive-primaries\t", @@ -7262,12 +7258,20 @@ RunResult RunFind( on_error("--summary-scope", status); return RunResult{.errors = 2}; } + const auto workers = ResolveJobs(command.globals, style); + if (!workers.ok()) { + on_error("--jobs", workers.status()); + return RunResult{.errors = 2}; + } + // The calling coordinator is one participant. All discovery, stat, content and comparison + // phases reuse at most N-1 background workers until this command returns. + RunExecutor run_executor(*workers > 1 ? *workers - 1 : std::size_t{0}); if (compare) { - return RunTreeCompare(command, histograms, fs, emit, on_error, style, table_width, executor); + return RunTreeCompare(command, histograms, fs, emit, on_error, style, table_width, run_executor, executor); } return RunFindCore( command, histograms, command.roots, fs, emit, on_error, style, mbo::types::OptionalRef{}, - /*compare_listing=*/false, executor); + /*compare_listing=*/false, run_executor, executor); } namespace { diff --git a/xff/engine/run.h b/xff/engine/run.h index 44bd8aa3ad5..9043dd3610b 100644 --- a/xff/engine/run.h +++ b/xff/engine/run.h @@ -83,6 +83,8 @@ struct RunResult { // The default factory prepares bound callbacks and typed operands once per run expression. // A null executor preserves the original tree directly for differential qualification. // Qualification factories are test-only and cannot be linked into the CLI. +// One command-owned executor serves traversal, eligible matching and file comparison. The resolved +// --jobs=N allowance includes the caller, so at most N-1 background workers are started lazily. RunResult RunFind( const parser::Command& command, const vfs::FileSystem& fs, diff --git a/xff/engine/run_executor.h b/xff/engine/run_executor.h index 7e8d959849b..f5116f10f9a 100644 --- a/xff/engine/run_executor.h +++ b/xff/engine/run_executor.h @@ -92,7 +92,9 @@ class RunPromise final { std::shared_ptr> state_ = std::make_shared>(); }; -// One lazily started worker executor shared by all parallel phases of a run. +// One lazily started worker executor shared by all parallel phases of a command. +// max_workers counts background threads; command owners reserve one participant +// for the calling coordinator and therefore pass N-1 for a --jobs=N allowance. // Start grows the pool but never shrinks it; destruction joins the workers once. // Start, Submit and worker_count are called only by the coordinator. Jobs stay // leaves and must not submit or wait on work in this executor. diff --git a/xff/engine/run_executor_test.cc b/xff/engine/run_executor_test.cc index 5f25f6f7187..3e0bf0381e1 100644 --- a/xff/engine/run_executor_test.cc +++ b/xff/engine/run_executor_test.cc @@ -47,6 +47,7 @@ TEST_F(RunExecutorTest, ReusesTheSameBoundedThreadsAcrossPhases) { const auto phase = [&] { absl::Barrier started(3); std::vector> tasks; + tasks.reserve(3); for (int worker = 0; worker < 3; ++worker) { tasks.push_back(executor.Submit([&] { started.Block(); @@ -68,8 +69,8 @@ TEST_F(RunExecutorTest, ReusesTheSameBoundedThreadsAcrossPhases) { TEST_F(RunExecutorTest, SupportsMoveOnlyJobsAndResults) { RunExecutor executor(1); executor.Start(1); - auto task = executor.Submit([value = std::make_unique(42)]() mutable { return std::move(value); }); - auto result = task.Get(); + auto task = executor.Submit([value = std::make_unique(42)] mutable { return std::move(value); }); + const auto result = task.Get(); EXPECT_THAT(*result, Eq(42)); EXPECT_THAT(RunTask{}.Valid(), IsFalse()); executor.Submit([] {}).Get(); diff --git a/xff/engine/walk.cc b/xff/engine/walk.cc index cec68b8ed2c..64476713aa3 100644 --- a/xff/engine/walk.cc +++ b/xff/engine/walk.cc @@ -94,11 +94,17 @@ class Walker { // A stop action may leave prefetched siblings unconsumed. Their drain jobs // still reference this walker, so join those jobs before its state is torn // down. Normal walks reach already-completed jobs here. - for (RunTask& drain : drain_tasks_) { + for (const RunTask& drain : drain_tasks_) { drain.Wait(); } } + // Leaf jobs capture this walker, so its identity must remain stable until drained. + Walker(const Walker&) = delete; + Walker& operator=(const Walker&) = delete; + Walker(Walker&&) = delete; + Walker& operator=(Walker&&) = delete; + // Whether `stated` is a FILE this walk should try to open as a container. `kRoots` offers only the // paths named on the command line (depth 0), `kAll` offers every file met; `kNone` never asks. bool ShouldTryMount(const Stated& stated, int depth) const { @@ -272,27 +278,29 @@ class Walker { return StatEntries(absl::MakeSpan(entries)); } const auto chunks = (entries.size() + kChunk - 1) / kChunk; - const auto pending = std::min(chunks, options_.workers); - executor_.Start(pending); + executor_.Start(std::min(chunks - 1, options_.workers - 1)); + const auto pending = std::min({chunks - 1, executor_.worker_count(), options_.workers - 1}); std::atomic next = 0; std::vector children(entries.size()); std::vector> reads; reads.reserve(pending); - for (std::size_t worker = 0; worker < pending; ++worker) { - reads.push_back(executor_.Submit([&] { - for (;;) { - const std::size_t index = next.fetch_add(1, std::memory_order_relaxed); - if (index >= chunks) { - return; - } - const std::size_t first = index * kChunk; - const std::size_t end = std::min(first + kChunk, entries.size()); - for (std::size_t entry_index = first; entry_index < end; ++entry_index) { - children[entry_index] = StatEntry(std::move(entries[entry_index])); - } + const auto drain = [&] { + for (;;) { + const std::size_t index = next.fetch_add(1, std::memory_order_relaxed); + if (index >= chunks) { + return; } - })); + const std::size_t first = index * kChunk; + const std::size_t end = std::min(first + kChunk, entries.size()); + for (std::size_t entry_index = first; entry_index < end; ++entry_index) { + children[entry_index] = StatEntry(std::move(entries[entry_index])); + } + } + }; + for (std::size_t worker = 0; worker < pending; ++worker) { + reads.push_back(executor_.Submit(drain)); } + drain(); for (RunTask& read : reads) { read.Get(); } @@ -530,7 +538,7 @@ class Walker { return reads; } std::erase_if(drain_tasks_, [](RunTask& drain) { return drain.Ready(); }); - executor_.Start(directories.size()); + executor_.Start(std::min(directories.size(), options_.workers - 1)); struct ReadBatch final { std::vector paths; @@ -538,7 +546,7 @@ class Walker { std::atomic next = 0; }; - auto batch = std::make_shared(); + const auto batch = std::make_shared(); batch->paths.reserve(directories.size()); batch->promises.resize(directories.size()); for (std::size_t batch_index = 0; batch_index < directories.size(); ++batch_index) { @@ -546,7 +554,7 @@ class Walker { 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()); + 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 (;;) { @@ -590,7 +598,7 @@ absl::Status Walk( const WalkOptions& options, Visitor visit, WalkErrorFn on_error) { - RunExecutor executor(options.workers > 1 ? options.workers : std::size_t{0}); + RunExecutor executor(options.workers > 1 ? options.workers - 1 : std::size_t{0}); return Walk(fs, roots, options, executor, visit, on_error); } @@ -613,7 +621,7 @@ absl::Status Walk( Visitor visit, WalkErrorFn on_error, ContainerMounter mount_container) { - RunExecutor executor(options.workers > 1 ? options.workers : std::size_t{0}); + RunExecutor executor(options.workers > 1 ? options.workers - 1 : std::size_t{0}); return Walk(fs, roots, options, executor, visit, on_error, mount_container); } diff --git a/xff/engine/walk.h b/xff/engine/walk.h index 54ba06e1e5c..622ac20411a 100644 --- a/xff/engine/walk.h +++ b/xff/engine/walk.h @@ -84,8 +84,9 @@ struct WalkOptions { SymlinkMode symlinks = SymlinkMode::kNever; // Sibling ordering within each directory (xff `--sort`); kNone is readdir order. SortOrder sort = SortOrder::kNone; - // Worker threads for directory read-ahead and broad eager-stat batches; `1` is the sequential - // walk. The visitor always runs on a single coordinator thread. A caller may + // Participants for directory read-ahead and broad eager-stat batches, including the calling + // coordinator; standalone walks start at most workers-1 background threads. `1` is sequential. + // The visitor always runs on the calling coordinator thread. A caller may // offload independent matching, but traversal controls and visitor ordering stay serial. std::size_t workers = 1; // Keep complete metadata by default for arbitrary visitors. A metadata-free visitor may diff --git a/xff/engine/walk_parallel_test.cc b/xff/engine/walk_parallel_test.cc index 0dc85f5b57f..d2d89ea65b6 100644 --- a/xff/engine/walk_parallel_test.cc +++ b/xff/engine/walk_parallel_test.cc @@ -26,10 +26,12 @@ namespace xff::engine { namespace { using ::mbo::testing::IsOk; +using ::testing::Contains; using ::testing::ElementsAre; using ::testing::Eq; using ::testing::Gt; using ::testing::IsEmpty; +using ::testing::Le; using ::testing::SizeIs; struct StatTrace { @@ -125,6 +127,21 @@ WalkLog Record(const WideFs& fs, const WalkOptions& options) { struct WalkParallelTest : ::testing::Test {}; +TEST_F(WalkParallelTest, EagerStatBudgetIncludesTheCallingCoordinator) { + const WideFs serial; + const auto expected = Record(serial, {.workers = 1}); + for (const std::size_t participants : {1, 2, 3, 10}) { + SCOPED_TRACE(participants); + const WideFs fs; + fs.trace->await_peer = participants > 1; + const auto actual = Record(fs, {.workers = participants}); + EXPECT_THAT(actual.visits, Eq(expected.visits)); + EXPECT_THAT(actual.errors, IsEmpty()); + EXPECT_THAT(fs.trace->threads, SizeIs(Le(participants))); + EXPECT_THAT(fs.trace->threads, Contains(std::this_thread::get_id())); + } +} + TEST_F(WalkParallelTest, EagerStatsOverlapAndKeepCoordinatorVisitOrderAndFields) { constexpr auto kFields = std::to_array({vfs::MetadataFields::kBasic, vfs::MetadataFields::kBirthTime}); for (const auto fields : kFields) { diff --git a/xff/engine/walk_test.cc b/xff/engine/walk_test.cc index 80b7032f556..4c5edea6e14 100644 --- a/xff/engine/walk_test.cc +++ b/xff/engine/walk_test.cc @@ -481,7 +481,7 @@ TEST_F(WalkFakeFsTest, StopWaitsForUnusedPrefetchBeforeReturningToSharedExecutor blocking_fs.release.Notify(); EXPECT_THAT(while_blocked, Eq(std::future_status::timeout)); EXPECT_THAT(walk.get(), IsOk()); - EXPECT_THAT(executor.worker_count(), Eq(3)); + EXPECT_THAT(executor.worker_count(), Eq(2)); // The walk's three participants include its caller. EXPECT_THAT(executor.Submit([] { return 42; }).Get(), Eq(42)); }