o ciL@sddlZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddlm Z ddl mZmZmZmZmZmZmZmZddlZddlZddlmZmZmZmZmZmZddl m!Z!dd l"m#Z#m$Z$dd l%m&Z&m'Z'ddl(Z(e)e*Z+d d Z,eZ-e&Gd dde Z.e&e/dddfdegefdeee/eee/fde0de0dee de1f ddZ2e&GdddZ3e&Gddde4Z5e&ddZ6dd Z7d!d"Z8e&d#efd$d%Z9d&ed'e:d(e:d)e:fd*d+Z;d,e:d-e:deefd.d/Ze&  dKde0fd?d@Z?dAdBZ@e' CdLdDee:dEeeeefdFedGe1fdHdIZAdS)MN) defaultdict)datetime)Number)Thread)AnyCallableDictListOptionalSequenceTypeUnion) deep_update flatten_dict merge_dictsunflatten_dictunflatten_list_dictunflattened_lookup)SafeFallbackEncoder)is_nan is_nan_or_inf) DeveloperAPI PublicAPIcCs(zddl}W|Styd}Y|SwNr)GPUtil ImportError)rrG/home/ubuntu/.local/lib/python3.10/site-packages/ray/tune/utils/util.py_import_gputil"s  rcsBeZdZdZdfdd ZddZdd Zd d Zd d ZZ S) UtilMonitora4Class for system usage utilization monitoring. It keeps track of CPU, RAM, GPU, VRAM usage (each gpu separately) by pinging for information every x seconds in a separate thread. Requires psutil and GPUtil to be installed. Can be enabled with Tuner(param_space={"log_sys_usage": True}). Tffffff?csd|_t}||_|dur|rtdtdur|rtd|dur)tdur)dStt|||_ t t |_ t |_d|_|rH|dSdS)NTz)Install gputil for GPU system monitoring.z-Install psutil to monitor system performance.)stoppedrrloggerwarningpsutilsuperr__init__delayrlistvalues threadingLocklockdaemonstart)selfr.r'r __class__rrr&8s"       zUtilMonitor.__init__c Cs|jrtdur#|jdttjdd|jdttj|jdureg}z|j }Wnt y?t dYnw|D]*}|jdt |jt|j|jdt |jt|jqBWddSWddS1sxwYdS)Ncpu_util_percent)intervalram_util_percentzGPUtil failed to retrieve GPUs.gpu_util_percentvram_util_percent)r,r$r)appendfloat cpu_percentvirtual_memorypercentrgetGPUs Exceptionr"debugstridload memoryUtil)r/gpu_listgpurrr_read_utilizationMs4     "zUtilMonitor._read_utilizationcCsp|jriS|jt|j}|jD] \}}|dd=qWdn1s(wYddd|DiS)NperfcSs(i|]\}}t|dkr|t|qS)r)lennpmean).0kvrrr ls(z(UtilMonitor.get_data..)r!r,copydeepcopyr)items)r/ ret_valueskeyvalrrrget_datads  zUtilMonitor.get_datacCs.d|_|js|t|j|jrdSdSNF)r!rEtimesleepr'r/rrrrunns  zUtilMonitor.runcCs d|_dS)NT)r!rXrrrstopts zUtilMonitor.stop)Tr ) __name__ __module__ __qualname____doc__r&rErTrYrZ __classcell__rrr0rr-s  rfnexception_type num_retries sleep_timetimeoutreturnc stfdd}t|D]>}tj|d}d|_||j|d|rBt d|dd|d t d dn sIdSt |qd S) Nc sFzWdSy"}zt|WYd}~dSd}~wwN)r"r#set)eerroredrcrbrr_try_fns  zretry_fn.._try_fn)targetT)rfzProcess timed out (try ra/z): r[F)r*Eventrangeclearrr-r.joinis_aliver"r>getattris_setrVrW)rbrcrdrerfrmiprocrrkrretry_fnxs$     ryc @s`eZdZdZeejddZdZ   dde de ed e e d e fd d Z d dZddZdS) warn_if_slowa)Prints a warning if a given operation is slower than 500ms. Example: >>> from ray.tune.utils.util import warn_if_slow >>> something = ... # doctest: +SKIP >>> with warn_if_slow("some_operation"): # doctest: +SKIP ... ray.get(something) # doctest: +SKIP TUNE_WARN_THRESHOLD_Sg?zTThe `{name}` operation took {duration:.3f} s, which may be a performance bottleneck.NFname thresholdmessagedisablecCs.||_|p|j|_|p |j|_d|_||_dSrU)r|DEFAULT_THRESHOLDr}DEFAULT_MESSAGEr~too_slowr)r/r|r}r~rrrrr&s    zwarn_if_slow.__init__cCst|_|Srh)rVr.rXrrr __enter__s zwarn_if_slow.__enter__cCsbt}|jr dS||j|jkr-|tdkr/d|_||j}t|jj |j |ddSdSdS)NgN@T)r|duration) rVrr.r} START_OF_TIMErr"r#r~formatr|)r/typevalue tracebacknowrrrr__exit__s zwarn_if_slow.__exit__)NNF)r[r\r]r^r8osenvirongetrrr?r boolr&rrrrrrrzs&   rzc@sxeZdZddZddZddZddZd d Zed d Z ed dZ eddZ ddZ ddZ ddZddZdS)TeecCs||_||_d|_dSrU)stream1stream2_handling_warning)r/rrrrrr&s z Tee.__init__cCsJ|jrdSd|d|d}|d|d|7}d|_t|d|_dS)NzValueError when calling 'z ' on stream (z). zargs: z kwargs: TF)rr"r#)r/opsargskwargsmsgrrr_warns  z Tee._warnc OL|j|jfD]}z |j|i|Wqty#|d|||YqwdS)Nseek)rrr ValueErrorrr/rrrrrrr zTee.seekc Or)Nwrite)rrrrrrrrrrrz Tee.writec Or)Nflush)rrrrrrrrrrrz Tee.flushcCt|jdr |jjS|jjS)Nencoding)hasattrrrrrXrrrr z Tee.encodingcCr)Nerror)rrrrrXrrrrrz Tee.errorcCr)Nnewlines)rrrrrXrrrrrz Tee.newlinescCtrhNotImplementedErrorrXrrrdetach z Tee.detachcOrrhrr/rrrrrreadrzTee.readcOrrhrrrrrreadlinerz Tee.readlinecOrrhrrrrrtellrzTee.tellN)r[r\r]r&rrrrpropertyrrrrrrrrrrrrs     rcCstdS)Nz%Y-%m-%d_%H-%M-%S)rtodaystrftimerrrrdate_strsrcCs|tdfS)zConverts obj to a form that can be pinned in object store memory. Currently only numpy arrays are pinned in memory, if you have a strong reference to the array value. ra)rHzerosobjrrr _to_pinnablesrcCs|dS)z"Retrieve from _to_pinnable format.rrrrrr_from_pinnable(sr trainablec s(ddlmm}fdd}td|dz|dt||dd td Wd StyA}z td |WYd }~nd }~wwtdt|dt|}t}|j rktdt |j d||j |dd|j rtdt |j d||j |dd|std|Std|d|S)aUtility for detecting why your trainable function isn't serializing. Args: trainable: The trainable object passed to tune.Tuner(trainable). Currently only supports Function API. Returns: bool | set of unserializable objects. Example: .. code-block:: python import threading # this is not serializable e = threading.Event() def test(): print(e) diagnose_serialization(test) # should help identify that 'e' should be moved into # the `test` scope. # correct implementation def test(): e = threading.Event() print(e) assert diagnose_serialization(test) is True r)_check_serializabilityregister_trainablec s|D]H\}}d}z ||d}Wn$ty7}zd}|jjdt|}||WYd}~nd}~ww|t|d|d||rL||qdS)NPASSEDFAILEDz: z[name='z'']... )rPr=r1r[r?add)objects failure_setprintervar_namevariablerstatusrjrrrcheck_variablesSs  z/diagnose_serialization..check_variableszTrying to serialize z...z__test:F)warnzSerialization succeeded!TzSerialization failed: NzIInspecting the scope of the trainable by running `inspect.getclosurevars(z)`...z Detected z. global variables. Checking serializability...cS td|SNz printrrrrt z(diagnose_serialization..z0 nonlocal variables. Checking serializability...cSrrrrrrrr{rzNothing was found to have failed the diagnostic test, though serialization did not succeed. Feel free to raise an issue on github.z Variable(s) z was found to be non-serializable. Consider either removing the instantiation/imports of these objects or moving them into the scope of the trainable. ) ray.tune.registryrrrr?r=inspectgetclosurevarsriglobalsrG nonlocals)rrrrjclosurerrrrdiagnose_serialization.sJ#   rstatecheckpoint_dir file_name tmp_file_namecCsjddlm}tj||}t|d}|||Wdn1s#wYt|tj||dS)aSAtomically saves the state object to the checkpoint directory. This is automatically used by Tuner().fit during a Tune job. Args: state: Object state to be serialized. checkpoint_dir: Directory location for the checkpoint. file_name: Final name of file. tmp_file_name: Temporary name of file. rNwb)ray.cloudpickle cloudpicklerpathrsopendumpreplace)rrrrrtmp_search_ckpt_pathfrrr _atomic_saves  rdirpath ckpt_patterncCsjddlm}ttj||}|sdSt|}t|d}||}Wd|S1s.wY|S)ahReturns the most recently modified checkpoint. Assumes files are saved with an ordered name, most likely by :obj:atomic_save. Args: dirpath: Directory in which to look for the checkpoint file. ckpt_pattern: File name pattern to match to find checkpoint files. Returns: (dict) Deserialized state dict. rNrb) rrglobrrrsmaxrrA)rrr full_pathsmost_recent_checkpointrcheckpoint_staterrr_load_newest_checkpoints    rbeta) stability{Gz?gpu_id target_utilretrydelay_sgpu_memory_limitc s&t}|dur tddurt}|std|ddttr0r-tndntts?tdt dfd d fd d | D}|vr^td |dt t|D]*}t fdd| D} | j |krtd|d| j dt|qddStd)aOChecks if a given GPU has freed memory. Requires ``gputil`` to be installed: ``pip install gputil``. Args: gpu_id: GPU id or uuid to check. Must be found within GPUtil.getGPUs(). If none, resorts to the first item returned from `ray.get_gpu_ids()`. target_util: The utilization threshold to reach to unblock. Set this to 0 to block until the GPU is completely free. retry: Number of times to check GPU limit. Sleeps `delay_s` seconds between checks. delay_s: Seconds to wait before check. Returns: bool: True if free. Raises: RuntimeError: If GPUtil is not found, if no GPUs are detected or if the check fails. Example: .. code-block:: python def tune_func(config): tune.utils.wait_for_gpu() train() tuner = tune.Tuner( tune.with_resources( tune_func, resources={"gpu": 1} ), tune_config=tune.TuneConfig(num_samples=10) ) tuner.fit() Nz3GPUtil must be installed if calling `wait_for_gpu`.zPNo GPU ids found from `ray.get_gpu_ids()`. Did you set Tune resources correctly?rr@uuidzgpu_id (z) must be type str/int.cs t|Srh)ru)g)gpu_attrrr gpu_id_fns zwait_for_gpu..gpu_id_fncsh|]}|qSrrrJr)rrr szwait_for_gpu..z% not found in set of available GPUs: zt. `wait_for_gpu` takes either GPU ordinal ID (e.g., '0') or UUID (e.g., 'GPU-04546190-b68d-65ac-101b-035f8faed77d').c3s |] }|kr|VqdSrhrr)rrrr szwait_for_gpu..zWaiting for GPU util to reach z. Util: z0.3fTzGPU memory was not freed.)r RuntimeErrorray get_gpu_ids isinstancer?isdigitintrrr<rqnextrBr"inforVrW) rrrrrr gpu_id_listgpu_idsrw gpu_objectr)rrrr wait_for_gpusD/      r  trainable_clsconfignum_gpusc CstsJdtj|d|}|j|d}|j|d}ddlm}tdD] }t|j}q&||s:Jdt|j|j t|j}||dksVJt|j}||d ksfJd S) aHelper method to check if your Trainable class will resume correctly. Args: trainable_cls: Trainable class for evaluation. config: Config to pass to Trainable when testing. num_gpus: GPU resources to allocate when testing. use_object_store: Whether to save and restore to Ray's object store. Recommended to set this to True if planning to use algorithms that pause training (i.e., PBT, HyperBand). zNeed Ray to be initialized.)r )r r)TRAINING_ITERATIONr`zQValidation will not pass because it requires `training_iteration` to be returned.rT) ris_initializedremoteray.air.constantsr rqrtrainrestoresave) r r r  remote_cls trainable_1 trainable_2r _resrrrvalidate_save_restore s      rc CsXt|}d}z|iW|Sty+}ztt|d}WYd}~|Sd}~ww)zCheck if func({}) works.TFN)r signaturebindr=r"r>r?)funcfunc_siguse_config_singlerjrrr_detect_config_singleJs  r Tparameter_namespoints_to_evaluateevaluated_rewardsvalidate_point_name_lengthscCs|rs    (   J  '+L  _ d )