o mi6@sdZddlZddlZddZddZddZejjj fd ej fd d Z d d Z dej ej fddZddejej defddZdddZddZd dejeeffddZdS)!zTorch distributed utilities.NcCtjr tjSdS)Nr)torch distributedis_initializedget_rankrrD/home/ubuntu/SpeechTokenizer/speechtokenizer/quantization/distrib.pyrank  r cCrN)rrrget_world_sizerrrr world_sizer rcCs tdkSr )rrrrris_distributeds rtensorcCstr tj||SdSN)rrr all_reduce)roprrrr srcCst|p t|Sr)ris_floating_point is_complex)rrrr_is_complex_or_float%srparamscCsbtr|sdStjt|g|djtjd}t||t|tkr/t dt|ddS)Nrdevicedtypez&Mismatch in number of params: ours is z*, at least one worker has a different one.) rrrlenrlongritemr RuntimeError)rrrrr_check_number_of_params)s rtensorssrccCsbtsdSdd|D}t|g}|D]}tjj|j|dd}||q|D]}|q(dS)zBroadcast the tensors from the given parameters to all workers. This can be used to ensure that all workers have the same model to start with. NcSsg|]}t|r|qSr)r).0rrrr >sz%broadcast_tensors..Tr!async_op)rrrr broadcastdataappendwait)r r!handlesrhandlerrrbroadcast_tensors8s  r,TcCstsdSg}|D])}t|jr2|r!tjj|jtjjjdd}n tjj|jddd}| ||fq |D]\}}| |rF|jt _q5dS)zU Sync grad for buffers. If average is False, broadcast instead of averaging. NTrr%rr$) rrrr'rrReduceOpSUMr&r(r)r)buffersaverager*bufferr+rrr sync_bufferIs(  r3cCsxtsdSg}|D]}|jdur%tjj|jjtjjjdd}|||fq |D]\}}| |jjt _q(dS)z Simpler alternative to DistributedDataParallel, that doesn't rely on any black magic. For simple models it can also be as fast. Just call this on your model parameters after the call to backward! NTr-) rgradrrrr'r.r/r(r)r)rr*pr+rrr sync_grad_s  r6?metricscCsts|St|\}}tjrdnd}tjt|dg|tjd}||9}t ||dd|d }t t||S)znAverage a dictionary of metrics across all workers, using the optional `count` as unormalized weight. cudacpur rN) rzipitemsrr9 is_availablerlistfloat32rr:tolistdict)r8countkeysvaluesrraveragedrrraverage_metricsrsrG)r)T)r7)__doc__typingtprr rrrr.r/TensorrrListrIterableintr,r3r6DictstrfloatrGrrrrs