diff --git a/.github/workflows/portability.yml b/.github/workflows/portability.yml index 75122ab0..0f238496 100644 --- a/.github/workflows/portability.yml +++ b/.github/workflows/portability.yml @@ -51,5 +51,22 @@ jobs: python -m pip install --upgrade pip pip install -r requirements-dev.txt + - name: DIAG 487 microbench (temporary) + if: matrix.python-version == '3.12' + run: python diag_487.py + + - name: DIAG 487 profile (temporary) + if: matrix.python-version == '3.12' + run: python diag_487_profile.py tests.test_database tests.test_review_schema tests.test_api_admin tests.test_lora_utils + - name: Run tests + env: + PM_TEST_SLOWEST: "40" run: python -m tests.run_suite -v + + - name: DIAG 487 microbench without Defender realtime (temporary) + if: always() && matrix.python-version == '3.12' && runner.os == 'Windows' + shell: pwsh + run: | + Set-MpPreference -DisableRealtimeMonitoring $true + python diag_487.py diff --git a/diag_487.py b/diag_487.py new file mode 100644 index 00000000..12c58b33 --- /dev/null +++ b/diag_487.py @@ -0,0 +1,108 @@ +"""Temporary #487 diagnostics (removed before merge).""" + +import logging +import os +import shutil +import sqlite3 +import statistics +import sys +import tempfile +import time + +logging.disable(logging.CRITICAL) +sys.path.insert(0, os.getcwd()) +from database import models # noqa: E402 +from database.models import PromptModel # noqa: E402 +from database.operations import PromptDatabase # noqa: E402 + + +def bench(label, fn, n): + times = [] + for i in range(n): + t = time.perf_counter() + fn(i) + times.append((time.perf_counter() - t) * 1000) + print( + f" {label:44s} median {statistics.median(times):8.2f} ms" + f" mean {statistics.mean(times):8.2f} ms" + ) + + +def run(base): + d = tempfile.mkdtemp(dir=base) + print(f"== base {base}") + + def small(i): + p = os.path.join(d, f"f{i}.txt") + with open(p, "wb") as h: + h.write(b"x" * 4096) + os.unlink(p) + + def fsync(i): + p = os.path.join(d, f"s{i}.txt") + with open(p, "wb") as h: + h.write(b"x" * 4096) + h.flush() + os.fsync(h.fileno()) + os.unlink(p) + + def raw(sync): + def f(i): + p = os.path.join(d, f"r{sync}{i}.db") + c = sqlite3.connect(p) + c.execute("PRAGMA journal_mode=WAL") + c.execute(f"PRAGMA synchronous={sync}") + c.execute("CREATE TABLE t(x)") + c.execute("INSERT INTO t VALUES(1)") + c.commit() + c.close() + for s in ("", "-wal", "-shm"): + if os.path.exists(p + s): + os.unlink(p + s) + + return f + + def pdb(i): + p = os.path.join(d, f"p{i}.db") + db = PromptDatabase(p) + db.save_prompt( + text=f"t{i}", + category=None, + tags=[], + rating=None, + notes=None, + prompt_hash=str(i), + ) + PromptModel.close_all_instances() + for s in ("", "-wal", "-shm"): + if os.path.exists(p + s): + os.unlink(p + s) + + def mkrm(i): + t = tempfile.mkdtemp(dir=d) + open(os.path.join(t, "a"), "w").close() + shutil.rmtree(t) + + bench("create+write+close+unlink 4k file", small, 100) + bench("same + fsync", fsync, 50) + bench("mkdtemp + file + rmtree", mkrm, 100) + bench("raw sqlite WAL db NORMAL", raw("NORMAL"), 30) + bench("raw sqlite WAL db OFF", raw("OFF"), 30) + bench("PromptDatabase create+save+close+unlink", pdb, 20) + orig = models.configure_connection + + def off(conn): + orig(conn) + conn.execute("PRAGMA synchronous = OFF") + + models.configure_connection = off + bench(" same with synchronous=OFF", lambda i: pdb(1000 + i), 20) + models.configure_connection = orig + shutil.rmtree(d, ignore_errors=True) + + +bases = [tempfile.gettempdir()] +if os.environ.get("RUNNER_TEMP"): + bases.append(os.environ["RUNNER_TEMP"]) +for b in bases: + run(b) diff --git a/diag_487_profile.py b/diag_487_profile.py new file mode 100644 index 00000000..e0fd1da3 --- /dev/null +++ b/diag_487_profile.py @@ -0,0 +1,18 @@ +"""Temporary #487 profile of a few modules (removed before merge).""" + +import cProfile +import logging +import pstats +import sys +import unittest + +logging.disable(logging.CRITICAL) +names = sys.argv[1:] +suite = unittest.defaultTestLoader.loadTestsFromNames(names) +pr = cProfile.Profile() +pr.enable() +unittest.TextTestRunner(stream=open(__import__("os").devnull, "w")).run(suite) +pr.disable() +st = pstats.Stats(pr) +st.sort_stats("tottime").print_stats(30) +st.sort_stats("cumtime").print_stats(60) diff --git a/tests/run_suite.py b/tests/run_suite.py index 9775d274..fc75199f 100644 --- a/tests/run_suite.py +++ b/tests/run_suite.py @@ -28,6 +28,10 @@ stack shows which). A dump after the summary is a hang at interpreter exit A limit is a measurement, not a dial: change it only with new measurements written next to it (TESTING.md "Time limits and hangs"). + +PM_TEST_SLOWEST=N (opt-in, #487) lists the N slowest tests after the summary. +A test's time runs from its ``start:`` line to the next one, so it includes +the class and module fixtures in between, like the hang limit. """ import argparse @@ -36,6 +40,7 @@ import os import shutil import sys import tempfile +import time import unittest from pathlib import Path @@ -63,23 +68,57 @@ def read_limit(name, default): return int(text) -def make_result_class(stream, per_test_limit, shutdown_limit): +class TestClock: + """Seconds from each test's start to the next start (or the run's end).""" + + def __init__(self, clock=time.perf_counter): + self._clock = clock + self._current = None + self._started = 0.0 + self.durations = [] + + def start(self, test_id): + self.stop() + self._current = test_id + self._started = self._clock() + + def stop(self): + if self._current is not None: + self.durations.append((self._clock() - self._started, self._current)) + self._current = None + + def report(self, count): + """Return the lines that list the ``count`` slowest tests.""" + slowest = sorted(self.durations, key=lambda item: item[0], reverse=True) + lines = [f"slowest {min(count, len(slowest))} of {len(slowest)} tests:"] + lines += [f"{seconds:8.2f}s {test_id}" for seconds, test_id in slowest[:count]] + return lines + + +def make_result_class(stream, per_test_limit, shutdown_limit, slowest=0): """Build a TextTestResult that names each test and re-arms the watchdog.""" def arm(seconds): faulthandler.dump_traceback_later(seconds, exit=True, file=stream) + clock = TestClock() + class WatchdogResult(unittest.TextTestResult): def startTest(self, test): # At default verbosity the progress dots share a line; start ours. prefix = "\n" if self.dots else "" stream.write(f"{prefix}start: {test.id()}\n") stream.flush() + clock.start(test.id()) super().startTest(test) arm(per_test_limit) def stopTestRun(self): + clock.stop() super().stopTestRun() + if slowest: + stream.write("\n".join(["", *clock.report(slowest), ""])) + stream.flush() arm(shutdown_limit) return WatchdogResult @@ -138,6 +177,7 @@ def main(argv=None): try: per_test_limit = read_limit("PM_TEST_HANG_LIMIT", DEFAULT_PER_TEST_LIMIT) shutdown_limit = read_limit("PM_TEST_SHUTDOWN_LIMIT", DEFAULT_SHUTDOWN_LIMIT) + slowest = read_limit("PM_TEST_SLOWEST", 0) except LimitError as error: stream.write(f"{error}\n") return USAGE_EXIT @@ -152,7 +192,9 @@ def main(argv=None): verbosity=args.verbosity, # As unittest.main does: show warnings unless -W/PYTHONWARNINGS set them. warnings=None if sys.warnoptions else "default", - resultclass=make_result_class(stream, per_test_limit, shutdown_limit), + resultclass=make_result_class( + stream, per_test_limit, shutdown_limit, slowest + ), ) result = runner.run(suite) finally: diff --git a/tests/test_api_history.py b/tests/test_api_history.py index f42e4b60..9e1d8ceb 100644 --- a/tests/test_api_history.py +++ b/tests/test_api_history.py @@ -25,6 +25,7 @@ try: except ImportError: from tests.test_api_images import ImageAPITestCase +from database import restore_stage from database.operations import PromptDatabase from py.api.library import _encode, _scope from tests.test_execution_context_database import bundle, snapshot @@ -43,6 +44,21 @@ OTHER_SESSION = "aaaaaaaa-3333-4333-8333-333333333333" MISSING_RUN = "bbbbbbbb-4444-4444-8444-444444444444" +@contextmanager +def no_wal_truncate_wait(): + """Restore with the restore notification held back, as these tests need. + + The read under test keeps its snapshot open across the restore, so the + publish's WAL truncate can never succeed; without the short wait it would + sit out the full CHECKPOINT_BUSY_MS (30 s) each time (#487). + """ + with ( + patch("database.execution_history.notify_history_restore"), + patch.object(restore_stage, "CHECKPOINT_BUSY_MS", 50), + ): + yield + + class HistoryAPITests(ImageAPITestCase): async def tearDownAsync(self): if hasattr(self.api, "close_history_reads"): @@ -346,7 +362,7 @@ class HistoryAPITests(ImageAPITestCase): def during_snapshot(*args, **kwargs): result = original(*args, **kwargs) - with patch("database.execution_history.notify_history_restore"): + with no_wal_truncate_wait(): writer.model.restore_from_file(source) return result @@ -890,7 +906,7 @@ class HistoryAPITests(ImageAPITestCase): def during_snapshot(*args, **kwargs): result = original(*args, **kwargs) - with patch("database.execution_history.notify_history_restore"): + with no_wal_truncate_wait(): writer.model.restore_from_file(source) return result diff --git a/tests/test_history_read_fence.py b/tests/test_history_read_fence.py index 9e60dd5c..9b6f354f 100644 --- a/tests/test_history_read_fence.py +++ b/tests/test_history_read_fence.py @@ -7,6 +7,7 @@ from contextlib import closing from pathlib import Path from unittest.mock import patch +from database import restore_stage from database.execution_history import HistoryEpochChanged, notify_history_restore from tests import test_execution_history_database as history_tests @@ -41,7 +42,12 @@ class HistoryReadFenceTests(unittest.TestCase): generation = fence.begin() with reader.recorded_history_snapshot() as epoch: original = reader.get_history_run(start.run_id) - with patch("database.execution_history.notify_history_restore"): + # The open snapshot keeps the WAL from being truncated; do not + # sit out the full 30 s checkpoint wait for it (#487). + with ( + patch("database.execution_history.notify_history_restore"), + patch.object(restore_stage, "CHECKPOINT_BUSY_MS", 50), + ): writer.model.restore_from_file(source) self.assertEqual(reader.get_execution_epoch(), epoch) self.assertEqual(original["run"]["storage_epoch"], epoch) diff --git a/tests/test_run_suite.py b/tests/test_run_suite.py index b35fe567..0ea3f859 100644 --- a/tests/test_run_suite.py +++ b/tests/test_run_suite.py @@ -18,11 +18,13 @@ from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent SUBPROCESS_TIMEOUT = 60 WATCHDOG_ENV = ("PM_TEST_HANG_LIMIT", "PM_TEST_SHUTDOWN_LIMIT") +# Integer settings the runner reads; a bad value is a usage error. +RUNNER_ENV = (*WATCHDOG_ENV, "PM_TEST_SLOWEST") def run_suite(folder, extra_env=None, *args, verbosity=("-v",)): """Run the runner on ``folder`` and return the completed process.""" - env = {k: v for k, v in os.environ.items() if k not in WATCHDOG_ENV} + env = {k: v for k, v in os.environ.items() if k not in RUNNER_ENV} # Windows writes a piped stderr in the ANSI codepage; ask for the UTF-8 # this helper decodes, so non-ASCII values and paths survive. env["PYTHONIOENCODING"] = "utf-8" @@ -269,8 +271,57 @@ class RunSuiteTests(unittest.TestCase): self.assertIn("ok", result.stderr, detail) self.assertIn("Timeout (", result.stderr, detail) + def test_slowest_tests_are_listed_when_asked(self): + self.write( + "test_probe_slowest.py", + """ + import time + import unittest + + class Probe(unittest.TestCase): + def test_fast(self): + pass + + def test_slow(self): + time.sleep(0.3) + + def test_slower(self): + time.sleep(0.6) + """, + ) + result = run_suite(self.folder, {"PM_TEST_SLOWEST": "2"}) + self.assertEqual(result.returncode, 0, result.stderr) + report = result.stderr.split("slowest 2 of 3 tests:\n", 1) + self.assertEqual(len(report), 2, result.stderr) + lines = [line.split() for line in report[1].strip().splitlines()[:2]] + self.assertEqual( + [line[1] for line in lines], + [ + "test_probe_slowest.Probe.test_slower", + "test_probe_slowest.Probe.test_slow", + ], + result.stderr, + ) + self.assertGreaterEqual(float(lines[0][0].rstrip("s")), 0.6) + self.assertNotIn("test_fast", report[1]) + + def test_slowest_tests_are_not_listed_by_default(self): + self.write( + "test_probe_quiet.py", + """ + import unittest + + class Probe(unittest.TestCase): + def test_ok(self): + pass + """, + ) + result = run_suite(self.folder) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertNotIn("slowest", result.stderr) + def test_invalid_limit_is_a_usage_error(self): - for name in WATCHDOG_ENV: + for name in RUNNER_ENV: for value in ("abc", "0", "-3", "1.5", "\u00b2"): with self.subTest(name=name, value=value): result = run_suite(self.folder, {name: value})