Update run defaults and make verification workers cancellable

core

@@ -1 +1 @@
-Subproject commit 989dc2dbe7fc787b2ad33353413c9430b8bd2446
+Subproject commit 7258a0c4fe81545b0de6baf4256d6cd6ee60e01d

core-revision.json

@@ -1,5 +1,5 @@
 {
   "repository": "https://github.com/edmcman/decompile-verify.git",
-  "commit": "989dc2dbe7fc787b2ad33353413c9430b8bd2446",
+  "commit": "7258a0c4fe81545b0de6baf4256d6cd6ee60e01d",
   "version": "0.0.0"
 }

demo/app.py

@@ -20,6 +20,7 @@ from demo.presentation import banner, format_bit_ranges, json_detail, mismatch_s
 from demo.session import (UISession, begin_ui, cancel_session, controls, edit_inputs, idle_controls,
                           reset_session, stream_ui)
 import verifier as api
+from demo.execution import verification_reports
 from verifier import Task, integer_text, parse_bindings, parse_function, yaml_mapping
 
 HEADERS = ["Seed", "Verdict", "Location", "Differing bits", "Original", "Candidate", "Reason"]
@@ -81,18 +82,20 @@ def show_assembly(binary, function, candidate, bindings, candidate_symbol,
 
 
 def run_verification(binary, function, candidate, bindings, candidate_symbol,
-                     seed_count, seed_start, instruction_budget, compiler, flags, interfaces="", *, report_directory=None):
+                     seed_count, seed_start, instruction_budget, compiler, flags, interfaces="", *, report_directory=None, cancelled=None):
     """Translate UI inputs, stream engine results, then supply the JSON download."""
     try:
         task = task_from_inputs(binary, function, candidate, bindings, candidate_symbol,
                                 seed_count, seed_start, instruction_budget, compiler, flags, interfaces)
-        rows, detail = [], None
-        with closing(api.verify(task, stream=True)) as iterator:
+        rows, detail, report = [], None, None
+        with closing(verification_reports(task, cancelled=cancelled)) as iterator:
             for report in iterator:
                 rows.append(result_row(report.results[-1]))
                 if detail is None and report.first_mismatch is not None:
                     detail = asdict(report.first_mismatch)
                 yield banner(report), list(rows), detail, None, report.original_assembly, report.candidate_assembly
+        if cancelled is not None and cancelled.is_set():
+            return
         output = report.to_json()
         directory = Path(report_directory) if report_directory else Path(tempfile.gettempdir()) / "decompile-verify-reports"
         directory.mkdir(parents=True, exist_ok=True)
@@ -177,7 +180,7 @@ aid; verification runs the binary.
                            value="globals: {}\ncallees: {}\n", render=False)
         seed_count = gr.Number(label="Seed count", value=100, precision=0, minimum=1, render=False)
         seed_start = gr.Textbox(label="First seed", value="0", info="Nonnegative decimal or hex integer", render=False)
-        instruction_budget = gr.Number(label="Instruction budget per run", value=100000, precision=0,
+        instruction_budget = gr.Number(label="Instruction budget per run", value=1000, precision=0,
                                        minimum=1, render=False)
         compiler = gr.Dropdown(label="Candidate compiler", choices=["gcc", "clang"], value="gcc", render=False)
         flags = gr.Textbox(label="Candidate flags", value="-O2",
@@ -245,9 +248,11 @@ aid; verification runs the binary.
 
         def verify_ui(binary, function, candidate, bindings, candidate_symbol,
                       seed_count, seed_start, instruction_budget, compiler, flags, interfaces, session, ticket):
-            for update in stream_ui(run_verification, (binary, function, candidate, bindings, candidate_symbol,
-                                                      seed_count, seed_start, instruction_budget, compiler, flags, interfaces), session, ticket):
-                yield update[0], update[1], json_detail(update[2]), *update[3:]
+            with closing(stream_ui(run_verification, (binary, function, candidate, bindings, candidate_symbol,
+                                                      seed_count, seed_start, instruction_budget, compiler, flags, interfaces),
+                                   session, ticket, cancellable=True)) as updates:
+                for update in updates:
+                    yield update[0], update[1], json_detail(update[2]), *update[3:]
 
         def reset_ui(session):
             reset_session(session)
@@ -275,10 +280,9 @@ aid; verification runs the binary.
         verify_start = run.click(begin_ui, [session, *inputs], [verdict, *buttons, ticket],
                                  queue=False, api_visibility="private", show_progress="hidden")
         preview_event = gr.on(triggers=[], fn=show_assembly, inputs=inputs, outputs=outputs,
-                              api_name="assembly", concurrency_limit=1,
-                              concurrency_id="workbench", show_progress="hidden")
-        event = verify_start.then(verify_ui, [*inputs, session, ticket], outputs, api_name="verify", concurrency_limit=1,
-                                  concurrency_id="workbench", show_progress="hidden")
+                              api_name="assembly", concurrency_limit=None, show_progress="hidden")
+        event = verify_start.then(verify_ui, [*inputs, session, ticket], outputs, api_name="verify",
+                                  concurrency_limit=None, show_progress="hidden")
         event.then(idle_controls, session, buttons, queue=False,
                    api_visibility="private", show_progress="hidden")
         gr.on([binary.upload, binary.delete, binary.clear, *[component.input for component in inputs[1:]]],

demo/execution.py

@@ -0,0 +1,68 @@
+"""Run verification in an interruptible process."""
+
+from contextlib import closing
+import multiprocessing
+import os
+import signal
+import tempfile
+
+
+def _verify_worker(task, connection, directory):
+    # Compilers inherit this group so cancellation also stops child processes.
+    os.setsid()
+    tempfile.tempdir = directory
+    from verifier import verify
+
+    try:
+        with closing(verify(task, stream=True)) as reports:
+            for report in reports:
+                connection.send(("report", report))
+        connection.send(("done", None))
+    except Exception as error:
+        connection.send(("error", str(error)))
+    finally:
+        connection.close()
+
+
+def verification_reports(task, *, cancelled=None, _worker=_verify_worker):
+    """Stream results until completion or cancellation."""
+    if cancelled is not None and cancelled.is_set():
+        return
+    context = multiprocessing.get_context("spawn")
+    receiver, sender = context.Pipe(duplex=False)
+    with tempfile.TemporaryDirectory(prefix="verify-request-") as directory:
+        process = context.Process(target=_worker, args=(task, sender, directory))
+        started = False
+        try:
+            process.start()
+            started = True
+            sender.close()
+            while True:
+                if cancelled is not None and cancelled.is_set():
+                    return
+                # Poll briefly so cancellation can stop a blocked worker.
+                if not receiver.poll(0.1):
+                    continue
+                if cancelled is not None and cancelled.is_set():
+                    return
+                try:
+                    kind, value = receiver.recv()
+                except EOFError as error:
+                    raise ValueError("Verification worker exited without a result") from error
+                if kind == "done":
+                    return
+                if kind == "error":
+                    raise ValueError(value)
+                yield value
+        finally:
+            sender.close()
+            receiver.close()
+            if started:
+                # Kill the group even when the worker has exited but a child remains.
+                try:
+                    os.killpg(process.pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    if process.is_alive():
+                        process.kill()
+                process.join()
+                process.close()

demo/session.py

@@ -1,7 +1,7 @@
 """Private, per-browser state for displayed inputs and cancellable result streams."""
 
 from contextlib import closing
-from threading import RLock
+from threading import Event, RLock
 
 from demo.presentation import ABI_SLOT, status, with_stale
 
@@ -13,6 +13,7 @@ class UISession:
         self.busy = False
         self.display = ""
         self.lock = RLock()
+        self.cancelled = Event()
 
     def __deepcopy__(self, memo):
         # Each Gradio session gets its own state and synchronization primitive.
@@ -33,6 +34,7 @@ def edit_inputs(session, *values):
 
 def reset_session(session):
     with session.lock:
+        session.cancelled.set()
         session.generation += 1
         session.snapshot = session.current = None
         session.busy = False
@@ -41,6 +43,7 @@ def reset_session(session):
 
 def cancel_session(session):
     with session.lock:
+        session.cancelled.set()
         session.generation += 1
         session.busy = False
         _, separator, abi = session.display.partition(ABI_SLOT)
@@ -61,14 +64,16 @@ def idle_controls(session):
 
 def begin_ui(session, *values):
     with session.lock:
+        session.cancelled.set()
         session.generation += 1
+        session.cancelled = Event()
         session.snapshot = session.current = tuple(values)
         session.busy = True
         session.display = status("Building")
         return session.rendered(), *controls(True), session.generation
 
 
-def stream_ui(handler, values, session, ticket=None):
+def stream_ui(handler, values, session, ticket=None, *, cancellable=False):
     """Keep the six API outputs while protecting this session from obsolete streams."""
     with session.lock:
         if ticket is not None:
@@ -78,10 +83,12 @@ def stream_ui(handler, values, session, ticket=None):
         else:
             begin_ui(session, *values)
         generation = session.generation
+        cancelled = session.cancelled
         initial = session.rendered()
     try:
         yield initial, [], None, None, "", ""
-        with closing(handler(*values)) as iterator:
+        options = {"cancelled": cancelled} if cancellable else {}
+        with closing(handler(*values, **options)) as iterator:
             for update in iterator:
                 with session.lock:
                     if session.generation != generation:

tests/test_execution.py

@@ -0,0 +1,53 @@
+"""Exercise worker cleanup without requiring Linux ELF compilers."""
+
+import os
+from time import sleep
+
+import pytest
+
+from demo.execution import verification_reports
+
+
+def successful_worker(task, connection, directory):
+    os.setsid()
+    connection.send(("report", task))
+    connection.send(("done", None))
+    connection.close()
+
+
+def failing_worker(task, connection, directory):
+    os.setsid()
+    connection.send(("error", "invalid candidate"))
+    connection.close()
+
+
+def stalled_worker(task, connection, directory):
+    os.setsid()
+    if task:
+        connection.send(("report", os.getpid()))
+    sleep(60)
+
+
+def test_reports_and_errors():
+    assert list(verification_reports("report", _worker=successful_worker)) == ["report"]
+    with pytest.raises(ValueError, match="invalid candidate"):
+        list(verification_reports(None, _worker=failing_worker))
+
+
+
+
+def test_closing_stream_stops_worker():
+    reports = verification_reports(True, _worker=stalled_worker)
+    pid = next(reports)
+    reports.close()
+    with pytest.raises(ProcessLookupError):
+        os.kill(pid, 0)
+
+
+def test_verifier_worker_reports_build_errors(tmp_path):
+    from verifier import Task
+
+    task = Task(binary=tmp_path / "missing.elf", function=0x401000,
+                candidate="unsigned target(void) { return 0; }")
+    with pytest.raises(ValueError, match="missing.elf"):
+        list(verification_reports(task))

tests/test_execution_cancel.py

@@ -0,0 +1,105 @@
+"""Exercise cancellation while a real spawned worker is blocked before a result."""
+from concurrent.futures import ThreadPoolExecutor
+import os
+from pathlib import Path
+import subprocess
+from threading import Event
+import time
+
+import pytest
+
+from demo.execution import verification_reports
+from demo.session import UISession, begin_ui, cancel_session, reset_session, stream_ui
+
+
+def blocked_worker(task, connection, directory):
+    os.setsid()
+    child = subprocess.Popen(["sleep", "30"])
+    Path(directory, "unfinished").write_text("temporary build")
+    Path(task).write_text(f"{os.getpid()} {child.pid}\n{directory}")
+    if not Path(task).name.startswith("build"):
+        connection.send(("report", "ready"))
+    time.sleep(30)
+
+
+def completed_worker(task, connection, directory):
+    os.setsid()
+    connection.send(("report", "completed"))
+    connection.send(("done", None))
+    connection.close()
+
+
+def assert_stopped(pid):
+    # A killed compiler child may remain briefly as a zombie until reaped by init.
+    result = subprocess.run(["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True)
+    assert not result.stdout.strip() or result.stdout.strip().startswith("Z")
+
+
+@pytest.mark.parametrize("stop", [cancel_session, reset_session])
+def test_cancel_blocked_build_releases_worker_and_allows_next_run(tmp_path, stop):
+    session = UISession()
+    marker = tmp_path / "worker"
+
+    def handler(*values, cancelled):
+        for value in verification_reports(marker, cancelled=cancelled, _worker=blocked_worker):
+            yield value, [], None, None, "", ""
+
+    ticket = begin_ui(session, "source")[-1]
+    old_signal = session.cancelled
+    updates = stream_ui(handler, (), session, ticket, cancellable=True)
+    next(updates)
+    assert next(updates)[0] == "ready"
+    pids, directory = marker.read_text().splitlines()
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        waiting = executor.submit(lambda: list(updates))
+        time.sleep(0.2)
+        assert not waiting.done()
+        stop(session)
+        new_ticket = begin_ui(session, "new source")[-1]
+        assert old_signal.is_set() and not session.cancelled.is_set()
+        assert waiting.result(timeout=3) == []
+    assert session.busy and session.generation == new_ticket
+    assert not Path(directory).exists()
+    for pid in map(int, pids.split()):
+        assert_stopped(pid)
+    assert list(verification_reports(None, _worker=completed_worker)) == ["completed"]
+
+
+
+
+def test_close_stream_cleans_up_worker(tmp_path):
+    marker = tmp_path / "worker"
+    reports = verification_reports(marker, _worker=blocked_worker)
+    assert next(reports) == "ready"
+    reports.close()
+    pids, directory = marker.read_text().splitlines()
+    assert not Path(directory).exists()
+    for pid in map(int, pids.split()):
+        assert_stopped(pid)
+
+
+def test_already_cancelled_does_not_start_worker(tmp_path):
+    cancelled = Event()
+    cancelled.set()
+    marker = tmp_path / "worker"
+    assert list(verification_reports(marker, cancelled=cancelled, _worker=blocked_worker)) == []
+    assert not marker.exists()
+
+
+def test_cancel_before_first_build_result(tmp_path):
+    marker = tmp_path / "build-worker"
+    cancelled = Event()
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        waiting = executor.submit(lambda: list(verification_reports(
+            marker, cancelled=cancelled, _worker=blocked_worker)))
+        deadline = time.monotonic() + 10
+        while not marker.exists():
+            assert time.monotonic() < deadline
+            time.sleep(0.02)
+        assert not waiting.done()
+        cancelled.set()
+        assert waiting.result(timeout=3) == []
+    pids, directory = marker.read_text().splitlines()
+    assert not Path(directory).exists()
+    for pid in map(int, pids.split()):
+        assert_stopped(pid)

tests/test_faultless.py

@@ -13,29 +13,30 @@ def test_faultless_frontend_stream(tmp_path, compiler, name, example_rows):
     rows = example_rows
     args = rows[EXAMPLE_NAMES.index(name)].copy()
     args[8] = compiler
+    seed_count = args[5]
     updates = list(run_verification(*args, report_directory=tmp_path / 'reports'))
     report = json.loads(Path(updates[-1][3]).read_text())
-    assert len(updates) == 11
-    assert report['completed'] == 10
+    assert len(updates) == seed_count + 1
+    assert report['completed'] == seed_count
     expected = example_metadata(name)['expected_verdict']
     if name == 'faultless-byte-reinterpretation' and compiler == 'clang':
         # The original is built with gcc, so the escaped frames differ in layout.
         expected = 'inconclusive'
         assert all(r['reason'] == 'escaped_frame_differs' for r in report['results'])
     assert report['verdict'] == expected
-    assert [r['seed'] for r in report['results']] == list(range(10))
+    assert [r['seed'] for r in report['results']] == list(range(seed_count))
     if name == 'faultless-linked-list':
-        assert report['counts']['inconclusive'] == 10
+        assert report['counts']['inconclusive'] == seed_count
     elif name == 'faultless-deep-pointer':
-        assert report['counts']['mismatch'] == 10
+        assert report['counts']['mismatch'] == seed_count
         assert all(r['divergence'] == 'call[1]' for r in report['results'])
     elif name == 'faultless-heap-loop':
         assert report['counts']['agree'] > 0
         assert report['counts']['mismatch'] > 0
     elif name == 'faultless-memory-dependency':
-        assert report['counts']['agree'] == 10
+        assert report['counts']['agree'] == seed_count
     elif name == 'faultless-byte-reinterpretation':
-        assert report['counts'][expected] == 10
+        assert report['counts'][expected] == seed_count
     else:
-        assert report['counts']['mismatch'] == 10
+        assert report['counts']['mismatch'] == seed_count
         assert all(r['divergence'] == 'call[0]' for r in report['results'])