U yh@sddlZddlmZmZddlmZddlmZddlm Z m Z m Z m Z m Z mZmZmZddlZddlmZddlmZddlmZmZmZdd lmZd d d d ddgZeeZ GdddeZ!Gddde Z"GdddeZ#d"e ej$ee%dddZ&d#e ej$ee%e e'ej(fdddZ)Gdd d e#Z*Gdd d e*Z+Gdd d e*Z,Gdd d e#Z-Gd dde-Z.Gd!dde-Z/dS)$N)ABCabstractmethod) defaultdict)Enum)AnyCallableDictList NamedTupleOptionalTupleUnion)record_function) merge_chunkssplit_args_kwargs_into_chunksTensorChunkSpec)_PipelineStageBasePipelineScheduleSinglePipelineScheduleMulti Schedule1F1B ScheduleGPipeScheduleInterleaved1F1BScheduleLoopedBFSc@seZdZdZdZddZdS)_ComputationTypercCs|tjkrdSdSdS)NFB)rFORWARDselfr!X/var/www/html/venv/lib/python3.8/site-packages/torch/distributed/pipelining/schedules.py__str__"s z_ComputationType.__str__N)__name__ __module__ __qualname__rBACKWARDr#r!r!r!r"rsrc@s.eZdZUeed<eed<eed<ddZdS)_Actioncomputation_typemicrobatch_index stage_indexcCs|j|jd|jS)NZ_s)r)r*r+rr!r!r"__repr__.sz_Action.__repr__N)r$r%r&r__annotations__intr,r!r!r!r"r()s r(c @s*eZdZdeeedejfeee dfee e e fee e e e fee fdddZddZdd Zd d Zedeeeeeeeed d dZedddeedddZdeeeeeeeed ddZddZd ee dfee e e fdddZee e dddZdS)!_PipelineScheduleN.n_microbatchesloss_fnargs_chunk_speckwargs_chunk_specoutput_merge_speccCsH||_||_||_||_||_|jdk |_g|_td|j j dS)NzUsing ) _n_microbatches_loss_fn_args_chunk_spec_kwargs_chunk_spec_output_merge_spec _has_backward_internal_lossesloggerinfo __class__r$)r r1r2r3r4r5r!r!r"__init__3s  z_PipelineSchedule.__init__cCs,|jr(|jr(||||}|j|dSN)is_lastr; _compute_lossr<append)r stageoutput target_mbsmb_indexlossr!r!r"_maybe_compute_lossPs z%_PipelineSchedule._maybe_compute_losscCsjd|kot|jkn}|jr8|jr8|r8|j|St|jdkrb|sbtd|d|jndSdS)NrzLoss for microbatch z6 is not available. Available losses for microbatches: )lenr<rBr; RuntimeError)r rErHZ valid_indexr!r!r"_maybe_get_lossUs z!_PipelineSchedule._maybe_get_losscCs|t|ts|g}tdd|D}|rn|dk rnt|j|jkrZtd|jdt|j|||j|jdS)zB Update the losses to those in the internal state css|] }|jVqdSrA)rB.0rEr!r!r" hsz3_PipelineSchedule._update_losses..N Expecting z losses but got ) isinstancelistanyrKr<r6rLclearextend)r stageslossesZcontains_last_stager!r!r"_update_lossesas   z _PipelineSchedule._update_lossesarg_mbs kwarg_mbsrGrXcCstdS)z Run one iteration of the pipeline schedule with list of microbatches. Will go through all the microbatches according to the schedule implementation. Args: microbatches: list of microbatch args. NNotImplementedError)r r[r\rGrXr!r!r"_step_microbatchesxsz$_PipelineSchedule._step_microbatchestargetrXrXcOstdSa Run one iteration of the pipeline schedule with *whole-batch* input. Will chunk the input into microbatches automatically, and go through the microbatches according to the schedule implementation. args: positional arguments to the model (as in non-pipeline case). kwargs: keyword arguments to the model (as in non-pipeline case). target: target for the loss function. losses: a list to store the losses for each microbatch. Nr])r rarXargskwargsr!r!r"steps z_PipelineSchedule.stepcstdfdd }|dk r&||dn dgj}|dk rF||dn igj}|dk rd||d|dk rt|tstd t|||fS) z* Pre-process/check inputs )namecsRt|ts t|dt|t|jkrNtdjd|dt|dS)Nz must be a list but got a rQ z but got )rRrS TypeErrortyperKr6 ValueError)Zmbsrgrr!r"check_type_and_lens  z;_PipelineSchedule._check_inputs..check_type_and_lenNr[r!r\rGz losses must be a list but got a )strr6rRrSrirj)r r[r\rGrXrlr!rr" _check_inputss       z_PipelineSchedule._check_inputscCs |||SrA)r7)r rFrar!r!r"rCsz_PipelineSchedule._compute_loss)rdrecCsF|s|r*t|||j|j|j\}}||fSdg|jig|jfSdS)zj Splits a full-batch input into chunks (i.e. microbatches) and returns the chunks r!N)rr6r8r9)r rdre args_split kwargs_splitr!r!r" _split_inputss z_PipelineSchedule._split_inputs) output_chunksreturncCs t||jS)z Merge output chunks back to a batch state. If output_merge_spec is None, the utility will merge output chunks by dimension 0 (batch dim). )rr:)r rrr!r!r"_merge_outputssz _PipelineSchedule._merge_outputs)NNNN)NNNN)NNNN)N)r$r%r&r.r rtorchZTensorr rrrmr rr@rJrMrYrr r_rfrnrCrqrtr!r!r!r"r/2sX   &  r/)p2p_opsdesccCsDt|dkrdS|r|dnd}td||t|S)zt Simple wrapper over batch_isend_irecv from torch.distributed, which just adds a descriptive logger on top. rNz, z batch_p2p )rKr=debugdistZbatch_isend_irecvpop)rvrwZdesc_strr!r!r" _batch_p2ps  r|)rvrwrscCs`tt}i}t|dkr|S|D]}||j|q t|D]\}}t||d||<qB|S)z Sorts the list of P2P ops by the peer rank, and then calls batch_isend_irecv. Return a dictionary of works by peer rank. This function helps us avoid hangs in case of skip connections. rrw)rrSrKpeerrDsorteditemsr|)rvrwZ ops_by_peerZ work_by_peeropr~opsr!r!r"_sorted_batch_p2ps  rc seZdZdZd eeeeeee dfee e e fee e e e fee fdfdd Zdddeedd d ZZS) rz Base class for single-stage schedules. Implements the `step` method. Derived classes should implement `_step_microbatches`. N.)rEr1r2r3r4r5csLtj|||||d||_|j|_|j|j_|||jrH||dS)Nr0) superr@_stage num_stages _num_stagesr; has_backward_prepare_forward_infra_prepare_backward_infra)r rEr1r2r3r4r5r?r!r"r@s   zPipelineScheduleSingle.__init__r`rbcOsh|j|||\}}|dk r6tt||j}nd}||||||jjr`| |jj SdSdSrc) rclear_runtime_statesrqrSru tensor_splitr6r_rBrtrr)r rarXrdrerorp targets_splitr!r!r"rf-s zPipelineScheduleSingle.step)NNNN)r$r%r&__doc__rr.r rr rrrmr rr@r rf __classcell__r!r!rr"r s c@s6eZdZdZdeeeeeeeedddZdS)rz^ The GPipe schedule. Will go through all the microbatches in a fill-drain manner. NrZc Cs|||||\}}g}t|jD]}td|r|j|}t|dd}|D] } | qV|j |||||} |j |}t|dd}| |W5QRXt d|jjd|||j| ||q"|D] } | q|jsdSg} t|jD]}td|z|j|}t|d d}|D]} | q8||j|} |jj|| d |j|}t|d d}| |W5QRXt d|jjd |q||j|| D]} | qdS) z Run one iteration of the pipeline schedule with list of microbatches. Will go through all the microbatches according to the GPipe schedule. Args: microbatches: list of microbatch args. zForward fwd_recvr}fwd_send[z] Forwarded microbatch Nz Backward bwd_recvrIbwd_sendz] Backwarded microbatch )rnranger6rrget_fwd_recv_opsrvalueswaitforward_one_chunkget_fwd_send_opsrVr=ryr+rJr;get_bwd_recv_opsrMbackward_one_chunkget_bwd_send_opsrY) r r[r\rGrXZfwd_sends_to_waitirZworksZworkrFZbwd_sends_to_waitrIr!r!r"r_UsL             z ScheduleGPipe._step_microbatches)NNNNr$r%r&rr r r_r!r!r!r"rOsc@s6eZdZdZdeeeeeeeedddZdS)rzo The 1F1B schedule. Will perform one forward and one backward on the microbatches in steady state. NrZcCsP|||||\}}t|j|j|jj}d}d}d}g} t|D]} |j|} t| dd} rj| |j |||||} |r| |j |} ||dkrt| dd}| |j| |||d7}qB|j |}t| |dd}r| ||j|}|jj||d|j|}|d7}||jkr8q|j|} t|| d d}rb| |j |||||} | |j| |||j |} |d7}qt|d d}||jkr0|j |}t|d d} r| ||j|}|jj||d|r | |j|}t|d d}|d7}q|r>| ||j|dS) z Run one iteration of the pipeline schedule with list of microbatches. Will go through all the microbatches according to the 1F1B schedule. Args: microbatches: list of microbatch args. rNrr}rrZfwd_send_bwd_recvrZbwd_send_fwd_recvrr)rnminr6rrr+rrr|rrrrJrrMrrrY)r r[r\rGrXZ warmup_chunks fwd_mb_index bwd_mb_indexZ send_workZ fwd_sends_Z fwd_recvsZ recv_workrFZ bwd_recvsZ fuse_workrIZ bwd_sendsr!r!r"r_sh                   zSchedule1F1B._step_microbatches)NNNNrr!r!r!r"rsc seZdZdZdeeeeeee e dfee e e fee e e efe efdfdd Zdddeedd d Zdeeeeeeeed d d ZZS)rzQ Base class for multi-stage schedules. Implements the `step` method. N.rWr1r2r3r4r5cst|dkrtdt|tj|||||d|_|dj_|dj_|dj _ jD] }j |_ qdfdd_ i_jD]}||j r||qdS)Nrz9Multi-stage schedule expects at least two stages but got r0rcs|jojdk SrA)rBr7)rErr!r"Oz0PipelineScheduleMulti.__init__..)rKrkrr@_stagesrr group_size pp_group_size group_rankrankr;rZ_should_compute_losspipeline_orderrr)r rWr1r2r3r4r5rErrr"r@0s0          zPipelineScheduleMulti.__init__r`rbc Osz|jD] }|q|||\}}|dk r@tt||j}nd}||||||jD]}|jrZ| |j SqZdSrc) rrrqrSrurr6r_rBrtrr) r rarXrdrerErorprr!r!r"rf\s   zPipelineScheduleMulti.steprZcCsH|||||\}}dd|jD}|jd|j}|jd|j}t|j|jD]\}} |j|} |j|} g} | dk r"| \} }}| tjkr||}||||||}| ||||| | |nN| tj kr||}| ||}|j||d| ||ntd| d}|t| kr<| |}|dk r|\} }}| tjkr||jdkr||d}| ||n| tj krntd| d}|t| kr| |}|dk r$|\} }}| tjkrnB| tj kr|dkr$||d}| ||ntd| | rTt| qT||j|dS)z Operate on the microbatches for looped schedules (multiple stages on each rank). TODO: Does not use sorted_batch_isend_irecv(). As a result, this schedule does not support models with skip connections. cSsi|] }|j|qSr!)r+rNr!r!r" sz.rNrzUnknown computation type r)rnrrr enumeraterrrrrJrVrr'rMrrrkrKrrrr|rrY)r r[r\rGrXZstage_index_to_stageZ prev_rankZ next_rankZ time_stepactionZ prev_rank_opsZ next_rank_opsrr)rHr+rErFrIZprev_rank_actionZnext_rank_actionr!r!r"r_sl                   z(PipelineScheduleMulti._step_microbatches)NNNN)NNNN)r$r%r&rr rr.r rr rrrmr rr@rfr_rr!r!rr"r*s0 ,%c sVeZdZdZdeeeeeee e e e fe e fdfdd ZddZZS) rai Breadth-First Pipeline Parallelism. See https://arxiv.org/abs/2211.05953 for details. Simliar to Interleaved 1F1B, Looped BFS supports multiple stages per rank. What is different is that when microbatches are ready for multiple local stages, Loops BFS will prioritizes the earlier stage, running all available microbatches at once. NrWr1r2r5csBtj||||di|_t|jD]}||}||j|<q$dS)Nr)rr@rrr!_calculate_single_rank_operations)r rWr1r2r5rrank_opsrr!r"r@s  zScheduleLoopedBFS.__init__c Cst|j}t||j||j}g}t|D]}|dq*|D](}t|jD]}|ttj||qLq>d|jd|}| dg|t |D],}t t|jD]}|ttj ||qq|S)Nrr) rKrrrrDr6r(rrrVreversedr') r rn_local_stagesZ stage_indicesrrr+rHpost_warmup_opsr!r!r"rs,      z3ScheduleLoopedBFS._calculate_single_rank_operations)NN)r$r%r&rr rr.r rr rrmrr r@rrr!r!rr"rs c seZdZdZd eeeeeee e dfee e e fee e e efe efdfdd Zeeeddd ZZS) rak The Interleaved 1F1B schedule. See https://arxiv.org/pdf/2104.04473 for details. Will perform one forward and one backward on the microbatches in steady state and supports multiple stages per rank. When microbatches are ready for multiple local stages, Interleaved 1F1B prioritizes the earlier microbatch (also called "depth first"). N.rc s|dj|_||jdkr2td|d|jdtj||||||dt||_|dj|_|dj |_ i|_ t |jD]}| |}||j |<q|dS)Nrz?Interleaved 1F1B schedule requires the number of microbatches (zD) to be a multiple of the number of pipeline ranks (z).r) rrrkrr@rKrrrgrouprrr) r rWr1r2r3r4r5rrrr!r"r@&s,     z ScheduleInterleaved1F1B.__init__)rscsfdd}|jj}|}||}||}td|||fdd}fdd}tt} tt} g} tD]} | dqjjdjd } t|D]}|kr(||}| |}d | |<| t t j |||d kr| dg| qȈ|krD|krnnb||}| |}d | |<| t t j ||||}| |}d | |<| t t j ||q| d||}| |}d | |<| t t j ||qtjd D]} | dq| S) Ncs8jdj}|djd|}t|jjS)Nrr)rrrr6)rZwarmups_ops_last_stage warmup_opsrr!r"get_rank_warmup_opsNszVScheduleInterleaved1F1B._calculate_single_rank_operations..get_rank_warmup_opsz=rank %s, warmup_ops %s, 1f1b %s, cooldown_ops %s total_ops %scs|jj}|jSrA)rrrfZ local_index)rr r!r"forward_stage_indexjszVScheduleInterleaved1F1B._calculate_single_rank_operations..forward_stage_indexcs,jd|jj}|jS)Nr)rrrrr rr!r"backward_stage_indexoszWScheduleInterleaved1F1B._calculate_single_rank_operations..backward_stage_indexrr)rr6r=ryrr.rrDrr(rrrVr')r rrZmicrobatch_opsZ fwd_bwd_opsZ cooldown_opsZ total_opsrrZfwd_stage_mb_indexZbwd_stage_mb_indexrrrrZfwd_stage_indexrHrZbwd_stage_indexrr!rr"rMs~                  z9ScheduleInterleaved1F1B._calculate_single_rank_operations)NNNN)r$r%r&rr rr.r rr rrrmr rr@r(rrr!r!rr"rs ')N)N)0loggingabcrr collectionsrenumrtypingrrrr r r r r ruZtorch.distributed distributedrzZtorch.profilerrZ microbatchrrrrEr__all__ getLoggerr$r=rr(r/ZP2POprmr|r.ZWorkrrrrrrrr!r!r!r"sH  (      2  FT3@