iDDSrSSKrSSKrSSKrSSKrSSKrSSKrSSKJr SSK J r SSK J r SSK JrJrJrJr SSKJrJrJrJrJr "S S 5rg) 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 \rSrSrSrSR 5VVVVs/sHn[ [SU-5PM snnnnrSr Sr Sr Sr S r S rS rS rS rSrSrSrSrSrSrSrSrSrgs snnnnf) DirtyWorker`z 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%scXlSUlX lX0lX@lXPlX`lSUlSUlSUl [U5Ul 0Ul SUl SUlSUlg)z 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 G/var/www/ias/venv/lib/python3.13/site-packages/gunicorn/dirty/worker.py__init__DirtyWorker.__init__ksf "&   S>   c"SURS3$)Nz )rr#s r$__str__DirtyWorker.__str__stxxj**r'c8URR5 g)zUpdate heartbeat timestamp.N)rnotifyr*s r$r.DirtyWorker.notifys r'cURR(a@URRR5HupU[RU'M [ R "URRURRURRS9 [ R"5 [ R"URR55 URR5 UR5 UR!5 [R""5UlURR'U5 SUlUR+5 g)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_processuidgidr1seed close_on_execrfilenor init_signals load_appsgetpidrdirty_worker_initrrun)r#kvs r$ init_processDirtyWorker.init_processs 88<< **, ! 1 - txx||TXX\\*.((*=*= ?  488??,-    99; ""4(   r'cJURH(n[R"U[R5 M* [R"[RUR5 [R"[R UR5 [R"[R UR5 [R"[RUR5 [R"[RUR5 g)zSet up signal handlers.N) SIGNALSsignalSIG_DFLSIGTERM_signal_handlerSIGQUITSIGINTSIGABRTSIGUSR1)r#sigs r$r<DirtyWorker.init_signalss<  s<AB8,B B8 B5B00B55B88 C$CC$c SSKJn URRnU"USURS3S9UlUR RSU5 [R"5Ul [R"UR5 URRUR55 UR!5 g![a&nUR RSU5 SnAN>$** + + IId&& '%66  " "!!   !!5)  7hh 0 0 2!,,T-A-A-CD |||ll00222$|  ! ! # $$$- $2$|||%%    %))    ! ! # $$$))  sA;G4=E!>A?G4>FE#FE)0E%1E)5 FE'FG4FFF G4#F%E)'F)F/E2 0 F<F?F9FFF9FF9FF63G45F66G49G1 GGGG1G.+G1-G..G11G4c# UR(aYUR5 [R"URR S- 5IShvN UR(aMXggN7f)zPeriodically update heartbeat.g@N)rr.rlsleepr dirty_timeoutr*s r$r{DirtyWorker._heartbeat_loopsDjj KKM-- 6 6 <= = =jjj =sAA0A.A0,A0c# URRS5 UR(aK[R"U5IShvN nURX25IShvN UR(aMKUR5 UR5IShvN gNZ![ R a MCf=fN^![a&nURRSU5 SnANtSnAff=fNW![a gf=f!UR5 UR5IShvN f![a ff=f=f7f)zT Handle a connection from the arbiter. Each connection can send multiple requests. zNew connection from arbiterNzConnection error: %s) rr[rr read_message_asyncrlIncompleteReadErrorhandle_requestr^r_rX wait_closed)r#readerwritermessagercs r$rxDirtyWorker.handle_connection s 45 **$1$D$DV$LLG ))':::*** LLN ((***M22 ; 6 HHNN11 5 5 6 +   LLN ((***  sECB' B% B'C#C$C;E C9C7 C9$E%B''B?;C<D >B??C C4C/*D /C44D 7C99 DEDE ED5.D1/D54E5 E?EEEEc N# URS[[R"555nURS5nU[R :wa7[ U[SU355n[R"X%5IShvN gURS5nURS5nURS/5nURS05n UR5 URXgX5IShvN n [R"U 5(aURX:U5IShvN g[R"U 5(aURX:U5IShvN g[!X:5n[R"X%5IShvN gGN NNjN6N!["ayn [$R&"5n UR(R+S XgX5 [ U[-[U 5XgU S 95n[R"X%5IShvN Sn A gSn A ff=f7f) a) 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 r^r format_excrr_r) r#rr request_idmsg_typeresponserrrrresultrctbs r$rDirtyWorker.handle_request;s[[s4::<'89 ;;v& }55 5* #9(!DEH 33FE E E ;;z*X&{{62&Xr*  F<<$GGF""6**11*fMMM##F++22:vNNN)<#77III- FHNNJ F%%'B HHNN:#Q 4*c!fx(*,H  33FE E E FsBH%FAH%$F9F:4F.F/F3H%40F$F%F)H%*$FFFH%FFFF H")A)HHHH%H""H%c ^^ # [5m U U4Sjn[R"5nURURU5IShvN nUT LaO:[ R "U[X55IShvN UR5 Md[ R "U[U55IShvN TR%5 gN}NQN![awn[R"5nURRSXx5 [U[![#U5US95n [ R "X95IShvN SnANSnAff=f!TR%5 f=f7f)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@>[T5$![a Ts$f=fN)next StopIteration) _EXHAUSTEDgensr$ _get_next5DirtyWorker._stream_sync_generator.._get_next~s' "Cy   "!! "s  NError during streaming: %s %sr)objectrlget_running_looprun_in_executorr"r rrr.rr^rrrr_r rrrX) r#rrrrloopchunkrcrrrs ` @r$r"DirtyWorker._stream_sync_generatorqs2X  "  ++-D"224>>9MMJ&#77.zA  33(4   IIK/N   F%%'B HHNN;Q C*c!f3H 33FE E E F IIKs}E)5C C  -C:C ;9C4C5C9E) C CC EA'E EE E EEE&&E)c l# UShvN n[R"U[X55IShvN UR5 MCN>N [R"U[ U55IShvN O![ awn[ R"5nURRSXV5 [U[[U5US95n[R"X75IShvN SnAOSnAff=fUR5IShvN g!UR5IShvN f=f7f)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.rr^rrrr_r rraclose)r#rrrrrcrrs r$r#DirtyWorker._stream_async_generators " e#77.zA   # 33(4   F%%'B HHNN;Q C*c!f3H 33FE E E F**,  #**,  sD4A8A A A $A8A A8 A  A8 $A81A42A87D8 C9A'C4)C,*C4/D4C99D<D4DD4D1*D-+D11D4c ^^^^# XR;a [U5eURUmURRS:aURROSn[R "5n[R "URURUUUU4Sj5US9IShvN nU$N![Ra0 URRSUTU5 [SUSTS3US9ef=f7f) 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 rNc>T"T/TQ70TD6$r)rrbrrsr$%DirtyWorker.execute..sC888r')timeoutz%Execution timeout for %s.%s after %dsz Execution of .z timed out) rrrrrlrwait_forrr" TimeoutErrorrwarningr ) r#rrrrrrrrbs ``` @r$rDirtyWorker.executes, 99 $'1 1ii!,0HH,B,BQ,F$((((D'') "++$$NN8 FM##  HH  7&' $z6(*=  s1A1D89B:1B82B:7D8B::AC>>DcUR(a!URRSSS9 SUlURR5H2upUR 5 UR R SU5 M4 URR 5 [RRUR5(a [R"UR5 UR RSUR 5 g![a'nUR RSX5 SnAMSnAff=f![a Nf=f![a Nwf=f)zClean up resources on shutdown.FT)waitcancel_futuresNzClosed dirty app: %szError closing dirty app %s: %szDirty worker %s exiting)r"shutdownrr3rXrr[r^r_rr4rarurrvr]rr`s r$rqDirtyWorker._cleanups >> NN # #t # D!DN*ID J 5t<+  HHNN   ww~~d..// $**+  /:! J?II J       s=,DEAE EEE EE E%$E%)r"r!r rrrrrrrrrrrrN)__name__ __module__ __qualname____firstlineno____doc__splitgetattrrGrFr%r+r.rCr<rJrTr=r@rpr{rxrrrrrq__static_attributes__).0xrrGs0000r$rr`s-2246412wvw{+46G8+"H<"<! ",B> 64Fl.`B2h;] 6sA7 r)rrlrr4rGrrgunicornrgunicorn.workers.workertmprrbrerrorsrrr r protocolr r r rrrrr'r$rsH ?B  0  r;r;r'