diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..cba2cca --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,125 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## What this is + +`mlperf-common` is a grab-bag of utilities shared across NVIDIA's MLPerf benchmark +submissions. Benchmarks consume it by adding +`git+https://github.com/NVIDIA/mlperf-common.git` to their `requirements.txt`. + +Two mostly independent halves live here: + +* **Logging / profiling** (`mlperf_common/logging.py`, `scaleoutbridge.py`, + `frameworks/`, `callbacks/`) — wraps the official `mlperf_logging` package. +* **Fast file I/O** (`mlperf_common/fileio/`, `client/`, `src/`) — O_DIRECT copy + and checksum tools, and the shared tree-walk and I/O primitives behind them. + +## Commands + +```bash +python3 tests/run_tests.py # whole suite (stdlib only, no pytest, no GPU) +python3 tests/test_copyplan.py # one test file, directly + +make -C src # build the C++ tools +make -C src install prefix=/usr/local + +pip install . # installs the package + client/ scripts into bin/ +``` + +There is no linter or formatter configured, and no CI in the repo. + +Tests run each file in a **separate interpreter** so nothing one leaves in +`sys.modules` reaches another. They are stdlib-only and need no GPU. See +`tests/README.md` for what they do and do not cover — notably not `direct_io` +and not O_DIRECT itself. + +## Architecture + +### Logging stack + +`MLLoggerWrapper` (`mlperf_common/logging.py`) is the entry point. It takes a +`CommunicationHandler` so the same wrapper works under either `torch.distributed` +or MPI; `frameworks/base.py` defines the `CommunicationHandler` / +`ProfilerHandler` interfaces and `frameworks/{pyt,mxnet,hugectr,base_mpi}.py` +implement them per framework. Adding framework support means implementing those +two interfaces, not touching the wrapper. + +`scaleoutbridge.py` layers profiling on top: `init_bridge(prof_handler, +comm_handler, mllogger)` picks a bridge implementation from env vars — +`TIME_TAGS` / `NVTX_FLAG` select `ScaleoutBridgeIterwise`, `EPOCH_PROF` selects +`ScaleoutBridgeEpochwise`, and with none set you get the no-op +`ScaleoutBridgeBase`. + +`callbacks/logging.py` is the Lightning/NeMo layer (`LoggingCallback`, +`MLPerfLogger`, `StatsLogCallback`). It imports `lightning.pytorch` with a +fallback to `pytorch_lightning`, and is knob-driven by env vars +(`FORCE_SUCCESS_STATUS`, `REDUCE_TP`, `LOG_EVERY_N_BATCHES`, `RUN_N_ITERS`, +`SEED`). This is the only part that pulls in Lightning and numpy. + +### fileio + +Two modules, with a deliberate dependency rule stated in +`mlperf_common/fileio/__init__.py`: **neither may import torch.** They stay +stdlib-only so the single-node `client/` scripts don't drag in a training stack. +Anything needing torch belongs in a module of its own. + +* `direct_io.py` — aligned-buffer `pread`/`pwrite` with retry loops. + `allocate_aligned_buffers` carves per-thread buffers out of one big ctypes + allocation and hands back **memoryviews**, so slicing a buffer to the length + actually read costs nothing. +* `copyplan.py` — source-tree walk and src→dst mapping. `plan_copy_operations` + raises `UnreadableEntries` listing *every* bad entry rather than dying on the + first, because a walk that silently skips an unlistable subtree yields a + partial copy that exits 0 — and then a checksum run that skips the same files + and agrees with it. + +Both tools align to 2 MiB, the Linux huge-page size. It is *not* a shared +constant -- it is hardcoded in each (`fastcp:90,180`, `fastmd5:44,121`). Worth +promoting into `direct_io` next to `round_up`, so the two cannot drift. + +This package was extracted from `fastcp` and `fastmd5`, which had been carrying +their own copies. Keep the two scripts going through it rather than reintroducing +private variants. + +### client/ and src/ + +`client/` holds the single-node scripts. `setup.py`'s `scripts=` installs +`bindpcie` (NUMA/IB affinity binding), `mgpurun`, `slurm2pytorch` (derives +PyTorch rendezvous env from SLURM), `fastcp`, `direct_io.py`, and the +log/telemetry shell helpers. + +**`fastmd5` and `dropcache` are in `client/` but are NOT in `scripts=`**, so a +`pip install` does not produce them -- the only supported way to run `fastmd5` +is in place from a checkout. That matters more now that it hard-depends on the +package, because its own ImportError advice says "install mlperf-common", a +remedy that never yields a `fastmd5`. Either add it to `scripts=` or fix the +message; don't leave both. + +`fastcp` and `fastmd5` are a threaded copy and a per-GB checksum. `fastcp` +opens with O_DIRECT; `fastmd5` does not, despite using the same aligned-buffer +machinery. + +**Don't delete `slurm2pytorch`.** Benchmarks outside this repo depend on it and +it stays installed, even though nothing in this repo calls it. + +These scripts import `direct_io` and `mlperf_common.fileio.copyplan` via a +`sys.path` dance that works both for a pip install and for a source tree with +`mlperf_common/` alongside `client/` (the "deploy the repo to a shared filesystem +and run in place" pattern). `client/direct_io.py` is a compatibility shim +re-exporting `mlperf_common.fileio.direct_io`. Copying individual scripts out of +`client/` on their own does not work — preserve that when touching imports, and +if you add a script remember to add it to `scripts=` in `setup.py`. + +`src/` is a separate, older C++ implementation (`fastcp`, `cp-into`, +`alloc-empty-file-buffer`) sharing `cmdline.h`. It is not built or installed by +`setup.py`; use the `Makefile` directly. + +## Notes + +* Commit subjects follow `area: lowercase imperative summary`. +* New files carry the Apache 2.0 header with an NVIDIA copyright line. +* The README's "Mount check" section documents `get-mount-info.sh` / + `verify-mounts.sh`, which no longer exist — `mlperf_common/mountcheck.py` + (`--initialize` / verify against `expected-mounts.csv`, sparse SHA256 + fingerprints) replaced them. Trust the module over the README there. diff --git a/FASTCP-BUGS.md b/FASTCP-BUGS.md new file mode 100644 index 0000000..1fff746 --- /dev/null +++ b/FASTCP-BUGS.md @@ -0,0 +1,82 @@ +# Known defects in fastcp / fastmd5 / fileio + +Found reviewing the `mlperf_common.fileio` extraction (PR #43). + +Everything here was **introduced by that extraction** — each item was checked +against `main`'s pre-refactor `client/fastcp` and does not occur there. The much +longer list of defects this code inherited from `main` is deliberately *not* +here: those predate the refactor and are not this branch's to fix. They are +recorded in `MAIN-BUGS.md`. + +`direct_io.py` is byte-identical to `main`'s modulo comments, so nothing in it +is a new defect. + +--- + +## 1. The "cannot import mlperf_common" advisory is unreachable + +`client/fastcp:35-42` and `client/fastmd5:20-27` wrap the `copyplan` import in a +`try`/`except ImportError` that exits with a one-line explanation naming both +remedies. It never runs: the *unguarded* `import direct_io` at `fastcp:32` / +`fastmd5:18` executes first and fails first. + +Reproduced — copy `client/fastcp` and `client/direct_io.py` somewhere with no +`mlperf_common` and run it: + +``` +Traceback (most recent call last): + File "./fastcp", line 32, in + import direct_io + ... +ImportError: No module named 'mlperf_common'. direct_io lives in the +mlperf_common package. ... +``` + +A chained three-frame traceback, in the exact scenario the friendly message was +written for. The shim's own text is reasonable, so this is cosmetic — but the +handler above it is dead code and reads as though it works. + +Fix: move `import direct_io` inside the same guarded block, or drop the +now-redundant handler. + +## 2. A stale `direct_io.py` beside the script silently shadows the package + +`sys.path.insert(0, _HERE)` puts the script's own directory ahead of everything, +so a leftover standalone `direct_io.py` in an install's `bin/` outranks the +shim. Reproduced with `main`'s pre-refactor copy dropped next to `fastcp`: + +``` +direct_io resolved to : .../bin/direct_io.py +same module object : False +same pread function : False +``` + +The copy then runs the *stale* I/O primitives while `copyplan` comes from the +package. Harmless today only because the two are byte-identical — which means +the first edit to `mlperf_common/fileio/direct_io.py` silently stops reaching +any install carrying a leftover `bin/direct_io.py`, with no warning. + +This is a live risk precisely because `setup.py` *does* install +`client/direct_io.py` into `bin/`, so upgrading from a pre-refactor install +leaves the old file there. + +Related, not reproduced: `os.path.abspath` does not resolve symlinks, so +`ln -s /client/fastcp ~/bin/fastcp` makes `_HERE` `~/bin` and both inserts +point at the wrong tree. `os.path.realpath` would fix both. + +## 3. `test_copyplan.py`'s refusal rows do not check *which* error was raised + +`tests/test_copyplan.py:106-110` asserts only that some `CopyArgumentError` +escaped: + +```python +try: + copyplan.validate_copy_args(sources, destination, **kwargs) + check(name, False, "no exception raised") +except copyplan.CopyArgumentError: + check(name, True) +``` + +A misspelled fixture path, or a validator that grew over-eager and rejects the +wrong thing for the wrong reason, leaves every one of these rows green. Match +the message, or at least a distinct exception subclass per rule. diff --git a/MAIN-BUGS.md b/MAIN-BUGS.md new file mode 100644 index 0000000..d206b73 --- /dev/null +++ b/MAIN-BUGS.md @@ -0,0 +1,225 @@ +# fastcp / fastmd5 / direct_io — defects present on `main` + +**Status: not filed anywhere.** Recorded here because `NVIDIA/mlperf-common` has +GitHub issues disabled (`hasIssuesEnabled: false`), so there is no tracker to +file into. This file is the record until there is one; move it wherever the team +actually tracks bugs, and delete it here when that happens. + +- Found: 2026-09-02, reviewing PR #43 (`mfrank/shared-fileio`). +- Verified against: `main` @ `b5014f3`, via `git show main:client/fastcp` run directly. +- **None of these is caused by PR #43.** Each reproduces identically before and + after that refactor; `client/direct_io.py` is byte-identical across the two + modulo comments. The three defects that PR *did* introduce are tracked in + `FASTCP-BUGS.md` on that branch and are not repeated here. +- Deliberately excluded: anything only reachable on the `mfrank/py-data-staging` + branch. + +To re-run any of these against main: + +```bash +cd /home/matt/work/jun-2026/mlperf-common +mkdir -p /tmp/mv && git show main:client/fastcp > /tmp/mv/fastcp +git show main:client/direct_io.py > /tmp/mv/direct_io.py +python3 /tmp/mv/fastcp ... +``` + +Ordered by severity. + +--- + +# DATA LOSS + +## 1. `fastcp -f SRC .` truncates the source to zero + +```bash +mkdir -p /tmp/t1 && cd /tmp/t1 && head -c 50000 /dev/urandom > a.bin +python3 /tmp/mv/fastcp -f a.bin . +ls -l a.bin # 0 bytes. exit was 0. +``` + +DEST resolves to `/tmp/t1/a.bin`; `fastcp` opens it `O_TRUNC` *before* fstat'ing +the source, so the size it then copies is 0. GNU cp refuses outright: +`'a.bin' and 'a.bin' are the same file`, exit 1. + +`validate_copy_args` has no same-file guard. Without `-f` the pre-existence +check saves you; with `-f` — which any re-runnable staging script needs — the +source is unrecoverable. + +**Fix:** compare `(st_dev, st_ino)` of source and resolved destination in +`validate_copy_args`, and refuse. + +## 2. `fastcp -r -f src/.. dst` destroys files outside the copy + +```bash +mkdir -p /tmp/t2/src /tmp/t2/dst && cd /tmp/t2 +echo SRC-DATA > src/a.txt && echo VICTIM-DATA > victim.txt +python3 /tmp/mv/fastcp -r -f src/.. dst +wc -c victim.txt src/a.txt # both 0. dst/ is empty. exit was 0. +``` + +Destinations are built from `os.path.basename(src.rstrip("/"))`, which returns +`".."`, so every planned destination is literally `dst/../` — i.e. back +outside DEST, onto the sources themselves. GNU cp refuses: +`cannot copy a directory, 'src/..', into itself`. + +`basename('/') == ''` collapses `/` into DEST the same way. Without `-f` the +same command writes files outside DEST rather than destroying them. + +**Fix:** normalise and reject any source whose basename is `.`, `..` or empty; +or refuse when the resolved destination is not strictly under DEST. + +--- + +# SILENT WRONG RESULTS + +## 3. `-n 0` writes a full-size file of zeros and exits 0 + +```bash +mkdir -p /tmp/t3 && cd /tmp/t3 && head -c 50000 /dev/urandom > a.bin +python3 /tmp/mv/fastcp -n 0 a.bin out.bin +ls -l out.bin # 50000 bytes +python3 -c "d=open('out.bin','rb').read(); print(d == b'\x00'*len(d))" # True +``` + +`--num-threads` is never validated. `range(0)` makes both the thread-start and +the join loops no-ops, then `os.ftruncate(fd_dst, file_size)` inflates the empty +destination to the source's size. + +The verification half fails the same way: `fastmd5 -n 0 ` emits **zero +lines** and exits 0, so diffing a source checksum run against a destination one +reports them **identical**. The two failures compound into a clean-looking +staging job that copied nothing. + +Reachable without anyone typing `0`: `-n $(nproc)` inside a constrained cgroup, +or an unset shell variable arithmetic-expanded to 0. + +Same missing guard on `-b`: `fastmd5 -b 0` dies with `ZeroDivisionError` instead +of its intended message. + +**Fix:** require `--num-threads >= 1` and `--buffer-size > 0` in both tools. + +## 4. A dead worker thread yields a same-size, wrong-bytes destination + +An exception inside `copy_worker` kills only that thread. `join()` returns +normally, `ftruncate` sets the correct size, and `fastcp` exits 0. Source and +destination sizes match; contents do not. + +`fastmd5`'s `checksum_worker` docstring states the rule verbatim — *"an exception +escaping a thread does not affect the process exit status"* — and `fastmd5` was +fixed for exactly this. `fastcp`, the tool that *writes* data, was not. + +Related: `fastcp`'s bare `except Exception` around `workpile.get_nowait()` +swallows non-`Empty` errors, where `fastmd5` correctly narrows to `queue.Empty`. + +**Fix:** port `fastmd5`'s pattern — collect worker exceptions into a shared list +and exit nonzero if it is non-empty. + +## 5. Empty directories are silently omitted + +```bash +mkdir -p /tmp/t5/src/hollow /tmp/t5/dst && echo x > /tmp/t5/src/plain +python3 /tmp/mv/fastcp -r /tmp/t5/src /tmp/t5/dst +ls /tmp/t5/dst/src # 'plain' only; 'hollow' is missing. exit was 0. +``` + +`cp -r` recreates it. Fixed on the PR #43 branch (commit `01f4634`) by planning +directories alongside files; `main` is still affected. + +**Correcting `01f4634`'s commit message,** which calls this "a regression the +fileio extraction introduced". That is only half true, and the halves differ: + +| case | `main` | before `01f4634` | after | +| --- | --- | --- | --- | +| DEST exists, source has an empty subdir | exit 0, subdir missing | exit 0, subdir missing | exit 0, created | +| DEST absent, source wholly empty | **exit 1**, loud | exit 0, DEST never created | exit 0, created | + +So the first row is a defect `main` already had and the extraction carried +across unchanged — it belongs in this file, which is why it is here. Only the +second row regressed, and it regressed in the way that matters most: `main` +failed loudly enough for a caller checking the exit status to notice, and the +refactor turned that into a silent success with an incomplete tree. + +The fix is correct for both rows. Only the commit message overstated its scope, +and it is not worth a force-push to reword. + +## 6. Duplicate destination names clobber without `--force` + +`fastcp x/same y/same dst` writes both to `dst/same`, last writer wins, exit 0. +GNU cp refuses with `will not overwrite just-created`. + +--- + +# HANGS + +## 7. A FIFO anywhere in the tree hangs the copy forever + +```bash +mkdir -p /tmp/t7/src /tmp/t7/dst && echo hi > /tmp/t7/src/a.txt +mkfifo /tmp/t7/src/pipe +timeout 10 python3 /tmp/mv/fastcp -r /tmp/t7/src /tmp/t7/dst # exit 124 +``` + +Blocks in `open(O_RDONLY)` with no diagnostic and no timeout. `cp -r` completes +in milliseconds and recreates it as a FIFO. `fastmd5` silently skips it, so the +checksum reports a clean tree the copier can never finish. In a multi-rank +staging job this is a hang on one rank while the others wait. + +**Fix:** `list_relative_files` should classify non-regular files and either skip +them with a warning or refuse the copy. + +## 8. Unbounded retry loops in `direct_io` + +`pread` and `pwrite` re-issue the identical syscall at the identical offset on a +short transfer, with no iteration bound and no progress check. A *permanent* +short read/write therefore spins at 100% CPU forever instead of raising. + +- Read side: a file that shrinks between `fastmd5`'s `os.path.getsize` at enqueue + and the worker's `pread` minutes later hangs that worker; `main()`'s `join()` + never returns. +- Write side: this is the ENOSPC shape. A destination filling up makes `fastcp` + spin forever rather than report `No space left on device`. + +Same area, two more: + +- The retry guards compare against `count` where they need `padded_count`, so a + legitimate partial transfer raises an assertion describing the opposite of + what happened, instead of reaching the retry that would have worked. +- Under `python -O` every `assert` in the module vanishes. They are the module's + **only** validation, and it has no test coverage at all — including + `allocate_aligned_buffers`' alignment check, whose failure leaves buffers + misaligned and every O_DIRECT call returning EINVAL. + +**Fix:** bound the retries, require forward progress between iterations, and +convert the asserts to real exceptions. + +--- + +# PERMISSIONS + +## 9. Destination mode is never preserved + +```bash +umask 002 +mkdir -p /tmp/t9/src /tmp/t9/dst && head -c 100 /dev/urandom > /tmp/t9/src/secret +chmod 600 /tmp/t9/src/secret +python3 /tmp/mv/fastcp -r /tmp/t9/src /tmp/t9/dst +stat -c '%a' /tmp/t9/dst/src/secret # 775, was 600 +``` + +`os.open` is called with no mode argument, so destinations are created +`0o777 & ~umask`. `os.makedirs` does the same for directories — a `0o700` source +subdirectory becomes `0o775`. GNU cp preserves the mode in both cases. + +Staging a restricted-license dataset onto shared node-local storage therefore +widens its permissions to every other user on the node. Worth treating as a +security issue, not a cosmetic one. + +--- + +# COSMETIC + +## 10. `fastmd5`'s tab-separated output is unescaped + +A filename containing a tab yields five fields instead of four, so any consumer +splitting on tabs mis-parses the line. An undecodable filename additionally makes +the error handler re-raise `UnicodeEncodeError` from its own diagnostic print. diff --git a/client/direct_io.py b/client/direct_io.py index 99424ed..e2f4713 100644 --- a/client/direct_io.py +++ b/client/direct_io.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 -# Copyright (c) 2024-2025, NVIDIA CORPORATION. All rights reserved. +# Copyright (c) 2024-2026, NVIDIA CORPORATION. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -14,137 +14,30 @@ # See the License for the specific language governing permissions and # limitations under the License. -import os -import ctypes - -def print_memoryview(mv): - """Print the address and size of a memoryview.""" - address = ctypes.addressof(ctypes.c_char.from_buffer(mv)) - size = mv.nbytes - print(f"Address: {hex(address)}, Size: {hex(size)}") - -def allocate_aligned_buffers(buffer_size, alignment, num_bufs): - """Allocate a sequence of aligned buffers. - - Args: - buffer_size (int): Size of each buffer in bytes. - alignment (int): Alignment boundary in bytes. - num_bufs (int): Number of buffers to allocate. - - Returns: - list[memoryview]: A list of memoryview objects, each representing an aligned buffer - raw_buf: the ctypes raw string_buffer. This *must* be kept to avoid the - gc prematurely collecting the buffers. - """ - assert buffer_size % alignment == 0, "buffer_size must be a multiple of alignment" - - # Allocate a single large buffer that can accommodate all buffers with proper alignment - total_size = (num_bufs * buffer_size) + alignment - raw_buf = ctypes.create_string_buffer(total_size) - base_address = ctypes.addressof(raw_buf) - - # Calculate the offset to align the first buffer - first_offset = (alignment - (base_address % alignment)) % alignment - aligned_base_address = base_address + first_offset - - # Create memoryviews for each buffer - buffers = [] - for i in range(num_bufs): - buf_address = aligned_base_address + i * buffer_size - aligned_buf = (ctypes.c_char * buffer_size).from_address(buf_address) - mv = memoryview(aligned_buf) - # print_memoryview(mv) - buffers.append(mv) - return buffers, raw_buf - -def round_up(value, multiple): - """Round up `value` to the next multiple of `multiple`.""" - if multiple == 0: - raise ValueError("Multiple must be greater than 0.") - return ((value + multiple - 1) // multiple) * multiple - -def pread(fd, aligned_memview, count, offset, fs_block_size, thread_id): - """Perform a direct pread. - - Returns: - number of bytes read. Unlike Posix pread this routine handles - interrupts, so return value should always be equal to count unless - there is an unrecoverable error, in which case it throws rather than returns - - Required: `count` <= aligned_memview size - """ - padded_count = round_up(count, fs_block_size) - assert padded_count <= aligned_memview.nbytes, "memview too small for requested read" - - while True: - try: - bytes_read = os.preadv(fd, [aligned_memview[:padded_count]], offset) +# Compatibility shim. direct_io now lives in the mlperf_common package so that +# it can be imported as mlperf_common.fileio.direct_io, and not only by the +# scripts installed alongside it in bin/. This shim keeps `import direct_io` +# working for anything that still expects a module next to fastcp. - if bytes_read == count: - # Expected case: all requested bytes were read. - return bytes_read +# The package is found either because mlperf-common is installed (setup.py +# installs the package and these scripts together, so this holds for any pip +# install) or because we are running from a tree with mlperf_common/ next to +# client/ -- which covers the source checkout and the "deploy the repo to a +# shared filesystem and run client/ scripts in place" pattern. Copying +# individual scripts out of client/ on their own does not work. - if bytes_read > 0: - # Retry case: partial read, retry the entire read to maintain alignment restrictions - assert bytes_read < count, "bytes_read cannot exceed count" - continue - - if bytes_read == 0: - # Unexpected EOF - raise RuntimeError(f"Unexpected EOF encountered at offset {offset}") - - if bytes_read < 0: - raise OSError("preadv returned a negative value") - - except InterruptedError: - # Retry on EINTR - continue - except OSError as e: - # Non-retriable errors - if e.errno == 14: # Errno 14 corresponds to "Bad address" - raise RuntimeError(f"preadv failed with error: {e}, this usually means out of memory") - raise RuntimeError(f"preadv failed with error: {e}") - -def pwrite(fd, aligned_memview, count, offset, fs_block_size): - """Perform a direct pwrite - - Returns: - number of bytes written. Unlike Posix pwrite this routine handles - interrupts, so return value should always be equal to count unless - there is an unrecoverable error, in which case it throws rather than - returns - - Required: `count <= aligned_memview size - """ - padded_count = round_up(count, fs_block_size) - assert padded_count <= aligned_memview.nbytes, "memview too small for requested write" - while True: - try: - bytes_written = os.pwritev(fd, [aligned_memview[:padded_count]], offset) - - if bytes_written == padded_count: - # Expected case: all requested bytes were written. - # in the case of the last block in the file we may have padded up, so return - # the bytes _requested_ rather than the padded count - return count - - if bytes_written > 0: - # Retry case: partial write, retry the entire write to maintain alignment restrictions - assert bytes_written < count, "bytes_written cannot exceed count" - continue - - if bytes_written == 0: - # Unexpected failure to write - raise RuntimeError(f"Unexpected failure to write at offset {offset}") - - if bytes_written < 0: - raise OSError("pwritev returned a negative value") - - except InterruptedError: - # Retry on EINTR - continue - except OSError as e: - # Non-retriable errors - if e.errno == 14: # Errno 14 corresponds to "Bad address" - raise RuntimeError(f"pwritev failed with error: {e}, this usually means out of memory") - raise RuntimeError(f"pwritev failed with error: {e}") +import os +import sys + +try: + from mlperf_common.fileio.direct_io import * # noqa: F401,F403 +except ImportError: + sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), os.pardir)) + try: + from mlperf_common.fileio.direct_io import * # noqa: F401,F403 + except ImportError as exc: + raise ImportError( + f"{exc}. direct_io lives in the mlperf_common package. Either install " + "mlperf-common, or keep this script in a tree with mlperf_common/ " + "alongside client/." + ) from exc diff --git a/client/fastcp b/client/fastcp index b7787ab..bdc8194 100755 --- a/client/fastcp +++ b/client/fastcp @@ -21,10 +21,26 @@ import tempfile import queue import threading -# find direct_io in the same directory as this program -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +# Find direct_io next to this program, and mlperf_common one level up. That +# covers an install (setup.py installs the package and these scripts together, +# so the package is importable) and a tree with mlperf_common/ next to client/, +# which is both the source checkout and the "deploy the repo to a shared +# filesystem and run client/ scripts in place" pattern. +_HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _HERE) +sys.path.insert(1, os.path.join(_HERE, os.pardir)) import direct_io +try: + from mlperf_common.fileio.copyplan import ( + CopyArgumentError, UnreadableEntries, plan_copy_operations, validate_copy_args) +except ImportError as exc: + sys.exit( + f"fastcp: cannot import mlperf_common ({exc}).\n" + "fastcp needs the mlperf_common package: either install mlperf-common, " + "or run this script from a tree with mlperf_common/ alongside client/." + ) + def copy_worker(fd_src, fd_dst, buffer_size, workpile, thread_buffer, fs_block_size, thread_id): """Worker function to copy chunks from the workpile.""" while True: @@ -147,68 +163,17 @@ def parse_and_validate_args(): # usually meant that to be the source file name, and forgot to give a target parser.error(f"{prog_name}: missing destination file operand after '{args.destination}'") - # Validate sources - for src in args.sources: - if not os.path.exists(src): - sys.exit(f"{prog_name}: cannot stat '{src}': No such file or directory") - if os.path.isdir(src) and not args.recursive: - sys.exit(f"{prog_name}: -r not specified; omitting directory '{src}'") - - # Validate destination - if args.target_directory or len(args.sources) > 1: - if not os.path.exists(args.destination) or not os.path.isdir(args.destination): - sys.exit(f"{prog_name}: target '{args.destination}' is not a directory") - else: - if os.path.exists(args.destination) and os.path.isfile(args.destination): - if os.path.isdir(args.sources[0]): # Only valid if both are files - sys.exit(f"{prog_name}: cannot overwrite '{args.destination}' with directory '{args.sources[0]}'") + # The cp argument rules live in copyplan, next to the mapping they make + # well defined, so that everything planning a copy agrees on them. + try: + validate_copy_args(args.sources, args.destination, + recursive=args.recursive, + into_directory=bool(args.target_directory)) + except CopyArgumentError as exc: + sys.exit(f"{prog_name}: {exc}") return args -def list_relative_files(root): - """Return all file paths under 'root' as relative paths, sorted alphabetically. - os.walk(followlinks=True) allows following symlinked directories, - but it does not guard against cycles. This may cause infinite loops - if symlinks form a directory cycle. - """ - file_list = [] - # FIXME: os.walk(followlinks) doesn't protect against cycles - # to fix this we'd need to write our own version that did a depth-first - # spanning tree. - for dirpath, _, filenames in os.walk(root, followlinks=True): - for fname in filenames: - full_path = os.path.join(dirpath, fname) - rel_path = os.path.relpath(full_path, root) - file_list.append(rel_path) - return sorted(file_list) - -def plan_copy_operations(args): - """Return list of (src_abs, dst_abs, size_bytes) file tuples to copy.""" - file_jobs = [] - dst_root = os.path.abspath(args.destination) - - if not os.path.isdir(dst_root): # case 1: single file copy - src_abs = os.path.abspath(args.sources[0]) - dst_abs = os.path.abspath(args.destination) - size = os.path.getsize(src_abs) - file_jobs.append((src_abs, dst_abs, size)) - else: - for src in args.sources: - src_abs = os.path.abspath(src) - base = os.path.basename(src.rstrip("/")) - if os.path.isdir(src): - for relpath in list_relative_files(src): - full_src = os.path.join(src_abs, relpath) - full_dst = os.path.join(dst_root, base, relpath) - size = os.path.getsize(full_src) - file_jobs.append((full_src, full_dst, size)) - else: - dst_path = os.path.join(dst_root, base) - size = os.path.getsize(src_abs) - file_jobs.append((src_abs, dst_path, size)) - - return file_jobs - if __name__ == "__main__": args = parse_and_validate_args() # round buffer size up to next multiple of 2 MiB @@ -219,14 +184,21 @@ if __name__ == "__main__": print(f"Sources: {args.sources}") print(f"Destination: {args.destination}") - file_jobs = plan_copy_operations(args) + try: + plan = plan_copy_operations(args.sources, args.destination) + except UnreadableEntries as exc: + sys.exit(f"fastcp: {exc}") if not args.force: - for _, dst, _ in file_jobs: + for _, dst, _ in plan.files: if os.path.exists(dst): sys.exit(f"{os.path.basename(sys.argv[0])}: will not overwrite existing file '{dst}' without --force") - for src, dst, size in file_jobs: + # Up front, and shallowest first, so that an empty source directory is + # recreated even though no file names it -- cp -r creates it too. + for dirpath in plan.directories: + os.makedirs(dirpath, exist_ok=True) + + for src, dst, size in plan.files: print(f"Copy {src} -> {dst} ({size} bytes)") - os.makedirs(os.path.dirname(dst), exist_ok=True) fastcp(src, dst, args.num_threads, args.buffer_size) diff --git a/client/fastmd5 b/client/fastmd5 index adf7644..710dfa8 100755 --- a/client/fastmd5 +++ b/client/fastmd5 @@ -6,12 +6,26 @@ import sys import queue import hashlib import threading -from pathlib import Path -# find direct_io in the same directory as this program -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +# Find direct_io next to this program, and mlperf_common one level up. That +# covers an install (setup.py installs the package and these scripts together, +# so the package is importable) and a tree with mlperf_common/ next to client/, +# which is both the source checkout and the "deploy the repo to a shared +# filesystem and run client/ scripts in place" pattern. +_HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _HERE) +sys.path.insert(1, os.path.join(_HERE, os.pardir)) import direct_io # for pread(), allocate_aligned_buffers(), round_up() +try: + from mlperf_common.fileio.copyplan import UnreadableEntries, list_relative_files +except ImportError as exc: + sys.exit( + f"fastmd5: cannot import mlperf_common ({exc}).\n" + "fastmd5 needs the mlperf_common package: either install mlperf-common, " + "or run this script from a tree with mlperf_common/ alongside client/." + ) + def parse_and_validate_args(): """Parse command-line arguments. Always assume recursive for directories.""" parser = argparse.ArgumentParser( @@ -47,30 +61,41 @@ def enqueue_file_chunks(display_name, actual_path, chunk_size, workpile): this_chunk = min(chunk_size, size - offset) workpile.put((display_name, offset, this_chunk, block_size)) -def checksum_worker(thread_id, buffer, workpile, buffer_size, print_lock): - """Worker function to compute md5 checksums of file chunks.""" +def checksum_worker(thread_id, buffer, workpile, buffer_size, print_lock, errors): + """Worker function to compute md5 checksums of file chunks. + + A failure here must not be silent: an exception escaping a thread does not + affect the process exit status, so a caller comparing two trees would see + a short but successful-looking result. Record it instead and let main() + exit nonzero. + """ while True: try: filepath, offset, size, fs_block_size = workpile.get_nowait() except queue.Empty: return - md5 = hashlib.md5() - - with open(filepath, 'rb') as f: - fd = f.fileno() - remaining = size - local_offset = offset - while remaining > 0: - this_read = min(remaining, buffer_size) - bytes_read = direct_io.pread(fd, buffer, this_read, local_offset, fs_block_size, thread_id) - assert bytes_read == this_read, f"pread returned {bytes_read}, expected {this_read}" - md5.update(buffer[:this_read]) - local_offset += this_read - remaining -= this_read - - with print_lock: - print(f"{filepath}\t{offset}\t{size}\t{md5.hexdigest()}") + try: + md5 = hashlib.md5() + + with open(filepath, 'rb') as f: + fd = f.fileno() + remaining = size + local_offset = offset + while remaining > 0: + this_read = min(remaining, buffer_size) + bytes_read = direct_io.pread(fd, buffer, this_read, local_offset, fs_block_size, thread_id) + assert bytes_read == this_read, f"pread returned {bytes_read}, expected {this_read}" + md5.update(buffer[:this_read]) + local_offset += this_read + remaining -= this_read + + with print_lock: + print(f"{filepath}\t{offset}\t{size}\t{md5.hexdigest()}") + except Exception as exc: # noqa: BLE001 - reported and re-surfaced by main() + errors.append((filepath, offset, exc)) + with print_lock: + print(f"fastmd5: {filepath} at offset {offset}: {exc}", file=sys.stderr) workpile.task_done() def main(): @@ -78,16 +103,19 @@ def main(): workpile = queue.Queue() CHUNK_SIZE = 1 << 30 # 1 GiB - for root in args.paths: - original_arg = root - path = Path(original_arg) - if path.is_dir(): - for file in path.rglob('*'): - if file.is_file() or file.is_symlink(): - display_name = os.path.join(original_arg, os.path.relpath(file, path)) - enqueue_file_chunks(display_name, file, CHUNK_SIZE, workpile) - elif path.is_file() or path.is_symlink(): - enqueue_file_chunks(original_arg, path, CHUNK_SIZE, workpile) + # Share the tree walk with fastcp, so that checksumming a copy compares the + # same set of files the copy was planned from. + for original_arg in args.paths: + if os.path.isdir(original_arg): + try: + relpaths = list_relative_files(original_arg) + except UnreadableEntries as exc: + sys.exit(f"fastmd5: {exc}") + for relpath in relpaths: + full_path = os.path.join(original_arg, relpath) + enqueue_file_chunks(full_path, full_path, CHUNK_SIZE, workpile) + else: + enqueue_file_chunks(original_arg, original_arg, CHUNK_SIZE, workpile) # Prepare aligned buffers for each thread alignment = 2 * 1024 * 1024 # 2 MiB @@ -100,14 +128,20 @@ def main(): # Launch worker threads print_lock = threading.Lock() + errors = [] # list.append is atomic, so no lock needed threads = [] for i in range(args.num_threads): - t = threading.Thread(target=checksum_worker, args=(i, thread_buffers[i], workpile, args.buffer_size, print_lock)) + t = threading.Thread(target=checksum_worker, args=(i, thread_buffers[i], workpile, args.buffer_size, print_lock, errors)) t.start() threads.append(t) for t in threads: t.join() + if errors: + print(f"fastmd5: {len(errors)} chunk(s) failed; output is incomplete", + file=sys.stderr) + sys.exit(1) + if __name__ == '__main__': main() diff --git a/mlperf_common/fileio/__init__.py b/mlperf_common/fileio/__init__.py new file mode 100644 index 0000000..b2c99d7 --- /dev/null +++ b/mlperf_common/fileio/__init__.py @@ -0,0 +1,24 @@ +# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Fast file I/O: O_DIRECT primitives and copy planning. + + direct_io O_DIRECT pread/pwrite with aligned buffers (fastcp, fastmd5) + copyplan source-tree walk and src->dst mapping (fastcp, fastmd5) + +Both are dependency-free -- stdlib only, no torch, no numpy -- so that the +single-node client scripts can import them without pulling in a training stack. +Keep it that way: anything needing torch belongs in a module of its own, not +here. +""" diff --git a/mlperf_common/fileio/copyplan.py b/mlperf_common/fileio/copyplan.py new file mode 100644 index 0000000..8780636 --- /dev/null +++ b/mlperf_common/fileio/copyplan.py @@ -0,0 +1,274 @@ +#!/usr/bin/env python3 + +# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Source-tree walk and src->dst mapping shared by fastcp and fastmd5. + +Both take the same cp/rsync-shaped arguments and need the same answer: given +SOURCE(s) and a DEST, which files get copied where. Keeping that in one place +means a copy and a checksum of that copy can never disagree about what +"copy this directory" means. + +The argument rules live here too, in validate_copy_args, rather than in each +tool's CLI. They are not decoration: the mapping below is only well defined +once they hold, and a caller that skipped them used to get a silently wrong +plan instead of an error. plan_copy_operations therefore applies them itself, +and the CLIs call them early only to fail before doing any other setup. + +The semantics are GNU cp's, which is what fastcp set out to match: + + cp -r src newdir newdir absent -> create it, src's *contents* inside + cp -r src existingdir -> existingdir/src/... + cp a b existingdir -> existingdir/a, existingdir/b + cp a b newdir newdir absent -> error, target is not a directory + cp -r src file -> error, cannot overwrite non-directory +""" + +import collections +import os + +__all__ = ["CopyArgumentError", "UnreadableEntries", "list_relative_files", + "plan_copy_operations", "validate_copy_args"] + +# How many bad paths to name before summarising the rest. +_MAX_REPORTED = 20 + + +class UnreadableEntries(Exception): + """Some entries under a source tree could not be stat'd. + + Raised instead of letting the first bad entry throw, so a caller learns the + full extent in one pass rather than fixing them one job at a time. The + usual cause is a dangling symlink: os.walk lists it among the filenames + because it is not a directory, and stat then follows it to nothing. + """ + + def __init__(self, entries): + self.entries = list(entries) + super().__init__(self._describe()) + + def _describe(self): + count = len(self.entries) + noun = "entry" if count == 1 else "entries" + lines = [f"{count} unreadable {noun}:"] + for path, reason in self.entries[:_MAX_REPORTED]: + lines.append(f" {path}: {reason}") + if count > _MAX_REPORTED: + lines.append(f" ... and {count - _MAX_REPORTED} more") + return "\n".join(lines) + + +TreeEntries = collections.namedtuple("TreeEntries", ("files", "directories")) + +CopyPlan = collections.namedtuple("CopyPlan", ("files", "directories")) + + +class CopyArgumentError(Exception): + """The SOURCE/DEST combination is not one cp would accept. + + Carries a message in cp's wording, without a program name; callers prefix + their own, as cp does. + """ + + +def validate_copy_args(sources, destination, recursive=True, into_directory=False): + """Check a SOURCE(s)/DEST combination, raising CopyArgumentError if bad. + + `recursive` is the caller's -r flag and `into_directory` its -t. Both only + tighten the check, so plan_copy_operations can re-apply this with the + defaults on arguments a CLI has already accepted and never disagree. + """ + if not sources: + raise CopyArgumentError( + f"missing destination file operand after '{destination}'") + + for src in sources: + if not os.path.exists(src): + raise CopyArgumentError(f"cannot stat '{src}': No such file or directory") + if os.path.isdir(src) and not recursive: + raise CopyArgumentError(f"-r not specified; omitting directory '{src}'") + + if os.path.isdir(destination): + return + + # Not a directory, so there is nowhere to put a second source, and no + # basename to place anything under. -t says the destination is meant to be + # a directory to copy into, so it has to already be one. + if into_directory or len(sources) > 1: + if os.path.exists(destination): + raise CopyArgumentError(f"target '{destination}' is not a directory") + raise CopyArgumentError(f"target '{destination}': No such file or directory") + + if os.path.exists(destination) and os.path.isdir(sources[0]): + raise CopyArgumentError( + f"cannot overwrite non-directory '{destination}' " + f"with directory '{sources[0]}'") + + +def _stat_or_problem(path, problems): + """stat(), following symlinks. On failure record it and return None.""" + try: + return os.stat(path) + except OSError as exc: + problems.append((path, exc.strerror or str(exc))) + return None + + +def list_relative_files(root): + """Return all file paths under 'root' as relative paths, sorted alphabetically. + + os.walk(followlinks=True) allows following symlinked directories, and + symlinks to files appear in filenames, so both are dereferenced and copied + as content rather than recreated as links. + + Raises UnreadableEntries if anything under root cannot be stat'd, or if a + directory under it cannot be listed, reporting every such entry rather than + dying on the first one. + + Nothing here detects symlink cycles (`latest -> .`, `up -> ..`), but they + do not hang: the kernel allows 40 symlink traversals per path resolution, + so the descent stops around 40 levels down, in milliseconds. What that + looks like is worth knowing, because os.walk hides it -- it wraps + entry.is_dir() in try/except OSError and treats a failure as "not a + directory", so the un-openable link is reclassified as a *file* and the + walk reports no error at all, not even through onerror. The ELOOP then + surfaces here, from the stat, as an unreadable entry, and the copy is + refused. + + That refusal is the safe outcome, so do NOT "fix" this by skipping the + unreadable entry: the walk enumerates the files under the cycle once per + level on the way down, so tolerating the ELOOP would turn a loud refusal + into ~40 redundant copies of everything beneath it. A real fix means + detecting the cycle -- a depth-first walk tracking visited (st_dev, st_ino) + -- which also drops the duplicates. Not worth it until a dataset actually + contains such a link. + """ + return list_relative_entries(root).files + + +def list_relative_entries(root): + """Return a TreeEntries(files, directories) of paths relative to `root`. + + One walk yields both. Directories matter because a copy has to recreate + an empty one -- there is no file under it to imply it -- and walking a + second time to find them would double the metadata load on the shared + filesystem, which for a dataset of millions of files is the expensive part. + """ + file_list = [] + dir_list = [] + problems = [] + + def unlistable(exc): + # The one unreadable-entry case with no entry to report: files under a + # directory we cannot list never reach the walk at all, so there is + # nothing to stat and nothing to collect. os.walk's default is to + # swallow this and carry on, which makes an unlistable subtree + # indistinguishable from an empty one -- a partial copy that exits 0, + # and, since fastmd5 enumerates through here too, a partial checksum + # that agrees with it. + problems.append((exc.filename or root, exc.strerror or str(exc))) + + for dirpath, dirnames, filenames in os.walk(root, followlinks=True, onerror=unlistable): + if dirpath != root: + dir_list.append(os.path.relpath(dirpath, root)) + for fname in filenames: + full_path = os.path.join(dirpath, fname) + if _stat_or_problem(full_path, problems) is not None: + file_list.append(os.path.relpath(full_path, root)) + dirnames.sort() + if problems: + raise UnreadableEntries(problems) + return TreeEntries(sorted(file_list), sorted(dir_list)) + + +def plan_copy_operations(sources, destination): + """Return a CopyPlan: (src_abs, dst_abs, size_bytes) jobs, and dirs to create. + + If `destination` is an existing directory each source is placed inside it + under its own basename (recursing into directories). Otherwise there is + exactly one source and `destination` names it directly: a file is copied to + that name, and a directory has its *contents* copied in under it, which is + what `cp -r src newdir` does when newdir does not yet exist. `files` is + sorted by destination path so that any two tools planning the same copy + walk it in the same order. + + `directories` holds every destination directory the copy needs, shallowest + first, and is not derivable from `files`: an empty source directory has no + file under it to imply its parent, but `cp -r` still creates it. Create + them before copying rather than calling makedirs per file. + + Raises CopyArgumentError if the arguments are not a combination cp would + accept, and UnreadableEntries if any source cannot be stat'd or any + directory under it cannot be listed, reporting all of them together. + """ + validate_copy_args(sources, destination) + + file_jobs = [] + dst_dirs = set() + problems = [] + dst_root = os.path.abspath(destination) + + def add(src_abs, dst_abs): + st = _stat_or_problem(src_abs, problems) + if st is not None: + file_jobs.append((src_abs, dst_abs, st.st_size)) + # Every file implies its parent. Collecting them here means the + # caller can create the whole tree up front instead of calling + # makedirs once per file. + dst_dirs.add(os.path.dirname(dst_abs)) + + if not os.path.isdir(dst_root): + # Validation has established there is exactly one source and, if it is + # a directory, that nothing is in the way. `destination` *is* the + # copy, so a directory's contents go directly under it -- no basename + # level, which is where cp -r and cp differ. + src_abs = os.path.abspath(sources[0]) + if os.path.isdir(src_abs): + try: + entries = list_relative_entries(src_abs) + except UnreadableEntries as exc: + problems.extend(exc.entries) + else: + dst_dirs.add(dst_root) + for relpath in entries.files: + add(os.path.join(src_abs, relpath), os.path.join(dst_root, relpath)) + for relpath in entries.directories: + dst_dirs.add(os.path.join(dst_root, relpath)) + else: + add(src_abs, dst_root) + else: + for src in sources: + src_abs = os.path.abspath(src) + base = os.path.basename(src.rstrip("/")) + if os.path.isdir(src): + try: + entries = list_relative_entries(src) + except UnreadableEntries as exc: + problems.extend(exc.entries) + continue + dst_dirs.add(os.path.join(dst_root, base)) + for relpath in entries.files: + add(os.path.join(src_abs, relpath), + os.path.join(dst_root, base, relpath)) + for relpath in entries.directories: + dst_dirs.add(os.path.join(dst_root, base, relpath)) + else: + add(src_abs, os.path.join(dst_root, base)) + + if problems: + raise UnreadableEntries(problems) + # Shallowest first, so creating them in order never needs a parent that + # does not exist yet. + return CopyPlan(sorted(file_jobs, key=lambda job: job[1]), sorted(dst_dirs)) diff --git a/mlperf_common/fileio/direct_io.py b/mlperf_common/fileio/direct_io.py new file mode 100644 index 0000000..99424ed --- /dev/null +++ b/mlperf_common/fileio/direct_io.py @@ -0,0 +1,150 @@ +#!/usr/bin/env python3 + +# Copyright (c) 2024-2025, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import os +import ctypes + +def print_memoryview(mv): + """Print the address and size of a memoryview.""" + address = ctypes.addressof(ctypes.c_char.from_buffer(mv)) + size = mv.nbytes + print(f"Address: {hex(address)}, Size: {hex(size)}") + +def allocate_aligned_buffers(buffer_size, alignment, num_bufs): + """Allocate a sequence of aligned buffers. + + Args: + buffer_size (int): Size of each buffer in bytes. + alignment (int): Alignment boundary in bytes. + num_bufs (int): Number of buffers to allocate. + + Returns: + list[memoryview]: A list of memoryview objects, each representing an aligned buffer + raw_buf: the ctypes raw string_buffer. This *must* be kept to avoid the + gc prematurely collecting the buffers. + """ + assert buffer_size % alignment == 0, "buffer_size must be a multiple of alignment" + + # Allocate a single large buffer that can accommodate all buffers with proper alignment + total_size = (num_bufs * buffer_size) + alignment + raw_buf = ctypes.create_string_buffer(total_size) + base_address = ctypes.addressof(raw_buf) + + # Calculate the offset to align the first buffer + first_offset = (alignment - (base_address % alignment)) % alignment + aligned_base_address = base_address + first_offset + + # Create memoryviews for each buffer + buffers = [] + for i in range(num_bufs): + buf_address = aligned_base_address + i * buffer_size + aligned_buf = (ctypes.c_char * buffer_size).from_address(buf_address) + mv = memoryview(aligned_buf) + # print_memoryview(mv) + buffers.append(mv) + return buffers, raw_buf + +def round_up(value, multiple): + """Round up `value` to the next multiple of `multiple`.""" + if multiple == 0: + raise ValueError("Multiple must be greater than 0.") + return ((value + multiple - 1) // multiple) * multiple + +def pread(fd, aligned_memview, count, offset, fs_block_size, thread_id): + """Perform a direct pread. + + Returns: + number of bytes read. Unlike Posix pread this routine handles + interrupts, so return value should always be equal to count unless + there is an unrecoverable error, in which case it throws rather than returns + + Required: `count` <= aligned_memview size + """ + padded_count = round_up(count, fs_block_size) + assert padded_count <= aligned_memview.nbytes, "memview too small for requested read" + + while True: + try: + bytes_read = os.preadv(fd, [aligned_memview[:padded_count]], offset) + + if bytes_read == count: + # Expected case: all requested bytes were read. + return bytes_read + + if bytes_read > 0: + # Retry case: partial read, retry the entire read to maintain alignment restrictions + assert bytes_read < count, "bytes_read cannot exceed count" + continue + + if bytes_read == 0: + # Unexpected EOF + raise RuntimeError(f"Unexpected EOF encountered at offset {offset}") + + if bytes_read < 0: + raise OSError("preadv returned a negative value") + + except InterruptedError: + # Retry on EINTR + continue + except OSError as e: + # Non-retriable errors + if e.errno == 14: # Errno 14 corresponds to "Bad address" + raise RuntimeError(f"preadv failed with error: {e}, this usually means out of memory") + raise RuntimeError(f"preadv failed with error: {e}") + +def pwrite(fd, aligned_memview, count, offset, fs_block_size): + """Perform a direct pwrite + + Returns: + number of bytes written. Unlike Posix pwrite this routine handles + interrupts, so return value should always be equal to count unless + there is an unrecoverable error, in which case it throws rather than + returns + + Required: `count <= aligned_memview size + """ + padded_count = round_up(count, fs_block_size) + assert padded_count <= aligned_memview.nbytes, "memview too small for requested write" + while True: + try: + bytes_written = os.pwritev(fd, [aligned_memview[:padded_count]], offset) + + if bytes_written == padded_count: + # Expected case: all requested bytes were written. + # in the case of the last block in the file we may have padded up, so return + # the bytes _requested_ rather than the padded count + return count + + if bytes_written > 0: + # Retry case: partial write, retry the entire write to maintain alignment restrictions + assert bytes_written < count, "bytes_written cannot exceed count" + continue + + if bytes_written == 0: + # Unexpected failure to write + raise RuntimeError(f"Unexpected failure to write at offset {offset}") + + if bytes_written < 0: + raise OSError("pwritev returned a negative value") + + except InterruptedError: + # Retry on EINTR + continue + except OSError as e: + # Non-retriable errors + if e.errno == 14: # Errno 14 corresponds to "Bad address" + raise RuntimeError(f"pwritev failed with error: {e}, this usually means out of memory") + raise RuntimeError(f"pwritev failed with error: {e}") diff --git a/tests/README.md b/tests/README.md new file mode 100644 index 0000000..d165ebe --- /dev/null +++ b/tests/README.md @@ -0,0 +1,41 @@ +# tests + +Stdlib only, no pytest, no numpy, no GPU. Run them with: + + python3 tests/run_tests.py + +or run any one directly: + + python3 tests/test_copyplan.py + +Each test runs in its own interpreter, so nothing one test leaves in +`sys.modules` can reach another. + +## What is here + +| file | covers | +| --- | --- | +| `test_copyplan.py` | tree walk, src→dst mapping, `cp` argument semantics, and the refusal to plan an unreadable tree | + +`mlperf_common.fileio.copyplan` is stdlib-only, so the test imports the real +module straight out of the checkout. There is nothing to stub. + +## What these do NOT cover + +`direct_io` has no test here. Its `pread`/`pwrite` retry loops and block-size +padding are exercised only indirectly, by running `fastcp` and `fastmd5` against +a real filesystem — and O_DIRECT's alignment demands are not reproducible +against a temp directory on every filesystem. That is a real gap, not a +deliberate omission. + +Nothing here covers throughput, Lustre, or node-local NVMe. + +## Why `copyplan` is worth testing at all + +`fastcp` and `fastmd5` both enumerate a source tree through `list_relative_files`, +so the two agree about which files exist only because they share this code. If +the walk silently skipped a subtree — which is `os.walk`'s default when it +cannot list a directory — a partial copy would exit 0 and then be "verified" by +a checksum run that skipped exactly the same files. `plan_copy_operations` +raises `UnreadableEntries` listing every bad entry instead, and +`check_unlistable_directory` is what holds that behaviour in place. diff --git a/tests/run_tests.py b/tests/run_tests.py new file mode 100644 index 0000000..d06e084 --- /dev/null +++ b/tests/run_tests.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 + +# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Run every test_*.py here in its own interpreter; exit nonzero if any failed. + +Separate processes so that no test can be influenced by what an earlier one left +in sys.modules or patched onto an imported module. Nothing here needs a GPU, a +launcher, or any package beyond the standard library. +""" + +import glob +import os +import subprocess +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) + + +def main(): + failed = [] + for path in sorted(glob.glob(os.path.join(HERE, "test_*.py"))): + name = os.path.basename(path) + completed = subprocess.run([sys.executable, path], cwd=HERE, check=False) + if completed.returncode != 0: + failed.append(name) + print() + if failed: + print(f"FAILED: {', '.join(failed)}") + return 1 + print("all tests passed") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_copyplan.py b/tests/test_copyplan.py new file mode 100644 index 0000000..e896bb4 --- /dev/null +++ b/tests/test_copyplan.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 + +# Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""copyplan's walk, mapping, and its refusal to plan an unreadable tree. + +fastcp and fastmd5 both enumerate through here, so a disagreement about which +files a copy covers would mean checksumming a different set of files than was +copied. +""" + +import os +import shutil +import sys +import tempfile + +# copyplan is stdlib-only, so it imports straight out of the checkout with no +# stubbing: nothing here needs torch. +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from mlperf_common.fileio import copyplan # noqa: E402 + +results = [] + + +def check(name, condition, detail=""): + results.append((name, condition, detail)) + + +def build_tree(root): + """A tree with each kind of entry the walk has to have an answer for.""" + os.makedirs(os.path.join(root, "sub")) + os.makedirs(os.path.join(root, "realdir")) + for path in ("a.bin", "sub/b.bin", "realdir/c.bin"): + with open(os.path.join(root, path), "wb") as handle: + handle.write(b"x" * 10) + os.symlink("a.bin", os.path.join(root, "link_to_file")) + os.symlink("realdir", os.path.join(root, "link_to_dir")) + + +def check_cp_semantics(root): + """The SOURCE/DEST rules, against what GNU cp actually does. + + fastcp was written to match cp, and fastmd5 inherits the same enumeration + through this module, so these are the cases where "what does copy mean here" + has to have exactly one answer. The rows marked ERROR are the ones cp refuses. + """ + base = os.path.join(root, "sem") + src = os.path.join(base, "src") + os.makedirs(os.path.join(src, "sub")) + for path in ("a.bin", "sub/b.bin"): + with open(os.path.join(src, path), "wb") as handle: + handle.write(b"x" * 10) + for name in ("f1", "f2"): + with open(os.path.join(base, name), "wb") as handle: + handle.write(b"y" * 10) + existing = os.path.join(base, "existingdir") + os.makedirs(existing) + + f1, f2 = os.path.join(base, "f1"), os.path.join(base, "f2") + absent = os.path.join(base, "newdir") + + def plan(sources, destination, **kwargs): + plan = copyplan.plan_copy_operations(sources, destination, **kwargs) + return sorted(os.path.relpath(dst, base) for _, dst, _ in plan.files) + + # cp -r src newdir, newdir absent: the contents land directly in newdir, + # with no basename level. This is the case that used to plan a single + # 4 KiB "copy the directory as a file" job. + check("cp -r src newdir puts contents directly under newdir", + plan([src], absent) == ["newdir/a.bin", "newdir/sub/b.bin"], + f"got {plan([src], absent)}") + + # cp -r src existingdir: basename level is kept. + check("cp -r src existingdir keeps the basename level", + plan([src], existing) == ["existingdir/src/a.bin", + "existingdir/src/sub/b.bin"], + f"got {plan([src], existing)}") + + check("cp a b existingdir places both under it", + plan([f1, f2], existing) == ["existingdir/f1", "existingdir/f2"]) + + check("cp a newname renames a single file", + plan([f1], os.path.join(base, "renamed.bin")) == ["renamed.bin"]) + + for name, sources, destination, kwargs in ( + ("cp a b newdir is refused when newdir is absent", [f1, f2], absent, {}), + ("cp -r src file is refused", [src], f1, {}), + ("cp -t newdir src is refused when newdir is absent", + [src], absent, {"into_directory": True}), + ("a directory source without -r is refused", + [src], existing, {"recursive": False}), + ): + try: + copyplan.validate_copy_args(sources, destination, **kwargs) + check(name, False, "no exception raised") + except copyplan.CopyArgumentError: + check(name, True) + + # The planner must not depend on its caller having validated first. + try: + copyplan.plan_copy_operations([f1, f2], absent) + check("planning applies the rules itself", False, "no exception raised") + except copyplan.CopyArgumentError: + check("planning applies the rules itself", True) + + +def check_unlistable_directory(root): + """A subtree we cannot list must be reported, not quietly left out. + + This is the one unreadable-entry case with no entry to report: the files + under an unlistable directory never reach the walk, so there is nothing to + stat and nothing to collect. os.walk's default is to swallow the error and + carry on, which makes such a subtree indistinguishable from an empty one -- + a partial copy that exits 0, and a partial checksum that agrees with it. + """ + source = os.path.join(root, "blocked") + os.makedirs(os.path.join(source, "readable")) + os.makedirs(os.path.join(source, "secret")) + for path in ("readable/good.bin", "secret/hidden.bin"): + with open(os.path.join(source, path), "wb") as handle: + handle.write(b"x" * 10) + os.chmod(os.path.join(source, "secret"), 0o000) + + try: + # Root ignores the mode bits, so the premise would not hold and the + # check below would pass without testing anything. + try: + os.listdir(os.path.join(source, "secret")) + except PermissionError: + pass + else: + check("unlistable directory is reported", True, + "SKIPPED: this user can list a 0o000 directory") + return + + try: + found = copyplan.list_relative_files(source) + check("unlistable directory is reported", False, + f"no exception; returned {found}") + except copyplan.UnreadableEntries as exc: + check("unlistable directory is reported", True) + check("the message names the directory it could not list", + "secret" in str(exc), f"got {exc}") + + # Same hole one level up: the root of the walk itself. + os.chmod(source, 0o000) + try: + found = copyplan.list_relative_files(source) + check("unlistable root is reported", False, + f"no exception; returned {found}") + except copyplan.UnreadableEntries: + check("unlistable root is reported", True) + finally: + # Otherwise rmtree cannot clean up after us. + os.chmod(source, 0o755) + os.chmod(os.path.join(source, "secret"), 0o755) + + +def check_empty_directories(root): + """Directories with no files under them still have to be created. + + Nothing in `files` implies them -- that is the whole point -- so a planner + that reported only files produced a destination tree quietly missing every + empty directory, and for a wholly empty source produced no jobs at all and + never created the destination. cp -r creates it, and exits 0 either way, + so a caller checking the exit status could not tell. + """ + base = os.path.join(root, "emptydirs") + source = os.path.join(base, "src") + os.makedirs(os.path.join(source, "sub", "deep", "leaf")) + os.makedirs(os.path.join(source, "lonely")) + with open(os.path.join(source, "sub", "f.txt"), "w") as fh: + fh.write("x") + + plan = copyplan.plan_copy_operations([source], os.path.join(base, "dst")) + rel = {os.path.relpath(d, os.path.join(base, "dst")) for d in plan.directories} + check("empty directories appear in the plan", + {"lonely", os.path.join("sub", "deep"), os.path.join("sub", "deep", "leaf")} <= rel, + f"got {sorted(rel)}") + check("a directory holding a file is planned too", "sub" in rel) + check("directories are shallowest first", + list(plan.directories) == sorted(plan.directories)) + + # A source with no files at all still has to produce the destination. + empty_src = os.path.join(base, "wholly-empty") + os.makedirs(empty_src) + empty_dst = os.path.join(base, "newdir") + plan = copyplan.plan_copy_operations([empty_src], empty_dst) + check("an empty source plans no file jobs", plan.files == []) + check("an empty source still plans its destination directory", + empty_dst in plan.directories, f"got {plan.directories}") + + # list_relative_entries shares one walk; the files half must not drift + # from what list_relative_files reports. + entries = copyplan.list_relative_entries(source) + check("list_relative_files agrees with list_relative_entries", + entries.files == copyplan.list_relative_files(source)) + + +def main(): + root = tempfile.mkdtemp(prefix="copyplan-") + try: + source = os.path.join(root, "src") + build_tree(source) + + found = copyplan.list_relative_files(source) + check("symlink to a file is followed and listed", "link_to_file" in found) + check("symlink to a directory is descended into", + "link_to_dir/c.bin" in found, f"got {found}") + check("the symlinked directory is not itself listed as a file", + "link_to_dir" not in found) + check("result is sorted", found == sorted(found)) + + destination = os.path.join(root, "dst") + os.makedirs(destination) + plan = copyplan.plan_copy_operations([source], destination) + check("every planned file lands under dest//", + all(dst.startswith(os.path.join(destination, "src") + os.sep) + for _, dst, _ in plan.files)) + check("plan is sorted by destination", + [j[1] for j in plan.files] == sorted(j[1] for j in plan.files)) + check("sizes come back with the plan", + all(size == 10 for _, _, size in plan.files)) + + single = copyplan.plan_copy_operations( + [os.path.join(source, "a.bin")], os.path.join(destination, "renamed.bin")) + check("file-to-file copy keeps the given destination name", + len(single.files) == 1 and single.files[0][1].endswith("renamed.bin")) + + # Dangling symlinks: os.walk lists them among the filenames because they + # are not directories, and stat then follows them to nothing. + os.symlink("/nowhere", os.path.join(source, "dead1")) + os.symlink("/also/nowhere", os.path.join(source, "sub", "dead2")) + try: + copyplan.list_relative_files(source) + check("dangling symlinks are refused", False, "no exception raised") + except copyplan.UnreadableEntries as exc: + check("dangling symlinks are refused", True) + check("every bad entry is reported, not just the first", + len(exc.entries) == 2, f"reported {len(exc.entries)}") + check("the message names the bad paths", + "dead1" in str(exc) and "dead2" in str(exc)) + try: + copyplan.plan_copy_operations([source], destination) + check("planning refuses an unreadable tree", False, "no exception raised") + except copyplan.UnreadableEntries: + check("planning refuses an unreadable tree", True) + + check_cp_semantics(root) + check_unlistable_directory(root) + check_empty_directories(root) + finally: + shutil.rmtree(root, ignore_errors=True) + + failures = [r for r in results if not r[1]] + for name, _, detail in failures: + print(f" FAIL {name}{': ' + detail if detail else ''}") + print(f"test_copyplan: {len(results)} checks, {len(failures)} failures") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main())