Context Navigation

Changes between Version 4 and Version 5 of Ticket #37434, comment 2

Timestamp:: Oct 16, 2026, 11:11:24 PM (3 weeks ago)
Author:: harshit gupta

Legend:

: Unmodified
: Added
: Removed
: Modified

Ticket #37434, comment 2

-              v4
+              v5
 > Please can you share benchmark scripts along with the claimed performance improvement from this change.
+>
+```
+"""Check and benchmark the local post-commit callback change without Git writes."""
+from __future__ import annotations
+import argparse
+import ast
+import gc
+import hashlib
+import importlib.metadata
+import json
+import logging
+import math
+import os
+import platform
+import runpy
+import statistics
+import subprocess
+import sys
+import time
+import tracemalloc
+import types
+import unittest
+from pathlib import Path
+if not __debug__:
+    raise RuntimeError(
+        "Run these checks without -O or PYTHONOPTIMIZE; assertions must be enabled."
+    )
+HERE = Path(__file__).resolve().parent
+REPO = Path(
+    os.environ.get("DJANGO_REVIEW_REPO", HERE.parent / "django-current")
+).resolve()
+BASE = "d430084791a61b31cecec132bce3dca19194f7f2"
+FILES = ("django/db/backends/base/base.py", "tests/transaction_hooks/tests.py")
+REPEATS = 5
+COUNTS = (0, 1, 2, 10, 100, 500, 1000, 2000, 5000, 10000)
+WORKLOADS = [(count, 0) for count in COUNTS] + [(100, 65536), (1000, 1024)]
+CPU_HELPER = HERE / "benchmark_cpu.py"
+if not CPU_HELPER.exists():
+    CPU_HELPER = HERE.parent / "benchmarks/benchmark_cpu.py"
+sys.path.insert(0, str(CPU_HELPER.parent))
+from benchmark_cpu import RotatingCores
+sys.path.insert(0, str(REPO))
+from django.db.backends.base import base as base_module
+if Path(base_module.__file__).resolve() != REPO / FILES[0]:
+    raise RuntimeError("The wrong Django checkout was imported")
+def digest(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+def metadata():
+    return {
+        "base_commit": BASE,
+        "python": platform.python_version(),
+        "platform": platform.platform(),
+        "source_hashes": {name: digest(REPO / name) for name in FILES},
+        "runner_sha256": digest(Path(__file__)),
+        "affinity_helper_sha256": digest(CPU_HELPER),
+        "dependencies": {
+            name: importlib.metadata.version(name)
+            for name in ("Django", "asgiref", "sqlparse", "tzdata")
+        },
+    }
+def methods():
+    source = subprocess.check_output(
+        ["git", "show", f"{BASE}:{FILES[0]}"],
+        cwd=REPO,
+        text=True,
+        encoding="utf-8",
+    )
+    cls = next(
+        node
+        for node in ast.parse(source).body
+        if isinstance(node, ast.ClassDef) and node.name == "BaseDatabaseWrapper"
+    )
+    node = next(
+        node
+        for node in cls.body
+        if isinstance(node, ast.FunctionDef)
+        and node.name == "run_and_clear_commit_hooks"
+    )
+    namespace = dict(vars(base_module))
+    exec(
+        compile(
+            ast.Module(body=[node], type_ignores=[]),
+            "<pinned upstream callback dispatch>",
+            "exec",
+        ),
+        namespace,
+    )
+    compiled = namespace[node.name]
+    original = types.FunctionType(
+        compiled.__code__, vars(base_module), compiled.__name__, compiled.__defaults__
+    )
+    return {
+        "base": original,
+        "pr": base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks,
+    }
+def save(name, value):
+    (HERE / name).write_text(json.dumps(value, indent=2), encoding="utf-8")
+def run_tests(version, cores):
+    choices = methods()
+    base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks = choices[version]
+    report = {
+        **metadata(),
+        "version": version,
+        "backend": "SQLite",
+        "parallel_workers": 1,
+    }
+    original_start = unittest.TestResult.startTest
+    original_run = unittest.TextTestRunner.run
+    count = 0
+    def start_test(result, test):
+        nonlocal count
+        if count % 50 == 0:
+            cores.rotate()
+        count += 1
+        return original_start(result, test)
+    def run_suite(runner, suite):
+        result = original_run(runner, suite)
+        report.update(
+            tests_run=result.testsRun,
+            failures=[test.id() for test, _ in result.failures],
+            errors=[test.id() for test, _ in result.errors],
+            skipped=[test.id() for test, _ in result.skipped],
+            expected_failures=len(result.expectedFailures),
+            unexpected_successes=[test.id() for test in result.unexpectedSuccesses],
+        )
+        return result
+    unittest.TestResult.startTest = start_test
+    unittest.TextTestRunner.run = run_suite
+    os.chdir(REPO / "tests")
+    sys.path.insert(0, str(REPO / "tests"))
+    sys.argv = [
+        "runtests.py",
+        "transaction_hooks",
+        "transactions",
+        "backends.base",
+        "--settings=test_sqlite",
+        "--parallel=1",
+        "--verbosity=1",
+    ]
+    try:
+        try:
+            runpy.run_path(str(REPO / "tests/runtests.py"), run_name="__main__")
+            code = 0
+        except SystemExit as error:
+            code = error.code or 0
+    finally:
+        unittest.TestResult.startTest = original_start
+        unittest.TextTestRunner.run = original_run
+        base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks = choices["pr"]
+    report["exit_code"] = code
+    save(f"tests_{version}.json", report)
+    return code
+def setup():
+    from django.conf import settings
+    settings.configure(
+        DATABASES={
+            "default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}
+        },
+        INSTALLED_APPS=[],
+        SECRET_KEY="local-callback-check",
+        USE_TZ=True,
+    )
+    import django
+    django.setup()
+    from django.db import connection
+    connection.ensure_connection()
+class CallbackAbort(BaseException):
+    pass
+def outcome(method, count, failure_at, robust_mode, nested_mode, exception_class):
+    from django.db import connection, transaction
+    previous = base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks
+    base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks = method
+    events, errors = [], []
+    class Capture(logging.Handler):
+        def emit(self, record):
+            errors.append(
+                (
+                    record.getMessage(),
+                    type(record.exc_info[1]).__name__,
+                    str(record.exc_info[1]),
+                )
+            )
+    handler = Capture()
+    logger = base_module.logger
+    logger.addHandler(handler)
+    old_propagate = logger.propagate
+    logger.propagate = False
+    error = None
+    def callback(index):
+        events.append(("callback", index))
+        if index == failure_at:
+            raise exception_class("callback failed")
+        if nested_mode != "none" and index % 3 == 0:
+            if nested_mode == "transaction":
+                with transaction.atomic():
+                    transaction.on_commit(lambda: events.append(("nested", index)))
+            else:
+                transaction.on_commit(lambda: events.append(("nested", index)))
+    try:
+        try:
+            with transaction.atomic():
+                for index in range(count):
+                    robust = robust_mode == "all" or (
+                        robust_mode == "alternating" and index % 2 == 0
+                    )
+                    transaction.on_commit(
+                        lambda index=index: callback(index), robust=robust
+                    )
+        except BaseException as caught:
+            error = type(caught).__name__, str(caught)
+        return (
+            events,
+            error,
+            errors,
+            len(connection.run_on_commit),
+            connection.in_atomic_block,
+            connection.get_autocommit(),
+        )
+    finally:
+        logger.removeHandler(handler)
+        logger.propagate = old_propagate
+        base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks = previous
+def verify(choices, cores):
+    comparisons = 0
+    for count in (0, 1, 2, 3, 10, 30, 100, 250):
+        cores.rotate()
+        positions = (
+            list(dict.fromkeys((None, 0, count // 2, count - 1))) if count else [None]
+        )
+        for failure_at in positions:
+            for robust in ("none", "all", "alternating"):
+                for nested in ("none", "immediate", "transaction"):
+                    for exception in (ValueError, CallbackAbort):
+                        old = outcome(
+                            choices["base"],
+                            count,
+                            failure_at,
+                            robust,
+                            nested,
+                            exception,
+                        )
+                        new = outcome(
+                            choices["pr"], count, failure_at, robust, nested, exception
+                        )
+                        if old != new:
+                            raise AssertionError(
+                                (count, failure_at, robust, nested, exception, old, new)
+                            )
+                        comparisons += 1
+    print(f"Differential checks: {comparisons} matched", flush=True)
+    result = {**metadata(), "comparisons": comparisons, "differences": []}
+    save("differential.json", result)
+    return result
+def workflow(method, count, payload_size):
+    from django.db import connection, transaction
+    previous = base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks
+    base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks = method
+    updates = {}
+    try:
+        with transaction.atomic():
+            for index in range(count):
+                if payload_size:
+                    transaction.on_commit(
+                        lambda index=index, payload=bytearray(
+                            payload_size
+                        ): updates.__setitem__(f"record:{index}", index + len(payload))
+                    )
+                else:
+                    transaction.on_commit(
+                        lambda index=index: updates.__setitem__(
+                            f"record:{index}", index
+                        )
+                    )
+        if connection.run_on_commit:
+            raise AssertionError("Pending callbacks after commit")
+        return list(updates.items())
+    finally:
+        base_module.BaseDatabaseWrapper.run_and_clear_commit_hooks = previous
+def write_report(report):
+    save("benchmark_results.json", report)
+    lines = [
+        "# Post-commit callback benchmark",
+        "",
+        "Final slot-clearing implementation, not the earlier plain-loop prototype.",
+        "",
+        f"Python {report['python']}, Windows, SQLite. Five paired timing samples and five separate tracemalloc peaks per version, workload and fixed CPU pair. Short calls are batched. GC is enabled. Only two physical cores are allowed at once; pairs rotate between rows. One process, not parallel workers. Different CPU pairs are not pooled.",
+        "",
+        "The complete workflow includes transaction begin/commit, callback registration and callbacks updating a dictionary. Payload cases give each callback a separate bytearray of the stated size. No model writes, network or task broker. Large queues are stress cases, not typical requests. Peak memory is newly traced Python allocations, not total process RAM. All measured rows, including losses, are shown.",
+        "",
+        "| Callbacks | Captured payload/callback | CPU mask | Base time | Changed time | Speed | Base peak | Changed peak | Memory |",
+        "|---:|---:|---|---:|---:|---|---:|---:|---|",
+    ]
+    for row in report["results"]:
+        old, new = (statistics.median(row[v]["seconds"]) for v in ("base", "pr"))
+        old_peak, new_peak = (
+            statistics.median(row[v]["peak_bytes"]) for v in ("base", "pr")
+        )
+        speed = f"{old / new:.3f}x faster" if new < old else f"{new / old:.3f}x slower"
+        memory = (
+            "same"
+            if old_peak == new_peak
+            else (
+                f"{new_peak / old_peak:.6f}x more"
+                if new_peak > old_peak
+                else f"{old_peak / new_peak:.6f}x less"
+            )
+        )
+        lines.append(
+            f"| {row['callbacks']:,} | {row['payload_bytes']:,} B | {row['mask']} | {old * 1000:.4f} ms | {new * 1000:.4f} ms | {speed} | {old_peak / 1024:.3f} KiB | {new_peak / 1024:.3f} KiB | {memory} |"
+        )
+    lines += [
+        "",
+        f"Completed {len(report['results'])}/{report['planned_rows']} rows. Complete: {report['complete']}.",
+        "",
+        "Raw samples and source, runner and helper hashes are in benchmark_results.json.",
+        "",
+    ]
+    (HERE / "benchmark_results.md").write_text("\n".join(lines), encoding="utf-8")
+def benchmark(choices, cores):
+    report = {
+        **metadata(),
+        "repeats": REPEATS,
+        "planned_rows": len(WORKLOADS) * len(cores.pairs),
+        "cpu_pairs": [hex(mask) for mask in cores.pairs],
+        "complete": False,
+        "results": [],
+    }
+    write_report(report)
+    for count, payload in WORKLOADS:
+        for _ in cores.pairs:
+            mask = cores.rotate()
+            expected = workflow(choices["base"], count, payload)
+            assert workflow(choices["pr"], count, payload) == expected
+            start = time.perf_counter()
+            workflow(choices["base"], count, payload)
+            batch = max(
+, min(1000, math.ceil(0.05 / max(time.perf_counter() - start, 1e-9)))
+            )
+            row = {
+                "callbacks": count,
+                "payload_bytes": payload,
+                "mask": mask,
+                "batch_size": batch,
+                "outputs_matched": True,
+                "base": {"seconds": [], "peak_bytes": []},
+                "pr": {"seconds": [], "peak_bytes": []},
+            }
+            for repeat in range(REPEATS):
+                for version in (("base", "pr") if repeat % 2 == 0 else ("pr", "base")):
+                    gc.collect()
+                    start = time.perf_counter()
+                    for _ in range(batch):
+                        output = workflow(choices[version], count, payload)
+                    elapsed = (time.perf_counter() - start) / batch
+                    assert output == expected
+                    row[version]["seconds"].append(elapsed)
+                progress = (
+                    (len(report["results"]) + (repeat + 1) / REPEATS)
+                    / report["planned_rows"]
+                    * 100
+                )
+                print(
+                    f"{count} callbacks, {payload} B, {mask}: base + changed {repeat + 1}/5; overall {progress:.1f}%",
+                    flush=True,
+                )
+            for repeat in range(REPEATS):
+                for version in (("base", "pr") if repeat % 2 == 0 else ("pr", "base")):
+                    gc.collect()
+                    tracemalloc.start()
+                    try:
+                        output = workflow(choices[version], count, payload)
+                        peak = tracemalloc.get_traced_memory()[1]
+                    finally:
+                        tracemalloc.stop()
+                    assert output == expected
+                    row[version]["peak_bytes"].append(peak)
+            report["results"].append(row)
+            write_report(report)
+            print(
+                f"Saved row {len(report['results'])}/{report['planned_rows']}",
+                flush=True,
+            )
+    report["complete"] = True
+    write_report(report)
+def validate():
+    current = metadata()
+    for name in (
+        "tests_base.json",
+        "tests_pr.json",
+        "differential.json",
+        "benchmark_results.json",
+    ):
+        result = json.loads((HERE / name).read_text(encoding="utf-8"))
+        for key in (
+            "base_commit",
+            "python",
+            "source_hashes",
+            "runner_sha256",
+            "affinity_helper_sha256",
+            "dependencies",
+        ):
+            if result[key] != current[key]:
+                raise RuntimeError(f"Stale {name}: {key} changed; run -Check again")
+        if name.startswith("tests_"):
+            assert result["exit_code"] == 0 and result["tests_run"] > 0
+            assert (
+                not result["failures"]
+                and not result["errors"]
+                and not result["unexpected_successes"]
+            )
+        elif name == "differential.json":
+            assert result["comparisons"] >= 300 and not result["differences"]
+        else:
+            assert (
+                result["complete"] and len(result["results"]) == result["planned_rows"]
+            )
+            assert result["planned_rows"] == len(WORKLOADS) * len(result["cpu_pairs"])
+            seen = set()
+            for row in result["results"]:
+                key = row["callbacks"], row["payload_bytes"], row["mask"]
+                assert key not in seen and row["outputs_matched"]
+                seen.add(key)
+                for version in ("base", "pr"):
+                    assert (
+                        len(row[version]["seconds"])
+                        == len(row[version]["peak_bytes"])
+                        == REPEATS
+                    )
+            write_report(result)
+    print(
+        "Source hashes, tests, differential checks and complete raw benchmarks validated."
+    )
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--tests", choices=("base", "pr"))
+    group.add_argument("--verify", action="store_true")
+    group.add_argument("--benchmark", action="store_true")
+    group.add_argument("--validate", action="store_true")
+    args = parser.parse_args()
+    if args.validate:
+        validate()
+        return 0
+    cores = RotatingCores()
+    try:
+        if args.tests:
+            return run_tests(args.tests, cores)
+        setup()
+        choices = methods()
+        if args.verify:
+            verify(choices, cores)
+        else:
+            benchmark(choices, cores)
+        return 0
+    finally:
+        cores.restore()
+if __name__ == "__main__":
+    raise SystemExit(main())
+```
 | .Callbacks. | .Captured payload/callback. | .CPU mask. | ..Base time.. | .Changed time. | ....Speed.... | .Base peak.. | .Changed peak. | .Memory. |
 | ----------- | --------------------------- | ---------- | ------------- | -------------- | ------------- | ------------ | -------------- | -------- |