o zi@sddlZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZmZmZddlZddlmZddlmZddlmZdd lmZdd lm Z m!Z!dd l"m#Z#dd l$m%Z%dd l&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1m2Z2ddl3m4Z4ddl5m6Z6e rddl7m8Z8edZ9edZ:Gddde'e+Z;deed!eej?ddfd"d#Z@d$e deAfd%d&ZBd$e6ddfd'd(ZCd)ee=0.14.1c[s~eZdZdZdddddddddd d dd d d ddddddddddddddejddddddddddddddf,deedede dee dedede de de de de de d e d!e d"e d#ed$ed%e d&ed'e d(ed)ed*ed+ed,e d-e d.ed/ee d0ee e e e effd1e d2eeejd3eed4ed5e d6e d7e d8e d9ed:ed;edeed?ee d@dfZfdAdB Zed@efdCdDZeed@e e e ffdEdFZeddHdIZedJedKeed@edGeeffdLdMZedJed@dGfdNdOZedPed@efdQdRZ eddSeed@e!ffdTdU Z"ed@e!fdVdWZ#e  ddXe dYe e e eeeffdZeed[ee e e$e egeffd@df d\d]Z%e  ddXe dYee eee e e eeefffd^ed@e e efffd_d` Z&e a ddJdGdPedbe ee fdce ee fdded@ej'f dedfZ(edJdGdPedge ee fd@dfdhdiZ)e*edje+d@dfdkdlZ, ddmedPeed@edGeffdndoZ-edfdpdq Z.eddrdsZ/ddtduZ0ddvdwZ1ddxdyZ2ddzd{Z3ded.ed/ee d9ed:ed;ed` feature. Defaults have been set to enable ZeRO-Offload and some have been taken from the link below. These defaults have been set generally, but may require tuning for optimum performance based on your model size. `For more information: https://www.deepspeed.ai/docs/config-json/#zero-optimizations-for-fp16-training`. Arguments: zero_optimization: Enable ZeRO optimization. This is compatible with either ``precision="16-mixed"`` or ``precision="bf16-mixed"``. stage: Different stages of the ZeRO Optimizer. 0 is disabled, 1 is optimizer state partitioning, 2 is optimizer+gradient state partitioning, 3 is optimizer+gradient_parameter partitioning using the infinity engine. remote_device: Device to instantiate the model on initially (``cpu`` or ``nvme``). Defaults to GPU. offload_optimizer: Enable offloading optimizer memory and computation to CPU or NVMe based on ``offload_optimizer_device``. offload_parameters: When using ZeRO Stage 3, Enable offloading parameter memory and computation to CPU or NVMe based on ``offload_params_device``. offload_params_device: When offloading parameters choose the device to offload to, ``cpu`` or ``nvme``. offload_optimizer_device: When offloading optimizer state choose the device to offload to, ``cpu`` or ``nvme``. params_buffer_count: Number of buffers in buffer pool for parameter offloading when ``offload_params_device`` is ``nvme``. params_buffer_size: Size of buffers in buffer pool for parameter offloading when ``offload_params_device`` is ``nvme``. max_in_cpu: Number of parameter elements to maintain in CPU memory when offloading to NVMe is enabled. nvme_path: Filesystem path for NVMe device for optimizer/parameter state offloading. optimizer_buffer_count: Number of buffers in buffer pool for optimizer state offloading when ``offload_optimizer_device`` is set to to ``nvme``. This should be at least the number of states maintained per parameter by the optimizer. For example, Adam optimizer has 4 states (parameter, gradient, momentum, and variance). block_size: When using NVMe Offloading, the I/O block size in bytes. queue_depth: When using NVMe Offloading, the I/O queue depth. single_submit: When using NVMe Offloading, submit requests to storage device as multiple individual requests, as opposed to one block of requests. overlap_events: When using NVMe Offloading, submit requests to storage device in an overlapped fashion without waiting for completion of earlier requests. thread_count: When using NVMe Offloading, Intra-request parallelism for each read/write submitted by a user thread. pin_memory: When using ZeRO stage 3, pin optimizer state memory on CPU. This could boost throughput at the cost of extra memory overhead. sub_group_size: When using ZeRO stage 3, defines the number of parameters within a sub group to offload at a time. Smaller numbers require more communication, but improve memory efficiency. contiguous_gradients: Copies gradients to a continuous buffer as they are produced. Avoids memory fragmentation during backwards. Useful when training large models. overlap_comm: Overlap the reduction (synchronization) of gradients with the backwards computation. This is a speed optimization when training across multiple GPUs/machines. allgather_partitions: All gather updated parameters at the end of training step, instead of using a series of broadcast collectives. reduce_scatter: Use reduce/scatter instead of allreduce to average gradients. allgather_bucket_size: Number of elements to allgather at once. Used to limit the memory required for larger model sizes, with a tradeoff with speed. reduce_bucket_size: Number of elements to reduce at once. Used to limit the memory required for larger model sizes, with a tradeoff with speed. zero_allow_untested_optimizer: Allow untested optimizers to be used with ZeRO. Currently only Adam is a DeepSpeed supported optimizer when using ZeRO. logging_batch_size_per_gpu: Config used in DeepSpeed to calculate verbose timing for logging on a per sample per second basis (only displayed if logging=logging.INFO). To obtain accurate logs when using datasets that do not support batch samplers, set this to the actual per gpu batch size. config: Pass in a deepspeed formatted config dict, or path to a deepspeed config: https://www.deepspeed.ai/docs/config-json. All defaults will be ignored if a config is passed in. logging_level: Set logging level for deepspeed. loss_scale: Loss scaling value for FP16 training. 0.0 results in dynamic loss scaling, otherwise static. initial_scale_power: Power of the initial dynamic loss scale value. Loss scale is computed by ``2^initial_scale_power``. loss_scale_window: Window in which to raise/lower the dynamic FP16 loss scaling value. hysteresis: FP16 Delay shift in Dynamic Loss scaling. min_loss_scale: The minimum FP16 dynamic loss scaling value. partition_activations: Enables partition activation when used with ZeRO stage 3 and model parallelism. Still requires you to wrap your forward functions in deepspeed.checkpointing.checkpoint. See `deepspeed tutorial `_. cpu_checkpointing: Offloads partitioned activations to CPU if ``partition_activations`` is enabled. contiguous_memory_optimization: Copies partitioned activations so that they are contiguous in memory. Not supported by all models. synchronize_checkpoint_boundary: Insert :func:`torch.cuda.synchronize` at each checkpoint boundary. load_full_weights: True when loading a single checkpoint file containing the model state dict when using ZeRO Stage 3. This differs from the DeepSpeed checkpoint which contains shards per worker. ztTo use the `DeepSpeedStrategy`, you must have DeepSpeed installed. Install it by running `pip install -U deepspeed`.)r,rJrKrVrWNr0r1r3r2r4r5r6r>r7r8r9r:r;r<r=rQrRrSrTr.r@rArBrCrDrEr?rF)_DEEPSPEED_AVAILABLE ImportErrorsuper__init___backward_sync_control _load_configrH_create_default_configr"_config_initializedutilsloggingloggersetLevelr/rUrLrMrNrOrP_deepspeed_engine).selfr,r-r.r/r0r1r2r3r4r5r6r7r8r9r:r;r<r=r>r?r@rArBrCrDrErFrGrHrIrJrKrLrMrNrOrPrQrRrSrTrUrVrWr" __class__Y/home/ubuntu/.local/lib/python3.10/site-packages/lightning_fabric/strategies/deepspeed.pyr\5s1       ! zDeepSpeedStrategy.__init__cCs2t|jtsJ|jd}|duo|ddkS)Nr-r.) isinstancerHdictget)rfr-ririrj zero_stage_3)s zDeepSpeedStrategy.zero_stage_3cCs|j|jdS)N) num_replicasrank) world_size global_rankrfririrjdistributed_sampler_kwargs/sz,DeepSpeedStrategy.distributed_sampler_kwargsr!cC|jSN)rertririrjmodel4szDeepSpeedStrategy.modelmodule optimizerscCsJt|dkrtdt|d|||d\|_}||j|gfS)aSet up a model and multiple optimizers together. Currently, only a single optimizer is supported. Return: The model wrapped into a :class:`deepspeed.DeepSpeedEngine` and a list with a single deepspeed optimizer. r*z>Currently only one optimizer is supported with DeepSpeed. Got z optimizers instead.r)len ValueError_initialize_enginere'_set_deepspeed_activation_checkpointing)rfryrz optimizerririrjsetup_module_and_optimizers8s  z-DeepSpeedStrategy.setup_module_and_optimizerscCs||\|_}|jS)zxSet up a module for inference (no optimizers). For training, see :meth:`setup_module_and_optimizers`. )r}re)rfry_ririrj setup_moduleOszDeepSpeedStrategy.setup_modulercCs t|)zOptimizers can only be set up jointly with the model in this strategy. Please use :meth:`setup_module_and_optimizers` to set up both module and optimizer together. )NotImplementedError_err_msg_joint_setup_required)rfrririrjsetup_optimizerYs z!DeepSpeedStrategy.setup_optimizer empty_initcsT|jr|durtd|d|}t}|js#|tj|d|||S)NFz `empty_init=zN` is not a valid choice with `DeepSpeedStrategy` when ZeRO stage 3 is enabled.)r)rormodule_sharded_contextr enter_contextr[module_init_context)rfrmodule_sharded_ctxstackrgrirjrbs  z%DeepSpeedStrategy.module_init_contextcCs*ddl}|js J|jj|j|j|jdS)Nr)enabledr/config_dict_or_path)r"r`zeroInitror/rHrfr"ririrjros z(DeepSpeedStrategy.module_sharded_contextpathstatestorage_optionsfiltercs|durtd|durtdt|}t|dkrtdt|dkr(td|d}||}|jdur;||jfn|ffdd |D}t||j|id }|j ||d d dS) aSave model, optimizer, and other state in a checkpoint directory. Args: path: A path to where the files should be saved state: A dictionary with contents to be saved. If the dict contains modules or optimizers, their state-dict will be retrieved and converted automatically. storage_options: Unused by this strategy, since it doesn't use a ``CheckpointIO`` plugin. filter: Unsupported. Raises: TypeError: If the unused ``storage_options`` gets passed. ValueError: When no :class:`deepspeed.DeepSpeedEngine` objects were found in the state, or when multiple :class:`deepspeed.DeepSpeedEngine` objects were found. Nz`DeepSpeedStrategy.save_checkpoint(..., storage_options=...)` is not supported because `DeepSpeedStrategy` does not use the `CheckpointIO`.z`DeepSpeedStrategy.save_checkpoint(..., filter=...)` is not supported because `DeepSpeedStrategy` manages the state serialization internally.raCould not find a DeepSpeed 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 DeepSpeed engine modules in the given state. Saving checkpoints with DeepSpeed 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.csi|] \}}|vr||qSriri).0kvexcluded_objectsrirj sz5DeepSpeedStrategy.save_checkpoint..)r checkpoint) client_statetag) TypeError!_get_deepspeed_engines_from_stater{r| broadcastritems_validate_state_keys"_convert_stateful_objects_in_statesave_checkpoint)rfrrrrenginesenginerirrjr}s0   z!DeepSpeedStrategy.save_checkpointstrictc st|ttfs |jr|jr||}tj|||dS|s%td|dt |t |}t |dkr7tdt |dkrAtd|d}t rNddl mnddlmtfd d |D}tj|j|d |d |d \}}|durztdt|t|ddh@} t||| d|S)aLoad the contents from a checkpoint and restore the state of the given objects. Args: path: A path to where the file is located state: A dictionary of objects whose state will be restored in-place from the checkpoint path. This should contain exactly one model, and the model must already be set up by DeepSpeed. strict: Whether to enforce that the keys in `state` match the keys in the checkpoint. Returns: Dictionary with the state inside DeepSpeed's engine Raises: ValueError: If no state is provided, when no :class:`deepspeed.DeepSpeedEngine` objects were found in the state, or when multiple :class:`deepspeed.DeepSpeedEngine` objects were found. RuntimeError: If DeepSpeed was unable to load the checkpoint due to missing files or because the checkpoint is not in the expected DeepSpeed format. )rrrz1Got DeepSpeedStrategy.load_checkpoint(..., state=z) but a state with at least a model instance to reload is required. Pass it in like so: DeepSpeedStrategy.load_checkpoint(..., state={'model': model, ...})raCould not find a DeepSpeed 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.r*zFound multiple DeepSpeed engine modules in the given state. Saving and loading checkpoints with DeepSpeed is currently limited to a single model per checkpoint. To load multiple model states, call the load method for each model checkpoint separately.DeepSpeedOptimizerc3s|] }t|tfVqdSrw)rlr)ritemrrirj sz4DeepSpeedStrategy.load_checkpoint..rF)rload_optimizer_statesload_lr_scheduler_statesload_module_strictNzDeepSpeed was unable to load the checkpoint. Ensure you passed in a DeepSpeed compatible checkpoint or a single checkpoint file by setting `DeepSpeedStrategy(..., load_full_weights=True)`.r lr_scheduler)source destinationkeys)rlrrrUrorr[load_checkpointr|_validate_checkpoint_directoryrr{_DEEPSPEED_GREATER_EQUAL_0_14_1 deepspeed.runtime.base_optimizerrdeepspeed.runtimeanyvaluestorchcuda empty_cache RuntimeErrorsetr) rfrrrrroptimzer_state_requestedrrrrgrrjrsJ       z!DeepSpeedStrategy.load_checkpoint@max_norm norm_typeerror_if_nonfinitecCtdNzDeepSpeed handles gradient clipping automatically within the optimizer. Make sure to set the `gradient_clipping` value in your Config.r)rfryrrrrririrjclip_gradients_norms z%DeepSpeedStrategy.clip_gradients_normclip_valcCrrr)rfryrrririrjclip_gradients_value sz&DeepSpeedStrategy.clip_gradients_valuestrategy_registryc Cs|jd|dd|jd|ddd|jd|d dd d |jd |d dd|jd|ddd d |jd|ddd|jd|ddd d d|jd|ddd d dddd dS)Nr"zDefault DeepSpeed Strategy) descriptiondeepspeed_stage_1z#DeepSpeed with ZeRO Stage 1 enabledr*)rr.deepspeed_stage_1_offloadz5DeepSpeed with ZeRO Stage 1 and optimizer CPU OffloadT)rr.r0deepspeed_stage_2z#DeepSpeed with ZeRO Stage 2 enabledr%deepspeed_stage_2_offloadz&DeepSpeed ZeRO Stage 2 and CPU Offloaddeepspeed_stage_3zDeepSpeed ZeRO Stage 3rkdeepspeed_stage_3_offloadz&DeepSpeed ZeRO Stage 3 and CPU Offload)rr.r0r1deepspeed_stage_3_offload_nvmez'DeepSpeed ZeRO Stage 3 and NVMe Offloadnvme)rr.r0r1r/r2r7)register)clsrririrjregister_strategies)sJ z%DeepSpeedStrategy.register_strategiesrxcCsNddl}tdd|}|jtj|jjd|j|||dd\}}}}||fS)zInitialize one model and one optimizer with an optional learning rate scheduler. This calls :func:`deepspeed.initialize` internally. rNcSrvrw) requires_grad)pririrj^sz6DeepSpeedStrategy._initialize_engine..) device_rankF)argsrHrxmodel_parametersrdist_init_required) r"r parameters initializeargparse Namespace root_deviceindexrH)rfrxrr"rdeepspeed_enginedeepspeed_optimizerrririrjr}Rs z$DeepSpeedStrategy._initialize_enginecs0t|jtstd|jjjdtdS)Nz;The DeepSpeed strategy is only supported on CUDA GPUs but `z ` is used.)rlr,rrrh__name__r[setup_environmentrtrgrirjris z#DeepSpeedStrategy.setup_environmentcCsJ|jdusJt|jt|||js#|d|_dSdSNT)rJ _validate_device_index_selectionr_set_world_ranks_init_deepspeed_distributedr`_format_configrtririrj_setup_distributedrs  z$DeepSpeedStrategy._setup_distributedcCsrddl}|jdus Jtdkr(|td|jd|jdd|j| |_ |j |j |jj ddS)NrWindowsz1initializing deepspeed distributed: GLOBAL_RANK: z , MEMBER: r*/)distributed_port) r"rKplatformsystem_set_node_environment_variablesrinforsrr_get_process_group_backend_process_group_backendinit_distributed main_portrririrjr}s  z-DeepSpeedStrategy._init_deepspeed_distributedcCsb|jdusJ|jjtjd<t|jjtjd<t|jtjd<t|jtjd<t|jtjd<dS)N MASTER_ADDR MASTER_PORTRANK WORLD_SIZE LOCAL_RANK) rK main_addressosenvironstrrrsrr local_rankrtririrjrs z1DeepSpeedStrategy._set_node_environment_variablescCsdddl}t|jts J|jdr0|jd}|jjd|d|d|d|dddSdS)Nractivation_checkpointingrQrSrRprofile)mpu_rQcontiguous_checkpointingcheckpoint_in_cpur)r"rlrHrmrn checkpointing configure)rfr"checkpoint_configririrjr~s   z9DeepSpeedStrategy._set_deepspeed_activation_checkpointingc CsJ|jdur td|jddt|j|jj|j|j|j|j|j ddS)NzTo use DeepSpeed you must pass in a DeepSpeed config dict, or a path to a JSON config. See: https://lightning.ai/docs/pytorch/stable/advanced/model_parallel.html#deepspeedtrain_micro_batch_size_per_gpur*rHrVrLrNrPrMrO) rHr| setdefault_format_precision_configrVrLrNrPrMrOrtririrjrs  z DeepSpeedStrategy._format_config zero_kwargscKsx||||d|||||dd}|r4|}|r|| ||d|d<| r,| | | | ||d|d<|||d|r:||d <|S) N)rQrRrSrT)r9r:r;r<r=)raio)devicer3 buffer_countr>r0)rr3r buffer_sizer6r> offload_param)rFr-r )update)rfr-rFrGrQrRrSrTr0r1r3r2r4r5r6r7r8r>r9r:r;r<r=rcfg zero_configririrjr_sF  z(DeepSpeedStrategy._create_default_configckptcs>ddld dtjjdtddffdd |dd dS) a8Overrides the normal load_state_dict behaviour in PyTorch to ensure we gather parameters that may be sharded across processes before loading the state dictionary when using ZeRO stage 3. This is then automatically synced across processes. Args: ckpt: The ckpt file. rNryprefixrXc sg}g}g}d}t|dd}|}|dur||_|dur!in ||ddi}jjt|jddddjrI|j |||d|||d Wdn1sSwY|j D]\}} | durn| ||d q]dS) N state_dict _metadataF)recurser) modifier_rankT)rrlocal_metadatar missing_keysunexpected_keys error_msgs.) getattrcopyrrnrGatheredParameterslistris_global_zero_load_from_state_dict_modulesr) ryrr!r"r#rmetadatar namechildrr"loadrfrirjr0s6   z3DeepSpeedStrategy._restore_zero_state..load)r)r)r"rnnrr)rfryrrir/rj_restore_zero_states & z%DeepSpeedStrategy._restore_zero_statecCs|dur|jtjvrtd|jdtj|j}t|ttfrFtj|s-t d|t | }t |}Wdn1sAwYt|t sQ|dusQJ|S)Nz"Loading DeepSpeed config from set z environment variablezHYou passed in a path to a DeepSpeed config but the path does not exist: )DEEPSPEED_ENV_VARrrrrlrrrisfileFileNotFoundErroropenjsonr0rm)rfrHfririrjr^%s    zDeepSpeedStrategy._load_config)rXr!rw)NNr)rT)rXN)9r __module__ __qualname__r3rbWARNr rboolintrrrr rr rrrfloatrr\propertyrorrurxrrr rrrrrrrrrTensorrr classmethodrrr}rrrrr~rr_r r2r^ __classcell__ririrgrjr#2s6      !"#$%&'()*+,-.u      @" S      *            D:-r#rrXr!cs4ddlmtdd|D}fdd|DS)Nrr css"|] }t|tr|VqdSrw)rlrmodules)rryririrjr7s z4_get_deepspeed_engines_from_state..csg|] }t|r|qSri)rl)rrr rirj 8sz5_get_deepspeed_engines_from_state..)r"r!rr)rrCrir rjr4s rcCs4hd}||}|rtdd|dSdS)N> ry ds_configr ds_version buffer_names global_stepsr param_shapes dp_world_size mp_world_size skipped_stepsglobal_samplessparse_tensor_module_nameszYour state has keys that collide with DeepSpeed's internal engine state. This could result in your values being overwritten by DeepSpeed. Consider changing the name of these keys to something else: z, ) intersectionrrjoin)rdeepspeed_internal_keyscolliding_keysririrjr;srrJcCsPdd|D}ttt|}||kr&td|dddd|DddS) NcSsg|]}|jqSri)r)rrririrjrDWsz4_validate_device_index_selection..zThe selected device indices z don't match the local rank values of processes. If you need to select GPUs at a specific index, set the `CUDA_VISIBLE_DEVICES` environment variable instead. For example: `CUDA_VISIBLE_DEVICES=,css|]}t|VqdSrw)r)riririrjr]sz3_validate_device_index_selection..z`.)r(ranger{rrP)rJselected_device_indicesexpected_device_indicesririrjrVsrrcCs|o |dS)zVHeuristic check whether the path points to a top-level DeepSpeed checkpoint directory.r)is_dir)rririrj_is_deepspeed_checkpointasrYcCszt|}t|}d|}|s;t|j}|r t|d|j|o)t|jj}|r7t|d|jjt|dS)zeValidates that the path points to a DeepSpeed checkpoint directory and suggests fixes for user error.z7The provided path is not a valid DeepSpeed checkpoint: ze. It looks like you passed the path to a subfolder. Try to load using this parent directory instead: z. It looks like you passed the path to a file inside a DeepSpeed checkpoint folder. Try to load using this parent directory instead: N)rrYparentr5is_file)rpath_is_ds_checkpointdefault_messageparent_is_ds_checkpointparent_parent_is_ds_checkpointririrjrfs(   rrHrVrLrNrPrMrOcCsbd|vr|dvrtdd|||||d|d<dSd|vr-|dvr/tdd di|d<dSdSdS) Nfp16)z16-mixedz16-truezOEnabling DeepSpeed FP16. Model parameters and inputs will be cast to `float16`.T)rrLrMrNrOrPbf16)z bf16-mixedz bf16-truezPEnabling DeepSpeed BF16. Model parameters and inputs will be cast to `bfloat16`.r)rr ririrjrs r)Grr7rbrr contextlibr itertoolsrpathlibrtypingrrrrr r r r r rr lightning_utilities.core.importsrtorch.nnr torch.optimrtyping_extensionsrlightning_fabric.acceleratorsrr9lightning_fabric.plugins.environments.cluster_environmentr"lightning_fabric.plugins.precisionrlightning_fabric.strategies.ddpr$lightning_fabric.strategies.registryr$lightning_fabric.strategies.strategyr&lightning_fabric.utilities.distributedrlightning_fabric.utilities.loadr$lightning_fabric.utilities.rank_zerorrlightning_fabric.utilities.seedr lightning_fabric.utilities.typesrr"r!rYrr#rrrrrr<rYrr>r=rriririrjst    0               #