a dD3@sddlZddlZddlZddlmZddlmZmZmZm Z m Z m Z m Z m Z ddlZddlmmZddlmZddlmZddlmZmZmZddlmZddlmZdd lmZdd l m!Z!ej"#rdd l$m%Z%nGd d d Z%e&e'Z(d0ee ee edddZ)eee*e edddZ+e,dddZ-d1ee ee e e!e.fedddZ/d2ee ee e e!e.fedddZ0Gdddej1j2Z3eeeddd Z4d3ee d"e,ed#d$d%Z5d4ee.e e*e e*edd&d'd(Z6ej7e.d)d*d+Z8Gd,d-d-eZ9Gd.d/d/eZ:dS)5N) nullcontext)AnyIterableIteratorListOptionalSizedTupleUnion)module_available)Tensor)DatasetDistributedSamplerSampler)ClusterEnvironment)_TORCH_GREATER_EQUAL_1_12)rank_zero_info)ReduceOpgroupc@seZdZdZdS)rN)__name__ __module__ __qualname__WORLDrro/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/lightning_fabric/utilities/distributed.pyrsr)resultrreturnc s`|durtjjj}|}tj|}tjj|d|jdkrJt|||Stj |j |j dfddt |D}tjj ||dt|jddjtfdd |D}|rt|||Sg}}t|D]}|d||qt||fd dt |D}tj ||t|D](\} } d d| D} || | || <q2|S) ahFunction to gather all tensors from several DDP processes onto a list that is broadcasted to all processes. Works on tensors that have the same number of dimensions, but where each dimension may differ. In this case tensors are padded, gathered and then trimmed to secure equal workload for all processes. Args: result: The value to sync group: The process group to gather results from. Defaults to all processes (world) Return: gathered_result: List with size equal to the process group where gathered_result[i] corresponds to result tensor from process i Nrrdevicecsg|]}tqSrtorchZ zeros_like.0_) local_sizerr :z'_gather_all_tensors..Zdimc3s|]}t|kVqdSN)all)r#Zls)max_sizerr =r'z&_gather_all_tensors..csg|]}tqSrr r") result_paddedrrr&Jr'cSsg|] }t|qSr)slice)r#Zdim_sizerrrr&Mr')r! distributedrr contiguousget_world_sizebarrierndim_simple_gather_all_tensorstensorshaperrange all_gatherstackmaxvaluesr*detachcpureversedappenditemFpad enumerate) rr world_sizeZ local_sizesZall_sizes_equalZpad_dimsZpad_byvalgathered_resultidxZ item_sizeZ slice_paramr)r%r+r-r_gather_all_tensorss4        rH)rrrDrcs*fddt|D}tj|||S)Ncsg|]}tqSrr r"rrrr&Sr'z._simple_gather_all_tensors..)r7r!r/r8)rrrDrFrrIrr4Rsr4rcCs&ddlm}tjr tjp$|S)Nr_tpu_distributed)Z!lightning_fabric.accelerators.tpurLr!r/ is_availableis_initializedrKrrr_distributed_availableXs rO)rr reduce_oprcCstrt|||dS|S)aFunction to reduce a tensor across worker processes during distributed training. Args: result: The value to sync and reduce (typically tensor or number) group: The process group to gather results from. Defaults to all processes (world) reduce_op: The reduction operation. Defaults to sum. Can also be a string of 'avg', 'mean' to calculate the mean during reduction. Return: reduced value )rrP)rO _sync_ddp)rrrPrrr_sync_ddp_if_available^srRcCsd}|durtjjj}t|trH|dvr8tj}d}qLt t| }n|}t drddl m }|rtjdd kr|d vrtd |}tjj|d tjj|||dd |r|tj|}|S)aFunction to reduce the tensors from several DDP processes to one main process. Args: result: The value to sync and reduce (typically tensor or number) group: The process group to gather results from. Defaults to all processes (world) reduce_op: The reduction operation. Defaults to sum. Can also be a string of 'avg', 'mean' to calculate the mean during reduction. Return: reduced value FN)avgmeanTz,habana_frameworks.torch.utils.library_loaderr)is_habana_avaialbleZHCCL_DISTRIBUTED_BACKEND1)ztorch.LongTensorztorch.hpu.LongTensorz0Long tensor unsupported on HPU, casting to floatr)oprasync_op)r!r/rr isinstancestrlowerrSUMgetattrupperr Z,habana_frameworks.torch.utils.library_loaderrUosenvirongettyperfloatr2 all_reducer1)rrrPZdivide_by_world_sizerWrUrrrrQqs0      rQc@sLeZdZeejfeeededddZ eeee edfdddZ dS) _AllGathertorch.distributed.ProcessGroup)ctxr5rrcsJ||_fddttjj|dD}tjj||dtj|dd}|S)Ncsg|]}tqSrr r"r5rrr&r'z&_AllGather.forward..rrr()rr7r!r/r1r8r9)rgr5rZgathered_tensorrrhrforwards  z_AllGather.forwardN)rg grad_outputrcGs8t|}tjj|tjjjd|jd|tjdfS)NF)rWrXr)r!catr/rdrr\rZget_rank)rgrjrrrbackwards z_AllGather.backward) rrr staticmethodrrrr rrir rlrrrrres re)r5rrcCs4tjdkrtst||Sddl}|jjj ||S)z!Compatibility layer with Windows.win32rN) sysplatformrreapplyZtorch.distributed.nnr/nn functionalr8)r5rr!rrr_functional_all_gathers rtFrf)r5r sync_gradsrcCsXts |S|}|rtntt||}Wdn1sD0Yt|S)amFunction to gather a tensor from several distributed processes. Args: tensor: Tensor of shape (batch, ...) group: The process group to gather results from. Defaults to all processes (world) sync_grads: Flag that allows users to synchronize gradients for all_gather op Return: A tensor of shape (world_size, batch, ...) N)rOr0rr!Zno_gradrtr9)r5rruZgathered_tensorsrrr_all_gather_ddp_if_availables (rv)cluster_environmenttorch_distributed_backend global_rankrDkwargsrc Kstjstdtjr*tddS|dur6|n|}|durJ|n|}|j t j d<t |j t j d<td|d|dd |tjj|f||d |td d |d |dd ddS)aUtility function to initialize distributed connection by setting env variables and initializing the distributed process group. Args: cluster_environment: ``ClusterEnvironment`` instance torch_distributed_backend: Backend to use (includes `nccl` and `gloo`) global_rank: Rank of the current process world_size: Number of processes in the group kwargs: Kwargs for ``init_process_group`` Raises: RuntimeError: If ``torch.distributed`` is not available zOtorch.distributed is not available. Cannot initialize distributed process groupz7torch.distributed is already initialized. Exiting earlyNZ MASTER_ADDRZ MASTER_PORTz'Initializing distributed: GLOBAL_RANK: z , MEMBER: /)ZrankrDzd----------------------------------------------------------------------------------------------------z distributed_backend=z5 All distributed processes registered. Starting with z processes  )r!r/rM RuntimeErrorrNlogdebugryrDZ main_addressr_r`rZZ main_portinfoZinit_process_groupr)rwrxryrDrzrrr_init_dist_connections(     r)rrcCs|jdkrdSdS)NZcudaZncclZgloo)rbrrrr-_get_default_process_group_backend_for_devicesrc@sTeZdZdZeeefddddZee dddZ ed d d Z dd d d Z dS)_DatasetSamplerWrapperz6Dataset to create indexes from `Sampler` or `Iterable`N)samplerrcCs:t|tstdt|tdkr*td||_d|_dS)NaYou seem to have configured a sampler in your DataLoader which does not provide `__len__` method. The sampler was about to be replaced by `DistributedSamplerWrapper` since `replace_sampler_ddp` is True and you are using distributed training. Either provide `__len__` method in your sampler, remove it from DataLoader or set `replace_sampler_ddp=False` if you want to handle distributed sampling yourself.infaYou seem to have configured a sampler in your DataLoader which does not provide finite `__len__` method. The sampler was about to be replaced by `DistributedSamplerWrapper` since `replace_sampler_ddp` is True and you are using distributed training. Either provide `__len__` method in your sampler which returns a finite number, remove it from DataLoader or set `replace_sampler_ddp=False` if you want to handle distributed sampling yourself.)rYr TypeErrorlenrc_sampler _sampler_list)selfrrrr__init__s z_DatasetSamplerWrapper.__init__)indexrcCs |jdurt|j|_|j|Sr))rlistr)rrrrr __getitem__s  z"_DatasetSamplerWrapper.__getitem__rJcCs t|jSr))rrrrrr__len__$sz_DatasetSamplerWrapper.__len__cCst|j|_dS)z4Reset the sampler list in order to get new sampling.N)rrrrrrrreset'sz_DatasetSamplerWrapper.reset) rrr__doc__r rrrintrrrrrrrrrs rcsFeZdZdZeeefeeddfdd Ze dfdd Z Z S) DistributedSamplerWrapperaWrapper over ``Sampler`` for distributed training. Allows you to use any sampler in distributed mode. It will be automatically used by Lightning in distributed mode if sampler replacement is enabled. Note: The purpose of this wrapper is to take care of sharding the sampler indices. It is up to the underlying sampler to handle randomness and shuffling. The ``shuffle`` and ``seed`` arguments on this wrapper won't have any effect. N)rargsrzrcs"tjt|g|Ri|dSr))superrr)rrrrz __class__rrr8sz"DistributedSamplerWrapper.__init__rJcs"jfddtDS)Nc3s|]}j|VqdSr))dataset)r#rrrrr,=r'z5DistributedSamplerWrapper.__iter__..)rrr__iter__rrrrr;s z"DistributedSamplerWrapper.__iter__) rrrrr rrrrrr __classcell__rrrrr,s r)N)NN)NN)NF)NN);loggingr_ro contextlibrtypingrrrrrrr r r!Ztorch.nn.functionalrrrsrAZ lightning_utilities.core.importsr r Ztorch.utils.datar rrZ9lightning_fabric.plugins.environments.cluster_environmentrZ"lightning_fabric.utilities.importsrZ$lightning_fabric.utilities.rank_zerorZ lightning_fabric.utilities.typesrr/rMZtorch.distributedr getLoggerrrrHrr4boolrOrZrRrQZautogradZFunctionrertrvrrrrrrrrrs\ (        5 &1    *'