94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
"""Bounded disposable-process observation; never used by production retirement."""
|
|
from dataclasses import dataclass
|
|
import os
|
|
import select
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
|
|
@dataclass
|
|
class Result:
|
|
returncode: int
|
|
stdout: bytes
|
|
stderr: bytes
|
|
discarded: int
|
|
|
|
|
|
class FixtureTimeout(TimeoutError):
|
|
def __init__(self, pid, output, errors, discarded, reaped):
|
|
self.pid, self.stdout, self.stderr = pid, output, errors
|
|
self.discarded, self.reaped = discarded, reaped
|
|
super().__init__(f'fixture deadline: pid={pid} reaped={reaped} discarded={discarded} '
|
|
f'stdout={output!r} stderr={errors!r}')
|
|
|
|
|
|
def run_bounded(args, *, seconds=5, observe=None, **kwargs):
|
|
"""One deadline includes sampling, simultaneous pipe draining and exit.
|
|
|
|
Retain WNOWAIT leader identity until descendant-group cleanup, then reap
|
|
nonblockingly. An additional finite cleanup bound belongs ONLY to fixtures;
|
|
failure reports an unreaped PID for the outer disposable unit to contain.
|
|
"""
|
|
end = time.monotonic() + seconds
|
|
process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
start_new_session=True, **kwargs)
|
|
streams = {process.stdout.fileno(): bytearray(), process.stderr.fileno(): bytearray()}
|
|
output, errors = streams.values()
|
|
pending = set(streams)
|
|
discarded = 0
|
|
timed_out = False
|
|
reaped = False
|
|
limit = 64 * 1024 # retain bounded diagnostics while always draining pipes
|
|
|
|
def kill_group():
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
try:
|
|
for fd in pending:
|
|
os.set_blocking(fd, False)
|
|
exited = False
|
|
while pending or not exited:
|
|
if time.monotonic() >= end:
|
|
timed_out = True
|
|
break
|
|
if observe is not None:
|
|
observe(process.pid)
|
|
# Never reap the leader before killing any remaining descendants:
|
|
# its PID/group identity cannot be reused during this observation.
|
|
status = os.waitid(os.P_PID, process.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT)
|
|
if status is not None and not exited:
|
|
exited = True
|
|
kill_group()
|
|
ready, _, _ = select.select(list(pending), [], [], min(0.01, max(0, end - time.monotonic())))
|
|
for fd in ready:
|
|
chunk = os.read(fd, 65536)
|
|
if not chunk:
|
|
pending.remove(fd)
|
|
continue
|
|
keep = min(len(chunk), limit - len(streams[fd]))
|
|
streams[fd].extend(chunk[:keep])
|
|
discarded += len(chunk) - keep
|
|
finally:
|
|
failure = sys.exc_info()[1]
|
|
# Also run on observer/assertion failures. All signals precede reaping.
|
|
kill_group()
|
|
cleanup_end = time.monotonic() + 2
|
|
while time.monotonic() < cleanup_end:
|
|
if process.poll() is not None:
|
|
reaped = True
|
|
break
|
|
time.sleep(0.005)
|
|
process.stdout.close()
|
|
process.stderr.close()
|
|
if failure is not None:
|
|
failure.add_note(f'fixture cleanup: pid={process.pid} reaped={reaped} discarded={discarded} '
|
|
f'stdout={bytes(output)!r} stderr={bytes(errors)!r}')
|
|
if timed_out or not reaped:
|
|
raise FixtureTimeout(process.pid, bytes(output), bytes(errors), discarded, reaped)
|
|
return Result(process.returncode, bytes(output), bytes(errors), discarded)
|