o ci(@sUddlZddlZddlZddlZddlmZmZddlmZddl m Z m Z m Z m Z mZddlZddlmZmZddlmZmZmZddlmZddlmZdd lmZdd lmZdd lm Z m!Z!dd l"m#Z#m$Z$e rydd l%m&Z&ddl'm(Z(e)e*Z+eddGdddZ,eddGdddZ-eddGdddZ.eGdddZ/da0ee/e1d<e2Z3de/fddZ4dddZ5dS) N) dataclassfield)Queue) TYPE_CHECKINGAnyDictListOptional) DataIteratorDataset) BackendConfig Checkpoint DataConfig)session)_TrainingResult)SynchronizationActor)StorageContext) _copy_docinvoke_context_managers) RunConfig ScalingConfig)TrainContextCallback) ThreadRunnerT)frozenc@s|eZdZUdZeddddZeed<eed<e e ee fed<e ed <e ed <e eefed <eed <d efddZdS)TrainRunContextz s zTrainRunContext.)initdefault_factoryrun_id run_configtrain_loop_configscaling_configbackend_configdatasetsdataset_configreturncC|jS)z3Returns the run config of the current training run.)r%selfrrr get_run_config4zTrainRunContext.get_run_configN)__name__ __module__ __qualname____doc__rr$str__annotations__rr rrrr r rr/rrrr rs rc@s6eZdZUeed<eed<eed<eed<eed<dS)DistributedContext world_rank world_size local_ranklocal_world_size node_rankN)r1r2r3intr6rrrr r79s  r7c@s6eZdZUdZeed<eed<ded<eded<dS) ExecutionContextzHolds the execution context for the current worker process. Every worker process has a single execution context accessed via the `TrainContext`, which includes the training thread that is actually running the user code. synchronization_actor result_queuertraining_thread_runnerrtrain_context_callbacksN)r1r2r3r4rr6rrrrrr r>Bs r>c @seZdZUeed<eed<eed<eed<ee e fed<dZ e e ed<eejde fd d Zeejdefd d Zeejdefd dZeejdefddZeejdefddZeejdefddZeejddZddZddZddZde de fddZded fd!d"Z d+d#e e de fd$d%Z d+d#e d&ee efde e de fd'd(Z!  d,d&ee efde e d#e e fd)d*Z"dS)- TrainContexttrain_run_contextdistributed_contextexecution_contextstorage_contextdataset_shardsN checkpointr+cCs |jjjSr)rDr%namer-rrr get_experiment_namebs z TrainContext.get_experiment_namecC|jjSr)rEr9r-rrr get_world_sizefzTrainContext.get_world_sizecCrLr)rEr8r-rrr get_world_rankjrNzTrainContext.get_world_rankcCrLr)rEr:r-rrr get_local_ranknrNzTrainContext.get_local_rankcCrLr)rEr;r-rrr get_local_world_sizerrNz!TrainContext.get_local_world_sizecCrLr)rEr<r-rrr get_node_rankvrNzTrainContext.get_node_rankcCr,r)rGr-rrr get_storagezr0zTrainContext.get_storagecCrLr)rFr@r-rrr get_result_queuezTrainContext.get_result_queuecCrLr)rFr?r-rrr get_synchronization_actorrUz&TrainContext.get_synchronization_actorcCr,r)rIr-rrr get_checkpointszTrainContext.get_checkpoint dataset_namec Cs<z|j|WStytd|dt|jdw)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. Args: dataset_name: Name of the dataset shard. Returns: The ``DataIterator`` shard with the given name for this worker. Raises: KeyError: If the dataset shard with the given name is not found. zDataset z not found. Available datasets: .)rHKeyErrorlistkeys)r.rXrrr get_dataset_shards   zTrainContext.get_dataset_shardrcCrLr)rFrBr-rrr get_context_callbacksrUz"TrainContext.get_context_callbackscheckpoint_dir_namecCs8|p|j}|}t|jj|jj|jj |ddS)zSync the checkpoint dir name across ranks. Args: checkpoint_dir_name: The checkpoint dir name to sync. Returns: The synced checkpoint dir name. zray.train.report)r8r9datacaller_method_name) rG make_default_checkpoint_dir_namerVraygetbroadcast_from_rank_zeroremoterEr8r9)r.r_ sync_actorrrr &_sync_checkpoint_dir_name_across_rankssz3TrainContext._sync_checkpoint_dir_name_across_ranksmetricscCs0|std|dS|j||}||_t||dS)zSave the checkpoint to remote storage. Returns: The training result object containing the persisted checkpoint. N)rIri)rrGpersist_current_checkpointrI)r.r_rirIpersisted_checkpointrrr _save_checkpoints  zTrainContext._save_checkpointcCsdtjvrddlm}||rtdtdd|jjD||}| |||}| |WddS1s=wYdS)a3 Upload checkpoint to remote storage and put a training result on the result queue of this worker process. Args: metrics: The metrics to report. checkpoint: The checkpoint to report. checkpoint_dir_name: The name of the checkpoint dir in this iteration. Note: If not set, the checkpoint will be stored in the default storage path. If set, make sure this value is unique for each iteration. TODO: the report function should be implemented in the worker instead of in the train context. The train context should only keep the train related information and not the worker related actions. This refactor would also require the `TrainContextCallback` to be updated as well. torchr)contains_tensoraPassing 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 (ex: `.numpy()`, `.item()`, etc.) or save tensors as part of the checkpoint files instead.cSsg|]}|jqSr) on_report).0callbackrrr sz'TrainContext.report..N) sysmodulesray.air._internal.torch_utilsrn ValueErrorrrFrBrhrlrTput)r.rirIr_rntraining_resultrrr reports&   "zTrainContext.reportr)NN)#r1r2r3rr6r7r>rrr5r rIr r rrrKrMr=rOrPrQrRrSrTrVrWr]rr^rhrrrlryrrrr rCYsf   !   rC_train_contextr+cCs<ttdur tdtWdS1swYdS)Nz&TrainContext has not been initialized.) _context_lockrz RuntimeErrorrrrr get_train_contexts $r}cCs.t |aWddS1swYdSr)r{rz)contextrrr set_train_contexts"r)r+N)6loggingrs threadingr dataclassesrrqueuertypingrrrrr rcray.datar r ray.trainr r rray.train._internalrray.train._internal.sessionr6ray.train.v2._internal.execution.checkpoint.sync_actorr(ray.train.v2._internal.execution.storagerray.train.v2._internal.utilrrray.train.v2.api.configrr)ray.train.v2._internal.execution.callbackr;ray.train.v2._internal.execution.worker_group.thread_runnerr getLogger__file__loggerrr7r>rCrzr6Lockr{r}rrrrr sB         8