+
    jDD                         R t ^ RIt^ RIt^ RIt^ RIt^ RIt^ RIt^ RIHt ^ RI	H
t
 ^RIHt ^RIHtHtHtHt ^RIHtHtHtHtHt  ! R R4      tR# )	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)DirtyAppErrorDirtyAppNotFoundErrorDirtyTimeoutErrorDirtyWorkerError)DirtyProtocolmake_responsemake_error_responsemake_chunk_messagemake_end_messagec                      a ] tR t^`t oRtRP                  4        U u. uF  p \        \        RV ,          4      NK  	  up tR t	R t
R tR tR tR	 tR
 tR tR tR tR tR tR tR tR tR tR tRtVtR# u up i )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.
zABRT HUP QUIT INT TERM USR1zSIG%sc                    Wn         RV n        W n        W0n        W@n        WPn        W`n        RV n        RV n        RV n	        \        V4      V n        / V n        RV n        RV n        RV n        R# )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)agepidppid	app_pathscfglogsocket_pathbootedabortedaliver   tmpapps_server_loop	_executor)selfr   r   r   r   r   r   s   &&&&&&&J/var/www/idalgo/venv/lib/python3.14/site-packages/gunicorn/dirty/worker.py__init__DirtyWorker.__init__k   sf     	"&
S>	
    c                "    R V P                    R2# )z<DirtyWorker >)r   r    s   &r!   __str__DirtyWorker.__str__   s    txxj**r$   c                :    V P                   P                  4        R# )zUpdate heartbeat timestamp.N)r   notifyr'   s   &r!   r+   DirtyWorker.notify   s    r$   c                   V P                   P                  '       dA   V P                   P                  P                  4        F  w  rV\        P                  V&   K  	  \
        P                  ! V P                   P                  V P                   P                  V P                   P                  R7       \
        P                  ! 4        \
        P                  ! V P                  P                  4       4       V P                  P                  4        V P                  4        V P!                  4        \        P"                  ! 4       V n        V P                   P'                  V 4       RV n        V P+                  4        R# )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)r   envitemsosenvironr   set_owner_processuidgidr.   seedclose_on_execr   filenor   init_signals	load_appsgetpidr   dirty_worker_initr   run)r    kvs   &  r!   init_processDirtyWorker.init_process   s     88<<<**, !

1 - 	txx||TXX\\*.((*=*=	? 			 	488??,-  	 	 99;""4( 
r$   c                L   V P                    F(  p\        P                  ! V\        P                  4       K*  	  \        P                  ! \        P                  V P                  4       \        P                  ! \        P
                  V P                  4       \        P                  ! \        P                  V P                  4       \        P                  ! \        P                  V P                  4       \        P                  ! \        P                  V P                  4       R# )zSet up signal handlers.N)	SIGNALSsignalSIG_DFLSIGTERM_signal_handlerSIGQUITSIGINTSIGABRTSIGUSR1)r    sigs   & r!   r9   DirtyWorker.init_signals   s     <<CMM#v~~.   	fnnd&:&:;fnnd&:&:;fmmT%9%9: 	fnnd&:&:; 	fnnd&:&:;r$   c                    V\         P                  8X  d   V P                  P                  4        R# RV n        V P
                  '       d(   V P
                  P                  V P                  4       R# R# )z(Handle signals by setting alive = False.NF)rD   rK   r   reopen_filesr   r   call_soon_threadsafe	_shutdown)r    rL   frames   &&&r!   rG   DirtyWorker._signal_handler   sL    &.. HH!!#
:::JJ++DNN; r$   c                b    V P                   '       d   V P                   P                  4        R# R# )zInitiate async shutdown.N)r   closer'   s   &r!   rQ   DirtyWorker._shutdown   s     <<<LL  r$   c                    \        V P                  4      V n        V P                  P                  4        FN  w  rV P                  P                  RV4        VP                  4        V P                  P                  RV4       KP  	  R#   \         d#   pT P                  P                  RY4       h Rp?ii ; i  \         d#   pT P                  P                  RT4       h Rp?ii ; i)zLoad all configured dirty apps.zLoaded dirty app: %szInitialized dirty app: %sz%Failed to initialize dirty app %s: %sNzFailed to load dirty apps: %s)
r   r   r   r0   r   debuginitinfo	Exceptionerrorr    pathappes   &   r!   r:   DirtyWorker.load_apps   s    	'7DI!YY__.	5t<HHJHHMM"=tD	 /
 ! HHNN#J#',  	HHNN:A>	s<   AB: ,B
B: 
B7B22B77B: :C'C""C'c                *   ^ RI Hp V P                  P                  pV! VRV P                   R2R7      V n        V P                  P                  RV4        \        P                  ! 4       V n
        \        P                  ! V P                  4       V P                  P                  V P                  4       4       V P!                  4        R#   \         d'   pT P                  P                  RT4        Rp?L>Rp?ii ; i  T P!                  4        i ; i)z Run the main asyncio event loop.)ThreadPoolExecutorzdirty-worker--)max_workersthread_name_prefixz#Created thread pool with %d threadszWorker error: %sN)concurrent.futuresrc   r   dirty_threadsr   r   r   rX   asyncionew_event_loopr   set_event_looprun_until_complete
_run_asyncr[   r\   _cleanup)r    rc   num_threadsr`   s   &   r!   r=   DirtyWorker.run   s     	: hh,,+#!.txxj:
 	<kJ	 //1DJ""4::.JJ))$//*;< MMO  	2HHNN-q11	2 MMOs+   A#C C=C83D  8C==D   Dc                  "   \         P                  P                  V P                  4      '       d!   \         P                  ! V P                  4       \
        P                  ! V P                  V P                  R7      G Rj  xL
 V n        \         P                  ! V P                  R4       V P                  P                  RV P                  V P                  4       \
        P                  ! V P                  4       4      p V P                  ;_uu_4       GRj  xL
  V P                  P                  4       G Rj  xL
  RRR4      GRj  xL
  TP#                  4         TG Rj  xL
  R#  L LU L5 L'  + GRj  xL 
 '       g   i     L>; i  \
        P                    d     LXi ; i LE  \
        P                    d     R# i ; i  TP#                  4         TG Rj  xL 
  i   \
        P                    d     i i ; i; i5i)z6Main async loop - start server and handle connections.)r^   Ni  zDirty worker %s listening on %s)r1   r^   existsr   unlinkri   start_unix_serverhandle_connectionr   chmodr   rZ   r   create_task_heartbeat_loopserve_foreverCancelledErrorcancel)r    heartbeat_tasks   & r!   rm   DirtyWorker._run_async   sy     77>>$**++IId&&' %66""!!
 
 	!!5)7hh 0 0	2 !,,T-A-A-CD
	||||ll00222 $|
 !!#$$$-
 $2 $|||%% 		
 %))  !!#$$$)) s  A?HE.A?HF E0F E6<E2=E6F E4F H"F/ 'F-(F/ ,H0F 2E64F 6F	<E?=
F	F	
F G F F*'G )F**G -F/ /GHGHHG)"G%#G)(H)H>H HHHc                   "   V P                   '       dL   V P                  4        \        P                  ! V P                  P
                  R,          4      G Rj  xL
  K]  R#  L5i)zPeriodically update heartbeat.g       @N)r   r+   ri   sleepr   dirty_timeoutr'   s   &r!   rx   DirtyWorker._heartbeat_loop  s@     jjjKKM-- 6 6 <=== =s   A$AA$A"	A$c                  "   V P                   P                  R4        V P                  '       d;    \        P                  ! V4      G Rj  xL
 pT P                  Y24      G Rj  xL
  KL   VP                  4         VP                  4       G Rj  xL
  R#  LK  \
        P                   d     KE  i ; i LP  \         d'   pT P                   P                  RT4        Rp?LwRp?ii ; i LZ  \         d     R# i ; i  TP                  4         TP                  4       G Rj  xL 
  i   \         d     i i ; i; i5i)zT
Handle a connection from the arbiter.

Each connection can send multiple requests.
zNew connection from arbiterNzConnection error: %s)r   rX   r   r	   read_message_asyncri   IncompleteReadErrorhandle_requestr[   r\   rU   wait_closed)r    readerwritermessager`   s   &&&  r!   ru   DirtyWorker.handle_connection   s     	45	***$1$D$DV$LLG ))':::  LLN((*** M22 
 ; 	6HHNN1155	6
 +  LLN((*** s   D?B7 B BB B7 $B5%B7 +C? ,D?=C- C+C- D?B B2.B7 /C? 1B22B7 7C(C#C? #C((C? +C- -C<8D?;C<<D??D<D+$D'%D+*D<+D96D<8D99D<<D?c                f  "   VP                  R\        \        P                  ! 4       4      4      pVP                  R4      pV\        P
                  8w  d9   \        V\        RV 24      4      p\        P                  ! W%4      G Rj  xL
  R# VP                  R4      pVP                  R4      pVP                  R. 4      pVP                  R/ 4      p	V P                  4         V P                  WgW4      G Rj  xL
 p
\        P                  ! V
4      '       d   V P                  W:V4      G Rj  xL
  R# \        P                  ! V
4      '       d   V P                  W:V4      G Rj  xL
  R# \!        W:4      p\        P                  ! W%4      G Rj  xL
  R#  EL L Ln L8 L  \"         d~   p\$        P&                  ! 4       pT P(                  P+                  R	YgY4       \        T\-        \        T4      YgTR
7      4      p\        P                  ! Y%4      G Rj  xL 
   Rp?R# Rp?ii ; i5i)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
idtypezUnknown message type: Napp_pathactionargskwargszError executing %s.%s: %s
%s)r   r   	traceback)getstruuiduuid4r	   MSG_TYPE_REQUESTr   r   write_message_asyncr+   executeinspectisgenerator_stream_sync_generator
isasyncgen_stream_async_generatorr
   r[   r   
format_excr   r\   r   )r    r   r   
request_idmsg_typeresponser   r   r   r   resultr`   tbs   &&&          r!   r   DirtyWorker.handle_request;  s     [[s4::<'89
;;v&}555* #9(!DEH  33FEEE;;z*X&{{62&Xr* 		F<<$GGF ""6**11*fMMM##F++22:vNNN )<#77III- F H NN J 		F%%'BHHNN:#Q4*c!fx(*,H
  33FEEE		Fs   BH1F	AH1&F& ;F<F& F& 1F 2F& 6H18F& F& )F"*F& .H10$F& F$F& H1F&  F& "F& $F& &H.1A,H)H H)#H1)H..H1c           	       aa
"   \        4       o
V
V3R lp \        P                  ! 4       p VP                  V P                  V4      G Rj  xL
 pVS
J d   M:\
        P                  ! V\        W4      4      G Rj  xL
  V P                  4        Kf  \
        P                  ! V\        V4      4      G Rj  xL
  SP%                  4        R#  L LR L  \         d{   p\        P                  ! 4       pT P                  P                  RYx4       \        T\!        \#        T4      TR7      4      p	\
        P                  ! Y94      G Rj  xL 
   Rp?LRp?ii ; i  SP%                  4        i ; i5i)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                  B   <  \        S4      #   \         d    S u # i ; iN)nextStopIteration)
_EXHAUSTEDgens   r!   	_get_next5DirtyWorker._stream_sync_generator.<locals>._get_next~  s'    "Cy   "!!"s   
 NError during streaming: %s
%sr   )objectri   get_running_looprun_in_executorr   r	   r   r   r+   r   r[   r   r   r   r\   r   r   r   rU   )r    r   r   r   r   loopchunkr`   r   r   r   s   &&f&      @r!   r   "DirtyWorker._stream_sync_generatorq  s+     X
	"	++-D"224>>9MMJ&#77.zA   33(4   IIK/ N  	F%%'BHHNN;QC*c!f3H  33FEEE	F IIKs}   E/5C C.C ;C<9C 5C6C :E/C C C EA*EE
EE EE E,,E/c           	     x  "    V  Rj  xL
  p\         P                  ! V\        W4      4      G Rj  xL
  V P                  4        KD   L? LDB\         P                  ! T\	        T4      4      G Rj  xL 
  M  \
         d{   p\        P                  ! 4       pT P                  P                  RYV4       \        T\        \        T4      TR7      4      p\         P                  ! Y74      G Rj  xL 
   Rp?MRp?ii ; iTP                  4       G Rj  xL 
  R#   TP                  4       G Rj  xL 
  i ; i5i)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
Nr   r   )r	   r   r   r+   r   r[   r   r   r   r\   r   r   r   aclose)r    r   r   r   r   r`   r   r   s   &&&&    r!   r   #DirtyWorker._stream_async_generator  s    	"  e#77.zA     #  33(4    	F%%'BHHNN;QC*c!f3H  33FEEE	F **,#**,s   D:A9 AA
A%A9 AA9 
AA9 $A9 2A53A9 8D 9C>A*C9.C1/C94D 9C>>D D:DD:D70D31D77D:c           	        aaaa"   WP                   9  d   \        V4      hV P                   V,          oV P                  P                  ^ 8  d   V P                  P                  MRp\        P
                  ! 4       p \        P                  ! VP                  V P                  VVVV3R l4      VR7      G Rj  xL
 pV#  L  \        P                   d4    T P                  P                  RTST4       \        RT RS R2TR7      hi ; i5i)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
Nc                     < S! S .SO5/ SB # r    )r   r_   r   r   s   r!   <lambda>%DirtyWorker.execute.<locals>.<lambda>  s    C888r$   )timeoutz%Execution timeout for %s.%s after %dszExecution of .z
 timed out)r   r   r   r   ri   r   wait_forr   r   TimeoutErrorr   warningr   )	r    r   r   r   r   r   r   r   r_   s	   &&fff   @r!   r   DirtyWorker.execute  s     , 99$'11ii!,0HH,B,BQ,F$((((D '')	"++$$NN8   F M ## 		HH7&' $z6(*= 		s1   A7D><C :C;C  DC ADDc                   V P                   '       d%   V P                   P                  RRR7       RV n         V P                  P                  4        F2  w  r VP	                  4        V P
                  P                  RV4       K4  	   V P                  P	                  4         \        P                  P                  V P                  4      '       d!   \        P                  ! V P                  4       T P
                  P                  RT P                   4       R#   \         d(   pT P
                  P                  RY4        Rp?K  Rp?ii ; i  \         d     Li ; i  \         d     L{i ; i)zClean up resources on shutdown.FT)waitcancel_futuresNzClosed dirty app: %szError closing dirty app %s: %szDirty worker %s exiting)r   shutdownr   r0   rU   r   rX   r[   r\   r   r1   r^   rr   r   rs   rZ   r   r]   s   &   r!   rn   DirtyWorker._cleanup  s    >>>NN##t#D!DN *IDJ		5t< +	HHNN
	ww~~d..//		$**+ 	/:!  J?IIJ  		  		sB   ,DE #.E   E  E%EEEE E.-E.)r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   N)__name__
__module____qualname____firstlineno____doc__splitgetattrrD   rC   r"   r(   r+   r@   r9   rG   rQ   r:   r=   rm   rx   ru   r   r   r   r   rn   __static_attributes____classdictcell__)x__classdict__s   0@r!   r   r   `   s      -22464 12wvw{+46G8+"H<"<!
",B>64Fl.`B2h; ;]6s    A:r   )r   ri   r   r1   rD   r   r   gunicornr   gunicorn.workers.workertmpr   r_   r   errorsr   r   r   r   protocolr	   r
   r   r   r   r   r   r$   r!   <module>r      sH   
?B   	     0    r; r;r$   