test: list the slowest tests on request and stop waiting 30 s on a held WAL (#487)
PM_TEST_SLOWEST=N makes tests/run_suite.py list the N slowest tests. Three history tests restored while their own read held a snapshot, so each publish sat out the 30 s WAL-truncate wait; they now use a short wait. Temporary diagnostics for the Windows timings ride along in this commit.
This commit is contained in:
@@ -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
|
||||
|
||||
+108
@@ -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)
|
||||
@@ -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)
|
||||
+44
-2
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
+53
-2
@@ -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})
|
||||
|
||||
Reference in New Issue
Block a user