&`i ~UddlZddlZddlZddlZddlZddlZddlZddlZddlZddl 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 r ddlm?Z?ddl@mAZAejBeCZDe GddZEGddZFGddZGe6GddZHeIZJeeKeLd<daMeeHeLd<deeNeOfd ee>fd!ZPdKd#ZQd"eeHfd$ZRd%ZSd&ZTd'ee#d"e#fd(ZUd)e#d"dfd*ZVdLd+efd,ZWe7d-.eWddd/d0ed1ee!d2eeNd"dfd3ZXe7d-.eWd"ee!fd4ZYe7d5.eWd"eeNeffd6ZZe7d5.eWd"eNfd7Z[e7d5.eWd"eNfd8Z\e7d5.eWd"eNfd9Z]e7d:.eWd"eNfd;Z^e7d5.eWdMd=Z_e7d5.eWd"eNfd>Z`e7d5.eWd?@d"eafdAZbe7d5.eWd@d"eafdBZce7d5.eWd@d"eafdCZde7d5.eWd@d"eafdDZee7d5.eWd@d"eafdEZfe7d-.eW dLdFeeNd"edGfdHZge6eWd"e%fdIZhd"eifdJZjdS)NN) 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)PlacementGroupFactoryceZdZUdZeed<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__rfloatr3r r4o/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/train/_internal/session.pyr,r,8s== III GGGCJ KKKNNN%)OXc]))) FHSM     r=r,cFeZdZdZdejfdZd dededfdZ 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. futurec||_dSN)rA)selfrAs r>__init__z_FutureTrainingResult.__init__Ns  r=Tblockreturn_TrainingResultc|rd}nd} tj|j|S#t$rYdSt$r(}t d|Yd}~dSd}~wwxYw)zResolve into ``_TrainingResult``. This will return None for function trainables if no checkpoint has been saved before. Ng& .>timeoutzError resolving result: )raygetrA TimeoutError Exceptionloggererror)rDrFrKexcs r>resolvez_FutureTrainingResult.resolveQs  GGG ;74;888 8    DD ; ; ; LL9C99 : : : : : : : : : ;s$ A" A"AA"N)T) r5r6r7r8rL ObjectRefrEboolr rSr<r=r>r@r@Fshs};;T;X6G-H;;;;;;r=r@cJeZdZdZdeedeeeffdZ defdZ dS)rHz4A (checkpoint, metrics) result reported by the user. checkpointmetricsc"||_||_dSrCrWrX)rDrWrXs r>rEz_TrainingResult.__init__gs$ r=rGc(d|jd|jdS)NzTrainingResult(checkpoint=z , metrics=)rZrDs r>__repr__z_TrainingResult.__repr__ksVDOVVt|VVVVr=N) r5r6r7r8r rrr9rrEr^r<r=r>rHrHdsk>>8J#7$sCx.W#WWWWWWr=rHceZdZdZ d3dedeedeedeedeed eed eed eee e fd ee e fd ee de deede fdZde de fdZde de fdZdZdZ d4ded edefdZdZd4deedee fdZdeefdZdZde deefd Zd!edefd"Zd!edefd#Zd5d$Zd%eddfd&Z d4d'ed ee ddfd(Z!e"de fd)Z#e"de fd*Z$e"de fd+Z%e"de fd,Z&e"d6d.Z'e"de fd/Z( d4d0ee ded1fd2Z)dS)7 _TrainSessionz.Holds information for training on each worker.NF training_func world_rank local_rank node_ranklocal_world_size world_size trial_info dataset_shardmetadatarWdetailed_autofilled_metricsstoragesynchronous_result_reportingc| |_||_| |_||_||_||_||_||_| sJt d|d| | ||| | | |_ tj |_ d|_d|_||_d|_i|_dS)Nz StorageContext on SESSION (rank=z): )rargrkloaded_checkpointrg)rlrhrirbrcrdrerfrPdebugresetrjtimelast_report_time iteration time_totalget_current_iplocal_ip accelerator_state)rDrarbrcrdrerfrgrhrirWrjrkrls r>rEz_TrainSession.__init__ts6-I)+  $$" 0$w Q QQQQRRR '!(    ,G( $ ++--  r=keyrGc6|j|SrC)rxrM)rDrys r> get_statez_TrainSession.get_states{s###r=valuec||j|<dSrC)rx)rDryr|s r> set_statez_TrainSession.set_states  Cr=cVtj|_|jSrC)rLutilget_node_ip_addressrvr]s r>ruz_TrainSession.get_current_ips4466 }r=cFd|_|jdS)zStarts the training thread.TN)training_startedtraining_threadstartr]s r>rz_TrainSession.starts% $ ""$$$$$r=ctjd|_tj|_t jd|_d|_t jd|_ t|d|j |_ ||_ ||_ ||_i|_d|_d|_d|_t'j|jdt-t/t&jt4dr=t6d|jt'j|jdSdS) NrT)targetdaemon error_queueF)exist_ok1z#Changing the working directory to: ) threading Semaphore continue_lockEvent stop_eventr"Queue result_queue_inter_actor_queuerr rrgrkrnrx ignore_reportr _first_reportosmakedirstrial_working_directoryrUintenvironrMrrProchdir)rDrargrkrns r>rpz_TrainSession.resets@'033$/++"KNNDH !;q>> , 4;K    % !2 " %! G3dCCCC BJNN#93??@@ A A 6 LLWg6UWW    HW4 5 5 5 5 5  6 6r=cd|_dS)z-Ignore all future ``session.report()`` calls.TN)rr]s r>pause_reportingz_TrainSession.pause_reportings!r=rKc|j|j|jdd}|jr|j|}|S)zSFinishes the training thread. Raises any Exception from training. TforceNrJ) rsetrreleaserkpersist_artifactsrrjoin)rDrKoutputs r>finishz_TrainSession.finishs{  ""$$$ &&T&222    @)..w.??F r=c2|jstd|jr'|js|jd|_d}|J|jr1|d}||j1||d}|| dn3|j st d|js|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.FNTrFzVRunner error waiting to be raised in main thread. Logging all available results first.)r RuntimeErrorrlrrrris_alive_get_result_from_queues_report_thread_runner_errorremptyrPro)rDresults r>get_nextz_TrainSession.get_next s@ $ MKLL L  , '% -"**,,,!&D n!5!>!>!@!@n111==Fn!5!>!>!@!@n >111>>F >  , ,4 , 8 8 8 8#))++  ?0 )   & & ( ( ( r=cX|jtjdddi|_|jS)z$Get or create the inter-actor queue.Nrnum_cpusr) actor_options)r ray_queuerr]s r> _get_or_create_inter_actor_queuez._TrainSession._get_or_create_inter_actor_queueBs0  " *&/oa TU&W&W&WD #&&r=rFcXd}|jh |j|t}|r.|j||n#t j$rYnwxYw |j|t}n#tj$rYnwxYw|S)zUGet result from result queue. Pass result from training actor result queue if needed.NrFrK) rrMrrrreportrEmptyrr")rDrFrinter_actor_items r>rz%_TrainSession._get_result_from_queuesHs  " . #'#:#>#>)>$?$$ $2&..000KK 0111?     &**@U*VVFF{    D  s$AAA/.A/3!BB'&B'rc tj}tj}t|vr|t}n ||jz }|xjdz c_|xj|z c_||_tttj | t|jttjtt!jt$|ji}|jsd|D}|}|||S)-Add autofilled metrics and update attributes.rc,i|]\}}|tv||Sr<)r.0kvs r> z4_TrainSession._auto_fill_metrics..qs4###Aq4441444r=)rqrnowrrrrsrtrrmktime timetuplerrrgetpidrplatformnoderrvrjitemscopyupdate)rDr current_timecurrent_datetimetime_this_iterauto_filled_metricss r>_auto_fill_metricsz _TrainSession._auto_fill_metrics\sy{{ #<>> v % %#$45NN)D,AAN ! >) , s4;'7'A'A'C'CDDEE $/  X]__ DM  / ##/5577###   )*** r=ctj}ttt j|i}|}|||S)r) rrrrrqrrrr)rDrrrs r>_auto_fill_checkpoint_metricsz+_TrainSession._auto_fill_checkpoint_metrics{sc#<>> s4;'7'A'A'C'CDDEE  )*** r=c |j|t}t|#tj$rYdSwxYw)Nr)rrMrr r"r)rDrFes r>rz)_TrainSession._report_thread_runner_errorsO  $$5:N$OOA a '{    DD s )+>>training_resultc&|jr |j|_|j|d|j|jr/|jtj 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) rWrnrputracquireris_setclearsysexit)rDrs r>_report_training_resultz%_TrainSession._report_training_results  % @%4%?D " oT::: ""$$$ ? ! ! # #  O ! ! # # # HQKKKKK  r=rXcdtjvr ddlm}||rt d|jrdS||}d}|rI|j||j |}|jj |t<n d|t<|o|jj j }|j||rX|jrQ|}|jD]\}}||vr|||<||t'||} || 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.rrZ)rmodulesray.air._internal.torch_utilsr ValueErrorrrrk_update_checkpoint_indexpersist_current_checkpointcheckpoint_dir_namer sync_configsync_artifacts_on_checkpointrri get_metadatar set_metadatarHr) rDrXrWrpersisted_checkpointforce_artifact_sync user_metadatarrrs r>rz_TrainSession.reports ck ! ! E E E E E Ew''  K    F))'22#  0 L 1 1' : : :$(<#J#J:#V#V +/<+KG' ( (+/G' ( ! F (E  &&-@&AAA  =DM =0==??M ++-- ) )1M))'(M!$ - -m < < < ,@'RRR $$V,,,,,r=c|jjSrC)rgr3r]s r>r3z_TrainSession.experiment_names ..r=c|jjSrC)rgr-r]s r> trial_namez_TrainSession.trial_names ##r=c|jjSrC)rgr.r]s r>trial_idz_TrainSession.trial_ids !!r=c|jjSrC)rgr4r]s r>r4z_TrainSession.run_id %%r=r*c|jjSrC)rgr/r]s r>trial_resourcesz_TrainSession.trial_resourcess ((r=c|jjSrC)rgr0r]s r> trial_dirz_TrainSession.trial_dirrr= dataset_namer)c|j}|tjdn;t|tr&|st d||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.)rhwarningswarn isinstancedictrrM)rDrshards r>get_dataset_shardz_TrainSession.get_dataset_shards{" = M     t $ $ + "1 99\** * r=)NNNNFNFrC)FrGr*)*r5r6r7r8rr rr,rr9rrrrUrrEr{r~rurrprr;rrHrrrrrrrrrpropertyr3rrr4rrrr<r=r>r`r`ps88+/6:#'+/,1,0-2<<<SM<SM < C= < #3- <SM<Y'< S'\ 23<sCx.<Z(<&*<.)<'+<<<<|$S$S$$$$!S!!!!!%%% /6/6/6/6 /6/6/6/6b"""huo#.3(?33333j''' Th6O($> D T    481-1-d1-0D1-PT1-1-1-1-f////X/$C$$$X$"#"""X"&&&&X&)))X)&3&&&X& '+sm . !r=r`_checked_resources_sessionr/strategyc td|D}|r |tvrdSt|tr|jdSt j}|r|jj |j krdSt |t}|j r|j dd}n|j dd}t||rdS|jr d}|jdz|jzdz|jz}nd}|jdz|jz}|j d} d|D}t%d |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. c(h|]\}}|dk ||fSrr<rs r> z3_tune_task_and_actor_launch_hook..s%CCC1QUUaVUUUr=Nrractor.taskc@i|]\}}|dk |t|Sr)r;rs r>rz4_tune_task_and_actor_launch_hook..Ds)DDDAa!eeE!HHeeer=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_grouprLrget_current_placement_groupr.addget_trial_resourceshead_bundle_is_empty bundle_specsr& class_name module_name function_namer) fnr/rrycur_pgpgfavailable_bundles submittedr-main_resourcess r> _tune_task_and_actor_launch_hookrs CC (9(9CCC D DC #+++ x!A B B  # +X 1 1 3 3F X-0FI==3   C 4"/3"/3Y(9:: }7 ~#bm3c9B init_sessionr&Ssn  )   +*******%RZ77#C ,L)d-f--HHHr=ctSrCrr<r=r> get_sessionr)es Or=c dadS)z#Shuts down the initialized session.Nr(r<r=r>shutdown_sessionr+is HHHr=c td)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>!_raise_accelerator_session_misuser-os  -  r=default_accelerator_clscxt}|t|j||_|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. )r)r-rw)r.sessions r>get_acceleratorr1wsAmmG)+++"5577  r=rwc|t}|t|jtd||_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.)r)r-rwr)rwr0s r>set_acceleratorr3sDmmG)+++&@AAA%Gr= default_valuec"dtffd }|S)zKWarns if fn is being used outside of session and returns ``default_value``.rc\jtjfd}|S)Nct}|s6ttdrtjdddS|i|S)N-`zb` is meant to only be called inside a function that is executed by a Tuner or Trainer. Returning `z`.)r)r%rrr)r$r%r0r4rfn_names r>wrapperz4_warn_session_misuse..inner..wrappers!mmG %:FFWFFGGMEGEE3@EEE %$2t&v&& &r=)r5 functoolswraps)rr;r:r4s` @r>innerz#_warn_session_misuse..innersK+    ' ' ' ' ' '   'r=)r)r4r>s` r>_warn_session_misuser?s/($ Lr=stable) stability)rWrrXrWrc|tdddlm}|r=ddl}t rt d|j||St||dS)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)rW) rPwarning%ray.tune.trainable.trainable_fn_utilsrDray.tunerr!tunerr))rXrWrrDrLs r>rrsp& -   GFFFFF ? ) + +  $B   xw:>>>MMZ88888r=cddlm}|r:ddl}trt d|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. rrCNz`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) rFrDrGrr!rHget_checkpointr)rn)rDrLs r>rJrJ$s{XGFFFFF ) ) + +  $B   x&&((( == **r=betac(tjS)z5User metadata dict passed to the Trainer constructor.)r)rir<r=r>rrb == !!r=c(tjS)z,Experiment name for the corresponding trial.)r)r3r<r=r>get_experiment_namerOi == ((r=c(tjS)z'Trial name for the corresponding trial.)r)rr<r=r>get_trial_namerRps == ##r=c(tjS)z%Trial id for the corresponding trial.)r)rr<r=r> get_trial_idrTwrMr=alphac(tjS)z0Unique Train Run id for the corresponding trial.)r)r4r<r=r> get_run_idrW~s == r=r*c(tjS)z,Trial resources for the corresponding trial.)r)rr<r=r>rrrPr=c(tjS)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 )r)rr<r=r> get_trial_dirrZs, == ""r=r)r4cjt}t|dstd|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: ... rfz`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`.)r)hasattrrrfr0s r>get_world_sizer^sA<mmG 7L ) )  9    r=cjt}t|dstd|jS)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: ... rbz`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`.)r)r\rrbr]s r>get_world_rankr`sA:mmG 7L ) )  9    r=cjt}t|dstd|jS)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: ... rcz`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`.)r)r\rrcr]s r>get_local_rankrbsB@mmG 7L ) )  9    r=cjt}t|dstd|jS)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: ... rez`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`.)r)r\rrer]s r>get_local_world_sizerdsB:mmG 7. / /  9    ##r=cjt}t|dstd|jS)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: ... rdz`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`.)r)r\rrdr]s r> get_node_rankrfEsA:mmG 7K ( (  9    r=rr)ct}t|dstd||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`.)r)r\rr)rr0s r>rrlsN^mmG 7/ 0 0  9    $ $\ 2 22r=c(tjS)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. )r)rkr<r=r> get_storageris ==  r=cbttotjduS)z6Check if the current process is a Ray Train V1 worker.N)rUr)rbr<r=r>_in_ray_train_workerrks%    G;==#;4#GGr=)rGNrCr)kr<loggingrrr"rrrqr dataclassesrrtypingrrrrr r r rLray.air._internal.utilr r ray.air.constantsrrrrrray.datar ray.trainrray.train._internal.acceleratorrray.train._internal.storagerray.train.constantsrrrrrrrrray.train.errorr ray.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* getLoggerr5rPr,r@rHr`rrrr:rr9r;rr&r)r+r-r1r3r?rrJrrOrRrTrWrrZrr^r`rbrdrfrrirUrkr<r=r>rs    !!!!!!JJJJJJJJJJJJJJJJJJ ???????? 777777666666                    /.....444444''''''88888888######:::::: J%%%%%%IIIIII  8 $ $  ! ! ! ! ! ! !  !;;;;;;;;<WWWWWWWWWWWWWWWWv &)SUUC N***%)(= !(((=U #=/78K/L====@....$Xm, T+->;"&&&&&&$0 X(,)- m9m9m9 m9$m9"# m9  m9m9m9m9` X9+,9+9+9+9+x V"d38n""""  V)S))))  V$$$$$  V"c""""  W C     V))))  V#s####. VA&&&####'&#L VA&&&""""'&"J VA&&&%%%%'&%P VA&&&"$c"$"$"$'&"$J VA&&&"s"""'&"J X"&43433-43 n43434343n!^!!!!HdHHHHHHr=