tni%dZddlZddlZddlZddlZddlmZmZddlm Z m Z m Z ddl m Z mZddlmZdd lmZmZeZd Zd ed efd ZGddZdS)zh runpod | serverless | rp_scale.py Provides the functionality for scaling the runpod serverless worker. N)AnyDict)AsyncClientSession ClientSessionTooManyRequests)get_job handle_job) RunPodLogger) JobsProgress IS_LOCAL_TESTcltj|||}td|dS)NzUncaught exception | ) tracebackformat_exceptionlogerror)exc_type exc_value exc_tracebackexcs _/var/www/html/gpu-tools/venv/lib/python3.11/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. )rs r_default_concurrency_modifierrs  rceZdZdZdeeeffdZdZdZ dZ dZ dZ d Z d efd Zd efd Zd efdZd edefdZdS) JobScalerz^ Job Scaler. This class is responsible for scaling the number of concurrent requests. configc0tj|_d|_||_t |_tj|j|_t|_ t|_ d|_ t|_|dx}r||_ t"sdS|jdx}r||_ |jdx}r||_ |jdx}r ||_dSdS)Nr maxsizeZconcurrency_modifier jobs_fetcherjobs_fetcher_timeout jobs_handler)asyncioEvent_shutdown_eventrr"r job_progressQueue jobs_queuerr'r r(r)r r*getr)selfr"r'r(r)r*s r__init__zJobScaler.__init__-s &}#$  (NN!-0HIII$A!#$&!&#)::.D#E#E E  =(:: :< - ,D #';??3I#J#J J  =(:: :< - ,D    - -rclK||j|_|jr|j|jjkrdS|dkrt jdd{V3t j|j|_t d|jdS)Nrr r$z.JobScaler.set_scale | New concurrency set to: ) r'rr0r%current_occupancyr+sleepr/rdebugr2s r set_scalezJobScaler.set_scaleJs#'#<# ? ? ? M&-)= > > > > Q Q Q KKO P P P P P Q  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)rr7 kill_worker)r2signumframes rr?zJobScaler.handle_shutdownns: 8v888999 rcHKt4d{V}tj||}tj||}||g}tj|d{Vdddd{VdS#1d{VswxYwYdS)N)rr+ create_taskget_jobsrun_jobsgather)r2session jobtake_task jobrun_tasktaskss rrCz JobScaler.run}s0%'' ) ) ) ) ) ) )7".t}}W/E/EFFL!-dmmG.D.DEEK!;/E.%( ( ( ( ( ( ( ( ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) ) )sA'B BBc6|j S)z< Return whether the worker is alive or not. )r-is_setr8s ris_alivezJobScaler.is_alives'..0000rcltd|jdS)z- Whether to kill the worker. z Kill worker.N)rr7r-setr8s rrGzJobScaler.kill_workers1 .!!!   """""rrc|j}|j}td|jd|d|||zS)Nz JobScaler.status | concurrency: z ; queue: z ; progress: )r0qsizer. get_job_countrr7r)r2current_queue_countcurrent_progress_counts rr5zJobScaler.current_occupancys"o3355!%!2!@!@!B!B  Lt/G L LRe L LtJ L L   &(;;;rrOc &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: rFz!Failed to get job. | Error Type: z | Error Message: )rUr9rr5rr7r+r6wait_forr(r)r0putr.addinforYrCancelledError TimeoutError TypeError Exceptionrtype__name__str)r2rO jobs_needed acquired_jobsjobrs rrLzJobScaler.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==KK7chKg}|s|jsjt||jkr|js|jd{V}t j|||}| |t||jkr|j|rat dt|t j |t j d{V\}fd|D}t jdd{V|P|jjt j|d{VdS)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. NzJobs in progress: ) return_whencg|]}|v| Srr).0tdones r z&JobScaler.run_jobs..s;;;qQd]]]]]rr)rUr0emptylenrr1r+rKr appendrrdwaitFIRST_COMPLETEDr6rN)r2rOrRrntaskpendingrts @rrMzJobScaler.run_jobss mmoo #T_%:%:%<%< #e**t777@U@U@W@W7 O//11111111*4??7C+H+HII T""" e**t777@U@U@W@W7 <:c%jj::;;;&-lw'>'''!!!!!! g <;;;E;;;-"" " " " " " " "+mmoo #T_%:%:%<%< #0ne$$$$$$$$$$rrncK td|d|||j|d{V|jddr|n8#t $r+}td||d|d}~wwxYw|j |j |td|ddS#|j |j |td|dwxYw)za Process an individual job. This function is run concurrently for multiple jobs. z Handling Jobr_Nrefresh_workerFzError handling job: z Finished Job) rr7r*r"r1rGrhrr0 task_doner.remove)r2rOrnerrs rr zJobScaler.handle_jobs[ 1 IInc$i 0 0 0##GT[#>> > > > > > > >{/77 #  """    II2S22CI > > >I  O % % ' ' '   $ $S ) ) ) IInc$i 0 0 0 0 0 O % % ' ' '   $ $S ) ) ) IInc$i 0 0 0 0s+A2A76D7 B,&B''B,,DAEN)rj __module__ __qualname____doc__rrkrr3r9rDr?rCrUrGintr5rrLrMdictr rrrr!r!(s-tCH~----:   "   &    ) ) )111 ###<3<<<<7'm7'7'7'7'r %m % % % %D1 1D111111rr!)rr+r=r;rtypingrr http_clientrrrrp_jobr r rp_loggerr worker_stater rrrrrr!rrrrs(  MMMMMMMMMM''''''''######55555555lnn--- s s    i1i1i1i1i1i1i1i1i1i1r