o 1 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 resourceslogdirZ driver_ipZdriver_node_idNexperiment_namerun_id) __name__ __module__ __qualname____doc__str__annotations__rfloatr/rr0r8r8g/home/app/PaddleOCR-VL-test/.venv_paddleocr/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;r8r8r9__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=r?rCexcr8r8r9resolveQs z_FutureTrainingResult.resolveN)T) r1r2r3r4rDZ ObjectRefr>boolrrKr8r8r8r9r:Fsr:c@s<eZdZdZdeedeeeffddZ defddZ d S) rAz4A (checkpoint, metrics) result reported by the user. checkpointmetricscCs||_||_dSr<rMrN)r=rMrNr8r8r9r>gs z_TrainingResult.__init__r@cCsd|jd|jdS)NzTrainingResult(checkpoint=z , metrics=)rOr=r8r8r9__repr__ksz_TrainingResult.__repr__N) r1r2r3r4rrrr5rr>rRr8r8r8r9rAdsrAc@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_shardmetadatarMdetailed_autofilled_metricsstoragesynchronous_result_reportingcCs| |_||_| |_||_||_||_||_||_| sJt d|d| |j ||| | d| |_ t |_ d|_d|_||_d|_i|_dS)Nz StorageContext on SESSION (rank=z): )rTrZr^loaded_checkpointrg)r_r[r\rUrVrWrXrYrHdebugresetr]timelast_report_time iteration time_totalget_current_iplocal_ip accelerator_state)r=rTrUrVrWrXrYrZr[r\rMr]r^r_r8r8r9r>ts.   z_TrainSession.__init__keyr@cCs |j|Sr<)rjrE)r=rkr8r8r9 get_states z_TrainSession.get_statevaluecCs||j|<dSr<)rj)r=rkrmr8r8r9 set_statesz_TrainSession.set_statecCstj|_|jSr<)rDutilZget_node_ip_addressrhrQr8r8r9rgs z_TrainSession.get_current_ipcCsd|_|jdS)zStarts the training thread.TN)training_startedtraining_threadstartrQr8r8r9rrsz_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_queuervr rqrZr^r`rj ignore_reportrp _first_reportosmakedirsZtrial_working_directoryrLintenvironrErrHrachdir)r=rTrZr^r`r8r8r9rbs,     z_TrainSession.resetcCs d|_dS)z-Ignore all future ``session.report()`` calls.TN)rrQr8r8r9pause_reportings z_TrainSession.pause_reportingrCcCs>|j|j|jjddd}|jr|jj|d}|S)zSFinishes the training thread. Raises any Exception from training. TforceNrB) r}setr{releaser^persist_artifactsrprqjoin)r=rCoutputr8r8r9finishs  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.FNTr?zVRunner error waiting to be raised in main thread. Logging all available results first.)rp RuntimeErrorr_rr{rrqis_alive_get_result_from_queues_report_thread_runner_errorrvemptyrHra)r=resultr8r8r9get_next s*     z_TrainSession.get_nextcCs$|jdurtjdddid|_|jS)z$Get or create the inter-actor queue.NrsZnum_cpusr)Z actor_options)r ray_queuer~rQr8r8r9 _get_or_create_inter_actor_queueBs z._TrainSession._get_or_create_inter_actor_queuer?cCs~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.Nr?rC) rrErr{rreportrEmptyrr!)r=r?rZinter_actor_itemr8r8r9rHs(   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.rscSsi|] \}}|tvr||qSr8)r.0kvr8r8r9 qs z4_TrainSession._auto_fill_metrics..)rcrnowrrdrerfrrmktime timetuplerrrgetpidrplatformnoderrhr]itemscopyupdate)r=r current_timecurrent_datetimeZtime_this_iterauto_filled_metricsr8r8r9_auto_fill_metrics\s*   z _TrainSession._auto_fill_metricscCs4t}ttt|i}|}|||S)r) rrrrrcrrrr)r=rrrr8r8r9_auto_fill_checkpoint_metrics{s  z+_TrainSession._auto_fill_checkpoint_metricscCs.z |jj|td}t|tjyYdSw)Nr)rvrEr r r!r)r=r?er8r8r9rs 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) rMr`rputr{acquirer}is_setclearsysexit)r=rr8r8r9_report_training_results    z%_TrainSession._report_training_resultrNc Csdtjvrddlm}||rtd|jrdS||}d}|r4|j||j |}|jj |t <nd|t <|o>|jj j }|jj|d|rf|jrf|}|jD] \}}||vr`|||<qT||t||d} || dS)NZtorchr)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.rrO)rmodulesZray.air._internal.torch_utilsr ValueErrorrrr^Z_update_checkpoint_indexZpersist_current_checkpointcheckpoint_dir_namerZ sync_configZsync_artifacts_on_checkpointrr\ get_metadatarZ set_metadatarAr) r=rNrMrZpersisted_checkpointZforce_artifact_syncZ user_metadatarrrr8r8r9rs8        z_TrainSession.reportcC|jjSr<)rZr/rQr8r8r9r/z_TrainSession.experiment_namecCrr<)rZr+rQr8r8r9 trial_namerz_TrainSession.trial_namecCrr<)rZr,rQr8r8r9trial_idrz_TrainSession.trial_idcCrr<)rZr0rQr8r8r9r0rz_TrainSession.run_idr)cCrr<)rZr-rQr8r8r9trial_resourcesrz_TrainSession.trial_resourcescCrr<)rZr.rQr8r8r9 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 isinstancedictrrE)r=rZshardr8r8r9get_dataset_shards  z_TrainSession.get_dataset_shard)NNNNFNFr<)Fr@r))*r1r2r3r4rrrr*rr5rrrrLrr>rlrnrgrrrbrr7rrArrrrrrrrrpropertyr/rrr0rrrr8r8r8r9rSps       >  15 3 rS_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||fqSrr8rr8r8r9 s z3_tune_task_and_actor_launch_hook..Nrrsactor.taskcSs"i|] \}}|dkr|t|qSr)r7rr8r8r9rDs"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&Zplacement_grouprDroZget_current_placement_groupr,addget_trial_resourcesZhead_bundle_is_emptyZ bundle_specsr% class_name module_nameZ function_namer) fnr-rrkZcur_pgZpgfZavailable_bundlesZ submittedr+Zmain_resourcesr8r8r9 _tune_task_and_actor_launch_hooksB      rr@cOsDtrtdddlm}m}dtjvrt|_t|_ t |i|adS)NzIA Train session is already in use. Do not call `init_session()` manually.r)rremote_functionZTUNE_DISABLE_RESOURCE_CHECKS) rrrDrrrrrZ_actor_launch_hookZ_task_launch_hookrS)argskwargsrrr8r8r9 init_sessionSs rcCstSr<rr8r8r8r9 get_sessionesrcCsdadS)z#Shuts down the initialized session.Nrr8r8r8r9shutdown_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`rr8r8r8r9!_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)rrri)rsessionr8r8r9get_acceleratorws   rricCs0t}|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.)rrrir)rirr8r8r9set_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)rrfn_namer8r9wrappersz4_warn_session_misuse..inner..wrapper)r1 functoolswraps)rrr)rrr9inners z#_warn_session_misuse..inner)r)rrr8rr9_warn_session_misusesrZstable)Z stability)rMrrNrMrcCs^|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)rM) rHwarning%ray.tune.trainable.trainable_fn_utilsrray.tunerr tunerr)rNrMrrrDr8r8r9rsX 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_checkpointrr`)rrDr8r8r9r$s , rbetacCtjS)z5User metadata dict passed to the Trainer constructor.)rr\r8r8r8r9rbrcCr)z,Experiment name for the corresponding trial.)rr/r8r8r8r9get_experiment_nameirrcCr)z'Trial name for the corresponding trial.)rrr8r8r8r9get_trial_nameprrcCr)z%Trial id for the corresponding trial.)rrr8r8r8r9 get_trial_idwrralphacCr)z0Unique Train Run id for the corresponding trial.)rr0r8r8r8r9 get_run_id~rrr)cCr)z,Trial resources for the corresponding trial.)rrr8r8r8r9rrrcCr)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 )rrr8r8r8r9 get_trial_dirsrrsrcCt}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: ... rYz`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`.)rhasattrrrYrr8r8r9get_world_sizes  rcCr)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: ... rUz`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`.)rrrrUrr8r8r9get_world_rank  rcCr)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: ... rVz`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`.)rrrrVrr8r8r9get_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: ... rXz`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`.)rrrrXrr8r8r9get_local_world_sizerr 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: ... rWz`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`.)rrrrWrr8r8r9 get_node_rankErr 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`.)rrrr)rrr8r8r9rls /  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^r8r8r8r9 get_storages r cCstto tjduS)z6Check if the current process is a Ray Train V1 worker.N)rLrrUr8r8r8r9_in_ray_train_workersr )r@Nr<r)krloggingrrr!rryrcr dataclassesrrtypingrrrrrr r rDZray.air._internal.utilr r Zray.air.constantsr rrrrZray.datarZ ray.trainrZray.train._internal.acceleratorrZray.train._internal.storagerZray.train.constantsrrrrrrrrZray.train.errorrZray.train.utilsr Zray.utilrZray.util.annotationsr"r#Zray.util.debugr$Zray.util.placement_groupr%Zray.util.scheduling_strategiesr&r'r(Z#ray.tune.execution.placement_groupsr) getLoggerr1rHr*r:rArSrrrr6rr5r7rrrrrrrrrrrrrrrrrrrrr r r rr rLr r8r8r8r9s   $    (           @p<&%(%%7