&`iR2ddlZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z m Z mZddlZddlmcmZddlmZddlmZmZddlmZddlmZddlmZdd lmZdd l m!Z!dd l"m#Z#dd l$m%Z%dd l&m'Z'm(Z(m)Z)ddl*m+Z+ddl,m-Z-m.Z.ddl/Z/ ej0Z0n#e1$r ej2Z0YnwxYwej3dkrP ddl4Z4ddl5Z5ddl6Z6nB#e7e8f$r8Z9dZ4dZ5dZ6ej:e;ZdS)N)FIRST_COMPLETED)AnyDictListOptional)CoreContextFilter) JSONFormatter TextFormatter) build_address)'NOSET_ASCEND_RT_VISIBLE_DEVICES_ENV_VAR)"NOSET_CUDA_VISIBLE_DEVICES_ENV_VAR)RAY_JOB_CONFIG_JSON_ENV_VAR)"remove_ray_internal_flags_from_env) GcsClient) ActorHandle)JOB_ID_METADATA_KEYJOB_NAME_METADATA_KEYJobInfoStorageClient)JobLogStorageClient) JobErrorType JobStatuswin32z`Failed to Import win32api. For best usage experience run 'conda install pywin32'. Import error: cfeZdZdZdZdZddgZ d!deded eeefd ed ed e ef d Z d"dZ d#de deee ffdZdZdededejfdZde deeeffdZdejdefdZdeejfdZdeejdejfdZ d$de ede fdZd Z dS)% JobSupervisoram Ray actor created by JobManager for each submitted job, responsible to setup runtime_env, execute given shell command in subprocess, update job status, persist job logs and manage subprocess group cleaning. One job supervisor actor maps to one subprocess, for one job_id. Job supervisor actor should fate share with subprocess it created. g?SIGINTSIGTERMNjob_id entrypoint user_metadata gcs_addresscluster_id_hexlogs_dirc||_t||}t|||_t |_||_t|t|i|_ |j |tj |_ d|_tjt"d||_|dS)N)address cluster_idz .supervisor-)_job_idrr_job_info_clientr _log_client _entrypointrr _metadataupdateasyncioEvent _stop_event_win32_job_objectlogging getLogger__name___logger_configure_logger)selfrrr r!r"r# gcs_clients |/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/dashboard/modules/job/job_supervisor.py__init__zJobSupervisor.__init__Gs {~NNN 4Z J J.00%.v7LfU m,,,#=??"&(H)J)J&)J)JKK       returnctjtjjjd|jd}tj tj |d|j ttj}tj|}t#}t%jt$jdrt+}|||||j ||j |d|j _dS)zx Configure self._logger object to write logs to file based on job submission ID and to console. zjobs/supervisor-z.logT)exist_okFN)ospathjoinray_privateworker _global_nodeget_logs_dir_pathr'makedirsdirnamer4 addFilterrr1 StreamHandler FileHandlerr ray_constantsenv_boolRAY_BACKEND_LOG_JSON_ENV_VARr setFormatter addHandler propagate)r6supervisor_log_file_namestream_handler file_handler formatters r8r5zJobSupervisor._configure_loggeres1 $&7<< L  , > > @ @ 1t| 1 1 1$ $   BGOO$<==MMMM 022333 .00*+CDD !OO  !-"Le T T (%I##I...!!),,, ///  ---!& r:Fresources_specifiedc4ttjj}|r|S|di}|t |t|tj ||d<|S)aGet the runtime env that should be set in the job driver. Args: resources_specified: Whether the user specified resources (CPUs, GPUs, custom resources) in the submit_job request. If so, we will skip the workaround for GPU detection introduced in #24546, so that the behavior matches that of the user specifying resources for any other actor. Returns: The runtime env that should be set in the job driver. env_vars) dictrAget_runtime_context runtime_envgetpopr r rKRAY_WORKER_NICENESS)r6rUcurr_runtime_envrWs r8_get_driver_runtime_envz%JobSupervisor._get_driver_runtime_env{s   7 9 9 EFF  $# #$'' B77 7888 <=== ]6777'/$r:cdS)z&Used to check the health of the actor.Nr6s r8pingzJobSupervisor.pings r:env logs_pathc 4t|d5}|d|jd|jdt j|jdd|tj|tjdkr&tj ddkrd nd }tj }|j }tjdkrg tj|}n#t$r|cYcd d d SwxYwt jd |d |dtjtjntjdkrt"rt%jd d|_t%j|jt$j}t$j|dd<t%j|jt$j|t#jt4jt4jzd|} t%j|j| |cd d d S#1swxYwYd S)a Runs the entrypoint command as a child process, streaming stderr & stdout to given log files. Unix systems: Meanwhile we start a demon process and group driver subprocess in same pgid, such that if job actor dies, entire process group also fate share with it. Windows systems: A jobObject is created to enable fate sharing for the entire process group. Args: logs_path: File path on head node's local disk to store driver command's stdout & stderr. Returns: child_process: Child process that runs the driver command. Can be terminated or killed upon user calling stop(). azRunning entrypoint for job z:  TrRAY_JOB_STOP_SIGNALrcVtjtjtjhSN)signalpthread_sigmask SIG_UNBLOCKrrar:r8z0JobSupervisor._exec_entrypoint..s 6".!!r:N)shellstart_new_sessionstdoutstderrrd preexec_fnzwhile kill -s 0 z; do sleep 1; done; kill -9 -)rprrrsBasicLimitInformation LimitFlagsF)openwriter'r* subprocessPopenSTDOUTsysplatformr>environr[getpidpidgetpgidProcessLookupErrorDEVNULLwin32apiwin32jobCreateJobObjectr0QueryInformationJobObject!JobObjectExtendedLimitInformation"JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSESetInformationJobObject OpenProcesswin32conPROCESS_TERMINATEPROCESS_SET_QUOTAAssignProcessToJobObject) r6rdre logs_file child_process parent_pid child_pid child_pgidwin32_job_info child_handles r8_exec_entrypointzJobSupervisor._exec_entrypoints,)S ! !F !Y OORdlRRd>NRRR   ', "& !(|w.. '<==II  #M(J%)I|w&&)!#I!6!6JJ))))(((CF !F !F !F !F !F !F !F !>) \z\\PZ\\%-%- ((X( *2)A$)K)K&!)!C*H,V"" ?67 0*>" (3.1KK   1$2H,WWW MF !F !F !F !F !F !F !F !F !F !F !F !F !F !F !F !F !F !s7B1H CH  C3$H 2C33D H  HHcztjtjvrdtjtj<tjjdtjjj j j }|Jttj|||jdtj|ddiS)z?Returns environment variables that should be set in the driver.autoN)rZmetadataPYTHONUNBUFFERED1)rK RAY_ADDRESS_ENVIRONMENT_VARIABLEr>rrArBservices%canonicalize_bootstrap_address_or_dierCrD _ray_paramstemp_dirrjsondumpsr_r+)r6rUray_addrs r8_get_driver_env_varsz"JobSupervisor._get_driver_env_varss  9RZ G GIOBJ}E F<(NN CL'4@I  ### (#'#?#?@S#T#T $**  :H !  r:rcK|;|}||Stj|jd{V|9dSdSrk)pollr-sleepSUBPROCESS_POLL_PERIOD_S)r6r return_codes r8_pollingzJobSupervisor._pollingsd'',,..K&""mD$ABBBBBBBBB'''''r: processescK tj|d\}}t|dkrdStj|jd{VN)z'Poll processes until all are completed.Tr)timeoutN)psutil wait_procslenr-rr)r6r_alives r8 _poll_allzJobSupervisor._poll_all#sd C*9a@@@JQ5zzQmD$ABBBBBBBBB  Cr:sigcf|D]-} tj|j|#t$rY*wxYwdS)z>Ensure each process is already finished or send a kill signal.N)r>killrr)r6rrprocs r8_kill_processeszJobSupervisor._kill_processes,sW  D #&&&&%      s ! .._start_signal_actorc K|j|jd{V}|td|jd|j}|j}|t jkrtd|jd|t jkrtd|jd|d|d|r|j d{Vtj j jj}dt!|j|j}tj}|j|jt j||d  d{V t,j} t3| | |||jd | t<j|j !|j} |"| | } | j#} tI|%| } tMj | tI|j' gtP d{V\}}|j')r| *tVj,d kr#|j-rt]j/|j-dntVj,d krt,j0dd}||j1vr|j2|dd}tgj4| }|g|5dz}|6|totp| tst,j0d|j:}tI|;|}tMj<||d{V|jd|jd|dn^#tLj=$rL|j2d|jd|d|d|6|tpj>YnwxYw|j|jt j?d{Vnt|dks Jd|\}|A}|jd|jd||dkr3|j|jt jB|d{Vn{|j C|jd{V}||dkr d |d!|z}nd |d"}|j|jt jD||tjF#d{Vn#t$r|jHd$tjJ |j|jt jDtjJtjK%d{Vn>#t$r1|jHd&tjJYnwxYwYnwxYwtjLMdS#tjLMwxYw)'aA Stop and start both happen asynchronously, coordinated by asyncio event and coroutine, respectively. 1) Sets job status as running 2) Pass runtime env and metadata to subprocess as serialized env variables. 3) Handle concurrent events of driver execution and Nz&Status could not be retrieved for job .zJob zN is already in RUNNING state. JobSupervisor.run() should only be called once. z, is not in PENDING state. Current status is z with message zhttp://)driver_agent_http_addressdriver_node_id)jobinfo_replace_kwargsz"Submitting job with RAY_ADDRESS = ) return_whenrrirz7 not a valid stop signal. Terminating job with SIGTERM.T) recursiveRAY_JOB_STOP_WAIT_TIME_Sz% has been terminated gracefully with z$Attempt to gracefully terminate job z through z has timed out after z5 seconds. Job is now being force-killed with SIGKILL.z#Should have only one coroutine donez% entrypoint command exited with code r)driver_exit_coderuz-Job entrypoint command failed with exit code z3, last available logs (truncated to 20,000 chars): z. No logs available.)messager error_typezAGot unexpected exception while trying to execute driver command. )rrz2Failed to update job status to FAILED. Exception: )Nr(get_infor' RuntimeErrorstatusrrRUNNINGPENDINGwaitremoterArBrC global_workernoder node_ip_addressdashboard_agent_listen_portrY get_node_id put_statusr>rcopyrr,rr4inforKrr)get_log_file_pathrr create_taskrr-r/ris_setcancelr}r~r0rTerminateJobObjectr[VALID_STOP_SIGNALSwarningrProcesschildrenrgetattrrlint DEFAULT_RAY_JOB_STOP_WAIT_TIME_Srwait_for TimeoutErrorSIGKILLSTOPPEDrresult SUCCEEDEDget_last_n_log_linesFAILEDrJOB_ENTRYPOINT_COMMAND_ERROR Exceptionerror traceback format_exc"JOB_ENTRYPOINT_COMMAND_START_ERRORactor exit_actor)r6rrU curr_info curr_status curr_messagerrrrdlog_pathrr polling_taskfinishedr stop_signal job_process proc_to_killstop_job_wait_timepoll_job_stop_taskchild_process_taskrlog_tailrs r8runzJobSupervisor.run5s/88FFFFFFFF  W WWWXX X&  ( )+ + +Dt|DDD  )+ + +Pt|PP%0PP@LPPP   4%*1133 3 3 3 3 3 3 3|"05$umD>@@#.. L  -F"0$$/         z #*//##C /s 3 3 3 JJt001DEE F F F L  I}EFII   '99$,GGH 11#x@@M%)I&t}}]'C'CDDL ' {4+;+@+@+B+BCCD+!!!KHa &&((N ##%%%<7**t/E*/0FKKKK\W,,"$*..1F "R"RK"$*AAA ,,*000'0 "(.";";K$/=;3G3GRV3G3W3W#WL((wv{7S7STTTK-0JNN : $ E..* .9 9U9U-V-V*%./ACUVVVVVVVVV ))34<33$/333#/KKK ,,94<99'2991999 ,,\6>JJJJJK+66t|YEVWWWWWWWWWW8}})))+P)))'/$#07799  !!64<66(366!##/:: !+)4; &*%5%J%J4<%X%XXXXXXXH+BQ5@QQQ'' W5@WWW /:: !( ')4#/#L ;    L  5%02255    +66L$%022+N 7     "";"+"6"8"8;;  & I " " " " "CI " " " "swH;W BP,+W ,ARW REW Z7 8ZAYZ8Z Z ZZZ7ZZ77 [c8|jdS)zCSet step_event and let run() handle the rest in its asyncio.wait().N)r/setrbs r8stopzJobSupervisor.stops r:rk)r;N)F)NF)!r3 __module__ __qualname____doc__rrrstrrrr9r5boolrr_rcrXrzr{rrrrrrrrrlSignalsrrrrrar:r8rr9s!()$""I.#'!!!!CH~ !  !  !3-!!!!<''''.+0  #' c3h    8   \!D\!S\!Z=M\!\!\!\!|  c3h    @CJ,<CCCCCCfn)=CCCCfn)=FN6:$) j#j#&k2j#" j#j#j#j#Xr:r)?r-rr1r>rlrzr}r asyncio.tasksrtypingrrrrrAray._private.ray_constantsrBrKray._common.filtersrray._common.formattersr r ray._common.network_utilsr ray._private.accelerators.npur $ray._private.accelerators.nvidia_gpur "ray._private.runtime_env.constantsrray._private.utilsr ray._rayletr ray.actorr ray.dashboard.modules.job.commonrrr0ray.dashboard.modules.job.job_log_storage_clientrray.job_submissionrrrrAttributeError ensure_futurer~rrrModuleNotFoundError ImportErrorer2r3loggerrrrar:r8r s   )))))),,,,,,,,,,,, 222222222111111????????333333QQQQQQSSSSSSJJJJJJAAAAAA!!!!!!!!!!!! QPPPPP66666666 (%KK((('KKK(<7    -    ""8,, :67 : :         jjjjjjjjjjs*B B,+B,; CD.DD