*i.UdZddlmZddlZddlZddlmZddlmZddl m Z m Z m Z m Z mZe rddlmZmZddlZddlmZgdZd ed <ej0eZed ZeGd d e eZddddd ddZ d ddZdf ddZddZ ddZ! ddZ" ddZ#d dZ$d!dZ%y)"zl A set of primitive functions for performing collective ops. Each should also handle single rank scenario. ) annotationsN) defaultdict) dataclass)AnycastGeneric TYPE_CHECKINGTypeVar)CallableIterable) SyncPayload broadcast all_gatherall_gather_object_enforce_typez list[str]__all__Tc:eZdZUded<ded<ded<dZded <y) r str | None stage_nameboolsuccessrpayloadNzException | None exception)__name__ __module__ __qualname____annotations__rd/var/www/pod-logistic/pod-ai/venv/lib/python3.12/site-packages/torch/distributed/collective_utils.pyr r &s M J"&I&rr T)rrrankpgcT|s | tdd}d}||dk(s|)|j|k(rt|r |}n|}t ||||}|E|g} t j | ||t| dk7rtdt| | d}|jsNd |d } || d |jz } |j| d |jz } t| |jtt|jS#t$r}d}|}Yd}~d}~wwxYw) aK Broadcasts the data payload from rank 0 to all other ranks. Or if a function is passed, execute it in rank 0 and broadcast result to all other ranks. Can be used to broadcast a failure signal to stop all ranks. If the function raises an exception, all ranks will raise. Args: data_or_fn: the data to broadcast or function to execute and broadcast result. success: False to stop all ranks. stage_name: the name of the logical stage for synchronization and debugging rank: rank to broadcast data or execute function and broadcast results. pg: the process group for sync Throws: RuntimeError from original exception trace Returns: the value after synchronization Example usage: >> id = broadcast(data_or_fn=allocate_id, rank=0, pg=ext_pg.my_pg) Nz9Data or Function is expected to be None if not successfulrFrrrr)srcgroupz7Expected broadcast_list to have exactly 1 element, got zRank z failedz: stage z : exception )AssertionErrorr!callable Exceptionr distbroadcast_object_listlenrrr RuntimeErrorrrr) data_or_fnrrr!r"rresync_objbroadcast_list error_msgs r rr.sk> z- G  G"&I tqybnd9J J  $, !G H ~" "">t2F ~ ! # I#nJ]I^_ "!$   D6)  ! 8H$7$7#89 9I    ) <(:(:';< > all_ids = all_gather(data_or_fn=allocate_id, pg=ext_pg.my_pg) NTFr$rz)Unexpected stage name received from rank z:  z!all_gather failed with exception )r)r*r r+get_world_sizerrrr enumeratelistrrappendrr-r.)r/rr"rrrr0r1 total_listexception_listret_listr3isps r rr~s*G"&IG   lG  H ~Vd11"55 &r:x@+a.*Q-8CC 68 tDQ$8*EF (EAr}} *?s"R]]OSTU ::",,":%%q",,&78 OOBJJ ' ( ~  ""!$ %3H4F4F3GH%% &  !![ GI sF F FFc.t|t|uS)N)type)xys r rDsDGtAws r rrs, ;26;H1}AI 1h I{1~6's$tKN/C.DE..29o->@ rcVt|}t|dkr tdtt |t|k7r tdd}g}|r|j d}||}nt |tr/||dzk(rt||dzd}n||z }t|||z|}nvt |ts td||jk(r9t|j|j|jz|j}n|j||}|rt |tr |jt||dzdn!t |tr|j|g}|D]}t|dk(r|j|j.|jdk(r+|j|jd|jh|j|jd|jd|jdj|S)Nrzranks should all be positivez#ranks should not contain duplicatesr'z!curr must be an instance of range:,)sortedminr(r-setpop isinstanceintrGstopstartstepr:join)rankscurrrangesrBrYresultrs r _summarize_ranksr`s 5ME 5zA~;<< 3u:#e*$BCC#D F  IIaL <D c "D1H}T1q5!,4xT1t8T2dE*$%HIIDII~TZZTYY)> J d## &$ eD$(A./ D%  d F  : q6Q; MMQWWI ' VVq[ MMQWWIQqvvh/ 0 MMQWWIQqvvhax8 9 : 88F rc@|j}t|jDcgc]}tj|}}tj j |||Dcgc]b}|ddjtjj|ddjtjjfd}}tt}t|D]\}\} } || | fj| |dfScc}wcc}w)Nz(Seed, Offset)) get_staterGsizetorch empty_like distributedrviewuint64itemrrSr8add) generatorr& local_state_ all_statesstate seeds_offsetsseed_offset_ranksr!seedoffsets r _check_philox_rng_syncru s%%'K9>uzz|9LMA%"";/MJM   [9   r % * * ,eABinnU\\.J.O.O.QRM$C( )- 84ntV4.)--d34 . ..Ns D.A'Dc|j}t|jDcgc]}tj|}}tj j ||tt}t|D]:\}}|tj|jj|<|dfScc}w)NzGenerator state hash) rcrGrdrerfrgrrrSr8 hash_tensorrjrk)rlr& state_tensorrnall_state_tensors state_ranksr!s r _check_cpu_rng_syncr{0s&&(LAFuzz|ATUA)),7UU   !2LAc"K'(9:Fl E%%l388:;??E F . ..VsCc|jjdk(r t||S|jjdk(r t||St d|jj)NcudacpuzUnsupported generator device: )devicerArur{NotImplementedError)rlr&s r _check_rng_sync_internalr@sj&%i77     % '"9e44!,Y-=-=-B-B,C D  rc`d|dg}|jDcgc]\}}t|t|g}}}tjj drddlm}|||Sdj|Dcgc] }t|c}}t|d|Scc}}wcc}w)NRanksz valuestabulater)r)headers )itemsr`str importlibutil find_specrrZ) tag value_ranksrvaluer[ rank_valuesrrowrow_strs r _desync_table_strrMs3%w(GBMBSBSBU2>% % #e*-K~~ +% W55ii[9cS9:G '"WI& '':s B%<B+ct||\}}d}t|dkDr$dt||}tj ||S)Nr'zGenerator desync detected: )rr-rloggererror)rlr&r value_headerlog_strs r _check_rng_syncrZsL 8E JKG ;!01Brs ##!==2    8 $ CL ''!*' '!#' M%#M%M% M%  M% ! M%M%d"#'I"#I"I" !I" I"l0O""" "-" "J+\ / /'8 / / / /'8 / /    '8     (r