o i 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_rr^/var/www/myenv/lib/python3.10/site-packages/tensorflow/python/distribute/cross_device_utils.py -z2aggregate_gradients_using_nccl..cSg|] \}\}}||fqSrrrrrvrrrr0)ziprall_sumappendlist) replica_gradsagg_all_g_and_vsingle_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') avail_devicesr(r, num_devices group_sizeir+group_0_main_devicegroup_1_main_device group_0_begin group_1_begingroup_0_device_gradsgroup_0_agg_gradsrgroup_1_device_gradsgroup_1_agg_gradsagg_total_gradsgroup_0_agg_grads_bcastgroup_1_agg_grads_bcastagg_grads_bcastjsrc_device_gradrrr+aggregate_gradients_using_hierarchical_copy7sh       rGcCszdd|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/rmultiply logical_not reduce_all is_finite) grad_and_varsuse_mean check_inf_nangradsgradr"has_nan_or_infrrrr2s    r2c@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. rHcCs ||_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)selfgroup_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)rYjoinrZ_get_new_group_key)r[devices 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. rH)rUrVINSTANCE_KEY_START_NUMBER)r[ranew_keyr1rrrr`s  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 rH)rYrVget ValueError)r[ group_keyr1groupr"rrrget_instance_keys$zCollectiveKeys.get_instance_keycCs"t}|j|_t|j||_|SN)rTrUcopydeepcopyrV)r[memocopiedrrr __deepcopy__ szCollectiveKeys.__deepcopy__N)rH) __name__ __module__ __qualname____doc__r]rcr`rjrprrrrrTs   rTc @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.Trhr7collective_keysr1optionsc Cs||_||_||_||_||_|rMt,t|t d|_ Wdn1s.wYWddSWddS1sFwYdSd|_ dS)Ng) rU _group_size_collective_keys_device_options_use_ordering_tokenr init_scoper1rResourceVariable_ordering_token)r[rhr7rvr1rwrrrr]sP z"CollectiveReplicaLauncher.__init__ control_inputcCs$|dur|st|gStSrk)r|r control_dependenciesNullContextmanager)r[rrrr_control_input&s z(CollectiveReplicaLauncher._control_inputcCtsdStjSNF)r #executing_eagerly_outside_functionsru_prefer_unique_instance_keyr[rrr_use_unique_instance_key,sz2CollectiveReplicaLauncher._use_unique_instance_keycCrr)r rru_prefer_ordering_tokenrrrrr|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.is_control_flow_graphFNzCPU:0)dtype)rr get_default_graphgetattr outer_graphr executing_eagerlybuilding_function as_defaultcapture_call_time_value_next_instance_keyr TensorSpecr int32ryrjrUrzr1convert_to_tensor)r[graph instance_keyrrrr8s(   $  $ z,CollectiveReplicaLauncher._next_instance_keycCs|r|jjSdSrk)r|rhandlerrrr_get_ordering_tokenWsz-CollectiveReplicaLauncher._get_ordering_tokencCs|S)z0Whether this launcher can order NCCL operations.)r|rrrrcan_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. communication_hinttimeoutordering_tokenN)rr{mergerr r1rzrr all_reduce_v2rxrUimplementationvaluetimeout_seconds)r[rrrwrrrrr 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. rN)rr{rrr r1rzr all_gather_v2rxrUrrr)r[rrwrrrrr _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)r reduce_prod)rsrrrrr)r{rr rr&rr r1rzrrCommunicationImplementationNCCLrconcatsplitr$r)r[rrwoutputspackr flat_tensorsshapesrreduced num_elements flat_outputsr 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.rHrr)permN)r r RuntimeErrorr r1rzrrr3rrr4rank transposerexpand_dims_v2shape_v2 reduce_max _pad_utilrxr&)r[rrrwperm_preinput_tensor_tgathered_shape first_dims full_axis_dimpadded_input_tensorgather_padded_out_tensor split_tensorsr8 start_pos out_tensor_t 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_fnrcsn|j}jtjjkr|g}ng}t||j}Wdn1s)wYtj ||j dS)z/Use all_gather_fn to aggregate `IndexedSlices`.N)rindices dense_shape) rrrrrr rrr IndexedSlicesr)r all_valuescontrol all_indices)rrwrrall_gather_indexed_slices s  zVCollectiveReplicaLauncher.all_reduce_indexed_slices..all_gather_indexed_slicesrrwcsbt}t||}||}g}tjD]}||}|||||qt|dS)z4all_gather tensors of different sizes using padding.r) rrrrr4rxr&rr)rrw max_length padded_tensorall_padded_tensorsrr8r) all_lengthsr[rrall_gather_with_padding6s     zTCollectiveReplicaLauncher.all_reduce_indexed_slices..all_gather_with_paddingcs jSrk)rr)rr[rrHs zECollectiveReplicaLauncher.all_reduce_indexed_slices..csSrkrr)rrrrrIsN)r{rr r1rzrr TensorLikerrOptionsTensorr rrrrrrrequalr reduce_min)r[rrwlengthr)rrrrrwr[rall_reduce_indexed_slicess6     $z3CollectiveReplicaLauncher.all_reduce_indexed_slices)NNrk)!rqrrrsrtrrintrTstrrrr]rrrr Operationrrr|rrrrrrrrrrr rrrrrrrusz      2 BrucCs$tdd|Drt|S||S)zGAggregate tensors using `accumulation_fn` and IndexedSlices via concat.cs|] }t|tjVqdSrk isinstancer rrr"rrr Nsz6aggregate_tensors_or_indexed_slices..)anyrAggregateIndexedSlicesGradients)raccumulation_fnrrr#aggregate_tensors_or_indexed_slicesLs rcCs6t|tjrt|}t|j||j|jS||Srk)rr rrFlattenNestedIndexedSlicesrrr)rnrrr%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 r1rr rrr3rrr)rr1 copied_valuescopied_indices copied_shaperesultrrr'copy_tensor_or_indexed_slices_to_device]s$           rcCs4t|tjrdSt|tjrtdd|jDSdS)NTcsrrkrrrrrrrs  z$is_indexed_slices..F)rr r value_libDistributedValuesallr)rrrris_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`. )r0rr&)r dense_values dense_indices sparse_valuessparse_indicesr8r"rrrsplit_by_sparsityws      r cCsld}|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$)values_and_indices_listrvalues_and_indicesrr"r8rrr stitch_valuess    r cCs|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)rrloggingwarningrsizer&) input_tensorsbytes_per_packpackslast_pack_sizerrrrrr 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`.rrHr.)rrr)rrrrzerosr rpad)rrmissing_axis_dim tensor_rank paddings_axispaddingsrrrrrs   r)0rtrlrWtypingrrrrtensorflow.python.distributerrrtensorflow.python.eagerrr tensorflow.python.frameworkr r r r tensorflow.python.opsrrrrrrtensorflow.python.platformrrtensorflow.python.typesrrdr-rGr2objectrTrurIrrrrr r rrrrrrsH                R$e<   (