diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index e68c27c7bc6..ffbda54d909 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -24,7 +24,7 @@ repos:
hooks:
- id: check-added-large-files
# Retain the exact lossless cross-platform performance evidence archives.
- exclude: ^docs/benchmarks/((executor-copy-audit|bounded-read-ahead)-20261101|(directory-task-slots|packed-visit-snapshots|bounded-match-results|useful-match-workers|direct-evaluator-workers|derived-content-handoff|immutable-default-catalogs|prepared-diff-configuration|prepared-hash-operands|aggregate-comparison-totals|byte-shingle-representations|archive-input-buffers)-20261102|(digest-and-record-storage|caller-owned-streams|direct-source-materialization|hash-field-binding|hash-field-binding-tag|forward-cursor-skipping|raw-record-reuse|similarity-prefix-view|similarity-reference-reuse|owned-word-moves|smaller-shingle-probes|native-zip-seek-evaluation|encoded-record-reuse|threshold-aware-similarity)-20261103|(column-row-reuse|word-set-insertion|field-context-owned-results|selected-summary-output|fixed-summary-metrics|direct-summary-cells|compact-comparison-entries|comparison-category-sequence|native-name-paths|uniform-native-case|compact-traversal-metadata|eager-native-paths|native-case-experiments)-20261104|(folded-ascii-globs|bulk-directory-metadata|comparison-retention-followups|comparison-storage-followups|archive-prefix-reuse|archive-directory-range|archive-directory-allocation|archive-directory-observations|native-symlink-observations|native-metadata-followups|symlink-target-demand|native-pgo|readlink-short-target|cctz-empty-format|readlink-input-path|readlink-owned-transfer|archive-lookup-trials|archive-member-keys)-20261105|archive-directory-child-capacity-20261106|shared-stage-workers-20261107|archive-index-insertion-20261107|transactional-rg-records-20261107|immutable-zip-locator-evaluation-20261107|queued-read-cancellation-20261107|archive-btree-historical-decision-20261107|compiled-field-stat-demand-20261107|unchanged-site-payloads-20261107|archive-read-buffer-clear-20261107|archive-read-buffer-requalification-20261108|language-list-reserve-20261107|language-list-reserve-requalification-20261108|ordered-output-credit-20261107)\.json\.gz$
+ exclude: ^docs/benchmarks/((executor-copy-audit|bounded-read-ahead)-20261101|(directory-task-slots|packed-visit-snapshots|bounded-match-results|useful-match-workers|direct-evaluator-workers|derived-content-handoff|immutable-default-catalogs|prepared-diff-configuration|prepared-hash-operands|aggregate-comparison-totals|byte-shingle-representations|archive-input-buffers)-20261102|(digest-and-record-storage|caller-owned-streams|direct-source-materialization|hash-field-binding|hash-field-binding-tag|forward-cursor-skipping|raw-record-reuse|similarity-prefix-view|similarity-reference-reuse|owned-word-moves|smaller-shingle-probes|native-zip-seek-evaluation|encoded-record-reuse|threshold-aware-similarity)-20261103|(column-row-reuse|word-set-insertion|field-context-owned-results|selected-summary-output|fixed-summary-metrics|direct-summary-cells|compact-comparison-entries|comparison-category-sequence|native-name-paths|uniform-native-case|compact-traversal-metadata|eager-native-paths|native-case-experiments)-20261104|(folded-ascii-globs|bulk-directory-metadata|comparison-retention-followups|comparison-storage-followups|archive-prefix-reuse|archive-directory-range|archive-directory-allocation|archive-directory-observations|native-symlink-observations|native-metadata-followups|symlink-target-demand|native-pgo|readlink-short-target|cctz-empty-format|readlink-input-path|readlink-owned-transfer|archive-lookup-trials|archive-member-keys)-20261105|archive-directory-child-capacity-20261106|shared-stage-workers-20261107|archive-index-insertion-20261107|transactional-rg-records-20261107|immutable-zip-locator-evaluation-20261107|queued-read-cancellation-20261107|archive-btree-historical-decision-20261107|compiled-field-stat-demand-20261107|unchanged-site-payloads-20261107|archive-read-buffer-clear-20261107|archive-read-buffer-requalification-20261108|language-list-reserve-20261107|language-list-reserve-requalification-20261108|ordered-output-credit-20261107|adaptive-ordered-output-20261109)\.json\.gz$
- id: check-merge-conflict
- id: check-yaml
# BINARY_TYPES below: archive containers are binary but can START with text (a phar begins with
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 028b4de7bfe..9df09fbad4b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,8 @@
# 0.11.0
+- Adapt ordered content-output scheduling to heavy batches, apply byte backpressure, and share the matcher/coordinator CPU allowance while preserving whole-file validation and cancellation.
+
- Avoid redundant 64 KiB buffer clearing in legacy archive extraction.
- Preserve filesystem-native name matching for dangling and cross-volume symlinks when case
diff --git a/TODO.md b/TODO.md
index bf3f4de90ba..347eb5564d0 100644
--- a/TODO.md
+++ b/TODO.md
@@ -352,7 +352,8 @@ Measured follow-ups (see the P07/P08 sections of `docs/performance-analysis.md`)
- [ ] Add archive read admission or serial extraction into byte-bounded matcher queues before enabling member workers.
- [ ] Investigate ZIP member restart/indexing cost separately from compressed TAR decoding.
-- [ ] Design a byte-bounded ordered output queue and shared CPU allowance before enabling an overlapping pipeline.
+- [ ] Qualify the byte-bounded ordered output queue and shared coordinator/matcher CPU allowance
+ before enabling overlap. Preserve the existing independent directory-I/O limit and JC scope.
## Rg integration review (PRs 916-919)
@@ -506,12 +507,25 @@ EP/JC/P work; candidates are not speedup claims until measured on both platforms
keep the specialized pools and retain [all variants and counterexamples](docs/benchmarks/shared-stage-workers-20261107.md).
Future reuse designs must preserve independent root/mount allowances, leaf jobs, owner
draining and JC gates, and remove the repeatable Linux comparison regression before adoption.
+- [ ] Qualify direct span reads in `content::ScanLines` separately from O18. Native
+ profiling attributes 43.94% of sampled baseline CPU to zeroing beneath its owning
+ 64 KiB read loop. Reuse `ReadInto` only with initialized-range, binary/EOF/error
+ tests and native/memory/archive controls; retain O07's rejected variants rather
+ than assuming removal of zeroing is always a win. See the current O18 qualification
+ notes in [the performance analysis](docs/performance-analysis.md).
- [ ] O18: Qualify byte-bounded overlap with backpressure, ordered publication and cancellation;
avoid per-batch throwaway threads and nested waits on the same saturated pool.
Retain the [ordered-output credit experiment](docs/benchmarks/ordered-output-credit-20261107.md).
- All runtime refinements remain unselected after repeatable ordinary Linux output costs,
- despite large-output/RSS gains. Solve synchronization costs and missing input/slot/source/
- renderer-temporary admission independently; native output and stage overlap remain open.
+ Adaptive native/rg output, owned-text admission, byte backpressure and a shared matcher/
+ coordinator CPU allowance pass cross-platform qualification in
+ [the adaptive-output report](docs/benchmarks/adaptive-ordered-output-20261109.md).
+ Retain ordinary-path costs and anomalous Linux populations alongside large-output gains.
+ Keep full source/decoder/derived-cache/renderer-temporary admission and broader JC gates
+ open; output credit is not a whole-process cap. No public crossover setting is exposed.
+- [ ] Benchmark qualification: distinguish repeated execution of one image from independent
+ file-backed baseline copies. Retain both controls when investigating unexplained native
+ costs, and compare loaded code/data identities before repeating full populations for a
+ build-ID-only change. Preserve all observations rather than choosing a favorable cohort.
- [ ] O18a: Redesign queued cancellation after both measured policies fail cross-platform
qualification. The deterministic nine-to-four backend-call proof and exact serial/recycling
tests pass, but acquire-load and explicit-stop screens expose ordinary Mac costs. Retain
diff --git a/XFF.md b/XFF.md
index ebb17de03ce..0a4e6176a24 100644
--- a/XFF.md
+++ b/XFF.md
@@ -565,7 +565,7 @@ See also: [--help=config](#topic-config), [--help=archive](#topic-archive), [--h
- `-j N, -j=N, --jobs=N|all` - directory-read, eligible content-match, comparison and -exec workers _(global, find, xff, rg)_
- `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories on up to `N` worker threads; large directory listings can split required eager metadata reads across the same pool. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches also match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core.
+ `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories on up to `N` worker threads; large directory listings can split required eager metadata reads across the same pool. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches also match in bounded parallel batches, with output kept in traversal order. For content output, matchers and the coordinator share an `N`-participant allowance; output-heavy batches may overlap collection, matching and publication. Saved output uses byte backpressure, with the current oversized record allowed to finish; this is not a process-memory limit. Small initial batches, archive members and 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: [--help=output](#topic-output), [--help=stats](#topic-stats)
diff --git a/docs/benchmarks/adaptive-ordered-output-20261109.json.gz b/docs/benchmarks/adaptive-ordered-output-20261109.json.gz
new file mode 100644
index 00000000000..4bd87628a3e
Binary files /dev/null and b/docs/benchmarks/adaptive-ordered-output-20261109.json.gz differ
diff --git a/docs/benchmarks/adaptive-ordered-output-20261109.md b/docs/benchmarks/adaptive-ordered-output-20261109.md
new file mode 100644
index 00000000000..ff9d28646d2
--- /dev/null
+++ b/docs/benchmarks/adaptive-ordered-output-20261109.md
@@ -0,0 +1,276 @@
+# Adaptive ordered output and pipeline overlap
+
+Select the adaptive output mechanism for PR #1152: the 1,000-file heavy-output rg
+workload improves 28.88% on Linux and 33.16% on Mac, with roughly half the process
+peak RSS. Small/unknown output stays batched. Ordinary results include costs and
+unstable Linux populations; retain them explicitly rather than claiming a universal
+speedup. Current-head CI remains the final merge gate.
+
+The [lossless evidence capsule](adaptive-ordered-output-20261109.json.gz) contains
+private variants, exact source/build/image identities, test records, all raw timing
+samples, independent arithmetic verification, fixture content recipes and profiling
+observations. SHA-keyed payloads deduplicate identical files without losing bytes.
+Release executable images and raw extracted executable sections are omitted; their
+hashes and original source records remain.
+
+## Scope and ownership
+
+The matcher retains its existing persistent workers. A completed batch selects
+between batch publication and collecting one next batch while matching proceeds:
+unknown/small output uses batch publication, while an average of at least 1 KiB
+per visited entry enables overlap. Later small output turns it off again. This is
+an internal measured policy, not a new command-line option or a claim that output
+size predicts every CPU-heavy matcher.
+
+Input admission accounts for actual owned text capacity, including inline storage
+and the terminator. It admits at most 1 MiB per collecting batch, with one oversized
+record alone and one transit record while checking its actual capacity. Entry-count
+limits remain 256 for content output and 1,024 for decision-only work. Cleared
+predecessor storage is recycled before workers wake. Output-free filters stay batched.
+
+Output transactions reserve capacity before allocating: 64 inline bytes per slot,
+a 4 KiB per-slot heap grant and 8 MiB of shared excess credit. Growth accounts for
+both old and replacement allocations until the old allocation is released. Only
+the current ordered head may exceed shared credit so arbitrary-size whole-file
+binary/read-error validation can finish. Excess storage is returned after each
+publication; only allocations within a slot grant are retained for reuse.
+
+The coordinator borrows a stable whole-file view after producer completion and
+releases it before advancing the ordered head. Native and rg rendering preserve
+whole-file binary/late-error suppression, headings, contexts and callback order.
+Archive-member and traversal-error barriers drain prior work. Cancellation wakes
+byte-credit waiters and drains producers before filesystem/input ownership changes.
+
+Matcher work and coordinator collection/publication share the requested `N` CPU
+units. Workers borrow once per generation, release while parked for output credit,
+and reacquire after unlocking the credit mutex. The coordinator releases its unit
+before waiting for results. Directory I/O retains its existing independent limit;
+this is not one process-wide `N`-operation budget. No per-file CPU lock, clock,
+throwaway worker or generic executor is introduced.
+
+The byte allowance covers retained output allocations, not source buffers, decoder
+workspace, derived caches, renderer temporaries or process RSS. Broader JC/source/
+temporary admission and active-I/O preemption remain open.
+
+## Evidence and protocol
+
+The runtime qualification head is `4245f78c00f483309b22f08d6ac00fdca54214c7`,
+based on main `52950f937ad371e1450538faf7983024b324688d`. The baseline build
+uses `efffe2529f26c1c7ca61763fa5fbb17efd573f6f`; only six backfill/workflow/docs
+files differ between that baseline and the current base, with no runtime or
+build-toolchain change. All images use `--config=clang_release`.
+
+Native qualification covers 76 cells: listing, names, metadata, content filtering,
+native lines, rg filenames and rg lines; flat trees from 10 through 10,000 files,
+a many-directory 10,000-file tree and heavy output at 10/100/1,000 files. A separate
+32-cell crossover matrix covers 128/512/1,024/2,048/8,192/32,768 bytes per file and
+small-heavy-small/heavy-small-heavy transitions, with native/rg lines and one/three
+workers. Each population has three sessions, nine rounds retaining the fastest
+seven per case/participant/session, a discarded warmup, rotated participants,
+reversed second-session cases and a fixed third-session shuffle. An identical
+baseline is executed independently as a drift control.
+
+Each child uses native fork/exec/wait timing and verifies complete output and exit
+status. The native driver compares the complete line multiset with independently
+constructed path/content/metadata oracles; committed tests separately prove ordering.
+Linux pins distinct physical cores and uses `/dev/shm` fixtures; Mac uses APFS without
+an affinity claim. Absolute host times are not compared.
+
+The in-memory screen uses 13 cases, four workers and three grouped sessions with
+nine repetitions retaining seven. It isolates VFS/engine work but includes no native
+process startup. Its Linux diagnostic affinity uses the first four allowed logical
+CPUs, unlike the physical-core native protocol. It is supporting evidence, not a
+substitute for native results.
+
+First-visible-output diagnostics use a parent monotonic clock from before `Popen`
+to the first `read1` bytes. They include parent launch/pipe observation and are not
+native throughput measurements. They verify complete ordered output hashes and
+retain all samples and copied-baseline controls.
+
+## Rejected and intermediate policies
+
+Fixed overlap and per-entry synchronization repeatedly regressed ordinary Linux
+output despite large-output/RSS gains. A 256-entry publication window still cost
+9.28% native lines, 8.38% rg filenames and 13.07% rg lines on flat 10,000-file
+searches; controls were -0.36%, -1.12% and -0.62%. Keep those counterexamples.
+
+The atomic-wait variant spent 7.56% of sampled CPU in clock reads beneath libc++
+polling; blocking conditions avoid that mechanism. A separate syscall diagnostic
+increased futex calls from 324 to 7,550 with per-entry completion; traced elapsed
+time is not used as performance evidence. Windowed publication drains early on
+actual credit backpressure so it cannot deadlock behind a later blocked producer.
+
+Returning live predecessor entries and clearing them after dispatch regressed flat
+work. Recycling cleared entries before dispatch avoids that overlap. A 1 MiB shared
+output policy regressed 32 KiB output; raising the output crossover alone did not
+solve both native and rg costs. Interleaved isolated policy measurements select
+8 MiB shared excess instead. Neither value is an eager reservation.
+
+Collection-only CPU reservation missed publication running alongside every matcher;
+the final shared allowance covers both phases. Prior results remain distinct from
+that implementation. Its explicit lifetime/immobility declaration fix changed whole-file image hashes,
+so Linux native measurements were repeated. Subsequent ELF inspection proves that
+all allocated code/data sections and their addresses are identical between those
+two builds; only the GNU build-ID note differs. The two complete populations are
+retained separately, with that equivalence recorded.
+
+The first fixed-policy Mac population's third session overlapped local clang-tidy;
+retain it as contaminated, not final qualification. The final Mac preflight records
+no compiler or lint process before timing. Earlier failed prototypes/builds and
+private source snapshots remain in the evidence capsule.
+
+## Correctness and remaining gates
+
+Both hosts pass all 270 root full-feature release targets and the 73-target
+standalone extension/API release invocation. Those are distinct invocations, not
+a claim that every underlying source test is unique. The six focused targets are
+CPU allowance, output credit, parallel matching, rg engine, run engine and conformance.
+They pass ASan/TSan on both hosts and MSan on Linux. Local Linux MSan preserves
+LLVM22 instrumentation and its instrumented standard library, using the existing
+LLVM23 `ld.lld` only because Ubuntu26 lacks the older linker's `libxml2.so.2`.
+CI uses its normal configured toolchain, with no override.
+
+Tests exercise actual blocked CPU/byte-credit waiters, cancellation and restart,
+ordered giant heads with full later credit, allocation growth/release, inline and
+long-path records, recycled storage, small/heavy transitions, archive/error barriers
+and owner teardown. The conformance GID tests query fixture ownership: a temporary
+directory need not inherit the process group. The unchanged baseline reproduces
+that distinction; production group matching is unchanged.
+
+The production source is qualified by these local checks; current-head CI must
+also pass before merging. Full source/decoder/temporary memory admission and
+active-I/O preemption remain open under O18/JC. This change does not close those
+gates or expose a public scheduling threshold.
+
+## Native results
+
+Values are the median of three session percentage changes; positive means slower.
+RSS is the median of each session's mean process peak for the same seven fastest
+children. It is not a sum of peaks or a general memory bound.
+
+| Host | Workload, three workers | Elapsed change | Repeated baseline | Baseline RSS MiB | Candidate RSS MiB |
+| ----- | ------------------------ | -------------: | ----------------: | ---------------: | ----------------: |
+| linux | flat10000 / content | -1.19% | -0.56% | 7.08 | 6.94 |
+| linux | flat10000 / native-lines | -0.90% | -1.82% | 6.70 | 6.69 |
+| linux | flat10000 / rg-filter | -0.66% | -1.61% | 6.64 | 6.69 |
+| linux | flat10000 / rg-lines | +8.70% | +3.63% | 6.67 | 6.73 |
+| linux | many10000 / content | +0.16% | -3.84% | 6.61 | 6.49 |
+| linux | many10000 / native-lines | -3.80% | -2.17% | 5.97 | 6.08 |
+| linux | many10000 / rg-filter | +3.78% | +0.02% | 6.25 | 6.28 |
+| linux | many10000 / rg-lines | +4.82% | +0.28% | 6.24 | 6.30 |
+| linux | output1000 / rg-lines | -28.88% | +0.08% | 38.80 | 18.52 |
+| macos | flat10000 / content | +2.15% | +0.22% | 8.64 | 6.82 |
+| macos | flat10000 / native-lines | +0.80% | -2.25% | 14.85 | 13.78 |
+| macos | flat10000 / rg-filter | +0.02% | +0.22% | 15.49 | 14.56 |
+| macos | flat10000 / rg-lines | +0.59% | +0.05% | 15.85 | 14.51 |
+| macos | many10000 / content | +0.73% | -0.07% | 8.35 | 6.38 |
+| macos | many10000 / native-lines | +0.43% | -0.14% | 13.93 | 12.95 |
+| macos | many10000 / rg-filter | +0.48% | -0.36% | 14.81 | 13.30 |
+| macos | many10000 / rg-lines | +0.46% | +0.55% | 15.42 | 14.38 |
+| macos | output1000 / rg-lines | -33.16% | -0.82% | 79.30 | 38.12 |
+
+`rg-filter` is filename output (`-l`). The complete 76-cell matrices include listing,
+names, metadata and one-worker controls, not just the rows shown here. The Mac
+ordinary three-worker line/filename cells are within about 1%; metadata/content
+controls also include small costs. Linux many-directory filename output costs
+3.78%, with a repeated-baseline change of +0.02%. Keep this tradeoff visible.
+
+## Linux unchanged-path confirmation
+
+The published-image population includes a 26.07% many-directory name-listing cost
+and an 8.70% flat rg-line cost. An interleaved seven-case confirmation with both
+preceding implementations still shows +13.85% for the name case, but not the rg-line
+cost. These are real observations, not discarded outliers.
+
+A subsequent confirmation uses separate executable copies, including an independently
+copied baseline. The current and preceding CPU implementations have identical
+allocated code/data and addresses, except for the build-ID note. That cohort changes
+the apparent name cost's sign and keeps the large-output gain. It demonstrates that
+those large differences cannot be assigned to the declaration edit. It does not
+prove that every small ordinary-path cost is zero.
+
+| Three-worker case | Published population | Repeat control | Independent-image confirmation | Independent baseline copy |
+| ----------------------- | -------------------: | -------------: | -----------------------------: | ------------------------: |
+| flat10000 / files-exact | +5.02% | -1.13% | +2.10% | -1.65% |
+| flat10000 / name-exact | +6.08% | -0.31% | +1.76% | -1.42% |
+| flat10000 / rg-lines | +8.70% | +3.63% | +0.29% | +0.02% |
+| many10000 / files-exact | +10.19% | +1.03% | +1.43% | -0.15% |
+| many10000 / name-exact | +26.07% | +3.89% | -8.60% | +1.67% |
+| many10000 / rg-lines | +4.82% | +0.28% | -1.36% | -0.29% |
+| output1000 / rg-lines | -28.88% | +0.08% | -29.90% | -2.70% |
+
+Separate `perf stat` diagnostics send output to `/dev/null` and are not the native
+capture/timing protocol. Median retired user instructions change about +0.02% for
+flat names and +0.69% for many-directory names; candidate cycle counts fall in that
+diagnostic. This rules out a 26% increase in retired instruction work in those
+fixtures, without proving a universal elapsed-time result or a specific host cause.
+No live profiler/compiler was found in the subsequent Linux process check; idle
+Bazel servers are not compilation activity.
+
+Most earlier baseline/control labels execute the same image independently; they
+are temporal controls, not independent file-backed copies. Only the explicit
+independent-image confirmation makes the latter claim. Retain this distinction in
+future qualification and inspect allocated sections before repeating a complete
+population for a build-ID-only change.
+
+## Crossover and changing output
+
+| Host | Three-worker case | Elapsed change | Repeated baseline |
+| ----- | -------------------------------- | -------------: | ----------------: |
+| linux | payload32768 / native-lines | -5.67% | -0.45% |
+| linux | payload32768 / rg-lines | -0.22% | -1.64% |
+| linux | small-heavy-small / native-lines | +4.41% | +1.00% |
+| linux | small-heavy-small / rg-lines | +3.47% | +1.44% |
+| linux | heavy-small-heavy / native-lines | +1.05% | +2.09% |
+| linux | heavy-small-heavy / rg-lines | +0.52% | -0.73% |
+| macos | payload32768 / native-lines | -5.38% | +0.95% |
+| macos | payload32768 / rg-lines | -0.20% | +0.07% |
+| macos | small-heavy-small / native-lines | -0.11% | +6.99% |
+| macos | small-heavy-small / rg-lines | +1.23% | +0.53% |
+| macos | heavy-small-heavy / native-lines | +1.45% | +3.94% |
+| macos | heavy-small-heavy / rg-lines | +1.07% | +2.65% |
+
+All 32 crossover cells, including one worker and intermediate payload sizes, remain
+in the capsule. The mixed small-heavy-small Linux cells retain 3-4% costs; the
+change is selected for its substantial large-output and retained-storage benefits,
+with these transitions and ordinary costs recorded for review and further tuning.
+
+## Supporting memory-backend and first-output diagnostics
+
+The memory-backend large-output case improves 7.75% on Linux and 16.52% on Mac;
+repeated-baseline changes are -0.18% and -2.72%. Four-worker 10,000-file filters
+improve 7.19%/13.94%. Small cases are mixed: Mac 1,000/10,000-file rg filename
+output costs 2.22%/2.65%, with controls +2.72%/+4.01%; Linux 1,000-file rg lines
+cost 2.44%, with control -2.27%. Preserve all 13 cells rather than extrapolating
+from the strongest result.
+
+| Host | First-visible-output fixture | Baseline ms | Candidate ms | Repeated baseline ms |
+| ----- | ---------------------------- | ----------: | -----------: | -------------------: |
+| linux | flat10000 | 2.846 | 2.990 | 2.800 |
+| linux | many10000 | 2.186 | 2.315 | 2.140 |
+| linux | output100 | 2.936 | 2.459 | 2.868 |
+| linux | output1000 | 5.967 | 3.098 | 5.778 |
+| macos | flat10000 | 16.483 | 16.436 | 16.610 |
+| macos | many10000 | 12.202 | 12.380 | 12.368 |
+| macos | output100 | 10.793 | 9.820 | 10.651 |
+| macos | output1000 | 15.019 | 10.492 | 15.005 |
+
+These pipe-observation values include parent overhead. They support earlier visible
+large-output progress; they do not establish native throughput or a small-input
+startup improvement.
+
+## Footprint and reproducibility
+
+The lean release CLI grows from 12,729,496 to 12,823,944 bytes on Linux (+94,448,
+0.74%) and from 4,196,976 to 4,222,336 bytes on Mac (+25,360, 0.60%). These are
+copied local release images; no full-feature distribution size is inferred.
+
+The capsule's path index references SHA-256-keyed UTF-8 or base64 blobs. Decode a
+blob to recover its original file bytes; every reference and payload is verified
+by a round trip. Source snapshots identify the base and exact modified files.
+The fixture recipe maps 30,866 relative file paths to ten unique content blobs;
+create parents, decode each content blob and write the exact bytes, then update
+manifest roots/launcher/image paths for a new machine. Native and memory drivers,
+launcher's source, all original commands/logs, test XML, failed prototypes and
+profiling observations remain available. Executables are reproducible from the
+recorded source and toolchain instead of embedded in the repository.
diff --git a/docs/design-parallel.md b/docs/design-parallel.md
index d1116888551..53f00afb2ca 100644
--- a/docs/design-parallel.md
+++ b/docs/design-parallel.md
@@ -79,6 +79,21 @@ A bounded directory-read worker pool with a single coordinator:
records for ordered emission. Full line/context output buffers each input; filename/count/quiet
selection streams where the VFS supports it, retaining the longest line and draining the source
for binary/error checks. Neither the entry-count batch bound nor streaming limits record size.
+- Eligible host content output shares one `N`-participant CPU allowance between matcher workers
+ and the coordinator. Each producer borrows once per generation. It returns its unit while
+ blocked on output credit and reacquires after dropping the credit mutex, including cancellation
+ drains. The coordinator returns its unit while awaiting results and borrows for collection or
+ ordered publication. Directory-read I/O retains its independent existing limit; this is not one
+ global `N`-operation budget. No per-file timer or CPU-admission lock is added.
+- Output volume already produced selects batch publication or overlap with one collecting input
+ batch. Unknown or small output starts batched. Input text admission uses actual owned capacity,
+ with one oversized record alone. Output slots have 4 KiB grants plus an 8 MiB shared excess
+ allowance, charged for old and new allocations during growth. The ordered head may exceed that
+ allowance so whole-file validation can finish. Fixed slot storage, source/decoder buffers,
+ derived caches and renderer temporaries remain separate; this is not a process-memory cap.
+ Full-file binary/late-error validation precedes publication. Serial archive and traversal-error
+ boundaries drain prior host output. Cancellation wakes credit waiters and drains producers
+ before input or filesystem ownership can be recycled.
- Each worker binds native and output RE2 state once to avoid shared DFA-cache contention. PCRE2
workers share compiled patterns and reuse private match scratch, with bounded retained interpreter
heap frames and independent reentrant leases. Entry content and lazy metadata have one owner and
@@ -142,7 +157,8 @@ actions still occur in traversal order.
## Parallelism control
A single knob, `-j N` (long form `--jobs`), configures independent limits: up to
-`N` directory-read workers, up to `N` eligible content-matcher workers, and at most
+`N` directory-read workers, up to `N` eligible content-matcher participants (including
+coordinator collection/publication when the content-output pipeline is active), and at most
`N` outstanding semicolon-form `-exec`/`-execdir` children. These are separate
limits, not one shared `N`-operation budget. Content matching does not run alongside
execution actions in the same expression. `-j 1` makes each part synchronous.
diff --git a/docs/performance-analysis.md b/docs/performance-analysis.md
index 173fd33dfff..85e0757c8ed 100644
--- a/docs/performance-analysis.md
+++ b/docs/performance-analysis.md
@@ -5066,3 +5066,147 @@ stage overlap open. No global memory bound or public-head CI claim follows.
[Current-main requalification](benchmarks/ordered-output-credit-requalification-20261108.md) reproduces ordinary Linux
three-worker costs (+12.85% filenames/+10.88% lines), so H2 remains unselected despite
large-output gains. The complete case population and copied controls remain available.
+
+### O18 current-main overlap qualification in progress
+
+The current experiment starts a matcher generation on the existing persistent workers
+while the coordinator collects one further input batch. It publishes completed entries
+in input order, drains before serial/archive and traversal-error boundaries, and wakes
+credit waiters before worker teardown. A single transit record lets input admission check
+actual owned text capacity before queue insertion; oversized records are admitted alone.
+Output growth charges both old and new allocations against per-slot grants plus shared
+credit. Only the current ordered head may exceed the shared allowance, preserving
+arbitrarily large transactional output without truncation or a second source read.
+
+The overlap-only Linux screen improves the 10,000-file content-filter and filename-output
+cases, but the integrated credited variants repeatedly regress ordinary line output.
+Changing completion notifications, polling and layout has not qualified those costs away.
+The implementation remains an unselected development experiment; the earlier rejection
+and the remaining source/temporary/storage admission gates still apply. Current macOS
+focused TSan tests cover delayed heads/full later output, blocked-credit cancellation,
+restart, filesystem-owner teardown and long/oversized paths. These are development
+checks, not public-head/full-platform qualification or a whole-process memory bound.
+
+Linux user-space sampling of the same 10,000-file rg-line workload identifies a
+specific wait implementation cost: the compact atomic-wait prototype spends 7.56%
+of sampled CPU in `clock_gettime` reached through libc++ `atomic::wait`'s
+`steady_clock` polling. This is diagnostic evidence, not a timing gain. The next
+variant uses blocking condition waits for completion and output credit; notifications
+remain conditional on a registered waiter. Its outcome still requires measurement.
+
+Native whole-command validation then exposes a separate synchronization cost: in a
+traced 10,000-file flat-tree rg-line invocation, futex calls rise from 324 in the
+baseline to 7,550 with per-entry blocking completion. Traced wall times are not
+performance measurements. The 76-case native Linux population retains repeatable
+ordinary flat-tree costs even though the in-memory screen improved. Do not select
+that variant from the in-memory results alone.
+
+The next prototype amortizes ordinary publication in 64-entry windows, but abandons
+the window whenever any producer needs output credit. A credit waiter explicitly
+wakes the coordinator's condition; the coordinator then drains the ordered head,
+which retains its oversize guarantee. This is necessary to avoid deadlock when a
+window's later target cannot finish until earlier records release bytes. Small/tail
+batches instead evaluate and release one credited transaction at a time inline.
+Neither implementation is qualified by this design description alone.
+
+The native user-space profile also identifies a separate read-loop candidate. In
+500 baseline rg-line invocations over the 10,000-file flat fixture, 43.94% of sampled
+CPU reaches `memset` beneath `ScanLines`. Source inspection confirms that this loop
+calls owning `Read(65,536)`; the host implementation constructs a zero-filled
+64 KiB string before each read, including the EOF probe. This is not the earlier
+`ReadContentRange` change and not proof that removing zeroing alone improves native
+elapsed time. Qualify a direct span-read line-scanner path independently, using the
+existing native `ReadInto` capability, and retain O07's rejected memory/ten-worker
+controls. Preserve binary sniffing, short reads, EOF and late-error transactions;
+never read beyond the returned initialized byte range.
+
+### O18: adapt scheduling to observed output volume
+
+The fixed credited-overlap policies improve output-heavy work but retain ordinary
+Linux output costs. A larger publication window alone does not qualify them: the
+256-entry-window experiment still slows flat 10,000-file native lines, rg filenames
+and rg lines by 9.28%, 8.38% and 13.07%, respectively. Its copied-baseline controls
+are -0.36%, -1.12% and -0.62%; each candidate cost occurs in all three sessions.
+Output-heavy 1,000-file rg output improves 24.41%, so one fixed scheduling policy
+is not an adequate decision for both populations.
+
+The next private prototype chooses between batch publication and stage overlap
+from output bytes already produced. Unknown output and output-free filters start in batch mode. A completed
+batch averaging at least 1 KiB of output per visited entry enables overlap for the
+next batch; a later small-output batch switches it off again. This threshold is an
+experimental crossover, not a public option or a qualified default. It adds no
+metadata request or timer. Low-output computational work might still benefit from
+overlap; output size alone is not claimed to estimate every kind of matching cost.
+
+Both paths retain the same per-slot grants, shared excess credit, ordered-head
+oversize exception, whole-file validation, cancellation and owner-draining rules.
+Batch mode normally waits once for the batch's final publication slot, but drains
+early when any producer blocks on credit. It never waits for every producer while
+withholding the bytes that those producers need to finish. Overlap mode retains
+smaller publication windows and reserves a participant while collection is active.
+
+Tests alternate small and large transactions on reused workers and compare exact
+serial/parallel native and rg output across small-heavy-small traversal batches.
+Qualification must include output-size crossover and mixed workloads, unchanged
+listing/metadata paths, and both native and memory backends on both hosts. A useful
+heavy-output result cannot overrule a repeated ordinary-work regression. Neither
+this prototype nor the preceding tests establish the wider JC input/source/
+renderer-temporary admission contract or a whole-process memory cap.
+
+The next Linux policy comparison keeps the same cleared-input-reuse implementation
+and interleaves the baseline, copied baseline, 1 MiB policy, 64 KiB crossover policy
+and 8 MiB policy within every case/session. Raising the crossover alone does not
+resolve both native and rg 32 KiB-output costs. An 8 MiB shared excess allowance
+instead changes those cells from +5.85%/+4.59% to -6.70%/-4.20%, while the copied
+controls are +2.10%/-1.56%. Output-heavy 1,000-file native/rg output improves
+27.65%/26.41% with copied controls -0.33%/-0.86%. These are Linux development
+measurements, not a complete cross-platform qualification.
+
+The selected next qualification source retains the 1 KiB observed-output crossover,
+uses an 8 MiB shared excess allowance, and returns cleared predecessor input storage
+for reuse. Clearing those entries before worker dispatch matters: returning live
+records and clearing them on the coordinator after waking workers regressed ordinary
+flat-tree work. Preserve that rejected source and its raw population. The 8 MiB
+allowance remains a bound on shared excess output capacity, in addition to per-slot
+grants and the ordered-head oversize exception. It does not reserve 8 MiB eagerly,
+change the pending-input allowance or bound source/temporary/process memory.
+
+The collection-only participant reservation does not account for publication running
+alongside every matcher. The next qualification source replaces that reservation
+with a shared `N`-unit allowance for coordinator and matcher work. Producers borrow
+once per generation and temporarily return their unit during output-credit waits;
+they drop the credit mutex before reacquiring CPU, even on cancellation. The
+coordinator returns its unit before waiting for a result, and borrows while collecting
+or publishing. This keeps the new pipeline inside its shared participant allowance
+without per-file CPU-admission locks or timers. Directory I/O retains the existing
+independent limit documented for `--jobs`; command-wide stage-policy/JC work remains
+separate. Deterministic tests cover actual CPU/credit waiters, mutex handoff, resumption
+and cancellation with a coordinator-owned unit. Both hosts pass the six focused
+release targets; full shared-CPU qualification remains in progress.
+
+### O18: adaptive output qualification
+
+The [adaptive-output report](benchmarks/adaptive-ordered-output-20261109.md) retains
+all fixed/adaptive/CPU-accounting variants and their source, tests and raw observations.
+The selected runtime keeps small/unknown output batched, shares CPU admission across
+coordinator/matchers, and applies output credit to native and rg transactions. Both
+hosts pass the 270-target root release invocation and the 73-target standalone
+extension/API invocation, with six focused ASan/TSan targets per host and Linux MSan.
+
+The 1,000-file heavy-output rg workload improves 28.88% on Linux and 33.16% on Mac;
+peak RSS falls from 38.80 to 18.52 MiB and from 79.30 to 38.12 MiB. First-visible
+pipe output also improves. Ordinary results remain mixed: Linux many-directory rg
+filename output costs 3.78%; mixed-output transitions retain small costs. Preserve
+all 76 native and 32 crossover cells per host and the independent-image confirmations.
+
+One Linux population reports a 26.07% ordinary name cost; later cohorts change its
+sign. All loaded sections/addresses of the preceding/current CPU builds match except
+for the build-ID note. Separate instruction diagnostics show about 0.02%/0.69% more
+retired user instructions for flat/many-directory names, not a 26% increase. This
+explains why that large observation cannot be assigned to the declaration edit;
+it does not establish zero ordinary elapsed cost or a specific host cause.
+
+The capsule preserves original observations, repeated-baseline and independent-image
+controls, the contaminated earlier Mac session, failed prototypes, fixture recipes
+and exact image identities. No whole-process bound, active-I/O preemption or complete
+JC/source/temporary admission is inferred. Those parts of O18 remain open.
diff --git a/release-site.json b/release-site.json
index 535b340171c..54585c6063d 100644
--- a/release-site.json
+++ b/release-site.json
@@ -154,7 +154,8 @@
"docs/performance-candidate-completion-audit.md": "docs/performance-candidate-completion-audit.html",
"docs/benchmarks/language-list-reserve-requalification-20261108.md": "docs/benchmarks/language-list-reserve-requalification-20261108.html",
"docs/benchmarks/ordered-output-credit-20261107.md": "docs/benchmarks/ordered-output-credit-20261107.html",
- "docs/benchmarks/ordered-output-credit-requalification-20261108.md": "docs/benchmarks/ordered-output-credit-requalification-20261108.html"
+ "docs/benchmarks/ordered-output-credit-requalification-20261108.md": "docs/benchmarks/ordered-output-credit-requalification-20261108.html",
+ "docs/benchmarks/adaptive-ordered-output-20261109.md": "docs/benchmarks/adaptive-ordered-output-20261109.html"
},
"generated_html": {
"XFF.md": {
@@ -199,6 +200,7 @@
"docs/benchmarks/language-list-reserve-20261107.json.gz": "docs/benchmarks/language-list-reserve-20261107.json.gz",
"docs/benchmarks/language-list-reserve-requalification-20261108.json.gz": "docs/benchmarks/language-list-reserve-requalification-20261108.json.gz",
"docs/benchmarks/ordered-output-credit-20261107.json.gz": "docs/benchmarks/ordered-output-credit-20261107.json.gz",
- "docs/benchmarks/ordered-output-credit-requalification-20261108.json.gz": "docs/benchmarks/ordered-output-credit-requalification-20261108.json.gz"
+ "docs/benchmarks/ordered-output-credit-requalification-20261108.json.gz": "docs/benchmarks/ordered-output-credit-requalification-20261108.json.gz",
+ "docs/benchmarks/adaptive-ordered-output-20261109.json.gz": "docs/benchmarks/adaptive-ordered-output-20261109.json.gz"
}
}
diff --git a/xff/cli/globals.cc b/xff/cli/globals.cc
index 4d7f75244dc..591add2b7e9 100644
--- a/xff/cli/globals.cc
+++ b/xff/cli/globals.cc
@@ -1026,8 +1026,12 @@ constexpr std::array kGlobals = std::to_array({
"across the same pool. Lazy and metadata-free listings do not start stat workers. "
"Independent content predicates and rg searches also match in bounded "
"parallel batches, "
- "with output kept in traversal order. Small initial batches, archive members, stateful expressions "
- "stay on the coordinator. Metadata-consuming output does not itself prevent eligible content "
+ "with output kept in traversal order. For content output, matchers and the coordinator share "
+ "an `N`-participant allowance; output-heavy batches may overlap collection, matching and "
+ "publication. Saved output uses byte backpressure, with the current oversized record allowed "
+ "to finish; this is not a process-memory limit. Small initial batches, archive members and "
+ "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 "
diff --git a/xff/conformance/conformance_test.cc b/xff/conformance/conformance_test.cc
index a2cd420b6fd..127fa8f777f 100644
--- a/xff/conformance/conformance_test.cc
+++ b/xff/conformance/conformance_test.cc
@@ -333,8 +333,10 @@ TEST_F(ConformanceTest, UidMatchesCurrentUser) {
ExpectXff({"-uid", std::to_string(::geteuid())}, All()); // every entry is owned by the test user
}
-TEST_F(ConformanceTest, GidMatchesCurrentGroup) {
- ExpectXff({"-gid", std::to_string(::getegid())}, All());
+TEST_F(ConformanceTest, GidMatchesFixtureGroup) {
+ // The temporary directory's inherited group can differ from the process group.
+ MBO_ASSERT_OK_AND_ASSIGN(const auto metadata, fs_.Stat(Rel(""), false));
+ ExpectXff({"-gid", std::to_string(metadata.gid)}, All());
}
TEST_F(ConformanceTest, UserMatchesCurrentUser) {
@@ -346,11 +348,12 @@ TEST_F(ConformanceTest, UserMatchesCurrentUser) {
ExpectXff({"-user", pw->pw_name}, All());
}
-TEST_F(ConformanceTest, GroupMatchesCurrentGroup) {
+TEST_F(ConformanceTest, GroupMatchesFixtureGroup) {
+ MBO_ASSERT_OK_AND_ASSIGN(const auto metadata, fs_.Stat(Rel(""), false));
// NOLINTNEXTLINE(concurrency-mt-unsafe): single-threaded CLI/test path
- const struct group* const gr = ::getgrgid(::getegid());
+ const struct group* const gr = ::getgrgid(static_cast(metadata.gid));
if (gr == nullptr) {
- GTEST_SKIP() << "no group entry for egid";
+ GTEST_SKIP() << "no group entry for fixture gid";
}
ExpectXff({"-group", gr->gr_name}, All());
}
diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel
index 72ebaef0265..f436945d90a 100644
--- a/xff/engine/BUILD.bazel
+++ b/xff/engine/BUILD.bazel
@@ -260,6 +260,7 @@ cc_library(
":expression_contract_cc",
":extract_cc",
":mount_cc",
+ ":output_credit_cc",
":walk_cc",
"//xff/content:line_match_cc",
"//xff/content:snapshot_cc",
@@ -368,13 +369,56 @@ cc_fuzz_test(
)
# Drives one find run: walk the roots, evaluate the expression, (implicit) print.
+cc_library(
+ name = "cpu_allowance_cc",
+ srcs = ["cpu_allowance.cc"],
+ hdrs = ["cpu_allowance.h"],
+)
+
+cc_test(
+ name = "cpu_allowance_test",
+ size = "small",
+ srcs = ["cpu_allowance_test.cc"],
+ deps = [
+ ":cpu_allowance_cc",
+ "@googletest//:gtest",
+ "@googletest//:gtest_main",
+ ],
+)
+
+cc_library(
+ name = "output_credit_cc",
+ srcs = ["output_credit.cc"],
+ hdrs = ["output_credit.h"],
+ deps = [
+ ":cpu_allowance_cc",
+ "@abseil-cpp//absl/synchronization",
+ "@mboworks_mbo//mbo/types:optional_ref_cc",
+ ],
+)
+
+cc_test(
+ name = "output_credit_test",
+ size = "small",
+ srcs = ["output_credit_test.cc"],
+ deps = [
+ ":cpu_allowance_cc",
+ ":output_credit_cc",
+ "@abseil-cpp//absl/synchronization",
+ "@googletest//:gtest_main",
+ "@mboworks_mbo//mbo/testing:matchers_cc",
+ ],
+)
+
cc_library(
name = "parallel_match_cc",
srcs = ["parallel_match.cc"],
hdrs = ["parallel_match.h"],
deps = [
+ ":cpu_allowance_cc",
":evaluate_cc",
":expression_execution_cc",
+ ":output_credit_cc",
":owned_visit_cc",
"//xff/content:snapshot_cc",
"//xff/fuzzy:fuzzy_cc",
@@ -714,6 +758,7 @@ cc_test(
srcs = ["rg_test.cc"],
deps = [
":evaluate_cc",
+ ":output_credit_cc",
":owned_visit_cc",
":parallel_match_cc",
":run_cc",
diff --git a/xff/engine/cpu_allowance.cc b/xff/engine/cpu_allowance.cc
new file mode 100644
index 00000000000..13184c0b8a2
--- /dev/null
+++ b/xff/engine/cpu_allowance.cc
@@ -0,0 +1,61 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "xff/engine/cpu_allowance.h"
+
+#include
+
+namespace xff::engine {
+
+CpuAllowance::CpuAllowance(std::size_t participants) : available_(std::max(participants, std::size_t{1})) {}
+
+void CpuAllowance::Acquire() {
+ std::unique_lock lock(mutex_);
+ if (available_ == 0) {
+ ++waiters_;
+ condition_.wait(lock, [&] { return available_ != 0; });
+ --waiters_;
+ }
+ --available_;
+}
+
+void CpuAllowance::Release() {
+ bool waiting = false;
+ {
+ const std::scoped_lock lock(mutex_);
+ ++available_;
+ waiting = waiters_ != 0;
+ }
+ if (waiting) {
+ condition_.notify_one();
+ }
+}
+
+CpuAllowance::Usage CpuAllowance::GetUsage() const {
+ const std::scoped_lock lock(mutex_);
+ return {.available = available_, .waiters = waiters_};
+}
+
+CpuAllowance::Lease::Lease(CpuAllowance& allowance) : allowance_(allowance) {
+ Resume();
+}
+
+CpuAllowance::Lease::~Lease() {
+ Pause();
+}
+
+void CpuAllowance::Lease::Pause() {
+ if (held_) {
+ held_ = false;
+ allowance_.Release();
+ }
+}
+
+void CpuAllowance::Lease::Resume() {
+ if (!held_) {
+ allowance_.Acquire();
+ held_ = true;
+ }
+}
+
+} // namespace xff::engine
diff --git a/xff/engine/cpu_allowance.h b/xff/engine/cpu_allowance.h
new file mode 100644
index 00000000000..a251162905e
--- /dev/null
+++ b/xff/engine/cpu_allowance.h
@@ -0,0 +1,60 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#ifndef XFF_ENGINE_CPU_ALLOWANCE_H_
+#define XFF_ENGINE_CPU_ALLOWANCE_H_
+
+#include
+#include
+#include
+
+namespace xff::engine {
+
+// Pipeline participants borrow at generation/publication boundaries, not per file.
+// A producer temporarily returns its unit while blocked on output credit.
+class CpuAllowance final {
+ public:
+ explicit CpuAllowance(std::size_t participants);
+ ~CpuAllowance() = default;
+ CpuAllowance(const CpuAllowance&) = delete;
+ CpuAllowance& operator=(const CpuAllowance&) = delete;
+ CpuAllowance(CpuAllowance&&) = delete;
+ CpuAllowance& operator=(CpuAllowance&&) = delete;
+
+ class Lease final {
+ public:
+ explicit Lease(CpuAllowance& allowance);
+ ~Lease();
+ Lease(const Lease&) = delete;
+ Lease& operator=(const Lease&) = delete;
+ Lease(Lease&&) = delete;
+ Lease& operator=(Lease&&) = delete;
+
+ void Pause();
+ void Resume();
+
+ private:
+ CpuAllowance& allowance_;
+ bool held_ = false;
+ };
+
+ struct Usage {
+ std::size_t available;
+ std::size_t waiters;
+ };
+
+ Usage GetUsage() const;
+
+ private:
+ void Acquire();
+ void Release();
+
+ mutable std::mutex mutex_;
+ std::condition_variable condition_;
+ std::size_t available_;
+ std::size_t waiters_ = 0;
+};
+
+} // namespace xff::engine
+
+#endif // XFF_ENGINE_CPU_ALLOWANCE_H_
diff --git a/xff/engine/cpu_allowance_test.cc b/xff/engine/cpu_allowance_test.cc
new file mode 100644
index 00000000000..bc208677c16
--- /dev/null
+++ b/xff/engine/cpu_allowance_test.cc
@@ -0,0 +1,110 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "xff/engine/cpu_allowance.h"
+
+#include
+#include
+#include
+#include
+#include
+#include
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace xff::engine {
+namespace {
+
+using ::testing::Eq;
+using ::testing::IsFalse;
+using ::testing::IsTrue;
+using ::testing::Le;
+
+struct CpuAllowanceTest : ::testing::Test {
+ static bool AwaitWaiters(const CpuAllowance& allowance, std::size_t count) {
+ const auto end = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while (allowance.GetUsage().waiters < count && std::chrono::steady_clock::now() < end) {
+ std::this_thread::yield();
+ }
+ return allowance.GetUsage().waiters >= count;
+ }
+};
+
+TEST_F(CpuAllowanceTest, ZeroUnitsRetainsOneUsableParticipantAndLeaseRestoresIt) {
+ CpuAllowance allowance(0);
+ EXPECT_THAT(allowance.GetUsage().available, Eq(1));
+ {
+ CpuAllowance::Lease lease(allowance);
+ EXPECT_THAT(allowance.GetUsage().available, Eq(0));
+ lease.Pause();
+ lease.Pause();
+ EXPECT_THAT(allowance.GetUsage().available, Eq(1));
+ lease.Resume();
+ lease.Resume();
+ EXPECT_THAT(allowance.GetUsage().available, Eq(0));
+ }
+ EXPECT_THAT(allowance.GetUsage().available, Eq(1));
+}
+
+TEST_F(CpuAllowanceTest, ABlockedProducerReturnsItsUnitForTheCoordinatorAndCanResume) {
+ CpuAllowance allowance(1);
+ CpuAllowance::Lease producer(allowance);
+ std::atomic entered = false;
+ std::latch leave(1);
+ std::jthread coordinator([&] {
+ const CpuAllowance::Lease lease(allowance);
+ entered.store(true);
+ leave.wait();
+ });
+ EXPECT_THAT(AwaitWaiters(allowance, 1), IsTrue());
+ EXPECT_THAT(entered.load(), IsFalse());
+ producer.Pause();
+ const auto end = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while (!entered.load() && std::chrono::steady_clock::now() < end) {
+ std::this_thread::yield();
+ }
+ EXPECT_THAT(entered.load(), IsTrue());
+ leave.count_down();
+ coordinator.join();
+ producer.Resume();
+ EXPECT_THAT(allowance.GetUsage().available, Eq(0));
+}
+
+TEST_F(CpuAllowanceTest, ConcurrentOwnersStayWithinTheSharedAllowance) {
+ CpuAllowance allowance(3);
+ std::latch start(8);
+ std::latch leave(1);
+ std::atomic active = 0;
+ std::atomic peak = 0;
+ std::vector participants;
+ participants.reserve(8);
+ for (std::size_t index = 0; index < 8; ++index) {
+ participants.emplace_back([&] {
+ start.arrive_and_wait();
+ const CpuAllowance::Lease lease(allowance);
+ const auto count = active.fetch_add(1) + 1;
+ auto previous = peak.load();
+ while (previous < count && !peak.compare_exchange_weak(previous, count)) {}
+ leave.wait();
+ active.fetch_sub(1);
+ });
+ }
+ EXPECT_THAT(AwaitWaiters(allowance, 5), IsTrue());
+ const auto end = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while (active.load() < 3 && std::chrono::steady_clock::now() < end) {
+ std::this_thread::yield();
+ }
+ EXPECT_THAT(active.load(), Eq(3));
+ EXPECT_THAT(allowance.GetUsage().available, Eq(0));
+ leave.count_down();
+ for (auto& participant : participants) {
+ participant.join();
+ }
+ EXPECT_THAT(peak.load(), Le(3));
+ EXPECT_THAT(active.load(), Eq(0));
+ EXPECT_THAT(allowance.GetUsage().available, Eq(3));
+}
+
+} // namespace
+} // namespace xff::engine
diff --git a/xff/engine/evaluate.cc b/xff/engine/evaluate.cc
index 89bf71eca05..fb570dccc94 100644
--- a/xff/engine/evaluate.cc
+++ b/xff/engine/evaluate.cc
@@ -66,6 +66,7 @@
#include "xff/content/line_match.h"
#include "xff/content/word.h"
#include "xff/datetime/datetime.h"
+#include "xff/engine/output_credit.h"
#include "xff/engine/walk.h"
#include "xff/exec/exec.h"
#include "xff/fuzzy/fuzzy.h"
@@ -2193,29 +2194,57 @@ void EmitSelectedGrepLine(
// Source bytes are consumed incrementally. Buffer rendered records until the complete scan
// has validated binary detection and read errors; source and all-line vectors are never retained.
-template
-using GrepLineResult = std::conditional_t;
+void AppendRecords(std::string& records, std::string_view text) {
+ records.append(text);
+}
+
+void AppendRecords(CreditedOutput& records, std::string_view text) {
+ records.Append(text);
+}
+
+std::string_view RecordsView(std::string_view records) {
+ return records;
+}
+
+std::string_view RecordsView(const CreditedOutput& records) {
+ return records.View();
+}
-template
-absl::StatusOr> RenderGrepLines(
+void ClearRecords(std::string& records) {
+ records.clear();
+}
+
+void ClearRecords(CreditedOutput& records) {
+ records.Clear();
+}
+
+void InsertHeading(std::string& records, std::string_view heading) {
+ records.insert(0, heading);
+}
+
+void InsertHeading(CreditedOutput& records, std::string_view heading) {
+ records.Prepend(heading);
+}
+
+template
+absl::StatusOr RenderGrepLinesInto(
const parser::Expr& expr,
EvalContext& ctx,
const GrepMatchers& matcher,
std::uint64_t sniff,
- std::string records = {}) {
+ Buffer& rendered) {
if (ctx.visit.metadata.type != vfs::FileType::kRegular) {
- return GrepLineResult{};
+ return false;
}
const bool only_matching = ctx.grep.only_matching && !(ctx.grep.rg_mode && ctx.grep.invert);
const bool with_context = !only_matching && (ctx.grep_before != 0 || ctx.grep_after != 0);
content::LineContext selection(with_context ? ctx.grep_before : 0, with_context ? ctx.grep_after : 0);
- std::string rendered = std::move(records);
- rendered.clear();
+ ClearRecords(rendered);
std::conditional_t, const std::vector> record_ends;
const auto buffer = [&](std::string_view text) {
- rendered.append(text);
+ AppendRecords(rendered, text);
if constexpr (kPublishRecords) {
- record_ends.push_back(rendered.size());
+ record_ends.push_back(RecordsView(rendered).size());
}
};
const auto sink = ctx.emit;
@@ -2241,7 +2270,7 @@ absl::StatusOr> RenderGrepLines(
if (ctx.content.Loaded()) {
MBO_ASSIGN_OR_RETURN(const auto text, ctx.content.Read(ctx.fs, ctx.visit.path));
if (text.substr(0, static_cast(std::min(text.size(), sniff))).contains('\0')) {
- return GrepLineResult{};
+ return false;
}
content::VisitLines(text, visit);
} else {
@@ -2249,22 +2278,17 @@ absl::StatusOr> RenderGrepLines(
MBO_ASSIGN_OR_RETURN(const auto stream, source->Open());
MBO_ASSIGN_OR_RETURN(const auto scan, content::ScanLines(*stream, sniff, visit));
if (scan.binary) {
- return GrepLineResult{};
+ return false;
}
}
if constexpr (kPublishRecords) {
std::size_t start = 0;
for (const std::size_t end : record_ends) {
- sink(std::string_view(rendered).substr(start, end - start));
+ sink(RecordsView(rendered).substr(start, end - start));
start = end;
}
- return matched;
- } else {
- return RenderedRgOutput{
- .selected = matched,
- .text = std::move(rendered),
- };
}
+ return matched;
}
bool EvalGrepMatchers(const parser::Expr& expr, EvalContext& ctx, const GrepMatchers& matcher) {
@@ -2275,7 +2299,8 @@ bool EvalGrepMatchers(const parser::Expr& expr, EvalContext& ctx, const GrepMatc
const auto summary = SummarizeGrep(ctx, matcher, content::kBinaryNulSniffBytes);
return summary.ok() && EmitGrepSummary(expr, ctx, *summary);
}
- return RenderGrepLines(expr, ctx, matcher, content::kBinaryNulSniffBytes).value_or(false);
+ std::string records;
+ return RenderGrepLinesInto(expr, ctx, matcher, content::kBinaryNulSniffBytes, records).value_or(false);
}
bool EvalGrep(const parser::Expr& expr, EvalContext& ctx) {
@@ -4895,23 +4920,23 @@ absl::StatusOr EmitMultilineRg(const MatchOutput& output, const parser::Rg
return selected_lines != 0;
}
-absl::StatusOr RenderRgRecords(
+template
+absl::StatusOr RenderRgRecordsInto(
const MatchOutput& output,
const parser::RgSearch& search,
EvalContext& context,
- std::string records) {
+ Buffer& records) {
+ ClearRecords(records);
if (context.visit.metadata.type != vfs::FileType::kRegular) {
- return RenderedRgOutput{};
+ return false;
}
if (!search.multiline && !ReducedGrepOutput(context)) {
const GrepMatchers matcher(output.matchers, search);
- return RenderGrepLines(
- *output.source, context, matcher, search.text ? 0 : std::numeric_limits::max(),
- std::move(records));
+ return RenderGrepLinesInto(
+ *output.source, context, matcher, search.text ? 0 : std::numeric_limits::max(), records);
}
- records.clear();
const auto sink = context.emit;
- const auto buffer = [&](std::string_view text) { records.append(text); };
+ const auto buffer = [&](std::string_view text) { AppendRecords(records, text); };
context.emit = buffer;
const auto result = [&] -> absl::StatusOr {
if (search.multiline) {
@@ -4924,34 +4949,78 @@ absl::StatusOr RenderRgRecords(
return EmitGrepSummary(*output.source, context, summary);
}();
context.emit = sink;
- MBO_ASSIGN_OR_RETURN(const bool selected, result);
- return RenderedRgOutput{
- .selected = selected,
- .text = std::move(records),
- };
+ return result;
}
-} // namespace
-absl::StatusOr RenderRgOutput(
+template
+absl::StatusOr RenderRgOutputInto(
const MatchOutput& output,
const parser::RgSearch& search,
EvalContext& context,
- std::string records) {
+ Buffer& records) {
const bool filename = context.grep.filename;
const bool heading =
context.grep.heading && filename && !context.grep_json && context.grep.output == GrepOptions::Output::kLines;
if (heading) {
context.grep.filename = false;
}
- auto result = RenderRgRecords(output, search, context, std::move(records));
+ const auto result = RenderRgRecordsInto(output, search, context, records);
context.grep.filename = filename;
- MBO_ASSIGN_OR_RETURN(auto rendered, std::move(result));
- if (heading && !rendered.text.empty()) {
- rendered.text.insert(
- 0, absl::StrCat(
- context.grep.color ? "\x1b[35m" : "", context.visit.path, context.grep.color ? "\x1b[0m" : "", "\n"));
+ MBO_ASSIGN_OR_RETURN(const bool selected, result);
+ if (heading && !RecordsView(records).empty()) {
+ InsertHeading(
+ records,
+ absl::StrCat(
+ context.grep.color ? "\x1b[35m" : "", context.visit.path, context.grep.color ? "\x1b[0m" : "", "\n"));
+ }
+ return selected;
+}
+} // namespace
+
+absl::StatusOr RenderRgOutput(
+ const MatchOutput& output,
+ const parser::RgSearch& search,
+ EvalContext& context,
+ std::string records) {
+ MBO_ASSIGN_OR_RETURN(const bool selected, RenderRgOutputInto(output, search, context, records));
+ if (!selected) {
+ records.clear();
+ }
+ return RenderedRgOutput{
+ .selected = selected,
+ .text = std::move(records),
+ };
+}
+
+absl::StatusOr RenderRgOutput(
+ const MatchOutput& output,
+ const parser::RgSearch& search,
+ EvalContext& context,
+ CreditedOutput& records) {
+ const auto selected = RenderRgOutputInto(output, search, context, records);
+ if (!selected.ok() || !*selected) {
+ records.Clear();
+ }
+ return selected;
+}
+
+absl::StatusOr RenderMatchOutput(const MatchOutput& output, EvalContext& context, CreditedOutput& records) {
+ records.Clear();
+ if (ReducedGrepOutput(context)) {
+ const auto sink = context.emit;
+ const auto buffer = [&](std::string_view text) { records.Append(text); };
+ context.emit = buffer;
+ const auto summary = SummarizeGrep(context, GrepMatchers(output.matchers), content::kBinaryNulSniffBytes);
+ const bool selected = summary.ok() && EmitGrepSummary(*output.source, context, *summary);
+ context.emit = sink;
+ return selected;
+ }
+ const auto selected = RenderGrepLinesInto(
+ *output.source, context, GrepMatchers(output.matchers), content::kBinaryNulSniffBytes, records);
+ if (!selected.ok() || !*selected) {
+ records.Clear();
}
- return rendered;
+ return selected;
}
absl::StatusOr EmitRgOutput(const MatchOutput& output, const parser::RgSearch& search, EvalContext& context) {
diff --git a/xff/engine/evaluate.h b/xff/engine/evaluate.h
index 108ddf74bf9..7486266450c 100644
--- a/xff/engine/evaluate.h
+++ b/xff/engine/evaluate.h
@@ -479,6 +479,8 @@ absl::StatusOr PrepareRgOutput(
bool fold_case,
bool smart_case);
+class CreditedOutput;
+
struct RenderedRgOutput {
bool selected = false;
std::string text;
@@ -492,6 +494,14 @@ absl::StatusOr RenderRgOutput(
EvalContext& context,
std::string records = {});
+// Pipeline transactions reserve their actual output allocation before growth.
+absl::StatusOr RenderRgOutput(
+ const MatchOutput& output,
+ const parser::RgSearch& search,
+ EvalContext& context,
+ CreditedOutput& records);
+absl::StatusOr RenderMatchOutput(const MatchOutput& output, EvalContext& context, CreditedOutput& records);
+
absl::StatusOr EmitRgOutput(const MatchOutput& output, const parser::RgSearch& search, EvalContext& context);
// Validates the --diff-ignore token list and the --diff-ignore-matching regex, returning the
diff --git a/xff/engine/output_credit.cc b/xff/engine/output_credit.cc
new file mode 100644
index 00000000000..b6ab635f9f7
--- /dev/null
+++ b/xff/engine/output_credit.cc
@@ -0,0 +1,199 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "xff/engine/output_credit.h"
+
+#include
+#include
+#include
+
+namespace xff::engine {
+
+void OutputCredit::Reset() {
+ const std::scoped_lock lock(mutex_);
+ usage_ = {};
+ head_.store(0);
+ cancelled_.store(false, std::memory_order_relaxed);
+}
+
+bool OutputCredit::Acquire(std::size_t index, std::size_t bytes, mbo::types::OptionalRef cpu) {
+ if (bytes == 0) {
+ return true;
+ }
+ std::unique_lock lock(mutex_);
+ const auto available = [&] {
+ return cancelled_.load() || index == head_.load()
+ || (bytes <= kSharedBytes && usage_.excess <= kSharedBytes - bytes);
+ };
+ bool paused = false;
+ if (!available()) {
+ // The coordinator needs a CPU unit to return output credit. Never park while
+ // holding that unit; reacquire it only after dropping the credit mutex.
+ if (cpu) {
+ cpu->Pause();
+ paused = true;
+ }
+ blocked_.fetch_add(1);
+ if (progress_) {
+ // A coordinator batching publication must wake and drain the ordered head
+ // when credit fills. Do not nest the credit and progress mutexes.
+ lock.unlock();
+ { const absl::MutexLock notify(*progress_); }
+ lock.lock();
+ }
+ condition_.wait(lock, available);
+ blocked_.fetch_sub(1);
+ }
+ const bool admitted = !cancelled_.load();
+ if (admitted) {
+ usage_.excess += bytes;
+ usage_.peak_excess = std::max(usage_.peak_excess, usage_.excess);
+ }
+ lock.unlock();
+ if (paused) {
+ cpu->Resume();
+ }
+ return admitted;
+}
+
+void OutputCredit::Wake() {
+ if (blocked_.load() != 0) {
+ // Serialize with a producer's predicate check and release-to-wait. A producer
+ // registering later observes the already advanced head/cancellation instead.
+ { const std::scoped_lock lock(mutex_); }
+ condition_.notify_all();
+ }
+}
+
+void OutputCredit::Release(std::size_t bytes) {
+ if (bytes != 0) {
+ {
+ const std::scoped_lock lock(mutex_);
+ usage_.excess -= bytes;
+ }
+ if (blocked_.load() != 0) {
+ condition_.notify_all();
+ }
+ }
+}
+
+void OutputCredit::Advance(std::size_t next) {
+ head_.store(next);
+ Wake();
+}
+
+void OutputCredit::Cancel() {
+ cancelled_.store(true);
+ Wake();
+}
+
+OutputCredit::Usage OutputCredit::GetUsage() const {
+ const std::scoped_lock lock(mutex_);
+ return {
+ .excess = usage_.excess,
+ .peak_excess = usage_.peak_excess,
+ .waiters = blocked_.load(),
+ };
+}
+
+// XFF_ABI_POINTER: std::unique_ptr's required deleter for raw operator-new storage.
+void CreditedOutput::StorageDeleter::operator()(char* storage) const noexcept {
+ ::operator delete(storage);
+}
+
+CreditedOutput::~CreditedOutput() {
+ heap_.reset();
+ credit_.Release(excess_);
+}
+
+CreditedOutput::CreditedOutput(CreditedOutput&& other) noexcept
+ : credit_(other.credit_),
+ index_(other.index_),
+ heap_(std::move(other.heap_)),
+ capacity_(std::exchange(other.capacity_, kInlineBytes)),
+ size_(std::exchange(other.size_, 0)),
+ excess_(std::exchange(other.excess_, 0)),
+ cancelled_(other.cancelled_),
+ cpu_(other.cpu_) {
+ other.cpu_.reset();
+ if (!heap_) {
+ std::ranges::copy(std::span(other.inline_).first(size_), inline_.begin());
+ }
+}
+
+std::span CreditedOutput::Storage() {
+ return heap_ ? std::span(heap_.get(), capacity_) : std::span(inline_);
+}
+
+std::string_view CreditedOutput::View() const {
+ return heap_ ? std::string_view(heap_.get(), size_) : std::string_view(inline_.data(), size_);
+}
+
+bool CreditedOutput::Grow(std::size_t size) {
+ if (cancelled_) {
+ return false;
+ }
+ if (size <= capacity_) {
+ return true;
+ }
+ const auto next = std::max(size, 2 * capacity_);
+ // The grant covers both old and new allocations. Inline bytes live in the slot
+ // and are already charged as slot storage, rather than heap output credit.
+ const auto old = heap_ ? capacity_ : 0;
+ const auto combined = old + next;
+ const auto combined_excess = combined > OutputCredit::kSlotGrant ? combined - OutputCredit::kSlotGrant : 0;
+ const auto additional = combined_excess - excess_;
+ if (!credit_.Acquire(index_, additional, cpu_)) {
+ cancelled_ = true;
+ size_ = 0;
+ return false;
+ }
+ // XFF_ABI_POINTER: consume operator new's storage with unique_ptr's character-pointer ABI.
+ auto replacement = HeapStorage(static_cast(::operator new(next)));
+ std::ranges::copy(Storage().first(size_), std::span(replacement.get(), next).begin());
+ heap_ = std::move(replacement);
+ capacity_ = next;
+ const auto retained = next > OutputCredit::kSlotGrant ? next - OutputCredit::kSlotGrant : 0;
+ credit_.Release(combined_excess - retained);
+ excess_ = retained;
+ return true;
+}
+
+void CreditedOutput::Release() {
+ size_ = 0;
+ cancelled_ = false;
+ // Keep only the allocation already covered by this slot's initial grant.
+ // Larger transactions return every excess byte before the ordered head advances.
+ if (excess_ != 0) {
+ heap_.reset();
+ capacity_ = kInlineBytes;
+ credit_.Release(std::exchange(excess_, 0));
+ }
+}
+
+void CreditedOutput::Begin(std::size_t index) {
+ Release();
+ index_ = index;
+}
+
+void CreditedOutput::Append(std::string_view text) {
+ if (!text.empty() && Grow(size_ + text.size())) {
+ std::ranges::copy(text, Storage().subspan(size_).begin());
+ size_ += text.size();
+ }
+}
+
+void CreditedOutput::Prepend(std::string_view text) {
+ if (!text.empty() && Grow(size_ + text.size())) {
+ const auto storage = Storage();
+ std::ranges::copy_backward(storage.first(size_), storage.subspan(0, size_ + text.size()).end());
+ std::ranges::copy(text, storage.begin());
+ size_ += text.size();
+ }
+}
+
+void CreditedOutput::Clear() {
+ size_ = 0;
+}
+
+} // namespace xff::engine
diff --git a/xff/engine/output_credit.h b/xff/engine/output_credit.h
new file mode 100644
index 00000000000..84d5fe0f811
--- /dev/null
+++ b/xff/engine/output_credit.h
@@ -0,0 +1,129 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#ifndef XFF_ENGINE_OUTPUT_CREDIT_H_
+#define XFF_ENGINE_OUTPUT_CREDIT_H_
+
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+
+#include "absl/synchronization/mutex.h"
+#include "mbo/types/optional_ref.h"
+#include "xff/engine/cpu_allowance.h"
+
+namespace xff::engine {
+
+// Per-slot initial grants cover ordinary records without synchronized accounting.
+// Excess allocations share a byte allowance. Only the ordered head may exceed it,
+// so a late-validating transaction always has room to finish. The allowance is
+// retained output storage, not a bound on active source/decoder or process memory.
+class OutputCredit final {
+ public:
+ static constexpr std::size_t kSlotGrant = 4'096;
+ static constexpr std::size_t kSharedBytes = 8'388'608;
+
+ OutputCredit() = default;
+
+ explicit OutputCredit(absl::Mutex& progress) : progress_(progress) {}
+
+ struct Usage {
+ std::size_t excess = 0;
+ std::size_t peak_excess = 0;
+ std::size_t waiters = 0;
+ };
+
+ void Reset();
+ bool Acquire(std::size_t index, std::size_t bytes, mbo::types::OptionalRef cpu = {});
+ void Release(std::size_t bytes);
+ void Advance(std::size_t next);
+ void Cancel();
+ Usage GetUsage() const;
+
+ bool Blocked() const { return blocked_.load() != 0; }
+
+ private:
+ void Wake();
+
+ mutable std::mutex mutex_;
+ Usage usage_;
+ std::atomic head_ = 0;
+ std::atomic blocked_ = 0;
+ std::condition_variable condition_;
+ mbo::types::OptionalRef progress_;
+ std::atomic cancelled_ = false;
+};
+
+// One worker owns this contiguous transaction until release-publishing completion.
+// The coordinator borrows its view, then releases storage before advancing the head.
+// Old and new allocations both hold credit during growth. Nothing is truncated or
+// published before read/binary validation. Cancellation discards subsequent appends.
+class CreditedOutput final {
+ public:
+ static constexpr std::size_t kInlineBytes = 64;
+
+ CreditedOutput(OutputCredit& credit, std::size_t index) : credit_(credit), index_(index) {}
+
+ ~CreditedOutput();
+ CreditedOutput(const CreditedOutput&) = delete;
+ CreditedOutput& operator=(const CreditedOutput&) = delete;
+ CreditedOutput(CreditedOutput&& other) noexcept;
+ CreditedOutput& operator=(CreditedOutput&&) = delete;
+
+ void Begin(std::size_t index);
+
+ // Bound only while the producer renders; cleared before completion publication.
+ void BindCpu(mbo::types::OptionalRef cpu) {
+ if (cpu) {
+ cpu_.set_ref(*cpu);
+ } else {
+ cpu_.reset();
+ }
+ }
+
+ void Release();
+ void Append(std::string_view text);
+ void Prepend(std::string_view text);
+ void Clear();
+
+ bool Empty() const { return size_ == 0; }
+
+ std::size_t Size() const { return size_; }
+
+ std::size_t Capacity() const { return capacity_; }
+
+ bool Cancelled() const { return cancelled_; }
+
+ std::string_view View() const;
+
+ private:
+ bool Grow(std::size_t size);
+ std::span Storage();
+
+ OutputCredit& credit_;
+ std::size_t index_;
+ // The bytes are initialized only by append/prepend; do not zero the full slot array.
+ std::array inline_;
+
+ struct StorageDeleter {
+ // XFF_ABI_POINTER: std::unique_ptr's required deleter for raw operator-new storage.
+ void operator()(char* storage) const noexcept;
+ };
+
+ using HeapStorage = std::unique_ptr;
+ HeapStorage heap_;
+ std::size_t capacity_ = kInlineBytes;
+ std::size_t size_ = 0;
+ std::size_t excess_ = 0;
+ bool cancelled_ = false;
+ mbo::types::OptionalRef cpu_;
+};
+
+} // namespace xff::engine
+
+#endif // XFF_ENGINE_OUTPUT_CREDIT_H_
diff --git a/xff/engine/output_credit_test.cc b/xff/engine/output_credit_test.cc
new file mode 100644
index 00000000000..76905ad991e
--- /dev/null
+++ b/xff/engine/output_credit_test.cc
@@ -0,0 +1,251 @@
+// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "xff/engine/output_credit.h"
+
+#include
+#include
+#include
+#include
+#include
+#include
+#include
+
+#include "absl/synchronization/mutex.h"
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+#include "mbo/testing/matchers.h"
+#include "xff/engine/cpu_allowance.h"
+
+namespace xff::engine {
+namespace {
+
+using ::mbo::testing::EqualsText;
+using ::testing::Eq;
+using ::testing::Ge;
+using ::testing::IsFalse;
+using ::testing::IsTrue;
+
+struct OutputCreditTest : ::testing::Test {
+ static void CheckCpuCreditWake(bool cancel) {
+ OutputCredit credit;
+ CpuAllowance cpu(1);
+ EXPECT_THAT(credit.Acquire(2, OutputCredit::kSharedBytes), IsTrue());
+ bool admitted = false;
+ std::atomic finished = false;
+ std::jthread producer([&] {
+ CpuAllowance::Lease lease(cpu);
+ admitted = credit.Acquire(1, 1, lease);
+ finished.store(true);
+ });
+ EXPECT_THAT(AwaitBlocked(credit), IsTrue());
+ std::optional coordinator;
+ coordinator.emplace(cpu);
+ if (cancel) {
+ credit.Cancel();
+ } else {
+ credit.Advance(1);
+ }
+ const auto end = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while (cpu.GetUsage().waiters == 0 && std::chrono::steady_clock::now() < end) {
+ std::this_thread::yield();
+ }
+ EXPECT_THAT(cpu.GetUsage().waiters, Eq(1));
+ EXPECT_THAT(finished.load(), IsFalse());
+ // A resumed producer cannot retain the credit mutex while waiting for CPU.
+ EXPECT_THAT(credit.GetUsage().excess, Eq(OutputCredit::kSharedBytes + (cancel ? 0UZ : 1UZ)));
+ coordinator.reset();
+ producer.join();
+ EXPECT_THAT(finished.load(), IsTrue());
+ if (cancel) {
+ EXPECT_THAT(admitted, IsFalse());
+ } else {
+ EXPECT_THAT(admitted, IsTrue());
+ }
+ credit.Release(OutputCredit::kSharedBytes + (cancel ? 0UZ : 1UZ));
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+ EXPECT_THAT(cpu.GetUsage().available, Eq(1));
+ }
+
+ static bool AwaitBlocked(const OutputCredit& credit) {
+ const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while (credit.GetUsage().waiters == 0 && std::chrono::steady_clock::now() < deadline) {
+ std::this_thread::yield();
+ }
+ return credit.GetUsage().waiters != 0;
+ }
+};
+
+TEST_F(OutputCreditTest, BlockedOutputReturnsCpuAndReacquiresItAfterDroppingTheCreditMutex) {
+ CheckCpuCreditWake(false);
+}
+
+TEST_F(OutputCreditTest, CancelledOutputReacquiresCpuBeforeDrainingTheProducer) {
+ CheckCpuCreditWake(true);
+}
+
+TEST_F(OutputCreditTest, ContiguousGrowthAndHeadingPreserveTextAndReleaseEveryAllocation) {
+ OutputCredit credit;
+ {
+ CreditedOutput output(credit, 0);
+ output.Append("selected line\n");
+ output.Prepend("heading\n");
+ EXPECT_THAT(output.View(), EqualsText("heading\nselected line\n"));
+ const std::string text(4 * OutputCredit::kSlotGrant, 'x');
+ output.Append(text);
+ EXPECT_THAT(output.View(), EqualsText("heading\nselected line\n" + text));
+ EXPECT_THAT(credit.GetUsage().excess, Ge(text.size() - OutputCredit::kSlotGrant));
+ EXPECT_THAT(credit.GetUsage().peak_excess, Ge(credit.GetUsage().excess));
+ CreditedOutput moved(std::move(output));
+ EXPECT_THAT(moved.View(), EqualsText("heading\nselected line\n" + text));
+ // NOLINTNEXTLINE(bugprone-use-after-move,clang-analyzer-cplusplus.Move): defined reusable source contract.
+ output.Append("reused");
+ EXPECT_THAT(output.View(), EqualsText("reused"));
+ moved.Clear();
+ EXPECT_THAT(moved.Empty(), IsTrue());
+ }
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+}
+
+TEST_F(OutputCreditTest, EmptyAndCancelledTransactionsRetainNoPublishedText) {
+ OutputCredit credit;
+ CreditedOutput output(credit, 1);
+ output.Append("");
+ output.Prepend("");
+ EXPECT_THAT(output.Empty(), IsTrue());
+ credit.Cancel();
+ output.Append(std::string(OutputCredit::kSlotGrant * 2, 'x'));
+ EXPECT_THAT(output.Cancelled(), IsTrue());
+ output.Append("discarded");
+ output.Prepend("discarded heading");
+ EXPECT_THAT(output.Empty(), IsTrue());
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+}
+
+TEST_F(OutputCreditTest, HeapGrowthReservesOldAndNewCapacityUntilTheOldAllocationIsReleased) {
+ OutputCredit credit;
+ {
+ CreditedOutput output(credit, 0);
+ const std::string text(OutputCredit::kSlotGrant, 'x');
+ output.Append(text);
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+ output.Prepend("heading\n");
+ EXPECT_THAT(output.View(), EqualsText("heading\n" + text));
+ EXPECT_THAT(credit.GetUsage().excess, Eq(output.Capacity() - OutputCredit::kSlotGrant));
+ EXPECT_THAT(credit.GetUsage().peak_excess, Eq(output.Capacity()));
+ }
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+}
+
+TEST_F(OutputCreditTest, ReuseKeepsOnlyHeapStorageCoveredByTheSlotGrant) {
+ OutputCredit credit;
+ CreditedOutput output(credit, 0);
+ output.Append(std::string(512, 'x'));
+ const auto capacity = output.Capacity();
+ output.Release();
+ EXPECT_THAT(output.Empty(), IsTrue());
+ EXPECT_THAT(output.Capacity(), Eq(capacity));
+ credit.Advance(1);
+ output.Begin(1);
+ output.Append("new transaction");
+ EXPECT_THAT(output.View(), EqualsText("new transaction"));
+ output.Append(std::string(2 * OutputCredit::kSlotGrant, 'y'));
+ output.Release();
+ EXPECT_THAT(output.Capacity(), Eq(CreditedOutput::kInlineBytes));
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+}
+
+TEST_F(OutputCreditTest, InlineMovesAndReusedSlotsDoNotCopyUninitializedBytes) {
+ OutputCredit credit;
+ CreditedOutput output(credit, 0);
+ output.Append("test");
+ const CreditedOutput moved(std::move(output));
+ EXPECT_THAT(moved.View(), EqualsText("test"));
+ // NOLINTNEXTLINE(bugprone-use-after-move,clang-analyzer-cplusplus.Move): defined empty source contract.
+ EXPECT_THAT(output.Empty(), IsTrue());
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+}
+
+TEST_F(OutputCreditTest, OrderedHeadCanFinishWhenLaterTransactionsExhaustSharedCredit) {
+ OutputCredit credit;
+ EXPECT_THAT(credit.Acquire(1, OutputCredit::kSharedBytes), IsTrue());
+ // A later entry cannot take another byte, but head admission must never wait.
+ EXPECT_THAT(credit.Acquire(0, 2 * OutputCredit::kSharedBytes), IsTrue());
+ EXPECT_THAT(credit.GetUsage().peak_excess, Eq(3 * OutputCredit::kSharedBytes));
+ credit.Release(2 * OutputCredit::kSharedBytes);
+ credit.Release(OutputCredit::kSharedBytes);
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+}
+
+TEST_F(OutputCreditTest, AdvancingTheHeadWakesAnOversizedLaterTransaction) {
+ OutputCredit credit;
+ std::atomic entered = false;
+ bool admitted = false;
+ std::jthread producer([&] {
+ entered.store(true);
+ entered.notify_one();
+ admitted = credit.Acquire(1, 2 * OutputCredit::kSharedBytes);
+ });
+ entered.wait(false);
+ EXPECT_THAT(AwaitBlocked(credit), IsTrue());
+ credit.Advance(1);
+ producer.join();
+ EXPECT_THAT(admitted, IsTrue());
+ credit.Release(2 * OutputCredit::kSharedBytes);
+}
+
+TEST_F(OutputCreditTest, BackpressureWakesARegisteredPublicationWindow) {
+ absl::Mutex progress;
+ OutputCredit credit(progress);
+ std::atomic entered = false;
+ std::atomic finished = false;
+ std::atomic stop = false;
+ bool observed = false;
+ std::jthread coordinator([&] {
+ const absl::MutexLock lock(progress);
+ entered.store(true);
+ entered.notify_one();
+ const auto ready = [&] { return credit.Blocked() || stop.load(); };
+ progress.Await(absl::Condition(&ready));
+ observed = credit.Blocked();
+ finished.store(true);
+ });
+ entered.wait(false);
+ bool admitted = true;
+ std::jthread producer([&] { admitted = credit.Acquire(1, 2 * OutputCredit::kSharedBytes); });
+ const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while (!finished.load() && std::chrono::steady_clock::now() < deadline) {
+ std::this_thread::yield();
+ }
+ EXPECT_THAT(finished.load(), IsTrue());
+ stop.store(true);
+ { const absl::MutexLock notify(progress); }
+ credit.Cancel();
+ coordinator.join();
+ producer.join();
+ EXPECT_THAT(observed, IsTrue());
+ EXPECT_THAT(admitted, IsFalse());
+}
+
+TEST_F(OutputCreditTest, CancellationWakesBlockedReservationsAndResetAllowsFreshWork) {
+ OutputCredit credit;
+ std::atomic entered = false;
+ bool admitted = true;
+ std::jthread producer([&] {
+ entered.store(true);
+ entered.notify_one();
+ admitted = credit.Acquire(1, 2 * OutputCredit::kSharedBytes);
+ });
+ entered.wait(false);
+ EXPECT_THAT(AwaitBlocked(credit), IsTrue());
+ credit.Cancel();
+ producer.join();
+ EXPECT_THAT(admitted, IsFalse());
+ credit.Reset();
+ EXPECT_THAT(credit.Acquire(1, 4), IsTrue());
+ credit.Release(4);
+ EXPECT_THAT(credit.GetUsage().excess, Eq(0));
+}
+
+} // namespace
+} // namespace xff::engine
diff --git a/xff/engine/owned_visit.h b/xff/engine/owned_visit.h
index 4a1d01b8031..cfdab368b17 100644
--- a/xff/engine/owned_visit.h
+++ b/xff/engine/owned_visit.h
@@ -35,6 +35,10 @@ class OwnedVisit final {
std::string_view Root() const;
// Logical lengths remain unchanged even when their physical byte ranges overlap.
std::size_t LogicalTextBytes() const;
+
+ // Retained text capacity, including the terminator and inline storage.
+ std::size_t TextStorageBytes() const { return bytes_.capacity() + 1; }
+
Visit AsVisit() const;
private:
diff --git a/xff/engine/parallel_match.cc b/xff/engine/parallel_match.cc
index 0397629d855..dae71e18cad 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
@@ -43,9 +44,12 @@ ParallelMatch::ParallelMatch(
scores_(scores),
output_(std::move(output)),
execution_(execution),
- derivation_(std::move(derivation)) {}
+ derivation_(std::move(derivation)),
+ output_credit_(mutex_),
+ cpu_allowance_(workers_) {}
ParallelMatch::~ParallelMatch() {
+ Cancel();
{
const absl::MutexLock lock(mutex_);
stop_ = true;
@@ -68,9 +72,47 @@ bool ParallelMatch::Finished() const {
}
const std::vector& ParallelMatch::Match(std::vector entries) {
+ Start(std::move(entries));
+ return Wait();
+}
+
+void ParallelMatch::Cancel() {
+ cancelled_.store(true, std::memory_order_relaxed);
+ output_credit_.Cancel();
+ const absl::MutexLock lock(mutex_);
+ mutex_.Await(absl::Condition(this, &ParallelMatch::Finished));
+ if (credited_output_ && !coordinator_entries_) {
+ // Skipped entries have no producer to publish readiness. Wake an ordered
+ // consumer only after all active reads and result writes have drained.
+ for (auto& completion : ready_) {
+ for (auto& state : completion.states) {
+ state.store(kReady, std::memory_order_release);
+ }
+ }
+ }
+}
+
+const std::vector& ParallelMatch::Wait() {
+ coordinator_cpu_.reset();
const absl::MutexLock lock(mutex_);
- entries_ = std::move(entries);
+ mutex_.Await(absl::Condition(this, &ParallelMatch::Finished));
+ return results_;
+}
+
+void ParallelMatch::PrepareBatch(std::vector& entries) {
+ cancelled_.store(false, std::memory_order_relaxed);
+ entries_.clear();
+ entries_.swap(entries);
results_.resize(entries_.size());
+ for (auto& slot : output_slots_) {
+ slot.Release();
+ }
+ if (coordinator_output_) {
+ coordinator_output_->Release();
+ }
+ credited_output_ = false;
+ observed_output_bytes_ = 0;
+ coordinator_entries_ = false;
if (!derivation_.Empty()) {
derived_.clear();
derived_.resize(entries_.size());
@@ -86,16 +128,42 @@ const std::vector& ParallelMatch::Match(std::vector
}
}
next_.store(0, std::memory_order_relaxed);
+}
+
+std::vector ParallelMatch::Start(std::vector entries, bool pipeline) {
+ coordinator_cpu_.reset();
+ const absl::MutexLock lock(mutex_, absl::Condition(this, &ParallelMatch::Finished));
+ PrepareBatch(entries);
// Very small batches cannot amortize waking the pool.
if (workers_ == 1 || entries_.size() < 16 || (threads_.empty() && entries_.size() < 64)) {
if (execution_ && !coordinator_) {
coordinator_.emplace(execution_->MakeWorker(ExpressionWorkerRole::kCoordinator));
}
+ if (pipeline && output_) {
+ // There is no worker to unblock a serial renderer. Defer each transaction
+ // until consumption, then release it before evaluating the next entry.
+ output_credit_.Reset();
+ credited_output_ = true;
+ coordinator_entries_ = true;
+ return entries;
+ }
EvaluateEntries(
coordinator_fuzzy_workspace_, {}, {},
coordinator_ ? mbo::types::OptionalRef{*coordinator_}
: mbo::types::OptionalRef{});
- return results_;
+ return entries;
+ }
+ output_credit_.Reset();
+ credited_output_ = pipeline && output_.has_value();
+ if (credited_output_) {
+ ready_.resize((entries_.size() + kChunkSize - 1) / kChunkSize);
+ output_slots_.reserve(std::max(output_slots_.size(), entries_.size()));
+ for (std::size_t index = 0; index < entries_.size(); ++index) {
+ CompletionState(index).store(kPending, std::memory_order_relaxed);
+ if (index == output_slots_.size()) {
+ output_slots_.emplace_back(output_credit_, index);
+ }
+ }
}
const std::size_t count = std::min(workers_, (entries_.size() + 15) / 16);
// An early partial batch must not permanently cap the pool for later full batches.
@@ -109,9 +177,85 @@ const std::vector& ParallelMatch::Match(std::vector
// workers beyond the number of available chunks are redundant for a later short tail.
active_workers_ = std::min(threads_.size(), (entries_.size() + kChunkSize - 1) / kChunkSize);
remaining_ = active_workers_;
+ if (pipeline && CollectNextBatch()) {
+ coordinator_cpu_.emplace(cpu_allowance_);
+ }
++generation_;
- mutex_.Await(absl::Condition(this, &ParallelMatch::Finished));
- return results_;
+ return entries;
+}
+
+const ParallelResult& ParallelMatch::AwaitResult(std::size_t index) {
+ if (coordinator_entries_) {
+ if (!cancelled_.load(std::memory_order_acquire)) {
+ if (!coordinator_cpu_) {
+ coordinator_cpu_.emplace(cpu_allowance_);
+ }
+ EvaluateEntry(
+ index, coordinator_fuzzy_workspace_, {}, {},
+ coordinator_ ? mbo::types::OptionalRef{*coordinator_}
+ : mbo::types::OptionalRef{},
+ *coordinator_cpu_);
+ }
+ } else if (credited_output_) {
+ const auto await = [&](std::size_t target, bool allow_backpressure) {
+ auto& state = CompletionState(target);
+ if (state.load(std::memory_order_acquire) != kReady) {
+ coordinator_cpu_.reset();
+ auto expected = kPending;
+ // A publication window may already have registered this same target.
+ static_cast(state.compare_exchange_strong(expected, kWaiting, std::memory_order_acq_rel));
+ const auto completed = [&] {
+ return state.load(std::memory_order_acquire) == kReady || (allow_backpressure && output_credit_.Blocked());
+ };
+ const absl::MutexLock lock(mutex_, absl::Condition(&completed));
+ }
+ };
+ // Amortize wakeups for ordinary short records. If an arbitrary large later
+ // record needs credit, bypass the window and drain the current head instead.
+ static constexpr std::size_t kPublicationWindow = 64;
+ const auto window = CollectNextBatch() ? kPublicationWindow : entries_.size();
+ if (index % window == 0) {
+ await(std::min(index + window, entries_.size()) - 1, true);
+ }
+ await(index, false);
+ if (!coordinator_cpu_) {
+ coordinator_cpu_.emplace(cpu_allowance_);
+ }
+ } else if (index == 0) {
+ Wait();
+ }
+ if (cancelled_.load(std::memory_order_acquire)) {
+ coordinator_cpu_.reset();
+ static const ParallelResult kCancelled{.evaluation = {.matched = false}};
+ return kCancelled;
+ }
+ return results_.at(index);
+}
+
+void ParallelMatch::ReleaseResult(std::size_t index) {
+ if (credited_output_) {
+ if (auto& content = results_.at(index).content; content) {
+ observed_output_bytes_ +=
+ std::min(content->Text().size(), std::numeric_limits::max() - observed_output_bytes_);
+ content->credited_text.reset();
+ content->selected = true;
+ }
+ if (coordinator_entries_) {
+ if (coordinator_output_) {
+ coordinator_output_->Release();
+ }
+ } else {
+ output_slots_.at(index).Release();
+ }
+ output_credit_.Advance(index + 1);
+ if (index + 1 == entries_.size()) {
+ // A 256-entry batch with at least 1 KiB of output per visited entry has
+ // enough publication work to try overlap. This threshold is experimental;
+ // it does not inspect file metadata or introduce a clock into the hot path.
+ static constexpr std::size_t kOutputBytesPerEntry = 1'024;
+ overlap_useful_ = observed_output_bytes_ / entries_.size() >= kOutputBytesPerEntry;
+ }
+ }
}
void ParallelMatch::Run(std::size_t worker) {
@@ -132,13 +276,20 @@ void ParallelMatch::Run(std::size_t worker) {
}
generation = generation_;
}
- EvaluateEntries(
- fuzzy_workspace,
- local ? mbo::types::OptionalRef>{*local}
- : mbo::types::OptionalRef>{},
- matchers,
- evaluator ? mbo::types::OptionalRef{*evaluator}
- : mbo::types::OptionalRef{});
+ {
+ std::optional cpu;
+ if (credited_output_) {
+ cpu.emplace(cpu_allowance_);
+ }
+ EvaluateEntries(
+ fuzzy_workspace,
+ local ? mbo::types::OptionalRef>{*local}
+ : mbo::types::OptionalRef>{},
+ matchers,
+ evaluator ? mbo::types::OptionalRef{*evaluator}
+ : mbo::types::OptionalRef{},
+ cpu ? mbo::types::OptionalRef{*cpu} : mbo::types::OptionalRef{});
+ }
{
const absl::MutexLock lock(mutex_);
--remaining_;
@@ -150,30 +301,92 @@ void ParallelMatch::EvaluateEntries(
fuzzy::ScoreWorkspace& workspace,
mbo::types::OptionalRef> output,
mbo::types::OptionalRef matchers,
- mbo::types::OptionalRef evaluator) {
+ mbo::types::OptionalRef evaluator,
+ mbo::types::OptionalRef cpu) {
// Small chunks spread clustered expensive entries while amortizing atomic scheduling.
for (;;) {
const std::size_t first = next_.fetch_add(kChunkSize, std::memory_order_relaxed);
- if (first >= entries_.size()) {
+ if (first >= entries_.size() || cancelled_.load(std::memory_order_relaxed)) {
return;
}
const std::size_t end = std::min(first + kChunkSize, entries_.size());
for (std::size_t index = first; index < end; ++index) {
- EvaluateEntry(index, workspace, output, matchers, evaluator);
+ EvaluateEntry(index, workspace, output, matchers, evaluator, cpu);
+ if (credited_output_) {
+ auto& state = CompletionState(index);
+ if (state.exchange(kReady, std::memory_order_release) == kWaiting) {
+ // Acquiring/releasing the condition's mutex prevents a lost wake if the
+ // coordinator is between registering its wait and actually parking.
+ const absl::MutexLock lock(mutex_);
+ }
+ }
}
}
}
+void ParallelMatch::RenderEntryOutput(
+ std::size_t index,
+ const ParallelContentOutput& specification,
+ EvalContext& context,
+ mbo::types::OptionalRef> output,
+ mbo::types::OptionalRef cpu) {
+ auto& result = results_.at(index);
+ auto& content = result.content ? *result.content : result.content.emplace();
+ content.selected = true;
+ // Forking is an optimization: retain the already validated shared matcher if it fails.
+ const auto& prepared = output && output->ok() ? **output : specification.output;
+ const auto sink = context.emit;
+ const auto buffer = [&](std::string_view text) { content.text.append(text); };
+ context.emit = buffer;
+ context.grep = specification.grep;
+ context.grep_json = specification.json;
+ context.grep_before = specification.before;
+ context.grep_after = specification.after;
+ if (specification.automatic_filename) {
+ context.grep.filename = specification.root_count != 1 || context.visit.depth != 0;
+ }
+ if (credited_output_) {
+ auto& records = coordinator_entries_ ? (coordinator_output_ ? *coordinator_output_
+ : coordinator_output_.emplace(output_credit_, index))
+ : output_slots_.at(index);
+ if (coordinator_entries_) {
+ records.Begin(index);
+ }
+ records.BindCpu(cpu);
+ if (specification.rg) {
+ content.selected = RenderRgOutput(prepared, *specification.rg, context, records);
+ } else {
+ static_cast(RenderMatchOutput(prepared, context, records));
+ }
+ content.credited_text = records.View();
+ records.BindCpu({});
+ } else if (specification.rg) {
+ auto rendered = RenderRgOutput(prepared, *specification.rg, context, std::move(content.text));
+ if (rendered.ok()) {
+ content.selected = rendered->selected;
+ content.text = std::move(rendered->text);
+ } else {
+ content.selected = rendered.status();
+ content.text.clear();
+ }
+ } else {
+ EmitMatchOutput(prepared, context);
+ }
+ context.emit = sink;
+}
+
void ParallelMatch::EvaluateEntry(
std::size_t index,
fuzzy::ScoreWorkspace& workspace,
mbo::types::OptionalRef> output,
mbo::types::OptionalRef matchers,
- mbo::types::OptionalRef evaluator) {
+ mbo::types::OptionalRef evaluator,
+ mbo::types::OptionalRef cpu) {
const Visit visit = entries_.at(index).AsVisit();
auto& result = results_.at(index);
if (result.content) {
result.content->text.clear();
+ result.content->credited_text.reset();
}
Control control;
std::optional fuzzy;
@@ -193,31 +406,7 @@ void ParallelMatch::EvaluateEntry(
: expression_ ? EvaluateDeferred(*expression_, context)
: EvaluationResult{.matched = true};
if (result.evaluation.matched && output_) {
- auto& content = result.content ? *result.content : result.content.emplace();
- content.selected = true;
- // Forking is an optimization: retain the already validated shared matcher if it fails.
- const auto& prepared = output && output->ok() ? **output : output_->output;
- const auto buffer = [&](std::string_view text) { content.text.append(text); };
- context.emit = buffer;
- context.grep = output_->grep;
- context.grep_json = output_->json;
- context.grep_before = output_->before;
- context.grep_after = output_->after;
- if (output_->automatic_filename) {
- context.grep.filename = output_->root_count != 1 || visit.depth != 0;
- }
- if (output_->rg) {
- auto rendered = RenderRgOutput(prepared, *output_->rg, context, std::move(content.text));
- if (rendered.ok()) {
- content.selected = rendered->selected;
- content.text = std::move(rendered->text);
- } else {
- content.selected = rendered.status();
- content.text.clear();
- }
- } else {
- EmitMatchOutput(prepared, context);
- }
+ RenderEntryOutput(index, *output_, context, output, cpu);
} else {
result.content.reset();
}
diff --git a/xff/engine/parallel_match.h b/xff/engine/parallel_match.h
index d3e2165ab71..83c62cc993e 100644
--- a/xff/engine/parallel_match.h
+++ b/xff/engine/parallel_match.h
@@ -4,8 +4,10 @@
#ifndef XFF_ENGINE_PARALLEL_MATCH_H_
#define XFF_ENGINE_PARALLEL_MATCH_H_
+#include
#include
#include
+#include
#include
#include
#include
@@ -18,8 +20,10 @@
#include "absl/synchronization/mutex.h"
#include "mbo/types/optional_ref.h"
#include "xff/content/snapshot.h"
+#include "xff/engine/cpu_allowance.h"
#include "xff/engine/evaluate.h"
#include "xff/engine/expression_execution.h"
+#include "xff/engine/output_credit.h"
#include "xff/engine/owned_visit.h"
#include "xff/fuzzy/fuzzy.h"
#include "xff/parser/ast.h"
@@ -28,7 +32,7 @@ namespace xff::engine {
// Workers evaluate only audited predicates and prepared content searches. They have no host
// output/mutation sinks, captures, or result-set state. The caller emits buffered records in
-// entry order after the batch completes. Source bytes die with each worker's evaluator.
+// entry order after each entry completes. Source bytes die with each worker's evaluator.
struct ParallelContentOutput {
const MatchOutput& output;
mbo::types::OptionalRef rg;
@@ -43,6 +47,9 @@ struct ParallelContentOutput {
struct ContentResult {
absl::StatusOr selected = true;
std::string text;
+ std::optional credited_text;
+
+ std::string_view Text() const { return credited_text.value_or(text); }
};
struct ParallelResult {
@@ -76,6 +83,26 @@ class ParallelMatch final {
// owning entries remain available for ordered output until the next call.
const std::vector& Match(std::vector entries);
+ // Start one batch on the persistent workers. The coordinator can collect the next
+ // batch until ordered consumption establishes the publication barrier. Returns the
+ // consumed predecessor's empty input storage for collection reuse. Credited small
+ // batches evaluate one entry at a time on that coordinator; Match() remains synchronous.
+ std::vector Start(std::vector entries, bool pipeline = false);
+ const std::vector& Wait();
+ // Stop claiming entries and drain every worker before releasing the batch's owners.
+ // A subsequent Start() begins a fresh generation.
+ void Cancel();
+ // Each entry remains producer-owned until AwaitResult() returns. ReleaseResult()
+ // ends the borrowed output view and advances the head before the next wait.
+ const ParallelResult& AwaitResult(std::size_t index);
+ void ReleaseResult(std::size_t index);
+
+ OutputCredit::Usage OutputUsage() const { return output_credit_.GetUsage(); }
+
+ // Observed output controls scheduling only; both policies retain the same byte
+ // credit and whole-file publication rules. Unknown output starts without overlap.
+ bool CollectNextBatch() const { return output_ && overlap_useful_; }
+
const std::vector& Entries() const { return entries_; }
// Consume one ordered entry's derived caches. No snapshot storage exists without a request.
@@ -83,6 +110,13 @@ class ParallelMatch final {
private:
static constexpr std::size_t kChunkSize = 4;
+ void PrepareBatch(std::vector& entries) ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+ void RenderEntryOutput(
+ std::size_t index,
+ const ParallelContentOutput& specification,
+ EvalContext& context,
+ mbo::types::OptionalRef> output,
+ mbo::types::OptionalRef cpu);
void Run(std::size_t worker);
bool Ready(std::size_t generation, std::size_t worker) const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
bool Finished() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
@@ -90,14 +124,16 @@ class ParallelMatch final {
fuzzy::ScoreWorkspace& workspace,
mbo::types::OptionalRef> output = {},
mbo::types::OptionalRef matchers = {},
- mbo::types::OptionalRef evaluator = {});
- // Each worker exclusively owns the claimed result slot until this batch completes.
+ mbo::types::OptionalRef evaluator = {},
+ mbo::types::OptionalRef cpu = {});
+ // Each worker exclusively owns its claimed slot until publishing its completion.
void EvaluateEntry(
std::size_t index,
fuzzy::ScoreWorkspace& workspace,
mbo::types::OptionalRef> output,
mbo::types::OptionalRef matchers,
- mbo::types::OptionalRef evaluator);
+ mbo::types::OptionalRef evaluator,
+ mbo::types::OptionalRef cpu);
const mbo::types::OptionalRef expression_;
const std::size_t workers_;
@@ -119,6 +155,33 @@ class ParallelMatch final {
std::vector results_;
std::vector derived_;
std::atomic next_ = 0;
+ std::atomic cancelled_ = false;
+ OutputCredit output_credit_;
+ CpuAllowance cpu_allowance_;
+ // Owned only by the coordinator. Cancel() must not mutate another thread's lease.
+ std::optional coordinator_cpu_;
+ std::vector output_slots_;
+ std::optional coordinator_output_;
+ bool coordinator_entries_ = false;
+ static constexpr unsigned char kPending = 0;
+ static constexpr unsigned char kWaiting = 1;
+ static constexpr unsigned char kReady = 2;
+
+ // A claimed chunk has exactly one producer. Keep different chunks on separate
+ // cache lines while avoiding a whole cache line per entry.
+ struct alignas(128) Completion {
+ std::array, kChunkSize> states;
+ };
+
+ std::atomic& CompletionState(std::size_t index) {
+ return ready_.at(index / kChunkSize).states.at(index % kChunkSize);
+ }
+
+ std::deque ready_;
+ bool credited_output_ = false;
+ // Coordinator-owned observations; workers never read or mutate these counters.
+ std::size_t observed_output_bytes_ = 0;
+ bool overlap_useful_ = false;
};
} // namespace xff::engine
diff --git a/xff/engine/parallel_match_test.cc b/xff/engine/parallel_match_test.cc
index be37a46797a..eac790c80da 100644
--- a/xff/engine/parallel_match_test.cc
+++ b/xff/engine/parallel_match_test.cc
@@ -24,6 +24,7 @@ namespace {
using ::mbo::testing::EqualsText;
using ::testing::Eq;
+using ::testing::Ge;
using ::testing::IsFalse;
using ::testing::IsTrue;
using ::testing::SizeIs;
@@ -63,6 +64,49 @@ TEST_F(ParallelMatchTest, RepeatedBatchesRetainEntryOrderAndDoNotReuseOldResults
}
}
+TEST_F(ParallelMatchTest, AsynchronousBatchesPreserveOrderedResultsAndCanRestartAfterCancellation) {
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-type", "f"}));
+ ParallelMatch matcher(*command.expression, 4, false);
+ // Without rendered output there is no observed-output signal for stage overlap.
+ EXPECT_THAT(matcher.CollectNextBatch(), IsFalse());
+ for (const std::size_t count : {1, 128, 16, 0, 1'024, 7}) {
+ matcher.Start(Entries(count));
+ const auto& results = matcher.Wait();
+ ASSERT_THAT(results, SizeIs(count));
+ for (std::size_t index = 0; index < count; ++index) {
+ EXPECT_THAT(results.at(index).evaluation.matched, Eq(index % 2 == 0));
+ EXPECT_THAT(matcher.Entries().at(index).Path(), EqualsText(std::to_string(index)));
+ }
+ matcher.Start(Entries(1'024));
+ matcher.Cancel();
+ }
+ // Destruction drains a still-running generation; owned filesystem observers stay live.
+ matcher.Start(Entries(1'024));
+}
+
+TEST_F(ParallelMatchTest, StartingTheNextGenerationReturnsEmptyReusableInputCapacity) {
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-type", "f"}));
+ ParallelMatch matcher(*command.expression, 4, false);
+ auto spare = matcher.Start(Entries(128), true);
+ EXPECT_THAT(spare, SizeIs(0));
+ matcher.Wait();
+ spare = matcher.Start(Entries(64), true);
+ EXPECT_THAT(spare, SizeIs(0));
+ EXPECT_THAT(spare.capacity(), Ge(128));
+ matcher.Wait();
+ // Populate the returned storage rather than rebuilding a fresh vector.
+ for (std::size_t index = 0; index < 128; ++index) {
+ spare.emplace_back(Visit{.path = std::to_string(index), .metadata = {.type = vfs::FileType::kRegular}, .fs = fs});
+ }
+ const auto previous = matcher.Start(std::move(spare), true);
+ EXPECT_THAT(previous, SizeIs(0));
+ EXPECT_THAT(previous.capacity(), Ge(64));
+ EXPECT_THAT(matcher.Wait(), SizeIs(128));
+ for (const auto& result : matcher.Wait()) {
+ EXPECT_THAT(result.evaluation.matched, IsTrue());
+ }
+}
+
TEST_F(ParallelMatchTest, ZeroWorkersIsClampedAndUnusedPoolStartsNoWork) {
MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"root", "-true"}));
const ParallelMatch unused(*command.expression, 4, false);
diff --git a/xff/engine/rg_test.cc b/xff/engine/rg_test.cc
index b2e5bf0ae6a..23f21b0b3e5 100644
--- a/xff/engine/rg_test.cc
+++ b/xff/engine/rg_test.cc
@@ -35,6 +35,7 @@
#include "mbo/testing/matchers.h"
#include "mbo/testing/status.h"
#include "xff/engine/evaluate.h"
+#include "xff/engine/output_credit.h"
#include "xff/engine/owned_visit.h"
#include "xff/engine/parallel_match.h"
#include "xff/engine/run.h"
@@ -103,7 +104,9 @@ struct SearchFs final : vfs::FileSystem {
std::string read_error_path;
bool streaming = false;
vfs::Source entry_source = vfs::Source::kLocalFs;
+ std::string serial_entry;
vfs::SharedReadSource source_override;
+ std::function before_read;
bool track_threads = false;
std::size_t expected_threads = 2;
mutable std::mutex read_mutex;
@@ -138,7 +141,7 @@ struct SearchFs final : vfs::FileSystem {
.path = name,
.name = name.substr(5),
.type = vfs::FileType::kRegular,
- .source = entry_source,
+ .source = name == serial_entry ? vfs::Source::kArchiveMember : entry_source,
});
}
}
@@ -158,7 +161,7 @@ struct SearchFs final : vfs::FileSystem {
}
return vfs::Metadata{
.type = vfs::FileType::kRegular,
- .source = entry_source,
+ .source = path == serial_entry ? vfs::Source::kArchiveMember : entry_source,
.size = found->second.size(),
};
}
@@ -197,6 +200,9 @@ struct SearchFs final : vfs::FileSystem {
}
absl::StatusOr ReadContent(std::string_view path) const override {
+ if (before_read) {
+ before_read(path);
+ }
TrackRead();
reads.fetch_add(1, std::memory_order_relaxed);
if (read_error || path == read_error_path) {
@@ -843,6 +849,335 @@ TEST_F(RgEngineTest, DerivedPoolSlotsClearAcrossUnmatchedEmptyAndInlineBatches)
EXPECT_THAT(fs.reads.load(), Eq(258));
}
+TEST_F(RgEngineTest, InputByteAdmissionPreservesLongPathsAcrossPartialBatches) {
+ fs.files.clear();
+ const std::string tail(5'000, 'x');
+ for (std::size_t index = 0; index < 300; ++index) {
+ fs.files.emplace("tree/" + std::to_string(index) + tail, "hit\n");
+ }
+ EXPECT_THAT(Run({"--jobs=1", "--archive=none", "tree", "-content", "hit"}, false).errors, Eq(0));
+ const auto expected = output;
+ EXPECT_THAT(Run({"--jobs=4", "--archive=none", "tree", "-content", "hit"}, false).errors, Eq(0));
+ EXPECT_THAT(output, EqualsText(expected));
+ fs.files.clear();
+ fs.files.emplace("tree/" + std::string(1'100'000, 'x'), "hit\n");
+ EXPECT_THAT(Run({"--jobs=4", "--archive=none", "tree", "-content", "hit"}, false).errors, Eq(0));
+ EXPECT_THAT(output, EqualsText(fs.files.begin()->first + "\n"));
+}
+
+TEST_F(RgEngineTest, ObservedOutputSelectsOverlapAndReturnsToBatchPublication) {
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"--rg", "hit", "tree"}));
+ MBO_ASSERT_OK_AND_ASSIGN(const auto prepared, PrepareRgOutput(command, fs, false, false));
+ const auto search = command.rg.value_or(parser::RgSearch{});
+ ParallelMatch matcher(
+ {}, 4, false,
+ ParallelContentOutput{
+ .output = prepared,
+ .rg = search,
+ .grep =
+ {
+ .line_number = false,
+ .filename = false,
+ },
+ });
+ EXPECT_THAT(matcher.CollectNextBatch(), IsFalse());
+ // Cross inline/worker and small/heavy boundaries on reused transactions.
+ constexpr auto kBatches = std::to_array>({
+ {128, 4},
+ {10, 2'048},
+ {64, 4},
+ {128, 8'192},
+ {7, 4},
+ });
+ for (const auto& [count, size] : kBatches) {
+ fs.files.at("tree/a") = std::string(size - 4, 'x') + "hit\n";
+ matcher.Start(RepeatedEntries(count), true);
+ for (std::size_t index = 0; index < count; ++index) {
+ const auto& result = matcher.AwaitResult(index);
+ EXPECT_THAT(result.content, Optional(Property(&ContentResult::Text, EqualsText(fs.files.at("tree/a")))));
+ matcher.ReleaseResult(index);
+ }
+ matcher.Wait();
+ EXPECT_THAT(matcher.CollectNextBatch(), Eq(size >= 1'024));
+ EXPECT_THAT(matcher.OutputUsage().excess, Eq(0));
+ }
+}
+
+TEST_F(RgEngineTest, MixedOutputBatchesPreserveNativeAndRgRecordsAcrossSchedulingChanges) {
+ fs.files.clear();
+ for (std::size_t index = 0; index < 768; ++index) {
+ const auto prefix = std::to_string(1'000 + index);
+ fs.files.emplace("tree/" + prefix, index >= 256 && index < 512 ? std::string(8'192, 'x') + "hit\n" : "hit\n");
+ }
+ EXPECT_THAT(Run({"--jobs=1", "--archive=none", "-M", "tree", "-content", "hit"}, false).errors, Eq(0));
+ const auto native = output;
+ EXPECT_THAT(Run({"--jobs=4", "--archive=none", "-M", "tree", "-content", "hit"}, false).errors, Eq(0));
+ EXPECT_THAT(output, EqualsText(native));
+ EXPECT_THAT(Run({"--rg", "--jobs=1", "--archive=none", "hit", "tree"}, false).errors, Eq(0));
+ const auto rg = output;
+ EXPECT_THAT(Run({"--rg", "--jobs=4", "--archive=none", "hit", "tree"}, false).errors, Eq(0));
+ EXPECT_THAT(output, EqualsText(rg));
+}
+
+TEST_F(RgEngineTest, AdaptivePipelineDrainsBeforeTraversalErrors) {
+ fs.files.clear();
+ for (std::size_t index = 0; index < 768; ++index) {
+ fs.files.emplace("tree/" + std::to_string(1'000 + index), std::string(8'192, 'x') + "hit\n");
+ }
+ EXPECT_THAT(Run({"--jobs=1", "--archive=none", "hit", "tree", "missing"}).errors, Eq(1));
+ const auto expected = output;
+ bool emitted_after_error = false;
+ after_emit = [&] { emitted_after_error |= !errors.empty(); };
+ EXPECT_THAT(Run({"--jobs=4", "--archive=none", "hit", "tree", "missing"}).errors, Eq(1));
+ EXPECT_THAT(output, EqualsText(expected));
+ EXPECT_THAT(errors, SizeIs(1));
+ EXPECT_THAT(emitted_after_error, IsFalse());
+}
+
+TEST_F(RgEngineTest, AdaptivePipelinePreservesRecordsAcrossSerialArchiveMemberBoundaries) {
+ fs.files.clear();
+ for (std::size_t index = 0; index < 768; ++index) {
+ fs.files.emplace("tree/" + std::to_string(1'000 + index), std::string(8'192, 'x') + "hit\n");
+ }
+ // This source kind cannot enter the matcher pool. Force that real dispatch
+ // boundary after the pipeline has observed enough output to overlap batches.
+ fs.serial_entry = "tree/1512-serial";
+ fs.files.emplace(fs.serial_entry, "serial hit\n");
+ EXPECT_THAT(Run({"--jobs=1", "--archive=none", "hit", "tree"}).errors, Eq(0));
+ const auto expected = output;
+ EXPECT_THAT(Run({"--jobs=4", "--archive=none", "hit", "tree"}).errors, Eq(0));
+ EXPECT_THAT(output, EqualsText(expected));
+}
+
+TEST_F(RgEngineTest, CreditedInlineEntriesReleaseEachTransactionBeforeReadingTheNext) {
+ fs.files.at("tree/a") = std::string(2'097'152, 'x') + "hit\n";
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"--rg", "hit", "tree"}));
+ MBO_ASSERT_OK_AND_ASSIGN(const auto prepared, PrepareRgOutput(command, fs, false, false));
+ const auto search = command.rg.value_or(parser::RgSearch{});
+ ParallelMatch matcher(
+ {}, 4, false,
+ ParallelContentOutput{
+ .output = prepared,
+ .rg = search,
+ .grep =
+ {
+ .line_number = false,
+ .filename = false,
+ },
+ });
+ fs.track_threads = true;
+ fs.expected_threads = 1;
+ matcher.Start(RepeatedEntries(10), true);
+ EXPECT_THAT(fs.reads.load(), Eq(0));
+ for (std::size_t index = 0; index < 10; ++index) {
+ const auto& result = matcher.AwaitResult(index);
+ EXPECT_THAT(result.content, Optional(Property(&ContentResult::Text, EqualsText(fs.files.at("tree/a")))));
+ EXPECT_THAT(fs.reads.load(), Eq(index + 1));
+ matcher.ReleaseResult(index);
+ EXPECT_THAT(matcher.OutputUsage().excess, Eq(0));
+ }
+ matcher.Wait();
+ EXPECT_THAT(fs.read_threads, ElementsAre(std::this_thread::get_id()));
+}
+
+TEST_F(RgEngineTest, CreditedTransactionsProgressPastDelayedHeadAndFullLaterOutput) {
+ fs.files.clear();
+ const std::string ordinary = std::string(65'536, 'x') + "hit\n";
+ const std::string giant = std::string(2'097'152, 'x') + "hit\n";
+ std::vector entries;
+ static constexpr std::size_t kEntries = 256;
+ entries.reserve(kEntries);
+ for (std::size_t index = 0; index < kEntries; ++index) {
+ const auto path = "tree/" + std::to_string(index);
+ fs.files.emplace(path, index == 0 ? giant : ordinary);
+ entries.emplace_back(Visit{.path = path, .metadata = {.type = vfs::FileType::kRegular}, .fs = fs});
+ }
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"--rg", "hit", "tree"}));
+ MBO_ASSERT_OK_AND_ASSIGN(const auto prepared, PrepareRgOutput(command, fs, false, false));
+ const auto search = command.rg.value_or(parser::RgSearch{});
+ ParallelMatch matcher(
+ {}, 4, false,
+ ParallelContentOutput{
+ .output = prepared,
+ .rg = search,
+ .grep =
+ {
+ .line_number = false,
+ .filename = false,
+ },
+ });
+ std::mutex mutex;
+ std::condition_variable condition;
+ bool released = false;
+ bool timed_out = false;
+ fs.before_read = [&](std::string_view path) {
+ if (path == "tree/0") {
+ std::unique_lock lock(mutex);
+ timed_out = !condition.wait_for(lock, std::chrono::seconds(5), [&] { return released; });
+ }
+ };
+ matcher.Start(std::move(entries), true);
+ const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while ((matcher.OutputUsage().waiters == 0 || matcher.OutputUsage().excess < OutputCredit::kSharedBytes - 131'072)
+ && std::chrono::steady_clock::now() < deadline) {
+ std::this_thread::yield();
+ }
+ EXPECT_THAT(matcher.OutputUsage().waiters, Gt(0));
+ EXPECT_THAT(matcher.OutputUsage().excess, Le(OutputCredit::kSharedBytes));
+ {
+ const std::scoped_lock lock(mutex);
+ released = true;
+ condition.notify_all();
+ }
+ for (std::size_t index = 0; index < kEntries; ++index) {
+ const auto& result = matcher.AwaitResult(index);
+ EXPECT_THAT(
+ result.content, Optional(AllOf(
+ Field("selected", &ContentResult::selected, IsOkAndHolds(IsTrue())),
+ Property(&ContentResult::Text, EqualsText(index == 0 ? giant : ordinary)))));
+ matcher.ReleaseResult(index);
+ }
+ matcher.Wait();
+ EXPECT_THAT(timed_out, IsFalse());
+ EXPECT_THAT(matcher.OutputUsage().excess, Eq(0));
+ EXPECT_THAT(matcher.OutputUsage().peak_excess, Gt(OutputCredit::kSharedBytes));
+}
+
+TEST_F(RgEngineTest, CancellationDrainsCreditWaitersAndUnstartedResultsBeforePoolReuse) {
+ fs.files.at("tree/a") = std::string(262'144, 'x') + "hit\n";
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"--rg", "hit", "tree"}));
+ MBO_ASSERT_OK_AND_ASSIGN(const auto prepared, PrepareRgOutput(command, fs, false, false));
+ const auto search = command.rg.value_or(parser::RgSearch{});
+ ParallelMatch matcher(
+ {}, 4, false,
+ ParallelContentOutput{
+ .output = prepared,
+ .rg = search,
+ .grep =
+ {
+ .line_number = false,
+ .filename = false,
+ },
+ });
+ matcher.Start(RepeatedEntries(128), true);
+ const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while (matcher.OutputUsage().waiters == 0 && std::chrono::steady_clock::now() < deadline) {
+ std::this_thread::yield();
+ }
+ EXPECT_THAT(matcher.OutputUsage().waiters, Gt(0));
+ matcher.Cancel();
+ EXPECT_THAT(matcher.AwaitResult(127).evaluation.matched, IsFalse());
+ EXPECT_THAT(matcher.OutputUsage().waiters, Eq(0));
+ fs.files.at("tree/a") = "hit\n";
+ matcher.Start(RepeatedEntries(64), true);
+ for (std::size_t index = 0; index < 64; ++index) {
+ const auto& result = matcher.AwaitResult(index);
+ EXPECT_THAT(result.content, Optional(Property(&ContentResult::Text, EqualsText("hit\n"))));
+ matcher.ReleaseResult(index);
+ }
+ matcher.Wait();
+ EXPECT_THAT(matcher.OutputUsage().excess, Eq(0));
+}
+
+TEST_F(RgEngineTest, CancellationCanDrainWhileTheCoordinatorOwnsOneCpuUnit) {
+ fs.files.at("tree/a") = std::string(262'144, 'x') + "hit\n";
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"--rg", "hit", "tree"}));
+ MBO_ASSERT_OK_AND_ASSIGN(const auto prepared, PrepareRgOutput(command, fs, false, false));
+ const auto search = command.rg.value_or(parser::RgSearch{});
+ ParallelMatch matcher(
+ {}, 2, false,
+ ParallelContentOutput{
+ .output = prepared,
+ .rg = search,
+ .grep =
+ {
+ .line_number = false,
+ .filename = false,
+ },
+ });
+ matcher.Start(RepeatedEntries(64), true);
+ for (std::size_t index = 0; index < 64; ++index) {
+ EXPECT_THAT(
+ matcher.AwaitResult(index).content,
+ Optional(Property(&ContentResult::Text, EqualsText(fs.files.at("tree/a")))));
+ matcher.ReleaseResult(index);
+ }
+ matcher.Wait();
+ EXPECT_THAT(matcher.CollectNextBatch(), IsTrue());
+ matcher.Start(RepeatedEntries(128), true);
+ const auto end = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+ while (matcher.OutputUsage().waiters == 0 && std::chrono::steady_clock::now() < end) {
+ std::this_thread::yield();
+ }
+ EXPECT_THAT(matcher.OutputUsage().waiters, Gt(0));
+ std::jthread cancel([&] { matcher.Cancel(); });
+ cancel.join();
+ EXPECT_THAT(matcher.AwaitResult(127).evaluation.matched, IsFalse());
+ EXPECT_THAT(matcher.OutputUsage().waiters, Eq(0));
+ fs.files.at("tree/a") = "hit\n";
+ matcher.Start(RepeatedEntries(64), true);
+ for (std::size_t index = 0; index < 64; ++index) {
+ EXPECT_THAT(matcher.AwaitResult(index).content, Optional(Property(&ContentResult::Text, EqualsText("hit\n"))));
+ matcher.ReleaseResult(index);
+ }
+ matcher.Wait();
+ EXPECT_THAT(matcher.OutputUsage().excess, Eq(0));
+}
+
+TEST_F(RgEngineTest, AsyncDestructionDrainsWorkersBeforeReleasingFilesystemOwners) {
+ auto owner = std::make_shared();
+ const std::weak_ptr weak = owner;
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"tree", "-content", "hit"}));
+ {
+ ParallelMatch matcher(*command.expression, 4, false);
+ std::vector entries;
+ entries.reserve(128);
+ for (std::size_t index = 0; index < 128; ++index) {
+ entries.emplace_back(
+ Visit{
+ .path = "tree/a",
+ .metadata = {.type = vfs::FileType::kRegular},
+ .fs = *owner,
+ .fs_owner = owner,
+ });
+ }
+ matcher.Start(std::move(entries));
+ owner.reset();
+ EXPECT_THAT(weak.expired(), IsFalse());
+ }
+ EXPECT_THAT(weak.expired(), IsTrue());
+}
+
+TEST_F(RgEngineTest, AsyncStartReturnsWhileAnOwnedReadIsStillBlocked) {
+ MBO_ASSERT_OK_AND_ASSIGN(const auto command, parser::Parse({"tree", "-content", "hit"}));
+ ParallelMatch matcher(*command.expression, 4, false);
+ std::mutex mutex;
+ std::condition_variable condition;
+ bool started = false;
+ bool released = false;
+ bool timed_out = false;
+ fs.before_read = [&](std::string_view) {
+ std::unique_lock lock(mutex);
+ started = true;
+ condition.notify_all();
+ timed_out |= !condition.wait_for(lock, std::chrono::seconds(2), [&] { return released; });
+ };
+ matcher.Start(RepeatedEntries(128));
+ {
+ std::unique_lock lock(mutex);
+ EXPECT_THAT(condition.wait_for(lock, std::chrono::seconds(2), [&] { return started; }), IsTrue());
+ released = true;
+ condition.notify_all();
+ }
+ const auto& results = matcher.Wait();
+ EXPECT_THAT(timed_out, IsFalse());
+ EXPECT_THAT(results, SizeIs(128));
+ for (const auto& result : results) {
+ EXPECT_THAT(result.evaluation.matched, IsTrue());
+ }
+ EXPECT_THAT(fs.reads.load(), Eq(128));
+}
+
TEST_F(RgEngineTest, MatcherPoolGrowsForLaterBatchesAndRunsShortTailsInline) {
fs.files.clear();
for (std::size_t index = 0; index < 256; ++index) {
diff --git a/xff/engine/run.cc b/xff/engine/run.cc
index de106300048..8d024cedaab 100644
--- a/xff/engine/run.cc
+++ b/xff/engine/run.cc
@@ -5981,7 +5981,7 @@ RunResult RunFindCore(
}
const auto selected = prepared ? prepared->selected : EmitRgOutput(*rg_output, *command.rg, context);
if (prepared && selected.ok()) {
- context.emit(prepared->text);
+ context.emit(prepared->Text());
}
content = std::move(context.content);
if (!selected.ok()) {
@@ -6084,7 +6084,7 @@ RunResult RunFindCore(
.content = std::move(content),
};
if (prepared) {
- emit(prepared->text);
+ emit(prepared->Text());
} else {
EmitMatchOutput(*match_output, context);
}
@@ -6266,14 +6266,18 @@ RunResult RunFindCore(
// Decision-only batches retain no file bytes or rendered line records. Give them more work
// between coordinator barriers; keep content-output batches smaller because a line is unbounded.
const std::size_t match_batch_size = rg_output || match_output ? 256 : 1'024;
- const auto flush_matches = [&] {
- if (pending_matches.empty()) {
+ // One collecting batch overlaps a matcher generation. Slot counts stay fixed;
+ // text admission includes the owned capacity, with one oversized input alone.
+ static constexpr std::size_t kPendingMatchTextBytes = 1'048'576;
+ std::size_t pending_match_bytes = 0;
+ bool matches_running = false;
+ const auto publish_matches = [&] {
+ if (!matches_running) {
return;
}
- const auto& results = parallel_match->Match(std::move(pending_matches));
- pending_matches.clear();
- for (std::size_t index = 0; index < results.size(); ++index) {
- const auto& result = results.at(index);
+ matches_running = false;
+ for (std::size_t index = 0; index < parallel_match->Entries().size(); ++index) {
+ const auto& result = parallel_match->AwaitResult(index);
const auto& evaluated = result.evaluation;
const Visit visit = parallel_match->Entries().at(index).AsVisit();
fuzzy_score = evaluated.fuzzy;
@@ -6299,6 +6303,29 @@ RunResult RunFindCore(
visit, outputs, evaluated.matched, std::nullopt, parallel_match->TakeDerived(index),
result.content ? mbo::types::OptionalRef{*result.content}
: mbo::types::OptionalRef{});
+ parallel_match->ReleaseResult(index);
+ }
+ parallel_match->Wait();
+ };
+ const auto submit_matches = [&] {
+ publish_matches();
+ if (!pending_matches.empty()) {
+ pending_matches = parallel_match->Start(std::move(pending_matches), true);
+ pending_matches.clear();
+ pending_match_bytes = 0;
+ matches_running = true;
+ if (!parallel_match->CollectNextBatch()) {
+ publish_matches();
+ }
+ }
+ };
+ const auto flush_matches = [&] {
+ submit_matches();
+ publish_matches();
+ };
+ const auto reserve_match_slot = [&] {
+ if (pending_matches.size() == pending_matches.capacity()) {
+ pending_matches.reserve(std::min(match_batch_size, std::max(16UZ, 2 * pending_matches.capacity())));
}
};
const absl::Status status = Walk(
@@ -6382,9 +6409,22 @@ RunResult RunFindCore(
return WalkAction::kContinue;
}
if (parallel_entry) {
+ // The last constructed slot is a transit record until its actual text
+ // capacity is admitted. Only a full predecessor needs an ownership move.
+ reserve_match_slot();
pending_matches.emplace_back(visit);
- if (pending_matches.size() >= match_batch_size) {
- flush_matches();
+ const auto text_bytes = pending_matches.back().TextStorageBytes();
+ if (pending_matches.size() > 1
+ && text_bytes > kPendingMatchTextBytes - std::min(pending_match_bytes, kPendingMatchTextBytes)) {
+ OwnedVisit entry(std::move(pending_matches.back()));
+ pending_matches.pop_back();
+ submit_matches();
+ reserve_match_slot();
+ pending_matches.push_back(std::move(entry));
+ }
+ pending_match_bytes += text_bytes;
+ if (pending_matches.size() >= match_batch_size || pending_match_bytes >= kPendingMatchTextBytes) {
+ submit_matches();
}
return WalkAction::kContinue;
}