Mfrank/py data staging - #41
Open
matthew-frank wants to merge 23 commits into
Open
Conversation
…aging
Adds datastage, a collective dataset stager. Every node in a training job
needs the same dataset, so copying it onto each node reads the whole thing
N times from shared storage and runs at the speed of the slowest reader.
Instead each rank reads a disjoint shard and NCCL fans the bytes out, so
the data crosses the fabric once per node.
The world is split into one process group per LOCAL_RANK, each holding one
rank per node and owning a contiguous 1/L slice of the file. Within a group
each rank reads a disjoint sub-shard and an all-gather assembles the slice
on every node. A node's ranks together write the whole file, so nothing is
exchanged or written twice within a node, and the concurrent all-gathers
drive every NIC without per-cluster transport tuning.
Groups the three related modules under mlperf_common/fileio rather than
adding them to the package root:
direct_io moved verbatim from client/, so package modules can import it
(setup.py installs it as a script, which is not importable)
copyplan source-tree walk and src->dst mapping, lifted out of fastcp so
fastcp and datastage cannot disagree about what a copy covers
datastage the stager and its cp/rsync-shaped CLI
Only datastage needs torch, so the single-node client scripts do not pull in
a training stack.
client/direct_io.py becomes a shim re-exporting the package module, keeping
`import direct_io` working for fastcp and fastmd5. Both resolve the package
either from an install or from a tree with mlperf_common/ next to client/,
and fail with an actionable message otherwise.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
fastmd5 walked the tree itself with Path.rglob and an `is_file() or is_symlink()` predicate. That predicate admitted entries that are not readable files: is_file() already follows symlinks, so the extra clause only added broken symlinks and symlinks to directories. A symlink to a directory then raised IsADirectoryError inside a worker thread, and an exception escaping a thread does not affect the process exit status -- so fastmd5 printed a traceback on stderr, skipped the file, and exited 0. Anything comparing two trees by parsing stdout saw a short but successful-looking result. Use copyplan.list_relative_files instead, so checksumming a staged copy enumerates the same files the copy was planned from. datastage dereferences symlinks, so a staged tree has different link structure from its source; the two walks have to agree for the comparison to mean anything. Also collect exceptions raised in worker threads and exit nonzero if there were any. fastmd5 still checksums everything it can and reports the partial result, it just no longer reports success. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…e first A dangling symlink is listed by os.walk among the filenames, because it is not a directory, and stat then follows it to nothing. Every caller reached that FileNotFoundError during enumeration, on the main thread, before any work started -- so one dead link aborted the whole run with a raw traceback and zero output, even when the rest of the tree was fine. For datastage that meant rank 0 dying during planning, after the allocation was granted, and only naming whichever bad entry os.walk happened to reach first. Collect them instead and raise UnreadableEntries naming all of them (capped at 20 with a count of the remainder). Behaviour is still to fail rather than to skip: a staging tool that quietly omits files is the worse failure. plan_copy_operations now stats each source once and reuses the result for the size, rather than calling getsize separately. fastcp and fastmd5 turn it into a clean message and a nonzero exit. datastage has to broadcast the failure from rank 0 instead of raising, since every other rank is already parked in the broadcast and would otherwise hang until the NCCL watchdog fired. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Two fixes to the staging pipeline. torch.cuda.synchronize() after each host/device copy is a device-wide barrier, so the host-to-device copy, the all-gather, the device-to-host copy and the NVMe writes ran one after another -- most of what the slot pipeline exists for. Record an event per copy instead. The collective is launched before waiting on the host-to-device event, since it is queued behind the copy on the same stream anyway, and the writers now wait on the device-to-host event themselves so the main thread returns to the next round while that copy is still in flight. --window named the assembled all-gather buffer, and divided it by the node count to get the per-rank read size. That silently inflated: the per-rank piece cannot go below the 2MiB O_DIRECT alignment, so at 2048 nodes a requested 2G window became 4G, and 8G at 4096. Take --buffer-size instead, the bytes each rank reads per round, exactly as fastcp's --buffer-size is what each thread reads. The window is then derived rather than requested, so nothing is silently exceeded, and it is rounded up to a 2MiB multiple the way fastcp does. Memory now visibly scales with the job, because an all-gather delivers the whole window to every participant: at 2048 nodes 8M gives a 16 GiB window. Rank 0 reports the footprint, and an oversized request is rejected up front with the largest workable value rather than failing in a CUDA OOM. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Copying the whole assembled window into pinned host memory before writing any of it made pinned host memory scale with the node count: two windows per rank, times DGXNGPU ranks per node, is 512 GiB per node at 2048 nodes with -b 32M. That was the binding limit, not GPU memory, which sat at 22%. The window is a concatenation of one segment per node, each bound for a different file offset, so it does not have to land in host memory whole. Copy it back a few segments at a time through a small fixed pool instead, and give the drain its own thread so the next round's collective overlaps the copy-back and the writes. Pinned host is now constant; the device holds two windows so the collective and the copy-back can overlap, which is the cheap resource here -- this is its own job step and exits before training. The device budget goes to 60% accordingly. Two bugs found by an end-to-end test of the pipeline with CUDA and NCCL stubbed out, both of which silently corrupted the staged file: _run_pipeline set the stop flag as soon as the main loop finished feeding the drain queue, and the drainer treated that as "abort" -- so every window still queued was dropped. The drain lags the main loop by design, so stop now means abort only, and the drainer is joined before teardown. Timing decided how much of a file survived. O_DIRECT writes are padded up to the block size, so the last write of a file runs past its end and nothing trimmed it back; any file whose size was not a multiple of the block size ended up padded with garbage. ftruncate after the barrier, where fastcp does the same after its copy. It has to be after the barrier because until then another rank may still be padding. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Stdlib only, no pytest and no numpy, so `python3 tests/run_tests.py` works anywhere. torch and torch.distributed are stubbed, which is what lets datastage run without a GPU or a job: tensors become memoryviews over ctypes buffers, CUDA events become no-ops, and the multi-node cases fake the all-gather by filling each node's segment with the bytes that node would have read. test_pipeline compares content rather than checking for exceptions, because the failure mode of a mishandled buffer handoff is a file of exactly the right length holding the wrong bytes. The two bugs fixed in the previous commit both pass a "did it throw?" check and both fail this one; reintroducing either makes it go red, which is the property worth keeping. test_layout checks the two invariants that corrupt data silently if broken: every byte claimed exactly once, and every interior segment boundary aligned, since direct_io pads writes up to the block size. tests/README.md records what this deliberately does not cover -- real NCCL, real CUDA events, pinned memory alignment, O_DIRECT, and anything about throughput. A green run says the arithmetic and the choreography are right, not that staging works on a cluster. tests/ has no __init__.py, so find_packages() does not pick it up. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
CLAUDE.md documents the two halves of the repo, the fileio dependency rule (only datastage may import torch), and the datastage decomposition, so a reader does not have to reconstruct the collective staging design from comments spread across 715 lines. It also corrects the record on srun --distribution=arbitrary. Topology's comments present support for it as a design invariant; it is not one. The requirement was a mistake carried over from the C version, and ranks are block- or cyclic-distributed in every supported launch. REVIEW-FINDINGS.md and .review-findings.json carry the results of a review of this branch: 15 verified findings, tiered, with repro steps and status boxes. They are a working record, not a deliverable -- delete them once the list is worked through. The headline item is a missing torch.cuda.set_device on the drainer thread, which records the copy-completion events on GPU 0 and lets writers race the D2H DMA on every rank but LOCAL_RANK 0. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
CUDA's current device is per host thread, and main() sets it on the main thread only, so the drain thread started by _run_pipeline was on device 0. The device-to-host copies still ran on the right device -- the tensors carry it -- but torch.cuda.Event binds to the calling thread's current device when recorded, so the copy-completion events landed on device 0's idle stream. Both synchronize() calls that depend on them then returned immediately: _write_one's, so the writer pool read a pinned chunk before the copy filled it, and the one guarding recv_free_q, so a device window was recycled while copies were still reading it and the next all-gather overwrote it. The result was a file of exactly the right length holding wrong bytes, no exception, on every rank except LOCAL_RANK 0 -- 7 of 8 slices per file on a DGX node. test_device.py guards it. The stubs now model a per-thread current device and have events remember which one they were recorded against, which is the part of this that is checkable without a GPU: it goes red on the unfixed code with 3 of 9 events on device 0, and green after. It does not check CUDA ordering semantics, so a multi-GPU confirmation is still worth doing. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The copy-back was issued on the default stream, the same one carrying the H2D copy and the all-gather, so it could not run beside them. non_blocking =True only promises the host will not wait; the copy still queues on the stream it was issued to, and ProcessGroupNCCL orders the collective against that stream in both directions. Every round therefore ran H2D, then the collective, then the copy-back, strictly in turn -- which made the second device window RECV_DEV_SLOTS allocates, and fights the memory budget check over, do nothing at all. The copies and their events now go to a dedicated stream, and the host-side assembled.synchronize() becomes drain_stream.wait_event(assembled), so the drain thread queues a round's copies while that round's collective is still in flight rather than blocking until it lands. That leaves the window-reuse handoff as the one piece of cross-stream safety with nothing but a host wait behind it: the next all-gather goes to the default stream, which has no ordering against drain_stream, so last_copy.synchronize() before recv_free_q.put is what stops it overwriting a window still being read. Commented in place. test_device.py now also requires that no drainer operation is issued on the default stream, and that a side stream waits on an event before its first copy -- the omission that would turn this from a throughput change into a race. Both confirmed red before the change. The speedup itself is unmeasured: the stubs make every copy instantaneous, and whether the GPU side is the bottleneck depends on Lustre and NVMe rates. Worth a profile on a real node. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
list_relative_files collected every entry it could not stat, but a directory it could not list produced nothing to collect: the files underneath never reach the walk, so there is no entry to stat and no problem to record. os.walk's default onerror swallows the error and carries on, which makes an unlistable subtree indistinguishable from an empty one. One dataset directory with a bad mode, an EIO, or a stale Lustre handle was therefore enough to stage a partial dataset and exit 0 -- and because fastmd5 enumerates through the same function, checksumming the staged tree omitted exactly the same files, so verification agreed. An unlistable root returned an empty list, equally silently. An onerror callback now appends to the same problems list, so these join the existing report and planning refuses the tree. This hole predates the fileio refactor: the same walk was in client/fastcp before copyplan existed. What changed is its reach, since fastcp, fastmd5 and datastage now share one enumeration and so share one blind spot. test_copyplan.py covers both the subtree and the root case, and verifies its own premise -- a user who can list a 0o000 directory gets a skip rather than a vacuous pass. Confirmed red first: the subtree case returned only the readable file, the root case returned nothing at all. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Planning moved into this module when fileio was extracted; the argument rules that make the planning well defined stayed behind in fastcp's CLI. plan_copy_operations documented its fallback as a fact -- "otherwise this is a single file-to-file copy" -- when it was really a precondition the caller had to establish. datastage was the first caller to arrive without it, and so planned one job and silently dropped sources[1:] when the destination was not an existing directory. The rules now live in validate_copy_args next to the mapping they govern, and plan_copy_operations applies them itself, so a caller cannot skip them. fastcp's destination block is deleted in favour of the shared call and datastage's parse_args gained it -- there rather than in build_plan, because parse_args runs on every rank before the process group exists, so a bad invocation has to kill the job uniformly instead of leaving rank 0 exiting while its peers block in a collective. That also fixes a case fastcp got wrong on its own. cp -r src newdir, with newdir absent, is legal and copies src's *contents* into newdir. fastcp correctly permitted it and the planner then treated the source directory as a 4 KiB file, so the invocation was broken in both tools -- and it is the natural way to stage onto empty node-local scratch. A directory source with a non-directory destination now maps contents directly under it, with no basename level. fastmd5 is unaffected: read-only, no destination. test_copyplan pins all six cases against what GNU cp actually does, and both CLIs were exercised end to end, error wording included. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The FIXME warned that os.walk(followlinks) does not protect against cycles and implied a custom depth-first traversal was needed to avoid running away. Measured, that is not what happens: the kernel allows 40 symlink traversals per path resolution, so a `up -> ..` tree stops at 82 directories and depth 81 in under 10 ms, and os.walk is iterative so there is no stack to blow. The path to the error is also not the obvious one, which is why it was worth writing down. os.walk 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 nothing at all -- not even through the onerror hook added in 8f66d5f. The ELOOP surfaces from the stat instead, as a single unreadable entry, and the copy is refused. The note now also records the trap: the walk lists the files under the cycle once per level on the way down, so tolerating the ELOOP rather than detecting the cycle would replace a loud refusal with ~40 redundant copies of everything beneath it -- turning the one loud failure in this area into a silent one. Comment only; no behaviour change. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
build_plan has rank 0 walk the tree and broadcast the answer, and the comment above it already said a failure has to be broadcast rather than raised, because every other rank is blocked in that broadcast. The catch was then narrowed to UnreadableEntries, which is the one failure the author went looking for. Anything else escaped: rank 0 unwound through destroy_process_group and exited while its peers waited for a message that would never come, so the job died on an NCCL watchdog timeout naming neither the file nor the reason, ten minutes of allocation later. The os.stat that collects st_mtime_ns re-stats files plan_copy_operations already stat'd and loses that race against anything modifying shared storage, which is the realistic trigger. Catching Exception fixes it. The bar is reaching the broadcast, not anticipating the cause -- narrowing the catch is what created the hang in the first place. There were two live paths, not one: the stat race, and CopyArgumentError, which became reachable here when 023586a moved the cp argument rules into plan_copy_operations. The new test caught the second on its own; it had only been written for the first. An operator now gets, on every rank together: cannot stage the source tree: FileNotFoundError: [Errno 2] No such file or directory: '/.../src/vanished.bin' test_buildplan.py drives rank 0's side and distinguishes a failure that was broadcast from one that escaped. torch.tensor joins the stubs so the success path is reachable. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
build_plan had every rank stat every file and all-reduce a mismatch count, to catch ranks seeing a different view of shared storage. That is W x F stats, issued as a synchronised burst right after the broadcast: 5.1M at 64 nodes and 10k files, 41M at 512 nodes, 1.6 billion at 2048 nodes and 100k files. At a generous 50k ops/s for a single MDT that is roughly fourteen minutes of pure metadata at 512 nodes before a byte of data moves, and at 2048 it is an outage for every other job on the filesystem, not just this one. The cost scaled with the file count and the detection power did not. What it looked for -- a stale handle, a failed mount, the wrong dataset at the same path -- is a property of a mount, and affects every file on that node identically, so the ten-thousandth stat says nothing the first did not. It was also imperfect, since size and mtime agree on same-size same-mtime content differences. Removed rather than reduced. Mount verification already belongs to mountcheck.py, once per job, where a sparse SHA256 fingerprint is both cheaper and stronger; errors of this kind have not been seen in six years on these clusters; and small file copies are to be parallelised so that each small file is handled, and its metadata touched, by a single rank -- which makes per-rank per-file metadata work the wrong shape whatever its cost. The reasoning is in build_plan's docstring so it does not get reinvented. torch.tensor, FakeCounter and dist.all_reduce leave the stubs with it, having no remaining user. Staging still opens every file on every rank, which is inherent to having W disjoint readers. This was pure addition on top of that. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
REVIEW-FINDINGS used W x F cold. W is datastage's own notation, from the module docstring, but F was invented in the F16 write-up and defined nowhere, so the numbers that justify removing the check were unreadable without guessing at them. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Topology gathered every rank's hostname and LOCAL_RANK to work out which ranks shared a node, so that --distribution=arbitrary would group correctly. That was never a real requirement: it came from the C version by mistake, and it could not have worked anyway, since MASTER_ADDR is the first node in the nodelist and rank 0 has to be there for the rendezvous to come up at all. Under slurm's default block distribution the layout is arithmetic. Node i holds ranks i*L through i*L+L-1, so a rank's node is RANK // L, its slot is RANK % L, and group_ranks[l] is [node*L + l for node in range(N)]. No collective, no hostnames, no socket import. The assumption is verified rather than assumed, and the check is free: slurm reports RANK as SLURM_PROCID and LOCAL_RANK as SLURM_LOCALID, two independently derived numbers that agree only under a block distribution. A modulo comparison on each rank rejects --distribution=cyclic and =arbitrary, and a ragged --ntasks-per-node, with no communication -- turning a launch this code cannot handle into an immediate error instead of groups whose all-gather assembles the right bytes in the wrong order. LOCAL_WORLD_SIZE is now required rather than defaulting to 1, which had turned a single 8-GPU node into eight single-rank "nodes". This also settles the new_group question. group_ranks[l] is ascending by construction, so new_group's internal sort is provably a no-op and each member's group position is its node_index, which is what the drainer assumes when it maps all-gather output position to a file offset. Noted where it matters, and checked. Topology needed all_gather_object, which the stubs could not sensibly fake, so nothing had ever constructed it. Now it needs only new_group. test_topology.py covers six layouts and the rejected launches, and was confirmed able to go red by reintroducing a mis-grouping. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
torch.distributed's env:// rendezvous wants RANK, WORLD_SIZE, MASTER_ADDR and MASTER_PORT, and by convention LOCAL_RANK and LOCAL_WORLD_SIZE beside them. Neither slurm nor mpirun sets those names, so something has to translate. client/slurm2pytorch does it in bash and execs the real command; this is the same translation in Python, so a program can do it for itself rather than being wrapped. The two are behaviourally compatible, and that is checked rather than asserted: fed the same pre-wrapper environment, the script and configure() produce identical values for all seven variables. A program launched under the wrapper finds them already set and takes them as given (source == "preset"), which is the case the test deliberately feeds contradictory slurm variables to catch. Two divergences, both refusing to proceed where the script would guess: slurm2pytorch defaults LOCAL_WORLD_SIZE to 1, but SLURM_NTASKS_PER_NODE is only set when --ntasks-per-node was actually passed. `srun -N2 -n16` has no source for it, and that default turns two 8-GPU nodes into sixteen single-rank "nodes" -- which reads as a valid job until the copy comes out wrong. We consult SLURM_TASKS_PER_NODE, which srun always sets, and refuse to guess on a multi-rank job. slurm2pytorch falls back to MASTER_ADDR=127.0.0.1, which its own comment says "will fail for multinode" -- as a rendezvous that hangs to the wall clock. We parse the address out of slurm's compressed nodelist instead, which is available in the container where `scontrol show hostnames` is not. Only the first name is needed, which is much less work than expanding the list, and it has to keep the zero padding: dgx[001-004] is dgx001, and dgx1 does not resolve. If that fails on a multi-node job we say so immediately, and name MLPERF_SLURM_FIRSTNODE as the fix. Beyond that it cross-checks what it is given, since everything here feeds a rendezvous and a wrong rendezvous does not error, it hangs: slurm's own NNODES x NTASKS_PER_NODE must equal NTASKS, slurm and mpirun must not disagree about which process this is, ranks must fall inside their worlds, and the world must divide evenly by the node size. Whether the resulting layout is the block distribution datastage needs stays Topology's business. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
main() now calls dist_env.configure() before anything else, so the rendezvous
variables are derived in-process and the bash wrapper is no longer part of the
launch. Running under it still works: the variables are then already set and
taken as given.
srun --ntasks-per-node=${DGXNGPU} ... python3 -m ...fileio.datastage ...
configure() runs on every rank before the process group exists and derives
from environment alone, so a bad launch fails the whole job identically
instead of leaving some ranks blocked in a collective -- the same reasoning
already written above the argument validation in parse_args.
Three details worth their comments:
The dry-run shortcut keyed on `"RANK" not in os.environ`, which this change
would have silently broken by always populating RANK. It now asks dist_env
which launcher it found and takes the shortcut only for source == "single".
Same behaviour, stated rather than inferred from a missing variable.
The rendezvous banner prints before init_process_group, not after. A wrong
MASTER_ADDR hangs inside the rendezvous, so a line printed afterwards never
appears -- and diagnosing that hang is the only reason to print it. Rank 0
always, every rank under NV_MLPERF_DEBUG, as slurm2pytorch's debug echo did.
set_device takes env.local_rank rather than os.environ.get("LOCAL_RANK", 0).
That silent default is the same class of guess this whole change removes.
Verified through the stub harness: with no launcher, --dry-run prints the plan
and returns 0; as `srun -N2 -n16` with no --ntasks-per-node, it exits naming
the flag instead of proceeding with sixteen imaginary single-rank nodes.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
datastage uses NCCL, through torch, purely as a network transport: the data is host-resident at both ends and nothing computes on it, so the trip through the GPU costs two PCIe crossings a host-side transfer would not. The stated reason for NCCL is that it drives every NIC without per-cluster tuning, which is an argument about NIC aggregation rather than about GPUs -- so if an MPI program can get one NIC per rank, the GPU may be droppable entirely. bindpcie --ib=single is supposed to provide exactly that, and has probably not been exercised in years. These are notes for a session on a real node. Read off the script, and solid: the ibdev list comes from ibv_devinfo order, the guard hard-exits when the device and GPU counts do not divide, and the mapping is ibdevs[local_rank * num_ibdevs / num_gpus]. Inferred from reading, and needing hardware to confirm: that mapping is index arithmetic with no topology query anywhere, despite --help promising a device "near its GPU", so locality holds only if ibv_devinfo happens to enumerate in GPU order -- which is the way this flag would be actively harmful rather than merely useless. OMPI_MCA_btl_openib_if_include targets a BTL removed in OpenMPI 5, so UCX_NET_DEVICES is doing all the work, with the port hardcoded to :1. The guard likely passes under enroot, which sets MELLANOX_VISIBLE_DEVICES, and likely would not bare, where storage NICs push the count past the GPU count. And every diagnostic in that block uses `2>&1` where `>&2` was meant, so the errors land on stdout -- one reason a broken --ib=single could go unnoticed. Also records how to observe this properly: per-device port_xmit_data counters rather than inferred bandwidth, and a warning that UCX_MAX_RNDV_RAILS defaults to 2, so an unbound rank may already use two NICs and only the aggregate across ranks is a fair comparison. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Three independent bugs, none of which reports itself as one. _chmod_parents walked up towards the destination root with `path = os.path.dirname(path)`, which stops making progress at "/". A destination root of "/" therefore satisfied the loop condition forever, chmod'ing the container root while every peer rank waited in the barrier on the next line, until the job hit its wall clock. Break when dirname(path) == path -- the general "walked off the top" condition, not a special case for "/". The temp file is ftruncate'd to the full source size before any data is written, and only the success path's rename ever removed it, so a failed copy left a near-full file on node-local scratch -- one per attempt, since the name carries the job id, until a resubmit loop against a flaky fabric filled the NVMe and attempts started failing with ENOSPC instead of the original error. stage_file now unlinks it best-effort on the way out. Any rank that raises does the unlink; peers still hold it open, but the inode survives until they close and the space returns when they die, and the rename that would have published it is not going to happen. Ranks parked in a collective never reach the handler, which is what the watchdog is for. The device-memory budget check computes device_bytes as piece * (2*nodes + 1) -- the +1 being the send buffer -- but suggested a replacement --buffer-size computed by dividing by 2*nodes, dropping it. The suggestion was therefore itself over budget, so an operator following the error's own advice got the identical error back, byte for byte, and burned another multi-node allocation. It also had no way to say "no size works here": the max() floor handed back 2 MiB even when 2 MiB could not fit either. Tests for all three, each confirmed able to go red. test_stager.py drives _chmod_parents against a stub with a call ceiling, so a runaway reports as a failure rather than hanging the suite, and checks the budget advice by the property that matters -- parse the suggested size out of the message, rebuild with it, require acceptance, and require that 2 MiB more would have been refused so the advice is not needlessly small. test_pipeline.py injects a failure into _run_pipeline and asserts no temp file survives, recording from inside the failure that one existed, so it cannot pass by never creating one. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The suite in tests/ fakes torch before importing datastage, so it never loads CUDA or NCCL and never talks to another rank. Running it on a GPU node changes nothing; launching it under srun just gets N independent copies of the same CPU test. Real NCCL, real CUDA events, pinned memory against a real block size, and O_DIRECT have never been exercised by anything in this repo. cluster-selftest.sh is that missing half. From inside an allocation it builds a dataset with the shapes that have historically broken things -- sizes either side of the 2 MiB alignment boundary where the write padding and the closing ftruncate interact, a zero-length file, two large enough to need several rounds at any plausible node count, 64 small ones, and a symlink -- stages it with a real multi-node srun, and checks on every node that what landed on node-local storage matches what was read from shared storage. It compares two things rather than one, which rehearsing it locally turned out to matter for: fastmd5 emits one line per GB-chunk, so a zero-length file produces no lines at all, and a destination missing empty.bin entirely compares equal on checksums alone. A size-and-path listing catches that, plus truncations and unexpected extra files. Verified by injecting both failure modes into a stand-in copy: the missing empty file shows up in the listing and a single flipped byte in the checksums, independently. It also asserts the things recently fixed stay fixed on real hardware: no .datastage.tmp.* files survive a successful run, and everything is mode 0777. mtime is deliberately not compared. datastage does set it from the source, but timestamp granularity varies by filesystem and a false failure on a first hardware run would cost more than the check is worth. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
refactor the fastcp program, take the core and work it into a new distributed copy program based on Vaino's c++ version.