From d4e373e5ce9f052354015451695676a2f789fc76 Mon Sep 17 00:00:00 2001 From: Everett Badeaux Date: Thu, 5 Nov 2026 15:00:55 -0600 Subject: [PATCH] Add a baseline performance test --- performance_tests/CMakeLists.txt | 4 +- .../writer_performance_test/CMakeLists.txt | 13 + .../writer_performance_test/README.md | 151 +++++ .../writer_performance_test/histogram.h | 102 ++++ .../writer_performance_test/plan.cpp | 215 +++++++ .../writer_performance_test/plan.h | 67 +++ .../writer_performance_test/receiver.cpp | 257 ++++++++ .../writer_performance_test/receiver.h | 63 ++ .../writer_performance_test/recording_run.cpp | 314 ++++++++++ .../writer_performance_test/recording_run.h | 19 + .../writer_performance_test/report.cpp | 510 ++++++++++++++++ .../writer_performance_test/report.h | 58 ++ .../writer_performance_test/scenario.h | 119 ++++ .../writer_performance_test.cpp | 569 ++++++++++++++++++ 14 files changed, 2460 insertions(+), 1 deletion(-) create mode 100644 performance_tests/writer_performance_test/CMakeLists.txt create mode 100644 performance_tests/writer_performance_test/README.md create mode 100644 performance_tests/writer_performance_test/histogram.h create mode 100644 performance_tests/writer_performance_test/plan.cpp create mode 100644 performance_tests/writer_performance_test/plan.h create mode 100644 performance_tests/writer_performance_test/receiver.cpp create mode 100644 performance_tests/writer_performance_test/receiver.h create mode 100644 performance_tests/writer_performance_test/recording_run.cpp create mode 100644 performance_tests/writer_performance_test/recording_run.h create mode 100644 performance_tests/writer_performance_test/report.cpp create mode 100644 performance_tests/writer_performance_test/report.h create mode 100644 performance_tests/writer_performance_test/scenario.h create mode 100644 performance_tests/writer_performance_test/writer_performance_test.cpp diff --git a/performance_tests/CMakeLists.txt b/performance_tests/CMakeLists.txt index 67afe21..fc7f158 100644 --- a/performance_tests/CMakeLists.txt +++ b/performance_tests/CMakeLists.txt @@ -1,4 +1,6 @@ add_executable(performance_test performance_test.cpp) target_link_libraries(performance_test PRIVATE spectator-registry -) \ No newline at end of file +) + +add_subdirectory(writer_performance_test) diff --git a/performance_tests/writer_performance_test/CMakeLists.txt b/performance_tests/writer_performance_test/CMakeLists.txt new file mode 100644 index 0000000..083900d --- /dev/null +++ b/performance_tests/writer_performance_test/CMakeLists.txt @@ -0,0 +1,13 @@ +# Measures recording through the writers for a list of configurations. See writer_performance_test.cpp. +add_executable(writer_performance_test + writer_performance_test.cpp + plan.cpp + receiver.cpp + recording_run.cpp + report.cpp +) + +target_link_libraries(writer_performance_test + PRIVATE + spectator-registry +) diff --git a/performance_tests/writer_performance_test/README.md b/performance_tests/writer_performance_test/README.md new file mode 100644 index 0000000..8f1107e --- /dev/null +++ b/performance_tests/writer_performance_test/README.md @@ -0,0 +1,151 @@ +# Writer performance test + +Measures what it costs to record meters through the writers, and how much of what was recorded arrives. It does so +for a list of configurations: the socket writer, the buffer size, the number of recording threads, the load, and the +way the meters are used. It needs nothing but this repository, because the test starts its own receiver in place of +spectatord. + +## Run it + +Performance numbers only mean something from an optimized build. Build one in a directory of its own, next to the +normal build: + +```shell +source venv/bin/activate +BUILD_DIR=cmake-build-release BUILD_TYPE=Release ./build.sh skiptest +``` + +The first time, this also builds the dependencies in release mode, which takes a few minutes. + +Then, to see the full performance: + +```shell +./cmake-build-release/performance_tests/writer_performance_test/writer_performance_test --full +``` + +There are three sizes of run: + +| Option | What it measures | Time | +|-----------|------------------------------------------------------------------------------------------|------------------| +| (none) | Both writers with 1, 4 and 8 threads, three buffer sizes, one paced rate, the meter uses | about 2 minutes | +| `--full` | The same with 1 to 16 threads, every buffer size at every thread count, two paced rates | about 12 minutes | +| `--smoke` | Two configurations at a low rate; fails if a recorded line does not arrive | a few seconds | + +Add `--list` to see the configurations of a run and how long they take, without running them. + +The test also runs from the normal build, for instance to check that it works +(`./cmake-build/performance_tests/writer_performance_test/writer_performance_test --smoke`), but it then warns that +its numbers say little. + +## How it works + +For every configuration, the test + +1. opens a socket of its own on this machine, a UDP port or a Unix socket in `/tmp`, and reads everything that + arrives there, counting datagrams and lines; +2. starts a second process, which creates a `Registry` for that socket through the public API and lets every thread + record with a `Counter`; +3. lets the threads record for a short warm-up, and then measures them for a fixed time; +4. lets that process exit normally, so that the library sends the lines it still holds, and compares what was + recorded with what arrived. + +Every configuration gets a process of its own, because the library sets its writer up once per process. Every +configuration is measured three times, in rotation, and the report shows the median. + +The threads record in one of two ways: + +- **Flat out:** as fast as they can. This shows the highest rate, how it changes with the number of threads, and how + long a call can take when the threads record faster than the lines are sent. +- **Paced:** a fixed number of lines per second and thread, 20,000 unless `--rate` says otherwise. This shows how + long a call takes under an ordinary load, and how long a line takes to reach the receiver. + +The counter is used in one of four ways: created once per thread, or created for every line as in +`registry.CreateCounter(name, tags).Increment()`, each with two tags or with none. Everything but the comparison of +these four uses a counter that was created once and has two tags. + +## Reading the report + +The report has one table for each kind of load, and one that compares the ways of using a counter. + +| Column | Meaning | +|------------------------------------------|----------------------------------------------------------------------------| +| `lines/s`, `per thread` | Lines recorded per second during the measurement, by all threads together and by each of them. The median of the measurements. | +| `speed-up` | `lines/s` as a multiple of the same configuration with one thread. | +| `spread` | How far the slowest and the fastest measurement are apart, as a share of the median. A difference between two configurations that is smaller than this is not a difference. | +| `call p50`, `p99`, `p99.9`, `longest` | How long a call that records one line took. Half of the calls took no longer than `p50`, and 99% no longer than `p99`. | +| `cpu/line` | Processor time of the recording process per line, on all its threads, which includes the thread that sends. | +| `arrived` | The share of the recorded lines that the receiver got. `100%` is shown only if every line arrived. | +| `lines/datagram` | How many lines a datagram held on average. | +| `delay p50`, `delay p99`, `longest delay` | Paced load: the time from recording a line to its arrival at the receiver. | +| `bytes/line` | The length of a line, which the tags change. | +| `vs baseline` | With `--baseline`: `lines/s` as a multiple of the earlier run's. | + +## Options + +| Option | Meaning | +|-------------------|----------------------------------------------------------------------------------------------| +| `--full` | The long run, see above | +| `--smoke` | The check that the test works, see above | +| `--list` | Print the configurations and the time they take, and stop | +| `--writers LIST` | `udp`, `unix` or both: `--writers unix` | +| `--threads LIST` | Numbers of recording threads: `--threads 1,2,4,8,16,32` | +| `--buffers LIST` | Buffer sizes in bytes, `0` for no buffering: `--buffers 0,1024,8192` | +| `--rate N` | Lines per second and thread under the paced load | +| `--seconds S` | How long each configuration is measured | +| `--warmup S` | How long the threads record before the measurement begins | +| `--repeats N` | How often each configuration is measured | +| `--csv FILE` | Also write every measurement to `FILE` | +| `--baseline FILE` | A file that an earlier run wrote with `--csv`: the report shows this run against it | +| `--label TEXT` | A name for the run, for the report and the CSV file | + +With `--threads` or `--buffers`, every buffer size is measured with every number of threads. + +## Comparing two versions of the library + +Run the test on the first version and keep its measurements. Then build the second version the same way and run the +test against them: + +```shell +# on the first version, for example main +./cmake-build-release/performance_tests/writer_performance_test/writer_performance_test --full --label main --csv main.csv + +# on the second version +./cmake-build-release/performance_tests/writer_performance_test/writer_performance_test --full --label my-change --baseline main.csv +``` + +The second report has a column `vs baseline`, in which `2.00x` means twice the lines per second of the first +version. Compare runs from the same machine only. + +## Numbers you can rely on + +- Use an optimized build. The top of the report says whether it is one. +- Run the test on a machine that is doing nothing else, and on Linux if the numbers are to say something about + production. +- Look at `spread` before comparing two rows. +- The receiver runs on the same machine as the recording threads and takes processor time too. + +## Configurations that are left out + +A system limits how large a datagram can be, and the test leaves out the buffer sizes that do not fit; it lists them +under the report. macOS, for example, sends at most 2048 bytes on a Unix datagram socket by default, so there the +Unix writer is measured with a buffer of 512 bytes only. Linux carries every size that the test uses. + +## Measuring a new setting of the writer + +1. Add the setting to `Scenario` in `scenario.h`: the field, and its part in `Key()`, `ToArguments()` and + `FromArguments()`. +2. Pass it to the `WriterConfig` in `recording_run.cpp`. +3. Add the values to measure to `Selection` in `plan.h`, and the scenarios to `MakePlan()` in `plan.cpp`. +4. Show it in the tables and in the CSV file, both in `report.cpp`. + +## Files + +| File | Contents | +|----------------------------------------|-----------------------------------------------------------------------| +| `writer_performance_test.cpp` | The command line, starting the recording process, the run | +| `scenario.h` | One configuration, and how it is handed to the recording process | +| `plan.h`, `plan.cpp` | Which configurations a run consists of | +| `recording_run.h`, `recording_run.cpp` | The recording process: the threads that record, and what they measure | +| `receiver.h`, `receiver.cpp` | The receiving socket that stands in for spectatord | +| `report.h`, `report.cpp` | The tables, the CSV file, the comparison with a baseline | +| `histogram.h` | Counting durations, for the percentiles | diff --git a/performance_tests/writer_performance_test/histogram.h b/performance_tests/writer_performance_test/histogram.h new file mode 100644 index 0000000..506d262 --- /dev/null +++ b/performance_tests/writer_performance_test/histogram.h @@ -0,0 +1,102 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace perf { + +// Counts values, such as durations in nanoseconds, in buckets that get wider as the values get larger: every value +// below 64 has a bucket of its own, and every power of two above that is split into 32 buckets, so a value is off by +// about 3% at most. Recording a value is one array increment and never allocates, which keeps the measurement out of +// the way of what it measures. +class Histogram +{ + public: + void Record(std::uint64_t value) noexcept + { + value = std::min(value, kLargestValue); + ++m_buckets[BucketOf(value)]; + ++m_count; + m_max = std::max(m_max, value); + } + + void Merge(const Histogram& other) noexcept + { + for (std::size_t bucket = 0; bucket < m_buckets.size(); ++bucket) + { + m_buckets[bucket] += other.m_buckets[bucket]; + } + m_count += other.m_count; + m_max = std::max(m_max, other.m_max); + } + + std::uint64_t Count() const noexcept { return m_count; } + + // The largest value recorded, or zero if nothing was recorded. + std::uint64_t Max() const noexcept { return m_max; } + + // The value that the given share of the recorded values does not exceed, for example 0.99 for the 99th + // percentile. Zero if nothing was recorded. + std::uint64_t Percentile(double share) const noexcept + { + if (m_count == 0) + { + return 0; + } + const auto wanted = std::max( + 1, static_cast(std::ceil(share * static_cast(m_count)))); + std::uint64_t seen = 0; + for (std::size_t bucket = 0; bucket < m_buckets.size(); ++bucket) + { + seen += m_buckets[bucket]; + if (seen >= wanted) + { + return std::min(LargestValueOf(bucket), m_max); + } + } + return m_max; + } + + private: + static constexpr int kSplitBits = 5; // 32 buckets per power of two + static constexpr std::uint64_t kSplit = std::uint64_t{1} << kSplitBits; // 32 + static constexpr std::uint64_t kExactBelow = kSplit * 2; // 64 + static constexpr int kLargestPower = 41; // 2^42 ns is over an hour + static constexpr std::uint64_t kLargestValue = (std::uint64_t{1} << (kLargestPower + 1)) - 1; + static constexpr std::size_t kBuckets = kExactBelow + (kLargestPower - kSplitBits) * kSplit; + + static std::size_t BucketOf(std::uint64_t value) noexcept + { + if (value < kExactBelow) + { + return static_cast(value); + } + const int power = std::bit_width(value) - 1; // 6 or more + const int shift = power - kSplitBits; + const std::uint64_t withinPower = (value >> shift) - kSplit; // 0 to 31 + return static_cast(kExactBelow + static_cast(power - kSplitBits - 1) * kSplit + + withinPower); + } + + static std::uint64_t LargestValueOf(std::size_t bucket) noexcept + { + if (bucket < kExactBelow) + { + return bucket; + } + const std::uint64_t above = bucket - kExactBelow; + const int power = static_cast(above / kSplit) + kSplitBits + 1; + const int shift = power - kSplitBits; + return ((kSplit + above % kSplit + 1) << shift) - 1; + } + + std::array m_buckets{}; + std::uint64_t m_count = 0; + std::uint64_t m_max = 0; +}; + +} // namespace perf diff --git a/performance_tests/writer_performance_test/plan.cpp b/performance_tests/writer_performance_test/plan.cpp new file mode 100644 index 0000000..55b2703 --- /dev/null +++ b/performance_tests/writer_performance_test/plan.cpp @@ -0,0 +1,215 @@ +#include "plan.h" + +#include "receiver.h" + +#include +#include +#include +#include + +namespace perf { + +namespace { + +// No line that this test records is longer. The library sends a buffer once it has reached the buffer size, so a +// datagram can be this much larger than the buffer size. +constexpr std::size_t kLongestLine = 128; + +// What every measurement takes on top of the warm-up and the measurement: starting the process, and its exit. +constexpr double kOverheadSeconds = 0.3; + +} // namespace + +Selection FullSelection() +{ + Selection selection; + selection.seconds = 2.0; + selection.pacedSeconds = 3.0; + selection.warmUp = 0.5; + selection.threads = {1, 2, 4, 8, 16}; + selection.everyCombination = true; + selection.pacedThreads = {1, 4, 16}; + selection.pacedBuffers = {512, kUsualBuffer}; + selection.rates = {1000, 50000}; + selection.meterThreads = {1, 4, 16}; + selection.meterWriters = {WriterKind::Udp, WriterKind::Unix}; + return selection; +} + +Selection SmokeSelection() +{ + Selection selection; + selection.seconds = 0.2; + selection.pacedSeconds = 0.3; + selection.warmUp = 0.1; + selection.repeats = 1; + selection.threads = {}; + selection.buffers = {512}; + selection.pacedThreads = {2}; + selection.pacedBuffers = {512}; + selection.rates = {2000}; + selection.meterThreads = {}; + return selection; +} + +Plan MakePlan(const Selection& selection) +{ + Plan plan; + std::set planned; + const auto add = [&](const Scenario& scenario) + { + if (planned.insert(scenario.Key()).second) + { + plan.scenarios.push_back(scenario); + } + }; + + for (const WriterKind writer : selection.writers) + { + // Whether this system sends datagrams of a buffer of that size on a socket of this kind. + std::map known; + const auto carries = [&](unsigned int buffer) + { + if (buffer == 0) + { + return true; + } + const auto found = known.find(buffer); + if (found != known.end()) + { + return found->second; + } + const bool fits = Receiver::Carries(writer, buffer + kLongestLine); + known[buffer] = fits; + if (!fits) + { + plan.leftOut.push_back(WriterName(writer) + " with a buffer of " + std::to_string(buffer) + + " bytes: this system does not send datagrams of that size on such a socket"); + } + return fits; + }; + + // The buffer size at which the thread counts and the meters are compared: the usual one, or else the + // largest of the sizes below it that fits, or else the first size that fits. + unsigned int usual = 0; + for (const unsigned int buffer : selection.buffers) + { + if (buffer != 0 && buffer <= kUsualBuffer && buffer > usual && carries(buffer)) + { + usual = buffer; + } + } + for (const unsigned int buffer : selection.buffers) + { + if (usual == 0 && buffer != 0 && carries(buffer)) + { + usual = buffer; + } + } + + // Flat out, with a counter that the thread created once and that has tags. + for (const unsigned int buffer : selection.buffers) + { + if (!carries(buffer)) + { + continue; + } + Scenario scenario; + scenario.writer = writer; + scenario.bufferSize = buffer; + if (buffer == 0) + { + // Without buffering the calling thread sends, and the library does not allow a second one. + if (!selection.threads.empty()) + { + add(scenario); + } + continue; + } + for (const int threads : selection.threads) + { + if (selection.everyCombination || buffer == usual || threads == selection.bufferThreads) + { + scenario.threads = threads; + add(scenario); + } + } + } + + // Paced. + for (const unsigned int wanted : selection.pacedBuffers) + { + const unsigned int buffer = wanted == kUsualBuffer ? usual : wanted; + if (buffer == 0 || !carries(buffer)) + { + continue; + } + for (const int threads : selection.pacedThreads) + { + for (const int rate : selection.rates) + { + Scenario scenario; + scenario.writer = writer; + scenario.bufferSize = buffer; + scenario.threads = threads; + scenario.load = Load::Paced; + scenario.linesPerSecond = rate; + add(scenario); + } + } + } + + // The ways to use a meter. + const bool compareMeters = std::find(selection.meterWriters.begin(), selection.meterWriters.end(), writer) != + selection.meterWriters.end(); + if (compareMeters && usual != 0) + { + for (const int threads : selection.meterThreads) + { + for (const bool createMeter : {false, true}) + { + for (const bool tags : {true, false}) + { + Scenario scenario; + scenario.writer = writer; + scenario.bufferSize = usual; + scenario.threads = threads; + scenario.createMeter = createMeter; + scenario.tags = tags; + add(scenario); + } + } + } + } + } + return plan; +} + +double MeasuredSeconds(const Scenario& scenario, const Selection& selection) +{ + return scenario.load == Load::Paced ? selection.pacedSeconds : selection.seconds; +} + +double EstimatedSeconds(const Plan& plan, const Selection& selection) +{ + double seconds = 0; + for (const Scenario& scenario : plan.scenarios) + { + seconds += selection.warmUp + MeasuredSeconds(scenario, selection) + kOverheadSeconds; + } + return seconds * selection.repeats; +} + +std::string Describe(const Scenario& scenario) +{ + std::string text = WriterName(scenario.writer) + ", buffer " + + (scenario.bufferSize == 0 ? std::string("none") : std::to_string(scenario.bufferSize)) + ", " + + std::to_string(scenario.threads) + (scenario.threads == 1 ? " thread, " : " threads, "); + text += scenario.load == Load::Paced ? "paced at " + std::to_string(scenario.linesPerSecond) + " lines/s" + : std::string("flat out"); + text += scenario.createMeter ? ", counter created per line" : ""; + text += scenario.tags ? "" : ", no tags"; + return text; +} + +} // namespace perf diff --git a/performance_tests/writer_performance_test/plan.h b/performance_tests/writer_performance_test/plan.h new file mode 100644 index 0000000..01a4467 --- /dev/null +++ b/performance_tests/writer_performance_test/plan.h @@ -0,0 +1,67 @@ +#pragma once + +#include "scenario.h" + +#include +#include + +namespace perf { + +// The buffer size of the existing performance test. The thread counts and the ways to use a meter are compared at +// this size. +inline constexpr unsigned int kUsualBuffer = 4096; + +// Which scenarios a run consists of, and how long each is measured. The default is the short run. +struct Selection +{ + double seconds = 1.0; // how long a scenario is measured under a flat-out load + double pacedSeconds = 2.0; // the same under a paced load, where fewer lines make up the percentiles + double warmUp = 0.3; // how long the threads record before the measurement begins + int repeats = 3; // how often every scenario is measured + + std::vector writers{WriterKind::Udp, WriterKind::Unix}; + + // Flat out. Buffer size 0 is no buffering, which the library allows from one thread only. + std::vector threads{1, 4, 8}; + std::vector buffers{0, 512, kUsualBuffer, 32768}; + // Whether every buffer size is measured with every number of threads. Otherwise the numbers of threads are + // measured at the usual buffer size, and the buffer sizes with `bufferThreads` threads. + bool everyCombination = false; + int bufferThreads = 4; + + // Paced. + std::vector pacedThreads{1, 4}; + std::vector pacedBuffers{kUsualBuffer}; + std::vector rates{20000}; // lines per second and thread + + // The ways to use a meter, compared flat out at the usual buffer size. + std::vector meterThreads{1, 4}; + std::vector meterWriters{WriterKind::Unix}; +}; + +// More thread counts, buffer sizes and rates, measured for longer. +Selection FullSelection(); + +// A few seconds, to see that the test itself works: two scenarios at a rate at which no line may be lost. +Selection SmokeSelection(); + +// The scenarios of a run, and what was left out of it and why. +struct Plan +{ + std::vector scenarios; + std::vector leftOut; +}; + +// The scenarios of a selection. Buffer sizes that this system's sockets cannot carry are left out. +Plan MakePlan(const Selection& selection); + +// How long the scenario is measured. +double MeasuredSeconds(const Scenario& scenario, const Selection& selection); + +// How long the whole run takes, roughly. +double EstimatedSeconds(const Plan& plan, const Selection& selection); + +// The scenario in words, for example: unix, buffer 4096, 4 threads, flat out. +std::string Describe(const Scenario& scenario); + +} // namespace perf diff --git a/performance_tests/writer_performance_test/receiver.cpp b/performance_tests/writer_performance_test/receiver.cpp new file mode 100644 index 0000000..11d742b --- /dev/null +++ b/performance_tests/writer_performance_test/receiver.cpp @@ -0,0 +1,257 @@ +#include "receiver.h" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace perf { + +namespace { + +// The receive buffer that is asked for, so that the receiver is not what limits the writer. The system may give +// less. +constexpr int kReceiveBufferBytes = 8 * 1024 * 1024; + +// Larger than any datagram. +constexpr std::size_t kLargestDatagram = 1024 * 1024; + +// Once the sender has gone, the receiver stops when the socket has stayed empty for this long. +constexpr int kQuietMillis = 50; + +std::system_error Failure(const std::string& what) +{ + return std::system_error(errno, std::generic_category(), what); +} + +// A new datagram socket, which the recording process does not inherit. +int OpenSocket(int family) +{ + const int descriptor = ::socket(family, SOCK_DGRAM, 0); + if (descriptor < 0) + { + throw Failure("Could not create a socket"); + } + ::fcntl(descriptor, F_SETFD, FD_CLOEXEC); + return descriptor; +} + +sockaddr_un UnixAddress(const std::string& path) +{ + sockaddr_un address{}; + address.sun_family = AF_UNIX; + std::strncpy(address.sun_path, path.c_str(), sizeof(address.sun_path) - 1); + return address; +} + +// A receiving socket on this machine. +struct Bound +{ + int socket = -1; + std::string location; + std::string socketPath; + sockaddr_in udpAddress{}; +}; + +Bound Bind(WriterKind kind) +{ + Bound bound; + if (kind == WriterKind::Udp) + { + bound.socket = OpenSocket(AF_INET); + bound.udpAddress.sin_family = AF_INET; + bound.udpAddress.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + bound.udpAddress.sin_port = 0; // any port that is free + socklen_t length = sizeof(bound.udpAddress); + if (::bind(bound.socket, reinterpret_cast(&bound.udpAddress), length) != 0 || + ::getsockname(bound.socket, reinterpret_cast(&bound.udpAddress), &length) != 0) + { + const std::system_error failure = Failure("Could not bind a UDP socket on 127.0.0.1"); + ::close(bound.socket); + throw failure; + } + bound.location = "udp://127.0.0.1:" + std::to_string(ntohs(bound.udpAddress.sin_port)); + } + else + { + static std::atomic sockets{0}; + bound.socket = OpenSocket(AF_UNIX); + bound.socketPath = "/tmp/spectator_perf_" + std::to_string(::getpid()) + "_" + + std::to_string(sockets.fetch_add(1)) + ".sock"; + ::unlink(bound.socketPath.c_str()); + const sockaddr_un address = UnixAddress(bound.socketPath); + if (::bind(bound.socket, reinterpret_cast(&address), sizeof(address)) != 0) + { + const std::system_error failure = Failure("Could not bind a Unix socket at " + bound.socketPath); + ::close(bound.socket); + throw failure; + } + bound.location = "unix://" + bound.socketPath; + } + + const int size = kReceiveBufferBytes; + ::setsockopt(bound.socket, SOL_SOCKET, SO_RCVBUF, &size, sizeof(size)); + return bound; +} + +void Release(const Bound& bound) +{ + ::close(bound.socket); + if (!bound.socketPath.empty()) + { + ::unlink(bound.socketPath.c_str()); + } +} + +// The same clock, read the same way, as the recording threads use for the value of a line. +std::uint64_t MicrosNow() +{ + return static_cast(std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count()); +} + +} // namespace + +Receiver::Receiver(WriterKind kind) +{ + const Bound bound = Bind(kind); + m_socket = bound.socket; + m_location = bound.location; + m_socketPath = bound.socketPath; +} + +Receiver::~Receiver() +{ + Stop(); + ::close(m_socket); + if (!m_socketPath.empty()) + { + ::unlink(m_socketPath.c_str()); + } +} + +void Receiver::Start(bool timestamps) +{ + m_timestamps = timestamps; + m_thread = std::thread(&Receiver::Read, this); +} + +Received Receiver::Stop() +{ + m_stopping.store(true); + if (m_thread.joinable()) + { + m_thread.join(); + } + return m_received; +} + +void Receiver::Read() +{ + std::vector datagram(kLargestDatagram); + pollfd waiting{m_socket, POLLIN, 0}; + for (;;) + { + // Looked at before the socket is read, so that nothing that arrived before Stop() is left behind. + const bool stopping = m_stopping.load(); + const ssize_t size = ::recv(m_socket, datagram.data(), datagram.size(), MSG_DONTWAIT); + if (size < 0) + { + if (errno == EINTR) + { + continue; + } + if (errno != EAGAIN && errno != EWOULDBLOCK) + { + return; + } + // The socket is empty. While the sender runs, more will come. Once it has gone, what it sent last may + // still take a moment to get here. + const int ready = ::poll(&waiting, 1, stopping ? kQuietMillis : 20); + if (ready == 0 && stopping) + { + return; + } + continue; + } + + const auto bytes = static_cast(size); + ++m_received.datagrams; + m_received.bytes += bytes; + m_received.lines += static_cast(std::count(datagram.data(), datagram.data() + bytes, '\n')); + if (m_timestamps) + { + KeepDelays(datagram.data(), bytes); + } + } +} + +void Receiver::KeepDelays(const char* datagram, std::size_t size) +{ + const std::uint64_t now = MicrosNow(); + const char* const end = datagram + size; + for (const char* line = datagram; line < end;) + { + const void* const newline = std::memchr(line, '\n', static_cast(end - line)); + const char* const lineEnd = newline != nullptr ? static_cast(newline) : end; + + // A line is "c::", and the whole part of the value is the time of recording in microseconds. + const char* value = lineEnd; + while (value > line && value[-1] != ':') + { + --value; + } + std::uint64_t recordedAt = 0; + for (; value < lineEnd && *value >= '0' && *value <= '9'; ++value) + { + recordedAt = recordedAt * 10 + static_cast(*value - '0'); + } + m_received.delayMicros.Record(now > recordedAt ? now - recordedAt : 0); + + line = lineEnd + 1; + } +} + +bool Receiver::Carries(WriterKind kind, std::size_t bytes) +{ + const Bound receiver = Bind(kind); + const std::vector datagram(bytes, 'x'); + ssize_t sent = -1; + try + { + // A socket as the library opens it, with the system's default limits. + const int sender = OpenSocket(kind == WriterKind::Udp ? AF_INET : AF_UNIX); + if (kind == WriterKind::Udp) + { + sent = ::sendto(sender, datagram.data(), datagram.size(), 0, + reinterpret_cast(&receiver.udpAddress), sizeof(receiver.udpAddress)); + } + else + { + const sockaddr_un address = UnixAddress(receiver.socketPath); + sent = ::sendto(sender, datagram.data(), datagram.size(), 0, reinterpret_cast(&address), + sizeof(address)); + } + ::close(sender); + } + catch (const std::system_error&) + { + Release(receiver); + throw; + } + Release(receiver); + return sent == static_cast(bytes); +} + +} // namespace perf diff --git a/performance_tests/writer_performance_test/receiver.h b/performance_tests/writer_performance_test/receiver.h new file mode 100644 index 0000000..ef67c4f --- /dev/null +++ b/performance_tests/writer_performance_test/receiver.h @@ -0,0 +1,63 @@ +#pragma once + +#include "histogram.h" +#include "scenario.h" + +#include +#include +#include +#include +#include + +namespace perf { + +// What arrived at a Receiver. +struct Received +{ + std::uint64_t datagrams = 0; + std::uint64_t bytes = 0; + std::uint64_t lines = 0; + // With timestamps: how long each line took from being recorded to being read here, in microseconds. + Histogram delayMicros; +}; + +// Stands in for spectatord: a socket on this machine for the library to send to, and a thread that reads it and +// counts what arrives. The test knows what was recorded, so with this it also knows what was lost on the way. +class Receiver +{ + public: + explicit Receiver(WriterKind kind); + ~Receiver(); + + Receiver(const Receiver&) = delete; + Receiver& operator=(const Receiver&) = delete; + + // Where the socket is, in the form that WriterConfig takes: udp://127.0.0.1: or unix://. + const std::string& Location() const { return m_location; } + + // Starts reading. With `timestamps`, the value of every line is taken to be the time at which the line was + // recorded, as the paced threads in recording_run.cpp set it, and the delay of every line is kept. + void Start(bool timestamps); + + // To be called when the sending process has exited: reads what is still on its way, stops reading and returns + // what arrived. + Received Stop(); + + // Whether a socket like the library's can send a datagram of `bytes` to a socket of this kind on this machine. + // The limit differs between systems: macOS, for one, takes 2048 bytes on a Unix datagram socket by default. + static bool Carries(WriterKind kind, std::size_t bytes); + + private: + void Read(); + void KeepDelays(const char* datagram, std::size_t size); + + int m_socket = -1; + std::string m_location; + std::string m_socketPath; // of a Unix socket, to remove it again + bool m_timestamps = false; + std::atomic m_stopping{false}; + std::thread m_thread; + Received m_received; // the reading thread's, until it has been joined +}; + +} // namespace perf diff --git a/performance_tests/writer_performance_test/recording_run.cpp b/performance_tests/writer_performance_test/recording_run.cpp new file mode 100644 index 0000000..d8df189 --- /dev/null +++ b/performance_tests/writer_performance_test/recording_run.cpp @@ -0,0 +1,314 @@ +#include "recording_run.h" + +#include "histogram.h" + +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace perf { + +namespace { + +using Clock = std::chrono::steady_clock; + +// What the recording threads are to do. Only the main thread changes it, so reading it costs the threads nothing. +enum Phase : int +{ + kWaiting, // until every thread is ready + kWarmingUp, // recording, not counted + kMeasuring, // recording, counted + kStopping +}; + +// Flat out, one call in this many is timed. Reading the clock twice takes about as long as the call itself, so +// timing every call would slow the threads down and change what is measured. +constexpr std::uint64_t kTimedCallEvery = 32; + +// How often a paced thread catches up with its rate. +constexpr std::chrono::milliseconds kPaceStep{1}; + +const std::string kCounterName = "perf_test_counter"; + +std::uint64_t Nanos(Clock::duration duration) +{ + return static_cast(std::chrono::duration_cast(duration).count()); +} + +// A time as the value of a line: microseconds on the steady clock, which every process on the machine reads the +// same. The receiver takes the delay of the line from it. +double Micros(Clock::time_point time) +{ + return static_cast(std::chrono::duration_cast(time.time_since_epoch()).count()); +} + +// What one thread did. Aligned so that no two threads write to the same cache line. +struct alignas(128) ThreadResult +{ + std::uint64_t calls = 0; // all calls, warm-up included + std::uint64_t measuredCalls = 0; // the calls in the measurement + double measuredSeconds = 0; // how long the thread was in the measurement + Histogram callNanos; // how long the timed calls in the measurement took +}; + +// Records as fast as the thread can. `recordLine` records one line with the given value. +template +void RecordFlatOut(const std::atomic& phase, const RecordLine& recordLine, ThreadResult& result) +{ + // Everything the loop writes is local to the thread. The result is filled in afterwards. + Histogram callNanos; + std::uint64_t calls = 0; + std::uint64_t measuredCalls = 0; + bool measuring = false; + Clock::time_point measuringSince; + + for (;;) + { + const int current = phase.load(std::memory_order_relaxed); + if (current == kStopping) + { + break; + } + if (current == kMeasuring && !measuring) + { + measuring = true; + measuringSince = Clock::now(); + } + + if (calls % kTimedCallEvery == 0) + { + const Clock::time_point before = Clock::now(); + recordLine(1.0); + const Clock::time_point after = Clock::now(); + if (measuring) + { + callNanos.Record(Nanos(after - before)); + } + } + else + { + recordLine(1.0); + } + ++calls; + measuredCalls += measuring ? 1 : 0; + } + + result.calls = calls; + result.measuredCalls = measuredCalls; + result.measuredSeconds = measuring ? std::chrono::duration(Clock::now() - measuringSince).count() : 0.0; + result.callNanos = callNanos; +} + +// Records `linesPerSecond` lines per second: every kPaceStep, the thread records the lines that are due by then. The +// value of each line is the time at which it is recorded. Every call is timed. +template +void RecordPaced(const std::atomic& phase, int linesPerSecond, const RecordLine& recordLine, + ThreadResult& result) +{ + Histogram callNanos; + std::uint64_t calls = 0; + std::uint64_t measuredCalls = 0; + bool measuring = false; + Clock::time_point measuringSince; + const Clock::time_point start = Clock::now(); + + for (;;) + { + const int current = phase.load(std::memory_order_relaxed); + if (current == kStopping) + { + break; + } + if (current == kMeasuring && !measuring) + { + measuring = true; + measuringSince = Clock::now(); + } + + const double elapsed = std::chrono::duration(Clock::now() - start).count(); + const auto due = static_cast(elapsed * linesPerSecond); + while (calls < due) + { + const Clock::time_point before = Clock::now(); + recordLine(Micros(before)); + const Clock::time_point after = Clock::now(); + if (measuring) + { + callNanos.Record(Nanos(after - before)); + ++measuredCalls; + } + ++calls; + } + std::this_thread::sleep_for(kPaceStep); + } + + result.calls = calls; + result.measuredCalls = measuredCalls; + result.measuredSeconds = measuring ? std::chrono::duration(Clock::now() - measuringSince).count() : 0.0; + result.callNanos = callNanos; +} + +// One recording thread. +void Record(const Scenario& scenario, const spectator::Registry& registry, const std::atomic& phase, + std::atomic& ready, ThreadResult& result) +{ + // The tags of the existing performance test. + const std::unordered_map tags{{"location", WriterName(scenario.writer)}, + {"version", "correct-horse-battery-staple"}}; + + const auto run = [&](const auto& recordLine) + { + ready.fetch_add(1); + while (phase.load(std::memory_order_acquire) == kWaiting) + { + std::this_thread::sleep_for(std::chrono::microseconds(100)); + } + if (scenario.load == Load::Paced) + { + RecordPaced(phase, scenario.linesPerSecond, recordLine, result); + } + else + { + RecordFlatOut(phase, recordLine, result); + } + }; + + if (!scenario.createMeter) + { + // The counter exists before the measurement begins, so the loop is the call that records and nothing else. + const spectator::Counter counter = + scenario.tags ? registry.CreateCounter(kCounterName, tags) : registry.CreateCounter(kCounterName); + run([&counter](double value) { counter.Increment(value); }); + } + else if (scenario.tags) + { + // What a caller does that writes registry.CreateCounter(name, tags).Increment() wherever it counts. + run([®istry, &tags](double value) { registry.CreateCounter(kCounterName, tags).Increment(value); }); + } + else + { + run([®istry](double value) { registry.CreateCounter(kCounterName).Increment(value); }); + } +} + +// How long reading the clock takes, in nanoseconds: what a timed call is measured too long by. It is the average +// over many readings, because one reading can be shorter than the steps in which the clock advances. +double ClockNanos() +{ + constexpr int kReadings = 100000; + const Clock::time_point begin = Clock::now(); + Clock::time_point last = begin; + for (int i = 0; i < kReadings; ++i) + { + last = Clock::now(); + } + return static_cast(Nanos(last - begin)) / kReadings; +} + +// The processor time that this process has used, in seconds, on all its threads. +double ProcessorSeconds() +{ + rusage usage{}; + getrusage(RUSAGE_SELF, &usage); + const auto seconds = [](const timeval& time) + { return static_cast(time.tv_sec) + static_cast(time.tv_usec) / 1e6; }; + return seconds(usage.ru_utime) + seconds(usage.ru_stime); +} + +} // namespace + +int RunRecording(const Arguments& arguments) +try +{ + const Scenario scenario = FromArguments(arguments); + const std::string& location = Text(arguments, "location"); + const std::chrono::duration warmUp(Number(arguments, "warmup")); + const std::chrono::duration measurement(Number(arguments, "seconds")); + + // The lines go where this test listens, whatever the environment says. + unsetenv("SPECTATOR_OUTPUT_LOCATION"); + // What the library logs is not part of what is measured, and a send that fails shows in the lines that arrive. + spectator::SetLogLevel(spdlog::level::off); + + // No tags from the environment, so that the lines are the same on every machine. + const spectator::WriterConfig writerConfig = scenario.bufferSize == 0 + ? spectator::WriterConfig(location) + : spectator::WriterConfig(location, scenario.bufferSize); + const spectator::Config config(writerConfig, {}, false); + const spectator::Registry registry(config); + + const double clockNanos = ClockNanos(); + + std::atomic phase{kWaiting}; + std::atomic ready{0}; + std::vector results(static_cast(scenario.threads)); + std::vector threads; + for (ThreadResult& result : results) + { + threads.emplace_back(Record, std::cref(scenario), std::cref(registry), std::cref(phase), std::ref(ready), + std::ref(result)); + } + while (ready.load() < scenario.threads) + { + std::this_thread::sleep_for(std::chrono::microseconds(100)); + } + + const Clock::time_point start = Clock::now(); + phase.store(kWarmingUp); + std::this_thread::sleep_for(warmUp); + phase.store(kMeasuring); + std::this_thread::sleep_for(measurement); + phase.store(kStopping); + for (std::thread& thread : threads) + { + thread.join(); + } + const double seconds = std::chrono::duration(Clock::now() - start).count(); + + // Each thread's rate over the time that the thread itself measured, so a thread that was held up at either end + // of the measurement is neither cut short nor counted for too long. + std::uint64_t recorded = 0; + std::uint64_t measured = 0; + double linesPerSecond = 0; + Histogram callNanos; + for (const ThreadResult& result : results) + { + recorded += result.calls; + measured += result.measuredCalls; + if (result.measuredSeconds > 0) + { + linesPerSecond += static_cast(result.measuredCalls) / result.measuredSeconds; + } + callNanos.Merge(result.callNanos); + } + + std::printf("%srecorded=%llu measured=%llu rate=%.1f call_p50=%llu call_p99=%llu call_p999=%llu call_max=%llu " + "timed=%llu seconds=%.4f cpu=%.4f clock=%.1f\n", + kResultPrefix, static_cast(recorded), static_cast(measured), + linesPerSecond, static_cast(callNanos.Percentile(0.5)), + static_cast(callNanos.Percentile(0.99)), + static_cast(callNanos.Percentile(0.999)), + static_cast(callNanos.Max()), static_cast(callNanos.Count()), + seconds, ProcessorSeconds(), clockNanos); + std::fflush(stdout); + return 0; +} +catch (const std::exception& error) +{ + std::fprintf(stderr, "The recording process failed: %s\n", error.what()); + return 2; +} + +} // namespace perf diff --git a/performance_tests/writer_performance_test/recording_run.h b/performance_tests/writer_performance_test/recording_run.h new file mode 100644 index 0000000..b9b6e01 --- /dev/null +++ b/performance_tests/writer_performance_test/recording_run.h @@ -0,0 +1,19 @@ +#pragma once + +#include "scenario.h" + +namespace perf { + +// The line that RunRecording() prints starts with this. The rest of it is settings in the form name=value. +inline constexpr const char* kResultPrefix = "PERF_RESULT "; + +// The recording side of one measurement. It runs in a process of its own, because the library's Writer is one per +// process: the process creates a Registry for the scenario in `arguments`, which sends to the location given there, +// lets the scenario's threads record during the warm-up and the measurement, and prints what it measured as one +// line. Returns the exit code of the process. +// +// Lines that are still buffered when this returns are sent by the library as the process exits, so a normal exit is +// part of every measurement. +int RunRecording(const Arguments& arguments); + +} // namespace perf diff --git a/performance_tests/writer_performance_test/report.cpp b/performance_tests/writer_performance_test/report.cpp new file mode 100644 index 0000000..0643f18 --- /dev/null +++ b/performance_tests/writer_performance_test/report.cpp @@ -0,0 +1,510 @@ +#include "report.h" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace perf { + +namespace { + +double Median(std::vector values) +{ + if (values.empty()) + { + return 0; + } + std::sort(values.begin(), values.end()); + const std::size_t middle = values.size() / 2; + return values.size() % 2 == 1 ? values[middle] : (values[middle - 1] + values[middle]) / 2; +} + +// The measurements of one scenario, taken together. +struct Summary +{ + int measurements = 0; // that are ok + double linesPerSecond = 0; // the median + double spread = 0; // from the slowest to the fastest measurement, as a share of the median + double callP50 = 0; // the medians of the measurements' percentiles + double callP99 = 0; + double callP999 = 0; + double callMax = 0; // the longest call in any measurement + double cpuNanosPerLine = 0; + double receivedShare = 0; + bool allReceived = false; + double linesPerDatagram = 0; + double bytesPerLine = 0; + double delayP50 = 0; + double delayP99 = 0; + double delayMax = 0; +}; + +Summary Summarize(const std::vector& measurements) +{ + Summary summary; + std::vector rates, p50, p99, p999, cpu, delayP50, delayP99; + std::uint64_t recorded = 0, received = 0, datagrams = 0, bytes = 0; + for (const Measurement& measurement : measurements) + { + if (!measurement.ok) + { + continue; + } + ++summary.measurements; + rates.push_back(measurement.linesPerSecond); + p50.push_back(measurement.callP50); + p99.push_back(measurement.callP99); + p999.push_back(measurement.callP999); + cpu.push_back(measurement.cpuNanosPerLine); + delayP50.push_back(measurement.delayP50); + delayP99.push_back(measurement.delayP99); + summary.callMax = std::max(summary.callMax, measurement.callMax); + summary.delayMax = std::max(summary.delayMax, measurement.delayMax); + recorded += measurement.recorded; + received += measurement.received; + datagrams += measurement.datagrams; + bytes += measurement.bytes; + } + if (summary.measurements == 0) + { + return summary; + } + + summary.linesPerSecond = Median(rates); + const auto [slowest, fastest] = std::minmax_element(rates.begin(), rates.end()); + summary.spread = summary.linesPerSecond > 0 ? (*fastest - *slowest) / summary.linesPerSecond : 0; + summary.callP50 = Median(p50); + summary.callP99 = Median(p99); + summary.callP999 = Median(p999); + summary.cpuNanosPerLine = Median(cpu); + summary.delayP50 = Median(delayP50); + summary.delayP99 = Median(delayP99); + summary.receivedShare = recorded == 0 ? 0 : static_cast(received) / static_cast(recorded); + summary.allReceived = received == recorded; + summary.linesPerDatagram = datagrams == 0 ? 0 : static_cast(received) / static_cast(datagrams); + summary.bytesPerLine = received == 0 ? 0 : static_cast(bytes) / static_cast(received); + return summary; +} + +std::string Format(const char* format, double value) +{ + char text[64]; + std::snprintf(text, sizeof(text), format, value); + return text; +} + +std::string Time(double nanos) +{ + if (nanos < 1e3) + { + return Format("%.0f ns", nanos); + } + if (nanos < 1e6) + { + return Format("%.1f us", nanos / 1e3); + } + if (nanos < 1e9) + { + return Format("%.1f ms", nanos / 1e6); + } + return Format("%.2f s", nanos / 1e9); +} + +// The share of the lines that arrived. It is never rounded up to 100%: that is shown only if every line arrived. +std::string Arrived(const Summary& summary) +{ + if (summary.allReceived) + { + return "100%"; + } + return Format("%.2f%%", std::min(99.99, std::floor(summary.receivedShare * 10000) / 100)); +} + +std::string Buffer(unsigned int bufferSize) { return bufferSize == 0 ? "none" : std::to_string(bufferSize); } + +std::vector Split(const std::string& text, char separator) +{ + std::vector parts; + std::stringstream stream(text); + for (std::string part; std::getline(stream, part, separator);) + { + parts.push_back(part); + } + return parts; +} + +// A table whose columns are as wide as their widest entry. The first `textColumns` columns are aligned to the left, +// the numbers after them to the right. +class Table +{ + public: + Table(std::vector header, std::size_t textColumns) + : m_rows{std::move(header)}, m_textColumns(textColumns) + { + } + + void Add(std::vector row) { m_rows.push_back(std::move(row)); } + + bool Empty() const { return m_rows.size() == 1; } + + void Print() const + { + std::vector widths(m_rows.front().size(), 0); + for (const std::vector& row : m_rows) + { + for (std::size_t column = 0; column < row.size(); ++column) + { + widths[column] = std::max(widths[column], row[column].size()); + } + } + for (std::size_t rowIndex = 0; rowIndex < m_rows.size(); ++rowIndex) + { + std::string line = " "; + for (std::size_t column = 0; column < m_rows[rowIndex].size(); ++column) + { + const std::string& entry = m_rows[rowIndex][column]; + const std::string padding(widths[column] - entry.size(), ' '); + line += " " + (column < m_textColumns ? entry + padding : padding + entry) + " "; + } + std::printf("%s\n", line.c_str()); + if (rowIndex == 0) + { + std::string rule = " "; + for (const std::size_t width : widths) + { + rule += " " + std::string(width, '-') + " "; + } + std::printf("%s\n", rule.c_str()); + } + } + } + + private: + std::vector> m_rows; + std::size_t m_textColumns; +}; + +// A counter that the thread created once and that has tags: what the scenarios use that are not about meters. +bool UsualMeter(const Scenario& scenario) { return !scenario.createMeter && scenario.tags; } + +} // namespace + +std::string Rate(double perSecond) +{ + if (perSecond >= 1e6) + { + return Format("%.2f M", perSecond / 1e6); + } + if (perSecond >= 1e3) + { + return Format("%.1f k", perSecond / 1e3); + } + return Format("%.0f", perSecond); +} + +void PrintHeader(const std::string& label, const Selection& selection, const Plan& plan) +{ + utsname system{}; + ::uname(&system); +#if defined(__OPTIMIZE__) + const bool optimized = true; +#else + const bool optimized = false; +#endif + + std::printf("Writer performance test%s\n", label.empty() ? "" : (" (" + label + ")").c_str()); + std::printf(" Machine: %u hardware threads, %s %s, %s\n", std::thread::hardware_concurrency(), system.sysname, + system.release, system.machine); + std::printf(" Compiler: %s, %s\n", __VERSION__, optimized ? "optimized" : "NOT OPTIMIZED"); + std::printf(" Run: %zu configurations, each measured %d x %.1f s (paced: %.1f s) after %.1f s of warm-up:" + " about %.0f s\n", + plan.scenarios.size(), selection.repeats, selection.seconds, selection.pacedSeconds, + selection.warmUp, EstimatedSeconds(plan, selection)); + if (!optimized) + { + std::printf("\n WARNING: this program was built without optimization, so its numbers say little about a" + "\n release build. performance_tests/writer_performance_test/README.md says how to build one.\n"); + } + std::printf("\n"); +} + +void PrintReport(const Plan& plan, const Measurements& measurements, const std::map& baseline) +{ + const std::vector& scenarios = plan.scenarios; + std::vector summaries; + double clockNanos = 0; + int failed = 0; + for (const std::vector& ofScenario : measurements) + { + summaries.push_back(Summarize(ofScenario)); + for (const Measurement& measurement : ofScenario) + { + failed += measurement.ok ? 0 : 1; + clockNanos = std::max(clockNanos, measurement.clockNanos); + } + } + + const bool compare = !baseline.empty(); + const auto againstBaseline = [&](const Scenario& scenario, const Summary& summary) -> std::string + { + const auto found = baseline.find(scenario.Key()); + if (found == baseline.end() || found->second <= 0) + { + return "-"; + } + return Format("%.2fx", summary.linesPerSecond / found->second); + }; + // The rate of the same scenario with one thread, or zero if that was not measured. + const auto withOneThread = [&](Scenario scenario) -> double + { + scenario.threads = 1; + for (std::size_t i = 0; i < scenarios.size(); ++i) + { + if (scenarios[i].Key() == scenario.Key() && summaries[i].measurements > 0) + { + return summaries[i].linesPerSecond; + } + } + return 0; + }; + + // Flat out. + std::vector header{"writer", "buffer", "threads", "lines/s", "per thread", "speed-up", "spread", + "call p50", "p99", "longest", "cpu/line", "arrived", "lines/datagram"}; + if (compare) + { + header.push_back("vs baseline"); + } + Table flatOut(header, 2); + for (std::size_t i = 0; i < scenarios.size(); ++i) + { + const Scenario& scenario = scenarios[i]; + const Summary& summary = summaries[i]; + if (scenario.load != Load::FlatOut || !UsualMeter(scenario) || summary.measurements == 0) + { + continue; + } + const double single = withOneThread(scenario); + std::vector row{WriterName(scenario.writer), + Buffer(scenario.bufferSize), + std::to_string(scenario.threads), + Rate(summary.linesPerSecond), + Rate(summary.linesPerSecond / scenario.threads), + single > 0 ? Format("%.2fx", summary.linesPerSecond / single) : "-", + Format("%.0f%%", summary.spread * 100), + Time(summary.callP50), + Time(summary.callP99), + Time(summary.callMax), + Time(summary.cpuNanosPerLine), + Arrived(summary), + Format("%.1f", summary.linesPerDatagram)}; + if (compare) + { + row.push_back(againstBaseline(scenario, summary)); + } + flatOut.Add(row); + } + if (!flatOut.Empty()) + { + std::printf("Flat out: every thread records as fast as it can, with a counter that it created once.\n\n"); + flatOut.Print(); + std::printf("\n"); + } + + // Paced. + Table paced({"writer", "buffer", "threads", "lines/s per thread", "call p50", "p99", "p99.9", "longest", + "delay p50", "delay p99", "longest delay", "arrived"}, + 2); + for (std::size_t i = 0; i < scenarios.size(); ++i) + { + const Scenario& scenario = scenarios[i]; + const Summary& summary = summaries[i]; + if (scenario.load != Load::Paced || summary.measurements == 0) + { + continue; + } + paced.Add({WriterName(scenario.writer), Buffer(scenario.bufferSize), std::to_string(scenario.threads), + std::to_string(scenario.linesPerSecond), Time(summary.callP50), Time(summary.callP99), + Time(summary.callP999), Time(summary.callMax), Time(summary.delayP50 * 1e3), + Time(summary.delayP99 * 1e3), Time(summary.delayMax * 1e3), Arrived(summary)}); + } + if (!paced.Empty()) + { + std::printf("Paced: every thread records a fixed number of lines per second. The delay is the time from\n" + "recording a line to its arrival.\n\n"); + paced.Print(); + std::printf("\n"); + } + + // The ways to use a meter: the scenarios that were also measured with another use of the meter. + std::set> compared; + for (const Scenario& scenario : scenarios) + { + if (scenario.load == Load::FlatOut && !UsualMeter(scenario)) + { + compared.insert({scenario.writer, scenario.bufferSize, scenario.threads}); + } + } + std::vector meterHeader{"writer", "buffer", "threads", "counter", "tags", "lines/s", "per thread", + "call p50", "p99", "cpu/line", "bytes/line"}; + if (compare) + { + meterHeader.push_back("vs baseline"); + } + Table meters(meterHeader, 5); + // The uses of a meter next to each other, for every writer, buffer size and number of threads. + std::vector order(scenarios.size()); + std::iota(order.begin(), order.end(), std::size_t{0}); + std::stable_sort(order.begin(), order.end(), [&](std::size_t left, std::size_t right) + { + const auto rank = [&](const Scenario& scenario) + { return std::make_tuple(scenario.writer, scenario.bufferSize, scenario.threads, + scenario.createMeter, !scenario.tags); }; + return rank(scenarios[left]) < rank(scenarios[right]); + }); + for (const std::size_t i : order) + { + const Scenario& scenario = scenarios[i]; + const Summary& summary = summaries[i]; + if (scenario.load != Load::FlatOut || summary.measurements == 0 || + compared.count({scenario.writer, scenario.bufferSize, scenario.threads}) == 0) + { + continue; + } + std::vector row{WriterName(scenario.writer), + Buffer(scenario.bufferSize), + std::to_string(scenario.threads), + scenario.createMeter ? "created per line" : "created once", + scenario.tags ? "two" : "none", + Rate(summary.linesPerSecond), + Rate(summary.linesPerSecond / scenario.threads), + Time(summary.callP50), + Time(summary.callP99), + Time(summary.cpuNanosPerLine), + Format("%.0f", summary.bytesPerLine)}; + if (compare) + { + row.push_back(againstBaseline(scenario, summary)); + } + meters.Add(row); + } + if (!meters.Empty()) + { + std::printf("Meters: flat out, with the counter created once per thread or for every line, with and without" + "\ntags.\n\n"); + meters.Print(); + std::printf("\n"); + } + + std::printf("Notes\n"); + std::printf(" - lines/s is the median of the measurements. spread is how far the slowest and the fastest of them" + "\n are apart: a difference between two configurations that is smaller than that is not one.\n"); + std::printf(" - The call times include reading the clock, about %.0f ns here. Flat out, one call in 32 is timed." + "\n", + clockNanos); + std::printf(" - cpu/line is the processor time of the whole recording process, on all its threads, per line.\n"); + std::printf(" - arrived is the share of the recorded lines that this test received.\n"); + if (!paced.Empty()) + { + std::printf(" - A paced thread records the lines that are due once every millisecond, a few at a time.\n"); + } + for (const std::string& reason : plan.leftOut) + { + std::printf(" - Left out: %s.\n", reason.c_str()); + } + if (failed > 0) + { + std::printf(" - %d measurements FAILED: see the lines above the tables.\n", failed); + } +} + +bool EveryLineArrived(const std::vector& measurements) +{ + return !measurements.empty() && + std::all_of(measurements.begin(), measurements.end(), [](const Measurement& measurement) + { return measurement.ok && measurement.received == measurement.recorded; }); +} + +void WriteCsv(const std::string& path, const std::string& label, const Plan& plan, const Measurements& measurements) +{ + std::ofstream file(path); + if (!file) + { + throw std::runtime_error("Could not write " + path); + } + file << "label,key,writer,buffer,threads,load,rate_per_thread,meter,tags,repeat,ok,lines_per_second," + "call_p50_ns,call_p99_ns,call_p999_ns,call_max_ns,clock_ns,cpu_ns_per_line,recorded_lines," + "received_lines,datagrams,bytes,delay_p50_us,delay_p99_us,delay_max_us\n"; + for (std::size_t i = 0; i < plan.scenarios.size(); ++i) + { + const Scenario& scenario = plan.scenarios[i]; + for (std::size_t repeat = 0; repeat < measurements[i].size(); ++repeat) + { + const Measurement& measurement = measurements[i][repeat]; + file << label << ',' << scenario.Key() << ',' << WriterName(scenario.writer) << ',' + << scenario.bufferSize << ',' << scenario.threads << ',' << LoadName(scenario.load) << ',' + << scenario.linesPerSecond << ',' << (scenario.createMeter ? "create" : "reuse") << ',' + << (scenario.tags ? 1 : 0) << ',' << repeat + 1 << ',' << (measurement.ok ? 1 : 0) << ',' + << Format("%.1f", measurement.linesPerSecond) << ',' << Format("%.0f", measurement.callP50) << ',' + << Format("%.0f", measurement.callP99) << ',' << Format("%.0f", measurement.callP999) << ',' + << Format("%.0f", measurement.callMax) << ',' << Format("%.1f", measurement.clockNanos) << ',' + << Format("%.1f", measurement.cpuNanosPerLine) << ',' << measurement.recorded << ',' + << measurement.received << ',' << measurement.datagrams << ',' << measurement.bytes << ',' + << Format("%.0f", measurement.delayP50) << ',' << Format("%.0f", measurement.delayP99) << ',' + << Format("%.0f", measurement.delayMax) << '\n'; + } + } +} + +std::map ReadBaseline(const std::string& path) +{ + std::ifstream file(path); + if (!file) + { + throw std::runtime_error("Could not read " + path); + } + std::string line; + std::getline(file, line); + const std::vector header = Split(line, ','); + const auto column = [&](const std::string& name) + { + const auto found = std::find(header.begin(), header.end(), name); + if (found == header.end()) + { + throw std::runtime_error(path + " has no column " + name); + } + return static_cast(found - header.begin()); + }; + const std::size_t key = column("key"); + const std::size_t ok = column("ok"); + const std::size_t rate = column("lines_per_second"); + + std::map> rates; + while (std::getline(file, line)) + { + const std::vector fields = Split(line, ','); + if (fields.size() > std::max({key, ok, rate}) && fields[ok] == "1") + { + rates[fields[key]].push_back(std::stod(fields[rate])); + } + } + std::map medians; + for (const auto& [scenario, values] : rates) + { + medians[scenario] = Median(values); + } + return medians; +} + +} // namespace perf diff --git a/performance_tests/writer_performance_test/report.h b/performance_tests/writer_performance_test/report.h new file mode 100644 index 0000000..5e37149 --- /dev/null +++ b/performance_tests/writer_performance_test/report.h @@ -0,0 +1,58 @@ +#pragma once + +#include "plan.h" + +#include +#include +#include +#include + +namespace perf { + +// What one measurement of a scenario gave. +struct Measurement +{ + bool ok = false; + std::string problem; // why it is not ok + + double linesPerSecond = 0; // recorded during the measurement, by all threads together + double callP50 = 0; // how long a call that records a line took, in nanoseconds + double callP99 = 0; + double callP999 = 0; + double callMax = 0; + double clockNanos = 0; // how long reading the clock takes, which the call times include + double cpuNanosPerLine = 0; // processor time of the recording process, on all its threads + + std::uint64_t recorded = 0; // lines, warm-up included + std::uint64_t received = 0; // lines + std::uint64_t datagrams = 0; + std::uint64_t bytes = 0; + + double delayP50 = 0; // paced load: from recording a line to its arrival, in microseconds + double delayP99 = 0; + double delayMax = 0; +}; + +// The measurements of a run: for every scenario of the plan, one per repeat. +using Measurements = std::vector>; + +// A rate in words, for example 12.35 M or 853.1 k. +std::string Rate(double perSecond); + +// What is being run, on what, and for how long. Warns if this program was built without optimization. +void PrintHeader(const std::string& label, const Selection& selection, const Plan& plan); + +// The tables and the notes below them. With a baseline (see ReadBaseline), the rates are also shown as a multiple +// of the baseline's. +void PrintReport(const Plan& plan, const Measurements& measurements, const std::map& baseline); + +// Whether every line that was recorded in these measurements arrived. +bool EveryLineArrived(const std::vector& measurements); + +// Writes every measurement to a file of comma-separated values, one row each. +void WriteCsv(const std::string& path, const std::string& label, const Plan& plan, const Measurements& measurements); + +// From a file that WriteCsv wrote: the median rate of every scenario, by Scenario::Key(). +std::map ReadBaseline(const std::string& path); + +} // namespace perf diff --git a/performance_tests/writer_performance_test/scenario.h b/performance_tests/writer_performance_test/scenario.h new file mode 100644 index 0000000..14ef4cb --- /dev/null +++ b/performance_tests/writer_performance_test/scenario.h @@ -0,0 +1,119 @@ +#pragma once + +#include +#include +#include +#include +#include + +namespace perf { + +// The socket writer that the library sends through. +enum class WriterKind +{ + Udp, + Unix +}; + +// How the recording threads call the library. +enum class Load +{ + FlatOut, // every thread records as fast as it can + Paced // every thread records a fixed number of lines per second +}; + +inline std::string WriterName(WriterKind kind) { return kind == WriterKind::Udp ? "udp" : "unix"; } + +inline std::string LoadName(Load load) { return load == Load::FlatOut ? "flat-out" : "paced"; } + +// One configuration to measure. To measure a new setting of the writer, add it here, to Key(), and to the places +// that hand a scenario to the recording process (ToArguments and FromArguments below) and to the library +// (recording_run.cpp). +struct Scenario +{ + WriterKind writer = WriterKind::Unix; + unsigned int bufferSize = 0; // 0: no buffering, the library sends every line as it is recorded + int threads = 1; // recording threads + Load load = Load::FlatOut; + int linesPerSecond = 0; // paced load: the rate of each thread + bool createMeter = false; // create the counter for every line, instead of once per thread + bool tags = true; // whether the counter has tags + + // Names the scenario in the CSV file. Two runs are compared scenario by scenario through this. + std::string Key() const + { + std::string key = WriterName(writer) + "/buffer=" + std::to_string(bufferSize) + + "/threads=" + std::to_string(threads) + "/" + LoadName(load); + if (load == Load::Paced) + { + key += "=" + std::to_string(linesPerSecond); + } + key += createMeter ? "/create" : "/reuse"; + key += tags ? "/tags" : "/no-tags"; + return key; + } +}; + +// Settings in the form name=value, which is how the test hands a scenario to the process that records, and how that +// process hands back what it measured. +using Arguments = std::map; + +inline Arguments ParseArguments(const std::vector& words) +{ + Arguments arguments; + for (const std::string& word : words) + { + const std::size_t equals = word.find('='); + if (equals == std::string::npos) + { + throw std::invalid_argument("Expected name=value, got: " + word); + } + arguments[word.substr(0, equals)] = word.substr(equals + 1); + } + return arguments; +} + +inline const std::string& Text(const Arguments& arguments, const std::string& name) +{ + const auto found = arguments.find(name); + if (found == arguments.end()) + { + throw std::invalid_argument("Missing setting: " + name); + } + return found->second; +} + +inline double Number(const Arguments& arguments, const std::string& name) { return std::stod(Text(arguments, name)); } + +inline std::uint64_t Count(const Arguments& arguments, const std::string& name) +{ + return std::stoull(Text(arguments, name)); +} + +inline std::vector ToArguments(const Scenario& scenario) +{ + return { + "writer=" + WriterName(scenario.writer), + "buffer=" + std::to_string(scenario.bufferSize), + "threads=" + std::to_string(scenario.threads), + "load=" + LoadName(scenario.load), + "rate=" + std::to_string(scenario.linesPerSecond), + "create=" + std::to_string(scenario.createMeter ? 1 : 0), + "tags=" + std::to_string(scenario.tags ? 1 : 0), + }; +} + +inline Scenario FromArguments(const Arguments& arguments) +{ + Scenario scenario; + scenario.writer = Text(arguments, "writer") == "udp" ? WriterKind::Udp : WriterKind::Unix; + scenario.bufferSize = static_cast(Count(arguments, "buffer")); + scenario.threads = static_cast(Count(arguments, "threads")); + scenario.load = Text(arguments, "load") == "paced" ? Load::Paced : Load::FlatOut; + scenario.linesPerSecond = static_cast(Count(arguments, "rate")); + scenario.createMeter = Count(arguments, "create") != 0; + scenario.tags = Count(arguments, "tags") != 0; + return scenario; +} + +} // namespace perf diff --git a/performance_tests/writer_performance_test/writer_performance_test.cpp b/performance_tests/writer_performance_test/writer_performance_test.cpp new file mode 100644 index 0000000..a2dd254 --- /dev/null +++ b/performance_tests/writer_performance_test/writer_performance_test.cpp @@ -0,0 +1,569 @@ +// Measures what it costs to record meters through the writers, for a list of configurations: the socket writer, the +// buffer size, the number of recording threads, the load, and the way the meters are used. +// +// For every configuration (a Scenario, see scenario.h and plan.h), the test +// - opens a socket of its own on this machine and reads what arrives there (receiver.h), +// - starts a process that creates a Registry for that socket and lets its threads record (recording_run.h), +// - and, once that process has exited, compares what was recorded with what arrived. +// The recording runs in a process of its own because the library's Writer is one per process and is set up once. +// +// It reports, per configuration (report.h): +// - the lines recorded per second, and how that grows with the number of threads, +// - how long a call that records a line takes: the median, the 99th percentile and the longest, +// - the processor time per line, on all threads of the recording process, +// - the share of the recorded lines that arrived, and how many lines a datagram held, +// - under a paced load, how long a line took from being recorded to arriving. +// +// The numbers only mean something from an optimized build on a machine that is doing nothing else. README.md, next +// to this file, says how to build and run the test and how to read its report. + +#include "plan.h" +#include "receiver.h" +#include "recording_run.h" +#include "report.h" +#include "scenario.h" + +#include +#include +#include +#include +#include +#include + +#if defined(__APPLE__) +#include +#endif + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +extern char** environ; + +namespace { + +using perf::Load; +using perf::Measurement; +using perf::Plan; +using perf::Scenario; +using perf::Selection; +using perf::WriterKind; + +// ---------- The command line ---------- + +struct Options +{ + bool full = false; + bool smoke = false; + bool list = false; + bool help = false; + double seconds = 0; // 0: as the selection has it + double warmUp = -1; // negative: as the selection has it + int repeats = 0; // 0: as the selection has it + int rate = 0; // 0: as the selection has it + std::vector writers; + std::vector threads; + std::vector buffers; + std::string csvPath; + std::string baselinePath; + std::string label; +}; + +void PrintUsage() +{ + std::printf( + "Usage: writer_performance_test [options]\n" + "\n" + "Measures recording through the writers for a list of configurations and prints a report.\n" + "Without options it takes about two minutes.\n" + "\n" + " --full more thread counts, buffer sizes and rates: about twelve minutes\n" + " --smoke a few seconds: checks that the test works, and that every line arrives\n" + " --list print the configurations and how long they take, and stop\n" + "\n" + " --writers LIST udp, unix or both, for example: --writers unix\n" + " --threads LIST numbers of recording threads, for example: --threads 1,2,4,8,16,32\n" + " --buffers LIST buffer sizes in bytes, 0 for no buffering, for example: --buffers 0,1024,8192\n" + " With --threads or --buffers, every size is measured with every number of threads.\n" + " Use --list to see what a combination of options comes to.\n" + " --rate N lines per second and thread under the paced load\n" + " --seconds S how long each configuration is measured\n" + " --warmup S how long the threads record before the measurement begins\n" + " --repeats N how often each configuration is measured; the report shows the median\n" + "\n" + " --csv FILE also write every measurement to FILE\n" + " --baseline FILE a file written with --csv by an earlier run: the report shows this run against it\n" + " --label TEXT a name for this run, for the report and the CSV file\n" + "\n" + "The numbers say little unless this program was built with optimization, and unless the machine is\n" + "doing nothing else. How to build it, and how to read the report, is in the README.md next to its source.\n"); +} + +std::vector Split(const std::string& text, char separator) +{ + std::vector parts; + std::stringstream stream(text); + for (std::string part; std::getline(stream, part, separator);) + { + parts.push_back(part); + } + return parts; +} + +// A number that was given to an option. Anything else is a mistake on the command line. +double NumberFor(const std::string& option, const std::string& text) +{ + try + { + std::size_t used = 0; + const double number = std::stod(text, &used); + if (used == text.size() && number >= 0) + { + return number; + } + } + catch (const std::exception&) + { + } + throw std::invalid_argument(option + " takes numbers that are not negative, not: " + text); +} + +Options ParseOptions(const std::vector& words) +{ + Options options; + for (std::size_t i = 0; i < words.size(); ++i) + { + const std::string& word = words[i]; + const auto value = [&]() -> const std::string& + { + if (i + 1 >= words.size()) + { + throw std::invalid_argument(word + " needs a value"); + } + return words[++i]; + }; + const auto number = [&]() { return NumberFor(word, value()); }; + + if (word == "--full") + { + options.full = true; + } + else if (word == "--smoke") + { + options.smoke = true; + } + else if (word == "--list") + { + options.list = true; + } + else if (word == "--help" || word == "-h") + { + options.help = true; + } + else if (word == "--writers") + { + for (const std::string& name : Split(value(), ',')) + { + if (name != "udp" && name != "unix") + { + throw std::invalid_argument("Unknown writer: " + name); + } + options.writers.push_back(name == "udp" ? WriterKind::Udp : WriterKind::Unix); + } + } + else if (word == "--threads") + { + for (const std::string& count : Split(value(), ',')) + { + options.threads.push_back(std::max(1, static_cast(NumberFor(word, count)))); + } + } + else if (word == "--buffers") + { + for (const std::string& size : Split(value(), ',')) + { + options.buffers.push_back(static_cast(NumberFor(word, size))); + } + } + else if (word == "--rate") + { + options.rate = std::max(1, static_cast(number())); + } + else if (word == "--seconds") + { + options.seconds = number(); + } + else if (word == "--warmup") + { + options.warmUp = number(); + } + else if (word == "--repeats") + { + options.repeats = std::max(1, static_cast(number())); + } + else if (word == "--csv") + { + options.csvPath = value(); + } + else if (word == "--baseline") + { + options.baselinePath = value(); + } + else if (word == "--label") + { + options.label = value(); + } + else + { + throw std::invalid_argument("Unknown option: " + word); + } + } + return options; +} + +// The selection that the options ask for: one of the three that plan.h has, with what the options change in it. +Selection SelectionFor(const Options& options) +{ + Selection selection = + options.smoke ? perf::SmokeSelection() : options.full ? perf::FullSelection() : Selection(); + if (options.seconds > 0) + { + selection.seconds = options.seconds; + selection.pacedSeconds = options.seconds; + } + if (options.warmUp >= 0) + { + selection.warmUp = options.warmUp; + } + if (options.repeats > 0) + { + selection.repeats = options.repeats; + } + if (options.rate > 0) + { + selection.rates = {options.rate}; + } + if (!options.writers.empty()) + { + selection.writers = options.writers; + // The meters are compared with the writers that were asked for. + const bool wasCompared = !selection.meterWriters.empty(); + std::erase_if(selection.meterWriters, [&](WriterKind writer) + { return std::find(options.writers.begin(), options.writers.end(), writer) == + options.writers.end(); }); + if (wasCompared && selection.meterWriters.empty()) + { + selection.meterWriters = {options.writers.front()}; + } + } + // The thread counts and buffer sizes that were asked for apply to every part of the run that the selection has. + if (!options.threads.empty()) + { + selection.everyCombination = true; + for (std::vector* threads : {&selection.threads, &selection.pacedThreads, &selection.meterThreads}) + { + if (!threads->empty()) + { + *threads = options.threads; + } + } + } + if (!options.buffers.empty()) + { + selection.everyCombination = true; + selection.buffers = options.buffers; + if (!selection.pacedBuffers.empty()) + { + // The paced load is for buffered writing, so it takes the sizes other than 0. + selection.pacedBuffers.clear(); + for (const unsigned int buffer : options.buffers) + { + if (buffer != 0) + { + selection.pacedBuffers.push_back(buffer); + } + } + } + } + return selection; +} + +// ---------- One measurement ---------- + +struct ProcessResult +{ + bool exited = false; // by itself, and in time + int exitCode = -1; + std::string output; // what it wrote to its standard output +}; + +// Runs a program and collects what it prints. The process is killed if it has not exited after `limit`. +ProcessResult RunProcess(const std::string& executable, const std::vector& arguments, + std::chrono::milliseconds limit) +{ + int ends[2]; + if (::pipe(ends) != 0) + { + throw std::system_error(errno, std::generic_category(), "Could not create a pipe"); + } + ::fcntl(ends[0], F_SETFD, FD_CLOEXEC); + + posix_spawn_file_actions_t actions; + posix_spawn_file_actions_init(&actions); + posix_spawn_file_actions_adddup2(&actions, ends[1], STDOUT_FILENO); + posix_spawn_file_actions_addclose(&actions, ends[1]); + + std::vector argv; + argv.push_back(const_cast(executable.c_str())); + for (const std::string& argument : arguments) + { + argv.push_back(const_cast(argument.c_str())); + } + argv.push_back(nullptr); + + pid_t process = 0; + const int error = posix_spawn(&process, executable.c_str(), &actions, nullptr, argv.data(), environ); + posix_spawn_file_actions_destroy(&actions); + ::close(ends[1]); + if (error != 0) + { + ::close(ends[0]); + throw std::system_error(error, std::generic_category(), "Could not start " + executable); + } + + ProcessResult result; + bool killed = false; + const auto deadline = std::chrono::steady_clock::now() + limit; + pollfd waiting{ends[0], POLLIN, 0}; + char text[4096]; + for (;;) + { + const auto left = + std::chrono::duration_cast(deadline - std::chrono::steady_clock::now()); + if (left.count() <= 0) + { + ::kill(process, SIGKILL); + killed = true; + break; + } + if (::poll(&waiting, 1, static_cast(std::min(left.count(), 100))) <= 0) + { + continue; + } + const ssize_t size = ::read(ends[0], text, sizeof(text)); + if (size <= 0) + { + break; // the process has closed its end: it has exited + } + result.output.append(text, static_cast(size)); + } + ::close(ends[0]); + + int status = 0; + ::waitpid(process, &status, 0); + result.exited = !killed && WIFEXITED(status); + result.exitCode = WIFEXITED(status) ? WEXITSTATUS(status) : -1; + return result; +} + +// Where this program is, to run it again as the recording process. +std::string ExecutablePath(const char* invokedAs) +{ + char path[PATH_MAX]; +#if defined(__APPLE__) + std::uint32_t size = sizeof(path); + if (_NSGetExecutablePath(path, &size) == 0) + { + return path; + } +#elif defined(__linux__) + const ssize_t length = ::readlink("/proc/self/exe", path, sizeof(path) - 1); + if (length > 0) + { + path[length] = '\0'; + return path; + } +#endif + return invokedAs; +} + +// Measures a scenario once: the recording in a process of its own, the receiving here. +Measurement Measure(const Scenario& scenario, const Selection& selection, const std::string& executable) +{ + Measurement measurement; + try + { + perf::Receiver receiver(scenario.writer); + receiver.Start(scenario.load == Load::Paced); + + const double seconds = perf::MeasuredSeconds(scenario, selection); + std::vector arguments{"--child"}; + for (const std::string& argument : perf::ToArguments(scenario)) + { + arguments.push_back(argument); + } + arguments.push_back("location=" + receiver.Location()); + arguments.push_back("warmup=" + std::to_string(selection.warmUp)); + arguments.push_back("seconds=" + std::to_string(seconds)); + + // Far longer than it takes, so that only a process that hangs is stopped. + const std::chrono::milliseconds limit(static_cast((selection.warmUp + seconds + 30) * 1000)); + const ProcessResult process = RunProcess(executable, arguments, limit); + const perf::Received received = receiver.Stop(); + + if (!process.exited) + { + measurement.problem = "the recording process crashed or did not exit in time"; + return measurement; + } + const std::size_t prefix = process.output.find(perf::kResultPrefix); + if (process.exitCode != 0 || prefix == std::string::npos) + { + measurement.problem = "the recording process failed"; + return measurement; + } + const std::size_t begin = prefix + std::string(perf::kResultPrefix).size(); + const perf::Arguments result = perf::ParseArguments( + Split(process.output.substr(begin, process.output.find('\n', begin) - begin), ' ')); + + measurement.linesPerSecond = perf::Number(result, "rate"); + measurement.callP50 = perf::Number(result, "call_p50"); + measurement.callP99 = perf::Number(result, "call_p99"); + measurement.callP999 = perf::Number(result, "call_p999"); + measurement.callMax = perf::Number(result, "call_max"); + measurement.clockNanos = perf::Number(result, "clock"); + measurement.recorded = perf::Count(result, "recorded"); + measurement.cpuNanosPerLine = measurement.recorded == 0 ? 0 + : perf::Number(result, "cpu") * 1e9 / + static_cast(measurement.recorded); + + measurement.received = received.lines; + measurement.datagrams = received.datagrams; + measurement.bytes = received.bytes; + measurement.delayP50 = static_cast(received.delayMicros.Percentile(0.5)); + measurement.delayP99 = static_cast(received.delayMicros.Percentile(0.99)); + measurement.delayMax = static_cast(received.delayMicros.Max()); + measurement.ok = true; + } + catch (const std::exception& error) + { + measurement.problem = error.what(); + } + return measurement; +} + +// ---------- The run ---------- + +int Run(const Options& options, const std::string& executable) +{ + const Selection selection = SelectionFor(options); + const Plan plan = perf::MakePlan(selection); + const std::map baseline = + options.baselinePath.empty() ? std::map() : perf::ReadBaseline(options.baselinePath); + + perf::PrintHeader(options.label, selection, plan); + if (options.list) + { + for (const Scenario& scenario : plan.scenarios) + { + std::printf(" %s\n", perf::Describe(scenario).c_str()); + } + for (const std::string& reason : plan.leftOut) + { + std::printf(" Left out: %s\n", reason.c_str()); + } + return 0; + } + + // Every scenario once, then every scenario again, each time beginning with another one. What changes slowly on + // the machine while the test runs, such as its temperature, then affects all scenarios alike. + const std::size_t count = plan.scenarios.size(); + const auto repeats = static_cast(selection.repeats); + perf::Measurements measurements(count); + bool failed = false; + for (std::size_t repeat = 0; repeat < repeats; ++repeat) + { + for (std::size_t step = 0; step < count; ++step) + { + const std::size_t index = (step + repeat * count / repeats) % count; + const Scenario& scenario = plan.scenarios[index]; + const Measurement measurement = Measure(scenario, selection, executable); + measurements[index].push_back(measurement); + failed = failed || !measurement.ok; + + std::printf("[%3zu/%zu] %-62s %s\n", repeat * count + step + 1, count * repeats, + perf::Describe(scenario).c_str(), + measurement.ok ? (perf::Rate(measurement.linesPerSecond) + " lines/s").c_str() + : ("FAILED: " + measurement.problem).c_str()); + std::fflush(stdout); + } + } + std::printf("\n"); + + perf::PrintReport(plan, measurements, baseline); + if (!options.csvPath.empty()) + { + perf::WriteCsv(options.csvPath, options.label, plan, measurements); + std::printf("\nEvery measurement is in %s\n", options.csvPath.c_str()); + } + + if (options.smoke) + { + // At the low rate of the smoke run, no line may be lost with either writer. + for (std::size_t i = 0; i < count; ++i) + { + if (!perf::EveryLineArrived(measurements[i])) + { + std::printf("\nSMOKE TEST FAILED: %s: not every recorded line arrived\n", + perf::Describe(plan.scenarios[i]).c_str()); + return 1; + } + } + std::printf("\nSmoke test passed: every recorded line arrived.\n"); + } + return failed ? 1 : 0; +} + +} // namespace + +int main(int argc, char* argv[]) +{ + const std::vector words(argv + 1, argv + argc); + try + { + // The recording side of one measurement, which Measure() starts. + if (!words.empty() && words.front() == "--child") + { + return perf::RunRecording(perf::ParseArguments({words.begin() + 1, words.end()})); + } + + const Options options = ParseOptions(words); + if (options.help) + { + PrintUsage(); + return 0; + } + return Run(options, ExecutablePath(argv[0])); + } + catch (const std::invalid_argument& error) + { + std::fprintf(stderr, "%s\n\n", error.what()); + PrintUsage(); + return 2; + } + catch (const std::exception& error) + { + std::fprintf(stderr, "%s\n", error.what()); + return 1; + } +}