&`i:)ddlmZddlZddlZddlmZmZmZmZddl Z ddl m Z ddl m Z mZerddlmZe GddZeGd d ZdS) ) annotationsN) TYPE_CHECKINGDequeIteratorOptional)ObjectRefStreamEndOfStreamError) DeveloperAPI PublicAPI)Workerc&eZdZd dZd dZd dZd S) DynamicObjectRefGeneratorrefsDeque['ray.ObjectRef']c8tj||_dSN) collectionsdeque_refs)selfrs u/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/_private/object_ref_generator.py__init__z"DynamicObjectRefGenerator.__init__s.9->t-D-D returnIterator('ray.ObjectRef')c#`K|jr$|jV|j"dSdSr)rpopleftrs r__iter__z"DynamicObjectRefGenerator.__iter__sJj '*$$&& & & &j ' ' ' ' 'rintc*t|jSr)lenrrs r__len__z!DynamicObjectRefGenerator.__len__s4:rN)rr)rr)rr)__name__ __module__ __qualname__rrr"rrr r sTEEEE ''''rr ceZdZdZd"dZd#d Zd$d Zd Zd Zd Z d#dZ dZ dZ dZ dZd$dZd%dZd%dZd$dZd&d'dZd(dZd&d)dZd Zd!ZdS)*ObjectRefGeneratoraA generator to obtain object references from a task in a streaming manner. The class is compatible with the Python generator and async generator interfaces. The class is not thread-safe. Do not initialize the class and create an instance directly. The instance should be created by `.remote`. .. testcode:: import ray from typing import Generator @ray.remote(num_returns="streaming") def gen() -> Generator[int, None, None]: for i in range(5): yield i obj_ref_gen: ray.ObjectRefGenerator = gen.remote() for obj_ref in obj_ref_gen: print("Got:", ray.get(obj_ref)) generator_ref'ray.ObjectRef'worker'Worker'c||_d|_||_|jt |dsJdS)NF core_worker)_generator_ref_generator_task_raisedr+check_connectedhasattr)rr)r+s rrzObjectRefGenerator.__init__9sI+&+#  ##%%%v}-------rr'ObjectRefGenerator'c|Srr&rs rrzObjectRefGenerator.__iter__E rc*|S)aWaits until a next ref is available and returns the object ref. Raises StopIteration if there's no more objects to generate. The object ref will contain an exception if the task fails. When the generator task returns N objects, it can return up to N + 1 objects (if there's a system failure, the last object will contain a system level exception). ) _next_syncrs r__next__zObjectRefGenerator.__next__Hs   rc td)Nz`gen.send` is not supported.NotImplementedErrorrvalues rsendzObjectRefGenerator.sendUs!"@AAArc td)Nz`gen.throw` is not supported.r:r<s rthrowzObjectRefGenerator.throwX!"ABBBrc td)Nz`gen.close` is not supported.r:rs rclosezObjectRefGenerator.close[rArc|Srr&rs r __aiter__zObjectRefGenerator.__aiter__^r5rc:K|d{VSr) _next_asyncrs r __anext__zObjectRefGenerator.__anext__as*%%'''''''''rc$Ktd)Nz`gen.asend` is not supported.r:r<s rasendzObjectRefGenerator.asendds!"ABBBrc$Ktd)Nz`gen.athrow` is not supported.r:r<s rathrowzObjectRefGenerator.athrowg!"BCCCrc$Ktd)Nz`gen.aclose` is not supported.r:rs raclosezObjectRefGenerator.aclosejrMrc|jS)aReturns an object ref that is ready when a generator task completes. If the task is failed unexpectedly (e.g., worker failure), the `ray.get(gen.completed())` raises an exception. The function returns immediately. )r/rs r completedzObjectRefGenerator.completedms ""rboolc|j|jj}|rdS||j\}}|rdSt j|gdd\}}t|dkS)a)If True, it means the output of next(gen) is ready and ray.get(next(gen)) returns immediately. False otherwise. It returns False when next(gen) raises a StopIteration (this condition should be checked using is_finished). The function returns immediately. FTrtimeout fetch_local) r+r1r. is_finishedpeek_object_ref_streamr/raywaitr!)rr. expected_refis_readyready_s r next_readyzObjectRefGenerator.next_readyxs ##%%%k-      5!,!C!CDDW!X!X h  48\NA5IIIq5zzA~rc|j|jj}||j}|r6|jrdS t j|jdS#t$rYdSwxYwdS)zIf True, it means the generator is finished and all output is taken. False otherwise. When True, if next(gen) is called, it will raise StopIteration or StopAsyncIteration The function returns immediately. TF) r+r1r.is_object_ref_stream_finishedr/r0rYget Exception)rr.finisheds rrWzObjectRefGenerator.is_finisheds ##%%%k- < ##%%%k- 11$2EFFqIIrN timeout_sOptional[int | float]c|jj}||j\}}|sLt j|g|d\}}t |dkrtjS | |j}| rJn_#t$rR|j rtd t j|jtd#t$rd|_ |jcYcYSwxYwwxYw|S)aTWaits for timeout_s and returns the object ref if available. If an object is not available within the given timeout, it returns a nil object reference. If -1 timeout is provided, it means it waits infinitely. Waiting is implemented as busy waiting. Raises StopIteration if there's no more objects to generate. The object ref will contain an exception if the task fails. When the generator task returns N objects, it can return up to N + 1 objects (if there's a system failure, the last object will contain a system level exception). Args: timeout_s: If the next object is not ready within this timeout, it returns the nil object ref. Returns: ObjectRef corresponding to the next result in the stream. FrTrNT)r+r.rXr/rYrZr! ObjectRefniltry_read_next_object_ref_streamis_nilrr0 StopIterationrbrc)rrgr.r[r\r^unreadyrefs rr7zObjectRefGenerator._next_syncs:2k- "-!C!CDDW!X!X h +<.)QVWWWJAw7||a}((*** .==d>QRRCzz|| # ## #. . . .* .#- .+,,, $-  + + +.2+****** + .  s090B**DC%D%D=DDDrpNonec>K |d{VdS#t$rYdSwxYwr)rc)rrps r_suppress_exceptionsz'ObjectRefGenerator._suppress_exceptionssB  IIIIIIIII    DD s  cVK|jj}||j\}}|svt jt j||g|d{V\}}t|dkrtj S | |j}| rJnS#t$rF|jrt d |jd{Vt d#t"$rd|_|jcYcYSwxYwwxYw|S)z4Same API as _next_sync, but it is for async context.)rUNrT)r+r.rXr/asynciorZ create_taskrsr!rYrjrkrlrmrr0StopAsyncIterationrc)rrgr.rpr\r^ros rrGzObjectRefGenerator._next_asyncsk- #::4;NOO X +&|$T%>%>s%C%CDDEy   JAw7||a}((*** 3==d>QRRCzz|| # ## #. 3 3 3* 3(d2 3)))))))) )d2  + + +.2+****** + 3" s0%0CD&0 D=D&D"D&!D""D&c|t|jdr&|jj|jdSdS)Nr.)r2r+r.async_delete_object_ref_streamr/rs r__del__zObjectRefGenerator.__del__!sG 4; . . X K # B B4CV W W W W W  X Xrc td)NzeYou cannot return or pass a generator to other task. Serializing a ObjectRefGenerator is not allowed.) TypeErrorrs r __getstate__zObjectRefGenerator.__getstate__)s ?   r)r)r*r+r,)rr3)rr*)rrRr)rgrhrr*)rpr*rrq)rgrh)r#r$r%__doc__rrr8r>r@rCrErHrJrLrOrQr_rWrfr7rsrGrzr}r&rrr(r(s0.... ! ! ! !BBBCCCCCC(((CCCDDDDDD # # # #.FJJJJ66666p!!!!!FXXX     rr() __future__rrurtypingrrrrrYray.exceptionsrray.util.annotationsr r ray._private.workerr r r(r&rrrs"""""";;;;;;;;;;;; ::::::88888888+******         M M M M M M M  M M M r