U /e0@sddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m ZddlmZddlmZddlmZddlZddlmZeeZd d Zd d ZGd ddZedddZGdddZddZ dS)) annotationsN)Callable)Queue)TypeVar)Future)IOLoop)get_mp_contextc GsNz|j|f|Wn4tk rH}ztdt|s8W5d}~XYnXdS)zQ Helper to silence "IOLoop is closing" exception on IOLoop.add_callback. zIOLoop is clos(ed|ing)N)Z add_callback RuntimeErrorresearchstr)loopfuncargsexcr7/tmp/pip-unpacked-wheel-g426oqom/distributed/process.py_loop_add_callbacks rc OsRz|||}Wn0tk r>}zt||j|W5d}~XYnXt||j|dSN) ExceptionrZ set_exception set_result)r futurerrkwargsresrrrr_call_and_set_future$s  rc@seZdZdZdZdZdS) _ProcessStateFN)__name__ __module__ __qualname__is_alivepidexitcoderrrrr/sr_T_async_process AsyncProcess)boundc@seZdZUdZded<d8ddZdd Zd d Zd d ZdddddZ e ddZ e ddZ e ddddZ e ddZddZddd d!Zddd"d#Zd9d$d%Zd&d'Zd(d)dd*d+d,Zd-d.Zed/d0Zed1d2Zed3d4Zed5d6Zejd7d6ZdS):r#z A coroutine-compatible multiprocessing.Process-alike. All normally blocking methods are wrapped in Tornado coroutines. zmultiprocessing.Process_processNrc Cs|pi}t|s"tdt|t|_|p8tjdd|_tj dd\}|_ tj |j ||||||j t jjfd|_|jj|_t|t|j|_t|_t|_d|_d|_|dS)Nz#`target` needs to be callable, not F)instance)Zduplextargetnamer)callable TypeErrortyper_statercurrent_looprZPipe_keep_child_aliveProcess_rundaskconfig global_configr%r)_nameweakreffinalize_asyncprocess_finalizerZ_proc_finalizerPyQueue_watch_qr _exit_future_exit_callback_closed_start_threads)selfr r(r)rrparent_alive_piperrr__init__@s8  zAsyncProcess.__init__cCsd|jjd|jdS)N< >) __class__rr6r@rrr__repr__jszAsyncProcess.__repr__cCs|jrtddS)Nz(invalid operation on closed AsyncProcess)r> ValueErrorrGrrr _check_closedmszAsyncProcess._check_closedc Csptj|jd|jt||j|j|j|j |j fd|_ d|j _ |j dd}tj|||j d|_d|j_dS)Nz#AsyncProcess %s watch message queuer'TcSs|ddidS)Nopstop) put_nowaitqrrr stop_threadsz0AsyncProcess._start_threads..stop_threadrNF) threadingThread_watch_message_queuer)r7refr%r/r-r;r<Z_watch_message_threaddaemonstartr8_thread_finalizeratexit)r@rPrrrr?qs   zAsyncProcess._start_threadsintNone)r!returncCs*d|_|jdk r|||j|dSr)r%r=r<r)r@r!rrr_on_exits  zAsyncProcess._on_exitcs*fdd}tj|d}d|_|dS)zP Immediately exit the process when parent_alive_pipe is closed. cs8z Wntk r*tdYn XtddS)Nz'unexpected state: should be unreachable)recvEOFErroros_exitr rrArrmonitor_parents   z@AsyncProcess._immediate_exit_when_closed..monitor_parent)r(TN)rQrRrUrV)clsrArctrrbr_immediate_exit_when_closeds  z(AsyncProcess._immediate_exit_when_closedcCs@|||dt_tjjtjj|dd|||dS)N MainThreadold)priority) closerfrQcurrent_threadr)r3r4updater5)rdr(rrrAr0Zinherit_configrrrr2s   zAsyncProcess._run)processc stjfdd}}tdd||d} | dkrht||d|q(| dkrt||djq(| d krt||djq(| d krqq(d s(t|q(dS) Ncs^tjtjdfd}d|_|d_j_t ddjdS)Nz"AsyncProcess %s watch process joinr'T[z] created process with pid ) rVrQrRr#_watch_processrUrr loggerdebug)threadr)rmrOrselfrefstaterr_starts z1AsyncProcess._watch_message_queue.._startrnz] got message rKrVr terminatekillrLr) reprr)getrprqrrxryAssertionError) rdrurmr rvrOZ exit_futurerwmsgrKrrsrrSs z!AsyncProcess._watch_message_queuec Cst|}||j}}|dkr(d}d|_||_|}z|dk rTt|j|j|W5d}X|dkrxtd||j nt d||j |dS)NFzE[%s] process %r exit status was already read will report exitcode 255z#[%s] process %r exited with code %r) rzjoinr!rrr/r\rpwarningr rq) rdrurmrvrOrtr!Zoriginal_exit_coder@rrrros&  zAsyncProcess._watch_processcCs$|t}|jd|d|S)zQ Start the child process. This method returns a future. rVrKrrJrr;rMr@ZfutrrrrVszAsyncProcess.startzasyncio.Future[None])r[cCs$|t}|jd|d|S)zTerminate the child process. This method returns a future. See also -------- multiprocessing.Process.terminate rxrrrrrrrxs zAsyncProcess.terminatecCs$|t}|jd|d|S)zSend SIGKILL to the child process. On Windows, this is the same as terminate(). This method returns a future. See also -------- multiprocessing.Process.kill ryrrrrrrry!s zAsyncProcess.killcsJ||jjdk std|jjdk r,dStt|j|IdHdS)z_ Wait for the child process to exit. This method returns a coroutine. Nzcan only join a started process) rJr-r r|r!asynciowait_forZshieldr<)r@timeoutrrrr0s  zAsyncProcess.joincCs|js|d|_d|_dS)z Stop helper thread and release resources. This method returns immediately and does not ensure the child process has exited. NT)r>rWr%rGrrrrj>szAsyncProcess.closer"z"Callable[[_T_async_process], None])r@rr[cCs@t|rtdt|s"td|jjdks6td||_dS)z Set a function to be called by the event loop when the process exits. The function is called with the AsyncProcess as sole argument. The function may not be a coroutine function. z-exit callback may not be a coroutine functionz exit callback should be callableNz5cannot set exit callback when process already started)inspectiscoroutinefunctionr|r*r-r r=)r@rrrrset_exit_callbackHs  zAsyncProcess.set_exit_callbackcCs|jjSr)r-rrGrrrr[szAsyncProcess.is_alivecCs|jjSr)r-r rGrrrr ^szAsyncProcess.pidcCs|jjSr)r-r!rGrrrr!bszAsyncProcess.exitcodecCs|jSr)r6rGrrrr)fszAsyncProcess.namecCs|jjSrr%rUrGrrrrUjszAsyncProcess.daemoncCs ||j_dSrr)r@valuerrrrUns)NNNrN)N)rrr__doc____annotations__rBrHrJr?r\ classmethodrfr2rSrorVrxryrrjrrpropertyr r!r)rUsetterrrrrr#8s@  *  *        cCs>|r:ztd||Wntk r8YnXdS)Nzreaping stray process )rrpinforxOSError)procrrrr9ss  r9)! __future__rrrloggingmultiprocessingr`r rQr7collections.abcrqueuerr:typingrZtornado.concurrentrZtornado.iolooprr3Zdistributed.utilsr getLoggerrrprrrr"r#r9rrrrs0           =