feat(data-plane): track data-plane time, latency percentiles and byte volume

[ZhiyuLi-Nvidia]

What this adds

Per-step visibility into what the TransferQueue data plane costs, so "is the
data plane my bottleneck, and which part of it" is answerable from a dashboard
instead of a profiler.

Off by default at the top level (data_plane.enabled: false); when the data
plane is on, observability.enabled: true wraps the client in
MetricsDataPlaneClient. Measured overhead is ~15-20 µs per op, and the
wrapper bills itself so that cost is visible rather than asserted.

The metrics

Everything is logged under data_plane/cluster/ (the driver plus every policy
worker, summed) or data_plane/driver/ when the fan-out reaches one process.

Series	Answers
step/frac_of_step	Is the data plane worth optimising at all? The busiest process's time over the step's own wall clock. Processes run concurrently inside one step, so summing them exceeds it and averaging dilutes the busiest — the step waits on one process
step/percent_of_dataplane/by_op/{put,get,clear,register}	Which call is expensive? Sums to 100
step/percent_of_dataplane/by_cause/{fixed_overhead,transfer}	Fixed per-request cost, or bandwidth? Sums to ≤100; only ops with an identifiable fit can be split
step/volume_mb/by_op/{get,put}	Which direction the traffic went
step/wall_ms, step/comm_volume_mb	How much time and traffic
now/bytes_outstanding_mb	Occupancy — bytes put and not yet cleared. Rising is the leak signal
step/self/{overhead_ms,frac}	What measuring cost
step/hash/*	Only with verify_tensor_hash on

Plus a per-op breakdown table (data_plane/{cluster,driver}/breakdown),
ordered worst-first, carrying the detail that would be noise as 32 line charts:
calls, mean/max/p50/p90 ms, the overhead/transfer split, and MB.

Two denominators, deliberately. frac_of_step divides by the step;
percent_of_dataplane divides by the data plane. A workload can be 43% put
and still not be worth touching — you need both.

Reading the numbers correctly

Three things are easy to misread, and are documented in
nemo_rl/data_plane/README.md:

• Volume counts transfers, not data. A byte written then read counts on
  both sides, and every process's transfers are summed. Correct for "what
  crossed the wire"; useless for "how big was the batch".
• put reads small. The rollout actor builds its own client and is not on
  the worker group, so its kv_first_write of the whole batch is in neither
  volume_mb nor comm_volume_mb.
• by_cause is often absent. Request sizes in RL are frequently uniform,
  which makes the affine fit unidentifiable. It reports nothing rather than an
  arbitrary split.

Optional wire-hash guard

verify_tensor_hash: false by default, in the recipes as well — the suites
that gate on it pass data_plane.observability.verify_tensor_hash=True on the
command line, so a recipe you copy does not silently opt into it.

When on, it fingerprints every tensor row with torch.hash_tensor on the way
in and out and reports divergence — the one failure class nothing else
notices, because a mis-sharded read trains happily on the wrong rows. Catches a
row served from the wrong sample, two rows swapped between DP ranks,
truncation, a bf16→fp32 change, and a single element in any dtype.

The wire-in reading travels with the row. Each field is mirrored by a
<field>_hash int64 column, written by the same put and declared alongside the
field. Holding the reading in the putting process only ever verified a
same-process round trip, and the transfer worth checking is not one: the
rollout actor writes what the policy workers read. The mirror is per top-level
field, so a multimodal images reduces its leaves to one images_hash.

Measured on an interleaved same-node A/B — off/on/off/on, 20 steps each,
all four arms in one allocation (Llama-3.2-1B, 1 node x 4 GB300, TQ simple,
24.4 MB/step). Medians over steps >= 5, which drops vLLM warm-up and the first
refit.

	off	on
hash/rows_recorded	—	1536
hash/rows_checked	—	2560 - more than it wrote, i.e. it verifies rows it did not produce
hash/rows_unverified	—	0 - was the normal case before the reading travelled
hash/mismatches, guard_failures, fields_skipped	—	0
step/wall_s	0.219 / 0.216 s	0.273 / 0.274 s
step/self/overhead_ms	3.8 / 3.9 ms	112 / 113 ms
step/frac_of_step	1.42 %	1.73 / 1.83 %
step/comm_volume_mb	24.45	24.45
accounted cost of the guard		+164 ms/step

wall_s and self/overhead_ms are disjoint — the first is the RPC, the second
is the wrapper with the RPC subtracted out — so the guard's cost is their sum:
+55 ms of extra <field>_hash columns on the wire, +109 ms of hashing.

The control is the two off arms, ~40 minutes apart in the same allocation.
They differ by 2.9 ms on wall_s and 0.005 pp on frac_of_step, against a
guard effect of 164 ms — a 50:1 ratio of signal to drift. Both pairs, (b-a)
and (d-c), agree within 2%. comm_volume_mb is identical across arms to four
decimals, which is the check that the two arms moved the same data.

Its effect on step time is not measurable, and that is not evidence it is
free. total_step_time reads 15.36 / 15.69 / 15.22 / 15.21 s; the two
identical off arms differ by 0.14 s and per-arm MAD runs 0.16-0.56 s, so a
0.16 s effect sits inside the noise. The data-plane counters are the quiet
channel and the step clock the loud one, which is why the cost is reported from
the counters. An earlier revision of this section claimed the two agreed "to
within 4 ms" — that was a single pair read against this floor, and it is
withdrawn.

This supersedes the 10.2 ms this section previously claimed, measured when the
digest stayed process-local and most rows were abstained on rather than
compared.

Blind spot, measured rather than assumed: the row fold is an XOR reduction, so
it cannot see a permutation within a row (0/200 for a two-token swap). The
per-field fold over leaves uses * 31 + rather than XOR, so identical leaves
cannot cancel there.

Verification

• 369 unit tests pass, 3 skipped (tests/unit/data_plane, run inside the
  nightly container so TransferQueue is real - simple and mooncake_cpu
  round trips both included), plus the two test_grpo.py cases covering both
  logging scopes. Re-run at the current head, 45 s. That includes 23/23
  adversarial hash-guard checks (10 corruption modes caught, zero false alarms
  over a 500-row soak, every shard grouping) and 48/48 metric-audit checks
  against an independently computed ground truth.

• Both accepted limits of the guard are pinned by name, so neither can
  become a surprise: a within-row permutation, and a constant-valued row of
  even length, whose digest is its dtype|row shape seed alone because the
  XOR fold cancels in pairs. advantages has exactly that shape.

• A real 25-step GRPO run — Llama-3.2-1B on OpenMathInstruct-2, 1 node ×
  4 GB200, TransferQueue simple, 5 processes reporting:
  https://wandb.ai/nvidia/nemo-rl-dataplane-obs/runs/w316chaj

  get 65% of data-plane time · 24 MB moved per step · 38,400 rows
  hash-checked, 0 mismatches. Its frac_of_step predates the reduction
  change and was a per-process mean; the current figure, measured on the
  four-arm A/B above, is 1.42% of a ~15.3 s step guard-off and 1.73-1.83%
  with the guard on — the slowest process's serial data-plane time, since that
  is the one the gradient all-reduce makes everyone wait for.

Notes for review

• Percentiles are gated on sample size: p50 needs 20 calls in the window, p90
  needs 40 (~4 observations above the rank). Below that the key is absent
  rather than reporting the maximum under a percentile's name. p90 rather than
  p99 because a step holds tens of calls — a p99 off 58 samples equalled the
  maximum 80% of the time.
• max_ms is scoped to the step by being reset by its reader, since a
  maximum cannot be differenced out of a cumulative counter.
• Supersedes the 345-line observability.py already on main.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[ZhiyuLi-Nvidia]

/ok to test 48ed054

[ZhiyuLi-Nvidia]

/ok to test 67be5a7

[ZhiyuLi-Nvidia]

/ok to test 501a7fd

[ZhiyuLi-Nvidia]

/ok to test 4ea3bc7

[ZhiyuLi-Nvidia]

/ok to test 2d5719b

[ZhiyuLi-Nvidia]

/ok to test 05348d2

[ZhiyuLi-Nvidia]

/ok to test dc3d609

[ZhiyuLi-Nvidia]

/ok to test 485a454

[ZhiyuLi-Nvidia]

/ok to test e1505b1

[ZhiyuLi-Nvidia]

/ok to test 96c161c

[ZhiyuLi-Nvidia]

/ok to test 2af2c3a

[terrykong]

Thanks for this — the measurement discipline behind it is better than most performance PRs get. Running the A/B interleaved on one node (off/on/off/on, 20 steps each) instead of a single pair, quoting the 687 ms gap between the two off-arms as the noise floor, and defaulting the expensive hash guard to false in all 24 recipes so it only ever turns on from the command line — that is the shape of evidence that makes a number believable. Every one of the five threads @zyzhou5 opened came back with a measurement.

The feature earns its keep. The data plane moves the whole training batch between processes and had no per-op timing at all; frac_of_step is the only number that says whether optimising it is worth anything. Land it once the blocking item is fixed.

One blocking item. The comment on ppo-qwen2.5-1.5b-gsm8k-2n8g-megatron-valuetp2sp-dynbatch-noncolocated-async-single-controller.sh:28 explains it: that suite's config chain is the only one of the 24 with no data_plane.observability block left in it, so the new override kills the nightly at config parse. The fix needs enabled: true as well as verify_tensor_hash — see the comment for why restoring just the deleted lines is not enough.

Why this reached head invisibly. No job on this PR runs any of the 24 modified suite scripts. The label is CI:L1, which runs tests/functional/, and test_all_tests_can_find_config_if_dryrun exits in common.env before any uv run, so the override is never parsed. gh pr checks 3616 is green and says nothing at all about these scripts.

The PR description has gone stale. It still advertises step/percent_of_dataplane/by_cause/{fixed_overhead,transfer}, a "Reading the numbers correctly" bullet about the affine fit, and an "overhead/transfer split" column. grep -rn by_cause over *.py, *.md, *.yaml and *.sh at head returns nothing — b80b24cb4 removed the fit, and _BREAKDOWN_COLUMNS (observability.py:975-984) has no such column. step/wall_ms in that table should be step/wall_s. The README itself is correct throughout; only the description needs updating.

What was run here. 518 unit tests passed (15 skipped, 0 failed) across tests/unit/data_plane and tests/unit/algorithms/test_grpo.py. The three pinned ruff hooks pass. No GPU on this machine, so no nightly, no real TransferQueue, and nothing exercising mooncake or RDMA.

One last small thing: the single-controller copy of the log-before-commit fix has no unit test, while the grpo_sync copy does.

Generated by Claude Code

[ZhiyuLi-Nvidia]

/ok to test d60550e