ó ßüÚ\c@sÎddlmZddlZddlmZmZddlmZmZddl m Z ddl m Z ddl mZdd lmZd efd „ƒYZd efd „ƒYZdefd„ƒYZdS(iÿÿÿÿ(tgenNi(tFuturetdefault_client(tget_thread_identitytQueue(t to_serialize(tsync(t WrappedKey(t get_workertActorcBs’eZdZd d„Zd„Zd„Zed„ƒZed„ƒZ ed„ƒZ ed„ƒZ d„Z d „Z d „Zed „ƒZRS( sF Controls an object on a remote worker An actor allows remote control of a stateful object living on a remote worker. Method calls on this object trigger operations on the remote object and return ActorFutures on which we can block to get results. Examples -------- >>> class Counter: ... def __init__(self): ... self.n = 0 ... def increment(self): ... self.n += 1 ... return self.n >>> from dask.distributed import Client >>> client = Client() You can create an actor by submitting a class with the keyword ``actor=True``. >>> future = client.submit(Counter, actor=True) >>> counter = future.result() >>> counter Calling methods on this object immediately returns deferred ``ActorFuture`` objects. You can call ``.result()`` on these objects to block and get the result of the function call. >>> future = counter.increment() >>> future.result() 1 >>> future = counter.increment() >>> future.result() 2 cCs¬||_||_||_d|_|r?||_d|_niytƒ|_Wntk rkd|_nXyt ƒ|_t |ƒ|_Wntk r§d|_nXdS(N( t_clst_addresstkeytNonet_futuret_workert_clientRt ValueErrorRR(tselftclstaddressR tworker((s0lib/python2.7/site-packages/distributed/actor.pyt__init__3s           cCsd|jj|jfS(Ns(R t__name__R (R((s0lib/python2.7/site-packages/distributed/actor.pyt__repr__FscCst|j|j|jffS(N(R R R R (R((s0lib/python2.7/site-packages/distributed/actor.pyt __reduce__IscCs!|jr|jjS|jjSdS(N(Rtio_loopR(R((s0lib/python2.7/site-packages/distributed/actor.pyt_io_loopLs  cCs!|jr|jjS|jjSdS(N(Rt schedulerR(R((s0lib/python2.7/site-packages/distributed/actor.pyt_scheduler_rpcSs  cCsU|jr|jj|jƒS|jjr;|jj|jƒSt|jj|jƒSdS(N(RtrpcR Rtdirect_to_workerstProxyRPCR(R((s0lib/python2.7/site-packages/distributed/actor.pyt _worker_rpcZs   cCs*|jr|jjStƒ|jjkSdS(N(Rt asynchronousRRt thread_id(R((s0lib/python2.7/site-packages/distributed/actor.pyt _asynchronousds  cOs<|jr|jj|||ŽSt|jj|||ŽSdS(N(RRRtloop(Rtfunctargstkwargs((s0lib/python2.7/site-packages/distributed/actor.pyt_syncks cCsBttt|ƒƒƒ}|jd„t|jƒDƒƒt|ƒS(Ncss$|]}|jdƒs|VqdS(t_N(t startswith(t.0tattr((s0lib/python2.7/site-packages/distributed/actor.pys ts(tsettdirttypetupdateR tsorted(Rto((s0lib/python2.7/site-packages/distributed/actor.pyt__dir__rs cs£tˆjˆƒ}ˆjrFˆjjdkrFtdˆjjƒ‚nt|ƒrwtj|ƒ‡‡fd†ƒ}|Stj ‡‡fd†ƒ}ˆj |ƒSdS(Ntfinishedtpendings(Worker holding Actor was lost. Status: csytj‡‡‡‡fd†ƒ‰ˆjr1ˆƒStƒ‰tj‡‡fd†ƒ}ˆjj|ƒtˆˆjƒSdS(Nc 3s£yXˆjjdˆdˆjdgˆD]}t|ƒ^q%dd„ˆjƒDƒƒV}Wn1tk r‹ˆjr|ˆjVqŒtdƒ‚nXtj|dƒ‚dS(NtfunctiontactorR'R(cSs%i|]\}}t|ƒ|“qS((R(R,tktv((s0lib/python2.7/site-packages/distributed/actor.pys Šs s Unable to contact Actor's workertresult( R!t actor_executeR RtitemstOSErrorRRtReturn(targR;(R'R R(R(s0lib/python2.7/site-packages/distributed/actor.pytrun_actor_function_on_workerƒs     c3sˆƒV}ˆj|ƒdS(N(tput(tx(tqRA(s0lib/python2.7/site-packages/distributed/actor.pytwait_then_add_to_queuešs (Rt coroutineR$RRt add_callbackt ActorFuture(R'R(RE(R R(R'R(RDRAs0lib/python2.7/site-packages/distributed/actor.pyR&s!  c3s6ˆjjdˆdˆjƒV}tj|dƒ‚dS(Nt attributeR8R;(R!tactor_attributeR RR?(RC(R R(s0lib/python2.7/site-packages/distributed/actor.pytget_actor_attribute_from_worker§s (R5R6( tgetattrR RtstatusRtcallablet functoolstwrapsRRFR)(RR R-R&RK((R Rs0lib/python2.7/site-packages/distributed/actor.pyt __getattr__ws !"cCs |jjS(N(Rtclient(R((s0lib/python2.7/site-packages/distributed/actor.pyRR°sN(Rt __module__t__doc__R RRRtpropertyRRR!R$R)R4RQRR(((s0lib/python2.7/site-packages/distributed/actor.pyR s%       9R cBs eZdZd„Zd„ZRS(sQ An rpc-like object that uses the scheduler's rpc to connect to a worker cCs||_||_dS(N(RR (RRR((s0lib/python2.7/site-packages/distributed/actor.pyRºs cstj‡‡fd†ƒ}|S(Nc;s<ˆ|d<ˆjjdˆjd|ƒV}tj|ƒ‚dS(NtopRtmsg(RtproxyR RR?(RWR;(R R(s0lib/python2.7/site-packages/distributed/actor.pyR&¿s (RRF(RR R&((R Rs0lib/python2.7/site-packages/distributed/actor.pyRQ¾s(RRSRTRRQ(((s0lib/python2.7/site-packages/distributed/actor.pyR µs RHcBs,eZdZd„Zdd„Zd„ZRS(s Future to an actor's method call Whenever you call a method on an Actor you get an ActorFuture immediately while the computation happens in the background. You can call ``.result`` to block and collect the full result See Also -------- Actor cCs||_||_dS(N(RDR(RRDR((s0lib/python2.7/site-packages/distributed/actor.pyRÔs cCs?y |jSWn-tk r:|jjd|ƒ|_|jSXdS(Nttimeout(t_cached_resulttAttributeErrorRDtget(RRY((s0lib/python2.7/site-packages/distributed/actor.pyR;Øs   cCsdS(Ns ((R((s0lib/python2.7/site-packages/distributed/actor.pyRßsN(RRSRTRR R;R(((s0lib/python2.7/site-packages/distributed/actor.pyRHÈs   (ttornadoRRORRRRt compatibilityRRtprotocolRtutilsRt utils_commRRRR tobjectR RH(((s0lib/python2.7/site-packages/distributed/actor.pyts ©