diff --git a/megatron/core/pipeline_parallel/schedules.py b/megatron/core/pipeline_parallel/schedules.py index b2c23807bea..2a6820b280a 100644 --- a/megatron/core/pipeline_parallel/schedules.py +++ b/megatron/core/pipeline_parallel/schedules.py @@ -45,7 +45,11 @@ Shape = Union[List[int], torch.Size] -def get_forward_backward_func(pp_size: Optional[int] = None, vp_size: Optional[int] = None): +def get_forward_backward_func( + pp_size: Optional[int] = None, + vp_size: Optional[int] = None, + schedule_pg_collection: Optional[MultiModuleProcessGroupCollection] = None, +): """Retrieves the appropriate forward_backward function given the configuration of parallel_state. @@ -138,8 +142,13 @@ def forward_step(data_iterator, model): vp_size (Optional[int]): Virtual pipeline model parallel size to use. If both pp_size and vp_size are None, both values fall back to parallel_state. Otherwise, provided values are used as-is and None is treated as an explicit input. + schedule_pg_collection (Optional[MultiModuleProcessGroupCollection]): When a + multi-module (cross-grid) collection is passed, select the bridge schedule. """ + if isinstance(schedule_pg_collection, MultiModuleProcessGroupCollection): + return forward_backward_pipelining_without_interleaving + if pp_size is None and vp_size is None: pp_size = parallel_state.get_pipeline_model_parallel_world_size() vp_size = parallel_state.get_virtual_pipeline_model_parallel_world_size() diff --git a/megatron/training/training.py b/megatron/training/training.py index a7af8071f22..c347dd7df5c 100644 --- a/megatron/training/training.py +++ b/megatron/training/training.py @@ -2234,8 +2234,8 @@ def train_step(forward_step_func, data_iterator, model, optimizer, opt_param_sch model_chunk.force_all_reduce = save_wgrads_in_this_iteration optimizer.zero_grad() - if has_nvidia_modelopt: - # [ModelOpt]: Pipeline-parallel Distillation stacks student and teacher tensors + if has_nvidia_modelopt and getattr(args, "modelopt_enabled", False): + # Distillation shape-adjust reads parallel_state; only for modelopt-enabled runs. adjust_tensor_shapes_fn = get_tensor_shapes_adjust_fn_for_distillation( model, seq_length=args.seq_length, @@ -2388,8 +2388,9 @@ def _save_state_dict(attr_name, label): if args.empty_unused_memory_level >= 2: torch.cuda.empty_cache() - if is_last_stage: + if is_last_stage and losses_reduced: # Average loss across microbatches. + # Last stage may have no loss (e.g. MIMO encoder-grid ranks). loss_reduced = {} for key in losses_reduced[0].keys(): val = [x[key].view(-1) for x in losses_reduced] @@ -3352,7 +3353,9 @@ def train( eval_duration = 0.0 eval_iterations = 0 # Wrap forward_backward_func for Full iteration CUDA graph - forward_backward_func = get_forward_backward_func() + forward_backward_func = get_forward_backward_func( + schedule_pg_collection=schedule_pg_collection + ) if args.cuda_graph_impl == "full_iteration": forward_backward_func = FullCudaGraphWrapper( forward_backward_func,