o i-@s@ddlZddlZddlZddlZddlZddlZddlmZddlm Z m Z ddl m Z ddl mZmZddlmZmZddlmZdd lmZd d lmZmZmZmZe rndd lmZzdd lmZWn e ymYnwGdddeZ!Gddde!Z"GdddeZ#Gddde!Z$Gddde!Z%Gddde!Z&dS)N)shuffle) TYPE_CHECKINGOptional)DEFAULT_WORKER_TTL)InvalidJobOperationShutDownImminentException)Job JobStatus)HorseMonitorTimeoutException)now)SHUTDOWN_SIGNAL BaseWorker WorkerStatus signal_name)Queue) struct_rusagec@sbeZdZefdejfddZdeee ee edffddZ dd d Z dddZ dddZ dS)Workersigc Csvztt|j||jd|j|jWdSty:}z|jtj kr.|j d|jnWYd}~dSd}~ww)zKill the horse but catch "No such process" error has the horse could already be dead. Args: sig (signal.Signals, optional): _description_. Defaults to SIGKILL. zWorker %s: killed horse pid %szWorker %s: horse already deadN) oskillpggetpgid horse_pidloginfonameOSErrorerrnoZESRCHdebug)selfrer"O/home/app/Keep/.python/lib/python3.10/site-packages/rq/worker/worker_classes.py kill_horses zWorker.kill_horsereturnrcCsTd}}}ttt|jd\}}}Wdn1s wY|||fS)zWaits for the horse process to complete. Uses `0` as argument as to include "any child in the process group of the current process". Nr) contextlibsuppressChildProcessErrorrwait4r)r pidstatrusager"r"r#wait_for_horse+s   zWorker.wait_for_horsejobr queuercCsnt}|jtjd<|jtjd<|dkr%t|||tddS||_| d|dt dS)zSpawns a work horse to perform the actual work and passes it a job. This is where the `fork()` actually happens. Args: job (Job): The Job that will be ran queue (Queue): The queue RQ_WORKER_ID RQ_JOB_IDrzForked  at N) rforkrenvironidsetpgrpZmain_work_horse_exit _horse_pidproclinetime)r r.r/ child_pidr"r"r#fork_work_horse4s   zWorker.fork_work_horsec Cs,d}}}t|_ z ||jt|\}}}Wdn1s%wYWnUtyc|t|j|jdkr\|j |jdkr\| |jd| |Yn%| |Ynt y}z|jtjkrq| WYd}~nd}~wwq |dd|_|jd|j|j|||tjkrdSz|}Wn tyYdSw|j|jkr|jd|j|j|jr||j|j||dd dS|tjtj fvrt|_!|rt"|rd t#|d nd }d ||d} |jd|j|j| |$|||||j||| d dSdS)zThe worker will monitor the work horse and make sure that it either executes successfully or the status of the job is set to failed Args: job (Job): _description_ queue (Queue): _description_ NT<rz@Worker %s: work horse finished for job %s: retpid=%s, ret_val=%szBWorker %s: job %s stopped by user, moving job to FailedJobRegistryz+Job stopped by user, work-horse terminated.)r/ exc_stringz (signal )z5Work-horse terminated unexpectedly; waitpid returned z; zWorker %s: job %s failed (%s))%r Z started_atZdeath_penalty_classZjob_monitoring_intervalr r-Zset_current_job_working_time total_secondstimeoutZcurrent_job_working_time heartbeatr$Zmaintain_heartbeatsrrZEINTRr8rrrr5rEX_OKZ get_statusrZ_stopped_job_idwarningZstopped_callbackZexecute_stopped_callbackZhandle_job_failurer FINISHEDZFAILEDZended_at WIFSIGNALEDWTERMSIGZhandle_work_horse_killed) r r.r/ZretpidZret_valr,r!Z job_statusZ signal_msgr?r"r"r#monitor_work_horseGs`         $zWorker.monitor_work_horsecCs2|||||||||tjdS)zSpawns a work horse to perform the actual work and passes it a job. The worker will wait for the work horse and make sure it executes within the given timeout bounds, or will end the work horse with SIGALRM. N)prepare_executionr<rJ set_staterIDLEr r.r/r"r"r# execute_jobs   zWorker.execute_jobNr.r r/r)__name__ __module__ __qualname__rsignalSignalsr$tuplerintr-r<rJrOr"r"r"r#rs $ Jrc@seZdZdZd ddZdS) SpawnWorkerzWorker implementation that uses os.spawn() instead of os.fork(). This implementation is intended for environments where `os.fork()` is not available. r.r r/rcCs|jtjd<|jtjd<|jjtjd<|jjj}|dr |d=|dr(|d=t tj t j t j dd|d|j d |jd |jd g}||_|d |d tdS)z@Spawns a work horse to perform the actual work using os.spawn().r0r1ZRQ_EXECUTION_IDretryZ driver_infoz-cz import os import sys from redis import Redis from rq import Worker, Queue from rq.job import Job from rq.executions import Execution # Recreate worker instance redis = Redis(**z) worker = Worker.find_by_key("zu", connection=redis) if not worker: sys.exit(1) # Reconstruct job, queue and execution objects job = Job.fetch("z0", connection=worker.connection) queue = Queue("a", connection=worker.connection) execution_id = os.environ.get('RQ_EXECUTION_ID') worker.execution = Execution.fetch(execution_id, job.id, connection=worker.connection) # Set up work horse os.setpgrp() worker._is_horse = True worker.main_work_horse(job, queue) zSpawned r2N)rrr4r5Z execution connectionZconnection_poolZconnection_kwargsgetspawnvP_NOWAITsys executablekeyr8r9r:)r r.r/Z redis_kwargsr;r"r"r#r<s4       "zSpawnWorker.fork_work_horseNrP)rQrRrS__doc__r<r"r"r"r#rXsrXc@s(eZdZd ddZdddefdd Zd S) SimpleWorkerr.r r/rcCs&||||||tjdS)z1Execute job in same thread/process, do not fork()N)rKZ perform_jobrLrrMrNr"r"r#rOs  zSimpleWorker.execute_jobr%cCs |jdkrtSt|jp tdS)z-1" means that jobs never timeout. In this case, we should _not_ do -1 + 60 = 59. We should just stick to DEFAULT_WORKER_TTL. Args: job (Job): The Job Returns: ttl (int): TTL r=r>)rCrrW)r r.r"r"r#get_heartbeat_ttls zSimpleWorker.get_heartbeat_ttlNrP)rQrRrSrOrWrcr"r"r"r#rbs rbc@s<eZdZdZdZgdZddZddZdd Zd d Z d S) HerokuWorkerz Modified version of rq worker which: * stops work horses getting killed with SIGTERM * sends SIGRTMIN to work horses on SIGTERM to the main process which in turn causes the horse to crash `imminent_shutdown_delay` seconds later )f_codef_lastif_linenof_localsf_tracecCs4ttj|jttjtjttjtjdS)z>Modified to ignore SIGINT and SIGTERM and only handle SIGRTMINN)rTSIGRTMINrequest_stop_sigrtminSIGINTSIG_IGNSIGTERMr r"r"r#setup_work_horse_signalssz%HerokuWorker.setup_work_horse_signalscCs<|jdkr|jd|j|jtjddS|jddS)z"If horse is alive send it SIGRTMINrzBWorker %s: warm shut down requested, sending horse SIGRTMIN signal)rz(Warm shut down requested, no horse foundN)rrrr`r$rTrkrFrpr"r"r#handle_warm_shutdown_requests z)HerokuWorker.handle_warm_shutdown_requestcCsf|jdkr|jd|||dS|jd|jttj|jttj|jt|jdS)Nrz@Imminent shutdown, raising ShutDownImminentException immediatelyzBImminent shutdown, raising ShutDownImminentException in %d seconds)imminent_shutdown_delayrrFrequest_force_stop_sigrtminrTrkSIGALRMalarm)r signumframer"r"r#rls  z"HerokuWorker.request_stop_sigrtmincs6fdd|jD}|jdtdt|d|)Ncsi|]}|t|qSr")getattr).0attrrxr"r# sz.z2raising ShutDownImminentException to cancel job...zshut down imminent (signal: r@)frame_propertiesrrFrr)r rwrxrr"r|r#rts z(HerokuWorker.request_force_stop_sigrtminN) rQrRrSrarsr~rqrrrlrtr"r"r"r#rds rdc@eZdZdZddZdS)RoundRobinWorkerze Modified version of Worker that dequeues jobs from the queues using a round-robin strategy. cCs6|j|}|j|dd|jd|d|_dS)Nr )_ordered_queuesindex)r reference_queueposr"r"r#reorder_queuess *zRoundRobinWorker.reorder_queuesNrQrRrSrarr"r"r"r#r rc@r) RandomWorkerz` Modified version of Worker that dequeues jobs from the queues using a random strategy. cCst|jdS)N)rr)r rr"r"r#r&szRandomWorker.reorder_queuesNrr"r"r"r#r!rr)'r&rrrTr^r:randomrtypingrrdefaultsr exceptionsrrr.r r Ztimeoutsr utilsr baserrrrr/rresourcer ImportErrorrrXrbrdrrr"r"r"r#s8      8+