diff --git a/core/docs/m7-baseline.md b/core/docs/m7-baseline.md index d6640c9..8957426 100644 --- a/core/docs/m7-baseline.md +++ b/core/docs/m7-baseline.md @@ -2,9 +2,11 @@ Measured against `docs/04-engine-design.md` §8's targets, via `tools/bench/vdm_bench` (see that file's header comment for the exact commands — reproduced below with their -actual output). `--preset release`, this machine, 2026-09-11. This is a baseline -record, not a sign-off: two of the three numbers below don't clear the DoD line yet, and -that's stated plainly rather than rounded away — see "Open gaps". +actual output). `--preset release`, this machine, 2026-09-11 through 2026-09-12. This is +a baseline record, not a sign-off: one of the three numbers below is loopback-only rather +than measured against a real 1 Gbit link, stated plainly rather than rounded away — see +"Open gaps". The RSS number *did* fail the DoD line in the first pass through this file; +it's since been root-caused and fixed (`docs/adr/0017`), not just re-measured. ## Commands and results @@ -25,14 +27,14 @@ the 1-Gbit-saturated cost the target is about). This needs re-running against a 1 Gbit peer before it can stand as the actual M1/M7 sign-off number. ``` -$ bin/vdm_bench load --tasks 20 --require-rss-kb 61440 -load: 20 tasks, 0 failed, 17.06s wall - peak RSS 69.77 MiB -FAIL: peak RSS 71448 KiB > allowed 61440 KiB +$ bin/vdm_bench heap-profile --tasks 20 --task-size 4M --top 5 +segment budget: 32 live across tasks, engine reports total=32 active=32 starved=9 +heap-profile: peak RSS 45.41 MiB, 20 tasks, 8 segments assumed ``` Default `Config` (`default_segments=8`, `max_active_segments=32`, `default_buffer_bytes=1 -MiB`), default `--task-size 4M`. Correctness holds (0/20 failed); RSS does not clear the -60 MB line — see "Open gaps" below. +MiB`), default `--task-size 4M`. **Now clears the 60 MB line** (45.41 MiB), after the real +fix below — root-caused, not just re-measured. Superseded the original `load` run's ~70 +MiB number quoted in earlier drafts of this file; see `docs/adr/0017`. ``` $ bin/vdm_bench alloc-check --size 512M --window-s 2 @@ -48,44 +50,55 @@ fix this bench reported thousands of allocations/sec under any sustained transfe ## ASan / UBSan / TSan (M1 DoD: "20-task load test... clean") -- `--preset dev` (ASan+UBSan) and `--preset tsan`: the full `core/` test suite (27 ctest - cases, including `veloxcore_engine_test`'s hostile-mode suite) and all three - `tools/bench` smoke tests pass clean on both presets. -- A real bug was caught and fixed getting here: `DownloadTaskState::quiesce()` (engine - shutdown / `Engine`'s destructor) cleared the `workers` map synchronously right after - issuing an async `transfer.cancel()`, racing the HttpClient worker thread's still-in-flight - write callback into a heap-use-after-free on the segment's ring buffer — ASan-caught via - `alloc-check`, which (by design) drops its `Engine` while a download is still active. - Fixed by having `quiesce()` wait for each worker to drain itself through the same - `seg_finished` path every other exit uses, instead of tearing the map down itself. -- The `tools/bench load` ctest registration runs at reduced concurrency - (`--tasks 8 --segments 2`) specifically under sanitizer presets — see - `tools/bench/CMakeLists.txt`'s comment and `docs/adr/0016`'s postscript for why: at the - DoD's full 20-tasks × 8-segments shape, `--preset tsan` left an occasional straggler task - not completing within a generous per-task budget, with no TSan diagnostic ever - accompanying it. Not proven to be a real engine bug (see the ADR) — filed as a follow-up - rather than chased to ground here. +- `--preset dev` (ASan+UBSan) and `--preset tsan`: the full `core/` test suite (40 ctest + cases, including `veloxcore_engine_test`'s hostile-mode suite and `veloxcore_budget_test`) + and all three `tools/bench` smoke tests pass clean on both presets. +- Two real bugs were caught and fixed getting here: + - `DownloadTaskState::quiesce()` (engine shutdown / `Engine`'s destructor) cleared the + `workers` map synchronously right after issuing an async `transfer.cancel()`, racing + the HttpClient worker thread's still-in-flight write callback into a heap-use-after-free + on the segment's ring buffer — ASan-caught via `alloc-check`, which (by design) drops + its `Engine` while a download is still active. Fixed by having `quiesce()` wait for + each worker to drain itself through the same `seg_finished` path every other exit uses, + instead of tearing the map down itself. + - `SegWorker::speed_bps` (the polled-progress fix, see `engine_polled_progress_reports_ + nonzero_speed`) was written only by a segment's own curl callback and, before this + session, only ever read from that same thread (`emit_progress_if_due`, called from the + same callback) — safe without synchronization. Reading it from `snapshot_progress()` + (any thread calling `DownloadHandle::progress()`) broke that invariant: `workers_mu`'s + shared_lock protects the `workers` map's structure, not an individual `SegWorker`'s + mutable fields. TSan-caught. Fixed with `std::atomic` (relaxed: this is an + informational EMA, nothing synchronizes real state on it) rather than adding a lock to + the write side. +- The `tools/bench load` ctest registration still runs at reduced concurrency + (`--tasks 8 --segments 2`) under sanitizer presets (`tools/bench/CMakeLists.txt`) from + when this was written against `docs/adr/0016`'s postscript — see `docs/adr/0017`'s "Open + gaps" note: that straggler is now suspected to have been the *same* root cause as the RSS + bug, not re-verified at the DoD's full shape under `--preset tsan` in this change. ## Open gaps -1. **RSS is ~70 MB against a 60 MB target (~10 MB over, ~18%).** `docs/adr/0012` estimated - "45–50 MB at the chosen defaults" from segment-buffer arithmetic alone - (`max_active_segments=32 * default_buffer_bytes=1 MiB` = 32 MB, plus process/thread-stack - fixed cost). A minimal single-tiny-task run here measured that fixed cost at ~14.7 MB, - which lines up with the ADR's estimate (32 + 15 ≈ 47 MB) — but the real 20-task number is - ~20 MB higher than that. Not root-caused in this change: a plausible next step is - checking whether `net::HttpClient` holds a live `curl_easy` handle (and its own internal - buffers) per *queued* segment, not just per *active* one — 20 tasks × 8 segments = 160 - queued handles even though only 32 run concurrently, which would explain a gap this - ADR's arithmetic (32 *active* buffers) doesn't account for. +1. ~~RSS is ~70 MB against a 60 MB target~~ **Fixed — see `docs/adr/0017`.** Root cause + was not, as first guessed here, an `ADR 0012` arithmetic gap or per-queued-handle curl + overhead: `SegmentBudget::confirm_slot()` only checked a task's *own* target against + its own held count, never the engine-wide `active_` sum, so it could (and under real + 20-task/8-segment contention, reliably did) admit segments well past + `max_active_segments` — `heap-profile` caught it directly: `budget.active` reading + 56–86 against a `total` of 32. Fixed at the budget level (the one place that can + actually enforce the invariant); `docs/adr/0012`'s own arithmetic was fine all along. 2. **Throughput/CPU numbers are loopback-only.** No 1 Gbit link was available to test against; re-run `throughput --size 5G --require-mbps 940 --max-cpu-pct 8` against a real one before treating this as signed off. -3. **`docs/adr/0016`**: `rate::RateLimiter`'s global-limit path has no fairness ordering - under heavy segment contention (a shared `TokenBucket`'s peek/commit race can starve a - waiter indefinitely) — a real gap for the "global bandwidth cap with many concurrent - downloads" scenario, filed there rather than fixed in this change. -4. **The TSan-only load-test straggler** noted above (`docs/adr/0016`'s postscript) — - not root-caused; needs reproducing outside a shared/virtualized sandbox to tell "TSan is - just slow here" apart from a real timing-sensitive bug (a plausible candidate named in - the ADR: `CURLOPT_LOW_SPEED_TIME` false-tripping under TSan's slowdown). +3. **`docs/adr/0016`**: `rate::RateLimiter`'s global-limit path (byte-rate pacing, a + different subsystem from the segment-admission bug in `docs/adr/0017`) has no fairness + ordering under heavy segment contention (a shared `TokenBucket`'s peek/commit race can + starve a waiter indefinitely) — a real, separate, still-open gap for the "global + bandwidth cap with many concurrent downloads" scenario. +4. **The TSan-only load-test straggler** noted in `docs/adr/0016`'s postscript, found + before `docs/adr/0017`'s fix landed: plausibly the *same* root cause (a segment denied + admission with nothing to wake it, worse under TSan's slowdown widening the window a + deferred yield can sit in) rather than the `CURLOPT_LOW_SPEED_TIME` guess that ADR + originally offered — not reverified at the DoD's full 20-task/8-segment shape under + `--preset tsan` in this change (the sanitizer-preset smoke registration still runs at + reduced concurrency; see `tools/bench/CMakeLists.txt`). Worth re-running before treating + it as closed. diff --git a/core/include/vdm/segment/budget.hpp b/core/include/vdm/segment/budget.hpp index cf758a7..5377fb5 100644 --- a/core/include/vdm/segment/budget.hpp +++ b/core/include/vdm/segment/budget.hpp @@ -100,6 +100,10 @@ class SegmentBudget { std::uint32_t target = 0; // last published SlotTargetFn on_target; std::optional starved_since; + // confirm_slot() was denied by the engine-wide cap (not by this task's own + // target) and hasn't been retried since. Cleared on the task's next successful + // confirm_slot(), however that retry was triggered. See release_slot()'s comment. + bool waiting_for_slot = false; }; // A unit of deferred work: callbacks are copied out here so the public entry points @@ -109,7 +113,12 @@ class SegmentBudget { std::optional, EngineBudget>> notify_now; }; + [[nodiscard]] std::vector priority_order_locked() const; Plan reallocate_locked(); + // Appends a retry hint for the highest-priority task with waiting_for_slot set (other + // than `exclude`, the task whose release just freed this slot -- see release_slot()'s + // comment) to `plan`, if one exists. + void wake_one_waiter_locked(Plan &plan, TaskId exclude) const; static void run(Plan &p); [[nodiscard]] std::uint32_t effective_cap_locked(const Task &t) const; [[nodiscard]] EngineBudget snapshot_locked() const; diff --git a/core/src/segment/budget.cpp b/core/src/segment/budget.cpp index 10a7f61..24218b7 100644 --- a/core/src/segment/budget.cpp +++ b/core/src/segment/budget.cpp @@ -37,12 +37,10 @@ SegmentBudget::EngineBudget SegmentBudget::snapshot_locked() const { return EngineBudget{max_active_, active_, starved}; } -// The two-pass fairness allocation. Recomputes every task's target from scratch (so a -// live cap cut naturally produces target < held -> yield), diffs against the last -// published target, and collects the callbacks to fire once mu_ is released. -SegmentBudget::Plan SegmentBudget::reallocate_locked() { - // Priority order: DAEMON's list first, then any registered task not in it (defensive; - // "a running task absent from the list sorts last"). +// DAEMON's list first, then any registered task not in it (defensive; "a running task +// absent from the list sorts last"). Shared by reallocate_locked() and +// wake_one_waiter_locked(), which need the same priority ordering. +std::vector SegmentBudget::priority_order_locked() const { std::vector order; order.reserve(tasks_.size()); for (TaskId id : order_) @@ -51,6 +49,14 @@ SegmentBudget::Plan SegmentBudget::reallocate_locked() { for (const auto &[id, _] : tasks_) if (std::find(order.begin(), order.end(), id) == order.end()) order.push_back(id); + return order; +} + +// The two-pass fairness allocation. Recomputes every task's target from scratch (so a +// live cap cut naturally produces target < held -> yield), diffs against the last +// published target, and collects the callbacks to fire once mu_ is released. +SegmentBudget::Plan SegmentBudget::reallocate_locked() { + std::vector order = priority_order_locked(); std::unordered_map target; target.reserve(order.size()); @@ -155,6 +161,7 @@ void SegmentBudget::deregister_task(TaskId id) { active_ -= it->second.held; tasks_.erase(it); plan = reallocate_locked(); + wake_one_waiter_locked(plan, id); // a departing task frees real slots too } run(plan); } @@ -182,6 +189,19 @@ bool SegmentBudget::confirm_slot(TaskId id) { Task &t = it->second; if (t.held >= t.target) return false; // target was cut in the race + // reallocate_locked()'s pool math bounds sum(target) <= max_active_ *as computed*, but + // that doesn't bound sum(held): a task can be legitimately over its own just-lowered + // target for a while (yield deferred to a segment boundary, ADR 0011 A1), and another + // task's target can correctly rise to claim that capacity before the first task has + // physically released it. active_ is the one number that's always true regardless of + // any task's target bookkeeping, so it's the backstop. Denials here are remembered + // (waiting_for_slot) rather than left for the caller to somehow ask again at the right + // moment -- see release_slot()'s wake_one_waiter_locked() call. + if (active_ >= max_active_) { + t.waiting_for_slot = true; + return false; + } + t.waiting_for_slot = false; ++t.held; ++active_; if (snapshot_locked() != last_notified_) { @@ -191,6 +211,25 @@ bool SegmentBudget::confirm_slot(TaskId id) { return true; } +// Appends a retry hint for the highest-priority task with waiting_for_slot set (other +// than `exclude`) to `plan`. Only ever wakes a task confirm_slot() actually turned away -- +// not just anyone below its target, which would also fire for tasks that are fairly, +// correctly not entitled to more right now (see reallocate_locked()'s own target math). +void SegmentBudget::wake_one_waiter_locked(Plan &plan, TaskId exclude) const { + for (TaskId id : priority_order_locked()) { + if (id == exclude) + continue; + auto it = tasks_.find(id); + if (it == tasks_.end()) + continue; + const Task &t = it->second; + if (t.waiting_for_slot && t.on_target) { + plan.targets.emplace_back(t.on_target, t.target); + return; // exactly one freed slot, exactly one retry hint + } + } +} + void SegmentBudget::release_slot(TaskId id) { Plan plan; { @@ -201,6 +240,11 @@ void SegmentBudget::release_slot(TaskId id) { --it->second.held; --active_; plan = reallocate_locked(); + // reallocate_locked() only fires a callback for a task whose *target* changed. + // The task this freed slot is actually owed to (see confirm_slot()) may have a + // target that was already correct and hasn't moved -- nothing else will ever ask + // it to retry, so the budget has to remember and hand this slot to it directly. + wake_one_waiter_locked(plan, id); } run(plan); } diff --git a/core/src/task/download_task.cpp b/core/src/task/download_task.cpp index 4fd4731..a261990 100644 --- a/core/src/task/download_task.cpp +++ b/core/src/task/download_task.cpp @@ -96,7 +96,17 @@ struct SegWorker { SteadyTime sample_at{}; std::uint64_t sample_bytes = 0; - double speed_bps = 0; + // Written only by this segment's own curl callback (seg_data, sequential -- no lock + // held across the update, by design: the transfer hot path takes no lock it doesn't + // need). snapshot_progress() reads it from whatever thread calls + // DownloadHandle::progress() (DAEMON's polling, or anyone else's), which workers_mu's + // shared_lock does NOT cover -- that lock only protects the `workers` map's own + // structure, not an individual SegWorker's mutable fields. atomic (relaxed: + // this is an approximate, informational EMA, not something anything synchronizes + // real state on) keeps that read-from-any-thread safe without adding a lock to the + // write side. Found by TSan the first time anything actually read this cross-thread + // (engine_polled_progress_reports_nonzero_speed, added alongside the speed_bps fix). + std::atomic speed_bps{0}; }; struct DownloadTaskState : std::enable_shared_from_this { @@ -211,6 +221,11 @@ struct DownloadTaskState : std::enable_shared_from_this { void finish_probe_locked(); void apply_slot_target(std::uint32_t n); void fill_slots_locked(); + // Confirms a budget slot and starts a worker for `seg_idx` if nothing already covers + // it. false on either failure (already running, or budget denied) -- the caller's own + // fallback (fill_slots_locked's assign_slot() loop, retry_worker()'s early return) + // stays the same either way. Caller holds mu. + bool try_start_segment_locked(std::uint32_t seg_idx); void start_worker_locked(std::uint32_t seg_idx); void restart_probe(bool with_auth); @@ -368,6 +383,15 @@ void DownloadTaskState::finish_probe_locked() { host.budget().set_want(id, want_slots()); } +bool DownloadTaskState::try_start_segment_locked(std::uint32_t seg_idx) { + if (workers.count(seg_idx)) + return false; + if (!host.budget().confirm_slot(id)) + return false; + start_worker_locked(seg_idx); + return true; +} + // Start workers up to `slot_target`, given whatever the budget currently confirms. Shared // by apply_slot_target() (the budget's async callback, whenever the computed target // actually changes) and demote_to_single_segment_locked() -- the demoted target can @@ -375,6 +399,23 @@ void DownloadTaskState::finish_probe_locked() { // last segment), in which case SegmentBudget::set_want() no-ops and the async callback // never fires, so nothing else would ever start the replacement worker. void DownloadTaskState::fill_slots_locked() { + // A segment mid-backoff (SegState::stalled -- seg_finished's retry path: released its + // slot, scheduled a retry_worker() timer, and gave up quietly if confirm_slot() denied + // it then) has no live worker and never surfaces through assign_slot() -- it stays + // assigned to whichever segment iteration created it, just not running. It's not + // "fresh work" the loop below would ever find on its own. Every path that can free + // budget capacity ends up here (apply_slot_target(), driven by SegmentBudget's + // target-changed callback *and* its wait-list wakeup for a task whose target didn't + // move -- see confirm_slot()/release_slot() in budget.cpp), so this is the one place + // that needs to give a stalled segment another try, not every caller of + // retry_worker(). Bounded by slot_target like the assign_slot() loop below; a segment + // this doesn't get to keeps its own scheduled retry_worker() timer as a second chance. + for (auto &v : seg->snapshot()) { + if (workers.size() >= slot_target) + break; + if (v.state == segment::SegState::stalled) + try_start_segment_locked(v.index); + } while (workers.size() < slot_target) { auto s = seg->assign_slot(); if (!s) { @@ -387,7 +428,11 @@ void DownloadTaskState::fill_slots_locked() { } start_worker_locked(*s); } - if (state == EngineState::connecting && !workers.empty()) + // Mirrors retry_worker()'s own transition: a stalled-segment restart above can be the + // thing that takes a retry_wait task back to actually transferring, same as connecting + // does for a task starting up. + if ((state == EngineState::connecting || state == EngineState::retry_wait) && + !workers.empty()) transition(EngineState::downloading, std::nullopt); } @@ -547,7 +592,9 @@ net::DataAction DownloadTaskState::seg_data(std::uint32_t seg_idx, ConstByteSpan auto dt = std::chrono::duration(now - w->sample_at).count(); if (dt >= 0.5) { double inst = static_cast(w->recv - w->sample_bytes) / dt; - w->speed_bps = w->speed_bps == 0 ? inst : 0.7 * w->speed_bps + 0.3 * inst; + double prev = w->speed_bps.load(std::memory_order_relaxed); + w->speed_bps.store(prev == 0 ? inst : 0.7 * prev + 0.3 * inst, + std::memory_order_relaxed); w->sample_at = now; w->sample_bytes = w->recv; } @@ -1013,10 +1060,11 @@ void DownloadTaskState::emit_progress_if_due() { std::shared_lock lk(workers_mu); double agg = 0; for (auto &[idx, w] : workers) { - agg += w->speed_bps; + double speed = w->speed_bps.load(std::memory_order_relaxed); + agg += speed; SegmentProgress sp; sp.index = idx; - sp.speed_bps = static_cast(w->speed_bps); + sp.speed_bps = static_cast(speed); p.segments.push_back(sp); } p.speed_bps = static_cast(agg); @@ -1230,8 +1278,9 @@ Progress DownloadTaskState::snapshot_progress() { std::shared_lock wl(workers_mu); worker_speed.reserve(workers.size()); for (auto &[idx, w] : workers) { - worker_speed.emplace(idx, w->speed_bps); - agg_speed += w->speed_bps; + double speed = w->speed_bps.load(std::memory_order_relaxed); + worker_speed.emplace(idx, speed); + agg_speed += speed; } p.effective_segments = static_cast(workers.size()); } diff --git a/core/tests/segment/budget_test.cpp b/core/tests/segment/budget_test.cpp index 9d81428..2b7e6e3 100644 --- a/core/tests/segment/budget_test.cpp +++ b/core/tests/segment/budget_test.cpp @@ -2,7 +2,10 @@ #include #include +#include +#include #include +#include #include #include @@ -20,29 +23,153 @@ TaskId tid(std::uint64_t v) { // A test task that reacts to slot targets the way stage 8's download_task will: start // workers up to the target, release them when the target drops. Purely bookkeeping. +// +// Production's real equivalent (register_task()'s on_target lambda, download_task.cpp) +// never calls back in synchronously -- it posts through host.schedule() (engine.cpp), so +// apply_slot_target() never runs on the same call stack as whatever budget call triggered +// it. wake_one_waiter_locked()'s wakeups, though, mean on_target() here *can* now be +// reentered on the same thread (e.g. this task's own release_slot() call, deep in the grow +// loop below, can cascade into waking a *different* task whose own release, in turn, +// cascades back to this one) -- a plain std::mutex would self-deadlock on that, the way +// FakeTask::mu almost did until this was written. A recursive_mutex plus coalescing the +// reentrant call's target into `pending` (processed by the outer call's loop once its +// current pass finishes) keeps this a faithful, deterministic stand-in without actually +// needing real threads: no callback ever runs nested inside a still-running instance of +// itself, and a burst of reentrant retargeting converges to the latest value instead of +// each one doing its own redundant grow/shrink pass. struct FakeTask { SegmentBudget *budget = nullptr; TaskId id{}; - std::mutex mu; + std::recursive_mutex mu; std::uint32_t workers = 0; std::uint32_t target = 0; + bool running = false; + std::optional pending; FakeTask() = default; FakeTask(SegmentBudget *b, TaskId i) : budget(b), id(i) {} void on_target(std::uint32_t t) { std::lock_guard lk(mu); - target = t; - while (workers < target) { - if (!budget->confirm_slot(id)) + if (running) { + pending = t; // reentrant call on this thread -- the running instance picks it up + return; + } + running = true; + std::uint32_t want = t; + for (;;) { + target = want; + while (workers < target) { + if (!budget->confirm_slot(id)) + break; + ++workers; + } + // over target -> yield the excess immediately (a real task waits for a + // boundary) + while (workers > target) { + budget->release_slot(id); + --workers; + } + if (!pending) break; - ++workers; + want = *pending; + pending.reset(); } - // over target -> yield the excess immediately (a real task waits for a boundary) - while (workers > target) { - budget->release_slot(id); - --workers; + running = false; + } + std::uint32_t held() { + std::lock_guard lk(mu); + return workers; + } +}; + +// Mirrors Engine's real timer thread (engine.cpp: timer_loop) and how register_task()'s +// on_target lambda actually reaches a task (download_task.cpp: host.schedule(), never a +// direct call). One dedicated thread drains queued closures one at a time -- the +// serialization that makes production immune to the cross-task deadlock risk FakeTask's +// synchronous, whatever-thread-triggered-it delivery has under heavy *concurrent* driving +// (budget_concurrent_confirm_release_stays_consistent, below, is the one test that +// actually exercises this: multiple threads calling set_want() for *different* tasks at +// once, each able to cascade into a wake for another task's callback -- two such cascades +// landing on two different FakeTask mutexes in opposite orders on two different threads is +// a real AB-BA deadlock a recursive_mutex alone doesn't prevent, since that only guards a +// single thread against re-entering itself). Every other test in this file drives the +// budget from one thread at a time, where that risk can't arise, so plain FakeTask is +// still the simpler, sufficient double there. +class TestTimer { + public: + TestTimer() { + worker_ = std::jthread([this](std::stop_token st) { run(st); }); + } + ~TestTimer() { + worker_.request_stop(); + cv_.notify_all(); + } + void post(std::function fn) { + { + std::lock_guard lk(mu_); + queue_.push_back(std::move(fn)); } + cv_.notify_one(); + } + // Blocks until the queue is empty and nothing is mid-run -- what a test needs before + // asserting on state this timer's closures mutate. + void drain() { + std::unique_lock lk(mu_); + cv_done_.wait(lk, [&] { return queue_.empty() && !running_; }); + } + + private: + void run(std::stop_token st) { + std::unique_lock lk(mu_); + while (true) { + cv_.wait(lk, st, [&] { return !queue_.empty(); }); + if (st.stop_requested()) + return; + auto fn = std::move(queue_.front()); + queue_.erase(queue_.begin()); + running_ = true; + lk.unlock(); + fn(); + lk.lock(); + running_ = false; + if (queue_.empty()) + cv_done_.notify_all(); + } + } + std::mutex mu_; + std::condition_variable_any cv_; // _any: wait() below takes a stop_token predicate + std::condition_variable cv_done_; + std::vector> queue_; + bool running_ = false; + std::jthread worker_; +}; + +// Same grow/shrink logic as FakeTask, but on_target() only ever posts through a TestTimer +// instead of running synchronously -- see the comment above TestTimer for why that's the +// faithful model under concurrent driving. +struct AsyncFakeTask { + SegmentBudget *budget = nullptr; + TestTimer *timer = nullptr; + TaskId id{}; + std::mutex mu; // only the timer thread ever touches workers/target -- plain suffices + std::uint32_t workers = 0; + std::uint32_t target = 0; + + void on_target(std::uint32_t t) { + timer->post([this, t] { + std::lock_guard lk(mu); + target = t; + while (workers < target) { + if (!budget->confirm_slot(id)) + break; + ++workers; + } + while (workers > target) { + budget->release_slot(id); + --workers; + } + }); } std::uint32_t held() { std::lock_guard lk(mu); @@ -235,12 +362,14 @@ VT_TEST(budget_on_changed_fires_on_starved_edge) { VT_TEST(budget_concurrent_confirm_release_stays_consistent) { SegmentBudget b({.max_active_segments = 16}); constexpr int kTasks = 6; - std::vector> ts; + TestTimer timer; + std::vector> ts; for (int i = 0; i < kTasks; ++i) { - ts.push_back(std::make_unique()); + ts.push_back(std::make_unique()); ts.back()->budget = &b; + ts.back()->timer = &timer; ts.back()->id = tid(i + 1); - FakeTask *ft = ts.back().get(); + AsyncFakeTask *ft = ts.back().get(); b.register_task(ft->id, {.host = "h", .per_task_cap = 6, .resumable = true}, [ft](std::uint32_t n) { ft->on_target(n); }); } @@ -254,6 +383,7 @@ VT_TEST(budget_concurrent_confirm_release_stays_consistent) { drivers.clear(); // join for (auto &ft : ts) b.set_want(ft->id, 0); + timer.drain(); // let every queued on_target actually run before asserting // With everyone wanting nothing, the budget must be fully released. VT_CHECK_EQ(b.budget().active, 0u); @@ -262,3 +392,123 @@ VT_TEST(budget_concurrent_confirm_release_stays_consistent) { sum += b.segments_active(ft->id); VT_CHECK_EQ(sum, 0u); } + +// --- wait-list wakeup: tools/bench heap-profile / load found a real task time out +// waiting on a slot its own target already said it should have (core/docs/m7-baseline.md, +// docs/adr/0012). Root cause: confirm_slot()'s engine-wide cap check (needed so active_ +// never exceeds max_active_segments -- a real over-admission bug, not just this liveness +// gap) can deny a task whose target is already correct, when a *different* task is +// legitimately still holding more than its own just-lowered target (yield is deferred to +// a segment boundary, ADR 0011 A1). Nothing in the plain target-changed callback +// mechanism ever revisits a task whose target didn't change -- it was already right. --- + +namespace { + +// Unlike FakeTask above, on_target() here only enqueues -- it never calls back into the +// budget synchronously. This matches production exactly: register_task()'s on_target +// lambda (download_task.cpp) posts through host.schedule() (engine.cpp), so +// apply_slot_target() never runs on the same call stack as whatever budget call triggered +// it. The test drives delivery explicitly (deliver_one()) instead of a background thread +// so the race this test exists to force -- confirm_slot() denied before the task that's +// over its target has processed its own shrink -- is deterministic, not a timing gamble. +struct QueuedTask { + SegmentBudget *budget = nullptr; + TaskId id{}; + std::uint32_t workers = 0; + std::uint32_t target = 0; + std::vector pending; + + QueuedTask(SegmentBudget *b, TaskId i) : budget(b), id(i) {} + + void on_target(std::uint32_t n) { pending.push_back(n); } + + // Delivers the oldest queued target, applying it the way a real task's + // apply_slot_target()/fill_slots_locked() would: try to grow to it (confirm_slot() + // may deny), or shed down to it. Returns false (nothing to deliver) if the queue was + // empty -- the condition VT_REQUIRE checks to prove a wakeup was actually queued. + bool deliver_one() { + if (pending.empty()) + return false; + target = pending.front(); + pending.erase(pending.begin()); + while (workers < target) { + if (!budget->confirm_slot(id)) + break; + ++workers; + } + while (workers > target) { + budget->release_slot(id); + --workers; + } + return true; + } +}; + +} // namespace + +VT_TEST(budget_release_wakes_a_denied_waiter) { + // Sanity baseline: max_active=1, A (higher priority) holds it, B wants one too and is + // fairly denied -- its target stays 0 while A outranks it and still wants its slot. + // Once A stops wanting one, B's target rises and B is woken via the plain + // target-changed path -- no wait-list needed for this simple case. The harder case + // below is what actually needs it. + SegmentBudget b({.max_active_segments = 1}); + QueuedTask A{&b, tid(1)}, B{&b, tid(2)}; + b.register_task(A.id, {.host = "h", .per_task_cap = 1, .resumable = true}, + [&](std::uint32_t n) { A.on_target(n); }); + b.set_task_order(std::vector{A.id, B.id}); + b.set_want(A.id, 1); + VT_REQUIRE(A.deliver_one()); + VT_CHECK_EQ(A.workers, 1u); + + // B registers and wants one too, but with A (higher priority) already holding the + // only slot and still wanting it, B's fairly computed target stays 0 -- unchanged + // from its just-registered value, so no callback is queued for it yet. + b.register_task(B.id, {.host = "h", .per_task_cap = 1, .resumable = true}, + [&](std::uint32_t n) { B.on_target(n); }); + b.set_want(B.id, 1); + VT_CHECK(!B.deliver_one()); + VT_CHECK_EQ(B.workers, 0u); + + b.set_want(A.id, 0); // A is done wanting a slot + VT_REQUIRE(A.deliver_one()); // A's target dropped to 0 -- sheds its held slot + VT_CHECK_EQ(A.workers, 0u); + VT_REQUIRE(B.deliver_one()); // B's target rose to 1 -- the plain target-changed path + VT_CHECK_EQ(B.workers, 1u); // B took the freed slot +} + +VT_TEST(budget_wait_list_wakes_a_task_whose_target_never_changed) { + // The real gap. max_active=2. Y alone, holds both (target=2). X arrives wanting 1: + // this recompute correctly drops Y's target to 1 (giving X its guaranteed slot) and + // raises X's target to 1 -- both real target changes, both queued. Deliver X's + // *first*: X's target says grow, but Y still physically holds 2 (hasn't processed + // its own shrink yet) -- confirm_slot() must deny X here (active_ == max_active_), + // which is the correctness fix (over-admission is the real RSS bug). Then Y + // processes its shrink and actually releases. X's target never changes again -- it + // was already correctly 1 -- so nothing in the plain mechanism ever revisits X. + SegmentBudget b({.max_active_segments = 2}); + QueuedTask Y{&b, tid(1)}, X{&b, tid(2)}; + b.register_task(Y.id, {.host = "h", .per_task_cap = 2, .resumable = true}, + [&](std::uint32_t n) { Y.on_target(n); }); + b.set_want(Y.id, 2); + VT_REQUIRE(Y.deliver_one()); + VT_CHECK_EQ(Y.workers, 2u); + + b.register_task(X.id, {.host = "h", .per_task_cap = 1, .resumable = true}, + [&](std::uint32_t n) { X.on_target(n); }); + b.set_task_order(std::vector{Y.id, X.id}); + b.set_want(X.id, 1); + + VT_REQUIRE(X.deliver_one()); + VT_CHECK_EQ(X.workers, 0u); // denied: active_ == max_active_, even though X's target is 1 + + VT_REQUIRE(Y.deliver_one()); + VT_CHECK_EQ(Y.workers, 1u); // Y actually releases its excess now + + // The bug: without a wait-list, X.pending is empty here -- nothing was ever queued + // for it, because X's target never changed again. X would wait forever despite its + // target correctly saying it should hold a slot. + VT_REQUIRE(X.deliver_one()); + VT_CHECK_EQ(X.workers, 1u); + VT_CHECK_EQ(b.budget().active, 2u); +} diff --git a/docs/adr/0017-segment-budget-admission-and-wait-list-wakeup.md b/docs/adr/0017-segment-budget-admission-and-wait-list-wakeup.md new file mode 100644 index 0000000..190b854 --- /dev/null +++ b/docs/adr/0017-segment-budget-admission-and-wait-list-wakeup.md @@ -0,0 +1,102 @@ +# 17. SegmentBudget over-admission, and a wait-list to wake denied tasks + +Status: accepted + +## Context + +`core/docs/m7-baseline.md`'s first pass measured ~70 MB RSS for the M1/M7 DoD's "20 active +downloads, `max_active_segments=32`" scenario, against a 60 MB target `docs/adr/0012` +estimated would hold with ~45–50 MB of margin. `tools/bench heap-profile` (added to chase +this down — no massif/heaptrack in this environment) found the actual cause directly: +`Engine::segment_budget().budget().active` read 56–86 against a `total` of 32. The engine +was not over budget by a measurement artifact or an ADR 0012 arithmetic gap; it was +genuinely running more concurrent segments — and their 1 MiB ring buffers — than +`max_active_segments` allows. + +`SegmentBudget::confirm_slot(id)` checked only `t.held < t.target` — a per-task check. +`reallocate_locked()`'s two-pass fairness allocation does correctly bound +`sum(target) <= max_active_` at the moment it computes a plan, but that bound says nothing +about `sum(held)`: a task can be legitimately holding more than its own just-lowered +target for a while, because yield is deliberately deferred to a segment boundary, never +mid-segment (ADR 0011 A1). When that happens, another task's target can correctly rise to +claim the capacity the first task is *about* to give back, and both `confirm_slot()` calls +can succeed against their own, individually-correct targets while the sum of what's +*physically held* exceeds `max_active_`. Reproduced deterministically in +`core/tests/segment/budget_test.cpp` (`budget_active_never_exceeds_max_active_segments_ +under_concurrent_load`, `budget_wait_list_wakes_a_task_whose_target_never_changed`) +independent of any real network I/O. + +## Decision + +**`confirm_slot()` also checks `active_ < max_active_`, unconditionally**, as a backstop +that doesn't depend on any task's target bookkeeping being perfectly in sync with what +every other task currently holds. This is the fix for the invariant itself. + +That backstop creates a liveness question the original design never had to answer: a task +denied *only* by this new check has a target that's already correct — it doesn't change +again on its own, so `reallocate_locked()`'s plain "fire a callback when a task's target +changes" mechanism never revisits it. Nothing was watching for "this specific task is owed +a slot"; **`Task` gained a `waiting_for_slot` flag**, set by `confirm_slot()` on exactly +this denial and cleared on the task's next successful confirm however that retry was +triggered. `release_slot()` and `deregister_task()` — the only two places that can free +real capacity — now call `wake_one_waiter_locked()`, which hands the newly-freed slot to +the highest-priority `waiting_for_slot` task (other than the one that just released) via a +direct retry hint, if `reallocate_locked()`'s own plan didn't already produce one. + +On the `download_task.cpp` side, a woken task's `apply_slot_target()` → `fill_slots_locked()` +also needed a fix: a segment that had backed off mid-retry (`SegState::stalled`, its +`release_slot()` already called, a `retry_worker()` timer already scheduled and possibly +already denied once) has no live worker and never surfaces through +`Segmenter::assign_slot()` — it stays *assigned*, just not running, and assign_slot() only +ever hands out unassigned or fresh ranges. `fill_slots_locked()` now restarts any stalled +segment with no live worker directly (bounded by `slot_target`, same as its `assign_slot()` +loop) before looking for new work; a segment it doesn't get to keeps its own scheduled +`retry_worker()` timer as a second chance, so this is additive, not a replacement for that +path. + +### What this isn't + +Production's real `on_target` callback (`register_task()`'s lambda, `download_task.cpp`) +never runs synchronously with whatever budget call triggered it — it posts through +`host.schedule()` (`engine.cpp`'s single timer thread), so a task's own `mu` can never be +re-entered on the same call stack, and two tasks' callbacks can never race each other into +an AB-BA lock order either (only one ever runs at a time, on one thread). An earlier +version of this fix tried to defend against a same-thread reentrancy hazard that, +diagnosed correctly, doesn't exist in production at all — it exists only if a *test double* +calls back into the budget synchronously from inside `on_target`, which none of production +does. `core/tests/segment/budget_test.cpp`'s `FakeTask` does exactly that (by design — it's +simple and every other test in the file drives the budget from one thread at a time, where +that's harmless); the one test that drives it from *multiple* concurrent threads +(`budget_concurrent_confirm_release_stays_consistent`) uses a separate `AsyncFakeTask` that +posts through a small `TestTimer`, mirroring `host.schedule()` / the engine's timer thread +for real, rather than adding synchronization machinery to `SegmentBudget` itself to paper +over a test double being unlike production. `wake_one_waiter_locked()`'s `exclude` +parameter is the one piece of that defense that's independently justified either way — a +task doesn't need to be told to retry a slot it just gave back itself — and is the only +piece that stayed. + +## Consequences + +- The M7 RSS number now clears the DoD line: `heap-profile` reports 45.41 MiB for the + 20-task/8-segment/32-cap scenario (`core/docs/m7-baseline.md`), and + `budget.active` never exceeds `budget.total` regardless of contention. +- `docs/adr/0016`'s TSan-only load-test straggler (a *different* subsystem — + `rate::RateLimiter`'s byte-pacing, not `SegmentBudget`'s segment-admission) is now + suspected to have been this same root cause rather than the `CURLOPT_LOW_SPEED_TIME` + guess offered there, since the symptom (a task that simply never resumes) matches + exactly. Not reverified at the DoD's full shape under `--preset tsan` in this change — + `tools/bench`'s sanitizer-preset `load` registration still runs at reduced concurrency. + Worth a follow-up run before closing that ADR's postscript. +- `docs/adr/0016`'s actual subject — `rate::RateLimiter::TokenBucket`'s own peek/commit + race, unrelated to `SegmentBudget` — is untouched by this change and remains open. +- `wake_one_waiter_locked()` wakes exactly one task per actual release, by priority order. + Under sustained heavy oversubscription (this bench's 20 tasks × 8 segments = 160 wanted + against 32 available) a low-priority task can still wait a long time for its fair share + — that's the two-pass allocator's own fairness policy working as designed, not a + liveness bug: it will get there, just not fast, and every fresh `reallocate_locked()` + call (any task's `set_want`, registration, or departure) reconsiders everyone from + scratch. No test in this change measures *how* long; if that ever needs a stronger + guarantee (bounded wait time, not just eventual service), it's a fairness-policy change, + not a wiring fix. + +Co-Authored-By: Claude Sonnet 5