a d,@sdZddlmZmZmZmZmZddlZddlmZddl m Z ddl m Z ddl mZddl mZdd l mZdd l mZdd l mZdd l mZdd lmZmZmZd-e eeeeeeddddZee edddZ!ee"efee efdddZ#eeedddZ$edddZ%eeedddZ&Gdd d ej'j(Z)eeedd!d"Z*eeedd#d$Z+eeedd%d&Z,eeedd'd(Z-e.dd)d*Z/eeedd+d,Z0dS).z5Utilities that can be used with distributed training.)AnyCallableDictOptionalTupleN)Tensor)DistributedDataParallel)_all_gather_ddp_if_available)_distributed_available)_gather_all_tensors)-_get_default_process_group_backend_for_device)_init_dist_connection) _sync_ddp)_sync_ddp_if_available)rank_zero_debugrank_zero_deprecationrank_zero_info)modelddp_comm_state ddp_comm_hookddp_comm_wrapperreturncCs^|dur dS|}|dur:td|jd|jd||}td|jd|j||ddS)a5 Function to register communication hook for DDP model https://pytorch.org/docs/master/ddp_comm_hooks.html. Args: model: DDP model ddp_comm_state: state is passed to the hook and can be used to maintain and update any state information that users would like to maintain as part of the training process. Examples: error feedback in gradient compression, peers to communicate with next in GossipGrad etc. ddp_comm_hook: hook(state: object, bucket: dist._GradBucket) -> torch.futures.Future This callable function is called once the bucket is ready. The hook can perform whatever processing is needed and return a Future indicating completion of any async work (ex: allreduce). If the hook doesn't perform any communication, it can also just return a completed Future. The Future should hold the new value of grad bucket's tensors. Once a bucket is ready, c10d reducer would call this hook and use the tensors returned by the Future and copy grads to individual parameters. ddp_comm_wrapper: communication hook wrapper to support a communication hook such as FP16 compression as wrapper, which could be combined with ddp_comm_hook Examples: >>> from torch.distributed.algorithms.ddp_comm_hooks import ( # doctest: +SKIP ... default_hooks as default, ... powerSGD_hook as powerSGD, ... post_localSGD_hook as post_localSGD, ... ) >>> >>> # fp16_compress_hook for compress gradients >>> ddp_model = ... >>> register_ddp_comm_hook( # doctest: +SKIP ... model=ddp_model, ... ddp_comm_hook=default.fp16_compress_hook, ... ) >>> >>> # powerSGD_hook >>> ddp_model = ... >>> register_ddp_comm_hook( # doctest: +SKIP ... model=ddp_model, ... ddp_comm_state=powerSGD.PowerSGDState( ... process_group=None, ... matrix_approximation_rank=1, ... start_powerSGD_iter=5000, ... ), ... ddp_comm_hook=powerSGD.powerSGD_hook, ... ) >>> >>> # post_localSGD_hook >>> subgroup, _ = torch.distributed.new_subgroups() # doctest: +SKIP >>> ddp_model = ... >>> register_ddp_comm_hook( # doctest: +SKIP ... model=ddp_model, ... state=post_localSGD.PostLocalSGDState( ... process_group=None, ... subgroup=subgroup, ... start_localSGD_iter=1_000, ... ), ... ddp_comm_hook=post_localSGD.post_localSGD_hook, ... ) >>> >>> # fp16_compress_wrapper combined with other communication hook >>> ddp_model = ... >>> register_ddp_comm_hook( # doctest: +SKIP ... model=ddp_model, ... ddp_comm_state=powerSGD.PowerSGDState( ... process_group=None, ... matrix_approximation_rank=1, ... start_powerSGD_iter=5000, ... ), ... ddp_comm_hook=powerSGD.powerSGD_hook, ... ddp_comm_wrapper=default.fp16_compress_wrapper, ... ) Nz$DDP comm wrapper is provided, apply (z).zRegistering DDP comm hook: .)statehook)r __qualname__rZregister_comm_hook)rrrrrp/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/pytorch_lightning/utilities/distributed.pyregister_ddp_comm_hook!sVr)objrankrcCs0tj|kr|ndg}tjj||d|dS)N)srcr)torch distributedget_rankZbroadcast_object_list)r r!Zobjectsrrr_broadcast_object_listsr&)rrcs*tsdiSfddttjDS)aTThis distributed utility collects dictionary state across all processes. Args: state: Dictionary containing the state of the current process Returns: states: On global rank 0, a dictionary where the primary keys are the process rank and the values their associated states. Otherwise, returns None. rcsi|]}|t|qSr)r&).0r!rrr z0_collect_states_on_rank_zero..)new_distributed_availableranger#r$get_world_sizer(rr(r_collect_states_on_rank_zeros r.)argskwargsrcOstdt|i|S)Nz`pytorch_lightning.utilities.distributed.all_gather_ddp_if_available` has been deprecated in v1.8.0 and will be removed in v2.0.0. This function is internal but you can copy over its implementation.)rnew_all_gather_ddp_if_availabler/r0rrrall_gather_ddp_if_availablesr3)rcCstdtS)Nz`pytorch_lightning.utilities.distributed.distributed_available` has been deprecated in v1.8.0 and will be removed in v2.0.0. This function is internal but you can copy over its implementation.)rr+rrrrdistributed_availablesr4cOstdt|i|S)Nz`pytorch_lightning.utilities.distributed.gather_all_tensors` has been deprecated in v1.8.0 and will be removed in v2.0.0. This function is internal but you can copy over its implementation.)rnew_gather_all_tensorsr2rrrgather_all_tensorssr6c@sLeZdZdZed eeededddZeeee edfddd Z dS) AllGatherGrada!Gathers tensors from the whole group and stacks them. This implementation is copied from PyTorch. .. deprecated:: v1.8.0 This function has been deprecated in v1.8.0 in favor of :func:`torch.distributed.nn.functional.all_gather` and will be removed in v2.0.0. Nztorch.distributed.ProcessGroup)ctxtensorgrouprcsRtddd||_fddttjD}tjj||dtj|dd}|S) Nz`AllGatherGrad` has been deprecated in v1.8.0 and will be removed in v2.0.0. Use `torch.distributed.nn.functional.all_gather` instead.) stacklevelcsg|]}tqSr)r#Z zeros_like)r'_r9rr r*z)AllGatherGrad.forward..)r:r)Zdim)rr:r,r#r$r-Z all_gatherstack)r8r9r:Zgathered_tensorrr>rforwardszAllGatherGrad.forward)r8 grad_outputrcGs8t|}tjj|tjjjd|jd|tjdfS)NF)opZasync_opr:)r#catr$Z all_reduceZReduceOpZSUMr:r%)r8rBrrrbackwards zAllGatherGrad.backward)N) __name__ __module__r__doc__ staticmethodrrrrArrErrrrr7s r7cOstdt|i|S)Na6`pytorch_lightning.utilities.distributed.get_default_process_group_backend_for_device` has been deprecated in v1.8.0 and will be removed in v2.0.0. This function is internal but you can copy over its implementation. `lightning_fabric.utilities.distributed.get_default_process_group_backend_for_device` instead.)r0new_get_default_process_group_backend_for_devicer2rrr,get_default_process_group_backend_for_devicesrKcOstdt|i|S)Nz`pytorch_lightning.utilities.distributed.init_dist_connection` has been deprecated in v1.8.0 and will be removed in v2.0.0. This function is internal but you can copy over its implementation.)rnew_init_dist_connectionr2rrrinit_dist_connectionsrMcOstdt|i|S)Nz`pytorch_lightning.utilities.distributed.sync_ddp` has been deprecated in v1.8.0 and will be removed in v2.0.0. This function is internal but you can copy over its implementation.)r new_sync_ddpr2rrrsync_ddpsrOcOstdt|i|S)Nz`pytorch_lightning.utilities.distributed.sync_ddp_if_available` has been deprecated in v1.8.0 and will be removed in v2.0.0. This function is internal but you can copy over its implementation.)rnew_sync_ddp_if_availabler2rrrsync_ddp_if_availablesrQcCstdddlm}|S)Nz`pytorch_lightning.utilities.distributed.tpu_distributed` has been deprecated in v1.8.0 and will be removed in v2.0.0. This function is internal but you can copy over its implementation.r_tpu_distributed)rZ!lightning_fabric.accelerators.tpurSrRrrrtpu_distributeds  rTcOs"tdddlm}||i|S)Nz`pytorch_lightning.utilities.distributed.rank_zero_only` has been deprecated in v1.8.1 and will be removed in v2.0.0. You can import it from `pytorch_lightning.utilities` instead.r)rank_zero_only)r%pytorch_lightning.utilities.rank_zerorU)r/r0Znew_rank_zero_onlyrrrrUs  rU)NNN)1rHtypingrrrrrr#rZtorch.nn.parallel.distributedrZ&lightning_fabric.utilities.distributedr r1r r+r r5r rJr rLrrNrrPrVrrrobjectrintr&strr.r3r4r6ZautogradZFunctionr7rKrMrOrQboolrTrUrrrr sD          e "