a h@sddlmZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z ddl m Z mZddlZddlmZddlmZddlmZd d lmZmZd d lmZd d lmZmZmZd d lm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&d dl'm(Z(e)e*Z+ej,-e*dZ.erddl/m0Z0dddddZ1dddddZ2dddddZ3dddddZ4e GdddZ5dddddZ6dd d d dd!d"d#Z7ddd$d%d&Z8d'd(d)d*d+Z9d,d-Z:d.d/Z;dddd0d1Zd:d;Z?ddqz@reorder_communication_preserving_peak_memory..csg|] }|qSr$r$r)r'r$r% rr/z@reorder_communication_preserving_peak_memory..csg|]}|jqSr$)movesr)r,r$r%r0sr/zAreorder_communication_preserving_peak_memory improved overlap by z ns after z reorders. )zCollective nodezinitial exposedz final exposedr(zlimiting factorr1cSs.g|]&\}}t||j|j|j|j|jgqSr$) node_summaryinitial_exposed final_exposedr(limiting_factorr1)r*r+Znode_reorder_infor$r$r%r0s tabulater)r6)headersz>Please `pip install tabulate` to nicely render overlap stats.  ZartifactcSs dddS)N,reorder_communication_preserving_peak_memorystring)nameencodingr$r$r$r$r%sz>reorder_communication_preserving_peak_memory..csSNr$r$)reorder_log_strr$r%r=r/)Z metadata_fnZ payload_fn) 6_reorder_communication_preserving_peak_memory_internalsumitems importlibutil find_specr6strjoinmap overlap_loginfor)rZreordered_snodesZtotal_improvement total_movesr7rowsr6r$)r(r-r?r%r9Os>        r9c@sNeZdZUdZdZded<dZded<dZded<d Zd ed <e d d Z dS) ReorderInfozE Debug info describing how an individual snode was reordered floatr3r4NonerFr5rintr1cCs |j|jSr>)r3r4)selfr$r$r%r(szReorderInfo.improvementN) __name__ __module__ __qualname____doc__r3__annotations__r4r5r1propertyr(r$r$r$r%rMs     rMzDtuple[list[BaseSchedulerNode], dict[BaseSchedulerNode, ReorderInfo]]csHt|d}d}t|}tjdur(tj}ttjj}ttj}t ||}t |||\}}dd|Di} fdd} t |D]\} } t | rt } | | <| | || dd| _| _||krd | _qt| dd d D]V}||}|td| |krd | _qt |r$d | _qtd d| jDtfdd|Drht|shd| _q|||||d||krd| _q| j| krd| _q| jd7_|d7}||}||d||<|||d<||d||}||||d}||||||<| | ||dd| _qq|| fS)z Internal testing helper that also returns debug info. Returns: - reordered snodes list - dict {snode: ReorderInfo} drNcSsi|]}|t|qSr$estimate_op_runtimer)r$r$r%r.r/zJ_reorder_communication_preserving_peak_memory_internal..csFt|}d}|D]&}t|rqt|r*q8||7}qtd||S)Nr)r[rrmax)Zcollective_snodeZremaining_snodesZ comm_timeZ compute_timer+)runtimesr$r%exposed_communication_timeszZ_reorder_communication_preserving_peak_memory_internal..exposed_communication_timer z move limitrNzprefetch limitzcollective orderingcSsg|] }|jqSr$r;)r*sr$r$r%r0r/zJ_reorder_communication_preserving_peak_memory_internal..c3s|]}|vVqdSr>)get_name)r*o) dep_namesr$r% szI_reorder_communication_preserving_peak_memory_internal..zdata dependencyz peak memoryzsufficient overlapping)lenr Zreorder_prefetch_limitr rgraph graph_inputskeysget_output_namesrr enumeraterrMr3r4r5ranger]unmet_dependenciesany get_outputsrr1)rZ MOVE_LIMITrKZPER_COLLECTIVE_PREFETCH_LIMITrh graph_outputsZname_to_freeable_input_buf peak_memoryZ curr_memorystatsr_ir+Z reorder_infojZ prev_snodetmpZj_plus_one_allocZj_allocr$)rdr^r%r@st       "  r@bool)rrrr rcs(i}iiii t|D]b\}}|D] }|||<q0|D] }||<qF||<|} tj| <d| <| | <q d} |D]h}|rt|r| |<|jD]"} | } t| | | <q| d7} q|rt |rd|<qG fddddd|D gt t dd|D D]B\}} t | dkrdt|| D]}||qhq>g fdd fd d  fd d }t rtj}|rt|r||n|q D]&\}} t | dksJd qS)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 cs,eZdZddfdd ZddZdS)z$_schedule_for_comm..RunnablerP)rcs>||_tt|}|}|||f|_dSr>)r+nextiterget_operation_namesrbscore)rRr+r;Z fused_namename_to_fused_nodescores_0scores_1scores_2r$r%__init__Qs z-_schedule_for_comm..Runnable.__init__cSs |j|jkSr>rz)rRotherr$r$r%__lt__[sz+_schedule_for_comm..Runnable.__lt__N)rSrTrUrrr$r{r$r%RunnablePs rcSs"i|]}|tdd|jDqS)css|] }|jVqdSr>r`)r*depr$r$r%re_r/z0_schedule_for_comm...)r rmr)r$r$r%r.^sz&_schedule_for_comm..cSsi|]}|t|qSr$rZr)r$r$r%r.er/csX||D]@}|D]2}||t|dkrt|qqdS)zU Schedules `snode` and put all unblocked nodes onto the ready queue. rN)appendget_buffer_namesremoverfheapqheappush)r+buf_name)r buffer_usersready scheduled unmet_depsr$r%scheduleos    z$_schedule_for_comm..schedulecs.ddD}t|dkrdSt|dddS)zh Return the next node in the ready queue that's neither a collective or a wait. cSs$g|]}t|jst|js|qSr$)rr+rr*xr$r$r%r0szI_schedule_for_comm..get_overlapping_candidate..rNcSs|jSr>rrr$r$r%r=r/zG_schedule_for_comm..get_overlapping_candidate..key)rfmin) candidates)rr$r%get_overlapping_candidatezs  z5_schedule_for_comm..get_overlapping_candidatecsdt|s J||}|dkrV}durV||j||j8}qtdS)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)rrr+rheapify)r+Zcollective_cost candidate)rrr snode_to_costr$r%schedule_collective_for_overlaps    z;_schedule_for_comm..schedule_collective_for_overlapz;Detected unscheduled nodes. Nodes with unmet dependencies: )rkrryrbsysmaxsizer ancestorsrrrr rBrfrraddheappopr+)rrrr Zbuf_name_to_snodeidxr+rZop_name node_nameZcomm_idxZancestorZanc_fused_namedepsrrr$) rrrr|rrrr}r~rrrr%r"sb&                  r")nodesrcCsttjs|Sdd|D}tdt|D]D}tt||}||dD]}||t ||dqRq*|S)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 cSsg|]}t|r|qSr$)r)r*nr$r$r%r0r/z3decide_global_ordering_of_comms..r  mutating_buf) torch distributed is_availablerlrfrwrxr add_fake_depr )r name_to_bufr|Z comm_nodesrsrbufr$r$r%decide_global_ordering_of_commss rrrO)r+rcCs0tjdkr|}nttjs"Jt|}|S)z: Returns estimated op runtime in nanoseconds (ns) default)r r[get_estimated_runtimecallable)r+Zruntimer$r$r%r[s    r[cCs|}t|dkrd}t|jtjtjfr.,cSs2g|]*}t|tjr*d|jd|jdndqS)z (size=z , stride=rr) isinstancer ZLayoutsizeZstride)r*Zlayoutr$r$r%r0s z.0fz ns): z, )Z get_nodesrfrrr ZExternKernelOutZ_CollectiveKernelZpython_kernel_namerGZmaybe_get_nameAttributeError __class__rSrrr2)r+rZdetailZlayoutsZout_tensor_inforZ summariesZ child_snoder$r$r%r2s(   *r2cCsd}d}dd}t|D]\}}|durnt|rD|t|7}|j}nt|jrPn |t|7}||t|qt|r|t|7}|j}||t|qt|jr||t|d}q||dt|qtd|dddS)Nr\cSst|dd|dS)Nz>6r)rIdebug)stepmsgr$r$r%step_logsz#visualize_overlap..step_logz| zEst. runtime (ms): i)rkrr[rrr2rIr)orderZtotal_est_runtimeZ cur_comm_noderrr+r$r$r%visualize_overlaps.     rc Cs|}ttjj}ttj}tjD]`}t|t rL|t vrLt |}t |sdJd|dt |t |||\}}tjdkrtd|d|dz t|Wn2ty}ztjd|dWYd}~n d}~00t}||}t|} tjdkrftd |d | d z t|Wn4tyd}ztjd|dWYd}~n d}~00t |t |||\}}td |q(|S) Nz3Invalid reorder_compute_and_comm_for_overlap pass: z is not callablerz.==== Visualize overlap before reordering pass z, peak_memory=z ====r)exc_infoz-==== Visualize overlap after reordering pass z (ran in z sec)====zfinal peak_memory=)r rrgrhrirjr Z'reorder_for_compute_comm_overlap_passesrrFglobalsrrrrrZget_rankrIrr Exceptiontimeprint) rrrhrpprq_et0tr$r$r%$reorder_compute_and_comm_for_overlapsH      $  $ rztorch.fx.Graph)rgc sJt|jtttttD]\}}|jdkr"|jtjjj j kr"|j djdksrJd|d|j dd|j d}|j d}|dkr| |q"| |q"fdd }tt}tD]j\}}|jdkr|jtjj jj kr|}|j djdks"Jd d |d|r| |qd d }ddD]Z}|jdkrNt|jtjjrN|jjjrN||sN||rNJd|dqN|D]\} t| D]\} } | }|j dusJ|j \} | d} | t| dkr| | dn td}| |}tfdd|DrfJdd|d|d|D]R}|jdkrj|j vrj|jtjjj j krjtfdd|j D}||_ qjqĐq|D]0\} t| D]\} } | }||qސqΈD]@}|jdkr|jtjjj j kr|j d|vr||qdS)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 r8r cs|g}|g}t|t|ksRtd|dt|dt|ddSt||D]H\}}||kr\td|d|d|d |d|d dSq\d S) 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)getrflogwarningzip) graph_inputZresized_to_full_idxesZresized_to_0_idxesZresize_to_full_idxZresize_to_0_idx)&graph_input_to_resized_to_0_node_idxes)graph_input_to_resized_to_full_node_idxes node_listr$r%check_resize_patternYsD  zLremove_fsdp2_unsharded_param_graph_input_usage..check_resize_patternz\ Assumed all FSDP2 `unsharded_param`s to be graph input, but it's not true! Offending node: z . Graph: cSs$|jtjjjjkp"|jtjjjjkSr>)targetropsfsdpcopy_rinductorresize_storage_bytes_rr$r$r%is_allowed_mutationszKremove_fsdp2_unsharded_param_graph_input_usage..is_allowed_mutationcsdtjtjjr(ddtjjjDng}tfdd|D}tdd|D}t ||@dkS)NcSs&g|]\}}|jdur|jjr|qSr>)Z alias_infoZis_write)r*rsrr$r$r%r0szyremove_fsdp2_unsharded_param_graph_input_usage..is_node_mutating_unsharded_param_or_its_alias..cs$g|]}tj|jdqSval)rargsmetauntyped_storage)r*rsrr$r%r0scSsg|]}t|jdqSr)rrr)r*unsharded_paramr$r$r%r0sr) rrr_ops OpOverloadrk_schema argumentsr rf)rZunsharded_paramsZmutated_arg_idxesZmutated_node_arg_storagesZstorages_of_unsharded_paramsr$rr%-is_node_mutating_unsharded_param_or_its_aliass"   zeremove_fsdp2_unsharded_param_graph_input_usage..is_node_mutating_unsharded_param_or_its_aliaszdUser mutation on FSDP2 unsharded param is not allowed when Traceable FSDP2 is used. Violating node: c3s|]}|gVqdSr>r$)r*r)rrr$r%reszAremove_fsdp2_unsharded_param_graph_input_usage..z(Assumed no ops mutating unsharded param z in subgraph z, but it's not true! Graph: c3s|]}|urn|VqdSr>r$)r*arg) replacementrr$r%resN)listrrrkoprrrrrrrrrrrrrrZ is_mutablerirBrfrntuple erase_node)rgrrrZnew_sizerZ'unsharded_param_to_fsdp_copy_node_idxesZfsdp_copy_noderZfsdp_copy_node_idxesrsZfsdp_copy_node_idxrZsubgraph_start_idxZsubgraph_end_idxZsubgraph_nodesnew_argsr$)rrrrrrr%.remove_fsdp2_unsharded_param_graph_input_usage7s    %       rrP)rgrc s z2ddljsJjjjr,jjjs0JWnttt fyLYdS0ddl m }m }m }m}m}fdd}|}||jjjj|tj|jjjj|d|d|d|d |d |d |d |d |d |d|d|d|dddddfdd }||||dS)Nrr ) CallFunction KeywordArgMatchPatternMatcherPassregister_graph_patterncsTt|j}|D]@}|jtjkr|jdjjjjj ur|jddkr| |qdS)Nrr ) rrroperatorgetitemrrrall_gather_copy_inrr)grrrr$r%remove_unused_getitem s   z8reinplace_fsdp_all_gather..remove_unused_getitemall_gather_inputsinp_split_sizesall_gather_input_numel world_sizerankdtypedevicegroup_name_inner"allocate_memory_from_process_groupitem_idx group_size group_namecSs|jddkS)Nrr)kwargsmatchr$r$r%r=Dr/z+reinplace_fsdp_all_gather..)Z pass_dictZ extra_checkrr cs^fdd}|||d|d|d|d|d|d|d |d |d |d |d g dS)NcsX|dd}|d}|d}jjjj|}|d}|d}jjjj||||d}|S)NrNrr )out)rrrr_c10d_functionalall_gather_into_tensor_out)rZ copy_in_argsr r rrZ getitem_1all_gather_into_tensorrr$r%replGs   zEreinplace_fsdp_all_gather..reinplace_all_gather..replrrrrrrrrrr r )Zreplace_by_example)r rr rrr$r%reinplace_all_gather-s  z7reinplace_fsdp_all_gather..reinplace_all_gather)Z5torch.distributed.fsdp._fully_shard._fsdp_collectivesrrrrrr ImportErrorrAssertionErrorZpattern_matcherrrrrrrrrrrapply) rgrrrrrrZ graph_passrr$rr%reinplace_fsdp_all_gathersP     $rcCs2t|tjjjtjjjfrJt|ddS)N)rrZ _inductor schedulerZFusedSchedulerNodeGroupedSchedulerNoderQrb)r+r$r$r% get_op_idxnsrz1list[torch._inductor.scheduler.BaseSchedulerNode]z4dict[str, torch._inductor.scheduler.SchedulerBuffer]zdict[str, BaseSchedulerNode])rrr|rc sddlmg}tt}d}d}i}i}ifdd} |D]8} t| jtjjj j drt fdd| j Drd }| } t} t | | |ttjjj j tjjjj tjjjj gt| | |fd d d t| d d d} t| }d}tt| D]<}| |}t|jtjjjj r(|d7}|dkr|}q)rcreaterb)Zsnodes_to_groupZ group_noder+)rsnode_name_to_final_snoder$r%_create_group_nodes   z:enforce_comm_ordering_for_fsdp.._create_group_node)rc3s&|]}t|jtjjjjVqdSr>)rrrrrrrr)r|r$r%resz1enforce_comm_ordering_for_fsdp..Tcs&t|jp"t|jo"|jjv Sr>)rZNopKernelSchedulerNodeZExternKernelSchedulerNoderZ op_overloadr) allowed_opsrr$r%r=s   z0enforce_comm_ordering_for_fsdp..)Z criteria_cbcSst|Sr>rrr$r$r%r=r/rrcSst|Sr>r!rr$r$r%r=r/r)$rrr rrrrrrrrrnrrZ wait_tensorrZsplit_with_sizes_copyrsortedrfrlrrr Z _WaitKernelZ chunk_catrbrrrBrwrxrrorr )rrr|Z new_orderrZ ag_existsZ rs_existsZ$ag_grouped_node_to_wait_grouped_nodeZ$rs_grouped_node_to_wait_grouped_noderr+Zag_snodeZag_related_snode_setZag_related_snodesZend_idx_of_current_ag_blockZcopy_out_countrsZ cur_snodeZ wait_node_idxZ ag_group_nodeZag_wait_group_nodeZrs_snodeZrs_related_snode_setZrs_related_snodesZ rs_group_nodeZrs_wait_group_nodeZ prev_ag_waitZwait_group_noderrcZ prev_rs_waitr$)r r|rrr%enforce_comm_ordering_for_fsdpys                    r#)A __future__rrrCloggingrrr collectionsrZ dataclassesrtypingrrrZtorch._loggingrZ torch.multiprocessing.reductionsrZtorch.utils._ordered_setr rr r Z dependenciesr ZmemoryrrrutilsrrrrrrrZ virtualizedr getLoggerrSrZ_loggingZgetArtifactLoggerrIZtorch._inductor.schedulerrrrr&r9rMr@r"rr[r2rrrrrr#r$r$r$r%sR       $     WY &)Es