Files
samiandClaude Sonnet 5 f2ee45f818 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
2026-09-12 10:58:32 +04:00

772 lines
34 KiB
C++

// tools/bench/vdm_bench.cpp — the M1/M7 performance gates from docs/04-engine-design.md §8,
// and the sanitizer-clean concurrent-load regression that falls out of the same harness.
//
// Three subcommands, one binary, one way of driving vdm::Engine and reading back
// wall-clock/CPU/RSS:
//
// vdm_bench throughput [--size 5G] [--segments N] [--require-mbps 125] [--max-cpu-pct 8]
// A single download. Reports achieved throughput and process CPU as a percentage of
// one core. The M1 DoD line is "5 GB saturates a 1 Gbit link at <=8% of one core" --
// pass --size 5G --require-mbps 940 --max-cpu-pct 8 against a real link for the
// actual sign-off. Against the bundled local server (see support/local_server.hpp)
// the numbers are still meaningful for regression tracking; they just aren't a
// real-network measurement, which is why the ctest-registered run below doesn't gate
// on them.
//
// vdm_bench load [--tasks 20] [--task-size 4M] [--segments N] [--require-rss-kb N]
// docs/04 §8's "<=60 MB RSS with 20 active downloads at default buffers, given
// max_active_segments=32" scenario: one Engine, default Config, N concurrent tasks,
// peak RSS read back via getrusage(). This is the same binary the ctest below runs
// under ASan/UBSan/TSan as the M1 DoD's "20-task load test" -- there the job is
// purely "no sanitizer error, every task completes correctly"; --require-rss-kb is
// for a --preset release run, where the number means something (a sanitizer roughly
// doubles-to-quadruples RSS via redzones/shadow memory).
//
// vdm_bench heap-profile [--tasks 20] [--task-size 4M] [--segments N] [--top N]
// Answers "where does the RSS actually go", for the M7 RSS target (docs/04 §8) --
// no massif/heaptrack in this environment (no root to install them), so this is a
// small in-process equivalent: the same operator new/delete override tracks, per
// allocation, which call site made it (one return address, via
// __builtin_return_address(0) right in operator new -- cheap enough to run for the
// whole scenario, unlike a full backtrace) and keeps a live (not-yet-freed) byte
// count per site. Runs the same 20-task concurrent scenario as `load`, waits for
// peak RSS to stop growing, then prints the top sites by live bytes -- the
// breakdown ADR 0012's "45-50 MB" estimate needs to be checked against, and
// core/docs/m7-baseline.md's ~70 MB measurement needs explained.
//
// vdm_bench alloc-check [--size 128M] [--window-s 2]
// docs/agents/AGENT-CORE.md: "no allocation in the curl write callback... checked in
// review and by a bench assertion." operator new/delete are overridden process-wide
// below; this samples the count across a steady mid-transfer window (no segment
// start/stop, so no probe/segmenter/sidecar activity) and requires it stay within a
// small time-proportional budget -- not a strict zero, because emit_progress_if_due()
// legitimately builds a Progress::segments vector up to 4x/sec regardless of
// throughput. What must NOT happen is that count scaling with bytes transferred; the
// budget is sized so it can't, while tolerating that fixed, small, rate-independent
// bookkeeping cost. The transfer is paced by tools/testserver's `throttled` mode
// (support/testserver_client.hpp), not the engine's own rate limiter -- the
// limiter's pause/resume path allocates on every throttle event, which would measure
// the limiter instead of the write path.
//
// Every subcommand is report-only (exit 0 once the download(s) complete correctly) unless
// its --require-* flag is passed, so the ctest registrations in CMakeLists.txt are stable
// under CI's shared, sanitizer-slowed, virtualized hardware.
#include "vdm/engine.hpp"
#include <cxxabi.h>
#include <dlfcn.h>
#include <sys/resource.h>
#include <unistd.h>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <future>
#include <mutex>
#include <new>
#include <string>
#include <unordered_map>
#include <vector>
#include "support/local_server.hpp"
#include "support/testserver_client.hpp"
using namespace vdm;
using namespace vdm::task;
using namespace std::chrono_literals;
// --- allocation tracking: a plain counter for alloc-check, and (only when heap-profile
// turns it on) a per-call-site live-byte tracker standing in for massif/heaptrack, neither
// of which is installable here (no root). Both share one operator new/delete override
// since a process gets exactly one. ---------------------------------------------------
namespace {
std::atomic<std::uint64_t> g_alloc_count{0};
std::atomic<bool> g_site_tracking{false};
// Re-entrancy guard: our own bookkeeping below calls into std::unordered_map, which
// allocates. Without this, that nested operator new call would recurse into the same
// tracking logic -- not infinitely (the nested call's own bookkeeping call would itself be
// guarded the same way, so it terminates), but it would count the tracker's own node
// allocations as if they were the program's, polluting the profile. When set, an
// allocation is just satisfied via malloc/free with no bookkeeping.
thread_local bool g_in_tracker = false;
struct LiveAlloc {
std::size_t size;
void *site; // __builtin_return_address(0) at the call to operator new
};
struct SiteStats {
std::uint64_t live_bytes = 0;
std::uint64_t alloc_count = 0; // lifetime, not just live -- how *often* a site fires
};
std::mutex g_track_mu;
std::unordered_map<void *, LiveAlloc> g_live; // pointer -> what/where
std::unordered_map<void *, SiteStats> g_site_stats; // call site -> aggregate
void track_alloc(void *p, std::size_t n) {
if (!g_site_tracking.load(std::memory_order_relaxed) || g_in_tracker)
return;
g_in_tracker = true;
void *site = __builtin_return_address(0);
{
std::lock_guard lk(g_track_mu);
g_live.emplace(p, LiveAlloc{n, site});
auto &st = g_site_stats[site];
st.live_bytes += n;
st.alloc_count += 1;
}
g_in_tracker = false;
}
void track_free(void *p) {
if (!p || !g_site_tracking.load(std::memory_order_relaxed) || g_in_tracker)
return;
g_in_tracker = true;
{
std::lock_guard lk(g_track_mu);
if (auto it = g_live.find(p); it != g_live.end()) {
g_site_stats[it->second.site].live_bytes -= it->second.size;
g_live.erase(it);
}
}
g_in_tracker = false;
}
std::string describe_site(void *site); // defined below; used as a per-site fallback here
// Resolves a batch of call-site addresses to "function (file:line)" via one `addr2line`
// invocation (reads .symtab + DWARF, unlike dladdr below -- most of this codebase's
// call sites are internal-linkage or file-static and never make it into the *dynamic*
// symbol table dladdr is limited to). `sites` and the returned vector are in the same
// order. dladdr's `dli_fbase` (the object's runtime load base -- works even when it can't
// name the specific symbol, which is exactly the "site not in .dynsym" case this exists
// for) converts each ASLR'd runtime address into the file-relative offset addr2line wants.
// Falls back to a raw "0x..." string per site if addr2line isn't on PATH or the batch call
// otherwise fails -- describe_site() below still has dladdr as a second-line fallback for
// an individual site in that case.
std::vector<std::string> resolve_sites_via_addr2line(const std::vector<void *> &sites) {
std::vector<std::string> out(sites.size());
for (std::size_t i = 0; i < sites.size(); ++i) {
char buf[32];
std::snprintf(buf, sizeof buf, "%p", sites[i]);
out[i] = buf;
}
if (sites.empty())
return out;
// `-e /proc/self/exe` would resolve inside the *addr2line* process this spawns, i.e.
// to addr2line's own binary, not vdm_bench's -- readlink it here, in-process, first.
char exe_path[4096];
ssize_t exe_len = ::readlink("/proc/self/exe", exe_path, sizeof(exe_path) - 1);
if (exe_len <= 0)
return out;
exe_path[exe_len] = '\0';
std::string cmd = std::string("addr2line -f -C -e '") + exe_path + "'";
for (void *site : sites) {
Dl_info info{};
std::uintptr_t off = reinterpret_cast<std::uintptr_t>(site);
if (::dladdr(site, &info) && info.dli_fbase)
off -= reinterpret_cast<std::uintptr_t>(info.dli_fbase);
char buf[24];
std::snprintf(buf, sizeof buf, " %zx", off);
cmd += buf;
}
cmd += " 2>/dev/null";
FILE *pipe = ::popen(cmd.c_str(), "r");
if (!pipe)
return out;
std::vector<std::string> lines;
char line[1024];
while (std::fgets(line, sizeof line, pipe)) {
std::string s = line;
while (!s.empty() && (s.back() == '\n' || s.back() == '\r')) s.pop_back();
lines.push_back(std::move(s));
}
::pclose(pipe);
// addr2line -f prints two lines per address: function name, then file:line.
for (std::size_t i = 0; i * 2 + 1 < lines.size() && i < sites.size(); ++i) {
const std::string &fn = lines[i * 2];
const std::string &loc = lines[i * 2 + 1];
if (fn == "??" && loc == "??:0")
out[i] = describe_site(sites[i]); // addr2line drew a blank -- try dladdr
else
out[i] = fn + " (" + loc + ")";
}
return out;
}
// Single-site fallback (dladdr only) for when addr2line can't place an address (it read
// only .symtab -- e.g. no debug/symbol data at all for that address) or isn't available.
std::string describe_site(void *site) {
Dl_info info{};
if (!::dladdr(site, &info) || !info.dli_sname) {
char buf[32];
std::snprintf(buf, sizeof buf, "%p", site);
return buf;
}
std::string name = info.dli_sname;
int status = 0;
if (char *dem = abi::__cxa_demangle(name.c_str(), nullptr, nullptr, &status);
dem && status == 0) {
name = dem;
std::free(dem);
}
auto off = static_cast<std::uintptr_t>(reinterpret_cast<char *>(site) -
reinterpret_cast<char *>(info.dli_saddr));
return name + " +0x" + [&] {
char buf[24];
std::snprintf(buf, sizeof buf, "%zx", off);
return std::string(buf);
}();
}
} // namespace
void *operator new(std::size_t n) {
g_alloc_count.fetch_add(1, std::memory_order_relaxed);
void *p = std::malloc(n ? n : 1);
if (!p)
throw std::bad_alloc();
track_alloc(p, n);
return p;
}
// The nothrow overload is a distinct allocation function from the plain one above, not an
// alternate spelling of it: leaving it un-overridden means it falls through to ASan's own
// default (nothrow) operator new, while a deallocation of that same pointer still reaches
// the plain `operator delete` override below -- a real alloc-dealloc-mismatch ASan catches
// (hit via std::stable_sort's std::get_temporary_buffer, which allocates this way; see
// core/src/segment/budget.cpp's SegmentBudget::run() for why that path no longer uses
// stable_sort at all, but this override is fixed too since anything else pulling in a
// nothrow allocation -- directly or via another std:: temp-buffer user -- would hit it
// again otherwise).
void *operator new(std::size_t n, const std::nothrow_t &) noexcept {
g_alloc_count.fetch_add(1, std::memory_order_relaxed);
void *p = std::malloc(n ? n : 1);
if (p)
track_alloc(p, n);
return p;
}
void operator delete(void *p) noexcept {
track_free(p);
std::free(p);
}
void operator delete(void *p, std::size_t) noexcept {
track_free(p);
std::free(p);
}
void operator delete(void *p, const std::nothrow_t &) noexcept {
track_free(p);
std::free(p);
}
namespace {
// --- small helpers ---------------------------------------------------------------------
std::uint64_t parse_size(std::string_view s) {
if (s.empty())
return 0;
char suffix = s.back();
std::uint64_t mult = 1;
std::string_view digits = s;
if (suffix == 'k' || suffix == 'K') {
mult = 1024;
digits.remove_suffix(1);
} else if (suffix == 'm' || suffix == 'M') {
mult = 1024ull * 1024;
digits.remove_suffix(1);
} else if (suffix == 'g' || suffix == 'G') {
mult = 1024ull * 1024 * 1024;
digits.remove_suffix(1);
}
return std::strtoull(std::string(digits).c_str(), nullptr, 10) * mult;
}
// Trivial `--flag value` / `--flag` (bool) parser: no library dependency worth adding for
// a handful of options across three subcommands.
class Args {
public:
Args(int argc, char **argv, int start) {
for (int i = start; i < argc; ++i) raw_.emplace_back(argv[i]);
}
[[nodiscard]] std::string get(std::string_view flag, std::string def) const {
for (std::size_t i = 0; i < raw_.size(); ++i)
if (raw_[i] == flag && i + 1 < raw_.size())
return raw_[i + 1];
return def;
}
[[nodiscard]] std::uint64_t get_size(std::string_view flag, std::string def) const {
return parse_size(get(flag, std::move(def)));
}
[[nodiscard]] double get_double(std::string_view flag, double def) const {
auto s = get(flag, "");
return s.empty() ? def : std::strtod(s.c_str(), nullptr);
}
[[nodiscard]] long get_long(std::string_view flag, long def) const {
auto s = get(flag, "");
return s.empty() ? def : std::strtol(s.c_str(), nullptr, 10);
}
private:
std::vector<std::string> raw_;
};
struct Rusage {
double cpu_s;
long peak_rss_kb;
};
Rusage sample_rusage() {
struct ::rusage ru {};
::getrusage(RUSAGE_SELF, &ru);
double cpu = (double)ru.ru_utime.tv_sec + ru.ru_utime.tv_usec / 1e6 +
(double)ru.ru_stime.tv_sec + ru.ru_stime.tv_usec / 1e6;
return {cpu, ru.ru_maxrss}; // ru_maxrss is KiB on Linux, and is a lifetime peak, not
// a snapshot -- fine for us, we only ever want the peak.
}
// Synchronous single-download driver: start it, block for on_finished. Used by throughput
// and alloc-check, which only ever run one transfer at a time.
Result<DownloadOutcome> run_one(Engine &eng, DownloadSpec spec, std::chrono::seconds timeout) {
std::promise<Result<DownloadOutcome>> p;
auto f = p.get_future();
std::atomic<bool> fired{false};
DownloadCallbacks cbs;
cbs.on_finished = [&](Result<DownloadOutcome> r) {
if (!fired.exchange(true))
p.set_value(std::move(r));
};
auto h = eng.start(std::move(spec), std::move(cbs));
if (f.wait_for(timeout) != std::future_status::ready)
return Err{Error::timeout, "vdm_bench: download did not finish in time"};
return f.get();
}
std::string tmp_workdir() {
std::string p = "/tmp/vdm_bench_XXXXXX";
return ::mkdtemp(p.data()) ? p : "/tmp";
}
void report(const char *label, double v, const char *unit) {
std::fprintf(stderr, " %-22s %10.2f %s\n", label, v, unit);
}
// --- subcommands -------------------------------------------------------------------
int cmd_throughput(const Args &a) {
const std::uint64_t size = a.get_size("--size", "256M");
const long require_mbps = a.get_long("--require-mbps", 0);
const long max_cpu_pct = a.get_long("--max-cpu-pct", 0);
const auto segments = static_cast<std::uint32_t>(a.get_long("--segments", 0));
std::string dir = tmp_workdir();
if (!vdm::bench::make_sparse_file(dir + "/payload.bin", size)) {
std::fprintf(stderr, "vdm_bench: could not create %llu-byte payload in %s\n",
(unsigned long long)size, dir.c_str());
return 2;
}
vdm::bench::LocalServer srv(dir);
if (!srv.available()) {
std::fprintf(stderr,
"vdm_bench: no local server available (busybox missing?) -- skipping "
"throughput bench.\n");
return 0; // not a failure of the engine; nothing to measure against
}
Engine eng;
DownloadSpec spec;
spec.url = srv.url("/payload.bin");
spec.save_path = dir + "/out.bin";
if (segments)
spec.segments = segments;
const auto cpu0 = sample_rusage().cpu_s;
const auto t0 = std::chrono::steady_clock::now();
auto r = run_one(eng, std::move(spec), 300s);
const auto t1 = std::chrono::steady_clock::now();
const auto ru1 = sample_rusage();
if (!r.has_value()) {
std::fprintf(stderr, "vdm_bench throughput: download failed: %s\n",
r.error().to_string().c_str());
return 1;
}
const double wall_s = std::chrono::duration<double>(t1 - t0).count();
const double cpu_s = ru1.cpu_s - cpu0;
const double mbps = (r.value().bytes * 8.0 / 1'000'000.0) / wall_s;
const double cpu_pct = wall_s > 0 ? (cpu_s / wall_s) * 100.0 : 0.0;
std::fprintf(stderr, "throughput: %llu bytes in %.2fs\n", (unsigned long long)r.value().bytes,
wall_s);
report("throughput", mbps, "Mbps");
report("cpu", cpu_pct, "% of one core");
report("peak RSS", ru1.peak_rss_kb / 1024.0, "MiB");
int rc = 0;
if (require_mbps > 0 && mbps < require_mbps) {
std::fprintf(stderr, "FAIL: %.2f Mbps < required %ld Mbps\n", mbps, require_mbps);
rc = 1;
}
if (max_cpu_pct > 0 && cpu_pct > max_cpu_pct) {
std::fprintf(stderr, "FAIL: %.2f%% CPU > allowed %ld%%\n", cpu_pct, max_cpu_pct);
rc = 1;
}
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");
const long require_rss_kb = a.get_long("--require-rss-kb", 0);
// 0 => engine default (default_segments=8, docs/04 §8's actual DoD scenario). The
// ctest-registered smoke run overrides this down -- see CMakeLists.txt for why.
const auto segments_override = static_cast<std::uint32_t>(a.get_long("--segments", 0));
std::string dir = tmp_workdir();
// docs/04 §8's scenario is default_segments=8 per task, so up to tasks*8 concurrent
// segments (160, at the default --tasks 20) -- that's the point: the RSS ceiling is
// about *concurrent* segment buffers under a realistic multi-segment spread, not one
// connection per task. busybox httpd (LocalServer, used for the throughput bench)
// could not sustain that many concurrent connections reliably: reproducible hangs past
// a 120s per-task wait at --tasks 20 --task-size 4M, though not always at 2M -- some
// connections simply never got serviced. tools/testserver's threaded server (already
// exercised at real concurrency by the hostile-mode suite in engine_test.cpp) doesn't
// have that ceiling, so it's the transport here despite being the slower-per-request
// choice noted in support/local_server.hpp -- for this bench "slower" is actually
// wanted anyway (see below).
//
// Pace via testserver's own `throttled` mode rather than the engine's
// rate::RateLimiter: on loopback even 160 segments would otherwise race to completion
// before there's any concurrent overlap to measure RSS against, and pacing externally
// avoids a separate, real finding -- rate::RateLimiter::set_global_limit() under this
// much segment contention was observed to starve a couple of tasks for 120s+ instead of
// completing in the few seconds the rate implies (single shared TokenBucket, no
// fairness ordering across peek/commit races -- see docs/adr/0016). Throttle per
// connection, not per task: each segment is its own connection, so divide the
// per-task rate across default_segments to land total task duration in the same
// ballpark regardless of how many segments the engine actually opens.
const std::uint64_t assumed_segments = segments_override ? segments_override : 8;
const std::uint64_t per_conn_bps =
std::max<std::uint64_t>(1, task_size / (5 * assumed_segments)); // ~5s per task
vdm::bench::TestServerProc srv(per_conn_bps);
if (!srv.available()) {
std::fprintf(stderr,
"vdm_bench: tools/testserver unavailable -- skipping load bench.\n");
return 0;
}
Engine eng; // default Config: default_segments=8, max_active_segments=32 (docs/04 §8)
std::vector<std::promise<Result<DownloadOutcome>>> proms(tasks);
std::vector<std::future<Result<DownloadOutcome>>> futs;
std::vector<std::atomic<bool>> fired(tasks);
futs.reserve(tasks);
for (auto &p : proms) futs.push_back(p.get_future());
std::vector<DownloadHandle> handles;
handles.reserve(tasks);
const auto t0 = std::chrono::steady_clock::now();
for (int i = 0; i < tasks; ++i) {
DownloadSpec spec;
spec.url = srv.url("/throttled/file/" + std::to_string(task_size));
spec.save_path = dir + "/out" + std::to_string(i) + ".bin";
if (segments_override)
spec.segments = segments_override;
DownloadCallbacks cbs;
cbs.on_finished = [&proms, &fired, i](Result<DownloadOutcome> r) {
if (!fired[i].exchange(true))
proms[i].set_value(std::move(r));
};
handles.push_back(eng.start(std::move(spec), std::move(cbs)));
}
// TSan's per-access instrumentation overhead is heavy enough (observed: a couple of
// stragglers past 120s at --tasks 20 --task-size 2M, no TSan report -- just slow, not
// stuck) that a tight per-task budget here isn't testing the engine, it's testing the
// sanitizer. 300s per straggler is still bounded, just generous enough that "slow under
// instrumentation" and "actually wedged" stay distinguishable.
const auto task_timeout = std::chrono::seconds(a.get_long("--task-timeout-s", 300));
int failures = 0;
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;
}
auto r = futs[i].get();
if (!r.has_value()) {
std::fprintf(stderr, "task %d: %s\n", i, r.error().to_string().c_str());
++failures;
}
}
const auto t1 = std::chrono::steady_clock::now();
const auto ru = sample_rusage();
std::fprintf(stderr, "load: %d tasks, %d failed, %.2fs wall\n", tasks, failures,
std::chrono::duration<double>(t1 - t0).count());
report("peak RSS", ru.peak_rss_kb / 1024.0, "MiB");
if (failures > 0)
return 1;
if (require_rss_kb > 0 && ru.peak_rss_kb > require_rss_kb) {
std::fprintf(stderr, "FAIL: peak RSS %ld KiB > allowed %ld KiB\n", ru.peak_rss_kb,
require_rss_kb);
return 1;
}
return 0;
}
int cmd_heap_profile(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");
const auto segments_override = static_cast<std::uint32_t>(a.get_long("--segments", 0));
const int top_n = static_cast<int>(a.get_long("--top", 20));
std::string dir = tmp_workdir();
const std::uint64_t assumed_segments = segments_override ? segments_override : 8;
const std::uint64_t per_conn_bps =
std::max<std::uint64_t>(1, task_size / (5 * assumed_segments));
vdm::bench::TestServerProc srv(per_conn_bps);
if (!srv.available()) {
std::fprintf(stderr,
"vdm_bench: tools/testserver unavailable -- skipping heap-profile.\n");
return 0;
}
// On from before Engine's own construction, so its fixed cost (http worker threads,
// curl_multi handles, ...) shows up in the profile too, not just what the tasks add.
g_site_tracking.store(true, std::memory_order_relaxed);
Engine eng;
std::vector<DownloadHandle> handles;
handles.reserve(tasks);
for (int i = 0; i < tasks; ++i) {
DownloadSpec spec;
spec.url = srv.url("/throttled/file/" + std::to_string(task_size));
spec.save_path = dir + "/out" + std::to_string(i) + ".bin";
if (segments_override)
spec.segments = segments_override;
handles.push_back(eng.start(std::move(spec), DownloadCallbacks{}));
}
// Wait for peak RSS to stop growing (5 consecutive 200ms samples with no increase, or
// a generous cap) -- that's the moment whose live-allocation breakdown answers "where
// does the RSS actually go", the same question core/docs/m7-baseline.md's ~70 MB
// number raises against ADR 0012's ~45-50 MB estimate.
long last_rss = 0, stable_for = 0;
for (int i = 0; i < 300 && stable_for < 5; ++i) {
std::this_thread::sleep_for(200ms);
long rss = sample_rusage().peak_rss_kb;
stable_for = (rss <= last_rss) ? stable_for + 1 : 0;
last_rss = rss;
}
// Cross-check against the budget's own accounting, independent of the allocator-site
// snapshot below: this is what caught the real find here (core/docs/m7-baseline.md,
// docs/adr/0012) -- sum(effective_segments) tracked budget.active well past
// budget.total before the SegmentBudget fix these numbers are now reported alongside.
std::uint32_t total_effective_segments = 0;
for (auto &h : handles) total_effective_segments += h.progress().effective_segments;
auto eb = eng.segment_budget().budget();
std::fprintf(stderr,
"segment budget: %u live across tasks, engine reports total=%u active=%u "
"starved=%u\n",
total_effective_segments, eb.total, eb.active, eb.tasks_starved);
// Turn tracking off *before* touching g_site_stats: the snapshot below allocates its
// own vector storage through the very same overridden operator new. With tracking
// still on and g_track_mu already held for the snapshot, that allocation would recurse
// into track_alloc() and try to lock g_track_mu again -- self-deadlock on a
// non-recursive mutex (hit this exact hang while first wiring this subcommand up:
// g_in_tracker only guards the tracker's own bookkeeping containers, not other code
// that allocates while holding g_track_mu directly).
g_site_tracking.store(false, std::memory_order_relaxed);
std::vector<std::pair<void *, SiteStats>> sites;
{
std::lock_guard lk(g_track_mu);
sites.assign(g_site_stats.begin(), g_site_stats.end());
}
std::sort(sites.begin(), sites.end(), [](const auto &x, const auto &y) {
return x.second.live_bytes > y.second.live_bytes;
});
std::fprintf(stderr, "heap-profile: peak RSS %.2f MiB, %d tasks, %llu segments assumed\n",
last_rss / 1024.0, tasks, (unsigned long long)assumed_segments);
std::fprintf(stderr, "%-10s %-10s %s\n", "live KiB", "#allocs", "call site");
std::uint64_t total_live = 0;
for (auto &[site, st] : sites) total_live += st.live_bytes;
std::fprintf(stderr, "total tracked live: %.2f MiB across %zu sites\n\n",
total_live / 1024.0 / 1024.0, sites.size());
std::vector<void *> top_addrs;
for (int i = 0; i < top_n && i < static_cast<int>(sites.size()); ++i) top_addrs.push_back(sites[i].first);
auto labels = resolve_sites_via_addr2line(top_addrs);
for (std::size_t i = 0; i < top_addrs.size(); ++i) {
auto &[site, st] = sites[i];
std::fprintf(stderr, "%-10.1f %-10llu %s\n", st.live_bytes / 1024.0,
(unsigned long long)st.alloc_count, labels[i].c_str());
}
// Handles/Engine go out of scope here: ~Engine quiesces every task (waits for its
// workers to actually drain -- see the quiesce() fix) rather than waiting for the
// downloads to run to completion, which isn't the point of this subcommand.
return 0;
}
int cmd_alloc_check(const Args &a) {
const std::uint64_t size = a.get_size("--size", "128M");
const double window_s = a.get_double("--window-s", 2.0);
const double budget_per_s = a.get_double("--budget-per-s", 5.0);
std::string dir = tmp_workdir();
// busybox httpd over loopback is fast enough that even a --size in the hundreds of MB
// can complete in well under --window-s (measured: 64M in ~0.06s) -- there'd be no
// steady-state middle to sample. Pace the transfer with tools/testserver's `throttled`
// mode instead of the engine's own rate::RateLimiter: the limiter's precision
// pause/resume path allocates a timer node + std::function on every throttle event (see
// support/testserver_client.hpp), which would be exactly the kind of cost this bench
// exists to catch -- pacing external to the engine keeps the sample honest.
const std::uint64_t target_bps = std::max<std::uint64_t>(1, size / std::max(1.0, window_s * 6));
vdm::bench::TestServerProc srv(target_bps);
if (!srv.available()) {
std::fprintf(stderr,
"vdm_bench: tools/testserver unavailable -- skipping alloc-check.\n");
return 0;
}
Engine eng;
DownloadSpec spec;
spec.url = srv.url("/throttled/file/" + std::to_string(size));
spec.save_path = dir + "/out.bin";
// testserver throttles each connection independently, not the aggregate -- with the
// default multi-segment split the N parallel connections would finish in ~1/N of the
// time target_bps was sized for. Pin to one segment so the pacing math above holds.
spec.segments = 1;
std::promise<Result<DownloadOutcome>> p;
auto f = p.get_future();
std::atomic<bool> fired{false};
DownloadCallbacks cbs;
cbs.on_finished = [&](Result<DownloadOutcome> r) {
if (!fired.exchange(true))
p.set_value(std::move(r));
};
auto h = eng.start(std::move(spec), std::move(cbs));
// Ramp-up: let the probe, segment split, and first buffer fills happen (all legitimate
// allocation) before we start counting. Bail out if it finishes (or fails) before we
// ever get a steady-state window to sample -- too small a --size for --window-s.
for (int i = 0; i < 500 && h.progress().downloaded == 0; ++i) {
if (f.wait_for(0s) == std::future_status::ready) {
std::fprintf(stderr,
"vdm_bench: download finished during ramp-up -- use a bigger "
"--size or a smaller --window-s\n");
return 2;
}
std::this_thread::sleep_for(10ms);
}
const std::uint64_t before = g_alloc_count.load(std::memory_order_relaxed);
std::this_thread::sleep_for(std::chrono::duration<double>(window_s));
const std::uint64_t after = g_alloc_count.load(std::memory_order_relaxed);
if (f.wait_for(0s) == std::future_status::ready) {
std::fprintf(stderr,
"vdm_bench: download finished during the sampling window -- use a "
"bigger --size or a smaller --window-s\n");
return 2;
}
const std::uint64_t delta = after - before;
const double budget = budget_per_s * window_s + 5; // +5: fixed slack for one-off events
std::fprintf(stderr, "alloc-check: %llu allocations in %.2fs (budget %.0f)\n",
(unsigned long long)delta, window_s, budget);
if (static_cast<double>(delta) > budget) {
std::fprintf(stderr,
"FAIL: allocation count scales with the transfer, not just periodic "
"bookkeeping -- something on the write path is allocating.\n");
return 1;
}
return 0;
}
void usage() {
std::fprintf(stderr,
"usage: vdm_bench <throughput|load|heap-profile|alloc-check> [options]\n"
" throughput [--size 5G] [--segments N] [--require-mbps N] "
"[--max-cpu-pct N]\n"
" load [--tasks 20] [--task-size 4M] [--segments N] "
"[--require-rss-kb N] [--task-timeout-s 300]\n"
" heap-profile [--tasks 20] [--task-size 4M] [--segments N] [--top 20]\n"
" alloc-check [--size 128M] [--window-s 2] [--budget-per-s 5]\n");
}
} // namespace
int main(int argc, char **argv) {
if (argc < 2) {
usage();
return 2;
}
std::string cmd = argv[1];
Args args(argc, argv, 2);
if (cmd == "throughput")
return cmd_throughput(args);
if (cmd == "load")
return cmd_load(args);
if (cmd == "heap-profile")
return cmd_heap_profile(args);
if (cmd == "alloc-check")
return cmd_alloc_check(args);
usage();
return 2;
}