o
    پi6                  	   @   sp  d Z 	 ddlZddlZddlZddlZddlZddlmZmZ ddl	m
Z
 ddlmZ ddlmZ ddlmZmZmZmZ ddlZddlZddlmZ eeZG d	d
 d
ejddddZG dd dejdddddZdZdZG dd deZG dd deZ G dd deZ!G dd deZ"G dd deZ#G dd de#Z$G dd  d e#Z%G d!d" d"eZ&G d#d$ d$Z'dS )%a(  
Copyright 2025 SGLang Team
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
    N)ABCabstractmethod)deque)count)Queue)AnyCallableOptionalUnion)	BaseModelc                   @   s2   e Zd ZU eed< ee ed< dZee	 ed< dS )
EventBatchtseventsNattn_dp_rank)
__name__
__module____qualname__float__annotations__listr   r   r	   int r   r   W/home/ubuntu/.local/lib/python3.10/site-packages/sglang/srt/disaggregation/kv_events.pyr   &   s   
 r   TF)
array_likeomit_defaultsgcc                   @   s   e Zd ZdZdS )KVCacheEventz*Base class for all KV cache-related eventsN)r   r   r   __doc__r   r   r   r   r   1   s    r   )r   r   r   tagGPU
CPU_PINNEDc                   @   sV   e Zd ZU ee ed< ee ed< ee ed< eed< ee ed< dZee ed< dS )BlockStoredblock_hashesparent_block_hash	token_ids
block_sizelora_idNmedium)	r   r   r   r   r   r   r	   r'   strr   r   r   r   r!   @   s   
 r!   c                   @   s*   e Zd ZU ee ed< dZee ed< dS )BlockRemovedr"   Nr'   )	r   r   r   r   r   r   r'   r	   r(   r   r   r   r   r)   I   s   
 r)   c                   @   s   e Zd ZdS )AllBlocksClearedN)r   r   r   r   r   r   r   r*   N   s    r*   c                   @   s$   e Zd ZU eeeeef  ed< dS )KVEventBatchr   N)	r   r   r   r   r
   r!   r)   r*   r   r   r   r   r   r+   R   s   
 r+   c                   @   sD   e Zd ZdZddefddZededdfd	d
ZedddZ	dS )EventPublishera  
    Lightweight publisher for EventBatch batches with
    support for DP attention.

    In DP attention - each rank has its own Scheduler and
    KV cache instance in order to avoid duplicate events
    and ensure proper event attribution. In our implementation

    - Each DP rank has its own EventPublisher
    - Publishers annotate events with the dp rank
    - This allows consumers to distinguish events from different DP ranks
    r   r   c                 C   s
   || _ d S N)_attn_dp_rank)selfr   r   r   r   __init__d   s   
zEventPublisher.__init__r   returnNc                 C      dS )zEmit events in order.

        Implementations should guarantee at-least-once delivery and
        monotonic ordering (e.g., via sequence numbers).
        Nr   r/   r   r   r   r   publishg       zEventPublisher.publishc                 C   r2   )zShutdown the publisher.Nr   r/   r   r   r   shutdowno   r5   zEventPublisher.shutdownr   r1   N)
r   r   r   r   r   r0   r   r   r4   r7   r   r   r   r   r,   V   s    r,   c                   @   s$   e Zd ZdZdddZdddZdS )	NullEventPublisherz-No-op implementation (default when disabled).r1   Nc                 C      d S r-   r   r3   r   r   r   r4   w      zNullEventPublisher.publishc                 C   r;   r-   r   r6   r   r   r   r7   z   r<   zNullEventPublisher.shutdownr9   )r   r   r   r   r4   r7   r   r   r   r   r:   t   s    
r:   c                       s   e Zd ZU dZdZeed< djddddZ			
				d&de	de
dee
 de	de	de	de
dd
f fddZdedd
fddZd'ddZd'ddZd'dd Zd'd!d"Zedee
 d#e	dee
 fd$d%Z  ZS )(ZmqEventPublishera  Reliable PUB/ROUTER publisher with an in-memory replay buffer.

    Spawns a separate thread to handle publishing from a queue.

    Parameters
    ----------
    endpoint:
        PUB address. Use ``tcp://*:5557`` to bind or ``tcp://host:5557`` to
        connect.
    replay_endpoint:
        Optional ROUTER address for replay requests. When given, subscribers can
        request missed batches by sending the starting sequence number as an
        8-byte big-endian integer.
    buffer_steps:
        Number of past batches to keep for replay.
    hwm:
        ZeroMQ high-water-mark for PUB socket.
    max_queue_size:
        Maximum number of events to buffer in memory.
    topic:
        Topic to publish events to.
    g      ?SHUTDOWN_TIMEOUT   bigT)signedtcp://*:5557N'  順  r   endpointreplay_endpointbuffer_stepshwmmax_queue_sizetopicr1   c                    s   t  | ttt  |d| _tttt	f  |d| _
tj | _d | _d | _|| _| || j| _| || j| _|| _|   t | _|d| _d| _td tj| j ddd| _!| j!"  t#$| j% d S )N)maxsize)maxlenzutf-8TzStarting ZMQ publisher threadzzmq-publisher)targetdaemonname)&superr0   r   r	   r   _event_queuer   tupler   bytes_bufferzmqContextinstance_ctx_pub_replay_dp_rankoffset_endpoint_port	_endpoint_replay_endpoint_hwm_socket_setupr   _seq_genencode_topic_bytes_runningloggerinfo	threadingThread_publisher_thread_threadstartatexitregisterr7   )r/   r   rG   rH   rI   rJ   rK   rL   	__class__r   r   r0      s,   

zZmqEventPublisher.__init__r   c                 C   s0   | j std|jd u r| j|_| j| d S )NzPublisher is closed)rf   RuntimeErrorr   r]   rS   putr3   r   r   r   r4      s
   
zZmqEventPublisher.publishc                 C   s   d| _ | jd t }d}|r2t | | jk r2| j  }|r'td |r2t | | jk s|r?td| j	 | j | j
 rL| j
j| jd z| jdurY| jjdd | jdurh| jjdd W dS W dS w )	z1Stop the publisher thread and clean up resources.FNT皙?z:Warning: Queue still has %s items after %s seconds timeouttimeoutr   )linger)rf   rS   
put_nowaittimer>   emptysleeprg   warningqsizerl   is_alivejoinr[   closer\   )r/   rm   pending_itemsr   r   r   r7      s2   



zZmqEventPublisher.shutdownc                 C   s   | j du rB| jtj| _ | j | j d| jv s*d| jv s*| jds*| jdr;t	
d| j  | j | j n| j | j | jdura| jtj| _t	
d| j  | j| j dS dS )zoInitialize sockets
        https://pyzmq.readthedocs.io/en/v19.0.0/morethanbindings.html#thread-safety
        N*z::zipc://z	inproc://z4ZmqEventPublisher socket publisher_endpoint bind to z1ZmqEventPublisher socket replay_endpoint bind to )r[   rZ   socketrW   PUBset_hwmra   r_   
startswithrg   debugbindconnectr`   ROUTERr\   r6   r   r   r   rb      s*   







zZmqEventPublisher._socket_setupc              
   C   sd  t j | _| jdusJ | js| j dkr| jdurA| j	drAz| 
  W n ty@ } ztd| W Y d}~nd}~ww z| jjdd}|du rPW dS W n
 tjy[   Y qw z*t| j}| j|}|dd}| j| j||f | j||f | j  W n ty } ztd| td W Y d}~nd}~ww | js| j dksdS dS )	z1Background thread that processes the event queue.Nr   zError in replay: %srt   ru   r@   rA   zError in publisher thread: %s)msgspecmsgpackEncoder_packr[   rf   rS   r}   r\   poll_service_replay	Exceptionrg   	exceptiongetqueueEmptynextrc   rd   to_bytessend_multipartre   rV   append	task_donery   r{   )r/   eeventseqpayload	seq_bytesr   r   r   rk   
  s>   
z#ZmqEventPublisher._publisher_threadc              	   C   s   | j dusJ | j  }t|dkrtd| dS |\}}}t|d}| jD ]\}}||kr>| j |d|	dd|f q(| j |d| j
df dS )z6If a replay request is waiting, send buffered batches.N   zInvalid replay request: %srA       r@   )r\   recv_multipartlenrg   r|   r   
from_bytesrV   r   r   END_SEQ)r/   frame	client_id_start_seq_bytes	start_seqr   bufr   r   r   r   /  s   

z!ZmqEventPublisher._service_replaydata_parallel_rankc                 C   s   | r|dkr| S d| v r|  d| S d| v r?| r=d| v r=|  d}| d| }t| |d d }|| }| d| S | S td)	a  Helper function to offset the port in an endpoint by
            the data parallel rank.

        Args:
            endpoint: The endpoint string
                (e.g., "tcp://*:5557" or "inproc://cache")
            data_parallel_rank: The data parallel rank to offset by

        Returns:
            The endpoint with the port offset by data_parallel_rank
                or suffix appended
        r   inproc_dptcp:N   z0Invalid endpoint: must contain 'inproc' or 'tcp')rfindr   
ValueError)rG   r   last_colon_idx	base_addr	base_portnew_portr   r   r   r^   F  s   
z&ZmqEventPublisher.offset_endpoint_port)rC   NrD   rE   rE   rF   r9   )r   r   r   r   r>   r   r   r   r   r   r(   r	   r0   r   r4   r7   rb   rk   r   staticmethodr^   __classcell__r   r   rp   r   r=   ~   sP   
 	*


!
%r=   c                   @   s   e Zd ZU dZdZeed< 	 dZeed< 	 dZe	e ed< 	 dZ
eed	< 	 d
Zeed< 	 d
Zeed< 	 dZeed< 	 ededd fddZdS )KVEventsConfigz&Configuration for KV event publishing.null	publisherrC   rG   NrH   rD   rI   rE   rJ   rK   rF   rL   	cli_valuer1   c                 C   s
   t |S )z3Parse the CLI value for the event publisher config.)r   model_validate_json)clsr   r   r   r   from_cli  s   
zKVEventsConfig.from_cli)r   r   r   r   r   r(   r   rG   rH   r	   rI   r   rJ   rK   rL   classmethodr   r   r   r   r   r   h  s$   
 r   c                   @   sr   e Zd ZU eedZeeede	f f e
d< ededede	f ddfdd	Zeddee dede	fddZdS )EventPublisherFactory)r   rW   .	_registryrQ   ctorr1   Nc                 C   s(   || j v rtd| d|| j |< d S )Nzpublisher 'z' already registered)r   KeyError)r   rQ   r   r   r   r   register_publisher  s   
z(EventPublisherFactory.register_publisherr   configr   c              
   C   st   |st  S t|}| }|dd}z| j| }W n ty0 } z	td| d|d}~ww |dd|i|S )z'Create publisher from a config mapping.r   r   zUnknown event publisher ''Nr   r   )r:   r   r   
model_dumppopr   r   r   )r   r   r   config_dictkindconstructorexcr   r   r   create  s   
zEventPublisherFactory.creater8   )r   r   r   r:   r=   r   dictr(   r   r,   r   r   r   r	   r   r   r   r   r   r   r     s   
  "r   )(r   rn   loggingr   ri   ry   abcr   r   collectionsr   	itertoolsr   r   typingr   r   r	   r
   r   rW   pydanticr   	getLoggerr   rg   Structr   r   
MEDIUM_GPU
MEDIUM_CPUr!   r)   r*   r+   r,   r:   r=   r   r   r   r   r   r   <module>   sR    


	
 k(