#!/usr/bin/env python3
"""Compare bytecode between CPython and RustPython.
Compiles all Python files under Lib/ with both interpreters and reports
differences in the generated bytecode instructions.
Usage:
python scripts/compare_bytecode.py
python scripts/compare_bytecode.py --detail
python scripts/compare_bytecode.py --filter "asyncio/*.py"
python scripts/compare_bytecode.py --summary-json report.json
"""
import argparse
import fnmatch
import json
import os
import random
import subprocess
import sys
import tempfile
from collections import defaultdict
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_ROOT = os.path.dirname(SCRIPT_DIR)
DIS_DUMP = os.path.join(SCRIPT_DIR, "dis_dump.py")
DEFAULT_REPORT = os.path.join(PROJECT_ROOT, "compare_bytecode.report")
DUMP_TIMEOUT = 600
def find_rustpython():
"""Locate the RustPython binary, allowing release builds only."""
if "RUSTPYTHON" in os.environ:
path = os.environ["RUSTPYTHON"]
normalized = os.path.normpath(path)
debug_fragment = os.path.join("target", "debug", "rustpython")
if normalized.endswith(debug_fragment):
raise ValueError(
"RUSTPYTHON must point to a release binary, not target/debug/rustpython"
)
return path
path = os.path.join(PROJECT_ROOT, "target", "release", "rustpython")
if os.path.isfile(path) and os.access(path, os.X_OK):
return path
return None
def collect_targets(lib_dir, pattern=None):
"""Collect Python files to compare, relative to lib_dir."""
targets = []
for root, dirs, files in os.walk(lib_dir):
dirs[:] = sorted(
d for d in dirs if d != "__pycache__" and not d.startswith(".")
)
for fname in sorted(files):
if not fname.endswith(".py"):
continue
fpath = os.path.join(root, fname)
relpath = os.path.relpath(fpath, lib_dir)
if pattern and not fnmatch.fnmatch(relpath, pattern):
continue
targets.append((relpath, fpath))
return targets
def _start_one(interpreter, targets, base_dir):
"""Start a single dis_dump.py subprocess."""
env = os.environ.copy()
if interpreter != sys.executable:
env["RUSTPYTHONPATH"] = base_dir
files_file = None
output_file = None
try:
files_file = tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
delete=False,
prefix="compare-bytecode-files-",
)
output_file = tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
delete=False,
prefix="compare-bytecode-output-",
)
for _, path in targets:
files_file.write(path)
files_file.write("\n")
files_file.close()
output_file.close()
cmd = [
interpreter,
DIS_DUMP,
"--base-dir",
base_dir,
"--files-from",
files_file.name,
"--output",
output_file.name,
"--progress",
"10",
]
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=None, # inherit stderr so progress dots appear on terminal
env=env,
cwd=PROJECT_ROOT,
)
return {
"proc": proc,
"files_file": files_file.name,
"output_file": output_file.name,
"targets": targets,
"interpreter": interpreter,
"base_dir": base_dir,
}
except Exception:
for handle in (files_file, output_file):
if handle is None:
continue
try:
handle.close()
finally:
if os.path.exists(handle.name):
os.unlink(handle.name)
raise
def _load_dump_output(output_file):
try:
with open(output_file, encoding="utf-8") as f:
content = f.read().strip()
except OSError as e:
print(" Failed to read dump output: %s" % e, file=sys.stderr)
return None
if not content:
return {}
try:
return json.loads(content)
except json.JSONDecodeError as e:
print(" JSON parse error: %s" % e, file=sys.stderr)
return None
def _run_sync_dump(interpreter, targets, base_dir, timeout=DUMP_TIMEOUT):
job = _start_one(interpreter, targets, base_dir)
proc = job["proc"]
stdout = b""
timed_out = False
try:
stdout = proc.communicate(timeout=timeout)[0]
except subprocess.TimeoutExpired:
proc.kill()
proc.communicate()
print(f" Timeout ({timeout}s)", file=sys.stderr)
timed_out = True
try:
data = _load_dump_output(job["output_file"])
finally:
for path in (job["files_file"], job["output_file"]):
if os.path.exists(path):
os.unlink(path)
if timed_out:
return {}
if proc.returncode != 0:
print(" Warning: exited with code %d" % proc.returncode, file=sys.stderr)
stray = stdout.decode(errors="replace").strip()
if stray:
print(" Warning: unexpected stdout from dump helper", file=sys.stderr)
return data
def _rerun_missing_targets(interpreter, targets, base_dir):
recovered = {}
failed = []
empty = []
for target in targets:
relpath = target[0]
data = _run_sync_dump(interpreter, [target], base_dir)
if data is None:
failed.append(relpath)
recovered[relpath] = {
"status": "error",
"error": "dump helper failed while rerunning target",
}
elif data:
recovered.update(data)
else:
empty.append(relpath)
recovered[relpath] = {
"status": "error",
"error": "dump helper produced no data while rerunning target",
}
if failed:
print(
" Warning: rerun failed for %d file(s): %s"
% (len(failed), ", ".join(failed[:5])),
file=sys.stderr,
)
if empty:
print(
" Warning: rerun produced no data for %d file(s): %s"
% (len(empty), ", ".join(empty[:5])),
file=sys.stderr,
)
return recovered
def _finish_one(job, timeout=DUMP_TIMEOUT):
"""Wait for a single dis_dump.py process and return parsed JSON."""
proc = job["proc"]
expected = {relpath for relpath, _ in job["targets"]}
stdout = b""
try:
stdout = proc.communicate(timeout=timeout)[0]
except subprocess.TimeoutExpired:
proc.kill()
proc.communicate()
print(
f" Timeout ({timeout}s), retrying {len(job['targets'])} file(s) serially",
file=sys.stderr,
)
data = None
else:
data = _load_dump_output(job["output_file"])
finally:
for path in (job["files_file"], job["output_file"]):
if os.path.exists(path):
os.unlink(path)
if proc.returncode != 0:
print(" Warning: exited with code %d" % proc.returncode, file=sys.stderr)
stray = stdout.decode(errors="replace").strip()
if stray:
print(" Warning: unexpected stdout from dump helper", file=sys.stderr)
if data is None:
return _rerun_missing_targets(
job["interpreter"], job["targets"], job["base_dir"]
)
missing = [
target
for target in job["targets"]
if target[0] not in data and target[0] in expected
]
if missing:
print(
" Re-running %d missing file(s) serially" % len(missing),
file=sys.stderr,
)
data.update(
_rerun_missing_targets(job["interpreter"], missing, job["base_dir"])
)
return data
def start_dump(interpreter, targets, base_dir, num_workers=1):
"""Start dis_dump.py under the given interpreter, split across workers."""
if num_workers 50:
p(" ... and %d more" % (len(rp_error_files) - 50))
p()
if diff_files:
p("-" * 60)
p(" Bytecode Differences")
p("-" * 60)
for fp, code_diffs in diff_files:
p()
p(" %s:" % fp)
for code_path, diffs in code_diffs:
shown = min(len(diffs), args.max_diffs)
p(" %s: %d difference(s)" % (code_path, len(diffs)))
for idx, cp_inst, rp_inst in diffs[:shown]:
if idx == -1:
p(" %s" % (cp_inst or rp_inst))
else:
p(" [%3d] CPython: %s" % (idx, cp_inst))
p(" RustPython: %s" % rp_inst)
if len(diffs) > shown:
p(" ... and %d more" % (len(diffs) - shown))
p()
else:
list_limit = 0 if args.summary_json else max(args.list_limit, 0)
if diff_summaries and list_limit:
shown = min(list_limit, len(diff_summaries))
total = len(diff_summaries)
p(f"Top differing files ({shown} shown of {total}):")
top = sorted(
diff_summaries,
key=lambda item: (
item["diff_instructions"],
item["diff_code_objects"],
item["path"],
),
reverse=True,
)[:list_limit]
for item in top:
p(
" %s (%d code objects, %d instruction diffs)"
% (
item["path"],
item["diff_code_objects"],
item["diff_instructions"],
)
)
p()
p("Use --detail to see specific instruction differences.")
p()
# Summary JSON output
if args.summary_json:
summary = {
"total": total,
"sample": args.sample,
"sample_seed": sample_seed,
"match": match,
"differ": differ,
"rp_error": rp_err,
"cp_error": cp_err,
"both_error": both_err,
"rp_missing": rp_miss,
"match_pct": round(100.0 * match / total, 2) if total else 0,
"diff_files": [fp for fp, _ in diff_files]
if need_detailed_diffs
else [item["path"] for item in diff_summaries],
"top_diff_files": sorted(
diff_summaries,
key=lambda item: (
item["diff_instructions"],
item["diff_code_objects"],
item["path"],
),
reverse=True,
)[: min(20, len(diff_summaries))],
"rp_error_files": [fp for fp, _ in rp_error_files],
}
with open(args.summary_json, "w") as f:
json.dump(summary, f, indent=2)
log("Summary JSON: %s" % args.summary_json)
log("Done: %d match, %d differ, %d errors" % (match, differ, rp_err))
sys.exit(0 if differ == 0 and rp_err == 0 else 1)
if __name__ == "__main__":
main()