From 43d7828dc2acd8dbcd9f735200877542db72c47f Mon Sep 17 00:00:00 2001 From: helly25 <6420169+helly25@users.noreply.github.com> Date: Sun, 1 Nov 2026 09:19:20 +0000 Subject: [PATCH] Isolate executor diagnostics and query download support --- .bazelrc | 5 + bazelmod/clang_tidy_targets.bzl | 1 + docs/benchmarks/executor-dispatch-probe.md | 35 +- .../native-executor-dispatch-controls.json | 2 +- docs/native-profiling.md | 9 +- tools/repository_tooling_test.py | 15 + xff/engine/BUILD.bazel | 67 ++++ xff/engine/executor_benchmark.cc | 79 ++++ xff/engine/executor_measurement.cc | 232 +++++++++++ xff/engine/executor_measurement.h | 54 +++ xff/engine/executor_measurement_test.cc | 102 +++++ xff/engine/run_executor.cc | 176 +++++++++ xff/engine/run_executor.h | 243 ++++++++++++ xff/engine/run_executor_test.cc | 367 ++++++++++++++++++ 14 files changed, 1367 insertions(+), 20 deletions(-) create mode 100644 xff/engine/executor_benchmark.cc create mode 100644 xff/engine/executor_measurement.cc create mode 100644 xff/engine/executor_measurement.h create mode 100644 xff/engine/executor_measurement_test.cc create mode 100644 xff/engine/run_executor.cc create mode 100644 xff/engine/run_executor.h create mode 100644 xff/engine/run_executor_test.cc diff --git a/.bazelrc b/.bazelrc index 831d59903f8..a31edb97a5d 100644 --- a/.bazelrc +++ b/.bazelrc @@ -140,6 +140,11 @@ common:clang_release --config=clang --config=release common:linux_perf --define=pfm=1 common:linux_perf --downloader_config=bazelmod/linux_perf_downloader.cfg +# Unconfigured source queries visit every select arm, including Google Benchmark's +# optional libpfm dependency. Repair that one retired mirror for queries too, +# without enabling hardware counters or changing default/release build settings. +query --downloader_config=bazelmod/linux_perf_downloader.cfg + # --config=clang-coverage : collect C++ coverage with the same hermetic Clang/libc++ toolchain # that owns the default build. Bazel otherwise defaults C++ coverage to the local gcov toolchain. common:clang-coverage --config=clang diff --git a/bazelmod/clang_tidy_targets.bzl b/bazelmod/clang_tidy_targets.bzl index 690ac7fb0bc..64e5e01a52c 100644 --- a/bazelmod/clang_tidy_targets.bzl +++ b/bazelmod/clang_tidy_targets.bzl @@ -12,6 +12,7 @@ CLANG_TIDY_MANUAL_TARGETS = [ "//xff/cli:globals_benchmark", "//xff/engine:read_accounting_test", "//xff/engine:compare_benchmark", + "//xff/engine:executor_benchmark", "//xff/engine:rg_benchmark", "//xff/engine:expression_benchmark", "//xff/engine:expression_run_benchmark", diff --git a/docs/benchmarks/executor-dispatch-probe.md b/docs/benchmarks/executor-dispatch-probe.md index 0c88c911378..d56094d2771 100644 --- a/docs/benchmarks/executor-dispatch-probe.md +++ b/docs/benchmarks/executor-dispatch-probe.md @@ -1,30 +1,29 @@ # Executor lifecycle and task-grain probe This page retains the measurement evidence from [PR #1000](https://github.com/mboworks/xff/pull/1000) -without merging its executor or diagnostic C++ targets. The probe target is not -available on main: reproducing the experiment requires its exact measured source, -`c8929d16c77e318b15d87c5bfcbf07af0fa7058e`. The lossless bundle can instead be -verified offline from main without building or executing any measured binary. - -At that source, `//xff/engine:executor_benchmark` is an explicit diagnostic, not a -shipping dependency or a new runtime admission rule. It separates thread-start calls, -first dispatch, warmed work and executor teardown. Use a fresh executable process -for every cold sample: a previous executor or other Abseil contention may already -have calibrated the process's clock. Ordinary in-process repetition cannot -establish fresh-process startup cost. - -In a separate checkout of the measured source, build before measuring, then -invoke the resolved executable directly. Do not build this target from main or -relabel a new build as verification of the historical executable bytes: +and makes its executor probe available as a test-only diagnostic. It separates +thread-start calls, first dispatch, warmed work and executor teardown. Use a +fresh executable process for every cold sample. + +Build before measuring, then invoke the resolved executable directly: ```sh -git fetch origin audit/executor-dispatch-costs -git worktree add --detach /tmp/xff-executor-probe-repro c8929d16c77e318b15d87c5bfcbf07af0fa7058e -cd /tmp/xff-executor-probe-repro bazel build --config=clang_release //xff/engine:executor_benchmark bazel-bin/xff/engine/executor_benchmark leaf 3 8192 16 128 32 ``` +The candidate executor support and probe libraries are private `testonly` +targets. Bazel prevents shipping binaries from depending on them. Production +traversal and matching keep their existing scheduling implementations; no +admission, shared-pool, frontier, caller-claim or remaining-work policy is +installed by this infrastructure extraction. + +Historical results below describe the exact measured source +`c8929d16c77e318b15d87c5bfcbf07af0fa7058e` and original flags. The current main +includes the Abseil steady-clock correction, so new builds must retain their +own provenance and must not be relabeled as verification of historical binary +bytes or used to justify historical cold-start thresholds. + The six arguments are dispatch style (`leaf`, `claimable`, `drain`), caller-inclusive participants (1..64), items per warmed sweep, sweep count, chunk grain and unsigned mixing iterations per item. All dimensions except mixing iterations must be positive; diff --git a/docs/benchmarks/native-executor-dispatch-controls.json b/docs/benchmarks/native-executor-dispatch-controls.json index 312d25b75b2..1e8af37e6cf 100644 --- a/docs/benchmarks/native-executor-dispatch-controls.json +++ b/docs/benchmarks/native-executor-dispatch-controls.json @@ -147,6 +147,6 @@ "Phase times include diagnostic bookkeeping; fastest seven selects warmed phase, not process wall time.", "CPU model, kernel version and per-sample thermal/load observations were not recorded by this harness.", "Guest-visible distinct package/core pairs do not independently verify physical-host cores or QEMU thread affinity.", - "Diagnostic C++ targets and their executor dependency are not part of this evidence-only extraction onto main." + "Historical evidence remains unchanged; the current diagnostic targets are isolated behind testonly support." ] } diff --git a/docs/native-profiling.md b/docs/native-profiling.md index e320bccd348..ec3da5d13ef 100644 --- a/docs/native-profiling.md +++ b/docs/native-profiling.md @@ -28,7 +28,7 @@ filter, or timing columns alone do not establish working counters. Use The short duration above is a capability probe, not performance evidence. Google Benchmark 1.9.5 pins libpfm 4.11.0.bcr.1. That BCR source names the retired -`netcologne.dl.sourceforge.net` mirror. The opt-in downloader configuration +`netcologne.dl.sourceforge.net` mirror. The downloader configuration redirects exactly that archive to `downloads.sourceforge.net`, preserving the pinned version, BCR patches and integrity check. No LLVM/toolchain dependency changes are involved. It does not rewrite other archives or disable checksum @@ -36,6 +36,13 @@ validation. If network access is unavailable, an independently verified copy of `libpfm-4.11.0.tar.gz` can be supplied through Bazel's `--distdir=PATH`; do not substitute an unverified system library. +Unconfigured `bazel query` traverses all `select` arms, including the optional +libpfm dependency, even when hardware counters are disabled. Queries therefore +use the same narrowly scoped download repair so compile-source indexing can +complete. This does not enable counters or change default/release builds, +compiler/linker selection or macOS runtime behavior. Source indexing still +fails closed on any unsuccessful query. + ## Kernel and external perf capability The `perf` executable is a separate host profiling tool, not a libpfm dependency. diff --git a/tools/repository_tooling_test.py b/tools/repository_tooling_test.py index 25a8e187edc..e96863be12c 100644 --- a/tools/repository_tooling_test.py +++ b/tools/repository_tooling_test.py @@ -69,6 +69,21 @@ def test_libpfm_downloader_rewrites_only_the_retired_archive(self): retired + "?download=1", canonical.removeprefix("https://")): self.assertIsNone(re.fullmatch(pattern, unrelated)) + def test_unconfigured_queries_repair_libpfm_without_enabling_counters(self): + commands = [shlex.split(line, comments=True) + for line in (self.workspace / ".bazelrc").read_text().splitlines()] + commands = [command for command in commands if command] + download_settings = [command for command in commands + if any(option.startswith("--downloader_config=") for option in command[1:])] + self.assertEqual(download_settings, [ + ["common:linux_perf", "--downloader_config=bazelmod/linux_perf_downloader.cfg"], + ["query", "--downloader_config=bazelmod/linux_perf_downloader.cfg"], + ]) + query_settings = [option for command in commands if command[0] == "query" + for option in command[1:]] + self.assertNotIn("--define=pfm=1", query_settings) + self.assertNotIn("--config=linux_perf", query_settings) + if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) diff --git a/xff/engine/BUILD.bazel b/xff/engine/BUILD.bazel index 589895d96c8..c38fd8601ee 100644 --- a/xff/engine/BUILD.bazel +++ b/xff/engine/BUILD.bazel @@ -19,6 +19,73 @@ load("//xff:cc.bzl", "cc_binary", "cc_library", "cc_test") package(default_visibility = ["//visibility:private"]) +# Experimental scheduling support is restricted to tests and diagnostics. +cc_library( + name = "run_executor_cc", + testonly = True, + srcs = ["run_executor.cc"], + hdrs = ["run_executor.h"], + deps = [ + "@abseil-cpp//absl/base:core_headers", + "@abseil-cpp//absl/functional:any_invocable", + "@abseil-cpp//absl/synchronization", + ], +) + +cc_test( + name = "run_executor_test", + size = "small", + srcs = ["run_executor_test.cc"], + deps = [ + ":run_executor_cc", + "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/time", + "@googletest//:gtest", + "@googletest//:gtest_main", + ], +) + +cc_library( + name = "executor_measurement_cc", + testonly = True, + srcs = ["executor_measurement.cc"], + hdrs = ["executor_measurement.h"], + deps = [ + ":run_executor_cc", + "@abseil-cpp//absl/status", + "@abseil-cpp//absl/status:statusor", + ], +) + +cc_test( + name = "executor_measurement_test", + size = "small", + srcs = ["executor_measurement_test.cc"], + deps = [ + ":executor_measurement_cc", + "@abseil-cpp//absl/status", + "@googletest//:gtest", + "@googletest//:gtest_main", + "@mboworks_mbo//mbo/testing:status_cc", + ], +) + +# Fresh-process lifecycle / warmed-dispatch diagnostic, not runtime admission. +cc_binary( + name = "executor_benchmark", + testonly = True, + srcs = ["executor_benchmark.cc"], + tags = [ + "clang-tidy", + "manual", + ], + deps = [ + ":executor_measurement_cc", + "@abseil-cpp//absl/strings", + "@nlohmann_json//:json", + ], +) + # Directory traversal over a vfs::FileSystem: a worker pool reads (readdir+lstat) # in parallel while a single coordinator thread visits in --sort order. cc_library( diff --git a/xff/engine/executor_benchmark.cc b/xff/engine/executor_benchmark.cc new file mode 100644 index 00000000000..5bc616a00d9 --- /dev/null +++ b/xff/engine/executor_benchmark.cc @@ -0,0 +1,79 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#include +#include +#include +#include +#include +#include + +#include "absl/strings/numbers.h" +#include "nlohmann/json.hpp" +#include "xff/engine/executor_measurement.h" + +namespace { + +using ::xff::engine::ExecutorProbePhase; + +nlohmann::json Phase(const ExecutorProbePhase& phase) { + return { + {"seconds", phase.seconds}, + {"items", phase.items}, + {"queued_jobs", phase.queued_jobs}, + {"observed_participants", phase.observed_participants}, + {"caller_participated", phase.caller_participated}, + {"checksum", phase.checksum}, + }; +} + +// XFF_HOST_IO: diagnostic executable's result/error stream adapter; no path I/O. +int Error(std::string_view message) { + std::cerr << message << '\n'; + return 2; +} + +// XFF_HOST_IO: diagnostic executable's result stream adapter; no path I/O. +void Emit(const nlohmann::json& report) { + std::cout << report.dump() << '\n'; +} + +} // namespace + +// XFF_ABI_POINTER: standard process entry point; argv is converted inside this adapter. +int main(int argc, char** argv) { + if (argc != 7) { + return Error("usage: executor_benchmark leaf|claimable|drain PARTICIPANTS ITEMS SWEEPS GRAIN WORK_UNITS"); + } + const std::span argument_view(argv, static_cast(argc)); + const std::vector arguments(argument_view.begin(), argument_view.end()); + const auto dispatch = xff::engine::ParseExecutorProbeDispatch(arguments.at(1)); + if (!dispatch.ok()) { + return Error(dispatch.status().message()); + } + xff::engine::ExecutorProbeConfig config{.dispatch = *dispatch}; + if (!absl::SimpleAtoi(arguments.at(2), &config.participants) || !absl::SimpleAtoi(arguments.at(3), &config.items) + || !absl::SimpleAtoi(arguments.at(4), &config.sweeps) || !absl::SimpleAtoi(arguments.at(5), &config.grain) + || !absl::SimpleAtoi(arguments.at(6), &config.work_units)) { + return Error("probe dimensions must be unsigned integers"); + } + const auto result = xff::engine::MeasureExecutor(config); + if (!result.ok()) { + return Error(result.status().message()); + } + Emit({ + {"dispatch", arguments.at(1)}, + {"participants", config.participants}, + {"items", config.items}, + {"sweeps", config.sweeps}, + {"grain", config.grain}, + {"work_units", config.work_units}, + {"background_workers", result->background_workers}, + {"start_seconds", result->start_seconds}, + {"first_dispatch", Phase(result->first_dispatch)}, + {"warmed", Phase(result->warmed)}, + {"teardown_seconds", result->teardown_seconds}, + {"measurement", "synthetic executor phases including bookkeeping; fresh process required for cold costs"}, + }); + return 0; +} diff --git a/xff/engine/executor_measurement.cc b/xff/engine/executor_measurement.cc new file mode 100644 index 00000000000..bcdd741283d --- /dev/null +++ b/xff/engine/executor_measurement.cc @@ -0,0 +1,232 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#include "xff/engine/executor_measurement.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "absl/status/status.h" +#include "xff/engine/run_executor.h" + +namespace xff::engine { +namespace { + +using Clock = std::chrono::steady_clock; + +double Elapsed(Clock::time_point start) { + return std::chrono::duration(Clock::now() - start).count(); +} + +struct Chunk { + std::size_t items = 0; + std::uint64_t checksum = 0; + std::thread::id participant = std::this_thread::get_id(); +}; + +// Runtime-dependent unsigned mixing keeps the work observable in the returned +// checksum. Zero units measures dispatch with only the seed and accumulation. +Chunk Work(std::size_t first, std::size_t end, std::size_t units) { + Chunk result; + for (std::size_t index = first; index < end; ++index) { + std::uint64_t value = index + 1; + for (std::size_t unit = 0; unit < units; ++unit) { + value ^= value << 7U; + value ^= value >> 9U; + value ^= value << 8U; + } + result.checksum += value; + ++result.items; + } + return result; +} + +class Accumulator final { + public: + void Add(const Chunk& chunk) { + phase_.items += chunk.items; + phase_.checksum += chunk.checksum; + if (chunk.items != 0) { + participants_.insert(chunk.participant); + } + } + + void Queued() { ++phase_.queued_jobs; } + + ExecutorProbePhase Finish(double seconds) && { + phase_.seconds = seconds; + phase_.observed_participants = participants_.size(); + phase_.caller_participated = participants_.contains(std::this_thread::get_id()); + return phase_; + } + + private: + ExecutorProbePhase phase_; + std::set participants_; +}; + +struct PendingChunk { + RunTask result; + RunTask wrapper; + std::shared_ptr work; + + Chunk Get() { + if (work) { + work->Run(); + } + auto chunk = result.Get(); + if (wrapper.Valid()) { + wrapper.Get(); + } + return chunk; + } +}; + +PendingChunk Queue( + RunExecutor& executor, + ExecutorProbeDispatch dispatch, + std::size_t first, + std::size_t end, + std::size_t units) { + const auto job = [first, end, units] { return Work(first, end, units); }; + if (dispatch != ExecutorProbeDispatch::kClaimable) { + return {.result = executor.Submit(job)}; + } + RunPromise promise; + auto result = promise.Task(); + const auto work = std::make_shared([job, promise = std::move(promise)] mutable { promise.SetValue(job()); }); + auto wrapper = executor.Submit([work] { work->Run(); }, RunTaskClass::kReadAhead); + return {.result = std::move(result), .wrapper = std::move(wrapper), .work = work}; +} + +ExecutorProbePhase FirstDispatch(RunExecutor& executor, std::size_t participants) { + Accumulator total; + std::vector pending; + pending.reserve(participants - 1); + const auto start = Clock::now(); + for (std::size_t index = 0; index + 1 < participants; ++index) { + pending.push_back(Queue(executor, ExecutorProbeDispatch::kLeaf, index, index + 1, 0)); + total.Queued(); + } + total.Add(Work(participants - 1, participants, 0)); + for (auto& task : pending) { + total.Add(task.Get()); + } + return std::move(total).Finish(Elapsed(start)); +} + +ExecutorProbePhase LeafSweeps(RunExecutor& executor, const ExecutorProbeConfig& config) { + Accumulator total; + std::vector pending; + pending.reserve(config.participants - 1); + const auto start = Clock::now(); + for (std::size_t sweep = 0; sweep < config.sweeps; ++sweep) { + std::size_t next = 0; + while (next < config.items) { + pending.clear(); + const std::size_t caller_first = next; + const std::size_t caller_end = next + std::min(config.grain, config.items - next); + next = caller_end; + for (std::size_t worker = 1; worker < config.participants && next < config.items; ++worker) { + const std::size_t end = next + std::min(config.grain, config.items - next); + pending.push_back(Queue(executor, config.dispatch, next, end, config.work_units)); + total.Queued(); + next = end; + } + total.Add(Work(caller_first, caller_end, config.work_units)); + for (auto& task : pending) { + total.Add(task.Get()); + } + } + } + return std::move(total).Finish(Elapsed(start)); +} + +ExecutorProbePhase DrainSweeps(RunExecutor& executor, const ExecutorProbeConfig& config) { + Accumulator total; + std::vector> pending; + pending.reserve(config.participants - 1); + // Only this atomic assigns chunk indices. Each drain owns its local checksum; + // only the coordinator merges returned chunks and changes total/pending. + std::atomic next = 0; + const std::size_t chunks = ((config.items - 1) / config.grain) + 1; + const auto drain = [&] { + Chunk result; + for (;;) { + const std::size_t index = next.fetch_add(1, std::memory_order_relaxed); + if (index >= chunks) { + return result; + } + const std::size_t first = index * config.grain; + const auto chunk = Work(first, first + std::min(config.grain, config.items - first), config.work_units); + result.items += chunk.items; + result.checksum += chunk.checksum; + } + }; + const auto start = Clock::now(); + for (std::size_t sweep = 0; sweep < config.sweeps; ++sweep) { + next.store(0, std::memory_order_relaxed); + pending.clear(); + for (std::size_t worker = 1; worker < config.participants; ++worker) { + pending.push_back(executor.Submit(drain)); + total.Queued(); + } + total.Add(drain()); + for (auto& task : pending) { + total.Add(task.Get()); + } + } + return std::move(total).Finish(Elapsed(start)); +} + +} // namespace + +absl::StatusOr ParseExecutorProbeDispatch(std::string_view name) { + if (name == "leaf") { + return ExecutorProbeDispatch::kLeaf; + } + if (name == "claimable") { + return ExecutorProbeDispatch::kClaimable; + } + if (name == "drain") { + return ExecutorProbeDispatch::kDrain; + } + return absl::InvalidArgumentError("dispatch must be leaf, claimable or drain"); +} + +absl::StatusOr MeasureExecutor(const ExecutorProbeConfig& config) { + if (config.participants == 0 || config.participants > 64 || config.items == 0 || config.sweeps == 0 + || config.grain == 0 || config.items > std::numeric_limits::max() / config.sweeps + || config.sweeps > std::numeric_limits::max() / config.participants + || config.items > std::numeric_limits::max() - config.participants) { + return absl::InvalidArgumentError("participants must be 1..64; items/sweeps/grain positive and representable"); + } + if (config.dispatch != ExecutorProbeDispatch::kLeaf && config.dispatch != ExecutorProbeDispatch::kClaimable + && config.dispatch != ExecutorProbeDispatch::kDrain) { + return absl::InvalidArgumentError("unknown dispatch"); + } + auto executor = std::make_unique(config.participants - 1); + ExecutorProbeMeasurement result; + auto start = Clock::now(); + executor->Start(config.participants - 1); + result.start_seconds = Elapsed(start); + result.background_workers = executor->worker_count(); + result.first_dispatch = FirstDispatch(*executor, config.participants); + result.warmed = + config.dispatch == ExecutorProbeDispatch::kDrain ? DrainSweeps(*executor, config) : LeafSweeps(*executor, config); + start = Clock::now(); + executor.reset(); + result.teardown_seconds = Elapsed(start); + return result; +} + +} // namespace xff::engine diff --git a/xff/engine/executor_measurement.h b/xff/engine/executor_measurement.h new file mode 100644 index 00000000000..09151dc5c22 --- /dev/null +++ b/xff/engine/executor_measurement.h @@ -0,0 +1,54 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#ifndef XFF_ENGINE_EXECUTOR_MEASUREMENT_H_ +#define XFF_ENGINE_EXECUTOR_MEASUREMENT_H_ + +#include +#include +#include + +#include "absl/status/statusor.h" + +namespace xff::engine { + +enum class ExecutorProbeDispatch { kLeaf, kClaimable, kDrain }; + +absl::StatusOr ParseExecutorProbeDispatch(std::string_view name); + +struct ExecutorProbeConfig { + ExecutorProbeDispatch dispatch = ExecutorProbeDispatch::kLeaf; + std::size_t participants = 1; + std::size_t items = 1'024; + std::size_t sweeps = 16; + std::size_t grain = 1; + std::size_t work_units = 0; +}; + +struct ExecutorProbePhase { + double seconds = 0; + std::size_t items = 0; + std::size_t queued_jobs = 0; + std::size_t observed_participants = 0; + bool caller_participated = false; + std::uint64_t checksum = 0; +}; + +struct ExecutorProbeMeasurement { + std::size_t background_workers = 0; + double start_seconds = 0; + ExecutorProbePhase first_dispatch; + ExecutorProbePhase warmed; + double teardown_seconds = 0; +}; + +// Diagnostic only: explicitly start N-1 workers, dispatch one cold batch, then +// sweep synthetic checksum work with a warm pool. Cold results require a fresh +// process: earlier Abseil contention may already have calibrated its clock. +// Phase times include probe bookkeeping, not just isolated queue operations. +// No VFS, traversal/frontier, storage or runtime admission policy is measured. +absl::StatusOr MeasureExecutor(const ExecutorProbeConfig& config); + +} // namespace xff::engine + +#endif // XFF_ENGINE_EXECUTOR_MEASUREMENT_H_ diff --git a/xff/engine/executor_measurement_test.cc b/xff/engine/executor_measurement_test.cc new file mode 100644 index 00000000000..af9a68a1842 --- /dev/null +++ b/xff/engine/executor_measurement_test.cc @@ -0,0 +1,102 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#include "xff/engine/executor_measurement.h" + +#include +#include +#include + +#include "absl/status/status.h" +#include "gmock/gmock.h" +#include "gtest/gtest.h" +#include "mbo/testing/status.h" + +namespace xff::engine { +namespace { +using ::mbo::testing::IsOkAndHolds; +using ::mbo::testing::StatusIs; +using ::testing::Eq; +using ::testing::Ge; +using ::testing::IsTrue; +using ::testing::Le; + +struct ExecutorMeasurementTest : ::testing::Test { + static constexpr std::array kDispatches = { + ExecutorProbeDispatch::kLeaf, + ExecutorProbeDispatch::kClaimable, + ExecutorProbeDispatch::kDrain, + }; +}; + +TEST_F(ExecutorMeasurementTest, ParsesOnlyDiagnosticDispatches) { + EXPECT_THAT(ParseExecutorProbeDispatch("leaf"), IsOkAndHolds(ExecutorProbeDispatch::kLeaf)); + EXPECT_THAT(ParseExecutorProbeDispatch("claimable"), IsOkAndHolds(ExecutorProbeDispatch::kClaimable)); + EXPECT_THAT(ParseExecutorProbeDispatch("drain"), IsOkAndHolds(ExecutorProbeDispatch::kDrain)); + EXPECT_THAT(ParseExecutorProbeDispatch("unknown"), StatusIs(absl::StatusCode::kInvalidArgument)); +} + +TEST_F(ExecutorMeasurementTest, CountsEveryPartialChunkAndKeepsTheCallerInTheBudget) { + for (const auto dispatch : kDispatches) { + SCOPED_TRACE(static_cast(dispatch)); + for (const std::size_t participants : {1, 2, 3, 10}) { + SCOPED_TRACE(participants); + for (const std::size_t grain : {1, 16, 128}) { + SCOPED_TRACE(grain); + ASSERT_OK_AND_ASSIGN( + const auto result, + MeasureExecutor( + {.dispatch = dispatch, .participants = participants, .items = 513, .sweeps = 2, .grain = grain})); + EXPECT_THAT(result.background_workers, Eq(participants - 1)); + EXPECT_THAT(result.first_dispatch.items, Eq(participants)); + EXPECT_THAT(result.first_dispatch.queued_jobs, Eq(participants - 1)); + EXPECT_THAT(result.first_dispatch.checksum, Eq(participants * (participants + 1) / 2)); + EXPECT_THAT(result.first_dispatch.observed_participants, Le(participants)); + EXPECT_THAT(result.first_dispatch.caller_participated, IsTrue()); + EXPECT_THAT(result.warmed.items, Eq(1'026)); + EXPECT_THAT(result.warmed.checksum, Eq(263'682)); + EXPECT_THAT(result.warmed.observed_participants, Le(participants)); + // Drain workers may claim every chunk before their coordinator does. + // Caller participation is forced by leaf sweeps, not assumed for drains. + if (dispatch != ExecutorProbeDispatch::kDrain) { + EXPECT_THAT(result.warmed.caller_participated, IsTrue()); + } + if (participants == 1) { + EXPECT_THAT(result.warmed.queued_jobs, Eq(0)); + } + EXPECT_THAT(result.start_seconds, Ge(0.0)); + EXPECT_THAT(result.first_dispatch.seconds, Ge(0.0)); + EXPECT_THAT(result.warmed.seconds, Ge(0.0)); + EXPECT_THAT(result.teardown_seconds, Ge(0.0)); + } + } + } +} + +TEST_F(ExecutorMeasurementTest, RuntimeWorkHasTheIndependentlyKnownUnsignedChecksum) { + for (const auto dispatch : kDispatches) { + ASSERT_OK_AND_ASSIGN( + const auto result, + MeasureExecutor( + {.dispatch = dispatch, .participants = 3, .items = 2, .sweeps = 3, .grain = 1, .work_units = 1})); + EXPECT_THAT(result.warmed.items, Eq(6)); + EXPECT_THAT(result.warmed.checksum, Eq(3 * (33'153 + 66'306))); + } +} + +TEST_F(ExecutorMeasurementTest, RejectsInvalidOrUnrepresentableWork) { + const auto invalid = [](const ExecutorProbeConfig& config) { + EXPECT_THAT(MeasureExecutor(config), StatusIs(absl::StatusCode::kInvalidArgument)); + }; + invalid({.participants = 0}); + invalid({.participants = 65}); + invalid({.items = 0}); + invalid({.sweeps = 0}); + invalid({.grain = 0}); + invalid({.items = std::numeric_limits::max(), .sweeps = 2}); + invalid({.participants = 3, .items = 1, .sweeps = std::numeric_limits::max()}); + invalid({.dispatch = static_cast(99)}); +} + +} // namespace +} // namespace xff::engine diff --git a/xff/engine/run_executor.cc b/xff/engine/run_executor.cc new file mode 100644 index 00000000000..6d5ffb5e316 --- /dev/null +++ b/xff/engine/run_executor.cc @@ -0,0 +1,176 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#include "xff/engine/run_executor.h" + +#include +#include +#include +#include + +namespace xff::engine { + +namespace executor_detail { +struct ReadAheadBudget final { + explicit ReadAheadBudget(std::size_t bytes) : limit(bytes) {} + + const std::size_t limit; + // The mutex guards slot and byte accounting, including peaks. A reservation + // exclusively owns its own charge; its shared budget outlives all reservations. + absl::Mutex mutex; + ReadAheadStats usage ABSL_GUARDED_BY(mutex); +}; +} // 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) {} + + ~Impl() { + const absl::MutexLock lock(budget->mutex); + if (unfinished) { + --budget->usage.in_use; + } + budget->usage.retained_bytes -= bytes; + } + + Impl(const Impl&) = delete; + Impl& operator=(const Impl&) = delete; + Impl(Impl&&) = delete; + Impl& operator=(Impl&&) = delete; + + const std::shared_ptr budget; + const std::size_t minimum_bytes; + std::size_t bytes = 0; + bool unfinished = true; +}; + +ReadAheadReservation::ReadAheadReservation(std::unique_ptr impl) : impl_(std::move(impl)) {} + +ReadAheadReservation::~ReadAheadReservation() = default; +ReadAheadReservation::ReadAheadReservation(ReadAheadReservation&&) noexcept = default; +ReadAheadReservation& ReadAheadReservation::operator=(ReadAheadReservation&&) noexcept = default; + +bool ReadAheadReservation::Retain(std::size_t bytes) { + const absl::MutexLock lock(impl_->budget->mutex); + auto& usage = impl_->budget->usage; + const std::size_t others = usage.retained_bytes - impl_->bytes; + if (bytes < impl_->minimum_bytes || bytes > impl_->budget->limit - others) { + return false; + } + usage.retained_bytes = others + bytes; + usage.peak_retained_bytes = std::max(usage.peak_retained_bytes, usage.retained_bytes); + impl_->bytes = bytes; + return true; +} + +void ReadAheadReservation::Finish() { + const absl::MutexLock lock(impl_->budget->mutex); + if (impl_->unfinished) { + --impl_->budget->usage.in_use; + impl_->unfinished = false; + } +} + +RunExecutor::RunExecutor(std::size_t max_workers, std::size_t read_ahead_bytes) + : max_workers_(max_workers), read_ahead_bytes_(read_ahead_bytes) {} + +RunExecutor::~RunExecutor() { + { + const absl::MutexLock lock(mutex_); + stop_ = true; + } + for (std::thread& thread : threads_) { + thread.join(); + } +} + +void RunExecutor::Start(std::size_t useful_workers) { + const std::size_t count = std::min(max_workers_, useful_workers); + if (count == 0) { + return; + } + if (threads_.empty()) { + const absl::MutexLock lock(mutex_); + queues_.emplace(); + } + threads_.reserve(count); + while (threads_.size() < count) { + threads_.emplace_back([this] { Run(); }); + } +} + +bool RunExecutor::Pending() const { + return stop_ || (queues_ && (!queues_->foreground.empty() || !queues_->read_ahead.empty())); +} + +std::optional RunExecutor::ReserveReadAhead(std::size_t initial_bytes) { + if (threads_.empty()) { + return std::nullopt; + } + if (!read_ahead_budget_) { + read_ahead_budget_ = std::make_shared(read_ahead_bytes_); + } + const absl::MutexLock lock(read_ahead_budget_->mutex); + const auto capacity = std::min(threads_.size(), std::numeric_limits::max() / 2) * 2; + if (read_ahead_budget_->usage.in_use >= capacity + || initial_bytes > read_ahead_bytes_ - read_ahead_budget_->usage.retained_bytes) { + return std::nullopt; + } + auto impl = std::make_unique(read_ahead_budget_, initial_bytes); + ++read_ahead_budget_->usage.in_use; + read_ahead_budget_->usage.retained_bytes += initial_bytes; + read_ahead_budget_->usage.peak_retained_bytes = + std::max(read_ahead_budget_->usage.peak_retained_bytes, read_ahead_budget_->usage.retained_bytes); + read_ahead_budget_->usage.peak_in_use = + std::max(read_ahead_budget_->usage.peak_in_use, read_ahead_budget_->usage.in_use); + return ReadAheadReservation(std::move(impl)); +} + +ReadAheadStats RunExecutor::ReadAheadUsage() const { + if (!read_ahead_budget_) { + return {}; + } + const absl::MutexLock lock(read_ahead_budget_->mutex); + return read_ahead_budget_->usage; +} + +void RunExecutor::Run() { + for (;;) { + absl::AnyInvocable job; + { + const absl::MutexLock lock(mutex_, absl::Condition(this, &RunExecutor::Pending)); + if (!queues_) { + return; // An unstarted executor has no queued work to dispatch. + } + auto& queues = queues_.value(); + if (stop_ && queues.foreground.empty() && queues.read_ahead.empty()) { + return; + } + // Alternating classes prevents an older speculative frontier from sitting + // ahead of all foreground work, without starving speculation in a busy pool. + const bool read_ahead = !queues.read_ahead.empty() && (queues.foreground.empty() || queues.prefer_read_ahead); + auto& queue = read_ahead ? queues.read_ahead : queues.foreground; + job = std::move(queue.front()); + queue.pop(); + queues.prefer_read_ahead = !read_ahead; + } + job(); + } +} + +} // namespace xff::engine diff --git a/xff/engine/run_executor.h b/xff/engine/run_executor.h new file mode 100644 index 00000000000..490d0c795fd --- /dev/null +++ b/xff/engine/run_executor.h @@ -0,0 +1,243 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#ifndef XFF_ENGINE_RUN_EXECUTOR_H_ +#define XFF_ENGINE_RUN_EXECUTOR_H_ + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/functional/any_invocable.h" +#include "absl/synchronization/mutex.h" +#include "absl/synchronization/notification.h" + +namespace xff::engine { + +namespace executor_detail { +struct ReadAheadBudget; + +template +struct TaskState final { + using Stored = std::conditional_t, std::monostate, Result>; + std::optional result; + absl::Notification completed; +}; +} // namespace executor_detail + +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; + // Pending/finished request bookkeeping plus retained speculative listing storage. + std::size_t retained_bytes = 0; + std::size_t peak_in_use = 0; + std::size_t peak_retained_bytes = 0; +}; + +// An owned speculative-read reservation. Retain replaces its total byte charge, +// never below the initial bookkeeping charge; a failed charge must discard the +// additional storage instead of publishing it. Finish releases the unfinished-read +// slot without releasing bytes. Destruction releases any remaining slot and bytes. +// The reservation can outlive its executor, but only one owner may mutate it. +class ReadAheadReservation final { + public: + ~ReadAheadReservation(); + ReadAheadReservation(ReadAheadReservation&&) noexcept; + ReadAheadReservation& operator=(ReadAheadReservation&&) noexcept; + ReadAheadReservation(const ReadAheadReservation&) = delete; + ReadAheadReservation& operator=(const ReadAheadReservation&) = delete; + + [[nodiscard]] bool Retain(std::size_t bytes); + // Idempotent. Completed cached results must not occupy execution slots. + void Finish(); + + private: + friend class RunExecutor; + struct Impl; + explicit ReadAheadReservation(std::unique_ptr impl); + std::unique_ptr impl_; +}; + +template +class RunPromise; + +// A single-consumer result. Wait, Ready and Get require a valid task; Get consumes +// the value once. Completion publishes through an Abseil notification; no worker +// may touch the value after signaling completion. +template +class RunTask final { + public: + RunTask() = default; + RunTask(RunTask&&) = default; + RunTask& operator=(RunTask&&) = default; + RunTask(const RunTask&) = delete; + RunTask& operator=(const RunTask&) = delete; + + bool Valid() const { return state_ != nullptr; } + + bool Ready() const { return state_->completed.HasBeenNotified(); } + + void Wait() const { state_->completed.WaitForNotification(); } + + Result Get() { + Wait(); + if constexpr (!std::is_void_v) { + return std::move(*state_->result); + } + } + + private: + friend class RunPromise; + + explicit RunTask(std::shared_ptr> state) : state_(std::move(state)) {} + + std::shared_ptr> state_; +}; + +// The producer side of a RunTask, used when a drain job produces indexed results. +// Obtain one Task and call SetValue exactly once before releasing the producer. +template +class RunPromise final { + public: + using Stored = typename executor_detail::TaskState::Stored; + + RunTask Task() const { return RunTask(state_); } + + void SetValue(Stored result) + requires(!std::is_void_v) + { + state_->result.emplace(std::move(result)); + state_->completed.Notify(); + } + + void SetValue() + requires(std::is_void_v) + { + state_->completed.Notify(); + } + + private: + std::shared_ptr> state_ = std::make_shared>(); +}; + +// One lazily started worker executor shared by all parallel phases of a command. +// max_workers counts background threads; command owners reserve one participant +// for the calling coordinator and therefore pass N-1 for a --jobs=N allowance. +// Start grows the pool but never shrinks it; destruction joins the workers once. +// Start, Submit and worker_count are called only by the coordinator. Jobs stay +// leaves and must not submit or wait on work in this executor. +class RunExecutor final { + public: + static constexpr std::size_t kReadAheadBytes = 8 * 1'024 * 1'024; + + explicit RunExecutor(std::size_t max_workers, std::size_t read_ahead_bytes = kReadAheadBytes); + + ~RunExecutor(); + + RunExecutor(const RunExecutor&) = delete; + RunExecutor& operator=(const RunExecutor&) = delete; + RunExecutor(RunExecutor&&) = delete; + RunExecutor& operator=(RunExecutor&&) = delete; + + void Start(std::size_t useful_workers); + + // Coordinator-only admission and inspection. All walks sharing this executor use + // the same byte allowance and at most twice the started worker count in unfinished reads. + // Initial bytes charge the request's bookkeeping/path before it can enter the queue. + // In-flight VFS calls and the coordinator's required listing are not retained lookahead. + [[nodiscard]] std::optional ReserveReadAhead(std::size_t initial_bytes = 0); + ReadAheadStats ReadAheadUsage() const; + + std::size_t ReadAheadLimit() const { return read_ahead_bytes_; } + + template + auto Submit(Job job, RunTaskClass task_class = RunTaskClass::kForeground) ABSL_LOCKS_EXCLUDED(mutex_) { + using Result = std::invoke_result_t; + RunPromise promise; + auto result = promise.Task(); + absl::AnyInvocable task = [promise = std::move(promise), job = std::move(job)]() mutable { + if constexpr (std::is_void_v) { + job(); + promise.SetValue(); + } else { + promise.SetValue(job()); + } + }; + if (threads_.empty()) { + task(); + return result; + } + const absl::MutexLock lock(mutex_); + auto& queues = queues_.value(); + if (task_class == RunTaskClass::kReadAhead) { + queues.read_ahead.emplace(std::move(task)); + } else { + queues.foreground.emplace(std::move(task)); + } + return result; + } + + std::size_t worker_count() const { return threads_.size(); } + + private: + bool Pending() const ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); + void Run() ABSL_LOCKS_EXCLUDED(mutex_); + + struct DispatchQueues { + std::queue> foreground; + std::queue> read_ahead; + bool prefer_read_ahead = false; + }; + + const std::size_t max_workers_; + const std::size_t read_ahead_bytes_; + // Initialized lazily by the coordinator before publishing the first reservation. + // Workers access only their shared reservation state, never this owning handle. + std::shared_ptr read_ahead_budget_; + // The mutex guards both queues, dispatch preference and shutdown. Jobs run unlocked; + // the coordinator alone grows threads_ and inspects their count. Dispatch and + // shared-budget mutexes are never held simultaneously by this implementation. + mutable absl::Mutex mutex_; + // std::queue's default deque allocates even while empty. Inline commands never + // construct these queues; Start initializes them before publishing any worker. + std::optional queues_ ABSL_GUARDED_BY(mutex_); + bool stop_ ABSL_GUARDED_BY(mutex_) = false; + std::vector threads_; +}; + +} // namespace xff::engine + +#endif // XFF_ENGINE_RUN_EXECUTOR_H_ diff --git a/xff/engine/run_executor_test.cc b/xff/engine/run_executor_test.cc new file mode 100644 index 00000000000..472e334f9bf --- /dev/null +++ b/xff/engine/run_executor_test.cc @@ -0,0 +1,367 @@ +// SPDX-FileCopyrightText: Copyright (c) M. Boerger, the MBO Works authors +// SPDX-License-Identifier: Apache-2.0 + +#include "xff/engine/run_executor.h" + +#include +#include +#include +#include +#include +#include +#include + +#include "absl/synchronization/barrier.h" +#include "absl/time/time.h" +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +namespace xff::engine { +namespace { +using ::testing::_; +using ::testing::Contains; +using ::testing::ElementsAre; +using ::testing::Eq; +using ::testing::IsFalse; +using ::testing::IsTrue; +using ::testing::Le; +using ::testing::Ne; +using ::testing::Not; +using ::testing::Optional; +using ::testing::SizeIs; + +struct RunExecutorTest : ::testing::Test {}; + +TEST_F(RunExecutorTest, RunsInlineUntilStartedAndRetainsWorkers) { + RunExecutor executor(3); + const std::thread::id coordinator = std::this_thread::get_id(); + EXPECT_THAT(executor.Submit([] { return std::this_thread::get_id(); }).Get(), Eq(coordinator)); + + executor.Start(2); + EXPECT_THAT(executor.worker_count(), Eq(2)); + EXPECT_THAT(executor.Submit([] { return std::this_thread::get_id(); }).Get(), Ne(coordinator)); + + executor.Start(3); + EXPECT_THAT(executor.worker_count(), Eq(3)); + executor.Start(1); + EXPECT_THAT(executor.worker_count(), Eq(3)); + executor.Start(100); + EXPECT_THAT(executor.worker_count(), Eq(3)); +} + +TEST_F(RunExecutorTest, ReusesTheSameBoundedThreadsAcrossPhases) { + RunExecutor executor(3); + executor.Start(3); + const auto phase = [&] { + absl::Barrier started(3); + std::vector> tasks; + tasks.reserve(3); + for (int worker = 0; worker < 3; ++worker) { + tasks.push_back(executor.Submit([&] { + started.Block(); + return std::this_thread::get_id(); + })); + } + std::set threads; + for (auto& task : tasks) { + threads.insert(task.Get()); + } + return threads; + }; + const auto first = phase(); + EXPECT_THAT(first, SizeIs(3)); + EXPECT_THAT(first, Not(Contains(std::this_thread::get_id()))); + EXPECT_THAT(phase(), Eq(first)); +} + +TEST_F(RunExecutorTest, SupportsMoveOnlyJobsAndResults) { + RunExecutor executor(1); + executor.Start(1); + auto task = executor.Submit([value = std::make_unique(42)] mutable { return std::move(value); }); + const auto result = task.Get(); + EXPECT_THAT(*result, Eq(42)); + EXPECT_THAT(RunTask{}.Valid(), IsFalse()); + executor.Submit([] {}).Get(); +} + +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)); + executor.Start(1); + auto first = executor.ReserveReadAhead(); + auto second = executor.ReserveReadAhead(); + ASSERT_THAT(first, Optional(_)); + ASSERT_THAT(second, Optional(_)); + // The assertions report absence; this guard also proves engagement to static analysis. + if (!first || !second) { + return; + } + EXPECT_THAT(executor.ReserveReadAhead(), Eq(std::nullopt)); + EXPECT_THAT(first.value().Retain(60), IsTrue()); + EXPECT_THAT(second.value().Retain(41), IsFalse()); + EXPECT_THAT(second.value().Retain(40), IsTrue()); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(100)); + EXPECT_THAT(first.value().Retain(std::numeric_limits::max()), IsFalse()); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(100)); + first.reset(); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(40)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(1)); + executor.Start(2); + std::vector additional; + additional.reserve(3); + for (int index = 0; index < 3; ++index) { + auto reservation = executor.ReserveReadAhead(); + ASSERT_THAT(reservation, Optional(_)); + if (!reservation) { + return; + } + additional.push_back(std::move(reservation.value())); + } + EXPECT_THAT(executor.ReserveReadAhead(), Eq(std::nullopt)); + EXPECT_THAT(executor.ReadAheadUsage().peak_in_use, Eq(4)); + EXPECT_THAT(executor.ReadAheadUsage().peak_retained_bytes, Eq(100)); + additional.clear(); + second.reset(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); +} + +TEST_F(RunExecutorTest, ConcurrentReadRetainersCannotExceedTheirSharedByteAllowance) { + RunExecutor executor(3, 100); + executor.Start(3); + absl::Barrier ready(3); + std::vector> tasks; + tasks.reserve(3); + for (int index = 0; index < 3; ++index) { + auto reservation = executor.ReserveReadAhead(); + ASSERT_THAT(reservation, Optional(_)); + tasks.push_back(executor.Submit([&ready, reservation = std::move(reservation)] mutable { + ready.Block(); + const bool retained = reservation.value().Retain(40); + reservation.reset(); + return retained; + })); + } + // Each job releases its charge at completion, so successes are not required to + // overlap. The measured peak must always fit, regardless of worker ordering. + for (auto& task : tasks) { + task.Get(); + } + EXPECT_THAT(executor.ReadAheadUsage().peak_retained_bytes, Le(100)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); +} + +TEST_F(RunExecutorTest, CompletedListingsRetainBytesWithoutBlockingDeeperReadSlots) { + RunExecutor executor(1, 100); + executor.Start(1); + auto first = executor.ReserveReadAhead(20); + auto second = executor.ReserveReadAhead(20); + ASSERT_THAT(first, Optional(_)); + ASSERT_THAT(second, Optional(_)); + if (!first || !second) { + return; + } + EXPECT_THAT(executor.ReserveReadAhead(1), Eq(std::nullopt)); + first.value().Finish(); + second.value().Finish(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(40)); + EXPECT_THAT(executor.ReserveReadAhead(61), Eq(std::nullopt)); + auto deeper = executor.ReserveReadAhead(60); + ASSERT_THAT(deeper, Optional(_)); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(1)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(100)); + deeper.reset(); + first.reset(); + second.reset(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); +} + +TEST_F(RunExecutorTest, FinishingIsIdempotentAndCannotDropTheBookkeepingCharge) { + RunExecutor executor(1, 100); + executor.Start(1); + auto reservation = executor.ReserveReadAhead(20); + ASSERT_THAT(reservation, Optional(_)); + if (!reservation) { + return; + } + EXPECT_THAT(reservation.value().Retain(19), IsFalse()); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(20)); + EXPECT_THAT(reservation.value().Retain(100), IsTrue()); + reservation.value().Finish(); + reservation.value().Finish(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(100)); + EXPECT_THAT(reservation.value().Retain(0), IsFalse()); + EXPECT_THAT(reservation.value().Retain(20), IsTrue()); + auto unfinished = executor.ReserveReadAhead(80); + ASSERT_THAT(unfinished, Optional(_)); + reservation.reset(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(1)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(80)); + unfinished.reset(); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(0)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(0)); +} + +TEST_F(RunExecutorTest, ForegroundAndReadAheadQueuesAlternateWithoutStarvation) { + RunExecutor executor(1); + executor.Start(1); + absl::Notification started; + absl::Notification release; + std::vector order; + auto initial = executor.Submit( + [&] { + started.Notify(); + release.WaitForNotification(); + order.push_back(0); + }, + RunTaskClass::kReadAhead); + const bool running = started.WaitForNotificationWithTimeout(absl::Seconds(5)); + std::vector> tasks; + tasks.reserve(5); + tasks.push_back(executor.Submit([&] { order.push_back(1); }, RunTaskClass::kReadAhead)); + tasks.push_back(executor.Submit([&] { order.push_back(2); }, RunTaskClass::kReadAhead)); + tasks.push_back(executor.Submit([&] { order.push_back(3); })); + tasks.push_back(executor.Submit([&] { order.push_back(4); })); + tasks.push_back(executor.Submit([&] { order.push_back(5); })); + release.Notify(); + initial.Get(); + for (auto& task : tasks) { + task.Get(); + } + EXPECT_THAT(running, IsTrue()); + EXPECT_THAT(order, ElementsAre(0, 3, 1, 4, 2, 5)); +} + +TEST_F(RunExecutorTest, AReservationCanMoveAcrossThreadsAndOutliveItsExecutor) { + std::optional retained; + { + RunExecutor executor(1, 100); + executor.Start(1); + auto reservation = executor.ReserveReadAhead(); + ASSERT_THAT(reservation, Optional(_)); + if (!reservation) { + return; + } + auto task = executor.Submit([reservation = std::move(reservation.value())] mutable { + EXPECT_THAT(reservation.Retain(50), IsTrue()); + return std::move(reservation); + }); + retained.emplace(task.Get()); + EXPECT_THAT(executor.ReadAheadUsage().in_use, Eq(1)); + EXPECT_THAT(executor.ReadAheadUsage().retained_bytes, Eq(50)); + } + EXPECT_THAT(retained.value().Retain(100), IsTrue()); + EXPECT_THAT(retained.value().Retain(101), IsFalse()); +} + +} // namespace +} // namespace xff::engine