o `۷iuv@sddlZddlZddlmZmZmZmZmZmZm Z m Z m Z ddl Z ddl mZddlmZddlmZmZmZmZddlmZddlmZddlmZmZdd lmZdd lm Z dd l!m"Z"erhdd l#m$Z$e%e&Z'e d Z(e"Gdddej)dZ*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)AlgorithmConfigTc@seZdZ     dbdddeedeed eed ed ed eeefd dfddZ ddddddeddedeedeed eeefd df ddZ dcdeded dfddZ d e fddZ dddedededdd ef ddZdddddddddddeed eed!eeeeefd"eeeefd#eeef d$d%Z    & ddd'eeed(eeed)fd*eeed+eed,eed df d-d.Zd/eed dfd0d1Zded2d3Zdddddddd4d5eeegefeeegefeeefded6ed7eed+eed8ed9ed eefd:d;Zddd<d5eeegefeeegefeeefd6ed7eed efd=d>Zd&ddd?d+eed8ed9ed eeeeffd@dAZd eefdBdCZe e!j"dDdEZ#e e!j"d efdFdGZ$e dfdHdIZ%e d efdJdKZ&e d eefdLdMZ'e e!j"d efdNdOZ(e d efdPdQZ)e d efdRdSZ*e d efdTdUZ+e d efdVdWZ,e d efdXdYZ-e e!j"dZd[Z.e e!j"d\d]Z/e e!j"d^d_Z0e e!j"d`daZ1dS)g RunnerGroupFNrTconfigr local_runnerlogdir tune_trial_id pg_offset_setupkwargsreturnc Kst|tr t|n|pt|_||_t|j|_||_ ||_ ||_ t |j dd|_|rWz|jd||j|d|WdStyV}z |jrP|jdjd|d}~wwdS)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) selfrrrrrr r!er(r(Z/home/ubuntu/vllm_env/lib/python3.10/site-packages/ray/rllib/utils/runners/runner_group.py__init__*s:   zRunnerGroup.__init__)rr&rvalidater&r;cKsdd|_|dkr d}||_|j|fd|dur|n|ji||r0|jdd||jd||_dSdS)NrTr; runner_indexr&rr() _local_runner_RunnerGroup__local_config add_runners$_validate_runners_after_construction _make_runner _local_config)r7rr&rr;r!r(r(r9r js,  zRunnerGroup._setupc sFjfddtD}j||r!dSdS)z?Creates and adds a number of remote runners to this runner set.cs2g|]}jd|djdqS)r#r<r()rBr/).0ir!r&old_num_runnersr7r(r9 s  z+RunnerGroup.add_runners..N)r4 num_actorsrange add_actorsr;)r7r&r;r! new_runnersr(rFr9r@s    zRunnerGroup.add_runnerscCsR|jddD]}|js&|}|jr$td|jjdt |q|qdS)NcS|SN)assert_healthywr(r(r9z&RunnerGroup.validate..zValidation of z failed! Error=) r4 foreach_actorokget_ignore_ray_errors_on_runnersloggererror runner_cls__name__str)r7resultr8r(r(r9r;szRunnerGroup.validate)recreated_runnerr=r^c Kstd|||||j|jd|}|dkr|jdi|Stjdur%dn|j|}tjdi|j |jj |djdi|S)N)r worker_index num_workersrecreated_workerlog_dirrr)placement_group_bundle_indexr() r*r2r0rZr-utilget_current_placement_groupr1remote _remote_argsoptions)r7r=r&r^rr! pg_bundle_idxr(r(r9rBs0  zRunnerGroup._make_runner) from_runnerenv_steps_sampledconnector_statesrl_module_staterunner_indices_to_update env_to_module module_to_envrkrlrmrnrocKs(|p|j}|jp|jdko|j} |j} |jdkr/|jr/|ji|dur't|ini|p,i| s5| s5dS| r|gkr>i} nt|durM|jddd|jd}dd |D} d d |D}|jdurt |jd rt |jd r|dusrJ|jj }|dus|J|jj }i} | r| t || i|r| t||in|durt |t|i} n|jt tgd } |dur||jpd| t<| s|jdur|j| n|| t i|| ti| t d| td| r|jdur| r|j| |r| ||jdt| d|ddddSdS)z>Synchronizes the connectors of this `RunnerGroup`'s `Runner`s. training_onlyrNcSs|jttgdS)N components) get_stater rrPr(r(r9rRs z0RunnerGroup.sync_runner_states..F)rtimeout_secondscSg|] }t|vr|tqSr()r rDsr(r(r9rH z2RunnerGroup.sync_runner_states..cSrwr()rrxr(r(r9rHrz_env_to_module_module_to_envrsr# set_statestate)r!remote_worker_idsrrv)rmerge_runner_states in_evaluationbroadcast_runner_statesnum_healthy_remote_runnersr}rforeach_runner)sync_filters_on_rollout_workers_timeout_shasattrr{r|updater merge_statesrrur&rVpopr*)r7rrkrlrmrnrorprqr!merge broadcast runner_statesenv_to_module_statesmodule_to_env_statesr(r(r9sync_runner_statess                  zRunnerGroup.sync_runner_statesrpoliciesfrom_worker_or_learner_grouprto_worker_indicesrvinference_onlyc Ks"|jdur |dur tdd}|js|dur||dur|n|j}|dur'td|dur2dd|Dntg} t|trI|jdd| D|dt}nt|t rZt |jj | |d}n|j| |d}dd | D}t |} |jd t| d d ||d |jdur|dur|j|dSdSdS)aSyncs model weights from the given weight source to all remote workers. Weight source can be either a (local) rollout worker or a learner_group. It should just implement a `get_weights` method. Args: policies: Optional list of PolicyIDs to sync weights for. If None (default), sync weights to/from all policies. from_worker_or_learner_group: Optional (local) `Runner` instance or LearnerGroup instance to sync from. If None (default), sync from this `Runner`Group's local worker. to_worker_indices: Optional list of worker indices to sync the weights to. If None (default), sync to all remote workers. global_vars: An optional global vars dict to set this worker to. If None, do not update the global_vars. timeout_seconds: Timeout in seconds to wait for the sync weights calls to complete. Default is 0.0 (fire-and-forget, do not wait for any sync calls to finish). Setting this to 0.0 might significantly improve algorithm performance, depending on the algo's `training_step` logic. inference_only: Sync weights with workers that keep inference-only modules. This is needed for algorithms in the new stack that use inference-only modules. In this case only a part of the parameters are synced to the workers. Default is False. NzhNo `local_runner` in `RunnerGroup`! Must provide `from_worker_or_learner_group` arg in `sync_weights()`!z}`from_worker_or_trainer` is None. In this case, `RunnerGroup`^ should have `local_runner`. But `local_runner` is also `None`.cSg|]}td|qS/)r)rDpr(r(r9rHz,RunnerGroup.sync_weights..cSrr)r)rDmr(r(r9rHr)rtrcSs"i|] \}}|ttfvr||qSr()rr)rDkvr(r(r9 s  z,RunnerGroup.sync_weights..r}r~F)funcr!rrrv)r TypeErrornum_remote_runners ValueErrorrr)rrurr r-rVrgitemsr.rr*r}) r7rrrrvrr!rn weights_srcmodulesrl_module_state_refr(r(r9 sync_weightszsh"     zRunnerGroup.sync_weightsnew_remote_runnerscCs|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)r4clearrK)r7rr(r(r9resets zRunnerGroup.resetcCsdz+z |jdddddWntytdYn wW|jdSW|jdS|jw)z8Calls `stop` on all `Runner`s (including the local one).cSrMrN)stoprPr(r(r9rRrSz"RunnerGroup.stop..FT) healthy_onlyrzFailed to stop workers!N)r ExceptionrX exceptionr4rr7r(r(r9rs  zRunnerGroup.stop)r!rrrrvreturn_obj_refs mark_healthyrrrrrc Cs|r|rJdg} |r>|jdur>|r|d} |dd}ni} |}t|tr4t|j|d i| g} n ||jfi| g} |jsE| S|jj|||||||d} tj| |j ddd| D} | | S) 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_idsrvrrignore_ray_errorscSsg|]}|qSr()rVrDrr(r(r9rHQsz.RunnerGroup.foreach_runner..r() rr)r\getattrr4 actor_idsrTr handle_remote_call_result_errorsrW ignore_errors) r7rr!rrrrvrr local_result local_kwargsremote_resultsr(r(r9rs@)   zRunnerGroup.foreach_runner)rrcCs|jj|||dS)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)r4foreach_actor_async)r7rrrr(r(r9foreach_runner_asyncUs z RunnerGroup.foreach_runner_asyncrvrrcCs4|jj|||d}tj||jddd|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. rrcSsg|] }|j|fqSr()actor_idrVrr(r(r9rHsz6RunnerGroup.fetch_ready_async_reqs..)r4fetch_ready_async_reqsrrrWr)r7rvrrrr(r(r9rtsz"RunnerGroup.fetch_ready_async_reqscCs|jj|jddS)zChecks for unhealthy workers and tries restoring their states. Returns: List of IDs of the workers that were restored. T)rvr)r4probe_unhealthy_actorsrunner_health_probe_timeout_srr(r(r9probe_unhealthy_runnerssz#RunnerGroup.probe_unhealthy_runnerscCdS)z>Number of seconds to wait for health probe calls to `Runner`s.Nr(rr(r(r9rz)RunnerGroup.runner_health_probe_timeout_scCr)zClass for each runner.Nr(rr(r(r9rZrzRunnerGroup.runner_clscC|jS)z(Returns the config for a local `Runner`.)r?rr(r(r9rCzRunnerGroup._local_configcCr)zReturns the local `Runner`.)r>rr(r(r9rrzRunnerGroup.local_runnercC |jS)z(Returns the list of remote `Runner` IDs.)r4healthy_actor_idsrr(r(r9healthy_runner_ids zRunnerGroup.healthy_runner_idscCr)z)Number of runners to schedule and manage.Nr(rr(r(r9r&rzRunnerGroup.num_runnerscCr)zNumber of remote `Runner`s.)r4rIrr(r(r9rrzRunnerGroup.num_remote_runnerscCr)z/Returns the number of healthy remote `Runner`s.)r4num_healthy_actorsrr(r(r9rrz&RunnerGroup.num_healthy_remote_runnerscCstt|j|S)z(Returns the number of healthy `Runner`s.)intboolr>rrr(r(r9num_healthy_runnersszRunnerGroup.num_healthy_runnerscCr)z/Returns the number of in-flight async requests.)r4num_outstanding_async_reqsrr(r(r9num_in_flight_async_reqsrz$RunnerGroup.num_in_flight_async_reqscCr)zIReturns the number of times managed remote `Runner`s have been restarted.)r4total_num_restartsrr(r(r9num_remote_runner_restartsrz&RunnerGroup.num_remote_runner_restartscCr)z!Remote arguments for each runner.Nr(rr(r(r9rhrzRunnerGroup._remote_argscCr)z'If errors in runners should be ignored.Nr(rr(r(r9rWrz)RunnerGroup._ignore_ray_errors_on_runnerscCr)z&Maximum requests in flight per runner.Nr(rr(r(r9r3rz.RunnerGroup._max_requests_in_flight_per_runnercCr)z.If runners should validated after constructed.Nr(rr(r(r9rArz0RunnerGroup._validate_runners_after_construction)FNNrT)F)NNNrF)r"N)r"r)2r[ __module__ __qualname__rrr\rrrr:r r@rr;r rBrrrrr floatrrrrrrrrrrpropertyabcabstractmethodrrZrCrrr&rrrrrrhrWr3rAr(r(r(r9r(sh     C  " )  '   t *   S* " '      r) metaclass)+rloggingtypingrrrrrrrr r r- ray.actorr ray.exceptionsr ray.rllib.corer rrr$ray.rllib.core.learner.learner_grouprray.rllib.utils.actor_managerrray.rllib.utils.metricsrrray.rllib.utils.runners.runnerrray.rllib.utils.typingrray.util.annotationsr%ray.rllib.algorithms.algorithm_configr getLoggerr[rXrABCMetarr(r(r(r9s&,