a
    dT                     @   s~  d dl Z d dlmZmZmZmZmZ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 d dlmZ d dlmZ d dlZd dlmZ d d	lm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$ d dl%m&Z& d dl'm(Z( d dl)m*Z* d dl+m,Z, d dl-m.Z. d dl/m0Z0 edZ1erfe1rfe  d dl2m3Z3 W d   n1 sZ0    Y  neZ3G dd de&Z4dS )    N)AnyCallableDictListMappingOptionalTYPE_CHECKINGUnion)RequirementCache)Tensor)Module)	Optimizer)OrderedDict)_patch_cuda_is_available)ClusterEnvironment)ReduceOp)CUDAAccelerator)_LightningModuleWrapperBase$_LightningPrecisionModuleWrapperBase)CheckpointIO)ColossalAIPrecisionPlugin)DDPStrategy)
TBroadcast)	TrainerFn)is_overridden)rank_zero_warn)STEP_OUTPUT
colossalaiColoInitContextc                       s"  e Zd ZdZdZdMeee eeee	eeee	e	e	e	eee	ed ee
ej  ee ee ee dd fddZeejdddZeedddZeedddZddddZdddd Zdd fd!d"Zd#dd$d%d&Zejdd'd(d)Zddd*d+Zdd fd,d-ZdNeeeg ef eed.e f  eed/d0d1Z!dOee"eef d2d3d4Z#e$eef dd5d6d7Z%eeee& d8d9d:Z'eeee& d8d;d<Z(eee&d8d=d>Z)e*e"dd?d@dAZ+dPe,ee eee-ef  e,dCdDdEZ.dQe/ee/dGdHdIZ0dRe,ee ee,dJdKdLZ1  Z2S )SColossalAIStrategya  ColossalAI strategy. It only supports a single optimizer, which must be
    :class:`colossalai.nn.optimizer.CPUAdam` or :class:`colossalai.nn.optimizer.HybridAdam` now. Your model must
    be created in the function ``LightningModule.configure_sharded_model()``. Thus, you should overwrite this function.
    More details can be found in the below example.

    It configures accelerator and precision, and you should not configure them when initializing ``Trainer``.
    CUDA is essential for this strategy. Please make sure CUDA is available.

    Example::

        class GLUETransformer(LightningModule):
            ...
            def configure_sharded_model(self) -> None:
                self.model = BertForSequenceClassification.from_pretrained('bert-base-uncased')
        trainer = Trainer(..., accelerator="gpu", precision=16, strategy="colossalai")

    Args:
        use_chunk: Whether to use chunk-based memory management.
            It can speed up training, but slightly more memory will be used.

        chunk_size: The size of a chunk.
            It will be ignored when ``use_chunk=False``.
            If it's None, a best chunk size will be searched out based on ``chunk_search_range``,
            ``chunk_search_n_grids`` and ``min_chunk_size``.

        enable_distributed_storage: Whether to storage model in a distributed manner.
            It reduces memory from 1 to 1/N, but it may slow down training.

        placement_policy: It can be "cpu", "cuda" and "auto".

            * If it's "cpu", parameters, gradients and optimizer states will be offloaded to CPU,
                which means min CUDA memory will be used.
            * If it's "cuda", they won't be offloaded, which means max CUDA memory will be used. It's the fastest.
            * If it's "auto", they are moving dynamically based on CPU and CUDA memory usage.
                It will utilize heterogeneous memory space evenly and well.
                Note that "auto" policy can only work well when no other processes use CUDA during your training.

        force_outputs_fp32: Whether to cast outputs to fp32.

        gpu_margin_mem_ratio: The ratio of GPU remaining memory (after the first forward-backward)
            which will be used by optimizer.
            This argument will be ignored when ``placement_policy`` is not "auto".

        chunk_search_range: The range of chunk size to search.
            The actual search range will be from
            ``max(min_chunk_size, max_param_size)`` to ``max(min_chunk_size, max_param_size) + chunk_search_range``.

        chunk_search_n_grids: The number of intervals in the search range.

        min_chunk_size: The minimum size for a chunk in bytes.

        initial_scale: The initial dynamic loss scale value.

        min_scale: The minimum dynamic loss scaling value.

        growth_factor: The multiplication factor for increasing loss scale.

        backoff_factor: The multiplication factor for decreasing loss scale.

        growth_interval: The number of steps to increase loss scale when no overflow occurs.

        hysteresis: The number of overflows before decreasing loss scale.

        max_scale: The maximum dynamic loss scaling value.

    .. _colossalai.nn.optimizer.CPUAdam:
        https://colossalai.readthedocs.io/en/latest/colossalai/colossalai.nn.optimizer.cpu_adam.html

    .. _colossalai.nn.optimizer.HybridAdam:
        https://colossalai.readthedocs.io/en/latest/colossalai/colossalai.nn.optimizer.hybrid_adam.html

    r   TNautoF                                ?          zpl.accelerators.Accelerator)	use_chunk
chunk_sizeenable_distributed_storageplacement_policyforce_outputs_fp32gpu_margin_mem_ratiochunk_search_rangechunk_search_n_gridsmin_chunk_sizeinitial_scale	min_scalegrowth_factorbackoff_factorgrowth_interval
hysteresis	max_scaleacceleratorparallel_devicescluster_environmentcheckpoint_ioprecision_pluginreturnc                    s   t stdt  ddlm} W d    n1 s40    Y  t j|||||d || _|| _|| _	|| _
|| _|| _|||	d| _|
||||||d| _d| _| | _d S )NzTo use the `ColossalAIStrategy`, please install `colossalai` first. Download `colossalai` by consulting `https://colossalai.org/download`.r   )get_dist_logger)r<   r=   r>   r?   r@   )search_rangen_gridsr4   )r5   r6   r7   r8   r9   r:   r;   r'   )_COLOSSALAI_AVAILABLEModuleNotFoundErrorr   colossalai.loggingrB   super__init__r,   r-   r.   r/   r0   r1   chunk_size_search_kwargs
amp_kwargsZ
_num_nodesZ_logger)selfr,   r-   r.   r/   r0   r1   r2   r3   r4   r5   r6   r7   r8   r9   r:   r;   r<   r=   r>   r?   r@   rB   	__class__ p/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/pytorch_lightning/strategies/colossalai.pyrI   {   sB    *	zColossalAIStrategy.__init__)rA   c                 C   sN   t   ddlm} W d    n1 s(0    Y  | jd urH| j| j S | S )Nr   get_current_device)r   colossalai.utilsrR   r=   
local_rank)rL   rR   rO   rO   rP   root_device   s
    *
zColossalAIStrategy.root_devicec                 C   s   dS )z<Whether the plugin handles gradient accumulation internally.TrO   rL   rO   rO   rP   handles_gradient_accumulation   s    z0ColossalAIStrategy.handles_gradient_accumulationc                 C   s   dS )zDOverride to delay restoring from checkpoint till after pre-dispatch.TrO   rV   rO   rO   rP   restore_checkpoint_after_setup   s    z1ColossalAIStrategy.restore_checkpoint_after_setupc                 C   s   t  4 ddlm} ddlm} ddlm} W d    n1 s@0    Y  | jd usXJ |   |	|j
s|  |j| j| jd| jj| jjd || j d S )Nr   ParallelModeglobal_context)disable_existing_loggersZnccl)Zrank
world_sizebackendhostport)r   colossalai.contextrZ   colossalai.corer\   rG   r]   r>   Zset_world_ranksZis_initializedGLOBALZinit_global_distZglobal_rankr^   Zmain_addressZ	main_portZ
set_devicerT   )rL   rZ   gpcr]   rO   rO   rP   setup_distributed   s     *z$ColossalAIStrategy.setup_distributedr   c                 C   sH   t   ddlm} W d   n1 s(0    Y  G dd d|}| S )a  Provide hook to create modules in a distributed aware context. This is useful for when we'd like to
        shard the model instantly, which is useful for extremely large models which can save memory and
        initialization time.

        Returns: Model parallel context.
        r   r   Nc                       s,   e Zd Zejjeedd fddZ  ZS )zEColossalAIStrategy.model_sharded_context.<locals>.ModelShardedContextN)moduleargskwargsrA   c                    sF   t |dddu rd S t j|g|R i | | D ]
}d|_q6d S )N_colossalai_moduleFT)getattrrH   _post_init_methodmodulesrj   )rL   rg   rh   ri   Z
sub_modulerM   rO   rP   rl      s
    zWColossalAIStrategy.model_sharded_context.<locals>.ModelShardedContext._post_init_method)	__name__
__module____qualname__torchnnr   r   rl   __classcell__rO   rO   rM   rP   ModelShardedContext   s   rt   )r   (colossalai.utils.model.colo_init_contextr   )rL   r   rt   rO   rO   rP   model_sharded_context   s    *z(ColossalAIStrategy.model_sharded_contextc              
      s  t  , ddlm}m} ddlm} W d    n1 s80    Y  t   | jd usZJ | jj	oj| jj	j
}|rt| jdkrtd| jd }t|||fstdt| jtjtfsJ | j}t|dst  ( ddlm} dd	lm} W d    n1 s0    Y  | jstd
| jdd}	|	d }
| jdd}t|	| }t| jd d }t| j}||| | jd| j |
||d| _| jd usJ | jg|_!n|j!d | _|r||| jfd| j"i| j#g| _d S )Nr   )CPUAdam
HybridAdam)ZeroOptimizerr'   z8`ColossalAIStrategy` only supports single Optimizer now.z`ColossalAIStrategy` only supports `colossalai.nn.optimizer.CPUAdam` and `colossalai.nn.optimizer.HybridAdam` as its optimizer._colossalai_zero)	GeminiDDPrQ   zB`ColossalAIStrategy` must use chunk in versions higher than 0.1.10rC   r%   i   rD   r$   r4   T)rg   devicer/   Z
pin_memoryr0   search_range_mbZ
hidden_dimmin_chunk_size_mbr1   )$r   Zcolossalai.nn.optimizerrw   rx   Zcolossalai.zerory   rH   setup_precision_pluginlightning_moduletrainerZtraininglen
optimizers
ValueError
isinstancemodelplLightningModuler   hasattrcolossalai.nn.parallelr{   rS   rR   r,   rJ   getmathceilintr   r/   r0   rz   r1   rK   )rL   rw   rx   ry   Zis_training	optimizer	pl_moduler{   rR   r2   r}   Zsearch_n_gridsZsearch_intervalr~   r   rM   rO   rP   r      sX    *

,
z)ColossalAIStrategy.setup_precision_pluginz
pl.Trainer)r   rA   c                 C   s   | j j}|dkr td|dt| jtsBtd| jjj d|jj	t
jkrtd|jrdtd |jdkrvtd	|j}|jd
gkrtdt| j tstd| j| | jd usJ | j| j_| | j | | |   |   d S )NZ16z*`Trainer(strategy='colossalai', precision=z5)` is not supported. Consider setting `precision=16`.zB`ColossalAIStrategy` is only supported on `CUDAAccelerator`, but `z
` is used.ZbackwardzYou have overridden the `LightningModule.backward` hook but it will be ignored since ColossalAI handles the backward logic internally.r'   zaColossalAI does not support gradient accumulation now. Please set `accumulate_grad_batches` to 1.r   z^ColossalAI currently does not support different `accumulate_grad_batches` at different epochs.zI`ColossalAIStrategy` is only compatible with `ColossalAIPrecisionPlugin`.)r@   	precisionr   r   r<   r   rN   rn   statefnr   ZFITTINGr   r   r   Zaccumulate_grad_batchesaccumulation_schedulerZepochsr   setuprU   Z_deviceignore_no_grad_parametersZsetup_optimizersr   model_to_device)rL   r   r   r   rO   rO   rP   r   *  sD    



zColossalAIStrategy.setup)running_devicerA   c                 C   sB   | j d usJ | j  D ]$}|jst|dd |j||_qd S )NZ_ddp_to_ignoreT)r   
parametersZrequires_gradsetattrdatato)rL   r   paramrO   rO   rP   r   V  s
    z,ColossalAIStrategy.ignore_no_grad_parametersc                 C   sF   | j d usJ | j }| D ]$}||urt|dds|| j qd S )Nrj   F)r   rm   rk   r   rU   )rL   r   childrO   rO   rP   r   _  s
    z"ColossalAIStrategy.model_to_devicec                    sF   | j }t | _ | j}d | _| j}d | _t   || _ || _|| _d S N)r   listr   Z_lightning_modulerH   teardown)rL   r   Z
zero_modelr   rM   rO   rP   r   f  s    
zColossalAIStrategy.teardownzpl.LightningModule)r   opt_idxclosurer   ri   rA   c                 K   s6   |p| j }t|tjsJ | jj|f|||d|S )N)r   Zoptimizer_idxr   )r   r   r   r   r@   optimizer_step)rL   r   r   r   r   ri   rO   rO   rP   r   t  s    
z!ColossalAIStrategy.optimizer_step)rank_zero_onlyrA   c                    s   t   ddlm} W d   n1 s(0    Y  t| j|sBJ | jj|d}t| j }t|dksnJ |d \}}|d7 }|| j	u sJ t
  | D ]}||d |< q fdd	| D S )
aX  Returns a dictionary containing a whole state of the module. But all the tensors in the dictionary are
        detached from their parameters and located in cpu memory.

        Args:
            rank_zero_only: If True, only process rank 0 gets the correct dictionary.
                Otherwise, all processes get the same dictionary.
        r   )ZeroDDPN)Zonly_rank_0r'   . c                    s   i | ]\}} | |qS rO   rO   .0keyvalueZmapping_dictrO   rP   
<dictcomp>      zBColossalAIStrategy.lightning_module_state_dict.<locals>.<dictcomp>)r   r   r   r   r   
state_dictr   named_childrenr   r   dictkeysreplaceitems)rL   r   r   Zorg_dictchildrenprefixr   r   rO   r   rP   lightning_module_state_dict  s    *z.ColossalAIStrategy.lightning_module_state_dict)
checkpointrA   c                    s   |d }| j d usJ t| j  }t|dks4J |d \}}|d7 }|| ju sVJ t  | D ]}||  |< qdt fdd| D }| j 	| d S )Nr   r'   r   r   c                    s   i | ]\}} | |qS rO   rO   r   r   rO   rP   r     r   z<ColossalAIStrategy.load_model_state_dict.<locals>.<dictcomp>)
r   r   r   r   r   r   r   r   r   Zload_state_dict)rL   r   Z	orig_dictr   r   r   r   	load_dictrO   r   rP   load_model_state_dict  s    z(ColossalAIStrategy.load_model_state_dict)rh   ri   rA   c                 O   sL   | j d usJ | j   | j |i |W  d    S 1 s>0    Y  d S r   )r   r@   Zval_step_contextrL   rh   ri   rO   rO   rP   validation_step  s    z"ColossalAIStrategy.validation_stepc                 O   sL   | j d usJ | j   | j |i |W  d    S 1 s>0    Y  d S r   )r   r@   Ztest_step_contextr   rO   rO   rP   	test_step  s    zColossalAIStrategy.test_stepc                 O   sL   | j d usJ | j   | j |i |W  d    S 1 s>0    Y  d S r   )r   r@   Zpredict_step_contextr   rO   rO   rP   predict_step  s    zColossalAIStrategy.predict_step)strategy_registryrA   c                 C   s   |j d| dd d S )Nr   zDefault ColossalAI Strategy)description)register)clsr   rO   rO   rP   register_strategies  s    z&ColossalAIStrategy.register_strategiessum)tensorgroup	reduce_oprA   c                 C   s   t  4 ddlm} ddlm} ddlm} W d    n1 s@0    Y  t|tsX|S t|t	r|
 dv rtj}|j|jd}t  || }W d    q1 s0    Y  ntt| }||d|j|d}|S )Nr   )reducerY   r[   )avgmean)parallel_mode)dstr   op)r   #colossalai.communication.collectiver   rb   rZ   rc   r\   r   r   strlowerr   ZSUMZget_world_sizerd   rq   Zno_gradrk   upper)rL   r   r   r   r   rZ   re   Z
div_factorrO   rO   rP   r     s    *


(zColossalAIStrategy.reducer   )objsrcrA   c                 C   s   t  4 ddlm} ddlm} ddlm} W d   n1 s@0    Y  t|trd||||j	dS |g}t
jj||||j	d |d S dS )zBroadcasts an object to all processes.

        Args:
            obj: the object to broadcast
            src: source rank
        r   )	broadcastrY   r[   N)r   r   )r   )r   r   r   rb   rZ   rc   r\   r   r   rd   rq   ZdistributedZbroadcast_object_listZ	get_group)rL   r   r   r   rZ   re   Zobj_listrO   rO   rP   r     s    *
zColossalAIStrategy.broadcast)r   r   
sync_gradsrA   c                 C   sZ   t  ( ddlm} ddlm} W d   n1 s40    Y  |du sJJ ||d|jdS )z&Perform a all_gather on all processes.r   )
all_gatherrY   NF)Zdimr   )r   r   r   rb   rZ   rd   )rL   r   r   r   r   rZ   rO   rO   rP   r     s
    *zColossalAIStrategy.all_gather)TNTr!   Fr"   r#   r$   r%   r&   r'   r(   r)   r*   r(   r+   NNNNN)N)F)Nr   )r   )NF)3rn   ro   rp   __doc__Zstrategy_nameboolr   r   r   floatr   rq   r|   r   r   r   rI   propertyrU   rW   rX   rf   rv   r   r   r   r   r   r   r   r   r	   r   r   r   r   r   r   r   r   r   r   classmethodr   r   r   r   r   r   r   rs   rO   rO   rM   rP   r    /   s   I                     ?6,	 
 r    )5r   typingr   r   r   r   r   r   r   r	   rq   Z lightning_utilities.core.importsr
   r   Ztorch.nnr   Ztorch.optim.optimizerr   Ztyping_extensionsr   Zpytorch_lightningr   Z"lightning_fabric.accelerators.cudar   Z9lightning_fabric.plugins.environments.cluster_environmentr   Z&lightning_fabric.utilities.distributedr   Z#pytorch_lightning.accelerators.cudar   Z pytorch_lightning.overrides.baser   r   Z.pytorch_lightning.plugins.io.checkpoint_pluginr   Z#pytorch_lightning.plugins.precisionr   Z pytorch_lightning.strategies.ddpr   Z%pytorch_lightning.strategies.strategyr   Z pytorch_lightning.trainer.statesr   Z)pytorch_lightning.utilities.model_helpersr   Z%pytorch_lightning.utilities.rank_zeror   Z!pytorch_lightning.utilities.typesr   rE   ru   r   r    rO   rO   rO   rP   <module>   s6   (.