diff --git a/src/maxtext/layers/decoders.py b/src/maxtext/layers/decoders.py index 5429169f8a..b28b6dcb7a 100644 --- a/src/maxtext/layers/decoders.py +++ b/src/maxtext/layers/decoders.py @@ -33,6 +33,7 @@ from maxtext.layers import mhc from maxtext.layers import normalizations from maxtext.layers import pipeline +from maxtext.layers.nnx_decoders import NNXDecoderLayer, NNXSequentialPipelineStage, NNXScannedPipelineStage from maxtext.layers import quantizations from maxtext.layers.attentions import attention_as_linen from maxtext.layers.embeddings import attend_on_embedding, embed_as_linen, positional_embedding_as_linen @@ -260,7 +261,7 @@ def __call__( slot=slot, ) if self.config.scan_layers: - inputs = inputs[0] # When scan_layers is True the decoder layers return (outputs, None). + inputs = inputs[0] # When scan_layers is True the decoder layers return (outputs, None). if self.config.scan_layers: return inputs, None # pytype: disable=bad-return-type else: @@ -305,10 +306,19 @@ def setup(self): self.decoder_layer = self.get_decoder_layers() self.norm_layer = self.get_norm_layer(num_features=self.config.emb_dim) if self.config.using_pipeline_parallelism: - pipeline_stage_module = self.get_pipeline_stage_module(self.decoder_layer) remat_policy = self.get_remat_policy() + nnx_blocks = self._get_nnx_decoder_block_classes() + + # Per-stage builder handed to create_pipeline: the pipeline transform invokes it + # once per pipeline stage, passing that stage's rngs, to construct the stage's + # decoder block(s). nnx_blocks (the selected decoder block classes) is captured from + # this setup scope; remat_policy is applied separately by create_pipeline. + def build_pipeline_stage_layers(rngs): + """Builds one pipeline stage module from the selected NNX decoder block classes.""" + return self._build_nnx_pipeline_stage(nnx_blocks, rngs) + self.pipeline_module = pipeline.create_pipeline( - config=self.config, mesh=self.mesh, layers=pipeline_stage_module, remat_policy=remat_policy + config=self.config, layers=build_pipeline_stage_layers, mesh=self.mesh, remat_policy=remat_policy ) def minimal_policy(self, with_context=False, with_quantization=False): @@ -498,6 +508,42 @@ def get_decoder_layers(self): # Default case to handle any unknown decoder block types. raise ValueError(f"Incorrect decoder_block name {self.config.decoder_block.value=}") + def _get_nnx_decoder_block_classes(self): + """Returns NNX decoder block classes for pipeline stage creation.""" + cfg = self.config + + def get_scannable(normal_cls, scannable_cls): + return [scannable_cls] if cfg.scan_layers else [normal_cls] + + layer_map = { + DecoderBlockType.DEFAULT: [NNXDecoderLayer], + DecoderBlockType.LLAMA2: [llama2.LlamaDecoderLayer], + DecoderBlockType.MISTRAL: [mistral.MistralDecoderLayer], + DecoderBlockType.MIXTRAL: [mixtral.MixtralDecoderLayer], + DecoderBlockType.GEMMA: [gemma.GemmaDecoderLayer], + DecoderBlockType.GEMMA2: [gemma2.Gemma2DecoderLayer], + DecoderBlockType.GEMMA3: [gemma3.Gemma3DecoderLayer], + DecoderBlockType.GEMMA4: get_scannable(gemma4.Gemma4DecoderLayer, gemma4.Gemma4ScannableBlock), + DecoderBlockType.GEMMA4_SMALL: [gemma4_small.Gemma4SmallDecoderLayer], + DecoderBlockType.GPT3: [gpt3.Gpt3DecoderLayer], + DecoderBlockType.GPT_OSS: get_scannable(gpt_oss.GptOssDecoderLayer, gpt_oss.GptOssScannableBlock), + DecoderBlockType.QWEN2: [qwen2.Qwen2DecoderLayer], + DecoderBlockType.QWEN3: [qwen3.Qwen3DecoderLayer], + DecoderBlockType.QWEN3_MOE: [qwen3.Qwen3MoeDecoderLayer], + DecoderBlockType.QWEN3_CUSTOM_MOE: [qwen3_custom.Qwen3CustomMoeDecoderLayer], + DecoderBlockType.QWEN3_5: get_scannable(qwen3_5.Qwen3_5DecoderLayer, qwen3_5.Qwen3_5ScannableBlock), + DecoderBlockType.QWEN3_NEXT: get_scannable(qwen3.Qwen3NextDecoderLayer, qwen3.Qwen3NextScannableBlock), + DecoderBlockType.SIMPLE: [simple_layer.SimpleDecoderLayer], + DecoderBlockType.SIMPLE_MLP: [simple_layer.SimpleMlpDecoderLayer], + DecoderBlockType.DEEPSEEK: [deepseek.DeepSeekDenseLayer, deepseek.DeepSeekMoELayer], + DecoderBlockType.LLAMA4: get_scannable(llama4.Llama4DecoderLayer, llama4.Llama4ScannableBlock), + DecoderBlockType.OLMO3: get_scannable(olmo3.Olmo3DecoderLayer, olmo3.Olmo3ScannableBlock), + } + + if cfg.decoder_block not in layer_map: + raise ValueError(f"Incorrect decoder_block name {cfg.decoder_block.value=}") + return layer_map[cfg.decoder_block] + def set_remat_policy(self, block_layers, policy): """Set remat policy""" RemattedBlockLayers = [] @@ -526,6 +572,58 @@ def map_fn(path, value): RemattedBlockLayers.append(layer) return RemattedBlockLayers + def _build_nnx_pipeline_stage(self, decoder_blocks, rngs): + """Creates a single NNX pipeline stage module.""" + cfg = self.config + base_stage_cls = decoder_blocks[1] if cfg.decoder_block == DecoderBlockType.DEEPSEEK else decoder_blocks[0] + + if cfg.num_layers_per_pipeline_stage == 1: + return base_stage_cls(config=cfg, mesh=self.mesh, quant=self.quant, model_mode=self.model_mode, rngs=rngs) + elif cfg.scan_layers_per_stage: + return NNXScannedPipelineStage( + base_stage_cls, cfg.num_layers_per_pipeline_stage, cfg, self.mesh, self.quant, self.model_mode, rngs=rngs + ) + return NNXSequentialPipelineStage( + base_stage_cls, cfg.num_layers_per_pipeline_stage, cfg, self.mesh, self.quant, self.model_mode, rngs=rngs + ) + + def get_pipeline_stage_module(self, decoder_blocks): + """get pipeline stage module""" + + def get_layer_to_pipeline(blocks, cfg): + if cfg.decoder_block == DecoderBlockType.DEEPSEEK: + return blocks[1] # return the sparse block + else: + return blocks[0] + + cfg = self.config + base_stage = get_layer_to_pipeline(decoder_blocks, cfg) + if cfg.set_remat_policy_on_layers_per_stage: + policy = self.get_remat_policy() + base_stage = self.set_remat_policy([base_stage], policy)[0] + if cfg.num_layers_per_pipeline_stage == 1: + stage_module = base_stage(config=cfg, mesh=self.mesh, quant=self.quant, model_mode=self.model_mode) + elif cfg.scan_layers_per_stage: + stage_module = self.scan_decoder_layers( + cfg, + base_stage, + cfg.num_layers_per_pipeline_stage, + "layers_per_stage", + self.mesh, + in_axes_tuple=(nn.broadcast,) * 4, + model_mode=self.model_mode, + ) + else: + stage_module = SequentialBlockDecoderLayers( + decoder_layer=base_stage, + num_decoder_layers=cfg.num_layers_per_pipeline_stage, + config=cfg, + mesh=self.mesh, + quant=self.quant, + model_mode=self.model_mode, + ) + return stage_module + def get_norm_layer(self, num_features: int): """get normalization layer (return type inherits from nn.Module)""" if self.config.decoder_block in ( @@ -586,42 +684,6 @@ def scan_decoder_layers(self, cfg, decoder_layer, length, metadata_axis_name, me config=cfg, mesh=mesh, name=metadata_axis_name, quant=self.quant, **kwargs # pytype: disable=wrong-keyword-args ) - def get_pipeline_stage_module(self, decoder_blocks): - """get pipeline stage module""" - - def get_layer_to_pipeline(blocks, cfg): - if cfg.decoder_block == DecoderBlockType.DEEPSEEK: - return blocks[1] # return the sparse block - else: - return blocks[0] - - cfg = self.config - base_stage = get_layer_to_pipeline(decoder_blocks, cfg) - if cfg.set_remat_policy_on_layers_per_stage: - policy = self.get_remat_policy() - base_stage = self.set_remat_policy([base_stage], policy)[0] - if cfg.num_layers_per_pipeline_stage == 1: - stage_module = base_stage(config=cfg, mesh=self.mesh, quant=self.quant, model_mode=self.model_mode) - elif cfg.scan_layers_per_stage: - stage_module = self.scan_decoder_layers( - cfg, - base_stage, - cfg.num_layers_per_pipeline_stage, - "layers_per_stage", - self.mesh, - in_axes_tuple=(nn.broadcast,) * 4, - ) - else: - stage_module = SequentialBlockDecoderLayers( - decoder_layer=base_stage, - num_decoder_layers=cfg.num_layers_per_pipeline_stage, - config=cfg, - mesh=self.mesh, - quant=self.quant, - model_mode=self.model_mode, - ) - return stage_module - @nn.compact def _apply_embedding( self, @@ -1301,11 +1363,17 @@ def _apply_gemma4_scanned_blocks( decoder_positions, deterministic, model_mode, - slot, - previous_chunk, - bidirectional_mask, ) + # Pass slot/previous_chunk/bidirectional_mask by keyword only (via layer_call_kwargs), + # never positionally in broadcast_args: Gemma4DecoderLayer and Gemma4ScannableBlock + # declare slot and previous_chunk in swapped order, so positional passing misroutes them. + layer_call_kwargs = { + "slot": slot, + "previous_chunk": previous_chunk, + "bidirectional_mask": bidirectional_mask, + } + if num_full_blocks > 0: ScannableBlockToLinen = gemma4.Gemma4ScannableBlockToLinen policy = self.get_remat_policy() @@ -1335,7 +1403,7 @@ def _apply_gemma4_scanned_blocks( num_of_layers=block_pattern_len, name="scanned_blocks", )( - y, *broadcast_args + y, *broadcast_args, **layer_call_kwargs ) # Process any remaining layers that don't fit into a full scanned block @@ -1349,7 +1417,7 @@ def _apply_gemma4_scanned_blocks( attention_type=attention_type, layer_idx=layer_id, ) - y = layer(y, *broadcast_args) + y = layer(y, *broadcast_args, **layer_call_kwargs) if cfg.scan_layers: y = y[0] diff --git a/src/maxtext/layers/nnx_decoders.py b/src/maxtext/layers/nnx_decoders.py index c527b2ade9..342405ec51 100644 --- a/src/maxtext/layers/nnx_decoders.py +++ b/src/maxtext/layers/nnx_decoders.py @@ -69,6 +69,7 @@ from maxtext.utils import max_logging, max_utils, maxtext_utils, sharding from maxtext.utils.maxtext_utils_nnx import nnx_ensure_scan_leading_axis from maxtext.utils.sharding import create_sharding +from maxtext.layers.pipeline import create_nnx_pipeline # ------------------------------------------------------------------------------ # The network: Decoder Definitions @@ -252,6 +253,126 @@ def deepstack_process(hidden_states, bidirectional_mask, visual_embeds): return hidden_states +class NNXSequentialPipelineStage(nnx.Module): + """Sequential unscanned series of decoder layers formatted for a single pipeline stage.""" + + def __init__( + self, + layer_cls, + num_layers: int, + config: Config, + mesh: Mesh, + quant: Quant, + model_mode: str, + *, + rngs: nnx.Rngs, + ): + self.config = config + self.scan_layers = config.scan_layers + self.num_layers = num_layers + # Dynamically assign layers with explicit string names to ensure correct PyTree paths (layers_0) + for i in range(num_layers): + layer = layer_cls(config=config, mesh=mesh, quant=quant, model_mode=model_mode, rngs=rngs) + setattr(self, f"layers_{i}", layer) + + def __call__( + self, + inputs, + decoder_segment_ids, + decoder_positions, + deterministic, + model_mode, + **kwargs, + ): + for i in range(self.num_layers): + layer = getattr(self, f"layers_{i}") + out = layer( + inputs, + decoder_segment_ids, + decoder_positions, + deterministic, + model_mode, + **kwargs, + ) + inputs = out[0] if isinstance(out, tuple) else out + if self.scan_layers: + return inputs, None + return inputs + + +class NNXScannedPipelineStage(nnx.Module): + """Scanned block of decoder layers formatted for a single pipeline stage.""" + + def __init__( + self, + layer_cls, + num_layers: int, + config: Config, + mesh: Mesh, + quant: Quant, + model_mode: str, + *, + rngs: nnx.Rngs, + ): + self.config = config + + def create_layer_fn(rng): + return layer_cls(config=config, mesh=mesh, quant=quant, model_mode=model_mode, rngs=rng) + + forked_rngs = rngs.fork(split=num_layers) + + out_axes = nnx.StateAxes({nnx.Param: config.param_scan_axis, ...: 0}) + self.scanned_layers = nnx.vmap( + create_layer_fn, + in_axes=0, + out_axes=out_axes, + axis_name="layers_per_stage", + transform_metadata={nnx.PARTITION_NAME: "layers_per_stage"}, + )(forked_rngs) + + def __call__( + self, + inputs, + decoder_segment_ids, + decoder_positions, + deterministic, + model_mode, + **kwargs, + ): + graphdef, params, state = nnx.split(self.scanned_layers, nnx.Param, ...) + + scan_axis = self.config.param_scan_axis + if scan_axis != 0: + params = jax.tree.map(lambda x: jnp.moveaxis(x, scan_axis, 0), params) + + def layer_fn(carry, scanned_vars): + current_params, current_state = scanned_vars + layer = nnx.merge(graphdef, current_params, current_state) + layer_out = layer( + carry, + decoder_segment_ids, + decoder_positions, + deterministic, + model_mode, + **kwargs, + ) + new_carry = layer_out[0] if isinstance(layer_out, tuple) else layer_out + return new_carry, nnx.state(layer) + + final_carry, scanned_state = jax.lax.scan(layer_fn, inputs, (params, state)) + + if scan_axis != 0: + scanned_params, scanned_other = scanned_state.split(nnx.Param, ...) + scanned_params = jax.tree.map(lambda x: jnp.moveaxis(x, 0, scan_axis), scanned_params) + scanned_state = nnx.State.merge(scanned_params, scanned_other) + + nnx.update(self.scanned_layers, scanned_state) + + if self.config.scan_layers: + return final_carry, None + return final_carry + + class NNXDecoder(nnx.Module): """A stack of decoder layers as a part of an encoder-decoder architecture, using NNX.""" @@ -324,11 +445,86 @@ def _init_decoder_layers(self, decoder_block_classes, rngs, mesh): if config.using_pipeline_parallelism or config.scan_layers: raise ValueError("gemma4_small (Gemma4 E2B/E4B) does not support pipeline parallelism or scan_layers.") self._init_gemma4_small_layers(rngs) + elif config.using_pipeline_parallelism: + self._init_pipeline_layers(decoder_block_classes, rngs, mesh) elif config.scan_layers: self._init_scanned_layers(decoder_block_classes, rngs, mesh) else: self._init_sequential_layers(decoder_block_classes, rngs) + def _init_pipeline_layers(self, decoder_block_classes, rngs, mesh): + """Initializes decoder layers with pipeline parallelism.""" + config = self.config + assert not (config.engram_layers and self.is_deepseek), ( + "engram_layers + DeepSeek + pipeline_parallelism is not supported. " + "engram interleaving is currently only implemented in the non-pipeline path." + ) + + def build_pipeline_stage_layers(rngs): + return self._get_pipeline_stage_module(decoder_block_classes, rngs) + + self.pipeline_module = create_nnx_pipeline( + config=config, + stage_factory=build_pipeline_stage_layers, + mesh=mesh, + remat_policy=self.get_remat_policy(), + rngs=rngs, + ) + + if self.is_deepseek: + self._init_pipeline_deepseek(decoder_block_classes, rngs) + else: + self._init_pipeline_generic(decoder_block_classes, rngs) + + def _init_pipeline_deepseek(self, decoder_block_classes, rngs): + """Initializes DeepSeek dense and MoE layers outside pipeline.""" + config = self.config + assert len(decoder_block_classes) == 2 + dense_cls, moe_cls = decoder_block_classes + if config.scan_layers: + self.dense_layers = self._create_scanned_layers( + dense_cls, + length=config.first_num_dense_layers, + metadata_axis_name="dense_layers", + rngs=rngs, + ) + num_moe_outside = (config.num_decoder_layers - config.first_num_dense_layers) - config.pipeline_parallel_layers + if num_moe_outside > 0: + self.moe_layers_outside_pipeline = self._create_scanned_layers( + moe_cls, + length=num_moe_outside, + metadata_axis_name="moe_layers", + rngs=rngs, + ) + else: + self.num_dense_layers = config.first_num_dense_layers + for i in range(self.num_dense_layers): + self._create_and_register_named_layer(dense_cls, rngs, "dense_layers", i) + self.num_moe_outside_pipeline = ( + config.num_decoder_layers - config.first_num_dense_layers + ) - config.pipeline_parallel_layers + if self.num_moe_outside_pipeline > 0: + for i in range(self.num_moe_outside_pipeline): + self._create_and_register_named_layer(moe_cls, rngs, "moe_layers_outside_pipeline", i) + + def _init_pipeline_generic(self, decoder_block_classes, rngs): + """Initializes generic decoder layers outside pipeline.""" + config = self.config + remaining_layers = config.num_decoder_layers - config.pipeline_parallel_layers + if remaining_layers > 0: + base_cls = decoder_block_classes[0] + if config.scan_layers: + self.layers_outside_pipeline = self._create_scanned_layers( + base_cls, + length=remaining_layers, + metadata_axis_name="layers", + rngs=rngs, + ) + else: + self.num_layers_outside_pipeline = remaining_layers + for i in range(self.num_layers_outside_pipeline): + self._create_and_register_named_layer(base_cls, rngs, "layers_outside_pipeline", i) + def _init_scanned_layers(self, decoder_block_classes, rngs, mesh): """Initializes decoder layers with scanning (non-pipeline).""" if self.is_deepseek: @@ -574,6 +770,33 @@ def _init_gemma4_small_layers(self, rngs): setattr(self, f"layers_{lyr}", layer) self.layers.append(layer) + def _get_pipeline_stage_module(self, decoder_blocks, rngs): + """Retrieves the wrapper module formatted for single pipeline stage execution.""" + cfg = self.config + base_stage_cls = decoder_blocks[1] if self.is_deepseek else decoder_blocks[0] + + if cfg.num_layers_per_pipeline_stage == 1: + return self._create_single_layer(base_stage_cls, rngs) + elif cfg.scan_layers_per_stage: + return NNXScannedPipelineStage( + base_stage_cls, + cfg.num_layers_per_pipeline_stage, + cfg, + self.mesh, + self.quant, + self.model_mode, + rngs=rngs, + ) + return NNXSequentialPipelineStage( + base_stage_cls, + cfg.num_layers_per_pipeline_stage, + cfg, + self.mesh, + self.quant, + self.model_mode, + rngs=rngs, + ) + def _create_and_register_layer(self, layer_cls, rngs, base_name, i, **layer_kwargs): attr_name = f"{base_name}_{i}" layer = self._create_single_layer(layer_cls, rngs, **layer_kwargs) @@ -1394,195 +1617,284 @@ def __call__( if attention_metadata is not None: layer_kwargs["attention_metadata"] = attention_metadata - if self.is_gemma4_small: - y, kv_caches = self._apply_gemma4_small_layers( - y, - decoder_input_tokens, - decoder_segment_ids, - decoder_positions, - deterministic, - model_mode, - multimodal_input=multimodal_input, - kv_caches=kv_caches, - attention_metadata=attention_metadata, - previous_chunk=previous_chunk, - slot=slot, + if cfg.using_pipeline_parallelism: + logical_partition_spec = ( + self.pipeline_module.get_weight_sharding() + if (cfg.pipeline_fsdp_ag_once or cfg.pipeline_fsdp_ag_per_repeat) + else None ) - elif cfg.scan_layers: + if self.is_deepseek: - layer_kwargs = { + # Pre-pipeline: dense layers + outside-pipeline MoE layers under PP-as-DP axis rules. + ds_layer_kwargs = { "previous_chunk": previous_chunk, "slot": slot, } - - if cfg.engram_layers: - common_kwargs = { - "layer_kwargs": layer_kwargs, - "decoder_input_tokens": decoder_input_tokens, - } - - y = self._apply_interleaved_scanned_layers( - y, - "dense_layers", - 0, - cfg.first_num_dense_layers, - cfg.engram_layers, - *layer_args, - **common_kwargs, - ) - - y = self._apply_interleaved_scanned_layers( - y, - "moe_layers", - cfg.first_num_dense_layers, - cfg.num_decoder_layers, - cfg.engram_layers, - *layer_args, - **common_kwargs, - ) - else: - y, self.dense_layers, _ = self._apply_layers_sequentially( - self.dense_layers, - y, - *layer_args, - length=cfg.first_num_dense_layers, - **layer_kwargs, - ) - - num_moe = cfg.num_decoder_layers - cfg.first_num_dense_layers - - if cfg.use_batch_split_schedule: - policy = self.get_remat_policy() - mock_params = self._build_linen_params(self.moe_layers) - - if cfg.use_qwix_quantization and not cfg.use_manual_quantization: - y = deepseek_batchsplit_fp8.scan_batch_split_layers( + logical_axis_rules_pp_as_dp = sharding.logical_axis_rules_pp_act_as_dp(cfg.logical_axis_rules) + with self.mesh, nn.partitioning.axis_rules(logical_axis_rules_pp_as_dp): + if cfg.scan_layers: + if getattr(self, "dense_layers", None) is not None and cfg.first_num_dense_layers > 0: + y, self.dense_layers, _ = self._apply_layers_sequentially( + self.dense_layers, y, - mock_params, - decoder_positions, - decoder_segment_ids, - model_mode=model_mode, - mesh=self.mesh, - quant=self.quant, - cfg=cfg, - policy=policy, + *layer_args, + length=cfg.first_num_dense_layers, + **ds_layer_kwargs, ) - else: - # bf16 code path - y = deepseek_batchsplit.scan_batch_split_layers( + if hasattr(self, "moe_layers_outside_pipeline") and self.moe_layers_outside_pipeline is not None: + num_moe_outside = (cfg.num_decoder_layers - cfg.first_num_dense_layers) - cfg.pipeline_parallel_layers + y, self.moe_layers_outside_pipeline, _ = self._apply_layers_sequentially( + self.moe_layers_outside_pipeline, y, - mock_params, - decoder_positions, - mesh=self.mesh, - cfg=cfg, - num_layers=num_moe, + *layer_args, + length=num_moe_outside, + **ds_layer_kwargs, ) else: - y, self.moe_layers, _ = self._apply_layers_sequentially( - self.moe_layers, - y, - *layer_args, - length=num_moe, - **layer_kwargs, - ) - elif self.is_gemma3: - y = self._apply_gemma3_scanned_blocks( + # Unscanned: iterate registered layers by name. + for i in range(getattr(self, "num_dense_layers", 0)): + layer = getattr(self, f"dense_layers_{i}") + out = layer(y, *layer_args, **ds_layer_kwargs) + y = out[0] if isinstance(out, tuple) else out + for i in range(getattr(self, "num_moe_outside_pipeline", 0)): + layer = getattr(self, f"moe_layers_outside_pipeline_{i}") + out = layer(y, *layer_args, **ds_layer_kwargs) + y = out[0] if isinstance(out, tuple) else out + + y = self.pipeline_module( y, decoder_segment_ids, decoder_positions, deterministic, model_mode, - bidirectional_mask, - previous_chunk, - slot, + logical_partition_spec=logical_partition_spec, ) - elif self.is_gemma4: - y = self._apply_gemma4_scanned_blocks( + else: + # Standard pipeline run (non-DeepSeek, incl. Gemma4 — matches Linen decoders.py). + # Gemma4 routes through the pipeline here; _apply_gemma4_scanned_blocks is + # non-pipeline-only (its layers/layers_remainder are not built when + # pipeline parallelism is enabled). + y = self.pipeline_module( y, decoder_segment_ids, decoder_positions, deterministic, model_mode, - bidirectional_mask, - previous_chunk, - slot, + logical_partition_spec=logical_partition_spec, ) - else: - scan_length = int(cfg.num_decoder_layers / cfg.inhomogeneous_layer_cycle_interval) - if kv_caches is not None: - # Pass the kv_caches list directly to avoid copying in jnp.stack, - # which breaks vLLM PagedAttention in-place memory updates. - # The _apply_layers_sequentially function will handle it by statically unrolling. - y, self.layers, _ = self._apply_layers_sequentially( - self.layers, - y, - *layer_args, - length=scan_length, - kv_caches_stacked=kv_caches, - **layer_kwargs, - ) - # kv_caches list is updated in-place inside _apply_layers_sequentially - else: - y, self.layers, _ = self._apply_layers_sequentially( - self.layers, - y, - *layer_args, - length=scan_length, - **layer_kwargs, - ) + + # Remaining standard layers (outside the pipeline) + if hasattr(self, "layers_outside_pipeline") or hasattr(self, "num_layers_outside_pipeline"): + logical_axis_rules_pp_as_dp = sharding.logical_axis_rules_pp_act_as_dp(cfg.logical_axis_rules) + with ( + self.mesh, + nn.partitioning.axis_rules(logical_axis_rules_pp_as_dp), + ): + if cfg.scan_layers and hasattr(self, "layers_outside_pipeline"): + remaining = cfg.num_decoder_layers - cfg.pipeline_parallel_layers + y, self.layers_outside_pipeline, _ = self._apply_layers_sequentially( + self.layers_outside_pipeline, + y, + *layer_args, + length=remaining, + **layer_kwargs, + ) + elif (not cfg.scan_layers) and hasattr(self, "num_layers_outside_pipeline"): + for i in range(self.num_layers_outside_pipeline): + layer = getattr(self, f"layers_outside_pipeline_{i}") + out = layer(y, *layer_args, **layer_kwargs) + y = out[0] if isinstance(out, tuple) else out + else: - prevent_cse = maxtext_utils.should_prevent_cse_in_remat(cfg) + if self.is_gemma4_small: + y, kv_caches = self._apply_gemma4_small_layers( + y, + decoder_input_tokens, + decoder_segment_ids, + decoder_positions, + deterministic, + model_mode, + multimodal_input=multimodal_input, + kv_caches=kv_caches, + attention_metadata=attention_metadata, + previous_chunk=previous_chunk, + slot=slot, + ) + elif cfg.scan_layers: + if self.is_deepseek: + layer_kwargs = { + "previous_chunk": previous_chunk, + "slot": slot, + } - # Hoisted function to preserve XLA cache ID - def pure_layer_fn(graphdef, state_in, y_in, kv_in): + if cfg.engram_layers: + common_kwargs = { + "layer_kwargs": layer_kwargs, + "decoder_input_tokens": decoder_input_tokens, + } - if cfg.parameter_memory_host_offload: - state_in = jax.tree.map( - lambda x: jax.device_put(x, max_utils.device_space()), - state_in, + y = self._apply_interleaved_scanned_layers( + y, + "dense_layers", + 0, + cfg.first_num_dense_layers, + cfg.engram_layers, + *layer_args, + **common_kwargs, + ) + + y = self._apply_interleaved_scanned_layers( + y, + "moe_layers", + cfg.first_num_dense_layers, + cfg.num_decoder_layers, + cfg.engram_layers, + *layer_args, + **common_kwargs, + ) + else: + y, self.dense_layers, _ = self._apply_layers_sequentially( + self.dense_layers, + y, + *layer_args, + length=cfg.first_num_dense_layers, + **layer_kwargs, + ) + + num_moe = cfg.num_decoder_layers - cfg.first_num_dense_layers + + if cfg.use_batch_split_schedule: + policy = self.get_remat_policy() + mock_params = self._build_linen_params(self.moe_layers) + + if cfg.use_qwix_quantization and not cfg.use_manual_quantization: + y = deepseek_batchsplit_fp8.scan_batch_split_layers( + y, + mock_params, + decoder_positions, + decoder_segment_ids, + model_mode=model_mode, + mesh=self.mesh, + quant=self.quant, + cfg=cfg, + policy=policy, + ) + else: + # bf16 code path + y = deepseek_batchsplit.scan_batch_split_layers( + y, + mock_params, + decoder_positions, + mesh=self.mesh, + cfg=cfg, + num_layers=num_moe, + ) + else: + y, self.moe_layers, _ = self._apply_layers_sequentially( + self.moe_layers, + y, + *layer_args, + length=num_moe, + **layer_kwargs, + ) + elif self.is_gemma3: + y = self._apply_gemma3_scanned_blocks( + y, + decoder_segment_ids, + decoder_positions, + deterministic, + model_mode, + bidirectional_mask, + previous_chunk, + slot, ) + elif self.is_gemma4: + y = self._apply_gemma4_scanned_blocks( + y, + decoder_segment_ids, + decoder_positions, + deterministic, + model_mode, + bidirectional_mask, + previous_chunk, + slot, + ) + else: + scan_length = int(cfg.num_decoder_layers / cfg.inhomogeneous_layer_cycle_interval) + if kv_caches is not None: + # Pass the kv_caches list directly to avoid copying in jnp.stack, + # which breaks vLLM PagedAttention in-place memory updates. + # The _apply_layers_sequentially function will handle it by statically unrolling. + y, self.layers, _ = self._apply_layers_sequentially( + self.layers, + y, + *layer_args, + length=scan_length, + kv_caches_stacked=kv_caches, + **layer_kwargs, + ) + # kv_caches list is updated in-place inside _apply_layers_sequentially + else: + y, self.layers, _ = self._apply_layers_sequentially( + self.layers, + y, + *layer_args, + length=scan_length, + **layer_kwargs, + ) + else: + prevent_cse = maxtext_utils.should_prevent_cse_in_remat(cfg) - merged_layer = nnx.merge(graphdef, state_in) - out_y, out_kv = merged_layer(y_in, *layer_args, kv_cache=kv_in, **layer_kwargs) - return out_y, out_kv, nnx.state(merged_layer) + # Hoisted function to preserve XLA cache ID + def pure_layer_fn(graphdef, state_in, y_in, kv_in): - checkpointed_fn = jax.checkpoint(pure_layer_fn, policy=policy, prevent_cse=prevent_cse) + if cfg.parameter_memory_host_offload: + state_in = jax.tree.map( + lambda x: jax.device_put(x, max_utils.device_space()), + state_in, + ) - for lyr, layer in enumerate(self.layers): - graphdef, state = nnx.split(layer) - if kv_caches is not None: - if cfg.decoder_block in (DecoderBlockType.QWEN3_NEXT, DecoderBlockType.QWEN3_5): - if (lyr + 1) % cfg.inhomogeneous_layer_cycle_interval == 0: - kv_cache = ( - kv_caches["key_cache"][lyr], - kv_caches["value_cache"][lyr], - ) + merged_layer = nnx.merge(graphdef, state_in) + out_y, out_kv = merged_layer(y_in, *layer_args, kv_cache=kv_in, **layer_kwargs) + return out_y, out_kv, nnx.state(merged_layer) + + checkpointed_fn = jax.checkpoint(pure_layer_fn, policy=policy, prevent_cse=prevent_cse) + + for lyr, layer in enumerate(self.layers): + graphdef, state = nnx.split(layer) + if kv_caches is not None: + if cfg.decoder_block in (DecoderBlockType.QWEN3_NEXT, DecoderBlockType.QWEN3_5): + if (lyr + 1) % cfg.inhomogeneous_layer_cycle_interval == 0: + kv_cache = ( + kv_caches["key_cache"][lyr], + kv_caches["value_cache"][lyr], + ) + else: + kv_cache = None else: - kv_cache = None + kv_cache = kv_caches[lyr] else: - kv_cache = kv_caches[lyr] - else: - kv_cache = None + kv_cache = None - input_tokens = decoder_input_tokens if cfg.engram_layers else None - if input_tokens is not None: - layer_kwargs["decoder_input_tokens"] = input_tokens + input_tokens = decoder_input_tokens if cfg.engram_layers else None + if input_tokens is not None: + layer_kwargs["decoder_input_tokens"] = input_tokens - y, kv_cache, new_state = checkpointed_fn(graphdef, state, y, kv_cache) - nnx.update(layer, new_state) + y, kv_cache, new_state = checkpointed_fn(graphdef, state, y, kv_cache) + nnx.update(layer, new_state) - if kv_caches is not None and kv_cache is not None: - if cfg.decoder_block in (DecoderBlockType.QWEN3_NEXT, DecoderBlockType.QWEN3_5): - if (lyr + 1) % cfg.inhomogeneous_layer_cycle_interval == 0: - kv_caches["key_cache"][lyr] = kv_cache[0] - kv_caches["value_cache"][lyr] = kv_cache[1] - else: - kv_caches[lyr] = kv_cache + if kv_caches is not None and kv_cache is not None: + if cfg.decoder_block in (DecoderBlockType.QWEN3_NEXT, DecoderBlockType.QWEN3_5): + if (lyr + 1) % cfg.inhomogeneous_layer_cycle_interval == 0: + kv_caches["key_cache"][lyr] = kv_cache[0] + kv_caches["value_cache"][lyr] = kv_cache[1] + else: + kv_caches[lyr] = kv_cache - if deepstack_visual_embeds is not None and lyr < len(deepstack_visual_embeds): - visual_embeds = deepstack_visual_embeds[lyr] - if bidirectional_mask is not None and visual_embeds is not None: - y = deepstack_process(y, bidirectional_mask, visual_embeds) + if deepstack_visual_embeds is not None and lyr < len(deepstack_visual_embeds): + visual_embeds = deepstack_visual_embeds[lyr] + if bidirectional_mask is not None and visual_embeds is not None: + y = deepstack_process(y, bidirectional_mask, visual_embeds) assert isinstance(y, jax.Array) diff --git a/src/maxtext/layers/pipeline.py b/src/maxtext/layers/pipeline.py index 62ea52782b..07083a824e 100644 --- a/src/maxtext/layers/pipeline.py +++ b/src/maxtext/layers/pipeline.py @@ -14,7 +14,6 @@ """Pipeline layer wrapping a decoder layer(s). Supports circular pipelining.""" -import functools from typing import Any import numpy as np @@ -24,9 +23,13 @@ import jax import jax.ad_checkpoint -from flax.core import meta +from aqt.jax.v2 import aqt_tensor from flax import linen as nn -from flax.linen.spmd import LogicallyPartitioned +from flax.core import lift as flax_lift +from flax.core import scope as flax_scope +from flax import nnx +from maxtext.layers import initializers +from maxtext.layers.nnx_wrappers import is_linen_initializing, to_linen_class from maxtext.common.common_types import Config, MODEL_MODE_TRAIN, ShardMode from maxtext.utils.sharding import ( @@ -39,26 +42,22 @@ from maxtext.utils import pipeline_utils -class PipelineBase(nn.Module): - """Base module that implements shared pipelining logic across stages.""" - - config: Config - layers: nn.Module - mesh: Mesh - remat_policy: Any = None +class PipelineBase(nnx.Module): + """ + Base module that implements shared pipelining logic across stages. + Contains pure JAX and mathematical utilities. + """ - def setup(self): + def _setup_pipeline_attributes(self): """Initializes the configuration, calculating num_stages, delay, axes, and partition specs.""" self.num_stages = self.config.ici_pipeline_parallelism * self.config.dcn_pipeline_parallelism self.forwarding_delay = 2 if self.config.pipeline_delay_activation_forwarding else 1 self.pipeline_microbatch_size = self.config.micro_batch_size_to_train_on // self.config.num_pipeline_microbatches - microbatches_per_stage = self.config.num_pipeline_microbatches // self.num_stages - self.microbatches_per_stage = microbatches_per_stage + self.microbatches_per_stage = self.config.num_pipeline_microbatches // self.num_stages self.use_circ_storage = self.need_circ_storage() self.batch_axis_name = "activation_batch" self.seq_len_axis_name = "activation_length" - self.spmd_axis_name = "stage" if self.config.shard_mode == ShardMode.AUTO else None self.stages_in_logical = ("activation_stage", self.batch_axis_name, self.seq_len_axis_name, "activation_embed") @@ -172,8 +171,7 @@ def select_state_or_input(first_stage_in, shift): # Selects input (from stream_io) for stage 0, other stages get from shift (the rotated previous output) stages_in = select_state_or_input(first_stage_in, shift) - stages_in = self._maybe_shard_with_logical(stages_in, self.stages_in_logical) - return stages_in + return self._maybe_shard_with_logical(stages_in, self.stages_in_logical) def get_microbatch_and_repeat_ids(self, loop_iteration): """Gets the microbatch_ids and repeat_ids for all stages on this loop_iteration. Works for both circular and @@ -189,148 +187,18 @@ def get_pipeline_remat_policy(self): """Returns the pipeline remat policy for this pipeline.""" if self.config.remat_policy == "custom": return self.remat_policy - save_input_policy = jax.checkpoint_policies.save_only_these_names("iteration_input", "decoder_layer_input") if self.remat_policy is not None: - remat_policy = jax.checkpoint_policies.save_from_both_policies(self.remat_policy, save_input_policy) - else: - remat_policy = save_input_policy - return remat_policy - - def get_weight_sharding(self, *init_args): - """get weight sharding function for this pipeline.""" - key = jax.random.PRNGKey(0) - keys = {"params": key, "dropout": key, "aqt": key} - weights = self.init(keys, *init_args) - - def get_partition_spec(pytree): - def _is_leaf(x): - return isinstance(x, nn.spmd.LogicallyPartitioned) - - def get_partition_spec_leaf(leaf): - return leaf.get_partition_spec() - - return jax.tree.map(get_partition_spec_leaf, pytree, is_leaf=_is_leaf) - - partition_spec_with_extra_layer = get_partition_spec(weights) - logical_partition_spec = {"params": partition_spec_with_extra_layer["params"]["layers"]} - return logical_partition_spec - - def get_vmap_func_for_init(self): - """This vmap func is used to initialize the weights only on init.""" - - def func_to_vmap(body_instance, stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode): - return body_instance(stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode) - - vmap_func = nn.vmap( - func_to_vmap, - in_axes=(0, 0, 0, None, None), - spmd_axis_name=self.spmd_axis_name, - variable_axes={"params": 0, "_overwrite_with_gradient": 0}, - split_rngs={"params": self.is_initializing(), "dropout": self.config.enable_dropout}, - metadata_params={ - nn.PARTITION_NAME: "layers", - "sub_weight_split_dims_mapping": (None), - "is_initializing": self.is_initializing(), - "x_times": self.num_stages, - }, - ) - return vmap_func - - def get_main_vmap_func_for_iterations(self): - """ - Returns main stage function vmapped by number of stages. - This becomes a vmap over a single layer instance if body_instance is a single layer, - else a set of layers if body_instance is a set of layers. - """ - - def func_to_vmap( - body_instance, weights, stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode - ): - weights = meta.remove_axis( - weights, - 0, - { - nn.PARTITION_NAME: "layers", - "sub_weight_split_dims_mapping": (None,), - "is_initializing": self.is_initializing(), - "x_times": self.num_stages, - }, - ) - return body_instance.apply(weights, stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode) - - vmap_func = nn.vmap( - func_to_vmap, - in_axes=(0, 0, 0, 0, None, None), - spmd_axis_name=self.spmd_axis_name, - variable_axes={"params": 0}, - split_rngs={"params": self.is_initializing(), "dropout": self.config.enable_dropout}, - metadata_params={ - nn.PARTITION_NAME: "layers", - "sub_weight_split_dims_mapping": (None), - "is_initializing": self.is_initializing(), - "x_times": self.num_stages, - }, - ) - return vmap_func - - def _run_weight_initialization( - self, example_inputs, example_segmentation, example_position, segment_idx, position_idx, deterministic, model_mode - ): - """Runs the initialization sequence mapping layers appropriately based on pipeline settings.""" - vmap_func = self.get_vmap_func_for_init() - - if self.config.num_pipeline_repeats > 1: - vmap_func = nn.vmap( - vmap_func, - in_axes=(0, segment_idx, position_idx, None, None), - variable_axes={"params": 0, "_overwrite_with_gradient": 0, "non_trainable": 0, "hyper_params": 0}, - split_rngs={"params": True, "dropout": self.config.enable_dropout}, - metadata_params={ - nn.PARTITION_NAME: "circular_repeats", - "sub_weight_split_dims_mapping": (None,), - "is_initializing": True, - "x_times": self.config.num_pipeline_repeats, - "optimizer_dims_mapping": None, - }, - ) - example_inputs = jax.lax.broadcast(example_inputs, [self.config.num_pipeline_repeats]) - example_segmentation = ( - jax.lax.broadcast(example_segmentation, [self.config.num_pipeline_repeats]) - if example_segmentation is not None - else None - ) - example_position = ( - jax.lax.broadcast(example_position, [self.config.num_pipeline_repeats]) - if example_position is not None - else None - ) - - example_inputs = self._maybe_shard_with_logical(example_inputs, (None, None, None, None)) - stage_outputs = vmap_func( - self.layers, example_inputs, example_segmentation, example_position, deterministic, model_mode - ) - if self.config.scan_layers: - stage_outputs = stage_outputs[0] - if self.config.num_pipeline_repeats > 1: - stage_outputs = stage_outputs[0] - broadcasted_stage_outpus = jax.lax.broadcast( - stage_outputs[0], [self.config.micro_batch_size_to_train_on // self.pipeline_microbatch_size] - ) - - return jnp.reshape( - broadcasted_stage_outpus, - [self.config.micro_batch_size_to_train_on, self.config.max_target_length, self.config.emb_dim], - out_sharding=self.output_sharding, - ) + return jax.checkpoint_policies.save_from_both_policies(self.remat_policy, save_input_policy) + return save_input_policy @staticmethod - def _remove_fsdp_from_physical_partition_spec(pps): + def _remove_fsdp_from_physical_partition_spec(physical_partition_spec): """Removes 'fsdp' and 'fsdp_transpose' from physical partition spec.""" - if isinstance(pps, P): + if isinstance(physical_partition_spec, P): new_spec = [] # Iterate through each axis in the original PartitionSpec. - for axis in pps: + for axis in physical_partition_spec: if axis is None: new_spec.append(None) elif isinstance(axis, str): @@ -341,17 +209,13 @@ def _remove_fsdp_from_physical_partition_spec(pps): new_spec.append(None) elif isinstance(axis, (list, tuple)): # If the axis is a collection, filter out 'fsdp'. - new_axis = [a for a in axis if a not in ("fsdp", "fsdp_transpose")] + new_axis = [axis_element for axis_element in axis if axis_element not in ("fsdp", "fsdp_transpose")] new_spec.append(tuple(new_axis)) else: raise ValueError(f"Unsupported_axis_type: {type(axis)}") # Return a new sharding object with the modified spec. return P(*new_spec) - return pps - - -class Pipeline(PipelineBase): - """Original Pipeline implementation.""" + return physical_partition_spec def init_states(self, inputs): """Initialize components of state: state_io, shift, circular_storage and circular_storage_mover @@ -385,6 +249,7 @@ def init_states(self, inputs): state_io = jnp.reshape( inputs, (self.num_stages, self.microbatches_per_stage) + inputs.shape[1:], out_sharding=self.state_io_sharding ) + # We shard the pipeline_microbatch_size axis by data/fsdp, not num_microbatches since those are looped over. state_io = self._maybe_shard_with_logical(state_io, self.state_io_logical) @@ -406,7 +271,7 @@ def init_states(self, inputs): circ_storage = None circ_storage_mover = None - init_loop_state = { + return { "state_io": state_io, "shift": shift, "circ_storage": circ_storage, @@ -414,7 +279,6 @@ def init_states(self, inputs): "loop_iteration": 0, "prev_outputs": prev_outputs, } - return init_loop_state def shard_dim_by_stages(self, x, dim: int, physical_partition_spec: P | None, is_stage_weight: bool = False): """Shards x using the provided partition_spec, but adds the "stage" mesh axis to the existing sharding at @@ -468,10 +332,9 @@ def _gather_one(x, repeat_id): stage_weights = jax.vmap(_gather_one, in_axes=(stages_dim_in_weights, 0), out_axes=gathered_weights_stage_dim)( weights, repeat_ids ) - stage_weights = self.shard_dim_by_stages( + return self.shard_dim_by_stages( stage_weights, gathered_weights_stage_dim, physical_partition_spec=physical_partition_spec, is_stage_weight=True ) - return stage_weights def vmap_gather(self, xs, ids, ids_dim): """Use vmap to implement a stage-wise sharded gather. @@ -488,9 +351,11 @@ def vmap_gather(self, xs, ids, ids_dim): The per-stage gathered values. The shape is xs.shape but with ids_dim size replaced with [num_stages]. """ + xs = jnp.asarray(xs) + ndim = xs.ndim def _gather_one(x, i): - idx = tuple(i if d == ids_dim else slice(None) for d in range(x.ndim)) + idx = tuple(i if d == ids_dim else slice(None) for d in range(ndim)) replicated_sharding = NamedSharding(self.mesh, P()) return x.at[idx].get(out_sharding=replicated_sharding) @@ -521,8 +386,7 @@ def _rotate_right(arr): # we use +1 for right shifting stage_size = jax.lax.axis_size("stage") perm = [(i, (i + 1) % stage_size) for i in range(stage_size)] - arr = jax.lax.ppermute(arr, axis_name="stage", perm=perm) - return arr + return jax.lax.ppermute(arr, axis_name="stage", perm=perm) @jax.shard_map(mesh=self.mesh, in_specs=self.stages_in_spec, out_specs=self.stages_in_spec, check_vma=True) def _shift_right(arr): @@ -554,8 +418,7 @@ def _update_shift(output_in): # circ_storage_mover still points to the output of PREVIOUS iteration, which should aid in allowing overlapped # compute/async transfers def _rotate_right_and_update(circ_storage_mover_in, circ_storage_in): - rotated = _rotate_right(circ_storage_mover_in) - rotated = jnp.expand_dims(rotated, 1) + rotated = jnp.expand_dims(_rotate_right(circ_storage_mover_in), 1) # We rotate the pushing index into circ storage, and ensure that microbatch 0 lands in index 0 offset = ( loop_iteration - self.iterations_to_complete_first_microbatch_one_repeat() - 1 @@ -598,7 +461,7 @@ def _update_state_io(state_in, stream_slice, output, stream_buf_idx): new_state = _update_state_io(old_state_io, stream_slice, output, stream_buf_idx) - new_loop_state = { + return { "state_io": new_state, "shift": new_shift, "circ_storage": new_circ_storage, @@ -606,7 +469,6 @@ def _update_state_io(state_in, stream_slice, output, stream_buf_idx): "loop_iteration": loop_iteration + 1, "prev_outputs": new_prev_outputs, } - return new_loop_state def permute_output_micro_per_stage_dim(self, output): """ @@ -622,8 +484,103 @@ def permute_output_micro_per_stage_dim(self, output): # state_io - it will land on a different index of state_io depending on the number of iterations. microbatch_0_idx = self.iterations_to_complete_first_microbatch() % self.microbatches_per_stage permutation = (np.arange(self.microbatches_per_stage) + microbatch_0_idx) % self.microbatches_per_stage - output = output[:, permutation] - return output + return output[:, permutation] + + def realign_output_microbatches(self, output): + """Reorders the output tensor to reverse the circular shifts applied during execution. + + Because the pipeline operates circularly, the output microbatches are shifted + out of order by the time the final stage is completed. This rolls them back + into their original sequential layout. + """ + microbatch_0_idx = self.iterations_to_complete_first_microbatch() % self.microbatches_per_stage + output = jnp.roll(output, shift=-microbatch_0_idx, axis=1) + return self._maybe_shard_with_logical(output, self.state_io_logical) + + def get_weight_sharding(self, *init_args): + """Returns a pytree of logical-name PartitionSpecs mirroring the params state.""" + + state = nnx.state(self.layers, pipeline_utils.is_static_param) + + def get_spec(x): + if not isinstance(x, nnx.Variable): + # Non-VariableState leaf (e.g., nnx.Empty): treat as replicated. + return P() + # _overwrite_with_gradient variables (FP8 amax history / scales) carry no + # partition metadata; return replicated to keep the tree aligned. + if x.type.__name__ == "_overwrite_with_gradient": + return P() + # AQT QTensor values are a pytree wrapping quantized data; mirror the + # skip-list in variable_to_logically_partitioned (initializers.py:81-83). + if isinstance(x.value, aqt_tensor.QTensor): + return P() + if isinstance(x.value, nn.spmd.LogicallyPartitioned): + # Dead in the NNX-first flow; retained as a forward-compat guard in + # case a Linen-wrapped param is ever merged into this module. + return x.value.partitions + metadata = x.get_metadata() + # Try each known metadata key in order; first hit wins. + sharding = metadata.get("out_sharding") + if sharding is None: + sharding = metadata.get("sharding_names") + if sharding is None: + sharding = metadata.get("sharding") + # Already a PartitionSpec - pass through. + if isinstance(sharding, P): + return sharding + # Happy path: tuple/list of logical axis names from nnx.Param(sharding=...). + if isinstance(sharding, (tuple, list)): + return P(*sharding) + # Non-PartitionSpec wrapper with an explicit ``.spec`` attribute (kept + # for forward compatibility with future Flax wrapper types). + if sharding is not None and hasattr(sharding, "spec"): + return sharding.spec + # Fallback: replicated sharding (valid for shard_map, unlike None). + return P() + + return jax.tree.map(get_spec, state, is_leaf=lambda x: isinstance(x, nnx.Variable)) + + def get_main_vmap_func_for_iterations(self): + """Returns vmapped function that runs one pipeline iteration across stages.""" + + def func_to_vmap(graph, state, stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode): + module = nnx.merge(graph, state) + out = module(stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode) + return out, nnx.state(module) + + # Use jax.vmap instead of nnx.vmap to avoid nnx.State threading overhead + # that produces 6.3x more compile-time temp memory. nnx.vmap adds extra + # dynamic-slice/dynamic-update-slice ops for State management that inflate + # gradient buffers. jax.vmap produces identical HLO to nn.vmap. + # + # spmd_axis_name is handled manually: in EXPLICIT mode, sharding is applied + # by with_sharding_constraint calls in the pipeline. In AUTO mode, we apply + # the axis name via jax.vmap's spmd_axis_name if available (JAX 0.4.31+). + vmap_kwargs = { + "in_axes": (None, 0, 0, 0, 0, None, None), + "out_axes": (0, 0), + } + if self.spmd_axis_name is not None: + vmap_kwargs["spmd_axis_name"] = self.spmd_axis_name + return jax.vmap(func_to_vmap, **vmap_kwargs) + + @staticmethod + def _stamp_at_current_trace(weights): + """Pass each leaf through a no-op dynamic_slice so JAX creates new arrays + at the *current* trace level. This prevents trace-level mismatches when + outer-trace values (e.g. closed-over by ``jax.lax.scan``) are later fed + into ``nnx.merge`` inside the scan body. + + The operation is semantically an identity: ``x[0 : x.shape[0]]`` along + axis 0, which XLA will optimise away. + """ + + def _identity_slice(x): + if hasattr(x, "shape") and len(x.shape) > 0: + return jax.lax.dynamic_slice_in_dim(x, 0, x.shape[0], axis=0) + return x # scalars / non-array leaves pass through unchanged + + return jax.tree.map(_identity_slice, weights) def get_current_stage_weights(self, pipeline_weights, loop_iteration, physical_partition_spec=None): """ @@ -636,57 +593,163 @@ def get_current_stage_weights(self, pipeline_weights, loop_iteration, physical_p return self.get_current_repeat_from_stages( pipeline_weights, loop_iteration, physical_partition_spec=physical_partition_spec ) + # Stamp weights at the current trace level so that nnx.merge inside + # func_to_vmap does not hit a trace-level mismatch when running under + # jax.lax.scan (the weights may originate from an outer trace). + return self._stamp_at_current_trace(pipeline_weights) + + def all_gather_over_fsdp(self, variables, logical_partition_spec): + """ + all-gathers the variables over fsdp if fsdp is in the logical partition spec. + """ + if logical_partition_spec is None: + return variables + + def _gather_leaf(var, spec): + if spec is None: + return var + physical = logical_to_mesh_axes(spec, self.mesh, rules=self.config.logical_axis_rules) + no_fsdp = self._remove_fsdp_from_physical_partition_spec(physical) + sharding = NamedSharding(self.mesh, no_fsdp) + if isinstance(var, nnx.Variable): + var.value = self._maybe_shard_with_name(var.value, sharding) + return var + return self._maybe_shard_with_name(var, sharding) + + # nnx.Variable and PartitionSpec are JAX pytree nodes — treat them as leaves + # so the two trees align at the dict level. None must also be a leaf to avoid + # being treated as an empty container (0 children) vs the Variable's 1 child. + def is_leaf(x): + return isinstance(x, (nnx.Variable, P)) or x is None + + return jax.tree.map(_gather_leaf, variables, logical_partition_spec, is_leaf=is_leaf) + + def get_logical_spec_repeats_removed(self, full_logical): + """Returns a new logical spec with 'circular_repeats' removed.""" + if full_logical is None or self.config.num_pipeline_repeats == 1: + return full_logical + + def _remove_from_spec(spec): + if not isinstance(spec, P): + return spec + if spec and (spec[0] == "circular_repeats" or spec[0] is None): + return jax.sharding.PartitionSpec(*spec[1:]) + return jax.sharding.PartitionSpec(*[dim for dim in spec if dim != "circular_repeats"]) + + return jax.tree.map(_remove_from_spec, full_logical, is_leaf=lambda x: isinstance(x, P)) + + def __init__( + self, + config: Config, + stage_factory: Any, + mesh: Mesh, + remat_policy: Any = None, + *, + rngs: nnx.Rngs, + ): + self.config = config + self.mesh = mesh + self.remat_policy = remat_policy + self._setup_pipeline_attributes() + + def build_batched_rngs(shape): + kwargs = {} + rng_state = nnx.state(rngs, nnx.RngState) + leaves, _ = jax.tree_util.tree_flatten_with_path(rng_state) + for path, key in leaves: + stream_name = getattr(path[0], "key", str(path[0])) + if not jax.dtypes.issubdtype(key.dtype, jax.dtypes.prng_key): + key = jax.random.key(key) + num_splits = int(np.prod(shape)) + flat_keys = jax.random.split(key, num_splits) + kwargs[stream_name] = flat_keys.reshape(shape + key.shape) + return nnx.Rngs(**kwargs) + + def create_stage_fn(r): + stage = stage_factory(r) + # Split into (GraphDef, Param State, Rest of State) + return nnx.split(stage, nnx.Param, ...) + + vmap_stages = nnx.vmap( + create_stage_fn, + in_axes=0, + out_axes=(None, 0, 0), + spmd_axis_name=self.spmd_axis_name, + transform_metadata={nnx.PARTITION_NAME: "layers"}, + ) + + if self.config.num_pipeline_repeats > 1: + vmap_repeats = nnx.vmap( + vmap_stages, + in_axes=0, + out_axes=(None, 0, 0), + transform_metadata={nnx.PARTITION_NAME: "circular_repeats"}, + ) + batched_rngs = build_batched_rngs((self.config.num_pipeline_repeats, self.num_stages)) + graphdef, params, rest = vmap_repeats(batched_rngs) else: - return pipeline_weights + batched_rngs = build_batched_rngs((self.num_stages,)) + graphdef, params, rest = vmap_stages(batched_rngs) + + # Merge the batched states back into the module + self.layers = nnx.merge(graphdef, params, rest) + + +class NNXPipeline(PipelineBase): + """Original Pipeline implementation adapted for NNX.""" + + def get_current_stage_weights(self, pipeline_weights, loop_iteration, physical_partition_spec=None): + if self.config.num_pipeline_repeats > 1: + return self.get_current_repeat_from_stages( + pipeline_weights, loop_iteration, physical_partition_spec=physical_partition_spec + ) + return self._stamp_at_current_trace(pipeline_weights) def get_current_repeat_from_stages(self, weights, loop_iteration, physical_partition_spec=None): """Fetches the weights for the current repeat from the stages.""" _, repeat_ids = self.get_microbatch_and_repeat_ids(loop_iteration) - circular_metadata_params = { - nn.PARTITION_NAME: "circular_repeats", - "sub_weight_split_dims_mapping": (None,), - "is_initializing": self.is_initializing(), - "x_times": self.config.num_pipeline_repeats, - "optimizer_dims_mapping": None, - } - # Remove the circular metadata axis, this axis will be removed when passed to the main vmap, - # only one circular entry per stage. - weights = meta.remove_axis(weights, 0, circular_metadata_params) - weights = self._remove_logically_partition(weights) def gather_weights_for_stages_in(w, spec=None): + if w is None: + return None return self.vmap_parallel_gather( w, repeat_ids=repeat_ids, repeat_dim_in_weights=0, stages_dim_in_weights=1, physical_partition_spec=spec ) if physical_partition_spec is None: - weights = jax.tree.map(gather_weights_for_stages_in, weights) - else: - weights = jax.tree.map(gather_weights_for_stages_in, weights, physical_partition_spec) - return weights + return jax.tree.map(gather_weights_for_stages_in, weights) + + _, weights_params, weights_rest = nnx.split(weights, pipeline_utils.is_static_param, ...) + + spec_leaves = jax.tree_util.tree_leaves(physical_partition_spec, is_leaf=pipeline_utils.is_spec_leaf) + assert len(spec_leaves) == len(jax.tree_util.tree_leaves(weights_params)), ( + f"Spec tree leaf count ({len(spec_leaves)}) != weights tree leaf count " + f"({len(jax.tree_util.tree_leaves(weights_params))}). " + "The _is_static_param predicate may have diverged between get_weight_sharding and __call__." + ) + spec_iter = iter(spec_leaves) + gathered_params = jax.tree.map( + lambda w: gather_weights_for_stages_in(w, next(spec_iter)), + weights_params, + ) + + # Non-params gathered without sharding hints. + gathered_rest = jax.tree.map(gather_weights_for_stages_in, weights_rest) + + return nnx.State.merge(gathered_params, gathered_rest) def run_one_iteration( self, loop_state, - pipeline_weights, + pipeline_weights_graph, + pipeline_weights_state, positions, segment_ids, deterministic, model_mode, - decoder_layer_instance, logical_partition_spec=None, ): - """Run one loop iteration - gets weights and inputs for each stage, run the stages in parallel, - and update the loop state. - - Args: - loop_state: Dictionary containing the current state of the pipeline (state_io, shift, etc.) - positions: Positional encodings. - segment_ids: Segment IDs for packed sequences. - deterministic: Boolean indicating if execution should be deterministic (e.g. for dropout). - model_mode: Current model mode (train/predict). - logical_partition_spec: Logical partition specification for weights. - """ + """Executes the logic for a single microbatch iteration, including routing inputs and weights, and advancing buffers.""" state_io = loop_state["state_io"] shift = loop_state["shift"] circ_storage = loop_state["circ_storage"] @@ -702,94 +765,56 @@ def run_one_iteration( vmap_func = self.get_main_vmap_func_for_iterations() - if self.config.num_pipeline_repeats > 1: - _, repeat_ids = self.get_microbatch_and_repeat_ids(loop_iteration) - - def prepare_vars_for_main_vmap(weights, physical_partition_spec=None): - circular_metadata_params = { - nn.PARTITION_NAME: "circular_repeats", - "sub_weight_split_dims_mapping": (None,), - "is_initializing": self.is_initializing(), - "x_times": self.config.num_pipeline_repeats, - "optimizer_dims_mapping": None, - } - weights = meta.remove_axis(weights, 0, circular_metadata_params) - weights = self._remove_logically_partition(weights) - - def gather_weights_for_stages_in(w, spec=None): - return self.vmap_parallel_gather( - w, repeat_ids=repeat_ids, repeat_dim_in_weights=0, stages_dim_in_weights=1, physical_partition_spec=spec - ) - - if physical_partition_spec is None: - weights = jax.tree.map(gather_weights_for_stages_in, weights) - else: - weights = jax.tree.map(gather_weights_for_stages_in, weights, physical_partition_spec) - return weights - - prepare_vars_for_main_vmap_partial = functools.partial( - prepare_vars_for_main_vmap, physical_partition_spec=physical_partition_spec - ) - vmap_func = nn.map_variables( - vmap_func, - mapped_collections=["params", "_overwrite_with_gradient", "non_trainable", "summaries", "intermediates"], - mutable=True, - trans_in_fn=prepare_vars_for_main_vmap_partial, - ) + stage_weights_state = self.get_current_stage_weights( + pipeline_weights_state, loop_iteration, physical_partition_spec=physical_partition_spec + ) - stage_weights = self.get_current_stage_weights( - pipeline_weights, loop_iteration, physical_partition_spec=physical_partition_spec + # Strip nnx.Variable wrappers to raw arrays before nnx.vmap. + # When called inside jax.lax.scan, outer-scope Variables have + # _can_update=False, causing check_consistent_aliasing to reject them. + # nnx.merge inside func_to_vmap creates fresh Variables from raw values. + stage_weights_state = jax.tree.map( + lambda x: x.value if isinstance(x, nnx.Variable) else x, + stage_weights_state, + is_leaf=lambda x: isinstance(x, nnx.Variable), ) - stages_output = vmap_func( - decoder_layer_instance, - stage_weights, + + stages_output, updated_stage_weights_state = vmap_func( + pipeline_weights_graph, + stage_weights_state, stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode, ) + if self.config.scan_layers: stages_output = stages_output[0] - new_state = self.get_new_loop_state(stages_output, loop_state) - return new_state - - @staticmethod - def get_logical_spec_repeats_removed(full_logical): - """Returns a new logical spec with 'circular_repeats' removed.""" - if full_logical is None: - return None - - def _remove_from_spec(spec): - return jax.sharding.PartitionSpec(*[dim for dim in spec if dim != "circular_repeats"]) + if self.config.num_pipeline_repeats > 1: + _, repeat_ids = self.get_microbatch_and_repeat_ids(loop_iteration) - return jax.tree.map(_remove_from_spec, full_logical) + def _scatter_update(full_tree, updated_tree, spec=None): + if full_tree is None or updated_tree is None: + return full_tree - @staticmethod - def _remove_logically_partition(weights): - """Removes LogicallyPartitioned wrappers from the variables.""" + def _update_one_stage(full_slice, updated_slice, repeat_id): + return jax.lax.dynamic_update_slice_in_dim(full_slice, jnp.expand_dims(updated_slice, 0), repeat_id, axis=0) - def _remove_logically_partition_leaf(v): - return getattr(v, "value") if isinstance(v, LogicallyPartitioned) else v + sharded_repeat_ids = self.shard_dim_by_stages(repeat_ids, 0, physical_partition_spec=None) + scattered = jax.vmap(_update_one_stage, in_axes=(1, 0, 0), out_axes=1)( + full_tree, updated_tree, sharded_repeat_ids + ) + return self.shard_dim_by_stages(scattered, 1, physical_partition_spec=spec, is_stage_weight=False) - return jax.tree.map(_remove_logically_partition_leaf, weights, is_leaf=lambda v: isinstance(v, LogicallyPartitioned)) + pipeline_weights_state = jax.tree.map(_scatter_update, pipeline_weights_state, updated_stage_weights_state) + else: + pipeline_weights_state = updated_stage_weights_state - def all_gather_over_fsdp(self, variables, logical_partition_spec): - """Gathers FSDP partitioned variables to reconstruct them fully.""" - physical_partition_spec = logical_to_mesh( - logical_partition_spec, mesh=self.mesh, rules=self.config.logical_axis_rules - ) - physical_partition_spec_no_fsdp = jax.tree.map( - self._remove_fsdp_from_physical_partition_spec, physical_partition_spec - ) - return jax.tree.map( - lambda w, p: self._maybe_shard_with_name(w, NamedSharding(self.mesh, p)), - variables, - physical_partition_spec_no_fsdp, - ) + new_state = self.get_new_loop_state(stages_output, loop_state) + return new_state, pipeline_weights_state - @nn.compact def __call__( self, inputs: jnp.ndarray, @@ -812,33 +837,26 @@ def __call__( ), out_sharding=self.input_sharding, ) - example_inputs = jax.lax.broadcast(inputs[0], [self.num_stages]) ag_sharding = jax.sharding.NamedSharding(self.mesh, jax.sharding.PartitionSpec(None, None)) - if positions is not None: - positions = self._maybe_shard_with_name(positions, ag_sharding) - positions = positions.reshape( + positions = self._maybe_shard_with_name(positions, ag_sharding).reshape( (self.config.num_pipeline_microbatches, self.pipeline_microbatch_size, self.config.max_target_length) ) - example_position = jax.lax.broadcast(positions[0], [self.num_stages]) - position_idx = 0 - else: - example_position = None - position_idx = None - if segment_ids is not None: - segment_ids = self._maybe_shard_with_name(segment_ids, ag_sharding) - segment_ids = segment_ids.reshape( + segment_ids = self._maybe_shard_with_name(segment_ids, ag_sharding).reshape( (self.config.num_pipeline_microbatches, self.pipeline_microbatch_size, self.config.max_target_length) ) - example_segmentation = jax.lax.broadcast(segment_ids[0], [self.num_stages]) - segment_idx = 0 - else: - example_segmentation = None - segment_idx = None loop_state = self.init_states(inputs) + # During Linen init, skip the pipeline compute + # and return a zeros placeholder with the output shape. + if is_linen_initializing(): + return jnp.zeros( + (self.config.micro_batch_size_to_train_on, self.config.max_target_length, self.config.emb_dim), + dtype=inputs.dtype, + ) + # Each microbatch should go through each stage (with repeats) - so there is num_micro * (num_stages * repeats) # compute to perform # Each iteration is vmapped by num_stages, so the number of iterations should be @@ -852,80 +870,100 @@ def __call__( real_iterations = self.config.num_pipeline_microbatches * self.config.num_pipeline_repeats total_iterations = real_iterations + bubble_iterations - if self.is_initializing(): - return self._run_weight_initialization( - example_inputs, example_segmentation, example_position, segment_idx, position_idx, deterministic, model_mode - ) + logical_partition_spec = self.get_logical_spec_repeats_removed(logical_partition_spec) - if self.config.pipeline_fsdp_ag_once: - variables = self._remove_logically_partition(self.layers.variables) - all_pipeline_weights = self.all_gather_over_fsdp(variables, logical_partition_spec) - else: - all_pipeline_weights = self.layers.variables + layers_graph, layers_state = nnx.split(self.layers) - logical_partition_spec = self.get_logical_spec_repeats_removed(logical_partition_spec) + def is_lp(x): + return isinstance(x, nn.spmd.LogicallyPartitioned) - def run_iteration_scannable(model, loop_state, xs): - # flax transforms like nn.scan and nn.remat can only be applied to nn.module classes or nn.module instances, so we - # explicitly wrap the run_one_iteration in this method - the 1st argument model (`self`) is a nn.module instance. - return ( - model.run_one_iteration( - loop_state, - all_pipeline_weights, - positions, - segment_ids, - deterministic, - model_mode, - model.layers, - logical_partition_spec=logical_partition_spec, - ), - None, + def unbox_val(x): + return x.value if is_lp(x) else x + + layers_state = jax.tree.map(unbox_val, layers_state, is_leaf=is_lp) + + # Split BEFORE all_gather_over_fsdp so the tree handed to it aligns with + # logical_partition_spec. logical_partition_spec comes from get_weight_sharding + # which filters to the same _is_static_param predicate (nnx.Param + + # _overwrite_with_gradient), so layers_params and the spec tree are + # structurally identical by construction. Passing the unfiltered layers_state + # would include dropout/RNG state that the spec tree lacks, causing + # jax.tree.map to raise "Mismatch custom node data". Mirrors Linen + # where all_gather_over_fsdp operates on + # self.layers.variables (the params collection only). + _, layers_params, layers_metrics, layers_mutables = nnx.split( + layers_state, pipeline_utils.is_static_param, nnx.Intermediate, ... + ) + + # layers_mutables catch-all should contain ONLY RngState variables (RngKey/RngCount). + # If non_trainable state (e.g. BatchStat) appears here, + # it is being carried through scan instead of broadcast. + # NOTE: is_leaf stops jax.tree.leaves from traversing *into* Variable nodes, + # so we see actual Variable instances (not raw arrays). + assert all( + isinstance(v, nnx.RngState) + for v in jax.tree.leaves(layers_mutables, is_leaf=lambda x: isinstance(x, nnx.Variable)) + if isinstance(v, nnx.Variable) + ), ( + "Non-RngState variable found in layers_mutables catch-all partition. " + "Only RngState variables (RngKey/RngCount) should be present." + ) + + if self.config.pipeline_fsdp_ag_once: + layers_params = self.all_gather_over_fsdp(layers_params, logical_partition_spec) + + def scan_body(carry, _): + current_loop_state, current_layer_mutables = carry + iteration = current_loop_state["loop_iteration"] + advanced_mutables = pipeline_utils.advance_rng_state(current_layer_mutables, iteration) + current_layer_state = nnx.State.merge(layers_params, layers_metrics, advanced_mutables) + + new_loop_state, new_layer_state = self.run_one_iteration( + current_loop_state, + layers_graph, + current_layer_state, + positions, + segment_ids, + deterministic, + model_mode, + logical_partition_spec, ) + _, _, new_layer_metrics, new_layer_mutables = nnx.split( + new_layer_state, pipeline_utils.is_static_param, nnx.Intermediate, ... + ) + return (new_loop_state, new_layer_mutables), new_layer_metrics + if self.config.set_remat_policy_on_pipeline_iterations: - run_iteration_scannable = nn.remat( - run_iteration_scannable, - prevent_cse=not self.config.scan_pipeline_iterations, # prevent_cse not used with scan - policy=self.get_pipeline_remat_policy(), + scan_body = jax.checkpoint( + scan_body, policy=self.get_pipeline_remat_policy(), prevent_cse=not self.config.scan_pipeline_iterations ) if self.config.scan_pipeline_iterations: - variable_carry = [] - variable_broadcast = [ - "params", - "_overwrite_with_gradient", - ] # All loop iterations need the weights for the full pipeline. - if self.is_mutable_collection("non_trainable"): - variable_carry.append("non_trainable") - else: - variable_broadcast.append("non_trainable") - run_all_iterations_scanned = nn.scan( - run_iteration_scannable, - variable_axes={"summaries": 0, "aux_loss": 0, "intermediates": 0, "hyper_params": 0}, - variable_broadcast=variable_broadcast, - variable_carry=variable_carry, - # Dropout/aqt keys will be split for each iteration. - split_rngs={"random": True}, - length=total_iterations, + (loop_state, final_layer_mutables), stacked_metrics = jax.lax.scan( + scan_body, (loop_state, layers_mutables), None, length=total_iterations ) - loop_state, _ = run_all_iterations_scanned(self, loop_state, None) else: + current_carry = (loop_state, layers_mutables) + metrics_history = [] for _ in range(total_iterations): - loop_state, _ = run_iteration_scannable(self, loop_state, None) + current_carry, step_metrics = scan_body(current_carry, None) + metrics_history.append(step_metrics) + loop_state, final_layer_mutables = current_carry + stacked_metrics = jax.tree.map(lambda *xs: jnp.stack(xs), *metrics_history) if metrics_history else layers_metrics + + final_layer_state = nnx.State.merge(layers_params, stacked_metrics, final_layer_mutables) + nnx.update(self.layers, final_layer_state) - # The final output is located in the input/output array, however the output microbatches may be permuted relative to - # the input final_output = self.permute_output_micro_per_stage_dim(loop_state["state_io"]) - # reshape outputs to match input shape of total batch instead of microbatches [batch, sequence, embed] - final_output = jnp.reshape( + return jnp.reshape( final_output, (self.config.micro_batch_size_to_train_on, self.config.max_target_length, self.config.emb_dim), out_sharding=self.output_sharding, ) - return final_output -class CircularPipeline(PipelineBase): +class NNXCircularPipeline(PipelineBase): """Implements an circular pipeline schedule with asynchronous weight prefetching. Circular pipelining reduces the pipeline "bubble" by interleaving multiple pipeline @@ -935,74 +973,38 @@ class CircularPipeline(PipelineBase): *current* repeat is executing. """ - def init_states(self, inputs): - """Initializes the pipeline execution state and communication buffers. - - This sets up the memory needed to pass activations between pipeline stages - (`state_io` and `shift`) and allocates the empty Buffer Sliding Window (BSW) - that will hold the gathered FSDP weights. + def get_main_vmap_func_for_iterations(self): + """ + Returns main stage function vmapped by number of stages. + This becomes a vmap over a single layer instance if body_instance is a single layer, + else a set of layers if body_instance is a set of layers. """ - shift = jnp.zeros((self.num_stages,) + inputs.shape[1:], dtype=inputs.dtype) - shift = self._maybe_shard_with_logical(shift, self.stages_in_logical) - - if self.config.pipeline_delay_activation_forwarding: - prev_outputs = jnp.zeros((self.num_stages,) + inputs.shape[1:], dtype=inputs.dtype) - prev_outputs = self._maybe_shard_with_logical(prev_outputs, self.stages_in_logical) - else: - prev_outputs = None - - state_io = jnp.reshape( - inputs, (self.num_stages, self.microbatches_per_stage) + inputs.shape[1:], out_sharding=self.state_io_sharding - ) - state_io = self._maybe_shard_with_logical(state_io, self.state_io_logical) - - if self.use_circ_storage: - circ_storage = jnp.zeros((self.num_stages,) + inputs.shape, dtype=inputs.dtype, out_sharding=self.state_io_sharding) - circ_storage_mover = shift - else: - circ_storage = None - circ_storage_mover = None - - def _init_empty_bsw_buffers(variables): - # BSW requires two buffers (current and next) for the sliding window - return ( - jax.tree.map(lambda x: jnp.zeros_like(x[0]), variables), - jax.tree.map(lambda x: jnp.zeros_like(x[0]), variables), - ) - if self.is_initializing(): - bsw = None - else: - variables = pipeline_utils.remove_logically_partition(self.layers.variables) - bsw = _init_empty_bsw_buffers(variables) + def func_to_vmap(graph, state, stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode): + module = nnx.merge(graph, state) + out = module(stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode) + _, _, updated_metrics, updated_mutables = nnx.split(module, pipeline_utils.is_static_param, nnx.Intermediate, ...) + return out, (updated_metrics, updated_mutables) - init_loop_state = { - "state_io": state_io, - "shift": shift, - "circ_storage": circ_storage, - "circ_storage_mover": circ_storage_mover, - "loop_iteration": 0, - "prev_outputs": prev_outputs, + vmap_kwargs = { + "in_axes": (None, 0, 0, 0, 0, None, None), + "out_axes": (0, (0, 0)), } - return init_loop_state, bsw - - def gather_weights_across_stages_vmap(self, weights, repeat_ids, repeat_dim_in_weights, stages_dim_in_weights): - """Uses jax.vmap to dynamically slice and gather weights for specific pipeline repeats.""" - - def _gather_single_repeat(x, repeat_id): - return jnp.squeeze(jax.lax.dynamic_slice_in_dim(x, repeat_id, 1, repeat_dim_in_weights), repeat_dim_in_weights) - - gathered_weights_stage_dim = 0 - stage_weights = jax.vmap( - _gather_single_repeat, in_axes=(stages_dim_in_weights, 0), out_axes=gathered_weights_stage_dim - )(weights, repeat_ids) - return stage_weights + if self.spmd_axis_name is not None: + vmap_kwargs["spmd_axis_name"] = self.spmd_axis_name + return jax.vmap(func_to_vmap, **vmap_kwargs) def gather_microbatch_inputs_vmap(self, xs, ids, ids_dim): """Slices out the specific sequence inputs (e.g., positions, segments) for the current microbatch.""" + if xs is None: + return None + + xs = jnp.asarray(xs) + ndim = xs.ndim + def _gather_one(x, i): - idx = tuple(i if d == ids_dim else slice(None) for d in range(x.ndim)) + idx = tuple(i if d == ids_dim else slice(None) for d in range(ndim)) positions_sharding = ( create_sharding(self.mesh, (None, "layers", "activation_length")) if self.config.shard_mode == ShardMode.EXPLICIT @@ -1012,158 +1014,29 @@ def _gather_one(x, i): return jax.vmap(_gather_one, in_axes=(None, 0), out_axes=ids_dim)(xs, ids) - def advance_circular_buffers(self, output, loop_state): - """Rotates pipeline activations to the next physical device stage. - - Uses `jax.lax.ppermute` to perform cross-device ring communication, shifting - the forward activations (`state_io` and `shift`) from stage $i$ to stage $i+1$. - """ - old_state_io = loop_state["state_io"] - old_circ_storage = loop_state["circ_storage"] - old_circ_storage_mover = loop_state["circ_storage_mover"] - loop_iteration = loop_state["loop_iteration"] - - @jax.shard_map(mesh=self.mesh, in_specs=self.stages_in_spec, out_specs=self.stages_in_spec, check_vma=True) - def _rotate_right(arr): - stage_size = jax.lax.axis_size("stage") - perm = [(i, (i + 1) % stage_size) for i in range(stage_size)] - return jax.lax.ppermute(arr, axis_name="stage", perm=perm) - - @jax.shard_map(mesh=self.mesh, in_specs=self.stages_in_spec, out_specs=self.stages_in_spec, check_vma=True) - def _shift_right(arr): - stage_idx = jax.lax.axis_index("stage") - stage_size = jax.lax.axis_size("stage") - perm = [(i, (i + 1) % stage_size) for i in range(stage_size)] - arr = jax.lax.ppermute(arr, axis_name="stage", perm=perm) - return jnp.where(stage_idx == 0, jnp.zeros_like(arr), arr) - - def _update_shift(output_in): - if self.config.num_pipeline_repeats == 1 or self.use_circ_storage: - return _shift_right(output_in) - else: - return _rotate_right(output_in) - - new_shift = _update_shift(output) - new_prev_outputs = None - - if self.use_circ_storage: - - def _rotate_right_and_update(circ_storage_mover_in, circ_storage_in): - rotated = _rotate_right(circ_storage_mover_in) - rotated = jnp.expand_dims(rotated, 1) - offset = ( - loop_iteration - self.iterations_to_complete_first_microbatch_one_repeat() - 1 - ) % self.config.num_pipeline_microbatches - return jax.lax.dynamic_update_slice_in_dim(circ_storage_in, rotated, offset, axis=1) - - new_circ_storage = _rotate_right_and_update(old_circ_storage_mover, old_circ_storage) - new_circ_storage_mover = output - else: - new_circ_storage = None - new_circ_storage_mover = None - - stream_buf_idx = loop_iteration % self.microbatches_per_stage - stream_slice = old_state_io[:, stream_buf_idx] - - def _rotate_left(arr, stage_size): - perm = [(i, (i - 1) % stage_size) for i in range(stage_size)] - return jax.lax.ppermute(arr, axis_name="stage", perm=perm) - - def _shift_left(arr, stage_size, output): - stage_idx = jax.lax.axis_index("stage") - arr = _rotate_left(arr, stage_size) - return jnp.where(stage_idx == stage_size - 1, output, arr) - - @jax.shard_map( - mesh=self.mesh, - in_specs=(self.state_io_spec, self.stages_in_spec, self.stages_in_spec, P()), - out_specs=self.state_io_spec, - check_vma=True, - ) - def _update_state_io(state_in, stream_slice, output, stream_buf_idx): - stage_size = jax.lax.axis_size("stage") - stream_slice = _shift_left(stream_slice, stage_size, output) - stream_slice = jnp.expand_dims(stream_slice, 1) - return jax.lax.dynamic_update_slice_in_dim(state_in, stream_slice, stream_buf_idx, axis=1) - - new_state = _update_state_io(old_state_io, stream_slice, output, stream_buf_idx) - new_loop_state = { - "state_io": new_state, - "shift": new_shift, - "circ_storage": new_circ_storage, - "circ_storage_mover": new_circ_storage_mover, - "loop_iteration": loop_iteration + 1, - "prev_outputs": new_prev_outputs, - } - return new_loop_state - - def realign_output_microbatches(self, output): - """Reorders the output tensor to reverse the circular shifts applied during execution. - - Because the pipeline operates circularly, the output microbatches are shifted - out of order by the time the final stage is completed. This rolls them back - into their original sequential layout. - """ - microbatch_0_idx = self.iterations_to_complete_first_microbatch() % self.microbatches_per_stage - output = jnp.roll(output, shift=-microbatch_0_idx, axis=1) - output = self._maybe_shard_with_logical(output, self.state_io_logical) - return output - - def fetch_active_stage_weights(self, bsw, loop_iteration, physical_partition_spec=None, is_initializing=None): - """The module fetches the actively prefetched weights - from the Buffer Sliding Window to avoid mid-iteration FSDP all-gathers. - """ - pipeline_weights = self.get_current_weights_from_bsw( - bsw, loop_iteration, physical_partition_spec=physical_partition_spec, is_initializing=is_initializing - ) - return pipeline_weights + def gather_weights_across_stages_vmap(self, weights_state, repeat_ids, repeat_dim_in_weights, stages_dim_in_weights): + """Uses jax.vmap to dynamically slice and gather weights for specific pipeline repeats.""" - def get_current_weights_from_bsw(self, bsw, loop_iteration, physical_partition_spec, is_initializing=None): - """Pulls the fully gathered parameters for the current repeat from the BSW dual-buffer.""" - bsw_pps = jax.tree.map(self._remove_fsdp_from_physical_partition_spec, physical_partition_spec) - _, repeat_ids = self.get_microbatch_and_repeat_ids(loop_iteration) - stage0_repeat_id = jnp.maximum(loop_iteration, 0) // self.config.num_pipeline_microbatches + def _gather_repeat_leaf(w_leaf, rep_id): + if w_leaf is None: + return None + return jnp.squeeze( + jax.lax.dynamic_slice_in_dim(w_leaf, rep_id, 1, axis=repeat_dim_in_weights), axis=repeat_dim_in_weights + ) - @jax.shard_map(mesh=self.mesh, in_specs=((bsw_pps, bsw_pps), P("stage")), out_specs=bsw_pps, check_vma=True) - def select_weights_from_bsw(bsw, repeat_id): - # Different stage uses different components in BSW. Stage 0 must use the new weight. - return jax.tree.map(lambda x, y: jax.lax.select(repeat_id[0] == stage0_repeat_id, y, x), bsw[0], bsw[1]) - - weights = select_weights_from_bsw(bsw, repeat_ids) - if is_initializing is None: - is_initializing = self.is_initializing() - - circular_metadata_params = { - nn.PARTITION_NAME: "circular_repeats", - "sub_weight_split_dims_mapping": (None,), - "is_initializing": is_initializing, - "x_times": self.config.num_pipeline_repeats, - "optimizer_dims_mapping": None, - } - weights = meta.remove_axis(weights, 0, circular_metadata_params) - return weights + vmap_gather = jax.vmap(_gather_repeat_leaf, in_axes=(stages_dim_in_weights, 0), out_axes=0) + return jax.tree.map(lambda w: vmap_gather(w, repeat_ids) if w is not None else None, weights_state) - def from_all_variables_to_repeat_weights(self, weights, loop_iteration): + def from_all_variables_to_repeat_weights(self, weights_state, loop_iteration): """Gathers weights corresponding to the repeat IDs for current iteration.""" - _, repeat_ids = self.get_microbatch_and_repeat_ids(loop_iteration) - - def gather_weights_for_stages_in(w): - return self.gather_weights_across_stages_vmap( - w, repeat_ids=repeat_ids, repeat_dim_in_weights=0, stages_dim_in_weights=1 - ) + if self.config.num_pipeline_repeats == 1: + return weights_state - weights = pipeline_utils.remove_logically_partition(weights) - weights = jax.tree.map(gather_weights_for_stages_in, weights) + _, repeat_ids = self.get_microbatch_and_repeat_ids(loop_iteration) - circular_metadata_params = { - nn.PARTITION_NAME: "circular_repeats", - "sub_weight_split_dims_mapping": (None,), - "is_initializing": self.is_initializing(), - "x_times": self.config.num_pipeline_repeats, - "optimizer_dims_mapping": None, - } - repeat_weights = meta.remove_axis(weights, 0, circular_metadata_params) - return repeat_weights + return self.gather_weights_across_stages_vmap( + weights_state, repeat_ids=repeat_ids, repeat_dim_in_weights=0, stages_dim_in_weights=1 + ) def from_repeat_weights_to_bsw( self, @@ -1175,14 +1048,21 @@ def from_repeat_weights_to_bsw( ): """Executes the FSDP-like all-gathers to fully materialize a block of weights for the BSW.""" axes_to_remove = ["fsdp", "fsdp_transpose", "context"] - bsw_pps = pipeline_utils.derive_stage_weight_partition_specs(physical_partition_spec, axes_to_remove) + if physical_partition_spec is not None: + bsw_partition_specs = pipeline_utils.derive_stage_weight_partition_specs(physical_partition_spec, axes_to_remove) + else: + bsw_partition_specs = None def _from_repeat_weights_to_bsw_shardmap( repeat_weights, physical_partition_spec, axes_to_gather, ): - repeat_weights_pps = jax.tree.map(lambda p: P(*p[1:]), physical_partition_spec) + repeat_weights_partition_specs = jax.tree.map( + lambda p: P(*p[1:]) if isinstance(p, P) else p, + physical_partition_spec, + is_leaf=pipeline_utils.is_spec_leaf, + ) # Dynamically gather the index pytrees for all specified axes axis_indices_dict = { @@ -1199,11 +1079,19 @@ def should_skip_gather(axis_name, path_keys): # Add more exclusion rules for other axes here if needed in the future return False - # Renamed to be more descriptive of its action + weights_treedef = jax.tree.structure(repeat_weights) + partition_spec_treedef = jax.tree.structure(repeat_weights_partition_specs, is_leaf=pipeline_utils.is_spec_leaf) + weights_leaves = jax.tree.leaves(repeat_weights) + assert partition_spec_treedef.num_leaves == len(weights_leaves), ( + f"repeat_weights/spec leaf count mismatch: specs={partition_spec_treedef.num_leaves}, " + f"weights={len(weights_leaves)}" + ) + raw_weights = partition_spec_treedef.unflatten(weights_leaves) + @jax.shard_map( mesh=self.mesh, - in_specs=(repeat_weights_pps, None), # 'None' covers the entire axis_pytrees list - out_specs=bsw_pps, + in_specs=(repeat_weights_partition_specs, None), + out_specs=bsw_partition_specs, check_vma=False, ) def _shard_map_gather_weights(sharded_weights, indices_pytrees_list): @@ -1220,12 +1108,13 @@ def _gather_tensor_along_axes(path, x, *indices): return jax.tree_util.tree_map_with_path(_gather_tensor_along_axes, sharded_weights, *indices_pytrees_list) - return _shard_map_gather_weights(repeat_weights, axis_pytrees) + raw_bsw = _shard_map_gather_weights(raw_weights, axis_pytrees) + return weights_treedef.unflatten(jax.tree.leaves(raw_bsw)) - def _from_repeat_weights_to_bsw_hint( - repeat_weights, - ): + def _from_repeat_weights_to_bsw_hint(repeat_weights): def _apply_sharding_hint(weight, pspec): + if pspec is None or weight is None: + return weight sharding_name = NamedSharding(self.mesh, pspec) return maybe_shard_with_name( weight, @@ -1235,27 +1124,113 @@ def _apply_sharding_hint(weight, pspec): extra_stack_level=0, ) - return jax.tree.map(_apply_sharding_hint, repeat_weights, bsw_pps) + spec_leaves = jax.tree_util.tree_leaves(bsw_partition_specs, is_leaf=pipeline_utils.is_spec_leaf) + spec_iter = iter(spec_leaves) + return jax.tree.map(lambda w: _apply_sharding_hint(w, next(spec_iter)), repeat_weights) + + if bsw_partition_specs is None: + return repeat_weights if use_shardmap: return _from_repeat_weights_to_bsw_shardmap(repeat_weights, physical_partition_spec, axes_to_gather=axes_to_gather) return _from_repeat_weights_to_bsw_hint(repeat_weights) - def weight_prefetching(self, weights, physical_partition_spec, loop_iteration): + def weight_prefetching(self, weights_state, physical_partition_spec, loop_iteration): """Triggers asynchronous FSDP-like all-gathers for the next pipeline steps. By gathering weights for `loop_iteration + 1` right now, the network communication can overlap with the compute happening in `loop_iteration`. """ - repeat_weights = self.from_all_variables_to_repeat_weights(weights, loop_iteration + 1) - return self.from_repeat_weights_to_bsw(repeat_weights, physical_partition_spec) + next_repeat_weights = self.from_all_variables_to_repeat_weights(weights_state, loop_iteration + 1) + return self.from_repeat_weights_to_bsw(next_repeat_weights, physical_partition_spec) + + def fetch_active_stage_weights(self, bsw, loop_iteration, physical_partition_spec=None): + """The module fetches the actively prefetched weights + from the Buffer Sliding Window to avoid mid-iteration FSDP all-gathers. + """ + return self.get_current_weights_from_bsw(bsw, loop_iteration, physical_partition_spec) + + def get_current_weights_from_bsw(self, bsw, loop_iteration, physical_partition_spec): + """Pulls the fully gathered parameters for the current repeat from the BSW dual-buffer.""" - def run_one_iteration(self, loop_state, bsw, positions, segment_ids, deterministic, model_mode, logical_partition_spec): - """Executes the forward/backward logic for a single microbatch inside the pipeline. + if bsw[0] is bsw[1]: + treedef = jax.tree.structure(bsw[0]) + leaves = jax.tree.leaves(bsw[0]) + return treedef.unflatten(leaves) - This acts as the core step function that our `jax.lax.scan` wrappers call. It routes - the active BSW weights, sequences, and position IDs into the layer blocks, and then - advances the pipeline communication buffers via `advance_circular_buffers`. + bsw_partition_specs = jax.tree.map(self._remove_fsdp_from_physical_partition_spec, physical_partition_spec) + _, repeat_ids = self.get_microbatch_and_repeat_ids(loop_iteration) + stage0_repeat_id = jnp.maximum(loop_iteration, 0) // self.config.num_pipeline_microbatches + + if bsw_partition_specs is not None: + bsw_treedef = jax.tree.structure(bsw[0]) + + partition_spec_treedef = jax.tree.structure(bsw_partition_specs, is_leaf=pipeline_utils.is_spec_leaf) + bsw0_leaves = jax.tree.leaves(bsw[0]) + bsw1_leaves = jax.tree.leaves(bsw[1]) + assert bsw_treedef == jax.tree.structure( + bsw[1] + ), "BSW half-tree structure mismatch: bsw[0] and bsw[1] must be structurally identical but differ." + assert partition_spec_treedef.num_leaves == len(bsw0_leaves) == len(bsw1_leaves), ( + f"BSW/spec leaf count mismatch: specs={partition_spec_treedef.num_leaves}, " + f"bsw0={len(bsw0_leaves)}, bsw1={len(bsw1_leaves)}" + ) + raw_bsw_0 = partition_spec_treedef.unflatten(bsw0_leaves) + raw_bsw_1 = partition_spec_treedef.unflatten(bsw1_leaves) + + @jax.shard_map( + mesh=self.mesh, + in_specs=((bsw_partition_specs, bsw_partition_specs), P("stage")), + out_specs=bsw_partition_specs, + check_vma=True, + ) + def select_weights_from_bsw(bsw_inner, repeat_id): + return jax.tree.map( + lambda first_slot, second_slot: jax.lax.select(repeat_id[0] == stage0_repeat_id, second_slot, first_slot), + bsw_inner[0], + bsw_inner[1], + ) + + raw_weights = select_weights_from_bsw((raw_bsw_0, raw_bsw_1), repeat_ids) + weights = bsw_treedef.unflatten(jax.tree.leaves(raw_weights)) + else: + + def select_weights_from_bsw(bsw_inner, repeat_id): + return jax.tree.map( + lambda first_slot, second_slot: jax.lax.select(repeat_id == stage0_repeat_id, second_slot, first_slot) + if first_slot is not None + else None, + bsw_inner[0], + bsw_inner[1], + ) + + weights = jax.vmap(select_weights_from_bsw, in_axes=((0, 0), 0), out_axes=0)(bsw, repeat_ids) + + return weights + + def run_one_iteration( + self, + loop_state, + bsw, + pipeline_weights_graph, + layers_metrics, + current_layer_mutables, + positions, + segment_ids, + deterministic, + model_mode, + logical_partition_spec=None, + ): + """Run one loop iteration - gets weights and inputs for each stage, run the stages in parallel, + and update the loop state. + + Args: + loop_state: Dictionary containing the current state of the pipeline (state_io, shift, etc.) + positions: Positional encodings. + segment_ids: Segment IDs for packed sequences. + deterministic: Boolean indicating if execution should be deterministic (e.g. for dropout). + model_mode: Current model mode (train/predict). + logical_partition_spec: Logical partition specification for weights. """ state_io = loop_state["state_io"] shift = loop_state["shift"] @@ -1267,29 +1242,74 @@ def run_one_iteration(self, loop_state, bsw, positions, segment_ids, determinist stages_inputs = self.get_iteration_inputs(loop_iteration, state_io, circ_storage, shift) stages_inputs = jax.ad_checkpoint.checkpoint_name(stages_inputs, "iteration_input") + stages_positions = self.gather_microbatch_inputs_vmap(positions, microbatch_ids, 0) if positions is not None else None stages_segment_ids = ( self.gather_microbatch_inputs_vmap(segment_ids, microbatch_ids, 0) if segment_ids is not None else None ) vmap_func = self.get_main_vmap_func_for_iterations() - stage_weights = self.fetch_active_stage_weights( + + # 1. Fetch params from BSW (params-only, tree matches physical_partition_spec) + stage_params = self.fetch_active_stage_weights( bsw, loop_iteration, physical_partition_spec=physical_partition_spec, - is_initializing=self.is_initializing(), ) - stages_output = vmap_func( - self.layers, stage_weights, stages_inputs, stages_segment_ids, stages_positions, deterministic, model_mode + # 2. Gather non-params (metrics, mutables) for current repeat directly + _, repeat_ids = self.get_microbatch_and_repeat_ids(loop_iteration) + if self.config.num_pipeline_repeats > 1: + stage_metrics = self.gather_weights_across_stages_vmap( + layers_metrics, repeat_ids=repeat_ids, repeat_dim_in_weights=0, stages_dim_in_weights=1 + ) + stage_mutables = self.gather_weights_across_stages_vmap( + current_layer_mutables, repeat_ids=repeat_ids, repeat_dim_in_weights=0, stages_dim_in_weights=1 + ) + else: + stage_metrics = self._stamp_at_current_trace(layers_metrics) + stage_mutables = current_layer_mutables + + # 3. Merge into full state for forward pass + stage_weights_state = nnx.State.merge(stage_params, stage_metrics, stage_mutables) + + stages_output, (updated_stage_metrics, updated_stage_mutables) = vmap_func( + pipeline_weights_graph, + stage_weights_state, + stages_inputs, + stages_segment_ids, + stages_positions, + deterministic, + model_mode, ) + if self.config.scan_layers: stages_output = stages_output[0] - new_state = self.advance_circular_buffers(stages_output, loop_state) - return new_state + # Scatter-back: only update mutables (params handled by AD/gradient, metrics returned directly) + if self.config.num_pipeline_repeats > 1: + + def _scatter_update_mutables(full_tree, updated_tree): + if full_tree is None or updated_tree is None: + return full_tree + + def _update_one_stage(full_slice, updated_slice, repeat_id): + return jax.lax.dynamic_update_slice_in_dim(full_slice, jnp.expand_dims(updated_slice, 0), repeat_id, axis=0) + + sharded_repeat_ids = self.shard_dim_by_stages(repeat_ids, 0, physical_partition_spec=None) + scattered = jax.vmap(_update_one_stage, in_axes=(1, 0, 0), out_axes=1)( + full_tree, updated_tree, sharded_repeat_ids + ) + return self.shard_dim_by_stages(scattered, 1, physical_partition_spec=None, is_stage_weight=False) + + new_mutables_state = jax.tree.map(_scatter_update_mutables, current_layer_mutables, updated_stage_mutables) + new_layer_state = nnx.State.merge(updated_stage_metrics, new_mutables_state) + else: + new_layer_state = nnx.State.merge(updated_stage_metrics, updated_stage_mutables) + + new_state = self.get_new_loop_state(stages_output, loop_state) + return new_state, new_layer_state - @nn.compact def __call__( self, inputs: jnp.ndarray, @@ -1299,7 +1319,8 @@ def __call__( model_mode=MODEL_MODE_TRAIN, logical_partition_spec=None, ) -> jnp.ndarray: - """Entry point for the Pipeline Module. Sets up microbatch schedules and executes scans.""" + """Entry point for the Circular Pipeline Module. Sets up microbatch schedules and executes scans.""" + inputs = inputs.reshape( ( self.config.num_pipeline_microbatches, @@ -1309,110 +1330,523 @@ def __call__( ), out_sharding=self.input_sharding, ) - example_inputs = jax.lax.broadcast(inputs[0], [self.num_stages]) - ag_sharding = jax.sharding.NamedSharding(self.mesh, jax.sharding.PartitionSpec(None, None)) + ag_sharding = jax.sharding.NamedSharding(self.mesh, jax.sharding.PartitionSpec(None, None)) if positions is not None: - positions = self._maybe_shard_with_name(positions, ag_sharding) - positions = positions.reshape( + positions = self._maybe_shard_with_name(positions, ag_sharding).reshape( (self.config.num_pipeline_microbatches, self.pipeline_microbatch_size, self.config.max_target_length) ) - example_position = jax.lax.broadcast(positions[0], [self.num_stages]) - position_idx = 0 - else: - example_position = None - position_idx = None - if segment_ids is not None: - segment_ids = self._maybe_shard_with_name(segment_ids, ag_sharding) - segment_ids = segment_ids.reshape( + segment_ids = self._maybe_shard_with_name(segment_ids, ag_sharding).reshape( (self.config.num_pipeline_microbatches, self.pipeline_microbatch_size, self.config.max_target_length) ) - example_segmentation = jax.lax.broadcast(segment_ids[0], [self.num_stages]) - segment_idx = 0 - else: - example_segmentation = None - segment_idx = None - loop_state, bsw = self.init_states(inputs) - physical_partition_spec = logical_to_mesh( + loop_state = self.init_states(inputs) + + if is_linen_initializing(): + return jnp.zeros( + (self.config.micro_batch_size_to_train_on, self.config.max_target_length, self.config.emb_dim), + dtype=inputs.dtype, + ) + + # Two spec variants needed: + # - Full spec (with circular_repeats axis) -> BSW creation + # - Stripped logical spec (circular_repeats removed) -> BSW consumption + physical_partition_spec_full = logical_to_mesh( logical_partition_spec, mesh=self.mesh, rules=self.config.logical_axis_rules ) + logical_partition_spec_stripped = pipeline_utils.strip_pipeline_repeat_logical_axis(logical_partition_spec) bubble_iterations = self.forwarding_delay * (self.num_stages - 1) + num_microbatches = self.config.num_pipeline_microbatches + num_repeats = self.config.num_pipeline_repeats + remat_policy = self.get_pipeline_remat_policy() + + layers_graph, layers_state = nnx.split(self.layers) + + def is_lp(x): + return isinstance(x, nn.spmd.LogicallyPartitioned) + + def unbox_val(x): + return x.value if is_lp(x) else x + + layers_state = jax.tree.map(unbox_val, layers_state, is_leaf=is_lp) + + _, layers_params, layers_metrics, layers_mutables = nnx.split( + layers_state, pipeline_utils.is_static_param, nnx.Intermediate, ... + ) + + # Validate: layers_mutables should contain ONLY RngState variables + assert all( + isinstance(v, nnx.RngState) + for v in jax.tree.leaves(layers_mutables, is_leaf=lambda x: isinstance(x, nnx.Variable)) + if isinstance(v, nnx.Variable) + ), ( + "Non-RngState variable found in layers_mutables catch-all partition. " + "Only RngState variables (RngKey/RngCount) should be present." + ) + + # Pre-capture Python config values needed inside the stage function. + scan_pipeline_iterations_enabled = self.config.scan_pipeline_iterations + + # ---- Flatten all NNX state into indexed arrays for Linen scope ---- + # + # Each NNX state partition is flattened to (arrays[], keys[], treedef, metadata). + # The arrays go into Linen collections; the metadata (treedef, is_var_flags, etc.) + # stays as Python objects (never enters JAX trace). + + # Params -> 'params' collection (broadcast -- loop-invariant) + param_arrays, param_treedef, param_is_var, param_var_types, param_meta = pipeline_utils.flatten_nnx_state( + layers_params + ) + param_keys = [f"p{i}" for i in range(len(param_arrays))] + + # Mutables -> 'carry_state' collection (carried through scan iterations) + mutables_arrays, mutables_treedef, mutables_is_var, mutables_var_types, mutables_meta = ( + pipeline_utils.flatten_nnx_state(layers_mutables) + ) + mutables_keys = [f"m{i}" for i in range(len(mutables_arrays))] + + # Metrics -> 'metrics_template' collection (broadcast -- template for output structure) + metrics_arrays, metrics_treedef, metrics_is_var, metrics_var_types, metrics_meta = pipeline_utils.flatten_nnx_state( + layers_metrics + ) + metrics_keys = [f"k{i}" for i in range(len(metrics_arrays))] + + # Input arrays -> 'broadcast_inputs' collection (broadcast -- loop-invariant) + input_arrays = [] + input_keys = [] + if positions is not None: + input_arrays.append(positions) + input_keys.append("positions") + if segment_ids is not None: + input_arrays.append(segment_ids) + input_keys.append("segment_ids") + + # ---- Non-pytree context: replaces `self` in closures ---- + # NNX modules are JAX pytrees. Capturing `self` in lift.scan body leaks + # JIT-level tracers from self.layers. _PipelineContext holds bound methods + # only — NOT a JAX pytree, invisible to JAX tracing. + ctx = pipeline_utils.PipelineContext(self) + + # ---- BSW mutable ref (closure, NOT carry -- carry would OOM) ---- + bsw_ref = [None] + + def _stage_fn_for_scope(scope, carry): + """One repeat: prefetch w_next, run microbatch scan, return updated carry. + + All JAX arrays read from scope. No NNX module accessed. + """ + loop_state_carry, w_curr = carry + + # ---- Read ALL JAX arrays from scope ---- + + # Read params from scope (broadcast -- isolated by _partial_pack) + local_param_arrays = [scope.get_variable("params", key) for key in param_keys] + local_layers_params = pipeline_utils.unflatten_nnx_state( + local_param_arrays, param_treedef, param_is_var, param_var_types, param_meta + ) + + # Read broadcast inputs from scope + local_positions = scope.get_variable("broadcast_inputs", "positions") if "positions" in input_keys else None + local_segment_ids = scope.get_variable("broadcast_inputs", "segment_ids") if "segment_ids" in input_keys else None + + # Read metrics template from scope (needed for gather_weights_across_stages_vmap) + local_metrics_arrays = [scope.get_variable("metrics_template", key) for key in metrics_keys] + local_layers_metrics = pipeline_utils.unflatten_nnx_state( + local_metrics_arrays, metrics_treedef, metrics_is_var, metrics_var_types, metrics_meta + ) - if self.is_initializing(): - return self._run_weight_initialization( - example_inputs, example_segmentation, example_position, segment_idx, position_idx, deterministic, model_mode + # Read mutables from scope (carried across repeats by lift.scan) + cur_mutables_arrays = [scope.get_variable("carry_state", key) for key in mutables_keys] + current_layer_mutables = pipeline_utils.unflatten_nnx_state( + cur_mutables_arrays, mutables_treedef, mutables_is_var, mutables_var_types, mutables_meta ) - logical_partition_spec = pipeline_utils.strip_pipeline_repeat_logical_axis(logical_partition_spec) + @jax.custom_vjp + def _run_iter_local(loop_state_arg, bsw_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg): + return _run_iter_local_fwd(loop_state_arg, bsw_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg)[0] - def run_iteration_scannable(model, loop_state, bsw): - return ( - model.run_one_iteration( + def _run_iter_local_fwd(loop_state_arg, bsw_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg): + def _run(loop_state, bsw): + return ctx.run_one_iteration( loop_state, bsw, - positions, - segment_ids, + layers_graph, + metrics_arg, + mutables_arg, + positions_arg, + segment_ids_arg, deterministic, model_mode, - logical_partition_spec=logical_partition_spec, - ), - None, + logical_partition_spec_stripped, + ) + + _run_remat = jax.remat(_run, policy=remat_policy) + (new_loop_state, new_layer_state), vjp_fn = jax.vjp(_run_remat, loop_state_arg, bsw_arg) + return (new_loop_state, new_layer_state), vjp_fn + + def _run_iter_local_bwd(vjp_fn, output_grads): + output_grad_loop_state, output_grad_layer_state = output_grads + input_grad_loop_state, input_grad_bsw = vjp_fn((output_grad_loop_state, output_grad_layer_state)) + # Mutables, metrics, positions, segment_ids have no gradient (stop_gradient or non-differentiable). + return input_grad_loop_state, input_grad_bsw, None, None, None, None + + _run_iter_local.defvjp(_run_iter_local_fwd, _run_iter_local_bwd) + + # ---- Level 2 custom_vjp: wraps the entire microbatch scan ---- + # Matches Linen's run_pipeline_microbatches_custom. + # Purpose: explicit d+g gradient accumulation for BSW weights. + # Without this, JAX keeps 3 copies (weights + grads + accumulators). + # With this, gradients accumulate in-place (2 copies). + # + # All args are explicit (no closure captures of JAX arrays). + + @jax.custom_vjp + def _run_microbatches(loop_state_arg, bsw_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg): + return _run_microbatches_fwd(loop_state_arg, bsw_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg)[ + 0 + ] + + def _run_microbatches_fwd(loop_state_arg, bsw_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg): + # vjp over the scan to get scan_vjp_fn + def scan_fn(scan_loop_state, scan_bsw): + # inner_body MUST use scan_bsw (positional arg), NOT bsw_arg (closure). + # jax.vjp tracks gradients through positional args only. + # Using bsw_arg (closure) would make d_bsw = 0 → lost gradients. + def inner_body(inner_carry, _): + loop_state, mutables = inner_carry + mutables = jax.lax.stop_gradient(mutables) + inner_microbatch_iteration = loop_state["loop_iteration"] + advanced_mutables = pipeline_utils.advance_rng_state(mutables, inner_microbatch_iteration) + new_loop_state, new_layer_state = _run_iter_local( + loop_state, scan_bsw, advanced_mutables, metrics_arg, positions_arg, segment_ids_arg + ) + _, _, new_metrics, new_mutables = nnx.split( + new_layer_state, pipeline_utils.is_static_param, nnx.Intermediate, ... + ) + new_metrics_arrays_inner, _, _, _, _ = pipeline_utils.flatten_nnx_state(new_metrics) + return (new_loop_state, new_mutables), pipeline_utils.arrays_to_linen_collection( + new_metrics_arrays_inner, metrics_keys + ) + + if scan_pipeline_iterations_enabled: + (final_loop_state, final_mutables), metrics = jax.lax.scan( + inner_body, + (scan_loop_state, mutables_arg), + None, + length=num_microbatches, + ) + else: + carry = (scan_loop_state, mutables_arg) + metrics_list = [] + for _ in range(num_microbatches): + carry, step_met = inner_body(carry, None) + metrics_list.append(step_met) + final_loop_state, final_mutables = carry + metrics_arrays_local, _, _, _, _ = pipeline_utils.flatten_nnx_state(metrics_arg) + fallback_met = pipeline_utils.arrays_to_linen_collection(metrics_arrays_local, metrics_keys) + metrics = jax.tree.map(lambda *xs: jnp.stack(xs), *metrics_list) if metrics_list else fallback_met + return (final_loop_state, final_mutables), metrics + + scan_output, scan_vjp_fn = jax.vjp( + scan_fn, + loop_state_arg, + bsw_arg, + ) + (final_loop_state, final_mutables), metrics = scan_output + + return ((final_loop_state, final_mutables, bsw_arg), metrics), scan_vjp_fn + + def _run_microbatches_bwd(scan_vjp_fn, output_grads): + (output_grad_loop_state_mutables_bubblesw, output_grad_metrics) = output_grads + output_grad_loop_state, output_grad_mutables, output_grad_bsw = output_grad_loop_state_mutables_bubblesw + # Backprop through the microbatch scan. + input_grad_loop_state, input_grad_bsw = scan_vjp_fn( + ((output_grad_loop_state, output_grad_mutables), output_grad_metrics) + ) + # Accumulate the scan-computed BSW gradient onto the direct output gradient + # (the "d + g" gradient-accumulation pattern: input_grad + output_grad). + input_grad_bsw = jax.tree.map( + lambda accumulated, incoming: accumulated + incoming if hasattr(accumulated, "shape") else accumulated, + input_grad_bsw, + output_grad_bsw, + ) + return input_grad_loop_state, input_grad_bsw, None, None, None, None + + _run_microbatches.defvjp(_run_microbatches_fwd, _run_microbatches_bwd) + + # ---- Level 3 custom_vjp: wraps weight prefetching + microbatch scan ---- + # Matches Linen's execute_pipeline_stage_pure. + # Purpose: use jax.linear_transpose for weight_prefetching backward + # (reduce-scatter dual of all-gather), avoiding storing forward intermediates. + + @jax.custom_vjp + def _execute_stage( + loop_state_arg, w_curr_arg, params_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg + ): + return _execute_stage_fwd( + loop_state_arg, w_curr_arg, params_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg + )[0] + + def _execute_stage_fwd( + loop_state_arg, w_curr_arg, params_arg, mutables_arg, metrics_arg, positions_arg, segment_ids_arg + ): + # vjp over the FULL stage: weight_prefetching + microbatch scan. + # Auto-diff handles weight_prefetching backward (reduce-scatter) + # correctly — jax.linear_transpose is NOT needed. + def _full_stage(loop_state, w_curr, params): + iteration = loop_state["loop_iteration"] + w_next = ctx.weight_prefetching(params, physical_partition_spec_full, iteration) + bsw = (w_curr, w_next) + (final_loop_state, final_mutables, _), metrics = _run_microbatches( + loop_state, bsw, mutables_arg, metrics_arg, positions_arg, segment_ids_arg + ) + return (final_loop_state, w_next, final_mutables), metrics + + (result, metrics), stage_vjp_fn = jax.vjp(_full_stage, loop_state_arg, w_curr_arg, params_arg) + + return (result, metrics), stage_vjp_fn + + def _execute_stage_bwd(stage_vjp_fn, output_grads): + # output_grads = (output_grad_stage, output_grad_metrics) where + # output_grad_stage = (output_grad_loop_state, output_grad_w_next, output_grad_mutables). + (output_grad_stage, output_grad_metrics) = output_grads + + input_grad_loop_state, input_grad_w_curr, input_grad_params = stage_vjp_fn( + (output_grad_stage, output_grad_metrics) + ) + + return input_grad_loop_state, input_grad_w_curr, input_grad_params, None, None, None, None + + _execute_stage.defvjp(_execute_stage_fwd, _execute_stage_bwd) + + # ---- Execute the stage ---- + (new_loop_state, w_next, new_layer_mutables), inner_metrics = _execute_stage( + loop_state_carry, + w_curr, + local_layers_params, + current_layer_mutables, + local_layers_metrics, + local_positions, + local_segment_ids, ) + # ---- Write updated mutables back to scope ---- + new_mutables_arrays, _, _, _, _ = pipeline_utils.flatten_nnx_state(new_layer_mutables) + for key, array in zip(mutables_keys, new_mutables_arrays): + scope.put_variable("carry_state", key, array) + + # ---- Metrics returned as ys (NOT written to scope) ---- + # variable_axes has a DUMMY pass-through collection ('_pe_split_hint') + # that is never written. Its outputs are KNOWN in partial eval, while + # carry outputs are UNKNOWN (from custom_vjp). This known/unknown split + # triggers _scan_partial_eval_custom to create 2 scan groups → 2 recomp pairs. + # Metrics return as ys to avoid tainting the PE split. + + return (new_loop_state, w_next), inner_metrics + + # ---- Build lift.scan(lift.checkpoint(stage_fn)) over REPEATS ---- if self.config.set_remat_policy_on_pipeline_iterations: - run_iteration_scannable = nn.remat( - run_iteration_scannable, + checkpointed_stage = flax_lift.checkpoint( + _stage_fn_for_scope, + variables=True, + rngs=True, prevent_cse=not self.config.scan_pipeline_iterations, - policy=self.get_pipeline_remat_policy(), + policy=remat_policy, ) + else: + checkpointed_stage = _stage_fn_for_scope + + scanned_stage = flax_lift.scan( + checkpointed_stage, + variable_broadcast=["params", "broadcast_inputs", "metrics_template"], + variable_carry="carry_state", + variable_axes={}, + split_rngs={}, + length=num_repeats, + ) - # base scannable function used twice for real and bubble runs - base_scannable = functools.partial( - pipeline_utils.create_pipeline_stage, - deterministic=deterministic, - model_mode=model_mode, - logical_partition_spec=logical_partition_spec, - physical_partition_spec=physical_partition_spec, - positions=positions, - segment_ids=segment_ids, + # ---- Execute via flax.core.apply ---- + apply_fn = flax_scope.apply( + scanned_stage, + mutable=["carry_state"], ) - run_one_repeat_scannable = base_scannable(length=self.config.num_pipeline_microbatches) - run_bubbles_scannable = base_scannable(length=bubble_iterations) + # Prepare initial Linen variables + initial_w_curr = jax.tree.map(lambda x: jnp.zeros(x.shape[1:], dtype=x.dtype), layers_params) + init_mutables_arrays, _, _, _, _ = pipeline_utils.flatten_nnx_state(layers_mutables) + linen_variables = { + "params": pipeline_utils.arrays_to_linen_collection(param_arrays, param_keys), + "broadcast_inputs": pipeline_utils.arrays_to_linen_collection(input_arrays, input_keys), + "metrics_template": pipeline_utils.arrays_to_linen_collection(metrics_arrays, metrics_keys), + "carry_state": pipeline_utils.arrays_to_linen_collection(init_mutables_arrays, mutables_keys), + } - run_repeats_scanned = pipeline_utils.create_flax_pipeline_scan( - pipeline_stage_fn=run_one_repeat_scannable, - length=self.config.num_pipeline_repeats, - remat_policy=self.get_pipeline_remat_policy(), - use_scan=self.config.scan_pipeline_repeats, + # Run the repeat scan + scan_result, mutated_vars = apply_fn( + linen_variables, + (loop_state, initial_w_curr), ) - run_bubbles_scanned = pipeline_utils.create_flax_pipeline_scan( - pipeline_stage_fn=run_bubbles_scannable, - length=1, - remat_policy=self.get_pipeline_remat_policy(), - use_scan=self.config.scan_pipeline_repeats, + # scan_result = (final_carry, stacked_ys) from lift.scan + (loop_state, final_w_curr), repeat_metrics_raw = scan_result + + # Extract final mutables from scope + final_carry_state = mutated_vars["carry_state"] + final_mutables_arrays = pipeline_utils.linen_collection_to_arrays(final_carry_state, mutables_keys) + final_layer_mutables = pipeline_utils.unflatten_nnx_state( + final_mutables_arrays, mutables_treedef, mutables_is_var, mutables_var_types, mutables_meta ) - initial_carry_repeats = (loop_state, bsw[0], self.layers.variables) - (loop_state, w_curr, pipeline_weights), _ = run_repeats_scanned(self, initial_carry_repeats) - initial_carry_bubbles = (loop_state, w_curr, pipeline_weights) - (loop_state, _, pipeline_weights), _ = run_bubbles_scanned(self, initial_carry_bubbles) + + # Extract stacked metrics (stacked over repeats by lift.scan) + # Each repeat produces (num_microbatches, ...) metrics from inner scan + # lift.scan stacks these -> (num_repeats, num_microbatches, ...) + repeat_metrics_arrays = pipeline_utils.linen_collection_to_arrays(repeat_metrics_raw, metrics_keys) + repeat_metrics = pipeline_utils.unflatten_nnx_state( + repeat_metrics_arrays, metrics_treedef, metrics_is_var, metrics_var_types, metrics_meta + ) + # Reshape from (num_repeats, num_microbatches, ...) -> (total, ...) + repeat_metrics = jax.tree.map( + lambda x: x.reshape((num_repeats * num_microbatches,) + x.shape[2:]), + repeat_metrics, + ) + + # ---- Bubble iterations (pipeline drain) ---- + if bubble_iterations > 0: + bsw_ref[0] = (final_w_curr, final_w_curr) + + @jax.custom_vjp + def _run_iter_bubble( + loop_state_bubble, bsw_bubble, mutables_bubble, metrics_bubble, positions_bubble, segment_ids_bubble + ): + return _run_iter_bubble_fwd( + loop_state_bubble, bsw_bubble, mutables_bubble, metrics_bubble, positions_bubble, segment_ids_bubble + )[0] + + def _run_iter_bubble_fwd( + loop_state_bubble, bsw_bubble, mutables_bubble, metrics_bubble, positions_bubble, segment_ids_bubble + ): + def _run(loop_state, bsw): + return ctx.run_one_iteration( + loop_state, + bsw, + layers_graph, + metrics_bubble, + mutables_bubble, + positions_bubble, + segment_ids_bubble, + deterministic, + model_mode, + logical_partition_spec_stripped, + ) + + _run_remat = jax.remat(_run, policy=remat_policy) + (new_loop_state, new_layer_state), vjp_fn = jax.vjp(_run_remat, loop_state_bubble, bsw_bubble) + return (new_loop_state, new_layer_state), vjp_fn + + def _run_iter_bubble_bwd(vjp_fn, output_grads): + output_grad_loop_state, output_grad_layer_state = output_grads + input_grad_loop_state, input_grad_bsw = vjp_fn((output_grad_loop_state, output_grad_layer_state)) + return input_grad_loop_state, input_grad_bsw, None, None, None, None + + _run_iter_bubble.defvjp(_run_iter_bubble_fwd, _run_iter_bubble_bwd) + + def bubble_body(inner_carry, _): + loop_state, mutables = inner_carry + mutables = jax.lax.stop_gradient(mutables) + current_bubble_iteration = loop_state["loop_iteration"] + advanced_mutables = pipeline_utils.advance_rng_state(mutables, current_bubble_iteration) + new_loop_state, new_layer_state = _run_iter_bubble( + loop_state, bsw_ref[0], advanced_mutables, layers_metrics, positions, segment_ids + ) + _, _, new_metrics, new_mutables = nnx.split( + new_layer_state, pipeline_utils.is_static_param, nnx.Intermediate, ... + ) + new_metrics_arrays_bubble, _, _, _, _ = pipeline_utils.flatten_nnx_state(new_metrics) + return (new_loop_state, new_mutables), pipeline_utils.arrays_to_linen_collection( + new_metrics_arrays_bubble, metrics_keys + ) + + if self.config.scan_pipeline_iterations: + (loop_state, final_layer_mutables), bubble_metrics_raw = jax.lax.scan( + bubble_body, + (loop_state, final_layer_mutables), + None, + length=bubble_iterations, + ) + else: + bubble_carry = (loop_state, final_layer_mutables) + bubble_metrics_list = [] + for _ in range(bubble_iterations): + bubble_carry, bubble_step_metrics = bubble_body(bubble_carry, None) + bubble_metrics_list.append(bubble_step_metrics) + loop_state, final_layer_mutables = bubble_carry + bubble_metrics_raw = ( + jax.tree.map(lambda *xs: jnp.stack(xs), *bubble_metrics_list) + if bubble_metrics_list + else pipeline_utils.arrays_to_linen_collection(metrics_arrays, metrics_keys) + ) + + bubble_metrics_arrays = pipeline_utils.linen_collection_to_arrays(bubble_metrics_raw, metrics_keys) + bubble_metrics = pipeline_utils.unflatten_nnx_state( + bubble_metrics_arrays, metrics_treedef, metrics_is_var, metrics_var_types, metrics_meta + ) + + stacked_metrics = jax.tree.map( + lambda repeat_leaf, bubble_leaf: jnp.concatenate([repeat_leaf, bubble_leaf], axis=0), + repeat_metrics, + bubble_metrics, + ) + else: + stacked_metrics = repeat_metrics + + final_layer_state = nnx.State.merge(layers_params, stacked_metrics, final_layer_mutables) + nnx.update(self.layers, final_layer_state) final_output = self.realign_output_microbatches(loop_state["state_io"]) - final_output = jnp.reshape( + return jnp.reshape( final_output, (self.config.micro_batch_size_to_train_on, self.config.max_target_length, self.config.emb_dim), out_sharding=self.output_sharding, ) - return final_output - -def create_pipeline(config: Config, layers: nn.Module, mesh: Mesh, remat_policy: Any = None) -> PipelineBase: - """Factory function to instantiate the correct Pipeline module based on config.""" +def create_nnx_pipeline( + config: Config, stage_factory: Any, mesh: Mesh, remat_policy: Any = None, *, rngs: nnx.Rngs +) -> NNXPipeline | NNXCircularPipeline: + """Factory function to instantiate the NNX Pipeline module.""" if config.pipeline_fsdp_ag_per_repeat: - return CircularPipeline(config=config, layers=layers, mesh=mesh, remat_policy=remat_policy) + return NNXCircularPipeline( + config=config, stage_factory=stage_factory, mesh=mesh, remat_policy=remat_policy, rngs=rngs + ) + return NNXPipeline(config=config, stage_factory=stage_factory, mesh=mesh, remat_policy=remat_policy, rngs=rngs) + - return Pipeline(config=config, layers=layers, mesh=mesh, remat_policy=remat_policy) +Pipeline = to_linen_class( + NNXPipeline, + base_metadata_fn=initializers.variable_to_logically_partitioned, +) +CircularPipeline = to_linen_class( + NNXCircularPipeline, + base_metadata_fn=initializers.variable_to_logically_partitioned, +) + + +def create_pipeline( + config: Config, + layers=None, + mesh: Mesh = None, + remat_policy: Any = None, +) -> nn.Module: + """Returns the ToLinen-wrapped NNX pipeline appropriate for the config. + + For raw NNX pipeline classes (no Linen wrapping), use create_nnx_pipeline() instead. + + Args: + config: Model configuration. + layers: Callable[[nnx.Rngs], nnx.Module] constructing one pipeline stage. + mesh: JAX device mesh for sharding. + remat_policy: Optional rematerialization policy. + """ + cls = CircularPipeline if config.pipeline_fsdp_ag_per_repeat else Pipeline + return cls(config=config, stage_factory=layers, mesh=mesh, remat_policy=remat_policy) diff --git a/src/maxtext/utils/pipeline_utils.py b/src/maxtext/utils/pipeline_utils.py index c02030f301..d1240a91b8 100644 --- a/src/maxtext/utils/pipeline_utils.py +++ b/src/maxtext/utils/pipeline_utils.py @@ -20,6 +20,7 @@ from flax import linen as nn from flax.linen.spmd import LogicallyPartitioned import jax.numpy as jnp +from flax import nnx def get_mesh_axis_dim_indices(physical_partition_spec, axis_name="fsdp"): @@ -399,3 +400,148 @@ def create_flax_pipeline_scan(pipeline_stage_fn, length, remat_policy, use_scan= length=length, unroll=unroll_length, ) + + +def is_static_param(path, v): + """Predicate matching nnx.Param and FP8 _overwrite_with_gradient variables. + + Used throughout the pipeline to split state into trainable params vs other state. + Must be consistent everywhere to prevent tree structure mismatches. + """ + return isinstance(v, nnx.Param) or type(v).__name__ == "_overwrite_with_gradient" + + +def advance_rng_state(state, iteration): + """Fold loop_iteration into all RNG keys to produce unique dropout masks per scan step. + + jax.lax.scan has no split_rngs mechanism (unlike Linen's nn.scan), so every + iteration would otherwise see the same dropout mask. This mirrors the effect + of ``nn.scan(split_rngs={"random": True})`` from the Linen pipeline. + + Only typed PRNG key variables (``RngKey``) are folded. RNG counters + (``RngCount``) are uint32 arrays and must be left untouched -- calling + ``jax.random.fold_in`` on raw uint32 data triggers a PRNG-impl shape + mismatch (e.g. shape ``(N, 2)`` vs ``unsafe_rbg`` expecting ``(4,)``). + + Args: + state: An ``nnx.State`` (or partition thereof) that may contain + ``nnx.RngState`` variable entries whose ``.value`` is a JAX PRNG key. + iteration: A scalar integer (the loop counter) folded into each key via + ``jax.random.fold_in``. + + Returns: + A new state with the same tree structure, where every typed PRNG key + entry has a unique key derived from the original key and *iteration*. + """ + + def _fold_if_rng(x): + if isinstance(x, nnx.Variable) and issubclass(x.type, nnx.RngState): + val = x.value + if jax.dtypes.issubdtype(val.dtype, jax.dtypes.prng_key): + + def folded(k): + return jax.random.fold_in(k, iteration) + + for _ in range(val.ndim): + folded = jax.vmap(folded) + return x.replace(value=folded(val)) + return x + + return jax.tree.map(_fold_if_rng, state, is_leaf=lambda x: isinstance(x, nnx.Variable)) + + +def is_spec_leaf(x): + """Predicate matching leaves in the bsw_pps treedef, which can be either P or None (if no sharding).""" + return isinstance(x, P) or x is None + + +def flatten_nnx_state(state): + """Flatten nnx.State to (arrays, treedef, is_var_flags, var_types, var_metadata). + + Returns raw arrays and Python-only metadata for reconstruction. + var_metadata: list of dicts with Variable field metadata (NO .raw_value). + Captures: nothing (pure function on inputs). + """ + + def _is_var(x): + return isinstance(x, nnx.Variable) + + leaves_with_path, treedef = jax.tree_util.tree_flatten_with_path(state, is_leaf=_is_var) + arrays = [] + is_var_flags = [] + var_types = [] + var_metadata = [] + for _, leaf in leaves_with_path: + if isinstance(leaf, nnx.Variable): + arrays.append(leaf.value) + is_var_flags.append(True) + var_types.append(type(leaf)) + var_metadata.append(dict(leaf._var_metadata)) # pylint: disable=protected-access + else: + arrays.append(leaf) + is_var_flags.append(False) + var_types.append(None) + var_metadata.append({}) + return arrays, treedef, is_var_flags, var_types, var_metadata + + +def unflatten_nnx_state(arrays, treedef, is_var_flags, var_types, var_metadata): + """Reconstruct nnx.State from flattened arrays + metadata. + + Does NOT reference any nnx.Variable objects from the original state. + var_metadata contains only Python objects (no JAX arrays). + Captures: nothing (pure function on inputs). + """ + new_leaves = [] + for arr, is_var, vtype, meta in zip(arrays, is_var_flags, var_types, var_metadata): + if is_var and vtype is not None: + new_leaves.append(vtype(arr, **meta)) + else: + new_leaves.append(arr) + return treedef.unflatten(new_leaves) + + +def arrays_to_linen_collection(arrays, keys): + """Convert list of arrays + key names to a Linen-style flat dict. + + Captures: nothing (pure function). + """ + return dict(zip(keys, arrays)) + + +def linen_collection_to_arrays(collection, keys): + """Extract arrays from Linen-style flat dict in key order. + + Captures: nothing (pure function). + """ + return [collection[k] for k in keys] + + +# --------------------------------------------------------------------------- +# Non-pytree context: holds Python-only attributes from the pipeline module. +# NNX modules are JAX pytrees — capturing them in lift.scan closures leaks +# JIT-level tracers from self.layers. This wrapper is NOT a JAX pytree, +# so JAX never tries to flatten it. +# --------------------------------------------------------------------------- + + +class PipelineContext: + """Non-pytree wrapper holding pipeline methods + Python config. + + Created from an NNXCircularPipeline ONCE before entering transforms. + Captures ONLY bound methods (which internally access config/mesh/Python attrs) + and Python objects. No nnx.Variable or JAX arrays. + """ + + __slots__ = ( + "weight_prefetching", + "run_one_iteration", + "from_all_variables_to_repeat_weights", + "from_repeat_weights_to_bsw", + ) + + def __init__(self, pipeline_module): + self.weight_prefetching = pipeline_module.weight_prefetching + self.run_one_iteration = pipeline_module.run_one_iteration + self.from_all_variables_to_repeat_weights = pipeline_module.from_all_variables_to_repeat_weights + self.from_repeat_weights_to_bsw = pipeline_module.from_repeat_weights_to_bsw diff --git a/tests/integration/pipeline_parallelism_test.py b/tests/integration/pipeline_parallelism_test.py index ea9189a0b7..3de6c0ce96 100644 --- a/tests/integration/pipeline_parallelism_test.py +++ b/tests/integration/pipeline_parallelism_test.py @@ -14,7 +14,9 @@ """Tests for pipeline parallelism.""" +import contextlib import functools +import os import os.path import sys import unittest @@ -52,6 +54,21 @@ def _adapt_parallelism(args, pipeline_stages=4): args.append(f"ici_data_parallelism={data_par}") +@contextlib.contextmanager +def _temporary_env(name, value): + """Temporarily set an environment variable for the duration of a context.""" + + old_value = os.environ.get(name) + os.environ[name] = value + try: + yield + finally: + if old_value is None: + os.environ.pop(name, None) + else: + os.environ[name] = old_value + + def assert_same_output_and_grad(f1, f2, *inputs): """check that the output and gradient are the same""" f1_value, f1_grad = jax.value_and_grad(f1)(*inputs) @@ -65,8 +82,20 @@ def pytree_ravel(pytree): f1_grad = pytree_ravel(f1_grad) f2_grad = pytree_ravel(f2_grad) - assert jax.numpy.allclose(f1_value, f2_value, rtol=1e-2, atol=1e-2, equal_nan=False) - assert jax.numpy.allclose(f1_grad, f2_grad, rtol=1e-1, atol=1e-1, equal_nan=False) + g_diff = jnp.abs(f1_grad - f2_grad) + v_diff = jnp.abs(f1_value - f2_value) + + value_close = bool(jax.numpy.allclose(f1_value, f2_value, rtol=1e-2, atol=1e-2, equal_nan=False)) + grad_close = bool(jax.numpy.allclose(f1_grad, f2_grad, rtol=1e-1, atol=2e-1, equal_nan=False)) + assert value_close, f"value mismatch: f1={float(f1_value)} vs f2={float(f2_value)}, " f"abs_diff={float(v_diff)}" + assert grad_close, ( + f"grad mismatch: abs_diff_max={float(g_diff.max())}, " + f"abs_diff_mean={float(g_diff.mean())}, " + f"f1_grad_norm={float(jnp.linalg.norm(f1_grad))}, " + f"f2_grad_norm={float(jnp.linalg.norm(f2_grad))}, " + f"grad_size={f1_grad.size}, " + f"tolerance=rtol=1e-1+atol=1.0" + ) @pytest.mark.integration_test @@ -75,22 +104,28 @@ class PipelineParallelismTest(unittest.TestCase): base_output_directory = get_test_base_output_directory() dataset_path = get_test_dataset_path() - def assert_pipeline_same_output_and_grad(self, config, single_pipeline_stage_class=None): + def assert_pipeline_matches_sequential_output_and_grad(self, config, single_pipeline_stage_class=None): """check that the output and gradient are the same""" devices_array = maxtext_utils.create_device_mesh(config) mesh = Mesh(devices_array, config.mesh_axes) model_mode = MODEL_MODE_TRAIN + + # `single_pipeline_stage_class` (when provided, e.g. `deepseek.DeepSeekMoELayerToLinen` + # for the deepseek test) controls BOTH the pipeline and the per-layer reference path. + rngs = nnx.Rngs(params=0) if single_pipeline_stage_class is None: - rngs = nnx.Rngs(params=0) single_pipeline_stage = simple_layer.SimpleDecoderLayerToLinen( config=config, mesh=mesh, model_mode=model_mode, rngs=rngs ) + raw_stage_class = simple_layer.SimpleDecoderLayer + elif issubclass(single_pipeline_stage_class, nnx_wrappers.ToLinen): + single_pipeline_stage = single_pipeline_stage_class(config=config, mesh=mesh, model_mode=model_mode, rngs=rngs) + # `to_linen_class` stores the wrapped NNX class as the `module_class` class attribute + # (see `nnx_wrappers.py:to_linen_class` -> `ToLinenPartial.module_class = base_nnx_class`). + raw_stage_class = single_pipeline_stage_class.module_class else: - if issubclass(single_pipeline_stage_class, nnx_wrappers.ToLinen): - rngs = nnx.Rngs(params=0) - single_pipeline_stage = single_pipeline_stage_class(config=config, mesh=mesh, model_mode=model_mode, rngs=rngs) - else: - single_pipeline_stage = single_pipeline_stage_class(config=config, mesh=mesh, model_mode=model_mode) + single_pipeline_stage = single_pipeline_stage_class(config=config, mesh=mesh, model_mode=model_mode, rngs=rngs) + raw_stage_class = single_pipeline_stage_class def get_inputs(batch_size, sequence, features): """Get random inputs, and random dummy targets @@ -114,16 +149,19 @@ def get_inputs(batch_size, sequence, features): config.global_batch_size_to_train_on, config.max_target_length, config.emb_dim ) deterministic = True - # We use a simpler single matmul decoder layer for fast compilation in these tests. - rngs = nnx.Rngs(params=0) - single_pipeline_stage = simple_layer.SimpleDecoderLayerToLinen( - config=config, mesh=mesh, model_mode=model_mode, rngs=rngs - ) - my_pipeline = pipeline.create_pipeline(config=config, layers=single_pipeline_stage, mesh=mesh) + + def stage_factory(stage_rngs): + return raw_stage_class(config=config, mesh=mesh, model_mode=model_mode, rngs=stage_rngs) + + my_pipeline = pipeline.create_pipeline(config=config, layers=stage_factory, mesh=mesh) init_pipeline_params = my_pipeline.init( jax.random.PRNGKey(0), inputs, inputs_position, inputs_segmentation, deterministic, model_mode ) - logical_partition_spec = my_pipeline.get_weight_sharding( + # `get_weight_sharding` is a compact method on `PipelineLinen` (callable as + # `my_pipeline.get_weight_sharding(...)` directly) but on the ToLinen-wrapped NNX + # pipeline it must be invoked inside a bound module context. Use `bind` so the + # same call shape works on both paths. + logical_partition_spec = my_pipeline.bind(init_pipeline_params).get_weight_sharding( inputs, inputs_position, inputs_segmentation, deterministic, model_mode ) @@ -204,6 +242,36 @@ def regular_sequential_layers_dummy_loss( dummy_targets, ) + def assert_pipeline_same_output_and_grad(self, config, single_pipeline_stage_class=None): + """ + Assert that the pipeline implementation produces the same output and + gradient as a sequential implementation of the same layers. + """ + self.assert_pipeline_matches_sequential_output_and_grad(config, single_pipeline_stage_class) + + def _assert_circular_pipeline_ag_per_repeat_core(self, core): + """ + Assert that the pipeline implementation produces the same output and + gradient as a sequential implementation of the same layers when using + pipeline activation gradient all-gather per repeat, for a given core implementation. + """ + # 2 stages, 8 microbatches, enable pipeline ag per repeat. + with _temporary_env("MAXTEXT_CIRCULAR_PIPELINE_CORE", core): + config = pyconfig.initialize( + [sys.argv[0], get_test_config_path()], + enable_checkpointing=False, + enable_goodput_recording=False, + run_name=f"circular_ag_per_repeat_{core}", + max_target_length=128, + base_emb_dim=28, + ici_pipeline_parallelism=2, + base_num_decoder_layers=8, + num_pipeline_microbatches=8, + per_device_batch_size=4, + pipeline_fsdp_ag_per_repeat=True, + ) + self.assert_pipeline_matches_sequential_output_and_grad(config) + @pytest.mark.tpu_only def test_circular_minimum_microbatches_same_output_and_grad(self): # 4 stages, 8 layers (2 repeats, 1 layer per stage), 4 microbatches @@ -219,7 +287,7 @@ def test_circular_minimum_microbatches_same_output_and_grad(self): num_pipeline_microbatches=4, per_device_batch_size=4, ) - self.assert_pipeline_same_output_and_grad(config) + self.assert_pipeline_matches_sequential_output_and_grad(config) @pytest.mark.tpu_only def test_circular_extra_microbatches_same_output_and_grad(self): @@ -236,11 +304,18 @@ def test_circular_extra_microbatches_same_output_and_grad(self): num_pipeline_microbatches=8, per_device_batch_size=4, ) - self.assert_pipeline_same_output_and_grad(config) + self.assert_pipeline_matches_sequential_output_and_grad(config) @pytest.mark.tpu_only def test_circular_deepseek_megablox_same_output_and_grad(self): - # 4 stages, 8 layers (2 repeats, 1 layer per stage), 8 microbatches + # 4 stages, 8 layers (2 repeats, 1 layer per stage), 8 microbatches. + # DeepSeek's MoE block (`moe.RoutedAndSharedMoE`) constructs a `shared_experts` + # `MlpBlock` with `intermediate_dim = config.shared_experts * moe_mlp_dim`, so + # `shared_experts >= 1` is required to avoid a zero-dim DenseGeneral kernel + # (`ZeroDivisionError` in `_compute_fans`). DeepSeek self-attention is + # `attention_mla.MLA`, which asserts `config.attention_type == "mla"` in + # `_init_projections` (`src/maxtext/layers/attention_mla.py:718-721`). + # MLA layer supplies sane defaults for the head-dim / lora-rank fields. config = pyconfig.initialize( [sys.argv[0], get_test_config_path()], enable_checkpointing=False, @@ -260,8 +335,12 @@ def test_circular_deepseek_megablox_same_output_and_grad(self): decoder_block="deepseek", base_moe_mlp_dim=1024, base_mlp_dim=1024, + attention_type="mla", + shared_experts=1, + ) + self.assert_pipeline_matches_sequential_output_and_grad( + config, single_pipeline_stage_class=deepseek.DeepSeekMoELayerToLinen ) - self.assert_pipeline_same_output_and_grad(config, single_pipeline_stage_class=deepseek.DeepSeekMoELayerToLinen) @pytest.mark.tpu_only def test_circular_ag_once(self): @@ -279,25 +358,15 @@ def test_circular_ag_once(self): per_device_batch_size=4, pipeline_fsdp_ag_once=True, ) - self.assert_pipeline_same_output_and_grad(config) + self.assert_pipeline_matches_sequential_output_and_grad(config) @pytest.mark.tpu_only def test_circular_pipeline_ag_per_repeat(self): - # 2 stages, 8 microbatches, enable pipeline ag per repeat - config = pyconfig.initialize( - [sys.argv[0], get_test_config_path()], - enable_checkpointing=False, - enable_goodput_recording=False, - run_name="circular_ag_per_repeat", - max_target_length=128, - base_emb_dim=28, - ici_pipeline_parallelism=2, - base_num_decoder_layers=8, - num_pipeline_microbatches=8, - per_device_batch_size=4, - pipeline_fsdp_ag_per_repeat=True, - ) - self.assert_pipeline_same_output_and_grad(config) + self._assert_circular_pipeline_ag_per_repeat_core("nnx_scan") + + @pytest.mark.tpu_only + def test_circular_pipeline_ag_per_repeat_jax_state_core(self): + self._assert_circular_pipeline_ag_per_repeat_core("jax_state") @pytest.mark.tpu_only def test_non_circular_same_output_and_grad(self): @@ -313,7 +382,7 @@ def test_non_circular_same_output_and_grad(self): num_pipeline_microbatches=4, per_device_batch_size=4, ) - self.assert_pipeline_same_output_and_grad(config) + self.assert_pipeline_matches_sequential_output_and_grad(config) @pytest.mark.integration_test @pytest.mark.tpu_only @@ -396,7 +465,7 @@ def test_delay_activation_forwarding_same_output_and_grad(self): per_device_batch_size=4, pipeline_delay_activation_forwarding=True, ) - self.assert_pipeline_same_output_and_grad(config) + self.assert_pipeline_matches_sequential_output_and_grad(config) @pytest.mark.integration_test @pytest.mark.tpu_only diff --git a/tests/unit/nnx_decoders_test.py b/tests/unit/nnx_decoders_test.py index b806ce5c11..acf9183108 100644 --- a/tests/unit/nnx_decoders_test.py +++ b/tests/unit/nnx_decoders_test.py @@ -45,6 +45,7 @@ from maxtext.layers.embeddings import Embed from maxtext.layers.nnx_decoders import NNXDecoder, NNXDecoderLayer, deepstack_process from maxtext.layers.normalizations import RMSNorm +from maxtext.models import gemma4_small from maxtext.models.gpt3 import Gpt3LayerNorm from maxtext.models.llama2 import LlamaDecoderLayer from maxtext.utils import maxtext_utils @@ -908,9 +909,6 @@ def test_gemma4_small_decoder_with_mock_cache_and_ple(self): rngs=rngs, ) - # Pass in kv_caches list to be populated - from maxtext.models import gemma4_small - layer_types = gemma4_small.build_layer_types(cfg.num_decoder_layers, cfg.model_name) cache_index_of = gemma4_small.kv_cache_slot_map(layer_types, cfg.num_kv_shared_layers) max_slot = max(cache_index_of.values())