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