from __future__ import annotations
import argparse
import dataclasses
import datetime as dt
import difflib
import json
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Iterable
ROOT = Path(__file__).resolve().parents[2]
BENCH_ROOT = Path(__file__).resolve().parent
FIXTURE_ROOT = BENCH_ROOT / "fixtures"
DEFAULT_ANTLR_JAR = Path("/tmp/antlr-cleanroom/tools/antlr-4.13.2-complete.jar")
DEFAULT_GRAMMARS_V4 = Path("/tmp/antlr-cleanroom/grammars-v4")
GO_ANTLR_RUNTIME = "v4.13.1"
@dataclasses.dataclass(frozen=True)
class LanguageSpec:
name: str
grammar_rel: Path
grammar_files: tuple[str, ...]
lexer_name: str
parser_name: str
rust_lexer_module: str
rust_parser_module: str
rust_lexer_type: str
rust_parser_type: str
rust_entry: str
python_entry: str
go_entry: str
tree_sitter_name: str
python_support: tuple[Path, ...] = ()
go_support: tuple[Path, ...] = ()
LANGUAGES: dict[str, LanguageSpec] = {
"kotlin": LanguageSpec(
name="kotlin",
grammar_rel=Path("kotlin/kotlin"),
grammar_files=("KotlinLexer.g4", "KotlinParser.g4", "UnicodeClasses.g4"),
lexer_name="KotlinLexer",
parser_name="KotlinParser",
rust_lexer_module="kotlin_lexer",
rust_parser_module="kotlin_parser",
rust_lexer_type="KotlinLexer",
rust_parser_type="KotlinParser",
rust_entry="kotlin_file",
python_entry="kotlinFile",
go_entry="KotlinFile",
tree_sitter_name="kotlin",
),
"csharp": LanguageSpec(
name="csharp",
grammar_rel=Path("csharp/v7"),
grammar_files=("CSharpLexer.g4", "CSharpParser.g4"),
lexer_name="CSharpLexer",
parser_name="CSharpParser",
rust_lexer_module="c_sharp_lexer",
rust_parser_module="c_sharp_parser",
rust_lexer_type="CSharpLexer",
rust_parser_type="CSharpParser",
rust_entry="compilation_unit",
python_entry="compilation_unit",
go_entry="Compilation_unit",
tree_sitter_name="csharp",
python_support=(
Path("csharp/v7/Python3/CSharpLexerBase.py"),
Path("csharp/v7/Python3/CSharpParserBase.py"),
),
go_support=(
Path("csharp/v7/Go/CSharpLexerBase.go"),
Path("csharp/v7/Go/CSharpParserBase.go"),
),
),
"java": LanguageSpec(
name="java",
grammar_rel=Path("java/java"),
grammar_files=("JavaLexer.g4", "JavaParser.g4"),
lexer_name="JavaLexer",
parser_name="JavaParser",
rust_lexer_module="java_lexer",
rust_parser_module="java_parser",
rust_lexer_type="JavaLexer",
rust_parser_type="JavaParser",
rust_entry="compilation_unit",
python_entry="compilationUnit",
go_entry="CompilationUnit",
tree_sitter_name="java",
),
"trino": LanguageSpec(
name="trino",
grammar_rel=Path("sql/trino"),
grammar_files=("TrinoLexer.g4", "TrinoParser.g4"),
lexer_name="TrinoLexer",
parser_name="TrinoParser",
rust_lexer_module="trino_lexer",
rust_parser_module="trino_parser",
rust_lexer_type="TrinoLexer",
rust_parser_type="TrinoParser",
rust_entry="parse",
python_entry="parse",
go_entry="Parse",
tree_sitter_name="sql",
),
}
RUNTIMES = ("rust-antlr", "python-antlr", "go-antlr", "tree-sitter")
PHASES = ("parse", "lex")
@dataclasses.dataclass(frozen=True)
class Fixture:
language: str
path: Path
source: str
license: str
description: str
phases: tuple[str, ...] = ("parse", "lex")
@property
def name(self) -> str:
return self.path.name
@property
def abs_path(self) -> Path:
return FIXTURE_ROOT / self.path
@dataclasses.dataclass(frozen=True)
class Measurement:
language: str
fixture: str
runtime: str
min_ns: int
avg_ns: int
bytes: int
source: str
license: str
description: str
def run(
cmd: list[str],
*,
cwd: Path | None = None,
env: dict[str, str] | None = None,
quiet: bool = False,
) -> subprocess.CompletedProcess[str]:
if not quiet:
print("+ " + " ".join(cmd))
completed = subprocess.run(
cmd,
cwd=cwd,
env=env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
if completed.returncode != 0:
if completed.stdout:
print(completed.stdout, file=sys.stderr)
if completed.stderr:
print(completed.stderr, file=sys.stderr)
raise subprocess.CalledProcessError(
completed.returncode,
cmd,
output=completed.stdout,
stderr=completed.stderr,
)
return completed
def parse_csv(value: str, allowed: Iterable[str], label: str) -> list[str]:
allowed_set = set(allowed)
items = [item.strip() for item in value.split(",") if item.strip()]
unknown = [item for item in items if item not in allowed_set]
if unknown:
raise SystemExit(f"unknown {label}: {', '.join(unknown)}")
return items
def require_path(path: Path, label: str) -> None:
if not path.exists():
raise SystemExit(f"{label} does not exist: {path}")
def load_fixtures(languages: set[str], phase: str) -> list[Fixture]:
manifest = json.loads((FIXTURE_ROOT / "manifest.json").read_text())
fixtures = []
for item in manifest["fixtures"]:
language = str(item["language"])
phases = tuple(str(value) for value in item.get("phases", PHASES))
if language not in languages or phase not in phases:
continue
fixture = Fixture(
language=language,
path=Path(str(item["path"])),
source=str(item["source"]),
license=str(item["license"]),
description=str(item["description"]),
phases=phases,
)
require_path(fixture.abs_path, f"fixture {fixture.path}")
fixtures.append(fixture)
return fixtures
def copy_grammar(spec: LanguageSpec, grammars_v4: Path, target: Path) -> None:
target.mkdir(parents=True, exist_ok=True)
source_dir = grammars_v4 / spec.grammar_rel
for grammar in spec.grammar_files:
source = source_dir / grammar
require_path(source, f"{spec.name} grammar")
shutil.copy2(source, target / grammar)
def transform_csharp_python(grammar_dir: Path) -> None:
for grammar in grammar_dir.glob("*.g4"):
text = grammar.read_text()
text = re.sub(r"!this\.", "not self.", text)
text = re.sub(r"this\.", "self.", text)
grammar.write_text(text)
def transform_csharp_go(grammar_dir: Path) -> None:
for grammar in grammar_dir.glob("*.g4"):
if grammar.name == "CSharpLexer.g4":
grammar.write_text(transform_go_lexer_actions(grammar.read_text()))
elif grammar.name == "CSharpParser.g4":
grammar.write_text(grammar.read_text().replace("this.", "p."))
def transform_go_lexer_actions(grammar: str) -> str:
def predicate_replacement(match: re.Match[str]) -> str:
body = match.group("body").replace("this.", "p.")
return "{" + body + "}" + match.group("suffix")
grammar = re.sub(
r"\{(?P<body>[^{}]*this\.[^{}]*)\}(?P<suffix>\s*\?)",
predicate_replacement,
grammar,
flags=re.DOTALL,
)
return grammar.replace("this.", "l.")
def transform_java(grammar_dir: Path) -> None:
parser = grammar_dir / "JavaParser.g4"
text = parser.read_text()
text = text.replace(" superClass = JavaParserBase;\n", "")
text = re.sub(
r"annotationFieldValue:\s*\{ this\.IsNotIdentifierAssign\(\) \}\? annotationValue\s*\|\s*identifier '=' annotationValue\s*;",
"annotationFieldValue:\n identifier '=' annotationValue\n | annotationValue\n ;",
text,
flags=re.MULTILINE,
)
text = text.replace(
"recordComponent (',' recordComponent)* { this.DoLastRecordComponent() }?",
"recordComponent (',' recordComponent)*",
)
parser.write_text(text)
def generate_antlr(
antlr_jar: Path,
spec: LanguageSpec,
grammar_dir: Path,
output_dir: Path,
language: str | None,
go_package: str | None = None,
) -> None:
output_dir.mkdir(parents=True, exist_ok=True)
cmd = ["java", "-jar", str(antlr_jar)]
if language is not None:
cmd.append(f"-Dlanguage={language}")
if language == "Go":
cmd.extend(["-package", go_package or "parser"])
cmd.extend(["-o", str(output_dir), "-Xexact-output-dir"])
cmd.extend(spec.grammar_files[:2])
run(cmd, cwd=grammar_dir)
def prepare_python_support(spec: LanguageSpec, grammars_v4: Path, py_gen_dir: Path) -> None:
for rel_path in spec.python_support:
source = grammars_v4 / rel_path
require_path(source, f"{spec.name} Python support")
target = py_gen_dir / source.name
shutil.copy2(source, target)
if spec.name == "csharp" and source.name == "CSharpLexerBase.py":
target.write_text(
target.read_text().replace(
'tok = CommonToken(type=CSharpLexer.SKIPPED_SECTION, text="".join(text))',
'tok = CommonToken(type=CSharpLexer.SKIPPED_SECTION)\n'
' tok.text = "".join(text)',
)
)
def prepare_go_support(
spec: LanguageSpec,
grammars_v4: Path,
go_parser_dir: Path,
package_name: str,
) -> None:
for rel_path in spec.go_support:
source = grammars_v4 / rel_path
require_path(source, f"{spec.name} Go support")
target = go_parser_dir / source.name
shutil.copy2(source, target)
target.write_text(target.read_text().replace("package parser", f"package {package_name}", 1))
def generate_rust_modules(
spec: LanguageSpec,
grammar_dir: Path,
interp_dir: Path,
rust_generated_dir: Path,
require_generated_parser: bool,
runtime_root: Path,
) -> None:
common = [
"cargo",
"run",
"--quiet",
"--release",
"--manifest-path",
str(runtime_root / "Cargo.toml"),
"--bin",
"antlr4-rust-gen",
"--",
"--out-dir",
str(rust_generated_dir),
]
run(
[
*common,
"--lexer",
str(interp_dir / f"{spec.lexer_name}.interp"),
"--grammar",
str(grammar_dir / spec.grammar_files[0]),
"--allow-unsupported-lexer-actions",
]
)
parser_cmd = [
*common,
"--parser",
str(interp_dir / f"{spec.parser_name}.interp"),
"--grammar",
str(grammar_dir / spec.grammar_files[1]),
]
if require_generated_parser:
parser_cmd.append("--require-generated-parser")
run(parser_cmd)
def write_rust_runner(
work_dir: Path,
specs: list[LanguageSpec],
runtime_root: Path,
rust_thin_lto: bool,
) -> Path:
runner = work_dir / "rust-runner"
generated = runner / "src" / "generated"
generated.mkdir(parents=True, exist_ok=True)
manifest = [
"[package]",
'name = "parse-bench-rust-runner"',
'version = "0.0.0"',
'edition = "2024"',
"",
"[features]",
"default = []",
'perf-counters = ["antlr-rust-runtime/perf-counters"]',
"",
"[dependencies]",
f'antlr-rust-runtime = {{ path = "{runtime_root}" }}',
"",
]
if rust_thin_lto:
manifest.extend(
[
"[profile.release]",
'lto = "thin"',
"codegen-units = 1",
"",
]
)
(runner / "Cargo.toml").write_text("\n".join(manifest))
modules = "\n".join(
f" pub mod {spec.rust_lexer_module};\n pub mod {spec.rust_parser_module};"
for spec in specs
)
arms = "\n".join(
f' "{spec.name}" => parse_{spec.name}(&src).map_err(|err| err.to_string())?,'
for spec in specs
)
lex_arms = "\n".join(f' "{spec.name}" => lex_{spec.name}(&src)?,' for spec in specs)
dump_arms = "\n".join(
f' "{spec.name}" => dump_tree_{spec.name}(&src, out)?,'
for spec in specs
)
stats_arms = "\n".join(
f' "{spec.name}" => prediction_stats_{spec.name}(&src).map_err(|err| err.to_string())?,'
for spec in specs
)
functions = "\n\n".join(rust_parse_function(spec) for spec in specs)
(runner / "src" / "main.rs").write_text(
f"""#![allow(clippy::print_stdout)]
use std::env;
use std::fs;
use std::hint::black_box;
use std::io::{{self, Write}};
use std::path::PathBuf;
use std::process::ExitCode;
use std::time::Instant;
use antlr4_runtime::{{CommonTokenStream, InputStream, Node, NodeKind, Parser}};
mod generated {{
#![allow(dead_code, unused_imports, unreachable_pub, unused_qualifications)]
{modules}
}}
{functions}
fn main() -> ExitCode {{
match run_main() {{
Ok(()) => ExitCode::SUCCESS,
Err(err) => {{
eprintln!("{{err}}");
ExitCode::from(1)
}}
}}
}}
fn run_main() -> Result<(), String> {{
let mut args = env::args().skip(1);
let mut language: Option<String> = None;
let mut input: Option<PathBuf> = None;
let mut dump_tree: Option<PathBuf> = None;
let mut phase = "parse".to_owned();
let mut iters = 1_usize;
let mut warmups = 0_usize;
while let Some(arg) = args.next() {{
match arg.as_str() {{
"--language" => language = args.next(),
"--input" => input = args.next().map(PathBuf::from),
"--dump-tree" => dump_tree = args.next().map(PathBuf::from),
"--phase" => phase = args.next().ok_or("missing value for --phase")?,
"--iters" => iters = parse_usize(args.next(), "--iters")?,
"--warmups" => warmups = parse_usize(args.next(), "--warmups")?,
other => return Err(format!("unknown argument: {{other}}")),
}}
}}
let language = language.ok_or("missing --language <name>")?;
let input = input.ok_or("missing --input <path>")?;
if iters == 0 {{
return Err("--iters must be greater than 0".to_owned());
}}
if phase != "parse" && phase != "lex" {{
return Err(format!("unsupported phase: {{phase}}"));
}}
let src = fs::read_to_string(&input)
.map_err(|err| format!("failed to read {{}}: {{err}}", input.display()))?;
if let Some(path) = dump_tree {{
if phase != "parse" {{
return Err("--dump-tree requires --phase parse".to_owned());
}}
let file = fs::File::create(&path)
.map_err(|err| format!("failed to create {{}}: {{err}}", path.display()))?;
let mut out = io::BufWriter::new(file);
dump_tree_once(&language, &src, &mut out)?;
out.flush()
.map_err(|err| format!("failed to flush {{}}: {{err}}", path.display()))?;
return Ok(());
}}
let collect_stats = env::var_os("ANTLR_PERF_DUMP").is_some();
for _ in 0..warmups {{
run_once(&phase, &language, &src)?;
}}
#[cfg(feature = "perf-counters")]
antlr4_runtime::reset_prediction_perf_counters();
let mut min_ns = u128::MAX;
let mut total_ns = 0_u128;
for _ in 0..iters {{
let started = Instant::now();
run_once(&phase, &language, &src)?;
let elapsed = started.elapsed().as_nanos();
min_ns = min_ns.min(elapsed);
total_ns += elapsed;
}}
let avg_ns = total_ns / iters as u128;
println!("min_ns={{min_ns}} avg_ns={{avg_ns}}");
#[cfg(feature = "perf-counters")]
if collect_stats {{
antlr4_runtime::dump_prediction_perf_counters();
}}
if collect_stats && phase == "parse" {{
let (context_stats, dfa_stats) = prediction_stats_once(&language, &src)?;
dump_prediction_context_stats(context_stats);
dump_parser_dfa_stats(dfa_stats);
}}
Ok(())
}}
fn parse_once(
language: &str,
src: &str,
) -> Result<(), String> {{
match language {{
{arms}
other => return Err(format!("unsupported language: {{other}}")),
}}
Ok(())
}}
fn lex_once(language: &str, src: &str) -> Result<(), String> {{
match language {{
{lex_arms}
other => return Err(format!("unsupported language: {{other}}")),
}}
Ok(())
}}
fn dump_tree_once(language: &str, src: &str, out: &mut dyn Write) -> Result<(), String> {{
match language {{
{dump_arms}
other => return Err(format!("unsupported language: {{other}}")),
}}
Ok(())
}}
fn run_once(phase: &str, language: &str, src: &str) -> Result<(), String> {{
match phase {{
"parse" => parse_once(language, src),
"lex" => lex_once(language, src),
other => Err(format!("unsupported phase: {{other}}")),
}}
}}
fn dump_node<S: AsRef<str>>(
out: &mut dyn Write,
tree: Node<'_>,
rule_names: &[S],
depth: usize,
) -> io::Result<()> {{
let pad = " ".repeat(depth);
match tree.kind() {{
NodeKind::Rule => {{
let rule = tree.as_rule().expect("rule node kind checked");
let name = rule_names
.get(rule.rule_index())
.map_or("<?>", |name| name.as_ref());
writeln!(out, "{{pad}}Rule({{name}}, children={{}})", rule.child_count())?;
for child in rule.children() {{
dump_node(out, child, rule_names, depth + 1)?;
}}
}}
NodeKind::Terminal => writeln!(
out,
"{{pad}}Term({{}})",
tree_text(tree.as_terminal().expect("terminal node kind checked").text())
)?,
NodeKind::Error => writeln!(
out,
"{{pad}}Err({{}})",
tree_text(tree.as_error().expect("error node kind checked").text())
)?,
}}
Ok(())
}}
fn tree_text(text: &str) -> String {{
let mut escaped = String::new();
escaped.push('"');
for ch in text.chars() {{
escaped.extend(ch.escape_unicode());
}}
escaped.push('"');
escaped
}}
fn prediction_stats_once(
language: &str,
src: &str,
) -> Result<(antlr4_runtime::PredictionContextStats, antlr4_runtime::ParserDfaStats), String> {{
let stats = match language {{
{stats_arms}
other => return Err(format!("unsupported language: {{other}}")),
}};
Ok(stats)
}}
fn parse_usize(value: Option<String>, flag: &str) -> Result<usize, String> {{
let value = value.ok_or_else(|| format!("missing value for {{flag}}"))?;
value
.parse::<usize>()
.map_err(|err| format!("invalid {{flag}} value {{value}}: {{err}}"))
}}
fn dump_prediction_context_stats(stats: antlr4_runtime::PredictionContextStats) {{
eprintln!("perf context_store.contexts_created={{}}", stats.contexts_created);
eprintln!("perf context_store.singleton_contexts={{}}", stats.singleton_contexts);
eprintln!("perf context_store.array_contexts={{}}", stats.array_contexts);
eprintln!("perf context_store.array_entries={{}}", stats.array_entries);
eprintln!("perf context_store.interner_hits={{}}", stats.interner_hits);
eprintln!("perf context_store.pooled_bytes={{}}", stats.pooled_bytes);
eprintln!("perf context_store.retained_bytes={{}}", stats.retained_bytes);
eprintln!("perf context_store.context_capacity={{}}", stats.context_capacity);
eprintln!("perf context_store.array_parent_capacity={{}}", stats.array_parent_capacity);
eprintln!("perf context_store.array_return_state_capacity={{}}", stats.array_return_state_capacity);
eprintln!("perf context_store.interner_capacity={{}}", stats.interner_capacity);
eprintln!("perf context_store.workspace_merge_cache_entries={{}}", stats.workspace_merge_cache_entries);
eprintln!("perf context_store.workspace_merge_cache_capacity={{}}", stats.workspace_merge_cache_capacity);
eprintln!("perf context_store.workspace_entry_capacity={{}}", stats.workspace_entry_capacity);
eprintln!("perf context_store.outer_context_cache_hits={{}}", stats.outer_context_cache_hits);
eprintln!("perf context_store.outer_context_cache_misses={{}}", stats.outer_context_cache_misses);
}}
fn dump_parser_dfa_stats(stats: antlr4_runtime::ParserDfaStats) {{
eprintln!("perf dfa_store.states={{}}", stats.states);
eprintln!("perf dfa_store.transitions={{}}", stats.transitions);
eprintln!("perf dfa_store.max_row_width={{}}", stats.max_row_width);
eprintln!("perf dfa_store.dense_rows={{}}", stats.dense_rows);
eprintln!("perf dfa_store.sparse_rows={{}}", stats.sparse_rows);
eprintln!("perf dfa_store.empty_rows={{}}", stats.empty_rows);
eprintln!("perf dfa_store.dense_slots={{}}", stats.dense_slots);
eprintln!("perf dfa_store.sparse_entries={{}}", stats.sparse_entries);
eprintln!("perf dfa_store.row_width.le_64={{}}", stats.row_width_histogram[0]);
eprintln!("perf dfa_store.row_width.le_128={{}}", stats.row_width_histogram[1]);
eprintln!("perf dfa_store.row_width.le_256={{}}", stats.row_width_histogram[2]);
eprintln!("perf dfa_store.row_width.le_512={{}}", stats.row_width_histogram[3]);
eprintln!("perf dfa_store.row_width.gt_512={{}}", stats.row_width_histogram[4]);
eprintln!("perf dfa_store.populated_edges.0={{}}", stats.populated_edge_histogram[0]);
eprintln!("perf dfa_store.populated_edges.1={{}}", stats.populated_edge_histogram[1]);
eprintln!("perf dfa_store.populated_edges.2_3={{}}", stats.populated_edge_histogram[2]);
eprintln!("perf dfa_store.populated_edges.4_7={{}}", stats.populated_edge_histogram[3]);
eprintln!("perf dfa_store.populated_edges.8_15={{}}", stats.populated_edge_histogram[4]);
eprintln!("perf dfa_store.populated_edges.ge_16={{}}", stats.populated_edge_histogram[5]);
eprintln!("perf dfa_store.edge_density.empty={{}}", stats.edge_density_histogram[0]);
eprintln!("perf dfa_store.edge_density.le_1_pct={{}}", stats.edge_density_histogram[1]);
eprintln!("perf dfa_store.edge_density.le_5_pct={{}}", stats.edge_density_histogram[2]);
eprintln!("perf dfa_store.edge_density.le_12_5_pct={{}}", stats.edge_density_histogram[3]);
eprintln!("perf dfa_store.edge_density.le_25_pct={{}}", stats.edge_density_histogram[4]);
eprintln!("perf dfa_store.edge_density.gt_25_pct={{}}", stats.edge_density_histogram[5]);
eprintln!("perf dfa_store.hot_bytes={{}}", stats.hot_bytes);
eprintln!("perf dfa_store.cold_bytes={{}}", stats.cold_bytes);
if stats.states > 0 {{
eprintln!("perf dfa_store.hot_bytes_per_state={{}}", stats.hot_bytes / stats.states);
eprintln!("perf dfa_store.cold_bytes_per_state={{}}", stats.cold_bytes / stats.states);
}}
eprintln!("perf dfa_store.states_created={{}}", stats.states_created);
eprintln!("perf dfa_store.states_deduplicated={{}}", stats.states_deduplicated);
eprintln!("perf dfa_store.fingerprint_candidates={{}}", stats.fingerprint_candidates);
eprintln!("perf dfa_store.fingerprint_collisions={{}}", stats.fingerprint_collisions);
}}
"""
)
return runner
def rust_codegen_flags(
*,
native: bool,
pgo_generate: Path | None,
pgo_use: Path | None,
) -> list[str]:
flags = []
if native:
flags.append("-Ctarget-cpu=native")
if pgo_generate is not None:
flags.append(f"-Cprofile-generate={pgo_generate}")
if pgo_use is not None:
flags.extend(
[
f"-Cprofile-use={pgo_use}",
"-Cllvm-args=-pgo-warn-missing-function",
]
)
return flags
def rust_build_environment(args: argparse.Namespace) -> dict[str, str] | None:
flags = rust_codegen_flags(
native=args.rust_native,
pgo_generate=args.rust_pgo_generate,
pgo_use=args.rust_pgo_use,
)
if not flags:
return None
env = os.environ.copy()
existing = env.get("RUSTFLAGS", "").strip()
env["RUSTFLAGS"] = " ".join([existing, *flags]).strip()
return env
def rust_parse_function(spec: LanguageSpec) -> str:
return f"""fn lex_{spec.name}(src: &str) -> Result<(), String> {{
let lexer = generated::{spec.rust_lexer_module}::{spec.rust_lexer_type}::new(InputStream::new(src));
let tokens = CommonTokenStream::new(lexer);
black_box(tokens.token_count());
Ok(())
}}
fn parse_{spec.name}(src: &str) -> Result<(), antlr4_runtime::AntlrError> {{
let lexer = generated::{spec.rust_lexer_module}::{spec.rust_lexer_type}::new(InputStream::new(src));
let tokens = CommonTokenStream::new(lexer);
let mut parser = generated::{spec.rust_parser_module}::{spec.rust_parser_type}::new(tokens);
let tree = parser.{spec.rust_entry}()?;
black_box(&tree);
Ok(())
}}
fn dump_tree_{spec.name}(src: &str, out: &mut dyn Write) -> Result<(), String> {{
let lexer = generated::{spec.rust_lexer_module}::{spec.rust_lexer_type}::new(InputStream::new(src));
let mut tokens = CommonTokenStream::new(lexer);
tokens.fill();
let lexer_errors = tokens.drain_source_errors().len();
let mut parser = generated::{spec.rust_parser_module}::{spec.rust_parser_type}::new(tokens);
let root = parser
.{spec.rust_entry}()
.map_err(|err| err.to_string())?;
let syntax_errors = parser.number_of_syntax_errors();
if lexer_errors != 0 || syntax_errors != 0 {{
return Err(format!(
"rust-antlr {spec.name} parse produced {{lexer_errors}} lexer error(s) and {{syntax_errors}} parser syntax error(s)"
));
}}
dump_node(
out,
parser.node(root),
generated::{spec.rust_parser_module}::rule_names(),
0,
)
.map_err(|err| err.to_string())
}}
fn prediction_stats_{spec.name}(
src: &str,
) -> Result<(antlr4_runtime::PredictionContextStats, antlr4_runtime::ParserDfaStats), antlr4_runtime::AntlrError> {{
let lexer = generated::{spec.rust_lexer_module}::{spec.rust_lexer_type}::new(InputStream::new(src));
let tokens = CommonTokenStream::new(lexer);
let mut parser = generated::{spec.rust_parser_module}::{spec.rust_parser_type}::new(tokens);
let tree = parser.{spec.rust_entry}()?;
let stats = (parser.prediction_context_stats(), parser.parser_dfa_stats());
black_box(&tree);
Ok(stats)
}}"""
def write_python_antlr_runner(work_dir: Path, specs: list[LanguageSpec], py_gen_dir: Path) -> Path:
runner = work_dir / "python-antlr-runner"
runner.mkdir(parents=True, exist_ok=True)
spec_lines = ",\n".join(
f' "{spec.name}": ("{spec.lexer_name}", "{spec.parser_name}", "{spec.python_entry}")'
for spec in specs
)
script = runner / "bench_antlr.py"
script.write_text(
f"""#!/usr/bin/env python3
from __future__ import annotations
import argparse
import importlib
import sys
import time
from pathlib import Path
from antlr4 import CommonTokenStream, InputStream
GEN_DIR = {str(py_gen_dir)!r}
SPECS = {{
{spec_lines}
}}
SINK = None
CLASSES = {{}}
def parser_classes(language: str):
if language not in CLASSES:
lexer_name, parser_name, rule_name = SPECS[language]
lexer_module = importlib.import_module(lexer_name)
parser_module = importlib.import_module(parser_name)
CLASSES[language] = (
getattr(lexer_module, lexer_name),
getattr(parser_module, parser_name),
rule_name,
)
return CLASSES[language]
def parse_once(language: str, src: str) -> None:
global SINK
lexer_cls, parser_cls, rule_name = parser_classes(language)
stream = CommonTokenStream(lexer_cls(InputStream(src)))
parser = parser_cls(stream)
tree = getattr(parser, rule_name)()
SINK = tree
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--language", required=True, choices=sorted(SPECS))
parser.add_argument("--input", required=True)
parser.add_argument("--iters", type=int, default=1)
parser.add_argument("--warmups", type=int, default=0)
args = parser.parse_args()
if args.iters <= 0:
raise SystemExit("--iters must be greater than 0")
sys.path.insert(0, GEN_DIR)
src = Path(args.input).read_text()
for _ in range(args.warmups):
parse_once(args.language, src)
elapsed = []
for _ in range(args.iters):
started = time.perf_counter_ns()
parse_once(args.language, src)
elapsed.append(time.perf_counter_ns() - started)
print(f"min_ns={{min(elapsed)}} avg_ns={{sum(elapsed) // len(elapsed)}}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
"""
)
return script
def write_tree_sitter_runner(work_dir: Path, specs: list[LanguageSpec]) -> Path:
runner = work_dir / "tree-sitter-runner"
runner.mkdir(parents=True, exist_ok=True)
spec_lines = ",\n".join(
f' "{spec.name}": "{spec.tree_sitter_name}"'
for spec in specs
)
script = runner / "bench_tree_sitter.py"
script.write_text(
f"""#!/usr/bin/env python3
from __future__ import annotations
import argparse
import time
from pathlib import Path
from tree_sitter_language_pack import get_parser
SPECS = {{
{spec_lines}
}}
SINK = None
def parse_once(parser, src):
global SINK
tree = parser.parse(src)
# tree-sitter-language-pack ships py-tree-sitter bindings on some
# platforms (attributes) and its own Rust binding on others (methods);
# support both so the runner works on every wheel of the pinned version.
root = tree.root_node() if callable(tree.root_node) else tree.root_node
has_error = root.has_error() if callable(root.has_error) else root.has_error
if has_error:
raise RuntimeError("tree-sitter parse produced ERROR nodes")
SINK = tree
def probe_source(parser, raw: bytes):
try:
parse_once(parser, raw)
return raw
except TypeError:
# The Rust-binding wheels accept str input, not bytes.
src = raw.decode("utf-8")
parse_once(parser, src)
return src
def main() -> int:
arg_parser = argparse.ArgumentParser()
arg_parser.add_argument("--language", required=True, choices=sorted(SPECS))
arg_parser.add_argument("--input", required=True)
arg_parser.add_argument("--iters", type=int, default=1)
arg_parser.add_argument("--warmups", type=int, default=0)
args = arg_parser.parse_args()
if args.iters <= 0:
raise SystemExit("--iters must be greater than 0")
parser = get_parser(SPECS[args.language])
src = probe_source(parser, Path(args.input).read_bytes())
for _ in range(args.warmups):
parse_once(parser, src)
elapsed = []
for _ in range(args.iters):
started = time.perf_counter_ns()
parse_once(parser, src)
elapsed.append(time.perf_counter_ns() - started)
print(f"min_ns={{min(elapsed)}} avg_ns={{sum(elapsed) // len(elapsed)}}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
"""
)
return script
def write_go_runner(work_dir: Path, specs: list[LanguageSpec]) -> Path:
runner = work_dir / "go-runner"
runner.mkdir(parents=True, exist_ok=True)
(runner / "go.mod").write_text(
"\n".join(
[
"module parsebench",
"",
"go 1.23",
"",
f"require github.com/antlr4-go/antlr/v4 {GO_ANTLR_RUNTIME}",
"",
]
)
)
imports = "\n".join(
f' {go_package_name(spec)} "parsebench/{go_package_name(spec)}"'
for spec in specs
)
arms = "\n".join(
f' case "{spec.name}":\n parse{go_func_name(spec.name)}(src)'
for spec in specs
)
dump_arms = "\n".join(
f' case "{spec.name}":\n return dumpTree{go_func_name(spec.name)}(src, out)'
for spec in specs
)
functions = "\n\n".join(go_parse_function(spec) for spec in specs)
(runner / "main.go").write_text(
f"""package main
import (
"fmt"
"io"
"os"
"strconv"
"strings"
"time"
"github.com/antlr4-go/antlr/v4"
{imports}
)
var sink any
type countingErrorListener struct {{
*antlr.DefaultErrorListener
count int
}}
func newCountingErrorListener() *countingErrorListener {{
return &countingErrorListener{{DefaultErrorListener: antlr.NewDefaultErrorListener()}}
}}
func (l *countingErrorListener) SyntaxError(
_ antlr.Recognizer,
_ interface{{}},
_, _ int,
_ string,
_ antlr.RecognitionException,
) {{
l.count++
}}
func main() {{
if err := runMain(); err != nil {{
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}}
}}
func runMain() error {{
language := ""
input := ""
dumpTreePath := ""
iters := 1
warmups := 0
for i := 1; i < len(os.Args); i++ {{
switch os.Args[i] {{
case "--language":
i++
language = requiredArg(i, "--language")
case "--input":
i++
input = requiredArg(i, "--input")
case "--dump-tree":
i++
dumpTreePath = requiredArg(i, "--dump-tree")
case "--iters":
i++
iters = parsePositiveInt(requiredArg(i, "--iters"), "--iters")
case "--warmups":
i++
warmups = parseNonNegativeInt(requiredArg(i, "--warmups"), "--warmups")
default:
return fmt.Errorf("unknown argument: %s", os.Args[i])
}}
}}
if language == "" {{
return fmt.Errorf("missing --language <name>")
}}
if input == "" {{
return fmt.Errorf("missing --input <path>")
}}
bytes, err := os.ReadFile(input)
if err != nil {{
return err
}}
src := string(bytes)
if dumpTreePath != "" {{
file, err := os.Create(dumpTreePath)
if err != nil {{
return err
}}
defer file.Close()
return dumpTreeOnce(language, src, file)
}}
for i := 0; i < warmups; i++ {{
parseOnce(language, src)
}}
minNs := int64(1<<63 - 1)
totalNs := int64(0)
for i := 0; i < iters; i++ {{
started := time.Now()
parseOnce(language, src)
elapsed := time.Since(started).Nanoseconds()
if elapsed < minNs {{
minNs = elapsed
}}
totalNs += elapsed
}}
fmt.Printf("min_ns=%d avg_ns=%d\\n", minNs, totalNs/int64(iters))
return nil
}}
func parseOnce(language string, src string) {{
switch language {{
{arms}
default:
panic("unsupported language: " + language)
}}
}}
func dumpTreeOnce(language string, src string, out io.Writer) error {{
switch language {{
{dump_arms}
default:
return fmt.Errorf("unsupported language: %s", language)
}}
}}
func dumpNode(out io.Writer, tree antlr.Tree, ruleNames []string, depth int) error {{
pad := strings.Repeat(" ", depth)
switch node := tree.(type) {{
case antlr.ErrorNode:
_, err := fmt.Fprintf(out, "%sErr(%s)\\n", pad, treeText(node.GetText()))
return err
case antlr.TerminalNode:
_, err := fmt.Fprintf(out, "%sTerm(%s)\\n", pad, treeText(node.GetText()))
return err
case antlr.RuleNode:
ruleIndex := node.GetRuleContext().GetRuleIndex()
name := "<?>"
if ruleIndex >= 0 && ruleIndex < len(ruleNames) {{
name = ruleNames[ruleIndex]
}}
children := node.GetChildren()
if _, err := fmt.Fprintf(out, "%sRule(%s, children=%d)\\n", pad, name, len(children)); err != nil {{
return err
}}
for _, child := range children {{
if err := dumpNode(out, child, ruleNames, depth+1); err != nil {{
return err
}}
}}
return nil
default:
return fmt.Errorf("unsupported tree node type %T", tree)
}}
}}
func treeHasErrorNode(tree antlr.Tree) bool {{
if _, ok := tree.(antlr.ErrorNode); ok {{
return true
}}
if _, ok := tree.(antlr.TerminalNode); ok {{
return false
}}
for _, child := range tree.GetChildren() {{
if treeHasErrorNode(child) {{
return true
}}
}}
return false
}}
func treeText(text string) string {{
var b strings.Builder
b.WriteByte('"')
for _, r := range text {{
fmt.Fprintf(&b, `\\u{{%x}}`, r)
}}
b.WriteByte('"')
return b.String()
}}
func requiredArg(index int, flag string) string {{
if index >= len(os.Args) {{
panic("missing value for " + flag)
}}
return os.Args[index]
}}
func parsePositiveInt(value string, flag string) int {{
parsed, err := strconv.Atoi(value)
if err != nil || parsed <= 0 {{
panic("invalid " + flag + " value: " + value)
}}
return parsed
}}
func parseNonNegativeInt(value string, flag string) int {{
parsed, err := strconv.Atoi(value)
if err != nil || parsed < 0 {{
panic("invalid " + flag + " value: " + value)
}}
return parsed
}}
{functions}
"""
)
return runner
def go_func_name(name: str) -> str:
return "".join(part.capitalize() for part in name.split("-"))
def go_package_name(spec: LanguageSpec) -> str:
return f"{spec.name}parser"
def go_parse_function(spec: LanguageSpec) -> str:
pkg = go_package_name(spec)
func_name = go_func_name(spec.name)
return f"""func parse{func_name}(src string) {{
input := antlr.NewInputStream(src)
lexer := {pkg}.New{spec.lexer_name}(input)
tokens := antlr.NewCommonTokenStream(lexer, antlr.TokenDefaultChannel)
p := {pkg}.New{spec.parser_name}(tokens)
tree := p.{spec.go_entry}()
sink = tree
}}
func dumpTree{func_name}(src string, out io.Writer) error {{
input := antlr.NewInputStream(src)
lexer := {pkg}.New{spec.lexer_name}(input)
lexerErrors := newCountingErrorListener()
lexer.RemoveErrorListeners()
lexer.AddErrorListener(lexerErrors)
tokens := antlr.NewCommonTokenStream(lexer, antlr.TokenDefaultChannel)
tokens.Fill()
p := {pkg}.New{spec.parser_name}(tokens)
parserErrors := newCountingErrorListener()
p.RemoveErrorListeners()
p.AddErrorListener(parserErrors)
tree := p.{spec.go_entry}()
if lexerErrors.count != 0 || parserErrors.count != 0 || treeHasErrorNode(tree) {{
return fmt.Errorf(
"go-antlr {spec.name} parse produced %d lexer error(s), %d parser syntax error(s), or error nodes",
lexerErrors.count,
parserErrors.count,
)
}}
return dumpNode(out, tree, p.GetRuleNames(), 0)
}}"""
def copy_generated_go(go_gen_dir: Path, go_parser_dir: Path) -> None:
for path in go_gen_dir.glob("*.go"):
shutil.copy2(path, go_parser_dir / path.name)
def prepare_work(
args: argparse.Namespace,
specs: list[LanguageSpec],
runtimes: set[str],
) -> dict[str, Path]:
work_dir = args.work_dir
clear_work_dir(work_dir, args.runtime_root)
work_dir.mkdir(parents=True, exist_ok=True)
if args.rust_pgo_generate is not None:
args.rust_pgo_generate.mkdir(parents=True, exist_ok=True)
rust_runner = write_rust_runner(
work_dir,
specs,
args.runtime_root,
args.rust_thin_lto,
)
rust_generated = rust_runner / "src" / "generated"
py_gen = work_dir / "python-gen"
go_runner = write_go_runner(work_dir, specs)
for spec in specs:
if "rust-antlr" in runtimes:
base_grammar = work_dir / "grammars" / spec.name / "base"
copy_grammar(spec, args.grammars_v4, base_grammar)
if spec.name == "java":
transform_java(base_grammar)
interp_dir = work_dir / "generated" / spec.name / "interp"
generate_antlr(args.antlr_jar, spec, base_grammar, interp_dir, None)
generate_rust_modules(
spec,
base_grammar,
interp_dir,
rust_generated,
args.rust_generated_only,
args.runtime_root,
)
if "python-antlr" in runtimes:
py_grammar = work_dir / "grammars" / spec.name / "python"
copy_grammar(spec, args.grammars_v4, py_grammar)
if spec.name == "csharp":
transform_csharp_python(py_grammar)
if spec.name == "java":
transform_java(py_grammar)
py_lang_gen = py_gen
generate_antlr(args.antlr_jar, spec, py_grammar, py_lang_gen, "Python3")
prepare_python_support(spec, args.grammars_v4, py_lang_gen)
if "go-antlr" in runtimes:
go_grammar = work_dir / "grammars" / spec.name / "go"
copy_grammar(spec, args.grammars_v4, go_grammar)
if spec.name == "csharp":
transform_csharp_go(go_grammar)
if spec.name == "java":
transform_java(go_grammar)
go_gen = work_dir / "generated" / spec.name / "go"
package_name = go_package_name(spec)
generate_antlr(args.antlr_jar, spec, go_grammar, go_gen, "Go", package_name)
go_parser_dir = go_runner / package_name
go_parser_dir.mkdir(parents=True, exist_ok=True)
copy_generated_go(go_gen, go_parser_dir)
prepare_go_support(spec, args.grammars_v4, go_parser_dir, package_name)
runners: dict[str, Path] = {}
if "rust-antlr" in runtimes:
rust_build_cmd = [
"cargo",
"build",
"--quiet",
"--release",
"--manifest-path",
str(rust_runner / "Cargo.toml"),
]
if os.environ.get("ANTLR_PERF_DUMP"):
rust_build_cmd.extend(["--features", "perf-counters"])
run(rust_build_cmd, env=rust_build_environment(args))
runners["rust-antlr"] = rust_runner / "target" / "release" / "parse-bench-rust-runner"
if "python-antlr" in runtimes:
runners["python-antlr"] = write_python_antlr_runner(work_dir, specs, py_gen)
if "go-antlr" in runtimes:
run(["go", "mod", "tidy"], cwd=go_runner)
run(["go", "build", "-o", "parse-bench-go-runner", "."], cwd=go_runner)
runners["go-antlr"] = go_runner / "parse-bench-go-runner"
if "tree-sitter" in runtimes:
runners["tree-sitter"] = write_tree_sitter_runner(work_dir, specs)
return runners
def clear_work_dir(work_dir: Path, runtime_root: Path) -> None:
if not work_dir.exists():
return
if not work_dir.is_dir():
raise SystemExit(f"Refusing to delete non-directory work path: {work_dir}")
if work_dir == ROOT or work_dir in ROOT.parents:
raise SystemExit(f"Refusing to delete project root or parent directory: {work_dir}")
if work_dir == runtime_root or work_dir in runtime_root.parents:
raise SystemExit(
"Refusing to delete work directory containing the runtime checkout: "
f"{work_dir} (runtime root: {runtime_root})"
)
shutil.rmtree(work_dir)
def ensure_python_dependencies(python: str, runtimes: set[str]) -> None:
modules = []
if "python-antlr" in runtimes:
modules.append("antlr4")
if "tree-sitter" in runtimes:
modules.append("tree_sitter_language_pack")
if not modules:
return
code = "\n".join(f"import {module}" for module in modules)
try:
run([python, "-c", code], quiet=True)
except subprocess.CalledProcessError as err:
print(err.stderr, file=sys.stderr)
raise SystemExit(
"missing Python benchmark dependencies; run:\n"
f" {python} -m pip install -r {BENCH_ROOT / 'requirements.txt'}"
) from err
RUNNER_OUTPUT = re.compile(r"min_ns=(?P<min>\d+)\s+avg_ns=(?P<avg>\d+)")
def measure_fixture(
fixture: Fixture,
runtime: str,
runner: Path,
args: argparse.Namespace,
) -> Measurement:
env = None
if runtime == "rust-antlr" and args.rust_generated_only:
env = os.environ.copy()
env["ANTLR4_RUST_GENERATED_ONLY"] = "1"
if runtime in {"python-antlr", "tree-sitter"}:
cmd = [args.python, str(runner)]
else:
cmd = [str(runner)]
cmd.extend(
[
"--language",
fixture.language,
"--input",
str(fixture.abs_path),
]
)
if runtime == "rust-antlr":
cmd.extend(["--phase", args.phase])
cmd.extend(
[
"--iters",
str(args.iters),
"--warmups",
str(args.warmups),
]
)
completed = run(cmd, env=env, quiet=True)
if runtime == "rust-antlr" and os.environ.get("ANTLR_PERF_DUMP") and completed.stderr:
print(completed.stderr, file=sys.stderr, end="")
match = RUNNER_OUTPUT.search(completed.stdout)
if match is None:
raise SystemExit(
f"{runtime} runner did not report timing for {fixture.path}:\n"
f"stdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
)
return Measurement(
language=fixture.language,
fixture=fixture.name,
runtime=runtime,
min_ns=int(match.group("min")),
avg_ns=int(match.group("avg")),
bytes=fixture.abs_path.stat().st_size,
source=fixture.source,
license=fixture.license,
description=fixture.description,
)
def dump_tree_has_error_nodes(dump: str) -> bool:
return any(line.lstrip().startswith("Err(") for line in dump.splitlines())
def format_tree_diff(rust_dump: str, go_dump: str, *, max_lines: int = 80) -> str:
diff = list(
difflib.unified_diff(
rust_dump.splitlines(),
go_dump.splitlines(),
fromfile="rust-antlr",
tofile="go-antlr",
lineterm="",
)
)
if len(diff) > max_lines:
omitted = len(diff) - max_lines
diff = diff[:max_lines] + [f"... ({omitted} more diff lines omitted)"]
return "\n".join(diff)
def dump_fixture_tree(
runtime: str,
runner: Path,
fixture: Fixture,
output: Path,
args: argparse.Namespace,
) -> str:
env = None
if runtime == "rust-antlr" and args.rust_generated_only:
env = os.environ.copy()
env["ANTLR4_RUST_GENERATED_ONLY"] = "1"
cmd = [
str(runner),
"--language",
fixture.language,
"--input",
str(fixture.abs_path),
"--dump-tree",
str(output),
]
if runtime == "rust-antlr":
cmd.extend(["--phase", "parse"])
run(cmd, env=env, quiet=True)
return output.read_text()
def validate_rust_go_ast_parity(
fixtures: list[Fixture],
runners: dict[str, Path],
args: argparse.Namespace,
) -> None:
if not args.ast_check:
return
if args.phase != "parse":
raise SystemExit("--ast-check requires --phase parse")
if "rust-antlr" not in runners or "go-antlr" not in runners:
raise SystemExit(
"--ast-check requires --runtimes to include both rust-antlr and go-antlr"
)
dump_root = args.work_dir / "ast-dumps"
dump_root.mkdir(parents=True, exist_ok=True)
failures: list[str] = []
for fixture in fixtures:
fixture_dir = dump_root / fixture.language
fixture_dir.mkdir(parents=True, exist_ok=True)
rust_path = fixture_dir / f"{fixture.name}.rust.txt"
go_path = fixture_dir / f"{fixture.name}.go.txt"
try:
rust_dump = dump_fixture_tree(
"rust-antlr", runners["rust-antlr"], fixture, rust_path, args
)
go_dump = dump_fixture_tree(
"go-antlr", runners["go-antlr"], fixture, go_path, args
)
except subprocess.CalledProcessError as err:
detail = (err.stderr or err.stdout or "").strip()
failures.append(
f"{fixture.language}/{fixture.name}: tree dump failed"
+ (f": {detail}" if detail else "")
)
continue
if dump_tree_has_error_nodes(rust_dump):
failures.append(
f"{fixture.language}/{fixture.name}: rust-antlr AST contains error nodes "
f"(dump: {rust_path})"
)
if dump_tree_has_error_nodes(go_dump):
failures.append(
f"{fixture.language}/{fixture.name}: go-antlr AST contains error nodes "
f"(dump: {go_path})"
)
if rust_dump != go_dump:
failures.append(
f"{fixture.language}/{fixture.name}: rust-antlr and go-antlr ASTs differ\n"
f"{format_tree_diff(rust_dump, go_dump)}"
)
else:
print(f"{fixture.language}/{fixture.name}: rust/go AST parity ok")
if failures:
print("rust/go AST parity check failed:", file=sys.stderr)
for failure in failures:
print(f" {failure}", file=sys.stderr)
raise SystemExit(1)
def print_table(results: list[Measurement]) -> None:
rust_by_fixture = {
(result.language, result.fixture): result.avg_ns
for result in results
if result.runtime == "rust-antlr"
}
rows = []
for result in results:
rust_avg = rust_by_fixture.get((result.language, result.fixture))
relative = ""
if rust_avg:
relative = format_relative(result.avg_ns / rust_avg)
rows.append(
[
result.language,
result.fixture,
result.runtime,
str(result.bytes),
f"{result.min_ns / 1_000_000:.3f}",
f"{result.avg_ns / 1_000_000:.3f}",
relative,
]
)
headers = ["language", "fixture", "runtime", "bytes", "min ms", "avg ms", "vs rust"]
widths = [
max(len(row[index]) for row in rows + [headers])
for index in range(len(headers))
]
print(" | ".join(header.ljust(widths[index]) for index, header in enumerate(headers)))
print("-|-".join("-" * width for width in widths))
for row in rows:
print(" | ".join(cell.ljust(widths[index]) for index, cell in enumerate(row)))
def write_json(results: list[Measurement], args: argparse.Namespace) -> None:
if args.json is None:
return
args.json.parent.mkdir(parents=True, exist_ok=True)
data = {
"generated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
"iters": args.iters,
"warmups": args.warmups,
"phase": args.phase,
"rust_generated_only": args.rust_generated_only,
"repo": str(args.runtime_root),
"repo_commit": git_rev(args.runtime_root),
"rust_native": args.rust_native,
"rust_thin_lto": args.rust_thin_lto,
"rust_pgo_generate": (
str(args.rust_pgo_generate) if args.rust_pgo_generate is not None else None
),
"rust_pgo_use": str(args.rust_pgo_use) if args.rust_pgo_use is not None else None,
"grammars_v4": {
"path": str(args.grammars_v4),
"commit": git_rev(args.grammars_v4),
},
"results": [dataclasses.asdict(result) for result in results],
}
args.json.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n")
print(f"wrote JSON report: {args.json}")
def write_markdown(results: list[Measurement], args: argparse.Namespace) -> None:
if args.markdown is None:
return
rust_by_fixture = {
(result.language, result.fixture): result.avg_ns
for result in results
if result.runtime == "rust-antlr"
}
lines = [
f"# {args.phase.title()} Benchmark",
"",
f"- Iterations: `{args.iters}`",
f"- Warmups: `{args.warmups}`",
f"- Phase: `{args.phase}`",
f"- Runtime checkout: `{git_rev(args.runtime_root) or args.runtime_root}`",
f"- Native CPU: `{'yes' if args.rust_native else 'no'}`",
f"- ThinLTO / one codegen unit: `{'yes' if args.rust_thin_lto else 'no'}`",
f"- PGO profile generation: `{args.rust_pgo_generate or 'no'}`",
f"- PGO profile use: `{args.rust_pgo_use or 'no'}`",
f"- Rust generated-only: `{'yes' if args.rust_generated_only else 'no'}`",
f"- grammars-v4: `{git_rev(args.grammars_v4) or args.grammars_v4}`",
"",
"| Language | Fixture | Runtime | Bytes | Min ms | Avg ms | vs Rust |",
"| --- | --- | --- | ---: | ---: | ---: | ---: |",
]
for result in results:
rust_avg = rust_by_fixture.get((result.language, result.fixture))
relative = format_relative(result.avg_ns / rust_avg) if rust_avg else ""
lines.append(
f"| {result.language} | {result.fixture} | {result.runtime} | "
f"{result.bytes} | {result.min_ns / 1_000_000:.3f} | "
f"{result.avg_ns / 1_000_000:.3f} | {relative} |"
)
args.markdown.parent.mkdir(parents=True, exist_ok=True)
args.markdown.write_text("\n".join(lines) + "\n")
print(f"wrote Markdown report: {args.markdown}")
def format_relative(value: float) -> str:
if value < 0.01:
return "<0.01x"
return f"{value:.2f}x"
def git_rev(path: Path) -> str | None:
if not (path / ".git").exists():
return None
try:
return run(["git", "rev-parse", "HEAD"], cwd=path, quiet=True).stdout.strip()
except subprocess.CalledProcessError:
return None
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--antlr-jar",
type=Path,
default=Path(os.environ.get("ANTLR4_JAR", DEFAULT_ANTLR_JAR)),
)
parser.add_argument(
"--grammars-v4",
type=Path,
default=Path(os.environ.get("GRAMMARS_V4", DEFAULT_GRAMMARS_V4)),
)
parser.add_argument(
"--runtime-root",
type=Path,
default=ROOT,
help="Runtime checkout to generate and link into the Rust benchmark runner.",
)
parser.add_argument("--work-dir", type=Path, default=ROOT / "target" / "parse-bench")
parser.add_argument("--python", default=os.environ.get("PYTHON", sys.executable))
parser.add_argument("--languages", default="kotlin,csharp,java")
parser.add_argument("--runtimes", default="rust-antlr,python-antlr,go-antlr,tree-sitter")
parser.add_argument(
"--phase",
choices=PHASES,
default="parse",
help="Measure complete parsing or lexing/token buffering only.",
)
parser.add_argument(
"--rust-native",
action="store_true",
help="Build the Rust runner with -C target-cpu=native.",
)
parser.add_argument(
"--rust-thin-lto",
action="store_true",
help="Build the Rust runner with ThinLTO and one codegen unit.",
)
rust_pgo = parser.add_mutually_exclusive_group()
rust_pgo.add_argument(
"--rust-pgo-generate",
type=Path,
metavar="DIR",
help="Build the Rust runner with -Cprofile-generate=DIR for PGO training.",
)
rust_pgo.add_argument(
"--rust-pgo-use",
type=Path,
metavar="FILE",
help="Build the Rust runner with -Cprofile-use=FILE.",
)
parser.add_argument("--iters", type=int, default=10)
parser.add_argument("--warmups", type=int, default=2)
parser.add_argument("--quick", action="store_true", help="Use 3 timed iterations and 1 warmup.")
parser.add_argument(
"--rust-generated-only",
action="store_true",
help=(
"Require all Rust parser rules to be generated, then run rust-antlr "
"with ANTLR4_RUST_GENERATED_ONLY=1 so missing coverage fails instead "
"of falling back to the interpreter."
),
)
parser.add_argument(
"--ast-check",
action="store_true",
help=(
"Before timing, dump each fixture's parse tree from rust-antlr and "
"go-antlr (Rule/Term/Err format) and require byte-identical dumps "
"with no error nodes. Requires both runtimes for --phase parse; "
"use this to gate fair throughput comparisons."
),
)
parser.add_argument("--json", type=Path, help="Write machine-readable results.")
parser.add_argument("--markdown", type=Path, help="Write a Markdown table report.")
return parser.parse_args()
def normalize_paths(args: argparse.Namespace) -> None:
args.antlr_jar = args.antlr_jar.expanduser().resolve()
args.grammars_v4 = args.grammars_v4.expanduser().resolve()
args.runtime_root = args.runtime_root.expanduser().resolve()
args.work_dir = args.work_dir.expanduser().resolve()
if args.json is not None:
args.json = args.json.expanduser().resolve()
if args.markdown is not None:
args.markdown = args.markdown.expanduser().resolve()
if args.rust_pgo_generate is not None:
args.rust_pgo_generate = args.rust_pgo_generate.expanduser().resolve()
if args.rust_pgo_use is not None:
args.rust_pgo_use = args.rust_pgo_use.expanduser().resolve()
def main() -> int:
args = parse_args()
normalize_paths(args)
if args.quick:
args.iters = 3
args.warmups = 1
if args.iters <= 0:
raise SystemExit("--iters must be greater than 0")
if args.warmups < 0:
raise SystemExit("--warmups must be >= 0")
languages = parse_csv(args.languages, LANGUAGES, "language")
runtimes = set(parse_csv(args.runtimes, RUNTIMES, "runtime"))
if args.phase == "lex" and runtimes != {"rust-antlr"}:
raise SystemExit("--phase lex currently requires --runtimes rust-antlr")
specs = [LANGUAGES[name] for name in languages]
fixtures = load_fixtures(set(languages), args.phase)
require_path(args.antlr_jar, "ANTLR jar")
require_path(args.grammars_v4, "grammars-v4 checkout")
require_path(args.runtime_root / "Cargo.toml", "runtime Cargo manifest")
if args.rust_pgo_generate is not None and args.rust_pgo_generate.exists():
if not args.rust_pgo_generate.is_dir():
raise SystemExit(
f"Rust PGO generation path is not a directory: {args.rust_pgo_generate}"
)
if args.rust_pgo_use is not None:
require_path(args.rust_pgo_use, "Rust PGO profile")
if not args.rust_pgo_use.is_file():
raise SystemExit(f"Rust PGO profile is not a file: {args.rust_pgo_use}")
ensure_python_dependencies(args.python, runtimes)
runners = prepare_work(args, specs, runtimes)
validate_rust_go_ast_parity(fixtures, runners, args)
results: list[Measurement] = []
for fixture in fixtures:
for runtime in RUNTIMES:
if runtime not in runtimes:
continue
result = measure_fixture(fixture, runtime, runners[runtime], args)
results.append(result)
print(
f"{fixture.language}/{fixture.name} {runtime}: "
f"avg {result.avg_ns / 1_000_000:.3f}ms"
)
print()
print_table(results)
write_json(results, args)
write_markdown(results, args)
return 0
if __name__ == "__main__":
sys.exit(main())