&`iEUddlZddlZddlZddlZddlmZddlmZmZddl m Z ddl m Z m Z mZmZmZmZddlZddlmZddlmZddlmZmZdd lmZdd lmZdd lmZm Z dd l!m"Z"m#Z#dd l$m%Z%m&Z&ddl'm(Z(m)Z)ddl*m+Z+m,Z,e r$ddl-m.Z.m/Z/m0Z0ddl1m2Z2m3Z3ddl4m5Z5ddl6m7Z7ddl8m9Z9ej:e;ZedGddZ?edGddZ@eGddZAdaBeeAeCd <ejDZEd!eAfd"ZFd$d#ZGdS)%N)ThreadPoolExecutor) dataclassfield)Queue) TYPE_CHECKINGAnyCallableDictListOptional)retry) ActorHandle) DataIteratorDataset)AWS_RETRYABLE_TOKENS)SynchronizationActor)StorageContextdelete_fs_path)_TrainingReport_ValidationSpec)'construct_user_exception_with_tracebackinvoke_context_managers) RunConfig ScalingConfig)CheckpointConsistencyModeCheckpointUploadMode) BackendConfig Checkpoint DataConfig)DatasetShardMetadataDatasetShardProvider)TrainContextCallback) ThreadRunner)ReportedCheckpointT)frozenceZdZUdZeddZeed<eed<e e ee fed<e ed<d ed <e ee fed <d ed <defdZdS)TrainRunContextzzTrainRunContext.6sDJLLrxs r0r|z)TrainContext.get_all_reported_checkpointss:w  ! > E E&      r/ dataset_infor c6|j|S)a2Returns 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_info: The shard metadata, including the dataset name and worker rank. Returns: The ``DataIterator`` shard with the given name for this worker. Raises: KeyError: If the dataset shard with the given name is not found. )r[get_dataset_shard)r>r~s r0rzTrainContext.get_dataset_shards*<<\JJJr/r"c|jjSr*)rXrSr=s r0get_context_callbacksz"TrainContext.get_context_callbackss%==r/checkpoint_dir_namec|p|j}|}tj|j|jj|jj |dS)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)rHrIdatacaller_method_name) rY make_default_checkpoint_dir_namerurzr{broadcast_from_rank_zeror}rWrHrI)r>r sync_actors r0&_sync_checkpoint_dir_name_across_ranksz3TrainContext._sync_checkpoint_dir_name_across_rankss~  G#DDFF  3355 w  / 6 63>3>(#5 7     r/zupload checkpoint) description max_attemptsmatchFmetrics$delete_local_checkpoint_after_uploadcheckpoint_upload_fnvalidation_specc|std|dS |r=|||}|t|tjjst dn|j||}n-#t$r t d|dwxYw|rI t|j |j n-#t$r t d|YnwxYwt|||S)aSave the checkpoint to remote storage. Args: checkpoint_dir_name: The checkpoint dir to persist to. metrics: The metrics to report. checkpoint: The checkpoint to report. delete_local_checkpoint_after_upload: Whether to delete the checkpoint after it is uploaded. checkpoint_upload_fn: A user defined function that will be called with the checkpoint to upload it. If not provided, defaults to using the `pyarrow.fs.copy_files` utility for copying to the destination `storage_path`. validation_spec: The validation specification. Returns: The training result object containing the persisted checkpoint. Nr\rrz:checkpoint_upload_fn must return a `ray.train.Checkpoint`.z Failed to find local checkpoint z when attempting to upload it. This could be caused by multiple workers on a node attempting to upload the same directory, and then one of the workers deletes the directory before the others finish.zAFailed to delete the local checkpoint after a successful upload: )r isinstancerztrainr ValueErrorrYpersist_current_checkpointFileNotFoundErrorlogger exceptionr filesystempath Exception)r>rrr\rrrpersisted_checkpoints r0_upload_checkpointzTrainContext._upload_checkpoints8 "$   # ';'; 3(($(/z(#)*>88/%T0(,';'V'V 3(($!      !:!!!      0  z4joFFFF     dXbdd  ++    sAA11*B!B<<'C&%C&training_reportcj5jfdtdd||jr |j_|xjdz c_jddddS#1swxYwYdS)aThread waits for its turn before reporting training result to result queue. It does this in order to guarantee the FIFO processing of checkpoints. The queue size is set to 1 to avoid accumulating unprocessed results. If the queue is full, the put operation blocks until a result is consumed. TODO: Add a metric to track the blocking time waiting for the training result to be consumed by the controller. c jdz kS)Nr%)r])r^r>sr0r1z0TrainContext._wait_then_report..5s15F5JJr/zReporting training result z: r%N) r_wait_forrinfor\rsputr] notify_all)r>rr^s` `r0_wait_then_reportzTrainContext._wait_then_report&s) ( 5 5  ' 0 0JJJJJ    KKS->SS/SS   ) ="1"<  ! ! # # ' ' 8 8 8  % % * % %  ' 2 2 4 4 4 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5sB"B99B=B=checkpoint_upload_mode validate_fnvalidate_configc Z dtjvr ddlm} | |rt dt djjD5|rt|| nd xj dz c_ j } |}|tj kr1 ||| } | | n|tjkr)t!|| } | | n|tjkr^d t$d t&t$t(fd t*d dt,ddf  fd } j| |||| nt d|ddddS#1swxYwYdS)a Upload checkpoint to remote storage and put a training result on the result queue of this worker process. 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.cg|] }|j Sr.) on_report).0callbacks r0 z'TrainContext.report..ds+   "   r/)rrNr%rrrr\rr^r;cJ ||| }||dS#t$rc}tdjjt|Yd}~dSd}~wwxYw)NzCheckpoint upload failed in the background thread. Raising eagerly to avoid training in a corrupted state with more potential progress lost due to checkpointing failures.) rrrrrrXrRget_exception_queuerr) rrr\r^rerrr>rs r0_upload_checkpoint_and_reportz:TrainContext.report.._upload_checkpoint_and_reports *.*A*A/#&@0+ ++..@QRRRRR$   ((B .EYY[[__CAFF s05 B"ABB"z Invalid checkpoint upload mode: )sysmodulesray.air._internal.torch_utilsrrrrXrSrr^rrSYNCrr NO_UPLOADrASYNCrDr rr rMrasubmit)r>rr\rrrrrrrr^rrrs` `` @r0reportzTrainContext.reportAs, ck ! ! E E E E E Ew''  O%   $ 6 N     P P   '"1 +$3### #'  " "a ' " " $ 6 #'"M"M###  &)=)BBB"&"9"9'8(# ##&&8IJJJJ'+?+III"1)#$3### &&8IJJJJ'+?+EEE),!#s(^!) 6(+   61881'% !O7MOO]P P P P P P P P P P P P P P P P P P sD=F  F$'F$r*)NFNN)4r@rArBr(rErGrOrrr\r r]rMr^ threading Conditionr_rMAX_CHECKPOINT_UPLOAD_THREADSrarDrdrfrirkrmrorqrsrurwr VALIDATEDr r|rrrrr rr rboolr rrrrrrrr.r/r0rUrUosD&&&&++++''''####!!!!2222*.J&--- !#!!!s2E)2E2G2GI/GGG7I7I1888 "46S66663333333333333339c99992s2222$$$333<<<### 7P6Y   3   " #     K.DKKKKK >t,B'C>>>>48  #+C=     < U*BVWWW .25: 59E E  E c3hE \* E /3 E ' lC(,6 7 E "/2E  E E E XWE N5.5CF5555<.2-17K7P?C PT*.rrc3hr\*r&c] r !5 r /7tn r' lC(,6 7 rh htn'Et'KLMr"$r rrrrrrr/rU_train_contextr;c~t5ttdtcdddS#1swxYwYdS)abGet the internal train context. Note: This should not be used directly by user-facing APIs. User-facing APIs should call :class:`~ray.train.v2._internal.execution.train_fn_utils.TrainFnUtils` or use :class:`~ray.train.v2.api.context.TrainContext` instead. Returns: The internal TrainContext for this worker. Nz&TrainContext has not been initialized.) _context_lockr RuntimeErrorr.r/r0get_train_contextrs   !GHH Hs 266cJt5|addddS#1swxYwYdSr*)rr)contexts r0set_train_contextrsq !! !!!!!!!!!!!!!!!!!!s )r;N)Hloggingrrr+concurrent.futuresr dataclassesrrqueuertypingrrr r r r rzray._common.retryr ray.actorrray.datarr ray.train.v2._internal.constantsr6ray.train.v2._internal.execution.checkpoint.sync_actorr(ray.train.v2._internal.execution.storagerr0ray.train.v2._internal.execution.training_reportrrray.train.v2._internal.utilrrray.train.v2.api.configrrray.train.v2.api.report_configrr ray.trainrrr2ray.train.v2._internal.data_integration.interfacesr r!)ray.train.v2._internal.execution.callbackr";ray.train.v2._internal.execution.worker_group.thread_runnerr#$ray.train.v2.api.reported_checkpointr$ getLogger__file__rrr(rGrOrUrrELockrrrr.r/r0rs  111111((((((((EEEEEEEEEEEEEEEE ######!!!!!!********AAAAAAWWWWWWSSSSSSSS=<<<<<<< H??????????ONNNNNXXXXXXGGGGGG  8 $ $!" $: $ $::::::::, CCCCCCC CN *.&---   <"!!!!!!r/