Repository navigation
[megatron] GLM-5.3-Flash DSA: shard the k-pool indexer's query rows across TP (opt-in) - #2422
Conversation
…n query chunks mcore_ext/dsa_kpool.py: fused_qk_topk_kpool scores and selects pools in query chunks. Verbatim #7522 materializes FP32 [sq, b, heads, sq / kpool] per-head scores (32 GiB/GPU at 32k), so GLM-5.3-Flash could not train past ~16k. Top-k is per query row, so chunking is exact. SKYRL_DSA_KPOOL_SCORE_CHUNK_ELEMS sets the chunk cap (default 2 GiB of FP32 scores, 128 queries per chunk at 512k). 8 GiB was ~5% faster per step at 512k (TP8 x DP8, 64 B200) with the same peak memory. Test: test_dsa_kpool.py (chunked selection == one-shot, ragged chunks). Split out of #2341 (first of: chunked indexer scoring, TP-sharded indexer, query-chunked attention). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
There was a problem hiding this comment.
Code Review
This pull request introduces tensor-parallel query sharding for the k-pool indexer, allowing query rows to be split across the tensor-parallel group to reduce redundant scoring overhead. The changes include adding an opt-in environment variable SKYRL_DSA_INDEXER_TP_SHARD, updating the fused_qk_topk_kpool function to shard queries and gather selections, and adding a multi-GPU unit test to verify exact matching. A critical issue was identified in the all_gather_into_tensor implementation where a dimension mismatch between the input and output tensors will cause a runtime error; reshaping the gathered tensor as suggested is required to fix this.
| gathered = torch.empty((shard_size, batch, rows_per_rank, budget), dtype=torch.int32, device=q.device) | ||
| torch.distributed.all_gather_into_tensor(gathered, local.contiguous(), group=query_shard_group) | ||
| local = gathered.permute(1, 0, 2, 3).reshape(batch, shard_size * rows_per_rank, budget)[:, :sq] |
There was a problem hiding this comment.
The all_gather_into_tensor API expects the output tensor to have the same number of dimensions as the input tensor, with the 0th dimension sized to world_size * input_tensor.size(0). Currently, local has 3 dimensions (batch, rows_per_rank, budget), but gathered is initialized with 4 dimensions (shard_size, batch, rows_per_rank, budget). This will cause a RuntimeError at runtime whenever batch > 1 (or even batch = 1 on strict PyTorch versions) due to dimension and size mismatches.\n\nTo fix this, initialize gathered with 3 dimensions (shard_size * batch, rows_per_rank, budget) and then use .view() to reshape it before permuting.
gathered = torch.empty((shard_size * batch, rows_per_rank, budget), dtype=torch.int32, device=q.device)\n torch.distributed.all_gather_into_tensor(gathered, local.contiguous(), group=query_shard_group)\n local = gathered.view(shard_size, batch, rows_per_rank, budget).permute(1, 0, 2, 3).reshape(batch, shard_size * rows_per_rank, budget)[:, :sq]
|
| varlen_ends=ends, | ||
| cu_seqlens_kv=cu, | ||
| always_select_tail=True, | ||
| query_shard_group=group, |
There was a problem hiding this comment.
Production wiring remains untested. This test passes a process group directly to
fused_qk_topk_kpool, but production must read SKYRL_DSA_INDEXER_TP_SHARD and pass Megatron's TP group through the attention path. The test would still pass if that wiring stopped working, leaving redundant scoring or a collective failure undetected. Please add coverage for the flag-enabled attention path.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
…cast, preallocated output Adopts three pieces of the query-chunked selection that landed in NVIDIA/Megatron-LM#7522 (b27efd8): past one chunk, score under no_grad with ReLU and head weighting in place (one FP32 score buffer per chunk instead of two or three), cast k_pooled to FP32 once instead of per chunk, and write into a preallocated [batch, sq, budget] selection instead of torch.cat-ing chunks. Selections are bitwise unchanged; the one-chunk path still returns full scores with autograd. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
61b12f0 to
4ba48c8
Compare
… global positions test_kpool_query_chunking_is_exact only shows the chunked and one-shot paths agree. Add test_kpool_selection_matches_reference: an FP64 reference built from the definition (visible complete pools of the query's own document, head-weighted q.k scores, top budget, incomplete tail) checks both paths past index_topk on packed multi-document inputs, with and without ReLU. It checks the top-k properties rather than an ordering, so FP32/FP64 rounding at a near-tie can't fail it. Mutation-checked on B200: selecting the lowest-scoring pools, or dropping the per-document bounds, fails all 12 cases, while the chunking test passes both. The packed tests built bounds with generate_varlen_mask_params_for_positions from per-document local positions; it takes global packed-row positions (as DSAttention passes them), so every document after the first was bounded as if it were the first. Pass torch.arange(seqlen). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…cross TP (opt-in) Under sequence parallelism every TP rank runs the k-pool indexer on the same gathered sequence with the same (frozen, replicated) indexer weights, so at TP8 all 8 ranks computed identical O(sq^2) pool scores. SKYRL_DSA_INDEXER_TP_SHARD=1 passes the TP group as query_shard_group: each rank scores and selects pools only for its contiguous slice of query rows, then the int32 selections are all-gathered. Per query row nothing changes, so selections are bitwise identical. Full GLM-5.3-Flash, 64 B200 (TCP NCCL), 512k tokens/seq, TP8: warm fwd+bwd 1545 -> 1094 s, i.e. ~1/3 of the step was replicated indexer scoring. Test: test_dsa_kpool_tp_shard.py (two ranks, ragged last shard, many score chunks: identical to unsharded). Stacked on the chunked-indexer-scoring PR (shards walk that chunk loop). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4ba48c8 to
e0f681b
Compare
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K4BmK8yuJTMbprizvT4Rja
…a-indexer-tp-shard Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K4BmK8yuJTMbprizvT4Rja
…n query chunks (#2413) ## Summary First of three PRs split from the original GLM-5.3-Flash DSA long-context change: - **this PR**: chunked k-pool indexer scoring (on by default, exact) - **#2422**: TP-sharded indexer (opt-in), stacked on this one - **#2423**: query-chunked sparse attention (opt-in), independent, on main `mcore_ext/dsa_kpool.py`: `fused_qk_topk_kpool` scores and selects pools in query chunks. Verbatim #7522 materializes FP32 `[sq, b, heads, sq / kpool]` per-head scores (32 GiB/GPU at 32k), so GLM-5.3-Flash could not train past ~16k. Top-k is per query row, so chunking is exact. Only affects GLM-5.3-Flash's k-pool path. `SKYRL_DSA_KPOOL_SCORE_CHUNK_ELEMS` sets the chunk cap (default 2 GiB of FP32 scores, which is 128 queries per chunk at 512k). ### Aligned with upstream #7522 @ `b27efd8` #7522 has since added its own query chunking (`_kpool_topk_from_score_chunks`). The second commit here (`15fde54d`) adopts its structure past one chunk: scoring under `no_grad` with ReLU and head weighting in place (one FP32 score buffer per chunk instead of two or three), `k_pooled` cast to FP32 once instead of per chunk, and a preallocated `[batch, sq, budget]` selection instead of `torch.cat`. Selections are bitwise unchanged; the one-chunk path still returns full scores with autograd. What stays SkyRL-specific is the configurable cap: upstream's is a fixed 256 MiB. One selection at 512k (one TP8 rank's 65,536 query rows x 131,072 pools, 32 heads, packed varlen masking, 1 B200; median of 3, spread <0.1%; all runs select bitwise-identical pools): | implementation | 256 MiB (16 queries/chunk) | 2 GiB (128) | 8 GiB (512) | |---|---|---|---| | upstream #7522 @ `b27efd8` | 3.308 s, 0.61 GiB | 2.640 s, 2.45 GiB | 2.598 s, 8.88 GiB | | this PR, first commit | 3.480 s, 0.76 GiB | 2.652 s, 4.31 GiB | 2.593 s, 16.5 GiB | | this PR, with `15fde54d` | 3.282 s, 0.57 GiB | **2.626 s, 2.38 GiB** | 2.585 s, 8.57 GiB | (GiB = peak extra memory of the call.) With 11 DSA layers under full recompute (22 selections per step), upstream's 256 MiB cap costs ~15 s of a ~217 s step versus 2 GiB; 8 GiB saves under 1 s more over 2 GiB for 6 GiB more memory, so the 2 GiB default stays. ## Testing - [x] `gpu_ci/patches/megatron/mcore_ext/test_dsa_kpool.py` (chunked selection == one-shot): 5 passed with `15fde54d` (B200) - [x] CPU: first commit vs `15fde54d` vs one-shot, 72 chunked cases (packed / dense / additive mask, ReLU on and off, 1 to n-1 queries per chunk): bitwise-equal indices, identical one-shot scores, gradients still flow through them - [x] Chunk-size benchmark above - [x] Logprobs vs vLLM on the real 4-layer GLM-5.3-Flash checkpoint (TP2 / EP4, 4 B200). Re-run 2026-10-06 with every split PR of #2341 merged onto main (this PR's first commit; `15fde54d` selects bitwise-identical pools): Megatron vs vLLM 0.0658 (short) and 0.0548 (`kpool_beyond_topk`); vLLM pre/post sync 0.255 and 0.358. All pass. Originally measured on the pre-split branch: | row | Megatron vs vLLM (thr 0.1) | main | vLLM pre/post sync | main | |---|---|---|---|---| | short, 128 tokens | 0.0651 | 0.0623 | 0.2402 (thr 0.3) | 0.2467 | | `kpool_beyond_topk` (past `index_topk`) | 0.0558 | 0.0546 | 0.3446 (thr 0.5) | 0.3382 | Split out of #2341. 🤖 Generated with [Claude Code](https://claude.com/claude-code) <!-- CURSOR_SUMMARY --> --- > [!NOTE] > **High Risk** > K-pool indexer mistakes change which tokens are attended without raising, affecting Megatron vs vLLM parity past `dsa_indexer_topk`; chunking is intended to be exact per query row but this path is explicitly called the riskiest patch entry. > > **Overview** > **GLM-5.3-Flash k-pool DSA indexer** no longer materializes full-sequence FP32 index scores in `fused_qk_topk_kpool`. Long contexts now **score and top-k select pools in query chunks** via `_kpool_topk_in_query_chunks`, with chunk size capped by **`SKYRL_DSA_KPOOL_SCORE_CHUNK_ELEMS`** (default ~2 GiB of FP32 scores). When one chunk covers the whole sequence, behavior stays on the original `_compute_index_scores` path so full scores (and autograd) are still returned; otherwise only pool indices are kept and `index_scores` is `None`. > > Docs in the Megatron patches README describe this deliberate deviation from verbatim #7522 and how it relates to upstream’s smaller fixed chunk cap. > > GPU tests add **`test_kpool_query_chunking_is_exact`** (chunked vs one-shot indices) and **`test_kpool_selection_matches_reference`** (one-shot and chunked vs a brute-force definition, including packed varlen and ReLU on/off). > > <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit 1105a1b. Bugbot is set up for automated code reviews on this repo. Configure [here](https://www.cursor.com/dashboard/bugbot).</sup> <!-- /CURSOR_SUMMARY --> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K4BmK8yuJTMbprizvT4Rja
Stacked on #2413 (chunked k-pool indexer scoring): the shards walk that PR's chunk loop. Review only the top commit.
Summary
Under sequence parallelism every TP rank runs the k-pool indexer on the same gathered sequence with the same frozen, replicated indexer weights, so at TP8 all 8 ranks computed identical O(sq^2) pool scores (every query scores every pool).
SKYRL_DSA_INDEXER_TP_SHARD=1passes the TP group asquery_shard_group. Each rank scores and selects pools only for its contiguous slice of query rows, then the int32 selections are all-gathered. Per query row nothing changes, so selections are bitwise identical. Opt-in; GLM-5.3-Flash only.Full GLM-5.3-Flash, 64 B200 (TCP NCCL), 512k tokens/seq, TP8: warm fwd+bwd 1545 -> 1094 s, i.e. ~1/3 of the step was replicated indexer scoring (sharding 8 ways removes 7/8 of it).
Testing
gpu_ci/patches/megatron/mcore_ext/test_dsa_kpool_tp_shard.py(two ranks, ragged last shard, many score chunks: identical to unsharded), together with A'stest_dsa_kpool.py: 7 passed on this branch (B200)#2418 (context parallelism) is stacked on this PR.
Split out of #2341.
🤖 Generated with Claude Code
Note
Medium Risk
Opt-in distributed path changes which tokens the sparse DSA indexer selects if sharding or gather is wrong; selections are intended to be bitwise-identical and are GPU-tested, but errors would not raise.
Overview
Adds an opt-in (
SKYRL_DSA_INDEXER_TP_SHARD=1) optimization for GLM-5.3-Flash k-pool DSA: tensor-parallel ranks no longer each run the full O(sq×pools) indexer scoring on identical gathered inputs.glm5_next/dsa.pypasses the TP process group intofused_qk_topk_kpoolasquery_shard_group. Inmcore_ext/dsa_kpool.py, each rank scores and top-k’s only its contiguous slice of query rows (the chunked path gainsq_lo/q_hi), pads the last shard when lengths don’t divide evenly, all-gathers int32 pool ids, and reassembles a full[batch, sq, budget]tensor—per-row behavior stays the same as unsharded.Docs in the Megatron patches README note the deviation and the new GPU test.
test_dsa_kpool_tp_shard.pyruns a 2-rank NCCL check (ragged shard + small score chunks) that sharded selection matches a single rank bitwise.Reviewed by Cursor Bugbot for commit d0637d4. Bugbot is set up for automated code reviews on this repo. Configure here.