AVh~dZddlZddlZddlZddlZddlZddlmZddlm Z ddl m Z ddl m Z ddl mZddlmZdd lmZdd lmZej(ZGd d eZGd deZGddeZdZGddeZ ddZdZ ddZdZdZ dZ!ddZ"ddejFdddddfdZ$y) z/A component for running distributed TensorFlow.N) config_pb2)session)distribute_coordinator_context)multi_worker_util) tf_logging) coordinator)monitored_session) server_libc eZdZdZdZdZdZdZy) _TaskTypepsworkerchief evaluatorclientN)__name__ __module__ __qualname__PSWORKERCHIEF EVALUATORCLIENTc/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/distribute_coordinator.pyr r $s " & %) &rr ceZdZdZdZdZy)CoordinatorModez(Specify how distribute coordinator runs.standalone_clientindependent_workerN)rrr__doc__STANDALONE_CLIENTINDEPENDENT_WORKERrrrrr.s0 * ,rrceZdZdZdZdZy)_Barrierz4A reusable barrier class for worker synchronization.c||_d|_d|_tj|_tj |_tj|_ y)zInitializes the barrier object. Args: num_participants: an integer which is the expected number of calls of `wait` pass to through this barrier. rFN) _num_participants_counter_flag threadinglocal _local_senseLock_lock Condition _condition)selfnum_participantss r__init__z_Barrier.__init__?sG.DDMDJ!)D!DJ))+DOrcV|j |j_|j5|xjdz c_|j|j k(r"d|_|jj|_ddd|j 5|j|jjk7r>|j j|j|jjk7r>|j jdddy#1swYxYw#1swYyxYw)z7Waits until all other callers reach the same wait call.rN) r)r,valuer.r(r'r0wait notify_allr1s rr7z _Barrier.waitMs"&**nD - mmqm $00 0 &&,, - # JJ$++11 1  JJ$++11 1 oo  "## -- ##sADA!D0DDD(N)rrrr!r3r7rrrr%r%<s< , #rr%c|syt|jjtjgt|jjtj gzS)z'Gets number of workers including chief.r)lenas_dictgetr rr) cluster_specs r_get_num_workersr?[s[  \ ! ! # ' ' (8(8" = >  "5B7 77rceZdZdZ ddZdZdZdZdZdZ d Z dd Z e d Z e d Ze d Ze dZe dZe dZe dZe dZe dZe dZe dZe dZy)_WorkerContexta$The worker context class. This context object provides configuration information for each task. One context manager with a worker context object will be created per invocation to the `worker_fn` where `get_current_worker_context` can be called to access the worker context object. Nc||_||_||_||_||_||_||_|j|_t||_ |j|_ y)agInitialize the worker context object. Args: strategy: a `DistributionStrategy` object. cluster_spec: a ClusterSpec object. It can be empty or None in the local training case. task_type: a string indicating the role of the corresponding task, such as "worker" or "ps". It can be None if it is local training or in-graph replicated training. task_id: an integer indicating id of the corresponding task. It can be None if it is local training or in-graph replicated training. session_config: an optional `tf.compat.v1.ConfigProto` object. rpc_layer: optional string specifying the RPC protocol for communication with worker masters. If None or empty, hosts in the `cluster_spec` will be used directly. worker_barrier: optional, the barrier object for worker synchronization. N) _strategy _cluster_spec _task_type_task_id_session_config_worker_barrier _rpc_layer_get_master_target_master_targetr? _num_workers _is_chief_is_chief_node)r1strategyr> task_typetask_idsession_config rpc_layerworker_barriers rr3z_WorkerContext.__init__lse2DN%DDODM)D)DDO113D(6D..*Drcp|jr*d|jd|jd|jdSy)Nz[cluster_spec: z , task_type: z , task_id: ]z[local])rDrPrQr9s r_debug_messagez_WorkerContext._debug_messages/    dnndll<<rctj}|rtd|jz|tj_y)Nz8You cannot run distribute coordinator in a `worker_fn`. )rget_current_worker_context ValueErrorrW_worker_contextcurrent)r1 old_contexts r __enter__z_WorkerContext.__enter__sF0KKMK  E      !!>B"22:rc.dtj_yN)rr[r\)r1unused_exception_typeunused_exception_valueunused_tracebacks r__exit__z_WorkerContext.__exit__s>B"22:rc |jr|jtjk(ry|jsttj|jj vrtj}d}nStj |jj vsJtj }d}n|j}|j}d}|jr|jdz}||jj||xsdzS)z$Return the master target for a task.r://) rDrEr rrjobsrrFrI job_tasks)r1rPrQprefixs rrJz!_WorkerContext._get_master_targets   I4G4G!G  ?? D..33 3OO 4#5#5#:#::::$$ //i g F &f D&&00;GLqI IIrc|jr-|jtjtjdfvrytj|jj vr-|jtj k(r|jdk(ryy)z,Return whether the task is the chief worker.NTrF)rDrEr rrrhrrFr9s rrMz_WorkerContext._is_chiefsh    IOOY-@-@$GG  t11666 9+++ 0B  rcR|jsy|jjy)zWaits for other workers to reach the same call to this method. Raises: ValueError: if `worker_barrier` is not passed to the __init__ method. N)rHr7r9s rwait_for_other_workersz%_WorkerContext.wait_for_other_workerss"    rc|r1tj|}|j|jn |j}|jr |jj j r:tjd|tj||j|||Stjd|tj||j||S)aReturns a session creator. The returned session creator will be configured with the correct master target and session configs. It will also run either init ops or ready ops by querying the `strategy` object when `create_session` is called on it. Args: scaffold: A `Scaffold` used for gathering or building supportive ops. If not specified a default one is created. It's used to finalize the graph. config: `ConfigProto` proto used to configure the session. checkpoint_dir: A string. Optional path to a directory where to restore variables. checkpoint_filename_with_path: Full file name path to the checkpoint file. Only one of `checkpoint_dir` or `checkpoint_filename_with_path` can be specified. max_wait_secs: Maximum time to wait for the session to become available. Returns: a descendant of SessionCreator. z.Creating chief session creator with config: %r)masterconfigcheckpoint_dircheckpoint_filename_with_pathz/Creating worker session creator with config: %r)rorp max_wait_secs) copydeepcopy MergeFromrGrCextendedexperimental_should_initlogginginfor ChiefSessionCreator master_targetWorkerSessionCreator)r1scaffoldrprqrrrsrRs rsession_creatorz_WorkerContext.session_creators4}}V,nt334++n >>T^^44MM llCVL  2 2 ##'(E GGllDfM  3 3 ##% ''rc@tj|jSr`)rtrurGr9s rrRz_WorkerContext.session_configs ==-- ..rc|jduS)z"Whether the barrier is set or not.N)rHr9s r has_barrierz_WorkerContext.has_barrier s   t ++rcjt|jxr|jtjk7S)z*Whether it is distributed training or not.)boolrDrEr rr9s rdistributed_modez_WorkerContext.distributed_modes) "" # N9;N;N(NNrc@tj|jS)z*Returns a copy of the cluster_spec object.)rtrurDr9s rr>z_WorkerContext.cluster_specs ==++ ,,rc|jS)z+Returns the role of the corresponding task.)rEr9s rrPz_WorkerContext.task_types ??rc|jS)z2Returns the id or index of the corresponding task.)rFr9s rrQz_WorkerContext.task_ids ==rc|jS)zDReturns the session master for the corresponding task to connect to.)rKr9s rr|z_WorkerContext.master_target#   rc|jS)z)Returns whether the task is a chief node.)rNr9s ris_chiefz_WorkerContext.is_chief(rrc|jS)z:Returns number of workers in the cluster, including chief.)rLr9s r num_workersz_WorkerContext.num_workers-s   rcB|jjjS)zWhether to run init ops.)rCrwrxr9s rrxz'_WorkerContext.experimental_should_init2s >> " " ; ;;rcB|jjjS)zWhether to save checkpoint.)rCrwshould_checkpointr9s rrz _WorkerContext.should_checkpoint7s >> " " 4 44rcB|jjjS)zWhether to save summaries.)rCrwshould_save_summaryr9s rrz"_WorkerContext.should_save_summary<s >> " " 6 66r)NgrpcN)NNNNi )rrrr!r3rWr^rdrJrMrmrpropertyrRrrr>rPrQr|rrrxrrrrrrArAcsB#""+HBB J2    $!%)48$( .'` / / , , O O - -           < < 5 5 7 7rrAc tj|}tj|}|tjk(r|r*|j |n|sJ|j ||||t |||||||} | 5|r+|j 5||cdddcdddS||cdddS#1swYnxYw dddy#1swYyxYw)z:Runs a single worker by calling `worker_fn` under context.)rRrSrTN)rtrur r configurerAstop_on_exception) worker_fnrOr>rPrQrRrSrTcoordcontexts r_run_single_workerrBs==0. ]]8 $()%%%( O8 ~|YH  ## %'!  " " $#"##!! x !!###!!!s*>CB= C,C=C CCcHtj|j}|tjk(r8tj|vsJtj|tji}n |j tjdtj|S)zESplit the cluster for evaluator since it needn't talk to other tasks.N)rnormalize_cluster_specr<r rpop)r>rPnew_cluster_specs r_split_cluster_for_evaluatorrgs'==GI)%%%   "2 22 2-i.A.AB,,d3  1 12B CCrcfttddtj|k(sJtj|k(sJtj|k(sJtj t k(sJtj|k(sJtj|k(sJtjSdt_ |t_|t_|t_t t_|t_|t_|sJ|j|||r|dzzGfddt}|dk(r|}nVrtjdntjd t||}t!j"|||| }|j%|t_ |S) zRuns a standard server.serverNTrgc$eZdZdZfdZdZy)$_run_std_server.._FakeServerz)A fake server that runs a master session.cbtjdtjy)NzVCreating a remote session to start a TensorFlow server, target = %r, session_config=%r)targetrp)ryrzrSession)r1rRrs rstartz*_run_std_server.._FakeServer.starts* ll +,2NDooVN;rc0 tjd)N)timesleepr9s rjoinz)_run_std_server.._FakeServer.joins  1  rN)rrrr!rr)rRrsr _FakeServerrs3<rrgooglezDStarting standard TensorFlow server, target = %r, session_config= %rz0Starting standard TensorFlow server, target = %r)job_name task_indexrpprotocol)getattr _thread_localr>rPrQsession_config_strreprrS environmentrserver_started task_addressobjectryrzrr Serverr) r>rPrQrRrSrrrrs ` @r_run_std_serverr}s ]Hd+7  % % 55 5  " "i // /  G ++ +  + +tN/C CC C  " "i // /  $ $ 33 3   $(M !-M'M#M'+N';M$'M +M   $ $Y 8&   'F F H ]F ll (llEvN/ iHL    F ,,.- -rc tj}d}tj|jvrCt j t|||tjd|f||d}|jg} tt|} tjtjfD]} tt|jj!| gD]I} t j t|||| | |f|| |d} | j| j#| K|r| |gz}n| }|j%|y)z7Runs a standalone client for between-graph replication.NrrSrrargskwargs)rSrTr)r Coordinatorr rrhr*Threadrrr%r?rrranger;r<r=appendr)rrOeval_fn eval_strategyr>rRrSr eval_threadthreadsrTrPrQtthreads_to_joins r_run_between_graph_clientrsH  ! ! #%+L---""!}lI4G4G#  K ',\:;.OOY%5%56 i\11377 2FGH    #8\9g % .  aggi nnQ   -OO**_ rc >tj}d}tj|jvrCt j t|||tjd|f||d}|jt|||dd|||} |r|j|g| S)z2Runs a standalone client for in-graph replication.Nrrr) rrr rrhr*rrrr) rrOrrr>rRrSrr worker_results r_run_in_graph_clientrs  ! ! #%+L---""!}lI4G4G#  K$   - JJ } rc|tjk(r|r=|j|n*tj|}|j|||||j dd=y)aCall strategy's `configure` to mutate the session_config. The session_config is currently needed as default config for a TensorFlow server. In the future, we should be able to remove this method and only pass the session config to a client session. )rR)rRr>rPrQN)r rrrtrudevice_filters)rOrrRr>rPrQs r)_configure_session_config_for_std_serversr se)%%%^<}}X&H %! ##A&rchtjtjj dd}d|vr t dt j|d}d|vr t d|d}d|vr t d|d}t|j d d }|j d d }|xstj}d |jvrd|j_ n*d|jvr t dd|j_ t|||||}|j|S)aStarts a standard TensorFlow server. This method parses configurations from "TF_CONFIG" environment variable and starts a TensorFlow server. The "TF_CONFIG" is typically a json string and must have information of the cluster and the role of the server in the cluster. One example is: TF_CONFIG='{ "cluster": { "worker": ["host1:2222", "host2:2222", "host3:2222"], "ps": ["host4:2222", "host5:2222"] }, "task": {"type": "worker", "index": 1} }' This "TF_CONFIG" specifies there are 3 workers and 2 ps tasks in the cluster and the current role is worker 1. Valid task types are "chief", "worker", "ps" and "evaluator" and you can have at most one "chief" and at most one "evaluator". An optional key-value can be specified is "rpc_layer". The default value is "grpc". Args: session_config: an optional `tf.compat.v1.ConfigProto` object. Users can pass in the session config object to configure server-local devices. Returns: a `tf.distribute.Server` object which has already been started. Raises: ValueError: if the "TF_CONFIG" environment is not complete. TF_CONFIG{}clusterz$"cluster" is not found in TF_CONFIG.taskz!"task" is not found in TF_CONFIG.typez9"task_type" is not found in the `task` part of TF_CONFIG.indexrrSrrz/job:chief/replica:0/task:0rz=You must have `chief` or `worker` jobs in the `cluster_spec`.z/job:worker/replica:0/task:0)r>rPrQrRrS)jsonloadsosenvironr=rZrrintr ConfigProtorh experimentalcollective_group_leaderrr)rR tf_configr>task_envrPrQrSrs rrun_standard_tensorflow_serverr'sFFjj T:;)i = >>"99)I:NO, 9 : ;; v ( 8 E GGv)  Wa( )'mmK0)!=Z%;%;%=. !!!%7|(((  I KK '7 #  &  ,,. -rrc tjtjj dd} | j d| } | j dd} |sS| j di}| j di} | r-| j d|}t | j d |}|rt j|}npt|jd rZ|jj} | j}| j}| jxs| } | j} | j}|xst!j"d }|r)t%j&d ||j)||| | |sPt%j&dt+||ddd|| |rt+||ddd|| yt%j,dy|t.j0k(r|st%j,d|xs|}|st%j,d|t2j4dfvr8|jj6rt9||||||| St;||||||| St=||||||t?||||| | }|jAy|t.jBk7rtEd|z|st%j,d|xs|}|st%j,dt=|||||||t2jFk7r(tI|jddst?||||| | }|t2jJt2jLfvrc|jj6rt+||||||| StO||||}|jPrt+|||dd|| SjAy|t2jFk(rt+||||||| S|t2jRk7rtEd|zjAy)a Runs the coordinator for distributed TensorFlow. This function runs a split coordinator for distributed TensorFlow in its default mode, i.e the STANDALONE_CLIENT mode. Given a `cluster_spec` specifying server addresses and their roles in a cluster, this coordinator will figure out how to set them up, give the underlying function the right targets for master sessions via a scope object and coordinate their training. The cluster consisting of standard servers needs to be brought up either with the standard server binary or with a binary running distribute coordinator with `task_type` set to non-client type which will then turn into standard servers. In addition to be the distribute coordinator, this is also the source of configurations for each job in the distributed training. As there are multiple ways to configure a distributed TensorFlow cluster, its context object provides these configurations so that users or higher-level APIs don't have to figure out the configuration for each job by themselves. In the between-graph replicated training, this coordinator will create multiple threads and each calls the `worker_fn` which is supposed to create its own graph and connect to one worker master given by its context object. In the in-graph replicated training, it has only one thread calling this `worker_fn`. Another mode is the INDEPENDENT_WORKER mode where each server runs a distribute coordinator which will start a standard server and optionally runs `worker_fn` depending whether it is between-graph training or in-graph replicated training. The `strategy` object is expected to be a DistributionStrategy object which has implemented methods needed by distributed coordinator such as `configure(session_config, cluster_spec, task_type, task_id)` which configures the strategy object for a specific task and `experimental_should_init` property which instructs the distribute coordinator whether to run init ops for a task. The distribute coordinator will make a copy of the `strategy` object, call its `configure` method and pass it to `worker_fn` as an argument. The `worker_fn` defines the training logic and is called under its own worker context which can be accessed to via `get_current_worker_context`. A worker context provides access to configurations for each task, e.g. the task_type, task_id, master target and so on. Since `worker_fn` will be called in a thread and possibly multiple times, caller should be careful when it accesses global data. For example, it is unsafe to define flags in a `worker_fn` or to define different environment variables for different `worker_fn`s. The `worker_fn` for the between-graph replication is defined as if there is only one worker corresponding to the `worker_fn` and possibly ps jobs. For example, when training with parameter servers, it assigns variables to parameter servers and all other operations to that worker. In the in-graph replication case, the `worker_fn` has to define operations for all worker jobs. Using a distribution strategy can simplify the `worker_fn` by not having to worry about the replication and device assignment of variables and operations. This method is intended to be invoked by high-level APIs so that users don't have to explicitly call it to run this coordinator. For those who don't use high-level APIs, to change a program to use this coordinator, wrap everything in a the program after global data definitions such as commandline flag definition into the `worker_fn` and get task-specific configurations from the worker context. The `cluster_spec` can be either passed by the argument or parsed from the "TF_CONFIG" environment variable. Example of a TF_CONFIG: ``` cluster = {'chief': ['host0:2222'], 'ps': ['host1:2222', 'host2:2222'], 'worker': ['host3:2222', 'host4:2222', 'host5:2222']} os.environ['TF_CONFIG'] = json.dumps({'cluster': cluster}) ``` If `cluster_spec` is not given in any format, it becomes local training and this coordinator will connect to a local session. For evaluation, if "evaluator" exists in the cluster_spec, a separate thread will be created to call `eval_fn` with its `task_type` set to "evaluator". If `eval_fn` is not defined, fall back to `worker_fn`. This implies that evaluation will be done on a single machine if there is an "evaluator" task. If "evaluator" doesn't exist in the cluster_spec, it entirely depends on the `worker_fn` for how to do evaluation. Args: worker_fn: the function to be called. The function should accept a `strategy` object and will be given access to a context object via a context manager scope. strategy: a DistributionStrategy object specifying whether it should run between-graph replicated training or not, whether to run init ops, etc. This object will also be configured given `session_config`, `cluster_spec`, `task_type` and `task_id`. eval_fn: optional function for "evaluator" task. If `eval_fn` is not passed in but a "evaluator" task is found in the `cluster_spec`, the `worker_fn` will be used for this task. eval_strategy: optional DistributionStrategy object for "evaluator" task. mode: in which mode this distribute coordinator runs. cluster_spec: a dict, ClusterDef or ClusterSpec specifying servers and roles in a cluster. If not set or empty, fall back to local training. task_type: the current task type, optional if this is a client. task_id: the current task id, optional if this is a client. session_config: an optional `tf.compat.v1.ConfigProto` object which will be passed to `strategy`'s `configure` method and used to create a session. rpc_layer: optional string, the protocol for RPC, e.g. "grpc". Raises: ValueError: if `cluster_spec` is supplied but not a dict or a ClusterDef or a ClusterSpec. Returns: In the client job, return the value returned by `worker_fn` if it is in-graph replication or INDEPENDENT_WORKER mode; return None otherwise. rrrSrNrrrr_cluster_resolverT)allow_soft_placementzRunning Distribute Coordinator with mode = %r, cluster_spec = %r, task_type = %r, task_id = %r, environment = %r, rpc_layer = %rz%Running local Distribute Coordinator.z4Skipped evaluation since `eval_fn` is not passed in.zf`eval_fn` is not passed in. The `worker_fn` will be used if an "evaluator" task exists in the cluster.zW`eval_strategy` is not passed in. No distribution strategy will be used for evaluation.)r>rPrQrRrSrzUnexpected coordinator mode: %r_std_server_startedFzUnexpected task_type: %r)*rrrrr=rrrhasattrrwrrPrQrSrr>rrryrzr<rwarningrr"r rexperimental_between_graphrrrrrr#rZrrrrrArr)rrOrrmoder>rPrQrRrSrrrcluster_resolverrrs rrun_distribute_coordinatorrssrjj T:;)mmK3) mT2+ ==B/L}}VR(H,,vy1iHLL'23g$;;LILx  "56((:: **I&&G **7iI"..K#002L"!Z%;%;&!. LL IJN 7KL  LL89y(D$n "-tT'4ooLM 000  ooMN"G  oo>? Y%%t,,    5 5(Hg)6 )7D D$Ix-$0.)M M0-0> 097D#'! #f kkm 111 84? @@  ooMN"G  oo>? .h .9F F!<GL   #Ixt$(.)E E ++- i)) )  i 'DD ill "3i?@@ kkmr)rfNN)NNNNNNr`)%r!rtrrr*rtensorflow.core.protobufrtensorflow.python.clientrtensorflow.python.distributerrtensorflow.python.platformrrytensorflow.python.trainingrr r r+rrr rr%r?rArrrrrrrr"rrrrrs6  /,G:<281 !  ,f ,#v#>7\7V\7J"$&*!"!JD,"&" #'" $ BJ(V@'4F\(,-1$3$E$E,0)-'+.2)/ur