i) VddlZddlZddlZddlZddlZddlmZmZddlm Z m Z ddl m Z ddl mZddlmZddlmZmZmZmZddlZddlmZddlmZdd lmZmZdd lm Z dd l!m"Z"d d l#m$Z$m%Z%d dl&m'Z'm(Z(m)Z)d dl*m+Z+gdZ,ejZe.Z/GddeZ0e0jbZ1e0jdZ2e0jfZ3e0jhZ4e0jjZ5e0jlZ6e0jnZ7e0jpZ8e0jrZ9e0jtZ:e0jvZ;e0jxZe3Z?e:Z@ejdZBGddeZCedeCdeDfdZE dTdeFeGeHeCdzfdeGdzdeDfdZIGddeZJdTdeHejdeDdzdeHejfd ZM dTdeHejdeDdzdeFeGeHejffd!ZNd"eHejfd#ZOGd$d%eJZPGd&d'ePZQGd(d)ePZRGd*d+ePZSd,e0deTfd-ZUd.eHeCdzd/eGdeHeCdzfd0ZV dUd1eHeCdzd2eGdeHeCfd3ZWd1eHeCdzdeHeCfd4ZXd1eFeGeHeCfd5e eGgeGfd6eGdeFeGeHeCffd7ZYd.eFeGeHeCdzfd8eGd6eGd9eGdeFeGeGff d:ZZGd;dZ\Gd?d@eZ]GdAdBe[Z^GdCdDe^Z_ dVdEZ`GdFdGe^ZaGdHdIe^ZbGdJdKe^ZcGdLdMe^ZddNeDfdOZed5e eGgeGfd6eGfdPZfdQZgdReHe+dNeDfdSZhy)WN)ABCabstractmethod)Counter defaultdict)Callable)Enum) lru_cache)Anycast NamedTupleProtocol)OptimizedModule) FSDPModule UnshardHandle)_Loss)record_function)generate_rank_to_stage_mappinggenerate_stage_to_rank_mapping) merge_chunkssplit_args_kwargs_into_chunksTensorChunkSpec)_PipelineStageBase) get_schedule_classPipelineScheduleSinglePipelineScheduleMulti Schedule1F1B ScheduleGPipeScheduleInterleaved1F1BScheduleLoopedBFSScheduleInterleavedZeroBubbleScheduleZBVZeroBubbleScheduleDualPipeVcReZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZedZy)_ComputationTyper ctjdtjdtjdtjdtj dtj dtjdtjdtjd tjd tjd tjd i }||S) NFIWUNSHARDRESHARDSEND_FRECV_FSEND_BRECV_BB OVERLAP_F_B REDUCE_GRAD) r%FORWARDBACKWARD_INPUTBACKWARD_WEIGHTr5r6r7r8r9r: FULL_BACKWARDr<r=)selfstr_maps `/var/www/html/engine/venv/lib/python3.12/site-packages/torch/distributed/pipelining/schedules.py__str__z_ComputationType.__str__;s  $ $c  + +S  , ,c  $ $i  $ $i  # #X  # #X  # #X  # #X  * *C  ( (-  ( (-  t}c|dk(rtjS|dk(rtjS|dk(rtjS|dk(rtjS|dk(rtj S|dk(rtj S|dk(rtjS|dk(rtjS|d k(rtjS|d k(rtjS|d k(rtjS|d k(rtjStd |)Nr2r3r4r5r6r7r8r9r:r;r<r=Invalid computation type )r%r>r?r@r5r6r7r8r9r:rAr<r= RuntimeErroractions rDfrom_strz_ComputationType.from_strLs S=#++ + s]#22 2 s]#33 3 y #++ + y #++ + x #** * x #** * x #** * x #** * s]#11 1 } $#// / } $#// /!:6(CD DrFN)__name__ __module__ __qualname__r>r?r@r5r6r7r8r9r:rAr<r=rE staticmethodrLrFrDr%r%,sYGNOGG F F F FMKK"EErFr%zK(\d+)(F|I|B|W|UNSHARD|RESHARD|REDUCE_GRAD|SEND_F|RECV_F|SEND_B|RECV_B)(\d*)ceZdZUeed<eed<dZedzed<dZeddzed<dZ dZ e d e fd Z ed efd Zy) _Action stage_indexcomputation_typeNmicrobatch_index)rS. sub_actionsc"|jSN)__repr__rBs rDrEz_Action.__str__s}}rFcP|jC|jDcgc] }t|}}ddj|d|jSt |j }|t |jz }|j |t |j z }|Scc}w)N(;))rWreprjoinrUstrrTrV)rB sub_actionsub_action_reprsrepr_strs rDrZz_Action.__repr__s    'CGCSCSTZZ 0T Tsxx 012!D4I4I3JK K4++,H D112 2H$$0C 5 566O UsB#returncP|jttttt fvSrY)rUr>rAr?r@r<r[s rD is_compute_opz_Action.is_compute_ops)$$      )   rF action_stringc|j}|dk(ry|jdrd|vr|jd}|d|}||dzd}g}|jrM|jdD]9}tj |j}|)|j |;t dtj |d|rt|SdStj|x}rW|j\}} } t t|tj | t| rt| SdS|dk(rytd |d ) z Reverse of __repr__ String should be formatted as [stage][action type][(microbatch)] e.g. `2F0`, `1UNSHARD`, `3SEND_F1` Nr]r_rr^)rTrUrVrWzInvalid action string: zD, should be formatted as [stage][action type][(microbatch)] e.g. 2F0)strip startswithfindsplitrSrLappendr%tuple _action_regexmatchgroupsintlenrI) ri bracket_endsub_partcomputation_type_partrWsub_strrcrtrTrUrVs rDrLz_Action.from_strs&++- B   # #C (SM-A',,S1K$+H%2a!% ! K~~'~~c27G!(!1!1'--/!BJ!-#**:67!1!:!:;P!Q!%2=E+.  DH  "'' 6 65 6>Clln ;K)+;K  ))*:;),-=)>$% EI  b %m_4x y  rF)rMrNrOrv__annotations__r%rVrWrrrErZpropertyboolrhrPrbrLrQrFrDrSrSsq&&#'cDj'04K~&-4  t  3 3 3 rFrSrKrfcdt|S)NzPP:rbrJs rD_get_profiler_function_namers V rFpipeline_ordererror_step_numberc ^tj|}|D]/}tt||D]}||| d|||<1t d|j D}t|Dcgc]4}dt |jtt |dz z6}}t|Dcgc]}|j|dg|z}}ttj|ddi}t|} t| Dcgc]}dt |z} }t|g|D cgc]} t d| Dc} dt|d d zzdjfd t| Dz} t||D cgc]T\} }| d djfd t|Dz|!t!| j#d|k(rdndzV}} }| dzdj|zdz}|Scc}wcc}wcc}wcc} wcc}} w)z Formats the pipeline order in a timestep (row) x rank (column) grid of actions and returns the formatted string. If `error_step_number` is passed in, an additional label will be added to signify which step that it is erroring on. rkc32K|]}t|ywrY)rw).0actionss rD z)_format_pipeline_order..sHWCLHszStep r fillvalueRank c3LK|]}|tt|ndyw)Nr)rwrb)ritems rDrz)_format_pipeline_order..s" F$d.CD NA 5 Fs"$ rr&c3:K|]\}}|d|dyw. s0<)1E5;q>" " #$<sz: c3LK|]\}}t|d|dywrr)rrrrs rDrz)_format_pipeline_order..s/R4c$i+a.!1123Rs!$z <-- ERROR HERE )copydeepcopyrangerwmaxvaluesrbzfillsortedgetlist itertools zip_longestzipra enumeratervrp)rrrankr num_steps step_labelskey rank_actionstransposed_actions num_ranks rank_labelscol header_rowrrowformatted_rowsformatted_tablers @rD_format_pipeline_orderrsc]]>2N-s>$/01 -Ad#A&.*,t$Q' --H0E0E0GHHIAFyAQ<=#a&,,s3y1}#5677K >DN=S693y 01L i33\PRPQN#I-29-=>7SV#>K>{8%78  F# FFK KN+a/0388<5>{5K<4Jk+=>  E3 ' ((R9S>R R S!,EKKM!$%)::    N !4'$))N*CCdJO I? s89HHH:H$AH)c eZdZ d dededej fdzdeedfdzde e efdzde e e fee zdzd e f d Z d Zd Zd Ze d dedzdedzdedzdedzde f dZedddddedzfdZddddedzfdZ d!dedzdedzdedzdedzdeeeff dZdZ d"dee dfde e e fdzfdZdee de fdZy)#_PipelineScheduleNTn_microbatchesloss_fn.args_chunk_speckwargs_chunk_specoutput_merge_spec scale_gradsc||_||_||_||_||_||_ |jdu|_g|_tjd|jjy)NzUsing %s) _n_microbatches_loss_fnr_args_chunk_spec_kwargs_chunk_spec_output_merge_spec _has_backward_internal_lossesloggerinfo __class__rM)rBrrrrrrs rD__init__z_PipelineSchedule.__init__sq . '!0"3"3 "]]$657 J 7 78rFc|jr>|j1|j|||}|jj |yyyrY)is_lastr _compute_lossrrq)rBstageoutput target_mbsmb_indexlosss rD_maybe_compute_lossz%_PipelineSchedule._maybe_compute_loss@sD ==T]]6%%fj.BCD  ! ! ( ( .7=rFcd|cxkxrt|jknc}|jr|j|r|j|St|jdk7r|st d|d|jy)NrzLoss for microbatch z6 is not available. Available losses for microbatches: )rwrrrrI)rBrr valid_indexs rD_maybe_get_lossz!_PipelineSchedule._maybe_get_lossEs8@c$*?*?&@@ ==T]]6;((2 2 && '1 ,[&xj166:6K6K5LN  rFct|ts|g}td|D}|r}|{t|j|j k7r.t d|j dt|j|j|j|j|jjy)zB Update the losses to those in the internal state c34K|]}|jywrYr)rrs rDrz3_PipelineSchedule._update_losses..Xs!DE%--!DsN Expecting z losses but got ) isinstanceranyrwrrrIclearextend)rBstageslossescontains_last_stages rD_update_lossesz _PipelineSchedule._update_lossesQs &$'XF!!DV!DD 6#54(()T-A-AA" !5!5 66Fs4K`K`GaFbc LLN MM$// 0 ##%rFarg_mbs kwarg_mbsrrreturn_outputsct)aG 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. return_outputs: whether to return the outputs from the last stage. NotImplementedError)rBrrrrrs rD_step_microbatchesz$_PipelineSchedule._step_microbatcheshs $"!rFtargetrrct)5 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. return_outputs: whether to return the outputs from the last stage. r)rBrrrargskwargss rDstepz_PipelineSchedule.step|s ("!rFrrcz|j} d|_|j|||d|||_S#||_wxYw)a Run one iteration of the pipeline schedule with *whole-batch* input. Will chunk the input into microbatches automatically, and go through the microbatches, calling forward only. args: positional arguments to the model (as in non-pipeline case). kwargs: keyword arguments to the model (as in non-pipeline case). target: target values for the loss function. losses: a list to store the losses for each microbatch. Fr)rr)rBrrrroriginal_has_backwards rDevalz_PipelineSchedule.evalsH!% 2 2 7!&D 499d6&KFK"7D !6D s1 :rfcdtffd }| ||dndgjz}| ||dnigjz}| ||d|'t|tst dt |||fS)z* Pre-process/check inputs namec t|tst|dt|t |j k7r't dj d|dt |y)Nz must be a list but got a rrz but got )rr TypeErrortyperwr ValueError)mbsrrBs rDcheck_type_and_lenz;_PipelineSchedule._check_inputs..check_type_and_lensmc4(4&(B49+ NOO3x4///  !5!5 6avYs3xjQ0rFrrQrrz losses must be a list but got a )rbrrrrr)rBrrrrrs` rD _check_inputsz_PipelineSchedule._check_inputss #    w 2dT111G  y+ 6t333I  ! z< 8  fd+"B4<. QRR !!rFc&|j||SrY)r)rBrrs rDrz_PipelineSchedule._compute_losss}}VV,,rFrrc|s|r4t|||j|j|j\}}||fSdg|jzig|jzfS)zj Splits a full-batch input into chunks (i.e. microbatches) and returns the chunks rQ)rrrr)rBrr args_split kwargs_splits rD _split_inputsz_PipelineSchedule._split_inputssm 6'D$$%%'' ( $J |+ +4$...t7K7K0KK KrF output_chunksc.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)rBrs rD_merge_outputsz _PipelineSchedule._merge_outputss    # #  rFNNNNTNNNNrY)rMrNrOrvrtorchTensorrrrdictrbr r~rrrrrrrrrrrrrrQrFrDrrs7;>B?C@D  9 9#u||+,t3 934t; 9  _ 45< 9  S>E#J6= 9 9D/ &. $!%"&"# ""$;"4K " t "  ""&" "t ""*"&T7td{7, $!%"&" $"$"$;$"4K $" t $" tTz  $"L- )-LCHoLS#X%L. DI # rFrp2p_opsdescct|dk(rgS|r|dnd}tjd||tj|S)zt Simple wrapper over batch_isend_irecv from torch.distributed, which just adds a descriptive logger on top. rz, rkzbatch_p2p %s%s)rwrdebugdistbatch_isend_irecv)rrdesc_strs rD _batch_p2pr sF 7|q "$r{H LL!8W5  ! !' **rFctt}i}t|dk(r|S|D] }||jj |"t |j D]\}}t||||<|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. rr)rrrwpeerrqritemsr )rr ops_by_peer work_by_peeropropss rD_sorted_batch_p2prs0;4/@K/1L 7|q(BGG##B'(K--/08 c'$7 T8 rFworkc2|D]}|jy)zX Waits for a list of dist.Work (typically from _batch_p2p / _sorted_batch_p2p). N)wait)rws rD_wait_batch_p2prs rFceZdZdZ ddedededzdeedfdzd e e efdzd e e e fee zdzd e ffd Z d Zdddddedzde fdZde eeedzfdzfdZxZS)ra Base class for single-stage schedules. Implements the `step` method. Derived classes should implement `_step_microbatches`. Gradients are scaled by num_microbatches depending on the `scale_grads` argument, defaulting to True. This setting should match the configuration of your loss_fn, which may either average losses (scale_grads=True) or sum losses (scale_grads=False). NTrrrr.rrrct|||||||||_|j|_d|_d|_||jkrtd|d|jd|j|_ y)NrrrrrrFzNumber of microbatches (z9) must be greater than or equal to the number of stages (z).) superr_stage num_stages _num_stages_stage_forward_initialized_stage_backward_initializedr_get_pipeline_orderr) rBrrrrrrrrs rDrzPipelineScheduleSingle.__init__*s )+//#    ++*/'+0( D,, ,*>*:;##'#3#3"4B8   $ $ & rFc|jsmg}|j|jjt t ||jj |j||d|_|jr:|js-|jj|jd|_ yyyNT) r"rr_get_init_p2p_neighbors_opsrr _prepare_forward_infrarrr#_prepare_backward_infra)rBrrall_opss rD_initialize_stagez(PipelineScheduleSingle._initialize_stageMs..)+G NN4;;BBD E Jw/ 0 KK . .t/C/CT6 R.2D +   d&F&F KK / /0D0D E/3D ,'G rFrrrc|jrtjs td|j|j_|jj |j||\}}|*ttj||j}nd}|j||||||jjr'|r%|j|jjSyrzstep() requires gradients to be enabled for backward computation; it should not be used under torch.no_grad() context. Please call eval() instead.N)rris_grad_enabledrIr has_backwardclear_runtime_statesrr tensor_splitrrrrr) rBrrrrrrr targets_splits rDrzPipelineScheduleSingle.step]s&   e&;&;&=. $(#5#5   ((*$(#5#5dF#C L   !3!3FD&&t{{'@'@A ArFrfcy)a Returns the pipeline execution order as a schedule IR. The returned IR is a dictionary mapping rank IDs to lists of actions. Each action is either an _Action object representing computation to perform, or None representing a deliberate idle step. The None values are used to represent pipeline bubbles where a rank must wait for dependencies from other ranks before proceeding. However during execution, with the _PipelineScheduleRuntime, these Nones are skipped since the relevant communication (send/recv) will be scheduled and waited on. Returns: A dictionary mapping rank -> list of actions NrQr[s rDr$z*PipelineScheduleSingle._get_pipeline_orders rFr)rMrNrO__doc__rrvrrrrrrbr r~rr+rrrSr$ __classcell__rs@rDrrs$(>B?C@D ! !! ! D ! 34t; !  _ 45< !  S>E#J6=! ! F4&"# 2t 2  2hT#tGdN/C*C%Dt%KrFrc PeZdZdZ d dedzdedzdedzdedzdef dZy) _ScheduleForwardOnlyzo The forward-only schedule. Will go through all the microbatches and perform only the forward pass Nrrrrrc|| td|j||||\}}|j|d|dg}t|jD]}t d|5|j j|}t|d} | jD] } t| |j j||||||j j|}t|d} |j| jdddtjd|j j ||D] } t| y#1swYIxYw) z< Run one iteration of the pipeline schedule Nz7Forward-only schedule does not support loss computationrForward fwd_recvrfwd_send[%s] Forwarded microbatch %s)rIrr+rrrrget_fwd_recv_opsrrrforward_one_chunkget_fwd_send_opsrrrrT) rBrrrrrfwd_sends_to_waitrrworksrs rDrz'_ScheduleForwardOnly._step_microbatchessf  !V%7I "//JPVW wqz9Q<846t++, UA 8A30 9kk2215)#J?!LLN*D#D)* --aYq\Jkk2215)#J?!((8 9 LL79P9PRS T U$& "D D ! "# 9 9s 'B3E""E+ r)rMrNrOr4rr~rrQrFrDr8r8s] $!%"&"# *"*"$;*"4K *" t *"  *"rFr8c xeZdZdZ d dedzdedzdedzdedzdef dZd eeee dzfdzfd Z y) rz^ The GPipe schedule. Will go through all the microbatches in a fill-drain manner. Nrrrrrc Z|j||||\}}|j|d|dg}t|jD]}t d|5|j j |}t|d} | jD] } t| |j j||||||} |j j|}t|d} |j| jdddtjd|j j||j!|j  |||D] } t| g} t|jD]}t d |5|j j#|}t|d } | jD] } t| |j%|j |} |j j'|| ||jd z k( |j j)|}t|d } | j| jdddtjd|j j|!| D] } t| |j+|j ||j j-|j.r|jyd y#1swYxYw#1swYxYw)a6 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. return_outputs: whether to return the outputs from the last stage. rr:r;rsave_forward_outputr<Nr=z Backward bwd_recvrr last_backwardbwd_sendz[%s] Backwarded microbatch %s)rr+rrrrr>rrrr?r@rrrrTrget_bwd_recv_opsrbackward_one_chunkget_bwd_send_opsrperform_reduce_gradr)rBrrrrrrArrrBrrbwd_sends_to_waitrs rDrz ScheduleGPipe._step_microbatchess "//JPVW wqz9Q<846t++, IA 8A30 9kk2215)#J?!LLN*D#D)*66wqz9Q<^7kk2215)#J?!((8 9 LL79P9PRS T  $ $T[[&*a H# I,& "D D ! " 46t++, VA 9QC1 9kk2215)#J?!LLN*D#D)*++DKK; .."#t';';a'?"?/ kk2215)#J?!((8 9" LL8$++:Q:QST U% V*& "D D ! " DKK0 ''@P@P(<(<XVWXm 9 98 9 9sB5LCL!L !L* rfc i}|j}t|D]}g}|}|jdg|zt|jD],}|j t |t j|.d|dz |z z}|jdg|zt|jD],}|j t |t j|.t||j||<|S)z Returns the pipeline order for GPipe schedule. See base method in PipelineScheduleSingle for details on the schedule IR format. Nr'r) r!rrrrqrSr%r>rA_add_reduce_grad)rBr pp_group_sizerr warmup_delaymb_idxbackward_delays rDr$z!ScheduleGPipe._get_pipeline_order-s (( -( SD,.G L NND6L0 1  4 45 Pwt-=-E-EvNO P-!"3d":;N NND6N2 3  4 45 Vwt-=-K-KVTU V$4GT=Q=Q#RN4 ' S*rFr rMrNrOr4rr~rrrvrSr$rQrFrDrrs $!%"&"# NYNY$;NY4K NY t NY  NY`T#tGdN/C*C%Dt%KrFrc xeZdZdZ d dedzdedzdedzdedzdef dZd eeee dzfdzfd Z y) rzo The 1F1B schedule. Will perform one forward and one backward on the microbatches in steady state. Nrrrrrct|j||||\}}|j|d|dt|j|j|j j z }d}d}g} g} t|D]} |j j|} tt| d|j j||||||} t| |j j|} ||dz k7r t| d} |j|j | |||dz } |j j|}tt| |zd|j|j |}|j j!||||jdz k(|j j#|}|dz }||jk(rn|j j|} tt|| zd |j j||||||} |j|j | |||j j|} |dz }Gt|d } ||jkr|j j|}tt|d |j|j |}|j j!||||jdz k(t| |j j#|}t|d } |dz }||jkrt| |j%|j ||j j'|j(r|jy dy ) a5 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. return_outputs: whether to return the outputs from the last stage. rr;rrErr<fwd_send_bwd_recvrHbwd_send_fwd_recvrJrGN)rr+minrr!rrTrr>rr r?r@rrKrrLrMrrNr)rBrrrrr warmup_chunks fwd_mb_index bwd_mb_index send_work fwd_sends_ fwd_recvsr bwd_recvsr bwd_sendss rDrzSchedule1F1B._step_microbatchesTs "//JPVW wqz9Q<8    t{{66 6   &(  }% A 44\BI JyzB C[[22 %,'$2 3F I & 44\BI}q00&yzB  $ $T[[&*l S A L; D 44\BI Jy9'I A L'T111,  " DKK0 ''@P@P(<(<XVWXrFrfc i}|j}t|D]}g}|jdg|z|dz |z }d}t|D].}|jt |t j ||}0tdd|dz |z z}|jdg|zd} |j|z } | dkDri|dz }|jt |t j || dz} |jt |t j| | dz } | dkDri|j| z } | dkDr||z dkDrK|jd| dkDri|jt |t j| | dz } | dz} n4|jt |t j| | dz } | dz} | dkDrt||j||<|S)z Returns the pipeline order for 1F1B schedule. See base method in PipelineScheduleSingle for details on the schedule IR format. Nrrr&) r!rrrqrSr%r>rrrArQ) rBrrRrr num_forward forward_mbr wait_for_1f1b backward_mbremaining_forwardremaining_backwards rDr$z Schedule1F1B._get_pipeline_orders (( -(8 SD,.G NND6D= ))1,4KJ;' wt-=-E-EqIJ    1 (9D(@#ABM NND6M1 2K $ 4 4{ B #a'a wt-=-E-EzRS!Q&!D"2"@"@+Nq $a'"&!5!5 !C $q("D(A-NN4()A-#D*:*H*H+V$q( *a/*NN&6&D&DkR 1$K&!+&%%q(($4GT=Q=Q#RN4 q8 SrrFrrVrQrFrDrrNs $!%"&"# NYNY$;NY4K NY t NY  NY`BT#tGdN/C*C%Dt%KBrFr action_typec|ttfvSrY)r4r;)rls rD_requires_reduce_gradrn)s 1a&  rFrrc Fg}ttfd}|D]}||j|g}|jtk(r'|j |j D] }||| n ||||D]"}|jt |td$|S)z REDUCE_GRAD refers to joint across minibatches grad reduction. reduce_grad frees memory and we want to schedule it just after the last "backward"-like stage. ct|jr4|j}|xxdz cc<|k(r|j|yyyNr)rnrUrTrq)a to_schedulerTcntrs rD _leaf_actionz&_add_reduce_grad.._leaf_action7sO !3!3 4--K   ! ;>1"";/2 5rFN)rrvrqrUr<rWrSr=) rractions_with_reduce_gradrurrschedule_reduce_grad_stage_idxsrc stage_idxrts ` @rDrQrQ-s68%c*C0 S 9  ''*57'   ,1Jmm J Z)HI J ; <8 SI $ + +GI{D,Q R S S $#rFcompute_actionsmax_active_stagesc dtdttdzdttfd}t g dtf fd }dtf fd }t |D]s\}}| ||||d tt fd  }tt fd  }|D] } ||  |D] } ||   j |ut D] } ||   S) a]Given a basic schedule involving only compute actions (F,B,W,OVERLAP_F_B), add UNSHARD/RESHARD actions for FSDP. UNSHARD refers to fetching the full contents of an FSDP-sharded layer, requiring an all-gather operation. RESHARD does the opposite, releasing memory (but doing no communication) We abandon the "timestep lock" during lowering max_active_stages controls how many prefetches we allow. It should be measured in mb and tuneable but in practice 3 stages is probably the thing we want? (to account for having one f and one b active, and something else prefetching?) count next_actionsNrfct}g}|D]}||jtk(rt|jh|jD]G}|j|vs|j |j|j |jIt||k\s|S|j|vs|j |j|j |jt||k(s|S|S)zdRemove duplicates (same stage, different microbatch), find next 'count' stages that will do compute.)setrUr<rWrTaddrqrw)r|r}seenretrrrcs rDnext_stage_indicesz0_add_unshard_reshard..next_stage_indices^s "A}%%49R&'mm? %11= HHZ%;%;<JJz'='=>?3x5( }}D0/ 1==1s8u,! # "" rFrTchj|jt|tdyrY)rrqrSr5rT active_stagesfsdp_aware_actionss rD_unshardz&_add_unshard_reshard.._unshardys(+&!!'+w"EFrFchj|jt|tdyrY)removerqrSr6rs rD_reshardz&_add_unshard_reshard.._reshard}s([)!!'+w"EFrFc |vSrYrQ)srs rDz&_add_unshard_reshard..s a}&<rFc |vSrYrQ)rnext_ns rDrz&_add_unshard_reshard..s avorF)rvrrSrrfilterrq) ryrzrrrrrKfetchevictrrrrs @@@rD_add_unshard_reshardrNs #T'D.5IdSVi0"eM(*GcGGcG/* 6 > $$5qr7JKV /"q'9'A    "/"q'9'A-0,@1,Doa($   # #~ 5',,?""k&=&=='';+G+GG  ! !**M6;R;RS     "  ! !& )1 2 rF stage_to_rankr c  |Dcic]}|g}}|Dcic] }|tc}dtdtffd dtdtttff fd }dtdzdttdtffd |rgd}t |D]It |d kDsJd d t ||d }||j |j }n|f}t fd |Dsk||j||D]} j|  | s || \} } |j| j| || jj| | jj| |jd t |d k(r|=d }L|sJd|rg|Scc}wcc}w)z Transforms a compute-only schedule into a complete schedule with communication actions. For actions with sub-actions (OVERLAP_F_B) we ensure that all the subactions have been computed and the communication is ready rKrfcF|jtk(r<|jdz k7xr(|jdz|jk7S|jttfvr9|jdk7xr(|jdz |jk7Sy)NrrF)rUr2rTr?rA)rKr rs rD _has_commsz"_add_send_recv.._has_commss  " "a '%%a73M""Q&=v112=3 3 $ $(G G%%*3}""Q&0v11203 3rFc|s J|d|j}|j}|j}t||tk(rt nt |}|tk(r|dzn|dz }t||tk(rtnt|}||fS)Nz is not a valid comm actionr) rTrUrVrSr2r7r9r8r:)rKrxctyperTsendrecv_stage_idxrecvrs rD _get_commsz"_add_send_recv.._get_commss&!IfX-H#II!&& ''((yEQJ&FFK*/1*Q)a-~!vPTzrFN prev_actionscJ|y|jtk(rc|jdk7rTt|jt|j |vryt|jdz t|j |vryy|jt tfvr|jdz k7rt|jt|j |vryt|jdzt |j |vryt|jdzt|j |vryyy)aWe don't put our own recv ops in the schedule, we let a sender on another rank put our recv ops in place. This helps ensure a sane (non-hanging) ordering of sends and recvs. But it also means we might not be able to schedule our next compute action yet. TrrF) rUr2rTrSr8rVr?rAr:)rKrr s rD_ready_to_schedulez*_add_send_recv.._ready_to_schedules  >  $ $ )f.@.@A.E**FF4K4KL **Q.63J3JK   # # 'F F""j1n4**FF4K4KL **Q.@W@WX **Q. v?V?VW rFFrrank=z, len(compute_actions[rank])=c36K|]}|ywrYrQ)rrrrrrs rDrz!_add_send_recv.."sVQ)!\$-?@VsTz6Malformed compute schedule, can't schedule sends/recvs) rrSr~rrrrwrWallrqrrTr)ryrr r comm_actionsrprogressrK all_actionsrrrrrrrs ``` @@@rD_add_send_recvrs;DS-S4dBh-SL-SET,UTT35[,UL 7 t 7uWg-='>(7T>(W(RV(T ?+ Dt,-1 4'7C 5689 1%T*1-F!f&8&8&D$00 %i V+VV!T"))&1$ PA &**1-!!}%/] d%T*11$7$T*..t4$]43C3C%DELLTR$]43C3C%DEII$O P D ! % %a (?4()Q.#D)H? @QQQxG H M.T,Us G0G5rRnum_microbatchesc t|k(sJd|dtt|D]}|vrJd|t|Dcic]=}|ttttt tt ti?c}idtdtdtffd }D]n}t|D][\}}| t|tsJd|d |d |d |j|jD] } || ||R||||]pD]} t| t} t| t} t| t } t| t }| |k(sJd | d td| d|| |k(sJd| d| d|| | |zdzz|k(rJd| d|d| d| d| Scc}w)Nz2Schedule has incorrect number of ranks - expected z , actual z%Schedule is missing actions for rank rKrrc 2|j}|j}|j}|tk(r |tj |n"|t k(rX| |tvr.d|d|d|d|d }t |}|d|}t| |t j |n|tk(rX| |tvr.d|d|d|d|d }t |}|d|}t| |tj |n`|tk(rW| |tvr.d|d|d |d|d }t |}|d|}t| |tj || vr| |<y |} || k(sJd|d|d |d |d | y)zPProcess a single action and update stage_actions and stage_index_to_rank_mappingr, step z": Running Full Backward for stage z , microbatch z without first running Forwardrz Full pipeline schedule: z#: Running Backward Input for stage z$: Running Backward Weight for stage z% without first running Backward Inputz: Stage z is assigned to both rank z and rank N) rTrUrVr2rr;rAssertionErrorr3r4) rKrrs_idrmb_id error_msgformatted_schedulefull_error_msg existing_rankr stage_actionsstage_index_to_rank_mappings rD_process_actionz+_validate_schedule.._process_actionTsc!!'''' A: $  " & &u - aZM$/22D6.PQUPVW""'(FH&<t&"!k!>?Q>RS%^44 $  " & &u - aZM$/22D6.QRVQWX""'(FH&<t&"!k!>?Q>RS%^44 $  " & &u - aZM$/22D6.RSWRXY""'(MO&<t&"!k!>?Q>RS%^44 $  " & &u - 2 204 ' -7=M=( vWTF(4&8RSWRXXbcpbqr (rFrrz: Got an invalid action: z, expected instance of _ActionzGot rz microbatches for stage z , expected z(Invalid backward microbatches for stage z8: I and W must have equal counts, but got I=z, W=r&z : expected z( total backwards, but got B=z, I=) rwrr2rr;r3r4rSrvrrrW)rrRr rrstage_idrrrKrcrf_mbb_mbi_mbw_mbrrs` @@rD_validate_scheduler;s w<= ( <]O9UXY`UaTbc (m$OwN"Gv NNOj)=   su su su su   =M#%99s9#9v4%gdm4 4LD&~fg. vWTF*CF8Kij . !!-"("4"4.s%--2LHrFzDeprecation warning: 'use_full_backward' is no longer supported. Simply stop passing it, and everything should still work fine.)rr_stagesr r! group_sizerR group_rankrrstage_index_to_group_rank_stages_forward_initialized_stages_backward_initializedr_should_compute_lossr_backward_requires_autogradrwarning) rBrrrrrrrrrrrrs @rDrzPipelineScheduleMulti.__init__s )+//#   !!9//#AY111I(( )G    0 0* &\\ ME.2.L.LE + M,1(,1)d2$L!@B ,F(  ( NNQ  )rFrc|jsg}|jD]!}|j|j#t t |t }|jD]I}|jr|j|j||}-|j|j||}Kd|_|jrA|js4|jD]}|j|jd|_ yyyr&) rrrr'rr rris_firstr(rrrr))rBrrr*rnext_stage_argss rD_initialize_stagesz(PipelineScheduleMulti._initialize_stagess//)+G Du@@BC D Jw/ 005wO >>&+&B&B,,dF'O',&B&B,,ov'O  04D ,   d&G&G D--d.B.BC D04D -'H rFrrfct||j|j|j|_|j D]}|j|_y)z] Allocates the stage index to rank mapping which is needed for communication N)rrRr!rrr)rBrrs rD_validate_and_set_stage_mappingz5PipelineScheduleMulti._validate_and_set_stage_mappingsT *<         * & \\ ME.2.L.LE + MrFct|dd5}tj|}|jD] }|j |j|" dddy#1swYyxYw)QDump a CSV representation of the schedule into a file with the provided filename.rrknewlineN)opencsvwriterrwriterow)rBfilenamecsvfilerrs rD _dump_csvzPipelineScheduleMulti._dump_csv"s_ (C , ;ZZ(F++ ; 3 3D 9: ; ; ; ;s AAA'cH|dk(sJt|d5}tj|}t|D]5\}}|Dcgc]}tj |c}|j |<7 ddd|j|j ycc}w#1swY*xYw)zLoad a CSV representation of the schedule from a file with the provided filename. This API will most likely get renamed/refactored so is marked as internal for now. format must be "compute_only" for PipelineScheduleMulti. compute_onlyrkrN)rrreaderrrSrLrr)rBrformatrrrrrs rD _load_csvzPipelineScheduleMulti._load_csv)s ''' (B ' O7ZZ(F&v. O cJM,NQW-=-=a-@,N##D) O O ,,T-@-@A -O O Os+BBBBB!rrrc"|jr+|jrtjs t d|j D]}|j|_|j D]}|j|j||\}}|*ttj||j} nd} |j||| |||j D].}|js|s|j|jcSyr-)rrrr.rIrr/r0rrr1rrrrr) rBrrrrrrrrr2s rDrzPipelineScheduleMulti.step9s(   00))+. \\ 4E!%!3!3E  4\\ )E  & & ( )$(#5#5dF#C L   !3!3FD+>?? @rFrrrc  |j||||\}}|j|d|d|jDcic]}|j|}}t }t } |D]\} | dkDr!|j |j | dz | |jdz ks<| j |j | dz^t} t|j|jD]\} } g}| (| j}| j}| j} |Jd|tjk(rV|| }|j!||||||}|j#|||||j%|j'|n|tj(k(r|| }|j+||}| | xxdz cc<| | |j,k(}|j.r |j,nd}|j1||d||r|j/||j%|j3|n|tj4k(rM|| }|j+||}|j1||dd|j%|j3|n|tj6k(re|| }| | xxdz cc<| | |j,k(}|j.r |j,nd}|j9|| |r |j/|nt;d ||D]}|j|}d}| t=|kr|| }|*|j}|j}|j} |Jd|tjk(r1| dz|vsr|| dz}|j%|j?||t(t4t6fvrt;d || D]}|j|}d}| t=|kr|| }|*|j}|j}|j} |Jd|tt6fvrf|t4t(fvr1| dz |vs||| dz }|j%|jA|t;d |tCtE||jU|j|ycc}w#tF$rs}tHjKd |j|jLjNtQ|| | tHjKd tS|j| |d}~wwxYw) 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. rrNzCAll currently supported action types require valid microbatch_indexrETr full_backwardrIFrIzUnknown computation type zi[Rank %s] pipeline schedule %s caught the following exception '%s' at time_step %s when running action %sz%sr)+rrrrTrrrr!rrrrrUrVr%r>r?rrr@rArrrrLrMr?r@backward_weight_one_chunkrrwr>rKrr  ExceptionrerrorrrMrbrr)rBrrrrrrstage_index_to_stageall_prev_ranksall_next_ranksrTbackward_counter time_steprKrrUrrrrIgrad_scale_factor prev_rank prev_rank_opsprev_rank_action next_rank next_rank_opsnext_rank_actiones rDrz(PipelineScheduleMulti._step_microbatchests"//JPVW  IaL9 37,,? ).E  u $? ? $'5#&5/ TKQ""4#A#A+PQ/#RST--11""4#A#A+PQ/#RS  T*1!*4+>+>tyy+I!JW  IvV (*%'-'>'>$%66H"("4"4K#/]/(+;+C+CC 4[ A!&!8!8$#H-%h/0> "9" 00 HU 5#9#9(#CD)-=-K-KK 4[ A#33E8D(5:5,[9T=Q=QQ&594D4DD00!*00$!%*.*7 1 )!--.?@ 5#9#9(#CD)-=-L-LL 4[ A#33E8D00$!%*/*/ 1  5#9#9(#CD)-=-M-MM 4[ A(5:5,[9T=Q=QQ&594D4DD00!*77$*78)!--.?@(+DEUDV)WXX"0I$($7$7 $BM'+$ 3}#55+8+C('3+;+L+L(#3#D#D&6&B&B '3a3,/?/G/GG*Q2FF)=[1_(M # 5+A+A(+K L-)*+2 !",";"0I$($7$7 $BM'+$ 3}#55+8+C('3+;+L+L(#3#D#D&6&B&B '3a3,/II -.-1PP*Q2FF)=[1_(M # 5+A+A(+K L",";B?C@D)- +/4'(44D 4 34t; 4  _ 45< 4 S>E#J6=4 $;44%)4l5uS#X5: MCgn!556 M  M;B&"# 9t 9  9z $!%"&"# }2}2$;}24K }2 t }2  }2rFrc VeZdZ ddedeedzdeedzdedzdedzf dZy) _PipelineContextN schedule_refrrrrcJ||_||_||_||_||_yrY)rrrrr)rBrrrrrs rDrz_PipelineContext.__init__5s() "$ rFr)rMrNrOrrrrrrrQrFrDrr4s_'+'+"&" ' et# :$  4K  t rFrc eZdZdededdfdZy)_CustomFunctionProtocolrKctxrfNcyrYrQ)rBrKrs rD__call__z _CustomFunctionProtocol.__call__EsrF)rMrNrOrSrrrQrFrDrrDsKwK-=K$KrFrc eZdZdZfdZdededdfdZ ddee e e dzfd e ffd Z dd e d e ffd Zdd e d e fd ZdZdefdZ dde dzde dzde dzde dzdef dZxZS)_PipelineScheduleRuntimea% Provides a simple runtime that requires a 'schedule IR' including specified communication operations. Can be instantiated directly by creating _PipelineScheduleRuntime and calling load_csv, or can be subclassed and the subclass can be responsible for creating a schedule IR. ct||i|i|_t|_i|_i|_tt|_ t|_ yrY) rr_comp_type_to_function_maprr bwd_recv_ops fwd_recv_opsrr unshard_opsrunsharded_stages)rBrrrs rDrz!_PipelineScheduleRuntime.__init__PsU $)&)LN'.5iEGDF None rHz. Only FORWARD, FULL_BACKWARD, BACKWARD_INPUT, BACKWARD_WEIGHT, OVERLAP_F_B, UNSHARD, RESHARD and REDUCE_GRAD are supported.zTComputation type %s is already registered. Overwriting the existing custom function.N) r>rAr?r@r<r5r6r=rr rr)rBrUrs rDregister_custom_functionz1_PipelineScheduleRuntime.register_custom_function_s          $ +,<+=>no  t>> > NN<   =L''(89rFrrc t|i_|dk(rC|D]=}gj|<||D]$}|Jj|j|&?y|dk(r|j D]<\}}t |D])\}}| |j rtd|d|d|d>|D]M}t||j|<tj|jj|<Otjfdj _ytd |d ) z Given an in-memory representation for a simple compute-only schedule, lower it to a complex schedule including communication actions. Stores the schedule in self, and must be called before running step_mo() compute_commsNrz?Expected compute-only schedule but found communication action 'z ' at rank z , position ze. Communication actions (e.g. SEND_F, RECV_F, etc.) should not be present when format='compute_only'.c"j|SrYrrrBs rDrzG_PipelineScheduleRuntime._prepare_schedule_with_comms..s(F(Fq(IrF)rr format= is not implemented)rrpipeline_order_with_commsrqrrrhrrrQrrr!r)rBrrrrK action_listrrs` rD_prepare_schedule_with_commsz5_PipelineScheduleRuntime._prepare_schedule_with_commss /8CE& _ $ H79..t4%dmHF!---2248??GH H ~ %%,]]_ !k!*;!7IAv)&2F2F( &xz${1#FPQ   7KDM8..t48H2248((8..t4  .<..I++.D * & 1D&EF FrFrc|dk(r+t |||j|jy|dk(rzi}t |d5}t j |}t|D]+\}}|Dcgc]}tj|c}||<-|j||dddytd|dcc}w#1swYyxYw) a Loads a csv in simple format and then lowers it to include communication actions format must be either "compute_only" or "compute_comms". If compute_only, the lowering passes will automatically be run to generate a compute_comms schedule. rrrkr)rNrr) rrrrrrrrrSrLr) rBrrrrrrrrrs rDrz"_PipelineScheduleRuntime._load_csvs ^ # G h '  - -d.A.A B  &Gh+ JwG,!*6!2GID#BE$FQW%5%5a%8$FGDMG11'&1I  J J & 1D&EF F%G J Js+C1C CCCc|dk(rp|jJdt|dd5}tj|}|jD] }|j |j|" dddy|dk(rp|j Jdt|dd5}tj|}|j D] }|j |j |" dddyy#1swYyxYw#1swYyxYw) rrNz'Compute only schedule must be availablerrkrrz6Must initialize compute_comms schedule before dump_csv)rrrrrr)rBrrrrrs rDrz"_PipelineScheduleRuntime._dump_csvs  ^ #&&2 9 2hR0 ?GG, //?DOOD$7$7$=>? ? ? &11= H =hR0 JGG, ::JDOOD$B$B4$HIJ J J '  ? ? J JsAC,AC8,C58DcLtjfdjS)Nc"j|SrYrrs rDrz4_PipelineScheduleRuntime._simulate..sd44Q7rF)_simulate_comms_computerr!r[s`rD _simulatez"_PipelineScheduleRuntime._simulates%&  * * 7     rFrc:t|jt}|r|j}||jvrL|j|D]}|j |j|=|j j|||j vs Jd|yy)zQIf an unshard is active for `stage_idx`, wait() it and mark `stage_idx` unshared.z*Attempted to compute on sharded stage_idx=N)rsubmodrrTrrrr)rBrstage_uses_fsdprxrs rD_assert_unshardedz*_PipelineScheduleRuntime._assert_unshardeds$U\\:> ))ID,,,**95BGGI$$Y/%%)))4 5 55 =9,? 5 rFrrrrrc  j|\jddjDcic]}|j|c}jJdg dt ddf fd }j jtjjD]\}} tjd||  tt| 5| jjvr0t!|} j| j| | nH| jt"k(r-| j$Jd| j$D] } ||  n|| ddd rt- j/ rt1j2dk(sJd j5j|ycc}w#1swYjxYw#t&$rH} tj)d || tj)t+j| | d} ~ wwxYw) rrNzLMust call _prepare_schedule_with_comms() before calling _step_microbatches()rKrfc |j}|j |jnd}|dk\s|tttfvs Jd|d|j }|}t |jt}|dzv}|dz v}|tk(r*jt|j|y|tk(r*jt|j|y|tk(rH||fj vsJd|d|dt|j#|j ||f<y|t$k(rH||fj&vsJd|d|d t|j)|j&||f<y|tk(r|r|j*vr|j,vs Jd |d |jj/D]Q}t |tst1t2|j5d } j,|j| Syy|tk(r|r|j*vs Jd|d|j,vs Jd|d|jj/D]#}t |ts|j7%j*j9|yy|t:k(rj=||j>sC|sA||fj vs Jd|dtAj jC||f|jE|||} jG|| ||r|dzjI| |yy|tJk(rj=||jLsC|sA||fj&vs Jd|dtAj&jC||fjO||} jP|xxdz cc<jP|jRk(} |jU|| d | |r(|dz jW|jY||yy|tZk(rj=||jLsC|sA||fj&vs Jd|dtAj&jC||fjO||} |jU|| dd|r(|dz jW|jY||yy|t\k(rXj=|jP|xxdz cc<jP|jRk(} |j_|| y|tk(r,j`r jRnd} |jc| yted|d)Nrlrzaction=z missing mb_indexrzRecv twice for stage_idx=z mb_index=z without executing forwardz without executing backwardzUnsharding the same stage_idx=z twiceT)async_opzResharding stage_idx=z without unshardingz before finishing unshardzComputing action=z before receiving inputrEz Attempted to run compute action=rFrz is unknown or unsupported)3rUrVr5r6r=rTrr$rr7rqr r@r9rMr8r r>r:r rKrrmodulesr runshardreshardrr>r&rrrr?rset_local_fwd_inputrArrrrrLset_local_bwd_inputget_local_bwd_outputr?r@rrrNr)rK comp_typerrxrr%is_next_stage_on_this_rankis_prev_stage_on_this_rank submodulehandlerrrIrrrrrBsend_opsrrs rD_perform_actionzD_PipelineScheduleRuntime._step_microbatches.._perform_actions//I+1+B+B+N''TV q=I2% -&+,  - **I(3E(zBO)2Q:N)N &)2Q:N)N &F" 5+A+A(+K LMf$ 5+A+A(+K LMf$**+1i\8+=WX + )>>%T-=-==A9i\@A>&+\\%9%9%;C ))Z@$!%mY5F5FPT5F5U!V((3::6B C #g%"$(=(==0i\1DE=%D,<,<<0i\1JK<&+\\%9%9%;, ))Z@$!))+,))00;#g%&&u-6" **+-VI-DE + $D$5$5$9$99h:O$PQ00H%h'(6 1 (( HM.(Q7KK. m+&&u- 6" **+rFc t|j}t||j|z|j}t|Dcgc]}d}}|D]/|j fdt|j D1d|jdz |z z}|j dg|zt |D]8|j fdt t|j D:|Scc}w)Nc3TK|]}ttj|!ywrY)rSr%r>rrrTs rDrzFScheduleLoopedBFS._calculate_single_rank_operations.." s) %5%=%=xH%(r&rc3TK|]}ttj|!ywrY)rSr%rAr>s rDrzFScheduleLoopedBFS._calculate_single_rank_operations..- s) %5%C%CXNr?)rwrrrRrrreversed)rBrn_local_stages stage_indicesrar;post_warmup_opsrTs @rDr:z3ScheduleLoopedBFS._calculate_single_rank_operations sT\\* $$$~5t7I7I 9>d )D1$)D)D( K OO %d&:&: ;  t11A5<=01#M2 K OO (t/C/C)D E   %*Es C9)NNTT)rMrNrOr4rrrvrrrrbr rrr~rr:r5r6s@rDr r s,0@D +/?'(??E!D( ?  S>E#J6= ?  ?%)?<rFr c tt} tt} tt} t|D cgc]} d}} ||zd|dz |z zz||zz }| r||z dz }||z|z}g}d}| rtnt}t|D]}||kr^||}| |x}dz| |<|j t |tj|||dz k(sQ|jdg|zg||cxkr||zkrnn||}| |x}dz| |<|j t |tj|||}| |x}dz| |<|j t ||||j || s||z |k\s|||}| |x}dz| |<|j t |tj||dz }O| s|j d||}| |x}dz| |<|j t ||||j || s||z |k\s|||}| |x}dz| |<|j t |tj||dz }| rf|t|krX|||}| |x}dz| |<|j t |tj||dz }| r|t|krX|Scc} w)Nr&rr) rrvrr?rArqrSr%r>rr@rw)rBrR warmup_ops fwd_bwd_ops cooldown_opsrforward_stage_indexbackward_stage_indexnum_1f1b_microbatchesenable_zero_bubblefwd_stage_mb_indexbwd_stage_mb_indexweight_stage_mb_indexrar;rD total_opsbackward_op_idsweight_op_countFULL_BACKWARD_OR_BACKWARD_INPUTrfwd_stage_indexrr]bwd_stage_indexr^weight_stage_indexweight_mb_indexs rD_get_1f1b_rank_opsrX4 s*5S)9)4S)9,7,<5:$K%@qd%@H%@ &ma.?$.F)GG d O'$.2[(<7IOO--$IM% ?1"5O/??3  / OO)9)A)A8L Z!^# 89 2 8 [ 8 81"5O 2? CC 3  / OO)9)A)A<P 326O 2? CC 3  / OO)H,W   " "2 &!b:o9N&N%9#O4&"(==O'PPO=%&89*(88' 1$ &%226O 2? CC 3  / OO)H,W   " "2 &!b:o9N&N%9#O4&"(==O'PPO=%&89*(88' 1$[M%^ 33G!G1//2RS45GH HO 501  "$4$D$Do  1 33G!G Og&As K2ceZdZdZ ddeedededzdee dfdzde e e fdzd e e e fee zdzd e d e ffd Zd eedzfdZxZS)ra 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"). This schedule is mostly similar to the original paper. It differs by being relaxing the requirement of num_microbatch % pp_size == 0. Using the flex_pp schedule, we will have num_rounds = max(1, n_microbatches // pp_group_size) and it works as long as n_microbatches % num_rounds is 0. As a few examples, support 1. pp_group_size = 4, n_microbatches = 10. We will have num_rounds = 2 and n_microbatches % 2 is 0. 2. pp_group_size = 4, n_microbatches = 3. We will have num_rounds = 1 and n_microbatches % 1 is 0. Nrrrr.rrrrc |dj|_t | ||||||||t ||_|dj |_td||jz|_ ||jz|_ ||jzdk7rtd|jd|di|_ t|jD]"} |j| } | |j| <$|j|jy)Nrrrrrrrrrrz_Interleaved 1F1B requires the number of microbatches to be a multiple of the number of rounds ( ), but got .)rrRrrrwrBrrrnumber_of_roundsmicrobatches_per_roundrrrr:r rBrrrrrrrrrr;rs rDrz ScheduleInterleaved1F1B.__init__ s-$AY11 )+//#'A  "&k1I(( #A~9K9K'K L&48M8M&M# D11 1Q 65595J5J4KL)*!- @B$,,- 1D==dCH(0D   % 1 ))$*=*=>rFrfc  fd}| jjz}| z }||z } |z|z}tjd |||fd} fd}t jj ||||S)Ncjdz jz}d}||jdz |z zz}t|jjzS)Nrr&rBr_rRr[rrwarmups_ops_last_stagemultiply_factorrFrBs rDget_rank_warmup_opszVScheduleInterleaved1F1B._calculate_single_rank_operations..get_rank_warmup_ops o##a'++&, " O//##a'4/3J z4#7#7$:M:M#MN NrF=rank %s, warmup_ops %s, 1f1b %s, cooldown_ops %s total_ops %sc`|jzjz}|jzzSrYr_rBrRr local_indexrrBs rDrIzVScheduleInterleaved1F1B._calculate_single_rank_operations..forward_stage_index 44#>#>>$BUBUUK$"4"44< .backward_stage_index! W##:%$*E*EE%%&&   $"4"44< B?C@D +/)?'()?)?D )? 34t; )?  _ 45< )? S>E#J6=)?)?%))?V9 gn9M9 rFrceZdZdZ ddeedededzdee dfdzde e e fdzd e e e fee zdzd e d e ffd Zd eedzfdZdZxZS)r!aw The Interleaved Zero Bubble schedule. See https://arxiv.org/pdf/2401.10241 for details. Will perform one forward and one backward on inputs for the microbatches in steady state and supports multiple stages per rank. Uses the backward for weights to fill in the pipeline bubble. In particular this is implementing the ZB1P schedule in the paper. Nrrrr.rrrrc t||jj|dj|_t |||||||||t||_|dj|_ td||jz|_ ||jz|_ ||jzdk7rtd|jd|di|_t!|jD]"} |j#| } | |j| <$|j%|j|jz|_|j'|jy)Nrr[rzZZero bubble requires the number of microbatches to be a multiple of the number of rounds (r\r])"_check_torch_compile_compatibilityrrMrrRrrrwrBrrrr^r_rrrr:_add_bubbles_to_actionsrr`s rDrz&ScheduleInterleavedZeroBubble.__init__A sj +64>>3J3JK#AY11 )+//#'A  "&k1I(( #A~9K9K'K L&48M8M&M# D11 1Q 65595J5J4KL)*!- @B$,,- 1D==dCH(0D   % 1#::   $"4"4 4  ))$*=*=>rFrfc  fd}| jjz}| z }||z } |z|z}tjd |||fd} fd}} t jj ||||| d S)Ncjdz jz}d}||jdz |z zz}t|jjzSrqrcrds rDrgz\ScheduleInterleavedZeroBubble._calculate_single_rank_operations..get_rank_warmup_opsv rhrFric`|jzjz}|jzzSrYrkrls rDrIz\ScheduleInterleavedZeroBubble._calculate_single_rank_operations..forward_stage_index rnrFcjdz |z jzjzz }|jzzSrqrprqs rDrJz]ScheduleInterleavedZeroBubble._calculate_single_rank_operations..backward_stage_index rrrFT)rLrs) rBrrgrtrGrHrPrIrJrKrFs `` @rDr:z?ScheduleInterleavedZeroBubble._calculate_single_rank_operationsu s O). ,,t/C/CC$z1 % 3 ,|;  K        =  =!%!            !#  rFc|j}d}t}i}i}i}d}t|jD]} g|| <d|| <d|| < d} t} t|jD]} || } | t || k\rd} || | || | } | J| \}}}}||||||s>|| j || | || j |||f|| xxdz cc<|| j d|| xxdz cc<|| xxdz cc<|| j d|j| | rn|dkDrtjd|||S)Nc|tjk(r|dk7r |dz ||f|vryy|tjk(r'||dz k(r|tj|f|vS|dz||f|vSy)NrrTF)r%r>rA)rr microbatchnum_stages_globalseen_opss rD need_bubblezJScheduleInterleavedZeroBubble._add_bubbles_to_actions..need_bubble s%---A:519b*"=X"M  '555-11!#3#;#;ZHPXXX 2z2(BBrFrTFrz?Non zero bubbles added: total_bubbles_added=%s bubbles_added=%s) rrrrRrwrqrupdaterr)rBrrrrresult next_pointer bubbles_addedtotal_bubbles_addedr should_stop temp_seen_ops timestamp temp_actionrTrrras rDrzz5ScheduleInterleavedZeroBubble._add_bubbles_to_actions s%% B?C@D +/2?'(2?2?D 2? 34t; 2?  _ 45< 2? S>E#J6=2?2?%)2?h> gn9M> @@rFr!ceZdZdZ ddeedededzdee dfdzde e e fdzd e e e fee zdzd e d e ffd Zd eedzfdZxZS)r"a The Zero Bubble schedule (ZBV variant). See https://arxiv.org/pdf/2401.10241 Section 6 for details. This schedules requires exactly two stages per rank. This schedule will perform one forward and one backward on inputs for the microbatches in steady state and supports multiple stages per rank. Uses backward with respect to weights to fill in the pipeline bubble. This ZB-V schedule would have the "zero bubble" property only if time forward == time backward input == time backward weights. In practice, this is not likely true for real models so alternatively a greedy scheduler could be implemented for unequal/unbalanced time. Nrrrr.rrrrc t||jj|dj|_t |||||||||t|j|jd|_ |jD]} |j| _ t||_ |jdk7rtd|jd|dj|_|dj |_i|_t%|jD]"} |j'| } | |j"| <$|j)|j"y)Nrr[vstyler&0ZBV requires exactly 2 stages per rank, but got r]ryrrMrrRrrrr!rrrwrBrrrr rrr:r rBrrrrrrrrrrr;rs rDrzScheduleZBVZeroBubble.__init__ s\ +64>>3J3JK#AY11 )+//#'A  *H    0 0* &\\ ME.2.L.LE + M"&k   ! #B&&'q*  1I((  ).. @B$,,- 1D==dCH(0D   % 1 ))$*=*=>rFrfctd|jzdz |j}t|Dcgc]}d}}d\}}}}d|j|z zdz } |} |jdz |z } t| D](}|j t | t||dz }*|} t| D]N}|j t | t||dz }|j t | t||dz }P|j|z } t| D]o}|j t | t||dz }|j t | t||j t | t||dz }q||ks||kr||kr&|j t | t||dz }|j t | t||j t | t||dz }|j t | t||dz }|j t | t||j t | t||dz }||kr||kr||}}|}t|D]N}|j t | t||dz }|j t | t||dz }P|j|z }t|D]N}|j t | t||dz }|j t | t||dz }P||kr,|j t | t||dz }||kr,||kr,|j t | t||dz }||kr,||k(r||k(sJ||k(r||k(sJ|Dcgc]-}|'|j|j|jkr|nd/}}|Scc}wcc}w)Nr&r)rrrr)rUrV) rrRrrr rqrSr2r3r4rV)rBrn_microrar;f0_cntf1_cntb0_cntb1_cnt warmup_n1stage_id_chunk0stage_id_chunk1 warmup_n2 warmup_n3w0_cntw1_cnt cooldown_n1 cooldown_n2rKs rDr:z7ScheduleZBVZeroBubble._calculate_single_rank_operations; s[a$,,,q0$2F2FG8=d )D1$)D)D*4&++d23a7 //A-4y! A OO!fU  aKF    y! A OO!fU  aKF OO!fU  aKF &&- y! A OO!fU  aKF OO!fU  OO!fU  aKF vo'!1'!f !  OO!fU  OO!fU  aKF OO!fU  aKF OO!fU  OO!fU  aKF5vo'!18  {# A OO!fU  aKF OO!fU  aKF ((4/ {# A OO!fU  aKF OO!fU  aKF vo OO!fU  aKF vo vo OO!fU  aKF vo Ff$444Ff$444#   %++7++d.B.BB     a*EL  s P2P rurvr6s@rDr"r" s &$(>B?C@D +/1?'(1?1?D 1? 34t; 1?  _ 45< 1? S>E#J6=1?1?%)1?ftgn9MtrFr"ceZdZdZ ddeedededzdee dfdzde e e fdzd e e e fee zdzd e d e ffd Zd eedzfdZxZS)r#z The DualPipeV schedule. A more efficient schedule variant based on the DualPipe schedule introduced by DeepSeek in https://arxiv.org/pdf/2412.19437 Based on the open sourced code from https://github.com/deepseek-ai/DualPipe Nrrrr.rrrrc t||jj|dj|_t |||||||||t|j|jd|_ |jD]} |j| _ t||_ |jdk7rtd|jd||jkrtd|d |jd |dj|_|dj |_i|_t%|jD]"} |j'| } | |j"| <$|j)|j"y) Nrr[rrr&rr]zDDualPipeV requires at least as many microbatches as stages, but got z microbatches and z stages.rrs rDrzScheduleDualPipeV.__init__ s +64>>3J3JK#AY11 )+//#'A  *H    0 0* &\\ ME.2.L.LE + M"&k   ! #B&&'q*  D,, ,V!""4T5E5E4FhP  1I((  ).. @B$,,- 1D==dCH(0D   % 1 ))$*=*=>rFrfcg}ig|j}|j}t||dzd}||\}}dtffd dtdtdtffd }dtdtd t ffd } dtffd } ||z d z dz} t | D]} | ||t|d z} t | D]} | ||t| ||t ||z d z }t |D]&} | ||t| || ||t(||dzz |zd z}t |D]B}|dk(r%||d z k(r| ||t| ||tn ||||||||D||z d z }t |D]} | ||t|||||d z}d}t |D]V}||dzk(r |dzd k(rd}|rtnt}| |||||dzk(r |dzdk(rd}|rtnt}| |||X||z d z }t |D]"} | ||rtnt}| |||$|d z}t |D] } | | |S)Nr&rrrTc|tf}|tf}j|ddz|<j|ddz|<y)zbHelper method to increment BACKWARD_INPUT and BACKWARD_WEIGHT counters when FULL_BACKWARD is used.rrN)r?r@r)rT input_key weight_keycounterss rDincrement_backward_countszVScheduleDualPipeV._calculate_single_rank_operations..increment_backward_counts sK$n5I%7J"*,,y!"#BHZ rFr forward_stagebackward_stagec|tf}|tf}j|d}j|d}t|t|t|t|f}|j tdt d||dz|< |y)zYHelper method to add an overlapped forward+backward action which tracks microbatch index.rrlNr)r>r?rrSrArqr<) rrr forward_key backward_keyrgrirWrrs rDadd_overlap_f_bzLScheduleDualPipeV._calculate_single_rank_operations..add_overlap_f_b s)'2K*N;L!k15J",,|Q7K w ; {CK NN72{D+F G%/NH[ ! %n 5rFrUc|tk7r||fn|tf}j|d}|jt ||||tk(r |y|tk(rj||f|dz|<y)Nrr)rAr?rrqrS)rrTrUrrrr weight_queues rD add_actionzGScheduleDualPipeV._calculate_single_rank_operations..add_action s$}4./!>2   ||C+H NN7;0@(K L =0)+6$~5 ''h(?@ (1  rFcsyjd\}}|jt|t||tf}j |ddz|<y)z4Helper method to add a weight action from the queue.Nrr)rrqrSr@r)ractual_stage_indexrWrrrs rDadd_weight_action_if_pendingzYScheduleDualPipeV._calculate_single_rank_operations..add_weight_action_if_pending7 sb2>2B2B12E /  NN&## -o>J#+<< A#>#BHZ rFrr)rrFT) rRrrrvrr%rr>r?rA)rBrrr num_chunksrank_to_stages stage0_index stage1_indexrrrstep_1rastep_2step_3step_4rstep_5step_6 enable_zbr0step_7step_8rrrs @@@rDr:z3ScheduleDualPipeV._calculate_single_rank_operations s (*   && )) 7 y1}C &4D%9" l C3 C 6 6 6  6. - - -/ -0 C$ C$d"Q&!+v 7A w g 6 7v 7A w g 6 w g 6 7 T!A%v 7A w n = ( 1 w g 6 7 i!m+d2Q6v AAv$)a-/7L':7L-@".#/ *+  $T!A%v A w m < *+   v 9AFaKD1HM *3I w i 8FaKD1HM *3I w i 8 9T!A%v 9A ( 1*3I w i 8 9 v 2A ( 1 2rFrurvr6s@rDr#r# s$(>B?C@D +/6?'(6?6?D 6? 34t; 6?  _ 45< 6? S>E#J6=6?6?%)6?p^gn9M^rFr# schedule_namec 0tttttt t ttd }|Dcic]}|j|}}|j}||vr(td|dt|j|||Scc}w)z Maps a schedule name (case insensitive) to its corresponding class object. Args: schedule_name (str): The name of the schedule. ) 1F1BInterleaved1F1BGPipe LoopedBFSInterleavedZeroBubblerr ZBVZeroBubble DualPipeVzUnknown schedule name 'z'. The valid options are ) rrrr r!rrr"r#lowerrrkeys)r schedule_mapklowercase_keyslowercase_schedule_names rDrr s2&!>"8!6.& L-99qaggil9N9+113n4%m_4MdS_SdSdSfNgMh i   '>? @@ :sBc  t|Dcic]}|||Dcgc]}|| c}}}}t|Dcic]}|gc} Dcic] }|tc} dtdtdzf fd }dtdzdtf fd }|r_d}t|D]O}t ||dk(r||d}||r"| |||||j dd }G||dQt|d D]} t || dk(s|| =t|D]`}t ||dk(r |d ||d}||s/|| |d < |j|||j dbt|d D]} t || dk(s|| =|s>td t |D]}td |d||dtd|r_ Scc}wcc}}wcc}wcc}w)aThis function dry-run simulates the actions in the schedule from the perspective of all ranks, and flags any deadlocks caused by missing or misordered communications. It also simulates any bubbles in time where a rank can not execute any action due to waiting for unmet dependencies. The total number of simulator steps can be used as a metric for unit tests involving IR optimization passes as reordering and merging of IR can reduce the number of simulated steps. The simulation is not high-fidelity and does not model overlapping of compute and communication, or cuda streams. Future work may be to enhance this and model the compute time, comms overlap, and even memory. NrrKc\|j|||j|yyrY)rqr)rrK_prev_ops_rank _schedules rDadd_to_schedulez0_simulate_comms_compute..add_to_schedule s3$v&   4 $ $V , rFrfc|y|j} |}|jtk(rd|jdk(ryt|jt|j |vryt|jdz t|j |vryy|jt tfvr|j dz k(ryt|jt|j |vryt|jdzt |j |vryt|jdzt|j |vryy|jtk(ry|jtk(r)t|jt|j }||vS|jtk(r-|dz }t|t|j }| |vS|jtk(rTt|jt |j }t|jt|j }||vxs||vS|jtk(r-|dz}t|t|j }| |vStd|)NTrrFzUnsupported action type ) rTrUr2rSr8rVr?rAr:r@r7r9r) rKrxprev_ops expected_fpeer_stage_idx expected_send expected_b expected_bwrr rs rDrz3_simulate_comms_compute.._ready_to_schedule sd >&& !- ":;  " "a '!!Q&**FF4K4KLPXX**Q.63J3JKxW  $ $(G G!!Z!^3v))663J3JKxW**Q.@W@WX**Q. v?V?VW  $ $ 7  $ $ . !3!3Q8O8OPJ) )  $ $ .&]N#NFF*  ..@Q!-q@@N $N32b2IHQ.QttSU{.QN-c-7T>- 8B7T>8Bd8Bt >* ,D>$'(A-#D)!,F!&)%#D&1t$((+d+ ,5 &A>!$%*"1% & >* ,D>$'(A-r".#D)!,F!&)%*0IdOB'"4(,,V4t$((+ ,5 &A>!$%*"1% & )+A)+L M& I~nT.B1.E-FGH I:; ;W Z kA2/Rs& G=G8G8G= HH8G=c Ng}t|D]\}t||D]I\}}| |jt||jt t tfvrdndd|||ddK^ddl}t|d5}|jd |i|dddy#1swYyxYw) a This function dumps a schedule IR into a chrometrace format so it can be visualized. It is currently very basic and only serves as a graphical alternative to dumping the schedule IR as text. As future work we may extend this to include more accurate heuristics for durations, or let users input durations, add 'flow events' to let the UI show the connection between sends and recvs, and model cuda streams for comm/compute as separate streams on the chrometrace view. N computation communicationXr)rcatphpidtidtsdurrr traceEvents) rrrqrbrUr2r;r4jsonrdump)schedulereventsrtimesteprKrfs rD_dump_chrometracer3 sFx  )(4. 9  Hf~ MMK"22q!Qi?&,"   & h . =&)1-...s =BB$rc |D]}t|jtjjs2|jj D]6}t|t std|dt|jy)z Check if the schedule is compatible with torch.compile. Args: stages: List of pipeline stages to check schedule_name: Name of the schedule for error message Raises: RuntimeError: If any stage uses torch.compile zThe ze schedule is not supported with stage modules that have used torch.compile. Found OptimizedModule in N) rr$rnnModuler*rrIrrM)rrrmodules rDryryW s %,,8 ll**, F&/2"=/*004V 0E0E/FH  rFrY)r')rF)irrrloggingreabcrr collectionsrrcollections.abcrenumr functoolsr typingr r r r rtorch.distributed distributedr  torch._dynamortorch.distributed.fsdprrtorch.nn.modules.lossrtorch.profilerr_utilsrrrrrrrr__all__ getLoggerrMrr%r>r?r@r5r6r7r8r9r:rAr<r=r2r3r4r;compilersrSrbrrrvrrrP2POpWorkr rrrr8rrr~rnrQrrrrrrrr r rXrr!r"r#rr!rryrQrFrDr s  #,$22 )<'*RTT%    8 $;Et;E|  " "!00"22  " "  " "         .. ** **   R S jS l C  %)<d7T>223<Tz< <~Q Q h+TZZ(+d +d499o+37 $** %(4Z #tDII 6$tyy/B.BJ0"10"ft*tnX)Xv!'7!D!$ 'D. !$36$ 'D.$FP'D.)PP ']Pf#'D.)# ']#Lq#tG},-qSE3J'qq #tG}  qhx' #tGdN++ ,x'x'x' x'  #s(^ x'v{2-{2|   LhLd24d2N B0B\FRv 6v r$<Dw4wt^0^BAcA6B#+SE3J#7BEHBJ!.H # $58rF