o ziuJ@sddlZddlmZddlmZddlmZmZmZm Z m Z m Z m Z m Z ddlZddlZddlmZddlmZddlmZddlmZdd lmZdd lmZddlZdd lmZm Z dd l!m"Z"dd l#m$Z$ddl%m&Z&m'Z'm(Z(m)Z)ddl%m*Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1ddl2m3Z3ddl4m5Z5ddl6m7Z7m8Z8m9Z9ddl:m;Z;ddlZ>ddl?m@Z@ddlAmBZBmCZCddlDmEZEddlFmGZGddlHmIZImJZJmZerddlKmLZLeMeNZOdZPGdd d e@ZQGd!d"d"eCZRdS)#N) nullcontext) timedelta) TYPE_CHECKINGAnyCallableDictListLiteralOptionalUnion)rank_zero_only)Tensor)Module)DistributedDataParallel) Optimizer)override) CheckpointIOClusterEnvironment)default_pg_timeout)_StrategyRegistry)_distributed_is_initialized-_get_default_process_group_backend_for_device_init_dist_connection_sync_ddp_if_availablegroup) _IS_WINDOWS)_optimizers_to_device) reset_seed)ReduceOp)LightningOptimizer)_register_ddp_comm_hook_sync_module_statesprepare_for_backward) Precision)_MultiProcessingLauncher_SubprocessScriptLauncher)ParallelStrategy) TBroadcast_ForwardRedirection) TrainerFn_augment_message)rank_zero_deprecationrank_zero_infor ) ModelAverager)ddp_fork%ddp_fork_find_unused_parameters_false$ddp_fork_find_unused_parameters_true ddp_notebook)ddp_notebook_find_unused_parameters_false(ddp_notebook_find_unused_parameters_truecseZdZdZddddddddddedf deddeeejdee dee d ee d ee d ee d ee d eedeedeededdeddffdd ZedefddZeedejfddZedefddZejdeddfddZedefddZeedeeeffd d!Zedeefd"d#Zed]d$d%Zed]fd&d' Z ed^d*d+Z!ed,e"de#fd-d.Z$d]d/d0Z%defd1d2Z&d]d3d4Z'd]d5d6Z(d]d7d8Z)e d_d9e*d:e gefd,ee+d;e"fdedef fdd?Z-deeefd@dAZ.edBededdfdCdDZ/ed`dFe0dGede0fdHdIZ1edJe2ddfdKdLZ3ed]dMdNZ4e OdadPe2dQeedRee+e5efde2fdSdTZ6e7edUe8ddfdVdWZ9edXe:ddfdYdZZ;ed]fd[d\ Z<Z=S)b DDPStrategyzKStrategy for multi-process single-device training on one or multiple nodes.Npopen acceleratorzpl.accelerators.Acceleratorparallel_devicescluster_environment checkpoint_ioprecision_pluginddp_comm_state ddp_comm_hookddp_comm_wrappermodel_averaging_periodprocess_group_backendtimeout start_method)r7spawnfork forkserverkwargsreturnc  srtj|||||dt|jjdt|_d|_| |_ ||_ ||_ ||_ | |_ d|_| |_| |_| |_dS)N)r8r9r:r;r<z: initializing DDP strategy)super__init__logdebug __class____name___DDPForwardRedirection_forward_redirection _num_nodes _ddp_kwargs_ddp_comm_state_ddp_comm_hook_ddp_comm_wrapper_model_averaging_period_model_averager_process_group_backend_timeout _start_method)selfr8r9r:r;r<r=r>r?r@rArBrCrGrNT/home/ubuntu/.local/lib/python3.10/site-packages/pytorch_lightning/strategies/ddp.pyrKGs& zDDPStrategy.__init__cCstdt|jddddS)z1Legacy property kept for backwards compatibility.`z3.is_distributed` is deprecated. Use is discouraged.) stacklevelT)r-typerOr\r^r^r_is_distributedkszDDPStrategy.is_distributedcCs|jdusJ|j|jSN)r9 local_rankrdr^r^r_ root_devicess zDDPStrategy.root_devicecC|jSrfrRrdr^r^r_ num_nodesyzDDPStrategy.num_nodesrkcCs ||_dSrfrj)r\rkr^r^r_rk}s cCs|jdur t|jSdSNr)r9lenrdr^r^r_ num_processesszDDPStrategy.num_processescCs|j|j|jdS)N) num_replicasrank)rkro global_rankrdr^r^r_distributed_sampler_kwargssz&DDPStrategy.distributed_sampler_kwargscCrirf)rYrdr^r^r_rArlz!DDPStrategy.process_group_backendcCsD|jdusJ|jdkrt|j|j|j|_dSt||jd|_dS)Nr7)rC)r:r[r&rork _launcherr%rdr^r^r__configure_launchers zDDPStrategy._configure_launchercst|dSrf)rJsetup_environmentsetup_distributedrdr]r^r_rvs  zDDPStrategy.setup_environmenttrainer pl.TrainercCs|jdusJ|j||jj}|jdusJ|tjkr(|jr(|j|j|_|j |j| |tjkrB| | |nt|j||tjkrst|j|jddlmmmm}t|j|jru|dSdSdSrm)r8setupstatefnmodelr*FITTING _layer_syncapplyr<convert_modulemodel_to_device configure_ddpsetup_optimizersr"setup_precision_pluginr optimizersrh>torch.distributed.algorithms.ddp_comm_hooks.post_localSGD_hook distributed algorithmsddp_comm_hookspost_localSGD_hook isinstancerTPostLocalSGDState_enable_model_averaging)r\rx trainer_fn post_localSGDr^r^r_rzs(      zDDPStrategy.setupr}cCs~|}td|d|j|durtjtjnt}|t d||d|jWdS1s8wYdS)z^Wraps the model into a :class:`~torch.nn.parallel.distributed.DistributedDataParallel` module.z&setting up DDP model with device ids: z , kwargs: N)module device_idsr^) determine_ddp_device_idsrLrMrStorchcudastreamStreamrr)r\r}rctxr^r^r_ _setup_models  $zDDPStrategy._setup_modelcCsRt|jjdt|||_|jdusJt |j|j|j ddS)Nz: setting up distributed...)rB) rLrMrNrOrset_world_ranks_get_process_group_backendrYr:rrZrdr^r^r_rws  zDDPStrategy.setup_distributedcCs|jpt|jSrf)rYrrhrdr^r^r_rsz&DDPStrategy._get_process_group_backendcCsJ|jdur|j|j|j|j|j|j|j|jt_ t _ dSrf) r:set_global_rank node_rankrorgset_world_sizerkrrr rqutils_rank_zero_onlyrdr^r^r_rs zDDPStrategy.set_world_rankscCsPt|jjd|jjdkr&t|jtsJt |j|j |j |j ddSdS)Nz: registering ddp hooksr)r}r=r>r?) rLrMrNrOrhrcrr}rr!rTrUrVrdr^r^r__register_ddp_hookss  zDDPStrategy._register_ddp_hookscCst|jjd|jdurtdddlm}m}m }|j D]&}t |t r*|j }ts1t ||nd}t |||fs<|rFtd|jjdq |jdusNJtjjjjj|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 .)period warmup_steps)rLrMrNrOrW ValueErrortorch.distributed.optimrrrrrr _optimizerrrTrrrmodel_averaging averagersPeriodicModelAveragerstart_localSGD_iterrX)r\rrr optimizeris_distributed_optimizerr^r^r_rs*      z#DDPStrategy._enable_model_averagingrclosurepl.LightningModulec sJtj|||fi|}|jdur|Sdd|jD}|jt||S)aIPerforms the actual optimizer step. Args: optimizer: the optimizer performing the step 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)grad).0rparamr^r^r_ s(z.DDPStrategy.optimizer_step..)rJoptimizer_steprX param_groupsaverage_parametersiter)r\rrr}rGoptimizer_outputrr]r^r_rs  zDDPStrategy.optimizer_stepcCs@t|jjdt|jtjsJ||j|_| dS)Nz%: configuring DistributedDataParallel) rLrMrNrOrr}plLightningModulerrrdr^r^r_rs zDDPStrategy.configure_ddpcCs|jjdkrdS|jjgS)Ncpu)rhrcindexrdr^r^r_rs  z$DDPStrategy.determine_ddp_device_idsargscOs<tsdStjdkrtjj|ddStjdS)Nnccl)r)rrr get_backendbarrierr)r\rrGr^r^r_r#s zDDPStrategy.barrierrobjsrccCs,ts|S|g}tjj||tjd|dS)Nrr)rrrbroadcast_object_list_groupWORLD)r\rrr^r^r_ broadcast-s zDDPStrategy.broadcast closure_losscCs:t|jtsdS|jdusJ|jjst|j|dSdS)z.Run before precision plugin executes backward.N)rr}rlightning_moduleautomatic_optimizationr#)r\rr^r^r_ pre_backward6s zDDPStrategy.pre_backwardcCs<t|jjd|jd|jdusJ|j|jdS)Nz: moving model to device [z]...)rLrMrNrOrhr}tordr^r^r_r?szDDPStrategy.model_to_devicemeantensorr reduce_opcCst|tr t|||dS|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)rr r)r\rrrr^r^r_reduceEs zDDPStrategy.reducestrategy_registryc Csfd}|D]\}}|j||d|d|dqd}|D]\}}}|j||d|d|d||dqdS) N))ddpr7) ddp_spawnrD)r0rE)r3rEz"DDP strategy with `start_method` '') descriptionrC)) ddp_find_unused_parameters_falseFr7)ddp_find_unused_parameters_trueTr7)&ddp_spawn_find_unused_parameters_falseFrD)%ddp_spawn_find_unused_parameters_trueTrD)r1FrE)r2TrE)r4FrE)r5TrEz.DDP strategy with `find_unused_parameters` as z and `start_method` ')rfind_unused_parametersrC)register)clsrentriesnamerCfupr^r^r_register_strategiesYs$   zDDPStrategy.register_strategies exceptioncCst|ddddS)Nz>.*Expected to have finished reduction in the prior iteration.*ayIt looks like your LightningModule has parameters that were not used in producing the loss returned by training_step. If this is intentional, you must enable the detection of unused parameters in DDP, either by setting the string value `strategy='ddp_find_unused_parameters_true'` or by setting the flag in the strategy with `strategy=DDPStrategy(find_unused_parameters=True)`.)pattern new_messager+)r\rr^r^r_ on_exception}s  zDDPStrategy.on_exceptioncst|jjd|j}t|jtr,|jjs)|j dr)t d|jjd||_|durO|j durO|j j jtjkrO|jrO|jdusGJ|j|j|_tdS)Nz: tearing down strategycan_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.)rLrMrNrOrrr}r static_graph_get_ddp_logging_datagetr._trainerr{r|r*r~rrevertrJteardown)r\ pl_moduler]r^r_rs$  zDDPStrategy.teardown)rHN)rxryrHNrf)r)Nr)>rO __module__ __qualname____doc__rr rrdevicerrr$objectrintstrrr rrKpropertyboolrerrhrksetterrorrsrArurvrzrrrrwrrrrrr rrrrr(rr rrrr classmethodrr BaseExceptionrr __classcell__r^r^r]r_r6Ds      $           " r6c@s@eZdZededdddfddZededdddfdd ZdS) rPwrapper_moduleoriginal_modulerrHNcC"t|tr |jsd|_dSdSdS)NFrrrrequire_backward_grad_syncr\rrr^r^r_on_after_inner_forwards z-_DDPForwardRedirection.on_after_inner_forwardcCr)NTrrr^r^r_on_after_outer_forwards z-_DDPForwardRedirection.on_after_outer_forward)rOrrrrrrr^r^r^r_rPs rP)Slogging contextlibrdatetimertypingrrrrrr r r rtorch.distributed"lightning_utilities.core.rank_zeror rr torch.nnrtorch.nn.parallel.distributedrtorch.optim.optimizerrtyping_extensionsrpytorch_lightningrlightning_fabric.pluginsrr5lightning_fabric.plugins.collectives.torch_collectiverlightning_fabric.strategiesr&lightning_fabric.utilities.distributedrrrrrr"lightning_fabric.utilities.importsr$lightning_fabric.utilities.optimizerrlightning_fabric.utilities.seedr lightning_fabric.utilities.typesr pytorch_lightning.core.optimizerr 'pytorch_lightning.overrides.distributedr!r"r##pytorch_lightning.plugins.precisionr$&pytorch_lightning.strategies.launchersr%r&%pytorch_lightning.strategies.parallelr'%pytorch_lightning.strategies.strategyr(r) pytorch_lightning.trainer.statesr*&pytorch_lightning.utilities.exceptionsr,%pytorch_lightning.utilities.rank_zeror-r.6torch.distributed.algorithms.model_averaging.averagersr/ getLoggerrOrL_DDP_FORK_ALIASESr6rPr^r^r^r_sN  (                     d