o ciw@sddlZddlZddlZddlmZddlmZddlmZm Z m Z m Z m Z m Z mZmZddlZddlmmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lm Z dd l!m"Z"m#Z#dd l$m%Z%m&Z&m'Z'm(Z(m)Z)ddl*m+Z+ddl,m-Z-ddl.m/Z/ddl0m1Z1ddl2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:ddl;me?e@ZAGdddeBZCGdddeBZDeGdddZEGdddZFGdddeBZGGdd d ZHd!eIfd"d#ZJdS)$N) defaultdict) dataclass)AnyCallableDictListOptionalTupleTypeTypeVar)HIP_VISIBLE_DEVICES_ENV_VAR)NEURON_RT_VISIBLE_CORES_ENV_VAR)!ASCEND_RT_VISIBLE_DEVICES_ENV_VAR)CUDA_VISIBLE_DEVICES_ENV_VAR) env_integer)Dataset) RayActorError) Checkpoint DataConfig) TrialInfo_TrainingResult get_session init_sessionshutdown_session)StorageContext)check_for_failure) WorkerGroup) BackendConfig)&ENABLE_DETAILED_AUTOFILLED_METRICS_ENV%ENABLE_SHARE_CUDA_VISIBLE_DEVICES_ENV$ENABLE_SHARE_HIP_VISIBLE_DEVICES_ENV)ENABLE_SHARE_NEURON_CORES_ACCELERATOR_ENV'ENABLE_SHARE_NPU_RT_VISIBLE_DEVICES_ENVRAY_TRAIN_ENABLE_STATE_TRACKINGTRAIN_ENABLE_WORKER_SPREAD_ENV#TRAIN_PLACEMENT_GROUP_TIMEOUT_S_ENV)get_current_placement_groupremove_placement_groupTc@eZdZdZdS)TrainBackendErrorz?Errors with BackendExecutor that should not be exposed to user.N__name__ __module__ __qualname____doc__r0r0X/home/ubuntu/.local/lib/python3.10/site-packages/ray/train/_internal/backend_executor.pyr*.r*c@r))TrainingWorkerErrorz)Raised if a worker fails during training.Nr+r0r0r0r1r32r2r3c@s*eZdZUdZeed<eed<eed<dS)ResourceConfiga Resource configuration for resource_ids to share between workers. Args: resource_name: The name of the resource to configure (Example: "neuron_cores" or "gpu"). resource_enable_sharing_env_var: The environment variable to check if the resource should be shared. share_resource_ids_env_var: The environment variable to configure for sharing the resources with other workers. resource_nameresource_enable_sharing_env_varshare_resource_ids_env_varN)r,r-r.r/str__annotations__r0r0r0r1r46s   r4c@seZdZdZ    dEdedeededeee e fd ef d d Z    dFd ee gdfd ee deedeefddZddZddZde de fddZde de fddZdeefdd Z dGd!e gefd"ee efd#ee efd$ed%ed&eeddfd'd(Zdeeefd)d*Zd+d,Zd-d.Z /  dHd0e!d1eed2ee fd3d4Z"d5d6Z#dId8e!fd9d:Z$d;d<Z%d=d>Z&d?d@Z'dAdBZ(dCdDZ)dS)JBackendExecutoraMain execution class for training backends. This class holds a worker group and is responsible for executing the training function on the workers, and collecting intermediate results from ``session.report()``. Args: backend_config: The configurations for this specific backend. num_workers: Number of workers to use for training. resources_per_worker (Optional[Dict[str, float]]): Dictionary specifying the resources that will be requested for each worker. Defaults to {"CPU": 1}. max_retries: Number of retries when Ray actors fail. Defaults to 3. Set to -1 for unlimited retries. Nbackend_config trial_info num_workersresources_per_worker max_retriescCs|dur ddi|_n||_||_||_||_||_|jdkr'td|_d|_d|_ d|_ d|_ ||_ t |_d|_ttjttttjttttjttg|_ttd|_tt d|_!dS)NCPUr;rinf)"_resources_per_workercopy_backend_config backend_cls_backend _num_workers _max_failuresfloat _num_failures _last_failure_initialization_hook_placement_group _trial_infoInactiveWorkerGroup worker_groupdataset_shardsr4 ray_constants NEURON_CORESr!r NPUr"rGPUr r _resource_configsinttime_start_time_msrr#state_tracking_enabled)selfr=r>r?r@rAr0r0r1__init__[sD     zBackendExecutor.__init__initialization_hook train_clstrain_cls_argstrain_cls_kwargsc s||jpd}t|j|j||||d|_|jr|jjnd}|j|z\|r1||_ |j |ddl m dffdd }|j | ttt|jj}|jd ddkra|ra||jD]} || j| jrv|| j| jqd|j|j|jWn&ty} ztt | t!d |"|#WYd} ~ nd} ~ ww|j$rdd l%m&} dd l'm(} | | d |_)dSdS)zStarts the worker group.default)r?r@ actor_clsactor_cls_argsactor_cls_kwargsplacement_groupNr DataContextctxcs|dSN) _set_current)rkrir0r1_set_driver_dataset_contextz:BackendExecutor.start.._set_driver_dataset_contextrXzXFailure occurred during startup. Restarting all workers and attempting to startup again.)TrainRunStateManager)get_state_actor) state_actor)*_create_placement_grouprPrrJrErSrQdriver_node_id"sort_workers_by_node_id_and_gpu_idrOexecuteray.datarj get_currentboolrrrIshare_cuda_visible_devicesget_share_cuda_visible_devicesrY_is_share_resources_enabledr5r6_share_resource_idsr7on_startrGrlogger exceptionr8warning_increment_failures_restartr]ray.train._internal.staterp%ray.train._internal.state.state_actorrq state_manager) r^r`rarbrcrhtrial_driver_node_idrn"share_cuda_visible_devices_enabledresource_configexcrprqr0rir1startst       zBackendExecutor.startc st}tjjj}|j}|dup| }|rcfddtjD}tt t d}|r+dnd}tj j ||d}t dt td } tj|g| d \} } | rSt d n td t|j|_dSdS) aCreates a placement group if it does not exist. If a placement group is already detected (Tune) this will be a no-op. By default the placement group will be created with PACK strategy. This is optimized for colocating GPUs on a minimal number of nodes. This behavior can be overridden to use the SPREAD strategy by defining ``TRAIN_ENABLE_WORKER_SPREAD_ENV`` If a placement group is created it will be stored as self._placement_group. Ncsg|]}jqSr0)rErF).0_r^r0r1 s z;BackendExecutor._create_placement_group..rSPREADPACK)strategyz%Waiting for placement group to start.d)timeoutzPlacement group has started.aPlacement group creation timed out. Make sure your cluster either has enough resources or use an autoscaling cluster. If you are running on a cluster, make sure you specify an address in `ray.init()`, for example, `ray.init("auto")`. You can also increase the timeout by setting the TRAIN_PLACEMENT_GROUP_TIMEOUT_S environment variable. Current resources available: {}, resources requested by the placement group: {})r&ray_privateworker global_worker-should_capture_child_tasks_in_placement_grouprangerJryrr$utilrhrdebugr%waitready TimeoutErrorformatavailable_resources bundle_specsrP) r^current_placement_grouprrshould_create_placement_groupbundles use_spreadrrhrrrr0rr1rss6        z'BackendExecutor._create_placement_groupcCs|tjtdS)aFSets CUDA_VISIBLE_DEVICES on all workers. For each worker, CUDA_VISIBLE_DEVICES will be set to the GPU IDs visible to all workers on that worker's node. This allows GPU workers on the same node to communicate with one another. Example: Setup: - Node1: - Worker1: {0, 1} - Worker2: {2, 3} - Node2: - Worker3: {0, 1} CUDA_VISIBLE_DEVICES: - Worker1: "0,1,2,3" - Worker2: "0,1,2,3" - Worker3: "0,1" N)r~rUrXrrr0r0r1r|sz+BackendExecutor._share_cuda_visible_devicesresourceenv_varc sfdd|jjD}tt}tt}t|D]\}\}}||||||qg} |D]%\}}t|}d |fdd} ||D] }| |j || qJq2t | dS)aSets the given env_var on all workers. For each worker, the cores/devices are visible to all the workers on that worker's node.This allows workers on the same node to communicate with one another. Example: Setup: - Node1: - Worker1: {0, 1} - Worker2: {2, 3} - Node2: - Worker3: {0, 1} NEURON_RT_VISIBLE_CORES/TPU_VISIBLE_CHIPS/...: - Worker1: "0,1,2,3" - Worker2: "0,1,2,3" - Worker2: "0,1" Args: resource: The name of the resource/accelerator. env_var: The name of the environment variable to set. cs g|] }|jj|jjfqSr0)metadatanode_id resource_ids)rw)rr0r1rMs  z7BackendExecutor._share_resource_ids..,cstj<dSrl)osenvironr0)all_resource_idsrr0r1set_resource_ids`roz=BackendExecutor._share_resource_ids..set_resource_idsN)rSworkersrset enumerateaddupdateitemssortedjoinappendexecute_single_asyncrr{) r^rrnode_ids_and_resource_idsnode_id_to_worker_idnode_id_to_resource_ids worker_idrrfuturesrr0)rrrr1r~4s&    z#BackendExecutor._share_resource_idsr5enable_sharing_envcCs"|j|ddk}|ot|dS)aWhether to share resource IDs on all workers based on enable_sharing_env. This will return true if resources are requested and greater than 0. Also, user can disable by configuring the `enable_sharing_env` to "0". Args: resource_name: The name of the resource/accelerator. enable_sharing_env: The name of the environment variable to check. rT)rEr{rUenv_bool)r^r5rhas_resource_requestedr0r0r1r}is z+BackendExecutor._is_share_resources_enabledreturnc sii}ii}d}tt}tt|jD],}|jj|}|jj}|||<||d7<||vr;|||<|d7}|||<qtt|jD]}|jj|}|jj}||||<qIdfddt |jjD}t d||fS)aoCreate rank and world size mappings for workers. There are three maps returned: - local_rank_map, which maps from worker world_rank to local_rank. - local_world_size_map, which maps from world_rank to local_world_size - node_rank_map, which maps from world rank to node rank Example: Worker 0: node 0 Worker 1: node 0 Worker 2: node 1 Worker 3: node 0 Worker 4: node 1 Workers 0, 1, 3 are on node 0. Workers 2, 4 are on node 1. Expected local_rank_map: { 0 -> 0, 1 -> 1, 2 -> 0, 3 -> 2, 4 -> 1 } Expected local_world_size_map: { 0 -> 3, 1 -> 3, 2 -> 2, 3 -> 3, 4 -> 2 } Expected node_rank_map: { 0 -> 0, 1 -> 0, 2 -> 1, 3 -> 0, 4 -> 1 } rr; csLg|]"\}}d|jjd|jjd|jjd|d|d| qS)z - (node_id=z, ip=z, pid=z ) world_rank=z , local_rank=z , node_rank=)rrnode_ippid)rirlocal_rank_map node_rank_mapr0r1rszDBackendExecutor._create_rank_world_size_mappings..z'Started distributed worker processes: ) rrZrlenrSrrrrrrinfo) r^local_world_size_mapnode_idsnode_cnt node_id_dict world_rankrr workers_infor0rr1 _create_rank_world_size_mappingszs8-      z0BackendExecutor._create_rank_world_size_mappings train_funcdatasetsr data_configstorage checkpointcsXttdfdd}|jdur/dd|jjD}dd|jjD} |j|t|j|| d|_|\} } } g} tt|jD]%}| |jj |||| || || |t|j|j ||j||||d q?|j |j|j|| |jrdd lm}tj}|jj|j j|j j||||j|j|j|jg|jd d d }|j |dS)aWExecutes a training function on all workers in a separate thread. ``finish_training`` should be called after this. Args: train_func: The training function to run on each worker. datasets: The base datasets. data_config: The config object for creating dataset shards for workers. checkpoint: The checkpoint data that should be loaded onto each worker and accessed by the training function via ``session.get_checkpoint()``. If this is ``None`` then no checkpoint will be loaded. rc s>zt||||||||| || d WdStytdw)N) training_funcr local_rank node_ranklocal_world_size world_sizer> dataset_shardrrdetailed_autofilled_metricsrzAttempting to start training but a previous training run is still ongoing. You must call `finish_training` before calling `start_training` again.)r ValueErrorr*) rrrrrrr>rrrruse_detailed_autofilled_metricsr0r1initialize_sessions(  z:BackendExecutor.start_training..initialize_sessionNcSsg|]}|jqSr0)actorrrr0r0r1r sz2BackendExecutor.start_training..cSsg|]}|jjqSr0)rrrr0r0r1r s)rworker_handlesworker_node_ids) rrrrrr>rrrrr) RunStatusEnum) run_idrun_namejob_idcontroller_actor_idrrS start_time_ms run_status resourcescSst}|dSrl)rrsessionr0r0r1 train_asyncFs z3BackendExecutor.start_training..train_async)!rrrTrSr configurerrrrrrQrIon_training_startrGget_with_failure_handlingr] ray.train._internal.state.schemarrruntime_contextget_runtime_contextrregister_train_runrexperiment_name get_job_id get_actor_idr\RUNNINGrErJ execute_async)r^rrrrrrractorsrrrrrindexr core_contextrr0rr1start_trainingsn  $      zBackendExecutor.start_trainingcCsRdd}|j|}||}tdd|Dr'tdd|Ds%tddS|S)azFetches the next ``_TrainingResult`` from each worker. Each ``_TrainingResult`` is expected to correspond to the same step from each worker (e.g. the same call to ``train.report()``). Returns: A list of ``_TrainingResult``s or ``None`` if there are no more results since the training function has exited on all workers. cSs.td}z|}W|Stytdw)Nget_next_resultszs`get_next_results` has been called before `start_training`. Please call `start_training` before `get_next_results`.) _get_sessionget_next RuntimeErrorr*)rresultr0r0r1rWs  z2BackendExecutor.get_next_results..get_nextcs|]}|duVqdSrlr0rrr0r0r1 kz3BackendExecutor.get_next_results..csr rlr0r r0r0r1r mrzSome workers returned results while others didn't. Make sure that `session.report()` are called the same number of times on all workers.N)rSrranyallr)r^rrresultsr0r0r1rLs  z BackendExecutor.get_next_resultscCs"dd}|j|}||dS)zDisable workers from enqueuing results from ``session.report()``. Note: Already reported results may still be enqueued at this point, and should be handled appropriately. cSstd}|S)Npause_reporting)rrrr0r0r1pause_session_reportingsz@BackendExecutor.pause_reporting..pause_session_reportingNrSrr)r^rrr0r0r1rzs zBackendExecutor.pause_reportingcCs"dd}|j|}||}|S)atFinish training and return final results. Propagate any exceptions. Blocks until training is finished on all workers. Assumes `start_training` has already been called. Returns: A list of return values from calling ``train_func`` on each worker. Each item corresponds to the return value from a single worker. cSs&td}z |}Wt|Stw)Nfinish_training)rfinishr)routputr0r0r1 end_trainings  z5BackendExecutor.finish_training..end_trainingr)r^rrrr0r0r1rs zBackendExecutor.finish_trainingFerrored failed_rank stack_tracecCs|jrEddlm}m}|r,|j}d}|dur|d|d7}|dur+||| d7}n|j}d}|jj|jj ||t t dddSdS) zJReport the final train run status, error, and end time to TrainStateActor.r)MAX_ERROR_STACK_TRACE_LENGTHrNzRank z worker raised an error. rD)rr status_detail end_time_ms) r]rrrERROREDFINISHEDr end_train_runrQrrZr[)r^rrrrrrrr0r0r1report_final_run_statuss& z'BackendExecutor.report_final_run_statuscCs>t|\}}|r t|S||_|td|t)aGets the remote values while handling for worker failures. This method should be called instead of ``ray.get()`` directly in order to handle worker failures. If a worker failure is identified, backend specific failure handling is executed and a ``TrainingWorkerError`` is raised. Args: remote_values: List of object refs representing functions that may fail in the middle of execution. For example, running a Train training loop in multiple parallel actor calls. Returns: The resolved objects represented by the passed in ObjectRefs. zjFailure identified during training. Restarting all workers and continuing training from latest checkpoint.) rrr{rNrrrrr3)r^ remote_valuessuccessrr0r0r1rs  z)BackendExecutor.get_with_failure_handlingTgraceful_terminationcCs~|rz |j|j|jWntytdYnw|r$|jn|jjddt|_|j r:t |j d|_ d|_ dS)zShuts down the workers in the worker group. Args: graceful_termination: If set to True, attempt to clean up the backend before terminating the Ray actors. zXGraceful shutdown of backend failed. This is expected if one of the workers has crashed.r) patience_sN) rI on_shutdownrSrGrrrshutdownrRrPr'rT)r^r&r0r0r1r)s     zBackendExecutor.shutdowncCst|jt Srl) isinstancerSrRrr0r0r1 is_startedrozBackendExecutor.is_startedcCsF|j|jdur|j}nd}|jrt|jd|_|j|ddS)N)r`)rSr)rOrPr'r)r^r`r0r0r1rs   zBackendExecutor._restartcCsV|jd7_|j|jkr)|j}d|_|jdkr'td|jd}|d||dS)Nr;rzTraining has failed after z attempts.)rMrKrNrwith_traceback)r^failurerr0r0r1r s    z#BackendExecutor._increment_failurescC|jSrl)rSrr0r0r1get_worker_groupz BackendExecutor.get_worker_groupcCr.rl)rMrr0r0r1_get_num_failuresr0z!BackendExecutor._get_num_failures)Nr;Nr<)NNNNrl)FNN)T)*r,r-r.r/rrrrZrr8rLr_rr r rrsr|r~r}rrr(rrrrrrrrrrryr#rr)r+rrr/r1r0r0r0r1r:Is 9 T45Y    .    r:c@r))InactiveWorkerGroupErrorz0Raised when underlying worker group is inactive.Nr+r0r0r0r1r2 r2r2c@s,eZdZddZddZddZddZd S) rRcCst|Srl)varsrr0r0r1 __getstate__)sz InactiveWorkerGroup.__getstate__cCst||dSrl)r3r)r^stater0r0r1 __setstate__,sz InactiveWorkerGroup.__setstate__cCtrlr2)r^namer0r0r1 __getattr__/r0zInactiveWorkerGroup.__getattr__cCr7rlr8rr0r0r1__len__2r0zInactiveWorkerGroup.__len__N)r,r-r.r4r6r:r;r0r0r0r1rR$s  rR method_namecCs$t}|std|d|d|S)N`zP` has been called before `start_training`. Please call `start_training` before `z`.)rr*)r<rr0r0r1r6sr)Kloggingrr[ collectionsr dataclassesrtypingrrrrrr r r rray._private.ray_constantsrrU!ray._private.accelerators.amd_gpur ray._private.accelerators.neuronr ray._private.accelerators.npur$ray._private.accelerators.nvidia_gpurrrwrray.exceptionsr ray.trainrrray.train._internal.sessionrrrrrray.train._internal.storagerray.train._internal.utilsr ray.train._internal.worker_grouprray.train.backendrray.train.constantsrrr r!r"r#r$r%ray.util.placement_groupr&r'r( getLoggerr,r Exceptionr*r3r4r:r2rRr8rr0r0r0r1sL  (           (  \