DecBench backend: checkpoint every 30 seconds and at the end, and resume from a checkpoint

The backend pickled the whole partial result after every function: 2.5 MB each
on bzip2 and 18 MB on bash, quadratic in the function count, which is why the
bash backend took 1,253 s and 1,723 s against about 140 s for r2s alone. It
now writes a checkpoint when R2SLEIGH_CHECKPOINT_SECONDS (default 30) have
passed since the last one, and once more when the run ends or fails, so a kill
loses one interval of answers at most.

A checkpoint carries the r2s build, the binary's sha256 and the targets with
their names. A re-run that finds one for the same task asks r2s only for the
addresses it has no answer for; one for another task is not read.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01456qmqCGpCWJkrg6JUZNNM

tests/decbench/README.md

@@ -75,8 +75,15 @@ function its own from the moment r2s starts on it
   or `harness: r2s <ending> while rendering <addr>` (a crash or deadline, after
   which r2s restarts at the next function) -- and
   `rendered + declined == requested` is checked.
-* **Checkpoints.** The partial result is pickled after every function, so the
-  driver's hard kill keeps what was answered.
+* **Checkpoints.** The partial result is pickled to the driver's progress path
+  every `R2SLEIGH_CHECKPOINT_SECONDS` (default 30) and at the end, so the driver's
+  hard kill loses one interval of answers at most. (One pickle per function was
+  quadratic: 2.5 MB each on bzip2, 18 MB on bash, and the backend took 9 to 13
+  times r2s alone.) A re-run that finds the checkpoint of the same r2s build,
+  binary bytes and targets asks r2s only for the addresses it has no answer for;
+  the official driver gives each attempt a fresh path, so this serves a manual
+  re-run. `r2s_batch.py` hands each answer over while r2s is still running, which
+  is what makes a checkpoint taken mid-process hold anything.
 * **Fails closed.** A binary carrying `.symtab` or any debug section
   (`.debug_*`, `.zdebug_*`, `.gnu_debugdata`) is declined
   whole: nothing r2s says about it is admissible.

tests/decbench/r2s_timing.py

@@ -4,9 +4,10 @@
     r2s_timing.py --r2s <r2s> --run <results tree> [--project NAME ...] > timing.tsv
 
 The official driver's wall time for r2s includes the backend's progress checkpoint, a pickle of the whole partial
-result after every function, which grows with the binary. This replays the same `s <addr>; pddj` stream through the
-same runner (`tests/equiv/r2s_batch.py`) with no checkpoint, and prints per binary: functions asked, rendered,
-declined, wall seconds (open, `aa`, every function) and the peak resident set of the r2s process in kB.
+result (every 30 s since 2026-10-23; after every function before, which grew with the binary). This replays the same
+`s <addr>; pddj` stream through the same runner (`tests/equiv/r2s_batch.py`) with no checkpoint, and prints per binary:
+functions asked, rendered, declined, wall seconds (open, `aa`, every function) and the peak resident set of the r2s
+process in kB.
 Targets are the addresses the run's own tomls name (rendered plus failed), so the two measure the same set.
 Each binary runs in its own Python process, serially, so the peak RSS is that binary's alone.
 """

tests/decbench/r2sleigh_raw.py

@@ -34,8 +34,14 @@ and the DWARF ``low_pc`` of every source function:
   or ``harness: <how r2s ended> while rendering <addr>`` (a crash or a
   deadline, after which r2s is restarted at the next function).
   ``rendered + declined == requested`` is checked before returning.
-* After every function the partial result is pickled to ``progress_path``, so
-  the driver's hard kill loses nothing already answered.
+* The partial result is pickled to ``progress_path`` every
+  ``$R2SLEIGH_CHECKPOINT_SECONDS`` (default 30) and once more at the end, so
+  the driver's hard kill loses the answers of one interval at most. A pickle
+  costs time proportional to every function answered so far, which is why it
+  is not written per function (2.5 MB each on bzip2, 18 MB on bash). A run that
+  finds a checkpoint of the same r2s, binary and targets at ``progress_path``
+  asks r2s only for the addresses it has no answer for; the official driver
+  gives each attempt a fresh path, so that is for a manual re-run.
 * It fails closed on a binary carrying ``.symtab`` or any debug section
   (``.debug_*``, compressed ``.zdebug_*``, or the MiniDebugInfo
   ``.gnu_debugdata``): r2s reads the declarations of the file it opens, and
@@ -54,7 +60,9 @@ from __future__ import annotations
 import hashlib
 import json
 import logging
+import math
 import os
+import pickle
 import re
 import shutil
 import struct
@@ -84,6 +92,9 @@ log = logging.getLogger(__name__)
 
 BACKEND = "r2sleigh_native"
 
+CHECKPOINT_SECONDS = 30.0
+_clock = time.monotonic  # replaced in tests, which never sleep
+
 # Sections whose presence means r2s would read facts DecBench grades against:
 # the symbol table, and debug information in any of its spellings.
 FAIL_CLOSED_SECTIONS = (".symtab", ".gnu_debugdata")
@@ -171,6 +182,35 @@ def discover(executable: Path, binary: Path, timeout: float) -> tuple[list[tuple
     return functions, ending
 
 
+def checkpoint_interval() -> float:
+    """Seconds between checkpoints: ``$R2SLEIGH_CHECKPOINT_SECONDS``, 30 by default, 0 for every answer."""
+    try:
+        seconds = float(os.environ.get("R2SLEIGH_CHECKPOINT_SECONDS", CHECKPOINT_SECONDS))
+    except ValueError:
+        return CHECKPOINT_SECONDS
+    return max(0.0, seconds) if math.isfinite(seconds) else CHECKPOINT_SECONDS
+
+
+def resume_state(progress_path: Path | None, identity: dict) -> tuple[dict, dict] | None:
+    """The rendered functions and declines of a checkpoint of this very task, or None.
+
+    ``identity`` names the r2s build, the binary's bytes and the targets with their names; a checkpoint
+    of anything else is not read, so a stale file cannot put another run's answers into this one.
+    """
+    if progress_path is None or not Path(progress_path).is_file():
+        return None
+    try:
+        saved = pickle.loads(Path(progress_path).read_bytes())
+        block = (saved.decompiler.extra or {}).get("resume")
+        if not isinstance(block, dict) or block.get("identity") != identity:
+            return None
+        declined = {int(address): (str(label), str(cause))
+                    for address, label, cause in block["declined"]}
+        return dict(saved.functions), declined
+    except Exception:  # noqa: BLE001 - a checkpoint that cannot be read is no checkpoint
+        return None
+
+
 def stack_offset(location: str) -> int | None:
     """``stack-0x14``, ``[rbp-0x14]`` or ``sp+8`` -> the signed offset."""
     found = _STACK_OFFSET.search(location.strip())
@@ -302,7 +342,7 @@ class R2sDecompiler(Decompiler):
         rendered: dict[str, FunctionDecompilation] = {}
         extra: dict = {"requested_from": source}
 
-        def result() -> DecompilationResult:
+        def result(resume: dict | None = None) -> DecompilationResult:
             extra.update({
                 "requested": len(requested),
                 "rendered": len(rendered),
@@ -317,7 +357,7 @@ class R2sDecompiler(Decompiler):
                     decompiler_version=self.get_version(),
                     total_time_seconds=time.time() - started,
                     failed_functions=sorted(label for label, _ in declined.values()),
-                    extra=dict(extra),
+                    extra={**extra, "resume": resume} if resume is not None else dict(extra),
                 ),
                 functions=dict(rendered),
                 output_dir=output_dir,
@@ -354,6 +394,34 @@ class R2sDecompiler(Decompiler):
                                + (f" ({ending})" if ending else ""))
                 return result()
 
+        # The checkpoint names its task, so only a re-run of the same r2s, bytes and targets reads it.
+        identity = {
+            "version": self.get_version(),
+            "binary": binary_path.name,
+            "sha256": hashlib.sha256(binary_path.read_bytes()).hexdigest()
+            if progress_path is not None else "",
+            "requested": [[address, name] for address, name in requested.items()],
+        }
+        resumed = resume_state(progress_path, identity)
+        if resumed is not None:
+            rendered.update(resumed[0])
+            declined.update(resumed[1])
+            extra["resumed"] = len(rendered) + len(declined)
+        answered = {function.address for function in rendered.values()} | set(declined)
+
+        interval = checkpoint_interval()
+        saved_at = _clock()
+
+        def checkpoint() -> None:
+            nonlocal saved_at
+            saved_at = _clock()
+            if progress_path is not None:
+                common.dump_progress(progress_path, result({
+                    "identity": identity,
+                    "declined": [[address, label, cause]
+                                 for address, (label, cause) in declined.items()],
+                }))
+
         def file_answer(answer: "r2s_batch.Answer") -> None:
             name = requested.get(answer.address)
             key = name or f"0x{answer.address:x}"
@@ -364,12 +432,16 @@ class R2sDecompiler(Decompiler):
                 rendered[function.name] = function
             else:
                 declined[answer.address] = (key, answer.cause or "harness: no answer")
-            common.dump_progress(progress_path, result())
+            if _clock() - saved_at >= interval:
+                checkpoint()
 
-        report = r2s_batch.run_batch(
-            executable, binary_path, list(requested), function_timeout=timeout,
-            startup_timeout=startup, on_answer=file_answer,
-        )
+        try:
+            report = r2s_batch.run_batch(
+                executable, binary_path, [a for a in requested if a not in answered],
+                function_timeout=timeout, startup_timeout=startup, on_answer=file_answer,
+            )
+        finally:
+            checkpoint()
         extra["processes"] = report.processes
         extra["process_endings"] = report.endings
         if len(rendered) + len(declined) != len(requested):

tests/decbench/test_r2sleigh_raw.py

@@ -14,15 +14,18 @@ r2s is replaced by ``tests/equiv/testdata/stub_r2s.py``, which answers
 
 from __future__ import annotations
 
+import itertools
 import os
 import pickle
 import shutil
 import subprocess
 import sys
 import tempfile
+import time
 import types
 import unittest
 from pathlib import Path
+from unittest import mock
 
 HERE = Path(__file__).resolve().parent
 REPO = HERE.parents[1]
@@ -107,6 +110,8 @@ TARGETS = [0x401150, 0x401160, 0x4011F0, 0x401210, 0x401230, 0x401290]
 
 
 class Backend(unittest.TestCase):
+    runs = itertools.count()
+
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
         self.env = {
@@ -154,24 +159,12 @@ class Backend(unittest.TestCase):
         self.assertEqual(result.decompiler.extra["processes"], 1)
         self.assertEqual(result.decompiler.failed_functions, [f"0x{TARGETS[1]:x}"])
 
-    def test_a_crash_costs_one_function_and_every_answer_is_checkpointed(self):
+    def test_a_crash_costs_one_function_and_the_end_is_checkpointed(self):
         # D4: abort on the third of six; r2s restarts at the fourth, and the
-        # progress pickle grows by one function per answer.
+        # pickle the driver would recover holds the five answers.
         os.environ["STUB_R2S_FAULTS"] = f"abort@{TARGETS[2]:x}"
-        sizes: list[int] = []
-        original = r2sleigh_raw.common.dump_progress
-
-        def spy(path, result):
-            sizes.append(len(result.functions) + len(result.decompiler.extra["decline_causes"]))
-            original(path, result)
-
-        r2sleigh_raw.common.dump_progress = spy
-        try:
-            progress = Path(self.tmp.name) / "progress.pkl"
-            result = self.decompile(function_names=set(TARGETS), progress_path=progress)
-        finally:
-            r2sleigh_raw.common.dump_progress = original
-        self.assertEqual(sizes, [1, 2, 3, 4, 5, 6])
+        progress = Path(self.tmp.name) / "progress.pkl"
+        result = self.decompile(function_names=set(TARGETS), progress_path=progress)
         causes = result.decompiler.extra["decline_causes"]
         self.assertEqual(list(causes), [f"0x{TARGETS[2]:x}"])
         self.assertIn("SIGABRT", causes[f"0x{TARGETS[2]:x}"])
@@ -181,6 +174,142 @@ class Backend(unittest.TestCase):
         saved = pickle.loads(progress.read_bytes())
         self.assertEqual(len(saved.functions), 5)
         self.assertEqual(result.decompiler.extra["processes"], 2)
+        self.assertNotIn("resume", result.decompiler.extra)
+
+    def checkpoint_rows(self, answer_times, interval="30"):
+        """The rows held by each checkpoint of a run whose answers arrive at ``answer_times`` seconds."""
+        clock = [0.0]
+        rows: list[int] = []
+        os.environ["R2SLEIGH_CHECKPOINT_SECONDS"] = interval
+        real_batch = r2sleigh_raw.r2s_batch.run_batch
+        real_dump = r2sleigh_raw.common.dump_progress
+
+        def timed(*args, on_answer=None, **kwargs):
+            moments = iter(answer_times)
+
+            def arrive(answer):
+                clock[0] = next(moments)
+                on_answer(answer)
+
+            return real_batch(*args, on_answer=arrive, **kwargs)
+
+        def spy(path, result):
+            rows.append(len(result.functions) + len(result.decompiler.extra["decline_causes"]))
+            real_dump(path, result)
+
+        try:
+            r2sleigh_raw.r2s_batch.run_batch = timed
+            r2sleigh_raw.common.dump_progress = spy
+            r2sleigh_raw._clock = lambda: clock[0]
+            # A path of its own: a checkpoint left by an earlier call would resume.
+            self.decompile(function_names=set(TARGETS),
+                           progress_path=Path(self.tmp.name) / f"cadence-{next(self.runs)}.pkl")
+        finally:
+            r2sleigh_raw.r2s_batch.run_batch = real_batch
+            r2sleigh_raw.common.dump_progress = real_dump
+            r2sleigh_raw._clock = time.monotonic
+        return rows
+
+    def test_a_checkpoint_is_written_by_elapsed_time_and_at_the_end(self):
+        # Six answers in one second are one pickle, not six; a pause of 30 seconds is one more.
+        self.assertEqual(self.checkpoint_rows([1, 2, 3, 4, 5, 6]), [6])
+        self.assertEqual(self.checkpoint_rows([1, 2, 31, 32, 33, 70]), [3, 6, 6])
+        self.assertEqual(self.checkpoint_rows([1, 2, 3, 4, 5, 6], interval="0"),
+                         [1, 2, 3, 4, 5, 6, 6])
+
+    def test_the_interval_is_thirty_seconds_unless_the_environment_says_otherwise(self):
+        os.environ.pop("R2SLEIGH_CHECKPOINT_SECONDS", None)
+        self.assertEqual(r2sleigh_raw.checkpoint_interval(), 30.0)
+        for given, expected in (("5", 5.0), ("0", 0.0), ("-3", 0.0), ("soon", 30.0),
+                                ("nan", 30.0), ("inf", 30.0)):
+            os.environ["R2SLEIGH_CHECKPOINT_SECONDS"] = given
+            self.assertEqual(r2sleigh_raw.checkpoint_interval(), expected, given)
+
+    def asked_addresses(self):
+        """Patch the runner to record the addresses it is asked about; returns the list and an undo."""
+        asked: list[int] = []
+        real_batch = r2sleigh_raw.r2s_batch.run_batch
+
+        def spy(executable, binary, addresses, **kwargs):
+            asked.extend(addresses)
+            return real_batch(executable, binary, addresses, **kwargs)
+
+        r2sleigh_raw.r2s_batch.run_batch = spy
+        self.addCleanup(setattr, r2sleigh_raw.r2s_batch, "run_batch", real_batch)
+        return asked
+
+    @unittest.skipUnless(hasattr(os, "fork"), "needs fork to kill a run without cleanup")
+    def test_a_killed_run_resumes_from_its_last_checkpoint_and_gives_the_same_rows(self):
+        progress = Path(self.tmp.name) / "progress.pkl"
+        os.environ["R2SLEIGH_CHECKPOINT_SECONDS"] = "30"
+        clock = [0.0]
+        real_batch = r2sleigh_raw.r2s_batch.run_batch
+
+        def dying(*args, on_answer=None, **kwargs):
+            given = []
+
+            def arrive(answer):
+                clock[0] += 10.0  # a checkpoint on every third answer
+                on_answer(answer)
+                given.append(answer)
+                if len(given) == 5:
+                    os._exit(9)  # SIGKILL's view of the run: no handler, no `finally`
+
+            return real_batch(*args, on_answer=arrive, **kwargs)
+
+        pid = os.fork()
+        if pid == 0:
+            code = 2
+            try:
+                r2sleigh_raw._clock = lambda: clock[0]
+                r2sleigh_raw.r2s_batch.run_batch = dying
+                self.decompile(function_names=set(TARGETS), progress_path=progress)
+            except BaseException:  # noqa: BLE001
+                code = 3
+            finally:
+                os._exit(code)  # never fall back into the test runner
+        _, status = os.waitpid(pid, 0)
+        self.assertEqual(os.WEXITSTATUS(status), 9)
+        # Five answers were given; the third's checkpoint is the last, so two are lost, one interval at most.
+        kept = pickle.loads(progress.read_bytes())
+        self.assertEqual(sorted(f.address for f in kept.functions.values()), TARGETS[:3])
+
+        asked = self.asked_addresses()
+        resumed = self.decompile(function_names=set(TARGETS), progress_path=progress)
+        self.assertEqual(asked, TARGETS[3:])
+        self.assertEqual(resumed.decompiler.extra["resumed"], 3)
+        fresh = self.decompile(function_names=set(TARGETS))
+
+        def rows(result):
+            return ({name: (f.address, f.decompiled_code, f.metadata)
+                     for name, f in result.functions.items()},
+                    result.decompiler.extra["decline_causes"],
+                    result.decompiler.failed_functions,
+                    (result.decompiler.extra["requested"], result.decompiler.extra["rendered"],
+                     result.decompiler.extra["declined"]))
+
+        self.assertEqual(rows(resumed), rows(fresh))
+
+    def test_a_checkpoint_of_another_task_is_not_read(self):
+        progress = Path(self.tmp.name) / "progress.pkl"
+        self.decompile(function_names=set(TARGETS[:3]), progress_path=progress)
+        asked = self.asked_addresses()
+        # The checkpoint holds three targets, this run has six: nothing carries over.
+        result = self.decompile(function_names=set(TARGETS), progress_path=progress)
+        self.assertEqual(asked, TARGETS)
+        self.assertNotIn("resumed", result.decompiler.extra)
+        # The same six, another r2s build: nothing carries over either.
+        other = Path(self.tmp.name) / "other-build.pkl"
+        shutil.copyfile(progress, other)
+        asked.clear()
+        with mock.patch.object(r2sleigh_raw.R2sDecompiler, "get_version",
+                               return_value="r2s sha256:another"):
+            self.decompile(function_names=set(TARGETS), progress_path=other)
+        self.assertEqual(asked, TARGETS)
+        # The same task: the end-of-run checkpoint answers everything.
+        asked.clear()
+        again = self.decompile(function_names=set(TARGETS), progress_path=progress)
+        self.assertEqual((asked, again.decompiler.extra["resumed"]), ([], 6))
 
     def test_thousands_of_targets_are_one_process_and_every_one_is_filed(self):
         # The sailr set's large binaries have thousands of functions. As one -c