g^(h~ddlmZddlZddlZddlZddlZddlmZddlm Z m Z ddl Z ddl m Z ddlmZddlmZmZdd lmZdd lmZmZmZmZmZmZmZej:eZe j@jCed Z"e rdd l#m$Z$dd Z%ddZ& ddZ' ddZ( ddZ)ddZ*dZ+dZ, ddZ-ddZ.ddZ/dZ0 d dZ1y)!) annotationsN) defaultdict)Any TYPE_CHECKING)StorageWeakRef) OrderedSet)configir)WeakDep)contains_collective contains_waitfind_recursive_deps_of_nodefind_recursive_users_of_node is_collectiveis_fallback_opis_waitoverlap)BaseSchedulerNodec t|dddS)z7 Greedily schedules waits as late as possible. FT raise_comms sink_waitsreorder_for_overlap_schedule_for_commsnodess S/var/www/skyplay_api_hub/venv/lib/python3.12/site-packages/torch/_inductor/comms.pyrr$s Ed c t|dddS)z8 Greedily schedules comms as early as possible. TFrrrs rrr-s DU r c t|dddS)a This achieves the following overall scheduling procedure: Step 1: Given that we've currently scheduled comm N, we now schedule all compute nodes that are required for comm N + 1 but do not depend on comm N, to run at the same time with comm N. Step 2: If all those compute nodes are sufficient to overlap comm N, we're done. Otherwise, we now need to look elsewhere to find compute that overlaps with comm N. We prioritize compute nodes that are needed sooner. Step 3: We schedule the compute nodes dependent on comm N and required for comm N + 1. Step 4: We schedule comm N + 1. Repeat this for subsequent comm nodes. Trrrs rreorder_compute_for_overlapr#6s DTt r c i}iiiict|D]y\}}|jD]}|||< |jD]}||< ||j<|j} tj | <d| <|| <{d} |D]}|rZt |rO| |j<|jD]'} | j} t| | | <)| dz } _|sbt|snd|j<Gfdd|Dcic]}|td|jD!c}gtt|Dcic]}|t|c}jD]J\}} t| dk(rt!j"|| D]}|j%|Lgfdfdfd}trIt!j&j(}|rt |r ||n|trIjD]\}} t| dk(rJd Scc}wcc}w) a Schedule `snodes` for various comm optimization objectives. Args: snodes: the nodes to be scheduled. raise_comms: whether to greedily schedule collectives as early as possible sink_wait: whether to greedily schedule waits as late as possible reorder_compute_for_overlap: whether to reorder compute nodes to optimize for compute/communication overlapping. Returns: The new schedule order. Some notes on the synergy between different options: - `raise_comms` provides more overlapping oppurtunies for `reorder_compute_for_overlap`. - When both `raise_comms` and `sink_waits` is `True`, `raise_comms` is prioritized. rr c&eZdZdfd ZdZy)$_schedule_for_comm..Runnablec||_tt|j}|j }|||f|_yN)snodenextiterget_operation_namesget_namescore)selfr)name fused_namename_to_fused_nodescores_0scores_1scores_2s r__init__z-_schedule_for_comm..Runnable.__init__sVDJU6689:D+D1::.Runnable.__lt__s:: + +r N)returnNone)__name__ __module__ __qualname__r6r:)r2r3r4r5srRunnabler&s   ,r r@c34K|]}|jywr()r0).0deps r z%_schedule_for_comm..sGs#((Gscj||jD]N}|D]D}|j|t|dk(s)t j |FPy)zU Schedules `snode` and put all unblocked nodes onto the ready queue. rN)appendget_buffer_namesremovelenheapqheappush)r)buf_namer@ buffer_usersready scheduled unmet_depss rschedulez$_schedule_for_comm..schedulesv ..0 ;H%h/ ;5!((2z%()Q.NN5(5/: ; ;r cDcgc].}t|jst|js|0}}t|dk(ryt |dScc}w)zh Return the next node in the ready queue that's neither a collective or a wait. rNc|jSr(r8xs rzG_schedule_for_comm..get_overlapping_candidate..s QWWr key)r r)rrImin)rU candidatesrNs rget_overlapping_candidatez5_schedule_for_comm..get_overlapping_candidates] &qww/ agg8N   z?a :#455 s3Act|sJ||}|dkDrMx}Dj||j||jz}|dkDr x}Dtjy)z Schedules collective node `snode`, along with one or more compute nodes to overlap with it. The strategy is described in the comment of `reorder_compute_for_overlap`. rN)r rHr)rJheapify)r)collective_cost candidater[rNrQ snode_to_costs rschedule_collective_for_overlapz;_schedule_for_comm..schedule_collective_for_overlaps #5)))'. a 799F LL # Y__ % }Y__= =O a 799F  er z;Detected unscheduled nodes. Nodes with unmet dependencies: ) enumeraterGr,r-sysmaxsizer ancestorsrYrrunmet_dependenciesrestimate_op_runtimeitemsrIrJrKaddheappopr))rrrrbuf_name_to_snodeidxr)rLop_name node_namecomm_idxancanc_fused_namedepsrCrar@rMr[r2rNrQrOr3r4r5r`rPs @@@@@@@@@@@@rrrIsL#%r2 Hh' " U..0 0H*/ h ' 0002 0G*/ w ' 0/45>>+,NN$ !kk! "H+ .u5)1HU^^% & S!3C!8!A!A!C+.x/G+R( S MH M%0)*HU^^% &+ , , <  zGe.F.FGGG<J E=H=TLDJK5U/66KM!'')) t t9> NN5(5/ 2 )C   ! !% ( )) I ; ; 6& e* e$** #6u#= +E 2 UO e*"'') t4yA~ I* V ~  Q<Ls 9$J5J cxtjjs|S|Dcgc]}t|s|}}t dt |D]a}t t||j}||dz jD]!}||jt||#c|Scc}w)z Decide global ordering of comms, by just enforcing the ordering that's in the input graph (might not be the same ordering as the eager mode program). TODO: Come up with a better approach r  mutating_buf) torch distributed is_availabler rangerIr*r+rG add_fake_depr )nodes name_to_bufr2n comm_nodesirubufs rdecide_global_ordering_of_commsrs    ) ) + "=&9!&s B7B7ctjdk(r|j}|SttjsJtj|}|S)z: Returns estimated op runtime in nanoseconds (ns) default)r rgget_estimated_runtimecallable)r)runtimes rrgrgsR!!Y.--/ N22333,,U3 Nr cd}t|jtjrd|jjd}d}|jj }t|tj rd|jd|jd}|jjxsd}|jjj||d|dS)Nz ()z (size=z , stride=) isinstancenoder ExternKernelOutpython_kernel_nameget_output_specLayoutsizestridemaybe_get_name __class__r=)r)detailout_tensor_infolayoutrns r node_summaryrs F%**b001ejj334A6O ZZ ' ' )F&"))$#FKK= &--J ))+1rIjj""++ ,VH_4ER {RS TTr cd}d}|D]}|tt|r|t|z }|j}n.t|jr t d|t|z }t j t|yt|r t dt|jr"t j t|d}t j dt|t j d|dz dz y)Ngz8Wait is not expected when there is no collective runningzkFound two collectives running at the same time. `visualize_overlap` needs to be updated to handle this casez| zEst. runtime (ms): i)r rgrrAssertionError overlap_logdebugr)ordertotal_est_runtime cur_comm_noder)s rvisualize_overlaprs"M>  "5)!%8%??! % $$N"%8%??!   e!4 5 7"5)$R$!!\%%8$9; $ !!B|E':&;"<=->. /$6=>?r c|}tjD]}t|tr|t vr t |}t j jdk(r%tjd|d t|||}t j jdk(stjd|d t||S#t$r(}tjt|Yd}~d}~wwxYw#t$r)}tjt|Yd}~&d}~wwxYw)Nrz.==== Visualize overlap before reordering pass z ====z-==== Visualize overlap after reordering pass ) r 'reorder_for_compute_comm_overlap_passesrstrglobalsrvrwget_rankrrr Exception)rrpes r$reorder_compute_and_comm_for_overlapr0s E  ; ;* a !wy. ! A    % % '1 ,   @5I  *!%(%    % % '1 ,   ?s%H  *!%(#*( L *!!#a&)) * *!!#a&)) *s0: C D  D !DD  D>D99D>c t|jtttttD]\}}|jdk(s|j t jjjjk(sR|jdjdk(sJd|d|jdd|jd}|jd}|dkDr|j||j|fd}tt}tD]\}}|jdk(s|j t jjjjk(sR|}|jdjdk(sJd d |d|s|j|d }d D]}|jdk(st|j t j j"sB|j j$j&sc||rl||j)sJd |d|j+D]!\} t| D] \} } | }|jdusJ|j\} | dz} | t-| dz kr| | dzn t-dz }| |}t/fd|DrJdd|d|d|D]}|jdk(s|jvs"|j t jjjjk7s^t1fd|jD}||_ $|j+D].\} t| D]\} } | }|j3|0D]q}|jdk(s|j t jjjjk(sO|jd|vsa|j3|sy)a This FX graph pass replaces uses of FSDP2 unsharded params with their corresponding graph intermediates that were fsdp.copy_ into the unsharded params in the original graph. NOTE: Can only apply this pass to any of the FSDP2 unsharded params that have this pattern (or repetition of): `resize_(full) -> copy_ -> resize_(0)`. Because of this, for partial-graph case where `resize_(full) -> copy_` is in one graph and `resize_(0)` is in another graph, we can't remove these resize and copy ops and thus we will have worse performance there. In other words, "do we try to remove all the resize_(full) -> copy_ -> resize_(0) nodes for this unsharded param" is actually a per-unsharded-param decision, since for each unsharded param, we look at its resize sequence pattern (in `check_resize_pattern()`) to determine if its set of resize and copy nodes can be removed. call_functionr placeholderz1Resize can only operate on graph inputs, but got z# which is resizing non-graph-input  r clj|g}j|g}t|t|k(s2tjd|dt|dt|dyt ||D]7\}}||k\s tjd|d|d|d |d|d yy ) NzH Unequal number of resize-to-full and resize-to-0 nodes for graph input z: z vs. zK. Skipping `remove_fsdp2_unsharded_param_graph_input_usage` FX graph pass. Fz For graph input z: resize-to-full node z at index z happens after resize-to-0 node zd. Skipping `remove_fsdp2_unsharded_param_graph_input_usage` FX graph pass for that unsharded param. T)getrIlogwarningzip) graph_inputresized_to_full_idxesresized_to_0_idxesresize_to_full_idxresize_to_0_idx&graph_input_to_resized_to_0_node_idxes)graph_input_to_resized_to_full_node_idxes node_lists rcheck_resize_patternzLremove_fsdp2_unsharded_param_graph_input_usage..check_resize_patternns !J M M ! DGG UWX()S1C-DD KKHHS}UE#&8"9!:; 47 !#54  / "_4  3I>P4Q3RR\]o\pq )/ :;:oEVW r z\ Assumed all FSDP2 `unsharded_param`s to be graph input, but it's not true! Offending node: z . Graph: c|jtjjjj k(xs;|jtjj jj k(Sr()targetrvopsfsdpcopy_rinductorresize_storage_bytes_)rs ris_allowed_mutationzKremove_fsdp2_unsharded_param_graph_input_usage..is_allowed_mutationsO KK599>>//77 7 O{{eii00FFNNN r c nt|jtjjr^t |jj jDcgc])\}}|j|jjr|+c}}ng}t|Dcgc]5}t|j|jdj7c}}t|Dcgc](}t|jdj*c}}t||zdkDScc}}wcc}wcc}w)Nvalr)rrrv_ops OpOverloadrb_schema arguments alias_infois_writerrargsmetauntyped_storagerI)runsharded_paramsrrUmutated_arg_idxesmutated_node_arg_storagesunsharded_paramstorages_of_unsharded_paramss r-is_node_mutating_unsharded_param_or_its_aliaszeremove_fsdp2_unsharded_param_graph_input_usage..is_node_mutating_unsharded_param_or_its_aliass$++uzz'<'<=&dkk&9&9&C&CD Aq<<+ 0E0E   %/+ tyy|007GGIJ % ! (2(8 #33E:JJLM ( $ ,/KKLqPP)    s.D':D-"-D2zdUser mutation on FSDP2 unsharded param is not allowed when Traceable FSDP2 is used. Violating node: c32K|]}|gywr()rBrrrs rrDzAremove_fsdp2_unsharded_param_graph_input_usage..s#>d_DUVsz(Assumed no ops mutating unsharded param z in subgraph z, but it's not true! Graph: c3.K|] }|urn|ywr(r)rBarg replacementrs rrDzAremove_fsdp2_unsharded_param_graph_input_usage..s%%(+o'= 3F%sN)listr{rrboprrvrrrrrrFrrrrrr is_mutablekeysrhrIanytuple erase_node)graphrlrrnew_sizer'unsharded_param_to_fsdp_copy_node_idxesfsdp_copy_noderfsdp_copy_node_idxesrfsdp_copy_node_idx_subgraph_start_idxsubgraph_end_idxsubgraph_nodesnew_argsrrrrrrs @@@@@@r.remove_fsdp2_unsharded_param_graph_input_usagerLsU[[!I1*y) P T GG & uyy11GGOOO99Q ##..'-D=BBD eeidjk  : 1 6 6 8") %./C%D ) !A!&'9:N!&&q)_< <<+00NA{!3a!7 s/0144%QU+^a'  ''9:JKN* ))8(9~FVW w ' )GG.'4994 uyy'9'9'O'O'W'WW$%#'99% H!)DI )) ) ")P 1 6 6 8- %./C%D - !A!&'9:N   ^ , - -# GG & uyy11GGOOO ! GG   T " #r c ddl jjsJ jjj r jjj sJ ddl m }m }m }m}m} fd}|}|| jjj j |t"j$| jj&j(j |d|d|d|d|d |d |d |d |d |d|dd fd }|||j+|y#tttf$rYywxYw)Nrr ) CallFunction KeywordArgMatchPatternMatcherPassregister_graph_patterncJt|j}|D]}|jtjk(s!|j djj jjjuse|j ddk(sx|j|y)Nrr ) rr{roperatorgetitemrrrall_gather_copy_inrr)grr}rvs rremove_unused_getitemz8reinplace_fsdp_all_gather..remove_unused_getitem5suM  AH,,,FF1I$$ (I(I(Q(QQFF1IN Q  r all_gather_inputsinp_split_sizesall_gather_input_numel world_sizerankdtypedeviceitem_idx group_size group_namec&|jddk(S)Nrr)kwargs)matchs rrVz+reinplace_fsdp_all_gather..Ws%,,z":a"?r ) pass_dict extra_checkc |fd}|j||d|d|d|d|d|d|d|d |d g y) Nc|dd}|d}|d}jjjj|}|d}|d}jjj j||||}|S)Nrr )out)rrrr_c10d_functionalall_gather_into_tensor_out) r copy_in_argsrrrr getitem_1all_gather_into_tensorrvs rreplzEreinplace_fsdp_all_gather..reinplace_all_gather..replZs 9LbJbJ!J!B!B!J!J" )+G*1-I **EEMMZN # * )r rrrrrrrrr)replace_by_example)rrrrrvs rreinplace_all_gatherz7reinplace_fsdp_all_gather..reinplace_all_gatherBsi0 *$   *+()/0|$vwx |$|$  r )rr)5torch.distributed.fsdp._fully_shard._fsdp_collectivesrwrxrr rr  ImportErrorAttributeErrorrpattern_matcherrrrrrrrrrrapply) rrrrrrr graph_passrrvs @rreinplace_fsdp_all_gatherrss D  --/// II & & = = **EE  FE   $%J II & & = = E E   IINN55==230178|,v&w'x( :&  | $ | $# &?+.  /.  D% UE  8sA"E((E?>E?ct|tjjjtjjj frJt |jddS)N)rrv _inductor schedulerFusedSchedulerNodeGroupedSchedulerNodeintr-)r)s r get_op_idxr"s] OO % % 8 8 OO % % : :   u~~# $$r c z  !ddlm g}tt}d}d}i}i}i! !fd} |D]} t | j t jjjjrtfd| jDrd}| } t} t| | |tt jjjjt jjjjt jjj jgt#| | | fd t%| d } t'| }d }t)t'| D]W}| |}t+|j t jjj jr|dz }|dkDsU|}n| d|} d}t)t'| dz D]3}t-| |dzj t.j0s.|dz}n|J| | d|}| | |d}|||<'t+| j t jjj2jskd}| }t}t#|||t%|d }d}t)t'|dz D]3}t-||dzj t.j0s.|dz}n|J| |d|}| ||d}|||<t'!d kDsJ|rt'|d kDsJ|rt'|d kDsJ|D]N} | j5!vr!| j5} | |vr-|j7| |j9| Pd}|j;D]j\}}|at=t?|jA}|jCD],}|jEtG|j5|.|}ld}|j;D]j\}}|at=t?|jA}|jCD],}|jEtG|j5|.|}l|S)Nr )rFcjj|}|D]}||j<||j<|Sr()r creater-)snodes_to_group group_noder)rsnode_name_to_final_snodes r_create_group_nodez:enforce_comm_ordering_for_fsdp.._create_group_nodesV33::?K $ EE:D %enn&6 7 E;E!*"5"5"78r )rc3K|]I}t|jtjjj j Kywr()rrrvrrrr)rBrUr2s rrDz1enforce_comm_ordering_for_fsdp..sD  "1%**EIINN,M,M,U,U  sAATct|jxs0t|jxr|jjv Sr()rNopKernelSchedulerNodeExternKernelSchedulerNoder op_overload)rU allowed_opsrs rrVz0enforce_comm_ordering_for_fsdp..sFq)"B"BC"1i&I&IJ>FF..+= 'r ) criteria_cbct|Sr(r"rTs rrVz0enforce_comm_ordering_for_fsdp.. JqMr rWrct|Sr(r2rTs rrVz0enforce_comm_ordering_for_fsdp..r3r rt)$rrrrrrrvrr r rrrer wait_tensorrsplit_with_sizes_copyrsortedrIryrrr _WaitKernel chunk_catr-rFrirhr*r+rG get_outputsrzr )"rr|r2 new_orderrO ag_exists rs_exists$ag_grouped_node_to_wait_grouped_node$rs_grouped_node_to_wait_grouped_noder)r)ag_snodeag_related_snode_setag_related_snodesend_idx_of_current_ag_blockcopy_out_countr cur_snode wait_node_idx ag_group_nodeag_wait_group_noders_snoders_related_snode_setrs_related_snodes rs_group_noders_wait_group_node prev_ag_waitwait_group_noderuo prev_rs_waitr/rr(s" ` @@@renforce_comm_ordering_for_fsdprRs )+I3!III+-(+-( "nU  JJ59955PPXX  __   IHLVL  ($"  %II..IIQQII..::BBIINN88@@K )$" !'$*A!  +..?*@ 'N3012 -a0 !NNEIINN$H$H$P$P#a'N!A%23/ !22N3N O !M301A56 /A6;;R^^L$%EM !, ,,./@-/PQM"44Emn4U!V BT 0 ?EJJ (@(@(H(H IIHMWL ($"  !'$*A!  !M301A56 /A6;;R^^L$%EM !, ,,./@-/PQM"44Emn4U!V BT 0 ?]nU` ( )A -- -781<<<781<<< >> 8 8-enn.>?E I   e L*N*T*T*V'&   #]%C%C%E FGL!--/ **AJJL|D ' 'L*N*T*T*V'&   #]%C%C%E FGL!--/ **AJJL|D ' ' r )rlist[BaseSchedulerNode]r;rS) rrSrboolrrTrrTr;rS)r{rSr;rS)r)rr;float)rtorch.fx.Graph)rrVr;r<)r1list[torch._inductor.scheduler.BaseSchedulerNode]r|z4dict[str, torch._inductor.scheduler.SchedulerBuffer]r2zdict[str, BaseSchedulerNode]r;rW)2 __future__rrJloggingrrc collectionsrtypingrrrv torch.multiprocessing.reductionsrtorch.utils._ordered_setrrr r dependenciesr utilsr rrrrrr getLoggerr=r_logginggetArtifactLoggerrrrrrr#rrrgrrrrrr"rRrr rrcs>#  #% ;/!g!nn..xC , #&W #WWW W  Wt ",  U> #8A#Hl^%n =nEn5n7 nr