o i ew>@sdZddlZddlZddlmZddlmZddlmZddl m Z ddl m Z ejjZejjZejjZejjZdd Zd d Zd d ZddZddZddZddZddZddZdS)z?Training utilities for Estimator to use Distribute Coordinator.N)distribute_coordinator)distribute_coordinator_context)multi_worker_util) tf_logging) server_libcCs |stdt|tgS)z7Counts the number of parameter servers in cluster_spec.z?Internal error: `_count_ps` does not expect empty cluster_spec.) RuntimeErrorlenas_dictgetPS) cluster_specr ^/var/www/myenv/lib/python3.10/site-packages/tensorflow/python/distribute/estimator_training.py _count_ps$s rcCs4|stdt|tgt||gS)z?Counts the number of workers (including chief) in cluster_spec.zCInternal error: `_count_worker` does not expect empty cluster_spec.)rrr r WORKER)r chief_task_typer r r _count_worker-srcs|sdSg}|jvrg}|fddt|jDt|jvr'|td}|D]}||kr7||S|t||7}q+td||)z:Returns the global id of the given task type in a cluster.rcs g|] }|kr|tkr|qSr )r ).0trr r Bsz"_get_global_id..z=Internal Error: `task_type` ({}) is not in cluster_spec ({}).) jobsextendsortedr appendr job_tasksrformat)r task_typetask_idrtask_type_ordered_listnext_global_idrr rr_get_global_id7s$    r!cCsd|_|j|_|j|_|j|_|j|_|j|_ |j |_ |jrP|jt kr?t |j|_t|jtd|_t|j|j|jtd|_dSti|_d|_d|_d|_dSd|_d|_d|_dS)zDInitializes run config from distribute coordinator's worker context.Nrr)_servicer _cluster_specr _task_typer_task_id master_target_evaluation_master_masteris_chief _is_chief EVALUATORr_num_ps_replicasrCHIEF_num_worker_replicasr!_global_id_in_clusterr ClusterSpec)configworker_contextr r r$_init_run_config_from_worker_contextVs4     r4cCs |jr|jjr|jrtd|jj|_|jr&|jjr&|jr!td|jj|_t|di}| i|jrU|jrU|jj rU|rDtdt j j |_|jj |_tddS|r_d|jvs_|jsrd|_| ||tddS|svJ||_t j j|_td dS) z2Initializes RunConfig for distribution strategies.zREither `train_distribute` or`experimental_distribute.train_distribute` can be set.zPEither `eval_distribute` or`experimental_distribute.eval_distribute` can be set.clusterzXCannot set both "cluster_spec" of TF_CONFIG and `experimental_distribute.remote_cluster`zLRunConfig initialized for Distribute Coordinator with STANDALONE_CLIENT modeNmaster!Not using Distribute Coordinator.zMRunConfig initialized for Distribute Coordinator with INDEPENDENT_WORKER mode)_experimental_distributetrain_distribute_train_distribute ValueErroreval_distribute_eval_distributerr1r ._init_distributed_setting_from_environment_varremote_clusterdcCoordinatorModeSTANDALONE_CLIENT_distribute_coordinator_moder$logginginfor8_maybe_overwrite_session_config_for_distributed_trainingINDEPENDENT_WORKER)r2 tf_configr r r rinit_run_config{sF          rIcCspt|dr |jdurtddSt|jtjr#|jtjj tjj fvr,t d|jdS|j s6t ddSdS)z?Checks the config to see whether to run distribute coordinator.rCNr7Fz*Unexpected distribute_coordinator_mode: %rzZRunning `train_and_evaluate` locally, ignoring `experimental_distribute_coordinator_mode`.T) hasattrrCrDrE isinstancesix string_typesr@rArBrGwarningr )r2r r r!should_run_distribute_coordinators"    rOc s|jjs tdfdd}fdd}jtjjkr*j}|s)Jnd}tj|j|j j|j ddS)aRun distribute coordinator for Estimator's `train_and_evaluate`. Args: estimator: An `Estimator` instance to train and evaluate. train_spec: A `TrainSpec` instance to specify the training specification. eval_spec: A `EvalSpec` instance to specify the evaluation and export specification. executor_cls: the evaluation executor class of Estimator. Raises: ValueError: if `distribute_coordinator_mode` is None in RunConfig. z._worker_fncs^t}||j_t|jttdt t |j||_ d|j_ |}| dS)zFunction for evaluator task.rQN)rUrVrWr=r4rXrYrDrErZr[_eval_distributionrC_start_continuous_evaluation)r_r`executor)rb eval_spec executor_clsrdr r_eval_fns    z$train_and_evaluate.._eval_fnNmoder session_config) r2rCr;r@rArBr run_distribute_coordinatorr9r<rn)rbrdrirjrerkr r )rbrirjrcrdrtrain_and_evaluates, !  rpcjjsJj}jjsJtjj}jjsJd|jvr%tdjjtj j kr1tdjjj j rAtdjjj jfdd}tj|jj|j||jdS)z:Run distribute coordinator for Estimator's `train` method. evaluatorF'evaluator' job is not supported if you don't use `train_and_evaluate`zJOnly `STANDALONE_CLIENT` mode is supported when you call `estimator.train`zL`Estimator.train` API is not supported for %s with `STANDALONE_CLIENT` mode.csbt}||j_t}t|j|tdt t |j||_ |j r'}ng}||||S)rPrQ) rUrVrWr:rXrYr4rDrErZr[r\r*r_r`ra chief_hooksrbrTtrain_distributed_fnr rre<s   z#estimator_train.._worker_fnrl)rWrCr rnormalize_cluster_specr:rr;r@rArBextendedexperimental_between_graph __class____name__ror9rn)rbrwrTrcr rer rvrestimator_train#s2      r}crq)z=Run distribute coordinator for Estimator's `evaluate` method.rrrszMOnly `STANDALONE_CLIENT` mode is supported when you call `Estimator.evaluate`zO`Estimator.evaluate` API is not supported for %s with `STANDALONE_CLIENT` mode.cs^t}||j_t}t|j|tdt t |j||_ |j r'}ng}|||S)zFunction for evaluation.rQ) rUrVrWr=rXrYr4rDrErZr[rfr*rtrbevaluate_distributed_fnrTr rrems   z&estimator_evaluate.._worker_fnrl)rWrCr rrxr=rr;r@rArBryrzr{r|ror<rn)rbrrTrcr rer r~restimator_evaluateTs2      r)__doc__rUrLtensorflow.python.distributerr@rrXrtensorflow.python.platformrrDtensorflow.python.trainingr _TaskTyper.r,r rrrr!r4rIrOrpr}rr r r rs*       %6\ 1