o ۷iG @sTdZddlmZddlZddlmZddlZddlmZGdddZ dddZ dS)z'ZMQ-based queue utilities for Omni IPC.) annotationsN)Any)make_zmq_socketc@sdeZdZdZdddddd#ddZd$ddZd$ddZd%d&ddZd'ddZd(dd Z d)d!d"Z dS)*ZmqQueuez#Queue-like wrapper on a ZMQ socket.N)bindconnectrecv_timeout_mssend_timeout_msctx zmq.Context socket_typeintr str | Nonerr int | Noner returnNonec Cs|dur|n|}|durtd|du}t||||dd|_t|_|j|jtj||_||_ |dur:||j_ |durB||j_ ||_ dS)Nz(Either bind or connect must be specifiedi)rlinger) ValueErrorr_socketzmqPoller_pollerregisterPOLLIN_default_recv_timeout_default_send_timeoutrcvtimeosndtimeoendpoint) selfr r rrrr path bind_moder"U/home/ubuntu/vllm_env/lib/python3.10/site-packages/vllm_omni/entrypoints/zmq_utils.py__init__s   zZmqQueue.__init__objrc Cs8z |j|WdStjy}zt|d}~ww)z:Send an object to the queue. Blocks until sent or timeout.N)r send_pyobjrAgainqueueFullrr%er"r"r#put4  z ZmqQueue.putc Cs>z |jj|tjdWdStjy}zt|d}~ww)z-Send an object to the queue without blocking.flagsN)rr&rNOBLOCKr'r(r)r*r"r"r# put_nowait;s  zZmqQueue.put_nowaittimeout float | NonecCsN|dur |jSt|jt|d}||jtjkr#|jSt )zBReceive an object from the queue with optional timeout in seconds.Ni) r recv_pyobjdictrpollr getrrr(Empty)rr2eventsr"r"r#r7Bs   z ZmqQueue.getc Cs8z |jjtjdWStjy}zt|d}~ww)z2Receive an object from the queue without blocking.r.N)rr4rr0r'r(r8)rr+r"r"r# get_nowaitMr-zZmqQueue.get_nowaitboolcCs"t|jd}||jtjkS)z-Check if the queue is empty without blocking.r)r5rr6r7rrr)rr9r"r"r#emptyTszZmqQueue.emptycCs|jddS)Nr)rclose)rr"r"r#r=YszZmqQueue.close)r r r r rrrrrrr rrr)r%rrr)N)r2r3rr)rr)rr;)rr) __name__ __module__ __qualname____doc__r$r,r1r7r:r<r=r"r"r"r#rs "   rr r rstrr r rcCst|||dS)z:Create a ZmqQueue from an endpoint string and socket type.)r)r)r rr r"r"r#create_zmq_queue]srC)r r rrBr r rr) rA __future__rr(typingrrvllm.utils.network_utilsrrrCr"r"r"r#s   N