Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 53 additions & 36 deletions bench-orchestrator/bench_orchestrator/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ def temporary_ingest_output_dir(enabled: bool):


def backend_ingest_output_path(temp_dir: Path | None, index: int, backend: Engine) -> Path | None:
"""Return the ingest JSONL path a backend should write, if enabled."""
"""Return the ingest JSONL path a single benchmark run should write, if enabled."""
if temp_dir is None:
return None
return temp_dir / f"{index:02d}-{backend.value}.jsonl"
Expand All @@ -152,6 +152,19 @@ def write_combined_ingest_output(output_path: Path, input_paths: list[Path]) ->
output.write(line)


def drop_os_caches() -> None:
"""Try to drop OS caches and "sync" if runners have passwordless sudo"""
try:
subprocess.run(["sync"], check=True)
subprocess.run(
["sudo", "-n", "sh", "-c", "echo 3 > /proc/sys/vm/drop_caches"],
check=True,
capture_output=True,
)
except (OSError, subprocess.CalledProcessError):
pass


def write_result_line(line: str, store_writer, compatibility_file) -> None:
"""Write a raw result line to the run store and optional compatibility output."""
store_writer(line)
Expand Down Expand Up @@ -347,42 +360,46 @@ def run(
temporary_ingest_output_dir(ingest_output is not None) as ingest_temp_dir,
):
ingest_output_parts: list[Path] = []
for backend_idx, (backend, backend_targets) in enumerate(backend_groups.items()):
run_idx = 0
for backend, backend_targets in backend_groups.items():
executor = BenchmarkExecutor(binary_paths[backend], backend, verbose=verbose)
backend_formats = [target.format for target in backend_targets]
backend_ingest_output = backend_ingest_output_path(ingest_temp_dir, backend_idx, backend)

try:
results = executor.run(
benchmark=benchmark,
formats=backend_formats,
queries=query_list,
exclude_queries=exclude_list,
iterations=iterations,
options=bench_opts,
track_memory=track_memory,
samply=samply,
sample_rate=sample_rate,
tracing=tracing,
runner=runner,
ingest_output=backend_ingest_output,
on_result=lambda line, store_writer=ctx.write_raw_json, compatibility=compatibility_file: (
write_result_line(
line,
store_writer,
compatibility,
)
),
)
if backend_ingest_output is not None:
ingest_output_parts.append(backend_ingest_output)
console.print(f"[green]{backend.value}: {len(results)} results[/green]")
except RuntimeError as exc:
ctx.metadata.partial = True
if strict_failures:
raise
console.print(f"[red]{backend.value} failed: {exc}[/red]")
soft_failures.append(str(exc))
for target in backend_targets:
part_ingest_output = backend_ingest_output_path(ingest_temp_dir, run_idx, backend)
run_idx += 1

drop_os_caches()

try:
results = executor.run(
benchmark=benchmark,
formats=[target.format],
queries=query_list,
exclude_queries=exclude_list,
iterations=iterations,
options=bench_opts,
track_memory=track_memory,
samply=samply,
sample_rate=sample_rate,
tracing=tracing,
runner=runner,
ingest_output=part_ingest_output,
on_result=lambda line, store_writer=ctx.write_raw_json, compatibility=compatibility_file: (
write_result_line(
line,
store_writer,
compatibility,
)
),
)
if part_ingest_output is not None:
ingest_output_parts.append(part_ingest_output)
console.print(f"[green]{target}: {len(results)} results[/green]")
except RuntimeError as exc:
ctx.metadata.partial = True
if strict_failures:
raise
console.print(f"[red]{target} failed: {exc}[/red]")
soft_failures.append(str(exc))

if ingest_output is not None:
write_combined_ingest_output(ingest_output, ingest_output_parts)
Expand Down
2 changes: 2 additions & 0 deletions bench-orchestrator/tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@ def test_run_writes_compatibility_results_output(tmp_path, monkeypatch) -> None:

monkeypatch.setattr(cli_module, "ResultStore", lambda: run_store)
monkeypatch.setattr(cli_module.BenchmarkBuilder, "get_binary_path", lambda self, backend: binary_path)
monkeypatch.setattr(cli_module, "drop_os_caches", lambda: None)

def fake_run(self, **kwargs):
kwargs["on_result"](sample_line)
Expand Down Expand Up @@ -119,6 +120,7 @@ def test_run_combines_ingest_output_per_backend(tmp_path, monkeypatch) -> None:

monkeypatch.setattr(cli_module, "ResultStore", lambda: run_store)
monkeypatch.setattr(cli_module.BenchmarkBuilder, "get_binary_path", lambda self, backend: binary_paths[backend])
monkeypatch.setattr(cli_module, "drop_os_caches", lambda: None)

seen_backend_paths = []

Expand Down
Loading