Skip to content

Commit 383cd54

Browse files
Apply MIMO SP/CP sharding with explicit groups and thread loss mask
Make MIMO language-model input sharding correct for non-colocated / heterogeneous parallelism, where the language model owns its own process-group grid. PartitionAdapter.shard() owns a single layout contract: - Sequence-first input. shard() consumes (S, B, H) embeddings -- the layout align_embeddings_by_token_positions produces and the LM consumes -- and returns (S/(cp*tp), B, H). It transposes to batch-first only inside the CP block (get_batch_on_this_cp_rank requires batch-first); the SP-only path scatters dim 0 directly with no transpose round-trip. - Explicit TP group. scatter_to_sequence_parallel_region is called with the adapter's tp_group instead of the global parallel_state group. - Loss mask threaded through the LM forward and CP-sharded alongside labels, so the terminal per-token loss aligns with the CP-local hidden states. MimoModel: - Build the adapter only on language-module ranks (encoder-only ranks never shard and must not read process groups they do not own). - Thread packing_kwargs/packed_seq_params through the non-colocated language path so packed (THD) sequences are CP-sharded and reach the LM, matching the colocated path (shared _build_packed_seq_params helper). - Reject a dense attention_mask under context parallelism: a dense [B, S] mask cannot line up with the CP-sharded sequence, so MIMO masks via a causal attn_mask_type or packed_seq_params. position_ids are passed full-length, as required by mRoPE (the rotary module CP-shards positions internally). Tests: real 8-GPU sharding tests (SP-only, CP-only, CP+SP) assert per-rank shape and content; plus the attention-mask-under-CP guard, loss-mask CP-sharding, and packed/THD handling. 122 passed, 2 skipped on 1 node x 8 GPUs. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent 669b747 commit 383cd54

4 files changed

Lines changed: 475 additions & 265 deletions

File tree

‎megatron/core/models/mimo/model/base.py‎

Lines changed: 124 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
import logging
44
import warnings
5-
from typing import Any, Dict, Optional
5+
from typing import Any, Dict, Optional, Tuple
66

77
import torch
88

@@ -77,8 +77,11 @@ def __init__(self, mimo_config: MimoModelConfig, cp_group=None, tp_group=None) -
7777
max_seq_len = mimo_config.language_model_spec.params.get('max_sequence_length', 4096)
7878

7979
self.partition_adapter: Optional[PartitionAdapter] = None
80-
# Create partition adapter only if parallelism is enabled
81-
if language_config.context_parallel_size > 1 or language_config.sequence_parallel:
80+
# Only on language-module ranks: encoder-only ranks never shard and would read
81+
# process groups they do not own.
82+
if self.role.has_language_module and (
83+
language_config.context_parallel_size > 1 or language_config.sequence_parallel
84+
):
8285
partition_config = PartitionConfig.from_mp_config(
8386
mp=language_config,
8487
max_seq_len=max_seq_len,
@@ -307,10 +310,24 @@ def get_text_embeddings(
307310
else:
308311
position_ids_text = position_ids[batch_idx, seq_idx].unsqueeze(0)
309312

310-
text_embeddings = (
311-
unwrap_model(self.language_model)
312-
.embedding(input_ids=input_ids_text, position_ids=position_ids_text)
313-
.squeeze(1)
313+
embedding_layer = unwrap_model(self.language_model).embedding
314+
# Combined embeddings are SP-scattered later in PartitionAdapter; a second scatter
315+
# here would split the flat text tokens across TP ranks before alignment.
316+
if (
317+
self.partition_adapter is not None
318+
and self.partition_adapter.cfg.seq_parallel
319+
and getattr(embedding_layer, 'scatter_to_sequence_parallel', False)
320+
):
321+
raise RuntimeError(
322+
"MIMO sequence parallelism requires the language embedding scatter to be "
323+
"disabled; pass scatter_embedding_sequence_parallel=False when constructing "
324+
"the language model."
325+
)
326+
327+
text_embeddings = embedding_layer(
328+
input_ids=input_ids_text, position_ids=position_ids_text
329+
).squeeze(
330+
1
314331
) # Shape: [num_text_tokens, hidden_dim]
315332
return text_embeddings
316333

@@ -382,11 +399,14 @@ def forward(
382399
return self._forward_encoders(input_ids, modality_inputs, input_tensors), loss_mask
383400

384401
if self.role.has_language_module:
385-
return (
386-
self._forward_language_module(
387-
input_ids, position_ids, attention_mask, labels, input_tensors
388-
),
402+
return self._forward_language_module(
403+
input_ids,
404+
position_ids,
405+
attention_mask,
389406
loss_mask,
407+
labels,
408+
input_tensors,
409+
packing_kwargs,
390410
)
391411

392412
raise RuntimeError(f"Rank has no modules assigned in role: {self.role}")
@@ -497,28 +517,86 @@ def _empty_encoder_output(self, encoder_name: str) -> torch.Tensor:
497517
requires_grad=True,
498518
)
499519

520+
def _build_packed_seq_params(self, packing_kwargs: Optional[dict]) -> Optional[PackedSeqParams]:
521+
"""Build THD ``PackedSeqParams`` from ``packing_kwargs`` (None if not packing)."""
522+
if packing_kwargs is None:
523+
return None
524+
for key in packing_kwargs:
525+
if 'cu_seqlens' in key and packing_kwargs[key] is not None:
526+
packing_kwargs[key] = packing_kwargs[key].to(dtype=torch.int32)
527+
packed_seq_params = PackedSeqParams(**packing_kwargs)
528+
packed_seq_params.qkv_format = 'thd'
529+
return packed_seq_params
530+
531+
def _shard_language_inputs(
532+
self,
533+
embeddings: Optional[torch.Tensor],
534+
labels: Optional[torch.Tensor],
535+
loss_mask: Optional[torch.Tensor],
536+
packed_seq_params: Optional[PackedSeqParams] = None,
537+
) -> Tuple[
538+
Optional[torch.Tensor],
539+
Optional[torch.Tensor],
540+
Optional[torch.Tensor],
541+
Optional[PackedSeqParams],
542+
]:
543+
"""Apply CP/SP sharding via the partition adapter, or pass through if inactive.
544+
545+
``embeddings`` are sequence-first ``(S, B, H)`` (``None`` on non-first PP stages)
546+
and come back in ``(S/(cp*tp), B, H)``; labels/loss_mask are ``(B, S)``.
547+
"""
548+
if self.partition_adapter is None:
549+
return embeddings, labels, loss_mask, packed_seq_params
550+
551+
return self.partition_adapter.shard(
552+
embeddings=embeddings,
553+
labels=labels,
554+
loss_mask=loss_mask,
555+
packed_seq_params=packed_seq_params,
556+
)
557+
500558
def _forward_language_module(
501559
self,
502560
input_ids: torch.Tensor,
503561
position_ids: Optional[torch.Tensor],
504562
attention_mask: Optional[torch.Tensor],
563+
loss_mask: Optional[torch.Tensor],
505564
labels: Optional[torch.Tensor],
506565
input_tensors: Optional[Dict[str, torch.Tensor]],
507-
) -> torch.Tensor:
566+
packing_kwargs: Optional[dict] = None,
567+
) -> Tuple[Any, Optional[torch.Tensor]]:
508568
"""Forward pass for language module on this rank.
509569
510570
Args:
511571
input_ids: Token IDs
512572
position_ids: Position IDs
513-
attention_mask: Attention mask
573+
attention_mask: Attention mask. Must be ``None`` under context parallelism
574+
(CP-local hidden states cannot line up with a dense mask); mask via a
575+
causal ``attn_mask_type`` or ``packed_seq_params`` instead.
576+
loss_mask: Loss mask for per-token loss normalization
514577
labels: Labels for loss computation
515578
input_tensors: Hidden states or embeddings from previous stage
579+
packing_kwargs: Optional kwargs to construct packed (THD) sequence params.
516580
517581
Returns:
518-
Language model output (hidden states, logits, or loss depending on stage)
582+
Tuple of (language model output, possibly CP-sharded loss mask). The
583+
output is hidden states, logits, or loss depending on the stage.
519584
"""
520585
lang_name = MIMO_LANGUAGE_MODULE_KEY
521586

587+
if (
588+
self.partition_adapter is not None
589+
and self.partition_adapter.cfg.use_cp
590+
and attention_mask is not None
591+
):
592+
raise RuntimeError(
593+
"MIMO context parallelism requires attention_mask=None; mask via a causal "
594+
"attn_mask_type or packed_seq_params (a dense mask cannot line up with the "
595+
"CP-sharded sequence)."
596+
)
597+
598+
packed_seq_params = self._build_packed_seq_params(packing_kwargs)
599+
522600
if self.role.is_first_stage(lang_name):
523601
# First stage: receive encoder embeddings, combine with text, pass to LM
524602
# Build modality embeddings dict from encoder outputs
@@ -534,13 +612,21 @@ def _forward_language_module(
534612
)
535613
modality_embeddings["text"] = text_embeddings
536614

537-
# Combine all embeddings
615+
# Combine all embeddings ([S, B, H])
538616
combined_embeddings = self.align_embeddings_by_token_positions(
539617
modality_embeddings=modality_embeddings,
540618
input_ids=input_ids,
541619
special_token_ids=self.special_token_ids,
542620
)
543621

622+
# Apply CP/SP sharding; combined_embeddings returns in [S/(cp*tp), B, H].
623+
combined_embeddings, labels, loss_mask, packed_seq_params = self._shard_language_inputs(
624+
embeddings=combined_embeddings,
625+
labels=labels,
626+
loss_mask=loss_mask,
627+
packed_seq_params=packed_seq_params,
628+
)
629+
544630
lm_output = self.language_model(
545631
# decoder_input replaces the embedding lookup, so input_ids is
546632
# unused here; position_ids is still consumed by mRoPE in models
@@ -550,9 +636,19 @@ def _forward_language_module(
550636
decoder_input=combined_embeddings,
551637
labels=labels,
552638
attention_mask=attention_mask,
639+
packed_seq_params=packed_seq_params,
553640
)
554641
else:
555-
# Non-first stage: receive hidden states from previous LM stage
642+
# Non-first stage: receive hidden states from previous LM stage.
643+
# Labels/loss_mask still need CP sharding so the loss on the last stage
644+
# lines up with the CP-local hidden states.
645+
_, labels, loss_mask, packed_seq_params = self._shard_language_inputs(
646+
embeddings=None,
647+
labels=labels,
648+
loss_mask=loss_mask,
649+
packed_seq_params=packed_seq_params,
650+
)
651+
556652
hidden_states = input_tensors.get(lang_name) if input_tensors else None
557653

558654
# Set input tensor on language model for PP (unwrap DDP to reach GPTModel)
@@ -569,13 +665,14 @@ def _forward_language_module(
569665
decoder_input=None,
570666
labels=labels,
571667
attention_mask=attention_mask,
668+
packed_seq_params=packed_seq_params,
572669
)
573670

574671
# Key output for non-last stages so schedule can route to next LM stage
575672
if not self.role.is_last_stage(lang_name):
576-
return {lang_name: lm_output}
673+
return {lang_name: lm_output}, loss_mask
577674

578-
return lm_output
675+
return lm_output, loss_mask
579676

580677
def _build_colocated_communicators(self):
581678
grid_map = self.mimo_config.module_to_grid_map
@@ -632,16 +729,7 @@ def _forward_all_modules(
632729
633730
This is the original behavior, preserved for backward compatibility.
634731
"""
635-
# If packing_kwargs is provided, construct PackedSeqParams
636-
packed_seq_params = None
637-
if packing_kwargs is not None:
638-
# Ensure correct dtype for seqlens tensors
639-
for key in packing_kwargs:
640-
if 'cu_seqlens' in key and packing_kwargs[key] is not None:
641-
packing_kwargs[key] = packing_kwargs[key].to(dtype=torch.int32)
642-
packed_seq_params = PackedSeqParams(**packing_kwargs)
643-
packed_seq_params.qkv_format = 'thd'
644-
logger.debug(f"Packed sequence parameters: {packed_seq_params}")
732+
packed_seq_params = self._build_packed_seq_params(packing_kwargs)
645733

646734
# 1. Process each modality to get embeddings
647735
modality_embeddings = {}
@@ -679,25 +767,14 @@ def _forward_all_modules(
679767
)
680768
logger.debug(f"Combined embeddings shape: {combined_embeddings.shape}")
681769

682-
# 3. If sharding is needed, apply PartitionAdapter.
683-
# combined_embeddings is [S, B, H]; transpose to [B, S, H] for shard() which expects
684-
# batch-first layout (required by get_batch_on_this_cp_rank). After CP sharding each
685-
# rank holds [B, S/cp, H]; transpose back to [S/cp, B, H] for the language model.
686-
if self.partition_adapter is not None:
687-
combined_embeddings = combined_embeddings.transpose(0, 1).contiguous() # [B, S, H]
688-
combined_embeddings, labels, loss_mask, _, packed_seq_params = (
689-
self.partition_adapter.shard(
690-
embeddings=combined_embeddings,
691-
labels=labels,
692-
loss_mask=loss_mask,
693-
attention_mask=attention_mask,
694-
packed_seq_params=packed_seq_params,
695-
)
696-
)
697-
# shard() returns embeddings in [B, S/cp, H]; transpose to [S/cp, B, H]
698-
# which is what the language model expects.
699-
if combined_embeddings is not None:
700-
combined_embeddings = combined_embeddings.transpose(0, 1).contiguous()
770+
# 3. Apply CP/SP sharding. combined_embeddings is [S, B, H] and returns
771+
# [S/(cp*tp), B, H] for the LM (the adapter handles the CP batch-first transpose).
772+
combined_embeddings, labels, loss_mask, packed_seq_params = self._shard_language_inputs(
773+
embeddings=combined_embeddings,
774+
labels=labels,
775+
loss_mask=loss_mask,
776+
packed_seq_params=packed_seq_params,
777+
)
701778

702779
# 5. Forward pass through language model
703780
lm_output = self.language_model(

0 commit comments

Comments
 (0)