o Zh~@sPddlmZddlZddlZddlZddlZddlmZddlm Z m Z ddl Z ddl m Z ddlmZddlmZmZdd lmZdd lmZmZmZmZmZmZmZeeZe j !ed Z"e rhdd l#m$Z$d8ddZ%d8ddZ&d8ddZ'd9ddZ(d:ddZ)d;d d!Z*d"d#Z+d$d%Z,d8d&d'Z-dd6d7Z1dS)?) 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)BaseSchedulerNodesnodeslist[BaseSchedulerNode]returncCst|ddddS)z7 Greedily schedules waits as late as possible. FT raise_comms sink_waitsreorder_for_overlap_schedule_for_commrrD/var/www/auris/lib/python3.10/site-packages/torch/_inductor/comms.pyr$rcCst|ddddS)z8 Greedily schedules comms as early as possible. TFrrrrrr r-r!rcCst|ddddS)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. Trrrrrr reorder_compute_for_overlap6sr"boolrcsi}iiii t|D]1\}}|D]}|||<q|D]}||<q#||<|} tj| <d| <| | <qd} |D]4}|rnt|rn| |<|jD]} | } t| | | <qW| d7} qF|rzt |rzd|<qFG fddddd|D gt t dd|D D]\}} t | dkrt|| D] }||qqg fdd fd d  fd d }t rtj}|rt|r||n|t s݈ D]\}} t | dks Jd 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. rrcs(eZdZdfdd ZddZdS) z$_schedule_for_comm..RunnablerNonecs>||_tt|}|}|||f|_dSN)snodenextiterget_operation_namesget_namescore)selfr&nameZ fused_namename_to_fused_nodescores_0scores_1scores_2rr __init__s  z-_schedule_for_comm..Runnable.__init__cSs |j|jkSr%r+)r,otherrrr __lt__s z+_schedule_for_comm..Runnable.__lt__N)rr$)__name__ __module__ __qualname__r3r6rr.rr Runnables r:cSs"i|] }|tdd|jDqS)css|]}|jVqdSr%)r-).0deprrr sz0_schedule_for_comm...)rZunmet_dependenciesr;r&rrr sz&_schedule_for_comm..cSsi|]}|t|qSr)estimate_op_runtimer>rrr r?scsX||D] }|D]}||t|dkr(t|qq dS)zU Schedules `snode` and put all unblocked nodes onto the ready queue. rN)appendget_buffer_namesremovelenheapqheappush)r&buf_name)r: buffer_usersready scheduled unmet_depsrr schedules   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)r r&r r;xrrr szI_schedule_for_comm..get_overlapping_candidate..rNcSs|jSr%r4rNrrr szG_schedule_for_comm..get_overlapping_candidate..key)rDmin) candidates)rIrr get_overlapping_candidates  z5_schedule_for_comm..get_overlapping_candidatecsxt|sJ||}|dkr5}dur5||j||j8}|dkr5}dustdS)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 rCr&rEheapify)r&Zcollective_cost candidate)rVrIrL snode_to_costrr schedule_collective_for_overlaps   z;_schedule_for_comm..schedule_collective_for_overlapz;Detected unscheduled nodes. Nodes with unmet dependencies: ) enumeraterBr)r*sysmaxsizer ancestorsrTr rritemsrDrErFaddheappopr&)rrrrZbuf_name_to_snodeidxr&rGZop_name node_nameZcomm_idxZancZanc_fused_namedepsr<rZr) r:rHrVr/rIrLrJr0r1r2rYrKr rIsh&                   rnodescCsttjs|Sdd|D}tdt|D]"}tt||}||dD] }||t ||dq)q|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;nrrr rOsz3decide_global_ordering_of_comms..r mutating_buf) torch distributed is_availablerangerDr'r(rB add_fake_depr )re name_to_bufr/Z comm_nodesirhbufrrr decide_global_ordering_of_commss rqr&rfloatcCs2tjdkr |}|SttjsJt|}|S)z: Returns estimated op runtime in nanoseconds (ns) default)r r@Zget_estimated_runtimecallable)r&Zruntimerrr r@s  r@cCs~d}t|jtjrd|jjd}d}|j}t|tjr)d|jd|jd}|j p/d}|jj j ||d|dS)Nz ()z (size=z , stride=) isinstancenoder ZExternKernelOutZpython_kernel_nameZget_output_specZLayoutsizeZstrideZmaybe_get_name __class__r7)r&detailZout_tensor_infoZlayoutrcrrr node_summarys  r|cCsd}d}|D]N}|dur2t|r|t|7}|j}nt|jr#td|t|7}tt|qt|r:tdt|jrJtt|d}qtdt|qtd|dddS)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 r@rxrAssertionError overlap_logdebugr|)orderZtotal_est_runtimeZ cur_comm_noder&rrr visualize_overlaps0    rc Cs|}tjD]u}t|tr|tvrt|}tjdkrFt d|dzt |Wnt yE}z t t|WYd}~nd}~ww||}tjdkrzt d|dzt |Wqt yy}z t t|WYd}~qd}~wwq|S)Nrz.==== Visualize overlap before reordering pass z ====z-==== Visualize overlap after reordering pass ) r Z'reorder_for_compute_comm_overlap_passesrwstrglobalsrirjZget_rankr~rr Exception)rrperrr $reorder_compute_and_comm_for_overlap0s6      rgraphtorch.fx.Graphc s4t|jtttttD]E\}}|jdkrV|jtjjj j krV|j djdks9Jd|d|j dd|j d}|j d}|dkrO| |q| |qfdd }tt}tD]4\}}|jdkr|jtjj jj kr|}|j djdksJd d |d|r| |qgd d }ddD]'}|jdkrt|jtjjr|jjjr||s͈||rJd|dq|D]\} t| D]|\} } | }|j dusJ|j \} | d} | t| dkr| | dntd}| |}tfdd|Dr*Jdd|d|d|D])}|jdkrT|j vrT|jtjjj j krTtfdd|j D}||_ q,qq|D]\} t| D]\} } | }||qdq\D] }|jdkr|jtjjj j kr|j d|vr||qwdS)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. Z call_functionr placeholderz1Resize can only operate on graph inputs, but got z# which is resizing non-graph-input  rcs|g}|g}t|t|ks)td|dt|dt|ddSt||D]$\}}||krRtd|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)getrDlogwarningzip) 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_listrr check_resize_patternnsF   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%)targetriopsfsdpcopy_rsinductorresize_storage_bytes_rxrrr is_allowed_mutationszKremove_fsdp2_unsharded_param_graph_input_usage..is_allowed_mutationcsdtjtjjrddtjjjDng}tfdd|D}tdd|D}t ||@dkS)NcSs&g|]\}}|jdur|jjr|qSr%)Z alias_infoZis_write)r;rorNrrr rOs zyremove_fsdp2_unsharded_param_graph_input_usage..is_node_mutating_unsharded_param_or_its_alias..cs$g|]}tj|jdqSval)rargsmetauntyped_storage)r;rorrr rOscSsg|] }t|jdqSr)rrr)r;unsharded_paramrrr rOsr) rwrri_ops OpOverloadr[_schema argumentsrrD)rxZunsharded_paramsZmutated_arg_idxesZmutated_node_arg_storagesZstorages_of_unsharded_paramsrrr -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;rx)rrrr r=s  zAremove_fsdp2_unsharded_param_graph_input_usage..z(Assumed no ops mutating unsharded param z in subgraph z, but it's not true! Graph: c3s |] }|ur n|VqdSr%r)r;arg) replacementrrr r=s  N)listrerr[oprrirrrrsrrArrrwrrrZ is_mutablekeysr_rDanytuple erase_node)rrbrxrnew_sizerZ'unsharded_param_to_fsdp_copy_node_idxesZfsdp_copy_noderZfsdp_copy_node_idxesroZfsdp_copy_node_idx_Zsubgraph_start_idxZsubgraph_end_idxZsubgraph_nodesnew_argsr)rrrrrrr .remove_fsdp2_unsharded_param_graph_input_usageLs    %         $  rr$c szddljs JjjjrjjjsJWn ttt fy&YdSwddl m }m }m }m}m} fdd}|}||jjjj|tj|jjjj|d|d|d|d |d |d |d |d |d|d|ddddfdd }||||dS)Nrr) CallFunction KeywordArgMatchPatternMatcherPassregister_graph_patterncsTt|j}|D] }|jtjkr'|jdjjjjj ur'|jddkr'| |qdS)Nrr) rreroperatorgetitemrrrall_gather_copy_inrsr)grrfrirr remove_unused_getitem5s   z8reinplace_fsdp_all_gather..remove_unused_getitemall_gather_inputsinp_split_sizesall_gather_input_numel world_sizerankdtypedeviceitem_idx group_size group_namecSs|jddkS)Nrr)kwargs)matchrrr rQWsz+reinplace_fsdp_all_gather..)Z pass_dictZ extra_checkrrc sRfdd}|||d|d|d|d|d|d|d |d |d g dS) NcsX|dd}|d}|d}jjjj|}|d}|d}jjjj||||d}|S)Nrr)out)rrrrs_c10d_functionalall_gather_into_tensor_out)rZ copy_in_argsrrrrZ getitem_1all_gather_into_tensorrrr replZs   zEreinplace_fsdp_all_gather..reinplace_all_gather..replrrrrrrrrr)Zreplace_by_example)rrrrrrr reinplace_all_gatherBs z7reinplace_fsdp_all_gather..reinplace_all_gather)rr)Z5torch.distributed.fsdp._fully_shard._fsdp_collectivesrjrkrrrr ImportErrorAttributeErrorr}Zpattern_matcherrrrrrrsrrrrapply) rrrrrrrZ graph_passrrrr reinplace_fsdp_all_gathersP      "rcCs2t|tjjjtjjjfrJt|ddS)N)rwriZ _inductor schedulerZFusedSchedulerNodeGroupedSchedulerNodeintr*)r&rrr get_op_idxsr1list[torch._inductor.scheduler.BaseSchedulerNode]rn4dict[str, torch._inductor.scheduler.SchedulerBuffer]r/dict[str, BaseSchedulerNode]c sddlmg}tt}d}d}i}i}ifdd} |D]} 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|}nq}| d|} d}tt| dD]}t| |djtjr|d}nq|dusJ| | d|}| | |d}|||<q t| jtjjjj r5d }| }t}t|||t|dd d}d}tt|dD]}t||djtjr|d}nq|dus!J| |d|}| ||d}|||<q tdks?J|rKt|dksKJ|rWt|dksWJ|D]!} | vrh| } | |vroqY|| || qYd}|D]'\}}|durtt| }|!D]}|"t#||dq|}qd}|D]'\}}|durtt| }|!D]}|"t#||dq|}q|S)Nr)rFcs2j|}|D]}||<q||<|Sr%)rcreater*)Zsnodes_to_groupZ group_noder&)rsnode_name_to_final_snoderr _create_group_nodes  z:enforce_comm_ordering_for_fsdp.._create_group_node)rc3s(|]}t|jtjjjjVqdSr%)rrxrirrrrsrM)r/rr r=s z1enforce_comm_ordering_for_fsdp..Tcs&t|jpt|jo|jjv Sr%)rwZNopKernelSchedulerNodeZExternKernelSchedulerNoderxZ op_overloadrP) allowed_opsrrr rQs   z0enforce_comm_ordering_for_fsdp..)Z criteria_cbcSt|Sr%rrPrrr rQrRrcSrr%rrPrrr rQrrg)$rurrrrrxrirrrrsrr^rZ wait_tensorrZsplit_with_sizes_copyrsortedrDrlrrwr Z _WaitKernelZ chunk_catr*rAr`r_r'r(rBZ get_outputsrmr )rrnr/Z new_orderrJZ 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_countroZ 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_noderhoZ prev_rs_waitr)rr/rrr enforce_comm_ordering_for_fsdps                  r)rrrr) rrrr#rr#rr#rr)rerrr)r&rrrr)rr)rrrr$)rrrnrr/rrr)2 __future__rrEloggingrr\ collectionsrtypingrrriZ torch.multiprocessing.reductionsrZtorch.utils._ordered_setrrur r dependenciesr utilsr r rrrrr getLoggerr7rZ_loggingZgetArtifactLoggerr~rrrrr"rrqr@r|rrrrrrrrrr s@     $          Eo