o 1 io @sxddlZddlZddlZddlZddlmZddlmZddlm Z ddl m Z m Z ddl mZmZmZmZmZmZmZmZmZmZmZmZmZmZmZddlmZddlm Z m!Z!m"Z"dd l#m$Z$dd l%m&Z&dd l'm(Z(dd l)m*Z*dd l+m,Z,e-e&Z.e/e0gZ1eGdddZ2eGdddZ3eGdddZ4deeedeeeeeefffddZ5GdddZ6GdddZ7ddZ8ddZ9d e:ddfd!d"Z;ed#d$d%ZGd(d)d)eeefZ?Gd*d+d+eeefZ@ed,eee=gee>fdee=ge>ffd-d.ZAed/eee=geeeee>ffdee=geeee>fffd0d.ZAed1e?efdeeffd2d.ZAed3e@efdeefffd4d.ZAe  5 6 7dId8e:d9eBd e:d:eddd;f dd?  5 6 7dId8e:d9eBd e:d@eedef dAd.ZAdBejDdCefdDdEZEdBejDdFefdGdHZFdS)JN)deque) dataclass)wraps)isasyncgenfunctioniscoroutinefunction)AnyAsyncGeneratorCallable CoroutineDictGenericIterableListLiteralOptionalProtocolSetTupleTypeVaroverload)serve)extract_signature flatten_args recover_args)get_or_create_event_loop)SERVE_LOGGER_NAME)extract_self_if_method_call)RayServeException) PublicAPIc@s8eZdZUeed<eeed<ejed<ej j ed<dS)_SingleRequestself_argflattened_argsfuturerequest_contextN) __name__ __module__ __qualname__r__annotations__rasyncioFuturercontextZ_RequestContextr+r+^/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/serve/batching.pyr+s   rc@s eZdZUeed<ejed<dS)_GeneratorResultresult next_futureN)r$r%r&rr'r(r)r+r+r+r,r-3s r-c@sneZdZUeeed<edeefddZedeefddZ edeefddZ ede fd d Z d S) _RuntimeSummaryStatistics start_timesreturncC|jrt|jSdSN)r1minselfr+r+r,min_start_time=z(_RuntimeSummaryStatistics.min_start_timecCs|jr t|jt|jSdSr4)r1sumlenr6r+r+r,mean_start_timeAsz)_RuntimeSummaryStatistics.mean_start_timecCr3r4)r1maxr6r+r+r,max_start_timeGr9z(_RuntimeSummaryStatistics.max_start_timecCs t|jSr4)r;r1r6r+r+r, num_requestsKs z&_RuntimeSummaryStatistics.num_requestsN) r$r%r&rfloatr'propertyrr8r<r>intr?r+r+r+r,r09s r0list_of_flattened_argsr2cs|dd|D}t|dksJd|}g}t|D]ddkr-||dq|fdd|Dqt|S) z@Batch a list of flatten args and returns regular args and kwargscSsh|]}t|qSr+)r;).0argsr+r+r, Wz%_batch_args_kwargs..z=All batch requests should have the same number of parameters.rcsg|]}|qSr+r+rDitemidxr+r, crGz&_batch_args_kwargs..)r;poprangeappendr)rCZ arg_lengthsZ arg_lengthZbatched_flattened_argsr+rLr,_batch_args_kwargsPs  rRc @s4eZdZ d-dedededeeddf ddZd d Zd eddfd d Z de e e j fddfddZdee fddZdeededdfddZdedee j deddfddZde j dee j defddZdeddfd d!Zded"ee ddfd#d$Zd%e jddfd&d'Zed(eeddfd)d*Zd+d,ZdS). _BatchQueueNmax_batch_sizebatch_wait_timeout_smax_concurrent_batcheshandle_batch_funcr2cCsvt|_||_||_||_t||_t|_ t |_ i|_ d|_ t|_|dur5|j|||_ |dS)aOAsync queue that accepts individual items and returns batches. Respects max_batch_size and batch_wait_timeout_s; a batch will be returned when max_batch_size elements are available or the timeout has passed since the previous get. If handle_batch_func is passed in, a background coroutine will run to poll from the queue and call handle_batch_func on the results. Cannot be pickled. Arguments: max_batch_size: max number of elements to return in a batch. batch_wait_timeout_s: time to wait before returning an incomplete batch. max_concurrent_batches: max number of batches to run concurrently. handle_batch_func(Optional[Callable]): callback to run in the background to handle batches if provided. N)r(QueuequeuerTrUrV Semaphore semaphoreEventrequests_available_eventsettaskscurr_iteration_start_times_handle_batch_taskr_loop create_task_process_batches4_warn_if_max_batch_size_exceeds_max_ongoing_requestsr7rTrUrVrWr+r+r,__init__js    z_BatchQueue.__init__c CsFtjj}||j|jkr!td|jd|jd|ddSdS)zHelper to check whether the max_batch_size is bounded. Log a warning to configure `max_ongoing_requests` if it's bounded. z`max_batch_size` (z) * `max_concurrent_batches` (z)) is larger than `max_ongoing_requests` (z). This means the replica will never achieve the configured `max_batch_size` concurrently. Please update `max_ongoing_requests` to be >= `max_batch_size` * `max_concurrent_batches`.N)rZget_replica_contextZ_deployment_configmax_ongoing_requestsrTrVloggerwarning)r7rhr+r+r,res  z@_BatchQueue._warn_if_max_batch_size_exceeds_max_ongoing_requestsnew_max_batch_sizecCs||_|dSzUpdates queue's max_batch_size.N)rTrer7rkr+r+r,set_max_batch_sizes z_BatchQueue.set_max_batch_sizerequestcCs|j||jdSr4)rY put_nowaitr]r^)r7ror+r+r,puts z_BatchQueue.putcsg}||jIdH|j}|j}t} t|t|d}zt|j |IdHWn tj y<Ynwt ||kr[|j s[||jt ||kr[|j rH|j re|j t||ksst ||krv |Sq)aWait for batch respecting self.max_batch_size and self.timeout_s. Returns a batch of up to self.max_batch_size items. Waits for up to to self.timeout_s after receiving the first request that will be in the next batch. After the timeout, returns as many items as are ready. Always returns a batch with at least one item - will block indefinitely until an item comes in. NTr)rQrYgetrTrUtimer=r(wait_forr]wait TimeoutErrorr;empty get_nowaitclear)r7batchrTrUZbatch_start_timeZremaining_batch_time_sr+r+r,wait_for_batchs6    z_BatchQueue.wait_for_batchresultsinput_batch_lengthcCs*t||krtd|dt|ddS)NzHBatched function doesn't preserve batch size. The input list has length z" but the returned list has length .)r;r)r7r|r}r+r+r,_validate_resultss z_BatchQueue._validate_resultsfunc_generatorinitial_futuresc s"d}zlt|}t||ksJ|2zK3dHW}|||t|D]:}|||d}} | |ur7||n!|tvrFt| t||nt } t | t || || | q"q6|D] } | |urlt| tqaWdSt y} z|D] } | |urt| | qyWYd} ~ dSd} ~ ww)zConsumes batch function generator. This function only runs if the function decorated with @serve.batch is a generator. Nr)rr;rrPrQUSER_CODE_STREAMING_SENTINELS_set_exception_if_not_doneStopAsyncIterationr create_future_set_result_if_not_doner-popleft Exception) r7rrr}ZFINISHED_TOKENfuturesr|rMr.r"r/er+r+r,_consume_func_generatorsF            z#_BatchQueue._consume_func_generator func_futurerc sxz|IdH}|||t||D] \}}t||qWdSty;}z|D]}t||q(WYd}~dSd}~ww)z.Assigns func's results to the list of futures.N)rziprrr)r7rrr}r|r.r"rr+r+r,_assign_func_results#s    z _BatchQueue._assign_func_resultsfunccsrtj|js7|IdH}|||}t|}|j |t |j |<| |j|jr dSdS)z6Loops infinitely and processes queued request batches.N)rr*Z_unset_request_contextrb is_closedr{_process_batchr(rcr_addrsr`add_done_callback_handle_completed_task)r7rrzZpromisetaskr+r+r,rd4s      z_BatchQueue._process_batchesrzc s|j4IdHdd|D}t|dkr" WdIdHdSdd|D}z[|dj}tdd|D\}}|durJ||g|Ri|}n||i|}tjdd|Dt|ro|}|||t|IdHn|} | | |t|IdHtjgWn!t y} zt d|D]} t | | qWYd} ~ nd} ~ wwWdIdHdSWdIdHdS1IdHswYdS) zProcesses queued request batch.NcSsg|] }|js|qSr+)r" cancelledrDreqr+r+r,rNHsz._BatchQueue._process_batch..rcSg|]}|jqSr+)r"rJr+r+r,rNLcSrr+)r!rJr+r+r,rNTrcSrr+)r#rr+r+r,rN`rz0_process_batch ran into an unexpected exception.)r[r;r rRrr*Z_set_batch_request_contextrrrrri exceptionr) r7rrzrr rEkwargsZfunc_future_or_generatorrrrr"r+r+r,rAsJ         -.z_BatchQueue._process_batchrcCs&|j||j|=||dSr4)r_remover`_log_if_exceptionr)r7rr+r+r,rts z"_BatchQueue._handle_completed_taskexception_maybecCs4|durt|tjrtddStddSdS)NzTask was cancelledzTask failed unexpectedly) isinstancer(CancelledErrorridebugr)rr+r+r,rys  z_BatchQueue._log_if_exceptioncCs&|jdus ts dS|jdSr4)rar is_runningcancelr6r+r+r,__del__s z_BatchQueue.__del__r4)r$r%r&rBr@rr rgrernrrr(r)rqrr{r rrrrrrdrTaskr staticmethod BaseExceptionrrr+r+r+r,rSis\ -5   3  3 rSc @seZdZdZ    ddededed eefd d Ze d e fd dZ ded dfddZ ded dfddZ d efddZd efddZd efddZd efddZd eefddZdS) _LazyBatchQueueWrapperzStores a _BatchQueue and updates its settings. _BatchQueue cannot be pickled, you must construct it lazily at runtime inside a replica. This class initializes a queue only upon first access. rHNrTrUrVrWcCs"d|_||_||_||_||_dSr4)_queuerTrUrVrWrfr+r+r,rgs  z_LazyBatchQueueWrapper.__init__r2cCs(|jdurt|j|j|j|j|_|jS)zXReturns _BatchQueue. Initializes queue when called for the first time. N)rrSrTrUrVrWr6r+r+r,rYs z_LazyBatchQueueWrapper.queuerkcCs$||_|jdur|j|dSdSrl)rTrrnrmr+r+r,rns z)_LazyBatchQueueWrapper.set_max_batch_sizenew_batch_wait_timeout_scCs ||_|jdur||j_dSdSr4)rUr)r7rr+r+r,set_batch_wait_timeout_ss  z/_LazyBatchQueueWrapper.set_batch_wait_timeout_scC|jSr4rTr6r+r+r,get_max_batch_sizez)_LazyBatchQueueWrapper.get_max_batch_sizecCrr4rUr6r+r+r,get_batch_wait_timeout_srz/_LazyBatchQueueWrapper.get_batch_wait_timeout_scCstt|jjS)z;Gets summary statistics of current iteration's start times.)r0listrYr`valuesr6r+r+r,_get_curr_iteration_start_timessz6_LazyBatchQueueWrapper._get_curr_iteration_start_timescs t|jdr|jj SdS)zGets whether default _BatchQueue's background task is alive. Returns False if the batch handler doesn't use a default _BatchQueue. raF)hasattrrYradoner6r+r+r,_is_batching_task_alives z._LazyBatchQueueWrapper._is_batching_task_alivecs2t|jdrt}|jjj|d|SdS)zGets the stack for the default _BatchQueue's background task. Returns empty string if the batch handler doesn't use a default _BatchQueue. ra)fileN)rrYioStringIOra print_stackgetvalue)r7Z str_bufferr+r+r,_get_handling_task_stacks  z/_LazyBatchQueueWrapper._get_handling_task_stack)rrrHN)r$r%r&__doc__rBr@rr rgrArSrYrnrrrr0rboolrstrrr+r+r+r,rs0    rcCsNt|tst|tr|rt|}ntd||dkr%td|dS)Nz)max_batch_size must be integer >= 1, got rHz,max_batch_size must be an integer >= 1, got )rrBr@ is_integer TypeError ValueErrorrr+r+r,_validate_max_batch_sizes  rcCs6t|ttfstd||dkrtd|dS)Nz/batch_wait_timeout_s must be a float >= 0, got r)rr@rBrrrr+r+r,_validate_batch_wait_timeout_ssrrVcCs$t|tr |dkrtd|dS)NrHz4max_concurrent_batches must be an integer >= 1, got )rrBr)rVr+r+r, _validate_max_concurrent_batchess rSelfTypeT) contravariantTRc@*eZdZdedeedeefddZdS)_SyncBatchingMethodself__SyncBatchingMethod__batchr2cCdSr4r+)r7rrr+r+r,__call__ sz_SyncBatchingMethod.__call__Nr$r%r&rrrrrr+r+r+r,r "rc@r)_AsyncBatchingMethodr_AsyncBatchingMethod__batchr2csdSr4r+)r7rrr+r+r,rsz_AsyncBatchingMethod.__call__Nrr+r+r+r,rrr _sync_funccCrr4r+)rr+r+r,rzrz _async_funccCrr4r+)rr+r+r,rz _sync_methcCrr4r+)rr+r+r,rz"r _async_methcCrr4r+)rr+r+r,rz)rr{Gz?rHrTrU__BatchDecoratorcCrr4r+)rrTrUrVr+r+r,rz0sc @seZdZdZedeeegeefdeegeffddZ edeeege e e eeffdeege e e efffddZ ede e eefdee egeffd dZ ed ee eefdee ege e e efffd dZ d S) rzJDescibes behaviour of decorator produced by calling `batch` with argumentsrr2cCrr4r+)r7rr+r+r,r>rz_BatchDecorator.__call__rcCrr4r+)r7rr+r+r,rBrrcCrr4r+)r7rr+r+r,rHrrcCrr4r+)r7rr+r+r,rNrN)r$r%r&rrr rrrrr rrrrr+r+r+r,r;s,0    Zstable)Z stability_funccsd|durt|s tdt|stdtttfdd}t|r0||S|S)aConverts a function to asynchronously handle batches. The function can be a standalone function or a class method. In both cases, the function must be `async def` and take a list of objects as its sole argument and return a list of the same length as a result. When invoked, the caller passes a single object. These will be batched and executed asynchronously once there is a batch of `max_batch_size` or `batch_wait_timeout_s` has elapsed, whichever occurs first. `max_batch_size` and `batch_wait_timeout_s` can be updated using setter methods from the batch_handler (`set_max_batch_size` and `set_batch_wait_timeout_s`). Example: .. code-block:: python from ray import serve from starlette.requests import Request @serve.deployment class BatchedDeployment: @serve.batch(max_batch_size=10, batch_wait_timeout_s=0.1) async def batch_handler(self, requests: List[Request]) -> List[str]: response_batch = [] for r in requests: name = (await requests.json())["name"] response_batch.append(f"Hello {name}!") return response_batch def update_batch_params(self, max_batch_size, batch_wait_timeout_s): self.batch_handler.set_max_batch_size(max_batch_size) self.batch_handler.set_batch_wait_timeout_s(batch_wait_timeout_s) async def __call__(self, request: Request): return await self.batch_handler(request) app = BatchedDeployment.bind() Arguments: max_batch_size: the maximum batch size that will be executed in one call to the underlying function. batch_wait_timeout_s: the maximum duration to wait for `max_batch_size` elements before running the current batch. max_concurrent_batches: the maximum number of batches that can be executed concurrently. If the number of concurrent batches exceeds this limit, the batch handler will wait for a batch to complete before sending the next batch to the underlying function. Nz?@serve.batch can only be used to decorate functions or methods.z9Functions decorated with @serve.batch must be 'async def'cstdtjdtfdddtjffdd tfdd}tfd d }tr8|}n|}j|_j|_ j |_ j |_ j |_ j |_ j|_|S) N first_futurer2cSs<|} z|IdH}|j}|jVWn tyYdSwq)z1Generator that handles generator batch functions.TN)r/r.r)rr"Zasync_responser+r+r,batch_handler_generators   z@batch.._batch_decorator..batch_handler_generatorcs`tt||}t|}|dur|dd}j}t}tj}| t |||||S)NrI) rrrrYrrrr*Z_get_serve_request_contextrqr)rErr!r7Z batch_queuer"r#)rlazy_batch_queue_wrapperr+r,enqueue_requests     z8batch.._batch_decorator..enqueue_requestcs||}|Sr4r+)rErr)rrr+r,generator_batch_wrappers z@batch.._batch_decorator..generator_batch_wrappercs||IdHSr4r+)rEr)rr+r, batch_wrappersz6batch.._batch_decorator..batch_wrapper)rr(r)rrrrZ_get_max_batch_sizerZ_get_batch_wait_timeout_srnrrrr)rrrwrapperrUrTrV)rrrrr,_batch_decorators@ zbatch.._batch_decorator)callablerrrrr)rrTrUrVrr+rr,rzUs=Ur"r.cC|s ||dSdS)z3Sets the future's result if the future is not done.N)r set_result)r"r.r+r+r,rrrcCr)z6Sets the future's exception if the future is not done.N)r set_exception)r"rr+r+r,rrr)NrrrH)Gr(rloggingrs collectionsr dataclassesr functoolsrinspectrrtypingrrr r r r r rrrrrrrrZrayrZray._common.signaturerrrZray._common.utilsrZray.serve._private.constantsrZray.serve._private.utilsrZray.serve.exceptionsrZray.util.annotationsr getLoggerri StopIterationrrrr-r0rRrSrrrrBrrrrrrrzr@rr)rrr+r+r+r,s   D         'W 0        "