From 576dd6ed62a75177a7713336a449d5db84343f2d Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:18:29 -0400 Subject: [PATCH 1/6] fix(asap-tools): read a stopped client's truncated results A client stopped mid-run leaves its gzipped JSONL unterminated; compare_arms now keeps the complete lines before the cut. Co-Authored-By: Claude Opus 5.5 --- .../multi_experiment/compare_arms.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py b/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py index dc7bf986..5d33e6f9 100644 --- a/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py +++ b/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py @@ -26,8 +26,17 @@ def read_jsonl(path): - with gzip.open(path, "rt") as f: - return [json.loads(line) for line in f if line.strip()] + """The complete lines of a gzipped JSONL file. A client stopped mid-run + leaves the gzip stream unterminated: keep what precedes the cut.""" + rows = [] + try: + with gzip.open(path, "rt") as f: + for line in f: + if line.endswith("\n"): + rows.append(json.loads(line)) + except EOFError: + pass + return rows def sql_labels(sql): From 550e3faa7fa79ec6e51dc9ed9876da8ec433ea9a Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:29:40 -0400 Subject: [PATCH 2/6] fix(asap-tools): monitor only the ClickHouse server in ClickHouse runs remote_monitor matched processes by "clickhouse", which also matched a paced feeder's `--sink clickhouse` on the same node: the feeder's CPU was summed into ClickHouse's, and when the feed ended the monitor took it for ClickHouse exiting and stopped the query client mid-run. The pattern clickhouse.server matches the server in Docker and bare metal only. Co-Authored-By: Claude Opus 5.5 --- asap-tools/experiments/constants.py | 4 +++ .../services/remote_monitor_service.py | 4 +-- .../tests/test_remote_monitor_keywords.py | 27 +++++++++++++++++-- 3 files changed, 31 insertions(+), 4 deletions(-) diff --git a/asap-tools/experiments/constants.py b/asap-tools/experiments/constants.py index 6c7ee31a..ded60da9 100644 --- a/asap-tools/experiments/constants.py +++ b/asap-tools/experiments/constants.py @@ -14,6 +14,10 @@ QUERY_ENGINE_RS_BINARY_NAME = "query_engine_rust" QUERY_ENGINE_RS_FP_BINARY_NAME = "query_engine_rust_fp" QUERY_ENGINE_RS_CONTAINER_NAME = "sketchdb-queryengine-rust" +# remote_monitor's grep -E pattern for the ClickHouse server: `clickhouse-server` +# in Docker, `clickhouse server` bare metal. Plain "clickhouse" also matches a +# paced feeder's `--sink clickhouse`, whose exit would end the run. +CLICKHOUSE_SERVER_PROCESS_KEYWORD = "clickhouse.server" PROMETHEUS_THROUGHPUT_POLLING_INTERVAL = 5 PROMETHEUS_HEALTH_POLLING_INTERVAL = 5 diff --git a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py index d054d896..219acd70 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -90,7 +90,7 @@ def start( else: keywords = [constants.QUERY_ENGINE_RS_PROCESS_KEYWORD] else: - keywords = ["clickhouse"] + keywords = [constants.CLICKHOUSE_SERVER_PROCESS_KEYWORD] else: keywords = list(config_keywords) @@ -258,7 +258,7 @@ def start_clickhouse_ingest_monitor( (``constants.INGEST_MONITOR_STOP_FILE``) is created, or SIGTERM/SIGINT, then writes ``monitor_output_file`` under ``remote_monitor_output/``. """ - keywords = ["clickhouse"] + keywords = [constants.CLICKHOUSE_SERVER_PROCESS_KEYWORD] cmd = ( "python3 -u remote_monitor.py " "--execution_mode ingest " diff --git a/asap-tools/experiments/tests/test_remote_monitor_keywords.py b/asap-tools/experiments/tests/test_remote_monitor_keywords.py index 0cd080cd..71d070b3 100644 --- a/asap-tools/experiments/tests/test_remote_monitor_keywords.py +++ b/asap-tools/experiments/tests/test_remote_monitor_keywords.py @@ -1,8 +1,11 @@ """Tests that remote_monitor.py receives its keyword list as one clean argument.""" +import re import shlex import unittest +import constants + from experiment_utils.services.remote_monitor_service import RemoteMonitorService @@ -71,13 +74,33 @@ def test_remote_provider_keeps_escaped_quotes_for_ssh(self): def test_clickhouse_ingest_local_provider_gets_unescaped_keywords(self): self.assertEqual( - clickhouse_ingest_keywords_seen_by_monitor(remote=False), "clickhouse" + clickhouse_ingest_keywords_seen_by_monitor(remote=False), + "clickhouse.server", ) def test_clickhouse_ingest_remote_provider_keeps_escaped_quotes_for_ssh(self): self.assertEqual( - clickhouse_ingest_keywords_seen_by_monitor(remote=True), "clickhouse" + clickhouse_ingest_keywords_seen_by_monitor(remote=True), + "clickhouse.server", + ) + + +class ClickHouseKeywordTest(unittest.TestCase): + def test_matches_the_server_but_not_a_feeder_into_it(self): + # "clickhouse" also matched the paced feeder's `--sink clickhouse`; + # when the feed ended, the monitor took it for ClickHouse exiting and + # stopped the query client mid-run. + pattern = re.compile(constants.CLICKHOUSE_SERVER_PROCESS_KEYWORD) + for server in ( + "/usr/bin/clickhouse-server --config-file=/etc/clickhouse-server/config.xml", + "/usr/bin/clickhouse server --config-file=/etc/clickhouse-server/config.xml", + ): + self.assertTrue(pattern.search(server), server) + feeder = ( + "./target/release/paced_feeder --sink clickhouse " + "--url http://localhost:8123/?database=default --metric data" ) + self.assertIsNone(pattern.search(feeder)) if __name__ == "__main__": From 09c5eb40f753c7b4af969f7da3afa0768f7a277f Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:32:27 -0400 Subject: [PATCH 3/6] feat(paced-feeder): stop at the next tick on SIGTERM, writing stats A run can now stop its feeders when its queries are done instead of waiting out their duration, without losing --stats-out: SIGTERM ends the feed before the next tick and the stats cover the ticks sent. Achieved rows/s spans those ticks, not the planned duration. Co-Authored-By: Claude Opus 5.5 --- .../data-sources/paced-feeder/src/main.rs | 28 +++++++++++++++---- 1 file changed, 23 insertions(+), 5 deletions(-) diff --git a/asap-tools/data-sources/paced-feeder/src/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index 1a106ba2..d60a41a6 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -139,7 +139,7 @@ struct RunStats { rows_sent: usize, bytes_sent: usize, /// Rows sent divided by wall-clock time from the first tick's schedule to - /// the later of the last tick's end and the run's nominal end. + /// the later of the last tick's end and the end of its second. achieved_rows_per_sec: f64, /// Ticks whose work took longer than one second. overrun_ticks: usize, @@ -455,6 +455,17 @@ async fn main() -> Result<()> { workload.rows_per_tick(), ); + // SIGTERM ends the feed at the next tick, still writing its stats: a run + // stops its feeders when its queries are done, before their duration. + let stop = Arc::new(tokio::sync::Notify::new()); + let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; + let on_sigterm = Arc::clone(&stop); + tokio::spawn(async move { + sigterm.recv().await; + // Stores a permit, so a tick already sending stops at the next one. + on_sigterm.notify_one(); + }); + let base = wait_for_wall_clock(start_ms).await; let mut ticks = Vec::with_capacity(duration_secs as usize); let mut errors: Vec = Vec::new(); @@ -462,7 +473,13 @@ async fn main() -> Result<()> { for tick in 0..duration_secs { let scheduled = base + Duration::from_secs(tick); - tokio::time::sleep_until(scheduled.into()).await; + tokio::select! { + _ = tokio::time::sleep_until(scheduled.into()) => {} + _ = stop.notified() => { + eprintln!("SIGTERM: stopping after {tick} ticks"); + break; + } + } let started = Instant::now(); let (mut rows, mut bytes) = (0, 0); @@ -532,9 +549,10 @@ async fn main() -> Result<()> { } } - // A run that keeps up still spans the full duration, even though the last - // tick's work ends before its second does. - let wall_secs = base.elapsed().as_secs_f64().max(duration_secs as f64); + // A run that keeps up still spans its ticks' seconds, even though the last + // tick's work ends before its second does. A SIGTERM'd run spans the + // ticks it sent, not its duration. + let wall_secs = base.elapsed().as_secs_f64().max(ticks.len() as f64); let rows_sent: usize = ticks.iter().map(|t| t.rows).sum(); let grouped = args.workload_table.is_none(); let run = RunStats { From eb0b29cddd765b168a53a95235349ccf989f9afd Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:33:11 -0400 Subject: [PATCH 4/6] test(asap-query-engine): give the avg-of-quantiles test its own port It shared 19429 with e2e_topk_of_sum_ranks_groups_from_one_planned_heap; run in parallel, one of them got connection refused. Co-Authored-By: Claude Opus 5.5 --- asap-query-engine/tests/e2e_precompute_equivalence.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/asap-query-engine/tests/e2e_precompute_equivalence.rs b/asap-query-engine/tests/e2e_precompute_equivalence.rs index b0a1f3b0..1a595dc7 100644 --- a/asap-query-engine/tests/e2e_precompute_equivalence.rs +++ b/asap-query-engine/tests/e2e_precompute_equivalence.rs @@ -2070,7 +2070,7 @@ async fn e2e_avg_of_per_series_quantile_over_time() { }) .collect(); let (engine, query) = NativeDagScenario { - port: 19429, + port: 19437, metric, query, aggregation_configs: vec![config], From 7f920792f9b49069b36375bb7e22576cb1c81410 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:34:50 -0400 Subject: [PATCH 5/6] fix(asap-tools): end fed ClickHouse runs with their queries The run waited out its feed and the feed's end could cut the queries short. Now it stops the feeders once the queries are done (SIGTERM, so they still write stats), and a feed's duration only bounds a hung run; the random-workload configs set it to 10x the query phase for every arm. The feeder runs on the next node, as in experiment_run_e2e.py, so it no longer takes ClickHouse's cores. Co-Authored-By: Claude Opus 5.5 --- .../fake_metric/random_workload_artifacts.py | 8 +++- .../experiments/experiment_run_clickhouse.py | 43 ++++++++++--------- .../tests/test_experiment_run_clickhouse.py | 32 ++++++-------- .../tests/test_random_workload_artifacts.py | 3 ++ 4 files changed, 46 insertions(+), 40 deletions(-) diff --git a/asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py b/asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py index 1ec9b934..75181522 100644 --- a/asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py +++ b/asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py @@ -27,6 +27,10 @@ ACCURACY_SLA = 0.95 PILOT_SUPPORT = "common-promql" PILOT_DURATION_MS = 10 * 60 * 1000 +# Runs stop their feeders once the queries are done, so a feed's duration +# only bounds a run that hangs: this many times the scheduled query phase, +# past the slowest pilot arm (raw ClickHouse, ~4x). +FEED_BOUND_QUERY_PHASES = 10 QUERY_FRESHNESS_DELAY_SECS = 10 # Relative to EXPERIMENTS_DIR, where the runners read configs and SQL. CONFIG_DIR = "config/experiment_type/random_workload" @@ -139,6 +143,7 @@ def pilot_configs(workload, pilot_dir): lookback_secs = max(q["lookback_ms"] for q in queries) // 1000 starting_delay = lookback_secs + QUERY_FRESHNESS_DELAY_SECS duration = PILOT_DURATION_MS // 1000 + feed_secs = starting_delay + FEED_BOUND_QUERY_PHASES * duration seed = workload["data_facts"]["feeder_seed"] table = pilot_dir / "eval_table.json" out_dir = Path(CONFIG_DIR) / workload["id"] @@ -149,6 +154,7 @@ def pilot_configs(workload, pilot_dir): starting_delay, duration, QUERY_FRESHNESS_DELAY_SECS, + feeder_duration_secs=feed_secs, prebuilt_output_dir=str(pilot_dir), ) # One monitoring tool per run, so VictoriaMetrics is its own baseline run. @@ -205,7 +211,7 @@ def pilot_configs(workload, pilot_dir): "workload_table": str(table.resolve()), "metrics": list(metrics), "seed": seed, - "duration_secs": starting_delay + duration + sbw.FEEDER_SLACK_SECS, + "duration_secs": feed_secs, } for mode, init, name in ( (sbw.BASELINE, "init.sql", "clickhouse.yaml"), diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 8934b04d..041e431d 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -380,8 +380,9 @@ def _start_clickhouse_feed( start_ms = feeder_service.start( config=feeder_config, sinks={ - SINK_CLICKHOUSE: f"http://localhost:{clickhouse_http_port}/" - f"?database={CLICKHOUSE_DATABASE}" + # The feeder runs on the next node (see main()). + SINK_CLICKHOUSE: f"http://{provider.get_node_ip(node_offset)}:" + f"{clickhouse_http_port}/?database={CLICKHOUSE_DATABASE}" }, experiment_output_dir=os.path.join(experiment_root_output_dir, experiment_mode), local_experiment_dir=os.path.join(local_experiment_root_dir, experiment_mode), @@ -406,16 +407,16 @@ def _start_clickhouse_feed( return start_ms -def _finish_clickhouse_feed( - feeder_service: PacedFeederService, start_ms: int, duration_secs: int -) -> None: - """Let the feeders run to the end of their feed, so they write their - stats (they write none when killed), then stop any that overstay.""" - remaining_secs = start_ms / 1000 + duration_secs - time.time() - if not feeder_service.wait_for_exit( - max(remaining_secs, 0) + PACED_FEEDER_STOP_TIMEOUT_SECS - ): - feeder_service.stop() +def _finish_clickhouse_feed(feeder_service: PacedFeederService) -> None: + """Stop the feeders once the queries are done: the run ends with its + queries, however slow, and its feed's duration is only an upper bound. + SIGTERM lets a feeder finish its tick and write its stats.""" + feeder_service.stop() + if not feeder_service.wait_for_exit(PACED_FEEDER_STOP_TIMEOUT_SECS): + print( + f"Warning: paced_feeder still running {PACED_FEEDER_STOP_TIMEOUT_SECS} s " + "after SIGTERM; the next mode's feed start refuses to run over it" + ) def _run_clickhouse_ingest_with_monitor( @@ -645,8 +646,14 @@ def main(cfg: DictConfig) -> None: yaml.safe_load(f), minimum_experiment_running_time, ) - # Next to ClickHouse, so the feed crosses no network. - paced_feeder_service = PacedFeederService(provider, [node_offset]) + # On its own node, as in experiment_run_e2e.py, so the feeder neither + # takes ClickHouse's cores nor shows up as its CPU. + if num_nodes < 1: + raise ValueError( + "dataset.paced_feeder runs the feeder on its own node: " + "providers.*.num_nodes must be at least 1" + ) + paced_feeder_service = PacedFeederService(provider, [node_offset + 1]) if not no_teardown: # Otherwise a failed run leaves feeders inserting into the next # run's tables. @@ -883,7 +890,7 @@ def generate_controller_output() -> None: else: # --- baseline mode --- if paced_feeder_service is not None: - feed_start_ms = _start_clickhouse_feed( + _start_clickhouse_feed( provider=provider, node_offset=node_offset, data_loader=data_loader, @@ -918,11 +925,7 @@ def generate_controller_output() -> None: monitor_interval_seconds=monitor_interval_seconds, ) if paced_feeder_service is not None: - _finish_clickhouse_feed( - paced_feeder_service, - feed_start_ms, - paced_feeder_config["duration_secs"], - ) + _finish_clickhouse_feed(paced_feeder_service) sync.rsync_experiment_data( provider, diff --git a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py index 743bdb01..65ab509f 100644 --- a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py +++ b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py @@ -172,8 +172,10 @@ def start(self, calls, exited=True): with open(config_path, "w") as f: yaml.safe_dump(aligned_client_config(60), f) + provider = Mock() + provider.get_node_ip.return_value = "10.10.1.1" start_ms = _start_clickhouse_feed( - provider=Mock(), + provider=provider, node_offset=0, data_loader=calls.loader, feeder_service=calls.feeder, @@ -209,7 +211,8 @@ def test_tables_are_created_before_the_feed_and_queries_follow_it(self): self.assertEqual(start_kwargs["align_secs"], 60) self.assertEqual( start_kwargs["sinks"], - {"clickhouse": "http://localhost:8123/?database=default"}, + # From the feeder's own node to ClickHouse's. + {"clickhouse": "http://10.10.1.1:8123/?database=default"}, ) self.assertEqual(start_kwargs["experiment_output_dir"], "/exp/baseline") self.assertEqual( @@ -229,23 +232,14 @@ def test_feeders_that_outlive_pkill_stop_the_run(self): class FinishClickHouseFeedTest(unittest.TestCase): - def finish(self, exited): - feeder = Mock() - feeder.wait_for_exit.return_value = exited - with patch("experiment_run_clickhouse.time.time", return_value=1700000600): - _finish_clickhouse_feed(feeder, 1700000000000, duration_secs=720) - return feeder - - def test_waits_out_the_feed_so_feeders_write_their_stats(self): - # A killed feeder writes no stats; 120 s of feed remain. - feeder = self.finish(exited=True) - feeder.wait_for_exit.assert_called_once_with( - 120 + PACED_FEEDER_STOP_TIMEOUT_SECS - ) - feeder.stop.assert_not_called() - - def test_stops_feeders_that_overstay(self): - self.finish(exited=False).stop.assert_called_once() + def test_stops_the_feed_once_the_queries_are_done(self): + # Waiting out the feed's duration let it, not the queries, end the run; + # SIGTERM still lets feeders write their stats. + calls = Mock() + calls.wait_for_exit.return_value = True + _finish_clickhouse_feed(calls) + self.assertEqual([c[0] for c in calls.mock_calls], ["stop", "wait_for_exit"]) + calls.wait_for_exit.assert_called_once_with(PACED_FEEDER_STOP_TIMEOUT_SECS) class FedClickHouseConfigValidationTest(unittest.TestCase): diff --git a/asap-tools/experiments/tests/test_random_workload_artifacts.py b/asap-tools/experiments/tests/test_random_workload_artifacts.py index 35ef2be6..9f192828 100644 --- a/asap-tools/experiments/tests/test_random_workload_artifacts.py +++ b/asap-tools/experiments/tests/test_random_workload_artifacts.py @@ -157,6 +157,9 @@ def test_every_arm_feeds_the_same_data_and_schedule(self): (feeder["workload_table"], feeder["seed"], feeder["metrics"]), (str((PILOT_DIR / "eval_table.json").resolve()), 862, ["data"]), ) + # Runs stop feeders when their queries are done; the duration + # only bounds a hung run, past a slow arm's lagging queries. + self.assertEqual(feeder["duration_secs"], 1210 + 10 * 600) # The longest lookback (20 min) plus the freshness delay, then 10 min. schedules = [ (g["repetition_delay_ms"], g["client_options"]) From 17063a82c3598d8e4e0741f08a439f0215b7ccb6 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:35:35 -0400 Subject: [PATCH 6/6] feat(asap-tools): split each arm's CPU into warm-up and queries compare_arms reports CPU-seconds over the whole monitored run and its split at the client's starting delay (warm-up, ingest only) and after it (queries plus any lag), with the monitored length. The per-second samples stay as recorded. Co-Authored-By: Claude Opus 5.5 --- .../multi_experiment/compare_arms.py | 27 ++++++++++++++----- 1 file changed, 21 insertions(+), 6 deletions(-) diff --git a/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py b/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py index 5d33e6f9..7578016f 100644 --- a/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py +++ b/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py @@ -85,13 +85,17 @@ def __init__(self, mode_dir): client = mode_dir.parent / "controller_client_configs" / f"{mode_dir.name}.yaml" groups = yaml.safe_load(client.read_text())["query_groups"] self.interval_secs = [g["repetition_delay_ms"] / 1000 for g in groups] + # Queries start after this; until then the systems only ingest. + self.warmup_secs = min(g["client_options"]["starting_delay"] for g in groups) self.monitor = mode_dir / "remote_monitor_output" / "monitor_output.json" hydra = yaml.safe_load((mode_dir.parent / "hydra_config.yaml").read_text()) self.monitor_interval_secs = float(hydra["flow"]["monitor_interval_seconds"]) def resources(self): - """(total CPU-seconds, peak memory bytes), summed over monitored - processes at each sample.""" + """CPU-seconds over the whole monitored run, its warm-up and the rest + (queries and any lag), and peak memory bytes; each sample sums the + monitored processes. The monitor starts with the query client, so + the warm-up is its first `warmup_secs`.""" processes = json.loads(self.monitor.read_text()).values() n = max(len(p["cpu_percent"]) for p in processes) cpu, mem = [0.0] * n, [0.0] * n @@ -100,7 +104,15 @@ def resources(self): cpu[t] += c for t, m in enumerate(p["memory_info"]): mem[t] += m - return sum(cpu) / 100 * self.monitor_interval_secs, max(mem) + secs = self.monitor_interval_secs + warm = int(self.warmup_secs / secs) + return { + "monitored_secs": n * secs, + "cpu_secs": sum(cpu) / 100 * secs, + "warmup_cpu_secs": sum(cpu[:warm]) / 100 * secs, + "query_cpu_secs": sum(cpu[warm:]) / 100 * secs, + "peak_memory_bytes": max(mem), + } def percentile(xs, p): @@ -164,11 +176,14 @@ def report(arms, reference): f"missing series {missing}, empty reference windows {empty}" ) - print("\n== Resources (whole monitored run)") + print("\n== Resources: CPU-seconds (whole run = warm-up + queries), peak memory") for arm_name, arm in arms.items(): - cpu_secs, peak = arm.resources() + r = arm.resources() print( - f" {arm_name:16s} CPU {cpu_secs:9.0f} s peak memory {peak / 2**30:5.2f} GiB" + f" {arm_name:16s} {r['monitored_secs']:6.0f} s monitored CPU " + f"{r['cpu_secs']:7.0f} = {r['warmup_cpu_secs']:6.0f} + " + f"{r['query_cpu_secs']:6.0f} peak memory " + f"{r['peak_memory_bytes'] / 2**30:5.2f} GiB" )