AVhjzdZddlZddlZddlmZmZmZmZddlm Z ddlm Z ddl m Z ddl mZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZddlmZddlmZdZ dZ!dZ"dZ#Gdde$Z%Gdde$Z&ejNfdZ(dZ)dZ*dZ+d Z,d!Z-d"Z.d#Z/y)$zUtilities for cross_device_ops.N)CallableListOptionalUnion)collective_util)values) backprop_util)context)dtypes)indexed_slices)ops) tensor_spec) array_ops)collective_ops)cond)math_ops)nccl_ops)resource_variable_ops) tf_logging)coredc g}t|D]`}|Dcgc]\}}| }}}tj|}|jt||Dcgc] \}\}}||fc}}}bt t|}|Scc}}wcc}}}w)z)Aggregate gradients using nccl allreduce.)ziprall_sumappendlist) replica_gradsagg_all_g_and_vsingle_g_and_vg_ single_grads agg_gradsvs _/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/cross_device_utils.pyaggregate_gradients_using_ncclr&)s/]+Bn"01$!QA1L1  .I!$Y!?@@IAv1!Q@BB o.// 2 As BBc g}t|}|dz}tt|D]\}}||z}||z|z}||krd} |} n|} d} || | |z} tj||5t | dd\} } ddd|| | |z}tj||5t |dd\}} dddtj||5t  gdd\\}} } dddtj||5t j}dddtj||5t j}dddg}tt|D]W}tj||5||k||kk(r}n}|jt j|dddY|jt||D cgc] \}\} }||fc}} }tt|}|S#1swYxYw#1swYtxYw#1swYKxYw#1swY"xYw#1swYxYw#1swYxYwcc}} }w)aAggregate gradients using hierarchical copies. Args: avail_devices: available GPU devices. replica_grads: List of lists of (gradient, variable) tuples. The outer list is over replicas. The inner list is over individual gradients. Returns: The list of (aggregated_gradient, variable), where the gradient has been summed across all replicas and the variable is chosen from the first replica. rFN) len enumeraterr device$aggregate_single_gradient_using_copyridentityrangerr) avail_devicesrr# num_devices group_sizeir"group_0_main_devicegroup_1_main_device group_0_begin group_1_begingroup_0_device_gradsgroup_0_agg_gradsr!group_1_device_gradsgroup_1_agg_gradsagg_total_gradsgroup_0_agg_grads_bcastgroup_1_agg_grads_bcastagg_grads_bcastjsrc_device_gradr r$s r%+aggregate_gradients_using_hierarchical_copyrA7s0)M"+a*"3 #67.Foak/.;{JZ'm m mm( (5 (BD M"56 7.A u.. ( (5 (BD M"56 7.A u.. M"56 7@ D / 0%!@A@ M"56 7D ) 2 2? CD M"56 7D ) 2 2? CDO 3|$ %D ::mA& 'D * ,!j. A3/3/y11/BC DDD!$_l!CDDIAv1!QDF[.F`3 ?#) K.... @@ DDDD DD EsN&HH"H/H<<I  5I%I!H "H, /H9 <I I I cf|Dcgc]\}}| }}}tj|}|r0t|dkDr"tj|dt|z }|dd}|rAtj tj tj|}||f|fS||fdfScc}}w)a&Calculate the average gradient for a shared variable across all replicas. Note that this function provides a synchronization point across all replicas. Args: grad_and_vars: A list or tuple of (gradient, variable) tuples. Each (gradient, variable) pair within the outer list represents the gradient of the variable calculated for a single replica, and the number of pairs equals the number of replicas. use_mean: if True, mean is taken, else sum of gradients is taken. check_inf_nan: check grads for nans and infs. Returns: The tuple ([(average_gradient, variable),], has_nan_or_inf) where the gradient has been averaged across all replicas. The variable is chosen from the first replica. The has_nan_or_inf indicates the grads has nan or inf. g?rN)radd_nr)rmultiply logical_not reduce_all is_finite) grad_and_varsuse_mean check_inf_nanr r!gradsgradr$has_nan_or_infs r%r,r,s(' 'A1 '% '  $ #e*q.   dC#e*$4 5DAq!**Y0078:N !9n $$ !9d? (s B-c0eZdZdZddZdZdZdZdZy) CollectiveKeysaClass that manages collective keys. We need to manage three different keys for collective: *Group key*: an integer key to identify the set of cooperative devices. Collective ops work under the same set of devices must using the same group key. *Instance key*: an integer key to identify the set of same counterpart of tensors on different devices in a device group that need to be all-reduced. This class is thread safe. c`||_i|_tj|_i|_y)zaInitializes the object. Args: group_key_start: the starting integer of group key. N) _group_key_instance_key_table threadingLock_lock _known_groups)selfgroup_key_starts r%__init__zCollectiveKeys.__init__s) &DO!D!DJDc|j5dj|}||jvr|j||j|<|j|cdddS#1swYyxYw)aReturns a group key for the list of local devices. The same group key is returned if the list of local devices is the same. Args: devices: a list of local canonical device strings in a collective group. Returns: a group key. ,N)rVjoinrW_get_new_group_key)rXdevices devices_keys r% get_group_keyzCollectiveKeys.get_group_keyse -HHW%k D.. .*.*A*A'*J;'    , ---s A A##A,c|j}|xjdz c_i|j|<|D]}t|j||<|S)aJReturns a new group key. The caller should store and reuse the same group key for the same set of devices. Calling this method always returns a new group key. This method is not thread-safe. Args: devices: a list of canonical device strings in a collective group. Returns: a new group key. rC)rRrSINSTANCE_KEY_START_NUMBER)rXr`new_keyr+s r%r_z!CollectiveKeys._get_new_group_keysVooGOOqO(*DW%L2Kdw'/L Nr[c|j5|jj|d}|td|d||vrtd|d|||}||xxdz cc<|cdddS#1swYyxYw)aReturns a new instance key for use in defining a collective op. You should call this once per each collective op of a collective instance. Args: group_key: the group key returned by get_group_key(). You should not assign the group key yourself. device: a canonical device string. It should be the device this collective op is on. Returns: a new instance key. Raises: ValueError: when the group key is invalid or the device is not in the group. NzGroup z is not found.zDevice z is not present in group rC)rVrSget ValueError)rX group_keyr+groupr$s r%get_instance_keyzCollectiveKeys.get_instance_keys$ &&**9d;e 6)N;<< u 76(*CI;OPP -a Fmqm s AA--A6ct}|j|_tj|j||_|SN)rPrRcopydeepcopyrS)rXmemocopieds r% __deepcopy__zCollectiveKeys.__deepcopy__ s6 FF!%t/G/G!NF Mr[N)rC) __name__ __module__ __qualname____doc__rZrbr_rkrrr[r%rPrPs   -"*8r[rPc eZdZdZdZdZdedededede jf dZ d e e jej ffd Zd Zd Zd ZdZdZ dde jd ee e jej fdee jde j0fdZde jdee jde j0fdZ ddeee jdee jde j0fdZ dde jde jdee jde j0fdZ ddej>dee jdej>fdZ y)CollectiveReplicaLauncherz"Launch collectives on one replica.Trir1collective_keysr+optionscT||_||_||_||_||_|j rVt j5t j|5tjd|_ ddddddyd|_ y#1swYxYw#1swYyxYw)Ng) rR _group_size_collective_keys_device_options_use_ordering_tokenr init_scoper+rResourceVariable_ordering_token)rXrir1rzr+r{s r%rZz"CollectiveReplicaLauncher.__init__s DO!D+DDLDM ! >> JSZZ/J4EEbIJJJ"dJJJJs$BB9BB BB' control_inputcz|&|jstj|gStjSrm)rr control_dependenciesNullContextmanager)rXrs r%_control_inputz(CollectiveReplicaLauncher._control_input&s5 )A)A)C  % %}o 66  ! ! ##r[cLtjsytjSNF)r #executing_eagerly_outside_functionsry_prefer_unique_instance_keyrXs r%_use_unique_instance_keyz2CollectiveReplicaLauncher._use_unique_instance_key,s  2 2 4  $ @ @@r[cLtjsytjSr)r rry_prefer_ordering_tokenrs r%rz-CollectiveReplicaLauncher._use_ordering_token1s  2 2 4  $ ; ;;r[c|jr'tj}t|ddr|j}t|ddrt j sd|jrX|j5|j|jtjgtjcdddS|jj!|j"|j$}tj&d5tj(|tjcdddS|jj!|j"|j$S#1swYyxYw#1swYyxYw)zReturns the next instance key.is_control_flow_graphFNzCPU:0)dtype)rr get_default_graphgetattr outer_graphr executing_eagerlybuilding_function as_defaultcapture_call_time_value_next_instance_keyr TensorSpecr int32r~rkrRrr+convert_to_tensor)rXgraph instance_keys r%rz,CollectiveReplicaLauncher._next_instance_key8sA $$& ##%e E2E :!! E2E :  & & (U-D-D     Q ..%%{'='=b&,,'OQ  Q Q,,== OOT\\+ ZZ  I&&|6<<H I I " " 3 3DOO48LLBB Q Q I Is=>E) %E5)E25E>cP|jr|jjSyrm)rrhandlers r%_get_ordering_tokenz-CollectiveReplicaLauncher._get_ordering_tokenWs% !  ! ! ( (("r[c"|jS)z0Whether this launcher can order NCCL operations.)rrs r%can_order_ncclz(CollectiveReplicaLauncher.can_order_nccl[s  # # %%r[N input_tensorreturnc |j}|jj|}|j}t j |j 5|j|5tj||j|j||jj|j|cdddcdddS#1swYnxYwdddy#1swYyxYw)aAll-reduce a dense tensor. Args: input_tensor: a dense tensor. It must have the same shape on all replicas. control_input: if not None, add control edges between control_input and the all-reduce. options: an optional tf.distribute.experimental.CommunicationOptions. If provided, it overrides the default options. Returns: The reduced tensor. communication_hinttimeoutordering_tokenN)rrmergerr r+rrr all_reduce_v2r}rRimplementationvaluetimeout_seconds)rXrrr{rrs r% all_reducez$CollectiveReplicaLauncher.all_reduce_s"**,Lmm!!'*G--/N DLL ! )  ] + )  ) )     // $3399))') ) ) ) ) ) ) )s%C#-AC; C#C C##C,c ~|j}|jj|}|j}t j |j 5tj||j|j||jj|j|cdddS#1swYyxYw)a&All-gather a dense tensor. Args: input_tensor: a dense tensor. It must have the same shape on all replicas. options: an optional tf.distribute.experimental.CommunicationOptions. If provided, it overrides the default options. Returns: The reduced tensor. rN)rrrrr r+rr all_gather_v2r}rRrrr)rXrr{rrs r% _all_gatherz%CollectiveReplicaLauncher._all_gather~s**,Lmm!!'*G--/N DLL !)  ) )     // $3399))'))))s AB33B<input_tensor_packsc ||jj|}g}|D]|}tjr*|D]$}|j |j |d|&Bt j|j5|Dcgc]}tj|dg}}|Dcgc]}tj|}}|jtjjk(r|r|d} nd} |j tj |d| |} |D cgc]} t#j$| } } tj&| | d} t)|| D]*\}}|j tj||, ddd|Scc}wcc}wcc} w#1swYxYw)aBatch all-reduce dense tensors. This takes a list of batches of tensors. Using multiple batches have the benefit that it doesn't need to wait for all inputs to be ready to start the all-reduce. Args: input_tensor_packs: a list of lists of dense tensors. options: an optional tf.distribute.experimental.CommunicationOptions. If provided, it overrides the default options. Returns: A flat list of reduced tensors. Nraxis)rrr rrrr r+rrreshapeshaperrCommunicationImplementationNCCLconcatr reduce_prodsplitr)rXrr{outputspackrt flat_tensorsshapesrreduceds num_elements flat_outputsr flat_outputs r%batch_all_reducez*CollectiveReplicaLauncher.batch_all_reduces$mm!!'*GG"B  " " $! GL ..tWE F G ZZ % B>BC)++At4C,C0451IOOA&5&5$$ <<AABFM#BKM MOO|!4mWN';ABa(..q1B,B",QG,$' $= B e[ NN9,,[%@ A B B BB0 ND5C B Bs8F1F"$F1*F'AF1%F,AF1"F11F; rc tjr tdtj|j 5tj tj|g5tj|gtj|tj|dztj|fd}tj||}|jtjtj |d|}|dddf}tj"|}t%||} |j| |} g} t|j&D]!} | |z} | j)| | | || z#tj| d}tjtjd|dzdgtj|dztj|fd}tj||cdddcdddS#1swYnxYwdddy#1swYyxYw)aJAll-gather a dense tensor. This method must be called inside a tf.function. Args: input_tensor: a dense tensor. It must have the same rank on all replicas, and dimensions other than `axis` need to be the same as well. axis: 0-D int32 Tensor. Dimension along which to gather. Must be in the range [0, rank(value)). options: an optional tf.distribute.experimental.CommunicationOptions. If provided, it overrides the default options. Returns: The gathered Tensor. Raises: RuntimeError: if called in eager mode. z*all_gather is not supported in eager mode.rCrr)permN)r r RuntimeErrorr r+rrrr-rrr.rank transposerexpand_dims_v2shape_v2 reduce_max _pad_utilr}r)rXrrr{perm_preinput_tensor_tgathered_shape first_dims full_axis_dimpadded_input_tensorgather_padded_out_tensor split_tensorsr2 start_pos out_tensor_t perm_afters r% all_gatherz$CollectiveReplicaLauncher.all_gathers.  " E FF DLL !#@ ! !9#5#5l#C"D E#@!! 68>>$' >>$(INN<$@ A Ch!**>!TAX & >>$(INN>$B C Ej J ?G#@#@#@#@#@#@#@s$*H:)F)H% H:%H. *H::I input_slicesc jjtjj5dt t jttjgt jfdtjffd tjj }j#|dt jdttjdt jffd t%j$t'j(t'j*t'j,fdfdcd d d S#1swYy xYw) a9All-reduce an IndexedSlices. This method can be called outside tf.function. Args: input_slices: an IndexedSlices. options: an optional tf.distribute.experimental.CommunicationOptions. If provided, it overrides the default options. Returns: The reduced IndexedSlices. all_gather_fnrcB|j}jtjjk(r|g}ng}t j |5|j}dddtj|jS#1swY+xYw)z/Use all_gather_fn to aggregate `IndexedSlices`.N)rindices dense_shape) rrrrrr rrr IndexedSlicesr)r all_valuescontrol all_indicesrr{s r%all_gather_indexed_sliceszVCollectiveReplicaLauncher.all_reduce_indexed_slices..all_gather_indexed_slices s #<#6#6@  " "  7 7 < < =L''  % %g . E%l&:&:GD+ E++$002 2 E Es BBrr{c tj}t||} j||}g}t j D]!}||z}|j ||||z#tj|dS)z4all_gather tensors of different sizes using padding.r) rrrrr.r}rrr) rr{ max_length padded_tensorall_padded_tensorsrr2r all_lengthsrXs r%all_gather_with_paddingzTCollectiveReplicaLauncher.all_reduce_indexed_slices..all_gather_with_padding6s((5 !, ; !--mWE t''( CA*n)   1)I2=a.=A B C C q11r[c(jSrm)r)rrXsr%zECollectiveReplicaLauncher.all_reduce_indexed_slices..Hs+D,<,<=r[cSrmrw)rrsr%rzECollectiveReplicaLauncher.all_reduce_indexed_slices..Is+,CDr[N)rrr r+rrr TensorLikerrOptionsTensorr rrrrrrrequalr reduce_min)rXrr{lengthrrrs``` @@@r%all_reduce_indexed_slicesz3CollectiveReplicaLauncher.all_reduce_indexed_slicess2mm!!'*G DLL !+F2!)@)@ ABDKKOQ2  ' '2&|334f$$VW5k 2 2O334 29= 2YY ..!!+.!!+. 0 > D FM+F+F+Fs DE((E1)NNrm)!rsrtrurvrrintrPstrrrrZrrrr Operationrrrrrrrrrrrrrr rrrwr[r%ryrys* $ " " " . "8; "'// "$%03 1>+?$ A <B>)&HL37 )OO)eDOOS]]$BCD)//0 )=AKK )>)doo)#O$;$;<)AE)848,tDOO45,//0,=AKK,d48 =@OO=@ OO=@//0 =@=AKK =@D48EF"00EF//0EF## EFr[ryc`td|Drtj|S||S)zGAggregate tensors using `accumulation_fn` and IndexedSlices via concat.c3PK|]}t|tj ywrm isinstancer r.0r$s r% z6aggregate_tensors_or_indexed_slices..NsEA~33 4E$&)anyr AggregateIndexedSlicesGradients)raccumulation_fns r%#aggregate_tensors_or_indexed_slicesr Ls,EfEE  8 8 @@ 6 ""r[ct|tjrMtj|}tj|j |z |j |jS||z Srm)rr rr FlattenNestedIndexedSlicesrrr)rns r%%divide_by_n_tensors_or_indexed_slicesrTs[~334  4 4U ;E  ' ' q(8%--(-(9(9 ;; 19r[ctj|5t|tjrt j |j}t j |j}|j t j |j}nd}tj|||}nt j |}ddd|S#1swYSxYw)z-Copies a tensor or IndexedSlices to a device.N) r r+rr rrr-rrr)rr+ copied_valuescopied_indices copied_shaperesults r%'copy_tensor_or_indexed_slices_to_devicer]s zz& )%556((6m ))%--8n    & ))%*;*;<  ++M>,8:f!!%(f ) - ) -s B4CCct|tjryt|tjrt d|j DSy)NTc3PK|]}t|tj ywrmrrs r%rz$is_indexed_slices..rs$K89 1n223KrF)rr r value_libDistributedValuesallr)rs r%is_indexed_slicesrnsJ~334 y223 K=B\\K KK r[cg}g}g}g}t|D]U\}}t|r#|j||j|4|j||j|W||||fS)a Split values into dense and sparse values. Args: values: a list of tensors or `PerReplica`s. Returns: Four lists: a list of dense values, a list of their indices in `values` and a list of sparse values, a list of their indices in `values`. )r*rr)r dense_values dense_indices sparse_valuessparse_indicesr2r$s r%split_by_sparsityr"ws,--.da1A!1  }m^ CCr[cd}|D]}|t|dz }dg|z}|D]'}|s|ds t|D]\}}||J|||<)|S)zStitch values together according to their indices. Args: values_and_indices_list: a list of tuples of values and indices indicating the values and positions in the returned list. Returns: a stitched list of values. rN)r)r)values_and_indices_listrvalues_and_indicesrr$r2s r% stitch_valuesr&s &3) c$Q'((F) 6F?&303)*$!Qay   q  -r[c2|dk(r|gSg}d}|D]}|jj}|tjd||gcS||jj z}|r||kDr|j gd}|dj |||z }|S)aGroups `input_tensors` into chunks of `bytes_per_pack`. The method preserves the original order of `input_tensors`. The grouping is best effort, each pack could have more or less bytes than `bytes_per_pack`. It only groups values with known shape. Args: input_tensors: a list of Tensor. bytes_per_pack: an integer. Returns: A list of packs of Tensor. All values are grouped into one pack if `bytes_per_pack` is zero or any of the value has unknown shape. rzAnot packing values due to the unknown or inconsistent shape of %sr)rrloggingwarningrsizer) input_tensorsbytes_per_packpackslast_pack_sizerrr*s r% group_by_sizer/s q ? %.e;;++-L oo M _ %++** *D N^3 ll2n "IUdN!" ,r[c|tj|dz }tj|}d|gg}tj|tj|dz dft j gd}tj||}|S)z?Pad the `input_tensor`'s first dimension to be `full_axis_dim`.rrCr()rrr)rrrrzerosr rpad)rrmissing_axis_dim tensor_rank paddings_axispaddingsrs r%rrs"Y%7%7 %Ea%HH|,+'()-   oo[1_a0 E$% &( " lH= r[)0rvrnrTtypingrrrrtensorflow.python.distributerrrtensorflow.python.eagerr r tensorflow.python.frameworkr r r rtensorflow.python.opsrrrrrrtensorflow.python.platformrr(tensorflow.python.typesrrdr&rAr,objectrPryrDr rrrr"r&r/rrwr[r%r?s& 228<1+.6+3+0&**7<( Od HbVbJwFwFt AI#"D2.%P r[