o 8wif@s*ddlZddlZddlmZddlmZmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZddlZddlmZddlmZdd lmZdd lmZdd lmZmZdd lm Z dd l!m"Z"ddl#m$Z$m%Z%m&Z&m'Z'm(Z(ddl)m*Z*ddl+m,Z,ddl-m.Z.m/Z/m0Z0m1Z1ddl2m3Z3m4Z4m5Z5m6Z6m7Z7ddl2m8Z9ddl:m;Z;mZ>ddl?m@Z@mAZAmBZBddlCmZddlDmEZEddlFmGZGmHZHe rddlImJZJededZKGddde,ZLGddde0ZMGd d!d!eZN dEd"e d#eOePeeee ffd$eQd%eRd&eeOePeePe geQffd'df d(d)ZS * +dFd"e d#eOePeeee ffd,eQd-eQd'eOePe ff d.d/ZTd0eRd1eRd2eRd3ejUd'd4f d5d6ZVd7eWd'eefd8d9ZXdGd"e d7ed2eRd,eQd'df d:d;ZY *dHd=eOePe fd7ed2eRd,eQd'df d>d?ZZd7ed'efd@dAZ[dBeOePe fd7ed'eOePe ffdCdDZ\dS)IN) Generator)AbstractContextManager ExitStack) timedelta)Path) TYPE_CHECKINGAnyCallableLiteralOptionalTypeVarUnion)rank_zero_only)Tensor)Module) Optimizer) TypeGuardoverride) CheckpointIO)default_pg_timeout)_distributed_checkpoint_load_distributed_checkpoint_save_get_full_state_dict_context_is_full_checkpoint_is_sharded_checkpoint)_SubprocessScriptLauncher)ParallelStrategy) TBroadcast _apply_filter_BackwardSyncControl!_validate_keys_for_strict_loading)ReduceOp_distributed_is_initialized-_get_default_process_group_backend_for_device_init_dist_connection_sync_ddp_if_availablegroup)_TORCH_GREATER_EQUAL_2_3_TORCH_GREATER_EQUAL_2_4)_materialize_distributed_module)_METADATA_FILENAME _lazy_load_move_state_into) reset_seed)_PATH _Stateful) DeviceMeshTModel)boundcseZdZdZddddefdeedgefdeede fdeede fd e d e e d e e d dffd d ZedKddZeed efddZejeded dfddZeed ejfddZed e fddZejde d dfddZed e fddZeed ee effddZed e e fdd ZedLd!d"ZedLfd#d$ Zed%e d e fd&d'Z!ed%e d dfd(d)Z"edMd*e e d e#fd+d,Z$e -dNd.e%d/e ed0e ee&e fd e%fd1d2Z'ed3ed4ed dfd5d6Z(edOd8e)d9e d e)fd:d;Z*e  dPde ed?e ee ee ege ffd df d@dAZ-e  dQd` feature. Currently supports up to 2D parallelism. Specifically, it supports the combination of Fully Sharded Data-Parallel 2 (FSDP2) with Tensor Parallelism (DTensor). These PyTorch APIs are currently still experimental in PyTorch. Requires PyTorch 2.4 or newer. Arguments: parallelize_fn: A function that applies parallelisms to a module. The strategy will provide the model and device mesh as input. data_parallel_size: The number of devices within a data-parallel group. Defaults to ``"auto"``, which sets this size to the number of nodes in the cluster. tensor_parallel_size: The number of devices within a tensor-parallel group. Defaults to ``"auto"``, which sets this size to the number of GPUs in a single node. save_distributed_checkpoint: If ``True``, each rank saves its shard of weights and optimizer states to a file. The checkpoint is a folder with as many files as the world size. If ``False``, the full weights and optimizer states get assembled on rank 0 and saved to a single file. autoTNparallelize_fnr1data_parallel_sizetensor_parallel_sizesave_distributed_checkpointprocess_group_backendtimeoutreturncs^ttstt|jd||_||_||_d|_ ||_ ||_ ||_ t |_d|_dS)Nz requires PyTorch 2.4 or higher.)super__init__r) ImportErrortype__name___parallelize_fn_data_parallel_size_tensor_parallel_size _num_nodes_save_distributed_checkpoint_process_group_backend_timeout_ParallelBackwardSyncControl_backward_sync_control _device_mesh)selfr6r7r8r9r:r; __class__g/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/lightning_fabric/strategies/model_parallel.pyr?Ys  zModelParallelStrategy.__init__cCs|jdur td|jS)NzKAccessing the device mesh before processes have initialized is not allowed.)rL RuntimeErrorrMrPrPrQ device_meshps z!ModelParallelStrategy.device_meshcCtdt|jd)NThe `z3` does not use the `CheckpointIO` plugin interface.NotImplementedErrorrArBrSrPrPrQ checkpoint_iovz#ModelParallelStrategy.checkpoint_ioiocCrU)NrVz3` does not support setting a `CheckpointIO` plugin.rW)rMr[rPrPrQrY{rZcCs|jdusJ|j|jSN)parallel_devices local_rankrSrPrPrQ root_devices z!ModelParallelStrategy.root_devicecC|jSr\rFrSrPrPrQ num_nodeszModelParallelStrategy.num_nodesrbcCs ||_dSr\ra)rMrbrPrPrQrbs cCs|jdur t|jSdS)Nr)r]lenrSrPrPrQ num_processessz#ModelParallelStrategy.num_processescCs*|jdusJ|jd}||dS)N data_parallel) num_replicasrank)rTsizeget_local_rank)rMdata_parallel_meshrPrPrQdistributed_sampler_kwargss z0ModelParallelStrategy.distributed_sampler_kwargscCr`r\)rHrSrPrPrQr:rcz+ModelParallelStrategy.process_group_backendcCs2|jdusJ|jjst|j|j|j|_dSdSr\)cluster_environmentcreates_processes_externallyrrerb _launcherrSrPrPrQ_configure_launchersz)ModelParallelStrategy._configure_launchercsRt||jdkr|j|_|jdkr|j|_t|j|j|j|j |_ dS)Nr5) r>setup_environment_setup_distributedrDrbrEre_setup_device_mesh world_sizer_rLrSrNrPrQrqs    z'ModelParallelStrategy.setup_environmentmodulecsvddlmtfdd|Drtd|jjd|||j}t |t s3tdt |jt ||j |S)NrFullyShardedDataParallelc3|]}t|VqdSr\ isinstance).0modrvrPrQ z5ModelParallelStrategy.setup_module..z\Found modules that are wrapped with `torch.distributed.fsdp.FullyShardedDataParallel`. The `z5` only supports the new FSDP2 APIs in PyTorch >= 2.4.zBThe `parallelize_fn` must return a `nn.Module` instance, but got: )torch.distributed.fsdprwanymodules TypeErrorrOrBrCrTrzrrAr*r_rMrurPrvrQ setup_modules   z"ModelParallelStrategy.setup_modulecCsdSr\rPrrPrPrQmodule_to_devicesz&ModelParallelStrategy.module_to_device empty_initcCs2|j}t}|r|td|||S)Nmeta) precisionmodule_init_contextr enter_contexttorchdevice)rMrprecision_init_ctxstackrPrPrQrs  z)ModelParallelStrategy.module_init_contextmeantensorr' reduce_opcCst|tr t|||dS|S)N)r)rzrr%)rMrr'rrPrPrQ all_reduces z ModelParallelStrategy.all_reduceargskwargscOs>tsdStjdkrtjj|jjgddStjdS)Nnccl) device_ids)r"r distributed get_backendbarrierr_index)rMrrrPrPrQrs zModelParallelStrategy.barrierrobjsrccCs,ts|S|g}tjj||tjd|dS)Nr&r)r"rrbroadcast_object_list_groupWORLD)rMrrrPrPrQ broadcasts zModelParallelStrategy.broadcastpathstatestorage_optionsfiltercCsv|durtdt|jdt|jd|dur&|jr&tt|jdt||}t|||j |j|ddS)aNSave model, optimizer, and other state to a checkpoint on disk. If distributed checkpointing is enabled (default), the checkpoint gets saved as a directory containing one file per process, with model- and optimizer shards stored per file. Additionally, it creates a metadata file `meta.pt` with the rest of the user's state (only saved from rank 0). If distributed checkpointing is disabled (``save_distributed_checkpoint=False``), the checkpoint will be written to a single file containing the weights, optimizer state and other metadata. N`zF.save_checkpoint(..., storage_options=...)` is not supported because `z"` does not use the `CheckpointIO`.zV doesn't support loading distributed filtered checkpoints, so saving them is disabled.)rrfull_state_dictrhr) rrArBrGrXrr_save_checkpoint global_rank)rMrrrrrPrPrQsave_checkpoints$ z%ModelParallelStrategy.save_checkpointstrictcCs|stdt|jd|dt|jdt||}t|tr-t|||j|diSt|t r=t dt|jdt |||dS) zOLoad the contents from a checkpoint and restore the state of the given objects.zGot z.load_checkpoint(..., state=zY) but a state with at least a model instance to reload is required. Pass it in like so: z2.load_checkpoint(..., state={'model': model, ...}))rurtrzNLoading a single optimizer object from a checkpoint is not supported yet with .)rrr) ValueErrorrArBrrrzr _load_raw_module_state_from_pathrtrrX_load_checkpoint)rMrrrrPrPrQload_checkpoints  z%ModelParallelStrategy.load_checkpointcCs>t|||_|jdusJt|j|j|jddS)N)r;)r._set_world_ranks_get_process_group_backendrHrmr$rIrSrPrPrQrr,s  z(ModelParallelStrategy._setup_distributedcCs|jpt|jSr\)rHr#r_rSrPrPrQr3z0ModelParallelStrategy._get_process_group_backendcCsJ|jdur|j|j|j|j|j|j|j|jt_ t _ dSr\) rmset_global_rank node_rankrer^set_world_sizerbrrrhutils_rank_zero_onlyrSrPrPrQr6s z&ModelParallelStrategy._set_world_ranks)r<r1r<Nr\)Nr)r)NN)NT)3rB __module__ __qualname____doc__rr r2r r intboolr strrr?propertyrTrrrYsetterrrr_rbredictrrlr:rprqrrrrrrr!rrrrr/rrrrrrr __classcell__rPrPrNrQr4Cs      %"  r4c@s&eZdZedededefddZdS)rJruenabledr<cCs t||dS)z9Blocks gradient synchronization inside the FSDP2 modules.)rur) _FSDPNoSyncrMrurrPrPrQno_backward_sync@s z-_ParallelBackwardSyncControl.no_backward_syncN)rBrrrrrrrrPrPrPrQrJ?srJc@sXeZdZdededdfddZdeddfdd Zdd d Zd ed ededdfddZ dS)rrurr<NcCs||_||_dSr\)_module_enabledrrPrPrQr?Gs z_FSDPNoSync.__init__requires_grad_synccCs8ddlm}|jD]}t||r|j|ddq dS)Nr) FSDPModuleFrecurse)"torch.distributed._composable.fsdprrrrzset_requires_gradient_sync)rMrrr|rPrPrQ_set_requires_grad_syncKs  z#_FSDPNoSync._set_requires_grad_synccCs||j dSr\rrrSrPrPrQ __enter__Rsz_FSDPNoSync.__enter__exc_type exc_value tracebackcCs||jdSr\r)rMrrrrPrPrQ__exit__Urz_FSDPNoSync.__exit__r) rBrrrrr?rrrrrPrPrPrQrFs  rrrrrhrr<cCs|r|rt|std|dd|D}t|dkr$tdt|dkr.td|d}ddlm}m}m } ||d d } i} i} | D]7\} }t |t r\||| d }| }nt |t rk| ||| d }| }n t |trt|n|}| }t| |p}i||qJ|rt|rt|| | |dkrt| |dSdS|r||jd d d t| ||dkrt| |tdSdS) Nz/The checkpoint path exists and is a directory: cSsg|]}t|r|qSrP_has_dtensor_modules)r{rurPrPrQ csz$_save_checkpoint..raCould not find a distributed model in the provided checkpoint state. Please provide the model as part of the state like so: `save_checkpoint(..., state={'model': model, ...})`. Make sure you set up the model (and optimizers if any) through the strategy before saving the checkpoint.r=zFound multiple distributed models in the given state. Saving distributed checkpoints is currently limited to a single model per checkpoint. To save multiple models, call the save method for each model separately with a different path.)StateDictOptionsget_model_state_dictget_optimizer_state_dictT)r cpu_offloadoptions)parentsexist_ok)is_dirrIsADirectoryErrorvaluesrdr'torch.distributed.checkpoint.state_dictrrritemsrzrrr0 state_dictrshutilrmtreeupdatersaveis_fileunlinkmkdirrr+)rrrrhrrrurrrstate_dict_optionsconverted_statemetadatakeyr converted target_dictrPrPrQrYsP         rTFroptimizer_states_from_listcCs,ddlm}m}m}m}dd|D}t|dkrtddd|D} t|dkr2tdt|d\} } t |r|d d } | || i} t | || j | | |d | D]\}}||| |i}t |||| |||| d q]t |t}||| }t|||d |D]}||vrq||||<q|St|r t j |d d dd}t|| | |d |d d |d} t| D]!\}\}}|r|d|}n||}t|| }|| ||| d q||| }t|||d t|||d|Stdt|d)Nr)rrrset_optimizer_state_dictcSsi|] \}}t|r||qSrPr)r{rrurPrPrQ sz$_load_checkpoint..aCould not find a distributed model in the provided checkpoint state. Please provide the model as part of the state like so: `load_checkpoint(..., state={'model': model, ...})`. Make sure you set up the model (and optimizers if any) through the strategy before loading the checkpoint.cSs i|] \}}t|tr||qSrP)rzr)r{roptimrPrPrQrs r=zFound multiple distributed models in the given state. Loading distributed checkpoints is currently limited to a single model per checkpoint. To load multiple models, call the load method for each model separately with a different path.T)rr)optim_state_dictrcpuF)mmap map_location weights_onlybroadcast_from_rank0rroptimizer_states)source destinationkeysz The path z does not point to a valid checkpoint. Make sure the path points to either a directory with distributed checkpoint shards, or a single file with a full checkpoint.)rrrrrrrdrlistrrload_state_dictrloadr+rr popr_load_raw_module_state enumerate _rekey_optimizer_state_if_neededr-r)rrrrrrrrr optimizers module_keyrur module_state optim_keyr optim_staterrequested_metadata_keysr checkpoint optimizer_idxoptimizer_name optimizeroptimizer_staterPrPrQrsn         rr7r8rtrr1cCsHddlm}|||krtd|d|d|d||j||fddS) Nr)init_device_meshzThe sizes `data_parallel_size=z` and `tensor_parallel_size=z*` multiplied should equal the world size (z).)rftensor_parallel) device_type mesh_shapemesh_dim_names)torch.distributed.device_meshrrRrA)r7r8rtrrrPrPrQrss  rsrucs0ddlmt|totfdd|DS)NrDTensorc3rxr\ry)r{trrPrQr} r~z'_has_dtensor_modules..)torch.distributed._tensorrrzrr parameters)rurPrrQr s $rcCsFt|s td|trtj|dddnt|}t||||ddS)z;Loads the state dict from a file path into the FSDP module.zxFailed to load checkpoint directly into the model. The given path must be a single file containing the full state dict: Tr)rr)rrurtrN)rrr(rrr,r)rrurtrrrPrPrQrsrr=rcCsddlm}t|rUddlm}m}|dddd}|D]5\}} t| D],\} } ||r/dnd| } | |vrD|ss    $              } @  ^     $*