AVh,dZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z ddl mZddl mZddl mZddl mZdd l mZdd l mZdd l mZdd l mZdd lmZddlmZddlmZddlmZddlmZddlmZddlm Z ddl!m"Z"ddl!m#Z#ddl!m$Z$ddl%m&Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-dZ.dZ/ d@dZ0dZ1dZ2d Z3d@d!Z4d@d"Z5d@d#Z6 dAd$Z7d%Z8d&Z9e+d'Gd(d)e:Z;e+d*Gd+d,e;ZGd/d0e:Z?dCd1Z@dDd2ZAGd3d4e;ZBeBZCejd5d6ZEe+d7Gd8d9eBZFe+d:Gd;de;ZJdDd?ZKy)Ez?Classes for different algorithms of reduction and broadcasting.N) device_lib)collective_util)cross_device_utils) device_util)distribute_utils) ps_values) reduce_util) tpu_values)values) values_util)context) def_function)indexed_slices)kernels)ops)tensor) tensor_util) array_ops)math_ops)resource_variable_ops) tf_logging)nest) tf_export) doc_controlsct|tjtjfrt |j St |S)zChecks whether `destinations` is not empty. Args: destinations: a `DistributedValues`, variable, or string object. Returns: Boolean which is True if `destinations` is not empty. ) isinstancerBaseResourceVariable tensor_libTensorbooldevice destinationss ]/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/cross_device_ops.pycheck_destinationsr%4sC &;;""$%  ## $$ l c <t|tjtjt j tjtjtjfs tj|s tdt!|s tdy)z5Validates the `destination` is one of expected types.zcdestinations must be one of a `DistributedValues` object, a tf.Variable object, or a device string.zdestinations can not be emptyN)r value_libDistributedValuesrrr IndexedSlicesrAggregatingVariablesix string_typesr TPUMirroredVariableris_resource_variable ValueErrorr%r"s r$validate_destinationsr1Es ""J$5$5##Y%B%B 77 ';;LI B CC L ) 4 55 *r&ct|tjr tdt j |sCt j|dk(r+t j|j|jS|tjjk(r|S|dk7rtd|d|dt|t|||S) z8Reduce a non-DistributedValue `value` to `destinations`.z^You are passing a `DistributedValues` to `reduce_non_distributed_value`, which is not allowed.r)dtypezA non-DistributedValues value z, cannot be reduced with the given reduce op .)canonicalize_devices)rr(r)r0r is_tf_typenpallzerosshaper3r ReduceOpMEANr1simple_broadcast) reduce_opvaluer#num_replicas_in_graphr6s r$reduce_non_distributed_valuerBTs y223 M NN    &266%1*+= 88EKKu{{ 33+&&+++ L! 27D EE,'  |2F HHr&ct|tjr|Stj|st j |}t|drtj|fStd)z2Converts a single tensor into a PerReplica object.r!zZCannot convert `input_tensor` to a `PerReplica` object because it doesn't have device set.) rr(r)r is_tensorrconvert_to_tensorhasattr PerReplicar0) input_tensors r$_make_tensor_into_per_replicarIusg i99:    | ,((6L \8$    009 ::r&c g}t|}t|ttfs td|D]Z}t|ts tdt |dk7r tdt |d}|j ||df\|S)z@Converts each tensor into a PerReplica object in the input list.z3`value_destination_pairs` should be a list or tuplezz4_validate_value_destination_pairs..s IZe $ Isc3VK|]!}t|dtj#ywrN)rr(rG)rXvs r$rYz4_validate_value_destination_pairs..s) /!i22 3 /')T)rrLrMr9)rPs r$!_validate_value_destination_pairsr^sG ! +dE] ;E I1H I I   /- / /  r&ct|tjr |jS|rPt|tj rt j|fSt j|jfSt|tj rt j|fSt j|jfSrW) rr(r)_devicesr,r-rresolver!!canonicalize_without_job_and_task)r#r6s r$get_devices_fromrcs i99:   , 0 01!!,/ 11    3 3 4 66 c../  9 9, G II  7 7 8K8K L NNr&cd||uxs+tt||tt||k(SrW)setrc)leftrightr6s r$_devices_matchrhs@  9#. "#&) 5"6 7'999r&cdtfdDsytfdddDsyy)Nc3>K|]\}}t||ywrWrh)rXr\dr6s r$rYz%_all_devices_match..s' + !QQ/0 +sFc3JK|]\}}t|ddywr[rk)rXr\_r6rPs r$rYz%_all_devices_match..s2 / !QQ/2157KL /s #r4T)r9)rPr6s``r$_all_devices_matchrosA  +) + +   /)!"- / /  r&ct||}t|dk(r|stj||dSg}|D]'}|j tj||)t j |tjS)z8Broadcast `value` to `destinations` using simple copies.r4r wrap_class) rcrNr'copy_tensor_or_indexed_slices_to_devicerOrregroupr(Mirrored)r@r#always_mirroredr6devices value_updatesrls r$r>r>s \+? @'\Q  E E wqz M P  D DUA NPP  # #M/8/A/A CCr&c|j}|s tdt|}tj|5t j t j5tj||}|tjjk(rtj||}n(|tjjk7r tdddddddS#1swYxYw#1swYSxYw)z3Reduces the value by accumulation_fn and reduce_op.%`per_replica_value` must be non-emptyz.`reduce_op` must be Reduce.SUM or Reduce.MEAN.N)r r0rNrr!r device_policyDEVICE_PLACEMENT_SILENTr#aggregate_tensors_or_indexed_slicesr r<r=%divide_by_n_tensors_or_indexed_slicesSUM)per_replica_valuereduce_to_deviceaccumulation_fnr? all_valuescountreduceds r$_simple_reducers!''*  < == j/% zz"#K   w>> ?K"FF o'g k**// /$JJ U  ,,00 0IJJKK .KKK .s$$C/A3C#C/#C, (C//C9c*|j}|s tdtj|5t j tj 5tj||}ddddddS#1swYxYw#1swYSxYw)zAConcatenate all values in the DistributedValues input and return.rzN) r r0rr!r r{r|rconcat)rraxisrgathereds r$_simple_gatherrs ''*  < == zz"#4   w>> ?4!!*d3h44 /444 /s#$BA<*B<B BBzdistribute.CrossDeviceOpsceZdZdZdZedZddZddZdZ ddZ d Z e jd Ze jd Ze jd Zd Zy)CrossDeviceOpsaBase class for cross-device reduction and broadcasting algorithms. The main purpose of this class is to be passed to `tf.distribute.MirroredStrategy` in order to choose among different cross device communication implementations. Prefer using the methods of `tf.distribute.Strategy` instead of the ones of this class. Implementations: * `tf.distribute.ReductionToOneDevice` * `tf.distribute.NcclAllReduce` * `tf.distribute.HierarchicalCopyAllReduce` cd|_yNT)_canonicalize_devicesselfs r$__init__zCrossDeviceOps.__init__ s !%Dr&cy)Nr4rs r$_num_between_graph_workersz)CrossDeviceOps._num_between_graph_workerss r&Nc<|tj}t|}t||jdk(rt |j dk(rt|||jr}tj|j dj5tj|j d}dddtjftj S|tj}|j#||||S#1swYYxYw)aReduce `per_replica_value` to `destinations`. See `tf.distribute.StrategyExtended.reduce_to`. This can only be called in the cross-replica context. Args: reduce_op: a `tf.distribute.ReduceOp` specifying how values should be combined. per_replica_value: a `tf.distribute.DistributedValues`, or a `tf.Tensor` like object. destinations: a `tf.distribute.DistributedValues`, a `tf.Variable`, a `tf.Tensor` alike object, or a device string. It specifies the devices to reduce to. To perform an all-reduce, pass the same to `value` and `destinations`. Note that if it's a `tf.Variable`, the value is reduced to the devices of that variable, and this method doesn't update the variable. options: a `tf.distribute.experimental.CommunicationOptions`. See `tf.distribute.experimental.CommunicationOptions` for details. Returns: A `tf.Tensor` or `tf.distribute.DistributedValues`. Raises: ValueError: if per_replica_value can't be converted to a `tf.distribute.DistributedValues` or if destinations is not a string, `tf.Variable` or `tf.distribute.DistributedValues`. Nr4rrq)rOptionsrIr1rrNr rhrrr!ridentityrrtr(rureduce_implementation)rr?rr#optionsr\s r$reducezCrossDeviceOps.reduces8'')g56GH,' &&!+  1"%&1'+9 |T-G-G,I ::'..q188 9<   077: ;<  % %qdy7I7I JJ'')g  % %i1B&2G == <Gather `per_replica_value` to `destinations`. Args: per_replica_value: a `tf.distribute.DistributedValues`, or a `tf.Tensor` like object. destinations: a `tf.distribute.DistributedValues`, a `tf.Variable`, a `tf.Tensor` alike object, or a device string. It specifies the devices to gather to. To perform an all-gather, pass the same to `value` and `destinations`. Note that if it's a `tf.Variable`, the value is gathered to the devices of that variable, and this method doesn't update the variable. axis: specifies the dimension to gather along within each replica's tensor. options: a `tf.distribute.experimental.CommunicationOptions`. See `tf.distribute.experimental.CommunicationOptions` for details. Returns: A `tf.Tensor` or `tf.distribute.DistributedValues` Raises: ValueError: if per_replica_value can't be converted to a `tf.distribute.DistributedValues` or if destinations is not a string, `tf.Variable` or `tf.distribute.DistributedValues`. z0gather/all_gather does not support IndexedSlicesNr4rrq)rrr*NotImplementedErrorrrrIr1rrNr rhrrr!rrrrtr(ru_gather_implementation)rrr#rrr\s r$_gatherzCrossDeviceOps._gatherCs2#^%A%AB !0 11'')g56GH,' &&!+  1"%&1'+9 |T-G-G,I ::'..q188 9<   077: ;<  % %qdy7I7I JJ  & &'8,'. 00 <>r&c|tj}t|s t|}|D]\}}t ||j dk(rvt ||jr`t|ddjdk(rB|Dcgc]4\}}tj|jtj6c}}S|tj}|j|||Scc}}w)aBReduce values to destinations in batches. See `tf.distribute.StrategyExtended.batch_reduce_to`. This can only be called in the cross-replica context. Args: reduce_op: a `tf.distribute.ReduceOp` specifying how values should be combined. value_destination_pairs: a sequence of (value, destinations) pairs. See `tf.distribute.CrossDeviceOps.reduce` for descriptions. options: a `tf.distribute.experimental.CommunicationOptions`. See `tf.distribute.experimental.CommunicationOptions` for details. Returns: A list of `tf.Tensor` or `tf.distribute.DistributedValues`, one per pair in `value_destination_pairs`. Raises: ValueError: if `value_destination_pairs` is not an iterable of tuples of `tf.distribute.DistributedValues` and destinations. r4rrq)rrr^rTr1rrorrNr rrtr(rubatch_reduce_implementation)rr?rPrrnrlr\s r$ batch_reducezCrossDeviceOps.batch_reduces,'')g --D E!C !!#(1A &&!+0B!;!;1=AD #A &q ) 0 0B256B7.a  " "188 8J8J K '')g  + +I7N,3 55s 9C0c<t||j||S)aCBroadcast `tensor` to `destinations`. This can only be called in the cross-replica context. Args: tensor: a `tf.Tensor` like object. The value to broadcast. destinations: a `tf.distribute.DistributedValues`, a `tf.Variable`, a `tf.Tensor` alike object, or a device string. It specifies the devices to broadcast to. Note that if it's a `tf.Variable`, the value is broadcasted to the devices of that variable, this method doesn't update the variable. Returns: A `tf.Tensor` or `tf.distribute.DistributedValues`. )r1broadcast_implementationrrr#s r$ broadcastzCrossDeviceOps.broadcasts ,'  ( ( >>r&ctd)amImplementation of `reduce`. Overriding this method is useful for subclass implementers. Args: reduce_op: a `tf.distribute.ReduceOp` specifying how values should be combined. per_replica_value: a `tf.distribute.DistributedValues`, or a `tf.Tensor` like object. destinations: a `tf.distribute.DistributedValues`, a `tf.Variable`, a `tf.Tensor` alike object, or a device string. It specifies the devices to reduce to. To perform an all-reduce, pass the same to `value` and `destinations`. Note that if it's a `tf.Variable`, the value is reduced to the devices of that variable, this method doesn't update the variable. options: a `tf.distribute.experimental.CommunicationOptions`. See `tf.distribute.experimental.CommunicationOptions` for details. Returns: A `tf.Tensor` or `tf.distribute.DistributedValues`. Raises: ValueError: if per_replica_value can't be converted to a `tf.distribute.DistributedValues` or if destinations is not a string, `tf.Variable` or `tf.distribute.DistributedValues`. z2_reduce method must be implemented in descendants.rrr?rr#rs r$rz$CrossDeviceOps.reduce_implementations: < >>r&ctd)aImplementation of `batch_reduce`. Overriding this method is useful for subclass implementers. Args: reduce_op: a `tf.distribute.ReduceOp` specifying how values should be combined. value_destination_pairs: a sequence of (value, destinations) pairs. See `reduce` for descriptions. options: a `tf.distribute.experimental.CommunicationOptions`. See `tf.distribute.experimental.CommunicationOptions` for details. Returns: A list of `tf.Tensor` or `tf.distribute.DistributedValues`, one per pair in `value_destination_pairs`. Raises: ValueError: if `value_destination_pairs` is not an iterable of tuples of `tf.distribute.DistributedValues` and destinations. zFbatch_reduce_implementation method must be implemented in descendants.r)rr?rPrs r$rz*CrossDeviceOps.batch_reduce_implementations. P r&c4t||d|jS)aImplementation of `broadcast`. Args: tensor: a `tf.Tensor` like object. The value to broadcast. destinations: a `tf.distribute.DistributedValues`, a `tf.Variable`, a `tf.Tensor` alike object, or a device string. It specifies the devices to broadcast to. `destinations`. Note that if it's a `tf.Variable`, the value is broadcasted to the devices of that variable, this method doesn't update the variable. Returns: A `tf.Tensor` or `tf.distribute.DistributedValues`. T)rvr6)r>rrs r$rz'CrossDeviceOps.broadcast_implementations# !77  99r&ctd)aAll-reduce the `value` across all replicas so that all get the result. `value` can be a nested structure of tensors or `IndexedSlices`. The implementation should generally batch the all-reduces when possible. `options` can be set to hint the batching behavior. This API must be called in a replica context. Args: reduce_op: A `tf.distribute.ReduceOp` value specifying how values should be combined. value: Value to be reduced. A tensor or a nested structure of tensors or `IndexedSlices`. replica_id: An integer indicating the id of the replica where this all_reduce is called under. This is the local replica id that ranges from 0 to len(local_devices) - 1. options: A `tf.distribute.experimental.CommunicationOptions`. Returns: A tensor/IndexedSlices or a nested strucutre of tensors/IndexedSlices with the reduced values. The structure is the same as `value`. z/_all_reduce must be implemented in descendants.r)rr?r@ replica_idrs r$ _all_reducezCrossDeviceOps._all_reduce+s. O PPr&rW)__name__ __module__ __qualname____doc__rpropertyrrrrrrrfor_subclass_implementersrrrrrr&r$rrs      .=`,0\>>/5b?&))>*>>))*4))9*96Qr&rzdistribute.ReductionToOneDevicec6eZdZdZdfd ZdZdZdZxZS)ReductionToOneDeviceaA CrossDeviceOps implementation that copies values to one device to reduce. This implementation always copies values to one device to reduce them, then broadcast reduced values to the destinations. It doesn't support efficient batching. Here is how you can use `ReductionToOneDevice` in `tf.distribute.MirroredStrategy`: ``` strategy = tf.distribute.MirroredStrategy( cross_device_ops=tf.distribute.ReductionToOneDevice()) ``` cj||_|xstj|_tt |y)aHInitializes with a device to reduce to and a way to accumulate. Args: reduce_to_device: the intermediate device to reduce to. If None, reduce to the first device in `destinations` of the `reduce` method. accumulation_fn: a function that does accumulation. If None, `tf.math.add_n` is used. N)rradd_nrsuperrr)rrr __class__s r$rzReductionToOneDevice.__init__Vs,-D*G >>'< 00r&c0~t|rt||j}nt||j}|jxs|d}t j tj d|d|ddt|||}|j||S)Nrz Gather to rr5r) r%rcrrrrrrr)rrr#rrrwrrs r$rz+ReductionToOneDevice._gather_implementationrs,' t/I/IJg !2D4N4NOg,,:   0@'JBP/1A4HH >>(L 11r&c ^|Dcgc]\}}|j||||c}}Scc}}w)N)r#r)r)rr?rPrtr\s r$rz0ReductionToOneDevice.batch_reduce_implementationsC ,  Aq "" qq' # ;  s))NN) rrrrrrrr __classcell__rs@r$rrEs  1 1 2r&rc|dj}tt|Dcgc]}g}}|D]F}t|jD],\}}|j|k(sJ||j |df.H|Scc}w)aeGroup values into sublists by their devices. This grouping is needed to call the all-reduce library because it expects a list of the following form: [[(grad0_gpu0, v0_gpu0), (grad1_gpu0, v1_gpu0), (grad2_gpu0, v2_gpu0) ...], [(grad0_gpu1, v0_gpu1), (grad1_gpu1, v1_gpu1), (grad2_gpu1, v2_gpu1) ...], [(grad0_gpu2, v0_gpu2), (grad1_gpu0, v1_gpu2), (grad2_gpu0, v2_gpu2) ...], ... ] Args: per_replica_values: a list of PerReplica objects. Returns: a list of lists, each sublist has components for its corresponding device of PerReplica objects, paired with a None. rN)r`rangerN enumerater rO)per_replica_valuesr#rngroupedrir\s r$_group_value_by_devicers$$A&//,s<01 2AR 2' 2-#+223#1  ' '< 77 7 ajD "## . 3s A>c&tt||z}tt|dDcgc]}g}}|D]}t|D]y\}\} }|tj j k(r@tj| j5||j| |z dddf||j| {|D cgc]'} tj| tj)c} Scc}w#1swYxYwcc} w)aUngroup results from all-reduce and make Mirrored objects. Each all-reduce result will be divided by the number of destinations before Mirrored objects are created if reduce_op is "mean". Args: grouped_reduced: a list of lists, each sublist has components for each device, paired with a None. It is the result from cross_device_utils.aggregate_gradients_using*. destinations: a value to colocate the result with. reduce_op: Indicates how values will be aggregated. Accepted values are `tf.distribute.ReduceOp.SUM`, `tf.distribute.ReduceOp.MEAN`. num_between_graph_workers: number of workers in the between-graph replication. Returns: a list of Mirrored objects. rNrq)rNrcrrr r<r=rr!rOrrtr(ru) grouped_reducedr#r?num_between_graph_workers num_replicasrnindexper_replica_reducedrr\s r$_ungroup_and_make_mirroredrs ,%l347PP,S!345 6!2 6% 6,23 6Aq k**// / ZZ ! , (//!l* + , , a 27 8,-  " "I&& ( 88 7 , , 8s C=D,DD c$eZdZdZddZdZdZy)_ConcatAndSplitPackerz,Concatenate and split tensors for reduction.c2|dkr td||_y)zInitialize the _ConcatAndSplitPacker object. Args: num_packs: specifies the number of split packs that will be formed. Raises: ValueError: if num_packs is not greater than 0. rz$num_packs must be greater than zero.N)r0 num_packs)rrs r$rz_ConcatAndSplitPacker.__init__sA~ = >>DNr&c ||_g|_g|_g}|D]}tj|dd5|Dcgc]\}}t j |dg}}}|Dcgc]\}}t j|}}}|Dcgc]\}}t j|}}}t j|d} |j} td|Dr7t|Dcgc]\}}|jj!c}}} nt j| } | | z} | | | dz zz } | g| dz z| gz}t j| |}|jt!|dg| z|jj||jj|ddd|Scc}}wcc}}wcc}}wcc}}w#1swYxYw)z Pack tensors.rc3VK|]!\}}|jj#ywrW)r;is_fully_defined)rXgrns r$rYz-_ConcatAndSplitPacker.pack..s!Ldaqww'')Lr]r4N)grouped_grads_and_varsall_device_shapesall_device_sizesr colocate_withrreshaper;sizerrr9sum num_elementssplitrOzip)rrdevice_grad_packsdevice_grads_and_varsrrn flat_grads device_shapes device_sizes concat_grads num_splitstotal_grad_size split_sizesplit_size_last split_sizes grad_packss r$packz_ConcatAndSplitPacker.packs"8DDD!7)3   215a8 9(34I +/1aI  a" &  9NN1+N N6KLda q)L L '' A6 ^^ L6KL L2GH$!Qqww##%HJ/&NN<8/$ 2 )J*q.,II!lj1n58II __\;?   Z$*1D!EF %%m4 $$\2Q(3(3)3V Q OL"I3(3(3sBG!G G(G GG .AG/$GB&GGG( c Lg}t||j|j|jD]\}}}}t j |dd5|Dcgc]\}}| }}}t j|d} t j| |} t|| D cgc]\} } t j| | } } } t| |Dcgc] \}\}}||f}}}}|j|ddd|Scc}}wcc} } wcc}}}w#1swYxYw)zReverse the pack.rN) rrrrrrrrrrrO)rsummed_device_grad_packsaggregated_device_gradsrrrrrnrdevice_grads_concatgrads_with_sizesr;gradgrads_with_shapesr\summed_device_gradss r$unpackz_ConcatAndSplitPacker.unpack sX @C %t'B'B  # #T%:%:A<< = !  |   5a8; <<+CD41aQDD(../@!D%??+> M #=2BC t   dE *  %((9(=%?   q&1aQF   &&':;+<<<: #")E  #<D" D DD &DDD# Nr4)rrrrrrrrr&r$rrs4 2h #r&rc\|dkDr t|}|j|}||fSd}|}||fS)zPack tensors if specified.rN)rr) device_gradsr tensor_packerrs r$ _pack_tensorsr0sI]))4M%**<8 M ))M$ M ))r&c,|r|j|S|S)z4Unpack tensors if they are packed before all-reduce.)r)rrs r$_unpack_tensorsr;s    (( .r&cHeZdZdZd fd ZdZdZdZdZdZ dZ xZ S) AllReduceCrossDeviceOpsaIAll-reduce implementation of CrossDeviceOps. It performs all-reduce when applicable using NCCL or hierarchical copy. For the batch API, tensors will be repacked or aggregated for more efficient cross-device transportation. For reduces that are not all-reduce, it falls back to `tf.distribute.ReductionToOneDevice`. cd||_||_t|_tt |y)aInitializes the object. Args: all_reduce_alg: the all-reduce algorithm to use, currently only "nccl" or "hierarchical_copy" are supported. num_packs: a non-negative integer. The number of packs to split values into. If zero, no packing will be done. N)_all_reduce_alg _num_packsr_simple_cross_replica_opsrrr)rall_reduce_algrrs r$rz AllReduceCrossDeviceOps.__init__Ms-*DDO%9%;D" !413r&c~t||r1tdt|Ds|j||gdS|jj |||S)Nc3@K|]}d|jvyw)cpuNlowerrXrls r$rYz@AllReduceCrossDeviceOps.reduce_implementation..asKq"Kr)rhanyrc_batch_all_reducer rrs r$rz-AllReduceCrossDeviceOps.reduce_implementation[se (,7 K,<\,JK K  # #I0A/B CA FF  + + 2 29>O3?AAr&c t|r#|j||Dcgc]}|d c}S|Dcgc]\}}|j||||c}}Scc}wcc}}w)Nr)rorr)rr?rPrr\r@dests r$rz3AllReduceCrossDeviceOps.batch_reduce_implementationgsq12  # #I3J$KaQqT$KMM 5eT  $ $YtW E%Ls A Actj|\}}}}|r|j||}ng}|r|j||}ng}tj||f||ffS)z All-reduce algorithm in a batch.)rsplit_by_sparsity_do_batch_all_reduce_do_batch_all_reduce_sparse stitch_values) rr?r dense_values dense_indices sparse_valuessparse_indices dense_resultssparse_resultss r$rz)AllReduceCrossDeviceOps._batch_all_reducers ,,-?@?L-// <Hmm77 8EGnn  + +m]-K. 63r&rAllReduceSpecTuplezalg shards limitzdistribute.NcclAllReducec$eZdZdZdfd ZxZS) NcclAllReduceaNCCL all-reduce implementation of CrossDeviceOps. It uses Nvidia NCCL for all-reduce. For the batch API, tensors will be repacked or aggregated for more efficient cross-device transportation. For reduces that are not all-reduce, it falls back to `tf.distribute.ReductionToOneDevice`. Here is how you can use `NcclAllReduce` in `tf.distribute.MirroredStrategy`: ``` strategy = tf.distribute.MirroredStrategy( cross_device_ops=tf.distribute.NcclAllReduce()) ``` cn|dkrtdj|tt|d|y)zInitializes the object. Args: num_packs: a non-negative integer. The number of packs to split values into. If zero, no packing will be done. Raises: ValueError: if `num_packs` is negative. rz= 0, but {} is specifiedr%r rN)r0formatrr.rrrrs r$rzNcclAllReduce.__init__sE1}  H O O   -'(4r&rrrrrrrrs@r$r.r.s"44r&r.z$distribute.HierarchicalCopyAllReducec$eZdZdZdfd ZxZS)HierarchicalCopyAllReduceaAHierarchical copy all-reduce implementation of CrossDeviceOps. It reduces to one GPU along edges in some hierarchy and broadcasts back to each GPU along the same path. For the batch API, tensors will be repacked or aggregated for more efficient cross-device transportation. This is a reduction created for Nvidia DGX-1 which assumes GPUs connects like that on DGX-1 machine. If you have different GPU inter-connections, it is likely that it would be slower than `tf.distribute.ReductionToOneDevice`. For reduces that are not all-reduce, it falls back to `tf.distribute.ReductionToOneDevice`. Here is how you can use `HierarchicalCopyAllReduce` in `tf.distribute.MirroredStrategy`: ``` strategy = tf.distribute.MirroredStrategy( cross_device_ops=tf.distribute.HierarchicalCopyAllReduce()) ``` cn|dkrtdj|tt|d|y)zInitializes the object. Args: num_packs: a non-negative integer. The number of packs to split values into. If zero, no packing will be done. Raises: ValueError if `num_packs` is negative. rz=HierarchicalCopy requires num_packs >= 0, but {} is specifiedhierarchical_copyr0N)r0r1rr5rr2s r$rz"HierarchicalCopyAllReduce.__init__sC1}  I 6)   #T3*4r&rr3rs@r$r5r5s,r&r5cbeZdZdZ d fd ZedZdZdZdZ dZ dZ d Z d Z xZS) CollectiveAllReducezAll-reduce cross device ops using collective ops. In the between-graph replicated training, it will still do all-reduces across all workers and then put results on the right destinations. c|t|zdkDr td||_||_|xst j |_tj|_ |rtd|D|_ ntd|D|_ |j j|j}g|_ d|_|jD]X}t j|||j ||}|jj!||j#rRd|_Zt$t&|S||_y)aInitializes the object. Args: devices: a list of device strings to run collectives on. group_size: the global group size. For between-graph replicated training it's the total number of devices across all workers. options: a `tf.distribute.experimental.CommunicationOptions`. collective_keys: an optional CollectiveKey object. canonicalize_devices: Whether to canonicalize devices for workers or not. rz6group_size must be divisible by the number of devices.c3FK|]}tj|ywrWr canonicalizers r$rYz/CollectiveAllReduce.__init__..CsIAK44Q7I!c3FK|]}tj|ywrW)rrbrs r$rYz/CollectiveAllReduce.__init__..Es"M?@+ 7 7 :Mr>FTN)rNr0 _group_size_optionsrCollectiveKeys_collective_keys threadingLock_lockrMr` get_group_key _launchers _limited_ncclCollectiveReplicaLauncherrOcan_order_ncclrr9rr) rrw group_sizercollective_keysr6 group_keyr!launcherrs r$rzCollectiveAllReduce.__init__s- CL 1$ O PP!DDM,A/>>@ !DJIIIdmMDKMMdm%%33DMMBIDOD--"#== Z!6!6Ih ooX&  $ $ &! " t-/!5Dr&cF|jt|jz SrW)r@rNr`rs r$rz.CollectiveAllReduce._num_between_graph_workersWs   c$--0 00r&c tj|}|jrr|jtj j k(rKt|dk(r=|jt jtj j}|j|}tj|\}}} } g} g} |r+|jtj||j } t#j$sX|dk(rSt'j(dt|t|j|j*|jt| |j-| |} |t.j0j2k(rPt5| D]B\}}t7j8|j:|5||j*z | |<dddD| j| r4t#j$sN|dk(rIt'j(dt| t|j|j*|j| D]#}| j=|j?||%|t.j0j2k(rt5| D]\}}t7j8|j:|5tAjB| |jD|j*z | |jF| |jH| |<dddtjJ| |f| | ff}tjL||S#1swYxYw#1swYxYw)z%Implements CrossDeviceOps.all_reduce.r4implementationrzuCollective all_reduce tensors: %d all_reduces, num_devices = %d, group_size = %d, implementation = %s, num_packs = %dNzjCollective all_reduce IndexedSlices: %d all_reduces, num_devices =%d, group_size = %d, implementation = %s)r indices dense_shape)'rflattenrIrSrCommunicationImplementationNCCLrNmergerRINGrHrrreverse group_by_sizebytes_per_packr executing_eagerlyrinfor@batch_all_reducer r<r=rrr!r`rOall_reduce_indexed_slicesrr*r rTrUrpack_sequence_as)rr?r@rr flat_valuesrOrrr r!r"r#packsrr\ indexed_slice flat_resultss r$rzCollectiveAllReduce._all_reduce\s,,u%K w55  6 6 ; ; < KA  ! !,HHMM OPgz*H,,[9?L-MN ..|/6/E/EGe & & (Z1_  C  s4??3T5E5E  " "CJ  0 //w?m k**// /m, 4DAqzz$-- 34 4 4#3#33M!  4 4 4  & & (Z1_  78;M8J  $"2"2G4J4J L )H-  . .}g F HH k**// /n- ;DAqzz$-- 34 ; . < <%a(//$2B2BB&q)11*1-99!;N1  ; ; ;&33  '..)IJLL   553 4 4" ; ;sM AM$M! $M- c h jDcgc]}gc} tj}|D]3}t|D]#} |j|j|%5t j r fd}j5tjjtj} | j|tt|} | jdddnNg} j5t|D](}| jj ||* dddg} t D]6} | jt!j"| t$j&8| Scc}w#1swYSxYw#1swY_xYw)z'All reduce a list of per_replica_value.ctj5j||cdddS#1swYyxYwrW)r eager_moder) device_idrr?rvalues_by_devices r$ thread_fnzECollectiveAllReduce._all_reduce_per_replica_values..thread_fnsA    ! 6!!)-=i-H"+W6 6 6 6s 7ANrq)r`rNrrOr r r^rFmultiprocessingpool ThreadPoolmaprLcloserrrrtr(ru)rr?rrrn num_devicesrSrrlrnoutputs_by_devicerQr rks`` ` @r$_all_reduce_per_replica_valuesz2CollectiveAllReduce._all_reduce_per_replica_valuess%)MM2q2dmm$K): [!:!"";#5#5a#89::  "6 ::##..s4==/AB HHYU;5G0HI   ::L{# LA  " "y*:1*=q'J L LL F()K mm  " "6i6H6H IKK M93 LLs FA'F7F(F%(F1c tj|j||g|d}t||j}t |||jr|St |tjstj|g}g}tj|j5|D]}tj|5|jD]7} | j|k(s|jtj| n/|jtj|j ddd dddt#j$|tjS#1swYxYw#1swY:xYwNrrq)r mark_as_unsaveablertrcrrhrr(rurcontrol_dependenciesr r!rOrr_primaryrrt) rr?rr#r all_reducedrwrrlr\s r$rz)CollectiveAllReduce.reduce_implementationsk""$55i7H6I6=??@BK|T-G-GHG'002  k9#5#5 6&& }5k E ! !+"4"4 5 C C! ZZ] C%% Caxx1}ll9--a01 C LL++K,@,@A B C C C C  # #Ei6H6H II C C C Cs+,FE4'AE4< F4E= 9FF c Ttjt||j}|r$|j ||Dcgc]}|d c}|S|s%t j t jdd|Dcgc]\}}|j||||c}}Scc}wcc}}w)NrzFEfficient batch_reduce is not supported if destinations are different.r) r rwrorrtrrr)r)rr?rPrall_devices_matchr\r@rs r$rz/CollectiveAllReduce.batch_reduce_implementations""$*+B+/+E+EG  0 0 $;}|j j|j|j|@|j |v dddddd|S#1swYxYw#1swY|SxYw)N allgather) rFr name_scoperrNr`rOrH all_gatherr )gathered_valuesrSoutputsrrrrrs r$compute_gathered_valueszFCollectiveAllReduce._batch_all_gather..compute_gathered_values-so ::*s~~k2*- *K'T]]+, 7a NN4??1-88""1%tW6 7 7   )  ******s#B?A:B3!B?3B< 8B??C rq)rNrIrSrrWrXrYrrZrrrr`r@r r^rfunctionrOrrtr(ru) rrrr batch_sizerrmirroredr@s ```` r$r~z%CollectiveAllReduce._batch_all_gathers-'(J w55  6 6 ; ; <a  ! !,HHMM OPg  C S MM ,,g.D.D FFHJ K   "F --.EFHo/1oH Joo  " "5Y5G5G HJJ Or&ctj|j|}t|j|j |j ||jSrW)copydeepcopyrCr9r`r@rAr)rmemorMs r$ __deepcopy__z CollectiveAllReduce.__deepcopy__CsHmmD$9$94@O t}}d.>.> .0J0J LLr&r)rrrrrrrrrtrrrr~rrrs@r$r9r9sU $$( 96v 1 1H6TB JD&J<'RLr&r9c td|D}tjrtjj d}tjj d}t |t |k7rtjdtStjj }ntj|}t}|D]?}tj|j|vs%|j|jAt |t |k7r1tjddj!t#||z t%d|Drtjd tSt'j(d r t+d Stjd tS)aJFind the best `CrossDeviceOps` locally given a `tf.compat.v1.ConfigProto`. Args: devices: a list of devices passed to `tf.distribute.Strategy`. session_config: a `tf.compat.v1.ConfigProto` or `None`. If `None`, it will make decision based on all logical devices. Returns: A subclass of `CrossDeviceOps`. c3FK|]}tj|ywrWr<rs r$rYz*select_cross_device_ops..VsG!+2215Gr>GPU) device_typezUNCCL is not supported when using virtual GPUs, fallingback to reduction to one device)session_configzTSome requested devices in `tf.distribute.Strategy` are not visible to TensorFlow: %s,c3@K|]}d|jvyw)gpuNrrs r$rYz*select_cross_device_ops..ms;Aaggi ;rzPThere are non-GPU devices in `tf.distribute.Strategy`, not using nccl allreduce.r.r4)rz3Nccl kernel is not found, not using nccl allreduce.)rer#executing_eagerly_outside_functionsr list_logical_deviceslist_physical_devicesrNrwarningrrlist_local_devicesrr=nameaddjoinrLrrget_registered_kernels_for_opr.)rwrrequested_devices logical_gpus physical_gpusmachine_devices using_devicesrls r$select_cross_device_opsrKsGwGG,,.??$99e9LLOO%;;;NM <C .. oo89 ! ##oo'<<>O 33%'O%-  a'+<<  3011 OO  XXd+<}+L&MNP ;):;; OO01  !! **?; 1 %% OOIJ  !!r&)T)FTr)rrW)Lr collectionsrmultiprocessing.dummyrmmultiprocessing.poolrDnumpyr8r,tensorflow.python.clientrtensorflow.python.distributerrrrrr r r r(r tensorflow.python.eagerr rtensorflow.python.frameworkrrrrrrtensorflow.python.opsrrrtensorflow.python.platformrrtensorflow.python.utilr tensorflow.python.util.tf_exportrtensorflow.tools.docsrr%r1rBrIrTr^rcrhror>rrobjectrrrrrrrrAllReduceCrossTowerOps namedtupler,r.r5rWCollectiveCommunicationr9rrr&r$rsF  /8;49243<4+06/+<3+*7<'6." 6&7; HB : *  O9 &+*.C$(  &'FQVFQ(FQR  ,-@>@.@F>:; 8Fe#Fe#P*r3nr3l1,[++,@,>@ %&!4+!4'!4H 12' 7'3'V*EE-IIsL.sLl +"r&