o ?eWN@sJdZddlZddlZddlZddlZddlZddlZddlmZddl Z ddl m Z ddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZddlm Z ddl!m"Z#dd l$mZ%ddl&m'Z'ddl&m(Z(ddl&m)Z)ddl*m+Z+ddl*m,Z-ddl*m.Z.ddl*m/Z/ddl*m0Z0ddl*m1Z1ddl*m2Z2ddl*m3Z3ddl*m4Z4dd l5m6Z6dd!l5m7Z7dd"l5m8Z8dd#l5m9Z9dd$l5m:Z;dd%lm?Z?dd'l@mAZBdd(l@mCZCdd)l@mDZDdd*l@mEZEdd+lFmGZGdd,lHmIZIdd-lHmJZJdd.lHmKZKdd/lLmMZMd0ZNd1ZOd2d3ZPejQd4d5ZRd6d7ZSd8d9ZTeMd:gd;Gdgd;eIWd>Gd?d@d@ejUZXeMd>gd;GdAdBdBejYZZGdCdDdDej[Z\dEdFZ]e/j^e/j_e/j`e/jae/jbe/jcfZdGdGdHdHejeZfdIdJZgdS)Kz TPU Strategy.N)logging)ag_ctx)api) xla_sharding)cross_device_ops) device_util)distribute_lib)distribute_utils) input_lib) input_util) numpy_dataset) reduce_util)tpu_replicated_variable)tpu_util) tpu_values)values)tpu_cluster_resolver)context) def_function)function) constant_op)device) device_spec)dtypes)indexed_slices)ops) sparse_tensor) tensor_shape) tensor_util) array_ops)control_flow_ops)math_ops)resource_variable_ops) variables) ragged_tensor) save_context)device_assignment)tpu)tpu_hardware_feature) training_loop)tpu_ops) deprecation)nest) tf_inspect) tf_exportFcCsto to t S)z8Whether to batch variable initialization in tf.function.)/_EXPERIMENTAL_TPU_BATCH_VARIABLE_INITIALIZATIONrexecuting_eagerlyr%Zin_save_contextr2r2j/home/www/facesmatcher.com/pyenv/lib/python3.10/site-packages/tensorflow/python/distribute/tpu_strategy.py$enable_batch_variable_initializationPs r4ccsHtr dVdSt dVWddS1swYdSN)rZ#executing_eagerly_outside_functionsZ init_scoper2r2r2r3maybe_init_scopeYs   "r6cCsNtrt|tjs!t|tjs#t|rt|jtjs%t ddSdSdSdS)z/Validate the function passed into strategy.run.zTPUStrategy.run(fn, ...) does not support pure eager execution. please make sure the function passed into `strategy.run` is a `tf.function` or `strategy.run` is called inside a `tf.function` if eager behavior is enabled.N) rr1 isinstancerFunctionrZConcreteFunctioncallable__call__NotImplementedError)fnr2r2r3validate_run_functionbs   r=c sddi}i}|rfdd|D}|r"fdd|D}g}d}tt|jD]W\}}|dkr>|jdkr>q0|jtjj krL| |jq0|jtjj krV|}q0|jtjj kr|sjt fd d |Drtd t|d tfd d |Dd|||fSq0g} d} t|D]K\}} | r|dur||krtdt||krtd| |||<d} q|dur||kr| rtd| | qt||krtd| |||<q|rtj|fi|| |fS|||fS)a|Inspects arguments to partially apply any DistributedVariable. This avoids an automatic cast of the current variable value to tensor. Note that a variable may be captured implicitly with Python scope instead of passing it to run(), but supporting run() keeps behavior consistent with MirroredStrategy. Since positional arguments must be applied from left to right, this function does some tricky function inspection to move variable positional arguments into kwargs. As a result of this, we can't support passing Variables as *args, nor as args to functions which combine both explicit positional arguments and *args. Args: fn: The function to run, as passed to run(). args: Positional arguments to fn, as passed to run(). kwargs: Keyword arguments to fn, as passed to run(). Returns: A tuple of the function (possibly wrapped), args, kwargs (both possibly filtered, with members of args possibly moved to kwargs). If no variables are found, this function is a noop. Raises: ValueError: If the function signature makes unsupported use of *args, or if too many arguments are passed. cSst|}|ot|dtjSNr)r,flattenr7rDistributedVariable)xZflatr2r2r3is_distributed_vars z:_maybe_partial_apply_variables..is_distributed_varcsi|] \}}|r||qSr2r2.0kvrBr2r3 sz2_maybe_partial_apply_variables..csi|] \}}|s||qSr2r2rCrGr2r3rHs  Nrselfc3|]}|VqdSr5r2rDarGr2r3 z1_maybe_partial_apply_variables..zWMixing Variables and positional-only parameters not supported by TPUStrategy. Received z& DistributedVariables in **kwargs and c3rJr5r2rKrGr2r3rMrNz" in *args, expected zero for both.FzTPUStrategy.run() cannot handle Variables passed to *args. Either name the function argument, or capture the Variable implicitly.zBToo many positional arguments passed to call to TPUStrategy.run().TzTPUStrategy.run() cannot handle both Variables and a mix of positional args and *args. Either remove the *args, or capture the Variable implicitly.)items enumerater- signature parametersrnamekind ParameterPOSITIONAL_OR_KEYWORDappendVAR_POSITIONALPOSITIONAL_ONLYany ValueErrorlensum functoolspartial) r<argskwargsZ var_kwargsZ nonvar_kwargsZpositional_argsZindex_of_star_argsip star_argsZhave_seen_var_argrLr2rGr3_maybe_partial_apply_variables}st       rezdistribute.TPUStrategy)v1csVeZdZdZ   dfdd ZdddZed d Zd d Zd dZ ddZ Z S) TPUStrategyV2aSynchronous training on TPUs and TPU Pods. To construct a TPUStrategy object, you need to run the initialization code as below: >>> resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='') >>> tf.config.experimental_connect_to_cluster(resolver) >>> tf.tpu.experimental.initialize_tpu_system(resolver) >>> strategy = tf.distribute.TPUStrategy(resolver) While using distribution strategies, the variables created within the strategy's scope will be replicated across all the replicas and can be kept in sync using all-reduce algorithms. To run TF2 programs on TPUs, you can either use `.compile` and `.fit` APIs in `tf.keras` with TPUStrategy, or write your own customized training loop by calling `strategy.run` directly. Note that TPUStrategy doesn't support pure eager execution, so please make sure the function passed into `strategy.run` is a `tf.function` or `strategy.run` is called inside a `tf.function` if eager behavior is enabled. See more details in https://www.tensorflow.org/guide/tpu. `distribute_datasets_from_function` and `experimental_distribute_dataset` APIs can be used to distribute the dataset across the TPU workers when writing your own training loop. If you are using `fit` and `compile` methods available in `tf.keras.Model`, then Keras will handle the distribution for you. An example of writing customized training loop on TPUs: >>> with strategy.scope(): ... model = tf.keras.Sequential([ ... tf.keras.layers.Dense(2, input_shape=(5,)), ... ]) ... optimizer = tf.keras.optimizers.SGD(learning_rate=0.1) >>> def dataset_fn(ctx): ... x = np.random.random((2, 5)).astype(np.float32) ... y = np.random.randint(2, size=(2, 1)) ... dataset = tf.data.Dataset.from_tensor_slices((x, y)) ... return dataset.repeat().batch(1, drop_remainder=True) >>> dist_dataset = strategy.distribute_datasets_from_function( ... dataset_fn) >>> iterator = iter(dist_dataset) >>> @tf.function() ... def train_step(iterator): ... ... def step_fn(inputs): ... features, labels = inputs ... with tf.GradientTape() as tape: ... logits = model(features, training=True) ... loss = tf.keras.losses.sparse_categorical_crossentropy( ... labels, logits) ... ... grads = tape.gradient(loss, model.trainable_variables) ... optimizer.apply_gradients(zip(grads, model.trainable_variables)) ... ... strategy.run(step_fn, args=(next(iterator),)) >>> train_step(iterator) For the advanced use cases like model parallelism, you can set `experimental_device_assignment` argument when creating TPUStrategy to specify number of replicas and number of logical devices. Below is an example to initialize TPU system with 2 logical devices and 1 replica. >>> resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='') >>> tf.config.experimental_connect_to_cluster(resolver) >>> topology = tf.tpu.experimental.initialize_tpu_system(resolver) >>> device_assignment = tf.tpu.experimental.DeviceAssignment.build( ... topology, ... computation_shape=[1, 1, 1, 2], ... num_replicas=1) >>> strategy = tf.distribute.TPUStrategy( ... resolver, experimental_device_assignment=device_assignment) Then you can run a `tf.add` operation only on logical device 0. >>> @tf.function() ... def step_fn(inputs): ... features, _ = inputs ... output = tf.add(features, features) ... ... # Add operation will be executed on logical device 0. ... output = strategy.experimental_assign_to_logical_device(output, 0) ... return output >>> dist_dataset = strategy.distribute_datasets_from_function( ... dataset_fn) >>> iterator = iter(dist_dataset) >>> strategy.run(step_fn, args=(next(iterator),)) `experimental_spmd_xla_partitioning` enables the experimental XLA SPMD feature for model parallelism. This flag can reduce the compilation time and HBM requirements. When running in this mode, every input tensor must either be partitioned (via `strategy.experimental_split_to_logical_devices`) or fully replicated (via `strategy.experimental_replicate_to_logical_devices`) to all logical devices. And calling `strategy.experimental_assign_to_logical_device` will result in a ValueError in this mode. NFc sjtt|t|||||dudtjddtjd|j j tjd|j j d|_ dS)aKSynchronous training in TPU donuts or Pods. Args: tpu_cluster_resolver: A `tf.distribute.cluster_resolver.TPUClusterResolver` instance, which provides information about the TPU cluster. If None, it will assume running on a local TPU worker. experimental_device_assignment: Optional `tf.tpu.experimental.DeviceAssignment` to specify the placement of replicas on the TPU cluster. experimental_spmd_xla_partitioning: If True, enable the SPMD (Single Program Multiple Data) mode in XLA compiler. This flag only affects the performance of XLA compilation and the HBM requirement of the compiled TPU program. Ceveat: if this flag is True, calling `tf.distribute.TPUStrategy.experimental_assign_to_logical_device` will result in a ValueError. N)r&use_spmd_for_xla_partitioningenable_data_reorderV2 TPUStrategy num_workersnum_replicas_per_workerT) superrg__init__ TPUExtendedrdistribution_strategy_gaugeget_cellset#distribution_strategy_replica_gaugeextended num_hostsnum_replicas_per_host%_enable_packed_variable_in_eager_mode)rIrZexperimental_device_assignmentZ"experimental_spmd_xla_partitioning __class__r2r3ro[s&     zTPUStrategyV2.__init__r2cCHt|t|||\}}}t|t}|pt}|j ||||S)a3Run the computation defined by `fn` on each TPU replica. Executes ops specified by `fn` on each replica. If `args` or `kwargs` have `tf.distribute.DistributedValues`, such as those produced by a `tf.distribute.DistributedDataset` from `tf.distribute.Strategy.experimental_distribute_dataset` or `tf.distribute.Strategy.distribute_datasets_from_function`, when `fn` is executed on a particular replica, it will be executed with the component of `tf.distribute.DistributedValues` that correspond to that replica. `fn` may call `tf.distribute.get_replica_context()` to access members such as `all_reduce`. All arguments in `args` or `kwargs` should either be nest of tensors or `tf.distribute.DistributedValues` containing tensors or composite tensors. Example usage: >>> resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='') >>> tf.config.experimental_connect_to_cluster(resolver) >>> tf.tpu.experimental.initialize_tpu_system(resolver) >>> strategy = tf.distribute.TPUStrategy(resolver) >>> @tf.function ... def run(): ... def value_fn(value_context): ... return value_context.num_replicas_in_sync ... distributed_values = ( ... strategy.experimental_distribute_values_from_function(value_fn)) ... def replica_fn(input): ... return input * 2 ... return strategy.run(replica_fn, args=(distributed_values,)) >>> result = run() Args: fn: The function to run. The output must be a `tf.nest` of `Tensor`s. args: (Optional) Positional arguments to `fn`. kwargs: (Optional) Keyword arguments to `fn`. options: (Optional) An instance of `tf.distribute.RunOptions` specifying the options to run `fn`. Returns: Merged return value of `fn` across replicas. The structure of the return value is the same as the return value from `fn`. Each element in the structure can either be `tf.distribute.DistributedValues`, `Tensor` objects, or `Tensor`s (for example, if running on a single replica). r=re autographZ tf_convert autograph_ctxZcontrol_status_ctxrZ RunOptionsrutpu_runrIr<r`raoptionsr2r2r3runs 0 zTPUStrategyV2.runcC|jjS)a[Returns the cluster resolver associated with this strategy. `tf.distribute.TPUStrategy` provides the associated `tf.distribute.cluster_resolver.ClusterResolver`. If the user provides one in `__init__`, that instance is returned; if the user does not, a default `tf.distribute.cluster_resolver.TPUClusterResolver` is provided. ru_tpu_cluster_resolverrIr2r2r3cluster_resolvers zTPUStrategyV2.cluster_resolvercCsN|jjrtd|jjjd}|dks||krtd||tj||ddS)ayAdds annotation that `tensor` will be assigned to a logical device. This adds an annotation to `tensor` specifying that operations on `tensor` will be invoked on logical core device id `logical_device_id`. When model parallelism is used, the default behavior is that all ops are placed on zero-th logical device. ```python # Initializing TPU system with 2 logical devices and 4 replicas. resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='') tf.config.experimental_connect_to_cluster(resolver) topology = tf.tpu.experimental.initialize_tpu_system(resolver) device_assignment = tf.tpu.experimental.DeviceAssignment.build( topology, computation_shape=[1, 1, 1, 2], num_replicas=4) strategy = tf.distribute.TPUStrategy( resolver, experimental_device_assignment=device_assignment) iterator = iter(inputs) @tf.function() def step_fn(inputs): output = tf.add(inputs, inputs) # Add operation will be executed on logical device 0. output = strategy.experimental_assign_to_logical_device(output, 0) return output strategy.run(step_fn, args=(next(iterator),)) ``` Args: tensor: Input tensor to annotate. logical_device_id: Id of the logical core to which the tensor will be assigned. Raises: ValueError: The logical device id presented is not consistent with total number of partitions specified by the device assignment or the TPUStrategy is constructed with `experimental_spmd_xla_partitioning=True`. Returns: Annotated tensor with identical value as `tensor`. zCannot assign a tensor to a logical device in SPMD mode. To disable SPMD, Please construct the TPUStrategy with `experimental_spmd_xla_partitioning=False`rz`logical_core_id` to assign must be lower then total number of logical devices per replica. Received logical device id {} but there are only total of {} logical devices in replica.TZuse_sharding_op)ru_use_spmd_for_xla_partitioningr[ _tpu_devicesshapeformatrZ assign_device)rItensorlogical_device_idnum_logical_devices_per_replicar2r2r3%experimental_assign_to_logical_devices .z3TPUStrategyV2.experimental_assign_to_logical_devicec Cs|jjjd}t|}|j}t|}|t|kr#td|t|t|D]\}}|dur0q'||} || dkrCtd|| |q'||krQtd|||t | |} t j || ddS) a Adds annotation that `tensor` will be split across logical devices. This adds an annotation to tensor `tensor` specifying that operations on `tensor` will be split among multiple logical devices. Tensor `tensor` will be split across dimensions specified by `partition_dimensions`. The dimensions of `tensor` must be divisible by corresponding value in `partition_dimensions`. For example, for system with 8 logical devices, if `tensor` is an image tensor with shape (batch_size, width, height, channel) and `partition_dimensions` is [1, 2, 4, 1], then `tensor` will be split 2 in width dimension and 4 way in height dimension and the split tensor values will be fed into 8 logical devices. ```python # Initializing TPU system with 8 logical devices and 1 replica. resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='') tf.config.experimental_connect_to_cluster(resolver) topology = tf.tpu.experimental.initialize_tpu_system(resolver) device_assignment = tf.tpu.experimental.DeviceAssignment.build( topology, computation_shape=[1, 2, 2, 2], num_replicas=1) # Construct the TPUStrategy. Since we are going to split the image across # logical devices, here we set `experimental_spmd_xla_partitioning=True` # so that the partitioning can be compiled in SPMD mode, which usually # results in faster compilation and smaller HBM requirement if the size of # input and activation tensors are much bigger than that of the model # parameters. Note that this flag is suggested but not a hard requirement # for `experimental_split_to_logical_devices`. strategy = tf.distribute.TPUStrategy( resolver, experimental_device_assignment=device_assignment, experimental_spmd_xla_partitioning=True) iterator = iter(inputs) @tf.function() def step_fn(inputs): inputs = strategy.experimental_split_to_logical_devices( inputs, [1, 2, 4, 1]) # model() function will be executed on 8 logical devices with `inputs` # split 2 * 4 ways. output = model(inputs) return output strategy.run(step_fn, args=(next(iterator),)) ``` Args: tensor: Input tensor to annotate. partition_dimensions: An unnested list of integers with the size equal to rank of `tensor` specifying how `tensor` will be partitioned. The product of all elements in `partition_dimensions` must be equal to the total number of logical devices per replica. Raises: ValueError: 1) If the size of partition_dimensions does not equal to rank of `tensor` or 2) if product of elements of `partition_dimensions` does not match the number of logical devices per replica defined by the implementing DistributionStrategy's device specification or 3) if a known size of `tensor` is not divisible by corresponding value in `partition_dimensions`. Returns: Annotated tensor with identical value as `tensor`. rzvLength of `partition_dimensions` must equal to the rank of `tensor.shape` ({}). Received len(partition_dimensions)={}.NrzTensor shape at `partition_dimensions[{}]` must be divisible by corresponding value specified by `partition_dimensions` ({}). Received: {}.zThe product of `partition_dimensions` should be the same as the number of logical devices (={}). Received `partition_dimensions`={},and their product is {}.Tr) rurrnpprodr\r[rrPZarangereshapertile) rIrZpartition_dimensionsrZnum_partition_splitsZ input_shapeZ tensor_rankZ dim_indexZdim_sizeZ split_sizeZtile_assignmentr2r2r3%experimental_split_to_logical_devicessBC    z3TPUStrategyV2.experimental_split_to_logical_devicescCstj|ddS)aAdds annotation that `tensor` will be replicated to all logical devices. This adds an annotation to tensor `tensor` specifying that operations on `tensor` will be invoked on all logical devices. ```python # Initializing TPU system with 2 logical devices and 4 replicas. resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='') tf.config.experimental_connect_to_cluster(resolver) topology = tf.tpu.experimental.initialize_tpu_system(resolver) device_assignment = tf.tpu.experimental.DeviceAssignment.build( topology, computation_shape=[1, 1, 1, 2], num_replicas=4) strategy = tf.distribute.TPUStrategy( resolver, experimental_device_assignment=device_assignment) iterator = iter(inputs) @tf.function() def step_fn(inputs): images, labels = inputs images = strategy.experimental_split_to_logical_devices( inputs, [1, 2, 4, 1]) # model() function will be executed on 8 logical devices with `inputs` # split 2 * 4 ways. output = model(inputs) # For loss calculation, all logical devices share the same logits # and labels. labels = strategy.experimental_replicate_to_logical_devices(labels) output = strategy.experimental_replicate_to_logical_devices(output) loss = loss_fn(labels, output) return loss strategy.run(step_fn, args=(next(iterator),)) ``` Args: tensor: Input tensor to annotate. Returns: Annotated tensor with identical value as `tensor`. Tr)r replicate)rIrr2r2r3)experimental_replicate_to_logical_devicesls.z7TPUStrategyV2.experimental_replicate_to_logical_devices)NNFr2NN) __name__ __module__ __qualname____doc__rorpropertyrrrr __classcell__r2r2ryr3rgsf (:  ?ergz#distribute.experimental.TPUStrategycs<eZdZdZ  d fdd Zd ddZedd ZZS) rkaSynchronous training on TPUs and TPU Pods. To construct a TPUStrategy object, you need to run the initialization code as below: >>> resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='') >>> tf.config.experimental_connect_to_cluster(resolver) >>> tf.tpu.experimental.initialize_tpu_system(resolver) >>> strategy = tf.distribute.experimental.TPUStrategy(resolver) While using distribution strategies, the variables created within the strategy's scope will be replicated across all the replicas and can be kept in sync using all-reduce algorithms. To run TF2 programs on TPUs, you can either use `.compile` and `.fit` APIs in `tf.keras` with TPUStrategy, or write your own customized training loop by calling `strategy.run` directly. Note that TPUStrategy doesn't support pure eager execution, so please make sure the function passed into `strategy.run` is a `tf.function` or `strategy.run` is called inside a `tf.function` if eager behavior is enabled. Ncsrtdtt|t||||dudtjd dtj d |j j tj d |j j d|_dS) aPSynchronous training in TPU donuts or Pods. Args: tpu_cluster_resolver: A tf.distribute.cluster_resolver.TPUClusterResolver, which provides information about the TPU cluster. device_assignment: Optional `tf.tpu.experimental.DeviceAssignment` to specify the placement of replicas on the TPU cluster. z`tf.distribute.experimental.TPUStrategy` is deprecated, please use the non-experimental symbol `tf.distribute.TPUStrategy` instead.N)r&rirjrkrlrmT)rwarningrnrkrorprrqrrrsrtrurvrwrx)rIrr&ryr2r3ros*     zTPUStrategy.__init__r2cCr{)zSee base class.r|rr2r2r3rs  zTPUStrategy.runcCr)alReturns the cluster resolver associated with this strategy. `tf.distribute.experimental.TPUStrategy` provides the associated `tf.distribute.cluster_resolver.ClusterResolver`. If the user provides one in `__init__`, that instance is returned; if the user does not, a default `tf.distribute.cluster_resolver.TPUClusterResolver` is provided. rrr2r2r3rs zTPUStrategy.cluster_resolver)NNr) rrrrrorrrrr2r2ryr3rks $ rkcs>eZdZdZ   d fdd ZeddZd dd ZZS) TPUStrategyV1z)TPU distribution strategy implementation.Ncsbtt|t||||tjddtjd|j j tjd|j j d|_ dS)aInitializes the TPUStrategy object. Args: tpu_cluster_resolver: A tf.distribute.cluster_resolver.TPUClusterResolver, which provides information about the TPU cluster. steps_per_run: Number of steps to run on device before returning to the host. Note that this can have side-effects on performance, hooks, metrics, summaries etc. This parameter is only used when Distribution Strategy is used with estimator or keras. device_assignment: Optional `tf.tpu.experimental.DeviceAssignment` to specify the placement of replicas on the TPU cluster. Currently only supports the usecase of using a single core within a TPU cluster. ZV1rkrlrmTN) rnrrorprrqrrrsrtrurvrwrx)rIr steps_per_runr&ryr2r3ros    zTPUStrategyV1.__init__cCr)z0DEPRECATED: use .extended.steps_per_run instead.)Z _extendedrrr2r2r3rszTPUStrategyV1.steps_per_runr2cCr{)aRun `fn` on each replica, with the given arguments. Executes ops specified by `fn` on each replica. If `args` or `kwargs` have "per-replica" values, such as those produced by a "distributed `Dataset`", when `fn` is executed on a particular replica, it will be executed with the component of those "per-replica" values that correspond to that replica. `fn` may call `tf.distribute.get_replica_context()` to access members such as `all_reduce`. All arguments in `args` or `kwargs` should either be nest of tensors or per-replica objects containing tensors or composite tensors. Users can pass strategy specific options to `options` argument. An example to enable bucketizing dynamic shapes in `TPUStrategy.run` is: >>> resolver = tf.distribute.cluster_resolver.TPUClusterResolver(tpu='') >>> tf.config.experimental_connect_to_cluster(resolver) >>> tf.tpu.experimental.initialize_tpu_system(resolver) >>> strategy = tf.distribute.experimental.TPUStrategy(resolver) >>> options = tf.distribute.RunOptions( ... experimental_bucketizing_dynamic_shape=True) >>> dataset = tf.data.Dataset.range( ... strategy.num_replicas_in_sync, output_type=dtypes.float32).batch( ... strategy.num_replicas_in_sync, drop_remainder=True) >>> input_iterator = iter(strategy.experimental_distribute_dataset(dataset)) >>> @tf.function() ... def step_fn(inputs): ... output = tf.reduce_sum(inputs) ... return output >>> strategy.run(step_fn, args=(next(input_iterator),), options=options) Args: fn: The function to run. The output must be a `tf.nest` of `Tensor`s. args: (Optional) Positional arguments to `fn`. kwargs: (Optional) Keyword arguments to `fn`. options: (Optional) An instance of `tf.distribute.RunOptions` specifying the options to run `fn`. Returns: Merged return value of `fn` across replicas. The structure of the return value is the same as the return value from `fn`. Each element in the structure can either be "per-replica" `Tensor` objects or `Tensor`s (for example, if running on a single replica). r|rr2r2r3rs 3 zTPUStrategyV1.run)NNNr) rrrrrorrrrr2r2ryr3rs rcseZdZdZ     d\fdd ZddZdd Zd d Zej j fd d Z ddZ ddZ ddZddZddZddZ d]ddZddZejddZd d!Zd"d#Zed$d%Zd&d'Zd(d)Zd*d+Zd,d-Zd.d/Zd0d1Z d2d3Z!d4d5Z"ed6d7Z#ed8d9Z$ed:d;Z%edd?Z'ed@dAZ(edBdCZ)edDdEZ*edFdGZ+edHdIZ,dJdKZ-dLdMZ.    d^dNdOZ/dPdQZ0edRdSZ1d]dTdUZ2dVdWZ3dXdYZ4dZd[Z5Z6S)_rpzImplementation of TPUStrategy.NFc s tt|||durtd}|durd}t|_||_|j |_ ||_ dd|j j D}|durCt jdd|Dtd|_n8tj|dj}g} t|jD]} g} t|jD]} | t|j| | |dq\| | qSt j| td|_t|jdd|_t|_ t|_!|jdddfD]%} t| }|j "|g|j || |j!"|g|j!||q|r|#|jdddfnd|_$||_%d |_&d |_'dg|_(t)*rt+,t)j-| |_.||_/d |_0|jdd|j(d f}|D]}t)1|r d |_0dSqdS) NrcSsg|] }d|jvr|jqS)z device:TPU:)rSrDdr2r2r3 zs  z(TPUExtended.__init__..cSsg|]}|gqSr2r2rr2r2r3rsdtyper)replica logical_corejobTF)2rnrprotpu_cluster_resolver_libZTPUClusterResolverweakrefWeakKeyDictionary_tpu_function_cacherZget_tpu_system_metadata _tpu_metadata_device_assignmentdevicesrarrayobjectrrZ DeviceSpecV2 from_stringrrange num_replicasnum_cores_per_replicarWr canonicalize tpu_deviceZget_host_for_device _host_device collections OrderedDict_device_input_worker_devices_host_input_worker_devices setdefault_get_replica_order_replica_orderrZ_require_static_shapesZ(experimental_enable_get_next_as_optional_logical_device_stackrr1atexitregisterZ async_waitZ_use_var_policyr_using_custom_deviceZis_custom_device)rIZcontainer_strategyrrr&rhriZtpu_devices_flatZjob_name tpu_devices replica_idZreplica_devicesrr host_devicerrryr2r3ro^sz           zTPUExtended.__init__cCsVg}t|D]\}}tj|}||j|j|j|j|j f|fqddt |DS)aGet the replica order based on the tpu device order. For example, if the tpu_devices are: '/job:worker/replica:0/task:0/device:TPU:0', '/job:worker/replica:0/task:0/device:TPU:2', '/job:worker/replica:0/task:1/device:TPU:0', '/job:worker/replica:0/task:1/device:TPU:2', '/job:worker/replica:0/task:1/device:TPU:6', '/job:worker/replica:0/task:1/device:TPU:4', '/job:worker/replica:0/task:0/device:TPU:6', '/job:worker/replica:0/task:0/device:TPU:4', the returned replica order will be: [0, 1, 7, 6, 2, 3, 5, 4] This replica order will be used to reorder the data returned by the iterators, so that they can be placed on the same node as their computation graphs. Args: tpu_devices (List[str]): A list of tpu device names in the order of replicas. Returns: A list containing the order ids of corresponding TPU devices. cSsg|]\}}|qSr2r2)rD_rbr2r2r3rsz2TPUExtended._get_replica_order..) rP tf_deviceZ DeviceSpecrrWrrZ device_typetaskZ device_indexsorted)rIrZdevices_with_idsrbrspecr2r2r3rs  zTPUExtended._get_replica_ordercCst||dSr5)r Zvalidate_colocate)rIZcolocate_with_variabler2r2r3 _validate_colocate_with_variablesz,TPUExtended._validate_colocate_with_variablecCs,tt|j}tj||||jdS)z)Make iterators for each of the TPU hosts.)num_replicas_in_sync) r InputWorkerstuplerrO input_lib_v1ZDatasetIterator_container_strategy_num_replicas_in_sync)rIdataset input_workersr2r2r3_make_dataset_iterators z"TPUExtended._make_dataset_iteratorcCsXg}tt|j}|j}t|D]}|tj |||j dqt |||| S)NZnum_input_pipelinesZinput_pipeline_idr)r rrrrOrlrrWr InputContextrrZInputFunctionIteratorr)rIZinput_fnZreplication_modeinput_contextsrrlrbr2r2r3_make_input_fn_iterators"  z#TPUExtended._make_input_fn_iteratorcCst|t|j|Sr5)r Zone_host_numpy_dataset SingleDevicer)rIZ numpy_inputsessionr2r2r3 _experimental_make_numpy_datasets z,TPUExtended._experimental_make_numpy_datasetcCs2|r|jrtt|jStt|jSr5)experimental_fetch_to_devicer rrrrOr)rIrr2r2r3_get_input_workerss   zTPUExtended._get_input_workerscCsTt|tjr |j}t|}|D]\}}t|tjtj fr't d |t |qdS)Na>Found tensor {} with spec {}. TPUStrategy does not support distributed datasets with device prefetch when using sparse or ragged tensors. If you intend to use sparse or ragged tensors, please pass a tf.distribute.InputOptions object with experimental_fetch_to_device set to False to your dataset distribution function.) r7rZPerReplicaSpecZ_component_specsr,Z flatten_with_joined_string_pathsrZSparseTensorSpecr$ZRaggedTensorSpecr[rtype)rI element_specspecspathrr2r2r3 _check_spec s    zTPUExtended._check_speccCsX|r |jtjjkr td|dus|jr||jtj || || |j ||j dS)NzgInputReplicationMode.PER_REPLICA is only supported in `experimental_distribute_datasets_from_function`.)rr replica_order)experimental_replication_moderInputReplicationMode PER_REPLICAr;rrrr Zget_distributed_datasetrrrr)rIrrr2r2r3 _experimental_distribute_dataset/s  z,TPUExtended._experimental_distribute_datasetcCs|r |jtjjkr td||}g}|j}t|D]}|tj |||j dqt j |||| ||jd}|dus?|jrE||j|S)NzInputReplicationMode.PER_REPLICA is only supported in `experimental_distribute_datasets_from_function` of tf.distribute.MirroredStrategyr)rr)rrrrr;rrlrrWrrr Z&get_distributed_datasets_from_functionrrrrr)rIZ dataset_fnrrrrlrbZdistributed_datasetr2r2r3"_distribute_datasets_from_functionCs4    z.TPUExtended._distribute_datasets_from_functioncCs:g}t|jD]}||t||jqtj|ddS)NT)Z always_wrap)rrrWrZ ValueContextr regroup)rIZvalue_fnZper_replica_valuesrr2r2r3-_experimental_distribute_values_from_functionbsz9TPUExtended._experimental_distribute_values_from_functioncs|duri}t|}tfddt_fdd}t|t s,J|j }t j j dkrA|}nt|||}Wdn1sRwY`t|_t|t rdd|Dtj fddtDngtS) Ncs\|}tj}|r,t|gdd|DWdS1s%wYdS|S)zSingle step on the TPU device.cSsg|]}t|qSr2)ridentity)rDfr2r2r3r{szSTPUExtended._experimental_run_steps_on_iterator..run_fn..N)r,r?last_step_outputsrZcontrol_dependencies)inputsZ fn_resultZflat_last_step_outputs)ctxr<r2r3run_fnus   $z?TPUExtended._experimental_run_steps_on_iterator..run_fncsx~}g}tjD]fdd}|t||fq tj|jtj j dd}t |dt r:t |}|S)z%The rewritten step fn running on TPU.cs t|Sr5)r select_replicarArr2r3szUTPUExtended._experimental_run_steps_on_iterator..rewrite_fn..rh)r& xla_optionsr)Zget_nextrrrWr, map_structurer'rr XLAOptionsrr7listr?)r`Zper_replica_inputsreplicate_inputsrreplicate_outputs)multi_worker_iteratorrrIrr3 rewrite_fns&   zCTPUExtended._experimental_run_steps_on_iterator..rewrite_fnrcSg|] }t|tjs|qSr2r7r Operation)rDrAr2r2r3rs  zCTPUExtended._experimental_run_steps_on_iterator..csg|] }|dqSr5r2)rDrb)last_step_tensor_outputs output_numr2r3rs)r,r?r ZMultiStepContextrZget_default_graphZ_get_control_flow_contextZ_outer_control_flow_contextr7rrrrrr)repeatr groupZrun_opr\r_set_last_step_outputs)rIr<rZ iterationsZinitial_loop_valuesrrr2)rr<rrrrrIr3#_experimental_run_steps_on_iteratorms>        z/TPUExtended._experimental_run_steps_on_iteratorcCs>t|||i|WdS1swYdSr5)_TPUReplicaContextr)rIr<r`rar2r2r3_call_for_each_replicas $z"TPUExtended._call_for_each_replicaccs|jjd}||krtd|||j|z=tdur$dVn#t t | dVWdn1s:wYW|j dSW|j dSW|j dS|j w)9Places variables and ops on the specified logical device.rz]`logical_device_id` not in range (was {}, but there are only {} logical devices per replica).N) rrr[rrrWrenclosing_tpu_contextrrr'corepop)rIrrr2r2r3experimental_logical_devices*   z'TPUExtended.experimental_logical_devicecCst|jdS)zExperimental method added to be used by Estimator. This is a private method only to be used by Estimator. Other frameworks should directly be calling `tf.tpu.experimental.initialize_tpu_system` N)rZinitialize_tpu_systemrrr2r2r3_experimental_initialize_systemsz+TPUExtended._experimental_initialize_systemc  s|ddr di|S|dd}|dur|fi|S|dd}|dur6jddjdfn't|tjrZt|jdi|WdS1sTwYn|jjj \fddfd d fd d fd d}fdd}j st rj rdkr|}n|}d|d<n j rdkr}n}t j|t jt jfi|}j st rt|dj|S)z?Create a TPUMirroredVariable. See `DistributionStrategy.scope`.Zskip_mirrored_creatorFcustom_tpu_variable_creatorN colocate_withrc sd}g}tD]\}}t|p|dkr6|d}tt|r%|n|}Wdn1s1wY|dkrL|djdd}d||f|d<||d<ttj di|}Wdn1shwYt |t j ruJ| |Wdn1swYq|S)Returns a list of `tf.Variable`s. The list contains `number_replicas` `tf.Variable`s and can be used to initialize a `TPUMirroredVariable`. Args: **kwargs: the keyword arguments for creating a variable Nr initial_value:%s/replica_%d/rSr2)rPrrr6r9rSsplitr device_policyDEVICE_PLACEMENT_SILENTr7rTPUMirroredVariablerW)rar value_listrbrvar0namerF)r next_creatorr2r3_create_mirrored_tpu_variabless2    zDTPUExtended._create_variable.._create_mirrored_tpu_variablesc s|d}tt|r|n|}Wdn1swYg}tD]I}g}tD],}tj||||d<di|}Wdn1sPwY||q.d|d|}tj ||d}||q&|S)Returns a list of `TPUReplicatedVariable`s. The list consists of `num_replicas` `TPUReplicatedVariable`s and can be used to initialize a `TPUMirroredVariable`. Each `TPUReplicatedVariable` contains a list of `tf.Variable`s which are replicated to `num_cores_per_replica` logical cores to enable XLA SPMD compilation. Args: **kwargs: the keyword arguments for creating a variable rN{}/r:{}rSr#rSr2) r6r9rrrrrWrrTPUReplicatedVariable) rarmirrored_replicated_var_listrreplicated_var_list logic_core_idrF replica_nametpu_replicated_var)r rrrIr2r3)_create_mirrored_tpu_replicated_variables,s.      zOTPUExtended._create_variable.._create_mirrored_tpu_replicated_variablescs.tjdi|}j|t|dj|S)N _lazy_scoper2)rZTPUUninitializedVariablelazy_variable_trackerZadd_uninitialized_varsetattr)raZuninitialized_variablerr2r3uninitialized_variable_creatorRs zDTPUExtended._create_variable..uninitialized_variable_creatorc s|dddurd i|Sg}tD]\}}t||dkrR|dd}t|durCt|r8|}tj||ddd}Wdn1sMwY|dkrh|djdd}d||f|d<||d<|dddur{|dj |d<|d ddur|dj |d <t t j d i|}Wdn1swYt|tjrJ||Wdn1swYq|S) rrNrrrrrrSrr2)getrPrrr6r9convert_to_tensorrSrrrrrrr7rrrW)rarrbrrrrF)r!rr/r2r3,_create_uninitialized_mirrored_tpu_variables[s@     !zRTPUExtended._create_variable.._create_uninitialized_mirrored_tpu_variablesc sT|dd}|dd}|dd}|durd i|St4|durMt|r,|}tj||d}||d<|durB|dj|d<|durM|dj|d<Wdn1sWwYg}tD]E}g}tD](}tj ||d i|}Wdn1swY| |qjd |d|} t j || d} | | qb|S) r"rNrrrr#rSr$r2)r0r6r9rr1rrrrrrWrrr%) rarrrr&rr'r(rFr)r*)r+rrrIr/r2r37_create_uninitialized_mirrored_tpu_replicated_variablessB      z]TPUExtended._create_variable.._create_uninitialized_mirrored_tpu_replicated_variablesrTZ!experimental_batch_initializationr,r2)rrrr7r rrrZ_devicesrrr4rr Zcreate_mirrored_variablerZTPU_VARIABLE_CLASS_MAPPINGZTPU_VARIABLE_POLICY_MAPPINGr.r-) rIr rarrr2r3Z real_creatorZmirrored_variabler2)r+r!rr rrrIr/r3_create_variablesP    " % & 2 2  zTPUExtended._create_variablecCst|dds t|_|jS)N_lazy_variable_tracker)getattrrZLazyVariableTrackerr5rr2r2r3r-s  z!TPUExtended.lazy_variable_trackercsfdd}td|S)Nc sdt}jD] }t|||i|||<Wdn1s$wYq t|Sr5) rrrkeysrrrZPerWorkerResourcer)r r`raZ host_to_tablerrr2r3lookup_creators z;TPUExtended._resource_creator_scope..lookup_creatorZStaticHashTable)rZresource_creator_scope)rIr8r2rr3_resource_creator_scopes  z#TPUExtended._resource_creator_scopecsttjsStj}ttjr%jdur%tfddjjD}tjtkr4t j ||d}n*t j |dt|d}t tt|tdD]}t j |g|||td|d}qI| ||}|S)Nc3|] }j|VqdSr5_packed_variableZ on_devicervaluer2r3rM  z8TPUExtended._gather_to_implementation..axisr) r7rDistributedValuesrr@r<rr\_XLA_OP_BY_OP_INPUTS_LIMITrconcatr_broadcast_output)rIr> destinationsrArroutputrbr2r=r3_gather_to_implementations4      z%TPUExtended._gather_to_implementationcCst|}t|dkr9t|d}t|j}||kr7t|t |}Wd|S1s2wY|St ||}|S)Nrr) cross_device_ops_libZget_devices_fromr\rrrrrrrZsimple_broadcast)rIrFrGrZdest_canonicalZhost_canonicalr2r2r3rE s       zTPUExtended._broadcast_outputcsJttjs tr4tdur4|tjj kr!t d|j n|tjj kr/td|dtSttjsCt|||j Sj}ttjr^jdur^tfddjjD}tjtkrkt |}n$tj|d|djd}tdt|tD]}|t |||t7}q|tjj kr|dt|9}|||}|S)Ng?z `reduce_op`=z[ is not supported. Currently we only support ReduceOp.SUM and ReduceOp.MEAN in TPUStrategy.c3r:r5r;rr=r2r3rM7r?z)TPUExtended._reduce_to..rr)r7rrBr is_tf_typerrr ReduceOpZMEANr!Z scalar_mulrSUMr;r*Zcross_replica_sumrIZreduce_non_distributed_valuer@r<rrr\rCZadd_nrZ zeros_likerrrE)rI reduce_opr>rFrrrGrbr2r=r3 _reduce_tosF            zTPUExtended._reduce_toc Cst|tjst|tjsJtdur-|r!||g|Ri|S||g|Ri|fS|j}|durQt sQ|rE||g|Ri|S||g|Ri|fSg}g}|durg|j D] } | || fq\n|j D] } | | | j fqj|jtjjkr|jtjjkrt|t|t|D]e\} } | d} | d} d| } t | Ht| 3t| | || gt| |Rit| |Wdn1swYWdn1swYWdn1swYqt|||S)Nrrz update_%d)r7rTPUVariableMixinr"BaseResourceVariablerrr<rr1rrWrrZsynchronization variables_libZVariableSynchronizationZON_READZ aggregationZVariableAggregationNONEr Zassert_mirroredrPrr UpdateContextZ name_scoperZupdate_regroup)rIvarr<r`rar Z packed_varZupdatesZvalues_and_devicesrr>rbZvalue_and_devicerSr2r2r3_updateMs\        zTPUExtended._updatecCs$t|tjst|tjsJ|Sr5)r7rrOr"rPZ read_value)rIrTr2r2r3read_var{szTPUExtended.read_varcC|Sr5r2)rIr>r2r2r3value_containerzTPUExtended.value_containercsX~tttfr Stdur*fddt|jD}tj|dd|jd}|dSS)Ncsg|]}qSr2r2)rDrrr2r3rsz-TPUExtended._broadcast_to..rZconcat_dimensionZsplit_dimensionZ split_count) r7floatintrrrrr* all_to_all)rIrrFZbroadcast_tensorresultr2rZr3 _broadcast_tos zTPUExtended._broadcast_tocs4jdur jjSttfddtjjDS)Ncsg|]}j|qSr2)rr)rDrrr2r3rsz)TPUExtended.num_hosts..)rrrvr\rsrrrr2rr3rvs   zTPUExtended.num_hostscCs0|jdur |jjS|jj|jj}t|jj|Sr5)rrZnum_of_cores_per_hostrminr)rIZmax_models_per_hostr2r2r3rws z!TPUExtended.num_replicas_per_hostcCs|jdur |jjS|jjSr5)rrZ num_coresrrr2r2r3rs z!TPUExtended._num_replicas_in_synccCdS)NFr2rr2r2r3experimental_between_graphz&TPUExtended.experimental_between_graphcCrcNTr2rr2r2r3experimental_should_initrez$TPUExtended.experimental_should_initcCrcrfr2rr2r2r3should_checkpointrezTPUExtended.should_checkpointcCrcrfr2rr2r2r3should_save_summaryrezTPUExtended.should_save_summarycCst|jdd|jdfS)Nr)rrrrr2r2r3worker_devicesszTPUExtended.worker_devicescC|jSr5)rjrr2r2r3parameter_devicesszTPUExtended.parameter_devicescCst|jjS)z7Return the `tf.tpu.experimental.HardwareFeature` class.)r(ZHardwareFeaturerrr2r2r3r(sz TPUExtended.tpu_hardware_featurecCrkr5)r)rIZvar_listr2r2r3non_slot_devicesszTPUExtended.non_slot_devicesc Cs~t|jEtd/||i|}|r'|WdWdSt|j|WdWdS1s?wYWddS1sOwYdSr5)rrrrrSr,rZ_local_results)rIrr<r`rar r_r2r2r3_update_non_slots RzTPUExtended._update_non_slotcCs"~~~|r|||dSdSr5)CopyFrom_update_config_proto)rIZsession_config cluster_specZ task_typetask_idr2r2r3 _configureszTPUExtended._configurecCs2t|}d|_|j}|r|j||Srf)copydeepcopyZisolate_session_staterrqZ cluster_defroZas_cluster_def)rIZ config_protoZupdated_configrqr2r2r3rps  z TPUExtended._update_config_protocCrc)z`make_dataset_iterator` and `make_numpy_iterator` use global batch size. `make_input_fn_iterator` assumes per-replica batching. Returns: Boolean. Tr2rr2r2r3_global_batch_sizes zTPUExtended._global_batch_sizecCs|||}|||Sr5)_tpu_function_creator)rIr<r`rarfuncr2r2r3rs  zTPUExtended.tpu_runcsVtrjvrjSfdd}tr)t|}|j<|S)Nc stdd|||duri}ggfdd}g}tjD]}|tj|tjdt ||t ||gq j r||r|g}t |d}|D])}t|rU|jj}nt|}|duretd|tdg|} || qIt |d|}nd}jrtjj} nd} jptjj d } tj!||j"|| | d } Wdn1swYd d t#dt$rňdd<ddust#dt%j&rdgt'| } n fd d| D} t (| S)z3TF Function used to replicate the user computation.rz8`TPUStrategy.run` is called with [args: %s] [kwargs: %s]NcsLt|d|i|d<WddS1swYdS)z>Wraps user function to provide replica ID and `Tensor` inputs.replica_id_in_sync_grouprN)r )rZ replica_argsZreplica_kwargs)r<r_strategyr2r3 replicated_fns  zNTPUExtended._tpu_function_creator..tpu_function..replicated_fnrrzKinput tensor {} to TPUStrategy.run() has unknown rank, which is not allowedr)r&maximum_shapes padding_specrcSsdd|DS)NcSrr2r)rDor2r2r3rFsz]TPUExtended._tpu_function_creator..tpu_function....r2rr2r2r3rFszITPUExtended._tpu_function_creator..tpu_function..c s&g|]}tdt|qSr)r,pack_sequence_asr?)rDrGZ filter_opsr_r2r3rNszKTPUExtended._tpu_function_creator..tpu_function..))rZvlogrrrWrZconstantrint32r rZ&experimental_enable_dynamic_batch_sizer,r?rrJrrankrndimr[rrZ TensorShaperZ&experimental_bucketizing_dynamic_shaper'Z PaddingSpecZ POWER_OF_TWOscopeZexperimental_xla_optionsrrrrr7rrrr\r) r`rar|rrbr}Zflattened_listZ input_tensorrZ maximum_shaper~rrr<rrIr{rr3 tpu_functionsv              z7TPUExtended._tpu_function_creator..tpu_function)rr1rrrr)rIr<rrr2rr3rws N  z!TPUExtended._tpu_function_creatorcCrc)zAWhether this strategy indicates working in multi-worker settings.Fr2rr2r2r3_in_multi_worker_modeYsz!TPUExtended._in_multi_worker_modecCrWr5r2)rIrzr2r2r3_get_local_replica_idbrYz!TPUExtended._get_local_replica_id)NNNFFr5)NNNN)7rrrrrorrrrrZ PER_WORKERrrrrrrrr r  contextlibcontextmanagerrrr4rr-r9rHrErNrUrVrXr`rvrwrrdrgrhrirjrlr(rmrnrsrprvrrwrrrr2r2ryr3rp[sl*  ` l  1.               Y rpcCs8t|tr|dkr |S||Stt|d|||Sr>)r7r]rwhere_v2r!Z greater_equal)rArr2r2r3_make_axis_nonnegativefs  rc@s@eZdZdZdddZeddZddZd d Zdd d Z d S)r z+Replication Context class for TPU Strategy.rcCstjj|||ddS)Nry)rReplicaContextro)rIr{rzr2r2r3ros z_TPUReplicaContext.__init__cCs>t||j}t|j}|durtdfS|jj |fSr>) rZrequire_replica_contextZ _strategyrZconstant_valuerzr'rrurj)rIZdsrr2r2r3rs   z_TPUReplicaContext.devicescCs|jj|S)r)r{rur)rIrr2r2r3rsz._TPUReplicaContext.experimental_logical_devicecCsNt|trt|}|||j9<|Sttt|||t j|}|Sr5) r7r]rrrrr!equalrr)rI value_shape value_rankrA output_shaper2r2r3 _compute_all_gather_output_shapes z3_TPUReplicaContext._compute_all_gather_output_shapeNcsX~t|D] }t|tjrtdqfddfddt|D}t||S)Nz)all_gather does not support IndexedSlicescst|}|jjdurt|}t|}n#|jj}|j}t|}tt|D]}||dur8||||<q*t||}t |t rPdg|d}j ||<nt t t |d|j d}|||}|jtvrtjj }t||}t ||j}t|||} tjj| } t| |Stj||d} t| |} tj| ||j d} tjdg} t| j g} tj| ddj d} t jt| j dd}tj| ||d}t||S)Nrr@r[r) rr1rrras_listrr\rr7r]rrr!rrr&_DTYPES_SUPPORTED_BY_CROSS_REPLICA_SUMZone_hotrzrcastZ expand_dimsZ all_reducer rKrLrr*r^Zargmaxgather)r>rArrZvalue_shape_tensorrbZreplica_broadcast_shaperZreplica_id_maskZgathered_valuerZunordered_outputZconcat_replica_idZxla_to_replica_context_idZreplica_context_to_xla_idZsorted_with_extra_dimrr2r3_all_gather_tensors                z9_TPUReplicaContext.all_gather.._all_gather_tensorcsg|]}|dqS)r@r2)rDt)rrAr2r3rsz1_TPUReplicaContext.all_gather..)r,r?r7rZ IndexedSlicesr;r)rIr>rAZexperimental_hintsrFZysr2)rrArIr3 all_gathers  [ z_TPUReplicaContext.all_gatherrr5) rrrrrorrrrrr2r2r2r3r s    r cCs\t|j|}|jD]\}}||}|dur t|||<q |d||<q ||dS)z0Sets the last step outputs on the given context.Nr)r,rrZ_last_step_outputs_reduce_opsrOrZ PerReplicar )rrZlast_step_tensor_outputs_dictrSrMrGr2r2r3r sr )hrrrrrtr^rZabslrnumpyrZ tensorflow.python.autograph.corerr~Z tensorflow.python.autograph.implrr}Z+tensorflow.python.compiler.xla.experimentalrZtensorflow.python.distributerrIrrr r r r r rrrrZ-tensorflow.python.distribute.cluster_resolverrrZtensorflow.python.distribute.v1rZtensorflow.python.eagerrrrZtensorflow.python.frameworkrrrrrrrrrrZtensorflow.python.opsrr r!r"r#rQZtensorflow.python.ops.raggedr$Ztensorflow.python.saved_modelr%Ztensorflow.python.tpur&Zdevice_assignment_libr'r(r)Ztensorflow.python.tpu.opsr*Ztensorflow.python.utilr+r,r-Z tensorflow.python.util.tf_exportr.rCr0r4rr6r=reZStrategyrgZdeprecated_endpointsrkZ StrategyV1rZStrategyExtendedV1rprZbfloat16Zfloat16Zfloat32Zfloat64rZuint32rrr r r2r2r2r3s                                                w + Uf