o ?߱id9 @sddlmZddlZddlmZddlmZddlmZddl m Z m Z ddl m ZddlmZddlmZmZmZmZmZdd lmZdd lmZdd lmZdd lmZej ej!ej"d Z#dej$dededefddZ%dej$dede&de&de&f ddZ'ej(j)j*j+ej(j)j,j+ej(j)j-j+ej(j.j/j+ej(j)j0j+hZ1dej$fddZ2dej$defddZ3dej$fd d!Z4dej$d"efd#d$Z5dej$d"ed%e&d&e&fd'd(Z6dS))) defaultdictN) fully_shard) replicate) ReplicateShard)checkpoint_wrapper) DeviceMesh)ColwiseParallelPrepareModuleInputRowwiseParallelSequenceParallelparallelize_module)log)ActivationCheckpointConfig)FSDP2ModelConfig) ParallelDims)float16float32bfloat16model world_mesh parallel_dims job_configcCs|jr|jjr|jjstdt||d|j|jj |jjd|j j dkr+t ||j |jjr3t ||js9|jrm|jr?d}nd}t||t||jrStdntd|jr`td |jjrktd dSdS|jr|jd krytd t|||jj|jjd dSdS)z Apply tensor parallelism, activation checkpointing, torch.compile, and data parallelism to the model. NOTE: The passed-in model preferably should be on meta device. Otherwise, the model must fit on GPU or CPU memory. z$Async TP requires --training.compiletp) loss_parallel enable_float8enable_async_tpnone) dp_replicate dp_shard_cp)rzApplied HSDP to the modelzApplied FSDP to the modelz%Applied Context Parallel to the modelz#Applied CPU Offloading to the modelz&DDP has not supported > 1D parallelism)enable_compileenable_compiled_autogradN) tp_enabled experimentalenable_async_tensor_paralleltrainingcompile RuntimeErrorapply_tploss_parallel_enabledfloat8enable_float8_linearactivation_checkpointmodeapply_ac apply_compiledp_shard_enabled cp_enableddp_replicate_enabled apply_fsdptupleloggerinfoenable_cpu_offloadndim apply_ddpr")rrrrdp_mesh_dim_namesr<d/data/cameron/vidgen/cosmos-predict2.5/cosmos_predict2/_src/reason1/parallelisms/parallelize_qwen.pyparallelize_qwen0sV        r>tp_meshrrrc Cst||tttdddtttd|rtdnt| dd|r7ddlm}m}m }|||}} } nttt }} } |j j D]j} t| tddddddddftdddddddfd | | | |tdd t| tdftfd | |tdd | d } d d dddddddddd } i}| D] \}}||| |<q~ |} t| || dqC|rddlm}dtjj_||jtd|rdnd|rdndddS)zApply tensor parallelism.r F) input_layoutsoutput_layoutsuse_local_output)zmodel.embed_tokensz model.normlm_headr)Float8ColwiseParallelFloat8RowwiseParallelPrepareFloat8ModuleInputN)r@desired_input_layouts)rA) Zattention_norm attentionz attention.wqz attention.wkz attention.wvz attention.woffn_normZ feed_forwardzfeed_forward.w1zfeed_forward.w2zfeed_forward.w3input_layernorm self_attnzself_attn.q_projzself_attn.k_projzself_attn.v_projzself_attn.o_projpost_attention_layernormmlpz mlp.gate_projz mlp.down_projz mlp.up_proj)module device_meshparallelize_plan)enable_symm_mem_for_groupTApplied zFloat8 zAsync zTensor Parallelism to the model)r r rrr r Z%torchao.float8.float8_tensor_parallelrErFrGr rlayersitems#torch.distributed._symmetric_memoryrRtorch _inductorconfig_micro_pipeline_tp get_group group_namer6r7)rr?rrrrErFrGZrowwise_parallelZcolwise_parallelZprepare_module_inputtransformer_blockZ layer_planZ names_mappingZnew_layer_plankeyvaluerRr<r<r=r)rs      '   r)rOcs d}|j|vrtd|jd||jdkrt|ddS|jdks(J|j|jdk}|j}|s?|s?td |jd |r]d d lmmfd dfdd}t||ddS|rt|j}tj dd tj d7_ |r{tj |d krt|ddS|SdS)N)full selectivezInvalid AC mode: z. Valid modes: raFpreserve_rng_staterbopzInvalid selective AC option: zD. Valid options: 'op' or a positive int representing layer frequencyr)CheckpointPolicy$create_selective_checkpoint_contextscsfdd}|S)Ncsr|jrdnd}|d}|tjjjjkr|d7<|tvo0|tjjjjko/|ddk }|r6jSjS)N recomputeforwardZ _mm_countr r) is_recomputerXopsatenmmdefault _save_list MUST_SAVEPREFER_RECOMPUTE)ctxfuncargskwargsr.Z mm_count_keyto_save)rfmetar<r=_custom_policys  *zR_apply_ac_to_transformer_block.._get_custom_policy.._custom_policyr<)rxry)rfrxr=_get_custom_policys z:_apply_ac_to_transformer_block.._get_custom_policycstt}|S)N)rintrz)r{rgr<r="selective_checkpointing_context_fn$s zJ_apply_ac_to_transformer_block..selective_checkpointing_context_fn) context_fnrd_countr ) r. ValueErrorptd_checkpoint_wrapperselective_ac_optionisdigittorch.utils.checkpointrfrgr|__dict__ setdefaultr)rO ac_configZvalid_ac_modesZ use_op_sacZ use_layer_sacr}Zac_freqr<)rfr{rgr=_apply_ac_to_transformer_blocks@         rrcCsd|jks d|jkr-t|dr-|jdur-|jjD]\}}t|dd}|jj||qd|jks7d|jkrO|jjD]\}}t ||}|jj||q=t d|j d dS) z,Apply activation checkpointing to the model.visionvlmvisualNFrcllmrSz& activation checkpointing to the model) modelshasattrrblocksnamed_childrenrregister_modulerrUrr6r7r.)rrlayer_idblockr^r<r<r=r/>s   r/cCs>|jD]\}}tj|dd}|j||qtddS)z Apply torch.compile to each TransformerBlock, which makes compilation efficient due to repeated structure. Alternatively one can compile the whole model (after applying DP). T) fullgraphz2Compiling each TransformerBlock with torch.compileN)rUrrXr'rr6r7)rrr^r<r<r=r0Ssr0dp_meshcCs^|jdurt|jjD] \}}t||dq t|jjD] \}}t||dqt||ddS)z Apply data parallelism (via FSDP2) to the model. Args: model (nn.Module): The model to apply data parallelism to. dp_mesh (DeviceMesh): The device mesh to use for data parallelism. N)mesh)r enumeraterrrrU)rrrrr^r<r<r=r4_s r4r!r"cCs:|r|r dtjj_ndtjj_t||ddtddS)N'python_reducer_without_compiled_forward ddp_optimizerd)rP bucket_cap_mbzApplied DDP to the model)rX_dynamorZ optimize_ddprr6r7)rrr!r"r<r<r=r:vs   r:)7 collectionsrrXtorch.nnnn"torch.distributed._composable.fsdprZ'torch.distributed._composable.replicatertorch.distributed._tensorrr;torch.distributed.algorithms._checkpoint.checkpoint_wrapperrrtorch.distributed.device_meshrZ!torch.distributed.tensor.parallelr r r r r %cosmos_predict2._src.imaginaire.utilsrr69cosmos_predict2._src.reason1.configs.default.model_configrr JobConfig7cosmos_predict2._src.reason1.parallelisms.parallel_dimsrrrrZTORCH_DTYPE_MAPModuler>boolr)rlrmrnro'_scaled_dot_product_efficient_attention#_scaled_dot_product_flash_attention_c10d_functionalreduce_scatter_tensormaxrprr/r0r4r:r<r<r<r=sz           B       :