diff --git a/scripts/qualify-devguard.py b/scripts/qualify-devguard.py index ec1b65b..467e735 100644 --- a/scripts/qualify-devguard.py +++ b/scripts/qualify-devguard.py @@ -39,6 +39,7 @@ from pathlib import Path import platform import re +import resource import secrets import shutil import signal @@ -727,9 +728,10 @@ def scenario_authority_unavailable(bench): def scenario_runner_socket(bench): """`uds` with a worker the gateway did not start: governed executions are unsupported.""" socket = bench.work / 'standalone.sock' - worker = subprocess.Popen([str(bench.args.runtime_bin), str(socket)], stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, stderr=open(bench.evidence.file(f'{bench.mode}/standalone-worker.log'), 'ab'), - env=host.minimal_env(bench.work), start_new_session=True) + with open(bench.evidence.file(f'{bench.mode}/standalone-worker.log'), 'ab') as log: + worker = subprocess.Popen([str(bench.args.runtime_bin), str(socket)], stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=log, env=host.minimal_env(bench.work), + start_new_session=True) bench.live.workers.append(worker) deadline = time.monotonic() + 15 while not socket.exists() and time.monotonic() < deadline and worker.poll() is None: @@ -2258,7 +2260,7 @@ def teardown(self): if gateway.process.poll() is None: gateway.process.kill() gateway.process.wait() - gateway.log_reader.join(timeout=10) + gateway.join_readers() stopped = [runner.stop(limit=20) for runner in self.runners if runner.alive()] killed = kill_leftovers(self.roots.values(), self.began_unix) self.evidence.append(f'{self.prefix}/teardown.jsonl', {'runners_stopped': stopped, 'leftovers_killed': killed}) @@ -2667,10 +2669,10 @@ def recovery_invalid_settings(bench): managed_dir = bench.work / 'mw' managed_dir.mkdir(mode=0o700) managed_socket = managed_dir / 'runner.sock' - worker = subprocess.Popen([str(bench.args.runtime_bin), str(managed_socket)], stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=open(bench.evidence.file(f'{bench.prefix}/managed-worker.log'), 'ab'), - env=host.minimal_env(bench.work), start_new_session=True) + with open(bench.evidence.file(f'{bench.prefix}/managed-worker.log'), 'ab') as log: + worker = subprocess.Popen([str(bench.args.runtime_bin), str(managed_socket)], stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=log, env=host.minimal_env(bench.work), + start_new_session=True) bench.live.workers.append(worker) # A managed worker serves the first connection it accepts and no other, so it is not probed. deadline = time.monotonic() + 15 @@ -4574,14 +4576,18 @@ def drain(self): found = re.findall(r'stopped; (\d+) processes were terminated', self.runner.log()) drain['terminated'] = int(found[-1]) if found else None if self.governed: - drain['retired'], drain['instances'] = wait_retired(self.bench, self.runner.pid) - deadline = time.monotonic() + 60 - while True: - charged = host.journal_charged(self.args.journal)['attempts'].get(host.CONSUMER, 0) - if not charged or time.monotonic() >= deadline: - break - time.sleep(1.0) - drain['charged'] = charged + try: + drain['retired'], drain['instances'] = wait_retired(self.bench, self.runner.pid) + deadline = time.monotonic() + 60 + while True: + charged = host.journal_charged(self.args.journal)['attempts'].get(host.CONSUMER, 0) + if not charged or time.monotonic() >= deadline: + break + time.sleep(1.0) + drain['charged'] = charged + except (OSError, sqlite3.Error) as error: + drain['journal_error'] = f'{type(error).__name__}: {error}' + self.invalid.append(f"the journal could not be read: {drain['journal_error']}") return drain @@ -4952,8 +4958,10 @@ def verify_recovery(directory, problems): def harness(): - """The harness that ran: its files' SHA-256 and the CodeSpace commit.""" - found = {'files': {name: host.sha256(ROOT / 'scripts' / name) for name in HARNESS}} + """The harness that ran: its files' SHA-256, the CodeSpace commit, and the open-file limit it + and every process it started ran under (soft and hard; a macOS terminal's soft limit is 256).""" + found = {'files': {name: host.sha256(ROOT / 'scripts' / name) for name in HARNESS}, + 'open_files_limit': list(resource.getrlimit(resource.RLIMIT_NOFILE))} try: found.update(commit=git('rev-parse', 'HEAD'), dirty=dirty()) except (OSError, subprocess.CalledProcessError): diff --git a/scripts/qualify_host.py b/scripts/qualify_host.py index 11b5ba5..26b7b8b 100644 --- a/scripts/qualify_host.py +++ b/scripts/qualify_host.py @@ -8,6 +8,7 @@ run in a minimal environment (HOME, PATH, TMPDIR, LANG and CodeSpace's own settings) in their own sessions, so a terminal's interrupt reaches only the harness, which then settles what it started. """ +from collections import OrderedDict import hashlib import json import os @@ -40,12 +41,28 @@ NO_ANSWER = 'no_answer' # Codes that are the transport's, not CodeSpace's answer. TRANSPORT_CODES = (TRANSPORT, NO_ANSWER) +# The evidence writer keeps at most this many files open, closing the least recently written +# first, so that a run that writes to many files stays far below the open-file limit (256 by +# default in a macOS terminal). A recovery interval writes about 17 files at once, which all +# stay open. +WRITER_OPEN_FILES = 32 def default_socket(): return Path(f'/private/tmp/devguard-{os.getuid()}/authority.sock') +def close_quietly(handle): + """Close a pipe or file that may be closed already or whose other end is gone; the + descriptor is released either way.""" + if handle is None: + return + try: + handle.close() + except (OSError, ValueError): + pass + + def sha256(path): digest = hashlib.sha256() with open(path, 'rb') as handle: @@ -136,12 +153,13 @@ def verify_manifest(directory): class Writer(threading.Thread): """Appends JSON lines on its own thread, as DevGuard's measure.py does, so no sampler waits on - a disk write: a disk the load saturates delays the evidence, never a sample.""" + a disk write: a disk the load saturates delays the evidence, never a sample. It keeps at most + WRITER_OPEN_FILES files open and opens a closed one again to append to it.""" def __init__(self): super().__init__(daemon=True) self.lines = queue.Queue() - self.files = {} + self.files = OrderedDict() self.errors = [] self.written = 0 self.lag_max = 0.0 @@ -164,10 +182,7 @@ def run(self): batch = [item for item in batch if item is not None] for _, path, line in batch: try: - if path not in self.files: - path.parent.mkdir(parents=True, exist_ok=True) - self.files[path] = open(path, 'a') - self.files[path].write(line) + self._file(path).write(line) self.written += 1 except OSError as error: self.errors.append(f'{path.name}: {error}') @@ -179,6 +194,22 @@ def run(self): if batch: self.lag_max = max(self.lag_max, time.monotonic() - batch[0][0]) + def _file(self, path): + """The open file for `path`, the most recently written last; the least recently written + one is closed when too many are open.""" + if path in self.files: + self.files.move_to_end(path) + return self.files[path] + while len(self.files) >= WRITER_OPEN_FILES: + name, handle = self.files.popitem(last=False) + try: + handle.close() + except OSError as error: + self.errors.append(f'{name.name}: {error}') + path.parent.mkdir(parents=True, exist_ok=True) + self.files[path] = open(path, 'a') + return self.files[path] + def close(self, limit=120): """Write every line handed over and close the files; a disk that does not take them within `limit` seconds is reported, never waited for without end.""" @@ -288,6 +319,11 @@ def _read(self): self.pending.clear() for waiter in waiters: waiter['event'].set() + # The connection is over, so this side of both pipes is closed: a run of many gateways + # must not keep a descriptor of each. + close_quietly(self.process.stdout) + with self.write_lock: + close_quietly(self.process.stdin) def _answer_server(self, message): if 'id' not in message: @@ -461,14 +497,22 @@ def __init__(self, binary, flags, env, log_path, record=None, cwd=None): self.info = None def _log(self): - with open(self.log_path, 'ab') as log: - for line in self.process.stderr: - log.write(line) - log.flush() - entry = launch_line(line.decode('utf-8', 'replace')) - if entry: - with self.lock: - self.executions.setdefault(entry['process_id'], []).append(entry) + try: + with open(self.log_path, 'ab') as log: + for line in self.process.stderr: + log.write(line) + log.flush() + entry = launch_line(line.decode('utf-8', 'replace')) + if entry: + with self.lock: + self.executions.setdefault(entry['process_id'], []).append(entry) + finally: + close_quietly(self.process.stderr) + + def join_readers(self, limit=10): + """Wait for the readers of the gateway's output and log, which close its pipes as they end.""" + self.log_reader.join(timeout=limit) + self.client.reader.join(timeout=limit) def initialize(self, timeout=60): result, latency = self.client.request('initialize', { @@ -599,7 +643,7 @@ def close(self, limit=60): stopped = 'SIGKILL' self.process.kill() self.process.wait() - self.log_reader.join(timeout=10) + self.join_readers() return {'returncode': self.process.returncode, 'stopped': stopped, 'malformed_lines': self.client.malformed} @@ -631,7 +675,7 @@ def end_gateway(gateway, how, limit=60): stopped = 'SIGKILL' gateway.process.kill() gateway.process.wait() - gateway.log_reader.join(timeout=10) + gateway.join_readers() return {'how': how, 'returncode': gateway.process.returncode, 'stopped': stopped, 'exit_ms': round((time.monotonic() - began) * 1000, 3)} @@ -1200,16 +1244,22 @@ def __init__(self, binary, base, env, log_path, limit=2, timeout=60): stdout=subprocess.PIPE, stderr=self.log, env=env, text=True, start_new_session=True) self.lines = queue.Queue() - threading.Thread(target=self._read, daemon=True).start() + self.reader = threading.Thread(target=self._read, daemon=True) + self.reader.start() self.settings = self._answer(timeout) if self.settings is None: self.kill() raise RuntimeError('the fixture authority did not start') def _read(self): - for line in self.process.stdout: - self.lines.put(line) - self.lines.put(None) + try: + for line in self.process.stdout: + self.lines.put(line) + except (OSError, ValueError): + pass + finally: + close_quietly(self.process.stdout) + self.lines.put(None) def _answer(self, timeout): try: @@ -1237,15 +1287,16 @@ def kill(self): if self.process.poll() is None: self.process.kill() self.process.wait() + close_quietly(self.process.stdin) + close_quietly(self.log) + self.reader.join(timeout=10) def stop(self, limit=30): - try: - self.process.stdin.close() - except OSError: - pass + close_quietly(self.process.stdin) try: self.process.wait(timeout=limit) except subprocess.TimeoutExpired: self.kill() - self.log.close() + close_quietly(self.log) + self.reader.join(timeout=10) return self.process.returncode diff --git a/scripts/tests/test_qualify_devguard.py b/scripts/tests/test_qualify_devguard.py index b1c689e..32b8e15 100644 --- a/scripts/tests/test_qualify_devguard.py +++ b/scripts/tests/test_qualify_devguard.py @@ -8,6 +8,7 @@ import math import os from pathlib import Path +import resource import socket import sqlite3 import struct @@ -317,6 +318,21 @@ def test_writer(self): lines = (self.dir / 'w/rows.jsonl').read_text().splitlines() self.assertEqual([json.loads(line)['n'] for line in lines], list(range(100))) + def test_the_writer_keeps_few_files_open(self): + writer = host.Writer() + paths = [self.dir / f'w/{n:02d}.jsonl' for n in range(3 * host.WRITER_OPEN_FILES)] + for number in range(4): + for path in paths: + writer.write(path, {'round': number}) + deadline = time.monotonic() + 30 + while writer.written < 4 * len(paths) and time.monotonic() < deadline: + time.sleep(0.01) + self.assertLessEqual(len(writer.files), host.WRITER_OPEN_FILES) + report = writer.close() + self.assertEqual((report['lines'], report['error_count']), (4 * len(paths), 0)) + for path in paths: + self.assertEqual([json.loads(line)['round'] for line in path.read_text().splitlines()], [0, 1, 2, 3]) + def journal(self): database = self.dir / 'authority.sqlite' connection = sqlite3.connect(database) @@ -410,8 +426,6 @@ def setUp(self): def tearDown(self): self.gateway.close(limit=10) - self.gateway.process.stdout.close() - self.gateway.process.stderr.close() self.scratch.cleanup() def test_answers_errors_and_server_requests(self): @@ -438,6 +452,82 @@ def test_no_answer_and_a_lost_connection(self): self.assertTrue(raised.exception.transport) +FAKE_FIXTURE = textwrap.dedent(""" + import json, sys + print(json.dumps({'base': sys.argv[1]}), flush=True) + for line in sys.stdin: + print(json.dumps({'command': line.strip()}), flush=True) +""") + + +def open_descriptors(expected=None, limit=5.0): + """The test process's open descriptors; with `expected`, once they are that many or `limit` + seconds have passed, so that another test's thread that is still ending cannot make a leak + seem to come or go.""" + deadline = time.monotonic() + limit + while True: + count = len(os.listdir('/dev/fd')) + if expected is None or count == expected or time.monotonic() >= deadline: + return count + time.sleep(0.05) + + +class DescriptorTests(unittest.TestCase): + """Gateways and fixture authorities that ended leave none of their pipes open in the harness, + which starts hundreds of them in one run under a terminal's limit of 256 open files and keeps + each one referenced until the run settles.""" + + def setUp(self): + self.scratch = tempfile.TemporaryDirectory() + self.dir = Path(self.scratch.name) + self.server = self.dir / 'server.py' + self.server.write_text(FAKE_SERVER) + + def tearDown(self): + self.scratch.cleanup() + + def gateway(self, name): + gateway = host.Gateway(sys.executable, [str(self.server)], dict(os.environ), self.dir / f'{name}.log') + gateway.initialize(timeout=30) + return gateway + + def test_ended_gateways_close_their_pipes(self): + started = [self.gateway('first')] + started[0].close(limit=10) + before = open_descriptors() + for _ in range(3): + for how in host.GATEWAY_ENDS: + started.append(self.gateway(how)) + self.assertIsNone(host.end_gateway(started[-1], how, limit=10)['stopped']) + started.append(self.gateway('killed')) + started[-1].process.kill() + started[-1].process.wait() + started[-1].join_readers() + started.append(self.gateway('closed')) + started[-1].close(limit=10) + self.assertEqual(open_descriptors(before), before) + + def test_stopped_fixture_authorities_close_their_pipes(self): + binary = self.dir / 'fixture' + binary.write_text(f'#!{sys.executable}\n' + FAKE_FIXTURE) + binary.chmod(0o755) + + def start(name): + return host.FixtureAuthority(binary, self.dir / name, dict(os.environ), self.dir / f'{name}.log') + started = [start('first')] + started[0].stop() + before = open_descriptors() + for number in range(5): + started.append(start(f'fixture-{number}')) + self.assertEqual(started[-1].settings, {'base': str(self.dir / f'fixture-{number}')}) + self.assertEqual(started[-1].charged(), {'command': 'charged'}) + if number % 2: + started[-1].kill() + else: + self.assertEqual(started[-1].stop(), 0) + self.assertEqual(open_descriptors(before), before) + + class HarnessTests(unittest.TestCase): def setUp(self): self.scratch = tempfile.TemporaryDirectory() @@ -481,6 +571,9 @@ def test_consumer_settings_never_return_a_digest(self): self.assertNotIn('a' * 64, json.dumps(settings)) self.assertEqual(qualify.consumer_settings('[consumers.dev-cli]\nrole = "workload"\n'), {}) + def test_the_harness_records_its_open_file_limit(self): + self.assertEqual(qualify.harness()['open_files_limit'], list(resource.getrlimit(resource.RLIMIT_NOFILE))) + def test_documentation_keeps_the_candidate(self): repo = self.dir / 'repo' (repo / 'docs').mkdir(parents=True) @@ -1152,6 +1245,24 @@ def judged(change): self.assertEqual(analysis.recovery_verdict('incomplete', 'qualified'), 'inconclusive') self.assertEqual(analysis.recovery_verdict('passed', 'failed'), 'failed') + def test_an_unreadable_journal_at_the_drain_is_recorded(self): + interval = qualify.RecoveryInterval.__new__(qualify.RecoveryInterval) + interval.governed, interval.number, interval.invalid = True, 6, [] + interval.args = SimpleNamespace(journal=self.dir / 'missing/authority.sqlite') + + def no_gateway(*args, **kwargs): + raise OSError(24, 'Too many open files') + interval.bench = SimpleNamespace(gateway=no_gateway, socket=self.dir / 'runner.sock', + authority={'consumer': host.CONSUMER}, + database=self.dir / 'missing/authority.sqlite') + interval.runner = SimpleNamespace(pid=4242, wait=lambda limit: 0, + log=lambda: 'stopped; 0 processes were terminated\n') + drain = interval.drain() + self.assertEqual(drain['error'], 'OSError: [Errno 24] Too many open files') + self.assertTrue(drain['journal_error'].startswith('OperationalError: '), drain) + self.assertEqual(interval.invalid, [f"the journal could not be read: {drain['journal_error']}"]) + self.assertEqual((drain['runner_exit'], drain['socket_left'], drain['terminated']), (0, False, 0)) + def test_verify_recovery(self): path, _ = self.recovery_run() result = qualify.verify(path)