&`i&ddlZddlZddlmZmZddlmZmZmZm Z m Z m Z m Z ddl Z ddlmZmZejeZGddZ dd e e e eeefd e ed ed ee eeffd Z dde edeedede eed e ed eef dZ ddeed e ed e eeeeffdZdeed eefdZdS)N) defaultdictdeque)AnyDictListOptionalSequenceTupleType) ActorClass ActorHandlecJeZdZdZdZdZd dZd dZdZe d Z d S) TaskPoolzCHelper class for tracking the status of many in-flight actor tasks.cHi|_i|_t|_dSN)_tasks_objectsr _fetchingselfs j/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/rllib/utils/actors.py__init__zTaskPool.__init__s  cnt|tr |d}n|}||j|<||j|<dS)Nr) isinstancelistrr)rworker all_obj_refsobj_refs raddz TaskPool.addsB lD ) ) #"1oGG"G% G!- grFc#:Kt|j}|rtj|t |d\}}|s|rtj|dd\}}|D]:}|j||j|fV9dSdS)Nr) num_returnstimeoutg$@)rrraywaitlenpopr)r blocking_waitpendingready_rs r completedzTaskPool.completedst{##  MxS\\1MMMHE1 J] J8GDIIIq  M M{w//1B1B71K1KLLLLL  M M M Mrc#K||D]!\}}|j||f"t|D]'}|jsdS|jV(dS)zhSimilar to completed but only returns once the object is local. Assumes obj_ref only is one id.)r)N)r-rappendrangepopleft)rr) max_yieldrrr,s rcompleted_prefetchzTaskPool.completed_prefetch%s $~~M~JJ 5 5OFG N ! !67"3 4 4 4 4y!! + +A> .((** * * * *  + +rcP|jD]\}}||vr|j|=|j|=t t |jD]>}|j\}}||vr|j||f?dS)z(Notify that some workers may be removed.N) rcopyitemsrr1r'rr2r0)rworkersrevr,s r reset_workerszTaskPool.reset_workers3s;++--3355 + +KGR  K(M'*s4>**++ 5 5A.0022KBW}}%%r7m444  5 5rc*t|jSr)r'rrs rcountzTaskPool.countBs4;rN)F)Fr.) __name__ __module__ __qualname____doc__rr r-r4r:propertyr<rrrr sMM!!! ...MMMM + + + + 5 5 5  X   rr localhost actor_specsnode max_attemptsreturnc |dkrtj}dtt|D}t|D]#}d}t |D]\}\}}} } |pg}| pi} t||| krt ||| | |dzz|} |1t j| dj }|| | t||| krd}t||| kr?||| dD]} | j ||d| ||< |r|cS%td ) aOCreate co-located actors of any type(s) on any node. Args: actor_specs: Tuple/list with tuples consisting of: 1) The (already @ray.remote) class(es) to construct, 2) c'tor args, 3) c'tor kwargs, and 4) the number of actors of that class with given args/kwargs to construct. node: The node to co-locate the actors on. By default ("localhost"), place the actors on the node the caller of this function is located on. Use None for indicating that any (resource fulfilling) node in the cluster may be used. max_attempts: The maximum number of co-location attempts to perform before throwing an error. Returns: A dict mapping the created types to the list of n ActorHandles created (and co-located) for that type. rCcg|]}gSrBrB).0r,s r z+create_colocated_actors..bs . . ." . . .rTr$)clsargskwargsr<rFNrFz5Unable to create enough colocated actors -> aborting.) platformrFr1r' enumeratetry_create_colocatedr%getget_hostremoteextend__ray_terminate__ Exception) rErFrGokattemptall_goodityprNrOr< co_locatedas rcreate_colocated_actorsr`Gs. {} / .eC ,,-- . . .B &&&&-6{-C-C & & )A)T65:2D\rF2a5zzE!!1!7Q;/  <7:a=#9#@#@#B#BCCD1 Z(((r!u::%%$H2a5zzE!!Auvv11A'..00001fuf 1  III  K L LLrrMrNr<rOcZ|dkrtj}pifdt|D}t||\}}tdt|||D]}|j |S)aTries to co-locate (same node) a set of Actors of the same type. Returns a list of successfully co-located actors. All actors that could not be co-located (with the others on the given node) will not be in this list. Creates each actor via it's remote() constructor and then checks, whether it has been co-located (on the same node) with the other (already created) ones. If not, terminates the just created actor. Args: cls: The Actor class to use (already @ray.remote "converted"). args: List of args to pass to the Actor's constructor. One item per to-be-created actor (`count`). count: Number of actors of the given `cls` to construct. kwargs: Optional list of kwargs to pass to the Actor's constructor. One item per to-be-created actor (`count`). node: The node to co-locate the actors on. By default ("localhost"), place the actors on the node the caller of this function is located on. If None, will try to co-locate all actors on any available node. Returns: List containing all successfully co-located actor handles. rCc*g|]}jiSrB)rU)rKr,rNrMrOs rrLz(try_create_colocated..s* @ @ @ajcj$)&)) @ @ @r)rFzGot {} colocated actors of {}) rPrFr1split_colocatedloggerinfoformatr'rWrU) rMrNr<rOrFactorsr^non_co_locatedr_s `` ` rrRrRs@ {} \rF @ @ @ @ @ @5<< @ @ @F!0d!C!C!CJ KK/66s:NNOOO %% ""$$$$ rrgc|dkrtj}tjd|D}|t t }t ||D] \}}|||!d}d}|D]2}|t||krt||}|}3g}|D]0}||kr(| t||1t|||fSg} g}t ||D]6\}} ||kr| | !| | 7| |fS)aMSplits up given actors into colocated (on same node) and non colocated. The co-location criterion depends on the `node` given: If given (or default: platform.node()): Consider all actors that are on that node "colocated". If None: Consider the largest sub-set of actors that are all located on the same node (whatever that node is) as "colocated". Args: actors: The list of actor handles to split into "colocated" and "non colocated". node: The node defining "colocation" criterion. If provided, consider thos actors "colocated" that sit on this node. If None, use the largest subset within `actors` that are sitting on the same (any) node. Returns: Tuple of two lists: 1) Co-located ActorHandles, 2) non co-located ActorHandles. rCc@g|]}|jSrB)rTrU)rKr_s rrLz#split_colocated..s&999QQZ&&((999rN) rPrFr%rSrsetzipr r'rVrr0) rgrFhosts node_groupshostactormax_ largest_grouprhr^r_s rrcrcs0 {} G99&999 : :E |!#&& uf-- ) )KD%   ! !% ( ( ( (  % %Dc+d+,,,,;t,-- $  ? ?D}$$%%d;t+<&=&=>>>K .//??  5&)) ) )GD!t||!!!$$$$%%a((((>))rcft|\}}|D]}|j|Sr)rcrWrU)rg colocated non_colocatedr_s rdrop_colocatedrwsA.v66I} %% ""$$$$ r)rCrD)NrC)rC)loggingrP collectionsrrtypingrrrrr r r r% ray.actorr r getLoggerr=rdrintstrr`rRrcrwrBrrrs@********CCCCCCCCCCCCCCCCCC --------  8 $ $8 8 8 8 8 8 8 8 z&IMIM%c3 345IM 3-IMIM $[! !" IMIMIMIM`#'% )) j ) s)) ) T#Y  ) 3- )  + ))))\&<*<*  <* 3-<* 4 d;/ /0<*<*<*<*~4 ,k1Br