diff --git a/benchmarks/experiments/memory_probe.py b/benchmarks/experiments/memory_probe.py index 45a504911c8..b1011f0f146 100644 --- a/benchmarks/experiments/memory_probe.py +++ b/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 `/dev/block//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 diff --git a/benchmarks/experiments/runner.py b/benchmarks/experiments/runner.py index 16a2fb642ba..d51d2700929 100644 --- a/benchmarks/experiments/runner.py +++ b/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 diff --git a/benchmarks/experiments/test_memory_probe.py b/benchmarks/experiments/test_memory_probe.py index cb5a2186589..a9275cf9d78 100644 --- a/benchmarks/experiments/test_memory_probe.py +++ b/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/ (a symlink like the real one) and dev/block/.""" + 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