ó ßüÚ\c@`s9ddlmZmZmZddlZddlmZddlZddlZddl Z ddl Z ddl Z ddl m Z mZddlmZddlmZddlmZdd lmZejeƒZd „Zd „Zd efd „ƒYZdefd„ƒYZe j ƒZ!ej"d„ƒZ#dS(i(tprint_functiontdivisiontabsolute_importN(t timedeltai(tfinalizetQueue(t mp_context(tgen(tFuture(tIOLoopcG`sLy|j||ŒWn1tk rG}tjdt|ƒƒsH‚qHnXdS(sQ Helper to silence "IOLoop is closing" exception on IOLoop.add_callback. sIOLoop is clos(ed|ing)N(t add_callbackt RuntimeErrortretsearchtstr(tlooptfunctargstexc((s2lib/python2.7/site-packages/distributed/process.pyt_loop_add_callbacks cO`sSy|||Ž}Wn&tk r;}t||j|ƒnXt||j|ƒdS(N(t ExceptionRt set_exceptiont set_result(RtfutureRRtkwargstresR((s2lib/python2.7/site-packages/distributed/process.pyt_call_and_set_future!s t _ProcessStatecB`seZeZdZdZRS(N(t__name__t __module__tFalsetis_alivetNonetpidtexitcode(((s2lib/python2.7/site-packages/distributed/process.pyR,st AsyncProcesscB`s%eZdZddddid„Zd„Zd„Zd„Zd„Ze d„ƒZ e d„ƒZ e d„ƒZ e d „ƒZe d „ƒZd „Zd „Zejdd „ƒZd„Zd„Zd„Zed„ƒZed„ƒZed„ƒZed„ƒZejd„ƒZRS(s„ A coroutine-compatible multiprocessing.Process-alike. All normally blocking methods are wrapped in Tornado coroutines. c C`s÷t|ƒs(tdt|ƒfƒ‚ntƒ|_|pItjdtƒ|_t j dtƒ\}|_ t j d|j d|d|||||j fƒ|_tj|jƒ|jj|_tƒ|_tƒ|_d|_t|_|jƒdS(Ns%`target` needs to be callable, not %rtinstancetduplexttargettnameR(tcallablet TypeErrorttypeRt_stateR tcurrentRt_loopRtPipet_keep_child_alivetProcesst_runt_processt _danglingtaddR't_nametPyQueuet_watch_qRt _exit_futureR t_exit_callbackt_closedt_start_threads(tselfRR&R'RRtparent_alive_pipe((s2lib/python2.7/site-packages/distributed/process.pyt__init__8s          cC`sd|jj|jfS(Ns<%s %s>(t __class__RR5(R<((s2lib/python2.7/site-packages/distributed/process.pyt__repr__WscC`s|jrtdƒ‚ndS(Ns(invalid operation on closed AsyncProcess(R:t ValueError(R<((s2lib/python2.7/site-packages/distributed/process.pyt _check_closedZs c C`s¢tjd|jdd|jdtj|ƒ|j|j|j|j |j fƒ|_ t |j _ |j jƒd„}t||d|j ƒ|_t|j_dS(NR&R's#AsyncProcess %s watch message queueRcS`s|jidd6ƒdS(Ntstoptop(t put_nowait(tq((s2lib/python2.7/site-packages/distributed/process.pyt stop_threadnsRF(t threadingtThreadt_watch_message_queueR'tweakreftrefR2R-R+R7R8t_watch_message_threadtTruetdaemontstartRt _finalizerRtatexit(R<RG((s2lib/python2.7/site-packages/distributed/process.pyR;^s       cC`s<d|_|jdk r(|j|ƒn|jj|ƒdS(N(R R2R9R8R(R<R"((s2lib/python2.7/site-packages/distributed/process.pyt_on_exitvs c`s8‡fd†}tjd|ƒ}t|_|jƒdS(sP Immediately exit the process when parent_alive_pipe is closed. c`s?yˆjƒWntk r.tjdƒn Xtdƒ‚dS(Niÿÿÿÿs'unexpected state: should be unreachable(trecvtEOFErrortost_exitR ((R=(s2lib/python2.7/site-packages/distributed/process.pytmonitor_parentƒs  R&N(RHRIRNRORP(tclsR=RXtt((R=s2lib/python2.7/site-packages/distributed/process.pyt_immediate_exit_when_closed}s cC`sKxDtjjjjƒD]-}x$tj|ƒjD]}|jƒq/WqWdS(sw Python 2's logger's locks don't survive a fork event https://github.com/dask/distributed/issues/1491 N(tloggingtLoggertmanagert loggerDicttkeyst getLoggerthandlerst createLock(R'thandler((s2lib/python2.7/site-packages/distributed/process.pytreset_logger_locksscC`sA|jƒ|jƒ|j|ƒdtjƒ_|||ŽdS(Nt MainThread(RetcloseR[RHtcurrent_threadR'(RYR&RRR=R/((s2lib/python2.7/site-packages/distributed/process.pyR1§s    c `sâtˆƒƒ‰ˆƒj‰‡‡‡‡‡‡fd†}x¢tr݈jƒ}tjdˆ|fƒ|d} | dkr’t||d|ƒq<| dkr¸t||dˆjƒq<| dkrÈPq<ds<t|ƒ‚q<WdS( Nc `sˆjƒtjdtjddˆdˆˆˆˆfƒ}t|_|jƒtˆ_ˆjˆ_t j dˆˆjfƒdS(NR&R's"AsyncProcess %s watch process joinRs [%s] created process with pid %r( RPRHRIR#t_watch_processRNRORR!tloggertdebug(tthread(R'tprocessRFtrtselfreftstate(s2lib/python2.7/site-packages/distributed/process.pyt_start¿s        s[%s] got message %rRDRPRt terminateRCi( treprR'RNtgetRjRkRRrtAssertionError( RYRoRmRRpRFt exit_futureRqtmsgRD((R'RmRFRnRoRps2lib/python2.7/site-packages/distributed/process.pyRJ·s       cC`sŸt|ƒƒ}|jƒ|j}|dk s4t‚tjd||j|ƒt|_ ||_|ƒ}z)|dk rt |j |j |ƒnWdd}XdS(Ns#[%s] process %r exited with code %r( RstjoinR"R RuRjRkR!RRRR-RS(RYRoRmRpRFRnR"R<((s2lib/python2.7/site-packages/distributed/process.pyRiÛs      cC`s5|jƒtƒ}|jjidd6|d6ƒ|S(sO Start the child process. This method is a coroutine. RPRDR(RBRR7RE(R<tfut((s2lib/python2.7/site-packages/distributed/process.pyRPîs  cC`s5|jƒtƒ}|jjidd6|d6ƒ|S(sS Terminate the child process. This method is a coroutine. RrRDR(RBRR7RE(R<Ry((s2lib/python2.7/site-packages/distributed/process.pyRrùs  cc`s‘|jƒ|jjdk s(tdƒ‚|jjdk r>dS|dkrU|jVn8y!tjt d|ƒ|jƒVWntj k rŒnXdS(sZ Wait for the child process to exit. This method is a coroutine. scan only join a started processNtseconds( RBR+R!R RuR"R8Rt with_timeoutRt TimeoutError(R<ttimeout((s2lib/python2.7/site-packages/distributed/process.pyRxs   !cC`s,|js(|jƒd|_t|_ndS(s– Stop helper thread and release resources. This method returns immediately and does not ensure the child process has exited. N(R:RQR R2RN(R<((s2lib/python2.7/site-packages/distributed/process.pyRgs   cC`sCt|ƒstdƒ‚|jjdks6tdƒ‚||_dS(sÑ 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. s exit callback should be callables5cannot set exit callback when process already startedN(R(RuR+R!R R9(R<R((s2lib/python2.7/site-packages/distributed/process.pytset_exit_callback!s cC`s |jjS(N(R+R(R<((s2lib/python2.7/site-packages/distributed/process.pyR/scC`s |jjS(N(R+R!(R<((s2lib/python2.7/site-packages/distributed/process.pyR!2scC`s |jjS(N(R+R"(R<((s2lib/python2.7/site-packages/distributed/process.pyR"6scC`s|jS(N(R5(R<((s2lib/python2.7/site-packages/distributed/process.pyR':scC`s |jjS(N(R2RO(R<((s2lib/python2.7/site-packages/distributed/process.pyRO>scC`s||j_dS(N(R2RO(R<tvalue((s2lib/python2.7/site-packages/distributed/process.pyROBsN((RRt__doc__R R>R@RBR;RSt classmethodR[t staticmethodReR1RJRiRPRrRt coroutineRxRgR~RtpropertyR!R"R'ROtsetter(((s2lib/python2.7/site-packages/distributed/process.pyR#2s.      $   cC`sixbttƒD]T}|jr |jƒr y"tjd|fƒ|jƒWqatk r]qaXq q WdS(Nsreaping stray process %s(tlistR3RORRjtwarningRrtOSError(tproc((s2lib/python2.7/site-packages/distributed/process.pyt_cleanup_danglingJs ($t __future__RRRRRtdatetimeRR\RVR RHRKt compatibilityRRR6tutilsRttornadoRttornado.concurrentRttornado.ioloopR RaRRjRRtobjectRR#tWeakSetR3tregisterRŠ(((s2lib/python2.7/site-packages/distributed/process.pyts(       ÿ