o 8wiiJ@sddlZddlmZddlmZddlmZmZmZm Z m Z m Z ddl Z ddl Z ddlmZddl mZddlmZddlmZdd lmZdd lmZddlZdd lmZmZdd lm Z dd l!m"Z"ddl#m$Z$m%Z%m&Z&m'Z'ddl#m(Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1ddl2m3Z3ddl4m5Z5m6Z6m7Z7ddl8m9Z9ddl:m;Z;mZ>ddl?m@Z@mAZAddlBmCZCddlDmEZEddlFmGZGmHZHmZerddlImJZJeKeLZMdZNGdd d e>ZOGd!d"d"eAZPdS)#N) nullcontext) timedelta) TYPE_CHECKINGAnyCallableLiteralOptionalUnion)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)r5spawnfork forkserverkwargsreturnc  srtj|||||dt|jjdt|_d|_| |_ ||_ ||_ ||_ | |_ d|_| |_| |_| |_dS)N)r6r7r8r9r: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)selfr6r7r8r9r:r;r<r=r>r?r@rArErL]/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/pytorch_lightning/strategies/ddp.pyrIGs& zDDPStrategy.__init__cCstdt|jddddS)z1Legacy property kept for backwards compatibility.`z3.is_distributed` is deprecated. Use is discouraged.) stacklevelT)r+typerMrZr\r\r]is_distributedkszDDPStrategy.is_distributedcCs|jdusJ|j|jSN)r7 local_rankrbr\r\r] root_devicess zDDPStrategy.root_devicecC|jSrdrPrbr\r\r] num_nodesyzDDPStrategy.num_nodesricCs ||_dSrdrh)rZrir\r\r]ri}s cCs|jdur t|jSdSNr)r7lenrbr\r\r] num_processesszDDPStrategy.num_processescCs|j|j|jdS)N) num_replicasrank)rirm global_rankrbr\r\r]distributed_sampler_kwargssz&DDPStrategy.distributed_sampler_kwargscCrgrd)rWrbr\r\r]r?rjz!DDPStrategy.process_group_backendcCsD|jdusJ|jdkrt|j|j|j|_dSt||jd|_dS)Nr5)rA)r8rYr$rmri _launcherr#rbr\r\r]_configure_launchers zDDPStrategy._configure_launchercst|dSrd)rHsetup_environmentsetup_distributedrbr[r\r]rts  zDDPStrategy.setup_environmenttrainer pl.TrainercCs|jdusJ|j||jj}|jdusJ|tjkr(|jr(|j|j|_|j |j| |tjkrB| | |nt|j||tjkrst|j|jddlmmmm}t|j|jru|dSdSdSrk)r6setupstatefnmodelr(FITTING _layer_syncapplyr:convert_modulemodel_to_device configure_ddpsetup_optimizersr setup_precision_pluginr optimizersrf>torch.distributed.algorithms.ddp_comm_hooks.post_localSGD_hook distributed algorithmsddp_comm_hookspost_localSGD_hook isinstancerRPostLocalSGDState_enable_model_averaging)rZrv trainer_fn post_localSGDr\r\r]rxs(      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_idsrJrKrQtorchcudastreamStreamrr )rZr{rctxr\r\r] _setup_models  $zDDPStrategy._setup_modelcCsRt|jjdt|||_|jdusJt |j|j|j ddS)Nz: setting up distributed...)r@) rJrKrLrMrset_world_ranks_get_process_group_backendrWr8rrXrbr\r\r]rus  zDDPStrategy.setup_distributedcCs|jpt|jSrd)rWrrfrbr\r\r]rsz&DDPStrategy._get_process_group_backendcCsJ|jdur|j|j|j|j|j|j|j|jt_ t _ dSrd) r8set_global_rank node_rankrmreset_world_sizerirpr routils_rank_zero_onlyrbr\r\r]rs zDDPStrategy.set_world_rankscCsPt|jjd|jjdkr&t|jtsJt |j|j |j |j ddSdS)Nz: registering ddp hooksr)r{r;r<r=) rJrKrLrMrfrarr{r rrRrSrTrbr\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)rJrKrLrMrU ValueErrortorch.distributed.optimrrrrrr _optimizerrrRrrrmodel_averaging averagersPeriodicModelAveragerstart_localSGD_iterrV)rZrrr 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..)rHoptimizer_steprV param_groupsaverage_parametersiter)rZrrr{rEoptimizer_outputrr[r\r]rs  zDDPStrategy.optimizer_stepcCs@t|jjdt|jtjsJ||j|_| dS)Nz%: configuring DistributedDataParallel) rJrKrLrMrr{plLightningModulerrrbr\r\r]rs zDDPStrategy.configure_ddpcCs|jjdkrdS|jjgS)Ncpu)rfraindexrbr\r\r]rs  z$DDPStrategy.determine_ddp_device_idsargscOs<tsdStjdkrtjj|ddStjdS)Nnccl)r)rrr get_backendbarrierr)rZrrEr\r\r]r#s zDDPStrategy.barrierrobjsrccCs,ts|S|g}tjj||tjd|dS)Nrr)rrrbroadcast_object_list_groupWORLD)rZrrr\r\r] broadcast-s zDDPStrategy.broadcast closure_losscCs:t|jtsdS|jdusJ|jjst|j|dSdS)z.Run before precision plugin executes backward.N)rr{r lightning_moduleautomatic_optimizationr!)rZrr\r\r] pre_backward6s zDDPStrategy.pre_backwardcCs<t|jjd|jd|jdusJ|j|jdS)Nz: moving model to device [z]...)rJrKrLrMrfr{torbr\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)rZrrrr\r\r]reduceEs zDDPStrategy.reducestrategy_registryc Csfd}|D]\}}|j||d|d|dqd}|D]\}}}|j||d|d|d||dqdS) N))ddpr5) ddp_spawnrB)r.rC)r1rCz"DDP strategy with `start_method` '') descriptionrA)) ddp_find_unused_parameters_falseFr5)ddp_find_unused_parameters_trueTr5)&ddp_spawn_find_unused_parameters_falseFrB)%ddp_spawn_find_unused_parameters_trueTrB)r/FrC)r0TrC)r2FrC)r3TrCz.DDP strategy with `find_unused_parameters` as z and `start_method` ')rfind_unused_parametersrA)register)clsrentriesnamerAfupr\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))rZrr\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.)rJrKrLrMrrr{r static_graph_get_ddp_logging_datagetr,_trainerryrzr(r|r}revertrHteardown)rZ pl_moduler[r\r]rs$  zDDPStrategy.teardown)rFN)rvrwrFNrd)r)Nr)>rM __module__ __qualname____doc__rrlistrdevicerrr"objectrintstrrrrrIpropertyboolrcrrfrisetterrmdictrqr?rsrtrxr r rrurrrrrr rrrrr&rr rrrr classmethodrr BaseExceptionrr __classcell__r\r\r[r]r4Ds      $           " r4c@s@eZdZededdddfddZededdddfdd ZdS) rNwrapper_moduleoriginal_modulerrFNcC"t|tr |jsd|_dSdSdS)NFrr rrequire_backward_grad_syncrZrrr\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)rMrrrr rrr\r\r\r]rNs rN)Qlogging contextlibrdatetimertypingrrrrrr rtorch.distributed"lightning_utilities.core.rank_zeror rr torch.nnr torch.nn.parallel.distributedr torch.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.distributedrr 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- getLoggerrMrJ_DDP_FORK_ALIASESr4rNr\r\r\r]sN                        d