o iDD@sdZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZmZmZmZddlmZmZmZmZmZGd d d ZdS) 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@seZdZdZdddDZddZddZd d Zd d Z d dZ ddZ ddZ ddZ ddZddZddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'S)( 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. cCsg|] }ttd|qS)zSIG%s)getattrsignal).0xrI/home/ubuntu/.local/lib/python3.10/site-packages/gunicorn/dirty/worker.py hszDirtyWorker.zABRT HUP QUIT INT TERM USR1cCsb||_d|_||_||_||_||_||_d|_d|_d|_ t ||_ i|_ d|_ d|_d|_dS)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)selfrrrrrrrrr__init__ks   zDirtyWorker.__init__cCsd|jdS)Nz )rr&rrr__str__szDirtyWorker.__str__cCs|jdS)zUpdate heartbeat timestamp.N)r!notifyr)rrrr+szDirtyWorker.notifycCs|jjr|jjD] \}}|tj|<q tj|jj|jj|jj dt t |j |j ||t|_|j|d|_|dS)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_execr!filenor init_signals load_appsgetpidrdirty_worker_initrrun)r&kvrrr init_processs     zDirtyWorker.init_processcCsn|jD] }t|tjqttj|jttj|jttj|jttj|jttj|jdS)zSet up signal handlers.N) SIGNALSrSIG_DFLSIGTERM_signal_handlerSIGQUITSIGINTSIGABRTSIGUSR1)r&sigrrrr7s zDirtyWorker.init_signalscCs:|tjkr |jdSd|_|jr|j|jdSdS)z(Handle signals by setting alive = False.NF)rrFr reopen_filesr r$call_soon_threadsafe _shutdown)r&rGframerrrrBs  zDirtyWorker._signal_handlercCs|jr |jdSdS)zInitiate async shutdown.N)r#closer)rrrrJszDirtyWorker._shutdownc Csz._get_nextTNError during streaming: %s %sr})objectr\get_running_looprun_in_executorr%r rr r+rrPr}rrrQr rrrL) r&rrrurloopchunkrUrrrrrrqs>     z"DirtyWorker._stream_sync_generatorc sznz'|2z3dHW}t|t||IdH|q6t|t|IdHWn2ty[}z&t}|j d||t |t t ||d}t||IdHWYd}~nd}~wwW| IdHdSW| IdHdS| IdHw)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 rr r+rrPr}rrrQr rraclose)r&rrrurrUrrrrrrs2    "z#DirtyWorker._stream_async_generatorc s||jvr t||j||jjdkr|jjnd}t}ztj||jfdd|dIdH}|WStj yV|j d||t d|dd |dw) 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 rNcsgRiSrrrrzrTr{r|rrsz%DirtyWorker.execute..)timeoutz%Execution timeout for %s.%s after %dsz Execution of .z timed out) r"rrror\rwait_forrr% TimeoutErrorrwarningr)r&ryrzr{r|rrrrrrrs2   zDirtyWorker.executec Cs|jr|jjdddd|_|jD]+\}}z ||jd|Wqty>}z|jd||WYd}~qd}~wwz|j Wn tyOYnwzt j |j r^t |j Wn tyhYnw|jd|jdS)zClean up resources on shutdown.FT)waitcancel_futuresNzClosed dirty app: %szError closing dirty app %s: %szDirty worker %s exiting)r%shutdownr"r.rLrrMrPrQr!r/rSrcrrdrOrrRrrrras0   zDirtyWorker._cleanupN)__name__ __module__ __qualname____doc__splitr?r'r*r+r>r7rBrJr8r;r`rirfrrrrrrarrrrr`s,$ !60! 4r)rr\rr/rr}rgunicornrgunicorn.workers.workertmprrTrerrorsrrrr protocolr r r r rrrrrrsA