Fix scan platform probes: own-cgroup memory limit, OMP_PLACES binding, scan file growth on rerun

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).

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).

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).

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,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. 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. 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:

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)
@@ -450,27 +454,78 @@ 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:
             continue
         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:

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.

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,18 +548,36 @@ 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
         ----------
         consolidate : bool, optional
             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.

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)

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