o $i @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 l1m2Z2ddl3mZ4ddl5m6Z6m7Z7ddl8m9Z9ddl:m;Z;ddlZ>e rddlm?Z?ddl@mAZAeBeCZDe GdddZEGdddZFGdddZGe6GdddZHeIZJeeKeLd<daMeeHeLd<deeNeOfd ee>fd!d"ZPddd$d%ZQd#eeHfd&d'ZRd(d)ZSd*d+ZTd,ee#d#e#fd-d.ZUd/e#d#dfd0d1ZVded2efd3d4ZWe7d5d6eWddd7d8ed9ee!d:eeNd#dfd;d<ZXe7d5d6eWd#ee!fd=d>ZYe7d?d6eWd#eeNeffd@dAZZe7d?d6eWd#eNfdBdCZ[e7d?d6eWd#eNfdDdEZ\e7d?d6eWd#eNfdFdGZ]e7dHd6eWd#eNfdIdJZ^e7d?d6eWdfdLdMZ_e7d?d6eWd#eNfdNdOZ`e7d?d6eWdPdQd#eafdRdSZbe7d?d6eWddQd#eafdTdUZce7d?d6eWddQd#eafdVdWZde7d?d6eWddQd#eafdXdYZee7d?d6eWddQd#eafdZd[Zfe7d5d6eW ded\eeNd#ed]fd^d_Zge6eWd#e%fd`daZhd#eifdbdcZjdS)gN) 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)queue) 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__rfloatr1rr2r:r:X/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/train/_internal/session.pyr*8s 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=r:r:r;__init__Ns 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?rArEexcr:r:r;resolveQs z_FutureTrainingResult.resolveN)T) r3r4r5r6rF ObjectRefr@boolrrMr:r:r:r;r<Fsr<c@s<eZdZdZdeedeeeffddZ defddZ d S) rCz4A (checkpoint, metrics) result reported by the user. checkpointmetricscCs||_||_dSr>rPrQ)r?rPrQr:r:r;r@gs z_TrainingResult.__init__rBcCsd|jd|jdS)NzTrainingResult(checkpoint=z , metrics=)rRr?r:r:r;__repr__ksz_TrainingResult.__repr__N) r3r4r5r6rrrr7rr@rUr:r:r:r;rCdsrCc@seZdZdZ       dJdedeedeedeedeed eed eed eee e fd ee e fd ee de deede fddZde de fddZde de fddZddZddZ dKded edefddZd d!ZdKd"eedee fd#d$Zdeefd%d&Zd'd(Zd)e deefd*d+Zd,edefd-d.Zd,edefd/d0ZdLd1d2Zd3eddfd4d5Z dKd6ed ee ddfd7d8Z!e"de fd9d:Z#e"de fd;d<Z$e"de fd=d>Z%e"de fd?d@Z&e"dMdBdCZ'e"de fdDdEZ( dKdFee dedGfdHdIZ)dS)N _TrainSessionz.Holds information for training on each worker.NF training_func world_rank local_rank node_ranklocal_world_size world_size trial_info dataset_shardmetadatarPdetailed_autofilled_metricsstoragesynchronous_result_reportingcCs| |_||_| |_||_||_||_||_||_| sJt d|d| |j ||| | d| |_ t |_ d|_d|_||_d|_i|_dS)Nz StorageContext on SESSION (rank=z): )rWr]raloaded_checkpointrg)rbr^r_rXrYrZr[r\rJdebugresetr`timelast_report_time iteration time_totalget_current_iplocal_ip accelerator_state)r?rWrXrYrZr[r\r]r^r_rPr`rarbr:r:r;r@ts.   z_TrainSession.__init__keyrBcCs |j|Sr>)rmrG)r?rnr:r:r; get_states z_TrainSession.get_statevaluecCs||j|<dSr>)rm)r?rnrpr:r:r; set_statesz_TrainSession.set_statecCstj|_|jSr>)rFutilget_node_ip_addressrkrTr:r:r;rjs z_TrainSession.get_current_ipcCsd|_|jdS)zStarts the training thread.TN)training_startedtraining_threadstartrTr:r:r;rvsz_TrainSession.startcCstd|_t|_td|_d|_td|_ t |d|j d|_ ||_ ||_ ||_i|_d|_d|_d|_tj|jddtttjtdr\td|jt|jdSdS) NrT)targetdaemon error_queueF)exist_ok1z#Changing the working directory to: ) threading Semaphore continue_lockEvent stop_eventr!Queue result_queue_inter_actor_queuerzr rur]rarcrm ignore_reportrt _first_reportosmakedirstrial_working_directoryrOintenvironrGrrJrdchdir)r?rWr]rarcr:r:r;res,     z_TrainSession.resetcCs d|_dS)z-Ignore all future ``session.report()`` calls.TN)rrTr:r:r;pause_reportings z_TrainSession.pause_reportingrEcCs>|j|j|jjddd}|jr|jj|d}|S)zSFinishes the training thread. Raises any Exception from training. TforceNrD) rsetrreleaserapersist_artifactsrtrujoin)r?rEoutputr:r:r;finishs  z_TrainSession.finishcCs|jstd|jr|js|jd|_d}|dur/|jr/|jdd}|dur/|js |dur9|jdd}|durD|j ddn |j sNt d|jsV|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.FNTrAzVRunner error waiting to be raised in main thread. Logging all available results first.)rt RuntimeErrorrbrrrruis_alive_get_result_from_queues_report_thread_runner_errorrzemptyrJrd)r?resultr:r:r;get_next s*     z_TrainSession.get_nextcCs$|jdurtjdddid|_|jS)z$Get or create the inter-actor queue.Nrwnum_cpusr) actor_options)r ray_queuerrTr:r:r; _get_or_create_inter_actor_queueBs z._TrainSession._get_or_create_inter_actor_queuerAcCs~d}|jdur(z|jj|td}|r|j||Wn tjy'Ynwz |jj|td}W|St jy>Y|Sw)zUGet result from result queue. Pass result from training actor result queue if needed.NrArE) rrGrrrreportrEmptyrr!)r?rArinter_actor_itemr:r:r;rHs(   z%_TrainSession._get_result_from_queuesrc 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.rwcSsi|] \}}|tvr||qSr:)r.0kvr:r:r; qs z4_TrainSession._auto_fill_metrics..)rfrnowrrgrhrirrmktime timetuplerrrgetpidrplatformnoderrkr`itemscopyupdate)r?r current_timecurrent_datetimetime_this_iterauto_filled_metricsr:r:r;_auto_fill_metrics\s*   z _TrainSession._auto_fill_metricscCs4t}ttt|i}|}|||S)r) rrrrrfrrrr)r?rrrr:r:r;_auto_fill_checkpoint_metrics{s  z+_TrainSession._auto_fill_checkpoint_metricscCs.z |jj|td}t|tjyYdSw)Nr)rzrGr r r!r)r?rAer:r:r;rs 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) rPrcrputracquireris_setclearsysexit)r?rr:r:r;_report_training_results    z%_TrainSession._report_training_resultrQc 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.rrR)rmodulesray.air._internal.torch_utilsr ValueErrorrrra_update_checkpoint_indexpersist_current_checkpointcheckpoint_dir_namer sync_configsync_artifacts_on_checkpointrr_ get_metadatar set_metadatarCr) r?rQrPrpersisted_checkpointforce_artifact_sync user_metadatarrrr:r:r;rs8        z_TrainSession.reportcC|jjSr>)r]r1rTr:r:r;r1z_TrainSession.experiment_namecCrr>)r]r+rTr:r:r; trial_namerz_TrainSession.trial_namecCrr>)r]r,rTr:r:r;trial_idrz_TrainSession.trial_idcCrr>)r]r2rTr:r:r;r2rz_TrainSession.run_idr)cCrr>)r]r-rTr:r:r;trial_resourcesrz_TrainSession.trial_resourcescCrr>)r]r.rTr:r:r; 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 isinstancedictrrG)r?rshardr:r:r;get_dataset_shards  z_TrainSession.get_dataset_shard)NNNNFNFr>)FrBr))*r3r4r5r6rrrr*rr7rrrrOrr@rorqrjrvrerr9rrCrrrrrrrrrpropertyr1rrr2rrrr:r:r:r;rVps       >  15 3 rV_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||fqSrr:rr:r:r; s z3_tune_task_and_actor_launch_hook..Nrrwactor.taskcSs"i|] \}}|dkr|t|qSr)r9rr:r:r;rDs"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_grouprFrrget_current_placement_groupr,addget_trial_resourceshead_bundle_is_empty bundle_specsr% class_name module_name function_namer) fnr-rrncur_pgpgfavailable_bundles submittedr+main_resourcesr:r:r; _tune_task_and_actor_launch_hooksB      rrBcOsDtrtdddlm}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) rrrFrrrrr_actor_launch_hook_task_launch_hookrV)argskwargsrrr:r:r; init_sessionSs rcCstSr>rr:r:r:r; get_sessionesrcCsdadS)z#Shuts down the initialized session.Nrr:r:r:r;shutdown_sessionisrcCstd)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`rr:r:r:r;!_raise_accelerator_session_misuseosrdefault_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)rrrl)rsessionr:r:r;get_acceleratorws   rrlcCs0t}|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.)rrrlr)rlrr:r:r;set_accelerators    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)r rfn_namer:r;wrappersz4_warn_session_misuse..inner..wrapper)r3 functoolswraps)rrr )rr r;inners z#_warn_session_misuse..inner)r)r rr:rr;_warn_session_misusesrstable) stability)rPrrQrPrcCs^|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)rP) rJwarning%ray.tune.trainable.trainable_fn_utilsrray.tunerr tunerr)rQrPrrrFr:r:r;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_checkpointrrc)rrFr:r:r;r$s , rbetacCtjS)z5User metadata dict passed to the Trainer constructor.)rr_r:r:r:r;rbrcCr)z,Experiment name for the corresponding trial.)rr1r:r:r:r;get_experiment_nameirr cCr)z'Trial name for the corresponding trial.)rrr:r:r:r;get_trial_nameprr!cCr)z%Trial id for the corresponding trial.)rrr:r:r:r; get_trial_idwrr"alphacCr)z0Unique Train Run id for the corresponding trial.)rr2r:r:r:r; get_run_id~rr$r)cCr)z,Trial resources for the corresponding trial.)rrr:r:r:r;rrrcCr)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 )rrr:r:r:r; get_trial_dirsr%rwrcCt}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\rr:r:r;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: ... rXz`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'rrXr(r:r:r;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: ... rYz`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'rrYr(r:r:r;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: ... r[z`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'rr[r(r:r:r;get_local_world_sizer+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: ... rZz`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'rrZr(r:r:r; get_node_rankEr+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)rrr:r:r;rls /  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. )rrar:r:r:r; get_storages r/cCstto tjduS)z6Check if the current process is a Ray Train V1 worker.N)rOrrXr:r:r:r;_in_ray_train_workersr0)rBNr>r)krloggingrrr!rr}rfr dataclassesrrtypingrrrrrr r rFray.air._internal.utilr r ray.air.constantsr rrrrray.datar ray.trainrray.train._internal.acceleratorrray.train._internal.storagerray.train.constantsrrrrrrrrray.train.errorrray.train.utilsr ray.utilrray.util.annotationsr"r#ray.util.debugr$ray.util.placement_groupr%ray.util.scheduling_strategiesr&r'r(#ray.tune.execution.placement_groupsr) getLoggerr3rJr*r<rCrVrrrr8rr7r9rrrrrrr rrrrr r!r"r$rr%rr)r*r,r-r.rr/rOr0r:r:r:r;s   $    (           @p<&%(%%7