o ?ej@srdZddlZddlZddlmZmZmZmZddlm Z ddlm Z ddl m Z ddl mZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZddlmZddlmZdZ ddZ!ddZ"ddZ#Gddde$Z%Gddde$Z&ej'fdd Z(d!d"Z)d#d$Z*d%d&Z+d'd(Z,d)d*Z-d+d,Z.d-d.Z/dS)/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)coredcCsTg}t|D]}dd|D}t|}|ddt||Dqtt|}|S)z)Aggregate gradients using nccl allreduce.cSg|]\}}|qSr.0g_rrp/home/www/facesmatcher.com/pyenv/lib/python3.10/site-packages/tensorflow/python/distribute/cross_device_utils.py -z2aggregate_gradients_using_nccl..cSg|] \}\}}||fqSrrrrrvrrrr0)ziprZall_sumappendlist) replica_gradsZagg_all_g_and_vZsingle_g_and_v single_grads agg_gradsrrraggregate_gradients_using_nccl)s   r*c CsFg}t|}|d}tt|D] \}}||}|||}||kr(d} |} n|} d} || | |} t||t| dd\} } Wdn1sNwY|| | |}t||t|dd\}} Wdn1suwYt||t| |gdd\\}} } Wdn1swYt|| t|}Wdn1swYt|| t|}Wdn1swYg}tt|D]/}t||||k||kkr|}n|}| t|Wdn 1swYq| ddt||Dqt t|}|S)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. rFNcSr rrr!rrrrr#z?aggregate_gradients_using_hierarchical_copy..) len enumerater$r device$aggregate_single_gradient_using_copyridentityranger%r&)Z avail_devicesr'r)Z num_devices group_sizeir(Zgroup_0_main_deviceZgroup_1_main_deviceZ group_0_beginZ group_1_beginZgroup_0_device_gradsZgroup_0_agg_gradsrZgroup_1_device_gradsZgroup_1_agg_gradsZagg_total_gradsZgroup_0_agg_grads_bcastZgroup_1_agg_grads_bcastZagg_grads_bcastjZsrc_device_gradrrr+aggregate_gradients_using_hierarchical_copy7sh       r5cCszdd|D}t|}|rt|dkrt|dt|}|dd}|r7ttt|}||f|fS||fdfS)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. cSrrrrrrrrrz8aggregate_single_gradient_using_copy..g?rN)radd_nr,rmultiplyZ logical_notZ reduce_all is_finite)Z grad_and_varsZuse_meanZ check_inf_nanZgradsZgradr"Zhas_nan_or_infrrrr/s    r/c@s:eZdZdZdddZddZddZd d Zd d Zd S)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. r6cCs ||_i|_t|_i|_dS)zaInitializes the object. Args: group_key_start: the starting integer of group key. N) _group_key_instance_key_table threadingLock_lock _known_groups)selfZgroup_key_startrrr__init__s  zCollectiveKeys.__init__cCsX|jd|}||jvr|||j|<|j|WdS1s%wYdS)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)r?joinr@_get_new_group_key)rAdevicesZ devices_keyrrr get_group_keys   $zCollectiveKeys.get_group_keycCs:|j}|jd7_i|j|<|D] }t|j||<q|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. r6)r;r<INSTANCE_KEY_START_NUMBER)rArFZnew_keyr.rrrrEs  z!CollectiveKeys._get_new_group_keycCs|j7|j|d}|durtd|d||vr%td|d|||}||d7<|WdS1s=wYdS)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 r6)r?r<get ValueError)rA group_keyr.groupr"rrrget_instance_keys$zCollectiveKeys.get_instance_keycCs"t}|j|_t|j||_|SN)r:r;copydeepcopyr<)rAmemoZcopiedrrr __deepcopy__ szCollectiveKeys.__deepcopy__N)r6) __name__ __module__ __qualname____doc__rBrGrErMrRrrrrr:s   r:c @sNeZdZdZdZdZdedededede j f dd Z d e e jejffd d Zd dZddZddZddZddZ  d'de jd ee e jejfdee j de jfddZde jdee j de jfddZ d(deee jdee j de jfdd Z d(de jd!e jdee j de jfd"d#Z d(d$ejdee j dejfd%d&Z dS))CollectiveReplicaLauncherz"Launch collectives on one replica.TrKr2collective_keysr.optionsc Cs||_||_||_||_||_|rMt,t|t d|_ Wdn1s.wYWddSWddS1sFwYdSd|_ dS)Ng) r; _group_size_collective_keys_device_options_use_ordering_tokenr Z init_scoper.rZResourceVariable_ordering_token)rArKr2rXr.rYrrrrBsP z"CollectiveReplicaLauncher.__init__ control_inputcCs$|dur|st|gStSrN)r^r control_dependenciesZNullContextmanager)rAr`rrr_control_input&s z(CollectiveReplicaLauncher._control_inputcCtsdStjSNF)r #executing_eagerly_outside_functionsrW_prefer_unique_instance_keyrArrr_use_unique_instance_key,sz2CollectiveReplicaLauncher._use_unique_instance_keycCrcrd)r rerW_prefer_ordering_tokenrgrrrr^1sz-CollectiveReplicaLauncher._use_ordering_tokencCs|rht}t|ddr|j}t|ddsts@|jr@|| |j t gt jWdS1s9wYdS|j|j|j}tdtj|t jdWdS1sawYdS|j|j|jS)zReturns the next instance key.Zis_control_flow_graphFNzCPU:0)dtype)rhr Zget_default_graphgetattrZ outer_graphr executing_eagerlyZbuilding_functionZ as_defaultZcapture_call_time_value_next_instance_keyr Z TensorSpecr int32r[rMr;r\r.Zconvert_to_tensor)rAgraph instance_keyrrrrm8s(   $  $ z,CollectiveReplicaLauncher._next_instance_keycCs|r|jjSdSrN)r^r_handlergrrr_get_ordering_tokenWsz-CollectiveReplicaLauncher._get_ordering_tokencCs|S)z0Whether this launcher can order NCCL operations.)r^rgrrrcan_order_nccl[sz(CollectiveReplicaLauncher.can_order_ncclN input_tensorreturnc Cs|}|j|}|}t|j6|| tj ||j |j ||j j |j|dWdWdS1s=wYWddS1sMwYdS)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. Zcommunication_hinttimeoutordering_tokenN)rmr]mergerrr r.r\rbrZ all_reduce_v2rZr;implementationvaluetimeout_seconds)rArtr`rYrprxrrr all_reduce_s  Rz$CollectiveReplicaLauncher.all_reducec Csn|}|j|}|}t|jtj||j |j ||j j |j |dWdS1s0wYdS)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. rvN)rmr]ryrrr r.r\rZ all_gather_v2rZr;rzr{r|)rArtrYrprxrrr _all_gather~s $z%CollectiveReplicaLauncher._all_gatherinput_tensor_packsc Cs |j|}g}|D]x}tr |D] }|||d|qq t|jTdd|D}dd|D}|j t j j krC|rC|d}nd}|t j|dd||} dd|D} t j| | dd} t|| D] \} } |t | | qeWdn1s}wYq |S) 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. NcSsg|] }t|dgqS))rreshapertrrrrsz>CollectiveReplicaLauncher.batch_all_reduce..cSg|]}t|qSr)rshaperrrrrrraxiscSrr)rZ reduce_prod)rsrrrrr)r]ryr rlr%r}r r.r\rzrCommunicationImplementationNCCLrconcatsplitr$r)rArrYZoutputspackrtZ flat_tensorsZshapesr`Zreduced num_elementsZ flat_outputsrZ flat_outputrrrbatch_all_reduces8  z*CollectiveReplicaLauncher.batch_all_reducerc Cstrtdt|jtt|gtj |gt |t |dt |fdd}tj ||d}|tjt|dd|}|dddf}t |}t||} || |} g} t |jD]} | |} | | | | || qdt | d}tj t d|ddgt |dt |fdd}tj ||dWdWdS1swYWddS1swYdS)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.r6rr)permN)r rl RuntimeErrorr r.r\rarr0rrr1rankZ transposer~Zexpand_dims_v2shape_v2 reduce_max _pad_utilrZr%)rArtrrYZperm_preZinput_tensor_tZgathered_shapeZ first_dims full_axis_dimpadded_input_tensorZgather_padded_out_tensor split_tensorsr3 start_posZ out_tensor_tZ perm_afterrrr all_gathersF       Rz$CollectiveReplicaLauncher.all_gather input_slicesc sjtjZdttjtt j gtj fdt j ffdd tj}|dtjdtt j dtj ffdd ttttfd d fd d Wd S1siwYd S) 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_fnrucsn|j}jtjjkr|g}ng}t||j}Wdn1s)wYtj ||j dS)z/Use all_gather_fn to aggregate `IndexedSlices`.N)rindices dense_shape) rrzrrrr rarr IndexedSlicesr)rZ all_valuescontrolZ all_indices)rrYrrall_gather_indexed_slices s  zVCollectiveReplicaLauncher.all_reduce_indexed_slices..all_gather_indexed_slicesrtrYcsbt}t||}||}g}tjD]}||}|||||qt|dS)z4all_gather tensors of different sizes using padding.r) rrrr~r1rZr%rr)rtrY max_lengthZ padded_tensorZall_padded_tensorsrr3r) all_lengthsrArrall_gather_with_padding6s     zTCollectiveReplicaLauncher.all_reduce_indexed_slices..all_gather_with_paddingcs jSrN)r~r)rrArrHs zECollectiveReplicaLauncher.all_reduce_indexed_slices..csSrNrr)rrrrrIsN)r]ryr r.r\rr TensorLikerrOptionsTensorr rrrrr~rrequalrZ reduce_min)rArrYlengthr)rrrrrYrArall_reduce_indexed_slicess6     $z3CollectiveReplicaLauncher.all_reduce_indexed_slices)NNrN)!rSrTrUrVrfriintr:strrrrBrrrr Z Operationrbrhr^rmrrrsrrr}r~rrrr rrrrrrrWsz      2 BrWcCs$tdd|Drt|S||S)zGAggregate tensors using `accumulation_fn` and IndexedSlices via concat.cs|] }t|tjVqdSrN isinstancer rrr"rrr Nsz6aggregate_tensors_or_indexed_slices..)anyrZAggregateIndexedSlicesGradients)rZaccumulation_fnrrr#aggregate_tensors_or_indexed_slicesLs rcCs6t|tjrt|}t|j||j|jS||SrN)rr rrZFlattenNestedIndexedSlicesrrr)r{nrrr%divide_by_n_tensors_or_indexed_slicesTs  rcCst|>t|tjr.t|j}t|j}|j dur$t|j }nd}t|||}n t|}Wd|SWd|S1sFwY|S)z-Copies a tensor or IndexedSlices to a device.N) r r.rr rrr0rrr)r{r.Z copied_valuesZcopied_indicesZ copied_shaperesultrrr'copy_tensor_or_indexed_slices_to_device]s$           rcCs4t|tjrdSt|tjrtdd|jDSdS)NTcsrrNrrrrrrrs  z$is_indexed_slices..F)rr r value_libZDistributedValuesallr)r{rrris_indexed_slicesns  rcCs`g}g}g}g}t|D]\}}t|r||||q ||||q ||||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%)rZ dense_valuesZ dense_indicesZ sparse_valuesZsparse_indicesr3r"rrrsplit_by_sparsityws      rcCsld}|D] }|t|d7}qdg|}|D]}|r3|dr3t|D]\}}||dus.J|||<q"q|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$)Zvalues_and_indices_listrZvalues_and_indicesrr"r3rrr stitch_valuess    rcCs|dkr|gSg}d}|D]4}|j}|dur#td||gS||jj}|r/||kr6|gd}|d|||7}q |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. rNzAnot packing values due to the unknown or inconsistent shape of %sr)rrloggingwarningrjsizer%)Z input_tensorsZbytes_per_packZpacksZlast_pack_sizer{rrrrr group_by_sizes&      rcCs\|t|d}t|}d|gg}tj|tj|ddftjdgdd}t||}|S)z?Pad the `input_tensor`'s first dimension to be `full_axis_dim`.rr6r+)rrjr)rrrrZzerosr rnpad)rtrZmissing_axis_dimZ tensor_rankZ paddings_axisZpaddingsrrrrrs   r)0rVrOr=typingrrrrZtensorflow.python.distributerrrZtensorflow.python.eagerrr Ztensorflow.python.frameworkr r r r Ztensorflow.python.opsrrrrrrZtensorflow.python.platformrrZtensorflow.python.typesrrHr*r5r/objectr:rWr7rrrrrrrrrrrrsH                R$e<   (