B F.\+@sddlmZmZmZddlZddlmZddlZddlZddl Z ddl Z ddl Z ddl m Z mZddlmZddlmZddlmZdd lmZeeZd d Zd d ZGdddeZGdddeZe Z!ej"ddZ#dS))print_functiondivisionabsolute_importN) timedelta)finalizeQueue) mp_context)gen)Future)IOLoopc GsNy|j|f|Wn4tk rH}ztdt|s8Wdd}~XYnXdS)zQ Helper to silence "IOLoop is closing" exception on IOLoop.add_callback. zIOLoop is clos(ed|ing)N)Z add_callback RuntimeErrorresearchstr)loopfuncargsexcr2lib/python3.7/site-packages/distributed/process.py_loop_add_callbacks rc OsRy|||}Wn0tk r>}zt||j|Wdd}~XYnXt||j|dS)N) ExceptionrZ set_exception set_result)rfuturerrkwargsZresrrrr_call_and_set_future!s  rc@seZdZdZdZdZdS) _ProcessStateFN)__name__ __module__ __qualname__is_alivepidexitcoderrrrr,src@seZdZdZddddifddZddZdd Zd d Zd d Ze ddZ e ddZ e ddZ e ddZe ddZddZddZejd-ddZddZd d!Zd"d#Zed$d%Zed&d'Zed(d)Zed*d+Zejd,d+ZdS). AsyncProcessz A coroutine-compatible multiprocessing.Process-alike. All normally blocking methods are wrapped in Tornado coroutines. NrcCst|stdt|ft|_|p0tjdd|_tj dd\}|_ tj |j ||||||j fd|_ t|j |j j|_t|_t|_d|_d|_|dS)Nz%`target` needs to be callable, not %rF)instance)Zduplex)targetnamer)callable TypeErrortyper_stater Zcurrent_loopr ZPipe_keep_child_aliveZProcess_run_process _danglingaddr'_namePyQueue_watch_qr _exit_future_exit_callback_closed_start_threads)selfrr&r'rrparent_alive_piperrr__init__8s"    zAsyncProcess.__init__cCsd|jj|jfS)Nz<%s %s>) __class__rr2)r9rrr__repr__WszAsyncProcess.__repr__cCs|jrtddS)Nz(invalid operation on closed AsyncProcess)r7 ValueError)r9rrr _check_closedZszAsyncProcess._check_closedc Csntj|jd|jt||j|j|j|j |j fd|_ d|j _ |j dd}t|||j d|_d|j_dS)Nz#AsyncProcess %s watch message queue)r&r'rTcSs|ddidS)Nopstop) put_nowait)qrrr stop_threadgsz0AsyncProcess._start_threads..stop_thread)rCF) threadingThread_watch_message_queuer'weakrefrefr/r,r+r4r5Z_watch_message_threaddaemonstartr _finalizeratexit)r9rDrrrr8^s zAsyncProcess._start_threadscCs*d|_|jdk r|||j|dS)N)r/r6r5r)r9r#rrr_on_exitos  zAsyncProcess._on_exitcs*fdd}tj|d}d|_|dS)zP Immediately exit the process when parent_alive_pipe is closed. cs8y Wntk r*tdYn XtddS)Nz'unexpected state: should be unreachable)ZrecvEOFErroros_exitr r)r:rrmonitor_parent{s   z@AsyncProcess._immediate_exit_when_closed..monitor_parent)r&TN)rErFrJrK)clsr:rStr)r:r_immediate_exit_when_closedvs  z(AsyncProcess._immediate_exit_when_closedcCs:x4tjjjD]"}xt|jD] }|q"WqWdS)zw Python 2's logger's locks don't survive a fork event https://github.com/dask/distributed/issues/1491 N)loggingZLoggerZmanagerZ loggerDictkeys getLoggerZhandlersZ createLock)r'Zhandlerrrrreset_logger_locksszAsyncProcess.reset_logger_lockscCs2||||dt_|||dS)NZ MainThread)rZcloserVrEZcurrent_threadr')rTr&rrr:r-rrrr.s   zAsyncProcess._runc stjfdd}xt}td|f|d} | dkrft||d|q*| dkrt||djq*| dkrPq*d s*t|q*WdS) NcsZtjtjdfd}d|_|d_j_t djfdS)Nz"AsyncProcess %s watch process join)r&r'rTz [%s] created process with pid %r) rKrErFr$_watch_processrJr!r"loggerdebug)Zthread)r'processrCrselfrefstaterr_startsz1AsyncProcess._watch_message_queue.._startz[%s] got message %rr@rKr terminaterAr)reprr'getr]r^rrdAssertionError) rTrar_rrbrCZ exit_futurercmsgr@r)r'r_rCr`rarbrrGs z!AsyncProcess._watch_message_queuecCspt|}||j}|dk s$ttd||j|d|_||_|}z|dk rbt|j |j |Wdd}XdS)Nz#[%s] process %r exited with code %rF) rejoinr#rgr]r^r"r!rr,rN)rTrar_rbrCr`r#r9rrrr\s   zAsyncProcess._watch_processcCs$|t}|jd|d|S)zO Start the child process. This method is a coroutine. rK)r@r)r?r r4rB)r9futrrrrKszAsyncProcess.startcCs$|t}|jd|d|S)zS Terminate the child process. This method is a coroutine. rd)r@r)r?r r4rB)r9rjrrrrdszAsyncProcess.terminateccst||jjdk std|jjdk r,dS|dkr>|jVn2ytt|d|jVWntj k rnYnXdS)zZ Wait for the child process to exit. This method is a coroutine. Nzcan only join a started process)Zseconds) r?r+r"rgr#r5r Z with_timeoutr TimeoutError)r9Ztimeoutrrrris  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)r7rLr/)r9rrrr[szAsyncProcess.closecCs.t|std|jjdks$td||_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 be a coroutine function. z exit callback should be callableNz5cannot set exit callback when process already started)r(rgr+r"r6)r9rrrrset_exit_callbackszAsyncProcess.set_exit_callbackcCs|jjS)N)r+r!)r9rrrr!%szAsyncProcess.is_alivecCs|jjS)N)r+r")r9rrrr"(szAsyncProcess.pidcCs|jjS)N)r+r#)r9rrrr#,szAsyncProcess.exitcodecCs|jS)N)r2)r9rrrr'0szAsyncProcess.namecCs|jjS)N)r/rJ)r9rrrrJ4szAsyncProcess.daemoncCs ||j_dS)N)r/rJ)r9valuerrrrJ8s)N)rrr __doc__r;r=r?r8rN classmethodrV staticmethodrZr.rGr\rKrdr coroutinerir[rlr!propertyr"r#r'rJsetterrrrrr$2s.   #          r$c CsVxPttD]D}|jr |r ytd|f|Wq tk rLYq Xq WdS)Nzreaping stray process %s)listr0rJr!r]ZwarningrdOSError)procrrr_cleanup_dangling@s rw)$Z __future__rrrrMZdatetimerrWrQrrErHZ compatibilityrrr3Zutilsr Ztornador Ztornado.concurrentr Ztornado.ioloopr rYrr]rrobjectrr$WeakSetr0registerrwrrrrs*