o die@sTddlmZddlmZmZddlmZmZmZddl Z ddl m Z ddl mZddlmZmZmZddlmZmZmZddlmZdd l mZdd lmZmZdd lmZmZm Z m!Z!m"Z"dd l#m$Z$m%Z%m&Z&m'Z'm(Z(Gd ddeZ)Gddde)Z*Gddde*Z+Gddde*Z,Gddde)Z-Gddde)Z.dddZ/GdddZ0dS)) annotations)ABCabstractmethod)AnyDictListN)Graph GraphModuleNode)ConfigParallelExecutionCtx ParameterMeta)decompose_and_functionalize)scatter)REGISTRY$FallbackParallelAxisPropagateHandler)ColumnParallelLinearRowParallelLinearVocabParallelCrossEntropyLossVocabParallelEmbedding sharded_cross_entropy_wrapper_fn)is_cross_entropy is_embedding is_linearis_shape_consumerstable_topological_sortc@sDeZdZUdZdZded<edddZedddZ dddZ dS)PassBasez9 Base class for parallelization targeted passes. Tboolneed_rerun_when_recompilereturnstrcCs|jSN)__name__clsr&o/lsinfo/ai/hellotax_ai/llm_service/venv_embed/lib/python3.10/site-packages/optimum/fx/parallelization/passes.py signature4szPassBase.signature graph_moduler ctxr configr cCst)a Args: graph_module (`GraphModule`): graph module before processing. ctx (`ParallelExecutionCtx`): dynamic execution context which gathers and preserves information along processing. config (`Config`): static config to include instructions which persists the whole process. Returns: GraphModule: graph module after processed by the current pass. )NotImplementedErrorselfr)r*r+r&r&r'run8sz PassBase.runcCs@|js |jdkr |S|j|||d}|jr|j||S)Nr)r*r+)r compile_timesr/lint_and_recompilegraphlint recompiler-r&r&r'__call__Hs zPassBase.__call__Nr r!r)r r*r r+r r r ) r# __module__ __qualname____doc__r__annotations__ classmethodr(rr/r5r&r&r&r'r-s    rc@sZeZdZdZedddZeddd dZed ddZd!ddZd"ddZ d#ddZ dS)$ AnalyzeBasea= Base class for passes which only runs for analytical purposes and preserve graph structure during processing. Analytical passes are often prerequisite passes which provide information for passes later on to actually change the graph. Passes inheriting from `AnalyzeBase` places the class signature as a meta key in `node.meta`, which is a dict storing meta information related with a fx Node, such as the shape and dtype of output. Look-up APIs are exposed as classmethod so that passes using them won't need to create concrete instances. r r!cCs|Sr")r(r$r&r&r'meta_key`szAnalyzeBase.meta_keyFnoder fieldr must_haverc Cs|||s|s dStd|jd|d|jd|j|}||vr:|r8td|d|jdt|dS||S)Nz$Can't find information related with z in the current node `z ` make sure z has run and marked itzInvalid query field z for z, valid fields are )already_executed_per_node RuntimeErrorr#metar>KeyErrorlistkeys)r%r?r@rAinfor&r&r'get_stored_field_infoes $z!AnalyzeBase.get_stored_field_infocCs||jvSr")r>rD)r%r?r&r&r'rBxsz%AnalyzeBase.already_executed_per_noderHDict[Any, Any]NonecCs,||r td|d||j|<dS)NzNode zq has already been marked by the current pass, check if the current pass has already been executed in the pipeline)rBrCrDr>)r.r?rHr&r&r'place_marker_per_node|s  z!AnalyzeBase.place_marker_per_nodecCs&|}||jvr|j|dSdSr")r>rDpop)r.r?keyr&r&r'clear_marker_per_nodes z!AnalyzeBase.clear_marker_per_noder)r cCs |j}|jD]}||qdSr")r2nodesrO)r.r)gr?r&r&r' clean_alls  zAnalyzeBase.clean_allNr6)F)r?r r@rrArr r)r?r r r)r?r rHrJr rK)r?r r rK)r)r r rK) r#r8r9r:r<r>rIrBrLrOrRr&r&r&r'r=Ts    r=c@s$eZdZdZddd ZdddZdS)ParallelAxisSolverPassaF A pass which tries to automatically identify parallel layers in the graph. There are three steps involved to find a possible parallel solution given the traced graph module and process group. - Decompostion & Functionalization The vanilla graph traced by dynamo frontend is a high-level graph which contains high-level pytorch ops, and there could be thousands of them, which makes graph analysis hard in order to cover all cases. So we decompose the high-level graph into low-level graph which only conrtains core aten ops, which is a much smaller set. And functionalization is also needed to remove inplace ops in the graph so that we get `aten.Add` instead of `aten.Add_` in the graph, which furthur reduces the op set we need to consider. - Parallel Axis Propagation We need to write parallel axis propagation rules for aten ops in the decomposed and functionalized graph, note that we don't need to cover every possible parallelization strategy because in general only certain ops(usually involves computation) can be parallelized in transformer models. And we just need to write rules for a subset of core aten op set in order to support most of the transformer models. - Backtracking Search After we have defined parallel axis propagation rules for each op in the graph, we do a brute force backtracking search to try to find a possible solution which respects the propagation rule of every op in the graph. Note that there are several practical concerns - Time Complexity. Although brute force backtracking introduces an exponential time complexity, we reduces the search space by injecting human heuristics. First, we only consider parallelization on the head dimension (for tensor parallel) or the sequence dimension(to support sequence parallel), then at any time the tensor is parallelized on at most one dimension. Second, we only allow axis switch around certain layers(like `nn.Linear` or `nn.Embedding), and all other ops fall into their places by the parallel axis of their input and rules we write. - Optimal Solution. Note that since we return the first solution we find, then it might not be optimal in terms of memory consumption and communication overhead. But again we can adjust the order of search and try parallelize as much as we can first before fall back to non-parallelized search paths. And we don't pay too much attention on calculating communication overhead because in practice they are bounded under the constraint that only certain layers are allowed to communicate. Our goal is not to solve an optimization problem which tries to give a best solution of parallelizing any model under memory/hardware constraints, but rather a cheap solution which relieves you from writing boilerplate code for parallelizing layers of different models. r)r decomp_graphrr rKc Csdd|jjD}|jD]2}d|jvr>|jdd\}}||vs(Jd|d||}||||d|j|ddiq dS) NcSsi|]}|j|qSr&)name).0r?r&r&r' sz5ParallelAxisSolverPass.trace_back.. traced_fromrzun-recognized node origin z not in graph being traced parallel_axisr@)r2rPrDrOrLrI)r.r)rTnode_mapr? node_name_ orig_noder&r&r' trace_backs   z!ParallelAxisSolverPass.trace_backr*r r+r csTt||j}t|t|jdfdd ds"td|||S)Nidxintcs|tkrdS|}|jdkrt|jrtj|j}nt}||}|}|D]} |d|i|drAdS |q.dS)NT call_functionrYr F) lenopr is_supportedtargetmappingrr> propagaterLrO)r`r?prop_clspropaxis_candidatesaxisr+rPsearchr.r&r'rns   z*ParallelAxisSolverPass.run..searchrzRFailed to find a solution to automatically parallelize ops in graph in greedy way.)r`ra)rexample_inputsrrFrPrCr_)r.r)r*r+r2r&rmr'r/s  zParallelAxisSolverPass.runN)r)r rTrr rKr7)r#r8r9r:r_r/r&r&r&r'rSs * rSc@eZdZdZd d d Zd S) ParallelLayerAnnotatePassa This pass annotates layers which have different parallel axis(requires communication inside the layer) in their input and output tensors. Since heuristics applied during the searching process respect traditional classical ways of parallelizing layers(like Megatron-style `ColumnLinear` or `RowLinear`), we are guaranteed to match a valid replacement annotation according to parallelization strategy of input and output tensors. r)r r*r r+r r cCs|jjD]}t|rst|jdd}t|d}i}|dur.d|d<|dur)dnd|d<n>|dkrL|js9Jd d|d<d|d <|durGdnd|d<n |d krld |d<d|d <|dkrh|jscJd d|d <nd|d <|||qt|rt|jdd}t|d}|dur|dvsJddi}|dkr|jsJd d|d <nd|d <|||qt |rt|jdd}|dur||ddiq|S)NrrYcolumnrlTF gather_outputr zBillegal parallel axis for sequence parallelism deactivated settingsequence_parallelrowinput_is_parallel)r Nvocab) r2rPrrSrIargsenable_sequence_parallelrLrr)r.r)r*r+r? axis_before axis_afterrHr&r&r'r/s\     zParallelLayerAnnotatePass.runNr7r#r8r9r:r/r&r&r&r'rqsrqc@sReZdZdZeddd Zedd d Zedd d ZedddZdddZ dS)ParallelLayerReplacePassa^ A pass which modifies graph according to information provided by previous analytical passes, in general it does two things for now: 1. replaces linears and embedding layers with their parallel counterparts. 2. modifies hard-coded arguments like the number of attention heads in the graph by dividing it by parallelism level. r?r r*r r rKc Cs|jj}tj|dd}|durdS|dvsJ|jjddd}t|dkr1||d}|d}n|}|j}||j}|j|j}} || vrL| |} n%|d kr_tj|d d d } t ||| } ntj|d d d } t ||| } | | |<t ||| dS)NrlrZ>rvrr.r maxsplitrurrrrsT)r@rArw) r2 owning_modulerqrIrfrsplitrc get_submoduleparallel_layer_cacherrsetattr) r?r*r)rlprefix_and_field parent_modr@modrN layer_cachenew_modrsrwr&r&r' handle_linear*s4      z&ParallelLayerReplacePass.handle_linearc Cs|jj}tj|dd}|durdS|dvsJd|jjddd}t|dkr3||d }|d}n|}|j}||j}|j|j}} || vrN| |} n|j d ksWJd t ||} | | |<t ||| dS) NrlrZrx2Only support parallelization on vocab dim for now.rr rrurillegal path for recompilation) r2rrqrIrfrrcrrr0rr) r?r*r)rlrrr@rrNrrr&r&r'handle_embeddingLs&     z)ParallelLayerReplacePass.handle_embeddingc Cstj|dd}|dur dS|dvsJd|jdkro|jj}|jjddd}t|d kr8||d }|d}n|}|j}||j}|j|j }} || vrS| |} n|j d ks\Jd t ||j d } | | |<t ||| dSt|jd |_dS)NrlrZrr call_modulerr rrurr) reduction) process_group)rqrIrdr2rrfrrcrrr0rrrrtp_group) r?r*rlr)rrr@rrNrrr&r&r'handle_cross_entropyfs*     z-ParallelLayerReplacePass.handle_cross_entropycsddd}dfd d }tj|dd }|durdS|||tks%Jt|tr2|d kr4dS|j}||dksCJ|||<|||dS)Nr?r r List[Any]cSs`d|jvr t|jdSd|jvrt|jdSt|jdtr't|jdSt|jddSNsizeshaper )kwargsrF isinstancerytuple)r?r&r&r'extract_shape_from_nodes  zVParallelLayerReplacePass.handle_hard_coded_axis_param..extract_shape_from_node new_shaperYracsxd|jvr|dt|dSd|jvr|dt|dSt|jdtr0|dt|dS||d|dSr)r update_kwargrrry update_arg)r?rrYrr&r'updates  zEParallelLayerReplacePass.handle_hard_coded_axis_param..updaterZr)r?r r r)r?r rrrYra)rSrIrcrrarr)r?r*rrrY world_sizer&rr'handle_hard_coded_axis_params    z5ParallelLayerReplacePass.handle_hard_coded_axis_paramr)r r+r cCsh|jjD]-}t|r|||qt|r|||qt|r'|||qt|r1| ||q|Sr") r2rPrrrrrrrr)r.r)r*r+r?r&r&r'r/s  zParallelLayerReplacePass.runN)r?r r*r r rKr7) r#r8r9r: staticmethodrrrrr/r&r&r&r'r~#s !   "r~c@rp) InitializeOrLoadWeightsPassz Weights loading and intialization pass, will initialize parameters on current rank and load weights from disk if necessary. r)r r*r r+r r c st|jt|j}gi|j}}}t|jddD]s\}||vr0||||fqtdj rJt |vrJ|||t fqfddt j D} j sdj|jkrd} ntjtj| j|jdjd} tjD]u\j|jvrdd lm} | |jjd d d R} | j} fd dt j D}fddt j D}| |}t||}t | j!||Wdn1swYWdn1swYq|j"rqtjD]s\j|jvr qj r|dkr2tj#j$jd d}j%r$j%n|j&}|||'|j}nd}fddt j D}t "j rXt(|j|| j!|j)dn| j!||Wdn 1skwYqt*| dt | t kr||| fj r| |t <q|D]2\}} |j+ddd}t,|dkr|-|d}|d}n|}|}||vr| ||<t*||| q|S)NF)remove_duplicaterDcs4g|]}|jkrjr|n|qSr&)dim is_parallelrrVr)param param_metarr&r' s&z3InitializeOrLoadWeightsPass.run..)dtypedevice) requires_gradr) safe_openptcpu) frameworkrc*g|]}|jkr ntdddqSr"rto_sliceslicerrsourcer&r'rcs(g|]}|jkr jntdddqSr")rindexrr)rrfr&r'rscrr"rrrr&r'rr) scatter_dimrr rru).distget_world_sizerget_rank param_cachesortednamed_parametersappendgetattris_tiedidrangendimrrcurrent_devicenn Parametertorchzerosrrrgitemsr weight_map safetensorsr get_slice contiguous empty_likecopy_no_graddataneed_initializeemptyrinit_fnweight_init_fntorrrrrcr)r.r)r*r+tp_ranknew_parameterstied_parametersrrUr new_paramrfp tensor_slice source_index load_indextensorweightrrrrr@r&)rrrrfrr'r/s                 zInitializeOrLoadWeightsPass.runNr7r}r&r&r&r'rsrr PassPipelinecCstttttgS)a8 Ensemble a pass pipeline which contains the following passes: 1. `ParallelAxisSolverPass` to find a parallelization solution of tensors in the graph. 2. `ParallelLayerAnnotatePass` to annotate parallelized layers according to the solution found in the first step. 3. `ParallelLinearReplacePass` to do the actual replacement and modification of hard-coded attributes. 4. `InitializeOrLoadWeightsPass` to load or initialize weights for parameters. Returns: PassPipeline: the pipeline used for automatic parallelism. )rrSrqr~rr&r&r&r'build_parallel_pass_pipelines rc@s:eZdZdZgfdddZdd Zdd d ZdddZdS)rz `PassPipeline` ensembles a list of passes and execute them one by one as provided in the list, it can be iterated and appended after initialization for flexibility. passesList[PassBase]r rKcCs ||_dSr")_passes)r.rr&r&r'__init__4s zPassPipeline.__init__cCs |jSr")r__iter__)r.r&r&r'r7s zPassPipeline.__iter__PASSrcCs|j|dSr")rr)r.rr&r&r'r<szPassPipeline.appendr)r r*r r+r cCsD|jD] }||||d}q|jr |jD] }t|tr||q|S)N)r)r*r+)rclean_markers_after_all_passesrr=rR)r.r)r*r+rr&r&r'r5?s    zPassPipeline.__call__N)rrr rK)rrr rKr7)r#r8r9r:rrrr5r&r&r&r'r.s  )r r)1 __future__rabcrrtypingrrrrtorch.distributed distributedrtorch.nnrtorch.fxrr r corer r rdecomprr op_registryrrparallel_layersrrrrrutilsrrrrrrr=rSrqr~rrrr&r&r&r's,      '<Y: e