iddlmZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZmZmZmZddlZddlmZddlmZddlmZd d lmZmZmZd d lmZerd d lmZm Z d d l!m"Z"m#Z#m$Z$m%Z%m&Z&d dl'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.d dl/m0Z0ejbe2Z3ejhjke2dZ6erddl7m8Z8 dCdZ9dDdZ:dDdZ; dDdZ< dDdZ=e GddZ>dEdZ?dFdZ@dGdZA dHdZB dIdZCdJdKdZDdZEdLd ZFd!ZG dMd"ZH dNd#ZId$ZJd%ZKd&ZL dOd'ZM dPd(ZN dQd)ZO dRd*ZP dSd+ZQ dTd,ZR dUd-ZS dVd.ZT dWd/ZU dXd0ZVe Gd1d2ZW dTd3ZX dYd4ZY dZd5ZZ dPd6Z[ d[d7Z\ dSd8Z] d\d9Z^dDd:Z_d]d;Z`d<Zad=Zb dDd>Zcd^d?Zdd_d@ZedAZf d`dBZgy)a) annotationsN) defaultdict) dataclass)AnyOptional TYPE_CHECKINGUnion)trace_structured)StorageWeakRef) OrderedSet)config config_commsir)WeakDep)IRNode Operation)estimate_peak_memoryestimate_peak_memory_allocfreeFreeableInputBufferget_freeable_input_buf SNodeMemory)contains_collective contains_waitfind_recursive_deps_of_nodefind_recursive_users_of_node is_collectiveis_fallback_opis_wait)Voverlap)BaseSchedulerNodeci}|D]}|j||<ddlm}ddlm}|j }|}t |Dcgc]}g}}|j|t|j|tjtj|djj} t t|D]} | | || _ycc}w)Nr)_get_default_group)dim)get_estimated_runtimetorch.distributed distributed"torch.distributed.distributed_c10dr$get_world_sizerangeall_gather_objectlistvaluestorchmediantensortolistlenoverride_estimated_runtime) snodesruntime_estimationssnodedistr$ world_sizepg_gathered_runtime_estimationsmedian_runtime_estimationsis O/var/www/html/engine/venv/lib/python3.12/site-packages/torch/_inductor/comms.py6align_runtime_estimations_across_all_distributed_ranksr@6sC%*%@%@%BE"C$E$$&J  BCHCT6Uar6U 6U$d+>+E+E+G&H""' 12" fVVX3v; M/I!/Lq ,M7Vs C+c t|dddS)z7 Greedily schedules waits as late as possible. FT raise_comms sink_waitsreorder_for_overlap_schedule_for_commr5s r?rDrDLs Ed c t|dddS)z8 Greedily schedules comms as early as possible. TFrBrFrHs r?rCrCUs DU rIc 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. TrBrFrHs r?reorder_compute_for_overlaprL^s DTt rIc"t|\}}|S)a Reorders communication ops relative to computation ops to improve communication-compute overlapping and hide comm latency. Stops moving a particular op if it reaches a point that would have increased the peak memory footprint. Currently, follows these heuristics (subject to change or tune): - never reorders collectives relative to one another, for SPMD safety - has an option for per-collective prefetch limit, but does not enable it by default - limits the total number of reorder steps to some factor of the graph size to prevent worst-case quadratic performance Prerequisite: sink_comms_and_waits - ensure comm and wait nodes are scheduled as late as possible, respecting data dependencies. That allows reorder_communication_preserving_peak_memory to take a best case peak-memory snapshot, and then monotonically improve latency by moving collectives backward in time. Peak memory impact is computed in an iterative fashion. First, memory use at each timestep is computed, and global peak memory is computed as a max over timesteps. Then, when swapping any two adjacent nodes, only the curr-memory for the earlier of the nodes after the swap is affected. This enables checking step by step whether a swap is peak-memory-safe, and bailing out if not. Example: 0 n0 C0 1 n1 C0 + Allocs(n1) - Frees(n1) 2 n2 C0 + Allocs(n1) - Frees(n1) + Allocs(n2) - Frees(n2) 0 n0 C0 1 n2 C0 + Allocs(n2) - Frees(n2) <-- After moving n2 to Time 1, only time1 memory changes 2 n1 C0 + Allocs(n2) - Frees(n2) + Allocs(n1) - Frees(n1) )6_reorder_communication_preserving_peak_memory_internal)r5reordered_snodes node_statss r?,reorder_communication_preserving_peak_memoryrQqs@ ?vF!j rIceZdZUdZdZded<dZded<dZded<d Zded <d Z d ed <d Z d ed<d Z d ed<d Z d ed<dZ ded<edZy) ReorderInfozE Debug info describing how an individual snode was reordered Nonestrlimiting_factorrintmovesgrouped grouped_infofloat comm_time comp_timeinitial_exposed final_exposed overlap_infoc4|j|jz SNr`raselfs r? improvementzReorderInfo.improvement##d&8&888rIN)__name__ __module__ __qualname____doc__rV__annotations__rXrYr[r^r_r`rarbpropertyrhrIr?rSrSst"OS!E3NGSL#IuIu!OU!M5L# 99rIrSc|yt|tjjjj ryt |ddx}rd|vryy)NFTpython_kernel_nameextern_kernels)rr/opsaten#_scaled_dot_product_flash_attentiondefaultgetattr)noderrs r? is_gemm_likerzsX |  ::BB&d,@$GGG 0 0 rIcddlm}t||rtd|jDSt |j S)NrGroupedSchedulerNodec32K|]}t|ywrd)contains_gemm_like).0xs r? z%contains_gemm_like..s?Q%a(?s)torch._inductor.schedulerr} isinstanceanyr5rzry)r7r}s r?rrs4>%-.?%,,???EJJ''rIcddlm}t||r&|jr|jD] }|| y||y)Nrr|)rr}r temp_groupingr5)r7fnr}_snodes r?_temp_group_visit_leavesrs;>%-.53F3Fll F vJ  5 rIc |d}t|jsJt|dkDsJt|dz }d}d d}g}t|ddD]}||}t |r|j |t |rNt||r|}|d|jdz }nZt|sd g|D]} t|| sd  } d fd } t||  } |d |jd | | z dz }| |fS) a Calculate exposed communication time for a wait operation by finding its corresponding collective and accumulating overlapping compute time between them. The Wait node must be the last in snodes_to_wait. Compute time between corresponding Collective and Wait is accumulated. If there is another pair of Collective and Wait inside, Only compute before first such Wait' is considered as overlapping. Multiple process groups are not modeled so far. r rZz->C[]c|z yrdrprr_runtimess r?accumulate_timez8wait_exposed_communication_time..accumulate_time &) )IrI+[rr"returnrT) rryr3r+rappendr is_corresponding_collective_waitget_namecontains_async_collectiver)snodes_to_waitr wait_snodeidxr^rb waits_foundr>cwcomp_time_beforercomp_time_afterr_s ` @r?wait_exposed_communication_timersO #J :?? ## # ~  "" " n  !CIILK 3B Q 1      q ! q !/:>$QK $qzz|nA 66 ,Q/ $!A71=%(  !% * !O4#!AJJL>?=M+M*NaPP ?QB i --rIc|d}|}dtd|jD}d}g}|ddD]}td|jD}||zr|d|jd z }nt |rt |snp|j |gt|rd } |D]} t||sd } n| rn6} dfd } t|| } |d |jd| | z d z }||fS)a Calculate exposed communication time for a collective operation by finding its corresponding wait and accumulating compute time that can overlap with communication. The Collective node must be the first in snodes. Compute time between corresponding Collective and Wait is accumulated. If there is another pair of Collective and Wait inside, Only compute before first such Wait' is considered as overlapping. Multiple process groups are not modeled so far. rrc3<K|]}|jywrdrros r?rz2coll_exposed_communication_time..s2 2rZr Nc3LK|]}t|r|jywrd) _is_fake_depnamerds r?rz2coll_exposed_communication_time..$s LOAFF s$$z->W[rFTc|z yrdrprs r?rz8coll_exposed_communication_time..accumulate_time>rrIrrr) r get_outputsunmet_dependenciesrrrrrrr)r5rcollective_snoder^collective_outsrbcollectives_foundr7 unmet_depshas_wait_for_collectives_found_collrrrr_s ` @r?coll_exposed_communication_timer s]ay)*II'12.::<2(OL13#U  !44    ' d5>>#3"4A6 6L  u %,U3!((/  -2 ** 34DeL592 .$ * !8#!ENN,-QAQ/Q.RRSTT G#UH i --rIcd}|jD]<}|r|dz }||jz }|s |t|jz }>|S)NrZr;)r5rr-get_buffer_names)r7 with_bufsretns r? _group_namerHs[ C \\7  3JC qzz|  d511345 6C 7 JrIc>t|txr |jSrd)rris_fake)rs r?rrSs a ! /aii/rIcfdj|Dcgc]}|jc}Scc}w)N~)joinr)gnsgns r? _group_namesrWs% 88S1rR[[]1 221s.ct||}t|||\}}}}tt||}d|d<|||||fS)z*Initialize memory tracking data structures)rrN)rrdictzip) r5 graph_inputs graph_outputsname_to_freeable_input_buf peak_memorysnodes_curr_memorysnodes_allocfreebuf_to_snode_last_use _curr_memorys r?_initialize_memory_trackingr[sg!7 !M& .  MK#%57L F$678LL"  rIci}i}t|D]5\}}|dkDr||dz nd||<|t|dz kr||dznd||<7|d}|||fS)z/Create double-linked list structure from snodesrr N) enumerater3)r5_prev_nextr>r7_heads r?_initialize_double_linked_listrns{ E Ef%F5()Ava!e}4e ()CK!O(;va!e}e F 1IE % rIctd|jD}td|jD}t||zS)z Check if a wait node corresponds to a given collective node by verifying if the wait depends on outputs from the collective. c3<K|]}|jywrdrrs r?rz3is_corresponding_collective_wait..s V! Vrc34K|]}|jywrdrrs r?rz3is_corresponding_collective_wait..sJqAFFJ)r rrbool)rrrrs r?rrsD! V7G7S7S7U VVOJJ,I,IJJJ  _, --rIcXt|rtjStjSrd)rr*reorder_sink_runtime_estimations_comm_mult.reorder_sink_runtime_estimations_non_comm_multr7s r?_op_runtime_estimate_multrs$5!FFF  F FFrIc@t|jddx}rd|vryy)z Filtering out ops that contain Collective and Wait inside and considered as Collectives. See contains_collective function. If the op contains Wait inside - consider as Synchronous compute. rrNz-torch.ops._dtensor.shard_dim_alltoall.defaultFT)rxry)r7rrs r?is_async_collectivers- %UZZ1EtLLL :>P P rIc"t|tSrd)rrrs r?rrs u&9 ::rIcNg}|} ||j|||k(r |S||}!)aA Traverse doubly-linked list from head to tail and return nodes as a list. Args: head: Starting node of the segment tail: Ending node of the segment (inclusive) next_dict: Dictionary mapping each node to its next node Returns: List of nodes from head to tail (inclusive) )r)headtail next_dictrrs r?_group_nodes_from_linked_listrs@ C A  = JJqM 9  J aL rIcl||}|r|||<|||<||}|r|||<|||<|||<|||<||k(r|S|S)a Swap positions of candidate and group in doubly-linked list. Transforms: candidate_prev -> candidate -> group_head...group_tail -> group_tail_next Into: candidate_prev -> group_head...group_tail -> candidate -> group_tail_next Args: candidate: Node to swap with group group_head: First node of group group_tail: Last node of group prev_dict: Dictionary mapping nodes to their previous nodes next_dict: Dictionary mapping nodes to their next nodes head: Current head of the linked list Returns: New head of the linked list (may change if candidate was the head) rp) candidate group_head group_tail prev_dictrrcandidate_prevgroup_tail_nexts r? _perform_double_linked_list_swaprss8y)N$. .!*Ij +O%. /"*Ii&Ii%Ij y KrIcni}d} |s)t||z ||d|z |jz} | |fS| } |D]R} || d| z} | || <t| | } |j| } | 3| D]}| |jjz } T||d| z|jz}|||<t| |} | |fS)a Calculate potential peak memory after swapping candidate with group (reorder version). Computes new memory levels for all affected nodes and returns the potential peak memory along with cached post-allocation memory values for each node. Args: candidate: Node being moved gns: Group nodes group_tail: Last node of group group_peak_memory: Current peak memory within the group candidate_delta_mem: Net memory change from candidate (alloc - free) candidate_allocfree: Candidate's allocation/free info group_n_to_bufs_after_swap_dealloc_by_candidate: Buffers whose deallocation moves to candidate curr_memory: Current memory state dict Returns: Tuple of (potential_peak_memory, post_alloc_update_dict) rr )max size_allocget mpi_buffer size_free)rrrgroup_peak_memorycandidate_delta_memcandidate_allocfree/group_n_to_bufs_after_swap_dealloc_by_candidate curr_memory_post_alloc_updatepotential_peakmem_after_reorder_deltargn_post_alloc_membufsbufcandidate_mem_post_allocs r?(_calculate_potential_peak_memory_reorderrs+>8:N :  3 3  #A &! "!,, -  111%8#7 D'OA.1HH!22^->?>BB2F   D'3>>+C+CC' D D J" ! "  ( ( ) %=y!)ABN - --rIc |sK|D]} || } | d|z | d|z f|| <||d|jz} | |jz } | | f||<y|jD]}|D]}|||< d}|D]R}||}td||D}||z }| |xj|zc_|| |jz }||f||<T||} | |xj|z c_| | |jz }| |f||<y)aN Update memory tracking structures after swap (reorder version). Updates curr_memory, buf_to_snode_last_use, and snodes_allocfree dictionaries to reflect the new memory state after swapping candidate with group. Args: candidate: Node that was moved gns: Group nodes group_tail: Last node of group candidate_delta_mem: Net memory change from candidate (alloc - free) candidate_allocfree: Candidate's allocation/free info group_n_to_bufs_after_swap_dealloc_by_candidate: Buffers whose deallocation moves to candidate post_alloc_update: Cached post-allocation memory values curr_memory: Current memory state dict (mutated) buf_to_snode_last_use: Buffer to last-use node mapping (mutated) snodes_allocfree: Node allocation/free info dict (mutated) rr Nc3HK|]}|jjywrdrrrrs r?rz=_update_memory_tracking_after_swap_reorder..ms"2  NN $ $2  ")rrr.sum)rrrrrrpost_alloc_updaterrrrcm_candidate_post_alloc_mem_candidate_post_free_memrr"size_free_to_move_to_candidate_sumr_gn_post_alloc_memsize_free_to_move_to_candidategn_post_free_memcandidate_post_free_mems r?*_update_memory_tracking_after_swap_reorderr4s< ; BRB1++1++KO   #A &)<)G)G G " &(;(E(E E ! & $" I @FFH3 3C)2 !# & 33/0&  @"3A"6.12 FqI2 / & +.LL*%%)GG% 25Ea5H5R5R R,.>? A @!2) <Y))-OO)!$4Y$?$I$II "K rIctt}|jD]I\}|jj}||vr!t fd|Ds6|j |K|S)a< Find buffers whose last use will change after swapping candidate with group. When we swap [candidate [group]] to [[group] candidate], some buffers that were last used by a group node will now be last used by candidate instead. This affects memory deallocation timing. Args: candidate: The node being moved gns: Group nodes being swapped with candidate buf_to_snode_last_use: Mapping of buffers to their current last-use nodes Returns: Dict mapping group nodes to buffers that will change their last-use node c3(K|] }|k( ywrdrp)rrsnode_last_uses r?rz6_find_buffers_with_changed_last_use..s6B2'6s)rr-itemsr succ_nodesrr)rrrrrrrs @r?#_find_buffers_with_changed_last_users}, D4  $ $ & T  ^^.. J & 6#66 7GNNsS T ;:rIctjs t|rdd|jfStjs t |ryy)aa Check if a candidate node can be grouped with collective during reordering. This pass processes collectives left to right, so we avoid grouping with already-processed collectives based on configuration. Args: candidate: Node to check for groupability Returns: Tuple of (is_groupable, reason_if_not_groupable) Fzcandidate contains_collective )FrTN)r(reorder_iterative_group_with_collectivesrr)reorder_iterative_use_runtime_estimationsrrs r?_is_node_groupable_for_reorderr$sO$  @ @ $Y /01C1C1E0FG   A A i (. rIc|}|Dcic]}|||j} }t| Dcgc]}| | c}} t|Dcgc]}||jc}} d| d| dgd} |jD cgc]\}} t || j dz | j dz | jdz | jdz | jdz | j| j| j| j| jg }}} tjjdrddlm}||| z n8d z t#| d zz d j%t't"|z t)|d |}t+||k(sJt-|||\}}}}d |z d|z t.j1t3ddfd|Scc}wcc}wcc}wcc} }w)a Format reordering statistics, log them, and return final node list. Computes improvement metrics, creates a formatted table (using tabulate if available), validates the reordered node count, recalculates peak memory, and logs all information. Args: stats: Per-node reordering statistics head: Head of the reordered linked list next_dict: Linked list next pointers original_snodes_num: Original number of nodes (for validation) peak_memory: Initial peak memory before reordering name_to_freeable_input_buf: Buffer memory tracking info graph_outputs: Graph output names Returns: Final reordered list of scheduler nodes zAreorder_communication_preserving_peak_memory improved overlap by z ns after z reorders. ) zCollective node comm_time(us) comp_time(us)initial exposed(us)final exposed(us)improvement(us)limiting factorrXrYr[rb@@tabulaterr-headers>Please `pip install tabulate` to nicely render overlap stats.  Nz peak_memory_before:z peak_memory_after:artifactc dddS)NrQstringrencodingrprprIr?z2_format_and_log_reordering_stats.."sB  rIcSrdrp)reorder_log_strsr?r8z2_format_and_log_reordering_stats..&s?rI metadata_fn payload_fn)rhrrXr node_summaryr^r_r`rarVrYr[rb importlibutil find_specr-rUrmaprr3r overlap_loginfor )statsrroriginal_snodes_numrrrrPr7rhtotal_improvement total_movesr0 node_inforowsr- new_snodesnew_peak_memoryr;r:s @r? _format_and_log_reordering_statsrMsF8JEOPE5*U+777PKP[IE[/IJJG5z%(..GHK LL]K^_l , G6!+ 0 0 2  E9     # %   # %  % % +  # #c )  ! !C '  % % OO     " "  " "  D  ~~ +%8     M  3w<$..499Sd^44.tT9EJ z?1 11 1=.  OQ1/ }==O..?@@O_% +  MQIG& sG, G1G6BG;c n d}|D]}t|sd}n|s|ifSddlm}t|}t t j jj}t t j j}t|||\}}} } } |Dcic]}|t|t|z} }i} d}t|\}}}tj}d}|}tj }d}|||||}|rn t#|s|} |||k\rn|dz }t%x}| |<t't)|d|| \}}}||_||_||z x|_|_||_||}|}|}i}d} ||d}!|~tj4r4|j0tj6 |j*zkr d|_n:tj:st|r d |_nt)|||}"||j<|"d }#|#j>D$cic]}$tA|$r|$jB|$}%}$|jE}&d}'|&D](}(|%jG|(jIdx}$s&|$}'n|'tK|\})}*|)r|}tj4r@tM|r"tOt)|||| \}}}+||f||<t#|s| | |z } tQ|!||d}!|xjRdz c_)tU|"|_+||}d |'d tY|%jd |jId|j[gdtU|"d|* },|,|_ntj4r_t#|s| |}-|-dkDrIt|dkDr:tQd|j*|j,z }.tQd|j*|j,z |-z }/|/|.z }0|j]D].\}1}2tQd|1|2z }3tQd|1|2z |-z}4|0|4|3z z }00|0dkDrd|-d|0|_n|j_D]\}5\}1}2|1|2|-z f||5<nq| dkDrlt't)|d|| \}}}+tQd||z }.tQd||z | z}/|/|.z }0|0dkDr&d|jId| d|0d|d| |_n| |}6|6j`|6jbz }7te||"| }8tg||"||!|7|6|8|\}9}:|9|z |tjhzkDrd|9d||_n|xjjdz c_5|dz }tm||||||}t't)|d|| \}}}||_||_||_||z |_to||"||7|6|8|:|| | |r/ddl8m9};|;||"tU|"t)|d|| |||| d|8 }|rn ||}|~|}|||tu| ||||| |}<|<| fScc}wcc}$w)z Internal testing helper that also returns debug info. Returns: - reordered snodes list - dict {snode: ReorderInfo} FTrr|Nr r unexposed by runtime estimationszcollective orderingrdata dependency (dep_names: ) candidate:z(outs:z)dep on  non_group_reason:candidate has compute z,, group contains waits, total_exposed_delta z candidate z is collective, group_runtime:z, exposed_delta:z c_comm_time:z c_comp_time:peak memory new: vs base:!_debug_iterative_memory_recomputerQ);rrr}r3r r graphrkeysget_output_namesrestimate_op_runtimerrr(reorder_iterative_debug_limit_to_reorder(reorder_iterative_debug_memory_recomputerrSrrr^r_r`rarbr")reorder_iterative_extra_comm_comp_overlaprV,reorder_iterative_unsafe_collectives_reorder schedulerrrrrrrr$rrrrYrr[r-rr.rrrrr$reorder_iterative_peak_memory_budgetrXrr comms_debugrYrM)=r5has_collectivesr7r}rFrrrrrrrrrErHrrr debug_num_collectives_to_reordernum_processed_collectivescurr debug_iterative_memory_recomputeiterative_recompute_error _next_currrDr^r_rbrrr group_waits group_runtimerrgroupr data_depscandidate_outsdata_depris_groupable_resultgrouping_reasonr;msg c_runtimeexposed_before exposed_after exposed_delta gw_comm_time gw_comp_timegw_exposed_beforegw_exposed_aftergwrrrrrrYrKs= r?rNrN,sO u %"O  rz>f+$.qww/C/C/H/H/J$KL%/0H0H0J%KM $FL-H " 0  "5),Ee,LLL0H0 35EK8@E5% ==%&'(-D==%!&  uT{64[ $ (.D  + 7 %)I I !Q&!(]*uT{-L )$e J) )6(F :E:L:L:NR6L,03A|l7R0S-/2 !<,#>#J0,*-=@Q-QQM R)1,"8 DNN[_!^!0" "-!2!2!4"!% , ,%1$09$<3" B "%q(2Q9)T5Q$3/ 9a *-Q I0E)F(+Ay9/D}/T(U (5(F (1,",Y-?-?-A,BC22?A22? i[Xefoep!r!0 "/? /J #..1D1N1NN 4s$9 <9%''C   /N., Q QQR '~&6i }M$ JJ!OJ 1 K4:z5%E2Q-dD%@(2 .Iy,'DN&DN ,D !*Y!6D  6##?"%  0K,M %1%uE.! $BC -)-j)II#JI  uT{6L2   "J u M 0bsX-'X28X2ci}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}r@t!j&j(}|rt |r ||n|r@j+D]} 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|_yrd)r7nextiterget_operation_namesrscore)rgr7r fused_namename_to_fused_nodescores_0scores_1scores_2s r?__init__z-_schedule_for_comm..Runnable.__init__sVDJU6689:D+D1::.Runnable.__lt__s:: + +rIN)rrT)rjrkrlrr)rrrrsr?Runnablers   ,rIrc34K|]}|jywrdr)rdeps r?rz%_schedule_for_comm..sGs#((Grcj||jD]N}|D]D}|j|t|dk(s)t j |FPy)zU Schedules `snode` and put all unblocked nodes onto the ready queue. rN)rrremover3heapqheappush)r7buf_namer buffer_usersready scheduledrs r?schedulez$_schedule_for_comm..schedulesv ..0 ;H%h/ ;5!((2z%()Q.NN5(5/: ; ;rIcDcgc].}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|jSrdrrs r?r8zG_schedule_for_comm..get_overlapping_candidate..s QWWrIkey)rr7rr3min)r candidatesrs r?get_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)rrr7rheapify)r7collective_costrrrr snode_to_costs r?schedule_collective_for_overlapz;_schedule_for_comm..schedule_collective_for_overlaps #5)))'. a 799F LL # Y__ % }Y__= =O a 799F  erIz;Detected unscheduled nodes. Nodes with unmet dependencies: )rrrrsysmaxsizer ancestorsrrr rrr]rr3rraddheappopr7r.)r5rCrDrEbuf_name_to_snoderr7rop_name node_namecomm_idxancestoranc_fused_namedepsrrrrrrrrrrrrrrs @@@@@@@@@@@@r?rGrGtsL#%r2 Hh' " U..0 0H*/ h ' 0002 0G*/ w ' 0/45>>+,NN$ !kk! "H+ .u5)1HU^^% &!OO S!3H!=!F!F!H+.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$** #6u#= +E 2 UO !!# 4yA~ I* V ~  U<Ls 9$I15I6c ztjjs|S|Dcgc]}t|s|}}t dt |D]b}t t||j}||dz jD]"}||jt||d$d|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 T mutating_bufr) r/r( is_availablerr+r3rrr add_fake_depr)nodes name_to_bufrr comm_nodesr>rrs r?decide_global_ordering_of_commsrs    ) ) + "=&9!&s B8B8ceZdZUdZded<dZded<dZded<dZded<d Zded <d Z d ed <d Z d ed<d Z d ed<d Z d ed<d Z ded<edZy) SinkWaitInforrWrYrZrUr[rX moves_inforTrVr\r]r^r_r`rarbc4|j|jz Srdrerfs r?rhzSinkWaitInfo.improvement5rirIN)rjrkrlrYrnr[rXrrVr^r_r`rarbrorhrprIr?rr(swGSL#E3NJ!OS!IuIu!OU!M5L# 99rIrc$t|rdd|jfSt|rdd|jfStjs@t |rdd|jfSt |rdd|jfSy)aP Check if a candidate node can be grouped during sink_waits pass. Sink Waits traverses waits right to left, so we don't group with processed waits on the right or with async collectives. Args: candidate: Node to check for groupability Returns: Tuple of (is_groupable, reason_if_not_groupable) Fzcandidate contains wait $candidate contains_async_collective zcandidate contains collective zcandidate contains gemm_like r )rrrr&sink_iterative_use_runtime_estimationsrrr#s r?!_is_node_groupable_for_sink_waitsr:s"Y01C1C1E0FGGG + 293E3E3G2H I   > > y )01C1C1E0FG  i (/ 0B0B0D/EF  rIc d|d} || d|| jz } |sC| |jz} | | |jz f||<|D]} || } | d|z| d|zf|| <y|g|D]F}||}||xj|j|dz c_||||jz f||<Hy)a Update memory tracking structures after swap (sink_waits version). Updates curr_memory and snodes_allocfree dictionaries to reflect the new memory state after swapping candidate with group. Args: candidate: Node that was moved gns: Group nodes candidate_delta_mem: Net memory change from candidate (alloc - free) candidate_allocfree: Candidate's allocation/free info group_n_to_bufs_after_swap_dealloc_instead_of_candidate: Buffers whose deallocation moves from candidate to group post_alloc_update: Cached post-allocation memory values size_free_delta_update: Cached size-free delta values curr_memory: Current memory state dict (mutated) snodes_allocfree: Node allocation/free info dict (mutated) rr N)rrr)rrrr7group_n_to_bufs_after_swap_dealloc_instead_of_candidatersize_free_delta_updaterrr pre_group_memcandidate_post_allocrrr post_allocs r?-_update_memory_tracking_after_swap_sink_waitsrfs 8QJ +A.1A*1M1X1XXM B,/B/M/MM #6#@#@ @" I BRB1++1++KO   #  &q) %%)?)C)CAq)II%  )!,66 6 A rIc ||d||jz } i} i} d} |s!t||z| |jz} | | | fS| |jz} | | |<| } tdtjj |j D}| | |<||z}|D]S}||d|z}|| |<t| |} d}||vr*||}|D]}||jjz }|| |<||z}U| | | fS)a Calculate potential peak memory after swapping candidate with group (sink_waits version). Computes new memory levels for all affected nodes and returns the potential peak memory along with cached post-allocation and size-free delta values. Args: candidate: Node being moved gns: Group nodes group_head: First node of group group_peak_memory: Current peak memory within the group candidate_delta_mem: Net memory change from candidate (alloc - free) candidate_allocfree: Candidate's allocation/free info group_n_to_bufs_after_swap_dealloc_instead_of_candidate: Buffers whose deallocation moves from candidate to group curr_memory: Current memory state dict snodes_allocfree: Allocation/free info for all nodes Returns: Tuple of (potential_peak_memory, post_alloc_update_dict, size_free_delta_update_dict) rc3HK|]}|jjywrdr r s r?rz>_calculate_potential_peak_memory_sink_waits..s"&    &r ) rrr itertoolschain from_iterabler.rr)rrrrrrrrrrr_size_free_delta_updaterrcandidate_size_free_to_move delta_memr gn_post_allocgn_size_free_to_addrrs r?+_calculate_potential_peak_memory_sink_waitsrs{> +A.1A*1M1X1XXM79<>N B  3 3 /:: : 13JJJ(+>+I+II$8y!)N"%&??00 C J J L &# +F)EI&#&AAI )#B*Y6 !.2^]; H HJ2ND @#s~~'?'??# @*= #B '((  ) -/F FFrIcl||}|r|||<|||<||}|r|||<|||<|||<|||<||k(r|S|S)a Swap positions of candidate and group in doubly-linked list (sink_waits version). Transforms (moves candidate to the left): group_head_prev -> group_head...group_tail -> candidate -> candidate_next Into: group_head_prev -> candidate -> group_head...group_tail -> candidate_next Args: candidate: Node to swap with group group_head: First node of group group_tail: Last node of group prev_dict: Dictionary mapping nodes to their previous nodes next_dict: Dictionary mapping nodes to their next nodes head: Current head of the linked list Returns: New head of the linked list (may change if group_head was the head) rp)rrrrrrgroup_head_prevcandidate_nexts r?+_perform_double_linked_list_swap_sink_waitsrss8 +O%. /"*Iiy)N$. .!*Ij&Ij%IiT KrIc*gd}|jD cgc]\}} t|| jdz | jdz | jdz | j dz | j dz | j| j| j| j| j| jg } }} dtjjdrddlm} | | |z n8dz t#|d zz d j%t't"| z t(j+t-|d |} t/| |k(sJt1| ||\} }}}d |z d | z t3d dfd| Scc} }w)a Format sink_waits statistics, log them, and return final node list. Computes improvement metrics, creates a formatted table (using tabulate if available), validates the reordered node count, recalculates peak memory, and logs all information. Args: stats: Per-node sink_waits statistics head: Head of the reordered linked list next_dict: Linked list next pointers original_snodes_num: Original number of nodes (for validation) peak_memory: Initial peak memory before reordering name_to_freeable_input_buf: Buffer memory tracking info graph_outputs: Graph output names Returns: Final reordered list of scheduler nodes ) z Wait noder&r'r(r)r*r+rYr[rXrrbr,rZr-rr.r/r1r2Nz* sink_waits_iterative peak_memory_before:z) sink_waits_iterative peak_memory_after:r3c dddS)Nsink_waits_iterative_infor5r6rprprIr?r8z2_format_and_log_sink_waits_stats..ds/  rIcSrdrp)log_strsr?r8z2_format_and_log_sink_waits_stats..hs7rIr;)rr>r^r_r`rarhrVrYr[rXrrbr?r@rAr-rUrrBrCrDrr3rr )rErrrFrrrr0r7rDrJr-rKrLr;rs @r? _format_and_log_sink_waits_statsrs8 G:!;;=  E4   NNS NNS  3 &    $   s "  LL    JJ OO     D "G~~ +%8    TT3w<$&&499Sd^,,W.tT9EJ z?1 11 1=.  OQ1 <[MJJG ;O;LMMG #  ] sBFctt}|jD]H\}}|jj}||k7r"d}|D] }||vs|} |5||j |J|S)a[ Find buffers whose last use will change after swapping in sink_waits pass. When we swap [group] candidate to candidate [group], some buffers that were last used by candidate will now be last used by a group node instead. This is the opposite direction from the reorder version. Args: candidate: The node being moved (currently last use) gns: Group nodes being swapped with candidate buf_to_snode_last_use: Mapping of buffers to their current last-use nodes Returns: Dict mapping group nodes to buffers that will change their last-use node N)rr-rrrr) rrrrrrr last_succ_gnrs r?._find_buffers_with_changed_last_use_sink_waitsrms, D<  $ $ &   ^^.. Y &   "BZ!  "    @ MTT  # * CBrIc ddlm}t|}|dk(r|ifSttj j j}ttj j}t|||\}}}}} t|\} } } i} |Dcic]}|t|t|z}}|d}t}tj}tj}d}|| || |}|rn|t||k\rnt!|r||vs|}6|j#|t%x}| |<t't)| || |\}}}||z x|_|_||_||_||_| |}|}|}|}i}d}||d} | tj4r4|j,tj6 |j.zkr d|_nt)||| }!||j:|!d}"|j<D#cic]}#t?|#r|#j@|#}$}#|"jC}%d}&|%D](}'|$jE|'jGdx}#s&|#}&ntjHs%tK|rd |jG|_n|&tM|"xr tM|x}(rQtO|\})}*|)r|}tj4r@tM|r5tQt)|d| |\}}}+||f||<tK|s|||z }tS| ||d} |xjTd z c_*tW|!|_,| |}|&8tjZs(rd tW|!d |jG|_n d |&dt]|$jd|jGd|j_gdtW|!d|%D'cgc]}'|'jGc}'d|*|_ntj4rHta|jbrPt't)| || |\}}}+tSd||z },tSd||z |z}-|-|,kDrd|,d|-|_ntK|s||}.|.dkDrt|dkDrtSd|j.|j0z }/tSd|j.|j0z |.z  |jeD]*\}0}1|/tSd|0|1z tSd|0|1z |.zz z }/,|/dkDrd|.d|/|_n`|jgD]\}2\}0}1|0|1|.z f||2<||}3|3jh|3jjz }4tm||!|}5to||!|| |4|3|5|| \}6}7}8|6|z |tjpzkDrd|6d||_n|xjrd z c_9|xjtd|jGz c_:tw|||| | | } t't)| || |\}}}||_||_||z |_||_ty||!|4|3|5|7|8|| |r/d dl=m>}9|9||!tW|!t)| d| | ||||d|5 }|rn | |}| |}|| |t| | | ||| |}:|:| fScc}wcc}#wcc}'w)Nrr|rFrrOTrPrr zcollective ordering z with candidate:rQrRrSz(os:z ) dep on z outs:rTz"candidate is wait, exposed_before:z vs exposed_after:rUz2, group contains collectives, total_exposed_delta rVrWrrXsink_waits_iterative)@rr}r3r r rZrr[r\rrr]rrr_(sink_waits_iterative_debug_limit_to_sinkrrrrrr`rar^r_rbr&sink_iterative_extra_comm_comp_overlaprVrbrrrrrr*sink_waits_iterative_swap_with_collectivesrrrrrrYrr[/sink_waits_iterative_unsafe_collectives_reorderr-rrryr.rrrrr!sink_iterative_peak_memory_budgetrXrrrrdrYr);r5r}rFrrrrrrrrrrrEr7rrhprocessed_waitsridebug_num_sink_waits_to_reorderrj _prev_currrDr^r_rbrrrr group_collsrmrrrnrro group_outsrqrboth_contain_comms _is_groupablegroupable_reasonr;rvrwrurx gc_comm_time gc_comp_timegcrrrrrrrYrKs; r?_sink_waits_iterative_internalrs?f+arz$.qww/C/C/H/H/J$KL%/0H0H0J%KM $FL-H "9@E5%35E0  "5),Ee,LLL0H0 )/r D lO==% ==$!&  uT{64[ $  + 7O$(GG d#O(CD D!)^+uT{-L )%u =x. * 9l5> 4IIt1""($K      (.q1#BB""FFFWX(J$+HJ,C)$$"E$-#?#?|TU I**,JH ! ajjlD9919 H  JJ,Y7>y?Q?Q?S>TU(#'.Q3Fy3Q"3T3/ /!!*J%KK/ :2Q9)T5Q$3/ 9a3>*/N5eYNPX/+Iy!&)I ,A%BN$'9y+@=+P$QM%~5//=.>>PQ^P_a,1;!) 3I 1}[)9A)=), NNT^^;) Q ?) KLL:E:L:L:N6L,)SL<4O-PSV !<,#>#JT.M)1,"8 D88E!H!0" "-!2!2!4"!% , ,%1$09$<3" B "0@ /J #..1D1N1NN ?s$9 D<%''K $  HN.0G, N NNO '~&6i }M$ JJ!OJ OO9#5#5#7"89 9O?:z5%E2Q-eT5A82 .Iy,'DN&DN!*Y!6D  ,D  9##G"'  0J,M %1%uE.! $*K -)-j)I[#\_  uT{6b2   "J u W 0JJ$Fs[ [#[0[ct|dS)a Similarly to reorder_communication_preserving_peak_memory this pass will try to iteratively push Wait nodes later, recomputing estimated peak memory before each swap, and preventing peak memory regressions. Pass will be applied to every Wait node. If there are immediate dependencies with next node, pass will try to group them together and on the next step to swap the group with next candidate. If _inductor.config_comms.sink_iterative_use_runtime_estimations is set True, pass will stop reordering of Wait once corresponding Collective is unexposed, based on runtime estimations. inductor.config_comms.sink_iterative_peak_memory_budget allows to tune how much pass can regress initial peak memory. E.g.: sink_iterative_peak_memory_budget == 0.0 - No regression of initial peak memory is allowed sink_iterative_peak_memory_budget == 0.2 - Pass can improve comm-compute overlap, sacrificing 20% of initial peak memory value. inductor.config_comms.sink_iterative_extra_comm_comp_overlap config allows to more aggressively sink waits, stopping only when overlap_compute >= (1 + extra_comm_comp_overlap) * comm_time r)rrHs r?rrs. *& 1! 44rIctjdk(r|j}|SttjsJtj|}|S)z; Returns estimated op runtime in milliseconds (ms) rw)rr]r&callable)r7runtimes r?r]r]sR!!Y.--/ N22333,,U3 NrIc b|j}t|dk(rd}t|jtj tj frd|jDcgc]}|jc}}d|jDcgc]}|jc}}d|jd|jjd|d|d }|jDcgc]}|jj}}d j|D cgc];} t| tjrd | jd | j d nd=c} } |jj#} |jj&j(|| d| d|j+d dSg} |D]} | j-t/| |j&j(ddj| Scc}wcc}wcc}wcc} w#t$$rd} YwxYw)Nr rZzouts:zins: z (z) (),z (size=z , stride=z.0fz ns): z, ) get_nodesr3rryrExternKernelOut_CollectiveKernelrrrrrrget_output_specrLayoutsizestridemaybe_get_nameAttributeError __class__rjr&rr>)r7r5detailrouts_strrins_strchildlayoutslayoutout_tensor_infor summaries child_snodes r?r>r>s __ F 6{a ejj2#5#5r7K7K"L Me6G6G6IJ JKLHe.F.FGaffGHIG)*"UZZ-J-J,K4PXzYZ[bZccdeF=B__=NOE5::--/OO(( &  fbii0&++i a@    113I**&&//08II;VXY^YtYtYvwzX{{A AI4 k234oo&& 'r$))I*>)? @@/ KGO  I s+)H H/!H!AH)H H.-H.c&d}d}d}t|D]\}}|^t|r|t|z }|j}n$t |jrn|t|z }||t |ft|r.|t|z }|j}||t |t |jr||t |d}||dt |t jd|dz dz y)Nrc:tj|dd|y)Nz>6r)rCdebug)steprts r?step_logz#visualize_overlap..step_log,sT"IRu-.rIz| zEst. runtime (ms): i)rrr]ryrr>rCr)ordertotal_est_runtime cur_comm_noderrr7s r?visualize_overlapr%s #M/!'; e  "5)!%8%??! % $!%8%??! Tl512 4"5)!%8%??! % ,u"5!68$,u"5!68 $ L$7#89:-;. /$6=>?rIc(|}ttjjj }ttjj }t jD]I}t|tr|tvr t|}t|s Jd|dt|t|||\}}tjj!dk(r(t"j%d|d|d t'|t+j*}||}t+j*|z } tjj!dk(r(t"j%d |d | d  t'|t|t|||\}}t-d |L|S#t($r!}t"j%d|Yd}~d}~wwxYw#t($r!}t"j%d|Yd}~d}~wwxYw) Nz3Invalid reorder_compute_and_comm_for_overlap pass: z is not callablerz.==== Visualize overlap before reordering pass z, peak_memory=z ====rZ)exc_infoz-==== Visualize overlap after reordering pass z (ran in z sec)====zfinal peak_memory=)r r rZrr[r\r'reorder_for_compute_comm_overlap_passesrrUglobalsrrrr/r(get_rankrCrr Exceptiontimeprint) r5rrrprr;et0ts r?$reorder_compute_and_comm_for_overlapr*Ks E$.qww/C/C/H/H/J$KL%/0H0H0J%KM  ; ;' a !wy. ! A{ A!DT U {. *6<@-  Q    % % '1 ,   @?k^SXY  2!%(YY[% IIK"     % % '1 ,   ?s)A3iX  2!%(. *6<@-  Q #{n%&?'B L% 2!!"q!11 2 2!!"q!11 2s0 F: G': G$GG$' H0H  Hc t|jtttttD]\}}|jdk(s|j t jjjjusQ|jdjdk(sJd|d|jdd|jd}|jd}|dkDr|j||j|fd}tt}tD]\}}|jdk(s|j t jjjjusQ|}|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}||_ $|j3D]} | D]} | }|j5|!D]p}|jdk(s|j t jjjjusN|jd|vs`|j5|ry)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 r2r clj|g}j|g}t|t|k7r2tjd|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)rr3logwarningr) 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 r?check_resize_patternzLremove_fsdp2_unsharded_param_graph_input_usage..check_resize_patterns !J M M ! DGG UWX $ %-?)@ @ KKHHS}UE#&8"9!:; 47 !#54  / "_4  3I>P4Q3RR\]o\pq )/ :;:oEVW rIz\ Assumed all FSDP2 `unsharded_param`s to be graph input, but it's not true! Offending node: z . Graph: c|jtjjjj uxs:|jtjj jj uSrd)targetr/rtfsdpcopy_rwinductorresize_storage_bytes_)rys r?is_allowed_mutationzKremove_fsdp2_unsharded_param_graph_input_usage..is_allowed_mutationsO KK599>>//77 7 O{{eii00FFNNN rIc 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)rr;r/_ops OpOverloadr_schema arguments alias_infois_writer r argsmetauntyped_storager3)ryunsharded_paramsr>rmutated_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|]}|gywrdrp)rryrQrOs r?rzAremove_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|ywrdrp)rarg replacementrOs r?rzAremove_fsdp2_unsharded_param_graph_input_usage.. s%%(+o'= 3F%sN)r-rrropr;r/rtr>r?rwrIrr<r=rrCrDrE is_mutabler[rr3rtupler. erase_node)rZrryr1new_sizer9'unsharded_param_to_fsdp_copy_node_idxesfsdp_copy_noder@fsdp_copy_node_idxesr>fsdp_copy_node_idxr;subgraph_start_idxsubgraph_end_idxsubgraph_nodesnew_argsr6r7rQr8rUrOs @@@@@@r?.remove_fsdp2_unsharded_param_graph_input_usagercus{U[[!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 )) ) ")J!H N N P-"6 - &'9:N   ^ , -- # GG & uyy11GGOOO ! GG   T " #rIc 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 fd }|||j+|y#tttf$rYywxYw)Nrr ) CallFunction KeywordArgMatchPatternMatcherPassregister_graph_patterncHt|j}|D]}|jtjus |j djj jjjusd|j ddk(sw|j|y)Nrr ) r-rr;operatorgetitemrIrtr<all_gather_copy_inrwrY)gr8rr/s r?remove_unused_getitemz8reinplace_fsdp_all_gather..remove_unused_getitem[ suM  AH,,,FF1I$$ (I(I(Q(QQFF1IN Q  rIall_gather_inputsall_gather_outputinp_split_sizesall_gather_input_numelrankitem_idx group_size group_namec&|jddk(S)Nrur)kwargs)matchs r?r8z+reinplace_fsdp_all_gather..| s%,,z":a"?rI) pass_dict extra_checkc lfd}|j||d|d|d|d|d|d|dgy) Nc|dd}|d}|d}jjjj|}|d}|d}jjj j||||}|S)Nrrr )out)rtr<rmrw_c10d_functionalall_gather_into_tensor_out) rI copy_in_argsrvrwrmrl getitem_1all_gather_into_tensorr/s r?replzEreinplace_fsdp_all_gather..reinplace_all_gather..repl s 9LbJbJ!J!B!B!J!J" )+G*1-I **EEMMZN # * )rIrprqrrrsrtrvrw)replace_by_example)rzrIryrr/s r?reinplace_all_gatherz7reinplace_fsdp_all_gather..reinplace_all_gatherh s[. *$   *+*+()/0v|$|$  rI)rzrg)5torch.distributed.fsdp._fully_shard._fsdp_collectivesr(rrtrrr ImportErrorr AssertionErrorpattern_matcherrerfrgrhrirwrkrlr<rmapply) rZrerfrgrhriro graph_passrr/s @r?reinplace_fsdp_all_gatherr6 sd D  --/// II & & = = **EE  FE   $%J II & & = = E E   IINN55==23230178v& :&  | $ | $ $?), -, B% UA  8sA"EE10E1ct|tjjjtjjj frJt |jddS)Nr)rr/ _inductorrbFusedSchedulerNoder}rWrrs r? get_op_idxr s] OO % % 8 8 OO % % : :   u~~# $$rIc ~  !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]k\}}|bt=t?|jA}|jCD]-}|jEtG|j5|d/|}md}|j;D]k\}}|bt=t?|jA}|jCD]-}|jEtG|j5|d/|}m|S)Nr )rbFcjj|}|D]}||j<||j<|Srd)r}creater)snodes_to_group group_noder7rbsnode_name_to_final_snodes r?_create_group_nodez:enforce_comm_ordering_for_fsdp.._create_group_node sV33::?K $ EE:D %enn&6 7 E;E!*"5"5"78rI)rVc3K|]I}t|jtjjj j Kywrd)rryr/rtr<rmrw)rrrs r?rz1enforce_comm_ordering_for_fsdp.. sD  "1%**EIINN,M,M,U,U  sAATct|jxs0t|jxr|jjv Srd)rNopKernelSchedulerNodeExternKernelSchedulerNodery op_overload)r allowed_opsrbs r?r8z0enforce_comm_ordering_for_fsdp.. sFq)"B"BC"1i&I&IJ>FF..+= 'rI) criteria_cbct|Srdrrs r?r8z0enforce_comm_ordering_for_fsdp.. JqMrIrrct|Srdrrs r?r8z0enforce_comm_ordering_for_fsdp..$ rrIr)$rZrbr rrryr/rtrrrwrrr wait_tensorr<split_with_sizes_copyrsortedr3r+rrr _WaitKernel chunk_catrrrrrrrrrr)"r5rr new_orderr ag_exists rs_exists$ag_grouped_node_to_wait_grouped_node$rs_grouped_node_to_wait_grouped_noderr7ag_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_noderr prev_rs_waitrrbrs" ` @@@r?enforce_comm_ordering_for_fsdpr s )+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|TR ' 'L*N*T*T*V'&   #]%C%C%E FGL!--/ **AJJL|TR ' ' rI)r5list[BaseSchedulerNode])r5rrr)ryz"Optional[Union[IRNode, Operation]]rr)r7r"rr)r7r")rrrdict[BaseSchedulerNode, float]rtuple[float, float, str])r5rrrrr)F)rrU)rrrrU)r5rrztuple[dict[BaseSchedulerNode, Optional[BaseSchedulerNode]], dict[BaseSchedulerNode, Optional[BaseSchedulerNode]], BaseSchedulerNode])rr"rr"rr)rOptional[BaseSchedulerNode]rrr4dict[BaseSchedulerNode, Optional[BaseSchedulerNode]]rr)rr"rr"rr"rrrrrr"rr")rr"rrrr"rrWrrWrrrrrrrz(tuple[int, dict[BaseSchedulerNode, int]])rr"rrrr"rrWrrrrrdict[BaseSchedulerNode, int]rrrrrrrrT)rr"rrrrrz>dict[BaseSchedulerNode, list[Union[FreeableInputBuffer, Any]]])rr"rztuple[bool, Optional[str]])rEz$dict[BaseSchedulerNode, ReorderInfo]rr"rrrFrWrrWrrrOrderedSet[str]rr)r5rrzDtuple[list[BaseSchedulerNode], dict[BaseSchedulerNode, ReorderInfo]]) r5rrCrrDrrErrr)rrrr)rr"rrrrWrrrrrrrrrrrrrrT)rr"rrrr"rrWrrWrrrrrrrrrzFtuple[int, dict[BaseSchedulerNode, int], dict[BaseSchedulerNode, int]])rEz%dict[BaseSchedulerNode, SinkWaitInfo]rr"rrrFrWrrWrrrrrr)r5rrzEtuple[list[BaseSchedulerNode], dict[BaseSchedulerNode, SinkWaitInfo]])r7r"rr])rZtorch.fx.Graph)rZrrrT)r51list[torch._inductor.scheduler.BaseSchedulerNode]rz4dict[str, torch._inductor.scheduler.SchedulerBuffer]rzdict[str, BaseSchedulerNode]rr)h __future__rrr?rloggingrkrr$ collectionsr dataclassesrtypingrrrr r/torch._loggingr torch.multiprocessing.reductionsr torch.utils._ordered_setr rZrrr dependenciesrrrmemoryrrrrrutilsrrrrrrr virtualizedr getLoggerrjr/_logginggetArtifactLoggerrCrr"r@rDrCrLrQrSrzrrrrrrrrrrrrrrrrrrr$rMrNrGrrrrrrrrrrr]r>rr*rcrrrrprIr?rsb#  #!66 +;/&&!%g!nn..xC ;M #M, #&# ###L 99 9("(7.+7.7U7.7.t;. #;.,;.;.|03& #" .' .5F .  .G ; % %D 6. .!.".D . D .  ..b@. @. @."@. @.  @. % @.6:@.@..@.FJ J J"J J % J 6: J4JJ JJ JZ$; $; $; $;D $;N B 2 4 2 92 2 2  2 jCG CG CG"CG CG  CG % CG>BCGCGCGLCGL. .!.".D . D .  ..bX 0X XDX X  X !% X#XXv,C ,C ,C ,CD ,C^A #AJAH 54 A>#L' #''T~#BjZ%n =nEn5n7 nrI