B F.\S%@sddlmZmZmZddlmZddlZddlZddlZddl m Z ddl Z ddl m Z mZmZddlmZmZmZddlmZeeZGd d d eZGd d d eZdS) )print_functiondivisionabsolute_import) defaultdictN)gen)Future_get_global_clientClient)tokeysync thread_state) get_clientc@s`eZdZdZddZdddZddd Zejdd d Z dd d Z ejdddZ dddZ dS)QueueExtensionz 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 cCsl||_t|_t|_tdd|_|jj|j|j |j |j d|jj |j |jd||jjd<dS)NcSsdS)Nrrrr1lib/python3.7/site-packages/distributed/queues.py"sz)QueueExtension.__init__..) queue_create queue_put queue_get queue_qsize)zqueue-future-release queue_releasequeues) schedulerdictrclient_refcountrfuture_refcountZhandlersupdatecreateputgetqsizeZstream_handlersfuture_releaserelease extensions)selfrrrr__init__s  zQueueExtension.__init__NrcCs@||jkr*tjj|d|j|<d|j|<n|j|d7<dS)N)maxsizer)rtornadoQueuer)r%streamnameclientr'rrrr2s  zQueueExtension.createcCsl||jkrdS|j|d8<|j|dkrh|j|=|j|j}|j|=|jjdd|Dd|ddS)NrrcSs g|]}|ddkr|dqS)typervaluer).0drrr Csz*QueueExtension.release..zqueue-%s)keysr,)rrZ_queuerclient_releases_keys)r%r*r+r,Zfuturesrrrr#9s   zQueueExtension.releaseccsx|dk r@d|d}|j||fd7<|jj|gd|dn d|d}|dk r^tj|d}|j|j||dVdS) Nr)r-r.rzqueue-%s)r2r,Zmsgpack)seconds)timeout)rrZclient_desires_keysdatetime timedeltarr)r%r*r+keydatar,r5recordrrrrGs   zQueueExtension.putcCsN|j||fd8<|j||fdkrJ|jj|gd|d|j||f=dS)Nrrzqueue-%s)r2r,)rrr3)r%r+r8r,rrrr"Ss   zQueueExtension.future_releaseFc #sfdd|rj|}g}|dkrJx\|sF|V}||q(Wn:|dk r^d} t| x$t|D]} |V}||qhWfdd|D}t|n:|dk rtj |d}j|j|dV}|}t|dS) Ncsh|ddkrd|}|d}jj|}|dk r8|jnd}||d<|dkrd|jj|d<|jj|d <|S) z Add task status if known r-rr.NZloststateerred exception traceback)copyrZtasksr r;Zexception_blamer=r>)r:r8Ztsr;)r%rrprocess]s   z#QueueExtension.get..processTzNDask queues don't support simultaneous use of integer batch sizes and timeoutscsg|] }|qSrr)r/o)r@rrr1{sz&QueueExtension.get..)r4)r5) remptyr appendNotImplementedErrorrangerReturnr6r7) r%r*r+r,r5batchqoutr:msgir)r@r%rr Zs*       zQueueExtension.getcCs|j|S)N)rr!)r%r*r+r,rrrr!szQueueExtension.qsize)NNNr)NNN)NNNNNN)NNN)NNNNF)NNN) __name__ __module__ __qualname____doc__r&rr#r coroutinerr"r r!rrrrrs     (rc@seZdZdZdddZddZejddd Zdd d Z dd dZ ddZ ejdddZ ejddZ ddZddZddZdS) r)a Distributed Queue This allows multiple clients to share futures or small bits of data between each other with a multi-producer/multi-consumer queue. All metadata is sequentialized through the scheduler. Elements of the Queue must be either Futures or msgpack-encodable data (ints, strings, lists, dicts). All data is sent through the scheduler so it is wise not to send large objects. To share large objects scatter the data and share the future instead. .. warning:: This object is experimental and has known issues in Python 2 Examples -------- >>> 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 NrcCst|pt|_|pdtj|_|jjs4ttddrL|jj j |j|d|_ n$t |jj |jj j |j|dtj|_ dS)Nzqueue-Zon_event_loop_threadF)r+r')r r,uuidZuuid4hexr+Z asynchronousgetattrr rr_startedr ZlooprZmoment)r%r+r,r'rrrr&s    zQueue.__init__cstjfdd}|S)Nc3sjVtdS)N)rTrrFr)r%rr_szQueue.__await__.._)rrP __await__)r%rUr)r%rrVszQueue.__await__ccsFt|tr*|jjjt|j||jdVn|jjj|||jdVdS)N)r8r5r+)r9r5r+) isinstancerr,rrr r8r+)r%r.r5rrr_puts  z Queue._putcKs|jj|j|fd|i|S)z Put data into the queue r5)r,r rX)r%r.r5kwargsrrrrsz Queue.putFcKs|jj|jf||d|S)a 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 )r5rG)r,r _get)r%r5rGrYrrrr s z Queue.getcKs|jj|jf|S)z) Current number of elements in the queue )r,r _qsize)r%rYrrrr!sz Queue.qsizec#sRjjj|j|dV}fdd}|dkr6||}ntt||}t|dS)N)r5r+rGcsp|ddkrdt|djd|dd}|ddkrH|j|d|d jd j|dd n|d}|S) Nr-rr.Tr;)Zinformr;r<r=r>zqueue-future-release)opr+r8)rr,Z_stateZ set_error_send_to_schedulerr+)r0r.)r%rrr@s   zQueue._get..processF)r,rrr+listmaprrF)r%r5rGZrespr@resultr)r%rrZs    z Queue._getccs"|jjj|jdV}t|dS)N)r+)r,rrr+rrF)r%r`rrrr[sz Queue._qsizecCs$|jjdkr |jd|jddS)NZrunningr)r\r+)r,Zstatusr]r+)r%rrrcloses z Queue.closecCs|j|jjjfS)N)r+r,raddress)r%rrr __getstate__szQueue.__getstate__c Cs\|\}}yt|}|jj|ks"tWn$ttfk rHt|dd}YnX|j||ddS)NF)Zset_as_default)r+r,)rrrbAssertionErrorAttributeErrorr r&)r%r;r+rbr,rrr __setstate__szQueue.__setstate__)NNr)N)N)NF)NF)rLrMrNrOr&rVrrPrXrr r!rZr[rarcrfrrrrr)s    r))Z __future__rrr collectionsrr6ZloggingrQr(rZtornado.queuesr,rr r Zutilsr r r ZworkerrZ getLoggerrLZloggerobjectrr)rrrrs    v