ԋgW2ddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m ZddlmZddlmZddlmZddlZddlmZmZerdd lmZej6eZd Zd Zd Z d Z!GddZ"GddZ#dZ$y)) annotationsN)Callable)Queue) TYPE_CHECKING)Future)IOLoop)get_mp_contextwait_for)Selfc |j|g|y#t$r*}tjdt |sYd}~yd}~wwxYw)zQ Helper to silence "IOLoop is closing" exception on IOLoop.add_callback. zIOLoop is clos(ed|ing)N) add_callback RuntimeErrorresearchstr)loopfuncargsexcs 3lib/python3.12/site-packages/distributed/process.py_loop_add_callbackrsG$&& yy13s8< =s A  AA cH|js|j|yyN) cancelled set_result)futurevalues r#_future_set_result_unless_cancelledr)s!    %  cv|js|j|ytjd|y)Nz$Exception after Future was cancelled)exc_info)r set_exceptionloggererror)rrs r&_future_set_exception_unless_cancelledr%.s-    S! ;c Jrc ||i|}t|t||y#t$r}t|t||Yd}~yd}~wwxYwr)rr Exceptionr%)rrrrkwargsresrs r_call_and_set_futurer*5sQSD#F# 4!DfcR V 4!GQTUUVs A=AceZdZdZdZdZy) _ProcessStateFN)__name__ __module__ __qualname__is_alivepidexitcoderrr,r,@sH CHrr,ceZdZUdZded<ddZdZdZdZdd Z e d Z e d Z e dd Z e d ZdZddZddZddZdZddZdZedZedZedZedZej6dZy) AsyncProcessz A coroutine-compatible multiprocessing.Process-alike. All normally blocking methods are wrapped in Tornado coroutines. multiprocessing.Process_processNc |xsi}t|stdt|t|_|xst j d|_tjd\}|_ tj|j||||||jtjjf|_|j j"|_t'j(|t*|j |_t/|_t3|_d|_d|_|j;y)Nz#`target` needs to be callable, not F)instance)duplextargetnamer)callable TypeErrortyper,_statercurrent_loopr Pipe_keep_child_aliveProcess_rundaskconfig global_configr7r=_nameweakreffinalize_asyncprocess_finalizer_proc_finalizerPyQueue_watch_qr _exit_future_exit_callback_closed_start_threads)selfrr<r=rr(parent_alive_pipes r__init__zAsyncProcess.__init__Ns2A$v,AQRS S#o ;V^^U; 5C4D4I4IQV4I4W141&(0099!&& )) 1   ]]'' &// )4==   "H"  rcPd|jjd|jdS)N< >) __class__r-rKrVs r__repr__zAsyncProcess.__repr__xs&4>>**+1TZZL::rc2|jr tdy)Nz(invalid operation on closed AsyncProcess)rT ValueErrorr^s r _check_closedzAsyncProcess._check_closed{s <<GH H rc tj|jd|jzt j ||j |j|j|j|jf|_ d|j_ |jjd}t j|||j|_d|j_y)Nz#AsyncProcess %s watch message queuer;Tc*|jddiy)Nopstop) put_nowaitqs r stop_threadz0AsyncProcess._start_threads..stop_threads LL$ (rrhF) threadingThread_watch_message_queuer=rLrefr7rCrArQrR_watch_message_threaddaemonstartrM_thread_finalizeratexit)rVrjs rrUzAsyncProcess._start_threadss%.%5%5,,6B D!    !!  & "-1"") ""((* ) ")!1!1$ t}}!U(-%rcd|_|j|j||jj|yr)r7rSrRr)rVr2s r_on_exitzAsyncProcess._on_exits7    *    % $$X.rcjfd}tj|}d|_|jy)zP Immediately exit the process when parent_alive_pipe is closed. c jtd#t$rtjdYywxYw)Nz'unexpected state: should be unreachable)recvrEOFErroros_exit)rWsrmonitor_parentz@AsyncProcess._immediate_exit_when_closed..monitor_parentsC N"&&(##LMM   s ??)r<TN)rkrlrprq)clsrWr}ts ` r_immediate_exit_when_closedz(AsyncProcess._immediate_exit_when_closeds+  N,   N 3  rc|j|j|dtj_t j jt j j|d||i|y)N MainThreadold)priority) closerrkcurrent_threadr=rHrIupdaterJ)r~r<rr(rWrEinherit_configs rrGzAsyncProcess._runsb ! ''(9:*6   "' 4;;44nuUrc t j  fd} j}tj d d||d} | dk(rt ||d|nM| dk(rt ||dj n-| dk(rt ||djn | d k(ryJ|) Nc(jtjtjdzf}d|_|jd_j_tjddjy)Nz"AsyncProcess %s watch process joinr;T[z] created process with pid ) rqrkrlr5_watch_processrpr0r1r#debug)threadr=processrirselfrefstates r_startz1AsyncProcess._watch_message_queue.._starts{ MMO%%#229D@wq1F !FM LLN!EN EI LL1QC:599-H Irrz] got message rerqr terminatekillrf)reprr=getr#rr*rr) r~rrrrri exit_futurermsgrer=rs `` `` @@rrmz!AsyncProcess._watch_message_queues Oy~~ J J%%'C LL1QC~cW5 6TBW}$T3x=&A{"$T3x=':K:KLv$T3x=',,Gv# q!rcxt|}|j|jx}}|d}d|_||_|} |!t |j |j |d}|"tjd||jytjd||j|y#d}wxYw)NFzE[%s] process %r exit status was already read will report exitcode 255z#[%s] process %r exited with code %r) rjoinr2r0rrCrur#warningr1r) r~rrrrirr2original_exit_coderVs rrzAsyncProcess._watch_processs O (/(8(88%  H!y "4::t}}hGD  % NNW   LL>599h WDs #B55B9cv|jt}|jjd|d|S)zQ Start the child process. This method returns a future. rqrerrbrrQrgrVfuts rrqzAsyncProcess.starts3 h   3!?@ rcv|jt}|jjd|d|S)zTerminate the child process. This method returns a future. See also -------- multiprocessing.Process.terminate rrrrs rrzAsyncProcess.terminate!s3 "H   s!CD rcv|jt}|jjd|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 rrrrs rrzAsyncProcess.kill/s3 "H   #!>? rcK|j|jjJd|jjyt t j |j|d{y7w)z_ Wait for the child process to exit. This method returns a coroutine. Nzcan only join a started process)rbrAr1r2r asyncioshieldrR)rVtimeouts rrzAsyncProcess.join>sb {{*M,MM* ;;   + w~~d&7&78'BBBsA1A;3A94A;cZ|js|jd|_d|_yy)z Stop helper thread and release resources. This method returns immediately and does not ensure the child process has exited. NT)rTrrr7r^s rrzAsyncProcess.closeLs* ||  " " $ DMDLrctj|rJdt|sJd|jjJd||_y)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>rAr1rS)rVrs rset_exit_callbackzAsyncProcess.set_exit_callbackVsc..   ; : ; ~AAA~ KKOO # C B C #"rc.|jjSr)rAr0r^s rr0zAsyncProcess.is_alivegs{{###rc.|jjSr)rAr1r^s rr1zAsyncProcess.pidjs{{rc.|jjSr)rAr2r^s rr2zAsyncProcess.exitcodens{{###rc|jSr)rKr^s rr=zAsyncProcess.namers zzrc.|jjSrr7rpr^s rrpzAsyncProcess.daemonvs}}###rc&||j_yrr)rVrs rrpzAsyncProcess.daemonzs$ r)NNNr3N)r2intreturnNone)rr6)rzasyncio.Future[None]r)rVr rzCallable[[Self], None]rr)r-r.r/__doc____annotations__rXr_rbrUru classmethodrrGrmrrqrrrrrr0propertyr1r2r=rpsetterr3rrr5r5Fs &%(T;I.0/>    (6((TXX<    C #"$$$$$ ]]%%rr5c|jr* tjd||jyy#t$rYywxYw)Nzreaping stray process )r0r#inforOSError)procs rrNrNsG }}  KK07 8 NN    s(< AA)% __future__rrrloggingmultiprocessingr{rrkrLcollections.abcrqueuerrPtypingrtornado.concurrentrtornado.iolooprrHdistributed.utilsr r typing_extensionsr getLoggerr-r#rrr%r*r,r5rNr3rrrs" $" %! 6&   8 $! KS v%v%r r