\c@`sddlmZmZmZddlmZddlZddlmZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlZddlmZddlmZmZddlmZdd lmZmZmZdd lmZmZmZmZdd l m!Z!dd l"m#Z#dd l$m%Z%ddl&m'Z'ddl(m)Z)ddl*m+Z+m,Z,m-Z-m.Z.m/Z/ddl0m1Z1m2Z2m3Z3m4Z4ej5e6Z7de#fdYZ8de9fdYZ:dS(i(tprint_functiontdivisiontabsolute_import(t timedeltaN(tEmpty(tgen(tIOLoopt TimeoutError(tEventi(tget_address_hosttget_local_address_fortunparse_host_port(trpct RPCClosedtCommClosedErrortcoerce_to_address(ttime(t ServerNode(t AsyncProcess(tenable_proctitle_on_children(tSecurity(tget_ipt mp_contexttsilence_loggingtjson_load_robusttPeriodicCallback(t_ncorestruntparse_memory_limittWorkertNannycB`sveZdZdZdZddddddddddeeeddddgddddddZdZ e j ddZ e dZe d Ze j dd Zd Zdd Ze j dd dZe j ddZe j dd edZdZdZdZe j dZe dZdZe j ddddZRS(s A process to manage worker processes The nanny spins up Worker processes, watches then, and kills or restarts them as necessary. isdask-worker-spacetautocK`s|r"t|}|d|_ni|dkrXtjjdrXtjjd|_n3|dkrvt||_nt||f|_||_|pt|_ | |_ | |_ ||_ ||_ ||_||_|dkrtn||_|pi|_||_||_tjjd|_|p6t|_t|jtsTt|jjd|_|jjd|_||_|ptj|_ t!|jd|j|_"||_#| |_$| |_%t&|_'t(| |j |_)|rt*d|n||_+i|j,d6|j-d6|j.d 6|j/d 6|j0d 6}t1t2|j3|d |j d|j|j)rt4|j5d d |j }||j6d(tworker_addressR1(RT((s0lib/python2.7/site-packages/distributed/nanny.pyt__repr__si cc`s|jdkrdS|jj}|dkr/dStjtttf}y6tjt d||j j d|jd|VWn|k rnXdS(NtsecondsR tquiet_exceptions( tprocessR,R^RRRtEnvironmentErrorR t with_timeoutRREt unregister(RTttimeoutR^tallowed_errors((s0lib/python2.7/site-packages/distributed/nanny.pyt _unregisters       cC`s|jdkrdS|jjS(N(RbR,R^(RT((s0lib/python2.7/site-packages/distributed/nanny.pyR^scC`s|jdkrdS|jjS(N(RbR,t worker_dir(RT((s0lib/python2.7/site-packages/distributed/nanny.pyRiscc`s |s=|jt|jjd|jt|j|_ntt|trt t|jj|_|j|j|fd|jn(|j|d|jt|j|_t j d|j|j V}|dkr|j std|_n |jV|jtj|dS(s2 Start nanny, start local process, start watching RAs Start Nanny at: %rtrunningN(tlistenR RER RAR tipR=tintRtloggertinfoR$R^R>RSRMtstart_periodic_callbacksRtReturn(RTt addr_or_porttresponse((s0lib/python2.7/site-packages/distributed/nanny.pyt_starts$ "     cC`s|jjS(N(Rtt __await__(RT((s0lib/python2.7/site-packages/distributed/nanny.pyRuscC`s|jj|j|dS(N(RDt add_callbackRt(RTRr((s0lib/python2.7/site-packages/distributed/nanny.pytstartsicc`st|_|jdkr*tjdn|jj|}|jjdd||jjV|j ||jjVdS(s Kill the local worker process Blocks until both the process is down and the scheduler is properly informed tOKRfg?N( tFalseRJRbR,RRqRDRR%Rh(RTtcommRftdeadline((s0lib/python2.7/site-packages/distributed/nanny.pyR%s  %c!c`s|jr|j}n,|jjd}|jjt||j}|jdkrKtd|j d|j d|j d|j di|j d6d|jd |jd |jd |jd |jd |jd|jd|jd|jd|jd|j}|j|jtdtd|d|fd |jd|jd|jd|j|_nt |_!|jry,t"j#t$d|j|jj%V}Wqt"j&k r|j'd|jVt"j(dqXn|jj%V}t"j(|dS(su Start a local worker process Blocks until the process is up and the scheduler is properly informed iRUR1RBRFt service_portstnannyRGRKR2R4R3RLR5R6R7R<R:t worker_argsR9tworker_start_argston_exitR!R8R`Rfs timed outN()RRtlistenert bound_addresstprefixR R0RbR,tdictR+R1RBRFtportRGRKR2R4R3RLR5R6R7R<R:tupdateR9t WorkerProcessttuplet_on_exitRR8RIRJRRdRRwRRMRq(RTRzt start_argthostR9tresult((s0lib/python2.7/site-packages/distributed/nanny.pyR$sT                         &c#`st}tjfd}y!tjtd||VWn0tjk rttjdtjdnXtjddS(Nc3`s,jdk r(jVjVndS(N(RbR,R%R$((RT(s0lib/python2.7/site-packages/distributed/nanny.pyt_!s R`s,Restart timed out, returning before finisheds timed outRx( RRt coroutineRdRRRnterrorRq(RTRzRft executor_waitRwR((RTs0lib/python2.7/site-packages/distributed/nanny.pyR&s ! cC`s|jdkrdS|jj}|dkr/dSytj|j}Wntjk r\dSX|jj}||j }|j r||j krt j dd|j |j ndS(sE Track worker's memory. Restart if it goes above terminate fraction RjNs.Worker exceeded %d%% memory budget. Restartingid(RSRbR,tpsutiltProcesstpidt NoSuchProcesst memory_infotrssRKR;RntwarningR'(RTRbtprocR)tfrac((s0lib/python2.7/site-packages/distributed/nanny.pyRP/s    cC`s|jdk o|jjdkS(NRj(RbR,RS(RT((s0lib/python2.7/site-packages/distributed/nanny.pytis_aliveCscO`st|||S(N(R(RTtargstkwargs((s0lib/python2.7/site-packages/distributed/nanny.pyRFscc`s|jdkry|jjd|jVWn/ttfk r[|js\|jVdSnXy:|jdkr|jrt j d|j VqnWqt k rt j ddtqXndS( NtclosingtclosedR sRestarting workers1Failed to restart worker after its process exitedtexc_info(RR(RR(RSREReR^RcRR2RMRJRnRR$t ExceptionRRI(RTtexitcode((s0lib/python2.7/site-packages/distributed/nanny.pyRIs     cC`s|jo|jjS(N(RbR(RT((s0lib/python2.7/site-packages/distributed/nanny.pyR]scO`s#tjddd|j||S(Ns'Worker._close has moved to Worker.closet stackleveli(twarningstwarnRM(RTRR((s0lib/python2.7/site-packages/distributed/nanny.pyt_closeasicc`s|jdkr!tjdnd|_tjd|j|jy'|jdk rm|j d|VnWnt k rnXd|_|j j |j jd|_tjddS(s; Close the worker process, stop all comms. RRRxsClosing Nanny at %rRfN(RR(RSRRqRnRoR tstopRbR,R%RR RMREt close_rpc(RTRzRftreport((s0lib/python2.7/site-packages/distributed/nanny.pyRMes       N((t__name__t __module__t__doc__R,RbRSRIRyROR_RRRhtpropertyR^RiRtRuRwR%R$R&RPRRRRRRM(((s0lib/python2.7/site-packages/distributed/nanny.pyR%sb F   8    RcB`seZdZejdZdZdZdZe dZ dZ ejde dZ ejd Zed ZRS( cC`sgd|_||_||_||_||_||_d|_||_||_ d|_ d|_ dS(NR*( RSRLR~R9RRR,RbRR8RiR^(RTR~R9RRLRR!R8((s0lib/python2.7/site-packages/distributed/nanny.pyRO|s          cc`st|jdkr+tj|jn|jdkr]|jjVtj|jntj|_}tj|_ t j j }t d|jdtd|jd|jd|jd|jd |jd |j d |d |jd |j |_t|j_|jj|jt|_t|_d|_|jjV|j|V}|sstj|jn|d|_|d|_ |jst!d|_|jj"|j#tj|jdS(s7 Ensure the worker process is started. RjtstartingttargetRR~R9RRLt init_result_qt child_stop_qtuidRR8R tdirN($RRSRRqRjtwaitRtQueueRRtuuidtuuid4thexRt_runRR~R9RRLRR8RbRItdaemontset_exit_callbackRRtstoppedRwt_wait_until_connectedR^RiR>tsetRM(RTtinit_qRtmsg((s0lib/python2.7/site-packages/distributed/nanny.pyRwsJ                 cC`s!||jk rdS|jdS(N(Rbt mark_stopped(RTR((s0lib/python2.7/site-packages/distributed/nanny.pyRscC`sV|dk st|dkr)d|fS|dkrCd||fSd|| fSdS(Nis.Worker process %d was killed by unknown signalis'Worker process %d exited with status %ds)Worker process %d was killed by signal %d(R,R>(RTRR((s0lib/python2.7/site-packages/distributed/nanny.pyt_death_messages    cC`s|jdk o|jjS(N(RbR,R(RT((s0lib/python2.7/site-packages/distributed/nanny.pyRscC`s&|jr"|jjr"|jjSdS(N(RbRRR,(RT((s0lib/python2.7/site-packages/distributed/nanny.pyRscC`s|jdkr|jj}|dk s-t|dkra|j|jj|}tj|nd|_|j j |jj d|_ d|_ d|_|jrtjj|jrtj|jdtnd|_|jdk r|j|qndS(NRit ignore_errors(RSRbRR,R>RRRnRRRRMRRRitostpathtexiststshutiltrmtreeRIR(RTtrR((s0lib/python2.7/site-packages/distributed/nanny.pyRs"         icc`sKtj}|j|}|jdkr/dS|jdkrP|jjVdS|jdksetd|_|j}|jj idd6t d||jd d 6|d 6|jj x0|j r|j|krt jd VqW|j rGtjd |y|jVWqGtk rC}tjd|qGXndS(s Ensure the worker process is stopped, waiting at most *timeout* seconds before terminating it abruptly. RNtstoppingRRjRtopig?RfRg?s4Worker process still alive after %d seconds, killings!Failed to kill worker process: %s(RRj(RRCRRSRRR>RbRtputtmaxRMRRtsleepRnRR'RR(RTRfRRDR{Rbte((s0lib/python2.7/site-packages/distributed/nanny.pyR%s2      !  cc`sd}xtr|jdkr"dSy|jj}Wn"tk rYtj|Vq nX|d|krpq nd|krtjd|d|j j V|q tj |q WdS(Ng?RRt exceptions/Failed while trying to start worker process: %s( RIRSRt get_nowaitRRRRnRRbtjoinRq(RTRtdelayR((s0lib/python2.7/site-packages/distributed/nanny.pyRs"    c `s<tjj|yddlm} Wntk r7nX| |rUtj|ntj tj | ||t j dt fdfd} tjd| dd} t | _| jt j fd } yj| Wn!tk r'ntk r7nXdS( Ni(tinitialize_worker_processic 3`s9z'jdtdtd|d|VWdjXdS(NRR}RRf(RMRyR(RfR(RDR!(s0lib/python2.7/site-packages/distributed/nanny.pytdo_stopMs  c`sqxjtrlyjdd}Wntk r2qXj|jddksXtj|PqWdS(si Wait for an incoming stop message and then stop the worker cleanly. RfiRRN(RIR/RRMtpopR>Rv(R(RRRD(s0lib/python2.7/site-packages/distributed/nanny.pyt watch_stop_qYs   RRGsNanny stop queue watchc3`syjVWnEtk rY}tjdjid6|d6jnZXjsitjijd6jd6d6jj Vtj ddS(sK Try to start worker and inform parent of outcome. sFailed to start workerRRR Rs Worker closedN( RtRRnRRRMR R>RBtwait_until_closedRo(R(RRR!R(s0lib/python2.7/site-packages/distributed/nanny.pyRms "  (RtenvironRtdask.multiprocessingRt ImportErrorRntsetLevelRtclear_instancet make_currentRRRIt threadingtThreadRRwtrun_syncRtKeyboardInterrupt(tclsR~R9RRLRRRR8RRRttR((RRRRDRR!Rs0lib/python2.7/site-packages/distributed/nanny.pyR0s2        !  (RRRORRRwRRRRRRRIR%Rt classmethodR(((s0lib/python2.7/site-packages/distributed/nanny.pyR{s 1   &(;t __future__RRRtdatetimeRtloggingtmultiprocessing.queuesRRRRRRRR-ttornadoRttornado.ioloopRRt tornado.locksRRzR R R tcoreR R RRtmetricsRtnodeRRbRt proctitleRR<RtutilsRRRRRR!RRRRt getLoggerRRnRtobjectR(((s0lib/python2.7/site-packages/distributed/nanny.pyts4        "("W