feat(benchmark): per-node cost and bottleneck attribution for a run
--benchmark <path> reports cumulative CPU and wall time per node and names the node pacing the run. The pacing node is located from sampled channel occupancy, not from time-in-node: backpressure inflates time-in-node for everything downstream of the real bottleneck, so the obvious measure names the victim rather than the cause. Sampling starts with the network and stops before it is destroyed. Channel fill is instantaneous and everything has drained by shutdown, so a single read at the end reports an idle pipeline however congested it was. kill -USR1 dumps the table from a running or wedged process. Channel occupancy identifies a stalled node -- full input, empty output -- without a debug build or a debugger, which is the difference between diagnosing the AR-004 hang in seconds and reproducing it under gdb. Two knobs this exposes for measurement rather than sets: SAE_CV_THREADS, because OpenCV's TBB arena and KPN's thread-per-node are two schedulers unaware of each other on the same cores; and SAE_CUDA_BLOCKING_SYNC, because the default spin-wait held the embedder thread at 99.7% user time while nvidia-powerd cut the GPU's clock from 1005 to 210 MHz. Neither default changes until a measurement says it should. TRACES: VR-015 | PR-004
This commit is contained in:
@@ -19,6 +19,7 @@
|
||||
|
||||
#include <NvInfer.h>
|
||||
#include <cuda_runtime_api.h>
|
||||
#include <cstdlib>
|
||||
#include <opencv2/dnn.hpp>
|
||||
#include <opencv2/imgproc.hpp>
|
||||
|
||||
@@ -45,6 +46,40 @@ inline void check_cuda(cudaError_t e, const char* what) {
|
||||
throw CudaError(std::string(what) + ": " + cudaGetErrorString(e));
|
||||
}
|
||||
|
||||
/// TRACES: VR-015 | PR-004
|
||||
/// Select how a thread waits for the GPU. Must run before the CUDA context is
|
||||
/// created, so every engine constructor calls it and the first one wins.
|
||||
///
|
||||
/// The default (`cudaDeviceScheduleAuto`) spin-waits: `cudaStreamSynchronize`
|
||||
/// burns the calling thread's CPU for the whole of the device's work. Measured
|
||||
/// here, the embedder thread sat at 99.7% *user* time with 0.5 s of system time
|
||||
/// across 183 s — i.e. no blocking syscalls at all — while the GPU ran flat out.
|
||||
///
|
||||
/// On this laptop that is not merely wasted CPU. `nvidia-powerd` arbitrates one
|
||||
/// power budget across CPU and GPU, and the GPU's ceiling was observed dropping
|
||||
/// from 20 W idle to 15 W under our load, with the SM clock *falling* from
|
||||
/// 1005 MHz to 210 MHz once work started. Spinning may therefore be buying
|
||||
/// watts away from the device the pipeline is actually waiting on.
|
||||
///
|
||||
/// SAE_CUDA_BLOCKING_SYNC=1 switches to a blocking wait so the A/B needs no
|
||||
/// rebuild. Default is unchanged until the measurement says otherwise.
|
||||
inline void configure_cuda_sync_once() {
|
||||
static const bool done = [] {
|
||||
const char* env = std::getenv("SAE_CUDA_BLOCKING_SYNC");
|
||||
if (env && env[0] == '1') {
|
||||
cudaError_t e = cudaSetDeviceFlags(cudaDeviceScheduleBlockingSync);
|
||||
std::cerr << "[cuda] sync policy: BlockingSync"
|
||||
<< (e == cudaSuccess ? "" : " (FAILED — context already created)")
|
||||
<< "\n";
|
||||
} else {
|
||||
std::cerr << "[cuda] sync policy: default (spin) — "
|
||||
"set SAE_CUDA_BLOCKING_SYNC=1 to compare\n";
|
||||
}
|
||||
return true;
|
||||
}();
|
||||
(void)done;
|
||||
}
|
||||
|
||||
class TrtLogger : public nvinfer1::ILogger {
|
||||
public:
|
||||
void log(Severity sev, const char* msg) noexcept override {
|
||||
@@ -109,6 +144,7 @@ public:
|
||||
(output_is_fp16_ ? 2 : 4);
|
||||
check_cuda(cudaMalloc(&d_input_, in_bytes), "cudaMalloc input");
|
||||
check_cuda(cudaMalloc(&d_output_, out_bytes), "cudaMalloc output");
|
||||
configure_cuda_sync_once();
|
||||
check_cuda(cudaStreamCreate(&stream_), "cudaStreamCreate");
|
||||
|
||||
context_->setTensorAddress(input_name_.c_str(), d_input_);
|
||||
@@ -293,6 +329,7 @@ public:
|
||||
out_elem_counts_[oi] = count;
|
||||
}
|
||||
|
||||
configure_cuda_sync_once();
|
||||
check_cuda(cudaStreamCreate(&stream_), "cudaStreamCreate");
|
||||
|
||||
std::cerr << "[TrtScrfd] loaded: " << engine_path
|
||||
@@ -462,6 +499,7 @@ public:
|
||||
const std::size_t out_count = static_cast<std::size_t>(kWindow);
|
||||
check_cuda(cudaMalloc(&d_input_, in_count * 4), "cudaMalloc input");
|
||||
check_cuda(cudaMalloc(&d_output_, out_count * 4), "cudaMalloc output");
|
||||
configure_cuda_sync_once();
|
||||
check_cuda(cudaStreamCreate(&stream_), "cudaStreamCreate");
|
||||
context_->setTensorAddress(input_name_.c_str(), d_input_);
|
||||
context_->setTensorAddress(output_name_.c_str(), d_output_);
|
||||
|
||||
@@ -0,0 +1,590 @@
|
||||
#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
|
||||
@@ -32,6 +32,14 @@ struct Config {
|
||||
// scripts/optimizer/SCHEMA.md) for offline threshold-sweep replay via sae_kpn.
|
||||
std::string dump_embeddings_path;
|
||||
|
||||
/// TRACES: VR-015 | PR-004
|
||||
// When set, write a per-node timing and bottleneck report here (src/
|
||||
// benchmark.hpp) and print it at shutdown. Costs one background thread
|
||||
// reading relaxed atomics on a timer, so it is safe to leave on, but a
|
||||
// measurement run should still be isolated (nothing else on the GPU).
|
||||
std::string benchmark_path;
|
||||
int benchmark_interval_ms{100}; // channel-occupancy sampling period
|
||||
|
||||
// ── Sampling ─────────────────────────────────────────────────────────────
|
||||
float sample_fps{1.0f}; // frames to analyse per second of movie
|
||||
float max_decode_fps{0.f}; // wall-clock cap on source decode rate (0 = uncapped)
|
||||
|
||||
+105
-2
@@ -44,10 +44,15 @@
|
||||
// --expand-band-hi <p> store admission ceiling, P(same person) (default: 0.95)
|
||||
// --expand-min-anchor <N> accepted frames before a track confirms (default: 3)
|
||||
// --expand-debug-dir <p> dump promoted mugshots + embeddings here (SAE_DEBUG)
|
||||
// --benchmark <path> write a per-node timing + bottleneck report (JSON) and
|
||||
// print it at shutdown. Says where the run's time went
|
||||
// and which node is pacing it. See src/benchmark.hpp.
|
||||
// --benchmark-interval-ms <N> channel-occupancy sampling period (default: 100)
|
||||
// (SAE_DEBUG only)
|
||||
// --debug-dir <path> debug frames output dir (default: debug_frames)
|
||||
// --crop-context <f> bbox expansion factor for context crops (default: 1.5)
|
||||
|
||||
#include "benchmark.hpp"
|
||||
#include "config.hpp"
|
||||
#include "types.hpp"
|
||||
#include "gallery/embedder_stamp.hpp"
|
||||
@@ -71,9 +76,14 @@
|
||||
|
||||
#include <kpn/kpn.hpp>
|
||||
|
||||
#include <opencv2/core/utility.hpp> // cv::setNumThreads (SAE_CV_THREADS)
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <csignal>
|
||||
#include <cstdlib>
|
||||
#include <cstring>
|
||||
#include <fstream>
|
||||
#include <iostream>
|
||||
#include <map>
|
||||
#include <mutex>
|
||||
@@ -96,6 +106,20 @@ static constexpr std::size_t kSceneJoinDepth = 256;
|
||||
/// actually worked.
|
||||
static std::shared_ptr<SceneBoundaries> scene_stats;
|
||||
|
||||
/// TRACES: VR-015, AR-004 | PR-004
|
||||
/// Set by SIGUSR1, serviced by the wait loop. `kill -USR1 <pid>` on a running
|
||||
/// or WEDGED run prints the benchmark table immediately — channel occupancy
|
||||
/// names the stalled node (full input, empty output) without a debug build or a
|
||||
/// debugger, which is the difference between diagnosing the AR-004 hang in
|
||||
/// seconds and reproducing it under gdb.
|
||||
///
|
||||
/// The handler only stores a flag; all printing happens on the main thread,
|
||||
/// since nothing in the report is async-signal-safe.
|
||||
static std::atomic<bool> g_dump_request{false};
|
||||
extern "C" void sae_on_dump_signal(int) {
|
||||
g_dump_request.store(true, std::memory_order_relaxed);
|
||||
}
|
||||
|
||||
static Config parse_args(int argc, char** argv) {
|
||||
Config cfg;
|
||||
cfg.detector_model = kDefaultDetectorModel;
|
||||
@@ -114,6 +138,8 @@ static Config parse_args(int argc, char** argv) {
|
||||
else if (arg("--gallery")) cfg.gallery_path = next();
|
||||
else if (arg("--output")) cfg.output_path = next();
|
||||
else if (arg("--dump-embeddings")) cfg.dump_embeddings_path = next();
|
||||
else if (arg("--benchmark")) cfg.benchmark_path = next();
|
||||
else if (arg("--benchmark-interval-ms")) cfg.benchmark_interval_ms = std::stoi(next());
|
||||
else if (arg("--fps")) cfg.sample_fps = std::stof(next());
|
||||
else if (arg("--max-decode-fps")) cfg.max_decode_fps = std::stof(next());
|
||||
else if (arg("--start")) cfg.start_sec = std::stod(next());
|
||||
@@ -174,6 +200,21 @@ static Config parse_args(int argc, char** argv) {
|
||||
// ── Main ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
/// TRACES: VR-015 | PR-004
|
||||
// OpenCV here is built against TBB, so cv::parallel_for_ opens an arena of
|
||||
// nproc-1 workers (19 on a 20-core box) *on top of* KPN's one thread per
|
||||
// node. Two schedulers, neither aware of the other, on the same cores.
|
||||
//
|
||||
// SAE_CV_THREADS=1 hands concurrency entirely to KPN, which is where this
|
||||
// pipeline's parallelism is supposed to come from. Worth measuring rather
|
||||
// than assuming: TBB fan-out inside warpAffine is free speed when the
|
||||
// pipeline is otherwise idle, so this can cut either way. Unset = default.
|
||||
if (const char* t = std::getenv("SAE_CV_THREADS")) {
|
||||
const int n = std::atoi(t);
|
||||
cv::setNumThreads(n);
|
||||
std::cerr << "[opencv] cv::setNumThreads(" << n << ")\n";
|
||||
}
|
||||
|
||||
Config cfg;
|
||||
try {
|
||||
cfg = parse_args(argc, argv);
|
||||
@@ -236,7 +277,15 @@ int main(int argc, char** argv) {
|
||||
// AR-016: a film ends with faces on screen and those tracks have not timed
|
||||
// out. Without this flush the closing scene's cast is silently never
|
||||
// emitted — a loss that reads as a recognition miss, not a bookkeeping bug.
|
||||
sink_fn.set_pre_write_hook([registry](double last_ts) { registry->flush(last_ts); });
|
||||
/// TRACES: VR-015 | PR-004
|
||||
// Last timestamp the pipeline reached, latched on the way out. It is what
|
||||
// turns wall-clock seconds into the number that matters — seconds of film
|
||||
// per second of run — and the sink is the only node that knows it.
|
||||
std::atomic<double> film_sec{0.0};
|
||||
sink_fn.set_pre_write_hook([registry, &film_sec](double last_ts) {
|
||||
film_sec.store(last_ts, std::memory_order_release);
|
||||
registry->flush(last_ts);
|
||||
});
|
||||
#ifdef SAE_DEBUG
|
||||
DebugRendererFunc debug_fn {cfg};
|
||||
#endif
|
||||
@@ -302,8 +351,23 @@ int main(int argc, char** argv) {
|
||||
return false;
|
||||
});
|
||||
|
||||
/// TRACES: VR-015 | PR-004
|
||||
// Sampling must start with the network and stop before it is destroyed:
|
||||
// channel fill is instantaneous, and by the time a run ends everything
|
||||
// has drained, so a single read at shutdown reports an idle pipeline no
|
||||
// matter how congested it was.
|
||||
sae::bench::BenchmarkRecorder bench{cfg.benchmark_interval_ms};
|
||||
const bool benchmarking = !cfg.benchmark_path.empty();
|
||||
|
||||
std::cerr << "[main] starting pipeline…\n";
|
||||
net.start();
|
||||
if (benchmarking) {
|
||||
bench.start([&net] { return net.network_snapshot(); });
|
||||
std::signal(SIGUSR1, sae_on_dump_signal);
|
||||
std::cerr << "[benchmark] sampling every " << cfg.benchmark_interval_ms
|
||||
<< "ms — `kill -USR1 " << getpid()
|
||||
<< "` to dump the table now (works while hung)\n";
|
||||
}
|
||||
|
||||
// Wait until BOTH terminal branches finish: result_sink (face pipeline)
|
||||
// and, when enabled, scene_detector (the dense TransNetV2 branch, which
|
||||
@@ -311,12 +375,51 @@ int main(int argc, char** argv) {
|
||||
// pre-set true when scene detection is disabled.
|
||||
while ((!done.load(std::memory_order_acquire) ||
|
||||
!scene_done.load(std::memory_order_acquire)) &&
|
||||
!node_crashed.load(std::memory_order_acquire))
|
||||
!node_crashed.load(std::memory_order_acquire)) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(100));
|
||||
/// TRACES: VR-015, AR-004 | PR-004
|
||||
if (g_dump_request.exchange(false, std::memory_order_relaxed))
|
||||
bench.dump_live(std::cerr, film_sec.load(std::memory_order_acquire));
|
||||
}
|
||||
|
||||
// Latch the counters before stop(): they stay readable afterwards, but
|
||||
// only while the network object is alive, and this keeps the numbers
|
||||
// describing the run rather than the teardown.
|
||||
if (benchmarking) bench.stop();
|
||||
|
||||
net.stop();
|
||||
net.print_diagnostics();
|
||||
|
||||
/// TRACES: VR-015 | PR-004
|
||||
if (benchmarking && bench.has_data()) {
|
||||
const double film = film_sec.load(std::memory_order_acquire);
|
||||
bench.print(std::cerr, film);
|
||||
|
||||
nlohmann::json run_cfg{
|
||||
{"movie", cfg.movie_path},
|
||||
{"gallery", cfg.gallery_path},
|
||||
{"gallery_actors", gallery.actors.size()},
|
||||
{"sample_fps", cfg.sample_fps},
|
||||
{"min_face_px", cfg.min_face_px},
|
||||
{"max_faces", cfg.max_faces},
|
||||
{"embed_batch", cfg.embed_batch_size},
|
||||
{"expand_gallery", cfg.expand_gallery},
|
||||
{"scene_detect", cfg.scene_detect},
|
||||
{"detector_engine", cfg.detector_engine},
|
||||
{"arcface_engine", cfg.arcface_engine},
|
||||
{"detector_model", cfg.detector_model},
|
||||
{"arcface_model", cfg.arcface_model},
|
||||
};
|
||||
std::ofstream bf(cfg.benchmark_path);
|
||||
if (bf) {
|
||||
bf << bench.to_json(run_cfg, film).dump(2) << "\n";
|
||||
std::cerr << "[benchmark] wrote " << cfg.benchmark_path << "\n";
|
||||
} else {
|
||||
std::cerr << "[benchmark] ERROR: could not write "
|
||||
<< cfg.benchmark_path << "\n";
|
||||
}
|
||||
}
|
||||
|
||||
/// TRACES: AR-004 | SR-002
|
||||
// A dropped frame does not degrade a result, it silently changes one —
|
||||
// the output is a claim about footage that was never analysed, and
|
||||
|
||||
Reference in New Issue
Block a user