&`iwTddlZddlZddlZddlmZddlmZddlmZm Z m Z m Z m Z m Z mZmZddlZddlmcmZddlmZddlmZddlmZddlmZdd 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;mej?e@ZAGddeBZCGddeBZDeGddZEGddZFGddeBZGGdd ZHd!eIfd"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_groupTceZdZdZdS)TrainBackendErrorz?Errors with BackendExecutor that should not be exposed to user.N__name__ __module__ __qualname____doc__x/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/train/_internal/backend_executor.pyr+r+.sIIIIr2r+ceZdZdZdS)TrainingWorkerErrorz)Raised if a worker fails during training.Nr,r1r2r3r5r52s3333r2r5c2eZdZUdZeed<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/r0str__annotations__r1r2r3r7r76sB  %(((( ######r2r7ceZdZdZ d2dedeededeee e fd ef d Z d3d ee gdfd ee d eedeefdZdZdZde de fdZde de fdZdeefdZ d4de gefdee efdee efdededeeddfd Zdeeefd!Zd"Zd#Z d5d%e!d&eed'ee fd(Z"d)Z#d6d+e!fd,Z$d-Z%d.Z&d/Z'd0Z(d1Z)dS)7BackendExecutoraMain 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_retriesc| ddi|_n||_||_||_||_||_|jdkrtd|_d|_d|_ d|_ d|_ ||_ t|_d|_t!t"jt&t(t!t"jt,t.t!t"jt2t4g|_t9t;jdz|_t?t@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_shardsr7 ray_constants NEURON_CORESr"rNPUr#rGPUr!r _resource_configsinttime_start_time_msrr$state_tracking_enabled)selfrArBrCrDrEs r3__init__zBackendExecutor.__init__[sD '*/D & &)=)B)B)D)DD &-&2244 '(   ! !!&uD !$(! $%/11" *9/   !71   !4+  " *"$)++"455&12QST&U&U###r2initialization_hook train_clstrain_cls_argstrain_cls_kwargsc ||jpd}t|j|j|||||_|jr |jjnd}|j| |r!||_ |j |ddl m d f fd }|j | ttt |jj}|jdddkr|r||jD]B} || j| jr || j| jC|j|j|jn#t:$rs} t<tA| t<!d |"|#Yd} ~ nd} ~ wwxYw|j$r'dd l%m&} dd l'm(} | | |_)dSdS) zStarts the worker group.default)rCrD actor_clsactor_cls_argsactor_cls_kwargsplacement_groupNr) DataContextctxc2|dSN) _set_current)rpros r3_set_driver_dataset_contextz:BackendExecutor.start.._set_driver_dataset_contexts((-----r2r]zXFailure occurred during startup. Restarting all workers and attempting to startup again.)TrainRunStateManager)get_state_actor) state_actor)*_create_placement_grouprUrrOrJrXrVdriver_node_id"sort_workers_by_node_id_and_gpu_idrTexecuteray.dataro get_currentboolrr rNshare_cuda_visible_devicesget_share_cuda_visible_devicesr^_is_share_resources_enabledr8r9_share_resource_idsr:on_startrLrlogger exceptionr;warning_increment_failures_restartrbray.train._internal.stateru%ray.train._internal.state.state_actorrv state_manager)rcrerfrgrhrntrial_driver_node_idrt"share_cuda_visible_devices_enabledresource_configexcrurvros @r3startzBackendExecutor.starts $$&&&/<9')!%!;)-+    "04/? ID  + +T  <<=QRRR- " ?,?)!))*=>>> - , , , , , . . . . . . .   % %+''))    269M<22 .*..ua881<<6=00222#'#9  33#1#C,,'5'B M " "4#4d6J K K K K      SXX & & & NN/     $ $ & & & MMOOOOOOOO   & U F F F F F F M M M M M M!5!5//BSBS!T!T!TD     U Us3D.F"" H,A)HHct}tjjj}|j}|dup| }|r#fdt jD}tttd}|rdnd}tj ||}tdttd} tj|g| \} } | rtd n9t%d tj|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. NcBg|]}jSr1)rJrK).0_rcs r3 z;BackendExecutor._create_placement_group..s567*//11r2rSPREADPACK)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_grouprangerOr~rr%utilrnrdebugr&waitready TimeoutErrorformatavailable_resources bundle_specsrU) rccurrent_placement_grouprrshould_create_placement_groupbundles use_spreadrrnrrrs ` r3rxz'BackendExecutor._create_placement_groups~#>"?"?$2  @ 6 $t + A@@ & ) 4;@AR;S;SGk*H!LLMMJ#-9xx6H!h66w6RRO LL@ A A A!"EsKKGx!6!6!8!8 97KKKHE1  ;<<<<"*+1&/11?3O++   %4D ! ! !7 4 4r2cP|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)rrZr]rrcs r3rz+BackendExecutor._share_cuda_visible_devicess$0   !24PQQQQQr2resourceenv_varcN fd|jjD}tt}tt}t |D]>\}\}}||||||?g} |D]h\}}t|}d | fd} ||D]0}| |j || 1itj | 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. cLg|] }|jj|jjf!Sr1)metadatanode_id resource_ids)rwrs r3rz7BackendExecutor._share_resource_ids..MsB% % %  " '1 % % % r2,c&tj<dSrr)osenviron)all_resource_idsrsr3set_resource_idsz=BackendExecutor._share_resource_ids..set_resource_ids`s&6 7###r2N)rXworkersrset enumerateaddupdateitemssortedjoinappendexecute_single_asyncrr) rcrrnode_ids_and_resource_idsnode_id_to_worker_idnode_id_to_resource_ids worker_idrrfuturesrrs `` @r3rz#BackendExecutor._share_resource_ids4st2% % % % &. % % % ! +3//"-c"2"22; 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? cg|]C\}}d|jjd|jjd|jjd|d|d| DS)z - (node_id=z, ip=z, pid=z ) world_rank=z , local_rank=z , node_rank=)rrnode_ippid)rirlocal_rank_map node_rank_maps r3rzDBackendExecutor._create_rank_world_size_mappings..s   AqPaj0PPqz7IPPz~PP45PP,Q/PP=J1=MPP   r2z'Started distributed worker processes: ) rr_rlenrXrrrrrrinfo) rclocal_world_size_mapnode_idsnode_cnt node_id_dict world_rankrr workers_inforrs @@r3 _create_rank_world_size_mappingsz0BackendExecutor._create_rank_world_size_mappingszsZ! "    D$5 6 677 : :J&.z:Fo-G)5g)>N: &  ! ! !Q & ! ! !h&&$,!A (0(9M* % %D$5 6 677 E EJ&.z:Fo-G/;G/D  , ,yy     &d&7&?@@        M|MMNNN3]BBr2 train_funcdatasetsr data_configstorage checkpointcNttdfd}|j\d|jjD}d|jjD} ||t |j|| |_|\} } } g} tt |jD]p}| |j |||| || || |t |j|j ||j|||| q|j |j|j|| |jrddlm}t&j}|j|j j|j j||||j|j|j|jg|jz 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 z t||||||||| | |  dS#t$rtdwxYw)N) training_funcr local_rank node_ranklocal_world_size world_sizerB 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+) rrrrrrrBrrrruse_detailed_autofilled_metricss r3initialize_sessionz:BackendExecutor.start_training..initialize_sessions ",))'%5))"/%)0O#         '6 s :Ncg|] }|j Sr1)actorrrs r3rz2BackendExecutor.start_training.. sKKKvflKKKr2c&g|]}|jjSr1)rrrs r3rz2BackendExecutor.start_training.. sXXXF/XXXr2)rworker_handlesworker_node_ids) rrrrrrBrrrrr) RunStatusEnum) run_idrun_namejob_idcontroller_actor_idrrX start_time_ms run_status resourcescJt}|dSrr)rrsessions r3 train_asyncz3BackendExecutor.start_training..train_asyncFs!mmG MMOOOOOr2)!rrrYrXr configurerrrrrrVrNon_training_startrLget_with_failure_handlingrb ray.train._internal.state.schemarrruntime_contextget_runtime_contextrregister_train_runrexperiment_name get_job_id get_actor_idraRUNNINGrJrO execute_async)rcrrrrrrractorsrrrrrindexr core_contextrrs @r3start_trainingzBackendExecutor.start_trainingsz,+6 2A+ + ' " " " " " H   &KK1B1JKKKFXXd>O>WXXXH"-"7"7t011% ( #8##D   1 1 3 3   3t01122  E NN!66&$-e4+E2%9%%@"4#455#/)"&"5e"<%)#7    $ ''(94;OPPP &&w///  &  F F F F F F.BBDDL   1 1'.)9#..00$0$=$=$?$?!!."1(0569JJ 2       '' 44444r2cd}|j|}||}td|Dr*t d|Dst ddS|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. ctd} |}n#t$rtdwxYw|S)Nget_next_resultszs`get_next_results` has been called before `start_training`. Please call `start_training` before `get_next_results`.) _get_sessionget_next RuntimeErrorr+)rresults r3rz2BackendExecutor.get_next_results..get_nextWsa"#566G  ))++   '* Ms &Ac3K|]}|duV dSrrr1rrs r3 z3BackendExecutor.get_next_results..ks&**QqDy******r2c3K|]}|duV dSrrr1rs r3r z3BackendExecutor.get_next_results..ms&22QqDy222222r2zSome workers returned results while others didn't. Make sure that `session.report()` are called the same number of times on all workers.N)rXrranyallr)rcrrresultss r3rz BackendExecutor.get_next_resultsLs    #11(;;0099 **'*** * * 22'22222 ";tr2cjd}|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. cHtd}|S)Npause_reporting)rr'rs r3pause_session_reportingz@BackendExecutor.pause_reporting..pause_session_reportings""#455G**,, ,r2NrXrr)rcr(rs r3r'zBackendExecutor.pause_reportingzsD - - -#112IJJ &&w/////r2cjd}|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. ctd} |}tn#twxYw|S)Nfinish_training)rfinishr)routputs r3 end_trainingz5BackendExecutor.finish_training..end_trainingsM"#455G # ))!"""" """"Ms 4Ar))rcr/rr$s r3r,zBackendExecutor.finish_trainingsA   #11,??0099r2Ferrored failed_rank stack_tracec |jrddlm}m}|r%|j}d}| |d|dz }|||| dz }n |j}d}|j|jj ||ttj dzdSdS) zJReport the final train run status, error, and end time to TrainStateActor.r)MAX_ERROR_STACK_TRACE_LENGTHrNzRank z worker raised an error. rI)rr status_detail end_time_ms) rbr r4rERROREDFINISHEDr end_train_runrVrr_r`)rcr0r1r2r4rrr6s r3report_final_run_statusz'BackendExecutor.report_final_run_statuss  &           #*2 " *!%U[%U%U%UUM*![2N1N1O1O%PPM*3 "   , ,'.%+ d 233 -     %  r2ct|\}}|rtj|S||_|t d|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.) rrrrSrrrrr5)rc remote_valuessuccessrs r3rz)BackendExecutor.get_with_failure_handlingsx /}==  &7=)) )!*D   $ $ & & & NN>    MMOOO% %r2Tgraceful_terminationc|rQ |j|j|jn*#t$rt dYnwxYw|r|jn|jdt|_|j rt|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) rN on_shutdownrXrLrrrshutdownrWrUr(rY)rcr?s r3rCzBackendExecutor.shutdowns    ))$*;T=QRRRR    B   5   & & ( ( ( (   & &! & 4 4 4/11   ) "4#8 9 9 9$(D !"s%*$AAc8t|jt Srr) isinstancerXrWrs r3 is_startedzBackendExecutor.is_startedsd/1DEEEEr2c|j|j|j}nd}|jrt |jd|_||dS)N)re)rXrCrTrUr(r)rcres r3rzBackendExecutor._restartsq ""$$$  $ 0"&";  "&    ) "4#8 9 9 9$(D ! ': ;;;;;r2c|xjdz c_|j|jkrI|j}d|_|jdkr.td|jd}|d||dS)Nr?rzTraining has failed after z attempts.)rRrPrSrwith_traceback)rcfailurers r3rz#BackendExecutor._increment_failures s a  !3 3 3(G!%D !A%%"O1COOO((..G;  4 3r2c|jSrr)rXrs r3get_worker_groupz BackendExecutor.get_worker_groups   r2c|jSrr)rRrs r3_get_num_failuresz!BackendExecutor._get_num_failuress !!r2)Nr?Nr@)NNNNrr)FNN)T)*r-r.r/r0rr rr_rr;rQrdrr r rrxrrrrrr)rrrrrrrrr'r,r~r;rrCrFrrrLrNr1r2r3r>r>Is7*+/;?5V5V%5VY' 5V  5V 'tCJ'78 5V5V5V5V5Vr=A$(*.+/ RURU%hr4x&89RUD>RU! RU #4. RURURURUh242424hRRR43C3#3333j  RU    "PC$t*PCPCPCPCr,0~5~5RUO~5sG|$~5sCx. ~5  ~5  ~5Z(~5 ~5~5~5~5@,(4+@"A,,,,\ 0 0 0<%)%) c]c] @&&&:##T####:FFF < < <   !!!"""""r2r>ceZdZdZdS)InactiveWorkerGroupErrorz0Raised when underlying worker group is inactive.Nr,r1r2r3rPrP s::::r2rPc&eZdZdZdZdZdZdS)rWc t|Srr)varsrs r3 __getstate__z InactiveWorkerGroup.__getstate__)sDzzr2cJt||dSrr)rSr)rcstates r3 __setstate__z InactiveWorkerGroup.__setstate__,s" T %     r2ctrrrP)rcnames r3 __getattr__zInactiveWorkerGroup.__getattr__/&(((r2ctrrrYrs r3__len__zInactiveWorkerGroup.__len__2r\r2N)r-r.r/rTrWr[r^r1r2r3rWrW$sP !!!))))))))r2rW method_namecRt}|std|d|d|S)N`zP` has been called before `start_training`. Please call `start_training` before `z`.)rr+)r_rs r3rr6sQmmG           Nr2)Kloggingrr` collectionsr dataclassesrtypingrrrrr r r r rray._private.ray_constantsrrZ!ray._private.accelerators.amd_gpur ray._private.accelerators.neuronrray._private.accelerators.npur$ray._private.accelerators.nvidia_gpurrr|rray.exceptionsr ray.trainrrray.train._internal.sessionrrrrrray.train._internal.storagerray.train._internal.utilsr ray.train._internal.worker_grouprray.train.backendrray.train.constantsrr r!r"r#r$r%r&ray.util.placement_groupr'r(r) getLoggerr-r Exceptionr+r5r7r>rPrWr;rr1r2r3rvs ######!!!!!!LLLLLLLLLLLLLLLLLLLL 222222222IIIIIILLLLLLKKKKKKMMMMMM222222((((((,,,,,,,,766666777777888888++++++                    YXXXXXXX GCLL  8 $ $JJJJJ JJJ44444)444 $$$$$$$ $$T "T "T "T "T "T "T "T "n;;;;;y;;;))))))))$ c      r2