Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 121 additions & 29 deletions megatron/core/models/hybrid/hybrid_block.py
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,63 @@ def _inner_is_partial_moe_capture(self) -> bool:
and CudaGraphModule.moe_router in self.config.cuda_graph_modules
)

def _can_group_te_cuda_graph_with(self, next_layer: MegatronModule) -> bool:
"""Whether this attention layer and the following MoE prefix can share one TE graph.

HybridStack represents attention and MoE as separate layers, unlike the GPT path where
both scopes live in one TransformerLayer callable. Group only the matching mHC pair and
only when HybridStack executes its normal per-layer loop. Full/MHC-selective recompute
use different layer scheduling and retain the existing per-layer graphs.
"""
if not isinstance(next_layer, HyperConnectionHybridLayer):
return False
if self.config.recompute_granularity == 'full' or (
self.config.recompute_granularity == 'selective'
and 'mhc' in (self.config.recompute_modules or [])
):
return False
if not isinstance(self.inner_layer, TransformerLayer):
return False
is_attention_only = not (
isinstance(self.inner_layer.self_attention, IdentityOp)
and isinstance(self.inner_layer.cross_attention, IdentityOp)
) and isinstance(self.inner_layer.mlp, IdentityOp)
return is_attention_only and next_layer._inner_is_partial_moe_capture()

def _set_te_cuda_graph_group_tail(self, next_layer: MegatronModule) -> None:
"""Attach a non-registered capture-only tail while preserving checkpoint keys."""
assert self._can_group_te_cuda_graph_with(next_layer)
# Bypass nn.Module.__setattr__: next_layer remains registered exactly once in
# HybridStack.layers, so this capture-only reference cannot alter state_dict keys.
object.__setattr__(self, '_te_cuda_graph_group_tail', next_layer)

def _get_te_cuda_graph_group_tail(self) -> Optional['HyperConnectionHybridLayer']:
"""Return the capture-only MoE tail, if discovery grouped this layer."""
return getattr(self, '_te_cuda_graph_group_tail', None)

def parameters(self, recurse: bool = True):
"""Expose grouped-prefix parameters to TE without registering the group tail.

Transformer Engine derives a graphed callable's autograd input surface from
``callable.parameters()``. The grouped MoE tail stays registered only in
``HybridStack.layers`` for checkpoint compatibility, so include just its graph-covered
prefix parameters here. Parent model traversal and ``state_dict`` continue to use the
unchanged module hierarchy.
"""
seen = set()
for param in super().parameters(recurse=recurse):
seen.add(id(param))
yield param

group_tail = self._get_te_cuda_graph_group_tail()
if not recurse or group_tail is None:
return
for submodule in group_tail._get_submodules_under_cudagraphs():
for param in submodule.parameters():
if id(param) not in seen:
seen.add(id(param))
yield param

def _te_cuda_graph_capture(self, *args, **kwargs):
"""Capture the graph-safe portion of the wrapper forward.

Expand All @@ -222,6 +279,14 @@ def _te_cuda_graph_capture(self, *args, **kwargs):
"""
self._reconstruct_packed_seq_params_from_kwargs(kwargs)

group_tail = self._get_te_cuda_graph_group_tail()
if group_tail is not None:
hidden_states, context = self.forward(*args, **kwargs)
assert context is None, "Grouped hybrid CUDA graphs do not support cross-attention."
tail_kwargs = dict(kwargs)
tail_kwargs.pop("hidden_states", None)
return group_tail._te_cuda_graph_capture(hidden_states, **tail_kwargs)

if self._inner_is_partial_moe_capture():
hidden_states = args[0] if args else kwargs["hidden_states"]
aggregated, h_res, h_post, residual = self.hyper_connection(hidden_states)
Expand Down Expand Up @@ -256,39 +321,47 @@ def _te_cuda_graph_replay(self, *args, **kwargs):
"""
self._decompose_packed_seq_params_to_kwargs(kwargs)

group_tail = self._get_te_cuda_graph_group_tail()
if group_tail is not None:
out = list(super()._te_cuda_graph_replay(*args, **kwargs))
return group_tail._resume_partial_moe_cuda_graph(out)

if self._inner_is_partial_moe_capture():
out = list(super()._te_cuda_graph_replay(*args, **kwargs))
residual = out.pop() # n-stream [s, b, n*C] — graph output (see capture)
h_res = out.pop()
h_post = out.pop()
# Resume the inner MoE experts eagerly to the raw delta (mlp_output_with_bias),
# then let the n-stream BDA own the residual — identical to the eager forward
# (`_call_inner_transformer_layer_without_local_bda` fast path →
# fused_h_res_h_post_bda), just with the router/preprocess prefix graphed.
# Mirror the eager fast path's BDA args (it feeds the inner layer's
# `hidden_dropout` / `bias_dropout_fusion`) so replay == eager bit-for-bit.
mlp_output_with_bias = self.inner_layer.resume_moe_experts_after_partial_cudagraph(out)
hidden_states = self.hyper_connection.fused_h_res_h_post_bda(
h_res,
residual,
h_post,
mlp_output_with_bias,
dropout_prob=self.inner_layer.hidden_dropout,
training=self.training,
fused=self.inner_layer.config.bias_dropout_fusion,
manager=None,
)
if (
self.config.fp32_residual_connection
and self.config.params_dtype is not None
and hidden_states.dtype != self.config.params_dtype
):
hidden_states = hidden_states.to(self.config.params_dtype)
return hidden_states, None
return self._resume_partial_moe_cuda_graph(out)

cuda_graph_output = list(super()._te_cuda_graph_replay(*args, **kwargs))
return cuda_graph_output[0], None

def _resume_partial_moe_cuda_graph(self, out: List[Tensor]) -> Tuple[Tensor, Optional[Tensor]]:
"""Run the eager expert/BDA tail from captured router/preprocess outputs."""
assert self._inner_is_partial_moe_capture()
residual = out.pop() # n-stream [s, b, n*C] — graph output (see capture)
h_res = out.pop()
h_post = out.pop()
# Resume the inner MoE experts eagerly to the raw delta (mlp_output_with_bias),
# then let the n-stream BDA own the residual — identical to the eager forward
# (`_call_inner_transformer_layer_without_local_bda` fast path →
# fused_h_res_h_post_bda), just with the router/preprocess prefix graphed.
mlp_output_with_bias = self.inner_layer.resume_moe_experts_after_partial_cudagraph(out)
hidden_states = self.hyper_connection.fused_h_res_h_post_bda(
h_res,
residual,
h_post,
mlp_output_with_bias,
dropout_prob=self.inner_layer.hidden_dropout,
training=self.training,
fused=self.inner_layer.config.bias_dropout_fusion,
manager=None,
)
if (
self.config.fp32_residual_connection
and self.config.params_dtype is not None
and hidden_states.dtype != self.config.params_dtype
):
hidden_states = hidden_states.to(self.config.params_dtype)
return hidden_states, None

def _get_submodules_under_cudagraphs(self):
"""Submodules whose params are driven by the wrapper graph's manual hooks.

Expand All @@ -298,8 +371,17 @@ def _get_submodules_under_cudagraphs(self):
keep their normal forward hooks.
"""
if self._inner_is_partial_moe_capture():
return [self.hyper_connection] + self.inner_layer._get_submodules_under_cudagraphs()
return super()._get_submodules_under_cudagraphs()
submodules = [
self.hyper_connection,
*self.inner_layer._get_submodules_under_cudagraphs(),
]
else:
submodules = super()._get_submodules_under_cudagraphs()

group_tail = self._get_te_cuda_graph_group_tail()
if group_tail is not None:
submodules += group_tail._get_submodules_under_cudagraphs()
return submodules

def mamba_state_shapes_per_request(self) -> Optional[Tuple[Tuple[int], Tuple[int]]]:
"""Delegate Mamba inference state shape requests to the wrapped layer."""
Expand Down Expand Up @@ -946,7 +1028,12 @@ def get_inner_quant_context(config, layer_number):
use_inner_quantization_context=(use_inner_fp8_context or use_fp4_context),
)
else:
grouped_tail_to_skip = None
for l_no, layer in enumerate(self.layers):
if layer is grouped_tail_to_skip:
grouped_tail_to_skip = None
continue

# Layers have 1-indexed layer numbers attribute.
inner_quant_context = get_inner_quant_context(
self.config, layer.layer_number - 1
Expand Down Expand Up @@ -984,6 +1071,11 @@ def get_inner_quant_context(config, layer_number):
packed_seq_params=packed_seq_params,
)

if isinstance(layer, HyperConnectionHybridLayer) and getattr(
Comment thread
FDecaYed marked this conversation as resolved.
Resolved
layer, 'cuda_graphs', None
):
grouped_tail_to_skip = layer._get_te_cuda_graph_group_tail()

# The attention layer (currently a simplified transformer layer)
# outputs a tuple of (hidden_states, context). Context is intended
# for cross-attention, and is not needed in our model.
Expand Down
80 changes: 64 additions & 16 deletions megatron/core/transformer/cuda_graphs.py
Original file line number Diff line number Diff line change
Expand Up @@ -1701,6 +1701,55 @@ def _layer_is_graphable(layer, config):
return False


def _get_graphable_te_callables(layers, config):
"""Return graphable layers, grouping a supported adjacent hybrid attention/MoE pair."""
graphable_layers = []
layer_number = 0
while layer_number < len(layers):
layer = layers[layer_number]
if not _layer_is_graphable(layer, config):
layer_number += 1
continue

if layer_number + 1 < len(layers):
next_layer = layers[layer_number + 1]
can_group = getattr(layer, '_can_group_te_cuda_graph_with', None)
if (
can_group is not None
and _layer_is_graphable(next_layer, config)
and can_group(next_layer)
):
layer._set_te_cuda_graph_group_tail(next_layer)
Comment thread
FDecaYed marked this conversation as resolved.
Resolved
graphable_layers.append(layer)
layer_number += 2
continue

graphable_layers.append(layer)
layer_number += 1
return graphable_layers


def _get_mtp_te_callables(mtp_model_layer, config):
"""Expose graphable layers inside a hybrid MTP stack; GPT MTP remains one callable."""
from megatron.core.models.hybrid.hybrid_block import HybridStack

layers = (
mtp_model_layer.layers if isinstance(mtp_model_layer, HybridStack) else [mtp_model_layer]
)
return _get_graphable_te_callables(layers, config)


def _is_mtp_te_callable(layer, chunk_with_decoder):
"""Whether a callable is an MTP wrapper or a layer nested in a hybrid MTP stack."""
for mtp_layer in getattr(getattr(chunk_with_decoder, 'mtp', None), 'layers', []):
mtp_model_layer = mtp_layer.mtp_model_layer
if layer is mtp_model_layer or any(
layer is inner_layer for inner_layer in getattr(mtp_model_layer, 'layers', [])
):
return True
return False


class TECudaGraphHelper:
"""
Helper class to capture CUDA Graphs using TE make_graphed_callables().
Expand Down Expand Up @@ -1790,20 +1839,16 @@ def _discover_layers(self):
num_mtp_layers = len(chunk_with_decoder.mtp.layers)
else:
num_mtp_layers = 0
num_graphable_layers = 0
callables, callables_is_mtp = [], []
for layer_number in range(num_decoder_layers):
layer = chunk_with_decoder.decoder.layers[layer_number]
if _layer_is_graphable(layer, self.config):
num_graphable_layers += 1
callables.append(layer)
callables_is_mtp.append(False)
callables = _get_graphable_te_callables(
chunk_with_decoder.decoder.layers, self.config
)
callables_is_mtp = [False] * len(callables)
for layer_number in range(num_mtp_layers):
layer = chunk_with_decoder.mtp.layers[layer_number].mtp_model_layer
if _layer_is_graphable(layer, self.config):
num_graphable_layers += 1
callables.append(layer)
callables_is_mtp.append(True)
mtp_model_layer = chunk_with_decoder.mtp.layers[layer_number].mtp_model_layer
mtp_callables = _get_mtp_te_callables(mtp_model_layer, self.config)
callables.extend(mtp_callables)
callables_is_mtp.extend([True] * len(mtp_callables))
num_graphable_layers = len(callables)
log_on_each_pipeline_stage(
logger=logger,
tp_group=self.tp_group,
Expand Down Expand Up @@ -1924,8 +1969,8 @@ def _get_layer_static_inputs(layer, chunk_of_the_layer):
"""
Get the static inputs for a layer.
"""
assert layer in chunk_of_the_layer.decoder.layers or any(
layer is mtp_layer.mtp_model_layer for mtp_layer in chunk_of_the_layer.mtp.layers
assert layer in chunk_of_the_layer.decoder.layers or _is_mtp_te_callable(
layer, chunk_of_the_layer
), "Layer is not in the chunk"

def get_rotary_pos_emb(transformer_module, transformer_input):
Expand Down Expand Up @@ -2811,7 +2856,10 @@ def set_current_microbatch(model, microbatch_id):
assert hasattr(
layer, 'mtp_model_layer'
), f"MTP layer {layer} must have 'mtp_model_layer' attribute"
layer.mtp_model_layer.current_microbatch = microbatch_id
mtp_model_layer = layer.mtp_model_layer
mtp_model_layer.current_microbatch = microbatch_id
for inner_layer in getattr(mtp_model_layer, 'layers', []):
inner_layer.current_microbatch = microbatch_id

# Also set current_microbatch on vision encoder layers so that
# _te_cuda_graph_replay selects the correct graph index. Without this,
Expand Down
80 changes: 80 additions & 0 deletions tests/unit_tests/transformer/test_cuda_graphs.py
Original file line number Diff line number Diff line change
Expand Up @@ -911,6 +911,86 @@ def test_mhc_hybrid_layers_are_te_cudagraph_capturable(self):
)


class TestHybridTECudaGraphDiscovery:
def test_hybrid_mtp_layers_are_flattened_and_adjacent_layers_are_grouped(self, monkeypatch):
from megatron.core.transformer import cuda_graphs

class FakeGraphLayer(torch.nn.Module):
def __init__(self, group_with_next=False, graphable=True):
super().__init__()
self.group_with_next = group_with_next
self.graphable = graphable
self.group_tail = None

def _can_group_te_cuda_graph_with(self, next_layer):
return self.group_with_next and next_layer.graphable

def _set_te_cuda_graph_group_tail(self, next_layer):
self.group_tail = next_layer

head = FakeGraphLayer(group_with_next=True)
tail = FakeGraphLayer()
eager = FakeGraphLayer(graphable=False)
mtp_stack = HybridStack.__new__(HybridStack)
torch.nn.Module.__init__(mtp_stack)
mtp_stack.layers = torch.nn.ModuleList([head, tail, eager])

monkeypatch.setattr(
cuda_graphs, '_layer_is_graphable', lambda layer, config: layer.graphable
)
callables = cuda_graphs._get_mtp_te_callables(mtp_stack, object())

assert callables == [head]
assert head.group_tail is tail

gpt_mtp_layer = FakeGraphLayer()
assert cuda_graphs._get_mtp_te_callables(gpt_mtp_layer, object()) == [gpt_mtp_layer]

class Holder:
pass

mtp_layer = Holder()
mtp_layer.mtp_model_layer = mtp_stack
chunk = Holder()
chunk.mtp = Holder()
chunk.mtp.layers = [mtp_layer]
assert cuda_graphs._is_mtp_te_callable(head, chunk)
assert cuda_graphs._is_mtp_te_callable(tail, chunk)
assert not cuda_graphs._is_mtp_te_callable(eager, Holder())

def test_capture_group_tail_does_not_change_module_registration(self):
from megatron.core.transformer.identity_op import IdentityOp

class Config:
recompute_granularity = None
recompute_modules = []

inner = TransformerLayer.__new__(TransformerLayer)
torch.nn.Module.__init__(inner)
inner.self_attention = torch.nn.Linear(2, 2)
inner.cross_attention = IdentityOp()
inner.mlp = IdentityOp()

head = HyperConnectionHybridLayer.__new__(HyperConnectionHybridLayer)
torch.nn.Module.__init__(head)
head.config = Config()
head.inner_layer = inner

tail = HyperConnectionHybridLayer.__new__(HyperConnectionHybridLayer)
torch.nn.Module.__init__(tail)
tail.capture_weight = torch.nn.Parameter(torch.ones(1))
object.__setattr__(tail, '_inner_is_partial_moe_capture', lambda: True)
object.__setattr__(tail, '_get_submodules_under_cudagraphs', lambda: [tail])

state_dict_keys = tuple(head.state_dict())
head._set_te_cuda_graph_group_tail(tail)

assert head._get_te_cuda_graph_group_tail() is tail
assert '_te_cuda_graph_group_tail' not in head._modules
assert tuple(head.state_dict()) == state_dict_keys
assert any(param is tail.capture_weight for param in head.parameters())


# Global storage for comparing unique buffer counts across different num_microbatches,
# keyed by (pp_size, vpp_size)
_unique_buffer_counts = {}
Expand Down
Loading