From 412d08e4875a19778e2e03f75d21ec422e160b27 Mon Sep 17 00:00:00 2001 From: John Travers Date: Fri, 13 Nov 2026 17:43:04 +0000 Subject: [PATCH 1/3] Read the scan memory limit from the process's own cgroup path The scan admission read only the root cgroup's memory.max (and the v1 equivalent), so under a cgroup manager without Slurm variables the workers were admitted against the whole machine. cgroup_memory in solprop._process now follows /proc/self/cgroup up to the mount, as the CPU quota reader does, taking the tightest memory.max or memory.limit_in_bytes. An unreadable or malformed limit file and a malformed SOLPROP_SCAN_CPU_OFFSET or SOLPROP_SCAN_CPU_COUNT now raise DriverError naming the file or variable. Closes #170 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XGENPwwkUUG9KSAwneHjgH --- changelog.d/scan-cgroup-memory.fixed.md | 1 + docs/guide/workflows/scans.md | 12 ++-- src/solprop/_process.py | 65 +++++++++++++++++++--- src/solprop/driver/executors/pool.py | 34 ++++++++---- tests/driver/test_scan_resources.py | 74 ++++++++++++++++++++++--- tests/driver/test_scans.py | 10 +--- tests/test_process.py | 53 ++++++++++++++++++ 7 files changed, 209 insertions(+), 40 deletions(-) create mode 100644 changelog.d/scan-cgroup-memory.fixed.md diff --git a/changelog.d/scan-cgroup-memory.fixed.md b/changelog.d/scan-cgroup-memory.fixed.md new file mode 100644 index 000000000..33d5216cc --- /dev/null +++ b/changelog.d/scan-cgroup-memory.fixed.md @@ -0,0 +1 @@ +`Pool` and `Batch` scans now size their workers from the tightest control-group memory limit on the process's own path (version 1 and 2), not the root group's, so a scan under a systemd slice or other cgroup manager without Slurm variables is no longer admitted against the whole machine. An unreadable or malformed limit file, and a malformed `SOLPROP_SCAN_CPU_OFFSET` or `SOLPROP_SCAN_CPU_COUNT`, now raise `DriverError` naming the file or variable instead of a bare `OSError` or `ValueError` (#170). diff --git a/docs/guide/workflows/scans.md b/docs/guide/workflows/scans.md index c666a1e75..1ff8fe786 100644 --- a/docs/guide/workflows/scans.md +++ b/docs/guide/workflows/scans.md @@ -250,6 +250,10 @@ A backend/RSS probe sizes workers. The estimate takes the larger of measured RSS and baseline RSS plus the backend ledger, then reserves retained output and transfer buffers. The controller's RSS and 20 % memory headroom are reserved separately. The file records this calculation. It refuses an allocation that cannot fit one worker. +The allocation is the smallest of the physical memory, the Slurm limits and the +tightest control-group memory limit from the process's own group up to the root; a +limit file that cannot be read, or a malformed `SOLPROP_SCAN_CPU_OFFSET` or +`SOLPROP_SCAN_CPU_COUNT`, raises {py:class}`~solprop.exceptions.DriverError`. This is admission control, not an operating-system memory limit; adaptive periodic statistics and custom reducer working memory can exceed the probe estimate. @@ -504,12 +508,8 @@ Recovery has known limits. A live but wedged owner on the same host holds its po indefinitely, because the lease age is not consulted while its process is alive: stop that process to release the point. A `Local` or `Pool` scan interrupted by a signal or a crash, rather than stopped by its `stop=` flag, skips its `running` points for -`stale_age` seconds on restart. Container memory limits are read -from the root cgroup rather than the process's own cgroup, so on a shared node under a -cgroup manager without Slurm variables the workers can be given too much memory; a failing -platform probe raises outside the {py:class}`~solprop.exceptions.SolpropError` -hierarchy. The scan file grows on every resume, because HDF5 does not reclaim the -groups it recreates. +`stale_age` seconds on restart. The scan file grows on every resume, because HDF5 does +not reclaim the groups it recreates. ``` Inspect a generated Slurm script before submission: diff --git a/src/solprop/_process.py b/src/solprop/_process.py index 2b2de528a..106e49fae 100644 --- a/src/solprop/_process.py +++ b/src/solprop/_process.py @@ -450,8 +450,22 @@ def _read_cgroup_cpus(root: Path) -> int | None: lines = (root / "proc/self/cgroup").read_text(encoding="utf-8").splitlines() except OSError: return None + quotas = [ + quota + for directory in _cgroup_path(root, lines, "cpu") + if (quota := _quota(directory)) is not None + ] + return min(quotas) if quotas else None + + +def _cgroup_path(root: Path, lines: Sequence[str], controller: str) -> Iterator[Path]: + """Yield a controller's group directories from the process's own up to the mount. + + - ``lines``: ``/proc/self/cgroup``, one ``hierarchy:controllers:path`` per line; + a line of another form is skipped. Version 2's unified line serves every + controller, version 1's only its own. + """ mounts = root / "sys/fs/cgroup" - quotas: list[int] = [] for line in lines: fields = line.split(":", 2) if len(fields) != 3: @@ -459,18 +473,55 @@ def _read_cgroup_cpus(root: Path) -> int | None: controllers, path = fields[1], Path(fields[2].lstrip("/")) if not controllers: base = mounts - elif "cpu" in controllers.split(","): + elif controller in controllers.split(","): base = mounts / controllers - base = base if base.is_dir() else mounts / "cpu" + base = base if base.is_dir() else mounts / controller else: continue # A container without its own cgroup namespace names the host's path, which # its /sys/fs/cgroup does not have; the parents end at the mount, its own group. for directory in (path, *path.parents): - quota = _quota(base / directory) - if quota is not None: - quotas.append(quota) - return min(quotas) if quotas else None + yield base / directory + + +def cgroup_memory(root: Path | None = None) -> int | None: + """Return the tightest memory limit on this process's control-group path. + + Every group from the process's own up to the hierarchy's root is read -- + ``memory.max`` (version 2) and ``memory.limit_in_bytes`` (version 1) -- because a + parent's limit binds its children, and on a shared node the root group is often + unlimited while the process's own is not. Unlike :func:`cgroup_cpus` it is read + afresh at each call, and a limit file that exists but cannot be read or parsed is + an error, not "no limit": admitting workers against the whole machine is how the + group's limit kills them. + + - ``root``: filesystem root holding ``proc/self/cgroup`` and ``sys/fs/cgroup``; + ``/`` when unset. + - Returns: bytes, positive, or ``None`` where no limit is set (``max``, or zero) + or the process has no control groups (not Linux). + - Raises ``OSError`` when a file that exists cannot be read and ``ValueError`` when + a limit is neither ``max`` nor a whole number of bytes; both name the file. + """ + root = Path("/") if root is None else root + record = root / "proc/self/cgroup" + if not record.exists(): + return None + limits: list[int] = [] + for directory in _cgroup_path( + root, record.read_text(encoding="utf-8").splitlines(), "memory" + ): + for name in ("memory.max", "memory.limit_in_bytes"): + file = directory / name + if not file.exists(): + continue + value = file.read_text(encoding="utf-8").strip() + if value == "max": + continue + if not value.isdecimal(): + raise ValueError(f"{file} holds {value!r}, not a memory limit in bytes") + if int(value) > 0: + limits.append(int(value)) + return min(limits) if limits else None def cores_in_masks(masks: Sequence[int], cpus: Sequence[int]) -> int: diff --git a/src/solprop/driver/executors/pool.py b/src/solprop/driver/executors/pool.py index 468030728..907ba7043 100644 --- a/src/solprop/driver/executors/pool.py +++ b/src/solprop/driver/executors/pool.py @@ -17,13 +17,13 @@ from itertools import chain from multiprocessing.connection import Connection, wait from multiprocessing.process import BaseProcess -from pathlib import Path from typing import Protocol, cast, runtime_checkable from solprop._process import ( LOAD_TIME_BLAS_VARIABLES, THREAD_VARIABLES, allowed_cpus, + cgroup_memory, sizes_blas_after_load, spawn_environment, thread_budget, @@ -366,15 +366,14 @@ def _memory_limit() -> int: limits.append(_windows_physical_memory()) else: limits.append(os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE")) - for path in ( - "/sys/fs/cgroup/memory.max", - "/sys/fs/cgroup/memory/memory.limit_in_bytes", - ): - file = Path(path) - if file.exists(): - value = file.read_text(encoding="utf-8").strip() - if value.isdecimal() and int(value) > 0: - limits.append(int(value)) + try: + group = cgroup_memory() + except (OSError, ValueError) as error: + raise DriverError( + f"cannot read the control-group memory limit: {error}" + ) from error + if group is not None: + limits.append(group) slurm = os.environ.get("SLURM_MEM_PER_NODE") if slurm and slurm.isdecimal(): limits.append(int(slurm) * 1024**2) @@ -394,8 +393,8 @@ def _cpus() -> tuple[int, ...]: allocated = os.environ.get("SLURM_CPUS_PER_TASK") if allocated and allocated.isdecimal(): available = available[: int(allocated)] - offset = int(os.environ.get("SOLPROP_SCAN_CPU_OFFSET", "0")) - count = int(os.environ.get("SOLPROP_SCAN_CPU_COUNT", str(len(available)))) + offset = _integer("SOLPROP_SCAN_CPU_OFFSET", 0) + count = _integer("SOLPROP_SCAN_CPU_COUNT", len(available)) if offset < 0 or count < 1: raise DriverError( "scan instance CPU offsets/counts must be non-negative/positive" @@ -405,6 +404,17 @@ def _cpus() -> tuple[int, ...]: return available[offset : offset + count][: budget.threads] +def _integer(variable: str, default: int) -> int: + """Read an integer environment variable, refusing a malformed one by name.""" + value = os.environ.get(variable) + if value is None: + return default + try: + return int(value) + except ValueError: + raise DriverError(f"{variable}={value!r} is not an integer") from None + + @dataclass(frozen=True) class MemoryBudget: """Record the admission calculation without treating it as a hard RSS limit. diff --git a/tests/driver/test_scan_resources.py b/tests/driver/test_scan_resources.py index 9957b9098..aeb43add5 100644 --- a/tests/driver/test_scan_resources.py +++ b/tests/driver/test_scan_resources.py @@ -34,13 +34,20 @@ def _fake_cpus_are_cores(monkeypatch): monkeypatch.setattr(_process, "top_tier_cores", lambda: None) -@pytest.mark.parametrize("value", ["max", "0", "4294967296"]) -def test_linux_memory_limits_ignore_unlimited_or_zero_cgroups( - tmp_path, monkeypatch, value -): - record = tmp_path / "memory.max" - record.write_text(value) - monkeypatch.setattr(pool, "Path", lambda name: record) +def _cgroup_tree(root, line, groups): + """Write ``/proc/self/cgroup`` and each group's ``memory.max`` under ``root``.""" + (root / "proc/self").mkdir(parents=True) + (root / "proc/self/cgroup").write_text(line + "\n") + for directory, value in groups.items(): + group = root / "sys/fs/cgroup" / directory + group.mkdir(parents=True, exist_ok=True) + (group / "memory.max").write_text(value) + return root + + +def _linux(monkeypatch, root): + """Simulate a Linux machine of 4 GiB whose control groups live under ``root``.""" + monkeypatch.setattr(pool, "cgroup_memory", lambda: _process.cgroup_memory(root)) monkeypatch.setattr(pool.sys, "platform", "linux") monkeypatch.setattr( pool.os, @@ -50,9 +57,62 @@ def test_linux_memory_limits_ignore_unlimited_or_zero_cgroups( ) monkeypatch.delenv("SLURM_MEM_PER_NODE", raising=False) monkeypatch.delenv("SLURM_MEM_PER_CPU", raising=False) + + +@pytest.mark.parametrize("value", ["max", "0", "4294967296", "8589934592"]) +def test_linux_memory_limits_ignore_unlimited_or_zero_cgroups( + tmp_path, monkeypatch, value +): + _linux(monkeypatch, _cgroup_tree(tmp_path, "0::/", {".": value})) assert pool._memory_limit() == 2**32 +def test_the_memory_limit_is_the_process_group_not_the_root(tmp_path, monkeypatch): + # A nested version 2 group limited to 1 GiB under a root that says max: a + # systemd slice on a shared node, with no Slurm variables. + root = _cgroup_tree( + tmp_path, + "0::/user.slice/job.scope", + {".": "max", "user.slice/job.scope": str(2**30)}, + ) + _linux(monkeypatch, root) + assert pool._memory_limit() == 2**30 + + +def test_a_slurm_memory_limit_below_the_group_still_binds(tmp_path, monkeypatch): + _linux(monkeypatch, _cgroup_tree(tmp_path, "0::/job", {"job": str(2**31)})) + monkeypatch.setenv("SLURM_MEM_PER_NODE", "1024") + assert pool._memory_limit() == 2**30 + + +@pytest.mark.parametrize("readable", [True, False]) +def test_an_unreadable_memory_limit_is_a_driver_error_naming_it( + tmp_path, monkeypatch, readable +): + root = _cgroup_tree(tmp_path, "0::/job", {"job": "garbled"}) + if not readable: + # A directory in the file's place fails to read on any platform and as any + # user, root included, unlike a permission bit. + record = root / "sys/fs/cgroup/job/memory.max" + record.unlink() + record.mkdir() + _linux(monkeypatch, root) + with pytest.raises(DriverError, match=r"job.memory\.max"): + pool._memory_limit() + + +@pytest.mark.parametrize( + "variable", ["SOLPROP_SCAN_CPU_OFFSET", "SOLPROP_SCAN_CPU_COUNT"] +) +def test_a_malformed_cpu_slice_is_a_driver_error_naming_it(monkeypatch, variable): + monkeypatch.setattr(_process, "cgroup_cpus", lambda: None) + monkeypatch.delenv("SOLPROP_SCAN_CPU_OFFSET", raising=False) + monkeypatch.delenv("SOLPROP_SCAN_CPU_COUNT", raising=False) + monkeypatch.setenv(variable, "x") + with pytest.raises(DriverError, match=variable): + pool._cpus() + + @pytest.mark.parametrize("offset,count", [(-1, 1), (0, 0)]) def test_cpu_instance_refuses_invalid_slice(monkeypatch, offset, count): monkeypatch.setattr(_process, "cgroup_cpus", lambda: None) diff --git a/tests/driver/test_scans.py b/tests/driver/test_scans.py index 1a257999f..84845f66f 100644 --- a/tests/driver/test_scans.py +++ b/tests/driver/test_scans.py @@ -627,20 +627,14 @@ def test_linux_allocation_uses_affinity_slurm_and_cgroup(tmp_path, monkeypatch): (root / "proc/self/cgroup").write_text("0::/\n") (root / "sys/fs/cgroup").mkdir(parents=True) (root / "sys/fs/cgroup/cpu.max").write_text("200000 100000") + (root / "sys/fs/cgroup/memory.max").write_text(str(1024**3)) monkeypatch.setattr( _process, "_own_cgroup_cpus", lambda: _process.cgroup_cpus(root) ) + monkeypatch.setattr(pool, "cgroup_memory", lambda: _process.cgroup_memory(root)) monkeypatch.delenv("SOLPROP_THREADS", raising=False) monkeypatch.setattr(_process, "physical_cores", lambda cpus: None) monkeypatch.setattr(_process, "top_tier_cores", lambda: None) - memory = tmp_path / "memory.max" - memory.write_text(str(1024**3)) - missing = tmp_path / "missing" - monkeypatch.setattr( - pool, - "Path", - lambda name: {"/sys/fs/cgroup/memory.max": memory}.get(name, missing), - ) monkeypatch.setattr(pool.sys, "platform", "linux") monkeypatch.setattr( pool.os, "sched_getaffinity", lambda pid: {2, 4, 6, 8}, raising=False diff --git a/tests/test_process.py b/tests/test_process.py index 466c7ce76..6ddfdd38f 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -194,6 +194,59 @@ def test_no_control_groups_means_no_quota(tmp_path): assert _process.cgroup_cpus(tmp_path) is None +GiB = 2**30 + + +@pytest.mark.parametrize( + ("line", "groups", "expected"), + [ + # Version 2: the process's own group binds although the root says max. + ( + "0::/user.slice/job.scope", + { + ".": {"memory.max": "max"}, + "user.slice/job.scope": {"memory.max": str(GiB)}, + }, + GiB, + ), + # A parent's limit binds its child. + ( + "0::/a/b", + {"a": {"memory.max": str(GiB)}, "a/b": {"memory.max": str(2 * GiB)}}, + GiB, + ), + ("0::/a", {"a": {"memory.max": "max"}}, None), + ("0::/a", {"a": {"memory.max": "0"}}, None), + # A container without its own namespace: the mount's group is its own. + ("0::/docker/abc", {".": {"memory.max": str(3 * GiB)}}, 3 * GiB), + # Version 1, under the memory controller's own mount, and not the CPU's. + ( + "4:cpu,cpuacct:/slice\n3:memory:/slice", + { + "cpu,cpuacct/slice": {"memory.limit_in_bytes": str(GiB)}, + "memory": {"memory.limit_in_bytes": "9223372036854771712"}, + "memory/slice": {"memory.limit_in_bytes": str(2 * GiB)}, + }, + 2 * GiB, + ), + ("garbled\n0::/a", {"a": {"memory.max": str(GiB)}}, GiB), + ("0::/a", {}, None), + ], +) +def test_the_memory_limit_is_the_tightest_on_the_path(tmp_path, line, groups, expected): + assert _process.cgroup_memory(_cgroup(tmp_path, line, groups)) == expected + + +def test_no_control_groups_means_no_memory_limit(tmp_path): + assert _process.cgroup_memory(tmp_path) is None + + +def test_a_malformed_memory_limit_names_its_file(tmp_path): + root = _cgroup(tmp_path, "0::/a", {"a": {"memory.max": "lots"}}) + with pytest.raises(ValueError, match=r"a.memory\.max"): + _process.cgroup_memory(root) + + @pytest.mark.parametrize( ("cpus", "quota", "variable", "expected"), [ From 5ce957718a26debb454374e4bb168dd0f0f741a2 Mon Sep 17 00:00:00 2001 From: John Travers Date: Fri, 13 Nov 2026 17:43:04 +0000 Subject: [PATCH 2/3] Treat OMP_PLACES alone as OpenMP binding when sizing the thread budget libgomp binds the main thread to its first place when OMP_PLACES is set and OMP_PROC_BIND is unset (measured with OMP_PLACES=cores: one core's two hardware threads). allowed_cpus checked only OMP_PROC_BIND, so that binding was believed as the grant and every pool sized to one core. OMP_PROC_BIND=false still turns binding off. Closes #172 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XGENPwwkUUG9KSAwneHjgH --- changelog.d/omp-places-binding.fixed.md | 1 + src/solprop/_process.py | 16 ++++++++------ tests/test_process.py | 28 +++++++++++++++++++------ 3 files changed, 33 insertions(+), 12 deletions(-) create mode 100644 changelog.d/omp-places-binding.fixed.md diff --git a/changelog.d/omp-places-binding.fixed.md b/changelog.d/omp-places-binding.fixed.md new file mode 100644 index 000000000..d32737467 --- /dev/null +++ b/changelog.d/omp-places-binding.fixed.md @@ -0,0 +1 @@ +Setting `OMP_PLACES` without `OMP_PROC_BIND` no longer caps every solprop thread pool and scan at the one core GNU OpenMP binds the main thread to: the thread budget now treats `OMP_PLACES` alone as binding on, as libgomp does, and still believes a narrowing when `OMP_PROC_BIND=false` (#172). diff --git a/src/solprop/_process.py b/src/solprop/_process.py index 106e49fae..8f69e1625 100644 --- a/src/solprop/_process.py +++ b/src/solprop/_process.py @@ -5,8 +5,8 @@ """The CPUs this process may run on, as the scheduler granted them. Linux reports affinity per thread. An OpenMP runtime started with ``OMP_PROC_BIND`` -set -- GNU libgomp, which Kokkos's builds load -- binds the **main** thread to the first -of its places when it starts, so after the first parallel region +or ``OMP_PLACES`` set -- GNU libgomp, which Kokkos's builds load -- binds the **main** +thread to the first of its places when it starts, so after the first parallel region ``os.sched_getaffinity(0)`` names one CPU although the process was granted many. A thread pool sized from that answer runs on one core (measured: an 18-core allocation gave a pool sized this way one thread). The affinity is therefore read once, when @@ -325,19 +325,23 @@ def _at_import() -> tuple[int, ...]: def allowed_cpus() -> tuple[int, ...]: """Return the CPUs this process may run on, undoing an OpenMP main-thread binding. - The current affinity, unless ``OMP_PROC_BIND`` is set and the current answer is a + The current affinity, unless OpenMP binding is on and the current answer is a single CPU, or the hardware threads of a single core, of the wider set the process had when ``solprop`` was imported: that is the runtime's binding of the main thread to its first place (``OMP_PLACES=threads`` or ``cores``) and not the scheduler's - grant, so the import-time set is returned. A deliberate narrowing to several cores, - or any narrowing without ``OMP_PROC_BIND``, is believed. + grant, so the import-time set is returned. Binding is on when ``OMP_PROC_BIND`` is + set to anything but ``false``, or when it is unset and ``OMP_PLACES`` is set: + libgomp then binds as if it were ``true`` (measured: ``OMP_PLACES=cores`` alone + bound the main thread to one core's two hardware threads). A deliberate narrowing + to several cores, or any narrowing with binding off, is believed. - Returns: sorted CPU indices, at least one. """ current = _query() binding = os.environ.get("OMP_PROC_BIND", "").strip().lower() + places = os.environ.get("OMP_PLACES", "").strip() if ( - binding not in {"", "false"} + (binding not in {"", "false"} or (not binding and places)) and len(_AT_IMPORT) > len(current) and set(current) <= set(_AT_IMPORT) and (len(current) == 1 or physical_cores(current) == 1) diff --git a/tests/test_process.py b/tests/test_process.py index 6ddfdd38f..a1ee581e9 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -29,8 +29,10 @@ def _one_kind_of_core(monkeypatch): @pytest.fixture def granted(monkeypatch): - """A process granted CPUs 0-17 at import.""" + """A process granted CPUs 0-17 at import, with no OpenMP placement chosen.""" monkeypatch.setattr(_process, "_AT_IMPORT", tuple(range(18))) + monkeypatch.delenv("OMP_PROC_BIND", raising=False) + monkeypatch.delenv("OMP_PLACES", raising=False) def _affinity(monkeypatch, cpus): @@ -49,14 +51,28 @@ def test_a_main_thread_bound_by_openmp_still_sees_the_grant( assert _process.allowed_cpus() == tuple(range(18)) -@pytest.mark.parametrize("binding", [None, "false"]) +@pytest.mark.parametrize("places", ["cores", "threads", "{0},{1}"]) +def test_a_main_thread_bound_by_places_alone_still_sees_the_grant( + monkeypatch, granted, places +): + # libgomp with OMP_PLACES and no OMP_PROC_BIND binds as if it were true: one + # core's two hardware threads for OMP_PLACES=cores. + monkeypatch.setenv("OMP_PLACES", places) + monkeypatch.setattr(_process, "physical_cores", lambda cpus: 1) + _affinity(monkeypatch, {0, 9}) + assert _process.allowed_cpus() == tuple(range(18)) + + +@pytest.mark.parametrize( + ("binding", "places"), [(None, None), ("false", None), ("false", "cores")] +) def test_without_a_binding_the_current_affinity_is_believed( - monkeypatch, granted, binding + monkeypatch, granted, binding, places ): - if binding is None: - monkeypatch.delenv("OMP_PROC_BIND", raising=False) - else: + if binding is not None: monkeypatch.setenv("OMP_PROC_BIND", binding) + if places is not None: + monkeypatch.setenv("OMP_PLACES", places) _affinity(monkeypatch, {3}) assert _process.allowed_cpus() == (3,) From 263d05feedc42ec5ec5613de74e675595469a747 Mon Sep 17 00:00:00 2001 From: John Travers Date: Fri, 13 Nov 2026 17:43:04 +0000 Subject: [PATCH 3/3] Keep current scan layouts on a rerun that commits nothing Every close deleted and rebuilt the consolidated layouts, and HDF5 does not reclaim a deleted group's space, so an unchanged rerun grew the file. A commit now marks the layouts stale durably before it writes, and close rebuilds only stale layouts, so a run killed before consolidating is still rebuilt by the next close. The guide states what a rerun that commits points costs. Closes #169 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XGENPwwkUUG9KSAwneHjgH --- changelog.d/scan-rerun-growth.fixed.md | 1 + docs/guide/workflows/scans.md | 8 +++-- src/solprop/output/scan.py | 24 +++++++++++++- tests/output/test_scan.py | 43 ++++++++++++++++++++++++++ 4 files changed, 73 insertions(+), 3 deletions(-) create mode 100644 changelog.d/scan-rerun-growth.fixed.md diff --git a/changelog.d/scan-rerun-growth.fixed.md b/changelog.d/scan-rerun-growth.fixed.md new file mode 100644 index 000000000..aa26bc3e6 --- /dev/null +++ b/changelog.d/scan-rerun-growth.fixed.md @@ -0,0 +1 @@ +Rerunning a `Local` or `Pool` scan with every point already done no longer grows the scan file by rebuilding its consolidated `layouts`: they are rebuilt only when a point has been committed since they were last built, including by an earlier run that stopped before consolidating. Such a rerun now adds only its `executions` row (#169). diff --git a/docs/guide/workflows/scans.md b/docs/guide/workflows/scans.md index 1ff8fe786..a4c10f045 100644 --- a/docs/guide/workflows/scans.md +++ b/docs/guide/workflows/scans.md @@ -508,8 +508,12 @@ Recovery has known limits. A live but wedged owner on the same host holds its po indefinitely, because the lease age is not consulted while its process is alive: stop that process to release the point. A `Local` or `Pool` scan interrupted by a signal or a crash, rather than stopped by its `stop=` flag, skips its `running` points for -`stale_age` seconds on restart. The scan file grows on every resume, because HDF5 does -not reclaim the groups it recreates. +`stale_age` seconds on restart. A rerun that commits no point leaves the consolidated +`layouts` as they are, and adds only its row to `executions`, a few kilobytes. A rerun +that commits points rebuilds `layouts`, a copy of every completed point's arrays, and +HDF5 does not reclaim the space of the copy it replaces, so the file grows by about the +size of all the completed points' arrays as well as the new points' own: with 32 points +of 256 kB of fields each, a rerun that committed one point grew the file by 8.7 MB. ``` Inspect a generated Slurm script before submission: diff --git a/src/solprop/output/scan.py b/src/solprop/output/scan.py index ead5e7beb..b920e5098 100644 --- a/src/solprop/output/scan.py +++ b/src/solprop/output/scan.py @@ -376,6 +376,7 @@ def commit(self, result: PointResult) -> None: return if result.status not in ("done", "failed"): raise OutputError(f"invalid point status {result.status!r}") + self._layouts_stale() if not 0 <= result.index < len(_dataset(self.file, "points/index")): raise OutputError("result index is outside the scan") # A committed point is the only copy of its arrays on the Local, Pool and @@ -547,10 +548,24 @@ def record_execution(self, executor: str, policy: Mapping[str, object]) -> None: ) self.file.flush() + def _layouts_stale(self) -> None: + """Mark the layouts out of date, durably, before a commit changes staging. + + A rebuild that follows clears the mark; one that never comes (a crash, a + kill) leaves it for the next writer's close to see. + """ + layouts = self.file.require_group("layouts") + if layouts.attrs.get("current", False): + layouts.attrs["current"] = False + self.file.flush() + def close(self, *, consolidate: bool = True) -> None: """Build layout arrays and CSR statistics, then close the writer. A hard interruption before this point is readable from committed staging. + Layouts already current -- no point committed since they were built, as on a + rerun with every point done -- are kept rather than rebuilt: HDF5 does not + reclaim the space of a deleted group, so each rebuild grows the file. Parameters ---------- @@ -558,7 +573,11 @@ def close(self, *, consolidate: bool = True) -> None: False closes an unchanged collection without rebuilding its layouts. """ try: - if consolidate: + layouts = self.file.get("layouts") + current = isinstance(layouts, h5py.Group) and bool( + layouts.attrs.get("current", False) + ) + if consolidate and not current: self._collect_layouts() finally: # Closes the file, then releases the sidecar lock a waiting reader @@ -612,6 +631,9 @@ def _collect_layouts(self) -> None: target[row] = _dataset(file, f"staged/{index}/{key}")[...] self._common_fields(done) self._rectangular(done, next(iter(by_layout)) if len(by_layout) == 1 else None) + # Only after the whole replacement is flushed, which _rectangular ends with. + layouts.attrs["current"] = True + file.flush() def _rectangular(self, done: np.ndarray, layout_name: str | None) -> None: """Expose uniform arrays over the complete global point axis. diff --git a/tests/output/test_scan.py b/tests/output/test_scan.py index 3b7cf0ca3..eb1bc8633 100644 --- a/tests/output/test_scan.py +++ b/tests/output/test_scan.py @@ -415,3 +415,46 @@ def test_partial_staging_never_exposes_unvalidated_metadata(tmp_path): writer.close() assert ScanOutput(path).point(0)["status"] == "todo" assert "metadata" not in ScanOutput(path).point(0) + + +def _materialised(path): + """Read the global indices each consolidated layout holds.""" + with h5py.File(path) as file: + return [ + _dataset(layout, "index")[...].tolist() + for layout in _group(file, "layouts").values() + ] + + +def test_a_rerun_that_commits_nothing_keeps_the_layouts_and_the_file_size(tmp_path): + # Every close used to delete and rebuild the layouts, and HDF5 does not reclaim + # a deleted group's space: an unchanged rerun grew the file each time. + path = tmp_path / "scan.h5" + scan = Scan("scan", power=[1, 2, 3]) + writer = ScanWriter(path, scan, {}) + for index in range(3): + writer.commit(result(index)) + writer.close() + size = path.stat().st_size + for _ in range(3): + writer = ScanWriter(path, scan, {}) + writer.commit(dataclasses.replace(result(0), status="skipped")) + writer.close() + assert path.stat().st_size == size + assert _materialised(path) == [[0, 1, 2]] + + +def test_a_commit_after_an_interrupted_close_still_rebuilds_the_layouts(tmp_path): + path = tmp_path / "scan.h5" + scan = Scan("scan", power=[1, 2]) + writer = ScanWriter(path, scan, {}) + writer.commit(result(0)) + writer.close() + writer = ScanWriter(path, scan, {}) + writer.commit(result(1)) + # Killed before its close consolidated: the next writer, with nothing of its + # own to commit, rebuilds what the last one left. + writer.close(consolidate=False) + assert _materialised(path) == [[0]] + ScanWriter(path, scan, {}).close() + assert _materialised(path) == [[0, 1]]