&`iddlZddlZddlZddlmZddlmZmZmZm Z m Z m Z m Z m Z mZddlZddlmZddlmZddlmZddlmZmZddlmZerdd lmZejeZ ed Gd d Z!ed Gdde!Z"e!dZ#e!dZ$ed GddeZ%dS)N)deepcopy) TYPE_CHECKINGAnyCallableDictListOptionalSetTupleUnion _sum_bundles)PlacementGroupFactory)Trial) FIFOSchedulerTrialScheduler) PublicAPI)TuneControllerbeta) stabilitycXeZdZdZ d7dedeeeefde deeeeffd Z d e d eee fd efd Z ddd eeeffdZded eeeffdZdeeeefd eeeffdZdeeefd efdZ d8deeeefdeeeefdeeefdede f dZ d9deeefdedede d e f d Zdeeeefdeeefd!e d eeeeffd"Zdeeefd!e d eeeeffd#Zddd$deeeefdeeefd%ed&ede de f d'Zdeeeefd(eeeefd eeeeffd)Z d:deeeefd(eeeeefd*eed+eefd,Zd-ed.eed(eeeefdeeefd/ed0ed1ed2ed eeeeffd3Zddd-ed eee fd4d5d ee f d6ZdS);DistributeResourcesaeThis class creates a basic uniform resource allocation function. The function naively balances free resources (CPUs and GPUs) between trials, giving them all equal priority, ensuring that all resources are always being used. The free resources will be placed in new bundles. The function assumes that all bundles are equal (there is no "head" bundle). If for some reason a trial ends up with more resources than there are free ones, it will adjust downwards. It will also ensure that trial as at least as many resources as it started with (``base_trial_resource``). The function returns a new ``PlacementGroupFactory`` with updated resource requirements, or None. If the returned ``PlacementGroupFactory`` is equal by value to the one the trial has currently, the scheduler will skip the update process internally (same with None). If you wish to implement your own resource distribution logic, you can do so by extending this class, as it provides several generic methods. You can also implement a function instead. Args: add_bundles: If True, create new bundles from free resources. Otherwise, spread them among base_trial_resource bundles. increase_by: A dict with key-value pairs representing an atomic unit of resources (name-amount) the trial will be increased by. If not set, the trial will increase by 1 CPU/GPU. increase_by_times: If set to >=1 and ``increase_by`` is set, the trial will increase by maximum of ``increase_by_times * increase_by`` resources. If set to <1, no upper limit is set. Ignored if ``increase_by`` is not set. reserve_resources: A dict of resource_name-amount pairs representing the resources that will not be allocated to resized trials. FN add_bundles increase_byincrease_by_timesreserve_resourcescF||_|pi|_||_|pi|_dSN)rrrr)selfrrrrs /home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/tune/schedulers/resource_changing_scheduler.py__init__zDistributeResources.__init__>s4'&,"!2!2!8bbase_trial_resourceresultreturnc t|tst|jjd|js=t |jdkr%tdt |jd|ddkrdSdS)z>Return False if we should keep the current resources outright.z' only supports PlacementGroupFactories.zYIf `add_bundles` is False, the number of bundles in `resources_per_trial` must be 1 (got z).training_iterationFT) isinstancer ValueError __class____name__rlenbundles)r r$r%s r! _validatezDistributeResources._validateJs-/DEE >*SSS  C(;(C$D$Dq$H$H=/788===  & '! + +5tr#tune_controllerrc|j|jddz }|j|jddz }||fS)zBGet the number of CPUs and GPUs avaialble in total (not just free)CPUrGPU)_resource_updater get_num_cpusrget get_num_gpus)r r1total_available_cpustotal_available_gpuss r!_get_total_available_resourcesz2DistributeResources._get_total_available_resources_sx  - : : < <$((22 3   - : : < <$((22 3 $%999r#tc|jjdd|jjddfS)z7Check how many CPUs and GPUs a trial is using currentlyr3rr4)placement_group_factoryrequired_resourcesr7)r r<s r!_get_used_cpus_and_gpusz+DistributeResources._get_used_cpus_and_gpusms@  % 8 < 0 and ``max_increase_by_times`` > 0, ensure that the resulting number of bundles is not above ``min(max_increase_by_times, self.increase_by_times)``. If ``limit_to_increase_by_times`` is True and ``self.increase_by_times`` > 0, ensure that the resulting number of bundles is not above `self.increase_by_times``. rc>g|]}||SrF.0br s r! z8DistributeResources._add_two_bundles..s,LLLQ43H3H3K3KLqLLLr#c>g|]}||SrMrNrOs r!rRz8DistributeResources._add_two_bundles..sCPPPD,A,A!,D,DPPPPr#Nr3r4rB)rminrr7)r rGrHrrIrJr/s` r!_add_two_bundlesz$DistributeResources._add_two_bundless$ & ?$q((T-Ca-G-G(+)4+A))%%'!++(,(>%   LLLL)LLLPPPP$PPPG%q((!"8#8"89!)bTI!)bTI%Q<++E155 ! 8H8HPQ8R8RR$Q<++E155 ! 8H8HPQ8R8RRG %q(($'AJu%OOE1--0EE%% 5!%(AJu%OOE1--0EE%% 5! r#rcpusgpusmax_multiplierc|ddrU|ddr?t||ddz||ddz}nI|ddr||ddz}n||ddz}|dkr|dkrt||}t|S)zRGet how many times ``increase_by`` bundles occur in ``cpus`` and ``gpus``.r3rr4)r7rTint)r rrVrWrX multipliers r!_get_multiplierz#DistributeResources._get_multipliers ??5! $ $ ;)B)B ; q111 q111JJ__UA & & ;!:!::JJ!:!::J A  *q..^Z88J:r#r[ct| }|jr |d|}nt|}|ddxx|dd|zz cc<|ddxx|dd|zz cc<t |ddd|dd<t |ddd|dd<|S)z?Remove ``multiplier`` ``increase_by`` bundles from ``bundles``.Nrr3r4)absrrr7max)r r/rr[s r!_remove_bundlesz#DistributeResources._remove_bundless*oo%   :kzk*GGw''G AJu   !:!:Z!G G    AJu   !:!:Z!G G    #GAJu$5q 9 9GAJu  #GAJu$5q 9 9GAJu r#ct|}|jr|gt|z}nGig}|dd|z|dd<|dd|z|dd<|S)zICreate a list of new bundles containing ``increase_by`` * ``multiplier``.r3rr4)r^rrZr7)r rr[r/s r!_create_new_bundlesz'DistributeResources._create_new_bundless} __   G"mc*oo5GGdG +q 9 9J FGAJu  +q 9 9J FGAJu r#)rXrJ free_cpus free_gpusc|||||}|dkr||||}n5|dkr/|||}||||d|}|S)zUGiven free resources, increase/decrease the number of bundles in ``bundles``.rT)r\r`rbrU) r r/rrcrdrXrJr[bundles_to_adds r!#_modify_bundles_with_free_resourcesz7DistributeResources._modify_bundles_with_free_resourcess)) I~   >>**7KLLGG !^^!55k:NNN++d>%33l1o6I6I%QR6S6SS"1:>>%33l1o6I6I%QR6S6SSM r#max_added_cpusmax_added_gpusc|s;|jdkr$|jdd|jz}n tj}|s;|jdkr$|jdd|jz}n tj}||r|||n|}|dtj |kp$|dtj |k}|S)Nrr3r4)rrr7npinfrCrk)r r/rhrlrmadded_resourcesrets r!_are_bundles_below_limitz,DistributeResources._are_bundles_below_limits (%))!%!1!5!5eQ!?!?$BX!X!# (%))!%!1!5!5eQ!?!?$BX!X!#::>J WD # #G\ : : :PW     w / /. @ D""526'22^C  r#trial all_trialsr9r: used_cpus used_gpusc dtt|D} ||z } ||z } ||} || ddt|zz } || ddt|zz }d}d}t }t|t|kr| |ddkrm||ddkrR|t| z}t | |}|| ||| |d| |<||| ||}|dds,|dds||n |dkr|dz }| |ddz} ||ddz}|dz }t|t|kr5| |ddkr||ddkR|||j j ||| | |S)zReturns updated added bundles.c*g|]}tSrM)list)rP_s r!rRz>DistributeResources._get_new_added_bundles..?s)Q)Q)QQ$&&)Q)Q)Qr#r3rr4r()rX)rJ) ranger.rCr7setrrgrkaddr>r/)r rtrurhrr9r:rvrwupper_limit_all_trials_bundlesrcrdbase_resourcesupper_limit_cpus_to_distributeupper_limit_gpus_to_distributerJitrials_at_limitidx old_bundlesrqs r!_get_new_added_bundlesz*DistributeResources._get_new_added_bundles3s*R)Q%J:P:P)Q)Q)Q&(94 (94 99,GG)=   ua ( (3z?? :* &*>   ua ( (3z?? :* &!" %%  3z?? 2 2/+//%2K2KKK.+//%2K2KKKc8999C"#A##FGGK88.s3.. 9 + #>>''237O #&&ua00 +9L9LUTU9V9V +##C((((%*% *o.A.A%.K.K K * *o.A.A%.K.K K * FA9  3z?? 2 2/+//%2K2KKK.+//%2K2KKK877  # #-5|     "78   r# schedulerResourceChangingSchedulerc \|j}||sdS|tdddg}jrUj}|rJ|dddkr|dddksJn?jr|jd}n*|jdddrddi}nddi}t|j} | \}} | } fd | D} t| \} } t| } t| } || |||| | | }|||d }t|g|jRd |ji|j}|j|_|S) aRun resource allocation logic. Returns a new ``PlacementGroupFactory`` with updated resource requirements, or None. If the returned ``PlacementGroupFactory`` is equal by value to the one the trial has currently, the scheduler will skip the update process internally (same with None). Args: tune_controller: Trial runner for this Tune run. Can be used to obtain information about other trials. trial: The trial to allocate new resources to. result: The latest results of trial. scheduler: The scheduler calling the function. )r$r%Nr(rrBr3r4r)r1c:g|]}|SrM)r@)rPr<r s r!rRz0DistributeResources.__call__..s'RRR!d::1==RRRr#Fstrategy)base_trial_resourcesr0rrrFr7rr/rr;get_live_trialszipsumrrU_argsr_kwargs_head_bundle_is_empty)r r1rtr%rr$rrhr9r:ruused_cpus_and_gpusrvrwrj new_bundlespgfs` r!__call__zDistributeResources.__call__~s*2(<~~2Ef~UU 4  &"719M9M8N"O"O    %*K,,[99 9 99??5!,,11kooeQ6O6OST6T6T6TT6T   %-5b9KK ( + / /q 9 9 % !*KK !*K 3 ;<<  / / / P P %4466 RRRRzRRR"$67 9 NN  NN 33         ++ -e  $  ! &   (1 ")    %8$M! r#)FNrN)r)rrr)NNN)r- __module__ __qualname____doc__boolr rstrfloatrZr"rrr0r r;rr@rrCrFrUr\r`rbrgrkrsrrrMr#r!rrs7%%R"26!#8< 9 9 9d3:./ 9 9 $De$45 9 9 9 9#8BFsCx. * :/ : ue|  : : : :  53F    %De,-% c5j %%%%BtCJ'7BDBBBB&( 33S%Z()3S%Z()3#u*% 3 %) 3 # 3333p #u*%    .d3:&'#u*%  d3:  $#u*% d3:  0!%'d3:&'#u*%    #0De,-=A$sEzBR=S d3: ,:>*.*. d3:&'tDe$456!  ! 4I I KI 4U +, I #u*% I $ I $I I I  d3: I I I I VQ)QQS#X Q / Q ' ( QQQQQQr#rc6eZdZdZ ddedeeeefde deeeefd eed eef fd Z e d efd Z de dee deeeefdeeefdedededed eeeeffdZxZS)DistributeResourcesToTopJobaThis class creates a "TopJob" resource allocation function. The function will assign all of the free resources to the best performing trial (as defined by ``metric`` and ``mode``). The previous best trials will not have their resources deallocated, unless in the case outlined below. If for some reason a trial ends up with more resources than there are free ones, it will adjust downwards. It will also ensure that trial as at least as many resources as it started with (``base_trial_resource``). The function returns a new ``PlacementGroupFactory`` with updated resource requirements, or None. If the returned ``PlacementGroupFactory`` is equal by value to the one the trial has currently, the scheduler will skip the update process internally (same with None). Args: add_bundles: If True, create new bundles from free resources. Otherwise, spread them among base_trial_resource bundles. increase_by: A dict with key-value pairs representing an atomic unit of resources (name-amount) the trial will be increased by. If not set, the trial will increase by 1 CPU/GPU. increase_by_times: If set to >=1 and ``increase_by`` is set, the trial will increase by maximum of ``increase_by_times * increase_by`` resources. If set to <1, no upper limit is set. Ignored if ``increase_by`` is not set. reserve_resources: A dict of resource_name-amount pairs representing the resources that will not be allocated to resized trials. is that the attribute should increase monotonically. metric: The training result objective value attribute. Stopping procedures will use this attribute. If None, will use the metric of the scheduler. mode: One of {min, max}. Determines whether objective is minimizing or maximizing the metric attribute. If None, will use the metric of the scheduler. FNrrrrrmetricmodeclt||||||_||_dSr)superr"rr)r rrrrrrr,s r!r"z$DistributeResourcesToTopJob.__init__s7 k3DFWXXX  r#r&cP|jdvrtd|jdkrdSdS)N)rTr_z1The mode parameter can only be either min or max.r_g?g)rr+r s r! _metric_opz&DistributeResourcesToTopJob._metric_op s5 9N * *PQQ Q 9  3tr#rtrurhr9r:rvrwc ~jtd||z } ||z } t|fd} |jj} t fd| D| d} |j| jkr|| | dkr| S | || | S)NzThe metric parameter cannot be None. The parameter can be set in either `DistributeResourcesToTopJob`, the base scheduler or in `tune.TuneConfig()` (highest to lowest priority).cjj |jjtjzSr)r last_resultr7rrorp)r<r s r!zDDistributeResourcesToTopJob._get_new_added_bundles..*s'4?*Q]->->t{BF-S-SSr#)keyc3\K|]&}|jj"|V'dSr)rsr>r/)rPr<rhr s r! zEDistributeResourcesToTopJob._get_new_added_bundles..2sX  00-5|       r#r) rr+sortedrkr>r/nexttrial_idr\rg)r rtrurhrr9r:rvrwrcrd sorted_trialsrj best_trials` ` r!rz2DistributeResourcesToTopJob._get_new_added_bundless  ; D  )94 (94  SSSS   //  ) 1<       &    !     Nj1 1 1$$[)YGG1LL 77        r#)FNrNNN)r-rrrrr rrrrZr"propertyrrrr __classcell__r,s@r!rrs{((X"26!#8< $"   d3:./   $De$45   sm      EX5 5 K5 4U +, 5 #u*% 5 $ 5 $5 5 5  d3: 5 5 5 5 5 5 5 5 r#rF)rTc eZdZdZdefdeedeedee e e fdgee fddffd Z edZedee fd Zd ee d ee defd Zd ddefdZd ddefdZd ddede de fdZd ddede fdZd ddefdZd ddeefdZde fdZde fdZde fdZdedee e fdefdZdedefdZd ddede deeee ffdZ xZ!S)ra A utility scheduler to dynamically change resources of live trials. .. versionadded:: 1.5.0 .. note:: Experimental. API may change in future releases. The ResourceChangingScheduler works by wrapping around any other scheduler and adjusting the resource requirements of live trials in response to the decisions of the wrapped scheduler through a user-specified ``resources_allocation_function``. An example of such a function can be found in :doc:`/tune/examples/includes/xgboost_dynamic_resources_example`. If the functional API is used, the current trial resources can be obtained by calling `tune.get_trial_resources()` inside the training function. The function should be able to :ref:`load and save checkpoints ` (the latter preferably every iteration). If the Trainable (class) API is used, you can obtain the current trial resources through the ``Trainable.trial_resources`` property. Cannot be used if ``reuse_actors`` is True in ``tune.TuneConfig()``. A ValueError will be raised in that case. Args: base_scheduler: The scheduler to provide decisions about trials. If None, a default FIFOScheduler will be used. resources_allocation_function: The callable used to change live trial resource requiements during tuning. This callable will be called on each trial as it finishes one step of training. The callable must take four arguments: ``TrialRunner``, current ``Trial``, current result :class:`dict` and the ``ResourceChangingScheduler`` calling it. The callable must return a ``PlacementGroupFactory`` or None (signifying no need for an update). If ``resources_allocation_function`` is None, no resource requirements will be changed at any time. By default, :class:`DistributeResources` will be used, distributing available CPUs and GPUs over all running trials in a robust way, without any prioritization. Warning: If the ``resources_allocation_function`` sets trial resource requirements to values bigger than possible, the trial will not run. Ensure that your callable accounts for that possibility by setting upper limits. Consult :class:`DistributeResources` to see how that may be done. Example: .. code-block:: python base_scheduler = ASHAScheduler(max_t=16) def my_resources_allocation_function( tune_controller: "TuneController", trial: Trial, result: Dict[str, Any], scheduler: "ResourceChangingScheduler" ) -> Optional[Union[PlacementGroupFactory, Resource]]: # logic here # usage of PlacementGroupFactory is strongly preferred return PlacementGroupFactory(...) scheduler = ResourceChangingScheduler( base_scheduler, my_resources_allocation_function ) See :doc:`/tune/examples/includes/xgboost_dynamic_resources_example` for a more detailed example. Nbase_schedulerresources_allocation_functionrr&c t|tjd||_|p t |_d|_i|_t|_ d|_ d|_ dS)Nz`resources_allocation_function` is None. No resource requirements will be changed at any time. Pass a correctly defined function to enable functionality.) rr"warningswarn_resources_allocation_functionr_base_scheduler_base_trial_resources_trials_to_reallocater}_reallocated_trial_ids_metric_mode)r rrr,s r!r"z"ResourceChangingScheduler.__init__s  ( 0 MF    /L+-@FJ"  "14#  r#c|jjSr)rrrs r!rz ResourceChangingScheduler.metrics#++r#c|jSr)rrs r!rz.ResourceChangingScheduler.base_trial_resourcess ))r#rrc F||_||_|jj||fi|Sr)rrrset_search_properties)r rrspecs r!rz/ResourceChangingScheduler.set_search_propertiess2  9t#9&$OO$OOOr#r1rtc |j |j|_n>|j|jvr0|j}||jkrt d|jd|d|d|jj||fi|S)Nz^ResourceChangingScheduler doesn't support trials with varying base resources. First trial had z, trial z has .)rr>rr RuntimeErrorr on_trial_add)r r1rtkwargstrial_resourcess r!rz&ResourceChangingScheduler.on_trial_adds  % -).)FD & & ^4#> > >#;O$"<<<"*1**;@**'***1t#0%RR6RRRr#c *|jj||fi|Sr)ron_trial_errorr r1rtrs r!rz(ResourceChangingScheduler.on_trial_errors"2t#2?ETTVTTTr#r%c|j|||}|tjkr/||||}|r||j|<tjS|Sr)ron_trial_resultrCONTINUE$reallocate_trial_resources_if_neededrPAUSE)r r1rtr%base_scheduler_decision new_resourcess r!rz)ResourceChangingScheduler.on_trial_resultsv#'"6"F"F UF# #  #n&= = = EEM ,4A*51%++&&r#c ,|jj|||fi|Sr)ron_trial_complete)r r1rtr%rs r!rz+ResourceChangingScheduler.on_trial_completes26t#5 UF  .4   r#c *|jj||fi|Sr)ron_trial_removers r!rz)ResourceChangingScheduler.on_trial_removes$4t#3OUUUfUUUr#c `t|ddrtdd}i}|jD]U\}}|jt jkr#|||<t|d=|p| ||}V||_|j j |fi|}|S)N _reuse_actorsFzResourceChangingScheduler cannot be used with `reuse_actors=True`. FIX THIS by setting `reuse_actors=False` in `tune.TuneConfig()`.z# is still running, skipping for now) getattrr+ritemsstatusrRUNNINGloggerdebugset_trial_resourcesrchoose_trial_to_run)r r1rany_resources_changednew_trials_to_reallocatertrs r!rz-ResourceChangingScheduler.choose_trial_to_runs ?OU ; ; ?  !&#% $($>$D$D$F$F   E=|u},,2?(/ JJJKKK$9%T=U=U}>> ! !&>"8$8SSFSS r#c:d|jS)Nz(ResourceChangingScheduler) )r debug_stringrs r!rz&ResourceChangingScheduler.debug_stringsV1E1R1R1T1TVVVr#checkpoint_pathc|j}t|d5}tj||ddddS#1swxYwYdS)Nwb)__dict__openpickledump)r r save_object outputFiles r!savezResourceChangingScheduler.savesm /4 ( ( 1J K Z 0 0 0 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1s ;??ct|d5}tj|}dddn #1swxYwY|j|dS)Nrb)rrloadrupdate)r r inputFilers r!restorez!ResourceChangingScheduler.restore!s /4 ( ( 1I +i00K 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 [)))))s 266rc|retd|d|d|jd|_|||j|jdSdS)z'Returns True if new_resources were set.zSetting trial z resource to z with NTF)rinfo_bundlesr>update_resourcesrr~rr rtrs r!rz-ResourceChangingScheduler.set_trial_resources&s   KK111]11%.11   -1E )  " "= 1 1 1  ' + +EN ; ; ;4ur#ct|tr?|j|kr4t|d|jjd|jddSdS)zReturns True if trial's resources are value equal to new_resources. Only checks for PlacementGroupFactories at this moment. z PGF z and z are the same, skippingTF)r*rr>rrr?rs r!_are_resources_the_samez1ResourceChangingScheduler._are_resources_the_same6s| }&; < < ->> LL++0C++%8+++    45r#ch|jdSt|jdds%t|jd|j|j_t|jdds%t|jd|j|j_|||||}|r|||s|SdS)zCalls user defined resources_allocation_function. If the returned resources are not none and not the same as currently present, returns them. Otherwise, returns None.Nrrrr)rrrrrrrr)r r1rtr%rs r!rz>ResourceChangingScheduler.reallocate_trial_resources_if_neededMs  . 64t:HdKK 9@$i::D / 6t:FDII 7>$gtz88D / 4;; UFD   !!=!=e]!S!S ! tr#)"r-rrr_DistributeResourcesDefaultr rrrrrrrr"rrrrrrrrrrrrrrr rrdictrrrs@r!rrOs3GGV48 ( 0(0 $cN/  ./ 1  (  @,,X,*h/D&E***X*PsmP+3C=P PPPPS,<SUSSSS(U.>UuUUUU '/ '8= 'GK '  ' ' ' '  )         V/V8=VVVV / %4WcWWWW1C1111 *s**** +07L1L+M    ./8=GK %334 5r#r)&loggingrrcopyrtypingrrrrrr r r r numpyro#ray.air.execution.resources.requestr#ray.tune.execution.placement_groupsrray.tune.experimentr#ray.tune.schedulers.trial_schedulerrrray.util.annotationsr"ray.tune.execution.tune_controllerr getLoggerr-rrrr&_DistributeResourcesDistributedDefaultrrMr#r!rs( XXXXXXXXXXXXXXXXXXXXXX<<<<<<EEEEEE%%%%%%MMMMMMMM******BAAAAAA  8 $ $ Vyyyyyyyyx  Vu u u u u "5u u u p21eDDD)<)<)N)N)N& VWWWWWWWWWWr#