diff --git a/XFF.md b/XFF.md
index e3bcef0551a..9beb8ce3541 100644
--- a/XFF.md
+++ b/XFF.md
@@ -564,7 +564,7 @@ See also: [Configuration](#topic-config), [Archives](#topic-archive), [Output](#
- `-j N, -j=N, --jobs=N|all` - directory-read, eligible content-match, comparison and -exec workers _(global, find, xff, rg)_
- `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories using one command-owned, lazily started pool of up to `N-1` background workers plus the calling coordinator, reused for directory reads, required eager metadata, eligible content matching and file comparison. Comparison trees are coordinated sequentially within that same budget. The caller participates in directory reads, eligible content and comparison work. It reads an uncached current directory while workers read later siblings, and can claim a required queued read once. Directory read-ahead uses a bounded ordered window and a shared retained-storage allowance; oversized speculative listings are read on demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight filesystem call must finish. Workers reuse the same executor for independent work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core.
+ `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 directory reads, eligible content and comparison work. It reads an uncached current directory while workers read later siblings, and can claim a required queued read once. Directory read-ahead uses a bounded ordered window and a shared retained-storage allowance; oversized speculative listings are read on demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight filesystem call must finish. Workers reuse the same executor for independent work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches match in bounded parallel batches, with output kept in traversal order. A short timed sample estimates the remaining batch's work; admission accounts for cold startup, warm dispatch and pool growth. Completed batches do not justify starting workers for a small tail. Small 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 aa31c563182..79d2d4d90dc 100644
--- a/docs/benchmark-correctness-audit.md
+++ b/docs/benchmark-correctness-audit.md
@@ -388,13 +388,45 @@ Python worker's high floor. The forked launcher's inherited high-water mark must
be corrected or explicitly excluded before using those reported peaks to choose
arena sizes. This is separate from the worker scheduler experiment.
+### Remaining matcher-work experiment
+
+A regression against the caller-participation candidate processes 8,192 entries
+in sixteen-entry batches, then a 32-entry tail. The unchanged matcher starts one
+worker for that tail because already-completed work bypasses its cold probe.
+The regression fails before the change and passes after it; completed work cannot
+repay the new worker's cost on that tail.
+
+The separate follow-up removes the completed-entry counter. Four timed chunks
+evaluate up to sixteen entries exactly once; the middle two observations estimate
+only the remaining owned batch. A pure cost planner includes cold startup,
+dispatch, pool growth and minimum saving, retains caller-inclusive budgeting,
+and limits participants by both cost and independent remaining entries. Warm
+reuse is not automatically beneficial. Adaptive atomic-claim grain aims to
+amortize scheduling while retaining several balancing waves. Workers remain
+leaves, ordered results and private evaluator state remain unchanged, and tiny
+batches never require a timing observation or background dispatch.
+
+The allowances are experimental, not portable measured constants: 3.5 ms cold,
+50 us per added worker/dispatch, 100 us saving and a 32 us chunk target. The
+default Linux/x86 executor diagnostic in PR #1000 informs the cold cliff and
+the need to distinguish cold/warm/grain; it is not a file-cost calibration or
+native M5 Pro validation. Controlled slow evaluator tests exercise actual caller
+participation, exact-once results and worker reuse without touching a filesystem;
+they are not physical slow-storage acceptance. A trimmed small prefix still
+cannot prove uniform costs in an arbitrary batch. Directory/stat policies,
+candidate allocation profiles, repeated/reordered actual-workload timing and
+native cross-platform controls remain required. No performance acceptance is
+claimed by this implementation or its correctness tests.
+
The complete audit remains open. Final shared-executor timing must be collected
after every build has finished, using the complete prepared grid. Repeated
controls must cover content, eager metadata and slow storage. Separate evidence
in [PR #995](https://github.com/mboworks/xff/pull/995) now retains the exact-source
Linux baseline, enumeration-only allocation profiles and coarse counters/stacks
-for 20k/50k/100k single-worker workloads. Those baseline diagnostics do not complete
-precise phase-level attribution or allocation controls for the other workloads,
-platforms or new scheduler candidates. Hosted macOS checks provide portability evidence for PR #991;
+for 20k/50k/100k single-worker workloads. Its later allocation controls also cover
+tree output, collection, lazy/eager metadata and content matching for both layouts
+at those sizes, with separate native RSS and exact-output checks. Those baseline
+diagnostics do not complete precise phase-level attribution, multiworker/candidate
+allocation controls or native cross-platform acceptance. Hosted macOS checks provide portability evidence for PR #991;
they do not replace the specified native M5 Pro A/B and sampling controls for
the shared-executor experiment.
diff --git a/docs/design-parallel.md b/docs/design-parallel.md
index 335c46e0370..65748695880 100644
--- a/docs/design-parallel.md
+++ b/docs/design-parallel.md
@@ -56,10 +56,10 @@ background workers under `--jobs=N`:
- The coordinator applies traversal controls and writes the output sink in traversal order.
Independent content predicates and rg content searches reuse that executor, with the caller also
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
+- Content matching schedules adaptive chunks of owned entries, so clustered expensive files can use
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
+ six run inline even after workers exist. Later sufficiently costly 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
@@ -212,12 +212,51 @@ disappears, leaving the executor reusable by the next phase.
Content matching retains private evaluator and regex state for each logical
worker. Decision-only matching collects up to 8,192 entries before flushing;
rendered-content batches remain bounded at 256 entries because their output can
-be large. A cold executor keeps cheap small batches on the coordinator until
-8,192 entries have accumulated. A timed 16-entry prefix admits slower work
-earlier when that prefix takes at least 500 microseconds. Once traversal or
-matching has started the executor, batches of at least 16 entries may reuse it.
-Each submitted matcher job drains entry chunks, and results are consumed in
-original entry order.
+be large. Matcher admission now estimates only the owned batch's remaining
+work; completed batches do not justify starting workers for a small tail.
+Batches below six entries stay on the caller. Larger batches evaluate up to sixteen
+entries as four timed chunks of one to four entries, retain those results once, and use
+the middle two elapsed observations for a remaining-work estimate. Discarding
+the fastest/slowest chunk reduces sensitivity to one already-paid cold/outlier
+entry; it does not prove that the rest of a heterogeneous batch has equal cost.
+
+The pure planner compares remaining serial cost against predicted caller-inclusive
+parallel cost. It accounts separately for cold startup, waking the executor,
+new worker creation and a minimum saving allowance. Already-started workers are
+reused, but cheap warm batches may still remain serial. Worker growth must improve
+the estimate; the full requested budget is not automatically activated. The count
+of independent remaining items limits participants, but even a few expensive
+items may repay dispatch; there is no minimum of sixteen items per worker. A convex
+cost model permits checking only boundaries and the two integers around its
+continuous minimum, avoiding a linear search over a large requested worker count.
+
+The experimental default allowances are 3.5 ms cold startup, 50 us per added
+background worker, 50 us dispatch and 100 us minimum predicted saving. These are
+planning allowances, not portable measurements or worst-case guarantees. The cold
+allowance includes the separately observed default Linux/x86 Abseil calibration
+cliff; native M5 Pro and actual matcher setup/dispatch need independent validation.
+No runtime flag selects these testable internal policies, and neither synthetic
+executor controls nor controlled slow evaluators constitute storage acceptance.
+
+After admission, each background worker gets one drain and the caller also drains.
+Atomic-claim grain targets 32 us of estimated useful work, capped at 128 entries
+and, where possible, at a quarter of each participant's average share to retain balancing waves.
+Slow entries reduce the grain toward one. Workers never recursively submit or
+wait on their own executor; results are still consumed in original entry order.
+These drains are bounded by the matcher batch, not an entire tree. They do not
+make queue fairness preemptive or interrupt a running filesystem call. Directory
+and eager-stat admission remain separate inherited policies, requiring their own
+remaining-cost, metadata and slow-storage controls before acceptance.
+
+This candidate forecasts only the owned matcher batch. It does not establish
+eligible work in an unvisited frontier or credit future batches merely because
+earlier ones were expensive. Cheap bounded batches can therefore stay serial
+even when a larger command has more work ahead. Command-wide known-frontier cost
+planning and heterogeneous/cached workload controls remain part of the audit;
+this matcher experiment is not a completed adaptive command scheduler.
+Private matcher/evaluator setup and contention with other phases are not
+individually calibrated by this cost model; a warm shared executor does not imply
+that a newly encountered matcher's private state is already prepared.
Under `-j > 1` the serial `-exec ... ;` / `-execdir ... ;` form launches its child
on a bounded runner (at most `N` outstanding) instead of running it synchronously.
diff --git a/xff/cli/globals.cc b/xff/cli/globals.cc
index 4d6c560356e..a9489a3f159 100644
--- a/xff/cli/globals.cc
+++ b/xff/cli/globals.cc
@@ -915,7 +915,9 @@ constexpr std::array kGlobals = std::to_array({
"filesystem call must finish. Workers reuse the same executor for independent work; small workloads "
"stay on the coordinator. Lazy and metadata-free listings do not start "
"stat workers. Independent content predicates and rg searches match in bounded parallel batches, "
- "with output kept in traversal order. Small initial batches, archive members, stateful expressions "
+ "with output kept in traversal order. A short timed sample estimates the remaining batch's work; "
+ "admission accounts for cold startup, warm dispatch and pool growth. Completed batches do not "
+ "justify starting workers for a small tail. Small 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 "
diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel
index 9e923681db1..639c39452b5 100644
--- a/xff/engine/BUILD.bazel
+++ b/xff/engine/BUILD.bazel
@@ -272,6 +272,22 @@ cc_fuzz_test(
],
)
+cc_library(
+ name = "match_work_cc",
+ srcs = ["match_work.cc"],
+ hdrs = ["match_work.h"],
+)
+
+cc_test(
+ name = "match_work_test",
+ size = "small",
+ srcs = ["match_work_test.cc"],
+ deps = [
+ ":match_work_cc",
+ "@googletest//:gtest_main",
+ ],
+)
+
# Drives one find run: walk the roots, evaluate the expression, (implicit) print.
cc_library(
name = "parallel_match_cc",
@@ -281,6 +297,7 @@ cc_library(
":collect_cc",
":evaluate_cc",
":expression_execution_cc",
+ ":match_work_cc",
":run_executor_cc",
"//xff/parser:parser_cc",
"//xff/registry:registry_cc",
@@ -299,10 +316,14 @@ cc_test(
deps = [
":evaluate_cc",
":expression_execution_cc",
+ ":match_work_cc",
":parallel_match_cc",
"//xff/matching/regex:regex_cc",
"//xff/parser:parser_cc",
"//xff/vfs:local_fs_cc",
+ "@abseil-cpp//absl/base:core_headers",
+ "@abseil-cpp//absl/synchronization",
+ "@abseil-cpp//absl/time",
"@googletest//:gtest_main",
"@mboworks_mbo//mbo/testing:status_cc",
],
@@ -600,7 +621,10 @@ cc_test(
"//xff/parser:parser_cc",
"//xff/vfs:local_fs_cc",
"//xff/vfs:vfs_cc",
+ "@abseil-cpp//absl/base:core_headers",
"@abseil-cpp//absl/status",
+ "@abseil-cpp//absl/synchronization",
+ "@abseil-cpp//absl/time",
"@googletest//:gtest",
"@googletest//:gtest_main",
"@mboworks_mbo//mbo/testing:matchers_cc",
diff --git a/xff/engine/command_workers_test.cc b/xff/engine/command_workers_test.cc
index 6e38d0884b4..888ba6e42a9 100644
--- a/xff/engine/command_workers_test.cc
+++ b/xff/engine/command_workers_test.cc
@@ -16,6 +16,7 @@
#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
#include "absl/time/time.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
@@ -42,12 +43,17 @@ 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.
+// No host filesystem enters these controls. The matcher control supplies uniform
+// simulated latency and keeps its rendezvous outside the caller's admission sample.
class CommandFs final : public vfs::FileSystem {
public:
- explicit CommandFs(std::size_t participants, std::size_t content_participants = 0)
+ explicit CommandFs(
+ std::size_t participants,
+ std::size_t content_participants = 0,
+ absl::Duration content_delay = absl::ZeroDuration())
: participants_(participants),
- content_participants_(content_participants == 0 ? participants : content_participants) {}
+ content_participants_(content_participants == 0 ? participants : content_participants),
+ content_delay_(content_delay) {}
absl::StatusOr> ReadDir(std::string_view path) const override {
if (!IsDirectory(path)) {
@@ -96,6 +102,7 @@ class CommandFs final : public vfs::FileSystem {
return absl::NotFoundError(path);
}
Record(2, content_participants_);
+ absl::SleepFor(content_delay_);
return vfs::MemoryReadSource(bytes_);
}
@@ -128,6 +135,9 @@ class CommandFs final : public vfs::FileSystem {
EXPECT_THAT(synchronized_, IsTrue());
EXPECT_THAT(phase_readers_.at(2), SizeIs(Ge(content_participants_)));
EXPECT_THAT(phase_readers_.at(2), Contains(coordinator_));
+ if (content_delay_ > absl::ZeroDuration()) {
+ EXPECT_THAT(content_reads_, Eq(64));
+ }
}
void CheckWalkReuse() const {
@@ -150,7 +160,12 @@ class CommandFs final : public vfs::FileSystem {
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) {
+ const bool first = phase_readers_.at(phase).insert(std::this_thread::get_id()).second;
+ const bool sampled_content = phase == 2 && content_delay_ > absl::ZeroDuration();
+ if (sampled_content && ++content_reads_ <= 16) {
+ return; // Sampling cannot rendezvous with workers not yet admitted.
+ }
+ if (!first && !sampled_content) {
return;
}
// First calls rendezvous so one fast worker cannot drain the whole fixture before
@@ -164,13 +179,16 @@ class CommandFs final : public vfs::FileSystem {
const std::size_t participants_;
const std::size_t content_participants_;
+ const absl::Duration content_delay_;
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.
+ // Guards reader identities, content observation count and rendezvous outcomes.
+ // Filesystem data/configuration is immutable; no lock spans simulated latency.
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;
+ mutable std::size_t content_reads_ ABSL_GUARDED_BY(mutex_) = 0;
};
struct CommandWorkersTest : ::testing::Test {
@@ -219,7 +237,7 @@ TEST_F(CommandWorkersTest, ContentMatchingIncludesTheCallingCoordinatorInTheAllo
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}));
+ const CommandFs fs(participants, std::min(participants, std::size_t{2}), absl::Milliseconds(2));
ASSERT_OK_AND_ASSIGN(
const auto output,
Run({"--jobs=" + std::to_string(participants), "--sort=tree", "left", "-content", "needle"}, fs));
diff --git a/xff/engine/match_work.cc b/xff/engine/match_work.cc
new file mode 100644
index 00000000000..244ac4cae66
--- /dev/null
+++ b/xff/engine/match_work.cc
@@ -0,0 +1,95 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "xff/engine/match_work.h"
+
+#include
+#include
+#include
+#include
+#include
+#include
+
+namespace xff::engine {
+namespace {
+
+constexpr std::size_t kMaxGrain = 128;
+constexpr std::size_t kBalancingWaves = 4;
+
+} // namespace
+
+MatchWorkEstimate EstimateMatchWork(std::array elapsed, std::size_t chunk_entries) {
+ if (chunk_entries == 0 || chunk_entries > std::numeric_limits::max() / 2
+ || std::ranges::any_of(elapsed, [](auto duration) { return duration < std::chrono::nanoseconds::zero(); })) {
+ return {};
+ }
+ std::ranges::sort(elapsed);
+ const auto sampled = elapsed.at(1) > std::chrono::nanoseconds::max() - elapsed.at(2) ? std::chrono::nanoseconds::max()
+ : elapsed.at(1) + elapsed.at(2);
+ return {.sampled_entries = 2 * chunk_entries, .sampled_elapsed = sampled};
+}
+
+MatchWorkPlan PlanMatchWork(
+ std::size_t remaining_entries,
+ std::size_t participant_limit,
+ std::size_t started_workers,
+ MatchWorkEstimate estimate,
+ MatchWorkCosts costs) {
+ const auto zero = std::chrono::nanoseconds::zero();
+ const std::size_t capacity = std::min(participant_limit, remaining_entries);
+ if (capacity < 2 || estimate.sampled_entries == 0 || estimate.sampled_elapsed <= zero || costs.cold_start < zero
+ || costs.worker_start < zero || costs.dispatch < zero || costs.target_chunk < zero
+ || costs.minimum_saving < zero) {
+ return {};
+ }
+ // Floating nanoseconds avoid integral overflow when extrapolating a slow sample
+ // over a large remaining batch. Counts are clamped before conversion back.
+ const double per_entry =
+ static_cast(estimate.sampled_elapsed.count()) / static_cast(estimate.sampled_entries);
+ const double serial = per_entry * static_cast(remaining_entries);
+ const double cold = started_workers == 0 ? static_cast(costs.cold_start.count()) : 0;
+ const auto start = static_cast(costs.worker_start.count());
+ const auto dispatch = static_cast(costs.dispatch.count());
+ MatchWorkPlan best;
+ double best_cost = serial;
+ const auto consider = [&](std::size_t participants) {
+ const std::size_t background = participants - 1;
+ const std::size_t added = background > started_workers ? background - started_workers : 0;
+ const double predicted =
+ (serial / static_cast(participants)) + cold + dispatch + (static_cast(added) * start);
+ if (predicted < best_cost || (predicted == best_cost && participants < best.participants)) {
+ best.participants = participants;
+ best_cost = predicted;
+ }
+ };
+ // Among already started workers, the model improves monotonically. Beyond
+ // that boundary, serial/P + P*start is convex with its minimum at sqrt(serial/start).
+ // Inspect only boundaries and neighboring integers, even for huge input counts.
+ consider(2);
+ consider(capacity);
+ consider(started_workers >= capacity - 1 ? capacity : std::max(std::size_t{2}, started_workers + 1));
+ if (start > 0) {
+ const double optimum = std::sqrt(serial / start);
+ const std::size_t lower = optimum >= static_cast(capacity) ? capacity
+ : optimum <= 2 ? std::size_t{2}
+ : static_cast(optimum);
+ consider(lower);
+ if (lower < capacity) {
+ consider(lower + 1);
+ }
+ }
+ if (best.participants == 1 || serial - best_cost < static_cast(costs.minimum_saving.count())) {
+ return {};
+ }
+ // Aim for useful work between atomic claims, but preserve multiple scheduling
+ // waves for uneven entries. A slow item naturally reduces the grain toward one.
+ const std::size_t grain_limit =
+ std::max(std::size_t{1}, std::min(kMaxGrain, remaining_entries / best.participants / kBalancingWaves));
+ const double desired = std::ceil(static_cast(costs.target_chunk.count()) / per_entry);
+ best.grain = desired >= static_cast(grain_limit) ? grain_limit
+ : desired <= 1 ? std::size_t{1}
+ : static_cast(desired);
+ return best;
+}
+
+} // namespace xff::engine
diff --git a/xff/engine/match_work.h b/xff/engine/match_work.h
new file mode 100644
index 00000000000..307aafb95be
--- /dev/null
+++ b/xff/engine/match_work.h
@@ -0,0 +1,52 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#ifndef XFF_ENGINE_MATCH_WORK_H_
+#define XFF_ENGINE_MATCH_WORK_H_
+
+#include
+#include
+#include
+
+namespace xff::engine {
+
+struct MatchWorkEstimate {
+ std::size_t sampled_entries = 0;
+ std::chrono::nanoseconds sampled_elapsed = {};
+};
+
+// All four chunks were evaluated once. Use the middle two elapsed observations
+// to avoid letting one already-completed cold/outlier chunk justify a cheap tail.
+// Invalid dimensions/times return an empty estimate; duration addition saturates.
+MatchWorkEstimate EstimateMatchWork(std::array elapsed, std::size_t chunk_entries);
+
+// Experimental planning allowances, not portable measurements or hard latency bounds.
+// The cold allowance covers the observed default Linux/x86 calibration cliff; native
+// macOS and actual matcher setup/dispatch still require acceptance measurements.
+struct MatchWorkCosts {
+ std::chrono::nanoseconds cold_start = std::chrono::microseconds(3'500);
+ std::chrono::nanoseconds worker_start = std::chrono::microseconds(50);
+ std::chrono::nanoseconds dispatch = std::chrono::microseconds(50);
+ std::chrono::nanoseconds target_chunk = std::chrono::microseconds(32);
+ std::chrono::nanoseconds minimum_saving = std::chrono::microseconds(100);
+};
+
+struct MatchWorkPlan {
+ std::size_t participants = 1; // Includes the calling coordinator.
+ std::size_t grain = 4;
+};
+
+// Plan only the work still available after sampling. Existing background threads
+// are reusable, but waking them and growing the pool are not free. Invalid or
+// insufficient observations fall back to the caller. There is no completed-work
+// counter and no assumption of independent work beyond this owned batch.
+MatchWorkPlan PlanMatchWork(
+ std::size_t remaining_entries,
+ std::size_t participant_limit,
+ std::size_t started_workers,
+ MatchWorkEstimate estimate,
+ MatchWorkCosts costs = {});
+
+} // namespace xff::engine
+
+#endif // XFF_ENGINE_MATCH_WORK_H_
diff --git a/xff/engine/match_work_test.cc b/xff/engine/match_work_test.cc
new file mode 100644
index 00000000000..906878f22b5
--- /dev/null
+++ b/xff/engine/match_work_test.cc
@@ -0,0 +1,191 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "xff/engine/match_work.h"
+
+#include
+#include
+#include
+#include
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace xff::engine {
+namespace {
+
+using ::std::chrono::microseconds;
+using ::std::chrono::nanoseconds;
+using ::testing::AllOf;
+using ::testing::Eq;
+using ::testing::FieldsAre;
+using ::testing::Ge;
+using ::testing::Le;
+using ::testing::Message;
+
+struct MatchWorkTest : ::testing::Test {};
+
+TEST_F(MatchWorkTest, OneSlowCompletedChunkDoesNotInflateRemainingEstimate) {
+ EXPECT_THAT(
+ EstimateMatchWork({nanoseconds(400), microseconds(80'000), nanoseconds(800), nanoseconds(400)}, 4),
+ FieldsAre(8, nanoseconds(1'200)));
+ EXPECT_THAT(
+ EstimateMatchWork({nanoseconds(400), nanoseconds(800), nanoseconds(800), nanoseconds(400)}, 4),
+ FieldsAre(8, nanoseconds(1'200)));
+}
+
+TEST_F(MatchWorkTest, InvalidSampleChunksAndSaturatingDurationAdditionAreSafe) {
+ EXPECT_THAT(EstimateMatchWork({}, 0), FieldsAre(0, nanoseconds::zero()));
+ EXPECT_THAT(EstimateMatchWork({}, std::numeric_limits::max()), FieldsAre(0, nanoseconds::zero()));
+ EXPECT_THAT(EstimateMatchWork({nanoseconds(-1), {}, {}, {}}, 4), FieldsAre(0, nanoseconds::zero()));
+ EXPECT_THAT(
+ EstimateMatchWork({nanoseconds::max(), nanoseconds::max(), nanoseconds::max(), nanoseconds::max()}, 4),
+ FieldsAre(8, nanoseconds::max()));
+}
+
+TEST_F(MatchWorkTest, NoIndependentRemainingWorkStaysOnCallerEvenWithWarmPool) {
+ const MatchWorkEstimate estimate{.sampled_entries = 16, .sampled_elapsed = microseconds(16'000)};
+ constexpr std::array kRemaining{std::size_t{0}, std::size_t{1}};
+ for (const auto remaining : kRemaining) {
+ EXPECT_THAT(PlanMatchWork(remaining, 10, 0, estimate), FieldsAre(1, 4));
+ EXPECT_THAT(PlanMatchWork(remaining, 10, 9, estimate), FieldsAre(1, 4));
+ }
+}
+
+TEST_F(MatchWorkTest, FewExpensiveRemainingItemsCanRepayColdStartup) {
+ EXPECT_THAT(
+ PlanMatchWork(2, 10, 0, {.sampled_entries = 16, .sampled_elapsed = microseconds(160'000)}), FieldsAre(2, 1));
+ EXPECT_THAT(
+ PlanMatchWork(2, 10, 0, {.sampled_entries = 16, .sampled_elapsed = microseconds(16'000)}), FieldsAre(1, 4));
+ EXPECT_THAT(
+ PlanMatchWork(2, 10, 9, {.sampled_entries = 16, .sampled_elapsed = microseconds(16'000)}), FieldsAre(2, 1));
+}
+
+TEST_F(MatchWorkTest, ColdStartupCanOutweighUsefulWarmWork) {
+ const MatchWorkEstimate estimate{.sampled_entries = 16, .sampled_elapsed = microseconds(160)};
+ EXPECT_THAT(PlanMatchWork(256, 10, 0, estimate), FieldsAre(1, 4));
+ EXPECT_THAT(PlanMatchWork(256, 3, 2, estimate), FieldsAre(3, 4));
+ EXPECT_THAT(PlanMatchWork(256, 10, 9, estimate), FieldsAre(10, 4));
+}
+
+TEST_F(MatchWorkTest, EnoughRemainingCostRepaysColdStartupAndHonorsParticipantCap) {
+ const MatchWorkEstimate estimate{.sampled_entries = 16, .sampled_elapsed = microseconds(160)};
+ EXPECT_THAT(PlanMatchWork(1'024, 3, 0, estimate), FieldsAre(3, 4));
+ EXPECT_THAT(PlanMatchWork(1'024, 10, 0, estimate), FieldsAre(10, 4));
+ EXPECT_THAT(PlanMatchWork(1'024, 1, 0, estimate), FieldsAre(1, 4));
+ EXPECT_THAT(PlanMatchWork(1'024, 0, 0, estimate), FieldsAre(1, 4));
+}
+
+TEST_F(MatchWorkTest, ReusingThreadsStillMustRepayDispatchAndSavingMargin) {
+ const MatchWorkEstimate estimate{.sampled_entries = 16, .sampled_elapsed = microseconds(1)};
+ EXPECT_THAT(PlanMatchWork(1'024, 10, 9, estimate), FieldsAre(1, 4));
+}
+
+TEST_F(MatchWorkTest, NewWorkersNeedEnoughAdditionalBenefit) {
+ const MatchWorkEstimate estimate{.sampled_entries = 16, .sampled_elapsed = microseconds(16)};
+ EXPECT_THAT(PlanMatchWork(256, 10, 2, estimate).participants, Eq(3));
+ const MatchWorkEstimate slower{.sampled_entries = 16, .sampled_elapsed = microseconds(160)};
+ EXPECT_THAT(PlanMatchWork(256, 10, 1, slower).participants, Eq(7));
+}
+
+TEST_F(MatchWorkTest, GrainReflectsPerEntryCostAndPreservesBalancingWaves) {
+ const MatchWorkCosts free_dispatch{
+ .cold_start = {},
+ .worker_start = {},
+ .dispatch = {},
+ .target_chunk = microseconds(32),
+ .minimum_saving = {},
+ };
+ EXPECT_THAT(
+ PlanMatchWork(8'192, 3, 0, {.sampled_entries = 16, .sampled_elapsed = microseconds(1)}, free_dispatch),
+ FieldsAre(3, 128));
+ EXPECT_THAT(
+ PlanMatchWork(32, 3, 0, {.sampled_entries = 16, .sampled_elapsed = microseconds(1)}, free_dispatch),
+ FieldsAre(3, 2));
+ EXPECT_THAT(
+ PlanMatchWork(512, 3, 0, {.sampled_entries = 16, .sampled_elapsed = microseconds(1'600)}, free_dispatch),
+ FieldsAre(3, 1));
+}
+
+TEST_F(MatchWorkTest, SavingMarginBoundaryUsesRemainingCost) {
+ MatchWorkCosts costs{
+ .cold_start = {},
+ .worker_start = {},
+ .dispatch = microseconds(500),
+ .target_chunk = microseconds(32),
+ .minimum_saving = microseconds(300),
+ };
+ const MatchWorkEstimate estimate{.sampled_entries = 16, .sampled_elapsed = microseconds(800)};
+ EXPECT_THAT(PlanMatchWork(32, 2, 0, estimate, costs), FieldsAre(2, 1));
+ costs.dispatch += nanoseconds(1);
+ EXPECT_THAT(PlanMatchWork(32, 2, 0, estimate, costs), FieldsAre(1, 4));
+}
+
+TEST_F(MatchWorkTest, InvalidObservationsAndCostsFallBackToCaller) {
+ const MatchWorkEstimate valid{.sampled_entries = 16, .sampled_elapsed = microseconds(160)};
+ EXPECT_THAT(PlanMatchWork(1'024, 3, 0, {}), FieldsAre(1, 4));
+ EXPECT_THAT(PlanMatchWork(1'024, 3, 0, {.sampled_elapsed = microseconds(160)}), FieldsAre(1, 4));
+ EXPECT_THAT(PlanMatchWork(1'024, 3, 0, {.sampled_entries = 16}), FieldsAre(1, 4));
+ EXPECT_THAT(PlanMatchWork(1'024, 3, 0, {.sampled_entries = 16, .sampled_elapsed = nanoseconds(-1)}), FieldsAre(1, 4));
+ const std::array invalid{
+ MatchWorkCosts{.cold_start = nanoseconds(-1)}, MatchWorkCosts{.worker_start = nanoseconds(-1)},
+ MatchWorkCosts{.dispatch = nanoseconds(-1)}, MatchWorkCosts{.target_chunk = nanoseconds(-1)},
+ MatchWorkCosts{.minimum_saving = nanoseconds(-1)},
+ };
+ for (const auto costs : invalid) {
+ EXPECT_THAT(PlanMatchWork(1'024, 3, 0, valid, costs), FieldsAre(1, 4));
+ }
+}
+
+TEST_F(MatchWorkTest, MaximumCountsDoNotOverflowDurationOrSearchWorkerCountsLinearly) {
+ constexpr auto kMaximum = std::numeric_limits::max();
+ const auto plan =
+ PlanMatchWork(kMaximum, kMaximum, kMaximum, {.sampled_entries = 1, .sampled_elapsed = nanoseconds::max()});
+ EXPECT_THAT(plan.participants, Eq(kMaximum));
+ EXPECT_THAT(plan.grain, Eq(1));
+ const auto cold = PlanMatchWork(kMaximum, kMaximum, 0, {.sampled_entries = 1, .sampled_elapsed = nanoseconds::max()});
+ EXPECT_THAT(cold.participants, AllOf(Ge(2), Le(kMaximum)));
+ EXPECT_THAT(cold.grain, AllOf(Ge(1), Le(128)));
+}
+
+TEST_F(MatchWorkTest, ConstantTimeSearchAgreesWithExhaustiveBoundedCostModel) {
+ const MatchWorkCosts costs;
+ constexpr std::array kRemaining{2UZ, 4UZ, 16UZ, 256UZ, 8'192UZ};
+ constexpr std::array kLimits{2UZ, 3UZ, 10UZ};
+ constexpr std::array kStarted{0UZ, 1UZ, 9UZ};
+ constexpr std::array kElapsed{microseconds(1), microseconds(160), microseconds(1'600)};
+ for (const auto remaining : kRemaining) {
+ for (const auto limit : kLimits) {
+ for (const auto started : kStarted) {
+ for (const auto elapsed : kElapsed) {
+ SCOPED_TRACE(Message() << remaining << ":" << limit << ":" << started << ":" << elapsed.count());
+ const double serial = static_cast(nanoseconds(elapsed).count()) * static_cast(remaining) / 16;
+ double best_cost = serial;
+ std::size_t expected = 1;
+ const std::size_t capacity = remaining < limit ? remaining : limit;
+ for (std::size_t participants = 2; participants <= capacity; ++participants) {
+ const std::size_t added = participants - 1 > started ? participants - 1 - started : 0;
+ const double predicted = (serial / static_cast(participants))
+ + static_cast(started == 0 ? costs.cold_start.count() : 0)
+ + static_cast(costs.dispatch.count())
+ + (static_cast(added) * static_cast(costs.worker_start.count()));
+ if (predicted < best_cost) {
+ expected = participants;
+ best_cost = predicted;
+ }
+ }
+ if (serial - best_cost < static_cast(costs.minimum_saving.count())) {
+ expected = 1;
+ }
+ EXPECT_THAT(
+ PlanMatchWork(remaining, limit, started, {.sampled_entries = 16, .sampled_elapsed = elapsed})
+ .participants,
+ Eq(expected));
+ }
+ }
+ }
+ }
+}
+
+} // namespace
+} // namespace xff::engine
diff --git a/xff/engine/parallel_match.cc b/xff/engine/parallel_match.cc
index 612af3428f8..2fdf4af9f47 100644
--- a/xff/engine/parallel_match.cc
+++ b/xff/engine/parallel_match.cc
@@ -4,6 +4,7 @@
#include "xff/engine/parallel_match.h"
#include
+#include
#include
#include
#include
@@ -15,11 +16,9 @@
namespace xff::engine {
namespace {
-// A cold executor has a fixed creation and teardown cost. Accumulate enough matcher work to
-// repay it; once another run phase has started the executor, ordinary per-batch admission applies.
-inline constexpr std::size_t kMinColdParallelEntries = 8'192;
-inline constexpr std::size_t kColdProbeEntries = 16;
-inline constexpr auto kSlowColdProbe = std::chrono::microseconds(500);
+inline constexpr std::size_t kProbeChunks = 4;
+inline constexpr std::size_t kProbeChunkEntries = 4;
+inline constexpr std::size_t kMinRemainingEntries = 2;
} // namespace
@@ -61,13 +60,15 @@ ParallelMatch::ParallelMatch(
RunExecutor& executor,
bool scores,
std::optional output,
- mbo::types::OptionalRef execution)
+ mbo::types::OptionalRef execution,
+ MatchWorkCosts costs)
: expression_(expression),
workers_(std::max(workers, std::size_t{1})),
executor_(executor),
scores_(scores),
output_(std::move(output)),
- execution_(execution) {}
+ execution_(execution),
+ costs_(costs) {}
ParallelMatch::~ParallelMatch() = default;
@@ -76,7 +77,7 @@ const std::vector& ParallelMatch::Match(std::vectorMakeWorker(ExpressionWorkerRole::kCoordinator));
@@ -89,42 +90,42 @@ const std::vector& ParallelMatch::Match(std::vector{});
};
// Very small batches cannot amortize waking the executor.
- if (workers_ == 1 || entries_.size() < 16) {
+ if (workers_ == 1 || entries_.size() < kProbeChunks + kMinRemainingEntries) {
ensure_coordinator();
evaluate_coordinator();
- serial_entries_ += entries_.size();
return results_;
}
- std::size_t parallel_entries = entries_.size();
- if (cold && serial_entries_ + entries_.size() < kMinColdParallelEntries) {
- // A count threshold protects fast local storage, but must not suppress useful concurrency on
- // slow filesystems. Measure a small prefix, then either parallelize the rest of this batch or
- // finish inline and retain the evidence for later bounded batches.
- ensure_coordinator();
- const std::size_t probe = std::min(kColdProbeEntries, entries_.size());
+ ensure_coordinator();
+ // Observe this batch, even with an already warm executor. Discard the fastest
+ // and slowest of four timed chunks so one cold/outlier entry does not by itself
+ // justify dispatching a cheap tail. Every sampled result is retained exactly once.
+ std::array timings{};
+ const std::size_t chunk_entries =
+ std::min(kProbeChunkEntries, (entries_.size() - kMinRemainingEntries) / kProbeChunks);
+ const std::size_t probe_entries = kProbeChunks * chunk_entries;
+ for (std::size_t chunk = 0; chunk < kProbeChunks; ++chunk) {
const auto started = std::chrono::steady_clock::now();
- for (std::size_t index = 0; index < probe; ++index) {
+ const std::size_t first = chunk * chunk_entries;
+ for (std::size_t index = first; index < first + chunk_entries; ++index) {
results_.at(index) = EvaluateEntry(
entries_.at(index).AsVisit(), {}, {},
coordinator_ ? mbo::types::OptionalRef{*coordinator_}
: mbo::types::OptionalRef{});
}
- next_.store(probe, std::memory_order_relaxed);
- if (std::chrono::steady_clock::now() - started < kSlowColdProbe) {
- evaluate_coordinator();
- serial_entries_ += entries_.size();
- return results_;
- }
- parallel_entries -= probe;
+ timings.at(chunk) =
+ std::chrono::duration_cast(std::chrono::steady_clock::now() - started);
}
- const std::size_t participants = std::min(workers_, (parallel_entries + 15) / 16);
- if (participants <= 1) {
- ensure_coordinator();
+ next_.store(probe_entries, std::memory_order_relaxed);
+ const auto plan = PlanMatchWork(
+ entries_.size() - probe_entries, workers_, executor_.worker_count(), EstimateMatchWork(timings, chunk_entries),
+ costs_);
+ if (plan.participants == 1) {
evaluate_coordinator();
return results_;
}
- executor_.Start(participants - 1);
- const std::size_t count = std::min(participants - 1, executor_.worker_count());
+ grain_ = plan.grain;
+ executor_.Start(plan.participants - 1);
+ const std::size_t count = std::min(plan.participants - 1, executor_.worker_count());
while (worker_states_.size() < count) {
worker_states_.push_back(std::make_unique(*this));
}
@@ -157,14 +158,13 @@ void ParallelMatch::EvaluateEntries(
mbo::types::OptionalRef> output,
mbo::types::OptionalRef matchers,
mbo::types::OptionalRef evaluator) {
- // Small chunks spread clustered expensive entries while amortizing atomic scheduling.
- constexpr std::size_t kChunk = 4;
+ // Planned chunks amortize atomic scheduling while retaining multiple balancing waves.
for (;;) {
- const std::size_t first = next_.fetch_add(kChunk, std::memory_order_relaxed);
+ const std::size_t first = next_.fetch_add(grain_, std::memory_order_relaxed);
if (first >= entries_.size()) {
return;
}
- const std::size_t end = std::min(first + kChunk, entries_.size());
+ const std::size_t end = std::min(first + grain_, entries_.size());
for (std::size_t index = first; index < end; ++index) {
results_.at(index) = EvaluateEntry(entries_.at(index).AsVisit(), output, matchers, evaluator);
}
diff --git a/xff/engine/parallel_match.h b/xff/engine/parallel_match.h
index 13a3197164c..3b26f3961f2 100644
--- a/xff/engine/parallel_match.h
+++ b/xff/engine/parallel_match.h
@@ -17,6 +17,7 @@
#include "xff/engine/collect.h"
#include "xff/engine/evaluate.h"
#include "xff/engine/expression_execution.h"
+#include "xff/engine/match_work.h"
#include "xff/engine/run_executor.h"
#include "xff/parser/ast.h"
@@ -57,7 +58,8 @@ class ParallelMatch final {
RunExecutor& executor,
bool scores,
std::optional output = std::nullopt,
- mbo::types::OptionalRef execution = {});
+ mbo::types::OptionalRef execution = {},
+ MatchWorkCosts costs = {});
~ParallelMatch();
ParallelMatch(const ParallelMatch&) = delete;
ParallelMatch& operator=(const ParallelMatch&) = delete;
@@ -89,12 +91,14 @@ class ParallelMatch final {
const bool scores_;
const std::optional output_;
const mbo::types::OptionalRef execution_;
+ const MatchWorkCosts costs_;
std::optional coordinator_;
std::vector> worker_states_;
std::vector entries_;
std::vector results_;
std::atomic next_ = 0;
- std::size_t serial_entries_ = 0;
+ // Published before dispatch and unchanged until all jobs join.
+ std::size_t grain_ = 4;
};
} // namespace xff::engine
diff --git a/xff/engine/parallel_match_test.cc b/xff/engine/parallel_match_test.cc
index 71df891c414..fc6ec29046b 100644
--- a/xff/engine/parallel_match_test.cc
+++ b/xff/engine/parallel_match_test.cc
@@ -3,12 +3,19 @@
#include "xff/engine/parallel_match.h"
+#include
#include
#include
#include
+#include
#include
+#include
#include
+#include "absl/base/thread_annotations.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/time/clock.h"
+#include "absl/time/time.h"
#include "gmock/gmock.h"
#include "gtest/gtest.h"
#include "mbo/testing/status.h"
@@ -21,12 +28,67 @@
namespace xff::engine {
namespace {
+using ::testing::Contains;
+using ::testing::Each;
using ::testing::Eq;
using ::testing::IsFalse;
using ::testing::IsTrue;
-using ::testing::Le;
using ::testing::SizeIs;
+struct EvaluationTrace {
+ struct Snapshot {
+ std::vector counts;
+ std::set contributors;
+ };
+
+ explicit EvaluationTrace(std::size_t entries) : counts_(entries) {}
+
+ EvaluationResult Evaluate(EvalContext& context) ABSL_LOCKS_EXCLUDED(mutex_) {
+ const std::size_t index = context.visit.metadata.size;
+ {
+ const absl::MutexLock lock(mutex_);
+ ++counts_.at(index);
+ contributors_.insert(std::this_thread::get_id());
+ }
+ // A controlled slow evaluator, not a claim about physical storage latency.
+ absl::SleepFor(absl::Milliseconds(2));
+ return {.matched = index % 2 == 0};
+ }
+
+ Snapshot Read() const ABSL_LOCKS_EXCLUDED(mutex_) {
+ const absl::MutexLock lock(mutex_);
+ return {.counts = counts_, .contributors = contributors_};
+ }
+
+ private:
+ // Guards only count and contributor observations; no lock spans the delay.
+ mutable absl::Mutex mutex_;
+ std::vector counts_ ABSL_GUARDED_BY(mutex_);
+ std::set contributors_ ABSL_GUARDED_BY(mutex_);
+};
+
+struct TracedWorker final : ExpressionExecution::Worker::State {
+ explicit TracedWorker(EvaluationTrace& trace) : trace_(trace) {}
+
+ EvaluationResult Evaluate(EvalContext& context) const override { return trace_.Evaluate(context); }
+
+ private:
+ EvaluationTrace& trace_;
+};
+
+struct TracedPlan final : ExpressionExecution::Plan {
+ explicit TracedPlan(EvaluationTrace& trace) : trace_(trace) {}
+
+ ExpressionExecution::Worker MakeWorker(ExpressionWorkerRole) const override {
+ return ExpressionExecution::Worker(std::make_unique(trace_));
+ }
+
+ bool UsesIndexedMatchers() const override { return true; }
+
+ private:
+ EvaluationTrace& trace_;
+};
+
struct ParallelMatchTest : ::testing::Test {
// Type-only expressions never call this backend. No real paths are visited.
const vfs::LocalFs fs;
@@ -37,7 +99,7 @@ struct ParallelMatchTest : ::testing::Test {
for (std::size_t index = 0; index < count; ++index) {
entries.push_back({
.path = std::to_string(index),
- .metadata = {.type = index % 2 == 0 ? vfs::FileType::kRegular : vfs::FileType::kDirectory},
+ .metadata = {.type = index % 2 == 0 ? vfs::FileType::kRegular : vfs::FileType::kDirectory, .size = index},
.fs = fs,
});
}
@@ -104,10 +166,17 @@ TEST_F(ParallelMatchTest, PreparedCoordinatorAndWorkersRetainRegexStateAcrossPoo
parser::BindMatchers(command, regex::Grammar::kRe2, parser::CaseMode::kSensitive);
ASSERT_OK_AND_ASSIGN(const auto execution, PrepareExpressionExecution(*command.expression));
RunExecutor executor_pool(4);
- ParallelMatch matcher(*command.expression, 4, executor_pool, false, std::nullopt, execution);
+ const MatchWorkCosts free_dispatch{
+ .cold_start = {},
+ .worker_start = {},
+ .dispatch = {},
+ .minimum_saving = {},
+ };
+ ParallelMatch matcher(*command.expression, 4, executor_pool, false, std::nullopt, execution, free_dispatch);
// Small batches run on the coordinator, large batches activate private worker state;
// returning to the coordinator must not reuse a worker's previous match or scratch.
- for (const std::size_t count : {1, 128, 7, 512, 8'192, 10}) {
+ constexpr std::array kBatches{1UZ, 128UZ, 5UZ, 512UZ, 8'192UZ, 5UZ};
+ for (const std::size_t count : kBatches) {
const auto& results = matcher.Match(Entries(count));
ASSERT_THAT(results, SizeIs(count));
for (std::size_t index = 0; index < count; ++index) {
@@ -118,20 +187,58 @@ TEST_F(ParallelMatchTest, PreparedCoordinatorAndWorkersRetainRegexStateAcrossPoo
EXPECT_THAT(executor_pool.worker_count(), Eq(3));
}
-TEST_F(ParallelMatchTest, ColdExecutorWaitsForAccumulatedWork) {
+TEST_F(ParallelMatchTest, RepeatedAdmittedBatchesKeepCallerInclusiveBudget) {
MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-true"}));
RunExecutor executor(4);
- ParallelMatch matcher(*command.expression, 4, executor, false);
+ const MatchWorkCosts free_dispatch{
+ .cold_start = {},
+ .worker_start = {},
+ .dispatch = {},
+ .minimum_saving = {},
+ };
+ ParallelMatch matcher(*command.expression, 4, executor, false, std::nullopt, {}, free_dispatch);
for (int batch = 0; batch < 7; ++batch) {
EXPECT_THAT(matcher.Match(Entries(1'024)), SizeIs(1'024));
- // 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(executor.worker_count(), Eq(3));
}
EXPECT_THAT(matcher.Match(Entries(1'024)), SizeIs(1'024));
EXPECT_THAT(executor.worker_count(), Eq(3));
}
+TEST_F(ParallelMatchTest, CompletedSmallBatchesDoNotAdmitSmallTail) {
+ ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-true"}));
+ RunExecutor executor(9);
+ ParallelMatch matcher(*command.expression, 10, executor, false);
+ constexpr std::size_t kHistoryBatches = 512;
+ for (std::size_t batch = 0; batch < kHistoryBatches; ++batch) {
+ EXPECT_THAT(matcher.Match(Entries(16)), SizeIs(16));
+ }
+ EXPECT_THAT(matcher.Match(Entries(32)), SizeIs(32));
+ EXPECT_THAT(executor.worker_count(), Eq(0));
+}
+
+TEST_F(ParallelMatchTest, SlowRemainingWorkUsesCallerAndReusesBoundedWorkers) {
+ ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-true"}));
+ constexpr std::size_t kEntries = 64;
+ EvaluationTrace trace(kEntries);
+ const ExpressionExecution execution(std::make_unique(trace));
+ RunExecutor executor(2);
+ ParallelMatch matcher(*command.expression, 3, executor, false, std::nullopt, execution);
+ constexpr std::size_t kBatches = 2;
+ for (std::size_t batch = 0; batch < kBatches; ++batch) {
+ const auto& results = matcher.Match(Entries(kEntries));
+ ASSERT_THAT(results, SizeIs(kEntries));
+ for (std::size_t index = 0; index < kEntries; ++index) {
+ EXPECT_THAT(results.at(index).evaluation.matched, Eq(index % 2 == 0));
+ }
+ EXPECT_THAT(executor.worker_count(), Eq(2));
+ const auto observed = trace.Read();
+ EXPECT_THAT(observed.counts, Each(Eq(batch + 1)));
+ EXPECT_THAT(observed.contributors, SizeIs(3));
+ EXPECT_THAT(observed.contributors, Contains(std::this_thread::get_id()));
+ }
+}
+
TEST_F(ParallelMatchTest, StatefulMetadataAndActionExpressionsCannotEnterWorkers) {
const std::vector> expressions{
{"root", "-first", "1"},
diff --git a/xff/engine/rg_test.cc b/xff/engine/rg_test.cc
index 7682f3fc4e8..86dc9a485c7 100644
--- a/xff/engine/rg_test.cc
+++ b/xff/engine/rg_test.cc
@@ -15,12 +15,9 @@
#include
#include
-#include
-#include
#include
#include