o oiD @sddlZddlZddlZddlZddlZddlZddlmZddlmZddl m Z ddl m Z m Z mZmZmZmZmZmZddlZddlmmZddlmZddlmZddlmZmZmZdd l m!Z!m"Z"m#Z#dd l$m%Z%dd l&m'Z'dd l(m)Z)dd l*m+Z+ddl,m-Z-m.Z.ej/0rddl1m2Z2nGdddZ2e rddl3m4Z4ddl5m6Z6ddl7m8Z8e9e:Z;dKdddee-dedLdedee deefddZ?dede d ez'_gather_all_tensors..)dimc3s|] }t|kVqdSNr.)rQls)max_sizer!r" sz&_gather_all_tensors..crLr!rMrP) result_paddedr!r"rTrUcSsg|]}t|qSr!)slice)rQdim_sizer!r!r"rTs)rN distributedrr contiguousget_world_sizer;ndim_simple_gather_all_tensorstensorshaperKrange all_gatherstackmaxvaluesr/detachcpureversedappenditemFpad enumerate) rIrr, local_sizesall_sizes_equalpad_dimspad_byvalgathered_resultidx item_size slice_paramr!)rSrYr[r"_gather_all_tensorsfs4        r{r,cs*fddt|D}tj|||S)NcrLr!rMrPrIr!r"rTrUz._simple_gather_all_tensors..)rerNr^rf)rIrr,rwr!r|r"rbsrb reduce_opcCstr t|||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 )rr})_distributed_is_initialized _sync_ddp)rIrr}r!r!r"_sync_ddp_if_availablesrcCsd}|dur tjjjn|}t|tr6|dkrdn|}|dkr.tj|dkr.tj }d}n t t| }n|}t drRt jdd krR|d vrRtd |}tjj|d tjj|||dd tj|}|sm|St|sy|||S||S)a%Reduces a tensor across several distributed processes. This operation is performed in-place, meaning the result will be placed back into the input tensor on all processes. 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: The reduced value. FNmeanavgglooThabana_frameworksHCCL_DISTRIBUTED_BACKEND1)ztorch.LongTensorztorch.hpu.LongTensorz0Long tensor unsupported on HPU, casting to floatr)oprasync_op)rNr^rr isinstancestrlower get_backendrSUMgetattrupperr osenvirongettyperfloatr; all_reducer`is_floating_pointcopy_div_)rIrr}divide_by_world_sizerr,r!r!r"rs2    rFrcztorch.distributed.ProcessGroup sync_gradscCsdts|Sddlm}|}|rtnt |||}Wdn1s(wYt|S)anFunction 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, ...) r)rfN)r~torch.distributed.nn.functionalrfr_rrNno_gradrg)rcrrrfgathered_tensorsr!r!r"_all_gather_ddp_if_availables   rcluster_environmentr%torch_distributed_backend global_rankkwargsc Kstjs tdtjrtddS|dur|n|}|dur%|n|}|j t j d<t |j t j d<td|d|dd |tjj|f||d ||d kr]tttd 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 earlyN MASTER_ADDR MASTER_PORTz'Initializing distributed: GLOBAL_RANK: z , MEMBER: r-/)rankr,ncclzd----------------------------------------------------------------------------------------------------z distributed_backend=z5 All distributed processes registered. Starting with z processes  )rNr^ is_available RuntimeErroris_initializedlogdebugrr, main_addressrrr main_portinfoinit_process_groupatexitregister_destroy_dist_connectionr)rrrr,rr!r!r"_init_dist_connection s,      rcCs4ttjtjtrtjttjtjdSrW)signalSIGINTSIG_IGNr~rNr^destroy_process_groupSIG_DFLr!r!r!r"r8s rrKcCs|jdkrdSdS)Ncudarr)rrJr!r!r"-_get_default_process_group_backend_for_device@src@sXeZdZdZdeeefddfddZede de fdd Z de fd d Z dd d Z dS)_DatasetSamplerWrapperz6Dataset to create indexes from `Sampler` or `Iterable`samplerr+NcCs:t|ts tdt|tdkrtd||_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 `use_distributed_sampler` is True and you are using distributed training. Either provide `__len__` method in your sampler, remove it from DataLoader or set `use_distributed_sampler=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 `use_distributed_sampler` 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 `use_distributed_sampler=False` if you want to handle distributed sampling yourself.)rr TypeErrorlenr_sampler _sampler_list)selfrr!r!r"__init__Gs  z_DatasetSamplerWrapper.__init__indexcCs |jdur t|j|_|j|SrW)rlistr)rrr!r!r" __getitem__^s   z"_DatasetSamplerWrapper.__getitem__cCs t|jSrW)rrrr!r!r"__len__ds z_DatasetSamplerWrapper.__len__cCst|j|_dS)z4Reset the sampler list in order to get new sampling.N)rrrrr!r!r"resetgsz_DatasetSamplerWrapper.resetr+N)rrr__doc__r rrrrintrrrrr!r!r!r"rDsrcsPeZdZdZdeeefdededdffdd Ze de ffd d 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. rargsrr+Ncs"tjt|g|Ri|dSrW)superrr)rrrr __class__r!r"rys"z"DistributedSamplerWrapper.__init__cs"jfddtDS)Nc3s|]}j|VqdSrW)dataset)rQrrr!r"rZsz5DistributedSamplerWrapper.__iter__..)rrr__iter__rrrr"r|s z"DistributedSamplerWrapper.__iter__) rrrrr rrrrrrr __classcell__r!r!rr"rls & rr- num_processescCs(|dkr td|dtdt|S)Nr-z$`num_processes` should be >= 1, got .) ValueErrorrhr)rr!r!r"_suggested_max_num_threadssrcCs2dtjvrt|}t|t|tjd<dSdS)NOMP_NUM_THREADS)rrrrNset_num_threadsr)r num_threadsr!r!r"_set_num_threads_if_neededs  rcCstjo tjSrW)rNr^rrr!r!r!r"r~sr~c@sLeZdZdZdddZdddZdefdd Zd ed ed eddfd dZ dS)_InfiniteBarrieraPA barrier with an infinite timeout. Creates a new process group with the GLOO backend with a very high timeout that makes the barrier effectively wait forever. This is useful in cases where you want to execute a long-running operation on a subset of ranks that should not be subject to the regular collective timeout. r+NcCsd|_dd|_dS)NcSsdSrWr!r!r!r!r"sz+_InfiniteBarrier.__init__..)rr;rr!r!r"rsz_InfiniteBarrier.__init__cCs |dSrW)r;rr!r!r"__call__s z_InfiniteBarrier.__call__cCs,trtjjdtddd|_|jj|_|S)Nri')days)backendr*)r~rNr^ new_grouprrmonitored_barrierr;rr!r!r" __enter__s z_InfiniteBarrier.__enter__exc_type exc_value tracebackcCs(||jdurtj|jdSdSrW)r;rrNr^r)rrrrr!r!r"__exit__s z_InfiniteBarrier.__exit__r) rrrrrrrrrrr!r!r!r"rs   rr$cCstr ddlm}t||SdS)Nrr#F)rtorch.distributed._tensorr$r)rcr$r!r!r" _is_dtensors  r)Nr'rW)NN)NFr)r-)Prr@loggingrrr>rdatetimerpathlibrtypingrrrrr r r r rNtorch.nn.functionalnn functionalro lightning_utilities.core.importsr rtorch.utils.datarrrtyping_extensionsrrr#lightning.fabric.utilities.cloud_iorlightning.fabric.utilities.datar"lightning.fabric.utilities.importsr$lightning.fabric.utilities.rank_zeror lightning.fabric.utilities.typesrrr^rtorch.distributedrrr$lightning.fabric.pluginsr%lightning.fabric.strategiesr& getLoggerrrrboolrHr{rbrrrrrrrKrrrrrr~rrr!r!r!r"s   (             ;6 ,;  /(