tjgjddlZddlZddlZddlZddlZddlmZmZddlm Z m Z ddl m Z ddl mZddlmZddlmZddlmZmZmZmZmZddlZddlmZdd lmZdd lm Z m!Z!dd l"m#Z#dd l$m%Z%d dl&m'Z'm(Z(m)Z)d dl*m+Z+m,Z,m-Z-m.Z.d dl/m0Z0m1Z1m2Z2gdZ3ej4e5Z6Gdde7eZ8e8j9Z9e8j:Z:e8j;Z;e8j<ZZ>e8j?Z?e8j@Z@e8jAZAe8jBZBe8jCZCe8jDZDe.dZEe9ZFe:ZGe;ZHeBZIejJdZKGddeZLedeLde7fdZM d^deNeOePeLdzfdeOdzde7fdZQGddeZRd^dePejSd e7dzdePejTfd!ZU d^dePejSd e7dzdeNeOePejTffd"ZVd#ePejTfd$ZWGd%d&eRZXGd'd(eXZYGd)d*eXZZGd+d,eXZ[d-e8de\fd.Z]d/ePeLdzd0eOdePeLdzfd1Z^ d_d3ePeLdzd4eOdePeLfd5Z_d3ePeLdzdePeLfd6Z`d3eNeOePeLfd7e eOgeOfd8eOdeNeOePeLffd9Zad/eNeOePeLfd7e eOgeOfdeNeOePeLffd:Zbd/eNeOePeLdzfd;eOd8eOdd?eRZdeGd@dAZeGdBdCeZfGdDdEedZgGdFdGegZh d`dIZi dadKeOdLeOdMeOd;eOd0eOdNeOdeOfdOZjGdPdQegZkGdRdSegZlGdTdUegZmGdVdWegZndXe7fdYZod7e eOgeOfd8eOfdZZpd[Zqd\ePe0dXe7fd]ZrdS)bN)ABCabstractmethod)Counter defaultdict)Callable) dataclass)Enum) lru_cache)AnycastLiteral NamedTupleProtocol)OptimizedModule) FSDPModule UnshardHandle)_Loss)record_function)generate_rank_to_stage_mappinggenerate_stage_to_rank_mapping InferenceMode) _split_tensor merge_chunkssplit_args_kwargs_into_chunksTensorChunkSpec)_PipelineStageBase _RecvInfo PipelineStage) get_schedule_classPipelineScheduleSinglePipelineScheduleMulti Schedule1F1B ScheduleGPipeScheduleInterleaved1F1BScheduleLoopedBFSScheduleInterleavedZeroBubbleScheduleZBVZeroBubbleScheduleDualPipeVc^eZdZdZdZdZdZdZdZdZ dZ d Z d Z d Z d Zed eddfdZdS)_ComputationTypeFIWUNSHARDRESHARDSEND_FRECV_FSEND_BRECV_BB OVERLAP_F_B REDUCE_GRADactionreturncl t|S#t$r}td||d}~wwxYw)NInvalid computation type )r+ ValueError RuntimeError)r8excs d/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/torch/distributed/pipelining/schedules.pyfrom_strz_ComputationType.from_strEsR N#F++ + N N NC6CCDD# M Ns 3.3N)__name__ __module__ __qualname__FORWARDBACKWARD_INPUTBACKWARD_WEIGHTr/r0r1r2r3r4 FULL_BACKWARDr6r7 staticmethodstrr@r?r+r+6sGNOGG F F F FMKKNN!3NNN\NNNrKr+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 ZdS) _Action stage_indexcomputation_typeNmicrobatch_index)rM. sub_actionsc*|SN)__repr__selfs r?__str__z_Action.__str__ss}}rKc|j6d|jD}dd|d|jjSt |j}||jjz }|j|t |jz }|S)Nc,g|]}t|SrJ)repr).0 sub_actions r? z$_Action.__repr__..ys TTTZZ 0 0TTTrK(;))rQjoinrOvaluerIrNrP)rVsub_action_reprsrepr_strs r?rTz_Action.__repr__vs   'TT4CSTTT Qsxx 011QQD4I4OQQ Q4+,,H -3 3H$0C 5666OrKr9cP|jttttt fvSrS)rOrDrGrErFr6rUs r? is_compute_opz_Action.is_compute_ops&$      )   rK action_stringcd|}|dkrdS|drd|vr|d}|d|}||dzd}g}|r[|dD]E}t|}|||Ft dt|d|rt|ndSt |x}rl| \}} } t t|t| t| rt| ndS|dkrdStd |d ) z Reverse of __repr__ String should be formatted as [stage][action type][(microbatch)] e.g. `2F0`, `1UNSHARD`, `3SEND_F1` Nr^r`rr_)rNrOrPrQzInvalid action string: zD, should be formatted as [stage][action type][(microbatch)] e.g. 2F0)strip startswithfindsplitrMr@appendr+tuple _action_regexmatchgroupsintlenr=) rg bracket_endsub_partcomputation_type_partrQsub_strr\rrrNrOrPs r?r@z_Action.from_strs&++-- B  4  # #C ( ( SM-A-A',,S11K$+ H%2a!!% ! K~~ 7'~~c2277G!(!1!1'--//!B!BJ!-#**:666!1!:!:;P!Q!Q!%2=GE+...4  "'' 66 65 >Cllnn ;K)+;K   ))*:;;),-=)>)>H$%%%D  b 4 ym y y y   rK)rArBrCrt__annotations__r+rPrQrprWrTpropertyboolrfrHrIr@rJrKr?rMrMms&&&&#'cDj'''04K~&-444    t   X 3 3 3 3 \3 3 3 rKrMr8r9c&dt|S)NzPP:rI)r8s r?_get_profiler_function_namers V  rKpipeline_ordererror_step_numbercD tjD]@}tt|D]}|| d||<At dD fdt D} fdt D}ttj |ddi}t}dt|D}dt|g|RD d t|d d zzd fd t|Dz}  fd t||D} | dzd | zdz} | S)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. Nric34K|]}t|VdSrS)ru)r[actionss r? z)_format_pipeline_order..s(HHWCLLHHHHHHrKc g|]D}dt|ttdz zES)zStep r)rIzfillru)r[i num_stepss r?r]z*_format_pipeline_order..sQ<=#a&&,,s3y1}#5#566777rKcDg|]}|dgzS)ri)get)r[keyrrs r?r]z*_format_pipeline_order..s=693y 011rK fillvaluec2g|]}dt|zS)Rank r~)r[rs r?r]z*_format_pipeline_order..s#>>>7SVV#>>>rKc@g|]}td|DS)c3VK|]$}|tt|ndV%dS)Nr)rurI)r[items r?rz4_format_pipeline_order...s7 F F$d.CD NNNA F F F F F FrK)max)r[cols r?r]z*_format_pipeline_order..s?  F F# F F FFFrK rc38K|]\}}|d|VdS.sJ<<)1E5$;q>$$$<<<<<<rKc g|]k\}}|ddfdt|Dz-t|dkrdndzlS)z: rc3RK|]!\}}t|d|V"dSrr~)r[rrrs r?rz4_format_pipeline_order...s?RR4c$ii3+a.333RRRRRRrKNrz <-- ERROR HEREri)ra enumeratertrn)r[rrowrrs r?r]z*_format_pipeline_order..s    E3  ((RRRR9S>>RRR R R S!,EKKMM!$%%):::       rK ) copydeepcopyrangerurvaluessortedlist itertools zip_longestziprar)rrrankr step_labels rank_actionstransposed_actions num_ranks rank_labels header_rowformatted_rowsformatted_tablerrs`` @@r?_format_pipeline_orderrs9]>22N--s>$/0011 - -Ad#A&.*,t$Q' - HH0E0E0G0GHHHHHIAFyAQAQK=CN=S=SL i3\PRPPQQN##I>>U9-=-=>>>K{8%7888K KN++a/0388<<<<5>{5K5K<<<44J     k+=>>   N!4'$))N*C*CCdJO rKceZdZ d)dededejfdzdeedfdzde e efdzde e e fee zdzd e f d Z d*d Zd Zd Zdeede de ddfdZ d*deedee dfe zde e e fdzde de de de e e fdzdee e ffdZe d+dedzdedzdedzdedzde de e e fdzf dZeddddd dedzde e e fdzfd!Zddd"dedzfd#Z d,dedzdedzdedzdedzdeeeff d$Zd*d%Z d*dee dfde e e fdzfd&Zd'ee de fd(ZdS)-_PipelineScheduleNTn_microbatchesloss_fn.args_chunk_speckwargs_chunk_specoutput_merge_spec scale_gradsc||_||_||_||_||_||_ |jdu|_g|_t d|j j dS)NzUsing %s) _n_microbatches_loss_fnr_args_chunk_spec_kwargs_chunk_spec_output_merge_spec _has_backward_internal_lossesloggerinfo __class__rA)rVrrrrrrs r?__init__z_PipelineSchedule.__init__ sv . '!0"3"3 "]$657 J 788888rKc|jr@|j;|||||}|j|dSdSdSrS)is_lastr _compute_lossrro)rVstageoutput target_mbsmb_index loss_kwargslosss r?_maybe_compute_lossz%_PipelineSchedule._maybe_compute_loss+s^ = /T]6%%fj.BKPPD  ! ( ( . . . . . / /66rKcd|cxkot|jknc}|jr|j|r |j|St|jdkr|st d|d|jdS)NrzLoss for microbatch z6 is not available. Available losses for microbatches: )rurrrr=)rVrr valid_indexs r?_maybe_get_lossz!_PipelineSchedule._maybe_get_loss2s8@@@@c$*?&@&@@@@@ = T]6;6(2 2 & ' '1 , ,[ ,NxNN6:6KNN  4rKct|ts|g}td|D}|ry|wt|j|jkr,t d|jdt|j|||j|jdS)zB Update the losses to those in the internal state c3$K|] }|jV dSrSrr[rs r?rz3_PipelineSchedule._update_losses..Es$!D!DE%-!D!D!D!D!D!DrKN Expecting z losses but got ) isinstanceranyrurrr=clearextend)rVstageslossescontains_last_stages r?_update_lossesz _PipelineSchedule._update_losses>s &$'' XF!!D!DV!D!D!DDD  16#54())T-AAA"c!5ccs4K`GaGacc LLNNN MM$/ 0 0 0 ##%%%%%rKr has_backwardp2p_doner9ctd|Drd}ttt|D]}|||}|}d}t ttt|D]`}||}|td|dkr tj n tj }||_ atd|r|jndt!|dS|sLg}|D])}||*t't)|dSdS) uRun the P2P warm-up protocol for the given stages. For ``PipelineStage`` instances this executes the forward/backward vote protocol (which warms up 2-rank sub-communicators) and sets each stage's ``_inference_mode``. For other stage types it falls back to the legacy ``_get_init_p2p_neighbors_ops`` + ``_batch_p2p`` path. Args: stages: The pipeline stages owned by this rank. has_backward: Whether the schedule includes a backward pass. p2p_done: ``True`` if P2P neighbours have already been initialised (avoids redundant init on eval↔train mode switches). c3@K|]}t|tVdSrS)rrrs r?rz0_PipelineSchedule._warmup_p2p..hs,DDEz%//DDDDDDrKN) received_acc)received_resultzP2P warm-up voting failedrz1Rank determined inference_mode=%s for %d stage(s)None)allr rr_warmup_forward_votereversed_warmup_backward_resultr=rrSTATICDYNAMIC_inference_moderdebugrbrur_get_init_p2p_neighbors_ops_wait_batch_p2p _batch_p2p) rVrrraccrresultdetermined_modeall_opss r? _warmup_p2pz_PipelineSchedule._warmup_p2pUs& DDVDDD D D 1'+Cd=16:: Q Q00C0PP*-F48O!$tM':F"C"CDD 8 866v6NN>&'BCCC{{}}))"((&. )8%% LLC)8D%%fF        1(*G D Du@@BBCCCC Jw// 0 0 0 0 0  1 1rKargskwargstargetfwd_initializedbwd_initializedrc |}|r|j|krd}d}| } |jo| } | s| s||fS| r|||j|td|D} | r | djnd9t fd| Dsd| D} t d| t j| 5| r7d} |D]0}|j r|n| }| |j |||j } 1d }| r:d}t|D]&}| |j |j||| }'d }dddn #1swxYwY|D]+}t|t r|,||fS) u'Common stage initialization shared by Single and Multi schedules. Handles mode-change detection (eval↔train), P2P warm-up, RNG forking, forward / backward metadata inference, and FSDP cleanup. Returns the updated ``(fwd_initialized, bwd_initialized)`` flags. Fch|]=}tj|jjdk$tj|j>S)cpu)torchdevicetypers r? z:_PipelineSchedule._initialize_pp_stages..sH   < --2e;; U\**;;;rKrNc3.K|]}|jkVdSrSr)r[r device_types r?rz:_PipelineSchedule._initialize_pp_stages..s;/ / +1FK; &/ / / / / / rKch|] }|j SrJr)r[rs r?rz:_PipelineSchedule._initialize_pp_stages..s>>>FFK>>>rKzOAll stages must have the same device type for RNG forking. Found device types: )devicesr)rT)rrreceived_grad_metar)rrrrrAssertionErrorrrandomfork_rngis_first_prepare_forward_infrarr_prepare_backward_infrarrr _post_metadata_inference_cleanup)rVrrrrrrrr needs_fwd needs_bwdr  device_typesnext_stage_argsr stage_argsprev_stage_grad_metars @r?_initialize_pp_stagesz'_PipelineSchedule._initialize_pp_stagess(#  $ 2o E E#O#O'' &>+>  4 4"O3 3  C   VT%7 B B B  #     *1:gajood  "3/ / / / 5</ / / , , "?>g>>>L 6'366 \ " "7 " L L ' ' ''+#E).!L_J&+&B&B,"%)%7 'C''OO #' ',0$%f--E+0+H+H, $ %+?$/ ,I,,((#'/ ' ' ' ' ' ' ' ' ' ' ' ' ' ' '2 9 9E%// 966888//sA6EEEarg_mbs kwarg_mbsrrreturn_outputsct)a 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. loss_kwargs: extra keyword arguments forwarded to the loss function. NotImplementedError)rVrrrrrrs r?_step_microbatchesz$_PipelineSchedule._step_microbatchess ("!rKrrrrct) 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. loss_kwargs: extra keyword arguments forwarded to the loss function. r)rVrrrrrrs r?stepz_PipelineSchedule.steps ,"!rKrrcf|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#)rVrrrroriginal_has_backwards r?evalz_PipelineSchedule.evalsW!% 2 7!&D 49d6&KKFKK"7D  !6D  6 6 6 6s' 0cdtffd }| ||dn dgjz}| ||dn igjz}| ||d|4t|tst dt |||fS) z* Pre-process/check inputs namec t|ts!t|dt|t |jkr*t djd|dt |dS)Nz must be a list but got a rrz but got )rr TypeErrorrrurr<)mbsr)rVs r?check_type_and_lenz;_PipelineSchedule._check_inputs..check_type_and_len/sc4(( P4 N N499 N NOOO3xx4/// Q!5QQQQs3xxQQ0/rKNrrJrrz losses must be a list but got a )rIrrrr+r)rVrrrrr-s` r? _check_inputsz_PipelineSchedule._check_inputs$s #          w 2 2 2 2dT11G   y+ 6 6 6 6t33I  !  z< 8 8 8  fd++ S Q4<< Q QRRR !!rKc$|j||fi|piSrS)r)rVrrrs r?rz_PipelineSchedule._compute_lossJs#t}VVCC 0ArCCCrKc|s|r)t|||j|j|j\}}||fSdg|jzig|jzfS)zj Splits a full-batch input into chunks (i.e. microbatches) and returns the chunks rJ)rrrr)rVrr args_split kwargs_splits r? _split_inputsz_PipelineSchedule._split_inputsMso  L6 L'D$%' (( $J |+ +4$..t7K0KK KrK 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)rVr4s r?_merge_outputsz _PipelineSchedule._merge_outputsds    #   rKNNNNTrSNNNNTN)NNNN)rArBrCrtrrTensorrprdictrIr r|rrrrrrrrrrr#r'r.rr3r6rJrKr?rrs7;>B?C@D  9 9 9#u|+,t3 934t; 9  _ 45< 9  S>E#J6= 9 9 9 9 9F@D////   &&&.,1'(,1,1 ,1  ,1,1,1,1D.2Q0Q0'(Q0CHo#Q0S#X% Q0  Q0  Q0Q0#s(^d*Q0 tTz Q0Q0Q0Q0f $!%"&"#-1"""$;"4K " t "  "#s(^d*"""^"*"-1 """t " #s(^d* """^"."&T777td{7777, $!%"&" $"$"$"$;$"4K $" t $" tTz  $"$"$"$"LDDDD )-LLCHoLS#X%LLLL. DI #      rKrp2p_opsdescc*t|dkrgS|r|dnd}td||d|D}|tjhkr d|DS|tjhkr d|DStj|S)aK Wrapper over batch_isend_irecv that avoids coalescing for homogeneous batches (all-send or all-recv). Coalescing serializes ops on a single CUDA stream, which causes head-of-line blocking when independent P2P ops could otherwise overlap. Mixed batches still use batch_isend_irecv for deadlock avoidance. rz, rizbatch_p2p %s%sch|] }|j SrJ)opr[ps r?rz_batch_p2p..|s&&&&&&rKchg|]/}||j|j|j|j0S))grouptag group_dstr?tensorrCrD group_peerr@s r?r]z_batch_p2p..~D    DDaeq|D L L   rKchg|]/}||j|j|j|j0S))rCrD group_srcrFr@s r?r]z_batch_p2p..rIrK)rurrdistisendirecvbatch_isend_irecv)r;r<desc_strop_typess r?rros 7||q "*${{{{H LL!8W555&&g&&&HDJ<      DJ<       !' * **rKctt}i}t|dkr|S|D]"}||j|#t |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<)rrrupeerroritemsr)r;r< ops_by_peer work_by_peerr?rTopss r?_sorted_batch_p2prYs0;4/@/@K/1L 7||q((BG##B''''K--//0088 c'$777 T rKworkc8|D]}|dS)zX Waits for a list of dist.Work (typically from _batch_p2p / _sorted_batch_p2p). N)wait)rZws r?rrs, rKceZdZdZ ddedededzdeedfdzd e e efdzd e e e fee zdzd e ffd Z dd Zddddddedzde de e e fdzfdZde eeedzfdzfdZxZS)r!a 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|_||_dS)NrrrrrrF) superr_stage num_stages _num_stages_stage_forward_initialized_stage_backward_initialized_get_pipeline_orderr rVrrrrrrrrs r?rzPipelineScheduleSingle.__init__s{ )+//#      +*/'+0(  $ $ & & rKc x||jg||||j|j|\|_|_dSNr)rrbrerf)rVrrrrs r?_initialize_stagez(PipelineScheduleSingle._initialize_stagesR  & & [M     +  ,# '    +  , , ,rKr rrrc|jr"tjstd|j|j_|j|||\}}|(tt|t|j nd} | ||| ||||jj r!|r||jjSdSr"zstep() requires gradients to be enabled for backward computation; it should not be used under torch.no_grad() context. Please call eval() instead.Nrk)rris_grad_enabledr=rbrclear_runtime_statesr3rr_TARGET_CHUNK_SPECrrrr6r4) rVrrrrrrr1r2 targets_splits r?r#zPipelineScheduleSingle.steps *   e&;&=&= . $(#5   ((***$(#5#5dF#C#C L ! v'94;OPP Q Q Q       #     ;  > &&t{'@AA A4rKr9cdS)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 NrJrUs r?rgz*PipelineScheduleSingle._get_pipeline_order s trKr7NN)rArBrC__doc__rrtrrprr:rIr r|rrlrr#rMrg __classcell__rs@r?r!r!s|$(>B?C@D   !  D  34t;   _ 45<   S>E#J6=       :     ""#-1 :::t :  : #s(^d* ::::xT#tGdN/C*C%Dt%KrKr!cneZdZdZ d dedzdedzdedzdedzded eeefdzf d Z dS) _ScheduleForwardOnlyzo The forward-only schedule. Will go through all the microbatches and perform only the forward pass NTrrrrrrcT||td|||||\}}||dnd}||d|d|g}t|jD]} t d| 5|j| } t| d} | D]} t| |j | || || |j | } t| d} | | dddn #1swxYwYtd|jj|  |D]} t| dS) z< Run one iteration of the pipeline schedule Nz7Forward-only schedule does not support loss computationrForward fwd_recvrSfwd_send[%s] Forwarded microbatch %s)r=r.rlrrrrbget_fwd_recv_opsrYrrforward_one_chunkget_fwd_send_opsrrrrN) rVrrrrrrmaybe_first_targetfwd_sends_to_waitrrXworksrZs r?rz'_ScheduleForwardOnly._step_microbatches9s   !V%7I "//JPVWW.8.DZ]]$ wqz9Q<9KLLL46t+,, U UA A00 9 9k22155)#J???!LLNN**D#D)))) --aYq\JJJk22155)#J???!((888 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 LL79PRS T T T T & " "D D ! ! ! ! " "s C E!!E% (E% r8) rArBrCrurr|r:rIr rrJrKr?ryry3s $!%"&"#-1,",","$;,"4K ," t ,"  ,"#s(^d*,",",",",","rKryceZdZdZ d dedzdedzdedzdedzded eeefdzf d Z d ee ee dzfdzfd Z dS)r$z^ The GPipe schedule. Will go through all the microbatches in a fill-drain manner. NTrrrrrrc |||||\}}||dnd}||d|d||g}t|jD]>} t d| 5|j| } t| d} | D]} t| |j | || || |} |j | } t| d} | | dddn #1swxYwYtd|jj| ||j| || |@|D]} t| g}t|jD]:} t d | 5|j| } t| d } | D]} t| ||j| }|j| || |jd z k |j| } t| d } | | dddn #1swxYwYtd|jj| <|D]} t| ||j||j|jr|jnd dS)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. Nrr{r|rSsave_forward_outputr}r~z Backward bwd_recvrr last_backwardbwd_sendz[%s] Backwarded microbatch %s)r.rlrrrrbrrYrrrrrrrrNrget_bwd_recv_opsrbackward_one_chunkget_bwd_send_opsrperform_reduce_gradr)rVrrrrrrrrrrXrrZrbwd_sends_to_waitrs r?rz ScheduleGPipe._step_microbatchesns ""//JPVWW.8.DZ]]$  AJ ! &8+   46t+,, V VA A00 9 9k22155)#J???!LLNN**D#D))))66wqz9Q<^7k22155)#J???!((888 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9 LL79PRS T T T  $ $T[&*a U U U U & " "D D ! ! ! !46t+,, V VA Q11 9 9k22155)#J???!LLNN**D#D))))++DK;; .."#t';a'?"?/ k22155)#J???!((888 9 9 9 9 9 9 9 9 9 9 9 9 9 9 9" LL8$+:QST U U U U& " "D D ! ! ! ! DK000 ''@P(W(<(88#'#3888  - ,rKrrrrrrc|||||\}}||dnd}||d|d||t|j|j|jjz }d} d} g} g} t|D]} |j| }tt|d|j | || || |}t| |j | } | |dz krt| d} | |j||| || dz } |j| }tt| |zd ||j| }|j| || |jdz k |j| }| dz } | |jkrn|j| }tt||zd |j | || || |}| |j||| ||j | } | dz } Nt|d } | |jkr|j| }tt|d ||j| }|j| || |jdz k t| |j| }t|d } | dz } | |jkt| ||j||j|jr|jnddS)a 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. loss_kwargs: extra keyword arguments forwarded to the loss function. Nrr|rSrrr}Tfwd_send_bwd_recvrbwd_send_fwd_recvrr)r.rlminrrdrbrNrrrrrrrrrrrrrr)rVrrrrrrr warmup_chunks fwd_mb_index bwd_mb_index send_work fwd_sends_ fwd_recvsr bwd_recvsr bwd_sendss r?rzSchedule1F1B._step_microbatchess^$"//JPVWW.8.DZ]]$  AJ ! &8+      t{6 6    &(  }%%  A 44\BBI JyzBBB C C C[22 %,'$2 3F I & & & 44\BBI}q000&yzBBB  $ $ VZ{    A LL ,  44\BBI Jy9'>>I A L'T111,  """ DK000 ''@P(W(<(B?C@D !D  34t;   _ 45<  S>E#J6=6 $!%"&"#-1WYWYWY$;WY4K WY t WY  WY#s(^d*WYWYWYWYrBT#tGdN/C*C%Dt%KBBBBBBBBrKr# action_typec"|ttfvSrS)r.r5)rs r?_requires_reduce_gradrs 1a&  rKrrc Pg}ttfd}|D]}|||g}|jtkr|j|jD]}|||n ||||D]+}|t |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|jr:|j}|xxdz cc<|kr||dSdSdSNr)rrOrNro)a to_schedulerNcntrs r? _leaf_actionz&_add_reduce_grad.._leaf_actionsu !3 4 4 0-K     !   ;>11"";/////  0 021rKN)rrtrorOr6rQrMr7) rractions_with_reduce_gradrrschedule_reduce_grad_stage_idxsr\ stage_idxrs ` @r?rrs68%c**C000000 S S 9  ''***57'  , ,1Jm J J  Z)HIIII J L; < < <8 S SI $ + +GI{D,Q,Q R R R R S ##rKrcompute_actionsmax_active_stagescD dtdttdzdttfd}t g dtf fd }dtf fd }t |D]\}}|||||d tt fd  }tt fd  }|D] } || |D] } ||  |t 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_actionsNr9ct}g}|D]}||jtkrd|j]|jD]?}|j|vr4||j||j@t||krnTx|j|vrI||j||jt||krn|S)zdRemove duplicates (same stage, different microbatch), find next 'count' stages that will do compute.)setrOr6rQrNaddroru)rrseenretrr\s r?next_stage_indicesz0_add_unshard_reshard..next_stage_indicess " "A}%449R&'m?? %1== HHZ%;<<<JJz'=>>>3xx5(()}D00/// 1=111s88u,,!E rKrNc|t|tddSrS)rrorMr/rN active_stagesfsdp_aware_actionss r?_unshardz&_add_unshard_reshard.._unshard0s=+&&&!!'+w"E"EFFFFFrKc|t|tddSrS)removerorMr0rs r?_reshardz&_add_unshard_reshard.._reshard4s=[)))!!'+w"E"EFFFFFrKc |vSrSrJ)srs r?z&_add_unshard_reshard..?s a}&<rKc |vSrSrJ)rnext_ns r?rz&_add_unshard_reshard..As avorK)rtrrMrrfilterro) rrrrrrr8fetchevictrrrrs @@@r?_add_unshard_reshardrs #T'D.5IdSVi0"eeM(*GcGGGGGGGGcGGGGGGG//** 6 > $#$5qrr7JKKV<<< /"" #q'9'A    " " "/"" #q'9'A-0,@,@1,D,Doa(($   #~ 5 5',??"k&===';+GGG  ! !*M6;RSS       " " " "  ! !& ) ) )1 *2 rK stage_to_rankrcc  d|D}d|D dtdtffd dtdtttff fd }dtdzdttdtffd |rd }t |D]t |d ks(t d d t ||d }||j|j}n|f}t fd|Ds|| ||D]}  | |r||\} } | |   | || j  |  | j  | | d t |d kr|=d}|st d||S)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 ci|]}|gSrJrJr[rs r? z"_add_send_recv..s-S-S-S4dB-S-S-SrKc,i|]}|tSrJrrs r?rz"_add_send_recv..s,U,U,UTT355,U,U,UrKr8r9c|jtkr3|jdz ko$|jdz|jkS|jttfvr0|jdko$|jdz |jkSdS)NrrF)rOr,rNrErG)r8rcrs r? _has_commsz"_add_send_recv.._has_commss  "a ' '%a73MM"Q&==v122=3 3 $(G G G%*3}}"Q&00v12203 3urKc:|st|d|j}|j}|j}t ||t krt nt|}|t kr|dzn|dz }t ||t krtnt|}||fS)Nz is not a valid comm actionr) r rNrOrPrMr,r1r3r2r4)r8rctypersendrecv_stage_idxrecvrs r? _get_commsz"_add_send_recv.._get_commssz&!! I F!G!G!GHH H& '(yEQJJ&&FFKK*/1**Q)a-~!vvPPTzrKN prev_actionsc|dS|jtkrX|jdkrMt|jt|j|vrdSt|jdz t|j|vrdSdS|jt tfvr|jdz krtt|jt|j|vrdSt|jdzt |j|vrdSt|jdzt|j|vrdSdSdS)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. NTrrF) rOr,rNrMr2rPrErGr4)r8rrcs r?_ready_to_schedulez*_add_send_recv.._ready_to_schedules< >4  $ ) )f.@A.E.E*FF4KLL  t*Q.63JKK  t5  # 'F F F"j1n44*FF4KLL  t*Q.@WXX  t*Q. v?VWW  t54rKFrrank=z, len(compute_actions[rank])=c3<K|]}|VdSrSrJ)r[rrrrs r?rz!_add_send_recv..s6VVQ))!\$-?@@VVVVVVrKTz6Malformed compute schedule, can't schedule sends/recvs) rMr|rprrrur rQrrorrNr)rrrc comm_actionsrprogressr8 all_actionsrrrrrrrs `` @@@@r?_add_send_recvr~s.T-S?-S-S-SL,U,U_,U,U,UL 7 t        7 uWg-='>      (7T>(W(RV((((((T %?++  D-..22$%N%N%N_T5J1K1K%N%NOOO$T*1-F!f&8&D$0 %i VVVVVV+VVVVV !T"))&111$ P PA &**1---!z!}}P%/Z]] d%T*11$777$T*..t444$]]43C%D%DELLTRRR$]]43C%D%DEII$OOO D ! % %a ( ( (?4())Q..#D)HH  H G %L rKc :tjtj}tjtj}dt dt ffd dt dt ffd }i}|D]\}}g}i} |D]Y} | j|fvr| j | j| j f} | | | <(| j|fvr_|| |krNfd| D} | D]*} | | | +| j | j n| f} | D]}|jtkr<|j |j f} | | vr(| | | N|jtt fvr;|j ||j f} | | vr(| | | | | [| r5t#d|dt%| d |||<|S) a Defers RECV operations to reduce interference with unrelated compute ops, while maintaining deadlock-safe ordering via rank-parity P2P ordering. By default, the schedule places RECV ops as early as possible (ASAP) to overlap P2P communication with computation. However, on some platforms (e.g., AMD ROCm), a pending RECV can block unrelated compute ops that also use the communication fabric (e.g., FSDP allgather inside a forward pass), creating pipeline bubbles. This function defers each RECV to as late as possible, subject to: 1. A RECV must appear before the compute op that consumes its data. 2. Deadlock avoidance via rank-parity ordering (see pytorch/pytorch#172668): - When current rank > peer rank: SEND before RECV is safe, so deferred RECVs from that peer are NOT flushed before SENDs to that peer. - When current rank < peer rank: RECV before SEND is required, so deferred RECVs from that peer ARE flushed before SENDs to that peer. This breaks circular waits: the lower-ranked side always posts RECV first, providing the matching target for the higher-ranked side's SEND. r8r9cf|jkr|jdz S|jdzSrrOrN)r8r2rs r?_recv_peer_rankz(_defer_recv_ops.._recv_peer_rankA  "f , , =!3a!788 8 =!3a!788 8rKcf|jkr|jdzS|jdz Srr)r8r1rs r?_send_peer_rankz(_defer_recv_ops.._send_peer_rankrrKc8g|]\}}|k|SrJrJ)r[kvrrTs r?r]z#_defer_recv_ops..2s9   "a//!:L:LPT:T:T:T:T:TrKNz!Malformed input schedule on rank z,: leftover RECV ops with no consumer found: z9. Every RECV must be consumed by a downstream compute op.)r+r2r4r1r3rMrtrUrOrNrProrrQrDrGrEr rr)rrr4r3rrr action_list new_actionsdeferredr8rto_flush consumerssubr2r1rrTs ` @@@@r?_defer_recv_opsrs0 $F  $F  $F  $F99C9999999 99C9999999 (*F$]]__0#0#k%' LN!# '# 'F&66*:::&++ !' &66*:::&v..$;;     &.nn&6&6   H (>>#**8<<+<+<===='-&8&D""6) ! > >'722?FC4HICh#**8<<+<+<===)m^-LLL?FC4HICh#**8<<+<+<===   v & & & &   JDJJ+/0A0A+B+BJJJ  #t MrKrnum_microbatchesc t|ks"td|dtt|D]}|vrtd|dt|Didtdtdtffd }D]z}t |D]b\}}|t |tstd |d |d |d |j|jD]}||||U||||c{D]} t| t} t| t} t| t} t| t} | |ks td | dtd| d|| | kstd| d| d| | | | zdzz|kstd| d|d| d| d| S)Nz2Schedule has incorrect number of ranks - expected z , actual z%Schedule is missing actions for rank c i|]Q}|ttttttttiRSrJ)r,rr5r-r.)r[stage_ids r?rz&_validate_schedule..csS===   suu suu suu suu  ===rKr8rr#c |j}|j}|j}|tkr( |t|nv|t krr| |tvr6d|d|d|d|d }t |}|d|}t| |t |n|tkrr| |tvr6d|d|d|d|d }t |}|d|}t| |t|n||tkrq| |tvr6d|d|d |d|d }t |}|d|}t| |t|| vr| |<dS |} || kstd|d|d |d |d | dS)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) rNrOrPr,rr5rr r-r.) r8rr#s_idrmb_id error_msgformatted_schedulefull_error_msg existing_rankr stage_actionsstage_index_to_rank_mappings r?_process_actionz+_validate_schedule.._process_actionns(!'' A:: $  " & &u - - - - aZZM$/222HDHHHHQUHH"'HHH&<t&&&"!SS?QSS%^444 $  " & &u - - - - aZZM$/222HDHHHHRVHH"'HHH&<t&&&"!SS?QSS%^444 $  " & &u - - - - aZZM$/222ODOOOOSWOO"'OOO&<t&&&"!SS?QSS%^444 $  " & &u - - - 2 2 204 ' - - -7=MM))$vDvvvvtvvW[vvgtvv*)rKrr z: 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=rz : expected z( total backwards, but got B=z, I=) rur rrMrtrrrQr,r5r-r.)rrrcrrrr#r8r\r f_mbb_mbi_mbw_mbrrs` @@r?_validate_schedulerSs? LLM ) ) g g gY\]dYeYe g g   m$$QQ w   !O!O!OPP P  ==j))===M#%::s:#::::::::x44%gdm44 4 4LD&~fg.. $nDnnnnnnn !-"("4< '&rKceZdZdZ d$deedededzdee dfdzd e e e fdzd e e e fee zdzd e dzd e d e ffd Z d%dee dffdZd&dZde eeedzfddfdZdZ d'de dedfdZddddddedzde de e e fdzfdZ d(d edzd!edzd"edzdedzde de e e fdzf d#ZxZS))r"aX Base class for multi-stage schedules. Implements the `step` method. 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.rruse_full_backwardrbackward_requires_autogradc  t||||||||_|dj|_|dj|_|dj|_t|j|j|_ |jD]} |j | _ d|_ d|_ |j du fd|_i|_| |_|t"ddSdS)Nr`rFc|joSrSr)rhas_losss r?rz0PipelineScheduleMulti.__init__..s%-2LHrKzDeprecation warning: 'use_full_backward' is no longer supported. Simply stop passing it, and everything should still work fine.)rar_stagesrcrd group_sizer group_rankrrstage_index_to_group_rank_stages_forward_initialized_stages_backward_initializedr_should_compute_lossr_backward_requires_autogradrwarning) rVrrrrrrrrrrr rs @r?rzPipelineScheduleMulti.__init__s( )+//#     !!9/#AY11I( )G   0* * &\ M ME.2.LE + ++0(,1)d2$L$L$L$L!@B ,F(  ( NNQ      ) (rKrc |jo|j|jk}|j}||j||||j|j|\|_|_|jr|r|r|dSdSdSrj)r%rr&rr!&_validate_adjacent_stage_communication)rVrrrrreinit_for_mode_switchforward_initialized_befores r?_initialize_stagesz(PipelineScheduleMulti._initialize_stagess"&!A"  $"C C &*%E"  & & L     ,  -# '    ,  -  + :* :.D :  7 7 9 9 9 9 9 : : : :rKr9c dtdtdttdttddf d}|jD]}|j}d|jd D}|jrtn|d z h}||d ||d |jD}|j rtn|d zh}||d ||dS)zGValidate that stage communication follows adjacent-stage topology only.r directionactual_stage_indicesexpected_stage_indicesr9Nc||z }|rFtd|d|dt|dt|dt|d dS)NzLPipelineScheduleMulti only supports adjacent-stage communication, but stage z has z stages z with non-adjacent stages z (allowed adjacent stages: z\). This commonly indicates skip connections, which are unsupported in this schedule runtime.)r=r)rr0r1r2non_adjacent_stage_indicess r?_check_stage_indiceszZPipelineScheduleMulti._validate_adjacent_stage_communication.._check_stage_indices1s *>@V)V &) "-09--@I--$%9::--,22L+M+M-- 455 ---     rKcTh|]%}t|tr|j|j&SrS)rrsource)r[rs r?rzOPipelineScheduleMulti._validate_adjacent_stage_communication..Fs@111dI..148;3J 3J3J3JrKrrz forward recvc h|] }|D]}|| SrSrJ)r[dstsdsts r?rzOPipelineScheduleMulti._validate_adjacent_stage_communication..Us;/////?#???rKz forward send) rtrIrr!rNargs_recv_infor act_send_inforr)rVr5rractual_fwd_recv_sourcesexpected_fwd_recv_sourcesactual_fwd_send_destsexpected_fwd_send_destss r?r+z.s&,N,N,NQW-=-=a-@-@,N,N,NrKN)r rGrHreaderrrrB)rVrKrOrLrVrrs r? _load_csvzPipelineScheduleMulti._load_csvys  ^ # # !Ov!O!OPP P (B ' ' ' O7Z((F&v.. O O c,N,N#,N,N,N#D)) O O O O O O O O O O O O O O O O ,,T-@AAAAAs>A55A9<A9r rrrc|jr)|jr"tjst d|jD]}|j|_|jD]}||||\}} |(tt|t|j nd} | || | ||||jD]'}|jr|r||jcS(dSrn)rr(rror=r!rrpr3rrrqrrrr6r4) rVrrrrrrrr1r2rrs r?r#zPipelineScheduleMulti.stepsW,   0 )++  . \ 4 4E!%!3E  \ ) )E  & & ( ( ( ($(#5#5dF#C#C L ! v'94;OPP Q Q Q       #    \ @ @E} @ @**5+>?????trKrrrc J |||||\}}||dnd}||d|d||d|jD}t} t} |D]\} | dkr#| |j| dz | |jdz kr#| |j| dz]t} t|j |j D]\} } g}|X|j }|j }|j } |td|tjkrp|| }|||||||}|||||||||n|tjkr|| }|||}| | xxdz cc<| | |jk}|jr|jnd}|||d||r|||||n|tjkr`|| }|||}|||d d |||n|tjkrh|| }| | xxdz cc<| | |jk}|jr|jnd}||| |r||nt=d || D]}|j |}d}| t?|kr|| }||j }|j }|j } |td|tjkr;| dz|vr3|| dz}|| ||t*t6t8fvrt=d || D]}|j |}d}| t?|kr|| }||j }|j }|j } |td|t t8fvre|t6t*fvr;| dz |vr3|| dz }||!|t=d |tEtG|#tH$rd}tJ&d |j |j'j(|| |tJ&d tS|j | |d}~wwxYw|*|j|dS) 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. Nrci|] }|j| SrJrNrs r?rz.*? ? ? ).E u? ? ? rKrzCAll currently supported action types require valid microbatch_indexrTr full_backwardrFrzUnknown computation type zi[Rank %s] pipeline schedule %s caught the following exception '%s' at time_step %s when running action %sz%sr )+r.r.r!rrr$rdrrrrrOrPrNr r+rDrrrrrGrrrrrrErFbackward_weight_one_chunkr<rurrrr ExceptionrerrorrrArr)rVrrrrrrrstage_index_to_stageall_prev_ranksall_next_ranksrNbackward_counter time_stepr8rXrOrrrrrgrad_scale_factor prev_rank prev_rank_opsprev_rank_action next_rank next_rank_opsnext_rank_actiones r?rz(PipelineScheduleMulti._step_microbatchess"//JPVWW.8.DZ]]$  AJ ! &8+   ? ? 26,? ? ?  $'55#&55/ T TKQ""4#A+PQ/#RSSST-111""4#A+PQ/#RSSS)0!*4+>ty+I!J!J\ \  Iv[ (*%'-'>$%6H"("4K',a(+;+CCC 4[ A!&!8!8$#H-%h/0> "9"" 00!6:x 5#9#9(#C#CDDDD)-=-KKK 4[ A#33E8DD(555:555,[9T=QQ&594DKD00!*00$!%*.*7 1 )A!--.?@@@ 5#9#9(#C#CDDDD)-=-LLL 4[ A#33E8DD00$!%*/*/ 1  5#9#9(#C#CDDDD)-=-MMM 4[ A(555:555,[9T=QQ&594DKD00!*77$*78)A!--.?@@@()WEU)W)WXXX"0I$($7 $BM'+$ 3}#5#555+8+C('3+;+L(#3#D&6&B #+"0 e##,/?/GGG*Q2FFF)=[1_(M # 5+A+A(+K+K L L L-)*+2 !", NB?C@D)- +/44'(44D 4 34t; 4  _ 45< 4 S>E#J6=4 $;44%)444444nGK::#s(O::::23333j MCgn!556 M  M M M M;;; None r;z. 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) rDrGrErFr6r/r0r7r<rrr))rVrOrs r?register_custom_functionz1_PipelineScheduleRuntime.register_custom_functions          $ o,<ooo  t> > > NN<     =L'(8999rKrNrrOrPc t|i_|dkr]|D]X}gj|<||D]C}|tdt |j||DYdS|dkr|D]<\}}t|D]'\}}| |jstd|d|d|d(=|D]G}t||j|<tj|j j|<Htjfd j _jr t!jfd  _dSdSt#d |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() rQNz$Expected action to be not None, got rNz?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'.cj|SrSr$rrVs r?rzG_PipelineScheduleRuntime._prepare_schedule_with_comms.. s(Fq(IrK)rrccj|SrSrrs r?rzG_PipelineScheduleRuntime._prepare_schedule_with_comms.. sD,J1,MrK)rformat= is not implemented)rarBpipeline_order_with_commsr rrorUrrfr<rrrrrdr~rr)rVrrOrr8rrrs` r?_prepare_schedule_with_commsz5_PipelineScheduleRuntime._prepare_schedule_with_commssU //888CE& _ $ $ H H79.t4%dmHHF~,Q4<<QQ248??GGGG H H H~ % %%,]]__  !k!*;!7!7IAv)%3",!U$*!U!U6:!U!UGH!U!U!U##   7KDM88.t48H248(88.t44 .<.IIII+...D * " 1@2"M"M"M"M222...   &&E&E&E&EFF FrKrKc|dkr=t|||jdS|dkr}i}t |d5}t j|}t|D]\}}d|D||<|||ddddS#1swxYwYdStd|d ) 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. rNrQrirEcBg|]}t|SrJrTrUs r?r]z6_PipelineScheduleRuntime._load_csv..7 s&$F$F$FQW%5%5a%8%8$F$F$FrK)rONrr) rarWrrrGrHrVrr) rVrKrOrrLrVrrrs r?rWz"_PipelineScheduleRuntime._load_csv# sV ^ # # GG  h ' ' '  - -d.A B B B B B  & &Gh+++ JwG,,!*6!2!2GGID#$F$F#$F$F$FGDMM11'&1III  J J J J J J J J J J J J J J J J J J &&E&E&E&EFF FsAB;;B?B?rQc&|dkr|jtdt|dd5}tj|}|jD]"}||j|# ddddS#1swxYwYdS|dkr|jtdt|dd5}tj|}|jD]"}||j|# ddddS#1swxYwYdSdS) rDrNNz'Compute only schedule must be availabler]rirErQz6Must initialize compute_comms schedule before dump_csv)rr rGrHrIrJr)rVrKrOrLrIrs r?rMz"_PipelineScheduleRuntime._dump_csv< s ^ # #"*$%NOOOhR000 ?GG,, /??DOOD$7$=>>>>? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? & &-5$LhR000 JGG,, :JJDOOD$B4$HIIIIJ J J J J J J J J J J J J J J J J J J ' &s#?A<<BB7?DD Dc@tjfdjS)Ncj|SrSrrs r?rz4_PipelineScheduleRuntime._simulate..V sd4Q7rK)_simulate_comms_computerrdrUs`r? _simulatez"_PipelineScheduleRuntime._simulateS s-&  * 7 7 7 7     rKrc$t|jt}|rq|j}||jvrF|j|D]}||j|=|j|||jvrtd|dSdS)zQIf an unshard is active for `stage_idx`, wait() it and mark `stage_idx` unshared.z*Attempted to compute on sharded stage_idx=N) rsubmodrrNrr\rrr )rVrstage_uses_fsdprr?s r?_assert_unshardedz*_PipelineScheduleRuntime._assert_unshardedZ s$U\:>>  V)ID,,,*95BGGIIII$Y/%)))444 555$%T %T%TUUU V V65rKTrrrrrrc |\dnd}dd|djDjt dgdt ddffd }jtjj D]9\} } t d| |  tt| 5| jjvr0t!|} j| j| | nG| jt"kr,| jt d | jD] } || n || dddn #1swxYwY#t&$rQ} td | | tt+j| | d} ~ wwxYwr#t-#t1jdkrt d j|dS) rZNrci|] }|j| SrJr\rs r?rz?_PipelineScheduleRuntime._step_microbatches..~ r]rKzLMust call _prepare_schedule_with_comms() before calling _step_microbatches()r8r9c |j}|j|jnd}|dks)|tttfvst d|d|j}|}t|jt}|dzv}|dz v}|tkr7 t| |dS|tkr7 t||dS|t krO||fjvrt d|d|dt||j||f<dS|t&krO||fjvrt d|d|d t||j||f<dS|tkr|r|jvr |jvst d |d |jD]c}t|tst3t4|d } j| | bdSdS|tkr|r|jvrt d|d|jvrt d|d|jD],}t|ts|-j|dSdS|t<krȉ||j sI|sG||fjvrt d|dtCj"||f|#|||} $|| ||r!|dz%| |dSdS|tLkr||j'sI|sG||fjvrt d|dtCj"||f(||} j)|xxdz cc<j)|j*k} |+|| d | |r4|dz ,|-||dSdS|t\kr̉||j'sI|sG||fjvrt d|dtCj"||f(||} |+|| dd|r4|dz ,|-||dSdS|t^krY|j)|xxdz cc<j)|j*k} |0|| dS|tkr'j1rj*nd} |2| dStgd|d)Nrjrzaction=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 inputrz Attempted to run compute action=r^Fr`z is unknown or unsupported)4rOrPr/r0r7r rNrrrr1rorrr3rr2rrr4rrrrmodulesr runshardreshardrrDrrrrrrset_local_fwd_inputrGrrrgrrset_local_bwd_inputget_local_bwd_outputrErFrarrr<)r8 comp_typerrrris_next_stage_on_this_rankis_prev_stage_on_this_rank submodulehandlerrrrirrrrrVsend_opsrdrs r?_perform_actionzD_PipelineScheduleRuntime._step_microbatches.._perform_action sP/I+1+B+N''TV A %%B%B%B%BCCC*I(3E(zBBO)2Q:N)N &)2Q:N)N &F"" 5+A+A(+K+K L LMMMMMf$$ 5+A+A(+K+K L LMMMMMf$$x(D,===(\)\\x\\\>>%T-===,-VI-V-V-VWWW%*\%9%9%;%;CC ))Z@@%$!%mY5F5FPT5F5U5U!V!V(3::6BBBB C C CC g%%" < (===,I)III!D$444,O)OOO&+\%9%9%;%;,, ))Z@@%$!))++++)00;;;;; < <g%%&&u--- R7 R "8,D4EEE,IIII$D$5$9$99h:O$P$PQQQ00H%h'(6 1 ((6:x .(Q7KK m++&&u---  R7 R "8,D4EEE,XXXX$D$5$9$99h:O$P$PQQQ++E8<<%i000A5000 $ 5i @DDX X (("&"/ ).(Q7KK228<.F( BF F(F F(#F $F(( H2A G>>Hrq)rQr8)rArBrCrurr+rvrr:rtrrMr rrIrWrMrrrr|r rrvrws@r?r{r{s#&&&&& %L*%L1%L  %L%L%L%LT>>>>rKc t|j}t||j|z|j}dt|D}|D]5|fdt|jD6d|jdz |z z}|dg|zt |D]B|fdt t|jDC|S)Ncg|]}dSrSrJr[rs r?r]zGScheduleLoopedBFS._calculate_single_rank_operations.. )D)D)D1$)D)D)DrKc3NK|]}ttj|V dSrS)rMr+rDr[rrNs r?rzFScheduleLoopedBFS._calculate_single_rank_operations.. sG %5%=xHHrKrrc3NK|]}ttj|V dSrS)rMr+rGrs r?rzFScheduleLoopedBFS._calculate_single_rank_operations.. sG %5%CXNNrK)rur!rrrrr)rVrn_local_stages stage_indicesrpost_warmup_opsrNs @r?rz3ScheduleLoopedBFS._calculate_single_rank_operations sNT\** $$~5t7I  *E)Dd )D)D)D(  K OO %d&: ; ;    t1A5<=0111#M22  K OO (t/C)D)D E E    rK)NNTTF)rArBrCrurrrtrrr:rIr rpr|rrrvrws@r?r&r&h s,0@D +/#??'(??E!D( ?  S>E#J6= ?  ?%)???????@rKr&Fc tt} tt} tt} dt|D} ||zd|dz |z zz||zz }| r||z dz }||z|z}g}d}| rtnt}t|D]Q}||krl||}| |x}dz| |<| t |tj|||dz kr| dg|zu||cxkr ||zkrnn||}| |x}dz| |<| t |tj|||}| |x}dz| |<| t |||||| r]||z |krT|||}| |x}dz| |<| t |tj ||dz }| s| d||}| |x}dz| |<| t |||||| r]||z |krT|||}| |x}dz| |<| t |tj ||dz }S| r||t|kri|||}| |x}dz| |<| t |tj ||dz }| r|t|ki| S)Ncg|]}dSrSrJrs r?r]z&_get_1f1b_rank_ops.. s%@%@%@qd%@%@%@rKrrr) rrtrrErGrorMr+rDrrFru)rr warmup_ops fwd_bwd_ops cooldown_opsrforward_stage_indexbackward_stage_indexnum_1f1b_microbatchesenable_zero_bubblefwd_stage_mb_indexbwd_stage_mb_indexweight_stage_mb_indexrr total_opsbackward_op_idsweight_op_countFULL_BACKWARD_OR_BACKWARD_INPUTr?fwd_stage_indexrrbwd_stage_indexrweight_stage_indexweight_mb_indexs r?_get_1f1b_rank_opsr s*5S)9)9)4S)9)9,7,<,<&A%@E$KK%@%@%@H &ma.?$.F)GG d O3'$.2[(<7IOO-?-$IM%M% ??11"55O/??3  / OO)9)A8LL   Z!^## 8999 2 8 8 8 8 [ 8 8 8 8 8 811"55O 2? CC 3  / OO)9)A<PP   32266O 2? CC 3  / OO)H,WW     " "2 & & &! %b:o9N&N&N%9%9#O4&&"(==O'PPO=%&89*(8' 1$ & &%%%22266O 2? CC 3  / OO)H,WW     " "2 & & &! %b:o9N&N&N%9%9#O4&&"(==O'PPO=%&89*(8' 1$  33G3G!G!G11//2RSS45GH HO 501  "$4$Do     1  33G3G!G!G OrKrrrmicrobatches_per_roundmultiply_factorcT|dz |z}|||dz |z zz}t|||zS)zN Calculate the number of warmup operations for interleaved schedules. r)r)rrrrrrwarmups_ops_last_stagers r?_get_warmup_opsr8 sD-q04JJ'/ma>OSW=W*XXJ z>N: ; ;;rKceZdZdZ ddeedededzdee d fdzd e e e fdzd e e e fee zdzd e d e de ffd ZdeedzfdZxZS)r%a 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. NTFrrrr.rrrrr}c $|dj|_t|||||||||  t ||_|dj|_td||jz|_ ||j z|_ ||j zdkrtd|j d|di|_ t|jD]!} || } | |j | <"||j dS)Nr rrrrrrrrr}rz_Interleaved 1F1B requires the number of microbatches to be a multiple of the number of rounds ( ), but got .)r"rrarrurr#rrnumber_of_roundsrr<rrrr rVrrrrrrrrr}rrrs r?rz ScheduleInterleaved1F1B.__init__] sL$AY1 )+//#'A'  "&kk1I( #A~9K'K L L&48M&M# D1 1Q 6 6-595J--)--- @B$,-- 1 1D==dCCH(0D  % % ))$*=>>>>>rKr9c 4tjjjjdjjz}|z }||z }|z|z}t d|||fd}fd}tjj||||S)Nrr=rank %s, warmup_ops %s, 1f1b %s, cooldown_ops %s total_ops %scB|jzjz}|jzzSrSrrrr# local_indexrrVs r?rzVScheduleInterleaved1F1B._calculate_single_rank_operations..forward_stage_index +4#>>$BUUK$"44< .backward_stage_index K#:%$*EE%&&   $"44< B?C@D +/#+?+?'(+?+?D +? 34t; +?  _ 45< +? S>E#J6=+?+?%)+?+?+?+?+?+?+?Z2 gn9M2 2 2 2 2 2 2 2 rKr%ceZdZdZ ddeedededzdee d fdzd e e e fdzd e e e fee zdzd e d e de ffd ZdeedzfdZdZxZS)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. NTFrrrr.rrrrr}c t||jj|dj|_t |||||||||  t||_|dj |_ td||jz|_ ||j z|_ ||j zdkrtd|j d|di|_t!|jD]!} || } | |j| <"||j|jz|_||jdS)NrrrzZZero bubble requires the number of microbatches to be a multiple of the number of rounds (rr)"_check_torch_compile_compatibilityrrAr"rrarrurr#rrrrr<rrr_add_bubbles_to_actionsrrs r?rz&ScheduleInterleavedZeroBubble.__init__ s +64>3JKKK#AY1 )+//#'A'  "&kk1I( #A~9K'K L L&48M&M# D1 1Q 6 6-595J--)--- @B$,-- 1 1D==dCCH(0D  % % #::  $"4 4   ))$*=>>>>>rKr9c > tjjjjd jjz}| z }||z } |z|z}t d |||fd} fd}}tjj |||||d S)NrrrcB|jzjz}|jzzSrSrrs r?rz\ScheduleInterleavedZeroBubble._calculate_single_rank_operations..forward_stage_index rrKc^jdz |z jzjzz }|jzzSrrrs r?rz]ScheduleInterleavedZeroBubble._calculate_single_rank_operations..backward_stage_index! rrKT)rr) rVrrrrrrrrrs `` @r?rz?ScheduleInterleavedZeroBubble._calculate_single_rank_operations s$     '        ,t/CC$z1 % 3 ,|;  K          = = = = = =  = = = = = = =!%!          !#    rKc|j}d}t}i}i}i}d}t|jD]} g|| <d|| <d|| < d} t} t|jD]#} || } | t || kr%d} || | || | } | t dt | | \}}}}||||||sR|| || | || |||f|| xxdz cc<|| d|| xxdz cc<|| xxdz cc<|| d%| | | rnc|dkrt d|||S)Nc|tjkr|dkr |dz ||f|vrdSn4|tjkr$||dz kr|tj|f|vS|dz||f|vSdS)NrrTF)r+rDrG)rr? microbatchnum_stages_globalseen_opss r? need_bubblezJScheduleInterleavedZeroBubble._add_bubbles_to_actions..need_bubble< s%---A::519b*"=X"M"M4'555-111!#3#;ZHPXXX 2z2(BB5rKrTFz)Expected temp_action to be not None, got rz?Non zero bubbles added: total_bubbles_added=%s bubbles_added=%s) rrrrrur rrorupdaterr))rVrrrrr next_pointer bubbles_addedtotal_bubbles_addedr should_stop temp_seen_ops timestamp temp_actionrNr?rrs r?rz5ScheduleInterleavedZeroBubble._add_bubbles_to_actions9 sj%   B?C@D +/#4?4?'(4?4?D 4? 34t; 4?  _ 45< 4? S>E#J6=4?4?%)4?4?4?4?4?4?4?l7 gn9M7 7 7 7 rCCCCCCCrKr'ceZdZdZ ddeedededzdee d fdzd e e e fdzd e e e fee zdzd e d e de ffd ZdeedzfdZxZS)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. NTFrrrr.rrrrr}c t||jj|dj|_t |||||||||  t|j|jd|_ |j D]} |j | _ t||_ |j dkrtd|j d|dj|_|dj|_i|_t%|jD]!} || } | |j| <"||jdS)Nrrrstyler0ZBV requires exactly 2 stages per rank, but got rrrrAr"rrarrrdr$r!rurr<r#rrcrrrrrVrrrrrrrrr}rrrrs r?rzScheduleZBVZeroBubble.__init__ s} +64>3JKKK#AY1 )+//#'A'  *H   0* * * &\ M ME.2.LE + +!&kk  ! # #*&***  1I(  ). @B$,-- 1 1D==dCCH(0D  % % ))$*=>>>>>rKr9cN tdjzdz j}dt|D}d\}}}}dj|z zdz }|} jdz |z } t|D]1} |t | t||dz }2|} t| D]`} |t | t||dz }|t | t||dz }aj|z } t| D]} |t | t||dz }|t | t||t | t||dz }||ks||kr$||kr/|t | t||dz }|t | t||t | t||dz }|t | t||dz }|t | t||t | t||dz }||k||k$||}}|}t|D]`} |t | t||dz }|t | t||dz }aj|z }t|D]`} |t | t||dz }|t | t||dz }a||kr5|t | t||dz }||k5||kr5|t | t||dz }||k5||kr||kstd|d|d|||kr||kstd |d |d |fd |D}|S) Nrrcg|]}dSrSrJrs r?r]zKScheduleZBVZeroBubble._calculate_single_rank_operations.. rrK)rrrr)rOrPz0Expected w0_cnt == b0_cnt == f0_cnt, got w0_cnt=z , b0_cnt=z , f0_cnt=z0Expected w1_cnt == b1_cnt == f1_cnt, got w1_cnt=z , b1_cnt=z , f1_cnt=cJg|]}||j|jjkr|nd SrS)rPr)r[r8rVs r?r]zKScheduleZBVZeroBubble._calculate_single_rank_operations..4 sT     %+7+d.BBB    rK) rrrrrcrorMr,r-r.r )rVrn_microrf0_cntf1_cntb0_cntb1_cnt warmup_n1stage_id_chunk0stage_id_chunk1r warmup_n2 warmup_n3w0_cntw1_cnt cooldown_n1 cooldown_n2s` r?rz7ScheduleZBVZeroBubble._calculate_single_rank_operations sa$,,q0$2FGG)D)Dd )D)D)D*4&+d23a7 /A-4y!!  A OO!fUUU    aKFF y!!  A OO!fUUU    aKF OO!fUUU    aKFF&- y!!  A OO!fUUU    aKF OO!fUUU    OO!fUUU    aKFFvoo'!1!1'!f !  OO!fUUU    OO!fUUU    aKF OO!fUUU    aKF OO!fUUU    OO!fUUU    aKF5voo'!1!18  {##  A OO!fUUU    aKF OO!fUUU    aKFF(4/ {##  A OO!fUUU    aKF OO!fUUU    aKFFvoo OO!fUUU    aKF voo voo OO!fUUU    aKF voo &  Vv%5%5 m6mmTZmmekmm &  Vv%5%5 m6mmTZmmekmm      #    rKrrrws@r?r(r( s'  &$(>B?C@D +/#3?3?'(3?3?D 3? 34t; 3?  _ 45< 3? S>E#J6=3?3?%)3?3?3?3?3?3?3?jzgn9MzzzzzzzzrKr(ceZdZdZ ddeedededzdee d fdzd e e e fdzd e e e fee zdzd e d e de ffd ZdeedzfdZxZS)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 NTFrrrr.rrrrr}c t||jj|dj|_t |||||||||  t|j|jd|_ |j D]} |j | _ t||_ |j dkrtd|j d||jkrtd|d |jd |dj|_|dj|_i|_t%|jD]!} || } | |j| <"||jdS) NrrrrrrrzDDualPipeV requires at least as many microbatches as stages, but got z microbatches and z stages.rrs r?rzScheduleDualPipeV.__init__I s +64>3JKKK#AY1 )+//#'A'  *H   0* * * &\ M ME.2.LE + +!&kk  ! # #*&***  D, , ,P!PP595EPPP  1I(  ). @B$,-- 1 1D==dCCH(0D  % % ))$*=>>>>>rKr9cg}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]1} | ||t| || ||t2||dzz |zd z}t |D]R}|dkr.||d z kr%| ||t| ||tn||||||||S||z d z }t |D]"} | ||t||||#|d z}d}t |D]d}||dzkr |dzd krd}|rtnt}| |||||dzkr |dzdkrd}|rtnt}| |||e||z d z }t |D]*} | ||rtnt}| |||+|d z}t |D] } | ||S)NrrrrNc|tf}|tf}|ddz|<|ddz|<dS)zbHelper method to increment BACKWARD_INPUT and BACKWARD_WEIGHT counters when FULL_BACKWARD is used.rrN)rErFr)rN input_key weight_keycounterss r?increment_backward_countszVScheduleDualPipeV._calculate_single_rank_operations..increment_backward_counts sW$n5I%7J"*,,y!"<"#>#BHZ rKr forward_stagebackward_stagecV|tf}|tf}|d}|d}t|t|t|t|f}|tdt d||dz|< |dS)zYHelper method to add an overlapped forward+backward action which tracks microbatch index.rrjNr)rDrErrMrGror6) rr!r" forward_key backward_keyrrrQrr s r?add_overlap_f_bzLScheduleDualPipeV._calculate_single_rank_operations..add_overlap_f_b s)'2K*N;L!k155J",,|Q77K w ;; {CCK NN72{D+FF G G G%/NH[ ! % %n 5 5 5 5 5rKrOc0|tkr||fn|tf}|d}|t ||||tkr |dS|tkr||f|dz|<dS)Nrr)rGrErrorM)rrNrOrrrr  weight_queues r? add_actionzGScheduleDualPipeV._calculate_single_rank_operations..add_action s$}44.//!>2   ||C++H NN7;0@(KK L L L =00))+66666$~55 ''h(?@@@ (1  rKcsdSd\}}|t|t||tf}|ddz|<dS)z4Helper method to add a weight action from the queue.Nrr)rrorMrFr)ractual_stage_indexrrrr(s r?add_weight_action_if_pendingzYScheduleDualPipeV._calculate_single_rank_operations..add_weight_action_if_pending s 2>2B2B12E2E /  NN&##   -o>J#+<< A#>#>#BHZ rKrr)r!r"FT) rrrrtrr+rrDrErG)rVrrr num_chunksrank_to_stages stage0_index stage1_indexr&r)r,step_1rstep_2step_3step_4rstep_5step_6 enable_zbrstep_7step_8rr r(s @@@r?rz3ScheduleDualPipeV._calculate_single_rank_operations s(*   & ) 7 y1}C   &4D%9" l C3 C C C C C C 6 6 6  6 6 6 6 6 6 6. - - -/ - - - - - - - -0 C$ C C C C C C C$d"Q&!+v 7 7A Jw g 6 6 6 6v 7 7A Jw g 6 6 6 Jw g 6 6 6 6T!A%v 7 7A Jw n = = = ( ( 1 1 1 Jw g 6 6 6 6i!m+d2Q6v  AAvv$)a-// 7L'::: 7L-@@@@".#/ O*+     T!A%v  A Jw m < < < O*+      v 9 9AFaKD1HMM *3FI Jw i 8 8 8FaKD1HMM *3FI Jw i 8 8 8 8T!A%v 9 9A ( ( 1 1 1*3FI Jw i 8 8 8 8v 2 2A ( ( 1 1 1 1rKrrrws@r?r)r)A s'$(>B?C@D +/#8?8?'(8?8?D 8? 34t; 8?  _ 45< 8? S>E#J6=8?8?%)8?8?8?8?8?8?8?t^gn9M^^^^^^^^rKr) schedule_namec @tttttt t ttd }d|D}| }||vr4td|dt| |||S)z Maps a schedule name (case insensitive) to its corresponding class object. Args: schedule_name (str): The name of the schedule. ) 1F1BInterleaved1F1BGPipe LoopedBFSInterleavedZeroBubbler!r" ZBVZeroBubble DualPipeVc8i|]}||SrJ)lower)r[rs r?rz&get_schedule_class..6s"999qaggii999rKzUnknown schedule name 'z'. The valid options are ) r#r%r$r&r'r!r"r(r)rDr<rkeys)r: schedule_maplowercase_keyslowercase_schedule_names r?r r $s2&!>"8!6.&  L:9L999N+1133n44 im i idS_SdSdSfSfNgNg i i    '>? @@rKc fdtDdtD d D dtdtdzf fd }dtdzdtf fd }rd }tD]m}t |d kr|d }||r,| ||||d d }a||dntd D]}t |d kr|=tD]}t |d kr |d+|d }||rC|&| |d< |||d td D]}t |d kr|=|sRtdt D]#}td|d|d $td͉ S)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. c6i|]}|d|DS)cg|]}||SrSrJ)r[rs r?r]z6_simulate_comms_compute...Ls@@@Q!-q---rKrJ)r[rrs r?rz+_simulate_comms_compute..Ks?  @@..@@@rKci|]}|gSrJrJrs r?rz+_simulate_comms_compute..Os'222b222rKc,i|]}|tSrJrrs r?rz+_simulate_comms_compute..Ss.Q.Q.QttSUU.Q.Q.QrKrr8Nc||||||dSdSrS)ror)rr8_prev_ops_rank _schedules r?add_to_schedulez0_simulate_comms_compute..add_to_scheduleUsH$v&&&   4 $ $V , , , , ,  rKr9c|dS|j} |}|jtkrZ|jdkrdSt|jt|j|vrdSt|jdz t|j|vrdSdS|jt tfvr|j dz krdSt|jt|j|vrdSt|jdzt |j|vrdSt|jdzt|j|vrdSdS|jtkrdS|jtkr$t|jt|j}||vS|jtkr3|dz }t|t|j}| |vS|jtkrHt|jt |j}t|jt|j}||vp||vS|jtkr3|dz}t|t|j}| |vStd|)NTrrFzUnsupported action type ) rNrOr,rMr2rPrErGr4rFr1r3r<) r8rprev_ops expected_fpeer_stage_idx expected_send expected_b expected_bwrOrcrs r?rz3_simulate_comms_compute.._ready_to_scheduleZs >4& !-- ":":;  "a ' '!Q&&t*FF4KLLPXXXt*Q.63JKKxWWt5  $(G G G!Z!^33tv)663JKKxWWt*Q.@WXXt*Q. v?VWWt5  $ 7 74  $ . . !3Q8OPPJ) )  $ . .&]N#NFF**N22#N33222I/R.Qy.Q.Q.QN-c-7T>------- 8B7T>8Bd8B8B8B8B8B8B8B8Bt +<>** , ,D>$'((A--#D)!,F!!&)) ,%#OD&111t$((+++d++++555 & &A>!$%%**"1%>** , ,D>$'((A--r".#D)!,F!!&)) ,%*0IdOB'"4(,,V444t$((+++555 & &A>!$%%**"1% < )+A)+L+L M M M& I IGGGnT.B1.EGGHHHH:;; ;W +}t|t r'td|dt|j ?dS)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) rrrnnModulerrr=rrA)rr:rmodules r?rrs  %,88  l**,,  F&/22 "H=HH04V 0EHH    rKrS)r)rF)r)srrHrloggingreabcrr collectionsrrcollections.abcr dataclassesrenumr functoolsr typingr r r rrrtorch.distributed distributedrL torch._dynamortorch.distributed.fsdprrtorch.nn.modules.lossrtorch.profilerr_utilsrrrrrrrrrrrr__all__ getLoggerrArrIr+rDrErFr/r0r1r2r3r4rGr6r7rqr,r-r.r5compilerqrMrr:rtrrrP2POpWorkrrYrr!ryr$r#r|rrrrrrrr"rsrvr{r&rrr%r'r(r)r rrlrrJrKr?rs$   ########,,,,,,,,$$$$$$!!!!!!;;;;;;;;;;;;;; ))))))<<<<<<<<''''''******  @?????????     8 $ $NNNNNsDNNN.  "!0"2  "  "         . * * %_Q''  R T T T T T jT T T n C  %)<<d7T>223<Tz< <<<<~d d d d d d d d N ++TZ(+d +d49o++++:37 $* %(4Z #tDI 6$ty/BBBBB.BBBJ2"2"2"2"2"12"2"2"jxxxxx*xxxvzzzzz)zzzz!'7!D!!!!$ 'D. !$36$ 'D.$$$$FPP'D.)PP ']PPPPf#'D.)# ']####Lt#tG},-tSE3J'tt #tG}  ttttn[ #tG}$ %[SE3J'[ #tG} [[[[|' #tGdN++ ,''' '  #s(^ ''''DA2A2A2A2A2-A2A2A2H  LLLLLhLLL}2}2}2}2}24}2}2}2@DDDDD0DDD`FFFF^ << << < <  <  < <<<<&q q q q q 6q q q h}}}}}$<}}}@4D`````0```FAcAAAA6B#+SE3J#7BEHBBBBJ!.!.!.H # $58rK