benchmarks: count a stacked block device once in the container io figures (#555)

benchmarks/experiments/memory_probe.py

@@ -137,17 +137,93 @@ def parse_heap_info(text):
     return {"used": used, "committed": committed} if found else {}
 
 
-def parse_io_stat(text):
-    """cgroup v2 io.stat -> {"rbytes": n, "wbytes": n}, summed over devices. Cumulative since the cgroup was created."""
-    out = {"rbytes": 0, "wbytes": 0}
+SYSFS = "/sys"
+
+
+def _sysfs_read(path):
+    try:
+        with open(path) as fh:
+            return fh.read().strip()
+    except OSError:
+        return None
+
+
+def _whole_disk(sysfs, name):
+    """(maj:min, sysfs dir) of block device `name`, resolved to its whole disk when it is a partition.
+
+    cgroup v2 io.stat lists whole disks and device-mapper/md devices, never partitions: IO to a partition is charged to its disk.
+    """
+    d = os.path.realpath(os.path.join(sysfs, "class", "block", name))
+    if os.path.exists(os.path.join(d, "partition")):
+        d = os.path.dirname(d)
+    mm = _sysfs_read(os.path.join(d, "dev"))
+    return (mm, d) if mm else (None, d)
+
+
+def _stacked_below(sysfs, mm):
+    """Every maj:min that device `mm` is built on (its slaves, their slaves, ...), partitions resolved to whole disks.
+
+    None when sysfs has no entry for `mm` at all (the relation is unknown, not empty).
+    """
+    start = os.path.join(sysfs, "dev", "block", mm)
+    if not os.path.isdir(start):
+        return None
+    below, seen, todo = set(), set(), [start]
+    while todo:
+        d = todo.pop()
+        try:
+            names = os.listdir(os.path.join(d, "slaves"))
+        except OSError:
+            continue
+        for name in names:
+            smm, sdir = _whole_disk(sysfs, name)
+            if smm is None or smm in seen:
+                continue
+            seen.add(smm)
+            below.add(smm)
+            todo.append(sdir)
+    return below
+
+
+def parse_io_stat(text, sysfs=SYSFS):
+    """cgroup v2 io.stat -> {"rbytes": n, "wbytes": n}, cumulative since the cgroup was created.
+
+    io.stat has one line per block device the cgroup touched, and a stacked device is charged twice: a write to a device-mapper
+    device (LVM, dm-crypt) is counted on it AND on the disk under it, with identical bytes (the mini: `259:0` NVMe and `252:0` dm,
+    both wbytes=8691597312). Summing every line, as this function once did, doubles the IO of any container on such a stack.
+
+    Rule: a listed device is skipped when it is built on (is a slave of, directly or through further layers, partitions resolved to
+    their whole disk) another device that is also listed, per `<sysfs>/dev/block/<maj:min>/slaves`. The TOP of a stack is kept, so
+    the total is what the container wrote once. Unrelated devices (two independent disks) are still summed, so no real IO is lost.
+
+    When sysfs has no entry for a listed device (not readable, a different host's file, a device that has gone), nothing can be said
+    about it and it is counted, which is the old behaviour. That errs high by at most the stacked duplicate and never loses bytes;
+    the maximum single device was rejected as the fallback because it undercounts a container that really writes to two disks.
+
+    `sysfs` is injectable for tests; production reads the real `/sys`.
+    """
+    per = {}
     for line in (text or "").splitlines():
-        for tok in line.split()[1:]:
+        parts = line.split()
+        if not parts:
+            continue
+        acc = per.setdefault(parts[0], {"rbytes": 0, "wbytes": 0})
+        for tok in parts[1:]:
             k, _, v = tok.partition("=")
-            if k in out:
+            if k in acc:
                 try:
-                    out[k] += int(v)
+                    acc[k] += int(v)
                 except ValueError:
                     pass
+    below = set()
+    for mm in per:
+        below |= _stacked_below(sysfs, mm) or set()
+    out = {"rbytes": 0, "wbytes": 0}
+    for mm, v in per.items():
+        if mm in below:
+            continue
+        for k in out:
+            out[k] += v[k]
     return out
 
 

benchmarks/experiments/runner.py

@@ -2374,9 +2374,11 @@ def read_cpu_stat(cg):
     return out
 
 
-def read_io_stat(cg):
-    """Bytes actually read from and written to disk by a container, summed
-    over devices, from cgroup v2 io.stat.
+def read_io_stat(cg, sysfs=memory_probe.SYSFS):
+    """Bytes actually read from and written to disk by a container, from cgroup v2 io.stat.
+
+    Stacked devices are counted once (the dm device on top, not also the NVMe under it): the rule and its fallback are in
+    memory_probe.parse_io_stat. Before 2026-11-06 every device line was summed, which doubled the IO on a stacked disk (the mini).
 
     Pairs with the writable-layer size: that says how big the data ENDED UP,
     this says how much IO it took to get there. The gap between them is write
@@ -2385,16 +2387,12 @@ def read_io_stat(cg):
 
     Free to collect: two reads of one file per cell.
     """
-    out = {"rbytes": 0, "wbytes": 0}
     try:
-        for line in open(os.path.join(cg, "io.stat")):
-            for tok in line.split()[1:]:
-                k, _, v = tok.partition("=")
-                if k in out:
-                    out[k] += int(v)
-    except (OSError, ValueError):
+        with open(os.path.join(cg, "io.stat")) as fh:
+            text = fh.read()
+    except OSError:
         return {}
-    return out
+    return memory_probe.parse_io_stat(text, sysfs)
 
 
 def read_memory_stat(cg):
@@ -3201,6 +3199,10 @@ def run_cell(job, rep, scale, cpuset, tier, net_name):
            # cell ran under; make_paper_tables refuses to mix two in a table.
            "bench_host": os.environ.get("BENCH_HOST"),
            "instrument": bench_common.INSTRUMENT,
+           # THE IO FIGURES OF THIS ROW COUNT A STACKED BLOCK DEVICE ONCE (#555, memory_probe.parse_io_stat). Rows written before
+           # the fix lack this key, and the freeze halves their io figures on a stacked-device host (io_stat_correction.py);
+           # a row carrying it is never halved. Every io_* field of a row is written by this runner, so the stamp lives here.
+           "io_stat_dedup": 1,
            **({"arcadedb_http_client_role": "sensitivity"} if job.get("http_client") else {}),
            # The class the cell ASKED for. The lane also stamps it, from the
            # same variable, beside the string the engine itself reports; both

benchmarks/experiments/test_memory_probe.py

@@ -55,10 +55,100 @@ def test_parsers():
     assert mp.parse_nmt_other("Native memory tracking is not enabled") is None
 
 
-def test_io_stat_is_summed_over_devices():
-    assert mp.parse_io_stat("8:0 rbytes=1024 wbytes=2048 rios=1 wios=2\n259:0 rbytes=1 wbytes=2 dbytes=0\n") == {
+# The real io.stat of a cell container on the mini (2026-11-06): 259:0 is the NVMe, 252:0 the device-mapper device stacked on it,
+# both charged the same 8691597312 written bytes, and 8:0 a small unrelated disk.
+MINI_IO_STAT = """8:0 rbytes=0 wbytes=0 rios=0 wios=5 dbytes=0 dios=0
+259:0 rbytes=0 wbytes=8691597312 rios=0 wios=82042 dbytes=0 dios=0
+252:0 rbytes=0 wbytes=8691597312 rios=0 wios=22387 dbytes=0 dios=0
+"""
+
+
+def _block(sysfs, mm, name, slaves=(), partition_of=None):
+    """One block device in a synthetic sysfs: class/block/<name> (a symlink like the real one) and dev/block/<mm>."""
+    base = sysfs / "devices" / "block"
+    d = base / name if partition_of is None else base / partition_of / name
+    d.mkdir(parents=True, exist_ok=True)
+    (d / "dev").write_text(mm + "\n")
+    if partition_of is not None:
+        (d / "partition").write_text("3\n")
+    (sysfs / "class" / "block").mkdir(parents=True, exist_ok=True)
+    (sysfs / "class" / "block" / name).symlink_to(d)
+    (sysfs / "dev" / "block").mkdir(parents=True, exist_ok=True)
+    (sysfs / "dev" / "block" / mm).symlink_to(d)
+    for sl in slaves:
+        (d / "slaves").mkdir(exist_ok=True)
+        (d / "slaves" / sl).symlink_to(sysfs / "class" / "block" / sl)
+
+
+def _mini_sysfs(tmp_path):
+    """nvme0n1 (259:0) with partition nvme0n1p3 (259:3); dm-0 (252:0) on that partition; sda (8:0) alone."""
+    sysfs = tmp_path / "sys"
+    _block(sysfs, "259:0", "nvme0n1")
+    _block(sysfs, "259:3", "nvme0n1p3", partition_of="nvme0n1")
+    _block(sysfs, "252:0", "dm-0", slaves=["nvme0n1p3"])
+    _block(sysfs, "8:0", "sda")
+    return str(sysfs)
+
+
+def test_io_stat_is_summed_over_unrelated_devices(tmp_path):
+    empty = str(tmp_path / "no-sysfs")                 # no relation info at all: every listed device is counted
+    assert mp.parse_io_stat("8:0 rbytes=1024 wbytes=2048 rios=1 wios=2\n259:0 rbytes=1 wbytes=2 dbytes=0\n", empty) == {
         "rbytes": 1025, "wbytes": 2050}
-    assert mp.parse_io_stat("") == {"rbytes": 0, "wbytes": 0}
+    assert mp.parse_io_stat("", empty) == {"rbytes": 0, "wbytes": 0}
+    assert mp.parse_io_stat("\n\n", empty) == {"rbytes": 0, "wbytes": 0}
+
+
+def test_io_stat_counts_a_stacked_pair_once(tmp_path):
+    sysfs = _mini_sysfs(tmp_path)
+    # the real mini lines: dm-0 on top of the NVMe is one write, not two; the unrelated 8:0 adds its 0
+    assert mp.parse_io_stat(MINI_IO_STAT, sysfs) == {"rbytes": 0, "wbytes": 8691597312}
+    # only the dm device listed, or only the disk listed: nothing to dedupe, the same single count
+    assert mp.parse_io_stat("252:0 rbytes=7 wbytes=9\n", sysfs) == {"rbytes": 7, "wbytes": 9}
+    assert mp.parse_io_stat("259:0 rbytes=7 wbytes=9\n", sysfs) == {"rbytes": 7, "wbytes": 9}
+    # the lower device's own bytes differ from the top's (it also served other IO): the top is kept, the lower one dropped
+    two = "259:0 rbytes=100 wbytes=500\n252:0 rbytes=40 wbytes=300\n"
+    assert mp.parse_io_stat(two, sysfs) == {"rbytes": 40, "wbytes": 300}
+
+
+def test_io_stat_unrelated_third_device_is_still_counted(tmp_path):
+    sysfs = _mini_sysfs(tmp_path)
+    text = MINI_IO_STAT.replace("8:0 rbytes=0 wbytes=0", "8:0 rbytes=4096 wbytes=1024")
+    assert mp.parse_io_stat(text, sysfs) == {"rbytes": 4096, "wbytes": 8691597312 + 1024}
+
+
+def test_io_stat_two_layer_stack(tmp_path):
+    sysfs = tmp_path / "sys2"
+    _block(sysfs, "259:0", "nvme0n1")
+    _block(sysfs, "252:0", "dm-0", slaves=["nvme0n1"])
+    _block(sysfs, "252:1", "dm-1", slaves=["dm-0"])       # LUKS on LVM on the disk
+    text = "259:0 rbytes=1 wbytes=1\n252:0 rbytes=1 wbytes=1\n252:1 rbytes=1 wbytes=1\n"
+    assert mp.parse_io_stat(text, str(sysfs)) == {"rbytes": 1, "wbytes": 1}
+
+
+def test_io_stat_without_sysfs_info_counts_every_device(tmp_path):
+    # the old behaviour, and the documented fallback: with nothing known about the devices, no bytes are dropped
+    assert mp.parse_io_stat(MINI_IO_STAT, str(tmp_path / "missing")) == {"rbytes": 0, "wbytes": 2 * 8691597312}
+    partial = tmp_path / "partial"                        # sysfs knows the NVMe only: 252:0 has no entry, so no relation is found
+    _block(partial, "259:0", "nvme0n1")
+    assert mp.parse_io_stat(MINI_IO_STAT, str(partial)) == {"rbytes": 0, "wbytes": 2 * 8691597312}
+
+
+def test_io_stat_tolerates_bad_input(tmp_path):
+    sysfs = _mini_sysfs(tmp_path)
+    assert mp.parse_io_stat(None, sysfs) == {"rbytes": 0, "wbytes": 0}
+    assert mp.parse_io_stat("8:0 rbytes=oops wbytes=5\n", sysfs) == {"rbytes": 0, "wbytes": 5}
+
+
+def test_runner_read_io_stat_uses_the_stacked_rule(tmp_path):
+    import runner as RN
+    sysfs = _mini_sysfs(tmp_path)
+    cg = tmp_path / "cg"
+    cg.mkdir()
+    assert RN.read_io_stat(str(cg), sysfs) == {}                     # no io.stat: empty, as before
+    (cg / "io.stat").write_text(MINI_IO_STAT)
+    assert RN.read_io_stat(str(cg), sysfs) == {"rbytes": 0, "wbytes": 8691597312}
+    (cg / "io.stat").write_text("")
+    assert RN.read_io_stat(str(cg), sysfs) == {"rbytes": 0, "wbytes": 0}
 
 
 def test_phase_marks():
@@ -237,6 +327,14 @@ def test_runner_hooks_are_gated():
     assert src.index("    _mp = None") < src.index("memory_probe.HostProbe.maybe(")
 
 
+def test_runner_stamps_io_stat_dedup_where_it_builds_the_row():
+    src = open(os.path.join(HERE, "runner.py")).read()
+    assert src.count('"io_stat_dedup": 1,') == 1
+    # in the one row literal that also writes the instrument, before any io_* field is assigned to it
+    row_start = src.index('row = {"run_id": run_id, "engine_commit"')
+    assert row_start < src.index('"io_stat_dedup": 1,') < src.index('row[f"{name}_io_read_mib"]')
+
+
 def test_export_web_prints_the_per_phase_memory_column():
     pytest.importorskip("numpy")
     import export_web as ew