o ci` @sUddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl m Z mZmZmZmZmZmZddlZddlmZmZddlmZmZmZmZmZddlmZddl m!Z!dd l"m#Z#dd l$m%Z%dd l&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.dd l/m0Z0dd l1m2Z2ddl3m4Z4m5Z5ddl6m7Z7ddl8m9Z9ddl:m;Z;mm?Z?e@eAZBe GdddZCGdddZDGdddZEe4GdddZFeGZHeeIeJd<daKeeFeJd<deeLeMfdee<fd d!ZNdad#d$ZOd"eeFfd%d&ZPd'd(ZQd)d*ZRd+ee#d"e#fd,d-ZSd.e#d"dfd/d0ZTdbd1efd2d3ZUe5d4d5eUddd6d7ed8ee!d9eeLd"dfd:d;ZVe5d4d5eUd"ee!fdd5eUd"eeLeffd?d@ZXe5d>d5eUd"eLfdAdBZYe5d>d5eUd"eLfdCdDZZe5d>d5eUd"eLfdEdFZ[e5dGd5eUd"eLfdHdIZ\e5d>d5eUdcdKdLZ]e5d>d5eUd"eLfdMdNZ^e5d>d5eUdOdPd"e_fdQdRZ`e5d>d5eUddPd"e_fdSdTZae5d>d5eUddPd"e_fdUdVZbe5d>d5eUddPd"e_fdWdXZce5d>d5eUddPd"e_fdYdZZde5d4d5eU dbd[eeLd"ed\fd]d^Zee4eUd"e%fd_d`ZfdS)dN) dataclass)datetime) TYPE_CHECKINGAnyCallableDictOptionalSetType) RunnerThreadStartTraceback)_ERROR_FETCH_TIMEOUT_RESULT_FETCH_TIMEOUTSESSION_MISUSE_LOG_ONCE_KEYTIME_THIS_ITER_S TIMESTAMP)Dataset) Checkpoint) Accelerator)StorageContext)CHECKPOINT_DIR_NAMEDETAILED_AUTOFILLED_KEYSRAY_CHDIR_TO_TRIAL_DIR TIME_TOTAL_SWORKER_HOSTNAMEWORKER_NODE_IP WORKER_PID_v2_migration_warnings_enabledSessionMisuseError)_log_deprecation_warning) DeveloperAPI PublicAPI)log_once)_valid_resource_shape) PlacementGroupSchedulingStrategySchedulingStrategyT) DataIterator)PlacementGroupFactoryc@sjeZdZUdZeed<eed<eeefed<eed<eed<eed<dZe eed <dZ e eed <dS) TrialInfoz3The trial information to propagate to TrainSession.nameid resourceslogdir driver_ipdriver_node_idNexperiment_namerun_id) __name__ __module__ __qualname____doc__str__annotations__rfloatr0rr1r9r9O/home/ubuntu/.local/lib/python3.10/site-packages/ray/train/_internal/session.pyr)7s r)c@s8eZdZdZdejfddZd dededfd d Z d S) _FutureTrainingResultzA future that will be resolved to a `_TrainingResult`. This is needed for specific schedulers such as PBT that schedule saves. This wrapper should be removed after refactoring PBT to not schedule saves anymore. futurecCs ||_dSN)r<)selfr<r9r9r:__init__Ms z_FutureTrainingResult.__init__Tblockreturn_TrainingResultc Csj|rd}nd}z tj|j|dWStyYdSty4}ztd|WYd}~dSd}~ww)zResolve into ``_TrainingResult``. This will return None for function trainables if no checkpoint has been saved before. Ng& .>timeoutzError resolving result: )raygetr< TimeoutError Exceptionloggererror)r>r@rDexcr9r9r:resolvePs z_FutureTrainingResult.resolveN)T) r2r3r4r5rE ObjectRefr?boolrrLr9r9r9r:r;Esr;c@s<eZdZdZdeedeeeffddZ defddZ d S) rBz4A (checkpoint, metrics) result reported by the user. checkpointmetricscCs||_||_dSr=rOrP)r>rOrPr9r9r:r?fs z_TrainingResult.__init__rAcCsd|jd|jdS)NzTrainingResult(checkpoint=z , metrics=)rQr>r9r9r:__repr__jsz_TrainingResult.__repr__N) r2r3r4r5rrrr6rr?rTr9r9r9r:rBcsrBc@seZdZdZ       dEdedeedeedeedeed eed eed eee e fd ee e fd ee de deede fddZde de fddZde de fddZddZddZ dFded edefddZd d!ZdFd"eedee fd#d$Zdeefd%d&Zd'edefd(d)Zd'edefd*d+ZdGd,d-Zd.eddfd/d0ZdFd1ed ee ddfd2d3Ze de fd4d5Z!e de fd6d7Z"e de fd8d9Z#e de fd:d;Z$e dHd=d>Z%e de fd?d@Z& dFdAee dedBfdCdDZ'dS)I _TrainSessionz.Holds information for training on each worker.NF training_func world_rank local_rank node_ranklocal_world_size world_size trial_info dataset_shardmetadatarOdetailed_autofilled_metricsstoragesynchronous_result_reportingcCs| |_||_| |_||_||_||_||_||_| sJt d|d| |j ||| | d| |_ t |_ d|_d|_||_d|_i|_dS)Nz StorageContext on SESSION (rank=z): )rVr\r`loaded_checkpointrg)rar]r^rWrXrYrZr[rIdebugresetr_timelast_report_time iteration time_totalget_current_iplocal_ip accelerator_state)r>rVrWrXrYrZr[r\r]r^rOr_r`rar9r9r:r?ss.   z_TrainSession.__init__keyrAcCs |j|Sr=)rlrF)r>rmr9r9r: get_states z_TrainSession.get_statevaluecCs||j|<dSr=)rl)r>rmror9r9r: set_statesz_TrainSession.set_statecCstj|_|jSr=)rEutilget_node_ip_addressrjrSr9r9r:ris z_TrainSession.get_current_ipcCsd|_|jdS)zStarts the training thread.TN)training_startedtraining_threadstartrSr9r9r:rusz_TrainSession.startcCstd|_t|_td|_td|_t |d|jd|_ ||_ ||_ ||_ i|_d|_d|_d|_tj|jddtttjtdrYtd|jt|jdSdS) NrT)targetdaemon error_queueF)exist_ok1z#Changing the working directory to: ) threading Semaphore continue_lockEvent stop_eventqueueQueue result_queueryr rtr\r`rbrl ignore_reportrs _first_reportosmakedirstrial_working_directoryrNintenvironrFrrIrcchdir)r>rVr\r`rbr9r9r:rds*     z_TrainSession.resetcCs d|_dS)z-Ignore all future ``session.report()`` calls.TN)rrSr9r9r:pause_reportings z_TrainSession.pause_reportingrDcCs>|j|j|jjddd}|jr|jj|d}|S)zSFinishes the training thread. Raises any Exception from training. TforceNrC) rsetr~releaser`persist_artifactsrsrtjoin)r>rDoutputr9r9r:finishs  z_TrainSession.finishcCs|jstd|jr|js|jd|_d}|dur>|jr>z |jj dt d}Wn t j y4Ynw|dur>|js |durWz |jj dt d}Wn t j yVYnw|durb|j ddn |jsltd|jst|j|S)zGets the next ``_TrainingResult`` from the result queue. If the result queue is empty, then this function returns ``None``. z*Please call start before calling get_next.FNTr@rDr@zVRunner error waiting to be raised in main thread. Logging all available results first.)rs RuntimeErrorrarr~rrtis_aliverrFrrEmpty_report_thread_runner_errorryemptyrIrc)r>resultr9r9r:get_next sB      z_TrainSession.get_nextrc Cst}t}t|vr|t}n||j}|jd7_|j|7_||_ttt | t |jt t ttt|ji}|jsLdd|D}|}|||S)-Add autofilled metrics and update attributes.rvcSsi|] \}}|tvr||qSr9)r.0kvr9r9r: ]s z4_TrainSession._auto_fill_metrics..)rernowrrfrgrhrrmktime timetuplerrrgetpidrplatformnoderrjr_itemscopyupdate)r>r current_timecurrent_datetimetime_this_iterauto_filled_metricsr9r9r:_auto_fill_metricsHs*   z _TrainSession._auto_fill_metricscCs4t}ttt|i}|}|||S)r) rrrrrerrrr)r>rrrr9r9r:_auto_fill_checkpoint_metricsgs  z+_TrainSession._auto_fill_checkpoint_metricscCs.z |jj|td}t|tjyYdSw)Nr)ryrFr r rr)r>r@er9r9r:rrs z)_TrainSession._report_thread_runner_errortraining_resultcCsN|jr|j|_|jj|dd|j|jr%|jt ddSdS)aPlace a training result on the result queue for the main thread to process, then block until the main thread signals that training should continue. NOTE: This is used internally to report results from Train to Tune without persisting checkpoints to storage 2 times. `report` is the public API that directly persists to storage, which should only be called by user code. TrrN) rOrbrputr~acquireris_setclearsysexit)r>rr9r9r:_report_training_resultys    z%_TrainSession._report_training_resultrPc Csdtjvrddlm}||rtd|jrdS||}d}|r4|j||j |}|jj |t <nd|t <|o>|jj j }|jj|d|rf|jrf|}|jD] \}}||vr`|||<qT||t||d} || dS)Ntorchr)contains_tensorzPassing objects containg Torch tensors as metrics is not supported as it will throw an exception on deserialization. You can either convert the tensors to Python objects or report a `train.Checkpoint` with `ray.train.report` to store your Torch objects.rrQ)rmodulesray.air._internal.torch_utilsr ValueErrorrrr`_update_checkpoint_indexpersist_current_checkpointcheckpoint_dir_namer sync_configsync_artifacts_on_checkpointrr^ get_metadatar set_metadatarBr) r>rPrOrpersisted_checkpointforce_artifact_sync user_metadatarrrr9r9r:reports8        z_TrainSession.reportcC|jjSr=)r\r0rSr9r9r:r0z_TrainSession.experiment_namecCrr=)r\r*rSr9r9r: trial_namerz_TrainSession.trial_namecCrr=)r\r+rSr9r9r:trial_idrz_TrainSession.trial_idcCrr=)r\r1rSr9r9r:r1rz_TrainSession.run_idr(cCrr=)r\r,rSr9r9r:trial_resourcesrz_TrainSession.trial_resourcescCrr=)r\r-rSr9r9r: trial_dirrz_TrainSession.trial_dir dataset_namer'cCs@|j}|durtd|St|tr|std||S|S)NziNo dataset passed in. Returning None. Make sure to pass in a Dataset to Trainer.run to use this function.zMultiple datasets were passed into ``Trainer``, but no ``dataset_name`` is passed into ``get_dataset_shard``. Please specify which dataset shard to retrieve.)r]warningswarn isinstancedictrrF)r>rshardr9r9r:get_dataset_shards  z_TrainSession.get_dataset_shard)NNNNFNFr=)FrAr()(r2r3r4r5rrrr)rr6rrrrNrr?rnrprirurdrr8rrBrrrrrrrpropertyr0rrr1rrrr9r9r9r:rUos       >  .? 3 rU_checked_resources_sessionr,strategyc Cstdd|D}|r|tvrdSt|tr|jdurdStj}|r-|jj |j kr/dSt |t }|j rB|j dd}n|j dd}t||rPdS|jrcd}|jd|jd|j}n d}|jd|j}|j d} d d |D}td |d |d | d|d|d )zLaunch hook to catch nested tasks that can't fit in the placement group. This gives users a nice warning in case they launch a nested task in a Tune trial without reserving resources in the trial placement group to fit it. cSs h|] \}}|dkr||fqSrr9rr9r9r: s z3_tune_task_and_actor_launch_hook..Nrrvactor.taskcSs"i|] \}}|dkr|t|qSr)r8rr9r9r:r0s"z4_tune_task_and_actor_launch_hook..z3No trial resources are available for launching the z `zg`. To resolve this, specify the Tune option: > resources_per_trial=tune.PlacementGroupFactory( > [z] + [zB] * N > ) Where `N` is the number of slots to reserve for trial zs. If you are using a Ray training library, there might be a utility function to set this automatically for you. For more information, refer to https://docs.ray.io/en/latest/tune/tutorials/tune-resources.html) frozensetrrrr%placement_grouprErqget_current_placement_groupr+addget_trial_resourceshead_bundle_is_empty bundle_specsr$ class_name module_name function_namer) fnr,rrmcur_pgpgfavailable_bundles submittedr*main_resourcesr9r9r: _tune_task_and_actor_launch_hooksB      rrAcOsDtrtdddlm}m}dtjvrt|_t|_ t |i|adS)NzIA Train session is already in use. Do not call `init_session()` manually.r)rremote_functionTUNE_DISABLE_RESOURCE_CHECKS) rrrErrrrr_actor_launch_hook_task_launch_hookrU)argskwargsrrr9r9r: init_session?s rcCstSr=rr9r9r9r: get_sessionQsrcCsdadS)z#Shuts down the initialized session.Nrr9r9r9r:shutdown_sessionUsrcCstd)zKRaises a SessionMisuseError because a utility function was used improperly.zjprepare/accelerate utility functions should be called inside a training function executed by `Trainer.run`rr9r9r9r:!_raise_accelerator_session_misuse[srdefault_accelerator_clscCs,t}|dur t|jdur||_|jS)zThe accelerator for this training session. If an accelerator has not been set, then this method will construct an accelerator using the provided accelerator class. Raises: SessionMisuseError: if the session is uninitialized. N)rrrk)rsessionr9r9r:get_acceleratorcs   rrkcCs0t}|dur t|jdurtd||_dS)aSets the accelerator for this training session. Args: accelerator: The accelerator to use for training. Raises: SessionMisuseError: if the session is unitialized. RuntimeError: if the accelerator has already been set. Nz#Cannot change accelerator once set.)rrrkr)rkrr9r9r:set_acceleratorts    r default_valuecsdtffdd }|S)zKWarns if fn is being used outside of session and returns ``default_value``.rcs$jtfdd}|S)NcsFt}|sttdrtdddS|i|S)N-`zb` is meant to only be called inside a function that is executed by a Tuner or Trainer. Returning `z`.)rr#rrr)rrr)rrfn_namer9r:wrappersz4_warn_session_misuse..inner..wrapper)r2 functoolswraps)rrr)rrr:inners z#_warn_session_misuse..inner)r)rr r9r r:_warn_session_misusesr stable) stability)rOrrPrOrcCs^|dur tdddlm}|r%ddl}trtd|jj||dSt j||ddS)a Report metrics and optionally save a checkpoint. If a checkpoint is provided, it will be :ref:`persisted to storage `. If this is called in multiple distributed training workers: - Only the metrics reported by the rank 0 worker will be tracked by Ray Train. See :ref:`the metrics logging guide `. - A checkpoint will be registered as long as one or more workers reports checkpoint that is not None. See the :ref:`checkpointing guide `. - Checkpoints from multiple workers will be merged into one directory in persistent storage. See :ref:`the distributed checkpointing guide `. .. note:: Each invocation of this method will automatically increment the underlying ``training_iteration`` number. The physical meaning of this "iteration" is defined by user depending on how often they call ``report``. It does not necessarily map to one epoch. .. warning:: All workers must call `ray.train.report` the same number of times so that Ray Train can properly synchronize the training state across workers. Otherwise, your training will hang. .. warning:: This method does NOT act as a barrier for distributed training workers. Workers will upload their checkpoint, then continue training immediately. If you need to synchronize workers, you can use a framework-native barrier such as `torch.distributed.barrier()`. Example: .. testcode:: import tempfile from ray import train from ray.train import Checkpoint from ray.train.torch import TorchTrainer def train_func(config): start_epoch = 0 checkpoint = train.get_checkpoint() if checkpoint: with checkpoint.as_directory() as checkpoint_dir: # Load back training state ... for epoch in range(start_epoch, config.get("num_epochs", 10)): # Do training... metrics = {"loss": ...} with tempfile.TemporaryDirectory() as temp_checkpoint_dir: # Save the checkpoint... # torch.save(...) checkpoint = Checkpoint.from_directory(temp_checkpoint_dir) # Example: Only the rank 0 worker uploads the checkpoint. if ray.train.get_context().get_world_rank() == 0: train.report(metrics, checkpoint=checkpoint) else: train.report(metrics, checkpoint=None) trainer = TorchTrainer( train_func, scaling_config=train.ScalingConfig(num_workers=2) ) Args: metrics: The metrics you want to report. checkpoint: The optional checkpoint you want to report. Nz`checkpoint_dir_name` is only supported in the new Ray Train implementation, which can be enabled with `RAY_TRAIN_V2_ENABLED=1`. This argument will be ignored.r_in_tune_sessionz`ray.train.report` should be switched to `ray.tune.report` when running in a function passed to Ray Tune. This will be an error in the future. See this issue for more context: https://github.com/ray-project/ray/issues/49454)rO) rIwarning%ray.tune.trainable.trainable_fn_utilsrray.tunerr tunerr)rPrOrrrEr9r9r:rsX rcCs:ddlm}|rddl}trtd|jStjS)aAccess the latest reported checkpoint to resume from if one exists. Example: .. testcode:: import tempfile from ray import train from ray.train import Checkpoint from ray.train.torch import TorchTrainer def train_func(config): start_epoch = 0 checkpoint = train.get_checkpoint() if checkpoint: with checkpoint.as_directory() as checkpoint_dir: # Load back training state ... for epoch in range(start_epoch, config.get("num_epochs", 10)): # Do training... metrics = {"loss": ...} with tempfile.TemporaryDirectory() as temp_checkpoint_dir: # Save the checkpoint... checkpoint = Checkpoint.from_directory(temp_checkpoint_dir) train.report(metrics, checkpoint=checkpoint) trainer = TorchTrainer( train_func, scaling_config=train.ScalingConfig(num_workers=2) ) Returns: Checkpoint object if the session is currently being resumed. Otherwise, return None. rrNz`ray.train.get_checkpoint` should be switched to `ray.tune.get_checkpoint` when running in a function passed to Ray Tune. This will be an error in the future. See this issue for more context: https://github.com/ray-project/ray/issues/49454) rrrrr rget_checkpointrrb)rrEr9r9r:rs , rbetacCtjS)z5User metadata dict passed to the Trainer constructor.)rr^r9r9r9r:rNrcCr)z,Experiment name for the corresponding trial.)rr0r9r9r9r:get_experiment_nameUrrcCr)z'Trial name for the corresponding trial.)rrr9r9r9r:get_trial_name\rrcCr)z%Trial id for the corresponding trial.)rrr9r9r9r: get_trial_idcrralphacCr)z0Unique Train Run id for the corresponding trial.)rr1r9r9r9r: get_run_idjrrr(cCr)z,Trial resources for the corresponding trial.)rrr9r9r9r:rqrrcCr)aLog directory corresponding to the trial directory for a Tune session. If calling from a Train session, this will give the trial directory of its parent Tune session. .. testcode:: import ray.tune def train_func(config): print(ray.tune.get_context().get_trial_dir()) tuner = ray.tune.Tuner(train_func) tuner.fit() .. testoutput:: :options: +MOCK /Users/root/ray_results/train_func_2023-07-19_15-01-37/train_func_d620c_00000_0_2023-07-19_15-01-40 )rrr9r9r9r: get_trial_dirxsrrvr cCt}t|ds td|jS)aGet the current world size (i.e. total number of workers) for this run. .. testcode:: import ray from ray import train from ray.train import ScalingConfig from ray.train.tensorflow import TensorflowTrainer NUM_WORKERS = 2 def train_loop_per_worker(config): assert train.get_context().get_world_size() == NUM_WORKERS train_dataset = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") trainer = TensorflowTrainer( train_loop_per_worker, scaling_config=ScalingConfig(num_workers=NUM_WORKERS), datasets={"train": train_dataset} ) trainer.fit() .. testoutput:: :hide: ... r[z`get_world_size` can only be called for TrainSession! Make sure you only use that in `train_loop_per_worker` functionthat is passed into `DataParallelTrainer`.)rhasattrrr[rr9r9r:get_world_sizes  r"cCr)aGet the world rank of this worker. .. testcode:: import ray from ray import train from ray.train import ScalingConfig from ray.train.tensorflow import TensorflowTrainer def train_loop_per_worker(config): if train.get_context().get_world_rank() == 0: print("Worker 0") train_dataset = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") trainer = TensorflowTrainer( train_loop_per_worker, scaling_config=ScalingConfig(num_workers=2), datasets={"train": train_dataset} ) trainer.fit() .. testoutput:: :hide: ... rWz`get_world_rank` can only be called for TrainSession! Make sure you only use that in `train_loop_per_worker` functionthat is passed into `DataParallelTrainer`.)rr rrWr!r9r9r:get_world_rank  r#cCr)aGet the local rank of this worker (rank of the worker on its node). .. testcode:: import torch import ray from ray import train from ray.train import ScalingConfig from ray.train.torch import TorchTrainer def train_loop_per_worker(config): if torch.cuda.is_available(): torch.cuda.set_device(train.get_context().get_local_rank()) ... train_dataset = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") trainer = TorchTrainer( train_loop_per_worker, scaling_config=ScalingConfig(num_workers=2, use_gpu=True), datasets={"train": train_dataset} ) trainer.fit() .. testoutput:: :hide: ... rXz`get_local_rank` can only be called for TrainSession! Make sure you only use that in `train_loop_per_worker` functionthat is passed into `DataParallelTrainer`.)rr rrXr!r9r9r:get_local_ranks  r%cCr)aGet the local world size of this node (i.e. number of workers on this node). Example: .. testcode:: import ray from ray import train from ray.train import ScalingConfig from ray.train.torch import TorchTrainer def train_loop_per_worker(): print(train.get_context().get_local_world_size()) train_dataset = ray.data.from_items( [{"x": x, "y": x + 1} for x in range(32)]) trainer = TorchTrainer(train_loop_per_worker, scaling_config=ScalingConfig(num_workers=1), datasets={"train": train_dataset}) trainer.fit() .. testoutput:: :hide: ... rZz`get_local_world_size` can only be called for TrainSession! Make sure you only use that in `train_loop_per_worker` functionthat is passed into `DataParallelTrainer`.)rr rrZr!r9r9r:get_local_world_size r$r&cCr)aGet the rank of this node. Example: .. testcode:: import ray from ray import train from ray.train import ScalingConfig from ray.train.torch import TorchTrainer def train_loop_per_worker(): print(train.get_context().get_node_rank()) train_dataset = ray.data.from_items( [{"x": x, "y": x + 1} for x in range(32)]) trainer = TorchTrainer(train_loop_per_worker, scaling_config=ScalingConfig(num_workers=1), datasets={"train": train_dataset}) trainer.fit() .. testoutput:: :hide: ... rYz`get_node_rank` can only be called for TrainSession! Make sure you only use that in `train_loop_per_worker` functionthat is passed into `DataParallelTrainer`.)rr rrYr!r9r9r: get_node_rank1r$r'rr'cCs"t}t|ds td||S)a?Returns the :class:`ray.data.DataIterator` shard for this worker. Call :meth:`~ray.data.DataIterator.iter_torch_batches` or :meth:`~ray.data.DataIterator.to_tf` on this shard to convert it to the appropriate framework-specific data type. .. testcode:: import ray from ray import train from ray.train import ScalingConfig from ray.train.torch import TorchTrainer def train_loop_per_worker(config): ... for epoch in range(2): # Trainer will automatically handle sharding. data_shard = train.get_dataset_shard("train") for batch in data_shard.iter_torch_batches(): ... train_dataset = ray.data.read_csv("s3://anonymous@ray-example-data/iris.csv") trainer = TorchTrainer( train_loop_per_worker, scaling_config=ScalingConfig(num_workers=2), datasets={"train": train_dataset} ) trainer.fit() .. testoutput:: :hide: ... Args: dataset_name: If a Dictionary of Datasets was passed to ``Trainer``, then specifies which dataset shard to return. Returns: The ``DataIterator`` shard to use for this worker. If no dataset is passed into Trainer, then return None. rz`get_dataset_shard` can only be called for TrainSession! Make sure you only use that in `train_loop_per_worker` functionthat is passed into `DataParallelTrainer`.)rr rr)rrr9r9r:rXs /  rcCr)a2Returns the :class:`~ray.train._internal.storage.StorageContext` storage context which gives advanced access to the filesystem and paths configured through `RunConfig`. NOTE: This is a developer API, and the `StorageContext` interface may change without notice between minor versions. )rr`r9r9r9r: get_storages r()rANr=r)grloggingrrrrr|rer dataclassesrrtypingrrrrrr r rEray.air._internal.utilr r ray.air.constantsr rrrrray.datar ray.trainrray.train._internal.acceleratorrray.train._internal.storagerray.train.constantsrrrrrrrrray.train.errorrray.train.utilsr ray.util.annotationsr!r"ray.util.debugr#ray.util.placement_groupr$ray.util.scheduling_strategiesr%r&r'#ray.tune.execution.placement_groupsr( getLoggerr2rIr)r;rBrUrrrr7rr6r8rrrrrrrr rrrrrrrrrrr"r#r%r&r'rr(r9r9r9r:s   $    (           @p<&%(%%7