diff --git a/src/fi/alk/harness/agent_scan.py b/src/fi/alk/harness/agent_scan.py index 821263a45..2a192cd67 100644 --- a/src/fi/alk/harness/agent_scan.py +++ b/src/fi/alk/harness/agent_scan.py @@ -141,6 +141,9 @@ def _check_job(job: HarnessJob) -> HarnessFailure | None: # The most the source look's list of folders still to visit may hold, by its count (TH-9525). _LOOK_HELD = 67_108_864 +# Inside one folder, the clock is read again after this many entries, so a huge folder can't keep +# the look past its deadline. +_LOOK_CLOCK_EVERY = 4_096 def _path_cost(path: str) -> int: @@ -158,7 +161,9 @@ def _look( `_LOOK_HELD` first) or ``"clock"`` (the deadline passed first), with the folders read and what the list held. Each folder is read once, entry by entry: its subfolders are taken on until the bound, and the read goes on past the bound for a regular file, which wins wherever it is in the - folder. A folder or entry that can't be read is passed over, as os.walk does.""" + folder. A folder or entry that can't be read is passed over, as os.walk does. The clock is read + before each folder after the source's own, and every `_LOOK_CLOCK_EVERY` entries inside any + folder; an empty source never reaches either, so it is "none" whatever the time.""" try: if not source.is_dir(): return "none", 0, 0 @@ -179,7 +184,9 @@ def _look( over = False try: with os.scandir(folder) as entries: - for entry in entries: + for index, entry in enumerate(entries, 1): + if not index % _LOOK_CLOCK_EVERY and clock() > deadline: + return "clock", folders, held try: if entry.is_dir(follow_symlinks=False): if not over: diff --git a/src/fi/alk/harness/scan_walk.py b/src/fi/alk/harness/scan_walk.py index 1c3741dad..db6891b9a 100644 --- a/src/fi/alk/harness/scan_walk.py +++ b/src/fi/alk/harness/scan_walk.py @@ -212,11 +212,14 @@ class _Unread(Exception): pass -# CPython 3.12's sizes on a 64-bit build, as asked of the allocator: a str (`_text`), a tuple +# CPython's sizes on a 64-bit build, as asked of the allocator, measured on 3.12, the Python the +# hosted image builds the kit's venv with (`_SIZES_MEASURED_ON`): a str (`_text`, which uses +# 3.11's 49-byte header, so on 3.12 and 3.13, at 41, it is 8 bytes high), a tuple # (`_TUPLE` and a slot per item), a set's own object, one dict or set entry with its share of # the table, a facts object, and the join's share for a file that could be an agent (its sign, # its component bookkeeping, its row) beside three path strings `_project_of` and # `_subfolder` make. Each is then charged at the block the allocator gives it (`_block`). +_SIZES_MEASURED_ON = (3, 12) _TUPLE = 40 _SLOT = 8 _SET = 216 @@ -224,7 +227,8 @@ class _Unread(Exception): _FACTS = 512 _JOIN_SHARE = 768 # What the join holds for every kept file however it's joined: its place in the sorted list of -# paths and in one search's set of files seen. +# paths and in one search's set of files seen. Held by +# `test_a_fan_out_join_is_charged_without_its_base`, which fails with this at 0. _JOIN_PER_FILE = 64 # What the join holds whatever was kept: its result, and up to `_MAX_AGENTS` rows with their # names and evidence. Charged once, when the walk starts. A build hazard's own look is @@ -915,11 +919,20 @@ def _source_files( Installed trees and the folders a build skips are never entered. ``side`` False gives only files outside test, example and other side folders; True gives only the files in them. ``charge`` counts each folder kept to enter, at its full path, which is what `os.walk` - holds until it enters it.""" + holds until it enters it, and each folder's own listing, its subfolder and file names, which + `os.walk` holds while the walk is in that folder. With ``side`` True, the second pass, only + folders the first pass never entered are counted: side folders, and folders below one.""" for directory, directories, files in os.walk(start, followlinks=False): directories.sort() files.sort() relative = directory[len(top) :].split(os.sep)[1:] + first_pass = side is not True or any(_is_side_part(part) for part in relative) + if charge is not None and first_pass: + # Summed list by list: one tuple of every name would itself hold a slot a name. + charge( + sum(_SLOT + _text(name) for name in directories) + + sum(_SLOT + _text(name) for name in files) + ) if relative and any(name.endswith(".dist-info") for name in directories): # An installed tree, not the customer's code. directories[:] = [] @@ -928,9 +941,10 @@ def _source_files( if side is False: directories[:] = [name for name in directories if not _is_side_part(name)] if charge is not None and directories: - charge( - sum(2 * _SLOT + _text(os.path.join(directory, name)) for name in directories) - ) + new = directories + if not first_pass: + new = [name for name in directories if _is_side_part(name)] + charge(sum(2 * _SLOT + _text(os.path.join(directory, name)) for name in new)) folder = "/".join(relative) for name in files: if side is None or _read_with_side_files(_join(folder, name)) is side: @@ -1473,13 +1487,18 @@ def walk_repo( logger.warning("agent walk: joining files into agents ran out of time") walk.facts.clear() return _empty_result(walk, "clock") - except (_ResultTooBig, _CapReached): + except _ResultTooBig: # Too many candidates reach each other to hold: the same end as the walk's other caps. - # The join's lookups all meet folders the walk already charged, so none charges again; - # if one ever did, past a held stop, it ends here too rather than in the crash path. logger.warning("agent walk: joining files into agents passed its edge budget") walk.facts.clear() return _empty_result(walk, "limit") + except _CapReached: + # The join's lookups all meet folders the walk already charged, so none charges again; + # if one ever did, past a held stop, it ends here rather than in the crash path. The + # rows found so far are dropped with the facts, unlike a stop during the walk. + logger.warning("agent walk: joining files into agents passed its memory budget") + walk.facts.clear() + return _empty_result(walk, "limit") logger.info( "agent walk: files_read=%d unread=%d reader_faults=%d agents=%d left_out=%d stopped=%s", walk.reader.files_read, @@ -1542,9 +1561,9 @@ def _hazard( # folders below it. if index and clock() > deadline: return "" - # Sorted, so that a look stopped by a cap stops at the same folder on every file system. + # Folders sorted, so that a look stopped by a cap stops at the same folder on every file + # system. Files need no order: only their count and `agent.py`'s are used. directories.sort() - files.sort() directories[:] = [name for name in directories if name not in _BUILD_IGNORED] found += files.count("agent.py") if found > 1: diff --git a/tests/harness/test_agent_scan.py b/tests/harness/test_agent_scan.py index 74ad60709..7e6193235 100644 --- a/tests/harness/test_agent_scan.py +++ b/tests/harness/test_agent_scan.py @@ -1536,6 +1536,54 @@ def test_the_clock_is_first_read_after_the_sources_own_folder(tmp_path) -> None: assert (word, folders) == ("clock", 1) +def test_the_clock_is_read_inside_a_huge_folder(tmp_path) -> None: + """One folder of 5,000 empty folders, past the deadline: the look stops partway through that + folder's listing, not after reading all of it.""" + source = tmp_path / "src" + folders = _empty_folders(source, [f"f{index:05}" for index in range(5_000)]) + whole = _cost(str(source)) + sum(_cost(folder) for folder in folders) + + word, read, held = agent_scan._look(source, deadline=0.0, clock=lambda: 1.0) + + assert (word, read) == ("clock", 1) + # Read at the 4,096th entry, after 4,095 folders were taken on: where and how often. + assert held == (agent_scan._LOOK_CLOCK_EVERY - 1) * _cost(folders[0]) < whole + + +def test_the_clock_is_read_inside_a_huge_folder_of_links(tmp_path, monkeypatch) -> None: + """Links are entries too: 5,000 dangling links past the deadline stop at the clock without + each being looked at.""" + source = tmp_path / "src" + source.mkdir() + for index in range(5_000): + os.symlink("nowhere", source / f"l{index:05}") + calls = [0] + real = os.lstat + + def counted(path, *args, **kwargs): + calls[0] += 1 + return real(path, *args, **kwargs) + + monkeypatch.setattr(agent_scan.os, "lstat", counted) + + word, read, _ = agent_scan._look(source, deadline=0.0, clock=lambda: 1.0) + + assert (word, read) == ("clock", 1) + assert calls[0] < agent_scan._LOOK_CLOCK_EVERY + + +def test_the_clock_is_read_inside_a_huge_folder_past_the_bound(tmp_path, monkeypatch) -> None: + """Past the bound the read goes on for a regular file; the clock still stops it.""" + source = tmp_path / "src" + folders = _empty_folders(source, [f"f{index:05}" for index in range(5_000)]) + monkeypatch.setattr(agent_scan, "_LOOK_HELD", 50 * _cost(folders[0])) + + word, read, held = agent_scan._look(source, deadline=0.0, clock=lambda: 1.0) + + assert (word, read) == ("clock", 1) + assert held > agent_scan._LOOK_HELD + + def test_the_look_says_why_it_stopped_in_one_line_of_counts(tmp_path, monkeypatch, caplog) -> None: source = tmp_path / "src" folders = _empty_folders(source, [f"folder_{index:03}" for index in range(200)]) diff --git a/tests/harness/test_framework_detection.py b/tests/harness/test_framework_detection.py index dfbbf5ea3..c834526c3 100644 --- a/tests/harness/test_framework_detection.py +++ b/tests/harness/test_framework_detection.py @@ -1756,7 +1756,8 @@ def _watch_the_stack(monkeypatch: pytest.MonkeyPatch) -> list[int]: def watching(*args: Any, **kwargs: Any) -> Any: walk = real_walk(*args, **kwargs) for item in walk: - stack = walk.gi_frame.f_locals.get("stack", []) if walk.gi_frame else [] + # A missing `stack` fails loudly rather than reading nothing. + stack = walk.gi_frame.f_locals["stack"] held.append(sum(_allocated(sys.getsizeof(path)) + 8 for path in stack if isinstance(path, str))) yield item @@ -1764,6 +1765,12 @@ def watching(*args: Any, **kwargs: Any) -> Any: return held +# os.walk keeps its folders still to visit in a `stack` local only from 3.12; before, it recurses. +_STACK_WALK = pytest.mark.skipif( + sys.version_info < (3, 12), reason="os.walk has a `stack` list to watch only from Python 3.12" +) + + _STACK_CASES = { # top length (None: tmp_path itself), depth, siblings, chain stem, sibling stem # Sized so the whole list passes 2 MiB even under a Linux tmp_path of about 50 characters, and @@ -1777,6 +1784,7 @@ def watching(*args: Any, **kwargs: Any) -> Any: } +@_STACK_WALK @pytest.mark.parametrize("case", list(_STACK_CASES)) def test_the_walks_own_list_never_passes_the_folder_bound( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, case: str @@ -1800,6 +1808,7 @@ def test_the_walks_own_list_never_passes_the_folder_bound( assert max(held) <= bound, (max(held), bound) +@_STACK_WALK def test_the_walks_own_list_grows_past_the_test_bound_without_it( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/harness/test_scan_walk.py b/tests/harness/test_scan_walk.py index 0ebd273a3..d7095fae6 100644 --- a/tests/harness/test_scan_walk.py +++ b/tests/harness/test_scan_walk.py @@ -2297,3 +2297,123 @@ def charging(self: object, **kwargs: object) -> object: monkeypatch.setattr(scan_walk._Walk, "result", charging) assert _walk(tmp_path).stopped == "limit" + + +def test_each_folder_is_charged_once_across_the_two_passes(tmp_path: Path) -> None: + """The side pass walks the customer's own folders again; only the folders the first pass + never entered (side folders, and folders below one) are charged there.""" + for index in range(40): + (tmp_path / f"app{index:02}" / "inner").mkdir(parents=True) + (tmp_path / "tests" / f"t{index:02}" / "deep").mkdir(parents=True) + (tmp_path / f"app{index:02}" / "examples" / "e").mkdir(parents=True) + (tmp_path / f"app{index:02}" / ".hidden" / "x").mkdir(parents=True) + (tmp_path / ".github" / "workflows").mkdir(parents=True) + charged: dict[str, int] = {"own": 0, "side": 0, "once": 0} + + def into(key: str): + def charge(size: int) -> None: + charged[key] += size + + return charge + + top = str(tmp_path) + for _ in scan_walk._source_files(top, top, side=False, charge=into("own")): + pass + for _ in scan_walk._source_files(top, top, side=True, charge=into("side")): + pass + for _ in scan_walk._source_files(top, top, charge=into("once")): + pass + + assert charged["own"] > 0 and charged["side"] > 0 + assert charged["own"] + charged["side"] == charged["once"], charged + + +def test_the_estimates_sizes_were_measured_on_the_hosted_kits_python() -> None: + """The sizes are CPython 3.12's. On 3.14 a tuple's header is larger, so the estimate would be + low there. This fails when the hosted image's kit venv moves to another + Python, so the sizes are measured again before the bound is trusted.""" + dockerfile = (Path(__file__).resolve().parents[2] / "Dockerfile.hosted").read_text() + major, minor = scan_walk._SIZES_MEASURED_ON + made = [ + line + for line in dockerfile.splitlines() + if "/opt/alk-venv" in line and ("-m venv" in line or "uv venv" in line or "virtualenv" in line) + ] + assert len(made) == 1, made + assert f"python{major}.{minor} -m venv /opt/alk-venv" in made[0], made + assert made[0].count("-m venv") == 1, made + + +def test_a_memory_budget_stop_while_joining_says_so( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A held-budget stop during the join ends at `limit`, with its own cause in the log, not the + edge budget's.""" + _files(tmp_path, {"bot.py": CLIENT}) + + def reached(self, *, upload: bool): + raise scan_walk._CapReached() + + monkeypatch.setattr(scan_walk._Walk, "result", reached) + kept: list[int] = [] + real_empty = scan_walk._empty_result + + def empty(walk, stopped): + kept.append(len(walk.facts)) + return real_empty(walk, stopped) + + monkeypatch.setattr(scan_walk, "_empty_result", empty) + with caplog.at_level("WARNING", logger=scan_walk.logger.name): + result = scan_walk.walk_repo(tmp_path, deadline=FOREVER, clock=lambda: 0.0) + + assert result.stopped == "limit" + assert "passed its memory budget" in caplog.text + assert "edge budget" not in caplog.text + assert kept == [0] # the facts are let go before the empty result is built + + +@pytest.mark.parametrize("shape", ["one_entry", "two_entries"]) +def test_a_fan_out_join_is_charged_without_its_base(tmp_path: Path, shape: str) -> None: + """110 hubs of leaves that each import one shared module, about 22,000 files: the join walks + every leaf. One entry puts the import table just past its growth at 21,846 entries. Two + entries are dearer still: the first fills that table, then the second's own search grows its + seen set while the table stays full (about 59.5 bytes a file). What the join really holds is + covered by the per-file charge alone, without `_JOIN_BASE`, and only with `_JOIN_PER_FILE` + at 64: at 56 the two-entry shape fails. Leaves that import nothing share + CPython's empty tuple, which would hide the per-file cost.""" + per_hub = 199 if shape == "one_entry" else 198 + hubs = { + f"h{h:03}.py": "".join(f"import l{h:03}_{n:03}\n" for n in range(per_hub)) for h in range(110) + } + leaves = {f"l{h:03}_{n:03}.py": "import common\n" for h in range(110) for n in range(per_hub)} + entries = {"a.py": "".join(f"import h{h:03}\n" for h in range(110)) + CLIENT} + if shape == "two_entries": + entries["b.py"] = "".join(f"import h{h:03}\n" for h in range(100)) + CLIENT + _files(tmp_path, {**entries, "common.py": "x = 1\n", **hubs, **leaves}) + + seen = _measure(tmp_path) + + assert seen["join_rise"] > 1_000_000, seen + assert seen["join_rise"] <= seen["join_charged"] - scan_walk._JOIN_BASE, seen + + +@pytest.mark.parametrize("entries", ["folders", "files", "links"]) +def test_a_wide_folder_is_charged_its_listing(tmp_path: Path, entries: str) -> None: + """One folder of many entries: what `os.walk` holds for the folder it is in, its names, is + charged once when the walk reaches it, before any name is filtered out (links to a folder are + listed as folders and then left out), so the charge is never below the walk's peak.""" + target = tmp_path / "target" + target.mkdir() + wide = tmp_path / "wide" + wide.mkdir() + for index in range(80_000 if entries == "files" else 30_000): + if entries == "folders": + (wide / f"f{index:05}").mkdir() + elif entries == "files": + (wide / f"f{index:05}.txt").touch() + else: + (wide / f"f{index:05}").symlink_to(target) + + seen = _measure(tmp_path) + + assert seen["walk_peak"] <= seen["charged"], seen