o i eO@sVdZddlZddlZddlmmZddlm Z ddlm Z ddlm Z ddlm Z ddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZddZ   dpddZddZ dqddZ!ddZ"ddZ# drddZ$ddZ%dd Z&d!d"Z'd#d$Z(d%d&Z)d'd(Z*d)d*Z+ +dsd,d-Z, drd.d/Z-d0d1Z.d2d3Z/d4d5Z0d6d7Z1d8d9Z2d:d;Z3dd?Z5 dtd@dAZ6dtdBdCZ7dtdDdEZ8dFdGZ9dHdIZ:dJdKZ;dLdMZdRdSZ?dTdUZ@dVdWZAdXdYZBdZd[ZCd\d]ZDd^d_ZEd`daZFdbdcZGdddeZHejIdfdgZJdhdiZKdjdkZLdldmZMdndoZNdS)uz*Utilities related to distributed training.N)backend) callbacks)metrics) optimizers)distribute_coordinator_utils)distributed_training_utils)training_utils_v1) optimizer_v2) tf_contextlib)ModeKeys) tf_loggingc Csg}|jD]5}t|j}|d|}t|j|D]\}}tjjr(||q| | ||q||d}qtjjsKt | |dSdS)aSets the weights of the replicated models. The weights of the replicated models are set to the weights of the original model. The weights of the replicated model are Mirrored variables and hence we need to use the `update` call within a DistributionStrategy scope. Args: distribution_strategy: DistributionStrategy used to distribute training and validation. dist_model: The replicated models on the different devices. weights: The weights of the original model. N)layerslenweightsziptfcompatv1#executing_eagerly_outside_functionsassignappendunwrapr get_sessionrun) distribution_strategy dist_modelr assign_opslayer num_param layer_weightsswwr"a/var/www/myenv/lib/python3.10/site-packages/keras/src/distribute/distributed_training_utils_v1.py set_weights%s      r$Fc Csvt||}t|||}|rt||}nd}i} |r5|d} | r't|| | d<|d} | r5t|| | d<|||| fS)aUnwrap the list of values contained in the PerReplica parameters. This function calls `flatten_per_replica_values` to parse each of the input parameters into a list of values on the different devices. If we set `with_loss_tensor` to be True, we also call `reduce` on the list of losses on the different devices to give us one loss tensor. Args: distribution_strategy: DistributionStrategy used to distribute training and validation. grouped_inputs: PerReplica inputs returned from the train or test function that we ran on each device. grouped_outputs: PerReplica outputs returned from the train or test function that we ran on each device. grouped_updates: PerReplica updates returned from the train or test function that we ran on each device. grouped_session_args: PerReplica session args returned from the train or test function that we ran on each device. with_loss_tensor: Boolean that indicates if we need to add the reduced loss tensor as one of the outputs. Returns: Values of each of the PerReplica parameters. N feed_dictfetches)flatten_per_replica_valuesunwrap_outputsget) rgrouped_inputsgrouped_outputsgrouped_updatesgrouped_session_argswith_loss_tensor all_inputs all_outputs all_updatesall_session_argsgrouped_feed_dictgrouped_fetchesr"r"r# unwrap_valuesAs0#   r5cCs|tjkr t||S|jtjjj|dddd}t||d}t||d}|jtjjj|ddd}t |rOtj j rO|dd|j }|dd|j }|g|||dS) zBUnwrap the list of outputs contained in the PerReplica parameters. total_lossrNaxis output_lossesr batch_size)r6r9rr:)r PREDICTr'reducer distributeReduceOpSUMris_tpu_strategyrrrnum_replicas_in_sync)strategyr+moder6r9rr:r"r"r#unwrap_output_dicts.    rDcCsz|st||St|ts|g}|jtjjj|ddd}t||dd}t |r8tj j r8|dd|j }|g|S)aUnwrap the list of outputs contained in the PerReplica parameters. This function calls `flatten_per_replica_values` to parse each of the input parameters into a list of outputs on the different devices. If we set `with_loss_tensor` to be True, we also call `reduce` on the list of losses on the different devices to give us one loss tensor. Args: distribution_strategy: DistributionStrategy used to distribute training and validation. grouped_outputs: PerReplica outputs returned from the train or test function that we ran on each device. with_loss_tensor: Boolean that indicates if we need to add the reduced loss tensor as one of the outputs. Returns: Values of each of the PerReplica outputs. rNr7)r' isinstancelistr<rr=r>r?rr@rrrrA)rr+r.lossr0r"r"r#r(s$    r(csfddtj|DS)aUnwraps and flattens a nest of PerReplica parameters. PerReplica values have one value associated with each device. Each entry in the PerReplica dict has a device `key` and the corresponding value on the device as the `value`. In this function we take a PerReplica value or a list of PerReplica values and return all the values in the PerReplica dict. Args: distribution_strategy: DistributionStrategy used to distribute training and validation. per_replica_values: List of PerReplica object or a single PerReplica object. Returns: List of values of all the PerReplica objects. cs g|] }|D]}|q qSr")r).0 flattenederr"r# sz.flatten_per_replica_values..)rnestflatten)rper_replica_valuesr"rLr#r's  r'cCsl|r2|D]/}t|tjtjfrt|tjstd|t|tjr1t|ddr1t t dd|_ qdSdS)aValidate whether given callbacks are supported by DistributionStrategy. Args: input_callbacks: List of callbacks passed by the user to fit. optimizer: Optimizer instance used to train the model. Raises: ValueError: If `LearningRateScheduler` or `ReduceLROnPlateau` is one of the callbacks passed. ValueError: If `write_grads` is one of the parameters passed as part of the TensorBoard callback. zWYou must specify a Keras Optimizer V2 when using %s callback with DistributionStrategy. write_gradsFz}`write_grads` in the TensorBoard callback is not supported when using DistributionStrategy. Setting `write_grads` to `False`.N) rFrLearningRateSchedulerReduceLROnPlateaur OptimizerV2 ValueError TensorBoardgetattrloggingwarning UserWarningrQ)input_callbacks optimizercallbackr"r"r#validate_callbackss.     r^cCsDt||}|durt||}nd}|durt||}nd}|||fS)a/Validate all the components of a DistributedValue Dataset input. Args: distribution_strategy: The current DistributionStrategy used to call `fit`/`evaluate`. x: Input Dataset DistributedValue object. For example, when we use `MirroredStrategy` this is a PerReplica object with a tensor for each device set in the dict. x can also be a tuple or dict. The keys of the dict should match the names of the input layers of the model. y: Target Dataset DistributedValue object. For example, when we use `MirroredStrategy` this is a PerReplica object with a tensor for each device set in the dict. y can also be a tuple or dict. The keys of the dict should match the names of the output layers of the model. sample_weights: Sample weights Dataset DistributedValue object. For example, when we use `MirroredStrategy` this is a PerReplica object with a tensor for each device set in the dict. Returns: The unwrapped values list of the x and y DistributedValues inputs. Raises: ValueError: If x and y do not have support for being evaluated as tensors. or if x and y contain elements that are not tensors or if x and y contain elements that have a shape or dtype mismatch. N)validate_per_replica_inputs)rxysample_weights x_values_list y_values_listsample_weights_listr"r"r##validate_distributed_dataset_inputs!s "  rfcCsxtj|}g}|D]/}||}|D]}t|s#tdt|qts-t ||t ||| |dq |S)aValidates PerReplica dataset input list. Args: distribution_strategy: The current DistributionStrategy used to call `fit`, `evaluate` and `predict`. x: A list of PerReplica objects that represent the input or target values. Returns: List containing the first element of each of the PerReplica objects in the input list. Raises: ValueError: If any of the objects in the `per_replica_list` is not a tensor. zHDataset input to the model should be tensors instead they are of type {}r) rrNrOr is_tensorrUformattypeexecuting_eagerlyvalidate_all_tensor_shapesvalidate_all_tensor_typesr)rr`per_replica_listrcx_valuesvaluer"r"r#r_Us      r_cCs>|dj}tdt|D]}|||jkrtd|q dS)NrrEzAInput tensor dtypes do not match for distributed tensor inputs {})dtyperangerrUrh)r`rnx_dtypeir"r"r#rl~s rlcCsF|dj}tdt|D]}|||jkr td|qdS)NrrEzAInput tensor shapes do not match for distributed tensor inputs {})shapeas_listrqrrUrh)r`rnx_shapersr"r"r#rksrkcCstt}g}|D] }t|dds||q |sdS |dd|D}g}t||D]\}}|s:||d|_q/|sBdSq)z0Utility to wait for variables to be initialized._keras_initializedFNTcSsg|] }tjj|qSr")rrris_variable_initialized)rIvr"r"r#rMsz5_wait_for_variable_initialization..)r_get_variables get_graphrWrrrrw)session all_variablescandidate_varsryis_initializeduninitialized_varsflagr"r"r#!_wait_for_variable_initializations(    rcCsttdS)zLInitialize or restore variables or wait for variables to be initialized.N)r_initialize_variables _get_sessionr"r"r"r#"init_restore_or_wait_for_variablessrcCs0t|tjjjjst|tjjjjrtddS)zValidate inputs when using DistributionStrategy. Args: x: Model Inputs. y: Model Targets. Raises: ValueError: if input is not a Dataset or a numpy array(when we use MirroredStrategy). z`DistributionStrategy` does not support inputs of type Iterator. You must pass a `tf.data.Dataset` object or a numpy array as input.N)rFrrrdataIteratorrU)r`rar"r"r#validate_inputss rcCs,tjtjjj|}dd|D}| S)z9Returns whether a dataset contains a final partial batch.cSsg|]}|s|qSr")is_fully_defined)rIsr"r"r#rMsz2is_dataset_shape_fully_defined..)rrNrOrrrget_output_shapes)datasetshapesunknown_shapesr"r"r#is_dataset_shape_fully_definedsrcCsptj|d}t|tjr4|jd}|r)d|krdkr)nnt|d|}t|||||d\}}||fS)zFProcess the batch size and step size based on input and dist strategy.rrg?rErC) rrNrOrFnpndarrayrtintget_input_params)rBinputsr:steps_per_epochrCvalidation_split first_x_value num_samplesr"r"r#process_batch_and_step_sizes    rc Cszt| }tr|tjkpt| }n|tjkp)|tjks%|tj ko)t|}|dur_|dur8t |d}n |}|rA||j 9}|rNt ||t}nP||rZtd||f||}n?|durt||rotd||f||}n*|}|r}||j 9}||}|r|dkr||ddnd}||krtd|||f|r||j rtd||j f||j }||fS|}||fS) a Calculate the number of batches and steps/steps_per_epoch. Args: distribution_strategy: The DistributionStrategy used to compile the model. num_samples: The number of samples from which we determine the batch size and steps. steps: The specified number of steps. batch_size: The specified batch_size. mode: ModeKey representing whether input will be used for training, evaluation, or prediction. This is used to relax the constraints on consuming all the training samples to keep compatibility till we support partial batches. If none, then partial batches are not allowed. Returns: steps: The steps or steps_per_epoch argument depending on if a user is calling `fit`, `evaluate` or `predict`. If the is_training flag is set we don't require the number of samples to be used completely. batch_size: The batch size to be used in model iterations. Raises: ValueError: If the number of batches or steps evaluates to 0. N z;The number of samples %s is not divisible by batch size %s.zThe number of samples %s is not divisible by steps %s. Please change the number of steps to a value that can consume all the samplesrErz[Number of samples %s is less than samples required for specified batch_size %s and steps %szkThe batch size (%s) could not be sharded evenly across the sync replicas (%s) in the distribution strategy.) dist_utilsglobal_batch_size_supportedrrjr TRAINrr@r;TESTminrArceilastyperrU) rrstepsr:rCuse_per_replica_batchallow_partial_batchglobal_batch_sizemin_num_samplesr"r"r#rsz          rcCs2tjtjjj|}|dj}|r|dSdSNr)rrNrOrrrrdims)iteratorrrr"r"r#get_batch_dimensionZs rcCs@| ||}Wdn1swYt|||SN)scopemake_dataset_iteratorinitialize_iterator)rrrr"r"r# get_iteratorbs   rcCsd|$t|j}ts t|f|WddSWddS1s+wYdSr)rrgroup initializerrjrrr)rrinit_opr"r"r#ris  "rcCs|}ttj|t|jkr|}d}d}nttj|t|jt|jkr1|\}}d}n|\}}}t|j||||||fS)zCGet elements from the iterator and verify the input shape and type.N) get_nextrrrNrOroutputsrf_distribution_strategy)rmodel next_elementr`rarbr"r"r#_get_input_from_iteratorps   rcs|j}t|\}}t|r|durtdttr'fdd|jDt|rBt |t ||}t j t j|f\}nt |tjkrSg}g}n|durkt|rkt rf|jsftdt ||}||g}t|S)aPrepare feed values to the model execution function. Args: model: Model to prepare feed values for. inputs: List or dict of model inputs. targets: Optional list of model targets. sample_weights: Optional list of sample weight arrays. mode: One of ModeKeys.TRAIN/ModeKeys.TEST/ModeKeys.PREDICT. Returns: Feed values for the model in the given mode. Nz,TPUStrategy does not support sample weights.csg|]}|qSr"r")rIkeyrr"r#rMz(_prepare_feed_values..zb`sample_weight` is not supported when using tf.distribute.Strategy in eager mode and cloning=True.)rrrr@rUrFdict_feed_input_namesis_distributing_by_cloningr'rrN map_structurerstandardize_single_array ModelInputsrur r;rj_compile_distributionNotImplementedErrortuple)rrtargetsrbrCrBinsr"rr#_prepare_feed_valuess2          rcCs0t|jr tjr dStjjrt|j SdS)a)Decide whether this model is going to be distributed via cloning. We are going to distribute the model by cloning in graph mode. Args: model: Keras model to distribute. Returns: True if the `model` is going to be distributed using cloning and False otherwise. FT) rr@rrrjrrrboolrrr"r"r#rs   rcCs,|jsdSd|_d|_d|_d|_d|_dS)z$Custom compile for TPU predict mode.NT)built _is_compiledr6train_function test_functionpredict_functionrr"r"r#_custom_compile_for_predicts rc sddlm}ddlm}t||jr|j|||jd}n |j|||jd}|j |_ ddfdd|j D|_ t|t rCt j |}|tjkrR|d urRt||S|j|j|jt|j|j|jt|j|d |S) a6Build an updated model on replicas. We create a new Keras model while sharing the variables from the old graph. Building a new sub-graph is required since the original keras model creates placeholders for the input and the output that are not accessible till we call iterator.get_next() inside the step_fn for `fit`/`evaluate`/`predict`. The sharing of weights and layers between the old and the new model guarantee that we're using Strategy variables and any updates on either model are reflected correctly in callbacks and loop iterations. We need to make sure we share the optimizers between the old and the new model as well so that optimizer state is not lost if the user is running fit multiple times. Args: model: Model to be replicated across Replicas mode: Which of fit/eval/predict is building the distributed network inputs: Input variables to be passed to the model targets: Target tensor to be passed to model.compile Returns: A new model with shared layers with the old model. rmodels) sequential) input_tensorslayer_fncS|jtjkr t|tjS|Srrprbfloat16castfloat32outputr"r"r#_upcast_low_precision_outputs z@_build_network_on_replica.._upcast_low_precision_outputscg|]}|qSr"r"rIorr"r#rMz-_build_network_on_replica..Nr loss_weightssample_weight_modeweighted_metricstarget_tensors) keras.srcrkeras.src.enginerrF Sequential_clone_sequential_model share_weights_clone_functional_model_callable_lossesrrrrNrOr r;rcompiler\rHmetrics_module clone_metrics_compile_metricsrr_compile_weighted_metrics)rrCrrrr updated_modelr"rr#_build_network_on_replicas>         rc Cst7||jjt||||fd}t|||Wdn1s(wYWddSWddS1s@wYdSz&Create a cloned model on each replica.argsN)rr{ as_defaultrextendedcall_for_each_replicarset_distributed_modelrrBrCrrdistributed_modelr"r"r#_build_distributed_network/s  Prc sddlm}|j||d}t|jtjr|j}n |j}|jj |}ddfdd|j D|_ t|t r>t j |}|tjkrM|durMt||S|j||jt|j|j|jt|j|d |S) z&Clone and build the given keras_model.rr)rcSrrrrr"r"r#rLrz=_clone_and_build_model.._upcast_low_precision_outputscrr"r"rrr"r#rMRrz*_clone_and_build_model..Nr)rr clone_modelrFr\r TFOptimizer get_config __class__ from_configrrrrNrOr r;rrrHrrrrrr)rrCrrr cloned_modelr\optimizer_configr"rr#_clone_and_build_model:s6        r c Cst.||jjt||||fd}t|||Wdn1s(wYWdn1s7wY|tj krH| |dSdSr) rr{rrrrr rr r_make_callback_modelrr"r"r#clone_model_on_replicasis  r cCs>t|r t||St||}|r|St||}t||||S)zLMakes or reuses function to run one step of distributed model execution.)r%_make_execution_function_with_cloningget_distributed_function(_make_execution_function_without_cloningset_distributed_function)rrCdistributed_functiondistribution_functionr"r"r#_make_execution_functionts   rcsr|j(t|fdd|js$tfdd}n}|WdS1s2wYdS)zBCreates a function to run one step of distributed model execution.cs8|\}}}j|||fd}t|tjkd}|S)z;A single step of the distributed execution across replicas.rr.)rr(r r;)input_fnr`rarbrr0)rCper_replica_functionrBr"r#rs   zF_make_execution_function_without_cloning..distributed_functioncsdd|DS)NcSg|]}|qSr")numpy)rIoutr"r"r#rMrzX_make_execution_function_without_cloning..execution_function..r")r)rr"r#execution_functionszD_make_execution_function_without_cloning..execution_functionN)rr _make_replica_execution_function run_eagerlyrfunction)rrCrr")rrCrrBr#rs   $rcsR|tjkr j}n|tjkrj}n dfdd }|}|tjkr'tj|dd}|S)z8A single step of the distributed execution on a replica.Ncs~~|Sr)predict_on_batch)r`rarbrr"r#rs z:_make_replica_execution_function..predict_on_batchF) reset_metricsNN)r rtrain_on_batchr test_on_batchr; functoolspartial)rrCfuncrr"rr#rs   rcCs,|j}|jrt|||dSt|||dS)zBuild models on each replica.N)rrr r)rrCrBr"r"r#$_make_replicated_models_with_cloningsr%cCs|t||}|rt|drt|dr|js|jS|s#t||t||}|s'Jtr1t||}nt||}||_d|_|S)zLClones or re-uses models to run one step of distributed model execution._distribution_function_recompile_exec_functionF) get_distributed_modelhasattrr'_distributed_functionr%rrj_make_eager_execution_function_make_graph_execution_function)rrCrrr"r"r#r s(      r c sfdd}|j}|=|jj|t|fd\}}}}tt|||||tjkd\}} } } t j || f| ddd| WdS1sMwYdS) zBMakes function to run one step of distributed model in graph mode.cs|}|j|j|j|jfSr)rrr updates_opsession_kwargsrfrr"r#_per_replica_functions z=_make_graph_execution_function.._per_replica_functionrr distributed_ _function)updatesnameN) rrrrr(rr5r r;rr) rrCr1rBr*r+r,r-r/r0r1r2r"rr#r,sF       $r,c sfdd}|j}t}|d|Ot|'|jj|t|fd}|\}}t |||t j kd\}} } } Wdn1sFwYtj || dddWdWdS1shwYWddS1sxwYdS) zDMakes function to run one step of distributed model eager execution.cs|}|j|jfSr)rrrr/rr"r#r1,s  z=_make_eager_execution_function.._per_replica_functionrrNeager_distributed_r3)r5) rrr{rr_scratch_graphrrr(r5r r;r) rrCr1rB global_graphgroupedr*r+r/r0_r"rr#r+)s.    Rr+cCs>|j}t||}|r|}||d}t|||dSdS)z9Copies weights from original model to distributed models.rN)rr( get_weightsrr$)original_modelrCrBrorig_model_weights first_modelr"r"r#"_copy_weights_to_distributed_modelXs r?cCsD|jr|tjkr t||}|j|d}||dSdSdS)zCCopies weights from first distributed model back to original model.rN)rr rr(rr;r$)rrCrupdated_weightsr"r"r#_copy_weights_to_original_modelds rAcCsf|dur1|tjkr1g}tt|jD]}|j}||||||}|ttj |q|S|S)zKAggregates the per-replica batch-level outputs from a distributed step.N) r r;rqrrrArconcat_along_batch_dimensionrrNrO)rB batch_outsrrCtotal_batch_outsrs num_replicas nested_outsr"r"r#_per_replica_aggregate_batchnsrGcCsJ|jr!tjtjtjfD]}t||}|r |j|d}|q dSdSr)rr rrr;r(rr)rrCrr>r"r"r#_reset_metricss rHcCt|}|j|dSr)_generate_cache_key_distributed_model_cacher)rrCrr"r"r#r(r(cCt|}||j|<dSr)rJrK)rrCrrr"r"r#rrMrcCrIr)rJ_distributed_function_cacher)rLr"r"r#r rMr cCrNr)rJrO)rrCrrr"r"r#rrMrcCs t|}|Sr)hash)rCrr"r"r#rJsrJc csp|)t| dVWdn1swYWddSWddS1s1wYdSr)rrlearning_phase_scope)rBlearning_phaser"r"r#distributed_scopesPrScCs tjSr)dcget_current_worker_contextis_chiefr"r"r"r#is_current_worker_chiefs rWcCsP|std|p g}dd|Dstd|dustr!|Sdd|DS)aFilter Callbacks based on the worker context when running multi-worker. Args: callbacks_list: A list of `Callback` instances. model: Keras model instance. Returns: The list of `Callback` instances that should be run on this worker. zXfilter_distributed_callbacks() should only be called when Keras is in multi worker mode.cSsg|] }t|tjr|qSr")rFrModelCheckpoint)rIcr"r"r#rMs  z0filter_distributed_callbacks..z]ModelCheckpoint callback is not provided. Workers will need to restart training if any fails.NcSsg|]}|js|qSr")_chief_worker_only)rIr]r"r"r#rMs )_in_multi_worker_moderUrXrYrW)callbacks_listrr"r"r#filter_distributed_callbackss  r]cCst|rCt||}|st||t||}tdd|jD|_|rEt|j|}|d}|rGd|vrIt||D]\}}|j |gdq7dSdSdSdSdS)z3Update sample_weight_mode of the distributed model.cSrr")sample_weights_mismatch)rIrKr"r"r#rMrz/_update_sample_weight_modes..rN)rb) rr(r%any_training_endpointsr'r'rr_update_sample_weight_modes)rrCrbrdistributed_modelsmr r"r"r#ras(     racCsHt|dtjrtjjd|dSt|dtjrtj|ddSt|S)z5Concats prediction outputs along the batch dimension.r)r8 sp_inputsr7)rFr SparseTensorsparseconcat RaggedTensorr concatenate)rr"r"r#rBs  rB)NNF)Fr)rr)O__doc__r"rrtensorflow.compat.v2rv2rrrrrrrkeras.src.distributerrTrrrrkeras.src.optimizers.legacyr keras.src.utilsr keras.src.utils.mode_keysr tensorflow.python.platformr rXr$r5rDr(r'r^rfr_rlrkrrrrrrrrrrrrrrrr r rrrr%r r,r+r?rArGrHr(rr rrJcontextmanagerrSrWr]rarBr"r"r"r#s             C' 2* 4)    {4 M / $ (7/    '