ó ßüÚ\c@`sóddlmZmZmZddlmZddlZddlZddlZddl m Z ddl Z ddl m Z mZmZddlmZmZmZddlmZejeƒZd efd „ƒYZd efd „ƒYZdS( i(tprint_functiontdivisiontabsolute_import(t defaultdictN(tgeni(tFuturet_get_global_clienttClient(ttokeytsynct thread_state(t get_clienttQueueExtensioncB`s§eZdZd„Zd d d dd„Zd d d d„Zejd d d d d d d„ƒZ d d d d„Z ejd d d d e d„ƒZ d d d d„Z RS( sÈ An extension for the scheduler to manage queues This adds the following routes to the scheduler * queue_create * queue_release * queue_put * queue_get * queue_size cC`s©||_tƒ|_tƒ|_td„ƒ|_|jjji|jd6|j d6|j d6|j d6ƒ|jj ji|j d6|jd6ƒ||jjd"tt queue_createt queue_putt queue_gett queue_qsizesqueue-future-releaset queue_releasetqueues(t schedulertdictRtclient_refcountRtfuture_refcountthandlerstupdatetcreatetputtgettqsizetstream_handlerstfuture_releasetreleaset extensions(tselfR((s1lib/python2.7/site-packages/distributed/queues.pyt__init__s        icC`sR||jkr;tjjd|ƒ|j|>> from dask.distributed import Client, Queue # doctest: +SKIP >>> client = Client() # doctest: +SKIP >>> queue = Queue('x') # doctest: +SKIP >>> future = client.submit(f, x) # doctest: +SKIP >>> queue.put(future) # doctest: +SKIP See Also -------- Variable: shared variable between clients icC`s®|p tƒ|_|p(dtjƒj|_|jjsLttdt ƒrs|jj j d|jd|ƒ|_ n7t |jj|jj j d|jd|ƒtj|_ dS(Nsqueue-ton_event_loop_threadR)R%(RR*tuuidtuuid4thexR)t asynchronoustgetattrR RVRRt_startedR tloopRtmoment(R#R)R*R%((s1lib/python2.7/site-packages/distributed/queues.pyR$¨s  c`s%tj‡fd†ƒ}|ƒjƒS(Nc3`sˆjVtjˆƒ‚dS(N(R]RRK((R#(s1lib/python2.7/site-packages/distributed/queues.pyt_»s(RRUt __await__(R#R`((R#s1lib/python2.7/site-packages/distributed/queues.pyRaºscc`skt|tƒrA|jjjdt|jƒd|d|jƒVn&|jjjd|d|d|jƒVdS(NR9R4R)R:(t isinstanceRR*RRRR9R)(R#R-R4((s1lib/python2.7/site-packages/distributed/queues.pyt_putÂs #cK`s|jj|j|d||S(s Put data into the queue R4(R*R Rc(R#R-R4tkwargs((s1lib/python2.7/site-packages/distributed/queues.pyRÍscK`s"|jj|jd|d||S(sœ Get data from the queue Parameters ---------- timeout: Number (optional) Time in seconds to wait before timing out batch: boolean, int (optional) If True then return all elements currently waiting in the queue. If an integer than return that many elements from the queue If False (default) then return one item at a time R4RL(R*R t_get(R#R4RLRd((s1lib/python2.7/site-packages/distributed/queues.pyRÑs cK`s|jj|j|S(s) Current number of elements in the queue (R*R t_qsize(R#Rd((s1lib/python2.7/site-packages/distributed/queues.pyRßsc#`szˆjjjd|dˆjd|ƒV}‡fd†}|tkrR||ƒ}ntt||ƒƒ}tj|ƒ‚dS(NR4R)RLc`s¡|ddkr“t|dˆjdtd|dƒ}|ddkrd|jj|d|dƒnˆjjid d 6ˆjd 6|dd 6ƒn |d}|S( NR,RR-tinformR=R>R?R@squeue-future-releasetopR)R9(RR*RFt_statet set_errort_send_to_schedulerR)(R1R-(R#(s1lib/python2.7/site-packages/distributed/queues.pyREés& & ( R*RRR)RVtlisttmapRRK(R#R4RLtrespREtresult((R#s1lib/python2.7/site-packages/distributed/queues.pyReãs cc`s/|jjjd|jƒV}tj|ƒ‚dS(NR)(R*RRR)RRK(R#Ro((s1lib/python2.7/site-packages/distributed/queues.pyRfýscC`s:|jjdkr6|jjidd6|jd6ƒndS(NtrunningRRhR)(R*tstatusRkR)(R#((s1lib/python2.7/site-packages/distributed/queues.pytclosescC`s|j|jjjfS(N(R)R*Rtaddress(R#((s1lib/python2.7/site-packages/distributed/queues.pyt __getstate__scC`sz|\}}y(t|ƒ}|jj|ks3t‚Wn)ttfk r_t|dtƒ}nX|jd|d|ƒdS(Ntset_as_defaultR)R*(R RRstAssertionErrortAttributeErrorRRVR$(R#R=R)RsR*((s1lib/python2.7/site-packages/distributed/queues.pyt __setstate__ s  N(RRRSRTR5R$RaRRURcRRVRRReRfRrRtRx(((s1lib/python2.7/site-packages/distributed/queues.pyR'‹s     (t __future__RRRt collectionsRR7tloggingRXR&Rttornado.queuesR*RRRtutilsRR R tworkerR t getLoggerRRtloggertobjectR R'(((s1lib/python2.7/site-packages/distributed/queues.pyts    y