o
    ci+                     @   s   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Z	d dl
mZ d dlmZ d dlmZmZ d dlmZ d dlmZ edd	d
dZG dd dejZG dd dZG dd deZdS )    N)TupleUnion)ActorHandle)
RuntimeEnv)	Blackholerun_throughput_benchmark)PayloadPydantic)DeploymentHandlehttpdebug)
GRPC_TRACEGRPC_VERBOSITY)env_varsc                   @   s   e Zd ZdZdZdS )IOModeSYNCASYNCN)__name__
__module____qualname__r   r    r   r   b/home/ubuntu/.local/lib/python3.10/site-packages/ray/serve/_private/benchmarks/streaming/common.pyr      s    r   c                   @   s*   e Zd ZdefddZdd Zdd ZdS )	Endpointtokens_per_requestc                 C   s.   || _ tdtj tdtj d S Nrayz	ray.serve)_tokens_per_requestlogging	getLoggersetLevelWARNING)selfr   r   r   r   __init__   s   zEndpoint.__init__c                 c   sL    t ddd tddD ttddt dd}t| jD ]}|V  qd S )NTest outputc                 S      g | ]}t |qS r   float.0fr   r   r   
<listcomp>$       z#Endpoint.stream.<locals>.<listcomp>   d   Success!textfloatsintstsreasonr   rangelisttimer   r    payloadir   r   r   stream!      zEndpoint.streamc                 C  sL   t ddd tddD ttddt dd}t| jD ]}|V  qd S )Nr"   c                 S   r#   r   r$   r&   r   r   r   r)   0   r*   z'Endpoint.aio_stream.<locals>.<listcomp>r+   r,   r-   r.   r4   r8   r   r   r   
aio_stream-   r<   zEndpoint.aio_streamN)r   r   r   intr!   r;   r=   r   r   r   r   r      s    r   c                   @   sr   e Zd Zdeeef dededededefddZ	d	d
 Z
ejdd Zdd Zdd Zdeeef fddZdS )Caller
downstreammoder   
batch_size
num_trialstrial_runtimec                C   sR   || _ || _|| _|| _|| _|| _g | _td	tj
 td	tj
 d S r   )_h_moder   _batch_size_num_trials_trial_runtime
_durationsr   r   r   r   )r    r@   rA   r   rB   rC   rD   r   r   r   r!   ;   s   
zCaller.__init__c                 C   s:   | j tjkr
| jjS | j tjkr| jjS td| j  d)NzStreaming mode not supported ())rF   r   r   rE   r;   r   r=   NotImplementedErrorr    r   r   r   _get_remote_methodQ   s
   zCaller._get_remote_methodc                    s   d S )Nr   rM   r   r   r   _consume_single_streamY   s   zCaller._consume_single_streamc                    s6   t j fddt jD  I d H } j| d S )Nc                    s   g | ]}   jqS r   )_executerO   )r'   _rM   r   r   r)   _   s    
z+Caller._do_single_batch.<locals>.<listcomp>)asynciogatherr5   rG   rJ   extend)r    	durationsr   rM   r   _do_single_batch]   s   

zCaller._do_single_batchc                    s*   t  }| I d H  t  | }|d S )Ni  )r7   	monotonic)r    fnstartdur_sr   r   r   rP   g   s
   zCaller._executereturnc                    s`   t | j| j| j | j| jd}|I d H }t| jg d\}}}t	d| d| d|  |S )N)rX   
multiplierrC   rD   )2   K   c   z#Individual request quantiles:
	P50=z
	P75=z
	P99=)
r   rV   rG   r   rH   rI   np
percentilerJ   print)r    corototal_runtimep50p75p99r   r   r   run_benchmarkm   s   

zCaller.run_benchmarkN)r   r   r   r   r   r	   r   r>   r%   r!   rN   abcabstractmethodrO   rV   rP   r   rh   r   r   r   r   r?   :   s(    



r?   )ri   rR   enumr   r7   typingr   r   numpyr`   	ray.actorr   ray.runtime_envr   $ray.serve._private.benchmarks.commonr   r   2ray.serve._private.benchmarks.serialization.commonr   ray.serve.handler	   GRPC_DEBUG_RUNTIME_ENVEnumr   r   r?   r   r   r   r   <module>   s$     