o oi+ @s^ddlZddlmZmZmZmZmZmZmZm Z m Z m Z ddl Z ddl m Z ddlmZddlmZmZddlmZmZddlmZddlmZmZdd lmZd e e eeeefd e ee ej ffd d Z!deded dfddZ"   d dedee#deedeed df ddZ$de j%j&d dfddZ'GdddeZ(Gddde(Z)GdddZ*dS)!N) AnyCallableDictIterableIteratorListOptionalSizedUnioncast)Tensor)DistributedDataParallel)DistributedSamplerSampler)Selfoverride)_DatasetSamplerWrapper)rank_zero_debugrank_zero_info)_SizedIterableobjreturncCsPt|tr|gSt|ttfrtjtt|St|tr&tjtt| SgS)z?Recursively find all tensors contained in the specified object.) isinstancer listtuple itertoolschainmap _find_tensorsdictvalues)rr![/home/ubuntu/.local/lib/python3.10/site-packages/lightning/pytorch/overrides/distributed.pyrs  rmodeloutputcCsbtr,|jr,d|_|jr|jstt|ng}ttj j j |j }| ||dSd|_dS)NTF)torchis_grad_enabledrequire_backward_grad_syncrequire_forward_param_syncfind_unused_parameters static_graphrrr _C_distributed_c10dReducerreducer_rebuild_bucketsprepare_for_backward)r#r$argsr.r!r!r"r0,s r0ddp_comm_state ddp_comm_hookddp_comm_wrappercCs^|durdS|}|durtd|jd|jd||}td|jd|j||ddS)a 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 ( 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( model=ddp_model, ddp_comm_hook=default.fp16_compress_hook, ) # powerSGD_hook ddp_model = ... _register_ddp_comm_hook( 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() ddp_model = ... _register_ddp_comm_hook( 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( 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__rregister_comm_hook)r#r2r3r4r!r!r"_register_ddp_comm_hook=sWr;modulecCsJt|dr t|jnt}ddlm}ddlm}|||dd|ddS)zeTaken from https://github.com/pytorch/pytorch/blob/v2.0.0/torch/nn/parallel/distributed.py#L675-L682.!_ddp_params_and_buffers_to_ignorer)_get_default_group)_sync_module_statesi)srcparams_and_buffers_to_ignoreN)hasattrsetr="torch.distributed.distributed_c10dr>torch.distributed.utilsr?)r<parameters_to_ignorer>torch_sync_module_statesr!r!r"r?s   r?csHeZdZdZdededdffdd Zedeee fdd Z Z S) UnrepeatedDistributedSampleraGA fork of the PyTorch DistributedSampler that doesn't repeat data, instead allowing the number of batches per process to be off-by-one from each other. This makes this sampler usable for predictions (it's deterministic and doesn't require shuffling). It is potentially unsafe to use this sampler for training, because during training the DistributedDataParallel syncs buffers on each forward pass, so it could freeze if one of the processes runs one fewer batch. During prediction, buffers are only synced on the first batch, so this is safe to use as long as each process runs at least one batch. We verify this in an assert. Taken from https://github.com/jpuigcerver/PyLaia/blob/v1.0.0/laia/data/unpadded_distributed_sampler.py and https://github.com/pytorch/pytorch/issues/25162#issuecomment-634146002 r1kwargsrNcsntj|i|t|jtstdtt|jt|j|j |_ t|j|_ |j dks3|j dks5JdSdS)N6The given dataset must implement the `__len__` method.r) super__init__rdatasetr TypeErrorlenrangerank num_replicas num_samples total_size)selfr1rI __class__r!r"rMs    z%UnrepeatedDistributedSampler.__init__cCst|jts td|jr$t}||jtj t |j|d }n t t t |j}t ||jks6J||j|j|j}t ||jksIJt|S)NrJ) generator)rrNr rOshuffler% Generator manual_seedepochrandpermrPtolistrrQrUrRrSrTiter)rVgindicesr!r!r"__iter__s  z%UnrepeatedDistributedSampler.__iter__) __name__ __module__r9__doc__rrMrrrintrc __classcell__r!r!rWr"rHs   rHcsPeZdZdZdeeefdededdffdd Ze de ffd d Z Z S) #UnrepeatedDistributedSamplerWrapperz_Equivalent class to ``DistributedSamplerWrapper`` but for the ``UnrepeatedDistributedSampler``.samplerr1rIrNcs"tjt|g|Ri|dSN)rLrMr)rVrjr1rIrWr!r"rMs"z,UnrepeatedDistributedSamplerWrapper.__init__cs"jfddtDS)Nc3s|]}j|VqdSrk)rN).0indexrVr!r" sz?UnrepeatedDistributedSamplerWrapper.__iter__..)rNresetrLrcrnrWrnr"rcs z,UnrepeatedDistributedSamplerWrapper.__iter__) rdrer9rfr rrrrMrrrcrhr!r!rWr"ris &ric@sfeZdZdZdeddfddZdeefddZde fd d Z defd d Z de e effd dZdS)_IndexBatchSamplerWrapperz\This class is used to wrap a :class:`torch.utils.data.BatchSampler` and capture its indices. batch_samplerrNcCs,g|_dd|jD|_||_d|_dS)NcSsi|] \}}|dvr||qS))__next__rc__len__ __getstate__r!)rlkvr!r!r" s z6_IndexBatchSamplerWrapper.__init__..)seen_batch_indices__dict__items_batch_sampler _iterator)rVrrr!r!r"rMs  z"_IndexBatchSamplerWrapper.__init__cCs(|jdusJt|j}|j||Srk)r}nextryappend)rVbatchr!r!r"rss  z"_IndexBatchSamplerWrapper.__next__cCsg|_t|j|_|Srk)ryr`r|r}rnr!r!r"rcs z"_IndexBatchSamplerWrapper.__iter__cCs t|jSrk)rPr|rnr!r!r"rts z!_IndexBatchSamplerWrapper.__len__cCs|j}d|d<|S)Nr})rzcopy)rVr7r!r!r"ru s z&_IndexBatchSamplerWrapper.__getstate__)rdrer9rfrrMrrgrsrrcrtrstrrrur!r!r!r"rqs rq)NNN)+rtypingrrrrrrrr r r r%r torch.nn.parallel.distributedr torch.utils.datarrtyping_extensionsrr&lightning.fabric.utilities.distributedr%lightning.pytorch.utilities.rank_zerorr!lightning.pytorch.utilities.typesrrrrrrr0objectr;nnModuler?rHrirqr!r!r!r"sD 0      f,