o Ti@sdZddlZddlZddlZddlmZddlZddlZddlm Z ddl m Z ddl m Z ddlmZmZmZddlmZmZdd lmZdd lmZdd lmZd adadad ada da!ga"ga#ga$ga%da&d a'd a(d a)d a*d a+dZ,dBddZ-dCddZ.GdddZ/e/Z0ddZ1ddZ2ddZ3ddZ4ddZ5dBd d!Z6d"d#Z7d$d%Z8d&d'Z9d(d)Z:d*d+Z;d,d-ZZ?d0d1Z@ejAd2d3ZBd4d5ZCd6d7ZDd8d9ZEdBd:d;ZFdd?ZHd@dAZIdS)EaV Use to partition the activations stored for backward propagation Therefore reduces the memory consumption Also implements CPU checkpointing and contiguous memory checkpointing Reduces memory consumption and memory fragmentation Code for rng checkpointing taken from NVIDIA Megatron-LM mpu/random.py b886b7bb972afe72bac0f5de4f42a4a7bae8ebef N)comm)_C)DeepSpeedConfig)logger)copy_to_devicemove_to_devicesee_memory_usage)SynchronizedWallClockTimerFORWARD_GLOBAL_TIMER)bwc_tensor_model_parallel_rank)get_accelerator)compilerFzmodel-parallel-rngcCst|tr8g}|D]*}t|tjs||q |j}|dur%|j|d}n|}|}||_||q t|Stdt |j )Ndevicez@Only tuple of tensors is supported. Got Unsupported input type: ) isinstancetupletorchTensorappend requires_gradtodetach RuntimeErrortype__name__)inputsroutinprxr l/home/ubuntu/.local/lib/python3.10/site-packages/deepspeed/runtime/activation_checkpointing/checkpointing.pydetach_variableDs    r"csttdrttjrfdd}n-dkrttntt r*tntt r8ttfdd}t |dS)aUSets the random number generator state of the current GPU. Arguments: new_state (torch.ByteTensor): The desired state This function is adapted from PyTorch repo (torch.cuda.set_rng_state) #ignore-cuda with a single change: the input state is not cloned. Cloning caused major performance issues for +4 GPU cases. _cuda_setRNGStatecs<ttWddS1swYdSN)r rrr$r r new_stater r!cbfs "z_set_cuda_rng_state..cbr#cs2j}|dur t}t|}|dSr%)indexr current_devicedefault_generator set_state)idxr+r&r r!r(rs   N) hasattrrcallabler$rrr device_namerstrint lazy_call)r'rr(r r&r!_set_cuda_rng_state[s    r4c@sJeZdZdZddZddZddZdd Zd d Ze j e fd d Z dS)CudaRNGStatesTrackeraTracker for the cuda RNG states. Using the `add` method, a cuda rng state is initialized based on the input `seed` and is assigned to `name`. Later, by forking the rng state, we can perform operations and return to our starting cuda state. cCsi|_t|_dSr%states_setseeds_selfr r r!__init__s zCudaRNGStatesTracker.__init__cCsi|_t|_dS)z&Set to the initial state (no tracker).Nr6r:r r r!resets zCudaRNGStatesTracker.resetcCs t|jS)z{Get rng states. Copy the dictionary so we have direct pointers to the states, not just a pointer to the dictionary.)copyr7r:r r r! get_statess zCudaRNGStatesTracker.get_statescCs ||_dS)zhSet the rng states. For efficiency purposes, we do not check the size of seed for compatibility.N)r7)r;statesr r r! set_statess zCudaRNGStatesTracker.set_statescCsn||jvr td||j|||jvrtd|t}t|t|j|<t|dS)zTrack the rng state.zseed {} already existsz cuda rng state {} already existsN) r9 Exceptionformataddr7r get_rng_state manual_seedr4)r;nameseedorig_rng_stater r r!rDs      zCudaRNGStatesTracker.addc csr||jvr td|t}t|j|zdVWt|j|<t|dSt|j|<t|w)zVFork the cuda rng state, perform operations, and exit with the original state.zcuda rng state {} is not addedN)r7rBrCr rEr4)r;rGorig_cuda_rng_stater r r!forks    zCudaRNGStatesTracker.forkN) r __module__ __qualname____doc__r<r=r?rArD contextlibcontextmanager _MODEL_PARALLEL_RNG_TRACKER_NAMErKr r r r!r5|sr5cCtS)zGet cuda rng tracker.)_CUDA_RNG_STATE_TRACKERr r r r!get_cuda_rng_trackersrTc Csltt}|d}||}|}tdkr$tdt|t||t t |t t |dS)apInitialize model parallel cuda seed. This function should be called after the model parallel is initialized. Also, no get_accelerator().manual_seed should be called after this function. Basically, this is replacement for that function. Two set of RNG states are tracked: default state: This is for data parallelism and is the same among a set of model parallel GPUs but different across different model parallel groups. This is used for example for dropout in the non-model-parallel regions. model-parallel state: This state is different among a set of model parallel GPUs, but the same across data parallel groups. This is used for example for dropout in model parallel regions. rz> initializing model parallel cuda seeds on global rank {}, model parallel rank {}, and data parallel rank {} with model parallel seed: {} and data parallel seed: {}N)r mpudistget_rankrinforCget_data_parallel_rankrSr=r rFrDrQ)rHtp_rankoffsetmodel_parallel_seeddata_parallel_seedr r r!model_parallel_cuda_manual_seeds   r_cCsNtt}|d|}tt|WddS1s wYdS)NrU)r rVrSrKr rF)rHr[r]r r r!"model_parallel_reconfigure_tp_seeds   "r`cCs |}|t}|t}t|Sr%)numelmp_sizemp_rankr2)itemsizepartition_sizestartr r r!get_partition_startsrhcCs,|}|tdksJd|t}t|S)NrzJDoesn't handle if partition activation if item is not divisible by mp size)rarbr2)rdrerfr r r!get_partition_sizesric CsPt|ddksJdt|g}tt|d}t|D]}|d|}|d|d}t|s9||qtdusAtdkrY|t| }|durS| |}||q| }|t}|durpt j |g|j|d} n t j |g|j|jd} | d|t|} | |tj| | td| t| } | j|_||qt|S)Nrz,Expected even count of tensors, instead got rdtyper)group)lenr2rangeis_activation_to_checkpointrmp_grouprbviewlistnumpyrrarzerosrlrnarrowrccopy_rWall_gather_into_tensordatar) tensorsrrnum_argsirdrerf tensor_size flat_tensorpart input_tensorr r r!gather_partitioned_activations s6"       rcCsVdd|D}dd|D}dd|D}t|tur&t|t|t|fS|||fS)a Separate objects in list/tuple into tensors and non-tensors and create a mapping to enable re-aggregation. The order of tensors and non-tensors is preserved in their respective output groups. Parameters: all_objects (list/tuple): Objects containing tensors and non-tensors to be split. Returns: tuple: Containing tensors, non-tensors, and bools of whether each position in original list/tuple was a tensor. cSsg|] }t|r|qSr r is_tensor.0vr r r! =z#extract_tensors..cSsg|] }t|s|qSr rrr r r!r>rcSsg|]}t|qSr rrr r r!r?s)rr) all_objectstensor_objectsnon_tensor_objects tensor_flagsr r r!extract_tensors1s    rc Csg}d}d}d}tr g}d}|D]}|rd}q|}||qn|}|D]} | r4||||d7}q$||||d7}q$t|S)a Merge two lists (or tuples) of tensors and non-tensors using a mapping of positions in merged list (or tuple). Parameters: tensor_objects (list/tuple): Tensors to merge. non_tensor_objects (list/tuple): Non-tensors to merge. tensor_flags (list/tuple): Indicates whether each position in output is a tensor. Returns: tuple: Merge of tensors and non-tensors rNFr)PARTITION_ACTIVATIONSrr) rrrmerged_objects tensor_idxnon_tensor_idxreal_tensor_flags previous_flagflagrr r r! merge_tensorsEs,    rcCsBt|d pt|do|jdk}t|o |o |tko |S)z1 Is an activation to be checkpointed no_checkpointingF)r.rrris_floating_pointrarb)rd extra_flagr r r!rpos"rpc sg}d}t|D]\}}t|s|||d7}q||}t||ddt| |r;t dnj |r|t t kr^fddttD}t |tdnt |duryfddttD}|t |<dt|<dt |t|jtdt |t|jjdttjt |t|j<t |t|jj} t|dt|<|| qtrn|q|S)Nrrr#cpuc&g|]}tdjgjdqSr rkrtensor new_emptyrlr_ buffer_device partitionrfr r!rz)partition_activations..crrrrrr r!rr) enumeraterprrir contiguousrrrvrhclonerrrncontiguous_data_buffersro num_layers data_offsetsryshaper2mmapPAGESIZE element_sizerwCPU_CHECKPOINTr) argscpu_checkpointcontiguous_checkpointrnum_non_fp_tensors arg_indexrdr| tensor_listcontiguous_partitionr rr!partition_activationsysF $       rc Csrg}d}tt||D]\}\}}t|rt|nd}t|s2|||||d7}q tjg|j dj |_ |j |_ ||||} |r| } | t tkrptd} t| j| tg|j|j dtdnt| durtd} | j| tg|j|j dt| <dt| <t| dt| | j |j } | |} t| | t| <|| q ||q |S)Nrrrr rk)rziprrrrerpremptyrry saved_datararncontiguous_size_buffersrrrl size_offsetsrvrwview_as) rrrnew_argsrrargrrer|ratmpcontiguous_sizer r r!(get_partitioned_activations_for_backwards>            rcCs^g}tt||D]#\}\}}t|s||q tjg|jdj|_|j|_||q |S)Nr) rrrprrrrryr)rrrr|rrr r r! get_cpu_activations_for_backwards  rc@s(eZdZdZeddZeddZdS)CheckpointFunctionaThis function is adapted from torch.utils.checkpoint with two main changes: 1) torch.cuda.set_rng_state is replaced with `_set_cuda_rng_state` #ignore-cuda 2) the states in the model parallel tracker are also properly tracked/set/reset. 3) Performance activation partitioning, contiguous memory optimization 4) CPU Checkpointing 5) Profile forward and backward functions c sfdd}tr ttdurtrtatrtt|_t }tj |d}t r6t |t t}n t rBt|tdtd}t||td}t_t_t_tdddt ||} Wdn1sswYtd dd~t rt||t} t| d d ksJd t| || nt rt||} || n||trttttgtrtt | r| !s| gng} nd d| D} j"| t | r|| g7}| S|| 7}t#| d\} } } t$| S)Ncs&t|d\}}}|_|_|_dS)Nr)rdeepspeed_saved_tensorsnon_tensor_argsr)all_args tensor_argsrrctxr r!save_args_for_backwards z:CheckpointFunction.forward..save_args_for_backwardrrrcriterion_funcz#Before running forward on the layerFforce"After running forward on the layerrjr2save_for_backward called with odd number of args, cSs"g|] }t|r|s|qSr )rrr)ror r r!r9s"z.CheckpointFunction.forward..r)% SYNCHRONIZEr synchronizetimers PROFILE_TIMETimersr rg run_functioncurrent_device_nameStreamrrrCONTIGUOUS_CHECKPOINTINGrrrrprEfwd_cpu_rng_statefwd_cuda_rng_staterTr?fwd_cuda_rng_state_trackerrno_gradrrnrstoplogrrmark_non_differentiablerr) rr all_outputsrr cuda_devicetransport_streamr inputs_cudaoutputsrnon_grad_outputsrr rr!forwards^             "          zCheckpointFunction.forwardcGstdddtr ttrtdtr&tD]}g}qgaga ga ga tdddt j s5tdt}tj|d}|jD]}|dur_t|dr_|jdur_|j|j|_d|_qDtrrt|jtri|ndd}t|}ntrt|j|t}t|}n|j}t|}t||j|j d }t !}t!} t"#} t $|j%t&|j't"(|j)td ddt * |j+|} Wdn1swYtd ddt $|t&| t"(| t,| t j-r| f} t.| d \} } } g} g}t/| |D]\}}|j0r| 1||1|qtd ddt j 2| |d|_d|_d|_ tdddtr.save_args_for_backwardrrrrjrrc@seZdZdZdS)z(non_reentrant_checkpoint..Holderz:the place holder object used as activations to save memoryN)rrLrMrNr r r r!Holdersrcs0}t||jr|jr||S)zused to record the activation order in the `weak_holder_list` the activation order in holder list is consistent between the first forward and recomputing forward. * the jit compiled forward will break the order consistency * )rweakrefrefris_leaf)tensor_from_forwardres)r leaf_tensorsweak_holder_listr r!checkpoint_packs   z1non_reentrant_checkpoint..checkpoint_packc stdkrd fdd}dd}tdddtr!ttr)td tr:t D]}g}q-ga ga ga ga td ddt jsItd t}tj|d }trfttr]|nd d }t|}ntrst|t}t|}n}t|}t|d}t }t} t} t tttdddt $t jj!"|| |} Wd n1swYWd n1swYtdddt |t| t| d d d |vrtd|S)z'retrieve the activations from recomputercs<d7ddurdS|}|d<dS)zsave recompute activationsrN)r)tensor_from_replaydetached_activations)storageunpack_counterr r r! replay_pack)s zHnon_reentrant_checkpoint..checkpoint_unpack..replay_packcSstd)z$recompute graph need not to backwardz.checkpoint_unpack..replay_unpackrFrrrrrNrrrzAttempt to retrieve a tensor saved by autograd multiple times without checkpoint recomputation being triggered in between, this is not currently supported.)#rnrrr rrrrgrrrrrrrrrrrrrrr"rrprrErTr?rr4rArgraphsaved_tensors_hooks) holder_from_backwardrrrrrrrrrr_unused) rfunctionrrrrrrr )rr!checkpoint_unpack sr                    z3non_reentrant_checkpoint..checkpoint_unpackcsVd7tkr'tdddtrtdtdgtr)tdSdSdS)z'the hook registered to all leaf tensorsrrFrrN) rnrrrrrrr r) _nonuse_grads)backward_visited_leaf_nodesr r r!after_backward_hooks    z5non_reentrant_checkpoint..after_backward_hookrFrr)#rr rrrrr rgrrrrrrrrrrprErTr?rrnrrWeakKeyDictionaryrrr register_hookrrrrr)rrrrrrrrr rrr leaf_tensorrr ) rrrrrrrr rrrr r!non_reentrant_checkpointsn       "   k        r cGs4g}tj||g|Rt|dkr|dSt|S)zhCheckpoint a model or part of the model. This has been directly copied from torch.utils.checkpoint. rr)rapplyrnr)rrrr r r! checkpoints  r"cCs*|atdkrtdtddSdS)Nrz$**************Partition Activations z ************)rrWrXrrY)partition_activationr r r!#partition_activations_in_checkpoints r$cCs|adSr%)r)nlayersr r r!set_num_layersrr&cCs*trtD]}g}qgagagagadSdS)aResets memory buffers related to contiguous memory optimizations. Should be called during eval when multiple forward propagations are computed without any backward propagation that usually clears these buffers. Arguments: None Return: None N)rrrrr)rr r r!r=s r=cCsPt||dj}tdkrt||ja|j a |j a |j a|ja|jadS)NrVr)ractivation_checkpointing_configrWrXrrYreprrrcontiguous_memory_optimizationrnumber_checkpointsrcpu_checkpointingrsynchronize_checkpoint_boundaryrprofiler)configrVr r r!_configure_using_config_files  r0cCs dadadadadadadadS)NFT)rrrrrrdeepspeed_checkpointing_enabledr r r r!_configure_defaultssr2cCsNt|dur |a|durt|td|dur|a|dur|a|dur%|a|dur+|a|dur1|a|dur7|atr?ts?JdtrItdusIJdtdurkt tdr_t a t a tan ta ta tatdddtd krtd td td ttd tdtdtdttdtdSdS)aConfigure DeepSpeed Activation Checkpointing. Arguments: mpu_: Optional: An object that implements the following methods get_model_parallel_rank/group/world_size, and get_data_parallel_rank/group/world_size deepspeed_config: Optional: DeepSpeed Config json file when provided will be used to configure DeepSpeed Activation Checkpointing partition_activations: Optional: Partitions activation checkpoint across model parallel GPUs when enabled. By default False. Will overwrite deepspeed_config if provided contiguous_checkpointing: Optional: Copies activation checkpoints to a contiguous memory buffer. Works only with homogeneous checkpoints when partition_activations is enabled. Must provide num_checkpoints. By default False. Will overwrite deepspeed_config if provided num_checkpoints: Optional: Number of activation checkpoints stored during the forward propagation of the model. Used to calculate the buffer size for contiguous_checkpointing Will overwrite deepspeed_config if provided checkpoint_in_cpu: Optional: Moves the activation checkpoint to CPU. Only works with partition_activation. Default is false. Will overwrite deepspeed_config if provided synchronize: Optional: Performs get_accelerator().synchronize() at the beginning and end of each call to deepspeed.checkpointing.checkpoint for both forward and backward pass. By default false. Will overwrite deepspeed_config if provided profile: Optional: Logs the forward and backward time for each deepspeed.checkpointing.checkpoint invocation. Will overwrite deepspeed_config if provided Returns: None Nr'zContiguous Checkpointing is only available with partitioned activations. Set partitioned activations to true in deepspeed configzFMust specify the number of layers with contiguous memory checkpointingget_tensor_model_parallel_rankzAfter configurationFrrz$Activation Checkpointing Informationz----Partition Activations z, CPU CHECKPOINTING z$----contiguous Memory Checkpointing z with z total layersz----Synchronization z$----Profiling time in checkpointing )r2rVr0rrrrrrr.r3rc$get_tensor_model_parallel_world_sizerbget_tensor_model_parallel_grouprqget_model_parallel_rankget_model_parallel_world_sizeget_model_parallel_grouprrWrXrrY)mpu_deepspeed_configrcontiguous_checkpointingnum_checkpointscheckpoint_in_cpurr.r r r! configuresJ2       r>cCrR)zTrue if deepspeed activation checkpointing has been configured by calling deepspeed.checkpointing.configure, else returns false Arguments: None Return: True of configured, else False )r1r r r r! is_configuredls r?r%)r#)NNNNNNN)JrNr>rrO deepspeedrrWrrrdeepspeed.runtime.configrdeepspeed.utilsrdeepspeed.runtime.utilsrrrdeepspeed.utils.timerr rr deepspeed.utils.bwcr deepspeed.acceleratorr deepspeed.runtimer r1rVrcrbrqrrrrrrrrrrrrQr"r4r5rSrTr_r`rhrirrrrprrrrFunctionrr disabler"r$r&r=r0r2r>r?r r r r!s          !E) '* 7*Yt    g