diff --git a/XFF.md b/XFF.md
index 6044130d008..e3bcef0551a 100644
--- a/XFF.md
+++ b/XFF.md
@@ -564,7 +564,7 @@ See also: [Configuration](#topic-config), [Archives](#topic-archive), [Output](#
- `-j N, -j=N, --jobs=N|all` - directory-read, eligible content-match, comparison and -exec workers _(global, find, xff, rg)_
- `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories using one command-owned, lazily started pool of up to `N-1` background workers plus the calling coordinator, reused for directory reads, required eager metadata, eligible content matching and file comparison. Comparison trees are coordinated sequentially within that same budget. The caller participates in eligible content and comparison work. Directory read-ahead uses a bounded ordered window and a shared retained-storage allowance; oversized speculative listings are read on demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight filesystem call must finish. Workers reuse the same executor for independent work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core.
+ `N` is a positive integer; use `-j 4`, `-j=4`, or `--jobs=4`. The conventional attached short form `-j4` is also accepted. Every form accepts `all` in place of `N`. `-j 1` makes directory reads and `-exec ... ;` synchronous. At larger values, xff reads directories using one command-owned, lazily started pool of up to `N-1` background workers plus the calling coordinator, reused for directory reads, required eager metadata, eligible content matching and file comparison. Comparison trees are coordinated sequentially within that same budget. The caller participates in directory reads, eligible content and comparison work. It reads an uncached current directory while workers read later siblings, and can claim a required queued read once. Directory read-ahead uses a bounded ordered window and a shared retained-storage allowance; oversized speculative listings are read on demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight filesystem call must finish. Workers reuse the same executor for independent work; small workloads stay on the coordinator. Lazy and metadata-free listings do not start stat workers. Independent content predicates and rg searches match in bounded parallel batches, with output kept in traversal order. Small initial batches, archive members, stateful expressions stay on the coordinator. Metadata-consuming output does not itself prevent eligible content filters from running in workers; reductions and rendering remain serial. Independent regular-file comparison pairs can also use bounded workers, preserving relative-path output order. Batches of only small or metadata-only comparisons stay inline; archive and owned filesystem sources stay serial. Xff may independently keep up to `N` semicolon-form `-exec` / `-execdir` children outstanding. Reads and children can overlap; `N` is not one shared operation budget. The children's truth value is therefore success on launch. The `... +` batch forms still run once after the walk and propagate a failing exit status. With no flag, xff uses one fewer than the detected cores, capped at 15 and floored at 1; find and rg modes use every detected core. `all` always means every detected core.
See also: [Output](#topic-output), [Statistics](#topic-stats)
diff --git a/docs/benchmark-correctness-audit.md b/docs/benchmark-correctness-audit.md
index 42c719c16e8..aa31c563182 100644
--- a/docs/benchmark-correctness-audit.md
+++ b/docs/benchmark-correctness-audit.md
@@ -352,6 +352,33 @@ not valid evidence of sufficient remaining work. The bookkeeping estimate, retai
byte policy and task granularity need allocation profiles and repeated native
measurements. No timing, total-memory improvement or macOS acceptance is claimed.
+### Caller-participation control
+
+A deterministic 65-sibling in-memory control identified a separate gap in the
+bounded-frontier candidate: with `--jobs=2`, only one background reader touched
+child directories, peak concurrent reads were one, and the two-directory
+rendezvous timed out. The caller never read a child directory. With larger
+budgets, the observed reader identities likewise excluded the caller. Ordered
+output, exact-once reads and the thread cap still passed; those properties alone
+did not demonstrate useful caller participation.
+
+The follow-up reserves the first uncached required directory for the caller and
+queues later siblings. A required queued read is represented by move-only work
+that either the caller or worker can claim once. An already-claimed read still
+requires waiting for its result; no arbitrary unrelated job is run on that wait.
+The command budget remains caller-inclusive, with at most `N-1` background
+threads, reused across both comparison inventories and pair comparisons.
+
+Controls verify caller/worker overlap for budgets 2/3/10, serial-equivalent
+ordered visits and exactly one read per child. An occupied-worker control forces
+the caller to read its required queued directory before releasing that worker.
+Executor controls cover queued claims, already-running claims and simultaneous
+caller/worker claims, including move-only producer ownership. Cancellation,
+shared byte/slot accounting, error reporting and borrowed-state lifetime remain
+part of the required validation. The extra claim state and callback allocation
+need native allocation and timing evidence. Remaining-cost admission and useful
+task size are unchanged; no performance or native macOS acceptance is claimed.
+
### Memory-accounting finding and remaining acceptance work
Direct `/usr/bin/time -v` runs show approximately 5.3-5.4 MiB peak RSS for a small
diff --git a/docs/design-parallel.md b/docs/design-parallel.md
index c4972fd41b7..335c46e0370 100644
--- a/docs/design-parallel.md
+++ b/docs/design-parallel.md
@@ -159,6 +159,14 @@ The caller claims independent stat, content and comparison chunks rather than
waiting idle while `N` additional workers run. Sequential comparison inventories
remove the second coordinator, but may reduce overlap when each tree has a
narrow frontier on slow storage; native A/B controls must quantify that tradeoff.
+For directory reads, the first uncached required directory stays on the caller,
+while the bounded frontier queues later siblings for background workers. A queued
+read that becomes required can be claimed by its caller or its dispatched worker,
+exactly once. The claim mutex protects callback ownership only; filesystem access
+and result publication happen after unlocking. If a worker already owns that
+required read, the caller waits for its result. It does not execute an unrelated
+slow sibling merely to help the queue. This improves available participation, not
+a guarantee of full utilization or sufficient useful work for each dispatch.
The bounded-frontier follow-up separates foreground stat/content/comparison jobs
from speculative directory reads. When both queues contain work, dispatch
alternates their classes, retaining FIFO order within each class. This is fairness
diff --git a/xff/cli/globals.cc b/xff/cli/globals.cc
index 6b5179e6efb..4d6c560356e 100644
--- a/xff/cli/globals.cc
+++ b/xff/cli/globals.cc
@@ -907,7 +907,9 @@ constexpr std::array kGlobals = std::to_array({
"using one command-owned, lazily started pool of up to `N-1` background workers plus the calling "
"coordinator, reused for directory reads, required eager metadata, eligible content matching "
"and file comparison. Comparison trees are coordinated sequentially within that same budget. "
- "The caller participates in eligible content and comparison work. Directory read-ahead uses a bounded "
+ "The caller participates in directory reads, eligible content and comparison work. It reads an uncached "
+ "current directory while workers read later siblings, and can claim a required queued read once. "
+ "Directory read-ahead uses a bounded "
"ordered window and a shared retained-storage allowance; oversized speculative listings are read on "
"demand instead. Unneeded queued reads are canceled after pruning or stopping, but an in-flight "
"filesystem call must finish. Workers reuse the same executor for independent work; small workloads "
diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel
index f9b3a913751..9e923681db1 100644
--- a/xff/engine/BUILD.bazel
+++ b/xff/engine/BUILD.bazel
@@ -171,6 +171,7 @@ cc_test(
":walk_cc",
"//xff/vfs:local_fs_cc",
"//xff/vfs:vfs_cc",
+ "@abseil-cpp//absl/base:core_headers",
"@abseil-cpp//absl/functional:function_ref",
"@abseil-cpp//absl/status",
"@abseil-cpp//absl/status:statusor",
diff --git a/xff/engine/command_workers_test.cc b/xff/engine/command_workers_test.cc
index ac94152fca4..6e38d0884b4 100644
--- a/xff/engine/command_workers_test.cc
+++ b/xff/engine/command_workers_test.cc
@@ -64,7 +64,7 @@ class CommandFs final : public vfs::FileSystem {
}
return entries;
}
- Record(path.starts_with("left/") ? 0 : 1, participants_ > 1 ? participants_ - 1 : 1);
+ Record(path.starts_with("left/") ? 0 : 1, participants_);
return std::vector{
{
.path = std::string(path) + "/file",
@@ -133,7 +133,8 @@ class CommandFs final : public vfs::FileSystem {
void CheckWalkReuse() const {
const absl::MutexLock lock(mutex_);
EXPECT_THAT(phase_readers_.at(0), Eq(phase_readers_.at(1)));
- EXPECT_THAT(phase_readers_.at(0), SizeIs(participants_ > 1 ? participants_ - 1 : 1));
+ EXPECT_THAT(phase_readers_.at(0), SizeIs(participants_));
+ EXPECT_THAT(phase_readers_.at(0), Contains(coordinator_));
}
private:
diff --git a/xff/engine/run_executor.cc b/xff/engine/run_executor.cc
index ce734f03d9a..6d5ffb5e316 100644
--- a/xff/engine/run_executor.cc
+++ b/xff/engine/run_executor.cc
@@ -22,6 +22,20 @@ struct ReadAheadBudget final {
};
} // namespace executor_detail
+RunWork::RunWork(absl::AnyInvocable work) : work_(std::move(work)) {}
+
+absl::AnyInvocable RunWork::Take() {
+ const absl::MutexLock lock(mutex_);
+ return std::exchange(work_, absl::AnyInvocable{});
+}
+
+void RunWork::Run() {
+ auto work = Take();
+ if (work) {
+ work();
+ }
+}
+
struct ReadAheadReservation::Impl final {
Impl(std::shared_ptr state, std::size_t initial_bytes)
: budget(std::move(state)), minimum_bytes(initial_bytes), bytes(initial_bytes) {}
diff --git a/xff/engine/run_executor.h b/xff/engine/run_executor.h
index 7a5ab66b1fd..490d0c795fd 100644
--- a/xff/engine/run_executor.h
+++ b/xff/engine/run_executor.h
@@ -34,6 +34,30 @@ struct TaskState final {
enum class RunTaskClass { kForeground, kReadAhead };
+// A leaf job claimable by its coordinator or a dispatched worker. Exactly one
+// Run call executes the owned job, outside its mutex; other calls return without
+// waiting. Result publication, if needed, remains the job's responsibility.
+// Use shared ownership while both caller and queued worker can access the job.
+class RunWork final {
+ public:
+ explicit RunWork(absl::AnyInvocable work);
+
+ RunWork(const RunWork&) = delete;
+ RunWork& operator=(const RunWork&) = delete;
+ RunWork(RunWork&&) = delete;
+ RunWork& operator=(RunWork&&) = delete;
+
+ void Run() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ private:
+ absl::AnyInvocable Take() ABSL_LOCKS_EXCLUDED(mutex_);
+
+ // This mutex guards ownership of work_ only and is released before invoking
+ // the job. It is never held with an executor dispatch or read-budget mutex.
+ absl::Mutex mutex_;
+ absl::AnyInvocable work_ ABSL_GUARDED_BY(mutex_);
+};
+
struct ReadAheadStats {
// Unfinished reads. Completed results keep their byte charge, not an execution slot.
std::size_t in_use = 0;
diff --git a/xff/engine/run_executor_test.cc b/xff/engine/run_executor_test.cc
index 10c8cf768f0..472e334f9bf 100644
--- a/xff/engine/run_executor_test.cc
+++ b/xff/engine/run_executor_test.cc
@@ -3,6 +3,7 @@
#include "xff/engine/run_executor.h"
+#include
#include
#include
#include
@@ -83,6 +84,112 @@ TEST_F(RunExecutorTest, SupportsMoveOnlyJobsAndResults) {
executor.Submit([] {}).Get();
}
+TEST_F(RunExecutorTest, CallerClaimsRequiredWorkWhileItsQueuedWrapperWaits) {
+ RunExecutor executor(1);
+ executor.Start(1);
+ absl::Notification started;
+ absl::Notification release;
+ auto occupied = executor.Submit([&] {
+ started.Notify();
+ release.WaitForNotification();
+ });
+ const bool running = started.WaitForNotificationWithTimeout(absl::Seconds(5));
+ std::size_t calls = 0;
+ RunPromise promise;
+ auto result = promise.Task();
+ const auto work =
+ std::make_shared([&calls, value = std::make_unique(42), promise = std::move(promise)] mutable {
+ ++calls;
+ promise.SetValue(*value);
+ });
+ auto queued = executor.Submit([work] { work->Run(); }, RunTaskClass::kReadAhead);
+ work->Run();
+ const bool ready_before_release = result.Ready();
+ release.Notify();
+ occupied.Get();
+ queued.Get();
+ work->Run();
+ EXPECT_THAT(running, IsTrue());
+ EXPECT_THAT(ready_before_release, IsTrue());
+ EXPECT_THAT(result.Get(), Eq(42));
+ EXPECT_THAT(calls, Eq(1));
+}
+
+TEST_F(RunExecutorTest, CallerDoesNotDuplicateOrWaitInsideWorkerClaimedWork) {
+ RunExecutor executor(1);
+ executor.Start(1);
+ absl::Notification started;
+ absl::Notification release;
+ std::atomic calls = 0;
+ RunPromise promise;
+ auto result = promise.Task();
+ const auto work = std::make_shared([&calls, &started, &release, promise = std::move(promise)] mutable {
+ calls.fetch_add(1);
+ started.Notify();
+ // Bound a broken claim implementation so it reports failure rather than hanging.
+ promise.SetValue(release.WaitForNotificationWithTimeout(absl::Seconds(5)) ? 42 : 0);
+ });
+ auto dispatched = executor.Submit([work] { work->Run(); });
+ const bool running = started.WaitForNotificationWithTimeout(absl::Seconds(5));
+ work->Run();
+ const bool ready_before_release = result.Ready();
+ release.Notify();
+ dispatched.Get();
+ EXPECT_THAT(running, IsTrue());
+ EXPECT_THAT(ready_before_release, IsFalse());
+ EXPECT_THAT(result.Get(), Eq(42));
+ EXPECT_THAT(calls.load(), Eq(1));
+}
+
+TEST_F(RunExecutorTest, SimultaneousCallerAndWorkerClaimsExecuteMoveOnlyWorkOnce) {
+ RunExecutor executor(2);
+ executor.Start(2);
+ absl::Barrier ready(3);
+ std::atomic calls = 0;
+ RunPromise promise;
+ auto result = promise.Task();
+ const auto work =
+ std::make_shared([&calls, value = std::make_unique(42), promise = std::move(promise)] mutable {
+ calls.fetch_add(1);
+ promise.SetValue(*value);
+ });
+ const auto claim = [&] {
+ ready.Block();
+ work->Run();
+ };
+ auto first = executor.Submit(claim);
+ auto second = executor.Submit(claim);
+ claim();
+ first.Get();
+ second.Get();
+ EXPECT_THAT(result.Get(), Eq(42));
+ EXPECT_THAT(calls.load(), Eq(1));
+}
+
+TEST_F(RunExecutorTest, WorkerWrapperDoesNotWaitForCallerClaimedWork) {
+ RunExecutor executor(1);
+ executor.Start(1);
+ absl::Notification started;
+ absl::Notification release;
+ RunPromise promise;
+ auto result = promise.Task();
+ const auto work = std::make_shared([&started, &release, promise = std::move(promise)] mutable {
+ started.Notify();
+ // A queued wrapper may finish while its caller-owned read is still running.
+ // The coordinator owns that inline lifetime, not the wrapper's completion.
+ promise.SetValue(release.WaitForNotificationWithTimeout(absl::Seconds(5)) ? 42 : 0);
+ });
+ auto wrapper = executor.Submit([&started, &release, work] {
+ const bool caller_started = started.WaitForNotificationWithTimeout(absl::Seconds(5));
+ work->Run();
+ release.Notify();
+ return caller_started;
+ });
+ work->Run();
+ EXPECT_THAT(wrapper.Get(), IsTrue());
+ EXPECT_THAT(result.Get(), Eq(42));
+}
+
TEST_F(RunExecutorTest, ReadAheadSlotsAndBytesAreSharedAndReleasedByTheirOwner) {
RunExecutor executor(2, 100);
EXPECT_THAT(executor.ReserveReadAhead(), Eq(std::nullopt));
diff --git a/xff/engine/walk.cc b/xff/engine/walk.cc
index 3c75edd8b50..6cd03eb3b6f 100644
--- a/xff/engine/walk.cc
+++ b/xff/engine/walk.cc
@@ -87,8 +87,8 @@ class PrefetchedRead final {
public:
PrefetchedRead() = default;
- PrefetchedRead(RunTask task, std::shared_ptr demand)
- : task_(std::move(task)), demand_(std::move(demand)) {}
+ PrefetchedRead(RunTask task, std::shared_ptr demand, std::shared_ptr work)
+ : task_(std::move(task)), demand_(std::move(demand)), work_(std::move(work)) {}
~PrefetchedRead() {
if (demand_) {
@@ -107,12 +107,16 @@ class PrefetchedRead final {
// Only the coordinator can require a listing, immediately before waiting for
// it. That one result belongs to ordinary traversal working memory, not cache.
demand_->required.store(true, std::memory_order_relaxed);
+ // Claim only this required read if the worker has not already claimed it.
+ // Do not execute an unrelated slow sibling while waiting for this result.
+ work_->Run();
return task_.Get();
}
private:
RunTask task_;
std::shared_ptr demand_;
+ std::shared_ptr work_;
};
std::size_t AddBytes(std::size_t lhs, std::size_t rhs) {
@@ -176,7 +180,8 @@ class Walker {
~Walker() {
// A stop action may leave prefetched siblings unconsumed. Their drain jobs
// still reference this walker, so join those jobs before its state is torn
- // down. Normal walks reach already-completed jobs here.
+ // down. Caller-claimed reads finish inline before traversal can return, even
+ // if their queued no-op wrappers have already completed on a worker.
for (const RunTask& drain : drain_tasks_) {
drain.Wait();
}
@@ -424,7 +429,6 @@ class Walker {
window_ = 2 * std::min(walker_.options_.workers - 1, walker_.executor_.worker_count());
canceled_ = std::make_shared>(false);
requests_.emplace();
- Fill();
}
~ReadFrontier() {
@@ -446,7 +450,9 @@ class Walker {
while (!requests.empty() && requests.front().index < index) {
requests.pop_front();
}
- next_ = std::max(next_, index);
+ // An uncached current directory belongs to the caller. Start speculation
+ // at its next sibling, so --jobs=2 can overlap a caller read with one worker.
+ next_ = std::max(next_, index + 1);
Fill();
if (requests.empty() || requests.front().index != index) {
return {};
@@ -491,32 +497,32 @@ class Walker {
RunPromise promise;
auto task = promise.Task();
const auto demand = std::make_shared();
- walker_.drain_tasks_.push_back(walker_.executor_.Submit(
- [&walker = walker_, canceled = canceled_, demand, request_bytes, path = children_.at(next_).path,
- promise = std::move(promise), reservation = std::move(*reservation)] mutable {
- std::optional listing;
- if (!canceled->load(std::memory_order_relaxed) && !demand->canceled.load(std::memory_order_relaxed)) {
- auto read = walker.ReadDir(path);
- if (!canceled->load(std::memory_order_relaxed) && !demand->canceled.load(std::memory_order_relaxed)
- && (demand->required.load(std::memory_order_relaxed)
- || reservation.Retain(AddBytes(request_bytes, ListingBytes(read))))) {
- listing.emplace(std::move(read));
- }
- }
- reservation.Finish();
- // Release the producer handle before the executor signals this job's
- // completion. Otherwise an unobserved result can keep its credit alive
- // briefly after Walk has joined the job and inspected the shared budget.
- auto completed = std::move(promise);
- completed.SetValue({
- .listing = std::move(listing),
- .reservation = std::move(reservation),
- });
- },
- RunTaskClass::kReadAhead));
+ const auto work = std::make_shared([&walker = walker_, canceled = canceled_, demand, request_bytes,
+ path = children_.at(next_).path, promise = std::move(promise),
+ reservation = std::move(*reservation)] mutable {
+ std::optional listing;
+ if (!canceled->load(std::memory_order_relaxed) && !demand->canceled.load(std::memory_order_relaxed)) {
+ auto read = walker.ReadDir(path);
+ if (!canceled->load(std::memory_order_relaxed) && !demand->canceled.load(std::memory_order_relaxed)
+ && (demand->required.load(std::memory_order_relaxed)
+ || reservation.Retain(AddBytes(request_bytes, ListingBytes(read))))) {
+ listing.emplace(std::move(read));
+ }
+ }
+ reservation.Finish();
+ // Release the producer at publication: worker-owned callbacks drop it
+ // before their wrapper completes; caller-owned callbacks finish inline
+ // before Get returns or traversal can inspect the released budget.
+ auto completed = std::move(promise);
+ completed.SetValue({
+ .listing = std::move(listing),
+ .reservation = std::move(reservation),
+ });
+ });
+ walker_.drain_tasks_.push_back(walker_.executor_.Submit([work] { work->Run(); }, RunTaskClass::kReadAhead));
requests.push_back({
.index = next_,
- .task = PrefetchedRead(std::move(task), demand),
+ .task = PrefetchedRead(std::move(task), demand, work),
});
++next_;
}
diff --git a/xff/engine/walk_test.cc b/xff/engine/walk_test.cc
index 65146a9af63..ab93b529058 100644
--- a/xff/engine/walk_test.cc
+++ b/xff/engine/walk_test.cc
@@ -15,6 +15,7 @@
#include "xff/engine/walk.h"
+#include
#include
#include
#include
@@ -25,17 +26,21 @@
#include
#include