a dGS@s.ddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl m Z m Z m Z mZmZmZddlZddlZddlmZddlmZddlmZddlmZddlZdd lmZmZdd lm Z dd l!m"Z"m#Z#m$Z$m%Z%dd l!m&Z'dd l(m)Z)m*Z*ddl+m,Z,ddl-m.Z.ddl/m0Z0ddl1m2Z2ddl3m4Z4ddl5m6Z6ddl7m8Z8ddl9m:Z:ddl;mZ>ddl?m@Z@ddlAmBZBddlCmDZDddlEmFZFddlGmHZHddlImJZJmKZKmLZLddlMmNZNmOZOmPZPmQZQe:rddlRmSZSneTZSejUVrdd lWmXZXeYeZZ[Gd!d"d"e@Z\dS)#N) timedelta)Path)AnyCallableDictListOptionalUnion)Tensor)Module)DistributedDataParallel) Optimizer) CheckpointIOClusterEnvironment)default_pg_timeout)_distributed_available-_get_default_process_group_backend_for_device_init_dist_connection_sync_ddp_if_availablegroup) _IS_WINDOWS_TORCH_GREATER_EQUAL_1_11)_optimizers_to_device) reset_seed)ReduceOp)LightningOptimizer)LightningDistributedModule)$_LightningPrecisionModuleWrapperBase)prepare_for_backward)_FAIRSCALE_AVAILABLE)PrecisionPlugin)_SubprocessScriptLauncher)ParallelStrategy) TBroadcast) TrainerFn)register_ddp_comm_hook)DeadlockDetectedException)rank_zero_inforank_zero_onlyrank_zero_warn) PredictStep STEP_OUTPUTTestStepValidationStep)OSS) ModelAveragercs@eZdZdZdZddddddddddef edeeej ee ee ee ee eeeeeeeeeeedd fdd Zeedd d Zeej dd d Zeedd dZejeddddZeedddZeeeefdddZeedddZeeedddZddddZddfdd Z ddddd Z!e"e#d!d"d#Z$ddd$d%Z%edd&d'Z&ddd(d)Z'ddd*d+Z(ddd,d-Z)ddd.d/Z*d`e+eegefee,d0e"feed1fd2d3 Z-ddd4d5Z.eeedd6d7Z/eedd8d9d:Z0dae1ee1d<d=d>Z2e3dd?d@dAZ4dddBdCZ5dbe3eeee,e6efe3dEdFdGZ7eee8d8dHdIZ9eeee8d8dJdKZ:eeee8d8dLdMZ;eee8d8dNdOZeddRdSdTZ?eddUdVZ@dddWdXZAdddYdZZBedd[d\d]ZCddfd^d_ ZDZES)c DDPStrategyzKStrategy for multi-process single-device training on one or multiple nodes.ZddpNzpl.accelerators.Accelerator) acceleratorparallel_devicescluster_environment checkpoint_ioprecision_pluginddp_comm_state ddp_comm_hookddp_comm_wrappermodel_averaging_periodprocess_group_backendtimeoutkwargsreturnc  svtj|||||dt|jjdd|_| |_||_||_ ||_ | |_ d|_ g|_ d|_d|_| |_| |_dS)N)r2r3r4r5r6z: initializing DDP pluginF)super__init__logdetail __class____name__ _num_nodes _ddp_kwargs_ddp_comm_state_ddp_comm_hook_ddp_comm_wrapper_model_averaging_period_model_averager_pids _sync_dir"_rank_0_will_call_children_scripts_process_group_backend_timeout) selfr2r3r4r5r6r7r8r9r:r;r<r=rDi/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/pytorch_lightning/strategies/ddp.pyrAKs(zDDPStrategy.__init__)r>cCsdSNTrTrRrTrTrUis_distributedoszDDPStrategy.is_distributedcCs|jdusJ|j|jSN)r3 local_rankrWrTrTrU root_devicesszDDPStrategy.root_devicecCs|jSrYrFrWrTrTrU num_nodesxszDDPStrategy.num_nodes)r]r>cCs ||_dSrYr\)rRr]rTrTrUr]|scCs|jdurt|jSdSNr)r3lenrWrTrTrU num_processesszDDPStrategy.num_processescCst|j|j|jdS)N)Z num_replicasrank)dictr]r` global_rankrWrTrTrUdistributed_sampler_kwargssz&DDPStrategy.distributed_sampler_kwargscCsdSrVrTrWrTrTrU _is_single_process_single_devicesz,DDPStrategy._is_single_process_single_devicecCs|jSrY)rPrWrTrTrUr;sz!DDPStrategy.process_group_backendcCs4|jdusJ|jjs0t|j|j|j|_d|_dSrV)r4Zcreates_processes_externallyr"r`r]Z _launcherrOrWrTrTrU_configure_launcherszDDPStrategy._configure_launchercs|tdSrY)setup_distributedr@setup_environmentrWrSrTrUrhszDDPStrategy.setup_environmentz pl.Trainer)trainerr>cCst||j|_|r"||jdus0J|j|||jj }|t j krz|j rz|j dusjJ|j |j |_ ||t j kr|||t|j|j|t j krddlmmmm}t|j|jr|dSr^)bool broadcastrO_should_run_deadlock_detection&_share_information_to_prevent_deadlockr2setupmodel_to_devicestatefnr%FITTING _layer_syncmodelapplyZsetup_precision_plugin configure_ddpZsetup_optimizersr optimizersr[Z>torch.distributed.algorithms.ddp_comm_hooks.post_localSGD_hook distributed algorithmsZddp_comm_hooksZpost_localSGD_hook isinstancerHZPostLocalSGDState_enable_model_averaging)rRriZ trainer_fnZ post_localSGDrTrTrUrns(     zDDPStrategy.setup)rtr>cCs6|}td|d|jtf||d|jS)z^Wraps the model into a :class:`~torch.nn.parallel.distributed.DistributedDataParallel` module.z&setting up DDP model with device ids: z , kwargs: )module device_ids)determine_ddp_device_idsrBrCrGr )rRrtr}rTrTrU _setup_modelszDDPStrategy._setup_modelcCsZt|jjdt||jt_| |_ |j dusBJt |j |j |j ddS)Nz: setting up distributed...)r<)rBrCrDrErset_world_ranksrcr)ra_get_process_group_backendrPr4rrQrWrTrTrUrgs zDDPStrategy.setup_distributedcCs|jpt|jSrY)rPrr[rWrTrTrUrsz&DDPStrategy._get_process_group_backendcCsL|jdurdS|j|j|j|j|j|j|j|jt_ dSrY) r4Zset_global_rank node_rankr`rZZset_world_sizer]rcr)rarWrTrTrUrs  zDDPStrategy.set_world_rankscCs|jdd|jd<dS)Nfind_unused_parametersT)rGgetrWrTrTrUpre_configure_ddpszDDPStrategy.pre_configure_ddpcCsRt|jjd|jjdkrN|jrNt|jt s6Jt |j|j |j |j ddS)Nz: registering ddp hooksZcuda)rtr7r8r9)rBrCrDrEr[typererzrtr r&rHrIrJrWrTrTrU_register_ddp_hooksszDDPStrategy._register_ddp_hookscCst|jjd|jdur&tdddlm}m}m }|j D]`}t |t rT|j }tsbt ||nd}|st ||strt |tst ||r@td|jjdq@|jdusJtjjjjj|j|jjd|_dS) Nz.: reinitializing optimizers with post localSGDz\Post-localSGD algorithm is used, but model averaging period is not provided to DDP strategy.r)DistributedOptimizerPostLocalSGDOptimizerZeroRedundancyOptimizerFzKCurrently model averaging cannot work with a distributed optimizer of type .)ZperiodZ warmup_steps)rBrCrDrErK ValueErrorZtorch.distributed.optimrrrrwrzrZ _optimizerrr r/rHtorchrxryZmodel_averagingZ averagersZPeriodicModelAveragerZstart_localSGD_iterrL)rRrrr optimizerZis_distributed_optimizerrTrTrUr{s:     z#DDPStrategy._enable_model_averagingzpl.LightningModule)ropt_idxclosurertr=r>c sLtj||||fi|}|jdur(|Sdd|jD}|jt||S)a|Performs the actual optimizer step. Args: optimizer: the optimizer performing the step opt_idx: index of the current optimizer closure: closure calculating the loss value model: reference to the model, optionally defining optimizer step related hooks **kwargs: Any extra arguments to ``optimizer.step`` NcSs(g|] }|dD]}|jdur|qqS)paramsN)Zgrad).0rparamrTrTrU z.DDPStrategy.optimizer_step..)r@optimizer_steprLZ param_groupsZaverage_parametersiter)rRrrrrtr=Zoptimizer_outputrrSrTrUrs  zDDPStrategy.optimizer_stepcCsPt|jjd|t|jtjt fs2J| t |j|_| dS)Nz%: configuring DistributedDataParallel) rBrCrDrErrzrtplZLightningModulerrrrrWrTrTrUrv"s zDDPStrategy.configure_ddpcCs|jjdkrdS|jjgS)Ncpu)r[rindexrWrTrTrUr~)s z$DDPStrategy.determine_ddp_device_ids)argsr=r>cOs:ts dStjdkr,tjj|dn tjdS)NZnccl)r})rrrx get_backendbarrierr~rRrr=rTrTrUr.s zDDPStrategy.barrierr)objsrcr>cCs2|g}|j|krdg}tjj||tjd|dS)Nrr)rcrrxZbroadcast_object_list_groupZWORLD)rRrrrTrTrUrk6s  zDDPStrategy.broadcast) closure_lossr>cCs6t|jtsdS|jdusJ|jjs2t|j|dS)z.Run before precision plugin executes backward.N)rzrtr lightning_moduleautomatic_optimizationr)rRrrTrTrU pre_backward=s  zDDPStrategy.pre_backwardcCs<t|jjd|jd|jdus*J|j|jdS)Nz: moving model to device [z]...)rBrCrDrEr[rttorWrTrTrUroEszDDPStrategy.model_to_devicemean)tensorr reduce_opr>cCst|trt|||d}|S)aReduces a tensor from several distributed processes to one aggregated tensor. Args: tensor: the tensor to sync and reduce group: the process group to gather results from. Defaults to all processes (world) reduce_op: the reduction operation. Defaults to 'mean'/'avg'. Can also be a string 'sum' to calculate the sum during reduction. Return: reduced value, except when the input was not a tensor the output remains is unchanged )r)rzr r)rRrrrrTrTrUreduceJs zDDPStrategy.reducecOsL|jdusJ|j |j|i|WdS1s>0YdSrY)rtr6Ztrain_step_contextrrTrTrU training_step\s zDDPStrategy.training_stepcOs|j|jdusJ|jdus(J|jjjjtjkrX|j|i|WdSt |jt shJ|jj |i|WdSWdn1s0YdSrY) r6Zval_step_contextrrtrirprqr%rrrzr.validation_steprrTrTrUras zDDPStrategy.validation_stepcOsP|j2t|jtsJ|jj|i|WdS1sB0YdSrY)r6Ztest_step_contextrzrtr- test_steprrTrTrUrms zDDPStrategy.test_stepcOsP|j2t|jtsJ|jj|i|WdS1sB0YdSrY)r6Zpredict_step_contextrzrtr+ predict_steprrTrTrUrrs zDDPStrategy.predict_stepcCs0|jdusJ|jjs,|jdus$Jd|j_dSrV)rrrtZrequire_backward_grad_syncrWrTrTrUpost_training_stepwszDDPStrategy.post_training_step)strategy_registryr>cCs.|jd|ddd|j|j||jjddS)NZ ddp_find_unused_parameters_falsez3DDP Strategy with `find_unused_parameters` as FalseF) descriptionr)r)register strategy_namerDrE)clsrrTrTrUregister_strategies}szDDPStrategy.register_strategiescCstdddkp|jS)aDetermines whether the plugin will perform process reconciliation in case of errors. If the environment variable `PL_RECONCILE_PROCESS` is set, run detection regardless of the cluster environment. By default this is disabled. Otherwise, if the cluster environment creates the processes, allow the scheduler / parent process to perform the process termination, external to Lightning. ZPL_RECONCILE_PROCESS01)osgetenvrOrWrTrTrUrlsz*DDPStrategy._should_run_deadlock_detectioncCsh||jdkrt|_g}d}t|jD](}|||j|||j |j7}q.||j |_dSr^) _share_pidsrZtempfilemkdtemprNranger]appendrk world_sizer)rRZ sync_dirsZglobal_node_rank_zero_rTrTrUrms  z2DDPStrategy._share_information_to_prevent_deadlockcCsL||tjt|jd}| }t |t r@|n|g|_ dS)z3Make all DDP processes aware of all processes pids.)deviceN) rZ all_gatherrrrgetpidr[rnumpytolistrzlistrM)rRZpidsrTrTrUrszDDPStrategy._share_pids)tracer>cCs|jdkrdS|sdS|j}|s0tddSt|jdddtdtj ||j dt dtt||j|jkrdS|jD]}|tkrt|tjqt|td|j d|dS) NzQError handling mechanism for deadlock detection is uninitialized. Skipping check.T)parentsexist_okz.plzDeadLock detected from rank: z )rrlrNr*rmkdirrsaverpathjoinrctimesleepr_listdirr]rMrkillsignalSIGKILLshutilrmtreer')rRrZsync_dirpidrTrTrUreconciliate_processess$     z"DDPStrategy.reconciliate_processescst|jjd|j}t|jtr\trV|jj sV|j drVt d|jjd||_|dur|j dur|j jjtjkr|jr|jdusJ|j|j|_tdS)Nz: tearing down strategyZcan_set_static_graphzyYour model can run with static graph optimizations. For future training runs, we suggest you pass `Trainer(..., strategy=z%(static_graph=True))` to enable them.)rBrCrDrErrzrtr rZ static_graphZ_get_ddp_logging_datarr(Z_trainerrprqr%rrrsrevertr@teardown)rRZ pl_modulerSrTrUrs4 zDDPStrategy.teardown)N)r)Nr)FrE __module__ __qualname____doc__rrrrrrrrr!objectrintstrrrrApropertyrjrXr[r]setterr`rrdrer;rfrhrnr r rrgrrrrr{r r rrvr~rr$rkr rrorrr,rrrrr classmethodrrlrmrrr __classcell__rTrTrSrUr1Fs $#  #     #r1)]loggingrrrrrdatetimerpathlibrtypingrrrrrr rZtorch.distributedr Ztorch.nnr Ztorch.nn.parallel.distributedr Ztorch.optim.optimizerr Zpytorch_lightningrZlightning_fabric.pluginsrrZ5lightning_fabric.plugins.collectives.torch_collectiverZ&lightning_fabric.utilities.distributedrrrrrrZ"lightning_fabric.utilities.importsrrZ$lightning_fabric.utilities.optimizerrZlightning_fabric.utilities.seedrZ lightning_fabric.utilities.typesrZ pytorch_lightning.core.optimizerrZpytorch_lightning.overridesrZ pytorch_lightning.overrides.baserZ'pytorch_lightning.overrides.distributedrZ%pytorch_lightning.overrides.fairscaler Z#pytorch_lightning.plugins.precisionr!Z8pytorch_lightning.strategies.launchers.subprocess_scriptr"Z%pytorch_lightning.strategies.parallelr#Z%pytorch_lightning.strategies.strategyr$Z pytorch_lightning.trainer.statesr%Z'pytorch_lightning.utilities.distributedr&Z&pytorch_lightning.utilities.exceptionsr'Z%pytorch_lightning.utilities.rank_zeror(r)r*Z!pytorch_lightning.utilities.typesr+r,r-r.Zfairscale.optimr/rrxZ is_availableZ6torch.distributed.algorithms.model_averaging.averagersr0 getLoggerrErBr1rTrTrTrUsX