U /eB&@sddlmZddlZddlZddlZddlmZddlmZm Z ddl m Z m Z ddl mZmZeeZGdddZGd d d ZdS) ) annotationsN) defaultdict)parse_timedelta stringify)ClientFuture) get_client get_workerc@sTeZdZdZddZdddZddd Zdd d Zdd d ZdddZ dddZ dS)QueueExtensionzAn 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 cCs`||_t|_t|_tdd|_|jj|j|j |j |j d|jj |j |jddS)NcSsdS)Nrr r r 6/tmp/pip-unpacked-wheel-g426oqom/distributed/queues.py z)QueueExtension.__init__..) queue_create queue_put queue_get queue_qsize)queue-future-release queue_release) schedulerdictqueuesclient_refcountrfuture_refcounthandlersupdatecreateputgetqsizeZstream_handlersfuture_releaserelease)selfrr r r __init__s  zQueueExtension.__init__NrcCsNtd|||jkr8tj|d|j|<d|j|<n|j|d7<dS)Nz Queue name: )maxsize)loggerdebugrasyncioQueuerr"nameclientr$r r r r/s   zQueueExtension.createcCst||jkrdS|j|d8<|j|dkrp|j|=|j|j}|j|=dd|D}|rp|jj|d|ddS)Nr%rcSs g|]}|ddkr|dqS)typervaluer ).0dr r r @s z*QueueExtension.release..queue-%skeysr,)rr_queuerclient_releases_keys)r"r+r,Zfuturesr4r r r r!7s  zQueueExtension.releasecsn|dk r@d|d}|j||fd7<|jj|gd|dn d|d}tj|j|||dIdHdS)Nr)r-r.r%r2r3msgpacktimeout)rrZclient_desires_keysr(wait_forrr)r"r+keydatar,r9recordr r r rDs   zQueueExtension.putcCsN|j||fd8<|j||fdkrJ|jj|gd|d|j||f=dS)Nr%rr2r3)rrr6)r"r+r;r,r r r r MszQueueExtension.future_releaseFc sfdd|rj|}g}|dkrJ|s|IdH}||q&n:|dk r^d}t|t|D]} |IdH}||qffdd|D}|Stjj||dIdH}|}|SdS)Ncsh|ddkrd|}|d}jj|}|dk r8|jnd}||d<|dkrd|jj|d<|jj|d <|S) zAdd task status if knownr-rr.NZloststateerred exception traceback)copyrZtasksrr>Zexception_blamer@rA)r=r;tsr>r"r r processTs   z#QueueExtension.get..processTzNDask queues don't support simultaneous use of integer batch sizes and timeoutscsg|] }|qSr r )r/o)rEr r r1tsz&QueueExtension.get..r8)remptyrappendNotImplementedErrorranger(r:) r"r+r,r9batchqoutr=msg_r )rEr"r rSs(    zQueueExtension.getcCs|j|SN)rr)r"r+r,r r r r{szQueueExtension.qsize)NNr)NN)NNNNN)NNN)NNNF)NN) __name__ __module__ __qualname____doc__r#rr!rr rrr r r r r s    (r c@szeZdZdZdddZddZdd Zdd d Zdd d Zd ddZ ddZ d!ddZ ddZ ddZ ddZddZdS)"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 Parameters ---------- name: string (optional) Name used by other clients and the scheduler to identify the queue. If not given, a random name will be generated. client: Client (optional) Client used for communication with the scheduler. If not given, the default global client will be used. maxsize: int (optional) Number of items allowed in the queue. If 0 (the default), the queue size is unbounded. 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 NrcCsxz|p t|_Wntk r0tj|_YnX|pBdtj|_||_ |jj rft | |_n|j|j dS)Nzqueue-)rcurrentr, ValueErrorr uuidZuuid4hexr+r$Z asynchronousr(Z ensure_future_start_startedsyncr*r r r r#szQueue.__init__cs |jjj|j|jdIdH|S)N)r+r$)r,rrr+r$rDr r r rYsz Queue._startcs.tdrjSfdd}|SdS)NrZcsSrPr r rDr r rOszQueue.__await__.._)hasattrrZ __await__)r"rOr rDr r]s   zQueue.__await__csNt|tr.|jjjt|j||jdIdHn|jjj|||jdIdHdS)N)r;r9r+)r<r9r+) isinstancerr,rrrr;r+)r"r.r9r r r _puts z Queue._putcKs$t|}|jj|j|fd|i|S)a9Put data into the queue Parameters ---------- timeout : number or string or timedelta, optional Time in seconds to wait before timing out. Instead of number of seconds, it is also possible to specify a timedelta in string format, e.g. "200ms". r9)rr,r[r_)r"r.r9kwargsr r r rs z Queue.putFcKs$t|}|jj|jf||d|S)a4Get data from the queue Parameters ---------- timeout : number or string or timedelta, optional Time in seconds to wait before timing out. Instead of number of seconds, it is also possible to specify a timedelta in string format, e.g. "200ms". 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 )r9rK)rr,r[_get)r"r9rKr`r r r rsz Queue.getcKs|jj|jf|S)z'Current number of elements in the queue)r,r[_qsize)r"r`r r r rsz Queue.qsizecsLjjj|j|dIdH}fdd}|dkr:||}ntt||}|S)N)r9r+rKcsp|ddkrdt|djd|dd}|ddkrH|j|d|d jd j|dd n|d}|S) Nr-rr.Tr>)Zinformr>r?r@rAr)opr+r;)rr,_stateZ set_error_send_to_schedulerr+)r0r.rDr r rEs  zQueue._get..processF)r,rrr+listmap)r"r9rKresprEresultr rDr ras   z Queue._getcs|jjj|jdIdH}|S)N)r+)r,rrr+)r"rir r r rb sz Queue._qsizecCs$|jjdkr |jd|jddS)NZrunningr)rcr+)r,statusrer+rDr r r close s z Queue.closecCs|j|jjjfSrP)r+r,raddressrDr r r __getstate__szQueue.__getstate__c Cs\|\}}zt|}|jj|ks"tWn$ttfk rHt|dd}YnX|j||ddS)NF)Zset_as_default)r+r,)rrrlAssertionErrorAttributeErrorrr#)r"r>r+rlr,r r r __setstate__szQueue.__setstate__)NNr)N)N)NF)NF)rQrRrSrTr#rYr]r_rrrrarbrkrmrpr r r r r)s(   r)) __future__rr(loggingrW collectionsrZ dask.utilsrrZdistributed.clientrrZdistributed.workerrr getLoggerrQr&r r)r r r r s   o