o ih@sdZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z ddl mZmZmZmZddlmZmZmZmZmZmZmZmZmZmZmZmZmZm Z m!Z!ddl"m#Z#Gd d d Z$dS) z Dirty Arbiter Process Asyncio-based arbiter that manages the dirty worker pool and routes requests from HTTP workers to available dirty workers. N)util)get_app_workers_attributeparse_dirty_app_spec) DirtyErrorDirtyNoWorkersAvailableErrorDirtyTimeoutErrorDirtyWorkerError) DirtyProtocolmake_error_response make_response STASH_OP_PUT STASH_OP_GETSTASH_OP_DELETE STASH_OP_KEYSSTASH_OP_CLEAR STASH_OP_INFOSTASH_OP_ENSURESTASH_OP_DELETE_TABLESTASH_OP_TABLESSTASH_OP_EXISTS MANAGE_OP_ADDMANAGE_OP_REMOVE) DirtyWorkerc@s.eZdZdZdddDZdZdIddZd d Zd d Z d dZ ddZ ddZ ddZ ddZddZddZddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(ZdJd)d*Zd+d,Zd-d.Zd/d0Zd1d2Zd3d4Zd5d6ZdKd8d9Zd:d;Z dd?Z"d@dAZ#dBdCZ$dLdEdFZ%dGdHZ&dS)M DirtyArbitera4 Dirty arbiter that manages the dirty worker pool. The arbiter runs an asyncio event loop and handles: - Spawning and managing dirty worker processes - Accepting connections from HTTP workers - Routing requests to available dirty workers - Monitoring worker health via heartbeat cCsg|] }ttd|qS)zSIG%s)getattrsignal).0xrJ/home/ubuntu/.local/lib/python3.10/site-packages/gunicorn/dirty/arbiter.py <szDirtyArbiter.z*HUP QUIT INT TERM TTIN TTOU USR1 USR2 CHLDNcCs||_||_d|_t|_||_tjdd|_ |p!tj |j d|_ i|_ i|_i|_i|_i|_d|_d|_d|_|jj|_d|_d|_i|_i|_i|_i|_i|_g|_i|_| dS)z Initialize the dirty arbiter. Args: cfg: Gunicorn config log: Logger socket_path: Path to the arbiter's Unix socket pidfile: Well-known PID file location for orphan detection Nzgunicorn-dirty-)prefixz arbiter.sockrT)!cfglogpidosgetpidppidpidfiletempfilemkdtemptmpdirpathjoin socket_pathworkersworker_socketsworker_connections worker_queuesworker_consumers_worker_rr_index worker_agealive dirty_workers num_workers_server_loop_pending_requests app_specsapp_worker_mapworker_app_map_app_rr_indices_pending_respawns stash_tables_parse_app_specs)selfr$r%r0r*rrr __init__Bs8     zDirtyArbiter.__init__c Cs|jjD]<}t|\}}|dur1zt|}Wnty0}z|jd||WYd}~nd}~ww|||d|j|<t|j |<qdS)a Parse all app specifications from config. Populates self.app_specs with parsed information about each app, including the import path and worker count limits. Worker count priority: 1. Config override (e.g., "module:Class:2") - highest priority 2. Class attribute (e.g., workers = 2 on the class) 3. None (all workers) - default Nz,Could not read workers attribute from %s: %s) import_path worker_count original_spec) r$ dirty_appsrr Exceptionr%warningr>setr?)rEspecrGrHerrr rDys$   zDirtyArbiter._parse_app_specscCs2d}|jD]}|d}|durt||}q|S)a Calculate minimum number of workers required by app specs. Returns the maximum worker_count across all apps that have limits. Apps with worker_count=None don't impose a minimum. Returns: int: Minimum workers required (at least 1) rrHN)r>valuesmax)rE min_requiredrNrHrrr _get_minimum_workerss  z!DirtyArbiter._get_minimum_workerscCs^g}|jD]%\}}|d}t|j|t}|dur#||q||kr,||q|S)a Determine which apps a new worker should load. Returns a list of import paths for apps that need more workers. Apps with workers=None (all workers) are always included. Apps with worker limits are included only if they haven't reached their limit yet. Returns: List of import paths to load, or empty list if no apps need workers rHN)r>itemslenr?getrMappend)rE app_pathsrGrNrHcurrent_workersrrr _get_apps_for_new_workers   z%DirtyArbiter._get_apps_for_new_workercCsBt||j|<|D]}||jvrt|j|<|j||q dS)a? Register which apps a worker has loaded. Updates both app_worker_map and worker_app_map to track the bidirectional relationship between workers and apps. Args: worker_pid: The PID of the worker app_paths: List of app import paths loaded by this worker N)listr@r?rMaddrE worker_pidrXapp_pathrrr _register_worker_appss    z"DirtyArbiter._register_worker_appscCs6|j|g}|D]}||jvr|j||q dS)z Unregister a worker's apps when it exits. Removes the worker from all tracking maps. Args: worker_pid: The PID of the worker to unregister N)r@popr?discardr]rrr _unregister_workers   zDirtyArbiter._unregister_workerc Cs t|_|jd|j|jrJz t|jd}|t|jWdn1s*wYWnt yI}z |j d|WYd}~nd}~ww|j tj d<|j ||tdz!z t|Wn tyrYnwW|dSW|dS|w)z&Run the dirty arbiter (blocking call).z Dirty arbiter starting (pid: %s)wNzFailed to write PID file: %sGUNICORN_DIRTY_SOCKETz dirty-arbiter)r'r(r&r%infor*openwritestrIOErrorrLr0environr$on_dirty_starting init_signalsr _setproctitleasynciorun _run_asyncKeyboardInterrupt _cleanup_sync)rEfrOrrr rps2      zDirtyArbiter.runcCs|jD] }t|tjqttj|jttj|jttj|jttj|jttj|jttj |jttj |jttj |jdS)zSet up signal handlers.N) SIGNALSrSIG_DFLSIGTERM_signal_handlerSIGQUITSIGINTSIGHUPSIGUSR1SIGCHLDSIGTTINSIGTTOU)rEsigrrr rm s zDirtyArbiter.init_signalscs:|tjkrjrjfdddS|tjkr jdS|tjkr4jr2jfdddS|tjkrWj d7_ j dj jrUjfdddS|tj kr }j |krnj d|dSj d8_ j d j jrjfd ddSd _jrjjdSdS) zHandle signals.ctSN)ro create_task_handle_sigchldrrErr z.DirtyArbiter._signal_handler..Ncrr)rorreloadrrrr r+rrz'SIGTTIN: Increasing dirty workers to %scrrrormanage_workersrrrr r6rzASIGTTOU: Cannot decrease below %s workers (required by app specs)z'SIGTTOU: Decreasing dirty workers to %scrrrrrrr rIrF)rr}r<call_soon_threadsafer|r% reopen_filesr{r~r:rfrrSrLr8 _shutdown)rErframe min_workersrrr rxs\           zDirtyArbiter._signal_handlercCs|jr |jdSdS)zInitiate async shutdown.N)r;closerrrr rRszDirtyArbiter._shutdowncs`t|_tj|jrt|jtj|j |jdIdH|_ t |jd|j d|j|IdHt|}zTz'|j 4IdH|j IdHWdIdHn 1IdHs_wYWn tjtfyqYnwW|z|IdHWn tjyYnw|IdHdS|z|IdHWn tjyYnw|IdHw)z/Main async loop - start server, manage workers.)r.NizDirty arbiter listening on %s)roget_running_loopr<r'r.existsr0unlinkstart_unix_server handle_clientr;chmodr%rfrr_worker_monitor serve_foreverCancelledError RuntimeErrorcancelstop)rE monitor_taskrrr rqWsF  (zDirtyArbiter._run_asynccsn|jr5tdIdHt|jkr"|jdd|_|dS| IdH| IdH|jsdSdS)z1Periodically check worker health and manage pool.g?Nz+Parent changed, shutting down dirty arbiterF) r8rosleepr'getppidr)r%rLrmurder_workersrrrrr rs zDirtyArbiter._worker_monitorcs&||jr|IdHdSdS)z#Handle SIGCHLD - reap dead workers.N) reap_workersr8rrrrr rs zDirtyArbiter._handle_sigchldcsV|jdzzX|jr_z t|IdH}Wn tjy Yn?w|d}|tjkr5| ||IdHn'|tj krD| ||IdHn|tj krS| ||IdHn |||IdH|js Wntyy}z |jd|WYd}~nd}~wwW|z |IdHWdStyYdSw|z |IdHWwtyYww)a  Handle a connection from an HTTP worker. Routes requests to available dirty workers and returns responses. Supports both regular responses and streaming (chunk-based) responses. Also handles stash (shared state) operations. z&New client connection from HTTP workerNtypezClient connection error: %s)r%debugr8r read_message_asyncroIncompleteReadErrorrVMSG_TYPE_STASHhandle_stash_requestMSG_TYPE_STATUShandle_status_requestMSG_TYPE_MANAGEhandle_manage_request route_requestrKerrorr wait_closed)rEreaderwritermessagemsg_typerOrrr rsH       zDirtyArbiter.handle_clientc s$|dd}|d}||IdH}|dur>|js td}n|r*|jr*t|}ntd}t||}t||IdHdS||j vrK| |IdH|j |}t } |||| fIdHz| IdHWdSty} zt|td| |d}t||IdHWYd} ~ dSd} ~ ww)a Route a request to an available dirty worker via queue. Each worker has a dedicated queue and consumer task. Requests are submitted to the queue and processed sequentially by the consumer. For streaming responses, messages (chunks) are forwarded directly to the client_writer as they arrive from the worker. Args: request: Request message dict client_writer: StreamWriter to send responses to client idunknownr_NzNo dirty workers availablezRequest failed:  worker_id)rV_get_available_workerr1rr>rr r write_message_asyncr4_start_worker_consumerror create_futureputrKr ) rErequest client_writer request_idr_r^rresponsequeuefuturerOrrr rs8          zDirtyArbiter.route_requestcs>tj<fdd}t|}|j<dS)z3Start a consumer task for a worker's request queue.c sjr^zIIdH\}}}z7z||IdH|s$|dWnty@}z|s6||WYd}~nd}~wwWnwWn tj yXYdSwjsdSdSr) r8rV_execute_on_workerdone set_resultrK set_exception task_doneror)rrrrOrrEr^rr consumers,   z5DirtyArbiter._start_worker_consumer..consumerN)roQueuer4rr5)rEr^rtaskrrr rs   z#DirtyArbiter._start_worker_consumerc s|dd}z||IdH\}}t||IdH ztjt||jjdIdH}Wntj yLt |t d|jj}t||IdHYWdSw|d} | tj krat||IdHq| tj krrt||IdHWdS| tjtjfvrt||IdHWdS|jd| qty} z)|jd || ||t |td | |d }t||IdHWYd} ~ dSd} ~ ww) a Execute request on a specific worker (called by consumer). Handles both regular responses and streaming (chunk-based) responses. For streaming, chunk and end messages are forwarded directly to the client_writer as they arrive from the worker. rrNT)timeoutzWorker timeoutrz$Unknown message type from worker: %sz Error executing on worker %s: %szWorker communication failed: r)rV_get_worker_connectionr rrowait_forrr$ dirty_timeout TimeoutErrorr rMSG_TYPE_CHUNK MSG_TYPE_ENDMSG_TYPE_RESPONSEMSG_TYPE_ERRORr%rLrKr_close_worker_connectionr ) rEr^rrrrrrrrrOrrr r sZ     #   zDirtyArbiter._execute_on_workercs|r|jr||jvrt|j|}n dSt|j}|s dS|r8|jr8|j|d}|dt||j|<n |j}|dt||_||t|S)a Get an available worker PID using round-robin selection. If app_path is provided, only returns workers that have loaded that specific app. Uses per-app round-robin to ensure fair distribution among eligible workers. Args: app_path: Optional import path of the target app. If None, returns any worker using global round-robin. Returns: Worker PID or None if no eligible workers are available. Nrr) r>r?r[r1keysrArVrUr6)rEr_ eligible_pidsidxrrr rGs   z"DirtyArbiter._get_available_workercs||jvr |j|S|j|}|std|tdD]}tj|r(nt dIdHqtd|t |IdH\}}||f|j|<||fS)z%Get or create connection to a worker.zNo socket for worker 2皙?NzWorker socket not ready: ) r3r2rVrranger'r.rroropen_unix_connection)rEr^r0_rrrrr rqs     z#DirtyArbiter._get_worker_connectioncCs*||jvr|j|\}}|dSdS)zClose connection to a worker.N)r3rar)rEr^_readerrrrr rs  z%DirtyArbiter._close_worker_connectionc s|dd}t}g}|jD]5\}}z|j}t||d} Wntt t fy2d} Ynw| ||j t |dgt |dd| dq|jd d d |j|t||jr`t|jngd } t|| } t|| IdHdS) z Handle a status query request. Returns information about the dirty arbiter and its workers. Args: message: Status request message client_writer: StreamWriter to send response to client rrNrXbootedF)r&ageappsrlast_heartbeatcSs|dS)Nrr)rdrrr rsz4DirtyArbiter.handle_status_request..key) arbiter_pidr1rHr)rVtime monotonicr1rTtmp last_updateroundOSError ValueErrorAttributeErrorrWrrsortr&rUr>r[rr r r) rErrrnow workers_infor&workerrrresultrrrr rs4      z"DirtyArbiter.handle_status_requestc sH|dd}|d}tdt|dd}z|tkrad}t|D]}}|dur7jd7_|d7}tdIdHq"|dkrSd d |dd t j jd }n|d d ||t j jd }nn|t kr } d} t|D]9}j| krxn1t j dkrn(jd8_t j fddd} | tj| d7} tdIdHqod d|| t j jd}ntd|} t|| } t|| IdHWdSjd|tkrd nd||d|ddt||} t|| IdHWdSty#}z jd|t|tt|} t|| IdHWYd}~dSd}~ww)z Handle a worker management request. Supports adding or removing dirty workers via protocol messages. Args: message: Manage request message client_writer: StreamWriter to send response to client rroprcountrNrTr\z)All apps have reached their worker limits)success operation requestedspawnedreason total_workerstarget_workers)rrrrrrc j|jSrr1rprrr r z4DirtyArbiter.handle_manage_request..rremove)rrrremovedrrzUnknown manage operation: z6Worker management: %s %d workers (spawned/removed: %d)rrzManage operation error: %s)rVrQintrr spawn_workerr:rorrUr1rrSminr kill_workerrrwrr r rr%rfr rKrri)rErrrrrrrrrr oldest_pidrrrOrrr rs           z"DirtyArbiter.handle_manage_requestc s`|dd}|d}|dd}|d}|d}|dz`d }|tkr=||jvr3i|j|<||j||<d }n|tkr_||jvrKd d i}n||j|vrWd d i}n|j||}n|tkr{||jvrx||j|vrx|j||=d }nd }n|tkr||jvrg}nt|j|} rfdd| D} | }n|tkr||jvr|j| d }n|t kr||jvrd di}nqt |j||d}nf|t kr||jvri|j|<d }nU|t kr||jvr|j|=d }nEd }nB|tkrt|j}n6|tkr||jvrd }n(|d urd }n ||j|v}ntd|} t|| } t|| Id HWd St|trrd |vrr|d } | dkrLtd|} n| d krYtd|} ntt|} d| ddd| _t|| } nt||} t|| Id HWd Sty} z |jd| t|tt| } t|| Id HWYd } ~ d Sd } ~ ww)a! Handle a stash operation directly in the arbiter. All stash tables are stored in arbiter memory for simplicity and fast access. Args: message: Stash operation message client_writer: StreamWriter to send response to client rrrtablervaluepatternNTr key_not_foundFcs g|] }tt|r|qSr)fnmatchri)rkr rr r!Hs z5DirtyArbiter.handle_stash_request..table_not_found)sizerzUnknown stash operation: zTable not found: zKey not found: StashrErrorzStash operation error: %s)rVr rCrrrr[rrclearrrUrrrrrr r r isinstancedictrititlereplace error_typer rKr%r)rErrrrrrr rall_keysrrrrOrrr rs                             z!DirtyArbiter.handle_stash_requestcsjsdSj}jr.tj|kr.}|durntdIdHjr.tj|kstj|krZtjfddd} |t j tdIdHtj|ks5dSdS)z%Maintain the number of dirty workers.Nrcrrrrrrr rrz-DirtyArbiter.manage_workers..r) r8r:rUr1rrorrrrrrw)rEr:rrrrr rs"  zDirtyArbiter.manage_workersFc Cs|jr |jd}n|rt|j}n|}|s"|jddS|jd7_t j |j d|jd}t |j|j||j|j|d}t }|dkrp||_||j|<||j|<||||j|||jd|||St |_ztd |jjd |t dWdSty}zt |jdur|jndWYd}~dSd}~wty|jd |j st |j!t dYdSw) a Spawn a new dirty worker. Worker app assignment follows these priorities: 1. If there are pending respawns (from dead workers), use those apps 2. Otherwise, determine apps for a new worker based on allocation 3. If force_all_apps=True, spawn with all apps regardless of limits Args: force_all_apps: If True, spawn worker with all apps ignoring limits Returns: Worker PID in parent process, or None if no apps need workers rz)No apps need more workers, skipping spawnNrzworker-z.sock)rr)rXr$r%r0z,Spawned dirty worker (pid: %s) with apps: %szdirty-worker []z!Exception in dirty worker process)"rBrar[r>rrZr%rr7r'r.r/r-rr&r$forkr1r2r`dirty_post_forkrfr(rrn proc_name init_process_exit SystemExitcoderK exceptionrWORKER_BOOT_ERROR)rEforce_all_appsrXr0rr&rOrrr rsZ      (   zDirtyArbiter.spawn_workerc Cs^z t||WdSty.}z|jtjkr#||WYd}~dSWYd}~dSd}~ww)zKill a worker by PID.N)r'killrerrnoESRCH_cleanup_worker)rEr&rrOrrr rs zDirtyArbiter.kill_workercCs||||jvr|j||j|=|j|d||jvr0t|j|}|r0|j|| ||j |d}|rE|j |||j |d}|rgtj|rizt|WdStyfYdSwdSdS)z Clean up after a worker exits. Saves the dead worker's app list to pending respawns so the replacement worker gets the same apps. N)rr5rr4rar@r[rBrWrcr1r$dirty_worker_exitr2r'r.rrr)rEr& dead_appsrr0rrr r)s*      zDirtyArbiter._cleanup_workerc s|jjsdSt|jD]=\}}zt|j|jjkr"WqWn t t fy.Yqw|j sD|j d|d|_ ||tjq||tjqdS)z!Kill workers that have timed out.NzDIRTY WORKER TIMEOUT (pid:%s)T)r$rr[r1rTrrrrrrabortedr%criticalrrSIGABRTSIGKILL)rEr&rrrr rs"zDirtyArbiter.murder_workersc CszG tdtj\}}|sWdSd}t|rt|}nt|r/t|}|jd||||j kr;|j d|| ||j d|qt ya}z|jtjkrVWYd}~dSd}~ww)zReap dead worker processes.TNz)Dirty worker (pid:%s) killed by signal %sz$Dirty worker failed to boot (pid:%s)zDirty worker exited (pid:%s))r'waitpidWNOHANG WIFEXITED WEXITSTATUS WIFSIGNALEDWTERMSIGr%rLr$rr)rfrr'ECHILD)rEwpidstatusexitcoderrOrrr r*s0       zDirtyArbiter.reap_workerscsn|jdt|jjD]}|tdIdHq t|j }||jjdD] }| |t j q+dS)z!Reload workers (SIGHUP handling).zReloading dirty workersrN)r%rfrr$r9rrorr[r1rrrrw)rEr old_workersr&rrr rCs zDirtyArbiter.reloadTcs|jD]}|q|rtjntj}t|jj}t |j D]}| ||q$|j rKt|krK| tdIdH|j rKt|ks6t |j D] }| |tjqR| dS)zStop all workers.rN)r5rPrrrwryrr$dirty_graceful_timeoutr[r1rrrrorr/)rEgracefulrrlimitr&rrr rQs  zDirtyArbiter.stopcCs|jrtj|jrzt|jWn tyYnwtj|jr5zt|jWn ty4Ynwzt|jD] }ttj |j|qs D