a,j~ 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,dkd&Z-dld*Z.Gd+d,Z/Gd-d.eZ0d/d0dmd3Z1d4Z2 dndod<Z3 dndpd>Z4edGd?d@Z5dqdBZ6e ddCdrdIZ7e dsdMZ7d/dCdtdOZ7dudvdPZ8e dwdxdSZ9e dwdydUZ9 dudzdWZ9d/d/dXd{d`Z:dd/dXd|dcZ;d}djZs_ fg & & )\ ,==??, .     r"devicetorch.device | strcZt||}||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_metarequires_grad_r4)rr<ts r to_tensorz_TensorMeta.to_tensorVs/ #4 0 0 +,,,r"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/appendr1r2rrCdiffss r get_diffz_TensorMeta.get_diffcs 5==I : $ $ LLIDJIIEKII J J J ;%, & & LLLT[LLelLL M M M : $ $ LLIDJIIEKII J J J r"N)r5r6rr-)r<r=rr6rCr-rrD) r#r$r%r&__annotations__ staticmethodr;rBrJrr"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/r1r2r4rUrWrXrr) device_meshrO _local_tensorr/r1r2r4_specrXrtuple_layout)rYr[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")rYrrrO)rr')r<r=rdrrrrK)r#r$r%r&rrUrLrWrXrrrMr`propertyrcrirJrr"r rOrO{sF   %u5K5KLLLLLLLL%*U2%6%6%6M6666).)))J ',eB&7&7&7N7777&+e'''K   \ 8777X7    2******r"rO 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")anyrprqrsrtrbs 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)r9rOrwms r ryz*_StageMeta.has_dtensors..s1MMQ1MZ<88MMMMMMr"TF)rprqrz)rmetass r has_dtensorsz_StageMeta.has_dtensorssMk4<0  E MM%MMMMM ttur"c&|jduo|jduS)z-Check if forward metadata is fully populated.N)rprqrbs r is_complete_for_forwardz"_StageMeta.is_complete_for_forwards{$&C4.Rs_  ?  !'!(!'QVWWWW r")r^)rs r _derive_grad_metasrJs4    r"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__as68 '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))rrrrrds r get_meshz_MeshCache.get_meshes $+  ;s# #&)#   $)OnOO*OOO    == <)H"0HH@KHHH    C r"rdc||j|<dS)zAdd a mesh to the cache.Nr)rrrds 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)rrr5rd 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)lenrrbs r __len__z_MeshCache.__len__s4;r"r)rrrr)rr'rr)rr'rdrrr)rrrr)rr'rr3)rr) r#r$r%r&rrrrrrrr"r rrZs (((((    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. staticdynamicrrnstage_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)rrrsrt)clsrrs r needs_dynamiczInferenceMode.needs_dynamicsf++-- 4  "" 5" 5   #t'8'@4ur"N)rrnrr3rr3)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)r9rQTensorrr@r4)rwas 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. rrxrz 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. )ritemsrGvaluessort)rrr stage_to_rankrank_to_stagesstage_idrankstagess r generate_rank_to_stage_mappingrs37JNNM,.N'--//..$ ~ % %#%N4 t##H----!'')) r"c2eZdZUdZded<ded<ded<dS) PipeInfoz> Captures information for a pipeline (`Pipe` object). zfx.Graphgraphrrr3has_loss_and_backwardNrrr"r rr6s<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. )r9rrOr`r-r;r:s r extract_tensor_metarFs>&'""/((000&&v...r") allow_nonertuple[torch.Tensor, ...] | NonerLiteral[False]rocdSrrrrs r extract_tensor_metasrZs %(Cr"&tuple[torch.Tensor | None, ...] | None Literal[True]rrcdSrrrs r rrbs ,/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.)r9rQrrGrr)r^)rrmetas_with_nonehas_nonerAs r rrjs"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 r rDcrt|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. ) r9rQrrrr#r)warningswarn UserWarningrJjoin)r r rr r  actual_meta type_diffrIs 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"r,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)rrrrziprGrextend) r r rr r msg all_diffsrexpactrIs r validate_tensors_metadatar)-s. 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).TrNr 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#r4rr#rrrr9r) rrr+r,kind args_name grads_namerrgrads r 'validate_static_arg_grad_correspondencer6fs4 ,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)r5r6rrl)rrrrrro)rrrrrrr)rrrr3rrr)F)r5r6rr3rr6).)rrrrrr)rrrrrr)rrrr3rr) r rr rlrrr r3r r3rrD) r rr rrrr r3r r3rrD) rrrr*r+rr,r3rr)= __future__rloggingr dataclassesrrenumrtypingrrr r r r rQr torch.distributed._mesh_layoutrtorch.distributed.tensorrtorch.utils._pytreerrtorch.distributed.device_meshr(torch.distributed.tensor.placement_typesr getLoggerr#loggerrr^rr'rL RuntimeErrorr)r-rOrlrnrrr?rrrrrrrrrrrrrr)r6rr"r rDs#""""""((((((((NNNNNNNNNNNNNNNN 666666,,,,,,<<<<<<<<C888888BBBBBB  8 $ $h c3ht1C CD DDDDGGGGGlGGG $d###GGGGGGG$#GT $d###@@@@@;@@$#@H$l2 2222  DDDDDDDD6 $d###*******$#*  $d###       $# *     @ @ @ @ @ @ @ @ P00000D000p*/6+++ 17F174         ////( "%((((( ( /// /!"!"!"!"!"!"H!!!!!( "%**** *  !$ 1111 1 - - - - - t$" BBBBBBT#" 666666rDDDDDDr"