
    sg
>                         U d dl Z d dlZd dlmZmZmZ d dlZd dlmZ	 d dlm
Z
 d dlmZmZ d dlmZ d dlmZ g dZdaed   ed	<   d
edefdZded
eddfdZ G d d      Z G d de      Z G d de      Zy)    N)DictListOptional)Tensor)_get_device_handle
DeviceMesh)DTensorSpec)Shard)is_rng_supported_meshmanual_seedOffsetBasedRNGTrackerTensorParallelRNGTracker_RNGStateTracker_rng_trackerdevice_meshreturnc                     t        | j                        }|rt        |d      ryt        j                  d| j                   d       y)aF  Checks if the current device of ``device_mesh`` supports DTensor's random APIs.
    Currently DTensor Random APIs only supports cuda/cuda-like devices. We suggest
    users call this API to test the availability before using our random APIs.

    Args:
        device_mesh (:class:`DeviceMesh`): The device mesh on which we check if the
            random ops APIs are supported.

    Returns:
        A bool value. True if ``device_mesh`` supports DTensor Random APIs; False otherwise.

    .. warning::
        Currently we only support correct RNG on cuda/cuda-like devices.
    set_rng_stateTz:DTensor random operators may not have complete support on z device meshF)r   device_typehasattrwarningswarn)r   device_handles     S/var/www/html/venv/lib/python3.12/site-packages/torch/distributed/tensor/_random.pyr   r      sJ     '{'>'>?M@ 	HI`I`Haamn	
     seedc                    t        |j                        }|st        d|j                         | gt        j                         z  }t        j
                  ||        t        |      D ]B  \  }}| t        |      k7  st        d| ddt        j                          d|  d| d| d	       t        st        |j                        a
|j                         gt        t        t              rt        j                  ||        yt        t        t              rt        j                  |        yt        d	t               y)
a  Sets the seed for generating random numbers for the calling rank.

    Args:
        seed (int): The desired seed.
        device_mesh (:class:`DeviceMesh`): The device mesh to set the seed.

    Returns:
        None

    .. warning::
        When calling this function, :func:`manual_seed` must be called from all ranks of the
        default ``ProcessGroup`` even if some ranks may not be a part of the ``device_mesh``,
        with the same ``seed`` value.
        If ``device_mesh`` is a sub-mesh and the calling rank is not a part of it,
        ``manual_seed`` will not set its GPU device's generator seed.
        Current implementation only supports a GPU device mesh.
    zEDTensor randomness only supports cuda/cuda-like device type, but got z"calling manual_seed function over z- but received different seed values on ranks:zseed on rank z is z, and seed on rank !Nz7Unknown type of cuda RNG state tracker: _rng_tracker = )r   r   NotImplementedErrordistget_world_sizeall_gather_object	enumerateintRuntimeErrorget_rankr   r   get_coordinate
isinstancer   _manual_seed)r   r   r   object_listrankobjects         r   r   r   3   s;   $ '{'>'>?M!ST_TkTkSlm
 	

 &4..00K;-!+. f3v;4[MAno0TF:MdVSWX^W__`a  ,[-D-DE !!#/l$<=%%k48&;<%%d+I,X  0r   c                       e Zd ZdZddefdZedeeef   fd       Z	ede
fd       Zej                  dd       Zde
fd	Zd
edefdZd
ededdfdZdefdZy)r   a  
    _RNGStateTracker stores Random Number Generator (RNG) state (a ByteTensor object)
    in a dict, mapping from a corresponding tag to each state tensor. It also provides
    a set of convenient utility methods to help access/modify the state tensors. The most
    important interface is _distribute_region which will be used when DTensor executes
    a random op (an operator that calls RNG).
    r   c                    || _         t        |      | _        | j                  r| j                  j                         s"t	        | j
                  j                   d      i | _        | j                  j                         g| _	        d| _
        y )Nz= instantiation requires the presence of CUDA/CUDA-like deviceT)_device_typer   _device_handleis_availabler%   	__class____name___statescurrent_device_devices_use_distribute_region)selfr   s     r   __init__z_RNGStateTracker.__init__o   s~    '0=##(;(;(H(H(J>>**++hi  +-,,;;=>&*#r   r   c                     | j                   S N)r4   r8   s    r   
rng_statesz_RNGStateTracker.rng_states{   s    ||r   c                     | j                   S r;   r7   r<   s    r   distribute_region_enabledz*_RNGStateTracker.distribute_region_enabled   s    ***r   Nc                     || _         y r;   r?   )r8   values     r   r@   z*_RNGStateTracker.distribute_region_enabled   s
    &+#r   c                     || j                   v S r;   )r=   )r8   names     r   rng_state_is_syncz"_RNGStateTracker.rng_state_is_sync   s    t&&r   rD   c                     || j                   vr$t        | j                  j                   d|       | j                   |   dd j	                  t
        j                        }t        |j                               S )N  does not have random state for r      dtype	r=   r%   r2   r3   viewtorchint64r$   item)r8   rD   seed_tensors      r   get_seedz_RNGStateTracker.get_seed   sr    t&>>**++KD6R  t,a277ekk7J;##%&&r   r   c                    t        j                  |g      j                  t         j                        }t        j                  dg      j                  t         j                        }t        j                  ||g      | j
                  |<   y )Nr   )rM   tensorrL   uint8catr=   )r8   rD   r   rP   offset_tensors        r   set_seedz_RNGStateTracker.set_seed   s\    llD6*//<aS)..u{{; %		;*F Gr   specc                      y r;    r8   rX   s     r   _distribute_regionz#_RNGStateTracker._distribute_region   s    r   cuda)r   N)r3   
__module____qualname____doc__strr9   propertyr   r   r=   boolr@   setterrE   r$   rQ   rW   r	   r\   rZ   r   r   r   r   f   s    
+C 
+ Df-   +4 + + %%, &,' ''S 'S 'HS H H H
{ r   c                        e Zd ZdZddef fdZdeddfdZej                  de
fd	       Zd
edefdZd
ededdfdZde
ddfdZde
deddfdZdee   dee   defdZ xZS )r   z
    This subclass of ``_RNGStateTracker`` defines the default policy of how RNG states
    should be shared and synchronized among all ranks to respect the semantics of DTensor
    random operators.
    r   c                     t         |   |       | j                  j                         j	                  |      }t        j                  |d       |j	                  d      | j                  d<   y )Nr   cpuparallel-rng)superr9   r0   get_rng_statetor    	broadcastr=   )r8   r   	rng_stater2   s      r   r9   zOffsetBasedRNGTracker.__init__   sT    %''557::;G	y!$*3,,u*='r   parallel_seedr   Nc                 (    | j                  d|       y )Nri   )rW   )r8   ro   s     r   r)   z"OffsetBasedRNGTracker._manual_seed   s    nm4r   rX   c              #     K   | j                  d      st        d      | j                  r| j                  d      }| j	                  |       t
        j                  j                  | j                  | j                        5  | j                  j                  | j                  d          	 d  | j                  ||       	 d d d        y d  y # | j                  ||       w xY w# 1 sw Y   y xY ww)Nri   zlOffsetBasedRNGTracker requires the random state to be synchronized before entering into a distribute region!r   )rE   r%   r@   
get_offset_set_pre_op_offsetrM   randomfork_rngr6   r/   r0   r   r=   _set_post_op_offset)r8   rX   
old_offsets      r   r\   z(OffsetBasedRNGTracker._distribute_region   s      %%n5< 
 ))8J##D)&&t}}$BSBS&T ?##11$//.2QR? ,,T:>? ?  ,,T:>? ?s6   B C4)C(,C0C(C4C%%C((C1-C4rD   c                     || j                   vr$t        | j                  j                   d|       | j                   |   dd  j	                  t
        j                        }t        |j                               S )NrG   rH   rI   rK   )r8   rD   rV   s      r   rs   z OffsetBasedRNGTracker.get_offset   sr    t&>>**++KD6R  .388u{{8K=%%'((r   offsetc                 :   || j                   vr$t        | j                  j                   d|       | j                   |   dd }t	        j
                  |g      j                  t        j                        }t	        j                  ||g      | j                   |<   y )NrG   r   rH   )	r=   r%   r2   r3   rM   rS   rL   rT   rU   )r8   rD   rz   rP   rV   s        r   
set_offsetz OffsetBasedRNGTracker.set_offset   s    t&>>**++KD6R  t,a2fX.33EKK@ %		;*F Gr   c                    |j                   }|j                  }|j                  }|j                         }|J |D cg c]  }|dk\  r||   nd }}|D cg c]  }|dk\  r|j	                  |      nd }}| j                  ||      }	t        |      }
t        |j                        D ]Q  \  }}t        |t              s|j	                  |      }|j                  }|j                  ||   |dd      d   |
|<   S ddlm}  ||
      }| j                  d      }|	|z  dz   d	z  d	z  }| j!                  d||z          yc c}w c c}w )
a
  Set the starting RNG offset for current device's local shard before actual
        op execution. The pre_op_offset value should start from the current RNG offset
        and increment by the size of local shard until it reaches the size of the whole
        DTensor. For different ranks that hold the same DTensor shard, their pre_op_offset
        will be the same.

        Args:
            spec (:class:`DTensorSpec`): the spec of the DTensor object on which
                we prepare the offset for running random ops.

        Returns:
            None

        .. warning::
            Note that, current implementation does not consider DTensor's continguity.

        Example:
            take a DTensor of shape [8, 16] as an example. Assume that the DTensor
            is placed on a device mesh with placements ([Shard(1), Replicate(), Shard(0)]),
            and the mesh is:
                [[[0, 1], [2, 3]], [[4, 5], [6, 7]]]
            ``spec.mesh.get_coordinate()`` provides the coordinate of the current rank
            in the mesh. For example, the coordinate of rank 5 is (1, 0, 1).

            Another concept to introduce besides rank coordinate is shard coordinate.
            Each rank holds a local shard of the DTensor. In the example, the DTensor
            is partitioned into 4 [4, 8] shards. The first shard has 2 replicas and
            rank 0 (coord (0, 0, 0)) and rank 2 (coord (0, 1, 0)) have 1 replica each.
            That being said, the local shard on rank 0 and rank 2 correspond to the same
            shard of the DTensor. To denote each DTensor shard, we use a shard coordinate
            (in the example, it will be a tuple (i, j) where shard (i, j) has the slice
            DTensor[4 * i : 4 * (i + 1), 8 * j : 8 * (j + 1)], 0 <= i < 2, 0 <= j < 2).

            Once we have rank coordinate and shard coordinate, we can calculate on each rank
            what shard of the DTensor the rank holds, with the help of dim_map. The dim_map
            of the above DTensor is [2, 0] so the shard coordinate of a rank with rank coord
            (x, y, z) is simply (z, x) by taking(rank_coord[dim_map[0]],rank_coord[dim_map[1]]).
            Following this calculation,
            rank 0 and rank 2 holds the shard of coord (0, 0);
            rank 1 and rank 3 holds the shard of coord (0, 1);
            rank 4 and rank 6 holds the shard of coord (1, 0);
            rank 5 and rank 7 holds the shard of coord (1, 1);

            The last value to calculate before obtaining the starting offset is the shard linear index.
            The starting offset for each rank will be its shard_linear_index * local_tensor_numel.
        Nr      F)return_offsetprodri         )shapemeshdim_mapr'   size_calc_shard_linear_idxlistr#   
placementsr(   r
   dim_local_shard_size_on_dim#torch.distributed.tensor._ops.utilsr   rs   r|   )r8   rX   dtensor_shaper   r   
coordinatemesh_dimshard_coord
shard_sizeshard_linear_idxlocal_size_on_rank_0idx	placementmesh_dim_size	shard_dimr   
local_sizecurrent_offsetoffset_incrs                      r   rt   z(OffsetBasedRNGTracker._set_pre_op_offset   s   ^ 

yy,, ((*
%%%IP
=EHMJx q8
 
 IP
<D8q=DIIha7

 

  66{JO  $M2'8 		NC)U+ $		#%MM	2;2T2T!),!"'	 3U 3
 3$Y/			 	=./
 8 (*4q8Q>B(DEC

s   E Erx   c                 x    |j                   }ddlm}  ||      }|dz   dz  dz  }| j                  d||z          y)a  Sets the RNG to a synchronized state after running the local random op. Every
        rank should set its RNG offset to `old_offset + DTensor.numel()` where old_offset is
        the offset before calling `set_pre_op_offset` i.e. the offset before running DTensor
        random ops.

        Args:
            spec (:class:`DTensorSpec`): the spec of the DTensor object on which
                we post-process the offset for running random ops.

        Returns:
            None
        r   r   r   r   ri   N)r   r   r   r|   )r8   rX   rx   r   r   numels         r   rw   z)OffsetBasedRNGTracker._set_post_op_offset2  sB     

<]# q 1$
U(:;r   r   r   c                 t    d}d}t        t        |      t        |            D ]  \  }}|||z  z  }||z  } |S )Nr   r~   )zipreversed)r8   r   r   r   shard_coord_strider   r   s          r   r   z,OffsetBasedRNGTracker._calc_shard_linear_idxI  sX     Xk2HZ4HI 	'IC&8 88$&	'  r   r]   )r3   r_   r`   ra   rb   r9   r$   r)   
contextlibcontextmanagerr	   r\   rs   r|   rt   rw   r   r   __classcell__r2   s   @r   r   r      s    >C >5# 5$ 5 {  *)s )s )Hs HC HD HZF{ ZFt ZFx< < < <.
 9
 26s)
 	
 r   r   c                   d     e Zd Zddef fdZ	 d	dedefdZej                  de
fd       Z xZS )
r   r   c                 r    t         |   |       | j                  j                         | j                  d<   y )Ntensor-parallel-rng)rj   r9   r0   rk   r=   )r8   r   r2   s     r   r9   z!TensorParallelRNGTracker.__init__W  s.    %151D1D1R1R1T-.r   tp_mesh	base_seedc                 \    |j                         }d}||z   |z   }| j                  d|       y )Ni
  r   )get_local_rankrW   )r8   r   r   tensor_parallel_rankMegatronMagicNumtensor_parallel_seeds         r   r)   z%TensorParallelRNGTracker._manual_seed\  s=    
  '557  (+;;>RR+-ABr   rX   c              #     K   | j                  d      st        d      | j                  rt        j                  j                  | j                  | j                        5  | j                  j                  | j                  d          	 d  | j                  j                         | j                  d<   	 d d d        y d  y # | j                  j                         | j                  d<   w xY w# 1 sw Y   y xY ww)Nr   zoTensorParallelRNGTracker requires the random state to be synchronized before entering into a distribute region!rr   )rE   r%   r@   rM   ru   rv   r6   r/   r0   r   r=   rk   r[   s     r   r\   z+TensorParallelRNGTracker._distribute_regionh  s      %%&;<< 
 ))&&t}}$BSBS&T 	<##11OO$9:< ++99; OO-	< 	<  ++99; OO-	< 	<s6   AC< )C0
C'C06C<)C--C00C95C<r]   )i  )r3   r_   r`   rb   r9   r   r$   r)   r   r   r	   r\   r   r   s   @r   r   r   V  sQ    UC U 
C
C 
C {  r   r   )r   r   typingr   r   r   rM   torch.distributeddistributedr    r   torch.distributed.device_meshr   r   &torch.distributed.tensor._dtensor_specr	   (torch.distributed.tensor.placement_typesr
   __all__r   __annotations__rd   r   r$   r   r   r   r   rZ   r   r   <module>r      s      ' '     H > : .2h)* 1z d 40c 0
 0t 0f3 3lw , w t'/ 'r   