&`iddlZddlZddlZddlZddlmZddlmZmZddl m Z m Z m Z m Z mZmZmZmZddlZddlmZddlmZmZddlmZddlmZejeZeGd d ZeeGd d Z eGd dZ!eGddZ"eGddZ#dS)N) defaultdict) dataclassfield)AnyCallableDictIteratorListOptionalTupleUnion) ActorHandle)RayError RayTaskError)T) DeveloperAPIc@eZdZdZddedefdZedZdZ dS) ResultOrErrorzA wrapper around a result or a RayError thrown during remote task/actor calls. This is used to return data from `FaultTolerantActorManager` that allows us to distinguish between RayErrors (remote actor related) and valid results. Nresulterrorct||_t|tr|n||_dS)a1One and only one of result or error should be set. Args: result: The result of the computation. Note that None is a valid result if the remote function does not return anything. error: Alternatively, the error that occurred during the computation. N)_result isinstanceras_instanceof_cause_error)selfrrs q/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/rllib/utils/actor_manager.py__init__zResultOrError.__init__s> %.. E % % ' ' ' c|jduSN)rrs rokzResultOrError.ok*s{d""rc,|jr|jS|jS)z Returns the result or the error.)rrr"s rgetzResultOrError.get.s ; ; < r)NN) __name__ __module__ __qualname____doc__r Exceptionrpropertyr#r%rrrrso   s )    ##X#     rrcNeZdZUdZeed<eed<eed<edZ dZ dS) CallResultzRepresents a single result from a call to an actor. Each CallResult contains the index of the actor that was called plus the result or error from the call. actor_idresult_or_errortagc|jjS)z8Passes through the ok property from the result_or_error.)r0r#r"s rr#z CallResult.okCs#&&rc4|jS)z7Passes through the get method from the result_or_error.)r0r%r"s rr%zCallResult.getHs#'')))rN) r&r'r(r)int__annotations__rstrr+r#r%r,rrr.r.6sh MMM"""" HHH ''X'*****rr.ceZdZdZGddZdZdededefdZ d e efd Z d efd Z d e efd Z d e efd ZdS)RemoteCallResultsaRepresents a list of results from calls to a set of actors. CallResults provides convenient APIs to iterate over the results while skipping errors, etc. .. testcode:: :skipif: True manager = FaultTolerantActorManager( actors, max_remote_requests_in_flight_per_actor=2, ) results = manager.foreach_actor(lambda w: w.call()) # Iterate over all results ignoring errors. for result in results.ignore_errors(): print(result.get()) cNeZdZdZdeefdZdeefdZdefdZ dS)RemoteCallResults._Iteratorz.An iterator over the results of a remote call. call_resultsc||_dSr!) _call_results)rr;s rrz$RemoteCallResults._Iterator.__init__ds!-D   rreturnc|Sr!r,r"s r__iter__z$RemoteCallResults._Iterator.__iter__gsKrcR|jst|jdSNr)r= StopIterationpopr"s r__next__z$RemoteCallResults._Iterator.__next__js*% $##%))!,, ,rN) r&r'r(r)r r.rr r@rEr,rr _Iteratorr:asu<< .j)9 . . . . hz2     -j - - - - - -rrFcg|_dSr!)result_or_errorsr"s rrzRemoteCallResults.__init__os24rr/r0r1cX|jt|||dS)zAdd index of a remote actor plus the call result to the list. Args: actor_id: ID of the remote actor. result_or_error: The result or error from the call. tag: A description to identify the call. N)rHappendr.)rr/r0r1s r add_resultzRemoteCallResults.add_resultrs- $$Z/3%O%OPPPPPrr>cZ|tj|jS)z$Return an iterator over the results.)rFcopyrHr"s rr@zRemoteCallResults.__iter__|s#~~di(=>>???rc*t|jSr!)lenrHr"s r__len__zRemoteCallResults.__len__s4()))rcJ|d|jDS)z9Return an iterator over the results, skipping all errors.c g|] }|j | Sr,)r#.0rs r z3RemoteCallResults.ignore_errors..sHHHQ14HqHHHrrFrHr"s r ignore_errorszRemoteCallResults.ignore_errorss&~~HH$*?HHHIIIrcJ|d|jDS)aCReturn an iterator over the results, skipping only Ray errors. Similar to ignore_errors, but only skips Errors raised because of remote actor problems (often get restored automatcially). This is useful for callers that want to handle application errors differently from Ray errors. c`g|]+}t|t)|,Sr,)rr%rrSs rrVz7RemoteCallResults.ignore_ray_errors..s/ S S S1Z5R5R SQ S S SrrWr"s rignore_ray_errorsz#RemoteCallResults.ignore_ray_errorss.~~ S S- S S S   rN)r&r'r(r)rFrr4rr6rKr r@rPrXr[r,rrr8r8Ms $ - - - - - - - -555Q3QQSQQQQ@(=1@@@@ *****Jx 6JJJJ  8M#:       rr8cxeZdZedefdZedeeeeeege fde fdZ dS)FaultAwareApplyr>cdS)zuPing the actor. Can be used as a health check. Returns: "pong" if actor is up and well. pongr,r"s rpingzFaultAwareApply.pings vrfuncc ||g|Ri|S#t$ri}|jjrPtd|t j|jjtj dn|Yd}~dSd}~wwxYw)aCalls the given function with this Actor instance. A generic interface for applying arbitrary member functions on a remote actor. Args: func: The function to call, with this actor as first argument, followed by args, and kwargs. args: Optional additional args to pass to the function call. kwargs: Optional additional kwargs to pass to the function call. Returns: The return value of the function call. z*Worker exception caught during `apply()`: N) r*configrestart_failed_env_runnerslogger exceptiontimesleep#delay_between_env_runner_restarts_ssysexit)rraargskwargses rapplyzFaultAwareApply.applys* 4.t...v.. .   {5   !Qa!Q!QRRR 4;JKKK   s  BAA<<BN) r&r'r(rr6r`rrr rrpr,rrr]r]sc\ Xc]HSM:A=>    \   rr]ceZdZdZeGddZ d>deeede d e fd Z e d ee fd Z e d ee fd Z e deefdZe de d efdZe d e fdZe d e fdZe d e fdZe d?deed e fdZe de d efdZe de ded dfdZe dZe ddddddddeeegefeeegefeeefdeeeeefeeeeffded eee d!eed"ed#ed efd$Ze d?dddd%deeegefeeegefeeefdeedeeeeefeeeeffded eee d e f d&Z e d'd(ddd)d*eeeee!ed+ffd!eed"ed#ed ef d,Z"e d?dd(ddddddd-deeegefeeegefeeefdeedeeeeefeeeeffd!eed"ed#eded eee d.ed/ed eee!e efeffd0Z#e$d1ed.ed dfd2Z%e d@d!eed#ed ee fd3Z&ddd4deeegefeeegefeeefdeeeeefeeeeffd ee d ee'j(fd5Z)e dddd6d ee d7ee'j(d*eed!eed"ed#ed e!ee'j(effd8Z*dd9deeegefeeegeffdeeeeefd ee fd:Z+ d?d*eeeeee!ed+ffd e!ee'j(eeeeffd;Z,de fd<Z-d=Z.dS)AFaultTolerantActorManagera}A manager that is aware of the healthiness of remote actors. .. testcode:: import time import ray from ray.rllib.utils.actor_manager import FaultTolerantActorManager @ray.remote class MyActor: def apply(self, func): return func(self) def do_something(self): return True actors = [MyActor.remote() for _ in range(3)] manager = FaultTolerantActorManager( actors, max_remote_requests_in_flight_per_actor=2, ) # Synchronous remote calls. results = manager.foreach_actor(lambda actor: actor.do_something()) # Print results ignoring returned errors. print([r.get() for r in results.ignore_errors()]) # Asynchronous remote calls. manager.foreach_actor_async(lambda actor: actor.do_something()) time.sleep(2) # Wait for the tasks to finish. for r in manager.fetch_ready_async_reqs(): # Handle result and errors. if r.ok: print(r.get()) else: print("Error: {}".format(r.get())) ceZdZUdZeeZeee e fe d<dZ e e d<d dee de fd Zd dee ddfd Zd dee ddfd ZdS) %FaultTolerantActorManager._ActorStatezState of a single actor.)default_factory#num_in_flight_async_requests_by_tagT is_healthyNr1r>c|&t|jS|j|dS)z@Get number of in-flight requests for a specific tag or all tags.Nr)sumrvvaluesr%rr1s rget_num_in_flight_requestsz@FaultTolerantActorManager._ActorState.get_num_in_flight_requestss={4CJJLLMMM;??QGG GrcV||jvr d|j|<|j|xxdz cc<dS)z4Increment the count of in-flight requests for a tag.rrcNrvr{s rincrement_requestsz8FaultTolerantActorManager._ActorState.increment_requestssB$BBB@A8=  4S 9 9 9Q > 9 9 9 9 9rc|||jvr0|j|xxdzcc<|j|dkr |j|=dSdSdS)z4Decrement the count of in-flight requests for a tag.rcrNr~r{s rdecrement_requestsz8FaultTolerantActorManager._ActorState.decrement_requestssad>>>8===B===;C@AEE@EEE?>EErr!)r&r'r(r)rdictrvrr r6r4r5rwboolr|rrr,rr _ActorStaterts&&IN I I I +T(3-2D-E    D H H(3- H3 H H H H  ? ?(3- ?4 ? ? ? ?  F F(3- F4 F F F F F FrrNractors'max_remote_requests_in_flight_per_actorinit_idc|x|_|_i|_i|_t |_||pgi|_||_d|_ dS)aConstruct a FaultTolerantActorManager. Args: actors: A list of ray remote actors to manage on. These actors must have an ``apply`` method which takes a function with only one parameter (the actor instance itself). max_remote_requests_in_flight_per_actor: The maximum number of remote requests that can be in flight per actor. Any requests made to the pool that cannot be scheduled because the limit has been reached will be dropped. This only applies to the asynchronous remote call mode. init_id: The initial ID to use for the next remote actor. Default is 0. rN) _next_id_current_actor_id_actors_remote_actor_statesset_restored_actors add_actors_in_flight_req_to_actor_id(_max_remote_requests_in_flight_per_actor_num_actor_restarts)rrrrs rrz"FaultTolerantActorManager.__init__ so(298 .02 AC! #  "%%%EG' 4 5 $%   rr>cNt|jS)z2Returns a list of all worker IDs (healthy or not).)listrkeysr"s r actor_idsz#FaultTolerantActorManager.actor_ids0s DL%%''(((rcHd|jDS)z.Returns a list of worker IDs that are healthy.c&g|]\}}|j |Sr,rw)rTkvs rrVz?FaultTolerantActorManager.healthy_actor_ids..8s#PPPda1<PPPPr)ritemsr"s rhealthy_actor_idsz+FaultTolerantActorManager.healthy_actor_ids5s(QPd7==??PPPPrc|D]B}||j|j<||j|j<|xjdz c_CdS)zAdd a list of actors to the pool. Args: actors: A list of ray remote actors to be added to the pool. rcN)rrrr)rractors rrz$FaultTolerantActorManager.add_actors:sX  E*/DL '7;7G7G7I7ID %dm 4 MMQ MMM  rr/c|j|}|j|=|j|=|j||||S)zRemove an actor from the pool. Args: actor_id: ID of the actor to remove. Returns: Handle to the actor that was removed. )rrrdiscard_remove_async_state)rr/rs r remove_actorz&FaultTolerantActorManager.remove_actorFsW X& L "  %h / %%h///   *** rc*t|jS)z.Return the total number of actors in the pool.)rOrr"s r num_actorsz$FaultTolerantActorManager.num_actorsZs4<   rcbtd|jDS)z+Return the number of healthy remote actors.c3$K|] }|jV dSr!r)rTss r z?FaultTolerantActorManager.num_healthy_actors..bs$LLA1<LLLLLLrryrrzr"s rnum_healthy_actorsz,FaultTolerantActorManager.num_healthy_actors_s0LL)B)I)I)K)KLLLLLLrc|jS)z.lsE    ( ( - -      rrr{s `rnum_outstanding_async_reqsz4FaultTolerantActorManager.num_outstanding_async_reqsisL    .5577      rc\||jvrtd||j|jS)zWhether a remote actor is in healthy state. Args: actor_id: ID of the remote actor. Returns: True if the actor is healthy, False otherwise. Unknown actor id: )r ValueErrorrw)rr/s ris_actor_healthyz*FaultTolerantActorManager.is_actor_healthyqs: 44 4 4<(<<== =(2==rhealthyc0||jvrtd||j|j}|s|r|j|n|r|s|j|||j|_|s||dSdS)zUpdate activate state for a specific remote actor. Args: actor_id: ID of the remote actor. healthy: Whether the remote actor is healthy. rN)rrrwraddrr)rr/r was_healthys rset_actor_statez)FaultTolerantActorManager.set_actor_states 44 4 4<(<<== =/9D  4w 4  ! % %h / / / /  4 4  ! ) )( 3 3 39@!(+6 /  $ $X . . . . . / /rc.|jD]}tj||j|j|j|jdS)zClean up managed actors.N)rrzraykillclearrrr)rrs rrzFaultTolerantActorManager.clears\((**  E HUOOOO  !''))) ##%%% '--/////rTF)rn healthy_onlyremote_actor_idstimeout_secondsreturn_obj_refs mark_healthyrarnrrrrrc|p|}|r||||\}}}||||}|||dgt |z|||\} } | S)a Calls the given function with each actor instance as arg. Automatically marks actors unhealthy if they crash during the remote call. Args: func: A single Callable applied to all specified remote actors or a list of Callables, that get applied on the list of specified remote actors. In the latter case, both list of Callables and list of specified actors must have the same length. Alternatively, you can use the name of the remote method to be called, instead, or a list of remote method names. kwargs: An optional single kwargs dict or a list of kwargs dict matching the list of provided `func` or `remote_actor_ids`. In the first case (single dict), use `kwargs` on all remote calls. The latter case (list of dicts) allows you to define individualized kwarg dicts per actor. healthy_only: If True, applies `func` only to actors currently tagged "healthy", otherwise to all actors. If `healthy_only=False` and `mark_healthy=True`, will send `func` to all actors and mark those actors "healthy" that respond to the request within `timeout_seconds` and are currently tagged as "unhealthy". remote_actor_ids: Apply func on a selected set of remote actors. Use None (default) for all actors. timeout_seconds: Time to wait (in seconds) for results. Set this to 0.0 for fire-and-forget. Set this to None (default) to wait infinitely (i.e. for synchronous execution). return_obj_refs: whether to return ObjectRef instead of actual results. Note, for fault tolerance reasons, these returned ObjectRefs should never be resolved with ray.get() outside of the context of this manager. mark_healthy: Whether to mark all those actors healthy again that are currently marked unhealthy AND that returned results from the remote call (within the given `timeout_seconds`). Note that actors are NOT set unhealthy, if they simply time out (only if they return a RayActorError). Also not that this setting is ignored if `healthy_only=True` (b/c this setting only affects actors that are currently tagged as unhealthy). Returns: The list of return values of all calls to `func(actor)`. The values may be actual data returned or exceptions raised during the remote call in the format of RemoteCallResults. rarnrNr remote_callstagsrrr)r_filter_by_healthy_state _call_actors _fetch_resultrO) rrarnrrrrrr_remote_resultss r foreach_actorz'FaultTolerantActorManager.foreach_actorsh,?t~~/?/?  -1-J-J&;K.K.. *D&* ((-)  !..-%#l+++++% /  >r)rnrrc|s}t|trt|n2t|trt|nt|}|t|krMfdt |D}xj|z c_xjzc_|r|||\}}}td}t|trt|t|ksJg}g} g} tt||D]\} \} } j |  |}||| zj krp|| xxdz cc<t|tr|| n|pi}|| | || | nd|}|} g} |D][} j |  |}||| zj kr%|| xxdz cc<| | \| sdS|| | }t| |D]1\}}j ||||fj|<2t|S)aCalls given functions against each actors without waiting for results. Args: func: A single Callable applied to all specified remote actors or a list of Callables, that get applied on the list of specified remote actors. In the latter case, both list of Callables and list of specified actors must have the same length. Alternatively, you can use the name of the remote method to be called, instead, or a list of remote method names. tag: A tag to identify the results from this async call. kwargs: An optional single kwargs dict or a list of kwargs dict matching the list of provided `func` or `remote_actor_ids`. In the first case (single dict), use `kwargs` on all remote calls. The latter case (list of dicts) allows you to define individualized kwarg dicts per actor. healthy_only: If True, applies `func` only to actors currently tagged "healthy", otherwise to all actors. If `healthy_only=False` and later, `self.fetch_ready_async_reqs()` is called with `mark_healthy=True`, will send `func` to all actors and mark those actors "healthy" that respond to the request within `timeout_seconds` and are currently tagged as "unhealthy". remote_actor_ids: Apply func on a selected set of remote actors. Note, for fault tolerance reasons, these returned ObjectRefs should never be resolved with ray.get() outside of the context of this manager. Returns: The number of async requests that are actually fired. cNg|]!}j|zz"Sr,)rrrTirs rrVzAFaultTolerantActorManager.foreach_actor_async..+s@   '!+t/@/@@   rrcdSrBr,r,rrz?FaultTolerantActorManager.foreach_actor_async..8srrcr)rrrrOrangerrrr enumerateziprr|rrJrrr)rrar1rnrr num_callsnum_calls_to_make limited_funclimited_kwargslimited_remote_actor_idsrfraidnum_outstanding_reqs_for_tagrridcalls` rforeach_actor_asyncz-FaultTolerantActorManager.foreach_actor_asyncsZ  0#~~// $%% 'CIII&$'''V%&&  ,-- - -    y))      " "i / " "  " "doo&7&7 7 " "  -1-J-J&;K.K.. *D&*-8 ,B,B dD ! ! :t99$4 5 55555LN') $ )#d4D*E*E F F : : 9At/3/H0,,S11-13DT3JJCDD&d+++q0+++%/%=%=Qq FLbA ''***"))!,,,,33D999 : L#N') $( : :/3/H0,,S11-13DT3JJCDD&d+++q0+++,33D999' 1((!5)  4lCC > >HB  %b ) < .s'RRRaQZ)RRRrc6g|]}|Sr,)r%rSs rrVzMFaultTolerantActorManager.foreach_actor_async_fetch_ready..s DDDAEEGGDDDr)rrrr handle_remote_call_result_errorsrX) rrar1rnrrrrrr[rrs rforeach_actor_async_fetch_readyz9FaultTolerantActorManager.foreach_actor_async_fetch_readys^44++% 5     %- !    "BB / C    ERR>3O3O3Q3QRRR RDD^%A%A%C%CDDD Drresults_or_errorsc|D]L}|jr |r-t|9|dS)aChecks given results for application errors and raises them if necessary. Args: results_or_errors: The results or errors to check. ignore_ray_errors: Whether to ignore RayErrors within the elements of `results_or_errors`. N)r#rfrgr%)rr[r0s rrz:FaultTolerantActorManager.handle_remote_call_result_errorssj 1 , ,O! ," ,  !4!4!6!67777&))+++ , ,rctj}fdD}g}|r(d|d|d|}d|D}j||zS)aPing all unhealthy actors to try bringing them back. Args: timeout_seconds: Timeout in seconds (to avoid pinging hanging workers indefinitely). mark_healthy: Whether to mark all those actors healthy again that are currently marked unhealthy AND that respond to the `ping` remote request (within the given `timeout_seconds`). Note that actors are NOT set to unhealthy, if they simply time out, meaning take a longer time to fulfil the remote request. We only ever mark and actor unhealthy, if they return a RayActorError from the remote request. Also note that this settings is ignored if `healthy_only=True` (b/c this setting only affects actors that are currently tagged as unhealthy). Returns: A list of actor IDs that were restored by the `ping.remote()` call PLUS those actors that were previously restored via other remote requests. The cached set of such previously restored actors will be erased in this call. c>g|]}||Sr,r)rTr/rs rrVzDFaultTolerantActorManager.probe_unhealthy_actors..(s=   ((22    rc*|Sr!)r`)rs rrzBFaultTolerantActorManager.probe_unhealthy_actors..2s5::<<rF)rarrrrrc*g|]}|j |jSr,)r#r/)rTrs rrVzDFaultTolerantActorManager.probe_unhealthy_actors..9s1$$$$* $$$$r)rrrrr)rrralready_restored_actorsunhealthy_actor_idsjust_restored_actorsrs` rprobe_unhealthy_actorsz0FaultTolerantActorManager.probe_unhealthy_actorss:#'t'<"="=     NN,,    "  !////!4" / %) 0N$$.<$$$ ##%%%')===r)rnrc ||}g}t|trt|t|ks Jdt|ts Jdt t ||D]\}\}}t|t rY|t|j ||j dit|tr||n|piv||j |j |nt|t rnt |D]]\}}|t|j ||j dit|tr||n|pi^n9|D]6}||j |j j dd|i|pi7|S)aApply functions on a list of remote actors. Args: func: A single Callable applied to all specified remote actors or a list of Callables, that get applied on the list of specified remote actors. In the latter case, both list of Callables and list of specified actors must have the same length. Alternatively, you can use the name of the remote method to be called, instead, or a list of remote method names. kwargs: An optional single kwargs dict or a list of kwargs dict matching the list of provided `func` or `remote_actor_ids`. In the first case (single dict), use `kwargs` on all remote calls. The latter case (list of dicts) allows you to define individualized kwarg dicts per actor. remote_actor_ids: Apply func on this selected set of remote actors. Returns: A list of ObjectRefs returned from the remote calls. Nz>Funcs must have the same number of callables as actor indices.zBIf func is a list of functions, kwargs has to be a list of kwargs.rar,) rrrrOrrr6rJgetattrrremoterp)rrarnrcallsrrrs rrz&FaultTolerantActorManager._call_actorsDsu0  ##~~//  dD ! ! Y'((C--O T TS T T!*#.>*E*E F F E E 9D!a%% ELL= T 2A66=$.fd#;#;!4q &,l LLd!3!9!@!@!C!CDDDD Ec " " Y$%566  4  > >))(M4J4J4JCPPP((22LLXc!ffXXhXXX$$Xu$======== >87||s>222222&&sE G+!A?G&&G+)rnct|trt|t|ks Jdg}g}g}tt ||D]}\}\}} | r`t|tr||n|pi} |||| || ~|}|}|}nfd|D}|||fS)aFilter out func and remote worker ids by actor state. Args: func: A single, or a list of Callables. kwargs: An optional single kwargs dict or a list of kwargs dicts matching the list of provided `func` or `remote_actor_ids`. In case of a single dict, uses `kwargs` on all remote calls. In case of a list of dicts, the given kwarg dicts are per actor `func` or per `remote_actor_ids`. remote_actor_ids: IDs of potential remote workers to apply func on. Returns: A tuple of (filtered func, filtered remote worker ids). z@Func must have the same number of callables as remote actor ids.c>g|]}||Sr,rrs rrVzFFaultTolerantActorManager._filter_by_healthy_state..s-XXXat?T?TUV?W?WXXXXr)rrrOrrrrJ) rrarnr temp_functemp_remote_actor_ids temp_kwargsrrrrs ` rrz2FaultTolerantActorManager._filter_by_healthy_statesB( dD ! ! Y'((C--Q I$& !K )#d4D*E*E F F 7 7 9At((..7%/%=%=Qq FLbA$$Q'''&&q))))00666D F4   YXXX+;XXX V---rc|t}net|tr|h}nLt|ttfrt|}n t dt |dg}g}g}|jD]^\}\}}t|dks||vr?| || || |_|||fS)aReturn all the in flight requests that match the given tags, if any. Args: tags: A str or a list/tuple of str. If tags is empty or None, return all the in flight requests. Returns: A tuple consisting of a list of the remote calls that match the tag(s), a list of the corresponding remote actor IDs for these calls (same length), and a list of the tags corresponding to these calls (same length). Nz6tags must be either a str or a list/tuple of str, got .r) rrr6rtuplertyperrrOrJ)rrrrrrr1r/s rrz.FaultTolerantActorManager._filter_calls_by_tags <55DD c " " 6DD tUm , , t99DDVdVVV   %)%D%J%J%L%L ' ' !D/34yyA~~##D))) ''111!!#&&&-z99rct|jD]\}\}}||kr|j|=||jvr&|j|jdSdS)zRemove internal async state of for a given actor. This is called when an actor is removed from the pool or being marked unhealthy. Args: actor_id: The id of the actor. N)rrrrrvr)rr/reqr1rs rrz-FaultTolerantActorManager._remove_async_state6s~#4#B#H#H#J#JKK 9 9NC#rX~~3C8 t0 0 0  % 1%%''''' 1 0rc|jSr!)rr"s rrz FaultTolerantActorManager.actorsLs |r)Nrrr!)NF)/r&r'r(r)rrr r rr4rrrrrrrrrr6rrrrrr rrrrr8rrr rr staticmethodrrr ObjectRefrrrrrrr,rrrrrrsJ ##JFFFFFFFYF>/378 %%%%k*+%%25%% %%%%%%%%N)49)))\)Q49QQQ\Q k!2   \ S[\&!C!!!\!MCMMM\M(C(((\(  hsm s   \  > > > > >\ >//d/t///\/000\0 IM!04+/ %"JJJHcUCZ($xs /C*Dc4PS9TUJtCH~tDcN/CCDE J  J #49- J"%JJJ JJJ\JX"y! IM!04y!y!y!HcUCZ($xs /C*Dc4PS9TUy!c]y! tCH~tDcN/CCDE y!  y!#49-y! y!y!y!\y!v8:+. %" 888CcE#s(O348"% 8  8  8 888\8t"GE IM+. %"!04"&!&GEGEGEHcUCZ($xs /C*Dc4PS9TUGEc]GE tCH~tDcN/CCDE GE "% GEGEGEGE#49-GE GEGE eE#s(OS() *GEGEGE\GER,,, ,  ,,,\,.,0"9>9>!%9>9> c 9>9>9>\9>~IM&* ===HcUCZ($xs /C*Dc4PS9TU=tCH~tDcN/CCDE = s) = cm  ====~,0 %"`'`'`'s)`'3=) `' 3i `' "% `'`'`' tCM"$55 6`'`'`'\`'L59 *.*.*.HcUCZ($xs /C*DDE*.tT$Z/01 *. s) *.*.*.*.ZHL":":U3S 5c?#BCD": tCM"D$5tCy@ A":":":":H:C::::,rrr)$rMloggingrkrh collectionsr dataclassesrrtypingrrrr r r r r r ray.actorrray.exceptionsrrray.rllib.utils.typingrray.util.annotationsr getLoggerr&rfrr.r8r]rrr,rrrs%  ######((((((((NNNNNNNNNNNNNNNNNNNN !!!!!!11111111$$$$$$------  8 $ $          F ******* **D D D D D D D D N++++++++\JJJJJJJJJJr