Skip to content

Commit 5242bdf

Browse files
authored
Enable per-layer compile with or without MoE (#2741)
per-layer compile becomes possible after applying fully_shard at layer-level (no more moe level): #2281 `apply_compile`: consolidate apply_compile_dense and apply_compile_sparse into one. The only difference was capture_scalar_outputs which is harmless for dense models. Also removed the _run_experts_grouped_mm separate compile boundary and EP wrapper. FSDP2 + EP: `NGPU=8 MODULE=deepseek_v3 CONFIG=deepseek_v3_debugmodel ./run_train.sh --compile.enable --compile.components model,loss --parallelism.expert_parallel_degree 4` FSDP2 + EP + TP: `NGPU=8 MODULE=deepseek_v3 CONFIG=deepseek_v3_debugmodel ./run_train.sh --compile.enable --compile.components model,loss --parallelism.expert_parallel_degree 4 --parallelism.tensor_parallel_degree 2` FSDP2 + EP + TP&ETP: `NGPU=8 MODULE=deepseek_v3 CONFIG=deepseek_v3_debugmodel ./run_train.sh --compile.enable --compile.components model,loss --parallelism.expert_parallel_degree 4 --parallelism.tensor_parallel_degree 2 --parallelism.expert_tensor_parallel_degree 2` qwen3-vl: `NGPU=8 MODULE=qwen3_vl CONFIG=qwen3_vl_debugmodel_moe ./run_train.sh --compile.enable --compile.components model,loss --parallelism.expert_parallel_degree 4 --training.steps 20` H100 works <img width="610" height="173" alt="Screenshot 2026-04-02 at 16 28 08" src="https://github.com/user-attachments/assets/d60d3c40-5c64-4399-b7ea-b2914f5c5b82" /> <img width="1226" height="356" alt="Screenshot 2026-04-03 at 00 32 26" src="https://github.com/user-attachments/assets/55313369-6c48-481c-8889-1dac145b473f" /> A100 also works <img width="885" height="225" alt="Screenshot 2026-04-03 at 00 26 14" src="https://github.com/user-attachments/assets/0908fcd7-7eab-4adb-ac60-e9d4ffa200b9" />
1 parent 74485ea commit 5242bdf

13 files changed

Lines changed: 50 additions & 154 deletions

File tree

tests/unit_tests/test_compile_moe.py

Lines changed: 13 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
import torch
1010

1111
from torchtitan.config import CompileConfig
12-
from torchtitan.distributed.compile import apply_compile_sparse
12+
from torchtitan.distributed.compile import apply_compile
1313
from torchtitan.models.common.linear import Linear
1414
from torchtitan.protocols.module import Module, ModuleDict
1515

@@ -42,41 +42,32 @@ def forward(self, x):
4242

4343

4444
class TestApplyCompile(unittest.TestCase):
45-
def test_patched_once(self):
46-
"""
47-
Calls apply_compile multiple times, as in the case with PP.
48-
But patches should only happen once
49-
"""
50-
unused_model1 = TinyModel(num_layers=2, dim=128)
51-
unused_model2 = TinyModel(num_layers=2, dim=128)
52-
compile_config = CompileConfig(backend="eager")
53-
54-
apply_compile_sparse(unused_model1, compile_config, ep_enabled=True)
55-
apply_compile_sparse(unused_model2, compile_config, ep_enabled=True)
45+
@unittest.skipUnless(torch.cuda.is_available(), "requires CUDA")
46+
def test_grouped_mm_compiles_and_runs(self):
47+
model = TinyModel(num_layers=2, dim=128).cuda()
48+
compile_config = CompileConfig(backend="inductor")
49+
50+
apply_compile(model, compile_config)
5651

5752
from torchtitan.models.common import moe as moe_module
5853

59-
# Generate sample inputs for _run_experts_grouped_mm
6054
num_experts = 8
6155
dim = 128
6256
hidden_dim = 256
63-
w1 = torch.randn(num_experts, hidden_dim, dim)
64-
w2 = torch.randn(num_experts, dim, hidden_dim)
65-
w3 = torch.randn(num_experts, hidden_dim, dim)
57+
w1 = torch.randn(num_experts, hidden_dim, dim, device="cuda")
58+
w2 = torch.randn(num_experts, dim, hidden_dim, device="cuda")
59+
w3 = torch.randn(num_experts, hidden_dim, dim, device="cuda")
6660
num_tokens_per_expert = torch.tensor(
67-
[10, 8, 12, 9, 11, 7, 10, 13], dtype=torch.int32
61+
[10, 8, 12, 9, 11, 7, 10, 13], dtype=torch.int32, device="cuda"
6862
)
6963
total_tokens = num_tokens_per_expert.sum().item()
70-
x = torch.randn(total_tokens, dim)
64+
x = torch.randn(total_tokens, dim, device="cuda")
7165

72-
# Call the function, should not error
7366
output = moe_module._run_experts_grouped_mm(
7467
w1, w2, w3, x, num_tokens_per_expert
7568
)
7669

77-
print(f"Input shape: {x.shape}")
78-
print(f"Output shape: {output.shape}")
79-
print(f"Num tokens per expert: {num_tokens_per_expert}")
70+
self.assertEqual(output.shape, x.shape)
8071

8172

8273
if __name__ == "__main__":

torchtitan/distributed/compile.py

Lines changed: 4 additions & 109 deletions
Original file line numberDiff line numberDiff line change
@@ -7,12 +7,8 @@
77
import torch
88
import torch.nn as nn
99
from torch._subclasses.fake_tensor import FakeTensorMode
10-
from torch.distributed.algorithms._checkpoint.checkpoint_wrapper import (
11-
CheckpointWrapper,
12-
)
1310

1411
from torchtitan.config import CompileConfig
15-
from torchtitan.models.common import moe as moe_module
1612
from torchtitan.tools.logging import logger
1713

1814

@@ -24,43 +20,13 @@
2420
)
2521

2622

27-
def apply_compile_dense(model: nn.Module, compile_config: CompileConfig) -> None:
28-
"""
29-
Apply torch.compile to each TransformerBlock, which makes compilation efficient due to
30-
repeated structure. Alternatively one can compile the whole model (after applying DP).
31-
32-
This is for dense (non-MoE) models. It compiles each TransformerBlock as a whole.
33-
"""
34-
# Skip replaying forward side effects (e.g. RoPE cache updates) during
35-
# the AC recompute in backward. Eager AC replays the forward python
36-
# side-effects in backward, but torch.compile has no easy way to reapply
37-
# python mutations in the backward. Setting this flag accepts this eager
38-
# and compile divergence by skipping reapplication of side effects.
39-
torch._dynamo.config.skip_fwd_side_effects_in_bwd_under_checkpoint = (
40-
True # pyrefly: ignore [bad-assignment]
41-
)
42-
43-
# pyrefly: ignore [missing-attribute]
44-
for layer_id, transformer_block in model.layers.named_children():
45-
transformer_block.compile(backend=compile_config.backend, fullgraph=True)
46-
# pyrefly: ignore [missing-attribute]
47-
model.layers.register_module(layer_id, transformer_block)
48-
49-
logger.info("Compiling each TransformerBlock with torch.compile")
50-
51-
52-
def apply_compile_sparse(
53-
model: nn.Module, compile_config: CompileConfig, ep_enabled: bool
54-
) -> None:
23+
def apply_compile(model: nn.Module, compile_config: CompileConfig) -> None:
5524
"""
5625
Apply torch.compile to each TransformerBlock, which makes compilation efficient due to
5726
repeated structure. Alternatively one can compile the whole model (after applying DP).
58-
59-
This is for MoE (sparse) models. It compiles sub-modules individually to avoid
60-
graph breaks from FSDP(GroupedExperts).
6127
"""
62-
# Needed for torch.compile to avoid graph breaking on dynamic shapes in
63-
# token-choice MoE, but it is experimental.
28+
# Needed for torch.compile to handle data-dependent dynamic shapes in
29+
# token-choice MoE dispatch. Harmless for dense models.
6430
torch._dynamo.config.capture_scalar_outputs = True
6531
# Skip replaying forward side effects (e.g. RoPE cache updates) during
6632
# the AC recompute in backward. Eager AC replays the forward python
@@ -73,77 +39,6 @@ def apply_compile_sparse(
7339

7440
# pyrefly: ignore [missing-attribute]
7541
for layer_id, transformer_block in model.layers.named_children():
76-
if transformer_block.moe_enabled:
77-
# If it is a MoE layer, FSDP(GroupedExperts) will cause a graph break
78-
# So we must weave compile wrappers around those FSDP hooks to
79-
# prevent AC from falling back the whole graph to eager.
80-
# TODO: Fix Compile(AC(graph break))
81-
82-
if isinstance(transformer_block, CheckpointWrapper):
83-
# TODO: Make CheckpointWrapper a transparent wrapper
84-
# unwrap so that .named_children() works
85-
block = transformer_block._checkpoint_wrapped_module
86-
else:
87-
block = transformer_block
88-
89-
for attr_name, submod in block.named_children():
90-
assert getattr(block, attr_name) == getattr(
91-
transformer_block, attr_name
92-
)
93-
94-
if isinstance(submod, moe_module.MoE):
95-
# avoid graph breaking on the GroupedExperts' FSDP hooks
96-
# by wrapping each submod's forward instead of their __call__
97-
moe = submod
98-
for attr_name, submod in moe.named_children():
99-
if attr_name == "experts":
100-
# NOTE: We don't compile token dispatch and token combine due to an issue on B200:
101-
# https://github.com/pytorch/torchtitan/issues/1940
102-
continue
103-
submod.compile(backend=compile_config.backend, fullgraph=True)
104-
else:
105-
submod.compile(backend=compile_config.backend, fullgraph=True)
106-
else:
107-
# If it's not a MoE layer, there is no FSDP(GroupedExperts)
108-
# So we can compile the whole block
109-
transformer_block.compile(
110-
backend=compile_config.backend,
111-
fullgraph=True,
112-
)
113-
114-
# pyrefly: ignore [missing-attribute]
115-
model.layers.register_module(layer_id, transformer_block)
116-
117-
# Patch some globals only once (apply_compile_sparse is called multiple times for PP setup)
118-
already_patched = (
119-
"_run_experts_grouped_mm_dynamic"
120-
in moe_module._run_experts_grouped_mm.__qualname__
121-
)
122-
if not already_patched:
123-
moe_module._run_experts_grouped_mm = torch.compile(
124-
moe_module._run_experts_grouped_mm,
125-
backend=compile_config.backend,
126-
fullgraph=True,
127-
)
128-
129-
if ep_enabled:
130-
compiled_fn = moe_module._run_experts_grouped_mm
131-
132-
# keep function logic in sync with `already_patched` above
133-
def _run_experts_grouped_mm_dynamic(
134-
w1: torch.Tensor,
135-
w2: torch.Tensor,
136-
w3: torch.Tensor,
137-
x: torch.Tensor,
138-
num_tokens_per_expert: torch.Tensor,
139-
) -> torch.Tensor:
140-
# dynamic number of tokens in expert parallel
141-
torch._dynamo.mark_dynamic(x, 0)
142-
return compiled_fn(w1, w2, w3, x, num_tokens_per_expert)
143-
144-
moe_module._run_experts_grouped_mm = _run_experts_grouped_mm_dynamic
145-
146-
# NOTE: We don't compile for loop code path due to an issue with unbacked symints:
147-
# https://github.com/pytorch/pytorch/issues/166460
42+
transformer_block.compile(backend=compile_config.backend, fullgraph=True)
14843

14944
logger.info("Compiling each TransformerBlock with torch.compile")

torchtitan/distributed/expert_parallel.py

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -160,10 +160,14 @@ def _token_dispatch(
160160
num_tokens_per_expert_group = torch.ops._c10d_functional.wait_tensor(
161161
num_tokens_per_expert_group
162162
)
163+
# non_blocking=True is safe in eager, but under torch.compile the
164+
# async D2H transfer can race with the subsequent .tolist()/.item()
165+
# calls, producing stale values and failing unbacked-symint guards.
166+
non_blocking = not torch.compiler.is_compiling()
163167
input_splits = (
164168
num_tokens_per_expert.view(ep_degree, -1)
165169
.sum(dim=1)
166-
.to(torch.device("cpu"), non_blocking=True)
170+
.to(torch.device("cpu"), non_blocking=non_blocking)
167171
)
168172
# NOTE: this would incur a device-to-host sync
169173
output_splits = (
@@ -498,10 +502,14 @@ def _token_dispatch(
498502
num_tokens_per_expert_group = torch.ops._c10d_functional.wait_tensor(
499503
num_tokens_per_expert_group
500504
)
505+
# non_blocking=True is safe in eager, but under torch.compile the
506+
# async D2H transfer can race with the subsequent .tolist()/.item()
507+
# calls, producing stale values and failing unbacked-symint guards.
508+
non_blocking = not torch.compiler.is_compiling()
501509
input_splits = (
502510
num_tokens_per_expert.view(ep_degree, -1)
503511
.sum(dim=1)
504-
.to(torch.device("cpu"), non_blocking=True)
512+
.to(torch.device("cpu"), non_blocking=non_blocking)
505513
)
506514
# NOTE: this would incur a device-to-host sync
507515
output_splits = (

torchtitan/experiments/rl/models/parallelize.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@
2626
from torchtitan.config import ParallelismConfig
2727
from torchtitan.config.configs import CompileConfig
2828
from torchtitan.distributed import ParallelDims
29-
from torchtitan.distributed.compile import apply_compile_dense
29+
from torchtitan.distributed.compile import apply_compile
3030
from torchtitan.distributed.tensor_parallel import NoParallel
3131

3232
logger = logging.getLogger(__name__)
@@ -72,7 +72,7 @@ def parallelize_qwen3(
7272
and compile_config.enable
7373
and "model" in compile_config.components
7474
):
75-
apply_compile_dense(model, compile_config)
75+
apply_compile(model, compile_config)
7676

7777
return model
7878

torchtitan/experiments/transformers_modeling_backend/parallelize.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@
2727
)
2828
from torchtitan.distributed import ParallelDims
2929
from torchtitan.distributed.activation_checkpoint import apply_ac
30-
from torchtitan.distributed.compile import apply_compile_dense
30+
from torchtitan.distributed.compile import apply_compile
3131
from torchtitan.distributed.fsdp import get_fsdp_reshard_after_forward_policy
3232
from torchtitan.distributed.tensor_parallel import maybe_enable_async_tp, NoParallel
3333
from torchtitan.models.llama3.parallelize import disable_fsdp_gradient_division
@@ -93,7 +93,7 @@ def parallelize_hf_transformers(
9393

9494
# turn on per-TransformerBlock compile after AC wrapping and before FSDP
9595
if model_compile_enabled:
96-
apply_compile_dense(model, compile_config)
96+
apply_compile(model, compile_config)
9797

9898
dp_mesh_dim_names = (
9999
["dp_replicate", "fsdp"] if parallel_dims.dp_replicate_enabled else ["fsdp"]

torchtitan/experiments/vlm/infra/parallelize.py

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
)
2121
from torchtitan.distributed import ParallelDims
2222
from torchtitan.distributed.activation_checkpoint import apply_ac
23-
from torchtitan.distributed.compile import apply_compile_dense
23+
from torchtitan.distributed.compile import apply_compile
2424
from torchtitan.distributed.fsdp import get_fsdp_reshard_after_forward_policy
2525
from torchtitan.models.llama3.parallelize import disable_fsdp_gradient_division
2626
from torchtitan.protocols.model_converter import ModelConvertersContainer
@@ -75,8 +75,8 @@ def parallelize_vlm(
7575

7676
# turn on per-TransformerBlock compile after AC wrapping and before FSDP
7777
if compile_config.enable:
78-
apply_compile_dense(model, compile_config)
79-
apply_compile_dense(model.encoder, compile_config)
78+
apply_compile(model, compile_config)
79+
apply_compile(model.encoder, compile_config)
8080

8181
names = ["dp_replicate", "fsdp"] if parallel_dims.dp_replicate_enabled else ["fsdp"]
8282
apply_fsdp(

torchtitan/models/common/moe.py

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,10 @@ def _run_experts_for_loop(
3333

3434
# a tuple of tensors indexed by experts
3535
# each with shape (tokens_per_expert(varying), dim)
36+
# NOTE: x is not sliced because padding was removed in #2774, so
37+
# sum(num_tokens_per_expert) == x.shape[0] always holds.
3638
x_splits = torch.split(
37-
x[: sum(num_tokens_per_expert_list)],
39+
x,
3840
split_size_or_sections=num_tokens_per_expert_list,
3941
dim=0,
4042
)

torchtitan/models/deepseek_v3/parallelize.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
)
2626
from torchtitan.distributed import ParallelDims
2727
from torchtitan.distributed.activation_checkpoint import apply_ac
28-
from torchtitan.distributed.compile import apply_compile_sparse
28+
from torchtitan.distributed.compile import apply_compile
2929
from torchtitan.distributed.context_parallel import apply_cp_to_attention_module
3030
from torchtitan.distributed.tensor_parallel import maybe_enable_async_tp, NoParallel
3131
from torchtitan.models.deepseek_v3 import DeepSeekV3Model
@@ -134,7 +134,7 @@ def parallelize_deepseekv3(
134134
)
135135

136136
if model_compile_enabled:
137-
apply_compile_sparse(model, compile_config, parallel_dims.ep_enabled)
137+
apply_compile(model, compile_config)
138138

139139
dp_mesh_names = (
140140
["dp_replicate", "fsdp"] if parallel_dims.dp_replicate_enabled else ["fsdp"]

torchtitan/models/gpt_oss/parallelize.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
)
2929
from torchtitan.distributed import ParallelDims
3030
from torchtitan.distributed.activation_checkpoint import apply_ac
31-
from torchtitan.distributed.compile import apply_compile_sparse
31+
from torchtitan.distributed.compile import apply_compile
3232
from torchtitan.distributed.context_parallel import apply_cp_to_attention_module
3333
from torchtitan.distributed.expert_parallel import (
3434
ExpertParallel,
@@ -125,7 +125,7 @@ def parallelize_gptoss(
125125

126126
# turn on per-TransformerBlock compile after AC wrapping and before FSDP
127127
if model_compile_enabled:
128-
apply_compile_sparse(model, compile_config, parallel_dims.ep_enabled)
128+
apply_compile(model, compile_config)
129129

130130
dp_mesh_names = (
131131
["dp_replicate", "fsdp"] if parallel_dims.dp_replicate_enabled else ["fsdp"]

torchtitan/models/llama3/parallelize.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@
3030
)
3131
from torchtitan.distributed import ParallelDims
3232
from torchtitan.distributed.activation_checkpoint import apply_ac
33-
from torchtitan.distributed.compile import apply_compile_dense
33+
from torchtitan.distributed.compile import apply_compile
3434
from torchtitan.distributed.context_parallel import apply_cp_to_attention_module
3535
from torchtitan.distributed.fsdp import (
3636
disable_fsdp_gradient_division,
@@ -117,7 +117,7 @@ def parallelize_llama(
117117

118118
# turn on per-TransformerBlock compile after AC wrapping and before FSDP
119119
if model_compile_enabled:
120-
apply_compile_dense(model, compile_config)
120+
apply_compile(model, compile_config)
121121

122122
names = ["dp_replicate", "fsdp"] if parallel_dims.dp_replicate_enabled else ["fsdp"]
123123
dp_mesh = parallel_dims.get_mesh(names)

0 commit comments

Comments
 (0)