AxjK7dZddlZddlZddlZddlZddlmZmZmZddl m Z m Z m Z ddl mZmZmZmZddlmZmZmZdd lmZmZeZd Zd ed efd ZGddZdS)zh runpod | serverless | rp_scale.py Provides the functionality for scaling the runpod serverless worker. N)AnyDictSet)AsyncClientSession ClientSessionTooManyRequests) _job_stop_urlget_jobget_stop_signals handle_job) RunPodLogger_reset_batch_id _set_batch_id) JobsProgress IS_LOCAL_TESTcltj|||}td|dS)NzUncaught exception | ) tracebackformat_exceptionlogerror)exc_type exc_value exc_tracebackexcs SC:\Projects\Historical\venv\Lib\site-packages\runpod/serverless/modules/rp_scale.py_handle_uncaught_exceptionrs7  $Xy- H HCII+c++,,,,,current_concurrencyreturnc|S)z Default concurrency modifier. This function returns the current concurrency without any modification. Args: current_concurrency (int): The current concurrency. Returns: int: The current concurrency. )r s r_default_concurrency_modifierr$s  rceZdZdZdeeeffdZdZdZ dZ dZ dZ d Z d efd Zd efd Zd efdZd efdZded efdZd edefdZdS) JobScalerz^ Job Scaler. This class is responsible for scaling the number of concurrent requests. configctj|_d|_||_t |_i|_t|_ d|_ tj |j|_ t|_t|_d|_t$|_|dx}r||_t*sdS|jdx}r||_|jdx}r||_|jdx}r||_|jdx}r||_ |jd x}r ||_ dSdS) Nr Zmaxsizeconcurrency_modifier jobs_fetcherjobs_fetcher_timeout jobs_handlerstop_signals_fetcherstop_signals_fetcher_timeout)asyncioEvent_shutdown_eventr r'r job_progress jobs_tasksr r0r1Queue jobs_queuer$r,r r-r.rr/getr)selfr'r,r-r.r/r0r1s r__init__zJobScaler.__init__-ss&}#$  (NN46$4!,.)!-0HIII$A!#$&!&#)::.D#E#E E  =(:: :< - ,D #';??3I#J#J J  =(:: :< - ,D #';??3I#J#J J  =( ? ? ? M&-)= > > > > N N N HHL M M M M M N  DHHJJsAA$BBcjtd|d|dS)a Called when the worker is signalled to shut down. This function is called when the worker receives a signal to shut down, such as SIGTERM or SIGINT. It sets the shutdown event, which will cause the worker to exit its main loop and shut down gracefully. Args: signum: The signal number that was received. frame: The current stack frame. zReceived shutdown signal: .N)rr? kill_worker)r:signumframes rrGzJobScaler.handle_shutdown{s: 8v888999 rcKt4d{V}tj||}tj||}tj||}|||g}tj|d{Vdddd{VdS#1d{VswxYwYdS)N)rr2 create_taskget_jobsrun_jobsmonitor_stop_signalsgather)r:session jobtake_task jobrun_task jobstop_tasktaskss rrKz JobScaler.runsP%'' ) ) ) ) ) ) )7".t}}W/E/EFFL!-dmmG.D.DEEK".t/H/H/Q/QRRL!; =E.%( ( ( ( ( ( ( ( ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) )sBB99 CCc6|j S)z< Return whether the worker is alive or not. )r4is_setr@s ris_alivezJobScaler.is_alives'..0000rcltd|jdS)z- Whether to kill the worker. z Kill worker.N)rr?r4setr@s rrOzJobScaler.kill_workers1 .!!!   """""rr!c|j}|j}td|jd|d|||zS)Nz JobScaler.status | concurrency: z ; queue: z ; progress: )r8qsizer5 get_job_countrr?r )r:current_queue_countcurrent_progress_counts rr=zJobScaler.current_occupancys"o3355!%!2!@!@!B!B  Lt/G L LRe L LtJ L L   &(;;;rrXc &K|r|d{V|j|z }|dkr5tdt jdd{V tdt j| |||j d{V}|s7td t jdd{V|D]]}|j |d{V|j |td|d ^td |j n#t"$r7td t jd d{VYnt j$rtd t j$rtdYnt($r(}td|dYd}~nYd}~wt*$rI}tdt/|jdt3|Yd}~nd}~wwxYwt jdd{Vn#t jdd{VwxYw|dSdS)z Retrieve multiple jobs from the server in batches using blocking requests. Runs the block in an infinite loop while the worker is alive. Adds jobs to the JobsQueue Nrz2JobScaler.get_jobs | Queue is full. Retrying soon.r z.JobScaler.get_jobs | Starting job acquisition.timeoutz&JobScaler.get_jobs | No jobs acquired.z Job QueuedidzJobs in queue: z?JobScaler.get_jobs | Too many requests. Debounce for 5 seconds.z+JobScaler.get_jobs | Request was cancelled.z9JobScaler.get_jobs | Job acquisition timed out. Retrying.z'JobScaler.get_jobs | Unexpected error: rNz!Failed to get job. | Error Type:  | Error Message: )r_rAr r=rr?r2r>wait_forr-r.r8putr5addinforcr CancelledError TimeoutError TypeError Exceptionrtype__name__str)r:rX jobs_needed acquired_jobsjobrs rrTzJobScaler.get_jobssmmoo/ '.."" " " " " " " "2T5K5K5M5MMKa NOOOmA&&&&&&&&&& ' JKKK'.&6%%g{;; 5'''!!!!!! %IIFGGG6mA&&&&&&&&&&3)77C/--c222222222%))#...IIlCI6666D4?+@+@+B+BDDEEEE" ' ' ' UmA&&&&&&&&&&&)    GHHH' W W W UVVVVV N N N LELLLMMMMMMMM    lU 8Lll`cdi`j`jll  mA&&&&&&&&&&gmA&&&&&&&&&&_mmoo/ '/ '/ '/ '/ 'sc A+F(BF(&K(>J=&K(AJ=9K; J=I'"K' J=4?J83K8J==KK7cpKt}d}|s|jst ||jkr|js|jd{V}tj| ||}| |||j |d<t ||jkr|j|rtt |}||krt d||}tj|tjdd{V\}}||ntjdd{V|s|jtj|ddid{V} | D]N} t'| t(r7t'| tjstd | OdS) z Retrieve jobs from the jobs queue and process them concurrently. Runs the block in an infinite loop while the worker is alive or jobs queue is not empty. rNrjzJobs in progress: g?) return_whenrireturn_exceptionsTz8JobScaler.run_jobs | Task failed during shutdown drain: )rar_r8emptylenr r9r2rSrror6rrpwaitFIRST_COMPLETEDdifference_updater>rW isinstancertrqr) r:rXr\last_task_countrztaskcurrent_task_countdonependingresultsresults rrUzJobScaler.run_jobss_ $'55mmoo )T_%:%:%<%< )e**t777@U@U@W@W7 O//11111111*4??7C+H+HII $-1D * e**t777@U@U@W@W7 )%(ZZ"%88HHF2DFFGGG&8O&-lw'>'''!!!!!! g ''----mC(((((((((3mmoo )T_%:%:%<%< )> FFFFFFFFF _ _F&),, _ZH^5_5_ _ ]U[]]^^^ _ _rc K|jtur*ttddS|r t j|||jd{V}|D]}| |d{V|st j dd{Vn#t$rt j dd{VYnt j $rt dt j$rt dYnot$rc}tdt#|jd t'|t j dd{VYd}~nd}~wwxYwt j d d{Vn#t j d d{VwxYw|dSdS) a` Long-polls the dedicated stop channel and stops signalled jobs. Runs in an infinite loop while the worker is alive. The Runpod server signals a request to be stopped (for example when it is cancelled or times out) and this loop stops just that in-progress job, leaving the worker's other jobs running. NzJobScaler.monitor_stop_signals | Stop channel could not be derived from the job-take URL; per-job stop is disabled for this worker.rhr rkz7JobScaler.monitor_stop_signals | Request was cancelled.z?JobScaler.monitor_stop_signals | Stop poll timed out. Retrying.z-JobScaler.monitor_stop_signals | Error Type: rlr)r0r r rrJr_r2rmr1stop_jobr>r rqr?rrrtrrurvrw)r:rXjob_idsjob_idrs rrVzJobScaler.monitor_stop_signalss  $(8 8 8]__=T HHS    Fmmoo ' '!( 0--g66 =!!!&00F--//////////+"-*********" ' ' 'mA&&&&&&&&&&&)    STTT' ] ] ] [\\\\\ ' ' ' xDKKDXxxlopulvlvxxmA&&&&&&&&&&&&&&  ' mA&&&&&&&&&&gmA&&&&&&&&&&9mmoo ' ' ' ' 'sKA0CG$F'&G(AF'9G; F'AF"G"F''GG!rcK|j|}| td|ddStd||dS)a Stop a single in-progress job by cancelling its running task. Args: job_id: The id of the job to stop. Returns: True if a matching in-progress job was found and stopped, False otherwise. Nz,JobScaler.stop_job | No in-progress job for rNFz Stopping job.T)r6r9rr?rpcancel)r:rrs rrzJobScaler.stop_jobAsk""6** < IINVNNN O O O5 &))) trrzcKt|d} td|d|||j|d{V|jddr|ng#tj$r#t d|dt$r*}t d||dd}~wwxYw|j |j||j|ddtd |dt%|dS#|j |j||j|ddtd |dt%|wxYw) za Process an individual job. This function is run concurrently for multiple jobs. batchIdz Handling JobrjNrefresh_workerFz Job stopped.zError handling job: z Finished Job)rr9rr?r/r'rOr2rqrprtrr8 task_doner5remover6popr)r:rXrzbatch_id_tokenerrs rrzJobScaler.handle_jobUs'swwy'9'9:: , IInc$i 0 0 0##GT[#>> > > > > > > >{/77 #  """%    HH^SY / / /     II2S22CI > > >   O % % ' ' '   $ $S ) ) ) O  D 4 0 0 0 IInc$i 0 0 0 N + + + + + O % % ' ' '   $ $S ) ) ) O  D 4 0 0 0 IInc$i 0 0 0 N + + + +s+A2BF:C=%C88C==FBH N)rv __module__ __qualname____doc__rrwrr;rArLrGrKr_rOintr=rrTrUrVboolrdictrr#rrr&r&(sY(MtCH~(M(M(M(MT   "   &    ) ) )111 ###<3<<<<7'm7'7'7'7'r+_m+_+_+_+_Z,'-,',',','\ST(, ,D,,,,,,rr&)rr2rErCrtypingrrr http_clientrrr rp_jobr r r r rp_loggerrrr worker_staterrrrrr$r&r#rrrsF  !!!!!!!!!!MMMMMMMMMMHHHHHHHHHHHHCCCCCCCCCC55555555lnn--- s s    K,K,K,K,K,K,K,K,K,K,r