j^LUdZddlZddlZddlZddlZddlZddlZddlZddlm Z m Z m Z ddl m Z mZGddZGdd ZGd d ZejZejd Zejeed <dadZdZddefdZddefdZdZdZdS)zu Dirty Client Client for HTTP workers to communicate with the dirty worker pool. Provides both sync and async APIs. N)DirtyConnectionError DirtyErrorDirtyTimeoutError) DirtyProtocol make_requestczeZdZdZddZdZdZdZdZdZ d Z d Z d Z d Z d ZdZdZdZdZdZdZdS) DirtyClientz Client for calling dirty workers from HTTP workers. Provides both sync and async APIs. The sync API is for traditional sync workers (sync, gthread), while the async API is for async workers (asgi, gevent). >@c|||_||_d|_d|_d|_t j|_dS)z Initialize the dirty client. Args: socket_path: Path to the dirty arbiter's Unix socket timeout: Default timeout for operations in seconds N) socket_pathtimeout_sock_reader_writer threadingLock_lock)selfr rs H/var/www/html/venv/lib/python3.11/site-packages/gunicorn/dirty/client.py__init__zDirtyClient.__init__(s;'    ^%% cp|jdS tjtjtj|_|j|j|j|jdS#tjtf$r&}d|_td||j|d}~wwxYw)z Establish sync socket connection to arbiter. Raises: DirtyConnectionError: If connection fails N$Failed to connect to dirty arbiter: r ) rsocketAF_UNIX SOCK_STREAM settimeoutrconnectr errorOSErrorrres rr zDirtyClient.connect;s : ! F v~v7IJJDJ J ! !$, / / / J  t/ 0 0 0 0 0 g&   DJ&:q:: ,  sA,A99B5!B00B5ct|j5|||||cdddS#1swxYwYdS)a Execute an action on a dirty app (sync/blocking). Args: app_path: Import path of the dirty app (e.g., 'myapp.ml:MLApp') action: Action to call on the app *args: Positional arguments **kwargs: Keyword arguments Returns: Result from the dirty app action Raises: DirtyConnectionError: If connection fails DirtyTimeoutError: If operation times out DirtyError: If execution fails N)r_execute_lockedrapp_pathactionargskwargss rexecutezDirtyClient.executePs$Z H H''&$GG H H H H H H H H H H H H H H H H H Hs -11c@|j|ttj}t |||||} t j|j|t j|j}| |S#tj $r+| td|j t$rB}| t|t rt#d||d}~wwxYw)zExecute while holding the lock.N request_idr(r)r*r+&Timeout waiting for dirty app responserCommunication error: )rr struuiduuid4rr write_message read_message_handle_responserr _close_socketr Exception isinstancerr rr(r)r*r+r/requestresponser$s rr&zDirtyClient._execute_lockedes@ :  LLNNN&& !     K  ' G < < <%1$*==H((22 2~       #8   K K K    !Z(( &'Bq'B'BCC J  KsABAD=DDc(t|||||S)a* Stream results from a dirty app action (sync). This method returns an iterator that yields chunks from a streaming response. Use this for actions that return generators. Args: app_path: Import path of the dirty app (e.g., 'myapp.ml:MLApp') action: Action to call on the app *args: Positional arguments **kwargs: Keyword arguments Yields: Chunks of data from the streaming response Raises: DirtyConnectionError: If connection fails DirtyTimeoutError: If operation times out DirtyError: If execution fails Example:: for chunk in client.stream("myapp.llm:LLMApp", "generate", prompt): print(chunk, end="", flush=True) )DirtyStreamIteratorr's rstreamzDirtyClient.streams4#464HHHrc|d}|tjkr|dS|tjkr,|di}t j|}|t d|)zmsg_type error_infor!s rr8zDirtyClient._handle_responses<<'' }6 6 6<<)) ) 5 5 5!gr22J(44EKAxAABB Brc||j4 |jn#t$rYnwxYwd|_dSdS)zClose the socket connection.N)rcloser:rs rr9zDirtyClient._close_socketsY : !    """"    DJJJ " !s # 00cn|j5|ddddS#1swxYwYdS)zClose the sync connection.N)rr9rMs rrLzDirtyClient.closes Z ! !     ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! !s *..cVK|jdS tjtj|j|jd{V\|_|_dS#tj$rtd|jttf$r}td||j|d}~wwxYw)z Establish async connection to arbiter. Raises: DirtyConnectionError: If connection fails Nr1z#Timeout connecting to dirty arbiterrr) rasynciowait_foropen_unix_connectionr rr TimeoutErrorrr"ConnectionErrorrr#s r connect_asynczDirtyClient.connect_asyncs < # F /6/?,T-=>> 000****** &DL$,,,#   #5  )   &:q:: ,  sAA5B( B##B(cK|j|d{Vttj}t |||||} t j|j|d{Vtj t j |j |j d{V}| |S#tj$r1|d{Vt!d|j t"$rH}|d{Vt%|t&rt)d||d}~wwxYw)a Execute an action on a dirty app (async/non-blocking). Args: app_path: Import path of the dirty app action: Action to call on the app *args: Positional arguments **kwargs: Keyword arguments Returns: Result from the dirty app action Raises: DirtyConnectionError: If connection fails DirtyTimeoutError: If operation times out DirtyError: If execution fails Nr.r1r0r2)rrUr3r4r5rrwrite_message_asyncrPrQread_message_asyncrrr8rS _close_asyncrr:r;rrr<s r execute_asynczDirtyClient.execute_asyncs& < $$&& & & & & & & &&& !     K3DL'JJ J J J J J J J%-0>> H ((22 2#   ##%% % % % % % % %#8   K K K##%% % % % % % % %!Z(( &'Bq'B'BCC J  KsA,CAEAEEc(t|||||S)a8 Stream results from a dirty app action (async). This method returns an async iterator that yields chunks from a streaming response. Use this for actions that return generators. Args: app_path: Import path of the dirty app (e.g., 'myapp.ml:MLApp') action: Action to call on the app *args: Positional arguments **kwargs: Keyword arguments Yields: Chunks of data from the streaming response Raises: DirtyConnectionError: If connection fails DirtyTimeoutError: If operation times out DirtyError: If execution fails Example:: async for chunk in client.stream_async("myapp.llm:LLMApp", "generate", prompt): await response.write(chunk) )DirtyAsyncStreamIteratorr's r stream_asynczDirtyClient.stream_asyncs4(hfMMMrcK|jZ |j|jd{Vn#t$rYnwxYwd|_d|_dSdSzClose the async connection.N)rrL wait_closedr:rrMs rrYzDirtyClient._close_async3s < #  ""$$$l..0000000000    DLDLLL $ #s8A AAc>K|d{VdSr_)rYrMs r close_asynczDirtyClient.close_async>s0!!!!!!!!!!!rc.||SN)r rMs r __enter__zDirtyClient.__enter__Fs  rc.|dSrd)rLrexc_typeexc_valexc_tbs r__exit__zDirtyClient.__exit__Js rc>K|d{V|Srd)rUrMs r __aenter__zDirtyClient.__aenter__Ms/  """"""""" rc>K|d{VdSrd)rbrgs r __aexit__zDirtyClient.__aexit__Qs0           rNr )__name__ __module__ __qualname____doc__rr r,r&rAr8r9rLrUrZr]rYrbrerkrmrorrr r s6 & & & &&*HHH*#K#K#KJIII8 C C C!!!46K6K6KpNNN8    """!!!!!rr c<eZdZdZdZdZ d dZdZdZdZ d Z dS) r@a Iterator for streaming responses from dirty workers (sync). This class is returned by `DirtyClient.stream()` and yields chunks from a streaming response until the end message is received. Uses a deadline-based timeout approach: - Total stream timeout: limits entire stream duration - Idle timeout: limits gap between chunks (defaults to total timeout) r @Nc||_||_||_||_||_d|_d|_d|_d|_d|_ ||nt|j |j |_ dSNFclientr(r)r*r+_started _exhausted _request_id _deadline_last_chunk_timeminDEFAULT_IDLE_TIMEOUTr _idle_timeoutrr{r(r)r*r+ idle_timeouts rrzDirtyStreamIterator.__init__mv        $)4LLT.?? rc|SrdrurMs r__iter__zDirtyStreamIterator.__iter__ rc|jrt|js|d|_|SNT)r} StopIterationr|_start_request_read_next_chunkrMs r__next__zDirtyStreamIterator.__next__sG ?  } !    ! ! ! DM$$&&&rc|jj5|jj|jt j}||jjz|_||_ttj |_ t|j |j|j|j|j}t%j|jj|ddddS#1swxYwYdSz(Send the initial request to the arbiter.N)r*r+)r{rrr time monotonicrrrr3r4r5r~rr(r)r*r+rr6rnowr=s rrz"DirtyStreamIterator._start_requests* [  D D{ ( ##%%%.""C 4;#66DN$'D !"4:<<00D "   Y{ G  ' (97 C C C# D D D D D D D D D D D D D D D D D DsCCC#&C#c|jj5tj}||jkr"d|_t d|jj|j|z }||jkr|j}nt||j } |jj |tj|jj }n#tj$rmtj}||jkr"d|_t d|jj||jz }d|_t d|dd|j t"$r8}d|_|jt'd||d}~wwxYwtj|_|d }|tjkr!|d cdddS|tjkrd|_t.|tjkr1d|_|d i}t3j||tjkrd|_t.d|_t3d |#1swxYwYdS) &Read the next message from the stream.TStream exceeded total timeoutr1%Timeout waiting for next chunk (idle .1fs)r2NrCdatar!Unknown message type: )r{rrrrr}rr_TIMEOUT_THRESHOLDrrrrrr7rrr:r9rrEMSG_TYPE_CHUNK MSG_TYPE_ENDrrGrrHrF) rr remaining read_timeoutr> idle_durationr$rIrJs rrz$DirtyStreamIterator._read_next_chunks [ F BF B.""Cdn$$"&'3 K/ ,I4222#6 "9d.@AA  O !,,\:::(5dk6GHH>   n&&$.((&*DO+7 $ 3!$d&; ; "&'QMQQQQ . O O O"& ))+++*+F1+F+FGGQN O %)N$4$4D !||F++H=777||F++cF BF BF BF BF BF BF BF Bh=555"&##=777"&%\\'266  *:666=:::"&###DO@h@@AA AMF BF BF BF BF BF BF BF BF BF Bs?A3I2=B?>I2?BE;3E66E;;AI2BI22I69I6rd) rqrrrsrtrrrrrrrrurrr@r@Zs   #    $'''DDD*HBHBHBHBHBrr@c<eZdZdZdZ d dZdZdZdZdZ d Z dS) r\a Async iterator for streaming responses from dirty workers. This class is returned by `DirtyClient.stream_async()` and yields chunks from a streaming response until the end message is received. Uses a deadline-based timeout approach for efficiency: - Total stream timeout: limits entire stream duration - Idle timeout: limits gap between chunks (defaults to total timeout) This avoids the overhead of asyncio.wait_for() on every chunk read. r Nc||_||_||_||_||_d|_d|_d|_d|_d|_ ||nt|j |j |_ dSryrzrs rrz!DirtyAsyncStreamIterator.__init__rrc|SrdrurMs r __aiter__z"DirtyAsyncStreamIterator.__aiter__rrcK|jrt|js!|d{Vd|_|d{VSr)r}StopAsyncIterationr|rrrMs r __anext__z"DirtyAsyncStreamIterator.__anext__so ? %$ $} !%%'' ' ' ' ' ' ' ' DM**,,,,,,,,,rcK|jj|jd{Vtj}||jjz|_||_ttj |_ t|j |j |j|j|j}t#j|jj|d{VdSr)r{rrUrrrrrr3r4r5r~rr(r)r*r+rrWrs rrz'DirtyAsyncStreamIterator._start_requests ;  &+++-- - - - - - - -nt{22 #tz||,,   M K;    / 0CWMMMMMMMMMMMrrwcKtj}||jkr"d|_t d|jj|j|z } ||jkr%tj |jj d{V}nMt||j }tjtj |jj |d{V}n#tj$rfd|_tj}||jkrt d|jj||jz }t d|dd|j t"$r>}d|_|jd{Vt'd||d}~wwxYwtj|_|d }|tjkr|d S|tjkrd|_t.|tjkr1d|_|d i}t3j||tjkrd|_t.d|_t3d |) rTrr1Nrrrr2rCrr!r)rrrr}rr{rrrrXrrrrPrQrSrr:rYrrErrrrGrrHrF) rrrr>rrr$rIrJs rrz)DirtyAsyncStreamIterator._read_next_chunk4sn $. "DO#/ +  NS(  K4222!.!AK'"" #9d.@AA !(!1!4T[5HII("""#   "DO.""Cdn$$'3 K/ $"77M#M MMMM*  K K K"DO+**,, , , , , , , ,&'Bq'B'BCC J K !% 0 0<<'' }3 3 3<<'' ' }1 1 1"DO$ $ }3 3 3"DO!gr22J&z22 2 }6 6 6"DO$ $<(<<===sA=C A=F 9FFrd) rqrrrsrtrrrrrrrrurrr\r\s   #    $---NNN,J>J>J>J>J>rr\ dirty_client_async_client_varc,|addlm}||dS)z@Set the global dirty socket path (called during initialization).r)set_stash_socket_pathN)_dirty_socket_pathstashr)pathrs rset_dirty_socket_pathrs5-,,,,,$rct2tjd}|r|St dtS)zGet the dirty socket path.NGUNICORN_DIRTY_SOCKETz\Dirty socket path not configured. Make sure dirty_workers > 0 and dirty_apps are configured.)rosenvironrEr)rs rget_dirty_socket_pathrsI!z~~566  K I    rr returncttdd}|+t}t||}|t_|S)a Get or create a thread-local sync client. This is the recommended way to get a client in sync HTTP workers. Args: timeout: Timeout for operations in seconds Returns: DirtyClient: Thread-local client instance Example:: from gunicorn.dirty import get_dirty_client def my_view(request): client = get_dirty_client() result = client.execute("myapp.ml:MLApp", "inference", data) return result rNr1)getattr _thread_localrr rrr{r s rget_dirty_clientrsD*]ND 9 9F ~+-- [':::%+ " MrcK t}nI#t$r<t}t ||}t|YnwxYw|S)a Get or create a context-local async client. This is the recommended way to get a client in async HTTP workers. Args: timeout: Timeout for operations in seconds Returns: DirtyClient: Context-local client instance Example:: from gunicorn.dirty import get_dirty_client_async async def my_view(request): client = await get_dirty_client_async() result = await client.execute_async("myapp.ml:MLApp", "inference", data) return result r1)rrE LookupErrorrr setrs rget_dirty_client_asyncrsw*&"&&(( &&&+-- [':::f%%%%%& MsAA$#A$czttdd}|"|dt_dSdS)z4Close the thread-local client (call on worker exit).rN)rrrLrr{s rclose_dirty_clientrs< ]ND 9 9F  %) """rcK t}|d{VdS#t$rYdSwxYw)z%Close the context-local async client.N)rrErbrrs rclose_dirty_client_asyncrsh "&&((  """""""""""      s39 AArp)rtrP contextvarsrrrrr4errorsrrrprotocolrrr r@r\localr ContextVarr__annotations__rrrrrrrrurrrs#     s!s!s!s!s!s!s!s!v OBOBOBOBOBOBOBOBdR>R>R>R>R>R>R>R>t !! :P9O::;)+6       k:+<***     r