Skip to content

Commit 902468d

Browse files
[PyTorch] Add no-load-balance THD all-gather CP (#3221)
* Add packed THD partitioning for all-gather CP Allow causal THD attention to shard the complete packed token buffer with mirrored context-parallel chunks while retaining document boundaries through sequence metadata. This supports workloads whose individual documents are not divisible by twice the CP size. Keep the existing per-document partition as the default and reject backend or attention combinations that the prototype has not validated, so existing paths remain unchanged. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Refine THD super-sequence partitioning Reuse the existing THD CUDA partition and reorder kernels by representing the complete physical token buffer as one partitioning sequence. Preserve CPU and mixed-device dataloader behavior with a reference fallback. Rename the opt-in policy to packed_super_sequence to distinguish physical partitioning from THD packing, and add a narrowly gated matched-input benchmark path so the two policies can be compared without workload or timing asymmetry. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Add contiguous THD all-gather partition Allow relaxed THD all-gather to assign one contiguous chunk per CP rank while preserving the mirrored policy for compatibility and comparison. Rank-major ownership needs no KV reorder and reduces each rank to one attention step; keep it opt-in while the performance tradeoffs are evaluated. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Remove mirrored packed THD partition Keep per-document partitioning as the default and packed-contiguous as the only packed opt-in so the experimental API has one global ownership contract. Delete the unused 2*CP global metadata and reorder paths, simplify packed metadata to one chunk and one attention step per rank, and retain a negative test for the retired selector. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Gate packed THD partition with env flag Keep the experimental packed-contiguous selection internal to context parallelism so public TransformerLayer and attention APIs retain their existing signatures. Per-document partitioning remains the default unless NVTE_EXPERIMENTAL_CP_AG_THD_PACKED_CONTIGUOUS=1 is set. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Remove experimental packed THD tests from PR Keep the initial upstream review focused on the production context-parallel implementation. The test development remains available in the preceding commits while the branch tip restores the existing test suite unchanged. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Rename experimental THD policy and add coverage Name the single contiguous-chunk policy after its deliberate lack of causal load balancing so the performance tradeoff is explicit. Remove the CPU reference partitioner and require the existing CUDA path for per-document metadata. Capture the selected layout through backward and add focused helper plus CP2 forward/backward coverage so mutable environment state cannot make the two passes use different token orders. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Move THD no-load-balance checks to CP dispatch Validate the experimental policy where context-parallel communication is selected so unsupported combinations fail before entering custom autograd. Passing the captured mode into the internal all-gather call also prevents a second environment lookup from selecting a different layout. Restore the default CPU dataloader slicing behavior, keep one stream dependency per format path, and exercise padded feature execution so the lean experimental path does not regress existing callers. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Expose CP attention load-balancing strategy Make token partitioning explicit so input slicing and attention cannot diverge through mutable process state. Reuse native THD indices for CUDA while preserving CPU dataloader behavior, and cover the supported FusedAttention and unpadded FlashAttention 3 paths. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Rename CP load-balancing strategy Use the shorter public name because the strategy governs both input partitioning and attention execution. Consolidate redundant low-level partition tests into the existing end-to-end slicing coverage. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Mark no-load-balance CP strategy experimental Document the experimental contract at the public selection points. Preserve the legacy child-setter invocation for the default strategy so the additive API does not disrupt existing extension modules. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Skip known sm90 deterministic THD OOM The focused no-load-balance test uses cp_2_0, which reaches the same known cuDNN deterministic THD backward workspace limit as the generic CP matrix. Apply the equivalent sm90 skip so the standalone coverage does not bypass that guard. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Move deterministic THD guard to backend selection Apply the known sm90 cuDNN backward workspace restriction in the common selector so every caller avoids choosing FusedAttention for unsupported large THD training configurations. Keep inference, smaller shapes, nondeterministic execution, and other architectures unaffected. Remove test-local skips and make the focused CP case use normal backend availability preflight so coverage follows the production selection path. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> * Limit deterministic THD guard to F16 backend The empirical large-workspace limitation applies to deterministic F16/BF16 THD backward. Keep the FP8 fused-attention backend available for the same geometry. Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com> --------- Signed-off-by: Sudhakar Singh <sudhakars@nvidia.com>
1 parent 2fd4604 commit 902468d

11 files changed

Lines changed: 624 additions & 113 deletions

File tree

‎tests/pytorch/attention/run_attention_with_cp.py‎

Lines changed: 55 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -11,16 +11,17 @@
1111
import torch.distributed as dist
1212
from transformer_engine.pytorch.attention.dot_product_attention.context_parallel import (
1313
get_cu_seqlens_on_cp_rank,
14+
get_thd_partitioned_indices,
1415
)
1516
from transformer_engine.pytorch.attention.dot_product_attention.utils import combine_and_quantize
16-
import transformer_engine_torch as tex
1717
from transformer_engine.pytorch import DType
1818
from test_attention_with_cp import (
1919
model_configs_flash_attn,
2020
model_configs_fused_attn,
2121
)
2222
from transformer_engine.pytorch import (
2323
autocast,
24+
CPLoadBalancingStrategy,
2425
DotProductAttention,
2526
Float8Quantizer,
2627
Float8CurrentScalingQuantizer,
@@ -52,6 +53,7 @@ def generate_input_shapes(
5253
world_size: int,
5354
kernel_backend: str,
5455
fa_pad_between_seqs: str = "False",
56+
load_balancing_strategy=CPLoadBalancingStrategy.DUAL_CHUNK_SWAP,
5557
):
5658
if qkv_format == "bshd":
5759
q_input_shape = (
@@ -110,8 +112,33 @@ def generate_input_shapes(
110112
cu_seqlens_q_padded = None
111113
cu_seqlens_kv_padded = None
112114
elif qkv_format == "thd":
113-
seqlens_q = torch.randint(0, config.max_seqlen_q + 1, [config.batch_size]).to(torch.int32)
114-
seqlens_q_padded = (seqlens_q + 2 * world_size - 1) // (world_size * 2) * (world_size * 2)
115+
if load_balancing_strategy is CPLoadBalancingStrategy.NO_LOAD_BALANCE:
116+
assert config.batch_size == 2
117+
if kernel_backend == "FlashAttention" and fa_pad_between_seqs == "False":
118+
seqlens_q = torch.tensor(
119+
[config.max_seqlen_q - 2, config.max_seqlen_q], dtype=torch.int32
120+
)
121+
assert seqlens_q.sum().item() % world_size == 0
122+
seqlens_q_padded = seqlens_q
123+
else:
124+
# Exercise both document padding and a CP chunk boundary inside a document.
125+
seqlens_q = torch.tensor(
126+
[config.max_seqlen_q - 2, config.max_seqlen_q - 1],
127+
dtype=torch.int32,
128+
)
129+
padded_total = 2 * config.max_seqlen_q
130+
assert padded_total % world_size == 0
131+
seqlens_q_padded = torch.tensor(
132+
[config.max_seqlen_q - 1, config.max_seqlen_q + 1],
133+
dtype=torch.int32,
134+
)
135+
else:
136+
seqlens_q = torch.randint(0, config.max_seqlen_q + 1, [config.batch_size]).to(
137+
torch.int32
138+
)
139+
seqlens_q_padded = (
140+
(seqlens_q + 2 * world_size - 1) // (world_size * 2) * (world_size * 2)
141+
)
115142
cu_seqlens_q_padded = torch.cat(
116143
[
117144
torch.zeros([1], dtype=torch.int32),
@@ -206,10 +233,12 @@ def run_dpa_with_cp(
206233
is_training="True",
207234
fa_pad_between_seqs="False",
208235
deterministic="False",
236+
load_balancing_strategy="DUAL_CHUNK_SWAP",
209237
log_level=logging.WARNING,
210238
):
211239
"""Test DotProductAttention module with context parallelism"""
212240
logging.root.setLevel(log_level)
241+
load_balancing_strategy = CPLoadBalancingStrategy[load_balancing_strategy]
213242
# When is_training is False, gradient outputs are None.
214243
is_training = is_training == "True"
215244
pad_between_seqs = None
@@ -331,7 +360,14 @@ def run_dpa_with_cp(
331360
cu_seqlens_kv,
332361
cu_seqlens_q_padded,
333362
cu_seqlens_kv_padded,
334-
) = generate_input_shapes(qkv_format, config, world_size, kernel_backend, fa_pad_between_seqs)
363+
) = generate_input_shapes(
364+
qkv_format,
365+
config,
366+
world_size,
367+
kernel_backend,
368+
fa_pad_between_seqs,
369+
load_balancing_strategy,
370+
)
335371
q_orig = torch.clamp(torch.randn(q_input_shape, dtype=dtypes[dtype]), min=-1, max=1).cuda()
336372
k_orig = torch.clamp(torch.randn(k_input_shape, dtype=dtypes[dtype]), min=-1, max=1).cuda()
337373
v_orig = torch.clamp(torch.randn(v_input_shape, dtype=dtypes[dtype]), min=-1, max=1).cuda()
@@ -468,11 +504,21 @@ def run_dpa_with_cp(
468504
x.view(*x.shape[:seq_dim], -1, *x.shape[(seq_dim + 2) :]) for x in [q_, k_, v_, dout_]
469505
]
470506
elif qkv_format == "thd":
471-
seq_idx_q = tex.thd_get_partitioned_indices(
472-
cu_seqlens_q_padded, q_.shape[0], world_size, rank
507+
seq_idx_q = get_thd_partitioned_indices(
508+
cu_seqlens_q_padded,
509+
q_.shape[0],
510+
world_size,
511+
rank,
512+
device=q_.device,
513+
load_balancing_strategy=load_balancing_strategy,
473514
)
474-
seq_idx_kv = tex.thd_get_partitioned_indices(
475-
cu_seqlens_kv_padded, k_.shape[0], world_size, rank
515+
seq_idx_kv = get_thd_partitioned_indices(
516+
cu_seqlens_kv_padded,
517+
k_.shape[0],
518+
world_size,
519+
rank,
520+
device=k_.device,
521+
load_balancing_strategy=load_balancing_strategy,
476522
)
477523
q_, dout_ = [x.index_select(0, seq_idx_q) for x in [q_, dout_]]
478524
k_, v_ = [x.index_select(0, seq_idx_kv) for x in [k_, v_]]
@@ -516,6 +562,7 @@ def run_dpa_with_cp(
516562
cp_comm_ranks,
517563
torch.cuda.Stream(),
518564
cp_comm_type,
565+
load_balancing_strategy,
519566
)
520567
if config.softmax_type != "vanilla":
521568
core_attn.softmax_offset.grad.zero_()

‎tests/pytorch/attention/test_attention_with_cp.py‎

Lines changed: 53 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -684,23 +684,6 @@ def test_cp_with_fused_attention(
684684
pytest.skip("Deterministic mode does not support non-vanilla softmax with FusedAttention")
685685
if _deterministic and config.attn_bias_type == "post_scale_bias" and is_training:
686686
pytest.skip("Deterministic mode does not support post_scale_bias with requires_grad")
687-
# Observed: cuDNN det THD backward asks for ~128 * bHSS bytes of workspace
688-
# on sm90; at 1<<30 that's 128 GiB, won't fit on H100's 80 GB. Held exactly
689-
# at b=2 + power-of-2 S in our sweep; for b>=3 the workspace was observed to
690-
# grow super-linearly (b=4 took ~4x the b=2 amount, not 2x) — revisit if a
691-
# config uses b>2.
692-
SM90_DET_FUSED_THD_BWD_MAX_BHSS = 1 << 30
693-
if (
694-
_deterministic
695-
and qkv_format == "thd"
696-
and get_device_compute_capability() == (9, 0)
697-
and config.batch_size * config.num_heads * config.max_seqlen_q * config.max_seqlen_kv
698-
>= SM90_DET_FUSED_THD_BWD_MAX_BHSS
699-
):
700-
pytest.skip(
701-
"Deterministic FusedAttention backward with THD format OOMs on sm90"
702-
" for large bHSS configs (known cuDNN issue)."
703-
)
704687

705688
_submit(
706689
pool,
@@ -718,3 +701,56 @@ def test_cp_with_fused_attention(
718701
deterministic=_deterministic,
719702
log_level=pytest_logging_level,
720703
)
704+
705+
706+
@pytest.mark.skipif(get_cudnn_version() < (8, 9, 7), reason="cuDNN 8.9.7+ is required.")
707+
@pytest.mark.skipif(
708+
get_device_compute_capability() < (9, 0), reason="FusedAttention THD requires sm90+."
709+
)
710+
def test_cp_with_fused_attention_no_load_balance(cp_pool):
711+
"""Check experimental single-chunk forward/backward."""
712+
config = copy.deepcopy(model_configs_fused_attn["cp_2_0"])
713+
config.context_parallel = True
714+
config.cp_comm_type = "all_gather"
715+
config.attn_mask_type = "padding_causal"
716+
available_backends, _, _ = get_available_attention_backends(
717+
config,
718+
qkv_dtype=torch.bfloat16,
719+
qkv_layout="thd_thd_thd",
720+
pad_between_seqs=True,
721+
is_training=True,
722+
deterministic=_deterministic,
723+
)
724+
if not available_backends[1]:
725+
pytest.skip("No attention backend available.")
726+
_submit(
727+
cp_pool(2),
728+
dtype="bf16",
729+
model="cp_2_0",
730+
qkv_format="thd",
731+
kernel_backend="FusedAttention",
732+
cp_comm_type="all_gather",
733+
load_balancing_strategy="NO_LOAD_BALANCE",
734+
deterministic=_deterministic,
735+
log_level=pytest_logging_level,
736+
)
737+
738+
739+
@pytest.mark.skipif(
740+
get_device_compute_capability() != (9, 0) or not FlashAttentionUtils.v3_is_installed,
741+
reason="FlashAttention 3 requires sm90 and an installed FA3 package.",
742+
)
743+
def test_cp_with_flash_attention_3_no_load_balance(cp_pool):
744+
"""Check the supported unpadded FlashAttention 3 path."""
745+
_submit(
746+
cp_pool(2),
747+
dtype="bf16",
748+
model="cp_2_0",
749+
qkv_format="thd",
750+
kernel_backend="FlashAttention",
751+
cp_comm_type="all_gather",
752+
fa_pad_between_seqs=False,
753+
load_balancing_strategy="NO_LOAD_BALANCE",
754+
deterministic=_deterministic,
755+
log_level=pytest_logging_level,
756+
)

‎tests/pytorch/attention/test_cp_utils.py‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,9 +7,14 @@
77
import itertools
88
import torch
99
import unittest
10+
from transformer_engine.pytorch import CPLoadBalancingStrategy
1011
from transformer_engine.pytorch.attention.dot_product_attention.context_parallel import (
1112
_zero_thd_padding,
1213
get_batch_on_this_cp_rank,
14+
get_no_load_balance_thd_causal_metadata,
15+
get_thd_partitioned_indices,
16+
restore_thd_gathered_kv,
17+
unrestore_thd_gathered_kv,
1318
pad_thd_sequences_for_cp,
1419
generate_positional_ids_for_cp,
1520
)
@@ -21,6 +26,49 @@
2126
tex = None
2227

2328

29+
class TestTHDPartitioning(unittest.TestCase):
30+
def test_default_partition_rejects_cpu_metadata(self):
31+
with self.assertRaisesRegex(AssertionError, "requires CUDA cu_seqlens"):
32+
get_thd_partitioned_indices(torch.tensor([0, 8]), 8, 2, 0)
33+
34+
def test_no_load_balance_metadata_handles_document_padding_boundary(self):
35+
cu_seqlens = torch.tensor([0, 6, 10], dtype=torch.int32)
36+
cu_seqlens_padded = torch.tensor([0, 8, 12], dtype=torch.int32)
37+
38+
q_cu, q_cu_padded, kv_cu = get_no_load_balance_thd_causal_metadata(
39+
cu_seqlens,
40+
cu_seqlens_padded,
41+
total_tokens=12,
42+
cp_size=2,
43+
cp_rank=1,
44+
)
45+
46+
self.assertEqual(len(q_cu), 1)
47+
self.assertTrue(torch.equal(q_cu[0], torch.tensor([0, 0, 4], dtype=torch.int32)))
48+
self.assertTrue(torch.equal(q_cu_padded[0], torch.tensor([0, 2, 6], dtype=torch.int32)))
49+
self.assertTrue(torch.equal(kv_cu[0], torch.tensor([0, 6, 10], dtype=torch.int32)))
50+
51+
def test_no_load_balance_restore_uses_captured_mode(self):
52+
tokens = torch.arange(8)
53+
cu_seqlens_padded = torch.tensor([0, 8], dtype=torch.int32)
54+
55+
restored = restore_thd_gathered_kv(
56+
tokens,
57+
cu_seqlens_padded,
58+
2,
59+
CPLoadBalancingStrategy.NO_LOAD_BALANCE,
60+
)
61+
unrestored = unrestore_thd_gathered_kv(
62+
tokens,
63+
cu_seqlens_padded,
64+
2,
65+
CPLoadBalancingStrategy.NO_LOAD_BALANCE,
66+
)
67+
68+
self.assertIs(restored, tokens)
69+
self.assertIs(unrestored, tokens)
70+
71+
2472
class TestSequencePadding(unittest.TestCase):
2573
def test_padding_with_custom_padding_values_sequences_shorter_than_divisibility_factor(
2674
self,
@@ -586,6 +634,46 @@ def test_cp_rank_slicing_simple_case(self):
586634
self.assertTrue(torch.equal(labels_r1, expected_labels_r1))
587635
self.assertTrue(torch.equal(pos_ids_r1, expected_pos_ids_r1))
588636

637+
@unittest.skipUnless(torch.cuda.is_available() and tex is not None, "CUDA extension required")
638+
def test_cp_rank_slicing_dual_chunk_swap_on_cuda(self):
639+
"""CUDA inputs use the native DualChunkSwap partition indices."""
640+
input_ids = torch.arange(16, device="cuda").unsqueeze(0)
641+
labels = input_ids + 100
642+
position_ids = torch.arange(16, device="cuda")
643+
cu_seqlens = torch.tensor([0, 8, 16])
644+
645+
self._mock_distributed_env(cp_size=2, cp_rank=0)
646+
input_ids_rank, labels_rank, position_ids_rank = get_batch_on_this_cp_rank(
647+
cu_seqlens, input_ids, labels, position_ids
648+
)
649+
650+
expected_indices = torch.tensor([0, 1, 6, 7, 8, 9, 14, 15], device="cuda")
651+
self.assertTrue(torch.equal(input_ids_rank, input_ids.index_select(1, expected_indices)))
652+
self.assertTrue(torch.equal(labels_rank, labels.index_select(1, expected_indices)))
653+
self.assertTrue(
654+
torch.equal(position_ids_rank, position_ids.index_select(0, expected_indices))
655+
)
656+
657+
def test_cp_rank_slicing_no_load_balance_on_cpu(self):
658+
"""The experimental policy assigns one contiguous CPU chunk per rank."""
659+
input_ids = torch.arange(12).unsqueeze(0)
660+
labels = input_ids + 100
661+
position_ids = torch.arange(12)
662+
cu_seqlens = torch.tensor([0, 5, 12])
663+
664+
self._mock_distributed_env(cp_size=4, cp_rank=2)
665+
input_ids_rank, labels_rank, position_ids_rank = get_batch_on_this_cp_rank(
666+
cu_seqlens,
667+
input_ids,
668+
labels,
669+
position_ids,
670+
load_balancing_strategy=CPLoadBalancingStrategy.NO_LOAD_BALANCE,
671+
)
672+
673+
self.assertTrue(torch.equal(input_ids_rank, torch.tensor([[6, 7, 8]])))
674+
self.assertTrue(torch.equal(labels_rank, torch.tensor([[106, 107, 108]])))
675+
self.assertTrue(torch.equal(position_ids_rank, torch.tensor([6, 7, 8])))
676+
589677
def test_cp_rank_slicing_multiple_sequences(self):
590678
"""Test CP rank slicing with multiple sequences."""
591679
# Setup: Two sequences of length 8 each, CP size = 2

‎transformer_engine/pytorch/__init__.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
load_framework_extension("torch")
1919
from transformer_engine.pytorch import constants
20-
from transformer_engine.pytorch.constants import DType
20+
from transformer_engine.pytorch.constants import CPLoadBalancingStrategy, DType
2121
from transformer_engine.pytorch.module import LayerNormLinear
2222
from transformer_engine.pytorch.module import Linear
2323
from transformer_engine.pytorch.module import LayerNormMLP

‎transformer_engine/pytorch/attention/dot_product_attention/backends.py‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
)
3838
from transformer_engine.pytorch.tensor.float8_tensor import Float8Tensor
3939
from transformer_engine.pytorch.constants import (
40+
CPLoadBalancingStrategy,
4041
QKVLayouts,
4142
dist_group_type,
4243
)
@@ -900,6 +901,7 @@ def forward(
900901
num_splits: Optional[int] = 1,
901902
cu_seqlens_q_padded: Optional[torch.Tensor] = None,
902903
cu_seqlens_kv_padded: Optional[torch.Tensor] = None,
904+
load_balancing_strategy=CPLoadBalancingStrategy.DUAL_CHUNK_SWAP,
903905
) -> torch.Tensor:
904906
"""flash-attn fprop"""
905907

@@ -1134,6 +1136,7 @@ def forward(
11341136
pad_between_seqs=pad_between_seqs,
11351137
use_flash_attn_3=use_flash_attn_3,
11361138
fp8_output=fp8_output,
1139+
load_balancing_strategy=load_balancing_strategy,
11371140
)
11381141
else:
11391142
if is_cpu_offload_enabled():
@@ -2116,6 +2119,7 @@ def forward(
21162119
packed_qkv: Optional[torch.Tensor] = None,
21172120
packed_kv: Optional[torch.Tensor] = None,
21182121
bf16_backward: bool = False,
2122+
load_balancing_strategy=CPLoadBalancingStrategy.DUAL_CHUNK_SWAP,
21192123
) -> torch.Tensor:
21202124
"""fused attention fprop"""
21212125
assert (
@@ -2282,6 +2286,7 @@ def forward(
22822286
fp8_output=fp8_output,
22832287
layer_number=self.layer_number,
22842288
return_max_logit=self.return_max_logit,
2289+
load_balancing_strategy=load_balancing_strategy,
22852290
)
22862291
elif score_mod is not None:
22872292
output = FusedAttentionWithScoreModFunc.apply(

0 commit comments

Comments
 (0)