ó ßüÚ\c@`s)ddlmZmZmZddlmZddlZddlZddlm Z ddl Zyddl m Z Wn!e k r—ddlm Z nXddlmZmZmZddlmZdd lmZmZdd lmZejeƒZd efd „ƒYZd efd„ƒYZdS(i(tprint_functiontdivisiontabsolute_import(t defaultdictN(tgen(tmergei(tFuturet_get_global_clienttClient(ttime(ttokeyt log_errors(t get_clienttVariableExtensioncB`seZdZd„Zdddddd„Zejd„ƒZddddd„Z ejddddd„ƒZ ejdddd„ƒZ RS(sª An extension for the scheduler to manage queues This adds the following routes to the scheduler * variable-set * variable-get * variable-delete cC`s¬||_tƒ|_ttƒ|_ttjjƒ|_ tjjƒ|_ |jj j i|jd6|j d6ƒ|j|jjd<|j|jjd<||jjd>> from dask.distributed import Client, Variable # doctest: +SKIP >>> client = Client() # doctest: +SKIP >>> x = Variable('x') # doctest: +SKIP >>> x.set(123) # docttest: +SKIP >>> x.get() # docttest: +SKIP 123 >>> future = client.submit(f, x) # doctest: +SKIP >>> x.set(future) # doctest: +SKIP See Also -------- Queue: shared multi-producer/multi-consumer queue between clients icC`s2|p tƒ|_|p(dtjƒj|_dS(Ns variable-(RR'R?R@RAR/(R"R/R'tmaxsize((s3lib/python2.7/site-packages/distributed/variable.pyR#–scc`s_t|tƒr;|jjjdt|jƒd|jƒVn |jjjd|d|jƒVdS(NR0R/R1(t isinstanceRR'RRR R0R/(R"R%((s3lib/python2.7/site-packages/distributed/variable.pyt_setšscK`s|jj|j||S(s° Set the value of this variable Parameters ---------- value: Future or object Must be either a Future or a msgpack-encodable value (R'tsyncRQ(R"R%tkwargs((s3lib/python2.7/site-packages/distributed/variable.pyR£scc`sé|jjjd|d|jd|jjƒV}|ddkrÌt|d|jdtd|dƒ}|dd kr’|jj|d |d ƒn|jj id d 6|jd6|dd6|dd6ƒn |d}t j |ƒ‚dS(NR8R/R'R$RR%tinformR:R;R<R=svariable-future-releasetopR0R7( R'RRR/tidRtTruet_statet set_errort_send_to_schedulerRRE(R"R8tdR%((s3lib/python2.7/site-packages/distributed/variable.pyt_get­s&    cK`s|jj|jd||S(s Get the value of this variable R8(R'RRR\(R"R8RS((s3lib/python2.7/site-packages/distributed/variable.pyRÂscC`s:|jjdkr6|jjidd6|jd6ƒndS(sn Delete this variable Caution, this affects all clients currently pointing to this variable. trunningRRUR/N(R'tstatusRZR/(R"((s3lib/python2.7/site-packages/distributed/variable.pyR ÆscC`s|j|jjjfS(N(R/R'Rtaddress(R"((s3lib/python2.7/site-packages/distributed/variable.pyt __getstate__ÎscC`sz|\}}y(t|ƒ}|jj|ks3t‚Wn)ttfk r_t|dtƒ}nX|jd|d|ƒdS(Ntset_as_defaultR/R'(R RR_tAssertionErrortAttributeErrorRtFalseR#(R"R:R/R_R'((s3lib/python2.7/site-packages/distributed/variable.pyt __setstate__Ñs  N(RJRKRLR)R#RRMRQRR\RR R`Re(((s3lib/python2.7/site-packages/distributed/variable.pyRNvs    ( t __future__RRRt collectionsRtloggingR?RRt tornado.lockstcytoolzRt ImportErrorttoolzR'RRRtmetricsR tutilsR R tworkerR t getLoggerRJtloggertobjectR RN(((s3lib/python2.7/site-packages/distributed/variable.pyts     _