AҐi^DUdZddlZddlZddlZddlZddlZddlZddlZddlm Z m Z m Z ddl m Z mZGddZGdd ZGd d Zej$Zej(d Zej(eed <dadZdZddefdZddefdZdZdZy)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_requestcxeZdZdZddZdZdZdZdZdZ dZ d Z d Z d Z d Zd ZdZdZdZdZdZy) 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, eventlet). c|||_||_d|_d|_d|_t j |_y)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 r s N/var/www/descvideos/venv/lib/python3.12/site-packages/gunicorn/dirty/client.py__init__zDirtyClient.__init__(s6'    ^^% c|jy tjtjtj|_|jj |j |jj |jy#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 settimeoutr connectr errorOSErrorrres rrzDirtyClient.connect;s :: !  v~~v7I7IJDJ JJ ! !$,, / JJ  t// 0 g& DJ&6qc: ,,  sBBC*"C  Ccn|j5|j||||cdddS#1swYyxYw)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.executePs8$ZZ H''&$G H H Hs+4c*|j|jttj}t |||||} t j|j|t j|j}|j|S#tj$r(|jtd|jt$r5}|jt|t rt#d||d}~wwxYw)zExecute while holding the lock.N request_idr'r(r)r*&Timeout waiting for dirty app responser Communication error: )rrstruuiduuid4rr write_message read_message_handle_responserr _close_socketr Exception isinstancerr rr'r(r)r*r.requestresponser#s rr%zDirtyClient._execute_lockedes ::  LLN& !   K  ' ' G <%11$**=H((2 2~~     #8   K    !Z(&)>qc'BC J  Ks ABAD0D  Dc 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#464HHrc|jd}|tjk(r|jdS|tjk(r)|jdi}t j |}|t d|)z 0* &DL$,*## #5  ) &6qc: ,,  s;CAA'A%A'$C%A''8B?B::B??CcK|j|jd{ttj}t |||||} t j|j|d{tjt j|j|jd{}|j|S77]7#tj$r1|jd{7t!d|jt"$r>}|jd{7t%|t&rt)d||d}~wwxYww)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-r0r/r1)rrTr2r3r4rrwrite_message_asyncrOrPread_message_asyncrr r7rR _close_asyncrr9r:rrr;s r execute_asynczDirtyClient.execute_asyncsD& << $$& & && !   K33DLL'J J J%--00> H ((2 2/ ' K## ##% % %#8   K##% % %!Z(&)>qc'BC J  Ksp E&C1E&#C8C9AC>C?CE&CC&E#D$E#%E8D;9%EE##E&c 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(hfMMrcK|jL |jj|jjd{d|_d|_yy7#t$rYwxYwwzClose the async connection.N)rrK wait_closedr9rrLs rrXzDirtyClient._close_async3sf << #  ""$ll..000 DLDL $1  s: A-7AAA A-A A*'A-)A**A-c@K|jd{y7wr^)rXrLs r close_asynczDirtyClient.close_async>s!!! c&|j|SN)rrLs r __enter__zDirtyClient.__enter__Fs  rc$|jyrd)rKrexc_typeexc_valexc_tbs r__exit__zDirtyClient.__exit__Js  rcBK|jd{|S7wrd)rTrLs r __aenter__zDirtyClient.__aenter__Ms"  """  #s c@K|jd{y7wrd)rargs r __aexit__zDirtyClient.__aexit__Qs   rbN>@)__name__ __module__ __qualname____doc__rrr+r%r@r7r8rKrTrYr\rXrarerkrmrorrr r sd &&*H*#KJI8 C!46KpN8  "!rr c:eZdZdZdZdZ d dZdZdZdZ d Z y) 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) rq@Nc||_||_||_||_||_d|_d|_d|_d|_d|_ |||_ yt|j|j|_ yNFclientr'r(r)r*_started _exhausted _request_id _deadline_last_chunk_timeminDEFAULT_IDLE_TIMEOUTr _idle_timeoutrr|r'r(r)r* idle_timeouts rrzDirtyStreamIterator.__init__mx        $)4L T..? rc|SrdrvrLs r__iter__zDirtyStreamIterator.__iter__ rc|jrt|js|jd|_|j SNT)r~ StopIterationr}_start_request_read_next_chunkrLs r__next__zDirtyStreamIterator.__next__s8 ?? }}    ! DM$$&&rcH|jj5|jj|jjt j }||jj z|_||_ttj|_ t|j|j|j|j |j"}t%j&|jj|dddy#1swYyxYwz(Send the initial request to the arbiter.N)r)r*)r|rrrtime monotonicr rrr2r3r4rrr'r(r)r*rr5rnowr<s rrz"DirtyStreamIterator._start_requests [[   D{{  ( ##%.."C 4;;#6#66DN$'D !"4::<0D "    YY{{ G  ' ' (9(97 C# D D Ds C8DD!cx|jj5tj}||jk\r(d|_t d|jj|j|z }||jkDr |j}nt||j} |jjj|tj|jj}tj|_|j)d }|tj*k(r|j)d cdddS|tj,k(r d|_t.|tj0k(r.d|_|j)d i}t3j4||tj6k(r d|_t.d|_t3d |#tj$r~tj}||jk\r(d|_t d|jj||j z }d|_t d|dd|jt"$r5}d|_|jj%t'd||d}~wwxYw#1swYyxYw) &Read the next message from the stream.TStream exceeded total timeoutr0%Timeout waiting for next chunk (idle .1fs)r1NrBdatar Unknown message type: )r|rrrrr~rr _TIMEOUT_THRESHOLDrrrrrr6rrr9r8rrDMSG_TYPE_CHUNK MSG_TYPE_ENDrrFrrGrE) rr remaining read_timeoutr= idle_durationr#rHrIs rrz$DirtyStreamIterator._read_next_chunksy [[  F B.."Cdnn$"&'3 KK// ,I4222#66 "9d.@.@A  O !!,,\:(55dkk6G6GH,%)NN$4D !||F+H=777||F+cF BF Bh=555"&##=777"&%\\'26  **:66=:::"&###DO5hZ@A Aa>> nn&$..(&*DO+7 $ 3 3!$d&;&; ; "&';M#;NbQ .. O"& ))+*-B1#+FGQN OKF BF Bs:B J0%AG 3A J0 BJ0 BJ-80J((J--J00J9rd) rrrsrtrurrrrrrrrvrrr?r?Zs8  # $'D*HBrr?c:eZdZdZdZ d dZdZdZdZdZ d Z y) 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. rqNc||_||_||_||_||_d|_d|_d|_d|_d|_ |||_ yt|j|j|_ yrzr{rs rrz!DirtyAsyncStreamIterator.__init__rrc|SrdrvrLs r __aiter__z"DirtyAsyncStreamIterator.__aiter__rrcK|jrt|js|jd{d|_|j d{S7#7wr)r~StopAsyncIterationr}rrrLs r __anext__z"DirtyAsyncStreamIterator.__anext__sP ??$ $}}%%' ' ' DM**,,, (-s!2AAAAAAc"K|jj"|jjd{tj}||jj z|_||_ttj|_ t|j|j|j|j|j }t#j$|jj|d{y77wr)r|rrTrrr rrr2r3r4rrr'r(r)r*rrVrs rrz'DirtyAsyncStreamIterator._start_requests ;;   &++++- - -nnt{{222 #tzz|,    MM KK;;  // 0C0CWMMM . Ns"4DD CDD D DrxcpKtj}||jk\r(d|_t d|j j |j|z } ||jkDr2tj|j jd{}n\t||j}tjtj|j j|d{}tj|_|j)d }|tj*k(r|j)d S|tj,k(r d|_t.|tj0k(r.d|_|j)d i}t3j4||tj6k(r d|_t.d|_t3d |7F7#tj$rwd|_tj}||jk\r!t d|j j ||j z }t d|dd|jt"$r>}d|_|j j%d{7t'd||d}~wwxYww) rTrr0Nrrrr1rBrr r)rrrr~rr|r rrrWrrrrOrPrRrr9rXrrDrrrrFrrGrE) rrrr=rrr#rHrIs rrz)DirtyAsyncStreamIterator._read_next_chunk4srnn $.. "DO#/ ++  NNS(  K4222!.!A!AKK''" #9d.@.@A !(!1!1!44T[[5H5HI(".!% 0<<' }33 3<<' ' }11 1"DO$ $ }33 3"DO!gr2J&&z2 2 }66 6"DO$ $1(<==o ## "DO.."Cdnn$'3 KK// $"7"77M#7 c7J"M**  K"DO++**, , ,&)>qc'BC J Ks]AJ6;G$GAG$6G"7G$;C$J6G$"G$$BJ35$J.JJ..J33J6rd) rrrsrtrurrrrrrrrvrrr[r[s7  # $-N,J>rr[ dirty_client_async_client_varc$|addlm}||y)z@Set the global dirty socket path (called during initialization).r)set_stash_socket_pathN)_dirty_socket_pathstashr)pathrs rset_dirty_socket_pathrs-$rcvt.tjjd}|r|St dtS)zGet the dirty socket path.GUNICORN_DIRTY_SOCKETz\Dirty socket path not configured. Make sure dirty_workers > 0 and dirty_apps are configured.)rosenvironrDr)rs rget_dirty_socket_pathrs>!zz~~56 K I   rreturncpttdd}|"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 rNr0)getattr _thread_localrr rr r|r s rget_dirty_clientrs8*]ND 9F ~+- [':%+ " MrcK tj}|S#t$r0t}t ||}tj |Y|SwxYww)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 r0)rrD LookupErrorrr setrs rget_dirty_client_asyncrsW*&"&&( M &+- [':f% M &s%AA5AAAAcbttdd}||jdt_yy)z4Close the thread-local client (call on worker exit).rN)rrrKrr|s rclose_dirty_clientrs, ]ND 9F  %) "rcK tj}|jd{y7#t$rYywxYww)z%Close the context-local async client.N)rrDrarrs rclose_dirty_client_asyncrs< "&&(  """    s,A'313A3 ?A?Arp)rurO contextvarsrrrrr3errorsrrrprotocolrrr r?r[localr ContextVarr__annotations__rrrrrrrrvrrrs     s!s!v OBOBdR>R>t ! :P9O9O:;))+6   k:+<* r