&`iuvDddlZddlZddlmZmZmZmZmZmZm Z m Z m Z ddl Z ddl mZddlmZddlmZmZmZmZddlmZddlmZddlmZmZdd lmZdd lm Z dd l!m"Z"erdd l#m$Z$ej%e&Z'e d Z(e"Gddej)Z*dS)N) TYPE_CHECKINGAnyCallableDictListOptionalTupleTypeVarUnion) ActorHandle) RayActorError)!COMPONENT_ENV_TO_MODULE_CONNECTORCOMPONENT_LEARNER!COMPONENT_MODULE_TO_ENV_CONNECTORCOMPONENT_RL_MODULE) LearnerGroup)FaultTolerantActorManager)NUM_ENV_STEPS_SAMPLED_LIFETIMEWEIGHTS_SEQ_NO)Runner)PolicyID) DeveloperAPI)AlgorithmConfigTcDeZdZ dFdddeedeed eed ed ed eeefd dfdZ ddddddeddedeedeed eeefd df dZ dGdeded dfdZ d e fdZ dddedededdd ef dZdddddddddddeedeedeeeeefdeeeefdeeef dZ dHd!eeed"eeed#fd$eeed%eed&eed df d'Zd(eed dfd)ZdId*Zdddddddd+d,eeegefeeegefeeefded-ed.eed%eed/ed0ed eefd1Zddd2d,eeegefeeegefeeefd-ed.eed efd3Zd ddd4d%eed/ed0ed eeeeffd5Zd eefd6Ze e!j"d7Z#e e!j"d efd8Z$e dJd9Z%e d efd:Z&e d eefd;Z'e e!j"d efd<Z(e d efd=Z)e d efd>Z*e d efd?Z+e d efd@Z,e d efdAZ-e e!j"dBZ.e e!j"dCZ/e e!j"dDZ0e e!j"dEZ1dS)K RunnerGroupFNrTconfigr local_runnerlogdir tune_trial_id pg_offset_setupkwargsreturnc t|trtj|n|p t|_||_t j|j|_||_ ||_ ||_ t|j d|_|rL |jd||j|d|dS#t"$r&}|jr|jdjd|d}~wwxYwdS)N)'max_remote_requests_in_flight_per_actorinit_id)r num_runnersrr) isinstancedictr from_dictr_remote_configrayput_remote_config_obj_ref_tune_trial_id _pg_offset_logdirr"_max_requests_in_flight_per_runner_worker_managerr"r)r actor_init_failedargs) selfrrrr r!r"r#es x/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/rllib/utils/runners/runner_group.py__init__zRunnerGroup.__init__*s3$&$'' /O %f - - --O-- %&)gd.A&B&B#+# 8484[       ! $ 0!- !    & &).++G!   sB// C9!CC)rr)rvalidater)r>c d|_|dkrd}||_|j|fd||n|ji||r|jdd||jd||_dSdS)NrTr> runner_indexr)rr+) _local_runner_RunnerGroup__local_config add_runners$_validate_runners_after_construction _make_runner _local_config)r:rr)rr>r#s r<r"zRunnerGroup._setupjs" !  L$    #X:        !2!2"')"" ""D     c jfdtD}j||rdSdS)z?Creates and adds a number of remote runners to this runner set.c Ng|]!}jd|zdzzjd"S)r&r@r+)rFr2).0ir#r)old_num_runnersr:s r< z+RunnerGroup.add_runners..sk     D  ,q014+k92         rHN)r7 num_actorsrange add_actorsr>)r:r)r>r# new_runnersrMs`` ` @r<rDzRunnerGroup.add_runnerss.99;;        ;''     '' 444   MMOOOOO  rHc |jdD]^}|jsU|}|jr8t d|jjdt|\|_dS)Nc*|SN)assert_healthyws r<z&RunnerGroup.validate..s1CSCSCUCUrHzValidation of z failed! Error=) r7 foreach_actorokget_ignore_ray_errors_on_runnersloggererror runner_cls__name__str)r:resultr;s r<r>zRunnerGroup.validates*889U9UVV  F9 JJLL5LLZ)AZZRUVWRXRXZZG   rH)recreated_runnerrArdc <td|||||j|jd|}|dkr |jdi|Stjdn |j|z}t jdi|j |j |jdi|S)N)r worker_index num_workersrecreated_workerlog_dirr r)placement_group_bundle_indexr+) r-r5r3r`r0utilget_current_placement_groupr4remote _remote_argsoptions)r:rAr)rdrr# pg_bundle_idxs r<rFzRunnerGroup._make_runners %#-L-      1  "4?,,V,, , x3355= B</    +CJ + +* + +DO < < W-W @ @      rH) from_runnerenv_steps_sampledconnector_statesrl_module_staterunner_indices_to_update env_to_module module_to_envrrrsrtrurvc |p|j}|jp|jdko|j} |j} |jdkr2|jr+|ji| t |ini|pi| s| sdS| r|gkri} nG||dd|j}d|D} d|D}|jJt|jd r5t|jd r |J|jj }|J|jj }i} | r/| t|| i|r/| t||inY|5t|t|i} n"|ttg } |||jpd z| t <| s|j|j| n\|| ti|| ti| td| td| rd|j| r|j| |r| ||d t)| |dddSdS)z>Synchronizes the connectors of this `RunnerGroup`'s `Runner`s. training_onlyrNcF|ttgS)N components) get_staterrrWs r<rYz0RunnerGroup.sync_runner_states..s$!++ A A(#.##rHF)rtimeout_secondsc:g|]}t|v |tSr+)rrKss r<rNz2RunnerGroup.sync_runner_states..3(((8A==78===rHc:g|]}t|v |tSr+)rrs r<rNz2RunnerGroup.sync_runner_states..rrH_env_to_module_module_to_envr|r& set_statestate)r#remote_worker_idsrr)rmerge_runner_states in_evaluationbroadcast_runner_statesnum_healthy_remote_runnersrrforeach_runner)sync_filters_on_rollout_workers_timeout_shasattrrrupdater merge_statesrr~r)r\popr-)r:rrrrsrtrurvrwrxr#merge broadcast runner_statesenv_to_module_statesmodule_to_env_statess r<sync_runner_stateszRunnerGroup.sync_runner_statess "6T%6 *  &/ 9 Rf>R 2  *a / /D4E /   ' '-889JKK  ',"     Y  F E 2%% " #+'+':': &+"L(; ( ($((-((($ ((-((($%1 13CDD2 13CDD2)000$($5$DM(000$($5$DM " '!((= - : :;O P P(!((= - : :;O P P"5}7N7N7P7P5}7N7N7P7P! !, 5 599 !6!!   (.s#AAA1$s*Q.AAArHc(g|]}tdz|zSr)r)rKms r<rNz,RunnerGroup.sync_weights..s#MMM 1C 7! ;MMMrH)r}rc:i|]\}}|ttfv||Sr+)rr)rKkvs r< z,RunnerGroup.sync_weights..s9Aq,n===1===rHrrF)funcr#rrr)r TypeErrornum_remote_runners ValueErrorrr,rr~rr r0r\rnitemsr1rr-r) r:rrrrrr#ru weights_srcmodulesrl_module_state_refs r< sync_weightszRunnerGroup.sync_weightszsD   $)E)MJ    "B &B&N0;-,&  " U'BAAAAA)*  +|44 "-"7"7MMWMMM#1#8##$#%k;77 &)g#-44'.+95''OO'2&;&;#*'5'<''O+1133O#&'/":":     "5666""3 /       (+7!++O<<<<< ) (77rHnew_remote_runnerscl|j|j|dS)zHard overrides the remote `Runner`s in this set with the provided ones. Args: new_remote_workers: A list of new `Runner`s (as `ActorHandles`) to use as new remote workers. N)r7clearrQ)r:rs r<resetzRunnerGroup.resets7 ""$$$ ''(:;;;;;rHc |dddn*#t$rtdYnwxYw|jdS#|jwxYw)z8Calls `stop` on all `Runner`s (including the local one).c*|SrU)stoprWs r<rYz"RunnerGroup.stop..s!&&((rHFT) healthy_onlyrzFailed to stop workers!N)r Exceptionr^ exceptionr7rr:s r<rzRunnerGroup.stops )   ""T      8 8 8   6 7 7 7 7 7 8  & & ( ( ( ( (D & & ( ( ( (s&A!$AA!AA!!A<)r#rrrrreturn_obj_refs mark_healthyrrrrrc |r |r Jdg} |ra|jZ|r|d} |dd}ni} |}t|trt|j|di| g} n||jfi| g} |js| S|j|||||||} tj| |j d| D} | | zS) aCalls the given function with each `Runner` as its argument. Args: func: The function to call for each `Runner`s. The only call argument is the respective `Runner` instance. local_env_runner: Whether to apply `func` to local `Runner`, too. Default is True. healthy_only: Apply `func` on known-to-be healthy `Runner`s only. remote_worker_ids: Apply `func` on a selected set of remote `Runner`s. Use None (default) for all remote `Runner`s. 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 this `RunnerGroup`. mark_healthy: Whether to mark all those `Runner`s healthy again that are currently marked unhealthy AND that returned results from the remote call (within the given `timeout_seconds`). Note that `Runner`s are NOT set unhealthy, if they simply time out (only if they return a `RayActorError`). Also note that this setting is ignored if `healthy_only=True` (b/c `mark_healthy` only affects `Runner`s that are currently tagged as unhealthy). Returns: The list of return values of all calls to `func([worker])`. z-Can not return `ObjectRef` from local worker.Nrr&)r#rremote_actor_idsrrrignore_ray_errorsc6g|]}|Sr+)r\rKrs r<rNz.RunnerGroup.foreach_runner..Qs JJJa!%%''JJJrHr+) rr,rbgetattrr7 actor_idsrZr handle_remote_call_result_errorsr] ignore_errors) r:rr#rrrrrr local_result local_kwargsremote_resultss r<rzRunnerGroup.foreach_runnerscR  ;'3 ; ; : ; ; 3  ID-9 %ay ! $$$ I @(94 @ @ P P< P PQ $T%6 G G, G GH #--//  -;; %.++%<   "B d.P    KJ>+G+G+I+IJJJn,,rH)rrc<|j|||S)aCalls the given function asynchronously with each `Runner` as the argument. Does not return results directly. Instead, `fetch_ready_async_reqs()` can be used to pull results in an async manner whenever they are available. Args: func: The function to call for each `Runner`s. The only call argument is the respective `Runner` instance. healthy_only: Apply `func` on known-to-be healthy `Runner`s only. remote_worker_ids: Apply `func` on a selected set of remote `Runner`s. Returns: The number of async requests that have actually been made. This is the length of `remote_worker_ids` (or self.num_remote_workers()` if `remote_worker_ids` is None) minus the number of requests that were NOT made b/c a remote `Runner` already had its `max_remote_requests_in_flight_per_actor` counter reached. )rr)r7foreach_actor_async)r:rrrs r<foreach_runner_asyncz RunnerGroup.foreach_runner_asyncUs-2#77 %.8   rHrrrc|j|||}tj||jd|DS)aGet esults from outstanding asynchronous requests that are ready. Args: timeout_seconds: Time to wait for results. Default is 0, meaning those requests that are already ready. return_obj_refs: Whether to return ObjectRef instead of actual results. mark_healthy: Whether to mark all those workers healthy again that are currently marked unhealthy AND that returned results from the remote call (within the given `timeout_seconds`). Note that workers are NOT set unhealthy, if they simply time out (only if they return a RayActorError). Also note that this setting is ignored if `healthy_only=True` (b/c `mark_healthy` only affects workers that are currently tagged as unhealthy). Returns: A list of results successfully returned from outstanding remote calls, paired with the indices of the callee workers. rrcDg|]}|j|fSr+)actor_idr\rs r<rNz6RunnerGroup.fetch_ready_async_reqs..s'NNN!QUUWW%NNNrH)r7fetch_ready_async_reqsrrr]r)r:rrrrs r<rz"RunnerGroup.fetch_ready_async_reqstss4-DD++%E   "B "@    ON~/K/K/M/MNNNNrHcD|j|jdS)zChecks for unhealthy workers and tries restoring their states. Returns: List of IDs of the workers that were restored. T)rr)r7probe_unhealthy_actorsrunner_health_probe_timeout_srs r<probe_unhealthy_runnersz#RunnerGroup.probe_unhealthy_runnerss- #:: >;   rHcdS)z>Number of seconds to wait for health probe calls to `Runner`s.Nr+rs r<rz)RunnerGroup.runner_health_probe_timeout_srHcdS)zClass for each runner.Nr+rs r<r`zRunnerGroup.runner_clsrrHc|jS)z(Returns the config for a local `Runner`.)rCrs r<rGzRunnerGroup._local_configs ""rHc|jS)zReturns the local `Runner`.)rBrs r<rzRunnerGroup.local_runners !!rHc4|jS)z(Returns the list of remote `Runner` IDs.)r7healthy_actor_idsrs r<healthy_runner_idszRunnerGroup.healthy_runner_idss#55777rHcdS)z)Number of runners to schedule and manage.Nr+rs r<r)zRunnerGroup.num_runnersrrHc4|jS)zNumber of remote `Runner`s.)r7rOrs r<rzRunnerGroup.num_remote_runnerss#..000rHc4|jS)z/Returns the number of healthy remote `Runner`s.)r7num_healthy_actorsrs r<rz&RunnerGroup.num_healthy_remote_runners#66888rHcntt|j|zS)z(Returns the number of healthy `Runner`s.)intboolrBrrs r<num_healthy_runnerszRunnerGroup.num_healthy_runnerss.4*++,,t/N/N/P/PPPrHc4|jS)z/Returns the number of in-flight async requests.)r7num_outstanding_async_reqsrs r<num_in_flight_async_reqsz$RunnerGroup.num_in_flight_async_reqss#>>@@@rHc4|jS)zIReturns the number of times managed remote `Runner`s have been restarted.)r7total_num_restartsrs r<num_remote_runner_restartsz&RunnerGroup.num_remote_runner_restartsrrHcdS)z!Remote arguments for each runner.Nr+rs r<rozRunnerGroup._remote_argsrrHcdS)z'If errors in runners should be ignored.Nr+rs r<r]z)RunnerGroup._ignore_ray_errors_on_runnersrrHcdS)z&Maximum requests in flight per runner.Nr+rs r<r6z.RunnerGroup._max_requests_in_flight_per_runnerrrHcdS)z.If runners should validated after constructed.Nr+rs r<rEz0RunnerGroup._validate_runners_after_constructionrrH)FNNrT)F)NNNrF)r$N)r$r)2ra __module__ __qualname__rrrbrrrr=r"rDrr>r rFrrrrr floatrrrrrrrr rrpropertyabcabstractmethodrr`rGrrr)rrrrrror]r6rEr+rHr<rr(s(- $'+>>!> tn > > }>>>sCx.> >>>>F/3',#'   *+   tn  4. sCx.      DsdQU0 )    $"' # # # #  #  # " #  # # # # R)-+/;?488<bbb"bf% b $C= b #4S#X#78 b"$sCx.1b#+49"5bbbbL.2PT15+.). r=r=4>*r='/uV^5K/L&Mr=$DI. r= "% r= ! r= r=r=r=r=h<[(9r s                        !!!!!!(((((( >=====CCCCCCRRRRRRRR111111++++++------FEEEEEE  8 $ $ GCLLF =F =F =F =F =CKF =F =F =F =F =F =rH