o ڷi_@sdZddlmZddlZddlmZddlmZddlm Z ddl m Z ddl m Z mZmZmZmZmZddlZdd lmZmZmZGd d d eZGd d d ZGdddeejZGdddZedddZGdddeejeZdS)z%Future-returning APIs for coroutines.) annotationsN)Future)deque)partial)chain)Any AwaitableCallable NamedTupleTypeVarcast)EVENTSPOLLINPOLLOUTc@s>eZdZUded<ded<ded<ded<d ed <d ed <d S) _FutureEventrfuturestrkindtupleargsdictkwargsrmsgtimerN)__name__ __module__ __qualname____annotations__rrA/home/ubuntu/vllm_env/lib/python3.10/site-packages/zmq/_future.pyrs  rc@sFeZdZUdZdZded<ded<ddd Zdd d Zddd dZdS)_AsynczMixin for common async logicNr _current_loopz type[Future]_FuturereturncCsL|jdur||_||j|jS|}||jur$||_|||S)zGet event loop Notice if event loop has changed, and register init_io_state on activation of a new event loop N)r! _default_loop_init_io_state)self current_looprrr _get_loop1s     z_Async._get_loopcCstd)Nz!Must be implemented in a subclassNotImplementedError)r&rrrr$Bsz_Async._default_loopNonecCdSNrr&looprrrr%Esz_Async._init_io_state)r#rr-)r#r+) rrr__doc__r!rr(r$r%rrrrr +s    r csZeZdZUdZded<ded<ded<ded<dddZdddZddfdd ZZS) _AsyncPollerz:Poller that returns a Future on poll, instead of blocking.ztype[_AsyncSocket] _socket_classint_READ_WRITEz list[Any] raw_socketsr/rsocketevtfr r#r+cCt)z"Schedule callback for a raw socketr))r&r/r7r8r9rrr_watch_raw_socketQz_AsyncPoller._watch_raw_socketsocketscGr:)z$Unschedule callback for a raw socketr))r&r/r=rrr_unwatch_raw_socketsUr<z!_AsyncPoller._unwatch_raw_sockets Awaitable[list[tuple[Any, int]]]c s|dkr0ztd}Wnty(}z |WYd}~Sd}~ww|Sgfdd}fddgfdd}jD]W\}}t |t j rt |j sqj |}||t j@r}|jd d |t j@r|jd d qV|d}|t j@r|jO}|t j@r|jO}|||qVfd d } | rÈ||dur|dkrfd d} d|| fdd} | fdd} | S)z Return a Future for a poll eventrNcs ddSdSr-done set_result)rwatcherrrwake_rawmz#_AsyncPoller.poll..wake_rawcsjgRSr-)r>r9)r/r6r&rrrsz#_AsyncPoller.poll..csD]}|qdSr-)_clear_io_state)r9s)wrapped_socketsrr_clear_wrapper_iows z,_AsyncPoller.poll.._clear_wrapper_iopollrc srdSrzWdStyYdSwr)dSz ttd}Wnt yK}z |WYd}~dSd}~ww |dSNr) rC cancelledcancel RuntimeError exception set_exceptionsuperr1rO ExceptionrD)r9resulte) __class__rr&rFrr on_poll_readys&  z(_AsyncPoller.poll..on_poll_readycrAr-rBrrErrtrigger_timeoutrHz*_AsyncPoller.poll..trigger_timeoutMbP?cs$tdr dSdS)NrS)hasattrrSremove_timeoutrI)r/timeout_handlerrcancel_timeouts  z)_AsyncPoller.poll..cancel_timeoutcss dSdSr-)rCrSrIrErrcancel_watchers z)_AsyncPoller.poll..cancel_watcher)r"rWrOrXrVrDr(add_done_callbackr= isinstance_zmqSocketr2 from_socketappendr_add_recv_eventr_add_send_eventr4r5r; call_later) r&timeoutrYrZrGrNr7maskr8r\r]rbrcr[)rr/r6r&rarFrMrrOYsb                    z_AsyncPoller.poll) r/rr7rr8r3r9r r#r+)r/rr=rr#r+)r?)r#r@) rrrr0rr;r>rO __classcell__rrrorr1Is   r1c@seZdZeddZdS)_NoTimercCr,r-rrrrrrSsz_NoTimer.cancelN)rrr staticmethodrSrrrrrqsrqT _AsyncSocket)boundcs|eZdZUdZdZdZded<eZdZ    d]d^fd d Z e d_d`ddZ d_dafdd Z ejj je _fddZejjje_ dbdcd"d#Z dbddd%d&Zddd'ded*d+Z dbdfd.d/Z   dbdgd2d3Zd4d5Zdejfdhd6d7Zd8d9Zd:d;Zed_ddidGdHZdjdIdJZdKdLZ dMdNZ!dkdOdPZ"d_dQdRZ#dSdTZ$dUdVZ%dWdXZ&d_dYdZZ'd[d\Z(Z)S)lrtNr _zmq.Socket _shadow_sockr? _from_socket_zmq.Socket | Noner#r+c st|tjr d|}}|durtj|jd||_ntj||fi|tj|j|_|dur?tj |j j dt ddt |_t |_d|_|jj|_dS)N)shadowz^(io_loop) argument is deprecated in pyzmq 22.2. The currently active loop will always be used.) stacklevelr)rerfrgrW__init__ underlyingrwrzwarningswarnr[rDeprecationWarningr _recv_futures _send_futures_stateFD_fd)r&context socket_typeio_looprxrrorrr}s"   z_AsyncSocket.__init__clstype[T]r7rrrscCs |||dS)z.Create an async socket from an existing Socket)rxrr)rr7rrrrrhs z_AsyncSocket.from_socketlinger int | Nonec sz|js4|jdur4tt|jpg|jpg}|D]}|js/z|jWqt y.Yqwq| t j |ddS)N)r) closedrlistrrrrrCrSrTrKrWclose)r&r event_listeventrorrrs  z_AsyncSocket.closecs"t|}|tkr|||Sr-)rWgetr _schedule_remaining_events)r&keyrYrorrrs  z_AsyncSocket.getTFflagsr3copybooltrack)Awaitable[list[bytes] | list[_zmq.Frame]]cC|jdt|||ddS)zvReceive a complete multipart zmq message. Returns a Future whose result will be a multipart message. recv_multipartrrrrrjrr&rrrrrrrsz_AsyncSocket.recv_multipartAwaitable[bytes | _zmq.Frame]cCr)zReceive a single zmq frame. Returns a Future, whose result will be the received frame. Recommend using recv_multipart instead. recvrrrrrrrrs z_AsyncSocket.recvnbytesrrAwaitable[int]cCs|jd|ft||ddS)zReceive a single zmq frame into a pre-allocated buffer. Returns a Future, whose result will be the number of bytes received. recv_intor)rrr)r&bufrrrrrr&sz_AsyncSocket.recv_into msg_parts%Awaitable[_zmq.MessageTracker | None]cKs(||d<||d<||d<|jd||dS)zqSend a complete multipart zmq message. Returns a Future that resolves when sending is complete. rrrsend_multipartrr)rk)r&rrrrrrrrr1sz_AsyncSocket.send_multipartdatarcKs<||d<||d<||d<|t|||d|jd||dS)zSend a single zmq frame. Returns a Future that resolves when sending is complete. Recommend using send_multipart instead. rrrrsendr)updaterrk)r&rrrrrrrrr=s z_AsyncSocket.sendcs>|fdd}|fdd}|S)zDeserialize with Futuresc srsdurtdddtdSr*dS}z|}WntyL}z |WYd}~dSd}~ww |dS)z+Chain result through serialization to recvdNzFuture z completed while awaiting z. A message has been dropped!) rCrRrUrrRuntimeWarningrVrYrXrD)_rloadedrZr9loadrecvdrr_chainUs" z)_AsyncSocket._deserialize.._chaincs$rdSrdSdS)z"Chain cancellation from f to recvdN)rCrRrS)r)r9rrr _chain_cancelps  z0_AsyncSocket._deserialize.._chain_cancel)r"rd)r&rrrrrrr _deserializeQs   z_AsyncSocket._deserializecsjr ttj}||tt|| fdd} r0|n |fdd} |S)zSpoll the socket for events returns a Future for the poll results. cstrdSrzWdStyYdSw|r)dSt}| ddSrQ) rCrRrSrTrUrVrrYrDr)r9evtsr poll_futurer&rr unwrap_results   z(_AsyncSocket.poll..unwrap_resultcs0szWdStyYdSwdS)z4Cancel underlying poll if request has been cancelledN)rCrSrTrP)rrr cancel_polls z&_AsyncSocket.poll..cancel_poll) rrfZMQErrorENOTSUP _poller_classregisterr rrOr"rCrd)r&rmrprrrrrrO{s      z_AsyncSocket.pollcsfdd}|||S)z'Add a timeout for a send or recv FuturecsrdStdSr-)rCrVrfAgainrrPrrfuture_timeoutsz1_AsyncSocket._add_timeout..future_timeout) _call_later)r&rrmrrrPr _add_timeouts  z_AsyncSocket._add_timeoutcCs|||S)zSchedule a function to be called later Override for different IOLoop implementations Tornado and asyncio happen to both have ioloop.call_later with the same signature. )r(rl)r&delaycallbackrrrrsz_AsyncSocket._call_latercCs.|sdSz||WdStyYdSw)zMake sure that futures are removed from the event list when they resolve Avoids delaying cleanup until the next send/recv event, which may never come. N)remove ValueError)rrrrrr_remove_finished_futures z$_AsyncSocket._remove_finished_future)rrrrrr tuple | Nonedict[str, Any] | Noner Future | Nonerc Cs:|p|}|dur d}|duri}|drO|ddtj@rOt|j|}z ||i|}WntyG}z ||WYd}~|Sd}~ww| ||St } t tdrf|jj } | dkrf| || d} t||||d| d} |j| |jtt@r||jr| |jvr|t|j|j| d |t|S) z4Add a recv event, returning the corresponding FutureNrrrrRCVTIMEOr^rrrrrr)r" startswithrrfDONTWAITgetattrrwrXrVrDrqr_rcvtimeorrrrir r _handle_recvrdrr _add_io_state) r&rrrrr9rrrZr timeout_ms _future_eventrrrrjsJ       z_AsyncSocket._add_recv_eventc Csh|p|}|dvrw|jsw|dd}|}|tjB|d<t|j|}d} z ||fi|} Wn6tjyP} z|tj@rD| | nd} WYd} ~ n d} ~ wt yf} z | | WYd} ~ n d} ~ ww| | | rw|j ru| |St} ttdr|jtj} | dkr||| d} t||d ||| d }|j||t|j|j|d |t|S) z4Add a send event, returning the corresponding Future)rrrrTFNSNDTIMEOr^rrr)r"rrrrfrrrwrrVrXrDrrrqr_rrrrirdrrrr)r&rrrrr9r nowait_kwargsr finish_earlyrrZrrrrrrrk sR          z_AsyncSocket._add_send_eventc Cs,|jtt@s dSd}|jr%|j\}}}}}}|r!d}nn|js|js-|t|dur3dS||dkrB| ddS|dkrK|jj }n|dkrT|jj }n|dkr]|jj }nt d||dtjO<z ||i|}Wnty} z || WYd} ~ dSd} ~ ww| |dS)zHandle recv eventsNrOrrrzUnhandled recv event type: r)rwrr rrpopleftrC_drop_io_staterSrDrrrrrfrrXrV) r&r9rrrrrrrYrZrrrrJs@     z_AsyncSocket._handle_recvc Cs|jtt@s dSd}|jr%|j\}}}}}}|r!d}nn|js|js-|t|dur3dS||dkrB| ddS|dkrK|jj }n|dkrT|jj }nt d||dt jO<z ||fi|}Wnty} z || WYd} ~ dSd} ~ ww| |dS)NrOrrzUnhandled send event type: r)rwrr rrrrCrrSrDrrrrfrrXrV) r&r9rrrrrrrYrZrrr _handle_sendus<    z_AsyncSocket._handle_sendcCsH|jjrdS|jt}|tj@r||tj@r|| dS)z(Dispatch IO events to _handle_recv, etc.N) rwrrr rfrrrrr)r&fdevents zmq_eventsrrr_handle_eventss    z_AsyncSocket._handle_eventscCsB|jdkrdS|dur|jt}||j@r|d|jdSdS)zkSchedule a call to handle_events next loop iteration If there are still events to handle. rN)rrwrr rr)r&rrrrrs   z'_AsyncSocket._schedule_remaining_eventscCs*|j|kr |j|B}|_||jdS)zAdd io_state to poller.Nr_update_handlerr&staterrrrs z_AsyncSocket._add_io_statecCs(|j|@r |j|@|_||jdS)z&Stop poller from watching an io_state.Nrrrrrrs z_AsyncSocket._drop_io_statecCs|r||dS)zOUpdate IOLoop handler with state. zmq FD is always read-only. N)r(rrrrrrs z_AsyncSocket._update_handlercCs6|dur|}||j|j|j|d|jdS)z#initialize the ioloop event handlerNr)r( add_handlerrwrr4rr.rrrr%sz_AsyncSocket._init_io_statecCs2|j}|jjr |j}|jdur|j|dSdS)zNunregister the ioloop event handler called once during close N)rwrrr!remove_handler)r&rrrrrKs  z_AsyncSocket._clear_io_state)Nr?NN)rxryr#r+r-)rrr7rvrrr#rs)rrr#r+)rTF)rr3rrrrr#r)rr3rrrrr#r)rr3rr3r#r)rrrr3rrr#r) rrrr3rrrrrrr#r)r#r) rrrrrrrrr#r)NNN)rr)*rrrrrrrr1rrr} classmethodrhrrfrgr0rrrrrrrrrOrrrrrrjrkrrrrrrrr%rKrprrrorrtsd        *2   7=+ )  ) r0 __future__rrasyncior collectionsr functoolsr itertoolsrtypingrrr r r r zmqrfr rrrr Pollerr1rqrsrgrtrrrrs        w