AҐihdZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z m Z ddl mZmZmZmZddlmZmZmZmZmZmZmZmZmZmZmZmZmZm Z m!Z!ddl"m#Z#Gd d Z$y) 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 VeZdZdZdj Dcgc]}t t d|zc}}}}ZdZd&dZ dZ dZ d Z d Z d Zd Zd ZdZdZdZdZdZdZdZdZdZd'dZdZdZdZdZdZdZ d(dZ!dZ"d Z#d!Z$d"Z%d#Z&d)d$Z'd%Z(ycc}}}}w)* 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 z*HUP QUIT INT TERM TTIN TTOU USR1 USR2 CHLDzSIG%sNcN||_||_d|_tj|_||_tjd|_ |xs*tjj|jd|_ i|_ i|_i|_i|_i|_d|_d|_d|_|jj*|_d|_d|_i|_i|_i|_i|_i|_g|_i|_|jAy)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!r,r&s O/var/www/descvideos/venv/lib/python3.12/site-packages/gunicorn/dirty/arbiter.py__init__zDirtyArbiter.__init__BsIIK  &&.?@ & "'',, KK+   "$ " ! 8811  !#  !!# c,|jjD]H}t|\}}| t|}|||d|j|<t|j|<Jy#t$r'}|j j d||Yd}~Xd}~wwxYw)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;)rAspecrFrGes rBr@zDirtyArbiter._parse_app_specsysHH'' 5D(  ?rDcg}|jjD]b\}}|d}t|jj |t }||j |L||ksR|j |d|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 rG)r:itemslenr;getrLappend)rA app_pathsrFrMrGcurrent_workerss rB_get_apps_for_new_workerz%DirtyArbiter._get_apps_for_new_workers !%!5!5!7 . K/L!$"5"5"9"9+su"MNO#  - </  - .rDct||j|<|D]E}||jvrt|j|<|j|j |Gy)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;rLaddrA worker_pidrYapp_paths rB_register_worker_appsz"DirtyArbiter._register_worker_appss`+/y/J'! :Ht22203##H-    ) - -j 9 :rDc|jj|g}|D]/}||jvs|j|j|1y)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_s rB_unregister_workerzDirtyArbiter._unregister_workersV''++J; " BH4...##H-55jA BrDctj|_|jj d|j|j rD t |j d5}|jt|jddd|jtjd<|jj||jt!j"d t%j&|j)|j-y#1swYxYw#t$r&}|jjd|Yd}~d}~wwxYw#t*$rYZwxYw#|j-wxYw)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&openwritestrIOErrorrKr,environr on_dirty_starting init_signalsr _setproctitleasynciorun _run_asyncKeyboardInterrupt _cleanup_sync)rAfrNs rBrtzDirtyArbiter.runs(99;  8$((C << D$,,,+GGCM*+ /3.>.> *+ ""4(  ?+ ! KK) *    -++ D  !?CC D"!       sT D.#%D"D..#E "D+'D.. E7EE E,)E/+E,,E//FcN|jD]&}tj|tj(tjtj|jtjtj |jtjtj |jtjtj|jtjtj|jtjtj|jtjtj|jtjtj|jy)zSet up signal handlers.N) SIGNALSsignalSIG_DFLSIGTERM_signal_handlerSIGQUITSIGINTSIGHUPSIGUSR1SIGCHLDSIGTTINSIGTTOU)rAsigs rBrqzDirtyArbiter.init_signals s<< /C MM#v~~ . /  fnnd&:&:; fnnd&:&:; fmmT%9%9: fmmT%9%9: fnnd&:&:; fnnd&:&:; fnnd&:&:; fnnd&:&:;rDc(|tjk(r+jrjjfdy|tjk(rj j y|tjk(r+jrjjfdy|tjk(rfxjdz c_ j jdjjrjjfdy|tjk(rj}j|krj jd|yxjdzc_ j jdjjrjjfd yd _jr&jjjyy) zHandle signals.cJtjjSN)rs create_task_handle_sigchldrAsrBz.DirtyArbiter._signal_handler..sG//0D0D0FGrDNcJtjjSr)rsrreloadrsrBrz.DirtyArbiter._signal_handler..+sG// >rDrz'SIGTTIN: Increasing dirty workers to %scJtjjSrrsrmanage_workersrsrBrz.DirtyArbiter._signal_handler..6G//0C0C0EFrDzASIGTTOU: Cannot decrease below %s workers (required by app specs)z'SIGTTOU: Decreasing dirty workers to %scJtjjSrrrsrBrz.DirtyArbiter._signal_handler..IrrDF)r{rr8call_soon_threadsaferr! reopen_filesrrr6rjrrSrKr4 _shutdown)rArframe min_workerss` rBr~zDirtyArbiter._signal_handlers &.. zz //G  &.. HH ! ! #  &-- zz //>  &..    !  HHMMC** ,zz //F  &.. 335K;.  .     !  HHMMC** ,zz //F  :: JJ + +DNN ; rDcR|jr|jjyy)zInitiate async shutdown.N)r7closers rBrzDirtyArbiter._shutdownRs << LL    rDcKtj|_tjj |j rtj|j tj|j|j d{|_ tj|j d|jjd|j |jd{tj|j!} |j4d{|jj#d{dddd{|j) |d{|j+d{y777o7O7A#1d{7swYQxYw#tj$t&f$rYowxYw7\#tj$$rYowxYw7_#|j) |d{7n#tj$$rYnwxYw|j+d{7wxYww)z/Main async loop - start server, manage workers.)r*NizDirty arbiter listening on %s)rsget_running_loopr8r#r*existsr,unlinkstart_unix_server handle_clientr7chmodr!rjrr_worker_monitor serve_foreverCancelledError RuntimeErrorcancelstop)rA monitor_tasks rBruzDirtyArbiter._run_asyncWs--/  77>>$** + IId&& '%66   !!   !!5)  5t7G7GH!!###**4+?+?+AB  || 3 3ll00222 3 3    ! """))+  ;  $  32 3 3 3 3&& 5    #))       ! """))  ))+  s-BI FA"I 5F6'I F9.F/F92F$F F$ F9 F"!F9%I 6G;G<GI G3I I F9 F$"F9$F6*F- +F62F99GG5GG5GG0-I /G00I 5IH H HIH)&I(H))I?III cK|jrtjdd{tj|j k7r3|j jdd|_|jy|jd{|jd{|jryy77-7w)z1Periodically check worker health and manage pool.g?Nz+Parent changed, shutting down dirty arbiterF) r4rssleepr#getppidr%r!rKrmurder_workersrrs rBrzDirtyArbiter._worker_monitorsjj--$ $ $zz|tyy(  !NO"  %%' ' '%%' ' 'jj $ ( 's:%CB?A+CCC+C,C=CCCczK|j|jr|jd{yy7w)z#Handle SIGCHLD - reap dead workers.N) reap_workersr4rrs rBrzDirtyArbiter._handle_sigchlds3  ::%%' ' '  's 0;9;cK|jjd |jr tj|d{}|jd}|tjk(r|j||d{nv|tjk(r|j||d{nH|tjk(r|j||d{n|j||d{|jr|j# |j%d{y7#t j $rYAwxYw777u7\#t$r&}|jj!d|Yd}~wd}~wwxYw7Z#t$rYywxYw#|j# |j%d{7w#t$rYwwxYwxYww)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!debugr4r read_message_asyncrsIncompleteReadErrorrWMSG_TYPE_STASHhandle_stash_requestMSG_TYPE_STATUShandle_status_requestMSG_TYPE_MANAGEhandle_manage_request route_requestrJerrorr wait_closed)rAreaderwritermessagemsg_typerNs rBrzDirtyArbiter.handle_clients ?@ **$1$D$DV$LLG#;;v.};;;33GVDDD!>!>>44WfEEE!>!>>44WfEEE,,Wf==='**. LLN ((***/M22EFF> = HHNN8! < < = +   LLN ((***  sG ED7D5D7 9EE-E0E1-EEE9E:E GF /F 0F 4G5D77E  E E  EEEE F!F=FFF F FGFGG-GGGG GGGGGcK|jdd}|jd}|j|d{}|h|js td}n%|r|jr t |}n td}t ||}tj||d{y||jvr|j|d{|j|}tjj} |j||| fd{ | d{y777a77 #t$rC} t |t!d| |}tj||d{7Yd} ~ yd} ~ wwxYww)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 idunknownraNzNo dirty workers availablezRequest failed:  worker_id)rW_get_available_workerr-rr:rr r write_message_asyncr0_start_worker_consumerrsr create_futureputrJr ) rArequest client_writer request_idrar`rresponsequeuefuturerNs rBrzDirtyArbiter.route_requestsp[[y1 ;;z* 55h??  <<"#?@dnn4X>"#?@*:u=H33M8L L L  T// /--j9 9 9"":.))+99;ii-8999 MLL5@ M : :  M* #3A3!7:NH 33M8L L L  Ms8FD)A(F#D+$'F D- A FD/FD3#D1$D3(F+F-F/F1D33 E?<3E:/E20E:5F:E??FcKtjj<fd}tj|}|j<yw)z3Start a consumer task for a worker's request queue.cKjry jd{\}}} j||d{|js|j dj jrxyy7e7G#t $r+}|js|j |Yd}~Sd}~wwxYw#jwxYw#tj$rYywxYwwr) r4rW_execute_on_workerdone set_resultrJ set_exception task_donersr)rrrrNrrAr`s rBconsumerz5DirtyArbiter._start_worker_consumer..consumers**;@99;5F2G]F *"55& &{{}"--d3 )**5F %4%{{}"0034)--s C3CB CBB %B*C: C3C3 C B C!B=8C=CCCCC0-C3/C00C3N)rsQueuer0rr1)rAr`rtaskrs`` @rBrz#DirtyArbiter._start_worker_consumersK ).:& $""8:.,0j)sAAcK|jdd} |j|d{\}}tj||d{ t j tj ||jjd{}|jd} | tjk(rtj||d{| tjk(rtj||d{y| tjtjfvrtj||d{y|j j#d| 7B7$7#tj$rLt|td|jj}tj||d{7YywxYw777#t$$rq} |j j'd|| |j)|t|t+d | | }tj||d{7Yd} ~ yd} ~ wwxYww) 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. rrN)timeoutzWorker timeoutrz$Unknown message type from worker: %sz Error executing on worker %s: %szWorker communication failed: r)rW_get_worker_connectionr rrswait_forrr dirty_timeout TimeoutErrorr r MSG_TYPE_CHUNK MSG_TYPE_ENDMSG_TYPE_RESPONSEMSG_TYPE_ERRORr!rKrJr_close_worker_connectionr ) rAr`rrrrrrrrrNs rBrzDirtyArbiter._execute_on_worker s%[[y1 0 M#'#>#>z#JJNFF33FGD D D $+$4$4%88@ $ 6 6%G#;;v.};;;';;M7SSS}999';;M7SSS ? ? - < < >>';;M7SSS  !GRC K D ++2")*:DHH +v~ %Is$BC C,C ?CC C c~||jvr/|jj|\}}|jyy)zClose connection to a worker.N)r/rdr)rAr`_readerrs rBrz%DirtyArbiter._close_worker_connections8 00 0"5599*EOGV LLN 1rDc K|jdd}tj}g}|jj D]f\}} |j j }t||z d} |j||jt|dgt|dd| dh|jd |j|t!||j"r#t%|j"j'ngd } t)|| } t+j,|| d{y#tttf$rd} YwxYw7!w) 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 rrNrYbootedF)r"ageappsrlast_heartbeatc |dS)Nr)rhs rBrz4DirtyArbiter.handle_status_request..s %rDkey) arbiter_pidr-rGr)rWtime monotonicr-rUtmp last_updateroundOSError ValueErrorAttributeErrorrXrgetattrsortr"rVr:r]rr r r) rArrrnow workers_infor"workerrrresultrs rBrz"DirtyArbiter.handle_status_requests>[[y1 nn <<--/ KC &$jj446 !&s['8!!<   zz R8!&(E:"0 !   01 88# -37>>D,,./r  !V4// xHHH+Z8 &!% &* Is7A E )D55B:E/E0E5E E EEc TK|jdd}|jd}tdt|jdd} |tk(rd}t |D]K}j }|xj dz c_|dz }tjdd{M|dk(r)d d |dd tjj d }n]d d ||tjj d }n5|tk(rj} d} t |D]}j | krntjdkrnxj dzc_tjjfd} j| t j"| dz } tjdd{d d|| tjj d}n9t%d|} t'|| } t)j*|| d{yj,j/d|tk(rd nd||jd|jddt1||} t)j*|| d{y777~7 #t2$rc}j,j5d|t'|t%t7|} t)j*|| d{7Yd}~yd}~wwxYww)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)rrrrrrc6j|jSrr-rprAs rBrz4DirtyArbiter.handle_manage_request..s4<<?3F3FrDrremove)rrrremovedrrzUnknown manage operation: z6Worker management: %s %d workers (spawned/removed: %d)rrzManage operation error: %s)rWrQintrr spawn_workerr6rsrrVr-rrSminr kill_workerr{r}rr r rr!rjr rJrrm)rArrrr rrrr rr oldest_pidrrrNs` rBrz"DirtyArbiter.handle_manage_requests[[y1 [[ As7;;w234N M]"u-A!..0F)((A-(1 !--,,, -a<#'%*%*#$"M),T\\):*.*:*:F$(%*%*#*),T\\):*.*:*: F''"779 u -A'';64<<(A-$$)$"%T\\%6%6%8)F"HJ$$Z@qLG!--,,, - $!)!&&%(%6&*&6&6 #%?t#DE.z5A#77 xPPP HHMMR#%#6%H **Y 9a0HI K %Z8H33M8L L LA-R-Q M M HHNN7 ;*:z#a&7IJH33M8L L L MsA L( AJ9*J0+D*J9J3A J96J57J9;L(rd#CD.z5A#77 xPPP&$'Gv,=#G_ !22&):5''BCE?2&'>?E&s6{3E%*:+;+;+=+E+Ec2+N*Ou#U .z5A(V<33M8L L L{$IXQ" M M HHNN6 :*:z#a&7IJH33M8L L L Ms|A)P0,D O9(N8!EO<N==OP0B0O2N?3O7P08O?O P- AP(P P(#P0(P--P0chKjsyj}jrmtj|krUj }|nBt j dd{jrtj|krUtj|kDrtjjfd}j|tjt j dd{tj|kDr~yy77!w)z%Maintain the number of dirty workers.Nrc6j|jSrrrs rBrz-DirtyArbiter.manage_workers..s4<<?+>+>rDr) r4r6rVr-rrsrrrrr{r})rAr6r r s` rBrzDirtyArbiter.manage_workersszz && jjS.<&&(F~--$ $ $ jjS.<$,,+-T\\..0!>@J   Z 8--$ $ $ $,,+- % %s1A)D2,D.-(D2A8D2D0D2,D20D2cF|jr|jjd}n6|r$t|jj }n|j }|s|j jdy|xjdz c_tjj|jd|jd}t|j|j||j|j |}tj }|dk7rr||_||j"|<||j$|<|j'|||jj)|||j j+d|||Stj,|_ t/j0d |jj2d |j5tj6dy#t8$r7}tj6|j: |j:ndYd}~yd}~wt<$r^|j j?d |j@stj6|jBtj6dYywxYw) 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%rYr r!r,z,Spawned dirty worker (pid: %s) with apps: %szdirty-worker []z!Exception in dirty worker process)"r>rdr]r:rr[r!rr3r#r*r+r)rr"r forkr-r.rbdirty_post_forkrjr$rrr proc_name init_process_exit SystemExitcoderJ exceptionrWORKER_BOOT_ERROR)rAforce_all_appsrYr,r r"rNs rBrzDirtyArbiter.spawn_workers  ! !..2215I T^^0023I557I HHNNF G 1ggll KK74??"359 #  ggi !8FJ &DLL '2D   $  & &sI 6 HH $ $T6 2 HHMMHy *JYY[     0B0B/C1E F    ! HHQK : HHqvv1QVVq 9 9  HH  B C==//0 HHQK  s (AG;; J -H66A'J J c tj||y#t$r=}|jtjk(r|j |Yd}~yYd}~yd}~wwxYw)zKill a worker by PID.N)r#killrerrnoESRCH_cleanup_worker)rAr"rrNs rBrzDirtyArbiter.kill_workersK * GGC  *ww%++%$$S))& *s A.AAc|j|||jvr*|j|j|j|=|jj |d||j vr5t |j |}|r|jj||j||jj |d}|r|jj|||jj |d}|r7tjj!|r tj"|yyy#t$$rYywxYw)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)rr1rr0rdr<r]r>rXrfr-r dirty_worker_exitr.r#r*rrr)rAr" dead_appsr r,s rBrFzDirtyArbiter._cleanup_workers. %%c* $'' '  ! !# & - - /%%c* sD) $%% %T0056I&&--i8 $!!#t,  HH & &tV 4))--c48 277>>+6  +&7;  s2E EEc,K|jjsyt|jj D]\}} t j |jjz |jjkrN |jsD|jjd|d|_ |j|tj |j|tj"y#ttf$rYwxYww)z!Kill workers that have timed out.NzDIRTY WORKER TIMEOUT (pid:%s)T)r rr]r-rUrrrrrrabortedr!criticalrr{SIGABRTSIGKILL)rAr"r s rBrzDirtyArbiter.murder_workerssxx%%  2 2 45 6KC >>#fjj&<&<&>>$((BXBXXY >>!!"A3G!%  fnn5  fnn5 6Z(  s,ADAC? A5D?DDDDc\ tjdtj\}}|syd}tj|rtj|}nGtj |r2tj |}|jjd||||jk(r|jjd||j||jjd|#t$r(}|jtjk7rYd}~yd}~wwxYw)zReap dead worker processes.Nz)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!rKr@rrFrjrrDECHILD)rAwpidstatusexitcoderrNs rBrzDirtyArbiter.reap_workers*s !zz"bjj9 f<<'!~~f5H^^F+++f-CHH$$%P%)30t555HHNN#I4P$$T* >$** +  $**+  ZZ , 8 "'',,t{{A67 8 HHT[[ !  7B%       s6EE BE% EE E"!E"% E10E1)NNr)F)T))__name__ __module__ __qualname____doc__splitrr{rzr@rCr@rSr[rbrfrtrqr~rrurrrrrrrrrrrrrrrrFrrrrrw).0xrr{s0000rBrr1s<AACEEwvw{+EG5 n 5D"6:$B"!@ <8ros]    @ " OCOCrD