tj UddlmZddlZddlZddlmZmZddlmZddl m Z m Z m Z m Z mZmZddlZddlmZddlmZddlmZdd lmZmZer dd lmZdd lmZejeZGd d e Z e!e!e"dfedzfZ#de$d<Gdde%Z&eddGddZ'eddGdde'Z(e'e(zZ)de$d<edGddZ*eddGddZ+eddGdd Z,djd&Z-dkd*Z.Gd+d,Z/Gd-d.eZ0d/d0dld3Z1d4Z2 dmdnd<Z3 dmdod>Z4edGd?d@Z5dpdBZ6e ddCdqdIZ7e drdMZ7d/dCdsdOZ7dtdudPZ8e dvdwdSZ9e dvdxdUZ9 dtdydWZ9d/d/dXdzd`Z:dd/dXd{dbZ;d|diZs_ fg & & )\ ,==??, .     r"devicetorch.device | strct||}|r||j|S)zReconstruct a tensor on ``device`` from this metadata. Args: device: Target device for the tensor. Returns: An empty strided tensor on ``device``. )_make_tensor_from_metais_floating_pointrequires_grad_r4)rr<ts r to_tensorz_TensorMeta.to_tensorVsC #4 0 0    1  T/ 0 0 0r"other list[str]cX||krgSg}|j|jkr%|d|jd|j|j|jkr%|d|jd|j|j|jkr%|d|jd|j|S)zReturn field-by-field differences with ``other``. Args: other: Metadata to compare against. Returns: List of human-readable difference strings (empty if equal). zshape mismatch:  vs zstride mismatch: zdtype mismatch: )r/appendr1r2rrDdiffss r get_diffz_TensorMeta.get_diffds 5==I : $ $ LLIDJIIEKII J J J ;%, & & LLLT[LLelLL M M M : $ $ LLIDJIIEKII J J J r"N)r5r6rr-)r<r=rr6rDr-rrE) r#r$r%r&__annotations__ staticmethodr;rCrKrr"r r-r-0s   \ .    r"r-ceZdZUdZedZded<edZded <edZd ed <edZ d ed <edZ ded<e d dZ e d!dZd"dZd#dZdS)$ _DTensorMetaaDTensor metadata extending :class:`_TensorMeta` with distribution info. Inherited fields (shape, stride, etc.) are LOCAL shard attributes. Additional fields capture global shape and placement information needed to reconstruct a :class:`DTensor` via ``DTensor.from_local()``. The :class:`DeviceMesh` is **not** stored (not serializable for P2P); it is looked up from :class:`_MeshCache` using ``(mesh_dim_names, mesh_layout)`` as the key. c*tjgSr)torchSizerr"r z_DTensorMeta.sUZ^^r")default_factoryr. global_shaper)defaultr0 global_strideztuple[Placement, ...] placementsrrNrrdtensorrrc |j}t|jj|j|j|j|j||jj|j rt|j nd|j  S)zCreate metadata from a DTensor. Args: dtensor: The DTensor to extract metadata from. Returns: Metadata capturing both local and global attributes. r) r/r1r2r4rVrXrYrr) device_meshrP _local_tensorr/r1r2r4_specrYrtuple_layout)rZr\s r from_dtensorz_DTensorMeta.from_dtensors) '-(//11-!/ !..**}/5@5OWk0111UW#+    r"r'c|j|jfS)z E$666 Yd.?YYUEWYY!U%888 \t/A\\uGZ\\%"222 SDOSSAQSS"e&::: _0C__I]__5#444 VT-=VV5CTVV LL< = = = r")rZrrrP)rr')r<r=rerrrrL)r#r$r%r&rrVrMrXrYrrrNrapropertyrdrkrKrr"r rPrP|sF   %u5K5KLLLLLLLL%*U2%6%6%6M6666).)))J ',eB&7&7&7N7777&+e'''K   \ 8777X7!!!!2******r"rP TensorMeta)r+cdeZdZUdZdZded<dZded<dZded<dZded<dd Z dd Z dd Z dS) _StageMetazPConsolidated tensor metadata for a pipeline stage's forward and backward passes.Ntuple[TensorMeta, ...] | Noneinputsoutputs$tuple[TensorMeta | None, ...] | None input_grads output_gradsrr3cdtd|j|j|j|jfDS)z)Check if any metadata field is populated.c3K|]}|duV dSrr).0vs r z%_StageMeta.has_any..s:   TM      r")anyrrrsrurvrcs r has_anyz_StageMeta.has_anysC  k4<1A4CTU      r"cd|j|jfD] }|rtd|DrdS!dS)z3Check if any input/output metadata is DTensor type.c3DK|]}|t|tVdSr)r9rP)ryms r r{z*_StageMeta.has_dtensors..s1MMQ1MZ<88MMMMMMr"TF)rrrsr|)rmetass r has_dtensorsz_StageMeta.has_dtensorssMk4<0  E MM%MMMMM ttur"c&|jduo|jduS)z-Check if forward metadata is fully populated.N)rrrsrcs r is_complete_for_forwardz"_StageMeta.is_complete_for_forwards{$&C4.derive_oneTsQ 4 a & & 4'8'     r"c3.K|]}|VdSrr)ryrrs r r{z%_derive_grad_metas..`s+551A555555r")rrnrr)r_)rrs @r _derive_grad_metasrKs<      5555 555 5 55r"cDeZdZdZdddZdd Zdd ZddZddZddZ dS) _MeshCachezCache for :class:`DeviceMesh` objects keyed by ``(mesh_dim_names, mesh_layout)``. Assumes all pipeline stages share the same rank tensor (true for TorchTitan-style frameworks where meshes derive from a common world). N get_mesh_cbGetMeshCallback | NonerNonec"i|_||_dSr)_cache _get_mesh_cb)rrs r __init__z_MeshCache.__init__js68 'r"keyr'rc||jvr |j|S|\}}|jtd|d|d|||}|td|d|d||j|<|S)aReturn a cached mesh, or create one via the callback. Args: key: Cache key ``(mesh_dim_names, mesh_layout)``. Returns: The ``DeviceMesh``. Raises: PipeliningMetadataError: If not cached and no callback provided. Nz+Mesh not found in cache for mesh_dim_names=z, mesh_layout=z`, and no get_mesh callback provided. Provide a get_mesh callback or use DTensors in static mode.z>Mesh lookup failed: callback returned None for mesh_dim_names=z6. Ensure all stages use meshes from the same universe.)rrr))rrrrres r get_meshz_MeshCache.get_meshns $+  ;s# #&)#   $)OnOO*OOO    == <)H"0HH@KHHH    C r"rec||j|<dS)zAdd a mesh to the cache.Nr)rrres r putz_MeshCache.puts Cr"tensorstuple[torch.Tensor | None, ...]c|D]Y}t|trB|j}|jrt |jnd}|j}||f}||jvr ||j|<ZdS)zJExtract and cache meshes from any :class:`DTensor` instances in *tensors*.rN)r9rr\rr_r`r)rrr5re dim_namesrrs r update_from_tensorsz_MeshCache.update_from_tensorss~ , ,F&'** ,):>:MUE$"5666SU "l  +.dk))'+DK$ , ,r"r3c||jvSrr)rrs r __contains__z_MeshCache.__contains__sdk!!r"intc*t|jSr)lenrrcs r __len__z_MeshCache.__len__s4;r"r)rrrr)rr'rr)rr'rerrr)rrrr)rr'rr3)rr) r#r$r%r&rrrrrrrr"r rrcs (((((    D     , , , ,""""      r"rc2eZdZdZdZdZed d Zd S) InferenceModeaPipeline-level metadata inference mode, determined collectively across all PP ranks. The mode is set by the schedule (not individual stages) because ``has_backward`` is only known at schedule creation time and all stages must agree to avoid P2P hangs. .. attribute:: STATIC All stages have sufficient metadata; runtime inference is skipped. .. attribute:: DYNAMIC At least one stage requires runtime metadata inference. staticdynamicrrpstage_has_backwardr3rc|sdS|sdS|sdS|j|jdSdS)a'Determine whether dynamic metadata inference is needed for a stage. Args: meta: Stage metadata from user-provided args. stage_has_backward: Whether a backward pass will be performed. Returns: ``True`` if dynamic inference is needed. TF)rrrurv)clsrrs r needs_dynamiczInferenceMode.needs_dynamicsf++-- 4  "" 5" 5   #t'8'@4ur"N)rrprr3rr3)r#r$r%r&STATICDYNAMIC classmethodrrr"r rrsH  FG[r"rFdetachrr3cnt|\}}|r d|D}t||}||fS|S)a;Flatten ``args`` into a list, optionally detaching tensors. Args: args: Nested arguments to flatten. detach: If ``True``, detach tensors while preserving ``requires_grad``. Returns: ``(new_args, flat_detached_args)`` when ``detach=True``; ``flat_args`` list otherwise. cg|]J}t|tjr,||jn|KSr)r9rRTensorrrAr4)ryas r z flatten_args..s[   !U\** AHHJJ % %ao 6 6 6   r"r)argsr flat_argstreespec flat_detachednew_argss r flatten_argsrs`'t,,Ix '      "-::&& r"c$t|dS)zHFlatten and detach. Deprecated: use ``flatten_args(args, detach=True)``.Tr)r)rs r flatten_args_detachrs T * * **r"looppp_sizer num_stagesstylestrdict[int, int]c8i}|dkrt|D] }||z||< nv|dkr]||zdkrtd|d|dd}t|D]+}|||<|dz|zdkr||zdzdkr|dz }&|dz},ntd |d |S) z Compute the stage id to rank mapping for either a looped or V-style schedule. Most commonly num_stages == pp_size * 2, but this function can be used to compute the mapping for any number of stages per rank. rrzrz num_stages z% must be evenly divisible by pp_size z for V scheduleszStyle z is not supported.)range ValueError)rrrmapping stage_index rank_indexs r generate_stage_to_rank_mappingrsG  ,, 9 9K#.#8GK  9 #  1 $ $hjhhwhhh   ,,  K#-GK a7*a//w&!+q00a a  ;%;;;<<< Nr"dict[int, list[int]]ct|||}i}|D])\}}||vrg||<|||*|D]}||S)a Compute the rank to stage id mapping for either a looped or V-style schedule. This function inverts the stage_to_rank_mapping to get which stages are assigned to each rank. Returns a dictionary mapping rank -> list of stage indices assigned to that rank. )ritemsrHvaluessort)rrr stage_to_rankrank_to_stagesstage_idrankstagess r generate_rank_to_stage_mappingr%s37JNNM,.N'--//..$ ~ % %#%N4 t##H----!'')) r"c2eZdZUdZded<ded<ded<dS) PipeInfoz> Captures information for a pipeline (`Pipe` object). zfx.Graphgraphrrr3has_loss_and_backwardNrrr"r rr?s<OOOOOOr"rr5ct|trt|St|S)aExtract metadata from a tensor. Handles both plain Tensor and DTensor correctly: DTensors are dispatched to ``_DTensorMeta.from_dtensor`` which captures local shard attributes plus global shape/placement info, while plain tensors use ``_TensorMeta.from_tensor``. Args: tensor: A plain tensor or DTensor. Returns: ``_TensorMeta`` for plain tensors, ``_DTensorMeta`` for DTensors. )r9rrPrar-r;r:s r extract_tensor_metarOs>&'""/((000&&v...r") allow_nonertuple[torch.Tensor, ...] | NonerLiteral[False]rqcdSrrrrs r extract_tensor_metasrcs %(Cr"&tuple[torch.Tensor | None, ...] | None Literal[True]rtcdSrrrs r rrks ,/3r"Atuple[torch.Tensor | None, ...] | tuple[torch.Tensor, ...] | Nonec|dSg}d}|D]V}t|tjr#|t |?d}|dW|s|rt dt |S)aExtract metadata from a tuple of tensors. Args: tensors: Tuple of tensors (may include ``None`` when ``allow_none=True``). allow_none: If ``True``, preserve ``None`` elements (for gradients). Returns: Tuple of ``TensorMeta``, or ``None`` if ``tensors`` is ``None``. Raises: PipeliningMetadataError: If ``None`` found and ``allow_none=False``. NFTz_None values are not allowed in tensor metadata tuples. Use allow_none=True for optional values.)r9rRrrHrr)r_)rrmetas_with_nonehas_nonerBs r rrss"t/1OH )) a & & )  " "#6q#9#9 : : : :H  " "4 ( ( ( (  ( % 7     ! !!r"c|r|n|}t|tr|S|S)uConvert a DTensor to its local shard, or return a plain tensor as-is. When ``detach=True``, the tensor is detached before conversion — this applies to both DTensors and plain tensors. Args: tensor: A tensor that may be a DTensor. detach: If ``True``, detach before ``to_local()`` to avoid redistribution during backward. Returns: The local tensor component. )rr9rto_local)r5rmaybe_detached_tensors r to_local_if_dtensorrsF06AFMMOOO6'110$--///  r"rCtorch.Tensor | tuple[torch.Tensor, ...] | list[torch.Tensor] | NonecdSrrrrs r validate_and_normalize_to_tuplers '*cr"Qtorch.Tensor | tuple[torch.Tensor | None, ...] | list[torch.Tensor | None] | NonecdSrrrs r rrs .1Sr"torch.Tensor | tuple[torch.Tensor, ...] | tuple[torch.Tensor | None, ...] | list[torch.Tensor] | list[torch.Tensor | None] | Nonec |dSt|tjr|fSt|ttfrt |D]_\}}||st d|dt|tjs(t d|dt|jd`t|trt|n|St dt|jd)aNormalize ``args`` to a tuple and validate that all elements are tensors. Args: args: A single tensor, tuple/list of tensors, or ``None``. allow_none: If ``True``, permit ``None`` elements (for gradients). Returns: Tuple of tensors, or ``None`` if ``args`` is ``None``. Raises: PipeliningMetadataError: On non-tensor values (or ``None`` when ``allow_none=False``). Nz Stage arg[zF] is None. Stage args must be tensors. Use kwargs for optional values.z ] has type zC. All stage args must be tensors. Use kwargs for non-tensor inputs.zuT{{{$>% a4PT::K^ a a a   r"raise_on_mismatchwarn_on_mismatchdescexpectedactualtorch.Tensor | TensorMetar rrEcrt|tjrt|}n|}t |t |urudt |jdt |jg}|rt |d|d|r(tj|d|ddtd|S| |}|r`|r't |dd ||r5tj|d d |dtd|S) al Compare expected metadata against actual tensor or metadata. This is the unified validation/comparison function that uses get_diff() from metadata classes. Works with both plain tensors and DTensors. For plain tensors: compares shape/stride/dtype/requires_grad. For DTensors: compares all properties including global shape and placements. Args: desc: Description for error/warning messages. expected: Expected tensor metadata (_TensorMeta or _DTensorMeta). actual: Actual tensor or metadata to compare against. raise_on_mismatch: If True, raise PipeliningMetadataError on mismatch. warn_on_mismatch: If True, issue a warning on mismatch. Returns: List of differences (empty if metadata matches). Raises: PipeliningMetadataError: If raise_on_mismatch=True and differences exist. ztype: expected , got : rz: Metadata type mismatch. z.. Using dynamically inferred metadata instead.r stacklevelz; z: Metadata mismatch. ) r9rRrrr r#r)warningswarn UserWarningrKjoin)rrrr r actual_meta type_diffrJs r validate_metadatars>&%,'')&11   H~~T+.... Yd8nn5 Y YT+=N=N=W Y Y   E)T*C*CYq\*C*CDD D   M@@9Q<@@@         k * *E   I)T*G*GTYYu5E5E*G*GHH H   M@@dii.>.>@@@      Lr",tuple[torch.Tensor | TensorMeta | None, ...]cjt|t|krV|dt|dt|}|rt||rtj|td|gSg}t t ||dD]\}\}} ||  || Z|d|d |d nd d | d nd }|rt||rtj|td||kt|d|d || ||} | | |S)a2Validate metadata for a tuple of tensors element-wise. Args: desc: Description prefix for error/warning messages. expected: Tuple of expected metadata (may include ``None`` for grads). actual: Tuple of actual tensors or metadata to compare against. raise_on_mismatch: If ``True``, raise on the first mismatch. warn_on_mismatch: If ``True``, issue warnings for mismatches. Returns: Aggregated list of difference strings. Raises: PipeliningMetadataError: If lengths differ or on mismatch. z : expected z tensors, got rrTstrictN[z ]: expected rmetadatar]r ) rr)rrrrziprHrextend) rrrr rmsg all_diffsr expactrJs r validate_tensors_metadatar,6s. 8}}F ##LL#h--LLs6{{LL  /)#.. .  : M#{q 9 9 9 9u I"3x#E#E#EFF  :C ;3;  ;#+??!??3;J??!$vv*?? ! 3-c222 > c;1====   S ! ! ! !NNaNNN  /-      r"rtuple[torch.Tensor, ...]gradsris_inputrc|rdnd}|d}|d}t|t|kr9td|d|dt|d|dt|d tt||d D]\}\}} |js6| 4td|d|d |d|d |dt | jd |jr.| ,tjd|d|d |d|d |d tdt|trR|jrK| It| ts4td|d|d |d|d |dt | jd d S)u/ Validate the args↔grads contract for static mode. Enforces four rules for each (arg, grad) pair: 1. len(args) must equal len(grads). 2. If arg.requires_grad is False, grad must be None. 3. If arg.requires_grad is True and grad is None, emit a warning (this is legal at pipeline boundaries but may indicate a bug). 4. If arg is a DTensor with requires_grad=True and grad is not None, grad must also be a DTensor. Args: stage_index: The stage index for error messages. args: Tuple of forward tensors. grads: Tuple of gradient tensors (can include None). is_input: True for input_args/input_grads, False for output_args/output_grads. Raises: PipeliningMetadataError: If any hard rule (1, 2, or 4) is violated. inputoutput_args_gradszStage rz length (z) does not match zo). Each forward tensor must have a corresponding gradient entry (use None for tensors that don't require grad).Tr!Nr#z] has requires_grad=False, but z] is not None (zE). Non-differentiable tensors must have None as their gradient entry.z] has requires_grad=True, but zT] is None. This is legal at pipeline boundaries but may indicate a missing gradient.rrz,] is a DTensor with requires_grad=True, but z] is za, expected DTensor or None. DTensor gradients may have different placements than forward tensors.) rr)rr&r4r r#rrrr9r) rrr.r/kind args_name grads_namer r grads r 'validate_static_arg_grad_correspondencer9os4 ,77HDIJ 4yyCJJ% \[ \ \J \ \U \ \ \ \#&t99 \ \ \   $CeD$A$A$ABB;C  T%5)VVV VVAVV!VV$%VV6:4jj6IVVV     M888 88A88!88$%888      sG $ $ !  tW--!*YYY YYAYY!YY$%YY,0JJ,?YYY 5r")rr-r<r=rr6)rrrr)rr3)r)rrrrrrrr)rrrrrrrr)r5r6rrn)rrrrrrq)rrrrrrt)rrrr3rrt)F)r5r6rr3rr6).)rrrrrr)rrrrrr)rrrr3rr) rrrrnrrr r3rr3rrE) rrrrrrr r3rr3rrE) rrrr-r.rr/r3rr)= __future__rloggingr dataclassesrrenumrtypingrrr r r r rRr torch.distributed._mesh_layoutrtorch.distributed.tensorrtorch.utils._pytreerrtorch.distributed.device_meshr(torch.distributed.tensor.placement_typesr getLoggerr#loggerrr_rr'rM RuntimeErrorr)r-rPrnrprrr?rrrrrrrrrrrrrr,r9rr"r rGs#""""""((((((((NNNNNNNNNNNNNNNN 666666,,,,,,<<<<<<<<C888888BBBBBB  8 $ $h c3ht1C CD DDDDGGGGGlGGG $d###HHHHHHH$#HV $d###@@@@@;@@$#@H$l2 2222  DDDDDDDD6 $d###*******$#*  $d###       $# *66660@ @ @ @ @ @ @ @ P00000D000p*/6+++ 17F174         ////( "%((((( ( /// /!"!"!"!"!"!"H!!!!!( "%**** *  !$ 1111 1 - - - - - t$" BBBBBBT#" 666666rDDDDDDr"