Spaces:
Running
Running
File size: 2,669 Bytes
16f323d 826c2df 16f323d 826c2df 16f323d 826c2df 16f323d 826c2df 16f323d 826c2df 16f323d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """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()
|