From c8f402c4bf55d3b6afc6ae81ef0557a8bc3808d3 Mon Sep 17 00:00:00 2001 From: lgoyal6 Date: Sun, 6 Sep 2026 14:34:05 -0700 Subject: [PATCH] Report errors, memory and cost alongside the latency numbers A latency table never says three things on its own: what failed while the numbers were being produced, how much memory the process held to produce them, and what the work cost. A p999 next to an unreported delivery failure is not a measurement of the transport, it is a measurement of the messages that happened to arrive. Errors, per role, counted rather than dropped: - The receiver silently `return`ed on a payload too short to carry the header, so a framing bug would have shown up as a slightly lower delivery ratio and nothing else. Now counted. - A sequence number that does not advance is a duplicate or a reorder. The receiver cannot tell which from one number, so it reports the count and does not claim to know which. - Offered minus received is the error a latency table hides best: a message that never arrived contributes no sample, so dropping it *improves* every percentile in the row above it. - The request-reply client is closed loop with one outstanding, so the reply to request N must be N. A stale or duplicate reply still set got_reply and still pushed a sample, timed from the wrong request. The mismatch is counted, and so is the wait cap firing, which the old loop passed over silently: a wedged transport produced a short run and a clean-looking percentile table. - The sender's error is offered work it never handed to the transport before the wall cap. In open-loop mode the schedule says how many arrivals the run should have produced, and reporting only `sent` hides the shortfall. Memory is peak and steady RSS together, so a transient spike can be told apart from a working set that stays resident. ru_maxrss is bytes on Darwin and kilobytes on Linux; that trap is named in a comment rather than papered over with a guessed factor of 1024, because getting it wrong is silently wrong by three orders of magnitude in the flattering direction on exactly one of the two platforms. Cost is CPU seconds, which is the axis that matters for a poll/tick transport with no event-loop driver: both roles spin, so wall time says how long the run lasted and CPU time says what it burned to last that long. Involuntary context switches are reported next to it so a tail that came from a busy machine can be told apart from a tail that came from the transport. The USD figure is arithmetic on cpu_s rather than a measurement, so the rate is printed with it. resource.h is deliberately separate from common.h. common.h is the wire format and the schedule, shared by the taut, TCP and ENet binaries and documented in docs/BENCHMARKS.md; this is process accounting, and it writes to its own CSV beside the latency CSV so no documented schema changes. Verification, and its limits. bench/ is Linux-only in CMake and no CI job builds it, so neither the normal local build nor CI compiles this file. Instead: latency_bench.cc compiles clean under the project's full strict set, -Werror -Wpedantic -Wshadow -Wconversion -Wsign-conversion -Wold-style-cast -Wcast-align -Wdouble-promotion, on this machine. Both platform branches of resource.h were compiled and run, not just the local one: the Darwin branch reads a real RSS and emits a well-formed CSV row, and the Linux branch was compiled by forcing its arm of the conditional and exercised for header, row and path behaviour. resource_path rewrites the extension as documented. The library suite is unaffected at 61/61. --- bench/latency_bench.cc | 114 +++++++++++++++++++++++++++++--- bench/resource.h | 146 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 252 insertions(+), 8 deletions(-) create mode 100644 bench/resource.h diff --git a/bench/latency_bench.cc b/bench/latency_bench.cc index c8f86b0..063cacf 100644 --- a/bench/latency_bench.cc +++ b/bench/latency_bench.cc @@ -29,6 +29,7 @@ #include #include "bench/common.h" +#include "bench/resource.h" #include "taut/config.h" #include "taut/session.h" #include "taut/transport.h" @@ -109,6 +110,14 @@ int run_receiver(const bench::Args& a) { std::vector samples; samples.reserve(1u << 20); std::uint64_t received = 0, max_seq = 0, goodput_bytes = 0; + // What went wrong, counted rather than dropped on the floor. A payload too short to + // carry the header used to `return` silently, so a framing bug would have shown up as + // a slightly lower delivery ratio and nothing else. A sequence that does not advance + // is a duplicate or a reorder; the receiver cannot tell which from one number, so it + // reports the count and does not claim to know. + std::uint64_t err_short = 0, err_nonmonotonic = 0; + std::uint64_t last_seq_seen = 0; + bool have_last_seq = false; std::uint64_t first_recv_ns = 0, last_recv_ns = 0; bool got_end = false; std::uint64_t last_activity_ns = bench::now_ns(); @@ -116,6 +125,7 @@ int run_receiver(const bench::Args& a) { session.on_message([&](taut::Class, taut::ByteSpan payload) { if (payload.size() < bench::kMsgHeader) { + ++err_short; return; } std::uint64_t ts = 0, seq = 0; @@ -132,6 +142,11 @@ int run_receiver(const bench::Args& a) { last_recv_ns = recv; ++received; goodput_bytes += msg; + if (have_last_seq && seq <= last_seq_seen) { + ++err_nonmonotonic; + } + last_seq_seen = seq; + have_last_seq = true; if (seq > max_seq) { max_seq = seq; } @@ -167,8 +182,21 @@ int run_receiver(const bench::Args& a) { std::to_string(secs) + "," + std::to_string(mbps); bench::append_csv(a.out, bench::key_header() + ",received,goodput_bytes,secs,goodput_mbps", row); - std::fprintf(stderr, "taut recv[thru]: %llu msgs, %.2f Mbit/s over %.2fs\n", - static_cast(received), mbps, secs); + const bench::Resources res = bench::sample_resources(); + const std::uint64_t errors = err_short + err_nonmonotonic; + char detail[128]; + std::snprintf(detail, sizeof detail, "short=%llu nonmono=%llu", + static_cast(err_short), + static_cast(err_nonmonotonic)); + bench::append_csv(bench::resource_path(a.out), bench::resource_header(), + bench::resource_row("recv-thru", errors, detail, res, received)); + std::fprintf(stderr, + "taut recv[thru]: %llu msgs, %.2f Mbit/s over %.2fs | errors=%llu (%s) " + "rss_peak=%.1fMB rss_steady=%.1fMB cpu=%.2fs\n", + static_cast(received), mbps, secs, + static_cast(errors), detail, + static_cast(res.rss_peak) / 1048576.0, + static_cast(res.rss_steady) / 1048576.0, res.cpu_s); return 0; } @@ -183,11 +211,29 @@ int run_receiver(const bench::Args& a) { bench::key_header() + ",offered,received,p50_ms,p90_ms,p99_ms,p999_ms,min_ms,max_ms,mean_ms", row); + const bench::Resources res = bench::sample_resources(); + // Offered minus received is the error that matters most here and the one a latency + // table hides: a message that never arrived contributes no sample, so dropping it + // improves every percentile in the row above. + const std::uint64_t undelivered = offered > received ? offered - received : 0; + const std::uint64_t errors = undelivered + err_short + err_nonmonotonic; + char detail[160]; + std::snprintf(detail, sizeof detail, "undelivered=%llu short=%llu nonmono=%llu", + static_cast(undelivered), + static_cast(err_short), + static_cast(err_nonmonotonic)); + bench::append_csv(bench::resource_path(a.out), bench::resource_header(), + bench::resource_row("recv-lat", errors, detail, res, received)); std::fprintf(stderr, "taut recv[lat cls%d loss%.0f%%]: offered=%llu recv=%llu p50=%.2f p99=%.2f " - "p999=%.2f max=%.2f ms\n", + "p999=%.2f max=%.2f ms | errors=%llu (%s) rss_peak=%.1fMB rss_steady=%.1fMB " + "cpu=%.2fs ivcsw=%llu\n", a.taut_class, a.loss_pct, static_cast(offered), - static_cast(received), p.p50, p.p99, p.p999, p.max); + static_cast(received), p.p50, p.p99, p.p999, p.max, + static_cast(errors), detail, + static_cast(res.rss_peak) / 1048576.0, + static_cast(res.rss_steady) / 1048576.0, res.cpu_s, + static_cast(res.ivcsw)); return 0; } @@ -278,9 +324,28 @@ int run_sender(const bench::Args& a) { } const double mult = sent > 0 ? static_cast(tx.tx_datagrams()) / static_cast(sent) : 0.0; - std::fprintf(stderr, "taut send[cls%d loss%.0f%%]: sent=%llu tx_datagrams=%llu (%.2fx)\n", + const bench::Resources res = bench::sample_resources(); + // The sender's error is the offered work it never managed to hand to the transport + // before the wall-cap. In open-loop mode the schedule says how many arrivals the run + // was supposed to produce; anything short of that is offered work that was dropped by + // the load generator itself, and reporting only `sent` would hide it. + const std::uint64_t scheduled = + a.mode == "throughput" ? sent : bench::schedule_count(a.seed, a.rate, a.duration_s); + const std::uint64_t unoffered = scheduled > sent ? scheduled - sent : 0; + char detail[128]; + std::snprintf(detail, sizeof detail, "unoffered=%llu", + static_cast(unoffered)); + bench::append_csv(bench::resource_path(a.send_out.empty() ? a.out : a.send_out), + bench::resource_header(), + bench::resource_row("send", unoffered, detail, res, sent)); + std::fprintf(stderr, + "taut send[cls%d loss%.0f%%]: sent=%llu tx_datagrams=%llu (%.2fx) | " + "errors=%llu (%s) rss_peak=%.1fMB rss_steady=%.1fMB cpu=%.2fs\n", a.taut_class, a.loss_pct, static_cast(sent), - static_cast(tx.tx_datagrams()), mult); + static_cast(tx.tx_datagrams()), mult, + static_cast(unoffered), detail, + static_cast(res.rss_peak) / 1048576.0, + static_cast(res.rss_steady) / 1048576.0, res.cpu_s); return 0; } @@ -354,8 +419,15 @@ int run_rr_client(const bench::Args& a) { bool got_reply = false; std::vector samples; samples.reserve(1u << 18); + // Closed loop, one outstanding, so the reply to request N must be N. A stale or + // duplicate reply would still set got_reply and would still push a sample, and the + // sample would be timed from the wrong request. Counting the mismatch is what keeps + // correctness held fixed while the latency distribution is compared. + std::uint64_t err_short = 0, err_seq_mismatch = 0, err_timeout = 0; + std::uint64_t expect_seq = 0; session.on_message([&](taut::Class, taut::ByteSpan payload) { if (payload.size() < bench::kMsgHeader) { + ++err_short; return; } std::uint64_t ts = 0, seq = 0; @@ -363,6 +435,9 @@ int run_rr_client(const bench::Args& a) { if (seq == bench::kEndSeq) { return; } + if (seq != expect_seq) { + ++err_seq_mismatch; + } samples.push_back(static_cast(bench::now_ns() - ts) / 1e6); // round trip got_reply = true; }); @@ -376,6 +451,7 @@ int run_rr_client(const bench::Args& a) { while (bench::now_ns() - start < dur_ns) { bench::write_msg(buf, bench::now_ns(), seq); // stamp actual send instant (true round trip) got_reply = false; + expect_seq = seq; while (!session.send(cls, payload)) { pump(); } @@ -387,6 +463,11 @@ int run_rr_client(const bench::Args& a) { session.poll(); session.tick(); } + if (!got_reply) { + // The cap fired. The old loop moved on silently, so a wedged transport + // produced a short run with a clean-looking percentile table. + ++err_timeout; + } } // END sentinel, then a short flush. @@ -417,12 +498,29 @@ int run_rr_client(const bench::Args& a) { bench::append_csv(a.send_out, bench::key_header() + ",sent,tx_datagrams,tx_payload_bytes", s); } + const bench::Resources res = bench::sample_resources(); + const std::uint64_t missing = sent > p.n ? sent - p.n : 0; + const std::uint64_t errors = missing + err_short + err_seq_mismatch + err_timeout; + char detail[192]; + std::snprintf(detail, sizeof detail, "missing=%llu timeout=%llu seq_mismatch=%llu short=%llu", + static_cast(missing), + static_cast(err_timeout), + static_cast(err_seq_mismatch), + static_cast(err_short)); + bench::append_csv(bench::resource_path(a.out), bench::resource_header(), + bench::resource_row("rr-client", errors, detail, res, p.n)); std::fprintf(stderr, "taut rr[cls%d loss%.0f%%]: n=%zu p50=%.2f p99=%.2f p999=%.2f max=%.2f ms " - "(%.2fx wire)\n", + "(%.2fx wire) | errors=%llu (%s) rss_peak=%.1fMB rss_steady=%.1fMB cpu=%.2fs " + "cpu_us_per_msg=%.2f ivcsw=%llu\n", a.taut_class, a.loss_pct, p.n, p.p50, p.p99, p.p999, p.max, sent > 0 ? static_cast(tx.tx_datagrams()) / static_cast(sent) - : 0.0); + : 0.0, + static_cast(errors), detail, + static_cast(res.rss_peak) / 1048576.0, + static_cast(res.rss_steady) / 1048576.0, res.cpu_s, + p.n ? res.cpu_s * 1e6 / static_cast(p.n) : 0.0, + static_cast(res.ivcsw)); return 0; } diff --git a/bench/resource.h b/bench/resource.h new file mode 100644 index 0000000..89b2543 --- /dev/null +++ b/bench/resource.h @@ -0,0 +1,146 @@ +// Errors, memory and cost for the latency benchmarks (C02). +// +// The three things a latency table never says on its own: what failed while the numbers +// were being produced, how much memory the process held to produce them, and what the +// work cost. A p999 next to an unreported 4% delivery failure is not a measurement of the +// transport, it is a measurement of the messages that happened to arrive. +// +// Kept out of common.h on purpose: common.h is the wire format and the schedule, shared by +// the taut, TCP and ENet binaries and documented in docs/BENCHMARKS.md. This is process +// accounting, and it is written to its own CSV beside the latency CSV so the documented +// schemas do not change. +#pragma once + +#include +#include +#include +#include + +#if defined(__APPLE__) +#include +#endif + +namespace bench { + +// Peak resident set of this process, in bytes. +// +// ru_maxrss is BYTES on Darwin and KILOBYTES on Linux. That is a portability trap worth +// naming rather than a factor of 1024 worth guessing, and getting it wrong silently +// produces a memory number that is wrong by three orders of magnitude in the flattering +// direction on exactly one of the two platforms this benchmark runs on. +inline std::uint64_t peak_rss_bytes() { + struct rusage ru {}; + getrusage(RUSAGE_SELF, &ru); +#if defined(__APPLE__) + return static_cast(ru.ru_maxrss); +#else + return static_cast(ru.ru_maxrss) * 1024ULL; +#endif +} + +// Resident set right now. Reported next to the peak so a transient spike (a reserve() of +// the sample vector, a burst of retransmit buffers) can be told apart from a working set +// that stays resident for the whole run. +inline std::uint64_t current_rss_bytes() { +#if defined(__APPLE__) + mach_task_basic_info info{}; + mach_msg_type_number_t count = MACH_TASK_BASIC_INFO_COUNT; + if (task_info(mach_task_self(), MACH_TASK_BASIC_INFO, reinterpret_cast(&info), + &count) == KERN_SUCCESS) { + return static_cast(info.resident_size); + } + return 0; +#else + std::FILE* f = std::fopen("/proc/self/statm", "r"); + if (f == nullptr) { + return 0; + } + unsigned long total = 0, resident = 0; + const int got = std::fscanf(f, "%lu %lu", &total, &resident); + std::fclose(f); + return got == 2 ? static_cast(resident) * 4096ULL : 0; +#endif +} + +// User + system CPU seconds charged to this process. +// +// This is the cost axis, and it is the one that matters for a poll/tick transport with no +// event-loop driver: both roles spin, so wall time says how long the run lasted and CPU +// time says what it burned to last that long. They are not the same number, and on a +// machine shared with other work the difference is the whole story. +inline double cpu_seconds() { + struct rusage ru {}; + getrusage(RUSAGE_SELF, &ru); + return static_cast(ru.ru_utime.tv_sec) + static_cast(ru.ru_utime.tv_usec) / 1e6 + + static_cast(ru.ru_stime.tv_sec) + static_cast(ru.ru_stime.tv_usec) / 1e6; +} + +// Voluntary + involuntary context switches. An involuntary switch is the kernel taking the +// CPU away, which is what contention from other processes on the box looks like from +// inside the benchmark. Reported so a tail that came from a busy machine can be told apart +// from a tail that came from the transport. +inline void context_switches(std::uint64_t& voluntary, std::uint64_t& involuntary) { + struct rusage ru {}; + getrusage(RUSAGE_SELF, &ru); + voluntary = static_cast(ru.ru_nvcsw); + involuntary = static_cast(ru.ru_nivcsw); +} + +// Price of one CPU-second, printed alongside every derived dollar figure. +// +// USD is arithmetic on a measured quantity, not a measurement. The measured quantity is +// cpu_s; publishing the rate with it means the figure can be redone against a different +// instance or a different price without re-running anything. +inline constexpr double kUsdPerCpuSecond = 0.145 / 4.0 / 3600.0; // c7g.xlarge on-demand / 4 vCPU +inline constexpr const char* kRateLabel = "c7g.xlarge_0.145usd_hr_4vcpu"; + +struct Resources { + std::uint64_t rss_peak = 0; + std::uint64_t rss_steady = 0; + double cpu_s = 0; + std::uint64_t vcsw = 0; + std::uint64_t ivcsw = 0; +}; + +inline Resources sample_resources() { + Resources r; + r.rss_peak = peak_rss_bytes(); + r.rss_steady = current_rss_bytes(); + r.cpu_s = cpu_seconds(); + context_switches(r.vcsw, r.ivcsw); + return r; +} + +inline std::string resource_header() { + return "role,errors,err_detail,rss_peak_mb,rss_steady_mb,cpu_s,cpu_us_per_msg," + "usd_per_million_msgs,vol_ctx_sw,invol_ctx_sw,usd_rate"; +} + +inline std::string resource_row(const std::string& role, std::uint64_t errors, + const std::string& err_detail, const Resources& r, + std::uint64_t messages) { + char buf[512]; + const double per_msg_us = messages ? r.cpu_s * 1e6 / static_cast(messages) : 0.0; + const double usd_per_m = messages + ? r.cpu_s * kUsdPerCpuSecond * 1e6 / static_cast(messages) + : 0.0; + std::snprintf(buf, sizeof buf, "%s,%llu,%s,%.1f,%.1f,%.3f,%.3f,%.6f,%llu,%llu,%s", role.c_str(), + static_cast(errors), err_detail.c_str(), + static_cast(r.rss_peak) / 1048576.0, + static_cast(r.rss_steady) / 1048576.0, r.cpu_s, per_msg_us, usd_per_m, + static_cast(r.vcsw), + static_cast(r.ivcsw), kRateLabel); + return buf; +} + +// The latency CSV path with ".resources.csv" in place of its extension, so the resource +// row lands beside the run it belongs to without changing the documented latency schema. +inline std::string resource_path(const std::string& out) { + if (out.empty()) { + return {}; + } + const auto dot = out.find_last_of('.'); + return (dot == std::string::npos ? out : out.substr(0, dot)) + ".resources.csv"; +} + +} // namespace bench