Skip to content

Commit 370c11f

Browse files
Address review: gate the MTP freeze on skip_lm_loss, agree on the EP guard across ranks
- Freeze the MTP head only when its loss is actually skipped (skip_lm_loss=True); with skip_lm_loss=False it still trains. - Reject SequentialMLP experts at EP>1 in an all-rank preflight, so stages without them don't block in the next collective; compute the writer predicate after the IQ guards. - Add regressions for the F32 router bias, the vision exclude_modules entry, a source generation_config that save_pretrained rejects, and MLA head_dim. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Signed-off-by: Keval Morabia <28916987+kevalmorabia97@users.noreply.github.com>
1 parent 905cb1b commit 370c11f

4 files changed

Lines changed: 109 additions & 26 deletions

File tree

‎modelopt/torch/distill/plugins/megatron.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -590,7 +590,7 @@ def _sharded_state_dict(self, *args, **kwargs) -> "ShardedStateDict":
590590
# An MTP head left out of quantization is exempt from that: there is no quantization
591591
# error to recover there, and its CE materialises an fp32 [seq, vocab] tensor.
592592
skip_mtp_loss = _mtp_excluded_from_quantization(model)
593-
if skip_mtp_loss:
593+
if distill_cfg.skip_lm_loss and skip_mtp_loss:
594594
# Freeze the untrained MTP head: DDP's overlapped grad reduce asserts on params with no grad.
595595
for name, param in model.named_parameters():
596596
if "mtp" in name.split("."):

‎modelopt/torch/export/unified_export_megatron.py‎

Lines changed: 23 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -321,11 +321,6 @@ def save_pretrained(
321321
# medusa_heads and eagle_module only exist in the last stage.
322322
is_last_stage_main_rank = pp_rank == pp_size - 1 and tp_rank == 0
323323
is_writer_rank = self._is_sidecar_writer_rank(is_last_stage_main_rank)
324-
# One writer per pipeline stage: other TP / DP / EP ranks hold the same layers (EP>1 ranks
325-
# hold no gathered experts at all), and writing them too would race on the same files.
326-
writes_layers = (
327-
tp_rank == 0 and get_data_parallel_rank() == 0 and get_expert_model_parallel_rank() == 0
328-
)
329324

330325
quantization_format = self._get_quantization_format(self.model)
331326
if self._any_rank_uses_iq_quantization():
@@ -345,6 +340,22 @@ def save_pretrained(
345340
"Megatron IQ1_S/IQ2_XS unified export currently requires pipeline model "
346341
"parallel size 1"
347342
)
343+
# SequentialMLP rules index experts by local position, so EP>1 would collide. Agreed across
344+
# ranks so that stages without such experts don't block in the collectives below.
345+
if (
346+
torch.distributed.is_initialized()
347+
and get_expert_model_parallel_world_size() > 1
348+
and self._any_rank(any(hasattr(m, "local_experts") for m in self.model.modules()))
349+
):
350+
raise NotImplementedError(
351+
"Export at expert parallel size > 1 needs grouped-GEMM experts; "
352+
"export SequentialMLP (--no_moe_grouped_gemm) checkpoints at EP=1."
353+
)
354+
# One writer per pipeline stage: other TP / DP / EP ranks hold the same layers (EP>1 ranks
355+
# hold no gathered experts at all), and writing them too would race on the same files.
356+
writes_layers = (
357+
tp_rank == 0 and get_data_parallel_rank() == 0 and get_expert_model_parallel_rank() == 0
358+
)
348359

349360
# Main export process
350361
layer_state_dicts = self.layer_state_dicts
@@ -747,15 +758,6 @@ def _get_transformer_layer_state_dict(self, layer, layer_id, is_mtp=False):
747758
layer.mlp.shared_experts.gate_weight, layer_id, is_mtp=is_mtp
748759
)
749760
if hasattr(layer.mlp.experts, "local_experts"):
750-
# SequentialMLP rules index experts by local position, so EP>1 would collide.
751-
if (
752-
torch.distributed.is_initialized()
753-
and get_expert_model_parallel_world_size() > 1
754-
):
755-
raise NotImplementedError(
756-
"Export at expert parallel size > 1 needs grouped-GEMM experts; "
757-
"export SequentialMLP (--no_moe_grouped_gemm) checkpoints at EP=1."
758-
)
759761
if not self.rules.get("use_packed_local_experts", False):
760762
for expert_id, expert in enumerate(layer.mlp.experts.local_experts):
761763
self.rules["local_experts.linear_fc1"](
@@ -1183,11 +1185,15 @@ def _any_rank_uses_iq_quantization(self) -> bool:
11831185
holding no IQ layer would skip the raise and then block in the next collective while its
11841186
peers exit. Agree across ranks first, mirroring ``_gather_exclude_modules``.
11851187
"""
1186-
local_uses_iq = uses_iq_quantization(self.model)
1188+
return self._any_rank(uses_iq_quantization(self.model))
1189+
1190+
@staticmethod
1191+
def _any_rank(local: bool) -> bool:
1192+
"""Whether ``local`` holds on any rank, so callers can raise everywhere or nowhere."""
11871193
if not torch.distributed.is_initialized():
1188-
return local_uses_iq
1194+
return local
11891195
per_rank = [None] * torch.distributed.get_world_size()
1190-
torch.distributed.all_gather_object(per_rank, local_uses_iq)
1196+
torch.distributed.all_gather_object(per_rank, local)
11911197
return any(per_rank)
11921198

11931199
def _get_quantization_format(self, module: torch.nn.Module):

‎tests/gpu_megatron/torch/distill/plugins/test_distill_megatron.py‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -216,7 +216,7 @@ def _test_topk_logits_kl_loss(top_k, rank, size):
216216
loss["kd_loss"].backward()
217217

218218

219-
def _test_skip_lm_loss_with_mtp(quantized, rank, size):
219+
def _test_skip_lm_loss_with_mtp(quantized, skip_lm_loss, rank, size):
220220
"""Test that skip_lm_loss only zeroes the main LM head, unless MTP is left out of quantization."""
221221
set_seed(SEED)
222222

@@ -264,7 +264,7 @@ def _test_skip_lm_loss_with_mtp(quantized, rank, size):
264264
student_model.decoder.weight_quantizer = TensorQuantizer()
265265

266266
distill_cfg = setup_distillation_config(
267-
config_or_path=DistillationConfig(skip_lm_loss=True),
267+
config_or_path=DistillationConfig(skip_lm_loss=skip_lm_loss),
268268
student_cfg=student_model.config,
269269
teacher_cfg=teacher_model.config,
270270
)
@@ -277,10 +277,11 @@ def _test_skip_lm_loss_with_mtp(quantized, rank, size):
277277
adjust_distillation_model_for_mcore(distillation_model, distill_cfg)
278278

279279
# An untrained MTP head is frozen so DDP doesn't wait for its grads.
280+
mtp_skipped = quantized and skip_lm_loss
280281
with distillation_model.hide_teacher_model():
281282
mtp_params = [p for n, p in distillation_model.named_parameters() if "mtp" in n.split(".")]
282283
assert mtp_params
283-
assert all(p.requires_grad != quantized for p in mtp_params)
284+
assert all(p.requires_grad != mtp_skipped for p in mtp_params)
284285

285286
# Intercept each call to compute_language_model_loss and record return values.
286287
recorded_losses = []
@@ -314,8 +315,8 @@ def _recording_loss(labels, logits):
314315
f"Expected {mtp_num_layers + 1} loss calls, got {len(recorded_losses)}"
315316
)
316317
for i, loss in enumerate(recorded_losses[:-1]):
317-
assert loss.any() != quantized, f"MTP head {i} loss should be zero iff MTP is skipped"
318-
assert not recorded_losses[-1].any(), "Main LM head loss should be zero with skip_lm_loss=True"
318+
assert loss.any() != mtp_skipped, f"MTP head {i} loss should be zero iff MTP is skipped"
319+
assert recorded_losses[-1].any() != skip_lm_loss, "Main LM head loss is zero iff skip_lm_loss"
319320

320321

321322
def test_logits_kl_loss(dist_workers):
@@ -328,10 +329,12 @@ def test_topk_logits_kl_loss(dist_workers, top_k: int = 5):
328329
dist_workers.run(partial(_test_topk_logits_kl_loss, top_k))
329330

330331

331-
@pytest.mark.parametrize("quantized", [False, True])
332-
def test_skip_lm_loss_with_mtp(dist_workers, quantized):
332+
@pytest.mark.parametrize(
333+
("quantized", "skip_lm_loss"), [(False, True), (True, True), (True, False)]
334+
)
335+
def test_skip_lm_loss_with_mtp(dist_workers, quantized, skip_lm_loss):
333336
"""Test that skip_lm_loss only zeroes the main LM head, unless MTP is left out of quantization."""
334-
dist_workers.run(partial(_test_skip_lm_loss_with_mtp, quantized))
337+
dist_workers.run(partial(_test_skip_lm_loss_with_mtp, quantized, skip_lm_loss))
335338

336339

337340
def test_mtp_excluded_from_quantization():

‎tests/gpu_megatron/torch/export/test_unified_export_megatron.py‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
from _test_utils.torch.megatron.models import get_mcore_gpt_model, get_mcore_hybrid_model
3131
from _test_utils.torch.megatron.utils import get_forward
3232
from _test_utils.torch.transformers_models import (
33+
create_tiny_deepseek_v3_dir,
3334
create_tiny_llama_dir,
3435
create_tiny_nemotron_dir,
3536
create_tiny_nemotron_h_dir,
@@ -340,6 +341,51 @@ def test_megatron_iq_export_rejects_tensor_parallelism(qformat):
340341
exporter.save_pretrained("unused", "unused")
341342

342343

344+
def test_megatron_export_rejects_sequential_experts_at_ep():
345+
"""SequentialMLP experts are numbered by local position, so EP>1 export must refuse them."""
346+
experts = torch.nn.Module()
347+
experts.local_experts = torch.nn.ModuleList()
348+
exporter = object.__new__(GPTModelExporter)
349+
exporter.model = torch.nn.Sequential(experts)
350+
351+
with (
352+
patch.object(exporter, "_is_sidecar_writer_rank", return_value=False),
353+
patch.object(GPTModelExporter, "_any_rank", staticmethod(lambda local: local)),
354+
patch.object(uem.torch.distributed, "is_initialized", return_value=True),
355+
patch.object(uem, "get_expert_model_parallel_world_size", return_value=2),
356+
patch.object(uem, "get_pipeline_model_parallel_rank", return_value=0),
357+
patch.object(uem, "get_pipeline_model_parallel_world_size", return_value=1),
358+
patch.object(uem, "get_tensor_model_parallel_rank", return_value=0),
359+
pytest.raises(NotImplementedError, match="grouped-GEMM experts"),
360+
):
361+
exporter.save_pretrained("unused", "unused")
362+
363+
364+
def _test_mla_export_keeps_hf_head_dim(model_dir, rank, size):
365+
model = get_mcore_gpt_model(
366+
tensor_model_parallel_size=size,
367+
pipeline_model_parallel_size=1,
368+
initialize_megatron=True,
369+
num_layers=2,
370+
hidden_size=128,
371+
num_attention_heads=2,
372+
vocab_size=128,
373+
max_sequence_length=128,
374+
transformer_impl="transformer_engine",
375+
multi_latent_attention=True,
376+
).cuda()
377+
hf_head_dim = transformers.AutoConfig.from_pretrained(model_dir).head_dim
378+
# kv_channels is MLA's V head dim, which the exporter used to write as HF's head_dim.
379+
assert model.config.kv_channels != hf_head_dim
380+
exporter = GPTModelExporter(model, str(model_dir))
381+
assert exporter._hf_text_config.head_dim == hf_head_dim
382+
383+
384+
def test_mla_export_keeps_hf_head_dim(dist_workers_size_1, tmp_path):
385+
model_dir = create_tiny_deepseek_v3_dir(tmp_path)
386+
dist_workers_size_1.run(partial(_test_mla_export_keeps_hf_head_dim, model_dir))
387+
388+
343389
@pytest.mark.parametrize("qformat", IQ_FORMAT_NAMES)
344390
def test_megatron_iq_export_rejects_pipeline_parallelism(qformat):
345391
"""IQ packing requires PP=1 so the fused-MoE rejection reaches every rank.
@@ -370,6 +416,27 @@ def test_megatron_iq_export_rejects_pipeline_parallelism(qformat):
370416
exporter.save_pretrained("unused", "unused")
371417

372418

419+
def _verify_exported_metadata(export_dir: Path, model_type, quant_config, extra_module):
420+
# The router bias decides expert routing: FP32 like Megatron's buffer and HF's checkpoints.
421+
bias_dtypes = set()
422+
for shard in export_dir.glob("*.safetensors"):
423+
with safe_open(str(shard), framework="pt") as f:
424+
for key in f.keys(): # noqa: SIM118 - safe_open is not iterable
425+
if key.endswith(("expert_bias", "e_score_correction_bias")):
426+
bias_dtypes.add(f.get_slice(key).get_dtype())
427+
assert bias_dtypes <= {"F32"}, bias_dtypes
428+
if model_type == "qwen3_moe":
429+
assert bias_dtypes
430+
if model_type == "qwen3vl" and quant_config:
431+
# The vision tower is copied through unquantized.
432+
quant = json.loads((export_dir / "hf_quant_config.json").read_text())["quantization"]
433+
assert "model.visual*" in quant["exclude_modules"]
434+
if model_type == "llama" and quant_config is None and extra_module is None:
435+
# The source generation config (top_p without do_sample) is copied through as-is.
436+
generation_config = json.loads((export_dir / "generation_config.json").read_text())
437+
assert generation_config["top_p"] == 0.5
438+
439+
373440
def _test_unified_export_megatron(
374441
tmp_path,
375442
model_type,
@@ -531,6 +598,9 @@ def _test_unified_export_megatron(
531598
if quant_config:
532599
_verify_model_quant_config(tmp_export_dir, quant_config, kv_cache_quant_cfg)
533600

601+
if rank == 0:
602+
_verify_exported_metadata(tmp_export_dir, model_type, quant_config, extra_module)
603+
534604
if rank == 0 and extra_module is None:
535605
# Names / shapes only: these Megatron weights are random, not loaded from model_dir.
536606
allow_missing = ()
@@ -596,6 +666,10 @@ def test_unified_export_megatron(
596666
):
597667
if model_type == "llama":
598668
model_dir = create_tiny_llama_dir(tmp_path)
669+
# Valid to load, but rejected by GenerationConfig.save_pretrained's validation.
670+
(model_dir / "generation_config.json").write_text(
671+
json.dumps({"top_p": 0.5, "do_sample": False})
672+
)
599673
elif model_type == "qwen3vl":
600674
model_dir = create_tiny_qwen3vl_dir(tmp_path)
601675
elif model_type == "nemotron":

0 commit comments

Comments
 (0)