&`i':ddlZddlZddlmZddlmZmZmZmZddl Z ddl m Z e dGddej Z e dGd d ej Z e dGd d ZGd dZdS)N)Iterable)AnyDictListOptional) PublicAPIbeta) stabilityceZdZdS)EmptyN__name__ __module__ __qualname__b/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/util/queue.pyr r Drr ceZdZdS)FullNr rrrrrrrrc deZdZdZd dedeeddfdZdefdZdefd Z defd Z de fd Z de fd Z d!dede deeddfdZ d!dede deeddfdZd!de deedefdZ d!de deedefdZdeddfdZdeddfdZdefdZdedeefdZd"de deddfdZdS)#QueueaA first-in, first-out queue implementation on Ray. The behavior and use cases are similar to those of the asyncio.Queue class. Features both sync and async put and get methods. Provides the option to block until space is available when calling put on a full queue, or to block until items are available when calling get on an empty queue. Optionally supports batched put and get operations to minimize serialization overhead. Args: maxsize (optional, int): maximum size of the queue. If zero, size is unbounded. actor_options (optional, Dict): Dictionary of options to pass into the QueueActor during creation. These are directly passed into QueueActor.options(...). This could be useful if you need to pass in custom resource requirements, for example. Examples: .. testcode:: from ray.util.queue import Queue q = Queue() items = list(range(10)) for item in items: q.put(item) for item in items: assert item == q.get() # Create Queue with the underlying actor reserving 1 CPU. q = Queue(actor_options={"num_cpus": 1}) rNmaxsize actor_optionsreturncddlm}|d|pi}||_tjt jdi||j|_dS)Nr)record_library_usagez util.Queuer)ray._common.usage.usage_librrrayremote _QueueActoroptionsactor)selfrrrs r__init__zQueue.__init__7sqDDDDDD\***%+  +CJ{ # # + <   -44T::;;;;;$     "w{{ !JKKK --dG<<=====s 17Ac K|sB |jj|d{VdS#tj$rt wxYw||dkrt d|jj||d{VdS)aAdds an item to the queue. If block is True and the queue is full, blocks until the queue is no longer full or until timeout. There is no guarantee of order if multiple producers put to the same full queue. Raises: Full: if the queue is full and blocking is False. Full: if the queue is full, blocking is True, and it timed out. ValueError: if timeout is negative. Nrr:)r#r;r r<r=rr>r?r@s r put_asynczQueue.put_asyncps  ; j+22488888888888$     "w{{ !JKKKjn++D':::::::::::s %-Ac,|sK tj|jjS#t j$rtwxYw||dkrtdtj|jj|S)a Gets an item from the queue. If block is True and the queue is empty, blocks until the queue is no longer empty or until timeout. There is no guarantee of order if multiple consumers get from the same empty queue. Returns: The next item in the queue. Raises: Empty: if the queue is empty and blocking is False. Empty: if the queue is empty, blocking is True, and it timed out. ValueError: if timeout is negative. Nrr:) rr.r# get_nowaitr r< QueueEmptyr r>r$r7r8s rr.z Queue.gets" ? wtz4;;==>>>%     "w{{ !JKKKwtz~44W==>>>s /4A cK|s? |jjd{VS#tj$rt wxYw||dkrt d|jj|d{VS)aGets an item from the queue. There is no guarantee of order if multiple consumers get from the same empty queue. Returns: The next item in the queue. Raises: Empty: if the queue is empty and blocking is False. Empty: if the queue is empty, blocking is True, and it timed out. ValueError: if timeout is negative. Nrr:)r#rDr r<rEr r>r.rFs r get_asynczQueue.get_asyncs < !Z299;;;;;;;;;%     "w{{ !JKKK!Z^227;;;;;;;;;s #*Ac0||dS)zgEquivalent to put(item, block=False). Raises: Full: if the queue is full. Fr7)r?r$r6s rr;zQueue.put_nowaits xxEx***ritemsct|tstdtj|jj|dS)zTakes in a list of items and puts them into the queue in order. Raises: Full: if the items will not fit in the queue z$Argument 'items' must be an IterableN) isinstancer TypeErrorrr.r#put_nowait_batchr )r$rLs rrPzQueue.put_nowait_batchsO %** DBCC C  +22599:::::rc.|dS)zcEquivalent to get(block=False). Raises: Empty: if the queue is empty. FrJ)r.r*s rrDzQueue.get_nowaits xxex$$$r num_itemsct|tstd|dkrtdt j|jj|S)zGets items from the queue and returns them in a list in order. Raises: Empty: if the queue does not contain the desired number of items z#Argument 'num_items' must be an intrz'num_items' must be nonnegative) rNintrOr>rr.r#get_nowait_batchr r$rRs rrUzQueue.get_nowait_batchsa)S)) CABB B q==>?? ?wtz299)DDEEErFforcegrace_period_sc|jrs|rtj|jdnU|jj}tj|g|\}}|rtj|jdd|_dS)aTerminates the underlying QueueActor. All of the resources reserved by the queue will be released. Args: force: If True, forcefully kill the actor, causing an immediate failure. If False, graceful actor termination will be attempted first, before falling back to a forceful kill. grace_period_s: If force is False, how long in seconds to wait for graceful termination before falling back to forceful kill. T) no_restart)r8N)r#rkill__ray_terminate__r wait)r$rXrYdone_refdonenot_dones rshutdownzQueue.shutdowns : : :55555:7>>@@!$8*n!M!M!Mh:HTZD9999 r)rN)TN)FrW)rrr__doc__rTrrr%r+r)r/boolr3r5rfloatr?rBr.rHr;rrPrDrrUrbrrrrrsB       RV     2c2222s2t22221d1111 IM>>> $>6>uo> >>>>8IM;;; $;6>uo; ;;;;6???x?#????:>B<<<+3E?< <<<<4+s+t++++ ;h ;4 ; ; ; ;%C%%%% F# F$s) F F F FdCrrcNeZdZdZdZdZdZd dZd dZdZ d Z d Z d Z dS) r!cP||_tj|j|_dSr')rr<rqueue)r$rs rr%z_QueueActor.__init__s  ]4<00 rc4|jSr')rhr/r*s rr/z_QueueActor.qsizez!!!rc4|jSr')rhr3r*s rr3z_QueueActor.empty rjrc4|jSr')rhr5r*s rr5z_QueueActor.fullsz   rNcK tj|j||d{VdS#tj$rt wxYwr')r<wait_forrhr? TimeoutErrorr)r$r6r8s rr?z_QueueActor.putsg "4:>>$#7#7AA A A A A A A A A A#   J s 39AcK tj|j|d{VS#tj$rt wxYwr')r<rnrhr.ror )r$r8s rr.z_QueueActor.gets_  )$*..*:*:GDDDDDDDD D#   K s 16A c:|j|dSr')rhr;rKs rr;z_QueueActor.put_nowaits d#####rc .|jdkrjt||z|jkr=tdt|d|d|jd|D]}|j|dS)Nrz Cannot add z items to queue of size z and maxsize .)rlenr/rrhr;)r$rLr6s rrPz_QueueActor.put_nowait_batch s c%jj>>::<<>>.2l>>>  ( (D J ! !$ ' ' ' ' ( (rc4|jSr'rhrDr*s rrDz_QueueActor.get_nowait*sz$$&&&rc|kr(td|ddfdt|DS)Nz Cannot get z items from queue of size rscBg|]}jSrrv).0_r$s r z0_QueueActor.get_nowait_batch..2s'BBBA %%''BBBr)r/r rangerVs` rrUz_QueueActor.get_nowait_batch-sk tzz|| # #ViVVtzz||VVV CBBBy1A1ABBBBrr') rrrr%r/r3r5r?r.r;rPrDrUrrrr!r!s111""""""!!!  $$$((('''CCCCCrr!)r<rhcollections.abcrtypingrrrrrray.util.annotationsrr rrr!rrrrs{ $$$$$$,,,,,,,,,,,, ****** V     EK    V     5:    Vkkkkkkkk\/C/C/C/C/C/C/C/C/C/Cr