o ?e@sdZddlZddlZddlZddlZddlZddlZddl 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-ddZ.ddZ/ dTd d!Z0d"d#Z1d$d%Z2d&d'Z3dTd(d)Z4dTd*d+Z5dTd,d-Z6 . dUd/d0Z7d1d2Z8d3d4Z9e+d5Gd6d7d7e:Z;e+d8Gd9d:d:e;Zd?Z>Gd@dAdAe:Z?dWdBdCZ@dXdDdEZAGdFdGdGe;ZBeBZCeDdHdIZEe+dJGdKdLdLeBZFe+dMGdNdOdOeBZGe jHZIe jHZHGdPdQdQe;ZJdXdRdSZKdS)Yz?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_controlscCs$t|tjtjfrt|jSt|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. ) isinstancerZBaseResourceVariable tensor_libTensorbooldevice destinationsr!n/home/www/facesmatcher.com/pyenv/lib/python3.10/site-packages/tensorflow/python/distribute/cross_device_ops.pycheck_destinations4s  r#cCsHt|tjtjtjtjt j t j fst |stdt|s"tddS)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 IndexedSlicesrZAggregatingVariablesix string_typesr ZTPUMirroredVariablerZis_resource_variable ValueErrorr#rr!r!r"validate_destinationsEsr*TcCs|t|tjr tdt|st|dkrtj|j |j dS|t j j kr'|S|dkr3td||ft|t|||dS)z8Reduce a non-DistributedValue `value` to `destinations`.z^You are passing a `DistributedValues` to `reduce_non_distributed_value`, which is not allowed.r)dtypezOA non-DistributedValues value %s cannot be reduced with the given reduce op %s.canonicalize_devices)rr$r%r)rZ is_tf_typenpallZzerosshaper+rReduceOpMEANr*simple_broadcast) reduce_opvaluer Znum_replicas_in_graphr.r!r!r"reduce_non_distributed_valueTs  r7cCsBt|tjr|St|st|}t|drt|fSt d)z2Converts a single tensor into a PerReplica object.rzZCannot convert `input_tensor` to a `PerReplica` object because it doesn't have device set.) rr$r%rZ is_tensorrZconvert_to_tensorhasattr PerReplicar))Z input_tensorr!r!r"_make_tensor_into_per_replicaus     r:cCstg}t|}t|ttfstd|D]$}t|tstdt|dkr(tdt|d}|||dfq|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..css |] }t|dtjVqdSrN)rr$r9rFvr!r!r"rGsT)rr<r=r0)r@r!r!r"!_validate_value_destination_pairssrLcCsbt|tjr |jS|rt|tjrt|fSt|jfSt|tjr*t |fSt |jfSrE) rr$r%_devicesr'r(rresolver!canonicalize_without_job_and_task)r r.r!r!r"get_devices_froms     rPcCs$||uptt||tt||kSrE)setrP)leftrightr.r!r!r"_devices_matchs rTcsBtfddDs dStfddddDsdSdS)Nc3s |] \}}t||VqdSrErT)rFrKdr-r!r"rGs  z%_all_devices_match..Fc3s(|]\}}t|ddVqdSrIrUrFrK_r.r@r!r"rGs  r,T)r0)r@r.r!rYr"_all_devices_matchs  rZFcCsZt||}t|dkr|st||dSg}|D] }|t||qtj|tjdS)z8Broadcast `value` to `destinations` using simple copies.r,rZ wrap_class) rPr>rZ'copy_tensor_or_indexed_slices_to_devicer?rregroupr$Mirrored)r6r always_mirroredr.devicesZ value_updatesrVr!r!r"r4s  r4c Cs|j}|s tdt|}t|Dttj%t ||}|t j j kr-t ||}n |t j jkr7tdWdn1sAwYWd|SWd|S1sYwY|S)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 r)r>rrr device_policyDEVICE_PLACEMENT_SILENTrZ#aggregate_tensors_or_indexed_slicesrr2r3Z%divide_by_n_tensors_or_indexed_slicesZSUM)per_replica_valuereduce_to_deviceaccumulation_fnr5 all_valuescountreducedr!r!r"_simple_reduces0        ric Cs|j}|s tdt|-ttjt||}Wdn1s&wYWd|SWd|S1s>wY|S)zAConcatenate all values in the DistributedValues input and return.r`N) r r)rrr rarbrconcat)rcrdaxisrfgatheredr!r!r"_simple_gathers   rmzdistribute.CrossDeviceOpsc@seZdZdZddZeddZdddZdd d Zd d Z dd dZ ddZ e j ddZe j ddZe j ddZddZdS)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` cCs d|_dSNT)_canonicalize_devicesselfr!r!r"__init__ szCrossDeviceOps.__init__cCsdS)Nr,r!rqr!r!r"_num_between_graph_workerssz)CrossDeviceOps._num_between_graph_workersNcCs|durt}t|}t||jdkrMt|jdkrMt|||jrMt |jdj t |jd}Wdn1s?wYt j|ftjdS|durUt}|||||S)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`. Nr,rr[)rOptionsr:r*rtr>r rTrprrridentityrr\r$r]reduce_implementation)rrr5rcr optionsrKr!r!r"reduces, zCrossDeviceOps.reducecCst|tjr td|durt}t|}t||jdkrWt |j dkrWt |||j rWt |j djt|j d}Wdn1sIwYtj|ftjdS|||||S)a>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 IndexedSlicesNr,rr[)rrr&NotImplementedErrorrrur:r*rtr>r rTrprrrrvrr\r$r]_gather_implementation)rrrcr rkrxrKr!r!r"_gatherCs,   zCrossDeviceOps._gathercCtd)aImplementation of `gather` method of `tf.distribute.CrossDeviceOps`. Overriding this method is useful for subclass implementers. 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, 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`. z2_gather method must be implemented in descendants.rzrrrcr rkrxr!r!r"r{qsz%CrossDeviceOps._gather_implementationcCs|durt}t|st|}|D]\}}t|q|jdkr8t||jr8t|ddj dkr8dd|DS|dur@t}| |||S)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. Nr,rcSs"g|] \}}tj|jtjdqSr[)rr\r r$r]rWr!r!r" sz/CrossDeviceOps.batch_reduce..) rrurLrDr*rtrZrpr>r batch_reduce_implementation)rrr5r@rxrXrVr!r!r" batch_reduces2    zCrossDeviceOps.batch_reducecCst||||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`. )r*broadcast_implementationrrrr r!r!r" broadcasts zCrossDeviceOps.broadcastcCr})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.r~rrr5rcr rxr!r!r"rwsz$CrossDeviceOps.reduce_implementationcCr})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~rrr5r@rxr!r!r"rsz*CrossDeviceOps.batch_reduce_implementationcCst||d|jdS)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)r^r.)r4rprr!r!r"rs z'CrossDeviceOps.broadcast_implementationcCr})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 interger 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~)rrr5r6 replica_idrxr!r!r" _all_reduce+szCrossDeviceOps._all_reducerE)__name__ __module__ __qualname____doc__rspropertyrtryr|r{rrrZfor_subclass_implementersrwrrrr!r!r!r"rns"    0. 1    rnzdistribute.ReductionToOneDevicecs:eZdZdZd fdd ZddZddZd d ZZS) 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()) ``` Ncs$||_|ptj|_tt|dS)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)rdrZadd_nresuperrrs)rrrdre __class__r!r"rsVs zReductionToOneDevice.__init__cCsf~t|r t||j}nt||j}|jp|d}ttjd||fdt|||j|}| ||S)Nrz"Reduce to %s then broadcast to %r. ) r#rPrprdlogging log_first_nINFOrirer)rrr5rcr rxr_rdrhr!r!r"rwcs   z*ReductionToOneDevice.reduce_implementationcCsb~t|r t||j}nt||j}|jp|d}ttjd||fdt|||}|||S)Nrz"Gather to %s then broadcast to %r.r) r#rPrprdrrrrmr)rrrcr rkrxr_rdrlr!r!r"r{rs    z+ReductionToOneDevice._gather_implementationcsfdd|DS)Ncs"g|] \}}j||dqS))r rxrw)rFtrKrxr5rrr!r"rs zDReductionToOneDevice.batch_reduce_implementation..r!rr!rr"rsz0ReductionToOneDevice.batch_reduce_implementation)NN) rrrrrsrwr{r __classcell__r!r!rr"rEs  rcCsb|dj}ddtt|D}|D]}t|jD]\}}|j|ks$J|||dfqq|S)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. rcSg|]}gqSr!r!rFrXr!r!r"rz*_group_value_by_device..N)rMranger> enumerater r?)per_replica_valuesr groupedrcirKr!r!r"_group_value_by_devices rr,c Cstt||}ddtt|dD}|D]:}t|D]3\}\}} |tjjkrIt|j|| ||Wdn1sCwYq|| |qqdd|DS)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. cSrr!r!rr!r!r"rrz._ungroup_and_make_mirrored..rNcSsg|] }tj|tjdqSr)rr\r$r]rJr!r!r"rs  ) r>rPrrrr2r3rrr?) Zgrouped_reducedr r5Znum_between_graph_workersZ num_replicasindexZper_replica_reducedrrKrXr!r!r"_ungroup_and_make_mirroreds rc@s*eZdZdZd ddZddZddZd S) _ConcatAndSplitPackerz,Concatenate and split tensors for reduction.r,cCs|dkrtd||_dS)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)r) num_packsrrrr!r!r"rss  z_ConcatAndSplitPacker.__init__c Cs(||_g|_g|_g}|D]}t|ddqdd|D}dd|D}dd|D}t|d}|j}tdd|DrJt dd|D} nt |} | |} | | |d } | g|d | g} t || } | t | d g||j ||j |Wd n1swYq |S) z Pack tensors.rcSsg|] \}}t|dgqS)rZreshaperFgrXr!r!r"rsz._ConcatAndSplitPacker.pack..cSg|] \}}t|qSr!)rr1rr!r!r"rcSrr!)rsizerr!r!r"rrcss|] \}}|jVqdSrE)r1Zis_fully_definedrr!r!r"rGsz-_ConcatAndSplitPacker.pack..cSsg|] \}}|jqSr!)r1Z num_elementsrr!r!r"rrr,N)grouped_grads_and_varsall_device_shapesall_device_sizesr colocate_withrrjrr0sumrsplitr?zip)rrrdevice_grad_packsdevice_grads_and_varsZ flat_grads device_shapes device_sizesZ concat_gradsZ num_splitsZtotal_grad_sizeZ split_sizeZsplit_size_lastZ split_sizesZ grad_packsr!r!r"packs8     *z_ConcatAndSplitPacker.packc Csg}t||j|j|jD]K\}}}}t|dd4dd|D}t|d}t||}ddt||D} ddt| |D} | | Wdn1sRwYq |S)zReverse the pack.rcSsg|]\}}|qSr!r!rr!r!r"rz0_ConcatAndSplitPacker.unpack..cSsg|] \}}t||qSr!r)rFr1Zgradr!r!r"r"s cSsg|] \}\}}||fqSr!r!)rFrrXrKr!r!r"r(sN) rrrrrrrrjrr?) rrZsummed_device_grad_packsZaggregated_device_gradsrrrrZdevice_grads_concatZgrads_with_sizesZgrads_with_shapesZsummed_device_gradsr!r!r"unpack s.   z_ConcatAndSplitPacker.unpackNr,)rrrrrsrrr!r!r!r"rs   4rcCs2|dkrt|}||}||fSd}|}||fS)zPack tensors if specified.rN)rr)Z device_gradsr tensor_packerrr!r!r" _pack_tensors0s rcCs|r||S|S)z4Unpack tensors if they are packed before all-reduce.)r)rhrr!r!r"_unpack_tensors;s rcsReZdZdZdfdd ZddZdd Zd d Zd d ZddZ ddZ Z 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`. ncclr,cs&||_||_t|_tt|dS)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_opsrrrs)rrall_reduce_algrrr!r"rsMs z AllReduceCrossDeviceOps.__init__cCsD~t||rtddt|Ds|||gdS|j|||S)Ncss|] }d|vVqdS)cpuNlowerrFrVr!r!r"rGaz@AllReduceCrossDeviceOps.reduce_implementation..r)rTanyrP_batch_all_reducerryrr!r!r"rw[s  z-AllReduceCrossDeviceOps.reduce_implementationcs4t|rdd|DSfdd|DS)NcSg|]}|dqSrr!rJr!r!r"rkrzGAllReduceCrossDeviceOps.batch_reduce_implementation..c g|] \}}||qSr!rrFr6destrr!r"rm)rZrrr!rr"rgs z3AllReduceCrossDeviceOps.batch_reduce_implementationc CsTt|\}}}}|r|||}ng}|r|||}ng}t||f||ffS)z All-reduce algorithm in a batch.)rsplit_by_sparsity_do_batch_all_reduce_do_batch_all_reduce_sparse stitch_values) rrr5r dense_values dense_indices sparse_valuessparse_indices dense_resultssparse_resultsr!r!r"rrs  z)AllReduceCrossDeviceOps._batch_all_reducecCsttjdt||j|jfd|dj}t|}t||j\}}|jdkr-t |}nt ||}t ||}t ||d|S)zRun batch all-reduces.zDbatch_all_reduce: %d all-reduces with algorithm = %s, num_packs = %drrr)rrrr>rrrMrrrZaggregate_gradients_using_ncclZ+aggregate_gradients_using_hierarchical_copyrr)rrr5rr rrrrhr!r!r"rs(   z,AllReduceCrossDeviceOps._do_batch_all_reducecCs,ttjdt|d|j|t||S)z'Run batch all-reduce for sparse values.z9Efficient allreduce is not supported for %d IndexedSlicesr)rrWARNr>rrr)rrr5rr!r!r"rs z3AllReduceCrossDeviceOps._do_batch_all_reduce_sparsecCs"ttjddt||||S)Nzgather/all_gather with NCCL or HierarchicalCopy is not supported. Falling back to gather on one device and then broadcast. We're working on a more efficient implementation.)rrrrr|rr!r!r"r{s z.AllReduceCrossDeviceOps._gather_implementation)rr,) rrrrrsrwrrrrr{rr!r!rr"rBs    rAllReduceSpecTuplezalg shards limitzdistribute.NcclAllReducec"eZdZdZdfdd ZZS) 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()) ``` r,c.|dkr td|tt|jd|ddS)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 specifiedrrrN)r)formatrrrsrrr!r"rss   zNcclAllReduce.__init__rrrrrrsrr!r!rr"rsrz$distribute.HierarchicalCopyAllReducecr)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()) ``` r,cr)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 specifiedZhierarchical_copyrN)r)rrrrsrrr!r"rss   z"HierarchicalCopyAllReduce.__init__rrr!r!rr"rsrcsjeZdZdZ  dfdd ZeddZdd Zd d Zd d Z ddZ ddZ ddZ ddZ ZS)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. NTc s|t|dkr td||_||_|pt|_t|_ |r+t dd|D|_ n t dd|D|_ |j |j }g|_ d|_|j D]}t|||j||}|j ||s^d|_qEtt|||_dS) 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.cs|]}t|VqdSrEr canonicalizerr!r!r"rGCrHz/CollectiveAllReduce.__init__..csrrE)rrOrr!r!r"rGEs  FTN)r>r) _group_size_optionsrZCollectiveKeys_collective_keys threadingLock_lockr=rMZ get_group_key _launchers _limited_ncclZCollectiveReplicaLauncherr?Zcan_order_ncclrrrsrp) rrr_Z group_sizerxcollective_keysr.Z group_keyrlauncherrr!r"rss4      zCollectiveAllReduce.__init__cCs|jt|jSrE)rr>rMrqr!r!r"rtWsz.CollectiveAllReduce._num_between_graph_workersc Cst|}|jr |jtjjkr t|dkr |tj tjj d}|j |}t |\}}} } g} g} |r|t ||j} tsZ|dkrZtdt|t|j |j|jt| || |} |tjjkrt| D]#\}}t|j|||j| |<Wdn1swYqj| | rts|dkrtdt| t|j |j|j| D] }| |||q|tjjkrt| D]2\}}t|j|t j!| |j"|j| |j#| |j$d| |<Wdn1swYqt %| |f| | ff}t&||S)z%Implements CrossDeviceOps.all_reduce.r,implementationrzuCollective 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)'rflattenrrrCommunicationImplementationNCCLr>mergeruRINGrrrreverseZ group_by_sizeZbytes_per_packr executing_eagerlyrinforZbatch_all_reducerr2r3rrrrMr?Zall_reduce_indexed_slicesrr&r rrrZpack_sequence_as)rrr5r6rrxZ flat_valuesrrrrrrrZpacksrrKZ indexed_sliceZ flat_resultsr!r!r"r\sx             zCollectiveAllReduce._all_reducec s4ddjDtj}|D]}t|D] }||j|qqtr[fdd}jtj tj}| |t t|} | Wdn1sUwYn)g} jt|D]}| ||qeWdn1swYg} t| D] } | tj| tjdq| S)z'All reduce a list of per_replica_value.cSrr!r!rr!r!r"rrzFCollectiveAllReduce._all_reduce_per_replica_values..cs@t||WdS1swYdSrE)r Z eager_moder)Z device_idrxr5rrZvalues_by_devicer!r" thread_fns  $zECollectiveAllReduce._all_reduce_per_replica_values..thread_fnNr[)rMr>rr?r r rrmultiprocessingpoolZ ThreadPoolmapr<closerrrr\r$r]) rrr5rrxZ num_devicesrCrrrZoutputs_by_devicerAr r!rr"_all_reduce_per_replica_valuess8     z2CollectiveAllReduce._all_reduce_per_replica_valuesc Cst|||g|d}t||j}t|||jr|St|tjs)t|g}g}t |j @|D]5}t |&|j D]} | j |krP| t| qZq?| t|jWdn1sdwYq4Wdn1stwYtj|tjdSNrr[)r mark_as_unsaveabler rPrprTrr$r]rcontrol_dependenciesr rr?rrv_primaryrr\) rrr5rcr rxZ all_reducedr_rrVrKr!r!r"rws<       z)CollectiveAllReduce.reduce_implementationcsZtt|j}|rdd|DS|s"ttjddfdd|DS)NcSrrr!rJr!r!r"rrzCCollectiveAllReduce.batch_reduce_implementation..zFEfficient batch_reduce is not supported if destinations are different.rcrr!rrrr!r"rr)r r rZrpr rrr)rrr5r@rxZall_devices_matchr!rr"rs z/CollectiveAllReduce.batch_reduce_implementationc Cs||g||d}tt||j}t|||jr|St|tjs)t|g}g}t |j @|D]5}t |&|j D]} | j |krP| t| qZ| t|jq?Wdn1sdwYq4Wdn1stwYtj|tjdSr )_batch_all_gatherr r rPrprTrr$r]rr r rr?rrvrrr\) rrrcr rkrxZ all_gatheredr_rrVrKr!r!r"r{s0       z*CollectiveAllReduce._gather_implementationc st}jrjtjjkr|dkrtjtjjdt t j d|tj j jfdfdd}trDt|}n|}g}|D] }|tj|tjdqK|S)z'all gather multiple per-replica-values.r,rzeCollective batch_all_gather: %d all-gathers, num_devices = %d, group_size = %d, implementation = %s, rc sg}jLtd.D]#}g}ttjD]}|j||j |q||qWdn1srMr?rZ all_gatherr )gathered_valuesrCZoutputsrrkrxrrrr!r"compute_gathered_values-s   ( zFCollectiveAllReduce._batch_all_gather..compute_gathered_valuesr[)r>rrrrrrrurrrrrMrr rr functionr?rr\r$r]) rrrrkrxZ batch_sizerrmirroredr6r!rr"rs>  z%CollectiveAllReduce._batch_all_gathercCs&t|j|}t|j|j|j||jSrE)copydeepcopyrrrMrrrp)rrmemorr!r!r" __deepcopy__Csz CollectiveAllReduce.__deepcopy__ro)rrrrrsrrtrr rwrr{rrrr!r!rr"rs ; J!")rcCstdd|D}tr4tjdd}tjdd}t|t|kr-tdt St}nt j |d}t}|D]}t |j|vrO||jq?t|t|kretddt||td d|Drvtd t Std rtd d Stdt S)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`. csrrErrr!r!r"rGVrHz*select_cross_device_ops..ZGPU)Z 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,css|] }d|vVqdS)ZgpuNrrr!r!r"rGmrzPThere are non-GPU devices in `tf.distribute.Strategy`, not using nccl allreduce.rr,)rz3Nccl kernel is not found, not using nccl allreduce.)rQrZ#executing_eagerly_outside_functionsr Zlist_logical_devicesZlist_physical_devicesr>rwarningrrZlist_local_devicesrrnameaddjoinr<rrZget_registered_kernels_for_opr)r_rZrequested_devicesZ logical_gpusZ physical_gpusZmachine_devicesZ using_devicesrVr!r!r"select_cross_device_opsKs8       r)T)FTrrrE)Lr collectionsrZmultiprocessing.dummyrZmultiprocessing.poolrnumpyr/r'Ztensorflow.python.clientrZtensorflow.python.distributerrrrrrr r r$r Ztensorflow.python.eagerr r Ztensorflow.python.frameworkrrrrrrZtensorflow.python.opsrrrZtensorflow.python.platformrrZtensorflow.python.utilrZ tensorflow.python.util.tf_exportrZtensorflow.tools.docsrr#r*r7r:rDrLrPrTrZr4rirmobjectrnrrrrrrrZAllReduceCrossTowerOps namedtuplerrrrZCollectiveCommunicationrrr!r!r!r"s                         !     KC # h v$+8