From e7b02cb6e554f8991d7f886eda28f221b675cd12 Mon Sep 17 00:00:00 2001 From: Matt Frank Date: Tue, 1 Sep 2026 09:45:41 -0500 Subject: [PATCH 1/8] fileio: add direct_io and copyplan, extracted from fastcp and fastmd5 fastcp and fastmd5 each carried their own copy of the aligned-buffer I/O and the source-tree walk, so a fix to one did not reach the other. Move both into mlperf_common.fileio, where anything else that needs them can import them too. The package is deliberately stdlib-only -- no torch, no numpy -- so that the single-node client scripts can use it without pulling in a training stack. copyplan.plan_copy_operations reports every unreadable entry rather than dying on the first, and refuses a tree it cannot fully list. os.walk's default is to swallow that and carry on, which makes an unlistable subtree indistinguishable from an empty one: a partial copy that exits 0, and then a checksum run that skips the same files and agrees with it. Co-Authored-By: Claude Opus 5 --- mlperf_common/fileio/__init__.py | 24 +++ mlperf_common/fileio/copyplan.py | 233 ++++++++++++++++++++++++++++++ mlperf_common/fileio/direct_io.py | 150 +++++++++++++++++++ 3 files changed, 407 insertions(+) create mode 100644 mlperf_common/fileio/__init__.py create mode 100644 mlperf_common/fileio/copyplan.py create mode 100644 mlperf_common/fileio/direct_io.py 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..e3ee36a --- /dev/null +++ b/mlperf_common/fileio/copyplan.py @@ -0,0 +1,233 @@ +#!/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 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) + + +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. + """ + file_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, _, filenames in os.walk(root, followlinks=True, onerror=unlistable): + 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)) + if problems: + raise UnreadableEntries(problems) + return sorted(file_list) + + +def plan_copy_operations(sources, destination): + """Return a list of (src_abs, dst_abs, size_bytes) tuples to copy. + + 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. The result is + sorted by destination path so that every rank of a collective copy walks + the files in the same order. + + 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 = [] + 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)) + + 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: + for relpath in list_relative_files(src_abs): + add(os.path.join(src_abs, relpath), os.path.join(dst_root, relpath)) + except UnreadableEntries as exc: + problems.extend(exc.entries) + 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: + relpaths = list_relative_files(src) + except UnreadableEntries as exc: + problems.extend(exc.entries) + continue + for relpath in relpaths: + add(os.path.join(src_abs, relpath), + os.path.join(dst_root, base, relpath)) + else: + add(src_abs, os.path.join(dst_root, base)) + + if problems: + raise UnreadableEntries(problems) + return sorted(file_jobs, key=lambda job: job[1]) 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}") From ce8bfc9782c906942195d7e8edbbdf89a74ea9f2 Mon Sep 17 00:00:00 2001 From: Matt Frank Date: Tue, 1 Sep 2026 09:45:50 -0500 Subject: [PATCH 2/8] fastcp, fastmd5: share the extracted I/O and copy planner Both scripts now import direct_io and copyplan from mlperf_common.fileio instead of carrying private copies, so the two agree by construction about which files a copy covers -- which matters because fastmd5 is what verifies a tree fastcp wrote. Checksumming a different set of files than was copied would look exactly like success. client/direct_io.py becomes a compatibility shim re-exporting the package module, so `import direct_io` keeps working for anything that still expects a module sitting next to fastcp. The sys.path dance in each script covers both ways these are run: a pip install that puts the package and the scripts together, and a source tree with mlperf_common/ next to client/, which is the "deploy the repo to a shared filesystem and run in place" pattern. Copying a single script out of client/ on its own does not work; preserve that when touching the imports. fastmd5's output is byte-identical to the pre-refactor version. Co-Authored-By: Claude Opus 5 --- client/direct_io.py | 159 ++++++++------------------------------------ client/fastcp | 92 +++++++++---------------- client/fastmd5 | 98 ++++++++++++++++++--------- 3 files changed, 122 insertions(+), 227 deletions(-) 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..ee07c2c 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,7 +184,10 @@ if __name__ == "__main__": print(f"Sources: {args.sources}") print(f"Destination: {args.destination}") - file_jobs = plan_copy_operations(args) + try: + file_jobs = 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: 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() From f678751cfdd25296fa31052444fc1472e4c84af6 Mon Sep 17 00:00:00 2001 From: Matt Frank Date: Tue, 1 Sep 2026 09:45:58 -0500 Subject: [PATCH 3/8] tests: cover copy planning, and document the shared I/O layer test_copyplan.py checks the src->dst mapping against what GNU cp actually does, and checks that an unlistable directory makes planning fail loudly instead of quietly yielding a short file list. copyplan is stdlib-only, so the test imports it straight out of the checkout; there is nothing to stub. run_tests.py still runs each file in its own interpreter, so nothing one test leaves in sys.modules can reach another. direct_io has no test of its own. Its retry loops and block-size padding are exercised only indirectly, by running fastcp and fastmd5 against a real filesystem. tests/README.md says so rather than leaving the gap implied. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 114 ++++++++++++++++++++ tests/README.md | 41 ++++++++ tests/run_tests.py | 48 +++++++++ tests/test_copyplan.py | 233 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 436 insertions(+) create mode 100644 CLAUDE.md create mode 100644 tests/README.md create mode 100644 tests/run_tests.py create mode 100644 tests/test_copyplan.py diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..726bb31 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,114 @@ +# 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. + +`BUFFER_ALIGN = 2 MiB` (the huge-page size) is the shared alignment constant +across `fastcp` and `fastmd5`. + +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 scripts installed onto `PATH` by `setup.py`: `bindpcie` (NUMA/IB +affinity binding), `mgpurun`, `slurm2pytorch` (derives PyTorch rendezvous env +from SLURM), `fastcp` / `fastmd5` (threaded copy and per-GB checksum; `fastcp` opens with +O_DIRECT, `fastmd5` does not), +`dropcache`, plus log/telemetry shell helpers. + +**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/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..5d27386 --- /dev/null +++ b/tests/test_copyplan.py @@ -0,0 +1,233 @@ +#!/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): + jobs = copyplan.plan_copy_operations(sources, destination, **kwargs) + return sorted(os.path.relpath(dst, base) for _, dst, _ in jobs) + + # 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 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) + jobs = 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 jobs)) + check("plan is sorted by destination", + [j[1] for j in jobs] == sorted(j[1] for j in jobs)) + check("sizes come back with the plan", all(size == 10 for _, _, size in jobs)) + + 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) == 1 and single[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) + 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()) From 01f463498a05ed3094fc8301dafcfb5beeca589c Mon Sep 17 00:00:00 2001 From: Matt Frank Date: Tue, 1 Sep 2026 11:58:11 -0500 Subject: [PATCH 4/8] copyplan: plan the directories too, so empty ones get created plan_copy_operations reported only files, so a directory with nothing under it was never created: nothing in the file list implied it. `fastcp -r src newdir` where src holds an empty subdir produced a destination quietly missing it, and a wholly empty source produced no jobs at all, so newdir was never created and fastcp still exited 0. cp -r creates it in both cases. This was a regression the fileio extraction introduced. Before it, the same input failed loudly -- the old planner tried to copy the directory as a file and died with "Failed to open source file ... Invalid argument", exit 1 -- so a caller checking the exit status noticed. Exiting 0 with an incomplete tree is strictly worse, and it is the shape this repo cares about most: fastmd5 emits no line for a directory either, so a checksum comparison of the two trees also reports them identical. plan_copy_operations now returns a CopyPlan(files, directories). The directories come from the same os.walk that already yields dirnames and threw them away, so finding them costs no extra metadata work on the shared filesystem -- walking a second time would have doubled it. They are sorted shallowest first, so creating them in order never needs an absent parent, and they include each file's parent, which lets fastcp create the tree up front instead of calling makedirs once per file. list_relative_files keeps its signature and is now a wrapper over list_relative_entries; fastmd5 wants only the files and is unchanged. Co-Authored-By: Claude Opus 5 --- client/fastcp | 12 ++++-- mlperf_common/fileio/copyplan.py | 63 ++++++++++++++++++++++++++------ tests/test_copyplan.py | 57 +++++++++++++++++++++++++---- 3 files changed, 110 insertions(+), 22 deletions(-) diff --git a/client/fastcp b/client/fastcp index ee07c2c..bdc8194 100755 --- a/client/fastcp +++ b/client/fastcp @@ -185,16 +185,20 @@ if __name__ == "__main__": print(f"Destination: {args.destination}") try: - file_jobs = plan_copy_operations(args.sources, args.destination) + 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/mlperf_common/fileio/copyplan.py b/mlperf_common/fileio/copyplan.py index e3ee36a..8780636 100644 --- a/mlperf_common/fileio/copyplan.py +++ b/mlperf_common/fileio/copyplan.py @@ -36,6 +36,7 @@ cp -r src file -> error, cannot overwrite non-directory """ +import collections import os __all__ = ["CopyArgumentError", "UnreadableEntries", "list_relative_files", @@ -69,6 +70,11 @@ def _describe(self): 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. @@ -149,7 +155,19 @@ def list_relative_files(root): -- 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): @@ -162,26 +180,34 @@ def unlistable(exc): # that agrees with it. problems.append((exc.filename or root, exc.strerror or str(exc))) - for dirpath, _, filenames in os.walk(root, followlinks=True, onerror=unlistable): + 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 sorted(file_list) + return TreeEntries(sorted(file_list), sorted(dir_list)) def plan_copy_operations(sources, destination): - """Return a list of (src_abs, dst_abs, size_bytes) tuples to copy. + """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. The result is - sorted by destination path so that every rank of a collective copy walks - the files in the same order. + 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 @@ -190,6 +216,7 @@ def plan_copy_operations(sources, destination): validate_copy_args(sources, destination) file_jobs = [] + dst_dirs = set() problems = [] dst_root = os.path.abspath(destination) @@ -197,6 +224,10 @@ 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 @@ -206,10 +237,15 @@ def add(src_abs, dst_abs): src_abs = os.path.abspath(sources[0]) if os.path.isdir(src_abs): try: - for relpath in list_relative_files(src_abs): - add(os.path.join(src_abs, relpath), os.path.join(dst_root, relpath)) + 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: @@ -218,16 +254,21 @@ def add(src_abs, dst_abs): base = os.path.basename(src.rstrip("/")) if os.path.isdir(src): try: - relpaths = list_relative_files(src) + entries = list_relative_entries(src) except UnreadableEntries as exc: problems.extend(exc.entries) continue - for relpath in relpaths: + 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) - return sorted(file_jobs, key=lambda job: job[1]) + # 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/tests/test_copyplan.py b/tests/test_copyplan.py index 5d27386..e896bb4 100644 --- a/tests/test_copyplan.py +++ b/tests/test_copyplan.py @@ -73,8 +73,8 @@ def check_cp_semantics(root): absent = os.path.join(base, "newdir") def plan(sources, destination, **kwargs): - jobs = copyplan.plan_copy_operations(sources, destination, **kwargs) - return sorted(os.path.relpath(dst, base) for _, dst, _ in jobs) + 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 @@ -169,6 +169,47 @@ def check_unlistable_directory(root): 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: @@ -185,18 +226,19 @@ def main(): destination = os.path.join(root, "dst") os.makedirs(destination) - jobs = copyplan.plan_copy_operations([source], 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 jobs)) + for _, dst, _ in plan.files)) check("plan is sorted by destination", - [j[1] for j in jobs] == sorted(j[1] for j in jobs)) - check("sizes come back with the plan", all(size == 10 for _, _, size in jobs)) + [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) == 1 and single[0][1].endswith("renamed.bin")) + 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. @@ -219,6 +261,7 @@ def main(): check_cp_semantics(root) check_unlistable_directory(root) + check_empty_directories(root) finally: shutil.rmtree(root, ignore_errors=True) From f26bdebcd81942a4fd318d56a2dd315106434489 Mon Sep 17 00:00:00 2001 From: Matt Frank Date: Tue, 1 Sep 2026 11:58:21 -0500 Subject: [PATCH 5/8] docs: correct three claims the fileio split invalidated All three were introduced or preserved by trimming CLAUDE.md for this branch, and all three are checkable and were wrong. BUFFER_ALIGN was described as "the shared alignment constant across fastcp and fastmd5". It does not exist here: it was defined in datastage.py, which this branch does not carry, and grep matched nothing but the sentence asserting it. 2 MiB is hardcoded in four places across the two scripts. Say that instead, and note that direct_io is where it belongs. CLAUDE.md listed fastmd5 and dropcache among the scripts setup.py installs. It installs neither. That is worth stating rather than quietly correcting, because this branch makes fastmd5 hard-depend on the package while its own ImportError text advises "install mlperf-common" -- a remedy that cannot produce a fastmd5. copyplan's docstring still justified its sort order by "every rank of a collective copy", which has no consumer on this branch. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 726bb31..cba2cca 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -74,8 +74,9 @@ Anything needing torch belongs in a module of its own. partial copy that exits 0 — and then a checksum run that skips the same files and agrees with it. -`BUFFER_ALIGN = 2 MiB` (the huge-page size) is the shared alignment constant -across `fastcp` and `fastmd5`. +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 @@ -83,11 +84,21 @@ private variants. ### client/ and src/ -`client/` holds scripts installed onto `PATH` by `setup.py`: `bindpcie` (NUMA/IB -affinity binding), `mgpurun`, `slurm2pytorch` (derives PyTorch rendezvous env -from SLURM), `fastcp` / `fastmd5` (threaded copy and per-GB checksum; `fastcp` opens with -O_DIRECT, `fastmd5` does not), -`dropcache`, plus log/telemetry shell helpers. +`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. From 3114ec807eb46b76f9b5436c33ec4d17352fe816 Mon Sep 17 00:00:00 2001 From: Matt Frank Date: Wed, 2 Sep 2026 07:51:27 -0500 Subject: [PATCH 6/8] docs: record the defects this refactor introduced Three, all reproduced, all absent from main's pre-refactor fastcp: the friendly "cannot import mlperf_common" advisory is dead code because the unguarded import above it fails first; a leftover standalone direct_io.py in an install's bin/ outranks the new shim, which matters because setup.py installs that file and an upgrade leaves the old one behind; and test_copyplan's refusal rows assert only that some CopyArgumentError escaped, not which. Deliberately excludes the longer list of defects inherited from main -- direct_io.py is byte-identical there, and fastcp's destructive cases (same-file -f, src/.., -n 0) reproduce identically against main. Those predate this branch and are tracked elsewhere. Co-Authored-By: Claude Opus 5 --- FASTCP-BUGS.md | 82 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) create mode 100644 FASTCP-BUGS.md diff --git a/FASTCP-BUGS.md b/FASTCP-BUGS.md new file mode 100644 index 0000000..dd16731 --- /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, are not this branch's to fix, and are tracked +separately. + +`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. From 4b45440bff8c374af3b80901683518a4787c3a78 Mon Sep 17 00:00:00 2001 From: Matt Frank Date: Wed, 2 Sep 2026 09:49:55 -0500 Subject: [PATCH 7/8] docs: record the defects inherited from main Ten of them, severity-ordered, each with a runnable reproduction against `git show main:client/fastcp` rather than against this branch, because none is caused by the fileio extraction -- they all predate it, and direct_io.py is byte-identical across the two. Two lose data silently and exit 0: `fastcp -f a.bin .` truncates the source it was asked to copy, and `fastcp -r -f src/.. dst` writes every destination back outside DEST onto the sources themselves. A third pair compounds -- `-n 0` produces a full-size all-zero destination while `fastmd5 -n 0` produces no output, so the checksum comparison that would have caught it reports the two trees identical. This goes in the tree rather than a bug tracker because the repo has GitHub issues disabled. It is the record until there is somewhere better; the header says to move it and delete this file when that happens. Co-Authored-By: Claude Opus 5 --- FASTCP-BUGS.md | 4 +- MAIN-BUGS.md | 208 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 210 insertions(+), 2 deletions(-) create mode 100644 MAIN-BUGS.md diff --git a/FASTCP-BUGS.md b/FASTCP-BUGS.md index dd16731..1fff746 100644 --- a/FASTCP-BUGS.md +++ b/FASTCP-BUGS.md @@ -5,8 +5,8 @@ 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, are not this branch's to fix, and are tracked -separately. +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. diff --git a/MAIN-BUGS.md b/MAIN-BUGS.md new file mode 100644 index 0000000..8671b51 --- /dev/null +++ b/MAIN-BUGS.md @@ -0,0 +1,208 @@ +# 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. + +## 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. From f2609de847fd65b52dd865254050b7827189d004 Mon Sep 17 00:00:00 2001 From: Matt Frank Date: Wed, 2 Sep 2026 09:51:05 -0500 Subject: [PATCH 8/8] docs: correct 01f4634's claim about the empty-directory bug That commit message says the bug "was a regression the fileio extraction introduced". Only one of its two cases was. With DEST already a directory, main silently omits an empty subdirectory and exits 0, and the extraction carried that across unchanged -- a pre-existing defect, now recorded in MAIN-BUGS.md alongside the others main has. Only the wholly-empty-source case regressed, where main exited 1 and the extraction turned that into exit 0 with the destination never created. The fix in 01f4634 is correct for both; only its description was too broad. Correcting it here rather than force-pushing a reworded commit onto an open PR. Co-Authored-By: Claude Opus 5 --- MAIN-BUGS.md | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/MAIN-BUGS.md b/MAIN-BUGS.md index 8671b51..d206b73 100644 --- a/MAIN-BUGS.md +++ b/MAIN-BUGS.md @@ -125,6 +125,23 @@ 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.