core: add a state dump to tools/bench load on task timeout

Per-task engine state, downloaded/effective_segments, every segment's own
state, and the engine's SegmentBudget snapshot -- printed once, when a task
times out, instead of needing to re-run the bench under a debugger or add
throwaway instrumentation to find out why. This is what actually diagnosed
the SegmentBudget over-admission bug fixed in the previous commit: the dump
showed budget.active pinned at max_active_segments while multiple tasks sat
starved, which is what pointed straight at confirm_slot()'s missing
engine-wide check rather than a per-task target bug.

Also updates the vdm_bench_load20 ctest registration's comment: the
straggler under --preset tsan that motivated running it at reduced
concurrency (docs/adr/0016's postscript) is now suspected to have been the
same SegmentBudget bug (docs/adr/0017), not the TSan-timing artifact first
guessed -- not reverified at the DoD's full shape under TSan in this change,
so the reduced-concurrency registration stays for now.

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Q3QrF7rCt21bkAjt9BCDFQ

tools/bench/CMakeLists.txt

@@ -37,17 +37,18 @@ if(VELOX_BUILD_TESTS)
     set_tests_properties(vdm_bench_throughput_smoke PROPERTIES LABELS "bench" TIMEOUT 120)
 
     # --tasks 8 --segments 2 (not the DoD's 20 tasks * default_segments=8 = 160 concurrent
-    # segments): at the full shape, TSan's per-access instrumentation overhead was observed
+    # segments): at the full shape under --preset tsan, a straggler task was observed not
-    # to leave a straggler task not just slow but still incomplete past a 300s-per-task
+    # just slow but still incomplete past a 300s-per-task budget (docs/adr/0016's
-    # budget -- reproduced at tasks=20/segments=8 (2 stragglers) and, smaller but still
+    # postscript, written before docs/adr/0017's SegmentBudget fix landed) -- now suspected
-    # present, at tasks=20/segments=2 (1 straggler); tasks=8/segments=2 (16 concurrent
+    # to have been that same over-admission bug (a segment denied a slot with nothing to
-    # connections) was reliable across repeated runs. No TSan report ever accompanied a
+    # wake it), not a TSan-timing artifact as first guessed, since the symptom -- a task
-    # straggler (this isn't a race -- see docs/adr/0016's postscript), so it reads as some
+    # that simply never resumes -- matches exactly. Left at this reduced concurrency rather
-    # combination of TSan's overhead and this environment's scheduling, not an engine bug;
+    # than reverting: re-verifying the full shape is clean under TSan wasn't done as part
-    # still, "every task completes" is exactly what this smoke test is supposed to check
+    # of that fix (see docs/adr/0017's "Consequences"), so this is still the bar that's
-    # (see the split above), so the bar it runs at has to be one that actually holds. The
+    # actually known to hold. "Every task completes" is what this smoke test exists to
-    # DoD's real 20-task/default-segments/60MB-RSS shape is exercised by the manual/CI M7
+    # check (see the split above); the DoD's real 20-task/default-segments/60MB-RSS shape
-    # sign-off run in this file's header comment, at --preset release, where it passes.
+    # is exercised by the manual/CI M7 sign-off run in this file's header comment, at
+    # --preset release, where it passes (core/docs/m7-baseline.md).
     add_test(NAME vdm_bench_load20 COMMAND vdm_bench load --tasks 8 --task-size 2M --segments 2)
     set_tests_properties(vdm_bench_load20 PROPERTIES LABELS "bench" TIMEOUT 300)
 

tools/bench/vdm_bench.cpp

@@ -424,6 +424,42 @@ int cmd_throughput(const Args &a) {
     return rc;
 }
 
+const char *seg_state_name(segment::SegState s) {
+    switch (s) {
+        case segment::SegState::idle: return "idle";
+        case segment::SegState::connecting: return "connecting";
+        case segment::SegState::downloading: return "downloading";
+        case segment::SegState::stalled: return "stalled";
+        case segment::SegState::complete: return "complete";
+        case segment::SegState::failed: return "failed";
+    }
+    return "?";
+}
+
+// Printed once, on a task timeout, for whatever hang it wasn't designed to reproduce --
+// per task: engine state, worker/segment count, and every segment's own state; plus the
+// budget's own view, engine-wide. Enough to tell "stuck starved" (budget.active at cap,
+// nobody holding this task's segments) from "stuck stalled" (a segment sitting in
+// `stalled` with no worker, budget has room) from anything else, without re-deriving it by
+// re-running the bench.
+void dump_state(Engine &eng, const std::vector<DownloadHandle> &handles, int timed_out_task) {
+    std::fprintf(stderr, "---- state dump (task %d timed out) ----\n", timed_out_task);
+    auto eb = eng.segment_budget().budget();
+    std::fprintf(stderr, "budget: total=%u active=%u starved=%u\n", eb.total, eb.active,
+                eb.tasks_starved);
+    for (std::size_t i = 0; i < handles.size(); ++i) {
+        auto p = handles[i].progress();
+        std::fprintf(stderr, "task %zu: state=%d downloaded=%llu/%llu effective_segments=%u\n",
+                    i, static_cast<int>(handles[i].state()), (unsigned long long)p.downloaded,
+                    (unsigned long long)p.total.value_or(0), p.effective_segments);
+        for (auto &sp : p.segments)
+            std::fprintf(stderr, "  seg %u: state=%s completed=%llu speed_bps=%llu\n", sp.index,
+                        seg_state_name(sp.state), (unsigned long long)sp.completed,
+                        (unsigned long long)sp.speed_bps);
+    }
+    std::fprintf(stderr, "----------------------------------------\n");
+}
+
 int cmd_load(const Args &a) {
     const int tasks = static_cast<int>(a.get_long("--tasks", 20));
     const std::uint64_t task_size = a.get_size("--task-size", "4M");
@@ -501,6 +537,7 @@ int cmd_load(const Args &a) {
     for (int i = 0; i < tasks; ++i) {
         if (futs[i].wait_for(task_timeout) != std::future_status::ready) {
             std::fprintf(stderr, "task %d: timed out\n", i);
+            dump_state(eng, handles, i);
             ++failures;
             continue;
         }