AVhdZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z ddl Z ddlmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZej<Z ddlZ ddl!Z! ddl"Z#e#jHjKejLd gd Z'ejLdddgZ(ejLdgdZ)ejLdgdZ*dZ+dZ,Gdde-Z.Gddej^Z0Gdde-Z1e jdZ3dZ4dZ5Gdd e-Z6d!Z7d"Z8ed#g$Gd%d&e9Z:ed'g$Gd(d)e9Z;ed*g$Gd+d,e9Zed/g$ddd0e+ddfd1Z?da@ed2g$d3ZAdaBejZDd4ZEed5g$d6ZFy#e $rdZYxwxYw#e $rdZ!YwxYw#e $rYnwxYw)8z)Multi-process runner for testing purpose.N)logging)queue)tf2) v2_compat)multi_worker_util)multi_process_lib)context) test_util) tf_export_ProcessStatusInfo task_typetask_id is_successfulexc_info return_valueMultiProcessRunnerResultrstdoutTestEnvironmentrr cluster_spec rpc_layergrpc_fail_fast v2_enabledexecuting_eagerly visible_gpus Resourcesprocess_status_queueparent_to_sub_queuestreaming_pipe_wbarrierceZdZdZ ddZddZdZ ddZdZdZ dd Z d Z d Z d Z d ZdZdZdZefdZdZdZddZddZy)MultiProcessRunneraA utility class to start multiple processes to simulate a cluster. We need to use multiple processes to simulate a cluster in TF 2.0 tests because TF 2.0 has some process-global data structures that have to be separated by processes. We also need child processes to test out our fault tolerance because shutting down a standard TensorFlow server within its process is not supported. Note: the main test program that uses this runner class must run main program via `test_main` defined in this file. Using this runner in non-test binaries is not supported yet. This class is not thread-safe. Child processes will inherit TF2 behavior flag. Nctjrtjd|Jd|vr7t |ddkDr&t dj t |dtt|s t d||_ ||_ |xsd|_ ||_ ||_ ||_||_| |_||_| |_| |_| xsd|_|xsi|_| |_t t/j.j1d |_t5j6|_t/j:|_d |_tAjB|_"i|_#tI|_%g|_&tO|_(|jPjS|_*|jPjS|_+tYd |jj[D}|jPj]||_/|jPjS|_0d|_1y) a Instantiation of a `MultiProcessRunner`. Args: fn: Function to be run on child processes. This will be run on processes for all task types. cluster_spec: Dict for cluster spec. The utility function `tf.__internal__.distribute.multi_process_runner.create_cluster_spec` can be conveniently used to create such dict. The following is an example of cluster with three workers and two ps's. {"worker": ["worker0.example.com:2222", "worker1.example.com:2222", "worker2.example.com:2222"], "ps": ["ps0.example.com:2222", "ps1.example.com:2222"]} rpc_layer: RPC layer to use. Default value is 'grpc'. max_run_time: `None` or integer. If not `None`, child processes are forced to exit at approximately this many seconds after this utility is called. We achieve this through `signal.alarm()` api. Note that this is best effort at Python level since Python signal handler does not get executed when it runs lower level C/C++ code. So it can be delayed for arbitrarily long time. If any of the child process is still running when `max_run_time` is up, they will be force-terminated and an `UnexpectedSubprocessExitError` may be raised. If `None`, child processes are not forced to exit. grpc_fail_fast: Whether GRPC connection between processes should fail without retrying. Defaults to None, in which case the environment variable is not explicitly set. stream_output: True if the output/error from the subprocesses should be streamed to be printed in parent process' log. Defaults to True. return_output: If True, the output/error from the subprocesses should be collected to be attached to the resulting namedtuple returned from `join()`. The list of output can be retrieved via `stdout` attribute. Defaults to False. use_dill_for_args: Whether to use dill to pickle `args` and `kwargs`. dill can pickle more objects, but doesn't work with types in `multiprocessing` library like `Mutex`. daemon: Whether to start processes as daemons. dependence_on_chief: Whether to terminates the cluster if the chief exits. If auto_restart is True, it only terminates the cluster if the chief exits with a zero exit code. auto_restart: Whether to automatically restart processes that exit with non-zero exit code. share_gpu: Whether to share GPUs among workers. If False, each worker is assigned different GPUs in a roundrobin fashion. This should be True whenever possible for better test execution coverage; some situations that need it to be False are tests that runs NCCL. args: Positional arguments to be sent to `fn` run on subprocesses. kwargs: Keyword arguments to be sent to `fn` run on subprocesses. Raises: RuntimeError: if `multi_process_runner.test_main()` is not called. ValueError: if there are more than one chief in the `cluster_spec`. SkipTest: if thread sanitizer is enabled (which is incompatible with MPR). z:ThreadSanitizer is not compatible with MultiProcessRunner.NchiefzfIf chief exists in the cluster, there must be at most one chief. Current `cluster_spec` has {} chiefs.zfn is not a callablegrpcGPUFc32K|]}t|ywN)len).0 addressess a/home/dcms/DCMS/lib/python3.12/site-packages/tensorflow/python/distribute/multi_process_runner.py z.MultiProcessRunner.__init__..sNY#i.Ns)2r is_tsan_enabledunittestSkipTestr/ ValueErrorformat_check_initializationcallable_fn _cluster_spec _rpc_layer _max_run_time_grpc_fail_fast_stream_output_return_output_dependence_on_chief_use_dill_for_args_daemon _auto_restart_args_kwargs _share_gpur list_physical_devices _total_gpurenabled _v2_enabledr_executing_eagerly_joined threadingLock _process_lock _processesset _terminated_reading_threadsmanager_managerQueue_process_status_queue_parent_to_sub_queuesumvaluesBarrier_barrier_streaming_queue_watchdog_thread)selffnrr max_run_timer stream_output return_outputuse_dill_for_argsdaemondependence_on_chief auto_restart share_gpuargskwargspartiess r2__init__zMultiProcessRunner.__init__{sJ  "    F HH  ## #,3|G'<#=#A Js<#89: << B< - ..DH%D)6DO%D)D'D'D 3D/DDL%DDJ.handlerfs E ro) rQrRr7rNr<items enumeraterr>signalSIGALRMalarm)rarr1r_rs` r2rzMultiProcessRunner.startQs    H >?? ?@ @#'"4"4":":"<H )Y#I. HJGQ  3 3Iw G HH H % mmFNNG, ll4%%&& H Hs A6C##C,c|jr td|j5|jr td|jj D]5\}}t |D]"\}}||k(r||k(r|j||$7 dddt|||j|j|j|ji|jy#1swYRxYw)aStart the processes, with the specified task run in main process. This is similar to `start()` except that the task with task_type `as_task_type` and task_id `as_task_id` is run in the main process. This method is particularly useful when debugging tool such as `pdb` is needed in some specific task. Note that since this method is blocking until that specific task exits, additional actions would need a thread to be called: ```python def fn(): # user code to be run import pdb; pdb.set_trace() def follow_ups(): time.sleep(5) mpr.start_single_process( task_type='evaluator', task_id=0) mpr = multi_process_runner.MultiProcessRunner( fn, multi_worker_test_base.create_cluster_spec( has_chief=True, num_workers=1)) threading.Thread(target=follow_ups).start() mpr.start_in_process_as(as_task_type='chief', as_task_id=0) mpr.join() ``` Note that if `return_output=True`, the logs/stdout by task run by the main process is not available in result.stdout. Args: as_task_type: The task type to be run in the main process. as_task_id: The task id to be run in the main process. rrN) rRr7rQrNr<rrr_set_tf_configr=r;rFrG)ra as_task_type as_task_idrr1rrs r2start_in_process_asz&MultiProcessRunner.start_in_process_asmsJ  < ==  J ?@ @"&"4"4":":"<J )Y#I. JJGQ|+:0E  5 5i I JJ J<T-?-???$ DHHdjj)DLL)JJsAC 8C  C)c |j5|jr td|j|||||xsd|xsidddy#1swYyxYw)aCStarts a single process. This starts a process in the cluster with the task type, task id, and the process function (`fn`). If process function is `None`, the function provided at `__init__` will be used. If `cluster_spec` is `None`, the cluster spec provided at `__init__` will be used. TODO(rchao): It is meant that all subprocesses will be updated with the new cluster spec, but this has yet to be implemented. At this time only the newly started subprocess picks up this updated cluster spec. Args: task_type: The task type. task_id: The task id. cluster_spec: The cluster spec to be used on the newly started process. If `None`, the cluster spec provided at `__init__` will be used. fn: The process function to be run on the newly started process. If specified, specify `args` and `kwargs` as well. If `None`, the function provided at `__init__` will be used. args: Optional positional arguments to be supplied in `fn`. kwargs: Optional keyword arguments to be supplied in `fn`. rr+)rrbrkrlN)rQrNr7r)rarrrrbrkrls r2start_single_processz'MultiProcessRunner.start_single_processsj<     ?@ @ //  # zr2 0   s 7A  Acg} |j|jd##tj$rY|SwxYw)z Convert `queue.Queue` to `list`.Fblock)rgetrXEmpty)raqueue_to_convertlist_to_returns r2_queue_to_listz!MultiProcessRunner._queue_to_listsON .222?@ [[ s !'>>ci}|j|jD]}|||j|jf<|Sr.)rrYrr)rastatusesstatuss r2_get_process_statusesz(MultiProcessRunner._get_process_statusessEH%%d&@&@A<5;h  &..12< Oroc|j5|jj||fd}dddr |jSdS#1swYxYw)z:Returns the subprocess id given the task type and task id.N)rQrRrpidrarrrs r2get_process_idz!MultiProcessRunner.get_process_idsL  : //  y'2D 9a:1554::s AA c|j5|j||f}dddr |jSdS#1swYxYw)aZReturns the subprocess exit code given the task type and task id. Args: task_type: The task type. task_id: The task id. Returns: The subprocess exit code; `None` if the subprocess has not exited yet. Raises: KeyError: If the corresponding subprocess is not found with `task_type` and `task_id`. N)rQrRexitcoders r2get_process_exit_codez(MultiProcessRunner.get_process_exit_codesF   0 //9g. /a01::$$00s 7Ac*|j||duS)a Returns whether the subprocess still exists given the task type and id. Args: task_type: The task type. task_id: The task id. Returns: Boolean; whether the subprocess still exists. If the subprocess has exited, this returns False. N)r)rarrs r2process_existsz!MultiProcessRunner.process_existss  % %i 9T AAroc tjd|j5|jj dd}|r|j r|j |j dk(s |jsz|jjD]}|jd|j|jjD]}|j dddy|jr|d}|jjD]P\\}}}|j |j dk7s&d}tjd |||j||R|r dddvtd |jjDr dddy ddd#1swY xYw) aSimulates a cluster management system. - If auto_restart is True, it restarts processes that exit with a non-zero exit code. Note that when join() times out it overrides auto_restart to False. - If dependence_on_chief is True, it terminates all processes once the chief exits. If auto_restart is also True, it only terminates all processes if the chief exit with a zero exit code, otherwise it restarts the chief. This runs in self._watchdog_thread. Tr))r(rNr)timeoutFzRestarting failed %s-%dc38K|]}|jduywr.)r)r0rs r2r3z7MultiProcessRunner._process_watchdog..,sH!qzz%Hs)timesleeprQrRrrBrrEr\join_terminate_allrrinforall)rar(r has_failurerrs r2rz$MultiProcessRunner._process_watchdogs  jjm   ##L$7 T..5>>3M ^^q ););__++- ffQf     !__++- ffh  "   +)-)>)>)@ L%"y'Azz%!**/ kll4iI77 7K  L  38 Ht/E/E/GH H ;8 I9 s$CF6/=F6-F6=0F67+F66F?c|jD]`}t|tsJ|jr"|j ||j d_tj|j by)Nr)) r\ isinstancer r_get_mpr_resultr mpr_resultsixreraise)raprocess_statusesprocess_statuss r2_reraise_if_subprocess_errorz/MultiProcessRunner._reraise_if_subprocess_error/sg*113. (: ;; ;  ) )040D0D 1"- ^,,- .roc|rt|ts td|j5|jr tdd|_ddd|j j ||j jr|j5d|_dddtjd|jtj|j j t|j jr?tjd|j|j j |j}|j!|t#d j%||j'||j(j+D]*\\}}}tj,d |||j.,|j}|j!||j(j+D]a\\}}}|j.J|j.d kDs'||f|j0vs8t3d |||j.fz|j'|tj,d |j4D]}|j tj,dtj6d |j'|S#1swYxYw#1swYfxYw)a Joins all the processes with timeout. If any of the subprocesses does not exit approximately after `timeout` seconds has passed after `join` call, this raises a `SubprocessTimeoutError`. Note: At timeout, it uses SIGTERM to terminate the subprocesses, in order to log the stack traces of the subprocesses when they exit. However, this results in timeout when the test runs with tsan (thread sanitizer); if tsan is being run on the test targets that rely on timeout to assert information, `MultiProcessRunner.terminate_all()` must be called after `join()`, before the test exits, so the subprocesses are terminated with SIGKILL, and data race is removed. Args: timeout: optional integer or `None`. If provided as an integer, and not all processes report status within roughly `timeout` seconds, a `SubprocessTimeoutError` exception will be raised. If `None`, `join` never times out. Returns: A `MultiProcessRunnerResult` object, which has two attributes, `return_value` and `stdout`. `return_value` always contains a list of return values from the subprocesses, although the order is not meaningful. If `return_output` argument is True at `__init__`, `stdout` is available that contains a list of all messages from subprocesses' stdout and stderr. Raises: SubprocessTimeoutError: if not all processes report status approximately within `timeout` seconds. When this is raised, a `MultiProcessRunnerResult` object can be retrieved by `SubprocessTimeoutError`'s mpr_result attribute, which has the same structure as above 'Returns' section describes. UnexpectedSubprocessExitError: If any of the subprocesses did not exit properly (for example, they exit on SIGTERM or SIGKILL signal). When this is raised, a `MultiProcessRunnerResult` object can be retrieved by `UnexpectedSubprocessExitError`'s mpr_result attribute, which has the same structure as above 'Returns' section describes. If `max_run_time` is not `None`, it is expected that some subprocesses may be force-killed when `max_run_time` is up, and this is raised in those cases. Exception: if there is an Exception propagated from any subprocess. When this is raised, a `MultiProcessRunnerResult` object can be retrieved by `UnexpectedSubprocessExitError`'s mpr_result attribute, which has the same structure as above 'Returns' section describes. z'`timeout` must be an integer or `None`.z)MultiProcessRunner can't be joined twice.TNFz8Timeout when joining for child processes. Terminating...)sigzPTimeout when waiting for child processes to print stacktrace. Sending SIGKILL...zOne or more subprocesses timed out, where timeout was set to {}s. Please change the `timeout` argument for `MultiProcessRunner.join()` or `multi_process_runner.run()` if it should be adjusted.z%s-%d exit code: %srz@Subprocess %s-%d exited with exit code %s. See logs for details.zJoining log reading threads.zJoined log reading threads.)rintr7rQrNr`rrrErerrorrrSIGTERM_FORCE_KILL_WAIT_SECrrSubprocessTimeoutErrorr8rrRrrrrTUnexpectedSubprocessExitErrorrUr)rarrrrrrs r2rzMultiProcessRunner.join7s^z'3/ @ AA   DEEdl  w' %%'   #"# mmNO V^^,   !56   ' ' ) = >  ""$335 ''(89 " &'-fWo   / 0  22$(??#8#8#:JGa ll()WajjIJ113%%&67$(??#8#8#:4GaZZ ## # **q.i19I9II+ N'1::. /  !1 24 4 4 LL/0'' kkm LL./ LLO    0 11m##sK2K?2K<?L c|j|j}g}|jD]*}|j|j |j,t ||S)N)rr)rr_r\rrr)rarr return_valuesrs r2rz"MultiProcessRunner._get_mpr_resultsd  !6!6 7FM*113:  $ $ 0^889: $6 NNrocl|j5|jj||fd}|tdj |||j j ||f|jjdj |||jdddy#1swYyxYw)aTerminates the process with `task_type` and `task_id`. If auto_retart=True, the terminated task will be restarted unless the chief has already exited with zero exit code. Args: task_type: the task type. task_id: the task id. Nz{}-{} does not existterminate {} {}) rQrRrr7r8rTaddrZr~rrs r2 terminatezMultiProcessRunner.terminates    //  y'2D 9a /66y'JKK Iw/0 ##$5$<$< W%ffhs BB**B3c|xsttdtj}|jj D]\\}}}|j t jd||- tj|j||jj||ft jd|||y#t$rt jd||YwxYw)zTerminates all subprocesses. The caller is required to hold self._process_lock. Args: sig: the signal used to terminate the process. The default is SIGKILL. SIGKILLNz*%s-%d has already exited. Not terminating.z %s-%d terminated with signal %r.z/Attempting to kill %s-%d but it does not exist.)getattrrrrRrrrrrykillrrTrProcessLookupError)rarrrrs r2rz!MultiProcessRunner._terminate_alls  ;FNN;C#'??#8#8#: )Ga   A9 ) s i12 7G  ) ) F ))s,AC C'&C'ch|j5|j|dddy#1swYyxYw)zTerminates all subprocesses.N)rQr)rars r2rz MultiProcessRunner.terminate_alls.   #s(1) NNNTFTFTFTNNNN)NNNNr.)__name__ __module__ __qualname____doc__rnrqrrrrrrrrrrrr_DEFAULT_TIMEOUT_SECrrrrrr+ror2r&r&ks $ "!"!%#'!~!@* 4 9=.20426 C$J'82*n)-" $"& (T  %$ B+Z..g2RO*)4ror&c(eZdZdZfdZdZxZS)rzKA modified `multiprocessing.Process` that can set up environment variables.c ~tt| di|||_t |d|_|j |_y)Nrunr+)superrrn _test_envr _actual_run_run_with_setenvr)rarrl __class__s r2rnz_Process.__init__s8 (D",V,DNtU+D$$DHroc|j}|j&t|jtjd<|j rCdj |j Dcgc] }t|c}tjd<t|j|j|j|j|jScc}w)NGRPC_FAIL_FAST,CUDA_VISIBLE_DEVICES) rrstrryenvironrrrrrrrr)raris r2rz_Process._run_with_setenvs~~H*%()@)@%Abjj!"+.88#00 1a3q6 1,3bjj'(8%%x'7'79N9N%%'     2s$C)rrrrrnr __classcell__rs@r2rrsS% rorcFeZdZdZej dZdZdZdZ y)rz-Represents a callable to run in a subprocess.c#K|r"tj5ddddytj5ddddy#1swYyxYw#1swYyxYwwr.)r eager_mode graph_mode)rars r2 _runtime_modez_ProcFunc._runtime_modes\          s2A!A A!A A! AA!AA!c : |jjjd}|jdst dj ||dj ||k(rn<|jjj |tjd |jjj t||dd d tjdy #tj$rtjdYtwxYw) z>A function that regularly checks messages from parent process.TFrrzUnrecognized message: {}rr)g?Nr ) _resourcesr r startswithr7r8r~rrrXrrr ry_exit)rarrmessages r2_message_checking_funcz _ProcFunc._message_checking_funcs //55999F!!+.5<|jA|jB5tE|j,|j.|||}|jjFjI||jJstMjN|jP|jSdddt#jTdy#1swYxYw) zAThe wrapper function that actually gets run in child process(es).NT)chainz5Subprocess with PID %d (%s, %d) is now being started.z TF_CONFIG: %r TF_CONFIG)rrkrgr)+rr"r^rloads faulthandlerenableregisterrrrset_stderrthresholdDEBUGrydup2r!r{rrrgetpidrrrrrOrrrrrenable_v2_behaviorrr_run_containedrr~rrrrrexit) rarrrbrkrlrfrrs r2__call__z_ProcFunc.__call__,s  DO&&H BB ZZ dzz&!fFNN$7  .GGI & & - - /1B1B1DEGGI & & - - /1B1B1DE ))+C LLH###X%5%57 LL"**["9:**  ("2"2 3UW""$  H66 7 H..0@0@"d"$d oo**..t4   T]]# $HHQK%s =BKK%N) rrrr contextlibcontextmanagerrrrr)r+ror2rrs,5 < -;rorc:tD]}|jyr.)_active_pool_runnersshutdown)pools r2_shutdown_all_pool_runnersr0ps"dMMOrochttjdk\xrdtjdvS)z*Returns whether the test is run under OSS.r)bazelr)r/rargvr+ror2is_ossr4us' SXX!  6388A; 66roc2eZdZdZddZdZdZdZd dZy) MultiProcessPoolRunnerzA utility class to start a process pool to simulate a cluster. It's similar to MultiProcessRunner, but uses a pool of processes to avoid the expensive initialization cost of Tensorflow. Ncttj|||_||_||_i|_d|_y)aCreates a multi-process pool runner. Args: cluster_spec: Dict for cluster spec. The following is an example of cluster with three workers. {"worker": ["worker0.example.com:2222", "worker1.example.com:2222", "worker2.example.com:2222"]} initializer: a callable to called at the startup of worker processes. share_gpu: Whether to share GPUs among workers. If False, each worker is assigned different GPUs in a roundrobin fashion. Raises: RuntimeError: if `multi_process_runner.test_main()` is not called. ValueError: if there are more than one chief in the `cluster_spec`. N)r-rr< _initializerrH_conn_runner)rar initializerrjs r2rnzMultiProcessPoolRunner.__init__s7"T"%D#DDODJDLroc$|jyr.)r.rs r2__del__zMultiProcessPoolRunner.__del__s MMOrocL|jjD]}|ji|_|j# |jj d|_yy#t j $rt$r }tjd|Yd}~@d}~wwxYw)zShuts down the worker pool.Nz@Ignoring exception when shutting down MultiProcessPoolRunner: %s) r9r\rr:rr5r6 Exceptionr exception)raconnes r2r.zMultiProcessPoolRunner.shutdowns !!# jjlDJ ||  dl      N    sA''B#BB#c ttjdtd|jd|j |_|jr/tj|jtj}nd}|jjD]i\}}t|D]V\}}tjd\}}||j||f<|j j||t ||||fXky) zStarts the worker pool.Nrcyr.r+r+ror2z/MultiProcessPoolRunner._start..sroF)rbrrfrjTr)rbrk)rr5r6r&r<rHr:r8rrrrrrr9r_pool_runner_worker)rar;rr1rrconn1conn2s r2_startzMultiProcessPoolRunner._starts  |    = ??& ''// #DL  JJt00$2G2GHkk $ 2 2 8 8 :; 9!),;*'1&++4+8 u+0 Iw'( ))  "Wk59 * ;;;rocttj|j|j t j |t j}|jjD]}|j||xsg|xsif g}|jjD]?\\}}}tjd|| |j|jAg}|D]d} t'| t(sJ| j*st-j.| j0| j2J|j| j2f|S#t $r|j#t%dwxYw)aRuns `fn` with `args` and `kwargs` on all jobs. Args: fn: The function to be run. args: Optional positional arguments to be supplied in `fn`. kwargs: Optional keyword arguments to be supplied in `fn`. Returns: A list of return values. z!Waiting for the result from %s-%dzAUnexpected EOF. Worker process may have died. Please report a bug)r9rProcessr:rIrrrr9r\sendrrrrrecvEOFErrorr. RuntimeErrorrr rrrrr) rarbrkrlrArrrrrs r2rzMultiProcessPoolRunner.runsf || kkm B-- .B !!#0 iiTZR2./0&*jj&6&6&8 2"Gd ll6 7K2 , 2M*: (: ;; ;  ) ) ^,,-  $ $ 0^889 : 2 12 2 2s E%%%F )NTr) rrrrrnr=r.rIrr+ror2r6r6zs  0";8)ror6cl|rtj|}| |j\}}}tj|}t |||||}t j jt jj|j|#t$rYywxYw)aFunction that runs on the workers in a pool. It listens for callables to run and returns the result until `conn` is closed. It captures the exceptions during executing the callable and return it through `conn`. Args: task_type: the task type. task_id: the task index. initializer: a callable to execute during startup. conn: a multiprocessing.Connection object to listen for tasks and send results. N) rrrMrNr'rrrxrrL)rrr;rArbrkrlrs r2rFrFs**[)KM  b$ BB )Wb$ ?DJJJJIIdO     sB'' B32B3cd}d}d} ||i|}d}t|||||S#t$r'tj}t|||||cYSwxYw)aRuns `fn` with `args` and `kwargs`. The function returns _ProcessStatusInfo which captures the return value and the exception. Args: task_type: the task type. task_id: the task index. fn: the function to be run. args: optional positional arguments to be supplied in `fn`. kwargs: optional keyword arguments to be supplied in `fn`. Returns: a _ProcessStatusInfo. FNTr )r r?rr)rrrbrkrlrrrs r2r'r's"-, (#t&v&LM #!  ## #||~H #!  ###s"-AAzC__internal__.distribute.multi_process_runner.SubprocessTimeoutError)v1c"eZdZdZfdZxZS)ralAn error that indicates there is at least one subprocess timing out. When this is raised, a namedtuple object representing the multi-process run result can be retrieved by `tf.__internal__.distribute.multi_process_runner.SubprocessTimeoutError`'s `mpr_result` attribute. See `tf.__internal__.distribute.multi_process_runner.run` for more information. c:tt| |||_yr.)rrrnrramsgrrs r2rnzSubprocessTimeoutError.__init__Is $05 DOrorrrrrnrr s@r2rr<s!!rorzJ__internal__.distribute.multi_process_runner.UnexpectedSubprocessExitErrorc"eZdZdZfdZxZS)ra|An error indicating there is at least one subprocess with unexpected exit. When this is raised, a namedtuple object representing the multi-process run result can be retrieved by `tf.__internal__.distribute.multi_process_runner .UnexpectedSubprocessExitError`'s `mpr_result` attribute. See `tf.__internal__.distribute.multi_process_runner.run` for more information. c:tt| |||_yr.)rrrnrrUs r2rnz&UnexpectedSubprocessExitError.__init__\s '7< DOrorWr s@r2rrNs!!rorz@__internal__.distribute.multi_process_runner.NotInitializedErrorceZdZdZy)NotInitializedErroraAn error indicating `multi_process_runner.run` is used without init. When this is raised, user is supposed to call `tf.__internal__.distribute.multi_process_runner.test_main()` within `if __name__ == '__main__':` block to properly initialize `multi_process_runner.run`. N)rrrrr+ror2r[r[as ror[cBtjs tdy)Nz`multi_process_runner` is not initialized. Please call `tf.__internal__.distribute.multi_process_runner.test_main()` within `if __name__ == '__main__':` block in your python module to properly initialize `multi_process_runner`.)r initializedr[r+ror2r9r9ns%  & & (  " ## )rocn|||dd}|||d<tj|tjd<y)z#Set TF_CONFIG environment variable.)typeindex)clustertaskNrr)jsonrryr)rrrrtf_config_dicts r2rrxsB."+N; JJ~6"**[roz0__internal__.distribute.multi_process_runner.runFc ht|||||||}|j|j|S)aRun `fn` in multiple processes according to `cluster_spec`. Given a callable `fn`, `tf.__internal__.distribute.multi_process_runner.run` launches multiple processes, each of which runs `fn`. These processes are referred to as "subprocesses" or "child processes". Each of those subprocesses will have their `TF_CONFIG` environment variable set, according to `cluster_spec` and their task types. The stdout of the subprocesses are streamed to the main process' and thus available in logs (if `stream_output` is True), with [type-id] prefix. `tf.__internal__.distribute.multi_process_runner.run` will block until all subprocesses have successfully exited, and return a namedtuple object that represents the run result. This object has a `return_value` attribute, which is a list that contains subprocesses `fn`'s return values, for those subprocesses that successfully returned from `fn`. The order of `return_value` list is not meaningful. If an optional arg `return_output` (default to False) is set to True, the namedtuple object will have an additional attribute `stdout`, which is a list containing the stdout of the subprocesses. If any subprocess' `fn` ends up raising an error, that error will be reraised from `tf.__internal__.distribute.multi_process_runner.run`, and the aforementioned namedtuple object will be available through the exception's `mpr_result` attribute. This utility is used for simulating running TensorFlow programs across multiple task types, and each of the task type may contain more than one task (except for "chief" where more than one task is prohibited). Test coverage of multi-worker training is the main application of this utility, where code written for multi-worker training can be realistically covered in unit tests. Any test module that uses `tf.__internal__.distribute.multi_process_runner.run()` must call `tf.__internal__.distribute.multi_process_runner.test_main()` instead of regular `test.main()` inside `if __name__ == '__main__':` block for proper initialization. Args: fn: Function to be run on child processes. This will be run on processes for all task types. cluster_spec: Dict for cluster spec. The utility function `tf.__internal__.distribute.multi_process_runner.create_cluster_spec` can be conveniently used to create such dict. The following is an example of cluster with three workers and two ps's. {"worker": ["worker0.example.com:2222", "worker1.example.com:2222", "worker2.example.com:2222"], "ps": ["ps0.example.com:2222", "ps1.example.com:2222"]} rpc_layer: RPC layer to use. Default value is 'grpc'. max_run_time: `None` or integer. If not `None`, child processes are forced to exit at approximately this many seconds after this utility is called. We achieve this through `signal.alarm()` api. Note that this is best effort at Python level since Python signal handler does not get executed when it runs lower level C/C++ code. So it can be delayed for arbitrarily long time. If any of the child process is still running when `max_run_time` is up, they will be force-terminated and an `tf.__internal__.distribute.multi_process_runner .UnexpectedSubprocessExitError` may be raised. If `None`, child processes are not forced to exit. return_output: If True, the output/error from the subprocesses should be collected to be attached to the resulting namedtuple returned from this utility. The list of output can be retrieved via `stdout` attribute. Defaults to False. timeout: optional integer or `None`. If provided as an integer, and not all processes report status within roughly `timeout` seconds, a `tf.__internal__.distribute.multi_process_runner.SubprocessTimeoutError` exception will be raised. If `None`, `tf.__internal__.distribute.multi_process_runner.run` never times out. Defaults to the constant `_DEFAULT_TIMEOUT_SEC` defined in `multi_process_runner` module. args: Positional arguments to be sent to `fn` run on subprocesses. kwargs: Keyword arguments to be sent to `fn` run on subprocesses. Returns: A namedtuple object, which has two attributes, `return_value` and `stdout`. `return_value` always contains a list of returnvalues from the subprocesses, although the order is not meaningful. If `return_output` argument is True, `stdout` is available that contains a list of all messages from subprocesses' stdout and stderr, and the order is mostly chronological. Raises: RuntimeError: if `tf.__internal__.distribute.multi_process_runner.test_main()` is not called in test's `if __name__ == '__main__':` block. ValueError: if there are more than one chief in the `cluster_spec`. tf.__internal__.distribute.multi_process_runner.SubprocessTimeoutError: if not all processes report status approximately within `timeout` seconds. When this is raised, a namedtuple object can be retrieved by `tf.__internal__.distribute.multi_process_runner.SubprocessTimeoutError`'s `mpr_result` attribute, which has the same structure as above 'Returns' section describes. tf.__internal__.distribute.multi_process_runner .UnexpectedSubprocessExitError: If any of the subprocesses did not exit properly (for example, they exit on SIGTERM or SIGKILL signal). When this is raised, a namedtuple object can be retrieved by `tf.__internal__.distribute.multi_process_runner .UnexpectedSubprocessExitError`'s `mpr_result` attribute, which has the same structure as above 'Returns' section describes. If `max_run_time` is not `None`, it is expected that some subprocesses may be force-killed when `max_run_time` is up, and this is raised in those cases. Exception: if there is an Exception propagated from any subprocess. When this is raised, a namedtuple object can be retrieved by `tf.__internal__.distribute.multi_process_runner .UnexpectedSubprocessExitError` `mpr_result` attribute, which has the same structure as above 'Returns' section describes. Examples: ```python class SimpleMultiProcessTest(tf.test.TestCase): def test_simple_printing_and_return(self): def fn(): resolver = tf.distribute.cluster_resolver.TFConfigClusterResolver() # This will print "[chief-0]: Task type: chief , task id: 0" # for chief, for example. logging.info('Task type: %s, task id: %d', resolver.task_type, resolver.task_id) return resolver.task_type result = tf.__internal__.distribute.multi_process_runner.run( fn=fn, cluster_spec=( tf.__internal__ .distribute.multi_process_runner.create_cluster_spec( has_chief=True, num_workers=2))) assert sorted(result.return_value) == ['chief', 'worker', 'worker'] def test_error_from_fn(self): def fn(): resolver = tf.distribute.cluster_resolver.TFConfigClusterResolver() raise ValueError('Task type {}, task id {} is errors out'.format( resolver.task_type, resolver.task_id)) with self.assertRaisesRegex(ValueError, 'Task type worker, task id 0 is errors out'): cluster_spec = ( tf.__internal__.distribute.multi_process_runner.create_cluster_spec( num_workers=1)) tf.__internal__.distribute.multi_process_runner.run( fn=fn, cluster_spec=cluster_spec) if __name__ == '__main__': tf.__internal__.distribute.multi_process_runner.test_main() ``` )rcrerkrl)r&rr) rbrrrcrerrkrlrunners r2rrs>J !   & ,,. W roz8__internal__.distribute.multi_process_runner.get_barrierc0t tdtS)aReturns a `multiprocessing.Barrier` for `multi_process_runner.run`. `tf.__internal__.distribute.multi_process_runner.get_barrier()` returns a `multiprocessing.Barrier` object which can be used within `fn` of `tf.__internal__.distribute.multi_process_runner` to wait with `barrier.wait()` call until all other tasks have also reached the `barrier.wait()` call, before they can proceed individually. Note that all tasks (subprocesses) have to reach `barrier.wait()` call to proceed. Currently it is not supported to block on only a subset of tasks in the cluster. Example: ```python def fn(): some_work_to_be_done_by_all_tasks() tf.__internal__.distribute.multi_process_runner.get_barrier().wait() # The barrier guarantees that at this point, all tasks have finished # `some_work_to_be_done_by_all_tasks()` some_other_work_to_be_done_by_all_tasks() result = tf.__internal__.distribute.multi_process_runner.run( fn=fn, cluster_spec=( tf.__internal__ .distribute.multi_process_runner.create_cluster_spec( num_workers=2))) ``` Returns: A `multiprocessing.Barrier` for `multi_process_runner.run`. zbarrier is not defined. It is likely because you are calling get_barrier() in the main process. get_barrier() can only be called in the subprocesses.)r^r7r+ror2 get_barrierrh;s$L    /roctt5ttj atcdddS#1swYyxYw)ajReturns the multiprocessing manager object for concurrency tools. The manager object is useful as it controls a server process that holds the python objects that can be shared across processes. This can be used for parent-subprocess communication: ```python manager = multi_process_runner.manager() some_event_happening_in_subprocess = manager.Event() mpr = multi_process_runner.MultiProcessRunner(fn, cluster_spec, args=(some_event_happening_in_subprocess,)) mpr.start() some_event_happening_in_subprocess.wait() # Do something that only should after some event happens in subprocess. ``` Note that the user of multi_process_runner should not create additional `multiprocessing.Manager()` objects; doing so can result in segfault in some cases. This method should only be called after multi_process_runner.test_main() is called. N)r9 _manager_lockrWrManagerr+ror2rVrVns90 ((*h s ;Az6__internal__.distribute.multi_process_runner.test_maincttjdddfd}ttjdd|t j y)aMain function to be called within `__main__` of a test file. Any test module that uses `tf.__internal__.distribute.multi_process_runner.run()` must call this instead of regular `test.main()` inside `if __name__ == '__main__':` block, or an error will be raised when `tf.__internal__.distribute.multi_process_runner.run()` is used. This method takes care of needed initialization for launching multiple subprocesses. Example: ```python class MyTestClass(tf.test.TestCase): def testSomething(self): # Testing code making use of # `tf.__internal__.distribute.multi_process_runner.run()`. if __name__ == '__main__': tf.__internal__.distribute.multi_process_runner.test_main() ``` __main__tearDownModuleNc.tyyr.)r0)old_tear_down_modulesr2tear_down_modulez#test_main..tear_down_modules '(ro)rrmodulessetattrr test_main)rqrps @r2rtrtsK6!Z!8:J!%'  #++j !#35EFror.)Gr collectionsr*rcryrrrOrr5weakrefabslrr six.movesrrXtensorflow.pythonrtensorflow.python.compatrtensorflow.python.distributerrtensorflow.python.eagerr tensorflow.python.frameworkr tensorflow.python.util.tf_exportr rr ImportErrorrtblib.pickling_supporttblibpickling_supportinstall namedtupler rrrrrobjectr&rKrrWeakSetr-r0r4r6rFr'rOrrr[r9rrr^rhrWrPrjrVrtr+ror2rs0   $!.::+16#33    ",[++IK 2;112L3A82LN)+(():= #K " ";1   o o d ((2qqn'w( 7 xVxv:'#T  % !\ ! !  , !L ! !  F2O,O# 7 ="E$ mFmb  E"M+N+\   @ CK# L# C+,  $s62G$7G2<H$G/.G/2G=<G=H H