dtourolle
88c42573a5
Each of these was recorded as done and was done in one place out of two. AR-011 -- the TransNetV2 dedup window. The derived window (dedup_window_sec, median observed interval halved) reached scenes.json and nothing else. SceneBoundaries, the path that actually feeds is_scene_boundary to the tracker, kept the literal 0.04 s under a comment claiming it "matches the dedup scenes.json applies, so the two views agree". They did not agree. 0.04 is one frame at 25 fps and wider than a frame at 30, so two cuts on consecutive frames merged into one and the loss was invisible: the pipeline simply saw fewer boundaries. The detector now supplies the window it derived. AR-019 -- ownership. The register says ownership "comes from the registry, not a second local tally". Both existed: promotion fired on a local accepted-frame count and fell back to a local per-actor plurality when the registry had not yet claimed the track. That fallback was reachable in the live pipeline, not just in tests -- three accepted frames arrive well before a posterior crosses the ownership threshold -- so in practice the plurality usually decided, and it could not see the AR-025 correlation discounting it was meant to defer to. The tally is gone; promotion now requires the registry's verdict, with the accepted -frame count demoted to an explicit evidence floor. AR-017 -- the route. DeadTrack carried belief but no route, and the sink wrote the literal string "live", so a field the schema publishes could not distinguish anything. AR-017's own verification asks for "deferred and pooled routes distinguishable". Route is now an enum on the claim. Only `live` occurs today; `deferred` exists so AR-020's pass has somewhere to write instead of a serialisation change to make. Also: TrackGallery::forget had no callers, under a comment asserting the matcher called it "on a cut or track disappearance". The cut half was true by another route; the disappearance half was not, so a track that died quietly kept its diversity buffer until the next cut cleared everything. Replaced with prune_dead against the registry's own liveness, the same shape as the tracker's prune_boxes -- a second opinion about which tracks exist is a second thing that can be wrong. Removes dead logistic/logit helpers and fixes five TRACES tags that used a comma where a pipe separates requirement types, which the gate had been reporting as diagnostics. TRACES: AR-011, AR-017, AR-019 | IR-002 | SR-002, SR-005
591 lines
28 KiB
C++
591 lines
28 KiB
C++
#pragma once
|
||
/// TRACES: VR-015 | PR-004
|
||
///
|
||
/// Pipeline throughput benchmark — how much of a run is spent in each node.
|
||
///
|
||
/// The KPN network already counts most of what an optimiser needs, and
|
||
/// `print_diagnostics()` throws nearly all of it away: it prints frames and
|
||
/// `ema` per node, and passes `elapsed_s = 0`, which zeroes throughput. Two
|
||
/// things had to change before "time per node" could be answered honestly.
|
||
///
|
||
/// **`ema` is not a total.** It is an exponentially weighted average, so
|
||
/// `frames * ema` tracks the end of the run rather than the whole of it. On a
|
||
/// film that is a real difference — a detector costs one thing in a crowd scene
|
||
/// and another over a landscape. `NodeStats::total_exec_us` (added alongside
|
||
/// this) is the true sum.
|
||
///
|
||
/// **Wall time inside a node is not all work.** `PoolObjectNode::fire_once`
|
||
/// times the functor *and* `push_outputs`, and `push_outputs` parks on a full
|
||
/// downstream channel (AR-004). A node that is merely backpressured therefore
|
||
/// bills the time it spent waiting to whoever is ahead of it: SuperHero's
|
||
/// `frame_source` reported 141.9 ms/frame against a decoder logging 12-18 ms.
|
||
/// Optimising against that number means optimising the fastest node in the
|
||
/// graph.
|
||
///
|
||
/// So each node is reported three ways, and the three together are what
|
||
/// identify a cost:
|
||
///
|
||
/// - `exec_ms` — cumulative wall time in the node, work *and* parked pushes
|
||
/// - `cpu_ms` — thread CPU time (CLOCK_THREAD_CPUTIME_ID). Backpressure
|
||
/// cannot inflate it, because a parked node holds no thread.
|
||
/// - `pressure` — mean fill of its input channels minus that of its outputs
|
||
///
|
||
/// **`cpu_ms` cannot tell real work from a spinning GPU wait.** CUDA's default
|
||
/// sync policy (`cudaDeviceScheduleAuto`) spin-waits before yielding, so
|
||
/// `cudaStreamSynchronize` burns the calling thread's CPU while the GPU works.
|
||
/// A node that is purely GPU-bound can therefore report a high `cpu_ms` and read
|
||
/// as CPU-bound. `cudaSetDeviceFlags(cudaDeviceScheduleBlockingSync)` settles it
|
||
/// in one line: if a node's `cpu_ms` collapses under blocking sync, that CPU was
|
||
/// spin, not work.
|
||
///
|
||
/// **`cpu_ms` counts one thread only.** `CLOCK_THREAD_CPUTIME_ID` is per-thread,
|
||
/// and OpenCV here is built against TBB, so any node whose functor goes through
|
||
/// `cv::parallel_for_` (histogram compare, `warpAffine`, colour conversion) has
|
||
/// that work executed on TBB's arena — 19 workers on a 20-core box — and billed
|
||
/// to those threads rather than to the node. Such a node reads *cheaper* than it
|
||
/// is, and the difference shows up in `stall/f` instead, indistinguishable from a
|
||
/// GPU wait. `exec_ms` does capture it, since the functor does not return until
|
||
/// the parallel region joins: a node where `exec/f` greatly exceeds `cpu/f`
|
||
/// while its output channel is empty is fanning out, not waiting.
|
||
///
|
||
/// Work piles up *in front of* a bottleneck and starves everything *after* it,
|
||
/// so `pressure` is maximal at the node setting the pace. `cpu_ms` then says
|
||
/// which repair applies: high pressure with a saturated thread is CPU-bound and
|
||
/// the work must get cheaper, while high pressure with an idle thread is
|
||
/// waiting on a device, where batch size and engine precision are the knobs.
|
||
///
|
||
/// Occupancy has to be sampled during the run. `current_fill` is instantaneous
|
||
/// and every channel has drained by the time the network stops, so a single
|
||
/// read at the end reports an idle pipeline however congested it was.
|
||
///
|
||
/// Nothing here is specific to this pipeline's topology: the node graph is
|
||
/// recovered from KPN's channel names, so it keeps working when the graph
|
||
/// changes.
|
||
|
||
#include <kpn/diagnostics.hpp>
|
||
|
||
#include <nlohmann/json.hpp>
|
||
|
||
#include <algorithm>
|
||
#include <atomic>
|
||
#include <chrono>
|
||
#include <cstdint>
|
||
#include <functional>
|
||
#include <iomanip>
|
||
#include <map>
|
||
#include <ostream>
|
||
#include <sstream>
|
||
#include <string>
|
||
#include <thread>
|
||
#include <vector>
|
||
|
||
namespace sae::bench {
|
||
|
||
// ── Edge naming ──────────────────────────────────────────────────────────────
|
||
|
||
/// TRACES: VR-015 | PR-004
|
||
/// KPN names a channel "<src>:<idx> → <dst>:<idx>" (static_network.hpp).
|
||
/// Recovering the two node names from it is what keeps attribution
|
||
/// topology-agnostic: the graph is read back out of the channel names rather
|
||
/// than hard-coded here, so a new node or a re-wired branch needs no change.
|
||
/// Leaves both outputs untouched if the name does not carry an arrow.
|
||
inline void split_edge_name(const std::string& name,
|
||
std::string& producer, std::string& consumer) {
|
||
static const std::string kArrow = " \xe2\x86\x92 "; // " → "
|
||
const auto arrow = name.find(kArrow);
|
||
if (arrow == std::string::npos) return;
|
||
auto strip_port = [](std::string s) {
|
||
const auto colon = s.rfind(':');
|
||
return colon == std::string::npos ? s : s.substr(0, colon);
|
||
};
|
||
producer = strip_port(name.substr(0, arrow));
|
||
consumer = strip_port(name.substr(arrow + kArrow.size()));
|
||
}
|
||
|
||
// ── Channel occupancy, time-averaged ─────────────────────────────────────────
|
||
|
||
/// TRACES: VR-015 | PR-004
|
||
/// One channel's fill level integrated over the run. `peak_fill` is already
|
||
/// cumulative in `ChannelStats`, but a peak cannot distinguish "full once" from
|
||
/// "full throughout", and those are opposite diagnoses. A mean can.
|
||
struct ChannelOccupancy {
|
||
std::string name; // "src:0 → dst:0", as KPN names it
|
||
std::string producer; // node name left of the arrow
|
||
std::string consumer; // node name right of the arrow
|
||
std::size_t capacity{0};
|
||
std::uint64_t samples{0};
|
||
double fill_sum{0.0};
|
||
std::uint64_t samples_full{0};
|
||
std::uint64_t samples_empty{0};
|
||
|
||
// Final-snapshot totals (monotonic counters, so the last read is the total).
|
||
std::size_t peak_fill{0};
|
||
std::uint64_t pushes{0};
|
||
std::uint64_t pops{0};
|
||
std::uint64_t drops{0};
|
||
std::uint64_t overflows{0};
|
||
std::uint64_t bytes_pushed{0};
|
||
|
||
double mean_fill() const { return samples ? fill_sum / static_cast<double>(samples) : 0.0; }
|
||
double mean_fill_pct() const { return capacity ? 100.0 * mean_fill() / static_cast<double>(capacity) : 0.0; }
|
||
double peak_pct() const { return capacity ? 100.0 * static_cast<double>(peak_fill) / static_cast<double>(capacity) : 0.0; }
|
||
double full_pct() const { return samples ? 100.0 * static_cast<double>(samples_full) / static_cast<double>(samples) : 0.0; }
|
||
double empty_pct() const { return samples ? 100.0 * static_cast<double>(samples_empty) / static_cast<double>(samples) : 0.0; }
|
||
double bandwidth_mbs(double wall_s) const {
|
||
return wall_s > 0.0 ? static_cast<double>(bytes_pushed) / wall_s / 1e6 : 0.0;
|
||
}
|
||
};
|
||
|
||
// ── Per-node attributed cost ─────────────────────────────────────────────────
|
||
|
||
/// TRACES: VR-015 | PR-004
|
||
struct NodeCost {
|
||
std::string name;
|
||
std::uint64_t frames{0};
|
||
|
||
// Cumulative wall time in the node — the answer to "where did the run go",
|
||
// but only for a node that is not backpressured; it includes parked pushes.
|
||
double exec_ms{0.0};
|
||
double exec_ms_per_frame{0.0}; // true mean, not the EMA
|
||
double exec_share{0.0}; // exec_ms / wall_ms, 0..1
|
||
double ema_exec_ms{0.0}; // KPN's EMA, kept for continuity with the old report
|
||
double max_exec_ms{0.0};
|
||
|
||
// Thread CPU time: excludes sleeping, parking and waiting on a device, so it
|
||
// is the one number backpressure cannot inflate.
|
||
double cpu_ms{0.0};
|
||
double cpu_ms_per_frame{0.0};
|
||
double cpu_share{0.0}; // cpu_ms / wall_ms — thread saturation, 0..1
|
||
double cpu_pct_of_pipeline{0.0}; // this node's share of all nodes' CPU time
|
||
|
||
// Per frame, time inside the node not spent on its own CPU: parked on a
|
||
// full output channel, or waiting on the GPU. `pressure` separates those —
|
||
// a backpressured node has a full output, a device-bound one does not.
|
||
double stall_ms_per_frame{0.0};
|
||
|
||
// Queue occupancy either side of the node, in percent of capacity.
|
||
double in_fill_pct{0.0};
|
||
double out_fill_pct{0.0};
|
||
double pressure{0.0}; // in − out; maximal at the pacing node
|
||
bool has_input{false};
|
||
bool has_output{false};
|
||
|
||
double queue_wait_ms{0.0};
|
||
bool is_bottleneck{false};
|
||
|
||
/// TRACES: VR-015 | AR-004 | PR-004
|
||
/// Live scheduling state, so a wedged run says *why* it is wedged rather
|
||
/// than only that it is. With `queued=0, wake=1` a wake was recorded and
|
||
/// never consumed; with `queued=0, wake=0` and a full input, no wake was
|
||
/// ever generated. Those are different bugs in different files.
|
||
bool queued{false};
|
||
bool wake_pending{false};
|
||
};
|
||
|
||
/// TRACES: VR-015 | PR-004
|
||
/// Attribute cost to nodes from a KPN node snapshot plus sampled channel
|
||
/// occupancy. Pure — no clocks, no threads, no network — so the ranking is
|
||
/// unit-testable on CI hardware that can never run the pipeline itself.
|
||
///
|
||
/// A node with several inputs takes the **minimum** input fill: it can only run
|
||
/// once every input has data, so the emptiest one gates it, and a full sibling
|
||
/// channel means that channel's producer is blocked rather than this node being
|
||
/// slow. A node with several outputs takes the **maximum** output fill, since
|
||
/// parking on any one branch stops the node.
|
||
///
|
||
/// Terminals are the infinite-reservoir limit of the same rule: a source has
|
||
/// unlimited work available (input treated as 100% full) and a sink unlimited
|
||
/// drain (output treated as empty), so both stay rankable against the interior
|
||
/// nodes instead of dropping out of the comparison.
|
||
inline std::vector<NodeCost> attribute_cost(
|
||
const std::vector<kpn::NodeSnapshot>& nodes,
|
||
const std::vector<ChannelOccupancy>& channels,
|
||
double wall_sec)
|
||
{
|
||
const double wall_ms = wall_sec * 1000.0;
|
||
|
||
double cpu_total = 0.0;
|
||
for (const auto& n : nodes) cpu_total += n.total_cpu_ms;
|
||
|
||
std::vector<NodeCost> out;
|
||
out.reserve(nodes.size());
|
||
|
||
for (const auto& n : nodes) {
|
||
NodeCost c;
|
||
c.name = n.name;
|
||
c.frames = n.frames_processed;
|
||
c.exec_ms = n.total_exec_ms;
|
||
c.ema_exec_ms = n.ema_exec_ms;
|
||
c.max_exec_ms = n.max_exec_ms;
|
||
c.cpu_ms = n.total_cpu_ms;
|
||
c.queue_wait_ms = n.queue_wait_ms;
|
||
c.queued = n.queued;
|
||
c.wake_pending = n.wake_pending;
|
||
|
||
c.exec_ms_per_frame = c.frames ? c.exec_ms / static_cast<double>(c.frames) : 0.0;
|
||
c.cpu_ms_per_frame = c.frames ? c.cpu_ms / static_cast<double>(c.frames) : 0.0;
|
||
c.exec_share = wall_ms > 0.0 ? c.exec_ms / wall_ms : 0.0;
|
||
c.cpu_share = wall_ms > 0.0 ? c.cpu_ms / wall_ms : 0.0;
|
||
c.cpu_pct_of_pipeline = cpu_total > 0.0 ? 100.0 * c.cpu_ms / cpu_total : 0.0;
|
||
c.stall_ms_per_frame = c.exec_ms_per_frame - c.cpu_ms_per_frame;
|
||
if (c.stall_ms_per_frame < 0.0) c.stall_ms_per_frame = 0.0;
|
||
|
||
double in_min = 0.0; bool have_in = false;
|
||
double out_max = 0.0; bool have_out = false;
|
||
for (const auto& ch : channels) {
|
||
if (ch.consumer == n.name) {
|
||
const double f = ch.mean_fill_pct();
|
||
if (!have_in || f < in_min) in_min = f;
|
||
have_in = true;
|
||
}
|
||
if (ch.producer == n.name) {
|
||
const double f = ch.mean_fill_pct();
|
||
if (!have_out || f > out_max) out_max = f;
|
||
have_out = true;
|
||
}
|
||
}
|
||
c.has_input = have_in;
|
||
c.has_output = have_out;
|
||
c.in_fill_pct = have_in ? in_min : 100.0; // source: always has work
|
||
c.out_fill_pct = have_out ? out_max : 0.0; // sink: never blocks
|
||
c.pressure = c.in_fill_pct - c.out_fill_pct;
|
||
out.push_back(std::move(c));
|
||
}
|
||
|
||
// Rank, but only among nodes that actually ran: a node with zero frames has
|
||
// no cost to attribute and its neighbouring channels never moved.
|
||
auto best = out.end();
|
||
for (auto it = out.begin(); it != out.end(); ++it) {
|
||
if (it->frames == 0) continue;
|
||
if (best == out.end() || it->pressure > best->pressure) best = it;
|
||
}
|
||
if (best != out.end()) best->is_bottleneck = true;
|
||
|
||
return out;
|
||
}
|
||
|
||
/// TRACES: VR-015 | PR-004
|
||
/// One line of plain English about the winning node, since the point of the
|
||
/// report is to say what to change next. A saturated thread means the node's
|
||
/// own work is the limit; an idle thread under pressure means it is waiting on
|
||
/// a device, and those are different repairs.
|
||
inline std::string verdict(const std::vector<NodeCost>& costs) {
|
||
for (const auto& c : costs) {
|
||
if (!c.is_bottleneck) continue;
|
||
std::ostringstream os;
|
||
os << std::fixed << c.name << " sets the pace: ";
|
||
|
||
// A source's 100% input is the infinite-reservoir convention, not a
|
||
// measured queue — saying "work is backed up in front of it" would be
|
||
// asserting something no counter observed.
|
||
if (!c.has_input)
|
||
os << "nothing downstream is waiting on it (output "
|
||
<< std::setprecision(1) << c.out_fill_pct
|
||
<< "% full), so the pipeline is running as fast as this node can feed it. ";
|
||
else if (!c.has_output)
|
||
os << std::setprecision(1) << c.in_fill_pct
|
||
<< "% full input and nothing to block on, so it is the drain. ";
|
||
else
|
||
os << std::setprecision(1) << c.in_fill_pct << "% full input, "
|
||
<< c.out_fill_pct << "% full output. ";
|
||
|
||
os << std::setprecision(2) << c.cpu_ms_per_frame << " ms/frame on CPU. ";
|
||
|
||
if (c.cpu_share >= 0.85)
|
||
os << "CPU-bound — its thread is busy " << std::setprecision(0)
|
||
<< (100.0 * c.cpu_share) << "% of the run, so the work itself has to get"
|
||
" cheaper or be split across more threads.";
|
||
else if (c.cpu_share <= 0.35 && c.stall_ms_per_frame > c.cpu_ms_per_frame)
|
||
os << "Device-bound — its thread is busy only " << std::setprecision(0)
|
||
<< (100.0 * c.cpu_share) << "% of the run and it spends "
|
||
<< std::setprecision(2) << c.stall_ms_per_frame
|
||
<< " ms/frame off-CPU, so it is waiting on the GPU or the disk: batch size,"
|
||
" engine precision and the decode path are the knobs, not the C++.";
|
||
else
|
||
os << "Mixed — thread busy " << std::setprecision(0) << (100.0 * c.cpu_share)
|
||
<< "% of the run, " << std::setprecision(2) << c.stall_ms_per_frame
|
||
<< " ms/frame off-CPU.";
|
||
return os.str();
|
||
}
|
||
return "no node processed a frame — nothing to attribute";
|
||
}
|
||
|
||
// ── Recorder ─────────────────────────────────────────────────────────────────
|
||
|
||
/// TRACES: VR-015 | PR-004
|
||
/// Samples the live network on a timer and emits the report at the end.
|
||
///
|
||
/// The sampler only reads relaxed atomics, so it does not perturb what it
|
||
/// measures — which matters, since this exists to be trusted as a timing
|
||
/// measurement.
|
||
class BenchmarkRecorder {
|
||
public:
|
||
using Sampler = std::function<kpn::NetworkSnapshot()>;
|
||
|
||
explicit BenchmarkRecorder(int sample_interval_ms = 100)
|
||
: interval_(std::chrono::milliseconds(sample_interval_ms)) {}
|
||
|
||
~BenchmarkRecorder() { stop(); }
|
||
|
||
void start(Sampler sampler) {
|
||
sampler_ = std::move(sampler);
|
||
running_.store(true, std::memory_order_release);
|
||
thread_ = std::thread([this] {
|
||
while (running_.load(std::memory_order_acquire)) {
|
||
accumulate(sampler_());
|
||
std::this_thread::sleep_for(interval_);
|
||
}
|
||
});
|
||
}
|
||
|
||
/// Stops sampling and latches the final counter values. Call while the
|
||
/// network object is still alive: the monotonic counters stay valid after
|
||
/// `net.stop()`, but they die with the object.
|
||
void stop() {
|
||
if (!running_.exchange(false, std::memory_order_acq_rel)) return;
|
||
if (thread_.joinable()) thread_.join();
|
||
if (sampler_) {
|
||
final_ = sampler_();
|
||
// Occupancy is deliberately NOT accumulated from this last read:
|
||
// the pipeline has drained by now, and folding an idle sample into
|
||
// the mean biases every channel toward "never congested".
|
||
for (const auto& ch : final_.channels) {
|
||
auto& occ = occupancy_[ch.name];
|
||
if (occ.name.empty()) { // a channel that never moved
|
||
occ.name = ch.name;
|
||
occ.capacity = ch.capacity;
|
||
split_edge_name(ch.name, occ.producer, occ.consumer);
|
||
}
|
||
occ.peak_fill = ch.peak_fill;
|
||
occ.pushes = ch.pushes;
|
||
occ.pops = ch.pops;
|
||
occ.drops = ch.drops;
|
||
occ.overflows = ch.overflows;
|
||
occ.bytes_pushed = ch.bytes_pushed;
|
||
}
|
||
}
|
||
stopped_ = true;
|
||
}
|
||
|
||
bool has_data() const { return stopped_ && !final_.nodes.empty(); }
|
||
double wall_sec() const { return final_.elapsed_s; }
|
||
|
||
std::vector<ChannelOccupancy> channels() const {
|
||
std::vector<ChannelOccupancy> v;
|
||
v.reserve(occupancy_.size());
|
||
for (const auto& [_, occ] : occupancy_) v.push_back(occ);
|
||
return v;
|
||
}
|
||
|
||
std::vector<NodeCost> costs() const {
|
||
return attribute_cost(final_.nodes, channels(), final_.elapsed_s);
|
||
}
|
||
|
||
/// TRACES: VR-015 | PR-004
|
||
/// Machine-readable report, for sweeping configurations and diffing runs.
|
||
/// `film_sec` is the last timestamp the pipeline reached, so
|
||
/// `realtime_factor` answers what the optimiser is really asking: seconds
|
||
/// of film per second of wall clock. It is 0 for a topology with no result
|
||
/// sink (the dump-only path), and the field is then omitted rather than
|
||
/// reported as zero throughput.
|
||
nlohmann::json to_json(const nlohmann::json& run_config, double film_sec) const {
|
||
using nlohmann::json;
|
||
const double wall = final_.elapsed_s;
|
||
const auto chans = channels();
|
||
const auto cost = attribute_cost(final_.nodes, chans, wall);
|
||
|
||
json j;
|
||
j["schema_version"] = 1;
|
||
j["config"] = run_config;
|
||
|
||
json summary;
|
||
summary["wall_sec"] = wall;
|
||
summary["sample_count"] = sample_count_;
|
||
summary["sample_interval_ms"] = interval_.count();
|
||
if (film_sec > 0.0) {
|
||
summary["film_sec"] = film_sec;
|
||
summary["realtime_factor"] = wall > 0.0 ? film_sec / wall : 0.0;
|
||
}
|
||
for (const auto& c : cost)
|
||
if (c.is_bottleneck) { summary["bottleneck"] = c.name; break; }
|
||
summary["verdict"] = verdict(cost);
|
||
j["summary"] = summary;
|
||
|
||
json jnodes = json::array();
|
||
for (const auto& c : cost) {
|
||
jnodes.push_back({
|
||
{"name", c.name},
|
||
{"frames", c.frames},
|
||
{"fps", wall > 0.0 ? c.frames / wall : 0.0},
|
||
{"exec_ms", c.exec_ms},
|
||
{"exec_ms_per_frame", c.exec_ms_per_frame},
|
||
{"exec_share", c.exec_share},
|
||
{"ema_exec_ms", c.ema_exec_ms},
|
||
{"max_exec_ms", c.max_exec_ms},
|
||
{"cpu_ms", c.cpu_ms},
|
||
{"cpu_ms_per_frame", c.cpu_ms_per_frame},
|
||
{"cpu_share", c.cpu_share},
|
||
{"cpu_pct_of_pipeline", c.cpu_pct_of_pipeline},
|
||
{"stall_ms_per_frame", c.stall_ms_per_frame},
|
||
{"queue_wait_ms", c.queue_wait_ms},
|
||
{"in_fill_pct", c.in_fill_pct},
|
||
{"out_fill_pct", c.out_fill_pct},
|
||
{"pressure", c.pressure},
|
||
{"is_bottleneck", c.is_bottleneck},
|
||
{"queued", c.queued},
|
||
{"wake_pending", c.wake_pending},
|
||
});
|
||
}
|
||
j["nodes"] = std::move(jnodes);
|
||
|
||
json jch = json::array();
|
||
for (const auto& ch : chans) {
|
||
jch.push_back({
|
||
{"name", ch.name},
|
||
{"producer", ch.producer},
|
||
{"consumer", ch.consumer},
|
||
{"capacity", ch.capacity},
|
||
{"mean_fill", ch.mean_fill()},
|
||
{"mean_fill_pct", ch.mean_fill_pct()},
|
||
{"peak_fill", ch.peak_fill},
|
||
{"peak_pct", ch.peak_pct()},
|
||
{"full_pct", ch.full_pct()},
|
||
{"empty_pct", ch.empty_pct()},
|
||
{"pushes", ch.pushes},
|
||
{"pops", ch.pops},
|
||
{"drops", ch.drops},
|
||
{"overflows", ch.overflows},
|
||
{"mb_per_sec", ch.bandwidth_mbs(wall)},
|
||
});
|
||
}
|
||
j["channels"] = std::move(jch);
|
||
return j;
|
||
}
|
||
|
||
/// TRACES: VR-015 | PR-004
|
||
/// Human-readable form of the same data, so a run is legible without
|
||
/// opening the JSON.
|
||
void print(std::ostream& os, double film_sec) const {
|
||
print_impl(os, final_, film_sec);
|
||
}
|
||
|
||
/// TRACES: VR-015 | AR-004 | PR-004
|
||
/// Dump the report from a LIVE snapshot, mid-run, without stopping anything.
|
||
///
|
||
/// A report that only exists at shutdown is no use against the failure this
|
||
/// pipeline actually has: a wedged run never reaches shutdown, so the one
|
||
/// moment the numbers matter most is the one moment they were unavailable.
|
||
/// Channel occupancy names the stalled node directly — it is the one whose
|
||
/// input is full and whose output is empty — which is otherwise a debug-build
|
||
/// and a gdb session away.
|
||
///
|
||
/// Safe to call from the wait loop while the pipeline is running or hung: it
|
||
/// takes the same lock-free snapshot the sampler does.
|
||
void dump_live(std::ostream& os, double film_sec) const {
|
||
if (!sampler_) { os << "[benchmark] no sampler — run with --benchmark\n"; return; }
|
||
print_impl(os, sampler_(), film_sec);
|
||
}
|
||
|
||
private:
|
||
void print_impl(std::ostream& os, const kpn::NetworkSnapshot& snap,
|
||
double film_sec) const {
|
||
const double wall = snap.elapsed_s;
|
||
const auto chans = channels();
|
||
const auto cost = attribute_cost(snap.nodes, chans, wall);
|
||
|
||
os << "\n┌─ Pipeline benchmark (VR-015) ──────────────────────────────────────────────\n";
|
||
os << "│ wall " << std::fixed << std::setprecision(1) << wall << "s";
|
||
if (film_sec > 0.0)
|
||
os << " film " << film_sec << "s realtime x" << std::setprecision(2)
|
||
<< (wall > 0.0 ? film_sec / wall : 0.0);
|
||
os << " samples " << sample_count_ << "\n│\n";
|
||
|
||
os << "│ node frames cpu_s cpu%run cpu%tot cpu/f"
|
||
" exec/f stall/f in% out% press q/w\n";
|
||
for (const auto& c : cost) {
|
||
os << "│ " << (c.is_bottleneck ? "▶ " : " ") << std::left << std::setw(16)
|
||
<< c.name << std::right
|
||
<< std::setw(7) << c.frames
|
||
<< std::setw(10) << std::setprecision(1) << (c.cpu_ms / 1000.0)
|
||
<< std::setw(9) << std::setprecision(0) << (100.0 * c.cpu_share)
|
||
<< std::setw(9) << std::setprecision(0) << c.cpu_pct_of_pipeline
|
||
<< std::setw(8) << std::setprecision(2) << c.cpu_ms_per_frame
|
||
<< std::setw(8) << std::setprecision(2) << c.exec_ms_per_frame
|
||
<< std::setw(9) << std::setprecision(2) << c.stall_ms_per_frame
|
||
<< std::setw(7) << std::setprecision(0) << c.in_fill_pct
|
||
<< std::setw(7) << std::setprecision(0) << c.out_fill_pct
|
||
<< std::setw(8) << std::setprecision(1) << c.pressure
|
||
<< " " << int(c.queued) << "/" << int(c.wake_pending)
|
||
<< "\n";
|
||
}
|
||
|
||
/// TRACES: VR-015 | AR-004 | PR-004
|
||
// Fires only when the scheduling state is actually wrong, so a healthy
|
||
// run stays quiet and a wedged one names the fault — instead of leaving
|
||
// it to be reconstructed under a debugger that suppresses the bug.
|
||
for (const auto& c : cost) {
|
||
if (c.queued || !c.has_input) continue;
|
||
if (c.wake_pending)
|
||
os << "│ !! " << c.name << " idle with a wake outstanding"
|
||
" (queued=0 wake=1): the wake was recorded and never"
|
||
" consumed — submit/release handshake.\n";
|
||
else if (c.in_fill_pct > 50.0)
|
||
os << "│ !! " << c.name << " idle with a "
|
||
<< std::setprecision(0) << c.in_fill_pct
|
||
<< "% full input and no wake pending: the wake was never"
|
||
" generated — channel edge detection.\n";
|
||
}
|
||
|
||
os << "│\n│ channel cap mean% peak% full%"
|
||
" empty% MB/s\n";
|
||
for (const auto& ch : chans) {
|
||
os << "│ " << std::left << std::setw(36) << ch.name << std::right
|
||
<< std::setw(5) << ch.capacity
|
||
<< std::setw(7) << std::setprecision(1) << ch.mean_fill_pct()
|
||
<< std::setw(7) << ch.peak_pct()
|
||
<< std::setw(7) << ch.full_pct()
|
||
<< std::setw(7) << ch.empty_pct()
|
||
<< std::setw(9) << std::setprecision(1) << ch.bandwidth_mbs(wall)
|
||
<< "\n";
|
||
}
|
||
os << "│\n│ " << verdict(cost) << "\n";
|
||
os << "└────────────────────────────────────────────────────────────────────────────\n";
|
||
os << " cpu_s / cpu%tot is where the run's compute actually went. exec/f is wall\n"
|
||
" time in the node INCLUDING time parked on a full output channel, so it\n"
|
||
" overstates a backpressured node — compare it against cpu/f, which cannot\n"
|
||
" be inflated that way. press = input fill − output fill, and locates the\n"
|
||
" node that work is queueing up in front of.\n"
|
||
" cpu_s counts THIS node's thread only: work OpenCV fans out via TBB is\n"
|
||
" billed to the TBB arena, so a node using cv::parallel_for_ reads cheaper\n"
|
||
" than it is and the difference surfaces in stall/f.\n";
|
||
}
|
||
|
||
private:
|
||
void accumulate(const kpn::NetworkSnapshot& snap) {
|
||
++sample_count_;
|
||
for (const auto& ch : snap.channels) {
|
||
auto& occ = occupancy_[ch.name];
|
||
if (occ.name.empty()) {
|
||
occ.name = ch.name;
|
||
occ.capacity = ch.capacity;
|
||
split_edge_name(ch.name, occ.producer, occ.consumer);
|
||
}
|
||
occ.fill_sum += static_cast<double>(ch.current_fill);
|
||
++occ.samples;
|
||
if (ch.capacity && ch.current_fill >= ch.capacity) ++occ.samples_full;
|
||
if (ch.current_fill == 0) ++occ.samples_empty;
|
||
}
|
||
}
|
||
|
||
std::chrono::milliseconds interval_;
|
||
Sampler sampler_;
|
||
std::thread thread_;
|
||
std::atomic<bool> running_{false};
|
||
bool stopped_{false};
|
||
std::uint64_t sample_count_{0};
|
||
std::map<std::string, ChannelOccupancy> occupancy_;
|
||
kpn::NetworkSnapshot final_{};
|
||
};
|
||
|
||
} // namespace sae::bench
|