Spaces:
Running
Running
| """Run verification in an interruptible process.""" | |
| from contextlib import closing | |
| import multiprocessing | |
| import os | |
| import signal | |
| import tempfile | |
| def _verify_worker(task, connection, directory, recipe=None): | |
| # Compilers inherit this group so cancellation also stops child processes. | |
| os.setsid() | |
| tempfile.tempdir = directory | |
| from verifier import replay, verify | |
| try: | |
| if recipe is not None: | |
| connection.send(("report", replay(task, recipe))) | |
| else: | |
| 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, recipe=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: | |
| args = (task, sender, directory) if recipe is None else (task, sender, directory, recipe) | |
| process = context.Process(target=_worker, args=args) | |
| 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() | |