diff --git a/CMakeLists.txt b/CMakeLists.txt index 07e6dfa7f1d9..0e3bd13d0fb3 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1482,6 +1482,39 @@ if(BUILD_TESTS) "1,write,300000,primary" ) + # Same workload as pi_basic_blocking, driven by locust rather than piccolo, + # so that the number of concurrent clients can be varied. + add_e2e_test( + NAME pi_basic_blocking_locust + PYTHON_SCRIPT ${CMAKE_SOURCE_DIR}/tests/basicperf_locust.py + LABEL perf + CONFIGURATIONS perf + ADDITIONAL_ARGS + --package + "samples/apps/basic/basic" + --perf-label + "Basic Blocking Locust" + # add_piccolo_test passes this for every piccolo perf test, but + # add_e2e_test does not, and the default is 10. Without it this + # benchmark snapshots every 10 transactions and measures the disk. + --snapshot-tx-interval + 10000 + --users + 320 + --spawn-rate + 320 + --locust-processes + 10 + --measure-time-s + 20 + # Blocking writes return on commit, and commit cannot outpace the + # signature interval, so this sweeps from latency-bound to capacity-bound. + --sig-ms-intervals + 2 + 100 + 1000 + ) + if(WORKER_THREADS) add_piccolo_test( NAME pi_basic_mt diff --git a/scripts/perf_compare_radar.py b/scripts/perf_compare_radar.py index ab43797f0fad..d698093314af 100644 --- a/scripts/perf_compare_radar.py +++ b/scripts/perf_compare_radar.py @@ -248,6 +248,49 @@ def axis_label_color(percent: float, higher_is_better: bool, within_noise: bool) return LABEL_GOOD if improved else LABEL_BAD +# U+2026 HORIZONTAL ELLIPSIS. One character wide, so eliding the middle of a +# word with it costs a single column rather than the three of "...". +ELLIPSIS = "\u2026" + +# Eliding a shorter word saves nothing, since "abc" and "a" + ELLIPSIS + "c" +# are both three characters. +MIN_ELIDABLE_WORD_LENGTH = 4 + + +def elide_word(word: str) -> str: + """Replace the middle of a word with a single ellipsis character.""" + return f"{word[0]}{ELLIPSIS}{word[-1]}" + + +def shorten_label(label: str, max_length: int) -> str: + """Shorten a label to fit by eliding the middles of its words. + + Words are elided from left to right, each keeping its first and last letter, + until the label fits. Benchmarks measured at several settings differ only in + their last word, for example the interval in "Basic Blocking Locust 100ms", + so eliding from the left keeps the part which tells them apart readable for + as long as possible. + """ + if len(label) <= max_length: + return label + + words = label.split(" ") + for index, word in enumerate(words): + if len(word) < MIN_ELIDABLE_WORD_LENGTH: + continue + words[index] = elide_word(word) + elided = " ".join(words) + if len(elided) <= max_length: + return elided + + # Every word is elided and it still does not fit. Keep the end, which is + # what distinguishes one setting of a benchmark from another. + elided = " ".join(words) + if max_length <= 1: + return elided[:max_length] + return ELLIPSIS + elided[len(elided) - (max_length - 1) :] + + def axis_label( benchmark: str, value: float, percent: float, unit: str, within_noise: bool ) -> str: @@ -257,10 +300,7 @@ def axis_label( f": {metric_label_value(value, unit)} " f"{format_delta_percent(percent, within_noise)}" ) - max_label_length = MAX_AXIS_LABEL_LENGTH - len(suffix) - if len(label) <= max_label_length: - return f"{label}{suffix}" - return f"{label[:max_label_length - 3]}...{suffix}" + return f"{shorten_label(label, MAX_AXIS_LABEL_LENGTH - len(suffix))}{suffix}" def normalized_percent(value: float, baseline: float) -> float: @@ -319,14 +359,33 @@ def render_mermaid_radar_chart( for data in trend if (value := metric_value(data, benchmark, metric)) is not None ] - if not main_values: - continue - baseline = ewma(main_values) + # A benchmark added by this branch has no main runs to build a baseline + # from. Rather than drop it, which would make a new benchmark invisible + # on the very pull request which adds it, use this branch's own earliest + # run as the reference, so the axis is normalised and plotted exactly + # like every other one. + if main_values: + baseline = ewma(main_values) + sigma = statistics.pstdev(main_values) if len(main_values) > 1 else 0.0 + else: + branch_values = [ + value + for data in branch_runs + if (value := metric_value(data, benchmark, metric)) is not None + ] + if not branch_values: + continue + baseline = branch_values[0] + # Spread is measured the same way as for a benchmark with main + # history, from the runs available, so that such an axis carries a + # band and a noise threshold like every other one rather than + # collapsing to a point at the baseline. + sigma = statistics.pstdev(branch_values) if len(branch_values) > 1 else 0.0 + if baseline <= 0: continue - sigma = statistics.pstdev(main_values) if len(main_values) > 1 else 0.0 branch_percent = normalized_percent(branch_value, baseline) sigma_percent = normalized_percent(sigma, baseline) within_noise = within_noise_band(branch_percent, sigma_percent) @@ -507,6 +566,14 @@ def render_comparison( "Higher is better for throughput and rate, lower for latency and memory._" ), "", + ( + "_A benchmark which does not exist on `main` yet has no baseline of its " + "own, so this branch's earliest run is used as its reference and its band " + "is measured across this branch's runs. Its axis is normalized, scaled and " + "coloured like any other, but the comparison is against this branch rather " + "than against `main`._" + ), + "", "", "", ] diff --git a/tests/basicperf_locust.py b/tests/basicperf_locust.py new file mode 100644 index 000000000000..8ca189690e68 --- /dev/null +++ b/tests/basicperf_locust.py @@ -0,0 +1,309 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the Apache 2.0 License. + +""" +"Basic Blocking Locust" benchmark. + +Measures the throughput of blocking writes (PUT /records/blocking/{key}), which +only return once the transaction has committed. This covers the same ground as +the piccolo-driven "Basic Blocking" benchmark, but drives load with locust +instead, so that the client count can be ramped up and the workload described +in Python rather than in a pre-generated parquet file. + +The load itself is defined in infra/basicperf_locustfile.py. This script owns +the network, runs locust against it, and converts locust's statistics into +bencher metrics. +""" + +import argparse +import csv +import math +import os +import subprocess + +import infra.bencher +import infra.e2e_args +import infra.interfaces +import infra.key_space +import infra.net +import infra.network +import infra.proc +from loguru import logger as LOG + +LOCUST_FILE_NAME = "basicperf_locustfile.py" + +# Prefix for the CSV files locust writes. Locust appends _stats.csv, +# _failures.csv, and so on. +CSV_PREFIX = "locust" + +# Name of the row holding totals across all request types in locust's stats CSV. +AGGREGATED_ROW_NAME = "Aggregated" + +# Slack allowed on top of the expected spawn and measurement time before the +# run is considered stuck and killed by locust's own --run-time. +RUN_TIME_MARGIN_S = 60 + +# Shortest measurement window, as a fraction of the one requested, which is +# still accepted as describing steady state. +MIN_MEASURED_FRACTION = 0.9 + + +def locust_file_path() -> str: + return os.path.join( + os.path.dirname(os.path.realpath(__file__)), "infra", LOCUST_FILE_NAME + ) + + +def run_locust(args, network, primary) -> dict: + """Run locust to completion against the given node, and return the + aggregated statistics it recorded.""" + csv_prefix = os.path.join(network.common_dir, CSV_PREFIX) + + session_auth = primary.session_auth("user0")["session_auth"] + host = "https://" + infra.interfaces.make_address( + primary.get_public_rpc_host(), primary.get_public_rpc_port() + ) + + cmd = ["locust"] + cmd += ["--headless"] + cmd += ["--locustfile", locust_file_path()] + cmd += ["--host", host] + + # Client authentication + cmd += ["--ca", primary.session_ca()["ca"]] + cmd += ["--cert", session_auth.cert] + cmd += ["--key", session_auth.key] + + cmd += ["--key-space-size", f"{args.key_space_size}"] + + # Load profile + cmd += ["--users", f"{args.users}"] + cmd += ["--spawn-rate", f"{args.spawn_rate}"] + cmd += ["--measure-time-s", f"{args.measure_time_s}"] + + # The run is normally ended by the locustfile, a fixed time after the last + # user has spawned. This is only a backstop against a run which never + # finishes spawning, so it is deliberately generous: locust's --run-time + # includes the ramp, and pre-empting the real deadline would silently + # shorten the measurement window. + spawn_time_s = math.ceil(args.users / args.spawn_rate) + run_time_ceiling_s = spawn_time_s + args.measure_time_s + RUN_TIME_MARGIN_S + cmd += ["--run-time", f"{run_time_ceiling_s}s"] + + # A single locust process cannot saturate the service, because it drives + # all of its users from one thread. Fork enough workers to keep the client + # from being the bottleneck. + cmd += ["--processes", f"{args.locust_processes}"] + + # Workers reach the master over TCP, on a fixed port 5557 by default. Pick + # a free one instead, so that a second locust run on the same machine, from + # another test or another checkout, does not fail to bind. + master_port = infra.net.probably_free_local_port("localhost") + cmd += ["--master-bind-port", f"{master_port}"] + cmd += ["--master-port", f"{master_port}"] + + # Discard everything recorded while users were still being spawned, so the + # reported numbers describe steady state at the full user count. + cmd += ["--reset-stats"] + + cmd += ["--csv", csv_prefix] + + LOG.info(f"Starting locust: {' '.join(cmd)}") + # Locust exits non-zero if any request failed, which should fail the test. + subprocess.run(cmd, check=True) + + return read_aggregated_stats(f"{csv_prefix}_stats.csv") + + +def read_aggregated_stats(stats_path: str) -> dict: + with open(stats_path, "r") as f: + rows = list(csv.DictReader(f)) + + for row in rows: + if row["Name"] == AGGREGATED_ROW_NAME: + return row + + raise RuntimeError(f"No {AGGREGATED_ROW_NAME} row found in {stats_path}") + + +def stat_as_float(stats: dict, column: str) -> float: + """Read one numeric column from locust's statistics. + + Locust writes "N/A" rather than a number when it has too few samples to + produce a percentile, so a run which barely gathered any data would + otherwise fail here with a bare ValueError, after the network has already + been torn down and the evidence lost. + """ + value = stats[column] + try: + return float(value) + except ValueError as e: + raise RuntimeError( + f"Locust reported {column!r} as {value!r} rather than a number, " + "which means it had too few samples to compute it. The run did " + "not gather enough data to describe steady state." + ) from e + + +def measure(args, sig_ms_interval: int) -> dict: + """Run the workload against a fresh network with the given signature + interval, and return the statistics locust recorded.""" + args.sig_ms_interval = sig_ms_interval + # A response is only sent once its transaction commits, and commit cannot + # be observed any faster than the primary sends consensus updates. Move + # that in step with the signature interval, as commit_latency.py does, + # otherwise the shorter intervals are gated by the 100ms default and the + # setting has no effect. + args.consensus_update_timeout_ms = sig_ms_interval + + LOG.info(f"Starting nodes on {args.nodes} with {sig_ms_interval}ms signatures") + with infra.network.network( + args.nodes, args.binary_dir, args.debug_nodes, pdb=args.pdb + ) as network: + network.start_and_open(args) + + primary, _ = network.find_primary() + + infra.key_space.create_and_fill_key_space(args.key_space_size, primary) + + stats = run_locust(args, network, primary) + + request_count = int(stats["Request Count"]) + failure_count = int(stats["Failure Count"]) + throughput = stat_as_float(stats, "Requests/s") + # Locust reports response times in milliseconds. + median_latency_ms = stat_as_float(stats, "Median Response Time") + p99_latency_ms = stat_as_float(stats, "99%") + min_latency_ms = stat_as_float(stats, "Min Response Time") + + LOG.info( + f"{request_count} requests ({failure_count} failures) " + f"=> {throughput:.1f} tx/s" + ) + LOG.info( + f"Latency: min={min_latency_ms:.1f}ms p50={median_latency_ms:.1f}ms " + f"p99={p99_latency_ms:.1f}ms" + ) + + if request_count == 0: + raise RuntimeError("Locust recorded no requests") + + # Locust reports throughput over the window its statistics cover, so + # the two together give the length of that window. Check it against the + # window that was asked for: a short window still yields a plausible + # looking throughput, so without this a truncated run would be reported + # as a valid result. + measured_duration_s = request_count / throughput + if measured_duration_s < args.measure_time_s * MIN_MEASURED_FRACTION: + raise RuntimeError( + f"Measured over {measured_duration_s:.1f}s, but expected " + f"{args.measure_time_s}s. The run was cut short, so this " + "result does not describe steady state." + ) + LOG.info(f"Measured over {measured_duration_s:.1f}s") + + # Locust should already have exited non-zero, but do not report a + # throughput figure built from failed requests under any circumstances. + if failure_count != 0: + raise RuntimeError( + f"Locust recorded {failure_count} failures out of {request_count} requests" + ) + + mem = infra.proc.get_proc_memory_stats(primary.remote.remote.proc.pid) + + network.stop_all_nodes() + + return { + "throughput": throughput, + "median_latency_ms": median_latency_ms, + "p99_latency_ms": p99_latency_ms, + "min_latency_ms": min_latency_ms, + "memory": mem, + } + + +def run(args): + # Each interval needs its own network, since the signature interval is + # fixed in the node's configuration at startup. + results = {} + for sig_ms_interval in args.sig_ms_intervals: + results[sig_ms_interval] = measure(args, sig_ms_interval) + + bf = infra.bencher.Bencher() + for sig_ms_interval, result in results.items(): + label = f"{args.perf_label} (sig_ms_interval={sig_ms_interval}ms)" + bf.set(label, infra.bencher.Throughput(round(result["throughput"], 1))) + bf.set( + label, + infra.bencher.Latency( + value=result["median_latency_ms"], + high_value=result["p99_latency_ms"], + low_value=result["min_latency_ms"], + ), + ) + if result["memory"] is not None: + bf.set_memory(label, result["memory"]) + + LOG.info("Summary:") + for sig_ms_interval, result in results.items(): + LOG.info( + f" {sig_ms_interval:>5}ms signatures: " + f"{result['throughput']:>9.1f} tx/s, " + f"p50 {result['median_latency_ms']:.1f}ms" + ) + + +def cli_args(): + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter + ) + parser.add_argument( + "--users", + help="Number of concurrent locust users, each sending one blocking write at a time", + type=int, + default=320, + ) + parser.add_argument( + "--spawn-rate", + help="Number of users to start per second", + type=int, + default=320, + ) + parser.add_argument( + "--measure-time-s", + help="Seconds to measure for, once all users have spawned", + type=int, + default=20, + ) + parser.add_argument( + "--locust-processes", + help="Number of locust worker processes to fork", + type=int, + default=10, + ) + parser.add_argument( + "--sig-ms-intervals", + help="Signature intervals, in milliseconds, to measure the workload at. " + "Each is measured against its own network.", + type=int, + nargs="+", + default=[2, 100, 1000], + ) + parser.add_argument( + "--key-space-size", + help="Size of the key space which is pre-populated and written to", + type=int, + default=1000, + ) + return infra.e2e_args.cli_args( + parser=parser, accept_unknown=False, ledger_chunk_bytes_override="5MB" + ) + + +if __name__ == "__main__": + args = cli_args() + # A single node is enough: the benchmark measures the time taken to commit + # on the primary, and additional nodes only add replication cost. + args.nodes = infra.e2e_args.min_nodes(args, f=0) + + run(args) diff --git a/tests/infra/basicperf.py b/tests/infra/basicperf.py index 7bb518517723..327cfb96c187 100644 --- a/tests/infra/basicperf.py +++ b/tests/infra/basicperf.py @@ -3,7 +3,6 @@ import argparse import datetime import hashlib -import http import json import os import random @@ -20,6 +19,7 @@ import infra.bencher import infra.e2e_args import infra.jwt_issuer +import infra.key_space import infra.proc import infra.remote_client @@ -214,22 +214,6 @@ def __call__( ) -def create_and_fill_key_space(size: int, primary: infra.node.Node) -> list[str]: - LOG.info(f"Creating and filling key space of size {size}") - space = [f"{i}" for i in range(size)] - mapping = {key: f"{hashlib.sha256(key.encode()).hexdigest()}" for key in space} - with primary.client("user0") as c: - r = c.post("/records", mapping) - assert r.status_code == http.HTTPStatus.NO_CONTENT, r - # Quick sanity check - for j in [0, -1]: - r = c.get(f"/records/{space[j]}") - assert r.status_code == http.HTTPStatus.OK, r - assert r.body.text() == mapping[space[j]], r - LOG.info("Key space created and filled") - return space - - def replace_primary(network, host, old_primary, snapshots_dir, statistics): LOG.info(f"Set up new node: {host}") node = network.create_node(host) @@ -297,7 +281,9 @@ def run(args): jwt = jwt_issuer.issue_jwt() additional_headers["Authorization"] = f"Bearer {jwt}" - key_space = create_and_fill_key_space(args.key_space_size, primary) + key_space = infra.key_space.create_and_fill_key_space( + args.key_space_size, primary + ) clients = [] client_idx = 0 diff --git a/tests/infra/basicperf_locustfile.py b/tests/infra/basicperf_locustfile.py new file mode 100644 index 000000000000..f1d8cfc4c893 --- /dev/null +++ b/tests/infra/basicperf_locustfile.py @@ -0,0 +1,151 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the Apache 2.0 License. + +""" +Locust workload for the "Basic Blocking Locust" benchmark. + +Each user issues blocking writes (PUT /records/blocking/{key}) one at a time, +waiting for each response before sending the next. That endpoint only responds +once the transaction has committed, so a single user's rate is bounded by +commit latency, and total throughput is driven by the number of users. + +FastHttpUser (geventhttpclient) is used rather than HttpUser (requests), +because the latter cannot drive enough requests per second to saturate the +service. +""" + +import hashlib +import logging +import random +import ssl + +import gevent +from locust import constant, events, task +from locust.contrib.fasthttp import FastHttpUser +from locust.runners import WorkerRunner + +# All requests are reported under a single name, so that locust aggregates them +# into one statistics entry rather than one per key. +REQUEST_NAME = "PUT /records/blocking/{key}" + +DEFAULT_KEY_SPACE_SIZE = 1000 + +DEFAULT_MEASURE_TIME_S = 20 + +EXPECTED_STATUS = 204 + +LOG = logging.getLogger(__name__) + +# Bodies are fixed per key, so they are built once per process and shared by +# every user in it, rather than rebuilt per user or per request. +_bodies: list[str] = [] + + +def _get_bodies(key_space_size: int) -> list[str]: + global _bodies + if len(_bodies) != key_space_size: + _bodies = [ + hashlib.sha256(f"{i}".encode()).hexdigest() for i in range(key_space_size) + ] + return _bodies + + +@events.init_command_line_parser.add_listener +def init_parser(parser): + parser.add_argument("--ca", help="Path to service certificate", required=True) + parser.add_argument("--cert", help="Path to client certificate", required=True) + parser.add_argument("--key", help="Path to client private key", required=True) + parser.add_argument( + "--key-space-size", + help="Number of distinct keys written to", + type=int, + default=DEFAULT_KEY_SPACE_SIZE, + ) + parser.add_argument( + "--measure-time-s", + help="Seconds to keep running once all users have spawned", + type=int, + default=DEFAULT_MEASURE_TIME_S, + ) + + +@events.init.add_listener +def on_init(environment, **_kwargs): + """Stop the run a fixed time after the last user has spawned. + + Locust's own --run-time starts counting when locust starts, so it includes + the ramp. --reset-stats discards the statistics gathered during the ramp + but does not extend the deadline, so the further the ramp is stretched, the + smaller the steady-state window becomes, until it disappears entirely. + Timing from spawning_complete instead keeps the measurement window the same + length whatever the spawn rate is. + """ + # In distributed mode the master tells the workers to fire this event too, + # but only the master decides when the run ends. + if isinstance(environment.runner, WorkerRunner): + return + + spawning_completed = False + + def stop_after_measurement_window(**_kwargs): + nonlocal spawning_completed + spawning_completed = True + # Statistics are reset by --reset-stats on this same event, so the + # window measured here is exactly the window reported. + gevent.spawn_later( + environment.parsed_options.measure_time_s, environment.runner.quit + ) + + def check_spawning_completed(**_kwargs): + # Reaching the end without spawning having completed means the run was + # ended by the --run-time backstop mid-ramp. The statistics then + # describe a partial ramp, but still look like a plausible result, so + # say so and exit non-zero rather than reporting them. + if not spawning_completed: + LOG.error( + "Run ended before all users had spawned. " + "The statistics do not describe steady state." + ) + environment.process_exit_code = 1 + + environment.events.spawning_complete.add_listener(stop_after_measurement_window) + environment.events.quitting.add_listener(check_spawning_completed) + + +class BlockingWriter(FastHttpUser): + # Send the next request as soon as the previous response arrives. Each + # response already waits for commit, so no additional pacing is wanted. + wait_time = constant(0) + + # Verify the service certificate, rather than skipping verification as + # FastHttpUser does by default. + insecure = False + + def __init__(self, environment): + super().__init__(environment) + self.key_space_size = environment.parsed_options.key_space_size + self.bodies = _get_bodies(self.key_space_size) + + def ssl_context_factory(self): + opts = self.environment.parsed_options + context = ssl.create_default_context(cafile=opts.ca) + context.load_cert_chain(certfile=opts.cert, keyfile=opts.key) + return context + + @task + def blocking_write(self): + index = random.randrange(self.key_space_size) + with self.client.put( + f"/records/blocking/{index}", + data=self.bodies[index], + headers={"content-type": "text/plain"}, + name=REQUEST_NAME, + catch_response=True, + ) as response: + # Anything other than the expected status is a failure, including + # the 5xx returned when a transaction is invalidated, and the 0 + # reported when the connection itself failed. + if response.status_code == EXPECTED_STATUS: + response.success() + else: + response.failure(f"Unexpected status {response.status_code}") diff --git a/tests/infra/key_space.py b/tests/infra/key_space.py new file mode 100644 index 000000000000..d7aebcaa7ec2 --- /dev/null +++ b/tests/infra/key_space.py @@ -0,0 +1,31 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the Apache 2.0 License. + +import hashlib +import http + +from loguru import logger as LOG + +import infra.node + + +def key_body(key: str) -> str: + """The value written for a given key. Fixed per key, so that a write is + idempotent and the value can be recomputed without storing it.""" + return hashlib.sha256(key.encode()).hexdigest() + + +def create_and_fill_key_space(size: int, primary: infra.node.Node) -> list[str]: + LOG.info(f"Creating and filling key space of size {size}") + space = [f"{i}" for i in range(size)] + mapping = {key: key_body(key) for key in space} + with primary.client("user0") as c: + r = c.post("/records", mapping) + assert r.status_code == http.HTTPStatus.NO_CONTENT, r + # Quick sanity check + for j in [0, -1]: + r = c.get(f"/records/{space[j]}") + assert r.status_code == http.HTTPStatus.OK, r + assert r.body.text() == mapping[space[j]], r + LOG.info("Key space created and filled") + return space