^jUddlmZddlZddlZddlmZmZddlmZddl m Z m Z m Z m Z mZmZddlZddlmZddlmZddlmZdd lmZmZer dd lmZdd lmZej:eZGd d e Z e!e!e"dfedzfZ#de$d<Gdde%Z&eddGddZ'eddGdde'Z(e'e(zZ)de$d<edGddZ*eddGddZ+eddGdd Z, dd)Z1d*Z2 d? d@d+Z3 d? dAd,Z4edGd-d.Z5dBd/Z6e dd0 dCd1Z7e dDd2Z7d'd0 dEd3Z7dFdGd4Z8e dH dId5Z9e dH dJd6Z9 dF dKd7Z9d'd'd8 dLd9Z:dd'd8 dMd:Z; dNd;ZsM fg &)\ ,,==?,, ..   rctt||}|jr|j|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_r3)rdevicets r to_tensorz_TensorMeta.to_tensorVs5 #4 0    T// 0rc||k(rgSg}|j|jk7r+|jd|jd|j|j|jk7r+|jd|jd|j|j|jk7r+|jd|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.appendr0r1rotherdiffss rget_diffz_TensorMeta.get_diffds 5=I :: $ LL+DJJtorch.device | strr!rHrEr,r! list[str]) r"r#r$r%__annotations__ staticmethodr9r@rGrrrr,r,0s=     . rr,ceZdZUdZedZded<edZded <edZd ed <edZ d ed <edZ ded<e ddZ e ddZddZddZy) _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)torchSizerrrz_DTensorMeta.sUZZ^r)default_factoryr- global_shaper)defaultr/ global_strideztuple[Placement, ...] placementsrrNr rc p|j}t|jj|jj |j |j |j|j |jj|jrt|jnd|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.r0r1r3rUrWrXrr) device_meshrO _local_tensorr.r0r1r3_specrXrtuple_layout)dtensorrZs r from_dtensorz_DTensorMeta.from_dtensors)) ''--((//1--!// !..*}}//5@5O5Ok001UW#++  rc2|j|jfS)zmesh local_tensordts r to_dtensorz_DTensorMeta.to_dtensorsw.dF;    ##%%     ""6"6"8""4(BGR  rc0||k(rgStj||}t|trV|j|jk7r+|j d|jd|j|j |j k7r+|j d|j d|j |j|jk7r+|j d|jd|j|j|jk7r+|j d|jd|j|j|jk7r+|j d|jd|j|S|j d|S)zReturn field-by-field differences, including DTensor-specific fields. Args: other: Metadata to compare against. Returns: List of human-readable difference strings (empty if equal). zglobal_shape mismatch: rBzglobal_stride mismatch: zplacements mismatch: zmesh_dim_names mismatch: zmesh_layout mismatch: z!type: _DTensorMeta vs _TensorMeta) r,rGr6rOrUrCrWrXrrrDs rrGz_DTensorMeta.get_diffs 5=I $$T51 e\ *  E$6$66 -d.?.?-@UEWEWDXY!!U%8%88 .t/A/A.B$uGZGZF[\%"2"22 +DOO+d5CTCTBUV   LL< = r)r_rr!rO)r!r&)r>rIrgrr!rrJ)r"r#r$r%rrUrLrWrXrrrMr`propertyrcrjrGrrrrOrO|s  %5KLL*L%*2%6M?6).)J% ',B&7NO7&+'K#  877!2*rrO TensorMeta)r*cbeZdZUdZdZded<dZded<dZded<dZded<d d Z d d Z d d Z y) _StageMetazPConsolidated tensor metadata for a pipeline stage's forward and backward passes.Ntuple[TensorMeta, ...] | Noneinputsoutputs$tuple[TensorMeta | None, ...] | None input_grads output_gradsc~td|j|j|j|jfDS)z)Check if any metadata field is populated.c3$K|]}|du ywrr).0vs r z%_StageMeta.has_any..s  TM s)anyrqrrrtrurbs rhas_anyz_StageMeta.has_anys9 kk4<<1A1A4CTCTU   rcl|j|jfD]}|std|Dsyy)z3Check if any input/output metadata is DTensor type.c3BK|]}|st|tywr)r6rO)rxms rrzz*_StageMeta.has_dtensors..sMQ1Z<8MsTF)rqrrr{)rmetass r has_dtensorsz_StageMeta.has_dtensorss6kk4<<0 EM%MM rc>|jduxr|jduS)z-Check if forward metadata is fully populated.N)rqrrrbs ris_complete_for_forwardz"_StageMeta.is_complete_for_forwards{{$&C4<)rQ empty_stridedr.r0r1)metar>s rr;r;6s/    ZZ{{jj  rc4ddtfd|DS)zDerive gradient metadata from tensor metadata. Returns metadata with the same shape/stride/dtype but ``requires_grad=False``. Entries where the source has ``requires_grad=False`` become ``None``. c|jsyt|tryt|j|j |j dS)NFr5)r3r6rOr,r.r0r1)rs r derive_onez&_derive_grad_metas..derive_oneTs> a &''88''   rc3.K|] }|ywrr)rxrrs rrzz%_derive_grad_metas..`s51A5s)rrmr!zTensorMeta | None)r]) tensor_metasrs @r_derive_grad_metasrKs   5 5 55rcBeZdZdZd d dZd dZd dZd dZddZddZ y) _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). Nc i|_||_yr)_cache _get_mesh_cb)r get_mesh_cbs r__init__z_MeshCache.__init__js68 'rc||jvr|j|S|\}}|jtd|d|d|j||}|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. z+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()rkeyrrrgs rget_meshz_MeshCache.get_meshns $++ ;;s# #&)#    $)=n=MN*m,NO    = <)""0!1 }MGH    C rc"||j|<y)zAdd a mesh to the cache.Nr)rrrgs rputz_MeshCache.puts Crc|D]p}t|ts|j}|jrt |jnd}|j }||f}||j vsb||j |<ry)zJExtract and cache meshes from any :class:`DTensor` instances in *tensors*.rN)r6rrZrr]r^r)rtensorsr8rg dim_namesrrs rupdate_from_tensorsz_MeshCache.update_from_tensorssn ,F&'*)):>:M:ME$"5"56SU "ll  +.dkk)'+DKK$ ,rc||jvSrr)rrs r __contains__z_MeshCache.__contains__sdkk!!rc,t|jSr)lenrrbs r__len__z_MeshCache.__len__s4;;rr)rzGetMeshCallback | Noner!None)rr&r!r)rr&rgrr!r)rtuple[torch.Tensor | None, ...]r!r)rr&r!r2)r!int) r"r#r$r%rrrrrrrrrrrcs& ( D  ," rrc*eZdZdZdZdZeddZy) 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. staticdynamicc|jsy|jsy|sy|j |jyy)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)rrrtru)clsrstage_has_backwards r needs_dynamiczInferenceMode.needs_dynamicsM++-  ""    #t'8'8'@rN)rrorr2r!r2)r"r#r$r%STATICDYNAMIC classmethodrrrrrrs% FGrrFdetachct|\}}|rb|Dcgc]G}t|tjr)|j j |j n|I}}t||}||fS|Scc}w)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. )rr6rQTensorrr=r3r)argsr flat_argstreespeca flat_detachednew_argss r flatten_argsrs't,Ix   !U\\* HHJ % %aoo 6   "-:&&  sA A5ct|dS)zHFlatten and detach. Deprecated: use ``flatten_args(args, detach=True)``.Tr)r)rs rflatten_args_detachrs T **rci}|dk(rt|D] }||z||< |S|dk(rU||zdk7rtd|d|dd}t|D])}|||<|dz|zdk(r||zdzdk(r|dz }%|dz}+|Std |d ) 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. loopryrz num_stages z% must be evenly divisible by pp_size z for V scheduleszStyle z is not supported.)range ValueError)pp_size num_stagesstylemapping stage_index rank_indexs rgenerate_stage_to_rank_mappingrsG  , 9K#.#8GK  9( N% #  1 $j\)NwiWgh   , K#-GK a7*a/w&!+q0a a   N6%(:;< list of stage indices assigned to that rank. )ritemsrCvaluessort)rrr stage_to_rankrank_to_stagesstage_idrankstagess rgenerate_rank_to_stage_mappingr%s37JNM,.N'--/.$ ~ %#%N4 t##H-. !'')  rc0eZdZUdZded<ded<ded<y) PipeInfoz> Captures information for a pipeline (`Pipe` object). zfx.Graphgraphrrr2has_loss_and_backwardNrrrrrr?s OOrrcvt|trtj|Stj |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. )r6rrOr`r,r9r7s rextract_tensor_metarOs0&'"((00&&v..r) allow_nonecyrrrrs rextract_tensor_metasrcs %(rcyrrrs rrrks ,/rc|yg}d}|D]J}t|tjr|jt |8d}|jdL|s |r t 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.)r6rQrrCrr(r])rrmetas_with_nonehas_noner?s rrrss"/1OH ) a &  " "#6q#9 :H  " "4 ( ) (% 7    !!rcn|r|jn|}t|tr|jS|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. )rr6rto_local)r8rmaybe_detached_tensors rto_local_if_dtensorrs406FMMO6'1$--//  rcyrrrrs rvalidate_and_normalize_to_tuplers'*rcyrrrs rrrs.1rc |yt|tjr|fSt|ttfrt |D]X\}}||st d|dt|tjr5t d|dt|jdt|tr t|S|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.z$>%J4PT:K^K^J__` a  rraise_on_mismatchwarn_on_mismatchc*t|tjr t|}n|}t |t |urmdt |j dt |j g}|rt |d|d|r%tj|d|ddtd|S|j|}|rT|rt |dd j||r1tj|d d j|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. ) r6rQrrrr"r(warningswarn UserWarningrGjoin)descexpectedactualrr actual_meta type_diffrFs rvalidate_metadatar s0>&%,,')&1   H~T+..d8n556fT+=N=W=W-?@?@   Lrct|t|k7rJ|dt|dt|}|r t||rtj|td|gSg}t t ||dD]\}\}} || || R|d|d|d nd d | d nd }|r t||rtj|td|j|dt|d|d || || } |j| |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 rrTstrict[z ]: expected rmetadatar]r) rr(rrrrziprCr extend) rr r rrmsg all_diffsrexpactrFs rvalidate_tensors_metadatar6s>. 8}F #k#h-s6{mL )#. .  MM#{q 9u I"3x#EF :C ;3;  ;#+&!L3;J(OP!$v*=? !-c22 c;1=   S ! !fAaSN  /-   + , rc|rdnd}|d}|d}t|t|k7r-td|d|dt|d|dt|d tt||d D]\}\}} |js3| 1td|d|d |d|d |dt | j d |jr.| ,tjd|d|d |d|d |d tdt|ts|js| t| trtd|d|d |d|d |dt | j d y )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).TrNrz] 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(rrr3rr"rrrr6r) rrgradsis_inputkind args_name grads_namerrgrads r'validate_static_arg_grad_correspondencer&os47HD&I6J 4yCJ%[MJrIr!rH)rrr!r)rr2)r)rrrrrstrr!zdict[int, int])rrrrrr'r!zdict[int, list[int]])r8rHr!rm)rtuple[torch.Tensor, ...] | NonerLiteral[False]r!rp)r&tuple[torch.Tensor | None, ...] | Noner Literal[True]r!rs)rAtuple[torch.Tensor | None, ...] | tuple[torch.Tensor, ...] | Nonerr2r!rs)F)r8rHrr2r!rH).)rzCtorch.Tensor | tuple[torch.Tensor, ...] | list[torch.Tensor] | Nonerr)r!r()rzQtorch.Tensor | tuple[torch.Tensor | None, ...] | list[torch.Tensor | None] | Nonerr+r!r*)rztorch.Tensor | tuple[torch.Tensor, ...] | tuple[torch.Tensor | None, ...] | list[torch.Tensor] | list[torch.Tensor | None] | Nonerr2r!r,) rr'r rmr ztorch.Tensor | TensorMetarr2rr2r!rK) rr'r rr z,tuple[torch.Tensor | TensorMeta | None, ...]rr2rr2r!rK) rrrztuple[torch.Tensor, ...]r rr!r2r!r)= __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]r'r&rL RuntimeErrorr(r,rOrmrorrr;rrrrrrrrrrrrr rr&rrrr:s #(NN 6,<8B   8 $h c3ht1C CD iDGlG $d#HH$HV $d#@;@$@H$l2 I2  DDD6 $d#**$*  $d#  $   *6(6"60@ @ P0D0p*/6+ 17 !*-F17 !*-4    /( "%( ,((# ( ( / 3//* / /!" N!"!"* !"H!( "%* M**%* *  !$ 1 1  1 , 1 1 -  - - G- t$" B BB &B  B  BBT#" 6 6+6 96  6  66rDD "D +D D  Dr