AҐiDDdZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZmZmZmZddlmZmZmZmZmZGd d Zy) a Dirty Worker Process Asyncio-based worker that loads dirty apps and handles requests from the DirtyArbiter. Threading Model --------------- Each dirty worker runs an asyncio event loop in the main thread for: - Handling connections from the arbiter - Managing heartbeat updates - Coordinating task execution Actual app execution runs in a ThreadPoolExecutor (separate threads): - The number of threads is controlled by ``dirty_threads`` config (default: 1) - Each thread can execute one app action at a time - The asyncio event loop is NOT blocked by task execution State and Global Objects ------------------------ Apps can maintain persistent state because: 1. Apps are loaded ONCE when the worker starts (in ``load_apps()``) 2. The same app instances are reused for ALL requests 3. App state (instance variables, loaded models, etc.) persists Example:: class MLApp(DirtyApp): def init(self): self.model = load_heavy_model() # Loaded once, reused self.cache = {} # Persistent cache def predict(self, data): return self.model.predict(data) # Uses loaded model Thread Safety: - With ``dirty_threads=1`` (default): No concurrent access, thread-safe by design - With ``dirty_threads > 1``: Multiple threads share the same app instances, apps MUST be thread-safe (use locks, thread-local storage, etc.) Heartbeat and Liveness ---------------------- The worker sends heartbeat updates to prove it's alive: 1. A dedicated asyncio task (``_heartbeat_loop``) runs independently 2. It updates the heartbeat file every ``dirty_timeout / 2`` seconds 3. Since tasks run in executor threads, they do NOT block heartbeats 4. The arbiter kills workers that miss heartbeat updates Timeout Control --------------- Execution timeout is enforced at two levels: 1. **Worker level**: Each task execution has a timeout (``dirty_timeout``). If exceeded, the worker returns a timeout error but the thread may continue running (Python threads cannot be cancelled). 2. **Arbiter level**: The arbiter also enforces timeout when waiting for worker response. Workers that don't respond are killed via SIGABRT. Note: Since Python threads cannot be forcibly cancelled, a truly stuck operation will continue until the worker is killed by the arbiter. N)util) WorkerTmp)load_dirty_apps) DirtyAppErrorDirtyAppNotFoundErrorDirtyTimeoutErrorDirtyWorkerError) DirtyProtocol make_responsemake_error_responsemake_chunk_messagemake_end_messagec eZdZdZdj Dcgc]}t t d|zc}}}}ZdZdZ dZ dZ dZ d Z d Zd Zd Zd ZdZdZdZdZdZdZdZycc}}}}w) DirtyWorkerz Dirty worker process that loads dirty apps and handles requests. Each worker runs its own asyncio event loop and listens on a worker-specific Unix socket for requests from the DirtyArbiter. zABRT HUP QUIT INT TERM USR1zSIG%sc||_d|_||_||_||_||_||_d|_d|_d|_ t||_ i|_ d|_ d|_d|_y)a? Initialize a dirty worker. Args: age: Worker age (for identifying workers) ppid: Parent process ID app_paths: List of dirty app import paths cfg: Gunicorn config log: Logger socket_path: Path to this worker's Unix socket z [booting]FTN)agepidppid app_pathscfglog socket_pathbootedabortedalivertmpapps_server_loop _executor)selfrrrrrrs N/var/www/descvideos/venv/lib/python3.12/site-packages/gunicorn/dirty/worker.py__init__zDirtyWorker.__init__ksr "&   S>   c"d|jdS)Nz )rr"s r#__str__zDirtyWorker.__str__stxxj**r%c8|jjy)zUpdate heartbeat timestamp.N)rnotifyr(s r#r+zDirtyWorker.notifys r%c|jjr?|jjjD]\}}|tj|<t j |jj|jj|jjt jt j|jj|jj|j|j!tj"|_|jj'|d|_|j+y)z Initialize the worker process after fork. This is called in the child process after fork. It sets up the environment, loads apps, and starts the main run loop. ) initgroupsTN)renvitemsosenvironrset_owner_processuidgidr-seed close_on_execrfilenor init_signals load_appsgetpidrdirty_worker_initrrun)r"kvs r# init_processzDirtyWorker.init_processs 88<< **, "1 ! 1  " txx||TXX\\*.((*=*= ?  488??,-    99; ""4(   r%c:|jD]&}tj|tj(tjtj|jtjtj |jtjtj |jtjtj|jtjtj|jy)zSet up signal handlers.N) SIGNALSsignalSIG_DFLSIGTERM_signal_handlerSIGQUITSIGINTSIGABRTSIGUSR1)r"sigs r#r8zDirtyWorker.init_signalss<< /C MM#v~~ . /  fnnd&:&:; fnnd&:&:; fmmT%9%9:  fnnd&:&:;  fnnd&:&:;r%c|tjk(r|jjyd|_|j r&|j j |jyy)z(Handle signals by setting alive = False.NF)rBrIr reopen_filesrr call_soon_threadsafe _shutdown)r"rJframes r#rEzDirtyWorker._signal_handlersJ &.. HH ! ! #  :: JJ + +DNN ; r%cR|jr|jjyy)zInitiate async shutdown.N)rcloser(s r#rNzDirtyWorker._shutdowns << LL    r%c t|j|_|jjD]N\}}|jj d| |j |jjd|Py#t$r#}|jjd||d}~wwxYw#t$r"}|jjd|d}~wwxYw)zLoad all configured dirty apps.zLoaded dirty app: %szInitialized dirty app: %sz%Failed to initialize dirty app %s: %sNzFailed to load dirty apps: %s) rrrr/rdebuginitinfo Exceptionerrorr"pathappes r#r9zDirtyWorker.load_appss '7DI!YY__.  c5t<HHJHHMM"=tD   !HHNN#J#',  HHNN:A >  s<AB7,BB7 B4B//B44B77 C"CC"c ddlm}|jj}||d|jd|_|j jd| tj|_ tj|j|jj|j|j!y#t$r&}|j jd|Yd}~;d}~wwxYw#|j!wxYw) z Run the main asyncio event loop.r)ThreadPoolExecutorz dirty-worker--) max_workersthread_name_prefixz#Created thread pool with %d threadszWorker error: %sN)concurrent.futuresr]r dirty_threadsrr!rrSasyncionew_event_loopr set_event_looprun_until_complete _run_asyncrVrW_cleanup)r"r] num_threadsr[s r#r<zDirtyWorker.runs :hh,, +#!.txxj:  >$** + IId&& '%66  " "!!   !!5)  7hh 0 0 2!,,T-A-A-CD || 3 3ll00222 3 3  ! ! # $$$-  32 3 3 3 3%%    %))    ! ! # $$$))  sA7G&9E:A=G&8E8E E8 E#*E+E#/ E8:E!;E8?G&FFFG&E8E#!E8#E5)E, *E51E88F F, FF,FF)&G&(F))G&,G#>G GG  G# G G#G  G##G&cK|jrR|jtj|jj dz d{|jrQyy7w)zPeriodically update heartbeat.g@N)rr+rcsleepr dirty_timeoutr(s r#rqzDirtyWorker._heartbeat_loopsCjj KKM-- 6 6 <= = =jj =sA A$A"A$ A$ctK|jjd |jrE tj|d{}|j||d{|jrE|j |jd{y7U#t j $rYAwxYw7V#t$r&}|jjd|Yd}~qd}~wwxYw7T#t$rYywxYw#|j |jd{7w#t$rYwwxYwxYww)zl Handle a connection from the arbiter. Each connection can send multiple requests. zNew connection from arbiterNzConnection error: %s) rrSrr read_message_asyncrcIncompleteReadErrorhandle_requestrVrWrQ wait_closed)r"readerwritermessager[s r#rnzDirtyWorker.handle_connection s 45 **$1$D$DV$LLG ))'6:::** LLN ((***M22 ; 6 HHNN11 5 5 6 +   LLN ((***  sD8 B7BBB B7B5B70D8C+C)C+D8BB2/B71B22B77 C&C!C:!C&&C:)C++ C74D86C77D8:D5 D&D" D&%D5& D2/D51D22D55D8c DK|jdttj}|jd}|tj k7r7t |td|}t j||d{y|jd}|jd}|jdg}|jdi} |j |j|||| d{} tj| r|j|| |d{ytj| r|j|| |d{yt!|| }t j||d{y777f777#t"$r~} t%j&} |j(j+d ||| | t |t-t| ||| }t j||d{7Yd} ~ yd} ~ wwxYww) ai Handle a single request message. Supports both regular (non-streaming) and streaming responses. For streaming, detects if the result is a generator and sends chunk messages followed by an end message. Args: message: Request dict from protocol writer: StreamWriter for sending responses idtypezUnknown message type: Napp_pathactionargskwargszError executing %s.%s: %s %s)rr traceback)getstruuiduuid4r MSG_TYPE_REQUESTr r write_message_asyncr+executeinspect isgenerator_stream_sync_generator isasyncgen_stream_async_generatorr rVr format_excrrWr) r"rr request_idmsg_typeresponserrrrresultr[tbs r#r|zDirtyWorker.handle_request;s[[s4::<'89 ;;v& }55 5* #9(!DEH 33FHE E E ;;z*X&{{62&Xr*  F<<&$GGF""6*11*ffMMM##F+22:vvNNN)V<#77III- FHNNJ F%%'B HHNN:#VQ 4*c!fx(*,H  33FHE E E FsBH F AH "F9F:/F)F*F.H /+FFFH  %FFF H FFFF HA.H HHH HH c  Kt fd} tj} |j|j|d{}| urn9t j |t||d{|jbt j |t|d{j%y7z7O7#t$rz}tj}|jjd||t|t!t#||} t j || d{7Yd}~d}~wwxYw#j%wxYww)z Stream chunks from a synchronous generator. Args: request_id: Request ID for the messages gen: Sync generator to iterate writer: StreamWriter for sending messages c> tS#t$rcYSwxYwN)next StopIteration) _EXHAUSTEDgensr# _get_nextz5DirtyWorker._stream_sync_generator.._get_next~s' "Cy   "!! "s  NError during streaming: %s %sr)objectrcget_running_looprun_in_executorr!r rrr+rrVrrrrWr rrrQ) r"rrrrloopchunkr[rrrs ` @r#rz"DirtyWorker._stream_sync_generatorqs;X  "  ++-D"224>>9MMJ&#77.z5A  33(4   IIK/N   F%%'B HHNN;Q C*c!f3H 33FHE E E F IIKs}E&4C C ,C 7C87C /C 0C 4E&C C C EA*E >E?E E EEE##E&c lK |23d{}tj|t||d{|jB7=76tj|t |d{7n#t $rz}t j}|jjd||t|tt||}tj||d{7Yd}~nd}~wwxYw|jd{7y#|jd{7wxYww)z Stream chunks from an asynchronous generator. Args: request_id: Request ID for the messages gen: Async generator to iterate writer: StreamWriter for sending messages Nrr)r rrr+rrVrrrrWr rraclose)r"rrrrr[rrs r#rz#DirtyWorker._stream_async_generators  "  e#77.z5A   # 33(4   F%%'B HHNN;Q C*c!f3H 33FHE E E F**,  #**,  sD4A6A AA $A6A A6A  A6 #A6/A20A65D6 C9?A*C4)C,*C4/D4C99D<D4DD4D1*D-+D11D4c K||jvr t||j||jjdkDr|jjnd}t j } t j |j|jfd|d{}|S7#tj$r3|jjd||td|dd|wxYww) a Execute an action on a dirty app. The action runs in a thread pool executor to avoid blocking the asyncio event loop. Execution timeout is enforced using ``dirty_timeout`` config. Args: app_path: Import path of the dirty app action: Action name to execute args: Positional arguments kwargs: Keyword arguments Returns: Result from the app action Raises: DirtyAppNotFoundError: If app is not loaded DirtyTimeoutError: If execution exceeds timeout DirtyAppError: If execution fails rNcgiSr)rrZrrsr#z%DirtyWorker.execute..sC888r%)timeoutz%Execution timeout for %s.%s after %dsz Execution of .z timed out) rrrrxrcrwait_forrr! TimeoutErrorrwarningr ) r"rrrrrrrrZs ``` @r#rzDirtyWorker.executes, 499 $'1 1ii!,0HH,B,BQ,F$((((D'') "++$$NN8 FM##  HH  7&' $z6(*=  s1A.D5:B8/B60B85D6B88AC>>Dc|jr$|jjddd|_|jjD]2\}} |j |j j d|4 |jj  tjj|jrtj|j|j jd|j y#t$r'}|j jd||Yd}~d}~wwxYw#t$rYwxYw#t$rYtwxYw)zClean up resources on shutdown.FT)waitcancel_futuresNzClosed dirty app: %szError closing dirty app %s: %szDirty worker %s exiting)r!shutdownrr/rQrrSrVrWrr0rYrkrrlrUrrXs r#rhzDirtyWorker._cleanups >> NN # #t # D!DN* JID# J 5t< J  HHNN   ww~~d../ $**+  /:! J?qII J       s=,D D>AE D;D66D;> E  E  EEN)__name__ __module__ __qualname____doc__splitgetattrrBrAr$r)r+r?r8rErNr9r<rgrqrnr|rrrrh).0xrrBs0000r#rr`s-22466wvw{+6G8+"H<"<! ",B> 64Fl.`B2h;] 6sA0 r)rrcrr0rBrrgunicornrgunicorn.workers.workertmprrZrerrorsrrr r protocolr r r rrrrr%r#rsH ?B  0  r;r;r%